From a5b89127320a9554dc3834cc83a374aa5e5e637d Mon Sep 17 00:00:00 2001 From: Steven Rhoods Date: Sat, 8 Aug 2026 16:13:47 +0100 Subject: [PATCH] Auto-retry parked shards once at phase end; add scan-rate timeline tab MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two independent UX improvements requested together: 1. advance()'s parked-shard guard used to hold every phase transition open indefinitely the instant any shard parked, requiring operator intervention even when the park cause was transient at the moment it happened (a mount blip, a brief NFS hiccup) rather than deterministic. On a long-running pass this could mean a job sitting for hours waiting on a condition that had long since cleared. advance() now calls store.RetryParkedByJob (resets attempt to 0, so the retry gets the same fresh 5-attempt budget as a new shard) once per pass before falling back to the original block-and-alert behavior. Bounded to exactly one automatic round via Controller.parkedAutoRetried (keyed by pass id, in-memory, same "safe direction to be wrong in" restart semantics as the existing parkedAlerted map) — an unbounded retry on a genuinely stuck shard would just be an infinite loop wearing a different name. checkParkedShards' operator alerting composes with this unchanged. 2. The "Aggregate throughput" panel now has a small tab toggle ("throughput" / "scan rate") that switches the same timeline canvas between bwHist and the already-tracked-but-never-graphed scanHist, via a new scanRate() formatter alongside the existing rate(). No new data plumbing needed — scanHist already fed the small KPI-strip sparkline; this just makes it available on the big auto-scaling graph too. Choice persists across reloads like unitsMode does. (A third requested item — bulk retry-all/drop-all buttons for parked shards — turned out to already be fully implemented, both the /api/v1/jobs/{name}/parked/retry|drop endpoints and the console's job-scoped bulk buttons; no change needed there.) New tests: TestAdvanceAutoRetriesParkedShardsOnce, TestAdvanceBlocksOnSecondParkAfterAutoRetry (the infinite-loop guard), TestAdvanceProceedsAfterAutoRetrySucceeds, and a console.test.mjs case driving the actual tab click end to end. All verified to fail against the pre-fix code before asserting the fix. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01PsdNZLfmAFrMX2VUtkLtmm --- .../internal/passctrl/parkedretry_test.go | 146 ++++++++++++++++++ coordinator/internal/passctrl/passctrl.go | 45 +++++- docs/DESIGN-coordinator.md | 31 +++- webui/console.html | 46 +++++- webui/test/console.test.mjs | 27 ++++ 5 files changed, 282 insertions(+), 13 deletions(-) create mode 100644 coordinator/internal/passctrl/parkedretry_test.go diff --git a/coordinator/internal/passctrl/parkedretry_test.go b/coordinator/internal/passctrl/parkedretry_test.go new file mode 100644 index 0000000..07401c0 --- /dev/null +++ b/coordinator/internal/passctrl/parkedretry_test.go @@ -0,0 +1,146 @@ +package passctrl + +import ( + "testing" + "time" + + "drsync/coordinator/internal/model" +) + +// TestAdvanceAutoRetriesParkedShardsOnce: a pass that would otherwise block +// indefinitely on parked work now gets one automatic retry round first — +// many park causes (a mount blip, a brief NFS hiccup) are transient at the +// moment a shard exhausts MaxShardAttempts, and by the time the rest of a +// large pass has drained, conditions may well have changed. +// RetryParkedByJob resets the attempt counter, so this call must requeue the +// shard (not silently no-op), and advance() must not treat that requeue as a +// completed phase this same tick — the shard needs a later tick to actually +// be granted and drain. +func TestAdvanceAutoRetriesParkedShardsOnce(t *testing.T) { + c := newController(t) + job := makeJob(t, c, []byte(baseSpec)) + pass, err := c.st.CreatePass(job.ID, 1, model.PassDirfix) + if err != nil { + t.Fatal(err) + } + parkOneShard(t, c, pass, "a/parked-dir") + + if err := c.advance(job); err != nil { + t.Fatal(err) + } + + counts, err := c.st.ShardStateCounts(pass.ID) + if err != nil { + t.Fatal(err) + } + if counts[model.ShardParked] != 0 { + t.Fatalf("parked count after first advance() = %d, want 0 (auto-retry should have requeued it)", + counts[model.ShardParked]) + } + if counts[model.ShardQueued] != 1 { + t.Fatalf("queued count after first advance() = %d, want 1 (the retried shard)", + counts[model.ShardQueued]) + } + // Pass must still be DIRFIX: the requeued shard has not been granted or + // drained yet, so the phase cannot have transitioned. + pass, err = c.st.PassByNo(job.ID, 1) + if err != nil { + t.Fatal(err) + } + if pass.State != model.PassDirfix { + t.Fatalf("pass state = %s, want still DIRFIX (retried shard not yet drained)", pass.State) + } +} + +// TestAdvanceBlocksOnSecondParkAfterAutoRetry: if the auto-retried shard +// parks again, the pass must fall back to the original block-and-alert +// behavior rather than retrying forever — an unbounded auto-retry loop on a +// genuinely stuck shard (e.g. a real permissions problem) would just be an +// infinite loop wearing a different name, and would silently mask a job that +// needs operator attention. +func TestAdvanceBlocksOnSecondParkAfterAutoRetry(t *testing.T) { + c := newController(t) + job := makeJob(t, c, []byte(baseSpec)) + pass, err := c.st.CreatePass(job.ID, 1, model.PassDirfix) + if err != nil { + t.Fatal(err) + } + parkOneShard(t, c, pass, "a/parked-dir") + + if err := c.advance(job); err != nil { + t.Fatal(err) + } + // Simulate the retried shard failing again: lease it and park it once more. + leased, err := c.st.LeaseShards("agent-1", 10, time.Minute) + if err != nil { + t.Fatal(err) + } + if len(leased) != 1 { + t.Fatalf("leased %d shards, want 1 (the retried shard)", len(leased)) + } + if err := c.st.ParkShard(leased[0].ID, leased[0].LeaseID, "EIO again"); err != nil { + t.Fatal(err) + } + + if err := c.advance(job); err != nil { + t.Fatal(err) + } + counts, err := c.st.ShardStateCounts(pass.ID) + if err != nil { + t.Fatal(err) + } + if counts[model.ShardParked] != 1 { + t.Fatalf("parked count after second advance() = %d, want 1 (must stay parked, "+ + "not retried a second time)", counts[model.ShardParked]) + } + if counts[model.ShardQueued] != 0 { + t.Fatalf("queued count after second advance() = %d, want 0 (no further auto-retry)", + counts[model.ShardQueued]) + } + pass, err = c.st.PassByNo(job.ID, 1) + if err != nil { + t.Fatal(err) + } + if pass.State != model.PassDirfix { + t.Fatalf("pass state = %s, want still DIRFIX (blocked on the re-parked shard)", pass.State) + } +} + +// TestAdvanceProceedsAfterAutoRetrySucceeds: the common case the feature +// exists for — the retried shard succeeds this time, and the phase advances +// normally once it drains, with no operator intervention needed at all. +func TestAdvanceProceedsAfterAutoRetrySucceeds(t *testing.T) { + c := newController(t) + job := makeJob(t, c, []byte(baseSpec)) + pass, err := c.st.CreatePass(job.ID, 1, model.PassDirfix) + if err != nil { + t.Fatal(err) + } + parkOneShard(t, c, pass, "a/parked-dir") + + if err := c.advance(job); err != nil { + t.Fatal(err) + } + leased, err := c.st.LeaseShards("agent-1", 10, time.Minute) + if err != nil { + t.Fatal(err) + } + if len(leased) != 1 { + t.Fatalf("leased %d shards, want 1 (the retried shard)", len(leased)) + } + if err := c.st.CompleteShard(leased[0].ID, leased[0].LeaseID, 0, nil, nil); err != nil { + t.Fatal(err) + } + + if err := c.advance(job); err != nil { + t.Fatal(err) + } + drainReaps(t, c) + pass, err = c.st.PassByNo(job.ID, 1) + if err != nil { + t.Fatal(err) + } + if pass.State != model.PassLinkfix { + t.Fatalf("pass state = %s, want LINKFIX (retried shard completed, phase should advance)", pass.State) + } +} diff --git a/coordinator/internal/passctrl/passctrl.go b/coordinator/internal/passctrl/passctrl.go index 11af790..f78963f 100644 --- a/coordinator/internal/passctrl/passctrl.go +++ b/coordinator/internal/passctrl/passctrl.go @@ -76,6 +76,20 @@ type Controller struct { // firing their own email, then flushed together once the window elapses. parkedRollup map[string]*parkedRollupState + // parkedAutoRetried tracks pass IDs whose parked backlog has already been + // given one automatic retry round (see advance()'s parked-shard gate) — + // so a persistently-failing shard that re-parks after the retry blocks + // the pass for operator attention as before, instead of looping forever. + // Guarded by parkedAlertMu (same lock as the other parked-tracking maps; + // a dedicated mutex would only ever be taken alongside it in practice). + // In-memory only, same "safe direction to be wrong in" reasoning as + // parkedAlerted: a coordinator restart re-attempts one retry round for a + // pass that was already retried before the restart — a redundant but + // harmless extra attempt, not a correctness issue. Forgotten once a pass + // leaves parked-blocked state (advances or the job ends), so a job that + // runs again later gets its own fresh retry round. + parkedAutoRetried map[int64]bool + // reapCh feeds runReapWorker — see its doc comment for why reaping runs on // its own goroutine instead of inline in advance(). reapCh chan reapRequest @@ -129,7 +143,8 @@ func (c *Controller) jobTerminal(jobID int64) { func New(st *store.Store, journalRoot string) *Controller { return &Controller{st: st, journalRoot: journalRoot, parkedAlerted: map[int64]bool{}, parkedRollup: map[string]*parkedRollupState{}, - reapCh: make(chan reapRequest, reapChanBuffer), reapInflight: map[reapKey]bool{}} + parkedAutoRetried: map[int64]bool{}, + reapCh: make(chan reapRequest, reapChanBuffer), reapInflight: map[reapKey]bool{}} } // SetNotifier wires an email sender for pass/job completion notifications. A @@ -442,6 +457,34 @@ func (c *Controller) advance(job *store.Job) error { return nil // phase still draining } if parked := counts[model.ShardParked]; parked > 0 { + // Give the backlog one automatic retry round before blocking for an + // operator: a parked shard exhausted MaxShardAttempts (5) grants, but + // many park causes are transient at the point they're hit (a mount + // blip, a brief NFS hiccup) rather than deterministic — by the time + // the rest of the pass has drained (potentially hours later on a + // large tree), conditions may well have changed, and RetryParkedByJob + // resets the attempt counter, so a retried shard gets the same fresh + // 5-attempt budget as any new shard rather than one bonus try. Only + // one round: parkedAutoRetried (keyed by pass, not shard — shard IDs + // change across the retry) stops this from firing every tick forever + // on a genuinely stuck shard, which would just be an infinite retry + // loop wearing a different name. If it re-parks after the one round, + // fall through to the existing block-and-alert behavior below. + c.parkedAlertMu.Lock() + alreadyRetried := c.parkedAutoRetried[pass.ID] + if !alreadyRetried { + c.parkedAutoRetried[pass.ID] = true + } + c.parkedAlertMu.Unlock() + if !alreadyRetried { + n, err := c.st.RetryParkedByJob(job.Name) + if err != nil { + return fmt.Errorf("auto-retry parked shards: %w", err) + } + slog.Info("auto-retried parked shards at end of pass", "job", job.Name, + "pass", pass.PassNo, "retried", n) + return nil // re-queued work needs a later tick to grant/drain + } // Do not advance past parked work silently; operator resolves via API. slog.Warn("pass blocked on parked shards", "job", job.Name, "pass", pass.PassNo, "parked", parked) diff --git a/docs/DESIGN-coordinator.md b/docs/DESIGN-coordinator.md index 134cc47..e6d0d80 100644 --- a/docs/DESIGN-coordinator.md +++ b/docs/DESIGN-coordinator.md @@ -91,9 +91,10 @@ PENDING ──▶ PROBING ──all probes ok──▶ SCANNING ──all shards synced into the underlying rootfs. A missing/misordered mount or a stub on any host is thus caught before bulk work runs — not just on whichever agent grabbed the root shard. A failed probe parks (like any shard), - and the parked-shard guard holds the pass until the operator fixes the mount and - retries. Probes pinned to an agent that departs after seeding are pruned so the phase - is not stalled. An empty fleet skips probing (nobody to probe or grant work to). + and the parked-shard guard holds the pass — after one automatic retry round, see + §2.3 — until the operator fixes the mount and retries. Probes pinned to an agent + that departs after seeding are pruned so the phase is not stalled. An empty fleet + skips probing (nobody to probe or grant work to). - `SCANNING` is the long phase: walk, diff, and copy are interleaved *per shard*, so data starts moving seconds after pass start; there is no global "scan first" barrier. - `DIRFIX` applies directory metadata deepest-first from the journal's dir records @@ -124,8 +125,8 @@ PENDING ──▶ PROBING ──all probes ok──▶ SCANNING ──all shards ``` QUEUED ──grant──▶ LEASED ──ShardResult ok──▶ DONE │ │ - lease expiry┘ └─ShardResult(err)──▶ PARKED ──(operator retry / auto after - ▼ transient-window)──▶ QUEUED + lease expiry┘ └─ShardResult(err)──▶ PARKED ──(operator retry, or one + ▼ automatic round at pass end)──▶ QUEUED QUEUED (attempt++) ``` @@ -134,6 +135,26 @@ QUEUED ──grant──▶ LEASED ──ShardResult ok──▶ DONE diagnosis breadcrumbs instead of poisoning the fleet forever. - Shards created by `ShardSplit` enter `QUEUED` in the same transaction that records the split against the parent (ordering invariant, protocol doc §4.2). +- **One automatic retry round at the end of a phase, then block for the operator.** + `advance()`'s parked-shard guard (§2.2) used to hold every phase transition open + indefinitely the instant any shard parked — correct, but on a long-running pass a + job could sit for hours waiting on a park caused by something transient at the + moment it happened (a mount blip, a brief NFS hiccup), not a shard that will fail + the same way forever. `advance()` now calls `store.RetryParkedByJob` once per pass + (`Controller.parkedAutoRetried`, keyed by pass id, in-memory — a coordinator restart + re-attempts one round for a pass already retried before the restart, a redundant + but harmless extra attempt, same "safe direction to be wrong in" reasoning as + `parkedAlerted` below) before falling back to the original block-and-alert + behavior. `RetryParkedByJob` resets `attempt` to 0, so a retried shard gets the + same fresh 5-attempt budget as any new shard, not one bonus try. Deliberately + bounded to exactly one round: an unbounded auto-retry on a genuinely stuck shard + (a real permissions problem, a permanently unreachable mount) would just be an + infinite retry loop wearing a different name, and would silently mask a job that + actually needs operator attention. `checkParkedShards`' alerting (`passctrl.go`, + the tick-driven parked-shard email digest) composes with this unchanged — a shard + that leaves and re-enters PARKED state naturally clears and re-sets its + `parkedAlerted` entry, so the operator is still alerted, just once the automatic + round has already been given a chance to make the alert unnecessary. ## 3. Schema (SQLite) diff --git a/webui/console.html b/webui/console.html index c9436ea..4543228 100644 --- a/webui/console.html +++ b/webui/console.html @@ -619,13 +619,17 @@

