Skip to main content
Version: Current

InMemoryStore

@rotorsoft/act-root

@rotorsoft/act-root / act/src / InMemoryStore

Class: InMemoryStore

Defined in: libs/act/src/adapters/in-memory-store.ts:320

In-memory event store implementation.

This is the default store used by Act when no other store is injected. It stores all events in memory and is suitable for:

  • Development and prototyping
  • Unit and integration testing
  • Demonstrations and examples

Not suitable for production - all data is lost when the process exits. Use PostgresStore for production deployments.

The in-memory store provides:

  • Full Store interface implementation
  • Optimistic concurrency control
  • Stream leasing for distributed processing simulation
  • Snapshot support
  • Fast performance (no I/O overhead)

Store.notify is intentionally not implemented. The notify hook is a cross-process wake-up signal — local commits already arm the drain via do(). An in-memory store is single-process by definition, so there is no remote writer to be notified of. The Act orchestrator detects the absence and falls back to the existing debounce/poll path.

Examples​

Using in tests

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

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

  it("increments", async () => {
    await app.do("increment", target, { by: 5 });
    const snapshot = await app.load(Counter, "counter-1");
    expect(snapshot.state.count).toBe(5);
  });
});

Explicit instantiation

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

const testStore = new InMemoryStore();
await testStore.seed();

// Use for specific test scenarios
await testStore.commit("test-stream", events, meta);

Querying events

const events: any[] = [];
await store().query(
  (event) => events.push(event),
  { stream: "test-stream" }
);
console.log(`Found ${events.length} events`);

See​

  • Store for the interface definition
  • PostgresStore for production use
  • store for injecting stores

Implements​

Constructors​

Constructor​

new InMemoryStore(): InMemoryStore

Returns​

InMemoryStore

Methods​

ack()​

ack(leases): Promise<object[]>

Defined in: libs/act/src/adapters/in-memory-store.ts:665

Acknowledge completion of processing for leased streams.

Parameters​

leases​

Lease[]

Leases to acknowledge, including last processed watermark and lease holder.

Returns​

Promise<object[]>

Implementation of​

Store.ack

block()​

block(leases): Promise<object[]>

Defined in: libs/act/src/adapters/in-memory-store.ts:682

Block a stream for processing after failing to process and reaching max retries with blocking enabled.

Parameters​

leases​

BlockedLease[]

Leases to block, including lease holder and last error message.

Returns​

Promise<object[]>

Blocked leases.

Implementation of​

Store.block

claim()​

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

Defined in: libs/act/src/adapters/in-memory-store.ts:558

Atomically discovers and leases streams for processing. Fuses poll + lease into a single operation.

Parameters​

lagging​

number

Max streams from lagging frontier.

leading​

number

Max streams from leading frontier.

by​

string

Lease holder identifier.

millis​

number

Lease duration in milliseconds.

lane?​

string

Returns​

Promise<Lease[]>

Granted leases.

Implementation of​

Store.claim

commit()​

commit<E>(stream, msgs, meta, expectedVersion?): Promise<Committed<E, keyof E>[]>

Defined in: libs/act/src/adapters/in-memory-store.ts:491

Commit one or more events to a stream.

Type Parameters​

E​

E extends Schemas

Parameters​

stream​

string

The stream name.

msgs​

Message<E, keyof E>[]

The events/messages to commit.

meta​

EventMeta

Event metadata.

expectedVersion?​

number

Optional optimistic concurrency check.

Returns​

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

The committed events with metadata.

Throws​

ConcurrencyError if expectedVersion does not match.

Implementation of​

Store.commit

defer()​

defer(input, deferred_at): Promise<number>

Defined in: libs/act/src/adapters/in-memory-store.ts:698

Hold the matched streams out of claim until deferred_at (ms since epoch) — see Store.defer. Accepts an explicit list of names or a StreamFilter, mirroring reset/unblock. Persisted store state (unlike in-process backoff), so the skip is honored by every competing worker. Unknown names are silently skipped.

Parameters​

input​

string[] | StreamFilter

deferred_at​

number

Returns​

Promise<number>

Count of streams whose deferred_at was set.

Implementation of​

Store.defer

dispose()​

dispose(): Promise<void>

Defined in: libs/act/src/adapters/in-memory-store.ts:380

Dispose of the store and clear all events.

Returns​

Promise<void>

Promise that resolves when disposal is complete.

Implementation of​

Store.dispose

drop()​

drop(): Promise<void>

Defined in: libs/act/src/adapters/in-memory-store.ts:397

Drop all data from the store.

Returns​

Promise<void>

Promise that resolves when the store is cleared.

Implementation of​

Store.drop

forget_pii()​

forget_pii(stream): Promise<number>

Defined in: libs/act/src/adapters/in-memory-store.ts:813

Wipe the sensitive-data payload for every event on the stream — see Store.forget_pii. O(1) drop of the stream's inner Map; the size of that Map is the count of events that had PII. Idempotent: a second call finds no inner Map and returns 0.

Parameters​

stream​

string

Target stream.

Returns​

Promise<number>

