Skip to main content
Version: Current

Act

@rotorsoft/act-root

@rotorsoft/act-root / act/src / Act

Class: Act<TSchemaReg, TEvents, TActions, TStateMap, TActor>

Defined in: libs/act/src/act.ts:138

See​

Store

The orchestrator: executes actions, loads state, queries events, and runs the reactions each commit frees (on its own, unless autoSettle: false).

const app = act().withState(Counter).build();
await app.do("increment", { stream: "counter1", actor }, { by: 1 });
const snapshot = await app.load(Counter, "counter1");

Type Parameters​

TSchemaReg​

TSchemaReg extends SchemaRegister<TActions>

SchemaRegister for state

TEvents​

TEvents extends Schemas

Schemas for events

TActions​

TActions extends Schemas

Schemas for actions

TStateMap​

TStateMap extends Record<string, Schema> = Record<string, never>

Map of state names to state schemas

TActor​

TActor extends Actor = Actor

Actor type extending base Actor

Implements​

  • IAct<TEvents, TActions, TActor, TSchemaReg>

Constructors​

Constructor​

new Act<TSchemaReg, TEvents, TActions, TStateMap, TActor>(registry, states?, batch_handlers?, options?, lanes?, patch_fn?): Act<TSchemaReg, TEvents, TActions, TStateMap, TActor>

Defined in: libs/act/src/act.ts:337

Use the act builder rather than constructing directly.

Parameters​

registry​

Registry<TSchemaReg, TEvents, TActions, keyof TStateMap & string>

states?​

Map<string, State<any, any, any>> = ...

batch_handlers?​

Map<string, BatchHandler<any>> = ...

options?​

ActOptions = {}

lanes?​

readonly LaneConfig[] = []

patch_fn?​

PatchFn = bare_patch

Returns​

Act<TSchemaReg, TEvents, TActions, TStateMap, TActor>

Properties​

registry​

readonly registry: Registry<TSchemaReg, TEvents, TActions, keyof TStateMap & string>

Defined in: libs/act/src/act.ts:167

Public registry — kept as-is per the no-prefix-on-public convention.

Accessors​

lanes​

Get Signature​

get lanes(): readonly LaneConfig[]

Defined in: libs/act/src/act.ts:332

Drain lanes declared via .withLane(...). Implicit default not included.

Returns​

readonly LaneConfig[]

Methods​

audit()​

audit(categories?, options?): AsyncIterable<AuditFinding>

Defined in: libs/act/src/act.ts:1385

Operator-driven store audit: walks the store and yields findings per category, each with the remediation it suggests. Never run by the framework. Pass a subset of categories to scope the work; the iterable can be broken early, and findings stream without buffering the report.

Parameters​

categories?​

AuditCategory[]

Subset of categories to run (default: all).

options?​

AuditOptions

Query window + per-category thresholds.

Returns​

AsyncIterable<AuditFinding>

Async iterable of AuditFinding.

Example​

Schema drift since the last scan

for await (const f of app.audit(["schema"], { query: { created_after: lastScan } }))
  await escalate(f);

blocked_streams()​

blocked_streams(options?): Promise<StreamPosition[]>

Defined in: libs/act/src/act.ts:1353

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

Results are ordered by stream name, paginated by limit (default 100). Pass after to fetch the next page (keyset cursor on the stream name). For richer queries — including blocked + source filters, or full unblocked introspection — drop to store().query_streams(...) directly.

Parameters​

options?​
after?​

string

limit?​

number

Returns​

Promise<StreamPosition[]>

Array of StreamPosition for currently-blocked streams.

Example​

Discover and recover

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-" });

close()​

close(targets): Promise<CloseResult>

Defined in: libs/act/src/act.ts:1466

Close the books: guard, archive, truncate, and optionally restart streams. A stream with pending reactions is skipped; an archive failure aborts the whole close while the streams are still guarded. Targets carrying before prune a rolling window instead and leave the stream live (needs .snap(...)). The phases and their failure semantics are in the close-cycle architecture page.

Parameters​

targets​

CloseTarget[]

Per-stream close options (stream, restart?, archive?, before?)

Returns​

Promise<CloseResult>

{ truncated: TruncateResult, skipped: string[] }

Example​

Archive and close

await app.close([
  { stream: "order-123", archive: async () => { await archiveToS3("order-123"); } },
  { stream: "order-456" },
]);

correlate()​

