diff --git a/packages/agent/src/dkg-agent-base.ts b/packages/agent/src/dkg-agent-base.ts index 83ab709584..16e7467c64 100644 --- a/packages/agent/src/dkg-agent-base.ts +++ b/packages/agent/src/dkg-agent-base.ts @@ -1664,6 +1664,7 @@ export class DKGAgentBase { /** Drain and checkpoint the inbox before releasing the broader persistence lifetime. */ protected async closeFinalizationRecoveryStore(): Promise { + await this.finalizationHandler?.stopRecoveryWorker(); const store = this.finalizationRuntime.detachRecoveryStore(); await store?.close(); } @@ -1679,6 +1680,7 @@ export class DKGAgentBase { degradedReason: 'not-configured', stateCounts: {}, livePayloadBytes: 0, + dueEntries: 0, }; } const health = await store.health(); @@ -1692,7 +1694,9 @@ export class DKGAgentBase { && canonicalReceiptCapability === 'supported', canonicalReceiptCapability, ...( - canonicalReceiptCapability === 'unsupported' && health.available + canonicalReceiptCapability === 'unsupported' + && health.available + && health.degradedReason === undefined ? { degradedReason: 'canonical-finalization-receipt-unsupported', } diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 34b190fa1b..53563fc756 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -3273,6 +3273,14 @@ export class LifecycleSyncMethods extends DKGAgentBase { }, MESSAGE_OUTBOX_TICK_MS); if (this.messengerOutboxTimer.unref) this.messengerOutboxTimer.unref(); + // The durable finalization inbox is an executable retry queue, not only a + // write-ahead journal. Its lifecycle is independent of chain-cursor + // progress so entries received after a watermark advance are still + // reconsidered. The worker batches SQLite reads but serializes graph work. + if (this.finalizationRuntime.getRecoveryStore()) { + this.getOrCreateFinalizationHandler().startRecoveryWorker(); + } + // Wire V10 Random Sampling prover. Edge nodes no-op. Core nodes with // transient identity/RPC startup failures retry in the background so // one flaky `getIdentityId()` call does not disable proving until the diff --git a/packages/agent/src/dkg-agent.ts b/packages/agent/src/dkg-agent.ts index 97e763d975..d87a7d7375 100644 --- a/packages/agent/src/dkg-agent.ts +++ b/packages/agent/src/dkg-agent.ts @@ -1754,6 +1754,10 @@ export class DKGAgent extends DKGAgentBase { ); } } + // Stop admission and await the active finalization recovery batch while + // chain and graph-store dependencies are still alive. No new retry may + // begin after this boundary. + await this.finalizationHandler?.stopRecoveryWorker(); // OT-RFC-64 Gate 1: unregister the public catalog protocols and drain the // receiver scheduler (awaiting in-flight durable stage writes) while the // router, node, and control-object store are all still live — before diff --git a/packages/agent/src/finalization-handler.ts b/packages/agent/src/finalization-handler.ts index a3f1cc5c7a..043c8f646d 100644 --- a/packages/agent/src/finalization-handler.ts +++ b/packages/agent/src/finalization-handler.ts @@ -79,6 +79,9 @@ import { type FinalizationRecoveryPreparedMaterialization, type FinalizationRecoveryReplayOutcome, } from './finalization-recovery.js'; +import { + FinalizationRecoveryWorker, +} from './finalization-recovery-worker.js'; import type { FinalizationRecoveryEntry, FinalizationRecoveryStore, @@ -381,6 +384,7 @@ export class FinalizationHandler { private readonly negativeSnapshotMemo = new Map(); /** Equivalent finalization/reconcile reads share one promise until it settles. */ private readonly scanSingleFlights = new Map>(); + private readonly recoveryWorker: FinalizationRecoveryWorker; constructor( store: TripleStore, @@ -448,6 +452,21 @@ export class FinalizationHandler { }, materializer, ); + this.recoveryWorker = new FinalizationRecoveryWorker( + (limit) => this.recovery.processDueBatch(limit), + { + info: (message) => this.log.info(createOperationContext('system'), message), + warn: (message) => this.log.warn(createOperationContext('system'), message), + }, + ); + } + + startRecoveryWorker(): void { + this.recoveryWorker.start(); + } + + stopRecoveryWorker(): Promise { + return this.recoveryWorker.stop(); } async handleFinalizationMessage( diff --git a/packages/agent/src/finalization-recovery-sqlite-store.ts b/packages/agent/src/finalization-recovery-sqlite-store.ts index c944415c8f..6963db5e5a 100644 --- a/packages/agent/src/finalization-recovery-sqlite-store.ts +++ b/packages/agent/src/finalization-recovery-sqlite-store.ts @@ -34,6 +34,17 @@ export type { SqliteFinalizationRecoveryStoreOptions, } from './finalization-recovery-sqlite-policy.js'; +const DUE_FINALIZATION_SQL_PREDICATE = ` + ( + state IN ('RECEIVED','VERIFIED','REORGED') + OR ( + state = 'SETTLED' + AND (publisher_upgrade_pending = 1 OR next_attempt_at IS NOT NULL) + ) + ) + AND (next_attempt_at IS NULL OR next_attempt_at <= ?) +`; + function sameFinalizationRecoveryIdentity( existing: FinalizationRecoveryEntry, input: FinalizationRecoveryReceiveInput, @@ -80,6 +91,16 @@ export class SqliteFinalizationRecoveryStore implements FinalizationRecoveryStor return this.#closed; } + async get(key: string): Promise { + if (this.#closed || this.#closing) return undefined; + await this.#mutationTail; + if (this.#closed) return undefined; + const row = this.database.prepare( + 'SELECT * FROM finalization_inbox_v1 WHERE key = ?', + ).get(key); + return row ? finalizationRecoveryRowToEntry(row) : undefined; + } + receive(input: FinalizationRecoveryReceiveInput): Promise { if (this.#closed || this.#closing) return Promise.resolve({ status: 'closed' }); return this.mutate(() => { @@ -422,6 +443,25 @@ export class SqliteFinalizationRecoveryStore implements FinalizationRecoveryStor return entry.nextAttemptAt === undefined || entry.nextAttemptAt <= this.#policy.now(); } + async listDue(limit: number): Promise { + if (this.#closed || this.#closing) return []; + if (!Number.isFinite(limit)) return []; + const boundedLimit = Math.min( + this.#policy.maxEntries, + Math.max(0, Math.trunc(limit)), + ); + if (boundedLimit === 0) return []; + await this.#mutationTail; + if (this.#closed) return []; + const now = this.#policy.now(); + return this.database.prepare(` + SELECT * FROM finalization_inbox_v1 + WHERE ${DUE_FINALIZATION_SQL_PREDICATE} + ORDER BY COALESCE(next_attempt_at, updated_at), updated_at, key + LIMIT ? + `).all(now, boundedLimit).map(finalizationRecoveryRowToEntry); + } + async listForKnowledgeAsset(input: { chainId: string; contextGraphId: string; @@ -497,17 +537,24 @@ export class SqliteFinalizationRecoveryStore implements FinalizationRecoveryStor return this.mutate(() => { if (this.#closed) return; const now = this.#policy.now(); + const nextAttemptAt = retryDelayMs === undefined ? null : now + retryDelayMs; this.database.prepare(` UPDATE finalization_inbox_v1 SET attempt_count = attempt_count + 1, last_error = ?, - next_attempt_at = ?, + next_attempt_at = CASE + WHEN ? IS NULL THEN next_attempt_at + WHEN next_attempt_at IS NULL THEN ? + ELSE MAX(next_attempt_at, ?) + END, updated_at = ? WHERE key = ? AND generation = ? AND state IN ('RECEIVED','VERIFIED','REORGED','SETTLED') `).run( lastError ?? null, - retryDelayMs === undefined ? null : now + retryDelayMs, + nextAttemptAt, + nextAttemptAt, + nextAttemptAt, now, key, generation, @@ -523,17 +570,31 @@ export class SqliteFinalizationRecoveryStore implements FinalizationRecoveryStor degradedReason: 'closing', stateCounts: {}, livePayloadBytes: 0, + dueEntries: 0, }; } await this.#mutationTail; if (this.#closed) { - return { available: false, closed: true, degradedReason: 'closed', stateCounts: {}, livePayloadBytes: 0 }; + return { + available: false, + closed: true, + degradedReason: 'closed', + stateCounts: {}, + livePayloadBytes: 0, + dueEntries: 0, + }; } const counts = this.database.prepare( 'SELECT state, COUNT(*) AS count FROM finalization_inbox_v1 GROUP BY state', ).all(); const stateCounts: Partial> = {}; for (const row of counts) stateCounts[String(row.state) as FinalizationRecoveryState] = Number(row.count); + const now = this.#policy.now(); + const due = this.database.prepare(` + SELECT COUNT(*) AS count, MIN(created_at) AS oldest + FROM finalization_inbox_v1 + WHERE ${DUE_FINALIZATION_SQL_PREDICATE} + `).get(now) as { count: number | bigint; oldest: number | bigint | null }; const capacity = readFinalizationRecoveryCapacity(this.database, this.#policy); return { available: true, @@ -542,9 +603,13 @@ export class SqliteFinalizationRecoveryStore implements FinalizationRecoveryStor ...(capacity.capacityExhausted ? { degradedReason: 'capacity-exhausted' } : {}), stateCounts, livePayloadBytes: capacity.livePayloadBytes, + dueEntries: Number(due.count), + ...(due.oldest === null + ? {} + : { oldestDueAgeMs: Math.max(0, now - Number(due.oldest)) }), ...(capacity.oldest === undefined ? {} - : { oldestPendingAgeMs: Math.max(0, this.#policy.now() - capacity.oldest) }), + : { oldestPendingAgeMs: Math.max(0, now - capacity.oldest) }), }; } diff --git a/packages/agent/src/finalization-recovery-store.ts b/packages/agent/src/finalization-recovery-store.ts index 977e4ad75e..e05f329b24 100644 --- a/packages/agent/src/finalization-recovery-store.ts +++ b/packages/agent/src/finalization-recovery-store.ts @@ -80,11 +80,15 @@ export interface FinalizationRecoveryHealth { degradedReason?: string; stateCounts: Partial>; livePayloadBytes: number; + dueEntries: number; + oldestDueAgeMs?: number; oldestPendingAgeMs?: number; } export interface FinalizationRecoveryStore { readonly closed: boolean; + /** Reloads one entry after waiting on an in-process serialization boundary. */ + get(key: string): Promise; receive(input: FinalizationRecoveryReceiveInput): Promise; recordTrustedPublisher( key: string, @@ -113,6 +117,12 @@ export interface FinalizationRecoveryStore { clearSettledRetry(key: string, generation: number): Promise; rejectSettled(key: string, generation: number, lastError: string): Promise; isAttemptDue(entry: FinalizationRecoveryEntry): boolean; + /** + * Returns a bounded, oldest-first snapshot of entries whose persisted retry + * gate is open. Callers must still rely on generation-checked transitions: + * live gossip or reconciliation may update an entry after this read. + */ + listDue(limit: number): Promise; listForKnowledgeAsset(input: { chainId: string; contextGraphId: string; diff --git a/packages/agent/src/finalization-recovery-worker.ts b/packages/agent/src/finalization-recovery-worker.ts new file mode 100644 index 0000000000..9498b99ecf --- /dev/null +++ b/packages/agent/src/finalization-recovery-worker.ts @@ -0,0 +1,100 @@ +export const FINALIZATION_RECOVERY_WORKER_BATCH_SIZE = 16; +export const FINALIZATION_RECOVERY_WORKER_POLL_INTERVAL_MS = 5_000; + +interface FinalizationRecoveryWorkerLog { + info(message: string): void; + warn(message: string): void; +} + +interface FinalizationRecoveryWorkerOptions { + batchSize?: number; + pollIntervalMs?: number; +} + +/** + * Lifecycle-owned asynchronous retry loop for the durable finalization inbox. + * + * SQLite reads are batched, while graph materialization remains serial inside + * `processDueBatch`. This is an I/O scheduler on the Node.js event loop, not a + * worker thread: it never blocks startup and never introduces concurrent + * Blazegraph finalization writes. + */ +export class FinalizationRecoveryWorker { + readonly #batchSize: number; + readonly #pollIntervalMs: number; + #running = false; + #timer: ReturnType | undefined; + #inFlight: Promise | undefined; + + constructor( + private readonly processDueBatch: (limit: number) => Promise, + private readonly log: FinalizationRecoveryWorkerLog, + options: FinalizationRecoveryWorkerOptions = {}, + ) { + this.#batchSize = Math.max( + 1, + Math.trunc(options.batchSize ?? FINALIZATION_RECOVERY_WORKER_BATCH_SIZE), + ); + this.#pollIntervalMs = Math.max( + 1, + Math.trunc( + options.pollIntervalMs ?? FINALIZATION_RECOVERY_WORKER_POLL_INTERVAL_MS, + ), + ); + } + + get running(): boolean { + return this.#running; + } + + start(): void { + if (this.#running) return; + this.#running = true; + this.schedule(0); + } + + async stop(): Promise { + this.#running = false; + if (this.#timer) { + clearTimeout(this.#timer); + this.#timer = undefined; + } + await this.#inFlight?.catch(() => undefined); + } + + private schedule(delayMs: number): void { + if (!this.#running || this.#timer) return; + this.#timer = setTimeout(() => { + this.#timer = undefined; + if (!this.#running) return; + const run = this.runBatch(); + this.#inFlight = run; + void run.finally(() => { + if (this.#inFlight === run) this.#inFlight = undefined; + }); + }, delayMs); + this.#timer.unref?.(); + } + + private async runBatch(): Promise { + let selected = 0; + try { + selected = await this.processDueBatch(this.#batchSize); + if (selected > 0) { + this.log.info( + `Finalization recovery worker inspected ${selected} due inbox entr` + + `${selected === 1 ? 'y' : 'ies'}`, + ); + } + } catch (error) { + this.log.warn( + `Finalization recovery worker batch failed: ` + + `${error instanceof Error ? error.message : String(error)}`, + ); + } finally { + if (this.#running) { + this.schedule(selected >= this.#batchSize ? 0 : this.#pollIntervalMs); + } + } + } +} diff --git a/packages/agent/src/finalization-recovery.ts b/packages/agent/src/finalization-recovery.ts index e944353acf..560547aeb3 100644 --- a/packages/agent/src/finalization-recovery.ts +++ b/packages/agent/src/finalization-recovery.ts @@ -6,6 +6,7 @@ import type { } from '@origintrail-official/dkg-chain'; import { decodeFinalizationMessage, + getMetrics, GRAPH_KA_CONTENT_SCOPE_VERSION, } from '@origintrail-official/dkg-core'; import { ethers } from 'ethers'; @@ -153,6 +154,29 @@ interface ResolveCanonicalReceiptOptions { const SETTLED_RECEIPT_RETRY_BASE_MS = 1_000; const SETTLED_RECEIPT_RETRY_MAX_MS = 60_000; const SETTLED_NOT_FOUND_RETRY_LIMIT = 5; +const DEFERRED_RETRY_BASE_MS = 1_000; +const DEFERRED_RETRY_MAX_MS = 60_000; +/** + * At the maximum retry delay this is approximately seven days of autonomous + * worker attempts. The wall-clock window below must also elapse so duplicate + * gossip and reconciliation cannot burn the budget early. + */ +export const FINALIZATION_RECOVERY_LIVE_RETRY_LIMIT = 7 * 24 * 60; +export const FINALIZATION_RECOVERY_LIVE_RETRY_WINDOW_MS = 7 * 24 * 60 * 60 * 1_000; + +export interface FinalizationRecoveryOptions { + liveRetryLimit?: number; + liveRetryWindowMs?: number; + now?: () => number; +} + +function deferredRetryDelayMs(attemptCount: number): number { + const exponent = Math.min(30, Math.max(0, attemptCount)); + return Math.min( + DEFERRED_RETRY_MAX_MS, + DEFERRED_RETRY_BASE_MS * (2 ** exponent), + ); +} export function finalizationRecoveryEntryKey(input: { chainId: string; @@ -176,14 +200,19 @@ export class FinalizationRecovery< string, Promise >(); + private readonly entryLockTails = new Map>(); private readonly store: FinalizationRecoveryStore | undefined; private readonly storeSource: FinalizationRecoveryStoreSource | undefined; + private readonly liveRetryLimit: number; + private readonly liveRetryWindowMs: number; + private readonly now: () => number; constructor( store: FinalizationRecoveryStore | FinalizationRecoveryStoreSource | undefined, private readonly chain: ChainAdapter | undefined, private readonly log: FinalizationRecoveryLog, private readonly materializer: FinalizationRecoveryMaterializer, + options: FinalizationRecoveryOptions = {}, ) { if (store && 'getRecoveryStore' in store) { this.storeSource = store; @@ -192,6 +221,17 @@ export class FinalizationRecovery< this.store = store; this.storeSource = undefined; } + this.liveRetryLimit = Math.max( + 1, + Math.trunc(options.liveRetryLimit ?? FINALIZATION_RECOVERY_LIVE_RETRY_LIMIT), + ); + this.liveRetryWindowMs = Math.max( + 1, + Math.trunc( + options.liveRetryWindowMs ?? FINALIZATION_RECOVERY_LIVE_RETRY_WINDOW_MS, + ), + ); + this.now = options.now ?? Date.now; } private getStore(): FinalizationRecoveryStore | undefined { @@ -251,39 +291,243 @@ export class FinalizationRecovery< || this.chain.chainId === 'none' || !this.chain.resolveCanonicalFinalizationReceipt ) return false; - let entry = await this.receive(input); - // A configured inbox fails closed: capacity, conflict, corruption, and - // write failures leave Oxigraph untouched. - if (!entry) return true; - if (entry.state === 'SETTLED') { - entry = await this.revalidateSettled(input, entry); + const key = finalizationRecoveryEntryKey({ + chainId: this.chain.chainId, + contextGraphId: input.contextGraphId, + ual: input.candidate.scope.ual, + txHash: input.candidate.msg.txHash, + }); + return this.withEntryLock(key, async () => { + let entry = await this.receive(input); + // A configured inbox fails closed: capacity, conflict, corruption, and + // write failures leave Oxigraph untouched. if (!entry) return true; - } + if (entry.state === 'SETTLED') { + entry = await this.revalidateSettled(input, entry); + if (!entry) return true; + } + // Terminal rows remain audited and inert until retention removes them. + // In particular, duplicate gossip must not revive an entry that exhausted + // the autonomous retry budget. + if (!this.isLiveEntry(entry)) return true; + + for (let attempt = 0; attempt < 2; attempt += 1) { + try { + const outcome = await this.materialize(input, { kind: 'recovery', entry }); + if (outcome === 'deferred') { + await this.recordDeferred( + entry, + 'finalization processing deferred', + ); + } else { + await this.settleEntry(entry, outcome); + } + return true; + } catch (error) { + if (!this.materializer.isRetryableError(error)) throw error; + if (attempt === 0) { + await new Promise((resolve) => setTimeout(resolve, 50)); + continue; + } + this.log.warn( + `Finalization recovery materialization remained busy for ${entry.ual}; ` + + 'keeping inbox entry', + ); + await this.recordDeferred( + entry, + 'store scheduler remained busy', + ); + return true; + } + } + return true; + }); + } - for (let attempt = 0; attempt < 2; attempt += 1) { + /** + * Replays a bounded due snapshot independently of chain-cursor progress. + * Entries are processed serially so a recovered backlog cannot recreate the + * store pressure that caused it. + */ + async processDueBatch(limit: number): Promise { + const store = this.getStore(); + if (!store) return 0; + try { + if (!this.chain || this.chain.chainId === 'none') return 0; + let entries: FinalizationRecoveryEntry[]; try { - const outcome = await this.materialize(input, { kind: 'recovery', entry }); - if (outcome === 'deferred') { - await this.recordDeferred(entry, 'finalization processing deferred'); - } else { - await this.settleEntry(entry, outcome); - } - return true; + entries = await store.listDue(limit); } catch (error) { - if (!this.materializer.isRetryableError(error)) throw error; - if (attempt === 0) { - await new Promise((resolve) => setTimeout(resolve, 50)); - continue; + this.log.warn( + `Finalization recovery due-inbox read failed: ` + + `${error instanceof Error ? error.message : String(error)}`, + ); + return 0; + } + for (const entry of entries) { + let outcome: FinalizationRecoveryReplayOutcome; + try { + outcome = await this.replayDueEntry(entry); + } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + this.log.warn( + `Finalization recovery due-entry replay failed for ${entry.ual}: ${reason}`, + ); + await this.recordDeferred( + entry, + `background replay failed: ${reason}`, + deferredRetryDelayMs(entry.attemptCount), + ); + outcome = 'retry-pending'; } + getMetrics().finalizationRecoveryAttemptsTotal?.add(1, { outcome }); + } + return entries.length; + } finally { + await this.recordDueMetrics(store); + } + } + + private async recordDueMetrics(store: FinalizationRecoveryStore): Promise { + try { + const health = await store.health(); + const metrics = getMetrics(); + metrics.finalizationRecoveryDueEntries?.record(health.dueEntries); + metrics.finalizationRecoveryOldestDueAgeMs?.record(health.oldestDueAgeMs ?? 0); + } catch (error) { + this.log.warn( + `Finalization recovery metrics snapshot failed: ` + + `${error instanceof Error ? error.message : String(error)}`, + ); + } + } + + async replayDueEntry( + snapshot: FinalizationRecoveryEntry, + ): Promise { + return this.withEntryLock( + snapshot.key, + () => this.replayDueEntryLocked(snapshot), + ); + } + + private async replayDueEntryLocked( + snapshot: FinalizationRecoveryEntry, + ): Promise { + const store = this.getStore(); + if (!store) return 'none'; + const entry = await store.get(snapshot.key); + if (!entry) return 'none'; + if (!store.isAttemptDue(entry)) return 'retry-pending'; + const liveRetryAgeMs = Math.max(0, this.now() - entry.createdAt); + if ( + this.isLiveEntry(entry) + && entry.attemptCount >= this.liveRetryLimit + && liveRetryAgeMs >= this.liveRetryWindowMs + ) { + const reason = 'autonomous retry budget exhausted after ' + + `${entry.attemptCount} attempts over ${liveRetryAgeMs}ms`; + const rejected = await this.transition(entry, 'REJECTED', reason); + if (rejected) { this.log.warn( - `Finalization recovery materialization remained busy for ${entry.ual}; ` - + 'keeping inbox entry', + `Finalization recovery rejected exhausted inbox entry for ${entry.ual}: ${reason}`, ); - await this.recordDeferred(entry, 'store scheduler remained busy'); - return true; + } + return 'invalidated'; + } + if ( + !this.chain + || this.chain.chainId === 'none' + || this.chain.chainId !== entry.chainId + || !this.chain.getKAContextGraphId + ) { + await this.recordDeferred( + entry, + 'background replay lacks the matching chain binding capability', + deferredRetryDelayMs(entry.attemptCount), + ); + return 'retry-pending'; + } + + try { + const boundContextGraphId = await this.chain.getKAContextGraphId( + BigInt(entry.kaId), + ); + if ( + boundContextGraphId === null + || boundContextGraphId === undefined + || BigInt(boundContextGraphId) <= 0n + ) { + await this.recordDeferred( + entry, + 'background replay chain binding is not available yet', + deferredRetryDelayMs(entry.attemptCount), + ); + return 'retry-pending'; + } + const replayInput: FinalizationRecoveryReplayInput = { + chainId: entry.chainId, + contextGraphId: entry.contextGraphId, + onChainCgId: BigInt(boundContextGraphId).toString(), + ual: entry.ual, + merkleRoot: entry.merkleRoot, + kaId: entry.kaId, + }; + const matches = await this.matchingEntries(replayInput); + const matchingEntry = matches.find((candidate) => candidate.key === entry.key); + const outcome = matchingEntry + ? await this.replayEntry( + matchingEntry, + replayInput, + deferredRetryDelayMs(matchingEntry.attemptCount), + ) + : 'none'; + if (outcome !== 'none') return outcome; + await this.recordDeferred( + entry, + 'background replay found no canonical finalization match', + deferredRetryDelayMs(entry.attemptCount), + ); + return 'retry-pending'; + } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + this.log.warn( + `Finalization recovery background replay failed for ${entry.ual}: ${reason}`, + ); + await this.recordDeferred( + entry, + `background replay failed: ${reason}`, + deferredRetryDelayMs(entry.attemptCount), + ); + return 'retry-pending'; + } + } + + private isLiveEntry(entry: FinalizationRecoveryEntry): boolean { + return entry.state === 'RECEIVED' + || entry.state === 'VERIFIED' + || entry.state === 'REORGED'; + } + + private async withEntryLock( + key: string, + operation: () => Promise, + ): Promise { + const previous = this.entryLockTails.get(key) ?? Promise.resolve(); + let release: (() => void) | undefined; + const current = new Promise((resolve) => { + release = resolve; + }); + this.entryLockTails.set(key, current); + await previous.catch(() => undefined); + try { + return await operation(); + } finally { + release?.(); + if (this.entryLockTails.get(key) === current) { + this.entryLockTails.delete(key); } } - return true; } /** Processes the compatibility path when no durable inbox can be used. */ @@ -603,11 +847,20 @@ export class FinalizationRecovery< return this.transition(entry, 'SETTLED'); } - async recordDeferred(entry: FinalizationRecoveryEntry, reason: string): Promise { + async recordDeferred( + entry: FinalizationRecoveryEntry, + reason: string, + retryDelayMs?: number, + ): Promise { const store = this.getStore(); if (!store) return; try { - await store.recordAttempt(entry.key, entry.generation, reason); + await store.recordAttempt( + entry.key, + entry.generation, + reason, + retryDelayMs, + ); } catch (error) { this.log.warn( `Finalization recovery attempt update failed for ${entry.ual}: ` @@ -1056,13 +1309,27 @@ export class FinalizationRecovery< async replayMatching( input: FinalizationRecoveryReplayInput, + options: { persistDeferredBackoff?: boolean } = {}, ): Promise { const entries = await this.matchingEntries(input); let outcome: FinalizationRecoveryReplayOutcome = 'none'; for (const entry of entries) { const replayKey = this.replaySingleFlightKey(entry, input); const existing = this.replaySingleFlights.get(replayKey); - const replay = existing ?? this.replayEntry(entry, input, replayKey); + const replay = existing ?? this.withEntryLock( + entry.key, + () => this.replayEntry( + entry, + input, + options.persistDeferredBackoff + ? deferredRetryDelayMs(entry.attemptCount) + : undefined, + ), + ).finally(() => { + if (this.replaySingleFlights.get(replayKey) === replay) { + this.replaySingleFlights.delete(replayKey); + } + }); if (!existing) this.replaySingleFlights.set(replayKey, replay); const entryOutcome = await replay; if (entryOutcome === 'recovered') outcome = 'recovered'; @@ -1099,6 +1366,7 @@ export class FinalizationRecovery< degradedReason: 'not-configured', stateCounts: {}, livePayloadBytes: 0, + dueEntries: 0, }; } return store.health(); @@ -1126,6 +1394,7 @@ export class FinalizationRecovery< entry: FinalizationRecoveryEntry, candidate: ParsedGraphScopedFinalization, input: FinalizationRecoveryReplayInput, + deferredRetryDelay?: number, ): Promise { const evidence = entry.verifiedEvidence; if ( @@ -1159,12 +1428,17 @@ export class FinalizationRecovery< sameCanonicalFinalizationPlacement(canonical.receipt, evidence) ? 'trusted publisher access semantics arrived after settlement' : 'trusted publisher access semantics and canonical placement changed after settlement', + deferredRetryDelay, ); if (!matchesReplayTarget) return 'none'; return recovered ? 'recovered' : 'retry-pending'; } if (!sameCanonicalFinalizationPlacement(canonical.receipt, evidence)) { - const recovered = await this.recoverSettledReorg(entry, candidate); + const recovered = await this.recoverSettledReorg( + entry, + candidate, + deferredRetryDelay, + ); if (!matchesReplayTarget) return 'none'; return recovered ? 'recovered' : 'retry-pending'; } @@ -1188,11 +1462,16 @@ export class FinalizationRecovery< entry, candidate, 'trusted publisher access semantics and canonical placement changed after settlement', + deferredRetryDelay, ); if (!matchesReplayTarget) return 'none'; return recovered ? 'recovered' : 'retry-pending'; } - const recovered = await this.recoverSettledReorg(entry, candidate); + const recovered = await this.recoverSettledReorg( + entry, + candidate, + deferredRetryDelay, + ); if (!matchesReplayTarget) return 'none'; return recovered ? 'recovered' : 'retry-pending'; } @@ -1237,6 +1516,7 @@ export class FinalizationRecovery< entry: FinalizationRecoveryEntry, candidate: ParsedGraphScopedFinalization, reason: string, + deferredRetryDelay?: number, ): Promise { const publisherPeerId = entry.trustedPublisherPeerId; if ( @@ -1265,7 +1545,11 @@ export class FinalizationRecovery< { kind: 'recovery', entry: rearmed }, ); if (outcome === 'deferred') { - await this.recordDeferred(rearmed, 'settled publisher upgrade recovery deferred'); + await this.recordDeferred( + rearmed, + 'settled publisher upgrade recovery deferred', + deferredRetryDelay, + ); return false; } return this.settleEntry(rearmed, outcome); @@ -1295,6 +1579,7 @@ export class FinalizationRecovery< private async recoverSettledReorg( entry: FinalizationRecoveryEntry, candidate: ParsedGraphScopedFinalization, + deferredRetryDelay?: number, ): Promise { if (!await this.markReorged( entry, @@ -1318,117 +1603,151 @@ export class FinalizationRecovery< { kind: 'recovery', entry: reorged }, ); if (outcome === 'deferred') { - await this.recordDeferred(reorged, 'settled reorg recovery deferred'); + await this.recordDeferred( + reorged, + 'settled reorg recovery deferred', + deferredRetryDelay, + ); return false; } return this.settleEntry(reorged, outcome); } - private replayEntry( - entry: FinalizationRecoveryEntry, + private async replayEntry( + snapshot: FinalizationRecoveryEntry, input: FinalizationRecoveryReplayInput, - replayKey: string, + deferredRetryDelay?: number, ): Promise { - const replay = (async () => { - try { - const candidate = this.decodeEntry(entry); - if (!candidate) return 'none' as const; - if (entry.state === 'SETTLED') { - return this.replaySettled(entry, candidate, input); - } + const store = this.getStore(); + if (!store) return 'none'; + const entry = await store.get(snapshot.key); + if (!entry || (!this.isLiveEntry(entry) && entry.state !== 'SETTLED')) { + return 'none'; + } + try { + // Only autonomous replay observes its persisted deadline. Chain + // reconciliation remains an immediate, authoritative recovery trigger. + if ( + deferredRetryDelay !== undefined + && entry.state !== 'SETTLED' + && !store.isAttemptDue(entry) + ) return 'retry-pending' as const; + const candidate = this.decodeEntry(entry); + if (!candidate) return 'none' as const; + if (entry.state === 'SETTLED') { + return this.replaySettled( + entry, + candidate, + input, + deferredRetryDelay, + ); + } - let outcome: FinalizationRecoveryApplyOutcome; - if (entry.state === 'VERIFIED' && entry.verifiedEvidence) { - const evidence = entry.verifiedEvidence; - const evidenceMatchesEnvelope = entry.generation > 0 - ? VerifiedGraphScopedFinalizationEvidenceCodec.matchesImmutableEnvelope( - evidence, - candidate, - entry, - ) - : VerifiedGraphScopedFinalizationEvidenceCodec.matchesEnvelope( - evidence, - candidate, - entry, - ); - if (!evidenceMatchesEnvelope) { - this.log.warn( - `Finalization recovery evidence does not match its envelope for ${entry.ual}`, + let outcome: FinalizationRecoveryApplyOutcome; + if (entry.state === 'VERIFIED' && entry.verifiedEvidence) { + const evidence = entry.verifiedEvidence; + const evidenceMatchesEnvelope = entry.generation > 0 + ? VerifiedGraphScopedFinalizationEvidenceCodec.matchesImmutableEnvelope( + evidence, + candidate, + entry, + ) + : VerifiedGraphScopedFinalizationEvidenceCodec.matchesEnvelope( + evidence, + candidate, + entry, ); - await this.rejectEntry(entry, 'verified evidence does not match immutable envelope'); - return 'none' as const; - } - const receiptStatus = await this.verifyPersistedReceipt( - candidate, - evidence, - entry.generation > 0, + if (!evidenceMatchesEnvelope) { + this.log.warn( + `Finalization recovery evidence does not match its envelope for ${entry.ual}`, ); - if (receiptStatus !== 'confirmed') { - if (receiptStatus === 'reorged') { - await this.markReorged( - entry, - 'persisted receipt disagrees with canonical chain truth', - ); - } else if (receiptStatus === 'rejected') { - await this.rejectEntry( - entry, - 'persisted transaction failed or contains no finalization event', - ); - } else if (receiptStatus === 'unsupported') { - await this.markUnsupported(entry); - } else { - await this.recordDeferred(entry, `persisted receipt is ${receiptStatus}`); - } - this.log.info( - `Finalization recovery receipt is ${receiptStatus} for ${entry.ual}`, + await this.rejectEntry(entry, 'verified evidence does not match immutable envelope'); + return 'none' as const; + } + const receiptStatus = await this.verifyPersistedReceipt( + candidate, + evidence, + entry.generation > 0, + ); + if (receiptStatus !== 'confirmed') { + if (receiptStatus === 'reorged') { + await this.markReorged( + entry, + 'persisted receipt disagrees with canonical chain truth', + ); + } else if (receiptStatus === 'rejected') { + await this.rejectEntry( + entry, + 'persisted transaction failed or contains no finalization event', + ); + } else if (receiptStatus === 'unsupported') { + await this.markUnsupported(entry); + } else { + await this.recordDeferred( + entry, + `persisted receipt is ${receiptStatus}`, + deferredRetryDelay, ); - return 'none' as const; } - const replayOutcome = await this.materializer.replayVerified({ - replay: input, - entry, - candidate, - evidence, - }); - outcome = replayOutcome === 'promoted' - ? 'applied' - : replayOutcome === 'already-confirmed' || replayOutcome === 'stale-target' - ? 'already-confirmed' - : 'deferred'; - } else { - const recoverySourcePeerId = entry.trustedPublisherPeerId ?? entry.sourcePeerId; - outcome = await this.materialize( - { - rawMessage: entry.rawMessage, - contextGraphId: entry.contextGraphId, - ...(recoverySourcePeerId ? { sourcePeerId: recoverySourcePeerId } : {}), - candidate, - }, - { kind: 'recovery', entry }, + this.log.info( + `Finalization recovery receipt is ${receiptStatus} for ${entry.ual}`, ); + return deferredRetryDelay !== undefined + && (receiptStatus === 'pending' || receiptStatus === 'not-found') + ? 'retry-pending' as const + : 'none' as const; } - - if (outcome === 'deferred') { - await this.recordDeferred(entry, 'replay processing deferred'); - return 'none' as const; - } - const settled = await this.settleEntry(entry, outcome); - return settled ? 'recovered' as const : 'none' as const; - } catch (error) { - if (!this.materializer.isRetryableError(error)) throw error; - this.log.info( - `Finalization recovery materialization remains busy for ${entry.ual}; ` - + 'keeping inbox entry', + const replayOutcome = await this.materializer.replayVerified({ + replay: input, + entry, + candidate, + evidence, + }); + outcome = replayOutcome === 'promoted' + ? 'applied' + : replayOutcome === 'already-confirmed' || replayOutcome === 'stale-target' + ? 'already-confirmed' + : 'deferred'; + } else { + const recoverySourcePeerId = entry.trustedPublisherPeerId ?? entry.sourcePeerId; + outcome = await this.materialize( + { + rawMessage: entry.rawMessage, + contextGraphId: entry.contextGraphId, + ...(recoverySourcePeerId ? { sourcePeerId: recoverySourcePeerId } : {}), + candidate, + }, + { kind: 'recovery', entry }, ); - await this.recordDeferred(entry, 'replay store scheduler remained busy'); - return 'none' as const; } - })().finally(() => { - if (this.replaySingleFlights.get(replayKey) === replay) { - this.replaySingleFlights.delete(replayKey); + + if (outcome === 'deferred') { + await this.recordDeferred( + entry, + 'replay processing deferred', + deferredRetryDelay, + ); + return deferredRetryDelay === undefined + ? 'none' as const + : 'retry-pending' as const; } - }); - return replay; + const settled = await this.settleEntry(entry, outcome); + return settled ? 'recovered' as const : 'none' as const; + } catch (error) { + if (!this.materializer.isRetryableError(error)) throw error; + this.log.info( + `Finalization recovery materialization remains busy for ${entry.ual}; ` + + 'keeping inbox entry', + ); + await this.recordDeferred( + entry, + 'replay store scheduler remained busy', + deferredRetryDelay, + ); + return deferredRetryDelay === undefined + ? 'none' as const + : 'retry-pending' as const; + } } private decodeEntry( diff --git a/packages/agent/test/finalization-recovery-sqlite-store.test.ts b/packages/agent/test/finalization-recovery-sqlite-store.test.ts index 733c53c5ee..4eae2e85f8 100644 --- a/packages/agent/test/finalization-recovery-sqlite-store.test.ts +++ b/packages/agent/test/finalization-recovery-sqlite-store.test.ts @@ -120,6 +120,100 @@ setInterval(() => {}, 60_000); } describe('SQLite finalization recovery store', () => { + it('lists only due live work in bounded oldest-first batches', async () => { + const directory = await temporaryDirectory(); + try { + let now = 1_000; + const store = await openSqliteFinalizationRecoveryStore(directory, { + now: () => now, + }); + await store.receive(received({ key: 'entry-1' })); + await store.receive(received({ key: 'entry-2' })); + await store.receive(received({ key: 'entry-3' })); + await store.recordAttempt('entry-1', 0, 'busy', 1_000); + await store.markVerified('entry-3', 0, evidence()); + await store.transition('entry-3', 0, 'SETTLED'); + + await expect(store.listDue(16)).resolves.toMatchObject([ + { key: 'entry-2', state: 'RECEIVED' }, + ]); + + now = 2_000; + await expect(store.listDue(16)).resolves.toMatchObject([ + { key: 'entry-2', state: 'RECEIVED', attemptCount: 0 }, + { key: 'entry-1', state: 'RECEIVED', attemptCount: 1 }, + ]); + await expect(store.listDue(0)).resolves.toEqual([]); + await store.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it('lists a SETTLED receipt retry only after its persisted deadline', async () => { + const directory = await temporaryDirectory(); + try { + let now = 1_000; + const store = await openSqliteFinalizationRecoveryStore(directory, { + now: () => now, + }); + await store.receive(received()); + await store.markVerified('entry-1', 0, evidence()); + await store.transition('entry-1', 0, 'SETTLED'); + await store.recordAttempt('entry-1', 0, 'receipt pending', 1_000); + + await expect(store.listDue(16)).resolves.toEqual([]); + now = 2_000; + await expect(store.listDue(16)).resolves.toMatchObject([ + { key: 'entry-1', state: 'SETTLED', lastError: 'receipt pending' }, + ]); + await store.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it('never narrows a persisted retry deadline and reports due backlog health', async () => { + const directory = await temporaryDirectory(); + try { + let now = 1_000; + const store = await openSqliteFinalizationRecoveryStore(directory, { + now: () => now, + }); + await store.receive(received()); + await expect(store.health()).resolves.toMatchObject({ + dueEntries: 1, + oldestDueAgeMs: 0, + }); + + await store.recordAttempt('entry-1', 0, 'long backoff', 10_000); + expect(await store.get('entry-1')).toMatchObject({ + nextAttemptAt: 11_000, + }); + await expect(store.health()).resolves.toMatchObject({ dueEntries: 0 }); + + now = 1_500; + await store.recordAttempt('entry-1', 0, 'duplicate without delay'); + await store.recordAttempt('entry-1', 0, 'shorter backoff', 100); + expect(await store.get('entry-1')).toMatchObject({ + nextAttemptAt: 11_000, + }); + + await store.recordAttempt('entry-1', 0, 'longer backoff', 20_000); + expect(await store.get('entry-1')).toMatchObject({ + nextAttemptAt: 21_500, + }); + now = 21_500; + await expect(store.health()).resolves.toMatchObject({ + dueEntries: 1, + oldestDueAgeMs: 20_500, + }); + await store.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + it('durably transitions RECEIVED to VERIFIED to SETTLED across reopen', async () => { const directory = await temporaryDirectory(); try { diff --git a/packages/agent/test/finalization-recovery-worker.test.ts b/packages/agent/test/finalization-recovery-worker.test.ts new file mode 100644 index 0000000000..32dbaa9584 --- /dev/null +++ b/packages/agent/test/finalization-recovery-worker.test.ts @@ -0,0 +1,92 @@ +import { describe, expect, it, vi } from 'vitest'; +import { + FINALIZATION_RECOVERY_WORKER_BATCH_SIZE, + FinalizationRecoveryWorker, +} from '../src/finalization-recovery-worker.js'; + +describe('FinalizationRecoveryWorker', () => { + it('continues full batches immediately and uses the configured SQLite batch size', async () => { + const processDueBatch = vi.fn() + .mockResolvedValueOnce(FINALIZATION_RECOVERY_WORKER_BATCH_SIZE) + .mockResolvedValueOnce(0); + const worker = new FinalizationRecoveryWorker( + processDueBatch, + { info: () => {}, warn: () => {} }, + { pollIntervalMs: 60_000 }, + ); + try { + worker.start(); + await vi.waitFor(() => expect(processDueBatch).toHaveBeenCalledTimes(2)); + expect(processDueBatch).toHaveBeenNthCalledWith( + 1, + FINALIZATION_RECOVERY_WORKER_BATCH_SIZE, + ); + expect(processDueBatch).toHaveBeenNthCalledWith( + 2, + FINALIZATION_RECOVERY_WORKER_BATCH_SIZE, + ); + } finally { + await worker.stop(); + } + }); + + it('never overlaps batches and waits for the active batch during shutdown', async () => { + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const processDueBatch = vi.fn(async () => { + await gate; + return FINALIZATION_RECOVERY_WORKER_BATCH_SIZE; + }); + const worker = new FinalizationRecoveryWorker( + processDueBatch, + { info: () => {}, warn: () => {} }, + { pollIntervalMs: 1 }, + ); + worker.start(); + await vi.waitFor(() => expect(processDueBatch).toHaveBeenCalledOnce()); + await new Promise((resolve) => setTimeout(resolve, 5)); + expect(processDueBatch).toHaveBeenCalledOnce(); + + let stopped = false; + const stopping = worker.stop().then(() => { + stopped = true; + }); + await Promise.resolve(); + expect(stopped).toBe(false); + release(); + await stopping; + expect(worker.running).toBe(false); + expect(processDueBatch).toHaveBeenCalledOnce(); + }); + + it('logs a transient batch failure and continues polling', async () => { + vi.useFakeTimers(); + const processDueBatch = vi.fn() + .mockRejectedValueOnce(new Error('busy')) + .mockResolvedValueOnce(0); + const warn = vi.fn(); + const worker = new FinalizationRecoveryWorker( + processDueBatch, + { info: () => {}, warn }, + { pollIntervalMs: 25 }, + ); + try { + worker.start(); + await vi.advanceTimersByTimeAsync(0); + + expect(processDueBatch).toHaveBeenCalledOnce(); + expect(warn).toHaveBeenCalledWith( + 'Finalization recovery worker batch failed: busy', + ); + expect(worker.running).toBe(true); + + await vi.advanceTimersByTimeAsync(25); + expect(processDueBatch).toHaveBeenCalledTimes(2); + } finally { + await worker.stop(); + vi.useRealTimers(); + } + }); +}); diff --git a/packages/agent/test/finalization-recovery.test.ts b/packages/agent/test/finalization-recovery.test.ts index 9fd6e17989..6a9d797d81 100644 --- a/packages/agent/test/finalization-recovery.test.ts +++ b/packages/agent/test/finalization-recovery.test.ts @@ -3,10 +3,18 @@ import { createHash } from 'node:crypto'; import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import { metrics } from '@opentelemetry/api'; +import { + AggregationTemporality, + InMemoryMetricExporter, + MeterProvider, + PeriodicExportingMetricReader, +} from '@opentelemetry/sdk-metrics'; import { decodeFinalizationMessage, encodeFinalizationMessage, GRAPH_KA_CONTENT_SCOPE_VERSION, + rebuildMetrics, type FinalizationMessageMsg, } from '@origintrail-official/dkg-core'; import { @@ -112,6 +120,548 @@ function recoveryMaterializer() { } describe('graph-scoped finalization recovery admission', () => { + it('autonomously settles a durable RECEIVED entry without a chain-cursor replay', async () => { + const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-due-worker-')); + try { + const store = await openSqliteFinalizationRecoveryStore(directory); + let applyCalls = 0; + const recovery = new FinalizationRecovery( + store, + recoveryChain(), + { info: () => {}, warn: () => {} }, + { + ...recoveryMaterializer(), + apply: async () => { + applyCalls += 1; + return 'applied' as const; + }, + }, + ); + await recovery.receive({ + rawMessage: encodeFinalizationMessage(message()), + contextGraphId: CONTEXT_GRAPH, + sourcePeerId: '12D3KooWPublisher', + candidate: parsedMessage(), + }); + expect(await store.list()).toMatchObject([{ + state: 'RECEIVED', + attemptCount: 0, + }]); + + await expect(recovery.processDueBatch(16)).resolves.toBe(1); + expect(applyCalls).toBe(1); + expect(await store.list()).toMatchObject([{ + state: 'SETTLED', + attemptCount: 0, + }]); + await store.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it('backs off a busy due entry before the autonomous worker retries it', async () => { + const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-due-backoff-')); + try { + let now = 1_000; + let busy = true; + const store = await openSqliteFinalizationRecoveryStore(directory, { + now: () => now, + }); + const recovery = new FinalizationRecovery( + store, + recoveryChain(), + { info: () => {}, warn: () => {} }, + { + ...recoveryMaterializer(), + apply: async () => { + if (busy) { + throw new StoreSchedulerBusyError( + 'queue_wait_timeout', + 'normal', + 'finalization-recovery-worker', + ); + } + return 'applied' as const; + }, + }, + ); + await recovery.receive({ + rawMessage: encodeFinalizationMessage(message()), + contextGraphId: CONTEXT_GRAPH, + sourcePeerId: '12D3KooWPublisher', + candidate: parsedMessage(), + }); + + await expect(recovery.processDueBatch(16)).resolves.toBe(1); + const [deferred] = await store.list(); + expect(deferred).toMatchObject({ + state: 'VERIFIED', + attemptCount: 1, + lastError: 'replay store scheduler remained busy', + }); + expect(deferred.nextAttemptAt).toBe(2_000); + await expect(recovery.processDueBatch(16)).resolves.toBe(0); + + busy = false; + now = deferred.nextAttemptAt!; + await expect(recovery.processDueBatch(16)).resolves.toBe(1); + expect(await store.list()).toMatchObject([{ state: 'SETTLED' }]); + await store.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it('continues a due batch after one entry throws and backs off the failed row', async () => { + const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-due-poison-')); + try { + let now = 1_000; + const secondTxHash = `0x${'bc'.repeat(32)}`; + const store = await openSqliteFinalizationRecoveryStore(directory, { + now: () => now, + }); + const recovery = new FinalizationRecovery( + store, + recoveryChain({ + resolveCanonicalFinalizationReceipt: async (txHash) => ({ + status: 'confirmed', + receipt: confirmedReceipt({ txHash }), + }), + }), + { info: () => {}, warn: () => {} }, + recoveryMaterializer(), + ); + await recovery.receive({ + rawMessage: encodeFinalizationMessage(message()), + contextGraphId: CONTEXT_GRAPH, + sourcePeerId: '12D3KooWPublisher', + candidate: parsedMessage(), + }); + await recovery.receive({ + rawMessage: encodeFinalizationMessage(message({ txHash: secondTxHash })), + contextGraphId: CONTEXT_GRAPH, + sourcePeerId: '12D3KooWPublisher', + candidate: parsedMessage({ txHash: secondTxHash }), + }); + const originalReplayDueEntry = recovery.replayDueEntry.bind(recovery); + const replayDueEntry = vi.spyOn(recovery, 'replayDueEntry'); + replayDueEntry + .mockRejectedValueOnce(new Error('poison replay')) + .mockImplementation(originalReplayDueEntry); + + await expect(recovery.processDueBatch(16)).resolves.toBe(2); + expect(replayDueEntry).toHaveBeenCalledTimes(2); + expect(await store.list()).toMatchObject([ + { + state: 'RECEIVED', + attemptCount: 1, + lastError: 'background replay failed: poison replay', + nextAttemptAt: 2_000, + }, + { + state: 'SETTLED', + txHash: secondTxHash, + }, + ]); + await store.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it('emits due metrics even when the due-inbox read fails', async () => { + const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-due-metrics-')); + const exporter = new InMemoryMetricExporter(AggregationTemporality.CUMULATIVE); + const meterProvider = new MeterProvider({ + readers: [ + new PeriodicExportingMetricReader({ + exporter, + exportIntervalMillis: 60_000, + }), + ], + }); + let store: Awaited> | undefined; + metrics.disable(); + expect(metrics.setGlobalMeterProvider(meterProvider)).toBe(true); + rebuildMetrics(); + try { + store = await openSqliteFinalizationRecoveryStore(directory); + vi.spyOn(store, 'listDue').mockRejectedValueOnce( + new Error('sqlite due read unavailable'), + ); + vi.spyOn(store, 'health').mockResolvedValue({ + available: true, + closed: false, + stateCounts: { RECEIVED: 7 }, + livePayloadBytes: 123, + dueEntries: 7, + oldestDueAgeMs: 4_321, + }); + const warn = vi.fn(); + const recovery = new FinalizationRecovery( + store, + recoveryChain(), + { info: () => {}, warn }, + recoveryMaterializer(), + ); + + await expect(recovery.processDueBatch(16)).resolves.toBe(0); + expect(warn).toHaveBeenCalledWith( + expect.stringContaining('due-inbox read failed'), + ); + expect(warn).not.toHaveBeenCalledWith( + expect.stringContaining('metrics snapshot failed'), + ); + await meterProvider.forceFlush(); + const datapoints = new Map(); + for (const resourceMetrics of exporter.getMetrics()) { + for (const scopeMetrics of resourceMetrics.scopeMetrics) { + for (const metric of scopeMetrics.metrics) { + for (const point of metric.dataPoints) { + if (typeof point.value === 'number') { + datapoints.set(metric.descriptor.name, point.value); + } + } + } + } + } + expect(datapoints.get('dkg.finalization_recovery.due_entries')).toBe(7); + expect(datapoints.get('dkg.finalization_recovery.oldest_due_age_ms')).toBe(4_321); + } finally { + await store?.close().catch(() => {}); + await meterProvider.shutdown().catch(() => {}); + metrics.disable(); + rebuildMetrics(); + await rm(directory, { recursive: true, force: true }); + } + }); + + it('autonomously applies a pending SETTLED publisher upgrade after restart', async () => { + const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-settled-upgrade-')); + let store: Awaited> | undefined; + try { + const chain = recoveryChain(); + store = await openSqliteFinalizationRecoveryStore(directory); + const initialRecovery = new FinalizationRecovery( + store, + chain, + { info: () => {}, warn: () => {} }, + recoveryMaterializer(), + ); + await initialRecovery.processLive({ + rawMessage: encodeFinalizationMessage(message()), + contextGraphId: CONTEXT_GRAPH, + sourcePeerId: '12D3KooWRelay', + candidate: parsedMessage(), + }); + const [settled] = await store.list(); + expect(settled).toMatchObject({ + state: 'SETTLED', + generation: 0, + publisherUpgradePending: false, + }); + await expect(store.recordSettledPublisherUpgrade( + settled!.key, + settled!.generation, + '12D3KooWPublisher', + )).resolves.toMatchObject({ status: 'recorded' }); + expect(await store.listDue(16)).toMatchObject([{ + state: 'SETTLED', + publisherUpgradePending: true, + }]); + + await store.close(); + store = await openSqliteFinalizationRecoveryStore(directory); + let upgradeApplyCalls = 0; + const restartedRecovery = new FinalizationRecovery( + store, + chain, + { info: () => {}, warn: () => {} }, + { + ...recoveryMaterializer(), + prepare: async () => ({ + onChainContextGraphId: '42', + localTopicOnChainContextGraphId: '42', + publisherPeerId: '12D3KooWPublisher', + accessPolicy: 'allowList' as const, + allowedPeers: ['12D3KooWReader'], + }), + apply: async () => { + upgradeApplyCalls += 1; + return 'applied' as const; + }, + }, + ); + + await expect(restartedRecovery.processDueBatch(16)).resolves.toBe(1); + expect(upgradeApplyCalls).toBe(1); + expect(await store.list()).toMatchObject([{ + state: 'SETTLED', + generation: 1, + publisherUpgradePending: false, + trustedPublisherPeerId: '12D3KooWPublisher', + }]); + } finally { + await store?.close().catch(() => {}); + await rm(directory, { recursive: true, force: true }); + } + }); + + it('rejects a VERIFIED poison entry only after count and age budgets expire', async () => { + const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-due-exhausted-')); + try { + let now = 1_000; + let applyCalls = 0; + const store = await openSqliteFinalizationRecoveryStore(directory, { + maxEntries: 1, + now: () => now, + }); + const recovery = new FinalizationRecovery( + store, + recoveryChain(), + { info: () => {}, warn: () => {} }, + { + ...recoveryMaterializer(), + apply: async () => { + applyCalls += 1; + throw new StoreSchedulerBusyError( + 'queue_wait_timeout', + 'normal', + 'finalization-recovery-worker', + ); + }, + replayVerified: async () => { + throw new StoreSchedulerBusyError( + 'queue_wait_timeout', + 'normal', + 'finalization-recovery-worker', + ); + }, + }, + { + liveRetryLimit: 2, + liveRetryWindowMs: 10_000, + now: () => now, + }, + ); + await recovery.receive({ + rawMessage: encodeFinalizationMessage(message()), + contextGraphId: CONTEXT_GRAPH, + sourcePeerId: '12D3KooWPublisher', + candidate: parsedMessage(), + }); + + await recovery.processDueBatch(16); + let [entry] = await store.list(); + expect(entry).toMatchObject({ state: 'VERIFIED', attemptCount: 1 }); + now = entry!.nextAttemptAt!; + await recovery.processDueBatch(16); + [entry] = await store.list(); + expect(entry).toMatchObject({ state: 'VERIFIED', attemptCount: 2 }); + now = entry!.nextAttemptAt!; + + await expect(recovery.processDueBatch(16)).resolves.toBe(1); + [entry] = await store.list(); + expect(entry).toMatchObject({ state: 'VERIFIED', attemptCount: 3 }); + + now = entry!.createdAt + 10_000; + await expect(recovery.processDueBatch(16)).resolves.toBe(1); + expect(await store.list()).toMatchObject([{ + state: 'REJECTED', + attemptCount: 3, + lastError: 'autonomous retry budget exhausted after 3 attempts over 10000ms', + }]); + await expect(recovery.processLive({ + rawMessage: encodeFinalizationMessage(message()), + contextGraphId: CONTEXT_GRAPH, + sourcePeerId: '12D3KooWPublisher', + candidate: parsedMessage(), + })).resolves.toBe(true); + expect(applyCalls).toBe(1); + expect(await store.list()).toMatchObject([{ + state: 'REJECTED', + attemptCount: 3, + }]); + await expect(store.receive({ + key: 'replacement', + chainId: 'base:84532', + contextGraphId: CONTEXT_GRAPH, + ual: UAL, + txHash: `0x${'ef'.repeat(32)}`, + assertionVersion: '1', + merkleRoot: `0x${'00'.repeat(32)}`, + kaId: PACKED_KA_ID.toString(), + batchId: PACKED_KA_ID.toString(), + rawMessage: encodeFinalizationMessage(message({ + txHash: `0x${'ef'.repeat(32)}`, + })), + })).resolves.toMatchObject({ status: 'inserted' }); + await store.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it('serializes live and autonomous materialization for the same inbox key', async () => { + const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-entry-lock-')); + try { + const store = await openSqliteFinalizationRecoveryStore(directory); + let releaseApply: (() => void) | undefined; + const applyGate = new Promise((resolve) => { + releaseApply = resolve; + }); + let applyCalls = 0; + let replayVerifiedCalls = 0; + let concurrentMaterializations = 0; + let maximumConcurrentMaterializations = 0; + const enterMaterializationSection = async (): Promise => { + concurrentMaterializations += 1; + maximumConcurrentMaterializations = Math.max( + maximumConcurrentMaterializations, + concurrentMaterializations, + ); + await applyGate; + concurrentMaterializations -= 1; + }; + const recovery = new FinalizationRecovery( + store, + recoveryChain(), + { info: () => {}, warn: () => {} }, + { + ...recoveryMaterializer(), + apply: async () => { + applyCalls += 1; + await enterMaterializationSection(); + return 'applied' as const; + }, + replayVerified: async () => { + replayVerifiedCalls += 1; + await enterMaterializationSection(); + return 'promoted' as const; + }, + }, + ); + const input = { + rawMessage: encodeFinalizationMessage(message()), + contextGraphId: CONTEXT_GRAPH, + sourcePeerId: '12D3KooWPublisher', + candidate: parsedMessage(), + }; + await recovery.receive(input); + + const live = recovery.processLive(input); + const worker = recovery.processDueBatch(16); + await vi.waitFor(() => expect(applyCalls).toBe(1)); + releaseApply?.(); + await Promise.all([live, worker]); + + expect(maximumConcurrentMaterializations).toBe(1); + expect(applyCalls).toBe(1); + expect(replayVerifiedCalls).toBe(1); + expect(await store.list()).toMatchObject([{ state: 'SETTLED' }]); + await store.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it('does not let duplicate live gossip erase a live-entry retry deadline', async () => { + const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-live-backoff-')); + try { + let now = 1_000; + let receiptCalls = 0; + const store = await openSqliteFinalizationRecoveryStore(directory, { + now: () => now, + }); + const recovery = new FinalizationRecovery( + store, + recoveryChain({ + resolveCanonicalFinalizationReceipt: async () => { + receiptCalls += 1; + return { status: 'pending' }; + }, + }), + { info: () => {}, warn: () => {} }, + recoveryMaterializer(), + { + liveRetryLimit: 1, + liveRetryWindowMs: 10_000, + now: () => now, + }, + ); + const input = { + rawMessage: encodeFinalizationMessage(message()), + contextGraphId: CONTEXT_GRAPH, + sourcePeerId: '12D3KooWPublisher', + candidate: parsedMessage(), + }; + + await recovery.receive(input); + await recovery.processDueBatch(16); + const [deferred] = await store.list(); + expect(deferred).toMatchObject({ + state: 'RECEIVED', + attemptCount: 1, + nextAttemptAt: 2_000, + }); + await recovery.processLive(input); + expect(receiptCalls).toBe(2); + expect(await store.list()).toMatchObject([{ + attemptCount: 2, + nextAttemptAt: 2_000, + }]); + + now = 2_000; + await recovery.processDueBatch(16); + expect(receiptCalls).toBe(3); + expect(await store.list()).toMatchObject([{ + state: 'RECEIVED', + attemptCount: 3, + }]); + await store.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + + it('lets chain reconciliation recover an entry before its worker deadline', async () => { + const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-reconcile-deadline-')); + try { + let now = 1_000; + const store = await openSqliteFinalizationRecoveryStore(directory, { + now: () => now, + }); + const chain = recoveryChain(); + const recovery = new FinalizationRecovery( + store, + chain, + { info: () => {}, warn: () => {} }, + recoveryMaterializer(), + ); + const entry = await recovery.receive({ + rawMessage: encodeFinalizationMessage(message()), + contextGraphId: CONTEXT_GRAPH, + sourcePeerId: '12D3KooWPublisher', + candidate: parsedMessage(), + }); + await store.recordAttempt(entry!.key, entry!.generation, 'worker busy', 60_000); + + await expect(recovery.replayMatching({ + chainId: chain.chainId, + contextGraphId: CONTEXT_GRAPH, + onChainCgId: '42', + ual: UAL, + merkleRoot: `0x${'00'.repeat(32)}`, + kaId: PACKED_KA_ID.toString(), + })).resolves.toBe('recovered'); + expect(await store.list()).toMatchObject([{ state: 'SETTLED' }]); + await store.close(); + } finally { + await rm(directory, { recursive: true, force: true }); + } + }); + it('returns a typed parsed envelope for a valid singleton finalization', () => { expect(parseGraphScopedFinalization(message(), CONTEXT_GRAPH)).toMatchObject({ ok: true, @@ -246,6 +796,7 @@ describe('graph-scoped finalization recovery admission', () => { it('persists RECEIVED before a store-heavy operation can fail', async () => { const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-recovery-wiring-')); let agent: DKGAgent | undefined; + let releaseWorkerStop: (() => void) | undefined; try { agent = await DKGAgent.create({ name: 'FinalizationRecoveryWiringBot', @@ -254,7 +805,10 @@ describe('graph-scoped finalization recovery admission', () => { chainAdapter: new MockChainAdapter(), }); const preStartHandler = agent.getOrCreateFinalizationHandler(); + const startWorker = vi.spyOn(preStartHandler, 'startRecoveryWorker'); + const stopWorker = vi.spyOn(preStartHandler, 'stopRecoveryWorker'); await agent.start(); + expect(startWorker).toHaveBeenCalledTimes(1); expect(agent.getOrCreateFinalizationHandler()).toBe(preStartHandler); const originalQuery = agent.store.query.bind(agent.store); agent.store.query = async () => { @@ -277,7 +831,22 @@ describe('graph-scoped finalization recovery admission', () => { available: true, stateCounts: { RECEIVED: 1 }, }); - await agent.stop(); + const workerStopGate = new Promise((resolve) => { + releaseWorkerStop = resolve; + }); + stopWorker.mockImplementationOnce(() => workerStopGate); + const nextTeardown = vi.spyOn( + agent as unknown as { + closeRfc64PublicCatalogBootstrapV1(): Promise; + }, + 'closeRfc64PublicCatalogBootstrapV1', + ); + const stopping = agent.stop(); + await vi.waitFor(() => expect(stopWorker).toHaveBeenCalledTimes(1)); + expect(nextTeardown).not.toHaveBeenCalled(); + releaseWorkerStop(); + await stopping; + expect(nextTeardown).toHaveBeenCalled(); expect(agent.getOrCreateFinalizationHandler()).toBe(preStartHandler); expect(await agent.getFinalizationRecoveryHealth()).toMatchObject({ available: false, @@ -285,6 +854,7 @@ describe('graph-scoped finalization recovery admission', () => { degradedReason: 'not-configured', }); } finally { + releaseWorkerStop?.(); await agent?.stop().catch(() => {}); await agent?.store.close().catch(() => {}); await rm(directory, { recursive: true, force: true }); @@ -645,11 +1215,11 @@ describe('graph-scoped finalization recovery admission', () => { ...replay, merkleRoot: `0x${'ff'.repeat(32)}`, }); - await vi.waitFor(() => expect(replayReceiptCalls).toBe(2)); releaseReceipts(); await expect(Promise.all([currentRootReplay, newerRootReplay])) .resolves.toEqual(['recovered', 'none']); + expect(replayReceiptCalls).toBe(2); } finally { releaseReceipts(); await store?.close().catch(() => {}); diff --git a/packages/agent/test/ka-graph-finalization-handler.test.ts b/packages/agent/test/ka-graph-finalization-handler.test.ts index 91dcc004ea..4acd8d8b03 100644 --- a/packages/agent/test/ka-graph-finalization-handler.test.ts +++ b/packages/agent/test/ka-graph-finalization-handler.test.ts @@ -1,4 +1,4 @@ -import { beforeEach, describe, expect, it } from 'vitest'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -1209,6 +1209,83 @@ describe('graph-scoped finalization handler', () => { } }); + it('autonomously drains a persisted busy finalization after restart', async () => { + const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-worker-restart-')); + let inbox: SqliteFinalizationRecoveryStore | undefined; + let restarted: FinalizationHandler | undefined; + const query = store.query.bind(store); + try { + const { message, vmGraph } = await stageGraph(); + const chain = { + chainId: 'base:84532', + getLatestMerkleRoot: async () => message.kcMerkleRoot, + getMerkleRootCount: async () => 1n, + getKAContextGraphId: async () => 42n, + resolveCanonicalFinalizationReceipt: async () => canonicalReceipt(message), + } as ChainAdapter; + inbox = await openSqliteFinalizationRecoveryStore(directory, { + maxPerContextGraph: 1, + }); + const pressured = new FinalizationHandler( + store, + chain, + recoveryOptions(inbox), + ); + let busyReads = 2; + store.query = async (sparql, options) => { + if (busyReads > 0) { + busyReads -= 1; + throw new StoreSchedulerBusyError( + 'queue_wait_timeout', + 'normal', + 'autonomous-finalization-recovery.query', + ); + } + return query(sparql, options); + }; + + await pressured.handleFinalizationMessage( + encodeFinalizationMessage(message), + CG, + '12D3KooWPublisher', + ); + expect(await inbox.list()).toMatchObject([{ + state: 'RECEIVED', + attemptCount: 1, + lastError: 'store scheduler remained busy', + }]); + expect(await inbox.health()).toMatchObject({ + ready: false, + degradedReason: 'capacity-exhausted', + }); + + store.query = query; + await inbox.close(); + inbox = await openSqliteFinalizationRecoveryStore(directory, { + maxPerContextGraph: 1, + }); + restarted = new FinalizationHandler( + store, + chain, + recoveryOptions(inbox), + ); + restarted.startRecoveryWorker(); + + await vi.waitFor(async () => { + expect(await inbox!.list()).toMatchObject([{ state: 'SETTLED' }]); + }); + expect(await store.countQuads(vmGraph)).toBe(2); + const health = await inbox.health(); + expect(health.ready).toBe(true); + expect(health).not.toHaveProperty('degradedReason'); + } finally { + store.query = query; + await restarted?.stopRecoveryWorker(); + await closeInbox(inbox); + await rm(directory, { recursive: true, force: true }); + } + }); + it('does not mutate Oxigraph when the VERIFIED transaction cannot commit', async () => { const directory = await mkdtemp(join(tmpdir(), 'dkg-finalization-recovery-')); let inbox: SqliteFinalizationRecoveryStore | undefined; @@ -1222,6 +1299,7 @@ describe('graph-scoped finalization handler', () => { inbox = await openSqliteFinalizationRecoveryStore(directory); const failingVerifiedStore: FinalizationRecoveryStore = { get closed() { return inbox!.closed; }, + get: inbox.get.bind(inbox), receive: inbox.receive.bind(inbox), recordTrustedPublisher: inbox.recordTrustedPublisher.bind(inbox), recordSettledPublisherUpgrade: @@ -1233,6 +1311,7 @@ describe('graph-scoped finalization handler', () => { clearSettledRetry: inbox.clearSettledRetry.bind(inbox), rejectSettled: inbox.rejectSettled.bind(inbox), isAttemptDue: inbox.isAttemptDue.bind(inbox), + listDue: inbox.listDue.bind(inbox), listForKnowledgeAsset: inbox.listForKnowledgeAsset.bind(inbox), transition: inbox.transition.bind(inbox), recordAttempt: inbox.recordAttempt.bind(inbox), @@ -1294,6 +1373,7 @@ describe('graph-scoped finalization handler', () => { } as ChainAdapter; const rejectedStore: FinalizationRecoveryStore = { closed: false, + get: async () => undefined, receive: async () => { if (failureMode === 'write-failure') throw new Error('disk full'); return { status: 'capacity' }; @@ -1314,6 +1394,7 @@ describe('graph-scoped finalization handler', () => { clearSettledRetry: async () => {}, rejectSettled: async () => false, isAttemptDue: () => true, + listDue: async () => [], listForKnowledgeAsset: async () => [], transition: async () => false, recordAttempt: async () => {}, @@ -1322,6 +1403,7 @@ describe('graph-scoped finalization handler', () => { closed: false, stateCounts: {}, livePayloadBytes: 0, + dueEntries: 0, }), close: async () => {}, }; diff --git a/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts b/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts index e5f011619c..14035787fc 100644 --- a/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts +++ b/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts @@ -263,6 +263,28 @@ describe('DKGAgent RFC-64 inventory lifecycle', () => { await expect(agent.closeRfc64PersistenceV1()).resolves.toBeUndefined(); }); + it('preserves capacity exhaustion when canonical receipt support is unavailable', async () => { + const agent = syntheticAgent(); + agent.chain = { chainId: 'none' }; + agent.finalizationRuntime.attachRecoveryStore({ + health: async () => ({ + available: true, + closed: false, + ready: false, + degradedReason: 'capacity-exhausted', + stateCounts: { RECEIVED: 64 }, + livePayloadBytes: 1, + dueEntries: 64, + }), + } as any); + + await expect(agent.getFinalizationRecoveryHealth()).resolves.toMatchObject({ + ready: false, + canonicalReceiptCapability: 'unsupported', + degradedReason: 'capacity-exhausted', + }); + }); + it('owns one persistent foundation and purges every stale candidate in bounded yielding batches', async () => { const dataDirectory = temporaryDataDirectory(); await seedStaleCandidateLoads(dataDirectory, 17); diff --git a/packages/agent/vitest.unit.config.ts b/packages/agent/vitest.unit.config.ts index f0a78ba659..7c60b9d4f2 100644 --- a/packages/agent/vitest.unit.config.ts +++ b/packages/agent/vitest.unit.config.ts @@ -99,6 +99,7 @@ export default defineConfig({ "test/finalization-handler-chain-truth.test.ts", "test/finalization-handler-defensive-cg-id.test.ts", "test/finalization-recovery.test.ts", + "test/finalization-recovery-worker.test.ts", "test/finalization-recovery-sqlite-store.test.ts", "test/named-ka-publish-recovery.test.ts", "test/ka-graph-finalization-handler.test.ts", diff --git a/packages/chain/test/strict-current-finalized-evm-rpc.unit.test.ts b/packages/chain/test/strict-current-finalized-evm-rpc.unit.test.ts index 808cfdb7e4..f1f6588098 100644 --- a/packages/chain/test/strict-current-finalized-evm-rpc.unit.test.ts +++ b/packages/chain/test/strict-current-finalized-evm-rpc.unit.test.ts @@ -244,8 +244,12 @@ describe('RFC-64 strict current-finalized raw JSON-RPC transport', () => { 'eth_getCode', 'eth_getCode', ]); - expect(server.calls[2]!.params[0]).toBe(TO); - expect(server.calls[3]!.params[0]).toBe(OTHER_TO); + // Distinct target probes run in parallel, so their server arrival order is + // intentionally unspecified. The contract is that every target is checked + // before the eth_call phase begins. + expect(server.calls.slice(2)).toHaveLength(2); + expect(new Set(server.calls.slice(2).map(({ params }) => params[0]))) + .toEqual(new Set([TO, OTHER_TO])); }); it('rejects an oversized generic return only after a stable fallback sandwich', async () => { diff --git a/packages/cli/src/daemon/routes/status.ts b/packages/cli/src/daemon/routes/status.ts index 32595b399b..4136881bcd 100644 --- a/packages/cli/src/daemon/routes/status.ts +++ b/packages/cli/src/daemon/routes/status.ts @@ -696,6 +696,7 @@ export async function handleStatusRoutes(ctx: RequestContext): Promise { degradedReason: reason, stateCounts: {}, livePayloadBytes: 0, + dueEntries: 0, }); let finalizationRecovery: Awaited< ReturnType diff --git a/packages/cli/test/status-route-rpc.test.ts b/packages/cli/test/status-route-rpc.test.ts index 1442c637fb..a0c2c80afd 100644 --- a/packages/cli/test/status-route-rpc.test.ts +++ b/packages/cli/test/status-route-rpc.test.ts @@ -189,6 +189,7 @@ describe('/api/status finalization recovery health', () => { stateCounts: { RECEIVED: 1 }, liveEntries: 1, livePayloadBytes: 4, + dueEntries: 1, }; const response = await requestStatusWithAgent({ @@ -215,6 +216,7 @@ describe('/api/status finalization recovery health', () => { degradedReason: 'finalization inbox health read failed', stateCounts: {}, livePayloadBytes: 0, + dueEntries: 0, }); }); }); diff --git a/packages/core/src/telemetry-api.ts b/packages/core/src/telemetry-api.ts index 7c3983c573..1b200cc95a 100644 --- a/packages/core/src/telemetry-api.ts +++ b/packages/core/src/telemetry-api.ts @@ -195,6 +195,12 @@ export interface DkgMetrics { storeCancellationCompletedTotal: Counter; /** scope and reason identify the bounded retry loop; attempt is capped */ storeRetryAttemptsTotal: Counter; + /** current durable finalization entries whose retry gate is open */ + finalizationRecoveryDueEntries: Gauge; + /** milliseconds since the oldest currently due finalization was received */ + finalizationRecoveryOldestDueAgeMs: Gauge; + /** outcome={recovered|invalidated|retry-pending|none} */ + finalizationRecoveryAttemptsTotal: Counter; /** process-local sync inflight sample */ syncGlobalInflight: Histogram; /** ms; lane and priority_class are bounded sync scheduler enums */ @@ -328,6 +334,21 @@ function buildMetrics(): DkgMetrics { storeRetryAttemptsTotal: meter.createCounter('dkg.store.retry_attempts_total', { description: 'Bounded expensive-work retry attempts', }), + finalizationRecoveryDueEntries: meter.createGauge( + 'dkg.finalization_recovery.due_entries', + { description: 'Durable finalization inbox entries currently eligible for retry' }, + ), + finalizationRecoveryOldestDueAgeMs: meter.createGauge( + 'dkg.finalization_recovery.oldest_due_age_ms', + { + unit: 'ms', + description: 'Age of the oldest durable finalization inbox entry eligible for retry', + }, + ), + finalizationRecoveryAttemptsTotal: meter.createCounter( + 'dkg.finalization_recovery.attempts_total', + { description: 'Autonomous finalization replay attempts by bounded outcome' }, + ), syncGlobalInflight: meter.createHistogram('dkg.sync.global_inflight', { description: 'Sampled process-local sync inflight count', }),