Skip to main content
Version: Current

Store

@rotorsoft/act-root

@rotorsoft/act-root / act/src / Store

Interface: Store

Defined in: libs/act/src/types/ports.ts:544

Interface for event store implementations.

The Store interface defines the contract for persistence adapters in Act. Implementations must provide event storage, querying, and distributed processing capabilities through leasing and watermark tracking.

Act includes two built-in implementations:

  • InMemoryStore: For development and testing
  • PostgresStore: For production use with PostgreSQL

Custom stores can be implemented for other databases or event log systems.

Example​

Using a custom store

import { store } from "@rotorsoft/act";
import { PostgresStore } from "@rotorsoft/act-pg";

// Replace the default in-memory store
store(new PostgresStore({
  host: "localhost",
  port: 5432,
  database: "myapp",
  user: "postgres",
  password: "secret"
}));

const app = act()
  .withState(Counter)
  .build();

See​

  • InMemoryStore for the default implementation
  • PostgresStore for the PostgreSQL implementation

Extends​

Properties​

ack​

ack: (leases) => Promise<Lease[]>

Defined in: libs/act/src/types/ports.ts:839

Finalizes leased streams atomically: acknowledges the ones that processed successfully and persists defer schedules for the ones that chose to be re-visited later — one call, one transaction.

Every entry advances the watermark to at (the last event handled this cycle) and releases the lease. An entry without Lease.due is a plain ack: it also resets retry to -1 and clears any schedule. An entry with due additionally defers the remainder: the stream's deferred_at is set to due (ms since epoch) and retry to the entry's own Lease.retry. Advance and defer are independent legs, so a partial-progress backoff/defer advances past the events it handled — that prefix never re-runs — while the failing tail waits for deferred_at (a hold passes at = the current watermark, making the advance a no-op). A caller deferring a deliberate re-visit passes retry: -1 (the same semantics as defer — a defer is not a failure); a caller pacing a retry backoff passes the climbing counter so the retry budget keeps accruing across windows until it blocks. All-or-nothing is the contract: a failure must leave every watermark and every schedule untouched, so a drain cycle's outcomes can never land partially (an acked close request must not survive a lost defer, and vice versa).

Parameters​

leases​

Lease[]

Leases to finalize; due-carrying entries defer, the rest ack. At most one entry per stream. Each lease is for one stream, and an adapter that applies the batch in one statement cannot apply two entries to the same row (UPDATE … FROM may touch a target row once, so Postgres keeps an arbitrary one). Callers de-duplicate before calling, as the drain does; adapters may assume uniqueness.

Returns​

Promise<Lease[]>

The acknowledged leases (deferred entries are not returned)

Example​

const leased = await store().claim(5, 5, randomUUID(), 10000);
// Ack most streams at ID 150; hold order-42 until half past
await store().ack(leased.map(l =>
  l.stream === "order-42"
    ? { ...l, due: Date.now() + 30 * 60_000 }
    : { ...l, at: 150 }
));

See​

  • claim for acquiring leases
  • defer for the standalone (operator-facing) schedule write

block​

block: (leases) => Promise<BlockedLease[]>

Defined in: libs/act/src/types/ports.ts:874

Blocks streams after persistent processing failures.

Blocked streams won't be returned by claim until manually unblocked. This prevents poison messages from repeatedly failing and consuming resources.

Streams are typically blocked when:

  • Max retries reached
  • blockOnError option is true
  • Handler throws an error

Parameters​

leases​

BlockedLease[]

Leases to block with error messages. At most one entry per stream, for the same reason as ack; adapters may assume uniqueness.

Returns​

Promise<BlockedLease[]>

Blocked leases

Example​

try {
  await processEvents(lease);
  await store().ack([lease]);
} catch (error) {
  if (lease.retry >= 3) {
    await store().block([{
      ...lease,
      error: error.message
    }]);
  }
}

See​

claim for lease management

claim​

claim: (lagging, leading, by, millis, lane?) => Promise<Lease[]>

Defined in: libs/act/src/types/ports.ts:704

Atomically discovers and leases streams for reaction processing.

Atomically discovers a stream and acquires a lease in one round-trip, eliminating the race that exists when discovery and locking are separate calls (a competing worker can grab the stream between the two).

