Write file:// blobs atomically via temp file and rename (closes #130)

The file:// backend streamed each object straight to its final key, so an upload cut off mid-stream left a truncated object there. The next backup saw that Stat succeeded, recorded the blob as complete, and produced a snapshot that reported success but could not be restored.

Writes now go to a temporary file with a .partial suffix in the destination directory, are synced, then renamed onto the key. List and ListStream skip .partial files, so a leftover is never trusted as a blob and is overwritten when the key is written again. S3 PutObject is already atomic.

Disclosure: the containing directory is not synced after the rename, so a host crash right after it could still lose the object on some filesystems.

model: claude-opus-4-8 (implementation, review); claude-fable-5-1 (merge)

internal/storage/file.go

@@ -46,31 +46,18 @@ func (f *FileStorer) SetFilesystem(fs afero.Fs) {
 // storage base path.
 const storageDirPerm = 0o755
 
+// tempSuffix marks a partially written object. writeAtomic streams into a
+// temp file carrying this suffix and only renames it onto the real key once
+// the whole object is on disk, so an interrupted write can never leave a
+// truncated object at the key a later run would Stat and trust as a complete
+// blob. List and ListStream skip these files, so a leftover from an
+// interrupted write is never listed or trusted as a blob; it is otherwise
+// harmless and is overwritten when the same key is written again.
+const tempSuffix = ".partial"
+
 // Put stores data at the specified key.
 func (f *FileStorer) Put(_ context.Context, key string, data io.Reader) error {
-	path := f.fullPath(key)
-
-	// Create parent directories
-	dir := filepath.Dir(path)
-
-	err := f.fs.MkdirAll(dir, storageDirPerm)
-	if err != nil {
-		return fmt.Errorf("creating directories: %w", err)
-	}
-
-	file, err := f.fs.Create(path)
-	if err != nil {
-		return fmt.Errorf("creating file: %w", err)
-	}
-
-	defer func() { _ = file.Close() }()
-
-	_, err = io.Copy(file, data)
-	if err != nil {
-		return fmt.Errorf("writing file: %w", err)
-	}
-
-	return nil
+	return f.writeAtomic(key, data, nil)
 }
 
 // PutWithProgress stores data with progress reporting.
@@ -78,35 +65,7 @@ func (f *FileStorer) PutWithProgress(
 	_ context.Context, key string, data io.Reader,
 	_ int64, progress ProgressCallback,
 ) error {
-	path := f.fullPath(key)
-
-	// Create parent directories
-	dir := filepath.Dir(path)
-
-	err := f.fs.MkdirAll(dir, storageDirPerm)
-	if err != nil {
-		return fmt.Errorf("creating directories: %w", err)
-	}
-
-	file, err := f.fs.Create(path)
-	if err != nil {
-		return fmt.Errorf("creating file: %w", err)
-	}
-
-	defer func() { _ = file.Close() }()
-
-	// Wrap with progress tracking
-	pw := &progressWriter{
-		writer:   file,
-		callback: progress,
-	}
-
-	_, err = io.Copy(pw, data)
-	if err != nil {
-		return fmt.Errorf("writing file: %w", err)
-	}
-
-	return nil
+	return f.writeAtomic(key, data, progress)
 }
 
 // Get retrieves data from the specified key.
@@ -188,7 +147,7 @@ func (f *FileStorer) List(ctx context.Context, prefix string) ([]string, error)
 		default:
 		}
 
-		if !info.IsDir() {
+		if !info.IsDir() && !strings.HasSuffix(info.Name(), tempSuffix) {
 			// Convert back to key (relative path from basePath)
 			relPath, err := filepath.Rel(f.basePath, path)
 			if err != nil {
@@ -245,7 +204,7 @@ func (f *FileStorer) ListStream(ctx context.Context, prefix string) <-chan Objec
 				return nil //nolint:nilerr // continue walking despite errors
 			}
 
-			if !info.IsDir() {
+			if !info.IsDir() && !strings.HasSuffix(info.Name(), tempSuffix) {
 				relPath, err := filepath.Rel(f.basePath, path)
 				if err != nil {
 					ch <- ObjectInfo{Err: fmt.Errorf("computing relative path: %w", err)}
@@ -275,6 +234,72 @@ func (f *FileStorer) Info() Info {
 	}
 }
 
+// writeAtomic streams data into a temp file in the destination directory,
+// fsyncs it, and renames it onto the final key. The key therefore appears
+// only once the whole object has been durably written; a failure part-way
+// leaves a temp file (removed here on the failing path) rather than a
+// truncated object at the key.
+func (f *FileStorer) writeAtomic(
+	key string, data io.Reader, progress ProgressCallback,
+) error {
+	path := f.fullPath(key)
+	dir := filepath.Dir(path)
+
+	err := f.fs.MkdirAll(dir, storageDirPerm)
+	if err != nil {
+		return fmt.Errorf("creating directories: %w", err)
+	}
+
+	tmp, err := afero.TempFile(f.fs, dir, filepath.Base(path)+"-*"+tempSuffix)
+	if err != nil {
+		return fmt.Errorf("creating temp file: %w", err)
+	}
+
+	tmpPath := tmp.Name()
+
+	// Remove the temp file unless the rename below claims it. On the success
+	// path renamed is true, so the deferred Close and Remove are harmless
+	// no-ops on a name that no longer exists.
+	renamed := false
+
+	defer func() {
+		_ = tmp.Close()
+
+		if !renamed {
+			_ = f.fs.Remove(tmpPath)
+		}
+	}()
+
+	var w io.Writer = tmp
+	if progress != nil {
+		w = &progressWriter{writer: tmp, callback: progress}
+	}
+
+	_, err = io.Copy(w, data)
+	if err != nil {
+		return fmt.Errorf("writing file: %w", err)
+	}
+
+	err = tmp.Sync()
+	if err != nil {
+		return fmt.Errorf("syncing temp file: %w", err)
+	}
+
+	err = tmp.Close()
+	if err != nil {
+		return fmt.Errorf("closing temp file: %w", err)
+	}
+
+	err = f.fs.Rename(tmpPath, path)
+	if err != nil {
+		return fmt.Errorf("renaming temp file: %w", err)
+	}
+
+	renamed = true
+
+	return nil
+}
+
 // fullPath returns the full filesystem path for a key.
 func (f *FileStorer) fullPath(key string) string {
 	return filepath.Join(f.basePath, key)