Articles · Projections & Read Models ·033

Persistent Named Subscriptions

Runnable sample on GitHub

Sample: samples/033-persistent-named-subscriptions — a durable, named position over the event log that resumes across restarts. Run its tests with dotnet test (needs Docker for PostgreSQL).

Prerequisites: 005 — Event Sourcing Basics, 007 — Projections.

Overview

A projection (article 007) is a catch-up reader: it walks the global event log forward, updates a read model, and remembers how far it got so a restart resumes instead of reprocessing two years of history. That “how far it got” is a checkpoint — a durable position on the log. The projection runner owns it and you never touch it directly.

But projections are not the only thing that wants to read the log forward and remember its place. You might ship events to an external system, feed an analytics pipeline, drive a webhook integration, or build a bespoke reader the framework’s projection machinery doesn’t model. All of those need the same primitive the projection runner uses internally: a durable position, identified by name, that only moves forward. That primitive is ISubscriptionStore, and a persistent named subscription is one named position recorded in it.

This article wires the PostgreSQL-backed store (EfCoreSubscriptionStore) and proves the property that makes it worth having: a named consumer’s position survives the process. A fresh process against the same database reads the stored position and resumes from exactly where the last one stopped.

Why this exists

A catch-up consumer that doesn’t persist its position has only bad options on restart. Replay from the start every time — correct but ruinously expensive once the log is large, and it re-delivers every event the consumer ever saw. Start from “now” and silently drop everything that arrived while it was down — fast, but it loses events, which is usually unacceptable for an integration feed. The only good answer is to record the position durably and resume from it: process the backlog exactly once on restart, then continue.

Projection checkpoints already do this, but they are sealed inside the projection runner. The moment you have a consumer that isn’t a Relay projection — an exporter, a custom feed, a bridge to another bus — you’d be tempted to hand-roll a last_processed_id column, and hand-rolled position tracking is where the subtle bugs live: positions that regress on a retry, races between two writers, or a position keyed on the wrong column so a late-committing transaction gets skipped forever. ISubscriptionStore gives you the framework’s correct, gap-safe, monotonic position-keeping for arbitrary named consumers, backed by the database you already run.

When to use this

  • An external feed or integration. You read events forward and push them somewhere — a partner API, another message bus, a data warehouse — and must resume cleanly after a deploy or crash.
  • A custom catch-up reader the projection runner doesn’t model. Anything that consumes the global log on its own cadence and needs a durable place to stand.
  • Multiple independent consumers of one log. Each gets its own named position, advancing at its own pace, with no coupling between them.

When not to use this

  • You’re building a read model. Use a projection (article 007). The projection runner already owns its checkpoint, gives you ordered delivery, failure handling, dead-lettering, and rebuilds — a named subscription is the raw position primitive underneath that, not a replacement for it. Reach here only when your consumer genuinely isn’t a projection.
  • You only ever need current state. Then you don’t need to read the log forward at all; query a projection or a table.
  • Per-request work. A subscription position is for a long-running forward reader, not for handling a single command.

Concepts

Named subscription. A consumer identified by a string name ("orders-export", "analytics-bridge"). The name is the primary key in relay_subscriptions — one row per consumer — so distinct consumers never collide and each resumes independently.

Stored position. The position is an EventCursor — a (TxId, Position) pair, not a bare offset. Position is the database-assigned global_position; TxId is the inserting transaction id. Both are needed because global_position is allocated at INSERT but made visible at COMMIT, and commit order is not allocation order. Ordering and checkpointing on (tx_id, position) is gap-safe: a later-committing transaction always has a strictly greater TxId, so a catch-up reader capped at the visibility horizon can never skip it. An unknown subscription reads back EventCursor.Start ((0, 0)) — “before the first event,” i.e. a full replay.

Monotonic advance. AdvanceAsync(name, cursor) only ever moves the stored position forward (by the (TxId, Position) order). A backward or equal ack — a retry that re-acknowledges an earlier place, a duplicate, an out-of-order callback — is ignored. The store serializes concurrent advances for a name with a transaction-scoped PostgreSQL advisory lock, so two writers can’t race the position backward.

Resume. On startup a consumer calls GetPositionAsync(name), gets back the last acknowledged cursor (or Start), and reads the log forward from there. Because the position is a row in your database, it outlives the process that wrote it — that is the whole point of “persistent.”

Architecture

