diff --git a/weed/command/filer_sync_jobs.go b/weed/command/filer_sync_jobs.go
index 5b664df96c..6e4fab3285 100644
--- a/weed/command/filer_sync_jobs.go
+++ b/weed/command/filer_sync_jobs.go
@@ -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
diff --git a/weed/command/filer_sync_jobs_test.go b/weed/command/filer_sync_jobs_test.go
index 20f5ab04da..e2ceaeb1bd 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"
+	"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)
+	}
+}