fix(plugin-email): send() 释放 insert 分配的行 id,不再泄漏进 managedRowIds (#5169)

.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.

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: '<x>' })) };
+ 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: '<live@x>' })) };
  const queue = makeQueue();

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: '<m@x>' })) };
+ 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: '<drained@x>' })) };
  const { p, rows } = makePersistence();

packages/plugins/plugin-email/src/email-service.ts

@@ -583,14 +583,29 @@ 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;
  if (this.options.persistence) {
  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);
  }
  }