PostgresStore uses FOR UPDATE SKIP LOCKED for zero-contention competing consumer semantics — workers never block each other, each grabbing different streams atomically. InMemoryStore fuses its poll+lease logic equivalently.

Claim reads only the subscription rows; it never matches a source against the event log. Correlate does that when it decides which targets to mark, and the drain uses the same rule when it fetches the claimed stream's events. A literal source (no anchor — the common case, and every autoclose/dynamic-resolver source) matches by string equality: the fast, index-friendly path, and exact so "s1" never matches "s12". A pattern source (starting with ^ or ending with $, e.g. a static ^(A|B)$ reaction) is compiled as a regex and matched against candidate stream names. Adapters that cannot faithfully run an arbitrary regex (SQLite) reject a non-portable pattern at subscribe time rather than silently never fetching its events.

Every granted lease counts against the stream's retry budget: claim increments the stream's retry counter and only ack resets it, so a timed-out lease reclaimed by any worker marches the stream toward blockOnError exactly like a handler failure.

Used by Act.drain() as the primary stream acquisition method.

Parameters​

lagging​

number

Max streams from the lagging frontier (ascending watermark)

leading​

number

Max streams from the leading frontier (descending watermark)

by​

string

Unique lease holder identifier (UUID)

millis​

number

How long the lease stands, in milliseconds. It bounds the expiry only — the holder (by) is recorded whatever the value, so a zero-length lease is granted, expires immediately, and its ack still lands. A store that skipped recording the holder would hand out a lease whose every ack is dropped, parking the watermark and redelivering every event on every drain.

lane?​

string

Optional lane filter

Returns​

Promise<Lease[]>

Array of successfully leased streams with metadata

Example​

const leased = await store().claim(5, 5, randomUUID(), 10000);
leased.forEach(({ stream, at, lagging }) => {
  console.log(`Leased ${stream} at ${at} (lagging: ${lagging})`);
});

See​

  • subscribe for registering new streams (used by correlate)
  • ack for acknowledging completion
  • block for blocking failed streams

commit​

commit: <E>(appends, meta) => Promise<Committed<E, keyof E>[]>

Defined in: libs/act/src/types/ports.ts:610

Appends events to one or more streams atomically: every append lands, or none does.

Each append names its stream, its events and, optionally, the version the stream must be at. The store must:

  • check every expectedVersion against its own stream before appending anything (-1 expects an empty stream, undefined skips the check), and throw ConcurrencyError naming the first stream that moved;
  • refuse a stream named twice and an append with no events (a plain Error), so a commit is never a version check alone;
  • assign global ids in append order, then event order, and make the events visible in that order;
  • continue each stream's versions from its head, and attach meta to every event.

No appends at all commits nothing and returns [].

Type Parameters​

E​

E extends Schemas

Event schemas

Parameters​

appends​

readonly Append<E>[]

One Append per stream

meta​

EventMeta

Event metadata (correlation, causation), shared by every event

Returns​

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

The committed events of every append, in append order

Throws​

If a stream's version doesn't match its expectedVersion

Example​

const events = await store().commit(
  [{ stream: "user-123", msgs: [{ name: "UserCreated", data: { email: "user@example.com" } }], expectedVersion: -1 }],
  { correlation: "req-456", causation: { action: {...} } }
);

defer​

defer: (input, deferred_at) => Promise<number>

Defined in: libs/act/src/types/ports.ts:913

Operator verb: bulk-pause streams until a future time without advancing their watermark. The drain itself never calls this — a reaction's defer outcome is persisted atomically by ack via due-marked leases; this standalone verb exists for operator-driven scheduling ("hold every webhook-.* stream until the maintenance window ends"), completing the recovery family: reset / unblock / prioritize / defer.

Sets deferred_at on each matched stream. claim skips any stream whose deferred_at is still in the future, so a paused stream is not re-claimed (and retry is never bumped) until the due-time passes, at which point the same pending events are re-delivered. Unlike in-process backoff, this is durable, shared store state — every competing worker honors the skip.

claim treats a deferred stream as due once the claiming caller's clock reaches deferred_at, not a clock of the store's own (such as a database NOW()): the orchestrator's defer timer wakes on the caller's clock, and the claim it triggers must agree with it.

