reckon_db_dcb (reckon_db v5.8.1)
View SourceDCB conditional-append primitive.
Implements append_if_no_tag_matches/4 as a khepri:transaction/2 body. The transaction:
1. Verifies the tag-filter context ({context_changed, _} on conflict). 2. For integrity-enabled stores: verifies the seq counter and chain-tip still match what the caller observed when pre-stamping MACs. On mismatch aborts with {dcb_state_changed, _} and the outer loop retries with a fresh snapshot. 3. Writes events under ?DCB_STREAM_PATH ++ [SeqKey] plus one tag-index entry per tag at ?BY_TAG_PATH ++ [Tag, SeqKey]. 4. Updates the seq counter (and chain-tip if integrity is enabled).
The whole sequence happens inside one Ra log entry; atomic across the cluster. Either everything commits, or an abort comes back and nothing changed.
Integrity (reckon-db 3.2.0+)
On integrity-enabled stores, DCB events carry prev_event_hash + mac like any other integrity-bearing event. Because Khepri's Horus extractor rejects code containing crypto:* calls (even on unreachable branches; extraction sees the whole function), MAC chains are pre-computed OUTSIDE the transaction. Inside the transaction we verify the chain-tip + counter are still what we saw, then write the pre-stamped records.
Two retry vectors: - {context_changed, _}: tag-filter conflict. Callers (e.g., evoq_decision_runtime) retry with fresh context. - {dcb_state_changed, _}: counter or chain-tip moved between our pre-stamp read and the transaction. The retry loop in this module handles it transparently, bounded by ?INTEGRITY_RETRY_BUDGET.
Summary
Functions
Return all event types present in the DCB log, sorted by count descending.
Return all tags present in the DCB log, sorted by event count descending.
Return DCB events where data[Key] = Value, up to Limit events in ascending seq order.
Return DCB events matching the composite payload field combination, up to Limit events in ascending seq order.
Read DCB events in ascending seq order.
Types
Functions
-spec all_event_types(atom()) -> {ok, [{binary(), non_neg_integer()}]} | {error, term()}.
Return all event types present in the DCB log, sorted by count descending.
Each type is returned as {EventType, EventCount}. Uses the by_event_type index.
-spec all_tags(atom()) -> {ok, [{binary(), non_neg_integer()}]} | {error, term()}.
Return all tags present in the DCB log, sorted by event count descending.
Each tag is returned as {Tag, EventCount}. Uses the by_tag index so it does not scan DCB events directly.
-spec append_if_no_tag_matches(StoreId :: atom() | binary(), TagFilter :: reckon_gater_types:tag_filter(), SeqCutoff :: reckon_gater_types:seq_cutoff(), Events :: [map()]) -> {ok, LastSeq :: non_neg_integer()} | {error, {context_changed, non_neg_integer()}} | {error, no_events} | {error, dcb_concurrent_writer_exhausted} | {error, term()}.
-spec read_by_payload(atom(), binary(), binary(), pos_integer()) -> {ok, [#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}]} | {error, term()}.
Return DCB events where data[Key] = Value, up to Limit events in ascending seq order.
Requires {payload, Key} declared in the store's index config.
-spec read_by_payload_hash(atom(), [binary()], [binary()], pos_integer()) -> {ok, [#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}]} | {error, term()}.
Return DCB events matching the composite payload field combination, up to Limit events in ascending seq order.
Requires {payload_hash, Keys} declared in the store's index config. Field order in Keys/Values is ignored (hash is order-independent).
-spec read_log(atom(), non_neg_integer(), pos_integer()) -> {ok, [#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}], non_neg_integer()} | {error, term()}.
Read DCB events in ascending seq order.
FromSeq is the first sequence number to include (0 for the beginning). Limit is the maximum number of events to return.
Returns {ok, Events, TotalCount} where TotalCount is the total number of events in the DCB log (used for pagination).