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 that the
next backup trusted.

On every remote with a server-side move, an object is now written under
a name ending in `.partial` and moved onto its key with rclone's
operations.Move, which removes an object already at the key first;
drive, dropbox and others will not move onto one. Listings skip
`.partial` names. Rclone's own copy also requires the 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.

Model: opus-5-5

README.md

@@ -428,6 +428,16 @@ a local or mounted filesystem. Useful for testing or backing up to a NAS.
 **Rclone** (`rclone://remote/path`): Uses rclone's 70+ supported cloud
 providers. Requires rclone to be configured separately (`rclone config`).
 
+An upload cut off part-way leaves nothing under the object's name on S3, which
+shows an object only once its upload has completed, and on the local filesystem
+backend, which writes a temporary file and renames it into place. Rclone
+remotes with a server-side move (local and sftp among them) are written under a
+temporary name ending in `.partial` and moved into place. Rclone remotes without
+one are written in place: where such a remote shows a file while it is still
+being written, a killed upload can leave a truncated object under its name,
+which a later backup takes for complete. A leftover `.partial` file is ignored
+and can be deleted.
+
 Legacy S3 configuration via `s3.*` fields (endpoint, bucket, prefix, etc.) is
 still supported for backward compatibility. `storage_url` takes precedence if
 both are set.

TODO.md

@@ -22,6 +22,26 @@ the tag exists and is exercised; what is left is merging `next` to
 
 # Completed Steps
 
