Skip to content

fix(sync): cancel chain event work during shutdown #2044

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
branarakic wants to merge 2 commits into
base: codex/rfc64-m1-cold-historical-binding
Choose a base branch
from
Open

fix(sync): cancel chain event work during shutdown #2044

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8fae889
fix(sync): cancel chain event work during shutdown
Aug 3, 2026
9d8c738
test(sync): model early shutdown admission close
Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions packages/agent/src/dkg-agent-lifecycle.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

@otReviewAgent otReviewAgent Aug 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Issue: The KACG abort signal wiring is not covered end to end

What's wrong
The new cancellation behavior depends on the signal being passed through multiple call boundaries. The added abort test verifies the handler when called directly, but not the production wiring that supplies the signal to that handler.

Example
A regression that changes await this.handleKARegisteredNudge(onChainId, kaId, ctx, signal) back to await this.handleKARegisteredNudge(onChainId, kaId, ctx) would leave the direct nudge abort test green, while production shutdown could still wait on the unresolved CG resolver.

Suggested direction
Add a regression test that drives a KnowledgeAssetRegisteredToContextGraph event through the real poller callback boundary and asserts the signal reaches handleKARegisteredNudge and becomes aborted during stop().

For Agents
Add a wiring-level test around ChainEventPoller plus the agent lifecycle callback, or a focused lifecycle test that stubs handleKARegisteredNudge and proves the callback receives the poller's stop signal. Preserve the existing payload behavior and assert that stopping the poller aborts the same signal observed by the nudge handler.

}
: undefined,
});
Expand Down
11 changes: 8 additions & 3 deletions packages/agent/src/dkg-agent-swm-host.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2636,17 +2636,19 @@ export class SwmHostModeMethods extends DKGAgentBase {
localCgId: string,
sub: ContextGraphSub,
targetOnChainId?: bigint,
signal?: AbortSignal,
): Promise<string | null> {
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; }
Expand Down Expand Up @@ -2680,7 +2682,9 @@ export class SwmHostModeMethods extends DKGAgentBase {
onChainId: string,
kaId: bigint,
ctx: OperationContext,
signal?: AbortSignal,
): Promise<string | null> {
if (signal?.aborted) return null;
let targetOnChain: bigint | null = null;
try { targetOnChain = BigInt(onChainId); } catch { targetOnChain = null; }

Expand All @@ -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);
Expand Down
18 changes: 18 additions & 0 deletions packages/agent/src/dkg-agent.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1720,8 +1720,26 @@
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();

Check failure on line 1731 in packages/agent/src/dkg-agent.ts

View workflow job for this annotation

GitHub Actions / SQLite lifecycle (Windows)

test/rfc64-agent-inventory-lifecycle.test.ts > DKGAgent RFC-64 inventory lifecycle > closes after network consumers and before the triple store, with idempotent stop 

TypeError: this.node.beginStop is not a function
 ❯ DKGAgent.beginStop src/dkg-agent.ts:1731:15
 ❯ DKGAgent.stop src/dkg-agent.ts:1742:10
 ❯ test/rfc64-agent-inventory-lifecycle.test.ts:539:17

Check failure on line 1731 in packages/agent/src/dkg-agent.ts

View workflow job for this annotation

GitHub Actions / SQLite lifecycle (Windows)

Unhandled error 

TypeError: this.node.beginStop is not a function
 ❯ DKGAgent.beginStop src/dkg-agent.ts:1731:15
 ❯ DKGAgent.stop src/dkg-agent.ts:1742:10
 ❯ test/rfc64-agent-inventory-lifecycle.test.ts:603:24

This error originated in "test/rfc64-agent-inventory-lifecycle.test.ts" test file. It doesn't mean the error was thrown inside the file itself, but while it was running.
The latest test that might've caused the error is "keeps shutdown active and the triple store open until persistence drain settles". It might mean one of the following:
- The error was thrown, while Vitest was running this test.
- If the error occurred after the test had been completed, this was the last documented test before it was thrown.

Check failure on line 1731 in packages/agent/src/dkg-agent.ts

View workflow job for this annotation

GitHub Actions / SQLite lifecycle (Windows)

Unhandled error 

