Concurrency

Concurrency control in Chronicle ensures that multiple operations don’t interfere with each other when appending events to the same event source. Chronicle provides a sophisticated concurrency control mechanism through the ConcurrencyScope concept, which allows you to define precisely how concurrency should be handled based on formalized event metadata tags.

Understanding ConcurrencyScope

A ConcurrencyScope defines the boundaries and constraints for concurrent operations when appending events. It uses formalized, well-known event metadata tags that are indexed by Chronicle to scope concurrency control to specific aspects of your events, providing fine-grained control over when concurrency violations should be detected. These tags are separate from any user-controlled event tags you may add for categorization. See Event Metadata Tags for details.

Formalized Metadata Tags for Concurrency

Chronicle uses the following formalized, indexed metadata tags to scope concurrency:

• EventSourceId: Unique identifier for the event source
• EventSourceType: Overarching, binding concept (e.g., Account)
• EventStreamType: A concrete process related to event source type (e.g., Onboarding, Transactions)
• EventStreamId: A marker to separate independent streams for a stream type (e.g., Monthly, Yearly)
• EventTypes: Specific event types to scope concurrency to

Basic Usage

Simple Concurrency Scope

The most basic form of concurrency control scopes to a specific sequence number for an event source:

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.EventSequences.Concurrency;

[EventType]
public record ConcurrencyAccountOpened(string AccountName);

public class ConcurrencyBankAccountService(IEventLog eventLog)
{
    public async Task OpenAccount(EventSourceId accountId, string accountName)
    {
        var concurrencyScope = new ConcurrencyScope(
            SequenceNumber: 42,
            EventSourceId: accountId);

        await eventLog.Append(
            accountId,
            new ConcurrencyAccountOpened(accountName),
            concurrencyScope: concurrencyScope);
    }
}

Elixir

defmodule MyApp.Events.ConcurrencyAccountOpened do
  use Chronicle.Events.EventType, id: "concurrency-account-opened"

  defstruct [:account_name]
end

defmodule MyApp.ConcurrencyBankAccountService do
  alias Chronicle.Events.ConcurrencyScope
  alias MyApp.Events.ConcurrencyAccountOpened

  def open_account(account_id, account_name) do
    scope = ConcurrencyScope.for_event_source(42)

    Chronicle.append(account_id, %ConcurrencyAccountOpened{account_name: account_name},
      concurrency_scope: scope
    )
  end
end

TypeScript

import { eventType, IEventStore } from '@cratis/chronicle';

@eventType()
class ConcurrencyAccountOpened {
    constructor(readonly accountName: string) {}
}

class ConcurrencyBankAccountService {
    constructor(private readonly store: IEventStore) {}

    async openAccount(accountId: string, accountName: string): Promise<void> {
        await this.store.eventLog.append(accountId, new ConcurrencyAccountOpened(accountName), {
            concurrencyScope: {
                sequenceNumber: 42n,
                eventSourceId: true
            }
        });
    }
}

Using ConcurrencyScopeBuilder

For more complex scenarios, use the ConcurrencyScopeBuilder to fluently construct concurrency scopes:

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.EventSequences.Concurrency;

[EventType]
public record ConcurrencyMoneyDeposited(decimal Amount);

[EventType]
public record ConcurrencyMoneyWithdrawn(decimal Amount);

public class ConcurrencyAccountTransactionService(IEventLog eventLog)
{
    public async Task ProcessTransaction(EventSourceId accountId, decimal amount)
    {
        var concurrencyScope = new ConcurrencyScopeBuilder()
            .WithEventSourceId(accountId)
            .WithSequenceNumber(15)
            .WithEventStreamType("Transactions")
            .WithEventType<ConcurrencyMoneyDeposited>()
            .WithEventType<ConcurrencyMoneyWithdrawn>()
            .Build();

        await eventLog.Append(
            accountId,
            new ConcurrencyMoneyDeposited(amount),
            concurrencyScope: concurrencyScope);
    }
}

Elixir

defmodule MyApp.Events.ConcurrencyMoneyDeposited do
  use Chronicle.Events.EventType, id: "concurrency-money-deposited"

  defstruct [:amount]
end

defmodule MyApp.Events.ConcurrencyMoneyWithdrawn do
  use Chronicle.Events.EventType, id: "concurrency-money-withdrawn"

  defstruct [:amount]
