[Bug]: QUeues get stuck in Active

[tariqbilal]

Version

v5.54.0

Platform

NodeJS

What happened?

Title: Jobs randomly get stuck in "active" state not even working after restart and fails.

Description:
Jobs in our BullMQ queues periodically get stuck in the "active" state without being processed or moved to failed state. This affects only specific queues while others continue working normally. There is no resolution I have found yet.

Environment:

• BullMQ Version: [v5.54.0]
• Node.js Version: [22]
• Redis Version: [redis-cli 7.0.15]
• Operating System: Ubuntu 24
• Deployment: VM

Reproduction Steps:

1. Start a BullMQ worker processing jobs normally
2. After several hours/days of operation, observe that:
   • New jobs enter the queue
   • Jobs are picked up and move to "active" state
   • Jobs remain stuck in "active" state indefinitely
   • No progress is made on stuck jobs
   • No error events are emitted

Expected Behavior:

• Jobs should either complete processing and move to "completed" state
• Or jobs should fail and move to "failed" state with appropriate error
• Stalled jobs should be automatically detected and restarted based on configuration

Actual Behavior:

• Jobs remain stuck in "active" state indefinitely
• No automatic stall detection occurs despite stalledInterval configuration
• No error events or logs are generated for stuck jobs

Configuration:

const queue = new Queue('affected-queue', {
  connection: redisConfig,
  settings: {
    stalledInterval: 30000,
    maxStalledCount: 2,
  },
  defaultJobOptions: {
    attempts: 3,
    backoff: {
      type: 'exponential',
      delay: 1000,
    },
    removeOnComplete: 100,
    removeOnFail: 100,
  },
});

### How to reproduce.

function initializeWorker(sessions) {
    const worker = new Worker('whatsapp-jobs', async (job) => {
        console.log(`Processing job ${job.id} of type ${job.name}`);
        
        // 🔴 ADD THIS: Simulate stuck condition (1% chance)
        if (Math.random() < 0.01) {
            console.log(`🚨 SIMULATING STUCK JOB: ${job.id} - This job will hang forever`);
            // This promise never resolves - simulating the exact stuck condition
            await new Promise(() => {});
            // Code never reaches here
        }
        
        // 🔴 ADD THIS: Simulate event loop blocking (0.5% chance)
        if (Math.random() < 0.005) {
            console.log(`🚨 SIMULATING BLOCKED EVENT LOOP: ${job.id}`);
            // Block the event loop for 5 minutes
            const start = Date.now();
            while (Date.now() - start < 300000) {
                // Busy wait - blocks event loop
            }
        }
        
        // Your existing code continues...
        try {
            const { sessionName, to, text, question, options, timestamp } = job.data;
            
            switch (job.name) {
                // ... your existing cases
            }
        } catch (error) {
            console.error(`Error processing job ${job.id}:`, error);
            throw error;
        }
    }, {
        // Your existing configuration
        connection: {
            host: process.env.REDIS_HOST || 'localhost',
            port: parseInt(process.env.REDIS_PORT) || 6379,
            maxRetriesPerRequest: null,
            enableReadyCheck: false
        },
        concurrency: 10,
        limiter: { max: 200, duration: 10000 },
        stalledInterval: 120000, // Should detect stalls every 2 minutes
        maxStalledCount: 3,
        lockDuration: 180000, // 3 minute lock
        lockRenewTime: 90000, // Renew every 90 seconds
        autorun: true
    });

    // Your existing event handlers...
    return worker;
}

<img width="1680" height="978" alt="Image" src="https://github.com/user-attachments/assets/af3e5819-6f6a-47dc-99cd-69a47a51d122" />

### Relevant log output

