Harden queue runtime correctness boundaries

.changeset/clean-effect-contracts.md

@@ -1,5 +1,5 @@
 ---
-"@effectmq/core": major
+"@effectmq/core": patch
 ---
 
 Make public Effect contracts honest and restructure the package around focused

.changeset/reviewed-runtime-correctness.md

@@ -0,0 +1,23 @@
+---
+"@effectmq/core": patch
+---
+
+Fix queue correctness and boundary validation found during a repository-wide
+review:
+
+- allow `Schema.Void` tasks to persist and recover successful completion;
+- keep `wait` and `execute` subscribed across retryable failures;
+- reject queue/task descriptors that do not match a persisted handle;
+- validate numeric offer, lease, and retry-timestamp inputs before mutating
+  Redis;
+- validate decoded storage values, preserve prototype-sensitive object keys,
+  and reject corrupt Redis numbers and cursors in typed error channels;
+- schema-validate built-in failure events and keep their public type precise;
+- keep stalled-attempt history out of handler retry schedules, settle attempts
+  when retry-policy evaluation fails, stop safely when retained history is too
+  short to replay, and retain terminal failures even when error history is
+  disabled;
+- preserve retry-policy interruption for lease recovery and avoid full
+  wait-list scans on creation, acquisition, and non-waiting transitions; and
+- reject unsafe worker concurrency, timing, and lease supervision options
+  before acquiring work or starting fibers.

.gitignore

@@ -3,6 +3,8 @@ node_modules/
 dist/
 .next/
 out/
+.vercel/
+.env.local
 *.tsbuildinfo
 .history/
 llms/

CLAUDE.md

@@ -14,16 +14,15 @@ pnpm build                        # tsc -> dist/ (tests and src/testing are excl
 pnpm typecheck                    # strict production + complete test typecheck
 pnpm typecheck:src                # production sources only
 pnpm typecheck:test               # all tests and src/testing support
