Something went wrong. Try again.
Automations and webhooks for the AT Protocol airglow.run
automation webhook atproto atprotocol
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182/** Floor on how long `stop()` waits for an in-flight tick. The shutdown * budget is the dominant value; this just guarantees a tick gets at least * one second to land even if the caller passes 0. Rows a runner leaves * claimed past its recovery window are reset on the next process's first * tick, so a missed wait is recoverable. */const SHUTDOWN_TICK_WAIT_MIN_MS = 1_000;
export type IntervalRunnerHandle = { /** Stop scheduling new ticks. If a tick is in flight, await it (up to * `timeoutMs`) so partial batches don't leave rows half-claimed. */ stop(timeoutMs?: number): Promise<void>;};
/** Drive `run` once on start (to drain anything that came due during * downtime) and then every `tickMs`, never overlapping two ticks. */export function startIntervalRunner(opts: { name: string; tickMs: number; run: () => Promise<void>;}): IntervalRunnerHandle { let tickInFlight = false; let activeTick: Promise<void> | null = null; let shuttingDown = false;
async function tick() { // Re-entry guard at the tick level. Any fan-out inside `run` is the // parallelism layer; this guard only prevents back-to-back ticks (a long // tick still running when the interval fires) from doubling up. if (tickInFlight || shuttingDown) return; tickInFlight = true; try { await opts.run(); } catch (err) { console.error(`[${opts.name}] tick failed:`, err); } finally { tickInFlight = false; } }
function startTrackedTick(): void { // Skipped ticks must not replace `activeTick`, or `stop()` would await // an instantly resolved no-op instead of the tick still doing work. if (tickInFlight || shuttingDown) return; const p = tick(); activeTick = p; void p.finally(() => { if (activeTick === p) activeTick = null; }); }
startTrackedTick(); let interval: ReturnType<typeof setInterval> | null = setInterval(startTrackedTick, opts.tickMs);
return { async stop(timeoutMs = 0) { // `shuttingDown` only blocks NEW ticks. A tick already running keeps // going; we wait below for it to finish naturally rather than abort // partway. shuttingDown = true; if (interval !== null) { clearInterval(interval); interval = null; } const pending = activeTick; if (!pending) return; const waitMs = Math.max(timeoutMs, SHUTDOWN_TICK_WAIT_MIN_MS); // Capture + clear the timeout so a clean shutdown (pending resolves // first) doesn't leave a dangling setTimeout holding the process alive // for up to waitMs. let timeoutHandle: ReturnType<typeof setTimeout> | undefined; const deadline = new Promise<void>((resolve) => { timeoutHandle = setTimeout(resolve, waitMs); }); try { await Promise.race([pending.catch(() => {}), deadline]); } finally { if (timeoutHandle !== undefined) clearTimeout(timeoutHandle); } }, };}