diff --git a/.changeset/messaging-dispatchers-stop-on-shutdown.md b/.changeset/messaging-dispatchers-stop-on-shutdown.md new file mode 100644 index 0000000000..c519aa3878 --- /dev/null +++ b/.changeset/messaging-dispatchers-stop-on-shutdown.md @@ -0,0 +1,13 @@ +--- +"@objectstack/service-messaging": patch +--- + +**Fix:** `MessagingServicePlugin` now releases its delivery dispatchers on `kernel.shutdown()`. Previously they kept running after shutdown had resolved (#9371). + +The plugin starts two `setInterval` dispatchers at `kernel:ready` — `NotificationDispatcher` over `sys_notification_delivery` and `HttpDispatcher` over `sys_http_delivery` — and released them from a method named `stop()`. The kernel's plugin teardown hook is `destroy()` (`Plugin.destroy?()` in `@objectstack/core`; the only teardown `ObjectKernel.performShutdown()` and `LiteKernel.destroy()` invoke), and `stop()` is not on that interface, so **nothing ever called it**. Both dispatchers went on claiming and updating delivery rows after `await kernel.shutdown()` returned. Measured on the new pin: 48 further delivery reads/writes in the 80 ms following a resolved shutdown. + +The teardown body now lives on `destroy()`. `stop()` is **retained as an alias** — it is public API of an exported class, and an embedder may well have learned to call it directly precisely because the kernel never did. No call site has to change, and no accept/reject behaviour of any contract moves. + +**Why it was invisible in production, and where the bill landed.** `start()` `unref()`s both timers, so a long-lived host process still exits and the leak is silent. Under vitest the worker process is alive throughout teardown, so a tick fires *after* a test file is over, reads a delivery table through a driver the suite already disconnected, and `SqlDriver`'s console fallback warns. `console.*` inside a vitest worker is an RPC to the main process (`onUserConsoleLog`); one issued after `rpcDone()` has snapshotted the pending set is rejected by `$rejectPendingCalls` as `EnvironmentTeardownError: [vitest-worker]: Closing rpc while "onUserConsoleLog" was pending`. Nothing awaits that promise, so it lands as an unhandled rejection and fails a run in which every test passed — twice measured on `examples/app-showcase` (334/334 and 337/337 green, exit 1, a merge-queue eviction each time). The width of the window is the duration of `rpcDone()`, which is why it only ever fired on a loaded queue runner and never on the PR-side run of the identical diff. + +Suites that boot a kernel with this plugin get quieter and finish cleaner as a result: over 48 loaded runs of the affected showcase file, console output emitted after the file's own `afterAll` went 3 → 0, and console RPC round-trips per run roughly halved (6574 → 3456 in aggregate). diff --git a/packages/services/service-messaging/src/messaging-service-plugin.ts b/packages/services/service-messaging/src/messaging-service-plugin.ts index 3317a3dfc9..c8369b4d42 100644 --- a/packages/services/service-messaging/src/messaging-service-plugin.ts +++ b/packages/services/service-messaging/src/messaging-service-plugin.ts @@ -378,11 +378,52 @@ export class MessagingServicePlugin implements Plugin { } } - /** Stop the dispatcher loop + retention sweep on shutdown. */ - async stop(): Promise { + /** + * The kernel's teardown hook (`Plugin.destroy`, core `types.ts`) — the ONLY + * teardown entry point `ObjectKernel.performShutdown()` and + * `LiteKernel.destroy()` invoke. + * + * [#9371] IT USED TO BE `stop()`, WHICH NOTHING CALLED. The method below + * carried the docblock "Stop the dispatcher loop … on shutdown" and was + * correct in every respect except its NAME: `Plugin` declares `destroy?()` + * and no `stop()`, so the kernel walked past this plugin at shutdown and + * both `setInterval` dispatchers went on ticking after `await + * kernel.shutdown()` resolved. `start()` `unref()`s the timers, so a + * long-lived process still EXITS — which is exactly why this stayed + * invisible in production and surfaced somewhere else entirely. + * + * WHERE IT SURFACED. Under vitest the worker process is very much alive + * during teardown, so a tick fires after the test file is over, reads + * `sys_notification_delivery` / `sys_http_delivery` through a driver the + * suite already disconnected, and `SqlDriver`'s console fallback warns. + * `console.*` inside a worker is an RPC to the main process + * (`onUserConsoleLog`); one created after `rpcDone()` has taken its + * snapshot is rejected by vitest's `$rejectPendingCalls` as + * `EnvironmentTeardownError: [vitest-worker]: Closing rpc while + * "onUserConsoleLog" was pending`. Nobody awaits that promise, so it lands + * as an UNHANDLED REJECTION and fails the run — with every test passing. + * That is #9371: `examples/app-showcase` green at 334/334 and 337/337, exit + * 1, a merge-queue eviction each time. The window is the duration of + * `rpcDone()`, which is why it only ever fired on a loaded queue runner and + * never on the PR-side run of the identical diff. + * + * So this is a teardown-contract fix, not a test fix: a plugin that owns + * timers must release them on the hook the kernel actually calls. + */ + async destroy(): Promise { await this.dispatcher?.stop(); this.dispatcher = undefined; await this.httpDispatcher?.stop(); this.httpDispatcher = undefined; } + + /** + * Retained alias for {@link destroy}. Kept because it is public API of an + * exported class and removing it would break any embedder that learned to + * call it precisely BECAUSE the kernel never did. Callers should prefer + * kernel shutdown; direct callers keep working. + */ + async stop(): Promise { + await this.destroy(); + } } diff --git a/packages/services/service-messaging/src/plugin-shutdown-stops-dispatchers.test.ts b/packages/services/service-messaging/src/plugin-shutdown-stops-dispatchers.test.ts new file mode 100644 index 0000000000..8906a8b5e7 --- /dev/null +++ b/packages/services/service-messaging/src/plugin-shutdown-stops-dispatchers.test.ts @@ -0,0 +1,146 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#9371] `await kernel.shutdown()` must leave NOTHING of this plugin still + * touching the database. + * + * THE DEFECT. `MessagingServicePlugin` starts two `setInterval` dispatchers at + * `kernel:ready` — `NotificationDispatcher` over `sys_notification_delivery` + * and `HttpDispatcher` over `sys_http_delivery`. The teardown that stops them + * was spelled `stop()`. The kernel's plugin teardown hook is `destroy()` + * (`Plugin.destroy?()` in `@objectstack/core`'s `types.ts`; the only teardown + * `ObjectKernel.performShutdown()` and `LiteKernel.destroy()` invoke), and + * `stop()` is not part of that interface — so nothing in the repo ever called + * it and both dispatchers went on ticking after shutdown had resolved. + * + * WHY IT WENT UNNOTICED, AND WHERE THE BILL LANDED. `start()` `unref()`s both + * timers, so a long-lived host process still exits and the leak is silent in + * production. Under vitest the worker process is alive throughout teardown, so + * a tick fires AFTER the test file is over, reads a delivery table through a + * driver the suite has already disconnected, and `SqlDriver`'s console + * fallback warns. `console.*` inside a worker is an RPC to the main process + * (`onUserConsoleLog`); one issued after `rpcDone()` has snapshotted the + * pending set is rejected by `$rejectPendingCalls` as `EnvironmentTeardownError: + * [vitest-worker]: Closing rpc while "onUserConsoleLog" was pending`. Nothing + * awaits that promise, so it surfaces as an unhandled rejection and fails a run + * in which every test passed — twice measured on `examples/app-showcase` + * (334/334 and 337/337 green, exit 1, a merge-queue eviction each time). + * + * WHAT THIS PINS, AND WHY IN THIS SHAPE. The assertion is behavioural — "after + * shutdown the plugin issues no further delivery reads/writes" — and not + * `expect(plugin.destroy).toBeDefined()`, because the hook merely EXISTING is + * not the property that was missing; being reached by the kernel is. The + * counter is installed on the very `IDataEngine` the dispatchers captured, so + * it observes the production call path (`outbox.claim()` → `engine.update()` / + * `engine.find()`) rather than a stand-in. + * + * The pre-shutdown leg is a POSITIVE CONTROL and load-bearing: without it a + * dispatcher that never started would satisfy the post-shutdown assertion + * vacuously, and this test would pass on a plugin that does nothing at all. + */ + +import { describe, it, expect, afterEach } from 'vitest'; +import { ObjectKernel } from '@objectstack/core'; +import { ObjectQLPlugin } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +import type { ObjectQL } from '@objectstack/objectql'; +import type { IDataEngine } from '@objectstack/spec/contracts'; +import { MessagingServicePlugin } from './messaging-service-plugin.js'; + +/** Fast enough that a handful of ticks fit in a short window; still real time. */ +const TICK_MS = 10; +/** ~8 ticks. Wide enough that "no reads" cannot be a scheduling coincidence. */ +const OBSERVE_MS = 80; + +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); + +const openKernels: ObjectKernel[] = []; +const openDrivers: Array<{ disconnect?: () => Promise }> = []; + +afterEach(async () => { + // Kernels first, drivers second: the kernel's own teardown still wants a + // live driver to drain against. (Reversing these is what makes a suite + // shout DATABASE_ERROR at teardown time.) + while (openKernels.length) { + try { await openKernels.pop()?.shutdown(); } catch { /* already stopped */ } + } + while (openDrivers.length) { + try { await openDrivers.pop()?.disconnect?.(); } catch { /* noop */ } + } +}); + +interface Booted { + kernel: ObjectKernel; + /** Reads/writes the dispatchers have made against the delivery tables. */ + deliveryCalls: () => number; +} + +async function bootMessagingKernel(): Promise { + const kernel = new ObjectKernel({ logger: { level: 'silent' } } as any); + openKernels.push(kernel); + + await kernel.use(new ObjectQLPlugin()); + await kernel.use( + new MessagingServicePlugin({ dispatchIntervalMs: TICK_MS, partitionCount: 1 }), + ); + await kernel.bootstrap(); + + const objectql = kernel.getService('objectql'); + const driver: any = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: ':memory:' }, + useNullAsDefault: true, + }); + await driver.connect(); + objectql.registerDriver(driver, true); + openDrivers.push(driver); + await objectql.syncSchemas(); + + // Count on the engine instance the dispatchers captured at `kernel:ready` + // (`getData()` resolves the `data` service, which is this same object), so + // the tally is of real `outbox.claim()` traffic. + type EngineCall = (name: string, ...rest: unknown[]) => unknown; + const engine = kernel.getService('data') as unknown as + Record<'find' | 'findOne' | 'update', EngineCall>; + let calls = 0; + for (const method of ['find', 'findOne', 'update'] as const) { + const orig = engine[method].bind(engine); + engine[method] = (name: string, ...rest: unknown[]) => { + if (String(name).includes('_delivery')) calls++; + return orig(name, ...rest); + }; + } + + return { kernel, deliveryCalls: () => calls }; +} + +describe('#9371 MessagingServicePlugin releases its dispatchers on kernel shutdown', () => { + it('stops touching the delivery tables once shutdown() has resolved', async () => { + const { kernel, deliveryCalls } = await bootMessagingKernel(); + + // POSITIVE CONTROL — the dispatchers really are running, so the + // post-shutdown assertion below is measuring silence and not absence. + await sleep(OBSERVE_MS); + expect(deliveryCalls()).toBeGreaterThan(0); + + await kernel.shutdown(); + + const atShutdown = deliveryCalls(); + await sleep(OBSERVE_MS); + + // The contract: shutdown() resolving means the plugin is done with the + // database. Before the fix both dispatchers kept ticking here and this + // count kept climbing. + expect(deliveryCalls()).toBe(atShutdown); + }); + + it('a second shutdown is a no-op rather than a throw', async () => { + const { kernel } = await bootMessagingKernel(); + await sleep(TICK_MS * 2); + await kernel.shutdown(); + // Idempotence matters because `destroy()` nulls the handles it stopped; + // a teardown that only works once is a teardown that fails in a suite. + const plugin = new MessagingServicePlugin(); + await expect(plugin.destroy()).resolves.toBeUndefined(); + }); +}); diff --git a/packages/services/service-messaging/vitest.config.ts b/packages/services/service-messaging/vitest.config.ts new file mode 100644 index 0000000000..58bc9ec2e3 --- /dev/null +++ b/packages/services/service-messaging/vitest.config.ts @@ -0,0 +1,26 @@ +import { defineConfig } from 'vitest/config'; +import path from 'node:path'; + +/** + * [#9371] `plugin-shutdown-stops-dispatchers.test.ts` asserts a property of the + * KERNEL's teardown contract — that `ObjectKernel.performShutdown()` reaches + * this plugin's `destroy()` — so the `@objectstack/core` it runs against has to + * be the source in this checkout. Unaliased, the workspace link resolves it to + * `dist/`, and the verdict becomes a function of build state: a `dist` merely + * BEHIND would run that test GREEN against a `performShutdown` that no longer + * matches the one shipping, which is precisely the reading it exists to pin. + * `pnpm check:test-source-alias` is the gate. + * + * ANCHORED regex, array form, deliberately: a bare string `find` matches by + * PREFIX, so with a FILE replacement it would also swallow any subpath and + * resolve it to `…/core/src/index.ts/` — `ENOTDIR` at run time, from a + * config that reads as correct. Mirrors the identical rule in + * `examples/app-showcase/vitest.config.ts`. + */ +export default defineConfig({ + resolve: { + alias: [ + { find: /^@objectstack\/core$/, replacement: path.resolve(__dirname, '../../core/src/index.ts') }, + ], + }, +});