Watchdog notice reader: a future stamp is eligible now; no production clock seam

A failed entry stamped ahead of the clock (host clock stepped forward then
corrected) is eligible now instead of waiting out the skew; the next attempt
rewrites the stamp. The WATCHDOG_NOTICE_NOW_MS env var is deleted: the reader
reads Date.now(), and tests fix it with a --import preload (tests/fake-clock.mjs).
The unknown-entry tests now rerun far past the backoff so a retried unknown fails.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

docs/watchdog-notice.md

@@ -27,7 +27,7 @@ node tools/watchdog-notice.mjs status --journal-store J
 A notice whose send failed (bridge exit 75) is never dropped. It is eligible
 again once `min(2^(attempts-1), 60)` minutes have passed since its last attempt
 in the journal: 1, 2, 4, 8, 16, 32, then every 60 minutes. A run that finds it
-inside its delay skips it without spawning the bridge. Every attempt carries the
+inside its delay skips it without spawning the bridge. An entry stamped ahead of the clock (a host clock stepped forward, then corrected) is eligible now, and its next attempt rewrites the stamp. Every attempt carries the
 same nonce, so a Discord outage of any length costs a late notice, not a lost
 one, and a retry that succeeds is the one post.
 

tests/fake-clock.mjs

@@ -0,0 +1,5 @@
+// Test preload: fixes Date.now() to FAKE_NOW_MS when a test sets it.
+if (process.env.FAKE_NOW_MS) {
+  const fixed = Number(process.env.FAKE_NOW_MS);
+  Date.now = () => fixed;
+}

tests/watchdog-notice.test.mjs

@@ -72,25 +72,27 @@ async function world(prefix, { incidents = FIXTURE, mode = "ok" } = {}) {
   if (incidents) await copyFile(incidents, w.incidents);
   w.setMode = (value) => writeFile(resolve(dir, "mode"), value);
   w.run = (extra = {}) => run(w, extra);
-  w.status = () => runStatus(w);
+  w.status = (env) => runStatus(w, env);
   w.posts = async () => (existsSync(w.calls) ? (await readFile(w.calls, "utf8")).trim().split("\n").map((line) => JSON.parse(line)) : []);
   return w;
 }
 
-// The reader's clock is WATCHDOG_NOTICE_NOW_MS; a test drives the backoff by
-// naming the minute (from T0) each run happens at.
+// The reader reads Date.now(). A test drives the backoff the way it stubs fetch:
+// a preload (fake-clock.mjs, passed to the reader as --import) fixes Date.now
+// to the minute (from T0) the run happens at. The reader has no clock seam.
 const T0 = 1_800_000_000_000;
