From f7761413a6721978ef485d3b03d735b50d5b8f54 Mon Sep 17 00:00:00 2001 From: tttboy123 <3383341447@qq.com> Date: Fri, 18 Sep 2026 11:06:42 +0800 Subject: [PATCH 1/2] fix(storage): truncate local files on overwrite --- packages/shared/pkg/storage/storage_fs.go | 7 ++- .../shared/pkg/storage/storage_fs_test.go | 52 +++++++++++++++++++ 2 files changed, 58 insertions(+), 1 deletion(-) diff --git a/packages/shared/pkg/storage/storage_fs.go b/packages/shared/pkg/storage/storage_fs.go index 9ed0303e23..926a444a6b 100644 --- a/packages/shared/pkg/storage/storage_fs.go +++ b/packages/shared/pkg/storage/storage_fs.go @@ -290,7 +290,12 @@ func (o *fsObject) getHandle(checkExistence bool) (*os.File, error) { } } - handle, err := os.OpenFile(o.path, os.O_RDWR|os.O_CREATE, 0o644) + flags := os.O_RDWR | os.O_CREATE + if !checkExistence { + flags |= os.O_TRUNC + } + + handle, err := os.OpenFile(o.path, flags, 0o644) if err != nil { return nil, err } diff --git a/packages/shared/pkg/storage/storage_fs_test.go b/packages/shared/pkg/storage/storage_fs_test.go index 176b4dbb90..4b3d49d342 100644 --- a/packages/shared/pkg/storage/storage_fs_test.go +++ b/packages/shared/pkg/storage/storage_fs_test.go @@ -63,6 +63,58 @@ func TestFSPut(t *testing.T) { require.Equal(t, payload, string(data)) } +func TestFSPutTruncatesExistingObject(t *testing.T) { + t.Parallel() + p := newTempProvider(t) + ctx := t.Context() + + obj, err := p.OpenBlob(ctx, "overwrite/dst.txt") + require.NoError(t, err) + + require.NoError(t, obj.Put(ctx, []byte("a longer initial payload"))) + require.NoError(t, obj.Put(ctx, []byte("short"))) + + seekable, ok := obj.(Seekable) + require.True(t, ok) + size, err := seekable.Size(ctx) + require.NoError(t, err) + require.EqualValues(t, len("short"), size) + + data, err := GetBlob(ctx, obj) + require.NoError(t, err) + require.Equal(t, []byte("short"), data) +} + +func TestFSStoreFileTruncatesExistingObject(t *testing.T) { + t.Parallel() + p := newTempProvider(t) + ctx := t.Context() + + srcPath := filepath.Join(t.TempDir(), "src.txt") + require.NoError(t, os.WriteFile(srcPath, []byte("a longer initial payload"), 0o600)) + + obj, err := p.OpenSeekable(ctx, "overwrite/dst.txt") + require.NoError(t, err) + + _, _, err = obj.StoreFile(ctx, srcPath) + require.NoError(t, err) + + const replacement = "short" + require.NoError(t, os.WriteFile(srcPath, []byte(replacement), 0o600)) + _, _, err = obj.StoreFile(ctx, srcPath) + require.NoError(t, err) + + size, err := obj.Size(ctx) + require.NoError(t, err) + require.EqualValues(t, len(replacement), size) + + blob, ok := obj.(Blob) + require.True(t, ok) + data, err := GetBlob(ctx, blob) + require.NoError(t, err) + require.Equal(t, []byte(replacement), data) +} + func TestDelete(t *testing.T) { t.Parallel() p := newTempProvider(t) From b30f27fb9c20de7b0e964333a4fcf9dbe93d0d91 Mon Sep 17 00:00:00 2001 From: tttboy123 <3383341447@qq.com> Date: Fri, 18 Sep 2026 11:56:26 +0800 Subject: [PATCH 2/2] fix(storage): clear stale size metadata on replacement --- packages/shared/pkg/storage/storage_fs.go | 36 +++- .../shared/pkg/storage/storage_fs_test.go | 183 ++++++++++++++---- 2 files changed, 174 insertions(+), 45 deletions(-) diff --git a/packages/shared/pkg/storage/storage_fs.go b/packages/shared/pkg/storage/storage_fs.go index 926a444a6b..8e2edd081f 100644 --- a/packages/shared/pkg/storage/storage_fs.go +++ b/packages/shared/pkg/storage/storage_fs.go @@ -125,8 +125,11 @@ func (o *fsObject) Put(_ context.Context, data []byte, _ ...PutOption) error { defer handle.Close() _, err = io.Copy(handle, bytes.NewReader(data)) + if err != nil { + return err + } - return err + return o.clearSizeSidecar() } func (o *fsObject) StoreFile(ctx context.Context, path string, opts ...PutOption) (*FullFrameTable, [32]byte, error) { @@ -162,16 +165,22 @@ func (o *fsObject) StoreFile(ctx context.Context, path string, opts ...PutOption defer handle.Close() n, err := io.Copy(handle, r) - if err == nil { - logger.L().Debug(ctx, "Stored file to filesystem", - zap.String("object", o.path), - zap.String("source", path), - zap.Int64("size_uncompressed", n), - zap.String("compression", "none"), - ) + if err != nil { + return nil, [32]byte{}, err } - return nil, [32]byte{}, err + if err = o.clearSizeSidecar(); err != nil { + return nil, [32]byte{}, err + } + + logger.L().Debug(ctx, "Stored file to filesystem", + zap.String("object", o.path), + zap.String("source", path), + zap.Int64("size_uncompressed", n), + zap.String("compression", "none"), + ) + + return nil, [32]byte{}, nil } func (o *fsObject) storeFileCompressed(ctx context.Context, localPath string, cfg CompressConfig, sink FrameSink) (*FullFrameTable, [32]byte, error) { @@ -303,6 +312,15 @@ func (o *fsObject) getHandle(checkExistence bool) (*os.File, error) { return handle, nil } +func (o *fsObject) clearSizeSidecar() error { + err := os.Remove(SizeSidecar(o.path)) + if err == nil || os.IsNotExist(err) { + return nil + } + + return fmt.Errorf("failed to remove uncompressed-size sidecar for %s: %w", o.path, err) +} + // fsPartUploader implements partUploader for local filesystem. // Embeds memPartUploader for concurrent-safe part collection, // then writes atomically on Complete. diff --git a/packages/shared/pkg/storage/storage_fs_test.go b/packages/shared/pkg/storage/storage_fs_test.go index 4b3d49d342..13ad853310 100644 --- a/packages/shared/pkg/storage/storage_fs_test.go +++ b/packages/shared/pkg/storage/storage_fs_test.go @@ -1,6 +1,7 @@ package storage import ( + "context" "os" "path/filepath" "testing" @@ -63,56 +64,166 @@ func TestFSPut(t *testing.T) { require.Equal(t, payload, string(data)) } -func TestFSPutTruncatesExistingObject(t *testing.T) { +func TestFSPutReplacesExistingObject(t *testing.T) { t.Parallel() - p := newTempProvider(t) - ctx := t.Context() - obj, err := p.OpenBlob(ctx, "overwrite/dst.txt") - require.NoError(t, err) + tests := []struct { + name string + initial string + replacement string + }{ + {"shorter", "a longer initial payload", "short"}, + {"longer", "short", "a longer replacement payload"}, + {"same_length", "first", "other"}, + {"empty", "a non-empty payload", ""}, + } - require.NoError(t, obj.Put(ctx, []byte("a longer initial payload"))) - require.NoError(t, obj.Put(ctx, []byte("short"))) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + p := newTempProvider(t) + ctx := t.Context() - seekable, ok := obj.(Seekable) - require.True(t, ok) - size, err := seekable.Size(ctx) - require.NoError(t, err) - require.EqualValues(t, len("short"), size) + obj, err := p.OpenBlob(ctx, filepath.Join("overwrite", tc.name+".txt")) + require.NoError(t, err) - data, err := GetBlob(ctx, obj) - require.NoError(t, err) - require.Equal(t, []byte("short"), data) + require.NoError(t, obj.Put(ctx, []byte(tc.initial))) + require.NoError(t, obj.Put(ctx, []byte(tc.replacement))) + + seekable, ok := obj.(Seekable) + require.True(t, ok) + size, err := seekable.Size(ctx) + require.NoError(t, err) + require.EqualValues(t, len(tc.replacement), size) + + data, err := GetBlob(ctx, obj) + require.NoError(t, err) + require.Equal(t, []byte(tc.replacement), data) + }) + } } -func TestFSStoreFileTruncatesExistingObject(t *testing.T) { +func TestFSStoreFileReplacesExistingObject(t *testing.T) { t.Parallel() - p := newTempProvider(t) - ctx := t.Context() - srcPath := filepath.Join(t.TempDir(), "src.txt") - require.NoError(t, os.WriteFile(srcPath, []byte("a longer initial payload"), 0o600)) + tests := []struct { + name string + initial string + replacement string + }{ + {"shorter", "a longer initial payload", "short"}, + {"longer", "short", "a longer replacement payload"}, + {"same_length", "first", "other"}, + {"empty", "a non-empty payload", ""}, + } - obj, err := p.OpenSeekable(ctx, "overwrite/dst.txt") - require.NoError(t, err) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + p := newTempProvider(t) + ctx := t.Context() - _, _, err = obj.StoreFile(ctx, srcPath) - require.NoError(t, err) + srcPath := filepath.Join(t.TempDir(), "src.txt") + require.NoError(t, os.WriteFile(srcPath, []byte(tc.initial), 0o600)) - const replacement = "short" - require.NoError(t, os.WriteFile(srcPath, []byte(replacement), 0o600)) - _, _, err = obj.StoreFile(ctx, srcPath) - require.NoError(t, err) + obj, err := p.OpenSeekable(ctx, filepath.Join("overwrite", tc.name+".txt")) + require.NoError(t, err) - size, err := obj.Size(ctx) - require.NoError(t, err) - require.EqualValues(t, len(replacement), size) + _, _, err = obj.StoreFile(ctx, srcPath) + require.NoError(t, err) - blob, ok := obj.(Blob) - require.True(t, ok) - data, err := GetBlob(ctx, blob) - require.NoError(t, err) - require.Equal(t, []byte(replacement), data) + require.NoError(t, os.WriteFile(srcPath, []byte(tc.replacement), 0o600)) + _, _, err = obj.StoreFile(ctx, srcPath) + require.NoError(t, err) + + size, err := obj.Size(ctx) + require.NoError(t, err) + require.EqualValues(t, len(tc.replacement), size) + + blob, ok := obj.(Blob) + require.True(t, ok) + data, err := GetBlob(ctx, blob) + require.NoError(t, err) + require.Equal(t, []byte(tc.replacement), data) + }) + } +} + +func TestFSUncompressedReplacementClearsCompressedSizeSidecar(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + replace func(t *testing.T, ctx context.Context, obj Seekable, replacement []byte) + }{ + { + name: "put", + replace: func(t *testing.T, ctx context.Context, obj Seekable, replacement []byte) { + t.Helper() + + blob, ok := obj.(Blob) + require.True(t, ok) + require.NoError(t, blob.Put(ctx, replacement)) + }, + }, + { + name: "store_file", + replace: func(t *testing.T, ctx context.Context, obj Seekable, replacement []byte) { + t.Helper() + + replacementPath := filepath.Join(t.TempDir(), "replacement.txt") + require.NoError(t, os.WriteFile(replacementPath, replacement, 0o600)) + _, _, err := obj.StoreFile(ctx, replacementPath) + require.NoError(t, err) + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + p := newTempProvider(t) + ctx := t.Context() + + objectPath := filepath.Join("overwrite", "compressed-object-"+tc.name) + obj, err := p.OpenSeekable(ctx, objectPath) + require.NoError(t, err) + + compressedSrcPath := filepath.Join(t.TempDir(), "compressed-src.txt") + compressedPayload := []byte("a longer payload written through the compressed filesystem path") + require.NoError(t, os.WriteFile(compressedSrcPath, compressedPayload, 0o600)) + + compression := CompressConfig{ + Enabled: true, + Type: CompressionLZ4.String(), + FrameSizeKB: 1, + MinPartSizeMB: 1, + FrameEncodeWorkers: 1, + } + _, _, err = obj.StoreFile(ctx, compressedSrcPath, WithCompressConfig(compression)) + require.NoError(t, err) + + sidecarPath := SizeSidecar(p.getPath(objectPath)) + _, err = os.Stat(sidecarPath) + require.NoError(t, err) + + replacement := []byte("short") + tc.replace(t, ctx, obj, replacement) + + size, err := obj.Size(ctx) + require.NoError(t, err) + require.EqualValues(t, len(replacement), size) + + blob, ok := obj.(Blob) + require.True(t, ok) + data, err := GetBlob(ctx, blob) + require.NoError(t, err) + require.Equal(t, replacement, data) + + _, err = os.Stat(sidecarPath) + require.True(t, os.IsNotExist(err)) + }) + } } func TestDelete(t *testing.T) {