feat: add graceful shutdown handling to scraper job scheduler
All checks were successful
CI/CD Pipeline - Apartment API / Scan Dependencies (pull_request) Successful in 13s
CI/CD Pipeline - Apartment API / Lint & Test (pull_request) Successful in 39s
CI/CD Pipeline - Apartment API / Send Webhook Notification (pull_request) Successful in 2s
CI/CD Pipeline - Apartment API / Build & Push Image (pull_request) Has been skipped
CI/CD Pipeline - Apartment API / Deploy to Production (pull_request) Has been skipped
All checks were successful
CI/CD Pipeline - Apartment API / Scan Dependencies (pull_request) Successful in 13s
CI/CD Pipeline - Apartment API / Lint & Test (pull_request) Successful in 39s
CI/CD Pipeline - Apartment API / Send Webhook Notification (pull_request) Successful in 2s
CI/CD Pipeline - Apartment API / Build & Push Image (pull_request) Has been skipped
CI/CD Pipeline - Apartment API / Deploy to Production (pull_request) Has been skipped
Add gracefulShutdown(), registerSignalHandlers(), and isShuttingDown() to enable clean process termination during scraper operations. Key changes: - Register SIGTERM and SIGINT signal handlers for process shutdown - Wait for in-progress scraper jobs to complete before exiting, with a configurable timeout (SCRAPER_SHUTDOWN_TIMEOUT, default 30s) - Release mutex lock on shutdown to prevent stale lock state - Track shutdown state to reject new job acquisitions during shutdown - Stop cron scheduler as part of the shutdown sequence - Add SHUTDOWN_TIMEOUT to scraper config module Adds 14 new tests covering shutdown lifecycle, timeout enforcement, idempotent shutdown calls, and lock rejection during shutdown (56 total).
This commit is contained in:
@ -11,6 +11,10 @@ let currentJobId = null;
|
||||
// Scheduler state
|
||||
let scheduledJob = null;
|
||||
|
||||
// Shutdown state
|
||||
let shuttingDown = false;
|
||||
let runningJobPromise = null;
|
||||
|
||||
/**
|
||||
* Check if scraper is currently running
|
||||
* @returns {boolean}
|
||||
@ -33,7 +37,7 @@ function getCurrentJobId() {
|
||||
* @returns {boolean} True if lock acquired
|
||||
*/
|
||||
function acquireLock(jobId) {
|
||||
if (isRunning) {
|
||||
if (isRunning || shuttingDown) {
|
||||
return false;
|
||||
}
|
||||
isRunning = true;
|
||||
@ -141,20 +145,23 @@ function initializeScheduler(db) {
|
||||
|
||||
jobLogger.info('Scheduled scrape triggered');
|
||||
|
||||
// Check if already running
|
||||
// Check if already running or shutting down
|
||||
if (!acquireLock(jobId)) {
|
||||
jobLogger.warn('Skipped - scrape already in progress');
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await runScrape(db, { trigger: 'scheduled', jobId });
|
||||
const jobExecution = runScrape(db, { trigger: 'scheduled', jobId });
|
||||
runningJobPromise = jobExecution;
|
||||
await jobExecution;
|
||||
} catch (error) {
|
||||
jobLogger.error('Scheduled scrape failed', {
|
||||
errorType: error.name,
|
||||
errorMessage: error.message
|
||||
});
|
||||
} finally {
|
||||
runningJobPromise = null;
|
||||
releaseLock();
|
||||
}
|
||||
}, {
|
||||
@ -179,6 +186,80 @@ function stopScheduler() {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if the scraper is in the process of shutting down
|
||||
* @returns {boolean}
|
||||
*/
|
||||
function isShuttingDown() {
|
||||
return shuttingDown;
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform a graceful shutdown of the scraper
|
||||
* - Stops the cron scheduler to prevent new jobs
|
||||
* - Waits for any running job to complete (with timeout)
|
||||
* - Releases the mutex lock
|
||||
* - Logs shutdown progress
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
async function gracefulShutdown() {
|
||||
const logger = createLogger('shutdown');
|
||||
|
||||
logger.info('Shutdown initiated');
|
||||
|
||||
// Mark as shutting down to prevent new jobs
|
||||
shuttingDown = true;
|
||||
|
||||
// Stop the cron scheduler
|
||||
stopScheduler();
|
||||
|
||||
// Wait for running job to complete (with timeout)
|
||||
if (isRunning && runningJobPromise) {
|
||||
logger.info('Waiting for running job to complete', {
|
||||
jobId: currentJobId,
|
||||
timeout: config.SHUTDOWN_TIMEOUT
|
||||
});
|
||||
|
||||
let timeoutHandle;
|
||||
const timeoutPromise = new Promise((resolve) => {
|
||||
timeoutHandle = setTimeout(() => resolve('timeout'), config.SHUTDOWN_TIMEOUT);
|
||||
});
|
||||
|
||||
const result = await Promise.race([
|
||||
runningJobPromise.then(() => 'completed').catch(() => 'completed'),
|
||||
timeoutPromise
|
||||
]);
|
||||
|
||||
clearTimeout(timeoutHandle);
|
||||
|
||||
if (result === 'timeout') {
|
||||
logger.warn('Shutdown wait for running job timed out', {
|
||||
jobId: currentJobId,
|
||||
timeout: config.SHUTDOWN_TIMEOUT
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Force release the lock
|
||||
releaseLock();
|
||||
|
||||
logger.info('Shutdown complete');
|
||||
}
|
||||
|
||||
/**
|
||||
* Register process signal handlers for graceful shutdown
|
||||
* Listens for SIGTERM and SIGINT signals
|
||||
*/
|
||||
function registerSignalHandlers() {
|
||||
process.on('SIGTERM', () => {
|
||||
gracefulShutdown();
|
||||
});
|
||||
|
||||
process.on('SIGINT', () => {
|
||||
gracefulShutdown();
|
||||
});
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
isScraperRunning,
|
||||
getCurrentJobId,
|
||||
@ -187,5 +268,8 @@ module.exports = {
|
||||
getScheduleExpression,
|
||||
getNextScheduledRun,
|
||||
initializeScheduler,
|
||||
stopScheduler
|
||||
stopScheduler,
|
||||
isShuttingDown,
|
||||
gracefulShutdown,
|
||||
registerSignalHandlers
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user