Merge pull request #54 from ccamel/fix/improvements

Fix/improvements

Author: ccamel

apps/es_kernel/src/es_kernel_aggregate.erl

@@ -319,11 +319,10 @@ apply_events(PayloadEvents, {AggregateModule, State0, Sequence0}) ->
     }
 ) -> ok.
 persist_events(PayloadEvents, {AggregateModule, StoreContext, StreamId, Sequence0, NowFun}) ->
-    %% Extract aggregate_type from StreamId (which is {aggregate_type, aggregate_id})
     {AggregateType, _AggregateId} = StreamId,
     {Events, _} =
-        lists:foldl(
-            fun(PayloadEvent, {Events, SequenceN}) ->
+        lists:mapfoldl(
+            fun(PayloadEvent, SequenceN) ->
                 Now = NowFun(),
                 SequenceN1 = SequenceN + 1,
                 EventType = AggregateModule:event_type(PayloadEvent),
@@ -336,9 +335,9 @@ persist_events(PayloadEvents, {AggregateModule, StoreContext, StreamId, Sequence
                         Now,
                         PayloadEvent
                     ),
-                {[Event | Events], SequenceN1}
+                {Event, SequenceN1}
             end,
-            {[], Sequence0},
+            Sequence0,
             PayloadEvents
         ),
     lists:foreach(

apps/es_kernel/src/es_kernel_mgr_aggregate.erl

@@ -77,6 +77,8 @@ Dispatches a command to the appropriate aggregate instance.
 - Command is the command to dispatch.
 
 Returns `ok` on success, or `{error, Reason}` if routing or execution fails.
+If the target aggregate dies during the call, returns `{error, {aggregate_down, Reason}}`
+and evicts the stale PID so a subsequent dispatch will start a fresh aggregate.
 """.
 -spec dispatch(ServerRef, Command) -> ok | {error, Reason} when
     ServerRef :: gen_server:server_ref(),
@@ -198,21 +200,34 @@ ensure_and_dispatch(
             end;
         Pid ->
             Result = forward(Pid, Command),
-            {ok, Result, State}
+            case Result of
+                {error, {aggregate_down, _}} ->
+                    NewPids = maps:remove(Target, Pids),
+                    {ok, Result, State#state{pids = NewPids}};
+                _ ->
+                    {ok, Result, State}
+            end
     end.
 
 -doc """
 Forwards a command to an aggregate process.
 
 Returns the result of the aggregate's `execute/2` function:
 `ok` on success or `{error, Reason}` on failure.
+
+If the target process dies before replying, returns `{error, {aggregate_down, Reason}}`.
 """.
 -spec forward(Pid, Command) -> ok | {error, Reason} when
     Pid :: pid(),
     Command :: es_contract_command:t(),
     Reason :: term().
 forward(Pid, Command) ->
-    es_kernel_aggregate:execute(Pid, Command).
+    try es_kernel_aggregate:execute(Pid, Command) of
+        Result -> Result
+    catch
+        exit:Reason ->
+            {error, {aggregate_down, Reason}}
+    end.
 
 -doc """
 Starts an aggregate process for a given stream ID.

apps/es_kernel/src/es_kernel_store.erl

@@ -92,12 +92,14 @@ This is a convenience wrapper around fold/5 that collects all events into a list
     Range :: es_contract_range:range(),
     Result :: [es_contract_event:t()].
 retrieve_events(StoreContext, StreamId, Range) ->
-    fold(
-        StoreContext,
-        StreamId,
-        fun(Event, Acc) -> Acc ++ [Event] end,
-        [],
-        Range
+    lists:reverse(
+        fold(
+            StoreContext,
+            StreamId,
+            fun(Event, Acc) -> [Event | Acc] end,
+            [],
+            Range
+        )
     ).
 
 -doc """

apps/es_kernel/test/es_kernel_mgr_aggregate_tests.erl

@@ -7,7 +7,8 @@ suite_test_() ->
         [
             {"aggregate_behaviour", fun aggregate_behaviour/0},
             {"aggregate_passivation", fun aggregate_passivation/0},
-            {"aggregate_invalid_command", fun aggregate_invalid_command/0}
+            {"aggregate_invalid_command", fun aggregate_invalid_command/0},
+            {"aggregate_survives_dead_worker", fun aggregate_survives_dead_worker/0}
         ],
     {foreach, fun setup/0, fun teardown/1, TestCases}.
 
@@ -173,6 +174,32 @@ aggregate_invalid_command() ->
 
     ?assertEqual(ok, es_kernel_mgr_aggregate:stop(Pid)).
 
+aggregate_survives_dead_worker() ->
+    Pid = start_mgr(5000),
+    Id = <<"123">>,
+
+    ?assertEqual(
+        ok,
+        es_kernel_mgr_aggregate:dispatch(Pid, cmd(deposit, Id, #{amount => 100}))
+    ),
+    AggPid = agg_id(Pid, bank_account, Id),
+    exit(AggPid, kill),
+
+    %% Dispatch while the cached pid is dead: should not crash manager and should return an error
+    ?assertMatch(
+        {error, {aggregate_down, _}},
+        es_kernel_mgr_aggregate:dispatch(Pid, cmd(deposit, Id, #{amount => 10}))
+    ),
+
+    %% Next dispatch should start a fresh aggregate and succeed (rehydrated to previous balance)
+    ?assertEqual(
+        ok,
+        es_kernel_mgr_aggregate:dispatch(Pid, cmd(deposit, Id, #{amount => 20}))
+    ),
+    ?assertState(agg_id(Pid, bank_account, Id), Id, #{balance := 120}, 2),
+
+    ?assertEqual(ok, es_kernel_mgr_aggregate:stop(Pid)).
+
 start_mgr(Timeout) ->
     StoreContext = es_kernel_app:get_store_context(),
     {ok, Pid} =