Repository: commanded/commanded
Pull request #140 (2018-02-08 12:25:25)

Aggregate subscribes to its own events
Each aggregate instance will subscribe to its own events to catch any events appended to an aggregate's stream that were _not_ persisted via the aggregate instance process.

This might happen when events are directly appended to the stream using the configured event store, or when Commanded is run on multiple nodes but they're not connected together to form a cluster.

This pull request will require the two event store adapters be updated as the `Commanded.EventStore` behaviour has been extended to support single stream transient subscriptions. 


Patch:
diff --git a/lib/commanded/aggregates/aggregate.ex b/lib/commanded/aggregates/aggregate.ex
index 49928de..7f899be 100644
--- a/lib/commanded/aggregates/aggregate.ex
+++ b/lib/commanded/aggregates/aggregate.ex
@@ -45,7 +45,7 @@ defmodule Commanded.Aggregates.Aggregate do
   alias Commanded.Aggregates.{Aggregate, ExecutionContext}
   alias Commanded.Event.Mapper
   alias Commanded.EventStore
-  alias Commanded.EventStore.SnapshotData
+  alias Commanded.EventStore.{RecordedEvent, SnapshotData}
 
   @read_event_batch_size 100
 
@@ -53,6 +53,7 @@ defmodule Commanded.Aggregates.Aggregate do
             aggregate_uuid: nil,
             aggregate_state: nil,
             aggregate_version: 0,
+            lifespan_timeout: nil,
             snapshot_every: nil,
             snapshot_module_version: 1,
             snapshot_version: 0
@@ -78,6 +79,10 @@ defmodule Commanded.Aggregates.Aggregate do
     # events from the event store
     :ok = GenServer.cast(self(), :populate_aggregate_state)
 
+    # subscribe to aggregate's events to catch any events appended to its stream
+    # by another process, such as directly appended to the event store.
+    :ok = GenServer.cast(self(), :subscribe_to_events)
+
     {:ok, state}
   end
 
@@ -122,6 +127,11 @@ defmodule Commanded.Aggregates.Aggregate do
     GenServer.call(via_name(aggregate_module, aggregate_uuid), :aggregate_version)
   end
 
+  @doc false
+  def take_snapshot(aggregate_module, aggregate_uuid) do
+    GenServer.cast(via_name(aggregate_module, aggregate_uuid), :take_snapshot)
+  end
+
   @doc false
   def shutdown(aggregate_module, aggregate_uuid) do
     GenServer.stop(via_name(aggregate_module, aggregate_uuid))
@@ -133,7 +143,18 @@ defmodule Commanded.Aggregates.Aggregate do
   end
 
   @doc false
-  def handle_cast({:snapshot_state, lifespan_timeout}, %Aggregate{} = state) do
+  def handle_cast(:subscribe_to_events, %Aggregate{} = state) do
+    %Aggregate{aggregate_uuid: aggregate_uuid} = state
+
+    :ok = EventStore.subscribe(aggregate_uuid)
+
+    {:noreply, state}
+  end
+
+  @doc false
+  def handle_cast(:take_snapshot, %Aggregate{} = state) do
+    %Aggregate{lifespan_timeout: lifespan_timeout} = state
+
     {:noreply, do_snapshot(state), lifespan_timeout}
   end
 
@@ -155,8 +176,10 @@ defmodule Commanded.Aggregates.Aggregate do
         {:stop, :normal, reply, state}
 
       _ ->
+        state = %Aggregate{state | lifespan_timeout: lifespan_timeout}
+
         if snapshotting_enabled?(state) && snapshot_required?(state) do
-          :ok = GenServer.cast(self(), {:snapshot_state, lifespan_timeout})
+          :ok = GenServer.cast(self(), :take_snapshot)
 
           {:reply, reply, state}
         else
@@ -179,11 +202,70 @@ defmodule Commanded.Aggregates.Aggregate do
     {:reply, aggregate_version, state}
   end
 
+  @doc false
+  def handle_info({:events, events}, %Aggregate{} = state) do
+    %Aggregate{lifespan_timeout: lifespan_timeout} = state
+
+    Logger.debug(fn -> describe(state) <> " received events: #{inspect(events)}" end)
+
+    try do
+      state = Enum.reduce(events, state, &handle_event/2)
+
+      {:noreply, state, lifespan_timeout}
+    catch
+      {:error, reason} ->
+        Logger.debug(fn -> describe(state) <> " stopping due to: #{inspect(reason)}" end)
+
+        # stop after event handling returned an error
+        {:stop, reason, state}
+    end
+  end
+
   @doc false
   def handle_info(:timeout, %Aggregate{} = state) do
+    Logger.debug(fn -> describe(state) <> " stopping due to inactivity timeout" end)
+
     {:stop, :normal, state}
   end
 