-pnpm exec biome check src/        # lint (what CI runs)
+pnpm lint                         # Biome lint across the repository
+pnpm check                        # formatting, lint, types, architecture, Lua, docs, release
 pnpm lint:fix                     # biome check --write --unsafe
 pnpm test                         # vitest run (integration tests, needs Docker — see below)
 pnpm test:watch                   # vitest watch mode
 pnpm vitest run src/TaskQueue.test.ts        # single test file
 pnpm vitest run -t "test name substring"     # single test by name
 ```
 
-Note: the `pnpm lint` script runs `turbo run lint`, but turbo is not a dependency — it's stale. Use `pnpm exec biome check src/` (matches CI).
-
 ### Tests need Docker
 
 Effectful tests use `@effect/vitest`. Redis integration suites install the suite-scoped `TestLayer` from `src/testing/redisLayer.ts`; it acquires and releases containers, clients, child processes, listeners, and temporary directories through Effect scopes. Never add a module-level `ManagedRuntime`, direct `Effect.runPromise` test runner, or top-level resource warming. Container acquisition has an explicit 60s hook timeout and normal tests have a 30s timeout (`vitest.config.ts`). The root `docker-compose.yml` Redis is for manual/local experimentation only — tests don't use it.
@@ -41,7 +40,7 @@ Flat `src/` with a strict layering, top to bottom:
 - **`TaskQueue.ts`** — the high-level API users live in: `make` (bind a queue name to a task definition), `offer` (enqueue), `complete` (take → run handler → report outcome, applying the task's retry schedule on failure), `stream` (typed lifecycle events), `wait` / `execute` (await a task's terminal result). Decodes engine tasks/events against the task's schemas.
 - **`Scheduler.ts`** — cron-driven durable task materialization. Competing processes idempotently offer the same tick task; managed workers execute it with at-least-once delivery.
 - **`Task.ts`** — `Task.make`: the schema-bearing task definition (payload/success/error schemas, `idempotencyKey` — which *is* the task id, so same key = same task — `retry` as an Effect `Schedule`, `maxRetries` default 5, `null` = unbounded).
-- **`TaskEngine.ts`** (the big one, ~1000 lines) — the low-level `Context.Service` implementing queue primitives as **atomic Lua scripts** (inline `/*lua*/` strings, built by `buildScripts`): create/take/writeSuccess/writeError, lock extend/remove, delayed + cron schedule state. Tasks move between Redis lists: `wait`, `scheduled`, `active`, `failed`, `success`. Every state change publishes to a per-queue Redis Stream (`<prefix>:<name>:events`, via `XADD`); `stream` polls it with `XREAD` (default 1s). Consumers rarely call the engine directly — go through `TaskQueue`/`Scheduler`.
+- **`TaskEngine.ts`** (the big one, ~1000 lines) — the low-level `Context.Service` implementing queue primitives through the atomic script generated from `src/lua/taskEngine.lua`: create/take/writeSuccess/writeError, lock extend/remove, delayed + cron schedule state. Tasks move between Redis indexes named `wait`, `scheduled`, `active`, `failed`, and `success`. Every state change publishes to a per-queue Redis Stream (`<prefix>:<name>:events`, via `XADD`); `stream` uses a blocking `XREAD` with a default two-second poll interval. Consumers rarely call the engine directly — go through `TaskQueue`/`Scheduler`.
 - **`RedisPool.ts`** — the minimal service the engine depends on: just `send` + `eval`. Any Redis client can implement it.
 - **`NodeRedisPool.ts`** — the bundled `RedisPool` implementation using node-redis `createClientPool` (lazy connect, closed on layer scope end). Tests provide `RedisPool` via ioredis + testcontainers instead (`src/testing/redisLayer.ts`) — proof the service boundary works.
 - **`TaskRecord.ts`** — public typed task record schemas and storage codecs.
@@ -50,7 +49,10 @@ Flat `src/` with a strict layering, top to bottom:
 
 Wiring: `TaskEngine.layer()` is the complete zero-requirement Node live graph and intentionally retains Redis operational services. `TaskEngine.layerNoDeps()` is the custom-client layer that requires `RedisPool`.
 
-The library deliberately has **no built-in concurrency/rate limiting** — one `complete` processes one task, and callers compose concurrency from Effect primitives (fibers, semaphores, schedules). Don't add worker-pool machinery.
+`Worker` provides built-in bounded local concurrency, lease supervision,
+maintenance, and graceful draining. It does not provide distributed/global
+concurrency or rate limiting; compose those policies explicitly at the
+application boundary.
 
 ## Conventions
 

README.md

@@ -85,8 +85,8 @@ A task is a *schema*, not a function. You declare what goes in (`payload`), what
 A tagged error makes failures pattern-matchable downstream, so reach for `Schema.TaggedError` rather than a bare struct.
 
 ```ts docs-check=email
