diff --git a/server/src/nest/mail-ingest/mail-ingest-imap.job.ts b/server/src/nest/mail-ingest/mail-ingest-imap.job.ts index 3154d7341..f9622ca57 100644 --- a/server/src/nest/mail-ingest/mail-ingest-imap.job.ts +++ b/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); diff --git a/server/src/nest/mail-ingest/mail-ingest-process.job.ts b/server/src/nest/mail-ingest/mail-ingest-process.job.ts index 8e074f978..2206f2ee5 100644 --- a/server/src/nest/mail-ingest/mail-ingest-process.job.ts +++ b/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; diff --git a/server/src/nest/mail-ingest/mail-ingest-relay.job.ts b/server/src/nest/mail-ingest/mail-ingest-relay.job.ts index 849be10d6..88d35892f 100644 --- a/server/src/nest/mail-ingest/mail-ingest-relay.job.ts +++ b/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); diff --git a/server/src/nest/mail-ingest/mail-ingest-webhook.service.ts b/server/src/nest/mail-ingest/mail-ingest-webhook.service.ts index 49537a2d5..da008a4c7 100644 --- a/server/src/nest/mail-ingest/mail-ingest-webhook.service.ts +++ b/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' }; } } diff --git a/server/tests/unit/nest/mail-ingest/mail-ingest-transports.test.ts b/server/tests/unit/nest/mail-ingest/mail-ingest-transports.test.ts index c1f46b508..34f282e3d 100644 --- a/server/tests/unit/nest/mail-ingest/mail-ingest-transports.test.ts +++ b/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 () => {