Eventual Consistency in Projections

Projections in Chronicle operate under the principle of eventual consistency, meaning that read models are updated asynchronously as events are processed. This design provides significant performance benefits but requires understanding of how and when data becomes consistent.

Understanding Eventual Consistency

When you append an event to Chronicle, the following sequence occurs:

Event is persisted to the event store immediately
Event append operation returns successfully to the caller
Projections are updated asynchronously in the background
Read models become consistent after processing completes

This means there’s a brief window where:

The event has been successfully stored
But projections may not yet reflect the changes

Asynchronous Processing

Chronicle processes events for projections asynchronously to ensure optimal performance and scalability:

Processing Pipeline

Benefits of Asynchronous Processing

High Throughput: Event appends don’t wait for projection updates
Scalability: Projection processing can be scaled independently
Resilience: Failed projection updates don’t affect event persistence
Performance: Read and write operations are optimized separately

Consistency Guarantees

What Chronicle Guarantees

Event Ordering: Events for the same event source are processed in order
At-Least-Once Processing: Every event will be processed (with retries on failure)
Partition Consistency: All projections for a single event source will be eventually consistent

What Chronicle Does NOT Guarantee

Immediate Consistency: Projections may lag behind events
Cross-Partition Ordering: Events from different event sources may be processed out of relative order
Synchronous Updates: Projection updates are always asynchronous

Handling Eventual Consistency

1. Design for Asynchronous Updates

Structure your application to work naturally with eventual consistency:

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.ReadModels;

[EventType]
public record EcBookCreated(string Title, string Author);

public record EcBookInventory(Guid Id, string Title, string Author);

public class EcBookService(IEventLog eventLog, IReadModels readModels)
{
    // Good — fire and forget: don't wait for the projection before returning
    public async Task<EventSourceId> CreateBook(string title, string author)
    {
        var bookId = EventSourceId.New();
        await eventLog.Append(bookId, new EcBookCreated(title, author));
        return bookId;
    }

    // Problematic — expecting immediate consistency
    public async Task<EcBookInventory> CreateBookAndReturn(string title, string author)
    {
        var bookId = EventSourceId.New();
        await eventLog.Append(bookId, new EcBookCreated(title, author));

        // The projection may not have run yet — this can return a stale or default instance
        return await readModels.GetInstanceById<EcBookInventory>(bookId);
    }
}

import io.cratis.chronicle.IEventStore
import io.cratis.chronicle.events.EventType
import java.util.UUID

@EventType
data class EcBookCreated(val title: String, val author: String)

data class EcBookInventory(val id: String = "", val title: String = "", val author: String = "")

class EcBookService(private val store: IEventStore) {
    // Good — fire and forget: don't wait for the projection before returning
    suspend fun createBook(title: String, author: String): String {
        val bookId = UUID.randomUUID().toString()
        store.eventLog.append(bookId, EcBookCreated(title, author))
        return bookId
    }

    // Problematic — expecting immediate consistency
    suspend fun createBookAndReturn(title: String, author: String): EcBookInventory? {
        val bookId = UUID.randomUUID().toString()
        store.eventLog.append(bookId, EcBookCreated(title, author))

        // The projection may not have run yet — this can return null or a stale instance
        return store.readModels.getInstanceByKey(EcBookInventory::class, bookId)
    }
}

import io.cratis.chronicle.IEventStore;
import io.cratis.chronicle.events.EventType;
import kotlin.jvm.JvmClassMappingKt;
import kotlinx.coroutines.BuildersKt;
import kotlin.coroutines.EmptyCoroutineContext;
import kotlin.coroutines.Continuation;
import java.util.UUID;

@EventType
record EcBookCreated(String title, String author) {}

record EcBookInventory(String id, String title, String author) {
    EcBookInventory() {
        this("", "", "");
    }
}

class EcBookService {
    private final IEventStore store;

    EcBookService(IEventStore store) {
        this.store = store;
    }

    // Good — fire and forget: don't wait for the projection before returning
    String createBook(String title, String author) throws InterruptedException {
        var bookId = UUID.randomUUID().toString();
        var eventLog = store.getEventLog();

        BuildersKt.runBlocking(EmptyCoroutineContext.INSTANCE, (scope, continuation) -> {
            @SuppressWarnings("unchecked")
            var appendContinuation = (Continuation<Object>) continuation;
            return eventLog.append(bookId, new EcBookCreated(title, author), null, appendContinuation);
        });

        return bookId;
    }

    // Problematic — expecting immediate consistency
    EcBookInventory createBookAndReturn(String title, String author) throws InterruptedException {
        var bookId = createBook(title, author);

        // The projection may not have run yet — this can return null or a stale instance
        return (EcBookInventory) BuildersKt.runBlocking(
            EmptyCoroutineContext.INSTANCE,
            (scope, continuation) -> {
                @SuppressWarnings("unchecked")
                var readContinuation = (Continuation<? super EcBookInventory>) continuation;
                return store.getReadModels().getInstanceByKey(
                    JvmClassMappingKt.getKotlinClass(EcBookInventory.class),
                    bookId,
                    readContinuation);
            });
    }
}

