filer.sync: move the watermark over jobs that finished out of order (#12554)

* filer.sync: move the watermark over jobs that finished ahead of an older one The watermark only took the finishing job's timestamp when it was the oldest in flight, so jobs that completed before an older one were never recorded and a quiet stream kept the offset behind finished events. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer.sync: keep one finished timestamp per gap behind a running job A long job kept every later completion in the done heap. Only the newest one between two running or failed timestamps can become the watermark. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * filer.sync: test done-heap compaction across several running and failed jobs Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-12 06:48:45 +01:00 · 2026-11-11 08:47:30 -08:00
1 parent 000baf8dfa
commit a4a68c885a
@@ -3,6 +3,8 @@ package command
 import (
 	"container/heap"
 	"path"
+	"slices"
+	"sort"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -125,6 +127,10 @@ type MetadataProcessor struct {
 	// tsHeap is a min-heap of active job timestamps with lazy deletion,
 	// used for O(log n) amortized watermark tracking.
 	tsHeap tsMinHeap
+	// doneTsHeap holds timestamps of jobs that succeeded while an older job
+	// was still running or failed; the watermark moves over them once
+	// nothing older is left.
+	doneTsHeap tsMinHeap

 	// failedTs records every event whose job returned an error and has not
 	// since completed, and oldestFailedTsNs caches its minimum (0 when empty).
@@ -539,13 +545,24 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse)
 			}
 			heap.Pop(&t.tsHeap)
 		}
