EFFECT 4 BETAREDIS-BACKEDMIT

Typed payloads.
Typed results.
Typed errors. All the way down.

Define background work with schemas and process it with handlers that are ordinary Effects. Redis keeps unfinished attempts recoverable; payloads, results and failures stay typed end to end.

$pnpm add @effectmq/core@rc
Read the docs →View on npm ↗

Durable work, explicit states.

A task is offered, queued, leased under a unique fence token, then acknowledged or rescheduled by its retry policy. If a worker disappears, maintenance recovers the unfinished attempt. This diagram shows the engine's state transitions.

waiting / scheduledactive (fenced attempt)retry / failedsuccess

The whole loop, in thirty seconds.

A task is a schema, not a function. Your handler receives a fully decoded payload—not a JSON string—and typed failures remain pattern-matchable downstream.

import { Cron, Effect, Schedule, Schema } from "effect";
import { NodeRuntime } from "@effect/platform-node";
import { Scheduler, Task, TaskEngine, TaskQueue, type TaskHandler, Worker } from "@effectmq/core";

class EmailRejected extends Schema.TaggedError<EmailRejected>()(
  "EmailRejected",
  { reason: Schema.String },
) {}

const SendEmail = Task.make({
  name: "send-email",
  payload: { to: Schema.String, subject: Schema.String },
  success: Schema.String,
  error: EmailRejected,                        // typed, pattern-matchable failure
  idempotencyKey: (p) => `email:${p.to}:${p.subject}`,
  retry: Schedule.exponential("1 second"),   // retries are just Schedules
});

Built in locally. Explicit globally.

Worker runs a bounded pool of local task slots with lease supervision, maintenance and graceful draining. Run more processes to fan out. Cross-process concurrency and rate limits need shared coordination; a local semaphore cannot enforce them.

// Five acquire/process loops in this worker process.
const worker = Worker.make(emails, handle, { concurrency: 5 });

yield* Worker.run(worker);

A baseline you can reproduce.

The release baseline measures full task lifecycles—atomic create, fenced acquire and acknowledge—by payload size and concurrency. Use it to detect regressions, not to size production infrastructure.

2,968 tasks/s

peak lifecycle throughput · 1 KiB @ c=32

1.44 ms p50

single-task lifecycle latency · 1 KiB @ c=1

39,097 items/s

due-backlog sweep, atomic Lua batches

Completed lifecycles per second — by payload × concurrency

592

2,180

2,922

c=1

c=8

c=32

64 B payload

641

2,347

2,968

c=1

c=8

c=32

1 KiB payload

483

1,624

2,036

c=1

c=8

c=32

16 KiB payload

baseline: Node 22 · Redis 8.0.6 (loopback) · Effect 4.0.0-beta.107 · macOS arm64 — committed as regression evidence, not a capacity claim.

What the queue handles.

Retries are Schedules

Declare retry on the task as an Effect Schedule — exponential, jittered, whatever composes. Exhausted schedule → your failure policy.

retry: Schedule.exponential("1 second")

Idempotency is the id

The idempotency key is the task id. Offer the same key twice and you get one task, not two — replacement is an explicit new generation.

idempotencyKey: (p) => `email:${p.to}`

Durable cron

Scheduler.make materializes a real queue task per tick — competing schedulers and crash recovery can safely re-offer it.

missed: { _tag: "coalesce" }

Typed lifecycle events

task.completed carries your success type; task.failed carries your typed error. Streams, wait and execute all decode against your schemas.

TaskQueue.stream(emails)

At-least-once, fenced

Every attempt is fenced with a unique lease token; stalled workers are separated from failed handlers. Unfinished work stays recoverable.

lease: unique token per attempt

One runtime layer

TaskEngine.layer wires the Redis pools, health services, cryptographic identity and Lua-backed engine. Provide it once; work through TaskQueue, Worker and Scheduler.

TaskEngine.layer({ redis })

Define the work. Run the worker.
Redis keeps unfinished attempts recoverable.

$pnpm add @effectmq/core@rc
Star on GitHub ↗