Observing Append Operations

IEventSequence exposes an AppendOperations observable that emits after every Append or AppendMany call on that sequence. Any code can subscribe to this observable to react to appended events in real time — without polling the event log.

AppendOperations

using Cratis.Chronicle.EventSequences;

public interface IObservingAppendsEventSequence
{
    IObservable<IEnumerable<AppendedEventWithResult>> AppendOperations { get; }
}

View C# snippet source on GitHub

import io.cratis.chronicle.eventSequences.AppendedEventWithResult
import kotlinx.coroutines.flow.SharedFlow

/**
 * The shape of [io.cratis.chronicle.eventSequences.IEventSequence.appendOperations] - a hot
 * [SharedFlow] that emits after every [io.cratis.chronicle.eventSequences.IEventSequence.append]
 * or [io.cratis.chronicle.eventSequences.IEventSequence.appendMany] call made through that instance.
 */
interface EventSequenceAppendOperationsShape {
    val appendOperations: SharedFlow<List<AppendedEventWithResult>>
}

View Kotlin snippet source on GitHub

import io.cratis.chronicle.eventSequences.AppendedEventWithResult;
import io.cratis.chronicle.eventSequences.IEventLog;

import java.util.List;
import java.util.function.Consumer;

import kotlinx.coroutines.Job;

// The shape of the Java bridge for observing append operations: EventLogJavaBridge.watchAppendOperations
// subscribes a callback that receives a list of AppendedEventWithResult after every append or
// appendMany call made through the event log, returning the Job backing the subscription.
interface EventSequenceAppendOperationsShape {
    Job watchAppendOperations(IEventLog eventLog, Consumer<List<AppendedEventWithResult>> callback);
}

View Java snippet source on GitHub

Elixir does not support this workflow yet.

View Elixir snippet source on GitHub

import { AppendedEventWithResult } from '@cratis/chronicle';

// eventLog.appendOperations is an AsyncIterable<AppendedEventWithResult[]> - a hot,
// multicast stream: iterating it yields every batch of events appended through this
// specific event log instance from the moment you start iterating (never for
// transactional appends, and never replayed for a late subscriber).
type ObservingAppendsShape = AsyncIterable<AppendedEventWithResult[]>;

View TypeScript snippet source on GitHub

The observable emits a collection of AppendedEventWithResult after each operation:

A single-event Append emits a collection containing one element.
A batch AppendMany emits the full batch as one collection.

Each AppendedEventWithResult pairs the appended event with its result:

Member	Type	Description
`Event`	`AppendedEvent`	The appended event, including context and deserialized content
`Event.Content`	`object`	The deserialized event object
`Event.Context`	`EventContext`	Metadata: event source, sequence number, correlation ID, causation chain
`Result`	`AppendResult`	Success flag, sequence number, violations, and errors

Subscribers receive the notification after the operation has completed, whether it succeeded or failed. This observable does not fire for transactional appends through ITransactionalEventSequence.

Subscribing

Inject IEventSequence (or IEventLog) and subscribe to AppendOperations:

using Cratis.Chronicle.EventSequences;

public class ObservingAppendsMonitor(IEventLog eventLog) : IDisposable
{
    readonly IDisposable _subscription = eventLog.AppendOperations.Subscribe(OnAppended);

    static void OnAppended(IEnumerable<AppendedEventWithResult> operations)
    {
        foreach (var item in operations)
        {
            Console.WriteLine($"Event {item.Event.Content.GetType().Name} appended: success={item.Result.IsSuccess}");
        }
    }

    public void Dispose() => _subscription.Dispose();
}

View C# snippet source on GitHub

import io.cratis.chronicle.eventSequences.IEventLog
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.launch

/**
 * Collects append operations for as long as [scope] is active, logging each appended event's
 * outcome.
 */
class AppendMonitor(eventLog: IEventLog, scope: CoroutineScope) {
    init {
        scope.launch {
            eventLog.appendOperations.collect { operations ->
                operations.forEach { println("Event ${it.event::class.simpleName} appended: success=${it.result.isSuccess}") }
            }
        }
    }
}

View Kotlin snippet source on GitHub

import io.cratis.chronicle.eventSequences.IEventLog;

import io.cratis.chronicle.java.EventLogJavaBridge;
import kotlinx.coroutines.Job;