end

defmodule MyApp.ConcurrencyAccountTransactionService do
  alias Chronicle.Events.ConcurrencyScope
  alias MyApp.Events.{ConcurrencyMoneyDeposited, ConcurrencyMoneyWithdrawn}

  def process_transaction(account_id, amount) do
    scope =
      ConcurrencyScope.for_event_source(15,
        event_stream_type: "Transactions",
        event_types: [ConcurrencyMoneyDeposited, ConcurrencyMoneyWithdrawn]
      )

    Chronicle.append(account_id, %ConcurrencyMoneyDeposited{amount: amount},
      concurrency_scope: scope
    )
  end
end

Scoping by Event Metadata Tags

EventSourceType and EventStreamType

You can scope concurrency to specific event source types and stream types:

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.EventSequences.Concurrency;

[EventType]
public record ConcurrencyAccountSettingsUpdated(string Settings);

public class ConcurrencyAccountManagementService(IEventLog eventLog)
{
    public async Task UpdateAccountSettings(EventSourceId accountId, string settings)
    {
        var concurrencyScope = new ConcurrencyScopeBuilder()
            .WithEventSourceId(accountId)
            .WithEventSourceType("BankAccount")
            .WithEventStreamType("AccountManagement")
            .WithSequenceNumber(10)
            .Build();

        await eventLog.Append(
            accountId,
            new ConcurrencyAccountSettingsUpdated(settings),
            eventSourceType: "BankAccount",
            eventStreamType: "AccountManagement",
            concurrencyScope: concurrencyScope);
    }
}

Elixir

defmodule MyApp.Events.ConcurrencyAccountSettingsUpdated do
  use Chronicle.Events.EventType, id: "concurrency-account-settings-updated"

  defstruct [:settings]
end

defmodule MyApp.ConcurrencyAccountManagementService do
  alias Chronicle.Events.ConcurrencyScope
  alias MyApp.Events.ConcurrencyAccountSettingsUpdated

  def update_account_settings(account_id, settings) do
    scope =
      ConcurrencyScope.for_event_source(10,
        event_source_type: "BankAccount",
        event_stream_type: "AccountManagement"
      )

    Chronicle.append(
      account_id,
      %ConcurrencyAccountSettingsUpdated{settings: settings},
      event_source_type: "BankAccount",
      event_stream_type: "AccountManagement",
      concurrency_scope: scope
    )
  end
end

EventStreamId

Use event stream IDs to scope concurrency to specific streams within a stream type:

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.EventSequences.Concurrency;

[EventType]
public record ConcurrencyMonthlyReportGenerated(string Month);

public class ConcurrencyMonthlyReportService(IEventLog eventLog)
{
    public async Task GenerateMonthlyReport(EventSourceId accountId, DateTime month)
    {
        var monthKey = month.ToString("yyyy-MM");

        var concurrencyScope = new ConcurrencyScopeBuilder()
            .WithEventSourceId(accountId)
            .WithEventStreamType("Reporting")
            .WithEventStreamId(monthKey)
            .WithSequenceNumber(5)
            .Build();

        await eventLog.Append(
            accountId,
            new ConcurrencyMonthlyReportGenerated(monthKey),
            eventStreamType: "Reporting",
            eventStreamId: monthKey,
            concurrencyScope: concurrencyScope);
    }
}

Elixir

defmodule MyApp.Events.ConcurrencyMonthlyReportGenerated do
  use Chronicle.Events.EventType, id: "concurrency-monthly-report-generated"

  defstruct [:month]
end

defmodule MyApp.ConcurrencyMonthlyReportService do
  alias Chronicle.Events.ConcurrencyScope
  alias MyApp.Events.ConcurrencyMonthlyReportGenerated

  def generate_monthly_report(account_id, month_key) do
    scope =
      ConcurrencyScope.for_event_source(5,
        event_stream_type: "Reporting",
        event_stream_id: month_key
      )

    Chronicle.append(
      account_id,
      %ConcurrencyMonthlyReportGenerated{month: month_key},
      event_stream_type: "Reporting",
      event_stream_id: month_key,
      concurrency_scope: scope
    )
  end
end

Event Types

Scope concurrency to specific event types to allow concurrent operations on different types of events:

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.EventSequences.Concurrency;

