Production Checklist

Before deploying, check how your workers handle retries, shutdown, and jobs left running after a crash.

Numeric scheduler options must be finite and non-negative. Concurrency, retry counts, and payload byte limits must be safe integers; durations accept fractional milliseconds. Polling, heartbeat, and retention check intervals must be at least 1 ms; these timers and shutdownTimeout cannot exceed 2,147,483,647 ms (about 24.8 days). Invalid options throw MonqueError when constructing Monque or registering a worker, including through Ts.ED.

Zero remains valid for limits and durations outside those repeating timers: for example, statsCacheTtlMs: 0 disables statistics caching, zero concurrency prevents new claims, and a retention age of zero makes terminal jobs eligible at the next cleanup.

Quick Reference

Topic	Key Point
Idempotency	Workers must handle duplicate execution safely
Deduplication	Use `uniqueKey` to prevent duplicate jobs
Replica Set	Required for Change Streams
Shutdown	Configure `shutdownTimeout` appropriately
Recovery	Choose absolute locks or renewable leases; allow for slow runs and renewal delays

1. Design for Idempotency

Jobs may be executed more than once due to:

Failed attempts retry with exponential backoff.
Recovery can return jobs with expired claims to pending, even while the original handler is still running.
A network failure can prevent Monque from recording a completed run.

Idempotency Patterns

For an external side effect, pass a stable business-operation key to a provider that supports idempotency. The following paymentProvider is an application adapter: it must forward the key to the provider and handle its response when an identical request is already in flight. A local status check before calling the provider cannot prevent two handlers from charging at once, or a second charge after a lost response.

monque.register("charge-payment", async (job) => {
  const { paymentId, amount, currency } = job.data;

  await paymentProvider.charge(
    { paymentId, amount, currency },
    { idempotencyKey: `charge:${paymentId}` },
  );
});

Reuse the same key and immutable payment details on every attempt. Keep retries within the provider’s key-retention period; after it expires, reconcile the payment before sending another charge. A new Job ID or claimId is not a stable payment key.

For MongoDB-only effects, commit the deduplication record and the mutations in one transaction. This example assumes validated string account and transfer identifiers, plus a positive amount in integer minor units. Use collections from the same MongoClient that owns the session. Create the collections and unique indexes before starting workers:

const transfers = db.collection("transfers");
const accounts = db.collection("accounts");
await transfers.createIndex({ transferId: 1 }, { unique: true });
await accounts.createIndex({ accountId: 1 }, { unique: true });

monque.register("transfer-funds", async (job) => {
  const { fromAccount, toAccount, amount, transferId } = job.data;
  const session = client.startSession();

  try {
    await session.withTransaction(
      async () => {
        const existing = await transfers.findOne({ transferId }, { session });
        if (existing) {
          if (
            existing.fromAccount !== fromAccount ||
            existing.toAccount !== toAccount ||
            existing.amount !== amount
          ) {
            throw new Error("Transfer identifier reused with different details");
          }
          return;
        }

        // The unique index prevents concurrent transactions from applying this twice.
        await transfers.insertOne(
          { transferId, fromAccount, toAccount, amount, status: "completed" },
          { session },
        );
        const debit = await accounts.updateOne(
          { accountId: fromAccount, balance: { $gte: amount } },
          { $inc: { balance: -amount } },
          { session },
        );
        const credit = await accounts.updateOne(
          { accountId: toAccount },
          { $inc: { balance: amount } },
          { session },
        );
        if (debit.matchedCount !== 1 || credit.matchedCount !== 1) {
          throw new Error("Account missing or insufficient funds");
        }
      },
      { readConcern: { level: "snapshot" }, writeConcern: { w: "majority" } },
    );
  } finally {
    await session.endSession();
  }
});

Account identifiers must also be unique. Keep completed transfer records for as long as the same transfer can be submitted again. If concurrent attempts race to insert the record, a duplicate-key error can escape the transaction; let the Job retry and read the committed record. Do not treat every duplicate-key error as a successful transfer. The record and both balance changes either commit together or roll back together.

MongoDB transactions require a replica set or sharded cluster. The driver can rerun the transaction callback, so keep external calls outside it and pass session to every operation. See the Node.js transaction guide.

See Idempotent Operations for more patterns.

2. Use `uniqueKey` for Deduplication

Prevent duplicate jobs from being enqueued using uniqueKey:

// Only one sync job per user at a time
await monque.enqueue(
  "sync-user",
  { userId: "123" },
  {
    uniqueKey: "sync-user-123",
  },
);

Always Use `uniqueKey` for Scheduled Jobs

// ✅ Safe for app restarts and scaling
await monque.schedule(
  "0 * * * *",
  "hourly-report",
  {},
  {
    uniqueKey: "hourly-report-singleton",
  },
);