```shell
I have already shared everything.
<img width="1680" height="978" alt="Image" src="https://github.com/user-attachments/assets/af3e5819-6f6a-47dc-99cd-69a47a51d122" />

Code of Conduct

• I agree to follow this project's Code of Conduct

[manast]

// 🔴 ADD THIS: Simulate stuck condition (1% chance)
if (Math.random() < 0.01) {
console.log(🚨 SIMULATING STUCK JOB: ${job.id} - This job will hang forever);
// This promise never resolves - simulating the exact stuck condition
await new Promise(() => {});
// Code never reaches here
}

There is no way in NodeJS to detect a condition like this, if you have a Promise that never resolves in your process function, that job will be active for ever or until the worker is closed (forced closed actually). Then another worker will mark it as stalled and picked again.

[tariqbilal]

@manast ok I have followed your approach. now they have started failing without completing. Previously all these jobs used to work so there is no issue with my code.

import { Worker } from 'bullmq';
import { sendText, sendImage } from './messageService.js';
import { sendPoll } from './pollService.js';
import sessions from '../routes/sessionStore.js';

/**
 * Timeout wrapper to prevent jobs from hanging forever
 * This addresses the issue where Promises never resolve, causing eternal stuck jobs
 */
function withTimeout(promise, timeoutMs = 60000, jobId, jobType) {
    return Promise.race([
        promise,
        new Promise((_, reject) => 
            setTimeout(() => {
                reject(new Error(`Job ${jobId} (${jobType}) timed out after ${timeoutMs}ms - Promise never resolved`));
            }, timeoutMs)
        )
    ]);
}

function initializeWorker(sessions) {
    const worker = new Worker('whatsapp-jobs', async (job) => {
        console.log(`Processing job ${job.id} of type ${job.name}`);
        
        // Log delay information if available
        if (job.data.delayInfo) {
            const { delayType, delay, description, estimatedSendTime } = job.data.delayInfo;
            const actualDelay = Date.now() - job.timestamp;
            console.log(`Delay info: ${delayType} - ${description}`);
            console.log(`Estimated send time: ${estimatedSendTime}`);
            console.log(`Actual delay: ${actualDelay}ms (difference: ${actualDelay - delay}ms)`);
        }
        
        try {
            const { sessionName, to, text, question, options, timestamp } = job.data;
            
            switch (job.name) {
                case 'test-connection':
                    console.log(`Test connection job ${job.id} processing at ${new Date().toISOString()}`);
                    console.log(`Job data:`, { timestamp });
                    return { success: true, message: 'Connection test successful' };

                case 'test-job':
                    console.log(`Test job ${job.id} processing at ${new Date().toISOString()}`);
                    console.log(`Job data:`, { sessionName, timestamp });
                    return { success: true, message: 'Test job completed' };

                case 'send-message':
                    const session = await sessions.get(sessionName);
                    if (!session) {
                        throw new Error(`Session ${sessionName} not found`);
                    }
                    if (session.status !== 'connected') {
                        throw new Error(`Session ${sessionName} is not connected`);
                    }
                    console.log(`Sending message to ${to} from session ${sessionName}`);
                    
                    // Wrap in timeout to prevent eternal hangs
                    const messageResult = await withTimeout(
                        sendText(session.sock, to, text, sessionName),
                        60000, // 60 second timeout
                        job.id,
                        'send-message'
                    );
                    
                    // Add delay information to result
                    if (job.data.delayInfo) {
                        messageResult.delayInfo = {
                            ...job.data.delayInfo,
                            actualSendTime: new Date().toISOString(),
                            actualDelay: Date.now() - job.timestamp
                        };
                    }
                    
                    return messageResult;
                
                case 'send-poll':
                    const pollSession = await sessions.get(sessionName);
                    if (!pollSession) {
                        throw new Error(`Session ${sessionName} not found`);
                    }
                    if (pollSession.status !== 'connected') {
                        throw new Error(`Session ${sessionName} is not connected`);
                    }
                    console.log(`Sending poll to ${to} from session ${sessionName}`);
                    const { order_id, order_number } = job.data;
                    
                    // Wrap in timeout to prevent eternal hangs
                    const pollResult = await withTimeout(
                        sendPoll(pollSession.sock, to, question, options, order_id, order_number),
                        90000, // 90 second timeout (polls can take longer)
                        job.id,
                        'send-poll'
                    );
                    
                    // Add delay information to result
                    if (job.data.delayInfo) {
                        pollResult.delayInfo = {
                            ...job.data.delayInfo,
                            actualSendTime: new Date().toISOString(),
                            actualDelay: Date.now() - job.timestamp
                        };
                    }
                    
                    return pollResult;
                
                case 'send-image':
                    const imageSession = await sessions.get(sessionName);
                    if (!imageSession) {
                        throw new Error(`Session ${sessionName} not found`);
                    }
                    if (imageSession.status !== 'connected') {
                        throw new Error(`Session ${sessionName} is not connected`);
                    }
                    console.log(`Sending image to ${to} from session ${sessionName}`);
                    
                    // Wrap in timeout to prevent eternal hangs
                    const imageResult = await withTimeout(
                        sendImage(imageSession.sock, job.data.to, job.data.imageUrl, job.data.caption, sessionName),
                        120000, // 120 second timeout (images can take longer to download/send)
                        job.id,
                        'send-image'
                    );
                    
                    if (job.data.delayInfo) {
                        imageResult.delayInfo = {
                            ...job.data.delayInfo,
                            actualSendTime: new Date().toISOString(),
                            actualDelay: Date.now() - job.timestamp
                        };
                    }
                    return imageResult;
                
                default:
                    throw new Error(`Unknown job type: ${job.name}`);
            }
        } catch (error) {
            console.error(`Error processing job ${job.id}:`, error);
            throw error;
        }
    }, {
        connection: {
            host: process.env.REDIS_HOST || 'localhost',
            port: parseInt(process.env.REDIS_PORT) || 6379,
            maxRetriesPerRequest: null, // Required by BullMQ - prevents job stalling
            enableReadyCheck: false
        },
        concurrency: 10, // Process 10 jobs at a time
        limiter: {
            max: 200, // Max 200 jobs per 10 seconds (increased to handle load)
            duration: 10000
        },
        stalledInterval: 120000, // Check for stalled jobs every 2 minutes
        maxStalledCount: 3, // Allow job to stall 3 times before failing
        lockDuration: 180000, // Lock jobs for 3 minutes
        lockRenewTime: 90000, // Renew lock every 90 seconds
        autorun: true, // Automatically start processing
        settings: {
            backoffStrategies: {
                exponential: (attemptsMade) => Math.min(Math.pow(2, attemptsMade) * 5000, 30000)
            },
            // Automatically retry stalled jobs
            stalledInterval: 120000,
            maxStalledCount: 3
        }
    });

    worker.on('completed', (job) => {
        console.log(`Job ${job.id} completed successfully`);
        
        // Log completion with delay information
        if (job.data.delayInfo) {
            const { delayType, delay } = job.data.delayInfo;
            const actualDelay = Date.now() - job.timestamp;
            console.log(`Job ${job.id} (${delayType} delay: ${delay}ms) completed with actual delay: ${actualDelay}ms`);
        }
    });

    worker.on('failed', async (job, error) => {
        console.error(`Job ${job.id} failed (attempt ${job.attemptsMade}/${job.opts.attempts}):`, error.message);
        
        // Log failure with delay information
        if (job.data.delayInfo) {
            const { delayType, delay } = job.data.delayInfo;
            console.error(`Failed job ${job.id} had ${delayType} delay: ${delay}ms`);
        }
        
        // Log if job will retry
        if (job.attemptsMade < job.opts.attempts) {
            const nextRetryIn = Math.pow(2, job.attemptsMade) * 5000; // Exponential backoff
            console.log(`⏳ Job ${job.id} will retry in ${nextRetryIn}ms (attempt ${job.attemptsMade + 1}/${job.opts.attempts})`);
        } else {
            console.error(`❌ Job ${job.id} exhausted all ${job.opts.attempts} attempts and will not retry`);
        }
    });

    worker.on('stalled', (job) => {
        console.warn(`⚠️ Job ${job.id} has stalled (stall count: ${job.stalledCount || 1}/${job.opts.maxStalledCount || 3})`);
        console.warn(`Job will ${(job.stalledCount || 1) < (job.opts.maxStalledCount || 3) ? 'be retried' : 'FAIL after next stall'}`);
    });

    worker.on('error', (error) => {
        console.error('Worker error:', error);
    });

    return worker;
}

export { initializeWorker }; 

this is my code

[manast]

Clearly sendText, sendPollor sendImage are not completing in the expected time and are thus throwing a timeout exception.