khepri_machine: Introduce {apply, MFA} trigger action

[Why]
It allows to execute an arbitrary function when a trigger is triggered.
This is like a stored procedure, except that the function must be
available on the cluster member that will execute it.

[How]
The `khepri:register_trigger/{3,4,5}` functions can take an MFA tuple in
addition to a stored procedure path or a PID. They can also accept the
`{sproc, StoreProcPath}`.

The function is executed with the given args list with the following
trigger descriptor appended:

    #khepri_trigger{type = EventType,
                    store_id = StoreId,
                    trigger_id = TriggerId,
                    event = EventProps,
		    action = ActionProps}

Author: dumbbell

src/khepri.erl

@@ -2843,9 +2843,11 @@ clear_many_payloads(StoreId, PathPattern, Options) ->
       TriggerId :: trigger_id(),
       EventFilter :: khepri_evf:event_filter_or_compat(),
       Action :: StoredProcPath |
+                MFA |
                 Pid |
                 khepri_event_handler:trigger_action(),
       StoredProcPath :: khepri_path:path(),
+      MFA :: {module(), atom(), list()},
       Pid :: pid(),
       Ret :: ok | error().
 %% @doc Registers a trigger.
@@ -2866,18 +2868,22 @@ register_trigger(TriggerId, EventFilter, Action) ->
       TriggerId :: trigger_id(),
       EventFilter :: khepri_evf:event_filter_or_compat(),
       Action :: StoredProcPath |
+                MFA |
                 Pid |
                 khepri_event_handler:trigger_action(),
       StoredProcPath :: khepri_path:path(),
+      MFA :: {module(), atom(), list()},
       Pid :: pid(),
       Ret :: ok | error();
 (TriggerId, EventFilter, Action, Options) -> Ret when
       TriggerId :: trigger_id(),
       EventFilter :: khepri_evf:event_filter_or_compat(),
       Action :: StoredProcPath |
+                MFA |
                 Pid |
                 khepri_event_handler:trigger_action(),
       StoredProcPath :: khepri_path:path(),
+      MFA :: {module(), atom(), list()},
       Pid :: pid(),
       Options :: command_options() | khepri:trigger_options(),
       Ret :: ok | error().
@@ -2910,9 +2916,11 @@ register_trigger(TriggerId, EventFilter, Action, Options)
       TriggerId :: trigger_id(),
       EventFilter :: khepri_evf:event_filter_or_compat(),
       Action :: StoredProcPath |
+                MFA |
                 Pid |
                 khepri_event_handler:trigger_action(),
       StoredProcPath :: khepri_path:path(),
+      MFA :: {module(), atom(), list()},
       Pid :: pid(),
       Options :: command_options() | khepri:trigger_options(),
       Ret :: ok | error().
@@ -2953,6 +2961,10 @@ register_trigger(TriggerId, EventFilter, Action, Options)
 %%
 %% The stored procedure is executed on the leader's Erlang node.
 %%
