Error Handling

Act defines four primary error types. Each signals a different class of problem with a distinct resolution strategy.

ValidationError

Thrown when an action or event payload fails Zod schema validation.

import { ValidationError } from "@rotorsoft/act";

try {
  await app.do("createUser", target, { email: 123 }); // wrong type
} catch (error) {
  if (error instanceof ValidationError) {
    console.error("Invalid payload:", error.details);
  }
}

Resolution: Fix the payload to match the schema. This is always a caller error.

PreconditionError

Thrown when a precondition from .requires() refuses an action before events are emitted. The message is "<action> failed precondition: <description>", and description carries the rule's description. A broken invariant (the state's schema, checked before every commit) throws ValidationError instead.

import { PreconditionError } from "@rotorsoft/act";

try {
  await app.do("CloseTicket", target, { reason: "Done" });
} catch (error) {
  if (error instanceof PreconditionError) {
    console.error("Rule violated:", error.description);
    console.error("Current state:", error.snapshot.state);
  }
}

Resolution: Check preconditions before dispatching, or handle gracefully in the UI. The state was not modified.

ConcurrencyError

Thrown when optimistic concurrency control detects a conflict — another process committed events to a stream the action appends to between your load() and commit().

import { ConcurrencyError } from "@rotorsoft/act";

try {
  await app.do("increment", target, { by: 1 });
} catch (error) {
  if (error instanceof ConcurrencyError) {
    console.error(`Stream ${error.stream}: expected v${error.expectedVersion}, found v${error.lastVersion}`);
  }
}

Resolution: Retry with fresh state. The cache is invalidated automatically on concurrency errors.

Retry Pattern — per-action policy

Declare the retry budget on the action itself. The orchestrator owns the loop: on ConcurrencyError it invalidates the cache, applies an optional backoff, and re-runs from load(). Any other error rethrows immediately and does not consume the budget.

The loop only retries framework-derived versions — where a concurrent writer advanced the head and the reload picks up the new head version. When the caller pins target.expectedVersion explicitly, a ConcurrencyError is a fixed conflict: every retry reloads and re-commits against the same pinned version and is guaranteed to fail again. So a caller-pinned conflict rethrows immediately, without burning the budget or sleeping out the backoff.

import { state } from "@rotorsoft/act";
import { z } from "zod";

const BankAccount = state({ BankAccount: z.object({ balance: z.number() }) })
  .init(() => ({ balance: 0 }))
  .emits({ Transferred: z.object({ amount: z.number() }) })
  .on(
    { transfer: z.object({ amount: z.number() }) },
    {
      maxRetries: 5,
      backoff: { strategy: "exponential", baseMs: 10, maxMs: 200, jitter: true },
    },
  )
    .emit((action) => ["Transferred", { amount: action.amount }])
  .build();

// Caller is unchanged — the retry is invisible to them.
await app.do("transfer", target, { amount: 100 });

The action author knows whether the action contends for a hot stream; the caller shouldn't have to. The same call site works for low-contention actions (omit options, surface ConcurrencyError on first conflict) and hot-stream actions (declare a budget, retry transparently).

maxRetries defaults to 0 (single attempt, current behavior). When backoff is omitted, retries run immediately — fine at low contention. On hot streams, jittered exponential backoff avoids the thundering herd of N writers re-racing in lockstep.

When to wrap manually instead

If you need different retry behavior than what the action declares — for instance, a UI mutation that should fail fast and surface to the user even on a hot action — wrap the call:

async function withRetry(action, target, payload, maxRetries) {
  for (let attempt = 0; attempt <= maxRetries; attempt++) {
    try {
      return await app.do(action, target, payload);
    } catch (error) {
      if (error instanceof ConcurrencyError && attempt < maxRetries) {
        continue;
      }
      throw error;
    }
  }
}

This is an escape hatch — the declarative form on the action is the primary mechanism.

StreamClosedError

Thrown when an action targets a stream that has been permanently closed (its head event is a __tombstone__). See Close cycle for how a stream gets tombstoned.

