Dynamic Consistency Boundary (DCB)

The Dynamic Consistency Boundary (DCB) pattern allows you to query and enforce consistency across events from multiple streams using tags -- strong-typed identifiers attached to events at append time. This is useful when your consistency boundary doesn't align with a single event stream.

Concept

In traditional event sourcing, consistency is enforced per-stream using optimistic concurrency on the stream version. DCB extends this by letting you:

1. Tag events with one or more strong-typed identifiers
2. Query events across streams by those tags
3. Aggregate tagged events into a view (like a live aggregation, but cross-stream)
4. Enforce consistency at save time -- detecting if new matching events were appended since you last read

Registering Tag Types

Tag types are strong-typed identifiers (typically record types wrapping a primitive). Register them during store configuration:

private void ConfigureStore()
{
    StoreOptions(opts =>
    {
        opts.Events.AddEventType<StudentEnrolled>();
        opts.Events.AddEventType<AssignmentSubmitted>();
        opts.Events.AddEventType<StudentDropped>();
        opts.Events.AddEventType<StudentGraded>();

        // Register tag types -- each gets its own table (mt_event_tag_student, mt_event_tag_course)
        opts.Events.RegisterTagType<StudentId>("student")
            .ForAggregate<StudentCourseEnrollment>();
        opts.Events.RegisterTagType<CourseId>("course")
            .ForAggregate<StudentCourseEnrollment>();

        opts.Projections.LiveStreamAggregation<StudentCourseEnrollment>();
    });
}

^{snippet source | anchor}

Each tag type gets its own table (mt_event_tag_student, mt_event_tag_course, etc.) with a composite primary key of (value, seq_id).

Automatic Tag Type Registration

When you register a SingleStreamProjection<TDoc, TId> or MultiStreamProjection<TDoc, TId> that uses a strong-typed identifier as its TId, Marten will automatically register that type as a tag type with ForAggregate() pointing to TDoc. This means you don't need to call RegisterTagType<TId>() explicitly in most cases:

// The projection's TId (TicketId) is auto-registered as a tag type
opts.Projections.Add<TicketSummaryProjection>(ProjectionLifecycle.Inline);

// No need for: opts.Events.RegisterTagType<TicketId>().ForAggregate<TicketSummary>();

Auto-discovery only applies to strong-typed identifiers (e.g., record struct TicketId(Guid Value)). Primitive types like Guid, string, int, long, and short are not auto-registered.

If you explicitly register a tag type before auto-discovery runs, your explicit registration takes precedence. This lets you customize the table suffix when needed:

// Explicit registration with custom table suffix — auto-discovery won't overwrite this
opts.Events.RegisterTagType<TicketId>("custom_ticket")
    .ForAggregate<TicketSummary>();
opts.Projections.Add<TicketSummaryProjection>(ProjectionLifecycle.Inline);

Tag Type Requirements

Tag types should be simple wrapper records around a primitive value:

// Strong-typed tag identifiers
public record StudentId(Guid Value);
public record CourseId(Guid Value);

^{snippet source | anchor}

Supported inner value types: Guid, string, int, long, short.

Tags work with both Rich (default) and Quick append modes. In Rich mode, tags are inserted using pre-assigned sequence numbers. In Quick mode, tags are inserted using a subquery that looks up the sequence from the event's id.

Tagging Events

Use BuildEvent and WithTag to attach tags before appending:

var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
enrolled.WithTag(studentId, courseId);
theSession.Events.Append(streamId, enrolled);
await theSession.SaveChangesAsync();

^{snippet source | anchor}

Events can have multiple tags of different types. Tags are persisted to their respective tag tables in the same transaction as the event.

Querying Events by Tags

Use EventTagQuery to build a query, then execute it with QueryByTagsAsync:

var query = new EventTagQuery().Or<StudentId>(studentId);
var events = await theSession.Events.QueryByTagsAsync(query);

^{snippet source | anchor}

Multiple Tags (OR)

// Query for either student
var query = new EventTagQuery()
    .Or<StudentId>(student1)
    .Or<StudentId>(student2);

var events = await theSession.Events.QueryByTagsAsync(query);

^{snippet source | anchor}

Filtering by Event Type

// Query only AssignmentSubmitted events for this student
var query = new EventTagQuery()
    .Or<AssignmentSubmitted, StudentId>(studentId);

var events = await theSession.Events.QueryByTagsAsync(query);

^{snippet source | anchor}

Events are always returned ordered by sequence number (global append order).

Aggregating by Tags

