From 8d49b64364285898770dd28c53b1bf75114fe436 Mon Sep 17 00:00:00 2001 From: ValentaTomas Date: Wed, 20 May 2026 18:42:50 -0700 Subject: [PATCH 1/3] fix(orchestrator): report real V4 header compression ratio StoreHeader now returns the actual pre-LZ4 vs stored byte sizes so the upload metrics expose the inner block compression that the V4 serializer applies, instead of trivially reporting a 1:1 ratio for every header. V3 (uncompressed) keeps reporting 1:1. --- .../pkg/sandbox/upload_metrics.go | 5 +- .../pkg/storage/header/serialization.go | 58 +++++++++++++++---- .../pkg/storage/header/serialization_v4.go | 20 ++++--- 3 files changed, 61 insertions(+), 22 deletions(-) diff --git a/packages/orchestrator/pkg/sandbox/upload_metrics.go b/packages/orchestrator/pkg/sandbox/upload_metrics.go index 499a85200e..72cf509495 100644 --- a/packages/orchestrator/pkg/sandbox/upload_metrics.go +++ b/packages/orchestrator/pkg/sandbox/upload_metrics.go @@ -47,12 +47,13 @@ func uploadRatioBp(compressed, uncompressed int64) int64 { } func storeHeaderWithMetrics(ctx context.Context, store storage.StorageProvider, path, fileType string, h *headers.Header) error { - size, err := headers.StoreHeader(ctx, store, path, h) + stats, err := headers.StoreHeader(ctx, store, path, h) if err != nil { return err } - recordUploadCompression(ctx, uploadArtifactHeader, fileType, storage.CompressConfig{}, size, size) + cfg := storage.CompressConfig{Type: stats.Compression.String()} + recordUploadCompression(ctx, uploadArtifactHeader, fileType, cfg, stats.Uncompressed, stats.Stored) return nil } diff --git a/packages/shared/pkg/storage/header/serialization.go b/packages/shared/pkg/storage/header/serialization.go index 9e25212495..16e59760fc 100644 --- a/packages/shared/pkg/storage/header/serialization.go +++ b/packages/shared/pkg/storage/header/serialization.go @@ -8,16 +8,52 @@ import ( "github.com/e2b-dev/infra/packages/shared/pkg/storage" ) +// HeaderStoreStats reports byte sizes for a serialized header. Stored is what +// was written to storage; Uncompressed is the same payload sized as if the +// inner LZ4 block had not been compressed (V3 has no inner compression, so +// Stored == Uncompressed). Compression is the inner-block codec actually used. +type HeaderStoreStats struct { + Stored int64 + Uncompressed int64 + Compression storage.CompressionType +} + // SerializeHeader serializes a header, dispatching to the version-specific format. // // V3 (Version <= 3): [Metadata] [v3 mappings…] // V4 (Version >= 4): [Metadata] [uint8 flags] [uint32 uncompressedSize] [LZ4( Builds + v4 mappings )] func SerializeHeader(h *Header) ([]byte, error) { + data, _, err := serializeHeaderWithStats(h) + + return data, err +} + +func serializeHeaderWithStats(h *Header) ([]byte, HeaderStoreStats, error) { if h.Metadata.Version <= 3 { - return serializeV3(h.Metadata, h.Mapping) + data, err := serializeV3(h.Metadata, h.Mapping) + if err != nil { + return nil, HeaderStoreStats{}, err + } + + return data, HeaderStoreStats{ + Stored: int64(len(data)), + Uncompressed: int64(len(data)), + Compression: storage.CompressionNone, + }, nil } - return serializeV4(h.Metadata, h.Builds, h.Mapping, h.IncompletePendingUpload) + data, blockUncompressed, err := serializeV4(h.Metadata, h.Builds, h.Mapping, h.IncompletePendingUpload) + if err != nil { + return nil, HeaderStoreStats{}, err + } + + overhead := int64(metadataSize + v4FlagsLen + v4SizePrefixLen) + + return data, HeaderStoreStats{ + Stored: int64(len(data)), + Uncompressed: overhead + blockUncompressed, + Compression: storage.CompressionLZ4, + }, nil } // DeserializeBytes auto-detects the header version and deserializes accordingly. @@ -57,33 +93,33 @@ func LoadHeader(ctx context.Context, s storage.StorageProvider, path string) (*H return DeserializeBytes(data) } -// StoreHeader serializes a header, uploads it, and returns the stored byte count. +// StoreHeader serializes a header, uploads it, and returns its size stats. // Refuses to persist a header still flagged as in-flight — the upload pipeline // must clear IncompletePendingUpload before reaching here. -func StoreHeader(ctx context.Context, s storage.StorageProvider, path string, h *Header) (int64, error) { +func StoreHeader(ctx context.Context, s storage.StorageProvider, path string, h *Header) (HeaderStoreStats, error) { if h == nil { - return 0, errors.New("header is nil") + return HeaderStoreStats{}, errors.New("header is nil") } if h.IncompletePendingUpload { - return 0, fmt.Errorf("refusing to persist incomplete header for %s", path) + return HeaderStoreStats{}, fmt.Errorf("refusing to persist incomplete header for %s", path) } - data, err := SerializeHeader(h) + data, stats, err := serializeHeaderWithStats(h) if err != nil { - return 0, fmt.Errorf("serialize header: %w", err) + return HeaderStoreStats{}, fmt.Errorf("serialize header: %w", err) } blob, err := s.OpenBlob(ctx, path, storage.MetadataObjectType) if err != nil { - return 0, fmt.Errorf("open blob %s: %w", path, err) + return HeaderStoreStats{}, fmt.Errorf("open blob %s: %w", path, err) } if err := blob.Put(ctx, data); err != nil { - return 0, fmt.Errorf("put blob %s: %w", path, err) + return HeaderStoreStats{}, fmt.Errorf("put blob %s: %w", path, err) } - return int64(len(data)), nil + return stats, nil } // Deserialize reads a header from a storage Blob (legacy API). diff --git a/packages/shared/pkg/storage/header/serialization_v4.go b/packages/shared/pkg/storage/header/serialization_v4.go index a88a772ede..cffef6814a 100644 --- a/packages/shared/pkg/storage/header/serialization_v4.go +++ b/packages/shared/pkg/storage/header/serialization_v4.go @@ -43,10 +43,12 @@ type v4SerializableBuildInfo struct { // serializeV4 writes [Metadata] [uint8 flags] [uint32 LZ4 size] [LZ4( Builds[] + Mappings[] )]. // Frame tables are sparse-trimmed to only frames referenced by mappings. -func serializeV4(metadata *Metadata, builds map[uuid.UUID]BuildData, mappings []BuildMap, incomplete bool) ([]byte, error) { +// Returns the serialized bytes and the uncompressed inner-block size (the +// LZ4 input length), so callers can report a meaningful compression ratio. +func serializeV4(metadata *Metadata, builds map[uuid.UUID]BuildData, mappings []BuildMap, incomplete bool) ([]byte, int64, error) { var metaBuf bytes.Buffer if err := binary.Write(&metaBuf, binary.LittleEndian, metadata); err != nil { - return nil, fmt.Errorf("failed to write metadata: %w", err) + return nil, 0, fmt.Errorf("failed to write metadata: %w", err) } var block bytes.Buffer @@ -61,7 +63,7 @@ func serializeV4(metadata *Metadata, builds map[uuid.UUID]BuildData, mappings [] }) if err := binary.Write(&block, binary.LittleEndian, uint32(len(buildIDs))); err != nil { - return nil, fmt.Errorf("failed to write build count: %w", err) + return nil, 0, fmt.Errorf("failed to write build count: %w", err) } buildRanges := extractRelevantRanges(mappings) @@ -75,17 +77,17 @@ func serializeV4(metadata *Metadata, builds map[uuid.UUID]BuildData, mappings [] } if err := binary.Write(&block, binary.LittleEndian, &entry); err != nil { - return nil, fmt.Errorf("failed to write build info: %w", err) + return nil, 0, fmt.Errorf("failed to write build info: %w", err) } trimmed := bd.FrameData.TrimToRanges(buildRanges[id]) if err := trimmed.Serialize(&block); err != nil { - return nil, fmt.Errorf("failed to write build frame data: %w", err) + return nil, 0, fmt.Errorf("failed to write build frame data: %w", err) } } if err := binary.Write(&block, binary.LittleEndian, uint32(len(mappings))); err != nil { - return nil, fmt.Errorf("failed to write mappings count: %w", err) + return nil, 0, fmt.Errorf("failed to write mappings count: %w", err) } for _, mapping := range mappings { @@ -97,7 +99,7 @@ func serializeV4(metadata *Metadata, builds map[uuid.UUID]BuildData, mappings [] } if err := binary.Write(&block, binary.LittleEndian, v4); err != nil { - return nil, fmt.Errorf("failed to write block mapping: %w", err) + return nil, 0, fmt.Errorf("failed to write block mapping: %w", err) } } @@ -105,7 +107,7 @@ func serializeV4(metadata *Metadata, builds map[uuid.UUID]BuildData, mappings [] blockBytes := block.Bytes() compressed, err := compressLZ4(blockBytes) if err != nil { - return nil, fmt.Errorf("failed to LZ4-compress v4 header block: %w", err) + return nil, 0, fmt.Errorf("failed to LZ4-compress v4 header block: %w", err) } var flags uint8 @@ -119,7 +121,7 @@ func serializeV4(metadata *Metadata, builds map[uuid.UUID]BuildData, mappings [] binary.LittleEndian.PutUint32(result[metadataSize+v4FlagsLen:], uint32(len(blockBytes))) copy(result[metadataSize+v4FlagsLen+v4SizePrefixLen:], compressed) - return result, nil + return result, int64(len(blockBytes)), nil } // deserializeV4 decompresses and reads the V4 block. From 66d10fe67b8997d331d33ccfd2915f4f1b41c686 Mon Sep 17 00:00:00 2001 From: ValentaTomas Date: Wed, 20 May 2026 18:51:53 -0700 Subject: [PATCH 2/3] minimize diff: drop stats struct, return ints directly --- .../pkg/sandbox/upload_metrics.go | 9 ++- .../pkg/storage/header/serialization.go | 75 +++++++------------ .../pkg/storage/header/serialization_v4.go | 3 +- 3 files changed, 33 insertions(+), 54 deletions(-) diff --git a/packages/orchestrator/pkg/sandbox/upload_metrics.go b/packages/orchestrator/pkg/sandbox/upload_metrics.go index 72cf509495..780b20f90a 100644 --- a/packages/orchestrator/pkg/sandbox/upload_metrics.go +++ b/packages/orchestrator/pkg/sandbox/upload_metrics.go @@ -47,13 +47,16 @@ func uploadRatioBp(compressed, uncompressed int64) int64 { } func storeHeaderWithMetrics(ctx context.Context, store storage.StorageProvider, path, fileType string, h *headers.Header) error { - stats, err := headers.StoreHeader(ctx, store, path, h) + stored, uncompressed, err := headers.StoreHeader(ctx, store, path, h) if err != nil { return err } - cfg := storage.CompressConfig{Type: stats.Compression.String()} - recordUploadCompression(ctx, uploadArtifactHeader, fileType, cfg, stats.Uncompressed, stats.Stored) + cfg := storage.CompressConfig{} + if h.Metadata.Version >= headers.MetadataVersionV4 { + cfg.Type = storage.CompressionLZ4.String() + } + recordUploadCompression(ctx, uploadArtifactHeader, fileType, cfg, uncompressed, stored) return nil } diff --git a/packages/shared/pkg/storage/header/serialization.go b/packages/shared/pkg/storage/header/serialization.go index 16e59760fc..f30f6dbb51 100644 --- a/packages/shared/pkg/storage/header/serialization.go +++ b/packages/shared/pkg/storage/header/serialization.go @@ -8,52 +8,18 @@ import ( "github.com/e2b-dev/infra/packages/shared/pkg/storage" ) -// HeaderStoreStats reports byte sizes for a serialized header. Stored is what -// was written to storage; Uncompressed is the same payload sized as if the -// inner LZ4 block had not been compressed (V3 has no inner compression, so -// Stored == Uncompressed). Compression is the inner-block codec actually used. -type HeaderStoreStats struct { - Stored int64 - Uncompressed int64 - Compression storage.CompressionType -} - // SerializeHeader serializes a header, dispatching to the version-specific format. // // V3 (Version <= 3): [Metadata] [v3 mappings…] // V4 (Version >= 4): [Metadata] [uint8 flags] [uint32 uncompressedSize] [LZ4( Builds + v4 mappings )] func SerializeHeader(h *Header) ([]byte, error) { - data, _, err := serializeHeaderWithStats(h) - - return data, err -} - -func serializeHeaderWithStats(h *Header) ([]byte, HeaderStoreStats, error) { if h.Metadata.Version <= 3 { - data, err := serializeV3(h.Metadata, h.Mapping) - if err != nil { - return nil, HeaderStoreStats{}, err - } - - return data, HeaderStoreStats{ - Stored: int64(len(data)), - Uncompressed: int64(len(data)), - Compression: storage.CompressionNone, - }, nil + return serializeV3(h.Metadata, h.Mapping) } - data, blockUncompressed, err := serializeV4(h.Metadata, h.Builds, h.Mapping, h.IncompletePendingUpload) - if err != nil { - return nil, HeaderStoreStats{}, err - } - - overhead := int64(metadataSize + v4FlagsLen + v4SizePrefixLen) + data, _, err := serializeV4(h.Metadata, h.Builds, h.Mapping, h.IncompletePendingUpload) - return data, HeaderStoreStats{ - Stored: int64(len(data)), - Uncompressed: overhead + blockUncompressed, - Compression: storage.CompressionLZ4, - }, nil + return data, err } // DeserializeBytes auto-detects the header version and deserializes accordingly. @@ -93,33 +59,44 @@ func LoadHeader(ctx context.Context, s storage.StorageProvider, path string) (*H return DeserializeBytes(data) } -// StoreHeader serializes a header, uploads it, and returns its size stats. -// Refuses to persist a header still flagged as in-flight — the upload pipeline -// must clear IncompletePendingUpload before reaching here. -func StoreHeader(ctx context.Context, s storage.StorageProvider, path string, h *Header) (HeaderStoreStats, error) { +// StoreHeader serializes a header, uploads it, and returns the stored and +// pre-LZ4 (uncompressed) byte counts. V3 has no inner compression so the two +// match. Refuses to persist a header still flagged as in-flight. +func StoreHeader(ctx context.Context, s storage.StorageProvider, path string, h *Header) (stored, uncompressed int64, err error) { if h == nil { - return HeaderStoreStats{}, errors.New("header is nil") + return 0, 0, errors.New("header is nil") } if h.IncompletePendingUpload { - return HeaderStoreStats{}, fmt.Errorf("refusing to persist incomplete header for %s", path) + return 0, 0, fmt.Errorf("refusing to persist incomplete header for %s", path) } - data, stats, err := serializeHeaderWithStats(h) - if err != nil { - return HeaderStoreStats{}, fmt.Errorf("serialize header: %w", err) + var data []byte + if h.Metadata.Version <= 3 { + data, err = serializeV3(h.Metadata, h.Mapping) + if err != nil { + return 0, 0, fmt.Errorf("serialize header: %w", err) + } + uncompressed = int64(len(data)) + } else { + var blockUncompressed int64 + data, blockUncompressed, err = serializeV4(h.Metadata, h.Builds, h.Mapping, h.IncompletePendingUpload) + if err != nil { + return 0, 0, fmt.Errorf("serialize header: %w", err) + } + uncompressed = int64(metadataSize+v4FlagsLen+v4SizePrefixLen) + blockUncompressed } blob, err := s.OpenBlob(ctx, path, storage.MetadataObjectType) if err != nil { - return HeaderStoreStats{}, fmt.Errorf("open blob %s: %w", path, err) + return 0, 0, fmt.Errorf("open blob %s: %w", path, err) } if err := blob.Put(ctx, data); err != nil { - return HeaderStoreStats{}, fmt.Errorf("put blob %s: %w", path, err) + return 0, 0, fmt.Errorf("put blob %s: %w", path, err) } - return stats, nil + return int64(len(data)), uncompressed, nil } // Deserialize reads a header from a storage Blob (legacy API). diff --git a/packages/shared/pkg/storage/header/serialization_v4.go b/packages/shared/pkg/storage/header/serialization_v4.go index cffef6814a..e89dab83e6 100644 --- a/packages/shared/pkg/storage/header/serialization_v4.go +++ b/packages/shared/pkg/storage/header/serialization_v4.go @@ -43,8 +43,7 @@ type v4SerializableBuildInfo struct { // serializeV4 writes [Metadata] [uint8 flags] [uint32 LZ4 size] [LZ4( Builds[] + Mappings[] )]. // Frame tables are sparse-trimmed to only frames referenced by mappings. -// Returns the serialized bytes and the uncompressed inner-block size (the -// LZ4 input length), so callers can report a meaningful compression ratio. +// Also returns the uncompressed inner-block size (LZ4 input length). func serializeV4(metadata *Metadata, builds map[uuid.UUID]BuildData, mappings []BuildMap, incomplete bool) ([]byte, int64, error) { var metaBuf bytes.Buffer if err := binary.Write(&metaBuf, binary.LittleEndian, metadata); err != nil { From 112f606e931d1f89542174c704b0effb0c68bf65 Mon Sep 17 00:00:00 2001 From: ValentaTomas Date: Wed, 20 May 2026 19:07:04 -0700 Subject: [PATCH 3/3] compose header compression config in StoreHeader Returns the cfg up so the metrics caller no longer duplicates the V4-knows-LZ4 dispatch. --- .../pkg/sandbox/upload_metrics.go | 6 +---- .../pkg/storage/header/serialization.go | 24 ++++++++++--------- 2 files changed, 14 insertions(+), 16 deletions(-) diff --git a/packages/orchestrator/pkg/sandbox/upload_metrics.go b/packages/orchestrator/pkg/sandbox/upload_metrics.go index 780b20f90a..6041a5f9f9 100644 --- a/packages/orchestrator/pkg/sandbox/upload_metrics.go +++ b/packages/orchestrator/pkg/sandbox/upload_metrics.go @@ -47,15 +47,11 @@ func uploadRatioBp(compressed, uncompressed int64) int64 { } func storeHeaderWithMetrics(ctx context.Context, store storage.StorageProvider, path, fileType string, h *headers.Header) error { - stored, uncompressed, err := headers.StoreHeader(ctx, store, path, h) + cfg, stored, uncompressed, err := headers.StoreHeader(ctx, store, path, h) if err != nil { return err } - cfg := storage.CompressConfig{} - if h.Metadata.Version >= headers.MetadataVersionV4 { - cfg.Type = storage.CompressionLZ4.String() - } recordUploadCompression(ctx, uploadArtifactHeader, fileType, cfg, uncompressed, stored) return nil diff --git a/packages/shared/pkg/storage/header/serialization.go b/packages/shared/pkg/storage/header/serialization.go index f30f6dbb51..cf0017d92b 100644 --- a/packages/shared/pkg/storage/header/serialization.go +++ b/packages/shared/pkg/storage/header/serialization.go @@ -59,44 +59,46 @@ func LoadHeader(ctx context.Context, s storage.StorageProvider, path string) (*H return DeserializeBytes(data) } -// StoreHeader serializes a header, uploads it, and returns the stored and -// pre-LZ4 (uncompressed) byte counts. V3 has no inner compression so the two -// match. Refuses to persist a header still flagged as in-flight. -func StoreHeader(ctx context.Context, s storage.StorageProvider, path string, h *Header) (stored, uncompressed int64, err error) { +// StoreHeader serializes a header, uploads it, and returns the effective +// compression config plus the stored and pre-compression byte counts. V3 has +// no inner compression so the counts match and cfg is the zero value. Refuses +// to persist a header still flagged as in-flight. +func StoreHeader(ctx context.Context, s storage.StorageProvider, path string, h *Header) (cfg storage.CompressConfig, stored, uncompressed int64, err error) { if h == nil { - return 0, 0, errors.New("header is nil") + return storage.CompressConfig{}, 0, 0, errors.New("header is nil") } if h.IncompletePendingUpload { - return 0, 0, fmt.Errorf("refusing to persist incomplete header for %s", path) + return storage.CompressConfig{}, 0, 0, fmt.Errorf("refusing to persist incomplete header for %s", path) } var data []byte if h.Metadata.Version <= 3 { data, err = serializeV3(h.Metadata, h.Mapping) if err != nil { - return 0, 0, fmt.Errorf("serialize header: %w", err) + return storage.CompressConfig{}, 0, 0, fmt.Errorf("serialize header: %w", err) } uncompressed = int64(len(data)) } else { var blockUncompressed int64 data, blockUncompressed, err = serializeV4(h.Metadata, h.Builds, h.Mapping, h.IncompletePendingUpload) if err != nil { - return 0, 0, fmt.Errorf("serialize header: %w", err) + return storage.CompressConfig{}, 0, 0, fmt.Errorf("serialize header: %w", err) } uncompressed = int64(metadataSize+v4FlagsLen+v4SizePrefixLen) + blockUncompressed + cfg.Type = storage.CompressionLZ4.String() } blob, err := s.OpenBlob(ctx, path, storage.MetadataObjectType) if err != nil { - return 0, 0, fmt.Errorf("open blob %s: %w", path, err) + return storage.CompressConfig{}, 0, 0, fmt.Errorf("open blob %s: %w", path, err) } if err := blob.Put(ctx, data); err != nil { - return 0, 0, fmt.Errorf("put blob %s: %w", path, err) + return storage.CompressConfig{}, 0, 0, fmt.Errorf("put blob %s: %w", path, err) } - return int64(len(data)), uncompressed, nil + return cfg, int64(len(data)), uncompressed, nil } // Deserialize reads a header from a storage Blob (legacy API).