@platformatic/job-queue

You can attempt to cancel a job using cancel(id). The behavior depends on the current state of the job:

• Not Found: Returns { status: 'not_found' }.
• Completed: Returns { status: 'completed' } (cannot cancel a finished job).
• Processing: Returns { status: 'processing' }. Jobs currently being handled by a worker cannot be cancelled mid-processing.
• Queued: The job is cancelled. The library performs an atomic MULTI block to:
  1. Delete the job entry from the {prefix}:jobs hash using HDEL.
  2. Publish a {id}:cancelled event to {prefix}:events.

Note on Implementation: The library does not perform an $O(n)$ removal from the Redis LIST directly. Instead, it deletes the job from the jobs hash. When a worker eventually picks up the job from the queue, it checks the jobs hash; if the entry is missing, the worker skips the job and removes it from its processing queue.

// Example cancellation
const status = await cancel('job-id');
if (status.status === 'cancelled') {
  console.log('Job was successfully cancelled');
}

The enqueueAndWait(id, payload, { timeout, resultTTL? }) method is an optimized pattern for waiting on a job result without polling. It uses Redis Pub/Sub for minimal latency.

The Workflow:

1. Subscription: The client subscribes to {prefix}:notifications:{id} before enqueuing to prevent race conditions.
2. Enqueue: The client calls enqueue(id, payload).
3. Wait: The client waits for a notification on the subscription.
   • If the job is already completed, it returns the result immediately.
   • If the job is a duplicate, it waits on the subscription.
   • If the job is queued, it blocks until a notification is received or the timeout is reached.
4. Resolution:
   • 'completed' notification: The client performs a GET to fetch the result and returns.
   • 'failed' notification: The client fetches the error and throws it.
   • timeout: The client throws a TimeoutError.

Key Optimization: The Pub/Sub mechanism only carries the notification (not the full payload), allowing the client to fetch the result via a single GET after the notification arrives, which handles large payloads efficiently.

// Example Request/Response usage
try {
  const result = await enqueueAndWait('job-id', { data: 'foo' }, { 
    timeout: 5000, 
    resultTTL: 30000 
  });
  console.log('Job finished:', result);
} catch (err) {
  if (err instanceof TimeoutError) {
    console.error('Job timed out');
  } else {
    console.error('Job failed:', err);
  }
}

The library uses an event-driven architecture and a background Reaper to recover jobs that were interrupted (e.g., due to a worker crash).

Event-Based Detection

All state changes are published to {prefix}:events. The Reaper subscribes to these events and maintains timers for jobs in the processing state.

The Reaper Process

1. Registration: When a worker starts, it registers itself in the {prefix}:workers set.
2. Monitoring: The Reaper monitors {id}:processing events.
3. Timer: For every processing event, the Reaper starts a timer based on the visibilityTimeout.
4. Requeue: If the timer fires and the job is still marked as processing in the {prefix}:jobs hash, the Reaper triggers recoverStalledJob(id, workerId).

Atomic Recovery

The recoverStalledJob function uses a Lua script to ensure atomicity:

• It checks if the job state in {prefix}:jobs starts with processing:.
• If yes, it moves the message from the worker's processing queue back to the main queue (LREM + LPUSH).
• It updates the job state to queued.
• It publishes an {id}:stalled event.

Recovery Timing

• Stalled Jobs: Recovery occurs approximately after the visibilityTimeout expires.
• Graceful Shutdown: If a worker shuts down gracefully, it is responsible for requeueing its own jobs immediately.

When building with @platformatic/job-queue, be aware of the following architectural constraints and requirements:

• Idempotency is mandatory: The library provides at-least-once delivery. Jobs may be re-processed due to crashes, lease expirations, or network issues. You must ensure your job handlers are idempotent (processing the same job twice results in the same state as processing it once).
• FIFO only: The queue does not support priority levels; all jobs are processed in First-In-First-Out order. For priority support, you must implement multiple queues.
• No delayed jobs: Jobs are processed immediately upon being added. If you need delayed execution, you must use an external scheduler.
• Single Redis instance: The library does not have built-in support for Redis clustering. To use Redis Cluster, you must use a proxy or a compatible client.
• Cancellation mechanism: Job removal is handled via cancellation marking rather than direct list removal to avoid $O(n)$ performance penalties.

When you call enqueue(id, payload, { resultTTL? }), the library performs an atomic operation to ensure reliability:

1. Deduplication Check: It checks if the job ID already exists in the {prefix}:jobs hash.
   • If the state is completed, it returns the cached result.
   • If the state is queued or processing, it returns a duplicate status.
   • If the job does not exist or is failed, it proceeds to enqueue.
2. Atomic Enqueue: Using a Redis MULTI block, it:
   • Sets the job state in {prefix}:jobs to queued:{timestamp}.
   • Pushes the message to the {prefix}:queue list.
   • Publishes a {id}:queued event to {prefix}:events.

TTL Resolution: The resultTTL (retention time for the result) is resolved as options.resultTTL ?? config.resultTTL. This value is serialized into the message so that workers use the producer's preferred TTL.

Validation: resultTTL must be an integer representing milliseconds and must be > 0. Invalid values will trigger a validation error.

// Example enqueue call
enqueue('job-123', { task: 'process-image' }, { resultTTL: 60000 });
// resultTTL is 60,000ms (1 minute)