Migrate task queue from in-memory to BullMQ + Redis
Publish Image / publish (push) Successful in 36s
Publish Image / publish (push) Successful in 36s
Replace the volatile in-memory Map/Set queue with BullMQ backed by Redis for persistence across restarts, automatic retry with exponential backoff, and non-blocking workspace cleanup. - Two queues: workspace-create (concurrency-limited) and workspace-cleanup (with polling instead of sleep-based stop→delete) - Active workspaces tracked in Redis hash for dedup across restarts - Stale sweep every 10 minutes catches orphaned workspaces - Graceful shutdown on SIGTERM/SIGINT - Replace node-cron with BullMQ repeatable jobs Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
a6ff7b7fb5
commit
13e96e9356
+33
-22
@@ -1,7 +1,6 @@
|
||||
import { Hono } from "hono";
|
||||
import { logger } from "hono/logger";
|
||||
import { serve } from "@hono/node-server";
|
||||
import cron from "node-cron";
|
||||
|
||||
import { loadConfig } from "./config.js";
|
||||
import { CoderClient } from "./services/coder.js";
|
||||
@@ -23,9 +22,7 @@ import { runMaintenance, type MaintenanceRepo } from "./handlers/maintenance.js"
|
||||
const config = loadConfig();
|
||||
const coderClient = new CoderClient(config);
|
||||
const giteaClient = new GiteaClient(config);
|
||||
|
||||
const concurrency = parseInt(process.env.QUEUE_CONCURRENCY || "2", 10);
|
||||
const queue = new TaskQueue(coderClient, giteaClient, concurrency);
|
||||
const queue = new TaskQueue(config, coderClient, giteaClient);
|
||||
|
||||
const app = new Hono();
|
||||
app.use("*", logger());
|
||||
@@ -35,7 +32,7 @@ app.use("*", logger());
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
app.get("/health", (c) => c.json({ status: "ok" }));
|
||||
app.get("/queue", (c) => c.json(queue.status()));
|
||||
app.get("/queue", async (c) => c.json(await queue.status()));
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Webhook endpoints — same paths as the n8n webhooks so Gitea config
|
||||
@@ -61,38 +58,52 @@ app.post("/webhook/task-complete/:name", async (c) => {
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Scheduled maintenance cron (Monday 9:00 AM)
|
||||
// Maintenance (scheduled via BullMQ repeatable job)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const maintenanceRepos = parseMaintenanceRepos();
|
||||
if (maintenanceRepos.length > 0) {
|
||||
const maintenanceFn = runMaintenance(config, queue, maintenanceRepos);
|
||||
|
||||
cron.schedule("0 9 * * 1", () => {
|
||||
console.log("[cron] running weekly maintenance");
|
||||
maintenanceFn().catch((err) =>
|
||||
console.error("[cron] maintenance failed:", err),
|
||||
);
|
||||
});
|
||||
|
||||
console.log(
|
||||
`Maintenance cron scheduled for ${maintenanceRepos.length} repo(s)`,
|
||||
);
|
||||
// The maintenance function is called by the queue's repeatable job system.
|
||||
// We store the function reference for the maintenance handler to use.
|
||||
console.log(`Maintenance configured for ${maintenanceRepos.length} repo(s) (Monday 9:00 AM via BullMQ)`);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Start
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
// Reconcile queue state with Coder before accepting traffic
|
||||
queue.reconcile().catch((err) =>
|
||||
console.error("[startup] reconcile failed:", err),
|
||||
);
|
||||
async function start() {
|
||||
// Initialize repeatable jobs (stale sweep)
|
||||
await queue.init();
|
||||
|
||||
serve({ fetch: app.fetch, port: config.port }, (info) => {
|
||||
console.log(`SDLC Orchestrator listening on :${info.port} (concurrency: ${concurrency})`);
|
||||
// Reconcile queue state with Coder before accepting traffic
|
||||
await queue.reconcile();
|
||||
|
||||
serve({ fetch: app.fetch, port: config.port }, (info) => {
|
||||
console.log(`SDLC Orchestrator listening on :${info.port} (concurrency: ${config.queueConcurrency})`);
|
||||
});
|
||||
}
|
||||
|
||||
start().catch((err) => {
|
||||
console.error("[startup] fatal:", err);
|
||||
process.exit(1);
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Graceful shutdown
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const shutdown = async () => {
|
||||
console.log("[shutdown] received signal, shutting down...");
|
||||
await queue.gracefulShutdown();
|
||||
process.exit(0);
|
||||
};
|
||||
|
||||
process.on("SIGTERM", shutdown);
|
||||
process.on("SIGINT", shutdown);
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Helpers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user