fix(email): compare queue timestamps in JS, and make retry actually send
Codex review round 1 on #1273. One of the four is a real bug on every SQLite deployment. The waiting-row query compared `created_at` against a bound ISO string. On SQLite that column does not hold a string: queueEmail writes a JS Date and the native binding stores epoch ms, and SQLite orders INTEGER before TEXT regardless of value -- so the comparison was true for EVERY row. Mail queued a second ago read as ten minutes overdue, and a scheduled_at years in the future read as already due. Confirmed directly against sqlite3: a 2026 row matches `created_at <= '2020-01-01T00:00:00.000Z'`. Binding a Date instead is not the fix, since knex hands sqlite3 a Date the same way and jest's sandbox Dates stringify to "[object Object]" (CLAUDE.md). So the engine-safe half of the predicate stays in SQL and the two time comparisons move into JS behind a toMillis() that accepts all three shapes this column really has -- Date from Postgres, ms-number from SQLite, ISO string from fixtures and older rows. The scan is capped at 1000 pending rows ordered oldest-first; everything overdue sorts into that window, and the response was already capped at 200. The existing tests missed this because they store ISO strings, which is what CLAUDE.md prescribes for jest -- so the new ones store epoch ms, the production shape, and one mixes both in a single queue. Retry was a no-op for the rows it most needed to help. It wrote pending / retry_count 0 / no schedule, which is exactly what a waiting row already is: the row came back unchanged while the toast said it had been re-queued. And since the usual reason a row is waiting is that nothing is working the queue, deferring it to the next pass is the one answer that cannot help. It now follows the reset with the same single-row flush the project cockpit uses. An idle pass no longer inherits the previous pass's totals -- the no-pending early return skipped the lastResult assignment, so System Health kept attributing an old sent/failed count to a run that did nothing. "All clear" now means the whole queue is clear, which is what the PR claimed and the code did not do. An empty waiting list is only reassuring when something is working the queue: a processor stopped a minute ago has no overdue rows yet either, and a green check there is the same false all-clear this branch exists to remove. 7 more tests. The 5 that pin new behaviour fail before this commit; the SQLite ones fail in the way the bug predicts rather than erroring. Both new tests stub the webhook transport with a spy rather than pointing it at a dead port: real connection attempts left open handles that destabilised unrelated suites in the same jest worker.
This commit is contained in:
Paul Nothaft
@@ -25,6 +25,35 @@ const jwt = require('jsonwebtoken');
|
||||
|
||||
const { bootCrmDb, seedMinimal, buildRouteApp } = require('../integration/helpers/crmDb');
|
||||
|
||||
/**
|
||||
* Stand up a transport without touching a socket.
|
||||
*
|
||||
* processEmailQueue bails before it reads a single row when there is no
|
||||
* transport, so the two tests below cannot otherwise reach the code they are
|
||||
* about. Spying on the webhook transport is the cheapest way in — and it must
|
||||
* be a spy rather than a dead URL, because real connection attempts leave open
|
||||
* handles that destabilise unrelated suites in the same worker.
|
||||
*
|
||||
* `send` rejects: a failed delivery is a delivery ATTEMPT, which is exactly
|
||||
* what these two need to observe.
|
||||
*/
|
||||
function stubWebhookTransport() {
|
||||
const transport = require('../../src/services/emailWebhookTransport');
|
||||
const savedFrom = process.env.EMAIL_FROM;
|
||||
process.env.EMAIL_FROM = '[email protected]';
|
||||
const enabled = jest.spyOn(transport, 'isEnabled').mockReturnValue(true);
|
||||
const send = jest.spyOn(transport, 'send').mockRejectedValue(new Error('transport down'));
|
||||
return {
|
||||
send,
|
||||
restore() {
|
||||
enabled.mockRestore();
|
||||
send.mockRestore();
|
||||
if (savedFrom === undefined) delete process.env.EMAIL_FROM;
|
||||
else process.env.EMAIL_FROM = savedFrom;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const MINUTE = 60 * 1000;
|
||||
const ago = (ms) => new Date(Date.now() - ms).toISOString();
|
||||
const ahead = (ms) => new Date(Date.now() + ms).toISOString();
|
||||
@@ -133,6 +162,66 @@ describe('GET /admin/system-health/failures — waiting emails (#1262)', () => {
|
||||
expect(body.stuckEmails).toEqual([]);
|
||||
});
|
||||
|
||||
// --- cross-engine timestamps (Codex review round 1) -----------------------
|
||||
//
|
||||
// Every test above stores ISO strings, which is what CLAUDE.md prescribes
|
||||
// for jest + SQLite. Production SQLite does not: queueEmail writes a JS Date
|
||||
// and the native binding stores it as a ms-NUMBER. SQLite orders INTEGER
|
||||
// before TEXT whatever the values are, so the original WHERE clause -- a
|
||||
// ms-number column compared against a bound ISO string -- was true for every
|
||||
// row. Fresh mail read as ten minutes overdue and future schedules read as
|
||||
// due, on every SQLite deployment.
|
||||
describe('rows stored as epoch ms, the SQLite production shape', () => {
|
||||
const ms = (offset) => Date.now() + offset;
|
||||
|
||||
it('does not report a mail queued seconds ago as waiting', async () => {
|
||||
await queue({ email_type: 'gallery_created', created_at: ms(-30 * 1000) });
|
||||
|
||||
const body = await failures();
|
||||
expect(body.waitingEmails).toEqual([]);
|
||||
expect(body.counts.waitingEmails).toBe(0);
|
||||
});
|
||||
|
||||
it('still reports a genuinely overdue one', async () => {
|
||||
await queue({ email_type: 'gallery_created', created_at: ms(-52 * MINUTE) });
|
||||
|
||||
const body = await failures();
|
||||
expect(typesOf(body.waitingEmails)).toEqual(['gallery_created']);
|
||||
});
|
||||
|
||||
it('leaves a future numeric scheduled_at alone', async () => {
|
||||
await queue({
|
||||
email_type: 'invoice_due',
|
||||
created_at: ms(-52 * MINUTE),
|
||||
scheduled_at: ms(3 * 24 * 60 * MINUTE),
|
||||
});
|
||||
|
||||
const body = await failures();
|
||||
expect(body.waitingEmails).toEqual([]);
|
||||
});
|
||||
|
||||
it('reports a past-due numeric scheduled_at', async () => {
|
||||
await queue({
|
||||
email_type: 'invoice_due',
|
||||
created_at: ms(-52 * MINUTE),
|
||||
scheduled_at: ms(-30 * MINUTE),
|
||||
});
|
||||
|
||||
const body = await failures();
|
||||
expect(typesOf(body.waitingEmails)).toEqual(['invoice_due']);
|
||||
});
|
||||
|
||||
it('mixes both storage shapes in one queue without confusing them', async () => {
|
||||
await queue({ email_type: 'gallery_created', created_at: ms(-52 * MINUTE) });
|
||||
await queue({ email_type: 'quote_sent', created_at: ago(52 * MINUTE) });
|
||||
await queue({ email_type: 'invoice_due', created_at: ms(-30 * 1000) });
|
||||
await queue({ email_type: 'customer_invitation', created_at: ago(30 * 1000) });
|
||||
|
||||
const body = await failures();
|
||||
expect(typesOf(body.waitingEmails)).toEqual(['gallery_created', 'quote_sent']);
|
||||
});
|
||||
});
|
||||
|
||||
it('reports what the queue processor last did', async () => {
|
||||
const body = await failures();
|
||||
// Never started in this process — which is the condition that makes a
|
||||
@@ -142,6 +231,59 @@ describe('GET /admin/system-health/failures — waiting emails (#1262)', () => {
|
||||
expect(body.processor).toHaveProperty('lastError');
|
||||
});
|
||||
|
||||
it('does not attribute a previous pass\'s totals to an idle one', async () => {
|
||||
// Codex review round 1. The no-pending early return skipped the lastResult
|
||||
// assignment, so after one pass that sent or failed something, every idle
|
||||
// pass afterwards advanced lastRunAt while still reporting the old totals.
|
||||
const { processEmailQueue, getQueueProcessorStatus } = require('../../src/services/emailProcessor');
|
||||
const transport = stubWebhookTransport();
|
||||
|
||||
try {
|
||||
await queue({ email_type: 'gallery_created' });
|
||||
await processEmailQueue();
|
||||
const worked = getQueueProcessorStatus().lastResult;
|
||||
expect(worked.processed).toBe(1);
|
||||
expect(worked.sent + worked.failed).toBe(1);
|
||||
|
||||
await db('email_queue').del();
|
||||
await processEmailQueue();
|
||||
|
||||
expect(getQueueProcessorStatus().lastResult).toEqual({ processed: 0, sent: 0, failed: 0 });
|
||||
} finally {
|
||||
transport.restore();
|
||||
}
|
||||
});
|
||||
|
||||
it('actually flushes the row on retry instead of leaving it as it was', async () => {
|
||||
// Codex review round 1. The retry endpoint only wrote pending /
|
||||
// retry_count 0 / no schedule — which is exactly what a WAITING row
|
||||
// already is, so nothing happened while the toast said "re-queued". And
|
||||
// the usual reason a row is waiting is that nothing is working the queue,
|
||||
// so "wait for the next pass" is the one answer that cannot help.
|
||||
const transport = stubWebhookTransport();
|
||||
|
||||
try {
|
||||
await queue({ email_type: 'gallery_created' });
|
||||
const { id } = await db('email_queue').first('id');
|
||||
|
||||
const res = await request(app)
|
||||
.post(`/admin/system-health/failures/email/${id}/retry`)
|
||||
.set('Authorization', `Bearer ${token}`);
|
||||
expect(res.status).toBe(200);
|
||||
|
||||
// A send was ATTEMPTED, which the old endpoint never did. It fails here
|
||||
// (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();
|
||||
expect(after.retry_count).toBe(1);
|
||||
expect(after.error_message).toBeTruthy();
|
||||
} finally {
|
||||
transport.restore();
|
||||
}
|
||||
});
|
||||
|
||||
it('surfaces the transport failure that makes every pass a no-op', async () => {
|
||||
const { processEmailQueue, getQueueProcessorStatus } = require('../../src/services/emailProcessor');
|
||||
await queue({ email_type: 'gallery_created' });
|
||||
|
||||
Reference in New Issue
Block a user