filer.backup, filer.sync: stop sharing resume checkpoints across destinations (#10934)

* filer.backup: key the checkpoint by source path and sink destination

The checkpoint id hashed only sink name + directory, so two backups to
different buckets or endpoints sharing a directory layout advanced one
checkpoint: whichever job was running pushed the shared offset forward,
and a stopped or failing job later resumed from the other's position,
silently skipping changes. Backups of different source paths to the same
destination shared a checkpoint the same way.

Each sink now reports a destination identity (endpoint or account,
bucket or container, directory) and the checkpoint is keyed by the
source path plus that identity. Reads fall back to the historical
name+directory key when the new key has no value, so existing backups
resume where they left off; writes go only to the new key.

* filer.sync: include the target path in the offset key

The offset stored on the target filer was keyed by source path and
source filer signature only, so two syncs from the same source cluster
and path to different directories on the same target cluster advanced
one shared checkpoint, and the slower one could resume past events it
never applied. The target path now participates in the key; "/" keeps
the historical form, and a sync with a non-root target path falls back
to the historical key once when its own key has no value yet.

* join checkpoint key fields with NUL so they cannot alias

A path or configuration value spelling out the separator could
concatenate two different field tuples to the same checkpoint key.
NUL cannot appear in a CLI path argument or any sane configuration
value, making the encoding injective.

weed/command/filer_backup.go

@@ -137,14 +137,14 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 
 	// get start time for the data sink
 	startFrom := time.Unix(0, 0)
-	sinkId := util.HashStringToLong(dataSink.GetName() + dataSink.GetSinkToDirectory())
+	sinkId, legacySinkId := backupCheckpointIds(sourcePath, dataSink)
 	runSnapshot := *backupOption.initialSnapshot && timeAgo == 0
 	if timeAgo == 0 {
 		if runSnapshot {
 			// snapshot below sets the start point; no checkpoint read needed
 			glog.V(0).Infof("initialSnapshot requested — walking live tree before subscribing")
 		} else {
-			lastOffsetTsNs, err := getOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, int32(sinkId))
+			lastOffsetTsNs, err := getOffsetWithFallback(grpcDialOption, sourceFiler, BackupKeyPrefix, sinkId, BackupKeyPrefix, legacySinkId)
 			if err != nil {
 				glog.V(0).Infof("starting from %v (offset read failed: %v)", startFrom, err)
 			} else if lastOffsetTsNs > 0 {
@@ -186,7 +186,7 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 		// The walk can take hours on large trees; retry the tiny KV write a
 		// handful of times before giving up so a flaky filer KV doesn't force
 		// the whole walk to repeat on the next retry loop iteration.
-		if err := persistSnapshotOffset(grpcDialOption, sourceFiler, int32(sinkId), snapshotTsNs); err != nil {
+		if err := persistSnapshotOffset(grpcDialOption, sourceFiler, sinkId, snapshotTsNs); err != nil {
 			glog.Errorf("initialSnapshot: FAILED to persist offset %d for sinkId %d after retries: %v — the next retry will redo the full walk", snapshotTsNs, sinkId, err)
 			return fmt.Errorf("persist initial snapshot offset: %w", err)
 		}
@@ -220,7 +220,7 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 
 	processEventFnWithOffset := pb.AddOffsetFunc(processEventFn, 3*time.Second, func(counter int64, lastTsNs int64) error {
 		glog.V(0).Infof("backup %s progressed to %v %0.2f/sec", sourceFiler, time.Unix(0, lastTsNs), float64(counter)/float64(3))
-		return setOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, int32(sinkId), lastTsNs)
+		return setOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, sinkId, lastTsNs)
 	})
 
 	if dataSink.IsIncremental() && *filerBackupOptions.retentionDays > 0 {
@@ -262,6 +262,24 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 
 }
 
+// backupCheckpointIds derives the checkpoint key for this backup and the
+// historical key kept for fallback reads. The checkpoint covers everything
+// that defines one backup stream: what is backed up (the source path) and
+// exactly where it lands (the sink's destination identity). The historical
+// key hashed only sink name + directory, so two backups to different buckets
+// or endpoints sharing a directory layout advanced one checkpoint and could
+// silently skip each other's changes. Writes go only to the new key; the
+// historical key is read once when the new key has no value yet, so an
+// existing backup resumes where it left off after an upgrade.
+//
+// NUL joins the fields because it cannot occur in a path or a configuration
+// value, so distinct field tuples cannot concatenate to the same hash input.
+func backupCheckpointIds(sourcePath string, dataSink sink.ReplicationSink) (sinkId, legacySinkId int32) {
+	sinkId = int32(util.HashStringToLong(sourcePath + "\x00" + dataSink.GetName() + "\x00" + dataSink.GetDestinationIdentity()))
+	legacySinkId = int32(util.HashStringToLong(dataSink.GetName() + dataSink.GetSinkToDirectory()))
+	return
+}
+
 func getSourceKey(resp *filer_pb.SubscribeMetadataResponse) string {
 	if resp == nil || resp.EventNotification == nil {
 		return ""

weed/command/filer_backup_offset_test.go

@@ -0,0 +1,112 @@
+package command
+
+import (
+	"context"
+	"net"
+	"sync"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// kvFilerServer implements just the KV calls of the filer gRPC API over an
+// in-memory map, mirroring the real server's behavior of returning an empty
+// response, not an error, for a missing key.
+type kvFilerServer struct {
+	filer_pb.UnimplementedSeaweedFilerServer
+	mu sync.Mutex
+	kv map[string][]byte
+}
+
+func (s *kvFilerServer) KvGet(_ context.Context, req *filer_pb.KvGetRequest) (*filer_pb.KvGetResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return &filer_pb.KvGetResponse{Value: s.kv[string(req.Key)]}, nil
+}
+
+func (s *kvFilerServer) KvPut(_ context.Context, req *filer_pb.KvPutRequest) (*filer_pb.KvPutResponse, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.kv[string(req.Key)] = req.Value
+	return &filer_pb.KvPutResponse{}, nil
+}
+
+func startKvFiler(t *testing.T) (pb.ServerAddress, grpc.DialOption) {
+	t.Helper()
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	grpcServer := grpc.NewServer()
+	filer_pb.RegisterSeaweedFilerServer(grpcServer, &kvFilerServer{kv: map[string][]byte{}})
+	go func() { _ = grpcServer.Serve(listener) }()
+	t.Cleanup(func() { grpcServer.Stop() })
+	grpcPort := listener.Addr().(*net.TCPAddr).Port
+	return pb.NewServerAddressWithGrpcPort("127.0.0.1:8888", grpcPort),
+		grpc.WithTransportCredentials(insecure.NewCredentials())
+}
+
+// An upgraded backup finds no destination-scoped checkpoint yet and must
+// resume from the historical name+directory key; once its own key holds a
+// value, the historical key — possibly still advanced by another backup that
+// shared it — must no longer influence it.
+func TestBackupOffset_HistoricalFallbackThenIndependence(t *testing.T) {
+	filerAddr, dial := startKvFiler(t)
+
+	s := &stubSink{name: "s3", dir: "/", destination: "endpoint-a|bucket-a|/"}
+	sinkId, legacySinkId := backupCheckpointIds("/", s)
+
+	// pre-upgrade state: only the shared historical checkpoint exists
+	if err := setOffset(dial, filerAddr, BackupKeyPrefix, legacySinkId, 111); err != nil {
+		t.Fatalf("seed historical offset: %v", err)
+	}
+	got, err := getOffsetWithFallback(dial, filerAddr, BackupKeyPrefix, sinkId, BackupKeyPrefix, legacySinkId)
+	if err != nil || got != 111 {
+		t.Fatalf("fallback read = (%d, %v), want (111, nil)", got, err)
+	}
+
+	// the backup checkpoints under its own key and prefers it from then on
+	if err := setOffset(dial, filerAddr, BackupKeyPrefix, sinkId, 222); err != nil {
+		t.Fatalf("write destination-scoped offset: %v", err)
+	}
+	got, err = getOffsetWithFallback(dial, filerAddr, BackupKeyPrefix, sinkId, BackupKeyPrefix, legacySinkId)
+	if err != nil || got != 222 {
+		t.Fatalf("destination-scoped read = (%d, %v), want (222, nil)", got, err)
+	}
+
+	// another backup advancing the historical key no longer moves this one
+	if err := setOffset(dial, filerAddr, BackupKeyPrefix, legacySinkId, 333); err != nil {
+		t.Fatalf("advance historical offset: %v", err)
+	}
+	got, err = getOffsetWithFallback(dial, filerAddr, BackupKeyPrefix, sinkId, BackupKeyPrefix, legacySinkId)
+	if err != nil || got != 222 {
+		t.Fatalf("read after historical advance = (%d, %v), want (222, nil)", got, err)
+	}
+}
+
+// The collision scenario from the report, over the real KV wire path: two
+// backups whose configurations differ only in endpoint and bucket write
+// their checkpoints without disturbing each other's.
+func TestBackupOffset_DistinctDestinationsDoNotShare(t *testing.T) {
+	filerAddr, dial := startKvFiler(t)
+
+	idA, legacyA := backupCheckpointIds("/", &stubSink{name: "s3", dir: "/", destination: "s3.us-west-004.backblazeb2.com|seaweed-backup-a|/"})
+	idB, legacyB := backupCheckpointIds("/", &stubSink{name: "s3", dir: "/", destination: "s3.us-east-005.backblazeb2.com|seaweed-backup-b|/"})
+
+	if err := setOffset(dial, filerAddr, BackupKeyPrefix, idA, 1000); err != nil {
+		t.Fatalf("write A: %v", err)
+	}
+	if err := setOffset(dial, filerAddr, BackupKeyPrefix, idB, 2000); err != nil {
+		t.Fatalf("write B: %v", err)
+	}
+
+	if got, err := getOffsetWithFallback(dial, filerAddr, BackupKeyPrefix, idA, BackupKeyPrefix, legacyA); err != nil || got != 1000 {
+		t.Fatalf("A reads (%d, %v), want (1000, nil)", got, err)
+	}
+	if got, err := getOffsetWithFallback(dial, filerAddr, BackupKeyPrefix, idB, BackupKeyPrefix, legacyB); err != nil || got != 2000 {
+		t.Fatalf("B reads (%d, %v), want (2000, nil)", got, err)
+	}
+}

weed/command/filer_backup_test.go

@@ -213,14 +213,17 @@ func TestEventSourceSuperseded_Guards(t *testing.T) {
 }
 
 // stubSink is a minimal ReplicationSink used to exercise initialSnapshotTargetKey
-// without standing up a real sink. Only the two methods read by the key builder
-// (GetName, IsIncremental) need meaningful behavior; the rest satisfy the interface.
+// and backupCheckpointIds without standing up a real sink; the methods those
+// read (GetName, IsIncremental, GetSinkToDirectory, GetDestinationIdentity)
+// reflect the fields, the rest satisfy the interface.
 type stubSink struct {
 	name          string
+	dir           string
+	destination   string
 	isIncremental bool
 }
 
-func (s *stubSink) GetName() string                           { return s.name }
+func (s *stubSink) GetName() string                             { return s.name }
 func (s *stubSink) Initialize(util.Configuration, string) error { return nil }
 func (s *stubSink) DeleteEntry(string, bool, bool, []int32) error {
 	return nil
@@ -229,9 +232,10 @@ func (s *stubSink) CreateEntry(string, *filer_pb.Entry, []int32) error { return
 func (s *stubSink) UpdateEntry(string, *filer_pb.Entry, string, *filer_pb.Entry, bool, []int32) (bool, error) {
 	return false, nil
 }
-func (s *stubSink) GetSinkToDirectory() string       { return "" }
+func (s *stubSink) GetSinkToDirectory() string         { return s.dir }
+func (s *stubSink) GetDestinationIdentity() string     { return s.destination }
 func (s *stubSink) SetSourceFiler(*source.FilerSource) {}
-func (s *stubSink) IsIncremental() bool              { return s.isIncremental }
+func (s *stubSink) IsIncremental() bool                { return s.isIncremental }
 
 var _ sink.ReplicationSink = (*stubSink)(nil)
 
@@ -274,3 +278,68 @@ func TestInitialSnapshotTargetKey(t *testing.T) {
 		t.Errorf("sourceKey == sourcePath (trailing slash mismatch): got %q, want %q", got, "/backup")
 	}
 }
+
+// The scenario from the collision report: two backups to different S3
+// endpoints/buckets that share the destination directory "/" must not share
+// a checkpoint, or the stopped one resumes from the other's position and
+// skips changes.
+func TestBackupCheckpointIds_DistinctDestinations(t *testing.T) {
+	backupA := &stubSink{name: "s3", dir: "/", destination: "s3.us-west-004.backblazeb2.com|seaweed-backup-a|/"}
+	backupB := &stubSink{name: "s3", dir: "/", destination: "s3.us-east-005.backblazeb2.com|seaweed-backup-b|/"}
+
+	idA, legacyA := backupCheckpointIds("/", backupA)
+	idB, legacyB := backupCheckpointIds("/", backupB)
+
+	if idA == idB {
+		t.Errorf("backups to different destinations share checkpoint id %d", idA)
+	}
+	// Both historically hashed to the same key — that is the bug the
+	// destination-scoped key fixes, and the shared value both fall back to.
+	if legacyA != legacyB {
+		t.Errorf("legacy ids differ: %d vs %d", legacyA, legacyB)
+	}
+}
+
+// Two backups of different source paths to the same destination must not
+// share a checkpoint either: each stream sees a different event subset, so a
+// shared offset lets the faster one push the slower one past unseen events.
+func TestBackupCheckpointIds_DistinctSourcePaths(t *testing.T) {
+	s := &stubSink{name: "s3", dir: "/", destination: "endpoint|bucket|/"}
+	idA, _ := backupCheckpointIds("/buckets/a", s)
+	idB, _ := backupCheckpointIds("/buckets/b", s)
+	if idA == idB {
+		t.Errorf("backups of different source paths share checkpoint id %d", idA)
+	}
+}
+
+// The fallback key must keep the exact historical formula
+// hash(GetName() + GetSinkToDirectory()) truncated to int32, or existing
+// backups lose their checkpoint on upgrade and replay from zero.
+func TestBackupCheckpointIds_LegacyFormulaUnchanged(t *testing.T) {
+	s := &stubSink{name: "s3", dir: "/data", destination: "endpoint|bucket|/data"}
+	_, legacy := backupCheckpointIds("/", s)
+	if want := int32(util.HashStringToLong("s3" + "/data")); legacy != want {
+		t.Errorf("legacy id = %d, want historical formula value %d", legacy, want)
+	}
+}
+
+// The NUL joins keep the hash input injective: field values spelling out
+// other fields' content must not concatenate to the same input.
+func TestBackupCheckpointIds_NoAliasing(t *testing.T) {
+	idA, _ := backupCheckpointIds("/src", &stubSink{name: "s3", dir: "/", destination: "/d=>s3|/other"})
+	idB, _ := backupCheckpointIds("/src=>s3|/d", &stubSink{name: "s3", dir: "/", destination: "/other"})
+	if idA == idB {
+		t.Errorf("field content spelling a separator aliases checkpoint id %d", idA)
+	}
+}
+
+// Restarting the same configuration must derive the same key, or every
+// restart would orphan its checkpoint.
+func TestBackupCheckpointIds_Stable(t *testing.T) {
+	s := &stubSink{name: "s3", dir: "/", destination: "endpoint|bucket|/"}
+	id1, legacy1 := backupCheckpointIds("/buckets/a", s)
+	id2, legacy2 := backupCheckpointIds("/buckets/a", s)
+	if id1 != id2 || legacy1 != legacy2 {
+		t.Errorf("ids not stable: (%d,%d) vs (%d,%d)", id1, legacy1, id2, legacy2)
+	}
+}

weed/command/filer_sync.go

@@ -74,6 +74,7 @@ type syncState struct {
 	grpcDialOption       grpc.DialOption
 	targetFiler          pb.ServerAddress
 	sourcePath           string
+	targetPath           string
 	sourceFilerSignature int32
 }
 
@@ -217,7 +218,7 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 			if offsetTsNs == 0 {
 				return
 			}
-			if err := setOffset(state.grpcDialOption, state.targetFiler, getSignaturePrefixByPath(state.sourcePath), state.sourceFilerSignature, offsetTsNs); err != nil {
+			if err := setOffset(state.grpcDialOption, state.targetFiler, getSignaturePrefixByPath(state.sourcePath, state.targetPath), state.sourceFilerSignature, offsetTsNs); err != nil {
 				glog.Errorf("failed to save checkpoint for %s on shutdown: %v", name, err)
 			} else {
 				glog.V(0).Infof("saved checkpoint for %s on shutdown: %v", name, time.Unix(0, offsetTsNs))
@@ -231,7 +232,7 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 	go func() {
 		// a->b
 		// set synchronization start timestamp to offset
-		initOffsetError := initOffsetFromTsMs(grpcDialOptionB, filerB, aFilerSignature, *syncOptions.aFromTsMs, getSignaturePrefixByPath(*syncOptions.aPath))
+		initOffsetError := initOffsetFromTsMs(grpcDialOptionB, filerB, aFilerSignature, *syncOptions.aFromTsMs, getSignaturePrefixByPath(*syncOptions.aPath, *syncOptions.bPath))
 		if initOffsetError != nil {
 			glog.Errorf("init offset from timestamp %d error from %s to %s: %v", *syncOptions.aFromTsMs, *syncOptions.filerA, *syncOptions.filerB, initOffsetError)
 			os.Exit(2)
@@ -273,7 +274,7 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 	if !*syncOptions.isActivePassive {
 		// b->a
 		// set synchronization start timestamp to offset
-		initOffsetError := initOffsetFromTsMs(grpcDialOptionA, filerA, bFilerSignature, *syncOptions.bFromTsMs, getSignaturePrefixByPath(*syncOptions.bPath))
+		initOffsetError := initOffsetFromTsMs(grpcDialOptionA, filerA, bFilerSignature, *syncOptions.bFromTsMs, getSignaturePrefixByPath(*syncOptions.bPath, *syncOptions.aPath))
 		if initOffsetError != nil {
 			glog.Errorf("init offset from timestamp %d error from %s to %s: %v", *syncOptions.bFromTsMs, *syncOptions.filerB, *syncOptions.filerA, initOffsetError)
 			os.Exit(2)
@@ -339,7 +340,10 @@ func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDi
 
 	// if first time, start from now
 	// if has previously synced, resume from that point of time
-	sourceFilerOffsetTsNs, err := getOffset(targetGrpcDialOption, targetFiler, getSignaturePrefixByPath(sourcePath), sourceFilerSignature)
+	// the historical key ignored the target path; falling back to it lets a
+	// sync that predates target-scoped keys survive the upgrade
+	signaturePrefix := getSignaturePrefixByPath(sourcePath, targetPath)
+	sourceFilerOffsetTsNs, err := getOffsetWithFallback(targetGrpcDialOption, targetFiler, signaturePrefix, sourceFilerSignature, getSignaturePrefixByPath(sourcePath, "/"), sourceFilerSignature)
 	if err != nil {
 		return err
 	}
@@ -387,6 +391,7 @@ func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDi
 			grpcDialOption:       targetGrpcDialOption,
 			targetFiler:          targetFiler,
 			sourcePath:           sourcePath,
+			targetPath:           targetPath,
 			sourceFilerSignature: sourceFilerSignature,
 		})
 	}
@@ -420,7 +425,7 @@ func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDi
 		lastProgressedTsNs = offsetTsNs
 		// collect synchronous offset
 		statsCollect.FilerSyncOffsetGauge.WithLabelValues(sourceFiler.String(), targetFiler.String(), clientName, sourcePath).Set(float64(offsetTsNs))
-		return setOffset(targetGrpcDialOption, targetFiler, getSignaturePrefixByPath(sourcePath), sourceFilerSignature, offsetTsNs)
+		return setOffset(targetGrpcDialOption, targetFiler, signaturePrefix, sourceFilerSignature, offsetTsNs)
 	})
 
 	prefix := sourcePath
@@ -452,14 +457,22 @@ func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDi
 
 }
 
-// When each business is distinguished according to path, and offsets need to be maintained separately.
-func getSignaturePrefixByPath(path string) string {
-	// compatible historical version
-	if path == "/" {
-		return SyncKeyPrefix
-	} else {
-		return SyncKeyPrefix + path
+// Offsets are kept per (source path, target path): two syncs from the same
+// source to different directories on the same target filer see the same events
+// but progress independently, so a shared key would let one push the other
+// past events it never processed. "/" contributes nothing on either side,
+// keeping the historical key form so existing deployments resume unchanged.
+func getSignaturePrefixByPath(sourcePath, targetPath string) string {
+	prefix := SyncKeyPrefix
+	if sourcePath != "/" {
+		prefix += sourcePath
 	}
+	if targetPath != "/" {
+		// NUL cannot occur in a path, so the combined key can alias neither a
+		// source-only key nor another (source, target) pair
+		prefix += "\x00" + targetPath
+	}
+	return prefix
 }
 
 func getOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, signaturePrefix string, signature int32) (lastOffsetTsNs int64, readErr error) {
@@ -489,6 +502,20 @@ func getOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, signature
 
 }
 
+// getOffsetWithFallback reads the offset under the current key, falling back
+// to the historical key when the current one has no value yet, so streams
+// created before a checkpoint key-scheme change resume where they left off
+// instead of replaying from zero. Writes must go only to the current key:
+// keeping the historical key warm would re-create the very sharing between
+// streams that a key-scheme change separates.
+func getOffsetWithFallback(grpcDialOption grpc.DialOption, filer pb.ServerAddress, signaturePrefix string, signature int32, legacySignaturePrefix string, legacySignature int32) (int64, error) {
+	lastOffsetTsNs, err := getOffset(grpcDialOption, filer, signaturePrefix, signature)
+	if err == nil && lastOffsetTsNs == 0 {
+		lastOffsetTsNs, err = getOffset(grpcDialOption, filer, legacySignaturePrefix, legacySignature)
+	}
+	return lastOffsetTsNs, err
+}
+
 func setOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, signaturePrefix string, signature int32, offsetTsNs int64) error {
 	return pb.WithFilerClient(false, signature, filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 

weed/command/filer_sync_offset_test.go

@@ -0,0 +1,76 @@
+package command
+
+import "testing"
+
+// The offset key is a persistence contract: a changed form orphans saved
+// checkpoints, so the historical forms must survive verbatim and the
+// target-scoped form must stay stable.
+func TestGetSignaturePrefixByPath_KeyForms(t *testing.T) {
+	cases := []struct {
+		sourcePath, targetPath, want string
+	}{
+		// historical forms, kept so existing deployments resume unchanged
+		{"/", "/", "sync."},
+		{"/data", "/", "sync./data"},
+		// the target path participates once it deviates from "/"
+		{"/", "/backup-a", "sync.\x00/backup-a"},
+		{"/data", "/backup-a", "sync./data\x00/backup-a"},
+	}
+	for _, tc := range cases {
+		if got := getSignaturePrefixByPath(tc.sourcePath, tc.targetPath); got != tc.want {
+			t.Errorf("getSignaturePrefixByPath(%q, %q) = %q, want %q", tc.sourcePath, tc.targetPath, got, tc.want)
+		}
+	}
+}
+
+// The NUL separator keeps the key injective: a source directory whose name
+// spells out a separator must not alias a different (source, target) pair.
+func TestGetSignaturePrefixByPath_NoAliasing(t *testing.T) {
+	if a, b := getSignaturePrefixByPath("/archive=>/backup", "/"), getSignaturePrefixByPath("/archive", "/backup"); a == b {
+		t.Errorf("source path spelling a separator aliases key %q", a)
+	}
+}
+
+// Two syncs from the same source filer and path to different directories on
+// the same target filer must keep separate checkpoints, or the running one
+// pushes the shared offset past events the other never processed.
+func TestGetSignaturePrefixByPath_DistinctTargets(t *testing.T) {
+	a := getSignaturePrefixByPath("/", "/backup-a")
+	b := getSignaturePrefixByPath("/", "/backup-b")
+	if a == b {
+		t.Errorf("different target paths share offset key %q", a)
+	}
+	if root := getSignaturePrefixByPath("/", "/"); a == root {
+		t.Errorf("non-root target path shares offset key %q with the root form", a)
+	}
+}
+
+// Over the real KV wire path: a sync that predates target-scoped keys resumes
+// from the historical key, and once each sync checkpoints under its own key
+// they no longer disturb each other.
+func TestSyncOffset_HistoricalFallbackThenIndependence(t *testing.T) {
+	filerAddr, dial := startKvFiler(t)
+	const sourceSig = int32(12345)
+	historical := getSignaturePrefixByPath("/data", "/")
+	prefixA := getSignaturePrefixByPath("/data", "/backup-a")
+	prefixB := getSignaturePrefixByPath("/data", "/backup-b")
+
+	// pre-upgrade state: both syncs shared the historical key
+	if err := setOffset(dial, filerAddr, historical, sourceSig, 111); err != nil {
+		t.Fatalf("seed historical offset: %v", err)
+	}
+	if got, err := getOffsetWithFallback(dial, filerAddr, prefixA, sourceSig, historical, sourceSig); err != nil || got != 111 {
+		t.Fatalf("A fallback read = (%d, %v), want (111, nil)", got, err)
+	}
+
+	// A checkpoints under its own key; B still resumes from the historical one
+	if err := setOffset(dial, filerAddr, prefixA, sourceSig, 222); err != nil {
+		t.Fatalf("write A offset: %v", err)
+	}
+	if got, err := getOffsetWithFallback(dial, filerAddr, prefixA, sourceSig, historical, sourceSig); err != nil || got != 222 {
+		t.Fatalf("A read = (%d, %v), want (222, nil)", got, err)
+	}
+	if got, err := getOffsetWithFallback(dial, filerAddr, prefixB, sourceSig, historical, sourceSig); err != nil || got != 111 {
+		t.Fatalf("B read = (%d, %v), want (111, nil)", got, err)
+	}
+}

weed/command/filer_sync_process_test.go

@@ -57,6 +57,7 @@ func equalSyncStrings(a, b []string) bool {
 	return true
 }
 func (s *recordingSyncSink) GetSinkToDirectory() string         { return "/dest" }
+func (s *recordingSyncSink) GetDestinationIdentity() string     { return "/dest" }
 func (s *recordingSyncSink) SetSourceFiler(*source.FilerSource) {}
 func (s *recordingSyncSink) IsIncremental() bool                { return s.incremental }
 

weed/replication/replicator_test.go

@@ -71,6 +71,10 @@ func (s *recordingSink) GetSinkToDirectory() string {
 	return s.sinkToDirectory
 }
 
+func (s *recordingSink) GetDestinationIdentity() string {
+	return s.sinkToDirectory
+}
+
 func (s *recordingSink) SetSourceFiler(*source.FilerSource) {}
 
 func (s *recordingSink) IsIncremental() bool {

weed/replication/sink/azuresink/azure_sink.go

@@ -24,6 +24,8 @@ import (
 
 type AzureSink struct {
 	client        *azblob.Client
+	accountName   string
+	endpoint      string
 	container     string
 	dir           string
 	filerSource   *source.FilerSource
@@ -42,6 +44,10 @@ func (g *AzureSink) GetSinkToDirectory() string {
 	return g.dir
 }
 
+func (g *AzureSink) GetDestinationIdentity() string {
+	return g.accountName + "\x00" + g.endpoint + "\x00" + g.container + "\x00" + g.dir
+}
+
 func (g *AzureSink) IsIncremental() bool {
 	return g.isIncremental
 }
@@ -63,6 +69,8 @@ func (g *AzureSink) SetSourceFiler(s *source.FilerSource) {
 }
 
 func (g *AzureSink) initialize(accountName, accountKey, clientID, endpoint, container, dir string) error {
+	g.accountName = accountName
+	g.endpoint = endpoint
 	g.container = container
 	g.dir = dir
 

weed/replication/sink/b2sink/b2_sink.go

@@ -34,6 +34,10 @@ func (g *B2Sink) GetSinkToDirectory() string {
 	return g.dir
 }
 
+func (g *B2Sink) GetDestinationIdentity() string {
+	return g.bucket + "\x00" + g.dir
+}
+
 func (g *B2Sink) IsIncremental() bool {
 	return g.isIncremental
 }

weed/replication/sink/filersink/filer_sink.go

@@ -75,6 +75,10 @@ func (fs *FilerSink) GetSinkToDirectory() string {
 	return fs.dir
 }
 
+func (fs *FilerSink) GetDestinationIdentity() string {
+	return fs.grpcAddress + "\x00" + fs.dir
+}
+
 func (fs *FilerSink) IsIncremental() bool {
 	return fs.isIncremental
 }

weed/replication/sink/gcssink/gcs_sink.go

@@ -39,6 +39,10 @@ func (g *GcsSink) GetSinkToDirectory() string {
 	return g.dir
 }
 
+func (g *GcsSink) GetDestinationIdentity() string {
+	return g.bucket + "\x00" + g.dir
+}
+
 func (g *GcsSink) IsIncremental() bool {
 	return g.isIncremental
 }

weed/replication/sink/localsink/local_sink.go

@@ -56,6 +56,10 @@ func (localsink *LocalSink) GetSinkToDirectory() string {
 	return localsink.Dir
 }
 
+func (localsink *LocalSink) GetDestinationIdentity() string {
+	return localsink.Dir
+}
+
 func (localsink *LocalSink) IsIncremental() bool {
 	return localsink.isIncremental
 }

weed/replication/sink/replication_sink.go

@@ -13,6 +13,12 @@ type ReplicationSink interface {
 	CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error
 	UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error)
 	GetSinkToDirectory() string
+	// GetDestinationIdentity distinguishes this sink's write destination from
+	// any other destination the same sink type could write to: endpoint or
+	// account, bucket or container, and directory. filer.backup keys its resume
+	// checkpoint on it, so two configurations writing to different places must
+	// not share a value.
+	GetDestinationIdentity() string
 	SetSourceFiler(s *source.FilerSource)
 	IsIncremental() bool
 }

weed/replication/sink/s3sink/s3_sink.go

@@ -53,6 +53,10 @@ func (s3sink *S3Sink) GetSinkToDirectory() string {
 	return s3sink.dir
 }
 
+func (s3sink *S3Sink) GetDestinationIdentity() string {
+	return s3sink.endpoint + "\x00" + s3sink.bucket + "\x00" + s3sink.dir
+}
+
 func (s3sink *S3Sink) IsIncremental() bool {
 	return s3sink.isIncremental
 }