Distributed Worker

The Distributed Worker pattern separates publishers (API nodes) from consumers (worker nodes). Publishers use Queue() without a handler. Consumers use Worker() which only processes — no .add() method.

Working example → See a complete Distributed Worker implementation with video encoding, publisher/consumer split, and crash-safe processing: apps/backend/src/triggers/distributed-workers.ts

Architecture

┌──────────────────┐     ┌──────────┐     ┌──────────────────┐
│  API Server      │────▶│  Redis/  │◀────│  Worker Server   │
│  queue() — push  │     │ Postgres │     │  worker() — pull │
│  zero CPU cost   │     └──────────┘     │  heavy compute   │
└──────────────────┘                      └──────────────────┘

Publisher Queue (API Side)

Use queue() without a handler to create a publisher that only pushes jobs. It consumes zero CPU and polling overhead.

triggers/video-encode.ts

import { Queue } from "oqronkit";

export type VideoMetadata = {
  videoId: string;
  s3ResourceUri: string;
  codec: "h264" | "hevc" | "av1";
  bitrate: number;
};

// Publisher only — no handler, no polling engine
export const videoEncodeQueue = Queue<VideoMetadata, string>({
  name: "video-encode-topic",
});

Push jobs from your API routes:

api-routes.ts

import { videoEncodeQueue } from "./triggers/video-encode.js";

app.post("/api/upload", async (req, res) => {
  const job = await videoEncodeQueue.add(
    {
      videoId: `vid_${Date.now().toString(36)}`,
      s3ResourceUri: req.body.filePath,
      codec: "hevc",
      bitrate: 4500,
    },
    { idempotencyKey: `vid-${req.body.fileHash}` }, // Idempotency
  );
  res.json({ trackingId: job.id });
});

Consumer Worker (Worker Side)

Use Worker() to create a consumer. Workers have no .add() method — they only process.

triggers/video-worker.ts

import { Worker } from "oqronkit";
import type { VideoMetadata } from "./video-encode.js";

export const videoEncodeWorker = Worker<VideoMetadata, string>({
  topic: "video-encode-topic",

  concurrency: 2, // 2 heavy videos per server
  guaranteedWorker: true, // Heartbeat crash-safety
  heartbeatMs: 5_000,
  lockTtlMs: 30_000,

  retries: {
    max: 1,
    strategy: "fixed",
    baseDelay: 10_000,
  },

  hooks: {
    onSuccess: (job, finalUrl) => {
      console.log(`Video ${job.data.videoId} uploaded to: ${finalUrl}`);
    },
    onFail: (job, err) => {
      console.error(`Video ${job.data.videoId} failed:`, err);
    },
  },

  handler: async (ctx) => {
    const { videoId, codec, s3ResourceUri } = ctx.data;

    ctx.log("info", `Fetching source from ${s3ResourceUri}`);
    ctx.progress(10, "Downloading source");
    await download(s3ResourceUri);

    if (ctx.signal.aborted) {
      throw new Error("Transcoding cancelled");
    }

    ctx.progress(40, `Transcoding to ${codec}`);
    await transcode(s3ResourceUri, codec);

    ctx.progress(90, "Uploading chunks");
    await uploadToCDN(videoId);

    ctx.progress(100, "Done");
    return `https://cdn.example.com/videos/${videoId}.mp4`;
  },
});

Bootstrap

API and worker servers initialize Oqron independently:

api-server.ts

import { Oqron } from "oqronkit";
import "./triggers/video-encode.js"; // Publisher-only definition

const oqron = new Oqron({
  mode: "redis",
  redis: "redis://localhost:6379",
});
await oqron.start();

worker-server.ts

import { Oqron } from "oqronkit";
import "./triggers/video-worker.js"; // Consumer worker definition

const oqron = new Oqron({
  mode: "redis",
  redis: "redis://localhost:6379",
});
await oqron.start();

Worker Configuration