flowchart TD
    subgraph Process A (before restart)
        CA[Consumer 'orders-export'] -->|read forward from cursor| LOG[(relay_events)]
        CA -->|AdvanceAsync name, cursor| STORE[(relay_subscriptions<br/>one row per name)]
    end

    CA -.process exits / crashes / redeploys.-> X[ ]

    subgraph Process B (after restart)
        CB[Consumer 'orders-export' again] -->|GetPositionAsync name| STORE
        STORE -->|stored cursor 42,4200| CB
        CB -->|resume read forward from 42,4200| LOG
    end

Process B is a brand-new process with nothing shared in memory with A. It learns where to resume only from the durable row in relay_subscriptions — so the position is the seam that makes the restart invisible.

Building it step by step

The full sample is in samples/033-persistent-named-subscriptions.

1. Map the relay tables onto your DbContext

protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    base.OnModelCreating(modelBuilder);
    modelBuilder.ApplyRelayEventStore();      // relay_events — the log a subscription reads forward over
    modelBuilder.ApplyRelaySubscriptions();   // relay_subscriptions — the durable per-name position
}

2. Register the store

// The store opens its own connection per call, so register a context factory:
services.AddDbContextFactory<FeedDbContext>(o => o.UseNpgsql(connectionString));
services.AddRelaySubscriptionStorePostgres<FeedDbContext>();   // ISubscriptionStore (singleton)

3. Use it from a consumer — read where you left off, work, acknowledge

public sealed class OrdersExportConsumer(ISubscriptionStore subscriptions, IEventStore events)
{
    private const string Name = "orders-export";

    public async Task PumpAsync(CancellationToken ct)
    {
        var cursor = await subscriptions.GetPositionAsync(Name, ct);   // EventCursor.Start on first run
        // ... read events forward from `cursor`, ship each one to the external system (idempotently) ...
        var lastProcessed = new EventCursor(txId, position);           // the last event you acknowledged
        await subscriptions.AdvanceAsync(Name, lastProcessed, ct);     // forward-only; safe to re-call
    }
}

AdvanceAsync is monotonic, so re-acknowledging after a partial failure never rewinds the feed, and two instances acking concurrently can’t move it backward.

Complete source code

File Shows DB?
FeedDbContext.cs ApplyRelayEventStore() + ApplyRelaySubscriptions() —
FeedFixture.cs Testcontainers Postgres + AddRelaySubscriptionStorePostgres; builds a fresh provider to simulate a restart Yes
PersistentSubscriptionTests.cs starts at 0, advance persists, monotonic, resume-across-restart Yes

Running the example

dotnet test samples/033-persistent-named-subscriptions/Feed.Subscriptions.Tests

This needs Docker — the fixture starts a real postgres:16 via Testcontainers, applies the schema with EnsureCreatedAsync(), and exercises the store against it. The restart test does the only honest thing: it disposes the first ServiceProvider entirely (every connection, cache, and the store singleton) and builds a second, independent provider against the same connection string, proving the position came from the database and not from in-process state. In production you replace EnsureCreatedAsync() with EF migrations (or the baseline SQL in libraries/nuvora-nexus-relay/docs/migrations/) and add the AddRelaySchema() readiness check.

Testing

The split mirrors the rest of Relay: the irreducible database interaction lives in one focused integration test. There is nothing pure to extract here — the entire value of a persistent position is that it survives a process, which is a database-level fact. So the test proves exactly that, against real PostgreSQL: an unknown name reads EventCursor.Start; AdvanceAsync persists; a backward or equal ack is ignored (monotonic); and a fresh provider against the same database resumes from the stored cursor. Mocking the store would test the mock, not the guarantee.

// Resume across restart — the property that justifies "persistent".
var processOne = FeedFixture.BuildProvider(connectionString);
await using (processOne)
    await processOne.GetRequiredService<ISubscriptionStore>()
        .AdvanceAsync("orders-export", new EventCursor(42, 4200));   // process #1 makes progress, then exits

var processTwo = FeedFixture.BuildProvider(connectionString);       // brand-new process, same database
await using (processTwo)
    (await processTwo.GetRequiredService<ISubscriptionStore>().GetPositionAsync("orders-export"))
        .Should().Be(new EventCursor(42, 4200));                     // resumes where #1 stopped

