Closing streams

Some facts end a story. An invoice is issued, a month is signed off, a customer is archived. After that, nothing should be appended to it, and a rule in your command code is not enough: two requests can race, and a bug in one client can still append.

Closing a stream moves that rule into the kernel. Once a scope is closed, every append that falls inside it is rejected with a constraint violation of type StreamClosed, no matter which client sends it.

What closing means

A closure is a record in the kernel’s constraint storage. It is not an event, and it does not change the events already in the log. Every append is checked against the closures of its event sequence and namespace before it is committed.

There are two ways to create a closure:

Way	Written by	Reopened by
Manual	Your code calls CompleteStream	Nothing in the normal flow. Only the admin repair path
Event-driven	The kernel, when it appends an event type declared with [ClosesStream]	An event type you name in ReopenedBy

Pick the event-driven form when a domain fact ends the scope. The closing fact and the closure are written in the same step, and the closure shows up in the log as the fact that caused it. Pick the manual form for ad-hoc closes that no domain event represents.

Choose a scope

A scope is any non-empty combination of four dimensions. A dimension you leave unset is not constrained:

Dimension	Matches
Event source id	One event source, for example one invoice
Event source type	Every event source of one type
Event stream type	A kind of stream, for example monthly-report
Event stream id	A stream within a type, for example 2026-04

An append is refused when every dimension set in a closure equals the append’s value. Closing the stream monthly-report / 2026-04 of event source report-7 refuses only appends that carry all three values. Another report that uses the same stream type and id is unaffected.

Some examples:

Scope	Refuses
Event source report-7, stream type monthly-report, stream id 2026-04	That one report’s April stream
Event source report-7	Every stream of report-7, including its default stream
Stream type monthly-report, stream id 2026-04	The April stream of every event source
Event source type Report	Every event source of that type

EventSourceId.Unspecified, EventSourceType.Unspecified and an empty stream type or stream id count as unset. EventStreamType.All and EventStreamId.Default are real values: the scope event source report-7 + All + Default closes exactly the default stream of report-7.

The default stream cannot be closed on its own

A scope that names no event source and whose stream values are only All and Default would block the whole event sequence. CompleteStream returns DefaultStreamCannotBeCompleted for it. The same applies when the event source type is unset or Default. Add an event source id, or any other specific value, to close a default stream.

Scoped closing shows how to close each combination from code.

Close a stream

The two-argument call closes one stream type and stream id pair across every event source:

C#

using Cratis.Chronicle.EventSequences;

public class ClosingStreamsInvoiceCloser(IEventLog eventLog)
{
    public async Task CloseInvoiceStream(EventStreamId invoiceStreamId)
    {
        var result = await eventLog.CompleteStream(new EventStreamType("invoices"), invoiceStreamId);

        result.Switch(
            sequenceNumber => Console.WriteLine($"Stream closed at sequence number {sequenceNumber}"),
            error => Console.WriteLine($"Failed to close stream: {error}"));
    }
}

View C# snippet source on GitHub

Kotlin

import io.cratis.chronicle.IEventStore
import io.cratis.chronicle.eventSequences.CompleteStreamResult

/**
 * Closes the invoice stream so no further line items can be appended to it once the invoice
 * has been issued.
 */
suspend fun closeInvoiceStream(store: IEventStore, invoiceStreamId: String) {
    when (val result = store.eventLog.completeStream("invoices", invoiceStreamId)) {
        is CompleteStreamResult.Success -> println("Stream closed at sequence ${result.sequenceNumber.value}")
        CompleteStreamResult.AlreadyCompleted -> println("Failed to close stream: already completed")
        CompleteStreamResult.DefaultStreamCannotBeCompleted -> println("Failed to close stream: the default stream cannot be completed")
    }
}

View Kotlin snippet source on GitHub

Java

import io.cratis.chronicle.EventStore;

import io.cratis.chronicle.java.EventLogJavaBridge;

class ClosingStreamsIndexCompleteStream {
    // Closes the invoice stream so no further line items can be appended to it. Returns false
    // when the stream was already completed, or when it is the default stream (which can never
    // be completed).
    boolean closeInvoiceStream(EventStore store, String invoiceStreamId) {
        return EventLogJavaBridge.completeStream(store.getEventLog(), "invoices", invoiceStreamId);
    }
}

View Java snippet source on GitHub

Elixir

defmodule MyApp.ClosingStreamsInvoiceCloser do
  alias Chronicle.EventSequences.EventLog

  def close_invoice_stream(invoice_stream_id) do
    case EventLog.complete_stream("invoices", invoice_stream_id) do
      {:ok, sequence_number} ->
        IO.puts("Stream closed at sequence number #{sequence_number}")

      {:error, reason} ->
        IO.puts("Failed to close stream: #{inspect(reason)}")
    end
  end
