From 83bef13c9fff0c9c57414ed68d8cd7f46b60912f Mon Sep 17 00:00:00 2001 From: shaunwarman Date: Mon, 6 Jul 2026 21:51:59 -0600 Subject: [PATCH 1/2] fix: correct TTI probe false failures and record timeout vs error --- app/models/tti.js | 18 +++++++++++++++++ helpers/get-message.js | 33 +++++++++++++++++++++++-------- jobs/tti.js | 45 +++++++++++++++++++++++++++++++++--------- 3 files changed, 79 insertions(+), 17 deletions(-) diff --git a/app/models/tti.js b/app/models/tti.js index 023359412..f40bc7ba7 100644 --- a/app/models/tti.js +++ b/app/models/tti.js @@ -35,6 +35,24 @@ const TTI = new mongoose.Schema({ type: Number, required: true, default: 0 + }, + // when *Ms is 0 these distinguish "did not arrive within 1m" + // (timeout) from IMAP connection/fetch failures (error) + directTimeout: { + type: Boolean, + default: false + }, + directError: { + type: Boolean, + default: false + }, + forwardingTimeout: { + type: Boolean, + default: false + }, + forwardingError: { + type: Boolean, + default: false } } ] diff --git a/helpers/get-message.js b/helpers/get-message.js index ddf19d1ae..42c49601f 100644 --- a/helpers/get-message.js +++ b/helpers/get-message.js @@ -9,6 +9,15 @@ const pWaitFor = require('p-wait-for'); async function getMessage(imapClient, info, provider) { let received; let err; + let timedOut = false; + + // + // NOTE: we match the full unique Message-ID (not just its domain) + // otherwise the concurrent Direct and Forward probes sent to the + // same inbox would match each other's message and skew timings + // + const messageId = info.messageId.replace('<', '').replace('>', ''); + try { await pWaitFor( async () => { @@ -20,17 +29,23 @@ async function getMessage(imapClient, info, provider) { // console.log('capabilities', imapClient.capabilities); try { - for await (const message of imapClient.fetch('*', { + // mailbox is empty (nothing has arrived yet) + if (imapClient.mailbox && imapClient.mailbox.exists === 0) + return false; + + // + // NOTE: we scan the entire mailbox (it is purged after every run) + // instead of only the newest message ('*'), otherwise any + // unrelated message arriving last would blind the probe + // for the full timeout and record a false failure + // + for await (const message of imapClient.fetch('1:*', { headers: ['Message-ID'] })) { if (received) continue; if ( message.headers && - message.headers - .toString() - .includes( - info.messageId.replace('<', '').replace('>', '').split('@')[1] - ) + message.headers.toString().includes(messageId) ) { // // NOTE: due to NTP time differences we cannot rely on @@ -50,15 +65,17 @@ async function getMessage(imapClient, info, provider) { return Boolean(received); }, { - interval: 0, + interval: ms('0.5s'), timeout: ms('1m') } ); } catch (_err) { err = _err; + // distinguish "message never arrived within 1m" from IMAP errors + timedOut = _err.name === 'TimeoutError'; } - return { provider, received, err }; + return { provider, received, err, timedOut }; } module.exports = getMessage; diff --git a/jobs/tti.js b/jobs/tti.js index b240ba71d..8d9405c0b 100644 --- a/jobs/tti.js +++ b/jobs/tti.js @@ -82,6 +82,12 @@ const IP_ADDRESS = ip.address(); const imapClients = new Map(); +function getFailureLabel(timedOut, errored) { + if (timedOut) return 'Timeout (>1m)'; + if (errored) return 'Error'; + return 'N/A'; +} + const graceful = new Graceful({ mongooses: [mongoose], redisClients: [client], @@ -165,6 +171,10 @@ async function checkTTI() { config.imapConfigurations.map(async (provider) => { let directMs = 0; let forwardingMs = 0; + let directTimeout = false; + let directError = false; + let forwardingTimeout = false; + let forwardingError = false; let imapClient; try { // https://github.com/postalsys/imapflow/blob/88e46d9bbcdc347d22df27bc591841431d8dc831/lib/imap-flow.js#L243-L247 @@ -337,7 +347,7 @@ Forward Email // rewrite messageId since `raw` overrides this info.messageId = messageId; - const { received, err } = await getMessage( + const { received, err, timedOut } = await getMessage( imapClient, info, provider @@ -361,18 +371,29 @@ Forward Email await logger.fatal(err); } - return _.isDate(received) - ? received.getTime() - date.getTime() - : 0; + return { + ms: _.isDate(received) + ? received.getTime() - date.getTime() + : 0, + timedOut: Boolean(timedOut), + errored: Boolean(err) && !timedOut + }; } ) ); - directMs = results[0]; - forwardingMs = results[1]; + directMs = results[0].ms; + directTimeout = results[0].timedOut; + directError = results[0].errored; + forwardingMs = results[1].ms; + forwardingTimeout = results[1].timedOut; + forwardingError = results[1].errored; } catch (err) { err.provider = provider; err.isCodeBug = true; logger.fatal(err); + // provider-level failure (e.g. IMAP connect) zeroes both legs + if (directMs === 0) directError = true; + if (forwardingMs === 0) forwardingError = true; } // delete all messages once done @@ -390,7 +411,11 @@ Forward Email return { name: provider.name, directMs, - forwardingMs + forwardingMs, + directTimeout, + directError, + forwardingTimeout, + forwardingError }; }) ); @@ -428,10 +453,12 @@ Forward Email .map( (p) => `${p.name}: Direct (${ - p.directMs === 0 ? 'N/A' : prettyMilliseconds(p.directMs) + p.directMs === 0 + ? getFailureLabel(p.directTimeout, p.directError) + : prettyMilliseconds(p.directMs) }) • Forwarding (${ p.forwardingMs === 0 - ? 'N/A' + ? getFailureLabel(p.forwardingTimeout, p.forwardingError) : prettyMilliseconds(p.forwardingMs) })` ) From 958d4fe88aabf34ee3ef4ba4e95a7e11d1dec38f Mon Sep 17 00:00:00 2001 From: shaunwarman Date: Mon, 6 Jul 2026 21:52:30 -0600 Subject: [PATCH 2/2] chore: log stage-timing breakdown for slow tmp payloads --- .env.defaults | 2 + .env.schema | 1 + helpers/parse-payload.js | 119 ++++++++++++++++++++++++++++++++++++++- 3 files changed, 121 insertions(+), 1 deletion(-) diff --git a/.env.defaults b/.env.defaults index deeb8c193..b7c999a31 100644 --- a/.env.defaults +++ b/.env.defaults @@ -141,6 +141,8 @@ SQLITE_FTS5_ENABLED=false DATABASE_MAP_MAX_SIZE=500 DATABASE_MAP_MAX_EVICTIONS_PER_SWEEP=50 SQLITE_DEBUG_TIMERS=true +# log a stage-timing breakdown for 'tmp' payloads slower than this (ms) +SLOW_TMP_PAYLOAD_MS=5000 ################ ## web server ## diff --git a/.env.schema b/.env.schema index a799bdb38..35332db1f 100644 --- a/.env.schema +++ b/.env.schema @@ -165,6 +165,7 @@ SQLITE_FTS5_ENABLED= DATABASE_MAP_MAX_SIZE= DATABASE_MAP_MAX_EVICTIONS_PER_SWEEP= SQLITE_DEBUG_TIMERS= +SLOW_TMP_PAYLOAD_MS= ################ ## web server ## diff --git a/helpers/parse-payload.js b/helpers/parse-payload.js index 6156ed7ba..910073054 100644 --- a/helpers/parse-payload.js +++ b/helpers/parse-payload.js @@ -115,6 +115,12 @@ const CHECKPOINTS = ['PASSIVE', 'FULL', 'RESTART', 'TRUNCATE']; const HOSTNAME = os.hostname(); const IP_ADDRESS = ip.address(); +// log a consolidated stage-timing breakdown for any 'tmp' payload +// whose queue-wait + handler time exceeds this threshold +const SLOW_TMP_PAYLOAD_MS = env.SLOW_TMP_PAYLOAD_MS + ? Number.parseInt(env.SLOW_TMP_PAYLOAD_MS, 10) + : ms('5s'); + const PAYLOAD_ACTIONS = new Set([ 'sync', // no db 'tmp', // no db @@ -630,12 +636,20 @@ async function parsePayload(data, ws) { // create the temporary message for each alias const errors = {}; + // stage timings for the consolidated slow-payload log below + const handlerStartedAt = Date.now(); + const queueWaitMs = payload.sent_at + ? Math.max(0, now - payload.sent_at) + : 0; + const aliasTimings = []; + // // rate limit the payload.remoteAddress from // sending more than 1 GB per day or 1000 messages per day // but attempt to use the reverse PTR root domain of the remoteAddress // let sender = payload.remoteAddress; + const reverseDnsStartedAt = Date.now(); try { const [clientHostname] = await this.resolver.reverse( payload.remoteAddress @@ -647,11 +661,14 @@ async function parsePayload(data, ws) { logger.warn(err); } + const reverseDnsMs = Date.now() - reverseDnsStartedAt; + const date = new Date().toISOString().split('T')[0]; // // parse headers from message // + const parseStartedAt = Date.now(); const splitter = new Splitter(); const joiner = new Joiner(); let headers; @@ -670,11 +687,23 @@ async function parsePayload(data, ws) { // (arguments = `session`, `headers`, `body`, `useSender`) // const fingerprint = getFingerprint({}, headers, payload.raw); + const parseMs = Date.now() - parseStartedAt; await pMap( payload.aliases, async (obj) => { + // per-alias stage timer: mark() records elapsed time since the + // previous mark under the given stage name (see slow-payload log) + const aliasStartedAt = Date.now(); + let stageStartedAt = aliasStartedAt; + const stages = {}; + const mark = (name) => { + const ts = Date.now(); + stages[name] = ts - stageStartedAt; + stageStartedAt = ts; + }; + try { const alias = await Aliases.findById(obj.id) .populate('domain', 'id name') @@ -688,6 +717,8 @@ async function parsePayload(data, ws) { .lean() .exec(); + mark('aliasLookup'); + if (!alias) throw new TypeError('Alias does not exist'); if (!alias.user) throw new TypeError('User does not exist'); @@ -745,6 +776,8 @@ async function parsePayload(data, ws) { const { isOverQuota, storageUsed, maxQuotaPerAlias } = await Aliases.isOverQuota(alias, 0, this.client); + mark('quota'); + if (isOverQuota) { const err = new Error( `${session.user.username} has exceeded quota with ${bytes( @@ -896,6 +929,8 @@ async function parsePayload(data, ws) { } } + mark('rateLimit'); + // check that we have available space const storagePath = getPathToDatabase({ id: alias.id, @@ -910,6 +945,8 @@ async function parsePayload(data, ws) { )} was available` ); + mark('diskSpace'); + // we should only use in-memory database is if was connected (IMAP session open) if ( this.databaseMap && @@ -1208,6 +1245,8 @@ async function parsePayload(data, ws) { }); } + mark('sieve'); + // Use Sieve-determined folder and flags const targetFolder = sieveResult.folder || 'INBOX'; const targetFlags = sieveResult.flags || []; @@ -1284,6 +1323,7 @@ async function parsePayload(data, ws) { } } + mark('directAppend'); // // sqlite_auth_request disabled — auth is handled directly // @@ -1406,6 +1446,8 @@ async function parsePayload(data, ws) { const tmpDb = await getTemporaryDatabase.call(this, session); + mark('tmpDbOpen'); + let err; try { @@ -1780,7 +1822,46 @@ async function parsePayload(data, ws) { logger.fatal(err); } - if (tmpDb) await closeDatabase(tmpDb); + mark('tmpStore'); + + if (tmpDb && !this.temporaryDatabaseMap) + await closeDatabase(tmpDb); + + mark('tmpDbClose'); + + // send user push notification + if (!err) + sendApn(this.client, alias.id, targetFolder || 'INBOX') + .then() + .catch((err) => + logger.fatal(err, { session, resolver: this.resolver }) + ); + + // send websocket push notification (enriched payload with eml) + if (!err) + sendWebSocketNotification( + this.client, + alias.id, + 'newMessage', + { + mailbox: targetFolder || 'INBOX', + message: { + folder_path: targetFolder || 'INBOX', + flags: targetFlags || [], + is_unread: !(targetFlags || []).includes('\\Seen'), + is_flagged: (targetFlags || []).includes('\\Flagged'), + is_deleted: (targetFlags || []).includes('\\Deleted'), + is_draft: (targetFlags || []).includes('\\Draft'), + is_encrypted: false, + eml: Buffer.isBuffer(messageRaw) + ? messageRaw.toString() + : typeof messageRaw === 'string' + ? messageRaw + : '', + object: 'message' + } + } + ); if (err) throw err; } @@ -1892,6 +1973,7 @@ async function parsePayload(data, ws) { remoteAddress: payload.remoteAddress }); } +<<<<<<< HEAD } catch (imipErr) { // Don't fail message delivery if iMIP processing fails logger.warn('iMIP processing failed', { @@ -1905,6 +1987,10 @@ async function parsePayload(data, ws) { recipient: session.user.username }); } +======= + + mark('imip'); +>>>>>>> d57fe2e9e (chore: log stage-timing breakdown for slow tmp payloads) } catch (err) { err.payload = _.omit(payload, 'raw'); err.isCodeBug = isCodeBug(err); @@ -1912,11 +1998,42 @@ async function parsePayload(data, ws) { errors[`${obj.address}`] = JSON.parse( safeStringify(parseErr(err)) ); + } finally { + aliasTimings.push({ + alias: obj.address, + aliasId: obj.id, + totalMs: Date.now() - aliasStartedAt, + stages + }); } }, { concurrency } ); + // + // one consolidated log line per slow tmp request so latency can be + // attributed to queue-wait vs a specific per-alias handler stage + // + const handlerMs = Date.now() - handlerStartedAt; + if (queueWaitMs + handlerMs >= SLOW_TMP_PAYLOAD_MS) { + logger.warn('slow tmp payload', { + // opt in to Mongo log persistence (warn-level logs are + // otherwise dropped by the hook in helpers/logger.js) + ignore_hook: false, + payloadId: payload.id, + queueWaitMs, + handlerMs, + endToEndMs: queueWaitMs + handlerMs, + reverseDnsMs, + parseMs, + byteLength, + aliasCount: payload.aliases.length, + aliasTimings, + remoteAddress: payload.remoteAddress, + hostname: HOSTNAME + }); + } + response = { id: payload.id, data: errors