wehub-resource-sync
106 lines
5.1 KiB
TypeScript
106 lines
5.1 KiB
TypeScript
import { MollifierDrainer } from "@trigger.dev/redis-worker";
|
|
import { prisma } from "~/db.server";
|
|
import { env } from "~/env.server";
|
|
import { engine as runEngine } from "~/v3/runEngine.server";
|
|
import { logger } from "~/services/logger.server";
|
|
import { singleton } from "~/utils/singleton";
|
|
import { getMollifierBuffer } from "./mollifierBuffer.server";
|
|
import {
|
|
createDrainerHandler,
|
|
createDrainerTerminalFailureHandler,
|
|
isRetryablePgError,
|
|
} from "./mollifierDrainerHandler.server";
|
|
import type { MollifierSnapshot } from "./mollifierSnapshot.server";
|
|
|
|
// Distinct error class for the deterministic "fail loud at boot" throws
|
|
// below. The bootstrap in `mollifierDrainerWorker.server.ts` catches
|
|
// transient/init errors and logs them so an unrelated Redis blip doesn't
|
|
// crash the webapp, but it RETHROWS this class — a misconfigured
|
|
// shutdown timeout or missing buffer is a deploy-time mistake that
|
|
// should fail health checks and roll back, not silently disable a
|
|
// half-rolled-out feature.
|
|
//
|
|
// The `name` getter is set explicitly so cross-realm `instanceof` checks
|
|
// (e.g. when Remix dev hot-reloads the module and the consumer keeps a
|
|
// reference to the old class) can fall back to `error.name === ...` and
|
|
// still recognise the marker.
|
|
export class MollifierConfigurationError extends Error {
|
|
constructor(message: string) {
|
|
super(message);
|
|
this.name = "MollifierConfigurationError";
|
|
}
|
|
}
|
|
|
|
function initializeMollifierDrainer(): MollifierDrainer<MollifierSnapshot> {
|
|
const buffer = getMollifierBuffer();
|
|
if (!buffer) {
|
|
// Unreachable in normal config: getMollifierDrainer() gates on the
|
|
// same env flag as getMollifierBuffer(). If we hit this, fail loud
|
|
// — the operator has set TRIGGER_MOLLIFIER_ENABLED=1 on a worker pod but
|
|
// the buffer can't initialise (e.g. TRIGGER_MOLLIFIER_REDIS_HOST resolves
|
|
// to nothing). Crashing surfaces the misconfig immediately rather
|
|
// than silently leaving entries un-drained.
|
|
throw new MollifierConfigurationError(
|
|
"MollifierDrainer initialised without a buffer — env vars inconsistent"
|
|
);
|
|
}
|
|
|
|
// Validate BEFORE start() so a misconfigured shutdown timeout fails
|
|
// loud at module-load time and the singleton is never cached. If start()
|
|
// ran first and the throw propagated out, the loop would already be
|
|
// polling with no SIGTERM handler registered by the caller — exactly
|
|
// the failure mode the validation is supposed to prevent.
|
|
//
|
|
// The SIGTERM handler in mollifierDrainerWorker.server.ts is sync fire-and-forget:
|
|
// `drainer.stop({ timeoutMs })` returns a promise that keeps the event
|
|
// loop alive, but in cluster mode the primary runs its own
|
|
// GRACEFUL_SHUTDOWN_TIMEOUT and will call `process.exit(0)`
|
|
// independently. If the drainer's deadline exceeds the primary's, the
|
|
// drainer is cut off mid-wait — "log a warning on timeout" turns into
|
|
// "hard exit with no log". 1s margin gives the primary room to finish
|
|
// its own teardown after the drainer settles.
|
|
const shutdownMarginMs = env.TRIGGER_MOLLIFIER_DRAIN_SHUTDOWN_MARGIN_MS;
|
|
if (
|
|
env.TRIGGER_MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS >=
|
|
env.GRACEFUL_SHUTDOWN_TIMEOUT - shutdownMarginMs
|
|
) {
|
|
throw new MollifierConfigurationError(
|
|
`TRIGGER_MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS (${env.TRIGGER_MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS}) must be at least ${shutdownMarginMs}ms below GRACEFUL_SHUTDOWN_TIMEOUT (${env.GRACEFUL_SHUTDOWN_TIMEOUT}); otherwise the primary's hard exit shadows the drainer's deadline.`
|
|
);
|
|
}
|
|
|
|
logger.debug("Initializing mollifier drainer", {
|
|
concurrency: env.TRIGGER_MOLLIFIER_DRAIN_CONCURRENCY,
|
|
maxAttempts: env.TRIGGER_MOLLIFIER_DRAIN_MAX_ATTEMPTS,
|
|
drainBatchSize: env.TRIGGER_MOLLIFIER_DRAIN_BATCH_SIZE,
|
|
});
|
|
|
|
const drainer = new MollifierDrainer<MollifierSnapshot>({
|
|
buffer,
|
|
handler: createDrainerHandler({ engine: runEngine, prisma }),
|
|
onTerminalFailure: createDrainerTerminalFailureHandler({ engine: runEngine, prisma }),
|
|
concurrency: env.TRIGGER_MOLLIFIER_DRAIN_CONCURRENCY,
|
|
maxAttempts: env.TRIGGER_MOLLIFIER_DRAIN_MAX_ATTEMPTS,
|
|
maxOrgsPerTick: env.TRIGGER_MOLLIFIER_DRAIN_MAX_ORGS_PER_TICK,
|
|
drainBatchSize: env.TRIGGER_MOLLIFIER_DRAIN_BATCH_SIZE,
|
|
pollIntervalMs: env.TRIGGER_MOLLIFIER_DRAIN_POLL_INTERVAL_MS,
|
|
maxBackoffMs: env.TRIGGER_MOLLIFIER_DRAIN_MAX_BACKOFF_MS,
|
|
backoffFloorMs: env.TRIGGER_MOLLIFIER_DRAIN_BACKOFF_FLOOR_MS,
|
|
isRetryable: isRetryablePgError,
|
|
});
|
|
|
|
return drainer;
|
|
}
|
|
|
|
// Returns a configured-but-stopped drainer. Callers MUST register their
|
|
// SIGTERM / SIGINT shutdown handlers before invoking `drainer.start()` —
|
|
// see `apps/webapp/app/v3/mollifierDrainerWorker.server.ts`. Starting
|
|
// inside the singleton factory would put the polling loop ahead of
|
|
// handler registration, leaving a narrow window where a SIGTERM landing
|
|
// between `start()` and `process.once("SIGTERM", ...)` would skip the
|
|
// graceful stop. The split is intentional.
|
|
export function getMollifierDrainer(): MollifierDrainer<MollifierSnapshot> | null {
|
|
if (env.TRIGGER_MOLLIFIER_ENABLED !== "1") return null;
|
|
return singleton("mollifierDrainer", initializeMollifierDrainer);
|
|
}
|