@juliascript/effectmq

It's a task queue built on Effect: typed payloads, typed results, typed errors, all the way down. You describe a unit of work as a schema, hand it to a queue, and process it with a handler that is just an Effect. Retries, delays, idempotency, cron schedules: handled. The available engine is backed by Redis, but, like many things in Effect, it can be swapped for a different implementation.

pnpm add @juliascript/effectmq [email protected] @effect/[email protected]

This library is built on the Effect 4 beta and doesn't work with the current stable Effect release. The examples below use the @effect/platform-node Redis layer. This is beta-era software riding beta-era Effect; pin accordingly.

In thirty seconds

Define a task, enqueue work, process it. The whole loop:

import { Effect, Layer, Schema } from "effect";
import { NodeRedis, NodeRuntime } from "@effect/platform-node";
import { Task, TaskEngine, TaskQueue } from "@juliascript/effectmq";

const SendEmail = Task.make({
  name: "send-email",
  payload: { to: Schema.String, subject: Schema.String },
  successSchema: Schema.String,
  errorSchema: Schema.Never,
});

const emails = TaskQueue.make("emails", SendEmail);

const program = Effect.gen(function* () {
  yield* TaskQueue.offer(emails, { to: "[email protected]", subject: "Welcome" });

  yield* TaskQueue.complete(emails, (task) =>
    sendViaProvider(task.payload), // returns the provider message id
  );
});

// The engine + its Redis layer: the only wiring you need to run the above.
const AppLayer = Layer.provideMerge(TaskEngine.layer(), NodeRedis.layer());

program.pipe(Effect.provide(AppLayer), NodeRuntime.runMain);

That's the shape of it. The rest of this README explains the pieces (typed errors, retries, worker pools, schedules) and the one thing the library deliberately doesn't do.

The setup, once

TaskEngine.layer() requires the Redis service. NodeRedis provides it:

import { Layer } from "effect";
import { NodeRedis } from "@effect/platform-node";
import { TaskEngine } from "@juliascript/effectmq";

const AppLayer = Layer.provideMerge(TaskEngine.layer(), NodeRedis.layer());

NodeRedis is the convenient default, but anything that provides the Redis service works: a pooled connection, an in-memory fake for tests, or a Redis-compatible server (Valkey, Dragonfly, and friends).

TaskEngine is the machinery underneath: atomic Lua scripts, locks, the lists tasks move between. Provide its layer and forget it; the API you live in is TaskQueue and Scheduler.

Define a task

A task is a schema, not a function. You declare what goes in (payload), what a success looks like, and what a failure looks like. The idempotencyKey decides what "the same task" means: offer the same key twice and you get one task, not two.

A tagged error makes failures pattern-matchable downstream, so reach for Schema.TaggedErrorClass rather than a bare struct.

import { Schema } from "effect";
import { Task, TaskQueue } from "@juliascript/effectmq";

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

const SendEmail = Task.make({
  name: "send-email",
  payload: { to: Schema.String, subject: Schema.String },
  successSchema: Schema.String, // e.g. a provider message id
  errorSchema: EmailRejected,
  // optional, but it's how you ensure the same job isn't enqueued twice
  idempotencyKey: (p) => `email:${p.to}:${p.subject}`,
});

const emails = TaskQueue.make("emails", SendEmail);

Offer work, then do it

offer enqueues a payload. complete takes the next task, runs your handler, and reports the outcome back to the engine, returning true on success and false when the handler fails. The task your handler receives is fully decoded: task.payload is the real object, not a JSON string.

import { Effect } from "effect";

const program = Effect.gen(function* () {
  yield* TaskQueue.offer(emails, {
    to: "[email protected]",
    subject: "Welcome",
  });

  // Take one task, run it, report success/failure.
  const ok = yield* TaskQueue.complete(emails, (task) =>
    Effect.gen(function* () {
      const id = yield* sendViaProvider(task.payload); // your code
      return id; // matches successSchema
      // ...or `yield* new EmailRejected({ reason })` to fail with the typed error
    }),
  );
  // ok === true  → handler succeeded, resolved per onSuccessPolicy
  // ok === false → handler failed, routed per onFailurePolicy
});

