chore: log stage-timing breakdown for slow tmp payloads

Author: shaunwarman

.env.defaults

@@ -139,6 +139,8 @@ SQLITE_PORT=3456
 SQLITE_RCLONE_ENABLED=false
 SQLITE_FTS5_ENABLED=false
 DATABASE_MAP_MAX_SIZE=500
+# log a stage-timing breakdown for 'tmp' payloads slower than this (ms)
+SLOW_TMP_PAYLOAD_MS=5000
 
 ################
 ## web server ##

.env.schema

@@ -163,6 +163,7 @@ SQLITE_PORT=
 SQLITE_RCLONE_ENABLED=
 SQLITE_FTS5_ENABLED=
 DATABASE_MAP_MAX_SIZE=
+SLOW_TMP_PAYLOAD_MS=
 
 ################
 ## web server ##

helpers/parse-payload.js

@@ -114,6 +114,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
@@ -649,12 +655,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
@@ -666,11 +680,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;
@@ -689,11 +706,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')
@@ -707,6 +736,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');
@@ -764,6 +795,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(
@@ -915,6 +948,8 @@ async function parsePayload(data, ws) {
                 }
               }
 
+              mark('rateLimit');
+
               // check that we have available space
               const storagePath = getPathToDatabase({
                 id: alias.id,
@@ -929,6 +964,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 &&
@@ -1226,6 +1263,8 @@ async function parsePayload(data, ws) {
                 });
               }
 
+              mark('sieve');
+
               // Use Sieve-determined folder and flags
               const targetFolder = sieveResult.folder || 'INBOX';
               const targetFlags = sieveResult.flags || [];
@@ -1302,6 +1341,8 @@ async function parsePayload(data, ws) {
                 }
               }
 
+              mark('directAppend');
+
               /*
               //
               // attempt to get in-memory password from IMAP servers
@@ -1393,6 +1434,8 @@ async function parsePayload(data, ws) {
               if (!session.db) {
                 const tmpDb = await getTemporaryDatabase.call(this, session);
 
+                mark('tmpDbOpen');
+
                 let err;
 
                 try {
@@ -1767,9 +1810,13 @@ async function parsePayload(data, ws) {
                   logger.fatal(err);
                 }
 
+                mark('tmpStore');
+
                 if (tmpDb && !this.temporaryDatabaseMap)
                   await closeDatabase(tmpDb);
 
+                mark('tmpDbClose');
+
                 // send user push notification
                 if (!err)
                   sendApn(this.client, alias.id, targetFolder || 'INBOX')
@@ -1916,18 +1963,51 @@ async function parsePayload(data, ws) {
                     recipient: session.user.username
                   });
                 }
+
+              mark('imip');
             } 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