Concurrent event handler acks a redelivered event as "already seen" without handling it

Start with test/event/event_handler_concurrency_test.exs and the Commanded.Event.Handler clause shown in the issue that skips events at or below last_seen_event. Reproduce the case by blocking event 1, delivering events 2 and 3 to the sibling, then killing the blocked instance. Done means the redelivered event reaches handle/2 and the regression test passes; the issue offers a patch approach and reports its suite checks.

Summary

An event handler started with concurrency: n (n > 1) can ack an event without calling handle/2. The event is lost. This happens when a handler instance exits while an event is in flight and the event is redelivered to a sibling instance.

Versions: commanded 1.4.11 (also current main) and eventstore 1.4.10, using commanded_eventstore_adapter 1.4.2. The in-memory adapter shows the same behaviour.

What happens

Commanded.Event.Handler skips any event whose number is at or below the instance's last_seen_event:

# Ignore already seen event.
defp handle_event(%RecordedEvent{event_number: event_number} = event, _context, %Handler{last_seen_event: last_seen_event} = state)
     when not is_nil(last_seen_event) and event_number <= last_seen_event do
  Logger.debug(describe(state) <> " has already seen event ##{inspect(event_number)}")

  confirm_receipt(event, state)
end

last_seen_event is per process. With one instance, that process sees every event in order, so the check is sound. With concurrency, each instance sees only some of the events. Its last_seen_event is the highest number among the events it was sent, not a position in the stream.

With the EventStore adapter:

1. Instance A is handling event N. Its sibling, instance B, handles later events in other partitions, or round-robin events when there is no partition_by/2. B's last_seen_event rises above N.
2. A exits mid-event: a crash, a kill, a linked exit, or an error/3 {:stop, reason}.
3. SubscriptionFsm.remove_subscriber/2 puts A's in-flight event N back at the head of its partition and sends it to the available subscriber with the lowest last_sent, which is B. A's restarted instance can't have subscribed yet: the subscription refuses it with :too_many_subscribers until A is removed, and the requeue happens during that removal.
4. B logs has already seen event #N, acks N and never calls handle/2. The subscription checkpoint moves past N.

This needs B to have capacity when A is removed. If B is still busy with its own event, N waits in the queue, and A's restarted instance may connect and take it first. In that case nothing is lost.

Only N is lost. That ack sets B's last_seen_event back down to N, so the later events in N's partition are handled normally.

A subscription reconnect can reach the same clause, because it sends events to the instances again from the last acked position. The in-memory adapter has its own redelivery path: on unsubscribe it rewinds the checkpoint to the in-flight event and republishes everything from there.

We hit this in production with a concurrency: 8 handler partitioned by key: an instance that exited mid-event lost that event. #556 reports the same symptom, the checkpoint moving past a failed event at concurrency: 5, and may have the same cause. I haven't run its reproduction.

Reproduction

This uses the in-memory adapter and the test setup in test/event/event_handler_concurrency_test.exs. The handler runs at concurrency: 2, and its handle/2 blocks on event 1 until the test sends :proceed:

1. Append event 1. One instance starts handling it and blocks.
2. Append events 2 and 3. Both go to the other instance, whose last_seen_event becomes 3.
3. Kill the blocked instance with Process.exit(pid, :kill).
4. Expected: another instance handles event 1. Actual: nothing handles it, and the log shows has already seen event #1.

On main this fails every time; I ran it with 25 different seeds.

It also reproduces with the EventStore adapter on Postgres: commanded 1.4.11, eventstore 1.4.10 and commanded_eventstore_adapter 1.4.2, with the same scenario as a test in the adapter's suite. Before killing the blocked instance, the test waits until the sibling has acked event 3. Both cases failed with all 25 seeds: a handler without partition_by/2, and one partitioned by key. The sibling logged has already seen event #1 and never handled it. I can share that test as well.

Possible fixes

1. Apply the already-seen skip only at concurrency: 1. At concurrency > 1, a redelivered event always goes to handle/2. The cost is that an instance may handle an event again that it already handled. That is within the at-least-once contract, and the skip only worked at concurrency > 1 when the event happened to come back to the same instance.
2. Track last_seen_event per partition. This is sound for partitioned handlers, because events in one partition are delivered in order. But without partition_by/2 every event has the same nil key, so the bug stays. The handler would also have to call partition_by/2 again for every event and keep an entry for every partition key it has ever seen.

Option 1 looks simpler and covers both partitioned and round-robin handlers. Neither option needs to touch handle_batch/3, since batching and concurrency can't be combined, or ProcessRouter, since process managers have no :concurrency option.

Offer

I have a patch for option 1: a guard on the clause head plus the reproduction above as a test. The full suite, credo --strict, Dialyzer and the format check pass with it. With the patch, the Postgres test passes with all 25 seeds, and the adapter's own suite still passes. I'm happy to open it as a PR if you'd like this fixed that way. If you'd rather fix it differently, or yourselves, that's fine too.