volume

func DownloadBlockChunks ¶

func DownloadBlockChunks(
	ctx context.Context,
	log *slog.Logger,
	chunkDir string,
	blockURL string,
	totalSize int64,
	chunkSize int64,
	workers int,
	fetcher *exporter.Fetcher,
	codec compress.Codec,
	onProgress func(n int),
) error

DownloadBlockChunks downloads all chunks of a block volume into chunkDir. Chunk k covers raw bytes [k*chunkSize, min((k+1)*chunkSize, totalSize)). Each chunk is fetched via a Range GET, encoded as an independent frame using codec, and atomically written as chunk_NNNNN[.<ext>] where ext is codec.Ext().

chunkDir is the absolute path to the chunk directory (the caller constructs it using archive.BlockChunksDirName).

Already-complete chunks (final file exists) are skipped. Stale *.tmp files are cleaned before a chunk is fetched. workers bounds parallelism; the first error cancels all in-flight work.

Durable sub-chunk resume: each chunk's raw bytes are streamed directly to a durable "<chunk>.part" file as they arrive (see partSuffix) instead of being buffered in memory for the whole chunk. An interrupted chunk resumes with a Range GET starting at the ".part" file's TRUSTED prefix — the offset last proven durable by a successful fsync (see partOffsetSuffix), never the raw file size alone — truncating away any tail the file may physically hold beyond that offset first. This bounds a kill mid-chunk to losing at most the bytes written since the last fsync (partSyncInterval), even when the process is killed hard enough that the file's on-disk SIZE outruns its DATA. The final codec frame is produced, and the ".part" file consumed, only once the raw bytes are fully durable on disk.

Memory note: once a chunk's ".part" file is complete, finalizeChunkFrame streams it through codec.EncodeFrameStream directly into the final chunk's AtomicWriter — the whole raw chunk is never read into memory as a []byte here. Finalize memory is bounded by each codec's own fixed windows and buffers, independently of chunkSize; zstd uses a fresh single-concurrency stream writer per frame. The outer pipeline still multiplies that per-frame bound by the concurrent-node and per-volume worker limits.

func DownloadBlockChunksRooted ¶

func DownloadBlockChunksRooted(
	ctx context.Context,
	destination *archive.RootedDestination,
	log *slog.Logger,
	chunkDir string,
	blockURL string,
	totalSize int64,
	chunkSize int64,
	workers int,
	fetcher *exporter.Fetcher,
	codec compress.Codec,
	onProgress func(n int),
) error

DownloadBlockChunksRooted downloads chunks through destination's locked view.

func DownloadFilesystemVolume ¶

func DownloadFilesystemVolume(
	ctx context.Context,
	log *slog.Logger,
	tarPath string,
	stagingDir string,
	filesRootURL string,
	workers int,
	chunkSize int64,
	fetcher *exporter.Fetcher,
	codec compress.Codec,
	setTotal func(total int64),
	onProgress func(n int),
) error

DownloadFilesystemVolume downloads all files from the data-exporter filesystem volume at filesRootURL, stages each file as a compressed blob named <relPath><codec.Ext()> under stagingDir, then assembles a single uncompressed PAX tar at tarPath whose file entries carry the compressed names and bytes.

If tarPath already exists, its parent directory is synced before the tar is trusted or staging is removed (resume: published tar becomes durable). An already-staged compressed file <relPath><ext> is not re-downloaded (partial resume). The stagingDir is removed on successful tar assembly.

workers bounds both active file downloads and the file-job queue. At most workers active fsItems plus workers queued fsItems are retained. Inventory sorting retains at most fsInventorySortBatchSize items; each merge opens at most fsInventoryMergeFanIn runs and retains one item per run. Every spool record is capped at fsInventoryMaxRecordSize. Total spool disk is linear in source metadata: one directory-queue record per directory plus at most two sorted-run copies during a merge pass. It lives under stagingDir/.d8-meta and is removed only after data.tar is durably committed. The first error cancels all in-flight downloads.

chunkSize bounds the size of each Range-based chunk used to stage a file whose declared size is known (item.size > 0): every such file is staged via stageChunkedFile, reusing DownloadBlockChunks/MergeBlockChunks (the same durable, ".part"-resumable machinery the block-volume path uses) — a single chunk when size <= chunkSize, multiple chunks otherwise — so an interrupted download of ANY known-size file resumes from its last durably-persisted offset instead of restarting from byte zero. chunkSize <= 0 falls back to DefaultChunkSize. A file whose declared size is unknown (item.size < 0) or exactly zero keeps the original single-shot GET + codec.EncodeStream path: chunk geometry needs a trustworthy total size up front, and there is no meaningful partial to resume for zero declared bytes.

