From 5927e1aa3d3f959e72abe87e55688330c60c5c78 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Mon, 21 Sep 2026 21:24:42 +0200 Subject: [PATCH] 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 | 133 ++++++++++++++++----------- internal/storage/file_atomic_test.go | 119 ++++++++++++++++++++++++ 2 files changed, 198 insertions(+), 54 deletions(-) create mode 100644 internal/storage/file_atomic_test.go diff --git a/internal/storage/file.go b/internal/storage/file.go index 36ed92e..5c31898 100644 --- a/internal/storage/file.go +++ b/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) diff --git a/internal/storage/file_atomic_test.go b/internal/storage/file_atomic_test.go new file mode 100644 index 0000000..9006695 --- /dev/null +++ b/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) + } +}