Fluens.Messaging.Sagas 0.9.2

Fluens.Messaging.Sagas

Process Manager / Saga abstraction layered on top of Fluens.Messaging. State lives in the module's own FluensDbContext, the saga write commits in the same SaveChangesAsync as the inbox ACK and any business outbox row, and the four governing ADRs are 0028 (saga abstraction), 0029 (portable uint Version concurrency), 0030 (timeouts via deferred outbox), and 0031 (operator surfaces + saga metrics).

Installation

dotnet add package Fluens.Messaging.Sagas

Usage

Registration

WithSagas(...) is the single entry point hung off MessagingModuleBuilder; invoke it from inside the AddModuleMessaging<TDbContext> callback so the builder's IServiceCollection + TDbContext are seeded. Every saga tuple is registered via RegisterSaga<TSaga, TData>(); WithSagas reflects over each saga's IStartedBy<T> / IHandleSagaMessage<T> / IHandleSagaTimeout<T> markers and wires one closed-generic SagaDispatcher<TSaga, TData, TMessage> per consumed message type as the module's IMessageConsumer<TMessage> (ADR 20260603000002 Rule 7).

module.AddMessaging<OrdersDbContext>(msg =>
{
    msg.Publishes<OrderCreatedMessage>();
    msg.WithSagas(s =>
    {
        s.RegisterSaga<OrderFulfillmentSaga, OrderFulfillmentData>();
    });
});