One complete processes one task. To process many, set your workers up accordingly.

On concurrency

Differently than other queue libraries, effectmq doesn't have builtin concurrency, rate limiting, backpressure. It doesn't need to, it works perfectly with the Effect primitives you are used to.

Effect gives you the fine control you need from your workers, so complete does exactly one task, and you decide how many run at once, with the same tools you use everywhere else:

import { Effect, Schedule, Semaphore } from "effect";

// Concurrency example with semaphore
const worker = Effect.gen(function* () {
  // At most 5 tasks in flight at any moment.
  const semaphore = yield* Semaphore.make(5);

  yield* Semaphore.withPermit(
    semaphore,
    TaskQueue.complete(emails, (task) => handle(task)),
  ).pipe(
    Effect.forkScoped,            // each worker is its own fiber
    Effect.repeat(Schedule.forever), // ...that keeps pulling work
  );
});

Want a rate limit instead of a raw permit count? Compose one from a Semaphore and a Schedule. Want retries with jitter? Schedule. Want to fan out across a cluster? Run more processes. None of it is our invention, all of it composes. The queue's job is to hand you the next task, exactly once, safely. What you do with your fibers is your business.

"But Effect already has Workflow"

It does, and it's excellent, for a different problem. Effect Workflow is durable execution: long-running, multi-step sagas that survive process death, resume exactly where they left off, and persist every intermediate step so the whole history can be replayed. It leans on clustering and sharding; nodes have to be live and coordinated; the durability is total because the use case demands it.

That power has a price that sometimes isn't worth paying. Sometimes you don't have a saga. You have a job. "Send this email." "Resize that image." "Spawn 5 AI agents to complete these tasks." There's no multi-step history worth replaying; there's a payload, a handler, and an outcome. Reaching for durable execution there is like renting a shipping container to mail a letter.

effectmq works whether you have a single worker running in a separate fiber or a hundred distributed across multiple processes.

So:

If Workflow is Temporal (durable, replayable, cluster-coordinated orchestration) then this is BullMQ: a queue. You put work in, workers take it out, it runs once, retries if it must, and then it's done. No replay log, no shard map, no requirement that the whole cluster be breathing. Just a queue, with Effect's types and Effect's primitives.

Pick durable execution when the process is the thing you can't afford to lose. Pick a queue when the work is.

Scheduling

For recurring work, Scheduler.make runs a handler on a cron expression. If you've used Effect's ClusterCron (effect/unstable/cluster/ClusterCron), it'll feel familiar. The schedule state lives in the engine, so multiple processes running the same named scheduler will collectively fire the handler once per tick, not once per process.

import { Cron } from "effect";
import { Scheduler } from "@juliascript/effectmq";

const nightlyReport = Scheduler.make({
  name: "nightly-report",
  cron: Cron.parseUnsafe("0 2 * * *"), // 02:00 every day
  handler: Effect.gen(function* () {
    yield* buildAndSendReport();
  }),
});

Notes

• Completion policies. offer accepts onSuccessPolicy and onFailurePolicy, each one of delete | keep | mark-as-success | mark-as-failure. They decide where a finished task lands: gone, quietly retained, or parked on the success/failed list for inspection. Defaults are delete.
• Retries. Set maxRetries on offer; a failing task goes back to the wait list until the count is exhausted, then the failure policy applies. A Canceled error skips remaining retries.
• Idempotency. The idempotencyKey is the task id. Same key, same task: offering again updates rather than duplicates.
• Delays. offer(..., { delay }) schedules the task for the future; it sits on the scheduled list until its time comes.
• The engine. TaskEngine is the low-level, Lua-backed layer all of this sits on. You provide its layer; you rarely call it directly.

License

MIT.