From 0debea9bfbc1b1308acd9397d644a8d92945586f Mon Sep 17 00:00:00 2001 From: Lev Brouk Date: Wed, 10 Jun 2026 10:51:14 -0700 Subject: [PATCH 1/3] refactor(storage): mark producer-side FrameTables as FullFrameTable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a FullFrameTable wrapper for the two narrow contexts where the FT covers a whole file: (a) fresh out of compressStream / CompressBytes / StoreFile / UploadFramed, and (b) the future StorageDiff fallback latch (#2919). Producers return *FullFrameTable; downcast via Table() to feed the hot read path or assign to BuildData.FrameData. FrameTable itself stays the authoritative type — what BuildData stores, what DeserializeFrameTable and TrimToRanges return, what ReadAt and OpenRangeReader accept. No on-disk format change, no behavior change. --- .../pkg/sandbox/block/streaming_chunk_test.go | 6 +- .../pkg/sandbox/build_upload_v4.go | 10 +-- .../sandbox/template/peerclient/seekable.go | 2 +- .../template/peerclient/seekable_test.go | 6 +- .../shared/pkg/storage/compress_encode.go | 2 +- .../pkg/storage/compress_frame_table.go | 63 ++++++++++++++----- .../pkg/storage/compress_frame_table_test.go | 12 ++-- .../shared/pkg/storage/compress_upload.go | 4 +- .../pkg/storage/compress_upload_test.go | 4 +- .../pkg/storage/header/serialization_test.go | 48 +++++++------- .../storage/header/serialization_v5_test.go | 2 +- packages/shared/pkg/storage/mock_seekable.go | 14 ++--- packages/shared/pkg/storage/storage.go | 4 +- packages/shared/pkg/storage/storage_aws.go | 2 +- .../pkg/storage/storage_cache_seekable.go | 2 +- .../storage/storage_cache_seekable_test.go | 4 +- packages/shared/pkg/storage/storage_fs.go | 4 +- packages/shared/pkg/storage/storage_google.go | 4 +- 18 files changed, 114 insertions(+), 79 deletions(-) diff --git a/packages/orchestrator/pkg/sandbox/block/streaming_chunk_test.go b/packages/orchestrator/pkg/sandbox/block/streaming_chunk_test.go index 2263dc7c4e..b9784d576f 100644 --- a/packages/orchestrator/pkg/sandbox/block/streaming_chunk_test.go +++ b/packages/orchestrator/pkg/sandbox/block/streaming_chunk_test.go @@ -78,7 +78,7 @@ func (s *fakeSeekable) Size(_ context.Context) (int64, error) { return int64(len(s.data)), nil } -func (s *fakeSeekable) StoreFile(context.Context, string, ...storage.PutOption) (*storage.FrameTable, [32]byte, error) { +func (s *fakeSeekable) StoreFile(context.Context, string, ...storage.PutOption) (*storage.FullFrameTable, [32]byte, error) { panic("not used") } @@ -146,7 +146,7 @@ func makeCompressedTestData(tb testing.TB, data []byte) (*storage.FrameTable, *f }) require.NoError(tb, err) - return ft, &fakeSeekable{data: compressed} + return ft.Table(), &fakeSeekable{data: compressed} } type chunkerTestCase struct { @@ -424,7 +424,7 @@ func (s *panicSeekable) Size(_ context.Context) (int64, error) { return int64(len(s.data)), nil } -func (s *panicSeekable) StoreFile(context.Context, string, ...storage.PutOption) (*storage.FrameTable, [32]byte, error) { +func (s *panicSeekable) StoreFile(context.Context, string, ...storage.PutOption) (*storage.FullFrameTable, [32]byte, error) { panic("not used") } diff --git a/packages/orchestrator/pkg/sandbox/build_upload_v4.go b/packages/orchestrator/pkg/sandbox/build_upload_v4.go index 77b20f0f6e..ad1a463a0c 100644 --- a/packages/orchestrator/pkg/sandbox/build_upload_v4.go +++ b/packages/orchestrator/pkg/sandbox/build_upload_v4.go @@ -75,16 +75,16 @@ func (u *Upload) uploadFramed( var selfBuild headers.BuildData if srcPath != "" { - ft, checksum, err := storage.UploadFramed(ctx, u.store, u.paths.DataFile(string(fileType), cfg.CompressionType()), seekableTypeFor(fileType), srcPath, storage.WithCompressConfig(cfg), storage.WithMetadata(u.objectMetadata), storage.WithChecksumSHA256()) + fullFT, checksum, err := storage.UploadFramed(ctx, u.store, u.paths.DataFile(string(fileType), cfg.CompressionType()), seekableTypeFor(fileType), srcPath, storage.WithCompressConfig(cfg), storage.WithMetadata(u.objectMetadata), storage.WithChecksumSHA256()) if err != nil { return fmt.Errorf("%s upload: %w", fileType, err) } // Compressed: frame-table byte count, since sparse memfile diffs stream // fewer bytes than they occupy on disk. Uncompressed has no table. - size := ft.UncompressedSize() - compressedSize := ft.CompressedSize() - if !ft.IsCompressed() { + size := fullFT.UncompressedSize() + compressedSize := fullFT.CompressedSize() + if !fullFT.IsCompressed() { info, statErr := os.Stat(srcPath) if statErr != nil { return fmt.Errorf("%s stat: %w", fileType, statErr) @@ -98,7 +98,7 @@ func (u *Upload) uploadFramed( dataFileType = uploadFileRootfs } recordUploadCompression(ctx, dataFileType, cfg, size, compressedSize) - selfBuild = headers.BuildData{Size: size, Checksum: checksum, FrameData: ft} + selfBuild = headers.BuildData{Size: size, Checksum: checksum, FrameData: fullFT.Table()} } h := srcHeader.CloneForUpload(u.headerVersion) diff --git a/packages/orchestrator/pkg/sandbox/template/peerclient/seekable.go b/packages/orchestrator/pkg/sandbox/template/peerclient/seekable.go index 4b9230da73..ad337225fe 100644 --- a/packages/orchestrator/pkg/sandbox/template/peerclient/seekable.go +++ b/packages/orchestrator/pkg/sandbox/template/peerclient/seekable.go @@ -154,7 +154,7 @@ func (s *peerSeekable) OpenRangeReader(ctx context.Context, off int64, length in return rc, err } -func (s *peerSeekable) StoreFile(context.Context, string, ...storage.PutOption) (*storage.FrameTable, [32]byte, error) { +func (s *peerSeekable) StoreFile(context.Context, string, ...storage.PutOption) (*storage.FullFrameTable, [32]byte, error) { // peerSeekable only exists when routingProvider routed this buildID to an // active peer at open time, i.e. the file is being P2P-served (the peer // owns the upload). Asking the local orchestrator to upload it is a diff --git a/packages/orchestrator/pkg/sandbox/template/peerclient/seekable_test.go b/packages/orchestrator/pkg/sandbox/template/peerclient/seekable_test.go index b5266e9094..8173ba376a 100644 --- a/packages/orchestrator/pkg/sandbox/template/peerclient/seekable_test.go +++ b/packages/orchestrator/pkg/sandbox/template/peerclient/seekable_test.go @@ -200,7 +200,7 @@ func TestPeerStorageProvider_FullTransitionFlow(t *testing.T) { // 1. Pre-transition read via peer. ft={ct=None} (V3 header). rc, err := seekable.OpenRangeReader(t.Context(), 0, int64(len(prePeerBytes)), - storage.NewFrameTable(storage.CompressionNone, nil)) + storage.NewFullFrameTable(storage.CompressionNone, nil).Table()) require.NoError(t, err) got, err := io.ReadAll(rc) require.NoError(t, err) @@ -210,14 +210,14 @@ func TestPeerStorageProvider_FullTransitionFlow(t *testing.T) { // 2. First post-transition call: retriable error, no peer/base contact. _, err = seekable.OpenRangeReader(t.Context(), 0, 1, - storage.NewFrameTable(storage.CompressionNone, nil)) + storage.NewFullFrameTable(storage.CompressionNone, nil).Table()) var transErr *storage.PeerTransitionedError require.ErrorAs(t, err, &transErr) // 3. Caller reloads V4 header and retries with ct=Zstd. This must hit the // compressed path on base. rc, err = seekable.OpenRangeReader(t.Context(), 0, int64(len(postBaseBytes)), - storage.NewFrameTable(storage.CompressionZstd, nil)) + storage.NewFullFrameTable(storage.CompressionZstd, nil).Table()) require.NoError(t, err) got, err = io.ReadAll(rc) require.NoError(t, err) diff --git a/packages/shared/pkg/storage/compress_encode.go b/packages/shared/pkg/storage/compress_encode.go index 98364b46c5..a996b337cf 100644 --- a/packages/shared/pkg/storage/compress_encode.go +++ b/packages/shared/pkg/storage/compress_encode.go @@ -109,7 +109,7 @@ func newCompressorPool(cfg CompressConfig) (*sync.Pool, error) { return pool, nil } -func CompressBytes(ctx context.Context, data []byte, cfg CompressConfig) (*FrameTable, []byte, [32]byte, error) { +func CompressBytes(ctx context.Context, data []byte, cfg CompressConfig) (*FullFrameTable, []byte, [32]byte, error) { up := &memPartUploader{} const compressBytesConcurrency = 1 diff --git a/packages/shared/pkg/storage/compress_frame_table.go b/packages/shared/pkg/storage/compress_frame_table.go index 9b8079ae7c..beb15e562a 100644 --- a/packages/shared/pkg/storage/compress_frame_table.go +++ b/packages/shared/pkg/storage/compress_frame_table.go @@ -65,21 +65,47 @@ type frameEntry struct { // FrameTable is the decompression index for a compressed diff file. // Immutable after construction; safe to share across goroutines. // Sparse tables (gaps between entries) are supported. +// +// FrameTable is the general type: it is what BuildData stores, what +// DeserializeFrameTable / TrimToRanges return, and what the hot read path +// (ReadAt / OpenRangeReader) accepts. The FullFrameTable wrapper below +// marks the two narrow producer/fallback contexts where intent matters at +// compile time. type FrameTable struct { compressionType CompressionType entries []frameEntry // sorted by StartU } +// FullFrameTable marks a FrameTable that covers an entire file with no gaps +// (entries[0].StartU == 0, frames contiguous). It appears in exactly two +// contexts: (a) fresh out of compressStream / CompressBytes / StoreFile / +// UploadFramed, and (b) latched to a StorageDiff as a fallback when no +// authoritative FrameTable from the header is available. Wraps FrameTable +// by value so layout and method set are unchanged; downcast with Table() +// to feed the hot read path or BuildData.FrameTable. +type FullFrameTable struct{ FrameTable } + +// Table returns the underlying *FrameTable, nil-safely. Use when handing the +// FT to a hot-path consumer (ReadAt / OpenRangeReader) or assigning to +// BuildData.FrameTable. +func (ft *FullFrameTable) Table() *FrameTable { + if ft == nil { + return nil + } + + return &ft.FrameTable +} + // newFrameTableFromEntries creates a FrameTable from pre-computed absolute-offset entries. -func newFrameTableFromEntries(ct CompressionType, entries []frameEntry) *FrameTable { - return &FrameTable{compressionType: ct, entries: entries} +func newFrameTableFromEntries(ct CompressionType, entries []frameEntry) FrameTable { + return FrameTable{compressionType: ct, entries: entries} } -// NewFrameTable creates a FrameTable from consecutive frame sizes, computing -// absolute offsets starting from zero. -func NewFrameTable(ct CompressionType, sizes []FrameSize) *FrameTable { +// NewFullFrameTable creates a FullFrameTable from consecutive frame sizes, +// computing absolute offsets starting from zero. Producer-side constructor. +func NewFullFrameTable(ct CompressionType, sizes []FrameSize) *FullFrameTable { if len(sizes) == 0 { - return newFrameTableFromEntries(ct, nil) + return &FullFrameTable{newFrameTableFromEntries(ct, nil)} } entries := make([]frameEntry, len(sizes)) @@ -96,7 +122,7 @@ func NewFrameTable(ct CompressionType, sizes []FrameSize) *FrameTable { c += int64(s.C) } - return newFrameTableFromEntries(ct, entries) + return &FullFrameTable{newFrameTableFromEntries(ct, entries)} } // CompressionType returns the compression type. Nil-safe: returns CompressionNone for nil. @@ -208,7 +234,8 @@ func (ft *FrameTable) LocateUncompressed(offset int64) (Range, error) { } // Serialize writes the frame table to w in binary little-endian format. -// Nil-safe: writes zeros for type and count. +// Nil-safe: writes zeros for type and count. The V4 writer trims before +// calling this, so headers never carry a full (untrimmed) table. func (ft *FrameTable) Serialize(w io.Writer) error { var ct CompressionType var n int @@ -277,16 +304,20 @@ func DeserializeFrameTable(r io.Reader) (*FrameTable, error) { } } - return newFrameTableFromEntries(CompressionType(ct), entries), nil + ft := newFrameTableFromEntries(CompressionType(ct), entries) + + return &ft, nil } // TrimToRanges returns a new FrameTable containing only the frames that -// overlap with at least one of the given U-space byte ranges. -// Used during V4 header serialization to keep headers compact when a build -// has many frames but only a few are referenced in the current layer. -// Nil-safe: returns ft unchanged when ft is nil or ranges is empty. +// overlap with at least one of the given U-space byte ranges. Used during +// V4 header serialization to keep headers compact when a build has many +// frames but only a few are referenced in the current layer. Nil-safe. func (ft *FrameTable) TrimToRanges(ranges []Range) *FrameTable { - if ft == nil || len(ft.entries) == 0 || len(ranges) == 0 { + if ft == nil { + return nil + } + if len(ft.entries) == 0 || len(ranges) == 0 { return ft } @@ -320,7 +351,9 @@ func (ft *FrameTable) TrimToRanges(ranges []Range) *FrameTable { } } - return newFrameTableFromEntries(ft.compressionType, trimmed) + out := newFrameTableFromEntries(ft.compressionType, trimmed) + + return &out } func (ct CompressionType) Suffix() string { diff --git a/packages/shared/pkg/storage/compress_frame_table_test.go b/packages/shared/pkg/storage/compress_frame_table_test.go index 4a10af5732..78914dfbe3 100644 --- a/packages/shared/pkg/storage/compress_frame_table_test.go +++ b/packages/shared/pkg/storage/compress_frame_table_test.go @@ -93,7 +93,7 @@ func TestLocate(t *testing.T) { func TestNewFrameTable(t *testing.T) { t.Parallel() - ft := NewFrameTable(CompressionZstd, []FrameSize{ + ft := NewFullFrameTable(CompressionZstd, []FrameSize{ {U: 1 << 20, C: 500_000}, {U: 1 << 20, C: 600_000}, }) @@ -118,7 +118,7 @@ func TestNewFrameTable(t *testing.T) { func TestFrameTable_TrimToRanges(t *testing.T) { t.Parallel() - ft := NewFrameTable(CompressionLZ4, []FrameSize{ + ft := NewFullFrameTable(CompressionLZ4, []FrameSize{ {U: 1 << 20, C: 500_000}, {U: 1 << 20, C: 600_000}, {U: 1 << 20, C: 400_000}, @@ -128,7 +128,7 @@ func TestFrameTable_TrimToRanges(t *testing.T) { t.Run("all frames retained", func(t *testing.T) { t.Parallel() trimmed := ft.TrimToRanges([]Range{{Offset: 0, Length: 4 << 20}}) - require.Same(t, ft, trimmed) + require.Equal(t, ft.NumFrames(), trimmed.NumFrames()) }) t.Run("single range trims to subset", func(t *testing.T) { @@ -190,13 +190,15 @@ func TestSerializeDeserializeFrameTable(t *testing.T) { t.Run("round-trip", func(t *testing.T) { t.Parallel() - ft := NewFrameTable(CompressionZstd, []FrameSize{ + ft := NewFullFrameTable(CompressionZstd, []FrameSize{ {U: 2048, C: 1024}, {U: 4096, C: 3500}, }) + // Persisted FTs are always Partial; trim with nil ranges keeps every frame. + partial := ft.TrimToRanges(nil) var buf bytes.Buffer - require.NoError(t, ft.Serialize(&buf)) + require.NoError(t, partial.Serialize(&buf)) got, err := DeserializeFrameTable(&buf) require.NoError(t, err) diff --git a/packages/shared/pkg/storage/compress_upload.go b/packages/shared/pkg/storage/compress_upload.go index 1ed8fcc113..f296945d4d 100644 --- a/packages/shared/pkg/storage/compress_upload.go +++ b/packages/shared/pkg/storage/compress_upload.go @@ -111,7 +111,7 @@ func (p *part) addFrame(ctx context.Context, buf inputBuf, n int, pool *sync.Poo }) } -func compressStream(ctx context.Context, in io.Reader, cfg CompressConfig, uploader partUploader, maxUploadConcurrency int, sink FrameSink) (*FrameTable, [32]byte, error) { +func compressStream(ctx context.Context, in io.Reader, cfg CompressConfig, uploader partUploader, maxUploadConcurrency int, sink FrameSink) (*FullFrameTable, [32]byte, error) { ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -179,7 +179,7 @@ func compressStream(ctx context.Context, in io.Reader, cfg CompressConfig, uploa return nil, [32]byte{}, fmt.Errorf("complete upload: %w", err) } - ft := NewFrameTable(cfg.CompressionType(), frameSizes) + ft := NewFullFrameTable(cfg.CompressionType(), frameSizes) return ft, sum256(hasher), nil } diff --git a/packages/shared/pkg/storage/compress_upload_test.go b/packages/shared/pkg/storage/compress_upload_test.go index 0721cddc92..fadee22173 100644 --- a/packages/shared/pkg/storage/compress_upload_test.go +++ b/packages/shared/pkg/storage/compress_upload_test.go @@ -186,7 +186,7 @@ func TestCompressStreamRoundTrip(t *testing.T) { // Round-trip: decompress and compare. compressed := up.Assemble() - decompressed, err := decompressAll(ft, compressed) + decompressed, err := decompressAll(ft.Table(), compressed) require.NoError(t, err) require.Equal(t, original, decompressed) }) @@ -300,7 +300,7 @@ func TestCompressStreamRace(t *testing.T) { return fmt.Errorf("stream %d: checksum mismatch", i) } - decompressed, err := decompressAll(ft, up.Assemble()) + decompressed, err := decompressAll(ft.Table(), up.Assemble()) if err != nil { return fmt.Errorf("stream %d: decompress: %w", i, err) } diff --git a/packages/shared/pkg/storage/header/serialization_test.go b/packages/shared/pkg/storage/header/serialization_test.go index 83adabcaf8..fe2a09e937 100644 --- a/packages/shared/pkg/storage/header/serialization_test.go +++ b/packages/shared/pkg/storage/header/serialization_test.go @@ -224,10 +224,10 @@ func TestSerializeDeserialize_V4_WithFrameTable(t *testing.T) { h.Builds = map[uuid.UUID]BuildData{ buildID: { Size: 12345, Checksum: checksum, - FrameData: storage.NewFrameTable(storage.CompressionLZ4, []storage.FrameSize{ + FrameData: storage.NewFullFrameTable(storage.CompressionLZ4, []storage.FrameSize{ {U: 2048, C: 1024}, {U: 2048, C: 900}, - }), + }).Table(), }, baseID: {Size: 67890}, } @@ -296,11 +296,11 @@ func TestSerializeDeserialize_V4_Zstd(t *testing.T) { // 3 frames; only the third [8192, 12288) overlaps the mapping. h.Builds = map[uuid.UUID]BuildData{ buildID: { - FrameData: storage.NewFrameTable(storage.CompressionZstd, []storage.FrameSize{ + FrameData: storage.NewFullFrameTable(storage.CompressionZstd, []storage.FrameSize{ {U: 4096, C: 2000}, {U: 4096, C: 3000}, {U: 4096, C: 3500}, - }), + }).Table(), }, } @@ -398,7 +398,7 @@ func TestSerializeDeserialize_V4_ManyFrames(t *testing.T) { h, err := NewHeader(metadata, mappings) require.NoError(t, err) h.Builds = map[uuid.UUID]BuildData{ - buildID: {FrameData: storage.NewFrameTable(storage.CompressionLZ4, frames)}, + buildID: {FrameData: storage.NewFullFrameTable(storage.CompressionLZ4, frames).Table()}, } data, err := SerializeHeader(h) @@ -506,7 +506,7 @@ func TestSerializeDeserialize_V4_MultiBuild_LocateCompressed(t *testing.T) { // Frame 0: U=[0,4096) C=[0,1000) // Frame 1: U=[4096,8192) C=[1000,2800) // Frame 2: U=[8192,12288) C=[2800,5100) - ftA := storage.NewFrameTable(storage.CompressionZstd, []storage.FrameSize{ + ftA := storage.NewFullFrameTable(storage.CompressionZstd, []storage.FrameSize{ {U: 4096, C: 1000}, {U: 4096, C: 1800}, {U: 4096, C: 2300}, @@ -515,7 +515,7 @@ func TestSerializeDeserialize_V4_MultiBuild_LocateCompressed(t *testing.T) { // Build B: 2 frames, each 4096 uncompressed. // Frame 0: U=[0,4096) C=[0,500) // Frame 1: U=[4096,8192) C=[500,1700) - ftB := storage.NewFrameTable(storage.CompressionLZ4, []storage.FrameSize{ + ftB := storage.NewFullFrameTable(storage.CompressionLZ4, []storage.FrameSize{ {U: 4096, C: 500}, {U: 4096, C: 1200}, }) @@ -545,8 +545,8 @@ func TestSerializeDeserialize_V4_MultiBuild_LocateCompressed(t *testing.T) { h, err := NewHeader(metadata, mappings) require.NoError(t, err) h.Builds = map[uuid.UUID]BuildData{ - buildA: {Size: 12288, Checksum: checksumA, FrameData: ftA}, - buildB: {Size: 8192, Checksum: checksumB, FrameData: ftB}, + buildA: {Size: 12288, Checksum: checksumA, FrameData: ftA.Table()}, + buildB: {Size: 8192, Checksum: checksumB, FrameData: ftB.Table()}, } data, err := SerializeHeader(h) @@ -618,7 +618,7 @@ func TestSerializeDeserialize_V4_TrimmedOffsets_Error(t *testing.T) { buildID := uuid.New() // 4 frames, each 4096 uncompressed. - ft := storage.NewFrameTable(storage.CompressionZstd, []storage.FrameSize{ + ft := storage.NewFullFrameTable(storage.CompressionZstd, []storage.FrameSize{ {U: 4096, C: 2000}, {U: 4096, C: 3000}, {U: 4096, C: 3500}, @@ -648,7 +648,7 @@ func TestSerializeDeserialize_V4_TrimmedOffsets_Error(t *testing.T) { h, err := NewHeader(metadata, mappings) require.NoError(t, err) h.Builds = map[uuid.UUID]BuildData{ - buildID: {FrameData: ft}, + buildID: {FrameData: ft.Table()}, } data, err := SerializeHeader(h) @@ -685,7 +685,7 @@ func TestSerializeDeserialize_V4_TrimmedOffsets_Error(t *testing.T) { func TestFrameTable_LocateCompressed(t *testing.T) { t.Parallel() - fd := storage.NewFrameTable(storage.CompressionZstd, []storage.FrameSize{ + fd := storage.NewFullFrameTable(storage.CompressionZstd, []storage.FrameSize{ {U: 2048, C: 1024}, {U: 2048, C: 900}, {U: 4096, C: 3500}, @@ -722,7 +722,7 @@ func TestFrameTable_LocateCompressed(t *testing.T) { func TestFrameTable_LocateUncompressed(t *testing.T) { t.Parallel() - fd := storage.NewFrameTable(storage.CompressionZstd, []storage.FrameSize{ + fd := storage.NewFullFrameTable(storage.CompressionZstd, []storage.FrameSize{ {U: 2048, C: 1024}, {U: 4096, C: 3500}, }) @@ -750,7 +750,7 @@ func TestSerializeDeserialize_V4_SparseTrimming(t *testing.T) { buildID := uuid.New() otherID := uuid.New() - ft := storage.NewFrameTable(storage.CompressionLZ4, []storage.FrameSize{ + ft := storage.NewFullFrameTable(storage.CompressionLZ4, []storage.FrameSize{ {U: 4096, C: 2000}, {U: 4096, C: 3000}, {U: 4096, C: 2500}, @@ -776,7 +776,7 @@ func TestSerializeDeserialize_V4_SparseTrimming(t *testing.T) { h, err := NewHeader(metadata, mappings) require.NoError(t, err) h.Builds = map[uuid.UUID]BuildData{ - buildID: {FrameData: ft, Size: 16384}, + buildID: {FrameData: ft.Table(), Size: 16384}, otherID: {Size: 8192}, } @@ -854,7 +854,7 @@ func TestSerializeDeserialize_V4_MixedChain(t *testing.T) { v3aID := uuid.New() v3bID := uuid.New() - midFT := storage.NewFrameTable(storage.CompressionZstd, []storage.FrameSize{ + midFT := storage.NewFullFrameTable(storage.CompressionZstd, []storage.FrameSize{ {U: 4096, C: 1234}, }) @@ -879,7 +879,7 @@ func TestSerializeDeserialize_V4_MixedChain(t *testing.T) { require.NoError(t, err) h.Builds = map[uuid.UUID]BuildData{ selfID: {}, - midID: {Size: 4096, FrameData: midFT}, + midID: {Size: 4096, FrameData: midFT.Table()}, olderID: {Size: 4096}, } @@ -916,7 +916,7 @@ func TestSerializeDeserialize_V4_MixedChain(t *testing.T) { // Layered chain with a compressed self entry: // C-v4 (self) → U-v4 (mid) → C-v4 (older) → V3 → V3. After a serialize round // trip, every virtual offset resolves to the right build via GetShiftedMapping -// and carries the expected compression (FrameData present iff compressed). +// and carries the expected compression (FrameTable present iff compressed). func TestSerializeDeserialize_V4_CompressedSelfChain(t *testing.T) { t.Parallel() @@ -926,11 +926,11 @@ func TestSerializeDeserialize_V4_CompressedSelfChain(t *testing.T) { v3aID := uuid.New() v3bID := uuid.New() - selfFT := storage.NewFrameTable(storage.CompressionZstd, []storage.FrameSize{ + selfFT := storage.NewFullFrameTable(storage.CompressionZstd, []storage.FrameSize{ {U: 4096, C: 1100}, {U: 4096, C: 1200}, }) - olderFT := storage.NewFrameTable(storage.CompressionZstd, []storage.FrameSize{ + olderFT := storage.NewFullFrameTable(storage.CompressionZstd, []storage.FrameSize{ {U: 4096, C: 1300}, }) @@ -955,9 +955,9 @@ func TestSerializeDeserialize_V4_CompressedSelfChain(t *testing.T) { h, err := NewHeader(metadata, mappings) require.NoError(t, err) h.Builds = map[uuid.UUID]BuildData{ - selfID: {Size: 8192, FrameData: selfFT}, + selfID: {Size: 8192, FrameData: selfFT.Table()}, midID: {Size: 4096}, - olderID: {Size: 4096, FrameData: olderFT}, + olderID: {Size: 4096, FrameData: olderFT.Table()}, } data, err := SerializeHeader(h) @@ -992,10 +992,10 @@ func TestSerializeDeserialize_V4_CompressedSelfChain(t *testing.T) { ft := got.GetBuildFrameData(m.BuildId) if tc.compressed { - require.NotNil(t, ft, "offset %d expected FrameData", tc.offset) + require.NotNil(t, ft, "offset %d expected FrameTable", tc.offset) require.Equal(t, storage.CompressionZstd, ft.CompressionType(), "offset %d", tc.offset) } else { - require.Nil(t, ft, "offset %d expected no FrameData", tc.offset) + require.Nil(t, ft, "offset %d expected no FrameTable", tc.offset) } } diff --git a/packages/shared/pkg/storage/header/serialization_v5_test.go b/packages/shared/pkg/storage/header/serialization_v5_test.go index 768fbdce38..42392125b0 100644 --- a/packages/shared/pkg/storage/header/serialization_v5_test.go +++ b/packages/shared/pkg/storage/header/serialization_v5_test.go @@ -61,7 +61,7 @@ func TestV5_RoundTrip(t *testing.T) { } checksum := sha256.Sum256([]byte("a")) h := v5Header(t, metadata, mappings, map[uuid.UUID]BuildData{ - a: {Size: 100, Checksum: checksum, FrameData: storage.NewFrameTable(storage.CompressionZstd, []storage.FrameSize{{U: int32(bs), C: 1000}, {U: int32(2 * bs), C: 2000}})}, + a: {Size: 100, Checksum: checksum, FrameData: storage.NewFullFrameTable(storage.CompressionZstd, []storage.FrameSize{{U: int32(bs), C: 1000}, {U: int32(2 * bs), C: 2000}}).Table()}, b: {Size: 200}, }) diff --git a/packages/shared/pkg/storage/mock_seekable.go b/packages/shared/pkg/storage/mock_seekable.go index 440836bc54..94d6b8712e 100644 --- a/packages/shared/pkg/storage/mock_seekable.go +++ b/packages/shared/pkg/storage/mock_seekable.go @@ -179,7 +179,7 @@ func (_c *MockSeekable_Size_Call) RunAndReturn(run func(ctx context.Context) (in } // StoreFile provides a mock function for the type MockSeekable -func (_mock *MockSeekable) StoreFile(ctx context.Context, path string, opts ...PutOption) (*FrameTable, [32]byte, error) { +func (_mock *MockSeekable) StoreFile(ctx context.Context, path string, opts ...PutOption) (*FullFrameTable, [32]byte, error) { var tmpRet mock.Arguments if len(opts) > 0 { tmpRet = _mock.Called(ctx, path, opts) @@ -192,17 +192,17 @@ func (_mock *MockSeekable) StoreFile(ctx context.Context, path string, opts ...P panic("no return value specified for StoreFile") } - var r0 *FrameTable + var r0 *FullFrameTable var r1 [32]byte var r2 error - if returnFunc, ok := ret.Get(0).(func(context.Context, string, ...PutOption) (*FrameTable, [32]byte, error)); ok { + if returnFunc, ok := ret.Get(0).(func(context.Context, string, ...PutOption) (*FullFrameTable, [32]byte, error)); ok { return returnFunc(ctx, path, opts...) } - if returnFunc, ok := ret.Get(0).(func(context.Context, string, ...PutOption) *FrameTable); ok { + if returnFunc, ok := ret.Get(0).(func(context.Context, string, ...PutOption) *FullFrameTable); ok { r0 = returnFunc(ctx, path, opts...) } else { if ret.Get(0) != nil { - r0 = ret.Get(0).(*FrameTable) + r0 = ret.Get(0).(*FullFrameTable) } } if returnFunc, ok := ret.Get(1).(func(context.Context, string, ...PutOption) [32]byte); ok { @@ -259,12 +259,12 @@ func (_c *MockSeekable_StoreFile_Call) Run(run func(ctx context.Context, path st return _c } -func (_c *MockSeekable_StoreFile_Call) Return(frameTable *FrameTable, bytes [32]byte, err error) *MockSeekable_StoreFile_Call { +func (_c *MockSeekable_StoreFile_Call) Return(frameTable *FullFrameTable, bytes [32]byte, err error) *MockSeekable_StoreFile_Call { _c.Call.Return(frameTable, bytes, err) return _c } -func (_c *MockSeekable_StoreFile_Call) RunAndReturn(run func(ctx context.Context, path string, opts ...PutOption) (*FrameTable, [32]byte, error)) *MockSeekable_StoreFile_Call { +func (_c *MockSeekable_StoreFile_Call) RunAndReturn(run func(ctx context.Context, path string, opts ...PutOption) (*FullFrameTable, [32]byte, error)) *MockSeekable_StoreFile_Call { _c.Call.Return(run) return _c } diff --git a/packages/shared/pkg/storage/storage.go b/packages/shared/pkg/storage/storage.go index 668ca49fb4..1515444d0e 100644 --- a/packages/shared/pkg/storage/storage.go +++ b/packages/shared/pkg/storage/storage.go @@ -153,7 +153,7 @@ type StreamingReader interface { type SeekableWriter interface { // Store entire file. Compression is opt-in via WithCompressConfig. - StoreFile(ctx context.Context, path string, opts ...PutOption) (*FrameTable, [32]byte, error) + StoreFile(ctx context.Context, path string, opts ...PutOption) (*FullFrameTable, [32]byte, error) } type Seekable interface { @@ -162,7 +162,7 @@ type Seekable interface { Size(ctx context.Context) (int64, error) } -func UploadFramed(ctx context.Context, provider StorageProvider, remotePath string, objType SeekableObjectType, localPath string, opts ...PutOption) (*FrameTable, [32]byte, error) { +func UploadFramed(ctx context.Context, provider StorageProvider, remotePath string, objType SeekableObjectType, localPath string, opts ...PutOption) (*FullFrameTable, [32]byte, error) { object, err := provider.OpenSeekable(ctx, remotePath, objType) if err != nil { return nil, [32]byte{}, err diff --git a/packages/shared/pkg/storage/storage_aws.go b/packages/shared/pkg/storage/storage_aws.go index 5e62a784b9..467c01c79f 100644 --- a/packages/shared/pkg/storage/storage_aws.go +++ b/packages/shared/pkg/storage/storage_aws.go @@ -162,7 +162,7 @@ func (o *awsObject) WriteTo(ctx context.Context, dst io.Writer) (int64, error) { return io.Copy(dst, resp.Body) } -func (o *awsObject) StoreFile(ctx context.Context, path string, opts ...PutOption) (*FrameTable, [32]byte, error) { +func (o *awsObject) StoreFile(ctx context.Context, path string, opts ...PutOption) (*FullFrameTable, [32]byte, error) { p := ApplyPutOptions(opts) if CompressConfigFromOpts(p).IsCompressionEnabled() { return nil, [32]byte{}, errors.New("compressed uploads are not supported on AWS (builds target GCP only)") diff --git a/packages/shared/pkg/storage/storage_cache_seekable.go b/packages/shared/pkg/storage/storage_cache_seekable.go index a3b20c1118..a7965fd636 100644 --- a/packages/shared/pkg/storage/storage_cache_seekable.go +++ b/packages/shared/pkg/storage/storage_cache_seekable.go @@ -296,7 +296,7 @@ func (c *cachedSeekable) Size(ctx context.Context) (n int64, e error) { return size, nil } -func (c *cachedSeekable) StoreFile(ctx context.Context, path string, opts ...PutOption) (_ *FrameTable, _ [32]byte, e error) { +func (c *cachedSeekable) StoreFile(ctx context.Context, path string, opts ...PutOption) (_ *FullFrameTable, _ [32]byte, e error) { ctx, span := c.tracer.Start(ctx, "write object from file system", trace.WithAttributes(attribute.String("path", path)), ) diff --git a/packages/shared/pkg/storage/storage_cache_seekable_test.go b/packages/shared/pkg/storage/storage_cache_seekable_test.go index d989eb084f..8524d59795 100644 --- a/packages/shared/pkg/storage/storage_cache_seekable_test.go +++ b/packages/shared/pkg/storage/storage_cache_seekable_test.go @@ -622,11 +622,11 @@ func TestCachedSeekable_StoreFile_Compressed_WriteThrough(t *testing.T) { // Stub inner.StoreFile: open the file and run compressStream with the sink // pulled from opts — mirrors what fs/GCS backends do for compressed puts. up := &memPartUploader{} - var capturedFT *FrameTable + var capturedFT *FullFrameTable inner := NewMockSeekable(t) inner.EXPECT(). StoreFile(mock.Anything, mock.Anything, mock.Anything). - RunAndReturn(func(ctx context.Context, path string, opts ...PutOption) (*FrameTable, [32]byte, error) { + RunAndReturn(func(ctx context.Context, path string, opts ...PutOption) (*FullFrameTable, [32]byte, error) { po := ApplyPutOptions(opts) require.NotNil(t, po.FrameSink, "cachedSeekable must attach a FrameSink for compressed+writeThrough") diff --git a/packages/shared/pkg/storage/storage_fs.go b/packages/shared/pkg/storage/storage_fs.go index 25b08f7b10..8f72834767 100644 --- a/packages/shared/pkg/storage/storage_fs.go +++ b/packages/shared/pkg/storage/storage_fs.go @@ -130,7 +130,7 @@ func (o *fsObject) Put(_ context.Context, data []byte, _ ...PutOption) error { return err } -func (o *fsObject) StoreFile(ctx context.Context, path string, opts ...PutOption) (*FrameTable, [32]byte, error) { +func (o *fsObject) StoreFile(ctx context.Context, path string, opts ...PutOption) (*FullFrameTable, [32]byte, error) { putOpts := ApplyPutOptions(opts) cfg := CompressConfigFromOpts(putOpts) if cfg.IsCompressionEnabled() { @@ -174,7 +174,7 @@ func (o *fsObject) StoreFile(ctx context.Context, path string, opts ...PutOption return nil, [32]byte{}, err } -func (o *fsObject) storeFileCompressed(ctx context.Context, localPath string, cfg CompressConfig, sink FrameSink) (*FrameTable, [32]byte, error) { +func (o *fsObject) storeFileCompressed(ctx context.Context, localPath string, cfg CompressConfig, sink FrameSink) (*FullFrameTable, [32]byte, error) { file, err := os.Open(localPath) if err != nil { return nil, [32]byte{}, fmt.Errorf("failed to open local file %s: %w", localPath, err) diff --git a/packages/shared/pkg/storage/storage_google.go b/packages/shared/pkg/storage/storage_google.go index 0556cecb73..dcada06468 100644 --- a/packages/shared/pkg/storage/storage_google.go +++ b/packages/shared/pkg/storage/storage_google.go @@ -401,7 +401,7 @@ func (o *gcpObject) WriteTo(ctx context.Context, dst io.Writer) (int64, error) { return n, nil } -func (o *gcpObject) StoreFile(ctx context.Context, path string, opts ...PutOption) (_ *FrameTable, _ [32]byte, e error) { +func (o *gcpObject) StoreFile(ctx context.Context, path string, opts ...PutOption) (_ *FullFrameTable, _ [32]byte, e error) { ctx, span := tracer.Start(ctx, "write to gcp from file system") defer func() { recordError(span, e) @@ -547,7 +547,7 @@ func (o *gcpObject) StoreFile(ctx context.Context, path string, opts ...PutOptio return nil, sum256(hasher), e } -func (o *gcpObject) storeFileCompressed(ctx context.Context, localPath string, cfg CompressConfig, maxConcurrency int, putOpts PutOptions) (*FrameTable, [32]byte, error) { +func (o *gcpObject) storeFileCompressed(ctx context.Context, localPath string, cfg CompressConfig, maxConcurrency int, putOpts PutOptions) (*FullFrameTable, [32]byte, error) { file, err := os.Open(localPath) if err != nil { return nil, [32]byte{}, fmt.Errorf("failed to open local file %s: %w", localPath, err) From 3be648d3eff0d2d7050a50afdaa6ea1073e413ca Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 10 Jun 2026 18:05:05 +0000 Subject: [PATCH 2/3] chore: auto-commit generated changes --- packages/shared/pkg/storage/mock_seekable.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/shared/pkg/storage/mock_seekable.go b/packages/shared/pkg/storage/mock_seekable.go index 94d6b8712e..610bddaca0 100644 --- a/packages/shared/pkg/storage/mock_seekable.go +++ b/packages/shared/pkg/storage/mock_seekable.go @@ -259,8 +259,8 @@ func (_c *MockSeekable_StoreFile_Call) Run(run func(ctx context.Context, path st return _c } -func (_c *MockSeekable_StoreFile_Call) Return(frameTable *FullFrameTable, bytes [32]byte, err error) *MockSeekable_StoreFile_Call { - _c.Call.Return(frameTable, bytes, err) +func (_c *MockSeekable_StoreFile_Call) Return(fullFrameTable *FullFrameTable, bytes [32]byte, err error) *MockSeekable_StoreFile_Call { + _c.Call.Return(fullFrameTable, bytes, err) return _c } From 61fb9dc1f39a054459f1260f5ef12fdc8692626f Mon Sep 17 00:00:00 2001 From: Lev Brouk Date: Wed, 10 Jun 2026 11:36:58 -0700 Subject: [PATCH 3/3] PR feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Make the FullFrameTable wrapper unembedded (private named field), so methods on the inner FrameTable are no longer promoted. The only way to use a *FullFrameTable is now its nil-safe Table() — calling methods directly on a nil *FullFrameTable becomes a compile error instead of a runtime panic. This catches the bug Gemini flagged on build_upload_v4.go: fullFT is nil for uncompressed uploads, and Go's embed selector would deref it before reaching the underlying nil-safe FrameTable methods. Now fixed by extracting ft := fullFT.Table() once and routing all calls through it. Same fix applied to debug-logging in storage_fs/storage_google. TrimToRanges, DeserializeFrameTable, Serialize, and newFrameTableFromEntries are all reverted to their original bodies (no behavior delta in FrameTable itself). --- .../pkg/sandbox/build_upload_v4.go | 9 +-- .../pkg/storage/compress_frame_table.go | 60 +++++++------------ .../pkg/storage/compress_frame_table_test.go | 10 ++-- .../pkg/storage/compress_upload_test.go | 8 ++- .../pkg/storage/header/serialization_test.go | 6 +- .../storage/storage_cache_seekable_test.go | 7 ++- packages/shared/pkg/storage/storage_fs.go | 7 ++- packages/shared/pkg/storage/storage_google.go | 5 +- 8 files changed, 49 insertions(+), 63 deletions(-) diff --git a/packages/orchestrator/pkg/sandbox/build_upload_v4.go b/packages/orchestrator/pkg/sandbox/build_upload_v4.go index ad1a463a0c..a6e8b87068 100644 --- a/packages/orchestrator/pkg/sandbox/build_upload_v4.go +++ b/packages/orchestrator/pkg/sandbox/build_upload_v4.go @@ -82,9 +82,10 @@ func (u *Upload) uploadFramed( // Compressed: frame-table byte count, since sparse memfile diffs stream // fewer bytes than they occupy on disk. Uncompressed has no table. - size := fullFT.UncompressedSize() - compressedSize := fullFT.CompressedSize() - if !fullFT.IsCompressed() { + ft := fullFT.Table() + size := ft.UncompressedSize() + compressedSize := ft.CompressedSize() + if !ft.IsCompressed() { info, statErr := os.Stat(srcPath) if statErr != nil { return fmt.Errorf("%s stat: %w", fileType, statErr) @@ -98,7 +99,7 @@ func (u *Upload) uploadFramed( dataFileType = uploadFileRootfs } recordUploadCompression(ctx, dataFileType, cfg, size, compressedSize) - selfBuild = headers.BuildData{Size: size, Checksum: checksum, FrameData: fullFT.Table()} + selfBuild = headers.BuildData{Size: size, Checksum: checksum, FrameData: ft} } h := srcHeader.CloneForUpload(u.headerVersion) diff --git a/packages/shared/pkg/storage/compress_frame_table.go b/packages/shared/pkg/storage/compress_frame_table.go index beb15e562a..ea937927cb 100644 --- a/packages/shared/pkg/storage/compress_frame_table.go +++ b/packages/shared/pkg/storage/compress_frame_table.go @@ -65,47 +65,36 @@ type frameEntry struct { // FrameTable is the decompression index for a compressed diff file. // Immutable after construction; safe to share across goroutines. // Sparse tables (gaps between entries) are supported. -// -// FrameTable is the general type: it is what BuildData stores, what -// DeserializeFrameTable / TrimToRanges return, and what the hot read path -// (ReadAt / OpenRangeReader) accepts. The FullFrameTable wrapper below -// marks the two narrow producer/fallback contexts where intent matters at -// compile time. type FrameTable struct { compressionType CompressionType entries []frameEntry // sorted by StartU } -// FullFrameTable marks a FrameTable that covers an entire file with no gaps -// (entries[0].StartU == 0, frames contiguous). It appears in exactly two -// contexts: (a) fresh out of compressStream / CompressBytes / StoreFile / -// UploadFramed, and (b) latched to a StorageDiff as a fallback when no -// authoritative FrameTable from the header is available. Wraps FrameTable -// by value so layout and method set are unchanged; downcast with Table() -// to feed the hot read path or BuildData.FrameTable. -type FullFrameTable struct{ FrameTable } - -// Table returns the underlying *FrameTable, nil-safely. Use when handing the -// FT to a hot-path consumer (ReadAt / OpenRangeReader) or assigning to -// BuildData.FrameTable. +// FullFrameTable marks a FrameTable that covers an entire file with no gaps. +// Produced by compressStream / CompressBytes / StoreFile / UploadFramed. +// The inner FrameTable is unexported and unembedded so methods are not +// promoted; consumers must reach functionality through nil-safe Table(). +type FullFrameTable struct{ ft FrameTable } + +// Table returns the underlying *FrameTable, nil-safely. func (ft *FullFrameTable) Table() *FrameTable { if ft == nil { return nil } - return &ft.FrameTable + return &ft.ft } // newFrameTableFromEntries creates a FrameTable from pre-computed absolute-offset entries. -func newFrameTableFromEntries(ct CompressionType, entries []frameEntry) FrameTable { - return FrameTable{compressionType: ct, entries: entries} +func newFrameTableFromEntries(ct CompressionType, entries []frameEntry) *FrameTable { + return &FrameTable{compressionType: ct, entries: entries} } // NewFullFrameTable creates a FullFrameTable from consecutive frame sizes, -// computing absolute offsets starting from zero. Producer-side constructor. +// computing absolute offsets starting from zero. func NewFullFrameTable(ct CompressionType, sizes []FrameSize) *FullFrameTable { if len(sizes) == 0 { - return &FullFrameTable{newFrameTableFromEntries(ct, nil)} + return &FullFrameTable{ft: *newFrameTableFromEntries(ct, nil)} } entries := make([]frameEntry, len(sizes)) @@ -122,7 +111,7 @@ func NewFullFrameTable(ct CompressionType, sizes []FrameSize) *FullFrameTable { c += int64(s.C) } - return &FullFrameTable{newFrameTableFromEntries(ct, entries)} + return &FullFrameTable{ft: *newFrameTableFromEntries(ct, entries)} } // CompressionType returns the compression type. Nil-safe: returns CompressionNone for nil. @@ -234,8 +223,7 @@ func (ft *FrameTable) LocateUncompressed(offset int64) (Range, error) { } // Serialize writes the frame table to w in binary little-endian format. -// Nil-safe: writes zeros for type and count. The V4 writer trims before -// calling this, so headers never carry a full (untrimmed) table. +// Nil-safe: writes zeros for type and count. func (ft *FrameTable) Serialize(w io.Writer) error { var ct CompressionType var n int @@ -304,20 +292,16 @@ func DeserializeFrameTable(r io.Reader) (*FrameTable, error) { } } - ft := newFrameTableFromEntries(CompressionType(ct), entries) - - return &ft, nil + return newFrameTableFromEntries(CompressionType(ct), entries), nil } // TrimToRanges returns a new FrameTable containing only the frames that -// overlap with at least one of the given U-space byte ranges. Used during -// V4 header serialization to keep headers compact when a build has many -// frames but only a few are referenced in the current layer. Nil-safe. +// overlap with at least one of the given U-space byte ranges. +// Used during V4 header serialization to keep headers compact when a build +// has many frames but only a few are referenced in the current layer. +// Nil-safe: returns ft unchanged when ft is nil or ranges is empty. func (ft *FrameTable) TrimToRanges(ranges []Range) *FrameTable { - if ft == nil { - return nil - } - if len(ft.entries) == 0 || len(ranges) == 0 { + if ft == nil || len(ft.entries) == 0 || len(ranges) == 0 { return ft } @@ -351,9 +335,7 @@ func (ft *FrameTable) TrimToRanges(ranges []Range) *FrameTable { } } - out := newFrameTableFromEntries(ft.compressionType, trimmed) - - return &out + return newFrameTableFromEntries(ft.compressionType, trimmed) } func (ct CompressionType) Suffix() string { diff --git a/packages/shared/pkg/storage/compress_frame_table_test.go b/packages/shared/pkg/storage/compress_frame_table_test.go index 78914dfbe3..9927d612c4 100644 --- a/packages/shared/pkg/storage/compress_frame_table_test.go +++ b/packages/shared/pkg/storage/compress_frame_table_test.go @@ -96,7 +96,7 @@ func TestNewFrameTable(t *testing.T) { ft := NewFullFrameTable(CompressionZstd, []FrameSize{ {U: 1 << 20, C: 500_000}, {U: 1 << 20, C: 600_000}, - }) + }).Table() require.Equal(t, 2, ft.NumFrames()) require.Equal(t, CompressionZstd, ft.CompressionType()) @@ -123,7 +123,7 @@ func TestFrameTable_TrimToRanges(t *testing.T) { {U: 1 << 20, C: 600_000}, {U: 1 << 20, C: 400_000}, {U: 1 << 20, C: 700_000}, - }) + }).Table() t.Run("all frames retained", func(t *testing.T) { t.Parallel() @@ -193,12 +193,10 @@ func TestSerializeDeserializeFrameTable(t *testing.T) { ft := NewFullFrameTable(CompressionZstd, []FrameSize{ {U: 2048, C: 1024}, {U: 4096, C: 3500}, - }) - // Persisted FTs are always Partial; trim with nil ranges keeps every frame. - partial := ft.TrimToRanges(nil) + }).Table() var buf bytes.Buffer - require.NoError(t, partial.Serialize(&buf)) + require.NoError(t, ft.Serialize(&buf)) got, err := DeserializeFrameTable(&buf) require.NoError(t, err) diff --git a/packages/shared/pkg/storage/compress_upload_test.go b/packages/shared/pkg/storage/compress_upload_test.go index fadee22173..8efe3f2568 100644 --- a/packages/shared/pkg/storage/compress_upload_test.go +++ b/packages/shared/pkg/storage/compress_upload_test.go @@ -160,7 +160,7 @@ func TestCompressStreamRoundTrip(t *testing.T) { up := &memPartUploader{} cfg := defaultCfg(tc.codec, tc.workers, tc.frameSize) - ft, checksum, err := compressStream( + fullFT, checksum, err := compressStream( t.Context(), bytes.NewReader(original), cfg, @@ -169,6 +169,7 @@ func TestCompressStreamRoundTrip(t *testing.T) { nil, ) require.NoError(t, err) + ft := fullFT.Table() if tc.dataSize == 0 { require.Equal(t, 0, ft.NumFrames()) @@ -186,7 +187,7 @@ func TestCompressStreamRoundTrip(t *testing.T) { // Round-trip: decompress and compare. compressed := up.Assemble() - decompressed, err := decompressAll(ft.Table(), compressed) + decompressed, err := decompressAll(ft, compressed) require.NoError(t, err) require.Equal(t, original, decompressed) }) @@ -417,10 +418,11 @@ func BenchmarkStoreFile(b *testing.B) { outPath := filepath.Join(outDir, "output.dat") obj := &fsObject{path: outPath} - ft, _, err := obj.StoreFile(b.Context(), inputPath, WithCompressConfig(compCfg)) + fullFT, _, err := obj.StoreFile(b.Context(), inputPath, WithCompressConfig(compCfg)) if err != nil { b.Fatal(err) } + ft := fullFT.Table() b.ReportMetric(float64(ft.CompressedSize())/float64(ft.UncompressedSize()), "ratio") } diff --git a/packages/shared/pkg/storage/header/serialization_test.go b/packages/shared/pkg/storage/header/serialization_test.go index fe2a09e937..022e0edecb 100644 --- a/packages/shared/pkg/storage/header/serialization_test.go +++ b/packages/shared/pkg/storage/header/serialization_test.go @@ -689,7 +689,7 @@ func TestFrameTable_LocateCompressed(t *testing.T) { {U: 2048, C: 1024}, {U: 2048, C: 900}, {U: 4096, C: 3500}, - }) + }).Table() // Frame 0: U=[0,2048), C=[0,1024) r, err := fd.LocateCompressed(0) @@ -725,7 +725,7 @@ func TestFrameTable_LocateUncompressed(t *testing.T) { fd := storage.NewFullFrameTable(storage.CompressionZstd, []storage.FrameSize{ {U: 2048, C: 1024}, {U: 4096, C: 3500}, - }) + }).Table() // Frame 0: U=[0,2048) r, err := fd.LocateUncompressed(0) @@ -916,7 +916,7 @@ func TestSerializeDeserialize_V4_MixedChain(t *testing.T) { // Layered chain with a compressed self entry: // C-v4 (self) → U-v4 (mid) → C-v4 (older) → V3 → V3. After a serialize round // trip, every virtual offset resolves to the right build via GetShiftedMapping -// and carries the expected compression (FrameTable present iff compressed). +// and carries the expected compression (FrameData present iff compressed). func TestSerializeDeserialize_V4_CompressedSelfChain(t *testing.T) { t.Parallel() diff --git a/packages/shared/pkg/storage/storage_cache_seekable_test.go b/packages/shared/pkg/storage/storage_cache_seekable_test.go index 8524d59795..26bac12f9a 100644 --- a/packages/shared/pkg/storage/storage_cache_seekable_test.go +++ b/packages/shared/pkg/storage/storage_cache_seekable_test.go @@ -651,10 +651,11 @@ func TestCachedSeekable_StoreFile_Compressed_WriteThrough(t *testing.T) { c.wg.Wait() - require.Equal(t, 3, capturedFT.NumFrames()) + ft := capturedFT.Table() + require.Equal(t, 3, ft.NumFrames()) assembled := up.Assemble() - for i := range capturedFT.NumFrames() { - _, _, startC, endC := capturedFT.FrameAt(i) + for i := range ft.NumFrames() { + _, _, startC, endC := ft.FrameAt(i) framePath := makeFrameFilename(c.path, Range{Offset: startC, Length: int(endC - startC)}) onDisk, err := os.ReadFile(framePath) require.NoError(t, err) diff --git a/packages/shared/pkg/storage/storage_fs.go b/packages/shared/pkg/storage/storage_fs.go index 8f72834767..828da55639 100644 --- a/packages/shared/pkg/storage/storage_fs.go +++ b/packages/shared/pkg/storage/storage_fs.go @@ -136,13 +136,14 @@ func (o *fsObject) StoreFile(ctx context.Context, path string, opts ...PutOption if cfg.IsCompressionEnabled() { ft, checksum, err := o.storeFileCompressed(ctx, path, cfg, putOpts.FrameSink) if err == nil { + t := ft.Table() logger.L().Debug(ctx, "Stored file to filesystem", zap.String("object", o.path), zap.String("source", path), - zap.Int64("size_uncompressed", ft.UncompressedSize()), - zap.Int64("size_compressed", ft.CompressedSize()), + zap.Int64("size_uncompressed", t.UncompressedSize()), + zap.Int64("size_compressed", t.CompressedSize()), zap.String("compression", cfg.CompressionType().String()), - zap.Int("frames", ft.NumFrames()), + zap.Int("frames", t.NumFrames()), ) } diff --git a/packages/shared/pkg/storage/storage_google.go b/packages/shared/pkg/storage/storage_google.go index dcada06468..1bfb689729 100644 --- a/packages/shared/pkg/storage/storage_google.go +++ b/packages/shared/pkg/storage/storage_google.go @@ -456,14 +456,15 @@ func (o *gcpObject) StoreFile(ctx context.Context, path string, opts ...PutOptio } else { timer.Success(ctx, fileInfo.Size()) + t := ft.Table() logger.L().Debug(ctx, "Uploaded file to GCS", zap.String("bucket", bucketName), zap.String("object", objectName), zap.String("source", path), zap.Int64("size_uncompressed", fileInfo.Size()), - zap.Int64("size_compressed", ft.CompressedSize()), + zap.Int64("size_compressed", t.CompressedSize()), zap.String("compression", cfg.CompressionType().String()), - zap.Int("frames", ft.NumFrames()), + zap.Int("frames", t.NumFrames()), zap.Int64("duration_ms", time.Since(start).Milliseconds()), ) }