diff --git a/infra/relay/src/hooks/HookInboxStore.test.ts b/infra/relay/src/hooks/HookInboxStore.test.ts
index 121ce84a686c..8a62fd539adc 100644
--- a/infra/relay/src/hooks/HookInboxStore.test.ts
+++ b/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 = (effect: Effect.Effect) =>
);
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();
+ const released = yield* Deferred.make();
+ const sent: Array = [];
+ 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* () {
diff --git a/infra/relay/src/hooks/HookInboxStore.ts b/infra/relay/src/hooks/HookInboxStore.ts
index 64049f6b86ab..ae57b0fccfef 100644
--- a/infra/relay/src/hooks/HookInboxStore.ts
+++ b/infra/relay/src/hooks/HookInboxStore.ts
@@ -321,6 +321,10 @@ export const deliverDue = Effect.fn("HookInboxStore.deliverDue")(function* (
"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) {
@@ -328,6 +332,11 @@ export const deliverDue = Effect.fn("HookInboxStore.deliverDue")(function* (
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* (
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");