Stop one lost extrinsic from silencing the noter bot (#30)

Every report waits behind the previous one, and a submission only settled
when it reached a block. One the pool dropped, rejected or replaced never
settled, so every report queued after it waited forever. The watchdog meant
to catch a silent bot was armed only after the first scan returned, so when
that first scan was the one that hung there was no watchdog at all: the
production container logged its last line one minute after starting on
2026-10-13 and stayed up, silent, for more than a day.

- submit.js: a submission settles on a terminal status (Dropped, Invalid,
  Usurped, FinalityTimeout) or after SUBMIT_TIMEOUT_MS, and releases its
  status subscription.
- The watchdog is armed before connecting and counts from the process start
  until a scan completes, so a heartbeat left by an earlier process in the
  same container does not count for this one.
- The noter's own account is skipped: the chain refuses a noter's report
  about themselves, so it failed in every scan.
- Unit tests for both rules, run inside the noter image in CI.

.github/workflows/ci.yml

@@ -77,6 +77,11 @@ jobs:
       - name: Crypto stack
         run: docker run --rm -v "$PWD/scripts/bot-smoke.mjs:/app/bot-smoke.mjs:ro" bot node bot-smoke.mjs ${{ matrix.mode }}
 
+      # Run inside the image, so the modules under test are the ones the image ships.
+      - name: Unit tests
+        if: matrix.bot == 'noter'
+        run: docker run --rm -v "$PWD/noter/test:/app/test:ro" bot node --test test/submit.test.mjs
+
       - name: The bot loads and stops at its missing mnemonic
         env:
           EXPECT: ${{ matrix.expect }}

noter/Dockerfile

@@ -10,6 +10,6 @@ WORKDIR /app
 COPY package.json package-lock.json ./
 RUN npm ci --omit=dev --omit=optional --no-audit --no-fund
 
-COPY bot.js ./
+COPY bot.js submit.js ./
 
 CMD ["node", "bot.js"]

noter/bot.js

@@ -22,6 +22,7 @@ import { ApiPromise, WsProvider } from '@pezkuwi/api';
 import { Keyring } from '@pezkuwi/keyring';
 import { cryptoWaitReady } from '@pezkuwi/util-crypto';
 import fs from 'fs';
+import { watchSubmission, isStale } from './submit.js';
 
 // ========================================
 // CONFIGURATION
@@ -45,6 +46,11 @@ const UNITS = BigInt('1000000000000'); // 10^12
 // restart policy turns a silent stall into a visibly crash-looping container.
 const HEARTBEAT_FILE = process.env.HEARTBEAT_FILE || '/tmp/noter-heartbeat';
 const STALE_AFTER = SCAN_INTERVAL * 3;
+const STARTED_AT = Date.now();
+// How long one submission may wait for a block before it is given up on. Every submission
+// waits behind the previous one, so this bounds how long a single lost extrinsic can hold
+// up all the reports queued after it.
+const SUBMIT_TIMEOUT = parseInt(process.env.SUBMIT_TIMEOUT_MS || '120000', 10);
 
 // ========================================
 // LOGGING
@@ -58,14 +64,14 @@ function touchHeartbeat() {
   }
 }
 
