Write rclone uploads under a temporary name and move them into place (closes #266)

The rclone backend wrote each object straight to its key, so killing an
upload to a local or sftp remote left a truncated object there. The next
backup found the key, skipped the upload and recorded a snapshot that
could not be restored.

On every remote with a server-side move, an object is now written under
a name ending in `.partial` and moved onto its key; listings skip such
names. Rclone's own copy also requires the remote's PartialUploads flag;
this does not, because hdfs shows a file while it is written without
setting it. Remotes without a move are written in place, as the README
now says.

Model: opus-5-5

internal/storage/rclone_test.go

@@ -1,28 +1,45 @@
 package storage_test
 
 import (
+	"bytes"
 	"context"
 	"errors"
+	"os"
+	"path/filepath"
+	"strings"
 	"testing"
 
+	"github.com/rclone/rclone/fs"
 	"sneak.berlin/go/vaultik/internal/storage"
 )
 
-// The rclone backend is a thin adapter over the rclone library: it turns a
-// (remote, path) pair into rclone's "remote:path" string, hands it to
-// rclone, and maps rclone's own results back to the Storer interface. What
-// can be tested in-process, without a configured remote or network, is that
-// adapter layer — how the arguments are shaped and how construction errors
-// are reported. The data-plane operations (Put/Get/List/Delete) are rclone's
-// own, exercised against a real provider (drive, s3-via-rclone, ...), which
-// needs a configured remote with credentials and network access and so is
-// out of reach of a unit test. The shared Storer conformance suite therefore
-// runs against the in-process file and s3 backends; the rclone backend
-// inherits that contract once a remote is configured.
-//
 // These tests use rclone's ":local:" on-the-fly backend, which addresses the
-// local filesystem directly without any configured remote, so construction
-// runs entirely in-process.
+// local filesystem directly without any configured remote, so they run
+// entirely in-process. A remote that needs credentials and network access
+// (drive, s3 via rclone, ...) is out of reach of a unit test.
+
+// newRcloneStorer builds an rclone backend on rclone's local backend,
+// rooted at a fresh temp directory.
+//
+//nolint:ireturn // conformance runs against the Storer interface by design
+func newRcloneStorer(t *testing.T) storage.Storer {
+	t.Helper()
+
+	s, err := storage.NewRcloneStorer(context.Background(), ":local", t.TempDir())
+	if err != nil {
+		t.Fatalf("NewRcloneStorer: %v", err)
+	}
+
+	return s
+}
+
+// TestRcloneStorer runs the shared Storer contract against the rclone
+// backend.
+//
+//nolint:paralleltest // NewRcloneStorer installs the process-global rclone config
+func TestRcloneStorer(t *testing.T) {
+	runStorerConformance(t, newRcloneStorer)
+}
 
 // TestNewRcloneStorerConstruction checks that a valid remote constructs a
 // backend and that Info() reports the shaped "remote:path" location.
@@ -44,6 +61,107 @@ func TestNewRcloneStorerConstruction(t *testing.T) {
 	}
 }
 
+// TestRcloneStorerObjectAppearsOnlyWhenComplete checks that on a remote
+// with a server-side move, such as local, nothing is at the key until the
+// upload has finished, so a killed upload cannot leave a truncated object
+// there.
+//
+//nolint:paralleltest // NewRcloneStorer installs the process-global rclone config
+func TestRcloneStorerObjectAppearsOnlyWhenComplete(t *testing.T) {
+	// Without this, rclone reads ahead of the write and holds a small
+	// upload in memory, so the progress callback would run before anything
+	// is written to the remote.
+	ctx, ci := fs.AddConfig(context.Background())
+	ci.BufferSize = 0
+	ci.StreamingUploadCutoff = 0
+
+	s, err := storage.NewRcloneStorer(ctx, ":local", t.TempDir())
+	if err != nil {
+		t.Fatalf("NewRcloneStorer: %v", err)
+	}
+
+	key := testBlobKey
+	data := bytes.Repeat([]byte("blob-bytes"), 1000)
+	seenEarly := false
+
+	err = s.PutWithProgress(ctx, key, bytes.NewReader(data), int64(len(data)),
+		func(int64) error {
+			_, statErr := s.Stat(ctx, key)
+			if statErr == nil {
+				seenEarly = true
+			}
+
+			return nil
+		})
+	if err != nil {
+		t.Fatalf("PutWithProgress: %v", err)
+	}
+
+	if seenEarly {
+		t.Error("object was at its key before the upload finished")
+	}
+
+	info, err := s.Stat(ctx, key)
+	if err != nil {
+		t.Fatalf("Stat after upload: %v", err)
+	}
+
+	if info.Size != int64(len(data)) {
+		t.Errorf("stored size = %d, want %d", info.Size, len(data))
+	}
+}
+
+// TestRcloneStorerListSkipsPartialFiles checks that a temporary file left
+// by a killed upload is never listed as a key.
+//
+//nolint:paralleltest // NewRcloneStorer installs the process-global rclone config
+func TestRcloneStorerListSkipsPartialFiles(t *testing.T) {
+	dir := t.TempDir()
+	ctx := context.Background()
+
+	s, err := storage.NewRcloneStorer(ctx, ":local", dir)
+	if err != nil {
+		t.Fatalf("NewRcloneStorer: %v", err)
+	}
+
+	realKey := testBlobKey
+
+	err = s.Put(ctx, realKey, strings.NewReader("blob-bytes"))
+	if err != nil {
+		t.Fatalf("Put: %v", err)
+	}
+
+	leftover := filepath.Join(dir, realKey+"-123456.partial")
+
+	err = os.WriteFile(leftover, []byte("half"), 0o600)
+	if err != nil {
+		t.Fatalf("writing leftover temp file: %v", err)
+	}
+
+	keys, err := s.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 s.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)
+	}
+}
+
 // TestNewRcloneStorerUnknownRemote checks that a remote that is not in the
 // rclone config fails construction with the ErrRemoteNotFound sentinel,
 // rather than silently returning a backend pointed nowhere.