class EventsObservingAppendsSubscribing {
    // Subscribes to append operations for the lifetime of the returned Job; cancel it to stop.
    Job subscribe(IEventLog eventLog) {
        return EventLogJavaBridge.watchAppendOperations(eventLog, operations ->
            operations.forEach(operation ->
                System.out.println("Event " + operation.getEvent().getClass().getSimpleName() +
                    " appended: success=" + operation.getResult().isSuccess())));
    }
}

View Java snippet source on GitHub

Elixir does not support this workflow yet.

View Elixir snippet source on GitHub

import { IEventLog } from '@cratis/chronicle';

async function monitorAppends(eventLog: IEventLog): Promise<void> {
    for await (const operations of eventLog.appendOperations) {
        for (const item of operations) {
            console.log(`Event ${item.event.eventType.id.value} appended: success=${item.result.isSuccess}`);
        }
    }
}

View TypeScript snippet source on GitHub

Always dispose the subscription when you no longer need it to avoid resource leaks.

Integration Testing

The most common use of AppendOperations in application code is through the IEventAppendCollection helper provided by Cratis.Chronicle.XUnit.Integration. It subscribes internally and provides a ready-to-assert collection of AppendedEventWithResult entries. See Chronicle.Testing.EventAppendCollection for full details.

Waiting for Observer Completion After Append

When you need to wait until observers affected by an append operation have completed, use WaitForCompletion() from the Cratis.Chronicle.Observation namespace.

With the current .NET client and server, the wait includes observers that handle the appended event types, plus observers subscribed to all events. For AppendMany, each observer waits only through its last matching event in the batch. An observer whose tag, event source type, or event stream type filters exclude the appended events does not hold up the wait. If an affected observer fails while catching up, the result includes failed partitions. Check the append result before you wait: an append that failed never reached the log, so there is nothing to wait for and the wait reports success.

Only clients sending the first appended sequence number get filter-aware waits. The current .NET client sends it, including when the batch starts at sequence zero. TypeScript, Elixir, Kotlin, Python, and older .NET clients do not; filtered observers can hold up their waits until the timeout. A batch larger than the server’s bounded scan also falls back to waiting through its tail.

The wait gives up after five seconds unless you pass a longer timeout. A timeout returns a result with IsSuccess false and TimedOut true instead of throwing. Newer servers name the observers still behind in OutstandingObservers. TypeScript, Elixir, Kotlin, and Python clients also send no event types and consider every observer on the sequence rather than narrowing by event type. An older server ignores the new fields, so a newer .NET client can still time out on a filtered observer or a mixed-type batch there.

Waiting only means something when there are observers behind the connection. The in-process testing surfaces (EventScenario and EventStoreForTesting) run the event sequence without a silo, so an append there never reaches a projection, reducer, or reactor. Rather than report that everything completed, WaitForCompletion() throws CannotWaitForObserverCompletion on an append result that carries no observer surface — assert on the scenario’s own surface, or move the check to an out-of-process integration spec where observers genuinely run.

Append

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.Observation;

[EventType]
public record ObservingAppendsSomeEvent(string Data);

public class ObservingAppendsCompletionWaiter(IEventLog eventLog)
{
    public async Task AppendAndWait(EventSourceId eventSourceId)
    {
        var appendResult = await eventLog.Append(eventSourceId, new ObservingAppendsSomeEvent("example"));

        // A failed append has nothing to wait for, and waiting on it reports success.
        if (!appendResult.IsSuccess)
        {
            return;
        }

        var completion = await appendResult.WaitForCompletion();

        if (!completion.IsSuccess)
        {
            foreach (var failedPartition in completion.FailedPartitions)
            {
                Console.WriteLine($"Observer {failedPartition.ObserverId} failed partition {failedPartition.Partition}");
            }
        }
    }
}

View C# snippet source on GitHub

Kotlin does not support this workflow yet.

View Kotlin snippet source on GitHub

Java does not support this workflow yet.

View Java snippet source on GitHub

defmodule MyApp.Events.ObservingAppendsSomeEvent do
  use Chronicle.Events.EventType, id: "observing-appends-some-event"

  defstruct [:data]
end

defmodule MyApp.ObservingAppendsCompletionWaiter do
  alias Chronicle.EventSequences.EventLog
  alias MyApp.Events.ObservingAppendsSomeEvent

  def append_and_wait(event_source_id) do
    case EventLog.append_and_wait_for_completion(event_source_id, %ObservingAppendsSomeEvent{
           data: "example"
         }) do
      {:ok, %{success: true}} ->
        :ok

      {:ok, %{success: false, failed_partitions: failed_partitions}} ->
        Enum.each(failed_partitions, fn failed_partition ->
          IO.puts("Observer failed partition: #{inspect(failed_partition)}")
        end)

      {:error, reason} ->
        {:error, reason}
    end
  end