-import { Effect, Schedule, Schema, Semaphore, Stream } from "effect";
-import { Task, TaskQueue, type TaskHandler } from "@effectmq/core";
+import { Effect, Schedule, Schema, Stream } from "effect";
+import { Task, TaskQueue, type TaskHandler, Worker } from "@effectmq/core";
 
 class EmailRejected extends Schema.TaggedError<EmailRejected>()(
   "EmailRejected",
@@ -158,7 +158,7 @@ const repeatedWorkerProgram = Effect.gen(function* () {
 
 ## Streaming & events
 
-The engine publishes a lifecycle event to a per-queue Redis Stream every time a task changes state. `TaskQueue.stream` hands you those events as an Effect `Stream`, decoded against your queue's schemas: `task.created` and `task.updated` carry fully-typed tasks, `task.failed` carries your typed error, `task.completed` carries your typed success value, and `task.moved` reports the list transition.
+The engine publishes a lifecycle event to a per-queue Redis Stream every time a task changes state. `TaskQueue.stream` hands you those events as an Effect `Stream`, decoded against your queue's schemas: `task.created` and `task.updated` carry fully-typed tasks, `task.failed` carries your typed error or a built-in stalled/canceled error, `task.completed` carries your typed success value, and `task.moved` reports the list transition.
 
 ```ts docs-check=email
 const watch = TaskQueue.stream(emails).pipe(
@@ -185,33 +185,26 @@ const offerAndWait = Effect.gen(function* () {
 });
 ```
 
-`wait` reads durable state, subscribes from the handle's authoritative Redis cursor, and rechecks state after subscription, so completion before or during subscription is observed. Streams poll Redis (default every second); persist a retained cursor when building a resumable event consumer.
+`wait` reads durable state, subscribes from the handle's authoritative Redis cursor, and rechecks state after subscription, so completion before or during subscription is observed. Streams use a blocking Redis read (default two-second block); persist a retained cursor when building a resumable event consumer.
 
 ---
 
 ## 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:
+`complete` processes exactly one task. For a long-running process, `Worker`
+provides bounded local concurrency, lease supervision, maintenance, and graceful
+draining:
 
 ```ts docs-check=email
-// 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, handleSendEmail),
-  ).pipe(
-    Effect.forkScoped,            // each worker is its own fiber
-    Effect.repeat(Schedule.forever), // ...that keeps pulling work
-  );
-});
+const worker = Worker.make(emails, handleSendEmail, { concurrency: 5 });
+const program = Worker.run(worker);
 ```
 
-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? Run more worker processes. None of it is our invention, all of it composes. The queue's job is to preserve eligible work and fence the current attempt; handlers remain at-least-once and must make external side effects idempotent.
+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.
 
 ---
 

biome.json

@@ -10,7 +10,8 @@
     "includes": [
       "**",
       "!node_modules",
-      "!.next",
+      "!**/.next",
+      "!**/.vercel",
       "!dist",
       "!build",
       "!src/scratchpad",

docs/api-reference.md

@@ -25,20 +25,23 @@ page describes the intended entry points and their contracts.
 - `execute(queue, payload, options?)` is offer followed by handle-based wait.
 - `completeOne(queue, handler, processing?)` acquires and supervises at most one
   attempt, returning whether work was processed.
-- `complete(queue, handler, processing?)` processes one task and returns its id.
+- `complete(queue, handler)` processes one task and returns its id.
 - `stream(queue, options?)` decodes versioned lifecycle events from a cursor.
 
 Important offer options include `taskId`, `delay`, `maxRetries`, completion
 policies, `onDuplicate`, and
-`retainResultUntil: "current-task-settles"`. Processing options configure lease
-duration, heartbeat interval, and bounded heartbeat transport retry.
+`retainResultUntil: "current-task-settles"`. Numeric overrides are validated
+before Redis is mutated. Processing options configure lease duration,
+heartbeat interval, and bounded heartbeat transport retry.
 
 ## `Worker`
 
 - `make(queue, handler, options?)` describes a managed worker.
 - `run(worker)` runs scoped acquisition slots plus maintenance until
   interrupted. Options include concurrency, poll/maintenance intervals, drain
-  timeout, and processing supervision.
+  timeout, and processing supervision. Invalid concurrency, durations, or
+  processing settings fail with `WorkerConfigurationError` before any fibers
+  start.
 
 ## `Scheduler`
 

docs/storage-protocol-v1.md

@@ -43,13 +43,16 @@ reading their values, and fails when the old namespace has not been drained.
 After a task's Effect Schema has encoded it, v1 accepts:
 
 - `null`, strings, booleans, and `Uint8Array`
-- finite numbers whose magnitude is at most `Number.MAX_SAFE_INTEGER`
+- finite numbers whose magnitude is at most `Number.MAX_SAFE_INTEGER`, except
+  negative zero
 - arrays containing supported values
 - plain string-keyed objects containing supported values
 
 Empty arrays and objects, nested nulls, Unicode, binary bytes, and fractional
-safe numbers round-trip losslessly. `undefined`, `bigint`, non-finite and unsafe
-numbers, class instances, symbols, functions, and cyclic objects are rejected.
+safe numbers round-trip losslessly. A top-level `undefined` success is accepted
+for `Schema.Void`; `undefined` payloads, failures, and nested values remain
+invalid. Negative zero, `bigint`, non-finite and unsafe numbers, class
+instances, symbols, functions, and cyclic objects are rejected.
 Encoded user values are limited to 1 MiB by default. A task definition can
 override `storageLimits.maxValueBytes`; the same bound applies independently to
 payload, success, and typed-failure envelopes.

src/EngineRecord.ts

@@ -38,16 +38,58 @@ export const TextFromBytes = Schema.Unknown.pipe(
   }),
 );
 
+const numberFromText = (integer: boolean) =>
+  SchemaGetter.transformOrFail((value: string, options) => {
+    const pattern = integer
+      ? /^-?(?:0|[1-9]\d*)$/
+      : /^-?(?:0|[1-9]\d*)(?:\.\d+)?(?:[eE][+-]?\d+)?$/;
+    const parsed = Number(value);
+    return pattern.test(value) &&
+      Number.isFinite(parsed) &&
+      Math.abs(parsed) <= Number.MAX_SAFE_INTEGER
+      ? Effect.succeed(parsed)
+      : Effect.fail(
+          new SchemaIssue.InvalidValue(
+            {
+              message: integer
+                ? "Expected a safe decimal integer"
+                : "Expected a safe decimal number",
+            },
+            value,
+            options,
+          ),
+        );
+  });
+
 export const NumberFromBytes = TextFromBytes.pipe(
-  Schema.decodeTo(Schema.Number, {
-    decode: SchemaGetter.transform(Number),
+  Schema.decodeTo(Schema.Finite, {
+    decode: numberFromText(false),
+    encode: SchemaGetter.transform(String),
+  }),
+);
+
+export const IntegerFromBytes = TextFromBytes.pipe(
+  Schema.decodeTo(Schema.Int, {
+    decode: numberFromText(true),
     encode: SchemaGetter.transform(String),
   }),
 );
 
 export const BooleanFromBytes = TextFromBytes.pipe(
   Schema.decodeTo(Schema.Boolean, {
-    decode: SchemaGetter.transform((value) => value === "1"),
+    decode: SchemaGetter.transformOrFail((value, options) =>
+      value === "1"
+        ? Effect.succeed(true)
+        : value === "0"
+          ? Effect.succeed(false)
+          : Effect.fail(
+              new SchemaIssue.InvalidValue(
+                { message: 'Expected Redis boolean "0" or "1"' },
+                value,
+                options,
+              ),
+            ),
+    ),
     encode: SchemaGetter.transform((value) => (value ? "1" : "0")),
   }),
 );
@@ -57,28 +99,28 @@ const msgpackListFromBytes = <S extends Schema.Top>(item: S) =>
 
 export const EngineTaskSchema = Schema.Struct({
   id: TextFromBytes,
-  protocolVersion: NumberFromBytes,
+  protocolVersion: IntegerFromBytes,
   schemaId: TextFromBytes,
-  generation: NumberFromBytes,
+  generation: IntegerFromBytes,
   name: TextFromBytes,
   delay: NumberFromBytes,
-  maxRetries: NumberFromBytes,
-  maxStalledCount: NumberFromBytes,
-  maxErrorEntries: NumberFromBytes,
-  maxRelationships: NumberFromBytes,
-  maxEventEntries: NumberFromBytes,
-  taskRecordRetentionMs: NumberFromBytes,
-  resultRetentionMs: NumberFromBytes,
-  terminalIndexRetentionMs: NumberFromBytes,
-  deadLetterRetentionMs: NumberFromBytes,
-  eventRetentionMs: NumberFromBytes,
-  attempt: NumberFromBytes,
-  handlerFailureCount: NumberFromBytes,
-  stalledAttemptCount: NumberFromBytes,
+  maxRetries: IntegerFromBytes,
+  maxStalledCount: IntegerFromBytes,
+  maxErrorEntries: IntegerFromBytes,
+  maxRelationships: IntegerFromBytes,
+  maxEventEntries: IntegerFromBytes,
+  taskRecordRetentionMs: IntegerFromBytes,
+  resultRetentionMs: IntegerFromBytes,
+  terminalIndexRetentionMs: IntegerFromBytes,
+  deadLetterRetentionMs: IntegerFromBytes,
+  eventRetentionMs: IntegerFromBytes,
+  attempt: IntegerFromBytes,
+  handlerFailureCount: IntegerFromBytes,
+  stalledAttemptCount: IntegerFromBytes,
   onSuccessPolicy: TextFromBytes.pipe(Schema.decodeTo(CompletionPolicySchema)),
   onFailurePolicy: TextFromBytes.pipe(Schema.decodeTo(CompletionPolicySchema)),
-  createdAt: NumberFromBytes.pipe(Schema.decodeTo(DateFromNumberSchema)),
-  updatedAt: NumberFromBytes.pipe(Schema.decodeTo(DateFromNumberSchema)),
+  createdAt: IntegerFromBytes.pipe(Schema.decodeTo(DateFromNumberSchema)),
+  updatedAt: IntegerFromBytes.pipe(Schema.decodeTo(DateFromNumberSchema)),
   payload: UnknownFromMsgpack,
   success: UnknownFromMsgpack.pipe(Schema.optional),
   errors: msgpackListFromBytes(
@@ -100,11 +142,11 @@ export const EngineTaskSchema = Schema.Struct({
 export type EngineTask = typeof EngineTaskSchema.Type;
 
 export const EngineTerminalResultSchema = Schema.Struct({
-  protocolVersion: NumberFromBytes,
+  protocolVersion: IntegerFromBytes,
   schemaId: TextFromBytes,
-  generation: NumberFromBytes,
+  generation: IntegerFromBytes,
   outcome: TextFromBytes.pipe(Schema.decodeTo(TaskOutcomeSchema)),
-  settledAt: NumberFromBytes,
+  settledAt: IntegerFromBytes,
   success: UnknownFromMsgpack.pipe(Schema.optional),
   failure: UnknownFromMsgpack.pipe(Schema.optional),
 });

src/PublicContracts.test.ts

@@ -1,8 +1,9 @@
-import { expect, it } from "vitest";
 import * as Context from "effect/Context";
 import type * as Effect from "effect/Effect";
 import type * as Layer from "effect/Layer";
 import * as Schema from "effect/Schema";
+import type * as Stream from "effect/Stream";
+import { expect, it } from "vitest";
 import type * as RedisPool from "./RedisPool.js";
 import * as TaskEngine from "./TaskEngine.js";
 import * as TaskQueue from "./TaskQueue.js";
@@ -71,7 +72,7 @@ const compilePublicContracts = () => {
     Equal<Effect.Success<typeof completeOne>, boolean>
   >;
   type CompleteOneError = Expect<
-    Equal<Effect.Error<typeof completeOne>, TaskQueue.CompleteError>
+    Equal<Effect.Error<typeof completeOne>, TaskQueue.CompleteOneError>
   >;
   type CompleteOneServices = Expect<
     Equal<Effect.Services<typeof completeOne>, Effect.Services<typeof complete>>
@@ -135,6 +136,18 @@ const compilePublicContracts = () => {
   >;
   type RejectAnyServices = ExpectFalse<IsAny<Effect.Services<typeof complete>>>;
 
+  const events = TaskQueue.stream(queue);
+  type FailedEvent = Extract<
+    Stream.Success<typeof events>,
+    { readonly _tag: "task.failed" }
+  >;
+  type FailedEventError = Expect<
+    Equal<
+      FailedEvent["payload"]["error"],
+      boolean | TaskRecord.StalledErrorSchema | TaskRecord.CanceledErrorSchema
+    >
+  >;
+
   const layerNoDeps = TaskEngine.layerNoDeps();
   const liveLayer = TaskEngine.layer();
   type LayerNoDepsRequirement = Expect<
@@ -164,6 +177,7 @@ const compilePublicContracts = () => {
     | RejectUnknownError
     | RejectErasedServices
     | RejectAnyServices
+    | FailedEventError
     | LayerNoDepsRequirement
     | LiveLayerRequirement;
 };

src/RetrySchedule.ts

@@ -49,7 +49,11 @@ export const nextRunAt = Effect.fnUntraced(function* <R>(
       (value) => Effect.succeed([value, -1] as const),
     );
     if (delay === -1) return undefined;
-    time = error.timestamp.getTime() + Duration.toMillis(delay);
+    const delayMillis = Duration.toMillis(delay);
+    time = error.timestamp.getTime() + delayMillis;
+    if (!Number.isFinite(time) || Math.abs(time) > Number.MAX_SAFE_INTEGER) {
+      return undefined;
+    }
   }
   return time;
 });