filer.backup: process metadata events through the concurrent processor (#11951)

* filer.backup: process metadata events through the concurrent processor

Each event was handled inline on the subscription loop, so one
fetch+write at a time capped the stream at a few entries per second and
a busy source kept building lag. Route events through the same
MetadataProcessor filer.sync uses: it fans jobs out behind -concurrency
while serializing same-path and subtree-barrier conflicts, and the
checkpoint now advances on the processor's contiguous watermark via
GetResumeTsNs, with Resubscribe replaying anything a failure pins.
-disableErrorRetry keeps its log-and-skip meaning by returning nil after
logging, since the processor would otherwise pin the checkpoint forever.

* filer.backup: drain in-flight jobs before the stream retry loop reconnects

FollowMetadata returns as soon as the stream ends while admitted jobs keep writing; the next doFilerBackup builds a fresh processor that cannot order its replayed writes against them, so an older upload could finish last and clobber a newer file.

* filer.backup: checkpoint the completed watermark on a ticker and at stream end

The offset callback only fires while events arrive, so work draining on a quiet stream stayed uncheckpointed until the next event and a restart recopied it. Persist the processed watermark every 3s when it advances, and once more after the post-stream drain.

* filer.backup: bound the checkpoint wait before reconnecting

setOffset has no context, so a stalled KvPut would hold persistDone and block the retry loop forever. The persist goroutine now performs the final write itself on stopPersist, and the caller waits at most 30s.

* filer.backup: carry the processed position across retry attempts

A follow attempt that ended with a pinned failure recomputed startFrom on retry, so a -timeAgo window could move past the failed event and skip it. doFilerBackup now returns the failure pin (or processed watermark) and the retry loop clamps startFrom to it.

* filer.backup: resume retries at the previous attempt's pinned position

The retry loop threaded resumeTsNs through but never consulted it, so a
-timeAgo run recomputed startFrom as now-minus-the-window on every attempt
— a failed event older than the window was skipped on reconnect. Use the
pinned position when one exists; -timeAgo and the persisted checkpoint
only choose the first starting point.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer.backup: resume below the failure pin so the failed event replays

The metadata reader treats the resume position as exclusive — events at
or below it are skipped — so resuming at the failed event's own timestamp
would skip it. Back one tick off the pin.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer.backup: label the progress line as the received-event rate

The counter increments when a job is admitted, not when its copy
finishes, so the old wording showed backup throughput during a burst of
slow copies while the processed watermark had not moved.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* replication: bound the offset checkpoint RPC

setOffset called KvPut on the background context, so a stalled write
could pin a caller's drain or persistence loop indefinitely.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer.backup: keep the resume pin when filer-source setup fails

A retried attempt carries resumeTsNs so the failed event replays from the
same position, but filer-source and SSE initialization failures returned 0,
so the next attempt recomputed -timeAgo and could move the window past the
pinned event.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* replication: serialize checkpoint writes per key

A checkpoint write can outlive its caller's wait and land beside the next
subscription attempt's writes, moving saved progress backward. Serialize
writes per key inside setOffset and drop any value older than one already
written from this process.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* filer.backup: gofmt

* replication: serialize checkpoint writes per filer and key

A single process-wide lock made a stalled checkpoint RPC block every
stream's checkpoint, including the opposite direction of a bidirectional
filer.sync. Keeping the last-written watermark per persisted key also let
one stream's offset shadow another stream checkpointing the same key to
a different target filer. Lock and track progress per (filer, key).

* replication: keep the resume position behind a cleared failure pin

A redelivery that clears failedTs lets the processed watermark advance
past the earliest failure, but events dropped while the processor was
stopped — possibly sharing the failure's timestamp — have to replay.
Latched earliestFailedTsNs floors the resume position one tick below
that failure so reconnects still cover them.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* replication: persist floored checkpoints through the write ordering

setOffset drops a value older than the last written, which also swallows
a deliberate regression: the resume floor, an operator-seeded offset, or
a snapshot high-water mark. allowRegress marks those callers so the
checkpoint can move backward when a failure forces replay.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* replication: order checkpoint writes by issuance, not value

The value-based stale guard could never fire — every caller passed
allowRegress because floors and seeds are deliberate regressions — and
ordering by value cannot tell a stale high watermark apart from real
progress: a write that outlives its attempt could still land after a
floored resume and skip events owed a replay. Order by issuance instead:
a write issued before the last one to land under the key is dropped, so
late writes from a superseded attempt can no longer move the checkpoint
in either direction while deliberate regressions still apply.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* replication: keep the resume position one tick below the frontier

* remote.sync: resume and persist through the floored watermark

* test: concurrent metadata jobs fan out up to the limit

Distinct-path events must all be in flight before any finishes — a
serial loop could only ever have one running — while the limit still
bounds the pool. Pins the property behind the backup-throughput fix.

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/command/filer_backup.go

@@ -42,6 +42,7 @@ type FilerBackupOptions struct {
 	timeAgo             *time.Duration
 	retentionDays       *int
 	initialSnapshot     *bool
+	concurrency         *int
 }
 
 var (
@@ -65,6 +66,7 @@ func init() {
 	filerBackupOptions.disableErrorRetry = cmdFilerBackup.Flag.Bool("disableErrorRetry", false, "disables errors retry, only logs will print")
 	filerBackupOptions.ignore404Error = cmdFilerBackup.Flag.Bool("ignore404Error", true, "ignore 404 errors from filer")
 	filerBackupOptions.initialSnapshot = cmdFilerBackup.Flag.Bool("initialSnapshot", false, "before subscribing to metadata updates, walk the live filer tree under -filerPath and seed the destination, then subscribe from the walk-start timestamp so concurrent changes are still captured. Runs on every start when -timeAgo is 0 and overwrites the saved checkpoint, so it also re-seeds a reinitialized destination. Remove it once the backup is caught up, otherwise it re-walks the whole tree on each restart.")
+	filerBackupOptions.concurrency = cmdFilerBackup.Flag.Int("concurrency", DefaultConcurrencyLimit, "the maximum number of files that will be backed up concurrently.")
 }
 
 var cmdFilerBackup = &Command{
@@ -105,10 +107,12 @@ func runFilerBackup(cmd *Command, args []string) bool {
 
 	clientId := util.RandomInt32()
 	var clientEpoch int32
+	var resumeTsNs int64
 
 	for {
 		clientEpoch++
-		err := doFilerBackup(grpcDialOption, &filerBackupOptions, reExcludeFileName, excludeFileNames, excludePathPatterns, clientId, clientEpoch)
+		var err error
+		resumeTsNs, err = doFilerBackup(grpcDialOption, &filerBackupOptions, reExcludeFileName, excludeFileNames, excludePathPatterns, clientId, clientEpoch, resumeTsNs)
 		if err != nil {
 			glog.Errorf("backup from %s: %v", *filerBackupOptions.filer, err)
 			time.Sleep(1747 * time.Millisecond)
@@ -120,12 +124,12 @@ const (
 	BackupKeyPrefix = "backup."
 )
 
-func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOptions, reExcludeFileName *regexp.Regexp, excludeFileNames []*wildcard.WildcardMatcher, excludePathPatterns []*wildcard.WildcardMatcher, clientId int32, clientEpoch int32) error {
+func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOptions, reExcludeFileName *regexp.Regexp, excludeFileNames []*wildcard.WildcardMatcher, excludePathPatterns []*wildcard.WildcardMatcher, clientId int32, clientEpoch int32, resumeTsNs int64) (int64, error) {
 
 	// find data sink
 	dataSink := findSink(util.GetViper())
 	if dataSink == nil {
-		return fmt.Errorf("no data sink configured in replication.toml")
+		return 0, fmt.Errorf("no data sink configured in replication.toml")
 	}
 
 	sourceFiler := pb.ServerAddress(*backupOption.filer)
@@ -139,7 +143,13 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 	startFrom := time.Unix(0, 0)
 	sinkId, legacySinkId := backupCheckpointIds(sourcePath, dataSink)
 	runSnapshot := *backupOption.initialSnapshot && timeAgo == 0
-	if timeAgo == 0 {
+	if resumeTsNs > 0 {
+		// The previous attempt pinned where it stopped — a failed event's
+		// timestamp or the processed watermark — which can sit behind both the
+		// persisted checkpoint and a recomputed -timeAgo window.
+		startFrom = time.Unix(0, resumeTsNs)
+		glog.V(0).Infof("resuming previous attempt at %v", startFrom)
+	} else 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")
@@ -166,11 +176,11 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 		sourceFiler.ToGrpcAddress(),
 		sourcePath,
 		*backupOption.proxyByFiler); err != nil {
-		return fmt.Errorf("filersource initialization failed: %v", err)
+		return resumeTsNs, fmt.Errorf("filersource initialization failed: %v", err)
 	}
 
 	if err := repl_util.InitializeSSEForReplication(filerSource); err != nil {
-		return fmt.Errorf("SSE initialization failed: %v", err)
+		return resumeTsNs, fmt.Errorf("SSE initialization failed: %v", err)
 	}
 	dataSink.SetSourceFiler(filerSource)
 
@@ -181,14 +191,14 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 	if runSnapshot {
 		snapshotTsNs, err := runInitialSnapshot(sourceFiler.ToGrpcAddress(), filerSource, sourcePath, targetPath, excludePaths, reExcludeFileName, excludeFileNames, excludePathPatterns, dataSink, *backupOption.ignore404Error)
 		if err != nil {
-			return fmt.Errorf("initial snapshot: %w", err)
+			return 0, fmt.Errorf("initial snapshot: %w", err)
 		}
 		// 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, 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)
+			return 0, fmt.Errorf("persist initial snapshot offset: %w", err)
 		}
 		startFrom = time.Unix(0, snapshotTsNs)
 		// walk once per process; retries resume from the persisted checkpoint
@@ -218,9 +228,74 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 		processEventFn = genProcessFunction(sourcePath, targetPath, excludePaths, reExcludeFileName, excludeFileNames, excludePathPatterns, dataSink, *backupOption.doDeleteFiles, debug)
 	}
 
-	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, sinkId, lastTsNs)
+	if *backupOption.disableErrorRetry {
+		// TrivialOnError semantics: log the failure and advance. Left to the
+		// processor a permanent failure would pin the checkpoint and replay
+		// forever, which is the opposite of what this flag asks for.
+		inner := processEventFn
+		processEventFn = func(resp *filer_pb.SubscribeMetadataResponse) error {
+			if err := inner(resp); err != nil {
+				glog.Errorf("process %v: %v", resp, err)
+			}
+			return nil
+		}
+	}
+
+	// Serial event handling caps throughput at one fetch+write at a time, so a
+	// busy source keeps outrunning the backup. The metadata processor fans
+	// events out while serializing same-path and subtree-barrier conflicts,
+	// and the checkpoint only advances on the contiguous processed watermark.
+	concurrency := *backupOption.concurrency
+	if concurrency < 1 || concurrency > 1024 {
+		glog.Warningf("invalid concurrency value %d, using default: %d", concurrency, DefaultConcurrencyLimit)
+		concurrency = DefaultConcurrencyLimit
+	}
+	processor := NewMetadataProcessor(processEventFn, concurrency, startFrom.UnixNano())
+
+	// AddOffsetFunc only checkpoints while events arrive; persist the
+	// completed watermark on a ticker too so work finishing on a quiet
+	// stream does not sit uncheckpointed until the next event.
+	var lastPersistedTsNs atomic.Int64
+	lastPersistedTsNs.Store(-1)
+	stopPersist := make(chan struct{})
+	persistDone := make(chan struct{})
+	go func() {
+		defer close(persistDone)
+		ticker := time.NewTicker(3 * time.Second)
+		defer ticker.Stop()
+		persist := func() {
+			if ts := resumeTsNsFor(processor); ts != lastPersistedTsNs.Load() {
+				if err := setOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, sinkId, ts); err == nil {
+					lastPersistedTsNs.Store(ts)
+				} else {
+					glog.Warningf("persist backup offset: %v", err)
+				}
+			}
+		}
+		for {
+			select {
+			case <-stopPersist:
+				persist()
+				return
+			case <-ticker.C:
+				persist()
+			}
+		}
+	}()
+
+	var lastLogTsNs = time.Now().UnixNano()
+	processEventFnWithOffset := pb.AddOffsetFunc(func(resp *filer_pb.SubscribeMetadataResponse) error {
+		processor.AddSyncJob(resp)
+		return nil
+	}, 3*time.Second, func(counter int64, lastTsNs int64) error {
+		offsetTsNs := processor.processedTsWatermark.Load()
+		if offsetTsNs == 0 {
+			return nil
+		}
+		now := time.Now().UnixNano()
+		glog.V(0).Infof("backup %s received %0.2f events/sec, processed through %v", sourceFiler, float64(counter)/(float64(now-lastLogTsNs)/1e9), time.Unix(0, offsetTsNs))
+		lastLogTsNs = now
+		return nil
 	})
 
 	if dataSink.IsIncremental() && *filerBackupOptions.retentionDays > 0 {
@@ -256,9 +331,29 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti
 		StartTsNs:              startFrom.UnixNano(),
 		StopTsNs:               0,
 		EventErrorType:         eventErrorType,
+		GetResumeTsNs: func() int64 {
+			return resumeTsNsFor(processor)
+		},
+		Resubscribe: processor.ResubscribeCh(),
 	}
 
-	return pb.FollowMetadata(sourceFiler, grpcDialOption, metadataFollowOption, processEventFnWithOffset)
+	err := pb.FollowMetadata(sourceFiler, grpcDialOption, metadataFollowOption, processEventFnWithOffset)
+	// Jobs admitted before the stream ended keep running; drain them so the
+	// retry loop's replay cannot race an older write past a newer one.
+	processor.Drain()
+	close(stopPersist)
+	// setOffset has no context, so a stalled checkpoint RPC must not hold the
+	// retry loop: bound the wait and let a late write land in the background.
+	select {
+	case <-persistDone:
+	case <-time.After(30 * time.Second):
+		glog.Warningf("backup %s offset checkpoint did not finish in 30s; reconnecting anyway", sourceFiler)
+	}
+	// Resume the next attempt at the floored watermark — never recompute a
+	// position that skips events. The floor sits one tick below the earliest
+	// failure and stays there even when a redelivery clears the pin, so
+	// events dropped while the processor was stopped still replay.
+	return resumeTsNsFor(processor), err
 
 }
 

weed/command/filer_remote_gateway_buckets.go

@@ -48,7 +48,7 @@ func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSo
 		now := time.Now().UnixNano()
 		glog.V(0).Infof("remote sync %s progressed to %v %0.2f/sec", *option.filerAddress, time.Unix(0, offsetTsNs), float64(counter)/(float64(now-lastLogTsNs)/1e9))
 		lastLogTsNs = now
-		return remote_storage.SetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, offsetTsNs)
+		return remote_storage.SetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), option.bucketsDir, resumeTsNsFor(processor))
 	})
 
 	option.clientEpoch++
