fix: re-arm the approver inbox with exponential backoff (R1.5) (#97)

The worker inbound path got a proper re-arming backoff (2s→60s, doubling,
re-arms on any failure). The approver inbox — added later as a deliberately
parallel pool — never did, and kept the original defect verbatim:

    onClose: () => {
      approverPool.delete(key);
      if (approverIdentities.has(vtaDid)) {
        setTimeout(() => void getApproverWarmSession(vtaDid).catch(() => undefined), 2000);
      }
    }

One fixed 2s retry, scheduled only from `onClose`, with its failure swallowed.
If that retry failed, no session ever opened — so `onClose` could not fire
again and nothing re-armed. A mediator outage longer than ~2s left the
approver's inbox permanently dead until the operator happened to re-run the
biometric unlock.

That inbox is the one that receives `task-consent/request`. A dead listener
there means gated actions silently never get their human check — exactly the
failure R1.5 exists to prevent, and a security control rather than a
convenience (R7.2).

Adds `ReconnectScheduler` to core: retry forever, cap the delay not the
attempt count, re-arm on every failure including first-connect (where no open
means no close), never stack timers per key, reset to base on success, and an
optional `shouldRetry` gate so an operator lock beats a retry already in
flight. Timers are injectable so the loop is testable without real waiting.

The worker path deliberately keeps its own copy for now. That pool is
documented as isolated so it "can never disturb the working worker inbound
path" — adopting the shared scheduler there is a change to that path, not to
this one, and belongs in its own PR.

Verified: 9 new tests covering re-arm, doubling+cap, 60s-outage recovery,
reset-on-success, timer non-stacking, lock-beats-retry, cancel-during-flight
and per-key independence. Confirmed they FAIL (4/9) against a mutant with the
re-arm removed, so they detect the original bug rather than merely passing.
Full suite 180/180, clean build and lint, MV3 single-bundle guard holds.

Signed-off-by: Glenn Gore <glenn.g@affinidi.com>

Author: stormer78

packages/core/src/inbound/index.ts

@@ -2,3 +2,4 @@ export * from "./confirm.js";
 export * from "./task-consent.js";
 export * from "./effect-format.js";
 export * from "./dedup.js";
+export * from "./reconnect.js";

packages/core/src/inbound/reconnect.ts

@@ -0,0 +1,114 @@
+/**
+ * Re-arming exponential backoff for inbound mediator sessions (R1.5).
+ *
+ * A listener that stops retrying is a listener that silently misses every
+ * consent request that arrives afterwards — the human check for a gated action
+ * never happens and nothing reports that it didn't (R7.2). So the contract
+ * here is deliberately narrow and deliberately relentless:
+ *
+ *   - retry forever; cap the DELAY, never the attempt count
+ *   - re-arm on EVERY failure, including the first-connect case where no
+ *     session ever opened (an `onClose`-driven retry cannot cover that — no
+ *     open means no close)
+ *   - never let two timers stack for the same key
+ *   - reset to the base delay on success, so a recovered session doesn't carry
+ *     a grown delay into its next outage
+ *
+ * `attempt` must not throw: it reports success as a boolean. A throw escaping
+ * the timer callback would kill the loop, which is precisely how the previous
+ * fire-and-forget `.catch(() => undefined)` retry gave up for good.
+ *
+ * Timer functions are injectable so the loop is testable without real waiting.
+ */
+
+export interface ReconnectTimers {
+  setTimeout: (fn: () => void, ms: number) => ReturnType<typeof setTimeout>;
+  clearTimeout: (handle: ReturnType<typeof setTimeout>) => void;
+}
+
+export interface ReconnectOptions {
+  /** First retry delay, and the value reset to on success. */
+  baseMs: number;
+  /** Ceiling for the doubling delay. Bounds the blind window after recovery. */
+  maxMs: number;
+  /** Bring the session up. Must resolve `true` on success, `false` on ANY
+   *  failure. Must not throw. */
+  attempt: (key: string) => Promise<boolean>;
+  /** Optional gate consulted before each attempt and again before re-arming.
+   *  Returning false abandons the loop and clears state — used for "the
+   *  operator locked this identity", which must beat a retry in flight. */
+  shouldRetry?: (key: string) => boolean;
+  timers?: ReconnectTimers;
+}
+
+interface Entry {
+  delayMs: number;
+  timer: ReturnType<typeof setTimeout> | undefined;
+}
+
+export class ReconnectScheduler {
+  readonly #opts: ReconnectOptions;
+  readonly #timers: ReconnectTimers;
+  readonly #entries = new Map<string, Entry>();
+
+  constructor(opts: ReconnectOptions) {
+    this.#opts = opts;
+    this.#timers = opts.timers ?? {
+      setTimeout: (fn, ms) => setTimeout(fn, ms),
+      clearTimeout: (h) => clearTimeout(h),
+    };
+  }
+
+  /** Queue a retry for `key`, or do nothing if one is already queued. */
+  schedule(key: string): void {
+    if (this.#opts.shouldRetry && !this.#opts.shouldRetry(key)) {
+      this.clear(key);
+      return;
+    }
+    const entry = this.#entries.get(key) ?? { delayMs: this.#opts.baseMs, timer: undefined };
+    if (entry.timer) return; // already queued — never stack timers for one key
+    entry.timer = this.#timers.setTimeout(() => {
+      entry.timer = undefined;
+      void this.#opts.attempt(key).then((ok) => {
+        if (ok) {
+          this.clear(key);
+          return;
+        }
+        // A lock (or equivalent) during the attempt wins over the retry.
+        if (this.#opts.shouldRetry && !this.#opts.shouldRetry(key)) {
+          this.clear(key);
+          return;
+        }
+        // Don't resurrect a backoff cancelled while this attempt was in flight
+        // — `clear()` dropped the entry, and re-arming would revive a loop the
+        // caller explicitly ended.
+        if (this.#entries.get(key) !== entry) return;
+        entry.delayMs = Math.min(entry.delayMs * 2, this.#opts.maxMs);
+        this.schedule(key);
+      });
+    }, entry.delayMs);
+    this.#entries.set(key, entry);
+  }
+
+  /** Cancel any pending retry for `key` and reset its delay to the base. */
+  clear(key: string): void {
+    const entry = this.#entries.get(key);
+    if (entry?.timer) this.#timers.clearTimeout(entry.timer);
+    this.#entries.delete(key);
+  }
+
+  /** Cancel every pending retry. */
+  clearAll(): void {
+    for (const key of [...this.#entries.keys()]) this.clear(key);
+  }
+
+  /** The delay the NEXT retry for `key` would use. Test/telemetry surface. */
+  pendingDelayMs(key: string): number | undefined {
+    return this.#entries.get(key)?.delayMs;
+  }
+
+  /** Whether a retry is currently queued for `key`. */
+  isArmed(key: string): boolean {
+    return this.#entries.get(key)?.timer !== undefined;
+  }
+}

packages/core/tests/inbound.reconnect.mjs

@@ -0,0 +1,255 @@
+import { test } from "node:test";
+import assert from "node:assert/strict";
+
+import { ReconnectScheduler } from "../dist/inbound/reconnect.js";
+
+/** A controllable clock. Real timers would make these tests slow and flaky;
+ *  worse, the behaviour under test IS the timing, so it has to be observable
+ *  rather than waited out. */
+function fakeTimers() {
+  let now = 0;
+  let seq = 0;
+  const queued = new Map(); // handle -> { at, fn }
+  return {
+    timers: {
+      setTimeout: (fn, ms) => {
+        const handle = ++seq;
+        queued.set(handle, { at: now + ms, fn });
+        return handle;
+      },
+      clearTimeout: (handle) => queued.delete(handle),
+    },
+    /** Advance the clock, firing due callbacks in time order. */
+    async advance(ms) {
+      const target = now + ms;
+      for (;;) {
+        let next = null;
+        for (const [handle, t] of queued) {
+          if (t.at <= target && (next === null || t.at < queued.get(next).at)) next = handle;
+        }
+        if (next === null) break;
+        const { at, fn } = queued.get(next);
+        queued.delete(next);
+        now = at;
+        fn();
+        // Let the promise chain inside the callback settle before the next tick.
+        await new Promise((r) => setImmediate(r));
+      }
+      now = target;
+    },
+    pending: () => queued.size,
+  };
+}
+
+test("a failing attempt re-arms instead of giving up after one retry", async () => {
+  // The original bug: ONE 2s retry, and if it failed nothing ever tried again
+  // because no session opened, so onClose could not fire a second time.
+  const clock = fakeTimers();
+  let attempts = 0;
+  const s = new ReconnectScheduler({
+    baseMs: 2_000,
+    maxMs: 60_000,
+    attempt: async () => {
+      attempts++;
+      return false;
+    },
+    timers: clock.timers,
+  });
+
+  s.schedule("vta");
+  await clock.advance(2_000);
+  assert.equal(attempts, 1);
+  // The whole point: still armed after the first failure.
+  assert.equal(s.isArmed("vta"), true, "must re-arm after a failed attempt");
+
+  await clock.advance(4_000);
+  assert.equal(attempts, 2);
+  await clock.advance(8_000);
+  assert.equal(attempts, 3);
+});
+
+test("the delay doubles and then caps, so retries never stop", async () => {
+  const clock = fakeTimers();
+  const s = new ReconnectScheduler({
+    baseMs: 2_000,
+    maxMs: 16_000,
+    attempt: async () => false,
+    timers: clock.timers,
+  });
+
+  s.schedule("vta");
+  const seen = [];
+  for (let i = 0; i < 8; i++) {
+    // Advance by exactly the queued delay so each pass fires one retry —
+    // a coarser jump would run several doublings inside one advance.
+    const delay = s.pendingDelayMs("vta");
+    seen.push(delay);
+    await clock.advance(delay);
+  }
+  assert.deepEqual(seen, [2_000, 4_000, 8_000, 16_000, 16_000, 16_000, 16_000, 16_000]);
+  assert.equal(s.isArmed("vta"), true, "a capped backoff is still a live backoff");
+});
+
+test("a mediator down for 60s recovers on its own once it returns", async () => {
+  const clock = fakeTimers();
+  let mediatorUp = false;
+  let connects = 0;
+  const s = new ReconnectScheduler({
+    baseMs: 2_000,
+    maxMs: 60_000,
+    attempt: async () => {
+      if (!mediatorUp) return false;
+      connects++;
+      return true;
+    },
+    timers: clock.timers,
+  });
+
+  s.schedule("vta");
+  await clock.advance(60_000);
+  assert.equal(connects, 0, "still down, so no connect");
+  assert.equal(s.isArmed("vta"), true, "but still trying");
+
+  mediatorUp = true;
+  await clock.advance(64_000);
+  assert.equal(connects, 1, "recovers without a worker reboot");
+  assert.equal(s.isArmed("vta"), false, "and stops retrying once connected");
+});
+
+test("success resets the delay, so the next outage starts from the base", async () => {
+  const clock = fakeTimers();
+  let up = false;
+  const s = new ReconnectScheduler({
+    baseMs: 2_000,
+    maxMs: 60_000,
+    attempt: async () => up,
+    timers: clock.timers,
+  });
+
+  s.schedule("vta");
+  await clock.advance(14_000); // fail a few times, growing the delay
+  up = true;
+  await clock.advance(60_000); // connect
+  assert.equal(s.isArmed("vta"), false);
+
+  up = false;
+  s.schedule("vta"); // next outage
+  assert.equal(s.pendingDelayMs("vta"), 2_000, "must not inherit the grown delay");
+});
+
+test("timers never stack for one key", async () => {
+  const clock = fakeTimers();
+  let attempts = 0;
+  const s = new ReconnectScheduler({
+    baseMs: 2_000,
+    maxMs: 60_000,
+    attempt: async () => {
+      attempts++;
+      return false;
+    },
+    timers: clock.timers,
+  });
+
+  s.schedule("vta");
+  s.schedule("vta");
+  s.schedule("vta");
+  assert.equal(clock.pending(), 1, "three schedules, one timer");
+  await clock.advance(2_000);
+  assert.equal(attempts, 1);
+});
+
+test("shouldRetry=false abandons the loop — a lock beats a retry in flight", async () => {
+  // The operator locking the approver must win: resurrecting the session would
+  // defeat an explicit security act.
+  const clock = fakeTimers();
+  let unlocked = true;
+  let attempts = 0;
+  const s = new ReconnectScheduler({
+    baseMs: 2_000,
+    maxMs: 60_000,
+    attempt: async () => {
+      attempts++;
+      unlocked = false; // locked during the attempt
+      return false;
+    },
+    shouldRetry: () => unlocked,
+    timers: clock.timers,
+  });
+
+  s.schedule("vta");
+  await clock.advance(2_000);
+  assert.equal(attempts, 1);
+  assert.equal(s.isArmed("vta"), false, "must not re-arm after a lock");
+
+  await clock.advance(60_000);
+  assert.equal(attempts, 1, "and must never attempt again");
+});
+
+test("scheduling while already locked does not arm anything", async () => {
+  const clock = fakeTimers();
+  let attempts = 0;
+  const s = new ReconnectScheduler({
+    baseMs: 2_000,
+    maxMs: 60_000,
+    attempt: async () => {
+      attempts++;
+      return false;
+    },
+    shouldRetry: () => false,
+    timers: clock.timers,
+  });
+
+  s.schedule("vta");
+  assert.equal(clock.pending(), 0);
+  await clock.advance(60_000);
+  assert.equal(attempts, 0);
+});
+
+test("clear() cancels a queued retry and a retry in flight cannot revive it", async () => {
+  const clock = fakeTimers();
+  let attempts = 0;
+  let release;
+  const gate = new Promise((r) => (release = r));
+  const s = new ReconnectScheduler({
+    baseMs: 2_000,
+    maxMs: 60_000,
+    attempt: async () => {
+      attempts++;
+      await gate; // still in flight when clear() lands
+      return false;
+    },
+    timers: clock.timers,
+  });
+
+  s.schedule("vta");
+  await clock.advance(2_000);
+  assert.equal(attempts, 1);
+  s.clear("vta"); // cancelled mid-attempt
+  release();
+  await new Promise((r) => setImmediate(r));
+  assert.equal(s.isArmed("vta"), false, "a cancelled backoff must stay cancelled");
+
+  await clock.advance(60_000);
+  assert.equal(attempts, 1);
+});
+
+test("independent keys back off independently", async () => {
+  const clock = fakeTimers();
+  const attempts = { a: 0, b: 0 };
+  const s = new ReconnectScheduler({
+    baseMs: 2_000,
+    maxMs: 60_000,
+    attempt: async (key) => {
+      attempts[key]++;
+      return key === "b"; // b connects, a keeps failing
+    },
+    timers: clock.timers,
+  });
+
+  s.schedule("a");
+  s.schedule("b");
+  await clock.advance(2_000);
+  assert.deepEqual(attempts, { a: 1, b: 1 });
+  assert.equal(s.isArmed("a"), true, "a is still retrying");
+  assert.equal(s.isArmed("b"), false, "b connected and stopped");
+});