Jobs

migration lifecycle
-

Aggregate throughput

bytes copied · fleet-wide · 90s
+

Aggregate throughput

bytes copied · fleet-wide · 90s + + + +
–
throughputpeak –
- +
@@ -857,6 +861,15 @@

New job

while (v >= 1000 && i < u.length - 1) { v /= 1000; i++; } return [i === 0 ? String(Math.round(v)) : v.toFixed(v < 10 ? 2 : 1), u[i]]; } + // scanRate: entries/s → [value, unit] — the same [value, unit] shape as + // rate() so drawTL can treat both metrics identically, just with SI counts + // (1000-based, matching countFmt) instead of a byte/bit rate. + function scanRate(eps) { + eps = +eps || 0; + const u = ["/s","K/s","M/s","B/s"]; let i = 0; + while (eps >= 1000 && i < u.length - 1) { eps /= 1000; i++; } + return [i === 0 ? String(Math.round(eps)) : eps.toFixed(eps < 10 ? 2 : 1), u[i]]; + } function countFmt(n) { n = +n || 0; if (n >= 1e9) return (n / 1e9).toFixed(2) + "B"; @@ -1718,8 +1731,25 @@

New job

: `
no parked shards
`; } - // ---------- throughput timeline (auto-scaling canvas) ---------- + // ---------- aggregate timeline (auto-scaling canvas) ---------- + // "bw" (default) plots bwHist through rate() (throughput, unit-mode aware); + // "scan" plots scanHist through scanRate() (entries/s). Same canvas, same + // draw routine — just a different history array and formatter, persisted + // like unitsMode so a reload keeps the operator's last choice. + let tlMode = localStorage.getItem("drsync.tlmode") === "scan" ? "scan" : "bw"; + function applyTLMode(m) { + tlMode = m; + document.querySelectorAll("#tl-mode button").forEach(b => { const on = b.dataset.tl === m; b.classList.toggle("on", on); b.setAttribute("aria-selected", on); }); + $("#tl-title").textContent = m === "scan" ? "Aggregate scan rate" : "Aggregate throughput"; + $("#tl-eyebrow").textContent = m === "scan" ? "entries scanned · fleet-wide · 90s" : "bytes copied · fleet-wide · 90s"; + cv.setAttribute("aria-label", m === "scan" ? "Aggregate scan rate timeline" : "Aggregate throughput timeline"); + } + $("#tl-mode").addEventListener("click", e => { + const b = e.target.closest("button[data-tl]"); if (!b) return; + applyTLMode(b.dataset.tl); localStorage.setItem("drsync.tlmode", tlMode); drawTL(); + }); const cv = $("#tl"), ctx = cv.getContext("2d"); + applyTLMode(tlMode); function sizeCanvas() { const dpr = Math.min(devicePixelRatio || 1, 2), w = cv.clientWidth, h = 150; cv.width = w * dpr; cv.height = h * dpr; ctx.setTransform(dpr, 0, 0, dpr, 0, 0); @@ -1728,14 +1758,16 @@

