diff --git a/packages/shared/pkg/storage/storage_fs.go b/packages/shared/pkg/storage/storage_fs.go index 9ed0303e23..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 + } + + if err = o.clearSizeSidecar(); err != nil { + return nil, [32]byte{}, err } - 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) { @@ -290,7 +299,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 } @@ -298,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 176b4dbb90..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,6 +64,168 @@ func TestFSPut(t *testing.T) { require.Equal(t, payload, string(data)) } +func TestFSPutReplacesExistingObject(t *testing.T) { + t.Parallel() + + 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", ""}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + p := newTempProvider(t) + ctx := t.Context() + + obj, err := p.OpenBlob(ctx, filepath.Join("overwrite", tc.name+".txt")) + require.NoError(t, err) + + 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 TestFSStoreFileReplacesExistingObject(t *testing.T) { + t.Parallel() + + 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", ""}, + } + + for _, tc := range tests { + t.Run(tc.name, func(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(tc.initial), 0o600)) + + obj, err := p.OpenSeekable(ctx, filepath.Join("overwrite", tc.name+".txt")) + require.NoError(t, err) + + _, _, err = obj.StoreFile(ctx, srcPath) + require.NoError(t, err) + + 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) { t.Parallel() p := newTempProvider(t)