Workers

A worker is a handler registered for a job name. Monque calls it when a matching job is due and a concurrency slot is available.

Quick Reference

Method / Type	Description
register()	Register a job handler
WorkerOptions	Worker concurrency, retries, payload validation, and replacement
start()	Start processing jobs
stop()	Graceful shutdown
pause() / resume()	Pause or resume local execution
now()	Enqueue for immediate execution
MonqueOptions.workerConcurrency	Global concurrency default

Registering a Worker

Register a handler for a job name with register():

monque.register("job-name", async (job) => {
  // Your processing logic
});

With Type Safety

Define the expected data type for better IDE support and compile-time checks:

interface ImageProcessJob {
  imageUrl: string;
  width: number;
  height: number;
  format: "jpeg" | "png" | "webp";
}

monque.register<ImageProcessJob>("process-image", async (job) => {
  const { imageUrl, width, height, format } = job.data;
  // TypeScript knows all property types
  await processImage(imageUrl, { width, height, format });
});

With Payload Validation

Pass a Standard Schema compatible schema to validate persisted input before your handler runs. For example, using Zod:

import { z } from "zod";

const schema = z.object({ email: z.email() });
monque.register(
  "send-email",
  async (job) => {
    await sendEmail(job.data.email);
  },
  { schema },
);

Monque accepts the standard interface from @standard-schema/spec; install your preferred compatible validator separately. Both synchronous and asynchronous schemas are supported. The handler’s payload type is inferred from the schema’s output, including transformations.

Validation happens on every claimed attempt, including jobs enqueued by another producer. Enqueueing does not validate against a local worker’s schema. The stored payload and lifecycle events retain the original input; only the handler receives transformed output. Retries validate that original input again.

If the schema returns validation issues, the job fails once with PayloadValidationError, without invoking the handler or automatically retrying. The job:fail error includes the schema’s issues, including their paths. Unexpected exceptions thrown by a validator follow the worker’s normal retry policy.

With Worker Options

Pass WorkerOptions to set concurrency or replace a handler:

monque.register(
  "heavy-computation",
  async (job) => {
    await compute(job.data);
  },
  {
    concurrency: 2, // Only 2 concurrent jobs (overrides workerConcurrency)
  },
);

Worker Options

Option	Type	Default	Description
concurrency	number	5 (from workerConcurrency)	Maximum concurrent jobs for this worker
replace	boolean	false	Replace an existing worker with the same name instead of throwing
schema	StandardSchemaV1	None	Validate input and pass the schema output to the handler

Workers also accept retry overrides.

Note

Each worker type maintains its own concurrency limit. A scheduler with 3 workers at concurrency 5 each could process up to 15 jobs simultaneously.

Worker Lifecycle

Pausing Local Processing

Pause one worker or all execution in this scheduler instance:

monque.pause("send-email");
monque.isPaused("send-email"); // true
monque.resume("send-email");

monque.pause();
monque.isPaused(); // true
monque.resume();

Running executions continue, including their heartbeats. Enqueueing, stale recovery, and other scheduler instances remain active. A claim that finishes arriving after a pause is released without starting its handler. Resuming wakes local polling promptly; it does not restart the scheduler.

Pauses are local memory state and are lost when the process exits. You can pause before initialization or worker registration, and repeated calls are safe. A stop/start cycle on the same instance preserves pauses. Resuming the whole instance preserves individually paused workers; resuming a worker does not override a whole-instance pause. isPaused(name) reports the effective state, while isPaused() reports the instance pause. getProcessingState(name?) also returns the scheduler’s instanceId and globallyPaused, so management clients can distinguish a worker pause from an instance-wide pause. The same method is available on Ts.ED’s MonqueService.

Use stop() when you need to drain and shut down. A paused, connected scheduler stays healthy. To stop execution across a fleet, pause each consumer instance.

Replacing a Handler

Pass { replace: true } to register a new handler for an existing job name:

monque.register("send-email", updatedEmailHandler, { replace: true, concurrency: 3 });

Running jobs finish with their original handler and remain visible in Queue Views. Subsequent polls count them against the replacement’s concurrency limit. A poll already in progress may finish acquiring work with the previous registration. Graceful shutdown also waits for jobs started by the previous handler.

1. Job Acquisition

When start() is called, Monque begins looking for jobs:

1. Change Streams (if available) provide near real-time notifications: triggering targeted, debounced polls for specific workers
2. Safety polling runs every safetyPollInterval (default 30 s) as a backup when Change Streams are active
3. If Change Streams are unavailable, regular polling runs every pollInterval (default 1 s)

2. Atomic Claim

Monque claims a due pending job before calling its handler. This prevents two scheduler instances from claiming the same pending job at the same time. See Atomic Claim Pattern for multi-instance deployment.

3. Execution

Your handler function is called with the full job document:

monque.register("example", async (job) => {
  console.log(job._id); // ObjectId
  console.log(job.name); // 'example'
  console.log(job.data); // Your payload
  console.log(job.failCount); // Previous failures
});

4. Completion or Failure

• Success: One-time jobs become completed; recurring jobs return to pending for their next cron time.
• Error thrown: Job retried or set to failed (see Retry & Backoff)

Error Handling

Automatic Retry

If your handler throws or returns a rejected promise, the job is automatically retried:

monque.register("flaky-api", async (job) => {
  const response = await fetch(job.data.apiUrl);
  if (!response.ok) {
    throw new Error(`API returned ${response.status}`);
  }
  // Job will be retried with exponential backoff
});