defmodule MyApp.Events.EcBookCreated do
  use Chronicle.Events.EventType, id: "ec-book-created"

  defstruct [:title, :author]
end

defmodule MyApp.ReadModels.EcBookInventory do
  defstruct id: "", title: "", author: ""
end

defmodule MyApp.EcBookService do
  alias MyApp.Events.EcBookCreated
  alias MyApp.ReadModels.EcBookInventory

  # Good — fire and forget: don't wait for the projection before returning
  def create_book(book_id, title, author) do
    Chronicle.append(book_id, %EcBookCreated{title: title, author: author})
    :ok
  end

  # Problematic — expecting immediate consistency
  def create_book_and_return(book_id, title, author) do
    create_book(book_id, title, author)

    # The projection may not have run yet — this can return nil or a stale instance
    Chronicle.read_model(EcBookInventory, book_id)
  end
end

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

@eventType()
class EcBookCreated {
    constructor(readonly title: string, readonly author: string) {}
}

class EcBookInventory {
    id: string = '';
    title: string = '';
    author: string = '';
}

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

    // Good — fire and forget: don't wait for the projection before returning
    async createBook(title: string, author: string): Promise<string> {
        const bookId = Guid.create().toString();
        await this.store.eventLog.append(bookId, new EcBookCreated(title, author));
        return bookId;
    }

    // Problematic — expecting immediate consistency
    async createBookAndReturn(title: string, author: string): Promise<EcBookInventory> {
        const bookId = Guid.create().toString();
        await this.store.eventLog.append(bookId, new EcBookCreated(title, author));

        // The projection may not have run yet — this can return a stale or default instance
        return this.store.readModels.getInstanceById(EcBookInventory, bookId);
    }
}

2. Watch Projection Changes

Chronicle provides a .Watch<TReadModel>() API that allows you to observe projection changes in real-time:

using System.Reactive.Linq;
using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.ReadModels;

[EventType]
public record EcWatchBookCreated(string Title, string Author);

public record EcWatchBookInventory(Guid Id, string Title, string Author);

public class EcWatchBookService(IEventLog eventLog, IReadModels readModels)
{
    public IObservable<ReadModelChangeset<EcWatchBookInventory>> WatchBookChanges() =>
        readModels.Watch<EcWatchBookInventory>();

    public async Task CreateBookAndWatch(string title, string author)
    {
        var bookId = EventSourceId.New();

        // Subscribe before appending so the update is observed once the projection catches up
        using var subscription = readModels.Watch<EcWatchBookInventory>()
            .Where(changeset => changeset.ModelKey.Value == bookId.Value)
            .Subscribe(changeset => Console.WriteLine($"Book projection updated: {changeset.ReadModel?.Title}"));

        await eventLog.Append(bookId, new EcWatchBookCreated(title, author));
    }
}

Kotlin does not support this workflow yet.
`IReadModelsService` only exposes `getInstanceByKey` — there is no reactive or
streaming API to watch a read model for changes. Track the client SDK issue
before relying on observable read models from Kotlin.

Java does not support this workflow yet.
`IReadModelsService` only exposes `getInstanceByKey` — there is no reactive or
streaming API to watch a read model for changes. Track the client SDK issue
before relying on observable read models from Java.

Elixir does not support this workflow yet.
There is no reactive or streaming API to watch a read model for changes —
only point-in-time reads via `Chronicle.read_model/2`. Track the client SDK
issue before relying on observable read models from Elixir.

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

@eventType()
class EcWatchBookCreated {
    constructor(readonly title: string, readonly author: string) {}
}

class EcWatchBookInventory {
    id: string = '';
    title: string = '';
    author: string = '';
}

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

    watchBookChanges() {
        return this.store.readModels.watch(EcWatchBookInventory);
    }

    async createBookAndWatch(title: string, author: string): Promise<void> {
        const bookId = Guid.create().toString();

        // Start watching before appending so the update is observed once the projection catches up
        const watchBook = async () => {
            for await (const changeset of this.store.readModels.watch(EcWatchBookInventory)) {
                if (changeset.key === bookId) {
                    console.log(`Book projection updated: ${changeset.readModel.title}`);
                    break;
                }
            }
        };
        const watching = watchBook();

        await this.store.eventLog.append(bookId, new EcWatchBookCreated(title, author));
        await watching;
    }
}

3. Monitor Database Changes

For applications that need to respond to database changes from external sources, you can:

Use database change streams: Most modern databases (MongoDB, PostgreSQL, SQL Server) provide change stream APIs
Implement polling mechanisms: Periodically check for changes using timestamps or version fields
Leverage Chronicle’s watch API: Use the .Watch<TReadModel>() method to observe projection updates regardless of their source

4. Use Command-Query Segregation

Separate operations that create data from those that read data:

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.ReadModels;

[EventType]
public record EcCqsBookCreated(string Title);

public record EcCqsBook(Guid Id, string Title);

// Commands — fire and forget, never return projected state
public class EcCqsBookCommandHandler(IEventLog eventLog)
{
    public Task Create(EventSourceId bookId, string title) =>
        eventLog.Append(bookId, new EcCqsBookCreated(title));
}