Without uniqueKey, calling schedule() multiple times (e.g., on each app restart) creates duplicate scheduled jobs.

3. Use a Replica Set for Change Streams

Without a replica set:

Monque falls back to polling only
Job pickup latency increases to pollInterval (default: 1000ms)
Real-time job notifications are unavailable

Local Development with Replica Set

Use Docker Compose for local development:

# docker-compose.yml
services:
  mongo:
    image: mongo:7
    command: ["--replSet", "rs0", "--bind_ip_all"]
    ports:
      - "27017:27017"

Initialize the replica set:

docker exec -it <container> mongosh --eval "rs.initiate()"

4. Configure Graceful Shutdown

Give running handlers time to finish before your application exits:

const monque = new Monque(db, {
  shutdownTimeout: 30000, // Wait up to 30s for jobs to finish
});

process.on("SIGTERM", async () => {
  console.log("Shutting down...");
  await monque.stop(); // Waits for in-progress jobs
  await client.close();
  process.exit(0);
});

Kubernetes Considerations

Set terminationGracePeriodSeconds higher than shutdownTimeout:

spec:
  terminationGracePeriodSeconds: 45 # > shutdownTimeout (30s)
  containers:
    - name: worker
      lifecycle:
        preStop:
          exec:
            command: ["sh", "-c", "sleep 5"]

5. Tune `lockTimeout` for Your Workload

By default, lockTimeout is an absolute limit measured from lockedAt. With recoverStaleJobs: true, initialization returns expired processing jobs to pending. Heartbeats update liveness information but do not extend an absolute lock. Expiry alone does not trigger recovery on a running scheduler that has leases disabled.

Set the timeout above your longest expected run, with headroom for delays. Twice that duration can be a starting point, but it cannot guarantee that a slow handler will finish before recovery. Another initializing scheduler can recover an expired claim while its original handler still runs.

const monque = new Monque(db, {
  lockTimeout: 3600000, // Absolute timeout: 1 hour
  heartbeatInterval: 30000, // Liveness update; does not extend lockTimeout
});

Renewable leases

Enable leaseDuration for jobs with variable runtimes or recovery without restarting a surviving scheduler. Upgrade every scheduler sharing the collection before enabling it. Older versions do not recognize renewed leases.

const monque = new Monque(db, {
  heartbeatInterval: 30000,
  leaseDuration: 90000, // Allow several heartbeats of renewal headroom
  recoverStaleJobs: true,
});

Heartbeats renew leaseExpiresAt using MongoDB’s clock. An expired lease cannot renew or record a late result. A running scheduler with leases and recovery enabled checks for expired claims after each heartbeat, including legacy absolute locks. Existing claims keep their original mode, so retain a suitable lockTimeout during migration.

Allow for renewal delays, database outages, and event-loop pauses when sizing leases. Recovery does not cancel the original handler or undo its side effects. See Heartbeat Mechanism for recovery and shutdown details.

Diagnose delayed or duplicate execution

Symptom	Check
Overlapping executions	Absolute lock expired during a slow run, or a lease could not renew
Jobs remain in `processing` after a crash	Recovery disabled, claim not expired, or no initialization or lease-enabled recovery pass
Slow recovery	Claim deadline, heartbeat interval, database availability, and event-loop delays

6. Set Appropriate Concurrency

Configure workerConcurrency and per-worker concurrency based on your workload:

const monque = new Monque(db, {
  workerConcurrency: 5,
});

// CPU-bound: low concurrency
monque.register("video-transcode", handler, { concurrency: 2 });

// I/O-bound: higher concurrency
monque.register("api-sync", handler, { concurrency: 10 });

// Cap simultaneous requests; enforce the provider's time-based quota separately
monque.register("stripe-webhook", handler, { concurrency: 5 });

Concurrency limits in-flight work in one Scheduler Instance. It does not enforce a requests-per-second or requests-per-minute limit: fast jobs can start many times per second, and each additional instance has its own concurrency allowance. Monque has no built-in shared rate limit. Use a limiter shared by all callers before each external request, including retries inside a handler.

Instance-level Throttling

Use instanceConcurrency to limit the total jobs processed by a single instance across all workers:

const monque = new Monque(db, {
  instanceConcurrency: 20, // Instance-wide limit
  workerConcurrency: 10, // Per-worker default
});

Choose a limit that fits the CPU, memory, and database connections available to one process.

See Concurrency Strategy for guidelines.

7. Consider Producer-Consumer Architecture

Run API servers and workers in separate processes when they need different resources or deployment schedules:

Benefits

You can add worker capacity or deploy new handlers without redeploying the API servers. Workers also get their own CPU and memory, reducing contention with HTTP requests.

Implementation

Ts.ED Integration:

// API Server (Producer-only)
@Configuration({
  monque: {
    dbFactory: async () => client.db("myapp"),
    disableJobProcessing: true, // Only enqueue, don't process
  },
})
export class ApiServer {}

