Cratis Chronicle reactors — handle events in batches

A batch method handles the matching events from a page of Cratis Chronicle events in one call. Use it when a downstream system accepts bulk work, such as updating a search index with one request instead of a request per event. You can still report failures or delayed retries for individual partitions.

A page is a delivery containing events for one or more partitions, each identified by an event source id. A batch method receives only the event types it observes, in sequence order, and runs once for a page that contains matching events. Per-event methods continue to run once per event, partition by partition.

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.Observation;
using Cratis.Chronicle.Reactors;

[EventType]
public record UserRegistered(string Name);

[EventType]
public record UserRenamed(string NewName);

[EventType]
public record UserEmailChanged(string Email);

public record UserDocument(EventSourceId Id, string Name);

public record BulkUpsertFailure(EventSourceId EventSourceId, Exception Exception);

public record BulkUpsertThrottled(EventSourceId EventSourceId, TimeSpan RetryAfter);

public record BulkUpsertResponse(IEnumerable<BulkUpsertFailure> Failures, IEnumerable<BulkUpsertThrottled> Throttled);

public class UnknownUserEvent(object @event) : Exception($"Unknown user event: {@event}");

public interface ISearchIndex
{
    Task<BulkUpsertResponse> BulkUpsert(IEnumerable<UserDocument> documents, CancellationToken cancellationToken);
}

public interface IEmailConfirmations
{
    Task Send(EventSourceId accountId, string email);
}

public class Users(ISearchIndex index, IEmailConfirmations confirmations) : IReactor
{
    [Observes<UserRegistered>]
    [Observes<UserRenamed>]
    public async Task Index(EventBatch batch, BatchContext context)
    {
        var documents = batch.Select(e => e.Event switch
        {
            UserRegistered registered => new UserDocument(e.Context.EventSourceId, registered.Name),
            UserRenamed renamed => new UserDocument(e.Context.EventSourceId, renamed.NewName),
            _ => throw new UnknownUserEvent(e.Event)
        });

        var response = await index.BulkUpsert(documents, context.CancellationToken);

        foreach (var failure in response.Failures)
        {
            batch.Fail(failure.EventSourceId, failure.Exception);
        }

        foreach (var throttled in response.Throttled)
        {
            batch.Defer(throttled.EventSourceId, throttled.RetryAfter);
        }
    }

    public Task EmailChanged(UserEmailChanged @event, EventContext context) =>
        confirmations.Send(context.EventSourceId, @event.Email);
}

View C# snippet source on GitHub

Kotlin

import io.cratis.chronicle.events.EventContext
import io.cratis.chronicle.events.EventType
import io.cratis.chronicle.observation.BatchContext
import io.cratis.chronicle.observation.EventBatch
import io.cratis.chronicle.observation.Observes
import io.cratis.chronicle.observation.Reactor
import java.time.Duration

@EventType
data class BatchUserRegistered(val name: String)

@EventType
data class BatchUserRenamed(val newName: String)

@EventType
data class BatchUserEmailChanged(val email: String)

data class BatchUserDocument(val id: String, val name: String)

data class BatchUpsertFailure(val eventSourceId: String, val error: Exception)

data class BatchUpsertThrottled(val eventSourceId: String, val retryAfter: Duration)

data class BatchUpsertResponse(val failures: List<BatchUpsertFailure>, val throttled: List<BatchUpsertThrottled>)

interface BatchSearchIndex {
    suspend fun bulkUpsert(documents: List<BatchUserDocument>): BatchUpsertResponse
}

interface BatchEmailConfirmations {
    suspend fun send(accountId: String, email: String)
}

@Reactor
class BatchUsers(private val index: BatchSearchIndex, private val confirmations: BatchEmailConfirmations) {
    @Observes(BatchUserRegistered::class)
    @Observes(BatchUserRenamed::class)
    suspend fun index(batch: EventBatch, context: BatchContext) {
        val documents = batch.map { e ->
            when (val event = e.event) {
                is BatchUserRegistered -> BatchUserDocument(e.context.eventSourceId, event.name)
                is BatchUserRenamed -> BatchUserDocument(e.context.eventSourceId, event.newName)
                else -> error("Unknown user event: $event")
            }
        }

        val response = index.bulkUpsert(documents)
        response.failures.forEach { batch.fail(it.eventSourceId, it.error) }
        response.throttled.forEach { batch.defer(it.eventSourceId, it.retryAfter) }
    }

    suspend fun emailChanged(event: BatchUserEmailChanged, context: EventContext) =
        confirmations.send(context.eventSourceId, event.email)
}

View Kotlin snippet source on GitHub

Java

