fix(relay): stop held webhook deliveries when the inbox is cleared

infra/relay/src/hooks/HookInboxStore.test.ts

@@ -2,8 +2,10 @@ import { describe, expect, it } from "@effect/vitest";
 import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
 import * as Clock from "effect/Clock";
 import * as DateTime from "effect/DateTime";
+import * as Deferred from "effect/Deferred";
 import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
+import * as Fiber from "effect/Fiber";
 import * as SqlClient from "effect/sql/SqlClient";
 import * as TestClock from "effect/testing/TestClock";
 
@@ -43,6 +45,41 @@ const withInbox = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
   );
 
 describe("HookInboxStore", () => {
+  it.effect.each(["delivered", "busy", "unreachable"] as const)(
+    "stops a prefetched batch cleared during a %s send",
+    (outcome) =>
+      withInbox(
+        Effect.gen(function* () {
+          for (const id of ["first", "second", "third"]) {
+            yield* HookInboxStore.hold(yield* hook(id), BASE_URL);
+          }
+          const started = yield* Deferred.make<void>();
+          const released = yield* Deferred.make<void>();
+          const sent: Array<string> = [];
+          const run = yield* HookInboxStore.deliverDue((_baseUrl, held) =>
+            Effect.gen(function* () {
+              sent.push(held.id);
+              if (held.id === "first") {
+                yield* Deferred.succeed(started, undefined);
+                yield* Deferred.await(released);
+              }
+              return outcome;
+            }),
+          ).pipe(Effect.forkChild);
+          yield* Deferred.await(started);
+          yield* HookInboxStore.clear;
+          yield* Deferred.succeed(released, undefined);
+          const nextAt = yield* Fiber.join(run);
+          expect(sent).toEqual(["first"]);
+          // The alarm must not be rescheduled, even if the in-flight send failed.
+          expect(nextAt).toBeNull();
+          const sql = yield* SqlClient.SqlClient;
+          expect(yield* sql`SELECT id FROM held_hooks`).toEqual([]);
+          expect(yield* sql`SELECT id FROM held_hooks_target`).toEqual([]);
+        }),
+      ),
+  );
+
   it.effect("delivers held requests oldest first, byte for byte", () =>
     withInbox(
       Effect.gen(function* () {

infra/relay/src/hooks/HookInboxStore.ts

@@ -321,13 +321,22 @@ export const deliverDue = Effect.fn("HookInboxStore.deliverDue")(function* <R>(
         "relay.inbox.held_bytes": backlog.bytes,
       });
     });
+  const stillHeld = (seq: number) =>
+    sql<{ readonly seq: number }>`SELECT seq FROM held_hooks WHERE seq = ${seq}`.pipe(
+      Effect.map((rows) => rows.length > 0),
+    );
   // Pages run out only once every non-busy request has been tried, so a
   // backlog behind busy hooks never hides another hook's requests.
   pages: while (batch.length > 0) {
     for (const row of batch) {
       if (sent === DELIVERIES_PER_RUN) break pages;
       // Later requests to a busy hook wait their turn, so its order is kept.
       if (busyHooks.has(row.hook_key)) continue;
+      // Clear can discard this prefetched page while an upstream send is in flight.
+      if (!(yield* stillHeld(row.seq))) {
+        yield* annotateRun("cleared");
+        return null;
+      }
       const headers = decodeHeaders(row.headers);
       if (Option.isNone(headers)) {
         // Unreadable: it can never be delivered, and must not block the rest.
@@ -347,6 +356,11 @@ export const deliverDue = Effect.fn("HookInboxStore.deliverDue")(function* <R>(
         headers: headers.value,
         body: row.body,
       });
+      // A cleared request must neither advance this run nor schedule a retry.
+      if (!(yield* stillHeld(row.seq))) {
+        yield* annotateRun("cleared");
+        return null;
+      }
       if (outcome === "unreachable") {
         failures = yield* countFailure;
         yield* annotateRun("unreachable");