Permanent Failure

Throw NonRetryableError when retrying cannot help, such as when a resource has been permanently deleted. The job becomes failed immediately and emits job:fail with willRetry: false. Ordinary errors, including other MonqueError instances, follow the configured retry policy.

import { NonRetryableError } from "@monque/core";

monque.register<{ resourceId: string }>("check-external-resource", async (job) => {
  const exists = await checkResourceExists(job.data.resourceId);
  if (!exists) {
    throw new NonRetryableError("Resource permanently deleted");
  }

  await processResource(job.data.resourceId);
});

Tip

Returning normally records a successful run. Throw NonRetryableError when you need to preserve a failure for inspection. After correcting the cause, you can use retryJob() to retry it manually. For recurring jobs, permanent failure stops the schedule until you retry it.

Concurrency Control

Global Default

Set in MonqueOptions:

const monque = new Monque(db, {
  workerConcurrency: 10, // All workers default to 10 concurrent jobs
});

Per-Worker Override

// Low-priority background tasks: higher concurrency
monque.register("log-analytics", handler, { concurrency: 20 });

// Resource-intensive tasks: lower concurrency
monque.register("video-transcode", handler, { concurrency: 2 });

// External API calls: limit simultaneous requests
monque.register("api-sync", handler, { concurrency: 5 });

Instance-level Concurrency

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

const monque = new Monque(db, {
  instanceConcurrency: 10, // Instance processes max 10 jobs total
  workerConcurrency: 5, // Each worker defaults to 5 concurrent jobs
});

// With 3 workers at concurrency 5, normally 15 jobs could run simultaneously.
// With instanceConcurrency: 10, only 10 jobs run at any time across all workers.
monque.register("emails", emailHandler); // Up to 5 concurrent
monque.register("notifications", notifyHandler); // Up to 5 concurrent
monque.register("webhooks", webhookHandler); // Up to 5 concurrent
// But total across all three never exceeds 10

When workers share an instanceConcurrency limit, polling resumes after the worker that last acquired a job. A busy worker cannot keep taking every new slot ahead of later-registered workers. Notifications for one job name also check other workers waiting for shared capacity. Running jobs are not interrupted, and jobs within each worker are still acquired in their scheduled order.

Tip

Set instanceConcurrency to fit the CPU, memory, and database connections available to one scheduler process.

Concurrency Strategy

Scenario	Recommended Concurrency
CPU-bound tasks	1-2 per core
I/O-bound tasks (database, API)	5-20
Memory-intensive tasks	Limited by available RAM
External API calls	Limit simultaneous requests; enforce time-based rate limits separately

When an execution finishes, waiting jobs can use the freed local slot without waiting for pollInterval, even when change streams are unavailable. Work written by other producers is discovered through change streams or, when unavailable, within pollInterval.

Multiple Workers

Register multiple workers for different job types:

// Email processing
monque.register<EmailJob>("send-email", async (job) => {
  await sendEmail(job.data);
});

// Image processing
monque.register<ImageJob>("resize-image", async (job) => {
  await resizeImage(job.data);
});

// Notification dispatch
monque.register<NotificationJob>("push-notification", async (job) => {
  await sendPushNotification(job.data);
});

Worker Registration Timing

Workers can be registered before or after start():

Before Start (Recommended)

// Register all workers first
monque.register('job-a', handlerA);
monque.register('job-b', handlerB);

// Then start processing
monque.start();

After Start

monque.start();

// Workers registered after start begin processing
// on the next poll cycle
monque.register('dynamic-worker', handler);

Caution

Registering a second worker with the same name throws WorkerRegistrationError. Pass { replace: true } to replace the handler, for example during a development reload.

Best Practices

1. Keep Handlers Focused

Keep unrelated tasks in separate handlers so they can be retried independently:

// ✅ Good - Single responsibility
monque.register("send-email", sendEmailHandler);
monque.register("update-stats", updateStatsHandler);

// ❌ Bad - Too much in one handler
monque.register("process-order", async (job) => {
  await sendConfirmationEmail();
  await updateInventory();
  await notifyShipping();
  await updateAnalytics();
});

2. Handle Partial Failures

Track progress so retries can resume unfinished work:

monque.register("sync-data", async (job) => {
  const { batchId } = job.data;

  // Track progress in external state
  const progress = await getProgress(batchId);

  // Resume from where we left off
  for (const item of progress.remaining) {
    await processItem(item);
    await updateProgress(batchId, item);
  }
});

A crash can occur between processItem and updateProgress, and concurrent executions can read the same progress. Each item operation must itself be idempotent, or its effect and progress update must commit in one transaction. See Idempotency patterns.

3. Log Strategically

Use job events for centralized logging:

monque.on("job:start", (job) => {
  logger.info("Job started", { jobId: job._id, name: job.name });
});

monque.on("job:complete", ({ job, duration }) => {
  logger.info("Job completed", {
    jobId: job._id,
    name: job.name,
    duration,
  });
});

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

Next Steps

• Scheduling - Configure recurring jobs
• Retry & Backoff - Understand failure handling
• Heartbeat Mechanism - Monitor heartbeats and recovery

API Reference

• Monque.register() - Register a worker handler
• Monque.start() - Start the scheduler
• Monque.stop() - Graceful shutdown
• Monque.now() - Immediate job execution
• WorkerOptions - Worker configuration
• MonqueOptions - Scheduler options
• Job - Job document passed to handlers
• MonqueEventMap - Available events