TypeError: this.node.beginStop is not a function
 ❯ DKGAgent.beginStop src/dkg-agent.ts:1731:15
 ❯ DKGAgent.stop src/dkg-agent.ts:1742:10
 ❯ test/rfc64-agent-inventory-lifecycle.test.ts:570:24

This error originated in "test/rfc64-agent-inventory-lifecycle.test.ts" test file. It doesn't mean the error was thrown inside the file itself, but while it was running.
The latest test that might've caused the error is "awaits finalization inbox drain before RFC-64 persistence and store teardown". It might mean one of the following:
- The error was thrown, while Vitest was running this test.
- If the error occurred after the test had been completed, this was the last documented test before it was thrown.
}

async stop(): Promise<void> {
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
Expand Down
53 changes: 49 additions & 4 deletions packages/agent/test/outbox-shutdown-lifecycle.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<void>((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;

@otReviewAgent otReviewAgent Aug 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Issue: Extract a shutdown-test fixture instead of adding another raw prototype stub

What's wrong
The new test repeats the full manual agent shape, and the surrounding diff shows why that structure is brittle: adding beginStop() and stopSampling() forced unrelated tests to grow more stub fields. This is maintainability debt in a lifecycle area that is already large and order-sensitive; the tests are becoming a mirror of private implementation details rather than a small harness for shutdown behavior.

Example
A narrow helper such as makeStartedStopAgent(overrides) could own the default started, timers, runtime, node, router, store, messenger, and logger stubs. Each test would override only the dependency it is asserting, such as chainPoller.stop or messenger.stopOutboxDrain.

Suggested direction
Create a local factory for the minimal started-agent shutdown harness and let individual tests pass overrides. That keeps these tests focused on the ordering they care about and prevents every new shutdown dependency from creating broad, repetitive fixture churn.

Confidence note
This is a maintainability finding about the new test structure, not about runtime behavior.

For Agents
In packages/agent/test/outbox-shutdown-lifecycle.test.ts, extract a local fixture/factory for the minimal started agent used by shutdown tests. Preserve each test's ordering assertions, but move default stop dependencies into the helper so future lifecycle fields do not require touching every case.

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;
Expand All @@ -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,
Expand All @@ -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() },
Expand Down Expand Up @@ -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,
Expand All @@ -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 },
Expand Down Expand Up @@ -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 () => {}),
Expand All @@ -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() },
Expand All @@ -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 () => {}),
Expand All @@ -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 },
Expand Down
53 changes: 52 additions & 1 deletion packages/agent/test/vm-reconcile-self-prime.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,13 @@ interface AgentInternals {
localCgId: string,
sub: { subscribed: boolean; coreHosted?: boolean; onChainId?: string },
targetOnChainId?: bigint,
signal?: AbortSignal,
): Promise<string | null>;
handleKARegisteredNudge(onChainId: string, kaId: bigint, ctx: unknown, signal?: AbortSignal): Promise<string | null>;
getContextGraphOnChainId(
localCgId: string,
options?: { signal?: AbortSignal; source?: string },
): Promise<string | null>;
handleKARegisteredNudge(onChainId: string, kaId: bigint, ctx: unknown): Promise<string | null>;
subscribedContextGraphs: Map<string, { subscribed: boolean; coreHosted?: boolean; onChainId?: string }>;
vmReconcileDispatcher: {
dispatch: (cg: string, reason: 'live' | 'periodic') => Promise<boolean>;
Expand Down Expand Up @@ -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<void>((resolve) => { resolverStarted = resolve; });
internals.getContextGraphOnChainId = async (_id, options = {}) => {
resolverStarted();
await new Promise<void>((_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([]);
});
});
7 changes: 6 additions & 1 deletion packages/chain/src/chain-adapter.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -1055,7 +1060,7 @@ export interface ChainAdapter {
getBlockNumber?(): Promise<number>;

// Events
listenForEvents(filter: EventFilter): AsyncIterable<ChainEvent>;
listenForEvents(filter: EventFilter, options?: ChainEventListenOptions): AsyncIterable<ChainEvent>;

// Context Graphs (name-hash commitment via ContextGraphNameRegistry)
createContextGraph(params: CreateContextGraphParams): Promise<TxResult>;
Expand Down
Loading
Loading