+  # Handle events appended to the aggregate's stream, received by its
+  # event store subscription, by applying any missed events to its state.
+  defp handle_event(%RecordedEvent{} = event, %Aggregate{} = state) do
+    %RecordedEvent{data: data, stream_version: stream_version} = event
+
+    %Aggregate{
+      aggregate_module: aggregate_module,
+      aggregate_state: aggregate_state,
+      aggregate_version: aggregate_version
+    } = state
+
+    expected_version = aggregate_version + 1
+
+    case stream_version do
+      ^expected_version ->
+        # apply event to aggregate's state
+        %Aggregate{
+          state
+          | aggregate_version: stream_version,
+            aggregate_state: aggregate_module.apply(aggregate_state, data)
+        }
+
+      already_seen_version when already_seen_version <= aggregate_version ->
+        # ignore events already applied to aggregate state
+        state
+
+      _unexpected_version ->
+        Logger.debug(fn ->
+          describe(state) <> " received an unexpected event: #{inspect(event)}"
+        end)
+
+        # throw an error when an unexpected event is received
+        throw({:error, :unexpected_event_received})
+    end
+  end
+
+  # Populate the aggregate's state from a snapshot, if present, and it's events.
+  #
   # Attempt to fetch a snapshot for the aggregate to use as its initial state.
   # If the snapshot exists, fetch any subsequent events to rebuild its state.
   # Otherwise start with the aggregate struct and stream all existing events for
diff --git a/lib/commanded/event/handler.ex b/lib/commanded/event/handler.ex
index a49b354..190d77a 100644
--- a/lib/commanded/event/handler.ex
+++ b/lib/commanded/event/handler.ex
@@ -496,7 +496,13 @@ defmodule Commanded.Event.Handler do
     %Handler{state | last_seen_event: event_number}
   end
 
-  defp ack_event(event, %Handler{consistency: consistency, handler_name: handler_name, subscription: subscription}) do
+  defp ack_event(event, %Handler{} = state) do
+    %Handler{
+      consistency: consistency,
+      handler_name: handler_name,
+      subscription: subscription
+    } = state
+
     :ok = EventStore.ack_event(subscription, event)
     :ok = Subscriptions.ack_event(handler_name, consistency, event)
   end
diff --git a/lib/commanded/event_store/adapters/in_memory.ex b/lib/commanded/event_store/adapters/in_memory.ex
index 3fc1457..96b4812 100644
--- a/lib/commanded/event_store/adapters/in_memory.ex
+++ b/lib/commanded/event_store/adapters/in_memory.ex
@@ -14,7 +14,7 @@ defmodule Commanded.EventStore.Adapters.InMemory do
       serializer: nil,
       persisted_events: [],
       streams: %{},
-      transient_subscribers: [],
+      transient_subscribers: %{},
       persistent_subscriptions: %{},
       snapshots: %{},
       next_event_number: 1
@@ -52,49 +52,61 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     GenServer.start_link(__MODULE__, state, name: __MODULE__)
   end
 
+  @impl GenServer
   def init(%State{} = state) do
     {:ok, state}
   end
 
+  @impl Commanded.EventStore
   def append_to_stream(stream_uuid, expected_version, events) do
     GenServer.call(__MODULE__, {:append_to_stream, stream_uuid, expected_version, events})
   end
 
+  @impl Commanded.EventStore
   def stream_forward(stream_uuid, start_version \\ 0, read_batch_size \\ 1_000)
+
   def stream_forward(stream_uuid, start_version, _read_batch_size) do
     GenServer.call(__MODULE__, {:stream_forward, stream_uuid, start_version})
   end
 
-  def subscribe(subscriber) do
-    GenServer.call(__MODULE__, {:subscribe, subscriber})
+  @impl Commanded.EventStore
+  def subscribe(stream_uuid) do
+    GenServer.call(__MODULE__, {:subscribe, stream_uuid, self()})
   end
 
+  @impl Commanded.EventStore
   def subscribe_to_all_streams(subscription_name, subscriber, start_from) do
     subscription = %Subscription{name: subscription_name, subscriber: subscriber, start_from: start_from}
 
     GenServer.call(__MODULE__, {:subscribe_to_all_streams, subscription})
   end
 
+  @impl Commanded.EventStore
   def ack_event(pid, event) do
     GenServer.cast(__MODULE__, {:ack_event, event, pid})
   end
 
+  @impl Commanded.EventStore
   def unsubscribe_from_all_streams(subscription_name) do
     GenServer.call(__MODULE__, {:unsubscribe_from_all_streams, subscription_name})
   end
 
+  @impl Commanded.EventStore
   def read_snapshot(source_uuid) do
     GenServer.call(__MODULE__, {:read_snapshot, source_uuid})
   end
 
+  @impl Commanded.EventStore
   def record_snapshot(snapshot) do
     GenServer.call(__MODULE__, {:record_snapshot, snapshot})
   end
 
