%%%------------------------------------------------------------------- %%% @doc Generic tagged outbox: a durable work queue written atomically %%% with the document batch. %%% %%% A write carrying the put option `outbox => [Tag]' appends one entry %%% per tag to the same RocksDB WriteBatch as the document, so the entry %%% and the doc commit or fail together. Entries are keyed %%% `prefix | db | tag | hlc' (the write's change HLC), giving %%% time-ordered scans per tag and exact-key acknowledgements. %%% %%% Rewrites of a tagged document replace the entry (the old HLC key is %%% deleted in the same batch, mirroring the changes feed), so the outbox %%% holds at most one pending entry per document per tag. Consumers fold %%% a tag (caller-side, no writer round-trip), process the CURRENT state %%% of each document, and ack by exact HLC key: if the document was %%% rewritten mid-processing, the new entry lives at a different key and %%% re-drives the consumer, so acking the processed key never loses work. %%% Un-acked entries are the consumer's checkpoint; there is nothing else %%% to persist. %%% %%% docdb knows nothing about what tags mean; producers (for example the %%% barrel's embedding indexer) define the semantics. %%% @end %%%------------------------------------------------------------------- -module(barrel_outbox). -include("barrel_docdb.hrl"). %% Batch op builders (called from the barrel_db_server write paths) -export([write_ops/7]). %% Consumer API -export([fold/5, fold/6]). %% Entry codec (exposed for tests) -export([encode_entry/3, decode_entry/1]). -type tag() :: binary(). -type entry() :: #{ hlc := barrel_hlc:timestamp(), id := docid(), rev := revid(), deleted := boolean() }. -export_type([tag/0, entry/0]). %%==================================================================== %% Batch op builders %%==================================================================== %% @doc Batch ops for a tagged write: one put per tag at the new HLC, %% plus a replace-delete of the previous entry when the doc existed. %% Deleting a key that was never written is a RocksDB no-op, so a %% previously untagged write is harmless. -spec write_ops(db_name(), [tag()], barrel_hlc:timestamp(), barrel_hlc:timestamp() | undefined, docid(), revid(), boolean()) -> [term()]. write_ops(_DbName, [], _NewHlc, _OldHlc, _DocId, _Rev, _Deleted) -> []; write_ops(DbName0, Tags, NewHlc, OldHlc, DocId, Rev, Deleted) when is_list(Tags) -> DbName = barrel_keyspace:resolve(DbName0), Entry = encode_entry(DocId, Rev, Deleted), lists:flatmap( fun(Tag) when is_binary(Tag) -> Put = {put, barrel_store_keys:outbox_key(DbName, Tag, NewHlc), Entry}, case OldHlc of undefined -> [Put]; _ -> [Put, {delete, barrel_store_keys:outbox_key(DbName, Tag, OldHlc)}] end end, Tags). %%==================================================================== %% Consumer API %%==================================================================== %% @doc Fold over the pending entries of a tag in HLC order. %% Runs in the caller's process (pure RocksDB range scan). `Fun' receives %% a decoded entry() and the accumulator and returns `{ok, Acc}' or %% `{stop, Acc}'. Options: `limit' (max entries to visit). -spec fold(barrel_store_rocksdb:db_ref(), db_name(), tag(), fun((entry(), term()) -> {ok, term()} | {stop, term()}), term()) -> term(). fold(StoreRef, DbName, Tag, Fun, Acc) -> fold(StoreRef, DbName, Tag, Fun, Acc, #{}). %% @doc Fold over the pending entries of a tag with options. -spec fold(barrel_store_rocksdb:db_ref(), db_name(), tag(), fun((entry(), term()) -> {ok, term()} | {stop, term()}), term(), map()) -> term(). fold(StoreRef, DbName0, Tag, Fun, Acc, Opts) -> DbName = barrel_keyspace:resolve(DbName0), Start = barrel_store_keys:outbox_prefix(DbName, Tag), End = barrel_store_keys:outbox_end(DbName, Tag), Limit = maps:get(limit, Opts, infinity), WrapFun = fun(Key, Value, {N, AccIn}) -> {_Tag, Hlc} = barrel_store_keys:decode_outbox_key(DbName, Key), Entry = (decode_entry(Value))#{hlc => Hlc}, case Fun(Entry, AccIn) of {ok, AccOut} when N + 1 >= Limit -> {stop, {N + 1, AccOut}}; {ok, AccOut} -> {ok, {N + 1, AccOut}}; {stop, AccOut} -> {stop, {N + 1, AccOut}} end end, {_Count, FinalAcc} = barrel_store_rocksdb:fold_range(StoreRef, Start, End, WrapFun, {0, Acc}), FinalAcc. %%==================================================================== %% Entry codec %%==================================================================== %% @doc Encode an outbox entry value (compact binary, same style as the %% change entry codec). -spec encode_entry(docid(), revid(), boolean()) -> binary(). encode_entry(DocId, Rev, Deleted) -> DelBit = case Deleted of true -> 1; false -> 0 end, <<(byte_size(DocId)):16, DocId/binary, (byte_size(Rev)):16, Rev/binary, DelBit:8>>. %% @doc Decode an outbox entry value (without the key-derived hlc). -spec decode_entry(binary()) -> #{id := docid(), rev := revid(), deleted := boolean()}. decode_entry(<>) -> #{id => DocId, rev => Rev, deleted => DelBit =:= 1}.