fix: make take snapshot sync call (#20)

Author: yordis

lib/commanded/aggregates/aggregate.ex

@@ -274,9 +274,9 @@ defmodule Commanded.Aggregates.Aggregate do
   end
 
   @doc false
-  def take_snapshot(application, aggregate_module, aggregate_uuid) do
+  def take_snapshot(application, aggregate_module, aggregate_uuid, timeout \\ 5_000) do
     name = via_name(application, aggregate_module, aggregate_uuid)
-    GenServer.cast(name, :take_snapshot)
+    GenServer.call(name, :take_snapshot, timeout)
   end
 
   @doc false
@@ -315,11 +315,14 @@ defmodule Commanded.Aggregates.Aggregate do
 
   @doc false
   @impl GenServer
-  def handle_cast(:take_snapshot, %Aggregate{} = state), do: do_take_snapshot(state)
+  def handle_call(:take_snapshot, _from, %Aggregate{} = state) do
+    case do_take_snapshot(state) do
+      {:ok, state} ->
+        {:reply, :ok, state}
 
-  @impl GenServer
-  def handle_cast({:take_snapshot, lifespan_timeout}, %Aggregate{} = state) do
-    do_take_snapshot(%Aggregate{state | lifespan_timeout: lifespan_timeout})
+      {:error, _reason} = error ->
+        {:reply, error, state}
+    end
   end
 
   @doc false
@@ -353,7 +356,7 @@ defmodule Commanded.Aggregates.Aggregate do
 
     response =
       if Snapshotting.snapshot_required?(snapshotting, aggregate_version) do
-        :ok = GenServer.cast(self(), {:take_snapshot, lifespan_timeout})
+        send(self(), {:take_snapshot, lifespan_timeout})
 
         # Don't reply with a lifetime because we just asked for a snapshot to
         # be taken. When it finishes, it will set the timeout.
@@ -410,6 +413,20 @@ defmodule Commanded.Aggregates.Aggregate do
     end
   end
 
+  @doc false
+  @impl GenServer
+  def handle_info({:take_snapshot, lifespan_timeout}, %Aggregate{} = state) do
+    state = %Aggregate{state | lifespan_timeout: lifespan_timeout}
+
+    case do_take_snapshot(state) do
+      {:ok, state} ->
+        noreply_with_lifespan(state)
+
+      {:error, _error} ->
+        noreply_with_lifespan(state)
+    end
+  end
+
   @doc false
   @impl GenServer
   def handle_info(:timeout, %Aggregate{} = state) do
@@ -622,18 +639,15 @@ defmodule Commanded.Aggregates.Aggregate do
 
     Logger.debug(describe(state) <> " recording snapshot")
 
-    state =
-      case Snapshotting.take_snapshot(snapshotting, aggregate_version, aggregate_state) do
-        {:ok, snapshotting} ->
-          %Aggregate{state | snapshotting: snapshotting}
+    case Snapshotting.take_snapshot(snapshotting, aggregate_version, aggregate_state) do
+      {:ok, snapshotting} ->
+        {:ok, %Aggregate{state | snapshotting: snapshotting}}
 
-        {:error, error} ->
-          Logger.warning(describe(state) <> " snapshot failed due to: " <> inspect(error))
+      {:error, reason} = error ->
+        Logger.warning(describe(state) <> " snapshot failed due to: " <> inspect(reason))
 
-          state
-      end
-
-    noreply_with_lifespan(state)
+        error
+    end
   end
 
   defp telemetry_wrong_expected_version(context, from, state) do

lib/commanded/commands/dispatcher.ex

@@ -67,6 +67,20 @@ defmodule Commanded.Commands.Dispatcher do
     struct(Pipeline, Map.from_struct(payload))
   end
 
+  # Ignoring this dialyzer warning until dialyxir catches up on Elixir 1.18.x-otp-28
+  # Please file a bug in https://github.com/jeremyjh/dialyxir/issues with this message.
+  # Unknown error occurred:
+  #   %FunctionClauseError{
+  #     module: Dialyxir.Warnings.CallWithoutOpaque,
+  #     function: :format_long,
+  #     arity: 1,
+  #     kind: nil,
+  #     args: nil,
+  #     clauses: nil
+  #   }
+  # Open issue in dialyxir:
+  # https://github.com/jeremyjh/dialyxir/issues/568
+  @dialyzer {:nowarn_function, execute: 3}
   defp execute(%Pipeline{} = pipeline, %Payload{} = payload, %ExecutionContext{} = context) do
     %Pipeline{application: application, assigns: %{aggregate_uuid: aggregate_uuid}} = pipeline
     %Payload{aggregate_module: aggregate_module, timeout: timeout} = payload

test/application/application_supervisor_test.exs

@@ -28,6 +28,9 @@ defmodule Commanded.Aggregates.ApplicationSupervisorTest do
   end
 
   defp assert_hibernated(pid) do
-    assert Process.info(pid, :current_function) == {:current_function, {:erlang, :hibernate, 3}}
+    pre_otp_28 = {:current_function, {:erlang, :hibernate, 3}}
+    post_otp_28 = {:current_function, {:gen_server, :loop_hibernate, 4}}
+
+    assert Process.info(pid, :current_function) in [pre_otp_28, post_otp_28]
   end
 end

test/event/event_handler_batch_telemetry_test.exs

@@ -85,8 +85,7 @@ defmodule Commanded.Event.EventHandlerBatchTelemetryTest do
       recorded_events = EventFactory.map_to_recorded_events(events, 1, metadata: metadata)
       state = setup_state(ErrorHandlingBatchHandler)
 
-      {:stop, _, _} =
-        Handler.handle_info({:events, recorded_events}, state)
+      {:stop, _, _} = Handler.handle_info({:events, recorded_events}, state)
 
       assert_receive {[:commanded, :event, :batch, :start], _measurements, _metadata}
       refute_received {[:commanded, :event, :batch, :stop], _measurements, _metadata}

test/event/handler_init_test.exs

@@ -116,7 +116,10 @@ defmodule Commanded.Event.HandlerInitTest do
   end
 
   defp assert_hibernated(pid) do
-    assert Process.info(pid, :current_function) == {:current_function, {:erlang, :hibernate, 3}}
+    pre_otp_28 = {:current_function, {:erlang, :hibernate, 3}}
+    post_otp_28 = {:current_function, {:gen_server, :loop_hibernate, 4}}
+
+    assert Process.info(pid, :current_function) in [pre_otp_28, post_otp_28]
   end
 
   defp send_subscribed(handler) do