setTotal, when non-nil, is called exactly once with the summed declared size of all file items in the listing before staging begins, so a progress sink can show a real denominator (mirrors the block path's stream.SetTotal after HeadVolume).

func DownloadFilesystemVolumeRooted ¶

func DownloadFilesystemVolumeRooted(
	ctx context.Context,
	destination *archive.RootedDestination,
	log *slog.Logger,
	tarPath string,
	stagingDir string,
	filesRootURL string,
	workers int,
	chunkSize int64,
	fetcher *exporter.Fetcher,
	codec compress.Codec,
	setTotal func(total int64),
	onProgress func(n int),
) error

DownloadFilesystemVolumeRooted writes every filesystem artifact through destination.

func FinalizeNode ¶

func FinalizeNode(nodeDir string, node *source.Node) error

FinalizeNode is FinalizeNodeContext with a non-cancellable context.

func FinalizeNodeContext ¶

func FinalizeNodeContext(ctx context.Context, nodeDir string, node *source.Node) error

FinalizeNodeContext computes the node integrity checksum over all current files in nodeDir (manifests/*.yaml, data.bin[.<ext>], data.tar, data/<pvc>.*) and atomically writes <nodeDir>/snapshot.yaml. It must be called after all manifests and volume data for the node are fully written.

The snapshot.yaml Volumes list is populated as follows:

• Nodes that captured their own volume (node.Data != nil): one VolumeInfo from status.data (Variant A, cardinality ≤1) — covers both non-aggregator domain nodes and orphan leaf volume nodes.
• All other nodes (aggregators and manifest-only): Volumes is nil (omitted).

The Volumes field does not affect ComputeNodeChecksum because snapshot.yaml is excluded from the integrity digest.

FinalizeNodeContext is idempotent: each call recomputes the checksum and overwrites snapshot.yaml with the fresh value. The pipeline calls it once per node after both WriteNodeManifests and any volume download have completed.

After snapshot.yaml is durably written, FinalizeNodeContext removes the resume identity marker (archive.NodeIdentityMarkerName). The marker exists only to prove a PARTIAL (snapshot.yaml-less) dir belongs to this snapshot (inv. #9); once snapshot.yaml — the authoritative identity record VerifyNode/ScanNode read — is on disk, the marker is redundant and leaving it would violate the documented final node layout (snapshot.yaml + manifests/ + optional snapshots/ + at most one volume payload). The remove happens strictly AFTER the snapshot.yaml write so a crash at any earlier point still leaves the marker in place and a partial dir always carries exactly one identity record. Removal is checksum-neutral (ComputeNodeChecksum/collectNodeFiles never read the marker), so it cannot perturb the checksum just written or any later VerifyNode.

func FinalizeNodeRootedContext ¶

func FinalizeNodeRootedContext(
	ctx context.Context,
	destination *archive.RootedDestination,
	nodeDir string,
	node *source.Node,
) error

FinalizeNodeRootedContext finalizes a node through destination's locked view.

func FinalizeNodeRootedContextWithChecksum ¶

func FinalizeNodeRootedContextWithChecksum(
	ctx context.Context,
	destination *archive.RootedDestination,
	nodeDir string,
	node *source.Node,
	checksum archive.NodeChecksum,
) error

FinalizeNodeRootedContextWithChecksum finalizes a node through destination's locked view using a checksum computed from that same view after content preparation completed.

func MergeBlockChunks ¶

func MergeBlockChunks(ctx context.Context, chunkDir, outPath string, totalSize, chunkSize int64, ext string) error

MergeBlockChunks assembles all chunk_%05d[.<ext>] files from chunkDir into a single outPath, in strict ascending-index order. ext is the codec extension (e.g. ".zst"); use "" for the none codec.

chunkDir is the absolute path to the directory containing the chunk files. outPath is the absolute destination path for the merged block file.

Pre-conditions (enforced):

• All chunks 0 .. ceil(totalSize/chunkSize)-1 must be present.
• If any chunk is missing, ErrMissingChunk is returned and no output is written.

MERGE STRATEGY is the same for every codec (gzip, lz4, none, and zstd): mergeByConcatenation copies each chunk's already-finalized, independently- encoded frame byte-for-byte into an unpublished AtomicWriter temporary file, in ascending order. Concatenation of independent frames is itself a valid multi-frame stream decodable by stock tools.

HISTORY: a native-zstd-seekable-format merge path (decoding each zstd chunk back to raw bytes and re-encoding it through a seekable.Writer to embed a seek table) was implemented and then reverted; see .agent/tasks.json's notes_on_plan_switch for the full history and the chunk-boundary-skipping redesign now being planned in its place.

Post-conditions on success:

• outPath is a fully durable (fsynced) stream that decodes to exactly totalSize raw bytes (verified; see ErrDecodedLengthMismatch).
• The chunk directory and all its contents are removed.
• No sidecar file is ever written alongside outPath, for any codec: the chunk-offset index sidecar this function used to write is superseded by the zstd path's own embedded seek table (the sidecar machinery itself has since been deleted — nothing produces or consumes it anymore).

chunkSize ≤ 0 falls back to DefaultChunkSize.

A totalSize of 0 is a first-class case: under the frame-concatenation format zero raw bytes are zero frames, i.e. an EMPTY file. It is committed atomically, the (empty or absent) chunkDir is removed, and decoded-length verification is SKIPPED. An empty stream trivially decodes to the 0 wanted raw bytes, and for gzip the empty concatenation of zero frames IS the correct on-disk representation even though kgzip.NewReader rejects an empty stream with EOF (a gzip member requires a header). Without this short-circuit a zero-size gzip volume loops forever: merge → verify-fail → remove → retry. The restore/import decode counterparts rely on this "zero frames == zero bytes" contract, so it must not change without updating them.

ctx is checked once per chunk during the copy loop and on every read during decoded-length verification. Cancellation aborts the in-progress AtomicWriter (so no partial file is ever visible at outPath) and returns a wrapped ctx.Err() (checkable via errors.Is). A hard kill or graceful cancellation never loses data: the chunk directory is only removed after verification and a full successful Commit, so the merge resumes from the same chunks on the next run.

func MergeBlockChunksRooted ¶

func MergeBlockChunksRooted(
	ctx context.Context,
	destination *archive.RootedDestination,
	chunkDir, outPath string,
	totalSize, chunkSize int64,
	ext string,
) error

MergeBlockChunksRooted merges chunks through destination's locked view.

func ScanBlockChunkProgress ¶

func ScanBlockChunkProgress(chunkDir, ext string) (int64, int64, error)

ScanBlockChunkProgress computes durably-committed raw bytes and the raw total byte size recorded for chunkDir's on-disk geometry, purely from local state — no network call, and NO filesystem mutation: it is a pure observation used to seed a progress display before any transfer starts. It mirrors downloadChunk/fetchChunkRaw's own chunk-boundary formula exactly: each already-final chunk contributes its full raw length, and a still-open chunk contributes its TRUSTED ".part" prefix (the offset proven durable by an fsync, capped at that chunk's raw length; never the raw file size alone). Unlike the download path's partialChunkSize, an oversized ".part" is left untouched on disk here — the file gets treated as contributing only its trusted (safe) prefix, not removed or truncated; that cleanup remains the download path's job (see partialChunkSize, used by fetchChunkRaw), which is the only place actually about to act on the chunk's geometry.

It returns (0, 0, nil) when chunkDir carries no trustworthy geometry yet (chunks.meta missing or corrupt) — the same case ensureChunkGeometry treats as "nothing to resume from", so there is nothing safe to seed either.

The pipeline uses this to seed a volume's progress stream with its already-downloaded bytes as soon as the stream is created — well before the DataExport becomes ready or a fresh HEAD confirms totalSize — and keeps the seeded value in place (no reset) once DownloadBlockChunks starts: downloadChunk/fetchChunkRaw's own resume-skip crediting re-derives and re-credits the identical already-committed bytes, and the pipeline wraps that crediting with skipSeededBytes(seeded, ...) so the re-derived bytes are discarded instead of double-counted, rather than dropping the stream to 0 first (see pipeline.seedStreamFromDisk / pipeline.skipSeededBytes). Because this scan never mutates chunkDir between the two calls (no worker has started yet), the two computations always agree exactly for the normal (already-trusted) case; the only divergence is deliberate: because this scan is strictly read-only, an oversized ".part" never disappears out from under a display-only scan. ScanBlockChunkProgress is the non-cancellable compatibility API for callers that have no live context. Cancellable production paths must use ScanBlockChunkProgressContext.

func ScanBlockChunkProgressContext ¶

func ScanBlockChunkProgressContext(ctx context.Context, chunkDir, ext string) (int64, int64, error)

ScanBlockChunkProgressContext is ScanBlockChunkProgress with prompt cancellation.

func ScanBlockChunkProgressRootedContext ¶

func ScanBlockChunkProgressRootedContext(
	ctx context.Context,
	destination *archive.RootedDestination,
	chunkDir, ext string,
) (int64, int64, error)

ScanBlockChunkProgressRootedContext scans progress through destination.

func ScanFSStagingProgress ¶

func ScanFSStagingProgress(ctx context.Context, stagingDir, ext string) (int64, error)

ScanFSStagingProgress computes durably-committed raw bytes across every still-open per-file chunk directory, purely from local state — no network call. Per-file chunk dirs live under the reserved metadata namespace (stagingDir/.d8-meta/chunks/<relPath>/<leaf><ext>.d, per archive.FsFileChunksDirName) so no server-provided path can alias one, so this scans ONLY that subtree. A per-file chunk directory is identified STRICTLY by its name matching archive.FsFileChunksLeafName(ext) — not merely by the presence of a chunks.meta sidecar inside it — and, once matched, its contribution is computed via the identical ScanBlockChunkProgress formula from the same marker createChunkDir writes for both block volumes and per-file FS chunks (see stageChunkedFile, which reuses DownloadBlockChunks/MergeBlockChunks unchanged).

Deliberately excluded:

• The sizes sidecar (also under .d8-meta) carries no chunks.meta, so it is naturally ignored; scanning only the chunks/ subtree makes that explicit.
• A file that is ALREADY fully staged (its chunk directory has already been merged away by MergeBlockChunks into a flat <relPath><ext> blob at the staging root) contributes nothing here, because its original raw declared size is not recoverable from disk once the chunk dir — the only place that size was ever recorded (chunks.meta) — is gone; the merged blob's own on-disk length is a compressed/frame-concatenated size, not the raw size the rest of the progress accounting uses. Such a file keeps being credited exactly once, at its true declared size, by stageCompressedFile's existing resume-skip path once the listing confirms it; the caller must not double-count that credit against this scan (see pipeline.downloadFS, which wraps its onProgress with pipeline.skipSeededBytes(seeded, ...) so that later re-derived credit is discarded instead of double-counted, rather than resetting the stream to 0 before staging begins).
• Two obsolete on-disk shapes are never scanned as leaves, both rejected by the same leaf-name check: the original flat layout written before any relocation (stagingDir/<relPath><ext>.d, entirely outside .d8-meta) is outside the chunks/ subtree this scan even descends into; a directory from the FIRST relocation, named after the file itself (.d8-meta/chunks/<relPath><ext>.d) rather than the fixed leaf name, sits INSIDE the scanned subtree but its name never matches archive.FsFileChunksLeafName(ext), so it is treated as an intermediate directory instead of a trusted leaf even though it may still carry its own chunks.meta. Both cases leave the affected file to simply re-download once, which is acceptable and preferable to crediting bytes for a directory the current download machinery will never merge.

func ScanFSStagingProgressRooted ¶

func ScanFSStagingProgressRooted(
	ctx context.Context,
	destination *archive.RootedDestination,
	stagingDir, ext string,
) (int64, error)

ScanFSStagingProgressRooted scans through destination's locked view.

func ScanFSStagingProgressWithHook ¶

func ScanFSStagingProgressWithHook(
	ctx context.Context,
	stagingDir string,
	ext string,
	hook archive.OpenBoundaryHook,
) (int64, error)

ScanFSStagingProgressWithHook is ScanFSStagingProgress with a deterministic descriptor-boundary hook for adversarial replacement tests.

func ScanFSStagingSizes ¶

func ScanFSStagingSizes(ctx context.Context, stagingDir, ext string) (int64, int64, bool, error)

ScanFSStagingSizes reads the sizes sidecar from stagingDir and, for every file it records, credits its persisted declared size when that file has ALREADY been fully staged as a flat <relPath><ext> blob — i.e. its chunk directory was already merged away by MergeBlockChunks, or it was written whole by stageWholeFile. This is the complement to ScanFSStagingProgress, which by construction can only see STILL-OPEN chunk directories: once a chunk dir is merged away, chunks.meta — the only on-disk record of that file's raw declared size — goes with it, so the sidecar is the only way to credit an already-completed file without a network round-trip.

found is false (with zero totals, no error) when no sidecar exists yet — a from-scratch run, or a staging dir predating this feature — so the caller knows to fall back to the network-driven total/credit path instead of trusting a zero total.

func ScanFSStagingSizesRooted ¶

func ScanFSStagingSizesRooted(
	ctx context.Context,
	destination *archive.RootedDestination,
	stagingDir, ext string,
) (int64, int64, bool, error)

ScanFSStagingSizesRooted scans sidecar sizes through destination.

func WriteNodeManifests ¶

func WriteNodeManifests(ctx context.Context, src source.ManifestSource, nodeDir string, node *source.Node) error

WriteNodeManifests fetches the own-scope manifests for node from src and writes each object as an uncompressed YAML file into <nodeDir>/manifests/ using archive.WriteManifest. Collision fallback (same kind+name but different API group) is handled transparently by WriteManifest.

PersistentVolumeClaims are excluded in two cases:

• Volume-leaf children (node.Children[i].IsVolumeLeaf()): the captured PVC manifest belongs in each leaf node's own manifests/ directory.
• The node's own captured volume (node.Data.SourceRef): the PVC data is already captured in the volume payload (data.bin[.<ext>] or data.tar); the PVC identity is recorded in snapshot.yaml Volumes[].Target.

Matching is by metadata.uid first; if the uid is absent in the captured manifest, it falls back to metadata.name.

The manifests are fetched from the node's own manifests-download subresource.

The operation is idempotent: rewriting an already-present object with the same kind, name, and API group is a no-op.

func WriteNodeManifestsRooted ¶

func WriteNodeManifestsRooted(
	ctx context.Context,
	destination *archive.RootedDestination,
	src source.ManifestSource,
	nodeDir string,
	node *source.Node,
) error

WriteNodeManifestsRooted writes node manifests through destination.

func WriteTar ¶

func WriteTar(ctx context.Context, outputPath string, stagingDir string, entries TarEntrySource) error

WriteTar writes a deterministic plain uncompressed PAX tar to outputPath. entries must emit RelPath-sorted entries; the filesystem inventory pipeline establishes that order with bounded on-disk merge sorting before staging. WriteTar never clones or retains the entry stream. Raw bytes for "file" entries are read from filepath.Join(stagingDir, filepath.FromSlash(entry.RelPath)). The output file is written atomically (.tmp → fsync → rename).

ctx is checked before every entry, around every fixed-size file read, and after temporary-file sync/close at AtomicWriter's pre-publication checkpoint. Cancellation observed before publication aborts the temporary output; after publication begins, rename/durability errors determine the result. A parent-directory sync error retains archive.PublicationPublished through wrapping. The staging directory is always untouched, so assembly or durability confirmation can be retried.

func WriteTarRooted ¶

func WriteTarRooted(
	ctx context.Context,
	destination *archive.RootedDestination,
	outputPath string,
	stagingDir string,
	entries TarEntrySource,
) error

WriteTarRooted assembles a tar through destination's locked view.

func WriteVolumeManifest ¶

func WriteVolumeManifest(ctx context.Context, src source.ManifestSource, volumeDir string, volNode *source.Node) error

WriteVolumeManifest fetches the manifests in the volume node's scope via src and writes the single PVC that corresponds to the volume node's captured source into <volumeDir>/manifests/persistentvolumeclaim_<name>.yaml.

For orphan leaf volume nodes the captured PVC manifest lives in the parent aggregator node's own manifests, so the scope ref is the parent ref (see source.Node.ManifestScopeRef).

The target PVC is matched by metadata.uid first (when both the node's captured source uid and the captured object's uid are non-empty); otherwise by metadata.name.

Returns an error if the target PVC is not present in the fetched manifests.

func WriteVolumeManifestRooted ¶

func WriteVolumeManifestRooted(
	ctx context.Context,
	destination *archive.RootedDestination,
	src source.ManifestSource,
	volumeDir string,
	volNode *source.Node,
) error

WriteVolumeManifestRooted writes one volume manifest through destination.