diff --git a/.changeset/email-managed-row-ids-release-persisted-id.md b/.changeset/email-managed-row-ids-release-persisted-id.md new file mode 100644 index 0000000000..38fecce886 --- /dev/null +++ b/.changeset/email-managed-row-ids-release-persisted-id.md @@ -0,0 +1,29 @@ +--- +"@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/packages/plugins/plugin-email/src/email-service.queue-delivery.test.ts b/packages/plugins/plugin-email/src/email-service.queue-delivery.test.ts index d59ea67328..be0c6a58ee 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,6 +189,30 @@ 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. + const queue = makeQueue(); + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + const persistence: EmailPersistence = { + async insert() { return { id: 'db-pk-9' }; }, + async update() { /* noop */ }, + }; + const svc = new EmailService({ + transport, defaultFrom: 'no@reply.com', persistence, queueDelivery: wiring(queue), + }); + + const res = await svc.send(MSG); + + // 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(svc.isServiceManaged('db-pk-9')).toBe(false); + }); + it('sendInline() bypasses the queue — the mail/test path', async () => { const transport = { send: vi.fn(async () => ({ messageId: '' })) }; const queue = makeQueue(); diff --git a/packages/plugins/plugin-email/src/email-service.test.ts b/packages/plugins/plugin-email/src/email-service.test.ts index 146238fe0a..dac42d4b6f 100644 --- a/packages/plugins/plugin-email/src/email-service.test.ts +++ b/packages/plugins/plugin-email/src/email-service.test.ts @@ -165,6 +165,35 @@ describe('EmailService', () => { 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; + const transport = { send: vi.fn(async () => ({ messageId: '' })) }; + let svc!: EmailService; + const persistence: EmailPersistence = { + // Ignores the minted id and hands back the row's real (DB) key. + 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)); + }, + }; + 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 + }); + it('deliverPersistedRow delivers an existing row WITHOUT inserting a new one', async () => { const transport = { send: vi.fn(async () => ({ messageId: '' })) }; const { p, rows } = makePersistence(); diff --git a/packages/plugins/plugin-email/src/email-service.ts b/packages/plugins/plugin-email/src/email-service.ts index 5098821676..2f3494b591 100644 --- a/packages/plugins/plugin-email/src/email-service.ts +++ b/packages/plugins/plugin-email/src/email-service.ts @@ -583,6 +583,18 @@ export class EmailService implements IEmailService { // Reserve the row id BEFORE persistence.insert so the drain hook // (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; this.managedRowIds.add(id); try { let persistedId: string | undefined; @@ -590,7 +602,10 @@ export class EmailService implements IEmailService { try { const res = await this.options.persistence.insert(baseRow); persistedId = typeof res === 'string' ? res : res?.id ?? id; - if (persistedId !== id) this.managedRowIds.add(persistedId); + if (persistedId !== id) { + this.managedRowIds.add(persistedId); + extraManagedId = persistedId; + } } catch (err: any) { this.options.logger?.warn('EmailService: sys_email persist failed (non-fatal)', { error: err?.message }); } @@ -625,7 +640,16 @@ export class EmailService implements IEmailService { // reclaimed afterwards, which is why the keys travel with the delivery. return await this.deliverNormalized(rowId, normalized, undefined, storageKeys); } finally { + // Release EXACTLY what was reserved above — both ids, 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 + // must still see the row as `send()`'s. Once this returns, ownership is + // over in both modes: inline delivery is finished, and a queued row is + // 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); } }