@@ -65,7 +65,7 @@ func (option *RemoteGatewayOptions) followBucketUpdatesAndUploadToRemote(filerSo
 		StopTsNs:               0,
 		EventErrorType:         pb.RetryForeverOnError,
 		GetResumeTsNs: func() int64 {
-			return processor.processedTsWatermark.Load()
+			return resumeTsNsFor(processor)
 		},
 		Resubscribe: processor.ResubscribeCh(),
 	}

weed/command/filer_remote_sync_dir.go

@@ -55,7 +55,7 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour
 		now := time.Now().UnixNano()
 		glog.V(0).Infof("remote sync %s progressed to %v %0.2f/sec", *option.filerAddress, time.Unix(0, offsetTsNs), float64(counter)/(float64(now-lastLogTsNs)/1e9))
 		lastLogTsNs = now
-		return remote_storage.SetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, offsetTsNs)
+		return remote_storage.SetSyncOffset(option.grpcDialOption, pb.ServerAddress(*option.filerAddress), mountedDir, resumeTsNsFor(processor))
 	})
 
 	option.clientEpoch++
@@ -77,7 +77,7 @@ func followUpdatesAndUploadToRemote(option *RemoteSyncOptions, filerSource *sour
 		StopTsNs:               0,
 		EventErrorType:         pb.RetryForeverOnError,
 		GetResumeTsNs: func() int64 {
-			return processor.processedTsWatermark.Load()
+			return resumeTsNsFor(processor)
 		},
 		Resubscribe: processor.ResubscribeCh(),
 	}