Build an aggregate from tagged events, similar to AggregateStreamAsync but across streams. First define an aggregate that applies the tagged events:

// Aggregate for DCB
public class StudentCourseEnrollment
{
    public Guid Id { get; set; }
    public string StudentName { get; set; } = "";
    public string CourseName { get; set; } = "";
    public List<string> Assignments { get; set; } = new();
    public bool IsDropped { get; set; }

    public void Apply(StudentEnrolled e)
    {
        StudentName = e.StudentName;
        CourseName = e.CourseName;
    }

    public void Apply(AssignmentSubmitted e)
    {
        Assignments.Add(e.AssignmentName);
    }

    public void Apply(StudentDropped e)
    {
        IsDropped = true;
    }
}

^{snippet source | anchor}

Then aggregate across streams by tag query:

var query = new EventTagQuery()
    .Or<StudentId>(studentId)
    .Or<CourseId>(courseId);

var aggregate = await theSession.Events.AggregateByTagsAsync<StudentCourseEnrollment>(query);

^{snippet source | anchor}

Returns null if no matching events are found.

Fetch for Writing (Consistency Boundary)

FetchForWritingByTags loads the aggregate and establishes a consistency boundary. At SaveChangesAsync time, Marten checks whether any new events matching the query have been appended since the read, throwing DcbConcurrencyException if so:

// Fetch for writing
await using var session2 = theStore.LightweightSession();
var query = new EventTagQuery().Or<StudentId>(studentId);
var boundary = await session2.Events.FetchForWritingByTags<StudentCourseEnrollment>(query);

// Read current state
var aggregate = boundary.Aggregate; // may be null if no events yet
var lastSequence = boundary.LastSeenSequence;

// Append via boundary
var assignment = session2.Events.BuildEvent(new AssignmentSubmitted("HW1", 95));
assignment.WithTag(studentId, courseId);
boundary.AppendOne(assignment);

// Save -- will throw DcbConcurrencyException if another session
// appended matching events after our read
await session2.SaveChangesAsync();

^{snippet source | anchor}

Handling Concurrency Violations

try
{
    await session1.SaveChangesAsync();
}
catch (DcbConcurrencyException ex)
{
    // Reload and retry -- the boundary's tag query had new matching events
    // ex.Query -- the original tag query
    // ex.LastSeenSequence -- the sequence at time of read
}

^{snippet source | anchor}

TIP

The consistency check only detects events that match the same tag query. Events appended to unrelated tags or streams will not cause a violation.

Checking Event Existence

If you only need to know whether any events matching a tag query exist -- without loading or deserializing them -- use EventsExistAsync. This is a lightweight SELECT EXISTS(...) query that avoids the overhead of fetching and materializing event data:

[Fact]
public async Task events_exist_returns_true_when_matching_events_found()
{
    var studentId = new StudentId(Guid.NewGuid());
    var courseId = new CourseId(Guid.NewGuid());
    var streamId = Guid.NewGuid();

    var enrolled = theSession.Events.BuildEvent(new StudentEnrolled("Alice", "Math"));
    enrolled.WithTag(studentId, courseId);
    theSession.Events.Append(streamId, enrolled);
    await theSession.SaveChangesAsync();

    // Check existence -- lightweight, no event loading
    var query = new EventTagQuery().Or<StudentId>(studentId);
    var exists = await theSession.Events.EventsExistAsync(query);
    exists.ShouldBeTrue();
}

^{snippet source | anchor}

This is useful for guard clauses and validation logic in DCB workflows where you need to check preconditions before appending new events.

EventsExistAsync is also available in batch queries via batch.Events.EventsExist(query).

How It Works

Storage

Each registered tag type creates a PostgreSQL table:

CREATE TABLE IF NOT EXISTS mt_event_tag_student (
    value uuid NOT NULL,
    seq_id bigint NOT NULL,
    CONSTRAINT pk_mt_event_tag_student PRIMARY KEY (value, seq_id),
    CONSTRAINT fk_mt_event_tag_student_events
        FOREIGN KEY (seq_id) REFERENCES mt_events(seq_id) ON DELETE CASCADE
);

Consistency Check

At SaveChangesAsync time, Marten executes an EXISTS query checking for new events matching the tag query with seq_id > lastSeenSequence. This runs in the same transaction as the event appends, providing serializable consistency for the tagged boundary.

Tag Routing

Events appended via IEventBoundary.AppendOne() are automatically routed to streams based on their tags. Each tag value becomes the stream identity, so events with the same tag value end up in the same stream.