feat: platform backbone and field capture, first part (checkpoint)
Platform backbone (plan 2026-10-13-plattform-rueckgrat), reviewed tasks: - run roles api/worker/job/all with shared startup checks (Task 1) - schema and migration for DomainEvent, deliveries, JobRun, ThrottlerHit (Task 2) - audit, notifications and domain events written in the same transaction (Task 3) - dispatcher with SKIP LOCKED claims, retries and stale release (Task 4) - worker building blocks: tick, LISTEN/NOTIFY wake-up, health (Task 5) - worker process and entry points (Task 6) - throttler counters in Postgres (Task 19) - separate keys per purpose with key ring (Task 20) - public origin page without exact location (Task 22) - shop keys split into read and write (Task 23) - Kustomize base, components, overlay encry, CI validation (Task 26) Field capture T3 (plan 2026-10-13-t3-erfassung-am-stand), reviewed tasks: - brood frames, read endpoints, apiary overview and offline pack - network basics, offline queue v3, navigation sheets, form building blocks - new inspection with last-time hints and weather, varroa per day, treatment prefill, apiary page with round - inspection photo (Task 10) is implemented, review pending Not rolled out. Remaining tasks follow in later commits. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
This commit is contained in:
Nick SzaunigandClaude Opus 5.5 committed
1 parent
10cbe9bf2d
commit
c01a38835a
340 files changed
+25695
-1808
No files matched your search
@@ -0,0 +1,254 @@
|
||||
import { Logger } from '@nestjs/common';
|
||||
import { Client } from 'pg';
|
||||
import { ListenClient, PgListener } from './pg-listener';
|
||||
import type { WakeChannel } from '../common/worker-wakeup';
|
||||
|
||||
jest.mock('pg', () => ({ Client: jest.fn() }));
|
||||
|
||||
const CONNECTION_STRING = 'postgresql://user:secret@db/app';
|
||||
const CHANNELS: readonly WakeChannel[] = [
|
||||
'domain_event',
|
||||
'calendar_sync',
|
||||
'webhook_delivery',
|
||||
];
|
||||
|
||||
class FakeClient implements ListenClient {
|
||||
connect = jest.fn<Promise<void>, []>().mockResolvedValue(undefined);
|
||||
query = jest.fn<Promise<unknown>, [string]>().mockResolvedValue(undefined);
|
||||
end = jest.fn<Promise<void>, []>().mockResolvedValue(undefined);
|
||||
private readonly handlers = new Map<string, (arg: unknown) => void>();
|
||||
|
||||
on(
|
||||
event: 'notification' | 'error' | 'end',
|
||||
cb: (arg: unknown) => void,
|
||||
): void {
|
||||
this.handlers.set(event, cb);
|
||||
}
|
||||
|
||||
emit(event: 'notification' | 'error' | 'end', arg?: unknown): void {
|
||||
this.handlers.get(event)?.(arg);
|
||||
}
|
||||
}
|
||||
|
||||
describe('PgListener', () => {
|
||||
let clients: FakeClient[];
|
||||
let connectFails: boolean;
|
||||
let logger: {
|
||||
error: jest.Mock<void, unknown[]>;
|
||||
warn: jest.Mock<void, unknown[]>;
|
||||
log: jest.Mock<void, unknown[]>;
|
||||
};
|
||||
let onNotify: jest.Mock;
|
||||
let listener: PgListener;
|
||||
|
||||
function createListener(
|
||||
clientFactory: (connectionString: string) => ListenClient,
|
||||
): PgListener {
|
||||
return new PgListener({
|
||||
connectionString: CONNECTION_STRING,
|
||||
channels: CHANNELS,
|
||||
onNotify,
|
||||
logger: logger as unknown as Logger,
|
||||
clientFactory,
|
||||
});
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
jest.useFakeTimers();
|
||||
clients = [];
|
||||
connectFails = false;
|
||||
logger = {
|
||||
error: jest.fn<void, unknown[]>(),
|
||||
warn: jest.fn<void, unknown[]>(),
|
||||
log: jest.fn<void, unknown[]>(),
|
||||
};
|
||||
onNotify = jest.fn();
|
||||
listener = createListener(() => {
|
||||
const client = new FakeClient();
|
||||
if (connectFails) {
|
||||
client.connect.mockRejectedValue(new Error('ECONNREFUSED'));
|
||||
}
|
||||
clients.push(client);
|
||||
return client;
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await listener.stop();
|
||||
jest.useRealTimers();
|
||||
});
|
||||
|
||||
it('LISTENs on every channel after connecting', async () => {
|
||||
await listener.start();
|
||||
|
||||
expect(clients).toHaveLength(1);
|
||||
expect(clients[0].connect).toHaveBeenCalledTimes(1);
|
||||
expect(clients[0].query.mock.calls.map((c) => c[0])).toEqual([
|
||||
'LISTEN "domain_event"',
|
||||
'LISTEN "calendar_sync"',
|
||||
'LISTEN "webhook_delivery"',
|
||||
]);
|
||||
});
|
||||
|
||||
it('passes the connection string to the client factory', async () => {
|
||||
const factory = jest.fn<ListenClient, [string]>(() => new FakeClient());
|
||||
const other = createListener(factory);
|
||||
await other.start();
|
||||
expect(factory).toHaveBeenCalledWith(CONNECTION_STRING);
|
||||
await other.stop();
|
||||
});
|
||||
|
||||
it('forwards notifications of known channels and ignores others', async () => {
|
||||
await listener.start();
|
||||
|
||||
clients[0].emit('notification', { channel: 'calendar_sync', payload: '' });
|
||||
clients[0].emit('notification', { channel: 'something_else' });
|
||||
clients[0].emit('notification', 'garbage');
|
||||
|
||||
expect(onNotify).toHaveBeenCalledTimes(1);
|
||||
expect(onNotify).toHaveBeenCalledWith('calendar_sync');
|
||||
});
|
||||
|
||||
it('reconnects after an error with 1 s, 2 s, 4 s … at most 30 s', async () => {
|
||||
await listener.start();
|
||||
connectFails = true;
|
||||
clients[0].emit('error', new Error('connection lost'));
|
||||
|
||||
for (const delay of [1000, 2000, 4000, 8000, 16000, 30000, 30000]) {
|
||||
const before = clients.length;
|
||||
await jest.advanceTimersByTimeAsync(delay - 1);
|
||||
expect(clients).toHaveLength(before);
|
||||
await jest.advanceTimersByTimeAsync(1);
|
||||
expect(clients).toHaveLength(before + 1);
|
||||
}
|
||||
});
|
||||
|
||||
it('resets the backoff after a successful reconnect', async () => {
|
||||
await listener.start();
|
||||
clients[0].emit('error', new Error('lost'));
|
||||
await jest.advanceTimersByTimeAsync(1000);
|
||||
expect(clients).toHaveLength(2);
|
||||
|
||||
clients[1].emit('end');
|
||||
await jest.advanceTimersByTimeAsync(999);
|
||||
expect(clients).toHaveLength(2);
|
||||
await jest.advanceTimersByTimeAsync(1);
|
||||
expect(clients).toHaveLength(3);
|
||||
});
|
||||
|
||||
it('reconnects only once when error and end both fire', async () => {
|
||||
await listener.start();
|
||||
clients[0].emit('error', new Error('lost'));
|
||||
clients[0].emit('end');
|
||||
await jest.advanceTimersByTimeAsync(1000);
|
||||
expect(clients).toHaveLength(2);
|
||||
});
|
||||
|
||||
it('retries when the initial connect fails and logs no credentials', async () => {
|
||||
connectFails = true;
|
||||
await listener.start();
|
||||
expect(clients).toHaveLength(1);
|
||||
|
||||
await jest.advanceTimersByTimeAsync(1000);
|
||||
expect(clients).toHaveLength(2);
|
||||
|
||||
const logged = JSON.stringify([
|
||||
...logger.error.mock.calls,
|
||||
...logger.warn.mock.calls,
|
||||
]);
|
||||
expect(logged).not.toContain('secret');
|
||||
});
|
||||
|
||||
it('stop() ends the client and does not reconnect', async () => {
|
||||
await listener.start();
|
||||
await listener.stop();
|
||||
|
||||
expect(clients[0].end).toHaveBeenCalledTimes(1);
|
||||
clients[0].emit('end');
|
||||
await jest.advanceTimersByTimeAsync(60_000);
|
||||
expect(clients).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('flushes a backlog by notifying every channel after a reconnect only', async () => {
|
||||
await listener.start();
|
||||
expect(onNotify).not.toHaveBeenCalled();
|
||||
|
||||
clients[0].emit('error', new Error('lost'));
|
||||
await jest.advanceTimersByTimeAsync(1000);
|
||||
|
||||
expect(
|
||||
(onNotify.mock.calls as [WakeChannel][]).map(([channel]) => channel),
|
||||
).toEqual([...CHANNELS]);
|
||||
});
|
||||
|
||||
it('stop() does not hang on a client whose end() never settles', async () => {
|
||||
await listener.start();
|
||||
clients[0].end.mockReturnValue(new Promise<void>(() => undefined));
|
||||
|
||||
const stopping = listener.stop();
|
||||
await jest.advanceTimersByTimeAsync(5000);
|
||||
await expect(stopping).resolves.toBeUndefined();
|
||||
});
|
||||
|
||||
it('does not recycle the connection when onNotify throws', async () => {
|
||||
onNotify.mockImplementation(() => {
|
||||
throw new Error('consumer failed');
|
||||
});
|
||||
await listener.start();
|
||||
|
||||
clients[0].emit('notification', { channel: 'domain_event' });
|
||||
await jest.advanceTimersByTimeAsync(60_000);
|
||||
expect(clients).toHaveLength(1);
|
||||
expect(clients[0].end).not.toHaveBeenCalled();
|
||||
|
||||
clients[0].emit('error', new Error('lost'));
|
||||
await jest.advanceTimersByTimeAsync(1000);
|
||||
expect(clients).toHaveLength(2);
|
||||
// Flush calls all threw, yet the new connection stays.
|
||||
await jest.advanceTimersByTimeAsync(60_000);
|
||||
expect(onNotify).toHaveBeenCalledTimes(1 + CHANNELS.length);
|
||||
expect(clients).toHaveLength(2);
|
||||
expect(clients[1].end).not.toHaveBeenCalled();
|
||||
expect(logger.error).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('creates the default pg client with keep-alive settings', async () => {
|
||||
const created = new FakeClient();
|
||||
(Client as unknown as jest.Mock).mockReturnValue(created);
|
||||
const defaulted = new PgListener({
|
||||
connectionString: CONNECTION_STRING,
|
||||
channels: CHANNELS,
|
||||
onNotify,
|
||||
logger: logger as unknown as Logger,
|
||||
});
|
||||
await defaulted.start();
|
||||
|
||||
expect(Client).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
connectionString: CONNECTION_STRING,
|
||||
keepAlive: true,
|
||||
keepAliveInitialDelayMillis: 10_000,
|
||||
}),
|
||||
);
|
||||
await defaulted.stop();
|
||||
});
|
||||
|
||||
it('pins the LISTEN session to UTC like every Prisma connection', async () => {
|
||||
(Client as unknown as jest.Mock).mockReturnValue(new FakeClient());
|
||||
const defaulted = new PgListener({
|
||||
connectionString: CONNECTION_STRING,
|
||||
channels: CHANNELS,
|
||||
onNotify,
|
||||
logger: logger as unknown as Logger,
|
||||
});
|
||||
await defaulted.start();
|
||||
|
||||
expect(Client).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
connectionString: CONNECTION_STRING,
|
||||
options: '-c TimeZone=UTC',
|
||||
}),
|
||||
);
|
||||
await defaulted.stop();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,149 @@
|
||||
import { Logger } from '@nestjs/common';
|
||||
import { Client } from 'pg';
|
||||
import { loggableError } from '../common/loggable-error';
|
||||
import { withTimeout } from '../common/with-timeout';
|
||||
import type { WakeChannel } from '../common/worker-wakeup';
|
||||
import { dbSessionConfig } from '../prisma/db-session';
|
||||
|
||||
const RECONNECT_INITIAL_DELAY_MS = 1000;
|
||||
const RECONNECT_MAX_DELAY_MS = 30_000;
|
||||
const CONNECTION_TIMEOUT_MS = 10_000;
|
||||
const KEEP_ALIVE_INITIAL_DELAY_MS = 10_000;
|
||||
const END_TIMEOUT_MS = 5000;
|
||||
|
||||
export interface ListenClient {
|
||||
connect(): Promise<void>;
|
||||
query(sql: string): Promise<unknown>;
|
||||
on(event: 'notification' | 'error' | 'end', cb: (arg: unknown) => void): void;
|
||||
end(): Promise<void>;
|
||||
}
|
||||
|
||||
export interface PgListenerOptions {
|
||||
connectionString: string;
|
||||
channels: readonly WakeChannel[];
|
||||
onNotify: (channel: WakeChannel) => void;
|
||||
logger: Logger;
|
||||
clientFactory?: (connectionString: string) => ListenClient;
|
||||
}
|
||||
|
||||
/**
|
||||
* Same connection string and session settings (UTC) as Prisma, so host,
|
||||
* TLS and time zone match.
|
||||
*/
|
||||
function createPgClient(connectionString: string): ListenClient {
|
||||
return new Client({
|
||||
...dbSessionConfig(connectionString),
|
||||
keepAlive: true,
|
||||
keepAliveInitialDelayMillis: KEEP_ALIVE_INITIAL_DELAY_MS,
|
||||
connectionTimeoutMillis: CONNECTION_TIMEOUT_MS,
|
||||
}) as unknown as ListenClient;
|
||||
}
|
||||
|
||||
function notificationChannel(arg: unknown): string | undefined {
|
||||
if (typeof arg !== 'object' || arg === null) return undefined;
|
||||
const { channel } = arg as { channel?: unknown };
|
||||
return typeof channel === 'string' ? channel : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Dedicated Postgres connection that LISTENs on the wake-up channels and
|
||||
* reconnects with exponential backoff (1 s, doubled, at most 30 s). A missed
|
||||
* notification is harmless: the polling loops catch up on their next tick.
|
||||
*/
|
||||
export class PgListener {
|
||||
private readonly createClient: (connectionString: string) => ListenClient;
|
||||
private client: ListenClient | undefined;
|
||||
private reconnectTimer: NodeJS.Timeout | undefined;
|
||||
private reconnectDelayMs = RECONNECT_INITIAL_DELAY_MS;
|
||||
private stopped = true;
|
||||
|
||||
constructor(private readonly options: PgListenerOptions) {
|
||||
this.createClient = options.clientFactory ?? createPgClient;
|
||||
}
|
||||
|
||||
async start(): Promise<void> {
|
||||
if (!this.stopped) return;
|
||||
this.stopped = false;
|
||||
await this.connect(false);
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
this.stopped = true;
|
||||
if (this.reconnectTimer) {
|
||||
clearTimeout(this.reconnectTimer);
|
||||
this.reconnectTimer = undefined;
|
||||
}
|
||||
const client = this.client;
|
||||
this.client = undefined;
|
||||
if (client) await this.endQuietly(client);
|
||||
}
|
||||
|
||||
private async connect(isReconnect: boolean): Promise<void> {
|
||||
const client = this.createClient(this.options.connectionString);
|
||||
this.client = client;
|
||||
client.on('notification', (arg) => this.handleNotification(arg));
|
||||
client.on('error', (error) => this.handleLoss(client, error));
|
||||
client.on('end', () => this.handleLoss(client));
|
||||
try {
|
||||
await client.connect();
|
||||
for (const channel of this.options.channels) {
|
||||
await client.query(`LISTEN "${channel}"`);
|
||||
}
|
||||
this.reconnectDelayMs = RECONNECT_INITIAL_DELAY_MS;
|
||||
// Notifications sent while disconnected are gone; wake everything.
|
||||
if (isReconnect) {
|
||||
this.options.channels.forEach((channel) => this.notify(channel));
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
this.handleLoss(client, error);
|
||||
}
|
||||
}
|
||||
|
||||
/** A failing consumer must never be mistaken for a lost connection. */
|
||||
private notify(channel: WakeChannel): void {
|
||||
try {
|
||||
this.options.onNotify(channel);
|
||||
} catch (error: unknown) {
|
||||
const { message } = loggableError(error);
|
||||
this.options.logger.error(`onNotify failed for ${channel}: ${message}`);
|
||||
}
|
||||
}
|
||||
|
||||
private handleNotification(arg: unknown): void {
|
||||
const channel = notificationChannel(arg);
|
||||
const known = this.options.channels.find((c) => c === channel);
|
||||
if (known) this.notify(known);
|
||||
}
|
||||
|
||||
private handleLoss(client: ListenClient, error?: unknown): void {
|
||||
// error and end fire for the same loss; only the current client counts.
|
||||
if (this.client !== client) return;
|
||||
this.client = undefined;
|
||||
if (error !== undefined) {
|
||||
const { message } = loggableError(error);
|
||||
this.options.logger.warn(`LISTEN connection lost: ${message}`);
|
||||
}
|
||||
void this.endQuietly(client);
|
||||
if (this.stopped) return;
|
||||
const delay = this.reconnectDelayMs;
|
||||
this.reconnectDelayMs = Math.min(delay * 2, RECONNECT_MAX_DELAY_MS);
|
||||
this.options.logger.log(`Reconnecting LISTEN connection in ${delay} ms`);
|
||||
this.reconnectTimer = setTimeout(() => {
|
||||
this.reconnectTimer = undefined;
|
||||
void this.connect(true);
|
||||
}, delay);
|
||||
}
|
||||
|
||||
private async endQuietly(client: ListenClient): Promise<void> {
|
||||
try {
|
||||
await withTimeout(
|
||||
client.end(),
|
||||
END_TIMEOUT_MS,
|
||||
'LISTEN client end timed out',
|
||||
);
|
||||
} catch (error: unknown) {
|
||||
const { message } = loggableError(error);
|
||||
this.options.logger.warn(`Closing LISTEN connection failed: ${message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,215 @@
|
||||
import { Logger } from '@nestjs/common';
|
||||
import { PollingLoop } from './polling-loop';
|
||||
|
||||
const INTERVAL_MS = 5000;
|
||||
|
||||
function deferred(): { promise: Promise<void>; resolve: () => void } {
|
||||
let resolve!: () => void;
|
||||
const promise = new Promise<void>((r) => {
|
||||
resolve = r;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
describe('PollingLoop', () => {
|
||||
let logger: { error: jest.Mock };
|
||||
|
||||
function createLoop(tick: () => Promise<unknown>): PollingLoop {
|
||||
return new PollingLoop({
|
||||
name: 'test-loop',
|
||||
intervalMs: INTERVAL_MS,
|
||||
tick,
|
||||
logger: logger as unknown as Logger,
|
||||
});
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
jest.useFakeTimers();
|
||||
logger = { error: jest.fn() };
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
jest.useRealTimers();
|
||||
});
|
||||
|
||||
it('ticks every intervalMs', async () => {
|
||||
const tick = jest.fn().mockResolvedValue(undefined);
|
||||
const loop = createLoop(tick);
|
||||
loop.start();
|
||||
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
expect(tick).toHaveBeenCalledTimes(1);
|
||||
await jest.advanceTimersByTimeAsync(INTERVAL_MS - 1);
|
||||
expect(tick).toHaveBeenCalledTimes(1);
|
||||
await jest.advanceTimersByTimeAsync(1);
|
||||
expect(tick).toHaveBeenCalledTimes(2);
|
||||
await jest.advanceTimersByTimeAsync(INTERVAL_MS);
|
||||
expect(tick).toHaveBeenCalledTimes(3);
|
||||
|
||||
await loop.stop();
|
||||
});
|
||||
|
||||
it('never runs two ticks at once', async () => {
|
||||
const gate = deferred();
|
||||
const tick = jest.fn().mockReturnValue(gate.promise);
|
||||
const loop = createLoop(tick);
|
||||
loop.start();
|
||||
|
||||
await jest.advanceTimersByTimeAsync(INTERVAL_MS * 3);
|
||||
expect(tick).toHaveBeenCalledTimes(1);
|
||||
|
||||
gate.resolve();
|
||||
await loop.stop();
|
||||
});
|
||||
|
||||
it('wake() during a tick runs exactly one more tick right after', async () => {
|
||||
const gate = deferred();
|
||||
const tick = jest
|
||||
.fn()
|
||||
.mockReturnValueOnce(gate.promise)
|
||||
.mockResolvedValue(undefined);
|
||||
const loop = createLoop(tick);
|
||||
loop.start();
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
|
||||
loop.wake();
|
||||
loop.wake();
|
||||
loop.wake();
|
||||
expect(tick).toHaveBeenCalledTimes(1);
|
||||
|
||||
gate.resolve();
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
expect(tick).toHaveBeenCalledTimes(2);
|
||||
|
||||
await loop.stop();
|
||||
});
|
||||
|
||||
it('wake() while waiting runs the next tick immediately', async () => {
|
||||
const tick = jest.fn().mockResolvedValue(undefined);
|
||||
const loop = createLoop(tick);
|
||||
loop.start();
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
expect(tick).toHaveBeenCalledTimes(1);
|
||||
|
||||
loop.wake();
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
expect(tick).toHaveBeenCalledTimes(2);
|
||||
|
||||
await jest.advanceTimersByTimeAsync(INTERVAL_MS - 1);
|
||||
expect(tick).toHaveBeenCalledTimes(2);
|
||||
await jest.advanceTimersByTimeAsync(1);
|
||||
expect(tick).toHaveBeenCalledTimes(3);
|
||||
|
||||
await loop.stop();
|
||||
});
|
||||
|
||||
it('logs a throwing tick via loggableError and keeps going', async () => {
|
||||
const tick = jest
|
||||
.fn()
|
||||
.mockRejectedValueOnce(new Error('tick exploded'))
|
||||
.mockResolvedValue(undefined);
|
||||
const loop = createLoop(tick);
|
||||
loop.start();
|
||||
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
expect(logger.error).toHaveBeenCalledTimes(1);
|
||||
expect(logger.error).toHaveBeenCalledWith(
|
||||
expect.stringContaining('tick exploded'),
|
||||
expect.any(String),
|
||||
);
|
||||
|
||||
await jest.advanceTimersByTimeAsync(INTERVAL_MS);
|
||||
expect(tick).toHaveBeenCalledTimes(2);
|
||||
|
||||
await loop.stop();
|
||||
});
|
||||
|
||||
it('stop() waits for the running tick and schedules nothing more', async () => {
|
||||
const gate = deferred();
|
||||
const tick = jest.fn().mockReturnValue(gate.promise);
|
||||
const loop = createLoop(tick);
|
||||
loop.start();
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
|
||||
let stopped = false;
|
||||
const stopping = loop.stop().then(() => {
|
||||
stopped = true;
|
||||
});
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
expect(stopped).toBe(false);
|
||||
|
||||
gate.resolve();
|
||||
await stopping;
|
||||
expect(stopped).toBe(true);
|
||||
|
||||
await jest.advanceTimersByTimeAsync(INTERVAL_MS * 3);
|
||||
expect(tick).toHaveBeenCalledTimes(1);
|
||||
expect(jest.getTimerCount()).toBe(0);
|
||||
});
|
||||
|
||||
it('wake() after stop() does nothing', async () => {
|
||||
const tick = jest.fn().mockResolvedValue(undefined);
|
||||
const loop = createLoop(tick);
|
||||
loop.start();
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
await loop.stop();
|
||||
|
||||
loop.wake();
|
||||
await jest.advanceTimersByTimeAsync(INTERVAL_MS);
|
||||
expect(tick).toHaveBeenCalledTimes(1);
|
||||
expect(jest.getTimerCount()).toBe(0);
|
||||
});
|
||||
|
||||
it('does not lose a wake() fired right at the end of a tick', async () => {
|
||||
const tick = jest
|
||||
.fn()
|
||||
.mockImplementationOnce(async () => {
|
||||
// Fires after the tick body, before the loop bookkeeping finished.
|
||||
await Promise.resolve();
|
||||
queueMicrotask(() => loop.wake());
|
||||
})
|
||||
.mockResolvedValue(undefined);
|
||||
const loop = createLoop(tick);
|
||||
loop.start();
|
||||
|
||||
await jest.advanceTimersByTimeAsync(1);
|
||||
expect(tick).toHaveBeenCalledTimes(2);
|
||||
|
||||
await loop.stop();
|
||||
});
|
||||
|
||||
it('never runs two ticks at once across stop() and start()', async () => {
|
||||
const gate = deferred();
|
||||
let active = 0;
|
||||
let maxActive = 0;
|
||||
const tick = jest
|
||||
.fn()
|
||||
.mockImplementationOnce(async () => {
|
||||
active++;
|
||||
maxActive = Math.max(maxActive, active);
|
||||
await gate.promise;
|
||||
active--;
|
||||
})
|
||||
.mockImplementation(() => {
|
||||
active++;
|
||||
maxActive = Math.max(maxActive, active);
|
||||
active--;
|
||||
return Promise.resolve();
|
||||
});
|
||||
const loop = createLoop(tick);
|
||||
loop.start();
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
|
||||
void loop.stop();
|
||||
loop.start();
|
||||
await jest.advanceTimersByTimeAsync(INTERVAL_MS);
|
||||
expect(tick).toHaveBeenCalledTimes(1);
|
||||
|
||||
gate.resolve();
|
||||
await jest.advanceTimersByTimeAsync(1);
|
||||
expect(tick).toHaveBeenCalledTimes(2);
|
||||
expect(maxActive).toBe(1);
|
||||
|
||||
await loop.stop();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,81 @@
|
||||
import { Logger } from '@nestjs/common';
|
||||
import { loggableError } from '../common/loggable-error';
|
||||
|
||||
export interface PollingLoopOptions {
|
||||
name: string;
|
||||
intervalMs: number;
|
||||
tick: () => Promise<unknown>;
|
||||
logger: Logger;
|
||||
}
|
||||
|
||||
/**
|
||||
* Single-flight polling loop. The next tick is scheduled with setTimeout
|
||||
* only after the previous one finished (no setInterval), so ticks never
|
||||
* overlap. `wake()` skips the wait; a wake during a running tick leads to
|
||||
* exactly one more tick right after it.
|
||||
*/
|
||||
export class PollingLoop {
|
||||
private timer: NodeJS.Timeout | undefined;
|
||||
private running: Promise<void> | undefined;
|
||||
private rerunRequested = false;
|
||||
private stopped = true;
|
||||
|
||||
constructor(private readonly options: PollingLoopOptions) {}
|
||||
|
||||
start(): void {
|
||||
if (!this.stopped) return;
|
||||
this.stopped = false;
|
||||
// A tick of the previous run may still be finishing; its bookkeeping
|
||||
// schedules the next one, so ticks never overlap across stop()/start().
|
||||
// The first tick after start() runs immediately, also in that case.
|
||||
if (this.running) this.rerunRequested = true;
|
||||
else this.schedule(0);
|
||||
}
|
||||
|
||||
wake(): void {
|
||||
if (this.stopped) return;
|
||||
if (this.running) {
|
||||
this.rerunRequested = true;
|
||||
return;
|
||||
}
|
||||
this.schedule(0);
|
||||
}
|
||||
|
||||
async stop(): Promise<void> {
|
||||
this.stopped = true;
|
||||
this.rerunRequested = false;
|
||||
if (this.timer) {
|
||||
clearTimeout(this.timer);
|
||||
this.timer = undefined;
|
||||
}
|
||||
await this.running;
|
||||
}
|
||||
|
||||
private schedule(delayMs: number): void {
|
||||
if (this.timer) clearTimeout(this.timer);
|
||||
this.timer = setTimeout(() => {
|
||||
this.timer = undefined;
|
||||
this.running = this.runTick().finally(() => {
|
||||
this.running = undefined;
|
||||
// Decided after `running` is cleared: a wake() before this point
|
||||
// set rerunRequested, one after it schedules itself.
|
||||
if (!this.stopped) {
|
||||
this.schedule(this.rerunRequested ? 0 : this.options.intervalMs);
|
||||
}
|
||||
});
|
||||
}, delayMs);
|
||||
}
|
||||
|
||||
private async runTick(): Promise<void> {
|
||||
this.rerunRequested = false;
|
||||
try {
|
||||
await this.options.tick();
|
||||
} catch (error: unknown) {
|
||||
const { message, stack } = loggableError(error);
|
||||
this.options.logger.error(
|
||||
`${this.options.name} tick failed: ${message}`,
|
||||
stack,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
import type { Server } from 'node:http';
|
||||
import {
|
||||
Injectable,
|
||||
Logger,
|
||||
OnApplicationBootstrap,
|
||||
OnApplicationShutdown,
|
||||
} from '@nestjs/common';
|
||||
import { startWorkerHealthServer, workerHealthPort } from './worker-health';
|
||||
import { WorkerRuntime } from './worker-runtime.service';
|
||||
|
||||
/**
|
||||
* GET /health of the worker process on WORKER_HEALTH_PORT (default 3002),
|
||||
* for the kubelet only - no Service, no Ingress. Only WorkerModule provides
|
||||
* it; in "all" the API's own /health endpoints apply.
|
||||
*/
|
||||
@Injectable()
|
||||
export class WorkerHealthServer
|
||||
implements OnApplicationBootstrap, OnApplicationShutdown
|
||||
{
|
||||
private readonly logger = new Logger(WorkerHealthServer.name);
|
||||
private server: Server | undefined;
|
||||
|
||||
constructor(private readonly runtime: WorkerRuntime) {}
|
||||
|
||||
async onApplicationBootstrap(): Promise<void> {
|
||||
const port = workerHealthPort();
|
||||
this.server = await startWorkerHealthServer({
|
||||
port,
|
||||
check: () => this.runtime.health(),
|
||||
});
|
||||
this.logger.log(`Worker health endpoint listening on port ${port}`);
|
||||
}
|
||||
|
||||
async onApplicationShutdown(): Promise<void> {
|
||||
const server = this.server;
|
||||
this.server = undefined;
|
||||
if (!server) {
|
||||
return;
|
||||
}
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
import * as http from 'node:http';
|
||||
import { AddressInfo } from 'node:net';
|
||||
import { Logger } from '@nestjs/common';
|
||||
import {
|
||||
DEFAULT_WORKER_HEALTH_PORT,
|
||||
HEALTH_STALE_MS,
|
||||
startWorkerHealthServer,
|
||||
WorkerHealth,
|
||||
workerHealthPort,
|
||||
} from './worker-health';
|
||||
|
||||
interface JsonResponse {
|
||||
status: number;
|
||||
body: unknown;
|
||||
}
|
||||
|
||||
describe('worker health', () => {
|
||||
let server: http.Server | undefined;
|
||||
|
||||
afterEach(async () => {
|
||||
jest.restoreAllMocks();
|
||||
const running = server;
|
||||
server = undefined;
|
||||
if (running) {
|
||||
await new Promise((resolve) => running.close(resolve));
|
||||
}
|
||||
});
|
||||
|
||||
async function start(check: () => Promise<WorkerHealth>): Promise<number> {
|
||||
server = await startWorkerHealthServer({ port: 0, check });
|
||||
return (server.address() as AddressInfo).port;
|
||||
}
|
||||
|
||||
function request(
|
||||
port: number,
|
||||
path: string,
|
||||
method = 'GET',
|
||||
): Promise<JsonResponse> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const req = http.request({ port, path, method }, (res) => {
|
||||
let raw = '';
|
||||
res.on('data', (chunk: Buffer) => (raw += chunk.toString()));
|
||||
res.on('end', () =>
|
||||
resolve({
|
||||
status: res.statusCode ?? 0,
|
||||
body: raw ? (JSON.parse(raw) as unknown) : undefined,
|
||||
}),
|
||||
);
|
||||
});
|
||||
req.on('error', reject);
|
||||
req.end();
|
||||
});
|
||||
}
|
||||
|
||||
it('exposes the documented constants', () => {
|
||||
expect(HEALTH_STALE_MS).toBe(120_000);
|
||||
expect(DEFAULT_WORKER_HEALTH_PORT).toBe(3002);
|
||||
});
|
||||
|
||||
it('answers 200 {"status":"ok"} on GET /health when healthy', async () => {
|
||||
const port = await start(() => Promise.resolve({ ok: true }));
|
||||
expect(await request(port, '/health')).toEqual({
|
||||
status: 200,
|
||||
body: { status: 'ok' },
|
||||
});
|
||||
});
|
||||
|
||||
it('answers 503 with only status and reason when unhealthy', async () => {
|
||||
const port = await start(() =>
|
||||
Promise.resolve({ ok: false, reason: 'dispatcher-stale' }),
|
||||
);
|
||||
expect(await request(port, '/health')).toEqual({
|
||||
status: 503,
|
||||
body: { status: 'unhealthy', reason: 'dispatcher-stale' },
|
||||
});
|
||||
});
|
||||
|
||||
it('answers 404 for any other path or method', async () => {
|
||||
const port = await start(() => Promise.resolve({ ok: true }));
|
||||
expect((await request(port, '/')).status).toBe(404);
|
||||
expect((await request(port, '/health/ready')).status).toBe(404);
|
||||
expect((await request(port, '/health', 'POST')).status).toBe(404);
|
||||
});
|
||||
|
||||
it('sends no JSON content type with a 404', async () => {
|
||||
const port = await start(() => Promise.resolve({ ok: true }));
|
||||
const headers = await new Promise<http.IncomingHttpHeaders>((resolve) => {
|
||||
http.get({ port, path: '/nope' }, (res) => {
|
||||
res.resume();
|
||||
resolve(res.headers);
|
||||
});
|
||||
});
|
||||
expect(headers['content-type']).toBeUndefined();
|
||||
});
|
||||
|
||||
it('treats a throwing check as 503 database', async () => {
|
||||
const error = jest
|
||||
.spyOn(Logger.prototype, 'error')
|
||||
.mockImplementation(() => undefined);
|
||||
const port = await start(() => Promise.reject(new Error('pw=hunter2')));
|
||||
expect(await request(port, '/health')).toEqual({
|
||||
status: 503,
|
||||
body: { status: 'unhealthy', reason: 'database' },
|
||||
});
|
||||
expect(error).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
describe('workerHealthPort', () => {
|
||||
it('defaults to 3002', () => {
|
||||
expect(workerHealthPort({})).toBe(3002);
|
||||
});
|
||||
|
||||
it('reads WORKER_HEALTH_PORT', () => {
|
||||
expect(workerHealthPort({ WORKER_HEALTH_PORT: '4100' })).toBe(4100);
|
||||
});
|
||||
|
||||
it.each([
|
||||
'abc',
|
||||
'0',
|
||||
'-1',
|
||||
'70000',
|
||||
'30.5',
|
||||
'',
|
||||
'0x10',
|
||||
'1e3',
|
||||
' 3002',
|
||||
'3002 ',
|
||||
'+3002',
|
||||
])('rejects invalid value %p', (value) => {
|
||||
expect(() => workerHealthPort({ WORKER_HEALTH_PORT: value })).toThrow(
|
||||
'WORKER_HEALTH_PORT',
|
||||
);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,94 @@
|
||||
import * as http from 'node:http';
|
||||
import { Logger } from '@nestjs/common';
|
||||
import { loggableError } from '../common/loggable-error';
|
||||
|
||||
/** The dispatcher must have completed a run within this window. */
|
||||
export const HEALTH_STALE_MS = 120_000;
|
||||
export const DEFAULT_WORKER_HEALTH_PORT = 3002;
|
||||
|
||||
const MAX_PORT = 65_535;
|
||||
const HEALTH_PATH = '/health';
|
||||
|
||||
export interface WorkerHealth {
|
||||
ok: boolean;
|
||||
reason?: 'database' | 'dispatcher-stale';
|
||||
}
|
||||
|
||||
export interface WorkerHealthServerOptions {
|
||||
port: number;
|
||||
check: () => Promise<WorkerHealth>;
|
||||
}
|
||||
|
||||
const logger = new Logger('WorkerHealth');
|
||||
|
||||
export function workerHealthPort(env: NodeJS.ProcessEnv = process.env): number {
|
||||
const raw = env.WORKER_HEALTH_PORT;
|
||||
if (raw === undefined) return DEFAULT_WORKER_HEALTH_PORT;
|
||||
const port = /^\d+$/.test(raw) ? Number(raw) : 0;
|
||||
if (port < 1 || port > MAX_PORT) {
|
||||
throw new Error(
|
||||
`WORKER_HEALTH_PORT must be an integer from 1 to ${MAX_PORT}`,
|
||||
);
|
||||
}
|
||||
return port;
|
||||
}
|
||||
|
||||
async function evaluate(
|
||||
check: () => Promise<WorkerHealth>,
|
||||
): Promise<WorkerHealth> {
|
||||
try {
|
||||
return await check();
|
||||
} catch (error: unknown) {
|
||||
const { message, stack } = loggableError(error);
|
||||
logger.error(`Worker health check failed: ${message}`, stack);
|
||||
return { ok: false, reason: 'database' };
|
||||
}
|
||||
}
|
||||
|
||||
function respond(
|
||||
res: http.ServerResponse,
|
||||
status: number,
|
||||
body?: Record<string, string>,
|
||||
): void {
|
||||
const headers: http.OutgoingHttpHeaders = { 'Cache-Control': 'no-store' };
|
||||
if (body) headers['Content-Type'] = 'application/json';
|
||||
res.writeHead(status, headers);
|
||||
res.end(body ? JSON.stringify(body) : undefined);
|
||||
}
|
||||
|
||||
/** Minimal probe endpoint (node:http, no Express): GET /health only. */
|
||||
export async function startWorkerHealthServer(
|
||||
options: WorkerHealthServerOptions,
|
||||
): Promise<http.Server> {
|
||||
const server = http.createServer((req, res) => {
|
||||
const path = (req.url ?? '').split('?')[0];
|
||||
if (req.method !== 'GET' || path !== HEALTH_PATH) {
|
||||
respond(res, 404);
|
||||
return;
|
||||
}
|
||||
evaluate(options.check)
|
||||
.then((health) => {
|
||||
if (health.ok) {
|
||||
respond(res, 200, { status: 'ok' });
|
||||
} else {
|
||||
respond(res, 503, {
|
||||
status: 'unhealthy',
|
||||
reason: health.reason ?? 'database',
|
||||
});
|
||||
}
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
const { message, stack } = loggableError(error);
|
||||
logger.error(`Worker health response failed: ${message}`, stack);
|
||||
res.destroy();
|
||||
});
|
||||
});
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once('error', reject);
|
||||
server.listen(options.port, () => {
|
||||
server.off('error', reject);
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
return server;
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { WorkerRuntime } from './worker-runtime.service';
|
||||
|
||||
/**
|
||||
* The worker loops. Imported by WorkerModule (dist/worker.js) and
|
||||
* AllInOneModule ("all"), never by AppModule: API pods run no background
|
||||
* loops. Its dependencies come from the global Prisma, DomainEvents and
|
||||
* CalendarSync modules of the importing root module.
|
||||
*/
|
||||
@Module({
|
||||
providers: [WorkerRuntime],
|
||||
exports: [WorkerRuntime],
|
||||
})
|
||||
export class WorkerRuntimeModule {}
|
||||
@@ -0,0 +1,330 @@
|
||||
import { Logger } from '@nestjs/common';
|
||||
import { Prisma } from '../../generated/prisma';
|
||||
import type { CalendarSyncService } from '../calendar-sync/calendar-sync.service';
|
||||
import type { WakeChannel } from '../common/worker-wakeup';
|
||||
import {
|
||||
DomainEventDispatcher,
|
||||
DRAIN_MAX_ROUNDS,
|
||||
} from '../domain-events/domain-event-dispatcher';
|
||||
import type { PrismaService } from '../prisma/prisma.service';
|
||||
import { PgListener, PgListenerOptions } from './pg-listener';
|
||||
import { HEALTH_STALE_MS } from './worker-health';
|
||||
import {
|
||||
CALENDAR_OUTBOX_POLL_MS,
|
||||
DOMAIN_EVENT_POLL_MS,
|
||||
HEALTH_QUERY_TIMEOUT_MS,
|
||||
WorkerRuntime,
|
||||
} from './worker-runtime.service';
|
||||
|
||||
interface FakeListener {
|
||||
start: jest.Mock<Promise<void>, []>;
|
||||
stop: jest.Mock<Promise<void>, []>;
|
||||
}
|
||||
|
||||
jest.mock('./pg-listener', () => ({ PgListener: jest.fn() }));
|
||||
|
||||
const DATABASE_URL = 'postgresql://worker:secret@db/app';
|
||||
|
||||
function deferred(): { promise: Promise<void>; resolve: () => void } {
|
||||
let resolve!: () => void;
|
||||
const promise = new Promise<void>((r) => {
|
||||
resolve = r;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
/** Advances lastTickAt like the real dispatcher: a new stamp per run. */
|
||||
class FakeDispatcher {
|
||||
lastTickAt: Date | null = null;
|
||||
releaseStaleClaims = jest.fn<Promise<number>, []>().mockResolvedValue(0);
|
||||
drain = jest.fn<Promise<number>, unknown[]>(() => {
|
||||
this.tick();
|
||||
return Promise.resolve(0);
|
||||
});
|
||||
|
||||
tick(): void {
|
||||
this.lastTickAt = new Date();
|
||||
}
|
||||
}
|
||||
|
||||
describe('WorkerRuntime', () => {
|
||||
const ListenerMock = PgListener as unknown as jest.Mock;
|
||||
const originalDatabaseUrl = process.env.DATABASE_URL;
|
||||
let listener: FakeListener;
|
||||
let prisma: { $queryRaw: jest.Mock };
|
||||
let dispatcher: FakeDispatcher;
|
||||
let calendarSync: {
|
||||
processOutbox: jest.Mock<Promise<void>, [AbortSignal?]>;
|
||||
};
|
||||
let runtime: WorkerRuntime;
|
||||
let logError: jest.SpyInstance;
|
||||
|
||||
const listenerOptions = (): PgListenerOptions =>
|
||||
(ListenerMock.mock.calls[0] as [PgListenerOptions])[0];
|
||||
const notify = (channel: WakeChannel) => listenerOptions().onNotify(channel);
|
||||
|
||||
/** Bootstraps the runtime and lets the first ticks run. */
|
||||
async function bootstrap(): Promise<void> {
|
||||
await runtime.onApplicationBootstrap();
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
jest.useFakeTimers();
|
||||
process.env.DATABASE_URL = DATABASE_URL;
|
||||
listener = {
|
||||
start: jest.fn<Promise<void>, []>().mockResolvedValue(undefined),
|
||||
stop: jest.fn<Promise<void>, []>().mockResolvedValue(undefined),
|
||||
};
|
||||
ListenerMock.mockReset().mockImplementation(() => listener);
|
||||
prisma = { $queryRaw: jest.fn().mockResolvedValue([{ ok: 1 }]) };
|
||||
dispatcher = new FakeDispatcher();
|
||||
calendarSync = {
|
||||
processOutbox: jest
|
||||
.fn<Promise<void>, [AbortSignal?]>()
|
||||
.mockResolvedValue(undefined),
|
||||
};
|
||||
logError = jest
|
||||
.spyOn(Logger.prototype, 'error')
|
||||
.mockImplementation(() => undefined);
|
||||
jest.spyOn(Logger.prototype, 'warn').mockImplementation(() => undefined);
|
||||
runtime = new WorkerRuntime(
|
||||
prisma as unknown as PrismaService,
|
||||
dispatcher as unknown as DomainEventDispatcher,
|
||||
calendarSync as unknown as CalendarSyncService,
|
||||
);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await runtime.onModuleDestroy();
|
||||
jest.restoreAllMocks();
|
||||
jest.useRealTimers();
|
||||
// Assigning undefined would store the string "undefined".
|
||||
if (originalDatabaseUrl === undefined) {
|
||||
delete process.env.DATABASE_URL;
|
||||
} else {
|
||||
process.env.DATABASE_URL = originalDatabaseUrl;
|
||||
}
|
||||
});
|
||||
|
||||
it('exposes the documented poll intervals', () => {
|
||||
expect(DOMAIN_EVENT_POLL_MS).toBe(5_000);
|
||||
expect(CALENDAR_OUTBOX_POLL_MS).toBe(60_000);
|
||||
expect(HEALTH_QUERY_TIMEOUT_MS).toBe(5_000);
|
||||
});
|
||||
|
||||
it('starts the listener and every loop at bootstrap', async () => {
|
||||
await runtime.onApplicationBootstrap();
|
||||
|
||||
expect(ListenerMock).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
connectionString: DATABASE_URL,
|
||||
channels: ['domain_event', 'calendar_sync'],
|
||||
}),
|
||||
);
|
||||
expect(listener.start).toHaveBeenCalledTimes(1);
|
||||
// LISTEN is up before the first tick: nothing committed in between
|
||||
// can be missed until the next poll.
|
||||
expect(dispatcher.drain).not.toHaveBeenCalled();
|
||||
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
|
||||
expect(dispatcher.releaseStaleClaims).toHaveBeenCalledTimes(1);
|
||||
expect(dispatcher.drain).toHaveBeenCalledTimes(1);
|
||||
// Every event of the database (no filter), stoppable by shutdown.
|
||||
expect(dispatcher.drain).toHaveBeenCalledWith(
|
||||
undefined,
|
||||
DRAIN_MAX_ROUNDS,
|
||||
expect.any(AbortSignal),
|
||||
);
|
||||
const [releaseOrder] =
|
||||
dispatcher.releaseStaleClaims.mock.invocationCallOrder;
|
||||
const [drainOrder] = dispatcher.drain.mock.invocationCallOrder;
|
||||
expect(releaseOrder).toBeLessThan(drainOrder);
|
||||
expect(calendarSync.processOutbox).toHaveBeenCalledWith(
|
||||
expect.any(AbortSignal),
|
||||
);
|
||||
});
|
||||
|
||||
it('polls domain events every 5 s and the calendar outbox every 60 s', async () => {
|
||||
await bootstrap();
|
||||
|
||||
await jest.advanceTimersByTimeAsync(DOMAIN_EVENT_POLL_MS);
|
||||
expect(dispatcher.drain).toHaveBeenCalledTimes(2);
|
||||
expect(calendarSync.processOutbox).toHaveBeenCalledTimes(1);
|
||||
|
||||
await jest.advanceTimersByTimeAsync(
|
||||
CALENDAR_OUTBOX_POLL_MS - DOMAIN_EVENT_POLL_MS,
|
||||
);
|
||||
expect(dispatcher.drain).toHaveBeenCalledTimes(
|
||||
CALENDAR_OUTBOX_POLL_MS / DOMAIN_EVENT_POLL_MS + 1,
|
||||
);
|
||||
expect(calendarSync.processOutbox).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('wakes the domain-events loop on a domain_event notification', async () => {
|
||||
await bootstrap();
|
||||
|
||||
notify('domain_event');
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
|
||||
expect(dispatcher.drain).toHaveBeenCalledTimes(2);
|
||||
expect(calendarSync.processOutbox).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('wakes the calendar-outbox loop on a calendar_sync notification', async () => {
|
||||
await bootstrap();
|
||||
|
||||
notify('calendar_sync');
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
|
||||
expect(calendarSync.processOutbox).toHaveBeenCalledTimes(2);
|
||||
expect(dispatcher.drain).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('is healthy after a recent tick and dispatcher-stale after 120 s without one', async () => {
|
||||
await bootstrap();
|
||||
expect(await runtime.health()).toEqual({ ok: true });
|
||||
|
||||
// From the next tick on the dispatcher hangs and stamps nothing more.
|
||||
const stuck = deferred();
|
||||
dispatcher.drain.mockReturnValue(stuck.promise.then(() => 0));
|
||||
|
||||
await jest.advanceTimersByTimeAsync(HEALTH_STALE_MS);
|
||||
expect(await runtime.health()).toEqual({ ok: true });
|
||||
await jest.advanceTimersByTimeAsync(1);
|
||||
expect(await runtime.health()).toEqual({
|
||||
ok: false,
|
||||
reason: 'dispatcher-stale',
|
||||
});
|
||||
|
||||
stuck.resolve();
|
||||
});
|
||||
|
||||
// lastTickAt advances per dispatched event: a long batch is progress.
|
||||
it('stays healthy while a long batch keeps dispatching events', async () => {
|
||||
await bootstrap();
|
||||
const batch = deferred();
|
||||
dispatcher.drain.mockReturnValue(batch.promise.then(() => 0));
|
||||
await jest.advanceTimersByTimeAsync(DOMAIN_EVENT_POLL_MS);
|
||||
|
||||
await jest.advanceTimersByTimeAsync(HEALTH_STALE_MS - 1_000);
|
||||
dispatcher.tick(); // one more event of the running batch dispatched
|
||||
expect(await runtime.health()).toEqual({ ok: true });
|
||||
|
||||
await jest.advanceTimersByTimeAsync(HEALTH_STALE_MS);
|
||||
expect(await runtime.health()).toEqual({ ok: true });
|
||||
await jest.advanceTimersByTimeAsync(1);
|
||||
expect(await runtime.health()).toEqual({
|
||||
ok: false,
|
||||
reason: 'dispatcher-stale',
|
||||
});
|
||||
|
||||
batch.resolve();
|
||||
});
|
||||
|
||||
// A wall-clock step must neither fake nor hide a stall.
|
||||
it('measures the dispatcher age on the monotonic clock', async () => {
|
||||
await bootstrap();
|
||||
const stuck = deferred();
|
||||
dispatcher.drain.mockReturnValue(stuck.promise.then(() => 0));
|
||||
|
||||
jest.setSystemTime(Date.now() + 10 * HEALTH_STALE_MS);
|
||||
expect(await runtime.health()).toEqual({ ok: true });
|
||||
|
||||
stuck.resolve();
|
||||
});
|
||||
|
||||
it('counts the time since bootstrap while the dispatcher never ran', async () => {
|
||||
dispatcher.drain.mockRejectedValue(new Error('database down'));
|
||||
await bootstrap();
|
||||
|
||||
await jest.advanceTimersByTimeAsync(HEALTH_STALE_MS);
|
||||
expect(await runtime.health()).toEqual({ ok: true });
|
||||
await jest.advanceTimersByTimeAsync(1);
|
||||
expect(await runtime.health()).toEqual({
|
||||
ok: false,
|
||||
reason: 'dispatcher-stale',
|
||||
});
|
||||
});
|
||||
|
||||
it('reports database when SELECT 1 fails or exceeds 5 s', async () => {
|
||||
await bootstrap();
|
||||
|
||||
prisma.$queryRaw.mockRejectedValueOnce(new Error('connection refused'));
|
||||
expect(await runtime.health()).toEqual({ ok: false, reason: 'database' });
|
||||
|
||||
prisma.$queryRaw.mockReturnValueOnce(new Promise(() => undefined));
|
||||
const slow = runtime.health();
|
||||
await jest.advanceTimersByTimeAsync(HEALTH_QUERY_TIMEOUT_MS);
|
||||
await expect(slow).resolves.toEqual({ ok: false, reason: 'database' });
|
||||
});
|
||||
|
||||
it('logs a failed health query without the Prisma message', async () => {
|
||||
await bootstrap();
|
||||
prisma.$queryRaw.mockRejectedValueOnce(
|
||||
new Prisma.PrismaClientKnownRequestError('password "hunter2"', {
|
||||
code: 'P1001',
|
||||
clientVersion: '7.9.1',
|
||||
}),
|
||||
);
|
||||
|
||||
await runtime.health();
|
||||
|
||||
const [message, stack] = logError.mock.calls[0] as [string, unknown];
|
||||
expect(message).toContain('P1001');
|
||||
expect(message).not.toContain('hunter2');
|
||||
expect(stack).toBeUndefined();
|
||||
});
|
||||
|
||||
// A running tick must not claim up to 19 more batches of 50 events or
|
||||
// 19 more calendar entries while the pod is being stopped.
|
||||
it('hands the running work a stop signal that shutdown aborts', async () => {
|
||||
await bootstrap();
|
||||
const [, , drainSignal] = dispatcher.drain.mock.calls[0] as [
|
||||
unknown,
|
||||
unknown,
|
||||
AbortSignal,
|
||||
];
|
||||
const [outboxSignal] = calendarSync.processOutbox.mock.calls[0];
|
||||
expect(drainSignal.aborted).toBe(false);
|
||||
expect(outboxSignal).toBe(drainSignal);
|
||||
|
||||
await runtime.onModuleDestroy();
|
||||
|
||||
expect(drainSignal.aborted).toBe(true);
|
||||
});
|
||||
|
||||
it('stops loops and listener on shutdown', async () => {
|
||||
await bootstrap();
|
||||
|
||||
await runtime.onModuleDestroy();
|
||||
|
||||
expect(listener.stop).toHaveBeenCalledTimes(1);
|
||||
notify('domain_event');
|
||||
await jest.advanceTimersByTimeAsync(CALENDAR_OUTBOX_POLL_MS * 2);
|
||||
expect(dispatcher.drain).toHaveBeenCalledTimes(1);
|
||||
expect(calendarSync.processOutbox).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
// A pass cut off mid-entry would only be taken over after the 5 min
|
||||
// stale-claim window - and Prisma disconnects right after this hook.
|
||||
it('lets shutdown wait for a running tick', async () => {
|
||||
const pass = deferred();
|
||||
calendarSync.processOutbox.mockReturnValueOnce(pass.promise);
|
||||
await bootstrap();
|
||||
let stopped = false;
|
||||
|
||||
const shutdown = runtime.onModuleDestroy().then(() => {
|
||||
stopped = true;
|
||||
});
|
||||
await jest.advanceTimersByTimeAsync(0);
|
||||
expect(stopped).toBe(false);
|
||||
// The running pass has been told to stop after its current entry.
|
||||
expect(calendarSync.processOutbox.mock.calls[0][0]?.aborted).toBe(true);
|
||||
|
||||
pass.resolve();
|
||||
await shutdown;
|
||||
expect(stopped).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,175 @@
|
||||
import {
|
||||
Injectable,
|
||||
Logger,
|
||||
OnApplicationBootstrap,
|
||||
OnModuleDestroy,
|
||||
} from '@nestjs/common';
|
||||
import { CalendarSyncService } from '../calendar-sync/calendar-sync.service';
|
||||
import { loggableError } from '../common/loggable-error';
|
||||
import { withTimeout } from '../common/with-timeout';
|
||||
import { WAKE_CHANNELS, WakeChannel } from '../common/worker-wakeup';
|
||||
import {
|
||||
DomainEventDispatcher,
|
||||
DRAIN_MAX_ROUNDS,
|
||||
} from '../domain-events/domain-event-dispatcher';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { PgListener } from './pg-listener';
|
||||
import { PollingLoop } from './polling-loop';
|
||||
import { HEALTH_STALE_MS, WorkerHealth } from './worker-health';
|
||||
|
||||
export const DOMAIN_EVENT_POLL_MS = 5_000;
|
||||
export const CALENDAR_OUTBOX_POLL_MS = 60_000;
|
||||
/** The health check's SELECT 1 must answer within this. */
|
||||
export const HEALTH_QUERY_TIMEOUT_MS = 5_000;
|
||||
|
||||
/** One background loop: its poll interval and the channel that wakes it. */
|
||||
interface WorkerLoop {
|
||||
name: string;
|
||||
intervalMs: number;
|
||||
channel: WakeChannel;
|
||||
tick: () => Promise<unknown>;
|
||||
}
|
||||
|
||||
function databaseUrl(): string {
|
||||
const url = process.env.DATABASE_URL;
|
||||
if (!url) {
|
||||
throw new Error('DATABASE_URL is required for the LISTEN connection');
|
||||
}
|
||||
return url;
|
||||
}
|
||||
|
||||
/**
|
||||
* The background work of the worker role (dist/worker.js; also "all" for
|
||||
* local development and tests): one single-flight polling loop per outbox,
|
||||
* woken early through LISTEN/NOTIFY. Every loop claims its rows, so more
|
||||
* worker pods are safe. The API role never loads this module
|
||||
* (runtime/root-module.ts).
|
||||
*/
|
||||
@Injectable()
|
||||
export class WorkerRuntime implements OnApplicationBootstrap, OnModuleDestroy {
|
||||
private readonly logger = new Logger(WorkerRuntime.name);
|
||||
private readonly loops: ReadonlyMap<WakeChannel, PollingLoop>;
|
||||
private listener: PgListener | undefined;
|
||||
/** Aborted at shutdown: running ticks claim no further work. */
|
||||
private readonly stopping = new AbortController();
|
||||
/** Monotonic time (performance.now) of the last dispatcher progress. */
|
||||
private progressAt = performance.now();
|
||||
/** The dispatcher's lastTickAt as last seen; a new value is progress. */
|
||||
private seenTickAt: Date | null = null;
|
||||
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly dispatcher: DomainEventDispatcher,
|
||||
private readonly calendarSync: CalendarSyncService,
|
||||
) {
|
||||
this.loops = new Map(
|
||||
this.workerLoops().map(({ channel, name, intervalMs, tick }) => [
|
||||
channel,
|
||||
new PollingLoop({ name, intervalMs, tick, logger: this.logger }),
|
||||
]),
|
||||
);
|
||||
}
|
||||
|
||||
/** One row per outbox the worker serves. */
|
||||
private workerLoops(): readonly WorkerLoop[] {
|
||||
return [
|
||||
{
|
||||
name: 'domain-events',
|
||||
intervalMs: DOMAIN_EVENT_POLL_MS,
|
||||
channel: WAKE_CHANNELS.domainEvents,
|
||||
tick: () => this.dispatchDomainEvents(),
|
||||
},
|
||||
{
|
||||
name: 'calendar-outbox',
|
||||
intervalMs: CALENDAR_OUTBOX_POLL_MS,
|
||||
channel: WAKE_CHANNELS.calendarSync,
|
||||
tick: () => this.calendarSync.processOutbox(this.stopping.signal),
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
async onApplicationBootstrap(): Promise<void> {
|
||||
this.progressAt = performance.now();
|
||||
this.listener = new PgListener({
|
||||
connectionString: databaseUrl(),
|
||||
channels: [...this.loops.keys()],
|
||||
onNotify: (channel) => this.loops.get(channel)?.wake(),
|
||||
logger: this.logger,
|
||||
});
|
||||
// LISTEN first: work committed from now on wakes a loop, and work
|
||||
// committed before is found by the first ticks, which run right away.
|
||||
await this.listener.start();
|
||||
this.loops.forEach((loop) => loop.start());
|
||||
}
|
||||
|
||||
// Not onApplicationShutdown: Nest runs every onModuleDestroy first, and
|
||||
// PrismaService disconnects there. A running tick finishes the entry or
|
||||
// batch it holds but claims nothing more; stopping waits for it - a tick
|
||||
// cut off mid-entry leaves a claim that is only taken over after the
|
||||
// 5 min stale-claim window.
|
||||
async onModuleDestroy(): Promise<void> {
|
||||
this.stopping.abort();
|
||||
const listener = this.listener;
|
||||
this.listener = undefined;
|
||||
await listener?.stop();
|
||||
await Promise.all([...this.loops.values()].map((loop) => loop.stop()));
|
||||
}
|
||||
|
||||
/**
|
||||
* Unhealthy if the database does not answer SELECT 1 within 5 s, or if
|
||||
* the dispatcher has made no progress for longer than HEALTH_STALE_MS.
|
||||
*/
|
||||
async health(): Promise<WorkerHealth> {
|
||||
try {
|
||||
await withTimeout(
|
||||
this.prisma.$queryRaw`SELECT 1`,
|
||||
HEALTH_QUERY_TIMEOUT_MS,
|
||||
'Worker health query timed out',
|
||||
);
|
||||
} catch (error: unknown) {
|
||||
// Prisma errors by name/code/model only (see loggableError).
|
||||
const { message, stack } = loggableError(error);
|
||||
this.logger.error(`Worker health query failed: ${message}`, stack);
|
||||
return { ok: false, reason: 'database' };
|
||||
}
|
||||
const idleMs = performance.now() - this.noteDispatcherProgress();
|
||||
if (idleMs > HEALTH_STALE_MS) {
|
||||
this.logger.warn(
|
||||
`Domain event dispatcher made no progress for ${Math.round(idleMs)} ms`,
|
||||
);
|
||||
return { ok: false, reason: 'dispatcher-stale' };
|
||||
}
|
||||
return { ok: true };
|
||||
}
|
||||
|
||||
/**
|
||||
* The dispatcher stamps lastTickAt with the wall clock on every claim and
|
||||
* every dispatched event. Its age is taken on the monotonic clock instead,
|
||||
* so a clock step can neither fake nor hide a stall: a new stamp counts
|
||||
* from the moment it is first seen - after every tick, and on every
|
||||
* health check while a long batch is still running. Before the first
|
||||
* stamp, the time since bootstrap counts.
|
||||
*/
|
||||
private noteDispatcherProgress(): number {
|
||||
const tickAt = this.dispatcher.lastTickAt;
|
||||
if (tickAt !== this.seenTickAt) {
|
||||
this.seenTickAt = tickAt;
|
||||
this.progressAt = performance.now();
|
||||
}
|
||||
return this.progressAt;
|
||||
}
|
||||
|
||||
private async dispatchDomainEvents(): Promise<void> {
|
||||
try {
|
||||
await this.dispatcher.releaseStaleClaims();
|
||||
// No filter: every event of the database.
|
||||
await this.dispatcher.drain(
|
||||
undefined,
|
||||
DRAIN_MAX_ROUNDS,
|
||||
this.stopping.signal,
|
||||
);
|
||||
} finally {
|
||||
this.noteDispatcherProgress();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { ConfigModule } from '@nestjs/config';
|
||||
import { AuditModule } from '../audit/audit.module';
|
||||
import { AutomationModule } from '../automation/automation.module';
|
||||
import { CalendarSyncModule } from '../calendar-sync/calendar-sync.module';
|
||||
import { DomainEventsModule } from '../domain-events/domain-events.module';
|
||||
import { NotificationsModule } from '../notifications/notifications.module';
|
||||
import { PrismaModule } from '../prisma/prisma.module';
|
||||
import { TasksModule } from '../tasks/tasks.module';
|
||||
import { WebhooksModule } from '../webhooks/webhooks.module';
|
||||
import { WorkerHealthServer } from './worker-health-server.service';
|
||||
import { WorkerRuntimeModule } from './worker-runtime.module';
|
||||
|
||||
/**
|
||||
* Root module of dist/worker.js, started with
|
||||
* NestFactory.createApplicationContext: the modules the worker loops and
|
||||
* the domain event handlers need, without the HTTP layer (no guards,
|
||||
* interceptors, throttler or routes - controllers of the imported modules
|
||||
* are never served). The only port is WorkerHealthServer's GET /health.
|
||||
*/
|
||||
@Module({
|
||||
imports: [
|
||||
ConfigModule.forRoot({ isGlobal: true }),
|
||||
PrismaModule,
|
||||
AuditModule,
|
||||
NotificationsModule,
|
||||
DomainEventsModule,
|
||||
WebhooksModule,
|
||||
CalendarSyncModule,
|
||||
TasksModule,
|
||||
AutomationModule,
|
||||
WorkerRuntimeModule,
|
||||
],
|
||||
providers: [WorkerHealthServer],
|
||||
})
|
||||
export class WorkerModule {}
|
||||
Reference in new issue
Block a user