end

View Elixir snippet source on GitHub

TypeScript

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

async function closeInvoiceStream(eventLog: IEventLog, invoiceStreamId: string): Promise<void> {
    const result = await eventLog.completeStream('invoices', invoiceStreamId);

    if (result.isSuccess) {
        console.log(`Stream closed at sequence number ${result.sequenceNumber.value}`);
    } else {
        console.log(`Failed to close stream: ${result.error}`);
    }
}

View TypeScript snippet source on GitHub

An empty stream type means All and an empty stream id means Default, so ("", "") is the default stream and returns DefaultStreamCannotBeCompleted. The call returns Result<EventSequenceNumber, CompleteStreamError>. On success the number is the tail of the event sequence at the moment of closing.

To close anything narrower or broader than “one pair for every event source”, pass a ClosedStreamScope to CompleteStream(scope, expectedTailSequenceNumber).

The optional expectedTailSequenceNumber makes the close race-free. The kernel compares it with the sequence number of the last event inside the scope and closes in the same step, so you close only if nothing landed after the last fact you know about. Scoped closing shows the pattern.

Errors

Error	Meaning
AlreadyCompleted	A manual closure already covers the scope. Nothing was written
DefaultStreamCannotBeCompleted	The scope would close the default stream on its own, see above
EmptyScope	No dimension is set
ExpectedTailMismatch	The scope’s tail differs from expectedTailSequenceNumber. Nothing was written

What “already completed” means

A scope is covered when an existing closure’s dimensions are a subset of the scope’s dimensions and their values match. A narrower scope than an existing manual closure is therefore AlreadyCompleted. A broader one is not covered and creates a new closure.

Coverage by an event-driven closure alone does not return AlreadyCompleted. The call succeeds and writes a manual closure next to it. Manual closures are final, so a later ReopenedBy event cannot undo what an operator decided explicitly.

Retrying a close is safe: a second call returns AlreadyCompleted before the expected-tail check runs.

After a stream is closed

An append inside a closed scope is rejected with a StreamClosed violation. The violation details name the owner (ClosedBy, empty for a manual closure) and the sequence number at which the scope closed (ClosedAtSequenceNumber):

C#

using Cratis.Chronicle.Events;
using Cratis.Chronicle.Events.Constraints;
using Cratis.Chronicle.EventSequences;
using System.Linq;

[EventType]
public record ClosingStreamsInvoiceLineAdded(string Description, decimal Amount);

public class ClosingStreamsInvoiceLineAppender(IEventLog eventLog)
{
    public async Task<bool> TryAppendLine(EventSourceId invoiceId)
    {
        var appendResult = await eventLog.Append(
            invoiceId,
            new ClosingStreamsInvoiceLineAdded("Consulting", 500m),
            new EventStreamType("invoices"),
            new EventStreamId("invoice-42"));

        if (!appendResult.IsSuccess)
        {
            var violation = appendResult.ConstraintViolations
                .FirstOrDefault(v => v.ConstraintType == ConstraintType.StreamClosed);
            return violation is null;
        }

        return true;
    }
}

View C# snippet source on GitHub

Kotlin

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

@EventType
data class ClosingStreamsInvoiceLineAdded(val description: String = "", val amount: Double = 0.0)

/**
 * Appends a line item to an invoice stream. Once the stream has been closed, the append is
 * rejected with a "closed-stream" constraint violation and no further lines can be added.
 */
suspend fun tryAppendLine(store: IEventStore, invoiceId: String, eventStreamId: String): Boolean {
    val result = store.eventLog.append(
        invoiceId,
        ClosingStreamsInvoiceLineAdded("Consulting", 500.0),
        AppendOptions(eventStreamType = "invoices", eventStreamId = eventStreamId)
    )

    if (!result.isSuccess) {
        if (result.constraintViolations.any { it.constraintId == "closed-stream" }) return false
        throw IllegalStateException("Append failed: $result")
    }

    return true
}

View Kotlin snippet source on GitHub

Java

import io.cratis.chronicle.EventStore;
import io.cratis.chronicle.events.EventType;
import io.cratis.chronicle.eventSequences.AppendOptions;
import io.cratis.chronicle.eventSequences.AppendResult;
import io.cratis.chronicle.eventSequences.ConstraintViolation;

import io.cratis.chronicle.java.AppendOptionsBuilder;
import io.cratis.chronicle.java.EventLogJavaBridge;

@EventType
record ClosingStreamsInvoiceLineAdded(String description, double amount) {}

