alikia2x/cvsa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update: handler for SIGINT & SIGTERM

Browse Source
This commit is contained in:
alikia2x (寒寒) 2025-04-14 00:49:57 +08:00
parent 6df6345ec1
commit 0614067278
Signed by: alikia2x
GPG Key ID: 56209E0CCD8420C6
1 changed files with 15 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
packages/crawler/src/worker.ts
View File

@ -16,8 +16,21 @@ import {
takeSnapshotForVideoWorker, takeSnapshotForVideoWorker,
} from "mq/exec/snapshotTick.ts"; } from "mq/exec/snapshotTick.ts";
const releaseLockForJob = async (name: string) => {
await lockManager.releaseLock(name);
logger.log(`Released lock: ${name}`, "mq");
}
const releaseAllLocks = async () => {
const locks = ["dispatchRegularSnapshots", "dispatchArchiveSnapshots", "getLatestVideos"];
for (const lock of locks) {
await releaseLockForJob(lock);
}
}
Deno.addSignalListener("SIGINT", async () => { Deno.addSignalListener("SIGINT", async () => {
logger.log("SIGINT Received: Shutting down workers...", "mq"); logger.log("SIGINT Received: Shutting down workers...", "mq");
await releaseAllLocks();
await latestVideoWorker.close(true); await latestVideoWorker.close(true);
await snapshotWorker.close(true); await snapshotWorker.close(true);
Deno.exit(); Deno.exit();
@ -25,6 +38,7 @@ Deno.addSignalListener("SIGINT", async () => {
Deno.addSignalListener("SIGTERM", async () => { Deno.addSignalListener("SIGTERM", async () => {
logger.log("SIGTERM Received: Shutting down workers...", "mq"); logger.log("SIGTERM Received: Shutting down workers...", "mq");
await releaseAllLocks();
await latestVideoWorker.close(true); await latestVideoWorker.close(true);
await snapshotWorker.close(true); await snapshotWorker.close(true);
Deno.exit(); Deno.exit();
@ -61,10 +75,6 @@ latestVideoWorker.on("error", (err) => {
logger.error(e.rawError, e.service, e.codePath); logger.error(e.rawError, e.service, e.codePath);
}); });
latestVideoWorker.on("closed", async () => {
await lockManager.releaseLock("getLatestVideos");
});
const snapshotWorker = new Worker( const snapshotWorker = new Worker(
"snapshot", "snapshot",
async (job: Job) => { async (job: Job) => {
@ -95,10 +105,4 @@ const snapshotWorker = new Worker(
snapshotWorker.on("error", (err) => { snapshotWorker.on("error", (err) => {
const e = err as WorkerError; const e = err as WorkerError;
logger.error(e.rawError, e.service, e.codePath); logger.error(e.rawError, e.service, e.codePath);
}); });
snapshotWorker.on("closed", async () => {
await lockManager.releaseLock("dispatchRegularSnapshots");
await lockManager.releaseLock("dispatchArchiveSnapshots");
logger.log(`Released lock: dispatchArchiveSnapshots, dispatchRegularSnapshots`, "mq", "snapshotWorker:on:closed");
});