import { StreamClosedError } from "@rotorsoft/act";

try {
  await app.do("AddNote", target, { text: "..." });
} catch (error) {
  if (error instanceof StreamClosedError) {
    console.error(`Stream ${error.stream} is closed`);
  }
}

Resolution: Closed streams are terminal. To re-open one, call app.close([{ stream, restart: true }]) — that seeds a fresh __snapshot__ and the stream accepts actions again.

StoreError

Thrown by a Store adapter when an infrastructure operation fails — a dropped connection, a transaction rollback, a query timeout. It's the typed boundary between "the store is unavailable/degraded" and the domain errors above (ConcurrencyError, StreamClosedError), which describe legitimate outcomes you branch on. The original driver error is preserved on cause, and the failed operation name on operation.

import { StoreError } from "@rotorsoft/act";

try {
  await app.do("AddNote", target, { text: "..." });
} catch (error) {
  if (error instanceof StoreError) {
    console.error(`store ${error.operation} failed`, error.cause);
  }
}

Adapters throw StoreError from claim / ack / block / subscribe (and the Postgres/SQLite adapters from their commit paths) instead of silently returning empty results — so a degraded backend can't be mistaken for "no work."

Store failures and the circuit breaker

The drain, settle, and autoclose loops all hit the store. When it goes down, every attempt throws a StoreError. Rather than hammer a dead database, each Act owns a circuit breaker shared by those loops:

• closed → normal; consecutive failures are counted.
• open → after failureThreshold consecutive failures, any drain/settle/autoclose attempt skips the store while open.
• half-open → once cooldownMs has passed, the next attempt is a trial; a pass closes the breaker, a failure re-opens it.

After every store failure, open or not, the breaker schedules one retry (a settle) cooldownMs later (plus the settle debounce, settleDebounceMs), so recovery is automatic: you don't need a new commit, a periodic poller, or a manual drain(). While open, the loops simply skip the store; the retry re-trials it and the breaker closes as soon as the store is healthy again. A failed drain pass also keeps its controller armed and retries on its own, never before the leases it took expire, since only then can another pass claim those streams.

If the store stays down, the breaker keeps probing — one attempt per cooldownMs, indefinitely (always exactly one pending timer, unref()'d so it never blocks process exit). It never gives up, so it recovers whenever the store returns; the trade-off is that each failed probe re-emits the error event (an outage heartbeat — raise cooldownMs or dedupe in your handler if that's too chatty). The timer is cleared on the first successful probe and on dispose(). Tune it via ActOptions.circuitBreaker (defaults: threshold 5, cooldown 30s).

Subscribe to the error lifecycle event to alert on a degraded store — it fires on every failed store cycle with the error and the breaker state:

app.on("error", ({ error, circuit }) => {
  if (error instanceof StoreError)
    metrics.increment("act.store_error", { op: error.operation, circuit });
  // circuit === "open" means the loops have backed off and will retry after the cooldown
});

