mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-10-19 05:26:47 +00:00
Maurice
724c88c201
A mechanical prettier --write over server/src, server/tests, server/scripts and shared/src with the workspaces' .prettierrc; nothing but layout and import order changes, and every file keeps the same set of imports. shared is formatted throughout and its format baseline is empty. The server baseline keeps the 35 files that Prettier would grow past their lint:size limit (they need a split first, which is not a formatting change), and the generated entity files stay as their generator writes them.
426 lines
19 KiB
TypeScript
426 lines
19 KiB
TypeScript
/**
|
|
* Supervisor lifecycle edge cases (#plugins). These drive the in-memory state
|
|
* machine directly (spawn stubbed) — no child fork — to prove the recovery paths
|
|
* an admin depends on: re-activating a plugin that died stays possible, and a
|
|
* crash-restart cycle doesn't leak the dead child's cron tasks.
|
|
*/
|
|
import { RpcRateLimiter, TokenBucket } from '../../../src/nest/plugins/host/rate-limit';
|
|
import { PluginSupervisor } from '../../../src/nest/plugins/supervisor/plugin-supervisor';
|
|
import { createTestOrm, type TestOrm } from '../../helpers/test-orm';
|
|
|
|
import Database from 'better-sqlite3';
|
|
import { describe, it, expect, vi, beforeAll, afterAll, afterEach } from 'vitest';
|
|
|
|
// task-6-fix-brief.md item 1: onMessage's 'req' branch now THROWS without a
|
|
// resolveOrm thunk (no more silent unwrapped dispatch), so the one case below
|
|
// that dispatches a real 'req' message ("refuses a throttled call...") needs a
|
|
// real ORM even though nothing it exercises reaches a repository.
|
|
const ormDb = new Database(':memory:');
|
|
let ormHandle: TestOrm;
|
|
beforeAll(async () => {
|
|
ormHandle = await createTestOrm(ormDb);
|
|
});
|
|
afterAll(async () => {
|
|
await ormHandle.close();
|
|
ormDb.close();
|
|
});
|
|
|
|
function makeSupervisor() {
|
|
const dispose = vi.fn();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const s = new PluginSupervisor((() => ({ dispose })) as any, {}, {}, () => ormHandle.orm);
|
|
const spawn = vi.fn();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).spawn = spawn;
|
|
return { s, spawn, dispose };
|
|
}
|
|
// Reaches into the private `running` map so a case can assert on the live entry.
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const entry = (s: PluginSupervisor, id: string): { status: string } => (s as any).running.get(id);
|
|
|
|
describe('supervisor re-activation after failure', () => {
|
|
const supers: PluginSupervisor[] = [];
|
|
afterEach(() => {
|
|
for (const s of supers) {
|
|
void s.shutdownAll().catch(() => {});
|
|
}
|
|
supers.length = 0;
|
|
});
|
|
|
|
it('re-activating a plugin left in error state re-spawns instead of silently no-op-ing', () => {
|
|
const { s, spawn } = makeSupervisor();
|
|
supers.push(s);
|
|
// A prior crash-auto-disable / load-error leaves a dead entry in running.
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', {
|
|
id: 'p',
|
|
status: 'error',
|
|
jobTasks: undefined,
|
|
pending: new Map(),
|
|
invocations: new Map(),
|
|
crashes: [],
|
|
});
|
|
|
|
void s.activate('p', new Set());
|
|
|
|
expect(spawn).toHaveBeenCalledTimes(1); // it re-spawned, not no-op
|
|
expect(entry(s, 'p').status).toBe('starting'); // fresh entry replaced the dead one
|
|
});
|
|
|
|
it('re-activating a LIVE plugin is an idempotent no-op', () => {
|
|
const { s, spawn } = makeSupervisor();
|
|
supers.push(s);
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', {
|
|
id: 'p',
|
|
status: 'active',
|
|
jobTasks: undefined,
|
|
pending: new Map(),
|
|
invocations: new Map(),
|
|
crashes: [],
|
|
});
|
|
|
|
void s.activate('p', new Set());
|
|
|
|
expect(spawn).not.toHaveBeenCalled();
|
|
expect(entry(s, 'p').status).toBe('active'); // untouched
|
|
});
|
|
});
|
|
|
|
describe('supervisor shutdownAll is a clean stop, not a crash', () => {
|
|
it('marks plugins stopped BEFORE killing, so a child exit during shutdown logs no phantom crash', async () => {
|
|
const onStatus = vi.fn();
|
|
const onLog = vi.fn();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const s = new PluginSupervisor((() => ({ dispose: vi.fn() })) as any, { onStatus, onLog }, {});
|
|
const sup = {
|
|
id: 'p',
|
|
status: 'active',
|
|
child: {},
|
|
rpcHost: { dispose: vi.fn() },
|
|
crashes: [],
|
|
jobTasks: undefined,
|
|
pending: new Map(),
|
|
invocations: new Map(),
|
|
};
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', sup);
|
|
// kill() simulates the child actually exiting during shutdown → fires onExit
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const internals = s as any;
|
|
internals.kill = vi.fn(async (x: unknown) => {
|
|
internals.onExit(x, 0, 'SIGTERM');
|
|
});
|
|
await s.shutdownAll();
|
|
expect(sup.status).toBe('stopped'); // stopped before kill
|
|
// onExit took the early-return path — no crash bookkeeping written to the DB hooks
|
|
expect(onLog).not.toHaveBeenCalledWith('p', 'warn', expect.stringContaining('crashed'));
|
|
expect(onStatus).not.toHaveBeenCalledWith('p', 'error', expect.anything());
|
|
expect(onStatus).not.toHaveBeenCalledWith('p', 'starting', expect.anything());
|
|
});
|
|
});
|
|
|
|
describe('supervisor crash-restart does not leak cron tasks', () => {
|
|
it("onExit stops the dead child's jobTasks before re-scheduling", () => {
|
|
const { s } = makeSupervisor();
|
|
const stop = vi.fn();
|
|
const sup = {
|
|
id: 'p',
|
|
status: 'active',
|
|
child: {},
|
|
jobTasks: [{ stop }],
|
|
pending: new Map(),
|
|
invocations: new Map(),
|
|
crashes: [],
|
|
};
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', sup);
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).onExit(sup, 1, null);
|
|
|
|
expect(stop).toHaveBeenCalledTimes(1); // the previous incarnation's tasks were stopped
|
|
expect(sup.jobTasks).toBeUndefined(); // reference cleared so it can't be double-stopped
|
|
});
|
|
});
|
|
|
|
describe('supervisor buffers events across a restart and replays on activation', () => {
|
|
// A subscriber with a grant + one subscription, at an arbitrary status.
|
|
const sub = (status: string) => ({
|
|
id: 'p',
|
|
status,
|
|
granted: new Set(['events:subscribe']),
|
|
events: ['trip:updated'],
|
|
});
|
|
|
|
it('an active subscriber gets the event immediately, nothing buffered', () => {
|
|
const { s } = makeSupervisor();
|
|
const invoke = vi.fn((..._a: unknown[]) => Promise.resolve());
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).invoke = invoke;
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', sub('active'));
|
|
s.deliverEvent(7, 'trip:updated');
|
|
expect(invoke).toHaveBeenCalledTimes(1);
|
|
expect(invoke.mock.calls[0][1]).toBe('invoke.event');
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
expect((s as any).pendingEvents.get('p')).toBeUndefined();
|
|
});
|
|
|
|
it('an event that fires while the subscriber is starting is buffered, then replayed on activation', () => {
|
|
const { s } = makeSupervisor();
|
|
const invoke = vi.fn(() => Promise.resolve());
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).invoke = invoke;
|
|
const sup = sub('starting');
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', sup);
|
|
s.deliverEvent(7, 'trip:updated');
|
|
expect(invoke).not.toHaveBeenCalled(); // held, not delivered mid-restart
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
expect((s as any).pendingEvents.get('p')).toHaveLength(1);
|
|
// plugin comes back
|
|
sup.status = 'active';
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).flushPendingEvents(sup);
|
|
expect(invoke).toHaveBeenCalledTimes(1); // replayed exactly once
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
expect((s as any).pendingEvents.get('p')).toBeUndefined(); // buffer drained
|
|
});
|
|
|
|
it('a deliberately stopped subscriber neither receives nor buffers the event', () => {
|
|
const { s } = makeSupervisor();
|
|
const invoke = vi.fn(() => Promise.resolve());
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).invoke = invoke;
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', sub('stopped'));
|
|
s.deliverEvent(7, 'trip:updated');
|
|
expect(invoke).not.toHaveBeenCalled();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
expect((s as any).pendingEvents.get('p')).toBeUndefined();
|
|
});
|
|
|
|
it('flush drops expired events and re-checks the grant (nothing leaks if it was revoked while down)', () => {
|
|
const { s } = makeSupervisor();
|
|
const invoke = vi.fn(() => Promise.resolve());
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).invoke = invoke;
|
|
const sup = sub('active');
|
|
// one fresh + one already-expired buffered event
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).pendingEvents.set('p', [
|
|
{ tripId: 1, event: 'trip:updated', expiresAt: Date.now() + 60_000 },
|
|
{ tripId: 2, event: 'trip:updated', expiresAt: Date.now() - 1 },
|
|
]);
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).flushPendingEvents(sup);
|
|
expect(invoke).toHaveBeenCalledTimes(1); // only the unexpired one
|
|
// grant revoked while down → flush delivers nothing
|
|
invoke.mockClear();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).pendingEvents.set('p', [{ tripId: 1, event: 'trip:updated', expiresAt: Date.now() + 60_000 }]);
|
|
sup.granted = new Set(); // grant gone
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).flushPendingEvents(sup);
|
|
expect(invoke).not.toHaveBeenCalled();
|
|
});
|
|
|
|
it('the buffer is bounded per plugin (drop-oldest past the cap)', () => {
|
|
const { s } = makeSupervisor();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).invoke = vi.fn(() => Promise.resolve());
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', sub('starting'));
|
|
for (let i = 0; i < 250; i++) s.deliverEvent(i, 'trip:updated');
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const q = (s as any).pendingEvents.get('p');
|
|
expect(q).toHaveLength(200); // capped
|
|
expect(q[0].tripId).toBe(50); // oldest 50 dropped
|
|
});
|
|
|
|
it("disable() clears a plugin's buffered events", () => {
|
|
const { s } = makeSupervisor();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', {
|
|
id: 'p',
|
|
status: 'starting',
|
|
granted: new Set(),
|
|
events: [],
|
|
jobTasks: undefined,
|
|
pending: new Map(),
|
|
invocations: new Map(),
|
|
crashes: [],
|
|
rpcHost: { dispose: vi.fn() },
|
|
});
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).pendingEvents.set('p', [{ tripId: 1, event: 'trip:updated', expiresAt: Date.now() + 60_000 }]);
|
|
void s.disable('p');
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
expect((s as any).pendingEvents.get('p')).toBeUndefined();
|
|
});
|
|
});
|
|
|
|
describe('supervisor GDPR user-data hooks are grant- and status-gated', () => {
|
|
const withGrant = (status: string) => ({ id: 'p', status, granted: new Set(['hook:user-data']) });
|
|
|
|
it('deliverUserErasure ACKs for an active plugin (a queued erasure is a duty, not grant-gated)', async () => {
|
|
const { s } = makeSupervisor();
|
|
const invoke = vi.fn(() => Promise.resolve({}));
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).invoke = invoke;
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', withGrant('active'));
|
|
expect(await s.deliverUserErasure('p', 42)).toBe(true);
|
|
expect(invoke).toHaveBeenCalledWith(
|
|
'p',
|
|
'invoke.deleteUserData',
|
|
{ userId: 42 },
|
|
expect.objectContaining({ actingUserId: undefined }),
|
|
);
|
|
// inactive → not delivered, not acked (stays queued by the caller)
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', withGrant('starting'));
|
|
invoke.mockClear();
|
|
expect(await s.deliverUserErasure('p', 42)).toBe(false);
|
|
expect(invoke).not.toHaveBeenCalled();
|
|
// active but the grant was later DROPPED → still delivered (the row was enqueued
|
|
// while it held the grant; refusing would strand the user's data forever)
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', { id: 'p', status: 'active', granted: new Set() });
|
|
expect(await s.deliverUserErasure('p', 42)).toBe(true);
|
|
});
|
|
|
|
it('deliverUserErasure resolves false when the child errors (so the row is retried)', async () => {
|
|
const { s } = makeSupervisor();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).invoke = vi.fn(() => Promise.reject(new Error('child died')));
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', withGrant('active'));
|
|
expect(await s.deliverUserErasure('p', 42)).toBe(false);
|
|
});
|
|
|
|
it('collectUserExport returns the plugin payload, undefined when ungranted/inactive', async () => {
|
|
const { s } = makeSupervisor();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).invoke = vi.fn(() => Promise.resolve({ ok: true, data: [{ v: 1 }] }));
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', withGrant('active'));
|
|
expect(await s.collectUserExport('p', 42)).toEqual({ ok: true, data: [{ v: 1 }] });
|
|
// ungranted/inactive → undefined (not applicable)
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', { id: 'p', status: 'active', granted: new Set() });
|
|
expect(await s.collectUserExport('p', 42)).toBeUndefined();
|
|
// active + granted but the export throws → { ok: false } (flagged incomplete, not omitted)
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).invoke = vi.fn(() => Promise.reject(new Error('timeout')));
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).running.set('p', withGrant('active'));
|
|
expect(await s.collectUserExport('p', 42)).toEqual({ ok: false });
|
|
});
|
|
});
|
|
|
|
describe('supervisor rate-limits ctx.* dispatch', () => {
|
|
it('refuses a throttled call with HOST_ERROR instead of dispatching it', async () => {
|
|
const { s } = makeSupervisor();
|
|
const dispatch = vi.fn(async () => ({ k: 'res', id: 'x', ok: true, result: 1 }));
|
|
const send = vi.fn();
|
|
const sup = {
|
|
id: 'p',
|
|
status: 'active',
|
|
child: { send },
|
|
rpcHost: { dispatch },
|
|
invocations: new Map(),
|
|
pending: new Map(),
|
|
rpcLimiter: new RpcRateLimiter({ burst: 1, perSec: 0, maxInFlight: 8 }, 0),
|
|
};
|
|
const req = { k: 'req', id: 'r1', method: 'db.query', params: {} };
|
|
// first call within the burst -> dispatched
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
await (s as any).onMessage(sup, req);
|
|
expect(dispatch).toHaveBeenCalledTimes(1);
|
|
// second call is over budget -> refused, never reaches dispatch
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
await (s as any).onMessage(sup, { ...req, id: 'r2' });
|
|
expect(dispatch).toHaveBeenCalledTimes(1);
|
|
const last = send.mock.calls[send.mock.calls.length - 1][0];
|
|
expect(last).toMatchObject({ k: 'res', id: 'r2', ok: false, error: { code: 'HOST_ERROR' } });
|
|
});
|
|
});
|
|
|
|
describe('supervisor throttles plugin log/stderr volume (host-thread DoS guard)', () => {
|
|
// Regression: the rpcLimiter only covers the `req` (ctx.*) channel. Log lines
|
|
// (ctx.log.*, stderr, unknown evt topics) reach a SYNCHRONOUS INSERT+prune on the
|
|
// host thread through a different branch, so a `while(true) ctx.log.error(...)` loop
|
|
// would freeze the instance unless that path is throttled too.
|
|
it('drops evt/log lines beyond the per-plugin bucket so a ctx.log flood cannot pin the host thread', async () => {
|
|
const onLog = vi.fn();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const s = new PluginSupervisor((() => ({ dispose: vi.fn() })) as any, { onLog }, {});
|
|
const sup = { id: 'p', status: 'active', logLimiter: new TokenBucket(3, 0, Date.now()), droppedLogs: 0 };
|
|
// 10 rapid error lines through the real evt→log path; the bucket holds only 3
|
|
// tokens (refill 0/sec for the test), so only 3 reach onLog's INSERT+prune.
|
|
for (let i = 0; i < 10; i++) {
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
await (s as any).onMessage(sup, { k: 'evt', topic: 'log', data: { level: 'error', msg: `line ${i}` } });
|
|
}
|
|
expect(onLog).toHaveBeenCalledTimes(3);
|
|
expect(sup.droppedLogs).toBe(7);
|
|
});
|
|
|
|
it('surfaces a single "N dropped" summary when logging resumes after throttling', () => {
|
|
const onLog = vi.fn();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const s = new PluginSupervisor((() => ({ dispose: vi.fn() })) as any, { onLog }, {});
|
|
const sup = { id: 'p', logLimiter: new TokenBucket(1, 0, Date.now()), droppedLogs: 0 };
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const rec = (level: string, msg: string) => (s as any).recordLog(sup, level, msg);
|
|
rec('error', 'a'); // spends the only token
|
|
rec('error', 'b'); // dropped
|
|
rec('error', 'c'); // dropped
|
|
expect(sup.droppedLogs).toBe(2);
|
|
// Simulate a refill; the next accepted line first reports the drop, then itself.
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(sup.logLimiter as any).tokens = 1;
|
|
rec('error', 'd');
|
|
expect(onLog).toHaveBeenCalledWith('p', 'warn', expect.stringContaining('2 log line(s) dropped'));
|
|
expect(onLog).toHaveBeenCalledWith('p', 'error', 'd', undefined);
|
|
expect(sup.droppedLogs).toBe(0);
|
|
});
|
|
});
|
|
|
|
/**
|
|
* The 'message' listener is where a rejection would escape: an EventEmitter discards
|
|
* what its callback returns, onMessage is async, and the host has no
|
|
* unhandledRejection handler — so Node 22 would end the process. A plugin getting to
|
|
* kill the supervisor that contains it defeats the point of the supervisor.
|
|
*/
|
|
describe('supervisor message handling never escapes as a rejection', () => {
|
|
it('logs a failed message against the plugin and keeps running', async () => {
|
|
const onLog = vi.fn();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const s = new PluginSupervisor((() => ({ dispose: vi.fn() })) as any, { onLog }, {});
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).onMessage = () => Promise.reject(new Error('malformed envelope'));
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
expect(() => (s as any).handleChildMessage({ id: 'p' }, { k: 'req', id: 'x' })).not.toThrow();
|
|
|
|
await new Promise((r) => setTimeout(r, 0));
|
|
expect(onLog).toHaveBeenCalledWith('p', 'error', expect.stringContaining('malformed envelope'));
|
|
});
|
|
|
|
it('says nothing when the message is handled normally', async () => {
|
|
const onLog = vi.fn();
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
const s = new PluginSupervisor((() => ({ dispose: vi.fn() })) as any, { onLog }, {});
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).onMessage = () => Promise.resolve();
|
|
|
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
(s as any).handleChildMessage({ id: 'p' }, { k: 'evt', topic: 'log' });
|
|
|
|
await new Promise((r) => setTimeout(r, 0));
|
|
expect(onLog).not.toHaveBeenCalled();
|
|
});
|
|
});
|