diff --git a/packages/agent/src/dkg-agent-lifecycle.ts b/packages/agent/src/dkg-agent-lifecycle.ts index 366d8c5056..4dd5210f23 100644 --- a/packages/agent/src/dkg-agent-lifecycle.ts +++ b/packages/agent/src/dkg-agent-lifecycle.ts @@ -2557,10 +2557,10 @@ export class LifecycleSyncMethods extends DKGAgentBase { // sweep is the safety net if this is missed. Only wired when reconciliation // is actually possible (chain + ordinal reads present). onKARegisteredToContextGraph: this.vmReconcileEnabled() - ? async ({ contextGraphId: onChainId, kaId }) => { + ? async ({ contextGraphId: onChainId, kaId }, signal) => { // GH #1098 — body extracted to `handleKARegisteredNudge` so the // bind-only-the-matching-CG branch is directly testable. - await this.handleKARegisteredNudge(onChainId, kaId, ctx); + await this.handleKARegisteredNudge(onChainId, kaId, ctx, signal); } : undefined, }); diff --git a/packages/agent/src/dkg-agent-swm-host.ts b/packages/agent/src/dkg-agent-swm-host.ts index 8cc2922c1c..15ff511624 100644 --- a/packages/agent/src/dkg-agent-swm-host.ts +++ b/packages/agent/src/dkg-agent-swm-host.ts @@ -2636,17 +2636,19 @@ export class SwmHostModeMethods extends DKGAgentBase { localCgId: string, sub: ContextGraphSub, targetOnChainId?: bigint, + signal?: AbortSignal, ): Promise { - if (!sub.subscribed || sub.onChainId) return null; + if (signal?.aborted || !sub.subscribed || sub.onChainId) return null; let resolved: string | null = null; try { resolved = await this.getContextGraphOnChainId(localCgId, { + signal, source: 'agent.vmReconcile.resolveOnChainId', }); } catch { return null; } - if (!resolved) return null; + if (signal?.aborted || !resolved) return null; if (targetOnChainId !== undefined) { let resolvedNum: bigint | null = null; try { resolvedNum = BigInt(resolved); } catch { return null; } @@ -2680,7 +2682,9 @@ export class SwmHostModeMethods extends DKGAgentBase { onChainId: string, kaId: bigint, ctx: OperationContext, + signal?: AbortSignal, ): Promise { + if (signal?.aborted) return null; let targetOnChain: bigint | null = null; try { targetOnChain = BigInt(onChainId); } catch { targetOnChain = null; } @@ -2693,7 +2697,8 @@ export class SwmHostModeMethods extends DKGAgentBase { // path); the sweep remains the safety net for a CG whose quad hasn't arrived. if (targetOnChain !== null) { for (const [lcg, sub] of this.subscribedContextGraphs) { - const bound = await this.selfPrimeSubscriptionOnChainId(lcg, sub, targetOnChain); + if (signal?.aborted) return null; + const bound = await this.selfPrimeSubscriptionOnChainId(lcg, sub, targetOnChain, signal); if (bound) { this.log.info(ctx, `Phase B: KACG nudge cg=${onChainId} ka=${kaId} -> bound + reconcile pre-subscribed "${lcg}"`); if (this.vmReconcileDispatcher) void this.vmReconcileDispatcher.triggerLive(lcg); diff --git a/packages/agent/src/dkg-agent.ts b/packages/agent/src/dkg-agent.ts index 89ce17c18b..32c9218a1f 100644 --- a/packages/agent/src/dkg-agent.ts +++ b/packages/agent/src/dkg-agent.ts @@ -1720,8 +1720,26 @@ export class DKGAgent extends DKGAgentBase { await this.chainPoller?.waitForCurrentPoll(); } + /** + * Close network admission before an outer runtime awaits any of its own + * shutdown dependencies. The daemon owns catch-up and publish workers that + * are drained before {@link stop}; those workers can themselves be blocked + * on this agent's protocol router, so waiting until stop() would recreate + * the same cancellation-order inversion one layer higher. + */ + beginStop(): void { + this.node.beginStop(); + } + async stop(): Promise { if (!this.started) return; + // Abort network-backed work before awaiting the chain poller. The poller + // dispatches VM-reconcile callbacks that may themselves be blocked in a + // sync read; waiting for the poll before DKGNode.stop() used to invert the + // cancellation order and force the daemon's shutdown watchdog to exit 100. + // beginStop() only closes admission/aborts reads; the actual libp2p stop + // remains at the existing dependency-safe point below. + this.beginStop(); this.syncCapacityRuntime.stopSampling(); if (this.chainPoller) { // Await so any in-flight poll (and its HTTP keep-alive socket) settles diff --git a/packages/agent/test/outbox-shutdown-lifecycle.test.ts b/packages/agent/test/outbox-shutdown-lifecycle.test.ts index c283e5cd1b..27e2d18da1 100644 --- a/packages/agent/test/outbox-shutdown-lifecycle.test.ts +++ b/packages/agent/test/outbox-shutdown-lifecycle.test.ts @@ -6,6 +6,47 @@ import { FinalizationRuntime } from '../src/finalization-runtime.js'; import { VmReconcileQueueClosedError } from '../src/vm-reconcile-service.js'; describe('DKGAgent outbox shutdown lifecycle', () => { + it('aborts network waits before awaiting the chain-event poller drain', async () => { + let releasePoller!: () => void; + const pollerDrain = new Promise((resolve) => { releasePoller = resolve; }); + const beginStop = vi.fn(); + const stopNode = vi.fn(async () => {}); + const chainPollerStop = vi.fn(async () => { + expect(beginStop).toHaveBeenCalledOnce(); + await pollerDrain; + }); + const agent = Object.create(DKGAgent.prototype) as any; + Object.assign(agent, { + started: true, + syncCapacityRuntime: { stopSampling: vi.fn() }, + chainPoller: { stop: chainPollerStop }, + coreHostRecordingsClosed: false, + drainCoreHostRecordings: vi.fn(async () => {}), + messenger: { stopOutboxDrain: vi.fn(async () => {}) }, + clearRandomSamplingBindRetry: vi.fn(), + clearStorageACKRegistrationRetry: vi.fn(), + storageACKRegistrationRetryInFlight: false, + randomSamplingHandle: null, + inFlightSubstrateFanOutCount: () => 0, + router: { closePooling: vi.fn(async () => {}) }, + node: { beginStop, stop: stopNode }, + finalizationRuntime: new FinalizationRuntime(), + store: { close: vi.fn(async () => {}) }, + log: { warn: vi.fn() }, + }); + + const stopping = agent.stop(); + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(beginStop).toHaveBeenCalledOnce(); + expect(chainPollerStop).toHaveBeenCalledOnce(); + expect(stopNode).not.toHaveBeenCalled(); + + releasePoller(); + await stopping; + expect(stopNode).toHaveBeenCalledOnce(); + }); + it('closes reconcile admission and cancels queued jobs before store teardown', async () => { let releaseActive!: () => void; let queuedStarted = false; @@ -23,6 +64,7 @@ describe('DKGAgent outbox shutdown lifecycle', () => { const agent = Object.create(DKGAgent.prototype) as any; Object.assign(agent, { started: true, + syncCapacityRuntime: { stopSampling: vi.fn() }, chainPoller: null, vmReconcileDispatcher: dispatcher, coreHostRecordingsClosed: false, @@ -34,7 +76,7 @@ describe('DKGAgent outbox shutdown lifecycle', () => { randomSamplingHandle: null, inFlightSubstrateFanOutCount: () => 0, router: { closePooling: vi.fn(async () => {}) }, - node: { stop: stopNode }, + node: { beginStop: vi.fn(), stop: stopNode }, finalizationRuntime: new FinalizationRuntime(), store: { close: closeStore }, log: { warn: vi.fn() }, @@ -75,6 +117,7 @@ describe('DKGAgent outbox shutdown lifecycle', () => { const agent = Object.create(DKGAgent.prototype) as any; Object.assign(agent, { started: true, + syncCapacityRuntime: { stopSampling: vi.fn() }, chainPoller: null, vmReconcileDispatcher: dispatcher, coreHostRecordingsClosed: false, @@ -86,7 +129,7 @@ describe('DKGAgent outbox shutdown lifecycle', () => { randomSamplingHandle: null, inFlightSubstrateFanOutCount: () => 0, router: { closePooling: vi.fn(async () => {}) }, - node: { stop: stopNode }, + node: { beginStop: vi.fn(), stop: stopNode }, finalizationRuntime: new FinalizationRuntime(), store: { close: closeStore }, log: { warn }, @@ -116,6 +159,7 @@ describe('DKGAgent outbox shutdown lifecycle', () => { const agent = Object.create(DKGAgent.prototype) as any; Object.assign(agent, { started: true, + syncCapacityRuntime: { stopSampling: vi.fn() }, chainPoller: null, coreHostRecordingsClosed: false, drainCoreHostRecordings: vi.fn(async () => {}), @@ -126,7 +170,7 @@ describe('DKGAgent outbox shutdown lifecycle', () => { randomSamplingHandle: null, inFlightSubstrateFanOutCount: () => 0, router: { closePooling: vi.fn(async () => {}) }, - node: { stop: stopNode }, + node: { beginStop: vi.fn(), stop: stopNode }, finalizationRuntime: new FinalizationRuntime(), store: { close: vi.fn(async () => {}) }, log: { warn: vi.fn() }, @@ -150,6 +194,7 @@ describe('DKGAgent outbox shutdown lifecycle', () => { const agent = Object.create(DKGAgent.prototype) as any; Object.assign(agent, { started: true, + syncCapacityRuntime: { stopSampling: vi.fn() }, chainPoller: null, coreHostRecordingsClosed: false, drainCoreHostRecordings: vi.fn(async () => {}), @@ -160,7 +205,7 @@ describe('DKGAgent outbox shutdown lifecycle', () => { randomSamplingHandle: null, inFlightSubstrateFanOutCount: () => 0, router: { closePooling: vi.fn(async () => {}) }, - node: { stop: stopNode }, + node: { beginStop: vi.fn(), stop: stopNode }, finalizationRuntime: new FinalizationRuntime(), store: { close: closeStore }, log: { warn }, diff --git a/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts b/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts index 7235010096..bf0301bec2 100644 --- a/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts +++ b/packages/agent/test/rfc64-agent-inventory-lifecycle.test.ts @@ -167,7 +167,10 @@ function minimalStartedAgent( randomSamplingHandle: null, inFlightSubstrateFanOutCount: () => 0, router: { closePooling: vi.fn(async () => {}) }, - node: { stop: vi.fn(async () => { order.push('node'); }) }, + node: { + beginStop: vi.fn(), + stop: vi.fn(async () => { order.push('node'); }), + }, syncVerifyWorker: { close: vi.fn(async () => { order.push('sync-worker'); }) }, rfc64PersistenceV1: { close: () => { diff --git a/packages/agent/test/vm-reconcile-self-prime.test.ts b/packages/agent/test/vm-reconcile-self-prime.test.ts index 58ed5e9d6b..22e0fc28f6 100644 --- a/packages/agent/test/vm-reconcile-self-prime.test.ts +++ b/packages/agent/test/vm-reconcile-self-prime.test.ts @@ -27,8 +27,13 @@ interface AgentInternals { localCgId: string, sub: { subscribed: boolean; coreHosted?: boolean; onChainId?: string }, targetOnChainId?: bigint, + signal?: AbortSignal, + ): Promise; + handleKARegisteredNudge(onChainId: string, kaId: bigint, ctx: unknown, signal?: AbortSignal): Promise; + getContextGraphOnChainId( + localCgId: string, + options?: { signal?: AbortSignal; source?: string }, ): Promise; - handleKARegisteredNudge(onChainId: string, kaId: bigint, ctx: unknown): Promise; subscribedContextGraphs: Map; vmReconcileDispatcher: { dispatch: (cg: string, reason: 'live' | 'periodic') => Promise; @@ -210,4 +215,50 @@ describe('GH #1098 — VM reconcile sweep self-primes onChainId for a pre-subscr expect(reconciled).toBe(CG_BOUND); expect(triggered).toEqual([`live:${CG_BOUND}`]); }); + + it('live KACG nudge handler aborts an in-flight CG resolver without binding or reconciling', async () => { + const chain = new MockChainAdapter(); + agent = await DKGAgent.create({ name: 'KacgNudgeAbort', chainAdapter: chain }); + stubNode(agent); + const internals = agent as unknown as AgentInternals; + const localCg = 'gh1098-nudge-abort'; + internals.subscribedContextGraphs.set(localCg, { subscribed: true }); + + let resolverStarted!: () => void; + const started = new Promise((resolve) => { resolverStarted = resolve; }); + internals.getContextGraphOnChainId = async (_id, options = {}) => { + resolverStarted(); + await new Promise((_resolve, reject) => { + const rejectAbort = () => { + const error = new Error('resolver aborted'); + error.name = 'AbortError'; + reject(error); + }; + if (options.signal?.aborted) rejectAbort(); + else options.signal?.addEventListener('abort', rejectAbort, { once: true }); + }); + return '9090'; + }; + const triggered: string[] = []; + internals.vmReconcileDispatcher = { + dispatch: async () => true, + triggerLive: (cg: string) => { triggered.push(cg); }, + triggerPeriodic: () => undefined, + tryTriggerPeriodic: () => true, + }; + + const controller = new AbortController(); + const pending = internals.handleKARegisteredNudge( + '9090', + 1n, + createOperationContext('system'), + controller.signal, + ); + await started; + controller.abort(); + + await expect(pending).resolves.toBeNull(); + expect(internals.subscribedContextGraphs.get(localCg)?.onChainId).toBeUndefined(); + expect(triggered).toEqual([]); + }); }); diff --git a/packages/chain/src/chain-adapter.ts b/packages/chain/src/chain-adapter.ts index e2fd29ed9d..8619d4efdc 100644 --- a/packages/chain/src/chain-adapter.ts +++ b/packages/chain/src/chain-adapter.ts @@ -346,6 +346,11 @@ export interface EventFilter { toBlock?: number; } +export interface ChainEventListenOptions { + /** Stop an in-flight event scan without allowing its caller to advance a cursor. */ + signal?: AbortSignal; +} + export interface CreateContextGraphParams { /** * Human-readable context graph name. The on-chain contextGraphId is derived as @@ -1055,7 +1060,7 @@ export interface ChainAdapter { getBlockNumber?(): Promise; // Events - listenForEvents(filter: EventFilter): AsyncIterable; + listenForEvents(filter: EventFilter, options?: ChainEventListenOptions): AsyncIterable; // Context Graphs (name-hash commitment via ContextGraphNameRegistry) createContextGraph(params: CreateContextGraphParams): Promise; diff --git a/packages/chain/src/evm-adapter-events.ts b/packages/chain/src/evm-adapter-events.ts index de5be757d5..e1ffed6342 100644 --- a/packages/chain/src/evm-adapter-events.ts +++ b/packages/chain/src/evm-adapter-events.ts @@ -11,7 +11,13 @@ import { EVMChainAdapterBase } from './evm-adapter-base.js'; import { ethers } from 'ethers'; -import type { EventFilter, ChainEvent } from './chain-adapter.js'; +import type { EventFilter, ChainEvent, ChainEventListenOptions } from './chain-adapter.js'; + +const eventScanAborted = (): Error => { + const error = new Error('Chain event scan aborted'); + error.name = 'AbortError'; + return error; +}; export class EventsMethods extends EVMChainAdapterBase { // ===================================================================== @@ -40,19 +46,47 @@ export class EventsMethods extends EVMChainAdapterBase { eventFilter: ethers.ContractEventName, fromBlock: ethers.BlockTag, toBlock?: ethers.BlockTag, + signal?: AbortSignal, ): Promise<(ethers.Log | ethers.EventLog)[]> { - return this.readContractWith( + if (signal?.aborted) return Promise.reject(eventScanAborted()); + + const pending = this.readContractWith( contract, label, (c) => c.queryFilter(eventFilter, fromBlock, toBlock), { policy: 'wideLogScan', skipPreferred: true }, ); + if (!signal) return pending; + + // ethers does not expose AbortSignal on Contract.queryFilter. Reject the + // caller-facing wait immediately while retaining handlers on the provider + // promise, so a later socket close cannot become an unhandled rejection. + return new Promise((resolve, reject) => { + const onAbort = () => { + signal.removeEventListener('abort', onAbort); + reject(eventScanAborted()); + }; + signal.addEventListener('abort', onAbort, { once: true }); + pending.then( + (logs) => { + signal.removeEventListener('abort', onAbort); + resolve(logs); + }, + (error) => { + signal.removeEventListener('abort', onAbort); + reject(error); + }, + ); + }); } - async *listenForEvents(filter: EventFilter): AsyncIterable { + async *listenForEvents(filter: EventFilter, options?: ChainEventListenOptions): AsyncIterable { + if (options?.signal?.aborted) return; await this.init(); + if (options?.signal?.aborted) return; for (const eventType of filter.eventTypes) { + if (options?.signal?.aborted) return; if (eventType === 'KnowledgeBatchCreated') { // V8-only event — emitted by archived KnowledgeAssetsStorage. When the // V8 contract is absent (the V10-only deploy path after this PR), this @@ -63,7 +97,7 @@ export class EventsMethods extends EVMChainAdapterBase { } const eventFilter = storage.filters.KnowledgeBatchCreated(); const logs = await this.queryFilterWithFailover( - storage, 'kasV9.queryFilter(KnowledgeBatchCreated)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, + storage, 'kasV9.queryFilter(KnowledgeBatchCreated)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, options?.signal, ); for (const log of logs) { @@ -92,7 +126,7 @@ export class EventsMethods extends EVMChainAdapterBase { if (cgStorage) { const eventFilter = cgStorage.filters.ContextGraphExpanded(); const logs = await this.queryFilterWithFailover( - cgStorage, 'cgStorage.queryFilter(ContextGraphExpanded)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, + cgStorage, 'cgStorage.queryFilter(ContextGraphExpanded)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, options?.signal, ); for (const log of logs) { @@ -122,7 +156,7 @@ export class EventsMethods extends EVMChainAdapterBase { if (cgStorage) { const eventFilter = cgStorage.filters.KnowledgeAssetRegisteredToContextGraph(); const logs = await this.queryFilterWithFailover( - cgStorage, 'cgStorage.queryFilter(KnowledgeAssetRegisteredToContextGraph)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, + cgStorage, 'cgStorage.queryFilter(KnowledgeAssetRegisteredToContextGraph)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, options?.signal, ); for (const log of logs) { @@ -171,7 +205,7 @@ export class EventsMethods extends EVMChainAdapterBase { const kcFilter = kaStorage.filters[createEventName](); const kcLogs = await this.queryFilterWithFailover( - kaStorage, 'kas.queryFilter(KnowledgeAssetCreated)', kcFilter, fromB, toB, + kaStorage, 'kas.queryFilter(KnowledgeAssetCreated)', kcFilter, fromB, toB, options?.signal, ); // Legacy mint range. `KnowledgeAssetsMinted` is still declared on the @@ -182,7 +216,7 @@ export class EventsMethods extends EVMChainAdapterBase { if (hasEvent('KnowledgeAssetsMinted')) { const mintFilter = kaStorage.filters.KnowledgeAssetsMinted(); const mintLogs = await this.queryFilterWithFailover( - kaStorage, 'kas.queryFilter(KnowledgeAssetsMinted)', mintFilter, fromB, toB, + kaStorage, 'kas.queryFilter(KnowledgeAssetsMinted)', mintFilter, fromB, toB, options?.signal, ); for (const ml of mintLogs) { const mp = kaStorage.interface.parseLog({ topics: [...ml.topics], data: ml.data }); @@ -206,7 +240,7 @@ export class EventsMethods extends EVMChainAdapterBase { try { const transferFilter = kaStorage.filters.Transfer(ethers.ZeroAddress); const transferLogs = await this.queryFilterWithFailover( - kaStorage, 'kas.queryFilter(Transfer)', transferFilter, fromB, toB, + kaStorage, 'kas.queryFilter(Transfer)', transferFilter, fromB, toB, options?.signal, ); for (const tl of transferLogs) { const tp = kaStorage.interface.parseLog({ topics: [...tl.topics], data: tl.data }); @@ -267,7 +301,7 @@ export class EventsMethods extends EVMChainAdapterBase { if (registry) { const eventFilter = registry.filters.NameClaimed(); const logs = await this.queryFilterWithFailover( - registry, 'cgNameRegistry.queryFilter(NameClaimed)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, + registry, 'cgNameRegistry.queryFilter(NameClaimed)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, options?.signal, ); for (const log of logs) { const parsed = registry.interface.parseLog({ topics: [...log.topics], data: log.data }); @@ -292,7 +326,7 @@ export class EventsMethods extends EVMChainAdapterBase { if (cgStorage) { const eventFilter = cgStorage.filters.ContextGraphCreated(); const logs = await this.queryFilterWithFailover( - cgStorage, 'cgStorage.queryFilter(ContextGraphCreated)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, + cgStorage, 'cgStorage.queryFilter(ContextGraphCreated)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, options?.signal, ); for (const log of logs) { const parsed = cgStorage.interface.parseLog({ topics: [...log.topics], data: log.data }); @@ -327,7 +361,7 @@ export class EventsMethods extends EVMChainAdapterBase { if (profileStorage) { const eventFilter = profileStorage.filters.RelayCapabilityUpdated(); const logs = await this.queryFilterWithFailover( - profileStorage, 'profileStorage.queryFilter(RelayCapabilityUpdated)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, + profileStorage, 'profileStorage.queryFilter(RelayCapabilityUpdated)', eventFilter, filter.fromBlock ?? 0, filter.toBlock, options?.signal, ); for (const log of logs) { const parsed = profileStorage.interface.parseLog({ topics: [...log.topics], data: log.data }); diff --git a/packages/chain/src/mock-adapter.ts b/packages/chain/src/mock-adapter.ts index bd67ca9b6a..3a107ad15f 100644 --- a/packages/chain/src/mock-adapter.ts +++ b/packages/chain/src/mock-adapter.ts @@ -518,10 +518,11 @@ export class MockChainAdapter implements ChainAdapter { // --- Events --- - async *listenForEvents(filter: EventFilter): AsyncIterable { + async *listenForEvents(filter: EventFilter, options?: { signal?: AbortSignal }): AsyncIterable { const from = filter.fromBlock ?? 0; const to = filter.toBlock ?? Infinity; for (const evt of this.events) { + if (options?.signal?.aborted) return; if (evt.blockNumber > to) break; if ( evt.blockNumber >= from && diff --git a/packages/chain/src/no-chain-adapter.ts b/packages/chain/src/no-chain-adapter.ts index 75b7933183..398fba30fd 100644 --- a/packages/chain/src/no-chain-adapter.ts +++ b/packages/chain/src/no-chain-adapter.ts @@ -34,7 +34,7 @@ export class NoChainAdapter implements ChainAdapter { async ensureProfile(_options?: { nodeName?: string; stakeAmount?: bigint; lockTier?: number }): Promise { noChain(); } async reserveUALRange(_count: number): Promise { noChain(); } async batchMintKnowledgeAssets(_params: BatchMintParams): Promise { noChain(); } - async *listenForEvents(_filter: EventFilter): AsyncIterable { noChain(); } + async *listenForEvents(_filter: EventFilter, _options?: { signal?: AbortSignal }): AsyncIterable { noChain(); } async createContextGraph(_params: CreateContextGraphParams): Promise { noChain(); } async submitToContextGraph(_kcId: string, _contextGraphId: string): Promise { noChain(); } async revealContextGraphMetadata(_contextGraphId: string, _name: string, _description: string): Promise { noChain(); } diff --git a/packages/chain/test/evm-adapter-tip-read-carveouts.unit.test.ts b/packages/chain/test/evm-adapter-tip-read-carveouts.unit.test.ts index f702b37bdc..f50d327ba4 100644 --- a/packages/chain/test/evm-adapter-tip-read-carveouts.unit.test.ts +++ b/packages/chain/test/evm-adapter-tip-read-carveouts.unit.test.ts @@ -129,6 +129,40 @@ describe('endpoint-stickiness carve-outs: tip-sensitive reads pass skipPreferred expect(readContractWith.calls).toHaveLength(1); expect(readContractWith.calls[0][3]).toMatchObject({ policy: 'wideLogScan', skipPreferred: true }); }); + + it('lets shutdown abandon an in-flight wide-log wait without an unhandled provider rejection', async () => { + const a = makeAdapter(); + a.contracts = { + knowledgeAssetsStorage: { + filters: { KnowledgeBatchCreated: () => ({}) }, + interface: { parseLog: () => null }, + }, + }; + let resolveStarted!: () => void; + const started = new Promise((resolve) => { resolveStarted = resolve; }); + let rejectProvider!: (error: Error) => void; + const providerWait = new Promise((_resolve, reject) => { rejectProvider = reject; }); + a.readContractWith = recorder(() => { + resolveStarted(); + return providerWait; + }); + + const controller = new AbortController(); + const iterator = a.listenForEvents( + { eventTypes: ['KnowledgeBatchCreated'], fromBlock: 0 }, + { signal: controller.signal }, + )[Symbol.asyncIterator](); + const next = iterator.next(); + await started; + controller.abort(); + + await expect(next).rejects.toMatchObject({ name: 'AbortError' }); + // The underlying ethers/provider operation may settle after shutdown. Its + // rejection remains observed by the abort wrapper rather than escaping as + // an unhandled promise. + rejectProvider(new Error('socket closed after adapter teardown')); + await new Promise((resolve) => setTimeout(resolve, 0)); + }); }); // getBlockTimestamp is a CONCRETE receipt-block read (NOT the tip), so it is NOT a diff --git a/packages/cli/src/daemon/lifecycle.ts b/packages/cli/src/daemon/lifecycle.ts index 7a1c16f1d2..cf90d19329 100644 --- a/packages/cli/src/daemon/lifecycle.ts +++ b/packages/cli/src/daemon/lifecycle.ts @@ -3730,6 +3730,13 @@ export async function runDaemonInner( if (shuttingDown) return; shuttingDown = true; log("Shutting down..."); + // Close protocol admission before draining daemon-owned publish/catch-up + // workers. Those workers may be awaiting agent router reads, while + // agent.stop() is intentionally later in the dependency-safe teardown. + // Without this early boundary, the outer daemon and inner agent each wait + // for the other layer to initiate cancellation and the watchdog must force + // exit even though DKGAgent.stop() itself has the correct ordering. + agent.beginStop(); // Tell the supervisor's liveness watcher (PR #664) that this is a graceful // shutdown before any slow cleanup runs. The watcher reads `api.port`'s // absence as "worker is intentionally going down — don't SIGKILL me @@ -3762,35 +3769,47 @@ export async function runDaemonInner( rateLimiter.destroy(); metricsCollector?.stop(); // Stops log exporters AND flushes + shuts down the OTel SDK. + log("[shutdown-stage] telemetry drain starting"); await stopTelemetry(); + log("[shutdown-stage] telemetry drain complete"); natStatusWatcherStop?.(); resetNatStatus(); + log("[shutdown-stage] publisher runtime drain starting"); await publisherState.runtime ?.stop() .catch((err: any) => log(`Publisher runtime stop error: ${err?.message ?? String(err)}`), ); + log("[shutdown-stage] publisher runtime drain complete"); // Drain the async-promote worker before closing the agent — once // `agent.stop()` runs the queue's underlying triple store goes // away. We let in-flight promotes complete (or hit // `shutdownTimeoutMs`); RFC §6.2 forbids marking `running → // queued` here so the next boot's `recoverOnStartup()` decides. + log("[shutdown-stage] promote worker drain starting"); await promoteWorkerLifecycle?.stop(shuttingDown ? 'daemon shutting down' : null); + log("[shutdown-stage] promote worker drain complete"); + log("[shutdown-stage] catch-up runner drain starting"); await daemonState.catchupRunner ?.close() .catch((err: any) => log(`Catch-up runner stop error: ${err?.message ?? String(err)}`), ); + log("[shutdown-stage] catch-up runner drain complete"); server.close(); + log("[shutdown-stage] agent drain starting"); await agent.stop(); + log("[shutdown-stage] agent drain complete"); // Stop the managed Oxigraph child AFTER the agent has stopped // issuing store queries, so an in-flight SPARQL request never // races the killed server. No-op when not using oxigraph-server. + log("[shutdown-stage] managed store drain starting"); await managedOxigraph ?.stop() .catch((err: any) => log(`Managed Oxigraph stop error: ${err?.message ?? String(err)}`), ); + log("[shutdown-stage] managed store drain complete"); dashDb.close(); log("Stopped."); } finally { diff --git a/packages/core/src/node.ts b/packages/core/src/node.ts index d954bec934..569d6111c1 100644 --- a/packages/core/src/node.ts +++ b/packages/core/src/node.ts @@ -705,6 +705,22 @@ export class DKGNode { return this.stopAbortController?.signal; } + /** + * Close network admission before a higher-level owner starts awaiting its + * own teardown dependencies. + * + * DKGAgent has cleanup work (notably the chain-event poller) that must run + * before libp2p itself is stopped. Some of those callbacks can already be + * waiting on ProtocolRouter reads. If the stop signal only fires inside + * {@link stop}, those reads keep the earlier cleanup step alive and the + * daemon eventually reaches its forced-shutdown watchdog. Aborting here is + * idempotent and leaves the node available for the remaining close calls; + * {@link stop} still owns the actual libp2p teardown. + */ + beginStop(): void { + this.stopAbortController?.abort(); + } + constructor(config: DKGNodeConfig = {}) { this.config = config; } @@ -1867,7 +1883,7 @@ export class DKGNode { // closing — doesn't deadlock waiting for those reads to finish. // Aborting + then awaiting libp2p.stop() is the graceful // counterpart to PR-1's hard-timeout safety net (#655). - this.stopAbortController?.abort(); + this.beginStop(); if (this.relayWatchdogTimer) { clearTimeout(this.relayWatchdogTimer); this.relayWatchdogTimer = null; diff --git a/packages/core/test/node-dht-wiring.test.ts b/packages/core/test/node-dht-wiring.test.ts index 7ac5d931be..36e19a077e 100644 --- a/packages/core/test/node-dht-wiring.test.ts +++ b/packages/core/test/node-dht-wiring.test.ts @@ -59,6 +59,26 @@ describe('DKGNode DHT network identity wiring', () => { await node.stop(); }); + it('can abort network admission before the actual libp2p stop', async () => { + const { DKGNode } = await import('../src/node.js'); + const node = new DKGNode({ + listenAddresses: ['/ip4/127.0.0.1/tcp/0'], + enableMdns: false, + }); + + await node.start(); + expect(node.stopSignal?.aborted).toBe(false); + + node.beginStop(); + expect(node.stopSignal?.aborted).toBe(true); + // Admission close is deliberately idempotent because both DKGAgent.stop + // and DKGNode.stop own a defensive call. + node.beginStop(); + + await node.stop(); + expect(node.stopSignal).toBeUndefined(); + }); + it('passes the active relay network gater into libp2p during start', async () => { const { DKGNode } = await import('../src/node.js'); const node = new DKGNode({ diff --git a/packages/publisher/src/chain-event-lane-runner.ts b/packages/publisher/src/chain-event-lane-runner.ts index 844e89a391..7b40c439eb 100644 --- a/packages/publisher/src/chain-event-lane-runner.ts +++ b/packages/publisher/src/chain-event-lane-runner.ts @@ -35,7 +35,7 @@ export interface ChainEventPollerLaneSpec { canUseLegacyAggregateCursor?(): boolean; liveSeedLookbackBlocks?: number; cadenceMs: number; - dispatch(event: ChainEvent, ctx: OperationContext): Promise; + dispatch(event: ChainEvent, ctx: OperationContext, signal?: AbortSignal): Promise; onBackfillFromGenesis?(ctx: OperationContext): void; } @@ -95,12 +95,14 @@ export class ChainEventLaneRunner { await this.restoreLaneCursors(this.activeLaneSpecs(), ctx); } - async poll(): Promise { + async poll(signal?: AbortSignal): Promise { + if (signal?.aborted) return; const ctx = createOperationContext('publish'); const activeLanes = this.activeLaneSpecs(); if (activeLanes.length === 0) return; await this.restoreLaneCursors(activeLanes, ctx); + if (signal?.aborted) return; const now = this.clock(); const dueLanes = activeLanes.filter((lane) => this.laneDue(lane, now)); @@ -110,10 +112,12 @@ export class ChainEventLaneRunner { if (this.chain.getBlockNumber) { try { head = await this.chain.getBlockNumber(); } catch { /* unavailable */ } } + if (signal?.aborted) return; const scanResults: ChainEventPollerLaneScanResult[] = []; for (const lane of dueLanes) { - scanResults.push(await this.scanLane(lane, head, now, ctx)); + if (signal?.aborted) break; + scanResults.push(await this.scanLane(lane, head, now, ctx, signal)); } await this.persistScanResults(scanResults, activeLanes); } @@ -235,6 +239,7 @@ export class ChainEventLaneRunner { head: number | undefined, now: number, ctx: OperationContext, + signal?: AbortSignal, ): Promise { const state = lane.state; @@ -269,14 +274,27 @@ export class ChainEventLaneRunner { let advanced = false; try { - for await (const event of this.chain.listenForEvents(filter)) { - await lane.spec.dispatch(event, ctx); + for await (const event of this.chain.listenForEvents(filter, { signal })) { + // A stopped partial lane must be replayed from its prior durable + // cursor. Event callbacks are idempotent, so replay is safer than + // advancing past events that were never dispatched. + if (signal?.aborted) { + return { lane, blockNumber: state.lastBlock, advanced: false }; + } + await lane.spec.dispatch(event, ctx, signal); + } + + if (signal?.aborted) { + return { lane, blockNumber: state.lastBlock, advanced: false }; } state.lastBlock = upperBound; advanced = true; this.applyLaneSchedule(lane, { kind: 'success', now, caughtUp }); } catch (err) { + if (signal?.aborted) { + return { lane, blockNumber: state.lastBlock, advanced: false }; + } this.log.error(ctx, `Poll lane ${lane.spec.name} failed: ${err instanceof Error ? err.message : String(err)}`); this.applyLaneSchedule(lane, { kind: 'failure', now }); } diff --git a/packages/publisher/src/chain-event-poller.ts b/packages/publisher/src/chain-event-poller.ts index 83cd3b63f5..1bcc12ac1c 100644 --- a/packages/publisher/src/chain-event-poller.ts +++ b/packages/publisher/src/chain-event-poller.ts @@ -67,7 +67,7 @@ export type OnKARegisteredToContextGraph = (info: { txHash: string; txIndex?: number; blockNumber: number; -}) => Promise; +}, signal?: AbortSignal) => Promise; /** * Callback for `KnowledgeAssetCreated` events — OT-RFC-43 Option-1 allocator @@ -133,6 +133,7 @@ export class ChainEventPoller { private readonly log = new Logger('ChainEventPoller'); private timer: ReturnType | null = null; private running = false; + private stopController: AbortController | null = null; /** * The currently-executing `poll()` promise (or `null` when idle). * @@ -172,6 +173,8 @@ export class ChainEventPoller { async start(): Promise { if (this.running) return; this.running = true; + const stopController = new AbortController(); + this.stopController = stopController; const ctx = createOperationContext('system'); @@ -192,7 +195,7 @@ export class ChainEventPoller { // chain is monotonic and the poll catches up via `MAX_RANGE`, so a // skipped tick is functionally identical to slightly longer cadence. if (this.inFlightPoll) return; - this.inFlightPoll = this.poll() + this.inFlightPoll = this.poll(this.stopController?.signal) .catch((err) => { const pollCtx = createOperationContext('system'); this.log.error(pollCtx, `Poll failed: ${err instanceof Error ? err.message : String(err)}`); @@ -201,7 +204,7 @@ export class ChainEventPoller { }, this.intervalMs); // Run first poll immediately, and track it so `stop()` can await it. - this.inFlightPoll = this.poll() + this.inFlightPoll = this.poll(stopController.signal) .catch(() => {}) .finally(() => { this.inFlightPoll = null; }); } @@ -232,6 +235,11 @@ export class ChainEventPoller { this.timer = null; } this.running = false; + // Stop after the current event callback and do not start another lane or + // dispatch another event from a large historical page. The current lane's + // cursor deliberately remains unchanged so any partial page is replayed + // safely on restart. + this.stopController?.abort(); const pending = this.inFlightPoll; if (pending) { // The `.catch(() => {})` chain at the call sites already swallows @@ -241,6 +249,7 @@ export class ChainEventPoller { // null out `inFlightPoll` once the await unblocks. try { await pending; } catch { /* already logged or swallowed */ } } + this.stopController = null; const ctx = createOperationContext('system'); this.log.info(ctx, 'Chain event poller stopped'); @@ -292,7 +301,7 @@ export class ChainEventPoller { eventTypes: () => ['KnowledgeAssetRegisteredToContextGraph'], requiresFullHistory: () => false, cadenceMs: this.intervalMs, - dispatch: (event, ctx) => this.handleKARegistered(event, ctx), + dispatch: (event, ctx, signal) => this.handleKARegistered(event, ctx, signal), }, { name: 'collectionUpdates', @@ -321,8 +330,8 @@ export class ChainEventPoller { ]; } - private async poll(): Promise { - await this.laneRunner.poll(); + private async poll(signal?: AbortSignal): Promise { + await this.laneRunner.poll(signal); } private async handleBatchCreated(event: ChainEvent, ctx: OperationContext): Promise { @@ -446,7 +455,11 @@ export class ChainEventPoller { } } - private async handleKARegistered(event: ChainEvent, ctx: OperationContext): Promise { + private async handleKARegistered( + event: ChainEvent, + ctx: OperationContext, + signal?: AbortSignal, + ): Promise { if (!this.onKARegisteredToContextGraph) return; const { data } = event; const contextGraphId = String(data['contextGraphId'] ?? ''); @@ -470,7 +483,7 @@ export class ChainEventPoller { txHash, txIndex, blockNumber: event.blockNumber, - }); + }, signal); } catch (err) { this.log.warn(ctx, `onKARegisteredToContextGraph callback failed: ${err instanceof Error ? err.message : String(err)}`); } diff --git a/packages/publisher/test/chain-event-lane-runner.unit.test.ts b/packages/publisher/test/chain-event-lane-runner.unit.test.ts index 472b3e9172..8c390c23d0 100644 --- a/packages/publisher/test/chain-event-lane-runner.unit.test.ts +++ b/packages/publisher/test/chain-event-lane-runner.unit.test.ts @@ -97,6 +97,53 @@ describe('ChainEventPoller lane runner and cursors', () => { expect(filters).toEqual([]); }); + it('stops a large event page after the active callback and replays the partial lane', async () => { + const events: ChainEvent[] = [1, 2, 3].map((id) => ({ + type: 'ContextGraphCreated', + blockNumber: 9_500 + id, + data: { + contextGraphId: String(id), + creator: '0x' + 'a1'.repeat(20), + accessPolicy: 0, + publishPolicy: 1, + nameHash: '0x' + id.toString(16).padStart(64, '0'), + }, + })); + const { adapter } = makeChain(10_000, events); + const seen: string[] = []; + let stopRequested = false; + let resolveStopStarted: () => void = () => {}; + const stopStarted = new Promise((resolve) => { resolveStopStarted = resolve; }); + let stopPromise: Promise | undefined; + let poller!: ChainEventPoller; + poller = new ChainEventPoller({ + chain: adapter, + publishHandler: makeHandler(), + intervalMs: 60_000, + onContextGraphCreated: async ({ contextGraphId }) => { + seen.push(contextGraphId); + if (!stopRequested) { + stopRequested = true; + stopPromise = poller.stop(); + resolveStopStarted(); + } + }, + }); + + await poller.start(); + await stopStarted; + await stopPromise; + expect(seen).toEqual(['1']); + + // The interrupted page never advanced its cursor. A restart therefore + // replays the first idempotent event and then processes the rest rather + // than skipping events 2 and 3. + await poller.start(); + await new Promise((resolve) => setTimeout(resolve, 20)); + await poller.stop(); + expect(seen).toEqual(['1', '1', '2', '3']); + }); + it('cold-starts a restored pending publish lane from block 0 without allocator callbacks', async () => { const merkleRoot = '0x' + '55'.repeat(32); const oldCreate: ChainEvent = {