Jobs stuck in waiting state.

[apoorv569]

I am using chokidar to monitor a specific directory for specific files (audio/music files), which when finds any such file sends it as a job to a queue for processing,

function get_file_hash(filepath: string) {
    const file_buffer = fs.readFileSync(filepath);

    const hash_sum = createHash('sha256');

    hash_sum.update(file_buffer);

    return hash_sum.digest('hex');
}

async function process_and_add_file_to_queue(file: string) {
    const file_hash = get_file_hash(file);

    const exisiting_processed_file = await get_processed_file_by_hash(file_hash);

    if (exisiting_processed_file) {
        logger.info(`File ${file} is already processed, skipping`);

        return;
    }

    await metadata_extractor_queue.add("extract_metadata", { file });

    const processed_file = await insert_processed_file({
        filepath: file,
        file_hash
    });

    if (processed_file) {
        logger.info(`File ${file} added to queue and marked as processed`);
    }
}

const audio_file_extensions = ['.mp3', '.wav', '.flac', '.ogg', '.aac', '.m4a'];

export function start_fs_watcher() {
    const watcher = chokidar.watch(DATA_DIRECTORY, {
        persistent: true,
        ignoreInitial: false,
        awaitWriteFinish: true,
        ignored: (path, stats) => stats?.isFile() && !audio_file_extensions.some((ext) => path.endsWith(ext)),
    });

    watcher
        .on('ready', async () => {
            logger.info("Initial filesystem scan complete. Ready for changes")
        })
        .on('error', async (error) => {
            logger.error(`Watcher error: ${error}`)
        })
        .on('add', async (file) => {
            try {
                await process_and_add_file_to_queue(file);
            } catch (error) {
                logger.error(error.message);
            }
        })
        .on('change', async (file) => {
            try {
                await process_and_add_file_to_queue(file);
            } catch (error) {
                logger.error(error.message);
            }
        })
        .on('unlink', async (file) => {
            logger.warn(`File ${file} has been removed`);
        });
}

here is my queue and worker setup,

const queue_opts = {
    connection,
    defaultJobOptions: {
        attempts: 3,
        backoff: {
            type: 'exponential',
            delay: 1000
        }
    }
};

const metadata_extractor_queue_name = "metadata_extractor_queue";
const save_metadata_to_db_queue_name = "save_metadata_to_db_queue";

export const metadata_extractor_queue = new Queue(metadata_extractor_queue_name, queue_opts);
export const save_metadata_to_db_queue = new Queue(save_metadata_to_db_queue_name, queue_opts);

const metadata_extractor_worker = new Worker(metadata_extractor_queue_name, async (job) => {
    const file = job.data.file;

    const metadata = await extract_metadata(file);

    if (metadata) {
        await save_metadata_to_db_queue.add('save_metadata', { metadata });
    }
}, { connection });

const save_metadata_to_db_worker = new Worker(save_metadata_to_db_queue_name, async (job) => {
    const metadata = job.data.metadata;

    if (metadata) {
        await save_metadata([metadata]);
    }
}, { connection });

const queues = [
    metadata_extractor_queue,
    save_metadata_to_db_queue
];

const workers = [
    metadata_extractor_worker,
    save_metadata_to_db_worker
];

for (const worker of workers) {
    worker.on('active', (job) => {
        logger.info(`⚙️ Job ${job.name} with id ${job.id} is being processed with ${worker.name} worker`);
    });

    worker.on('progress', (job) => {
        logger.info(`⏳ Job ${job.name}: ${job.progress}% done`);
    });

    worker.on('completed', (job) => {
        logger.info(`✅ Job ${job.name} with id ${job.id} completed via ${worker.name} worker`);
    });

    worker.on('failed', (job, error) => {
        logger.error(`❌ Job ${job?.name} with id ${job?.id} failed from ${worker.name} worker: ${error.message}`);
    });
}

connection is,

export const connection = new IORedis(
    REDIS_PORT,
    REDIS_HOST,
    {
        maxRetriesPerRequest: null,
    }
);

to all this, I call this,

start_fs_watcher();

now I have a separate temp project for bull-board setup,

import { createBullBoard } from "@bull-board/api";
import { BullAdapter } from '@bull-board/api/bullAdapter.js';
import { ExpressAdapter } from "@bull-board/express";
import { Queue } from "bullmq";
import express from "express";
import IORedis from 'ioredis';

(async () => {
    const connection = new IORedis(
        6379,
        "localhost",
        {
            maxRetriesPerRequest: null,
        }
    );

    const queue_opts = {
        connection,
        defaultJobOptions: {
            attempts: 3,
            backoff: {
                type: 'exponential',
                delay: 1000
            }
        }
    };

    const queues_list = [
        "metadata_extractor_queue",
        "save_metadata_to_db_queue"
    ];

    const server_adapter = new ExpressAdapter();
    server_adapter.setBasePath("/");

    const queues = queues_list
          .map((qs) => new Queue(qs, queue_opts))
          .map((q) => new BullAdapter(q));

    const { addQueue, removeQueue, setQueues, replaceQueues } = createBullBoard({
        queues,
        serverAdapter: server_adapter
    });

    const app = express();

    app.use("/", server_adapter.getRouter());

    app.listen(8090, () => {
        console.log("App is running on http://localhost:8090");
    });
})();

