fix(plugin-email): send() releases an insert-assigned row id from managedRowIds (#5169) (#5524)

`EmailPersistence` is a public interface whose `insert` may answer with an id of
its own — a database-assigned primary key, an external delivery system's receipt
id. `sendInternal()` reserved that id as service-managed (so the `sys_email`
afterInsert outbox drain skips a row `send()` is already delivering) but its
`finally` released only the id it had minted itself, so every such message left
one entry behind for the life of the process.

The memory is the smaller half. `isServiceManaged(persistedId)` stayed true
forever, and that is a standing "this row belongs to a live send()" assertion
which both the drain hook and the boot outbox sweep (#5161) trust and nothing
ever re-checks — a row stranded at `queued` under such an id would be skipped by
every future sweep.

The id now travels in a variable declared outside the try (`extraManagedId`) and
the `finally` releases exactly what was reserved, so the release condition can
never drift from the reservation condition. The window itself is unchanged: the
release still happens after inline delivery has finalized the row and after
queue mode has published its job.

No in-repo path reached the branch — the plugin's own persistence returns the id
it was given, because ObjectQL's insert echoes `row.id` — so this was reachable
only through a custom `EmailPersistence`. Two tests pin it, one per delivery
mode, and the inline one also asserts the managed window still covers the row's
`sent` finalize so a future "release earlier" cannot pass silently.

Claude-Session: https://claude.ai/code/session_01BWS4heBoAitLmzCLhcYdbK

Co-authored-by: Claude <noreply@anthropic.com>

Author: os-zhuang

.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);
     }
   }