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");