From f14197e64832993bf24d32f17db7b34800b8e2ab Mon Sep 17 00:00:00 2001 From: Peter Hedenskog Date: Fri, 15 May 2026 11:51:25 +0200 Subject: [PATCH] Heal stale active rows when a test runner dies mid-run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a test runner is killed mid-job (docker restart, OOM kill, host reboot), nothing flipped its DB row off status='active'. /search kept showing a ghost active entry indefinitely, and /result rendered "Streaming logs · running" over the dead worker's frozen log output — because the queue still keeps the job in its active set until its periodic stall check moves it back to wait (~30 s default), and getState() can't distinguish a real worker from an orphaned lock. Operators saw two jobs apparently running on one test runner after a restart even though per-worker concurrency=1 was intact: one was real, the other was the abandoned ghost. Three fixes close the gap from different sides. A global:stalled listener mirrors the queue back to the DB the moment a stalled job goes back to wait. A 60-second reconcile pass picks up rows the listener missed because the server itself was also restarting when the stall fired. And /result/:id peeks at the job's lock key in Redis so an orphaned-active job renders as "stalled, retrying" immediately instead of waiting 30 seconds for the stall check to catch up. Stalled rows are mirrored as waiting rather than failed, so a successful retry doesn't leave a phantom failure in the 24 h health pill. Co-authored-by: Claude noreply@anthropic.com --- server/locales/en.json | 1 + server/src/database/index.js | 22 ++++++ server/src/routes/html/result.js | 39 +++++++++-- server/src/server.js | 112 ++++++++++++++++++++++++++++++- 4 files changed, 168 insertions(+), 6 deletions(-) diff --git a/server/locales/en.json b/server/locales/en.json index e75dcd84..519a1950 100644 --- a/server/locales/en.json +++ b/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", diff --git a/server/src/database/index.js b/server/src/database/index.js index 1ecb9a1b..9abad809 100644 --- a/server/src/database/index.js +++ b/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. */ diff --git a/server/src/routes/html/result.js b/server/src/routes/html/result.js index 35cf2ee7..e1474377 100644 --- a/server/src/routes/html/result.js +++ b/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, diff --git a/server/src/server.js b/server/src/server.js index c14e8e0f..3f086c75 100644 --- a/server/src/server.js +++ b/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.