The schedule is cleared whenever the watermark moves or the stream is recovered: ack, block, reset, and unblock all reset deferred_at. Re-deferring simply overwrites it.

Accepts an explicit list of stream names or a StreamFilter (regex by default), the same shape as reset/unblock.

Parameters​

input​

string[] | StreamFilter

Stream names or a StreamFilter selecting streams

deferred_at​

number

Wall-clock time (ms since epoch) to revisit the streams

Returns​

Promise<number>

Count of streams whose deferred_at was set

Example​

// Pause every webhook delivery stream during downstream maintenance
await store().defer({ stream: "^webhook-" }, Date.now() + 30 * 60_000);

dispose​

dispose: Disposer

Defined in: libs/act/src/types/ports.ts:27

Inherited from​

Disposable.dispose

drop​

drop: () => Promise<void>

Defined in: libs/act/src/types/ports.ts:574

Drops all data from the store.

Dangerous operation that deletes all events and state. Use with extreme caution, primarily for testing or development environments.

Returns​

Promise<void>

Example​

// Clean up after tests
afterAll(async () => {
  await store().drop();
});

forget_pii?​

optional forget_pii?: (stream) => Promise<number>

Defined in: libs/act/src/types/ports.ts:1352

Wipe the sensitive-data payload for every event on the stream — the physical-erasure side of the sensitive-data epic. Sets the adapter's PII column (or equivalent) to NULL for the stream's events; events.data and the rest of the row are never touched.

Returns the count of rows wiped. Idempotent: a second call on an already-wiped stream returns 0 without error.

Capability-gated via pii_isolation in @rotorsoft/act-tck's StoreCapabilities. Adapters that can't UPDATE rows (Kafka, append-only object-storage logs) declare pii_isolation: false and omit this method; app.forget(stream) throws on those adapters at orchestrator level.

Append-only invariant on events.data is preserved — only the separately-isolated PII column is mutated. Disk reclamation is adapter-dependent (PG autovacuum reclaims lazily; SQLite needs PRAGMA incremental_vacuum or VACUUM). For strict-deletion jurisdictions the production checklist documents the operator step.

Encryption-at-rest is the operator's DB-layer concern (pgcrypto, RDS TDE, Cloud SQL TDE, SQLite SEE) — not an application-level port. The framework's job is isolation + erasure.

Parameters​

stream​

string

Target stream

Returns​

Promise<number>

Count of events whose PII column was set to NULL

notify?​

optional notify?: (handler) => NotifyDisposer | Promise<NotifyDisposer>

Defined in: libs/act/src/types/ports.ts:1319

Optional cross-process commit notifications.

When implemented, the Act orchestrator subscribes once at build time and routes notifications to wake settle() automatically — so a remote worker's commit triggers reactions on this process without waiting for the debounce/poll cycle. Subscribers also receive each notification on the "notified" lifecycle event for fan-out (SSE pushes, audit logs, dashboards).

Self-filtering contract: implementations must skip their own commits. The handler fires only for commits originating from a different process writing to the same backing store. This keeps the local fast path (do() already arms drain) free of duplicate wake-ups and gives "notified" a clean cross-process semantic. Enforced by the notify capability cases in @rotorsoft/act-tck, along with one-notification-per-commit batch delivery.

Hint, not a contract: the orchestrator never depends on notify for correctness. If absent, dropped, or the store omits it, the existing debounce/poll path still drains correctly — notify only lowers cross-process p99 reaction latency.

Adapter status (Act 0.x):

  • PostgresStore: implemented via LISTEN/NOTIFY on the act_commit channel
  • InMemoryStore: not implemented (single-process — no remote writers exist)
  • SqliteStore: not implemented (single-node by design)

Parameters​

handler​

(notification) => void

Callback invoked once per remote commit

Returns​

NotifyDisposer | Promise<NotifyDisposer>

Disposer releasing the underlying listener

prioritize​

prioritize: (filter, priority) => Promise<number>

Defined in: libs/act/src/types/ports.ts:1043

Bulk-update the scheduling priority of streams matching a filter.

Used by Act.prioritize for operator runtime control over lagging-frontier claim() ordering. Unlike subscribe, which keeps the per-stream priority at the max() of all registered reactions targeting that stream, prioritize sets the priority directly to priority for matching rows — letting operators override the build-time scheduling policy.