correlate(query?): Promise<{ last_id: number; subscribed: number; }>

Defined in: libs/act/src/act.ts:1079

Scans committed events from the correlation checkpoint, subscribes the target streams their reactions resolve to (static or computed from the event), and marks their work so the next drain can claim it.

Parameters​

query?​

Query = {}

Only limit is used: the most events the pass reads (10 when the query has none; 0 reads everything). The scan always starts at the correlation checkpoint, which every reaction shares, so filters and after are ignored.

Returns​

Promise<{ last_id: number; subscribed: number; }>

The number of newly subscribed streams and the last scanned event id

Example​

Manual correlation

const { subscribed } = await app.correlate({ limit: 100 });

See​

start_correlations for automatic periodic correlation

do()​

do<TKey>(action, target, payload, options?): Promise<Snapshot<TSchemaReg[TKey], TEvents>[]>

Defined in: libs/act/src/act.ts:720

Executes an action: validates the payload, loads the state, checks its preconditions, runs the handler, folds the events it emits, checks the state's invariants (its schema) on the result, and commits them with optimistic concurrency on every stream it appends to. The reactions to those events settle on their own (unless autoSettle: false).

Type Parameters​

TKey​

TKey extends string | number | symbol

Action name from registered actions

Parameters​

action​

TKey

The name of the action to execute

target​

Target<TActor>

Target specification with stream ID and actor context

payload​

Readonly<TActions[TKey]>

Action payload matching the action's schema

options?​

DoOptions<TEvents>

Per-call dispatch options (DoOptions) — reactingTo to thread correlation, correlator to override the framework or orchestrator-level correlator for this call only.

Returns​

Promise<Snapshot<TSchemaReg[TKey], TEvents>[]>

The snapshots of the action's own stream, one per event appended to it; the loaded snapshot alone when the action commits nothing

Throws​

If the payload doesn't match the action schema, or the state the action leaves breaks the state schema

Throws​

If business rules are violated

Throws​

If another process modified the stream

Example​

Basic action execution

const snapshots = await app.do(
  "increment",
  {
    stream: "counter-1",
    actor: { id: "user1", name: "Alice" }
  },
  { by: 5 }
);

console.log(snapshots[0].state.count); // Current count after increment

See​

Implementation of​

IAct.do

drain()​

drain(options?): Promise<Drain<TEvents>>

Defined in: libs/act/src/act.ts:990

Runs one drain cycle: claims streams with marked work (leasing them), fetches their events, runs the matching reactions, and acks or blocks each stream.

A stream is claimable only once correlate() has marked it, static targets included, so a commit no correlate has seen is not drainable. Apps rarely call this: each commit settles its reactions on its own. With autoSettle: false, tests walk the pipeline with correlate() then drain() (see the testing guide).

Parameters​

options?​

DrainOptions = {}

Drain configuration — see DrainOptions for fields (streamLimit, eventLimit, leaseMillis).

Returns​

Promise<Drain<TEvents>>

Drain statistics with fetched, leased, acked, and blocked counts

Example​

Walking the pipeline in a test

const app = act().withState(User).withProjection(Users).build({ autoSettle: false });
await app.do("createUser", target, payload);
await app.correlate();
await app.drain();

See​

  • settle for debounced correlate→drain with lifecycle events
  • correlate for dynamic stream discovery

emit()​

emit<E>(event, args): boolean

Defined in: libs/act/src/act.ts:186

Emit a lifecycle event. The payload type is inferred from the event name via ActLifecycleEvents.

Each listener is contained on its own: a throwing listener is logged, the remaining listeners still run, and the caller (do(), forget(), close(), the drain) never sees the throw, since its durable work has already landed.

Uses rawListeners so once wrappers still de-register themselves, and returns "had listeners" to preserve the EventEmitter.emit contract.

Type Parameters​

E​

E extends keyof ActLifecycleEvents<TSchemaReg, TEvents, TActions>

Parameters​

event​

E

args​

ActLifecycleEvents<TSchemaReg, TEvents, TActions>[E]

Returns​

boolean

forget()​

forget(stream): Promise<{ eventCount: number; }>

Defined in: libs/act/src/act.ts:946

Wipe the sensitive-data payload for every event on the stream — see IAct.forget.

Throws on adapters without Store.forget_pii, invalidates the cache entry for the stream, emits the forgotten lifecycle event with the row count. Idempotent: a second call returns {eventCount: 0} and does NOT re-emit.

