commanded/commanded-scheduler

Schedule one-off and recurring commands for Commanded CQRS/ES applications

Commanded

Use Commanded to build your own Elixir applications following the CQRS/ES pattern.

Provides support for:

• Command registration and dispatch.
• Hosting and delegation to aggregates.
• Event handling.
• Long running process managers.

Commanded provides a solid technical foundation for you to build on. It allows you to focus on modelling your domain, the most important part of your app, creating a better application at a faster pace.

You can use Commanded with one of the following event stores for persistence:

• EventStore - Elixir library using Postgres for persistence.
• EventStoreDB - a stream database built for Event Sourcing.
• In-memory event store - included for test use only.

Please refer to the CHANGELOG for features, bug fixes, and any upgrade advice included for each release.

Requires Erlang/OTP v21.0 and Elixir v1.11 or later.

---

Sponsors

• View sponsors & backers

---

• Changelog
• Wiki
• What is CQRS/ES?
• Frequently asked questions
• Getting help
• Latest published Hex package & documentation

MIT License

---

This README and the following guides follow the master branch which may not be the currently published version.

Read the documentation for the latest published version of Commanded on Hex.

Overview

• Getting started
• Choosing an event store
  • PostgreSQL-based EventStore
  • Greg Young's Event Store
• Using Commanded
  • Aggregates
    • Example aggregate
    • Commanded.Aggregate.Multi
    • Aggregate state snapshots
  • Commands
    • Command handlers
    • Command dispatch and routing
      • Define aggregate identity
      • Multi-command registration
      • Dispatch timeouts
      • Dispatch consistency guarantee
      • Dispatch returning execution result
      • Aggregate lifespan
      • Composite command routers
    • Middleware
      • Example middleware
    • Composite command routers
  • Events
    • Domain events
    • Event handlers
      • Consistency guarantee
    • Upcasting events
  • Process managers
    • Example process manager
  • Supervision
  • Serialization
    • Default JSON serializer
    • Configuring JSON serialization
    • Decoding event structs
    • Using an alternative serialization format
    • Customising serialization
  • Read model projections
• Deployment
  • Single node deployment
  • Multi node cluster deployment
  • Multi node, but not clustered deployment
• Testing with Commanded
• Used in production?
• Example application
• Learn Commanded in 20 minutes
• Choosing an event store provider
• Tooling
• Contributing
• Need help?

---

Used in production?

Yes, see the companies using Commanded.

Example application

Conduit is an open source, example Phoenix 1.3 web application implementing the CQRS/ES pattern in Elixir. It was built to demonstrate the implementation of Commanded in an Elixir application for the Building Conduit book.

Learn Commanded in 20 minutes

Watch Bernardo Amorim introduce CQRS and event sourcing at Code Beam SF 2018. Including a tutorial on how to implement an Elixir application using these concepts with Commanded.

Contributing

Pull requests to contribute new or improved features, and extend documentation are most welcome.

Please follow the existing coding conventions, or refer to the Elixir style guide.

You should include unit tests to cover any changes. Run mix test to execute the test suite.

Contributors

Commanded exists thanks to the following people who have contributed.

• Adil Yarulin
• Alexandre de Souza
• Andrey Akulov
• Andrzej Sliwa
• Ben Smith
• Benjamin Moss
• Bernardo Amorim
• Brenton Annan
• Chris Brodt
• Chris Martin
• Christophe Juniet
• Danilo Silva
• Dave Lucia
• David Carlin
• Damir Vandic
• Danni Friedland
• Dilaksun Bavarajan
• Ernesto
• Fernando Mendes
• Florian Ebeling
• Henry Hazan
• JC
• Joan Zapata
• Joao Gilberto Moura
• João Thallis
• John Wilger
• Joseph Lozano
• Kian-Meng Ang
• Kok J Sam
• Leif Gensert
• Luís Ferreira
• Marcelo Dominguez
• Matt Doughty
• Matthew Boehlig
• Michael Herold
• Miguel Palhas
• Nigel Thorne
• Olafur Arason
• Paolo Laurenti
• Patrick Detlefsen
• Phil Chen
• Raphaël Lustin
• Štefan Ľupták
• Tobiasz Małecki
• Vladimir Drobyshevskiy
• Willy Wombat
• Yordis Prieto
• Yuri de Figueiredo
• Zven

Need help?

Please open an issue if you encounter a problem, or need assistance. You can also seek help in the #commanded channel in the official Elixir Slack.

commanded-scheduler's Issues

Only a single instance of the jobs scheduler may run

Commanded scheduler should support running on a cluster of nodes. It must ensure that only a single jobs scheduler process may be running.

It's possible to use Erlang's :global module to enforce a global process name. Example given in this DockYard post on recurring jobs.

Recurring commands

Schedule a command to be executed on a recurring interval, using a cron-like format.

The API to schedule the recurring command is done, but execution is still pending.

Consider using SchedEx for job scheduling

SchedEx provides an API for scheduling recurring and one-off jobs where each job is its own GenServer process and execution is implemented using Process.send_after/4.