weed/command/filer_sync.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"regexp"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -222,7 +223,7 @@ func runFilerSynchronize(cmd *Command, args []string) bool {
 			if state == nil || state.processor == nil {
 				return
 			}
-			offsetTsNs := state.processor.processedTsWatermark.Load()
+			offsetTsNs := resumeTsNsFor(state.processor)
 			if offsetTsNs == 0 {
 				return
 			}
@@ -475,7 +476,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, signaturePrefix, sourceFilerSignature, offsetTsNs)
+		return setOffset(targetGrpcDialOption, targetFiler, signaturePrefix, sourceFilerSignature, resumeTsNsFor(processor))
 	})
 
 	prefix := sourcePath
@@ -495,7 +496,7 @@ func doSubscribeFilerMetaChanges(clientId int32, clientEpoch int32, sourceGrpcDi
 		StopTsNs:               0,
 		EventErrorType:         pb.RetryForeverOnError,
 		GetResumeTsNs: func() int64 {
-			return processor.processedTsWatermark.Load()
+			return resumeTsNsFor(processor)
 		},
 		Resubscribe: processor.ResubscribeCh(),
 		// While the source has only read activity it emits no metadata events, so
@@ -576,16 +577,77 @@ func getOffsetWithFallback(grpcDialOption grpc.DialOption, filer pb.ServerAddres
 	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 {
+// A checkpoint write can outlive its caller's wait and land beside the next
+// subscription attempt's writes; issued concurrently they may be applied out
+// of order, moving saved progress backward or forward past events still owed
+// a replay. Writes to the same persisted key are serialized and each is
+// ordered by issuance: a write issued before the last one to land is stale
+// and dropped. Deliberate regressions — resume floors, operator seeds — are
+// issued after the writes they supersede, so they still apply. State is keyed
+// by target filer so unrelated streams — or the two directions of a
+// bidirectional sync — never block or shadow each other.
+var offsetCheckpointWrites = struct {
+	sync.Mutex
+	byKey map[string]*offsetCheckpoint
+}{byKey: map[string]*offsetCheckpoint{}}
 
-		syncKey := []byte(signaturePrefix + "____")
-		util.Uint32toBytes(syncKey[len(signaturePrefix):len(signaturePrefix)+4], uint32(signature))
+type offsetCheckpoint struct {
+	sync.Mutex
+	lastSeq int64
+}
+
+// resumeTsNsFor floors the processed watermark by the earliest failure the
+// processor saw: a redelivery can clear the pin, but events dropped while it
+// was stopped only replay if the resume position stays behind it. It also
+// stays one tick below the newest timestamp seen — a watermark on the
+// frontier can have finished every job it admitted while more events at the
+// same timestamp still sit unread in the log, and an exclusive resume there
+// would skip them.
+func resumeTsNsFor(p *MetadataProcessor) int64 {
+	ts := p.processedTsWatermark.Load()
+	if newestTsNs := p.NewestSeenTsNs(); newestTsNs > 0 && ts >= newestTsNs {
+		ts = newestTsNs - 1
+	}
+	if failedTsNs := p.EarliestFailedTsNs(); failedTsNs > 0 && failedTsNs-1 < ts {
+		ts = failedTsNs - 1
+	}
+	return ts
+}
+
+// offsetWriteSeq numbers checkpoint writes in issuance order so a write that
+// lands late can be told apart from the fresher write it would shadow.
+var offsetWriteSeq atomic.Int64
+
+func setOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, signaturePrefix string, signature int32, offsetTsNs int64) error {
+	syncKey := []byte(signaturePrefix + "____")
+	util.Uint32toBytes(syncKey[len(signaturePrefix):len(signaturePrefix)+4], uint32(signature))
+	key := string(filer) + "\x00" + string(syncKey)
+
+	offsetCheckpointWrites.Lock()
+	checkpoint := offsetCheckpointWrites.byKey[key]
+	if checkpoint == nil {
+		checkpoint = &offsetCheckpoint{}
+		offsetCheckpointWrites.byKey[key] = checkpoint
+	}
+	offsetCheckpointWrites.Unlock()
+
+	seq := offsetWriteSeq.Add(1)
+	checkpoint.Lock()
+	defer checkpoint.Unlock()
+	if seq < checkpoint.lastSeq {
+		return nil
+	}
+	checkpoint.lastSeq = seq
+
+	return pb.WithFilerClient(false, signature, filer, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 
 		valueBuf := make([]byte, 8)
 		util.Uint64toBytes(valueBuf, uint64(offsetTsNs))
 
-		resp, err := client.KvPut(context.Background(), &filer_pb.KvPutRequest{
+		// a stalled KvPut must not hold a caller's drain or persistence loop
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		resp, err := client.KvPut(ctx, &filer_pb.KvPutRequest{
 			Key:   syncKey,
 			Value: valueBuf,
 		})
@@ -600,7 +662,6 @@ func setOffset(grpcDialOption grpc.DialOption, filer pb.ServerAddress, signature
 		return nil
 
 	})
-
 }
 
 func genProcessFunction(sourcePath string, targetPath string, excludePaths []string, reExcludeFileName *regexp.Regexp, excludeFileNames []*wildcard.WildcardMatcher, excludePathPatterns []*wildcard.WildcardMatcher, dataSink sink.ReplicationSink, doDeleteFiles bool, debug bool) func(resp *filer_pb.SubscribeMetadataResponse) error {

weed/command/filer_sync_jobs.go

@@ -137,6 +137,19 @@ type MetadataProcessor struct {
 	failedSticky     bool
 	oldestFailedTsNs int64
 
+	// earliestFailedTsNs is the smallest timestamp this processor ever failed
+	// on, latched for its lifetime: a redelivery can clear failedTs, but by
+	// then the processor is stopped and has already dropped unrecorded
+	// events — possibly sharing the failure's timestamp — so a resume
+	// position must still sit behind it.
+	earliestFailedTsNs int64
+
+	// newestSeenTsNs is the largest TsNs the stream has offered, job or
+	// marker. A watermark equal to it is not a safe resume point: more
+	// events at the same timestamp can still sit unread in the log, so the
+	// persisted position has to stay one tick below it.
+	newestSeenTsNs int64
+
 	// resubscribeCh closes once a failure has stopped the processor and all
 	// in-flight jobs have drained, asking the metadata follower to drop the
 	// stream so the caller's reconnect replays the pinned events in order —
@@ -199,6 +212,21 @@ func (t *MetadataProcessor) OldestFailedTsNs() int64 {
 	return t.oldestFailedTsNs
 }
 
+// EarliestFailedTsNs returns the smallest timestamp this processor has ever
+// failed on, including failures later cleared by a redelivery.
+func (t *MetadataProcessor) EarliestFailedTsNs() int64 {
+	t.activeJobsLock.Lock()
+	defer t.activeJobsLock.Unlock()
+	return t.earliestFailedTsNs
+}
+
+// NewestSeenTsNs returns the largest TsNs the stream has offered so far.
+func (t *MetadataProcessor) NewestSeenTsNs() int64 {
+	t.activeJobsLock.Lock()
+	defer t.activeJobsLock.Unlock()
+	return t.newestSeenTsNs
+}
+
 // ResubscribeCh closes once a job failure has stopped the processor and its
 // in-flight jobs have drained, signaling the metadata follower to drop the
 // stream so a reconnect replays what the watermark still covers.
@@ -206,6 +234,18 @@ func (t *MetadataProcessor) ResubscribeCh() <-chan struct{} {
 	return t.resubscribeCh
 }
 
+// Drain blocks until every admitted job has finished. Callers whose stream
+// just ended use it before reconnecting: a replayed event must not race a
+// still-running job for the same path, or the older write could finish last
+// and clobber the newer one.
+func (t *MetadataProcessor) Drain() {
+	t.activeJobsLock.Lock()
+	defer t.activeJobsLock.Unlock()
+	for t.activeJobCount > 0 {
+		t.activeJobsCond.Wait()
+	}
+}
+
 // pathAncestors returns all proper ancestor directories of p.
 // For "/a/b/c", returns ["/a/b", "/a", "/"].
 func pathAncestors(p util.FullPath) []util.FullPath {
@@ -339,6 +379,9 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse)
 		// to it so idle stretches still advance the resume point.
 		t.activeJobsLock.Lock()
 		defer t.activeJobsLock.Unlock()
+		if resp.TsNs > t.newestSeenTsNs {
+			t.newestSeenTsNs = resp.TsNs
+		}
 		if resp.TsNs > t.filteredTsNs {
 			t.filteredTsNs = resp.TsNs
 		}
@@ -354,6 +397,10 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse)
 	t.activeJobsLock.Lock()
 	defer t.activeJobsLock.Unlock()
 
+	if resp.TsNs > t.newestSeenTsNs {
+		t.newestSeenTsNs = resp.TsNs
+	}
+
 	p, newPath, kind := extractJobInfo(resp)
 	eventKey := failedEventKey{tsNs: resp.TsNs, path: p, newPath: newPath, kind: kind}
 
@@ -425,6 +472,9 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse)
 			// whole mechanism exists to remove.
 			t.stopped = true
 			t.activeJobsCond.Broadcast()
+			if t.earliestFailedTsNs == 0 || resp.TsNs < t.earliestFailedTsNs {
+				t.earliestFailedTsNs = resp.TsNs
+			}
 			if t.failedSticky {
 				if resp.TsNs < t.oldestFailedTsNs {
 					t.oldestFailedTsNs = resp.TsNs

weed/command/filer_sync_jobs_test.go

@@ -4,6 +4,7 @@ import (
 	"container/heap"
 	"errors"
 	"fmt"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -553,6 +554,61 @@ func TestSamePathBarrierSerialization(t *testing.T) {
 	})
 }
 
+// TestConcurrentJobsUpToLimit proves independent-path jobs run in parallel up
+// to the concurrency limit: every admitted job is in flight before any
+// finishes (a serial loop could only have one running), and admissions past
+// the limit wait for a slot rather than running ahead of it.
+func TestConcurrentJobsUpToLimit(t *testing.T) {
+	const limit = 3
+	started := make(chan struct{}, limit)
+	release := make(chan struct{})
+	var inFlight, maxInFlight atomic.Int32
+
+	p := NewMetadataProcessor(func(resp *filer_pb.SubscribeMetadataResponse) error {
+		cur := inFlight.Add(1)
+		for {
+			if old := maxInFlight.Load(); cur <= old || maxInFlight.CompareAndSwap(old, cur) {
+				break
+			}
+		}
+		started <- struct{}{}
+		<-release
+		inFlight.Add(-1)
+		return nil
+	}, limit, 0)
+
+	admitDone := make(chan struct{})
+	go func() {
+		defer close(admitDone)
+		for i := 0; i < limit+1; i++ {
+			p.AddSyncJob(makeResp("/d", fmt.Sprintf("f%d", i), false, int64(i+1), true))
+		}
+	}()
+
+	for i := 0; i < limit; i++ {
+		select {
+		case <-started:
+		case <-time.After(10 * time.Second):
+			t.Fatalf("only %d of %d jobs entered flight", i, limit)
+		}
+	}
+	// The extra admission must still be blocked on a full pool.
+	select {
+	case <-started:
+		t.Fatal("a job started with no free slot")
+	case <-admitDone:
+		t.Fatal("submission returned past the concurrency limit")
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	close(release)
+	<-admitDone
+	p.Drain()
+	if got := maxInFlight.Load(); got != limit {
+		t.Fatalf("peak in-flight = %d, want %d", got, limit)
+	}
+}
+
 // benchResult prevents the compiler from optimizing away the conflict check.
 var benchResult bool
 
@@ -1073,3 +1129,107 @@ func TestResubscribeWaitsForSameTsSibling(t *testing.T) {
 		t.Fatal("resubscribe channel never closed after the same-ts jobs drained")
 	}
 }
+
+// TestResumeStaysBehindClearedPin verifies the resume floor outlives the
+// ledger pin: a redelivery can clear failedTs, but events dropped while the
+// processor was stopped — including ones sharing the failure's timestamp —
+// still have to replay, so the resume position stays one tick below the
+// earliest failure.
+func TestResumeStaysBehindClearedPin(t *testing.T) {
+	fail := true
+	release := make(chan struct{})
+	p := NewMetadataProcessor(func(resp *filer_pb.SubscribeMetadataResponse) error {
+		name := resp.EventNotification.NewEntry.GetName()
+		if name == "slow.txt" {
+			<-release
+		}
+		if name == "bad.txt" && fail {
+			return errors.New("AccessDenied: Access Denied")
+		}
+		return nil
+	}, 10, 0)
+
+	// the slow job holds the drain — and the resubscribe — open so the
+	// redelivery still lands on this processor
+	p.AddSyncJob(makeResp("/dir", "slow.txt", false, 300, true))
+	p.AddSyncJob(makeResp("/dir", "bad.txt", false, 200, true))
+	deadline := time.Now().Add(10 * time.Second)
+	for p.OldestFailedTsNs() != 200 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+
+	// a same-ts sibling arriving after the stop is dropped
+	p.AddSyncJob(makeResp("/dir", "sib.txt", false, 200, true))
+	p.activeJobsLock.Lock()
+	dropped := p.activeJobTs[200] == 0
+	p.activeJobsLock.Unlock()
+	if !dropped {
+		t.Fatal("sibling admitted after the failure stopped the processor")
+	}
+
+	// the redelivery runs and clears the pin
+	fail = false
+	p.AddSyncJob(makeResp("/dir", "bad.txt", false, 200, true))
+	deadline = time.Now().Add(10 * time.Second)
+	for p.OldestFailedTsNs() != 0 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if got := p.OldestFailedTsNs(); got != 0 {
+		t.Fatalf("oldest failed = %d after the redelivery, want 0", got)
+	}
+
+	if got := p.EarliestFailedTsNs(); got != 200 {
+		t.Fatalf("earliest failed = %d, want the floor held at 200", got)
+	}
+	if got := resumeTsNsFor(p); got != 199 {
+		t.Fatalf("resume = %d, want 199 so the dropped sibling replays", got)
+	}
+
+	close(release)
+	waitForJobsToDrain(t, p)
+}
+
+// TestResumeStaysBelowNewestSeen covers the frontier case: when every admitted
+// job has finished, the watermark sits on the newest timestamp seen — but the
+// log can still hold same-timestamp events the stream has not delivered yet.
+// An exclusive resume there would skip them, so the position stays one tick
+// below until a newer timestamp proves the block is closed.
+func TestResumeStaysBelowNewestSeen(t *testing.T) {
+	p := NewMetadataProcessor(func(resp *filer_pb.SubscribeMetadataResponse) error {
+		return nil
+	}, 10, 0)
+
+	// the event at 100 finishes before a same-timestamp sibling could even be
+	// read; the watermark reaches the frontier but cannot persist there
+	p.AddSyncJob(makeResp("/dir", "a.txt", false, 100, true))
+	waitForJobsToDrain(t, p)
+	if got := resumeTsNsFor(p); got != 99 {
+		t.Fatalf("resume = %d at the frontier, want 99 so an unread same-ts event replays", got)
+	}
+
+	// once a newer timestamp is seen the 100-block is closed: a slow job at
+	// 200 in flight pins nothing below it, so 100 is now a safe position
+	release := make(chan struct{})
+	p2 := NewMetadataProcessor(func(resp *filer_pb.SubscribeMetadataResponse) error {
+		if resp.TsNs == 200 {
+			<-release
+		}
+		return nil
+	}, 10, 0)
+	p2.AddSyncJob(makeResp("/dir", "c.txt", false, 100, true))
+	p2.AddSyncJob(makeResp("/dir", "b.txt", false, 200, true))
+	deadline := time.Now().Add(10 * time.Second)
+	for p2.processedTsWatermark.Load() != 100 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if got := resumeTsNsFor(p2); got != 100 {
+		t.Fatalf("resume = %d below the frontier, want 100", got)
+	}
+
+	// finishing the 200 job lands the watermark back on the frontier
+	close(release)
+	waitForJobsToDrain(t, p2)
+	if got := resumeTsNsFor(p2); got != 199 {
+		t.Fatalf("resume = %d at the frontier, want 199", got)
+	}
+}

weed/command/filer_sync_offset_test.go

@@ -1,6 +1,10 @@
 package command
 
-import "testing"
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
 
 // The offset key is a persistence contract: a changed form orphans saved
 // checkpoints, so the historical forms must survive verbatim and the
@@ -74,3 +78,50 @@ func TestSyncOffset_HistoricalFallbackThenIndependence(t *testing.T) {
 		t.Fatalf("B read = (%d, %v), want (111, nil)", got, err)
 	}
 }
+
+// TestSyncOffset_StaleWriteDropped: a checkpoint write issued before a write
+// that already landed under the same key is stale and must not apply, while
+// a deliberately regressed value issued later — a resume floor or an
+// operator seed — still does.
+func TestSyncOffset_StaleWriteDropped(t *testing.T) {
+	filerAddr, dial := startKvFiler(t)
+	prefix := getSignaturePrefixByPath("/data", "/backup-stale")
+	sourceSig := int32(77)
+
+	if err := setOffset(dial, filerAddr, prefix, sourceSig, 300); err != nil {
+		t.Fatalf("seed offset: %v", err)
+	}
+
+	// Pin the checkpoint past the next write's issuance seq, simulating a
+	// write that was issued before a newer one already landed.
+	syncKey := []byte(prefix + "____")
+	util.Uint32toBytes(syncKey[len(prefix):len(prefix)+4], uint32(sourceSig))
+	mapKey := string(filerAddr) + "\x00" + string(syncKey)
+	offsetCheckpointWrites.Lock()
+	checkpoint := offsetCheckpointWrites.byKey[mapKey]
+	offsetCheckpointWrites.Unlock()
+	if checkpoint == nil {
+		t.Fatal("no checkpoint state for the seeded key")
+	}
+	checkpoint.Lock()
+	checkpoint.lastSeq = offsetWriteSeq.Load() + 1000
+	checkpoint.Unlock()
+
+	if err := setOffset(dial, filerAddr, prefix, sourceSig, 100); err != nil {
+		t.Fatalf("stale write: %v", err)
+	}
+	if got, err := getOffset(dial, filerAddr, prefix, sourceSig); err != nil || got != 300 {
+		t.Fatalf("after stale write = (%d, %v), want (300, nil)", got, err)
+	}
+
+	// a later-issued regression is deliberate and must apply
+	checkpoint.Lock()
+	checkpoint.lastSeq = 0
+	checkpoint.Unlock()
+	if err := setOffset(dial, filerAddr, prefix, sourceSig, 199); err != nil {
+		t.Fatalf("floored write: %v", err)
+	}
+	if got, err := getOffset(dial, filerAddr, prefix, sourceSig); err != nil || got != 199 {
+		t.Fatalf("after floored write = (%d, %v), want (199, nil)", got, err)
+	}
+}