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);
    }
}

Kotlin

import io.cratis.chronicle.IEventStore
import io.cratis.chronicle.eventSequences.AppendOptions
import io.cratis.chronicle.eventSequences.EventSequenceNumber
import io.cratis.chronicle.eventSequences.concurrency.ConcurrencyScopeBuilder
import io.cratis.chronicle.events.EventType

@EventType
data class ConcurrencyStockReserved(val sku: String = "", val quantity: Int = 0)

/**
 * Appends only if the event source is still at the expected sequence number — the kernel
 * rejects the append with a concurrency violation if another writer got there first.
 */
suspend fun reserveStockIfUnchanged(store: IEventStore, sku: String, expectedSequenceNumber: Long) {
    val scope = ConcurrencyScopeBuilder()
        .withSequenceNumber(EventSequenceNumber(expectedSequenceNumber))
        .withEventSourceId()
        .build()

    val result = store.eventLog.append(sku, ConcurrencyStockReserved(sku, 1), AppendOptions(concurrencyScope = scope))
    val violation = result.concurrencyViolation
    if (!result.isSuccess && violation != null) {
        println("Concurrency violation: expected ${violation.expectedSequenceNumber}, actual ${violation.actualSequenceNumber}")
    }
}

Java

import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.eventSequences.AppendOptions;
import io.cratis.chronicle.eventSequences.AppendResult;
import io.cratis.chronicle.eventSequences.concurrency.ConcurrencyScope;
import io.cratis.chronicle.eventSequences.concurrency.ConcurrencyScopeBuilder;
import io.cratis.chronicle.eventSequences.concurrency.ConcurrencyViolation;

import io.cratis.chronicle.java.ConcurrencyScopeBuilderJavaBridge;
import io.cratis.chronicle.java.EventLogJavaBridge;

@EventType
record ConcurrencyStockReserved(String sku, int quantity) {}

class EventsConcurrencyBasic {
    // Appends only if the event source is still at the expected sequence number — the kernel
    // rejects the append with a concurrency violation if another writer got there first.
    void reserveStockIfUnchanged(EventStore store, String sku, long expectedSequenceNumber) {
        ConcurrencyScope scope = ConcurrencyScopeBuilderJavaBridge
            .withSequenceNumber(new ConcurrencyScopeBuilder(), expectedSequenceNumber)
            .withEventSourceId()
            .build();

        AppendResult result = EventLogJavaBridge.append(
            store.getEventLog(), sku, new ConcurrencyStockReserved(sku, 1), new AppendOptions(null, scope));

        ConcurrencyViolation violation = result.getConcurrencyViolation();
        if (!result.isSuccess() && violation != null) {
            System.out.println("Concurrency violation for event source: " + violation.getEventSourceId());
        }
    }
}

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.

The First Append Into a Scope Is Not Checked

A concurrency check needs an expected sequence number to compare against, and the optimistic strategy gets one by reading the tail through the scope’s own narrowing. When nothing matching that narrowing has been appended yet there is no tail to read, so the strategy reports EventSequenceNumber.Unavailable — the same value that means “no expected sequence number was supplied at all”. The kernel cannot tell those two apart, so it skips the check, logs a warning, and lets the append through.

This is not limited to narrow scopes — it is every first append into a scope:

• The first event on a new event source has no tail, so its append is not checked.
• The first event carrying a new EventSourceType, EventStreamType, or EventStreamId is not checked either, even when the event source itself already has plenty of events under a different narrowing.

The second case is the one to watch. It is the append most exposed to a race — it opens a new partition on a stream that other writers are already using — and it is the one the check does not cover.

Expressing “This Must Not Exist Yet”

A concurrency scope protects a sequence you are extending. It cannot express “no event matching this scope may exist yet”, because EventSequenceNumber has no value meaning before the first event.

When two writers must not both create the first event in a scope, enforce that with a constraint rather than a concurrency scope. A unique event type constraint permits exactly one event of a given type per event source, and the kernel rejects the second append no matter which client made it.

Best Practices

1. Use specific scoping: Scope concurrency as narrowly as possible to avoid unnecessary blocking — but note that the narrower the scope, the more scopes there are, and the first append into each one is not checked
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