This would allow implementation of Recurring commands (#4).

Support for Commanded 1.0

Upgrading to v1.0 breaks commanded_scheduler with the following error:

== Compilation error in file lib/commanded/scheduler/scheduling/scheduling.ex ==
** (ArgumentError) Commanded.Scheduler.Scheduling expects :application option
    lib/commanded/event/handler.ex:332: Commanded.Event.Handler.compile_config/2
    lib/commanded/scheduler/scheduling/scheduling.ex:4: (module)
    (stdlib) erl_eval.erl:680: :erl_eval.do_apply/6

PR incoming with a naive patch, I'm not sure what the right pattern is for supporting multiple applications but it does work for a single application. I'd love feedback on how to do this right.

Migrations are incompatible

I setup an app to use Postgres EventStore. It's working well and tests are passing.

I added Commanded Scheduler and the migration can't run.

Case 1

$ MIX_ENV=test mix ecto.create # create the empty database
$ MIX_ENV=test mix event_store.init # create the tables events schema_migrations snapshots stream_events streams subscriptions
$ MIX_ENV=test mix ecto.migrate    
** (Postgrex.Error) ERROR 42703 (undefined_column) column s0.version does not exist
    query: SELECT s0."version"::bigint FROM "schema_migrations" AS s0 FOR UPDATE

Case 2

$ MIX_ENV=test mix ecto.create # create the empty database
$ MIX_ENV=test mix ecto.migrate # create the tables projection_versions schedules schema_migrations
$ MIX_ENV=test mix event_store.init 
** (Postgrex.Error) ERROR 42P07 (duplicate_table) relation "schema_migrations" already exists

case 3

$ MIX_ENV=test mix ecto.create # create the empty database
$ MIX_ENV=test mix ecto.migrate # create the tables projection_versions schedules schema_migrations
$ MIX_ENV=test mix event_store.migrate    
** (Postgrex.Error) ERROR 42703 (undefined_column) column "major_version" does not exist

I never used Ecto migrations before so maybe I'm missing something. For now I can't get all the tables I need for the EventStore and for Scheduler. :(

What should I do in order to share the same database?

Schedule batch commands API

Batch commands can be scheduled using the following Commanded.Scheduler.batch/2 function:

alias Commanded.Scheduler

Scheduler.batch(reservation_id, fn batch ->
  batch
  |> Scheduler.schedule_once(%TimeoutReservation{..}, timeout_due_at, name: "timeout")
  |> Scheduler.schedule_once(%ReleaseSeat{..}, release_due_at, name: "release")
end)

Rather than require an anonymous function be used to configure the batch, it might be simpler to add a new Scheduler.Batch module to construct the batch of scheduled commands:

alias Commanded.Scheduler
alias Commanded.Scheduler.Batch

batch = 
  reservation_id
  |> Batch.new()
  |> Batch.schedule_once(%TimeoutReservation{..}, timeout_due_at, name: "timeout")
  |> Batch.schedule_once(%ReleaseSeat{..}, release_due_at, name: "release")

This could then be scheduled after construction:

Scheduler.schedule_batch(batch)

Batch schedule multiple commands in a single request

It's currently only possible to schedule one-off and recurring commands one at a time.

It would be useful to allow scheduling multiple commands, grouped together, to provide an "all or none" guarantee.

Example

Scheduler.batch(reservation_id, fn batch ->
  batch
  |> Scheduler.schedule_once(%TimeoutReservation{..}, timeout_due_at)
  |> Scheduler.schedule_once(%ReleaseSeat{..}, release_due_at)
end)

Schedule idle timeout

Schedule aggregate should specify a life span allowing instances to be shutdown after an inactivity period.

Rescheduling commands

It is currently not possible to schedule a command using schedule_uuid that was previously cancelled:

iex> Scheduler.schedule_once(seat_id, %TimeoutSeatReservation{...}, ~N[2020-01-01 12:00:00])
:ok
iex> Scheduler.cancel_schedule(seat_id)
:ok
iex> Scheduler.schedule_once(seat_id, %TimeoutSeatReservation{...}, ~N[2020-01-01 14:00:00])
{:error, :already_scheduled}

This is because command handlers for scheduling are allowed only when the schedule aggregate instance does not exist yet:

  def execute(%Schedule{schedule_uuid: nil} = schedule, %ScheduleOnce{} = once) do
    ...
  end

  def execute(%Schedule{}, %ScheduleOnce{}), do: {:error, :already_scheduled}

https://github.com/commanded/commanded-scheduler/blob/master/lib/commanded/scheduler/schedule/schedule.ex#L27

This means that if we need rescheduling, we need to generate a new random schedule_uuid for each new schedule and track it client side. This makes it a bit more difficult and doesn't guarantee schedule uniqueness (in the example above that we have a single active schedule per single seat), e.g.:

iex> Scheduler.schedule_once(seat_id, %TimeoutSeatReservation{...}, ~N[2020-01-01 12:00:00])
:ok
iex> Scheduler.cancel_schedule(seat_id)
:ok
iex> Scheduler.schedule_once(seat_id, %TimeoutSeatReservation{...}, ~N[2020-01-01 14:00:00])
:ok
iex> Scheduler.schedule_once(seat_id, %TimeoutSeatReservation{...}, ~N[2020-01-01 16:00:00])
{:error, :already_scheduled}

I believe it might also be useful to have some kind of rescheduling of a command, something like:

Scheduler.reschedule(seat_id, ~N[2020-01-01 14:00:00])