Background tasks

Covara ships a distributed task queue backed by the KV store: Zod-validated task definitions, retry strategies, delayed/cron scheduling, idempotency, concurrency caps, a dead-letter queue, and a Cloudflare Queues adapter for the edge.

Quick start

import {
  defineTask, initializeTasks, getTaskScheduler, getTaskRegistry, startTaskWorkers,
} from "covara/tasks";
import { createKV } from "covara/kv";
import { z } from "zod";

const kv = await createKV({ type: "redis", redis: { url: "redis://localhost" } });
initializeTasks(kv);

const sendEmailTask = defineTask({
  name: "send-email",
  input: z.object({ to: z.string().email(), subject: z.string(), body: z.string() }),
  handler: async (ctx, input) => {
    await sendEmail(input.to, input.subject, input.body);
    return { sent: true };
  },
});

getTaskRegistry().register(sendEmailTask);
await startTaskWorkers(kv, getTaskRegistry(), 3);

const taskId = await getTaskScheduler().enqueue(sendEmailTask, {
  to: "user@example.com", subject: "Welcome!", body: "Thanks for signing up.",
});

Defining tasks

const myTask = defineTask({
  name: "my-task",
  input: z.object({ data: z.string() }),
  output: z.object({ result: z.number() }),
  handler: async (ctx, input) => { /* ... */ },

  retry: {
    maxAttempts: 5,
    backoff: "exponential",   // "exponential" | "linear" | "fixed"
    initialDelayMs: 1000,
    maxDelayMs: 60000,
    retryOn: (error) => error.name !== "ValidationError",
  },

  timeout: 30000,
  maxConcurrency: 10,          // enforced fleet-wide cap
  priority: 75,                // higher = sooner (0-100)
  debounce: { windowMs: 5000, key: (input) => input.userId },
  idempotencyKey: (input) => `order-${input.orderId}`,
  idempotencyRetentionMs: 24 * 60 * 60 * 1000,
  resultTtlMs: 60 * 60 * 1000,
});

Task context

Property	Description
`taskId`	Unique id
`attempt`	Current attempt (1-based)
`scheduledAt` / `startedAt`	Timestamps
`workerId`	Processing worker
`signal`	`AbortSignal` for cancellation
`reportProgress(percent, message?)`	Persist progress (0–100, clamped)

handler: async (ctx, input) => {
  for (let i = 0; i < input.rows.length; i++) {
    await importRow(input.rows[i]);
    await ctx.reportProgress((i / input.rows.length) * 100, `row ${i}`);
  }
}
// elsewhere: (await scheduler.getTask(id))?.progress → { percent, message?, updatedAt }

While running, the worker writes a heartbeat (lastHeartbeatAt, default every 10s) and extends its distributed lock; if the lock can't be extended, the task is aborted via ctx.signal and rescheduled.

Idempotency & concurrency (enforced)

Idempotency: when idempotencyKey is set, the first successful completion is recorded in KV. A second execution with the same key within idempotencyRetentionMs (default 24h) short-circuits and returns the stored result. Use a distributed KV so this holds across workers.
Concurrency: maxConcurrency is enforced via a distributed counter — workers reserve a slot before running and release on completion; over-cap tasks wait for free capacity.

Result TTL

A completed/failed task carries resultExpiresAt = completedAt + resultTtlMs (default 24h) and is deleted lazily on next read. Dead tasks live in the DLQ until purged.

Scheduling

const scheduler = getTaskScheduler();

await scheduler.enqueue(myTask, { data: "hello" });                       // ASAP
await scheduler.schedule(myTask, { data: "hello" }, { delay: 5 * 60_000 }); // delayed
await scheduler.schedule(myTask, { data: "hello" }, { at: new Date("2024-12-25T00:00:00Z") });
await scheduler.schedule(myTask, { data: "urgent" }, { delay: 1000, priority: 100 });

Recurring tasks

await scheduler.scheduleRecurring(dailyReportTask, {}, { interval: 60 * 60 * 1000 });
await scheduler.scheduleRecurring(dailyReportTask, {}, { cron: "0 0 * * *", timezone: "UTC" });
await scheduler.scheduleRecurring(weeklyDigestTask, {}, { cron: "0 9 * * 1", timezone: "America/New_York" });

Missed-occurrence catchup decides what happens when the scheduler comes back after downtime:

`catchup`	Behavior
`"skip"` (default)	Fire once for the current occurrence; drop missed ones
`"last"`	Run only the latest missed occurrence
`"all"`	One task per missed occurrence, capped at 1000

Managing tasks

const task = await scheduler.getTask(taskId);   // status: pending|scheduled|running|completed|failed|dead
await scheduler.cancel(taskId);
await scheduler.getTasks({ status: ["pending", "scheduled"], name: "send-email", limit: 100 });
await scheduler.getQueueDepth();

Workers

import { startTaskWorkers, createTaskWorker } from "covara/tasks";

const workers = await startTaskWorkers(kv, registry, 3, {
  concurrency: 5, pollIntervalMs: 1000, lockTtlMs: 30000, heartbeatMs: 10000,
});

const worker = createTaskWorker(kv, registry, { id: "worker-main", concurrency: 10, taskTypes: ["send-email"] });
await worker.start();

worker.pause(); worker.resume();
await worker.drain(30000);                       // stop claiming, wait for in-flight
await worker.stop({ drain: true, timeoutMs: 30000 });
worker.getStats();                               // { activeTasks, processedCount, failedCount, uptime, ... }