Parameters​

stream​

string

Target stream.

Returns​

Promise<{ eventCount: number; }>

{eventCount} — number of events whose PII column was wiped.

Implementation of​

IAct.forget

load()​

Call Signature​

load<TNewState, TNewEvents, TNewActions>(state, stream, callback?, asOf?): Promise<Snapshot<TNewState, TNewEvents>>

Defined in: libs/act/src/act.ts:798

Loads the current state snapshot for a specific stream.

Reconstructs the current state by replaying events from the event store. Uses snapshots when available to optimize loading performance.

Accepts either a State definition object or a state name string. When using a string, the merged state (from partial states registered via .withState()) is resolved by name.

Type Parameters​
TNewState​

TNewState extends Schema

State schema type

TNewEvents​

TNewEvents extends Schemas

Event schemas type

TNewActions​

TNewActions extends Schemas

Action schemas type

Parameters​
state​

State<TNewState, TNewEvents, TNewActions>

The state definition or state name to load

stream​

string

The stream ID (state instance identifier)

callback?​

(snapshot) => void

Optional callback invoked with the loaded snapshot

asOf?​

AsOf

Returns​

Promise<Snapshot<TNewState, TNewEvents>>

The current state snapshot for the stream

Example​

Load by state definition

const snapshot = await app.load(Counter, "counter-1");
console.log(snapshot.state.count);    // Current count
console.log(snapshot.patches);        // Events since last snapshot
See​

Snapshot for snapshot structure

Implementation of​

IAct.load

Call Signature​

load<TKey>(name, stream, callback?, asOf?): Promise<Snapshot<TStateMap[TKey], TEvents>>

Defined in: libs/act/src/act.ts:808

Loads the current state snapshot for a specific stream.

Reconstructs the current state by replaying events from the event store. Uses snapshots when available to optimize loading performance.

Accepts either a State definition object or a state name string. When using a string, the merged state (from partial states registered via .withState()) is resolved by name.

Type Parameters​
TKey​

TKey extends string

Parameters​
name​

TKey

stream​

string

The stream ID (state instance identifier)

callback?​

(snapshot) => void

Optional callback invoked with the loaded snapshot

asOf?​

AsOf

Returns​

Promise<Snapshot<TStateMap[TKey], TEvents>>

The current state snapshot for the stream

Example​

Load by state definition

const snapshot = await app.load(Counter, "counter-1");
console.log(snapshot.state.count);    // Current count
console.log(snapshot.patches);        // Events since last snapshot
See​

Snapshot for snapshot structure

Implementation of​

IAct.load

Call Signature​

load<TNewState, TNewEvents, TNewActions>(state, target, callback?): Promise<Snapshot<TNewState, TNewEvents>>

Defined in: libs/act/src/act.ts:815

Loads the current state snapshot for a specific stream.

Reconstructs the current state by replaying events from the event store. Uses snapshots when available to optimize loading performance.

Accepts either a State definition object or a state name string. When using a string, the merged state (from partial states registered via .withState()) is resolved by name.

Type Parameters​
TNewState​

TNewState extends Schema

State schema type

TNewEvents​

TNewEvents extends Schemas

Event schemas type

TNewActions​

TNewActions extends Schemas

Action schemas type

Parameters​
state​

State<TNewState, TNewEvents, TNewActions>

The state definition or state name to load

target​

LoadTarget<TActor>

callback?​

(snapshot) => void

Optional callback invoked with the loaded snapshot

Returns​

Promise<Snapshot<TNewState, TNewEvents>>

The current state snapshot for the stream

Example​

Load by state definition

const snapshot = await app.load(Counter, "counter-1");
console.log(snapshot.state.count);    // Current count
console.log(snapshot.patches);        // Events since last snapshot
See​

Snapshot for snapshot structure

Implementation of​

IAct.load

Call Signature​

load<TKey>(name, target, callback?): Promise<Snapshot<TStateMap[TKey], TEvents>>

Defined in: libs/act/src/act.ts:824

Loads the current state snapshot for a specific stream.

Reconstructs the current state by replaying events from the event store. Uses snapshots when available to optimize loading performance.

Accepts either a State definition object or a state name string. When using a string, the merged state (from partial states registered via .withState()) is resolved by name.

Type Parameters​
TKey​

TKey extends string

Parameters​
name​

TKey

