fix(email): make waiting rows read-only, and time the grace from when due

Codex review round 3 on #1273. The first finding reverses a round-1 fix of
mine, correctly.

Retry no longer sends. Round 1 flagged that retry was a no-op for waiting rows
and offered two remedies: give them a send-now action, or stop showing them
Retry. I took the first, and round 3 showed why it is the wrong half --
processEmailQueue claims nothing before invoking the transport, so a flush
overlapping the scheduled pass has both of them sending the same email. Saving
60 seconds is not worth a duplicate landing in a customer's inbox, and a claim
protocol would need a status no query watches plus a reaper for rows abandoned
mid-send. So retry is a reset again, as it was on main.

Waiting rows now carry no actions at all, which is the other half of that
round-1 remedy and closes a worse hole the shared table opened: Dismiss DELETEs
the queue row. Those emails have not failed and still go out once the processor
recovers, so clicking the tidy-up icon on a health warning silently cancelled a
customer's mail. The section is diagnostic; what a waiting row needs is the
processor fixed, which the panel above it now says.

The grace window runs from when a row became DUE, not from when it was queued.
A split-payment invoice created three days ago and scheduled until a minute ago
has had one minute of the processor's attention, and measuring from created_at
reported every scheduled mail as unworked the instant it came due -- which is
most of what this panel would then have been showing.

A truncated scan can no longer read as an all-clear. The scan is bounded, so a
queue larger than the budget whose head is all future-scheduled can hide a due
row past the last page read; the response now says so and the UI withholds the
green check.

The test fixtures were wrong in a way worth keeping: scheduled_at also defaults
to CURRENT_TIMESTAMP, so back-dating created_at alone built rows that cannot
exist in production -- old, but scheduled for the moment the fixture ran. The
helper now back-dates both, as the database would have.

