NATS JetStream implementation

NATS JetStream implementation

The platform event log is backed by NATS JetStream. External applications always go through the PubSub service, which exposes a REST API for publishing and SignalR hubs for subscribing (see Concepts). Platform services running inside the cluster connect to NATS directly using the shared eventing libraries — they publish and consume against the same stream as the PubSub service, without the extra hop through its REST/SignalR endpoints.

Transport abstraction

Publishing and consuming is built on transport-agnostic IEventWriter / IEventReader abstractions. The concrete transport is selected by the EventWriterType and EventReaderType configuration keys; besides NATS JetStream there are in-memory (local development default), Redis, Azure EventHub and Kafka providers. Deployed environments use NatsJetStreamEventWriter / NatsJetStreamEventReader.

The NATS integration uses the official NATS.Net v2 client.

Stream

There is one JetStream stream per environment, named SITE_EVENTS_{Environment}:

Setting	Value
Retention	Limits, MaxAge 7 days
Storage	File
Replicas	3 (1 locally)
Acknowledgements	Enabled (NoAck = false)
Duplicate window	2 minutes
Subjects	events.{environment}.*.*.>

The stream is provisioned lazily: whichever writer or reader connects first checks whether the stream exists and creates it if not. Concurrent creation is race-safe — a loser of the race falls back to the stream created by the winner. If a pre-provisioned stream does not cover the expected subject filter, a warning is logged instead of silently rewriting the stream configuration.

The stream is always addressed by its configured name; a stream that happens to cover the same subjects under a different name is never adopted, because that would put several environments on one stream where they overwrite each other's consumers (consumer names are per service, not per environment). If another stream already claims the environment's subjects, JetStream refuses to create ours and startup fails with an explicit error — the conflicting stream has to be removed or narrowed before deploying.

Subjects

Events are published to subjects following the convention

events.{environment}.{customerId}.{entityType}.{resourceId}

where entityType is the CloudEvent type and customerId comes from the caller's token. All tokens are sanitized to [A-Za-z0-9_-] so that NATS-reserved characters (., *, >, whitespace) in user-provided values cannot break out of their subject segment. The per-environment subject filter events.{environment}.*.*.> isolates environments sharing one NATS cluster, and the customer segment allows server-side filtering of subscriptions per tenant, event type and resource.

Publishing

External applications publish through POST /api/eventing/event (see Publish events). Platform services running inside the platform Kubernetes namespace use the internal endpoint POST /api/pubsub/eventing/impersonated-event, which publishes an event on behalf of a given tenant, user and project and is not part of the public API surface.

Events follow the CloudEvent specification and are serialized as structured JSON. Custom extension attributes carry platform context: customerId, projectId, userId, resourceId, operationId and offset.