[EventType]
public record ConcurrencyPaymentProcessed(decimal Amount);

[EventType]
public record ConcurrencyPaymentFailed(decimal Amount);

[EventType]
public record ConcurrencyPaymentRefunded(decimal Amount);

public class ConcurrencyAccountService(IEventLog eventLog)
{
    public async Task ProcessPayment(EventSourceId accountId, decimal amount)
    {
        // Only check concurrency for payment-related events
        var concurrencyScope = new ConcurrencyScopeBuilder()
            .WithEventSourceId(accountId)
            .WithSequenceNumber(20)
            .WithEventType<ConcurrencyPaymentProcessed>()
            .WithEventType<ConcurrencyPaymentFailed>()
            .WithEventType<ConcurrencyPaymentRefunded>()
            .Build();

        await eventLog.Append(
            accountId,
            new ConcurrencyPaymentProcessed(amount),
            concurrencyScope: concurrencyScope);
    }
}

Elixir

defmodule MyApp.Events.ConcurrencyPaymentProcessed do
  use Chronicle.Events.EventType, id: "concurrency-payment-processed"

  defstruct [:amount]
end

defmodule MyApp.Events.ConcurrencyPaymentFailed do
  use Chronicle.Events.EventType, id: "concurrency-payment-failed"

  defstruct [:amount]
end

defmodule MyApp.Events.ConcurrencyPaymentRefunded do
  use Chronicle.Events.EventType, id: "concurrency-payment-refunded"

  defstruct [:amount]
end

defmodule MyApp.ConcurrencyAccountService do
  alias Chronicle.Events.ConcurrencyScope
  alias MyApp.Events.{ConcurrencyPaymentFailed, ConcurrencyPaymentProcessed, ConcurrencyPaymentRefunded}

  def process_payment(account_id, amount) do
    # Only check concurrency for payment-related events
    scope =
      ConcurrencyScope.for_event_source(20,
        event_types: [ConcurrencyPaymentProcessed, ConcurrencyPaymentFailed, ConcurrencyPaymentRefunded]
      )

    Chronicle.append(account_id, %ConcurrencyPaymentProcessed{amount: amount},
      concurrency_scope: scope
    )
  end
end

AppendMany with Concurrency Scopes

When appending multiple events, you can specify different concurrency scopes for different event sources:

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.EventSequences.Concurrency;

[EventType]
public record ConcurrencyMoneyWithdrawnForTransfer(decimal Amount);

[EventType]
public record ConcurrencyMoneyDepositedForTransfer(decimal Amount);

public class ConcurrencyTransferService(IEventLog eventLog)
{
    public async Task TransferMoney(EventSourceId fromAccount, EventSourceId toAccount, decimal amount)
    {
        var events = new[]
        {
            new EventForEventSourceId(fromAccount, new ConcurrencyMoneyWithdrawnForTransfer(amount)),
            new EventForEventSourceId(toAccount, new ConcurrencyMoneyDepositedForTransfer(amount))
        };

        var concurrencyScopes = new Dictionary<EventSourceId, ConcurrencyScope>
        {
            [fromAccount] = new ConcurrencyScopeBuilder()
                .WithEventSourceId(fromAccount)
                .WithSequenceNumber(50)
                .WithEventType<ConcurrencyMoneyWithdrawnForTransfer>()
                .Build(),

            [toAccount] = new ConcurrencyScopeBuilder()
                .WithEventSourceId(toAccount)
                .WithSequenceNumber(25)
                .WithEventType<ConcurrencyMoneyDepositedForTransfer>()
                .Build()
        };

        await eventLog.AppendMany(events, concurrencyScopes: concurrencyScopes);
    }
}

Elixir

defmodule MyApp.Events.ConcurrencyMoneyWithdrawnForTransfer do
  use Chronicle.Events.EventType, id: "concurrency-money-withdrawn-for-transfer"

  defstruct [:amount]
end

defmodule MyApp.Events.ConcurrencyMoneyDepositedForTransfer do
  use Chronicle.Events.EventType, id: "concurrency-money-deposited-for-transfer"

  defstruct [:amount]
end

