%% @doc Stateful per-node Decision/Context actor (Part B). %% %% The transient aggregate for a dynamic boundary. For a keyed, hot %% consistency boundary (one seat, one account, one SKU) this actor: %% %% 1. Serialises commands for the boundary at one process, so the %% store's append condition passes first try instead of N-1 %% context_changed retries. %% 2. Caches the folded decision model (or the raw context events) and %% updates it incrementally after each successful append — no full %% context re-read per command. %% %% It is a per-node cache + serialiser, NEVER the correctness authority: %% reckon-db's append_if_no_tag_matches/4 stays the sole source of truth. %% A second node, a different decision sharing the boundary's tags, or a %% direct append can still write — so on context_changed the actor %% invalidates its cache, re-reads context, and retries. It never trusts %% its cache as truth. %% %% Opt in by implementing evoq_decision:boundary_key/1. Absent ⇒ the %% stateless optimistic runtime (evoq_decision_runtime) handles it. %% %% Context is loaded LAZILY on the first decide: the actor is keyed by %% boundary, but context/1 takes a command. Contract: for a given %% boundary_key, context/1 must resolve to the same filter for every %% command (see evoq_decision:boundary_key/1 docs). %% %% @author rgfaber -module(evoq_decision_actor). -behaviour(gen_server). %% API -export([start_link/3]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]). -define(DEFAULT_RETRY_BUDGET, 3). -record(st, { module :: module(), key :: binary(), store_id :: atom(), %% Folded model (folded=true) or the raw context-events list %% (folded=false). `undefined' until the first decide loads it. model :: term() | undefined, cutoff = -1 :: integer(), folded :: boolean(), lifespan :: module() }). %%==================================================================== %% API %%==================================================================== %% @doc Start an actor for the {Module, Key} boundary. -spec start_link(module(), binary(), atom()) -> {ok, pid()} | {error, term()}. start_link(Module, Key, StoreId) -> gen_server:start_link(?MODULE, {Module, Key, StoreId}, []). %%==================================================================== %% gen_server callbacks %%==================================================================== %% @private init({Module, Key, StoreId}) -> evoq_decision_registry:register(Module, Key, self()), Folded = erlang:function_exported(Module, init_decision_model, 0) andalso erlang:function_exported(Module, apply_context_event, 2), Lifespan = application:get_env( evoq, lifespan_module, evoq_aggregate_lifespan_default), State = #st{ module = Module, key = Key, store_id = StoreId, model = undefined, cutoff = -1, folded = Folded, lifespan = Lifespan }, {ok, State, Lifespan:after_command(#{})}. %% @private handle_call({decide, Command}, _From, State0) -> case ensure_loaded(Command, State0) of {error, Reason} -> {reply, {error, Reason}, State0, (State0#st.lifespan):after_error(Reason)}; {ok, State1} -> Budget = retry_budget(State1#st.module), {Reply, State2} = decide_loop(Command, State1, Budget), {reply, Reply, State2, next_timeout(Reply, State2)} end; handle_call(_Request, _From, State) -> {reply, {error, unknown_request}, State, next_idle(State)}. %% @private handle_cast(_Msg, State) -> {noreply, State, next_idle(State)}. %% @private Idle timeout: passivate (v1: stop and rebuild on next spawn). handle_info(timeout, State) -> case (State#st.lifespan):on_timeout(State#st.model) of {ok, infinity} -> {noreply, State}; {ok, hibernate} -> {noreply, State, hibernate}; {ok, passivate} -> {stop, normal, State}; {ok, stop} -> {stop, normal, State}; {ok, T} when is_integer(T)-> {noreply, State, T}; %% Decision-model snapshotting (proposal OQ3) is the documented %% extension point here. v1 default: ignore the snapshot, apply %% the action — cold boundaries rebuild on next spawn. {snapshot, Action} -> apply_action(Action, State) end; handle_info(_Info, State) -> {noreply, State, next_idle(State)}. %% @private terminate(_Reason, #st{module = Module, key = Key}) -> evoq_decision_registry:unregister(Module, Key), ok. %%==================================================================== %% Internal — decide loop %%==================================================================== decide_loop(_Command, State, 0) -> {{error, retry_budget_exhausted}, State}; decide_loop(Command, State, Retries) -> #st{module = Mod, model = Model} = State, decide_result(Mod:decide(Model, Command), Command, State, Retries). decide_result({error, _} = DecideError, _Command, State, _Retries) -> {DecideError, State}; decide_result({ok, Events}, Command, State, Retries) when is_list(Events) -> #st{module = Mod, store_id = StoreId, cutoff = Cutoff} = State, Filter = Mod:context(Command), Append = evoq_event_store:append_if_no_tag_matches(StoreId, Filter, Cutoff, Events), append_result(Append, Command, State, Retries, Events). append_result({ok, LastSeq}, _Command, State, _Retries, Events) -> {{ok, Events}, absorb(State, Events, LastSeq)}; append_result({error, {context_changed, _}}, Command, State, Retries, _Events) -> %% Cache went stale: re-read context, refresh model + cutoff, re-decide %% (the decision may now differ or become an error), bounded by budget. reload_result(reload(Command, State), Command, State, Retries); append_result({error, _} = BackendError, _Command, State, _Retries, _Events) -> {BackendError, State}. reload_result({ok, State2}, Command, _State, Retries) -> decide_loop(Command, State2, Retries - 1); reload_result({error, _} = ReloadError, _Command, State, _Retries) -> {ReloadError, State}. %% Fold just-appended events into the cache and advance the cutoff to the %% store's reported high-water — no re-read. absorb(#st{folded = true, module = Mod, model = Model} = State, Events, LastSeq) -> NewModel = lists:foldl( fun(E, M) -> Mod:apply_context_event(M, E) end, Model, Events), State#st{model = NewModel, cutoff = LastSeq}; absorb(#st{folded = false, model = Events0} = State, Events, LastSeq) -> State#st{model = Events0 ++ Events, cutoff = LastSeq}. %%==================================================================== %% Internal — context loading %%==================================================================== ensure_loaded(_Command, #st{model = Model} = State) when Model =/= undefined -> {ok, State}; ensure_loaded(Command, State) -> reload(Command, State). reload(Command, #st{module = Mod, store_id = StoreId} = State) -> Filter = Mod:context(Command), case evoq_decision_runtime:load_context(StoreId, Filter) of {ok, Events, Cutoff} -> {ok, State#st{model = build_model(State, Events), cutoff = Cutoff}}; {error, _} = Error -> Error end. build_model(#st{folded = true, module = Mod}, Events) -> lists:foldl(fun(E, M) -> Mod:apply_context_event(M, E) end, Mod:init_decision_model(), Events); build_model(#st{folded = false}, Events) -> Events. %%==================================================================== %% Internal — lifespan / timeouts %%==================================================================== next_timeout({ok, _}, State) -> (State#st.lifespan):after_command(#{}); next_timeout({error, Reason}, State) -> (State#st.lifespan):after_error(Reason). next_idle(State) -> (State#st.lifespan):after_command(#{}). apply_action(passivate, State) -> {stop, normal, State}; apply_action(stop, State) -> {stop, normal, State}; apply_action(hibernate, State) -> {noreply, State, hibernate}; apply_action(infinity, State) -> {noreply, State}; apply_action(T, State) when is_integer(T) -> {noreply, State, T}. retry_budget(Mod) -> case erlang:function_exported(Mod, retry_budget, 0) of true -> Mod:retry_budget(); false -> ?DEFAULT_RETRY_BUDGET end.