trails-cool/trails
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add cron job to sweep stale import batches every minute

Browse source 
Batches stuck in pending or running for more than 10 minutes (server
restart mid-import, pg-boss job dropped) are marked failed with a
user-visible message. Runs every minute via pg-boss cron with a 55s
expiry so overlapping runs are dropped.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Ullrich Schäfer 2026-05-23 19:08:35 +02:00
parent 64624cb13c
commit a181dfe6b8
No known key found for this signature in database
GPG key ID: A32FF691A0F752D9
2 changed files with 40 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

38
apps/journal/app/jobs/import-batches-sweep.ts Normal file
View file

@ -0,0 +1,38 @@
import type { JobDefinition } from "@trails-cool/jobs";
import { and, lt, inArray } from "drizzle-orm";
import { getDb } from "../lib/db.ts";
import { importBatches } from "@trails-cool/db/schema/journal";
import { logger } from "../lib/logger.server.ts";
// Batches stuck longer than this without completing are considered stale.
const STALE_MS = 10 * 60 * 1000;
export const importBatchesSweepJob: JobDefinition = {
name: "import-batches-sweep",
cron: "* * * * *",
retryLimit: 0,
expireInSeconds: 55,
async handler() {
const db = getDb();
const cutoff = new Date(Date.now() - STALE_MS);
const result = await db
.update(importBatches)
.set({
status: "failed",
errorMessage: "Import timed out — the server may have restarted mid-import. Click 'Run again' to retry.",
completedAt: new Date(),
})
.where(
and(
inArray(importBatches.status, ["pending", "running"]),
lt(importBatches.startedAt, cutoff),
),
)
.returning({ id: importBatches.id });
if (result.length > 0) {
logger.info({ count: result.length }, "import-batches-sweep: marked stale batches as failed");
}
},
};

3
apps/journal/server.ts
View file

@ -144,8 +144,9 @@ server.listen(port, async () => {
const { notificationsFanoutJob } = await import("./app/jobs/notifications-fanout.ts");
const { notificationsPurgeJob } = await import("./app/jobs/notifications-purge.ts");
const { komootBulkImportJob } = await import("./app/jobs/komoot-bulk-import.ts");
const { importBatchesSweepJob } = await import("./app/jobs/import-batches-sweep.ts");
// eslint-disable-next-line @typescript-eslint/no-explicit-any
jobs.push(notificationsFanoutJob, notificationsPurgeJob, komootBulkImportJob as any);
jobs.push(notificationsFanoutJob, notificationsPurgeJob, komootBulkImportJob as any, importBatchesSweepJob);
const boss = createBoss(process.env.DATABASE_URL ?? "postgres://trails:trails@localhost:5432/trails");
await startWorker(boss, jobs);