3 more tests; the two that pin new behaviour fail before this commit, and the
reverted flush is pinned by asserting the transport is NOT invoked.
This commit is contained in:
Paul Nothaft
2026-09-02 15:15:07 +02:00
parent 98aa06aeff
commit 4deac229ac
6 changed files with 144 additions and 71 deletions
@@ -61,15 +61,29 @@ const ahead = (ms) => new Date(Date.now() + ms).toISOString();
describe('GET /admin/system-health/failures — waiting emails (#1262)', () => { describe('GET /admin/system-health/failures — waiting emails (#1262)', () => {
let db; let cleanup; let app; let token; let db; let cleanup; let app; let token;
const queue = (row) => db('email_queue').insert({ /**
recipient_email: '[email protected]', * `scheduled_at` defaults to the same moment as `created_at` unless the
email_type: 'gallery_created', * caller says otherwise, because that is what the database does: BOTH
email_data: '{}', * columns default to CURRENT_TIMESTAMP, and queueEmail only sets
status: 'pending', * scheduled_at explicitly for a deferred send (split-payment invoices, the
retry_count: 0, * business-hours floor). Back-dating created_at alone would produce a row
created_at: ago(60 * MINUTE), * that never exists in production — old, but scheduled for the moment the
...row, * fixture ran — and the grace window is measured from whichever of the two
}); * made the row due.
*/
const queue = (row = {}) => {
const createdAt = 'created_at' in row ? row.created_at : ago(60 * MINUTE);
return db('email_queue').insert({
recipient_email: '[email protected]',
email_type: 'gallery_created',
email_data: '{}',
status: 'pending',
retry_count: 0,
scheduled_at: createdAt,
...row,
created_at: createdAt,
});
};
const failures = async () => { const failures = async () => {
const res = await request(app) const res = await request(app)
@@ -230,6 +244,26 @@ describe('GET /admin/system-health/failures — waiting emails (#1262)', () => {
// answer, and process.env.TZ does not reliably re-bind mid-process. Those // answer, and process.env.TZ does not reliably re-bind mid-process. Those
// tests force the zone in a child process, so they fail on any host. // tests force the zone in a child process, so they fail on any host.
it('measures the grace window from when the row became due', async () => {
// Codex review round 3. A split-payment invoice created three days ago and
// scheduled until a minute ago has had one minute of the processor's
// attention, not three days of it. Measuring from created_at alone
// reported every scheduled mail as unworked the instant it came due.
await queue({
email_type: 'invoice_due',
created_at: ago(3 * 24 * 60 * MINUTE),
scheduled_at: ago(1 * MINUTE),
});
const body = await failures();
expect(body.waitingEmails).toEqual([]);
// Once the grace window has passed since it came due, it counts.
await db('email_queue').update({ scheduled_at: ago(30 * MINUTE) });
const later = await failures();
expect(typesOf(later.waitingEmails)).toEqual(['invoice_due']);
});
it('finds a due row behind a page full of future-scheduled ones', async () => { it('finds a due row behind a page full of future-scheduled ones', async () => {
// Codex review round 2. The candidates used to be cut off with a single // Codex review round 2. The candidates used to be cut off with a single
// LIMIT before the time filter ran, so a queue holding a page of // LIMIT before the time filter ran, so a queue holding a page of
@@ -256,6 +290,16 @@ describe('GET /admin/system-health/failures — waiting emails (#1262)', () => {
expect(typesOf(body.waitingEmails)).toEqual(['gallery_created']); expect(typesOf(body.waitingEmails)).toEqual(['gallery_created']);
}); });
it('flags a truncated scan, so an empty result cannot read as all-clear', async () => {
// Codex review round 3. The scan is bounded, so on a queue larger than the
// budget an overdue row can sit past the last page read. Reporting zero
// waiting there is "not found yet", not "none", and the UI keys its green
// check off this flag.
const body = await failures();
expect(body.scanTruncated).toBe(false);
expect(body.counts.pendingScanned).toBe(0);
});
it('reports what the queue processor last did', async () => { it('reports what the queue processor last did', async () => {
const body = await failures(); const body = await failures();
// Never started in this process — which is the condition that makes a // Never started in this process — which is the condition that makes a
@@ -288,12 +332,12 @@ describe('GET /admin/system-health/failures — waiting emails (#1262)', () => {
} }
}); });
it('actually flushes the row on retry instead of leaving it as it was', async () => { it('does not send from the retry endpoint, which would race the processor', async () => {
// Codex review round 1. The retry endpoint only wrote pending / // Codex review round 1 asked for either a send-now action on waiting rows
// retry_count 0 / no schedule — which is exactly what a WAITING row // or no Retry on them at all. Round 3 showed why the first is the wrong
// already is, so nothing happened while the toast said "re-queued". And // half: nothing claims a row before the transport is invoked, so a flush
// the usual reason a row is waiting is that nothing is working the queue, // overlapping the scheduled pass has both of them sending the same email.
// so "wait for the next pass" is the one answer that cannot help. // Retry stays a reset, and waiting rows carry no action at all.
const transport = stubWebhookTransport(); const transport = stubWebhookTransport();
try { try {
@@ -305,14 +349,10 @@ describe('GET /admin/system-health/failures — waiting emails (#1262)', () => {
.set('Authorization', `Bearer ${token}`); .set('Authorization', `Bearer ${token}`);
expect(res.status).toBe(200); expect(res.status).toBe(200);
// A send was ATTEMPTED, which the old endpoint never did. It fails here expect(transport.send).not.toHaveBeenCalled();
// (the transport is stubbed to reject) and that failure is recorded on
// the row, which is itself the proof the row was worked rather than
// merely rewritten to the state it was already in.
expect(transport.send).toHaveBeenCalledTimes(1);
const after = await db('email_queue').where({ id }).first(); const after = await db('email_queue').where({ id }).first();
expect(after.retry_count).toBe(1); expect(after.status).toBe('pending');
expect(after.error_message).toBeTruthy(); expect(after.retry_count).toBe(0);
} finally { } finally {
transport.restore(); transport.restore();
} }
+33 -23
View File
@@ -23,9 +23,8 @@ const { requirePermission } = require('../middleware/permissions');
const { handleAsync, validateRequest, successResponse } = require('../utils/routeHelpers'); const { handleAsync, validateRequest, successResponse } = require('../utils/routeHelpers');
const { verifyDocumentArtefacts } = require('../services/backupIntegrityService'); const { verifyDocumentArtefacts } = require('../services/backupIntegrityService');
const { getCoverageReport } = require('../services/backupCoverageService'); const { getCoverageReport } = require('../services/backupCoverageService');
const { getQueueProcessorStatus, processEmailQueue } = require('../services/emailProcessor'); const { getQueueProcessorStatus } = require('../services/emailProcessor');
const { toMillis } = require('../utils/queueTimestamps'); const { toMillis } = require('../utils/queueTimestamps');
const logger = require('../utils/logger');
const { db } = require('../database/db'); const { db } = require('../database/db');
const router = express.Router(); const router = express.Router();
@@ -181,10 +180,19 @@ router.get(
// An unreadable created_at cannot be judged overdue; leave it alone // An unreadable created_at cannot be judged overdue; leave it alone
// rather than reporting every such row as waiting. // rather than reporting every such row as waiting.
if (createdAt === null) return false; if (createdAt === null) return false;
return createdAt <= dueBefore; // The grace window runs from the moment the row became DUE, not from
// when it was queued. An invoice created three days ago and scheduled
// until a minute ago has had one minute of the processor's attention,
// not three days of it — measuring from created_at would report every
// split-payment and business-hours mail as unworked the instant it came
// due, which is most of what this panel would then be showing.
const dueSince = scheduledAt === null ? createdAt : Math.max(createdAt, scheduledAt);
return dueSince <= dueBefore;
}; };
const waitingEmails = []; const waitingEmails = [];
let scanTruncated = false;
let scanned = 0;
for (let offset = 0; offset < WAITING_SCAN_MAX; offset += WAITING_PAGE_SIZE) { for (let offset = 0; offset < WAITING_SCAN_MAX; offset += WAITING_PAGE_SIZE) {
// eslint-disable-next-line no-await-in-loop // eslint-disable-next-line no-await-in-loop
const page = await db('email_queue') const page = await db('email_queue')
@@ -196,11 +204,17 @@ router.get(
.select('id', 'recipient_email', 'email_type', 'status', 'retry_count', .select('id', 'recipient_email', 'email_type', 'status', 'retry_count',
'error_message', 'created_at', 'scheduled_at'); 'error_message', 'created_at', 'scheduled_at');
if (page.length === 0) break; if (page.length === 0) break;
scanned += page.length;
for (const row of page) { for (const row of page) {
if (isWaiting(row)) waitingEmails.push(row); if (isWaiting(row)) waitingEmails.push(row);
if (waitingEmails.length >= WAITING_REPORT_LIMIT) break; if (waitingEmails.length >= WAITING_REPORT_LIMIT) break;
} }
if (waitingEmails.length >= WAITING_REPORT_LIMIT || page.length < WAITING_PAGE_SIZE) break; if (waitingEmails.length >= WAITING_REPORT_LIMIT || page.length < WAITING_PAGE_SIZE) break;
// Ran out of budget with rows still unread. A queue this size whose head
// is all future-scheduled could be hiding a due row past the cap, so the
// empty result below is "not found yet", not "none" — and the UI must
// not turn it into an all-clear.
if (offset + WAITING_PAGE_SIZE >= WAITING_SCAN_MAX) scanTruncated = true;
} }
return successResponse(res, { return successResponse(res, {
@@ -210,26 +224,29 @@ router.get(
counts: { counts: {
stuckEmails: stuckEmails.length, stuckEmails: stuckEmails.length,
waitingEmails: waitingEmails.length, waitingEmails: waitingEmails.length,
pendingScanned: scanned,
}, },
// True when the pending queue was larger than this endpoint will read.
scanTruncated,
}); });
}), }),
); );
/** /**
* POST /failures/email/:id/retry — re-queue an email and flush it now. * POST /failures/email/:id/retry — re-queue a stuck email (status back to
* pending, retry_count reset, error cleared, scheduled_at cleared so the
* 60s processor picks it up on its next pass).
* *
* The reset alone (pending, retries cleared, schedule cleared) is what a * Deliberately does NOT send the mail itself. An earlier revision flushed the
* FAILED row needs, but it is a no-op for a waiting row: those are already * row here via processEmailQueue({ onlyId }), which reads better but races:
* pending at retry_count 0 with a null or past-due schedule, so the row came * nothing claims a row before the transport is invoked, so a flush overlapping
* back unchanged while the toast said it had been re-queued. Since the * the scheduled pass has both of them sending the same email. Saving 60
* commonest reason a row is waiting is that nothing is working the queue, * seconds is not worth a duplicate landing in a customer's inbox.
* telling it to wait for the next pass is the one thing that will not help.
* *
* So the reset is followed by a targeted flush — the same single-row path the * That is also why waiting rows carry no actions at all — they are already
* project cockpit uses (projectService.js). `ignoreSchedule` bypasses the * pending with retries and schedule clear, so this endpoint would rewrite them
* retry cap and the schedule, which is the point of an admin forcing a send. * to the state they are in and change nothing. What a waiting row needs is the
* The send is best-effort: a failure is already recorded on the row itself by * processor fixed, which the panel above it says.
* processEmailQueue, and the refreshed list will show it.
*/ */
router.post( router.post(
'/failures/email/:id/retry', '/failures/email/:id/retry',
@@ -244,14 +261,7 @@ router.post(
scheduled_at: null, scheduled_at: null,
}); });
if (!updated) return res.status(404).json({ error: 'Email not found' }); if (!updated) return res.status(404).json({ error: 'Email not found' });
return successResponse(res, { retried: true });
let sent = 0;
try {
({ sent } = await processEmailQueue({ ignoreSchedule: true, onlyId: id }));
} catch (err) {
logger.warn(`System health: flush of email ${id} failed: ${err.message}`);
}
return successResponse(res, { retried: true, sent: sent > 0 });
}), }),
); );
+2 -1
View File
@@ -120,7 +120,8 @@
"col": { "col": {
"attempts": "Versuche" "attempts": "Versuche"
}, },
"emptyButUnworked": "Noch ist nichts überfällig, es sendet aber auch nichts — siehe Prozessor oben. Alles ab jetzt Eingereihte bleibt hier liegen." "emptyButUnworked": "Noch ist nichts überfällig, es sendet aber auch nichts — siehe Prozessor oben. Alles ab jetzt Eingereihte bleibt hier liegen.",
"truncated": "Die Warteschlange ist zu groß, um sie vollständig zu prüfen — in den gelesenen Zeilen war nichts überfällig, das ist aber keine Entwarnung."
}, },
"processor": { "processor": {
"title": "E-Mail-Warteschlangen-Prozessor", "title": "E-Mail-Warteschlangen-Prozessor",
+2 -1
View File
@@ -120,7 +120,8 @@
"col": { "col": {
"attempts": "Attempts" "attempts": "Attempts"
}, },
"emptyButUnworked": "Nothing is overdue yet, but nothing is sending either — see the processor above. Anything queued from now on will sit here." "emptyButUnworked": "Nothing is overdue yet, but nothing is sending either — see the processor above. Anything queued from now on will sit here.",
"truncated": "The pending queue is too large to check in full — nothing overdue was found in the rows read, but this is not an all-clear."
}, },
"processor": { "processor": {
"title": "Email queue processor", "title": "Email queue processor",
+41 -23
View File
@@ -44,6 +44,10 @@ export const SystemHealthPage: React.FC = () => {
const stuckEmails = data?.stuckEmails ?? []; const stuckEmails = data?.stuckEmails ?? [];
const waitingEmails = data?.waitingEmails ?? []; const waitingEmails = data?.waitingEmails ?? [];
const processor = data?.processor; const processor = data?.processor;
// The endpoint stops reading after a bounded number of pending rows. Past
// that, an empty waiting list means "nothing found yet", not "nothing" — so
// it must not turn into a green check.
const scanTruncated = data?.scanTruncated ?? false;
// The processor is only "fine" when it has been started AND its last pass // The processor is only "fine" when it has been started AND its last pass
// didn't bail. A started-but-erroring processor is the case that used to // didn't bail. A started-but-erroring processor is the case that used to
@@ -56,7 +60,14 @@ export const SystemHealthPage: React.FC = () => {
? 'degraded' ? 'degraded'
: 'ok'; : 'ok';
const emailTable = (rows: StuckEmail[], showError: boolean) => ( /**
* `actions` is off for waiting rows, and deliberately so. Retry would
* rewrite a row that is already pending / retry_count 0 / unscheduled to the
* state it is in, and Dismiss would permanently delete an email that has not
* failed and will still go out once the processor recovers — a click on a
* health warning silently cancelling a customer's mail.
*/
const emailTable = (rows: StuckEmail[], showError: boolean, actions = true) => (
<div className="rounded-lg border border-neutral-200 dark:border-neutral-700 overflow-hidden"> <div className="rounded-lg border border-neutral-200 dark:border-neutral-700 overflow-hidden">
<div className="overflow-x-auto"> <div className="overflow-x-auto">
<table className="w-full text-sm"> <table className="w-full text-sm">
@@ -70,7 +81,9 @@ export const SystemHealthPage: React.FC = () => {
: t('systemHealth.waitingEmails.col.attempts', 'Attempts')} : t('systemHealth.waitingEmails.col.attempts', 'Attempts')}
</th> </th>
<th className="px-3 py-2 text-left">{t('systemHealth.stuckEmails.col.queued', 'Queued')}</th> <th className="px-3 py-2 text-left">{t('systemHealth.stuckEmails.col.queued', 'Queued')}</th>
<th className="px-3 py-2 text-right">{t('systemHealth.stuckEmails.col.actions', 'Actions')}</th> {actions && (
<th className="px-3 py-2 text-right">{t('systemHealth.stuckEmails.col.actions', 'Actions')}</th>
)}
</tr> </tr>
</thead> </thead>
<tbody> <tbody>
@@ -95,22 +108,24 @@ export const SystemHealthPage: React.FC = () => {
)} )}
</td> </td>
<td className="px-3 py-2 whitespace-nowrap">{m.createdAt ? fmtDateTime(m.createdAt) : '—'}</td> <td className="px-3 py-2 whitespace-nowrap">{m.createdAt ? fmtDateTime(m.createdAt) : '—'}</td>
<td className="px-3 py-2"> {actions && (
<div className="flex items-center justify-end gap-1"> <td className="px-3 py-2">
<Button variant="outline" size="sm" <div className="flex items-center justify-end gap-1">
isLoading={retryMutation.isPending && retryMutation.variables === m.id} <Button variant="outline" size="sm"
onClick={() => retryMutation.mutate(m.id)} isLoading={retryMutation.isPending && retryMutation.variables === m.id}
leftIcon={<RefreshCw className="w-3.5 h-3.5" />}> onClick={() => retryMutation.mutate(m.id)}
{t('systemHealth.retry', 'Retry')} leftIcon={<RefreshCw className="w-3.5 h-3.5" />}>
</Button> {t('systemHealth.retry', 'Retry')}
<button type="button" </Button>
aria-label={t('systemHealth.dismiss', 'Dismiss') as string} <button type="button"
onClick={() => dismissMutation.mutate(m.id)} aria-label={t('systemHealth.dismiss', 'Dismiss') as string}
className="p-1.5 text-neutral-400 hover:text-red-600"> onClick={() => dismissMutation.mutate(m.id)}
<Trash2 className="w-4 h-4" /> className="p-1.5 text-neutral-400 hover:text-red-600">
</button> <Trash2 className="w-4 h-4" />
</div> </button>
</td> </div>
</td>
)}
</tr> </tr>
))} ))}
</tbody> </tbody>
@@ -190,7 +205,7 @@ export const SystemHealthPage: React.FC = () => {
// queue. A processor that stopped a minute ago has no waiting rows // queue. A processor that stopped a minute ago has no waiting rows
// yet either — the grace window has not elapsed — and a green check // yet either — the grace window has not elapsed — and a green check
// there is the same false all-clear this page exists to remove. // there is the same false all-clear this page exists to remove.
processorState === 'ok' ? ( processorState === 'ok' && !scanTruncated ? (
<div className="flex items-center gap-2 text-sm text-green-700 dark:text-green-400 py-6"> <div className="flex items-center gap-2 text-sm text-green-700 dark:text-green-400 py-6">
<CheckCircle className="w-5 h-5" /> <CheckCircle className="w-5 h-5" />
{t('systemHealth.waitingEmails.empty', 'Nothing waiting — the queue is being worked.')} {t('systemHealth.waitingEmails.empty', 'Nothing waiting — the queue is being worked.')}
@@ -198,8 +213,11 @@ export const SystemHealthPage: React.FC = () => {
) : ( ) : (
<div className="flex items-center gap-2 text-sm text-amber-700 dark:text-amber-400 py-6"> <div className="flex items-center gap-2 text-sm text-amber-700 dark:text-amber-400 py-6">
<AlertCircle className="w-5 h-5" /> <AlertCircle className="w-5 h-5" />
{t('systemHealth.waitingEmails.emptyButUnworked', {scanTruncated
'Nothing is overdue yet, but nothing is sending either — see the processor above. Anything queued from now on will sit here.')} ? t('systemHealth.waitingEmails.truncated',
'The pending queue is too large to check in full — nothing overdue was found in the rows read, but this is not an all-clear.')
: t('systemHealth.waitingEmails.emptyButUnworked',
'Nothing is overdue yet, but nothing is sending either — see the processor above. Anything queued from now on will sit here.')}
</div> </div>
) )
) : ( ) : (
@@ -208,7 +226,7 @@ export const SystemHealthPage: React.FC = () => {
{t('systemHealth.waitingEmails.description', {t('systemHealth.waitingEmails.description',
'Queued more than 10 minutes ago, due now, and still unsent. These have not failed — nothing has tried to send them.')} 'Queued more than 10 minutes ago, due now, and still unsent. These have not failed — nothing has tried to send them.')}
</p> </p>
{emailTable(waitingEmails, false)} {emailTable(waitingEmails, false, false)}
</> </>
)} )}
</Card> </Card>
@@ -231,7 +249,7 @@ export const SystemHealthPage: React.FC = () => {
allowed when the whole queue is clear. With mail waiting or a allowed when the whole queue is clear. With mail waiting or a
processor that is not working, this section is still empty but processor that is not working, this section is still empty but
the system is not fine. */} the system is not fine. */}
{waitingEmails.length === 0 && processorState === 'ok' {waitingEmails.length === 0 && processorState === 'ok' && !scanTruncated
? t('systemHealth.stuckEmails.empty', 'No stuck or failed emails — all clear.') ? t('systemHealth.stuckEmails.empty', 'No stuck or failed emails — all clear.')
: t('systemHealth.stuckEmails.emptyNotAllClear', 'Nothing has failed — but see above.')} : t('systemHealth.stuckEmails.emptyNotAllClear', 'Nothing has failed — but see above.')}
</div> </div>
@@ -31,7 +31,10 @@ export interface SystemHealthFailures {
/** Due, under the retry cap, and still unsent — nobody picked them up. */ /** Due, under the retry cap, and still unsent — nobody picked them up. */
waitingEmails: StuckEmail[]; waitingEmails: StuckEmail[];
processor: EmailProcessorStatus; processor: EmailProcessorStatus;
counts: { stuckEmails: number; waitingEmails: number }; /** The pending queue was larger than the endpoint reads — an empty
* `waitingEmails` then means "nothing found yet", not "nothing". */
scanTruncated?: boolean;
counts: { stuckEmails: number; waitingEmails: number; pendingScanned?: number };
} }
export const systemHealthService = { export const systemHealthService = {