From ca4f988af876571f8b910a2f4a623d12fb579836 Mon Sep 17 00:00:00 2001 From: Yordis Prieto Date: Wed, 19 Nov 2025 19:09:21 -0500 Subject: [PATCH] fix: add concurrency validation for Ecto projections Signed-off-by: Yordis Prieto --- guides/explanations/ecto-projections.md | 169 +++++ guides/explanations/read-model-projections.md | 2 +- .../howtos/building-read-models-with-ecto.md | 619 +++++++----------- .../ecto-projections-getting-started.md | 2 +- lib/commanded/projections/ecto.ex | 74 ++- test/projections/ecto_projection_test.exs | 232 +++++++ 6 files changed, 714 insertions(+), 384 deletions(-) create mode 100644 guides/explanations/ecto-projections.md diff --git a/guides/explanations/ecto-projections.md b/guides/explanations/ecto-projections.md new file mode 100644 index 00000000..dfc631a1 --- /dev/null +++ b/guides/explanations/ecto-projections.md @@ -0,0 +1,169 @@ +# Ecto Projections + +Ecto projections allow you to build read models from domain events using Ecto as the database layer. They provide automatic idempotency guarantees, transaction support, and efficient batch processing. + +## What are Ecto Projections? + +An Ecto projection is a specialized event handler that projects domain events into a relational database using Ecto. Unlike regular event handlers, Ecto projections: + +- **Guarantee idempotency** - Each event is projected exactly once using watermark-based tracking +- **Use database transactions** - All operations execute atomically via `Ecto.Multi` +- **Support batch processing** - Process multiple events in a single transaction for high throughput + +## Architecture + +### Event Handler Foundation + +Ecto projections are built on top of `Commanded.Event.Handler`, which means they: + +- Run as supervised GenServer processes +- Subscribe to the event store +- Receive events in order (within a single handler instance) +- Support the same lifecycle callbacks (init, error handling, etc.) + +### Idempotency Mechanism + +Ecto projections use a **watermark-based idempotency** strategy: + +1. A `projection_versions` table tracks the last seen event number for each projector +2. Before processing an event, the projector checks: `event_number > last_seen_event_number` +3. If true, the event is processed and the watermark is updated +4. If false, the event is skipped (already processed) + +This approach is: +- **Simple** - Single integer comparison +- **Fast** - One row per projector (not one row per event) +- **Correct** - Works perfectly for sequential processing + +### Transaction Semantics + +All projection operations happen within a database transaction: + +```elixir +Ecto.Multi.new() +|> Ecto.Multi.run(:track_projection_version, fn -> update_watermark() end) +|> Ecto.Multi.insert(:my_data, changeset) # Your projection logic +|> Repo.transaction() +``` + +If any step fails, the entire transaction rolls back, including the watermark update. This ensures consistency. + +### After-Update Callbacks + +The `after_update/3` and `after_update_batch/2` callbacks execute **AFTER** the transaction commits: + +- **Side effects only** - Use for notifications, pub/sub, external API calls +- **Cannot rollback** - Database changes are already committed +- **Errors propagate** - But the data is already saved + +This design prevents long-running side effects from blocking the transaction. + +## Batch Processing + +Batch processing allows high throughput by processing multiple events in a single database transaction: + +```elixir +use Commanded.Projections.Ecto, + batch_size: 50 # Process 50 events per transaction +``` + +**How it works:** + +1. Collect up to `batch_size` events from subscription +2. Start transaction +3. Lock projection version row (`FOR UPDATE`) +4. Filter events: keep only those with `event_number > watermark` +5. Update watermark to highest event number in batch +6. Execute user's projection logic for all unseen events +7. Commit transaction + +**Benefits:** + +- Reduced transaction overhead (1 transaction for N events instead of N transactions) +- Single fsync for the entire batch +- Better throughput for high-volume event streams + +**Trade-offs:** + +- Higher latency (wait for batch to fill or timeout) +- Mutually exclusive with concurrency +- Requires static schema prefix (no per-event dynamic schemas) + +## Why Concurrency Is Not Supported + +> #### Concurrency Not Supported {: .error} +> +> Ecto projections do not support `concurrency > 1` due to the watermark-based idempotency mechanism. + +**The Problem:** + +With concurrent workers processing events in parallel: + +``` +Time Worker 1 Worker 2 Watermark +---- -------- -------- --------- +T1 Event #3 Event #5 0 +T2 Updates: 3 Updates: 5 5 (Worker 2 commits first) +T3 Event #4 arrives 5 +T4 SKIPPED (4 < 5) ❌ 5 +``` + +Event #4 is permanently lost because the watermark already moved past it. + +**Why Regular Event Handlers Can Use Concurrency:** + +Regular event handlers don't use watermark idempotency - they rely on the event store subscription's checkpoint. With `partition_by/2`, they guarantee per-partition ordering while allowing cross-partition concurrency. + +**The Solution for Ecto Projections:** + +Use `:batch_size` instead of `:concurrency`: +- Maintains event ordering +- Provides high throughput +- Safe with watermark idempotency + +## Schema Prefixes + +Schema prefixes allow multi-tenant projections where each tenant's data lives in a separate PostgreSQL schema: + +```elixir +def schema_prefix(%Event{tenant: tenant}, _metadata), do: tenant +``` + +The `projection_versions` table will be read/written in the tenant's schema, ensuring complete isolation. + +> #### Batch Processing Limitation {: .warning} +> +> Batch projectors only support **static** schema prefixes (strings), not dynamic functions. +> This is because the entire batch must use the same schema - we can't mix events from +> different tenants in a single transaction. + +## Design Decisions + +### Why Watermark Instead of Per-Event Tracking? + +**Watermark approach:** +- Storage: 1 row per projector +- Lookup: Single integer comparison +- Write: 1 update per event/batch + +**Per-event tracking:** +- Storage: 1 row per projector per event (can be millions) +- Lookup: Check if event_number exists in table +- Write: 1 insert per event + +The watermark approach is simpler and more efficient for the 99% case (sequential processing). Per-event tracking would be needed only if we wanted to support concurrent processing in the future. + +### Why Batch Processing Over Concurrency? + +Batch processing provides: +- Ordering guarantees (required for correctness) +- High throughput (fewer transactions) +- Simpler code (single watermark update) + +Concurrency would require: +- Per-partition watermarks +- Complex coordination logic +- Higher storage overhead + +For read models, throughput via batching is sufficient. If you need parallel processing, consider multiple projectors subscribing to different streams. + diff --git a/guides/explanations/read-model-projections.md b/guides/explanations/read-model-projections.md index 34679210..ad145193 100644 --- a/guides/explanations/read-model-projections.md +++ b/guides/explanations/read-model-projections.md @@ -4,7 +4,7 @@ Your read model can be built using a Commanded event handler and whichever stora ## Ecto projections -Commanded includes built-in support for building read models using Ecto with one of the databases supported by Ecto (PostgreSQL, MySQL, et al). See the [Ecto Projections Getting Started](ecto-projections-getting-started.html) and [Building Read Models with Ecto](building-read-models-with-ecto.html) guides for details. +Commanded includes built-in support for building read models using Ecto with one of the databases supported by Ecto (PostgreSQL, MySQL, et al). See the [Ecto Projections Getting Started](../howtos/ecto-projections-getting-started.html) guide to set up, the [Building Read Models with Ecto](../howtos/building-read-models-with-ecto.html) how-to for practical examples, and the [Ecto Projections](ecto-projections.html) explanation for architectural details. ### Example diff --git a/guides/howtos/building-read-models-with-ecto.md b/guides/howtos/building-read-models-with-ecto.md index 6f62e5a5..071c4bff 100644 --- a/guides/howtos/building-read-models-with-ecto.md +++ b/guides/howtos/building-read-models-with-ecto.md @@ -1,496 +1,365 @@ -# Building read models with Ecto projections +# Building Read Models with Ecto Projections -## Creating a read model +Practical guide for common projection tasks. For conceptual understanding, see the [Ecto Projections explanation guide](../explanations/ecto-projections.html). -Use `Ecto.Schema` to define one or more read models: +## Create a Basic Projection + +**1. Define your read model schema:** ```elixir -defmodule ExampleProjection do +defmodule MyApp.Accounts.Projections.Account do use Ecto.Schema - schema "example_projections" do - field(:name, :string) + schema "accounts" do + field :account_number, :string + field :balance, :integer + + timestamps() end end ``` -## Creating a projector - -For each read model you will need to define a module that uses the `Commanded.Projections.Ecto` module and projects the appropriate domain events with the `project` macro. - -You must specify the following options when defining or starting an Ecto projector: - -- `:application` - (module or atom) the Commanded application (e.g. `MyApp.Application`). -- `:name` - (string) a unique name used to identify the event store subscription used by the projector. -- `:repo` - (module) an Ecto repo (e.g. `MyApp.Projections.Repo`). - -Once a projector has been deployed you _should not_ change its name. Doing so will cause a new event store subscription to be created and replay all existing events. - -**Note:** A read model projector is just a specialised Commanded event handler `GenServer` process. - -### Example +**2. Create the projector module:** ```elixir -defmodule MyApp.ExampleProjector do +defmodule MyApp.Accounts.Projectors.AccountProjector do use Commanded.Projections.Ecto, application: MyApp.Application, - repo: MyApp.Projections.Repo, - name: "example_projection" + repo: MyApp.Repo, + name: "account_projector" - project %AnEvent{name: name}, _metadata, fn multi -> - Ecto.Multi.insert(multi, :example_projection, %ExampleProjection{name: name}) - end + alias MyApp.Accounts.Projections.Account + alias MyApp.Accounts.Events.{AccountOpened, MoneyDeposited} - project %AnotherEvent{name: name}, fn multi -> - Ecto.Multi.insert(multi, :example_projection, %ExampleProjection{name: name}) + project %AccountOpened{account_number: number, initial_balance: balance}, fn multi -> + Ecto.Multi.insert(multi, :account, %Account{ + account_number: number, + balance: balance + }) end -end -``` - -#### Runtime configuration -The `:application` and `:name` options can be provided at runtime, but `:repo` must be specified at compile-time. - -```elixir -defmodule MyApp.ExampleProjector do - use Commanded.Projections.Ecto, - repo: MyApp.Projections.Repo + project %MoneyDeposited{account_number: number, amount: amount}, fn multi -> + Ecto.Multi.update_all( + multi, + :account, + where(Account, account_number: ^number), + inc: [balance: ^amount] + ) + end end ``` -Started with: - -```elixir -{:ok, pid} = ExampleProjector.start_link(application: MyApp.Application, name: "example_projection") -``` - -Or supervised: +**3. Add to your supervision tree:** ```elixir -Supervisor.start_link([ - {ExampleProjector, application: MyApp.Application, name: "example_projection"} -], strategy: :one_for_one) -``` +children = [ + MyApp.Accounts.Projectors.AccountProjector +] -Runtime configuration allows the same projector to be run more than once, with each instance using a separate application or name: - -```elixir -Supervisor.start_link([ - {ExampleProjector, application: App1, name: "App1.Projector"}, - {ExampleProjector, application: App2, name: "App2.Projector"} -], strategy: :one_for_one) +Supervisor.start_link(children, strategy: :one_for_one) ``` -### Subscription options +## Access Event Metadata -You can control how events are consumed from the event store using the `:subscription_opts` option. This provides a convenient way to group subscription-related configuration together: +Use the 3-arity version of `project/3` to access metadata: ```elixir -defmodule MyApp.ExampleProjector do - use Commanded.Projections.Ecto, - application: MyApp.Application, - repo: MyApp.Projections.Repo, - name: "example_projection", - subscription_opts: [ - start_from: :origin, # Start from the first event - subscribe_to: :all, # Subscribe to all streams - concurrency: 4 # Process with 4 workers - ] - - project %AnEvent{}, _metadata, fn multi -> - # Your projection logic - end +project %OrderPlaced{order_id: id} = event, metadata, fn multi -> + %{event_number: event_number, created_at: timestamp} = metadata + + Ecto.Multi.insert(multi, :order, %Order{ + id: id, + event_number: event_number, + placed_at: timestamp + }) end ``` -#### Available subscription options - -- **`:start_from`** - Where to begin reading events - - `:origin` - Start from the very first event (use for new projections) - - `:current` - Start from the current position (skip historical events) - - Positive integer - Start from a specific event number +## Use Batch Processing for High Throughput -- **`:subscribe_to`** - Which event stream(s) to subscribe to - - `:all` - Subscribe to all events (default) - - Stream name (string) - Subscribe to a specific stream - -- **`:concurrency`** - Number of concurrent workers for parallel processing - - Positive integer - Enable concurrent event processing - - **Note:** Mutually exclusive with `:batch_size` - -#### Alternative: Top-level options - -You can also pass these options at the top level instead of nesting them under `:subscription_opts`. Both formats are equivalent: +**Configure batch size:** ```elixir -defmodule MyApp.ExampleProjector do +defmodule MyApp.HighVolumeProjector do use Commanded.Projections.Ecto, application: MyApp.Application, - repo: MyApp.Projections.Repo, - name: "example_projection", - start_from: :origin, - subscribe_to: :all, - concurrency: 4 -end -``` - -If you provide both nested and top-level options, the top-level ones take precedence. - -### Using the `project` macro - -The `project/3` macro expects the domain event, metadata, and a single-arity function that takes and returns an `Ecto.Multi` data structure for grouping multiple Repo operations. These will all be executed within a single transaction. You can use `Ecto.Multi` to insert, update, and delete data. - -#### Examples - -Project an event and its metadata into a read model with `project/3`: - -```elixir -project %AnEvent{name: name}, metadata, fn multi -> - projection = %ExampleProjection{name: name, metadata: metadata} - - Ecto.Multi.insert(multi, :example_projection, projection) -end -``` - -Use `project/2` if you do not need to use the event metadata: - -```elixir -project %AnotherEvent{name: name}, fn multi -> - Ecto.Multi.insert(multi, :example_projection, %ExampleProjection{name: name}) -end -``` - -If you want to skip a projection event, you can return the `multi` transaction without further modifying it: - -```elixir -project %ItemUpdated{uuid: uuid} = event, _metadata, fn multi -> - case Repo.get(ItemProjection, uuid) do - nil -> multi - item -> Ecto.Multi.update(multi, :item, update_changeset(event, item)) + repo: MyApp.Repo, + name: "high_volume_projector", + batch_size: 100 + + # Use project_batch instead of project + project_batch fn events, multi -> + Enum.reduce(events, multi, fn + {%EventA{id: id, data: data}, _metadata}, multi -> + Ecto.Multi.insert(multi, {:event_a, id}, %ReadModel{ + id: id, + data: data + }) + + {%EventB{id: id}, _metadata}, multi -> + Ecto.Multi.update_all(multi, {:event_b, id}, + where(ReadModel, id: ^id), + set: [processed: true] + ) + + _other, multi -> + multi # Ignore other events + end) end end ``` -### Using the `project_batch` macro - -You can use `project_batch` to receive events in batches. To enable batching, you need to set the `batch_size` and use the `project_batch/1` macro. `project_batch/1` receives a function that takes a list of `{event, metadata}` tuples for all the events in the batch and an `Ecto.Multi` structure, similar to `project/3`. +## Handle Multiple Tables in One Projection -Note that there is currently no built in way to target a single type of event to be projected, and as such a single `project_batch` macro is expected to gracefully handle (or ignore) any events that it may receive. - -#### Example +Use `Ecto.Multi` operations: ```elixir -defmodule MyApp.Projections.BatchProjector do - use Commanded.Projections.Ecto, - application: MyApp.Application, - repo: MyApp.Projections.Repo, - name: "example_batch_projection", - batch_size: 10 - - project_batch fn events, multi -> - projections = events - |> Enum.map(fn - {%AnEvent{name: name}, _metadata} -> %{name: name} - _ -> nil - end) - |> Enum.reject(&is_nil/1) - - Ecto.Multi.insert_all(multi, :example_batch_projection, Projection, projections) - end +project %UserRegistered{user_id: id, email: email}, fn multi -> + multi + |> Ecto.Multi.insert(:user, %User{id: id, email: email}) + |> Ecto.Multi.insert(:audit, %AuditLog{ + action: "user_registered", + user_id: id + }) + |> Ecto.Multi.update_all(:stats, StatsQuery, inc: [user_count: 1]) end ``` -## Supervision +## Handle Projection Errors -Your projector module must be included in your application supervision tree: +Implement the `error/3` callback: ```elixir -defmodule MyApp.Projections.Supervisor do - use Supervisor - - def start_link(init_arg) do - Supervisor.start_link(__MODULE__, init_arg, name: __MODULE__) - end - - @impl true - def init(_init_arg) do - children = [ - MyApp.ExampleProjector - ] - - Supervisor.init(children, strategy: :one_for_one) - end -end -``` - -**Warning:** You should implement an [error handling](#error-handling) strategy in your projector module when supervising to prevent problematic events from causing cascading errors due too many restarts. - -## Error handling - -### `error/3` callback - -The `Commanded.Projections.Ecto` macro defines a Commanded event handler which means you can take advantage of the [`error/3` callback function](https://hexdocs.pm/commanded/Commanded.Event.Handler.html#module-error-3-callback) to handle any errors returned from a `project` function. The error function is passed the error returned by the event handler (e.g. `{:error, error}`), the event causing the error, and a context map containing state passed between retries. Use the context map to track any transient state you need to access between retried failures, such as the number of failed attempts. - -You can return one of the following responses depending upon the error severity: - -- `{:retry, context}` - retry the failed event, provide a context map containing any state passed to subsequent failures. This could be used to count the number of failures, stopping after too many. - -- `{:retry, delay, context}` - retry the failed event, after sleeping for the requested delay (in milliseconds). Context is a map as described in `{:retry, context}` above. - -- `:skip` - skip the failed event by acknowledging receipt. - -- `{:stop, reason}` - stop the projector with the given reason. - -#### Error handling example - -Here's an example projector module where an error tagged tuple is explicitly returned from a `project` function, but you can also handle exceptions caused by faulty `Ecto.Multi` database operations in a similar manner since the errors are caught and returned as tagged tuples (e.g. `{:error, %Ecto.ConstraintError{}}`). - -```elixir -defmodule MyApp.ExampleProjector do +defmodule MyApp.ResilientProjector do use Commanded.Projections.Ecto, application: MyApp.Application, - repo: MyApp.Projections.Repo, - name: "MyApp.ExampleProjector" + repo: MyApp.Repo, + name: "resilient_projector" require Logger - alias Commanded.Event.FailureContext - project %AnEvent{}, fn _multi -> - {:error, :failed} - end + # ... projection logic ... - def error({:error, :failed}, %AnEvent{}, %FailureContext{}) do + def error({:error, %Ecto.ConstraintError{}}, event, %FailureContext{}) do + Logger.warning("Constraint violation, skipping event: #{inspect(event)}") :skip end - def error({:error, %Ecto.ConstraintError{} = error}, _event, _failure_context) do - Logger.error("Failed due to constraint error: " <> inspect(error)) - - :skip - end - - def error({:error, _error}, _event, _failure_context) do - :skip + def error({:error, reason}, event, %FailureContext{context: context}) do + attempts = Map.get(context, :attempts, 0) + 1 + + if attempts < 3 do + Logger.warning("Projection failed, retrying (attempt #{attempts})") + {:retry, 5_000, %{attempts: attempts}} + else + Logger.error("Projection failed after 3 attempts, skipping") + :skip + end end end ``` -### `after_update/3` callback +## Notify After Projection Updates -You can define an `after_update/3` callback function in a projector to be called after each projected event. The function receives the event, its metadata, and all changes from the `Ecto.Multi` struct that were executed within the database transaction. +Use the `after_update/3` callback for side effects: ```elixir -defmodule MyApp.ExampleProjector do +defmodule MyApp.NotifyingProjector do use Commanded.Projections.Ecto, application: MyApp.Application, - repo: MyApp.Projections.Repo, - name: "MyApp.ExampleProjector" - - project %AnEvent{name: name}, fn multi -> - Ecto.Multi.insert(multi, :example_projection, %ExampleProjection{name: name}) - end - - @impl Commanded.Projections.Ecto - def after_update(event, metadata, changes) do - # Use the event, metadata, or `Ecto.Multi` changes and return `:ok` + repo: MyApp.Repo, + name: "notifying_projector" + + # ... projection logic ... + + def after_update(event, metadata, _changes) do + Phoenix.PubSub.broadcast( + MyApp.PubSub, + "projections", + {:projection_updated, event, metadata} + ) + :ok end end ``` -You could use this function to notify subscribers that the read model has been updated (e.g. pub/sub to Phoenix channels). - -#### ⚠️ Transaction Semantics - -**CRITICAL:** The `after_update/3` callback executes **AFTER** the database transaction has been committed. This means: - -- **Errors cannot rollback the transaction** - If this callback returns an error or raises an exception, the projection data is already persisted in the database. -- **Use for side effects only** - This callback is designed for notifications, pub/sub, external API calls, or other side effects that should happen after successful projection updates. -- **Error handling implications** - Returning `{:error, reason}` or raising an exception will propagate the error up, but the database changes are permanent. The event handler may retry, potentially causing duplicate side effects. - -If you need to perform validation or operations that should prevent the projection from being saved, do so within your `project/2` function (inside the `Ecto.Multi`), not in this callback. - -### `after_update_batch/2` callback - -Similarly for batching projectors, you can define an `after_update_batch/2` callback function in a projector to be called after a batch of events has been projected. The function receives a list of `{event, metadata}` tuples for each processed event and all changes from the `Ecto.Multi` struct. +For batch projectors, use `after_update_batch/2`: ```elixir -defmodule MyApp.BatchProjector do - use Commanded.Projections.Ecto, - application: MyApp.Application, - repo: MyApp.Projections.Repo, - name: "MyApp.BatchProjector", - batch_size: 10 - - project_batch fn events, multi -> - # Your batch projection logic - end - - @impl Commanded.Projections.Ecto - def after_update_batch(events, changes) do - # Notify subscribers about the batch update - :ok - end +def after_update_batch(events, _changes) do + count = length(events) + + Phoenix.PubSub.broadcast( + MyApp.PubSub, + "projections", + {:batch_projected, count} + ) + + :ok end ``` -#### ⚠️ Transaction Semantics - -**CRITICAL:** The `after_update_batch/2` callback executes **AFTER** the database transaction has been committed. This means: +## Configure Multi-Tenant Projections -- **Errors cannot rollback the transaction** - If this callback returns an error or raises an exception, the projection data is already persisted in the database. -- **Use for side effects only** - This callback is designed for notifications, pub/sub, external API calls, or other side effects that should happen after successful projection updates. -- **Error handling implications** - Returning `{:error, reason}` or raising an exception will propagate the error up, but the database changes are permanent. The event handler may retry, potentially causing duplicate side effects. +**Option 1: Static schema prefix** -If you need to perform validation or operations that should prevent the projection from being saved, do so within your `project_batch/2` function (inside the `Ecto.Multi`), not in this callback. - -## Schema prefix +```elixir +use Commanded.Projections.Ecto, + application: MyApp.Application, + repo: MyApp.Repo, + name: "tenant_a_projector", + schema_prefix: "tenant_a" +``` -When using a prefix for your Ecto schemas you might also want to change the prefix for the `ProjectionVersion` schema. There are a number of options to do this: +**Option 2: Dynamic per-event prefix** -1. Define a global static prefix via environment config: +```elixir +use Commanded.Projections.Ecto, + application: MyApp.Application, + repo: MyApp.Repo, + name: "multi_tenant_projector" - ```elixir - # config/config.exs - config :commanded, Commanded.Projections.Ecto, - schema_prefix: "example_schema_prefix" - ``` +def schema_prefix(%_{tenant_id: tenant_id}, _metadata) do + "tenant_#{tenant_id}" +end +``` -2. Provide a static `schema_prefix` as a projector option: +**Option 3: Function-based prefix** - ```elixir - defmodule MyApp.ExampleProjector do - use Commanded.Projections.Ecto, - application: MyApp.Application, - repo: MyApp.Projections.Repo, - name: "example_projection", - schema_prefix: "example_schema_prefix" - end - ``` +```elixir +use Commanded.Projections.Ecto, + application: MyApp.Application, + repo: MyApp.Repo, + name: "function_prefix_projector", + schema_prefix: fn event, _metadata -> + determine_schema(event) + end +``` -3. Provide a one-arity function as a `schema_prefix` projector option: +## Subscribe to Specific Streams - ```elixir - defmodule MyApp.ExampleProjector do - use Commanded.Projections.Ecto, - application: MyApp.Application, - repo: MyApp.Projections.Repo, - name: "example_projection", - schema_prefix: fn event -> "example_schema_prefix" end - end - ``` +```elixir +use Commanded.Projections.Ecto, + application: MyApp.Application, + repo: MyApp.Repo, + name: "account_projector", + subscription_opts: [ + subscribe_to: "account-*", + start_from: :origin + ] +``` - The function will receive the event as the single argument allowing you to use the same or a different schema for each event. +Or use top-level options: -4. Provide a two-arity function as a `schema_prefix` projector option: +```elixir +use Commanded.Projections.Ecto, + application: MyApp.Application, + repo: MyApp.Repo, + name: "account_projector", + subscribe_to: "account-*", + start_from: :origin +``` - ```elixir - defmodule MyApp.ExampleProjector do - use Commanded.Projections.Ecto, - application: MyApp.Application, - repo: MyApp.Projections.Repo, - name: "example_projection", - schema_prefix: fn event, metadata -> "example_schema_prefix" end - end - ``` +## Rebuild a Projection - The function will receive the event and its associated metadata as the two arguments allowing you to use the same or a different schema for each event. The metadata will also include the enriched fields such as the application, event handler name, and optional handler state. +**1. Stop the projector** -5. Define a `schema_prefix/1` callback function: +**2. Delete the projection version:** - ```elixir - defmodule MyApp.ExampleProjector do - use Commanded.Projections.Ecto, - application: MyApp.Application, - name: "example_projection" +```sql +DELETE FROM projection_versions +WHERE projection_name = 'account_projector'; +``` - @impl Commanded.Projections.Ecto - def schema_prefix(event), do: "example_schema_prefix" - end - ``` +**3. Clear the read model tables:** - The function will receive the event as the single argument allowing you to use the same or a different schema for each event. +```sql +TRUNCATE TABLE accounts RESTART IDENTITY CASCADE; +``` - An example usage could be for tenant specific projections where each tenant's data is projected and stored in a separate database schema: +**4. Reset the event store subscription:** - ```elixir - @impl Commanded.Projections.Ecto - def schema_prefix(%_{tenant: tenant}), do: tenant - ``` +For EventStore adapter: -6. Define a `schema_prefix/2` callback function: +```elixir +MyApp.EventStore.delete_subscription("account_projector") +``` - ```elixir - defmodule MyApp.ExampleProjector do - use Commanded.Projections.Ecto, - application: MyApp.Application, - name: "example_projection" +**5. Restart the projector** - @impl Commanded.Projections.Ecto - def schema_prefix(event, metadata), do: "example_schema_prefix" - end - ``` +The projector will replay all events from the beginning. - The function will receive the event and its associated metadata as the two arguments allowing you to use the same or a different schema for each event. The metadata will also include the enriched fields such as the application, event handler name, and optional handler state. +## Configure Runtime Options - An example usage could be for tenant specific projections where each tenant's data is projected and stored in a separate database schema: +Define projector without compile-time config: - ```elixir - @impl Commanded.Projections.Ecto - def schema_prefix(%_{tenant: tenant}, _metadata), do: tenant - ``` +```elixir +defmodule MyApp.RuntimeProjector do + use Commanded.Projections.Ecto, + repo: MyApp.Repo + + # ... projection logic ... +end +``` -### Migrations with a schema prefix +Start with runtime configuration: -1. Generate an Ecto migration in your app: +```elixir +{:ok, pid} = MyApp.RuntimeProjector.start_link( + application: MyApp.Application, + name: "runtime_projector" +) +``` - ```shell - mix ecto.gen.migration create_schema_projection_versions - ``` +Or in supervision tree: -2. Modify the generated migration, in `priv/repo/migrations`, to create the schema and a `projection_versions` table for the schema: +```elixir +children = [ + {MyApp.RuntimeProjector, + application: MyApp.Application, + name: "runtime_projector"} +] +``` - ```elixir - defmodule CreateSchemaProjectionVersions do - alias Commanded.Projections.Ecto.Migrations.V01CreateProjectionVersionsTable +## Skip Events Conditionally - use Ecto.Migration +Return the multi unchanged: - def up do - execute("CREATE SCHEMA example_schema_prefix") - V01CreateProjectionVersionsTable.up(prefix: "example_schema_prefix") - end +```elixir +project %ItemUpdated{id: id} = event, _metadata, fn multi -> + case Repo.get(Item, id) do + nil -> + multi # Skip - item doesn't exist + + item -> + changeset = update_changeset(item, event) + Ecto.Multi.update(multi, :item, changeset) + end +end +``` - def down do - V01CreateProjectionVersionsTable.down(prefix: "example_schema_prefix") - execute("DROP SCHEMA example_schema_prefix CASCADE") - end - end - ``` +## Common Options Reference - Note you will need to do this for each schema prefix you use. +### Required Options -## Rebuilding a projection +- `:application` - The Commanded application module +- `:name` - Unique projector name (string) +- `:repo` - Ecto repo module -The `projection_versions` table is used to ensure that events are only projected once. +### Subscription Options -To rebuild a projection you will need to: +- `:start_from` - `:origin`, `:current`, or event number +- `:subscribe_to` - `:all` or stream name/pattern -1. Delete the row containing the last seen event for the projection name: +### Performance Options - ```SQL - DELETE FROM projection_versions - WHERE projection_name = 'example_projection'; - ``` +- `:batch_size` - Number of events per transaction (integer) +- `:timeout` - Transaction timeout in milliseconds -2. Truncate the tables that are being populated by the projection, and restart their identity: +### Schema Options - ```SQL - TRUNCATE TABLE - example_projections, - other_projections - RESTART IDENTITY; - ``` +- `:schema_prefix` - String, 1-arity function, or 2-arity function -You will also need to reset the event store subscription for the commanded event handler. This is specific to whichever event store you are using. +See the [Ecto Projections explanation guide](../explanations/ecto-projections.html) for architectural details and the moduledoc for complete API reference. diff --git a/guides/howtos/ecto-projections-getting-started.md b/guides/howtos/ecto-projections-getting-started.md index 8aa25717..53f5d375 100644 --- a/guides/howtos/ecto-projections-getting-started.md +++ b/guides/howtos/ecto-projections-getting-started.md @@ -72,4 +72,4 @@ Commanded includes built-in support for read model projections using Ecto. You s end ``` -Refer to the [Building Read Models with Ecto](building-read-models-with-ecto.html) guide for more detail on how to configure and use a read model projector. +See the [Building Read Models with Ecto](building-read-models-with-ecto.html) how-to guide for practical examples, and the [Ecto Projections](../explanations/ecto-projections.html) explanation guide for architectural details. diff --git a/lib/commanded/projections/ecto.ex b/lib/commanded/projections/ecto.ex index eed81c70..40328f24 100644 --- a/lib/commanded/projections/ecto.ex +++ b/lib/commanded/projections/ecto.ex @@ -61,10 +61,27 @@ if Code.ensure_loaded?(Ecto) do - Only static `:schema_prefix` strings allowed (no dynamic functions) - Uses watermark-based idempotency for efficient batch processing + ## Concurrency Limitation + + > #### Concurrency Not Supported {: .error} + > + > Ecto projections do NOT support `:concurrency > 1` due to the risk of silent + > data loss from out-of-order event processing. + > + > **Why?** When events are processed concurrently across multiple workers, they may + > arrive out of order (e.g., event #5 processed before event #4). The watermark-based + > idempotency check (`last_seen_event_number`) will skip the earlier event permanently, + > causing data loss. + > + > **Solution:** Use `:batch_size` instead of `:concurrency` for high-throughput + > processing. Batch processing maintains event ordering while providing excellent + > performance by processing multiple events in a single database transaction. + ## Guides - [Getting started](ecto-projections-getting-started.html) - - [Building read models](building-read-models-with-ecto.html) + - [Building read models](building-read-models-with-ecto.html) (how-to) + - [Ecto Projections](ecto-projections.html) (explanation) """ @@ -80,10 +97,22 @@ if Code.ensure_loaded?(Ecto) do {_, nil} -> :ok - {_batch_size, _concurrency} -> + {_, 1} -> + # concurrency: 1 is the default (sequential) and doesn't conflict with batch_size + :ok + + {_batch_size, concurrency} when is_integer(concurrency) and concurrency > 1 -> {:error, - "cannot use both :batch_size and :concurrency options - they are mutually exclusive. " <> + "cannot use both :batch_size and :concurrency > 1 - they are mutually exclusive. " <> "Use :batch_size for batch processing OR :concurrency for concurrent processing, but not both."} + + _ -> + # Allow other combinations: + # - batch_size without concurrency + # - concurrency without batch_size + # - batch_size with concurrency: 1 (already handled above) + # Note: Invalid concurrency values are caught by validate_concurrency_compatibility/1 + :ok end end @@ -99,6 +128,36 @@ if Code.ensure_loaded?(Ecto) do end end + @doc false + def validate_concurrency_compatibility(opts) do + concurrency = Keyword.get(opts, :concurrency) + + case concurrency do + nil -> + :ok + + 1 -> + :ok + + value when is_integer(value) and value > 1 -> + {:error, + "Ecto projections do not support :concurrency > 1 due to out-of-order event processing that can cause silent data loss. " <> + "Events processed concurrently may arrive out of order (e.g., event #5 before event #4), " <> + "causing the watermark-based idempotency check to permanently skip earlier events. " <> + "Use :batch_size instead for high-throughput processing with ordering guarantees."} + + value when is_integer(value) -> + {:error, + "Invalid :concurrency value #{inspect(value)}. " <> + "Concurrency must be a positive integer (1 or greater)."} + + invalid_value -> + {:error, + "Invalid :concurrency value #{inspect(invalid_value)}. " <> + "Expected a positive integer, got: #{inspect(invalid_value)}"} + end + end + defp __define_update_projection__ do quote do def update_projection(event, metadata, multi_fn) do @@ -360,13 +419,16 @@ if Code.ensure_loaded?(Ecto) do Application.get_env(:commanded, Commanded.Projections.Ecto, []) |> Keyword.get(:schema_prefix) - # Validate mutual exclusivity + case validate_concurrency_compatibility(opts) do + :ok -> :ok + {:error, message} -> raise CompileError, description: message + end + case validate_mutual_exclusivity(opts) do :ok -> :ok {:error, message} -> raise CompileError, description: message end - # Validate batch schema prefix compatibility case validate_batch_schema_prefix_compatibility(batch_size, schema_prefix) do :ok -> :ok {:error, message} -> raise CompileError, description: message @@ -380,8 +442,6 @@ if Code.ensure_loaded?(Ecto) do Application.compile_env(:commanded, [Commanded.Projections.Ecto, :repo]) || raise("Commanded Ecto projections expects :repo to be configured in environment") @timeout @opts[:timeout] || :infinity - - # Pass through any other configuration to the event handler @handler_opts Keyword.drop(@opts, [:repo, :schema_prefix, :timeout]) unquote(__include_schema_prefix__(schema_prefix)) diff --git a/test/projections/ecto_projection_test.exs b/test/projections/ecto_projection_test.exs index 8b255643..f9736c8d 100644 --- a/test/projections/ecto_projection_test.exs +++ b/test/projections/ecto_projection_test.exs @@ -190,4 +190,236 @@ defmodule Commanded.Projections.EctoProjectionTest do UnnamedProjector.start_link() end end + + describe "concurrency validation" do + test "should allow concurrency: 1" do + defmodule TestConcurrency1Projector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestConcurrency1Projector", + repo: Commanded.Projections.Repo, + concurrency: 1 + + project(%AnEvent{name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :projection, %Projection{name: name}) + end) + end + + assert Code.ensure_loaded?(TestConcurrency1Projector) + end + + test "should allow no concurrency option" do + defmodule TestNoConcurrencyProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestNoConcurrencyProjector", + repo: Commanded.Projections.Repo + + project(%AnEvent{name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :projection, %Projection{name: name}) + end) + end + + assert Code.ensure_loaded?(TestNoConcurrencyProjector) + end + + test "should reject concurrency > 1 in top-level options" do + assert_raise CompileError, ~r/Ecto projections do not support :concurrency > 1/, fn -> + defmodule TestConcurrencyRejectedProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestConcurrencyRejectedProjector", + repo: Commanded.Projections.Repo, + concurrency: 4 + + project(%AnEvent{name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :projection, %Projection{name: name}) + end) + end + end + end + + test "error message should explain the risk" do + error = + assert_raise CompileError, fn -> + defmodule TestErrorMessageProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestErrorMessageProjector", + repo: Commanded.Projections.Repo, + concurrency: 2 + + project(%AnEvent{name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :projection, %Projection{name: name}) + end) + end + end + + assert error.description =~ "out-of-order event processing" + assert error.description =~ "silent data loss" + assert error.description =~ ":batch_size" + end + + test "should explain batch_size as alternative" do + error = + assert_raise CompileError, fn -> + defmodule TestBatchSizeAlternativeProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestBatchSizeAlternativeProjector", + repo: Commanded.Projections.Repo, + concurrency: 5 + + project(%AnEvent{name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :projection, %Projection{name: name}) + end) + end + end + + assert error.description =~ "Use :batch_size instead" + end + end + + describe "batch_size and concurrency mutual exclusivity" do + test "should allow batch_size with concurrency: 1" do + defmodule TestBatchAndConcurrency1Projector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestBatchAndConcurrency1Projector", + repo: Commanded.Projections.Repo, + batch_size: 10, + concurrency: 1 + + project_batch(fn events, multi -> + Enum.reduce(events, multi, fn {%AnEvent{name: name}, _metadata}, multi -> + Ecto.Multi.insert(multi, {:projection, name}, %Projection{name: name}) + end) + end) + end + + assert Code.ensure_loaded?(TestBatchAndConcurrency1Projector) + end + + test "should reject batch_size with concurrency > 1" do + assert_raise CompileError, ~r/Ecto projections do not support :concurrency > 1/, fn -> + defmodule TestBatchAndConcurrencyRejectedProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestBatchAndConcurrencyRejectedProjector", + repo: Commanded.Projections.Repo, + batch_size: 10, + concurrency: 4 + + project_batch(fn events, multi -> + Enum.reduce(events, multi, fn {%AnEvent{name: name}, _metadata}, multi -> + Ecto.Multi.insert(multi, {:projection, name}, %Projection{name: name}) + end) + end) + end + end + end + end + + describe "invalid concurrency values" do + test "should reject atom concurrency value" do + assert_raise CompileError, ~r/Invalid :concurrency value :foo/, fn -> + defmodule TestAtomConcurrencyProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestAtomConcurrencyProjector", + repo: Commanded.Projections.Repo, + concurrency: :foo + + project(%AnEvent{name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :projection, %Projection{name: name}) + end) + end + end + end + + test "should reject string concurrency value" do + assert_raise CompileError, ~r/Invalid :concurrency value "2"/, fn -> + defmodule TestStringConcurrencyProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestStringConcurrencyProjector", + repo: Commanded.Projections.Repo, + concurrency: "2" + + project(%AnEvent{name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :projection, %Projection{name: name}) + end) + end + end + end + + test "should reject list concurrency value" do + assert_raise CompileError, ~r/Invalid :concurrency value \[1, 2\]/, fn -> + defmodule TestListConcurrencyProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestListConcurrencyProjector", + repo: Commanded.Projections.Repo, + concurrency: [1, 2] + + project(%AnEvent{name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :projection, %Projection{name: name}) + end) + end + end + end + + test "should reject zero concurrency value" do + assert_raise CompileError, ~r/Invalid :concurrency value 0/, fn -> + defmodule TestZeroConcurrencyProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestZeroConcurrencyProjector", + repo: Commanded.Projections.Repo, + concurrency: 0 + + project(%AnEvent{name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :projection, %Projection{name: name}) + end) + end + end + end + + test "should reject negative concurrency value" do + assert_raise CompileError, + ~r/Invalid :concurrency value.*Expected a positive integer/, + fn -> + defmodule TestNegativeConcurrencyProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestNegativeConcurrencyProjector", + repo: Commanded.Projections.Repo, + concurrency: -1 + + project(%AnEvent{name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :projection, %Projection{name: name}) + end) + end + end + end + + test "error message should specify expected type" do + error = + assert_raise CompileError, fn -> + defmodule TestErrorMessageTypeProjector do + use Commanded.Projections.Ecto, + application: TestApplication, + name: "TestErrorMessageTypeProjector", + repo: Commanded.Projections.Repo, + concurrency: :bad_value + + project(%AnEvent{name: name}, _metadata, fn multi -> + Ecto.Multi.insert(multi, :projection, %Projection{name: name}) + end) + end + end + + assert error.description =~ "Expected a positive integer" + end + end end