Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 42 additions & 2 deletions svc/store/queue.ts
Original file line number Diff line number Diff line change
Expand Up @@ -120,15 +120,55 @@ export async function runReliableQueue(
}

/**
* Runs an async function on a loop, waiting the delay between each iteration
* Runs an async function on a loop, waiting the delay between each iteration.
*
* On failure, logs and sleeps with exponential backoff + jitter (capped at 30s)
* before retrying, so a sick dependency (e.g. Postgres at max_connections) is
* not hammered by tight-loop restarts.
*
* After MAX_CONSECUTIVE_FAILURES in a row, exits the process so PM2 can do a
* clean-slate restart (fresh Knex pool, cleared in-memory state).
*
* The lastRun health beacon is only refreshed on success, so the existing
* HEALTH_TIMEOUT alerting still detects a stuck worker before exit.
*
* @param func
* @param delay
*/
export async function runInLoop(func: () => Promise<void>, delay: number) {
const BASE_BACKOFF_MS = 250;
const MAX_BACKOFF_MS = 30_000;
Comment thread
ff137 marked this conversation as resolved.
const MAX_CONSECUTIVE_FAILURES = 10;
let consecutiveFailures = 0;
while (true) {
console.log("running %s", func.name);
const start = Date.now();
await func();
try {
await func();
} catch (e) {
consecutiveFailures += 1;
console.error(
"%s failed (consecutive failures: %d):",
func.name,
consecutiveFailures,
e,
);
if (consecutiveFailures >= MAX_CONSECUTIVE_FAILURES) {
console.error(
"%s exceeded %d consecutive failures, exiting for PM2 restart",
func.name,
MAX_CONSECUTIVE_FAILURES,
);
process.exit(1);
}
// 2 ** capped so the exponent doesn't overflow on long outages
const exp = Math.min(consecutiveFailures, 10);
Comment thread
ff137 marked this conversation as resolved.
const backoff = Math.min(MAX_BACKOFF_MS, BASE_BACKOFF_MS * 2 ** exp);
const jitter = Math.random() * 250;
await new Promise((resolve) => setTimeout(resolve, backoff + jitter));
continue;
Comment thread
ff137 marked this conversation as resolved.
}
consecutiveFailures = 0;
const end = Date.now();
console.log("%s: %dms", func.name, end - start);
await redis.setex(
Expand Down
Loading