reckon_db_streams (reckon_db v5.11.4)
View SourceStreams API facade for reckon-db
Provides the public API for stream operations: - append: Write events to a stream with optimistic concurrency - read: Read events from a stream - get_version: Get current stream version - exists: Check if stream exists - list_streams: List all streams in the store
Summary
Functions
Append events to a stream with expected version check
Conditionally append events under the DCB pseudo-stream (Dynamic Consistency Boundary, Phase 3, 2.4.0+).
Delete a stream and all its events
Check if a stream exists
Get current version of a stream
The store's monotonic total event count — an O(1) read of the counter that every append batch maintains (see bump_global_event_count/1). Absent counter (a store that predates the counter, or has never been appended to) reads as 0. This is the cheap source for ingest-rate dashboards, replacing a full-store scan.
Check if a store contains at least one event. Cannot rely on stream existence alone — streams can survive after all their events are deleted (truncation, GDPR erasure). Checks for actual event data by reading 1 event globally.
List all streams in the store
Read events from a stream
Read events from a stream with explicit options.
Read all events from a stream
Read all events across all streams in global epoch_us order.
Read all events of specific types from all streams using Khepri native filtering.
Read all events whose metadata key = value.
Read all events matching tags from all streams.
Types
-type direction() :: forward | backward.
-type event() :: #event{event_id :: binary(), event_type :: binary(), stream_id :: binary(), version :: non_neg_integer(), data :: map() | binary(), metadata :: map(), tags :: [binary()] | undefined, timestamp :: integer(), epoch_us :: integer(), data_content_type :: binary(), metadata_content_type :: binary(), prev_event_hash :: binary() | undefined, mac :: {KeyId :: non_neg_integer(), MacBytes :: binary()} | undefined, signature :: binary() | undefined}.
-type integrity_ctx() :: disabled | {enabled, Key :: binary(), ChainStart :: non_neg_integer()}.
-type read_opts() :: #{verify => verify_mode()}.
-type verify_mode() :: skip_legacy | strict | skip_all.
Functions
-spec append(atom(), binary(), integer(), [new_event()]) -> {ok, non_neg_integer()} | {error, term()}.
Append events to a stream with expected version check
Expected version semantics: -1 (NO_STREAM) - Stream must not exist (first write) -2 (ANY_VERSION) - No version check, always append N >= 0 - Stream version must equal N
Returns {ok, NewVersion} on success or {error, Reason} on failure.
-spec append_if_no_tag_matches(StoreId :: atom(), TagFilter :: reckon_gater_types:tag_filter(), SeqCutoff :: reckon_gater_types:seq_cutoff(), Events :: [reckon_db_log_backend:new_event()]) -> {ok, LastSeq :: non_neg_integer()} | {error, {context_changed, non_neg_integer()}} | {error, no_events} | {error, integrity_not_supported_in_dcb_v1} | {error, term()}.
Conditionally append events under the DCB pseudo-stream (Dynamic Consistency Boundary, Phase 3, 2.4.0+).
Unlike append/4,5, the precondition is NOT a stream-version check; it is a tag-filter context query. Returns {error, {context_changed, MaxSeq}} when any event matching TagFilter has seq above SeqCutoff.
v1 refuses on stores with integrity enabled (DCB v1 lacks HMAC chain). Returns {error, integrity_not_supported_in_dcb_v1} in that case.
See: plans/PLAN_DCB_IMPLEMENTATION.md
NOTE: This facade calls reckon_db_dcb directly. P3.4 will route via reckon_db_gateway_worker for transport-layer consistency with append/4,5.
Delete a stream and all its events
Check if a stream exists
Get current version of a stream
Returns: -1 - if stream doesn't exist or is empty N >= 0 - representing the version of the latest event
-spec global_event_count(atom()) -> {ok, non_neg_integer()}.
The store's monotonic total event count — an O(1) read of the counter that every append batch maintains (see bump_global_event_count/1). Absent counter (a store that predates the counter, or has never been appended to) reads as 0. This is the cheap source for ingest-rate dashboards, replacing a full-store scan.
Check if a store contains at least one event. Cannot rely on stream existence alone — streams can survive after all their events are deleted (truncation, GDPR erasure). Checks for actual event data by reading 1 event globally.
List all streams in the store
-spec read(atom(), binary(), non_neg_integer(), pos_integer(), direction()) -> {ok, [event()]} | {error, term()}.
Read events from a stream
Parameters: StoreId - The store identifier StreamId - The stream identifier StartVersion - Starting version (0-based) Count - Maximum number of events to read Direction - forward or backward
Returns {ok, [Event]} or {error, Reason}
-spec read(atom(), binary(), non_neg_integer(), pos_integer(), direction(), read_opts()) -> {ok, [event()]} | {error, term()}.
Read events from a stream with explicit options.
Currently supported options:
verify :: skip_legacy | strict | skip_all Tamper-resistance enforcement mode. Default: skip_legacy. - skip_legacy (default): events with version below the per-stream chain_start watermark are returned untouched (legacy data); events at or above the watermark are verified strictly and an integrity_violation is returned on any failure. - strict: every event must carry integrity fields and verify; legacy events surface as missing_integrity. - skip_all: no verification (dangerous; intended for migration tooling only).
Backward-direction reads always bypass chain verification in 2.1.0; the MAC alone could still be checked but is not in this release. Forward reads receive full chain + MAC verification.
Read all events from a stream
-spec read_all_global(atom(), non_neg_integer(), pos_integer()) -> {ok, [event()]} | {error, term()}.
Read all events across all streams in global epoch_us order.
Returns events sorted by epoch_us, skipping Offset events and returning up to BatchSize events. Used by catch-up subscriptions to replay historical events to a subscriber.
Parameters: StoreId - The store identifier Offset - Number of events to skip (0-based) BatchSize - Maximum number of events to return
Returns events sorted by epoch_us (global ordering).
Why this is cached, and why an index inside Khepri cannot replace it
This function has no true server-side pagination: Khepri's get_many/2 has no offset/limit primitive (if_name_matches/if_path_matches are regex-only, no numeric range condition), so ANY paginated design over it -- indexed or not -- pays a full-matching-subtree fetch on every call. Verified by building and benchmarking a secondary all index first (a natural-looking fix, since reckon_db_index already does exactly this for tags/event_type/{meta,_}): at 10k events it was NOT faster than the plain scan below, because those calls are ref+point-resolve (fine when a lookup matches a SMALL subset of the store, wrong when -- like read_all_global -- it touches nearly the whole store every time), and denormalizing full events into the index just duplicated the same full-subtree-fetch cost under a different path.
evoq_store_subscription:catch_up_historical/1 (the only real caller) pages through a store via a tight sequential burst of (StoreId, GrowingOffset, 1000) calls at subscription start -- which is exactly the access pattern this cache targets: pay the full scan-and-sort ONCE per burst (fingerprinted by global_event_count/1, already O(1) and already exact -- it only ever increments on append), not once per page. A store that grows mid-burst invalidates the cache on its own; the TTL below is a memory-hygiene bound for a burst that stalls or never completes, not a correctness mechanism.
Why the cache indexes one ETS row per event, not one row per store
A first cut of this cache (5.11.1) stored the WHOLE sorted list as a single ETS value and served a page via lists:nthtail/sublist on the list ets:lookup handed back. That still copies and walks the ENTIRE cached list on EVERY page: ets:lookup always deep-copies the full stored term out to the caller, so a "cache hit" against a real ~87k-event store (reproduced against a real evidence store, not a synthetic one -- see hecate-sentinel's CHANGELOG) cost ~110-130ms per page regardless of hit/miss, not the O(1) the fingerprinting was meant to buy. Against a store that size, blocking evoq_store_subscriptions synchronous catch-up for the whole burst was long enough on real (weaker-than-benchmark) hardware to plausibly trip an external liveness check mid-replay -- see evoq's async catch-up fix, same release train.
Indexing each event under its own {StoreId, Position} key makes a page read BatchSize independent point lookups (each O(1)/O(log N) in ETS, and each copies exactly one event, not the whole store) instead of one O(N) copy. The one full scan-and-sort per fingerprint change is unchanged -- sorting still requires seeing every event at least once -- only the PER-PAGE cost after that scan is fixed.
Two correctness properties a whole-list cache gets for free and a per-row one has to earn back, both closed by a per-rebuild Generation id every row (including the meta row) is tagged with:
1. The page bound must be the ACTUAL number of rows this generation wrote, never global_event_count/1. That counter is a fingerprint ("has anything changed"), not an authoritative row count -- it does not exist on stores that predate it (reads as 0, nothing backfilled), DCB appends (reckon_db_dcb) never bump it, and deletes never decrement it. Any of those makes it diverge from the real scanned length: paging against it would silently truncate catch-up (counter too low) or let a shrunk generation's stale trailing rows leak back in (counter too high). Len = length(SortedEvents) from THIS rebuild is the only authoritative bound. 2. A page is BatchSize independent ets:lookup calls, not one atomic read -- so while the WRITE side is atomic (the whole batch lands in one ets:insert/2, which OTP guarantees atomic and isolated for set tables), the READ side is not: a rebuild can still land between two of a page's lookups. If that rebuild's fresh sort reassigns a position this page already read (only possible if two events' epoch_us values are close enough, or a delete/DCB write races the scan, to change relative order -- sort_by_epoch/1 promises a stable sort of whatever it saw, nothing about positions surviving a later append), the page would silently mix two generations. Tagging each row with its Generation and checking it on every lookup turns that into a detectable condition instead of a silent one: a page that spans a rebuild is retried once, against the new generation, rather than returned torn.
-spec read_by_event_types(atom(), [binary()], pos_integer()) -> {ok, [event()]} | {error, term()}.
Read all events of specific types from all streams using Khepri native filtering.
This function uses Khepri's built-in #if_data_matches condition to filter events by type at the database level, avoiding loading all events into memory.
Parameters: StoreId - The store identifier EventTypes - List of event type binaries to match BatchSize - Maximum number of events to return (for pagination)
Returns events sorted by epoch_us (global ordering).
Read all events whose metadata key = value.
This is the sanctioned primitive applications build causation / correlation / saga read models on (e.g. read_by_metadata(Store, <<"causation_id">>, EventId)`). The store returns events matching a metadata key=value pair — bounded and indexed when `{meta, Key} is declared — and does NOT interpret what the key means. Lineage traversal, graphs, and read models are the application's job.
Indexed (O(matches)) when the store declared {meta, Key}; otherwise a one-time logger:warning is emitted and the query falls back to a whole-store scan (O(total events)).
-spec read_by_tags(atom(), [binary()], any | all, pos_integer()) -> {ok, [event()]} | {error, term()}.
Read all events matching tags from all streams.
Tags provide a mechanism for cross-stream querying without affecting stream-based concurrency control. This is useful for the process-centric model where you want to find all events related to specific participants.
Match Modes
any (default): Returns events containing ANY of the specified tags (union). Example: read_by_tags(Store, [<<"student:456">>, <<"student:789">>], any, 100) Returns events for either student.
all: Returns events containing ALL of the specified tags (intersection). Example: read_by_tags(Store, [<<"student:456">>, <<"course:CS101">>], all, 100) Returns only events tagged with both student 456 AND course CS101.
Parameters
StoreId - The store identifier Tags - List of tag binaries to match Match - any | all (matching strategy) BatchSize - Maximum number of events to return
Returns
Events sorted by epoch_us (global ordering).