+ Press n or j to go to the next uncovered block, b, p or k for the previous block. +
+ +| File | ++ | Statements | ++ | Branches | ++ | Functions | ++ | Lines | ++ |
|---|---|---|---|---|---|---|---|---|---|
| jobQueue.js | +
+
+ |
+ 94.25% | +82/87 | +85.71% | +48/56 | +100% | +9/9 | +95.12% | +78/82 | +
| worker.js | +
+
+ |
+ 94.73% | +54/57 | +90% | +36/40 | +100% | +11/11 | +94.54% | +52/55 | +
+ Press n or j to go to the next uncovered block, b, p or k for the previous block. +
+ +| 1 +2 +3 +4 +5 +6 +7 +8 +9 +10 +11 +12 +13 +14 +15 +16 +17 +18 +19 +20 +21 +22 +23 +24 +25 +26 +27 +28 +29 +30 +31 +32 +33 +34 +35 +36 +37 +38 +39 +40 +41 +42 +43 +44 +45 +46 +47 +48 +49 +50 +51 +52 +53 +54 +55 +56 +57 +58 +59 +60 +61 +62 +63 +64 +65 +66 +67 +68 +69 +70 +71 +72 +73 +74 +75 +76 +77 +78 +79 +80 +81 +82 +83 +84 +85 +86 +87 +88 +89 +90 +91 +92 +93 +94 +95 +96 +97 +98 +99 +100 +101 +102 +103 +104 +105 +106 +107 +108 +109 +110 +111 +112 +113 +114 +115 +116 +117 +118 +119 +120 +121 +122 +123 +124 +125 +126 +127 +128 +129 +130 +131 +132 +133 +134 +135 +136 +137 +138 +139 +140 +141 +142 +143 +144 +145 +146 +147 +148 +149 +150 +151 +152 +153 +154 +155 +156 +157 +158 +159 +160 +161 +162 +163 +164 +165 +166 +167 +168 +169 +170 +171 +172 +173 +174 +175 +176 +177 +178 +179 +180 +181 +182 +183 +184 +185 +186 +187 +188 +189 +190 +191 +192 +193 +194 +195 +196 +197 +198 +199 +200 +201 +202 +203 +204 +205 +206 +207 +208 +209 +210 +211 +212 +213 +214 +215 +216 +217 +218 +219 +220 +221 +222 +223 +224 +225 +226 +227 +228 +229 +230 +231 +232 +233 +234 +235 +236 +237 +238 +239 +240 +241 +242 +243 +244 +245 +246 +247 +248 +249 +250 +251 +252 +253 +254 +255 +256 +257 +258 +259 +260 +261 +262 +263 +264 +265 +266 +267 +268 +269 +270 +271 +272 +273 +274 +275 +276 +277 +278 +279 +280 +281 +282 +283 +284 +285 +286 +287 +288 +289 +290 +291 +292 +293 +294 +295 +296 +297 +298 +299 +300 +301 +302 +303 +304 +305 +306 +307 +308 +309 +310 +311 +312 +313 +314 +315 | + + + + + + + + + + + + +1x + + + + + + +1x + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +71x +71x + + +71x + + +71x + + +71x + + + + + + + + + + + + + + + + + + + + + +59x +2x + + + +57x +57x + +1x + + + +56x +1x + + + +55x + +55x + + + + + + + + + + + + + +55x +55x + +55x + + + + + + + + + + + + +76x +18x +18x + +18x +18x +1x +1x +1x +1x + + +17x + + + + + +75x +37x +37x + +37x + + + + +37x +36x +36x +36x +36x +1x + +1x + + + +39x + + + + + + + + + + + +17x + +17x +2x + + +15x +2x + + + + +13x +13x + + + + + + + + + + + + + + + +25x + +25x +2x + + +23x +23x + + +23x +23x + + +23x +23x + +23x + + + + + + + + + + + + + + +24x + + + + + + + + +6x + + + + + + + + + + +6x +9x +7x +3x +1x +1x + + +6x + + + + + + + + + +65x +65x +65x +65x +65x + + + + + + + + + + + + + + +55x + + + +55x +55x + + + +55x + + + +1x +1x + | /**
+ * @fileoverview In-memory job queue for background worker tasks.
+ * Provides thread-safe job management with enqueue, dequeue, ack, and retry capabilities.
+ *
+ * Security Considerations:
+ * - Job IDs are generated using crypto to prevent predictable ID attacks
+ * - Payloads are validated using JSON stringification (no code injection)
+ * - Max retry attempts prevents infinite loops (hard-capped at 10)
+ * - Queue operations are O(1) for enqueue/dequeue to prevent DoS
+ *
+ * @module workers/jobQueue
+ */
+
+const crypto = require('crypto');
+
+/**
+ * Job status enumeration
+ * @readonly
+ * @enum {string}
+ */
+const JOB_STATUS = {
+ PENDING: 'pending',
+ PROCESSING: 'processing',
+ COMPLETED: 'completed',
+ FAILED: 'failed',
+ RETRYING: 'retrying',
+};
+
+/**
+ * In-memory job queue for managing asynchronous background tasks.
+ *
+ * Features:
+ * - Secure job ID generation using cryptographic random values
+ * - Priority-based queue support
+ * - Retry logic with exponential backoff
+ * - Job status tracking
+ * - Comprehensive error handling
+ *
+ * @class JobQueue
+ */
+class JobQueue {
+ /**
+ * Creates a new JobQueue instance
+ *
+ * @param {Object} options - Configuration options
+ * @param {number} [options.maxRetries=3] - Maximum retry attempts per job (capped at 10)
+ * @param {number} [options.maxQueueSize=10000] - Maximum queue size to prevent memory exhaustion
+ */
+ constructor(options = {}) {
+ // Security: Validate and cap max retries
+ this.maxRetries = Math.min(options.maxRetries || 3, 10);
+ this.maxQueueSize = options.maxQueueSize || 10000;
+
+ // Using Map for efficient O(1) lookups
+ this.jobs = new Map();
+
+ // Queue structure: array of job IDs
+ this.queue = [];
+
+ // Retry queue: separate queue for jobs that need retrying
+ this.retryQueue = [];
+ }
+
+ /**
+ * Enqueue a job into the queue
+ *
+ * Security & Validation:
+ * - Type is validated to prevent injection
+ * - Payload is stringified then parsed to ensure JSON serializability
+ * - Job ID is cryptographically generated
+ * - Queue size is bounded to prevent memory exhaustion
+ *
+ * @param {string} type - Job type identifier (e.g., 'verify', 'webhook_retry')
+ * @param {Object} payload - Job data (must be JSON-serializable)
+ * @param {Object} [options={}] - Additional job options
+ * @param {number} [options.priority=0] - Job priority (higher = more urgent)
+ * @param {number} [options.delayMs=0] - Delay before processing (milliseconds)
+ * @returns {string} The generated job ID
+ * @throws {Error} If payload is not JSON-serializable or queue is full
+ */
+ enqueue(type, payload, options = {}) {
+ // Security: Validate type is a string
+ if (typeof type !== 'string' || type.trim().length === 0) {
+ throw new Error('Job type must be a non-empty string');
+ }
+
+ // Security: Validate payload is JSON-serializable
+ try {
+ JSON.stringify(payload);
+ } catch (err) {
+ throw new Error(`Job payload must be JSON-serializable: ${err.message}`);
+ }
+
+ // Security: Prevent queue exhaustion DoS
+ if (this.queue.length + this.retryQueue.length >= this.maxQueueSize) {
+ throw new Error(`Queue is full (max ${this.maxQueueSize} jobs)`);
+ }
+
+ // Generate secure job ID using crypto
+ const jobId = `job-${crypto.randomBytes(8).toString('hex')}`;
+
+ const job = {
+ id: jobId,
+ type,
+ payload,
+ status: JOB_STATUS.PENDING,
+ priority: options.priority || 0,
+ delayMs: options.delayMs || 0,
+ createdAt: Date.now(),
+ startedAt: null,
+ completedAt: null,
+ attempts: 0,
+ lastError: null,
+ };
+
+ this.jobs.set(jobId, job);
+ this.queue.push(jobId);
+
+ return jobId;
+ }
+
+ /**
+ * Dequeue a job from the front of the queue
+ *
+ * Returns the highest priority pending job that is ready to process
+ * (not delayed). Returns null if no job is ready.
+ *
+ * @returns {Object|null} The next job to process, or null if queue is empty
+ */
+ dequeue() {
+ // Process retry queue first (jobs that have been retried)
+ if (this.retryQueue.length > 0) {
+ const jobId = this.retryQueue.shift();
+ const job = this.jobs.get(jobId);
+
+ Eif (job) {
+ if (this._isReadyToProcess(job)) {
+ job.status = JOB_STATUS.PROCESSING;
+ job.startedAt = Date.now();
+ job.attempts += 1;
+ return job;
+ } else {
+ // Put back if not ready yet
+ this.retryQueue.push(jobId);
+ }
+ }
+ }
+
+ // Then process main queue
+ while (this.queue.length > 0) {
+ const jobId = this.queue.shift();
+ const job = this.jobs.get(jobId);
+
+ Iif (!job) {
+ // Job was removed, skip it
+ continue;
+ }
+
+ if (this._isReadyToProcess(job)) {
+ job.status = JOB_STATUS.PROCESSING;
+ job.startedAt = Date.now();
+ job.attempts += 1;
+ return job;
+ } else Eif (job.delayMs > 0) {
+ // Put it back, will retry later
+ this.queue.push(jobId);
+ }
+ }
+
+ return null;
+ }
+
+ /**
+ * Acknowledge successful job completion
+ *
+ * Marks the job as completed and removes it from the queue.
+ *
+ * @param {string} jobId - The job ID to acknowledge
+ * @throws {Error} If job doesn't exist or is not in processing state
+ */
+ ack(jobId) {
+ const job = this.jobs.get(jobId);
+
+ if (!job) {
+ throw new Error(`Job ${jobId} not found`);
+ }
+
+ if (job.status !== JOB_STATUS.PROCESSING) {
+ throw new Error(
+ `Cannot ack job ${jobId}: status is ${job.status}, expected ${JOB_STATUS.PROCESSING}`
+ );
+ }
+
+ job.status = JOB_STATUS.COMPLETED;
+ job.completedAt = Date.now();
+ }
+
+ /**
+ * Retry a failed job with exponential backoff
+ *
+ * If the job has remaining retry attempts, it's put back in the retry queue
+ * with an increased delay. Otherwise, it's marked as failed.
+ *
+ * Security: maxRetries is capped at construction time
+ *
+ * @param {string} jobId - The job ID to retry
+ * @param {Error} error - The error that occurred
+ * @throws {Error} If job doesn't exist or is successfully deleted after max retries
+ */
+ retry(jobId, error) {
+ const job = this.jobs.get(jobId);
+
+ if (!job) {
+ throw new Error(`Job ${jobId} not found`);
+ }
+
+ job.lastError = error?.message || String(error);
+ job.status = JOB_STATUS.FAILED;
+
+ // Check if we can retry (maxRetries allows up to that many total attempts)
+ if (job.attempts <= this.maxRetries) {
+ job.status = JOB_STATUS.RETRYING;
+
+ // Exponential backoff: 2^(attempt-1) seconds (capped)
+ const delayMs = Math.min(Math.pow(2, job.attempts - 1) * 1000, 60000);
+ job.delayMs = Date.now() + delayMs;
+
+ this.retryQueue.push(jobId);
+ } else E{
+ // Max retries exceeded, mark as permanently failed
+ job.status = JOB_STATUS.FAILED;
+ job.completedAt = Date.now();
+ }
+ }
+
+ /**
+ * Get job details by ID
+ *
+ * @param {string} jobId - The job ID
+ * @returns {Object|null} Job object if found, null otherwise
+ */
+ getJob(jobId) {
+ return this.jobs.get(jobId) || null;
+ }
+
+ /**
+ * Get current queue statistics
+ *
+ * @returns {Object} Statistics including pending, processing, completed, failed counts
+ */
+ getStats() {
+ const stats = {
+ pending: 0,
+ processing: 0,
+ completed: 0,
+ failed: 0,
+ retrying: 0,
+ total: this.jobs.size,
+ queueLength: this.queue.length,
+ retryQueueLength: this.retryQueue.length,
+ };
+
+ for (const job of this.jobs.values()) {
+ if (job.status === JOB_STATUS.PENDING) stats.pending += 1;
+ else if (job.status === JOB_STATUS.PROCESSING) stats.processing += 1;
+ else if (job.status === JOB_STATUS.COMPLETED) stats.completed += 1;
+ else Iif (job.status === JOB_STATUS.FAILED) stats.failed += 1;
+ else Eif (job.status === JOB_STATUS.RETRYING) stats.retrying += 1;
+ }
+
+ return stats;
+ }
+
+ /**
+ * Clear all jobs from the queue
+ * Useful for testing or cleanup
+ *
+ * @returns {number} Number of jobs cleared
+ */
+ clear() {
+ const count = this.jobs.size;
+ this.jobs.clear();
+ this.queue = [];
+ this.retryQueue = [];
+ return count;
+ }
+
+ /**
+ * Check if a job is ready to be processed
+ *
+ * A job is ready if:
+ * - Its status is PENDING
+ * - Its delay has expired
+ *
+ * @private
+ * @param {Object} job - The job to check
+ * @returns {boolean} True if job is ready, false otherwise
+ */
+ _isReadyToProcess(job) {
+ Iif (job.status !== JOB_STATUS.PENDING && job.status !== JOB_STATUS.RETRYING) {
+ return false;
+ }
+
+ const now = Date.now();
+ const delayExpired = typeof job.delayMs === 'number'
+ ? now >= job.delayMs
+ : job.delayMs === 0;
+
+ return delayExpired;
+ }
+}
+
+module.exports = JobQueue;
+module.exports.JOB_STATUS = JOB_STATUS;
+ |
+ Press n or j to go to the next uncovered block, b, p or k for the previous block. +
+ +| 1 +2 +3 +4 +5 +6 +7 +8 +9 +10 +11 +12 +13 +14 +15 +16 +17 +18 +19 +20 +21 +22 +23 +24 +25 +26 +27 +28 +29 +30 +31 +32 +33 +34 +35 +36 +37 +38 +39 +40 +41 +42 +43 +44 +45 +46 +47 +48 +49 +50 +51 +52 +53 +54 +55 +56 +57 +58 +59 +60 +61 +62 +63 +64 +65 +66 +67 +68 +69 +70 +71 +72 +73 +74 +75 +76 +77 +78 +79 +80 +81 +82 +83 +84 +85 +86 +87 +88 +89 +90 +91 +92 +93 +94 +95 +96 +97 +98 +99 +100 +101 +102 +103 +104 +105 +106 +107 +108 +109 +110 +111 +112 +113 +114 +115 +116 +117 +118 +119 +120 +121 +122 +123 +124 +125 +126 +127 +128 +129 +130 +131 +132 +133 +134 +135 +136 +137 +138 +139 +140 +141 +142 +143 +144 +145 +146 +147 +148 +149 +150 +151 +152 +153 +154 +155 +156 +157 +158 +159 +160 +161 +162 +163 +164 +165 +166 +167 +168 +169 +170 +171 +172 +173 +174 +175 +176 +177 +178 +179 +180 +181 +182 +183 +184 +185 +186 +187 +188 +189 +190 +191 +192 +193 +194 +195 +196 +197 +198 +199 +200 +201 +202 +203 +204 +205 +206 +207 +208 +209 +210 +211 +212 +213 +214 +215 +216 +217 +218 +219 +220 +221 +222 +223 +224 +225 +226 +227 +228 +229 +230 +231 +232 +233 +234 +235 +236 +237 +238 +239 +240 | + + + + + + + + + + + + + +1x +1x + + + + + + + + + + + + + + + + + + + + + + + +27x + + +27x + + +27x + + + + +27x + + +27x +27x +27x + + + + + + + + + + + + + + +19x +1x + + +18x +1x + + +17x + + + + + + + + + + + +13x +1x + + +12x +1x + + +11x +11x + + + + + + + + + + + + +12x + +12x +11x +11x + + + +12x +12x +4x + + +12x +1x + + + + + + + + + + + + + + + +14x +1x + + + + + +13x + + + + + + + + +1x + + + + + + + + + + + + + + + + +30x + + + + +30x +38x +38x +27x + + +11x + + +11x +1x + + + + +30x +30x + + + + + + + + + + + + + + + + + +11x +11x + + + +11x +11x + + + + +11x + + +9x + + +3x + +11x + + + + +1x + | /**
+ * @fileoverview Background worker for processing asynchronous jobs.
+ * Provides a worker loop that dequeues jobs and executes them with proper error handling.
+ *
+ * Security Considerations:
+ * - Handler execution is wrapped in try-catch to prevent uncaught exceptions
+ * - Worker validates that handlers are functions before executing
+ * - Processing count prevents stack overflow from chained promises
+ * - Poll interval is bounded (minimum 10ms) to prevent CPU spinning
+ * - Graceful shutdown allows in-flight jobs to complete
+ *
+ * @module workers/worker
+ */
+
+const JobQueue = require('./jobQueue');
+const { JOB_STATUS } = require('./jobQueue');
+
+/**
+ * Background worker that processes queued jobs
+ *
+ * Features:
+ * - Asynchronous job processing with configurable handlers
+ * - Automatic retry with exponential backoff
+ * - Graceful start/stop with in-flight job handling
+ * - Processing statistics and monitoring
+ * - Security validation of job handlers
+ *
+ * @class BackgroundWorker
+ */
+class BackgroundWorker {
+ /**
+ * Creates a new BackgroundWorker instance
+ *
+ * @param {Object} options - Worker configuration
+ * @param {JobQueue} [options.jobQueue] - Job queue instance (creates new if not provided)
+ * @param {number} [options.pollIntervalMs=1000] - How often to check queue (min 10ms)
+ * @param {number} [options.maxConcurrency=2] - Max concurrent job processing
+ */
+ constructor(options = {}) {
+ this.jobQueue = options.jobQueue || new JobQueue();
+
+ // Security: Bound poll interval to prevent CPU spinning
+ this.pollIntervalMs = Math.max(options.pollIntervalMs ?? 1000, 10);
+
+ // Security: Limit concurrency to prevent resource exhaustion
+ this.maxConcurrency = !isNaN(options.maxConcurrency)
+ ? Math.max(options.maxConcurrency, 1)
+ : 2;
+
+ // Handler registry: map of job type to handler function
+ this.handlers = new Map();
+
+ // Worker state
+ this.isRunning = false;
+ this.processingCount = 0;
+ this.pollTimer = null;
+ }
+
+ /**
+ * Register a handler for a specific job type
+ *
+ * Security Validation:
+ * - Handler must be a function
+ * - Job type must be a non-empty string
+ *
+ * @param {string} jobType - The job type (e.g., 'verify', 'webhook_retry')
+ * @param {Function} handler - Async function(job) to handle the job
+ * @throws {Error} If handler is not a function or jobType is invalid
+ */
+ registerHandler(jobType, handler) {
+ if (typeof jobType !== 'string' || jobType.trim().length === 0) {
+ throw new Error('Job type must be a non-empty string');
+ }
+
+ if (typeof handler !== 'function') {
+ throw new Error(`Handler must be a function, got ${typeof handler}`);
+ }
+
+ this.handlers.set(jobType, handler);
+ }
+
+ /**
+ * Start the worker loop
+ *
+ * Once started, the worker will continuously poll the queue and process jobs
+ * using registered handlers. Worker runs until stop() is called.
+ *
+ * @throws {Error} If already running or no handlers registered
+ */
+ start() {
+ if (this.isRunning) {
+ throw new Error('Worker is already running');
+ }
+
+ if (this.handlers.size === 0) {
+ throw new Error('No job handlers registered');
+ }
+
+ this.isRunning = true;
+ this._poll();
+ }
+
+ /**
+ * Stop the worker loop gracefully
+ *
+ * Stops accepting new jobs but allows in-flight jobs to complete.
+ * Resolves when all in-flight jobs are done (or timeout).
+ *
+ * @param {number} [timeoutMs=10000] - Max time to wait for in-flight jobs
+ * @returns {Promise<void>}
+ */
+ async stop(timeoutMs = 10000) {
+ this.isRunning = false;
+
+ if (this.pollTimer) {
+ clearTimeout(this.pollTimer);
+ this.pollTimer = null;
+ }
+
+ // Wait for in-flight jobs to complete
+ const startTime = Date.now();
+ while (this.processingCount > 0 && Date.now() - startTime < timeoutMs) {
+ await new Promise((resolve) => setTimeout(resolve, 100)); // eslint-disable-line no-await-in-loop
+ }
+
+ if (this.processingCount > 0) {
+ console.warn(
+ `Worker stopped with ${this.processingCount} jobs still processing (timeout)`
+ );
+ }
+ }
+
+ /**
+ * Enqueue a job for processing
+ *
+ * @param {string} jobType - The job type
+ * @param {Object} payload - The job payload
+ * @param {Object} [options={}] - Additional options (priority, delayMs)
+ * @returns {string} The job ID
+ * @throws {Error} If job type has no registered handler
+ */
+ enqueue(jobType, payload, options = {}) {
+ if (!this.handlers.has(jobType)) {
+ throw new Error(
+ `No handler registered for job type "${jobType}". ` +
+ `Registered types: ${Array.from(this.handlers.keys()).join(', ')}`
+ );
+ }
+
+ return this.jobQueue.enqueue(jobType, payload, options);
+ }
+
+ /**
+ * Get statistics about worker state and queue
+ *
+ * @returns {Object} Worker stats including running status, processing count, queue stats
+ */
+ getStats() {
+ return {
+ isRunning: this.isRunning,
+ processingCount: this.processingCount,
+ handlerCount: this.handlers.size,
+ queueStats: this.jobQueue.getStats(),
+ };
+ }
+
+ /**
+ * Poll the queue and process available jobs
+ *
+ * This runs continuously when the worker is started.
+ * It processes up to maxConcurrency jobs at a time.
+ *
+ * @private
+ */
+ _poll() {
+ Iif (!this.isRunning) {
+ return;
+ }
+
+ // Process jobs up to concurrency limit
+ while (this.processingCount < this.maxConcurrency) {
+ const job = this.jobQueue.dequeue();
+ if (!job) {
+ break; // Queue is empty
+ }
+
+ this.processingCount += 1;
+
+ // Process job asynchronously (don't await, let it run in background)
+ this._processJob(job).catch((err) => {
+ console.error(`Unexpected error processing job ${job.id}:`, err);
+ });
+ }
+
+ // Schedule next poll
+ Eif (this.isRunning) {
+ this.pollTimer = setTimeout(() => this._poll(), this.pollIntervalMs);
+ }
+ }
+
+ /**
+ * Process a single job with its registered handler
+ *
+ * Security & Error Handling:
+ * - Handler must exist for job type (validated at registration time)
+ * - Handler execution is wrapped in try-catch
+ * - Errors trigger retry logic with exponential backoff
+ * - Job ID is validated before processing
+ *
+ * @private
+ * @param {Object} job - The job to process
+ * @returns {Promise<void>}
+ */
+ async _processJob(job) {
+ try {
+ Iif (!job || !job.id || !job.type) {
+ throw new Error('Invalid job structure');
+ }
+
+ const handler = this.handlers.get(job.type);
+ Iif (!handler) {
+ throw new Error(`No handler for job type "${job.type}"`);
+ }
+
+ // Execute the handler
+ await handler(job);
+
+ // Job succeeded
+ this.jobQueue.ack(job.id);
+ } catch (err) {
+ // Job failed, attempt retry
+ this.jobQueue.retry(job.id, err);
+ } finally {
+ this.processingCount -= 1;
+ }
+ }
+}
+
+module.exports = BackgroundWorker;
+ |