diff --git a/internal/snapshot/archive/fsmetadata.go b/internal/snapshot/archive/fsmetadata.go index ac6895e3..cec3df59 100644 --- a/internal/snapshot/archive/fsmetadata.go +++ b/internal/snapshot/archive/fsmetadata.go @@ -18,8 +18,11 @@ package archive import ( "archive/tar" + "context" "errors" "fmt" + "io" + "math" "strconv" "strings" ) @@ -166,6 +169,59 @@ func (m FSMetadata) validate() error { return nil } +// SumTarRawSizes returns the sum of RawSize across every regular-file entry in the tar +// stream r, read via header-only seeks (tar.Reader skips each entry's body with Seek when +// r implements io.Seeker, never copying payload bytes). It is used to compute the exact +// byte total a filesystem volume's data.tar will produce on import, without a second full +// decode pass. ctx is checked once per entry so a large tar stays cancellable. +func SumTarRawSizes(ctx context.Context, r io.Reader) (int64, error) { + tr := tar.NewReader(r) + + var total int64 + + for { + if err := ctx.Err(); err != nil { + return 0, err + } + + hdr, err := tr.Next() + if errors.Is(err, io.EOF) { + break + } + + if err != nil { + return 0, fmt.Errorf("read tar entry: %w", err) + } + + if hdr.Typeflag != tar.TypeReg && hdr.Typeflag != 0 { + continue + } + + metadata, err := ParseFSMetadata(hdr) + if err != nil { + return 0, fmt.Errorf("entry %q: %w", hdr.Name, err) + } + + total, err = addRawSize(total, metadata.RawSize) + if err != nil { + return 0, fmt.Errorf("entry %q: %w", hdr.Name, err) + } + } + + return total, nil +} + +// addRawSize adds size to total, failing rather than silently wrapping when the sum would +// overflow int64 — the same overflow-safety contract snapimport's own raw-size accounting +// (addRawSize) applies to its running upload total. +func addRawSize(total, size int64) (int64, error) { + if size > math.MaxInt64-total { + return 0, fmt.Errorf("raw-size total overflows int64") + } + + return total + size, nil +} + func validateFSOriginalPath(originalPath string) error { if originalPath == "" { return fmt.Errorf("%w: original path is empty", ErrInvalidFSMetadata) diff --git a/internal/snapshot/archive/fsmetadata_test.go b/internal/snapshot/archive/fsmetadata_test.go index 9602d3fc..38632f11 100644 --- a/internal/snapshot/archive/fsmetadata_test.go +++ b/internal/snapshot/archive/fsmetadata_test.go @@ -19,6 +19,7 @@ package archive import ( "archive/tar" "bytes" + "context" "errors" "io" "os" @@ -167,6 +168,152 @@ func TestComputeNodeChecksum_CoversFSPAXMetadata(t *testing.T) { } } +// TestSumTarRawSizes covers SumTarRawSizes: sums only regular-entry PAX raw sizes, ignores +// directory/symlink entries, and propagates ParseFSMetadata's fail-closed errors. +func TestSumTarRawSizes(t *testing.T) { + t.Parallel() + + t.Run("success: empty tar sums to zero", func(t *testing.T) { + t.Parallel() + + var buf bytes.Buffer + + tw := tar.NewWriter(&buf) + if err := tw.Close(); err != nil { + t.Fatalf("close tar writer: %v", err) + } + + total, err := SumTarRawSizes(context.Background(), bytes.NewReader(buf.Bytes())) + if err != nil { + t.Fatalf("SumTarRawSizes: %v", err) + } + + if total != 0 { + t.Errorf("total = %d, want 0", total) + } + }) + + t.Run("success: sums regular entries, ignores directory and symlink entries", func(t *testing.T) { + t.Parallel() + + var buf bytes.Buffer + + tw := tar.NewWriter(&buf) + + if err := tw.WriteHeader(&tar.Header{Typeflag: tar.TypeDir, Name: "dir/", Mode: 0o755}); err != nil { + t.Fatalf("write directory header: %v", err) + } + + if err := tw.WriteHeader(&tar.Header{Typeflag: tar.TypeSymlink, Name: "link", Linkname: "dir/first", Mode: 0o777}); err != nil { + t.Fatalf("write symlink header: %v", err) + } + + writeRegularPAXEntry(t, tw, "dir/first", "none", 10) + writeRegularPAXEntry(t, tw, "second", "zstd", 25) + + if err := tw.Close(); err != nil { + t.Fatalf("close tar writer: %v", err) + } + + total, err := SumTarRawSizes(context.Background(), bytes.NewReader(buf.Bytes())) + if err != nil { + t.Fatalf("SumTarRawSizes: %v", err) + } + + if want := int64(10 + 25); total != want { + t.Errorf("total = %d, want %d", total, want) + } + }) + + t.Run("error: regular entry missing required PAX metadata", func(t *testing.T) { + t.Parallel() + + var buf bytes.Buffer + + tw := tar.NewWriter(&buf) + + // A regular entry with no PAX records at all: ParseFSMetadata must reject it, + // and SumTarRawSizes must propagate that failure rather than skip the entry. + if err := tw.WriteHeader(&tar.Header{ + Typeflag: tar.TypeReg, + Name: "broken.txt", + Mode: 0o600, + Size: 3, + }); err != nil { + t.Fatalf("write header: %v", err) + } + + if _, err := io.WriteString(tw, "abc"); err != nil { + t.Fatalf("write body: %v", err) + } + + if err := tw.Close(); err != nil { + t.Fatalf("close tar writer: %v", err) + } + + _, err := SumTarRawSizes(context.Background(), bytes.NewReader(buf.Bytes())) + if !errors.Is(err, ErrInvalidFSMetadata) { + t.Fatalf("SumTarRawSizes error = %v, want wrapping ErrInvalidFSMetadata", err) + } + }) + + t.Run("error: context canceled before completion", func(t *testing.T) { + t.Parallel() + + var buf bytes.Buffer + + tw := tar.NewWriter(&buf) + writeRegularPAXEntry(t, tw, "first", "none", 5) + + if err := tw.Close(); err != nil { + t.Fatalf("close tar writer: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := SumTarRawSizes(ctx, bytes.NewReader(buf.Bytes())) + if !errors.Is(err, context.Canceled) { + t.Fatalf("SumTarRawSizes error = %v, want context.Canceled", err) + } + }) +} + +// writeRegularPAXEntry writes one well-formed regular PAX entry of rawSize plaintext bytes. +// Body is always written raw (codec name is just metadata) so stored size always == rawSize. +func writeRegularPAXEntry(t *testing.T, tw *tar.Writer, originalPath, codec string, rawSize int64) { + t.Helper() + + metadata, err := NewFSMetadata(codec, originalPath, rawSize) + if err != nil { + t.Fatalf("NewFSMetadata: %v", err) + } + + storedPath, err := metadata.StoredPath() + if err != nil { + t.Fatalf("StoredPath: %v", err) + } + + hdr := &tar.Header{ + Format: tar.FormatPAX, + Typeflag: tar.TypeReg, + Name: storedPath, + Mode: 0o600, + Size: rawSize, + PAXRecords: metadata.PAXRecords(), + } + + // SumTarRawSizes never decodes the body, so any bytes of the declared Size work here; + // content correctness for a given codec is exercised elsewhere (fsmetadata round trip). + if err := tw.WriteHeader(hdr); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + + if _, err := tw.Write(make([]byte, rawSize)); err != nil { + t.Fatalf("Write: %v", err) + } +} + func writeTestMetadataTar(t *testing.T, path string, rawSize int64) { t.Helper() diff --git a/internal/snapshot/archive/snapshot_yaml.go b/internal/snapshot/archive/snapshot_yaml.go index b928ec16..93a6abfd 100644 --- a/internal/snapshot/archive/snapshot_yaml.go +++ b/internal/snapshot/archive/snapshot_yaml.go @@ -40,9 +40,14 @@ const ( // SnapshotFormatVersionLegacy identifies archives written before explicit envelope versioning. // Version zero is accepted only through an explicit unauthenticated compatibility option. SnapshotFormatVersionLegacy = 0 - // SnapshotFormatVersionCurrent is written by every snapshot.yaml marshal. Version 2 adds - // the mandatory authenticated direct-child commitment (ChildrenChecksum). - SnapshotFormatVersionCurrent = 2 + // SnapshotFormatVersionAuthenticatedChildren adds the mandatory ChildrenChecksum. An archive + // at this version has no RawSizeBytes/StoredSizeBytes; readers measure the payload instead. + SnapshotFormatVersionAuthenticatedChildren = 2 + // SnapshotFormatVersionPayloadSizes adds VolumeInfo.RawSizeBytes/StoredSizeBytes, the + // measured on-disk payload footprint (as opposed to Size's nominal restoreSize). + SnapshotFormatVersionPayloadSizes = 3 + // SnapshotFormatVersionCurrent is written by every snapshot.yaml marshal. + SnapshotFormatVersionCurrent = SnapshotFormatVersionPayloadSizes ) // sha256HexLen is the length of a hex-encoded SHA-256 digest (32 bytes → 64 hex chars). @@ -211,7 +216,7 @@ func validateSnapshotEnvelope(sy SnapshotYAML, options SnapshotYAMLReadOptions) } return validateChildrenChecksumPresent(sy) - case SnapshotFormatVersionCurrent: + case SnapshotFormatVersionAuthenticatedChildren, SnapshotFormatVersionCurrent: default: return fmt.Errorf("%d: %w", sy.FormatVersion, ErrUnsupportedSnapshotFormat) } @@ -319,10 +324,20 @@ type VolumeInfo struct { // StorageClassName records the source StorageClass of the captured volume. On re-import it // is sent as the PopulateData DataImport's spec.storageParams.storageClassName (required). StorageClassName string `json:"storageClassName,omitempty"` - // Size records the real allocated size of the captured volume (e.g. "10Gi"), taken from - // VolumeSnapshotContent.status.restoreSize. On re-import it is sent as the PopulateData - // DataImport's spec.storageParams.size (required). + // Size records the NOMINAL captured-volume quantity (e.g. "10Gi"), taken from + // VolumeSnapshotContent.status.restoreSize. Used only to size the scratch volume on + // re-import (spec.storageParams.size, required) — NOT the payload's real byte size, since + // a thin-provisioning backend can round the device up from it. See RawSizeBytes/StoredSizeBytes. Size string `json:"size,omitempty"` + // RawSizeBytes is the exact decoded byte count of the captured payload: for Block, the + // decoded length of data.bin[.]; for Filesystem, the sum of data.tar's regular-entry + // raw sizes. Unlike Size, this can exceed the nominal PVC quantity (thin-provisioning + // round-up). Recorded from SnapshotFormatVersionPayloadSizes onward; zero on an older + // archive means "not recorded", not "empty". + RawSizeBytes int64 `json:"rawSizeBytes,omitempty"` + // StoredSizeBytes is the on-disk size of the payload artifact (data.bin[.] or + // data.tar). Informational plus a corruption preflight; same presence rules as RawSizeBytes. + StoredSizeBytes int64 `json:"storedSizeBytes,omitempty"` } // NodeChecksum is a locally-computed integrity digest. SnapshotYAML.Checksum covers the node's diff --git a/internal/snapshot/archive/snapshot_yaml_test.go b/internal/snapshot/archive/snapshot_yaml_test.go index 394310ca..cae885c0 100644 --- a/internal/snapshot/archive/snapshot_yaml_test.go +++ b/internal/snapshot/archive/snapshot_yaml_test.go @@ -18,6 +18,8 @@ package archive_test import ( "context" + "crypto/sha256" + "encoding/json" "errors" "fmt" "io" @@ -1251,3 +1253,259 @@ func TestSnapshotYAML_ChecksumUnaffectedByVolumeField(t *testing.T) { t.Errorf("VerifyNode must pass after adding Volume field: %v", err) } } + +// TestSnapshotYAML_RoundTripV3 verifies that a fresh write stamps +// SnapshotFormatVersionCurrent (3) and that the new payload-size fields round-trip. +func TestSnapshotYAML_RoundTripV3(t *testing.T) { + t.Parallel() + + dir := makeSnapshotNodeDir(t) + + checksum, err := archive.ComputeNodeChecksum(dir) + if err != nil { + t.Fatalf("ComputeNodeChecksum: %v", err) + } + + want := archive.SnapshotYAML{ + APIVersion: "snapshot.storage.k8s.io/v1", + Kind: "VolumeSnapshot", + Name: "d8-ss-v3", + Checksum: checksum, + Volumes: []archive.VolumeInfo{{ + Target: archive.VolumeObjectRef{APIVersion: "v1", Kind: "PersistentVolumeClaim", Name: "pvc-v3"}, + Artifact: archive.VolumeObjectRef{APIVersion: "snapshot.storage.k8s.io/v1", Kind: "VolumeSnapshotContent", Name: "vsc-v3"}, + VolumeMode: archive.VolumeModeBlock, + StorageClassName: "sc-thin", + Size: "1Gi", + RawSizeBytes: 1077665792, + StoredSizeBytes: 900000000, + }}, + } + + if err := archive.WriteSnapshotYAML(dir, want); err != nil { + t.Fatalf("WriteSnapshotYAML: %v", err) + } + + got, err := archive.ReadSnapshotYAML(dir) + if err != nil { + t.Fatalf("ReadSnapshotYAML: %v", err) + } + + if got.FormatVersion != archive.SnapshotFormatVersionCurrent { + t.Errorf("FormatVersion = %d, want %d (SnapshotFormatVersionCurrent)", got.FormatVersion, archive.SnapshotFormatVersionCurrent) + } + + if got.MetadataChecksum == nil { + t.Fatal("MetadataChecksum must be set on a current-format write") + } + + if err := validateChecksumFieldTest(*got.MetadataChecksum); err != nil { + t.Errorf("MetadataChecksum invalid: %v", err) + } + + if len(got.Volumes) != 1 { + t.Fatalf("Volumes length: got %d, want 1", len(got.Volumes)) + } + + gotVol := got.Volumes[0] + if gotVol.RawSizeBytes != 1077665792 { + t.Errorf("RawSizeBytes = %d, want 1077665792", gotVol.RawSizeBytes) + } + + if gotVol.StoredSizeBytes != 900000000 { + t.Errorf("StoredSizeBytes = %d, want 900000000", gotVol.StoredSizeBytes) + } +} + +// validateChecksumFieldTest mirrors validateChecksum's invariants (algorithm/hex +// length/short consistency) without depending on unexported archive internals. +func validateChecksumFieldTest(c archive.NodeChecksum) error { + if c.Algorithm != archive.ChecksumAlgorithmSHA256 { + return fmt.Errorf("algorithm = %q, want %q", c.Algorithm, archive.ChecksumAlgorithmSHA256) + } + + if len(c.Hex) != 64 { + return fmt.Errorf("hex length = %d, want 64", len(c.Hex)) + } + + if c.Short != archive.ShortChecksum(c.Hex) { + return fmt.Errorf("short = %q, inconsistent with hex", c.Short) + } + + return nil +} + +// TestSnapshotYAML_RejectsUnsupportedVersion proves a format version beyond +// SnapshotFormatVersionCurrent is rejected outright, even with AllowUnauthenticatedLegacy. +func TestSnapshotYAML_RejectsUnsupportedVersion(t *testing.T) { + t.Parallel() + + dir := makeSnapshotNodeDir(t) + + future := archive.SnapshotFormatVersionCurrent + 1 + data := fmt.Appendf(nil, + "formatVersion: %d\napiVersion: snapshot.example.io/v1\nkind: Snapshot\nname: future\n"+ + "childrenChecksum: {algorithm: %s, hex: %q, short: %q}\n", + future, + archive.EmptyChildrenChecksum().Algorithm, + archive.EmptyChildrenChecksum().Hex, + archive.EmptyChildrenChecksum().Short) + + if err := os.WriteFile(filepath.Join(dir, archive.SnapshotYAMLName), data, 0o600); err != nil { + t.Fatalf("write future snapshot.yaml: %v", err) + } + + if _, err := archive.ReadSnapshotYAML(dir); !errors.Is(err, archive.ErrUnsupportedSnapshotFormat) { + t.Fatalf("ReadSnapshotYAML error = %v, want ErrUnsupportedSnapshotFormat", err) + } + + if _, err := archive.ReadSnapshotYAMLWithOptions(dir, archive.SnapshotYAMLReadOptions{AllowUnauthenticatedLegacy: true}); !errors.Is(err, archive.ErrUnsupportedSnapshotFormat) { + t.Fatalf("ReadSnapshotYAMLWithOptions(AllowUnauthenticatedLegacy) error = %v, want ErrUnsupportedSnapshotFormat "+ + "(the legacy opt-in must not widen acceptance of an unknown future major version)", err) + } +} + +// TestVolumeInfo_ZeroSizesOmittedFromJSON proves RawSizeBytes/StoredSizeBytes at their zero +// value are omitted from JSON entirely (omitempty), not serialized as 0 — required for a v2 +// archive's MetadataChecksum, computed over JSON that never had these keys, to stay stable. +func TestVolumeInfo_ZeroSizesOmittedFromJSON(t *testing.T) { + t.Parallel() + + vol := archive.VolumeInfo{ + Target: archive.VolumeObjectRef{APIVersion: "v1", Kind: "PersistentVolumeClaim", Name: "pvc"}, + Artifact: archive.VolumeObjectRef{APIVersion: "snapshot.storage.k8s.io/v1", Kind: "VolumeSnapshotContent", Name: "vsc"}, + VolumeMode: archive.VolumeModeBlock, + StorageClassName: "sc", + Size: "1Gi", + // RawSizeBytes/StoredSizeBytes deliberately left at their zero value. + } + + data, err := json.Marshal(vol) + if err != nil { + t.Fatalf("json.Marshal: %v", err) + } + + for _, key := range []string{"rawSizeBytes", "storedSizeBytes"} { + if strings.Contains(string(data), key) { + t.Errorf("marshaled VolumeInfo contains key %q with a zero value; want it omitted entirely (omitempty): %s", key, data) + } + } + + // Sanity: a non-zero value IS present. + vol.RawSizeBytes = 42 + data, err = json.Marshal(vol) + if err != nil { + t.Fatalf("json.Marshal: %v", err) + } + + if !strings.Contains(string(data), `"rawSizeBytes":42`) { + t.Errorf("marshaled VolumeInfo with RawSizeBytes=42 must contain the key: %s", data) + } +} + +// snapshotYAMLShadowV2 mirrors the private snapshotYAMLWire shape field-for-field, but +// without rawSizeBytes/storedSizeBytes — exactly what a real pre-fix v2 archive produced. +// Needed because SnapshotYAML.MarshalJSON always stamps the current version, so it can't +// itself produce v2 bytes. +type snapshotYAMLShadowV2 struct { + FormatVersion int `json:"formatVersion,omitempty"` + APIVersion string `json:"apiVersion"` + Kind string `json:"kind"` + Name string `json:"name"` + Namespace string `json:"namespace,omitempty"` + UID string `json:"uid,omitempty"` + SourceName string `json:"sourceName,omitempty"` + SourceObjectRef *archive.SourceObjectRef `json:"sourceObjectRef,omitempty"` + Checksum archive.NodeChecksum `json:"checksum"` + ChildrenChecksum *archive.NodeChecksum `json:"childrenChecksum,omitempty"` + MetadataChecksum *archive.NodeChecksum `json:"metadataChecksum,omitempty"` + Volumes []archive.VolumeInfo `json:"volumes,omitempty"` +} + +// TestSnapshotYAML_V2ArchiveStillValidates is the decisive compatibility test for this fix: +// it reproduces, byte for byte, a REAL pre-fix archive (formatVersion 2, no +// rawSizeBytes/storedSizeBytes keys in its JSON, metadataChecksum computed over exactly that +// canonical JSON) and proves it still validates. +// +// Without `omitempty` on RawSizeBytes/StoredSizeBytes, computeSnapshotMetadataChecksum would +// now serialize them as 0 even for this old document, disagreeing with the checksum the +// archive carries — failing every v2 archive in the wild with +// ErrSnapshotMetadataChecksumMismatch. This test proves that did not happen. +func TestSnapshotYAML_V2ArchiveStillValidates(t *testing.T) { + t.Parallel() + + childrenChecksum := archive.EmptyChildrenChecksum() + + shadow := snapshotYAMLShadowV2{ + FormatVersion: archive.SnapshotFormatVersionAuthenticatedChildren, + APIVersion: "snapshot.storage.k8s.io/v1", + Kind: "VolumeSnapshot", + Name: "pre-fix-v2-node", + Namespace: "ns-legacy", + Checksum: validChecksum(), + ChildrenChecksum: &childrenChecksum, + Volumes: []archive.VolumeInfo{{ + Target: archive.VolumeObjectRef{APIVersion: "v1", Kind: "PersistentVolumeClaim", Name: "pvc-legacy"}, + Artifact: archive.VolumeObjectRef{APIVersion: "snapshot.storage.k8s.io/v1", Kind: "VolumeSnapshotContent", Name: "vsc-legacy"}, + VolumeMode: archive.VolumeModeBlock, + StorageClassName: "sc-legacy", + Size: "1Gi", + // RawSizeBytes/StoredSizeBytes intentionally absent: the pre-fix wire type has no + // such fields — this is byte-for-byte what a real pre-fix archive wrote to disk. + }}, + } + + // canonicalWithoutChecksum is exactly what computeSnapshotMetadataChecksum hashes: + // the canonical envelope with metadataChecksum itself absent/nil. + canonicalWithoutChecksum, err := json.Marshal(shadow) + if err != nil { + t.Fatalf("json.Marshal shadow envelope: %v", err) + } + + for _, key := range []string{"rawSizeBytes", "storedSizeBytes"} { + if strings.Contains(string(canonicalWithoutChecksum), key) { + t.Fatalf("test fixture bug: canonical v2 bytes must not contain %q: %s", key, canonicalWithoutChecksum) + } + } + + sum := sha256.Sum256(canonicalWithoutChecksum) + hexDigest := fmt.Sprintf("%x", sum) + + metadataChecksum := archive.NodeChecksum{ + Algorithm: archive.ChecksumAlgorithmSHA256, + Hex: hexDigest, + Short: archive.ShortChecksum(hexDigest), + } + shadow.MetadataChecksum = &metadataChecksum + + final, err := json.Marshal(shadow) + if err != nil { + t.Fatalf("json.Marshal final shadow envelope: %v", err) + } + + dir := t.TempDir() + if err := os.WriteFile(filepath.Join(dir, archive.SnapshotYAMLName), final, 0o600); err != nil { + t.Fatalf("write v2 snapshot.yaml fixture: %v", err) + } + + got, err := archive.ReadSnapshotYAML(dir) + if err != nil { + t.Fatalf("CRITICAL: a real pre-fix v2 archive failed to validate after this fix: %v\n"+ + "this means RawSizeBytes/StoredSizeBytes are missing `omitempty` (or some other change "+ + "altered the canonical JSON shape), and EVERY v2 archive in the wild would now fail "+ + "ErrSnapshotMetadataChecksumMismatch on read", err) + } + + if got.FormatVersion != archive.SnapshotFormatVersionAuthenticatedChildren { + t.Errorf("FormatVersion = %d, want %d", got.FormatVersion, archive.SnapshotFormatVersionAuthenticatedChildren) + } + + if len(got.Volumes) != 1 { + t.Fatalf("Volumes length: got %d, want 1", len(got.Volumes)) + } + + if got.Volumes[0].RawSizeBytes != 0 || got.Volumes[0].StoredSizeBytes != 0 { + t.Errorf("a v2 archive's Volumes[0] payload sizes must read back as zero (never recorded), got RawSizeBytes=%d StoredSizeBytes=%d", + got.Volumes[0].RawSizeBytes, got.Volumes[0].StoredSizeBytes) + } +} diff --git a/internal/snapshot/compress/decode.go b/internal/snapshot/compress/decode.go index aa39d5b4..1025fbd4 100644 --- a/internal/snapshot/compress/decode.go +++ b/internal/snapshot/compress/decode.go @@ -18,6 +18,7 @@ package compress import ( "bufio" + "context" "errors" "fmt" "io" @@ -27,6 +28,10 @@ import ( "github.com/pierrec/lz4/v4" ) +// decodedSizeStreamBufferBytes bounds streamDecodedSize's read buffer (matches +// volume.copyBufferSize elsewhere in the snapshot packages). +const decodedSizeStreamBufferBytes = 32 << 10 + // NewReader returns a streaming decompressing io.ReadCloser for src, selecting // the codec by ext (the file-extension convention used by Codec.Ext: ".zst", // ".gz", ".lz4", or "" for no compression). It is the decode-side counterpart @@ -151,3 +156,107 @@ func (z *zstdReadCloser) Close() error { return nil } + +// DecodedSize returns the exact decoded byte length of source for codec ext. zstd reads it +// from frame headers (no payload decode); "" is just the stream length; gzip/lz4 have no +// decoded-size metadata, so they require a full streaming decode. source's position is restored. +func DecodedSize(ctx context.Context, ext string, source io.ReadSeeker) (int64, error) { + if err := ctx.Err(); err != nil { + return 0, err + } + + switch ext { + case "": + size, err := rawStreamSize(source) + if err != nil { + return 0, fmt.Errorf("determine raw stream size: %w", err) + } + + return size, nil + case ".zst": + size, err := ZstdDecodedSize(source) + if err != nil { + return 0, fmt.Errorf("determine zstd decoded size: %w", err) + } + + return size, nil + default: + size, err := streamDecodedSize(ctx, ext, source) + if err != nil { + return 0, fmt.Errorf("determine %s decoded size: %w", ext, err) + } + + return size, nil + } +} + +// rawStreamSize returns the byte length remaining in source from its current position +// through EOF, restoring that position before return. +func rawStreamSize(source io.ReadSeeker) (int64, error) { + start, err := source.Seek(0, io.SeekCurrent) + if err != nil { + return 0, fmt.Errorf("query current offset: %w", err) + } + + end, err := source.Seek(0, io.SeekEnd) + if err != nil { + return 0, fmt.Errorf("query end offset: %w", err) + } + + if _, err := source.Seek(start, io.SeekStart); err != nil { + return 0, fmt.Errorf("restore offset: %w", err) + } + + return end - start, nil +} + +// streamDecodedSize measures a non-zstd codec's decoded length by decoding the whole stream +// and counting bytes (gzip/lz4 have no decoded-size header). Cancellable; position is restored. +func streamDecodedSize(ctx context.Context, ext string, source io.ReadSeeker) (int64, error) { + start, err := source.Seek(0, io.SeekCurrent) + if err != nil { + return 0, fmt.Errorf("query current offset: %w", err) + } + + reader, err := NewReader(ext, source) + if err != nil { + return 0, err + } + + var total int64 + + buf := make([]byte, decodedSizeStreamBufferBytes) + + for { + if err := ctx.Err(); err != nil { + _ = reader.Close() + + return 0, err + } + + n, readErr := reader.Read(buf) + total += int64(n) + + if readErr != nil { + if errors.Is(readErr, io.EOF) { + break + } + + _ = reader.Close() + + return 0, fmt.Errorf("decode stream: %w", readErr) + } + } + + closeErr := reader.Close() + + if _, seekErr := source.Seek(start, io.SeekStart); seekErr != nil { + return 0, errors.Join(closeErr, fmt.Errorf("restore offset: %w", seekErr)) + } + + if closeErr != nil { + return 0, fmt.Errorf("close decoder: %w", closeErr) + } + + return total, nil +} diff --git a/internal/snapshot/compress/decode_test.go b/internal/snapshot/compress/decode_test.go index 68666d75..8f4f16bb 100644 --- a/internal/snapshot/compress/decode_test.go +++ b/internal/snapshot/compress/decode_test.go @@ -18,6 +18,7 @@ package compress_test import ( "bytes" + "context" "errors" "io" "testing" @@ -278,3 +279,215 @@ func TestNewReader_LZ4TruncatedFrameErrors(t *testing.T) { t.Errorf("truncated input must not be reported as a clean io.EOF: %v", err) } } + +// TestDecodedSize covers compress.DecodedSize across every codec, including multi-frame +// concatenation, error paths (corrupt/truncated zstd, cancellation), and position restoration. +func TestDecodedSize(t *testing.T) { + t.Parallel() + + t.Run("success: empty ext returns the raw stream length", func(t *testing.T) { + t.Parallel() + + data := []byte("raw uncompressed block bytes") + source := bytes.NewReader(data) + + got, err := compress.DecodedSize(context.Background(), "", source) + if err != nil { + t.Fatalf("DecodedSize: %v", err) + } + + if got != int64(len(data)) { + t.Errorf("size = %d, want %d", got, len(data)) + } + }) + + t.Run("success: empty payload of every codec decodes to zero", func(t *testing.T) { + t.Parallel() + + for _, tc := range decodeCases { + if tc.ext == ".zst" { + // zstd can't represent an empty payload as a frame at all (EncodeStream/ + // EncodeFrame produce zero bytes for empty input), so ZstdDecodedSize + // rejects it as "no frames" — exercised as an error case further down. + continue + } + + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + var frame []byte + + if tc.ext != "" { + codec, err := compress.New(tc.codecName, 0) + if err != nil { + t.Fatalf("compress.New(%q): %v", tc.codecName, err) + } + + var buf bytes.Buffer + if err := codec.EncodeStream(&buf, bytes.NewReader(nil)); err != nil { + t.Fatalf("EncodeStream(empty): %v", err) + } + + frame = buf.Bytes() + } + + got, err := compress.DecodedSize(context.Background(), tc.ext, bytes.NewReader(frame)) + if err != nil { + t.Fatalf("DecodedSize: %v", err) + } + + if got != 0 { + t.Errorf("size = %d, want 0", got) + } + }) + } + }) + + t.Run("error: zstd stream with no frames at all (empty payload)", func(t *testing.T) { + t.Parallel() + + _, err := compress.DecodedSize(context.Background(), ".zst", bytes.NewReader(nil)) + if err == nil { + t.Fatal("expected an error for a zero-byte zstd stream (no frames to prove a size from), got nil") + } + }) + + t.Run("success: multi-frame concatenated stream (block-chunk shape)", func(t *testing.T) { + t.Parallel() + + for _, tc := range decodeCases { + if tc.ext == "" { + continue // "" carries no per-chunk framing; covered by the raw-length case above. + } + + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + codec, err := compress.New(tc.codecName, 0) + if err != nil { + t.Fatalf("compress.New(%q): %v", tc.codecName, err) + } + + chunks := [][]byte{ + bytes.Repeat([]byte("chunk-one-"), 50), + bytes.Repeat([]byte("chunk-two-"), 30), + bytes.Repeat([]byte("chunk-three-"), 10), + } + + var ( + frames bytes.Buffer + total int64 + ) + + for _, chunk := range chunks { + if err := codec.EncodeFrameStream(&frames, bytes.NewReader(chunk), int64(len(chunk))); err != nil { + t.Fatalf("EncodeFrameStream: %v", err) + } + + total += int64(len(chunk)) + } + + source := bytes.NewReader(frames.Bytes()) + + got, err := compress.DecodedSize(context.Background(), tc.ext, source) + if err != nil { + t.Fatalf("DecodedSize: %v", err) + } + + if got != total { + t.Errorf("size = %d, want %d", got, total) + } + + // Position must be restored to where it started (beginning, here). + if pos, seekErr := source.Seek(0, io.SeekCurrent); seekErr != nil || pos != 0 { + t.Errorf("source position after DecodedSize = %d (err=%v), want 0 (position restored)", pos, seekErr) + } + }) + } + }) + + t.Run("success: position is restored from a non-zero starting offset", func(t *testing.T) { + t.Parallel() + + prefix := []byte("prefix-bytes-not-part-of-the-payload") + payload := []byte("the actual raw payload bytes") + + var buf bytes.Buffer + buf.Write(prefix) + buf.Write(payload) + + source := bytes.NewReader(buf.Bytes()) + if _, err := source.Seek(int64(len(prefix)), io.SeekStart); err != nil { + t.Fatalf("Seek: %v", err) + } + + got, err := compress.DecodedSize(context.Background(), "", source) + if err != nil { + t.Fatalf("DecodedSize: %v", err) + } + + if got != int64(len(payload)) { + t.Errorf("size = %d, want %d", got, len(payload)) + } + + if pos, seekErr := source.Seek(0, io.SeekCurrent); seekErr != nil || pos != int64(len(prefix)) { + t.Errorf("source position after DecodedSize = %d (err=%v), want %d (restored)", pos, seekErr, len(prefix)) + } + }) + + t.Run("error: truncated/corrupt zstd stream", func(t *testing.T) { + t.Parallel() + + codec, err := compress.New("zstd", 0) + if err != nil { + t.Fatalf("compress.New(zstd): %v", err) + } + + payload := bytes.Repeat([]byte("zstd payload bytes for truncation "), 100) + + var buf bytes.Buffer + if err := codec.EncodeFrameStream(&buf, bytes.NewReader(payload), int64(len(payload))); err != nil { + t.Fatalf("EncodeFrameStream: %v", err) + } + + truncated := buf.Bytes()[:len(buf.Bytes())/2] + + _, err = compress.DecodedSize(context.Background(), ".zst", bytes.NewReader(truncated)) + if err == nil { + t.Fatal("expected an error for a truncated zstd stream, got nil") + } + }) + + t.Run("error: unknown extension", func(t *testing.T) { + t.Parallel() + + _, err := compress.DecodedSize(context.Background(), ".xz", bytes.NewReader(nil)) + if !errors.Is(err, compress.ErrUnknownCodec) { + t.Fatalf("expected ErrUnknownCodec, got: %v", err) + } + }) + + t.Run("error: context canceled before any work (non-zstd codec, full decode path)", func(t *testing.T) { + t.Parallel() + + codec, err := compress.New("gzip", 0) + if err != nil { + t.Fatalf("compress.New(gzip): %v", err) + } + + payload := bytes.Repeat([]byte("gzip payload for cancellation "), 2000) + + var buf bytes.Buffer + if err := codec.EncodeFrameStream(&buf, bytes.NewReader(payload), int64(len(payload))); err != nil { + t.Fatalf("EncodeFrameStream: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err = compress.DecodedSize(ctx, ".gz", bytes.NewReader(buf.Bytes())) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got: %v", err) + } + }) +} diff --git a/internal/snapshot/pipeline/pipeline_test.go b/internal/snapshot/pipeline/pipeline_test.go index 8fa287ed..ae3bc38d 100644 --- a/internal/snapshot/pipeline/pipeline_test.go +++ b/internal/snapshot/pipeline/pipeline_test.go @@ -17,6 +17,7 @@ limitations under the License. package pipeline_test import ( + gotar "archive/tar" "bytes" "context" "crypto/md5" //nolint:gosec // test fixture digest, matches the exporter's hash.md5 contract @@ -2852,11 +2853,8 @@ func TestPipeline_BlockResumeAfterMerge(t *testing.T) { require.NoError(t, os.MkdirAll(filepath.Join(diskSnapDir, archive.ManifestsDirName), 0o755)) seedResumeIdentityMarker(t, diskSnapDir, diskSnapMarkerIdentity()) - require.NoError(t, os.WriteFile( - filepath.Join(diskSnapDir, archive.DataBlockName(".zst")), - []byte("pre-merged-block-data"), - 0o644, - )) + writeMergedZstdBlockFixture(t, filepath.Join(diskSnapDir, archive.DataBlockName(".zst")), + []byte("pre-merged-block-data")) cfg := pipeline.Config{ Namespace: testNS, @@ -2894,11 +2892,7 @@ func TestPipeline_FSResumeAfterTar(t *testing.T) { require.NoError(t, os.MkdirAll(filepath.Join(diskSnapDir, archive.ManifestsDirName), 0o755)) seedResumeIdentityMarker(t, diskSnapDir, diskSnapMarkerIdentity()) - require.NoError(t, os.WriteFile( - filepath.Join(diskSnapDir, archive.FsTarName), - []byte("pre-assembled-fs-tar"), - 0o644, - )) + writeAssembledFSTarFixture(t, filepath.Join(diskSnapDir, archive.FsTarName)) cfg := pipeline.Config{ Namespace: testNS, @@ -3646,6 +3640,39 @@ func TestPipeline_ForeignMergedBlock_NotLaunderedByResume(t *testing.T) { "collision dir must hold the correctly-downloaded bytes") } +// writeMergedZstdBlockFixture writes a real, decodable zstd frame at path, standing in for +// an already-merged block payload left by a prior run. FinalizeNode's MeasurePayload reads +// its frame header, which needs an explicit Frame_Content_Size (EncodeFrameStream stamps +// one; a placeholder wouldn't, and would fail finalize instead of testing resume-skip). +func writeMergedZstdBlockFixture(t *testing.T, path string, payload []byte) { + t.Helper() + + codec, err := compress.New("zstd", 0) + require.NoError(t, err, "compress.New(zstd, 0)") + + var buf bytes.Buffer + require.NoError(t, codec.EncodeFrameStream(&buf, bytes.NewReader(payload), int64(len(payload))), + "EncodeFrameStream") + + require.NoError(t, os.WriteFile(path, buf.Bytes(), 0o644)) +} + +// writeAssembledFSTarFixture writes a minimal, valid (empty) tar file at path, standing in +// for an already-assembled filesystem payload left by a prior run: FinalizeNode's +// MeasurePayload parses it via archive.SumTarRawSizes, so invalid tar content would fail +// finalize instead of testing resume-skip. An empty tar is enough since these tests only +// assert DataExport is skipped and the node finalizes, not the recorded size. +func writeAssembledFSTarFixture(t *testing.T, path string) { + t.Helper() + + var buf bytes.Buffer + + tw := gotar.NewWriter(&buf) + require.NoError(t, tw.Close()) + + require.NoError(t, os.WriteFile(path, buf.Bytes(), 0o644)) +} + // assertNodeComplete checks that snapshot.yaml exists in dir and VerifyNode passes. func assertNodeComplete(t *testing.T, dir string) { t.Helper() @@ -5777,9 +5804,13 @@ func TestPipeline_Progress_ClampStaleSeedToFreshTotal(t *testing.T) { require.NoError(t, os.MkdirAll(stagingDir, 0o755)) seedResumeIdentityMarker(t, diskSnapDir, diskSnapMarkerIdentity()) - // a.bin was fully staged as a flat blob under the OLD (larger) size, and - // the sizes sidecar records that stale size, so seedStreamFromDisk seeds - // both total (250) and current (250) — above the fresh listing total. + // a.bin was staged as a flat blob under the OLD (larger) size, so seedStreamFromDisk + // seeds total/current at 250 — above the fresh listing's 150, exercising the clamp + // below. The stale blob does NOT survive unchanged: with no MD5 for a.bin, + // stageCompressedFile measures its plaintext size, finds it disagrees with the + // listing's 150, and self-heals by re-fetching the true content in this same run. + // The clamp assertions below only cover progress-bar bookkeeping, not why the final + // 150 bytes end up correct. require.NoError(t, os.WriteFile(filepath.Join(stagingDir, "a.bin"+codec.Ext()), bytes.Repeat([]byte("A"), int(staleSize)), 0o644)) sizesJSON, err := json.Marshal(volume.FSSizesSidecar{ @@ -6869,11 +6900,8 @@ func TestPipeline_BlockAlreadyMerged_OwnDataRef_RemovesLeftoverChunkDir(t *testi archive.NodeDirName(childKind, diskSnapName)) require.NoError(t, os.MkdirAll(filepath.Join(diskSnapDir, archive.ManifestsDirName), 0o755)) seedResumeIdentityMarker(t, diskSnapDir, diskSnapMarkerIdentity()) - require.NoError(t, os.WriteFile( - filepath.Join(diskSnapDir, archive.DataBlockName(".zst")), - []byte("pre-merged-block-data"), - 0o644, - )) + writeMergedZstdBlockFixture(t, filepath.Join(diskSnapDir, archive.DataBlockName(".zst")), + []byte("pre-merged-block-data")) chunkDir := filepath.Join(diskSnapDir, archive.BlockChunksDirName) if tc.seedChunkDir { @@ -7039,11 +7067,8 @@ func TestPipeline_BlockAlreadyMerged_RemoveAllFailure_StillCompletes(t *testing. archive.NodeDirName(childKind, diskSnapName)) require.NoError(t, os.MkdirAll(filepath.Join(diskSnapDir, archive.ManifestsDirName), 0o755)) seedResumeIdentityMarker(t, diskSnapDir, diskSnapMarkerIdentity()) - require.NoError(t, os.WriteFile( - filepath.Join(diskSnapDir, archive.DataBlockName(".zst")), - []byte("pre-merged-block-data"), - 0o644, - )) + writeMergedZstdBlockFixture(t, filepath.Join(diskSnapDir, archive.DataBlockName(".zst")), + []byte("pre-merged-block-data")) chunkDir := seedLeftoverBlockChunkDir(t, diskSnapDir) require.NoError(t, os.Chmod(chunkDir, 0o555)) diff --git a/internal/snapshot/snapimport/fs.go b/internal/snapshot/snapimport/fs.go index 07d94028..a3c7ffdd 100644 --- a/internal/snapshot/snapimport/fs.go +++ b/internal/snapshot/snapimport/fs.go @@ -540,6 +540,9 @@ type fsTarScan struct { unsupportedEntryCount uint64 unsupportedEntrySummary string removeAll func(string) error + // rawTotal is the exact sum of every regular entry's PAX raw size, computed by this + // preflight pass — the upload's true total, known before any HEAD or PUT. + rawTotal int64 } func (s *fsTarScan) Close() error { @@ -1055,7 +1058,7 @@ func readFSTarRecord(reader io.Reader) (fsTarRecord, error) { // codec-geometry checks without activating a transfer. // // setTotal, when non-nil (nil disables reporting, matching onProgress's convention), is -// called with a running sum of exact PAX raw sizes as entries are walked. +// called exactly once, before any HEAD/PUT, with the exact sum of every regular entry's PAX raw size. // // activate, when non-nil, is called once per entry that actually needs a real PUT (the // NOT-done branch), never inside the `if done` server-side-skip branch above it. This is @@ -1276,6 +1279,12 @@ func uploadFSTarFromScanWithDependencies( slog.Int("directory_count", scan.ReservedEmptyDirectoryCount)) } + // scan.rawTotal was computed by the preflight pass; report it once, up front, so the + // bar's denominator is complete before the upload pass sends its first byte. + if setTotal != nil { + setTotal(scan.rawTotal) + } + info, err := source.Stat() if err != nil { return fmt.Errorf("inspect %s: %w", tarPath, err) @@ -1391,6 +1400,8 @@ func uploadFSTarFromScanWithDependencies( ) } + // runningTotal is no longer reported via setTotal (see the up-front call above); it + // only lets the post-loop check prove sizes didn't change since the preflight pass. runningTotal, err = addRawSize(runningTotal, metadata.RawSize) if err != nil { return errors.Join( @@ -1399,10 +1410,6 @@ func uploadFSTarFromScanWithDependencies( ) } - if setTotal != nil { - setTotal(runningTotal) - } - fileURL, err := fileUploadURL(baseURL, relPath) if err != nil { return errors.Join(err, closeFSTarSequence(sequence)) @@ -1466,6 +1473,16 @@ func uploadFSTarFromScanWithDependencies( ) } + // Extends the entry-count check above to bytes: the total already reported via setTotal + // must match what this pass actually walked, proving sizes didn't change since preflight. + if runningTotal != scan.rawTotal { + return errors.Join( + fmt.Errorf("%w: tar raw size total changed after preflight (%d/%d bytes)", + archive.ErrInvalidFSMetadata, runningTotal, scan.rawTotal), + closeFSTarSequence(sequence), + ) + } + if _, err := readFSTarSequenceDigest(sequenceReader); !errors.Is(err, io.EOF) { if err == nil { err = fmt.Errorf("%w: filesystem tar preflight sequence has an unexpected trailing record", @@ -1660,6 +1677,7 @@ func scanFSTarReaderWithOptions( var ( entryCount uint64 regularCount uint64 + rawTotal int64 ) for { @@ -1742,6 +1760,13 @@ func scanFSTarReaderWithOptions( return cleanup(fmt.Errorf("entry %q: %w", hdr.Name, err)) } + rawTotal, err = addRawSize(rawTotal, metadata.RawSize) + if err != nil { + _ = sequence.Close() + + return cleanup(fmt.Errorf("account tar entry %q: %w", hdr.Name, err)) + } + record.Kind = fsTarRecordRegular record.Path = path.Clean(metadata.OriginalPath) regularCount++ @@ -1803,6 +1828,7 @@ func scanFSTarReaderWithOptions( unsupportedEntryCount: diagnostics.count, unsupportedEntrySummary: diagnostics.summary(), removeAll: options.removeAll, + rawTotal: rawTotal, }, nil } diff --git a/internal/snapshot/snapimport/fs_test.go b/internal/snapshot/snapimport/fs_test.go index 022798d9..8b6db0c1 100644 --- a/internal/snapshot/snapimport/fs_test.go +++ b/internal/snapshot/snapimport/fs_test.go @@ -1217,6 +1217,108 @@ func TestScanFSTar_AcceptsStructuralDirectoryChain(t *testing.T) { } } +// TestScanFSTar_RawTotal covers scanFSTar's rawTotal accumulation: the exact sum of every +// regular entry's PAX raw size, ignoring directory and other non-regular entries entirely. +func TestScanFSTar_RawTotal(t *testing.T) { + t.Parallel() + + t.Run("success: empty tar sums to zero", func(t *testing.T) { + t.Parallel() + + var tarBuf bytes.Buffer + + tw := tar.NewWriter(&tarBuf) + if err := tw.Close(); err != nil { + t.Fatalf("close tar: %v", err) + } + + tarPath := filepath.Join(t.TempDir(), "data.tar") + if err := os.WriteFile(tarPath, tarBuf.Bytes(), 0o600); err != nil { + t.Fatalf("write data.tar: %v", err) + } + + scan, err := scanFSTar(context.Background(), tarPath) + if err != nil { + t.Fatalf("scanFSTar: %v", err) + } + t.Cleanup(func() { _ = scan.Close() }) + + if scan.rawTotal != 0 { + t.Errorf("rawTotal = %d, want 0", scan.rawTotal) + } + }) + + t.Run("success: only reserved empty directory sums to zero", func(t *testing.T) { + t.Parallel() + + var tarBuf bytes.Buffer + + tw := tar.NewWriter(&tarBuf) + if err := tw.WriteHeader(&tar.Header{Typeflag: tar.TypeDir, Name: "lost+found/", Mode: 0o700}); err != nil { + t.Fatalf("write reserved empty directory header: %v", err) + } + + if err := tw.Close(); err != nil { + t.Fatalf("close tar: %v", err) + } + + tarPath := filepath.Join(t.TempDir(), "data.tar") + if err := os.WriteFile(tarPath, tarBuf.Bytes(), 0o600); err != nil { + t.Fatalf("write data.tar: %v", err) + } + + scan, err := scanFSTar(context.Background(), tarPath) + if err != nil { + t.Fatalf("scanFSTar: %v", err) + } + t.Cleanup(func() { _ = scan.Close() }) + + if scan.rawTotal != 0 { + t.Errorf("rawTotal = %d, want 0 (no regular entries)", scan.rawTotal) + } + + if scan.ReservedEmptyDirectoryCount != 1 { + t.Errorf("ReservedEmptyDirectoryCount = %d, want 1", scan.ReservedEmptyDirectoryCount) + } + }) + + t.Run("success: sums only regular entries, ignoring directories", func(t *testing.T) { + t.Parallel() + + first := []byte("first entry content") + second := []byte("a somewhat longer second entry content") + + var tarBuf bytes.Buffer + + tw := tar.NewWriter(&tarBuf) + if err := tw.WriteHeader(&tar.Header{Typeflag: tar.TypeDir, Name: "dir/", Mode: 0o755}); err != nil { + t.Fatalf("write directory header: %v", err) + } + + addTarEntryMetadata(t, tw, "first.txt", "first.txt", "none", int64(len(first)), first, 0o600, 0, 0, time.Time{}) + addTarEntryMetadata(t, tw, "dir/second.txt", "dir/second.txt", "none", int64(len(second)), second, 0o600, 0, 0, time.Time{}) + + if err := tw.Close(); err != nil { + t.Fatalf("close tar: %v", err) + } + + tarPath := filepath.Join(t.TempDir(), "data.tar") + if err := os.WriteFile(tarPath, tarBuf.Bytes(), 0o600); err != nil { + t.Fatalf("write data.tar: %v", err) + } + + scan, err := scanFSTar(context.Background(), tarPath) + if err != nil { + t.Fatalf("scanFSTar: %v", err) + } + t.Cleanup(func() { _ = scan.Close() }) + + if want := int64(len(first) + len(second)); scan.rawTotal != want { + t.Errorf("rawTotal = %d, want %d", scan.rawTotal, want) + } + }) +} + func TestScanFSTar_AcceptsReservedEmptyDirectory(t *testing.T) { t.Parallel() @@ -3144,10 +3246,9 @@ func TestImportFSFromTar_SkipsAlreadyUploadedEntryWithoutTransfer(t *testing.T) t.Errorf("onProgress total = %d, want %d (skipped alpha.txt must still be credited at its exact decompressed size, plus beta.txt)", progressed, want) } - // setTotal must grow progressively: alpha.txt's exact size becomes known first from - // authenticated PAX metadata before its done-skip proof, then beta.txt adds its own - // exact size — never a single upfront call with the grand total. - wantTotals := []int64{int64(len(alphaPlain)), int64(len(alphaPlain) + len(betaPlain))} + // setTotal is called exactly once, before any HEAD/PUT, with the preflight pass's + // exact sum of both entries' PAX raw sizes — never a progressively growing total. + wantTotals := []int64{int64(len(alphaPlain) + len(betaPlain))} if len(totals) != len(wantTotals) { t.Fatalf("setTotal called %d times with %v, want %d calls with %v", len(totals), totals, len(wantTotals), wantTotals) } @@ -3168,6 +3269,118 @@ func TestImportFSFromTar_SkipsAlreadyUploadedEntryWithoutTransfer(t *testing.T) } } +// countingHTTPDoer counts every HTTPDo call made through it, so a test can prove an event +// (like a setTotal call) happened strictly before the first HTTP request. +type countingHTTPDoer struct { + inner httpDoer + calls int +} + +func (d *countingHTTPDoer) HTTPDo(req *http.Request) (*http.Response, error) { + d.calls++ + + return d.inner.HTTPDo(req) +} + +// TestImportFSFromTar_SetsExactTotalOnce proves setTotal is called exactly once, with the +// exact sum of every entry's PAX raw size, strictly before any HEAD or PUT request. +func TestImportFSFromTar_SetsExactTotalOnce(t *testing.T) { + t.Parallel() + + first := []byte("first entry, exact raw size known from PAX metadata") + second := []byte("second entry, a different exact raw size") + + var tarBuf bytes.Buffer + + tw := tar.NewWriter(&tarBuf) + addTarEntryMetadata(t, tw, "first.txt", "first.txt", "none", int64(len(first)), first, 0o600, 1, 2, time.Time{}) + addTarEntryMetadata(t, tw, "second.txt", "second.txt", "none", int64(len(second)), second, 0o600, 1, 2, time.Time{}) + + if err := tw.Close(); err != nil { + t.Fatalf("close tar: %v", err) + } + + dir := t.TempDir() + tarPath := filepath.Join(dir, "data.tar") + + if err := os.WriteFile(tarPath, tarBuf.Bytes(), 0o600); err != nil { + t.Fatalf("write data.tar: %v", err) + } + + imp := newFakeFileImporter() + srv := httptest.NewServer(imp) + t.Cleanup(srv.Close) + + doer := &countingHTTPDoer{inner: plainHTTPDoer{}} + + var ( + totalCalls int + lastTotal int64 + httpCallsAtSetTotal int + ) + + setTotal := func(n int64) { + totalCalls++ + lastTotal = n + httpCallsAtSetTotal = doer.calls + } + + if err := importFSFromTar(context.Background(), doer, srv.URL, tarPath, discardLogger(), setTotal, nil, nil); err != nil { + t.Fatalf("importFSFromTar: %v", err) + } + + if totalCalls != 1 { + t.Fatalf("setTotal called %d times, want exactly 1", totalCalls) + } + + if want := int64(len(first) + len(second)); lastTotal != want { + t.Errorf("setTotal value = %d, want %d (exact sum of both entries)", lastTotal, want) + } + + if httpCallsAtSetTotal != 0 { + t.Errorf("setTotal was called after %d HTTP request(s) had already been issued, want 0", httpCallsAtSetTotal) + } +} + +// TestImportFSTar_RawTotalMismatchAfterUpload proves the post-loop defensive check: if the +// upload pass's observed raw-size total disagrees with the preflight's scan.rawTotal +// (already reported via setTotal), the upload fails instead of reporting a wrong total. +func TestImportFSTar_RawTotalMismatchAfterUpload(t *testing.T) { + t.Parallel() + + content := []byte("single entry content for the raw-total mismatch fixture") + tarPath := writeSingleEntryFSTar(t, "none", content) + + file, err := os.Open(tarPath) + if err != nil { + t.Fatalf("open %s: %v", tarPath, err) + } + t.Cleanup(func() { _ = file.Close() }) + + scan, err := scanFSTarSource(context.Background(), file, tarPath) + if err != nil { + t.Fatalf("scanFSTarSource: %v", err) + } + t.Cleanup(func() { _ = scan.Close() }) + + // Simulate the preflight total disagreeing with what the upload pass walks, without + // touching any byte (which would also trip the digest revalidation, masking this check). + scan.rawTotal++ + + imp := newFakeFileImporter() + srv := httptest.NewServer(imp) + t.Cleanup(srv.Close) + + err = uploadFSTarFromScan(context.Background(), plainHTTPDoer{}, srv.URL, tarPath, file, discardLogger(), nil, nil, nil, &scan) + if err == nil { + t.Fatal("expected error for raw-size total mismatch, got nil") + } + + if !errors.Is(err, archive.ErrInvalidFSMetadata) { + t.Errorf("expected error wrapping archive.ErrInvalidFSMetadata, got: %v", err) + } +} + type countingReadSeeker struct { io.ReadSeeker bytes int64 @@ -4140,11 +4353,9 @@ func TestImportFSFromTar_PerCodecRoundTrip(t *testing.T) { t.Errorf("onProgress total = %d, want %d", reported, want) } - // setTotal must grow progressively across both not-done entries: first.dat's - // exact size is measured (or read from hdr.Size for codec "none") before - // second.dat is even reached, then second.dat's own exact size is added on - // top — proving the running sum, not a single grand total known up front. - wantTotals := []int64{int64(len(firstContent)), int64(len(firstContent) + len(secondContent))} + // setTotal is called exactly once, before any HEAD/PUT, with the preflight pass's + // exact sum of both entries' PAX raw sizes — not a progressively growing total. + wantTotals := []int64{int64(len(firstContent) + len(secondContent))} if len(totals) != len(wantTotals) { t.Fatalf("setTotal called %d times with %v, want %d calls with %v", len(totals), totals, len(wantTotals), wantTotals) } diff --git a/internal/snapshot/snapimport/plan.go b/internal/snapshot/snapimport/plan.go index 7610c406..66f98222 100644 --- a/internal/snapshot/snapimport/plan.go +++ b/internal/snapshot/snapimport/plan.go @@ -121,6 +121,20 @@ type PlannedNode struct { NodeChecksum string // SizeBytes is Size parsed once into its canonical byte count before cluster mutation. SizeBytes int64 + // PayloadRawSizeBytes/PayloadStoredSizeBytes are the measured on-disk payload footprint + // recorded in snapshot.yaml Volumes[0] (archive.VolumeInfo.RawSizeBytes/StoredSizeBytes), + // read verbatim from an archive at FormatVersion >= SnapshotFormatVersionPayloadSizes. On + // an older archive both are zero (never recorded); resolveBlockPayloadSize disambiguates + // that from a genuinely empty v3 payload using FormatVersion, and measures the upload + // size from the payload itself instead. Empty for structural/aggregator nodes. + PayloadRawSizeBytes int64 + PayloadStoredSizeBytes int64 + // FormatVersion is the archive envelope version this node's snapshot.yaml declared (see + // archive.SnapshotFormatVersion*). It exists solely to disambiguate + // PayloadRawSizeBytes == 0 meaning "not recorded" (FormatVersion < + // archive.SnapshotFormatVersionPayloadSizes) from "recorded, genuinely empty payload" + // (FormatVersion >= archive.SnapshotFormatVersionPayloadSizes). + FormatVersion int // PayloadKind and Codec are the classified on-disk upload representation. PayloadKind string Codec string @@ -659,6 +673,7 @@ func (b *planBuilder) readNode(source *archive.RootedSource) (PlannedNode, error Manifests: manifests, SourceObjectRef: sy.SourceObjectRef, NodeChecksum: sy.Checksum.Hex, + FormatVersion: sy.FormatVersion, snapshotDigest: sha256.Sum256(snapshotData), snapshotInfo: snapshotInfoBefore, manifestFiles: manifestFiles, @@ -672,6 +687,8 @@ func (b *planBuilder) readNode(source *archive.RootedSource) (PlannedNode, error node.StorageClassName = v.StorageClassName node.Size = v.Size node.VolumeMode = v.VolumeMode + node.PayloadRawSizeBytes = v.RawSizeBytes + node.PayloadStoredSizeBytes = v.StoredSizeBytes } blockPayload, found, err := archive.ClassifyBlockPayloadIn(source) diff --git a/internal/snapshot/snapimport/plan_test.go b/internal/snapshot/snapimport/plan_test.go index 20998bb5..69dcce30 100644 --- a/internal/snapshot/snapimport/plan_test.go +++ b/internal/snapshot/snapimport/plan_test.go @@ -1192,6 +1192,78 @@ func TestBuildPlan_LeafStorageParams(t *testing.T) { } } +// TestBuildPlan_ReadsPayloadSizeFields verifies that PlannedNode.PayloadRawSizeBytes/ +// PayloadStoredSizeBytes/FormatVersion are read verbatim from snapshot.yaml's +// Volumes[0].RawSizeBytes/StoredSizeBytes and its own formatVersion. +func TestBuildPlan_ReadsPayloadSizeFields(t *testing.T) { + t.Parallel() + + root := t.TempDir() + writeArchiveNode(t, root, archiveNode{ + apiVersion: "state-snapshotter.deckhouse.io/v1alpha1", + kind: "Snapshot", + name: "root", + }) + + leafDir := childDir(root, "VolumeSnapshot", "pvc-1") + writeArchiveNode(t, leafDir, archiveNode{ + apiVersion: "snapshot.storage.k8s.io/v1", + kind: "VolumeSnapshot", + name: "pvc-1", + blockData: []byte("rawbytes"), + blockExt: ".zst", + volumes: []archive.VolumeInfo{{ + StorageClassName: "sc-fast", + Size: "1Gi", + VolumeMode: "Block", + RawSizeBytes: 1077665792, + StoredSizeBytes: 900000000, + }}, + }) + finalizeArchiveChildrenChecksums(t, root) + + plan, err := BuildPlan(root) + if err != nil { + t.Fatalf("BuildPlan: %v", err) + } + + var leaf *PlannedNode + + for i := range plan { + if plan[i].Kind == "VolumeSnapshot" { + leaf = &plan[i] + + break + } + } + + if leaf == nil { + t.Fatal("VolumeSnapshot node not found in plan") + } + + if leaf.FormatVersion != archive.SnapshotFormatVersionCurrent { + t.Errorf("FormatVersion = %d, want %d", leaf.FormatVersion, archive.SnapshotFormatVersionCurrent) + } + + if leaf.PayloadRawSizeBytes != 1077665792 { + t.Errorf("PayloadRawSizeBytes = %d, want 1077665792", leaf.PayloadRawSizeBytes) + } + + if leaf.PayloadStoredSizeBytes != 900000000 { + t.Errorf("PayloadStoredSizeBytes = %d, want 900000000", leaf.PayloadStoredSizeBytes) + } + + // The nominal Size/SizeBytes fields (feeding scratch-volume provisioning and resume + // identity) must be entirely unaffected by the new payload-size fields. + if leaf.Size != "1Gi" { + t.Errorf("Size = %q, want %q", leaf.Size, "1Gi") + } + + if leaf.SizeBytes != 1024*1024*1024 { + t.Errorf("SizeBytes = %d, want %d", leaf.SizeBytes, int64(1024*1024*1024)) + } +} + func TestDataImportIdentity_CanonicalAndDimensionComplete(t *testing.T) { base := PlannedNode{ APIVersion: "snapshot.storage.k8s.io/v1", diff --git a/internal/snapshot/snapimport/volume.go b/internal/snapshot/snapimport/volume.go index 6106a2ed..e592c015 100644 --- a/internal/snapshot/snapimport/volume.go +++ b/internal/snapshot/snapimport/volume.go @@ -644,19 +644,9 @@ func (c *clusterVolumeImporter) sendVolumeDataFromSource( slog.String("namespace", namespace), slog.String("dataimport", diName)) - // The FS-upload total is reported PROGRESSIVELY, not as a single a-priori value - // like the block path's setTotal(totalSize) below: a tar header only records a - // compressed entry's STORED length, so the true (decompressed) size of a LATER - // file is not knowable until importFSFromTar has walked every entry before it. - // Computing a full total up front would mean decompressing every entry just to - // measure it -- exactly the extra work the two-pass streaming design exists to - // avoid when a resume never needs it (see importFSFromTar). Instead, setTotal is - // threaded straight through: importFSFromTar calls it with a running sum each - // time a new file's exact size becomes known (a skipped/already-done file's size - // from HEAD's Content-Length, or a not-done file's exact size from its measure - // step / hdr.Size), so the bar's denominator grows as work is discovered rather - // than staying at zero for the whole upload. Honest limitation: the total is not - // complete until the LAST entry in the tar has been processed. + // The FS-upload total, like the block path's setTotal(totalSize) below, is reported + // up front: importFSFromTar's header-only preflight pass already knows every entry's + // exact PAX raw size, so it sums them and calls setTotal once, before any HEAD/PUT. var err error if handle == nil { err = importFSFromTarWithDependencies( @@ -690,9 +680,9 @@ func (c *clusterVolumeImporter) sendVolumeDataFromSource( ) if handle == nil { - totalSize, err = blockTotalSize(leaf.DataFile, leaf.Size, ext) + totalSize, err = resolveBlockPayloadSize(ctx, leaf, nil) } else { - totalSize, err = blockTotalSizeFromSource(leaf.DataFile, leaf.Size, ext, handle) + totalSize, err = resolveBlockPayloadSize(ctx, leaf, handle) } if err != nil { @@ -709,9 +699,9 @@ func (c *clusterVolumeImporter) sendVolumeDataFromSource( slog.String("dataimport", diName), slog.Int64("bytes", totalSize)) - // totalSize is known up front (blockTotalSize never decompresses to measure it) - // and matches the onProgress increments (validated durable offsets advance the bar), - // so report it as the total before any bytes are sent. + // totalSize is known up front — trusted from the archive's recorded measurement, or + // else measured by resolveBlockPayloadSize without a full decode — and matches the + // onProgress increments, so report it as the total before any bytes are sent. if setTotal != nil { setTotal(totalSize) } @@ -1138,74 +1128,75 @@ func drainAndCloseResponseBody(resp *http.Response) error { return errors.Join(drainErr, closeErr) } -// ErrRawBlockSizeMismatch is returned by blockTotalSize when a raw (codec -// none) data.bin file's on-disk size does not match the size captured in the -// archive's VolumeInfo. Unlike a compressed payload, a raw payload has no -// separate decompressed size to fall back on — stat size and captured size -// are the SAME quantity — so any disagreement means a truncated, corrupted, or -// mismatched archive. Checking this before any HEAD/PUT keeps the failure -// deterministic and sends zero HTTP requests, instead of streaming a wrong -// byte count to the importer and only discovering the mismatch mid-transfer. +// ErrRawBlockSizeMismatch is returned by resolveBlockPayloadSize when a raw (codec none) +// data.bin's on-disk size disagrees with the archive's recorded StoredSizeBytes — for raw +// payloads the two are definitionally equal, so any mismatch means a corrupted archive. +// Checked before any HEAD/PUT. Never fires for an archive older than +// SnapshotFormatVersionPayloadSizes (nothing recorded to cross-check). var ErrRawBlockSizeMismatch = errors.New("raw block size mismatch") var errFailedBlockDecoderClose = errors.New("failed to close block decoder") -// blockTotalSize returns the exact decompressed byte count of a node's block-volume data -// file without decompressing it. size (a resource.Quantity string like "10Gi", sourced from -// VolumeSnapshotContent.status.restoreSize — see archive.VolumeInfo.Size) is parsed for -// EVERY codec, including raw: a block-volume capture always reads exactly the device's -// provisioned byte size, so the captured size is the total regardless of codec. Parsing -// size instead of decompressing avoids a full decompression pass purely to learn a byte -// count, which is the whole point of the streaming upload path. +// resolveBlockPayloadSize determines a block leaf's exact upload size (totalSize). // -// For a raw file (ext==""), the parsed size is additionally cross-checked against the -// file's actual on-disk size (os.Stat) — see ErrRawBlockSizeMismatch — because a raw -// payload's stat size and its captured size are definitionally the same number, so any -// difference is a corrupt/mismatched archive rather than a codec-driven size difference. -// A compressed file's on-disk (compressed) size is not comparable to the captured -// (decompressed) size at all, so no such check is possible or meaningful for ext != "". -func blockTotalSize(dataFile, size, ext string) (int64, error) { - return blockTotalSizeFromSource(dataFile, size, ext, nil) -} - -func blockTotalSizeFromSource(dataFile, size, ext string, source interface { - Stat() (os.FileInfo, error) -}) (int64, error) { - q, err := resource.ParseQuantity(size) - if err != nil { - return 0, fmt.Errorf("parsing captured volume size %q for %s: %w", size, dataFile, err) +// For a non-raw codec on a current-format archive (FormatVersion >= SnapshotFormatVersionPayloadSizes +// and PayloadRawSizeBytes > 0), it trusts the recorded measurement outright rather than +// re-decoding a possibly-huge file. A zero or legacy value always falls back to measuring +// (cheap header read for zstd, full decode otherwise). +// +// A raw (ext == "") payload is always measured — its on-disk length IS its decoded length, +// a cheap Seek — which also lets a current-format archive cross-check StoredSizeBytes +// (see ErrRawBlockSizeMismatch). +// +// source is the leaf's already-open reader when the caller holds one; nil means +// resolveBlockPayloadSize opens leaf.DataFile itself. +func resolveBlockPayloadSize(ctx context.Context, leaf PlannedNode, source io.ReadSeeker) (int64, error) { + if leaf.Ext != "" && leaf.FormatVersion >= archive.SnapshotFormatVersionPayloadSizes && leaf.PayloadRawSizeBytes > 0 { + return leaf.PayloadRawSizeBytes, nil } - captured := q.Value() + if source != nil { + return measureBlockPayloadSize(ctx, leaf, source) + } - if ext != "" { - return captured, nil + file, err := os.Open(leaf.DataFile) + if err != nil { + return 0, fmt.Errorf("open volume data %s: %w", leaf.DataFile, err) } - var info os.FileInfo - if source == nil { - info, err = os.Stat(dataFile) - } else { - info, err = source.Stat() + size, measureErr := measureBlockPayloadSize(ctx, leaf, file) + closeErr := file.Close() + + if err := errors.Join(measureErr, closeErr); err != nil { + return 0, err } + return size, nil +} + +// measureBlockPayloadSize measures leaf's block payload size from an already-open source. +// For ext == "" it also cross-checks the result against StoredSizeBytes, but only when that +// value is strictly positive — zero is ambiguous between "empty" and "never recorded". +func measureBlockPayloadSize(ctx context.Context, leaf PlannedNode, source io.ReadSeeker) (int64, error) { + size, err := compress.DecodedSize(ctx, leaf.Ext, source) if err != nil { - return 0, fmt.Errorf("stat volume data %s: %w", dataFile, err) + return 0, fmt.Errorf("determine block payload size for %s: %w", leaf.DataFile, err) } - if info.Size() != captured { - return 0, fmt.Errorf("%s: on-disk size %d does not match captured volume size %d (%q): %w", - dataFile, info.Size(), captured, size, ErrRawBlockSizeMismatch) + if leaf.Ext == "" && leaf.FormatVersion >= archive.SnapshotFormatVersionPayloadSizes && + leaf.PayloadStoredSizeBytes > 0 && size != leaf.PayloadStoredSizeBytes { + return 0, fmt.Errorf("%s: on-disk size %d does not match archive-recorded size %d: %w", + leaf.DataFile, size, leaf.PayloadStoredSizeBytes, ErrRawBlockSizeMismatch) } - return captured, nil + return size, nil } // putBlock streams the block-volume payload at dataFile to the importer's block // endpoint, honouring the server-reported X-Next-Offset for resumable progress. ext // selects the decode codec via compress.NewReader ("" for raw/no codec, matching // Codec.Ext); totalSize is the volume's exact decompressed byte count (see -// blockTotalSize). onProgress, when non-nil, is called as validated server offsets make +// resolveBlockPayloadSize). onProgress, when non-nil, is called as validated server offsets make // raw bytes known durable, including the initial HEAD prefix. activate, when non-nil, is called at the start of every // real transfer iteration (never when offset==totalSize short-circuits before any PUT is // attempted), so the caller's progress stream is activated only on a genuine transfer. @@ -2139,6 +2130,10 @@ func headBlockOffset(ctx context.Context, httpClient httpDoer, url string, total return 0, fmt.Errorf("invalid X-Next-Offset %q from %s: %w", next, url, err) } + if err := verifyDeviceCapacity(resp.Header.Get("X-Device-Size"), totalSize, url); err != nil { + return 0, err + } + return off, nil case http.StatusNotFound: if err := validateBlockOffset(0, totalSize); err != nil { @@ -2151,6 +2146,33 @@ func headBlockOffset(ctx context.Context, httpClient httpDoer, url string, total } } +// verifyDeviceCapacity fails BEFORE the first PUT when the importer's HEAD response +// (deviceSizeHeader, i.e. X-Device-Size) proves the target device is smaller than totalSize +// — otherwise this would only surface as a mid-transfer failure. +// +// A missing, empty, or unparsable header fails OPEN (returns nil): its absence isn't +// evidence the device is too small, only a proven shortfall is. +func verifyDeviceCapacity(deviceSizeHeader string, totalSize int64, url string) error { + if deviceSizeHeader == "" { + return nil + } + + deviceSize, err := strconv.ParseInt(deviceSizeHeader, 10, 64) + if err != nil { + return nil + } + + if deviceSize < totalSize { + return fmt.Errorf( + "target device size %d bytes is smaller than payload %d bytes (HEAD %s); "+ + "the target storage class may round up volumes differently than the source — check storageClassName", + deviceSize, totalSize, url, + ) + } + + return nil +} + // doBlockChunk performs one bounded PUT. Successful producer responses must acknowledge // exactly requestEnd; a conflict returns the producer's validated reposition offset. func doBlockChunk(httpClient httpDoer, req *http.Request, offset, requestEnd, totalSize int64) (int64, bool, error) { diff --git a/internal/snapshot/snapimport/volume_test.go b/internal/snapshot/snapimport/volume_test.go index 77c83496..18e5de4b 100644 --- a/internal/snapshot/snapimport/volume_test.go +++ b/internal/snapshot/snapimport/volume_test.go @@ -3378,7 +3378,16 @@ func TestSendVolumeData_CompressedFullSkipRequiresExactDecodedSize(t *testing.T) }, nil }) - leaf := PlannedNode{DataFile: dataFile, Ext: tc.ext, Size: strconv.FormatInt(tc.totalSize, 10)} + // FormatVersion/PayloadRawSizeBytes (not Size) drive resolveBlockPayloadSize's fast + // path; setting them to a wrong tc.totalSize simulates a lying manifest, which the + // exact-decoded-size proof below must still catch. + leaf := PlannedNode{ + DataFile: dataFile, + Ext: tc.ext, + Size: strconv.FormatInt(tc.totalSize, 10), + FormatVersion: archive.SnapshotFormatVersionPayloadSizes, + PayloadRawSizeBytes: tc.totalSize, + } importer := &clusterVolumeImporter{log: discardLogger()} err := importer.sendVolumeData( @@ -3562,7 +3571,15 @@ func TestSendVolumeData_TwoRunCompressedUndercountNeverFinalizes(t *testing.T) { } }) - leaf := PlannedNode{DataFile: dataFile, Ext: ".zst", Size: strconv.FormatInt(totalSize, 10)} + // FormatVersion/PayloadRawSizeBytes (not Size) drive resolveBlockPayloadSize's fast path; + // an undercounted totalSize simulates a manifest that under-reports, which must never finalize. + leaf := PlannedNode{ + DataFile: dataFile, + Ext: ".zst", + Size: strconv.FormatInt(totalSize, 10), + FormatVersion: archive.SnapshotFormatVersionPayloadSizes, + PayloadRawSizeBytes: totalSize, + } importer := &clusterVolumeImporter{log: discardLogger()} for run := 1; run <= 2; run++ { @@ -3722,7 +3739,16 @@ func TestSendVolumeData_ConflictToTotalRequiresExactDecodedSize(t *testing.T) { } }) - leaf := PlannedNode{DataFile: dataFile, Ext: ".zst", Size: strconv.FormatInt(tc.totalSize, 10)} + // FormatVersion/PayloadRawSizeBytes (not Size) drive resolveBlockPayloadSize's fast + // path; setting them to a wrong tc.totalSize simulates a lying manifest, which the + // exact-decoded-size proof below must still catch. + leaf := PlannedNode{ + DataFile: dataFile, + Ext: ".zst", + Size: strconv.FormatInt(tc.totalSize, 10), + FormatVersion: archive.SnapshotFormatVersionPayloadSizes, + PayloadRawSizeBytes: tc.totalSize, + } importer := &clusterVolumeImporter{log: discardLogger()} err := importer.sendVolumeData( @@ -5581,126 +5607,209 @@ func TestSendVolumeData_FSLeaf_UsesTarFile(t *testing.T) { } } -// TestBlockTotalSize covers every codec and every invalid-size shape -// blockTotalSize must handle: the raw (ext=="") on-disk size is cross-checked -// against the captured VolumeInfo.Size for BOTH a short and a long mismatch, -// while a compressed file's on-disk (compressed) size is never compared to -// the captured (decompressed) size at all. A missing or unparsable captured -// size fails regardless of codec. -func TestBlockTotalSize(t *testing.T) { +// poisonReadSeeker fails the test immediately if Read or Seek is called — proves +// resolveBlockPayloadSize's fast path trusts PayloadRawSizeBytes without touching the file. +type poisonReadSeeker struct{ t *testing.T } + +func (p poisonReadSeeker) Read(_ []byte) (int, error) { + p.t.Helper() + p.t.Fatal("unexpected Read: the fast path must not touch the payload file") + + return 0, nil +} + +func (p poisonReadSeeker) Seek(_ int64, _ int) (int64, error) { + p.t.Helper() + p.t.Fatal("unexpected Seek: the fast path must not touch the payload file") + + return 0, nil +} + +// TestResolveBlockPayloadSize covers its three decision paths: the current-format fast path +// for a non-raw codec (trusts PayloadRawSizeBytes, no I/O); the measured path for a legacy +// archive or any raw payload; and the v3 raw cross-check against PayloadStoredSizeBytes. +func TestResolveBlockPayloadSize(t *testing.T) { + t.Parallel() + tests := []struct { name string - ext string - size string - fileContent []byte // nil => no on-disk file at all + leaf PlannedNode + fileContent []byte // written to leaf.DataFile before the call + useSource bool // pass the file content as an explicit io.ReadSeeker wantTotal int64 - wantErr error // nil => any non-nil error is acceptable - wantErrNil bool + wantErr error }{ { - name: "raw exact match", - ext: "", - size: "10", - fileContent: []byte("0123456789"), - wantTotal: 10, - wantErrNil: true, + name: "success: v3 zstd trusts recorded raw size without reading the file", + leaf: PlannedNode{ + Ext: ".zst", + FormatVersion: archive.SnapshotFormatVersionPayloadSizes, + PayloadRawSizeBytes: 1077665792, + }, + wantTotal: 1077665792, }, { - name: "raw short mismatch (on-disk smaller than captured)", - ext: "", - size: "10", - fileContent: []byte("12345"), - wantErr: ErrRawBlockSizeMismatch, + name: "success: v3 gzip trusts recorded raw size without reading the file", + leaf: PlannedNode{ + Ext: ".gz", + FormatVersion: archive.SnapshotFormatVersionPayloadSizes, + PayloadRawSizeBytes: 4096, + }, + wantTotal: 4096, }, { - name: "raw long mismatch (on-disk larger than captured)", - ext: "", - size: "10", - fileContent: []byte("012345678901234567890123456789"), - wantErr: ErrRawBlockSizeMismatch, + name: "success: legacy zstd is measured via frame headers, not trusted from a field", + leaf: PlannedNode{ + Ext: ".zst", + FormatVersion: archive.SnapshotFormatVersionAuthenticatedChildren, + // A stale/absent field on a legacy archive must be ignored entirely. + PayloadRawSizeBytes: 999999, + }, }, { - name: "zstd: on-disk (compressed) size never compared to captured size", - ext: ".zst", - size: "10Gi", - fileContent: []byte("short-compressed-stand-in"), - wantTotal: 10 * 1024 * 1024 * 1024, - wantErrNil: true, + name: "success: legacy other codec is measured via a full decode from an explicit source", + leaf: PlannedNode{ + Ext: ".gz", + FormatVersion: archive.SnapshotFormatVersionAuthenticatedChildren, + }, + useSource: true, }, { - name: "gzip: captured size is authoritative", - ext: ".gz", - size: "5Mi", - fileContent: []byte("x"), - wantTotal: 5 * 1024 * 1024, - wantErrNil: true, + name: "success: legacy raw payload is measured via seek, no cross-check performed", + leaf: PlannedNode{ + Ext: "", + FormatVersion: archive.SnapshotFormatVersionAuthenticatedChildren, + // Deliberately mismatched: legacy archives never recorded StoredSizeBytes, + // so nothing to cross-check against — the measured length is trusted outright. + PayloadStoredSizeBytes: 999999, + }, }, { - name: "lz4: captured size is authoritative", - ext: ".lz4", - size: "1Ki", - fileContent: []byte("x"), - wantTotal: 1024, - wantErrNil: true, + name: "success: v3 raw payload matches recorded StoredSizeBytes", + leaf: PlannedNode{ + Ext: "", + FormatVersion: archive.SnapshotFormatVersionPayloadSizes, + PayloadStoredSizeBytes: 10, + }, + fileContent: []byte("0123456789"), + wantTotal: 10, }, { - name: "missing captured size", - ext: "", - size: "", + name: "error: v3 raw payload disagrees with recorded StoredSizeBytes (short)", + leaf: PlannedNode{ + Ext: "", + FormatVersion: archive.SnapshotFormatVersionPayloadSizes, + PayloadStoredSizeBytes: 10, + }, fileContent: []byte("12345"), - wantErrNil: false, + wantErr: ErrRawBlockSizeMismatch, }, { - name: "invalid captured size", - ext: "", - size: "not-a-quantity", - fileContent: []byte("12345"), - wantErrNil: false, + name: "error: v3 raw payload disagrees with recorded StoredSizeBytes (long)", + leaf: PlannedNode{ + Ext: "", + FormatVersion: archive.SnapshotFormatVersionPayloadSizes, + PayloadStoredSizeBytes: 10, + }, + fileContent: []byte("012345678901234567890123456789"), + wantErr: ErrRawBlockSizeMismatch, }, { - name: "raw file missing on disk", - ext: "", - size: "10", - wantErrNil: false, + name: "error: raw file missing on disk", + leaf: PlannedNode{ + Ext: "", + FormatVersion: archive.SnapshotFormatVersionAuthenticatedChildren, + }, + fileContent: nil, }, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { + t.Parallel() + dir := t.TempDir() - dataFile := filepath.Join(dir, "data.bin"+tc.ext) + dataFile := filepath.Join(dir, "data.bin"+tc.leaf.Ext) + tc.leaf.DataFile = dataFile + + if tc.wantTotal == 0 && tc.fileContent == nil && tc.wantErr == nil && + tc.leaf.FormatVersion != archive.SnapshotFormatVersionPayloadSizes { + // Legacy paths that must actually measure something: synthesize a valid + // encoded (or raw) payload whose decoded length becomes the expectation. + plain := []byte("legacy-measured-block-payload-bytes") + + if tc.leaf.Ext == "" { + tc.fileContent = plain + } else { + codec, err := compress.New(codecNameForExt(tc.leaf.Ext), 0) + if err != nil { + t.Fatalf("compress.New: %v", err) + } - if tc.fileContent != nil { - if err := os.WriteFile(dataFile, tc.fileContent, 0o600); err != nil { - t.Fatalf("write %s: %v", dataFile, err) + var buf bytes.Buffer + if err := codec.EncodeFrameStream(&buf, bytes.NewReader(plain), int64(len(plain))); err != nil { + t.Fatalf("EncodeFrameStream: %v", err) + } + + tc.fileContent = buf.Bytes() } + + tc.wantTotal = int64(len(plain)) } - got, err := blockTotalSize(dataFile, tc.size, tc.ext) + var source io.ReadSeeker - if tc.wantErrNil { - if err != nil { - t.Fatalf("unexpected error: %v", err) + switch { + case tc.leaf.Ext != "" && tc.leaf.FormatVersion >= archive.SnapshotFormatVersionPayloadSizes: + // Fast path: prove no I/O happens at all by handing over a poisoned source + // (or leaving DataFile pointed at a file that is never written). + source = poisonReadSeeker{t: t} + case tc.fileContent != nil: + if err := os.WriteFile(dataFile, tc.fileContent, 0o600); err != nil { + t.Fatalf("write %s: %v", dataFile, err) } - if got != tc.wantTotal { - t.Errorf("total = %d, want %d", got, tc.wantTotal) + if tc.useSource { + source = bytes.NewReader(tc.fileContent) + } + } + + got, err := resolveBlockPayloadSize(context.Background(), tc.leaf, source) + + if tc.wantErr != nil { + if !errors.Is(err, tc.wantErr) { + t.Fatalf("error = %v, want wrapping %v", err, tc.wantErr) } return } - if err == nil { - t.Fatal("expected error, got nil") + if err != nil { + t.Fatalf("unexpected error: %v", err) } - if tc.wantErr != nil && !errors.Is(err, tc.wantErr) { - t.Errorf("expected error wrapping %v, got: %v", tc.wantErr, err) + if got != tc.wantTotal { + t.Errorf("total = %d, want %d", got, tc.wantTotal) } }) } } +// codecNameForExt maps a compress.Codec.Ext-style extension back to its compress.New name, +// for building legacy-archive test fixtures directly from PlannedNode.Ext. +func codecNameForExt(ext string) string { + switch ext { + case ".zst": + return "zstd" + case ".gz": + return "gzip" + case ".lz4": + return "lz4" + default: + return "none" + } +} + // noHTTPDoer fails the test immediately if HTTPDo is ever called. It is used // to prove that a preflight failure (invalid/mismatched captured size) sends // zero HTTP requests -- the check must run strictly before any HEAD/PUT. @@ -5713,10 +5822,10 @@ func (d noHTTPDoer) HTTPDo(_ *http.Request) (*http.Response, error) { return nil, nil } -// TestSendVolumeData_Block_RawSizeMismatch_SendsNoHTTP verifies that a raw -// (codec none) block leaf whose on-disk data.bin size disagrees with its -// captured VolumeInfo.Size fails deterministically via blockTotalSize and -// never issues a single HTTP request (no HEAD, no PUT). +// TestSendVolumeData_Block_RawSizeMismatch_SendsNoHTTP verifies that a raw (codec none) +// block leaf from a v3 archive, whose on-disk data.bin size disagrees with its recorded +// PayloadStoredSizeBytes, fails deterministically via resolveBlockPayloadSize before any +// HTTP request (no HEAD, no PUT). func TestSendVolumeData_Block_RawSizeMismatch_SendsNoHTTP(t *testing.T) { dir := t.TempDir() dataFile := filepath.Join(dir, "data.bin") @@ -5726,12 +5835,13 @@ func TestSendVolumeData_Block_RawSizeMismatch_SendsNoHTTP(t *testing.T) { } leaf := PlannedNode{ - APIVersion: "snapshot.storage.k8s.io/v1", - Kind: "VolumeSnapshot", - Name: "pvc-1", - DataFile: dataFile, - Ext: "", - Size: "10", // disagrees with the 5-byte file actually on disk + APIVersion: "snapshot.storage.k8s.io/v1", + Kind: "VolumeSnapshot", + Name: "pvc-1", + DataFile: dataFile, + Ext: "", + FormatVersion: archive.SnapshotFormatVersionPayloadSizes, + PayloadStoredSizeBytes: 10, // disagrees with the 5-byte file actually on disk } imp := &clusterVolumeImporter{log: discardLogger()} @@ -5746,33 +5856,229 @@ func TestSendVolumeData_Block_RawSizeMismatch_SendsNoHTTP(t *testing.T) { } } -// TestSendVolumeData_Block_InvalidSize_SendsNoHTTP verifies that a block leaf -// with a missing/unparsable captured size fails before any HTTP request, -// for every codec (raw and compressed alike). -func TestSendVolumeData_Block_InvalidSize_SendsNoHTTP(t *testing.T) { +// TestSendVolumeData_Block_NominalSizeMismatchDoesNotFail is the regression test for the live +// bug: a thin-provisioning backend rounds the device up from the nominal captured size ("1Ki" +// below, standing in for 1Gi/1073741824), so the real payload (2900 bytes here, standing in for +// 1077665792) exceeds it. Before this fix, upload trusted nominal Size as totalSize and failed; +// now resolveBlockPayloadSize reads the archive's measured PayloadRawSizeBytes instead, so +// upload succeeds and setTotal reports the TRUE decoded size. +func TestSendVolumeData_Block_NominalSizeMismatchDoesNotFail(t *testing.T) { + t.Parallel() + + // Deliberately not a round number and deliberately larger than the nominal size below, + // standing in for a thin-provisioning device round-up. + payload := bytes.Repeat([]byte("thin-provisioned-block-bytes-"), 100) + + dir := t.TempDir() + dataFile := filepath.Join(dir, "data.bin.zst") + + writeEncodedBlockFile(t, dataFile, "zstd", payload) + + leaf := PlannedNode{ + APIVersion: "snapshot.storage.k8s.io/v1", + Kind: "VolumeSnapshot", + Name: "pvc-1", + DataFile: dataFile, + Ext: ".zst", + FormatVersion: archive.SnapshotFormatVersionPayloadSizes, + PayloadRawSizeBytes: int64(len(payload)), + // Nominal size, deliberately smaller than the real payload above (mirrors the live + // bug) — resolveBlockPayloadSize must never consult this field. + Size: "1Ki", + } + + imp := &fakeBlockImporter{} + srv := httptest.NewServer(imp) + t.Cleanup(srv.Close) + + var totals []int64 + + setTotal := func(n int64) { totals = append(totals, n) } + + importer := &clusterVolumeImporter{log: discardLogger()} + + err := importer.sendVolumeData(context.Background(), plainHTTPDoer{}, srv.URL, volumeModeBlock, leaf, targetNS, "pvc-1", setTotal, nil, nil) + if err != nil { + t.Fatalf("sendVolumeData must succeed despite the nominal/real size mismatch: %v", err) + } + + if got := imp.received(); !bytes.Equal(got, payload) { + t.Fatalf("server received %d bytes not matching the original %d-byte payload", len(got), len(payload)) + } + + if want := []int64{int64(len(payload))}; len(totals) != 1 || totals[0] != want[0] { + t.Errorf("setTotal calls = %v, want a single call with %v (the TRUE decoded size, not the nominal 1Ki)", totals, want) + } + + nominalBytes := int64(1024) + if len(totals) == 1 && totals[0] == nominalBytes { + t.Errorf("setTotal was called with the nominal size %d, not the measured payload size %d", nominalBytes, len(payload)) + } +} + +// TestHeadBlockOffset_DeviceSize covers verifyDeviceCapacity's fail-open/fail-closed +// contract as exercised through headBlockOffset's 200 OK branch. +func TestHeadBlockOffset_DeviceSize(t *testing.T) { + t.Parallel() + + const totalSize = int64(100) + + tests := []struct { + name string + deviceSizeHeader string + wantErr bool + }{ + {name: "success: header absent fails open", deviceSizeHeader: ""}, + {name: "success: device size equal to payload", deviceSizeHeader: "100"}, + {name: "success: device size larger than payload", deviceSizeHeader: "200"}, + {name: "error: device size smaller than payload", deviceSizeHeader: "50", wantErr: true}, + {name: "success: unparsable header fails open", deviceSizeHeader: "not-a-number"}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + doer := testHTTPDoer(func(_ *http.Request) (*http.Response, error) { + header := http.Header{} + header.Set("X-Next-Offset", "0") + + if tc.deviceSizeHeader != "" { + header.Set("X-Device-Size", tc.deviceSizeHeader) + } + + return &http.Response{ + StatusCode: http.StatusOK, + Status: "200 OK", + Header: header, + Body: io.NopCloser(bytes.NewReader(nil)), + }, nil + }) + + _, err := headBlockOffset(context.Background(), doer, "https://importer.local/block", totalSize) + + if tc.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + + if !strings.Contains(err.Error(), "50") || !strings.Contains(err.Error(), "100") { + t.Errorf("error should mention both device size and payload size, got: %v", err) + } + + return + } + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + }) + } +} + +// TestHeadBlockOffset_DeviceSize_SendsNoPUTOnShortfall proves the device-capacity check runs +// strictly before the first PUT: a doer that would fail the test on any PUT call never sees one. +func TestHeadBlockOffset_DeviceSize_SendsNoPUTOnShortfall(t *testing.T) { + t.Parallel() + + doer := testHTTPDoer(func(req *http.Request) (*http.Response, error) { + if req.Method == http.MethodPut { + t.Fatal("unexpected PUT: the device-capacity check must fail before any PUT is attempted") + } + + header := http.Header{} + header.Set("X-Next-Offset", "0") + header.Set("X-Device-Size", "5") + + return &http.Response{ + StatusCode: http.StatusOK, + Status: "200 OK", + Header: header, + Body: io.NopCloser(bytes.NewReader(nil)), + }, nil + }) + + err := putBlock(context.Background(), doer, "https://importer.local/api/v1/block", "/nonexistent/data.bin", "", 100, discardLogger(), nil, nil) + if err == nil { + t.Fatal("expected error for undersized target device, got nil") + } +} + +// TestEnsureDataImport_ResumeAcrossPayloadSizeFieldsAddition proves resume/dedup is unaffected: +// a DataImport created by a pre-fix binary (no PayloadRawSizeBytes/PayloadStoredSizeBytes/ +// FormatVersion anywhere) must still be recognised and reused when EnsureDataImport re-plans +// the same leaf with a fixed binary that now carries those fields. +func TestEnsureDataImport_ResumeAcrossPayloadSizeFieldsAddition(t *testing.T) { + leafOld := volumeSnapshotLeaf("pvc-1") + existing := dataImportObjForLeaf(targetNS, leafOld, false) + + leafNew := leafOld + leafNew.PayloadRawSizeBytes = 1077665792 + leafNew.PayloadStoredSizeBytes = 900000000 + leafNew.FormatVersion = archive.SnapshotFormatVersionPayloadSizes + + if dataImportIdentity(leafNew) != dataImportIdentity(leafOld) { + t.Fatal("dataImportIdentity must not incorporate PayloadRawSizeBytes/PayloadStoredSizeBytes/FormatVersion") + } + + dyn := newFakeDataImportDyn(existing) + imp := newTestVolumeImporter(dyn) + + name, err := imp.EnsureDataImport(context.Background(), leafNew, targetNS) + if err != nil { + t.Fatalf("EnsureDataImport must still match the pre-fix DataImport by its unaffected identity: %v", err) + } + + if want := imp.DataImportName(leafNew); name != want { + t.Errorf("name = %q, want %q", name, want) + } + + if want := imp.DataImportName(leafOld); name != want { + t.Errorf("DataImportName must be identity-stable across the payload-size fields addition: got %q, want %q", name, want) + } + + if c := countDataImportActions(dyn, "create"); c != 0 { + t.Errorf("a pre-fix DataImport must be reused, not recreated (creates=%d)", c) + } + + if c := countDataImportActions(dyn, "delete"); c != 0 { + t.Errorf("a pre-fix DataImport must not be deleted (deletes=%d)", c) + } +} + +// TestSendVolumeData_Block_CorruptCompressedPayload_SendsNoHTTP verifies that a legacy-archive +// compressed block leaf whose on-disk bytes aren't a valid frame fails during +// resolveBlockPayloadSize's measurement pass, before any HEAD/PUT. A legacy archive's +// compressed payload is always measured from the bytes themselves, so garbage bytes are the +// failure mode caught here (replaces the old missing/unparsable-Size preflight). +func TestSendVolumeData_Block_CorruptCompressedPayload_SendsNoHTTP(t *testing.T) { for _, tc := range blockCodecCases { + if tc.ext == "" { + continue // the raw path has its own dedicated mismatch test above. + } + t.Run(tc.codec, func(t *testing.T) { dir := t.TempDir() dataFile := filepath.Join(dir, "data.bin"+tc.ext) - if err := os.WriteFile(dataFile, []byte("irrelevant"), 0o600); err != nil { + if err := os.WriteFile(dataFile, []byte("not a valid "+tc.codec+" frame"), 0o600); err != nil { t.Fatalf("write %s: %v", dataFile, err) } leaf := PlannedNode{ - APIVersion: "snapshot.storage.k8s.io/v1", - Kind: "VolumeSnapshot", - Name: "pvc-1", - DataFile: dataFile, - Ext: tc.ext, - Size: "", // missing captured size + APIVersion: "snapshot.storage.k8s.io/v1", + Kind: "VolumeSnapshot", + Name: "pvc-1", + DataFile: dataFile, + Ext: tc.ext, + FormatVersion: archive.SnapshotFormatVersionAuthenticatedChildren, // legacy: always measured } imp := &clusterVolumeImporter{log: discardLogger()} err := imp.sendVolumeData(context.Background(), noHTTPDoer{t: t}, "https://importer.local", volumeModeBlock, leaf, targetNS, "pvc-1", nil, nil, nil) if err == nil { - t.Fatal("expected error for missing captured size, got nil") + t.Fatal("expected error for corrupt compressed payload, got nil") } }) } diff --git a/internal/snapshot/volume/fs.go b/internal/snapshot/volume/fs.go index 8cf2312a..efa2a28f 100644 --- a/internal/snapshot/volume/fs.go +++ b/internal/snapshot/volume/fs.go @@ -1544,9 +1544,12 @@ func defaultPortForScheme(scheme string) string { // and re-hashed via verifyStagedFileMD5 before the skip is trusted; on a // mismatch the bad blob is removed and staging falls through to re-fetch it in // this same run (a self-healing condition, not a hard error). When no MD5 is -// advertised the blob is still skipped, matching the fresh-path convention, -// with a one-line WARN. The verify costs one decode pass per already-staged -// file per resume run, bounded by staging size — the price of not trusting +// advertised, full content verification is impossible, but the already-staged +// blob is still decoded to measure its plaintext size via stagedFileRawSize, +// which is compared against the item's declared size below; a mismatch drives +// the same self-healing re-fetch as an MD5 mismatch, with a one-line WARN. +// The verify costs one decode pass per already-staged file per resume run, +// bounded by staging size — the price of not trusting // bytes we did not just write. A trusted skip still credits the item's // declared size to onProgress so the numerator can reach the denominator that // setTotal established from the inventory total — otherwise @@ -1573,14 +1576,10 @@ func stageCompressedFile( ) if item.md5 == "" { - log.Warn("no source MD5 available for file, skipping integrity verification", + log.Warn("no source MD5 available for file, verifying size only", slog.String("path", item.relPath)) - if item.size >= 0 { - rawSize = item.size - } else { - rawSize, verifyErr = stagedFileRawSize(view, destPath, codec.Ext()) - } + rawSize, verifyErr = stagedFileRawSize(view, destPath, codec.Ext()) } else { rawSize, verifyErr = verifyStagedFileMD5(view, destPath, codec.Ext(), item.md5) } diff --git a/internal/snapshot/volume/fs_test.go b/internal/snapshot/volume/fs_test.go index 716ef68f..c6d9bf7d 100644 --- a/internal/snapshot/volume/fs_test.go +++ b/internal/snapshot/volume/fs_test.go @@ -2669,15 +2669,15 @@ func TestDownloadFilesystemVolume_ResumeSkip_MismatchedBlobRestaged(t *testing.T } } -// TestDownloadFilesystemVolume_ResumeSkip_EmptyMD5SkipsWithWarn verifies that -// an already-staged blob for a listing item with no hash.md5 attribute is -// skipped WITHOUT verification (matching the fresh-path convention): the blob -// is not re-downloaded even though its bytes differ from the server's, its -// declared size is credited once, and a single WARN is logged. -func TestDownloadFilesystemVolume_ResumeSkip_EmptyMD5SkipsWithWarn(t *testing.T) { +// TestDownloadFilesystemVolume_ResumeSkip_EmptyMD5_SizeMatches verifies that, with no +// source MD5, an already-staged blob whose measured raw size matches the fresh listing's +// declared size is skipped without content verification: no file GET, the sentinel content +// survives in the tar, a single "verifying size only" WARN fires, and the size is credited +// to onProgress once. +func TestDownloadFilesystemVolume_ResumeSkip_EmptyMD5_SizeMatches(t *testing.T) { t.Parallel() - content := []byte("server content that must never be fetched on an empty-md5 skip") + content := []byte("server content that must never be fetched on a size-only skip!") codec := mustCodec(t, "none") srv, getCount := newFileGetCountingFSServer(t, "file.bin", content, "") @@ -2689,9 +2689,9 @@ func TestDownloadFilesystemVolume_ResumeSkip_EmptyMD5SkipsWithWarn(t *testing.T) t.Fatal(err) } - // Sentinel differs from the server content: with no advertised MD5 the skip - // branch must NOT verify it and must NOT re-download it. - sentinel := []byte("sentinel-not-server-content") + // Sentinel is the SAME length as the server content but differs byte for byte: content + // still can't be verified without MD5, but the size matches so the skip stands. + sentinel := bytes.Repeat([]byte("X"), len(content)) if err := os.WriteFile(filepath.Join(stagingDir, "file.bin"), sentinel, 0o644); err != nil { t.Fatal(err) } @@ -2718,24 +2718,24 @@ func TestDownloadFilesystemVolume_ResumeSkip_EmptyMD5SkipsWithWarn(t *testing.T) } if getCount() != 0 { - t.Errorf("empty-md5 staged file was re-downloaded: %d file GET(s), want 0", getCount()) + t.Errorf("size-matched empty-md5 staged file was re-downloaded: %d file GET(s), want 0", getCount()) } warnCount := 0 for _, msg := range lh.warnMessages() { - if msg == "no source MD5 available for file, skipping integrity verification" { + if msg == "no source MD5 available for file, verifying size only" { warnCount++ } } if warnCount != 1 { - t.Errorf("expected exactly 1 missing-digest WARN, got %d: %v", warnCount, lh.warnMessages()) + t.Errorf("expected exactly 1 verifying-size-only WARN, got %d: %v", warnCount, lh.warnMessages()) } entries := readTarContents(t, tarPath) if !bytes.Equal(entries["file.bin"], sentinel) { - t.Errorf("file.bin content = %q; want sentinel %q (skipped without verification)", entries["file.bin"], sentinel) + t.Errorf("file.bin content = %q; want sentinel %q (skipped, content unverifiable without MD5)", entries["file.bin"], sentinel) } if credited != int64(len(content)) { @@ -2743,6 +2743,94 @@ func TestDownloadFilesystemVolume_ResumeSkip_EmptyMD5SkipsWithWarn(t *testing.T) } } +// TestDownloadFilesystemVolume_ResumeSkip_EmptyMD5_SizeDiffers verifies that, with no +// source MD5, an already-staged blob whose measured size does NOT match the fresh listing +// is treated as stale and re-staged in the same run: at least one file GET fires, the tar +// carries the true server content, the "re-staging" WARN fires, progress reaches the +// declared size, and the resulting tar's PAX metadata stays consistent (SumTarRawSizes). +func TestDownloadFilesystemVolume_ResumeSkip_EmptyMD5_SizeDiffers(t *testing.T) { + t.Parallel() + + content := []byte("resume-skip true source content for the size-mismatched blob") + codec := mustCodec(t, "none") + srv, getCount := newFileGetCountingFSServer(t, "file.bin", content, "") + + nodeDir := t.TempDir() + tarPath := filepath.Join(nodeDir, archive.FsTarName) + stagingDir := filepath.Join(nodeDir, archive.FsTarStagingDirName) + + if err := os.MkdirAll(stagingDir, 0o755); err != nil { + t.Fatal(err) + } + + // Sentinel differs in LENGTH from the server content: with no MD5, the skip branch + // still catches this via the size check and re-fetches rather than trusting it. + sentinel := []byte("short-stale-blob") + if err := os.WriteFile(filepath.Join(stagingDir, "file.bin"), sentinel, 0o644); err != nil { + t.Fatal(err) + } + + lh := &warnCapture{} + log := slog.New(lh) + + var ( + progMu sync.Mutex + credited int64 + ) + + onProgress := func(n int) { + progMu.Lock() + credited += int64(n) + progMu.Unlock() + } + + if err := volume.DownloadFilesystemVolume( + context.Background(), log, tarPath, stagingDir, srv.URL+"/files/", + 1, 0, newFSFetcher(srv), codec, nil, onProgress, + ); err != nil { + t.Fatalf("DownloadFilesystemVolume: %v", err) + } + + if getCount() < 1 { + t.Errorf("size-mismatched empty-md5 staged file was not re-downloaded: %d file GET(s), want >= 1", getCount()) + } + + warnCount := 0 + + for _, msg := range lh.warnMessages() { + if msg == "staged file failed source MD5 re-check on resume, re-staging" { + warnCount++ + } + } + + if warnCount != 1 { + t.Errorf("expected exactly 1 re-staging WARN, got %d: %v", warnCount, lh.warnMessages()) + } + + entries := readTarContents(t, tarPath) + if !bytes.Equal(entries["file.bin"], content) { + t.Errorf("file.bin content = %q; want true source content %q (re-staged)", entries["file.bin"], content) + } + + if credited != int64(len(content)) { + t.Errorf("onProgress credited = %d; want %d (declared size once)", credited, len(content)) + } + + // The tar's PAX rawSize records must be honest: ParseFSMetadata rejects an entry whose + // stored byte count disagrees with its declared rawSize. A stale skip that never + // re-fetched would leave rawSize describing the listing's size while the tar stored the + // shorter sentinel, tripping exactly this check downstream. + f, err := os.Open(tarPath) + if err != nil { + t.Fatalf("open tar %s: %v", tarPath, err) + } + t.Cleanup(func() { _ = f.Close() }) + + if _, err := archive.SumTarRawSizes(context.Background(), f); err != nil { + t.Errorf("SumTarRawSizes: %v", err) + } +} + // ── path sanitization (sanitize-server-provided-paths) ───────────────────── // singleItemFSServer builds an httptest.Server exposing a one-item filesystem listing diff --git a/internal/snapshot/volume/manifest_worker.go b/internal/snapshot/volume/manifest_worker.go index 81a9e6e4..f130a524 100644 --- a/internal/snapshot/volume/manifest_worker.go +++ b/internal/snapshot/volume/manifest_worker.go @@ -267,6 +267,16 @@ func finalizeNodeWithChecksum( return fmt.Errorf("compute children checksum for %s/%s: %w", node.Kind, node.Name, err) } + // The recorded payload size cannot be threaded through in memory from wherever the + // volume was downloaded: this call may finalize a node that downloaded nothing in + // THIS run (a crash-resume or a re-publication triggered by a re-published child), so + // the only value correct in every case is measured fresh from the bytes already on + // disk, here, at finalize time. + payload, err := MeasurePayload(ctx, destination, nodeDir) + if err != nil { + return fmt.Errorf("measure payload for %s/%s: %w", node.Kind, node.Name, err) + } + sy := archive.SnapshotYAML{ APIVersion: node.APIVersion, Kind: node.Kind, @@ -277,7 +287,7 @@ func finalizeNodeWithChecksum( SourceObjectRef: buildSourceObjectRef(node.SourceRef), Checksum: checksum, ChildrenChecksum: &childrenChecksum, - Volumes: buildVolumesList(node), + Volumes: buildVolumesList(node, payload), } if err := writeSnapshotYAML(ctx, destination, nodeDir, sy); err != nil { @@ -327,18 +337,19 @@ func removePath(destination *archive.RootedDestination, path string) error { // buildVolumesList constructs the Volumes list for snapshot.yaml from a node. // Returns nil (omitted) when the node captured no volume. -func buildVolumesList(node *source.Node) []archive.VolumeInfo { +func buildVolumesList(node *source.Node, payload PayloadSize) []archive.VolumeInfo { if node.Data == nil { return nil } - return []archive.VolumeInfo{nodeDataToVolumeInfo(node.Data)} + return []archive.VolumeInfo{nodeDataToVolumeInfo(node.Data, payload)} } // nodeDataToVolumeInfo converts a namespaced status.data descriptor to a VolumeInfo. The // volume metadata (volumeMode/storageClassName/size) is carried through so the import side // can rebuild the DataImport spec for a re-import without re-reading the live cluster state. -func nodeDataToVolumeInfo(d *source.NodeData) archive.VolumeInfo { +// payload is the measured on-disk payload footprint (see MeasurePayload). +func nodeDataToVolumeInfo(d *source.NodeData, payload PayloadSize) archive.VolumeInfo { return archive.VolumeInfo{ Target: archive.VolumeObjectRef{ APIVersion: d.SourceRef.APIVersion, @@ -355,6 +366,8 @@ func nodeDataToVolumeInfo(d *source.NodeData) archive.VolumeInfo { VolumeMode: d.VolumeMode, StorageClassName: d.StorageClassName, Size: d.Size, + RawSizeBytes: payload.RawBytes, + StoredSizeBytes: payload.StoredBytes, } } diff --git a/internal/snapshot/volume/manifest_worker_test.go b/internal/snapshot/volume/manifest_worker_test.go index 0d83f1f0..85e975c2 100644 --- a/internal/snapshot/volume/manifest_worker_test.go +++ b/internal/snapshot/volume/manifest_worker_test.go @@ -17,6 +17,7 @@ limitations under the License. package volume_test import ( + "bytes" "context" "errors" "os" @@ -958,3 +959,168 @@ func TestFinalizeNode_NoVolumesOmitted(t *testing.T) { t.Errorf("VerifyNode: %v", err) } } + +// TestFinalizeNode_RecordsPayloadSizes_BlockZstd is the regression test for the live bug: +// nominal Size ("1Gi", standing in for a thin-provisioning round-up) disagrees with the real +// decoded zstd payload size. FinalizeNode must leave Size untouched while recording +// RawSizeBytes/StoredSizeBytes measured from the real payload, not derived from Size. +func TestFinalizeNode_RecordsPayloadSizes_BlockZstd(t *testing.T) { + t.Parallel() + + nodeDir := setupNodeDir(t) + + if err := archive.WriteManifest(nodeDir, makeObjWithUID("v1", "PersistentVolumeClaim", "pvc-thin", "uid-thin")); err != nil { + t.Fatalf("WriteManifest: %v", err) + } + + // Deliberately larger than the nominal "1Gi" below, standing in for the real bug's + // 1077665792-vs-1073741824 mismatch. + payload := bytes.Repeat([]byte("thin-provisioned-real-payload-bytes-"), 500) + storedBytes := writeZstdBlockPayload(t, nodeDir, payload) + + data := &source.NodeData{ + SourceRef: source.SourceRefIdentity{ + APIVersion: "v1", + Kind: "PersistentVolumeClaim", + Name: "pvc-thin", + Namespace: "ns", + UID: "uid-thin", + }, + ArtifactRef: source.ArtifactRef{ + APIVersion: "snapshot.storage.k8s.io/v1", + Kind: "VolumeSnapshotContent", + Name: "vsc-thin", + }, + VolumeMode: "Block", + StorageClassName: "linstor-thin-r1", + Size: "1Gi", // nominal, deliberately NOT equal to len(payload) + } + + node := &source.Node{ + APIVersion: "snapshot.storage.k8s.io/v1", + Kind: "VolumeSnapshot", + Name: "d8-ss-thin", + Namespace: "ns", + UID: "uid-vs-thin", + Data: data, + } + + if err := volume.FinalizeNode(nodeDir, node); err != nil { + t.Fatalf("FinalizeNode: %v", err) + } + + sy, err := archive.ReadSnapshotYAML(nodeDir) + if err != nil { + t.Fatalf("ReadSnapshotYAML: %v", err) + } + + if len(sy.Volumes) != 1 { + t.Fatalf("Volumes length: got %d, want 1", len(sy.Volumes)) + } + + vol := sy.Volumes[0] + + // The nominal size is NOT touched by this fix. + if vol.Size != "1Gi" { + t.Errorf("Size = %q, want unchanged %q", vol.Size, "1Gi") + } + + // The recorded payload sizes reflect the REAL bytes on disk, not the nominal size. + if vol.RawSizeBytes != int64(len(payload)) { + t.Errorf("RawSizeBytes = %d, want %d (the real decoded payload size)", vol.RawSizeBytes, len(payload)) + } + + if vol.StoredSizeBytes != storedBytes { + t.Errorf("StoredSizeBytes = %d, want %d (the real on-disk compressed size)", vol.StoredSizeBytes, storedBytes) + } + + nominalBytes := int64(1024 * 1024 * 1024) + if vol.RawSizeBytes == nominalBytes { + t.Error("RawSizeBytes must not equal the nominal 1Gi size — that is the exact bug this fix addresses") + } + + if err := archive.VerifyNode(nodeDir); err != nil { + t.Errorf("VerifyNode: %v", err) + } +} + +// TestFinalizeNode_RefinalizeDoneNodeKeepsSizes proves that re-finalizing an already-published +// node (no new download this run) still measures and records payload sizes correctly — +// MeasurePayload reads fresh from disk every time, so a second finalize must not drop them. +func TestFinalizeNode_RefinalizeDoneNodeKeepsSizes(t *testing.T) { + t.Parallel() + + nodeDir := setupNodeDir(t) + + if err := archive.WriteManifest(nodeDir, makeObjWithUID("v1", "PersistentVolumeClaim", "pvc-r", "uid-r")); err != nil { + t.Fatalf("WriteManifest: %v", err) + } + + payload := bytes.Repeat([]byte("refinalize-payload-bytes-"), 300) + storedBytes := writeZstdBlockPayload(t, nodeDir, payload) + + data := &source.NodeData{ + SourceRef: source.SourceRefIdentity{ + APIVersion: "v1", + Kind: "PersistentVolumeClaim", + Name: "pvc-r", + Namespace: "ns", + UID: "uid-r", + }, + ArtifactRef: source.ArtifactRef{ + APIVersion: "snapshot.storage.k8s.io/v1", + Kind: "VolumeSnapshotContent", + Name: "vsc-r", + }, + VolumeMode: "Block", + StorageClassName: "sc-r", + Size: "2Gi", + } + + node := &source.Node{ + APIVersion: "snapshot.storage.k8s.io/v1", + Kind: "VolumeSnapshot", + Name: "d8-ss-refinalize", + Namespace: "ns", + UID: "uid-vs-refinalize", + Data: data, + } + + if err := volume.FinalizeNode(nodeDir, node); err != nil { + t.Fatalf("first FinalizeNode: %v", err) + } + + first, err := archive.ReadSnapshotYAML(nodeDir) + if err != nil { + t.Fatalf("ReadSnapshotYAML after first finalize: %v", err) + } + + // Second finalize simulates a re-publication with no new download in this run: the + // payload bytes on disk are unchanged, so the re-measured sizes must be identical. + if err := volume.FinalizeNode(nodeDir, node); err != nil { + t.Fatalf("second FinalizeNode: %v", err) + } + + second, err := archive.ReadSnapshotYAML(nodeDir) + if err != nil { + t.Fatalf("ReadSnapshotYAML after second finalize: %v", err) + } + + if len(first.Volumes) != 1 || len(second.Volumes) != 1 { + t.Fatalf("Volumes length: first=%d second=%d, want 1/1", len(first.Volumes), len(second.Volumes)) + } + + if first.Volumes[0].RawSizeBytes != int64(len(payload)) || first.Volumes[0].RawSizeBytes != second.Volumes[0].RawSizeBytes { + t.Errorf("RawSizeBytes not preserved across refinalize: first=%d second=%d, want %d", + first.Volumes[0].RawSizeBytes, second.Volumes[0].RawSizeBytes, len(payload)) + } + + if first.Volumes[0].StoredSizeBytes != storedBytes || first.Volumes[0].StoredSizeBytes != second.Volumes[0].StoredSizeBytes { + t.Errorf("StoredSizeBytes not preserved across refinalize: first=%d second=%d, want %d", + first.Volumes[0].StoredSizeBytes, second.Volumes[0].StoredSizeBytes, storedBytes) + } + + if err := archive.VerifyNode(nodeDir); err != nil { + t.Errorf("VerifyNode after refinalize: %v", err) + } +} diff --git a/internal/snapshot/volume/payload_size.go b/internal/snapshot/volume/payload_size.go new file mode 100644 index 00000000..a9bf7b68 --- /dev/null +++ b/internal/snapshot/volume/payload_size.go @@ -0,0 +1,224 @@ +/* +Copyright 2026 Flant JSC + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package volume + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "path/filepath" + + "github.com/deckhouse/deckhouse-cli/internal/snapshot/archive" + "github.com/deckhouse/deckhouse-cli/internal/snapshot/compress" +) + +// Payload mode strings recorded in PayloadSize.Mode, matching the sibling +// snapimport.dataImportPayloadBlock/dataImportPayloadFilesystem naming. +const ( + payloadModeBlock = "block" + payloadModeFilesystem = "filesystem" +) + +// PayloadSize is the measured byte footprint of a snapshot node's captured volume +// payload, as it actually exists on disk — as opposed to VolumeInfo.Size, which is the +// nominal PVC quantity used to provision a scratch volume on re-import. +type PayloadSize struct { + // RawBytes is the exact decoded (plaintext) byte length of the payload: for a Block + // volume, data.bin[.]'s decoded length; for a Filesystem volume, the sum of + // data.tar's regular-entry raw sizes. + RawBytes int64 + // StoredBytes is the on-disk byte length of the payload artifact itself + // (data.bin[.] or data.tar). + StoredBytes int64 + // Mode is payloadModeBlock or payloadModeFilesystem for a node that carries a volume + // payload, "" for a node with no payload (aggregator node). + Mode string +} + +// MeasurePayload determines the exact raw and stored byte sizes of a snapshot node's +// captured payload by reading it from disk: for a Block volume, it decodes (or, for +// zstd, reads the frame headers of) data.bin[.]; for a Filesystem volume, it sums +// data.tar's regular-entry raw sizes and stats the tar file itself for StoredBytes. +// +// destination is nil for a plain-filesystem archive and non-nil for one opened through a +// locked rooted view (see archive.RootedDestination); both read the identical on-disk +// layout, only through different I/O primitives. Returns a zero PayloadSize with +// Mode == "" for a node with no payload (aggregator node) — not an error. +func MeasurePayload(ctx context.Context, destination *archive.RootedDestination, nodeDir string) (PayloadSize, error) { + blockPayload, hasBlock, err := classifyBlockPayload(destination, nodeDir) + if err != nil { + return PayloadSize{}, fmt.Errorf("classify block payload in %s: %w", nodeDir, err) + } + + if hasBlock { + size, measureErr := measureBlockPayload(ctx, destination, blockPayload) + if measureErr != nil { + return PayloadSize{}, fmt.Errorf("measure block payload %s: %w", blockPayload.Path, measureErr) + } + + return size, nil + } + + tarPath := filepath.Join(nodeDir, archive.FsTarName) + + hasTar, err := payloadFileExists(destination, tarPath) + if err != nil { + return PayloadSize{}, fmt.Errorf("inspect %s: %w", tarPath, err) + } + + if !hasTar { + return PayloadSize{}, nil + } + + size, err := measureFSPayload(ctx, destination, tarPath) + if err != nil { + return PayloadSize{}, fmt.Errorf("measure filesystem payload %s: %w", tarPath, err) + } + + return size, nil +} + +// classifyBlockPayload resolves nodeDir's block payload (see archive.ClassifyBlockPayload), +// through destination's locked view when set. +func classifyBlockPayload(destination *archive.RootedDestination, nodeDir string) (archive.BlockPayload, bool, error) { + if destination == nil { + return archive.ClassifyBlockPayload(nodeDir) + } + + return destination.FindBlockData(nodeDir) +} + +// payloadFileExists reports whether path exists, through destination's locked view when set. +func payloadFileExists(destination *archive.RootedDestination, path string) (bool, error) { + var err error + + if destination == nil { + _, err = os.Stat(path) + } else { + _, err = destination.Stat(path) + } + + if err == nil { + return true, nil + } + + if errors.Is(err, os.ErrNotExist) { + return false, nil + } + + return false, err +} + +// openPayloadFile opens path for reading, through destination's locked view when set. +func openPayloadFile(destination *archive.RootedDestination, path string) (*os.File, error) { + if destination == nil { + return os.Open(path) + } + + return destination.OpenRegular(path) +} + +// measureBlockPayload opens and measures one node's block-volume payload file. +func measureBlockPayload( + ctx context.Context, + destination *archive.RootedDestination, + payload archive.BlockPayload, +) (PayloadSize, error) { + file, err := openPayloadFile(destination, payload.Path) + if err != nil { + return PayloadSize{}, err + } + + size, measureErr := measureOpenBlockPayload(ctx, payload.Ext, file) + closeErr := file.Close() + + if err := errors.Join(measureErr, closeErr); err != nil { + return PayloadSize{}, err + } + + return size, nil +} + +// measureOpenBlockPayload measures an already-open block payload file. ext is the +// classified codec extension (see archive.BlockPayload.Ext) — callers MUST pass that +// value rather than re-deriving it from the filename, for the same reason +// archive.BlockPayload.Ext's doc comment gives. +func measureOpenBlockPayload(ctx context.Context, ext string, file *os.File) (PayloadSize, error) { + info, err := file.Stat() + if err != nil { + return PayloadSize{}, fmt.Errorf("stat block payload: %w", err) + } + + section := io.NewSectionReader(file, 0, info.Size()) + + rawBytes, err := compress.DecodedSize(ctx, ext, section) + if err != nil { + return PayloadSize{}, fmt.Errorf("decode block payload size: %w", err) + } + + return PayloadSize{ + RawBytes: rawBytes, + StoredBytes: info.Size(), + Mode: payloadModeBlock, + }, nil +} + +// measureFSPayload opens and measures one node's filesystem-volume payload file. +func measureFSPayload( + ctx context.Context, + destination *archive.RootedDestination, + tarPath string, +) (PayloadSize, error) { + file, err := openPayloadFile(destination, tarPath) + if err != nil { + return PayloadSize{}, err + } + + size, measureErr := measureOpenFSPayload(ctx, file) + closeErr := file.Close() + + if err := errors.Join(measureErr, closeErr); err != nil { + return PayloadSize{}, err + } + + return size, nil +} + +// measureOpenFSPayload measures an already-open data.tar file: the container itself is +// never compressed as a whole (only individual entries are, per-entry), so its own +// on-disk length is StoredBytes and RawBytes is the sum of every entry's decoded size. +func measureOpenFSPayload(ctx context.Context, file *os.File) (PayloadSize, error) { + info, err := file.Stat() + if err != nil { + return PayloadSize{}, fmt.Errorf("stat filesystem payload: %w", err) + } + + section := io.NewSectionReader(file, 0, info.Size()) + + rawBytes, err := archive.SumTarRawSizes(ctx, section) + if err != nil { + return PayloadSize{}, fmt.Errorf("sum filesystem payload raw sizes: %w", err) + } + + return PayloadSize{ + RawBytes: rawBytes, + StoredBytes: info.Size(), + Mode: payloadModeFilesystem, + }, nil +} diff --git a/internal/snapshot/volume/payload_size_test.go b/internal/snapshot/volume/payload_size_test.go new file mode 100644 index 00000000..ce249118 --- /dev/null +++ b/internal/snapshot/volume/payload_size_test.go @@ -0,0 +1,226 @@ +/* +Copyright 2026 Flant JSC + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package volume_test + +import ( + gotar "archive/tar" + "bytes" + "context" + "os" + "path/filepath" + "testing" + + "github.com/deckhouse/deckhouse-cli/internal/snapshot/archive" + "github.com/deckhouse/deckhouse-cli/internal/snapshot/compress" + "github.com/deckhouse/deckhouse-cli/internal/snapshot/volume" +) + +// writeZstdBlockPayload encodes payload as a single content-size-bearing zstd frame and +// writes it to /data.bin.zst, returning the on-disk (compressed) byte length. +func writeZstdBlockPayload(t *testing.T, nodeDir string, payload []byte) int64 { + t.Helper() + + codec, err := compress.New("zstd", 0) + if err != nil { + t.Fatalf("compress.New(zstd): %v", err) + } + + var buf bytes.Buffer + if err := codec.EncodeFrameStream(&buf, bytes.NewReader(payload), int64(len(payload))); err != nil { + t.Fatalf("EncodeFrameStream: %v", err) + } + + path := filepath.Join(nodeDir, archive.DataBlockName(".zst")) + if err := os.WriteFile(path, buf.Bytes(), 0o600); err != nil { + t.Fatalf("write %s: %v", path, err) + } + + return int64(buf.Len()) +} + +// writeRawBlockPayload writes payload verbatim to /data.bin (codec "none"). +func writeRawBlockPayload(t *testing.T, nodeDir string, payload []byte) { + t.Helper() + + path := filepath.Join(nodeDir, archive.DataBlockName("")) + if err := os.WriteFile(path, payload, 0o600); err != nil { + t.Fatalf("write %s: %v", path, err) + } +} + +// writeFSPayload writes a single-entry data.tar (codec "none") with rawSize == len(content), +// returning the on-disk tar byte length. +func writeFSPayload(t *testing.T, nodeDir string, content []byte) int64 { + t.Helper() + + tarPath := filepath.Join(nodeDir, archive.FsTarName) + tarBytes := buildSingleEntryDataTar(t, content) + + if err := os.WriteFile(tarPath, tarBytes, 0o600); err != nil { + t.Fatalf("write %s: %v", tarPath, err) + } + + return int64(len(tarBytes)) +} + +func TestMeasurePayload(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + setup func(t *testing.T, nodeDir string) (wantRaw, wantStored int64) + wantMode string + wantErr bool + }{ + { + name: "success: block zstd payload", + setup: func(t *testing.T, nodeDir string) (int64, int64) { + payload := bytes.Repeat([]byte("block-zstd-payload-bytes-"), 200) + stored := writeZstdBlockPayload(t, nodeDir, payload) + + return int64(len(payload)), stored + }, + wantMode: "block", + }, + { + name: "success: block raw (none) payload", + setup: func(t *testing.T, nodeDir string) (int64, int64) { + payload := []byte("raw block payload bytes") + writeRawBlockPayload(t, nodeDir, payload) + + return int64(len(payload)), int64(len(payload)) + }, + wantMode: "block", + }, + { + name: "success: empty block payload is legitimate, not an error", + setup: func(t *testing.T, nodeDir string) (int64, int64) { + writeRawBlockPayload(t, nodeDir, nil) + + return 0, 0 + }, + wantMode: "block", + }, + { + name: "success: filesystem tar payload", + setup: func(t *testing.T, nodeDir string) (int64, int64) { + content := []byte("filesystem entry content bytes") + stored := writeFSPayload(t, nodeDir, content) + + return int64(len(content)), stored + }, + wantMode: "filesystem", + }, + { + name: "success: node with no payload (aggregator) returns zero values, no error", + setup: func(*testing.T, string) (int64, int64) { + return 0, 0 + }, + wantMode: "", + }, + { + name: "error: corrupted block payload (truncated zstd frame)", + setup: func(t *testing.T, nodeDir string) (int64, int64) { + path := filepath.Join(nodeDir, archive.DataBlockName(".zst")) + if err := os.WriteFile(path, []byte("not a valid zstd frame"), 0o600); err != nil { + t.Fatalf("write corrupt payload: %v", err) + } + + return 0, 0 + }, + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + nodeDir := t.TempDir() + wantRaw, wantStored := tt.setup(t, nodeDir) + + got, err := volume.MeasurePayload(context.Background(), nil, nodeDir) + + if tt.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + + return + } + + if err != nil { + t.Fatalf("MeasurePayload: %v", err) + } + + if got.RawBytes != wantRaw { + t.Errorf("RawBytes = %d, want %d", got.RawBytes, wantRaw) + } + + if got.StoredBytes != wantStored { + t.Errorf("StoredBytes = %d, want %d", got.StoredBytes, wantStored) + } + + if got.Mode != tt.wantMode { + t.Errorf("Mode = %q, want %q", got.Mode, tt.wantMode) + } + }) + } +} + +// buildSingleEntryDataTar builds a minimal data.tar with one regular "none"-codec entry +// carrying the required D8 PAX metadata (matching archive.FSMetadata's contract). +func buildSingleEntryDataTar(t *testing.T, content []byte) []byte { + t.Helper() + + metadata, err := archive.NewFSMetadata("none", "file.bin", int64(len(content))) + if err != nil { + t.Fatalf("NewFSMetadata: %v", err) + } + + storedPath, err := metadata.StoredPath() + if err != nil { + t.Fatalf("StoredPath: %v", err) + } + + var buf bytes.Buffer + + tw := gotar.NewWriter(&buf) + + hdr := &gotar.Header{ + Format: gotar.FormatPAX, + Typeflag: gotar.TypeReg, + Name: storedPath, + Mode: 0o600, + Size: int64(len(content)), + PAXRecords: metadata.PAXRecords(), + } + + if err := tw.WriteHeader(hdr); err != nil { + t.Fatalf("WriteHeader: %v", err) + } + + if _, err := tw.Write(content); err != nil { + t.Fatalf("Write: %v", err) + } + + if err := tw.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + return buf.Bytes() +} diff --git a/internal/snapshot/volume/progress_test.go b/internal/snapshot/volume/progress_test.go index 6673b51d..804b23a3 100644 --- a/internal/snapshot/volume/progress_test.go +++ b/internal/snapshot/volume/progress_test.go @@ -33,6 +33,7 @@ import ( "testing" "time" + "github.com/klauspost/compress/zstd" "github.com/stretchr/testify/require" "github.com/deckhouse/deckhouse-cli/internal/snapshot/archive" @@ -331,14 +332,14 @@ func TestDownloadFilesystemVolume_ResumeSkipReachesFullTotal(t *testing.T) { require.NoError(t, os.MkdirAll(stagingDir, 0o755)) - // Simulate a prior partial run: root.txt was already staged (compressed - // blob written under stagingDir) but data.tar was never assembled. - // The staged bytes need not be a valid zstd stream for this assertion — - // stageCompressedFile's skip branch never decodes them, it only checks - // for the destination file's existence (see - // TestDownloadFilesystemVolume_SkipsExistingCompressedStaged in - // fs_test.go, which relies on the same property). - sentinel := []byte("sentinel-not-server-content") + // Simulate a prior partial run: root.txt was staged but data.tar wasn't assembled. With + // no source MD5, the resume-skip branch verifies size only — the pre-staged blob must + // decode to a plaintext of the same length as the listing's declared size (12 bytes) for + // the skip to stand, even though its content differs from the real source bytes. + sentinelPlaintext := bytes.Repeat([]byte("X"), len(files[0].content)) + sentinel, err := codec.EncodeFrame(sentinelPlaintext) + require.NoError(t, err) + preStaged := filepath.Join(stagingDir, "root.txt"+codec.Ext()) require.NoError(t, os.WriteFile(preStaged, sentinel, 0o644)) @@ -350,7 +351,7 @@ func TestDownloadFilesystemVolume_ResumeSkipReachesFullTotal(t *testing.T) { var counter progressCounter - err := volume.DownloadFilesystemVolume( + err = volume.DownloadFilesystemVolume( context.Background(), slog.Default(), tarPath, @@ -371,8 +372,8 @@ func TestDownloadFilesystemVolume_ResumeSkipReachesFullTotal(t *testing.T) { "resume still reaches 100%%") // The skip must still have avoided re-download: the tar entry for the - // pre-staged file carries the sentinel bytes, not freshly downloaded - // content. + // pre-staged file decodes back to the sentinel plaintext, not the real + // server content. f, err := os.Open(tarPath) require.NoError(t, err) @@ -398,6 +399,15 @@ func TestDownloadFilesystemVolume_ResumeSkipReachesFullTotal(t *testing.T) { require.NoError(t, readErr) require.Equal(t, sentinel, got, "pre-staged file must not be re-downloaded") + decoded, decodeErr := zstd.NewReader(bytes.NewReader(got)) + require.NoError(t, decodeErr) + + plaintext, readErr := io.ReadAll(decoded) + decoded.Close() + require.NoError(t, readErr) + require.Equal(t, sentinelPlaintext, plaintext, + "pre-staged file must decode to the sentinel plaintext, not freshly downloaded content") + foundSentinel = true }