From 6518fb3139815b7a7a53b2beb257464c5b2f3ab3 Mon Sep 17 00:00:00 2001 From: Julia Ortiz Date: Thu, 20 Aug 2026 16:26:54 -0300 Subject: [PATCH] docs: polish landing page and Effect examples --- .changeset/free-rabbits-clap.md | 2 + README.md | 94 ++++++---- apps/docs/app/landing.css | 18 +- apps/docs/app/page.tsx | 57 +++--- .../docs/components/landing/code-snippets.tsx | 167 +++++++----------- .../components/landing/concurrency-tabs.tsx | 12 +- .../components/landing/install-command.tsx | 25 ++- apps/docs/components/landing/sem-canvas.tsx | 16 +- .../docs/how-to/make-handlers-idempotent.mdx | 71 ++++++-- .../content/docs/how-to/operate-redis.mdx | 53 +++--- .../docs/tutorials/getting-started.mdx | 65 ++++--- docs/operations.md | 117 +++++++----- 12 files changed, 392 insertions(+), 305 deletions(-) create mode 100644 .changeset/free-rabbits-clap.md diff --git a/.changeset/free-rabbits-clap.md b/.changeset/free-rabbits-clap.md new file mode 100644 index 0000000..a845151 --- /dev/null +++ b/.changeset/free-rabbits-clap.md @@ -0,0 +1,2 @@ +--- +--- diff --git a/README.md b/README.md index 0670c00..085ff5c 100644 --- a/README.md +++ b/README.md @@ -1,18 +1,42 @@ -# @effectmq/core +# effectmq -It's a task queue built on [Effect](https://effect.website): 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. +**A typed, Redis-backed task queue for Effect 4.** Define work with schemas, run +handlers as Effects, and keep payloads, results, and failures typed from producer +to worker. + +[Website](https://docs-one-eta-87.vercel.app/) · +[Documentation](https://docs-one-eta-87.vercel.app/docs) · +[Getting started](https://docs-one-eta-87.vercel.app/docs/tutorials/getting-started) · +[API reference](https://docs-one-eta-87.vercel.app/docs/reference/task-queue) · +[npm](https://www.npmjs.com/package/@effectmq/core) + +effectmq is for background jobs that need durable Redis state without becoming a +workflow engine: send an email, resize an image, refresh a cache, or materialize +a scheduled report. It provides: + +- schema-checked payloads, successes, and failures; +- at-least-once delivery with fenced attempts and stalled-worker recovery; +- Effect `Schedule` retries, delayed offers, deduplication, and durable cron; +- bounded local worker concurrency, graceful draining, and maintenance; +- typed lifecycle streams plus `wait` and `execute` for durable results. + +## Install ```bash -pnpm add @effectmq/core effect@4.0.0-beta.107 @effect/platform-node@4.0.0-beta.107 +pnpm add @effectmq/core@0.3.0-rc.0 effect@4.0.0-beta.107 @effect/platform-node@4.0.0-beta.107 ``` -This library is built on the Effect 4 beta and doesn't work with the current stable Effect release. The examples below use the bundled `NodeRedisPool` layer, a connection-pooled Redis client that ships with the package (`@effect/platform-node` is only needed for `NodeRuntime`). This is beta-era software riding beta-era Effect; pin accordingly. +> [!IMPORTANT] +> effectmq currently targets the Effect 4 beta and is not compatible with the +> stable Effect 3 release. Pin the versions shown above. Node.js 22.19 or newer +> is required; CI verifies Node.js 22 and 24. -Node.js 22.19 or newer is required; the release matrix verifies Node.js 22 and 24. +The package includes its pooled `NodeRedisPool` implementation. +`@effect/platform-node` is only needed by these examples for `NodeRuntime`. --- -## In thirty seconds +## Quick start Define a task, enqueue work, process it. The whole loop: @@ -46,11 +70,13 @@ const AppLayer = TaskEngine.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. +That is the complete producer-to-worker loop. For a clean-room walkthrough with +Redis startup and expected output, follow the +[getting-started tutorial](https://docs-one-eta-87.vercel.app/docs/tutorials/getting-started). --- -## The setup, once +## Runtime setup `TaskEngine.layer()` is the complete Node live graph: it provides the engine, cryptographic identity generation, and the retained Redis pool, role, and @@ -74,7 +100,9 @@ ACL, bounded-pool, persistence, failover, health, and shutdown guidance. The tested platform matrix is in the [support policy](./docs/support-policy.md), and reproducible throughput/tail-latency results are published as [performance evidence](./docs/performance.md). -`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`. +`TaskEngine` is the machinery underneath: atomic Lua scripts, leases, and the +lists tasks move between. Provide its layer once; application code normally +lives in `TaskQueue`, `Worker`, and `Scheduler`. --- @@ -135,7 +163,10 @@ One `complete` processes one task. To process *many*, set your workers up accord ### Predefine the handler -Handlers are just functions and workers are just Effects, so both are values you can name once and reuse. Type a handler with `TaskHandler` to declare it next to the task definition, before any queue exists; bind it to a queue with `complete` and you have a worker effect you can run, repeat, or fork like any other: +Handlers are functions and workers are Effects, so both are values you can name +once and reuse. Type a handler with `TaskHandler` to declare it next to the task +definition before any queue exists. Bind it with `complete` for one task, or use +the managed `Worker` shown below for a long-running process: ```ts docs-check=email // Declared against the task definition — no queue in sight yet. @@ -189,7 +220,7 @@ const offerAndWait = Effect.gen(function* () { --- -## On concurrency +## Run a worker `complete` processes exactly one task. For a long-running process, `Worker` provides bounded local concurrency, lease supervision, maintenance, and graceful @@ -200,11 +231,11 @@ const worker = Worker.make(emails, handleSendEmail, { concurrency: 5 }); const program = Worker.run(worker); ``` -The built-in worker does not impose distributed/global concurrency or rate -limits. Compose those policies from Effect primitives or external coordination, -and run more worker processes to fan out. The queue preserves eligible work and -fences the current attempt; handlers remain at-least-once and must make external -side effects idempotent. +`concurrency` is local to one worker process (valid values: 1–1000). Run more +processes to fan out. Distributed/global concurrency and rate limits require +external coordination; effectmq does not pretend a process-local semaphore can +enforce them. The queue fences each attempt, but handlers remain at-least-once, +so make external side effects idempotent. --- @@ -260,28 +291,23 @@ const worker = Worker.make(reportQueue, ({ payload }) => ## 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.** Declare `retry` on the task definition (`Task.make`) as an Effect `Schedule` — or a `{ while, until, times, schedule }` options object. On failure the next run time is computed from the schedule and the task lands on the scheduled list until then; when the schedule is exhausted, the failure policy applies. `maxRetries` caps the attempts so an unbounded schedule (e.g. `Schedule.forever`) can't loop forever: it defaults to `5`, is overridable per-`offer` (the per-offer value wins), and set it to `null` for truly unbounded retries. A `Canceled` error skips remaining retries. +- **Retries.** Declare `retry` on the task definition (`Task.make`) as an Effect `Schedule` — or a `{ while, until, times, schedule }` options object. On failure the next run time is computed from the schedule and the task lands on the scheduled list until then; when the schedule is exhausted, the failure policy applies. The task-level `maxRetries` cap defaults to `5`; pass `null` there for an intentionally unbounded cap. An `offer` may override the cap with a finite non-negative number. Built-in canceled or stalled failures are not retried by the handler schedule. - **Idempotency.** The `idempotencyKey` is the task id. By default, offering the same key returns the existing generation unchanged; replacement requires explicit new-generation mode. - **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. -## Production guides - -- [Architecture](./docs/architecture.md) -- [Runtime boundaries](./docs/runtime-boundaries.md) -- [API reference](./docs/api-reference.md) -- [Delivery guarantees](./docs/delivery-guarantees.md) -- [Idempotent offers](./docs/idempotent-offers.md) -- [Task relationships](./docs/task-relationships.md) -- [Durable scheduling](./docs/scheduler.md) -- [Storage protocol v1](./docs/storage-protocol-v1.md) -- [Operations](./docs/operations.md) -- [Support policy](./docs/support-policy.md) -- [Upgrade and rollback](./docs/upgrade-and-rollback.md) -- [Performance evidence](./docs/performance.md) -- [Soak evidence](./docs/soak.md) -- [Release process](./docs/releasing.md) -- [Current release record](./docs/release-readiness.md) +## Go deeper + +| If you need to… | Read… | +| --- | --- | +| learn the library from a running example | [Getting started](https://docs-one-eta-87.vercel.app/docs/tutorials/getting-started) | +| process, schedule, retry, or await tasks | [How-to guides](https://docs-one-eta-87.vercel.app/docs/how-to/process-tasks) | +| look up exact API behavior | [API reference](./docs/api-reference.md) | +| understand delivery and task identity | [Delivery guarantees](./docs/delivery-guarantees.md) · [Idempotent offers](./docs/idempotent-offers.md) | +| operate Redis and plan upgrades | [Operations](./docs/operations.md) · [Upgrade and rollback](./docs/upgrade-and-rollback.md) | +| inspect architecture and storage contracts | [Architecture](./docs/architecture.md) · [Runtime boundaries](./docs/runtime-boundaries.md) · [Storage protocol v1](./docs/storage-protocol-v1.md) | +| evaluate support and performance | [Support policy](./docs/support-policy.md) · [Performance evidence](./docs/performance.md) · [Soak evidence](./docs/soak.md) | +| release the package | [Release process](./docs/releasing.md) · [Current release record](./docs/release-readiness.md) | --- diff --git a/apps/docs/app/landing.css b/apps/docs/app/landing.css index 6d46497..9adc520 100644 --- a/apps/docs/app/landing.css +++ b/apps/docs/app/landing.css @@ -222,6 +222,7 @@ .lp a.lp-ghost-btn { display: flex; align-items: center; + white-space: nowrap; color: #ededf2; border: 1px solid #2a2a33; border-radius: 10px; @@ -236,6 +237,15 @@ text-decoration: none; } +.lp a.lp-ghost-btn:focus-visible { + outline: 2px solid var(--accent); + outline-offset: 2px; +} + +.lp a.lp-ghost-btn:active { + border-color: var(--accent); +} + /* ---- sections ---- */ .lp-eyebrow { @@ -399,11 +409,15 @@ padding-top: 96px; padding-bottom: 96px; display: grid; - grid-template-columns: minmax(320px, 5fr) 6fr; + grid-template-columns: minmax(320px, 5fr) minmax(0, 6fr); gap: 56px; align-items: center; } +.lp-conc-grid > * { + min-width: 0; +} + .lp-conc-body { margin: 0 0 22px; font: @@ -708,7 +722,7 @@ @media (max-width: 960px) { .lp-conc-grid { - grid-template-columns: 1fr; + grid-template-columns: minmax(0, 1fr); gap: 36px; } .lp-stats, diff --git a/apps/docs/app/page.tsx b/apps/docs/app/page.tsx index 92aa34f..38db983 100644 --- a/apps/docs/app/page.tsx +++ b/apps/docs/app/page.tsx @@ -10,10 +10,11 @@ import "./landing.css"; export const metadata: Metadata = { title: "effectmq — typed task queue built on Effect", description: - "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.", + "A typed, Redis-backed task queue for Effect 4 with schema-checked payloads, results and failures, fenced attempts, retries, delays and durable cron.", }; const GITHUB_URL = "https://github.com/julia-script/effectmq"; +const NPM_URL = "https://www.npmjs.com/package/@effectmq/core"; const FEATURES = [ { @@ -43,8 +44,8 @@ const FEATURES = [ code: "lease: unique token per attempt", }, { - title: "Swappable engine", - body: "The Redis engine is a Layer built on atomic Lua scripts. Provide it and forget it — or swap it for a different implementation, like anything in Effect.", + title: "One runtime layer", + body: "TaskEngine.layer wires the Redis pools, health services, cryptographic identity and Lua-backed engine. Provide it once; work through TaskQueue, Worker and Scheduler.", code: "TaskEngine.layer({ redis })", }, ]; @@ -73,7 +74,7 @@ export default function HomePage() {
- BUILT ON EFFECT 4 + EFFECT 4 BETA REDIS-BACKED MIT
@@ -85,16 +86,18 @@ export default function HomePage() { Typed errors. All the way down.

- 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.

Read the docs → + + View on npm ↗ +
@@ -103,13 +106,12 @@ export default function HomePage() { className="lp-container" style={{ paddingTop: 96, paddingBottom: 84 }} > -

01 — Lifecycle

-

How a task lives.

+

Durable work, explicit states.

- 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.

@@ -138,12 +140,11 @@ export default function HomePage() { className="lp-container" style={{ paddingTop: 12, paddingBottom: 96 }} > -

02 — The API

The whole loop, in thirty seconds.

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.

@@ -158,12 +159,11 @@ export default function HomePage() { className="lp-container" style={{ paddingTop: 96, paddingBottom: 96 }} > -

04 — Receipts

-

Numbers, not vibes.

+

A baseline you can reproduce.

- 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.

@@ -207,9 +207,8 @@ export default function HomePage() { className="lp-container" style={{ paddingTop: 96, paddingBottom: 96 }} > -

05 — Batteries

- Handled, so you don't. + What the queue handles.

{FEATURES.map((f) => ( @@ -227,9 +226,10 @@ export default function HomePage() {

- Put work in. Take attempts. + Define the work. Run the worker.
- Redis keeps the rest recoverable. + Redis keeps unfinished attempts recoverable + .

@@ -250,10 +250,7 @@ export default function HomePage() { GitHub - + npm
diff --git a/apps/docs/components/landing/code-snippets.tsx b/apps/docs/components/landing/code-snippets.tsx index 686354d..dd70052 100644 --- a/apps/docs/components/landing/code-snippets.tsx +++ b/apps/docs/components/landing/code-snippets.tsx @@ -31,9 +31,14 @@ export const apiTabLabels = [ export const apiSnippets: ReadonlyArray = [ // 01 define 
-    import {"{ 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);"}
, // 03 worker 
-    {"// 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 = [
, // 04 concurrency 
-    {"// 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);"}
, // 05 schedule 
@@ -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 = [ // bounded 
-    {"// 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);"}
, // rate limit 
-    {"// 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 = [
, // fan-out 
-    {"// 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"}
, ]; 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 ( <>
-

03 — Concurrency

- Bring your own concurrency. + Built in locally. Explicit globally.

- 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.

{semTabLabels.map((label, i) => ( diff --git a/apps/docs/components/landing/install-command.tsx b/apps/docs/components/landing/install-command.tsx index 486496b..294e8b4 100644 --- a/apps/docs/components/landing/install-command.tsx +++ b/apps/docs/components/landing/install-command.tsx @@ -3,10 +3,10 @@ import { useEffect, useRef, useState } from "react"; const FULL_COMMAND = - "npm install @effectmq/core effect@4.0.0-beta.107 @effect/platform-node@4.0.0-beta.107"; + "pnpm add @effectmq/core@0.3.0-rc.0 effect@4.0.0-beta.107 @effect/platform-node@4.0.0-beta.107"; export function InstallCommand() { - const [copied, setCopied] = useState(false); + const [status, setStatus] = useState<"idle" | "copied" | "error">("idle"); const timer = useRef | undefined>(undefined); useEffect(() => () => clearTimeout(timer.current), []); @@ -15,19 +15,28 @@ export function InstallCommand() {
$ - npm install @effectmq/core + pnpm add @effectmq/core@rc
); diff --git a/apps/docs/components/landing/sem-canvas.tsx b/apps/docs/components/landing/sem-canvas.tsx index 14b45c9..ec2591f 100644 --- a/apps/docs/components/landing/sem-canvas.tsx +++ b/apps/docs/components/landing/sem-canvas.tsx @@ -24,9 +24,9 @@ const accentColor = () => .trim() || "#C6F94F"; /** - * Semaphore constellation: runners offer into a queue whose head is pulled - * through a column of permit stars. Three modes — bounded (5 permits), - * rate limit (1 permit on a clock), fan-out (6 permits). + * Worker constellation: producers offer into a queue whose head is pulled + * through local worker slots. Three modes — a five-slot local pool, a paced + * local loop, and process-level fan-out. */ export function SemCanvas({ mode }: { readonly mode: number }) { const canvasRef = useRef(null); @@ -64,13 +64,13 @@ export function SemCanvas({ mode }: { readonly mode: number }) { rateMs = m === 1 ? 620 : 0; gateLabel = m === 0 - ? "Semaphore.make(5)" + ? "Worker concurrency: 5" : m === 1 ? 'Schedule.spaced("100 millis")' - : "6 fibers · Effect.all"; + : "6 worker processes"; NODES = { - runnerA: { x: 0.09, y: 0.28, label: "runner A", flash: 0 }, - runnerB: { x: 0.09, y: 0.72, label: "runner B", flash: 0 }, + runnerA: { x: 0.09, y: 0.28, label: "producer A", flash: 0 }, + runnerB: { x: 0.09, y: 0.72, label: "producer B", flash: 0 }, queue: { x: 0.38, y: 0.5, label: "queue", flash: 0 }, done: { x: 0.92, y: 0.5, label: "done", flash: 0 }, }; @@ -94,7 +94,7 @@ export function SemCanvas({ mode }: { readonly mode: number }) { EDGES.push({ a: "queue", b: k, - label: j === midIdx ? (m === 2 ? "complete" : "withPermit") : "", + label: j === midIdx ? (m === 1 ? "complete()" : "completeOne()") : "", }); EDGES.push({ a: k, b: "done", label: "" }); }); diff --git a/apps/docs/content/docs/how-to/make-handlers-idempotent.mdx b/apps/docs/content/docs/how-to/make-handlers-idempotent.mdx index c396fa2..48e5711 100644 --- a/apps/docs/content/docs/how-to/make-handlers-idempotent.mdx +++ b/apps/docs/content/docs/how-to/make-handlers-idempotent.mdx @@ -15,39 +15,72 @@ the same key. ```ts import { Task, TaskQueue } from "@effectmq/core" import { Effect, Schema } from "effect" +import { + HttpClient, + HttpClientRequest, + HttpClientResponse +} from "effect/unstable/http" -const ChargeError = Schema.Struct({ reason: Schema.String }) +class ChargeRequestFailed extends Schema.TaggedError()( + "ChargeRequestFailed", + { + chargeId: Schema.String, + cause: Schema.Defect() + } +) {} + +const ChargeResponse = Schema.Struct({ chargeId: Schema.String }) const ChargeTask = Task.make({ name: "charge-card", + schemaId: "charge-card/v1", payload: { chargeId: Schema.String, amount: Schema.Number }, success: Schema.String, - error: ChargeError, + error: ChargeRequestFailed, idempotencyKey: ({ chargeId }) => chargeId }) const chargeHandler: TaskQueue.TaskHandler< typeof ChargeTask.payloadSchema, typeof ChargeTask.successSchema, - typeof ChargeTask.errorSchema -> = (task) => - Effect.tryPromise({ - try: async () => { - const response = await fetch("https://payments.example/charges", { - method: "POST", - headers: { - "content-type": "application/json", - "idempotency-key": `charges:${task.id}:${task.generation}` - }, - body: JSON.stringify(task.payload) - }) - if (!response.ok) throw new Error(`payment status ${response.status}`) - return Schema.decodeUnknownSync(Schema.String)(await response.text()) - }, - catch: (cause) => ({ reason: String(cause) }) - }) + typeof ChargeTask.errorSchema, + HttpClient.HttpClient +> = Effect.fn("chargeHandler")( + function* (task) { + return yield* Effect.gen(function* () { + const client = (yield* HttpClient.HttpClient).pipe( + HttpClient.filterStatusOk + ) + const request = yield* HttpClientRequest.post( + "https://payments.example/charges" + ).pipe( + HttpClientRequest.setHeader( + "idempotency-key", + `charges:${task.id}:${task.generation}` + ), + HttpClientRequest.schemaBodyJson(ChargeTask.payloadSchema)(task.payload) + ) + const response = yield* client.execute(request) + const body = yield* HttpClientResponse.schemaBodyJson(ChargeResponse)( + response + ) + return body.chargeId + }).pipe( + Effect.mapError( + (cause) => + new ChargeRequestFailed({ chargeId: task.payload.chargeId, cause }) + ) + ) + } +) ``` +The handler requires `HttpClient.HttpClient`; provide `FetchHttpClient.layer` +once in the application layer graph. Tests can provide a stub client without +patching global `fetch`. The client rejects non-2xx responses in the typed error +channel, and the schema decoder validates the provider response before the +handler returns it. + The downstream service must durably associate that key with the first outcome. If it accepts the side effect and the EffectMQ acknowledgement is lost, the next handler attempt receives the same stored outcome instead of creating a diff --git a/apps/docs/content/docs/how-to/operate-redis.mdx b/apps/docs/content/docs/how-to/operate-redis.mdx index cb36d1a..199d7c5 100644 --- a/apps/docs/content/docs/how-to/operate-redis.mdx +++ b/apps/docs/content/docs/how-to/operate-redis.mdx @@ -12,27 +12,38 @@ Provide the standard live graph with explicit timeouts and pool bounds: ```ts import { TaskEngine } from "@effectmq/core" - -const AppLive = TaskEngine.layer({ - engine: { - prefix: "my-service:effectmq:v1", - maintenanceBatchSize: 100 - }, - redis: { - topology: "standalone", - url: process.env.REDIS_URL, - username: process.env.REDIS_USERNAME, - password: process.env.REDIS_PASSWORD, - socket: { connectTimeout: 5_000 }, - commandOptions: { timeout: 2_000 }, - pool: { - minimum: 1, - maximum: 16, - acquireTimeout: 2_000, - cleanupDelay: 5_000 - } - } -}) +import { Config, Effect, Layer, Redacted } from "effect" + +const AppLive = Layer.unwrap( + Config.all({ + url: Config.string("REDIS_URL"), + username: Config.string("REDIS_USERNAME"), + password: Config.redacted("REDIS_PASSWORD") + }).pipe( + Effect.map(({ password, url, username }) => + TaskEngine.layer({ + engine: { + prefix: "my-service:effectmq:v1", + maintenanceBatchSize: 100 + }, + redis: { + topology: "standalone", + url, + username, + password: Redacted.value(password), + socket: { connectTimeout: 5_000 }, + commandOptions: { timeout: 2_000 }, + pool: { + minimum: 1, + maximum: 16, + acquireTimeout: 2_000, + cleanupDelay: 5_000 + } + } + }) + ) + ) +) ``` Use `rediss://` or node-redis TLS socket options for encrypted connections. diff --git a/apps/docs/content/docs/tutorials/getting-started.mdx b/apps/docs/content/docs/tutorials/getting-started.mdx index f7786b6..d343b17 100644 --- a/apps/docs/content/docs/tutorials/getting-started.mdx +++ b/apps/docs/content/docs/tutorials/getting-started.mdx @@ -37,7 +37,15 @@ docker run --name effectmq-tutorial-redis --publish 6379:6379 --detach redis:8-a Confirm that Redis is ready: ```bash -docker exec effectmq-tutorial-redis redis-cli ping +for ((attempt = 1; attempt <= 30; attempt++)); do + if docker exec effectmq-tutorial-redis redis-cli ping; then + break + fi + if ((attempt == 30)); then + exit 1 + fi + sleep 0.2 +done ``` The command prints: @@ -48,33 +56,16 @@ PONG ## Create the project -Create `package.json`: - -```json -{ - "name": "effectmq-hello", - "private": true, - "type": "module", - "scripts": { - "start": "tsx src/main.ts" - }, - "dependencies": { - "@effect/platform-node": "4.0.0-beta.107", - "@effectmq/core": "0.3.0-rc.0", - "effect": "4.0.0-beta.107", - "tsx": "4.22.4" - } -} -``` - -Install the dependencies and create the source directory: +Initialize the package and install the pinned EffectMQ dependencies: ```bash -pnpm install +pnpm init +pnpm add @effectmq/core@0.3.0-rc.0 effect@4.0.0-beta.107 @effect/platform-node@4.0.0-beta.107 +pnpm add --save-dev tsx@4.22.4 mkdir src ``` -The installation finishes with `Done` and creates `node_modules`. +The installation creates `package.json`, `pnpm-lock.yaml`, and `node_modules`. ## Define the task and queue @@ -82,8 +73,14 @@ Create `src/main.ts`: ```ts import { NodeRuntime } from "@effect/platform-node" -import { Task, TaskEngine, TaskQueue, Worker } from "@effectmq/core" -import { Console, Effect, Schema } from "effect" +import { + NodeRedisPool, + Task, + TaskEngine, + TaskQueue, + Worker +} from "@effectmq/core" +import { Console, Effect, Layer, Schema } from "effect" const Greet = Task.make({ name: "greet", @@ -96,6 +93,12 @@ const Greet = Task.make({ const greetings = TaskQueue.make("tutorial-greetings", Greet) +const EngineLive = TaskEngine.layer().pipe( + Layer.provideMerge( + NodeRedisPool.layer({ url: "redis://127.0.0.1:6379" }) + ) +) + const worker = Worker.make( greetings, ({ payload }) => @@ -125,11 +128,7 @@ const program = Effect.scoped( ) program.pipe( - Effect.provide( - TaskEngine.layer({ - redis: { url: "redis://127.0.0.1:6379" } - }) - ), + Effect.provide(EngineLive), NodeRuntime.runMain ) ``` @@ -142,7 +141,7 @@ operations. Run the application: ```bash -pnpm start +pnpm exec tsx src/main.ts ``` The output should contain: @@ -157,8 +156,8 @@ Notice that the producer receives a `TaskHandle` from `offer` and passes that handle to `wait`. The handle identifies the exact task generation whose result the caller expects. -Run `pnpm start` once more. The stable idempotency key causes the producer to -find the retained generation: +Run `pnpm exec tsx src/main.ts` once more. The stable idempotency key causes +the producer to find the retained generation: ```text Offered Ada as TaskExisting diff --git a/docs/operations.md b/docs/operations.md index 556e9e6..ac3e528 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -15,24 +15,35 @@ Every pool drains in-flight commands and closes when its Effect scope closes. ```ts import { NodeRedisPool } from "@effectmq/core" - -const RedisLive = NodeRedisPool.layer({ - topology: "standalone", - url: process.env.REDIS_URL, - username: process.env.REDIS_USERNAME, - password: process.env.REDIS_PASSWORD, - socket: { - connectTimeout: 5_000, - reconnectStrategy: (attempt) => Math.min(50 * 2 ** attempt, 2_000) - }, - commandOptions: { timeout: 2_000 }, - pool: { - minimum: 1, - maximum: 16, - acquireTimeout: 2_000, - cleanupDelay: 5_000 - } -}) +import { Config, Effect, Layer, Redacted } from "effect" + +const RedisLive = Layer.unwrap( + Config.all({ + url: Config.string("REDIS_URL"), + username: Config.string("REDIS_USERNAME"), + password: Config.redacted("REDIS_PASSWORD") + }).pipe( + Effect.map(({ password, url, username }) => + NodeRedisPool.layer({ + topology: "standalone", + url, + username, + password: Redacted.value(password), + socket: { + connectTimeout: 5_000, + reconnectStrategy: (attempt) => Math.min(50 * 2 ** attempt, 2_000) + }, + commandOptions: { timeout: 2_000 }, + pool: { + minimum: 1, + maximum: 16, + acquireTimeout: 2_000, + cleanupDelay: 5_000 + } + }) + ) + ) +) ``` For TLS, use a `rediss://` URL or node-redis socket TLS options. Supply CA and @@ -42,32 +53,50 @@ logs. For Sentinel, TLS and ACL settings for Redis nodes belong in ```ts import { NodeRedisPool } from "@effectmq/core" - -const SentinelRedisLive = NodeRedisPool.layer({ - topology: "sentinel", - sentinel: { - name: "effectmq-primary", - sentinelRootNodes: [ - { host: "sentinel-a.internal", port: 26379 }, - { host: "sentinel-b.internal", port: 26379 }, - { host: "sentinel-c.internal", port: 26379 } - ], - masterPoolSize: 16, - maxCommandRediscovers: 20, - scanInterval: 1_000, - commandOptions: { timeout: 2_000 }, - nodeClientOptions: { - username: process.env.REDIS_USERNAME, - password: process.env.REDIS_PASSWORD, - socket: { connectTimeout: 5_000 } - }, - sentinelClientOptions: { - username: process.env.SENTINEL_USERNAME, - password: process.env.SENTINEL_PASSWORD, - socket: { connectTimeout: 5_000 } - } - } -}) +import { Config, Effect, Layer, Redacted } from "effect" + +const SentinelRedisLive = Layer.unwrap( + Config.all({ + redisUsername: Config.string("REDIS_USERNAME"), + redisPassword: Config.redacted("REDIS_PASSWORD"), + sentinelUsername: Config.string("SENTINEL_USERNAME"), + sentinelPassword: Config.redacted("SENTINEL_PASSWORD") + }).pipe( + Effect.map( + ({ + redisPassword, + redisUsername, + sentinelPassword, + sentinelUsername + }) => + NodeRedisPool.layer({ + topology: "sentinel", + sentinel: { + name: "effectmq-primary", + sentinelRootNodes: [ + { host: "sentinel-a.internal", port: 26379 }, + { host: "sentinel-b.internal", port: 26379 }, + { host: "sentinel-c.internal", port: 26379 } + ], + masterPoolSize: 16, + maxCommandRediscovers: 20, + scanInterval: 1_000, + commandOptions: { timeout: 2_000 }, + nodeClientOptions: { + username: redisUsername, + password: Redacted.value(redisPassword), + socket: { connectTimeout: 5_000 } + }, + sentinelClientOptions: { + username: sentinelUsername, + password: Redacted.value(sentinelPassword), + socket: { connectTimeout: 5_000 } + } + } + }) + ) + ) +) ``` Use three Sentinel processes across independent failure domains and a quorum