Heal stale active rows when a test runner dies mid-run

server/locales/en.json

@@ -30,6 +30,7 @@
   "index.waitingtestrunners": "Waiting for test runners to come online.",
   "index.refresh": "Refresh",
   "index.waitingtorunnext": "Waiting to run next.",
+  "index.stalledretrying": "The previous test runner stopped before this test finished. The queue will hand it to the next available worker for a retry.",
   "index.inqueue": "There are %s jobs in the queue and your job is placed %s in the queue",
   "search.button.lasthour": "Last hour",
   "search.button.today": "Today",

server/src/database/index.js

@@ -87,6 +87,28 @@ export async function updateStatus(id, status, reason) {
   }
 }
 
+/**
+ * IDs of tests stuck at status='active' for longer than the given
+ * grace window. Used by the periodic reconcile pass to find rows that
+ * a dead testrunner left behind — a healthy worker updates the row
+ * within seconds, so anything still 'active' minutes later is a
+ * candidate to re-check against Bull's actual state.
+ */
+export async function getStaleActiveTestIds(graceMinutes) {
+  const select = `SELECT id FROM sitespeed_io_test_runs
+                   WHERE status = 'active'
+                     AND added_date < NOW() - ($1 || ' minutes')::interval`;
+  try {
+    const result = await DatabaseHelper.getInstance().query(select, [
+      String(graceMinutes)
+    ]);
+    return result.rows.map(row => row.id);
+  } catch (error) {
+    logError('Could not list stale active tests', error);
+    return [];
+  }
+}
+
 /**
  * Get the latests tests.
  */

server/src/routes/html/result.js

@@ -57,15 +57,44 @@ result.get('/:id', async function (request, response) {
             }
           }
         }
