diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 4de66e9..3b7e026 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -163,9 +163,9 @@ services: ASSET_HUB_RPC: wss://asset-hub-rpc.pezkuwichain.io PEOPLE_RPC: wss://people-rpc.pezkuwichain.io SCAN_INTERVAL_MS: "300000" - # The bot writes a heartbeat when a scan completes, not when it merely stays - # up: an endless reconnect loop kept the container "Up" for three weeks while - # no staking data reached the People chain. + # The bot writes a heartbeat when a scan makes progress, not when it merely + # stays up: an endless reconnect loop kept the container "Up" for three weeks + # while no staking data reached the People chain. healthcheck: test: ["CMD", "node", "-e", "const fs=require('fs');const i=parseInt(process.env.SCAN_INTERVAL_MS||'300000',10);let t=0;try{t=Number(fs.readFileSync(process.env.HEARTBEAT_FILE||'/tmp/noter-heartbeat','utf8'))}catch{};process.exit(Number.isFinite(t)&&Date.now()-t doSignAndSend(peopleApi, noterKeypair, tx, description)); +function queueSubmit(peopleApi, noterKeypair, tx, description, onInBlock) { + const task = submitQueue.then(() => doSignAndSend(peopleApi, noterKeypair, tx, description, onInBlock)); // Never let one caller's failure poison the queue for the next caller's submission. submitQueue = task.catch(() => {}); return task; } -function doSignAndSend(peopleApi, noterKeypair, tx, description) { +function doSignAndSend(peopleApi, noterKeypair, tx, description, onInBlock) { const decodeError = (dispatchError) => { if (dispatchError.isModule) { try { @@ -347,7 +351,7 @@ function doSignAndSend(peopleApi, noterKeypair, tx, description) { } return dispatchError.toString(); }; - return watchSubmission(tx, noterKeypair, { decodeError, timeoutMs: SUBMIT_TIMEOUT }).then( + return watchSubmission(tx, noterKeypair, { decodeError, timeoutMs: SUBMIT_TIMEOUT, onInBlock }).then( (block) => log('INFO', `TX success: ${description}`, { block }), (err) => { log('ERROR', 'TX failed', { description, error: err.message }); @@ -378,6 +382,52 @@ function submitStakingDetails(peopleApi, noterKeypair, updates) { return queueSubmit(peopleApi, noterKeypair, tx, `${updates.length} update(s) for [${addresses}]`); } +/** + * Submit a scan group's updates -- several accounts' -- as one `utility.forceBatch`, and + * return how many of them failed. + * + * One transaction per account made a scan's length the number of accounts with something to + * report times a block: every re-confirmation falls due in the same scan, so on mainnet that + * would be every tracked citizen at once, every few days. `forceBatch` rather than `batchAll` + * because the accounts are unrelated: one refused report must not take the others back with + * it, or the same refusal would block the whole group in every scan. + */ +async function submitScanGroup(peopleApi, noterKeypair, updates) { + if (updates.length === 1) { + await submitStakingDetails(peopleApi, noterKeypair, updates); + return 0; + } + const calls = updates.map(({ address, source, data }) => + peopleApi.tx.stakingScore.receiveStakingDetails( + address, + source, + data.stakedAmount.toString(), + data.nominationsCount, + data.unlockingChunksCount, + ) + ); + let failed = []; + await queueSubmit( + peopleApi, + noterKeypair, + peopleApi.tx.utility.forceBatch(calls), + `${updates.length} update(s) for ${new Set(updates.map(u => u.address)).size} account(s)`, + (events) => { failed = failedBatchItems(events); }, + ); + for (const { index, error } of failed) { + const { address, source } = updates[index]; + let reason = error?.toString?.() ?? String(error); + if (error?.isModule) { + try { + const decoded = peopleApi.registry.findMetaError(error.asModule); + reason = `${decoded.section}.${decoded.name}`; + } catch { /* keep the raw error */ } + } + log('ERROR', `Report refused for ${address.slice(0, 8)}... (${source})`, { error: reason }); + } + return failed.length; +} + /** * finalize_staking_details() is the permissionless step that promotes a matured * PendingStakingDetails entry into CachedStakingDetails (see the pallet's @@ -416,11 +466,15 @@ async function finalizeMaturedPending(peopleApi, noterKeypair) { // PROCESS SINGLE ACCOUNT // ======================================== -async function processAccount(relayApi, assetHubApi, peopleApi, noterKeypair, address) { +/** + * Work out what to report for one account, and report it unless `submit` is false -- a scan + * collects a whole group's updates and sends them together. Returns the updates. + */ +async function processAccount(relayApi, assetHubApi, peopleApi, noterKeypair, address, { submit = true } = {}) { // 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; + if (peopleApi.createType('AccountId', address).eq(noterKeypair.publicKey)) return []; const updates = []; @@ -501,11 +555,11 @@ async function processAccount(relayApi, assetHubApi, peopleApi, noterKeypair, ad } // 5. Submit all updates in a single batch - if (updates.length > 0) { + if (submit && updates.length > 0) { await submitStakingDetails(peopleApi, noterKeypair, updates); } - return updates.length; + return updates; } // ======================================== @@ -533,19 +587,39 @@ async function fullScan(relayApi, assetHubApi, peopleApi, noterKeypair) { const results = await Promise.allSettled( batch.map(([key]) => { const address = key.args[0].toString(); - return processAccount(relayApi, assetHubApi, peopleApi, noterKeypair, address); + return processAccount(relayApi, assetHubApi, peopleApi, noterKeypair, address, { submit: false }); }) ); + const groupUpdates = []; + let read = 0; for (const result of results) { if (result.status === 'fulfilled') { - updatedCount += result.value; + read++; + groupUpdates.push(...result.value); } else { errorCount++; log('ERROR', 'Account processing failed', { error: result.reason?.message }); } } + // Progress means the group was read and, if it had anything to say, said it. A group that + // could not be read, or whose report was refused outright, is not progress: a noter that + // reads everything and lands nothing is as mute as one that reads nothing. + let reported = groupUpdates.length === 0; + if (groupUpdates.length > 0) { + try { + const refused = await submitScanGroup(peopleApi, noterKeypair, groupUpdates); + updatedCount += groupUpdates.length - refused; + errorCount += refused; + reported = refused < groupUpdates.length; + } catch (err) { + errorCount += groupUpdates.length; + log('ERROR', 'Group submission failed', { updates: groupUpdates.length, error: err.message }); + } + } + if (read > 0 && reported) touchHeartbeat(); + // Small delay between batches to be gentle on RPC if (i + BATCH_SIZE < entries.length) { await new Promise(r => setTimeout(r, 500)); @@ -617,7 +691,7 @@ async function main() { 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', { + log('ERROR', 'No scan progress 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, }); @@ -668,13 +742,23 @@ async function main() { // Schedule periodic full scans log('INFO', `Periodic scan scheduled every ${SCAN_INTERVAL / 1000}s`); + // One scan at a time. A scan that outlasts the interval used to have the next one start + // beside it, reading the same accounts before the first one's reports landed and sending + // them again: 124 submissions for 98 accounts on 2026-10-19. + let scanning = false; setInterval(() => { + if (scanning) { + log('INFO', 'Previous scan still running; this interval is skipped'); + return; + } + scanning = true; finalizeMaturedPending(peopleApi, noterKeypair) .then(() => fullScan(relayApi, assetHubApi, peopleApi, noterKeypair)) .then(() => touchHeartbeat()) .catch(err => { log('ERROR', 'Periodic scan failed', { error: err.message }); - }); + }) + .finally(() => { scanning = false; }); }, SCAN_INTERVAL); } diff --git a/noter/submit.js b/noter/submit.js index 897b5ca..7b8c26a 100644 --- a/noter/submit.js +++ b/noter/submit.js @@ -18,9 +18,9 @@ const TERMINAL = ['isDropped', 'isInvalid', 'isUsurped', 'isFinalityTimeout']; * 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. + * answering. `onInBlock`, if given, sees the block's events for this extrinsic first. */ -export function watchSubmission(tx, signer, { decodeError, timeoutMs }) { +export function watchSubmission(tx, signer, { decodeError, timeoutMs, onInBlock }) { return new Promise((resolve, reject) => { let unsub = null; let settled = false; @@ -36,10 +36,14 @@ export function watchSubmission(tx, signer, { decodeError, timeoutMs }) { timeoutMs, ); - tx.signAndSend(signer, ({ status, dispatchError }) => { + tx.signAndSend(signer, ({ status, dispatchError, events }) => { if (status.isInBlock) { - if (dispatchError) settle(reject, new Error(decodeError(dispatchError))); - else settle(resolve, status.asInBlock.toHex()); + if (dispatchError) { + settle(reject, new Error(decodeError(dispatchError))); + } else { + if (!settled && onInBlock) onInBlock(events ?? []); + settle(resolve, status.asInBlock.toHex()); + } return; } const terminal = TERMINAL.find((flag) => status[flag]); @@ -65,3 +69,25 @@ export function isStale({ lastScanAt, startedAt, now, staleAfter }) { const since = Math.max(Number.isFinite(lastScanAt) ? lastScanAt : 0, startedAt); return now - since > staleAfter; } + +/** + * Which calls of a `utility.forceBatch` failed, by their position in the batch. + * + * `forceBatch` runs every call whatever the others do and reports each one, in call order, as + * `ItemCompleted` or `ItemFailed`. A failed item does not fail the transaction, so nothing else + * would ever surface it. + */ +export function failedBatchItems(events) { + const failed = []; + let index = 0; + for (const { event } of events) { + if (event.section !== 'utility') continue; + if (event.method === 'ItemCompleted') { + index++; + } else if (event.method === 'ItemFailed') { + failed.push({ index, error: event.data[0] }); + index++; + } + } + return failed; +} diff --git a/noter/test/submit.test.mjs b/noter/test/submit.test.mjs index 4a372c3..e7d9154 100644 --- a/noter/test/submit.test.mjs +++ b/noter/test/submit.test.mjs @@ -1,7 +1,7 @@ // node --test noter/test import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { watchSubmission, isStale } from '../submit.js'; +import { watchSubmission, isStale, failedBatchItems } 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. @@ -83,3 +83,47 @@ 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); }); + +test("a block's events reach onInBlock before the submission resolves", async () => { + const tx = fakeTx(); + let seen = null; + const p = watchSubmission(tx, {}, { ...opts, onInBlock: (events) => { seen = events; } }); + await tick(); + tx.emit({ isInBlock: true, type: 'InBlock', asInBlock: { toHex: () => '0xabc' } }, { events: ['e'] }); + assert.equal(await p, '0xabc'); + assert.deepEqual(seen, ['e']); +}); + +test('a dispatch error does not reach onInBlock', async () => { + const tx = fakeTx(); + let called = false; + const p = watchSubmission(tx, {}, { ...opts, onInBlock: () => { called = true; } }); + await tick(); + tx.emit({ isInBlock: true, type: 'InBlock', asInBlock: { toHex: () => '0xabc' } }, { dispatchError: 'E' }); + await assert.rejects(p, /decoded E/); + assert.equal(called, false); +}); + +const ev = (section, method, data = []) => ({ event: { section, method, data } }); + +test('a failed batch item is reported by its position among the calls', () => { + const events = [ + ev('stakingScore', 'StakingDetailsReceived'), + ev('utility', 'ItemCompleted'), + ev('utility', 'ItemFailed', ['NoterCannotAttestSelf']), + ev('stakingScore', 'StakingDetailsReceived'), + ev('utility', 'ItemCompleted'), + ev('utility', 'ItemFailed', ['NotAuthorized']), + ev('utility', 'BatchCompletedWithErrors'), + ev('system', 'ExtrinsicSuccess'), + ]; + assert.deepEqual(failedBatchItems(events), [ + { index: 1, error: 'NoterCannotAttestSelf' }, + { index: 3, error: 'NotAuthorized' }, + ]); +}); + +test('a batch whose every item landed reports no failure', () => { + const events = [ev('utility', 'ItemCompleted'), ev('utility', 'ItemCompleted'), ev('utility', 'BatchCompleted')]; + assert.deepEqual(failedBatchItems(events), []); +});