Filter semantics mirror query_streams: stream/source are regex by default, exact with the *_exact flags. blocked restricts to blocked or unblocked rows. Omitted fields don't filter. An empty filter ({}) updates every registered stream — useful for "reset all priorities to N" but a footgun otherwise.

Parameters​

filter​

StreamFilter

PrioritizeFilter selecting which streams to update. Required (use {} to target all).

priority​

number

New priority value. Set as-is — no max(), no clamp.

Returns​

Promise<number>

Count of streams whose priority was changed.

Example​

Boost a specific replay

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

See​

  • Act.prioritize for the orchestrator-level wrapper
  • claim for how priority biases stream scheduling

query​

query: <E>(callback, query?) => Promise<number>

Defined in: libs/act/src/types/ports.ts:645

Queries events from the store with optional filtering.

Calls the callback for each matching event. The callback approach allows processing large result sets without loading everything into memory.

The adapter returns what it stored. A Date is stored as its ISO string and comes back as that string; the framework turns the fields an event's schema declares z.date() back into Dates, so an ISO-shaped z.string() stays a string. See the Store contract in docs/docs/architecture/extension-points.md.

Type Parameters​

E​

E extends Schemas

Event schemas

Parameters​

callback​

(event) => void

Function invoked for each matching event

query?​

Readonly<{ after?: number; backward?: boolean; before?: number; correlation?: string; created_after?: Date; created_before?: Date; limit?: number; names?: string[]; stream?: string; stream_exact?: boolean; with_pii?: boolean; with_snaps?: boolean; }>

Optional filter criteria; see Query for the fields.

Returns​

Promise<number>

Total number of events processed

Example​

Query all events for a stream

let count = 0;
await store().query(
  (event) => {
    console.log(event.name, event.data);
    count++;
  },
  { stream: "user-123" }
);
console.log(`Found ${count} events`);

Overrides​

EventSource.query

query_stats​

query_stats: <E>(input, options?) => Promise<Map<string, StreamStats<E>>>

Defined in: libs/act/src/types/ports.ts:1281

Per-stream aggregated stats — single round trip per adapter.

Returns the latest event (head) plus opt-in extras (tail, count, names) for each stream selected by input. Streams with no qualifying events are absent from the result map.

Cost model. With no opt-in flags, the call uses an index-backed head lookup per stream — O(K) where K is the number of matched streams. tail alone stays in the cheap tier. Setting count and/or names triggers a full event scan over the matched streams (O(N) where N is total events); both stats share that scan, so requesting one or both is the same cost.

input. Either an explicit string[] of stream names, or a narrow event-stream selector { stream?, stream_exact? } for pattern-based or exact-name matching. Subscription-level filters (source, blocked) are intentionally not accepted here — they describe subscriptions, not events, and conflating the two would silently exclude unsubscribed event streams. For "stats for all blocked subscriptions" compose explicitly: query_streams({blocked: true}) → collect names → query_stats(names).

head vs tail naming. Follows the git-log convention: head is the latest event (highest id), tail is the earliest (lowest id). This is the opposite of the Unix head/tail commands.

Framework markers. Snapshots (__snapshot__) and tombstones (__tombstone__) are real events and are included by default — intentional, so schema-evolution tooling can count them. To exclude them, pass them in options.exclude (typed against EventName) so typos are compile-time errors.

Snapshot counts come from names. When names: true and snapshots are not in exclude, result.names["__snapshot__"] is the snapshot count for that stream — no separate field needed. Validates snapshot policy at scale: names["__snapshot__"] / count should match the configured snap predicate's expected ratio.

Time travel. options.before narrows to events with id < before, answering "what did this stream look like at event N?" without special call shape.

Ordering + pagination. The returned Map is ordered by stream name. options.after (exclusive cursor) + options.limit keyset- paginate over that order — pass the last key of one page as the after of the next. limit defaults to unbounded, so callers that omit it keep receiving every matching stream in one call.

Type Parameters​

E​

E extends Schemas

Event schemas. Narrow at the call site to type-check exclude against your event names (typos fail at compile time).

Parameters​

input​

string[] | Pick<StreamFilter, "stream" | "stream_exact">

Stream names or a filter selecting the streams to stat.

options?​

QueryStatsOptions<E>

Opt-in stat fields, event-name exclusions, and time-travel cutoff. See QueryStatsOptions.

