diff --git a/coordinator/internal/journal/reader.go b/coordinator/internal/journal/reader.go index ed2dd65..304664d 100644 --- a/coordinator/internal/journal/reader.go +++ b/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 diff --git a/coordinator/internal/journal/reader_test.go b/coordinator/internal/journal/reader_test.go index 7b1a674..17cb785 100644 --- a/coordinator/internal/journal/reader_test.go +++ b/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"]) + } +} diff --git a/coordinator/internal/passctrl/passctrl.go b/coordinator/internal/passctrl/passctrl.go index f78963f..4f2822a 100644 --- a/coordinator/internal/passctrl/passctrl.go +++ b/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 diff --git a/coordinator/internal/passctrl/seeddirfix_test.go b/coordinator/internal/passctrl/seeddirfix_test.go index 4d947d9..5f21fc4 100644 --- a/coordinator/internal/passctrl/seeddirfix_test.go +++ b/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)) + } +} diff --git a/coordinator/internal/passctrl/seedverify_test.go b/coordinator/internal/passctrl/seedverify_test.go index 8173da9..c7e5be9 100644 --- a/coordinator/internal/passctrl/seedverify_test.go +++ b/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) + } +} diff --git a/docs/DESIGN-coordinator.md b/docs/DESIGN-coordinator.md index 8b68feb..9c16e7d 100644 --- a/docs/DESIGN-coordinator.md +++ b/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)