+%% When giving an MFA tuple (`Module, Function, ArgsList}'), the designated
+%% function is executed. The `#khepri_trigger{}' trigger descriptor is
+%% appended to the list of arguments.
+%%
 %% When giving a PID as the action, a `#khepri_trigger{}' message is sent to
 %% this process.
 %%

src/khepri_event_handler.erl

@@ -29,13 +29,16 @@
          code_change/3]).
 
 -type trigger_action() :: {sproc, khepri_path:native_path()} |
+                          {apply, {module(), atom(), list()}} |
                           {send, pid(), any()}.
 %% The type of action associated with a trigger.
 %%
 %% It can be the following actions:
 %% <ul>
 %% <li>`sproc': the stored procedure pointed to by the given path will be
 %% executed</li>
+%% <li>`apply', the function described by the given module/function/arguments
+%% tuple will be executed</li>
 %% <li>`send': a message will be sent to the specified PID with the event
 %% details and the given term</li>
 %% </ul>
@@ -44,13 +47,16 @@
 %% execution of that trigger. See {@link trigger_descriptor/0}.
 
 -type triggered_action() :: {sproc, horus:horus_fun()} |
+                            {apply, {module(), atom(), list()}} |
                             {send, pid(), any()}.
 %% The action associated with a trigger once it is triggered.
 %%
 %% It can be the following actions:
 %% <ul>
 %% <li>`sproc': the stored procedure (extracted from the path given to the
 %% trigger registration) is executed</li>
+%% <li>`apply': the function described by the given module/function/arguments
+%% tuple is executed</li>
 %% <li>`send': a message is sent to the specified PID with the event details
 %% and the given term</li>
 %% </ul>
@@ -174,6 +180,8 @@ event_to_props(#ev_process{pid = Pid, change = Change}) ->
 
 action_to_props({sproc, _StoredProc}) ->
     #{};
+action_to_props({apply, _MFA}) ->
+    #{};
 action_to_props({send, _Pid, undefined}) ->
     #{};
 action_to_props({send, _Pid, Priv}) ->
@@ -191,6 +199,11 @@ run_triggered_action(
   ActionArg, State) ->
     run_triggered_sproc(
       StoreId, TriggeredAction, StoredProc, ActionArg, State);
+run_triggered_action(
+  StoreId,
+  #triggered_v2{action = {apply, MFA}} = TriggeredAction,
+  ActionArg, State) ->
+    run_triggered_apply(StoreId, TriggeredAction, MFA, ActionArg, State);
 run_triggered_action(
   StoreId,
   #triggered_v2{action = {send, Pid, _Priv}} = TriggeredAction,
@@ -210,6 +223,21 @@ run_triggered_sproc(StoreId, TriggeredAction, StoredProc, ActionArg, State) ->
               Class, Reason, Stacktrace, State)
     end.
 
+run_triggered_apply(StoreId, TriggeredAction, MFA, ActionArg, State) ->
+    try
+        %% TODO: Be flexible and accept a function with an arity of that fits
+        %% `Args0' only.
+        {Mod, Func, Args0} = MFA,
+        Args1 = Args0 ++ [ActionArg],
+        _ = erlang:apply(Mod, Func, Args1),
+        State
+    catch
+        Class:Reason:Stacktrace ->
+            handle_action_crash(
+              StoreId, TriggeredAction, ActionArg,
+              Class, Reason, Stacktrace, State)
+    end.
+
 handle_action_crash(
   StoreId, TriggeredAction, ActionArg, Class, Reason, Stacktrace,
   #?MODULE{trigger_crashes = Crashes} = State) ->
@@ -228,7 +256,7 @@ handle_action_crash(
                               #triggered_v2{event_filter = EF} -> EF
                           end,
             Msg = io_lib:format(
-                    "Triggered stored procedure crash~n"
+                    "Triggered action crash~n"
                     "  Store ID: ~s~n"
                     "  Trigger ID: ~s~n"
                     "  Event filter:~n"

src/khepri_machine.erl

@@ -678,9 +678,11 @@ handle_tx_exception(
       TriggerId :: khepri:trigger_id(),
       EventFilter :: khepri_evf:event_filter_or_compat(),
       Action :: StoredProcPath |
+                MFA |
                 Pid |
                 khepri_event_handler:trigger_action(),
       StoredProcPath :: khepri_path:path(),
+      MFA :: {module(), atom(), list()},
       Pid :: pid(),
       Options :: khepri:command_options() | khepri:trigger_options(),
       Ret :: ok | khepri:error().
@@ -739,6 +741,19 @@ maybe_convert_to_action(StoreId, _TriggerId, _EventFilter, StoredProcPath)
         false ->
             StoredProcPath1
     end;
+maybe_convert_to_action(
+  StoreId, TriggerId, EventFilter, {Mod, Func, Args} = MFA)
+  when is_atom(Mod) andalso is_atom(Func) andalso is_list(Args) ->
+    case does_api_comply_with(extended_trigger, StoreId) of
+        true ->
+            {apply, MFA};
+        false ->
+            ?khepri_misuse(
+               unsupported_trigger_action,
+               #{trigger_id => TriggerId,
+                 event_filter => EventFilter,
+                 action => MFA})
+    end;
 maybe_convert_to_action(StoreId, TriggerId, EventFilter, Pid)
   when is_pid(Pid) ->
     case does_api_comply_with(extended_trigger, StoreId) of

test/triggers.erl

@@ -491,7 +491,7 @@ a_buggy_sproc_does_not_crash_state_machine_test_() ->
                                   ok, khepri:put(?FUNCTION_NAME, [foo], 1))
                         end),
                 ?assertMatch(
-                   <<"Triggered stored procedure crash", _/binary>>, Log)
+                   <<"Triggered action crash", _/binary>>, Log)
             end)},
 
         {"Checking the procedure was executed",
@@ -551,7 +551,7 @@ a_buggy_sproc_does_not_crash_state_machine_test_() ->
                                           khepri:put(?FUNCTION_NAME, [foo], 6)
                                   end),
                 ?assertSubString(
-                  <<"Triggered stored procedure crash">>, Log),
+                  <<"Triggered action crash">>, Log),
                 ?assertSubString(
                   <<"(this crash occurred 6 times in the last 10 seconds)">>,
                   Log),
@@ -632,6 +632,33 @@ receive_sproc_msg_with_props(Key) ->
     after 1000                  -> timeout
     end.
 
+event_triggers_mfa_apply_test_() ->
+    EventFilter = khepri_evf:tree([foo]),
+    {setup,
+     fun() -> test_ra_server_helpers:setup(?FUNCTION_NAME) end,
+     fun(Priv) -> test_ra_server_helpers:cleanup(Priv) end,
+     [{inorder,
+       [{"Registering a trigger",
+         ?_assertEqual(
+            ok,
+            khepri:register_trigger(
+              ?FUNCTION_NAME,
+              ?FUNCTION_NAME,
+              EventFilter,
+              {erlang, send, [self()]}))},
+
+        {"Updating a node; should trigger the execution of the MFA",
+         ?_assertMatch(
+            ok,
+            khepri:put(?FUNCTION_NAME, [foo], value))},
+
+        {"Checking the procedure was executed",
+         ?_assert(receive
+                      #khepri_trigger{} ->
+                          true
+                  end)}]
+      }]}.
+
 event_triggers_message_send_test_() ->
     EventFilter = khepri_evf:tree([foo]),
     {setup,