defmodule MyApp.ConcurrencyTransferService do
  alias Chronicle.Events.ConcurrencyScope
  alias Chronicle.Transactions.UnitOfWork
  alias MyApp.Events.{ConcurrencyMoneyDepositedForTransfer, ConcurrencyMoneyWithdrawnForTransfer}

  def transfer_money(from_account, to_account, amount) do
    unit_of_work = UnitOfWork.begin()

    try do
      :ok =
        Chronicle.append(
          from_account,
          %ConcurrencyMoneyWithdrawnForTransfer{amount: amount},
          concurrency_scope: ConcurrencyScope.for_event_source(50)
        )

      :ok =
        Chronicle.append(
          to_account,
          %ConcurrencyMoneyDepositedForTransfer{amount: amount},
          concurrency_scope: ConcurrencyScope.for_event_source(25)
        )

      :ok = UnitOfWork.commit(unit_of_work)
    rescue
      exception ->
        UnitOfWork.rollback(unit_of_work)
        reraise exception, __STACKTRACE__
    end
  end
end

Event Source Operations with Concurrency

You can also use concurrency scopes with event source operations:

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.EventSequences.Operations;

[EventType]
public record ConcurrencyAccountValidated;

[EventType]
public record ConcurrencyAccountProcessed;

public class ConcurrencyBatchAccountProcessor(IEventLog eventLog)
{
    public async Task ProcessAccountBatch(EventSourceId accountId)
    {
        await eventLog
            .ForEventSourceId(accountId, source => source
                .WithConcurrencyScope(scope => scope
                    .WithSequenceNumber(30)
                    .WithEventType<ConcurrencyAccountProcessed>()
                    .WithEventType<ConcurrencyAccountValidated>())
                .Append(new ConcurrencyAccountValidated())
                .Append(new ConcurrencyAccountProcessed()))
            .Perform();
    }
}

Handling Concurrency Violations

When a concurrency violation occurs, Chronicle will return a ConcurrencyViolation in the append result:

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.EventSequences.Concurrency;

[EventType]
public record ConcurrencySafeAccountOpened(string AccountName);

public class ConcurrencySafeAccountService(IEventLog eventLog)
{
    public async Task<bool> TryOpenAccount(EventSourceId accountId, string accountName)
    {
        var concurrencyScope = new ConcurrencyScope(
            SequenceNumber: 0, // Expect this to be the first event
            EventSourceId: accountId);

        var result = await eventLog.Append(
            accountId,
            new ConcurrencySafeAccountOpened(accountName),
            concurrencyScope: concurrencyScope);

        if (result.HasConcurrencyViolations)
        {
            // result.ConcurrencyViolation describes the expected vs actual sequence number
            return false;
        }

        return result.IsSuccess;
    }
}

Concurrency Strategies

Chronicle provides built-in concurrency strategies:

Optimistic Concurrency Strategy

This strategy gets the current tail sequence number and uses it as the expected sequence number:

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.EventSequences.Concurrency;

[EventType]
public record ConcurrencyAccountNameChanged(string NewName);

// Configured automatically when using dependency injection
public class ConcurrencyOptimisticAccountService(IEventLog eventLog, IConcurrencyScopeStrategies strategies)
{
    public async Task UpdateAccount(EventSourceId accountId, string newName)
    {
        var strategy = strategies.GetFor(eventLog);
        var concurrencyScope = await strategy.GetScope(accountId);

        await eventLog.Append(
            accountId,
            new ConcurrencyAccountNameChanged(newName),
            concurrencyScope: concurrencyScope);
    }
}

Note: [EventStreamType] on a reactor or reducer scopes which events an observer receives (see Filter by event stream type) — it does not set the EventStreamType used for concurrency scoping on an append call. Set eventStreamType explicitly on Append/AppendMany, or through ConcurrencyScopeBuilder.WithEventStreamType(...), when you need concurrency scoped to a specific stream type.

Best Practices

1. Use specific scoping: Scope concurrency as narrowly as possible to avoid unnecessary blocking
2. Event type scoping: When possible, scope to specific event types to allow concurrent operations on different event types
3. Handle violations gracefully: Always check for concurrency violations and implement appropriate retry or fallback logic
4. Use builders for complex scopes: The ConcurrencyScopeBuilder provides a clear, fluent API for complex concurrency requirements
5. Use concurrency scope strategies: Inject IConcurrencyScopeStrategies to get a built-in strategy (such as optimistic concurrency) instead of hand-rolling sequence-number lookups