Events Store (Durable Publish)

Overview

By default, MassTransit Publish<T>() 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<T>() 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. This page documents the MassTransit transport's API surface only.

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

EventsStore can be turned on for the whole bus or for individual endpoints.

Global (All Publish Endpoints)

Calling cfg.UseEventsStore() inside the UsingKubeMQ callback makes every publish endpoint use EventsStore. All Publish<T>() calls become durable.

services.AddMassTransit(x =>
{
    x.AddConsumer<AuditEventConsumer>();

    x.UsingKubeMQ((ctx, cfg) =>
    {
        cfg.Host("localhost", 50000);
        cfg.UseEventsStore();  // All Publish<T>() 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

To make only certain channels durable, call UseEventsStore() on the receive endpoint configurator. Other endpoints continue to use volatile Events.

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

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.

cfg.ReceiveEndpoint("audit-events", e =>
{
    e.ConfigureKubeMQ(k =>
    {
        k.UseEventsStore();  // Enable via the transport-specific configurator
    });
});

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.

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.

cfg.ReceiveEndpoint("audit-events", e =>
{
    e.UseEventsStore();
    e.UseEventsStoreSubscription(opts =>
    {
        // No call -- StartFromNew is the default: only new messages
    });
});

StartFromFirst

Replay the entire stored history from sequence 1, then continue receiving new messages. This is the position to use for a full rebuild.

cfg.ReceiveEndpoint("audit-events", e =>
{
    e.UseEventsStore();
    e.UseEventsStoreSubscription(opts =>
    {
        opts.StartFromFirst();
    });
});

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.

cfg.ReceiveEndpoint("audit-events", e =>
{
    e.UseEventsStore();
    e.UseEventsStoreSubscription(opts =>
    {
        opts.StartFromLast();
    });
});

StartFromSequence

Start from a specific sequence number — for example, a checkpoint you persisted after the last successfully processed message.

cfg.ReceiveEndpoint("audit-events", e =>
{
    e.UseEventsStore();
    e.UseEventsStoreSubscription(opts =>
    {
        opts.StartFromSequence(42);  // Start from sequence 42 onward
    });
});

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.

// 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

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.

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

dotnet add package MassTransit.KubeMQ

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<InventoryChangeConsumer>();

    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<ReplayPublisher>();

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<InventoryChange>
{
    private readonly ILogger<InventoryChangeConsumer> _logger;

    public InventoryChangeConsumer(ILogger<InventoryChangeConsumer> logger) => _logger = logger;

    public Task Consume(ConsumeContext<InventoryChange> 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<ReplayPublisher> _logger;

    public ReplayPublisher(IBus bus, ILogger<ReplayPublisher> 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);
        }
    }
}

EventsStore vs Events

EventsStore and Events share fan-out and consumer-group semantics; the difference is persistence and 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.

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 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.

// 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

Next steps