Every publish sets the Nats-Msg-Id header (from the event's operationId, falling back to the event id). Together with the stream's 2 minute duplicate window this gives native JetStream deduplication, so a retried publish of the same operation is not stored twice.

Publishing retries up to 3 times with reconnect and backoff; connecting retries up to 10 times with capped backoff.

Consuming

The PubSub service consumes the stream with a durable pull consumer named pubsub-reader-{Environment}:

Setting	Value
Ack policy	Explicit
Ack wait	30 seconds
Max deliveries	5

Messages that fail processing are negatively acknowledged (Nak) and redelivered; after 5 failed deliveries the message is dropped (there is no dead-letter stream). Consumed events are fanned out to SignalR subscribers, filtered by event type, tenant and resource.

Subscribers created for SignalR clients use consumers named subscriber-{clientId} with an inactivity threshold of 5 minutes, so the server automatically cleans up consumers of disconnected clients.

A subscriber resuming from a stored offset is validated against the stream's last sequence first. If the offset is past the end of the stream — which means the stream was replaced and its sequence numbers restarted — the offset is dropped and the consumer starts from new messages only, instead of failing consumer creation.

Direct service connection

Platform services connect to NATS through drop-in implementations of the same abstractions they already use for the PubSub REST/SignalR path:

Abstraction	PubSub implementation	Direct NATS implementation
IPubSubClient (publish)	PubSubClient (HTTP to PubSub service)	NatsJetStreamPubSubClient
IPubSubSubscriber (subscribe)	PubSubSubscriber (SignalR to PubSub service)	NatsJetStreamSubscriber

Both direct implementations are configured from the NatsJetStream configuration section with Url, StreamName and Environment; the transport is selected by the EventingTransport setting:

"EventingTransport": "NatsJetStream",
"NatsJetStream": {
  "Url": "nats://localhost:4222",
  "StreamName": "SITE_EVENTS_local",
  "Environment": "local"
}

NatsJetStreamPubSubClient encodes events into the same structured CloudEvent JSON as the PubSub service — including the exact data bytes the HTTP path produces (the event payload is serialized with the same Newtonsoft settings PubSubClient uses on the wire) — so directly published messages are indistinguishable from messages published through the REST API. The caller identity stamped into the event (customerid, projectid, userid, identityid extension attributes) and used for the tenant segment of the subject comes either from the explicit tenant/user/project arguments (impersonated publishing) or is parsed from the cid/uid/pid/iid parts of the SAS token. Token expiry (ex) is checked, but the signature is not validated — the publishing service is already a trusted in-cluster caller. Publishing uses the same Nats-Msg-Id deduplication (from operationId) and retry behaviour as the PubSub service writer.

Each direct publish is tracked as an Application Insights NATS dependency (target = NATS URL, data = subject, with TenantId/ResourceId/OperationId properties), replacing the HTTP dependency and PubSub request telemetry of the REST path, and the current W3C trace context is propagated in the traceparent NATS message header. NatsJetStreamSubscriber restores this trace context around the message handler, so telemetry emitted while processing a directly consumed message is correlated with the publishing operation.

Two cases fall back to publishing through the PubSub REST API: transient events (they are distributed over Redis by the PubSub service and must not be persisted in the JetStream stream) and events whose identity cannot be resolved from the SAS token (missing cid, or a missing/expired ex — the PubSub service then rejects the token authoritatively).

NatsJetStreamSubscriber creates a consumer named subscriber-{serviceName} (the service mesh host name is used as client id) with an inactivity threshold of 5 minutes and filters by event types and optionally resource identifiers. Before handing a message to the service it stamps the JetStream stream sequence into the CloudEvent offset attribute — the same thing the PubSub service does for SignalR clients — so offset tracking and resume work identically on both transports. Note that transient events never reach the stream (they are distributed over Redis and SignalR only), so a service consuming through NatsJetStreamSubscriber does not receive them — services relying on transient events must keep the SignalR subscriber.

Offsets and replay

The subscription offset (see Concepts) maps directly to the JetStream deliver policy:

Offset	Deliver policy
beginning	All
specific point	ByStartSequence
end	New

Replay uses ephemeral consumers (named replay-{clientId}-{guid}, auto-removed after 5 minutes of inactivity) and supports either a bounded sequence range or replaying everything up to the current end of the stream. Transient events are not replayable.

Configuration

The PubSub service is configured through the writer/reader options, for example:

"EventWriterType": "NatsJetStreamEventWriter",
"NatsJetStreamEventWriter": {
  "TopicName": "SITE_EVENTS_{Environment}",
  "Options": {
    "natsUrl": "nats://nats.nats.svc.cluster.local:4222",
    "environment": "{Environment}",
    "natsReplicas": "3"
  }
},
"EventReaderType": "NatsJetStreamEventReader",
"NatsJetStreamEventReader": {
  "TopicName": "SITE_EVENTS_{Environment}",
  "Options": {
    "natsUrl": "nats://nats.nats.svc.cluster.local:4222",
    "environment": "{Environment}",
    "consumer.name": "pubsub-reader-{Environment}",
    "offset": "latest"
  }
}

The reader additionally supports subject.filter, replay.to.now, replay.from.sequence and replay.to.sequence options.

Health

The PubSub service registers a nats-jetstream health check that pings the NATS server and queries JetStream account info with a 5 second timeout. It is exposed through the standard platform health endpoint of the service.

Deployment

NATS runs as a 3 node cluster (official nats Helm chart) in the nats namespace with JetStream enabled and file storage on persistent volumes. Prometheus metrics are exported by the chart's exporter and NATS Surveyor. Locally, the Aspire runner starts a single-node NATS container with JetStream (-js) on port 4222.