diff --git a/.changeset/email-managed-row-ids-release-persisted-id.md b/.changeset/email-managed-row-ids-release-persisted-id.md deleted file mode 100644 index 38fecce886..0000000000 --- a/.changeset/email-managed-row-ids-release-persisted-id.md +++ /dev/null @@ -1,29 +0,0 @@ ---- -"@objectstack/plugin-email": patch ---- - -fix(plugin-email): `send()` releases an insert-assigned row id from `managedRowIds` instead of leaking it (#5169) - -`EmailPersistence.insert` is a **public** interface and may answer with an id of -its own — a database-assigned primary key, an external delivery system's receipt -id. `send()` reserves that id as service-managed too (so the `sys_email` -`afterInsert` outbox drain skips a row `send()` is already delivering), but its -`finally` only released the id `send()` had minted. The insert-assigned one was -reserved and never released. - -Two consequences, both now fixed: - -- **memory** — one leaked string per message in a `Set` that lives as long as the - process; -- **semantics** — `isServiceManaged(persistedId)` stayed true forever. Ids are - unique, so no other row was mistaken for a managed one, but that entry is a - standing "this row belongs to a live `send()`" assertion which the drain hook - and the boot outbox sweep (#5161) both trust and nothing ever re-checks: a row - stranded at `queued` under such an id would be skipped by every future sweep. - -The reservation window is unchanged — the release still happens in the same -`finally`, after inline delivery has finalized the row and after queue mode has -published the job, so nothing that relied on the row reading managed *during* -`send()` is affected. The in-repo persistence returns the id it was given -(ObjectQL echoes `row.id`), so no in-repo path ever reached the leaking branch; -this was reachable only by a custom `EmailPersistence` implementation. diff --git a/.changeset/email-persistence-insert-id-contract.md b/.changeset/email-persistence-insert-id-contract.md new file mode 100644 index 0000000000..c78d7bab1a --- /dev/null +++ b/.changeset/email-persistence-insert-id-contract.md @@ -0,0 +1,47 @@ +--- +"@objectstack/plugin-email": patch +--- + +fix(plugin-email)!: `EmailPersistence.insert` must return the row's own id — a substituted id is rejected instead of double-sending (#5523) + +**FROM** — `insert` could answer with an id of its own (a database-assigned +primary key, an external delivery system's receipt id) and `EmailService.send()` +adopted it: the substituted id was added to the service-managed set, used as the +queued job's `rowId`, and returned to the caller. + +**TO** — `insert` must confirm the id it was handed. Returning a different id +throws, naming the contract and the value returned, **before the message is +delivered**. + +**Fix, one line:** return `{ id: row.id }` (or `row.id`) from `insert`. If your +store assigns its own primary key, keep the service-minted id in the row's `id` +column and record the store's key in a column of its own. + +Why the contract tightened rather than the service accommodating both: the id is +minted by the service *before* the insert and is already load-bearing by the time +`insert` is called — out-of-row attachment content has been uploaded under +`sys_email/attachments//…`, so the row id is the only key that finds +those bytes again. Re-keying the row also broke delivery exactly-once: the +`sys_email` `afterInsert` outbox drain hook decides whether a freshly-inserted +row is the service's to deliver by asking `isServiceManaged()` about **the +inserted row's own id**, and that hook runs *inside* the insert — before `send()` +had seen, let alone reserved, the substituted id. So the hook read the row as an +application-inserted outbox entry and delivered it, while `send()` delivered it +again down its own path: one message sent twice, two terminal updates racing on +one row. The only thing that ever prevented it was the hook's `setTimeout(…, 0)` +losing a race to `send()`'s inline delivery — and `transport.send` is real +network I/O, so that race is normally lost. + +Scope of the check: it judges the confirmation's **value**, not its presence. An +implementation that returns no id at all leaves nothing to disagree with (the +drain hook reads the id off the inserted row, which is the minted one either +way), so the mail still goes. An insert that *throws* is unchanged — that stays +an operational condition the service rides out with a warning and inline +delivery; only a *successful* insert that renames the row is fatal. + +Breaking for external `EmailPersistence` implementations that re-key the row — +of which there are currently none: the in-repo implementation forwards the +engine's own answer and ObjectQL honours the id it is handed. Filed at `patch` +because the surface has no known external consumer and the declared TypeScript +signature is unchanged; a maintainer who counts a narrowed public-interface +contract as `minor`/`major` should relabel it. diff --git a/packages/plugins/plugin-email/src/email-plugin.outbox-sweep.test.ts b/packages/plugins/plugin-email/src/email-plugin.outbox-sweep.test.ts index 6d110f08ef..e98006df26 100644 --- a/packages/plugins/plugin-email/src/email-plugin.outbox-sweep.test.ts +++ b/packages/plugins/plugin-email/src/email-plugin.outbox-sweep.test.ts @@ -31,6 +31,13 @@ const flush = () => new Promise((r) => setTimeout(r, 5)); interface EngineOpts { /** Make `find` throw for the drain hook's own `where: { id }` re-read. */ failIdLookup?: boolean; + /** + * Store every `sys_email` insert under THIS id instead of the one it was + * handed, and answer with it — a datasource that assigns its own primary key + * (#5523). The row the afterInsert hook sees then carries an id `send()` never + * reserved, which is the whole mechanism the insert contract closes. + */ + reKeyInsert?: string; } /** @@ -84,12 +91,13 @@ function fakeEngine(opts: EngineOpts = {}) { return out; }, async insert(table: string, data: any) { - const row = { ...data }; + const reKeyed = opts.reKeyInsert && table === 'sys_email'; + const row = reKeyed ? { ...data, id: opts.reKeyInsert } : { ...data }; const t = rowsOf(table); t.push(row); tables.set(table, t); for (const fn of hooks.afterInsert ?? []) await fn({ object: table, result: row }); - return { id: data.id }; + return { id: row.id }; }, async update(table: string, patch: any) { const r = rowsOf(table).find((x) => x.id === patch.id); @@ -423,3 +431,50 @@ describe('normal delivery is unchanged', () => { expect(h.ctx.logger.error).not.toHaveBeenCalled(); }); }); + +// ── the insert id contract, at the seam that motivated it (#5523) ─────────── + +describe('a datasource that re-keys a sys_email insert', () => { + it('makes send() reject by name instead of double-sending through the drain hook', async () => { + // The full mechanism, end to end, through the REAL plugin persistence + // adapter (which forwards the engine's own answer rather than laundering it + // back into `row.id`): + // + // send() mints id → reserves it → engine.insert stores the row under + // 'db-pk-7' and fires afterInsert with THAT id → the drain hook asks + // isServiceManaged('db-pk-7'), gets false, and schedules a delivery of + // its own → insert returns 'db-pk-7'. + // + // Before the contract check, `send()` then adopted 'db-pk-7' and delivered + // the message a second time down its own path: two deliveries of one mail + // and two terminal updates racing on one row. The only thing that ever + // stopped it was the drain hook's `setTimeout(0)` losing to send()'s inline + // delivery — so this transport is deliberately SLOWER than that timer, + // which is what real network I/O is. Reverting the check turns the + // `toHaveBeenCalledTimes(1)` below into 2. + const transport = { + send: vi.fn(async () => { + await new Promise((r) => setTimeout(r, 25)); + return { messageId: '' }; + }), + }; + const h = await boot({ engine: { reKeyInsert: 'db-pk-7' }, transport }); + await h.ready(); + + await expect( + h.service().send({ to: 'a@b.com', from: 'x@y.com', subject: 'Hi', text: 'hello' }), + ).rejects.toThrow(/EmailPersistence\.insert must return the row's own id/); + + // Give the hook's already-scheduled delivery time to finish. It is NOT + // cancellable: the hook runs synchronously inside `engine.insert`, i.e. + // before `insert` has even returned to the service, so the contract check + // cannot pre-empt it. What the check removes is send()'s SECOND delivery — + // the row that did land is still delivered once, by the hook, and the + // caller is told loudly that its persistence is misconfigured. + await new Promise((r) => setTimeout(r, 120)); + + expect(transport.send).toHaveBeenCalledTimes(1); + expect(h.sysEmail()).toHaveLength(1); + expect(h.sysEmail()[0]).toMatchObject({ id: 'db-pk-7', status: 'sent' }); + }); +}); diff --git a/packages/plugins/plugin-email/src/email-plugin.ts b/packages/plugins/plugin-email/src/email-plugin.ts index 65b2506e82..4f5528c178 100644 --- a/packages/plugins/plugin-email/src/email-plugin.ts +++ b/packages/plugins/plugin-email/src/email-plugin.ts @@ -486,6 +486,14 @@ export class EmailServicePlugin implements Plugin { const created = await (engine as any).insert('sys_email', row, { context: SYSTEM_CTX, }); + // The engine's OWN answer is forwarded, not laundered into + // `row.id`. ObjectQL honours the id it is handed, so this confirms + // `row.id` — and on the day some driver re-keys the row instead, + // `EmailPersistence`'s insert contract (#5523) says so loudly here + // rather than letting the outbox drain hook double-send the + // message. `created` without a usable id means the engine reported + // no identity at all, which is not a disagreement: confirm the row + // we asked for. return created?.id ? { id: String(created.id) } : { id: String(row.id) }; }, async update(id, patch) { diff --git a/packages/plugins/plugin-email/src/email-service.queue-delivery.test.ts b/packages/plugins/plugin-email/src/email-service.queue-delivery.test.ts index be0c6a58ee..e4597eb04e 100644 --- a/packages/plugins/plugin-email/src/email-service.queue-delivery.test.ts +++ b/packages/plugins/plugin-email/src/email-service.queue-delivery.test.ts @@ -189,12 +189,17 @@ describe('EmailService — queue delivery on', () => { expect(svc.isServiceManaged(res.id)).toBe(false); }); - it('releases an insert-assigned id once the job is published (#5169)', async () => { - // Queue mode returns EARLY (right after the publish), so the release of an - // insert-assigned id happens on that path too — and it must, because the - // row is now the worker's: the boot outbox sweep decides whether to requeue - // it by asking `isServiceManaged`, and a permanently-true answer would make - // a stranded row unsweepable forever. + it('rejects an insert that returns a different id — and publishes NO job (#5523)', async () => { + // Re-judged from #5169's "releases an insert-assigned id once the job is + // published". That test pinned the release on the queue path's EARLY return + // for an insert-assigned id; `EmailPersistence.insert` may no longer assign + // one, so the scenario is now a rejection. The release-on-early-return fact + // it also carried is still pinned by the test above, which asserts + // `isServiceManaged(res.id) === false` after a queued send. + // + // Queue mode matters on its own here: a job that referenced the substituted + // id would send the mail from the WORKER as well, so the rejection has to + // land before the publish, not just before inline delivery. const queue = makeQueue(); const transport = { send: vi.fn(async () => ({ messageId: '' })) }; const persistence: EmailPersistence = { @@ -205,11 +210,10 @@ describe('EmailService — queue delivery on', () => { transport, defaultFrom: 'no@reply.com', persistence, queueDelivery: wiring(queue), }); - const res = await svc.send(MSG); + await expect(svc.send(MSG)).rejects.toThrow(/EmailPersistence\.insert must return the row's own id/); - // The job references the PERSISTED id, and that id is no longer managed. - expect(res).toMatchObject({ id: 'db-pk-9', status: 'queued' }); - expect(queue.published[0].data).toEqual({ rowId: 'db-pk-9' }); + expect(queue.published).toHaveLength(0); // nothing for a worker to double-send + expect(transport.send).not.toHaveBeenCalled(); expect(svc.isServiceManaged('db-pk-9')).toBe(false); }); diff --git a/packages/plugins/plugin-email/src/email-service.test.ts b/packages/plugins/plugin-email/src/email-service.test.ts index dac42d4b6f..26b606783f 100644 --- a/packages/plugins/plugin-email/src/email-service.test.ts +++ b/packages/plugins/plugin-email/src/email-service.test.ts @@ -147,6 +147,7 @@ describe('EmailService', () => { // afterInsert drain hook would fire — assert the row is flagged managed // there, and unflagged once send() resolves. let managedAtInsert: boolean | undefined; + let managedDuringDelivery: boolean | undefined; let insertedId: string | undefined; const transport = { send: vi.fn(async () => ({ messageId: '' })) }; let svc!: EmailService; @@ -156,42 +157,133 @@ describe('EmailService', () => { managedAtInsert = svc.isServiceManaged(insertedId); return { id: row.id }; }, - async update() { /* noop */ }, + async update(id) { + // The `sent` finalize runs INSIDE send(), i.e. while this row is still + // send()'s to deliver — the window the managed flag exists to protect + // must NOT shrink. (Inherited from the #5169 test this file used to + // carry on the insert-assigned-id path; that path is now a contract + // violation, so the fact is pinned here, on the compliant one.) + managedDuringDelivery = svc.isServiceManaged(String(id)); + }, }; svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence }); const res = await svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' }); expect(res.status).toBe('sent'); expect(managedAtInsert).toBe(true); // hook would skip it + expect(managedDuringDelivery).toBe(true); // window kept through finalize expect(svc.isServiceManaged(insertedId!)).toBe(false); // cleared after send }); - it('releases an insert-ASSIGNED row id from the managed set too (#5169)', async () => { - // `EmailPersistence` is public and its `insert` may answer with an id of - // its own — a database-assigned primary key, an external delivery system's - // receipt id. `send()` reserves that id as managed as well; the bug was - // that it never released it, so `isServiceManaged(persistedId)` stayed true - // forever: one leaked entry per message, and a "belongs to a live send()" - // assertion the drain hook and the boot sweep trust but nobody re-checks. - let managedDuringDelivery: boolean | undefined; + // ── EmailPersistence.insert id contract (#5523) ──────────────────── + // + // Re-judged from #5169's "releases an insert-ASSIGNED row id" test. That test + // pinned the *leak fix* on a capability — `insert` answering with an id of its + // own — that is now a contract violation, so the leak has no source left to + // fix. Same scenario, opposite verdict: rejected, by name, before delivery. + it('rejects an insert that returns a DIFFERENT id, naming the contract, before delivering', async () => { + // The double-send this closes: the sys_email afterInsert drain hook asks + // `isServiceManaged(hookCtx.result.id)` — the INSERTED row's id. A + // persistence that re-keys the row hands the hook an id send() never + // reserved, the hook reads the row as an app-inserted outbox entry and + // delivers it, and send() delivers it again down its own path. const transport = { send: vi.fn(async () => ({ messageId: '' })) }; - let svc!: EmailService; + const update = vi.fn(async () => { /* noop */ }); const persistence: EmailPersistence = { - // Ignores the minted id and hands back the row's real (DB) key. + // Ignores the minted id and hands back a DB primary key of its own. async insert() { return { id: 'db-pk-7' }; }, - async update(id) { - // The `sent` finalize runs INSIDE send(), i.e. while this row is still - // send()'s to deliver — the window the managed flag exists to protect - // must NOT shrink to make the release possible. - managedDuringDelivery = svc.isServiceManaged(String(id)); - }, + update, }; - svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence }); + const svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence }); + + await expect(svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' })) + .rejects.toThrow(/EmailPersistence\.insert must return the row's own id/); + + // Names the value actually returned, so the operator can find the + // implementation that returned it. + await expect(svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' })) + .rejects.toThrow(/'db-pk-7'/); + + // BEFORE delivery — not "double-send, then complain". This is the whole + // point of where the check sits: nothing was sent, nothing was finalized. + expect(transport.send).not.toHaveBeenCalled(); + expect(update).not.toHaveBeenCalled(); + + // And the throw does not leak the reservation — the `finally` still runs. + expect(svc.isServiceManaged('db-pk-7')).toBe(false); + }); + + it('accepts the bare-string confirmation form when it names the row id', async () => { + // `insert` may answer `row.id` instead of `{ id: row.id }` — same + // confirmation, and it must not be mistaken for a substitution. + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const persistence: EmailPersistence = { + async insert(row) { return String(row.id); }, + async update() { /* noop */ }, + }; + const svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence }); + + const res = await svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' }); + + expect(res.status).toBe('sent'); + expect(transport.send).toHaveBeenCalledTimes(1); + expect(svc.isServiceManaged(res.id)).toBe(false); + }); + + it('rejects a bare-string confirmation that names a different id', async () => { + // The string form is the same contract, not a loophole around it. + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const persistence: EmailPersistence = { + async insert() { return 'db-pk-8'; }, + async update() { /* noop */ }, + }; + const svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence }); + + await expect(svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' })) + .rejects.toThrow(/EmailPersistence\.insert must return the row's own id/); + expect(transport.send).not.toHaveBeenCalled(); + }); + + it('treats a confirmation that names NO id as confirming the row it inserted', async () => { + // Only reachable from an untyped JS implementation (the declared return + // type requires an id). There is no id to disagree with, and the drain hook + // reads the id off the inserted ROW — the minted one — so no double-send is + // reachable this way. The check judges the VALUE, not its presence, and an + // insert that silently returned nothing must not stop the mail. + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const persistence = { + async insert() { /* returns undefined */ }, + async update() { /* noop */ }, + } as unknown as EmailPersistence; + const svc = new EmailService({ transport, defaultFrom: 'no@reply.com', persistence }); const res = await svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' }); - expect(res).toMatchObject({ id: 'db-pk-7', status: 'sent' }); - expect(managedDuringDelivery).toBe(true); // window kept - expect(svc.isServiceManaged('db-pk-7')).toBe(false); // released, not leaked + expect(res.status).toBe('sent'); + expect(transport.send).toHaveBeenCalledTimes(1); + }); + + it('still rides out an insert that THROWS — an operational failure, not a contract violation', async () => { + // The two must stay distinguishable: a failing insert keeps its non-fatal + // warning path (mail goes inline), while a *renaming* insert is fatal. If + // the contract check ever moved inside the insert's catch, this test would + // still pass and the rejection test above would go green for the wrong + // reason — hence both, side by side. + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const warn = vi.fn(); + const persistence: EmailPersistence = { + async insert() { throw new Error('db down'); }, + async update() { /* noop */ }, + }; + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence, + logger: { info: vi.fn(), warn }, + }); + + const res = await svc.send({ to: 'a@b.com', subject: 'Hi', text: 'x' }); + + expect(res.status).toBe('sent'); + expect(transport.send).toHaveBeenCalledTimes(1); + expect(warn).toHaveBeenCalledWith(expect.stringContaining('persist failed'), expect.any(Object)); }); it('deliverPersistedRow delivers an existing row WITHOUT inserting a new one', async () => { diff --git a/packages/plugins/plugin-email/src/email-service.ts b/packages/plugins/plugin-email/src/email-service.ts index 2f3494b591..b6513b3b4e 100644 --- a/packages/plugins/plugin-email/src/email-service.ts +++ b/packages/plugins/plugin-email/src/email-service.ts @@ -115,12 +115,73 @@ export interface RowToNormalizedOptions { /** * Internal persistence shim — typed loosely so the service can run * without an ObjectQL engine wired (e.g. unit tests, serverless). + * + * ## `insert` must return the row's own id (#5523) + * + * The id is **minted by the service**, written into `row.id`, and is already + * load-bearing by the time `insert` is called: out-of-row attachment content + * has been uploaded under `sys_email/attachments//…` *before* the + * insert, so the row id is the only key that finds those bytes again. + * + * `insert` therefore **must** answer with that same id — `{ id: row.id }` (or + * `row.id`). The return value is a **confirmation**, not an opportunity to + * substitute an id of the implementation's own: returning a different one makes + * `send()` throw, naming this contract and the id returned, before the message + * is delivered. + * + * Why the contract is this way round, rather than the service adopting whatever + * id came back: the `sys_email` `afterInsert` outbox drain hook decides whether + * a freshly-inserted row is the service's to deliver by asking + * {@link EmailService.isServiceManaged} about **the inserted row's own id**. An + * implementation that re-keyed the row would hand the hook an id the service + * had never reserved, the hook would read the row as an application-inserted + * outbox entry, and the same message would go out twice — once from the hook, + * once from `send()` — with two terminal updates racing on one row. + * + * A store that insists on assigning its own primary key is still supportable: + * keep the service-minted id as the row's `id` and record the store's key in a + * column of its own. What is not supportable is silently changing the identity + * the rest of the pipeline has already committed to. */ export interface EmailPersistence { + /** + * Persist `row` and confirm it by returning **`row.id`** — see the contract + * on {@link EmailPersistence}. Returning any other id throws. + */ insert(row: Record): Promise<{ id: string } | string>; update?(id: string, patch: Record): Promise; } +/** + * Enforce {@link EmailPersistence}'s `insert` contract: the confirmation must + * name the row the service asked to have inserted. + * + * Scope is deliberate — this judges the **value** of the confirmation, not its + * presence. A confirmation that names no id at all (`undefined`/`null`, only + * reachable from an untyped JS implementation, since the declared return type + * requires one) leaves nothing to disagree with and is treated as confirmed: + * the drain hook reads the id off the **inserted row**, which is the minted one + * either way, so no double-send is reachable that way. The defect this closes + * is a *different* id, and that is what it rejects. + * + * @throws Error naming the contract, the id returned, and the id expected. + */ +function assertInsertConfirmedRowId(res: { id: string } | string | undefined, rowId: string): void { + const returned = typeof res === 'string' ? res : res == null ? undefined : res.id; + if (returned == null) return; + const confirmed = String(returned); + if (confirmed === rowId) return; + throw new Error( + 'EmailService: EmailPersistence.insert must return the row\'s own id — it returned ' + + `'${confirmed}' for a row inserted as '${rowId}'. The return value is a CONFIRMATION only: the id is ` + + 'minted by the service before the insert (out-of-row attachment content is already stored under ' + + `'sys_email/attachments/${rowId}/…'), and re-keying the row makes the sys_email outbox drain hook read it ` + + 'as an application-inserted entry — delivering the same message a second time. Fix: return ' + + '`{ id: row.id }` (or `row.id`) from insert; if your store assigns its own primary key, keep the ' + + 'service id in the row\'s `id` column and record the store\'s key in a column of its own.', + ); +} + /** * Naive RFC-5322 validator — good enough to catch obvious typos. * Defers full validation to the transport / receiving MTA. @@ -584,35 +645,40 @@ export class EmailService implements IEmailService { // (which fires synchronously inside that insert) sees it as managed // and skips it — `send()` owns this row's delivery. // - // `EmailPersistence` is a PUBLIC interface and its `insert` may answer with - // an id of its own (a database-assigned primary key, an external delivery - // system's receipt id). That id is reserved too — and it has to be released - // by the same `finally`, which is why it is held in a variable declared - // OUT here rather than recomputed from `persistedId` inside the try (#5169). - // Reserving without releasing would leave `isServiceManaged(persistedId)` - // permanently true: one leaked string per message for the life of the - // process, and — worse than the memory — a standing "this row belongs to a - // live send()" assertion that the drain hook and the boot outbox sweep both - // trust and nothing ever re-checks. - let extraManagedId: string | undefined; + // ONE id, reserved once and released once. `EmailPersistence.insert` is + // contractually required to confirm the id it was handed (#5523), so there + // is no second identity for this row to reserve: the drain hook, the boot + // outbox sweep, the attachment storage keys and the queued job all name + // `id`. An implementation that returns something else is rejected below, + // before delivery — the service does not adopt the substitute. this.managedRowIds.add(id); try { - let persistedId: string | undefined; + // Did the row land? That is the only thing the insert's answer tells us + // (the id is already known), so it is a boolean and not an id. + let rowPersisted = false; if (this.options.persistence) { + let res: { id: string } | string | undefined; + let insertThrew = false; try { - const res = await this.options.persistence.insert(baseRow); - persistedId = typeof res === 'string' ? res : res?.id ?? id; - if (persistedId !== id) { - this.managedRowIds.add(persistedId); - extraManagedId = persistedId; - } + res = await this.options.persistence.insert(baseRow); } catch (err: any) { + insertThrew = true; this.options.logger?.warn('EmailService: sys_email persist failed (non-fatal)', { error: err?.message }); } + if (!insertThrew) { + // Deliberately OUTSIDE the catch above. An insert that *fails* is an + // operational condition the service is built to ride out (the mail + // still goes, inline, with a warning). An insert that succeeds while + // renaming the row is a WIRING defect: tolerate it and the drain hook + // double-sends this very message. So it escapes the non-fatal path, + // and it is raised here — before any delivery below — rather than + // after the second copy has already gone out. + assertInsertConfirmedRowId(res, id); + rowPersisted = true; + } } - const rowId = persistedId ?? id; const storageKeys = encodedAttachments.kind === 'storage' ? encodedAttachments.keys : []; - if (persistedId === undefined && storageKeys.length > 0) { + if (!rowPersisted && storageKeys.length > 0) { // Content was uploaded for a row that does not exist. No row will ever // reference these bytes, so nothing will ever reclaim them either — // delete them here rather than create the one orphan class this design @@ -624,23 +690,23 @@ export class EmailService implements IEmailService { // job with nothing to reference. Deliver inline instead of publishing // a job that can only fail — the insert failure was already reported // above, and dropping the message would be the worse answer. - if (persistedId === undefined) { + if (!rowPersisted) { this.reportQueueDegradation( 'the sys_email row could not be persisted, so a queued job would have nothing to deliver', ); - } else if (await this.publishRow(queue, rowId)) { + } else if (await this.publishRow(queue, id)) { // The row is in the database at `queued` and the job is in the // queue's own store: a process death here loses neither. - return { id: rowId, status: 'queued' }; + return { id, status: 'queued' }; } } // Inline delivery of a message whose content IS in storage (the publish // failed, or queue mode is off for this send): the in-memory message is // still whole, so the mail goes out — and the content still has to be // reclaimed afterwards, which is why the keys travel with the delivery. - return await this.deliverNormalized(rowId, normalized, undefined, storageKeys); + return await this.deliverNormalized(id, normalized, undefined, storageKeys); } finally { - // Release EXACTLY what was reserved above — both ids, and only here. + // Release EXACTLY what was reserved above — the one id, and only here. // Here and not earlier: the reservation has to outlive the whole body, // because in inline mode the delivery (and the `sent`/`failed` update of // this very row) happens inside the try, and a sweep that ran mid-flight @@ -649,7 +715,6 @@ export class EmailService implements IEmailService { // the worker's — with the row committed at `queued`, re-checkable, which // is what makes the boot sweep a backstop rather than a double-send. this.managedRowIds.delete(id); - if (extraManagedId !== undefined) this.managedRowIds.delete(extraManagedId); } }