diff --git a/weed/command/filer_backup.go b/weed/command/filer_backup.go index 96b363cf55..060c5fc837 100644 --- a/weed/command/filer_backup.go +++ b/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 } diff --git a/weed/command/filer_remote_gateway_buckets.go b/weed/command/filer_remote_gateway_buckets.go index 569ef62430..7ef16cd284 100644 --- a/weed/command/filer_remote_gateway_buckets.go +++ b/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(), } diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index d36147191f..90d7514222 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/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(), } diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go index 529472139e..554f5e3f99 100644 --- a/weed/command/filer_sync.go +++ b/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 { diff --git a/weed/command/filer_sync_jobs.go b/weed/command/filer_sync_jobs.go index 69554fd9a4..a791158b40 100644 --- a/weed/command/filer_sync_jobs.go +++ b/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 diff --git a/weed/command/filer_sync_jobs_test.go b/weed/command/filer_sync_jobs_test.go index 968858638a..20f5ab04da 100644 --- a/weed/command/filer_sync_jobs_test.go +++ b/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) + } +} diff --git a/weed/command/filer_sync_offset_test.go b/weed/command/filer_sync_offset_test.go index 835bb70fa0..93caf23e5c 100644 --- a/weed/command/filer_sync_offset_test.go +++ b/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) + } +}