Change Streams

Monque uses MongoDB Change Streams to discover pending jobs without waiting for the next regular poll. Polling continues as a safety net. A job still needs a registered worker, an available concurrency slot, and a due nextRunAt before it can run.

Requirements

Use a replica set or sharded cluster with a MongoDB version compatible with the installed driver. See Installation for requirements and a local Docker setup.

Standalone MongoDB does not support Change Streams. Monque can still process jobs there using regular polling.

Configuration

const monque = new Monque(db, {
  pollInterval: 1_000,
  safetyPollInterval: 30_000,
});

await monque.initialize();
monque.start();

State	Periodic poll interval	Additional job discovery
Change Streams active	safetyPollInterval, default 30 seconds	Notifications for pending jobs and scheduled wakeups
Change Streams unavailable	pollInterval, default 1 second	Regular polling

At startup and each periodic poll, Monque reads the earliest pending deadline for each registered job name with available capacity, then atomically claims due jobs. Scheduling reads use the primary MongoDB node with local read concern to see current claim state. With the default binary collation, an idle collection needs one indexed discovery read instead of a separate claim attempt for every name. Startup discovery also schedules wakeups for future jobs already stored in MongoDB.

Collection collation is checked once. Collections with other collations retain per-worker atomic discovery and check all eligible workers for notifications so database name matching and shared concurrency fairness are preserved, including when stored and registered job names differ under the collection’s collation. Atomic discovery is also the fallback if collection metadata cannot be read.

Notifications for a job name trigger checks for its registered worker. Several notifications arriving together are batched for 100 milliseconds. Future jobs schedule a wakeup for their run time, retaining the relevant job names so unrelated workers need not be checked. Workers sharing an instance concurrency limit are checked in fair order across eligible names. Notifications that overlap an active poll retain their target names within a fixed memory budget; exceeding that budget requests full discovery. Dates more than approximately 25 days away are handled without timer overflow warnings or repeated immediate wakeups.

Runnable notifications claim jobs directly. When a claim finds no due job, discovery reads the next pending deadline for that name so persisted future jobs can use any remaining worker capacity without waiting for the safety poll.

Change Stream notifications contain scheduling metadata rather than job payloads. Claims that only change status to processing are filtered out. Worker handlers still receive the full claimed job and its payload.

Future wakeup tracking has a fixed memory budget for both deadlines and job names and prioritizes the nearest run times. A deadline with too many tracked names requests full discovery. If scheduling or rescheduling notifications exceed the deadline budget, later jobs are discovered through full polling. Discovery can then wait until the next safetyPollInterval (default 30 seconds) while Change Streams are active, or pollInterval while unavailable.

These intervals control scheduler job discovery. They are independent of dashboard refresh intervals. The dashboard reads the Management API; it does not open a MongoDB Change Stream.

Monitor connection state

Register event listeners before starting the scheduler:

monque.on("changestream:connected", () => {
  console.log("Change Stream connected");
});
monque.on("changestream:error", ({ error }) => {
  console.error("Change Stream error:", error.message);
});
monque.on("changestream:fallback", ({ reason }) => {
  console.warn("Using regular polling:", reason);
});
monque.on("changestream:closed", () => {
  console.log("Change Stream closed");
});

Monque attempts to reconnect after a stream error. After the retry limit is reached, it falls back to regular polling. Continue monitoring job failures and processing latency while the stream is unavailable.

isHealthy() reports whether the scheduler is initialized and started. It does not require an active Change Stream and is not a live database connectivity probe.

Measure pickup latency

Use a payload timestamp to measure how long a ready job waits before its handler starts:

monque.register<{ enqueuedAt: number }>("measure-latency", async (job) => {
  console.log("Pickup latency in ms:", Date.now() - job.data.enqueuedAt);
});

await monque.enqueue("measure-latency", { enqueuedAt: Date.now() });

Compare a standalone deployment using polling with a replica set using Change Streams. Available worker slots, handler duration, network latency, and database load all affect the result. Neither polling interval nor Change Stream notification time guarantees a maximum wait when workers are busy.

Troubleshooting

Symptom	Check
Stream never connects	Verify the deployment is a replica set or sharded cluster and the connection has the required MongoDB permissions.
Frequent reconnects	Inspect changestream:error events, network interruptions, and database availability.
Jobs wait despite an active stream	Check worker registration, concurrency limits, nextRunAt, and handler duration.
Jobs process but Health shows unavailable	Check that Management uses the started scheduler instance rather than a separate read-only instance.

Call await monque.stop() before closing the MongoDB client. This closes the stream and stops scheduler polling as part of shutdown.

See Workers for concurrency settings and Heartbeat Mechanism for crash recovery.