// Queries — always read from projections
public class EcCqsBookQueryHandler(IReadModels readModels)
{
    public Task<EcCqsBook> GetBook(EventSourceId bookId) =>
        readModels.GetInstanceById<EcCqsBook>(bookId);
}

import io.cratis.chronicle.IEventStore
import io.cratis.chronicle.events.EventType

@EventType
data class EcCqsBookCreated(val title: String)

data class EcCqsBook(val id: String = "", val title: String = "")

// Commands — fire and forget, never return projected state
class EcCqsBookCommandHandler(private val store: IEventStore) {
    suspend fun create(bookId: String, title: String) {
        store.eventLog.append(bookId, EcCqsBookCreated(title))
    }
}

// Queries — always read from projections
class EcCqsBookQueryHandler(private val store: IEventStore) {
    suspend fun getBook(bookId: String): EcCqsBook? =
        store.readModels.getInstanceByKey(EcCqsBook::class, bookId)
}

import io.cratis.chronicle.IEventStore;
import io.cratis.chronicle.events.EventType;
import kotlin.jvm.JvmClassMappingKt;
import kotlinx.coroutines.BuildersKt;
import kotlin.coroutines.EmptyCoroutineContext;
import kotlin.coroutines.Continuation;

@EventType
record EcCqsBookCreated(String title) {}

record EcCqsBook(String id, String title) {
    EcCqsBook() {
        this("", "");
    }
}

// Commands — fire and forget, never return projected state
class EcCqsBookCommandHandler {
    private final IEventStore store;

    EcCqsBookCommandHandler(IEventStore store) {
        this.store = store;
    }

    void create(String bookId, String title) throws InterruptedException {
        var eventLog = store.getEventLog();

        BuildersKt.runBlocking(EmptyCoroutineContext.INSTANCE, (scope, continuation) -> {
            @SuppressWarnings("unchecked")
            var appendContinuation = (Continuation<Object>) continuation;
            return eventLog.append(bookId, new EcCqsBookCreated(title), null, appendContinuation);
        });
    }
}

// Queries — always read from projections
class EcCqsBookQueryHandler {
    private final IEventStore store;

    EcCqsBookQueryHandler(IEventStore store) {
        this.store = store;
    }

    EcCqsBook getBook(String bookId) throws InterruptedException {
        return (EcCqsBook) BuildersKt.runBlocking(
            EmptyCoroutineContext.INSTANCE,
            (scope, continuation) -> {
                @SuppressWarnings("unchecked")
                var readContinuation = (Continuation<? super EcCqsBook>) continuation;
                return store.getReadModels().getInstanceByKey(
                    JvmClassMappingKt.getKotlinClass(EcCqsBook.class),
                    bookId,
                    readContinuation);
            });
    }
}

defmodule MyApp.Events.EcCqsBookCreated do
  use Chronicle.Events.EventType, id: "ec-cqs-book-created"

  defstruct [:title]
end

defmodule MyApp.ReadModels.EcCqsBook do
  defstruct id: "", title: ""
end

# Commands — fire and forget, never return projected state
defmodule MyApp.EcCqsBookCommandHandler do
  alias MyApp.Events.EcCqsBookCreated

  def create(book_id, title) do
    Chronicle.append(book_id, %EcCqsBookCreated{title: title})
  end
end

# Queries — always read from projections
defmodule MyApp.EcCqsBookQueryHandler do
  alias MyApp.ReadModels.EcCqsBook

  def get_book(book_id) do
    Chronicle.read_model(EcCqsBook, book_id)
  end
end

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

@eventType()
class EcCqsBookCreated {
    constructor(readonly title: string) {}
}

class EcCqsBook {
    id: string = '';
    title: string = '';
}

// Commands — fire and forget, never return projected state
class EcCqsBookCommandHandler {
    constructor(private readonly store: IEventStore) {}

    create(bookId: string, title: string): Promise<void> {
        return this.store.eventLog.append(bookId, new EcCqsBookCreated(title)).then(() => undefined);
    }
}

// Queries — always read from projections
class EcCqsBookQueryHandler {
    constructor(private readonly store: IEventStore) {}

    getBook(bookId: string): Promise<EcCqsBook> {
        return this.store.readModels.getInstanceById(EcCqsBook, bookId);
    }
}

Best Practices

1. Design for Eventual Consistency

Accept that reads may be slightly stale
Use optimistic UI updates where possible
Implement retry logic for critical consistency requirements

2. Test Eventual Consistency Scenarios

Test your application behavior during projection lag
Verify retry mechanisms work correctly
Ensure UI handles missing data gracefully

Summary

Eventual consistency in Chronicle projections provides excellent performance and scalability while requiring thoughtful application design. By understanding the asynchronous nature of projection updates and implementing appropriate patterns, you can build robust applications that work naturally with Chronicle’s event-driven architecture.

Remember: eventual consistency is a feature, not a limitation. It enables the high-performance, scalable systems that Chronicle is designed to support.