-const at = (minutes) => ({ WATCHDOG_NOTICE_NOW_MS: String(T0 + Math.round(minutes * 60_000)) });
+const CLOCK = resolve(import.meta.dirname, "fake-clock.mjs");
+const at = (minutes) => ({ FAKE_NOW_MS: String(T0 + Math.round(minutes * 60_000)) });
 
 function run(w, { env = {}, bridge = w.bridge } = {}) {
-  return spawnSync(process.execPath, [TOOL, "process", "--incident-store", w.incidents, "--journal-store", w.journal, "--receipt-store", w.receipts, "--bridge-cli", bridge], {
+  return spawnSync(process.execPath, ["--import", CLOCK, TOOL, "process", "--incident-store", w.incidents, "--journal-store", w.journal, "--receipt-store", w.receipts, "--bridge-cli", bridge], {
     encoding: "utf8",
     env: { ...process.env, DISCORD_OWNER_ID: RECIPIENT, ...env },
   });
 }
 
-function runStatus(w) {
-  const result = spawnSync(process.execPath, [TOOL, "status", "--journal-store", w.journal], { encoding: "utf8", env: process.env });
+function runStatus(w, env = {}) {
+  const result = spawnSync(process.execPath, ["--import", CLOCK, TOOL, "status", "--journal-store", w.journal], { encoding: "utf8", env: { ...process.env, ...env } });
   return { ...result, rows: result.status === null || !result.stdout ? [] : JSON.parse(result.stdout) };
 }
 
@@ -178,10 +180,23 @@ test("a running entry becomes unknown and is never re-posted", async () => {
   assert.equal(posts.filter((post) => opt(post, "cultmesh-command-id") === `${CLOSED_KEY}#opened`).length, 1, "the interrupted notice is not sent again");
   const rows = (await w.status()).rows;
   assert.equal(rows.find((row) => row.incidentKey === CLOSED_KEY && row.notice === "opened").status, "unknown");
-  assert.notEqual(w.run().status, 0, "still failing on the next run");
+  assert.notEqual(w.run({ env: at(100_000) }).status, 0, "still failing, and not re-posted, long past any backoff");
   assert.equal((await w.posts()).length, posts.length);
 });
 
+test("an entry stamped ahead of the clock is eligible now, not deferred by the skew", async () => {
+  const w = await world("wn-skew-", { mode: "fail" });
+  w.run({ env: at(1000) });
+  assert.equal((await w.posts()).length, 2, "both openings tried once, stamped at minute 1000");
+  w.run({ env: at(0) });
+  assert.equal((await w.posts()).length, 4, "the clock is behind the stamp: retried at once");
+  assert.equal((await w.status(at(0))).rows[0].nextEligibleAt, new Date(T0 + 2 * 60_000).toISOString(), "the retry rewrote the stamp from the corrected clock");
+  w.run({ env: at(1.5) });
+  assert.equal((await w.posts()).length, 4, "the cadence holds from the rewritten stamp");
+  w.run({ env: at(2) });
+  assert.equal((await w.posts()).length, 6);
+});
+
 const backoffMinutes = [0, 1, 3, 7, 15, 31, 63, 123, 183];
 
 test("a failed notice is retried after its backoff and never exhausted", async () => {
@@ -326,8 +341,8 @@ test("a bridge that dies after sending is unknown and never re-posted", async ()
     assert.equal(JSON.parse(first.stdout).unknown, 2);
     assert.notEqual((await w.status()).status, 0, "status exits non-zero while an entry is unknown");
     await w.setMode("ok");
-    assert.notEqual(w.run().status, 0);
-    assert.notEqual(w.run().status, 0);
+    assert.notEqual(w.run({ env: at(1000) }).status, 0, "far past any backoff");
+    assert.notEqual(w.run({ env: at(100_000) }).status, 0);
     const posts = await w.posts();
     assert.equal(posts.length, 2, "one opening post per incident, no retry, no closure");
     assert.ok(posts.every((post) => notice(post) === "opened"));
@@ -343,8 +358,8 @@ test("a spawn timeout and a spawn error are unknown, never failed, never re-post
     assert.notEqual(first.status, 0, mode);
     assert.equal(JSON.parse(first.stdout).unknown, 2, mode);
     await w.setMode("ok");
-    assert.notEqual(w.run().status, 0);
-    assert.notEqual(w.run().status, 0);
+    assert.notEqual(w.run({ env: at(1000) }).status, 0, "far past any backoff");
+    assert.notEqual(w.run({ env: at(100_000) }).status, 0);
     const posts = await w.posts();
     assert.equal(posts.length, 2, `${mode}: one opening post per incident, no retry, no closure`);
     assert.ok(posts.every((post) => notice(post) === "opened"));

tools/watchdog-notice.mjs

@@ -32,10 +32,8 @@ const JOURNAL_SCHEMA = "bifrost.watchdog_notice_execution.v1";
 // failures on, the exit status is the delivery-failure signal; retries go on.
 const MAX_DELAY_MINUTES = 60;
 const FAILING_ATTEMPTS = 3;
-// The overrides exist so a test can reach the timeout branch without a 60 s wait
-// and can drive the backoff without sleeping.
+// The override exists so a test can reach the timeout branch without a 60 s wait.
 const BRIDGE_TIMEOUT_MS = Number(process.env.WATCHDOG_NOTICE_BRIDGE_TIMEOUT_MS) || 60_000;
-const clockMs = () => Number(process.env.WATCHDOG_NOTICE_NOW_MS) || Date.now();
 const STATUSES = ["running", "completed", "failed", "unknown"];
 const NAME = /^[a-z][a-z-]{0,63}$/;
 // Idunn's require_id (GameCult/Idunn src/control_plane.rs:9122-9131 at 7f7528b):
@@ -88,7 +86,7 @@ async function processIncidents(options) {
     const outcome = await sendNotice({ bridgeCli, recipientId, receiptStore, key, nonce, record, content });
     await journal.put({ ...base, messageId: outcome.messageId ?? "", status: outcome.status, lastError: outcome.error ?? "", updatedAt: now() });
   };
-  const retryable = (entry) => !entry || (entry.status === "failed" && !(clockMs() < nextEligibleMs(entry)));
+  const retryable = (entry) => !entry || (entry.status === "failed" && Date.now() >= nextEligibleMs(entry));
   const entryFor = (record, notice) => journal.cache.get(journal.definition, `${record.incidentKey}#${notice}`);
 
   for (const record of records) {
@@ -121,9 +119,14 @@ async function status(options) {
 }
 
 const isFailing = (entry) => entry.status === "failed" && entry.attempts >= FAILING_ATTEMPTS;
-// Measured from the failed entry's last attempt in the journal.
-const nextEligibleMs = (entry) => Date.parse(entry.updatedAt) + Math.min(2 ** (entry.attempts - 1), MAX_DELAY_MINUTES) * 60_000;
-const now = () => new Date(clockMs()).toISOString();
+// Measured from the failed entry's last attempt in the journal. A stamp ahead of
+// the clock is skew (a host clock stepped forward, then corrected), not a delay
+// to wait out: that entry is eligible now, and its next attempt rewrites the stamp.
+const nextEligibleMs = (entry) => {
+  const stamp = Date.parse(entry.updatedAt);
+  return stamp > Date.now() ? Date.now() : stamp + Math.min(2 ** (entry.attempts - 1), MAX_DELAY_MINUTES) * 60_000;
+};
+const now = () => new Date(Date.now()).toISOString();
 
 // Reads P unlocked. An absent file is "nothing to do" (records: null). A
 // record of another schema is counted as skipped; a record of this schema