Current section
Files
Jump to
Current section
Files
src/barrel_db_server.erl
%%%-------------------------------------------------------------------
%%% @doc barrel_db_server - Individual database server process
%%%
%%% Manages a single database instance. Each database has its own
%%% gen_server process that handles all operations for that database.
%%% Opens both a document store (regular RocksDB) and an attachment
%%% store (RocksDB with BlobDB enabled).
%%% @end
%%%-------------------------------------------------------------------
-module(barrel_db_server).
-behaviour(gen_server).
-include("barrel_docdb.hrl").
%% API
-export([start_link/2]).
-export([info/1, stop/1]).
-export([get_store_ref/1, get_att_ref/1]).
-export([sweep_ttl/1]).
%% Document API
-export([
put_doc/3,
put_docs/3,
get_doc/3,
get_docs/3,
delete_doc/3,
fold_docs/3,
fold_docs/4,
resolve_conflict/4,
get_conflicts/2
]).
%% Tagged outbox API (acks are writes, so they go through the server)
-export([outbox_ack/3]).
%% Embedding column (computed vectors written back by an indexer)
-export([set_doc_embedding/4]).
%% Retention sweep (also runs on a timer; exposed for tests and tools)
-export([sweep_retention/1]).
%% Timeline fork (checkpoint both stores from inside the writer)
-export([checkpoint_to/3]).
%% Replication API
-export([
put_version/5,
diff_versions/2
]).
%% Local document API (for checkpoints, not replicated)
-export([
put_local_doc/3,
get_local_doc/2,
delete_local_doc/2,
fold_local_docs/4
]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2]).
-record(state, {
name :: binary(),
keyspace :: binary(), %% name used for storage keys (parent's on a branch)
parent :: binary() | undefined, %% timeline parent (branches only)
fork_hlc :: barrel_hlc:timestamp() | undefined, %% fork instant (branches only)
config :: map(),
db_path :: string(),
store_ref :: barrel_store_rocksdb:db_ref() | undefined,
att_ref :: barrel_att_store:att_ref() | undefined,
%% EncryptedEnv shared by both stores. The NIF frees the underlying
%% env when the handle is garbage collected, so it must stay
%% referenced for the whole db lifetime (also carried in the store
%% and att refs).
env :: rocksdb:env_handle() | undefined,
filter_pid :: pid() | undefined, %% Compaction filter handler for this database
compaction_timer :: reference() | undefined, %% Timer for periodic compaction checks
compaction_interval :: pos_integer(), %% Interval between checks in ms (default: 1 hour)
compaction_size_threshold :: pos_integer(), %% Size threshold in bytes to trigger compaction
retention_period :: non_neg_integer(), %% Retention window in seconds (0 = infinite)
retention_timer :: reference() | undefined, %% Timer for periodic retention sweeps
retention_interval :: pos_integer(), %% Interval between sweeps in ms
ttl_sweep_interval = 0 :: non_neg_integer(), %% Doc TTL sweep interval in ms (0 = off)
ttl_sweep_batch = 512 :: pos_integer(), %% Max expired docs per sweep pass
ttl_timer :: reference() | undefined %% Timer for periodic TTL sweeps
}).
%% Default compaction settings
-define(DEFAULT_COMPACTION_INTERVAL, 3600000). %% 1 hour in ms
-define(DEFAULT_COMPACTION_SIZE_THRESHOLD, 1073741824). %% 1 GB in bytes
%% Default retention settings: 30 days, swept every 10 minutes.
%% retention_period 0 means infinite (no sweeper started).
-define(DEFAULT_RETENTION_PERIOD, 2592000). %% 30 days in seconds
-define(DEFAULT_RETENTION_INTERVAL, 600000). %% 10 minutes in ms
%% Live conflict siblings per document are bounded; beyond the bound
%% the lowest versions are dropped deterministically.
-define(DEFAULT_MAX_CONFLICT_VERSIONS, 8).
%%====================================================================
%% API functions
%%====================================================================
%% @doc Start the database server
-spec start_link(binary(), map()) -> {ok, pid()} | {error, term()}.
start_link(Name, Config) ->
gen_server:start_link(?MODULE, [Name, Config], []).
%% @doc Get database info
-spec info(pid()) -> {ok, map()} | {error, term()}.
info(Pid) ->
gen_server:call(Pid, info).
%% @doc Checkpoint both stores into a branch's directories, minting
%% the fork instant inside the writer: every write applied before this
%% call is in the checkpoint and every later one has a larger HLC.
%% Blocks the writer for the memtable flush + hard links.
-spec checkpoint_to(pid(), string(), string()) ->
{ok, barrel_hlc:timestamp()} | {error, term()}.
checkpoint_to(Pid, DocsPath, AttPath) ->
gen_server:call(Pid, {checkpoint_to, DocsPath, AttPath}, infinity).
%% @doc Get the document store reference
-spec get_store_ref(pid()) -> {ok, barrel_store_rocksdb:db_ref()} | {error, term()}.
get_store_ref(Pid) ->
gen_server:call(Pid, get_store_ref).
%% @doc Get the attachment store reference
-spec get_att_ref(pid()) -> {ok, barrel_att_store:att_ref()} | {error, term()}.
get_att_ref(Pid) ->
gen_server:call(Pid, get_att_ref).
%% @doc Stop the database server
-spec stop(pid()) -> ok.
stop(Pid) ->
gen_server:stop(Pid).
%%====================================================================
%% Document API functions
%%====================================================================
%% @doc Put a document (create or update)
-spec put_doc(pid(), map(), map()) -> {ok, map()} | {error, term()}.
put_doc(Pid, Doc, Opts) ->
gen_server:call(Pid, {put_doc, Doc, Opts}).
%% @doc Put multiple documents (batch write)
-spec put_docs(pid(), [map()], map()) -> [{ok, map()} | {error, term()}].
put_docs(Pid, Docs, Opts) ->
gen_server:call(Pid, {put_docs, Docs, Opts}).
%% @doc Get a document
%% When raw_body => true in Opts, returns {ok, CborBin, Meta} for zero-copy responses
-spec get_doc(pid(), binary(), map()) -> {ok, map()} | {ok, binary(), map()} | {error, not_found} | {error, term()}.
get_doc(Pid, DocId, Opts) ->
gen_server:call(Pid, {get_doc, DocId, Opts}).
%% @doc Get multiple documents (batch read)
-spec get_docs(pid(), [binary()], map()) -> [{ok, map()} | {error, not_found} | {error, term()}].
get_docs(Pid, DocIds, Opts) ->
gen_server:call(Pid, {get_docs, DocIds, Opts}).
%% @doc Delete a document
-spec delete_doc(pid(), binary(), map()) -> {ok, map()} | {error, term()}.
delete_doc(Pid, DocId, Opts) ->
gen_server:call(Pid, {delete_doc, DocId, Opts}).
%% @doc Fold over all documents
-spec fold_docs(pid(), fun(), term()) -> {ok, term()}.
fold_docs(Pid, Fun, Acc) ->
gen_server:call(Pid, {fold_docs, Fun, Acc}, infinity).
%% @doc Fold over all documents with options
-spec fold_docs(pid(), fun(), term(), map()) -> {ok, term()}.
fold_docs(Pid, Fun, Acc, Opts) when is_map(Opts) ->
gen_server:call(Pid, {fold_docs, Fun, Acc, Opts}, infinity).
%% @doc Get conflicts for a document
%% Returns list of conflicting revision IDs (excluding the winner)
-spec get_conflicts(pid(), binary()) -> {ok, [binary()]} | {error, term()}.
get_conflicts(Pid, DocId) ->
gen_server:call(Pid, {get_conflicts, DocId}).
%% @doc Resolve a conflict by choosing a winning revision or providing a merged doc
%% Resolution can be:
%% - {choose, Rev} - keep this revision as winner, delete other branches
%% - {merge, Doc} - create new revision merging all conflicts
-spec resolve_conflict(pid(), binary(), binary(), {choose, binary()} | {merge, map()}) ->
{ok, map()} | {error, term()}.
resolve_conflict(Pid, DocId, BaseRev, Resolution) ->
gen_server:call(Pid, {resolve_conflict, DocId, BaseRev, Resolution}).
%% @doc Acknowledge processed outbox entries by their exact HLC keys.
%% Deleting a key that was already replaced by a newer write is a no-op,
%% so acks never lose work (see barrel_outbox).
-spec outbox_ack(pid(), barrel_outbox:tag(), [barrel_hlc:timestamp()]) ->
ok | {error, term()}.
outbox_ack(Pid, Tag, Hlcs) ->
gen_server:call(Pid, {outbox_ack, Tag, Hlcs}).
%% @doc Store a computed embedding as the document's embedding column.
%% CAS on the expected revision: embeddings are derived data, so this
%% never bumps the revision or emits a change; a conflict means a newer
%% write exists (which re-drives the computing indexer anyway).
-spec set_doc_embedding(pid(), binary(), binary(), [number()]) ->
ok | {error, conflict | not_found | term()}.
set_doc_embedding(Pid, DocId, ExpectedRev, Vector) ->
gen_server:call(Pid, {set_doc_embedding, DocId, ExpectedRev, Vector}).
%% @doc Run a retention sweep now. Prunes history entries, superseded
%% version bodies and expired tombstones older than the retention
%% window, then advances the history floor. No-op when retention is
%% infinite (retention_period 0).
-spec sweep_retention(pid()) -> {ok, map()} | {error, term()}.
sweep_retention(Pid) ->
gen_server:call(Pid, sweep_retention, infinity).
%% @doc Run one doc TTL sweep pass now (test and ops hook). Returns
%% the number of docs tombstoned.
-spec sweep_ttl(pid()) -> {ok, non_neg_integer()}.
sweep_ttl(Pid) ->
gen_server:call(Pid, ttl_sweep, infinity).
%%====================================================================
%% Replication API functions
%%====================================================================
%% @doc Put a document with explicit revision history (for replication)
%% Returns {ok, DocId, RevId} on success.
-spec put_version(pid(), map(), binary(), binary(), boolean()) ->
{ok, binary(), binary()} | {error, term()}.
put_version(Pid, Doc, VersionToken, VVBin, Deleted) ->
gen_server:call(Pid, {put_version, Doc, VersionToken, VVBin, Deleted}).
%% @doc Get revisions difference (for replication)
%% Returns {ok, Missing, PossibleAncestors}
-spec diff_versions(pid(), #{binary() => binary()}) ->
{ok, #{binary() => missing | have}} | {error, term()}.
diff_versions(Pid, TokenMap) when is_map(TokenMap) ->
gen_server:call(Pid, {diff_versions, TokenMap}).
%%====================================================================
%% Local Document API functions
%%====================================================================
%% @doc Put a local document (not replicated)
-spec put_local_doc(pid(), binary(), map()) -> ok | {error, term()}.
put_local_doc(Pid, DocId, Doc) ->
gen_server:call(Pid, {put_local_doc, DocId, Doc}).
%% @doc Get a local document
-spec get_local_doc(pid(), binary()) -> {ok, map()} | {error, not_found}.
get_local_doc(Pid, DocId) ->
gen_server:call(Pid, {get_local_doc, DocId}).
%% @doc Delete a local document
-spec delete_local_doc(pid(), binary()) -> ok | {error, not_found}.
delete_local_doc(Pid, DocId) ->
gen_server:call(Pid, {delete_local_doc, DocId}).
%% @doc Fold over local documents with a given prefix
-spec fold_local_docs(pid(), binary(), fun((binary(), map(), term()) -> term()), term()) ->
{ok, term()} | {error, term()}.
fold_local_docs(Pid, Prefix, Fun, Acc) ->
gen_server:call(Pid, {fold_local_docs, Prefix, Fun, Acc}).
%%====================================================================
%% gen_server callbacks
%%====================================================================
%% @doc Initialize the database server
init([Name, Config]) ->
process_flag(trap_exit, true),
%% Validate the channels config before opening anything: a bad
%% pattern set must fail the open, not the first write.
case barrel_channel:compile(maps:get(channels, Config, #{})) of
{ok, CompiledChannels} ->
init(Name, Config, CompiledChannels);
{error, ChanReason} ->
{stop, {invalid_channels, ChanReason}}
end.
init(Name, Config, CompiledChannels) ->
%% Get data directory from config. Fall back to the `data_dir' app env so a
%% node can relocate ALL of its dbs (including internal ones created with no
%% opts) off the shared default - lets two local nodes use distinct dirs.
DefaultDataDir = application:get_env(barrel_docdb, data_dir, "/tmp/barrel_data"),
DataDir = maps:get(data_dir, Config, DefaultDataDir),
DbPath = filename:join([DataDir, binary_to_list(Name)]),
%% A previous instance that crashed without terminate leaves its
%% store handle in persistent_term: the RocksDB LOCK stays held and
%% the database cannot reopen. Release it when its owner is gone.
ok = release_stale_refs(Name),
%% Branch identity comes from the TIMELINE sidecar file: it must be
%% readable BEFORE the compaction filter starts (the filter matches
%% key-embedded names) and before RocksDB opens.
case barrel_keyspace:read_meta(DbPath) of
{ok, #{keyspace := SidecarKs, parent := SidecarParent,
fork_hlc := SidecarForkHlc}} ->
init(Name, Config, CompiledChannels, DbPath,
SidecarKs, SidecarParent,
barrel_hlc:decode(SidecarForkHlc));
not_found ->
init(Name, Config, CompiledChannels, DbPath,
Name, undefined, undefined);
{error, corrupt_timeline_meta} ->
{stop, corrupt_timeline_meta}
end.
init(Name, Config, CompiledChannels, DbPath, Keyspace, Parent, ForkHlc) ->
%% Encryption resolves on the KEYSPACE (a branch opens with its
%% parent's key: the checkpoint is ciphertext under it) and is
%% checked against the CRYPTO marker before RocksDB opens, so a
%% wrong key or an encrypted/plaintext mismatch fails cleanly
%% instead of surfacing as corruption.
case init_encryption(Config, Keyspace, DbPath) of
{ok, Env} ->
init(Name, Config, CompiledChannels, DbPath, Keyspace,
Parent, ForkHlc, Env);
{error, EncReason} ->
{stop, EncReason}
end.
init(Name, Config, CompiledChannels, DbPath, Keyspace, Parent, ForkHlc, Env) ->
%% Start compaction filter handler BEFORE opening RocksDB
%% (handler pid is passed to CF options). The filter matches the
%% name embedded in keys, i.e. the keyspace.
PruneDepth = maps:get(prune_depth, Config, 1000),
{ok, FilterPid} = barrel_compaction_filter:start_link(#{
db_name => Keyspace,
prune_depth => PruneDepth
}),
%% Open document store (RocksDB with column families including body CF with BlobDB)
DocStorePath = filename:join(DbPath, "docs"),
StoreOpts0 = maps:get(store_opts, Config, #{}),
%% Pass the filter handler to the store options
StoreOpts = add_env_opt(StoreOpts0#{compaction_filter_handler => FilterPid},
Env),
case barrel_store_rocksdb:open(DocStorePath, StoreOpts) of
{ok, StoreRef} ->
%% Open attachment store (separate RocksDB with BlobDB)
AttStorePath = filename:join(DbPath, "attachments"),
AttOpts = add_env_opt(maps:get(att_opts, Config, #{}), Env),
case barrel_att_store:open(AttStorePath, AttOpts) of
{ok, AttRef} ->
%% Register in persistent_term for lookup
%% barrel_store is used by barrel_doc_body_store for body CF access
%% and by compaction filter handler for body deletion.
%% The keyspace installs BEFORE the store ref so no
%% reader ever sees a store without its keyspace.
case Keyspace of
Name -> ok;
_ -> ok = barrel_keyspace:install(Name, Keyspace)
end,
persistent_term:put({barrel_db, Name}, self()),
persistent_term:put({barrel_store, Name}, StoreRef),
persistent_term:put({barrel_source, Name},
ensure_source_id(StoreRef, Name)),
ok = barrel_channel:install(Name, CompiledChannels),
%% Compaction settings from config (or defaults)
CompactionInterval = maps:get(compaction_interval, Config,
?DEFAULT_COMPACTION_INTERVAL),
CompactionThreshold = maps:get(compaction_size_threshold, Config,
?DEFAULT_COMPACTION_SIZE_THRESHOLD),
%% Start periodic compaction check timer
TimerRef = erlang:send_after(CompactionInterval, self(), compaction_check),
%% Retention settings; the sweeper only runs for a
%% finite window
RetentionPeriod = maps:get(retention_period, Config,
?DEFAULT_RETENTION_PERIOD),
RetentionInterval = maps:get(retention_check_interval, Config,
?DEFAULT_RETENTION_INTERVAL),
RetentionTimer = case RetentionPeriod of
0 -> undefined;
_ -> erlang:send_after(RetentionInterval, self(),
retention_sweep)
end,
%% Doc TTL sweeper (opt-in): turns lazily expired
%% docs into real tombstones
TtlInterval = maps:get(ttl_sweep_interval, Config, 0),
TtlBatch = maps:get(ttl_sweep_batch, Config, 512),
TtlTimer = case TtlInterval of
0 -> undefined;
_ -> erlang:send_after(TtlInterval, self(),
ttl_sweep)
end,
logger:info("Database ~s started at ~s", [Name, DbPath]),
{ok, #state{
name = Name,
keyspace = Keyspace,
parent = Parent,
fork_hlc = ForkHlc,
config = Config,
db_path = DbPath,
store_ref = StoreRef,
att_ref = AttRef,
env = Env,
filter_pid = FilterPid,
compaction_timer = TimerRef,
compaction_interval = CompactionInterval,
compaction_size_threshold = CompactionThreshold,
retention_period = RetentionPeriod,
retention_timer = RetentionTimer,
retention_interval = RetentionInterval,
ttl_sweep_interval = TtlInterval,
ttl_sweep_batch = TtlBatch,
ttl_timer = TtlTimer
}};
{error, AttReason} ->
%% Close document store if attachment store fails
barrel_store_rocksdb:close(StoreRef),
exit(FilterPid, shutdown),
{stop, {att_store_open_failed, AttReason}}
end;
{error, Reason} ->
exit(FilterPid, shutdown),
{stop, {store_open_failed, Reason}}
end.
%% @doc Handle synchronous calls
handle_call(info, _From, #state{name = Name, keyspace = Keyspace,
parent = Parent, fork_hlc = ForkHlc,
config = Config, db_path = DbPath,
store_ref = StoreRef, att_ref = AttRef,
retention_period = RetentionPeriod} = State) ->
AttFloor = case barrel_att_store:supports_sync(AttRef) of
true -> barrel_att_store:att_floor(AttRef, Name);
false -> undefined
end,
Info0 = #{
name => Name,
keyspace => Keyspace,
config => Config,
db_path => DbPath,
pid => self(),
retention_period => RetentionPeriod,
history_floor => barrel_history:history_floor(StoreRef, Name),
att_floor => AttFloor
},
Info = case Parent of
undefined -> Info0;
_ -> Info0#{parent => Parent, fork_hlc => ForkHlc}
end,
{reply, {ok, Info}, State};
handle_call({checkpoint_to, DocsPath, AttPath}, _From,
#state{store_ref = StoreRef, att_ref = AttRef} = State) ->
ForkHlc = barrel_hlc:new_hlc(),
Result = case barrel_store_rocksdb:checkpoint(StoreRef, DocsPath) of
ok ->
case barrel_att_store:checkpoint(AttRef, AttPath) of
ok -> {ok, ForkHlc};
{error, _} = AttErr -> AttErr
end;
{error, _} = Err ->
Err
end,
{reply, Result, State};
handle_call(get_store_ref, _From, #state{store_ref = StoreRef} = State) ->
{reply, {ok, StoreRef}, State};
handle_call(get_att_ref, _From, #state{att_ref = AttRef} = State) ->
{reply, {ok, AttRef}, State};
%% Document operations
handle_call({put_doc, Doc, Opts}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = case validate_prov_opt(Opts) of
{ok, Opts1} -> do_put_doc(StoreRef, DbName, Doc, Opts1);
{error, _} = PErr -> PErr
end,
{reply, Result, State};
handle_call({put_docs, Docs, Opts}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = case validate_prov_opt(Opts) of
{ok, Opts1} -> do_put_docs(StoreRef, DbName, Docs, Opts1);
{error, _} = PErr -> PErr
end,
{reply, Result, State};
handle_call({get_doc, DocId, Opts}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = do_get_doc(StoreRef, DbName, DocId, Opts),
{reply, Result, State};
handle_call({get_docs, DocIds, Opts}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = do_get_docs(StoreRef, DbName, DocIds, Opts),
{reply, Result, State};
handle_call({delete_doc, DocId, Opts}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = case validate_prov_opt(Opts) of
{ok, Opts1} -> do_delete_doc(StoreRef, DbName, DocId, Opts1);
{error, _} = PErr -> PErr
end,
{reply, Result, State};
handle_call({fold_docs, Fun, Acc}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = do_fold_docs(StoreRef, DbName, Fun, Acc, #{}),
{reply, Result, State};
handle_call({fold_docs, Fun, Acc, Opts}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = do_fold_docs(StoreRef, DbName, Fun, Acc, Opts),
{reply, Result, State};
%% Embedding column write-back
handle_call({set_doc_embedding, DocId, ExpectedRev, Vector}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = do_set_doc_embedding(StoreRef, DbName, DocId, ExpectedRev, Vector),
{reply, Result, State};
%% Retention sweep (manual trigger; also runs on the timer)
handle_call(sweep_retention, _From,
#state{name = DbName, store_ref = StoreRef, att_ref = AttRef,
retention_period = RetentionPeriod} = State) ->
Result = case RetentionPeriod of
0 -> {ok, #{retention => infinite}};
_ -> do_retention_sweep(StoreRef, AttRef, DbName, RetentionPeriod)
end,
{reply, Result, State};
%% Doc TTL sweep (manual trigger; also runs on the timer)
handle_call(ttl_sweep, _From,
#state{name = DbName, store_ref = StoreRef,
ttl_sweep_batch = Batch} = State) ->
{reply, do_ttl_sweep(StoreRef, DbName, Batch), State};
%% Tagged outbox operations
handle_call({outbox_ack, Tag, Hlcs}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Ks = barrel_keyspace:resolve(DbName),
Result = case [{delete, barrel_store_keys:outbox_key(Ks, Tag, Hlc)}
|| Hlc <- Hlcs] of
[] -> ok;
Ops -> barrel_store_rocksdb:write_batch(StoreRef, Ops, #{})
end,
{reply, Result, State};
%% Conflict operations
handle_call({get_conflicts, DocId}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = do_get_conflicts(StoreRef, DbName, DocId),
{reply, Result, State};
handle_call({resolve_conflict, DocId, BaseRev, Resolution}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = do_resolve_conflict(StoreRef, DbName, DocId, BaseRev, Resolution),
{reply, Result, State};
%% Replication operations
handle_call({put_version, Doc, VersionToken, VVBin, Deleted}, _From,
#state{name = DbName, store_ref = StoreRef,
config = Config} = State) ->
Result = try
do_put_version(StoreRef, DbName, Config, Doc,
barrel_version:from_token(VersionToken),
barrel_vv:decode(VVBin), Deleted)
catch
Class:Reason ->
{error, {invalid_version, {Class, Reason}}}
end,
{reply, Result, State};
handle_call({diff_versions, TokenMap}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = try
do_diff_versions(StoreRef, DbName, TokenMap)
catch
Class:Reason ->
{error, {invalid_version, {Class, Reason}}}
end,
{reply, Result, State};
%% Local document operations
handle_call({put_local_doc, DocId, Doc}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = do_put_local_doc(StoreRef, DbName, DocId, Doc),
{reply, Result, State};
handle_call({get_local_doc, DocId}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = do_get_local_doc(StoreRef, DbName, DocId),
{reply, Result, State};
handle_call({delete_local_doc, DocId}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = do_delete_local_doc(StoreRef, DbName, DocId),
{reply, Result, State};
handle_call({fold_local_docs, Prefix, Fun, Acc}, _From,
#state{name = DbName, store_ref = StoreRef} = State) ->
Result = do_fold_local_docs(StoreRef, DbName, Prefix, Fun, Acc),
{reply, Result, State};
handle_call(_Request, _From, State) ->
{reply, {error, unknown_request}, State}.
%% @doc Handle asynchronous casts
handle_cast(_Msg, State) ->
{noreply, State}.
%% @doc Handle other messages
handle_info(compaction_check, #state{store_ref = StoreRef,
compaction_interval = Interval,
compaction_size_threshold = Threshold} = State) ->
%% Check database size and trigger compaction if needed
NewState = do_check_compaction(StoreRef, Threshold, State),
%% Restart timer for next check
TimerRef = erlang:send_after(Interval, self(), compaction_check),
{noreply, NewState#state{compaction_timer = TimerRef}};
handle_info(retention_sweep, #state{name = DbName, store_ref = StoreRef,
att_ref = AttRef,
retention_period = RetentionPeriod,
retention_interval = Interval} = State) ->
{ok, _Stats} = do_retention_sweep(StoreRef, AttRef, DbName,
RetentionPeriod),
TimerRef = erlang:send_after(Interval, self(), retention_sweep),
{noreply, State#state{retention_timer = TimerRef}};
handle_info(ttl_sweep, #state{name = DbName, store_ref = StoreRef,
ttl_sweep_interval = Interval,
ttl_sweep_batch = Batch} = State) ->
{ok, Swept} = do_ttl_sweep(StoreRef, DbName, Batch),
%% a full batch means more work is waiting: come back quickly
NextIn = case Swept >= Batch of
true -> 100;
false -> Interval
end,
TimerRef = erlang:send_after(NextIn, self(), ttl_sweep),
{noreply, State#state{ttl_timer = TimerRef}};
handle_info(_Info, State) ->
{noreply, State}.
%% @doc Clean up when terminating
terminate(_Reason, #state{name = Name, store_ref = StoreRef, att_ref = AttRef,
filter_pid = FilterPid,
compaction_timer = CompactionTimer,
retention_timer = RetentionTimer}) ->
%% Cancel compaction timer
_ =
case CompactionTimer of
undefined -> ok;
_ -> erlang:cancel_timer(CompactionTimer)
end,
%% Cancel retention timer
_ =
case RetentionTimer of
undefined -> ok;
_ -> erlang:cancel_timer(RetentionTimer)
end,
%% Stop compaction filter handler
_ =
case FilterPid of
undefined -> ok;
_ ->
try exit(FilterPid, shutdown)
catch _:_ -> ok
end
end,
%% Close attachment store
_ =
case AttRef of
undefined -> ok;
_ -> barrel_att_store:close(AttRef)
end,
%% Unregister BEFORE closing the store so caller-side readers
%% (barrel_docdb_reader, barrel_doc_body_store) stop picking up a ref
%% that is about to be closed. In-flight readers that already hold the
%% ref are handled by the reader's badarg guard.
persistent_term:erase({barrel_db, Name}),
persistent_term:erase({barrel_store, Name}),
persistent_term:erase({barrel_source, Name}),
ok = barrel_keyspace:uninstall(Name),
ok = barrel_channel:uninstall(Name),
%% Close document store (includes body CF)
_ =
case StoreRef of
undefined -> ok;
_ -> barrel_store_rocksdb:close(StoreRef)
end,
logger:info("Database ~s stopped", [Name]),
ok.
%%====================================================================
%% Internal functions
%%====================================================================
%% @private Resolve the encryption spec against the CRYPTO marker file
%% (a cleartext key-check token next to TIMELINE) and build the
%% EncryptedEnv. Fail-closed matrix: wrong key -> wrong_encryption_key,
%% plaintext open of an encrypted db -> db_is_encrypted, encrypted open
%% of an existing plaintext db -> cannot_encrypt_existing_db.
init_encryption(Config, Keyspace, DbPath) ->
Spec = maps:get(encryption, Config, disabled),
case barrel_keyprovider:key_for_db(Spec, Keyspace) of
{ok, plaintext} ->
case filelib:is_regular(crypto_marker_path(DbPath)) of
true -> {error, db_is_encrypted};
false -> {ok, undefined}
end;
{ok, Key} ->
check_crypto_marker(DbPath, Key);
{error, Reason} ->
{error, {encryption_key_error, Reason}}
end.
check_crypto_marker(DbPath, Key) ->
MarkerPath = crypto_marker_path(DbPath),
case file:read_file(MarkerPath) of
{ok, Token} ->
case barrel_crypto:key_check_verify(Key, Token) of
true -> new_encrypted_env(Key);
false -> {error, wrong_encryption_key}
end;
{error, enoent} ->
case filelib:is_dir(filename:join(DbPath, "docs")) of
true ->
{error, cannot_encrypt_existing_db};
false ->
case write_crypto_marker(MarkerPath, Key) of
ok ->
new_encrypted_env(Key);
{error, Reason} ->
{error, {crypto_marker_write_failed, Reason}}
end
end;
{error, Reason} ->
{error, {crypto_marker_read_failed, Reason}}
end.
%% rocksdb:new_env/1 cannot fail per its contract
new_encrypted_env(Key) ->
rocksdb:new_env({encrypted, Key}).
%% Written atomically (temp + rename) once, at the first encrypted open.
write_crypto_marker(MarkerPath, Key) ->
Token = barrel_crypto:key_check_new(Key),
ok = filelib:ensure_dir(MarkerPath),
Tmp = MarkerPath ++ ".tmp",
case file:write_file(Tmp, Token) of
ok -> file:rename(Tmp, MarkerPath);
{error, _} = Err -> Err
end.
crypto_marker_path(DbPath) ->
filename:join(DbPath, "CRYPTO").
add_env_opt(Opts, undefined) -> Opts;
add_env_opt(Opts, Env) -> Opts#{env => Env}.
%% @doc Check database size and trigger compaction if threshold exceeded
%% Called periodically by the compaction_check timer.
do_check_compaction(StoreRef, Threshold, State) ->
#state{name = Name} = State,
case barrel_store_rocksdb:get_db_size(StoreRef) of
{ok, Size} when Size >= Threshold ->
logger:info("Database ~s size ~p bytes exceeds threshold ~p, triggering compaction",
[Name, Size, Threshold]),
case barrel_store_rocksdb:compact_default_cf(StoreRef) of
ok ->
logger:info("Database ~s compaction completed", [Name]),
State;
{error, Reason} ->
logger:warning("Database ~s compaction failed: ~p", [Name, Reason]),
State
end;
{ok, Size} ->
logger:debug("Database ~s size ~p bytes below threshold ~p, skipping compaction",
[Name, Size, Threshold]),
State;
{error, Reason} ->
logger:warning("Failed to get database ~s size: ~p", [Name, Reason]),
State
end.
%% @private Release the store handle and registrations leaked by a
%% crashed instance. A live owner keeps everything (the open below then
%% fails on the RocksDB lock, which is the correct answer for a
%% double open).
release_stale_refs(Name) ->
Owner = persistent_term:get({barrel_db, Name}, undefined),
OwnerAlive = is_pid(Owner) andalso is_process_alive(Owner),
case persistent_term:get({barrel_store, Name}, undefined) of
undefined ->
ok;
_StoreRef when OwnerAlive ->
ok;
StoreRef ->
logger:warning(
"Database ~s: releasing store handle leaked by a crashed "
"instance", [Name]),
_ = try barrel_store_rocksdb:close(StoreRef)
catch _:_ -> ok
end,
persistent_term:erase({barrel_store, Name}),
persistent_term:erase({barrel_db, Name}),
persistent_term:erase({barrel_source, Name}),
ok = barrel_keyspace:uninstall(Name),
ok = barrel_channel:uninstall(Name),
ok
end.
%%====================================================================
%% Retention Sweep
%%====================================================================
%% @doc Prune everything older than the retention window and advance
%% the history floor. Walks the retained history up to the cutoff; each
%% swept entry drops its history row, plus:
%% - the superseded chain row and archived body of a version that is no
%% longer current (live conflict siblings are kept until resolved)
%% - the whole document when the entry is the current version of an
%% unconflicted tombstone (full forget: entity, bodies, chain, feed
%% row)
%% One pass of the doc TTL sweeper: fold the expiry index up to now,
%% re-check each doc's entity (a rewrite may have bumped or cleared the
%% TTL, leaving a stale row), tombstone the ones still due through the
%% normal delete path (feed, channels, and replication all correct),
%% and drop stale rows.
do_ttl_sweep(StoreRef, DbName, Batch) ->
Ks = barrel_keyspace:resolve(DbName),
NowMs = erlang:system_time(millisecond),
Start = barrel_store_keys:doc_expiry_prefix(Ks),
End = barrel_store_keys:doc_expiry_upto(Ks, NowMs),
Fold = fun(Key, _Value, Acc) ->
case length(Acc) >= Batch of
true -> {stop, Acc};
false -> {ok, [Key | Acc]}
end
end,
Keys = lists:reverse(
barrel_store_rocksdb:fold_range(StoreRef, Start, End, Fold, [])),
Swept = lists:foldl(
fun(Key, N) ->
{ExpiresMs, DocId} =
barrel_store_keys:decode_doc_expiry_key(Ks, Key),
Due = case read_current(StoreRef, DbName, DocId) of
#{expires_at := E} when E =:= ExpiresMs -> due;
undefined -> stale;
_Bumped -> stale
end,
case Due of
due ->
case do_delete_doc(StoreRef, DbName, DocId, #{}) of
{ok, _} -> N + 1;
{error, not_found} ->
drop_expiry_row(StoreRef, Key),
N
end;
stale ->
drop_expiry_row(StoreRef, Key),
N
end
end, 0, Keys),
{ok, Swept}.
drop_expiry_row(StoreRef, Key) ->
ok = barrel_store_rocksdb:write_batch(StoreRef, [{delete, Key}], #{}).
%% All deletes and the floor advance commit in one batch; re-running a
%% sweep is a no-op (deletes of missing keys do nothing).
do_retention_sweep(StoreRef, AttRef, DbName, RetentionPeriod) ->
Ks = barrel_keyspace:resolve(DbName),
CutoffMs = erlang:system_time(millisecond) - RetentionPeriod * 1000,
Cutoff = barrel_hlc:from_wall_time(CutoffMs),
%% Attachment feed tombstones follow the same window (separate
%% RocksDB; the two floors are independent)
AttSwept = case barrel_att_store:supports_sync(AttRef) of
true ->
{ok, #{tombstones_swept := N}} =
barrel_att_store:sweep_att_feed(AttRef, DbName, Cutoff),
N;
false ->
0
end,
Zero = #{history_swept => 0, superseded_removed => 0,
docs_forgotten => 0},
{Ops, Stats} = barrel_history:fold(
StoreRef, DbName,
fun(Entry, {OpsAcc, StatsAcc}) ->
{EntryOps, StatsAcc2} =
sweep_entry_ops(StoreRef, DbName, Entry, StatsAcc),
{ok, {EntryOps ++ OpsAcc, StatsAcc2}}
end,
{[], Zero},
#{to => Cutoff}),
%% Channel leave rows are events, not state: age them out
{LeaveOps, LeavesSwept} = sweep_channel_leaves(StoreRef, DbName, Cutoff),
FloorOp = {put, barrel_store_keys:db_meta(Ks, <<"history_floor">>),
barrel_hlc:encode(Cutoff)},
ok = barrel_store_rocksdb:write_batch(
StoreRef, [FloorOp | Ops ++ LeaveOps], #{}),
{ok, Stats#{floor => Cutoff, channel_leaves_swept => LeavesSwept,
att_tombstones_swept => AttSwept}}.
%% @private Delete leave rows older than the cutoff in every configured
%% channel (member rows are live state and are never age-swept).
sweep_channel_leaves(StoreRef, DbName, Cutoff) ->
Ks = barrel_keyspace:resolve(DbName),
lists:foldl(
fun(Channel, {OpsAcc, CountAcc}) ->
Start = barrel_store_keys:channel_prefix(Ks, Channel),
End = barrel_store_keys:channel_key(Ks, Channel, Cutoff),
barrel_store_rocksdb:fold_range(
StoreRef, Start, End,
fun(Key, Value, {InnerOps, InnerCount}) ->
case barrel_channel:decode_row(Value) of
#{flag := leave} ->
{ok, {[{delete, Key} | InnerOps],
InnerCount + 1}};
_ ->
{ok, {InnerOps, InnerCount}}
end
end,
{OpsAcc, CountAcc})
end,
{[], 0},
barrel_channel:names(DbName)).
%% @private Ops for one expired history entry.
sweep_entry_ops(StoreRef, DbName,
#{hlc := Hlc, id := DocId, version := Token}, Stats) ->
Ks = barrel_keyspace:resolve(DbName),
HistDel = {delete, barrel_store_keys:history_key(Ks, Hlc)},
VersionEnc = barrel_version:encode(barrel_version:from_token(Token)),
EntityKey = barrel_store_keys:doc_entity(Ks, DocId),
case barrel_store_rocksdb:get_entity(StoreRef, EntityKey) of
not_found ->
%% Doc already fully forgotten
{[HistDel], inc_sweep(history_swept, Stats)};
{ok, Columns} ->
CurrentEnc = proplists:get_value(?COL_VERSION, Columns),
Deleted = proplists:get_value(?COL_DELETED, Columns, <<"false">>)
=:= <<"true">>,
NConflicts = proplists:get_value(?COL_NCONFLICTS, Columns, 0),
case CurrentEnc of
VersionEnc when Deleted andalso NConflicts =:= 0 ->
%% Expired tombstone: the database forgets the doc
{forget_doc_ops(StoreRef, DbName, DocId, Columns)
++ [HistDel],
inc_sweep(docs_forgotten, Stats)};
VersionEnc ->
%% Still current (live doc, or tombstone with
%% unresolved conflicts): state is retained
{[HistDel], inc_sweep(history_swept, Stats)};
_Other ->
sweep_old_version_ops(StoreRef, DbName, DocId,
VersionEnc, HistDel, Stats)
end
end.
%% @private A version that is no longer current: drop its superseded
%% chain row and archived body. Live conflict siblings stay until
%% resolved.
sweep_old_version_ops(StoreRef, DbName, DocId, VersionEnc, HistDel, Stats) ->
Ks = barrel_keyspace:resolve(DbName),
ChainKey = barrel_store_keys:doc_version(Ks, DocId, VersionEnc),
case barrel_store_rocksdb:get(StoreRef, ChainKey) of
{ok, <<1:8, _/binary>>} ->
%% Live conflict sibling
{[HistDel], inc_sweep(history_swept, Stats)};
{ok, _Superseded} ->
BodyKey = barrel_store_keys:doc_body_rev(Ks, DocId, VersionEnc),
{[HistDel, {delete, ChainKey}, {body_delete, BodyKey}],
inc_sweep(superseded_removed, Stats)};
not_found ->
{[HistDel], inc_sweep(history_swept, Stats)}
end.
%% @private Full forget of an expired tombstone: entity, current body,
%% every chain row and archived body, and the live-feed row. The path
%% index rows were already removed when the doc was deleted.
forget_doc_ops(StoreRef, DbName, DocId, Columns) ->
Ks = barrel_keyspace:resolve(DbName),
ChangeHlc = barrel_hlc:decode(proplists:get_value(?COL_HLC, Columns)),
Start = barrel_store_keys:doc_version_prefix(Ks, DocId),
End = barrel_store_keys:doc_version_end(Ks, DocId),
ChainOps = barrel_store_rocksdb:fold_range(
StoreRef, Start, End,
fun(Key, _Value, Acc) ->
VEnc = barrel_store_keys:decode_doc_version_key(Ks, DocId, Key),
{ok, [{delete, Key},
{body_delete,
barrel_store_keys:doc_body_rev(Ks, DocId, VEnc)}
| Acc]}
end,
[]),
%% Channel rows live exactly at the doc's current change HLC (the
%% writer invariant), so the forget covers them too
ChannelDels = [{delete,
barrel_store_keys:channel_key(Ks, C, ChangeHlc)}
|| C <- barrel_channel:names(DbName)],
[{entity_delete, barrel_store_keys:doc_entity(Ks, DocId)},
{body_delete, barrel_store_keys:doc_body(Ks, DocId)},
{delete, barrel_store_keys:doc_hlc(Ks, ChangeHlc)}
| ChannelDels ++ ChainOps].
inc_sweep(Key, Stats) ->
maps:update_with(Key, fun(N) -> N + 1 end, Stats).
%%====================================================================
%% Document Operations
%%====================================================================
%% Wide column names for the document entity are shared with the caller-side
%% reader; they live in barrel_docdb.hrl (?COL_VERSION, ?COL_VV, ...).
%% @doc Put a document (create or update)
%% Accepts: Erlang map, indexed CBOR binary, or plain CBOR binary
%% Options:
%% - sync: boolean() - if true, sync to disk before returning (default: false)
do_put_doc(StoreRef, DbName, Doc, Opts) ->
%% Normalize input: map, indexed binary, or plain CBOR -> map for processing
DocMap = barrel_doc:to_map(Doc),
DocRecord = barrel_doc:make_doc_record(DocMap),
Old = read_current(StoreRef, DbName, maps:get(id, DocRecord)),
case cas_check(Old, maps:get(expected_version, DocRecord)) of
ok ->
{AllOps, {DocId, NewToken, NextHlc, Deleted, DocBody}} =
build_write_ops(StoreRef, DbName, DocRecord, Old, Opts),
Sync = maps:get(sync, Opts, false),
ok = barrel_store_rocksdb:write_batch(StoreRef, AllOps, #{sync => Sync}),
notify_subscribers(DbName, DocId, NewToken, NextHlc, Deleted, DocBody),
%% return_hlc => true adds the write's change HLC (internal atom
%% key), used by callers that ack outbox entries.
Result = #{
<<"id">> => DocId,
<<"ok">> => true,
<<"rev">> => NewToken
},
case maps:get(return_hlc, Opts, false) of
true -> {ok, Result#{hlc => NextHlc}};
false -> {ok, Result}
end;
{error, conflict} ->
{error, conflict}
end.
%% @doc Embedding entity columns for a write. Present only when the doc
%% carried an _embedding; absent otherwise, which clears any previous
%% (now stale) vector column since entity_put replaces the entity.
embedding_columns(DocRecord) ->
case maps:get(embedding, DocRecord, undefined) of
undefined ->
[];
Vector ->
[{?COL_EMBEDDING, barrel_doc:encode_embedding(Vector)},
{?COL_EMBEDDING_SRC, maps:get(embedding_src, DocRecord, ?EMBEDDING_SRC_CLIENT)}]
end.
%% @doc Provenance entity column for a write. Present only when the
%% caller supplied the provenance option; absent otherwise, which
%% clears any previous value since entity_put replaces the entity.
provenance_columns(undefined) -> [];
provenance_columns(ProvEnc) -> [{?COL_PROVENANCE, ProvEnc}].
%% @doc Validate the provenance and expires_at write options (if any);
%% caches the encoded provenance for the op builders.
validate_prov_opt(Opts) ->
case maps:find(expires_at, Opts) of
{ok, E} when not (is_integer(E) andalso E >= 0) ->
{error, {invalid_expires_at, E}};
_ ->
case maps:find(provenance, Opts) of
error ->
{ok, Opts};
{ok, Prov} ->
case barrel_provenance:validate(Prov) of
{ok, Enc} ->
{ok, Opts#{provenance_enc => Enc}};
{error, Reason} ->
{error, {invalid_provenance, Reason}}
end
end
end.
%% @doc TTL index maintenance for one write: the row moves with the
%% doc's expires_at so the sweeper folds exactly the expired range.
expiry_index_ops(_Ks, _DocId, Same, Same) ->
[];
expiry_index_ops(Ks, DocId, 0, New) ->
[{put, barrel_store_keys:doc_expiry(Ks, New, DocId), <<>>}];
expiry_index_ops(Ks, DocId, Old, 0) ->
[{delete, barrel_store_keys:doc_expiry(Ks, Old, DocId)}];
expiry_index_ops(Ks, DocId, Old, New) ->
[{delete, barrel_store_keys:doc_expiry(Ks, Old, DocId)},
{put, barrel_store_keys:doc_expiry(Ks, New, DocId), <<>>}].
%% @doc Read a document's current version state (undefined when absent).
read_current(StoreRef, DbName, DocId) ->
Ks = barrel_keyspace:resolve(DbName),
DocEntityKey = barrel_store_keys:doc_entity(Ks, DocId),
case barrel_store_rocksdb:get_entity(StoreRef, DocEntityKey) of
{ok, Columns} ->
VV = case proplists:get_value(?COL_VV, Columns, <<>>) of
<<>> -> barrel_vv:new();
VVBin -> barrel_vv:decode(VVBin)
end,
{Body, BodyCbor} = case barrel_store_rocksdb:body_get(StoreRef,
barrel_store_keys:doc_body(Ks, DocId)) of
{ok, OldCborBin} ->
{barrel_docdb_codec_cbor:decode_any(OldCborBin), OldCborBin};
not_found ->
{undefined, undefined}
end,
#{
version => barrel_version:decode(
proplists:get_value(?COL_VERSION, Columns)),
hlc => barrel_hlc:decode(proplists:get_value(?COL_HLC, Columns)),
vv => VV,
deleted => bin_to_deleted(
proplists:get_value(?COL_DELETED, Columns, <<"false">>)),
num_conflicts => proplists:get_value(?COL_NCONFLICTS, Columns, 0),
created_at => proplists:get_value(?COL_CREATED_AT, Columns, <<>>),
expires_at => proplists:get_value(?COL_EXPIRES_AT, Columns, 0),
tier => proplists:get_value(?COL_TIER, Columns, 0),
body => Body,
body_cbor => BodyCbor
};
not_found ->
undefined
end.
%% @doc This database's stable source id (the version author).
%% Per-database, not per-node: replicas on the same node must detect
%% each other's divergent writes as concurrent.
source_id(DbName) ->
persistent_term:get({barrel_source, DbName}).
%% @private Load or create the source id (persisted in db meta).
ensure_source_id(StoreRef, LogicalName) ->
%% Deliberately keyed by the LOGICAL name, never the keyspace: a
%% branch checkpoint carries only the parent-keyed source_id, so
%% this lookup misses on a fresh branch and mints a new author.
Key = barrel_store_keys:db_meta(LogicalName, <<"source_id">>),
case barrel_store_rocksdb:get(StoreRef, Key) of
{ok, SourceId} ->
SourceId;
not_found ->
SourceId = binary:encode_hex(crypto:strong_rand_bytes(8), lowercase),
ok = barrel_store_rocksdb:put(StoreRef, Key, SourceId),
SourceId
end.
%% @doc CAS check for local writes: the caller must supply the current
%% winner's version token when updating a live document. Creating a new
%% document, or recreating over a tombstone, needs no token.
cas_check(undefined, _Expected) ->
ok;
cas_check(#{deleted := true}, undefined) ->
ok;
cas_check(_Old, undefined) ->
{error, conflict};
cas_check(#{version := Current}, ExpectedToken) ->
case barrel_version:to_token(Current) =:= ExpectedToken of
true -> ok;
false -> {error, conflict}
end.
%% @doc Sibling entry stored in the per-doc version chain (0x1D):
%% `<<Flag:8, Deleted:8, VV/binary>>' where Flag 0 = superseded (covered
%% by the current winner) and 1 = conflict (a live concurrent sibling).
sibling_entry(Flag, Deleted, VV) ->
FlagByte = case Flag of superseded -> 0; conflict -> 1 end,
DelByte = case Deleted of true -> 1; false -> 0 end,
<<FlagByte:8, DelByte:8, (barrel_vv:encode(VV))/binary>>.
decode_sibling_entry(<<FlagByte:8, DelByte:8, VVBin/binary>>) ->
Flag = case FlagByte of 1 -> conflict; _ -> superseded end,
#{flag => Flag, deleted => DelByte =:= 1, vv => barrel_vv:decode(VVBin)}.
%% @doc Build the atomic op list for a local write (create or update).
%% Issues the version: for local writes the version HLC and the change
%% sequence HLC are the same fresh HLC. The superseded old winner goes to
%% the version chain and its body to the version-keyed archive.
%% Returns {AllOps, NotifyInfo} with NotifyInfo =
%% {DocId, NewToken, NextHlc, Deleted, DocBody}.
%% @private Channel feed ops for an applied write. Topics are derived
%% from the bodies only when channels are configured; tombstone bodies
%% have no topics.
channel_ops(DbName, NewHlc, DocInfo, DocBody, OldHlc, OldDocBody,
OldDeleted) ->
case barrel_channel:channels(DbName) of
Empty when map_size(Empty) =:= 0 ->
[];
_ ->
NewTopics = case maps:get(deleted, DocInfo) of
true -> [];
false -> barrel_channel:topics(DocBody)
end,
OldTopics = case OldDeleted of
true -> [];
false -> barrel_channel:topics(OldDocBody)
end,
barrel_channel:write_ops(DbName, NewHlc, DocInfo, NewTopics,
OldHlc, OldTopics)
end.
build_write_ops(StoreRef, DbName, DocRecord, Old, Opts) ->
Ks = barrel_keyspace:resolve(DbName),
#{id := DocId, deleted := Deleted, doc := DocBody} = DocRecord,
DocEntityKey = barrel_store_keys:doc_entity(Ks, DocId),
NextHlc = barrel_hlc:new_hlc(),
NewVersion = barrel_version:new(NextHlc, source_id(DbName)),
NewToken = barrel_version:to_token(NewVersion),
{OldVersion, OldHlc, OldVV, OldDeleted, OldDocBody, OldDocBodyCbor,
NConflicts, CreatedAt, ExpiresAt, Tier} =
case Old of
undefined ->
{undefined, undefined, barrel_vv:new(), false, undefined,
undefined, 0, barrel_hlc:encode(NextHlc), 0, 0};
#{version := OV, hlc := OH, vv := OVV, deleted := OD,
body := OB, body_cbor := OBC, num_conflicts := ONC,
created_at := OCA, expires_at := OEA, tier := OT} ->
{OV, OH, OVV, OD, OB, OBC, ONC, OCA, OEA, OT}
end,
NewVV = barrel_vv:bump(OldVV, NewVersion),
ProvEnc = maps:get(provenance_enc, Opts, undefined),
%% Doc TTL: absent = preserve, 0 = clear, N = expire at N (unix ms)
NewExpires = maps:get(expires_at, Opts, ExpiresAt),
ExpiryOps = expiry_index_ops(Ks, DocId, ExpiresAt, NewExpires),
DocColumns = [
{?COL_VERSION, barrel_version:encode(NewVersion)},
{?COL_DELETED, deleted_to_bin(Deleted)},
{?COL_HLC, barrel_hlc:encode(NextHlc)},
{?COL_VV, barrel_vv:encode(NewVV)},
{?COL_NCONFLICTS, NConflicts},
{?COL_CREATED_AT, CreatedAt},
{?COL_EXPIRES_AT, NewExpires},
{?COL_TIER, Tier}
] ++ provenance_columns(ProvEnc) ++ embedding_columns(DocRecord),
DocOps = [{entity_put, DocEntityKey, DocColumns}],
%% Change tracking info (rev carries the opaque version token)
DocInfo = #{
id => DocId,
rev => NewToken,
deleted => Deleted,
num_conflicts => NConflicts,
hlc => NextHlc
},
%% Supersede the live-feed row of the previous write
HlcDeleteOps = case OldHlc of
undefined -> [];
_ -> [{delete, barrel_store_keys:doc_hlc(Ks, OldHlc)}]
end,
%% Path index operations (if not deleted)
PathIndexOps = case Deleted of
true when OldDocBody =/= undefined ->
case barrel_ars_index:get_doc_paths(StoreRef, DbName, DocId) of
{ok, OldPaths} -> barrel_ars_index:remove_doc_ops(DbName, DocId, OldPaths);
not_found -> []
end;
true ->
[];
false when OldDocBody =:= undefined ->
barrel_ars_index:index_doc_ops(DbName, DocId, DocBody);
false ->
case OldDeleted of
true ->
%% Re-creating a deleted document. Delete only removed the
%% path index, not the body CF, so a body diff would see no
%% change and never re-add the paths. Index from scratch.
barrel_ars_index:index_doc_ops(DbName, DocId, DocBody);
false ->
barrel_ars_index:update_doc_ops(DbName, DocId, OldDocBody, DocBody)
end
end,
%% Change entry operations
ChangeInfo = DocInfo#{doc => DocBody},
ChangeOps = barrel_changes:write_change_ops(DbName, NextHlc, ChangeInfo),
%% Path-indexed change operations (for efficient filtered queries)
PathHlcOps = case OldHlc of
undefined ->
barrel_changes:write_path_index_ops(DbName, NextHlc, ChangeInfo);
_ ->
barrel_changes:update_path_index_ops(DbName, NextHlc, ChangeInfo,
OldHlc, OldDocBody)
end,
CborBody = barrel_docdb_codec_cbor:encode_cbor(DocBody),
%% The old winner is superseded: archive its body under its version
%% and record it in the per-doc version chain
{ArchiveOps, ChainOps} = case OldVersion of
undefined ->
{[], []};
_ ->
OldVersionEnc = barrel_version:encode(OldVersion),
Archive = case OldDocBodyCbor of
undefined -> [];
OldCbor ->
[{body_put,
barrel_store_keys:doc_body_rev(Ks, DocId, OldVersionEnc),
OldCbor}]
end,
Chain = [{put,
barrel_store_keys:doc_version(Ks, DocId, OldVersionEnc),
sibling_entry(superseded, OldDeleted, OldVV)}],
{Archive, Chain}
end,
%% Tagged outbox entries (generic durable work queue; see barrel_outbox)
OutboxOps = barrel_outbox:write_ops(DbName, maps:get(outbox, Opts, []),
NextHlc, OldHlc, DocId, NewToken, Deleted),
%% Retained history entry (resolutions pass cause `resolve')
HistoryOps = barrel_history:write_ops(
DbName, NextHlc, DocId, NewVersion, Deleted,
maps:get(history_cause, Opts, local), NewVV, ProvEnc),
%% Per-channel feed rows (write-time partial sync)
ChannelOps = channel_ops(DbName, NextHlc, DocInfo, DocBody, OldHlc,
OldDocBody, OldDeleted),
%% Write new body to current location (no version in key)
BodyKey = barrel_store_keys:doc_body(Ks, DocId),
BodyOp = {body_put, BodyKey, CborBody},
AllOps = DocOps ++ HlcDeleteOps ++ PathIndexOps ++ ChangeOps ++ PathHlcOps
++ ArchiveOps ++ ChainOps ++ OutboxOps ++ HistoryOps ++ ChannelOps
++ ExpiryOps ++ [BodyOp],
{AllOps, {DocId, NewToken, NextHlc, Deleted, DocBody}}.
%% @doc Put multiple documents in a single batch (batch write)
%% Options:
%% - sync: boolean() - if true, sync to disk before returning (default: false)
do_put_docs(StoreRef, DbName, Docs, Opts) ->
Sync = maps:get(sync, Opts, false),
%% Process each document to build operations and metadata
{AllOps, Notifications} = lists:foldl(
fun(Doc, {OpsAcc, NotifyAcc}) ->
case prepare_doc_ops(StoreRef, DbName, Doc, Opts) of
{ok, Ops, NotifyInfo} ->
{OpsAcc ++ Ops, [NotifyInfo | NotifyAcc]};
{error, _Reason} = Err ->
%% Skip failed docs, include error in notifications
{OpsAcc, [{error, Err} | NotifyAcc]}
end
end,
{[], []},
Docs
),
%% Write all operations in a single batch (includes body CF writes)
case AllOps of
[] -> ok;
_ ->
WriteOpts = #{sync => Sync},
ok = barrel_store_rocksdb:write_batch(StoreRef, AllOps, WriteOpts)
end,
%% Notify subscribers and build results (reverse to maintain order)
ReturnHlc = maps:get(return_hlc, Opts, false),
Results = lists:map(
fun({error, Err}) ->
Err;
({DocId, NewToken, NextHlc, Deleted, DocBody}) ->
notify_subscribers(DbName, DocId, NewToken, NextHlc, Deleted, DocBody),
Result = #{<<"id">> => DocId, <<"ok">> => true, <<"rev">> => NewToken},
case ReturnHlc of
true -> {ok, Result#{hlc => NextHlc}};
false -> {ok, Result}
end
end,
lists:reverse(Notifications)
),
Results.
%% @doc Prepare document operations without writing.
%% Batch-level Opts carry the outbox tags applied to every doc in the batch.
%% Returns {ok, Ops, NotifyInfo} or {error, Reason}
prepare_doc_ops(StoreRef, DbName, Doc, Opts) ->
try
DocMap = barrel_doc:to_map(Doc),
DocRecord = barrel_doc:make_doc_record(DocMap),
Old = read_current(StoreRef, DbName, maps:get(id, DocRecord)),
case cas_check(Old, maps:get(expected_version, DocRecord)) of
ok -> ok;
{error, conflict} -> throw(conflict)
end,
{AllOps, NotifyInfo} =
build_write_ops(StoreRef, DbName, DocRecord, Old, Opts),
{ok, AllOps, NotifyInfo}
catch
_:Reason ->
{error, Reason}
end.
%% @doc Store a computed embedding in the doc entity (CAS on revision).
%% Read-modify-write of the entity columns is safe here: this runs in
%% the database writer, which serializes all entity writes.
do_set_doc_embedding(StoreRef, DbName, DocId, ExpectedRev, Vector) ->
Ks = barrel_keyspace:resolve(DbName),
DocEntityKey = barrel_store_keys:doc_entity(Ks, DocId),
case barrel_store_rocksdb:get_entity(StoreRef, DocEntityKey) of
{ok, Columns} ->
CurrentToken = barrel_version:to_token(
barrel_version:decode(
proplists:get_value(?COL_VERSION, Columns))),
case CurrentToken of
ExpectedRev ->
Base = [C || {K, _} = C <- Columns,
K =/= ?COL_EMBEDDING,
K =/= ?COL_EMBEDDING_SRC],
NewColumns = Base ++
[{?COL_EMBEDDING, barrel_doc:encode_embedding(Vector)},
{?COL_EMBEDDING_SRC, ?EMBEDDING_SRC_COMPUTED}],
barrel_store_rocksdb:put_entity(StoreRef, DocEntityKey,
NewColumns);
_OtherRev ->
{error, conflict}
end;
not_found ->
{error, not_found};
{error, _} = Err ->
Err
end.
%% @doc Get a document by ID. The read implementation lives in
%% barrel_docdb_reader so callers can run it outside this process; the
%% server delegates to keep a single code path.
do_get_doc(StoreRef, DbName, DocId, Opts) ->
barrel_docdb_reader:get_doc(StoreRef, DbName, DocId, Opts).
%% @doc Get multiple documents by ID (batch read). Delegates to the
%% caller-side reader, same as do_get_doc/4.
do_get_docs(StoreRef, DbName, DocIds, Opts) ->
barrel_docdb_reader:get_docs(StoreRef, DbName, DocIds, Opts).
%% @doc Delete a document: a versioned tombstone write.
%% Options:
%% - rev: binary() - expected version token (optional, for conflict detection)
%% - sync: boolean() - if true, sync to disk before returning (default: false)
do_delete_doc(StoreRef, DbName, DocId, Opts) ->
Ks = barrel_keyspace:resolve(DbName),
case read_current(StoreRef, DbName, DocId) of
undefined ->
{error, not_found};
#{version := OldVersion, hlc := OldHlc, vv := OldVV,
deleted := OldDeleted, num_conflicts := NConflicts,
created_at := CreatedAt, expires_at := ExpiresAt, tier := Tier,
body := OldDocBody, body_cbor := OldDocBodyCbor} ->
CurrentToken = barrel_version:to_token(OldVersion),
case maps:get(rev, Opts, undefined) of
undefined -> ok;
CurrentToken -> ok;
_Other -> throw({error, {conflict, CurrentToken}})
end,
NextHlc = barrel_hlc:new_hlc(),
NewVersion = barrel_version:new(NextHlc, source_id(DbName)),
NewToken = barrel_version:to_token(NewVersion),
NewVV = barrel_vv:bump(OldVV, NewVersion),
DocEntityKey = barrel_store_keys:doc_entity(Ks, DocId),
ProvEnc = maps:get(provenance_enc, Opts, undefined),
%% a tombstone has no TTL; drop any pending expiry row
ExpiryOps = expiry_index_ops(Ks, DocId, ExpiresAt, 0),
DocColumns = [
{?COL_VERSION, barrel_version:encode(NewVersion)},
{?COL_DELETED, <<"true">>},
{?COL_HLC, barrel_hlc:encode(NextHlc)},
{?COL_VV, barrel_vv:encode(NewVV)},
{?COL_NCONFLICTS, NConflicts},
{?COL_CREATED_AT, CreatedAt},
{?COL_EXPIRES_AT, 0},
{?COL_TIER, Tier}
] ++ provenance_columns(ProvEnc),
DocOps = [{entity_put, DocEntityKey, DocColumns}],
%% Supersede the live-feed row of the previous write
HlcDeleteOps = [{delete, barrel_store_keys:doc_hlc(Ks, OldHlc)}],
%% Path index removal operations
PathIndexOps = case barrel_ars_index:get_doc_paths(StoreRef, DbName, DocId) of
{ok, Paths} -> barrel_ars_index:remove_doc_ops(DbName, DocId, Paths);
not_found -> []
end,
%% Build DocInfo for change tracking
NewDocInfo = #{
id => DocId,
rev => NewToken,
deleted => true,
num_conflicts => NConflicts,
hlc => NextHlc
},
%% Change entry operations
ChangeOps = barrel_changes:write_change_ops(DbName, NextHlc, NewDocInfo),
%% Path-indexed change operations for delete
PathHlcOps = barrel_changes:update_path_index_ops(DbName, NextHlc, NewDocInfo,
OldHlc, OldDocBody),
%% The superseded winner joins the version chain, body archived
OldVersionEnc = barrel_version:encode(OldVersion),
ArchiveOps = case OldDocBodyCbor of
undefined -> [];
OldCbor ->
[{body_put,
barrel_store_keys:doc_body_rev(Ks, DocId, OldVersionEnc),
OldCbor}]
end,
ChainOps = [{put,
barrel_store_keys:doc_version(Ks, DocId, OldVersionEnc),
sibling_entry(superseded, OldDeleted, OldVV)}],
%% Tagged outbox entries (deletes replace the pending entry too)
OutboxOps = barrel_outbox:write_ops(DbName, maps:get(outbox, Opts, []),
NextHlc, OldHlc, DocId, NewToken, true),
%% Retained history entry for the tombstone
HistoryOps = barrel_history:write_ops(
DbName, NextHlc, DocId, NewVersion, true, local, NewVV,
ProvEnc),
%% The tombstone lands in the channels the doc was in
ChannelOps = channel_ops(DbName, NextHlc, NewDocInfo, undefined,
OldHlc, OldDocBody, OldDeleted),
%% Write batch atomically
AllOps = DocOps ++ HlcDeleteOps ++ PathIndexOps ++ ChangeOps ++ PathHlcOps
++ ArchiveOps ++ ChainOps ++ OutboxOps ++ HistoryOps
++ ChannelOps ++ ExpiryOps,
WriteOpts = #{sync => maps:get(sync, Opts, false)},
ok = barrel_store_rocksdb:write_batch(StoreRef, AllOps, WriteOpts),
%% Notify path subscribers
notify_subscribers(DbName, DocId, NewToken, NextHlc, true, #{}),
DeleteResult = #{<<"id">> => DocId, <<"ok">> => true,
<<"rev">> => NewToken},
case maps:get(return_hlc, Opts, false) of
true -> {ok, DeleteResult#{hlc => NextHlc}};
false -> {ok, DeleteResult}
end
end.
%% @doc Fold over all documents (using wide column entity)
%% Options:
%% - include_deleted: boolean() - include deleted documents (default: false)
%% Note: Uses regular key iteration since wide columns are stored per-key
do_fold_docs(StoreRef, DbName, Fun, Acc, Opts) ->
Ks = barrel_keyspace:resolve(DbName),
StartKey = barrel_store_keys:doc_entity_prefix(Ks),
EndKey = barrel_store_keys:doc_entity_end(Ks),
PrefixLen = byte_size(StartKey),
IncludeDeleted = maps:get(include_deleted, Opts, false),
FoldFun = fun(Key, _Value, AccIn) ->
%% Extract DocId from key (after prefix)
DocId = binary:part(Key, PrefixLen, byte_size(Key) - PrefixLen),
%% Get entity to access columns
case barrel_store_rocksdb:get_entity(StoreRef, Key) of
{ok, Columns} ->
Rev = barrel_version:to_token(
barrel_version:decode(
proplists:get_value(?COL_VERSION, Columns))),
Deleted = bin_to_deleted(proplists:get_value(?COL_DELETED, Columns, <<"false">>)),
case {Deleted andalso not IncludeDeleted,
barrel_docdb_reader:expired(Columns)} of
{true, _} ->
%% Skip deleted documents
{ok, AccIn};
{_, true} ->
%% Lazily expired (the TTL sweeper will
%% tombstone it)
{ok, AccIn};
_ ->
%% Get document body from body CF (current body, no rev in key)
BodyKey = barrel_store_keys:doc_body(Ks, DocId),
DocBody = case barrel_store_rocksdb:body_get(StoreRef, BodyKey) of
{ok, CborBin} -> barrel_docdb_codec_cbor:decode_any(CborBin);
not_found -> #{}
end,
Doc = DocBody#{
<<"id">> => DocId,
<<"_rev">> => Rev
},
case Fun(Doc, AccIn) of
{ok, AccOut} -> {ok, AccOut};
{stop, AccOut} -> {stop, AccOut};
stop -> {stop, AccIn}
end
end;
not_found ->
{ok, AccIn}
end
end,
FinalAcc = barrel_store_rocksdb:fold_range(StoreRef, StartKey, EndKey, FoldFun, Acc),
{ok, FinalAcc}.
%%====================================================================
%% Replication Operations
%%====================================================================
%% @doc Apply a replicated version (the VV protocol write).
%% The remote version and its vector are preserved; only the change
%% sequence HLC is issued locally. Outcomes:
%% - already contained: idempotent no-op
%% - remote VV dominates: fast-forward, old winner superseded
%% - concurrent: the per-db conflict_merger may resolve it with a
%% merged body; otherwise LWW by version, the loser stays live as a
%% conflict sibling (bounded by max_conflict_versions) and the
%% vectors merge
do_put_version(StoreRef, DbName, Config, Doc, Version, RemoteVV, Deleted) ->
%% Advance the local clock past the remote write so subsequent local
%% change HLCs stay strictly increasing
_ = barrel_hlc:sync_hlc(barrel_version:hlc(Version)),
DocMap = barrel_doc:to_map(Doc),
DocId = case maps:find(<<"id">>, DocMap) of
{ok, Id} -> Id;
error -> erlang:error({missing_id, DocMap})
end,
DocBody = barrel_doc:doc_without_meta(DocMap),
%% Generic tags appended to replication-applied writes (barrel
%% sets [<<"embed">>] on record-mode databases so replicated docs
%% reach the indexer; docdb stays ignorant of why).
Tags = maps:get(outbox_tags_on_replication, Config, []),
case read_current(StoreRef, DbName, DocId) of
undefined ->
CreateVV = barrel_vv:bump(RemoteVV, Version),
apply_remote_current(StoreRef, DbName, DocId, DocBody, Version,
CreateVV, CreateVV, Deleted,
undefined, 0, [], Tags);
#{version := LocalV, vv := LocalVV} = Old ->
case barrel_vv:contains(LocalVV, Version) of
true ->
%% Idempotent redelivery: we already cover it
{ok, DocId, barrel_version:to_token(LocalV)};
false ->
MergedVV = barrel_vv:merge(LocalVV,
barrel_vv:bump(RemoteVV, Version)),
case barrel_vv:compare(RemoteVV, LocalVV) of
dominates ->
%% Fast-forward: old winner superseded
%% (dominating bump == merged vector)
Chain = [chain_op(DbName, DocId, LocalV,
superseded, Old)],
apply_remote_current(StoreRef, DbName, DocId,
DocBody, Version, MergedVV,
MergedVV, Deleted, Old,
maps:get(num_conflicts, Old),
Chain, Tags);
_Concurrent ->
handle_concurrent(StoreRef, DbName, Config, DocId,
DocBody, Version, RemoteVV,
Deleted, Old, MergedVV)
end
end
end.
%% @private A concurrent remote version: the merger hook gets the first
%% say; without one (or when it declines, crashes, or a side is a
%% tombstone) the deterministic LWW conflict path applies.
handle_concurrent(StoreRef, DbName, Config, DocId, DocBody, Version, RemoteVV,
Deleted, #{version := LocalV} = Old, MergedVV) ->
Tags = maps:get(outbox_tags_on_replication, Config, []),
case try_merge(Config, DocId, Old, DocBody, Deleted) of
{merge, MergedBody} ->
apply_merged(StoreRef, DbName, DocId, MergedBody, DocBody,
Version, RemoteVV, Deleted, Old, MergedVV, Tags);
conflict ->
NConflicts0 = maps:get(num_conflicts, Old) + 1,
Max = maps:get(max_conflict_versions, Config,
?DEFAULT_MAX_CONFLICT_VERSIONS),
case barrel_version:max(LocalV, Version) of
Version ->
%% Remote wins; local winner stays live as a
%% conflict sibling
{DropOps, NConflicts} = enforce_conflict_bound(
StoreRef, DbName, DocId, LocalV, NConflicts0, Max),
Chain = [chain_op(DbName, DocId, LocalV, conflict, Old)],
apply_remote_current(StoreRef, DbName, DocId, DocBody,
Version, MergedVV,
barrel_vv:bump(RemoteVV, Version),
Deleted, Old,
NConflicts, Chain ++ DropOps, Tags);
_LocalWins ->
{DropOps, NConflicts} = enforce_conflict_bound(
StoreRef, DbName, DocId, Version, NConflicts0, Max),
keep_local_add_conflict(StoreRef, DbName, DocId, DocBody,
Version, RemoteVV, Deleted, Old,
MergedVV, NConflicts, DropOps)
end
end.
%% @private Ask the configured merger to resolve a concurrent write.
%% Only meaningful when both sides are live documents. Any non-merge
%% answer or a crash falls back to the conflict path.
try_merge(Config, DocId, #{deleted := LocalDeleted, body := LocalBody},
RemoteBody, RemoteDeleted) ->
case maps:get(conflict_merger, Config, undefined) of
undefined ->
conflict;
_ when LocalDeleted; RemoteDeleted ->
conflict;
Merger ->
try call_merger(Merger, DocId, LocalBody, RemoteBody) of
{merge, Merged} when is_map(Merged) -> {merge, Merged};
_ -> conflict
catch
Class:Reason ->
logger:warning(
"conflict_merger failed for ~p: ~p:~p, "
"falling back to conflict", [DocId, Class, Reason]),
conflict
end
end.
call_merger({M, F}, DocId, LocalBody, RemoteBody) ->
M:F(DocId, LocalBody, RemoteBody);
call_merger(Fun, DocId, LocalBody, RemoteBody) when is_function(Fun, 3) ->
Fun(DocId, LocalBody, RemoteBody).
%% @private Apply a merger resolution in one batch: the remote version
%% is archived as superseded (with its own replicated history entry),
%% and a new local resolving write whose vector covers both sides
%% becomes current. The resolving write replicates like any other.
apply_merged(StoreRef, DbName, DocId, MergedBody, RemoteBody, Version,
RemoteVV, RemoteDeleted, Old, MergedVV, Tags) ->
Ks = barrel_keyspace:resolve(DbName),
%% The remote arrival: superseded chain row, archived body, history
RemoteChangeHlc = barrel_hlc:new_hlc(),
VersionEnc = barrel_version:encode(Version),
SiblingVV = barrel_vv:bump(RemoteVV, Version),
RemoteOps = [
{put, barrel_store_keys:doc_version(Ks, DocId, VersionEnc),
sibling_entry(superseded, RemoteDeleted, SiblingVV)},
{body_put, barrel_store_keys:doc_body_rev(Ks, DocId, VersionEnc),
barrel_docdb_codec_cbor:encode_cbor(RemoteBody)}
] ++ barrel_history:write_ops(DbName, RemoteChangeHlc, DocId, Version,
RemoteDeleted, replicated, SiblingVV),
%% The resolving write goes through the ordinary local pipeline
%% with the merged vector as its base (its bump then covers both
%% sides); the old local winner is archived by build_write_ops
DocRecord0 = barrel_doc:make_doc_record(MergedBody#{<<"id">> => DocId}),
DocRecord = DocRecord0#{expected_version => undefined},
Old2 = Old#{vv := MergedVV},
%% The resolving write is a new current body: tag it so the outbox
%% consumer (e.g. the record indexer) re-processes the merged text.
{WriteOps, {DocId, NewToken, NextHlc, Deleted, NotifyBody}} =
build_write_ops(StoreRef, DbName, DocRecord, Old2,
#{history_cause => resolve, outbox => Tags}),
ok = barrel_store_rocksdb:write_batch(StoreRef, RemoteOps ++ WriteOps, #{}),
notify_subscribers(DbName, DocId, NewToken, NextHlc, Deleted, NotifyBody),
{ok, DocId, NewToken}.
%% @private Deterministically bound the live conflict siblings: when a
%% new sibling would push the count past Max, the lowest versions (the
%% new one included) are dropped with their archived bodies. Every
%% replica seeing the same sibling set drops the same versions.
enforce_conflict_bound(_StoreRef, _DbName, _DocId, _NewSibling, NConflicts,
Max) when NConflicts =< Max ->
{[], NConflicts};
enforce_conflict_bound(StoreRef, DbName, DocId, NewSibling, NConflicts, Max) ->
Ks = barrel_keyspace:resolve(DbName),
Existing = [V || {V, _Entry} <- conflict_siblings(StoreRef, DbName, DocId)],
Sorted = lists:sort(
fun(A, B) -> barrel_version:compare(A, B) =:= lt end,
[NewSibling | Existing]),
Victims = lists:sublist(Sorted, NConflicts - Max),
Ops = lists:flatmap(
fun(V) ->
VEnc = barrel_version:encode(V),
[{delete, barrel_store_keys:doc_version(Ks, DocId, VEnc)},
{body_delete,
barrel_store_keys:doc_body_rev(Ks, DocId, VEnc)}]
end,
Victims),
{Ops, Max}.
%% @private Chain-row op for a version that is no longer current.
chain_op(DbName, DocId, Version, Flag, Old) ->
Ks = barrel_keyspace:resolve(DbName),
{put,
barrel_store_keys:doc_version(Ks, DocId, barrel_version:encode(Version)),
sibling_entry(Flag, maps:get(deleted, Old), maps:get(vv, Old))}.
%% @private The remote version becomes the current winner. `VV' is
%% the entity's vector (merged); `HistVV' is the arrival's OWN causal
%% vector (its bump), recorded in history. They differ only in the
%% concurrent remote-wins case, and keeping the causal one preserves
%% the concurrency information the timeline rewind reconstructs
%% conflict windows from.
apply_remote_current(StoreRef, DbName, DocId, DocBody, Version, VV, HistVV,
Deleted, Old, NConflicts, ChainOps, Tags) ->
Ks = barrel_keyspace:resolve(DbName),
ChangeHlc = barrel_hlc:new_hlc(),
Token = barrel_version:to_token(Version),
{OldHlc, OldDocBody, OldDocBodyCbor, CreatedAt, ExpiresAt, Tier} =
case Old of
undefined ->
{undefined, undefined, undefined,
barrel_hlc:encode(ChangeHlc), 0, 0};
#{hlc := OH, body := OB, body_cbor := OBC,
created_at := OCA, expires_at := OEA, tier := OT} ->
{OH, OB, OBC, OCA, OEA, OT}
end,
DocColumns = [
{?COL_VERSION, barrel_version:encode(Version)},
{?COL_DELETED, deleted_to_bin(Deleted)},
{?COL_HLC, barrel_hlc:encode(ChangeHlc)},
{?COL_VV, barrel_vv:encode(VV)},
{?COL_NCONFLICTS, NConflicts},
{?COL_CREATED_AT, CreatedAt},
{?COL_EXPIRES_AT, ExpiresAt},
{?COL_TIER, Tier}
],
DocOps = [{entity_put, barrel_store_keys:doc_entity(Ks, DocId),
DocColumns}],
DocInfo = #{id => DocId, rev => Token, deleted => Deleted,
num_conflicts => NConflicts, hlc => ChangeHlc},
HlcDeleteOps = case OldHlc of
undefined -> [];
_ -> [{delete, barrel_store_keys:doc_hlc(Ks, OldHlc)}]
end,
PathIndexOps = case Deleted of
true when OldDocBody =/= undefined ->
case barrel_ars_index:get_doc_paths(StoreRef, DbName, DocId) of
{ok, OldPaths} ->
barrel_ars_index:remove_doc_ops(DbName, DocId, OldPaths);
not_found -> []
end;
true ->
[];
false when OldDocBody =:= undefined ->
barrel_ars_index:index_doc_ops(DbName, DocId, DocBody);
false ->
case maps:get(deleted, Old) of
true ->
barrel_ars_index:index_doc_ops(DbName, DocId, DocBody);
false ->
barrel_ars_index:update_doc_ops(DbName, DocId,
OldDocBody, DocBody)
end
end,
ChangeInfo = DocInfo#{doc => DocBody},
ChangeOps = barrel_changes:write_change_ops(DbName, ChangeHlc, ChangeInfo),
PathHlcOps = case OldHlc of
undefined ->
barrel_changes:write_path_index_ops(DbName, ChangeHlc, ChangeInfo);
_ ->
barrel_changes:update_path_index_ops(DbName, ChangeHlc, ChangeInfo,
OldHlc, OldDocBody)
end,
ArchiveOps = case {Old, OldDocBodyCbor} of
{undefined, _} -> [];
{_, undefined} -> [];
{#{version := OldVersion}, OldCbor} ->
[{body_put,
barrel_store_keys:doc_body_rev(
Ks, DocId, barrel_version:encode(OldVersion)),
OldCbor}]
end,
HistoryOps = barrel_history:write_ops(
DbName, ChangeHlc, DocId, Version, Deleted, replicated, HistVV),
OutboxOps = barrel_outbox:write_ops(DbName, Tags, ChangeHlc, OldHlc,
DocId, Token, Deleted),
%% Replicated arrivals land in channel feeds too
OldDeleted = case Old of
undefined -> false;
_ -> maps:get(deleted, Old)
end,
ChannelOps = channel_ops(DbName, ChangeHlc, DocInfo, DocBody, OldHlc,
OldDocBody, OldDeleted),
CborBody = barrel_docdb_codec_cbor:encode_cbor(DocBody),
BodyOp = {body_put, barrel_store_keys:doc_body(Ks, DocId), CborBody},
AllOps = DocOps ++ HlcDeleteOps ++ PathIndexOps ++ ChangeOps ++ PathHlcOps
++ ArchiveOps ++ ChainOps ++ OutboxOps ++ HistoryOps ++ ChannelOps
++ [BodyOp],
ok = barrel_store_rocksdb:write_batch(StoreRef, AllOps, #{}),
notify_subscribers(DbName, DocId, Token, ChangeHlc, Deleted, DocBody),
{ok, DocId, Token}.
%% @private The local winner stays; the remote version is recorded as a
%% live conflict sibling (its body archived, queryable). The doc's state
%% changed (vector, conflict count), so a change event still fires.
%% ExtraOps carry sibling-bound drops, applied after the sibling puts.
keep_local_add_conflict(StoreRef, DbName, DocId, RemoteBody, Version, RemoteVV,
RemoteDeleted, Old, MergedVV, NConflicts, ExtraOps) ->
Ks = barrel_keyspace:resolve(DbName),
#{version := LocalV, hlc := OldHlc, deleted := LocalDeleted,
body := LocalBody, created_at := CreatedAt, expires_at := ExpiresAt,
tier := Tier} = Old,
ChangeHlc = barrel_hlc:new_hlc(),
LocalToken = barrel_version:to_token(LocalV),
DocColumns = [
{?COL_VERSION, barrel_version:encode(LocalV)},
{?COL_DELETED, deleted_to_bin(LocalDeleted)},
{?COL_HLC, barrel_hlc:encode(ChangeHlc)},
{?COL_VV, barrel_vv:encode(MergedVV)},
{?COL_NCONFLICTS, NConflicts},
{?COL_CREATED_AT, CreatedAt},
{?COL_EXPIRES_AT, ExpiresAt},
{?COL_TIER, Tier}
],
DocOps = [{entity_put, barrel_store_keys:doc_entity(Ks, DocId),
DocColumns}],
DocInfo = #{id => DocId, rev => LocalToken, deleted => LocalDeleted,
num_conflicts => NConflicts, hlc => ChangeHlc},
HlcDeleteOps = [{delete, barrel_store_keys:doc_hlc(Ks, OldHlc)}],
ChangeInfo = DocInfo#{doc => LocalBody},
ChangeOps = barrel_changes:write_change_ops(DbName, ChangeHlc, ChangeInfo),
PathHlcOps = barrel_changes:update_path_index_ops(DbName, ChangeHlc,
ChangeInfo, OldHlc,
LocalBody),
%% The remote loser: conflict chain row + archived body
VersionEnc = barrel_version:encode(Version),
SiblingVV = barrel_vv:bump(RemoteVV, Version),
SiblingEntry = <<1:8, %% conflict
(case RemoteDeleted of true -> 1; false -> 0 end):8,
(barrel_vv:encode(SiblingVV))/binary>>,
ChainOps = [{put, barrel_store_keys:doc_version(Ks, DocId, VersionEnc),
SiblingEntry}],
ArchiveOps = [{body_put,
barrel_store_keys:doc_body_rev(Ks, DocId, VersionEnc),
barrel_docdb_codec_cbor:encode_cbor(RemoteBody)}],
%% History records the arrival of the losing sibling
HistoryOps = barrel_history:write_ops(
DbName, ChangeHlc, DocId, Version, RemoteDeleted, replicated,
SiblingVV),
%% The winner's feed row moves to ChangeHlc; its channel rows move
%% with it (point-read move: a tombstone winner's membership cannot
%% be recomputed from its body)
ChannelOps = barrel_channel:move_ops(
fun(Key) -> barrel_store_rocksdb:get(StoreRef, Key) end,
DbName, OldHlc, ChangeHlc, DocInfo),
AllOps = DocOps ++ HlcDeleteOps ++ ChangeOps ++ PathHlcOps
++ ChainOps ++ ArchiveOps ++ HistoryOps ++ ChannelOps ++ ExtraOps,
ok = barrel_store_rocksdb:write_batch(StoreRef, AllOps, #{}),
notify_subscribers(DbName, DocId, LocalToken, ChangeHlc, LocalDeleted,
LocalBody),
{ok, DocId, LocalToken}.
%% @doc The replication diff: which offered versions does this database
%% not cover yet? `have' means the doc's version vector contains the
%% offered version.
do_diff_versions(StoreRef, DbName, TokenMap) ->
Result = maps:map(
fun(DocId, Token) ->
Version = barrel_version:from_token(Token),
case read_current(StoreRef, DbName, DocId) of
undefined ->
missing;
#{vv := LocalVV} ->
case barrel_vv:contains(LocalVV, Version) of
true -> have;
false -> missing
end
end
end,
TokenMap),
{ok, Result}.
%% @doc Put a local document (per-database)
do_put_local_doc(StoreRef, DbName, DocId, Doc) ->
Ks = barrel_keyspace:resolve(DbName),
Key = barrel_store_keys:local_doc_key(Ks, DocId),
Value = term_to_binary(Doc),
ok = barrel_store_rocksdb:local_put(StoreRef, Key, Value),
ok.
%% @doc Get a local document (per-database)
do_get_local_doc(StoreRef, DbName, DocId) ->
Ks = barrel_keyspace:resolve(DbName),
Key = barrel_store_keys:local_doc_key(Ks, DocId),
case barrel_store_rocksdb:local_get(StoreRef, Key) of
{ok, Value} ->
{ok, binary_to_term(Value)};
not_found ->
{error, not_found}
end.
%% @doc Delete a local document (per-database)
do_delete_local_doc(StoreRef, DbName, DocId) ->
Ks = barrel_keyspace:resolve(DbName),
Key = barrel_store_keys:local_doc_key(Ks, DocId),
case barrel_store_rocksdb:local_get(StoreRef, Key) of
{ok, _} ->
ok = barrel_store_rocksdb:local_delete(StoreRef, Key),
ok;
not_found ->
{error, not_found}
end.
%% @doc Fold over local documents matching a prefix
%% The prefix is applied to the DocId portion (after DbName).
do_fold_local_docs(StoreRef, DbName, DocIdPrefix, Fun, Acc0) ->
Ks = barrel_keyspace:resolve(DbName),
%% Build full key prefix: DbName + 0 + DocIdPrefix
KeyPrefix = barrel_store_keys:local_doc_key(Ks, DocIdPrefix),
%% Fold over matching keys, decode values, extract DocId
DbNameLen = byte_size(Ks),
Result = barrel_store_rocksdb:local_fold(
StoreRef,
KeyPrefix,
fun(Key, Value, Acc) ->
%% Extract DocId from key (skip DbName + 0)
<<_:DbNameLen/binary, 0, DocId/binary>> = Key,
Doc = binary_to_term(Value),
Fun(DocId, Doc, Acc)
end,
Acc0
),
{ok, Result}.
%%====================================================================
%% Subscription Notifications
%%====================================================================
%% @doc Notify subscribers of document changes
%% Extracts paths from document, matches against subscriptions,
%% and sends notifications to matching subscribers.
notify_subscribers(DbName, DocId, Rev, Hlc, Deleted, DocBody) ->
%% Extract paths from document body
Topics = case Deleted of
true ->
%% For deleted docs, just use the doc ID as a path
[DocId];
false ->
Paths = barrel_ars:analyze(DocBody),
barrel_ars:paths_to_topics(Paths)
end,
%% Find matching path subscribers
Pids = barrel_sub:match(DbName, Topics),
%% Build notification
Notification = {barrel_change, DbName, #{
id => DocId,
rev => Rev,
hlc => Hlc,
deleted => Deleted,
paths => Topics
}},
%% Send to each path subscriber
_ = [Pid ! Notification || Pid <- Pids],
%% Notify query subscribers (only for non-deleted docs)
case Deleted of
true ->
ok;
false ->
barrel_query_sub:notify_change(DbName, DocId, Rev, DocBody)
end,
ok.
%%====================================================================
%% Wide Column Helpers
%%====================================================================
%% @doc Convert boolean to binary for wide column storage
-spec deleted_to_bin(boolean()) -> binary().
deleted_to_bin(true) -> <<"true">>;
deleted_to_bin(false) -> <<"false">>.
%% @doc Convert binary back to boolean from wide column storage
-spec bin_to_deleted(binary()) -> boolean().
bin_to_deleted(<<"true">>) -> true;
bin_to_deleted(<<"false">>) -> false;
bin_to_deleted(_) -> false.
%% Conflict decoration for reads (maybe_add_conflicts*) moved to
%% barrel_docdb_reader with the read paths.
%%====================================================================
%% Conflict Operations
%%====================================================================
%% @doc Get list of conflicting version tokens for a document.
%% Fast path: the entity carries the live sibling count, so conflict-free
%% documents never scan the version chain.
do_get_conflicts(StoreRef, DbName, DocId) ->
case read_current(StoreRef, DbName, DocId) of
undefined ->
{error, not_found};
#{num_conflicts := 0} ->
{ok, []};
#{} ->
{ok, [barrel_version:to_token(V)
|| {V, _Entry} <- conflict_siblings(StoreRef, DbName, DocId)]}
end.
%% @doc Scan the version chain for live conflict siblings.
%% Returns [{version(), sibling_entry_map()}].
conflict_siblings(StoreRef, DbName, DocId) ->
Ks = barrel_keyspace:resolve(DbName),
Start = barrel_store_keys:doc_version_prefix(Ks, DocId),
End = barrel_store_keys:doc_version_end(Ks, DocId),
Fold = fun(Key, Value, Acc) ->
Entry = decode_sibling_entry(Value),
case maps:get(flag, Entry) of
conflict ->
VersionEnc = barrel_store_keys:decode_doc_version_key(
Ks, DocId, Key),
{ok, [{barrel_version:decode(VersionEnc), Entry} | Acc]};
superseded ->
{ok, Acc}
end
end,
lists:reverse(barrel_store_rocksdb:fold_range(StoreRef, Start, End, Fold, [])).
%% @doc Resolve a conflict with a resolving local write.
%% Resolution types:
%% {choose, Token} - the chosen sibling's body becomes the new current
%% {merge, Doc} - the supplied body becomes the new current
%% Either way ONE new version is written whose version vector covers the
%% winner and every conflict sibling, and all siblings flip to
%% superseded: the resolution is an ordinary write that replicates.
do_resolve_conflict(StoreRef, DbName, DocId, BaseRev, Resolution) ->
case read_current(StoreRef, DbName, DocId) of
undefined ->
{error, not_found};
#{num_conflicts := 0} ->
{error, no_conflicts};
#{version := Current, vv := CurrentVV} = Old ->
CurrentToken = barrel_version:to_token(Current),
case BaseRev of
CurrentToken ->
Siblings = conflict_siblings(StoreRef, DbName, DocId),
apply_resolution(StoreRef, DbName, DocId, Old, CurrentVV,
Siblings, Resolution);
_ ->
{error, {conflict, CurrentToken}}
end
end.
%% @private Apply a conflict resolution as a superseding write.
apply_resolution(StoreRef, DbName, DocId, Old, CurrentVV, Siblings, Resolution) ->
Ks = barrel_keyspace:resolve(DbName),
ResolvedBody = case Resolution of
{merge, Doc} when is_map(Doc) ->
{ok, barrel_doc:doc_without_meta(Doc)};
{choose, Token} ->
CurrentToken = barrel_version:to_token(maps:get(version, Old)),
case Token of
CurrentToken ->
{ok, maps:get(body, Old, #{})};
_ ->
case lists:filter(
fun({V, _E}) ->
barrel_version:to_token(V) =:= Token
end, Siblings) of
[{ChosenVersion, _Entry}] ->
read_version_body(StoreRef, DbName, DocId,
ChosenVersion);
[] ->
{error, {unknown_version, Token}}
end
end;
_ ->
{error, {invalid_resolution, Resolution}}
end,
case ResolvedBody of
{ok, Body} ->
%% Version vector of the resolving write covers everything
MergedVV = lists:foldl(
fun({_V, #{vv := SibVV}}, Acc) -> barrel_vv:merge(Acc, SibVV) end,
CurrentVV,
Siblings),
DocRecord0 = barrel_doc:make_doc_record(Body#{<<"id">> => DocId}),
DocRecord = DocRecord0#{expected_version => undefined},
%% Reuse the ordinary write pipeline with the merged VV and
%% conflict flips added on top
Old2 = Old#{vv := MergedVV, num_conflicts := 0},
{AllOps0, {DocId, NewToken, NextHlc, Deleted, DocBody}} =
build_write_ops(StoreRef, DbName, DocRecord, Old2,
#{history_cause => resolve}),
%% Flip every conflict sibling to superseded
FlipOps = [{put,
barrel_store_keys:doc_version(
Ks, DocId, barrel_version:encode(V)),
sibling_entry(superseded, maps:get(deleted, E),
maps:get(vv, E))}
|| {V, E} <- Siblings],
ok = barrel_store_rocksdb:write_batch(
StoreRef, AllOps0 ++ FlipOps, #{}),
notify_subscribers(DbName, DocId, NewToken, NextHlc, Deleted, DocBody),
{ok, #{id => DocId, rev => NewToken,
conflicts_resolved => length(Siblings)}};
{error, _} = Err ->
Err
end.
%% @private Read the archived body of a non-current version.
read_version_body(StoreRef, DbName, DocId, Version) ->
Ks = barrel_keyspace:resolve(DbName),
Key = barrel_store_keys:doc_body_rev(Ks, DocId,
barrel_version:encode(Version)),
case barrel_store_rocksdb:body_get(StoreRef, Key) of
{ok, CborBin} -> {ok, barrel_docdb_codec_cbor:decode_any(CborBin)};
not_found -> {error, {version_body_missing, barrel_version:to_token(Version)}};
{error, _} = Err -> Err
end.