From c648a5a6da2c07023cdc69909f2d5fe981d084c5 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 21 Aug 2026 18:04:35 -0400 Subject: [PATCH] Track running Granthi review queue jobs --- src/lib/db.js | 16 ++++++++++++++++ src/server.js | 7 +++++-- test/persistqueue.test.js | 15 ++++++++++++++- 3 files changed, 35 insertions(+), 3 deletions(-) diff --git a/src/lib/db.js b/src/lib/db.js index 0133cc2..84ce77b 100644 --- a/src/lib/db.js +++ b/src/lib/db.js @@ -35,6 +35,7 @@ const DDL = [ state TEXT NOT NULL DEFAULT 'queued', outcome TEXT, created_ts TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')), + started_ts TEXT, finished_ts TEXT )`, `CREATE INDEX IF NOT EXISTS idx_queue_jobs_state ON queue_jobs(state)` @@ -44,9 +45,17 @@ export function openDb(dbPath) { mkdirSync(dirname(dbPath), { recursive: true }); const db = new DatabaseSync(dbPath); for (const stmt of DDL) db.prepare(stmt).run(); + ensureColumn(db, 'queue_jobs', 'started_ts', 'TEXT'); return db; } +function ensureColumn(db, table, column, type) { + const columns = db.prepare(`PRAGMA table_info(${table})`).all(); + if (!columns.some((c) => c.name === column)) { + db.prepare(`ALTER TABLE ${table} ADD COLUMN ${column} ${type}`).run(); + } +} + export function insertReview(db, r) { const stmt = db.prepare(` INSERT INTO reviews (repo, sha, pr, event, pusher, authors, trailers, axes, @@ -76,6 +85,13 @@ export function enqueueJob(db, job) { return res.lastInsertRowid; } +export function startJob(db, id) { + db.prepare( + `UPDATE queue_jobs SET state = 'running', + started_ts = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE id = ?` + ).run(id); +} + export function finishJob(db, id, outcome = null) { db.prepare( `UPDATE queue_jobs SET state = 'done', outcome = ?, diff --git a/src/server.js b/src/server.js index 59b6d8f..cdddb9c 100644 --- a/src/server.js +++ b/src/server.js @@ -1,6 +1,6 @@ import { createServer } from 'node:http'; import { loadConfig } from './config.js'; -import { openDb, listReviews, enqueueJob, finishJob, recoverPendingJobs } from './lib/db.js'; +import { openDb, listReviews, enqueueJob, startJob, finishJob, recoverPendingJobs } from './lib/db.js'; import { GiteaClient } from './lib/gitea.js'; import { ReviewQueue } from './lib/queue.js'; import { verifySignature, jobFromWebhook } from './lib/webhook.js'; @@ -34,7 +34,10 @@ const queue = new ReviewQueue(); // whatever the outcome (runReview itself posts a final or fail-open status). function dispatchJob(queueRowId, job) { return queue.enqueue(`${job.owner}/${job.repo}`, (withGlobal) => - runReview({ cfg, gitea, adminGitea, db, job, withGlobal, log }) + { + startJob(db, queueRowId); + return runReview({ cfg, gitea, adminGitea, db, job, withGlobal, log }); + } ).then( (r) => finishJob(db, queueRowId, r?.ok ? (r.verdict || 'ok') : `error:${String(r?.error || '').slice(0, 200)}`), (e) => { diff --git a/test/persistqueue.test.js b/test/persistqueue.test.js index 3b3d3f2..ed0ed77 100644 --- a/test/persistqueue.test.js +++ b/test/persistqueue.test.js @@ -1,6 +1,6 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { openDb, enqueueJob, finishJob, recoverPendingJobs } from '../src/lib/db.js'; +import { openDb, enqueueJob, startJob, finishJob, recoverPendingJobs } from '../src/lib/db.js'; const JOB = (over = {}) => ({ event: 'push', owner: 'nirpa', repo: 'demo', sha: 'a'.repeat(40), @@ -32,6 +32,19 @@ test('finished jobs are not recovered again', () => { assert.equal(recoverPendingJobs(db).length, 0); }); +test('running jobs are visible and recovered on startup', () => { + const db = openDb(':memory:'); + const id = enqueueJob(db, JOB({ sha: '4'.repeat(40) })); + startJob(db, id); + + const row = db.prepare(`SELECT state, started_ts FROM queue_jobs WHERE id = ?`).get(id); + assert.equal(row.state, 'running'); + assert.match(row.started_ts, /^\d{4}-\d{2}-\d{2}T/); + + const pending = recoverPendingJobs(db); + assert.deepEqual(pending.map(p => p.id), [id]); +}); + test('unparseable persisted job is marked done, not re-run forever', () => { const db = openDb(':memory:'); db.prepare(`INSERT INTO queue_jobs (repo, sha, job) VALUES ('a/b', 'c', '{broken')`).run();