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

internal/storage/rclone.go

@@ -236,8 +236,7 @@ func (r *RcloneStorer) Info() Info {
 // 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 {
-	features := r.fsys.Features()
-	if features.Move == nil {
+	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)
@@ -258,7 +257,19 @@ func (r *RcloneStorer) upload(ctx context.Context, key string, data io.Reader) e
 		return fmt.Errorf("uploading object: %w", err)
 	}
 
-	_, err = features.Move(ctx, obj, key)
+	// 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)
 

internal/storage/rclone_test.go

@@ -4,12 +4,14 @@ 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"
 )
 
@@ -162,6 +164,85 @@ func TestRcloneStorerListSkipsPartialFiles(t *testing.T) {
 	}
 }
 
+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) {
+	fs.Register(&fs.RegInfo{
+		Name: "moverefusesexisting",
+		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 &moveRefusesExisting{Fs: local}, nil
+		},
+	})
+
+	ctx := context.Background()
+
+	s, err := storage.NewRcloneStorer(ctx, ":moverefusesexisting", t.TempDir())
+	if err != nil {
+		t.Fatalf("NewRcloneStorer: %v", err)
+	}
+
+	for _, content := range []string{"first", "second"} {
+		err = s.Put(ctx, testBlobKey, strings.NewReader(content))
+		if err != nil {
+			t.Fatalf("Put %q: %v", content, err)
+		}
+	}
+
+	rc, err := s.Get(ctx, testBlobKey)
+	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)
+	}
+
+	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.