WithSagas(...) auto-wires the SagaSaveChangesInterceptor singleton on every module FluensDbContext opting into sagas (the interceptor is registered both as its concrete type and as an ISaveChangesInterceptor enumerable entry so FluensDbContext.OnConfiguring picks it up without you touching OnConfiguring; mirrors AddOutboxInterceptor's idempotency pattern per ADR 20260603000003 Rule 4 — the double-registration guard lives on OnConfiguring, not on the extension method). An empty WithSagas block is a structural no-op (mirrors the AddXxx() convention).

Authoring a saga

A saga is an abstract subclass of Saga<TData> where TData : SagaData, new(). The base SagaData carries the infrastructure columns Guid Id / DateTimeOffset CreatedAt / DateTimeOffset? CompletedAt / [ConcurrencyCheck] uint Version — handler code never reads or writes them.

Lazy-loading proxies. SagaData itself is abstract, so EF Core's ProxyBindingRewriter skips it at model finalization — only your concrete SagaData subclass is affected. If the module's FluensDbContext enables UseLazyLoadingProxies() (see the Fluens.Messaging README), your SagaData subclass MUST NOT be sealed and MUST keep an accessible (non-private) parameterless constructor, or the host crashes at model finalization / first query / row materialization respectively — the same three requirements Fluens.Messaging's own entities satisfy (ADR 20260910000001).

public class OrderFulfillmentData : SagaData
{
    public Guid OrderId { get; set; }
    public string CustomerEmail { get; set; } = string.Empty;
    public bool PaymentCaptured { get; set; }
    public Guid? OutstandingTimeoutToken { get; set; }
}

public sealed class OrderFulfillmentSaga
    : Saga<OrderFulfillmentData>,
      IStartedBy<OrderPlaced>,
      IHandleSagaMessage<PaymentCaptured>,
      IHandleSagaTimeout<PaymentTimeout>
{
    protected override void Configure(ISagaCorrelation<OrderFulfillmentData> c)
    {
        c.Correlate<OrderPlaced>(m => m.OrderId, d => d.OrderId);
        c.Correlate<PaymentCaptured>(m => m.OrderId, d => d.OrderId);
    }

    public async Task Handle(OrderPlaced msg, ISagaContext ctx, CancellationToken ct)
    {
        Data.OrderId = msg.OrderId;
        Data.CustomerEmail = msg.CustomerEmail;
        Data.OutstandingTimeoutToken =
            ctx.RequestTimeout(new PaymentTimeout(msg.OrderId), TimeSpan.FromMinutes(15));
        await ctx.Publisher.Publish(new RequestPayment(msg.OrderId));
    }

    public Task Handle(PaymentCaptured msg, ISagaContext ctx, CancellationToken ct)
    {
        Data.PaymentCaptured = true;
        MarkAsComplete();
        return Task.CompletedTask;
    }

    public Task Handle(PaymentTimeout payload, ISagaContext ctx, CancellationToken ct)
    {
        var firedToken = Guid.Parse(ctx.Headers["fluens-saga-timeout-token"]);
        if (Data.OutstandingTimeoutToken != firedToken)
            return Task.CompletedTask;     // stale firing — silent-ignore

        // compensating action
        return Task.CompletedTask;
    }
}

Correlation

Saga lookup is by business correlation key declared in Configure via ISagaCorrelation<TData>.Correlate(messageKey, dataKey). The dispatcher emits a single SELECT … WHERE <dataKey> = @value and only uses Guid Id lookup when the saga author explicitly correlates on Id. The pure SagaCorrelationMap<TData> planner is unit-testable without a DB.

A non-start message arriving before its start message follows the silent-ignore path (ACK + Information log, no orphan row). The start message arriving twice (idempotent inbox retry) finds the existing row and re-enters Handle, not creating a second row.

SagaPublisherWrapper auto-stamps the fluens-saga-correlation: <sagaType>:<correlationKey> header on every non-timeout outbound message before publish — cross-app traceability is the default branch, there is no [CrossAppSaga] opt-in.

Completion and retention

MarkAsComplete() is soft — sets CompletedAt, never deletes the row. The partial active index WHERE completed_at IS NULL excludes completed rows from the saga's hot-path; it is created by FluensMigration.CreateSagaTable (see Saga table) — the EF model declares no index. Subsequent messages for a completed correlation key route to silent-ignore (ADR 20260603000002 Rule 6).

Retention of completed rows is handled by the per-saga-type SagaRetentionSweeper<TData> plugged into the existing per-module HousekeepingWorker via the generic IRetentionSweeper hook (Fluens.Messaging.Maintenance) — there is no separate saga retention worker. The cutoff is MessagingOptions.SagaRetentionDays (default 7, validated > 0).

Timeouts (ADR 20260603000004)

A saga timeout is a message type implementing the empty ISagaTimeout : IMessage marker handled via IHandleSagaTimeout<TTimeout> where TTimeout : ISagaTimeout. The two markers (IHandleSagaMessage<T> and IHandleSagaTimeout<T>) MUST NOT be mixed on one type.

ctx.RequestTimeout<TTimeout>(payload, delay) returns a fresh Guid token; storing it on TData for the later stale-check is the saga author's responsibility (the framework does not auto-stamp it onto state). SagaPublisherWrapper stamps fluens-saga-timeout-token: <Guid> + fluens-directed-module: <currentModule> (the bare module name — ADR 20260604000001 corrected the prior <currentApp>/<currentModule> shape that mismatched the in-app subscription registry and dropped every timeout) + the DeliverAfter(delay) offset on every ISagaTimeout outbound, so the timeout returns to the originating saga module via the existing partitioned inbox. The directed-module header is load-bearing: ISagaTimeout outbound NEVER crosses the mesh — LocalFanOut.DistributeLocallyAsync honours fluens-directed-module (ADR 20260601000005), and IngressDistributionProcessor honours it identically (ADR 20260530000005).

The framework does NOT pre-compare the timeout token against any saga field — the saga author writes the stale-check (if (Data.Token != ctx.Headers["fluens-saga-timeout-token"]) return;). A stale firing is observable via the saga.timeout.stale counter when the author flags it.

ctx.CancelTimeoutAsync(token) is best-effort + race-aware: DELETE FROM outbox WHERE … AND sent_at IS NULL over the named token; 0 rows affected does NOT throw. A timeout already delivered cannot be cancelled — the saga's stale-token check is the load-bearing invariant.

Sub-second timeout precision is OUT of scope — the cadence is TransportProcessorIntervalSeconds (default 60s) for the outbox-to-inbox leg and InboxProcessorIntervalSeconds (default 30s) for the inbox-to-handler leg.

Concurrency (ADR 20260603000003)

SagaData.Version is the portable optimistic-concurrency token: [ConcurrencyCheck] uint Version incremented by SagaSaveChangesInterceptor on EntityState.Modified entries only (Added keeps Version = 0, the INSERT carries that value). A DbUpdateConcurrencyException raised by the saga commit propagates to the outer LaneWorker retry chain unchanged — the two-stage RetryPolicy (ADR 20260530000004) classifies it as transient and replays on a fresh scope. The saga library never catches the exception, never classifies it as IPermanentConsumeFailure, and never implements its own retry loop. Each replay attempt observable on the saga.concurrency_conflict counter.

Provider-specific tokens (byte[] RowVersion / SQL Server rowversion/timestamp, PostgreSQL xmin) and replacement types (long / int / Guid) are forbidden — portability across SQL Server / PostgreSQL / future providers is the design goal.

Operator surfaces (ADR 20260603000005 Rule 1)

Three SRP-aligned interfaces mirror the ADR 20260531000001 messaging split — each internal sealed, per-module, scoped to the module's own FluensDbContext via IServiceScopeFactory. The "Admin" suffix is forbidden (ADR 20260531000001). The consumer depends on the narrowest interface needed; absence of a registration is a no-op / empty result, not an exception.

• ISagaMaintenance — read / diagnostics, SQL-side: QuerySagasAsync(SagaFilter) → IReadOnlyList<SagaSummary> runs one database query per registered SagaData subclass (found via dbContext.Model.GetEntityTypes(), the typed query closed once per type and cached), honouring every SagaFilter field in the WHERE clause (SagaType selects which types are queried), newest CreatedAt first, and SagaFilter.Limit as a row limit; the per-type results are merged newest-first and cut to Limit. GetSagaAsync(sagaId) → SagaDetail (a WHERE id = @id lookup per saga type until a hit) whose State is a reflection-based projection over EF-mapped properties skipping the four base columns (Id / CreatedAt / CompletedAt / Version); GetActiveSagaCountAsync(sagaType) is a single COUNT(*) WHERE completed_at IS NULL.
• ISagaRecovery — replay / retry, forward-looking: ReplaySagaAsync(sagaId) loads the saga by id, sets CompletedAt = NULL and stages a SagaReplaySignal inbox row in one transaction; an absent saga returns 0 without opening a transaction.
• ISagaControl — destructive override: ForceCompleteSagaAsync(sagaId, reason) sets CompletedAt = now() AND stages a SagaForceCompleteEvent outbox row carrying the operator's reason in one transaction (the event flows through the standard outbox path; consumers that care about saga-completion-audit subscribe normally); CancelTimeoutAsync(token) is best-effort + race-aware.

Resolving the calling module's own surface in a multi-saga-module host. WithSagas(...) registers each of these three interfaces both unkeyed (TryAddScoped — first-registered module wins, kept for single-saga-module back-compat) and keyed by module name (TryAddKeyedScoped). In a host with two or more saga-enabled modules, resolve the keyed registration to reach the calling module's own DbContext — serviceProvider .GetRequiredKeyedService<ISagaMaintenance>(moduleName) — never the unkeyed one, which silently resolves against whichever module registered first.

SagaFilter combines its set fields with AND semantics (an empty filter selects every saga):

var maintenance = sp.GetRequiredKeyedService<ISagaMaintenance>("orders");
var stuck = await maintenance.QuerySagasAsync(
    new SagaFilter { OnlyActive = true, CreatedBefore = DateTimeOffset.UtcNow.AddHours(-1), Limit = 50 },
    cancellationToken);

Saga table

The saga table and its indexes come from Fluens.Migrations: CreateSagaTable(tableName, schema) creates id (primary key), created_at, completed_at and version (typed by the provider conventions' SagaVersionColumnType, bigint on PostgreSQL and SQL Server), the partial ix_<tableName>_active (created_at DESC WHERE completed_at IS NULL), ix_<tableName>_created_at and ix_<tableName>_completed_at, and returns the table syntax so you chain your business columns. These indexes back the SQL-side operator queries above.

public class CreateOrderFulfillmentSaga : FluensMigration
{
    public CreateOrderFulfillmentSaga(IMigrationProviderConventions conventions) : base(conventions) { }

    public override void Up() =>
        CreateSagaTable("order_fulfillment", "orders")
            .WithColumn("order_id").AsGuid().NotNullable()
            .WithColumn("customer_email").AsString(200).NotNullable()
            .WithColumn("payment_captured").AsBoolean().NotNullable()
            .WithColumn("outstanding_timeout_token").AsGuid().Nullable();

    public override void Down() => DeleteSagaTable("order_fulfillment", "orders");
}

The saga base columns are verified at host start. When WithSagas registers a saga, the startup schema verification of Fluens.Messaging (see its README) also checks that each saga table exists and carries the SagaData base columns id, created_at, completed_at and version with the expected store types and nullability; a missing table is reported with CreateSagaTable("<table>", "<schema>") as the remedy. Saga business columns are not checked. Fluens:Messaging:VerifySchemaOnStartup set to false switches the check off.

Diagnostics (ADR 20260603000005 Rule 3)

Saga instruments ride the existing Fluens.Messaging BCL meter — no new static class, no new meter name. Seven new instruments added to MessagingMetrics itself, all tagged with module + saga.type:

• saga.active — absolute ObservableGauge<long> backed by a per-(module, saga.type) ConcurrentDictionary registry (mirroring inbox.lag's pattern, ADR 20260601000001 / ADR 20260603000005 Rule 4 — restart-safe, never a +1/-1 delta). Fed by the per-module hosted SagaActiveObserver<TDbContext> (cadence reuses InboxProcessorIntervalSeconds).
• saga.started, saga.completed, saga.timeout.fired, saga.timeout.stale, saga.concurrency_conflict — counters.
• saga.handle.duration — histogram (ms).

The recording methods on MessagingMetrics (SetSagaActive / RemoveSagaActive / RecordSagaStarted / RecordSagaCompleted / RecordSagaTimeoutFired / RecordSagaTimeoutStale / RecordSagaConcurrencyConflict / RecordSagaHandleDuration) take only primitive types so the static class stays type-agnostic (no saga type is referenced from Fluens.Messaging).

Cross-app caveat

Cross-app traceability is the default branch — there is no [CrossAppSaga] opt-in attribute. The cross-app posture of a given outbound message is decided per-message-type by the existing subscription topology; the saga abstraction stamps the correlation header unconditionally.

ISagaTimeout outbound is the single exception: the wrapper auto-stamps fluens-directed-module so the timeout returns to the originating saga module via the existing partitioned inbox — a timeout NEVER crosses the mesh to a non-originating app, by structural invariant (LocalFanOut + IngressDistributionProcessor honour the header symmetrically; the library does not expose a way to disable this stamping).

Configuration

The saga retention window is bound from the existing "Fluens:Messaging" configuration section:

No new configuration section — the saga library extends the existing MessagingOptions block.

Release notes

0.8.5 (breaking)

• ISagaMaintenance.QuerySagasAsync selects in SQL: every SagaFilter field becomes a WHERE predicate, results come back newest CreatedAt first, and the new SagaFilter.Limit bounds the row count. Callers that relied on the previous unspecified result order must sort themselves; Limit <= 0 throws ArgumentOutOfRangeException.
• GetSagaAsync, GetActiveSagaCountAsync, ReplaySagaAsync and ForceCompleteSagaAsync look the saga up by id in SQL instead of enumerating saga tables in memory.
• FluensMigration.CreateSagaTable / DeleteSagaTable (Fluens.Migrations) create the saga table with the partial active-saga index plus the created-at and completed-at indexes. The provider conventions gain SagaVersionColumnType, so a third-party IMigrationProviderConventions implementation must add it.