---
title: Observing Append Operations
editUrl: https://github.com/Cratis/Chronicle/edit/main/Documentation/events/observing-appends.mdx
description: Observe append operations on an event sequence, and wait for observers to catch up after an append.
---

import { Tabs, TabItem } from '@astrojs/starlight/components';


`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

<Tabs syncKey="chronicle-client">
<TabItem label="C#">

```csharp
using Cratis.Chronicle.EventSequences;

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

[View C# snippet source on GitHub](https://github.com/Cratis/Chronicle/blob/main/Documentation/client-snippets/events/observing-appends/shape.md)

</TabItem>
<TabItem label="Kotlin">

```kotlin
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](https://github.com/Cratis/Chronicle.Kotlin/blob/main/Documentation/client-snippets/events/observing-appends/shape.md)

</TabItem>
<TabItem label="Java">

```java
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](https://github.com/Cratis/Chronicle.Kotlin/blob/main/Documentation/client-snippets-java/events/observing-appends/shape.md)

</TabItem>
<TabItem label="Elixir">

```text
Elixir does not support this workflow yet.
```

[View Elixir snippet source on GitHub](https://github.com/Cratis/Chronicle.Elixir/blob/main/Documentation/client-snippets/events/observing-appends/shape.md)

</TabItem>
<TabItem label="TypeScript">

```typescript
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](https://github.com/Cratis/Chronicle.TypeScript/blob/main/Documentation/client-snippets/events/observing-appends/shape.md)

</TabItem>
</Tabs>

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`:

<Tabs syncKey="chronicle-client">
<TabItem label="C#">

```csharp
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](https://github.com/Cratis/Chronicle/blob/main/Documentation/client-snippets/events/observing-appends/subscribing.md)

</TabItem>
<TabItem label="Kotlin">

```kotlin
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](https://github.com/Cratis/Chronicle.Kotlin/blob/main/Documentation/client-snippets/events/observing-appends/subscribing.md)

</TabItem>
<TabItem label="Java">

```java
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](https://github.com/Cratis/Chronicle.Kotlin/blob/main/Documentation/client-snippets-java/events/observing-appends/subscribing.md)

</TabItem>
<TabItem label="Elixir">

```text
Elixir does not support this workflow yet.
```

[View Elixir snippet source on GitHub](https://github.com/Cratis/Chronicle.Elixir/blob/main/Documentation/client-snippets/events/observing-appends/subscribing.md)

</TabItem>
<TabItem label="TypeScript">

```typescript
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](https://github.com/Cratis/Chronicle.TypeScript/blob/main/Documentation/client-snippets/events/observing-appends/subscribing.md)

</TabItem>
</Tabs>

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

<Tabs syncKey="chronicle-client">
<TabItem label="C#">

```csharp
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](https://github.com/Cratis/Chronicle/blob/main/Documentation/client-snippets/events/observing-appends/wait-for-completion-append.md)

</TabItem>
<TabItem label="Kotlin">

```text
Kotlin does not support this workflow yet.
```

[View Kotlin snippet source on GitHub](https://github.com/Cratis/Chronicle.Kotlin/blob/main/Documentation/client-snippets/events/observing-appends/wait-for-completion-append.md)

</TabItem>
<TabItem label="Java">

```text
Java does not support this workflow yet.
```

[View Java snippet source on GitHub](https://github.com/Cratis/Chronicle.Kotlin/blob/main/Documentation/client-snippets-java/events/observing-appends/wait-for-completion-append.md)

</TabItem>
<TabItem label="Elixir">

```elixir
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](https://github.com/Cratis/Chronicle.Elixir/blob/main/Documentation/client-snippets/events/observing-appends/wait-for-completion-append.md)

</TabItem>
<TabItem label="TypeScript">

```typescript
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](https://github.com/Cratis/Chronicle.TypeScript/blob/main/Documentation/client-snippets/events/observing-appends/wait-for-completion-append.md)

</TabItem>
</Tabs>

### AppendMany

<Tabs syncKey="chronicle-client">
<TabItem label="C#">

```csharp
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](https://github.com/Cratis/Chronicle/blob/main/Documentation/client-snippets/events/observing-appends/wait-for-completion-append-many.md)

</TabItem>
<TabItem label="Kotlin">

```text
Kotlin does not support this workflow yet.
```

[View Kotlin snippet source on GitHub](https://github.com/Cratis/Chronicle.Kotlin/blob/main/Documentation/client-snippets/events/observing-appends/wait-for-completion-append-many.md)

</TabItem>
<TabItem label="Java">

```text
Java does not support this workflow yet.
```

[View Java snippet source on GitHub](https://github.com/Cratis/Chronicle.Kotlin/blob/main/Documentation/client-snippets-java/events/observing-appends/wait-for-completion-append-many.md)

</TabItem>
<TabItem label="Elixir">

```text
Elixir does not support this workflow yet.
```

[View Elixir snippet source on GitHub](https://github.com/Cratis/Chronicle.Elixir/blob/main/Documentation/client-snippets/events/observing-appends/wait-for-completion-append-many.md)

</TabItem>
<TabItem label="TypeScript">

```typescript
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](https://github.com/Cratis/Chronicle.TypeScript/blob/main/Documentation/client-snippets/events/observing-appends/wait-for-completion-append-many.md)

</TabItem>
</Tabs>