+
+        // Bull keeps a job in the 'active' set until its periodic
+        // stalled-jobs check moves it back to 'wait' — up to
+        // ~stalledInterval (default 30 s) after the worker dies.
+        // During that window getState() still says 'active' even
+        // though no live worker is touching it, and running.pug would
+        // happily render "Streaming logs · running" over stale log
+        // lines from the dead worker. Cheap inline fix: peek at the
+        // job's lock key in Redis. No lock owner means the worker
+        // has died and Bull hasn't caught up yet — render the
+        // stalled message instead.
+        let stalled = false;
+        if (status === 'active') {
+          try {
+            const lockOwner = await workQueue.client.get(job.lockKey());
+            if (!lockOwner) {
+              stalled = true;
+            }
+          } catch (error) {
+            logger.error('Lock probe failed for %s: %s', id, error.message);
+          }
+        }
+
+        let message;
+        if (stalled) {
+          message = getText('index.stalledretrying');
+        } else if (count > 1) {
+          message = getText('index.inqueue', count, placeInQueue);
+        } else {
+          message = getText('index.waitingtorunnext');
+        }
+
         response.header('Cache-Control', 'no-cache, no-store, must-revalidate');
         response.header('Pragma', 'no-cache');
         response.header('Expires', 0);
         return response.render('running', {
-          status: status,
-          message:
-            count > 1
-              ? getText('index.inqueue', count, placeInQueue)
-              : getText('index.waitingtorunnext'),
+          status: stalled ? 'stalled' : status,
+          message,
           id: id,
           url: testConfig.url,
           nconf,

server/src/server.js

@@ -7,6 +7,7 @@ import {
   onMessage,
   processJob,
   getExistingQueue,
+  getExistingQueueNames,
   addDeviceToQueue
 } from './queuehandler.js';
 import {
@@ -19,7 +20,8 @@ import {
   testConnection,
   updateStatus,
   updateTest,
-  getTest
+  getTest,
+  getStaleActiveTestIds
 } from './database/index.js';
 import DatabaseHelper from './database/databasehelper.js';
 import { testsCompletedTotal, testsFailedTotal } from './metrics.js';
@@ -37,6 +39,95 @@ async function setFailedStatus(jobid, error) {
   return updateStatus(jobid, 'failed', error);
 }
 
+async function setStalledStatus(jobid) {
+  // Bull moved this job from active back to wait because its lock
+  // expired (testrunner died mid-run). Mirror that in the DB so
+  // /search and /result stop showing the row as 'active' while the
+  // retry is queued — Bull will retry, and the next global:active /
+  // global:failed will write the final state. We do *not* mark it
+  // 'failed' here: a successful retry shouldn't leave a phantom
+  // failure in the 24h count.
+  return updateStatus(jobid, 'waiting');
+}
+
+// Cadence for the stale-active reconcile pass. Runs every 60 s, only
+// touching rows older than RECONCILE_GRACE_MINUTES so we never race a
+// freshly-active job whose `setActiveStatus` event is in flight.
+const RECONCILE_INTERVAL_MS = 60_000;
+const RECONCILE_GRACE_MINUTES = 5;
+
+// Catches stale active rows the global:stalled handler missed —
+// typically because the server was down when Bull fired the event,
+// so no listener was attached. Walks every `status='active'` row
+// older than the grace window, asks Bull what state the job is
+// actually in, and rewrites the DB to match. If Bull no longer has
+// the job at all (evicted by removeOnFail/removeOnComplete after a
+// completion we never recorded), we mark the row failed with an
+// explicit reason so it stops haunting /search.
+async function reconcileStaleActiveRows() {
+  const queueNames = getExistingQueueNames();
+  if (queueNames.length === 0) return;
+
+  const staleIds = await getStaleActiveTestIds(RECONCILE_GRACE_MINUTES);
+  if (staleIds.length === 0) return;
+
+  for (const id of staleIds) {
+    let job;
+    for (const queueName of queueNames) {
+      const queue = getExistingQueue(queueName);
+      if (!queue) continue;
+      try {
+        const candidate = await queue.getJob(id);
+        if (candidate) {
+          job = candidate;
+          break;
+        }
+      } catch (error) {
+        logger.error(
+          'Reconcile: error querying queue %s for job %s: %s',
+          queueName,
+          id,
+          error.message
+        );
+      }
+    }
+
+    if (!job) {
+      logger.info(
+        'Reconcile: %s is no longer in the queue, marking failed',
+        id
+      );
+      await updateStatus(id, 'failed', 'Reconciled: job no longer in queue');
+      continue;
+    }
+
+    let state;
+    try {
+      state = await job.getState();
+    } catch (error) {
+      logger.error('Reconcile: getState for %s failed: %s', id, error.message);
+      continue;
+    }
+
+    if (state === 'active') continue;
+
+    logger.info('Reconcile: %s Bull state is %s, updating DB', id, state);
+    if (state === 'failed') {
+      await updateStatus(
+        id,
+        'failed',
+        job.failedReason || 'Reconciled: queue state was failed'
+      );
+    } else if (state === 'completed') {
+      await updateStatus(id, 'completed');
+    } else {
+      // waiting / delayed / paused — Bull will run it again, mirror
+      // its current resting state instead of inventing one.
+      await updateStatus(id, 'waiting');
+    }
+  }
+}
+
 function setupLogging() {
   const logVerbose = nconf.get('log:verbose');
   configureLog({ level: logVerbose ? 'verbose' : 'info' });
@@ -97,6 +188,7 @@ async function setupTestRunnerQueue() {
           if (!queue) {
             onMessage(queueName, 'global:active', setActiveStatus);
             onMessage(queueName, 'global:failed', setFailedStatus);
+            onMessage(queueName, 'global:stalled', setStalledStatus);
             addDeviceToQueue(
               setup.deviceId,
               job.data.serverConfig.name,
@@ -155,6 +247,19 @@ export class SitespeedioServer {
     // lie.
     this.pruneTimer = setInterval(pruneStaleTestRunners, 60_000);
     this.pruneTimer.unref();
+    // Reconcile DB rows that were left at status='active' by a
+    // testrunner that died before its global:stalled event reached
+    // us — typically because the server itself was restarting at
+    // the same time, so no listener was attached. The grace window
+    // inside the helper avoids racing freshly-active jobs.
+    this.reconcileTimer = setInterval(
+      () =>
+        reconcileStaleActiveRows().catch(error =>
+          logger.error('Reconcile pass failed: %s', error.message)
+        ),
+      RECONCILE_INTERVAL_MS
+    );
+    this.reconcileTimer.unref();
     // Tell the world that we are starting
     await publish('server', 'start');
   }
@@ -167,6 +272,11 @@ export class SitespeedioServer {
       this.pruneTimer = undefined;
     }
 
+    if (this.reconcileTimer) {
+      clearInterval(this.reconcileTimer);
+      this.reconcileTimer = undefined;
+    }
+
     // Stop accepting new HTTP requests and drain the in-flight ones before
     // we tear down the database pool — otherwise a request mid-query will
     // hit a closed pool and 500 just before the process exits.