/** * 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 { describe, it, expect, vi, afterEach } from 'vitest'; import { PluginSupervisor } from '../../../src/nest/plugins/supervisor/plugin-supervisor'; import { RpcRateLimiter, TokenBucket } from '../../../src/nest/plugins/host/rate-limit'; function makeSupervisor() { const dispose = vi.fn(); // eslint-disable-next-line @typescript-eslint/no-explicit-any const s = new PluginSupervisor((() => ({ dispose })) as any, {}, {}); 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 (s as any).kill = vi.fn(async (x: any) => { (s as any).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); }); });