# Events Store (Durable Publish) (/integrations/masstransit/how-to/events-store) ## Overview [#overview] By default, MassTransit `Publish()` maps to KubeMQ **Events** — fire-and-forget fan-out where every active subscriber receives the message and nothing is persisted. If no subscriber is connected when an event is published, the message is silently dropped. **EventsStore** is the durable variant. When enabled, `Publish()` is routed to KubeMQ **EventsStore** instead: each published message is stored on the channel and can be replayed. Subscribers choose where they begin reading via a configurable start position, so a late-joining subscriber can catch up on everything it missed — ideal for rebuilding read models, onboarding new services, or recovering after an outage. → For what the Events Store is and KubeMQ's persistence and replay model, see [Events Store](/learn/events-store). This page documents the MassTransit transport's API surface only. EventsStore keeps the same fan-out semantics as Events — all subscribers receive each message — but adds persistence and replay. For volatile, in-the-moment fan-out where history is not needed, use [Events (Publish)](/integrations/masstransit/how-to/events) instead. The following diagram shows how a durable publish flows through EventsStore. The message is stored on the channel, fanned out to currently-connected subscribers, and remains available for a subscriber that connects later and replays from an earlier position. *A durable publish is stored and sequenced in EventsStore; a late-joining subscriber replays the history from `StartFromFirst`, then continues with new messages.* ## Enabling EventsStore [#enabling-eventsstore] EventsStore can be turned on for the whole bus or for individual endpoints. ### Global (All Publish Endpoints) [#global-all-publish-endpoints] Calling `cfg.UseEventsStore()` inside the `UsingKubeMQ` callback makes **every** publish endpoint use EventsStore. All `Publish()` calls become durable. ```csharp title="Program.cs" services.AddMassTransit(x => { x.AddConsumer(); x.UsingKubeMQ((ctx, cfg) => { cfg.Host("localhost", 50000); cfg.UseEventsStore(); // All Publish() calls use EventsStore // EventsStore is enabled globally; this endpoint inherits it. cfg.ReceiveEndpoint("audit-events", e => { e.ConfigureKubeMQ(t => { }); }); }); }); ``` `UseEventsStore()` on the bus factory configurator is the single global switch — there are no extra parameters. ### Per-Endpoint [#per-endpoint] To make only certain channels durable, call `UseEventsStore()` on the receive endpoint configurator. Other endpoints continue to use volatile Events. ```csharp title="Program.cs" services.AddMassTransit(x => { x.UsingKubeMQ((ctx, cfg) => { cfg.Host("localhost", 50000); cfg.ReceiveEndpoint("audit-events", e => { e.UseEventsStore(); // Only this endpoint uses EventsStore }); }); }); ``` ### Via Endpoint Transport Configurator [#via-endpoint-transport-configurator] The same per-endpoint switch is available through the KubeMQ-specific transport configurator, which is useful when you are already inside a `ConfigureKubeMQ` block setting other transport options. ```csharp title="Program.cs" cfg.ReceiveEndpoint("audit-events", e => { e.ConfigureKubeMQ(k => { k.UseEventsStore(); // Enable via the transport-specific configurator }); }); ``` **Do not combine `UseEventsStore()` and `UseVolatileEvents()` on the same endpoint.** This is not supported and throws `KubeMQTransportConfigurationException("UseVolatileEvents and UseEventsStore cannot be combined on the same endpoint.")` at bus startup. When EventsStore is enabled globally, every endpoint already has it on — calling `e.UseVolatileEvents()` to "opt back out" triggers exactly this error. To mix volatile and durable endpoints, **do not enable EventsStore globally**: leave it off so endpoints stay volatile by default, and call `e.UseEventsStore()` only on the endpoints that need persistence. ## Subscription Start Positions [#subscription-start-positions] When a subscriber attaches to an EventsStore channel, you control where it begins reading. Configure the position with `e.UseEventsStoreSubscription(opts => ...)`. Each call on `opts` maps to a KubeMQ `EventStoreStartPosition`. | Start position | Method | Behavior | Use case | | -------------- | ------------------------------------ | -------------------------------------------------------------------------- | ------------------------------------------------------------------------ | | New (default) | *(none)* | Only messages published after the subscription starts | Standard live fan-out; no replay | | First | `StartFromFirst()` | Replay all stored messages from the beginning, then continue with new ones | Full replay to rebuild a read model or populate a new service | | Last | `StartFromLast()` | Start from the most recently stored message, then continue | Resume roughly where you left off without replaying everything | | Sequence | `StartFromSequence(long sequence)` | Start at a specific sequence number and read onward | Resume from a known checkpoint after storing the last processed sequence | | Time | `StartFromTime(DateTimeOffset time)` | Start at or after an absolute timestamp | Replay from a specific moment, e.g. after a deployment or incident | | Time delta | `StartFromTimeDelta(int seconds)` | Start a relative number of seconds before now | "Give me the last N seconds of events" without computing absolute times | ### StartFromNew (Default) [#startfromnew-default] If you call `UseEventsStore()` without a start position, the subscriber receives only messages published after it connects — equivalent to volatile Events delivery, but with the channel still being persisted for other subscribers. ```csharp title="Program.cs" cfg.ReceiveEndpoint("audit-events", e => { e.UseEventsStore(); e.UseEventsStoreSubscription(opts => { // No call -- StartFromNew is the default: only new messages }); }); ``` ### StartFromFirst [#startfromfirst] Replay the entire stored history from sequence 1, then continue receiving new messages. This is the position to use for a full rebuild. ```csharp title="Program.cs" cfg.ReceiveEndpoint("audit-events", e => { e.UseEventsStore(); e.UseEventsStoreSubscription(opts => { opts.StartFromFirst(); }); }); ``` ### StartFromLast [#startfromlast] Begin at the most recently stored message, then continue. Useful for resuming approximately where a previous run stopped without re-reading the whole channel. ```csharp title="Program.cs" cfg.ReceiveEndpoint("audit-events", e => { e.UseEventsStore(); e.UseEventsStoreSubscription(opts => { opts.StartFromLast(); }); }); ``` ### StartFromSequence [#startfromsequence] Start from a specific sequence number — for example, a checkpoint you persisted after the last successfully processed message. ```csharp title="Program.cs" cfg.ReceiveEndpoint("audit-events", e => { e.UseEventsStore(); e.UseEventsStoreSubscription(opts => { opts.StartFromSequence(42); // Start from sequence 42 onward }); }); ``` ### StartFromTime and StartFromTimeDelta [#startfromtime-and-startfromtimedelta] Start from an absolute timestamp, or from a relative offset in seconds before now. The time-delta form avoids having to compute an absolute `DateTimeOffset`. ```csharp title="Program.cs" // Absolute: everything stored at or after one hour ago e.UseEventsStoreSubscription(opts => { opts.StartFromTime(DateTimeOffset.UtcNow.AddHours(-1)); }); // Relative: the last 3600 seconds (one hour) of events e.UseEventsStoreSubscription(opts => { opts.StartFromTimeDelta(3600); }); ``` ## Full Replay Example [#full-replay-example] This example enables EventsStore on an endpoint and replays the entire channel from the first stored message. A background publisher emits inventory changes; the consumer logs every historical and new event. It is taken from the transport's `examples/EventsStore/EventsStore.StartFromFirst` project. ### Start KubeMQ [#start-kubemq] MassTransit talks to KubeMQ over the native gRPC transport on port `50000`. Run a broker locally: Port `50000` is the gRPC port the transport connects to. Port `9090` is the shared HTTP server (REST, CloudEvents, MCP, and A2A connectors); it is not required for MassTransit but is exposed here for parity with the other examples. ### Add the Package [#add-the-package] The transport package targets .NET 8 and requires MassTransit 8.5.0 or later. ```bash dotnet add package MassTransit.KubeMQ ``` ### Configure, Publish, and Replay [#configure-publish-and-replay] ```csharp title="Program.cs" using MassTransit; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; var builder = Host.CreateApplicationBuilder(args); builder.Services.AddMassTransit(x => { x.AddConsumer(); x.UsingKubeMQ((context, cfg) => { cfg.Host("localhost", 50000); // Subscribe starting from the very first message in the EventsStore channel. // This replays all historical messages before receiving new ones. cfg.ReceiveEndpoint("eventsstore-start-from-first", e => { e.UseEventsStore(); e.UseEventsStoreSubscription(s => s.StartFromFirst()); }); }); }); builder.Services.AddHostedService(); var host = builder.Build(); await host.RunAsync(); // --- Message type --- public record InventoryChange(string Sku, int QuantityDelta, string Reason, DateTime ChangedAt); // --- Consumer --- public class InventoryChangeConsumer : IConsumer { private readonly ILogger _logger; public InventoryChangeConsumer(ILogger logger) => _logger = logger; public Task Consume(ConsumeContext context) { var msg = context.Message; _logger.LogInformation( "Replayed from first: SKU={Sku}, Delta={Delta}, Reason={Reason}", msg.Sku, msg.QuantityDelta, msg.Reason); return Task.CompletedTask; } } // --- Background publisher --- public class ReplayPublisher : BackgroundService { private readonly IBus _bus; private readonly ILogger _logger; public ReplayPublisher(IBus bus, ILogger logger) { _bus = bus; _logger = logger; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { await Task.Delay(3000, stoppingToken); var counter = 0; while (!stoppingToken.IsCancellationRequested) { counter++; var change = new InventoryChange( $"SKU-{(counter % 5) + 100}", counter % 2 == 0 ? 10 : -3, counter % 2 == 0 ? "restock" : "sale", DateTime.UtcNow); _logger.LogInformation("Publishing inventory change #{Counter}", counter); await _bus.Publish(change, stoppingToken); await Task.Delay(2000, stoppingToken); } } } ``` Run it once to seed the channel, then run it again: on the second run the consumer replays every previously stored `InventoryChange` from sequence 1 before it begins logging newly published messages. ## EventsStore vs Events [#eventsstore-vs-events] EventsStore and Events share fan-out and consumer-group semantics; the difference is persistence and replay. | Feature | Events | EventsStore | | ---------------------------- | ---------------------- | ---------------------------------------- | | Persistence | No | Yes | | Replay | No | Yes (configurable start position) | | Delivery when no subscribers | Message dropped | Message stored for later delivery | | Delayed delivery | Not supported | Not supported | | Fan-out | All active subscribers | All subscribers (including late joiners) | | Consumer groups | Yes | Yes | | Ordering | Per-channel | Per-channel with sequence numbers | ## Backpressure During Replay [#backpressure-during-replay] Replaying a large channel with `StartFromFirst` can deliver a burst of stored messages. Two settings act as flow control so consumers are not overwhelmed: * **`ConcurrentMessageLimit`** — MassTransit's standard concurrency limit. The transport honors it during replay, so no more than the configured number of messages are processed at once. * **`MaxPollMessages`** — the KubeMQ poll batch size on the receive endpoint. A smaller batch limits how many stored messages are pulled per poll, smoothing throughput during replay. ```csharp title="Program.cs" cfg.ReceiveEndpoint("audit-events", e => { e.MaxPollMessages = 50; // Pull at most 50 stored messages per poll e.UseEventsStore(); e.UseEventsStoreSubscription(opts => opts.StartFromFirst()); }); ``` ## Consumer Groups with EventsStore [#consumer-groups-with-eventsstore] Consumer groups behave with EventsStore exactly as they do with Events: multiple instances sharing the **same endpoint name** form a group, and KubeMQ load-balances stored events across the group members — each event is delivered to exactly one member rather than fanned out to all of them. ```csharp title="Program.cs" // Every instance of this service uses the same endpoint name ("audit-service"), // so they form one consumer group and share the load. cfg.ReceiveEndpoint("audit-service", e => { e.UseEventsStore(); e.UseEventsStoreSubscription(opts => { opts.StartFromFirst(); }); }); ``` This mirrors the transport's `examples/EventsStore/EventsStore.ConsumerGroup` project, where two endpoints bound to the channel `eventsstore-consumer-group` divide the replayed and new messages between them. ## Limitations [#limitations] * **No delayed delivery.** EventsStore does not support scheduled or delayed publish. Attempting to publish with a delay throws `KubeMQTransportConfigurationException`. If you need delayed delivery, use [Queues (Send)](/integrations/masstransit/how-to/queues) with `ctx.Delay`. * **No per-message ack/reject.** Unlike queue messages, stored events cannot be individually acknowledged or rejected. Processing failures are handled by MassTransit's standard retry pipeline rather than transport-level nack. * **Storage is server-governed.** How much history is retained depends on the KubeMQ server's EventsStore configuration, not on the transport. ## Next steps [#next-steps]