Skip to content

chore: tti and parse payload debug metrics #539

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shaunwarman wants to merge 2 commits into
base: master
Choose a base branch
from
Open

chore: tti and parse payload debug metrics #539

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
83bef13
fix: correct TTI probe false failures and record timeout vs error
shaunwarman Jul 7, 2026
958d4fe
chore: log stage-timing breakdown for slow tmp payloads
shaunwarman Jul 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .env.defaults
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 ##
Expand Down
1 change: 1 addition & 0 deletions .env.schema
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 ##
Expand Down
18 changes: 18 additions & 0 deletions app/models/tti.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
}
]
Expand Down
33 changes: 25 additions & 8 deletions helpers/get-message.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 () => {
Expand All @@ -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
Expand All @@ -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;
119 changes: 118 additions & 1 deletion helpers/parse-payload.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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;
Expand All @@ -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')
Expand All @@ -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');
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -896,6 +929,8 @@ async function parsePayload(data, ws) {
}
}

mark('rateLimit');

// check that we have available space
const storagePath = getPathToDatabase({
id: alias.id,
Expand All @@ -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 &&
Expand Down Expand Up @@ -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 || [];
Expand Down Expand Up @@ -1284,6 +1323,7 @@ async function parsePayload(data, ws) {
}
}

mark('directAppend');
//
// sqlite_auth_request disabled — auth is handled directly
//
Expand Down Expand Up @@ -1406,6 +1446,8 @@ async function parsePayload(data, ws) {

const tmpDb = await getTemporaryDatabase.call(this, session);

mark('tmpDbOpen');

let err;

try {
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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', {
Expand All @@ -1905,18 +1987,53 @@ 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);
logger.error(err);
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
Expand Down
Loading
Loading