The framework logs the error regardless of listeners; the event is emitted only when a listener is registered (Node's EventEmitter rethrows an unhandled "error").

Store failures only

The breaker counts store failures, not errors in your code. When application code fails inside one of the loops, the error stays in the loop that ran it, stops only what it breaks, and is logged a bounded number of times. It never emits error or opens the breaker:

• A resolver throws while correlate reads new events. The scan stops just before that event and is logged once per reaction and event. New events behind it are not matched to reactions; work already found keeps draining. Every later trigger (a commit, a settle) retries from that event, so nothing is lost: once the resolver stops throwing, correlation picks up where it stopped.
• A resolver throws while the drain matches a leased stream's events. Only that stream fails, exactly like a failing handler: it retries on the reaction's backoff and blocks once maxRetries is spent (with blockOnError). A dynamic target without a source reads every stream's events, so one bad event fails every such target. Give resolvers a source where you can.
• A reaction-requested close fails for any reason other than a StoreError (its archive callback throws, say). It is logged once and not retried; close the stream with app.close(...) once the cause is fixed.

Reaction handlers were already contained this way: a throwing handler retries and blocks its own stream. That includes a handler that throws a DeferSignal whose schedule the schema rejects ({ after: { minutes: 0 } }, an invalid Date): it fails like any other error instead of deferring.

All the loops share one count of consecutive store failures, and a success from any of them resets it. A store that fails only one kind of call (a single query timing out, say) while the others work can therefore keep the breaker closed. Each failure still logs and emits error, so alert on the rate of error events, broken down by error.operation, not on circuit === "open" alone.

Error Constants

Each error class sets its name to one of these constants, which is what serialized or logged errors carry. Match a thrown error in code with instanceof; the constants are on name, not message.

import { Errors } from "@rotorsoft/act";

// Errors.ValidationError    = "ERR_VALIDATION"
// Errors.PreconditionError  = "ERR_PRECONDITION"
// Errors.ConcurrencyError   = "ERR_CONCURRENCY"
// Errors.StreamClosedError  = "ERR_STREAM_CLOSED"
// Errors.NonRetryableError  = "ERR_NON_RETRYABLE"
// Errors.StoreError         = "ERR_STORE"

Production Error Handling

import {
  ConcurrencyError,
  PreconditionError,
  ValidationError,
} from "@rotorsoft/act";

// tRPC mutation
CreateItem: authedProcedure
  .input(z.object({ name: z.string() }))
  .mutation(async ({ input, ctx }) => {
    try {
      const snaps = await app.do("CreateItem", { stream: id, actor: ctx.actor }, input);
      // the commit settles its reactions on its own
      return { success: true, id };
    } catch (error) {
      if (error instanceof ValidationError) {
        throw new TRPCError({ code: "BAD_REQUEST", message: "Invalid input" });
      }
      if (error instanceof PreconditionError) {
        throw new TRPCError({ code: "PRECONDITION_FAILED", message: error.description });
      }
      if (error instanceof ConcurrencyError) {
        throw new TRPCError({ code: "CONFLICT", message: "Please retry" });
      }
      throw error;
    }
  }),

Blocked Streams

Streams block on three paths:

1. A reaction handler fails repeatedly and lease.retry reaches maxRetries (after maxRetries + 1 failed attempts). The lease is committed with blocked = true and stays out of claim() results.
2. A reaction handler throws NonRetryableError (or a subclass like NonRetryableWebhookError) — the drain finalizer blocks the stream on the first failed attempt without consuming the retry budget. See Non-retryable errors.
3. A reaction handler never fails but never finishes inside its lease. Each attempt completes, submits an ack the store drops (the lease is gone), and the next claim bumps retry again — the budget is spent without a single error, so paths 1 and 2 can't fire. The drain blocks such a stream at claim time, before dispatching, once retry is strictly greater than maxRetries. The fix is leaseMillis, not maxRetries: put the reaction on a lane whose leaseMillis is above the handler's real duration, then app.unblock. A leaseMillis passed to one app.drain() call covers only that call; the background drains (settle, the defer wake, cycleMs, start_correlations) use the lane's. See When every attempt loses its lease.

A failing store is deliberately not a fourth path. claim writes the retry count up before any handler runs, so a store that fails partway through a pass raises the count with no handler behind it, and path 3 would fire on that. Path 3 therefore stands down while the store is failing. See When the database, not the handler, spent the count.

Recovery uses app.unblock(input) (resume from where the stream stopped) or app.reset(input) (rebuild from event 0). Both accept either an explicit string[] or a StreamFilter for bulk operations. See Recovering a blocked stream and Discovering blocked streams.

Monitor blocked streams via the "blocked" lifecycle event:

app.on("blocked", (blocked) => {
  blocked.forEach(({ stream, error, retry }) => {
    console.error(`Stream ${stream} blocked after ${retry} retries: ${error}`);
    // Alert, log to monitoring, etc.
  });
});

Debugging

When something doesn't behave as expected, three knobs cover most cases.

Verbose logging. Set LOG_LEVEL=debug (or trace) before starting the process. The trace level wires breadcrumb logs into the load/action/drain hot paths via the tracing module:

LOG_LEVEL=trace pnpm dev

Lifecycle event subscriptions. Every Act instance emits a fixed set of lifecycle events; subscribe in dev to see what the framework is doing:

app.on("committed", (snaps) => console.log("committed", snaps.map((s) => s.event?.name)));
app.on("acked", (leases) => console.log("acked", leases));
app.on("blocked", (blocked) => console.error("blocked", blocked));
app.on("settled", (drain) => console.log("settled", drain));
app.on("closed", (result) => console.log("closed", result));
app.on("notified", (n) => console.log("cross-process commit", n));  // PostgresStore notify only
app.on("error", ({ error, circuit }) => console.error("store failure", error, circuit));

Direct event inspection. Bypass cache and reducers and look at what's actually in the store:

// All events on a stream (regex match by default)
const events = await app.query_array({ stream: "order-123" });

// Exact-string stream match — what load() uses internally
const exact = await app.query_array({ stream: "order-123", stream_exact: true });

// Time-travel: state as of a specific event id (see Cache and snapshots)
const snap = await app.load(Order, "order-123", undefined, { before: 5000 });

For introspecting reaction watermarks (per-stream at, retry, blocked, leased_by/leased_until) without taking a lease, use store().query_streams(...). The act-inspector tool is built on this primitive.

Per-reaction options

Each reaction handler accepts options that control retry and blocking behaviour:

.on("OrderPlaced")
  .do(handler, {
    maxRetries: 5,
    blockOnError: true,
    backoff: { strategy: "exponential", baseMs: 200, maxMs: 30_000, jitter: true },
  })
  .to(resolver)

• maxRetries (default 3) — how many times the framework re-claims a stream after a handler throws. Each failed cycle increments retry_count; the next claim() picks the stream up again with the same events.
• blockOnError (default true) — once a failure happens with retry_count at maxRetries (the maxRetries + 1th attempt), the framework calls block() to set blocked = true on the stream. Set false if your handler is idempotent and you'd rather keep retrying forever.
• backoff (default: a fixed 10 seconds) — paces inter-attempt timing so flaky receivers aren't hammered. See Backoff.

Set maxRetries: 0 for handlers that should never retry — typically those that already implement their own dead-letter strategy.

Backoff

Every failed attempt is paced by a backoff, persisted as a defer schedule on the stream, so no worker re-dispatches it until the delay elapses. A reaction that declares no backoff waits a fixed 10 seconds. Retry timing belongs to the reaction: leaseMillis never sets it, so a short lease can't turn a brief outage into an exhausted retry budget. Projections, which take no options, and a stream whose read fails use the same 10-second default. Declare a backoff for handlers that talk to external systems (HTTP, queues, third-party APIs):

backoff: {
  strategy: "exponential",  // "fixed" | "linear" | "exponential"
  baseMs: 200,              // base delay
  maxMs: 30_000,            // cap (only used by exponential)
  jitter: true,             // multiply by random factor in [0.5, 1.5)
}

Delay computation, where retry is the lease's retry counter at the failed attempt (0 is the first failure):

Strategy	Delay
fixed	baseMs
linear	baseMs * (retry + 1)
exponential	min(baseMs * 2^retry, maxMs)

With jitter: true, the final delay is multiplied by 0.5 + random() (range [0.5, 1.5)) to avoid lockstep thundering herds.

Per-stream schedule

A retry-with-backoff persists deferred_at = now + delay on the stream through a due-marked ack — the same store-level mechanism as an explicit defer, with one difference: the due-ack carries the climbing retry counter, so the retry budget keeps accruing toward blockOnError, whereas an explicit defer passes retry: -1 (a defer is a deliberate re-visit, not a failure).

Because the schedule lives in the store rather than in worker memory:

• Every competing worker honors the window — the stream is excluded from claim until deferred_at, so no worker re-attempts or re-dispatches during the delay.
• No worker re-claims the stream mid-window, so retry advances exactly once per real attempt. A stream blocks after maxRetries + 1 attempts (the first, then maxRetries retries), independent of worker count.
• The schedule is durable: a worker that restarts re-arms its drain at the persisted deferred_at.

On a partial-progress drain — a claimed stream carries several events and the handler succeeds on an earlier one before a later one fails — the due-ack advances the watermark to the last successfully-handled event and persists the window in one atomic entry (advance and defer are independent legs of ack). The succeeded prefix is never re-run when the stream is re-claimed after the window; only the failing tail is retried (#1278).

Independent of leaseMillis

The due-ack releases the lease and hands the store the schedule, so the backoff window is decoupled from leaseMillis: the effective delay is exactly the configured backoff, whether that's shorter or longer than the lease. (Earlier versions held the lease through the window, which floored the effective backoff up to leaseMillis and let mid-window re-claims phantom-bump the retry counter; the persisted schedule is now the sole gate — see #1262.)

Webhook delivery — @rotorsoft/act-http/webhook

The 80% pattern for external integration is "POST this event to a URL." Every team writes the same fetch wrapper — timeout, idempotency key, status-coded errors, JSON serialization. The @rotorsoft/act-http umbrella package ships that wrapper as webhook(), a reaction-handler factory that composes with the maxRetries / backoff options above:

import { webhook } from "@rotorsoft/act-http/webhook";

.on("OrderConfirmed")
  .do(
    webhook({
      url: "https://api.example.com/webhooks/orders",
      headers: (event) => ({ Authorization: "Bearer " + token }),
      body: (event) => ({ orderId: event.stream, total: event.data.total }),
      timeoutMs: 2_000,
    }),
    {
      maxRetries: 5,
      backoff: { strategy: "exponential", baseMs: 200, maxMs: 30_000, jitter: true },
    }
  )
  .to(resolver)

Behavior:

• POST by default; method configurable.
• Idempotency-Key derived from event.id (overridable per call, or return null to skip).
• 5xx, network errors, and timeouts throw WebhookError → drain retries per maxRetries / backoff.
• 4xx throws NonRetryableWebhookError (a subclass of NonRetryableError) → the drain finalizer blocks the stream on the first failed attempt when blockOnError is true. No wasted retries on permanent client errors.
• fetch is injectable for tests.

The two-class split lets handlers signal recoverability through the type system. NonRetryableError (exported from @rotorsoft/act) is the general primitive — any handler can throw it to bypass the retry budget for known-permanent failures (validation errors, "user deleted" 404s, business-rule violations). See Non-retryable errors below.

When webhook fits — and when it doesn't

webhook is built for fire-and-forget delivery to a cooperative receiver: short timeouts, retries paced by backoff, and idempotent endpoints that can absorb the occasional duplicate.

Keep timeoutMs below leaseMillis. The drain lease is what stops competing workers from re-dispatching while your handler is still in flight. If timeoutMs approaches or exceeds the lease, a slow receiver can hold the lease through expiry, at which point another worker will claim the stream and POST the same event in parallel. The downstream Idempotency-Key then becomes load-bearing — if your receiver doesn't dedup, you'll deliver twice. Rule of thumb: timeoutMs ≤ leaseMillis - safety_margin. If you need a longer window, bump leaseMillis globally on the Act options.

For heavy or long-running delivery, don't use webhook directly. Drain leases aren't free, and holding one for tens of seconds while a slow API churns is the wrong shape. The Act-native pattern is an outbox-style fan-out: emit a small "needs delivery" event (a cheap, local operation), and let a separate consumer — a downstream worker, a Kafka/SQS pipeline, an external scheduler — pick it up and do the long work at its own pace. Drain stays responsive; the slow path runs at its own schedule. See external integration (forthcoming) for the outbox pattern in detail.

Shape of work	Right tool
1–2s POST to a fast, idempotent API	webhook directly
Flaky-but-fast third party	webhook + aggressive backoff
Multi-second / multi-minute API call	Emit an event, drain hands off to a bus; bus worker calls the API
Bulk fan-out (10k+ receivers)	Emit a "fanout" event, let a dedicated consumer enumerate receivers
Streaming / long-poll / large file transfer	Not webhook — write a dedicated worker

Non-retryable errors

The drain pipeline retries on any thrown error by default — maxRetries is a budget, not a classifier. For failures the handler knows won't recover on retry — a 4xx from a webhook, a ZodError on malformed input, a "user deleted" 404, a business-rule violation — throwing a generic Error wastes the budget and delays the operator signal.

NonRetryableError (exported from @rotorsoft/act) is the handler-side signal. The drain finalizer checks error instanceof NonRetryableError and forces block = blockOnError regardless of lease.retry. The stream blocks on the first failed attempt; no retries, no backoff window.

import { NonRetryableError } from "@rotorsoft/act";

.on("PaymentReceived")
  .do(async function recordPayment(event) {
    const parsed = PaymentSchema.safeParse(event.data);
    if (!parsed.success) {
      throw new NonRetryableError("payment payload failed validation", {
        cause: parsed.error,
      });
    }
    // ... handle the parsed payload
  })

Important: NonRetryableError does not override blockOnError: false. If the operator has explicitly chosen "never block, retry forever," the framework respects that — NonRetryableError becomes equivalent to any other error. The class signal only matters on the block-when-budget-exhausted path.

@rotorsoft/act-http/webhook exports NonRetryableWebhookError (a subclass) for 4xx responses. The split lets generic catch sites use instanceof NonRetryableError while webhook-aware code reads the HTTP-specific status / url / responseBody fields.

Recovering a blocked stream — app.unblock

When a stream blocks — whether from NonRetryableError (first attempt) or from exhausting maxRetries — the operator's recovery path is app.unblock(input). The input is either an explicit list of stream names or a StreamFilter for bulk recovery:

// Single targeted recovery — by name.
await app.unblock(["webhooks-out-customer-42"]);

// Bulk recovery — by filter (all blocked streams matching a pattern).
await app.unblock({ stream: "^webhooks-out-" });

// Post-incident: unblock everything currently blocked.
await app.unblock({});

unblock clears the blocked flag, resets retry count, drops any stale lease, and settles the now-free stream on its own. The at watermark is not touched — the stream resumes from the next event after the last successful ack, not from the beginning.

The filter form always restricts to blocked = true regardless of what the caller passes — there's no use case for "unblock unblocked streams." Already-unblocked streams and unknown names are silently skipped; the return count reflects only streams that were actually flipped.

Filter patterns are regex-shaped, but stick to the portable grammar — ^ / $ anchors, . (any single character), .* (any run), and literal characters — which matches identically on every store. Richer regex (character classes, alternation, quantifiers, escapes) either matches with full regex semantics or throws ValidationError on adapters that can't express it exactly (SQLite): a filter that drives bulk recovery must fail loudly rather than silently select the wrong streams. See extension-points for the adapter contract.

Contrast with app.reset(input), which is for projection rebuilds. reset accepts the same string[] | StreamFilter shape but sets the watermark back to -1 and replays every event from the start:

Use case	Method
Recovered from a poison message, resume normally	app.unblock([stream]) or app.unblock(filter)
Bulk recovery across a family of streams	app.unblock({ stream: "^proj-" })
Deploy new projection logic, replay all events	app.reset([stream])
Rebuild every blocked stream from zero	app.reset({ blocked: true })

Discovering blocked streams — app.blocked_streams()

For the "show me what's broken" operational query, app.blocked_streams() returns every currently-blocked stream position. Convenience wrapper around store().query_streams(cb, { blocked: true }):

const blocked = await app.blocked_streams();
console.table(
  blocked.map(({ stream, retry, error }) => ({ stream, retry, error }))
);

// Operator investigates, then bulk-unblocks the family:
await app.unblock({ stream: "^webhooks-out-" });

Results paginate by limit (default 100) with an after keyset cursor on the stream name. For richer queries — source filters, unblocked introspection, custom pagination — drop to store().query_streams(...) directly.