Option	Type	Default	Description
`topic`	`string`	required	Queue name to listen on
`handler`	`(ctx) => Promise<R>`	optional	Job processor function (mutually exclusive with `processBatch`)
`processBatch`	`(jobs) => Promise<...>`	optional	Bulk processor for DataLoader patterns
`batchSize`	`number`	`10`	Claim limit for `processBatch`
`concurrency`	`number`	`1`	Max parallel job processing
`guaranteedWorker`	`boolean`	`true`	Heartbeat crash-safety — set `false` for lightweight tasks
`heartbeatMs`	`number`	`5000`	Lock renewal interval
`lockTtlMs`	`number`	`30000`	Lock time-to-live
`retries`	`RetryConfig`	—	Retry policy
`hooks`	`{ onSuccess, onFail }`	—	Lifecycle hooks
`throttle`	`{ max, duration }`	—	Dispatch rate cap per time window

Throttle

Cap the dispatch rate to avoid overwhelming external APIs:

const apiWorker = Worker<ApiRequest, ApiResponse>({
  topic: "external-api-calls",
  concurrency: 3,
  throttle: { max: 50, duration: 60_000 }, // 50 calls per minute

  handler: async (ctx) => {
    const res = await fetch(`https://api.vendor.com/process`, {
      method: "POST",
      body: JSON.stringify(ctx.data),
    });
    return res.json();
  },
});

throttle is per-process. For cluster-wide limiting, compose with rateLimiter.

Environment Targeting

You can limit which environments a worker consumes jobs in. For example, if you want a worker to only run in production and staging, use the environments array. In other environments, the worker won't poll or consume jobs, but the API can still publish them.

export const heavyWorker = Worker<JobData, string>({
  topic: "heavy-compute-topic",
  environments: ["production", "staging"], // Only active here
  handler: async (ctx) => {
    // ...
  },
});

Context API

The ctx object passed to your handler provides several powerful methods for managing the job lifecycle:

ctx.progress(percent, label?): Update job progress (0-100).
ctx.log(level, message): Write logs to the job run record.
ctx.spawnChild(queue, data, opts): Enqueue a linked child job.
ctx.requeue(delayMs?): Push the job back to the queue to run later without consuming a retry attempt.
ctx.discard(): Permanently fail the job without triggering further retries.

export const emailWorker = Worker<EmailData>({
  topic: "email-topic",
  handler: async (ctx) => {
    ctx.log("info", `Preparing to send to ${ctx.data.to}`);

    if (ctx.data.blocked) {
      // Fail permanently, don't retry
      ctx.discard();
      return;
    }

    try {
      await sendEmail(ctx.data);
      ctx.progress(100, "Sent");
    } catch (err) {
      if (err.isRateLimited) {
        // Try again in 60s without counting as a failed attempt
        ctx.requeue(60_000);
        return;
      }
      throw err; // Standard failure, triggers retries
    }
  },
});

Pre-Execution Gating

Use condition to prevent jobs from executing until certain requirements are met. If condition returns false, the job is re-queued with a delay (nack).

export const dependentWorker = Worker<JobData>({
  topic: "dependent-topic",
  condition: async (ctx) => {
    // Only run if the external service is healthy
    const isHealthy = await checkExternalService();
    return isHealthy;
  },
  handler: async (ctx) => {
    // ...
  },
});

const apiWorker = Worker<ApiRequest, ApiResponse>({
  topic: "external-api-calls",
  concurrency: 3,
  throttle: { max: 50, duration: 60_000 }, // 50 calls per minute

  handler: async (ctx) => {
    const res = await fetch(`https://api.vendor.com/process`, {
      method: "POST",
      body: JSON.stringify(ctx.data),
    });
    return res.json();
  },
});

throttle is per-process. For cluster-wide limiting, compose with rateLimiter.

Crash Recovery Flow

Worker claims job → writes workerId + TTL to lock adapter
Worker runs heartbeat setInterval to renew lock
If worker crashes (SIGKILL/OOM), heartbeat stops
Lock expires in Redis/Postgres after lockTtlMs
StallDetector finds expired lock → marks job as stalled
Job is re-queued and routed to a healthy worker within ~15 seconds

Next Steps

Crash Safety — Deep-dive into heartbeat locks and stall detection
Scheduler — Add time-based job execution

Distributed Worker

On this page