Returns​

Promise<Map<string, StreamStats<E>>>

Map keyed by stream name. Streams with no qualifying events (after exclude and before are applied) are absent.

Example​

Cheap heads — close-cycle pattern (one round trip, no scan)

const stats = await store().query_stats(streams, {
  exclude: [TOMBSTONE_EVENT],
});
for (const [stream, { head }] of stats) {
  // head.id, head.version, head.name
}

See​

query_streams​

query_streams: (callback, query?) => Promise<QueryStreamsResult>

Defined in: libs/act/src/types/ports.ts:1204

Streams registered subscription positions to a callback, plus the highest event id in the store.

Read-only introspection for operational dashboards (Store / Subscriptions tab, projection lag, blocked subscriptions). Avoids forcing apps to open a second connection and run raw SQL against adapter-specific schemas.

Mirrors the Store.query callback pattern: the callback is invoked once per matching position, allowing large result sets to be processed without buffering. Results are ordered by stream name; use query.after (last seen stream name) for keyset pagination on big tables (dynamic reactions can produce one subscription per aggregate).

Parameters​

callback​

(position) => void

Invoked once per matching StreamPosition.

query?​

QueryStreams

Optional QueryStreams filter (default limit: 100).

Returns​

Promise<QueryStreamsResult>

maxEventId and the count of positions emitted.

Example​

List blocked streams with their lag

const { maxEventId } = await store().query_streams(
  (s) => console.log(`${s.stream}: lag=${maxEventId - s.at} ${s.error}`),
  { blocked: true, limit: 50 }
);

reset​

reset: (input) => Promise<number>

Defined in: libs/act/src/types/ports.ts:953

Resets watermarks for the given streams to -1, making them eligible for replay from the beginning. Also clears retry, blocked, error, and lease state so the streams can be claimed immediately.

Prefer Act.reset() over calling this directly. This primitive only resets the store; it does not raise the orchestrator's internal drain-armed flag, so a settled Act instance will short-circuit and skip the replay. Act.reset() wraps this and arms the flag.

Accepts either an explicit list of stream names or a StreamFilter for bulk operations (e.g., "rebuild every blocked stream"). The filter form is the same shape used by prioritize and query_streams. An empty filter ({}) matches every registered stream — typically a footgun for reset; prefer narrower filters like { blocked: true }.

Parameters​

input​

string[] | StreamFilter

Stream names or a StreamFilter

Returns​

Promise<number>

Count of streams that were actually reset

Example​

// By name
await app.reset(["my-projection"]);

// By filter — rebuild every blocked stream in a projection family
await app.reset({ stream: "^proj-orders-", blocked: true });

// Low-level (does NOT trigger replay on settled apps)
await store().reset(["my-projection"]);

See​

Act.reset for the high-level rebuild API that wraps this primitive and schedules the replay

restore?​

optional restore?: (driver) => Promise<void>

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

Atomically wipe the store and commit a fresh sequence of events inside a single transaction.

Capability-gated. Adapters may or may not implement restore. Third-party stores that can't atomically wipe-and-rebuild in one transaction can omit it.

Driver pattern. The adapter is a thin transactional wrapper: open the transaction (PG BEGIN, SQLite BEGIN IMMEDIATE, an in-process snapshot for InMemoryStore), truncate the events + streams/subscriptions tables, hand the orchestrator a per-event insert callback by invoking driver(callback), then commit or roll back. Any throw inside driver rolls back the transaction — the store ends byte-for-byte unchanged from the pre-call state.

The framework's scan loop (in internal/event-sourcing.ts) is what calls callback repeatedly: it iterates the source, validates each event, applies drop_snapshots, fires on_progress, rewrites meta.causation.event.id through the per-call old → new map, and counts kept/dropped. Adapters never see that logic — their job is the transaction lifecycle plus the adapter-specific callback body.

Lossless created. The callback receives the event's original timestamp; adapters write it through verbatim. This is the property that makes restore a viable backup/migration primitive — distinct from commit, which always stamps now().

Renumbered id. Adapters reseed ids densely (1..N on PG/SQLite, 0..N-1 on InMemory). The source's original ids are used by the orchestrator as causation lookup keys but never written through.

