- A task queue built on Effect. - Describe work as a schema, process it with a handler that is just an - Effect. Retries, delays, idempotency, cron schedules: handled. Redis - keeps unfinished work recoverable. + 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.
01 — Lifecycle
-- The state machine every task moves through: offered, queued, leased - under a unique fence token, and acknowledged — or routed back through - its retry Schedule. Unfinished work survives worker loss. (An - illustration of the engine's internals, not a product dashboard.) + 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.
02 — The API
A task is a schema, not a function. Your handler receives a fully - decoded payload — the real object, not a JSON string — and its - failures are pattern-matchable typed errors. + decoded payload—not a JSON string—and typed failures remain + pattern-matchable downstream.
04 — Receipts
-- Every release ships reproducible performance evidence: full end-to-end - task lifecycles — atomic create, fenced acquire, acknowledge — - measured per payload size and concurrency. + 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.
05 — Batteries
-, // 03 workerimport {"{ Schedule, Schema }"}from "effect"+import {"{ Cron, Effect, Schedule, Schema }"}from {" "} +"effect"{";\n"} -import {"{ Task, TaskQueue }"}from "@effectmq/core"+import {"{ NodeRuntime }"}from "@effect/platform-node"+ {";\n"} +import {" "} + {"{ Scheduler, Task, TaskEngine, TaskQueue, type TaskHandler, Worker }"}{" "} +from "@effectmq/core"{";\n\n"}class EmailRejected extends Schema {".TaggedError<"} @@ -74,8 +79,8 @@ export const apiSnippets: ReadonlyArray= [ {".make("} "emails"{", SendEmail);\n\n"} -{"// Enqueue a payload — idempotent, durable.\n"} -yield +{"// Enqueue a payload and keep its durable handle.\n"} +const offered =yield {"* "}TaskQueue {".offer(emails, {\n to: "} @@ -95,39 +100,33 @@ export const apiSnippets: ReadonlyArray= [ {"// succeeds with your value — or fails with EmailRejected\n"} {"\n"}{"// Or await a task you already offered.\n"} -const task =yield - {"* "} -TaskQueue - {".offer(emails, payload);\n"}const outcome =yield {"* "}TaskQueue - {".wait(emails, task.handle);"} + {".wait(emails, offered.handle);"}
-, // 04 concurrency{"// Take one task, run it, report the outcome.\n"} -yield - {"* "} -TaskQueue - {".complete(emails, (task) =>\n "} +{"// Define the handler once, against the task schema.\n"} +const handleSendEmail:TaskHandler + { + "<\n typeof SendEmail.payloadSchema,\n typeof SendEmail.successSchema,\n typeof SendEmail.errorSchema\n> = (task) =>\n " + }Effect {".succeed("} {/* biome-ignore lint/suspicious/noTemplateCurlyInString: displayed as literal code */}{"`provider:${task.payload.to}`"}- {"),\n);\n\n"} -{"// Or name the handler once, and loop forever.\n"} -const - {" sendEmailWorker = "} -TaskQueue - {".complete(emails, handleSendEmail);\n"} -yield - {"* sendEmailWorker.pipe("} -Effect - {".repeat("} -Schedule - {".forever));\n\n"} -{"// The engine + its Redis layer: the only wiring you need.\n"} + {");\n\n"} ++ {"// A managed worker polls, supervises leases and drains cleanly.\n"} + +const worker =Worker + {".make(emails, handleSendEmail, { concurrency: "} +5 + {" });\n"} +const program =Worker + {".run(worker);\n\n"} +{"// Provide the complete Redis-backed runtime once.\n"} const AppLayer =TaskEngine {".layer({\n redis: { url: "}"redis://localhost:6379"@@ -139,35 +138,22 @@ export const apiSnippets: ReadonlyArray= [
-, // 05 schedule{"// No bespoke concurrency options. Just Effect.\n"} -const worker =Effect - {".gen("} -function - {"* () {\n "} -{"// At most 5 tasks in flight at any moment.\n"} - {" "} -const semaphore =yield - {"* "} -Semaphore - {".make("} +{"// Local concurrency is a Worker option, not a fiber recipe.\n"} +const worker =Worker + {".make(emails, handleSendEmail, {\n concurrency: "}5 - {");\n\n "} -yield - {"* "} -Semaphore - {".withPermit(\n semaphore,\n "} -TaskQueue - {".complete(emails, handleSendEmail),\n ).pipe(\n "} -Effect - {".forkScoped, "} -{"// each worker is its own fiber\n"} - {" "} -Effect - {".repeat("} -Schedule - {".forever), "} -{"// ...that keeps pulling work\n"} - {" );\n});"} + {",\n pollInterval: "} +"250 millis"+ {",\n drainTimeout: "} +"30 seconds"+ {",\n});\n\n"} ++ { + "// The limit is per process. Add shared coordination for a global cap.\n" + } + +const program =Worker + {".run(worker);"}
@@ -198,46 +184,30 @@ export const apiSnippets: ReadonlyArray, ]; -export const semTabLabels = ["bounded", "rate limit", "fan-out"]; +export const semTabLabels = ["worker pool", "local pacing", "fan-out"]; export const semSnippets: ReadonlyArray= [
-, // rate limit{"// At most 5 tasks in flight, across any number of runners.\n"} -const semaphore =yield - {"* "} -Semaphore - {".make("} +{"// Five acquire/process loops in this worker process.\n"} +const worker =Worker + {".make(emails, handle, { concurrency: "}5 - {");\n\n"} + {" });\n\n"}yield {"* "} -Semaphore - {".withPermit(\n semaphore,\n "} -TaskQueue - {".complete(emails, handle),\n).pipe("} -Effect - {".forkScoped, "} -Effect - {".repeat("} -Schedule - {".forever));"} +Worker + {".run(worker);"}
-, // fan-out{"// A rate limit is just a Semaphore + a Schedule.\n"} -const permits =yield - {"* "} -Semaphore - {".make("} -1 - {");\n\n"} ++ {"// Pace one local loop. Use shared coordination for a global limit.\n"} + +const runOne =TaskQueue + {".complete(emails, handle);\n\n"}yield - {"* "} -Semaphore - {".withPermit(\n permits,\n "} -TaskQueue - {".complete(emails, handle),\n).pipe(\n "} + {"* runOne.pipe(\n "}Effect {".repeat("}Schedule @@ -249,22 +219,19 @@ export const semSnippets: ReadonlyArray= [
-, ]; diff --git a/apps/docs/components/landing/concurrency-tabs.tsx b/apps/docs/components/landing/concurrency-tabs.tsx index febd463..fca87f7 100644 --- a/apps/docs/components/landing/concurrency-tabs.tsx +++ b/apps/docs/components/landing/concurrency-tabs.tsx @@ -10,20 +10,20 @@ export function ConcurrencyTabs() { return ( <>{"// Fan out: more fibers — or more processes. Same worker.\n"} -const worker =TaskQueue - {".complete(emails, handle).pipe(\n "} -Effect - {".repeat("} -Schedule - {".forever),\n);\n\n"} -yield - {"* "} ++ {"// Run the same worker program in more OS processes or containers.\n"} + +const worker =Worker + {".make(emails, handle, { concurrency: "} +5 + {" });\n\n"} +const program =Worker + {".run(worker).pipe(\n "}Effect - {".all(\n "} -Array - {".from({ length: "} -6 - {" }, () => worker),\n { concurrency: "} -"unbounded"- {" },\n);"} + {".provide(AppLayer),\n);\n\n"} +NodeRuntime + {".runMain(program); "} +{"// deploy N replicas"}
03 — Concurrency
- No builtin concurrency knobs, rate limiting or backpressure — it - doesn't need them. complete{" "} - does exactly one task; Semaphore, Schedule and fibers decide how many - run at once. None of it is our invention. All of it composes. + 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.