fix(orb): back off per-failure relay retries to avoid a fleet-wide storm (#1950)

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;
 

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).