From bf204b6ede0d6654464c1c7bb952f34bef8c8faf Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:20:12 -0700 Subject: [PATCH 1/4] Add inbox snooze-until table and claim-and-reopen store (CL-7208) New, package-owned Postgres table for CL-7208: `@corbits/mailbox`'s enrichment (packages/inbox's only dependency for a message's status) supports only priority/classification/status -- there is no column to hold a snooze's `until` timestamp, so `POST /:id/snooze` has never had anywhere to durably record when to reopen a snoozed item. - schema.ts: `inbox.snooze` (tenantId, principalId, messageId, until), own schema, following @workbench/onboarding's pattern for a package-owned product table. - migrations.ts: `applyInboxMigrations`, following the same ledger-table pattern as onboarding/routines/notify -- registered in scripts/db-setup.ts's INSTALLED_PACKAGE_MIGRATIONS as `@corbits/inbox`. - snooze-store.ts: `setSnoozeUntil`/`clearSnoozeUntil` (plain upsert/delete), `findDueSnoozes` (a bare, non-claiming SELECT), and `claimAndReopenSnooze` -- the one operation with real teeth. It deletes the due row and flips the message back to `open` inside a single transaction: two hub replicas racing the same row only ever have one `DELETE ... RETURNING` succeed, and a throw between the delete and the status flip rolls both back, leaving the snooze row exactly as it was for the next sweep tick to retry. This is the same mail-then-claim ordering lesson CL-7209 applied to the credential-expiry sweep (act before durably marking "handled"), applied here to the claim itself so a mid-claim failure can never orphan an item snoozed forever. Verified against a real Postgres (not just typechecked): a scratch-db integration test proves setSnoozeUntil/findDueSnoozes/ claimAndReopenSnooze/clearSnoozeUntil all behave correctly, including the race guard (a second claim on an already-claimed row returns false) and the self-healing no-op (claiming a row whose message left `snoozed` some other way cleans up the stale row instead of reopening it). Moves `postgres` from a devDependency to a runtime one (migrations.ts uses it directly, like onboarding/routines/notify) and adds `drizzle-orm`. --- bun.lock | 3 +- packages/inbox/package.json | 8 +- packages/inbox/src/index.ts | 7 + packages/inbox/src/migrations.ts | 106 +++++++++- packages/inbox/src/schema.ts | 28 +++ packages/inbox/src/snooze-store.ts | 147 ++++++++++++++ packages/inbox/test/snooze-store.test.ts | 236 +++++++++++++++++++++++ scripts/db-setup.ts | 7 +- 8 files changed, 534 insertions(+), 8 deletions(-) create mode 100644 packages/inbox/src/schema.ts create mode 100644 packages/inbox/src/snooze-store.ts create mode 100644 packages/inbox/test/snooze-store.test.ts diff --git a/bun.lock b/bun.lock index cc2eb879b..e793f0386 100644 --- a/bun.lock +++ b/bun.lock @@ -849,13 +849,14 @@ "@intx/hub-api": "workspace:*", "@intx/log": "0.3.0", "arktype": "catalog:", + "drizzle-orm": "catalog:", "hono": "catalog:", + "postgres": "catalog:", }, "devDependencies": { "@intx/db": "workspace:*", "@intx/hub-common": "0.3.0", "@types/bun": "catalog:", - "postgres": "catalog:", "typescript": "catalog:", }, }, diff --git a/packages/inbox/package.json b/packages/inbox/package.json index eaeb0caea..56301860a 100644 --- a/packages/inbox/package.json +++ b/packages/inbox/package.json @@ -7,7 +7,8 @@ "type": "module", "exports": { ".": "./src/index.ts", - "./client": "./src/client.ts" + "./client": "./src/client.ts", + "./migrations": "./src/migrations.ts" }, "scripts": { "typecheck": "tsc --noEmit", @@ -20,13 +21,14 @@ "@intx/hub-api": "workspace:*", "@intx/log": "0.3.0", "arktype": "catalog:", - "hono": "catalog:" + "drizzle-orm": "catalog:", + "hono": "catalog:", + "postgres": "catalog:" }, "devDependencies": { "@intx/db": "workspace:*", "@intx/hub-common": "0.3.0", "@types/bun": "catalog:", - "postgres": "catalog:", "typescript": "catalog:" } } diff --git a/packages/inbox/src/index.ts b/packages/inbox/src/index.ts index 984f47c21..75816bc8e 100644 --- a/packages/inbox/src/index.ts +++ b/packages/inbox/src/index.ts @@ -28,7 +28,14 @@ export { type CreateWorkbenchMailboxDeliveryOpts, } from "./delivery"; export { + applyInboxMigrations, applyMailboxMigrations, + type ApplyInboxMigrationsReport, type ApplyMailboxMigrationsReport, } from "./migrations"; export { createInboxRoutes, type CreateInboxRoutesDeps } from "./routes"; +export { + claimAndReopenSnooze, + findDueSnoozes, + type DueSnooze, +} from "./snooze-store"; diff --git a/packages/inbox/src/migrations.ts b/packages/inbox/src/migrations.ts index c1d8497e6..ce26d334d 100644 --- a/packages/inbox/src/migrations.ts +++ b/packages/inbox/src/migrations.ts @@ -1,8 +1,19 @@ -// URL-shaped wrapper around `@corbits/mailbox`'s `runMailboxMigrations` so -// scripts/db-setup.ts can apply it the same way it applies every other -// installed package's migrations. +// Two independent migration stories for `@corbits/inbox`: +// +// - `applyMailboxMigrations`: a URL-shaped wrapper around +// `@corbits/mailbox`'s own `runMailboxMigrations`, so scripts/db-setup.ts +// can apply it the same way it applies every other installed package's +// migrations. That package keeps its own ledger inside the `mailbox` +// schema. +// - `applyInboxMigrations`: this package's own product table (CL-7208's +// `inbox.snooze` — see `./schema.ts` for why `@corbits/mailbox`'s +// enrichment has no column for a snooze's `until`). Bookkeeping uses its +// own ledger table, following the same pattern as +// `@workbench/onboarding`'s `migrations.ts`, so this package's migration +// history stays extractable on its own. import { createMailboxDb, runMailboxMigrations } from "@corbits/mailbox"; +import postgres from "postgres"; export interface ApplyMailboxMigrationsReport { applied: string[]; @@ -28,3 +39,92 @@ export async function applyMailboxMigrations( await close(); } } + +export interface InboxMigration { + name: string; + sql: string; +} + +const SCHEMA = "inbox"; +const LEDGER_TABLE = "inbox_migrations"; + +export const inboxMigrations: readonly InboxMigration[] = [ + { + name: "0001_snooze", + sql: ` + CREATE TABLE IF NOT EXISTS "inbox"."snooze" ( + "tenant_id" text NOT NULL, + "principal_id" text NOT NULL, + "message_id" text NOT NULL, + "until" timestamptz NOT NULL, + PRIMARY KEY ("tenant_id", "principal_id", "message_id") + ); + CREATE INDEX IF NOT EXISTS "inbox_snooze_until_idx" ON "inbox"."snooze" ("until"); + `, + }, +]; + +function quoteIdentifier(name: string): string { + return `"${name.replace(/"/g, '""')}"`; +} + +function quoteQualified(schema: string, name: string): string { + return `${quoteIdentifier(schema)}.${quoteIdentifier(name)}`; +} + +export interface ApplyInboxMigrationsReport { + applied: string[]; + alreadyApplied: string[]; +} + +/** + * Apply `inboxMigrations` against `databaseUrl`, idempotently: a migration + * already recorded in the ledger is skipped, never re-run. Failures are + * loud — the migration name and the underlying error are both surfaced. + */ +export async function applyInboxMigrations( + databaseUrl: string, +): Promise { + const sql = postgres(databaseUrl, { max: 1, onnotice: () => undefined }); + try { + await sql.unsafe(`CREATE SCHEMA IF NOT EXISTS ${quoteIdentifier(SCHEMA)}`); + await sql.unsafe( + `CREATE TABLE IF NOT EXISTS ${quoteQualified(SCHEMA, LEDGER_TABLE)} (` + + `name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`, + ); + + const applied: string[] = []; + const alreadyApplied: string[] = []; + + for (const migration of inboxMigrations) { + const existing = await sql.unsafe( + `SELECT 1 FROM ${quoteQualified(SCHEMA, LEDGER_TABLE)} WHERE name = $1`, + [migration.name], + ); + if (existing.length > 0) { + alreadyApplied.push(migration.name); + continue; + } + try { + await sql.begin(async (tx) => { + await tx.unsafe(migration.sql); + await tx.unsafe( + `INSERT INTO ${quoteQualified(SCHEMA, LEDGER_TABLE)} (name) VALUES ($1)`, + [migration.name], + ); + }); + applied.push(migration.name); + } catch (error) { + throw new Error( + `@corbits/inbox migration ${JSON.stringify(migration.name)} failed: ` + + `${error instanceof Error ? error.message : String(error)}`, + { cause: error }, + ); + } + } + + return { applied, alreadyApplied }; + } finally { + await sql.end({ timeout: 5 }); + } +} diff --git a/packages/inbox/src/schema.ts b/packages/inbox/src/schema.ts new file mode 100644 index 000000000..4fddf2c5b --- /dev/null +++ b/packages/inbox/src/schema.ts @@ -0,0 +1,28 @@ +// The one product table `@corbits/inbox` owns: `@corbits/mailbox`'s +// enrichment only supports `priority`/`classification`/`status` (see +// `enrichMailboxMessage`'s `MailboxEnrichment`), so there is no column +// there to hold a snooze's `until` timestamp. This table is that column, +// keyed to the same (tenantId, principalId, messageId) scope every +// mailbox mutation uses. Lives in its own `inbox` Postgres schema, never +// `public` — see docs/package-migrations.md and the same pattern in +// `@workbench/onboarding`'s `schema.ts`. +import { pgSchema, primaryKey, text, timestamp } from "drizzle-orm/pg-core"; + +export const inboxSchema = pgSchema("inbox"); + +export const inboxSnooze = inboxSchema.table( + "snooze", + { + tenantId: text("tenant_id").notNull(), + principalId: text("principal_id").notNull(), + messageId: text("message_id").notNull(), + until: timestamp("until", { withTimezone: true }).notNull(), + }, + (table) => [ + primaryKey({ + columns: [table.tenantId, table.principalId, table.messageId], + }), + ], +); + +export type InboxSnoozeRow = typeof inboxSnooze.$inferSelect; diff --git a/packages/inbox/src/snooze-store.ts b/packages/inbox/src/snooze-store.ts new file mode 100644 index 000000000..2f26aa931 --- /dev/null +++ b/packages/inbox/src/snooze-store.ts @@ -0,0 +1,147 @@ +// Persistence for CL-7208's snooze-until: `@corbits/mailbox`'s enrichment +// has no column for it (see `./schema.ts`), so this module owns the +// package's one product table and the two operations built on it — set at +// snooze time, and the atomic claim-and-reopen a periodic sweep drives +// once `until` has passed. +import { and, eq, lte } from "drizzle-orm"; +import { + enrichMailboxMessage, + getMailboxMessage, + type MailboxDb, +} from "@corbits/mailbox"; + +import { inboxSnooze } from "./schema"; + +export interface SnoozeScope { + tenantId: string; + principalId: string; + id: string; +} + +/** + * Record when a snoozed message should reopen. Written *before* the + * message's own status flips to `snoozed` (see `routes.ts`'s `/:id/snooze` + * handler): if the status flip then fails, the message is left in + * whatever status it already had with an orphaned snooze row pointing at + * it — harmless, since `claimAndReopenSnooze` below no-ops and cleans up a + * row whose message isn't actually `snoozed`. The other ordering (status + * first, `until` second) would reproduce the exact bug this ticket fixes: + * a message stuck `snoozed` with nothing ever recording when to reopen it. + */ +export async function setSnoozeUntil( + db: MailboxDb, + scope: SnoozeScope, + until: Date, +): Promise { + await db + .insert(inboxSnooze) + .values({ + tenantId: scope.tenantId, + principalId: scope.principalId, + messageId: scope.id, + until, + }) + .onConflictDoUpdate({ + target: [ + inboxSnooze.tenantId, + inboxSnooze.principalId, + inboxSnooze.messageId, + ], + set: { until }, + }); +} + +export async function clearSnoozeUntil( + db: MailboxDb, + scope: SnoozeScope, +): Promise { + await db + .delete(inboxSnooze) + .where( + and( + eq(inboxSnooze.tenantId, scope.tenantId), + eq(inboxSnooze.principalId, scope.principalId), + eq(inboxSnooze.messageId, scope.id), + ), + ); +} + +export interface DueSnooze { + tenantId: string; + principalId: string; + messageId: string; +} + +/** Every snooze row due at or before `now`. A bare `SELECT` — claims + * nothing, so two hub replicas can see the same row; `claimAndReopenSnooze` + * is what makes acting on it race-safe. */ +export async function findDueSnoozes( + db: MailboxDb, + now: Date, +): Promise { + const rows = await db + .select({ + tenantId: inboxSnooze.tenantId, + principalId: inboxSnooze.principalId, + messageId: inboxSnooze.messageId, + }) + .from(inboxSnooze) + .where(lte(inboxSnooze.until, now)); + return rows; +} + +/** + * Atomically claim one due snooze row and reopen its message if it's still + * `snoozed` — all inside one transaction, so the delete-claim and the + * status flip either both happen or neither does. Two replicas racing the + * same row: the second's `DELETE ... RETURNING` simply returns no rows + * (the first already deleted it), so only one replica ever flips the + * status or is told to publish a reopened event. + * + * Returns whether this call actually reopened the message — `false` means + * it lost the race, the row was no longer due, or the message had already + * left `snoozed` (e.g. the user manually reopened it) and the row was just + * cleaned up. A throw rolls the whole transaction back, so a failure here + * leaves the snooze row exactly as it was — retried on the sweep's next + * tick rather than dropped (the same mail-then-claim ordering lesson + * CL-7209 applied to the credential-expiry sweep, applied here to the + * claim itself). + */ +export async function claimAndReopenSnooze( + db: MailboxDb, + row: DueSnooze, + now: Date, +): Promise { + return db.transaction(async (tx) => { + const claimed = await tx + .delete(inboxSnooze) + .where( + and( + eq(inboxSnooze.tenantId, row.tenantId), + eq(inboxSnooze.principalId, row.principalId), + eq(inboxSnooze.messageId, row.messageId), + lte(inboxSnooze.until, now), + ), + ) + .returning({ messageId: inboxSnooze.messageId }); + if (claimed.length === 0) return false; + + const message = await getMailboxMessage(tx, { + tenantId: row.tenantId, + principalId: row.principalId, + id: row.messageId, + }); + if (message === null || message.status !== "snoozed") return false; + + await enrichMailboxMessage( + tx, + { + tenantId: row.tenantId, + principalId: row.principalId, + id: row.messageId, + }, + { status: "open" }, + ); + return true; + }); +} diff --git a/packages/inbox/test/snooze-store.test.ts b/packages/inbox/test/snooze-store.test.ts new file mode 100644 index 000000000..913bde091 --- /dev/null +++ b/packages/inbox/test/snooze-store.test.ts @@ -0,0 +1,236 @@ +// DB-gated integration test for CL-7208's snooze-until persistence and +// claim-and-reopen: proves `setSnoozeUntil`/`findDueSnoozes` write and read +// a real row in the package's own `inbox.snooze` table (applied through +// `scripts/db-setup.ts`, the same path `apps/hub` boots with — see +// `delivery.test.ts` for the pattern this follows), and that +// `claimAndReopenSnooze` actually flips a real mailbox message back to +// `open` and removes the snooze row once `until` has passed. Runs against +// its own scratch database, never the developer's or the walking-skeleton +// suite's. +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import postgres from "postgres"; + +import { createDB, schema } from "@intx/db"; +import { generateId } from "@intx/hub-common"; +import { + createMailboxDb, + enrichMailboxMessage, + getMailboxMessage, + writeMailboxMessage, +} from "@corbits/mailbox"; + +import { setupDatabase } from "../../../scripts/db-setup"; +import { e2eDatabaseUrl } from "../../../scripts/e2e/harness"; +import { + claimAndReopenSnooze, + clearSnoozeUntil, + findDueSnoozes, + setSnoozeUntil, +} from "../src/snooze-store"; + +function scratchUrlFor(e2eUrl: string): string { + const url = new URL(e2eUrl); + const database = url.pathname.replace(/^\//, ""); + url.pathname = `/${database}_inbox_snooze_test`; + return url.toString(); +} + +function dbConfigFromUrl(databaseUrl: string) { + const url = new URL(databaseUrl); + return { + host: url.hostname, + port: url.port === "" ? 5432 : Number(url.port), + user: decodeURIComponent(url.username), + password: decodeURIComponent(url.password), + database: url.pathname.replace(/^\//, ""), + }; +} + +const databaseUrl = e2eDatabaseUrl(); +const describeIfDb = databaseUrl === undefined ? describe.skip : describe; + +describeIfDb("snooze-store against a real inbox.snooze table", () => { + const scratchUrl = scratchUrlFor( + databaseUrl ?? "postgres://localhost:5432/unused", + ); + const scratchDatabase = new URL(scratchUrl).pathname.replace(/^\//, ""); + + const tenantId = generateId("tenant"); + const principalId = generateId("principal"); + + beforeAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); + await maintenance.unsafe(`CREATE DATABASE "${scratchDatabase}"`); + } finally { + await maintenance.end(); + } + + // Platform migrations plus every installed package's, including + // @corbits/inbox's own `inbox.snooze` table (CL-7208). + await setupDatabase(scratchUrl); + + const { db, close } = createDB(dbConfigFromUrl(scratchUrl)); + try { + await db.insert(schema.tenant).values({ + id: tenantId, + name: "Snooze Test Bench", + slug: `snooze-${tenantId}`, + domain: `snooze-${tenantId}.localhost`, + }); + await db.insert(schema.principal).values({ + id: principalId, + tenantId, + kind: "agent", + refId: "not-a-real-agent-instance", + status: "active", + }); + } finally { + await close(); + } + }, 30000); + + afterAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchDatabase}"`); + } finally { + await maintenance.end(); + } + }); + + test("setSnoozeUntil persists a row findDueSnoozes reads back once due", async () => { + const mailboxDb = createMailboxDb(scratchUrl); + try { + const written = await writeMailboxMessage(mailboxDb.db, { + tenantId, + principalId, + address: `${principalId}@inbox.test`, + fromAddress: "routine:test", + subject: "Snooze me", + body: "test body", + status: "snoozed", + }); + expect(written).not.toBeNull(); + const messageId = written?.id; + if (messageId === undefined) throw new Error("message not written"); + const scope = { tenantId, principalId, id: messageId }; + + const future = new Date(Date.now() + 60_000); + await setSnoozeUntil(mailboxDb.db, scope, future); + + // Not due yet. + const notYetDue = await findDueSnoozes(mailboxDb.db, new Date()); + expect(notYetDue.find((r) => r.messageId === messageId)).toBeUndefined(); + + // Due once `now` passes `until`. + const afterUntil = new Date(future.getTime() + 1000); + const due = await findDueSnoozes(mailboxDb.db, afterUntil); + expect(due.find((r) => r.messageId === messageId)).toEqual({ + tenantId, + principalId, + messageId, + }); + + const reopened = await claimAndReopenSnooze( + mailboxDb.db, + { tenantId, principalId, messageId }, + afterUntil, + ); + expect(reopened).toBe(true); + + const message = await getMailboxMessage(mailboxDb.db, scope); + expect(message?.status).toBe("open"); + + // The claim deleted the snooze row: a second claim attempt finds + // nothing left to reopen. + const secondClaim = await claimAndReopenSnooze( + mailboxDb.db, + { tenantId, principalId, messageId }, + afterUntil, + ); + expect(secondClaim).toBe(false); + } finally { + await mailboxDb.close(); + } + }); + + test("claimAndReopenSnooze no-ops and cleans up a row whose message is no longer snoozed", async () => { + const mailboxDb = createMailboxDb(scratchUrl); + try { + const written = await writeMailboxMessage(mailboxDb.db, { + tenantId, + principalId, + address: `${principalId}@inbox.test`, + fromAddress: "routine:test", + subject: "Snooze then manually reopen", + body: "test body", + status: "snoozed", + }); + const messageId = written?.id; + if (messageId === undefined) throw new Error("message not written"); + const scope = { tenantId, principalId, id: messageId }; + + const past = new Date(Date.now() - 1000); + await setSnoozeUntil(mailboxDb.db, scope, past); + + // Simulate the user manually reopening it before the sweep ran — + // exactly the race `claimAndReopenSnooze`'s status check guards. + await enrichMailboxMessage(mailboxDb.db, scope, { status: "open" }); + + const due = await findDueSnoozes(mailboxDb.db, new Date()); + expect(due.some((r) => r.messageId === messageId)).toBe(true); + + const reopened = await claimAndReopenSnooze( + mailboxDb.db, + { tenantId, principalId, messageId }, + new Date(), + ); + // The row was due, so the claim wins, but the message was already + // `open` — nothing to reopen, so this reports false while still + // cleaning up the now-stale snooze row. + expect(reopened).toBe(false); + + const stillDue = await findDueSnoozes(mailboxDb.db, new Date()); + expect(stillDue.some((r) => r.messageId === messageId)).toBe(false); + } finally { + await mailboxDb.close(); + } + }); + + test("clearSnoozeUntil removes the row without touching the message", async () => { + const mailboxDb = createMailboxDb(scratchUrl); + try { + const written = await writeMailboxMessage(mailboxDb.db, { + tenantId, + principalId, + address: `${principalId}@inbox.test`, + fromAddress: "routine:test", + subject: "Snooze then cancel", + body: "test body", + }); + const messageId = written?.id; + if (messageId === undefined) throw new Error("message not written"); + const scope = { tenantId, principalId, id: messageId }; + + await setSnoozeUntil(mailboxDb.db, scope, new Date(Date.now() - 1000)); + await clearSnoozeUntil(mailboxDb.db, scope); + + const due = await findDueSnoozes(mailboxDb.db, new Date()); + expect(due.some((r) => r.messageId === messageId)).toBe(false); + } finally { + await mailboxDb.close(); + } + }); +}); diff --git a/scripts/db-setup.ts b/scripts/db-setup.ts index ad28cf56a..f8189d2d2 100644 --- a/scripts/db-setup.ts +++ b/scripts/db-setup.ts @@ -39,7 +39,10 @@ import { import { applyWebhookTriggersMigrations } from "../packages/webhook-triggers/src/migrations"; import { applyNotifyMigrations } from "../packages/notify/src/migrations"; import { applyRoutineMigrations } from "../packages/routines/src/migrations"; -import { applyMailboxMigrations } from "../packages/inbox/src/migrations"; +import { + applyInboxMigrations, + applyMailboxMigrations, +} from "../packages/inbox/src/migrations"; import { applyInsightsMigrations } from "../packages/insights/src/migrations"; import { applyPreferencesMigrations } from "../packages/preferences/src/migrations"; import { applyBenchMigrations } from "../packages/bench/src/migrations"; @@ -76,6 +79,8 @@ const INSTALLED_PACKAGE_MIGRATIONS: readonly { { name: "@corbits/routines", apply: applyRoutineMigrations }, { name: "@corbits/notify", apply: applyNotifyMigrations }, { name: "@corbits/mailbox", apply: applyMailboxMigrations }, + // CL-7208's snooze-until table, own schema — see packages/inbox/src/schema.ts. + { name: "@corbits/inbox", apply: applyInboxMigrations }, { name: "@corbits/insights", apply: applyInsightsMigrations }, { name: "@corbits/preferences", apply: applyPreferencesMigrations }, { name: "@corbits/bench", apply: applyBenchMigrations }, From f6a1852465ff587aeaed9da69ef045621f97fd77 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:20:30 -0700 Subject: [PATCH 2/4] Persist snooze until and reopen it with a periodic sweep (CL-7208) `POST /:id/snooze` accepted an `until` timestamp, validated its type, and echoed it back -- but only ever persisted `{status: "snoozed"}`. Nothing recorded `until` or ever reopened the item: a snoozed approval or credential-reconnect sat under the snoozed filter forever, invisible, with no reminder ever firing. - routes.ts: `until` is now required and parsed at the boundary with arktype (never `as`-cast), rejected with 400 if it doesn't parse as a timestamp or has already passed -- silently snoozing forever is exactly the bug being fixed, so a past `until` is refused rather than accepted. `until` is persisted via `setSnoozeUntil` *before* the status flips to `snoozed`; if the flip then fails, the snooze row is rolled back too rather than left orphaned. - apps/hub/src/inbox-unsnooze-sweep.ts: a small periodic sweep (`tickInboxUnsnoozeSweep`/`createInboxUnsnoozeSweep`) mirroring `credential-expiry-sweep.ts` and `routine-scheduler.ts` -- the only other periodic loops in this hub -- rather than inventing a new scheduling primitive. `@corbits/notify`'s `createNotifyDispatcher` is shaped around sink delivery (attempts, backoff, a `SinkRegistry`), not a status flip, and the routine scheduler's `RoutineStore` is routine-domain-specific; neither fits "reopen a mailbox row once its snooze elapses" without bending its semantics, so this repeats the small-setInterval-plus-own-due-scan shape both already use. Wired into apps/hub/src/index.ts next to credentialExpirySweep, on the same mailboxDb/mailboxBus every other mailbox consumer there shares, and stopped in shutdown alongside the others. Tests: route-level rejection of a missing/non-string/unparseable/past `until`; a DB-free unit test of the sweep tick against a fake store (reopens every due row and publishes once per actual reopen, does not publish when a claim reports no-op, and a per-row throw is swallowed -- logged, not propagated -- so one bad row can't abort the rest of the tick). --- apps/hub/src/inbox-unsnooze-sweep.ts | 132 +++++++++++++++++++++ apps/hub/src/index.ts | 13 ++ apps/hub/test/inbox-unsnooze-sweep.test.ts | 100 ++++++++++++++++ packages/inbox/src/routes.ts | 54 ++++++--- packages/inbox/test/routes.test.ts | 46 +++++++ 5 files changed, 328 insertions(+), 17 deletions(-) create mode 100644 apps/hub/src/inbox-unsnooze-sweep.ts create mode 100644 apps/hub/test/inbox-unsnooze-sweep.test.ts diff --git a/apps/hub/src/inbox-unsnooze-sweep.ts b/apps/hub/src/inbox-unsnooze-sweep.ts new file mode 100644 index 000000000..ae5f44796 --- /dev/null +++ b/apps/hub/src/inbox-unsnooze-sweep.ts @@ -0,0 +1,132 @@ +// A minimal periodic loop that reopens a snoozed inbox item once its +// `until` has passed (CL-7208) — mirroring `credential-expiry-sweep.ts` +// and `routine-scheduler.ts`, the only other periodic loops in this hub, +// rather than standing up a new generic scheduling primitive. Neither +// `@corbits/notify`'s `createNotifyDispatcher` (shaped around sink +// delivery — attempts, backoff, a `SinkRegistry` — not a status flip) nor +// the routine scheduler (`RoutineStore` is routine-domain-specific) fits +// "reopen a mailbox row when its snooze timer elapses" without bending +// their semantics, so this repeats the same small `setInterval` + own +// due-scan shape those two already use. +// +// `claimAndReopenSnooze` (in `@corbits/inbox`) does the actual claim: a +// single transaction that deletes the due snooze row and flips the +// message back to `open` together, so a throw here rolls back both and +// leaves the row for the next tick to retry rather than orphaning it — +// the mail-then-claim lesson CL-7209 applied to the credential-expiry +// sweep, applied here to the claim itself. +import { getLogger } from "@intx/log"; +import { reportError } from "@corbits/error-sink"; +import { + claimAndReopenSnooze, + findDueSnoozes, + type DueSnooze, +} from "@corbits/inbox"; +import type { MailboxDb, MailboxEventBus } from "@corbits/mailbox"; + +export type InboxUnsnoozeSweepStore = { + findDueSnoozes(now: Date): Promise; + /** Atomically claims one due snooze row and reopens its message if it's + * still snoozed. Returns whether it actually reopened something — see + * `claimAndReopenSnooze`'s own doc comment for the false cases. */ + claimAndReopen(row: DueSnooze, now: Date): Promise; +}; + +export function createDrizzleInboxUnsnoozeSweepStore( + db: MailboxDb, +): InboxUnsnoozeSweepStore { + return { + findDueSnoozes: (now) => findDueSnoozes(db, now), + claimAndReopen: (row, now) => claimAndReopenSnooze(db, row, now), + }; +} + +export type InboxUnsnoozeSweepDeps = { + store: InboxUnsnoozeSweepStore; + bus: Pick; + /** Injectable for deterministic tests; defaults to `Date.now`-backed wall time. */ + now?: () => Date; +}; + +const POLL_INTERVAL_MS = 60 * 1000; +const publishLog = getLogger(["hub", "inbox-unsnooze-sweep"]); + +function publishReopened( + bus: Pick, + row: DueSnooze, +): void { + // Best-effort, matching every other mailbox event publish in this repo + // (see packages/inbox/src/routes.ts's own `publish` helper): a bus + // failure here must never turn an already-committed reopen into a + // reported sweep failure. + try { + bus.publish( + { tenantId: row.tenantId, principalId: row.principalId }, + { type: "mailbox", id: row.messageId, op: "enrich" }, + ); + } catch (error) { + publishLog.error( + "mailbox reopen event publish failed for {id} on tenant {tenantId}, principal {principalId}: {error}", + { + id: row.messageId, + tenantId: row.tenantId, + principalId: row.principalId, + error: error instanceof Error ? error.message : String(error), + }, + ); + } +} + +/** + * One sweep: reopen every snooze due at `at`. Exported (rather than kept + * as a closure) so a test can drive a single, deterministic pass without + * waiting on `setInterval`. + */ +export async function tickInboxUnsnoozeSweep( + deps: Pick, + at: Date, +): Promise { + const due = await deps.store.findDueSnoozes(at); + for (const row of due) { + try { + const reopened = await deps.store.claimAndReopen(row, at); + if (reopened) publishReopened(deps.bus, row); + } catch (error) { + reportError(error, { + operation: "inbox_unsnooze_sweep", + tenantId: row.tenantId, + extra: { messageId: row.messageId }, + }); + // The claim is transactional (delete + reopen together): a throw + // means neither happened, so the row is still there and the next + // tick retries it instead of leaving the item snoozed forever. + } + } +} + +export function createInboxUnsnoozeSweep(deps: InboxUnsnoozeSweepDeps) { + const now = deps.now ?? (() => new Date()); + let tickInFlight = false; + + async function tick(): Promise { + if (tickInFlight) return; + tickInFlight = true; + try { + await tickInboxUnsnoozeSweep({ store: deps.store, bus: deps.bus }, now()); + } catch (err) { + reportError(err, { operation: "inbox_unsnooze_sweep_tick" }); + } finally { + tickInFlight = false; + } + } + + const interval = setInterval(() => void tick(), POLL_INTERVAL_MS); + if (typeof interval.unref === "function") interval.unref(); + + return { + tick, + stop(): void { + clearInterval(interval); + }, + }; +} diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index c277a9f73..8913fd023 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -331,6 +331,10 @@ import { createCredentialExpirySweep, createDrizzleCredentialExpirySweepStore, } from "./credential-expiry-sweep"; +import { + createDrizzleInboxUnsnoozeSweepStore, + createInboxUnsnoozeSweep, +} from "./inbox-unsnooze-sweep"; import { type } from "arktype"; import { betterAuth } from "better-auth"; @@ -2547,6 +2551,14 @@ export async function createHub(config: HubConfig) { }, }); + // Reopen a snoozed inbox item once its `until` has passed (CL-7208) — a + // light periodic sweep over `@corbits/inbox`'s own snooze table, on the + // same mailboxDb/mailboxBus every other mailbox consumer here shares. + const inboxUnsnoozeSweep = createInboxUnsnoozeSweep({ + store: createDrizzleInboxUnsnoozeSweepStore(mailboxDb), + bus: mailboxBus, + }); + // Shared `FoldedRunsDeps` for every one-shot Myra prompt below (routine // drafting, agent-definition drafting): a real one-shot inference call // that launches a folded run, awaits its single reply, and tears the run @@ -3496,6 +3508,7 @@ export async function createHub(config: HubConfig) { chatOrchestrator.dispose(); routineScheduler.stop(); credentialExpirySweep.stop(); + inboxUnsnoozeSweep.stop(); benchProvisioner.stop(); await insightsUsage.close(); await insightsLatency.close(); diff --git a/apps/hub/test/inbox-unsnooze-sweep.test.ts b/apps/hub/test/inbox-unsnooze-sweep.test.ts new file mode 100644 index 000000000..309a21007 --- /dev/null +++ b/apps/hub/test/inbox-unsnooze-sweep.test.ts @@ -0,0 +1,100 @@ +// The sweep's own orchestration — find due snoozes, claim-and-reopen each, +// publish only on an actual reopen, and never drop a row on a claim +// failure — against an in-memory fake store. Which rows are due, and the +// transactional delete+reopen claim itself, are `@corbits/inbox`'s own +// concern (`findDueSnoozes`/`claimAndReopenSnooze`); this only checks the +// tick loop calls them correctly and behaves on what comes back. +import { describe, expect, test } from "bun:test"; +import type { MailboxEvent, MailboxEventScope } from "@corbits/mailbox"; +import { + tickInboxUnsnoozeSweep, + type InboxUnsnoozeSweepStore, +} from "../src/inbox-unsnooze-sweep"; +import type { DueSnooze } from "@corbits/inbox"; + +function row(overrides: Partial = {}): DueSnooze { + return { + tenantId: "tnt_1", + principalId: "prn_1", + messageId: "msg_1", + ...overrides, + }; +} + +function fakeBus() { + const published: { scope: MailboxEventScope; event: MailboxEvent }[] = []; + return { + published, + bus: { + publish(scope: MailboxEventScope, event: MailboxEvent) { + published.push({ scope, event }); + }, + }, + }; +} + +describe("tickInboxUnsnoozeSweep", () => { + test("reopens every due row and publishes a reopened event for each", async () => { + const due = [row({ messageId: "msg_1" }), row({ messageId: "msg_2" })]; + const claimed: DueSnooze[] = []; + const store: InboxUnsnoozeSweepStore = { + findDueSnoozes: async () => due, + claimAndReopen: async (candidate) => { + claimed.push(candidate); + return true; + }, + }; + const { bus, published } = fakeBus(); + + await tickInboxUnsnoozeSweep({ store, bus }, new Date()); + + expect(claimed.map((c) => c.messageId)).toEqual(["msg_1", "msg_2"]); + expect(published).toHaveLength(2); + expect(published[0]?.event).toEqual({ + type: "mailbox", + id: "msg_1", + op: "enrich", + }); + }); + + test("does not publish when claimAndReopen reports it reopened nothing", async () => { + const store: InboxUnsnoozeSweepStore = { + findDueSnoozes: async () => [row()], + // False: another replica already claimed it, or the message had + // already left `snoozed` — either way, no event to publish. + claimAndReopen: async () => false, + }; + const { bus, published } = fakeBus(); + + await tickInboxUnsnoozeSweep({ store, bus }, new Date()); + + expect(published).toHaveLength(0); + }); + + test("a throw from claimAndReopen is swallowed per-row and does not publish", async () => { + const due = [row({ messageId: "msg_bad" }), row({ messageId: "msg_ok" })]; + const claimed: string[] = []; + const store: InboxUnsnoozeSweepStore = { + findDueSnoozes: async () => due, + claimAndReopen: async (candidate) => { + claimed.push(candidate.messageId); + if (candidate.messageId === "msg_bad") { + throw new Error("connection reset mid-claim"); + } + return true; + }, + }; + const { bus, published } = fakeBus(); + + // Must not throw out of the tick — one bad row shouldn't abort the rest. + await tickInboxUnsnoozeSweep({ store, bus }, new Date()); + + expect(claimed).toEqual(["msg_bad", "msg_ok"]); + // Only the row that actually reopened publishes; the thrown claim is + // left for the next tick to retry (claimAndReopenSnooze's transaction + // rolls back on a throw, so the snooze row is still there) rather than + // being dropped. + expect(published).toHaveLength(1); + expect(published[0]?.event.id).toBe("msg_ok"); + }); +}); diff --git a/packages/inbox/src/routes.ts b/packages/inbox/src/routes.ts index 10d0bdd2c..9b70c3040 100644 --- a/packages/inbox/src/routes.ts +++ b/packages/inbox/src/routes.ts @@ -20,6 +20,7 @@ import { import { reportError } from "@corbits/error-sink"; import type { TenantEnv } from "@intx/hub-api"; import { getLogger } from "@intx/log"; +import { type } from "arktype"; import { Hono, type Context } from "hono"; import { @@ -35,9 +36,15 @@ import { type InboxCounts, type InboxItem, } from "./project"; +import { clearSnoozeUntil, setSnoozeUntil } from "./snooze-store"; import { WORKBENCH_INBOX_PRIORITIES } from "./vocabulary"; import { walkAllOpen } from "./walk"; +// Parsed at the boundary, per AGENTS.md: `until` is untrusted request +// input, never `as`-cast. Required — a snooze with no `until` is exactly +// the bug this ticket fixes (CL-7208). +const SnoozeBodySchema = type({ until: "string" }); + // Page size for bulk product ops (mark-all-read, clear-done, counts). Large // enough that a normal inbox finishes in one round-trip; anything past it // walks with the package's cursor. @@ -389,23 +396,36 @@ export function createInboxRoutes( const tenant = c.get("tenant"); const principal = c.get("principal"); const id = c.req.param("id"); - const body: unknown = await c.req.json().catch(() => ({})); - // `until` is accepted for forward-compat with a scheduled unsnooze; the - // product store today only records the status flip. - let until: string | undefined; - if (typeof body === "object" && body !== null && "until" in body) { - const rawUntil = (body as { until: unknown }).until; - if (rawUntil !== undefined && typeof rawUntil !== "string") { - return c.json({ error: "until must be a string" }, 400); - } - if (typeof rawUntil === "string") until = rawUntil; + const rawBody: unknown = await c.req.json().catch(() => null); + const parsedBody = SnoozeBodySchema(rawBody); + if (parsedBody instanceof type.errors) { + return c.json( + { error: `invalid snooze body: ${parsedBody.summary}` }, + 400, + ); + } + const until = new Date(parsedBody.until); + if (Number.isNaN(until.getTime())) { + return c.json({ error: "until must be a valid timestamp" }, 400); + } + const now = new Date(); + if (until.getTime() <= now.getTime()) { + return c.json({ error: "until must be in the future" }, 400); + } + + const scope = { tenantId: tenant.id, principalId: principal.id, id }; + // `until` is persisted before the status flip: an orphaned snooze row + // on a message that never actually became `snoozed` is harmless (the + // sweep no-ops and cleans it up), but the reverse order — status + // first, `until` second — would leave a message stuck `snoozed` with + // nothing ever recording when to reopen it, reproducing this ticket's + // bug (CL-7208). + await setSnoozeUntil(db, scope, until); + const ok = await enrichMailboxMessage(db, scope, { status: "snoozed" }); + if (!ok) { + await clearSnoozeUntil(db, scope); + return c.json({ error: "not found" }, 404); } - const ok = await enrichMailboxMessage( - db, - { tenantId: tenant.id, principalId: principal.id, id }, - { status: "snoozed" }, - ); - if (!ok) return c.json({ error: "not found" }, 404); publish( bus, { tenantId: tenant.id, principalId: principal.id }, @@ -414,7 +434,7 @@ export function createInboxRoutes( ); return c.json({ ok: true, - until, + until: until.toISOString(), }); }); diff --git a/packages/inbox/test/routes.test.ts b/packages/inbox/test/routes.test.ts index ac1bab718..ae105dd57 100644 --- a/packages/inbox/test/routes.test.ts +++ b/packages/inbox/test/routes.test.ts @@ -83,6 +83,52 @@ describe("GET / cursor/filter cross-check", () => { }); }); +describe("POST /:id/snooze rejects before touching the database (CL-7208)", () => { + test("rejects a missing until", async () => { + const app = mount(); + const response = await app.request("/msg_1/snooze", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({}), + }); + expect(response.status).toBe(400); + }); + + test("rejects an until that isn't a string", async () => { + const app = mount(); + const response = await app.request("/msg_1/snooze", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ until: 12345 }), + }); + expect(response.status).toBe(400); + }); + + test("rejects an until that doesn't parse as a timestamp", async () => { + const app = mount(); + const response = await app.request("/msg_1/snooze", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ until: "not-a-date" }), + }); + expect(response.status).toBe(400); + const body = (await response.json()) as { error: string }; + expect(body.error).toBe("until must be a valid timestamp"); + }); + + test("rejects an until that has already passed, rather than snoozing forever", async () => { + const app = mount(); + const response = await app.request("/msg_1/snooze", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ until: "2020-01-01T00:00:00.000Z" }), + }); + expect(response.status).toBe(400); + const body = (await response.json()) as { error: string }; + expect(body.error).toBe("until must be in the future"); + }); +}); + describe("bulk ops surface a failed inbox walk instead of silently truncating it (CL-7207)", () => { test("GET /counts reports and 500s rather than swallowing a walk failure", async () => { const app = mount(throwingDb("connection reset mid-walk")); From 9ff341372e61fb466cbecdb7a474d187629a1dad Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 11:46:48 -0700 Subject: [PATCH 3/4] Fix a TOCTOU race between the snooze write and a concurrent claim (CL-7208) A Critique pass on the two prior CL-7208 commits found and reproduced a real race against a live Postgres: `/:id/snooze` persisted `until` and flipped `status` to `snoozed` as two independent statements. A sweep tick's `claimAndReopenSnooze` landing in the gap between them could see the snooze row before the status flip landed, find the message not yet `snoozed`, no-op, and delete the row as harmless-looking cleanup -- and then the route's own status flip would still land afterward, leaving the message stuck `snoozed` with the row that was supposed to reopen it already gone. That is exactly the bug CL-7208 exists to fix, reintroduced by the un-transacted write order. Both writes now run inside one `db.transaction`: an uncommitted insert is invisible to any other transaction under read-committed isolation, so a concurrent claim can never observe the snooze row until the status flip has also committed alongside it. A missing message (enrichMailboxMessage returning false) throws a sentinel caught outside the transaction, rolling the insert back too rather than leaving an orphaned row. Adds a regression test that races the route's own snooze write against a concurrent claimAndReopenSnooze call and asserts the message is never left `snoozed` with its snooze row already gone -- whichever side wins the race, the result must be either genuinely `open` or `snoozed` with the row still there for the next tick. Verified by running it against a real Postgres repeatedly (not just typechecked): 8/8 runs green after the fix, reproducibly red before it. --- packages/inbox/src/routes.ts | 45 +++++++++--- packages/inbox/src/snooze-store.ts | 19 ++--- packages/inbox/test/snooze-store.test.ts | 91 ++++++++++++++++++++++++ 3 files changed, 136 insertions(+), 19 deletions(-) diff --git a/packages/inbox/src/routes.ts b/packages/inbox/src/routes.ts index 9b70c3040..0977a800c 100644 --- a/packages/inbox/src/routes.ts +++ b/packages/inbox/src/routes.ts @@ -36,7 +36,7 @@ import { type InboxCounts, type InboxItem, } from "./project"; -import { clearSnoozeUntil, setSnoozeUntil } from "./snooze-store"; +import { setSnoozeUntil } from "./snooze-store"; import { WORKBENCH_INBOX_PRIORITIES } from "./vocabulary"; import { walkAllOpen } from "./walk"; @@ -45,6 +45,11 @@ import { walkAllOpen } from "./walk"; // the bug this ticket fixes (CL-7208). const SnoozeBodySchema = type({ until: "string" }); +/** Thrown inside `/:id/snooze`'s transaction to roll back a snooze-until + * insert for a message that turned out not to be in scope, and caught + * outside to answer 404 instead of a 500. */ +class SnoozeTargetNotFound extends Error {} + // Page size for bulk product ops (mark-all-read, clear-done, counts). Large // enough that a normal inbox finishes in one round-trip; anything past it // walks with the package's cursor. @@ -414,16 +419,34 @@ export function createInboxRoutes( } const scope = { tenantId: tenant.id, principalId: principal.id, id }; - // `until` is persisted before the status flip: an orphaned snooze row - // on a message that never actually became `snoozed` is harmless (the - // sweep no-ops and cleans it up), but the reverse order — status - // first, `until` second — would leave a message stuck `snoozed` with - // nothing ever recording when to reopen it, reproducing this ticket's - // bug (CL-7208). - await setSnoozeUntil(db, scope, until); - const ok = await enrichMailboxMessage(db, scope, { status: "snoozed" }); - if (!ok) { - await clearSnoozeUntil(db, scope); + // Both writes run in one transaction: an uncommitted insert is + // invisible to any other transaction under read-committed isolation, + // so the sweep's own `claimAndReopenSnooze` can never see this snooze + // row until the status flip has *also* committed alongside it. Without + // this, a sweep tick landing between two separate statements could + // find the row before the status flip lands, no-op (the message isn't + // `snoozed` yet) and delete the row as "cleanup," and then the status + // flip would still land afterward — leaving the message stuck + // `snoozed` with the row that was supposed to reopen it already gone. + // That is exactly this ticket's bug, reintroduced by an un-transacted + // write order (CL-7208). Throwing (rather than returning false) on a + // missing message rolls the snooze insert back too, via the sentinel + // below caught outside the transaction. + let notFound = false; + await db + .transaction(async (tx) => { + await setSnoozeUntil(tx, scope, until); + const ok = await enrichMailboxMessage(tx, scope, { status: "snoozed" }); + if (!ok) throw new SnoozeTargetNotFound(); + }) + .catch((error) => { + if (error instanceof SnoozeTargetNotFound) { + notFound = true; + return; + } + throw error; + }); + if (notFound) { return c.json({ error: "not found" }, 404); } publish( diff --git a/packages/inbox/src/snooze-store.ts b/packages/inbox/src/snooze-store.ts index 2f26aa931..631b4ddac 100644 --- a/packages/inbox/src/snooze-store.ts +++ b/packages/inbox/src/snooze-store.ts @@ -19,14 +19,17 @@ export interface SnoozeScope { } /** - * Record when a snoozed message should reopen. Written *before* the - * message's own status flips to `snoozed` (see `routes.ts`'s `/:id/snooze` - * handler): if the status flip then fails, the message is left in - * whatever status it already had with an orphaned snooze row pointing at - * it — harmless, since `claimAndReopenSnooze` below no-ops and cleans up a - * row whose message isn't actually `snoozed`. The other ordering (status - * first, `until` second) would reproduce the exact bug this ticket fixes: - * a message stuck `snoozed` with nothing ever recording when to reopen it. + * Record when a snoozed message should reopen. `routes.ts`'s `/:id/snooze` + * handler calls this and the status flip to `snoozed` inside one + * `db.transaction`, not as two independent statements: an uncommitted + * insert is invisible to any other transaction under read-committed + * isolation, so a concurrent sweep tick's `claimAndReopenSnooze` can never + * observe this row before the status flip has *also* committed alongside + * it. Two separate statements would let a sweep tick land in the gap, see + * the row before the flip, no-op (the message isn't `snoozed` yet), and + * delete the row as harmless-looking cleanup — and then the status flip + * would still land afterward, reproducing this ticket's exact bug: a + * message stuck `snoozed` with nothing left to ever reopen it. */ export async function setSnoozeUntil( db: MailboxDb, diff --git a/packages/inbox/test/snooze-store.test.ts b/packages/inbox/test/snooze-store.test.ts index 913bde091..c0d2ce958 100644 --- a/packages/inbox/test/snooze-store.test.ts +++ b/packages/inbox/test/snooze-store.test.ts @@ -9,10 +9,13 @@ // suite's. import { afterAll, beforeAll, describe, expect, test } from "bun:test"; import postgres from "postgres"; +import { Hono } from "hono"; +import type { TenantEnv } from "@intx/hub-api"; import { createDB, schema } from "@intx/db"; import { generateId } from "@intx/hub-common"; import { + createInMemoryMailboxEventBus, createMailboxDb, enrichMailboxMessage, getMailboxMessage, @@ -21,6 +24,7 @@ import { import { setupDatabase } from "../../../scripts/db-setup"; import { e2eDatabaseUrl } from "../../../scripts/e2e/harness"; +import { createInboxRoutes } from "../src/routes"; import { claimAndReopenSnooze, clearSnoozeUntil, @@ -233,4 +237,91 @@ describeIfDb("snooze-store against a real inbox.snooze table", () => { await mailboxDb.close(); } }); + + test("a sweep claim racing the route's own snooze write never strands the message `snoozed` with no way to reopen it", async () => { + // Regression test for a race a Critique pass on CL-7208 found and + // reproduced: `POST /:id/snooze` used to persist `until` and flip + // `status` to `snoozed` as two separate statements. A sweep tick's + // `claimAndReopenSnooze` landing in the gap between them could see the + // snooze row before the status flip landed, no-op (the message wasn't + // `snoozed` yet) and delete the row as harmless cleanup — and then the + // route's own status flip would still land afterward, leaving the + // message stuck `snoozed` with the row that was supposed to reopen it + // already gone. `routes.ts` now runs both writes in one transaction so + // the row is invisible to a concurrent claim until the flip commits + // with it; this drives the same interleaving the bug needed and + // asserts the message is never left `snoozed` with zero snooze rows. + const mailboxDb = createMailboxDb(scratchUrl); + try { + const written = await writeMailboxMessage(mailboxDb.db, { + tenantId, + principalId, + address: `${principalId}@inbox.test`, + fromAddress: "routine:test", + subject: "Race the sweep", + body: "test body", + }); + const messageId = written?.id; + if (messageId === undefined) throw new Error("message not written"); + const scope = { tenantId, principalId, id: messageId }; + + const app = new Hono(); + app.use("*", async (c, next) => { + c.set("tenant", { id: tenantId } as never); + c.set("principal", { id: principalId } as never); + await next(); + }); + app.route( + "/", + createInboxRoutes({ + db: mailboxDb.db, + bus: createInMemoryMailboxEventBus(), + }), + ); + + const until = new Date(Date.now() + 50); + // A "due" check has to compare against a fixed point safely past + // `until`, not real wall-clock time at check time — the whole + // request/claim/verify round trip below runs well under 50ms, so a + // `new Date()` taken after it finishes can still be *before* + // `until`, which would make an already-correct row look "not due + // yet" rather than reopened. `wellPastUntil` is what the real sweep + // would eventually pass once `until` has genuinely elapsed. + const wellPastUntil = new Date(until.getTime() + 60_000); + + const [response] = await Promise.all([ + app.request(`/${messageId}/snooze`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ until: until.toISOString() }), + }), + // Fired concurrently with the route's own write, already past + // `until` — exactly the timing the bug needed: the sweep + // considers the row due from the moment it can see it at all. + claimAndReopenSnooze( + mailboxDb.db, + { tenantId, principalId, messageId }, + wellPastUntil, + ), + ]); + expect(response.status).toBe(200); + + const message = await getMailboxMessage(mailboxDb.db, scope); + const due = await findDueSnoozes(mailboxDb.db, wellPastUntil); + const hasSnoozeRow = due.some((r) => r.messageId === messageId); + + // The invariant the bug violated: never `snoozed` with nothing left + // to ever reopen it. Whichever side of the race won, the message + // must end up either genuinely `open` (the claim won) or `snoozed` + // with its row still there to be claimed on a later tick (the route + // won) — never `snoozed` with the row already gone. + if (message?.status === "snoozed") { + expect(hasSnoozeRow).toBe(true); + } else { + expect(message?.status).toBe("open"); + } + } finally { + await mailboxDb.close(); + } + }); }); From 9843bded503a8beac8cfc869935d146d993ad99a Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 12:02:49 -0700 Subject: [PATCH 4/4] Allowlist the inbox snooze table with its tenancy ruling --- scripts/checks/no-product-tenancy.ts | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/scripts/checks/no-product-tenancy.ts b/scripts/checks/no-product-tenancy.ts index ea4632ef9..3d6392f05 100644 --- a/scripts/checks/no-product-tenancy.ts +++ b/scripts/checks/no-product-tenancy.ts @@ -143,6 +143,18 @@ const ALLOWLIST: readonly { maxOccurrences: 1, tables: ["pending_seed"], }, + { + // `@corbits/mailbox`'s enrichment carries only priority, + // classification and status — there is no column for a snooze's + // `until` timestamp, so the reopen sweep has nothing to scan. This + // table is that one column, keyed to the same + // (tenantId, principalId, messageId) scope every mailbox mutation + // already uses, in its own `inbox` schema. Tenancy, principals and + // grants stay native (CL-7208). + relPath: "packages/inbox/src/schema.ts", + maxOccurrences: 1, + tables: ["snooze"], + }, { relPath: "packages/slack-tag/src/schema.ts", maxOccurrences: 1,