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.

C#

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

Kotlin

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

Java

import io.cratis.chronicle.IEventStore;
import io.cratis.chronicle.eventSequences.AppendResult;
import kotlin.Unit;
import kotlin.coroutines.Continuation;
import kotlin.coroutines.EmptyCoroutineContext;
import kotlinx.coroutines.BuildersKt;

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

class TransactionalOrderWorkflow {
    void commitOrder(IEventStore store) throws InterruptedException {
        var unitOfWork = store.getUnitOfWorkManager().begin();

        try {
            BuildersKt.runBlocking(
                EmptyCoroutineContext.INSTANCE,
                (scope, continuation) -> {
                    @SuppressWarnings("unchecked")
                    var appendContinuation = (Continuation<? super AppendResult>) continuation;
                    return store.getEventLog().getTransactional().append(
                        "order-123",
                        new TransactionalOrderPlaced("order-123", 99.95),
                        null,
                        appendContinuation);
                });

            BuildersKt.runBlocking(
                EmptyCoroutineContext.INSTANCE,
                (scope, continuation) -> {
                    @SuppressWarnings("unchecked")
                    var appendContinuation = (Continuation<? super AppendResult>) continuation;
                    return store.getEventLog().getTransactional().append(
                        "inventory-widget",
                        new TransactionalInventoryReserved("widget", 1),
                        null,
                        appendContinuation);
                });

            BuildersKt.runBlocking(
                EmptyCoroutineContext.INSTANCE,
                (scope, continuation) -> {
                    @SuppressWarnings("unchecked")
                    var commitContinuation = (Continuation<? super Unit>) continuation;
                    return unitOfWork.commit(commitContinuation);
                });
        } catch (RuntimeException exception) {
            BuildersKt.runBlocking(
                EmptyCoroutineContext.INSTANCE,
                (scope, continuation) -> {
                    @SuppressWarnings("unchecked")
                    var rollbackContinuation = (Continuation<? super Unit>) continuation;
                    return unitOfWork.rollback(rollbackContinuation);
                });
            throw exception;
        }
    }
}

Elixir

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

TypeScript

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

Roll Back

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

Related Topics

• Appending events
• Appending many events
• Optimistic concurrency
• Constraints