%% This Source Code Form is subject to the terms of the Mozilla Public
%% License, v. 2.0. If a copy of the MPL was not distributed with this
%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
%%
%% Copyright © 2021-2022 VMware, Inc. or its affiliates. All rights reserved.
%%
%% @doc
%% Khepri private low-level API.
%%
%% This module exposes the private "low-level" API to the Khepri database and
%% state machine. Main functions correspond to Ra commands implemented by the
%% state machine. All functions in {@link khepri} are built on top of this
%% module.
%%
%% This module is private. The documentation is still visible because it may
%% help understand some implementation details. However, this module should
%% never be called directly outside of Khepri.
-module(khepri_machine).
-behaviour(ra_machine).
-include_lib("kernel/include/logger.hrl").
-include_lib("stdlib/include/assert.hrl").
-include("include/khepri.hrl").
-include("src/khepri_cluster.hrl").
-include("src/khepri_error.hrl").
-include("src/khepri_evf.hrl").
-include("src/khepri_fun.hrl").
-include("src/khepri_machine.hrl").
-include("src/khepri_ret.hrl").
-include("src/khepri_tx.hrl").
-include("src/khepri_projection.hrl").
-export([fold/5,
put/4,
delete/3,
transaction/5,
register_trigger/5,
register_projection/4]).
-export([get_keep_while_conds_state/2,
get_projections_state/2]).
%% ra_machine callbacks.
-export([init/1,
init_aux/1,
handle_aux/6,
apply/3,
state_enter/2,
overview/1]).
%% For internal use only.
-export([clear_cache/1,
ack_triggers_execution/2,
split_query_options/1,
split_command_options/1,
split_put_options/1,
find_matching_nodes/3,
find_matching_nodes/5,
collect_node_props_cb/3,
count_node_cb/3,
insert_or_update_node/5,
delete_matching_nodes/3,
handle_tx_exception/1,
process_query/3,
process_command/3,
walk_down_the_tree/6]).
-ifdef(TEST).
-export([are_keep_while_conditions_met/2,
get_root/1,
get_keep_while_conds/1,
get_keep_while_conds_revidx/1,
get_last_consistent_call_atomics/1]).
-endif.
-compile({no_auto_import, [apply/3]}).
-type tree_node() :: #node{}.
%% A node in the tree structure.
-type props() :: #{payload_version := khepri:payload_version(),
child_list_version := khepri:child_list_version()}.
%% Properties attached to each node in the tree structure.
-type triggered() :: #triggered{}.
-type command() :: #put{} |
#delete{} |
#tx{} |
#register_trigger{} |
#ack_triggered{} |
#register_projection{}.
%% Commands specific to this Ra machine.
-type machine_init_args() :: #{store_id := khepri:store_id(),
member := ra:server_id(),
snapshot_interval => non_neg_integer(),
commands => [command()],
atom() => any()}.
%% Structure passed to {@link init/1}.
-type machine_config() :: #config{}.
%% Configuration record, holding read-only or rarely changing fields.
-type keep_while_conds_map() :: #{khepri_path:native_path() =>
khepri_condition:native_keep_while()}.
%% Per-node `keep_while' conditions.
-type keep_while_conds_revidx() :: #{khepri_path:native_path() =>
#{khepri_path:native_path() => ok}}.
%% Internal reverse index of the keep_while conditions. If node A depends on a
%% condition on node B, then this reverse index will have a "node B => node A"
%% entry.
-type applied_changes() :: #{khepri_path:native_path() =>
khepri:node_props() | delete}.
%% Internal index of the per-node changes which happened during a traversal.
%% This is used when the tree is walked back up to determine the list of tree
%% nodes to remove after some keep_while condition evaluates to false.
-type projections_map() :: #{khepri_projection:projection() =>
khepri_path:native_pattern()}.
%% Internal mapping between {@link khepri_projection:projection()} records and
%% the native path patterns which trigger updates to each projection.
-type state() :: #?MODULE{}.
%% State of this Ra state machine.
-type aux_state() :: #khepri_machine_aux{}.
%% Auxiliary state of this Ra state machine.
-type query_fun() :: fun((state()) -> any()).
%% Function representing a query and used {@link process_query/3}.
-type walk_down_the_tree_extra() :: #{keep_while_conds =>
keep_while_conds_map(),
keep_while_conds_revidx =>
keep_while_conds_revidx(),
applied_changes =>
applied_changes()}.
-type walk_down_the_tree_fun() ::
fun((khepri_path:native_path(),
tree_node() | {interrupted, any(), map()},
Acc :: any()) ->
ok(tree_node() | keep | delete, any()) |
khepri:error()).
%% Function called to handle a node found (or an error) and used in {@link
%% walk_down_the_tree/6}.
-type ok(Type1, Type2) :: {ok, Type1, Type2}.
-type ok(Type1, Type2, Type3) :: {ok, Type1, Type2, Type3}.
-type common_ret() :: khepri:ok(khepri_adv:node_props_map()) |
khepri:error().
-type tx_ret() :: khepri:ok(khepri_tx:tx_fun_result()) |
khepri_tx:tx_abort() |
no_return().
-type async_ret() :: ok.
-export_type([common_ret/0,
tx_ret/0,
async_ret/0,
state/0,
machine_config/0,
tree_node/0,
props/0,
triggered/0,
keep_while_conds_map/0,
keep_while_conds_revidx/0,
projections_map/0]).
-define(HAS_TIME_LEFT(Timeout), (Timeout =:= infinity orelse Timeout > 0)).
-define(PROJECTION_PROPS_TO_RETURN, [payload_version,
child_list_version,
child_list_length,
child_names,
payload]).
%% -------------------------------------------------------------------
%% Machine protocol.
%% -------------------------------------------------------------------
%% TODO: Verify arguments carefully to avoid the construction of an invalid
%% command.
-spec fold(StoreId, PathPattern, Fun, Acc, Options) -> Ret when
StoreId :: khepri:store_id(),
PathPattern :: khepri_path:pattern(),
Fun :: khepri:fold_fun(),
Acc :: khepri:fold_acc(),
Options :: khepri:query_options() | khepri:tree_options(),
Ret :: khepri:ok(NewAcc) | khepri:error(),
NewAcc :: Acc.
%% @doc Returns all tree nodes matching the given path pattern.
%%
%% @param StoreId the name of the Ra cluster.
%% @param PathPattern the path (or path pattern) to the nodes to get.
%% @param Options query options such as `favor'.
%%
%% @returns an `{ok, NodePropsMap}' tuple with a map with zero, one or more
%% entries, or an `{error, Reason}' tuple.
fold(StoreId, PathPattern, Fun, Acc, Options)
when ?IS_STORE_ID(StoreId) andalso
is_function(Fun, 3) ->
PathPattern1 = khepri_path:from_string(PathPattern),
khepri_path:ensure_is_valid(PathPattern1),
{QueryOptions, TreeOptions} = split_query_options(Options),
Query = fun(#?MODULE{root = Root}) ->
try
find_matching_nodes(
Root, PathPattern1, Fun, Acc, TreeOptions)
catch
Class:Reason:Stacktrace ->
{exception, Class, Reason, Stacktrace}
end
end,
case process_query(StoreId, Query, QueryOptions) of
{exception, _, _, _} = Exception -> handle_tx_exception(Exception);
Ret -> Ret
end.
-spec put(StoreId, PathPattern, Payload, Options) -> Ret when
StoreId :: khepri:store_id(),
PathPattern :: khepri_path:pattern(),
Payload :: khepri_payload:payload(),
Options :: khepri:command_options() |
khepri:tree_options() |
khepri:put_options(),
Ret :: khepri_machine:common_ret() | khepri_machine:async_ret().
%% @doc Creates or modifies a specific tree node in the tree structure.
%%
%% @param StoreId the name of the Ra cluster.
%% @param PathPattern the path (or path pattern) to the node to create or
%% modify.
%% @param Payload the payload to put in the specified node.
%% @param Options command, tree and put options.
%%
%% @returns in the case of a synchronous put, an `{ok, NodePropsMap}' tuple
%% with a map with zero, one or more entries, or an `{error, Reason}' tuple;
%% in the case of an asynchronous put, always `ok' (the actual return value
%% may be sent by a message if a correlation ID was specified).
%%
%% @private
put(StoreId, PathPattern, Payload, Options)
when ?IS_STORE_ID(StoreId) andalso ?IS_KHEPRI_PAYLOAD(Payload) ->
PathPattern1 = khepri_path:from_string(PathPattern),
khepri_path:ensure_is_valid(PathPattern1),
Payload1 = khepri_payload:prepare(Payload),
{CommandOptions, TreeAndPutOptions} = split_command_options(Options),
Command = #put{path = PathPattern1,
payload = Payload1,
options = TreeAndPutOptions},
process_command(StoreId, Command, CommandOptions);
put(_StoreId, PathPattern, Payload, _Options) ->
?khepri_misuse(invalid_payload, #{path => PathPattern,
payload => Payload}).
-spec delete(StoreId, PathPattern, Options) -> Ret when
StoreId :: khepri:store_id(),
PathPattern :: khepri_path:pattern(),
Options :: khepri:command_options() | khepri:tree_options(),
Ret :: khepri_machine:common_ret() | khepri_machine:async_ret().
%% @doc Deletes all tree nodes matching the path pattern.
%%
%% @param StoreId the name of the Ra cluster.
%% @param PathPattern the path (or path pattern) to the nodes to delete.
%% @param Options command options such as the command type.
%%
%% @returns in the case of a synchronous delete, an `{ok, NodePropsMap}' tuple
%% with a map with zero, one or more entries, or an `{error, Reason}' tuple;
%% in the case of an asynchronous put, always `ok' (the actual return value
%% may be sent by a message if a correlation ID was specified).
delete(StoreId, PathPattern, Options) when ?IS_STORE_ID(StoreId) ->
PathPattern1 = khepri_path:from_string(PathPattern),
khepri_path:ensure_is_valid(PathPattern1),
{CommandOptions, TreeOptions} = split_command_options(Options),
%% TODO: Ensure `Extra' is unset.
Command = #delete{path = PathPattern1,
options = TreeOptions},
process_command(StoreId, Command, CommandOptions).
-spec transaction(StoreId, FunOrPath, Args, ReadWrite, Options) -> Ret when
StoreId :: khepri:store_id(),
FunOrPath :: Fun | PathPattern,
Fun :: khepri_tx:tx_fun(),
PathPattern :: khepri_path:pattern(),
Args :: list(),
ReadWrite :: ro | rw | auto,
Options :: khepri:command_options() | khepri:query_options(),
Ret :: khepri_machine:tx_ret() | khepri_machine:async_ret().
%% @doc Runs a transaction and returns the result.
%%
%% @param StoreId the name of the Ra cluster.
%% @param FunOrPath an arbitrary anonymous function or a path pattern pointing
%% to a stored procedure.
%% @param Args a list of arguments to pass to `FunOrPath'.
%% @param ReadWrite the read/write or read-only nature of the transaction.
%% @param Options command options such as the command type.
%%
%% @returns in the case of a synchronous transaction, `{ok, Result}' where
%% `Result' is the return value of `FunOrPath', or `{error, Reason}' if the
%% anonymous function was aborted; in the case of an asynchronous transaction,
%% always `ok' (the actual return value may be sent by a message if a
%% correlation ID was specified).
transaction(StoreId, Fun, Args, auto = ReadWrite, Options)
when ?IS_STORE_ID(StoreId) andalso
is_list(Args) andalso
is_function(Fun, length(Args)) andalso
is_map(Options) ->
case khepri_tx_adv:to_standalone_fun(Fun, ReadWrite) of
#standalone_fun{} = StandaloneFun ->
readwrite_transaction(StoreId, StandaloneFun, Args, Options);
_ ->
readonly_transaction(StoreId, Fun, Args, Options)
end;
transaction(StoreId, PathPattern, Args, auto, Options)
when ?IS_STORE_ID(StoreId) andalso
?IS_KHEPRI_PATH_PATTERN(PathPattern) andalso
is_list(Args) andalso
is_map(Options) ->
PathPattern1 = khepri_path:from_string(PathPattern),
khepri_path:ensure_is_valid(PathPattern1),
readwrite_transaction(StoreId, PathPattern1, Args, Options);
transaction(StoreId, Fun, Args, rw = ReadWrite, Options)
when ?IS_STORE_ID(StoreId) andalso
is_list(Args) andalso
is_function(Fun, length(Args)) andalso
is_map(Options) ->
StandaloneFun = khepri_tx_adv:to_standalone_fun(Fun, ReadWrite),
readwrite_transaction(StoreId, StandaloneFun, Args, Options);
transaction(StoreId, PathPattern, Args, rw, Options)
when ?IS_STORE_ID(StoreId) andalso
?IS_KHEPRI_PATH_PATTERN(PathPattern) andalso
is_list(Args) andalso
is_map(Options) ->
PathPattern1 = khepri_path:from_string(PathPattern),
khepri_path:ensure_is_valid(PathPattern1),
readwrite_transaction(StoreId, PathPattern1, Args, Options);
transaction(StoreId, Fun, Args, ro, Options)
when ?IS_STORE_ID(StoreId) andalso
is_list(Args) andalso
is_function(Fun, length(Args)) andalso
is_map(Options) ->
readonly_transaction(StoreId, Fun, Args, Options);
transaction(StoreId, PathPattern, Args, ro, Options)
when ?IS_STORE_ID(StoreId) andalso
?IS_KHEPRI_PATH_PATTERN(PathPattern) andalso
is_list(Args) andalso
is_map(Options) ->
PathPattern1 = khepri_path:from_string(PathPattern),
khepri_path:ensure_is_valid(PathPattern1),
readonly_transaction(StoreId, PathPattern1, Args, Options);
transaction(StoreId, Fun, Args, ReadWrite, Options)
when ?IS_STORE_ID(StoreId) andalso
is_function(Fun) andalso
is_list(Args) andalso
is_atom(ReadWrite) andalso
is_map(Options) ->
{arity, Arity} = erlang:fun_info(Fun, arity),
?khepri_misuse(
denied_tx_fun_with_invalid_args,
#{'fun' => Fun, arity => Arity, args => Args}).
-spec readonly_transaction(StoreId, FunOrPath, Args, Options) -> Ret when
StoreId :: khepri:store_id(),
FunOrPath :: Fun | PathPattern,
Fun :: khepri_tx:tx_fun(),
PathPattern :: khepri_path:pattern(),
Args :: list(),
Options :: khepri:query_options(),
Ret :: khepri_machine:tx_ret().
readonly_transaction(StoreId, Fun, Args, Options)
when is_list(Args) andalso is_function(Fun, length(Args)) ->
Query = fun(State) ->
%% It is a read-only transaction, therefore we assert that
%% the state is unchanged and that there are no side
%% effects.
{State, Ret, []} = khepri_tx_adv:run(
State, Fun, Args, false),
Ret
end,
case process_query(StoreId, Query, Options) of
{exception, _, _, _} = Exception ->
handle_tx_exception(Exception);
Ret ->
{ok, Ret}
end;
readonly_transaction(StoreId, PathPattern, Args, Options)
when ?IS_KHEPRI_PATH_PATTERN(PathPattern) andalso is_list(Args) ->
Query = fun(State) ->
%% It is a read-only transaction, therefore we assert that
%% the state is unchanged and that there are no side
%% effects.
{State, Ret, []} = locate_sproc_and_execute_tx(
State, PathPattern, Args, false),
Ret
end,
case process_query(StoreId, Query, Options) of
{exception, _, _, _} = Exception ->
handle_tx_exception(Exception);
Ret ->
{ok, Ret}
end.
-spec readwrite_transaction(StoreId, FunOrPath, Args, Options) -> Ret when
StoreId :: khepri:store_id(),
FunOrPath :: Fun | PathPattern,
Fun :: khepri_fun:standalone_fun(),
PathPattern :: khepri_path:pattern(),
Args :: list(),
Options :: khepri:command_options(),
Ret :: khepri_machine:tx_ret() | khepri_machine:async_ret().
readwrite_transaction(
StoreId, #standalone_fun{arity = Arity} = StandaloneFun, Args, Options)
when is_list(Args) andalso length(Args) =:= Arity ->
readwrite_transaction1(StoreId, StandaloneFun, Args, Options);
readwrite_transaction(
StoreId, Fun, Args, Options)
when is_list(Args) andalso is_function(Fun, length(Args)) ->
readwrite_transaction1(StoreId, Fun, Args, Options);
readwrite_transaction(
StoreId, PathPattern, Args, Options)
when ?IS_KHEPRI_PATH_PATTERN(PathPattern) andalso is_list(Args) ->
PathPattern1 = khepri_path:from_string(PathPattern),
khepri_path:ensure_is_valid(PathPattern1),
readwrite_transaction1(StoreId, PathPattern1, Args, Options).
readwrite_transaction1(StoreId, StandaloneFunOrPath, Args, Options) ->
Command = #tx{'fun' = StandaloneFunOrPath, args = Args},
case process_command(StoreId, Command, Options) of
{exception, _, _, _} = Exception ->
handle_tx_exception(Exception);
ok = Ret ->
CommandType = select_command_type(Options),
case CommandType of
sync -> {ok, Ret};
{async, _, _} -> Ret
end;
Ret ->
{ok, Ret}
end.
handle_tx_exception(
{exception, _, ?TX_ABORT(Reason), _}) ->
{error, Reason};
handle_tx_exception(
{exception, error, ?khepri_exception(_, _) = Reason, _Stacktrace}) ->
%% If the exception is a programming misuse of Khepri, we
%% re-throw a new exception instead of using `erlang:raise()'.
%%
%% The reason is that the default stacktrace is limited to 8 frames by
%% default (see `erlang:system_flag(backtrace_depth, Depth)' to reconfigure
%% it). Most if not all of those 8 frames might be taken by Khepri's
%% internal calls, making the stacktrace uninformative to the caller.
%%
%% By throwing a new exception, we increase the chance that there
%% is a frame pointing to the transaction function.
?khepri_misuse(Reason);
handle_tx_exception(
{exception, Class, Reason, Stacktrace}) ->
erlang:raise(Class, Reason, Stacktrace).
-spec register_trigger(
StoreId, TriggerId, EventFilter, StoredProcPath, Options) ->
Ret when
StoreId :: khepri:store_id(),
TriggerId :: khepri:trigger_id(),
EventFilter :: khepri_evf:event_filter() | khepri_path:pattern(),
StoredProcPath :: khepri_path:path(),
Options :: khepri:command_options(),
Ret :: ok | khepri:error().
%% @doc Registers a trigger.
%%
%% @param StoreId the name of the Ra cluster.
%% @param TriggerId the name of the trigger.
%% @param EventFilter the event filter used to associate an event with a
%% stored procedure.
%% @param StoredProcPath the path to the stored procedure to execute when the
%% corresponding event occurs.
%%
%% @returns `ok' if the trigger was registered, an `{error, Reason}' tuple
%% otherwise.
register_trigger(StoreId, TriggerId, EventFilter, StoredProcPath, Options)
when ?IS_STORE_ID(StoreId) ->
EventFilter1 = khepri_evf:wrap(EventFilter),
StoredProcPath1 = khepri_path:from_string(StoredProcPath),
khepri_path:ensure_is_valid(StoredProcPath1),
Command = #register_trigger{id = TriggerId,
sproc = StoredProcPath1,
event_filter = EventFilter1},
process_command(StoreId, Command, Options).
-spec register_projection(StoreId, PathPattern, Projection, Options) ->
Ret when
StoreId :: khepri:store_id(),
PathPattern :: khepri_path:pattern(),
Projection :: khepri_projection:projection(),
Options :: khepri:command_options(),
Ret :: ok | khepri:error().
%% @doc Registers a projection.
%%
%% @param StoreId the name of the Ra cluster.
%% @param PathPattern the pattern of tree nodes which should be projected.
%% @param Projection the projection record created with {@link
%% khepri_projection:new/3}.
%% @param Options command options such as the command type.
%%
%% @returns `ok' if the projection was registered, an `{error, Reason}' tuple
%% otherwise.
register_projection(
StoreId, PathPattern0,
#khepri_projection{name = Name,
projection_fun = #standalone_fun{},
ets_options = EtsOptions} = Projection,
Options0) when is_atom(Name) andalso is_list(EtsOptions) ->
Options = Options0#{reply_from => local},
PathPattern = khepri_path:from_string(PathPattern0),
khepri_path:ensure_is_valid(PathPattern),
Command = #register_projection{pattern = PathPattern,
projection = Projection},
process_command(StoreId, Command, Options).
-spec ack_triggers_execution(StoreId, TriggeredStoredProcs) ->
Ret when
StoreId :: khepri:store_id(),
TriggeredStoredProcs :: [triggered()],
Ret :: ok | khepri:error().
%% @doc Acknowledges the execution of a trigger.
%%
%% This is part of a mechanism to ensure that a trigger is executed at least
%% once.
%%
%% @private
ack_triggers_execution(StoreId, TriggeredStoredProcs)
when ?IS_STORE_ID(StoreId) ->
Command = #ack_triggered{triggered = TriggeredStoredProcs},
process_command(StoreId, Command, #{async => true}).
-spec get_keep_while_conds_state(StoreId, Options) -> Ret when
StoreId :: khepri:store_id(),
Options :: khepri:query_options(),
Ret :: khepri:ok(keep_while_conds_map()) | khepri:error().
%% @doc Returns the `keep_while' conditions internal state.
%%
%% The returned state consists of all the `keep_while' condition set so far.
%% However, it doesn't include the reverse index.
%%
%% @param StoreId the name of the Ra cluster.
%%
%% @returns the `keep_while' conditions internal state.
%%
%% @private
get_keep_while_conds_state(StoreId, Options)
when ?IS_STORE_ID(StoreId) ->
Query = fun(#?MODULE{keep_while_conds = KeepWhileConds}) ->
{ok, KeepWhileConds}
end,
Options1 = Options#{favor => consistency},
process_query(StoreId, Query, Options1).
-spec get_projections_state(StoreId, Options) -> Ret when
StoreId :: khepri:store_id(),
Options :: khepri:query_options(),
Ret :: khepri:ok(projections_map()) | khepri:error().
%% @doc Returns the `projections' internal state.
%%
%% The returned state consists of the mapping between path pattern and
%% {@link khepri_projection:projection()} records.
%%
%% @see khepri_projection.
%%
%% @private
get_projections_state(StoreId, Options)
when ?IS_STORE_ID(StoreId) ->
Query = fun(#?MODULE{projections = Projections}) ->
{ok, Projections}
end,
Options1 = Options#{favor => consistency},
process_query(StoreId, Query, Options1).
-spec split_query_options(Options) -> {QueryOptions, TreeOptions} when
Options :: QueryOptions | TreeOptions,
QueryOptions :: khepri:query_options(),
TreeOptions :: khepri:tree_options().
%% @private
split_query_options(Options) ->
Options1 = set_default_options(Options),
maps:fold(
fun
(Option, Value, {Q, T}) when
Option =:= timeout orelse
Option =:= favor ->
Q1 = Q#{Option => Value},
{Q1, T};
(props_to_return, [], {Q, T}) ->
{Q, T};
(Option, Value, {Q, T}) when
Option =:= expect_specific_node orelse
Option =:= props_to_return orelse
Option =:= include_root_props ->
T1 = T#{Option => Value},
{Q, T1}
end, {#{}, #{}}, Options1).
-spec split_command_options(Options) ->
{CommandOptions, TreeAndPutOptions} when
Options :: CommandOptions | TreeAndPutOptions,
CommandOptions :: khepri:command_options(),
TreeAndPutOptions :: khepri:tree_options() | khepri:put_options().
%% @private
split_command_options(Options) ->
Options1 = set_default_options(Options),
maps:fold(
fun
(Option, Value, {C, TP}) when
Option =:= reply_from orelse
Option =:= timeout orelse
Option =:= async ->
C1 = C#{Option => Value},
{C1, TP};
(props_to_return, [], Acc) ->
Acc;
(Option, Value, {C, TP}) when
Option =:= expect_specific_node orelse
Option =:= props_to_return orelse
Option =:= include_root_props ->
TP1 = TP#{Option => Value},
{C, TP1};
(keep_while, KeepWhile, {C, TP}) ->
%% `keep_while' is kept in `TreeAndPutOptions' here. The state
%% machine will extract it in `apply()'.
KeepWhile1 = khepri_condition:ensure_native_keep_while(
KeepWhile),
TP1 = TP#{keep_while => KeepWhile1},
{C, TP1}
end, {#{}, #{}}, Options1).
-spec split_put_options(TreeAndPutOptions) -> {TreeOptions, PutOptions} when
TreeAndPutOptions :: TreeOptions | PutOptions,
TreeOptions :: khepri:tree_options(),
PutOptions :: khepri:put_options().
%% @private
split_put_options(TreeAndPutOptions) ->
maps:fold(
fun
(keep_while, KeepWhile, {T, P}) ->
P1 = P#{keep_while => KeepWhile},
{T, P1};
(Option, Value, {T, P}) ->
T1 = T#{Option => Value},
{T1, P}
end, {#{}, #{}}, TreeAndPutOptions).
-define(DEFAULT_PROPS_TO_RETURN, [payload,
payload_version]).
set_default_options(Options) ->
%% By default, return payload-related properties. The caller can set
%% `props_to_return' to an empty map to get a minimal return value.
Options1 = case Options of
#{props_to_return := _} ->
Options;
_ ->
Options#{props_to_return => ?DEFAULT_PROPS_TO_RETURN}
end,
Options1.
-spec process_command(StoreId, Command, Options) -> Ret when
StoreId :: khepri:store_id(),
Command :: command(),
Options :: khepri:command_options(),
Ret :: any().
%% @doc Processes a command which is appended to the Ra log and processed by
%% this state machine code.
%%
%% `Command' may modify the state of the machine.
%%
%% The command associated code is executed in the context of the state machine
%% process on each Ra members.
%%
%% @param StoreId the name of the Ra cluster.
%%
%% @returns the result of the command or an "error" tuple.
%%
%% @private
process_command(StoreId, Command, Options) ->
CommandType = select_command_type(Options),
case CommandType of
sync ->
process_sync_command(StoreId, Command, Options);
{async, Correlation, Priority} ->
process_async_command(
StoreId, Command, Correlation, Priority)
end.
process_sync_command(StoreId, Command, Options) ->
Timeout = get_timeout(Options),
ReplyFrom = maps:get(reply_from, Options, leader),
CommandOptions = #{timeout => Timeout, reply_from => ReplyFrom},
T0 = khepri_utils:start_timeout_window(Timeout),
LeaderId = khepri_cluster:get_cached_leader(StoreId),
RaServer = use_leader_or_local_ra_server(StoreId, LeaderId),
case ra:process_command(RaServer, Command, CommandOptions) of
{ok, Ret, NewLeaderId} ->
khepri_cluster:cache_leader_if_changed(
StoreId, LeaderId, NewLeaderId),
just_did_consistent_call(StoreId),
?raise_exception_if_any(Ret);
{timeout, _} = TimedOut ->
{error, TimedOut};
{error, Reason}
when LeaderId =/= undefined andalso ?HAS_TIME_LEFT(Timeout) andalso
(Reason == noproc orelse Reason == nodedown) ->
%% The cached leader is no more. We simply clear the cache
%% entry and retry.
khepri_cluster:clear_cached_leader(StoreId),
NewTimeout = khepri_utils:end_timeout_window(Timeout, T0),
Options1 = Options#{timeout => NewTimeout},
process_sync_command(StoreId, Command, Options1);
{error, Reason} = Error
when LeaderId =:= undefined andalso ?HAS_TIME_LEFT(Timeout) andalso
(Reason == noproc orelse Reason == nodedown) ->
case khepri_utils:is_ra_server_alive(RaServer) of
true ->
%% The follower doesn't know about the new leader yet.
%% Retry again after waiting a bit.
NewTimeout0 = khepri_utils:end_timeout_window(Timeout, T0),
NewTimeout = khepri_utils:sleep(
?NOPROC_RETRY_INTERVAL, NewTimeout0),
Options1 = Options#{timeout => NewTimeout},
process_sync_command(StoreId, Command, Options1);
false ->
Error
end;
{error, _} = Error ->
Error
end.
process_async_command(StoreId, Command, Correlation, Priority) ->
LocalServerId = {StoreId, node()},
ra:pipeline_command(LocalServerId, Command, Correlation, Priority).
-spec select_command_type(Options) -> CommandType when
Options :: khepri:command_options(),
CommandType :: sync | {async, Correlation, Priority},
Correlation :: ra_server:command_correlation(),
Priority :: ra_server:command_priority().
%% @doc Selects the command type depending on what the caller wants.
%%
%% @private
-define(DEFAULT_RA_COMMAND_CORRELATION, no_correlation).
-define(DEFAULT_RA_COMMAND_PRIORITY, low).
-define(IS_RA_COMMAND_CORRELATION(Correlation),
(is_integer(Correlation) orelse is_reference(Correlation))).
-define(IS_RA_COMMAND_PRIORITY(Priority),
(Priority =:= normal orelse Priority =:= low)).
select_command_type(Options) when not is_map_key(async, Options) ->
sync;
select_command_type(#{async := false}) ->
sync;
select_command_type(#{async := true}) ->
{async, ?DEFAULT_RA_COMMAND_CORRELATION, ?DEFAULT_RA_COMMAND_PRIORITY};
select_command_type(#{async := Correlation})
when ?IS_RA_COMMAND_CORRELATION(Correlation) ->
{async, Correlation, ?DEFAULT_RA_COMMAND_PRIORITY};
select_command_type(#{async := Priority})
when ?IS_RA_COMMAND_PRIORITY(Priority) ->
{async, ?DEFAULT_RA_COMMAND_CORRELATION, Priority};
select_command_type(#{async := {Correlation, Priority}})
when ?IS_RA_COMMAND_CORRELATION(Correlation) andalso
?IS_RA_COMMAND_PRIORITY(Priority) ->
{async, Correlation, Priority}.
-spec process_query(StoreId, QueryFun, Options) -> Ret when
StoreId :: khepri:store_id(),
QueryFun :: query_fun(),
Options :: khepri:query_options(),
Ret :: any().
%% @doc Processes a query which is by the Ra leader.
%%
%% The `QueryFun' function takes the machine state as an argument and can
%% return anything. However, the machine state is never modified. The query
%% does not go through the Ra log and is not replicated.
%%
%% The `QueryFun' function is executed from a process on the leader Ra member.
%%
%% @param StoreId the name of the Ra cluster.
%%
%% @returns the result of the query or an "error" tuple.
%%
%% @private
process_query(StoreId, QueryFun, Options) ->
QueryType = select_query_type(StoreId, Options),
Timeout = get_timeout(Options),
case QueryType of
local -> process_local_query(StoreId, QueryFun, Timeout);
_ -> process_non_local_query(StoreId, QueryFun, QueryType, Timeout)
end.
-spec process_local_query(StoreId, QueryFun, Timeout) -> Ret when
StoreId :: khepri:store_id(),
QueryFun :: query_fun(),
Timeout :: timeout(),
Ret :: any().
process_local_query(StoreId, QueryFun, Timeout) ->
LocalServerId = {StoreId, node()},
Ret = ra:local_query(LocalServerId, QueryFun, Timeout),
process_query_response(
StoreId, LocalServerId, false, QueryFun, local, Timeout, Ret).
-spec process_non_local_query(StoreId, QueryFun, QueryType, Timeout) ->
Ret when
StoreId :: khepri:store_id(),
QueryFun :: query_fun(),
QueryType :: leader | consistent,
Timeout :: timeout(),
Ret :: any().
process_non_local_query(StoreId, QueryFun, QueryType, Timeout)
when QueryType =:= leader orelse
QueryType =:= consistent ->
T0 = khepri_utils:start_timeout_window(Timeout),
LeaderId = khepri_cluster:get_cached_leader(StoreId),
RaServer = use_leader_or_local_ra_server(StoreId, LeaderId),
Ret = case QueryType of
leader -> ra:leader_query(RaServer, QueryFun, Timeout);
consistent -> ra:consistent_query(RaServer, QueryFun, Timeout)
end,
NewTimeout = khepri_utils:end_timeout_window(Timeout, T0),
%% TODO: If the consistent query times out in the context of
%% `QueryType=compromise`, should we retry with a local query to
%% never block the query and let the caller continue?
process_query_response(
StoreId, RaServer, LeaderId =/= undefined, QueryFun, QueryType,
NewTimeout, Ret).
-spec process_query_response(
StoreId, RaServer, IsLeader, QueryFun, QueryType, Timeout,
Response) ->
Ret when
StoreId :: khepri:store_id(),
RaServer :: ra:server_id(),
IsLeader :: boolean(),
QueryFun :: query_fun(),
QueryType :: local | leader | consistent,
Timeout :: timeout(),
Response :: {ok, {RaIndex, any()}, NewLeaderId} |
{ok, any(), NewLeaderId} |
{error, any()} |
{timeout, ra:server_id()},
RaIndex :: ra:index(),
NewLeaderId :: ra:server_id(),
Ret :: any().
process_query_response(
StoreId, RaServer, IsLeader, _QueryFun, consistent, _Timeout,
{ok, Ret, NewLeaderId}) ->
case IsLeader of
true ->
khepri_cluster:cache_leader_if_changed(
StoreId, RaServer, NewLeaderId);
false ->
khepri_cluster:cache_leader(StoreId, NewLeaderId)
end,
just_did_consistent_call(StoreId),
?raise_exception_if_any(Ret);
process_query_response(
StoreId, RaServer, IsLeader, _QueryFun, _QueryType, _Timeout,
{ok, {_RaIndex, Ret}, NewLeaderId}) ->
case IsLeader of
true ->
khepri_cluster:cache_leader_if_changed(
StoreId, RaServer, NewLeaderId);
false ->
khepri_cluster:cache_leader(StoreId, NewLeaderId)
end,
?raise_exception_if_any(Ret);
process_query_response(
_StoreId, _RaServer, _IsLeader, _QueryFun, _QueryType, _Timeout,
{timeout, _} = TimedOut) ->
{error, TimedOut};
process_query_response(
StoreId, _RaServer, true = _IsLeader, QueryFun, QueryType, Timeout,
{error, Reason})
when QueryType =/= local andalso ?HAS_TIME_LEFT(Timeout) andalso
(Reason == noproc orelse Reason == nodedown) ->
%% The cached leader is no more. We simply clear the cache
%% entry and retry. It may time out eventually.
khepri_cluster:clear_cached_leader(StoreId),
process_non_local_query(StoreId, QueryFun, QueryType, Timeout);
process_query_response(
StoreId, RaServer, false = _IsLeader, QueryFun, QueryType, Timeout,
{error, Reason} = Error)
when QueryType =/= local andalso ?HAS_TIME_LEFT(Timeout) andalso
(Reason == noproc orelse Reason == nodedown)->
case khepri_utils:is_ra_server_alive(RaServer) of
true ->
%% The follower doesn't know about the new leader yet. Retry again
%% after waiting a bit.
NewTimeout = khepri_utils:sleep(?NOPROC_RETRY_INTERVAL, Timeout),
process_non_local_query(StoreId, QueryFun, QueryType, NewTimeout);
false ->
Error
end;
process_query_response(
_StoreId, _RaServer, _IsLeader, _QueryFun, _QueryType, _Timeout,
{error, _} = Error) ->
Error.
-spec select_query_type(StoreId, Options) -> QueryType when
StoreId :: khepri:store_id(),
Options :: khepri:query_options(),
QueryType :: local | leader | consistent.
%% @doc Selects the query type depending on what the caller favors.
%%
%% @private
select_query_type(StoreId, #{favor := Favor}) ->
do_select_query_type(StoreId, Favor);
select_query_type(StoreId, _Options) ->
do_select_query_type(StoreId, compromise).
-define(
LAST_CONSISTENT_CALL_TS_REF(StoreId),
{khepri, last_consistent_call_ts_ref, StoreId}).
do_select_query_type(StoreId, compromise) ->
Key = ?LAST_CONSISTENT_CALL_TS_REF(StoreId),
Idx = 1,
case persistent_term:get(Key, undefined) of
AtomicsRef when AtomicsRef =/= undefined ->
%% We verify when was the last time we did a command or a
%% consistent query (i.e. we made sure there was an active leader
%% in a cluster with a quorum of active members).
%%
%% If the last one was more than 10 seconds ago, we force a
%% consistent query to verify the cluster health at the same time.
%% Otherwise, we select a leader query which is a good balance
%% between freshness and latency.
Last = atomics:get(AtomicsRef, Idx),
Now = erlang:system_time(second),
ConsistentAgainAfter = application:get_env(
khepri,
consistent_query_interval_in_compromise,
10),
if
Now - Last < ConsistentAgainAfter -> leader;
true -> consistent
end;
undefined ->
consistent
end;
do_select_query_type(_StoreId, consistency) ->
consistent;
do_select_query_type(_StoreId, low_latency) ->
local.
just_did_consistent_call(StoreId) ->
%% We record the timestamp of the successful command or consistent query
%% which just returned. This timestamp is used in the `compromise' query
%% strategy to perform a consistent query from time to time, and leader
%% queries the rest of the time.
%%
%% We store the system time as seconds in an `atomics' structure. The
%% reference of that structure is stored in a persistent term. We don't
%% store the timestamp directly in a persistent term because it is not
%% suited for frequent writes. This way, we store the `atomics' reference
%% once and update the `atomics' afterwards.
Idx = 1,
AtomicsRef = case get_last_consistent_call_atomics(StoreId) of
Ref when Ref =/= undefined ->
Ref;
undefined ->
Key = ?LAST_CONSISTENT_CALL_TS_REF(StoreId),
Ref = atomics:new(1, []),
persistent_term:put(Key, Ref),
Ref
end,
Now = erlang:system_time(second),
ok = atomics:put(AtomicsRef, Idx, Now),
ok.
get_last_consistent_call_atomics(StoreId) ->
Key = ?LAST_CONSISTENT_CALL_TS_REF(StoreId),
persistent_term:get(Key, undefined).
-spec get_timeout(Options) -> Timeout when
Options :: khepri:command_options() | khepri:query_options(),
Timeout :: timeout().
%% @private
get_timeout(#{timeout := Timeout}) -> Timeout;
get_timeout(_) -> khepri_app:get_default_timeout().
use_leader_or_local_ra_server(_StoreId, LeaderId)
when LeaderId =/= undefined ->
LeaderId;
use_leader_or_local_ra_server(StoreId, undefined) ->
ThisNode = node(),
khepri_cluster:node_to_member(StoreId, ThisNode).
-spec clear_cache(StoreId) -> ok when
StoreId :: khepri:store_id().
%% @doc Clears the cached data for the given `StoreId'.
%%
%% @private
clear_cache(StoreId) ->
_ = persistent_term:erase(?LAST_CONSISTENT_CALL_TS_REF(StoreId)),
ok.
%% -------------------------------------------------------------------
%% ra_machine callbacks.
%% -------------------------------------------------------------------
-spec init(Params) -> State when
Params :: machine_init_args(),
State :: state().
%% @private
init(#{store_id := StoreId,
member := Member} = Params) ->
Config = case Params of
#{snapshot_interval := SnapshotInterval} ->
#config{store_id = StoreId,
member = Member,
snapshot_interval = SnapshotInterval};
_ ->
#config{store_id = StoreId,
member = Member}
end,
State = #?MODULE{config = Config},
%% Create initial "schema" if provided.
Commands = maps:get(commands, Params, []),
State3 = lists:foldl(
fun (Command, State1) ->
Meta = #{index => 0,
term => 0,
system_time => 0},
{S, _, _} = apply(Meta, Command, State1),
S
end, State, Commands),
reset_applied_command_count(State3).
-spec init_aux(StoreId :: khepri:store_id()) -> aux_state().
%% @private
init_aux(StoreId) ->
#khepri_machine_aux{store_id = StoreId}.
-spec handle_aux(RaState, Type, Command, AuxState, LogState, MachineState) ->
{no_reply, AuxState, LogState} when
RaState :: ra_server:ra_state(),
Type :: {call, ra:from()} | cast,
Command :: term(),
AuxState :: aux_state(),
LogState :: ra_log:state(),
MachineState :: state().
%% @private
handle_aux(
_RaState, cast,
#trigger_projection{path = Path,
old_props = OldProps,
new_props = NewProps,
projection = Projection},
AuxState, LogState, _MachineState) ->
khepri_projection:trigger(Projection, Path, OldProps, NewProps),
{no_reply, AuxState, LogState};
handle_aux(_RaState, cast, restore_projections, AuxState, LogState,
#?MODULE{projections = Projections, root = Root}) ->
maps:foreach(fun(Projection, PathPattern) ->
restore_projection(
Projection, Root, PathPattern)
end, Projections),
{no_reply, AuxState, LogState};
handle_aux(_RaState, _Type, _Command, AuxState, LogState, _MachineState) ->
{no_reply, AuxState, LogState}.
restore_projection(Projection, Root, PathPattern) ->
TreeOptions = #{props_to_return => ?PROJECTION_PROPS_TO_RETURN,
include_root_props => true},
case find_matching_nodes(Root, PathPattern, TreeOptions) of
{ok, MatchingNodes} ->
maps:foreach(fun(Path, Props) ->
khepri_projection:trigger(
Projection, Path, #{}, Props)
end, MatchingNodes);
Error ->
?LOG_DEBUG(
"Failed to recover projection ~s due to an error: ~p",
khepri_projection:name(Projection), Error),
ok
end.
-spec apply(Meta, Command, State) -> {State, Ret, SideEffects} when
Meta :: ra_machine:command_meta_data(),
Command :: command(),
State :: state(),
Ret :: any(),
SideEffects :: ra_machine:effects().
%% @private
%% TODO: Handle unknown/invalid commands.
apply(
Meta,
#put{path = PathPattern, payload = Payload, options = TreeAndPutOptions},
State) ->
{TreeOptions, Extra} = split_put_options(TreeAndPutOptions),
Ret = insert_or_update_node(
State, PathPattern, Payload, Extra, TreeOptions),
bump_applied_command_count(Ret, Meta);
apply(
Meta,
#delete{path = PathPattern, options = TreeOptions},
State) ->
Ret = delete_matching_nodes(State, PathPattern, TreeOptions),
bump_applied_command_count(Ret, Meta);
apply(
Meta,
#tx{'fun' = StandaloneFun, args = Args},
State) when ?IS_STANDALONE_FUN(StandaloneFun) ->
Ret = khepri_tx_adv:run(State, StandaloneFun, Args, true),
bump_applied_command_count(Ret, Meta);
apply(
Meta,
#tx{'fun' = PathPattern, args = Args},
State) when ?IS_KHEPRI_PATH_PATTERN(PathPattern) ->
Ret = locate_sproc_and_execute_tx(State, PathPattern, Args, true),
bump_applied_command_count(Ret, Meta);
apply(
Meta,
#register_trigger{id = TriggerId,
sproc = StoredProcPath,
event_filter = EventFilter},
#?MODULE{triggers = Triggers} = State) ->
StoredProcPath1 = khepri_path:realpath(StoredProcPath),
EventFilter1 = case EventFilter of
#evf_tree{path = Path} ->
Path1 = khepri_path:realpath(Path),
EventFilter#evf_tree{path = Path1}
end,
Triggers1 = Triggers#{TriggerId => #{sproc => StoredProcPath1,
event_filter => EventFilter1}},
State1 = State#?MODULE{triggers = Triggers1},
Ret = {State1, ok},
bump_applied_command_count(Ret, Meta);
apply(
Meta,
#ack_triggered{triggered = ProcessedTriggers},
#?MODULE{emitted_triggers = EmittedTriggers} = State) ->
EmittedTriggers1 = EmittedTriggers -- ProcessedTriggers,
State1 = State#?MODULE{emitted_triggers = EmittedTriggers1},
Ret = {State1, ok},
bump_applied_command_count(Ret, Meta);
apply(
Meta,
#register_projection{pattern = PathPattern, projection = Projection},
#?MODULE{projections = Projections, root = Root} = State) ->
Reply = khepri_projection:init(Projection),
State1 = case Reply of
ok ->
restore_projection(Projection, Root, PathPattern),
Projections1 = Projections#{Projection => PathPattern},
State#?MODULE{projections = Projections1};
_ ->
State
end,
Ret = {State1, Reply},
bump_applied_command_count(Ret, Meta).
-spec bump_applied_command_count(ApplyRet, Meta) ->
{State, Ret, SideEffects} when
ApplyRet :: {State, Ret} | {State, Ret, SideEffects},
State :: state(),
Ret :: any(),
Meta :: ra_machine:command_meta_data(),
SideEffects :: ra_machine:effects().
%% @private
bump_applied_command_count({State, Result}, Meta) ->
bump_applied_command_count({State, Result, []}, Meta);
bump_applied_command_count(
{#?MODULE{config = #config{snapshot_interval = SnapshotInterval},
metrics = Metrics} = State,
Result,
SideEffects},
#{index := RaftIndex}) ->
AppliedCmdCount0 = maps:get(applied_command_count, Metrics, 0),
AppliedCmdCount = AppliedCmdCount0 + 1,
case AppliedCmdCount < SnapshotInterval of
true ->
Metrics1 = Metrics#{applied_command_count => AppliedCmdCount},
State1 = State#?MODULE{metrics = Metrics1},
{State1, Result, SideEffects};
false ->
?LOG_DEBUG(
"Move release cursor after ~b commands applied "
"(>= ~b commands)",
[AppliedCmdCount, SnapshotInterval],
#{domain => [khepri, ra_machine]}),
State1 = reset_applied_command_count(State),
ReleaseCursor = {release_cursor, RaftIndex, State1},
SideEffects1 = [ReleaseCursor | SideEffects],
{State1, Result, SideEffects1}
end.
reset_applied_command_count(#?MODULE{metrics = Metrics} = State) ->
Metrics1 = maps:remove(applied_command_count, Metrics),
State#?MODULE{metrics = Metrics1}.
%% @private
state_enter(leader, State) ->
SideEffects1 = emitted_triggers_to_side_effects(State),
SideEffects1;
state_enter(recovered, _State) ->
SideEffect = {aux, restore_projections},
[SideEffect];
state_enter(_StateName, _State) ->
[].
%% @private
emitted_triggers_to_side_effects(
#?MODULE{config = #config{store_id = StoreId},
emitted_triggers = [_ | _] = EmittedTriggers}) ->
SideEffect = {mod_call,
khepri_event_handler,
handle_triggered_sprocs,
[StoreId, EmittedTriggers]},
[SideEffect];
emitted_triggers_to_side_effects(_State) ->
[].
%% @private
overview(#?MODULE{config = #config{store_id = StoreId},
root = Root,
triggers = Triggers,
keep_while_conds = KeepWhileConds}) ->
TreeOptions = #{props_to_return => [payload,
payload_version,
child_list_version,
child_list_length],
include_root_props => true},
{ok, NodePropsMap} = find_matching_nodes(
Root, [?KHEPRI_WILDCARD_STAR_STAR], TreeOptions),
MapFun = fun (#{sproc := Sproc} = Props) ->
Props#{sproc => khepri_fun:to_fun(Sproc)};
(Props) ->
Props
end,
Tree = khepri_utils:flat_struct_to_tree(NodePropsMap, MapFun),
#{store_id => StoreId,
tree => Tree,
triggers => Triggers,
keep_while_conds => KeepWhileConds}.
%% -------------------------------------------------------------------
%% Internal functions.
%% -------------------------------------------------------------------
-spec create_node_record(Payload) -> Node when
Payload :: khepri_payload:payload(),
Node :: tree_node().
%% @private
create_node_record(Payload) ->
#node{props = ?INIT_NODE_PROPS,
payload = Payload}.
-spec set_node_payload(Node, Payload) -> Node when
Node :: tree_node(),
Payload :: khepri_payload:payload().
%% @private
set_node_payload(#node{payload = Payload} = Node, Payload) ->
Node;
set_node_payload(#node{props = #{payload_version := DVersion} = Stat} = Node,
Payload) ->
Stat1 = Stat#{payload_version => DVersion + 1},
Node#node{props = Stat1, payload = Payload}.
-spec remove_node_payload(Node) -> Node when
Node :: tree_node().
%% @private
remove_node_payload(
#node{payload = ?NO_PAYLOAD} = Node) ->
Node;
remove_node_payload(
#node{props = #{payload_version := DVersion} = Stat} = Node) ->
Stat1 = Stat#{payload_version => DVersion + 1},
Node#node{props = Stat1, payload = khepri_payload:none()}.
-spec add_node_child(Node, ChildName, Child) -> Node when
Node :: tree_node(),
Child :: tree_node(),
ChildName :: khepri_path:component().
add_node_child(#node{props = #{child_list_version := CVersion} = Stat,
child_nodes = Children} = Node,
ChildName, Child) ->
Children1 = Children#{ChildName => Child},
Stat1 = Stat#{child_list_version => CVersion + 1},
Node#node{props = Stat1, child_nodes = Children1}.
-spec update_node_child(Node, ChildName, Child) -> Node when
Node :: tree_node(),
Child :: tree_node(),
ChildName :: khepri_path:component().
update_node_child(#node{child_nodes = Children} = Node, ChildName, Child) ->
Children1 = Children#{ChildName => Child},
Node#node{child_nodes = Children1}.
-spec remove_node_child(Node, ChildName) -> Node when
Node :: tree_node(),
ChildName :: khepri_path:component().
remove_node_child(#node{props = #{child_list_version := CVersion} = Stat,
child_nodes = Children} = Node,
ChildName) ->
?assert(maps:is_key(ChildName, Children)),
Stat1 = Stat#{child_list_version => CVersion + 1},
Children1 = maps:remove(ChildName, Children),
Node#node{props = Stat1, child_nodes = Children1}.
-spec remove_node_child_nodes(Node) -> Node when
Node :: tree_node().
remove_node_child_nodes(
#node{child_nodes = Children} = Node) when Children =:= #{} ->
Node;
remove_node_child_nodes(
#node{props = #{child_list_version := CVersion} = Stat} = Node) ->
Stat1 = Stat#{child_list_version => CVersion + 1},
Node#node{props = Stat1, child_nodes = #{}}.
-spec gather_node_props(Node, TreeOptions) -> NodeProps when
Node :: tree_node(),
TreeOptions :: khepri:tree_options(),
NodeProps :: khepri:node_props().
gather_node_props(#node{props = #{payload_version := PVersion,
child_list_version := CVersion},
payload = Payload,
child_nodes = Children},
#{props_to_return := WantedProps}) ->
lists:foldl(
fun
(payload_version, Acc) ->
Acc#{payload_version => PVersion};
(child_list_version, Acc) ->
Acc#{child_list_version => CVersion};
(child_list_length, Acc) ->
Acc#{child_list_length => maps:size(Children)};
(child_names, Acc) ->
Acc#{child_names => maps:keys(Children)};
(payload, Acc) ->
case Payload of
#p_data{data = Data} -> Acc#{data => Data};
#p_sproc{sproc = Fun} -> Acc#{sproc => Fun};
_ -> Acc
end;
(has_payload, Acc) ->
case Payload of
#p_data{data = _} -> Acc#{has_data => true};
#p_sproc{sproc = _} -> Acc#{is_sproc => true};
_ -> Acc
end;
(raw_payload, Acc) ->
Acc#{raw_payload => Payload}
end, #{}, WantedProps);
gather_node_props(#node{}, _Options) ->
#{}.
gather_node_props_from_old_and_new_nodes(OldNode, NewNode, TreeOptions) ->
OldNodeProps = case OldNode of
undefined -> #{};
_ -> gather_node_props(OldNode, TreeOptions)
end,
NewNodeProps0 = gather_node_props(NewNode, TreeOptions),
NewNodeProps1 = maps:remove(data, NewNodeProps0),
NewNodeProps2 = maps:remove(sproc, NewNodeProps1),
maps:merge(OldNodeProps, NewNodeProps2).
gather_node_props_for_error(Node) ->
gather_node_props(Node, #{props_to_return => ?DEFAULT_PROPS_TO_RETURN}).
-spec to_absolute_keep_while(BasePath, KeepWhile) -> KeepWhile when
BasePath :: khepri_path:native_path(),
KeepWhile :: khepri_condition:native_keep_while().
%% @private
to_absolute_keep_while(BasePath, KeepWhile) ->
maps:fold(
fun(Path, Cond, Acc) ->
AbsPath = khepri_path:abspath(Path, BasePath),
Acc#{AbsPath => Cond}
end, #{}, KeepWhile).
-spec are_keep_while_conditions_met(Node, KeepWhile) -> Ret when
Node :: tree_node(),
KeepWhile :: khepri_condition:native_keep_while(),
Ret :: true | {false, any()}.
%% @private
are_keep_while_conditions_met(_, KeepWhile)
when KeepWhile =:= #{} ->
true;
are_keep_while_conditions_met(Root, KeepWhile) ->
TreeOptions = #{props_to_return => [payload,
payload_version,
child_list_version,
child_list_length]},
maps:fold(
fun
(Path, Condition, true) ->
case find_matching_nodes(Root, Path, TreeOptions) of
{ok, Result} when Result =/= #{} ->
are_keep_while_conditions_met1(Result, Condition);
{ok, _} ->
{false, {pattern_matches_no_nodes, Path}};
{error, Reason} ->
{false, Reason}
end;
(_, _, False) ->
False
end, true, KeepWhile).
are_keep_while_conditions_met1(Result, Condition) ->
maps:fold(
fun
(Path, NodeProps, true) ->
khepri_condition:is_met(Condition, Path, NodeProps);
(_, _, False) ->
False
end, true, Result).
is_keep_while_condition_met_on_self(
Path, Node, #{keep_while_conds := KeepWhileConds}) ->
case KeepWhileConds of
#{Path := #{Path := Condition}} ->
khepri_condition:is_met(Condition, Path, Node);
_ ->
true
end;
is_keep_while_condition_met_on_self(_, _, _) ->
true.
update_keep_while_conds_extra(
#{keep_while_conds := KeepWhileConds,
keep_while_conds_revidx := KeepWhileCondsRevIdx} = Extra,
Path, KeepWhile) ->
AbsKeepWhile = to_absolute_keep_while(Path, KeepWhile),
KeepWhileCondsRevIdx1 = update_keep_while_conds_revidx(
KeepWhileConds, KeepWhileCondsRevIdx,
Path, AbsKeepWhile),
KeepWhileConds1 = KeepWhileConds#{Path => AbsKeepWhile},
Extra#{keep_while_conds => KeepWhileConds1,
keep_while_conds_revidx => KeepWhileCondsRevIdx1}.
-spec update_keep_while_conds_revidx(
KeepWhileConds, KeepWhileCondsRevIdx, Watcher, KeepWhile) ->
KeepWhileConds when
KeepWhileConds :: keep_while_conds_map(),
KeepWhileCondsRevIdx :: keep_while_conds_revidx(),
Watcher :: khepri_path:native_path(),
KeepWhile :: khepri_condition:native_keep_while().
update_keep_while_conds_revidx(
KeepWhileConds, KeepWhileCondsRevIdx, Watcher, KeepWhile) ->
%% First, clean up reversed index where a watched path isn't watched
%% anymore in the new keep_while.
OldWatcheds = maps:get(Watcher, KeepWhileConds, #{}),
KeepWhileCondsRevIdx1 = maps:fold(
fun(Watched, _, KWRevIdx) ->
Watchers = maps:get(Watched, KWRevIdx),
Watchers1 = maps:remove(Watcher, Watchers),
case maps:size(Watchers1) of
0 -> maps:remove(Watched, KWRevIdx);
_ -> KWRevIdx#{Watched => Watchers1}
end
end, KeepWhileCondsRevIdx, OldWatcheds),
%% Then, record the watched paths.
maps:fold(
fun(Watched, _, KWRevIdx) ->
Watchers = maps:get(Watched, KWRevIdx, #{}),
Watchers1 = Watchers#{Watcher => ok},
KWRevIdx#{Watched => Watchers1}
end, KeepWhileCondsRevIdx1, KeepWhile).
locate_sproc_and_execute_tx(
#?MODULE{root = Root} = State, PathPattern, Args, AllowUpdates) ->
TreeOptions = #{expect_specific_node => true,
props_to_return => [raw_payload]},
{StandaloneFun, Args1} =
case find_matching_nodes(Root, PathPattern, TreeOptions) of
{ok, Result} ->
case maps:values(Result) of
[#{raw_payload := #p_sproc{
sproc = StoredProc,
is_valid_as_tx_fun = ReadWrite}}]
when AllowUpdates andalso
(ReadWrite =:= ro orelse ReadWrite =:= rw) ->
{StoredProc, Args};
[#{raw_payload := #p_sproc{
sproc = StoredProc,
is_valid_as_tx_fun = ro}}]
when not AllowUpdates ->
{StoredProc, Args};
[#{raw_payload := #p_sproc{}}] ->
Reason = ?khepri_error(
sproc_invalid_as_tx_fun,
#{path => PathPattern}),
{fun failed_to_locate_sproc/1, [Reason]};
_ ->
Reason = ?khepri_error(
no_sproc_at_given_path,
#{path => PathPattern}),
{fun failed_to_locate_sproc/1, [Reason]}
end;
{error, Reason} ->
{fun failed_to_locate_sproc/1, [Reason]}
end,
khepri_tx_adv:run(State, StandaloneFun, Args1, AllowUpdates).
-spec failed_to_locate_sproc(Reason) -> no_return() when
Reason :: any().
%% @private
failed_to_locate_sproc(Reason) ->
khepri_tx:abort(Reason).
-spec find_matching_nodes(Root, PathPattern, TreeOptions) ->
Ret when
Root :: tree_node(),
PathPattern :: khepri_path:native_pattern(),
TreeOptions :: khepri:tree_options(),
Ret :: khepri_machine:common_ret().
%% @private
find_matching_nodes(Root, PathPattern, TreeOptions) ->
find_matching_nodes(
Root, PathPattern,
fun collect_node_props_cb/3, #{},
TreeOptions).
-spec collect_node_props_cb(Path, NodeProps, Map) ->
Ret when
Path :: khepri_path:native_path(),
NodeProps :: khepri:node_props(),
Map :: khepri_adv:node_props_map(),
Ret :: Map.
%% @private
collect_node_props_cb(Path, NodeProps, Map) when is_map(Map) ->
Map#{Path => NodeProps}.
-spec count_node_cb(Path, NodeProps, Count) ->
Ret when
Path :: khepri_path:native_path(),
NodeProps :: khepri:node_props(),
Count :: non_neg_integer(),
Ret :: Count.
%% @private
count_node_cb(_Path, _NodeProps, Count) when is_integer(Count) ->
Count + 1.
-spec find_matching_nodes(Root, PathPattern, Fun, Acc, TreeOptions) ->
Ret when
Root :: tree_node(),
PathPattern :: khepri_path:native_pattern(),
Fun :: khepri:fold_fun(),
Acc :: khepri:fold_acc(),
TreeOptions :: khepri:tree_options(),
Ret :: khepri:ok(Acc) | khepri:error().
%% @private
find_matching_nodes(Root, PathPattern, Fun, Acc, TreeOptions) ->
WalkFun = fun(Path, Node, Acc1) ->
find_matching_nodes_cb(
Path, Node, Fun, Acc1, TreeOptions)
end,
Ret = walk_down_the_tree(
Root, PathPattern, TreeOptions, #{}, WalkFun, Acc),
case Ret of
{ok, NewRoot, _, Acc2} ->
?assertEqual(Root, NewRoot),
{ok, Acc2};
Error ->
Error
end.
find_matching_nodes_cb(Path, #node{} = Node, Fun, Acc, TreeOptions) ->
NodeProps = gather_node_props(Node, TreeOptions),
Acc1 = Fun(Path, NodeProps, Acc),
{ok, keep, Acc1};
find_matching_nodes_cb(
_,
{interrupted, node_not_found = Reason, Info},
_Fun, _Acc,
#{expect_specific_node := true}) ->
%% If we are collecting node properties (the result is a map) and the path
%% targets a specific node which is not found, we return an error.
%%
%% If we are counting nodes, that's fine and the next function clause will
%% run. The walk won't be interrupted.
Reason1 = ?khepri_error(Reason, Info),
{error, Reason1};
find_matching_nodes_cb(_, {interrupted, _, _}, _, Acc, _) ->
{ok, keep, Acc}.
-spec insert_or_update_node(State, PathPattern, Payload, Extra, TreeOptions) ->
Ret when
State :: state(),
PathPattern :: khepri_path:native_pattern(),
Payload :: khepri_payload:payload(),
Extra :: khepri:put_options(),
TreeOptions :: khepri:tree_options(),
Ret :: {State, Result} | {State, Result, ra_machine:effects()},
Result :: khepri_machine:common_ret().
%% @private
insert_or_update_node(
#?MODULE{root = Root,
keep_while_conds = KeepWhileConds,
keep_while_conds_revidx = KeepWhileCondsRevIdx} = State,
PathPattern, Payload,
#{keep_while := KeepWhile},
TreeOptions) ->
Fun = fun(Path, Node, {_, _, Result}) ->
Ret = insert_or_update_node_cb(
Path, Node, Payload, TreeOptions, Result),
case Ret of
{ok, Node1, Result1} when Result1 =/= #{} ->
AbsKeepWhile = to_absolute_keep_while(
Path, KeepWhile),
KeepWhileOnOthers = maps:remove(Path, AbsKeepWhile),
KWMet = are_keep_while_conditions_met(
Root, KeepWhileOnOthers),
case KWMet of
true ->
{ok, Node1, {updated, Path, Result1}};
{false, Reason} ->
%% The keep_while condition is not met. We
%% can't insert the node and return an
%% error instead.
NodeName = case Path of
[] -> ?KHEPRI_ROOT_NODE;
_ -> lists:last(Path)
end,
Reason1 = ?khepri_error(
keep_while_conditions_not_met,
#{node_name => NodeName,
node_path => Path,
keep_while_reason => Reason}),
{error, Reason1}
end;
{ok, Node1, Result1} ->
{ok, Node1, {updated, Path, Result1}};
Error ->
Error
end
end,
Ret1 = walk_down_the_tree(
Root, PathPattern, TreeOptions,
#{keep_while_conds => KeepWhileConds,
keep_while_conds_revidx => KeepWhileCondsRevIdx,
applied_changes => #{}},
Fun, {undefined, [], #{}}),
case Ret1 of
{ok, Root1, #{applied_changes := AppliedChanges} = Extra,
{updated, ResolvedPath, Ret2}} ->
#{keep_while_conds := KeepWhileConds1,
keep_while_conds_revidx := KeepWhileCondsRevIdx1} =
update_keep_while_conds_extra(Extra, ResolvedPath, KeepWhile),
State1 = State#?MODULE{
root = Root1,
keep_while_conds = KeepWhileConds1,
keep_while_conds_revidx = KeepWhileCondsRevIdx1},
{State2, SideEffects} = create_tree_change_side_effects(
State, State1, Ret2, AppliedChanges),
{State2, {ok, Ret2}, SideEffects};
Error ->
?assertMatch({error, _}, Error),
{State, Error}
end;
insert_or_update_node(
#?MODULE{root = Root,
keep_while_conds = KeepWhileConds,
keep_while_conds_revidx = KeepWhileCondsRevIdx} = State,
PathPattern, Payload,
_Extra, TreeOptions) ->
Fun = fun(Path, Node, Result) ->
insert_or_update_node_cb(
Path, Node, Payload, TreeOptions, Result)
end,
Ret1 = walk_down_the_tree(
Root, PathPattern, TreeOptions,
#{keep_while_conds => KeepWhileConds,
keep_while_conds_revidx => KeepWhileCondsRevIdx,
applied_changes => #{}},
Fun, #{}),
case Ret1 of
{ok, Root1, #{keep_while_conds := KeepWhileConds1,
keep_while_conds_revidx := KeepWhileCondsRevIdx1,
applied_changes := AppliedChanges},
Ret2} ->
State1 = State#?MODULE{root = Root1,
keep_while_conds = KeepWhileConds1,
keep_while_conds_revidx =
KeepWhileCondsRevIdx1},
{State2, SideEffects} = create_tree_change_side_effects(
State, State1, Ret2, AppliedChanges),
{State2, {ok, Ret2}, SideEffects};
Error ->
?assertMatch({error, _}, Error),
{State, Error}
end.
insert_or_update_node_cb(
Path, #node{} = Node, Payload, TreeOptions, Result) ->
case maps:is_key(Path, Result) of
false ->
%% After a node is modified, we collect properties from the updated
%% `#node{}', except the payload which is from the old one.
Node1 = set_node_payload(Node, Payload),
NodeProps = gather_node_props_from_old_and_new_nodes(
Node, Node1, TreeOptions),
{ok, Node1, Result#{Path => NodeProps}};
true ->
{ok, Node, Result}
end;
insert_or_update_node_cb(
Path, {interrupted, node_not_found = Reason, Info}, Payload, TreeOptions,
Result) ->
%% We store the payload when we reached the target node only, not in the
%% parent nodes we have to create in between.
IsTarget = maps:get(node_is_target, Info),
case can_continue_update_after_node_not_found(Info) of
true when IsTarget ->
Node = create_node_record(Payload),
NodeProps = gather_node_props_from_old_and_new_nodes(
undefined, Node, TreeOptions),
{ok, Node, Result#{Path => NodeProps}};
true ->
Node = create_node_record(khepri_payload:none()),
{ok, Node, Result};
false ->
Reason1 = ?khepri_error(Reason, Info),
{error, Reason1}
end;
insert_or_update_node_cb(_, {interrupted, Reason, Info}, _, _, _) ->
Reason1 = ?khepri_error(Reason, Info),
{error, Reason1}.
can_continue_update_after_node_not_found(#{condition := Condition}) ->
can_continue_update_after_node_not_found1(Condition);
can_continue_update_after_node_not_found(#{node_name := NodeName}) ->
can_continue_update_after_node_not_found1(NodeName).
can_continue_update_after_node_not_found1(ChildName)
when ?IS_KHEPRI_PATH_COMPONENT(ChildName) ->
true;
can_continue_update_after_node_not_found1(#if_node_exists{exists = false}) ->
true;
can_continue_update_after_node_not_found1(#if_all{conditions = Conds}) ->
lists:all(fun can_continue_update_after_node_not_found1/1, Conds);
can_continue_update_after_node_not_found1(#if_any{conditions = Conds}) ->
lists:any(fun can_continue_update_after_node_not_found1/1, Conds);
can_continue_update_after_node_not_found1(_) ->
false.
-spec delete_matching_nodes(State, PathPattern, TreeOptions) -> Ret when
State :: state(),
PathPattern :: khepri_path:native_pattern(),
TreeOptions :: khepri:tree_options(),
Ret :: {State, Result} | {State, Result, ra_machine:effects()},
Result :: khepri_machine:common_ret().
%% @private
delete_matching_nodes(
#?MODULE{root = Root,
keep_while_conds = KeepWhileConds,
keep_while_conds_revidx = KeepWhileCondsRevIdx} = State,
PathPattern,
TreeOptions) ->
Ret1 = do_delete_matching_nodes(
PathPattern, Root,
#{keep_while_conds => KeepWhileConds,
keep_while_conds_revidx => KeepWhileCondsRevIdx,
applied_changes => #{}},
TreeOptions),
case Ret1 of
{ok, Root1, #{keep_while_conds := KeepWhileConds1,
keep_while_conds_revidx := KeepWhileCondsRevIdx1,
applied_changes := AppliedChanges},
Ret2} ->
State1 = State#?MODULE{
root = Root1,
keep_while_conds = KeepWhileConds1,
keep_while_conds_revidx = KeepWhileCondsRevIdx1},
{State2, SideEffects} = create_tree_change_side_effects(
State, State1, Ret2, AppliedChanges),
{State2, {ok, Ret2}, SideEffects};
Error ->
{State, Error}
end.
do_delete_matching_nodes(PathPattern, Root, Extra, TreeOptions) ->
Fun = fun(Path, Node, Result) ->
delete_matching_nodes_cb(Path, Node, TreeOptions, Result)
end,
walk_down_the_tree(Root, PathPattern, TreeOptions, Extra, Fun, #{}).
delete_matching_nodes_cb([] = Path, #node{} = Node, TreeOptions, Result) ->
Node1 = remove_node_payload(Node),
Node2 = remove_node_child_nodes(Node1),
NodeProps = gather_node_props(Node, TreeOptions),
{ok, Node2, Result#{Path => NodeProps}};
delete_matching_nodes_cb(Path, #node{} = Node, TreeOptions, Result) ->
NodeProps = gather_node_props(Node, TreeOptions),
{ok, delete, Result#{Path => NodeProps}};
delete_matching_nodes_cb(_, {interrupted, _, _}, _Options, Result) ->
{ok, keep, Result}.
create_tree_change_side_effects(
InitialState, NewState, Ret, KeepWhileAftermath) ->
%% We make a map where for each affected tree node, we indicate the type
%% of change.
Changes0 = maps:merge(Ret, KeepWhileAftermath),
Changes = maps:map(
fun
(_, NodeProps) when NodeProps =:= #{} -> create;
(_, #{} = _NodeProps) -> update;
(_, delete) -> delete
end, Changes0),
ProjectionEffects = create_projection_side_effects(
InitialState, NewState, Changes),
{NewState1, TriggerEffects} = create_trigger_side_effects(
InitialState, NewState, Changes),
{NewState1, ProjectionEffects ++ TriggerEffects}.
create_projection_side_effects(
#?MODULE{root = InitialRoot} = _InitialState,
#?MODULE{projections = Projections, root = NewRoot} = _NewState,
Changes) ->
%% Note: the order in which updates are applied to projections should not
%% be relied on.
Effects =
maps:fold(
fun(Path, Change, Effects) ->
maps:fold(
fun(Projection, Pattern, Effects1) ->
evaluate_projection(
InitialRoot, NewRoot,
Path, Change, Pattern, Projection, Effects1)
end, Effects, Projections)
end, [], Changes),
lists:reverse(Effects).
evaluate_projection(
InitialRoot, NewRoot, Path, Change, Pattern, Projection, Effects) ->
PathMatchingRoot = case Change of
Put when Put =:= create orelse Put =:= update ->
NewRoot;
delete ->
InitialRoot
end,
case does_path_match(Path, Pattern, [], PathMatchingRoot) of
true ->
FindOptions = #{props_to_return => ?PROJECTION_PROPS_TO_RETURN,
expect_specific_node => true},
InitialRet = find_matching_nodes(InitialRoot, Path, FindOptions),
InitialProps = case InitialRet of
{ok, #{Path := InitialProps0}} ->
InitialProps0;
_ ->
#{}
end,
NewRet = find_matching_nodes(NewRoot, Path, FindOptions),
NewProps = case NewRet of
{ok, #{Path := NewProps0}} ->
NewProps0;
_ ->
#{}
end,
Trigger = #trigger_projection{path = Path,
old_props = InitialProps,
new_props = NewProps,
projection = Projection},
Effect = {aux, Trigger},
[Effect | Effects];
false ->
Effects
end.
create_trigger_side_effects(
#?MODULE{triggers = Triggers} = _InitialState, NewState, _Changes)
when Triggers =:= #{} ->
{NewState, []};
create_trigger_side_effects(
%% We want to consider the new state (with the updated tree), but we want
%% to use triggers from the initial state, in case they were updated too.
%% In other words, we want to evaluate triggers in the state they were at
%% the time the change to the tree was requested.
#?MODULE{triggers = Triggers,
emitted_triggers = EmittedTriggers} = _InitialState,
#?MODULE{config = #config{store_id = StoreId},
root = Root} = NewState,
Changes) ->
TriggeredStoredProcs = list_triggered_sprocs(Root, Changes, Triggers),
%% We record the list of triggered stored procedures in the state
%% machine's state. This is used to guaranty at-least-once execution of
%% the trigger: the event handler process is supposed to ack when it
%% executed the triggered stored procedure. If the Ra cluster changes
%% leader in between, we know that we need to retry the execution.
%%
%% This could lead to multiple execution of the same trigger, therefore
%% the stored procedure must be idempotent.
NewState1 = NewState#?MODULE{
emitted_triggers =
EmittedTriggers ++ TriggeredStoredProcs},
%% We still emit a `mod_call' effect to wake up the event handler process
%% so it doesn't have to poll the internal list.
SideEffect = {mod_call,
khepri_event_handler,
handle_triggered_sprocs,
[StoreId, TriggeredStoredProcs]},
{NewState1, [SideEffect]}.
list_triggered_sprocs(Root, Changes, Triggers) ->
TriggeredStoredProcs =
maps:fold(
fun(Path, Change, TSP) ->
% For each change, we evaluate each trigger.
maps:fold(
fun(TriggerId, TriggerProps, TSP1) ->
evaluate_trigger(
Root, Path, Change, TriggerId, TriggerProps, TSP1)
end, TSP, Triggers)
end, [], Changes),
sort_triggered_sprocs(TriggeredStoredProcs).
evaluate_trigger(
Root, Path, Change, TriggerId,
#{sproc := StoredProcPath,
event_filter := #evf_tree{path = PathPattern,
props = EventFilterProps} = EventFilter},
TriggeredStoredProcs) ->
%% For each trigger based on a tree event:
%% 1. we verify the path of the changed tree node matches the monitored
%% path pattern in the event filter.
%% 2. we verify the type of change matches the change filter in the
%% event filter.
PathMatches = does_path_match(Path, PathPattern, [], Root),
DefaultWatchedChanges = [create, update, delete],
WatchedChanges = case EventFilterProps of
#{on_actions := []} ->
DefaultWatchedChanges;
#{on_actions := OnActions}
when is_list(OnActions) ->
OnActions;
_ ->
DefaultWatchedChanges
end,
ChangeMatches = lists:member(Change, WatchedChanges),
case PathMatches andalso ChangeMatches of
true ->
%% We then locate the stored procedure. If the path doesn't point
%% to an existing tree node, or if this tree node is not a stored
%% procedure, the trigger is ignored.
%%
%% TODO: Should we return an error or at least log something? This
%% could be considered noise if the trigger exists regardless of
%% the presence of the stored procedure on purpose (for instance
%% the caller code is being updated).
case find_stored_proc(Root, StoredProcPath) of
undefined ->
TriggeredStoredProcs;
StoredProc ->
%% TODO: Use a record to format
%% stored procedure arguments?
EventProps = #{path => Path,
on_action => Change},
Triggered = #triggered{
id = TriggerId,
event_filter = EventFilter,
sproc = StoredProc,
props = EventProps},
[Triggered | TriggeredStoredProcs]
end;
false ->
TriggeredStoredProcs
end;
evaluate_trigger(
_Root, _Path, _Change, _TriggerId, _TriggerProps, TriggeredStoredProcs) ->
TriggeredStoredProcs.
does_path_match(PathRest, PathRest, _ReversedPath, _Root) ->
true;
does_path_match([], _PathPatternRest, _ReversedPath, _Root) ->
false;
does_path_match(_PathRest, [], _ReversedPath, _Root) ->
false;
does_path_match(
[Component | Path], [Component | PathPattern], ReversedPath, Root)
when ?IS_KHEPRI_PATH_COMPONENT(Component) ->
does_path_match(Path, PathPattern, [Component | ReversedPath], Root);
does_path_match(
[Component | _Path], [Condition | _PathPattern], _ReversedPath, _Root)
when ?IS_KHEPRI_PATH_COMPONENT(Component) andalso
?IS_KHEPRI_PATH_COMPONENT(Condition) ->
false;
does_path_match(
[Component | Path], [Condition | PathPattern], ReversedPath, Root) ->
%% Query the tree node, required to evaluate the condition.
ReversedPath1 = [Component | ReversedPath],
CurrentPath = lists:reverse(ReversedPath1),
TreeOptions = #{expect_specific_node => true,
props_to_return => [payload,
payload_version,
child_list_version,
child_list_length]},
{ok, #{CurrentPath := Node}} = find_matching_nodes(
Root,
lists:reverse([Component | ReversedPath]),
TreeOptions),
case khepri_condition:is_met(Condition, Component, Node) of
true ->
ConditionMatchesGrandchildren =
case khepri_condition:applies_to_grandchildren(Condition) of
true ->
does_path_match(
Path, [Condition | PathPattern], ReversedPath1, Root);
false ->
false
end,
ConditionMatchesGrandchildren orelse
does_path_match(Path, PathPattern, ReversedPath1, Root);
{false, _} ->
false
end.
find_stored_proc(Root, StoredProcPath) ->
TreeOptions = #{expect_specific_node => true,
props_to_return => [payload,
payload_version,
child_list_version,
child_list_length]},
Ret = find_matching_nodes(
Root, StoredProcPath, TreeOptions),
%% Non-existing nodes and nodes which are not stored procedures are
%% ignored.
case Ret of
{ok, #{StoredProcPath := #{sproc := StoredProc}}} -> StoredProc;
_ -> undefined
end.
sort_triggered_sprocs(TriggeredStoredProcs) ->
%% We first sort by priority, then by triggered ID if priorities are equal.
%% The priority can be any integer (even negative integers). The default
%% priority is 0.
%%
%% A higher priority (a greater integer) means that the triggered stored
%% procedure will be executed before another one with lower priority
%% (smaller integer).
%%
%% If the priorities are equal, a trigger with an ID earlier in
%% alphabetical order will be executed before another one with an ID later
%% in alphabetical order.
lists:sort(
fun(#triggered{id = IdA, event_filter = EventFilterA},
#triggered{id = IdB, event_filter = EventFilterB}) ->
PrioA = khepri_evf:get_priority(EventFilterA),
PrioB = khepri_evf:get_priority(EventFilterB),
if
PrioA =:= PrioB -> IdA =< IdB;
true -> PrioA > PrioB
end
end,
TriggeredStoredProcs).
%% -------
-spec walk_down_the_tree(
Root, PathPattern, TreeOptions, Extra, Fun, FunAcc) -> Ret when
Root :: tree_node(),
PathPattern :: khepri_path:native_pattern(),
TreeOptions :: khepri:tree_options(),
Extra :: walk_down_the_tree_extra(),
Fun :: walk_down_the_tree_fun(),
FunAcc :: any(),
Node :: tree_node(),
Ret :: ok(Node, Extra, FunAcc) | khepri:error().
%% @private
walk_down_the_tree(Root, PathPattern, TreeOptions, Extra, Fun, FunAcc) ->
CompiledPathPattern = khepri_path:compile(PathPattern),
TreeOptions1 = case TreeOptions of
#{expect_specific_node := true} ->
TreeOptions;
_ ->
TreeOptions#{expect_specific_node => false}
end,
walk_down_the_tree1(
Root, CompiledPathPattern, TreeOptions1,
[], %% Used to remember the path of the node currently on.
[], %% Used to update parents up in the tree in a tail-recursive
%% function.
Extra, Fun, FunAcc).
-spec walk_down_the_tree1(
Root, CompiledPathPattern, TreeOptions,
ReversedPath, ReversedParentTree, Extra, Fun, FunAcc) -> Ret when
Root :: tree_node(),
CompiledPathPattern :: khepri_path:native_pattern(),
TreeOptions :: khepri:tree_options(),
ReversedPath :: khepri_path:native_pattern(),
ReversedParentTree :: [Node | {Node, child_created}],
Extra :: walk_down_the_tree_extra(),
Fun :: walk_down_the_tree_fun(),
FunAcc :: any(),
Node :: tree_node(),
Ret :: ok(Node, Extra, FunAcc) | khepri:error().
%% @private
walk_down_the_tree1(
CurrentNode,
[?KHEPRI_ROOT_NODE | PathPattern],
TreeOptions, ReversedPath, ReversedParentTree, Extra, Fun, FunAcc) ->
?assertEqual([], ReversedPath),
?assertEqual([], ReversedParentTree),
walk_down_the_tree1(
CurrentNode, PathPattern, TreeOptions,
ReversedPath,
ReversedParentTree,
Extra, Fun, FunAcc);
walk_down_the_tree1(
CurrentNode,
[?THIS_KHEPRI_NODE | PathPattern],
TreeOptions, ReversedPath, ReversedParentTree, Extra, Fun, FunAcc) ->
walk_down_the_tree1(
CurrentNode, PathPattern, TreeOptions,
ReversedPath, ReversedParentTree, Extra, Fun, FunAcc);
walk_down_the_tree1(
_CurrentNode,
[?PARENT_KHEPRI_NODE | PathPattern],
TreeOptions,
[_CurrentName | ReversedPath], [ParentNode0 | ReversedParentTree],
Extra, Fun, FunAcc) ->
ParentNode = case ParentNode0 of
{PN, child_created} -> PN;
_ -> ParentNode0
end,
walk_down_the_tree1(
ParentNode, PathPattern, TreeOptions,
ReversedPath, ReversedParentTree, Extra, Fun, FunAcc);
walk_down_the_tree1(
CurrentNode,
[?PARENT_KHEPRI_NODE | PathPattern],
TreeOptions,
[] = ReversedPath, [] = ReversedParentTree,
Extra, Fun, FunAcc) ->
%% The path tries to go above the root node, like "cd /..". In this case,
%% we stay on the root node.
walk_down_the_tree1(
CurrentNode, PathPattern, TreeOptions,
ReversedPath, ReversedParentTree, Extra, Fun, FunAcc);
walk_down_the_tree1(
#node{child_nodes = Children} = CurrentNode,
[ChildName | PathPattern],
TreeOptions, ReversedPath, ReversedParentTree, Extra, Fun, FunAcc)
when ?IS_KHEPRI_NODE_ID(ChildName) ->
case Children of
#{ChildName := Child} ->
walk_down_the_tree1(
Child, PathPattern, TreeOptions,
[ChildName | ReversedPath],
[CurrentNode | ReversedParentTree],
Extra, Fun, FunAcc);
_ ->
interrupted_walk_down(
node_not_found,
#{node_name => ChildName,
node_path => lists:reverse([ChildName | ReversedPath])},
PathPattern, TreeOptions,
[ChildName | ReversedPath],
[CurrentNode | ReversedParentTree],
Extra, Fun, FunAcc)
end;
walk_down_the_tree1(
#node{child_nodes = Children} = CurrentNode,
[Condition | PathPattern], #{expect_specific_node := true} = TreeOptions,
ReversedPath, ReversedParentTree, Extra, Fun, FunAcc)
when ?IS_KHEPRI_CONDITION(Condition) ->
%% We distinguish the case where the condition must be verified against the
%% current node (i.e. the node name is ?KHEPRI_ROOT_NODE or
%% ?THIS_KHEPRI_NODE in the condition) instead of its child nodes.
SpecificNode = khepri_path:component_targets_specific_node(Condition),
case SpecificNode of
{true, NodeName}
when NodeName =:= ?KHEPRI_ROOT_NODE orelse
NodeName =:= ?THIS_KHEPRI_NODE ->
CurrentName = special_component_to_node_name(
NodeName, ReversedPath),
CondMet = khepri_condition:is_met(
Condition, CurrentName, CurrentNode),
case CondMet of
true ->
walk_down_the_tree1(
CurrentNode, PathPattern, TreeOptions,
ReversedPath, ReversedParentTree,
Extra, Fun, FunAcc);
{false, Cond} ->
interrupted_walk_down(
mismatching_node,
#{node_name => CurrentName,
node_path => lists:reverse(ReversedPath),
node_props => gather_node_props_for_error(CurrentNode),
condition => Cond},
PathPattern, TreeOptions, ReversedPath,
ReversedParentTree, Extra, Fun, FunAcc)
end;
{true, ChildName} when ChildName =/= ?PARENT_KHEPRI_NODE ->
case Children of
#{ChildName := Child} ->
CondMet = khepri_condition:is_met(
Condition, ChildName, Child),
case CondMet of
true ->
walk_down_the_tree1(
Child, PathPattern, TreeOptions,
[ChildName | ReversedPath],
[CurrentNode | ReversedParentTree],
Extra, Fun, FunAcc);
{false, Cond} ->
interrupted_walk_down(
mismatching_node,
#{node_name => ChildName,
node_path => lists:reverse(
[ChildName | ReversedPath]),
node_props => gather_node_props_for_error(
Child),
condition => Cond},
PathPattern, TreeOptions,
[ChildName | ReversedPath],
[CurrentNode | ReversedParentTree],
Extra, Fun, FunAcc)
end;
_ ->
interrupted_walk_down(
node_not_found,
#{node_name => ChildName,
node_path => lists:reverse([ChildName | ReversedPath]),
condition => Condition},
PathPattern, TreeOptions,
[ChildName | ReversedPath],
[CurrentNode | ReversedParentTree],
Extra, Fun, FunAcc)
end;
{true, ?PARENT_KHEPRI_NODE} ->
%% TODO: Support calling Fun() with parent node based on
%% conditions on child nodes.
BadPathPattern =
lists:reverse(ReversedPath, [Condition | PathPattern]),
Exception = ?khepri_exception(
condition_targets_parent_node,
#{path => BadPathPattern,
condition => Condition}),
{error, Exception};
false ->
%% The caller expects that the path matches a single specific node
%% (no matter if it exists or not), but the condition could match
%% several nodes.
BadPathPattern =
lists:reverse(ReversedPath, [Condition | PathPattern]),
Exception = ?khepri_exception(
possibly_matching_many_nodes_denied,
#{path => BadPathPattern}),
{error, Exception}
end;
walk_down_the_tree1(
#node{child_nodes = Children} = CurrentNode,
[Condition | PathPattern] = WholePathPattern,
#{expect_specific_node := false} = TreeOptions,
ReversedPath, ReversedParentTree, Extra, Fun, FunAcc)
when ?IS_KHEPRI_CONDITION(Condition) ->
%% Like with "expect_specific_node =:= true" function clause above, We
%% distinguish the case where the condition must be verified against the
%% current node (i.e. the node name is ?KHEPRI_ROOT_NODE or
%% ?THIS_KHEPRI_NODE in the condition) instead of its child nodes.
SpecificNode = khepri_path:component_targets_specific_node(Condition),
case SpecificNode of
{true, NodeName}
when NodeName =:= ?KHEPRI_ROOT_NODE orelse
NodeName =:= ?THIS_KHEPRI_NODE ->
CurrentName = special_component_to_node_name(
NodeName, ReversedPath),
CondMet = khepri_condition:is_met(
Condition, CurrentName, CurrentNode),
case CondMet of
true ->
walk_down_the_tree1(
CurrentNode, PathPattern, TreeOptions,
ReversedPath, ReversedParentTree,
Extra, Fun, FunAcc);
{false, _} ->
StartingNode = starting_node_in_rev_parent_tree(
ReversedParentTree, CurrentNode),
{ok, StartingNode, Extra, FunAcc}
end;
{true, ?PARENT_KHEPRI_NODE} ->
%% TODO: Support calling Fun() with parent node based on
%% conditions on child nodes.
BadPathPattern =
lists:reverse(ReversedPath, [Condition | PathPattern]),
Exception = ?khepri_exception(
condition_targets_parent_node,
#{path => BadPathPattern,
condition => Condition}),
{error, Exception};
_ ->
%% There is a special case if the current node is the root node.
%% The caller can request that the root node's properties are
%% included. This is true by default if the path is `[]'. This
%% allows to get its props and payload atomically in a single
%% query.
IsRoot = ReversedPath =:= [],
IncludeRootProps = maps:get(
include_root_props, TreeOptions, false),
Ret0 = case IsRoot andalso IncludeRootProps of
true ->
walk_down_the_tree1(
CurrentNode, [], TreeOptions,
[], [],
Extra, Fun, FunAcc);
_ ->
{ok, CurrentNode, Extra, FunAcc}
end,
%% The result of the first part (the special case for the root
%% node if relevant) is used as a starting point for handling all
%% child nodes.
Ret1 = maps:fold(
fun
(ChildName, Child, {ok, CurNode, Extra1, FunAcc1}) ->
handle_branch(
CurNode, ChildName, Child,
WholePathPattern, TreeOptions,
ReversedPath,
Extra1, Fun, FunAcc1);
(_, _, Error) ->
Error
end, Ret0, Children),
case Ret1 of
{ok, CurrentNode, Extra, FunAcc2} ->
%% The current node didn't change, no need to update the
%% tree and evaluate keep_while conditions.
StartingNode = starting_node_in_rev_parent_tree(
ReversedParentTree, CurrentNode),
{ok, StartingNode, Extra, FunAcc2};
{ok, CurrentNode1, Extra2, FunAcc2} ->
CurrentNode2 = case CurrentNode1 of
CurrentNode ->
CurrentNode;
delete ->
CurrentNode1;
_ ->
%% Because of the loop, payload &
%% child list versions may have
%% been increased multiple times.
%% We want them to increase once
%% for the whole (atomic)
%% operation.
squash_version_bumps(
CurrentNode, CurrentNode1)
end,
walk_back_up_the_tree(
CurrentNode2, ReversedPath, ReversedParentTree,
Extra2, FunAcc2);
Error ->
Error
end
end;
walk_down_the_tree1(
#node{} = CurrentNode, [], _,
ReversedPath, ReversedParentTree, Extra, Fun, FunAcc) ->
CurrentPath = lists:reverse(ReversedPath),
case Fun(CurrentPath, CurrentNode, FunAcc) of
{ok, keep, FunAcc1} ->
StartingNode = starting_node_in_rev_parent_tree(
ReversedParentTree, CurrentNode),
{ok, StartingNode, Extra, FunAcc1};
{ok, delete, FunAcc1} ->
walk_back_up_the_tree(
delete, ReversedPath, ReversedParentTree, Extra, FunAcc1);
{ok, #node{} = CurrentNode1, FunAcc1} ->
walk_back_up_the_tree(
CurrentNode1, ReversedPath, ReversedParentTree, Extra, FunAcc1);
Error ->
Error
end.
-spec special_component_to_node_name(SpecialComponent, ReversedPath) ->
NodeName when
SpecialComponent :: ?KHEPRI_ROOT_NODE | ?THIS_KHEPRI_NODE,
ReversedPath :: khepri_path:native_path(),
NodeName :: khepri_path:component().
special_component_to_node_name(?KHEPRI_ROOT_NODE = NodeName, []) ->
NodeName;
special_component_to_node_name(?THIS_KHEPRI_NODE, [NodeName | _]) ->
NodeName;
special_component_to_node_name(?THIS_KHEPRI_NODE, []) ->
?KHEPRI_ROOT_NODE.
-spec starting_node_in_rev_parent_tree(ReversedParentTree) -> Node when
Node :: tree_node(),
ReversedParentTree :: [Node].
%% @private
starting_node_in_rev_parent_tree(ReversedParentTree) ->
hd(lists:reverse(ReversedParentTree)).
-spec starting_node_in_rev_parent_tree(ReversedParentTree, Node) -> Node when
Node :: tree_node(),
ReversedParentTree :: [Node].
%% @private
starting_node_in_rev_parent_tree([], CurrentNode) ->
CurrentNode;
starting_node_in_rev_parent_tree(ReversedParentTree, _) ->
starting_node_in_rev_parent_tree(ReversedParentTree).
-spec handle_branch(
Node, ChildName, Child, WholePathPattern, TreeOptions,
ReversedPath, Extra, Fun, FunAcc) -> Ret when
Node :: tree_node(),
ChildName :: khepri_path:component(),
Child :: tree_node(),
WholePathPattern :: khepri_path:native_pattern(),
TreeOptions :: khepri:tree_options(),
ReversedPath :: [Node | {Node, child_created}],
Extra :: walk_down_the_tree_extra(),
Fun :: walk_down_the_tree_fun(),
FunAcc :: any(),
Ret :: ok(Node, Extra, FunAcc) | khepri:error().
%% @private
handle_branch(
CurrentNode, ChildName, Child,
[Condition | PathPattern] = WholePathPattern,
TreeOptions, ReversedPath, Extra, Fun, FunAcc) ->
CondMet = khepri_condition:is_met(Condition, ChildName, Child),
Ret = case CondMet of
true ->
walk_down_the_tree1(
Child, PathPattern, TreeOptions,
[ChildName | ReversedPath],
[CurrentNode],
Extra, Fun, FunAcc);
{false, _} ->
{ok, CurrentNode, Extra, FunAcc}
end,
case Ret of
{ok, #node{child_nodes = Children} = CurrentNode1, Extra1, FunAcc1}
when is_map_key(ChildName, Children) ->
case khepri_condition:applies_to_grandchildren(Condition) of
false ->
Ret;
true ->
walk_down_the_tree1(
Child, WholePathPattern, TreeOptions,
[ChildName | ReversedPath],
[CurrentNode1],
Extra1, Fun, FunAcc1)
end;
{ok, _CurrentNode1, _Extra1, _FunAcc1} ->
%% The child node is gone, no need to test if the condition
%% applies to it or recurse.
Ret;
Error ->
Error
end.
-spec interrupted_walk_down(
Reason, Info, PathPattern, TreeOptions,
ReversedPath, ReversedParentTree,
Extra, Fun, FunAcc) -> Ret when
Reason :: mismatching_node | node_not_found,
Info :: map(),
PathPattern :: khepri_path:native_pattern(),
TreeOptions :: khepri:tree_options(),
ReversedPath :: khepri_path:native_path(),
Node :: tree_node(),
ReversedParentTree :: [Node | {Node, child_created}],
Extra :: walk_down_the_tree_extra(),
Fun :: walk_down_the_tree_fun(),
FunAcc :: any(),
Ret :: ok(Node, Extra, FunAcc) | khepri:error().
%% @private
interrupted_walk_down(
Reason, Info, PathPattern, TreeOptions, ReversedPath, ReversedParentTree,
Extra, Fun, FunAcc) ->
NodePath = lists:reverse(ReversedPath),
IsTarget = khepri_path:realpath(PathPattern) =:= [],
Info1 = Info#{node_is_target => IsTarget},
ErrorTuple = {interrupted, Reason, Info1},
case Fun(NodePath, ErrorTuple, FunAcc) of
{ok, ToDo, FunAcc1}
when ToDo =:= keep orelse ToDo =:= delete ->
?assertNotEqual([], ReversedParentTree),
StartingNode = starting_node_in_rev_parent_tree(
ReversedParentTree),
{ok, StartingNode, Extra, FunAcc1};
{ok, #node{} = NewNode, FunAcc1} ->
ReversedParentTree1 =
case Reason of
node_not_found ->
%% We record the fact the child is a new node. This is used
%% to reset the child's stats if it got new payload or
%% child nodes at the same time.
[{hd(ReversedParentTree), child_created}
| tl(ReversedParentTree)];
_ ->
ReversedParentTree
end,
case PathPattern of
[] ->
%% We reached the target node. We could call
%% walk_down_the_tree1() again, but it would call Fun() a
%% second time.
walk_back_up_the_tree(
NewNode, ReversedPath, ReversedParentTree1,
Extra, FunAcc1);
_ ->
%% We created a tree node automatically on our way to the
%% target. We want to add a `keep_while' condition for it
%% so it is automatically reclaimed when it becomes
%% useless (i.e., no payload and no child nodes).
Cond = #if_any{conditions =
[#if_child_list_length{count = {gt, 0}},
#if_has_payload{has_payload = true}]},
KeepWhile = #{NodePath => Cond},
Extra1 = update_keep_while_conds_extra(
Extra, NodePath, KeepWhile),
walk_down_the_tree1(
NewNode, PathPattern, TreeOptions,
ReversedPath, ReversedParentTree1,
Extra1, Fun, FunAcc1)
end;
Error ->
Error
end.
-spec reset_versions(Node) -> Node when
Node :: tree_node().
%% @private
reset_versions(#node{props = Stat} = CurrentNode) ->
Stat1 = Stat#{payload_version => ?INIT_DATA_VERSION,
child_list_version => ?INIT_CHILD_LIST_VERSION},
CurrentNode#node{props = Stat1}.
-spec squash_version_bumps(OldNode, NewNode) -> Node when
OldNode :: tree_node(),
NewNode :: tree_node(),
Node :: tree_node().
%% @private
squash_version_bumps(
#node{props = #{payload_version := DVersion,
child_list_version := CVersion}},
#node{props = #{payload_version := DVersion,
child_list_version := CVersion}} = CurrentNode) ->
CurrentNode;
squash_version_bumps(
#node{props = #{payload_version := OldDVersion,
child_list_version := OldCVersion}},
#node{props = #{payload_version := NewDVersion,
child_list_version := NewCVersion} = Stat} = CurrentNode) ->
DVersion = case NewDVersion > OldDVersion of
true -> OldDVersion + 1;
false -> OldDVersion
end,
CVersion = case NewCVersion > OldCVersion of
true -> OldCVersion + 1;
false -> OldCVersion
end,
Stat1 = Stat#{payload_version => DVersion,
child_list_version => CVersion},
CurrentNode#node{props = Stat1}.
-spec walk_back_up_the_tree(
Child, ReversedPath, ReversedParentTree, Extra, FunAcc) -> Ret when
Node :: tree_node(),
Child :: Node | delete,
ReversedPath :: khepri_path:native_path(),
ReversedParentTree :: [Node | {Node, child_created}],
Extra :: walk_down_the_tree_extra(),
FunAcc :: any(),
Ret :: ok(Node, Extra, FunAcc).
%% @private
walk_back_up_the_tree(
Child, ReversedPath, ReversedParentTree, Extra, FunAcc) ->
walk_back_up_the_tree(
Child, ReversedPath, ReversedParentTree, Extra, #{}, FunAcc).
-spec walk_back_up_the_tree(
Child, ReversedPath, ReversedParentTree, Extra, AppliedChanges,
FunAcc) -> Ret when
Node :: tree_node(),
Child :: Node | delete,
ReversedPath :: khepri_path:native_path(),
ReversedParentTree :: [Node | {Node, child_created}],
Extra :: walk_down_the_tree_extra(),
AppliedChanges :: #{khepri_path:native_path() => Node | delete},
FunAcc :: any(),
Ret :: ok(Node, Extra, FunAcc).
%% @private
walk_back_up_the_tree(
delete,
[ChildName | ReversedPath] = WholeReversedPath,
[ParentNode | ReversedParentTree], Extra, AppliedChanges, FunAcc) ->
%% Evaluate keep_while of nodes which depended on ChildName (it is
%% removed) at the end of walk_back_up_the_tree().
Path = lists:reverse(WholeReversedPath),
AppliedChanges1 = AppliedChanges#{Path => delete},
%% Evaluate keep_while of parent node on itself right now (its child_count
%% has changed).
ParentNode1 = remove_node_child(ParentNode, ChildName),
handle_keep_while_for_parent_update(
ParentNode1, ReversedPath, ReversedParentTree,
Extra, AppliedChanges1, FunAcc);
walk_back_up_the_tree(
Child,
[ChildName | ReversedPath],
[{ParentNode, child_created} | ReversedParentTree],
Extra, AppliedChanges, FunAcc) ->
%% No keep_while to evaluate, the child is new and no nodes depend on it
%% at this stage.
%% FIXME: Perhaps there is a condition in a if_any{}?
Child1 = reset_versions(Child),
%% Evaluate keep_while of parent node on itself right now (its child_count
%% has changed).
ParentNode1 = add_node_child(ParentNode, ChildName, Child1),
handle_keep_while_for_parent_update(
ParentNode1, ReversedPath, ReversedParentTree,
Extra, AppliedChanges, FunAcc);
walk_back_up_the_tree(
Child,
[ChildName | ReversedPath] = WholeReversedPath,
[ParentNode | ReversedParentTree],
Extra, AppliedChanges, FunAcc) ->
%% Evaluate keep_while of nodes which depend on ChildName (it is
%% modified) at the end of walk_back_up_the_tree().
Path = lists:reverse(WholeReversedPath),
TreeOptions = #{props_to_return => [payload,
payload_version,
child_list_version,
child_list_length]},
NodeProps = gather_node_props(Child, TreeOptions),
AppliedChanges1 = AppliedChanges#{Path => NodeProps},
%% No need to evaluate keep_while of ParentNode, its child_count is
%% unchanged.
ParentNode1 = update_node_child(ParentNode, ChildName, Child),
walk_back_up_the_tree(
ParentNode1, ReversedPath, ReversedParentTree,
Extra, AppliedChanges1, FunAcc);
walk_back_up_the_tree(
StartingNode,
[], %% <-- We reached the root (i.e. not in a branch, see handle_branch())
[], Extra, AppliedChanges, FunAcc) ->
Extra1 = merge_applied_changes(Extra, AppliedChanges),
handle_applied_changes(StartingNode, Extra1, FunAcc);
walk_back_up_the_tree(
StartingNode,
_ReversedPath,
[], Extra, AppliedChanges, FunAcc) ->
Extra1 = merge_applied_changes(Extra, AppliedChanges),
{ok, StartingNode, Extra1, FunAcc}.
handle_keep_while_for_parent_update(
ParentNode,
ReversedPath,
[{_GrandParentNode, child_created} | _] = ReversedParentTree,
Extra, AppliedChanges, FunAcc) ->
%% This is a freshly created node, we don't want to get rid of it right
%% away.
walk_back_up_the_tree(
ParentNode, ReversedPath, ReversedParentTree,
Extra, AppliedChanges, FunAcc);
handle_keep_while_for_parent_update(
ParentNode,
ReversedPath,
ReversedParentTree,
Extra, AppliedChanges, FunAcc) ->
ParentPath = lists:reverse(ReversedPath),
IsMet = is_keep_while_condition_met_on_self(
ParentPath, ParentNode, Extra),
case IsMet of
true ->
%% We continue with the update.
walk_back_up_the_tree(
ParentNode, ReversedPath, ReversedParentTree,
Extra, AppliedChanges, FunAcc);
{false, _Reason} ->
%% This parent node must be removed because it doesn't meet its
%% own keep_while condition. keep_while conditions for nodes
%% depending on this one will be evaluated with the recursion.
walk_back_up_the_tree(
delete, ReversedPath, ReversedParentTree,
Extra, AppliedChanges, FunAcc)
end.
merge_applied_changes(Extra, AppliedChanges) ->
OldKWA = maps:get(applied_changes, Extra, #{}),
NewKWA = maps:fold(
fun
(Path, delete, KWA1) ->
KWA1#{Path => delete};
(Path, NodeProps, KWA1) ->
case KWA1 of
#{Path := delete} -> KWA1;
_ -> KWA1#{Path => NodeProps}
end
end, OldKWA, AppliedChanges),
Extra#{applied_changes => NewKWA}.
handle_applied_changes(
Root,
#{applied_changes := AppliedChanges} = Extra,
FunAcc)
when AppliedChanges =:= #{} ->
{ok, Root, Extra, FunAcc};
handle_applied_changes(
Root,
#{keep_while_conds := KeepWhileConds,
keep_while_conds_revidx := KeepWhileCondsRevIdx,
applied_changes := AppliedChanges} = Extra,
FunAcc) ->
ToDelete = eval_keep_while_conditions(
AppliedChanges, KeepWhileConds, KeepWhileCondsRevIdx,
Root),
{KeepWhileConds1,
KeepWhileCondsRevIdx1} = maps:fold(
fun
(RemovedPath, delete, {KW, KWRevIdx}) ->
KW1 = maps:remove(RemovedPath, KW),
KWRevIdx1 = update_keep_while_conds_revidx(
KW, KWRevIdx,
RemovedPath, #{}),
{KW1, KWRevIdx1};
(_, _, Acc) ->
Acc
end, {KeepWhileConds, KeepWhileCondsRevIdx},
AppliedChanges),
Extra1 = Extra#{keep_while_conds => KeepWhileConds1,
keep_while_conds_revidx => KeepWhileCondsRevIdx1},
ToDelete1 = filter_and_sort_paths_to_delete(ToDelete, AppliedChanges),
remove_expired_nodes(ToDelete1, Root, Extra1, FunAcc).
eval_keep_while_conditions(
AppliedChanges, KeepWhileConds, KeepWhileCondsRevIdx, Root) ->
%% AppliedChanges lists all nodes which were modified or removed. We
%% want to transform that into a list of nodes to remove.
%%
%% Those marked as `delete' in AppliedChanges are already gone. We
%% need to find the nodes which depended on them, i.e. their keep_while
%% condition is not met anymore. Note that removed nodes' child nodes are
%% gone as well and must be handled (they are not specified in
%% AppliedChanges).
%%
%% Those modified in AppliedChanges must be evaluated again to decide
%% if they should be removed.
maps:fold(
fun
(RemovedPath, delete, ToDelete) ->
maps:fold(
fun(Path, Watchers, ToDelete1) ->
case lists:prefix(RemovedPath, Path) of
true ->
eval_keep_while_conditions_after_removal(
Watchers, KeepWhileConds, Root, ToDelete1);
false ->
ToDelete1
end
end, ToDelete, KeepWhileCondsRevIdx);
(UpdatedPath, NodeProps, ToDelete) ->
case KeepWhileCondsRevIdx of
#{UpdatedPath := Watchers} ->
eval_keep_while_conditions_after_update(
UpdatedPath, NodeProps,
Watchers, KeepWhileConds, Root, ToDelete);
_ ->
ToDelete
end
end, #{}, AppliedChanges).
eval_keep_while_conditions_after_update(
UpdatedPath, NodeProps, Watchers, KeepWhileConds, Root, ToDelete) ->
maps:fold(
fun(Watcher, ok, ToDelete1) ->
KeepWhile = maps:get(Watcher, KeepWhileConds),
CondOnUpdated = maps:get(UpdatedPath, KeepWhile),
IsMet = khepri_condition:is_met(
CondOnUpdated, UpdatedPath, NodeProps),
case IsMet of
true ->
ToDelete1;
{false, _} ->
case are_keep_while_conditions_met(Root, KeepWhile) of
true -> ToDelete1;
{false, _} -> ToDelete1#{Watcher => delete}
end
end
end, ToDelete, Watchers).
eval_keep_while_conditions_after_removal(
Watchers, KeepWhileConds, Root, ToDelete) ->
maps:fold(
fun(Watcher, ok, ToDelete1) ->
KeepWhile = maps:get(Watcher, KeepWhileConds),
case are_keep_while_conditions_met(Root, KeepWhile) of
true -> ToDelete1;
{false, _} -> ToDelete1#{Watcher => delete}
end
end, ToDelete, Watchers).
filter_and_sort_paths_to_delete(ToDelete, AppliedChanges) ->
Paths1 = lists:sort(
fun
(A, B) when length(A) =:= length(B) ->
A < B;
(A, B) ->
length(A) < length(B)
end,
maps:keys(ToDelete)),
Paths2 = lists:foldl(
fun(Path, Map) ->
case AppliedChanges of
#{Path := delete} ->
Map;
_ ->
case is_parent_being_removed(Path, Map) of
false -> Map#{Path => delete};
true -> Map
end
end
end, #{}, Paths1),
maps:keys(Paths2).
is_parent_being_removed([], _) ->
false;
is_parent_being_removed(Path, Map) ->
is_parent_being_removed1(lists:reverse(Path), Map).
is_parent_being_removed1([_ | Parent], Map) ->
case maps:is_key(lists:reverse(Parent), Map) of
true -> true;
false -> is_parent_being_removed1(Parent, Map)
end;
is_parent_being_removed1([], _) ->
false.
remove_expired_nodes([], Root, Extra, FunAcc) ->
{ok, Root, Extra, FunAcc};
remove_expired_nodes([PathToDelete | Rest], Root, Extra, FunAcc) ->
case do_delete_matching_nodes(PathToDelete, Root, Extra, #{}) of
{ok, Root1, Extra1, _} ->
remove_expired_nodes(Rest, Root1, Extra1, FunAcc)
end.
-ifdef(TEST).
get_root(#?MODULE{root = Root}) ->
Root.
get_keep_while_conds(
#?MODULE{keep_while_conds = KeepWhileConds}) ->
KeepWhileConds.
get_keep_while_conds_revidx(
#?MODULE{keep_while_conds_revidx = KeepWhileCondsRevIdx}) ->
KeepWhileCondsRevIdx.
-endif.
