Batching support #569

fmterrorf · 2024-09-20T21:39:24Z

This adds the option of having batched event handlers to speed things up. Please refer to the included documentation for details.

This PR is mostly taken from #509

Related PRs
commanded/commanded-ecto-projections#54

1. devbox init 2. devbox add elixir@1.16.3 3. use direnv for loading devbox env There is a .tool-versions but we got rid of asdf from our dev machines so setting this up instead in order to avoid each of us from having to setup asdf up on our machines again.

Co-authored-by: Cees de Groot <cg@cdegroot.com>

Skeleton code for batched handling, to be tested Basic skeleton test Ensure that no conflicting options/handlers are configured Move the module compilation test into describe block Add code to do error handling on batches (untested) Add tests for batching error handling Allow ack_events to take a list of events and acknowledge last event batch handler telemetry error handling fixes Update support batch handler logic Add batch handler telemetry tests Align naming of batch handler test module Update retry logic Update tests Add configuration test for ensuring batch_size and concurrency not test at same time Fix handle_batch comparison and add test case Add tests for handle_batch state and update last_event_seen logic Remove list implementation for InMemory#ack_event/3 Minor update for event_handler_batch_state_test Document handle_batch/2 for state update Add more event handler test coverage - upcast test for handle_batch/2 - batch_reset_event_handler_test Do not retry if :skip is received Commanded.options() -> Commanded.Application.options() Allow skipping events for batched handler Include commanded#493 and commanded#489 in CHANGELOG Include commanded#493 and commanded#489 in CHANGELOG Release v1.4.0 bugfix: retry command executing when the aggregator is down right before the execution chore: improve typespec for router dispatch resp Include commanded#494 in CHANGELOG Remove duplicate event apply when receiving missed events published to aggregate's event stream Update CHANGELOG Use Erlang v25.0.4 and Elixir v1.14.0-otp-25 Require at least Elixir v1.10 Fix typespec typo in Commanded.Application Use `:test` Mix env for GitHub workflow To catch dialyzer and credo errors in test files. Release v1.4.1 retry remaining batch when skipping event only retry for batch Reduce compile-time dependencies Reformat Cleanup some TODO comments that have been implemented Remove unused code in test Make Credo happy Update docs for :skip return in error callback Filter any already seen events from handle_batch Update docs on event given to batch error hander

Format Fix dialyzer error

@davydog187

* Pull in initial work by @davydog187. * Redefine how acknowledgement works Skeleton code for batched handling, to be tested Basic skeleton test Ensure that no conflicting options/handlers are configured Move the module compilation test into describe block Add code to do error handling on batches (untested) Add tests for batching error handling Allow ack_events to take a list of events and acknowledge last event batch handler telemetry error handling fixes Update support batch handler logic Add batch handler telemetry tests Align naming of batch handler test module Update retry logic Update tests Add configuration test for ensuring batch_size and concurrency not test at same time Fix handle_batch comparison and add test case Add tests for handle_batch state and update last_event_seen logic Remove list implementation for InMemory#ack_event/3 Minor update for event_handler_batch_state_test Document handle_batch/2 for state update Add more event handler test coverage - upcast test for handle_batch/2 - batch_reset_event_handler_test Do not retry if :skip is received Commanded.options() -> Commanded.Application.options() Allow skipping events for batched handler Include commanded#493 and commanded#489 in CHANGELOG Include commanded#493 and commanded#489 in CHANGELOG Release v1.4.0 bugfix: retry command executing when the aggregator is down right before the execution chore: improve typespec for router dispatch resp Include commanded#494 in CHANGELOG Remove duplicate event apply when receiving missed events published to aggregate's event stream Update CHANGELOG Use Erlang v25.0.4 and Elixir v1.14.0-otp-25 Require at least Elixir v1.10 Fix typespec typo in Commanded.Application Use `:test` Mix env for GitHub workflow To catch dialyzer and credo errors in test files. Release v1.4.1 retry remaining batch when skipping event only retry for batch Reduce compile-time dependencies Reformat Cleanup some TODO comments that have been implemented Remove unused code in test Make Credo happy Update docs for :skip return in error callback Filter any already seen events from handle_batch Update docs on event given to batch error hander * Drop support for {:error, reason, event} Format Fix dialyzer error * Update docs * Use delegate_event_to_handler & make confirm_receipt be more generic * Do not retry :skip events --------- Co-authored-by: Dave Lucia <davelucianyc@gmail.com> Co-authored-by: Cees de Groot <cg@evrl.com>

cdegroot · 2024-09-20T21:46:31Z

Note: review can wait until Calmwave has dogfooded this for a bit.

yordis · 2024-11-28T18:10:45Z

Any updates on this one?

drteeth

Really great work, I think it's very close. I think it needs 3 things:

The InMemory adapter needs to serve up batches of events
The branch needs a reqbase against master
Test suite to pass

drteeth · 2024-12-03T01:01:29Z

lib/commanded/event_store/adapter.ex

+  as well.
+
+  TODO Batching: this holds true for the PostgreSQL implementation, needs to be added to in_memory and
+       verified for EventStoreDB.


~~Looks like the InMemory implementation is done~~. I think we can go ahead without EventStoreDB as I don't think it's going to see much use going forward.

~~I'd like to see the InMemory adapter implement batching as well before we merge.~~ This is done.

Hi @drteeth,

From giving a second read at the code, it seems like the InMemory handler already implements the desired behavior of "acknowledging" the previous events in a batch here: https://github.com/calmwave-open-source/commanded/blob/batching-support/lib/commanded/event_store/adapters/in_memory/subscriber.ex#L17-L20