target​

LoadTarget<TActor>

callback?​

(snapshot) => void

Optional callback invoked with the loaded snapshot

Returns​

Promise<Snapshot<TStateMap[TKey], TEvents>>

The current state snapshot for the stream

Example​

Load by state definition

const snapshot = await app.load(Counter, "counter-1");
console.log(snapshot.state.count);    // Current count
console.log(snapshot.patches);        // Events since last snapshot
See​

Snapshot for snapshot structure

Implementation of​

IAct.load

off()​

Call Signature​

off<E>(event, listener): this

Defined in: libs/act/src/act.ts:276

Remove a previously registered lifecycle listener, with the state it was registered for when it has one.

Type Parameters​
E​

E extends keyof ActLifecycleEvents<TSchemaReg, TEvents, TActions>

Parameters​
event​

E

listener​

(args) => void

Returns​

this

Call Signature​

off<TNewState, TNewEvents, TNewActions>(event, state, listener): this

Defined in: libs/act/src/act.ts:282

Remove a previously registered lifecycle listener, with the state it was registered for when it has one.

Type Parameters​
TNewState​

TNewState extends Schema

TNewEvents​

TNewEvents extends Schemas

TNewActions​

TNewActions extends Schemas

Parameters​
event​

"committed"

state​

State<TNewState, TNewEvents, TNewActions>

listener​

(snapshots) => void

Returns​

this

Call Signature​

off<TKey>(event, state, listener): this

Defined in: libs/act/src/act.ts:291

Remove a previously registered lifecycle listener, with the state it was registered for when it has one.

Type Parameters​
TKey​

TKey extends string

Parameters​
event​

"committed"

state​

TKey

listener​

(snapshots) => void

Returns​

this

on()​

Call Signature​

on<E>(event, listener): this

Defined in: libs/act/src/act.ts:238

Register a listener for a lifecycle event. The listener receives the event-specific payload.

committed also takes a state, as a State or its name: the listener then hears only commits to that state's streams, whichever action appended them, with its snapshots typed from that state.

Type Parameters​
E​

E extends keyof ActLifecycleEvents<TSchemaReg, TEvents, TActions>

Parameters​
event​

E

listener​

(args) => void

Returns​

this

Example​
app.on("committed", Counter, (snapshots) => {
  console.log(snapshots.at(-1)?.state.count); // typed: number
});

Call Signature​

on<TNewState, TNewEvents, TNewActions>(event, state, listener): this

Defined in: libs/act/src/act.ts:244

Register a listener for a lifecycle event. The listener receives the event-specific payload.

committed also takes a state, as a State or its name: the listener then hears only commits to that state's streams, whichever action appended them, with its snapshots typed from that state.

Type Parameters​
TNewState​

TNewState extends Schema

TNewEvents​

TNewEvents extends Schemas

TNewActions​

TNewActions extends Schemas

Parameters​
event​

"committed"

state​

State<TNewState, TNewEvents, TNewActions>

listener​

(snapshots) => void

Returns​

this

Example​
app.on("committed", Counter, (snapshots) => {
  console.log(snapshots.at(-1)?.state.count); // typed: number
});

Call Signature​

on<TKey>(event, state, listener): this

Defined in: libs/act/src/act.ts:253

Register a listener for a lifecycle event. The listener receives the event-specific payload.

committed also takes a state, as a State or its name: the listener then hears only commits to that state's streams, whichever action appended them, with its snapshots typed from that state.

Type Parameters​
TKey​

TKey extends string

Parameters​
event​

"committed"

state​

TKey

listener​

(snapshots) => void

Returns​

this

Example​
app.on("committed", Counter, (snapshots) => {
  console.log(snapshots.at(-1)?.state.count); // typed: number
});

prioritize()​

prioritize(filter, priority): Promise<number>

Defined in: libs/act/src/act.ts:1419

Sets the scheduling priority of every stream matching filter, as-is: unlike build-time priorities (the highest across reactions wins), this overrides them. Useful to let a long replay jump ahead of other lagging streams, or to yield slots back. An empty filter ({}) updates every registered stream.

Parameters​

filter​

StreamFilter

Selection criteria (regex by default, like query).

priority​

number

New priority value. Set as-is — no clamp.

Returns​

Promise<number>

Count of streams whose priority changed.

Example​

Boost a specific projection mid-replay

await app.prioritize({ stream: "^proj-orders$", stream_exact: false }, 10);

