reckon_db_streams (reckon_db v5.11.7)
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
Idempotently create the read-all-global cache table.
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
-spec ensure_cache_table() -> ok.
Idempotently create the read-all-global cache table.
Exported (not just called lazily from ensure_cached/1) so the top-level supervisor's own init callback can call this itself, making the SUPERVISOR the table's owner instead of whichever transient gateway worker happens to call read_all_global/3 first.
That mattered: a public ETS table is owned by whoever calls ets:new/2, and the table dies with its owner regardless of how many other processes still reference it by name. A table created lazily by a short-lived worker vanishes the instant that worker exits for ANY reason -- including a routine one-off supervised restart unrelated to this table -- and the next caller's ets:lookup/2 in cached_or_rebuilt/2 then crashes with {badarg, "the table identifier does not refer to an existing ETS table"}. Reproduced on every boot in practice: gateway workers restart routinely during a store's own startup churn, and whichever one happened to win the race to create this table took it down with it moments later.
Racing calls to THIS function are still safe on their own terms (ets:new/2 raises badarg on a name collision, meaning another process already won -- nothing to do) -- the fix is giving the table a caller that is never itself the process racing to use it moments later.
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).