// Worker Server (Consumer)
@Configuration({
  monque: {
    dbFactory: async () => client.db("myapp"),
    instanceConcurrency: 20, // Limit total concurrent jobs
  },
})
export class WorkerServer {}

Core Package:

// Producer: Initialize but don't start
const producer = new Monque(db);
await producer.initialize();
// Don't call start() - just use enqueue()

// Consumer: Full lifecycle
const consumer = new Monque(db, { instanceConcurrency: 20 });
consumer.register("my-job", handler);
await consumer.initialize();
await consumer.start();

See Ts.ED Producer-only Mode for details.

8. Monitor Job Events

Log job failures and recovery events:

// Track failures
monque.on("job:fail", ({ job, error, willRetry }) => {
  logger.error("Job failed", {
    jobId: job._id,
    name: job.name,
    error: error.message,
    willRetry,
    failCount: job.failCount,
  });

  if (!willRetry) {
    alerting.notify(`Job ${job.name} permanently failed`);
  }
});

// Monitor stale recovery
monque.on("stale:recovered", ({ count }) => {
  if (count > 0) {
    metrics.increment("monque.stale_jobs_recovered", count);
    logger.warn(`Recovered ${count} stale jobs`);
  }
});

// Track Change Stream health
monque.on("changestream:fallback", ({ reason }) => {
  logger.warn("Change Streams unavailable, using polling", { reason });
});

See MonqueEventMap for all available events.

9. Configure Job Retention

Prevent unbounded database growth by configuring retention:

const monque = new Monque(db, {
  jobRetention: {
    completed: 24 * 60 * 60 * 1000, // 24 hours
    failed: 7 * 24 * 60 * 60 * 1000, // 7 days
    cancelled: 24 * 60 * 60 * 1000, // 24 hours
    interval: 60 * 60 * 1000, // Cleanup every hour
  },
});

10. Ensure Index Permissions

Monque creates MongoDB indexes during initialize(). Grant index creation permissions to that connection, or use skipIndexCreation: true and manage the full index set yourself. Precreating indexes alone does not skip index creation. See Indexes for listing and retention indexes.

Priority claims and deadline discovery need separate compound indexes. Keep both when managing indexes yourself, in addition to the other Monque indexes:

// Next due Job by priority, scheduled time, and identifier.
db.jobs.createIndex({ name: 1, status: 1, priority: -1, nextRunAt: 1, _id: 1 });
// Earliest deadline per Job Name and future wakeups.
db.jobs.createIndex({ name: 1, status: 1, nextRunAt: 1, claimedBy: 1 });

Use your configured collection name and preserve its collation. The priority index does not replace the deadline index. Existing deadline indexes remain usable; upgrading with automatic index creation adds the priority index without dropping them. With managed indexes, install the new index as part of the upgrade.

Index cost depends on the backlog. A priority scan can skip a group of future Jobs sharing one priority, but many distinct future priorities increase the index work. MongoDB may instead scan the due range through a deadline index and sort its candidates by priority. No one plan is cheapest for every distribution.

The MongoDB 8 integration test measures scheduler queries. After an urgent Job was claimed, the test collection held 100 overdue Jobs at priority -1 and 1,000 Jobs due far in the future.

When the future Jobs shared priority 100, the next claim examined 2 keys and 1 document. With 1,000 distinct future priorities, MongoDB sorted the due Jobs to select the first one, examining 100 keys and 100 documents. With skipIndexCreation: true and only the existing deadline index, that same distinct-priority case also selected the correct due Job, examining 100 keys and 100 documents. Deadline discovery examined 1 key and no documents in all three cases.

These results describe the test data; costs vary with your backlog. Inspect execution statistics on your own data. To reproduce the test, run this from packages/core:

vp test run tests/integration/priority-query.test.ts --reporter verbose --silent false

Initialization also sets missing Job priorities to 0, even when index creation is disabled. Grant write access for this automatic upgrade. It preserves existing priorities, timestamps, lifecycle state, and ownership metadata. Upgrade all producers and claiming Scheduler Instances before relying on consistent priority ordering.

11. Configure Management and Dashboard Access

If you serve the dashboard, mount authentication middleware before both the dashboard and Management routers. Set API permissions with authorize or readOnly, and redact payload fields that operators should not see. Protect the OpenAPI route with the same middleware.

Set a refresh interval appropriate for the number of open dashboards and your database load. Configure compression at the proxy or in Express, and preserve dashboard mount paths for direct links. See Dashboard configuration.

Pre-Deployment Checklist

API Reference

Monque - Main scheduler class
MonqueOptions - Configuration options
MonqueEventMap - Available events
EnqueueOptions - Job enqueue options
WorkerOptions - Worker configuration