khepri_machine: Support process-lifetime-based keep_while conditions

[Why]
This allows to link the lifetime of a tree node to that of a process.
When the process terminates, the tree node is deleted.

[How]
The `keep_while` put option now accepts a PID in addition to the
combination of a path and a condition.

Like the trigger equivalent, the process indicated in a `keep_while`
condition is monitored by the Ra leader. The handling of the DOWN
messages is exactly the same as for triggers, except that a tree node is
deleted (instead of executing a trigger).

Author: dumbbell

doc/overview.edoc

@@ -163,6 +163,8 @@ Relative paths are useful when putting conditions on
 
 === Tree node lifetime ===
 
+==== Lifetime based on conditions on tree nodes ====
+
 A tree node's lifetime starts when it is inserted the first time and ends when
 it is removed from the tree. However, intermediary tree nodes created on the
 way remain in the tree long after the leaf node was removed.
@@ -189,6 +191,47 @@ KeepWhileCondition = #{[stock, wood] => #if_child_list_length{count = {gt, 0}}}.
 `keep_while' conditions on self (like the example above) are not evaluated on
 the first insert though.
 
+==== Lifetime based on an Erlang process lifetime ====
+
+In addition to `keep_while' conditions based on other tree nodes, it is also
+possible to have a `keep_while' condition pointing to an Erlang process. The
+condition is the PID of that process. In this case, the lifetime of the tree
+node is linked to that of the process: when the process exits, the tree node is
+deleted.
+
+```
+khepri:put(
+  StoreId, [stock, wood, <<"walnut">>], 200,
+
+  %% The inserted tree node will be removed as soon as the processs inserting
+  %% it exits.
+  #{keep_while => self()}
+).
+'''
+
+This feature is based on Erlang monitors created by the Ra leader of the Khepri
+cluster.
+
+If the monitored process is on a different Erlang node than the Ra leader, the
+loss of connection between the two nodes (i.e. the `noconnection' exit reason)
+is handled in a specific way:
+<ul>
+<li>If the remote node is a member of the Khepri cluster, Khepri will start to
+monitor that node after receiving the `noconnection' exit reaison. Then two
+things can happen:
+<ul>
+<li>The node comes back online. In this case, processes that are supposed to
+run on it are monitored again. If the node was restarted and the processes were
+killed, this new monitoring will detect this situation.</li>
+<li>The node is removed from the cluster. In this case, it is unlikely to come
+back. Thus, the processes that were running on it are considered as dead.</li>
+</ul></li>
+<li>If the remote node is not a member of the Khepri cluster, Khepri will
+consider that the processes supposed to run on this node are dead, even if this
+is network partition and the node and its processes could be reachable again in
+the future.</li>
+</ul>
+
 == Stores ==
 
 A Khepri store corresponds to one Ra cluster. In fact, the name of the Ra

src/khepri_condition.erl

@@ -290,36 +290,64 @@
                                          if_child_list_version() |
                                          if_child_list_length().
 
--type keep_while() :: #{khepri_path:path() => condition()}.
-%% An association between a path and a condition. As long as the condition
-%% evaluates to true, the tree node is kept. Once the condition evaluates to
-%% false, the tree node is deleted.
+-type keep_while() :: #{khepri_path:path() => condition()} |
+                      pid().
+%% A condition that impacts the lifetime of a tree node.
 %%
-%% If the `keep_while' conditions are false at the time of the insert, the
-%% insert fails. The only exception to that is if the `keep_while' condition
-%% is on the inserted node itself.
+%% As long as the condition evaluates to true, the tree node is kept. Once the
+%% condition evaluates to false, the tree node is deleted.
+%%
+%% There are several types of conditions:
+%% <ul>
+%% <li>An association between a target path and a Khepri condition ({@link
+%% condition()}) that must be true on that target.
 %%
 %% Paths in the map can be native paths or Unix-like paths. However, having
 %% two entries that resolve to the same node (one native path entry and one
-%% Unix-like path entry for instance) is undefined behavior: one of them will
-%% overwrite the other.
+%% Unix-like path entry for instance) is undefined behaviour: one of them will
+%% overwrite the other.</li>
+%% <li>A process PID that must be alive.
+%%
+%% This can be a process running on a member of the cluster or outside of the
+%% cluster. If the node hosting the process is a member and the connection to
+%% this node is lost, Khepri will wait for the node to come back and monitor
+%% the process again, to ensure they are actually gone. If that node is
+%% removed from the cluster while it is still down, the process is considered
+%% gone. If the node is not a member of the cluster, the process is considered
+%% gone right away.</li>
+%% </ul>
 %%
-%% Example:
+%% If the `keep_while' conditions are false at the time of the insert, the
+%% insert fails. The only exception to that is if the `keep_while' condition
+%% is path-based and that path points to the inserted node itself.
+%%
+%% Example with a path/condition combination:
 %% ```
 %% khepri:put(
-%%   StoreId,
-%%   [foo],
-%%   Payload,
+%%   StoreId, [foo], Payload,
+%%
+%%   %% The node `[foo]' will be removed as soon as `[bar]' is removed
+%%   %% because the condition associated with `[bar]' will not be true
+%%   %% anymore.
 %%   #{keep_while => #{
-%%     %% The node `[foo]' will be removed as soon as `[bar]' is removed
-%%     %% because the condition associated with `[bar]' will not be true
-%%     %% anymore.
 %%     [bar] => #if_node_exists{exists = true}
 %%   }}
 %% ).
 %% '''