New job

const niceTop = v => { if (v <= 0) return 1; const p = Math.pow(10, Math.floor(Math.log10(v))), n = v / p; return (n <= 1 ? 1 : n <= 2 ? 2 : n <= 5 ? 5 : 10) * p * 1.05; }; function drawTL() { + const fmt = tlMode === "scan" ? scanRate : rate; + const src = tlMode === "scan" ? scanHist : bwHist; const { w, h } = sizeCanvas(); ctx.clearRect(0, 0, w, h); const pad = { t:12, r:6, b:16, l:6 }, gw = w - pad.l - pad.r, gh = h - pad.t - pad.b; - const data = bwHist.length ? bwHist : [0], n = data.length, hi = niceTop(Math.max(...data, 1)); + const data = src.length ? src : [0], n = data.length, hi = niceTop(Math.max(...data, 1)); const x = i => n < 2 ? pad.l + gw : pad.l + i / (n - 1) * gw; const y = v => pad.t + (1 - v / hi) * gh; ctx.strokeStyle = css("--grid"); ctx.lineWidth = 1; ctx.font = "10px ui-monospace, monospace"; ctx.fillStyle = css("--muted"); [0, hi / 2, hi].forEach(g => { ctx.beginPath(); ctx.moveTo(pad.l, y(g)); ctx.lineTo(w - pad.r, y(g)); ctx.stroke(); - const [rv, ru] = rate(g); ctx.fillText(rv + " " + ru, pad.l + 2, y(g) - 3); }); + const [rv, ru] = fmt(g); ctx.fillText(rv + " " + ru, pad.l + 2, y(g) - 3); }); const acc = css("--accent"), acc2 = css("--accent-2"); const grad = ctx.createLinearGradient(0, pad.t, 0, h - pad.b); grad.addColorStop(0, hexA(acc, .34)); grad.addColorStop(1, hexA(acc, .02)); @@ -1746,10 +1778,10 @@

