fix(mail-ingest): read a mail as soon as a transport takes it in

Only an upload started a processing run; a forwarded mail waited for the
next two-minute tick, and one that came in right after a run waited for two.
The IMAP and relay runs start one run once they took anything in, the
webhook once per mail it accepted.

server/src/nest/mail-ingest/mail-ingest-imap.job.ts

@@ -10,6 +10,7 @@ import {
   type ImapSession,
 } from './mail-ingest-imap.client';
 import { isPassingRefusal, MailIngestIntakeService } from './mail-ingest-intake.service';
+import { MailIngestProcessJob } from './mail-ingest-process.job';
 import { MailIngestSettingsService, type MailIngestImapConfig } from './mail-ingest-settings.service';
 import { InjectRepository } from '@mikro-orm/nestjs';
 import { Injectable, type OnApplicationBootstrap } from '@nestjs/common';
@@ -51,6 +52,7 @@ export class MailIngestImapJob implements OnApplicationBootstrap {
     private readonly intake: MailIngestIntakeService,
     private readonly imap: MailIngestImapClient,
     private readonly registrar: CronRegistrarService,
+    private readonly process: MailIngestProcessJob,
   ) {}
 
   onApplicationBootstrap(): void {
@@ -76,6 +78,8 @@ export class MailIngestImapJob implements OnApplicationBootstrap {
     try {
       const { taken, uidValidity, lastUid } = await this.poll(config, now);
       await this.cursor.recordOk('imap', config.mailbox, dbNow(new Date()), { uidValidity, lastUid });
+      // Read what came in now, not at the next processing tick.
+      if (taken > 0) this.process.runSoon();
       return taken;
     } catch (err) {
       const reason = describeImapFailure(err);

server/src/nest/mail-ingest/mail-ingest-process.job.ts

@@ -75,6 +75,8 @@ export class MailIngestProcessJob implements OnApplicationBootstrap {
    * Starts a run now instead of at the next tick, without waiting for it.
    * Where the crons are off (the test suites) nothing starts, as no tick
    * would either, so a run never outlives the request that asked for it.
+   *
+   * @txStandalone a run of its own, not awaited, under its own lease; nothing of the caller's writes rides on it.
    */
   runSoon(): void {
     if (!this.registrar.isEnabled()) return;

server/src/nest/mail-ingest/mail-ingest-relay.job.ts

@@ -4,6 +4,7 @@ import { dbNow } from '../../db/types';
 import { logError, logWarn } from '../audit/audit-log.logger';
 import { CronRegistrarService, LEASE_OWNER } from '../scheduling/cron-registrar.service';
 import { isPassingRefusal, MailIngestIntakeService } from './mail-ingest-intake.service';
+import { MailIngestProcessJob } from './mail-ingest-process.job';
 import { MailIngestRelayHost, type MailInboxRelay } from './mail-ingest-relay.service';
 import { MailIngestSettingsService } from './mail-ingest-settings.service';
 import { MAIL_INGEST_LIMITS } from './mail-ingest.constants';
@@ -69,6 +70,7 @@ export class MailIngestRelayJob implements OnApplicationBootstrap {
     private readonly intake: MailIngestIntakeService,
     private readonly relays: MailIngestRelayHost,
     private readonly registrar: CronRegistrarService,
+    private readonly process: MailIngestProcessJob,
   ) {}
 
   onApplicationBootstrap(): void {
@@ -91,6 +93,8 @@ export class MailIngestRelayJob implements OnApplicationBootstrap {
     try {
       const taken = await this.drain(relay, null, PAGES_PER_RUN);
       await this.cursor.recordOk('cloud', '', dbNow(new Date()));
+      // Read what came in now, not at the next processing tick.
+      if (taken > 0) this.process.runSoon();
       return taken;
     } catch (err) {
       const reason = err instanceof Error ? err.message : String(err);

server/src/nest/mail-ingest/mail-ingest-webhook.service.ts

@@ -1,5 +1,6 @@
 import { DomainError } from '../common/domain-error';
 import { MailIngestIntakeService, type MailIntakeRefusal } from './mail-ingest-intake.service';
+import { MailIngestProcessJob } from './mail-ingest-process.job';
 import { MailIngestSettingsService } from './mail-ingest-settings.service';
 import { Injectable } from '@nestjs/common';
 import type { MailIngestInboundResponse } from '@trek/shared';
@@ -45,6 +46,7 @@ export class MailIngestWebhookService {
   constructor(
     private readonly settings: MailIngestSettingsService,
     private readonly intake: MailIngestIntakeService,
+    private readonly process: MailIngestProcessJob,
   ) {}
 
   /**
@@ -92,6 +94,8 @@ export class MailIngestWebhookService {
       const [status, error] = REFUSALS[result.reason];
       throw new DomainError(status, error);
     }
+    // Read it now, not at the next processing tick.
+    if (result.status === 'accepted') this.process.runSoon();
     return { accepted: true, duplicate: result.status === 'duplicate' };
   }
 }

server/tests/unit/nest/mail-ingest/mail-ingest-transports.test.ts

@@ -6,7 +6,7 @@
  * failure the status card reads), the TREK Cloud relay (its page format,
  * paging, hash check, a mail it cannot read, a mail kept for now,
  * acknowledgement, idle without a relay or an import there) and the status
- * card and connection test.
+ * card and connection test. Whatever a transport takes in is read at once.
  */
 import { MailIngestCandidates } from '../../../../src/db/entities/MailIngestCandidates.entity';
 import { MailIngestCursor } from '../../../../src/db/entities/MailIngestCursor.entity';
@@ -19,6 +19,7 @@ import {
 } from '../../../../src/nest/mail-ingest/mail-ingest-imap.client';
 import { MailIngestImapJob } from '../../../../src/nest/mail-ingest/mail-ingest-imap.job';
 import { MailIngestIntakeService } from '../../../../src/nest/mail-ingest/mail-ingest-intake.service';
+import { MailIngestProcessJob } from '../../../../src/nest/mail-ingest/mail-ingest-process.job';
 import { MailIngestRelayJob } from '../../../../src/nest/mail-ingest/mail-ingest-relay.job';
 import { MailIngestRelayHost } from '../../../../src/nest/mail-ingest/mail-ingest-relay.service';
 import { MailIngestSettingsService } from '../../../../src/nest/mail-ingest/mail-ingest-settings.service';
@@ -72,6 +73,7 @@ const settings = {
   maxMailBytes: vi.fn(() => 1000),
 };
 const intake = { receive: vi.fn() };
+const processing = { runSoon: vi.fn() };
 const cursor = {
   acquireLease: vi.fn(async () => true),
   releaseLease: vi.fn(async () => undefined),
@@ -103,6 +105,7 @@ async function build() {
       { provide: MailIngestRelayHost, useValue: relays },
       { provide: MailIngestSettingsService, useValue: settings },
       { provide: MailIngestIntakeService, useValue: intake },
+      { provide: MailIngestProcessJob, useValue: processing },
       { provide: MailIngestImapClient, useValue: imap },
       { provide: CronRegistrarService, useValue: registrar },
       { provide: getRepositoryToken(MailIngestCursor), useValue: cursor },
@@ -155,10 +158,13 @@ describe('the signed webhook', () => {
     intake.receive.mockResolvedValue({ status: 'accepted' });
     expect(await service().receive(SECRET, RAW, signed(), NOW)).toEqual({ accepted: true, duplicate: false });
     expect(intake.receive).toHaveBeenCalledWith(RAW, 'webhook', ['trek+tok@example.com']);
+    expect(processing.runSoon).toHaveBeenCalledTimes(1);
     intake.receive.mockResolvedValue({ status: 'duplicate' });
     const noRecipient = signed({ recipient: '' });
     expect(await service().receive(SECRET, RAW, noRecipient, NOW)).toEqual({ accepted: true, duplicate: true });
     expect(intake.receive).toHaveBeenLastCalledWith(RAW, 'webhook', []);
+    // A duplicate is in already: nothing new to read.
+    expect(processing.runSoon).toHaveBeenCalledTimes(1);
   });
 
   it('refuses another secret, another recipient, a changed mail and a timestamp out of its window', async () => {
@@ -221,6 +227,8 @@ describe('the IMAP run', () => {
     expect(cursor.recordOk).toHaveBeenCalledWith('imap', 'INBOX', expect.any(String), { uidValidity: 9, lastUid: 42 });
     expect(session.close).toHaveBeenCalled();
     expect(cursor.releaseLease).toHaveBeenCalled();
+    // What came in is read at once, by one run for the whole box.
+    expect(processing.runSoon).toHaveBeenCalledTimes(1);
   });
 
   it('starts over when the server renumbered the box, and leaves a mail over the limit unread', async () => {
@@ -263,6 +271,7 @@ describe('the IMAP run', () => {
     expect(await job().tick(NOW)).toBe(0);
     expect(session.move).not.toHaveBeenCalled();
     expect(intake.receive).not.toHaveBeenCalled();
+    expect(processing.runSoon).not.toHaveBeenCalled();
     expect(cursor.recordOk).toHaveBeenCalledWith('imap', 'INBOX', expect.any(String), { uidValidity: 9, lastUid: 0 });
     settings.transport.mockResolvedValue('webhook');
     vi.clearAllMocks();
@@ -349,6 +358,7 @@ describe('the TREK Cloud relay', () => {
     // 2 did not arrive whole and stays with the relay; 3 is refused for good and dropped.
     expect(relay.ack.mock.calls).toEqual([[uuid(1)], [uuid(3)]]);
     expect(cursor.recordOk).toHaveBeenCalledWith('cloud', '', expect.any(String));
+    expect(processing.runSoon).toHaveBeenCalledTimes(1);
   });
 
   it('keeps a mail the import cannot take for now, and reads the mails after it', async () => {