by visiting this route I see that all jobs are successfully added to the metadata_extractor_queue but they stay in waiting state. I have tried waiting for a while, but they never to start processing.

[manast]

My recommendation here is that you start with a minimum test project with just one queue and one worker, when you get that working you can find your way up in where the issue is. Most likely the workers are either not connecting properly or they are listening to the wrong queue something like that, you will figure it out if you start simple and build from there.

[apoorv569]

Now I have this class I made for creating queue and worker at once, so there shouldn't be any naming issues or something.

export class TaskManager {
    private queues: Map<string, Queue> = new Map();
    private workers: Map<string, Worker> = new Map();

    public create_queue_and_worker(
        name: string,
        processor: Processor,
        queue_options?: QueueOptions,
        worker_options?: WorkerOptions) {
        let queue = this.queues.get(name);

        if (!queue) {
            queue = new Queue(name, queue_options);

            this.queues.set(name, queue);

            logger.info(`Created queue with name ${name}`);
        } else {
            logger.error(`Queue with name ${name} already exists`);
        }

        queue.on('error', (error) => {
            logger.error(`Queue ${queue.name} has failed: ${error}`);
        });

        let worker = this.workers.get(name);

        if (!worker) {
            worker = new Worker(name, processor, worker_options);

            this.workers.set(name, worker);

            logger.info(`Created worker for queue ${name}`);
        } else {
            logger.error(`Worker for queue ${name} already exists`);
        }

        worker.on('ready', () => {
            logger.info(`Worker ${worker.name} is ready`);
        });

        worker.on('error', (reason) => {
            logger.error(`Worker ${worker.name} failed: ${reason}`);
        });

        worker.on('stalled', (job_id, prev) => {
            logger.warn(`Worker ${worker.name} has stalled jobs ${job_id} ${prev}`);
        });

        worker.on('active', (job) => {
            logger.info(`⚙️ Job ${job.name} with id ${job.id} is being processed with ${worker.name} worker`);
        });

        worker.on('progress', (job) => {
            logger.info(`⏳ Job ${job.name}: ${job.progress}% done`);
        });

        worker.on('completed', (job) => {
            logger.info(`✅ Job ${job.name} with id ${job.id} completed via ${worker.name} worker`);
        });

        worker.on('failed', (job, error) => {
            logger.error(`❌ Job ${job?.name} with id ${job?.id} failed from ${worker.name} worker: ${error.message}`);
        });

        return { queue, worker };
    }
}

const task_manager = new TaskManager();

const queue_opts = {
    connection,
    prefix: "sonora_",
    defaultJobOptions: {
        attempts: 3,
        backoff: {
            type: 'exponential',
            delay: 1000
        }
    }
};

async function metadata_extract_job_handler(next_queue: Queue, job: Job) {
    const file = job.data.file;

    const metadata = await extract_metadata(file);

    if (metadata) {
        logger.info("Metadata exists.....................");

        if (next_queue) {
            await next_queue.add('save_metadata', { metadata });
            logger.info("Metadata send.....................");
        } else {
            logger.error("Unable to send metadata to next queue.......................");
        }
    } else {
        logger.error("Metadata does not exists.....................");
    }
}

async function save_metadata_to_db_job_handler(job: Job) {
    const metadata = job.data.metadata;

    if (metadata) {
        await save_metadata([metadata]);
    }
}

export const { queue: save_metadata_to_db_queue, worker: save_metadata_to_db_worker } =
    task_manager.create_queue_and_worker(
        "save_metadata_to_db",
        async (job) => {
            logger.info(`Processing job ${job.name} with id ${job.id}`);

            await save_metadata_to_db_job_handler(job);
        }, queue_opts, { connection });

export const { queue: metadata_extractor_queue, worker: metadata_extractor_worker } =
    task_manager.create_queue_and_worker(
        "metadata_extractor",
        async (job) => {
            logger.info(`Processing job ${job.name} with id ${job.id}`);

            await metadata_extract_job_handler(save_metadata_to_db_queue, job);
        }, queue_opts, { connection });

I never see any of these log statements being printed,

    if (metadata) {
        logger.info("Metadata exists.....................");
        await next_queue.add('save_metadata', { metadata });
        logger.info("Metadata send.....................");
    } else {
        logger.error("Metadata does not exists.....................");
    }

This is all I see from the logs,

[25-10-2024 07:19:54] INFO (26354): Created queue with name save_metadata_to_db
[25-10-2024 07:19:54] INFO (26354): Created worker for queue save_metadata_to_db
[25-10-2024 07:19:54] INFO (26354): Created queue with name metadata_extractor
[25-10-2024 07:19:54] INFO (26354): Created worker for queue metadata_extractor
[25-10-2024 07:19:54] INFO (26354): Worker metadata_extractor is ready
[25-10-2024 07:19:54] INFO (26354): Worker save_metadata_to_db is ready
[25-10-2024 07:20:28] INFO (26354): Initial filesystem scan complete. Ready for changes

[apoorv569]

The issue was prefix I had it set for queue but not for worker.