+%%
+%% Example with a PID:
+%% ```
+%% khepri:put(
+%%   StoreId, [foo], Payload,
+%%
+%%   %% The node `[foo]' will be removed as soon as the process inserting it
+%%   %% exits.
+%%   #{keep_while => self()}}
+%% ).
+%% '''
 
--type native_keep_while() :: #{khepri_path:native_path() => condition()}.
+-type native_keep_while() :: #{khepri_path:native_path() => condition()} |
+                             pid().
 %% An association between a native path and a condition.
 %%
 %% This is the same as {@link keep_while()} but the paths in the map keys were
@@ -354,7 +382,7 @@
       KeepWhile :: keep_while(),
       NativeKeepWhile :: native_keep_while().
 
-ensure_native_keep_while(KeepWhile) ->
+ensure_native_keep_while(KeepWhile) when is_map(KeepWhile) ->
     maps:fold(
       fun(Path, Condition, Acc) ->
               Path1 = khepri_path:from_string(Path),
@@ -364,7 +392,9 @@ ensure_native_keep_while(KeepWhile) ->
               %% Should we merge conditions in a `#if_all{}' condition? Return
               %% an error?
               Acc#{Path1 => Condition}
-      end, #{}, KeepWhile).
+      end, #{}, KeepWhile);
+ensure_native_keep_while(KeepWhile) when is_pid(KeepWhile) ->
+    KeepWhile.
 
 -spec compile(Condition) -> Condition when
       Condition :: khepri_path:pattern_component().

src/khepri_machine.erl

@@ -74,6 +74,10 @@
 %% khepri:trigger_options()}.</li>
 %% <li>Added support for process-lifetime-based triggers. The trigger can be
 %% executed when a process exits.</li>
+%% <li>Added support for process-lifetime-based `keep_while' condition. The
+%% `keep_while' put option can take a PID. The tree node will be deleted as
+%% soon as the associated process exits. See {@link
+%% khepri_condition:keep_while()}.</li>
 %% </ul>
 %% </td>
 %% </tr>
@@ -415,10 +419,27 @@ put(StoreId, PathPattern, Payload, Options)
     Payload1 = khepri_payload:prepare(Payload),
     {CommandOptions, NonCommandOptions} = split_command_options(
                                             StoreId, Options),
-    Command = #put{path = PathPattern1,
-                   payload = Payload1,
-                   options = NonCommandOptions},
-    process_command(StoreId, Command, CommandOptions);
+    ValidKeepWhile = case NonCommandOptions of
+                         #{keep_while := KW} when is_pid(KW) ->
+                             does_api_comply_with(
+                               process_based_keep_while, StoreId);
+                         _ ->
+                             true
+                     end,
+    case ValidKeepWhile of
+        true ->
+            Command = #put{path = PathPattern1,
+                           payload = Payload1,
+                           options = NonCommandOptions},
+            process_command(StoreId, Command, CommandOptions);
+        false ->
+            ?khepri_misuse(
+               unsupported_process_based_keep_while,
+               #{store_id => StoreId,
+                 node_path => PathPattern,
+                 payload => Payload,
+                 options => Options})
+    end;
 put(_StoreId, PathPattern, Payload, _Options) ->
     ?khepri_misuse(invalid_payload, #{path => PathPattern,
                                       payload => Payload}).
@@ -1556,7 +1577,8 @@ handle_aux(
 
     %% In `maybe_down_procs', we have processes that run on a member of the
     %% cluster and we are waiting for that node to come back up to monitor
-    %% these processes again (they are referenced by triggers).
+    %% these processes again (they are referenced by triggers and/or
+    %% `keep_while' conditions).
     %%
     %% However, the member might have been removed from the cluster after
     %% going down. Therefore, we need to verify on a regular basis if this
@@ -1774,9 +1796,27 @@ apply(
   #put{path = PathPattern, payload = Payload, options = NonCommandOptions},
   State) ->
     {TreeOptions, PutOptions} = split_put_options(NonCommandOptions),
-    Ret = insert_or_update_node(
-            State, PathPattern, Payload, PutOptions, TreeOptions, []),
-    post_apply(Ret, Meta);
+    Ret1 = insert_or_update_node(
+             State, PathPattern, Payload, PutOptions, TreeOptions, []),
+
+    %% If the operation was a success and the caller set a `keep_while'
+    %% condition on the inserted/updated tree node, we add the side effects
+    %% required by this condition, if any. For example, if the `keep_while'
+    %% condition is a PID, we need to monitor it.
+    Ret2 = case Ret1 of
+               {State1, {ok, _} = Result, SideEffects} ->
+                   case PutOptions of
+                       #{keep_while := KeepWhile} ->
+                           SideEffects1 = keep_while_to_side_effects(
+                                            KeepWhile, SideEffects),
+                           {State1, Result, SideEffects1};
+                       _ ->
+                           Ret1
+                   end;
+               _ ->
+                   Ret1
+           end,
+    post_apply(Ret2, Meta);
 apply(
   Meta,
   #delete{path = PathPattern, options = TreeOptions},
@@ -2003,12 +2043,35 @@ apply(Meta, {down,  _Pid, noconnection} = Command, State) ->
     post_apply(Ret, Meta);
 apply(Meta, {Down,  Pid, Reason}, State)
   when Down =:= down orelse Down =:= really_down ->
+    %% Handle keep_while conditions. We lookup tree nodes watching this exited
+    %% process and add delete them.
+    KeepWhileConds = get_keep_while_conds(State),
+    TreeOptions = #{},
+    {State1,
+     SideEffects1} = maps:fold(
+                       fun
+                           (Watcher, KeepWhile, {S, SE})
+                             when is_pid(KeepWhile) andalso
+                                  KeepWhile =:= Pid ->
+                               %% The process associated with this path
+                               %% exited. Therefore, we delete the
+                               %% corresponding tree node.
+                               %%
+                               %% We ignore the result here, because there is
+                               %% no explicit caller and no-one to return to.
+                               {S1, _R, SE1} = delete_matching_nodes(
+                                                 S, Watcher, TreeOptions, SE),
+                               {S1, SE1};
+                           (_Watcher, _KeepWhile, Acc) ->
+                               Acc
+                       end, {State, []}, KeepWhileConds),
+
     %% Handle triggers. We lookup triggers watching this exited process and add
     %% the side effects to execute the corresponding actions.
     Event = #ev_process{pid = Pid, change = {'DOWN', Reason}},
-    {State1, SideEffects} = add_trigger_side_effects(
-                              State, State, [Event], []),
-    Ret = {State1, Meta, SideEffects},
+    {State2, SideEffects2} = add_trigger_side_effects(
+                               State1, State1, [Event], SideEffects1),
+    Ret = {State2, Meta, SideEffects2},
     post_apply(Ret, Meta);
 apply(Meta, {nodeup, Node}, State) ->
     %% We can stop monitoring this node. If we get a `noconnection' from a
@@ -2018,8 +2081,19 @@ apply(Meta, {nodeup, Node}, State) ->
     %% We need to restore monitoring of processes that are hosted on that
     %% specific node. This will tell us if the process are still there are
     %% gone.
-    Triggers = get_triggers(State),
+    KeepWhileConds = get_keep_while_conds(State),
     SideEffects2 = maps:fold(
+                     fun
+                         (_Watcher, KeepWhile, SE)
+                           when is_pid(KeepWhile) andalso
+                                node(KeepWhile) =:= Node ->
+                             keep_while_to_side_effects(KeepWhile, SE);
+                         (_Watcher, _KeepWhile, SE) ->
+                             SE
+                     end, SideEffects1, KeepWhileConds),
+
+    Triggers = get_triggers(State),
+    SideEffects3 = maps:fold(
                      fun
                          (_TriggerId,
                           #{event_filter :=
@@ -2028,8 +2102,8 @@ apply(Meta, {nodeup, Node}, State) ->
                              event_filter_to_side_effect(EventFilter, SE);
                          (_TriggerId, _Trigger, SE) ->
                              SE
-                     end, SideEffects1, Triggers),
-    Ret = {State, Meta, SideEffects2},
+                     end, SideEffects2, Triggers),
+    Ret = {State, Meta, SideEffects3},
     post_apply(Ret, Meta);
 apply(Meta, {nodedown, _Node}, State) ->
     Ret = {State, Meta},
@@ -2177,7 +2251,8 @@ trigger_delayed_aux_queries_eval({State, Result, SideEffects}, _Meta) ->
 state_enter(leader, State) ->
     SideEffects1 = emitted_triggers_to_side_effects(State, leader),
     SideEffects2 = non_emitted_triggers_to_side_effects(State, SideEffects1),
-    SideEffects2;
+    SideEffects3 = keep_while_conds_to_side_effects(State, SideEffects2),
+    SideEffects3;
 state_enter(recovered, _State) ->
     SideEffect = {aux, restore_projections},
     [SideEffect];
@@ -2393,6 +2468,9 @@ does_api_comply_with(cached_members_list, MacVer)
 does_api_comply_with(extended_trigger, MacVer)
   when is_integer(MacVer) ->
     MacVer >= 3;
+does_api_comply_with(process_based_keep_while, MacVer)
+  when is_integer(MacVer) ->
+    MacVer >= 3;
 does_api_comply_with(_Behaviour, MacVer)
   when is_integer(MacVer) ->
     false;
@@ -2625,6 +2703,36 @@ evaluate_projection(
     Effect = {aux, Trigger},
     [Effect | Effects].
 
+-spec keep_while_conds_to_side_effects(State, SideEffects) ->
+    NewSideEffects when
+      State :: khepri_machine:state(),
+      SideEffects :: ra_machine:effects(),
+      NewSideEffects :: ra_machine:effects().
+%% @private
+
+keep_while_conds_to_side_effects(State, SideEffects) ->
+    KeepWhileConds = get_keep_while_conds(State),
+    SideEffects1 = maps:fold(
+                     fun(_Watcher, KeepWhile, SE) ->
+                             keep_while_to_side_effects(KeepWhile, SE)
+                     end, SideEffects, KeepWhileConds),
+    SideEffects1.
+
+-spec keep_while_to_side_effects(KeepWhile, SideEffects) ->
+    NewSideEffects when
+      KeepWhile :: khepri_condition:native_keep_while(),
+      SideEffects :: ra_machine:effects(),
+      NewSideEffects :: ra_machine:effects().
+%% @private
+
+keep_while_to_side_effects(KeepWhile, SideEffects) when is_map(KeepWhile) ->
+    SideEffects;
+keep_while_to_side_effects(KeepWhile, SideEffects) when is_pid(KeepWhile) ->
+    Pid = KeepWhile,
+    SideEffect = {monitor, process, Pid},
+    SideEffects1 = [SideEffect | SideEffects],
+    SideEffects1.
+
 -spec event_filter_to_side_effect(EventFilter, SideEffects) ->
     NewSideEffects when
       EventFilter :: khepri_evf:event_filter(),