fix(org-drain): classify /update 409 lock contention as drain-superseded, not drain-failed (#1823)

fix(org-drain): classify /update 409 as drain-superseded, not drain-failed

The OrgActor drain races the source's own SourceActor scrape. When a stale
scrape source's 4h SourceActor poll lands within the scrape window of a drain
dispatch, the SourceActor grabs the per-source scrape lock (#1814/#1815) a beat
before the OrgActor dispatches its /update — so /update returns 409 "Source has
an active MA session". The lock guard is working correctly (it prevents a
double-scrape), and the source drains anyway via the lock holder's scrape.

But the OrgActor logged every non-ok /update as `drain-failed` (warn), so this
expected, benign contention polluted the drain-error signal. Treat a 409 (the
only status the /update source-dedup lock returns) as `drain-superseded` at info
level; every other non-ok status (spend cap 429, kill switch 503, mint failure)
stays `drain-failed`.

Author: zachdunn

workers/api/src/org-actor.ts

@@ -12,8 +12,10 @@
  * No in-app budget: the discovery `/update` endpoint already enforces the per-org
  * ($2/day) + global ($15/day) dollar spend cap (checkSpendCap, #1055) and the
  * per-source scrape lock (#1815) before minting a session, so this actor just
- * dispatches. A rejected /update (cap hit / locked) is logged and dropped; the
- * source stays flagged and re-drains on a later SourceActor notify.
+ * dispatches. A rejected /update is logged and dropped; the source stays flagged
+ * and re-drains on a later SourceActor notify. A 409 (per-source scrape lock held
+ * by the source's own SourceActor scrape) is the expected benign race — logged as
+ * `drain-superseded`, not `drain-failed`, since the lock holder drains the source.
  */
 
 import { DurableObject } from "cloudflare:workers";
@@ -126,9 +128,17 @@ export class OrgActor extends DurableObject<OrgActorEnv> {
       });
       if (!res.ok) {
         const body = await res.text().catch(() => "");
-        logEvent("warn", {
+        // A 409 from /update is the per-source dedup scrape lock (#1814) firing:
+        // the source's own SourceActor scrape grabbed the lock a beat before this
+        // drain dispatched. That's benign — the lock holder drains the source, so
+        // our /update was redundant, not failed. Classify it as `drain-superseded`
+        // (info) rather than `drain-failed` so expected lock contention doesn't
+        // pollute the drain-error signal. Every other non-ok status (spend cap
+        // 429, kill switch 503, mint failure) is a genuine drop worth a warn.
+        const superseded = res.status === 409;
+        logEvent(superseded ? "info" : "warn", {
           component: "org-actor",
-          event: "drain-failed",
+          event: superseded ? "drain-superseded" : "drain-failed",
           orgId,
           status: res.status,
           detail: body.slice(0, 200),

workers/api/test/org-actor.test.ts

@@ -119,7 +119,51 @@ describe("OrgActor", () => {
     expect(h.dispatched.length).toBe(0);
   });
 
-  it("does not throw when /update returns an error (spend cap / lock)", async () => {
+  // Capture the JSON logEvent lines a run emits, by severity (info→log, warn→warn).
+  async function captureLogs(fn: () => Promise<void>) {
+    const info: any[] = [];
+    const warn: any[] = [];
+    const origLog = console.log;
+    const origWarn = console.warn;
+    const parse = (sink: any[]) => (line?: unknown) => {
+      try {
+        sink.push(JSON.parse(String(line)));
+      } catch {
+        /* non-JSON line, ignore */
+      }
+    };
+    console.log = parse(info) as typeof console.log;
+    console.warn = parse(warn) as typeof console.warn;
+    try {
+      await fn();
+    } finally {
+      console.log = origLog;
+      console.warn = origWarn;
+    }
+    return { info, warn };
+  }
+
+  it("classifies a 409 (scrape lock held) as drain-superseded, not drain-failed", async () => {
+    const db = mkDb();
+    seedFlaggedScrape(db, "src_a");
+    const h = mkActor(
+      db,
+      () =>
+        new Response("Source src_a has an active MA session (ma-abc)", {
+          status: 409,
+          headers: { "Retry-After": "900" },
+        }),
+    );
+    await h.actor.ensureDrainScheduled("org_x");
+    const { info, warn } = await captureLogs(() => h.actor.alarm()); // must not throw
+    expect(h.dispatched.length).toBe(1);
+    expect(h.alarmAt()).toBeNull();
+    // Benign race: emitted at info level as drain-superseded, never as an error.
+    expect(info.some((l) => l.event === "drain-superseded" && l.status === 409)).toBe(true);
+    expect(warn.some((l) => l.event === "drain-failed")).toBe(false);
+  });
+
+  it("classifies a non-409 error (spend cap 429) as drain-failed", async () => {
     const db = mkDb();
     seedFlaggedScrape(db, "src_a");
     const h = mkActor(
@@ -128,9 +172,11 @@ describe("OrgActor", () => {
         new Response(JSON.stringify({ error: "Daily global spend cap reached" }), { status: 429 }),
     );
     await h.actor.ensureDrainScheduled("org_x");
-    await h.actor.alarm(); // must not throw
+    const { info, warn } = await captureLogs(() => h.actor.alarm()); // must not throw
     expect(h.dispatched.length).toBe(1);
     expect(h.alarmAt()).toBeNull();
+    expect(warn.some((l) => l.event === "drain-failed" && l.status === 429)).toBe(true);
+    expect(info.some((l) => l.event === "drain-superseded")).toBe(false);
   });
 
   it("does NOT dispatch when the kill switch is off at alarm time", async () => {