import io.cratis.chronicle.events.EventContext;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.observation.BatchContext;
import io.cratis.chronicle.observation.EventAndContext;
import io.cratis.chronicle.observation.EventBatch;
import io.cratis.chronicle.observation.Observes;
import io.cratis.chronicle.observation.Reactor;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

@EventType
record BatchUserRegistered(String name) {}

@EventType
record BatchUserRenamed(String newName) {}

@EventType
record BatchUserEmailChanged(String email) {}

record BatchUserDocument(String id, String name) {}

record BatchUpsertFailure(String eventSourceId, Exception error) {}

record BatchUpsertThrottled(String eventSourceId, Duration retryAfter) {}

record BatchUpsertResponse(List<BatchUpsertFailure> failures, List<BatchUpsertThrottled> throttled) {}

interface BatchSearchIndex {
    BatchUpsertResponse bulkUpsert(List<BatchUserDocument> documents);
}

interface BatchEmailConfirmations {
    void send(String accountId, String email);
}

@Reactor
class BatchUsers {
    private final BatchSearchIndex index;
    private final BatchEmailConfirmations confirmations;

    BatchUsers(BatchSearchIndex index, BatchEmailConfirmations confirmations) {
        this.index = index;
        this.confirmations = confirmations;
    }

    @Observes(BatchUserRegistered.class)
    @Observes(BatchUserRenamed.class)
    public void index(EventBatch batch, BatchContext context) {
        List<BatchUserDocument> documents = new ArrayList<>();
        for (EventAndContext e : batch) {
            String id = e.getContext().getEventSourceId();
            if (e.getEvent() instanceof BatchUserRegistered registered) {
                documents.add(new BatchUserDocument(id, registered.name()));
            } else if (e.getEvent() instanceof BatchUserRenamed renamed) {
                documents.add(new BatchUserDocument(id, renamed.newName()));
            }
        }

        BatchUpsertResponse response = index.bulkUpsert(documents);
        response.failures().forEach(failure -> batch.fail(failure.eventSourceId(), failure.error()));
        response.throttled().forEach(throttled -> batch.defer(throttled.eventSourceId(), throttled.retryAfter()));
    }

    public void emailChanged(BatchUserEmailChanged event, EventContext context) {
        confirmations.send(context.getEventSourceId(), event.email());
    }
}

View Java snippet source on GitHub

Elixir

defmodule BatchUserRegistered do
  use Chronicle.Events.EventType, id: "batch-user-registered"

  defstruct name: ""
end

defmodule BatchUserRemoved do
  use Chronicle.Events.EventType, id: "batch-user-removed"

  defstruct name: ""
end

defmodule BatchUserEmailChanged do
  use Chronicle.Events.EventType, id: "batch-user-email-changed"

  defstruct email: ""
end

defmodule BatchUsersReactor do
  use Chronicle.Reactors.Reactor

  alias Chronicle.Observers.EventBatch

  # One call per page, with every BatchUserRegistered and BatchUserRemoved in it.
  @observes BatchUserRegistered
  @observes BatchUserRemoved
  def index(batch, _context) do
    operations =
      Enum.map(batch, fn
        {%BatchUserRegistered{name: name}, context} -> {:upsert, context.event_source_id, name}
        {%BatchUserRemoved{}, context} -> {:delete, context.event_source_id}
      end)

    {failed, throttled} = bulk_index(operations)

    batch = Enum.reduce(failed, batch, &EventBatch.fail(&2, &1, :index_failed))
    Enum.reduce(throttled, batch, &EventBatch.defer(&2, &1, 5_000))
  end

  # A per-event handler for a type no batch function observes.
  @handles BatchUserEmailChanged

  @impl true
  def handle(%BatchUserEmailChanged{}, _context), do: :ok

  # Send the operations to the search index; return the event source ids that failed and the
  # ones the index asked to retry later.
  defp bulk_index(_operations), do: {[], []}
end

View Elixir snippet source on GitHub

TypeScript

// Copyright (c) Cratis. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

import { BatchContext, EventBatch, EventContext, eventType, observes, reactor } from '@cratis/chronicle';
import { field } from '@cratis/fundamentals';

@eventType()
class UserRegistered {
    @field(String) readonly name: string;

    constructor(name: string = '') {
        this.name = name;
    }
}

@eventType()
class UserRenamed {
    @field(String) readonly newName: string;

    constructor(newName: string = '') {
        this.newName = newName;
    }
}

@eventType()
class UserEmailChanged {
    @field(String) readonly email: string;

    constructor(email: string = '') {
        this.email = email;
    }
}

interface UserDocument {
    readonly id: string;
    readonly name: string;
}