+  @impl Commanded.EventStore
   def delete_snapshot(source_uuid) do
     GenServer.call(__MODULE__, {:delete_snapshot, source_uuid})
   end
 
+  @impl GenServer
   def handle_call({:append_to_stream, stream_uuid, expected_version, events}, _from, %State{streams: streams} = state) do
     case Map.get(streams, stream_uuid) do
       nil ->
@@ -116,6 +128,7 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     end
   end
 
+  @impl GenServer
   def handle_call({:stream_forward, stream_uuid, start_version}, _from, %State{streams: streams} = state) do
     reply = case Map.get(streams, stream_uuid) do
       nil -> {:error, :stream_not_found}
@@ -128,25 +141,30 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     {:reply, reply, state}
   end
 
-  def handle_call({:subscribe, subscriber}, _from, %State{transient_subscribers: subscribers} = state) do
+  @impl GenServer
+  def handle_call({:subscribe, stream_uuid, subscriber}, _from, %State{} = state) do
+    %State{transient_subscribers: transient_subscribers} = state
+
     Process.monitor(subscriber)
 
-    state = %State{state |
-      transient_subscribers: [subscriber | subscribers],
-    }
+    transient_subscribers =
+      Map.update(transient_subscribers, stream_uuid, [subscriber], fn transient ->
+        [subscriber | transient]
+      end)
 
-    {:reply, :ok, state}
+    {:reply, :ok, %State{state | transient_subscribers: transient_subscribers}}
   end
 
+  @impl GenServer
   def handle_call({:subscribe_to_all_streams, %Subscription{name: subscription_name, subscriber: subscriber} = subscription}, _from, %State{persistent_subscriptions: subscriptions} = state) do
     {reply, state} = case Map.get(subscriptions, subscription_name) do
       nil ->
-        state = subscribe(subscription, state)
+        state = persistent_subscription(subscription, state)
 
         {{:ok, subscriber}, state}
 
       %Subscription{subscriber: nil} = subscription ->
-        state = subscribe(%Subscription{subscription | subscriber: subscriber}, state)
+        state = persistent_subscription(%Subscription{subscription | subscriber: subscriber}, state)
 
         {{:ok, subscriber}, state}
 
@@ -156,6 +174,7 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     {:reply, reply, state}
   end
 