+- 2026-10-07: Stopped a killed rclone upload from leaving a truncated
+  object under its key
+  ([issue #266](https://git.eeqj.de/sneak/vaultik/issues/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 with `Stat`, skipped the upload and recorded
+  a snapshot that could not be restored. On a remote with a server-side
+  move, an object is now written under a temporary name ending in
+  `.partial` and moved into place, and listings skip such names. Remotes
+  without one are still written in place.
+
+- 2026-10-07: Made an interrupted command exit 130 and say so
+  ([issue #267](https://git.eeqj.de/sneak/vaultik/issues/267)). Ctrl-C
+  or SIGTERM during `snapshot create`, `snapshot restore` or `snapshot
+  verify` exited 0 with no error line (`snapshot verify --json` exited
+  1, also without one), so a `--cron` run that never finished looked
+  like a success. A command stopped by either signal now exits 130 and
+  prints `interrupted before the command finished` on stderr, under
+  `--cron` and `--json` too.
+
 - 2026-10-07: Corrected documentation, help text and comments that were
   false about the code
   ([issue #233](https://git.eeqj.de/sneak/vaultik/issues/233)). A blob

internal/cli/app.go

@@ -206,6 +206,10 @@ func RunApp(ctx context.Context, app *fx.App) error {
 // RunOperation through cobra to Entry.
 var errReported = errors.New("operation failed")
 
+// errInterrupted marks an operation that SIGINT or SIGTERM stopped
+// before it finished. Entry shows it and returns exitCodeInterrupted.
+var errInterrupted = errors.New("interrupted before the command finished")
+
 // RunOperation runs op against the Vaultik instance inside the fx app
 // and turns a failure into a returned error rather than an os.Exit from
 // within the goroutine. An os.Exit there skipped main's deferred
@@ -220,17 +224,21 @@ var errReported = errors.New("operation failed")
 // interrupt OnStop cancels op and waits for the goroutine to return, so
 // op's cleanup (removing decrypted scratch files) runs before the
 // process exits; the wait is bounded by shutdownTimeout. report is
-// called with a non-canceled failure so the caller can show it to the
+// called with a failure so the caller can show it to the user before
-// user before it becomes errReported. A context cancellation is the
+// it becomes errReported.
-// interrupt path, not a failure: it is neither reported nor counted as
+//
-// one.
+// The run counts as interrupted unless op returned, without an
+// interrupt having cancelled it, before RunWithApp returned. An
+// interrupted op is not reported, whatever it returned; RunOperation
+// returns errInterrupted instead.
 func RunOperation(
 	ctx context.Context, opts AppOptions,
 	op func(v *vaultik.Vaultik) error, report func(err error),
 ) error {
 	var (
-		mu     sync.Mutex
+		mu       sync.Mutex
-		failed bool
+		finished bool // op returned before any interrupt cancelled it
+		failed   bool // op finished with an error
 	)
 
 	opts.Invokes = append(opts.Invokes,
@@ -241,11 +249,21 @@ func RunOperation(
 				OnStart: func(_ context.Context) error {
 					stop = v.StartOperation(func() {
 						err := op(v)
-						if err != nil && !errors.Is(err, context.Canceled) {
+
-							report(err)
+						// Only stop, called from OnStop below, cancels the
+						// Vaultik context, so a live context means no
+						// interrupt cancelled op. Check the context, not
+						// err: an interrupted op need not return
+						// context.Canceled (`snapshot verify --json`
+						// returns a verification failure).
+						if v.Context().Err() == nil {
+							if err != nil {
+								report(err)
+							}
 
 							mu.Lock()
-							failed = true
+							finished = true
+							failed = err != nil
 							mu.Unlock()
 						}
 
@@ -278,16 +296,23 @@ func RunOperation(
 		return err
 	}
 
-	// The goroutine sets failed before triggering the shutdown that lets
+	// RunWithApp returns only after the app was asked to stop, either by
-	// RunWithApp return, so the write is in place by the time we read it.
+	// an interrupt or by the goroutine's Shutdown call. When op finished
+	// without being cancelled, the goroutine set finished before that
+	// call. So if finished is unset here, an interrupt stopped the app,
+	// and op either returned after it was cancelled or is still running
+	// because the shutdown timed out.
 	mu.Lock()
 	defer mu.Unlock()
 
-	if failed {
+	switch {
+	case !finished:
+		return errInterrupted
+	case failed:
 		return errReported
+	default:
+		return nil
 	}
-
-	return nil
 }
 
 // runVaultikApp runs the standard single-operation command lifecycle

internal/cli/entry.go

@@ -15,6 +15,11 @@ import (
 // the startup banner.
 const shortCommitLen = 12
 
+// exitCodeInterrupted is the exit status of a command that SIGINT or
+// SIGTERM stopped. It is 128 plus SIGINT's number, 2, which is what a
+// shell reports for a command stopped by Ctrl-C.
+const exitCodeInterrupted = 130
+
 // Entry is the main entry point for the CLI application.
 // It prints the startup banner to stderr (unless a banner-suppressing
 // flag is present in os.Args — see bannerSuppressedInArgs), executes the
@@ -23,9 +28,10 @@ const shortCommitLen = 12
 // The banner goes to stderr because stdout carries only the output the
 // user asked for, such as a completion script or a `config get` value.
 //
-// It returns the process exit code (0 on success, 1 on error) rather
+// It returns the process exit code (0 on success, 130 when interrupted,
-// than calling os.Exit, so that main's deferred profile writers run
+// 1 on any other error) rather than calling os.Exit, so that main's
-// before the process ends. See run in cmd/vaultik/main.go.
+// deferred profile writers run before the process ends. See run in
+// cmd/vaultik/main.go.
 func Entry() int {
 	emitStartupBanner(os.Args[1:], os.Stderr)
 
@@ -39,11 +45,16 @@ func Entry() int {
 		// document instead); errReported says so. Printing it again
 		// here would double the error line.
 		// Every other error — bad arguments, a config that would not
-		// load — reaches Entry unreported, so it is shown here.
+		// load, an interrupt — reaches Entry unreported, so it is shown
+		// here.
 		if !errors.Is(err, errReported) {
 			ReportErrorf("%s", err.Error())
 		}
 
+		if errors.Is(err, errInterrupted) {
+			return exitCodeInterrupted
+		}
+
 		return 1
 	}
 

internal/cli/entry_interrupt_test.go

@@ -0,0 +1,203 @@
+package cli //nolint:testpackage // shares runEntry and the argument constants
+
+import (
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/adrg/xdg"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// stalledStoreConfig is hermeticConfig with an s3:// destination store
+// in place of the file:// one. The server behind it accepts any
+// credentials.
+const stalledStoreConfig = `age_recipients:
+  - age1278m9q7dp3chsh2dcy82qk27v047zywyvtxwnj4cvt0z65jw6a7q5dqhfj
+snapshots:
+  test:
+    paths:
+      - %s
+storage_url: s3://bucket?endpoint=%s&ssl=false
+s3:
+  access_key_id: key
+  secret_access_key: secret
+index_path: %s
+hostname: test-host
+`
+
+// interruptRepeat is how often interruptOnFirstRequest sends SIGINT.
+const interruptRepeat = 50 * time.Millisecond
+
+// TestEntryInterruptedRun sends SIGINT to the test process while a
+// command waits on the destination store, and checks that Entry returns
+// 130 and prints one line on stderr saying the run was interrupted. The
+// store is a local HTTP server that holds every request open, so the
+// command is always mid-operation when the signal arrives. The two
+// cases cover --cron and --json, which silence other output.
+//
+// Not parallel: it signals the process and replaces os.Args, os.Stdout,
+// os.Stderr and the xdg globals.
+//
+//nolint:paralleltest // signals the process and replaces process globals
+func TestEntryInterruptedRun(t *testing.T) {
+	for _, testCase := range []struct {
+		name string
+		args []string
+	}{
+		{
+			name: "snapshot create --cron",
+			args: []string{cmdSnapshot, cmdCreate, "--cron"},
+		},
+		{
+			name: "snapshot verify --json",
+			args: []string{cmdSnapshot, cmdVerify, someSnapshotID, flagJSON},
+		},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			endpoint, requestArrived := startStalledStore(t)
+			configPath := writeStalledStoreConfig(t, endpoint)
+			interruptOnFirstRequest(t, requestArrived)
+
+			code, _, stderr := runEntry(t,
+				append([]string{flagConfig, configPath}, testCase.args...)...)
+
+			assert.Equal(t, 130, code)
+			assert.Equal(t, 1,
+				strings.Count(stderr, errInterrupted.Error()), stderr)
+		})
+	}
+}
+
+// interruptOnFirstRequest sends SIGINT to the test process every
+// interruptRepeat, from the first request to the destination store until
+// the test ends. One signal is not enough: the command can reach the
+// store before fx has started catching signals. The test catches SIGINT
+// too, so that a signal fx is not catching does not kill the test
+// binary.
+func interruptOnFirstRequest(t *testing.T, requestArrived <-chan struct{}) {
+	t.Helper()
+
+	self, err := os.FindProcess(os.Getpid())
+	require.NoError(t, err)
+
+	caught := make(chan os.Signal, 1)
+	signal.Notify(caught, os.Interrupt)
+
+	testEnded := make(chan struct{})
+	senderDone := make(chan struct{})
+
+	// Stop catching SIGINT only after the sender has returned. The sender
+	// waits for each SIGINT it sends to arrive on caught; one still on
+	// its way after signal.Stop would kill the test binary.
+	t.Cleanup(func() {
+		close(testEnded)
+		<-senderDone
+		signal.Stop(caught)
+	})
+
+	go func() {
+		defer close(senderDone)
+
+		select {
+		case <-requestArrived:
+		case <-testEnded:
+			return
+		}
+
+		ticker := time.NewTicker(interruptRepeat)
+		defer ticker.Stop()
+
+		for {
+			// Empty caught, so that the receive below waits for this
+			// SIGINT rather than an earlier one.
+			select {
+			case <-caught:
+			default:
+			}
+
+			sendErr := self.Signal(os.Interrupt)
+			if sendErr != nil {
+				t.Errorf("sending SIGINT: %v", sendErr)
+
+				return
+			}
+
+			<-caught
+
+			select {
+			case <-testEnded:
+				return
+			case <-ticker.C:
+			}
+		}
+	}()
+}
+
+// startStalledStore starts an HTTP server that never answers: each
+// request is held until the client gives up on it or the test ends.
+// It returns the server's host:port and a channel that receives a value
+// when the first request arrives.
+func startStalledStore(t *testing.T) (string, <-chan struct{}) {
+	t.Helper()
+
+	requestArrived := make(chan struct{}, 1)
+	release := make(chan struct{})
+
+	server := httptest.NewServer(http.HandlerFunc(
+		func(_ http.ResponseWriter, r *http.Request) {
+			select {
+			case requestArrived <- struct{}{}:
+			default:
+			}
+
+			select {
+			case <-r.Context().Done():
+			case <-release:
+			}
+		}))
+
+	// Cleanups run last-registered first, so release lets any held
+	// request return before Close waits for it.
+	t.Cleanup(server.Close)
+	t.Cleanup(func() { close(release) })
+
+	return server.Listener.Addr().String(), requestArrived
+}
+
+// writeStalledStoreConfig writes a config whose destination store is the
+// server at endpoint and whose snapshot source holds one small file, so
+// that `snapshot create` has a blob to upload. Returns the config path.
+func writeStalledStoreConfig(t *testing.T, endpoint string) string {
+	t.Helper()
+
+	dir := t.TempDir()
+	configPath := filepath.Join(dir, "config.yml")
+	sourceDir := filepath.Join(dir, "source")
+
+	require.NoError(t, os.Mkdir(sourceDir, 0o750))
+	require.NoError(t, os.WriteFile(filepath.Join(sourceDir, "file.txt"),
+		[]byte("contents"), 0o600))
+
+	contents := fmt.Sprintf(stalledStoreConfig,
+		sourceDir, endpoint, filepath.Join(dir, "index.sqlite"))
+
+	require.NoError(t,
+		os.WriteFile(configPath, []byte(contents), configFileMode))
+
+	// The PID lock lives under xdg.DataHome, which xdg resolves at
+	// package init; point it at the temp dir so the test neither
+	// touches nor collides with the real one.
+	t.Setenv("XDG_DATA_HOME", filepath.Join(dir, "data"))
+	xdg.Reload()
+	t.Cleanup(xdg.Reload)
+
+	return configPath
+}

internal/storage/conformance_test.go

@@ -14,8 +14,9 @@ import (
 
 // runStorerConformance is the shared Storer contract. Every backend that
 // can run in-process is expected to pass it: TestFileStorer runs it against
-// file://, TestS3Storer against s3://. A new backend inherits this coverage
+// file://, TestS3Storer against s3://, TestRcloneStorer against rclone's
-// by passing its own constructor, so the contract is defined once.
+// local backend. A new backend inherits this coverage by passing its own
+// constructor, so the contract is defined once.
 //
 // It exercises the public Storer interface: round-trip, stat, list with
 // prefix filtering, overwrite, delete, delete-of-missing, and not-found on

internal/storage/file.go

@@ -50,9 +50,10 @@ const storageDirPerm = 0o755
 // 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
+// blob. The rclone backend's upload does the same on remotes with a
-// interrupted write is never listed or trusted as a blob; it is otherwise
+// server-side move. List and ListStream skip these files, so a leftover from
-// harmless and is overwritten when the same key is written again.
+// an interrupted write is never listed or trusted as a blob; it is otherwise
+// harmless.
 const tempSuffix = ".partial"
 
 // Put stores data at the specified key.

internal/storage/file_atomic_test.go

@@ -14,6 +14,9 @@ import (
 // errStreamInterrupted stands in for an upload cut off mid-stream.
 var errStreamInterrupted = errors.New("connection reset mid-upload")
 
+// testBlobKey is a key laid out as a blob's key is.
+const testBlobKey = "blobs/aa/bb/aabbccddeeff"
+
 // failingReader yields its data once, then fails.
 type failingReader struct {
 	data []byte
@@ -43,7 +46,7 @@ func TestFileStorer_InterruptedWriteLeavesNoTrustedObject(t *testing.T) {
 	}
 
 	ctx := context.Background()
-	key := "blobs/aa/bb/aabbccddeeff"
+	key := testBlobKey
 
 	err = f.PutWithProgress(ctx, key, &failingReader{data: []byte("partial")}, 4096, nil)
 	if err == nil {
@@ -79,7 +82,7 @@ func TestFileStorer_ListSkipsPartialFiles(t *testing.T) {
 	}
 
 	ctx := context.Background()
-	realKey := "blobs/aa/bb/aabbccddeeff"
+	realKey := testBlobKey
 
 	err = f.Put(ctx, realKey, strings.NewReader("blob-bytes"))
 	if err != nil {

internal/storage/rclone.go

@@ -3,6 +3,7 @@ package storage
 import (
 	"bytes"
 	"context"
+	"crypto/rand"
 	"errors"
 	"fmt"
 	"io"
@@ -68,14 +69,7 @@ func (r *RcloneStorer) Put(ctx context.Context, key string, data io.Reader) erro
 		return fmt.Errorf("reading data: %w", err)
 	}
 
-	// Upload the object
+	return r.upload(ctx, key, bytes.NewReader(buf))
-	_, err = operations.Rcat(ctx, r.fsys, key,
-		io.NopCloser(bytes.NewReader(buf)), time.Now(), nil)
-	if err != nil {
-		return fmt.Errorf("uploading object: %w", err)
-	}
-
-	return nil
 }
 
 // PutWithProgress stores data with progress reporting.
@@ -89,13 +83,7 @@ func (r *RcloneStorer) PutWithProgress(
 		callback: progress,
 	}
 
-	// Upload the object
+	return r.upload(ctx, key, pr)
-	_, err := operations.Rcat(ctx, r.fsys, key, io.NopCloser(pr), time.Now(), nil)
-	if err != nil {
-		return fmt.Errorf("uploading object: %w", err)
-	}
-
-	return nil
 }
 
 // Get retrieves data from the specified key.
@@ -173,6 +161,10 @@ func (r *RcloneStorer) List(ctx context.Context, prefix string) ([]string, error
 
 	err := operations.ListFn(ctx, r.fsys, func(obj fs.Object) {
 		key := obj.Remote()
+		if strings.HasSuffix(key, tempSuffix) {
+			return
+		}
+
 		if prefix == "" || strings.HasPrefix(key, prefix) {
 			keys = append(keys, key)
 		}
@@ -202,6 +194,10 @@ func (r *RcloneStorer) ListStream(
 			}
 
 			key := obj.Remote()
+			if strings.HasSuffix(key, tempSuffix) {
+				return
+			}
+
 			if prefix == "" || strings.HasPrefix(key, prefix) {
 				ch <- ObjectInfo{
 					Key:  key,
@@ -230,6 +226,59 @@ func (r *RcloneStorer) Info() Info {
 	}
 }
 
+// upload writes data to key. Where the remote has a server-side move, it
+// writes under a temporary name ending in tempSuffix and moves the object
+// onto key once it is complete, so a killed upload cannot leave a truncated
+// object at key; a remote without one is written in place. List and
+// ListStream skip a temporary object left behind.
+//
+// rclone's own copy does this only where the remote also sets
+// PartialUploads. That flag is not checked here: hdfs, for one, shows a
+// file while it is written without setting it.
+func (r *RcloneStorer) upload(ctx context.Context, key string, data io.Reader) error {
+	if r.fsys.Features().Move == nil {
+		_, err := operations.Rcat(ctx, r.fsys, key, io.NopCloser(data), time.Now(), nil)
+		if err != nil {
+			return fmt.Errorf("uploading object: %w", err)
+		}
+
+		return nil
+	}
+
+	tempKey := key + "-" + rand.Text() + tempSuffix
+
+	obj, err := operations.Rcat(ctx, r.fsys, tempKey, io.NopCloser(data), time.Now(), nil)
+	if err != nil {
+		// Rcat returns the object it wrote when the written data fails its check.
+		if obj != nil {
+			_ = obj.Remove(ctx)
+		}
+
+		return fmt.Errorf("uploading object: %w", err)
+	}
+
+	// On drive, dropbox, onedrive and others the remote's own move does not
+	// replace an object already at key. operations.Move removes the object
+	// it is given first, and copies where the remote refuses the move.
+	existing, err := r.fsys.NewObject(ctx, key)
+	if errors.Is(err, fs.ErrorObjectNotFound) {
+		existing = nil
+	} else if err != nil {
+		_ = obj.Remove(ctx)
+
+		return fmt.Errorf("looking up existing object: %w", err)
+	}
+
+	_, err = operations.Move(ctx, r.fsys, existing, key, obj)
+	if err != nil {
+		_ = obj.Remove(ctx)
+
+		return fmt.Errorf("moving object into place: %w", err)
+	}
+
+	return nil
+}
+
 // progressReader wraps an io.Reader to track read progress.
 type progressReader struct {
 	reader   io.Reader

internal/storage/rclone_test.go

@@ -1,28 +1,47 @@
 package storage_test
 
 import (
+	"bytes"
 	"context"
 	"errors"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
 	"testing"
 
+	"github.com/rclone/rclone/fs"
+	"github.com/rclone/rclone/fs/config/configmap"
 	"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
+// local filesystem directly without any configured remote, so they run
-// runs entirely in-process.
+// 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 +63,280 @@ func TestNewRcloneStorerConstruction(t *testing.T) {
 	}
 }
 
+// unbufferedUploadContext returns a context in which rclone neither reads
+// ahead of the write nor holds a small upload in memory. Without it, a
+// progress callback runs before anything is written to the remote.
+func unbufferedUploadContext() context.Context {
+	ctx, ci := fs.AddConfig(context.Background())
+	ci.BufferSize = 0
+	ci.StreamingUploadCutoff = 0
+
+	return ctx
+}
+
+// objectAtKeyDuringUpload uploads data to testBlobKey and reports whether
+// an object was at the key before the upload finished. It fails the test
+// unless the key then reads back as the uploaded data.
+func objectAtKeyDuringUpload(
+	ctx context.Context, t *testing.T, s *storage.RcloneStorer,
+) bool {
+	t.Helper()
+
+	data := bytes.Repeat([]byte("blob-bytes"), 1000)
+	seen := false
+
+	err := s.PutWithProgress(ctx, testBlobKey, bytes.NewReader(data),
+		int64(len(data)), func(int64) error {
+			_, statErr := s.Stat(ctx, testBlobKey)
+			if statErr == nil {
+				seen = true
+			}
+
+			return nil
+		})
+	if err != nil {
+		t.Fatalf("PutWithProgress: %v", err)
+	}
+
+	got := readObject(ctx, t, s, testBlobKey)
+	if !bytes.Equal(got, data) {
+		t.Errorf("object read back as %d bytes, want the %d uploaded",
+			len(got), len(data))
+	}
+
+	return seen
+}
+
+// readObject returns the contents of the object at key.
+func readObject(
+	ctx context.Context, t *testing.T, s *storage.RcloneStorer, key string,
+) []byte {
+	t.Helper()
+
+	rc, err := s.Get(ctx, key)
+	if err != nil {
+		t.Fatalf("Get: %v", err)
+	}
+
+	defer func() { _ = rc.Close() }()
+
+	got, err := io.ReadAll(rc)
+	if err != nil {
+		t.Fatalf("reading object: %v", err)
+	}
+
+	return got
+}
+
+// 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) {
+	ctx := unbufferedUploadContext()
+
+	s, err := storage.NewRcloneStorer(ctx, ":local", t.TempDir())
+	if err != nil {
+		t.Fatalf("NewRcloneStorer: %v", err)
+	}
+
+	if objectAtKeyDuringUpload(ctx, t, s) {
+		t.Error("object was at its key before the upload finished")
+	}
+}
+
+// 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)
+	}
+}
+
+// newRcloneStorerOnWrappedLocal registers name as rclone's local backend
+// wrapped by wrap, and builds an rclone backend on it rooted at a fresh
+// temp directory. wrap changes the features the local backend reports, so
+// that it behaves like a remote a unit test cannot reach.
+func newRcloneStorerOnWrappedLocal(
+	ctx context.Context, t *testing.T, name string, wrap func(fs.Fs) fs.Fs,
+) *storage.RcloneStorer {
+	t.Helper()
+
+	fs.Register(&fs.RegInfo{
+		Name: name,
+		NewFs: func(
+			ctx context.Context, _, root string, _ configmap.Mapper,
+		) (fs.Fs, error) {
+			local, err := fs.NewFs(ctx, ":local:"+root)
+			if err != nil {
+				return nil, err
+			}
+
+			return wrap(local), nil
+		},
+	})
+
+	s, err := storage.NewRcloneStorer(ctx, ":"+name, t.TempDir())
+	if err != nil {
+		t.Fatalf("NewRcloneStorer: %v", err)
+	}
+
+	return s
+}
+
+// withoutPartialUploads is rclone's local backend with the PartialUploads
+// flag cleared. Like hdfs, it then has a server-side move and shows a file
+// while it is written, without setting that flag.
+type withoutPartialUploads struct {
+	fs.Fs
+}
+
+func (f *withoutPartialUploads) Features() *fs.Features {
+	features := *f.Fs.Features()
+	features.PartialUploads = false
+
+	return &features
+}
+
+// TestRcloneStorerMovesIntoPlaceWithoutPartialUploads checks that on a
+// remote with a server-side move nothing is at the key until the upload has
+// finished, even when rclone does not mark the remote as showing partial
+// uploads.
+//
+//nolint:paralleltest // NewRcloneStorer installs the process-global rclone config
+func TestRcloneStorerMovesIntoPlaceWithoutPartialUploads(t *testing.T) {
+	ctx := unbufferedUploadContext()
+	s := newRcloneStorerOnWrappedLocal(ctx, t, "withoutpartialuploads",
+		func(local fs.Fs) fs.Fs { return &withoutPartialUploads{Fs: local} })
+
+	if objectAtKeyDuringUpload(ctx, t, s) {
+		t.Error("object was at its key before the upload finished")
+	}
+}
+
+// withoutMove is rclone's local backend reporting no server-side move.
+type withoutMove struct {
+	fs.Fs
+}
+
+func (f *withoutMove) Features() *fs.Features {
+	features := *f.Fs.Features()
+	features.Move = nil
+
+	return &features
+}
+
+// TestRcloneStorerWritesInPlaceWithoutMove checks that on a remote with no
+// server-side move an object is written straight to its key and reads back
+// from there.
+//
+//nolint:paralleltest // NewRcloneStorer installs the process-global rclone config
+func TestRcloneStorerWritesInPlaceWithoutMove(t *testing.T) {
+	ctx := unbufferedUploadContext()
+	s := newRcloneStorerOnWrappedLocal(ctx, t, "withoutmove",
+		func(local fs.Fs) fs.Fs { return &withoutMove{Fs: local} })
+
+	if !objectAtKeyDuringUpload(ctx, t, s) {
+		t.Error("object was not at its key while it was uploaded")
+	}
+}
+
+var errNameConflict = errors.New("an object with this name already exists")
+
+// moveRefusesExisting is rclone's local backend with a server-side move
+// that, like dropbox's or onedrive's, refuses to move onto an existing
+// object.
+type moveRefusesExisting struct {
+	fs.Fs
+}
+
+func (f *moveRefusesExisting) Features() *fs.Features {
+	features := *f.Fs.Features()
+	features.Move = f.move
+
+	return &features
+}
+
+//nolint:ireturn // the signature is rclone's
+func (f *moveRefusesExisting) move(
+	ctx context.Context, src fs.Object, remote string,
+) (fs.Object, error) {
+	_, err := f.NewObject(ctx, remote)
+	if err == nil {
+		return nil, errNameConflict
+	}
+
+	return f.Fs.Features().Move(ctx, src, remote)
+}
+
+// TestRcloneStorerOverwritesWhereMoveRefusesExisting checks that writing a
+// key twice replaces the object on a remote whose server-side move will not
+// replace an existing object.
+//
+//nolint:paralleltest // NewRcloneStorer installs the process-global rclone config
+func TestRcloneStorerOverwritesWhereMoveRefusesExisting(t *testing.T) {
+	ctx := context.Background()
+	s := newRcloneStorerOnWrappedLocal(ctx, t, "moverefusesexisting",
+		func(local fs.Fs) fs.Fs { return &moveRefusesExisting{Fs: local} })
+
+	for _, content := range []string{"first", "second"} {
+		err := s.Put(ctx, testBlobKey, strings.NewReader(content))
+		if err != nil {
+			t.Fatalf("Put %q: %v", content, err)
+		}
+	}
+
+	got := readObject(ctx, t, s, testBlobKey)
+	if string(got) != "second" {
+		t.Errorf("object = %q, want %q", got, "second")
+	}
+}
+
 // 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.