See​

Store.prioritize for the underlying primitive

query()​

query(query, callback?): Promise<{ count: number; first?: Committed<TEvents, keyof TEvents>; last?: Committed<TEvents, keyof TEvents>; }>

Defined in: libs/act/src/act.ts:883

Queries the event store for events matching a filter.

Use this for analyzing event streams, generating reports, or debugging. The callback is invoked for each matching event, and the method returns summary information (first event, last event, total count).

For small result sets, consider using query_array instead.

Parameters​

query​

Query

Filter criteria; see Query for the fields

callback?​

(event) => void

Optional callback invoked for each matching event

Returns​

Promise<{ count: number; first?: Committed<TEvents, keyof TEvents>; last?: Committed<TEvents, keyof TEvents>; }>

Object with first event, last event, and total count

Example​

Query all events for a stream

const { first, last, count } = await app.query(
  { stream: "counter-1" },
  (event) => console.log(event.name, event.data)
);
console.log(`Found ${count} events from ${first?.id} to ${last?.id}`);

See​

query_array for loading events into memory

Implementation of​

IAct.query

query_array()​

query_array(query): Promise<Committed<TEvents, keyof TEvents>[]>

Defined in: libs/act/src/act.ts:922

Queries the event store and returns all matching events in memory.

Use with caution - this loads all results into memory. For large result sets, use query with a callback instead to process events incrementally.

Parameters​

query​

Query

The query filter (same as query)

Returns​

Promise<Committed<TEvents, keyof TEvents>[]>

Array of all matching events

Example​

Load all events for a stream

const events = await app.query_array({ stream: "counter-1" });
console.log(`Loaded ${events.length} events`);
events.forEach(event => console.log(event.name, event.data));

See​

query for large result sets

Implementation of​

IAct.query_array

reset()​

reset(input): Promise<number>

Defined in: libs/act/src/act.ts:1200

Reset reaction stream watermarks and settle them, replaying their events through their reactions.

Use this to replay events through projections (or other reaction targets) after changing handler logic. It rewinds the same watermarks as store().reset(streams), and also wakes the drain and schedules a settle, which loops correlate→drain until the replay catches up; store().reset(...) alone leaves a settled app idle. Call app.settle(options) right after to give that pass batch options, or, with autoSettle: false, correlate() / drain() to drive the replay by hand.

Parameters​

input​

string[] | StreamFilter

Reaction target streams (e.g., projection names) to reset, or a StreamFilter for bulk operations

Returns​

Promise<number>

Count of streams that were actually reset

Example​

Rebuild a projection (production)

await app.reset(["my-projection"]);
app.settle({ eventLimit: 1000 }); // optional: bigger batches; emits "settled"

See​

  • Store.reset for the underlying store primitive
  • settle for the debounced full-catch-up loop

restore()​

restore(source, opts?, sink?): Promise<ScanResult>

Defined in: libs/act/src/act.ts:1290

Atomically wipe the store and rebuild it from an event source. The framework owns iteration, validation, drop_snapshots filtering, on_progress, and the per-call old → new causation remap; the adapter's Store.restore driver supplies the transaction lifecycle and per-event insert.

Safe in a live process: restoring into this Act's own store waits for its correlate passes in flight, replaces the log, then restarts correlation and settles. A correlate pass that starts meanwhile reads nothing. Other processes on the same store keep their correlation state and must be restarted.

Throws if the adapter has no restore capability. Throws on the first invalid event (negative version, malformed created) with the running index in the message; atomic transaction rollback in the adapter means a failing restore leaves the store byte-for-byte unchanged.

Parameters​

source​

EventSource

Any EventSource: a CsvFile backup or another store. Read in batches rather than buffered, so multi-million-event backups don't OOM. Each event's original id is used as a causation lookup key but never written through — adapters renumber densely.

opts?​

ScanOptions = {}

ScanOptions. drop_snapshots skips __snapshot__ events (counted in the result); on_progress fires once per event.

sink?​

EventSink

Where the events land. Defaults to this Act's store.

Returns​

Promise<ScanResult>

The ScanResult: what was kept, migrated and dropped, and how long it took.

Example​

Restore a CSV backup into the connected store

const result = await app.restore(new CsvFile({ path: "./backup.csv" }));
console.log(`Restored ${result.kept} events in ${result.duration_ms}ms`);
await cache().clear();   // operator's responsibility