class ClosingStreamsIndexAppendRejected {
    // Appends a line item to an invoice stream. Once the stream has been closed, the append is
    // rejected with a "closed-stream" constraint violation and no further lines can be added.
    boolean tryAppendLine(EventStore store, String invoiceId, String eventStreamId) {
        AppendOptions options = new AppendOptionsBuilder()
            .eventStreamType("invoices")
            .eventStreamId(eventStreamId)
            .build();

        AppendResult result = EventLogJavaBridge.append(
            store.getEventLog(), invoiceId, new ClosingStreamsInvoiceLineAdded("Consulting", 500.0), options);

        if (!result.isSuccess()) {
            for (ConstraintViolation violation : result.getConstraintViolations()) {
                if (violation.getConstraintId().equals("closed-stream")) {
                    return false;
                }
            }
            throw new IllegalStateException("Append failed: " + result);
        }

        return true;
    }
}

View Java snippet source on GitHub

Elixir

defmodule MyApp.Events.ClosingStreamsInvoiceLineAdded do
  use Chronicle.Events.EventType, id: "closing-streams-invoice-line-added"

  defstruct [:description, :amount]
end

defmodule MyApp.ClosingStreamsInvoiceLineAppender do
  alias Chronicle.EventSequences.EventLog
  alias MyApp.Events.ClosingStreamsInvoiceLineAdded

  def try_append_line(invoice_id) do
    case EventLog.append(
           invoice_id,
           %ClosingStreamsInvoiceLineAdded{description: "Consulting", amount: 500},
           event_stream_type: "invoices",
           event_stream_id: "invoice-42"
         ) do
      :ok ->
        true

      {:error, {:constraint_violations, violations}} ->
        not Enum.any?(violations, &stream_closed?/1)

      {:error, _reason} ->
        true
    end
  end

  # Mirrors the wire Constraint's Type field (seen as :Unique / :UniqueEventType
  # when registering constraints) — a rejection caused by a closed stream comes
  # back as a violation whose type is :StreamClosed.
  defp stream_closed?(violation) when is_map(violation) do
    Map.get(violation, :Type) == :StreamClosed or Map.get(violation, :type) == :stream_closed
  end

  defp stream_closed?(_violation), do: false
end

View Elixir snippet source on GitHub

TypeScript

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

@eventType()
class ClosingStreamsInvoiceLineAdded {
    @field(String) readonly description: string;
    @field(Number) readonly amount: number;

    constructor(description: string = '', amount: number = 0) {
        this.description = description;
        this.amount = amount;
    }
}

async function tryAppendLine(eventLog: IEventLog, invoiceId: string): Promise<boolean> {
    const [appendResult] = await eventLog.appendMany([{
        eventSourceId: invoiceId,
        event: new ClosingStreamsInvoiceLineAdded('Consulting', 500),
        eventStreamType: 'invoices',
        eventStreamId: 'invoice-42'
    }]);

    if (!appendResult.isSuccess) {
        const wasStreamClosed = appendResult.constraintViolations.some(violation => violation.constraintId === 'StreamClosed');
        return !wasStreamClosed;
    }

    return true;
}

View TypeScript snippet source on GitHub

The check is active for every event sequence without configuration.

Closing does not touch what is already stored or how it is read:

• Reads, observers, projections, reactors and replays work as before. A closed stream replays like any other.
• Events already in the stream stay in the log.
• A redacted closing event keeps its closure.

Closing only refuses new appends. The closing event itself is accepted.

Guarantees and limits

• No interleaving. The kernel writes an event-driven closure in the same turn as the durable append. No other append to that event sequence can land between the closing event and its closure. Unique constraints rely on the same mechanism.
• Not crash-atomic. If the kernel stops after it wrote the event and before it wrote the closure, the scope stays open until the closure is rebuilt.
• Reindex rebuilds event-driven closures. A constraint reindex that names a closing constraint rebuilds its closures from the events in the log. Manual closures are never touched. Registering or changing a declaration does not start a reindex and closes nothing in history. Removing a declaration leaves its closures in force.
• Redacted property-sourced events. When a closing or reopening event takes its stream id from a property and that event has been redacted, the rebuild cannot recover the scope. It leaves that constraint’s closures untouched, logs a warning with the constraint and event position, and continues with the other constraints. Closures whose dimensions come from the event context are unaffected.
• Event type generations. A property-sourced scope is read from the generation the event was appended with. If a new generation renames the property, update the declaration, otherwise closing and reopening fail for events of that generation.
• Kernel version. Scoped closing and closing events use new kernel calls. A client that uses them needs a kernel that has them. An older kernel could otherwise ignore the extra dimensions and close too broadly.

Next steps

• Scoped closing: close an event source, a stream or both.
• Closing events: let a domain fact close and reopen a scope.
• Inspecting and repairing: list closures and undo a mistaken manual close.
• Close the books: end a period and continue in the next one.
• Close a stream of a definition: take the scope from an event source definition.
• Dynamic Consistency Boundary in Chronicle: a closed stream makes a decision about it final.