interface BulkUpsertResponse {
    readonly failures: readonly { readonly eventSourceId: string; readonly error: Error }[];
    readonly throttled: readonly { readonly eventSourceId: string; readonly retryAfterMilliseconds: number }[];
}

interface ISearchIndex {
    bulkUpsert(documents: readonly UserDocument[], signal: AbortSignal): Promise<BulkUpsertResponse>;
}

interface IEmailConfirmations {
    send(accountId: string, email: string): Promise<void>;
}

@reactor('users')
class Users {
    constructor(private readonly index: ISearchIndex, private readonly confirmations: IEmailConfirmations) {}

    @observes(UserRegistered)
    @observes(UserRenamed)
    async indexUsers(batch: EventBatch, context: BatchContext): Promise<void> {
        const documents = batch.events.map(entry => {
            if (entry.is(UserRegistered)) return { id: entry.context.eventSourceId, name: entry.event.name };
            if (entry.is(UserRenamed)) return { id: entry.context.eventSourceId, name: entry.event.newName };
            throw new Error(`Unknown user event: ${entry.context.eventType.id.value}`);
        });

        const response = await this.index.bulkUpsert(documents, context.signal);

        for (const failure of response.failures) {
            batch.fail(failure.eventSourceId, failure.error);
        }

        for (const throttled of response.throttled) {
            batch.defer(throttled.eventSourceId, throttled.retryAfterMilliseconds);
        }
    }

    async userEmailChanged(event: UserEmailChanged, context: EventContext): Promise<void> {
        await this.confirmations.send(context.eventSourceId, event.email);
    }
}

The methods named after an event type, such as userEmailChanged, stay per-event handlers. A method with @observes is a batch method, and an event type reaches exactly one of them. Events in a batch carry their JSON content; use entry.is(EventType) or batch.ofType(EventType) to narrow them. Construct the reactor with its dependencies through your client artifact activator.

View TypeScript snippet source on GitHub

Here, Index receives UserRegistered and UserRenamed events together, while EmailChanged remains a per-event method. The bulk response identifies partitions that failed or need more time; Fail and Defer let the rest of the page succeed.

Declaring a batch method

A batch method takes an EventBatch as its first parameter and declares each event type it observes with [Observes<TEvent>], once per type. It may also take a BatchContext, a CancellationToken, and services from the service provider. Inside, pattern-match on the events.

Each event type assigned to a batch method must belong to that method alone. Chronicle refuses to register a reactor when:

• An event type is observed by both a batch method and a per-event method, or by two batch methods.
• A batch method has no [Observes<TEvent>], or a method with [Observes<TEvent>] does not take an EventBatch.
• A per-event method takes a BatchContext.
• A batch method returns anything but Task or void.

Returning events from a batch method is not supported. If the batch must produce events, append them through an event log supplied as a dependency.

Order

Each method sees its events in sequence order, and so in order within each partition. Two methods of the same reactor are not ordered relative to each other, even for the same partition. When two side effects must happen in a particular order, handle them in the same method.

The batch context

BatchContext describes the whole page, even when a method’s EventBatch contains only some of its events. It includes the observer, event store, namespace, event sequence, partitions, event count, and first and last sequence numbers. It also supplies a cancellation token and the deadline until which the kernel waits; a null deadline means it waits without a deadline.

Origin explains why Chronicle delivered the page:

Origin	Meaning
Live	The events were appended now and delivered to an observer that is up to date
CatchUp	The events were read from storage for an observer that is behind
Replay	The events were read from storage for an observer that was rewound
Retry	The events were read from storage to recover a failed or deferred partition

Reporting outcomes

Return normally without reporting a problem to mark every partition in the batch as successful. Throwing an exception fails every partition in that batch at its first event in the batch. Those partitions enter the normal failed partition recovery process.

For a partial result, report the affected partition on EventBatch:

Call	Result
Fail(partition, exception)	Fails that partition from its first event in the batch
Fail(eventContext, exception)	Fails that event’s partition from that event; earlier events count as handled
Defer(partition, retryAfter)	Delivers that partition again from its first event in the batch after the delay
Defer(eventContext, retryAfter)	Delivers that event’s partition again from that event after the delay

A deferral does not count as a failure toward quarantine. Use it when a downstream system asks you to back off. A kernel that does not deliver pages cannot defer a partition; the client fails the partition instead and says why.

When several methods report a problem for the same partition, Chronicle retries from the earliest event any of them reported. That can deliver events another method already handled. Keep side effects idempotent, meaning that repeating the same work does not repeat its effect, and use the delivery identity to recognize a redelivery.