fix(storage): truncate local files on overwrite

Co-authored-by: Boyce <95002055+tttboy123@users.noreply.github.com>
GitOrigin-RevId: 35cc85a6c6180b32415855a5a810f1ceda10b8b8

Author: 2 people

packages/shared/pkg/storage/storage_fs.go

@@ -118,13 +118,7 @@ func (o *fsObject) WriteTo(ctx context.Context, dst io.Writer) (n int64, err err
 }
 
 func (o *fsObject) Put(_ context.Context, data []byte, _ ...PutOption) error {
-	handle, err := o.getHandle(false)
-	if err != nil {
-		return err
-	}
-	defer handle.Close()
-
-	_, err = io.Copy(handle, bytes.NewReader(data))
+	_, err := o.replaceFrom(bytes.NewReader(data))
 
 	return err
 }
@@ -155,23 +149,27 @@ func (o *fsObject) StoreFile(ctx context.Context, path string, opts ...PutOption
 	}
 	defer r.Close()
 
-	handle, err := o.getHandle(false)
+	srcInfo, err := r.Stat()
 	if err != nil {
-		return nil, [32]byte{}, err
+		return nil, [32]byte{}, fmt.Errorf("failed to stat file %s: %w", path, err)
+	}
+	if dstInfo, statErr := os.Stat(o.path); statErr == nil && os.SameFile(srcInfo, dstInfo) {
+		return nil, [32]byte{}, nil
 	}
-	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"),
-		)
+	n, err := o.replaceFrom(r)
+	if 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) {
@@ -298,6 +296,60 @@ func (o *fsObject) getHandle(checkExistence bool) (*os.File, error) {
 	return handle, nil
 }
 
+func replaceFile(path string, r io.Reader) (int64, error) {
+	dir := filepath.Dir(path)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return 0, err
+	}
+
+	tmp, err := os.CreateTemp(dir, "."+filepath.Base(path)+".tmp-*")
+	if err != nil {
+		return 0, err
+	}
+	tmpPath := tmp.Name()
+	defer os.Remove(tmpPath)
+
+	if err := tmp.Chmod(0o644); err != nil {
+		tmp.Close()
+
+		return 0, err
+	}
+
+	n, err := io.Copy(tmp, r)
+	if err != nil {
+		tmp.Close()
+
+		return n, err
+	}
+	if err := tmp.Close(); err != nil {
+		return n, err
+	}
+
+	if err := os.Rename(tmpPath, path); err != nil {
+		return n, err
+	}
+
+	return n, nil
+}
+
+func (o *fsObject) replaceFrom(r io.Reader) (int64, error) {
+	n, err := replaceFile(o.path, r)
+	if err != nil {
+		return n, err
+	}
+
+	return n, o.clearSizeSidecar()
+}
+
+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.
@@ -312,7 +364,9 @@ func (u *fsPartUploader) Complete(_ context.Context) error {
 		return fmt.Errorf("failed to create directory: %w", err)
 	}
 
-	return os.WriteFile(u.fullPath, u.Assemble(), 0o644)
+	_, err := replaceFile(u.fullPath, bytes.NewReader(u.Assemble()))
+
+	return err
 }
 
 func (o *fsObject) OpenRangeReader(ctx context.Context, offsetU int64, length int64, frameTable *FrameTable) (_ RangeReader, _ Source, err error) {