# Synapse Events, Outbox & Observability > Event publishing, the outbox pattern, and observability. This file contains all documentation content in a single document following the llmstxt.org standard. ## Events Events let a handler notify the rest of the system that something happened, without knowing which components care. Multiple handlers can subscribe to the same event and all of them run. ## Define an event An event is a plain class or record that implements `IEvent`: ```csharp using UnambitiousFx.Synapse.Abstractions; public record TaskCreatedEvent(Guid TaskId, string Title) : IEvent; ``` ## Implement an event handler ```csharp public class SendWelcomeEmailOnTaskCreated : IEventHandler { private readonly IEmailService _email; public SendWelcomeEmailOnTaskCreated(IEmailService email) => _email = email; public async ValueTask HandleAsync( TaskCreatedEvent @event, CancellationToken ct = default) { await _email.SendAsync(@event.Title, ct); return Result.Success(); } } ``` ## Register event handlers ```csharp services.AddSynapse(cfg => { cfg.RegisterEventHandler(); cfg.RegisterEventHandler(); // second handler, same event }); ``` Multiple handlers for the same event type are all registered and all executed on every `EmitAsync` call. ## Publish an event Inject `IEmitter` and call `EmitAsync`: ```csharp await emitter.EmitAsync(new TaskCreatedEvent(task.Id, task.Title), ct); ``` Or publish from within a handler via `IContext`: ```csharp public class CreateTaskHandler : IRequestHandler { private readonly IContext _context; public CreateTaskHandler(IContext context) => _context = context; public async ValueTask> HandleAsync(CreateTaskCommand cmd, CancellationToken ct = default) { var id = Guid.NewGuid(); // ... persist ... await _context.PublishEventAsync(new TaskCreatedEvent(id, cmd.Title), ct); return Result.Success(id); } } ``` ## Emit modes `IEmitter.EmitAsync` accepts an optional `EmitMode` parameter: | Mode | Behaviour | | ------------------ | ------------------------------------------------------------------------------------------------------------- | | `EmitMode.Now` | Dispatches immediately. If any handler fails, the failure propagates back to the caller. | | `EmitMode.Outbox` | Stores the event in the outbox. Dispatched when `CommitEventsAsync` or `IOutboxCommit.CommitAsync` is called. | | `EmitMode.Default` | Resolved to the mode configured via `SetDefaultPublishingMode` (defaults to `Now`). | ```csharp // Dispatch immediately (default) await emitter.EmitAsync(new TaskCreatedEvent(id, title)); // Defer to outbox await emitter.EmitAsync(new TaskCreatedEvent(id, title), EmitMode.Outbox); // Use configured default await emitter.EmitAsync(new TaskCreatedEvent(id, title), EmitMode.Default); ``` See [Outbox Pattern](./outbox) for the full deferred-dispatch story. ## Event dispatch flow ```mermaid flowchart TD Emitter["IEmitter.EmitAsync(event, mode)"] ModeCheck{EmitMode?} Dispatcher["IEventDispatcher\n.DispatchAsync(event)"] Outbox["IEventOutboxStorage\n.AddAsync(event)"] Pipeline["IEventPipelineBehavior[]"] Orchestrator["IEventOrchestrator"] Sequential["SequentialEventOrchestrator\n(handlers run one after another)"] Concurrent["ConcurrentEventOrchestrator\n(handlers run in parallel)"] Handlers["IEventHandler<TEvent> ×N"] Emitter --> ModeCheck ModeCheck -- Now --> Dispatcher ModeCheck -- Outbox --> Outbox Dispatcher --> Pipeline Pipeline --> Orchestrator Orchestrator --> Sequential Orchestrator --> Concurrent Sequential --> Handlers Concurrent --> Handlers ``` ## Fan-out orchestration By default all handlers for an event run **sequentially** in registration order. You can switch to **concurrent** fan-out where all handlers run in parallel: ```csharp services.AddSynapse(cfg => { cfg.SetEventOrchestrator(); // all handlers run in parallel cfg.RegisterEventHandler(); cfg.RegisterEventHandler(); }); ``` :::tip Choosing an orchestrator Use `ConcurrentEventOrchestrator` when handlers are independent and latency matters. Use the default `SequentialEventOrchestrator` when order matters or handlers share mutable state. ::: ## Batch publishing Emit a collection of events in one call: ```csharp var events = new[] { new TaskCreatedEvent(id1, "A"), new TaskCreatedEvent(id2, "B") }; await emitter.EmitBatchAsync(events, ct); ``` Each event is dispatched individually through its own handler chain. ## See also - [Outbox Pattern](./outbox) — defer event dispatch for transactional safety. - [Pipeline Behaviors](./pipelines) — add cross-cutting concerns to event handling. - [Context](./context) — publish events from inside a handler via `IContext.PublishEventAsync`. --- ## Observability Synapse exposes structured logging, distributed tracing, metrics, and health checks out of the box. All hooks integrate with the standard .NET observability stack (Microsoft.Extensions.Logging, System.Diagnostics.ActivitySource, System.Diagnostics.Metrics, and `IHealthCheck`). ## Structured logging with built-in behaviors ### `SimpleLoggingBehavior` / `SimpleLoggingEventBehavior` Logs the name of each request or event and the elapsed time after the handler completes. The behaviors are typed; register a closed pair per handler (`SimpleLoggingBehavior` for void requests): ```csharp services.AddSynapse(cfg => { cfg.RegisterRequestPipelineBehavior< SimpleLoggingBehavior, CreateTaskCommand, Guid>(); cfg.RegisterEventPipelineBehavior< SimpleLoggingEventBehavior, TaskCreatedEvent>(); }); ``` :::tip To apply logging across every handler without naming each pair, make an open-generic behavior and mark it `[PipelineBehavior]` so the source generator emits the closed registrations for you. See [Pipeline Behaviors](./pipelines#applying-a-behavior-to-every-handler). ::: Sample output: ``` info: Synapse[0] Handling CreateTaskCommand info: Synapse[0] CreateTaskCommand completed in 4ms ``` ### `LoggingEnrichmentBehavior` Enriches the `ILogger` scope with `CorrelationId` and all context metadata so every log entry emitted during the request automatically includes them — no manual `BeginScope` calls needed: ```csharp cfg.RegisterRequestPipelineBehavior< LoggingEnrichmentBehavior, CreateTaskCommand, Guid>(); ``` Every `ILogger` call inside the handler or any downstream service will include: ```json { "CorrelationId": "019550a7-0000-7000-0000-000000000000", "OccuredAt": "2026-04-08T10:00:00Z" } ``` ## Distributed tracing Synapse creates an `ActivitySource` named `Unambitious.Synapse`. Any pipeline behavior or handler can start a child span using it: ```csharp using System.Diagnostics; using UnambitiousFx.Synapse.Observability; [PipelineBehavior] public sealed class TracingBehavior : IRequestPipelineBehavior where TRequest : IRequest where TResponse : notnull { public async ValueTask> HandleAsync( TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken = default) { using var activity = SynapseActivitySource.Source.StartActivity(typeof(TRequest).Name); var result = await next(request, cancellationToken); activity?.SetStatus(result.IsSuccess ? ActivityStatusCode.Ok : ActivityStatusCode.Error); return result; } } ``` `[PipelineBehavior]` lets the source generator close this open-generic behavior over every matching handler. See [Pipeline Behaviors](./pipelines#applying-a-behavior-to-every-handler). To export traces to an OpenTelemetry collector, register the activity source when configuring OpenTelemetry: ```csharp builder.Services.AddOpenTelemetry() .WithTracing(tracing => tracing .AddSource("Unambitious.Synapse") .AddOtlpExporter()); ``` ## Metrics `ISynapseMetrics` is backed by `System.Diagnostics.Metrics` and works with any OpenTelemetry-compatible metrics exporter. All instruments are published on the `Unambitious.Synapse` meter. Counters/histograms carry an `event.type` tag (and a `success` tag where applicable). | Instrument | Type | Description | | -------------------------------------- | --------------- | -------------------------------------- | | `mediator.events.dispatched` | Counter | Events dispatched (`event.type`, `success`) | | `mediator.events.dispatch_failures` | Counter | Event dispatch failures (`event.type`) | | `mediator.events.dispatch.duration` | Histogram (ms) | Dispatch duration (`event.type`) | | `mediator.outbox.events.processed` | Counter | Events processed from the outbox (`event.type`, `success`) | | `mediator.outbox.events.dead_lettered` | Counter | Events moved to the dead-letter queue (`event.type`) | | `mediator.outbox.queue_depth` | Gauge | Pending events in the outbox | | `mediator.outbox.processing_lag` | Gauge (seconds) | Age of the oldest pending event | | `mediator.outbox.retrying_count` | Gauge | Events that failed at least once and await retry | | `mediator.outbox.dead_letter_count` | Gauge | Events that exhausted retries | Export to Prometheus + OpenTelemetry: ```csharp builder.Services.AddOpenTelemetry() .WithMetrics(metrics => metrics .AddMeter("Unambitious.Synapse") .AddPrometheusExporter()); ``` Replace the default implementation to customise how the recorded operations are emitted. `ISynapseMetrics` exposes four record methods — `RecordEventDispatched`, `RecordDispatchLatency`, `RecordOutboxEventProcessed`, and `RecordOutboxDeadLettered` (the outbox gauges are observed internally by `SynapseMetrics`): ```csharp builder.Services.AddSingleton(); ``` ## Outbox health check `OutboxHealthCheck` implements `IHealthCheck` and reports the health of the event outbox. It monitors pending event count, dead-lettered event count, and queue lag (the age of the oldest pending event). ### Register ```csharp builder.Services.AddHealthChecks() .AddCheck("outbox", tags: ["ready"]); ``` ### Configure thresholds `OutboxHealthCheck` takes its `OutboxHealthCheckOptions` directly through its constructor (not via `IOptions`), so register a configured instance as a singleton — `AddCheck` then resolves it through the optional constructor parameter: ```csharp builder.Services.AddSingleton(new OutboxHealthCheckOptions { DegradedPendingThreshold = 50, CriticalPendingThreshold = 200, DegradedDeadLetterThreshold = 5, CriticalDeadLetterThreshold = 20, DegradedLagThreshold = TimeSpan.FromMinutes(5), CriticalLagThreshold = TimeSpan.FromMinutes(15), }); ``` | Threshold | Default | Effect when reached | | ----------------------------- | ------- | ------------------------- | | `DegradedPendingThreshold` | 1000 | Health status → Degraded | | `CriticalPendingThreshold` | 5000 | Health status → Unhealthy | | `DegradedDeadLetterThreshold` | 10 | Health status → Degraded | | `CriticalDeadLetterThreshold` | 50 | Health status → Unhealthy | | `DegradedLagThreshold` | 5 min | Health status → Degraded | | `CriticalLagThreshold` | 15 min | Health status → Unhealthy | ### Expose the health endpoint ```csharp app.MapHealthChecks("/health/ready", new HealthCheckOptions { Predicate = hc => hc.Tags.Contains("ready") }); ``` ## See also - [Pipeline Behaviors](./pipelines) — `SimpleLoggingBehavior` and `LoggingEnrichmentBehavior` registration. - [Outbox Pattern](./outbox) — outbox configuration and storage replacement. --- ## Outbox Pattern The outbox pattern decouples event publishing from handler execution. Instead of dispatching events immediately when `EmitAsync` is called, events are stored in an **outbox** and dispatched later in a controlled batch. This is useful when you need to guarantee that events are only dispatched after a database transaction commits successfully. ## How it works ```mermaid stateDiagram-v2 [*] --> Pending : EmitAsync (EmitMode.Outbox) Pending --> Processing : CommitEventsAsync / IOutboxCommit.CommitAsync Processing --> Completed : all event handlers succeed Processing --> Failed : handler fails after max retries Failed --> [*] Completed --> [*] ``` 1. `EmitAsync(event, EmitMode.Outbox)` stores the event in `IEventOutboxStorage`. No dispatch occurs. 2. Calling `context.CommitEventsAsync()` or `IOutboxCommit.CommitAsync()` moves pending events through the event pipeline and handlers. 3. If a handler fails, the outbox retries with exponential backoff up to `MaxRetryAttempts`. 4. After all retry attempts are exhausted, the event is marked `Failed` (dead-lettered). ## Using the outbox ### Store and defer an event ```csharp // Store event in outbox — nothing dispatched yet await context.PublishEventAsync(new TaskCreatedEvent(id, title), EmitMode.Outbox, ct); ``` ### Flush after your side-effects commit ```csharp public class CreateTaskHandler : IRequestHandler { private readonly ITaskRepository _repository; private readonly IContext _context; public CreateTaskHandler(ITaskRepository repository, IContext context) { _repository = repository; _context = context; } public async ValueTask> HandleAsync(CreateTaskCommand cmd, CancellationToken ct = default) { var id = Guid.NewGuid(); await _repository.SaveAsync(id, cmd.Title, ct); // ← side-effect await _context.PublishEventAsync( new TaskCreatedEvent(id, cmd.Title), EmitMode.Outbox, ct); // ← store, don't dispatch yet await _context.CommitEventsAsync(ct); // ← dispatch after save succeeded return Result.Success(id); } } ``` You can also inject `IOutboxCommit` directly and call it outside the handler: ```csharp await outboxCommit.CommitAsync(ct); ``` ## Configure the outbox Set outbox options in `AddSynapse`: ```csharp services.AddSynapse(cfg => { cfg.ConfigureOutbox(opts => { opts.MaxRetryAttempts = 5; opts.InitialRetryDelay = TimeSpan.FromSeconds(1); opts.BackoffFactor = 2.0; // exponential: 1s, 2s, 4s, 8s, 16s opts.BatchSize = 100; // max events per CommitAsync call }); }); ``` | Option | Default | Description | | ------------------- | --------------- | --------------------------------------------------------------- | | `MaxRetryAttempts` | `3` | Number of delivery attempts before marking the event as failed. | | `InitialRetryDelay` | `TimeSpan.Zero` | Delay before the first retry. | | `BackoffFactor` | `2.0` | Multiplier applied to the delay for each subsequent retry. | | `BatchSize` | unlimited | Cap the number of events dispatched per `CommitAsync` call. | ## Configure the default emit mode If most of your events should use the outbox, set it as the default: ```csharp cfg.SetDefaultPublishingMode(EmitMode.Outbox); ``` With this setting, `EmitAsync(event)` and `EmitAsync(event, EmitMode.Default)` store to the outbox automatically. :::warning Production storage The built-in `InMemoryEventOutboxStorage` is **not persistent** — all pending events are lost on restart. Replace it with a database-backed implementation before going to production. ::: ## Replace the storage for production The built-in `InMemoryEventOutboxStorage` is suitable for testing and simple scenarios. For production, replace it with a persistent implementation: ```csharp cfg.SetEventOutboxStorage(); ``` Register `EfCoreEventOutboxStorage` in DI with whatever lifetime your database context uses, then implement `IEventOutboxStorage`: ```csharp public class EfCoreEventOutboxStorage : IEventOutboxStorage { private readonly AppDbContext _db; public EfCoreEventOutboxStorage(AppDbContext db) => _db = db; public async ValueTask AddAsync(TEvent @event, CancellationToken ct = default) where TEvent : class, IEvent { _db.OutboxEvents.Add(OutboxEvent.From(@event)); await _db.SaveChangesAsync(ct); return Result.Success(); } // ... implement remaining members } ``` ## Health check Register the outbox health check to surface queue depth and lag in your `/health` endpoint: ```csharp builder.Services.AddHealthChecks() .AddCheck("outbox", tags: ["ready"]); ``` `OutboxHealthCheck` reads its `OutboxHealthCheckOptions` from its constructor (not `IOptions`), so register a configured instance as a singleton; `AddCheck` resolves it through the optional constructor parameter: ```csharp builder.Services.AddSingleton(new OutboxHealthCheckOptions { DegradedPendingThreshold = 50, // Degraded when pending >= 50 CriticalPendingThreshold = 200, // Unhealthy when pending >= 200 CriticalLagThreshold = TimeSpan.FromMinutes(5), }); ``` See [Observability](./observability#outbox-health-check) for the full threshold list (dead-letter and lag thresholds included). ## See also - [Events](./events) — `EmitMode` explained, how to publish events with `IEmitter`. - [Observability](./observability) — outbox metrics (queue depth, lag, failed count).