Packages
reckon_db
1.2.3
5.11.0
5.10.4
5.10.3
5.10.1
5.10.0
5.9.1
5.9.0
5.8.3
5.8.2
5.8.1
5.8.0
5.7.0
5.6.1
5.6.0
5.5.5
5.5.4
5.5.3
5.5.2
5.5.1
5.5.0
5.4.0
5.2.2
5.2.1
5.2.0
5.1.0
5.0.0
4.0.0
3.1.2
3.1.1
3.0.0
2.3.7
2.3.6
2.3.5
2.3.4
2.3.3
2.3.2
2.3.1
2.3.0
2.2.2
2.2.0
2.1.4
2.1.3
2.1.2
2.1.1
2.1.0
2.0.0
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.3
1.3.2
1.3.1
1.3.0
1.2.7
1.2.6
1.2.5
1.2.4
1.2.3
1.2.2
1.2.1
1.2.0
1.1.1
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
BEAM-native Event Store built on Khepri/Ra with Raft consensus. Event sourcing, persistent subscriptions, snapshots, and automatic cluster formation via UDP multicast discovery. Ships embedded Rust NIFs for 3-15x acceleration of crypto, hashing, compression, aggregation, filter matching, and grap...
Current section
Files
Jump to
Current section
Files
src/reckon_db_streams_writer.erl
%% @doc Streams writer worker for reckon-db
%%
%% A gen_server that handles write operations for streams.
%% Writers are temporary processes that terminate after a period of inactivity.
%%
%% Features:
%% - Partitioned by stream_id for concurrent writes to different streams
%% - Idle timeout to free up resources
%% - Swarm-like registration via pg groups
%%
%% @author rgfaber
-module(reckon_db_streams_writer).
-behaviour(gen_server).
-include("reckon_db.hrl").
-include("reckon_db_telemetry.hrl").
%% API
-export([start_link/1]).
-export([append/4]).
-export([get_writer/2]).
%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).
-define(DEFAULT_IDLE_TIMEOUT_MS, 10000). %% 10 seconds (matches ex-esdb)
-record(state, {
store_id :: atom(),
stream_id :: binary(),
partition :: non_neg_integer(),
idle_timeout :: pos_integer(),
last_activity :: integer()
}).
%%====================================================================
%% API
%%====================================================================
%% @doc Start a writer worker
-spec start_link({atom(), binary(), non_neg_integer()}) -> {ok, pid()} | {error, term()}.
start_link({StoreId, StreamId, Partition}) ->
gen_server:start_link(?MODULE, {StoreId, StreamId, Partition}, []).
%% @doc Append events to a stream via a writer worker
-spec append(atom(), binary(), integer(), [map()]) ->
{ok, non_neg_integer()} | {error, term()}.
append(StoreId, StreamId, ExpectedVersion, Events) ->
Writer = get_writer(StoreId, StreamId),
gen_server:call(Writer, {append, StoreId, StreamId, ExpectedVersion, Events}, infinity).
%% @doc Get or create a writer for a stream
-spec get_writer(atom(), binary()) -> pid().
get_writer(StoreId, StreamId) ->
GroupKey = writer_group_key(StoreId, StreamId),
case pg:get_members(?RECKON_DB_PG_SCOPE, GroupKey) of
[] ->
start_new_writer(StoreId, StreamId);
Writers ->
%% Pick a random writer if multiple exist
lists:nth(rand:uniform(length(Writers)), Writers)
end.
%%====================================================================
%% gen_server callbacks
%%====================================================================
%% @private
init({StoreId, StreamId, Partition}) ->
process_flag(trap_exit, true),
%% Join the pg group for this stream
GroupKey = writer_group_key(StoreId, StreamId),
ok = pg:join(?RECKON_DB_PG_SCOPE, GroupKey, self()),
IdleTimeout = application:get_env(reckon_db, writer_idle_timeout_ms, ?DEFAULT_IDLE_TIMEOUT_MS),
%% Schedule idle check
schedule_idle_check(IdleTimeout),
logger:debug("Streams writer started: store=~p stream=~s partition=~p",
[StoreId, StreamId, Partition]),
State = #state{
store_id = StoreId,
stream_id = StreamId,
partition = Partition,
idle_timeout = IdleTimeout,
last_activity = erlang:system_time(millisecond)
},
{ok, State}.
%% @private
handle_call({append, StoreId, StreamId, ExpectedVersion, Events}, _From, State) ->
Result = reckon_db_streams:do_append(StoreId, StreamId, ExpectedVersion, Events),
NewState = State#state{last_activity = erlang:system_time(millisecond)},
{reply, Result, NewState};
handle_call(_Request, _From, State) ->
{reply, {error, unknown_request}, State}.
%% @private
handle_cast(_Msg, State) ->
{noreply, State}.
%% @private
handle_info(check_idle, #state{idle_timeout = Timeout, last_activity = LastActivity} = State) ->
Now = erlang:system_time(millisecond),
IdleDuration = Now - LastActivity,
case IdleDuration >= Timeout of
true ->
logger:debug("Streams writer idle timeout reached, stopping"),
{stop, normal, State};
false ->
%% Schedule next check
schedule_idle_check(Timeout),
{noreply, State}
end;
handle_info(_Info, State) ->
{noreply, State}.
%% @private
terminate(_Reason, #state{store_id = StoreId, stream_id = StreamId}) ->
GroupKey = writer_group_key(StoreId, StreamId),
pg:leave(?RECKON_DB_PG_SCOPE, GroupKey, self()),
ok.
%%====================================================================
%% Internal functions
%%====================================================================
%% @private
-spec writer_group_key(atom(), binary()) -> term().
writer_group_key(StoreId, StreamId) ->
{StoreId, StreamId, streams_writer}.
%% @private
-spec start_new_writer(atom(), binary()) -> pid().
start_new_writer(StoreId, StreamId) ->
Partition = partition_for(StoreId, StreamId),
SupName = reckon_db_naming:streams_sup_name(StoreId),
ChildSpec = #{
id => make_ref(),
start => {?MODULE, start_link, [{StoreId, StreamId, Partition}]},
restart => temporary,
shutdown => 5000,
type => worker,
modules => [?MODULE]
},
case supervisor:start_child(SupName, ChildSpec) of
{ok, Pid} -> Pid;
{error, {already_started, Pid}} -> Pid;
{error, Reason} ->
logger:error("Failed to start streams writer: ~p", [Reason]),
error({failed_to_start_writer, Reason})
end.
%% @private
-spec partition_for(atom(), binary()) -> non_neg_integer().
partition_for(StoreId, StreamId) ->
Partitions = erlang:system_info(schedulers_online),
erlang:phash2({StoreId, StreamId}, Partitions).
%% @private
-spec schedule_idle_check(pos_integer()) -> reference().
schedule_idle_check(Timeout) ->
erlang:send_after(Timeout, self(), check_idle).