Retry strategies

retry: { backoff: "exponential", initialDelayMs: 1000, maxDelayMs: 60000 } // 1,2,4,8,16s (cap 60), +10-20% jitter
retry: { backoff: "linear", initialDelayMs: 2000, maxDelayMs: 30000 }      // 2,4,6,8s (cap 30)
retry: { backoff: "fixed", initialDelayMs: 5000 }                          // 5,5,5s
retry: { maxAttempts: 3, retryOn: (e) => e.name !== "ValidationError" }     // conditional
retry: { retryDelayMs: (e) => (e as { retryAfterMs?: number }).retryAfterMs } // per-error delay

retryOn returning false sends the task straight to the DLQ. retryDelayMs(error, attempt) overrides the backoff for one retry when it returns a non-negative number (e.g. a provider's Retry-After); undefined falls back to the configured backoff. Both apply to the KV worker and the Cloudflare Queues consumer. Durable email is built on these.

Dead letter queue

Tasks exceeding maxAttempts go to the DLQ.

import { createDeadLetterQueue } from "covara/tasks";
const dlq = createDeadLetterQueue(kv, requeue);

await dlq.list(100, 0);
await dlq.get(taskId);
const newId = await dlq.retry(taskId, "ops@example.com"); // fresh task, preserves lineage
await dlq.retryAll({ limit: 100, replayedBy: "ops@example.com" });
await dlq.purge(7 * 24 * 60 * 60 * 1000);
await dlq.count();
await dlq.audit(100); // [{ originalTaskId, newTaskId, replayedAt, replayedBy?, replayCount }]

Replay lineage: a replay creates a fresh task (new id, reset attempts) while preserving originalTaskId (always the first task), replayCount, replayedFromDlqAt, and replayedBy, and appends a durable audit entry.

Alerting: provide onDlqEnqueue (on the worker config or the DLQ) to be paged when a task is parked. Callback failures never affect persistence.

await startTaskWorkers(kv, registry, 3, {
  onDlqEnqueue: async (entry) => alert(`Task ${entry.taskId} dead: ${entry.reason}`),
});

Cloudflare Queues

On Workers there is no long-lived poller — use the Queues adapter. Retries are delegated to Cloudflare (message.retry()); idempotency, DLQ, and result storage still use your KV.

import { createCloudflareQueueProducer, createQueueConsumer, getTaskRegistry } from "covara/tasks";
import { createDurableObjectKV } from "covara/kv";

export default {
  async fetch(req, env) {
    const producer = createCloudflareQueueProducer(env.TASK_QUEUE);
    return new Response(await producer.enqueue(sendEmailTask, { to: "a@b.com", subject: "Hi", body: "..." }));
  },
  async queue(batch, env) {
    const consumer = createQueueConsumer({
      kv: createDurableObjectKV(env.KV_DO),
      registry: getTaskRegistry(),
      onDlqEnqueue: async (entry) => { /* alert */ },
    });
    await consumer.process(batch);
  },
};

[[queues.producers]]
queue = "tasks"
binding = "TASK_QUEUE"

[[queues.consumers]]
queue = "tasks"
max_batch_size = 10
max_retries = 3
dead_letter_queue = "tasks-dlq"

The producer maps delay/at to the queue's delaySeconds. DLQ entries produced by the consumer are not auto-requeued — replay with dlq.retry(...) or use a Cloudflare dead-letter queue binding. See Workers deployment.

Resource-trigger hooks

Fire tasks automatically when resources change:

import { createTaskTriggerHooks, composeHooks } from "covara/tasks";

const taskHooks = createTaskTriggerHooks({
  onCreate: [
    { task: sendWelcomeEmailTask, when: (data) => data.role === "user" },
    { task: auditLogTask, delay: 1000 },
  ],
  onUpdate: [{ task: auditLogTask, transform: (data) => ({ changes: data }) }],
  onDelete: [{ task: auditLogTask }],
});

useResource(usersTable, { id: usersTable.id, db, hooks: composeHooks(taskHooks, { /* ... */ }) });

ResourceTaskTrigger: { task, when?(data), transform?(data), delay? }.

Recurring scheduler management

import { createRecurringManager, startRecurringScheduler } from "covara/tasks";

const recurring = createRecurringManager(kv);
const scheduleId = await recurring.create(dailyReportTask, {}, { cron: "0 0 * * *", timezone: "America/New_York" });
await recurring.pause(scheduleId); await recurring.resume(scheduleId); await recurring.delete(scheduleId);

const stop = startRecurringScheduler(kv, async (name, input) => {
  const task = registry.get(name);
  return task ? scheduler.enqueue(task, input) : "";
}, 1000);

Best practices

Always set timeout. 2. Use idempotencyKey for payments/critical ops. 3. Don't retry non-transient errors (retryOn). 4. Alert on the DLQ. 5. Reserve high priority for time-sensitive work. 6. Scale workers horizontally. 7. Use a distributed KV (Redis / Durable Object) in production.

KV store · Email · Procedures & hooks
Workers deployment · Tasks contract

Quick start​

Defining tasks​

Task context​

Idempotency & concurrency (enforced)​

Result TTL​

Scheduling​

Recurring tasks​

Managing tasks​

Workers​

Retry strategies​

Dead letter queue​

Cloudflare Queues​

Resource-trigger hooks​

Recurring scheduler management​

Best practices​

Related​