Transactions and Unit of Work

A unit of work groups event appends in the client and commits them as one append batch. Use it when one logical operation must produce multiple events and the operation should have one commit point, one correlation id, and one success or failure outcome.

The transaction is still an event append operation. Chronicle validates constraints and concurrency when the unit of work commits, then persists the accepted events atomically for each committed batch.

When to Use a Unit of Work

Use a unit of work when:

One command or workflow emits more than one event.
The events should share the same correlation id.
The operation should not send partial appends before the workflow is ready.
The appends should be committed or rejected together.

Do not use a unit of work to hide modeling problems. If two facts can happen independently, append them independently. If one fact depends on another, model that invariant with constraints, concurrency, or the command decision that creates the events.

Commit Transactional Appends

Create a unit of work, append through the transactional event sequence, and commit when every event has been staged.

using Cratis.Chronicle;
using Cratis.Execution;

public static class TransactionalOrderWorkflow
{
    public static async Task CommitOrder(IEventStore store)
    {
        var unitOfWork = store.UnitOfWorkManager.Begin(CorrelationId.New());

        try
        {
            await store.EventLog.Transactional.Append(
                "order-123",
                new TransactionalOrderPlaced("order-123", 99.95m));

            await store.EventLog.Transactional.Append(
                "inventory-widget",
                new TransactionalInventoryReserved("widget", 1));

            await unitOfWork.Commit();
        }
        catch
        {
            await unitOfWork.Rollback();
            throw;
        }
    }
}

public record TransactionalOrderPlaced(string OrderId, decimal TotalAmount);
public record TransactionalInventoryReserved(string Sku, int Quantity);

import io.cratis.chronicle.IEventStore

data class TransactionalOrderPlaced(val orderId: String, val totalAmount: Double)
data class TransactionalInventoryReserved(val sku: String, val quantity: Int)

suspend fun commitOrder(store: IEventStore) {
    val unitOfWork = store.unitOfWorkManager.begin()

    try {
        store.eventLog.transactional.append(
            "order-123",
            TransactionalOrderPlaced("order-123", 99.95)
        )

        store.eventLog.transactional.append(
            "inventory-widget",
            TransactionalInventoryReserved("widget", 1)
        )

        unitOfWork.commit()
    } catch (exception: Exception) {
        unitOfWork.rollback()
        throw exception
    }
}

import io.cratis.chronicle.IEventStore;
import io.cratis.chronicle.java.BlockingEventStore;

record TransactionalOrderPlaced(String orderId, double totalAmount) {}
record TransactionalInventoryReserved(String sku, int quantity) {}

class TransactionalOrderWorkflow {
    void commitOrder(IEventStore store) {
        var eventStore = new BlockingEventStore(store);

        // Rolls back on the way out unless it was committed, so a throw needs no catch.
        try (var unitOfWork = eventStore.beginUnitOfWork()) {
            eventStore.getTransactional().append(
                "order-123",
                new TransactionalOrderPlaced("order-123", 99.95));

            eventStore.getTransactional().append(
                "inventory-widget",
                new TransactionalInventoryReserved("widget", 1));

            unitOfWork.commit();
        }
    }
}

defmodule MyApp.Events.TransactionalOrderPlaced do
  use Chronicle.Events.EventType, id: "TransactionalOrderPlaced"

  defstruct [:order_id, :total_amount]
end

defmodule MyApp.Events.TransactionalInventoryReserved do
  use Chronicle.Events.EventType, id: "TransactionalInventoryReserved"

  defstruct [:sku, :quantity]
end

defmodule MyApp.TransactionalOrderWorkflow do
  alias Chronicle.Transactions.UnitOfWork
  alias MyApp.Events.{TransactionalInventoryReserved, TransactionalOrderPlaced}

  def commit_order do
    unit_of_work = UnitOfWork.begin()

    try do
      :ok =
        Chronicle.append("order-123", %TransactionalOrderPlaced{
          order_id: "order-123",
          total_amount: 99.95
        })

      :ok =
        Chronicle.append("inventory-widget", %TransactionalInventoryReserved{
          sku: "widget",
          quantity: 1
        })

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

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

