Skip to content

Commit 9b6357e

Browse files
committed
feat: add worker timeout
setting timeout as 1 minute TODO: handle memory leaks from stalled jobs
1 parent 511832e commit 9b6357e

File tree

1 file changed

+37
-14
lines changed

1 file changed

+37
-14
lines changed

src/processor/index.ts

Lines changed: 37 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,38 @@
11
import type { Job } from "bullmq";
22
import type { SchedulerJobData } from "@wei/probot-scheduler";
3-
import type { Probot } from "probot";
3+
import type { Logger, Probot, ProbotOctokit } from "probot";
44
import logger from "@/src/utils/logger.ts";
55
import { getPullConfig } from "@/src/utils/get-pull-config.ts";
66
import { Pull } from "@/src/processor/pull.ts";
77

8+
const TIMEOUT = 60 * 1000;
9+
10+
function createTimeoutPromise(log: Logger) {
11+
return new Promise((_, reject) => {
12+
setTimeout(() => {
13+
log.warn("⏰ Job timed out after 1 minute");
14+
reject(new Error("Job timed out after 1 minute"));
15+
}, TIMEOUT);
16+
});
17+
}
18+
19+
async function processRepo(
20+
octokit: ProbotOctokit,
21+
jobData: SchedulerJobData,
22+
log: Logger,
23+
) {
24+
const { owner, repo } = jobData;
25+
26+
const config = await getPullConfig(octokit, log, jobData);
27+
if (!config) {
28+
log.info(`⚠️ No config found, skipping`);
29+
return;
30+
}
31+
32+
const pull = new Pull(octokit, { owner, repo, logger: log }, config);
33+
await pull.routineCheck();
34+
}
35+
836
export function getRepoProcessor(probot: Probot) {
937
return async function RepoJobProcessor(job: Job<SchedulerJobData>) {
1038
const log = logger.child({
@@ -14,23 +42,18 @@ export function getRepoProcessor(probot: Probot) {
1442

1543
log.info("🏃 Processing repo job");
1644

17-
const { installation_id, owner, repo } = job.data;
18-
1945
try {
20-
const octokit = await probot.auth(installation_id);
21-
22-
const config = await getPullConfig(octokit, log, job.data);
23-
if (!config) {
24-
log.info(`⚠️ No config found, skipping`);
25-
return;
26-
}
46+
const octokit = await probot.auth(job.data.installation_id);
2747

28-
const pull = new Pull(octokit, { owner, repo, logger: log }, config);
29-
await pull.routineCheck();
48+
await Promise.race([
49+
processRepo(octokit, job.data, log),
50+
createTimeoutPromise(log),
51+
]);
3052

31-
log.info(`✅ Repo job ${job.id} processed successfully`);
53+
log.info(`✅ Repo job processed successfully`);
3254
} catch (error) {
33-
log.error(error, "❌ Repo job failed");
55+
const message = error instanceof Error ? error.message : "Unknown error";
56+
log.error(error, `❌ Repo job failed: ${message}`);
3457
}
3558
};
3659
}

0 commit comments

Comments
 (0)