@monque/core

MongoDB-backed job scheduler with atomic locking, exponential backoff, and cron scheduling.

[!WARNING] This package is currently in pre-release. The public API may change between releases. Expect breaking changes until 1.0.0.

Installation

Using Bun:

bun add @monque/core mongodb

Or using npm/yarn/pnpm:

npm install @monque/core mongodb
yarn add @monque/core mongodb
pnpm add @monque/core mongodb

Usage

import { Monque } from '@monque/core';
import { MongoClient } from 'mongodb';

const client = new MongoClient('mongodb://localhost:27017');
await client.connect();

const monque = new Monque(client.db('myapp'), {
  collectionName: 'jobs',
  pollInterval: 1000,
  maxRetries: 10,
  defaultConcurrency: 5,
});

await monque.initialize();

// Register workers
monque.register('send-email', async (job) => {
  await sendEmail(job.data.to, job.data.subject);
});

// Start processing
monque.start();

// Enqueue jobs
await monque.enqueue('send-email', { to: 'user@example.com', subject: 'Hello' });

// Schedule recurring jobs
await monque.schedule('0 9 * * *', 'daily-report', { type: 'summary' });

// Graceful shutdown
await monque.stop();

API

new Monque(db, options?)

Creates a new Monque instance.

Options:

• collectionName - MongoDB collection name (default: 'monque_jobs')
• pollInterval - Polling interval in ms (default: 1000)
• maxRetries - Max retry attempts (default: 10)
• baseRetryInterval - Base backoff interval in ms (default: 1000)
• shutdownTimeout - Graceful shutdown timeout in ms (default: 30000)
• defaultConcurrency - Jobs per worker (default: 5)
• lockTimeout - Stale job threshold in ms (default: 1800000)
• recoverStaleJobs - Recover stale jobs on startup (default: true)

Methods

• initialize() - Set up collection and indexes
• enqueue(name, data, options?) - Enqueue a job
• now(name, data) - Enqueue for immediate processing
• schedule(cron, name, data) - Schedule recurring job
• register(name, handler, options?) - Register a worker
• start() - Start processing jobs
• stop() - Graceful shutdown
• isHealthy() - Check scheduler health

Events

monque.on('job:start', (job) => { /* job started */ });
monque.on('job:complete', ({ job, duration }) => { /* job completed */ });
monque.on('job:fail', ({ job, error, willRetry }) => { /* job failed */ });
monque.on('job:error', ({ error, job? }) => { /* unexpected error */ });
monque.on('stale:recovered', ({ count }) => { /* stale jobs recovered */ });

Development

Running Tests

# Run tests once (fresh container each time)
bun run test

# Run tests in watch mode with container reuse (faster iteration)
bun run test:dev

# Or enable reuse globally in your shell profile
export TESTCONTAINERS_REUSE_ENABLE=true
bun run test:watch

When TESTCONTAINERS_REUSE_ENABLE=true, the MongoDB testcontainer persists between test runs, significantly speeding up local development. Ryuk (the testcontainers cleanup daemon) remains enabled as a safety net for orphaned containers.

To manually clean up reusable containers:

docker ps -q --filter label=org.testcontainers=true | while read -r id; do docker stop "$id"; done

License

ISC