-		// If this was the oldest job, advance the watermark, but never to or
-		// past an event that failed: the offset is the durable resume point,
-		// and moving it over a failure drops that event for good.
-		if t.tsHeap.Len() == 0 || resp.TsNs < t.tsHeap[0] {
-			if t.oldestFailedTsNs == 0 || resp.TsNs < t.oldestFailedTsNs {
-				t.processedTsWatermark.Store(resp.TsNs)
+		// Advance the watermark over every finished job older than the oldest
+		// one still running, but never to or past an event that failed: the
+		// offset is the durable resume point, and moving it over a failure
+		// drops that event for good.
+		if jobErr == nil {
+			heap.Push(&t.doneTsHeap, resp.TsNs)
+		}
+		for t.doneTsHeap.Len() > 0 {
+			doneTsNs := t.doneTsHeap[0]
+			if (t.tsHeap.Len() > 0 && doneTsNs >= t.tsHeap[0]) ||
+				(t.oldestFailedTsNs != 0 && doneTsNs >= t.oldestFailedTsNs) {
+				break
 			}
+			heap.Pop(&t.doneTsHeap)
+			t.processedTsWatermark.Store(doneTsNs)
+		}
+		if len(t.doneTsHeap) > 2*(len(t.activeJobTs)+len(t.failedTs))+64 {
+			t.compactDoneTs()
 		}
 		if t.activeJobCount == 0 && t.filteredTsNs > t.processedTsWatermark.Load() &&
 			(t.oldestFailedTsNs == 0 || t.filteredTsNs < t.oldestFailedTsNs) {
@@ -560,6 +577,36 @@ func (t *MetadataProcessor) AddSyncJob(resp *filer_pb.SubscribeMetadataResponse)
 	}()
 }

+// compactDoneTs keeps only the newest finished timestamp between two
+// consecutive running or failed ones, since the watermark only lands on the
+// newest one below the oldest of them. Without it a long job keeps every later
+// completion. Must be called under activeJobsLock.
+func (t *MetadataProcessor) compactDoneTs() {
+	bounds := make([]int64, 0, len(t.activeJobTs)+len(t.failedTs)+1)
+	for ts := range t.activeJobTs {
+		bounds = append(bounds, ts)
+	}
+	for k := range t.failedTs {
+		bounds = append(bounds, k.tsNs)
+	}
+	if t.oldestFailedTsNs != 0 {
+		bounds = append(bounds, t.oldestFailedTsNs)
+	}
+	slices.Sort(bounds)
+	newest := make(map[int]int64)
+	for _, ts := range t.doneTsHeap {
+		i := sort.Search(len(bounds), func(i int) bool { return bounds[i] > ts })
+		if prev, ok := newest[i]; !ok || ts > prev {
+			newest[i] = ts
+		}
+	}
+	t.doneTsHeap = t.doneTsHeap[:0]
+	for _, ts := range newest {
+		t.doneTsHeap = append(t.doneTsHeap, ts)
+	}
+	heap.Init(&t.doneTsHeap)
+}
+
 // eventDataSize is the chunk data this event will copy: chunks on the new
 // entry that the old entry does not already have. Deletes, renames, and
 // attribute-only updates all come out zero, so byte rates reflect data
@@ -4,6 +4,7 @@ import (
 	"container/heap"
 	"errors"
 	"fmt"
+	"slices"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -1233,3 +1234,134 @@ func TestResumeStaysBelowNewestSeen(t *testing.T) {
 		t.Fatalf("resume = %d at the frontier, want 199", got)
 	}
 }
+
+func TestWatermarkAfterOutOfOrderCompletion(t *testing.T) {
+	gates := map[int64]chan struct{}{100: make(chan struct{}), 101: make(chan struct{}), 102: make(chan struct{})}
+	p := NewMetadataProcessor(func(resp *filer_pb.SubscribeMetadataResponse) error {
+		<-gates[resp.TsNs]
+		return nil
+	}, 10, 0)
+	for _, ts := range []int64{100, 101, 102} {
+		p.AddSyncJob(makeResp("/dir", fmt.Sprintf("f%d.txt", ts), false, ts, true))
+	}
+
+	close(gates[102])
+	waitForActiveJobs(t, p, 2)
+	close(gates[101])
+	waitForActiveJobs(t, p, 1)
+	if got := p.processedTsWatermark.Load(); got != 0 {
+		t.Fatalf("watermark = %d while the job at 100 runs, want 0", got)
+	}
+	close(gates[100])
+	waitForJobsToDrain(t, p)
+
+	if got := p.processedTsWatermark.Load(); got != 102 {
+		t.Fatalf("watermark = %d after every job finished, want 102", got)
+	}
+	if got := resumeTsNsFor(p); got != 101 {
+		t.Fatalf("resume = %d, want 101", got)
+	}
+}
+
+func TestDoneTimestampsStayBoundedBehindSlowJob(t *testing.T) {
+	release := make(chan struct{})
+	p := NewMetadataProcessor(func(resp *filer_pb.SubscribeMetadataResponse) error {
+		if resp.TsNs == 1 {
+			<-release
+		}
+		return nil
+	}, 4, 0)
+	p.AddSyncJob(makeResp("/dir", "slow.txt", false, 1, true))
+	for ts := int64(2); ts <= 2001; ts++ {
+		p.AddSyncJob(makeResp("/dir", fmt.Sprintf("f%d.txt", ts), false, ts, true))
+	}
+	if done := waitForActiveJobs(t, p, 1); done > 2*4+64+1 {
+		t.Fatalf("%d finished timestamps kept behind the slow job", done)
+	}
+	if got := p.processedTsWatermark.Load(); got != 0 {
+		t.Fatalf("watermark = %d while the oldest job runs, want 0", got)
+	}
+	close(release)
+	waitForJobsToDrain(t, p)
+	if got := p.processedTsWatermark.Load(); got != 2001 {
+		t.Fatalf("watermark = %d after every job finished, want 2001", got)
+	}
+}
+
+// waitForActiveJobs waits until exactly n jobs are in flight and returns how
+// many finished timestamps the processor is holding.
+func waitForActiveJobs(t *testing.T, p *MetadataProcessor, n int) int {
+	t.Helper()
+	deadline := time.Now().Add(10 * time.Second)
+	for time.Now().Before(deadline) {
+		p.activeJobsLock.Lock()
+		active, done := p.activeJobCount, len(p.doneTsHeap)
+		p.activeJobsLock.Unlock()
+		if active == n {
+			return done
+		}
+		time.Sleep(time.Millisecond)
+	}
+	t.Fatalf("timed out waiting for %d active jobs", n)
+	return 0
+}
+
+func TestCompactedDoneTimestampsKeepProgressBetweenRunningJobs(t *testing.T) {
+	gates := map[string]chan struct{}{"a10": make(chan struct{}), "a20": make(chan struct{}), "a30": make(chan struct{})}
+	p := NewMetadataProcessor(func(resp *filer_pb.SubscribeMetadataResponse) error {
+		if gate, ok := gates[resp.EventNotification.NewEntry.Name]; ok {
+			<-gate
+		}
+		return nil
+	}, 200, 0)
+	add := func(name string, ts int64) {
+		p.AddSyncJob(makeResp("/dir", name, false, ts, true))
+	}
+	add("a10", 10)
+	for ts := int64(11); ts <= 19; ts++ {
+		add(fmt.Sprintf("f%d", ts), ts)
+	}
+	add("a20", 20)
+	// a finished job sharing the running job's timestamp
+	add("f20", 20)
+	for ts := int64(21); ts <= 29; ts++ {
+		add(fmt.Sprintf("f%d", ts), ts)
+	}
+	add("a30", 30)
+	for ts := int64(31); ts <= 130; ts++ {
+		add(fmt.Sprintf("f%d", ts), ts)
+	}
+	if done := waitForActiveJobs(t, p, 3); done > 2*3+64+1 {
+		t.Fatalf("%d finished timestamps kept, want them compacted", done)
+	}
+
+	for _, step := range []struct {
+		gate string
+		want int64
+	}{{"a10", 19}, {"a20", 29}, {"a30", 130}} {
+		close(gates[step.gate])
+		deadline := time.Now().Add(10 * time.Second)
+		for p.processedTsWatermark.Load() != step.want && time.Now().Before(deadline) {
+			time.Sleep(time.Millisecond)
+		}
+		if got := p.processedTsWatermark.Load(); got != step.want {
+			t.Fatalf("watermark = %d after %s finished, want %d", got, step.gate, step.want)
+		}
+	}
+}
+
+func TestCompactDoneTsKeepsNewestPerGap(t *testing.T) {
+	p := NewMetadataProcessor(func(*filer_pb.SubscribeMetadataResponse) error { return nil }, 10, 0)
+	p.activeJobTs[10] = 1
+	p.activeJobTs[30] = 1
+	p.failedTs[failedEventKey{tsNs: 20, path: "/dir/f20"}] = struct{}{}
+	p.oldestFailedTsNs = 20
+	for _, ts := range []int64{5, 11, 15, 20, 22, 29, 30, 31, 40} {
+		heap.Push(&p.doneTsHeap, ts)
+	}
+	p.compactDoneTs()
+	got := slices.Sorted(slices.Values(p.doneTsHeap))
+	if want := []int64{5, 15, 29, 40}; !slices.Equal(got, want) {
+		t.Fatalf("compacted done timestamps = %v, want %v", got, want)
+	}
+}