diff --git a/src/orb/relay.ts b/src/orb/relay.ts index a7762c489d..59f512d841 100644 --- a/src/orb/relay.ts +++ b/src/orb/relay.ts @@ -141,6 +141,11 @@ export async function registerOrbRelay(env: Env, secret: string, relayUrl: strin const RELAY_RETRY_MAX_ATTEMPTS = 5; const RELAY_RETRY_BATCH_SIZE = 25; const RELAY_RETRY_CONCURRENCY = 5; +// Per-failure backoff (#1950): a row that just failed is not retried again until this window elapses, so a +// sustained outage does not re-attempt the whole failed-relay backlog on every ~2-min cron tick — which, fleet-wide, +// is a synchronized POST storm against the central Orb exactly when it is already degraded. Never-attempted rows +// (last_attempt_at IS NULL) stay immediately eligible, so a transient blip still recovers on the very next tick. +const RELAY_RETRY_BACKOFF_MINUTES = 5; // Pull-mode relay (#16): a brokered self-host behind NAT/tailnet can't receive PUSHED forwards, so the Orb instead // ENQUEUES its events here and the engine drains them outbound. The batch caps how many rows a single pull returns @@ -276,11 +281,14 @@ export async function retryFailedRelays(env: Env, opts?: { fetchImpl?: typeof fe if (pruned.meta.changes > 0) { console.error(JSON.stringify({ level: "error", event: "orb_relay_events_dropped", message: `${pruned.meta.changes} relay event(s) dropped after ${RELAY_RETRY_MAX_ATTEMPTS} retries or 1h TTL`, count: pruned.meta.changes })); } + // Skip rows still inside their per-failure backoff window (#1950): a row whose last attempt was under + // RELAY_RETRY_BACKOFF_MINUTES ago waits for a later tick, so a down container is not re-POSTed every ~2 min. + // The bound modifier keeps this portable (the pg-dialect rewrites datetime('now', ?) → now() + (?)::interval). const { results } = await env.DB .prepare( - "SELECT delivery_id, event_name, installation_id, raw_body FROM orb_relay_failures WHERE expires_at >= datetime('now') AND attempts < ? ORDER BY created_at, delivery_id LIMIT ?", + "SELECT delivery_id, event_name, installation_id, raw_body FROM orb_relay_failures WHERE expires_at >= datetime('now') AND attempts < ? AND (last_attempt_at IS NULL OR last_attempt_at <= datetime('now', ?)) ORDER BY created_at, delivery_id LIMIT ?", ) - .bind(RELAY_RETRY_MAX_ATTEMPTS, RELAY_RETRY_BATCH_SIZE) + .bind(RELAY_RETRY_MAX_ATTEMPTS, `-${RELAY_RETRY_BACKOFF_MINUTES} minutes`, RELAY_RETRY_BATCH_SIZE) .all<{ delivery_id: string; event_name: string; installation_id: number; raw_body: string }>(); if (!results.length) return; diff --git a/test/integration/orb-relay.test.ts b/test/integration/orb-relay.test.ts index 3093d52cfc..be793182fc 100644 --- a/test/integration/orb-relay.test.ts +++ b/test/integration/orb-relay.test.ts @@ -335,6 +335,41 @@ describe("retryFailedRelays", () => { expect(row?.attempts).toBe(1); }); + it("SKIPS a row still inside its per-failure backoff window (#1950)", async () => { + const e = brokeredEnv(); + const secret = await enroll(e, 9600); + await registerOrbRelay(e, secret, "https://c.example/v1/orb/relay"); + // A row that last failed 1 minute ago (attempts=1) is inside the 5-minute backoff window. + await db(e) + .prepare("INSERT INTO orb_relay_failures (delivery_id, event_name, installation_id, raw_body, attempts, last_attempt_at) VALUES (?, ?, ?, ?, ?, datetime('now', '-1 minutes'))") + .bind("backoff-recent", "pull_request", 9600, "{}", 1) + .run(); + let calls = 0; + const fetchFail = (() => { + calls += 1; + return Promise.resolve(new Response("bad", { status: 503 })); + }) as typeof fetch; + await retryFailedRelays(e, { fetchImpl: fetchFail }); + expect(calls).toBe(0); // backed off → not re-POSTed this tick, so no fleet-wide storm on a down container + const row = await db(e).prepare("SELECT attempts FROM orb_relay_failures WHERE delivery_id='backoff-recent'").first<{ attempts: number }>(); + expect(row?.attempts).toBe(1); // untouched + }); + + it("RETRIES a row once its per-failure backoff window has elapsed (#1950)", async () => { + const e = brokeredEnv(); + const secret = await enroll(e, 9601); + await registerOrbRelay(e, secret, "https://c.example/v1/orb/relay"); + // A row that last failed 10 minutes ago (attempts=1) is past the 5-minute backoff → eligible again. + await db(e) + .prepare("INSERT INTO orb_relay_failures (delivery_id, event_name, installation_id, raw_body, attempts, last_attempt_at) VALUES (?, ?, ?, ?, ?, datetime('now', '-10 minutes'))") + .bind("backoff-elapsed", "pull_request", 9601, "{}", 1) + .run(); + const fetchFail = (() => Promise.resolve(new Response("bad", { status: 503 }))) as typeof fetch; + await retryFailedRelays(e, { fetchImpl: fetchFail }); + const row = await db(e).prepare("SELECT attempts FROM orb_relay_failures WHERE delivery_id='backoff-elapsed'").first<{ attempts: number }>(); + expect(row?.attempts).toBe(2); // eligible → retried, attempts incremented + }); + it("DELETES a row when forwardOrbEvent skips it (event no longer forwardable)", async () => { const e = brokeredEnv(); // Store a failure for an event that was later removed from RELAY_FORWARD_EVENTS (e.g. check_run).