No subscription preservation. Both the events and the streams/subscriptions tables are wiped. Reactions re-subscribe via the orchestrator's normal correlate() path on the next settle cycle.

Correlate checkpoints reset. Every correlate checkpoint (the shared one and each keyed correlator's) is an event id, so restore sets them back to -1 in the same transaction; left at their old values they would sit above every renumbered event and correlate would never resolve the restored events' reactions. Correlator leases are kept. The Act that ran the restore restarts its own correlation; any other process running against the same store must be restarted, as for any offline restore.

Cache. Restore does not touch the Cache port — callers must cache().clear() after restore to avoid serving stale snapshots. Documented; not enforced.

Parameters​

driver​

(callback) => Promise<void>

Orchestrator-supplied iteration callback. The adapter calls driver(callback) exactly once, from inside its transaction. The callback argument is the adapter's per-event insert hook — it receives the event with meta.causation already rewritten to the new id space and returns the new id the adapter assigned. The driver is purely transactional from the adapter's perspective — kept/dropped counts and timing live in Act.restore.

Returns​

Promise<void>

See​

  • Act.restore for the public entry point.
  • truncate for the single-stream snapshot/tombstone primitive (different operation — restore wipes the whole store)

seed​

seed: () => Promise<void>

Defined in: libs/act/src/types/ports.ts:559

Initializes or resets the store.

Used primarily for testing to ensure a clean state between tests. For production stores, this might create necessary tables or indexes.

Returns​

Promise<void>

Example​

// Reset store between tests
beforeEach(async () => {
  await store().seed();
});

subscribe​

subscribe: (streams, correlated_at?, correlator?) => Promise<SubscribeResult>

Defined in: libs/act/src/types/ports.ts:746

Registers streams for event processing.

Upserts stream entries so they become visible to claim. Used by correlate() to register dynamically discovered reaction target streams.

Also returns the current maximum watermark across all subscribed streams, used internally for correlation checkpoint initialization on cold start.

Parameters​

streams​

SubscribeInput[]

Streams to register with optional source hint — a literal stream name (matched by equality when the drain fetches the stream's events) or a regex pattern (compiled and matched against stream names); non-portable patterns are rejected here on SQLite.

At most one entry per stream. A subscription row is keyed by stream and its source is set once at creation, so a second entry for the same stream could only ever contribute its priority — and an adapter that merges the batch in one statement cannot do even that (UPDATE ... FROM may touch a target row only once, so Postgres picks an arbitrary source row). Callers de-duplicate before calling; adapters may assume uniqueness.

correlated_at?​

number

Advance the correlate checkpoint to this event id — correlate's own watermark, in the same id space as a subscription's at.

correlate is the only component that knows how far it has read, and it already calls subscribe with the targets a scan discovered, so the checkpoint is written by its own producer in its own call. Nothing else may set it: a checkpoint derived from processing progress runs ahead of the read cursor and silently skips discovery.

Persists only when greater than the stored value. A lower or equal value is ignored rather than written, so a worker whose cursor lags cannot rewind it and re-sending the same value is a no-op. Omitted leaves it untouched.

correlator?​

Identifies the calling correlator, and asks for the correlation lease.

Correlation is duplicated work without it: every worker keeps its own in-memory scan position, so N workers wake on the same commit and each reads the whole range and writes the same marks — measured at exactly N reads and N mark-writes per committed event. The marks are idempotent, so this is waste rather than error, but the writes land on the rows claim locks and churn the partial index the design depends on.

It rides subscribe rather than a verb of its own because correlate already calls this every pass: the lease is a property of the checkpoint write that already happens.

  • key scopes the lease to interchangeable correlators. Workers running the same application are; a worker deployed with a subset of the reactions is not, and must scan for itself or its targets are never marked. Callers derive it from the registry; stores treat it as opaque, and it also selects which checkpoint row this caller reads and advances.
  • by identifies this worker. Passing the same value renews rather than fails, so acquiring and extending are one call.
  • millis is how long the lease is held from now.

Omitted entirely, no lease is taken and the caller reads and advances the shared checkpoint.

by​

string

key​

string

millis​

number

Returns​

Promise<SubscribeResult>

subscribed count of newly registered streams, watermark max at across all streams

Example​

const { subscribed, watermark } = await store().subscribe([
  { stream: "stats-user-1", source: "user-1" },
  { stream: "stats-user-2", source: "user-2", priority: 10 },
]);

See​

  • claim for discovering and leasing registered streams
  • prioritize for changing priority after subscription

truncate​

truncate: (targets) => Promise<TruncateResult>

Defined in: libs/act/src/types/ports.ts:1094

Atomically truncates streams and seeds each with a final event.

For each full target (no before), in a single transaction:

  1. Deletes all events for the stream
  2. Inserts a __snapshot__ (when snapshot is provided) or __tombstone__ event as the sole event on the stream

Subscriptions are not touched, so the event log and the subscriptions can live on different systems. A retired stream's subscription is inert (commits to a tombstoned stream are refused, so its work mark never rises and claim never returns it) and keeps each consumer's final watermark. Deleting those rows is optional housekeeping.

A windowed target (before set) is a pure prefix delete behind a real snapshot the app wrote — no seed, no tombstone, and the streams table is left untouched. The store finds the closest safe boundary — the latest __snapshot__ with created < before and, when max_id is given, id <= max_id — and deletes events with id < that snapshot's id, keeping the snapshot + tail. No qualifying snapshot ⇒ no-op (the stream is absent from the result). Because load() resets state at each snapshot on replay, events below the boundary contribute nothing to any load result — deleting them cannot change what load() returns. snapshot/meta must be omitted on windowed targets; before takes precedence when both appear.

Windowed targets are applied before full ones, and each full target pairs its own delete with its own seed. Both only matter for inputs the framework never produces — close-cycle dedups by stream before calling — but Store is a public port, so the order is contract rather than accident: a stream listed as both windowed and full ends with its FULL entry in the result, and a stream listed twice as full leaves exactly one seed carrying the last target's snapshot, reporting only that pass's own delete count. Deleting once for a deduped set and seeding once per entry would write two version: 0 rows, which the SQL adapters' unique (stream, version) constraint rejects outright.

Parameters​

targets​

object[]

Streams to truncate; full targets carry optional snapshot state and meta, windowed targets carry before (and optionally max_id, the min consumer watermark cap)

Returns​

Promise<TruncateResult>

Map keyed by stream name, each entry with deleted count and committed event (the new seed, or the surviving boundary snapshot on windowed entries, which also echo before)

See​

Act.close for the high-level close-the-books API that orchestrates safety checks, archive callbacks, and atomic truncate+seed

unblock​

unblock: (input) => Promise<number>

Defined in: libs/act/src/types/ports.ts:1007

Clears the blocked flag on streams without replaying their history. Sets blocked = false, retry = -1 (no attempt yet), error = null, and clears any lease bookkeeping. The at watermark stays where it was — the stream resumes from the next event after the last successful ack, not from zero.

The distinction from reset matters: reset() is for projection rebuilds (replay from event 0); unblock() is for recovering from a poison message after the operator fixes the underlying issue. Use unblock() when you don't want to re-process history.

Prefer Act.unblock() over calling this directly. Like reset(), this primitive doesn't raise the orchestrator's internal drain-armed flag — a settled Act instance will short-circuit and skip the resume. Act.unblock() wraps this and arms the flag.

Only streams that were actually blocked at call time count toward the return value; already-unblocked streams and unknown stream names are silently skipped. The atomic single-statement update makes the call safe to issue concurrently with claim() — workers holding a FOR UPDATE SKIP LOCKED lock won't see partial state.

Accepts either an explicit list of stream names or a StreamFilter for bulk recovery (e.g., "unblock every blocked order projection"). The blocked = true predicate is always applied — passing blocked: false in the filter matches nothing. An empty filter ({}) means "unblock everything that's blocked," which is a sane post-incident bulk recovery.

Parameters​

input​

string[] | StreamFilter

Stream names or a StreamFilter

Returns​

Promise<number>

Count of streams that were actually flipped (were blocked)

Example​

// By name (single targeted recovery)
await app.unblock(["webhooks-out-customer-42"]);

// By filter — unblock every blocked stream in a family
await app.unblock({ stream: "^webhooks-out-" });

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

// Low-level (does NOT trigger resume on settled apps)
await store().unblock(["webhooks-out-customer-42"]);

See​

  • Act.unblock for the high-level recovery API
  • reset for the rebuild-from-zero alternative