mirror of
https://github.com/mauriceboe/TREK.git
synced 2026-10-19 13:36:48 +00:00
Maurice
dc74d59b11
All three sit in code whose whole job is to survive a plugin that misbehaves. readTarGz trusted the tar size field, which parseInt reads with a leading minus. At -512 the cursor advance cancels out and below it goes backwards, so the same header is read forever while every pass appends a member — a 70-byte upload pins the main thread and exhausts the heap, synchronously, taking HTTP, the socket and the database with it. The entry budgets never get a say: they run on the list readTarGz has to finish building first. Sideload is the reachable path. The supervisor handed onMessage's promise to an EventEmitter listener, which discards it. A rejection there is an unhandledRejection, the host has no net for it, and Node ends the process — the supervisor killed by the plugin it contains. dispatch() read `'_inv' in req.params` before handle()'s try/catch, so params that are not an object threw a TypeError into exactly that discarded promise.
350 lines
18 KiB
TypeScript
350 lines
18 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 { 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);
|
|
});
|
|
});
|
|
|
|
/**
|
|
* 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();
|
|
});
|
|
});
|