Production considerations

  • One owner per subscription. A named subscription is a single logical consumer’s position. If two instances advance the same name concurrently, the store keeps the position monotonic but they’re competing — they’ll process overlapping ranges. For a singleton consumer, run it under leader election (article 023); to parallelize, give each worker its own partition and its own subscription name (article 023’s partitioner), not one shared name.
  • Delivery is at-least-once; consumers must be idempotent. You acknowledge after doing the work, so a crash between “did the work” and “advanced the position” replays the last range on restart. That is the safe direction (no event is lost), but it means the downstream effect can happen more than once — design the consumer so reprocessing an event is harmless (upsert by event id, dedupe at the sink).
  • Acknowledge after the side effect, never before. Advancing first and then doing the work turns a crash into lost events. Advance last so a crash only re-delivers.
  • Advance in sensible batches. Advancing after every single event is correct but chatty; advancing after a batch bounds the replay window to that batch on restart. Pick a batch size that trades replay cost against write traffic.
  • Schema is a deploy-time concern. Map relay_subscriptions (ApplyRelaySubscriptions()), ship it with EF migrations or the baseline script, and wire AddRelaySchema() so a missing table fails readiness rather than the first read.

Common mistakes

  • Acknowledging before the work is durable. The classic data-loss bug: advance the position, then the process dies before the event reaches the sink. Always advance after the effect.
  • Sharing one name across instances expecting partitioned parallelism. One name is one position; multiple advancers contend, they don’t shard. Use distinct names (per partition) or a single leader.
  • Assuming exactly-once delivery. It’s at-least-once. A consumer that isn’t idempotent will double-emit on every restart that replays a range. Dedupe at the sink.
  • Rolling your own last_id column “because it’s just one integer”. A bare offset checkpoints on position alone and will skip a late-committing transaction (see the EventCursor note above). The store checkpoints on (tx_id, position) for exactly this reason.
  • Treating it as a queue. A subscription is a position over a log you keep, not a queue that deletes consumed messages. The events stay; the position moves.

Tradeoffs

Benefits. Durable resume for any catch-up consumer, not just framework projections; gap-safe, monotonic position-keeping you don’t have to get right yourself; one named row per consumer with no cross-consumer coupling; and no new infrastructure — it rides the PostgreSQL you already run.

Costs. It is a low-level primitive: it stores a position and nothing else. Unlike the projection runner, it gives you no ordered delivery loop, no failure handling, no dead-lettering, no rebuild tooling — you write the read-forward loop and the idempotency. At-least-once delivery pushes idempotency onto every consumer. And it is single-position-per-name: parallel consumption needs partitioning you wire yourself.

Alternatives

  • Relay projections (article 007). If your consumer is a read model, use projections — the runner owns the checkpoint and gives you the delivery loop, failure handling, and rebuilds for free. Named subscriptions are the primitive beneath; don’t reimplement projections on top of them.
  • Kafka consumer offsets. Kafka commits a per-partition offset per consumer group — the same idea at the broker layer, with built-in partitioned parallelism and rebalancing. The right tool if your events already flow through Kafka; a named subscription is the equivalent for the PostgreSQL log Relay already keeps, with no extra system to run.
  • EventStoreDB persistent subscriptions. EventStoreDB offers server-managed persistent subscriptions with competing consumers, per-message acks/nacks, and retry/park semantics — richer than a single monotonic position. If you run EventStoreDB, use them; if you’re on Relay’s Postgres event store, ISubscriptionStore is the position-keeping half, and you supply the delivery loop.

Lessons from production systems

  • The integration that “replays from the beginning on every deploy” is a perennial incident: it works in staging with a small log and falls over in production where the log is enormous. A durable position fixes it the same afternoon.
  • “Start from now on restart” looks fine in a demo and quietly drops every event that arrived during the deploy. The teams that sleep well persist the position and accept at-least-once with idempotent consumers.
  • The position-regression bug — a retry that re-acks an earlier place and rewinds the feed — is brutal to diagnose because it only triggers under failure. Monotonic advance in the store eliminates the whole class.

Should you use this?

If you… Then…
build a read model no — use a projection (article 007); it owns the checkpoint
run a custom catch-up consumer / external feed that must resume yes — a named subscription
ship events to another system or bus yes — durable position + idempotent sink
need partitioned parallel consumption yes, but one name per partition (article 023), not one shared name
only ever query current state no — you don’t read the log forward at all

Next steps

  • 007 — Projections: the framework-owned catch-up reader this primitive sits beneath — ordered delivery, failure handling, dead-lettering, and rebuilds over the same gap-safe cursor.
  • 023 — Distributed Coordination: run a singleton consumer under leader election, or partition a keyed feed so each worker owns its own subscription.
  • 021 — Observability: track how far each named subscription lags the head of the log so a stalled feed is visible before it’s a backlog.