@eventType()
class TransactionalOrderPlaced {
    constructor(readonly orderId: string, readonly totalAmount: number) {}
}

@eventType()
class TransactionalInventoryReserved {
    constructor(readonly sku: string, readonly quantity: number) {}
}

async function commitOrder(store: IEventStore): Promise<void> {
    const unitOfWork = store.unitOfWorkManager.begin();

    try {
        await store.eventLog.transactional.append(
            'order-123',
            new TransactionalOrderPlaced('order-123', 99.95)
        );

        await store.eventLog.transactional.append(
            'inventory-widget',
            new TransactionalInventoryReserved('widget', 1)
        );

        await unitOfWork.commit();
    } catch (error) {
        await unitOfWork.rollback();
        throw error;
    }
}

Commit an ordered batch with exact concurrency

When a decision reads several streams, capture the exact revision it used and stage the resulting events with that scope. AddEvents materializes the events and scopes immediately, preserves cross-stream order, and still waits for the unit of work to commit before sending anything to Chronicle.

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

public static class TransactionalTransferWorkflow
{
    public static async Task<bool> TryCommitTransfer(
        IEventStore store,
        EventSequenceNumber expectedAuthorizationRevision)
    {
        using var unitOfWork = store.UnitOfWorkManager.Begin(CorrelationId.New());
        EventSourceId authorizationScopeLabel = "authorization-history";
        var authorizationScope = new ConcurrencyScope(
            expectedAuthorizationRevision,
            EventStreamType: "authorization");

        unitOfWork.AddEvents(
            EventSequenceId.Log,
            [
                new EventForEventSourceId(
                    "account-from",
                    new TransferDebited(100m)),
                new EventForEventSourceId(
                    "account-to",
                    new TransferCredited(100m))
            ],
            [new(authorizationScopeLabel, authorizationScope)]);

        await unitOfWork.Commit();
        return unitOfWork.IsSuccess;
    }
}

[EventType]
public record TransferDebited(decimal Amount);

[EventType]
public record TransferCredited(decimal Amount);

Kotlin does not support this workflow yet.

Java does not support this workflow yet.

Elixir does not support this workflow yet.

TypeScript does not support this workflow yet.

A concurrency-scope key can be an independent label when the scope does not narrow by event-source ID. It does not need to match either event target, so one decision-wide stream-, type-, or event-type-scoped revision can protect the entire ordered batch. If a scope does narrow by event-source ID, its key must be that event-source ID.

Every event target and scope label must contain a non-whitespace value; blank labels cannot survive transport and are rejected during enrollment.

Chronicle resolves the configured concurrency strategy for an event target that has no supplied scope or carries ConcurrencyScope.NotSet. An independent non-target label has no event target from which to resolve a strategy, so Chronicle rejects NotSet and incomplete scopes for those labels. Give an independent label a concrete exact scope or ConcurrencyScope.None.

AddEvents snapshots the event list, scope collection, and every scope’s nested event-type filter before it binds or stages anything in the unit of work. When ordered enrollment first participates in a unit of work, Chronicle also snapshots effective scopes from preceding AddEvent calls; later AddEvent scopes are snapshotted as they are enrolled. Later mutations to caller-owned collections therefore cannot change the eventual commit, and a lazy collection failure is reported during enrollment.

Dispose the unit of work even when enrollment validation fails, and inspect IsSuccess after commit. A completed Commit can still report a constraint or concurrency rejection rather than a successful append.

Consecutive AddEvent calls keep their existing source-grouped order. Each AddEvents call creates an exact-order segment between those legacy segments. Commit flattens the segments in enrollment-call order and sends all of them through one atomic append.

Legacy-only AddEvent calls retain their historical event-sequence behavior. Once AddEvents participates, every event in the unit of work must name the same event sequence; a mismatch is rejected before the ordered batch is staged.

Roll Back

Rollback discards the events staged in the unit of work. Nothing is sent to Chronicle for those staged events.

Transactions and Unit of Work

When to Use a Unit of Work

Commit Transactional Appends

Commit an ordered batch with exact concurrency

Roll Back

Related Topics