refactor(webapp): wire mollifier drainer shutdown through signalsEmitter
`process.once("SIGTERM", stopDrainer)` was the odd one out — every
other webapp service (runsReplicationInstance, llmPricingRegistry,
dynamicFlushScheduler, marqs, eventLoopMonitor) registers through
`signalsEmitter` from `~/services/signals.server`, an EventEmitter
backed by a single `process.on()` that fans out to all listeners.
Switching gets us:
- codebase consistency;
- `.on` (not `.once`) so a second SIGTERM, if the orchestrator emits
one before SIGKILL, still reaches us;
- if SIGTERM lands in the narrow gap between the listener attaching
and drainer.start() below, the first invocation no-ops (stop()
returns early because isRunning is false) but the listener stays
attached for any subsequent signal, instead of being consumed and
leaving the now-running drainer with no graceful-stop path.
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
import { env } from "~/env.server";
|
||||
import { logger } from "~/services/logger.server";
|
||||
import { signalsEmitter } from "~/services/signals.server";
|
||||
import { getMollifierDrainer } from "./mollifier/mollifierDrainer.server";
|
||||
|
||||
declare global {
|
||||
@@ -52,6 +53,19 @@ export function initMollifierDrainerWorker(): void {
|
||||
// entry.server.tsx, which Remix dev re-evaluates on every change).
|
||||
// Same guard owns both the handler registration and the start()
|
||||
// call so the two never get out of sync.
|
||||
//
|
||||
// Registers through `signalsEmitter` (the webapp-wide singleton in
|
||||
// `~/services/signals.server`) rather than `process.once` directly:
|
||||
// - matches the codebase convention (runsReplicationInstance,
|
||||
// llmPricingRegistry, dynamicFlushScheduler etc. all listen on
|
||||
// the same emitter);
|
||||
// - `.on` (not `.once`) means a second SIGTERM still reaches us if
|
||||
// the orchestrator delivers more than one signal before SIGKILL;
|
||||
// - if SIGTERM lands in the gap between this listener attaching
|
||||
// and `drainer.start()` below, the first invocation no-ops
|
||||
// (stop() returns early because the drainer isn't running yet)
|
||||
// but the listener stays attached for a subsequent signal,
|
||||
// rather than being consumed by `once`.
|
||||
const stopDrainer = () => {
|
||||
drainer
|
||||
.stop({ timeoutMs: env.TRIGGER_MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS })
|
||||
@@ -59,8 +73,8 @@ export function initMollifierDrainerWorker(): void {
|
||||
logger.error("Failed to stop mollifier drainer", { error });
|
||||
});
|
||||
};
|
||||
process.once("SIGTERM", stopDrainer);
|
||||
process.once("SIGINT", stopDrainer);
|
||||
signalsEmitter.on("SIGTERM", stopDrainer);
|
||||
signalsEmitter.on("SIGINT", stopDrainer);
|
||||
global.__mollifierShutdownRegistered__ = true;
|
||||
drainer.start();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user