fix(storage): truncate local files on overwrite

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,14 +299,28 @@ 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
 	}
 
 	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.

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)