-/// Milliseconds since the last completed scan, or `Infinity` if none has completed
-/// yet — an unwritable or missing heartbeat counts as stale rather than healthy.
-function heartbeatAge() {
+/// When the last scan completed, or `NaN` if none has — a missing or unreadable heartbeat.
+/// A heartbeat left by an earlier process in the same container is older than
+/// `STARTED_AT`, so `isStale` does not let it count for this one.
+function lastScanAt() {
   try {
-    const t = Number(fs.readFileSync(HEARTBEAT_FILE, 'utf8'));
-    return Number.isFinite(t) ? Date.now() - t : Infinity;
+    return Number(fs.readFileSync(HEARTBEAT_FILE, 'utf8'));
   } catch {
-    return Infinity;
+    return NaN;
   }
 }
 
@@ -317,26 +323,22 @@ function queueSubmit(peopleApi, noterKeypair, tx, description) {
 }
 
 function doSignAndSend(peopleApi, noterKeypair, tx, description) {
-  return new Promise((resolve, reject) => {
-    tx.signAndSend(noterKeypair, ({ status, dispatchError }) => {
-      if (status.isInBlock) {
-        if (dispatchError) {
-          let errMsg = dispatchError.toString();
-          if (dispatchError.isModule) {
-            try {
-              const decoded = peopleApi.registry.findMetaError(dispatchError.asModule);
-              errMsg = `${decoded.section}.${decoded.name}: ${decoded.docs.join(' ')}`;
-            } catch { /* use default */ }
-          }
-          log('ERROR', 'TX failed', { description, error: errMsg, block: status.asInBlock.toHex() });
-          reject(new Error(errMsg));
-        } else {
-          log('INFO', `TX success: ${description}`, { block: status.asInBlock.toHex() });
-          resolve();
-        }
-      }
-    }).catch(reject);
-  });
+  const decodeError = (dispatchError) => {
+    if (dispatchError.isModule) {
+      try {
+        const decoded = peopleApi.registry.findMetaError(dispatchError.asModule);
+        return `${decoded.section}.${decoded.name}: ${decoded.docs.join(' ')}`;
+      } catch { /* use default */ }
+    }
+    return dispatchError.toString();
+  };
+  return watchSubmission(tx, noterKeypair, { decodeError, timeoutMs: SUBMIT_TIMEOUT }).then(
+    (block) => log('INFO', `TX success: ${description}`, { block }),
+    (err) => {
+      log('ERROR', 'TX failed', { description, error: err.message });
+      throw err;
+    },
+  );
 }
 
 /**
@@ -400,6 +402,11 @@ async function finalizeMaturedPending(peopleApi, noterKeypair) {
 // ========================================
 
 async function processAccount(relayApi, assetHubApi, peopleApi, noterKeypair, address) {
+  // The chain refuses a noter's report about themselves (`NoterCannotAttestSelf`), so a
+  // noter that has opted in itself failed on that account in every scan. Its stake has to
+  // come from another noter; compared as account ids, since the address format may differ.
+  if (peopleApi.createType('AccountId', address).eq(noterKeypair.publicKey)) return 0;
+
   const updates = [];
 
   // 1. Collect ALL staking data first (direct + pool) before comparing
@@ -559,6 +566,24 @@ async function main() {
   // Wait for crypto WASM to be ready
   await cryptoWaitReady();
 
+  // Exiting is the point. Reconnecting forever looks like the bot is coping; a
+  // container that keeps dying does not, and the restart policy makes that
+  // visible in `docker ps` without anyone having to read the logs.
+  //
+  // Armed before anything that can hang — connecting, the first scan — and counted
+  // from the start of this process until a scan completes. Armed after the first
+  // scan, it never ran on a process whose first scan hung.
+  setInterval(() => {
+    const last = lastScanAt();
+    if (isStale({ lastScanAt: last, startedAt: STARTED_AT, now: Date.now(), staleAfter: STALE_AFTER })) {
+      log('ERROR', 'No scan completed within the staleness window — exiting so the restart policy takes over', {
+        last_scan_at: Number.isFinite(last) && last >= STARTED_AT ? new Date(last).toISOString() : null,
+        stale_after_ms: STALE_AFTER,
+      });
+      process.exit(1);
+    }
+  }, Math.min(SCAN_INTERVAL, 60_000)).unref?.();
+
   // Load noter keypair
   const mnemonic = loadNoterMnemonic();
   const keyring = new Keyring({ type: 'sr25519' });
@@ -611,19 +636,6 @@ async function main() {
       });
   }, SCAN_INTERVAL);
 
-  // Exiting is the point. Reconnecting forever looks like the bot is coping; a
-  // container that keeps dying does not, and the restart policy makes that
-  // visible in `docker ps` without anyone having to read the logs.
-  setInterval(() => {
-    const age = heartbeatAge();
-    if (age > STALE_AFTER) {
-      log('ERROR', 'No scan completed within the staleness window — exiting so the restart policy takes over', {
-        stale_for_ms: Number.isFinite(age) ? age : null,
-        stale_after_ms: STALE_AFTER,
-      });
-      process.exit(1);
-    }
-  }, Math.min(SCAN_INTERVAL, 60_000)).unref?.();
 }
 
 main().catch(err => {

noter/submit.js

@@ -0,0 +1,67 @@
+// Watching one extrinsic, and the liveness rule for the scan loop.
+//
+// Kept apart from bot.js, which connects to three chains the moment it is imported, so that
+// both can be tested without a chain.
+
+/**
+ * Statuses after which an extrinsic will never reach a block. Each of them used to be ignored:
+ * the promise below only settled on `isInBlock`, so a submission the pool dropped, rejected
+ * or replaced left it pending forever, and since every submission waits its turn behind the
+ * previous one, one lost extrinsic stopped every report after it.
+ */
+const TERMINAL = ['isDropped', 'isInvalid', 'isUsurped', 'isFinalityTimeout'];
+
+/**
+ * Sign and submit `tx`, settling once it is in a block or can never be.
+ *
+ * Resolves on `isInBlock` without a dispatch error, as before: an extrinsic on a fork that is
+ * later retracted is not lost, because the next scan finds neither a pending nor a cached
+ * entry and submits again. Rejects on a dispatch error, on a terminal status, and when nothing
+ * has settled after `timeoutMs`, which also covers a status subscription the node stopped
+ * answering.
+ */
+export function watchSubmission(tx, signer, { decodeError, timeoutMs }) {
+  return new Promise((resolve, reject) => {
+    let unsub = null;
+    let settled = false;
+    const settle = (fn, value) => {
+      if (settled) return;
+      settled = true;
+      clearTimeout(timer);
+      if (typeof unsub === 'function') unsub();
+      fn(value);
+    };
+    const timer = setTimeout(
+      () => settle(reject, new Error(`no block after ${timeoutMs} ms`)),
+      timeoutMs,
+    );
+
+    tx.signAndSend(signer, ({ status, dispatchError }) => {
+      if (status.isInBlock) {
+        if (dispatchError) settle(reject, new Error(decodeError(dispatchError)));
+        else settle(resolve, status.asInBlock.toHex());
+        return;
+      }
+      const terminal = TERMINAL.find((flag) => status[flag]);
+      if (terminal) settle(reject, new Error(`extrinsic ${status.type ?? terminal.slice(2)}`));
+    })
+      .then((u) => {
+        unsub = u;
+        if (settled) u();
+      })
+      .catch((err) => settle(reject, err));
+  });
+}
+
+/**
+ * Whether the bot has gone too long without completing a scan.
+ *
+ * Counted from the later of the last completed scan and the process start. Counting only from
+ * the last scan left a process whose *first* scan hung with no watchdog at all, because the
+ * watchdog was armed after that scan returned: on 2026-10-13 the production container stopped
+ * logging one minute after it started and stayed up, silent, for more than a day.
+ */
+export function isStale({ lastScanAt, startedAt, now, staleAfter }) {
+  const since = Math.max(Number.isFinite(lastScanAt) ? lastScanAt : 0, startedAt);
+  return now - since > staleAfter;
+}

noter/test/submit.test.mjs

@@ -0,0 +1,85 @@
+// node --test noter/test
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { watchSubmission, isStale } from '../submit.js';
+
+// A stand-in for an api-derive submittable: it hands the status callback to the test and
+// records whether the subscription was released.
+function fakeTx() {
+  const t = { emit: null, unsubscribed: false };
+  t.signAndSend = (_signer, cb) => {
+    t.emit = (flags, extra = {}) => cb({ status: { type: flags.type, ...flags }, ...extra });
+    return Promise.resolve(() => { t.unsubscribed = true; });
+  };
+  return t;
+}
+const opts = { decodeError: (e) => `decoded ${e}`, timeoutMs: 1_000 };
+const tick = () => new Promise((r) => setImmediate(r));
+
+for (const flag of ['isDropped', 'isInvalid', 'isUsurped', 'isFinalityTimeout']) {
+  test(`a submission that ends ${flag.slice(2)} settles instead of waiting forever`, async () => {
+    const tx = fakeTx();
+    const p = watchSubmission(tx, {}, opts);
+    await tick();
+    tx.emit({ isReady: true, type: 'Ready' });
+    tx.emit({ [flag]: true, type: flag.slice(2) });
+    await assert.rejects(p, new RegExp(flag.slice(2)));
+    assert.equal(tx.unsubscribed, true);
+  });
+}
+
+test('a submission that never reaches a block is given up after the timeout', async () => {
+  const tx = fakeTx();
+  const p = watchSubmission(tx, {}, { ...opts, timeoutMs: 50 });
+  await tick();
+  tx.emit({ isReady: true, type: 'Ready' });
+  await assert.rejects(p, /no block after 50 ms/);
+  assert.equal(tx.unsubscribed, true);
+});
+
+test('a submission in a block resolves with that block', async () => {
+  const tx = fakeTx();
+  const p = watchSubmission(tx, {}, opts);
+  await tick();
+  tx.emit({ isInBlock: true, type: 'InBlock', asInBlock: { toHex: () => '0xabc' } });
+  assert.equal(await p, '0xabc');
+  assert.equal(tx.unsubscribed, true);
+});
+
+test('a dispatch error in a block rejects with the decoded error', async () => {
+  const tx = fakeTx();
+  const p = watchSubmission(tx, {}, opts);
+  await tick();
+  tx.emit({ isInBlock: true, type: 'InBlock', asInBlock: { toHex: () => '0xabc' } }, { dispatchError: 'E' });
+  await assert.rejects(p, /decoded E/);
+});
+
+test('a submission the node refuses outright rejects', async () => {
+  const tx = { signAndSend: () => Promise.reject(new Error('1014: Priority is too low')) };
+  await assert.rejects(watchSubmission(tx, {}, opts), /Priority is too low/);
+});
+
+test('a later status cannot settle a submission twice', async () => {
+  const tx = fakeTx();
+  const p = watchSubmission(tx, {}, opts);
+  await tick();
+  tx.emit({ isInBlock: true, type: 'InBlock', asInBlock: { toHex: () => '0x1' } });
+  tx.emit({ isDropped: true, type: 'Dropped' });
+  assert.equal(await p, '0x1');
+});
+
+test('a first scan that hangs goes stale, counted from the process start', () => {
+  assert.equal(isStale({ lastScanAt: NaN, startedAt: 0, now: 901, staleAfter: 900 }), true);
+  assert.equal(isStale({ lastScanAt: NaN, startedAt: 0, now: 899, staleAfter: 900 }), false);
+});
+
+test("an earlier process's heartbeat does not keep this one alive", () => {
+  // Heartbeat from long before this process started: still counted from the start.
+  assert.equal(isStale({ lastScanAt: 10, startedAt: 1_000, now: 1_500, staleAfter: 900 }), false);
+  assert.equal(isStale({ lastScanAt: 10, startedAt: 1_000, now: 1_901, staleAfter: 900 }), true);
+});
+
+test('a recent scan keeps the bot alive', () => {
+  assert.equal(isStale({ lastScanAt: 5_000, startedAt: 0, now: 5_500, staleAfter: 900 }), false);
+  assert.equal(isStale({ lastScanAt: 5_000, startedAt: 0, now: 5_901, staleAfter: 900 }), true);
+});