From b3be2f5449b2f6089a91d4c1664b5267a571acb9 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 24 Aug 2026 19:30:20 -0700 Subject: [PATCH] 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 | 26 +++- weed/command/filer_backup_offset_test.go | 112 ++++++++++++++++++ weed/command/filer_backup_test.go | 79 +++++++++++- weed/command/filer_sync.go | 51 ++++++-- weed/command/filer_sync_offset_test.go | 76 ++++++++++++ weed/command/filer_sync_process_test.go | 1 + weed/replication/replicator_test.go | 4 + weed/replication/sink/azuresink/azure_sink.go | 8 ++ weed/replication/sink/b2sink/b2_sink.go | 4 + weed/replication/sink/filersink/filer_sink.go | 4 + weed/replication/sink/gcssink/gcs_sink.go | 4 + weed/replication/sink/localsink/local_sink.go | 4 + weed/replication/sink/replication_sink.go | 6 + weed/replication/sink/s3sink/s3_sink.go | 4 + 14 files changed, 362 insertions(+), 21 deletions(-) create mode 100644 weed/command/filer_backup_offset_test.go create mode 100644 weed/command/filer_sync_offset_test.go diff --git a/weed/command/filer_backup.go b/weed/command/filer_backup.go index c5adeed08..96b363cf5 100644 --- a/weed/command/filer_backup.go +++ b/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 "" diff --git a/weed/command/filer_backup_offset_test.go b/weed/command/filer_backup_offset_test.go new file mode 100644 index 000000000..7ae18a3c3 --- /dev/null +++ b/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) + } +} diff --git a/weed/command/filer_backup_test.go b/weed/command/filer_backup_test.go index addd06276..640a678bb 100644 --- a/weed/command/filer_backup_test.go +++ b/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) + } +} diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go index 284e037aa..e12ac6dc1 100644 --- a/weed/command/filer_sync.go +++ b/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 { diff --git a/weed/command/filer_sync_offset_test.go b/weed/command/filer_sync_offset_test.go new file mode 100644 index 000000000..835bb70fa --- /dev/null +++ b/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) + } +} diff --git a/weed/command/filer_sync_process_test.go b/weed/command/filer_sync_process_test.go index 477c223b3..73d760cae 100644 --- a/weed/command/filer_sync_process_test.go +++ b/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 } diff --git a/weed/replication/replicator_test.go b/weed/replication/replicator_test.go index 6c4947ef5..35646e950 100644 --- a/weed/replication/replicator_test.go +++ b/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 { diff --git a/weed/replication/sink/azuresink/azure_sink.go b/weed/replication/sink/azuresink/azure_sink.go index 9869d6f34..43374de44 100644 --- a/weed/replication/sink/azuresink/azure_sink.go +++ b/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 diff --git a/weed/replication/sink/b2sink/b2_sink.go b/weed/replication/sink/b2sink/b2_sink.go index c69dffa33..a18af0e4f 100644 --- a/weed/replication/sink/b2sink/b2_sink.go +++ b/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 } diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index a76923c41..0eaca800e 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/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 } diff --git a/weed/replication/sink/gcssink/gcs_sink.go b/weed/replication/sink/gcssink/gcs_sink.go index f0508f1fc..f2aa271b5 100644 --- a/weed/replication/sink/gcssink/gcs_sink.go +++ b/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 } diff --git a/weed/replication/sink/localsink/local_sink.go b/weed/replication/sink/localsink/local_sink.go index 8d1724170..f316b80ed 100644 --- a/weed/replication/sink/localsink/local_sink.go +++ b/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 } diff --git a/weed/replication/sink/replication_sink.go b/weed/replication/sink/replication_sink.go index 80eca0094..2dd9ec009 100644 --- a/weed/replication/sink/replication_sink.go +++ b/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 } diff --git a/weed/replication/sink/s3sink/s3_sink.go b/weed/replication/sink/s3sink/s3_sink.go index 310af9878..380a69d56 100644 --- a/weed/replication/sink/s3sink/s3_sink.go +++ b/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 }