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)

internal/storage/file_atomic_test.go

@@ -0,0 +1,119 @@
+package storage_test
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"sneak.berlin/go/vaultik/internal/storage"
+)
+
+// errStreamInterrupted stands in for an upload cut off mid-stream.
+var errStreamInterrupted = errors.New("connection reset mid-upload")
+
+// failingReader yields its data once, then fails.
+type failingReader struct {
+	data []byte
+	done bool
+}
+
+func (r *failingReader) Read(p []byte) (int, error) {
+	if r.done {
+		return 0, errStreamInterrupted
+	}
+
+	n := copy(p, r.data)
+	r.done = true
+
+	return n, nil
+}
+
+// TestFileStorer_InterruptedWriteLeavesNoTrustedObject checks that a write
+// cut off mid-stream leaves nothing at the destination key, so a later run
+// cannot Stat a truncated object and trust it as a complete blob.
+func TestFileStorer_InterruptedWriteLeavesNoTrustedObject(t *testing.T) {
+	t.Parallel()
+
+	f, err := storage.NewFileStorer(t.TempDir())
+	if err != nil {
+		t.Fatalf("NewFileStorer: %v", err)
+	}
+
+	ctx := context.Background()
+	key := "blobs/aa/bb/aabbccddeeff"
+
+	err = f.PutWithProgress(ctx, key, &failingReader{data: []byte("partial")}, 4096, nil)
+	if err == nil {
+		t.Fatal("expected the interrupted write to fail, got nil")
+	}
+
+	_, err = f.Stat(ctx, key)
+	if !errors.Is(err, storage.ErrNotFound) {
+		t.Fatalf("expected key absent after interrupted write, got Stat err %v", err)
+	}
+
+	keys, err := f.List(ctx, "blobs/")
+	if err != nil {
+		t.Fatalf("List: %v", err)
+	}
+
+	if len(keys) != 0 {
+		t.Fatalf("expected no keys listed after interrupted write, got %v", keys)
+	}
+}
+
+// TestFileStorer_ListSkipsPartialFiles checks that a leftover temp file (the
+// storage layer names them with a ".partial" suffix) is never surfaced as a
+// key by List or ListStream.
+func TestFileStorer_ListSkipsPartialFiles(t *testing.T) {
+	t.Parallel()
+
+	base := t.TempDir()
+
+	f, err := storage.NewFileStorer(base)
+	if err != nil {
+		t.Fatalf("NewFileStorer: %v", err)
+	}
+
+	ctx := context.Background()
+	realKey := "blobs/aa/bb/aabbccddeeff"
+
+	err = f.Put(ctx, realKey, strings.NewReader("blob-bytes"))
+	if err != nil {
+		t.Fatalf("Put: %v", err)
+	}
+
+	// A stray temp file, as an interrupted write would leave behind.
+	leftover := filepath.Join(base, "blobs/aa/bb/aabbccddeeff-123456.partial")
+
+	err = os.WriteFile(leftover, []byte("half"), 0o600)
+	if err != nil {
+		t.Fatalf("writing leftover temp file: %v", err)
+	}
+
+	keys, err := f.List(ctx, "blobs/")
+	if err != nil {
+		t.Fatalf("List: %v", err)
+	}
+
+	if len(keys) != 1 || keys[0] != realKey {
+		t.Fatalf("List should return only the real key, got %v", keys)
+	}
+
+	var streamed []string
+
+	for obj := range f.ListStream(ctx, "blobs/") {
+		if obj.Err != nil {
+			t.Fatalf("ListStream: %v", obj.Err)
+		}
+
+		streamed = append(streamed, obj.Key)
+	}
+
+	if len(streamed) != 1 || streamed[0] != realKey {
+		t.Fatalf("ListStream should return only the real key, got %v", streamed)
+	}
+}