New job

const ex = x(n - 1), ey = y(data[n - 1]); ctx.beginPath(); ctx.arc(ex, ey, 6, 0, 7); ctx.fillStyle = hexA(acc2, .22); ctx.fill(); ctx.beginPath(); ctx.arc(ex, ey, 3.2, 0, 7); ctx.fillStyle = acc2; ctx.fill(); - const [cvv, cu] = rate(data[n - 1]); $("#tl-now").textContent = cvv; + const [cvv, cu] = fmt(data[n - 1]); $("#tl-now").textContent = cvv; const uEl = document.querySelector(".tl-big .u"); if (uEl) uEl.textContent = cu; $("#tl-leg-u").textContent = cu; - const [pv, pu] = rate(Math.max(...data)); $("#tl-peak").textContent = "peak " + pv + " " + pu; + const [pv, pu] = fmt(Math.max(...data)); $("#tl-peak").textContent = "peak " + pv + " " + pu; } function hexA(hex, a) { hex = hex.replace("#", ""); if (hex.length === 3) hex = hex.split("").map(c => c + c).join(""); const n = parseInt(hex, 16); return `rgba(${(n >> 16) & 255},${(n >> 8) & 255},${n & 255},${a})`; } diff --git a/webui/test/console.test.mjs b/webui/test/console.test.mjs index d08655d..facb775 100644 --- a/webui/test/console.test.mjs +++ b/webui/test/console.test.mjs @@ -184,6 +184,33 @@ test("a closely-spaced poll does not blank the fleet's rates", async () => { assert.ok(/Mbps|Gbps|Kbps|MiB|GiB|KiB/.test(fleet), "fleet throughput column blanked"); }); +test("aggregate timeline toggles between throughput and scan rate", async () => { + // Default view is throughput: the scan-rate tab is present but inactive. + const bwTab = c.$('#tl-mode button[data-tl="bw"]'); + const scanTab = c.$('#tl-mode button[data-tl="scan"]'); + assert.ok(bwTab.classList.contains("on"), "throughput tab not selected by default"); + assert.ok(!scanTab.classList.contains("on"), "scan-rate tab selected before being clicked"); + assert.match(c.text("#tl-title"), /throughput/i); + + const bwNow = c.text("#tl-now"); + assert.notEqual(bwNow, "–", "timeline never populated in throughput mode"); + + scanTab.click(); + await c.tick(50); + + assert.ok(scanTab.classList.contains("on"), "scan-rate tab did not activate on click"); + assert.ok(!bwTab.classList.contains("on"), "throughput tab still marked active after switching"); + assert.match(c.text("#tl-title"), /scan rate/i); + const scanNow = c.text("#tl-now"); + assert.notEqual(scanNow, "–", "timeline blanked after switching to scan rate"); + + // Switching back restores the throughput reading, proving the two + // histories are genuinely independent series, not one relabeled. + bwTab.click(); + await c.tick(50); + assert.equal(c.text("#tl-now"), bwNow, "throughput reading changed after switching back"); +}); + // -------------------------------------------------------------------------- // Per-agent in-flight work // --------------------------------------------------------------------------