end

View Elixir snippet source on GitHub

import { eventType, IEventLog } from '@cratis/chronicle';
import { field } from '@cratis/fundamentals';

@eventType()
class ObservingAppendsSomeEvent {
    @field(String) readonly data: string;

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

async function appendAndWait(eventLog: IEventLog, eventSourceId: string): Promise<void> {
    const appendResult = await eventLog.append(eventSourceId, new ObservingAppendsSomeEvent('example'));
    const completion = await appendResult.waitForCompletion();

    if (!completion.isSuccess) {
        for (const failedPartition of completion.failedPartitions) {
            console.log(`Observer ${failedPartition.observerId} failed partition ${failedPartition.partition}`);
        }
    }
}

View TypeScript snippet source on GitHub

AppendMany

using Cratis.Chronicle.Events;
using Cratis.Chronicle.EventSequences;
using Cratis.Chronicle.Observation;

[EventType]
public record ObservingAppendsFirstEvent(string Data);

[EventType]
public record ObservingAppendsSecondEvent(string Data);

public class ObservingAppendsBatchCompletionWaiter(IEventLog eventLog)
{
    public async Task AppendManyAndWait(EventSourceId eventSourceId)
    {
        var appendManyResult = await eventLog.AppendMany(eventSourceId, new object[]
        {
            new ObservingAppendsFirstEvent("first"),
            new ObservingAppendsSecondEvent("second")
        });

        // A failed append has nothing to wait for, and waiting on it reports success.
        if (!appendManyResult.IsSuccess)
        {
            return;
        }

        var completion = await appendManyResult.WaitForCompletion();
        if (!completion.IsSuccess)
        {
            // Inspect failed partitions from affected observers
        }
    }
}

View C# snippet source on GitHub

Kotlin does not support this workflow yet.

View Kotlin snippet source on GitHub

Java does not support this workflow yet.

View Java snippet source on GitHub

defmodule MyApp.Events.ObservingAppendsFirstEvent do
  use Chronicle.Events.EventType, id: "observing-appends-first-event"

  defstruct [:data]
end

defmodule MyApp.Events.ObservingAppendsSecondEvent do
  use Chronicle.Events.EventType, id: "observing-appends-second-event"

  defstruct [:data]
end

defmodule MyApp.ObservingAppendsBatchCompletionWaiter do
  alias Chronicle.EventSequences.{AppendResult, EventLog}
  alias MyApp.Events.{ObservingAppendsFirstEvent, ObservingAppendsSecondEvent}

  def append_many_and_wait(event_source_id) do
    events = [
      %ObservingAppendsFirstEvent{data: "first"},
      %ObservingAppendsSecondEvent{data: "second"}
    ]

    case EventLog.append_many_with_result(event_source_id, events) do
      {:ok, %AppendResult{} = append_many_result} ->
        case AppendResult.wait_for_completion(append_many_result, timeout: 10_000) do
          {:ok, %{success: true}} ->
            :ok

          {:ok, %{success: false, failed_partitions: failed_partitions}} ->
            # Inspect failed partitions from affected observers
            {:error, {:failed_partitions, failed_partitions}}

          {:error, reason} ->
            {:error, reason}
        end

      # A failed append has nothing to wait for.
      {:error, %AppendResult{} = rejected} ->
        {:error, rejected}

      {:error, reason} ->
        {:error, reason}
    end
  end
end

View Elixir snippet source on GitHub

import { eventType, IEventLog } from '@cratis/chronicle';
import { field } from '@cratis/fundamentals';

@eventType()
class ObservingAppendsFirstEvent {
    @field(String) readonly data: string;

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

@eventType()
class ObservingAppendsSecondEvent {
    @field(String) readonly data: string;

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

async function appendManyAndWait(eventLog: IEventLog, eventSourceId: string): Promise<void> {
    const appendManyResults = await eventLog.appendMany(eventSourceId, [
        new ObservingAppendsFirstEvent('first'),
        new ObservingAppendsSecondEvent('second')
    ]);

    for (const appendResult of appendManyResults) {
        const completion = await appendResult.waitForCompletion();
        if (!completion.isSuccess) {
            // Inspect completion.failedPartitions from affected observers.
        }
    }
}

View TypeScript snippet source on GitHub