Dedup journal.Summary() against retried-shard duplicate records

coordinator/internal/journal/reader.go

@@ -97,10 +97,34 @@ func forEachRecord(raw []byte, fn func(*drsyncpb.JournalRecord) error) error {
 // ?summary=true query serve, factored out here so callers that need the
 // job-wide histogram (the completion email, the WebUI's journal-summary
 // panel) don't re-scan the journal with their own copy of the loop.
+//
+// Deduped per (pass, type, rel_path): journaling is at-least-once (a shard
+// whose lease expires mid-run is requeued and re-runs from scratch, see
+// ReadRecords' own doc comment and jrn.c — "readers dedup"), so the exact
+// same record can appear twice for one shard's worth of work with no
+// correctness issue on its own, but it inflates every count here 1:1 with
+// however many shards happened to need a retry — confirmed live: a single
+// lease expiry during a 22k-file pass alone inflated this by 2000. rel_path
+// is not unique enough on its own to dedup by (a legitimately different
+// second error on the same path, e.g. two different xattr names each
+// failing to apply, would collide) but collapsing that rare case is an
+// acceptable trade against the much more common shard-retry duplication
+// this exists to fix. Affordable here because Summary holds one small
+// per-type histogram (not a per-file batch to seed): passctrl.go's
+// seedVerify/seedDirfix hit the exact same duplication but deliberately do
+// NOT dedup it, since they stream in O(batch) memory and a whole-pass "seen"
+// set would reintroduce the O(N) memory that design exists to avoid — see
+// their own doc comments for that tradeoff.
 func Summary(root string, jobID int64, passNos []int) (byType map[string]int64, total int64, err error) {
 	byType = map[string]int64{}
 	for _, pn := range passNos {
+		seen := make(map[string]bool)
 		err := ReadRecords(root, jobID, pn, func(r *drsyncpb.JournalRecord) error {
+			key := r.Type.String() + "\x00" + string(r.RelPath)
+			if seen[key] {
+				return nil
+			}
+			seen[key] = true
 			byType[strings.TrimPrefix(r.Type.String(), "JR_")]++
 			total++
 			return nil

coordinator/internal/journal/reader_test.go

@@ -97,3 +97,42 @@ func TestSummaryMissingPassIsEmpty(t *testing.T) {
 		t.Errorf("byType = %+v, total = %d, want empty", byType, total)
 	}
 }
+
+// TestSummaryDedupsRetriedShard is the at-least-once-journaling regression:
+// a shard whose lease expires mid-run is requeued and re-runs from scratch
+// (ReadRecords' own doc comment: "a re-run shard re-emits its records"), so
+// the SAME (type, rel_path) can appear twice in one pass's journal —
+// confirmed live, a single lease expiry during a 22k-file pass alone
+// inflated counts by 2000. Without dedup, Summary (and therefore `drsync
+// journal cat --summary`, the completion email, and the WebUI's
+// journal-summary panel) would overcount by exactly the number of retried
+// shards' worth of records, with no way for an operator to tell that from a
+// real change in volume.
+func TestSummaryDedupsRetriedShard(t *testing.T) {
+	root := t.TempDir()
+	writeRecords(t, root, 1, 1, []*drsyncpb.JournalRecord{
+		{Type: drsyncpb.JournalRecord_JR_META_FIXED, RelPath: []byte("retried")},
+		{Type: drsyncpb.JournalRecord_JR_META_FIXED, RelPath: []byte("once")},
+		{Type: drsyncpb.JournalRecord_JR_META_FIXED, RelPath: []byte("retried")}, // duplicate: shard re-run
+		// A different TYPE on the same path is not a duplicate of the above —
+		// dedup keys on (type, rel_path), not rel_path alone, so a file that
+		// legitimately appears under two different record types in one pass
+		// (not possible from the walker today, but the key shape should not
+		// assume that) is still counted for each.
+		{Type: drsyncpb.JournalRecord_JR_ERROR, RelPath: []byte("retried")},
+	})
+
+	byType, total, err := Summary(root, 1, []int{1})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if total != 3 {
+		t.Errorf("total = %d, want 3 (2 distinct META_FIXED paths + 1 ERROR, the duplicate META_FIXED dropped)", total)
+	}
+	if byType["META_FIXED"] != 2 {
+		t.Errorf("byType[META_FIXED] = %d, want 2 — a retried shard's duplicate journal record was not deduped", byType["META_FIXED"])
+	}
+	if byType["ERROR"] != 1 {
+		t.Errorf("byType[ERROR] = %d, want 1", byType["ERROR"])
+	}
+}

coordinator/internal/passctrl/passctrl.go

@@ -953,6 +953,15 @@ func (c *Controller) seedDirfix(job *store.Job, pass *store.Pass) (int, error) {
 		return nil
 	}
 
+	// NOT deduped against a retried shard's duplicate JR_DIR_META (journaling
+	// is at-least-once, same mechanism as seedVerify's own doc comment
+	// below) — deliberately: this is streamed in O(batch) memory, and a
+	// whole-pass "seen" set would reintroduce the O(N) memory the streaming
+	// design exists to avoid. A duplicate DirMeta only means the same
+	// directory's metadata gets applied twice — wasteful but not incorrect,
+	// since dirfix.c's apply is idempotent. journal.Summary (a bounded
+	// per-pass histogram, not a per-file streaming consumer) does dedup this
+	// class of duplicate; this function cannot afford to the same way.
 	err := journal.ReadRecords(c.journalRoot, job.ID, pass.PassNo,
 		func(r *drsyncpb.JournalRecord) error {
 			if r.Type != drsyncpb.JournalRecord_JR_DIR_META || r.Src == nil {
@@ -1032,6 +1041,20 @@ func (c *Controller) seedVerify(job *store.Job, pass *store.Pass) (int, error) {
 	// file is hashed — and needs no whole-pass state.
 	firstCopied := true
 
+	// NOT deduped against a retried shard's duplicate records (journaling is
+	// at-least-once — a shard whose lease expires mid-run is requeued and
+	// re-runs from scratch, journal.go's own doc comment: "a re-run shard
+	// re-emits its records" — confirmed live: a single lease expiry during a
+	// 22k-file pass duplicated 2000 files' worth of records). Deliberately:
+	// this function is streamed in O(batch) memory (TestSeedVerifyMemoryBounded
+	// pins this at 1M files), and a whole-pass "seen" set to dedup would
+	// reintroduce the O(N) memory the streaming design exists to avoid. A
+	// duplicate VerifyEntry only means the same file gets verified twice —
+	// wasteful (double the read/hash cost for a checksum-sampled file) but not
+	// incorrect, since check_entry (agent/src/verify.c) is idempotent either
+	// way. journal.Summary (the reporting/audit view, not a per-file streaming
+	// consumer) does dedup, since a bounded per-pass type histogram can afford
+	// a "seen" set that seedVerify's per-file entries cannot.
 	err = journal.ReadRecords(c.journalRoot, job.ID, pass.PassNo,
 		func(r *drsyncpb.JournalRecord) error {
 			var checksum bool

coordinator/internal/passctrl/seeddirfix_test.go

@@ -162,3 +162,54 @@ func TestSeedDirfixEmpty(t *testing.T) {
 		t.Fatalf("seedDirfix seeded %d dirs from a journal with no DIR_META", n)
 	}
 }
+
+// TestSeedDirfixToleratesRetriedShard pins a deliberate design choice, not a
+// bug (same mechanism/rationale as TestSeedVerifyToleratesRetriedShard): a
+// shard whose lease expires mid-run is requeued and re-runs from scratch, so
+// the SAME directory's JR_DIR_META can appear twice in one pass's journal.
+// seedDirfix does NOT dedup this — it streams in O(batch) memory, and a
+// whole-pass "seen" set would reintroduce the O(N) memory that design
+// exists to avoid, in exchange for eliminating something that is wasteful
+// (the directory's metadata gets fetched and applied twice) but not
+// incorrect, since dirfix.c's apply is idempotent.
+func TestSeedDirfixToleratesRetriedShard(t *testing.T) {
+	c := newController(t)
+	job := makeJob(t, c, []byte(baseSpec))
+	pass, err := c.st.CreatePass(job.ID, 1, model.PassScanning)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// "a" appears twice (the shard that walked it ran twice); "b" appears once.
+	writeDirMetaJournal(t, c.journalRoot, job.ID, pass.PassNo, []dirRec{
+		{rel: "a", uid: 1, gid: 1, mode: 0o750, mt: 200},
+		{rel: "b", uid: 2, gid: 2, mode: 0o755, mt: 300},
+		{rel: "a", uid: 1, gid: 1, mode: 0o750, mt: 200}, // duplicate: shard re-run
+	})
+
+	n, err := c.seedDirfix(job, pass)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != 3 {
+		t.Fatalf("seedDirfix returned %d dirs, want 3 (duplicates are intentionally NOT collapsed — see comment)", n)
+	}
+
+	leased, err := c.st.LeaseShards("dirfix-reader", 1_000_000, time.Hour)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got []*drsyncpb.DirMeta
+	for _, sh := range leased {
+		if sh.Kind != model.KindDirfix {
+			continue
+		}
+		fb := &drsyncpb.DirFixBatch{}
+		if err := proto.Unmarshal(sh.Payload, fb); err != nil {
+			t.Fatal(err)
+		}
+		got = append(got, fb.Dirs...)
+	}
+	if len(got) != 3 {
+		t.Fatalf("dirfix shards cover %d dirs, want 3", len(got))
+	}
+}

coordinator/internal/passctrl/seedverify_test.go

@@ -69,6 +69,59 @@ func writeJournal(t *testing.T, root string, jobID int64, passNo int, rels []str
 	}
 }
 
+// writeJournalRecords is writeJournal's more general form: callers pass the
+// exact records to emit, in order, so a test can model a shard retry (the
+// same rel_path's record appearing twice in the journal — journaling is
+// at-least-once, see journal.ReadRecords' own doc comment) rather than
+// writeJournal's always-one-JR_COPIED-per-rel shape.
+func writeJournalRecords(t *testing.T, root string, jobID int64, passNo int, recs []*drsyncpb.JournalRecord) {
+	t.Helper()
+	w, err := journal.NewWriter(root)
+	if err != nil {
+		t.Fatal(err)
+	}
+	enc, err := zstd.NewWriter(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	const perBatch = 10_000
+	var raw []byte
+	var count int
+	appendBatch := func() {
+		if count == 0 {
+			return
+		}
+		if err := w.Append(&drsyncpb.JournalBatch{
+			JobId: uint64(jobID), PassNo: uint32(passNo),
+			RecordCount: uint32(count), RecordsZstd: enc.EncodeAll(raw, nil),
+		}); err != nil {
+			t.Fatal(err)
+		}
+		raw = raw[:0]
+		count = 0
+	}
+	for _, rec := range recs {
+		b, err := proto.Marshal(rec)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var hdr [binary.MaxVarintLen64]byte
+		n := binary.PutUvarint(hdr[:], uint64(len(b)))
+		raw = append(raw, hdr[:n]...)
+		raw = append(raw, b...)
+		if count++; count >= perBatch {
+			appendBatch()
+		}
+	}
+	appendBatch()
+	if err := w.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	if err := w.Close(); err != nil {
+		t.Fatal(err)
+	}
+}
+
 // countVerify reads back the verify shards seeded into a pass (by leasing the
 // queued shards) and returns the total verify entries and how many are
 // checksummed, plus the number of verify shards.
@@ -242,3 +295,48 @@ func TestSeedVerifyMemoryBounded(t *testing.T) {
 		t.Fatalf("peak heap grew %d MiB (> %d MiB budget) — not streaming?", grew>>20, budget>>20)
 	}
 }
+
+// TestSeedVerifyToleratesRetriedShard pins a deliberate design choice, not a
+// bug: a shard whose lease expires mid-run is requeued and re-runs from
+// scratch (journal.ReadRecords' own doc comment: "a re-run shard re-emits
+// its records"), so the SAME rel_path's JR_META_FIXED/JR_COPIED can appear
+// twice in one pass's journal — confirmed live, a single lease expiry during
+// a 22k-file pass alone produced exactly this for 2000 files. seedVerify
+// does NOT dedup this (unlike journal.Summary, a bounded per-pass histogram
+// that can afford to): it streams in O(batch) memory
+// (TestSeedVerifyMemoryBounded pins this at 1M files), and a whole-pass
+// "seen" set to dedup would reintroduce the O(N) memory the streaming design
+// exists to avoid, in exchange for eliminating something that is wasteful
+// (a duplicated file gets verified, and possibly checksummed, twice) but not
+// incorrect (check_entry, agent/src/verify.c, is idempotent either way).
+// This test exists so a future change doesn't silently reintroduce
+// unbounded memory while "fixing" this — see the comment at this call site
+// in passctrl.go for the full tradeoff.
+func TestSeedVerifyToleratesRetriedShard(t *testing.T) {
+	c := newController(t)
+	spec := []byte(baseSpec + "  verify:\n    checksum:\n      sample_rate: 1.0\n")
+	job := makeJob(t, c, spec)
+	pass, err := c.st.CreatePass(job.ID, 1, model.PassScanning)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// "retried" appears twice as JR_META_FIXED (the shard that found and
+	// fixed it ran twice); "once" appears once, normally.
+	writeJournalRecords(t, c.journalRoot, job.ID, pass.PassNo, []*drsyncpb.JournalRecord{
+		{Type: drsyncpb.JournalRecord_JR_META_FIXED, RelPath: []byte("retried")},
+		{Type: drsyncpb.JournalRecord_JR_META_FIXED, RelPath: []byte("once")},
+		{Type: drsyncpb.JournalRecord_JR_META_FIXED, RelPath: []byte("retried")}, // duplicate: shard re-run
+	})
+
+	got, err := c.seedVerify(job, pass)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got != 3 {
+		t.Fatalf("seedVerify returned %d entries, want 3 (duplicates are intentionally NOT collapsed — see comment)", got)
+	}
+	entries, _, _ := countVerify(t, c)
+	if entries != 3 {
+		t.Fatalf("verify shards cover %d entries, want 3", entries)
+	}
+}

docs/DESIGN-coordinator.md

@@ -872,6 +872,40 @@ Append-only, per (job, pass), the system of record for per-file outcomes:
   the page cache, so a coordinator crash would lose them. If an fsync fails,
   every ack for that cycle is withheld (counted by
   `drsync_journal_fsync_errors_total`) and retried on the next successful flush.
+- **At-least-once, not exactly-once:** a shard whose lease expires mid-run
+  (`ExpireLeases` in store.go) is requeued and re-runs from scratch — it
+  re-walks and re-emits every journal record it had already written the
+  first time, not just the remainder. `agent/src/jrn.c`'s own doc comment
+  names this as the design: "lease-expiry re-runs gives at-least-once
+  journaling (readers dedup)." That last parenthetical is a contract on
+  every reader, and as of 2026-08 only some of them honor it:
+  `journal.Orphans()` always deduped correctly (it's a `map[string]struct{}`
+  by construction); `journal.Summary()` did not, until **found live**
+  investigating a user report of stale destination ACLs surviving a
+  re-run — a 22k-file/POSIX-ACL reproduction that turned out to hit a lease
+  expiry mid-pass, which inflated `drsync journal cat --summary`, the
+  completion email, and the WebUI journal-summary panel by exactly the
+  retried shard's record count (2000, in that repro). Fixed by deduping
+  `Summary()` per `(pass, type, rel_path)`, same pattern as `Orphans()`.
+  `passctrl.seedVerify`/`seedDirfix`, by contrast, deliberately do **not**
+  dedup: both stream the journal in O(batch) memory
+  (`TestSeedVerifyMemoryBounded` pins this at a 128 MiB budget for 1M
+  files), and a whole-pass "seen" set to dedup would reintroduce the O(N)
+  memory that streaming design exists to avoid. Left un-deduped there
+  because it's provably safe to: a duplicate `VerifyEntry`/`DirMeta` just
+  means that one file/dir gets verified or metadata-applied twice
+  (`check_entry` in agent/src/verify.c and dirfix's apply are both
+  idempotent), wasteful but not incorrect. Net effect: reporting/audit
+  views (`Summary`) are now exact; shard-seeding views (`seedVerify`/
+  `seedDirfix`) tolerate duplicates by design and always have. This
+  duplication bug is **not** the same thing as the original stale-ACL
+  report that surfaced it — double-counting a retried shard's records
+  cannot explain files being silently missed with zero journal signal
+  (a re-run shard re-diffs from scratch; it can re-fix or correctly see
+  nothing to fix, never skip). That symptom remains open, with a
+  source-filesystem metadata-caching staleness (observed before with VAST
+  appliances) as the leading theory, pending the user's own larger-scale
+  reproduction.
 
 ## 6. REST API & WebSocket (day-1 surface, also the WebUI contract)