Given that, could you perhaps clarify what further work would be needed? The tests have been fixed and the branch updated with the latest upstream version.

Thanks!
Carlos

anderslemke · 2025-01-14T11:51:15Z

I'm rooting for this 🫶

Sync commanded to latest version

Update main to latest master

Update with latest master

Update with master

Update with latest master

drteeth

Looking good, just a little bit more to go.

drteeth · 2025-06-03T22:22:45Z

devbox.lock

@@ -0,0 +1,61 @@
+{


Can we leave the devbox stuff out?

I can remove it.

drteeth · 2025-06-03T22:26:31Z

lib/commanded/event/handler.ex

  @callback error(
              error :: term(),
-              failed_event :: domain_event,
+              failed_event :: domain_event | [domain_event] | nil,


While I get that you will only get the next behaviour when opting in to batching, it may break code that is depending on failed_event to definitely be there as per the old contract.

Is there a way to avoid changing this contract? Would having a batch_error callback for this new path make sense?

Talk to me about what we could do here.

You could eliminate the nil with List.wrap/1 so at least there is only that difference to deal with?

At the very least the custom error module case needs to be dealt with:

commanded/lib/commanded/event/handler.ex

Line 1346 in 2e11e27

module.error(error, data, failure_context)

Agree, we can do without the nil here.

drteeth · 2025-06-03T22:38:29Z

test/event/event_handler_state_test.exs

      assert match?(%{state: 0}, metadata)
+
+      # Make sure the event handler doesn't continue running logic after the test is done
+      Process.exit(handler, :kill)


Why are these needed? The handler is started with start_supervised, is there a reason that isn't enough to cleanup after the test?

It was causing sometimes error logs in the tests, since it would continue processing events in the background for a second or two, and the Mock was only expecting 1 event.

drteeth · 2025-06-03T22:59:56Z

lib/commanded/event/handler.ex

+    telemetry_metadata = batch_telemetry_metadata(events, context, state)
+    start_time = telemetry_start(telemetry_metadata, :batch)
+
+    case delegate_event_to_handler(events, state) do


Given that you are repeating the cases here from the single event path, should we consider trying to make this contract explicit somehow?

I'm thinking that these two branches will drift out of sync overtime, and be a source of bugs.

Is there a way to avoid this? My knee-jerk thoughts go to a protocol or behaviour that the handler depends on with implementations for single and batched paths?

Maybe that's over-thinking it, WDYT?

I think that the deeper issue is that we have too many return values to check for to begin with, and maybe this duplication just brings it out :-)

So my preference would be to let it slide, then think about better error handling, maybe collapse the cases down to a reasonable amount (2, ideally), then we can look at the code and see whether it's worth extracting.

badazz91 · 2025-08-26T14:05:43Z

hey @drteeth, @cdegroot - is there some documentation / guidelines on how to use this new feature? I would be really eager to try this out in our production app - if suitable :)

thomasdziedzic-calmwave and others added 9 commits August 25, 2024 10:00

aggregate improvements around retries and telemetry (#2)

3c084bd

Co-authored-by: Cees de Groot <cg@cdegroot.com>

Pull in initial work by @davydog187.

f9d2280

Drop support for {:error, reason, event}

450ee05

Format Fix dialyzer error

Update docs

fb46ba0

Use delegate_event_to_handler & make confirm_receipt be more generic

c77eb04

Do not retry :skip events

69ce480

fmterrorf mentioned this pull request Sep 20, 2024

Batching support commanded/commanded-ecto-projections#54

Open

fmterrorf marked this pull request as ready for review September 20, 2024 21:42

drteeth requested changes Dec 3, 2024

View reviewed changes

cblage and others added 14 commits January 15, 2025 16:27

Merge pull request #4 from commanded/master

b031bdd

Sync commanded to latest version

Ran devbox install

134606f

Merge branch 'master' into main-calmwave-1.4.7

fb7dc0f

Merge branch 'commanded:master' into main-calmwave-1.4.7

a7f486b

Merge pull request #6 from calmwave-open-source/main-calmwave-1.4.7

69b4a26

Update main to latest master

Merge branch 'main-calmwave' into batching-support-updated

7fd286d

Fix test errors, logging, 1 broken test

b2ee408

Tests running

9d199c8

Remove TODO

a206f79

Merge pull request #7 from commanded/master

144dd5c

Update with latest master

Merge pull request #8 from calmwave-open-source/update-with-master

d61b6cd

Update with master

Merge branch 'main-calmwave' into batching-support

2be16d5

Merge pull request #9 from commanded/master

60073ed

Update with latest master

Merge branch 'main-calmwave' into batching-support

2e11e27

drteeth requested changes Jun 3, 2025

View reviewed changes

drteeth approved these changes Jun 24, 2025

View reviewed changes

cdegroot merged commit e8de0e6 into commanded:master Jun 25, 2025

cdegroot deleted the batching-support branch June 25, 2025 13:58

Batching support #569

Batching support #569

Uh oh!

Conversation

fmterrorf commented Sep 20, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cdegroot commented Sep 20, 2024

Uh oh!

yordis commented Nov 28, 2024

Uh oh!

drteeth left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

drteeth Dec 3, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anderslemke commented Jan 14, 2025

Uh oh!

drteeth left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cblage Jun 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cdegroot Jun 6, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

badazz91 commented Aug 26, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

9 participants

fmterrorf commented Sep 20, 2024 •

edited

Loading

drteeth left a comment •

edited

Loading

drteeth Dec 3, 2024 •

edited

Loading

drteeth left a comment •

edited

Loading

cblage Jun 5, 2025 •

edited

Loading

cdegroot Jun 6, 2025 •

edited

Loading