Count of events whose isolated PII payload was deleted.

Implementation of​

Store.forget_pii

prioritize()​

prioritize(filter, priority): Promise<number>

Defined in: libs/act/src/adapters/in-memory-store.ts:849

Bulk-update priority of streams matching filter. Mirrors query_streams's filter semantics — see Store.prioritize. Unlike subscribe (which keeps max() of registered priorities), this sets the priority outright — operator override for the build-time scheduling policy.

Parameters​

filter​

StreamFilter

priority​

number

Returns​

Promise<number>

Count of streams whose priority changed.

Implementation of​

Store.prioritize

query()​

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

Defined in: libs/act/src/adapters/in-memory-store.ts:422

Query events in the store, optionally filtered by query options.

Type Parameters​

E​

E extends Schemas

Parameters​

callback​

(event) => void

Function to call for each 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_snaps?: boolean; }>

Optional query options.

Returns​

Promise<number>

The number of events processed.

Implementation of​

Store.query

query_stats()​

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

Defined in: libs/act/src/adapters/in-memory-store.ts:964

Per-stream aggregated stats — see Store.query_stats.

Single forward scan over the in-memory event list, accumulating per stream. The "cheap heads" cost tier from durable adapters doesn't apply here (InMemory has no indexes); correctness is the goal, perf is a non-issue.

Scope rules:

  • Array input — explicit stream names, regardless of subscription.
  • Filter input — stream/stream_exact match against event-bearing stream names; source/source_exact/blocked require a corresponding subscription in _streams (those are subscription concepts, not event concepts). Empty filter {} matches every event-bearing stream.

Type Parameters​

E​

E extends Schemas

Parameters​

input​

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

options?​

QueryStatsOptions<E>

Returns​

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

Implementation of​

Store.query_stats

query_streams()​

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

Defined in: libs/act/src/adapters/in-memory-store.ts:868

Streams registered subscription positions to the callback, ordered by stream name. Returns the highest event id in the store and the count of positions emitted.

Parameters​

callback​

(position) => void

query?​

QueryStreams

Returns​

Promise<QueryStreamsResult>

Implementation of​

Store.query_streams

reset()​

reset(input): Promise<number>

Defined in: libs/act/src/adapters/in-memory-store.ts:770

Reset watermarks to -1, clearing retry, blocked, error, and lease state so the matched streams can be replayed from the beginning. Accepts either an explicit list of names or a StreamFilter.

Parameters​

input​

string[] | StreamFilter

Stream names or a filter selecting the streams to reset.

Returns​

Promise<number>

Count of streams that were actually reset.

Implementation of​

Store.reset

restore()​

restore(driver): Promise<void>

Defined in: libs/act/src/adapters/in-memory-store.ts:1169

Atomically wipe-and-rebuild the store under an in-process snapshot.

Captures every index state up front, clears it, then hands the orchestrator a per-event insert callback via the driver. Any throw inside the driver restores the snapshot, leaving the store byte-for-byte unchanged from the operator's perspective.

ids are reassigned 0..N-1 as events arrive (dense — the monotonic id counter restarts at 0 for the rebuild). created is preserved verbatim from the source.

Parameters​

driver​

(callback) => Promise<void>

Returns​

Promise<void>

Implementation of​

Store.restore

seed()​

seed(): Promise<void>

Defined in: libs/act/src/adapters/in-memory-store.ts:389

Seed the store with initial data (no-op for in-memory).

Returns​

Promise<void>

Promise that resolves when seeding is complete.

Implementation of​

Store.seed

subscribe()​

subscribe(streams): Promise<{ subscribed: number; watermark: number; }>

Defined in: libs/act/src/adapters/in-memory-store.ts:626

Registers streams for event processing. When the same stream is resubscribed with a different priority, the maximum wins — so the highest-priority registered reaction sets the scheduling lane. Use prioritize for operator runtime overrides.

Parameters​

streams​

object[]

Streams to register with optional source + priority.

Returns​

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

subscribed count and current max watermark.

Implementation of​

Store.subscribe

truncate()​

truncate(targets): Promise<Map<string, { before?: Date; committed: Committed<Schemas, string>; deleted: number; }>>

Defined in: libs/act/src/adapters/in-memory-store.ts:1061

Atomically truncates streams and seeds each with a snapshot or tombstone. Windowed targets (before set) prune the prefix below the closest safe __snapshot__ instead — no seed, subscriptions untouched, no-op when no snapshot qualifies.

Parameters​

targets​

object[]

Streams to truncate with optional snapshot state and meta, or a before/max_id boundary for a windowed prefix delete.

Returns​

Promise<Map<string, { before?: Date; committed: Committed<Schemas, string>; deleted: number; }>>

Map keyed by stream name, each entry with deleted count and committed event.

Implementation of​

Store.truncate

unblock()​

unblock(input): Promise<number>

Defined in: libs/act/src/adapters/in-memory-store.ts:820

Clears the blocked flag on streams without replaying their history. Sets blocked = false, retry_count = 0, 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 "needs drain" 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

Implementation of​

Store.unblock