See​

Store.restore for the underlying driver-pattern primitive.

settle()​

settle(options?): void

Defined in: libs/act/src/act.ts:1505

Debounced, non-blocking correlate→drain cycle.

A local change that frees reaction work (a commit, reset(), unblock(), restore()) schedules this on its own unless the app sets autoSettle: false. Call it at startup, to run what was pending before a restart, or with options for the next pass; the scheduling rule is in the event-sourcing docs (§ settle). Runs correlate→drain passes until one makes no progress (see SettleOptions.maxPasses), then emits the "settled" lifecycle event, so one call fully catches up paginated streams without forcing callers to loop.

Parameters​

options?​

SettleOptions

Settle configuration — see SettleOptions for fields: debounceMs (default 10), correlate.limit (default 100, or the poll's limit while start_correlations runs; after is always the checkpoint), maxPasses (default Infinity; caps this settle's passes, not the passes a reaction chain's own commits schedule), streamLimit (default 10), eventLimit (default 10), leaseMillis (default 10000).

Returns​

void

Example​

Startup

app.settle(); // run what was pending before the restart

app.on("settled", (drain) => {
  // notify SSE clients, invalidate caches, etc.
});

See​

  • drain for single synchronous drain cycles
  • correlate for manual correlation

shutdown()​

shutdown(options?): Promise<void>

Defined in: libs/act/src/act.ts:591

Per-instance teardown: stops scheduling, removes the notify subscription, gives the cycles in flight up to graceMs to reach their ack (an abandoned handler holds its lease until it expires), then hands back the correlation lease and removes listeners. The budget is a ceiling, not a delay.

Idempotent: repeated calls return the same promise, with the first call's graceMs. Registered with the global dispose() registry, so dispose()() covers it; call it directly for prompt cleanup.

Parameters​

options?​

ShutdownOptions

See ShutdownOptions. Defaults the grace budget to the largest lane leaseMillis (capped at 30s).

Returns​

Promise<void>

start_correlations()​

start_correlations(query?, frequency?): boolean

Defined in: libs/act/src/act.ts:1135

Starts a background poll that settles on a timer: each tick looks for new events, subscribes the streams their reactions target, and drains them, along with any stream whose lease a dead worker let expire.

Use it when this process can miss commits — another process writes to a store without notify, or a notification is lost. It is the safety net under notify: a missed wakeup costs at most one frequency of latency.

Only one poll runs per Act instance. On a writer-only instance (drain: false) each tick is a no-op, like settle().

Parameters​

query?​

Query = {}

Only limit is used: the most events each correlate pass reads. With no limit, or 0, the passes use the background default (100): a poll never reads the whole log in one tick. The scan always starts at the correlation checkpoint, so after is ignored.

frequency?​

number = 10_000

Milliseconds between ticks (default: 10000)

Returns​

boolean

true if the poll started, false if one is already running

Example​

app.start_correlations({ limit: 100 }, 5000);
// later
app.stop_correlations();

See​

stop_correlations()​

stop_correlations(): void

Defined in: libs/act/src/act.ts:1154

Stops the poll started by start_correlations. Idempotent, and called by shutdown().

Returns​

void

See​

start_correlations

stop_settling()​

stop_settling(): void

Defined in: libs/act/src/act.ts:1171

Drops the settle pass scheduled but not yet started. A cycle already running finishes; later changes schedule new passes.

Returns​

void

See​

settle

unblock()​

unblock(input): Promise<number>

Defined in: libs/act/src/act.ts:1240

Clear the blocked flag on streams without replaying their history.

Use this to recover from a poison message after fixing the underlying issue — the stream resumes from the next event after the last successful ack, not from the beginning. Compare with reset, which rebuilds from event 0 (suitable for projection rebuilds, wrong for "I fixed the bug, please retry").

Wraps store().unblock(streams) and settles the now-free streams on its own. store().unblock(...) alone leaves them waiting for the next commit.

Parameters​

input​

string[] | StreamFilter

Stream names to unblock, or a StreamFilter for bulk recovery

Returns​

Promise<number>

Count of streams that were actually flipped (were blocked)

Example​

Recover from a 4xx webhook after fixing the bug

await app.unblock(["webhooks-out-customer-42"]);
// The stream resumes from the next event, not from zero.

See​

  • Store.unblock for the underlying store primitive
  • reset for the rebuild-from-zero alternative