Skip to main content

InMemoryStore

@rotorsoft/act-root

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

Class: InMemoryStore

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

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​

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);
  });
});
import { InMemoryStore } from "@rotorsoft/act";

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

// Use for specific test scenarios
await testStore.commit("test-stream", events, meta);
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:566

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:578

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:449

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:393

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

dispose()​

dispose(): Promise<void>

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

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:324

Drop all data from the store.

Returns​

Promise<void>

Promise that resolves when the store is cleared.

Implementation of​

Store.drop

prioritize()​

prioritize(filter, priority): Promise<number>

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

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:349

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:793

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:716

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:634

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:944

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 (matching the adapter's commit-id convention — InMemory uses _events.length). 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:316

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:527

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, { committed: Committed<Schemas, string>; deleted: number; }>>

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

Atomically truncates streams and seeds each with a snapshot or tombstone.

Parameters​

targets​

object[]

Streams to truncate with optional snapshot state and meta.

Returns​

Promise<Map<string, { 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:668

Clear the blocked flag (and retry / error / lease) on the matched streams without touching the watermark. Streams that aren't blocked at call time are silently skipped. Accepts either an explicit list of names or a StreamFilter. The filter form always restricts to blocked streams — passing blocked: false matches nothing. See Store.unblock.

Parameters​

input​

string[] | StreamFilter

Stream names or a filter selecting the streams to unblock.

Returns​

Promise<number>

Count of streams that were actually flipped (were blocked).

Implementation of​

Store.unblock