+  @impl GenServer
   def handle_call({:unsubscribe_from_all_streams, subscription_name}, _from, %State{persistent_subscriptions: subscriptions} = state) do
     state = %State{state |
       persistent_subscriptions: Map.delete(subscriptions, subscription_name),
@@ -164,6 +183,7 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     {:reply, :ok, state}
   end
 
+  @impl GenServer
   def handle_call({:read_snapshot, source_uuid}, _from, %State{snapshots: snapshots} = state) do
     reply = case Map.get(snapshots, source_uuid, nil) do
       nil -> {:error, :snapshot_not_found}
@@ -173,6 +193,7 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     {:reply, reply, state}
   end
 
+  @impl GenServer
   def handle_call({:record_snapshot, %SnapshotData{source_uuid: source_uuid} = snapshot}, _from, %State{snapshots: snapshots} = state) do
     state = %State{state |
       snapshots: Map.put(snapshots, source_uuid, serialize(snapshot, state)),
@@ -181,6 +202,7 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     {:reply, :ok, state}
   end
 
+  @impl GenServer
   def handle_call({:delete_snapshot, source_uuid}, _from, %State{snapshots: snapshots} = state) do
     state = %State{state |
       snapshots: Map.delete(snapshots, source_uuid)
@@ -189,6 +211,7 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     {:reply, :ok, state}
   end
 
+  @impl GenServer
   def handle_cast({:ack_event, event, subscriber}, %State{persistent_subscriptions: subscriptions} = state) do
     state = %State{state |
       persistent_subscriptions: ack_subscription_by_pid(subscriptions, event, subscriber),
@@ -197,10 +220,13 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     {:noreply, state}
   end
 
-  def handle_info({:DOWN, _ref, :process, pid, _reason}, %State{persistent_subscriptions: subscriptions, transient_subscribers: subscribers} = state) do
+  @impl GenServer
+  def handle_info({:DOWN, _ref, :process, pid, _reason}, %State{} = state) do
+    %State{persistent_subscriptions: persistent, transient_subscribers: transient} = state
+
     state = %State{state |
-      persistent_subscriptions: remove_subscriber_by_pid(subscriptions, pid),
-      transient_subscribers: subscribers -- [pid],
+      persistent_subscriptions: remove_subscriber_by_pid(persistent, pid),
+      transient_subscribers: remove_transient_subscriber_by_pid(transient, pid)
     }
 
     {:noreply, state}
@@ -261,7 +287,10 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     }
   end
 
-  defp subscribe(%Subscription{name: subscription_name, subscriber: subscriber} = subscription, %State{persistent_subscriptions: subscriptions, persisted_events: persisted_events} = state) do
+  defp persistent_subscription(%Subscription{} = subscription, %State{} = state) do
+    %Subscription{name: subscription_name, subscriber: subscriber} = subscription
+    %State{persistent_subscriptions: subscriptions, persisted_events: persisted_events} = state
+
     Process.monitor(subscriber)
 
     send(subscriber, {:subscribed, subscriber})
@@ -287,6 +316,13 @@ defmodule Commanded.EventStore.Adapters.InMemory do
     end)
   end
 
+  defp remove_transient_subscriber_by_pid(transient_subscriptions, pid) do
+    Enum.reduce(transient_subscriptions, transient_subscriptions, fn
+      {stream_uuid, subscribers}, transient ->
+        Map.put(transient, stream_uuid, subscribers -- [pid])
+    end)
+  end
+
   defp catch_up(%Subscription{subscriber: nil}, _persisted_events, _state), do: :ok
   defp catch_up(%Subscription{start_from: :current}, _persisted_events, _state), do: :ok
   defp catch_up(%Subscription{subscriber: subscriber, start_from: :origin, last_seen_event_number: last_seen_event_number}, persisted_events, %State{} = state) do
@@ -299,9 +335,11 @@ defmodule Commanded.EventStore.Adapters.InMemory do
       do: send(subscriber, {:events, Enum.map(events, &deserialize(&1, state))})
   end
 
-  defp publish_to_transient_subscribers(stream_uuid, events, %State{transient_subscribers: subscribers}) do
+  defp publish_to_transient_subscribers(stream_uuid, events, %State{transient_subscribers: transient}) do
+    subscribers = Map.get(transient, stream_uuid, [])
+
     for subscriber <- subscribers |> Enum.filter(&is_pid/1) do
-      send(subscriber, {:events, stream_uuid, events})
+      send(subscriber, {:events, events})
     end
   end
 
diff --git a/lib/commanded/event_store/event_store.ex b/lib/commanded/event_store/event_store.ex
index 91a696d..b2a105f 100644
--- a/lib/commanded/event_store/event_store.ex
+++ b/lib/commanded/event_store/event_store.ex
@@ -32,6 +32,16 @@ defmodule Commanded.EventStore do
     | {:error, :stream_not_found}
     | {:error, reason}
 
+  @doc """
+  Create a transient subscription to a single event stream.
+
+  The event store will publish any events appended to the given stream to the
+  `subscriber` process as an `{:events, events}` message.
+
+  The subscriber does not need to acknowledge receipt of the events.
+  """
+  @callback subscribe(stream_uuid) :: :ok | {:error, reason}
+
   @doc """
   Create a persistent subscription to all event streams.
 
@@ -98,6 +108,14 @@ defmodule Commanded.EventStore do
     event_store_adapter().stream_forward(stream_uuid, start_version, read_batch_size)
   end
 
+  @doc """
+  Create a transient subscription to a single event stream.
+  """
+  @spec subscribe(stream_uuid) :: :ok | {:error, reason}
+  def subscribe(stream_uuid) do
+    event_store_adapter().subscribe(stream_uuid)
+  end
+
   @doc """
   Create a persistent subscription to all event streams.
   """


Test patch:
diff --git a/test/aggregates/aggregate_concurrency_test.exs b/test/aggregates/aggregate_concurrency_test.exs
new file mode 100644
index 0000000..fc3568a
--- /dev/null
+++ b/test/aggregates/aggregate_concurrency_test.exs
@@ -0,0 +1,139 @@
+defmodule Commanded.Aggregates.AggregateConcurrencyTest do
+  use Commanded.MockEventStoreCase
+
+  alias Commanded.Aggregates.{Aggregate, ExecutionContext}
+  alias Commanded.EventStore.RecordedEvent
+  alias Commanded.ExampleDomain.{BankAccount, OpenAccountHandler, DepositMoneyHandler}
+  alias Commanded.ExampleDomain.BankAccount.Commands.{OpenAccount, DepositMoney}
+  alias Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited
+
+  setup do
+    expect(MockEventStore, :subscribe_to_all_streams, fn _handler_name, handler, _subscribe_from ->
+      {:ok, handler}
+    end)
+
+    expect(MockEventStore, :subscribe, fn _aggregate_uuid -> :ok end)
+
+    :ok
+  end
+
+  describe "concurrency error" do
+    setup [
+      :open_account
+    ]
+
+    test "should retry command", context do
+      %{account_number: account_number} = context
+
+      command = %DepositMoney{
+        account_number: account_number,
+        transfer_uuid: UUID.uuid4(),
+        amount: 100
+      }
+
+      context = %ExecutionContext{
+        command: command,
+        handler: DepositMoneyHandler,
+        function: :handle,
+        retry_attempts: 5
+      }
+
+      # fail to append once
+      expect(MockEventStore, :append_to_stream, fn ^account_number, 1, _event_data ->
+        {:error, :wrong_expected_version}
+      end)
+
+      # return "missing" event
+      expect(MockEventStore, :stream_forward, fn ^account_number, 2, _batch_size ->
+        [
+          %RecordedEvent{
+            event_id: UUID.uuid4(),
+            event_number: 2,
+            stream_id: account_number,
+            stream_version: 2,
+            event_type: "Elixir.Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited",
+            data: %MoneyDeposited{
+              account_number: account_number,
+              transfer_uuid: UUID.uuid4(),
+              amount: 500,
+              balance: 1_500
+            },
+            metadata: %{}
+          }
+        ]
+      end)
+
+      # succeed on second attempt
+      expect(MockEventStore, :append_to_stream, fn ^account_number, 2, event_data ->
+        {:ok, 2 + length(event_data)}
+      end)
+
+      assert {:ok, 3, _events} = Aggregate.execute(BankAccount, account_number, context)
+
+      assert Aggregate.aggregate_version(BankAccount, account_number) == 3
+
+      assert Aggregate.aggregate_state(BankAccount, account_number) == %BankAccount{
+               account_number: account_number,
+               balance: 1_600,
+               state: :active
+             }
+    end
+
+    test "should error after too many attempts", context do
+      %{account_number: account_number} = context
+
+      # fail to append to stream
+      expect(MockEventStore, :append_to_stream, 6, fn ^account_number, 1, _event_data ->
+        {:error, :wrong_expected_version}
+      end)
+
+      expect(MockEventStore, :stream_forward, 6, fn ^account_number, 2, _batch_size -> [] end)
+
+      command = %DepositMoney{
+        account_number: account_number,
+        transfer_uuid: UUID.uuid4(),
+        amount: 100
+      }
+
+      context = %ExecutionContext{
+        command: command,
+        handler: DepositMoneyHandler,
+        function: :handle,
+        retry_attempts: 5
+      }
+
+      assert {:error, :too_many_attempts} =
+               Aggregate.execute(BankAccount, account_number, context)
+    end
+
+    defp open_account(_context) do
+      account_number = UUID.uuid4()
+
+      expect(MockEventStore, :stream_forward, fn ^account_number, 1, _batch_size ->
+        []
+      end)
+
+      expect(MockEventStore, :append_to_stream, fn ^account_number, 0, event_data ->
+        {:ok, length(event_data)}
+      end)
+
+      {:ok, ^account_number} =
+        Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
+
+      command = %OpenAccount{account_number: account_number, initial_balance: 1_000}
+
+      context = %ExecutionContext{
+        command: command,
+        handler: OpenAccountHandler,
+        function: :handle,
+        retry_attempts: 1
+      }
+
+      {:ok, 1, _events} = Aggregate.execute(BankAccount, account_number, context)
+
+      [
+        account_number: account_number
+      ]
+    end
+  end
+end
diff --git a/test/aggregates/aggregate_lifespan_test.exs b/test/aggregates/aggregate_lifespan_test.exs
index 3df324a..c755acf 100644
--- a/test/aggregates/aggregate_lifespan_test.exs
+++ b/test/aggregates/aggregate_lifespan_test.exs
@@ -1,7 +1,7 @@
 defmodule Commanded.Aggregates.AggregateLifespanTest do
   use Commanded.StorageCase
 
-  alias Commanded.Aggregates.BankRouter
+  alias Commanded.Aggregates.{Aggregate, BankRouter}
   alias Commanded.ExampleDomain.BankAccount
 
   alias Commanded.ExampleDomain.BankAccount.Commands.{
@@ -11,6 +11,8 @@ defmodule Commanded.Aggregates.AggregateLifespanTest do
     WithdrawMoney
   }
 
+  alias Commanded.EventStore
+
   alias Commanded.Registration
 
   describe "aggregate started" do
@@ -23,7 +25,7 @@ defmodule Commanded.Aggregates.AggregateLifespanTest do
       pid = Registration.whereis_name({BankAccount, aggregate_uuid})
       ref = Process.monitor(pid)
 
-      %{aggregate_uuid: aggregate_uuid, ref: ref}
+      %{aggregate_uuid: aggregate_uuid, pid: pid, ref: ref}
     end
 
     test "should shutdown after timeout", %{aggregate_uuid: aggregate_uuid, ref: ref} do
@@ -53,7 +55,7 @@ defmodule Commanded.Aggregates.AggregateLifespanTest do
       refute_receive {:DOWN, ^ref, :process, _, :normal}, 30
     end
 
-    test "should stop process when requested" , %{
+    test "should stop process when requested", %{
       aggregate_uuid: aggregate_uuid,
       ref: ref
     } do
@@ -62,6 +64,46 @@ defmodule Commanded.Aggregates.AggregateLifespanTest do
 
       assert_receive {:DOWN, ^ref, :process, _, :normal}
     end
+
+    test "should adhere to aggregate lifespan when taking snapshot after receiving published event", %{
+      aggregate_uuid: aggregate_uuid,
+      pid: pid,
+      ref: ref
+    } do
+      :ok = BankRouter.dispatch(%OpenAccount{account_number: aggregate_uuid, initial_balance: 10})
+
+      events = aggregate_uuid |> EventStore.stream_forward() |> Enum.to_list()
+
+      assert Process.alive?(pid)
+
+      # publish events to aggregate before taking snapshot
+      send(pid, {:events, events})
+
+      :ok = Aggregate.take_snapshot(BankAccount, aggregate_uuid)
+
+      assert_receive {:DOWN, ^ref, :process, _, :normal}
+      assert {:ok, _snapshot} = EventStore.read_snapshot(aggregate_uuid)
+    end
+
+    test "should adhere to aggregate lifespan when receiving published events after taking snapshot", %{
+      aggregate_uuid: aggregate_uuid,
+      pid: pid,
+      ref: ref
+    } do
+      :ok = BankRouter.dispatch(%OpenAccount{account_number: aggregate_uuid, initial_balance: 10})
+
+      events = aggregate_uuid |> EventStore.stream_forward() |> Enum.to_list()
+
+      :ok = Aggregate.take_snapshot(BankAccount, aggregate_uuid)
+
+      assert Process.alive?(pid)
+
+      # publish events to aggregate after taking snapshot
+      send(pid, {:events, events})
+
+      assert_receive {:DOWN, ^ref, :process, _, :normal}
+      assert {:ok, _snapshot} = EventStore.read_snapshot(aggregate_uuid)
+    end
   end
 
   describe "deprecated `after_command/1` callback" do
diff --git a/test/aggregates/aggregate_subscription_test.exs b/test/aggregates/aggregate_subscription_test.exs
new file mode 100644
index 0000000..e427f04
--- /dev/null
+++ b/test/aggregates/aggregate_subscription_test.exs
@@ -0,0 +1,110 @@
+defmodule Commanded.Aggregates.AggregateSubscriptionTest do
+  use Commanded.StorageCase
+
+  alias Commanded.Aggregates.{Aggregate, ExecutionContext}
+  alias Commanded.Aggregates.Supervisor, as: AggregateSupervisor
+  alias Commanded.{EventStore, Registration}
+  alias Commanded.ExampleDomain.{BankAccount, OpenAccountHandler}
+  alias Commanded.ExampleDomain.BankAccount.Commands.OpenAccount
+  alias Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited
+
+  describe "append event directly to aggregate stream" do
+    setup [
+      :open_account,
+      :append_event_to_stream
+    ]
+
+    test "should notify aggregate and mutate its state", context do
+      %{account_number: account_number} = context
+
+      assert Aggregate.aggregate_version(BankAccount, account_number) == 2
+
+      assert Aggregate.aggregate_state(BankAccount, account_number) == %BankAccount{
+               account_number: account_number,
+               balance: 1_500,
+               state: :active
+             }
+    end
+
+    test "should ignore already seen events", context do
+      %{account_number: account_number} = context
+
+      pid = Registration.whereis_name({BankAccount, account_number})
+      events = account_number |> EventStore.stream_forward() |> Enum.to_list()
+
+      # send already seen events multiple times, they should be ignored
+      send(pid, {:events, events})
+      send(pid, {:events, events})
+      send(pid, {:events, events})
+      send(pid, {:events, events})
+
+      assert Aggregate.aggregate_version(BankAccount, account_number) == 2
+
+      assert Aggregate.aggregate_state(BankAccount, account_number) == %BankAccount{
+               account_number: account_number,
+               balance: 1_500,
+               state: :active
+             }
+    end
+
+    test "should stop aggregate process when unexpected event received", context do
+      %{account_number: account_number} = context
+
+      pid = Registration.whereis_name({BankAccount, account_number})
+      ref = Process.monitor(pid)
+
+      events =
+        account_number
+        |> EventStore.stream_forward()
+        |> Enum.to_list()
+        |> Enum.map(fn recorded_event ->
+          # specify invalid stream version
+          %EventStore.RecordedEvent{
+            recorded_event
+            | stream_version: 999
+          }
+        end)
+
+      # send invalid events, should stop the aggregate process
+      send(pid, {:events, events})
+
+      assert_receive {:DOWN, ^ref, :process, _, :unexpected_event_received}
+    end
+
+    defp open_account(_context) do
+      account_number = UUID.uuid4()
+
+      {:ok, ^account_number} = AggregateSupervisor.open_aggregate(BankAccount, account_number)
+
+      context = %ExecutionContext{
+        command: %OpenAccount{account_number: account_number, initial_balance: 1_000},
+        handler: OpenAccountHandler,
+        function: :handle,
+        retry_attempts: 1
+      }
+
+      {:ok, 1, _events} = Aggregate.execute(BankAccount, account_number, context)
+
+      [
+        account_number: account_number
+      ]
+    end
+
+    # Write an event to the aggregate's stream, bypassing the aggregate process
+    defp append_event_to_stream(%{account_number: account_number}) do
+      event = %Commanded.EventStore.EventData{
+        event_type: "Elixir.Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited",
+        data: %MoneyDeposited{
+          account_number: account_number,
+          transfer_uuid: UUID.uuid4(),
+          amount: 500,
+          balance: 1_500
+        }
+      }
+
+      {:ok, _} = EventStore.append_to_stream(account_number, 1, [event])
+
+      :ok
+    end
+  end
+end
diff --git a/test/aggregates/execute_command_test.exs b/test/aggregates/execute_command_test.exs
index 9fd6850..99c31ed 100644
--- a/test/aggregates/execute_command_test.exs
+++ b/test/aggregates/execute_command_test.exs
@@ -2,10 +2,9 @@ defmodule Commanded.Aggregates.ExecuteCommandTest do
   use Commanded.StorageCase
 
   alias Commanded.Aggregates.{Aggregate, ExecutionContext}
-  alias Commanded.EventStore
-  alias Commanded.ExampleDomain.{BankAccount, OpenAccountHandler, DepositMoneyHandler}
-  alias Commanded.ExampleDomain.BankAccount.Commands.{OpenAccount, DepositMoney}
-  alias Commanded.ExampleDomain.BankAccount.Events.{BankAccountOpened, MoneyDeposited}
+  alias Commanded.ExampleDomain.{BankAccount, OpenAccountHandler}
+  alias Commanded.ExampleDomain.BankAccount.Commands.OpenAccount
+  alias Commanded.ExampleDomain.BankAccount.Events.BankAccountOpened
   alias Commanded.Helpers.{ProcessHelper,Wait}
   alias Commanded.Registration
 
@@ -74,71 +73,6 @@ defmodule Commanded.Aggregates.ExecuteCommandTest do
     assert state_before == Aggregate.aggregate_state(BankAccount, account_number)
   end
 
-  describe "concurrency error" do
-    setup [
-      :open_account,
-      :append_event_to_stream
-    ]
-
-    test "should retry command", context do
-      %{account_number: account_number} = context
-
-      command = %DepositMoney{account_number: account_number, transfer_uuid: UUID.uuid4(), amount: 100}
-      context = %ExecutionContext{command: command, handler: DepositMoneyHandler, function: :handle, retry_attempts: 1}
-
-      assert {:ok, 3, _events} = Aggregate.execute(BankAccount, account_number, context)
-
-      assert Aggregate.aggregate_version(BankAccount, account_number) == 3
-      assert Aggregate.aggregate_state(BankAccount, account_number) == %BankAccount{
-        account_number: account_number,
-        balance: 1_600,
-        state: :active,
-      }
-    end
-
-    test "should error after too many attempts", context do
-      %{account_number: account_number} = context
-
-      command = %DepositMoney{account_number: account_number, transfer_uuid: UUID.uuid4(), amount: 100}
-      context = %ExecutionContext{command: command, handler: DepositMoneyHandler, function: :handle, retry_attempts: 0}
-
-      assert {:error, :too_many_attempts} = Aggregate.execute(BankAccount, account_number, context)
-    end
-
-    defp open_account(_context) do
-      account_number = UUID.uuid4()
-
-      {:ok, ^account_number} = Commanded.Aggregates.Supervisor.open_aggregate(BankAccount, account_number)
-
-      command = %OpenAccount{account_number: account_number, initial_balance: 1_000}
-      context = %ExecutionContext{command: command, handler: OpenAccountHandler, function: :handle, retry_attempts: 1}
-
-      {:ok, 1, _events} = Aggregate.execute(BankAccount, account_number, context)
-
-      [
-        account_number: account_number
-      ]
-    end
-
-    # Write an event to the aggregate's stream, bypassing the aggregate process
-    # to simulate a concurrency error.
-    defp append_event_to_stream(%{account_number: account_number}) do
-      {:ok, _} = EventStore.append_to_stream(account_number, 1, [
-        %Commanded.EventStore.EventData{
-          event_type: "Elixir.Commanded.ExampleDomain.BankAccount.Events.MoneyDeposited",
-          data: %MoneyDeposited{
-            account_number: account_number,
-            transfer_uuid: UUID.uuid4(),
-            amount: 500,
-            balance: 1_500
-          }
-        }
-      ])
-
-      :ok
-    end
-  end
-
   def assert_aggregate_exit(aggregate_module, aggregate_uuid, fun) do
     pid = spawn(fun)
 
diff --git a/test/event_store_adapter/subscription_test.exs b/test/event_store_adapter/subscription_test.exs
index 390b891..ada111f 100644
--- a/test/event_store_adapter/subscription_test.exs
+++ b/test/event_store_adapter/subscription_test.exs
@@ -55,6 +55,46 @@ defmodule Commanded.EventStore.Adapter.SubscriptionTest do
     end
   end
 
+  describe "transient subscription to single stream" do
+    test "should receive events appended to the stream" do
+      stream_uuid = UUID.uuid4()
+
+      assert :ok = EventStore.subscribe(stream_uuid)
+
+      {:ok, 1} = EventStore.append_to_stream(stream_uuid, 0, build_events(1))
+
+      assert_receive {:events, received_events}
+      assert Enum.map(received_events, &(&1.stream_id)) == [stream_uuid]
+      assert Enum.map(received_events, &(&1.stream_version)) == [1]
+
+      {:ok, 3} = EventStore.append_to_stream(stream_uuid, 1, build_events(2))
+
+      assert_receive {:events, received_events}
+      assert Enum.map(received_events, &(&1.stream_id)) == [stream_uuid, stream_uuid]
+      assert Enum.map(received_events, &(&1.stream_version)) == [2, 3]
+
+      {:ok, 6} = EventStore.append_to_stream(stream_uuid, 3, build_events(3))
+
+      assert_receive {:events, received_events}
+      assert Enum.map(received_events, &(&1.stream_id)) == [stream_uuid, stream_uuid, stream_uuid]
+      assert Enum.map(received_events, &(&1.stream_version)) == [4, 5, 6]
+
+      refute_receive {:events, _received_events}
+    end
+
+    test "should not receive events appended to another stream" do
+      stream_uuid = UUID.uuid4()
+      another_stream_uuid = UUID.uuid4()
+
+      assert :ok = EventStore.subscribe(stream_uuid)
+
+      {:ok, 1} = EventStore.append_to_stream(another_stream_uuid, 0, build_events(1))
+      {:ok, 3} = EventStore.append_to_stream(another_stream_uuid, 1, build_events(2))
+
+      refute_receive {:events, _received_events}
+    end
+  end
+
   describe "subscribe to all streams" do
     test "should receive `:subscribed` message once subscribed" do
       {:ok, subscription} = EventStore.subscribe_to_all_streams("subscriber", self(), :origin)
diff --git a/test/helpers/storage_case.ex b/test/helpers/storage_case.ex
index 1f57426..063f331 100644
--- a/test/helpers/storage_case.ex
+++ b/test/helpers/storage_case.ex
@@ -1,5 +1,6 @@
 defmodule Commanded.StorageCase do
   @moduledoc false
+  
   use ExUnit.CaseTemplate
 
   require Logger
diff --git a/test/process_managers/support/multi/todo_process_manager.ex b/test/process_managers/support/multi/todo_process_manager.ex
index 2b109dc..6df9641 100644
--- a/test/process_managers/support/multi/todo_process_manager.ex
+++ b/test/process_managers/support/multi/todo_process_manager.ex
@@ -26,5 +26,5 @@ defmodule Commanded.ProcessManagers.TodoProcessManager do
     %TodoProcessManager{state | todo_uuid: todo_uuid}
   end
 
-  def apply(%TodoProcessManager{} = state, event), do: state
+  def apply(%TodoProcessManager{} = state, _event), do: state
 end
diff --git a/test/support/mock_event_store_case.ex b/test/support/mock_event_store_case.ex
new file mode 100644
index 0000000..d952bff
--- /dev/null
+++ b/test/support/mock_event_store_case.ex
@@ -0,0 +1,35 @@
+defmodule Commanded.MockEventStoreCase do
+  @moduledoc false
+
+  use ExUnit.CaseTemplate
+
+  import Mox
+
+  alias Commanded.EventStore.Adapters.Mock, as: MockEventStore
+  
+  using do
+    quote do
+      import Mox
+
+      alias Commanded.EventStore.Adapters.Mock, as: MockEventStore
+    end
+  end
+
+  setup do
+    set_mox_global()
+
+    default_event_store_adapter = Application.get_env(:commanded, :event_store_adapter)
+
+    # use mock event store adapter
+    :ok = Application.put_env(:commanded, :event_store_adapter, MockEventStore)
+
+    Application.ensure_all_started(:commanded)
+
+    on_exit(fn ->
+      Application.put_env(:commanded, :event_store_adapter, default_event_store_adapter)
+      Application.stop(:commanded)
+    end)
+
+    :ok
+  end
+end