plexsphere

Appearance

Signed SSE Event Bus ​

This document is the authoritative bounded-context reference for the Signed SSE Event Bus that ships under internal/mesh/sse and exposes through GET /v1/nodes/{id}/events. It covers the ubiquitous language, the envelope schema and JetStream subject layout, the publish pipeline, the Last-Event-ID semantics, the in-memory nonce set, the threat model, the multi-replica handoff, and the operator runbook.

The bus is the plexsphere → plexd push channel: per-node long-lived SSE streams that carry signed envelopes with nonce/timestamp replay protection and a Last-Event-ID replay path on reconnect. Anything outside that surface — the underlying signer (internal/signing), the Postgres outbox, the JetStream broker, the audit sink — is a collaborator the bus orchestrates, not a concern of this document. This document is the implementation-side reference for the README's Signed Event Bus specification; the README pins the wire-level event-type catalogue and this document pins the in-process pipeline that emits it.

Status — a minimal slice is live behind an opt-in gate ​

The Signed SSE Event Bus was descoped from the first production release, and a minimal signed slice has since been partially un-descoped. The foundational primitives and the transport surface at GET /v1/nodes/{id}/events exist, and when the operator sets the opt-in NATS URL the composition root threads the full producer → relay → consumer pipeline so the surface streams signed envelopes; otherwise the endpoint stays on its fail-closed 501 stub. The push channel is an optimisation only — every dispatch converges through the reconciliation-pull baseline (GET /v1/nodes/{id}/state) with the bus switched off, so the stream never becomes a correctness dependency. See the un-descope decision for the rationale and the remaining deferred workstream.

Opt-in gate and the conditional 501 ​

GET /v1/nodes/{id}/events answers 501 with code: signed_event_bus_not_provisioned only when the bus is not configured — the 501 is conditional, not unconditional. The composition root in cmd/plexsphere/sse_factory_prod.go reads PLEXSPHERE_SSE_NATS_URL:

  • Unset (the default posture): BuildProductionSSEFactory returns a bundle that constructs only the in-memory NonceStore. With EventStream and SignatureVerifier nil, the handler mount gate in events_dispatch.go keeps the surface on its 501 stub — the unchanged posture for every deployment that has not opted in.
  • Set to a NATS URL: buildFullSSEWiring dials JetStream, ensures the PLEXSPHERE_NODE_EVENTS stream, builds the signer-backed per-Domain resolver and EnvelopeVerifier, the publisher, the relay over the sqlc-backed cursor store, and the consumer-side EventStreamAdapter, then populates the EventStream, SignatureVerifier, Publisher, and the relay + rotation run hooks. With the full security cone (EventStream, NonceStore, SignatureVerifier, RelationChecker, NodeRepo) wired, the surface streams signed envelopes instead of the 501.

The mount gate is all-or-nothing: a nil on ANY of the five load-bearing ports fails closed with the 501 rather than admitting traffic past a gate that is not in place.

The minimal five-event wire-type set ​

The relay fans out a closed five-member wire-type set in this slice, pinned by sseMinimalWireTypes in cmd/plexsphere/sse_factory_prod.go and passed as RelayConfig.AllowedWireTypes:

  • node_state_updated
  • policy_updated
  • bridge_config_updated
  • action_request
  • session_setup

A row whose source outbox event_type maps (via the publisher's wireTypeFor dispatch table) to a wire type outside this set is skipped-and-advanced: the relay never publishes it, but it advances the cursor past the row (bumping sse_relay_skip_total{reason=wire_type_not_allowed} and logging a WARN) so a single out-of-set row can never head-of-line block an in-set successor. A nil or empty AllowedWireTypes would allow every wire type; the composition root passes the closed five-member set so the bus carries exactly the node-facing wire literals a plexd agent consumes today.

The set bounds the node-facing bus only. The relay's signing-key rotation fan-out runs ahead of both skip-and-advance gates, so a signing_key_rotated row still reaches DomainSigningKeyResolver.Subscribe and invalidates the per-Domain public-key cache even though it is never published to a Node. Gating the invalidation on the allowed set would leave a retired key trusted until the process restarts.

Single-replica constraint ​

The slice runs single-replica only. The handler-side NonceStore is the process-local in-memory store, so the symmetric replay check would degrade silently across replicas if the nonce state were not shared. The composition root therefore refuses to boot when PLEXSPHERE_SSE_STREAM_REPLICAS > 1 is set with the in-memory store wired — the boot guard rejects the misconfiguration before /readyz greens and directs operators to pin replicas=1 or wire a distributed nonce store before scaling. PLEXSPHERE_SSE_STREAM_REPLICAS defaults to 1.

Opt-in configuration ​

The knobs cmd/plexsphere/sse_factory_prod.go reads to enable and tune the slice:

Env varDefaultEffect
PLEXSPHERE_SSE_NATS_URLunsetThe opt-in gate. Empty keeps the NonceStore-only wiring and the 501 stub; a non-empty NATS URL dials JetStream and assembles the full publish → relay → consume graph.
PLEXSPHERE_SSE_STREAM_REPLICAS1JetStream stream replica count. Boot refuses > 1 while the in-memory NonceStore is wired.
PLEXSPHERE_SSE_RELAY_INTERVAL250msCadence the relay loop ticks at when draining the outbox onto the bus. Parsed as a Go time.Duration; non-positive values are rejected at boot.
PLEXSPHERE_ACCESS_SIGNER_ENDPOINTunsetThe plexsphere-signer gRPC target the publisher signs envelopes through and the verifier resolves per-Domain public keys against. REQUIRED once the NATS gate is set — the factory refuses to build the slice without it. Reused from the access-issuance signer configuration, so a deployment configures one signer, not two.
PLEXSPHERE_ACCESS_SIGNER_CLIENT_CERT / PLEXSPHERE_ACCESS_SIGNER_CLIENT_KEY / PLEXSPHERE_ACCESS_SIGNER_SERVER_CAunsetThe mTLS file paths the signer client dials with; the signer always runs behind mTLS. Reused from the access-issuance signer configuration.

The nonce-set tuning knobs PLEXSPHERE_SSE_NONCE_TTL and PLEXSPHERE_SSE_MAX_LIVE_NONCES are documented under Nonce set design.

Deferred workstream ​

The un-descope is partial. The following stay deferred and are tracked in the roadmap:

  • Head-of-line-blocking mitigation — per-subject FilterSubject consumption and the skip-and-advance schema work on no_signing_key rows; today an un-provisioned Domain key halts that Domain's relay batch and the sse_relay_skip_total{reason=no_signing_key} counter is the operator signal.
  • A distributed nonce store — the prerequisite for relaxing the single-replica boot guard to replicas > 1.
  • Relaying session_revoked and node_secrets_updated over SSE — these are not in the five-event set, so revocation currently converges only through the reconciliation-pull sessions block draining.
  • The replay-window 410 behaviour — the Last-Event-ID out-of-window 410 last_event_id_outside_replay_window arm stays unreachable until the stream (low, high) bounds are threaded through the EventStream port.

The chainsaw E2E fixture at ../../../tests/e2e/mesh/dispatch-channels-sse/chainsaw-test.yaml now exercises the real plexsphere binary end-to-end with the bus opted in: it boots the production cmd/plexsphere with PLEXSPHERE_SSE_NATS_URL set (and PLEXSPHERE_SSE_STREAM_REPLICAS=1) against NATS JetStream and a live plexsphere-signer over mTLS, commits a dispatch to plexsphere.outbox_events, and asserts it arrives on GET /v1/nodes/{id}/events as a signer-signed action_request frame that the same NSK callback leg settles. A fallback step scales NATS to zero and proves a fresh dispatch still converges through the reconciliation pull alone, so the bus is a pure latency optimisation over a pull that never stops converging. The older wire-format compatibility fixture at ../../../tests/e2e/mesh/sse-signed-envelopes/chainsaw-test.yaml still deploys cmd/sse-stub-plexd as a stub producer/consumer pair to pin the wire format independently of the production binary.

What ships and is load-bearing today, wired into the production binary on the opt-in path:

  • The publish pipeline (sign-before-publish with Nats-Msg-Id dedup), the bounded-TTL nonce store with capacity-LRU eviction and Prometheus gauge, the outbox-relay with cursor advance, wire-type filtering, and ErrSigningKeyNotProvisioned skip-and-log, the Last-Event-ID parser with all four arms (empty / numeric / non-numeric / out-of-window; the out-of-window 410 stays unreachable until the stream bounds are threaded — see the deferred workstream above), the canonical stream-name + subject helpers, the plexsphere.sse_relay_cursor migration, the depguard allow-list, the OpenAPI surface byte-equality drift gate, the in-process EventStreamAdapter, the multi-replica boot guard refusing replicas > 1 with the in-memory NonceStore, and the consumer-side capture harness contract. The relay loop, stream provisioning, and the sqlc-backed RelayStore now have a production caller — buildFullSSEWiring in cmd/plexsphere/sse_factory_prod.go, behind the PLEXSPHERE_SSE_NATS_URL gate.
  • The node_state_updated default wire event-type, derived from a tenancy.NodeRegistered outbox row (and every other outbox literal that does not map onto a more specific wire type).
  • The policy_updated wire event-type. The publisher's wireTypeFor dispatch table maps the closed policy outbox literals (policy_revision_created and policy_deleted) onto the wire literal policy_updated; the producer-side fan-out runs from the compile-service arm — one publish per matched Node on revision-created, one per previously compiled Node on delete. See ../policy/events.md for the full dispatch table, fan-out algorithm, and payload schema.
  • The bridge_config_updated wire event-type. The wireTypeFor dispatch table collapses the closed seven-member set of bridge.* outbox literals onto the single wire literal bridge_config_updated, and the four bridge application services each emit one signed envelope per Node hosted by the changed bridge Resource, carrying the whole-object effective bridge config. A bridge-mode Node converges on the reconciliation-pull bridge block whether or not the bus is opted in. See ../bridge/events.md for the full dispatch table, the per-Node fan-out algorithm, the BridgeConfigUpdatedPayload schema, and the byte-equality contract with the pull snapshot.
  • The action_request and session_setup wire event-types — the two node-dispatch literals that complete the five-event set. They carry the per-Node action dispatch and per-Node session provisioning payloads the reconciliation pull also surfaces in its executions and sessions blocks, so a Node converges on either channel. See ../actions/events.md and ../access.md.

Because the bus is an optimisation only, downstream consumers MUST NOT treat the push channel as a correctness dependency: a Node converges through the reconciliation pull (GET /v1/nodes/{id}/state) whether or not the operator has opted the bus in, and the surface answers 501signed_event_bus_not_provisioned for any deployment that has not.

Cross-references ​

  • ../../../README.md#signed-event-bus — top-level specification of the bus, the per-node SSE surface, the event-type catalogue, and the Last-Event-ID replay contract.
  • ./peers.md — Key and Peer Manager bounded sub-context reference; the six peer event_type strings (peer_registered, peer_psk_assigned, peer_deregistered, peer_endpoint_changed, rotate_keys, peer_key_rotated) the relay routes verbatim.
  • ./key-rotation.md — the key-rotation workflow that emits the rotate_keys command and the peer_key_rotated notification onto this bus.
  • ../signing-rotation.md — the Signing Service key-rotation workflow that emits the signing_key_rotated event onto this bus.
  • ../policy/events.md — the policy events surface that maps the closed policy outbox set onto the policy_updated wire literal.
  • ../bridge/events.md — the bridge events surface that collapses the closed seven-member bridge outbox set onto the bridge_config_updated wire literal and fans it out one envelope per hosted Node.
  • ../../architecture/storage-topology.md — the Postgres node that backs plexsphere.sse_relay_cursor and the per-Domain outbox rows the relay drains.
  • ../../contributing/layout.md — bounded-context map row enumerating internal/mesh/sse and the depguard allow-list that keeps the bus boundary intact.
  • ../../how-to/mesh/inspect-the-event-bus.md — operator how-to for inspecting the live JetStream stream, the cursor row, and the publisher metrics.
  • ../../../api/openapi/plexsphere-v1.yaml — OpenAPI spec for GET /v1/nodes/{id}/events, including the Last-Event-ID header and the registered Problem-Details codes.
  • ../../../tests/workspace/mesh_sse_depguard_test.go — workspace alignment test pinning the depguard allow-list for the internal/mesh/sse boundary.

Ubiquitous language ​

The terms below travel together across the Go code, the OpenAPI contract, the audit log, the Prometheus metric label values, and operator-facing tooling. Names are preserved verbatim in error messages and structured-log attributes.

TermDefinition
EnvelopeThe signed-event value type (internal/signing/envelope) carried over the bus: {ID, Type, Scope, KeyID, IssuedAt, Payload, Signature}. The Signature is an Ed25519 detached signature over CanonicalBytes of the envelope with Signature excluded.
Domain EventThe business event a payload encodes (e.g. NodeRegistered from internal/identity/tenancy/events (node.go)). The bus does not interpret payloads; the EventType discriminator on the outbox row is reserved for forensic audit emission while the wire envelope's Type is fixed to node_state_updated.
Subscriber StreamA single authenticated plexd Node's SSE subscription on GET /v1/nodes/{id}/events. The subscribe is dual-credential: a plexd Node authenticates with its own NSK (an equality gate — the authenticated Node must equal the addressed {id} — with no ReBAC check; a cross-Node NSK is refused with 403 node_id_mismatch), or an operator authenticates with a session / API token / OIDC credential carrying the node-agent relation. An NSK-authenticated stream re-checks its credential on every keep-alive tick and closes once the NSK stops resolving, so revoking a Node's NSK tears down the streams it already holds open; an operator stream carries no NSK and skips the re-check. One stream per node; the HTTP handler increments sse_active_streams on Subscribe and decrements on the deferred close.
Signing Key ResolverThe SigningKeyResolver seam in internal/mesh/sse/signing_key_resolver.go that returns the (KeyID, Scope) for a Domain. Returns ErrSigningKeyNotProvisioned when the per-Domain key has not been minted yet; the relay logs and skips on this sentinel.
Nonce SetThe bounded LRU + TTL set in internal/mesh/sse/nonce.go that admits each envelope ID exactly once per freshness window. Add(nonce) → bool: true admits, false is a duplicate the consumer must drop.
Last-Event-IDThe Last-Event-ID HTTP header on a reconnect. Resolved by ParseLastEventID in internal/mesh/sse/replay_math.go to the JetStream OptStartSeq the consumer should resume from.
Replay WindowThe JetStream stream's [currentLow, currentHigh] retention bounds (24h MaxAge by default). A Last-Event-ID below currentLow has aged out; the handler returns 410 Gone with last_event_id_outside_replay_window.
Outbox CursorThe per-stream row in plexsphere.sse_relay_cursor advanced under SELECT … FOR UPDATE SKIP LOCKED by the relay. Carries (stream_name, last_outbox_id, last_xid, updated_at).
Stream Replica FactorThe JetStream stream replicas count read from PLEXSPHERE_SSE_STREAM_REPLICAS at boot. Composition refuses to start with replicas > 1 while the in-memory NonceStore is wired.
NodeRegisteredA representative Domain Event (the Tenancy bounded context emits it when a node joins). Reaches the bus as an outbox row whose payload the relay forwards into a single envelope.
NodeReachabilityChangedA Domain Event emitted by the per-Domain reachability evaluator (internal/mesh/reachability/evaluator.go) when a Node transitions between healthy, stale, and unreachable. Source: TypeNodeReachabilityChanged in internal/identity/tenancy/events/node.go. Wire-level Type discriminator is node_state_updated (the bus pins one wire type — same as every other Domain Event row). Payload contract: {from_state, to_state, node_id, domain_id, occurred_at}. See ./reachability.md for the state-machine semantics.
rotate_keysA peer-aggregate outbox event_type the Key and Peer Manager appends when an operator requests a Node mesh-key rotation. Unlike the past-tense peer notifications it is an imperative COMMAND addressed to a single Node, telling it to generate a fresh Curve25519 keypair and call POST /v1/keys/rotate. Source: TypeRotateKeys in internal/mesh/peers/events/events.go. Wire-level Type discriminator is node_state_updated (the bus pins one wire type — the per-event discriminator travels on the outbox event_type). See ./key-rotation.md for the rotation workflow.
peer_key_rotatedA peer-aggregate outbox event_type the Key and Peer Manager appends when a Node completes a mesh-key rotation; every peer of the rotated Node observes it. The payload carries the rotated Node's new 32-byte Curve25519 public key plus the (kid, wrap_key_version) PSK reference — never PSK plaintext or ciphertext. Source: TypePeerKeyRotated in internal/mesh/peers/events/events.go. Wire-level Type discriminator is node_state_updated. See ./key-rotation.md for the rotation workflow.
signing_key_rotatedA signing-aggregate outbox event_type the plexsphere-signer binary appends when a signing-key rotation opens. One outbox row per Node in the scope is written by OutboxEventPublisher.PublishKeyRotated in cmd/plexsphere-signer/event_publisher_outbox.go inside a dedicated publisher-owned pgx.Tx (NOT the rotation transaction — the rotation row commits first; the publisher's tx is independent). The per-Node rows are all-or-nothing within the publisher tx, but a crash between the rotation commit and the publisher tx leaves the rotation persisted with no fan-out (see signing-rotation.md two-phase trade-off). Source: TypeSigningKeyRotated in internal/signing/events/events.go. Payload contract: {event_id, occurred_at, domain_id, scope_kind, scope_id, old_key_id, new_key_id, new_public_key, valid_from, opened_at, closes_at} (the 32-byte new_public_key is base64-encoded on the wire). The row maps onto no member of the node-facing wire-type set, so the relay skips-and-advances it instead of publishing it to a Node. The consumer side is DomainSigningKeyResolver.Subscribe in internal/mesh/sse/signing_key_resolver.go, which calls Reset(DomainID) per event to invalidate the per-Domain public-key cache. See ../signing-rotation.md for the rotation workflow.
node_state_updatedThe default wire-level Type discriminator the publisher stamps onto every envelope that does NOT map onto a more specific wire literal. Pinned by EventTypeNodeStateUpdated in internal/mesh/sse/publisher.go; the per-row mapping is owned by the wireTypeFor dispatch table in the same file.
policy_updatedA second live wire-level Type discriminator the publisher stamps onto envelopes whose outbox-side event_type belongs to the closed policy aggregate set (policy_revision_created or policy_deleted). Pinned by EventTypePolicyUpdated in internal/mesh/sse/publisher.go; the publisher-side dispatch table and the producer-side fan-out are documented in ../policy/events.md.
bridge_config_updatedA third live wire-level Type discriminator the publisher stamps onto envelopes whose outbox-side event_type belongs to the closed seven-member bridge aggregate set (bridge.RelayConfigured, the user-access / public-ingress / site-to-site *Configured and *Removed literals). Pinned by EventTypeBridgeConfigUpdated in internal/mesh/sse/publisher.go; the producer-side per-Node fan-out is wired while the consumer-side stays deferred. The publisher-side dispatch table, the fan-out algorithm, and the BridgeConfigUpdatedPayload schema are documented in ../bridge/events.md.
PLEXSPHERE_NODE_EVENTSThe JetStream stream name. Pinned by StreamName in internal/mesh/sse/streamname.go; a single source of truth for the publisher's EnsureStream call and the consumer's Replay call.
plexsphere.node.events.<domain>.<node>The per-node JetStream subject template returned by SubjectFor(domainID, nodeID). The wire-format invariants (no ., *, >, no whitespace, no non-printables) are enforced by validateSubjectToken in the same file.

Envelope schema ​

The wire envelope is the envelope.Envelope value type defined in internal/signing/envelope/canonical.go. Field semantics:

FieldTypeMeaning
IDuuid.UUID (v7)Unique per envelope. The publisher allocates a fresh UUIDv7 per call so IDs sort naturally by issuance time across replicas without a separate sequence column. The ID also doubles as the nonce key on the consumer side (see Nonce set design).
TypestringThe wire-level discriminator. Always node_state_updated, pinned by the constant EventTypeNodeStateUpdated in internal/mesh/sse/publisher.go.
Scopesigning.ScopeThe signing scope the envelope was signed under — domain:<uuid> for every per-Domain envelope on this bus.
KeyIDsigning.KeyIDThe key that produced Signature. The Resolver returns this verbatim; the publisher never lets the signer pick its own active key.
IssuedAttime.Time (UTC)The publisher's wall-clock instant at envelope construction. Canonicalised to nine fractional digits via CanonicalTimeLayout.
Payloadjson.RawMessageThe Domain Event payload, taken verbatim from the outbox row. CanonicalBytes re-parses and re-emits the payload with sorted keys so a caller cannot leak its own key order into the signed pre-image.
Signature[]byteEd25519 detached signature over CanonicalBytes(env) with Signature excluded. Excluded by the canonicaliser unconditionally — a verifier re-runs CanonicalBytes on the received envelope without clearing the field manually.

The README's wire-level description of the envelope at README §Signed Event Bus lists a notional separate nonce field; the in-tree implementation uses the envelope ID (UUIDv7) as the nonce key on both publish and consume paths because the ID is already unique-per-envelope within the freshness window. A future dedicated Nonce field would slot into this same call site without touching the security control flow — see the DECISION block in internal/transport/http/v1/handlers/events.go around the nonceKey := ev.Envelope.ID.String() site.

CanonicalBytes is the byte-stable JSON pre-image the signer signs. Encoding rules — sorted object keys, no whitespace, HTML-safe escapes disabled, raw UTF-8, fixed-precision time layout — are pinned in the canonicaliser file's package doc; downstream re-canonicalisation by a verifier is therefore deterministic across Go versions, platforms, and clock resolutions.

JetStream subject layout ​

Two strings pin the wire contract between the publisher's EnsureStream call and the consumer's Replay call:

  • Stream name — the constant StreamName = "PLEXSPHERE_NODE_EVENTS" in internal/mesh/sse/streamname.go. This is the SINGLE SOURCE OF TRUTH; both the relay's EnsureStream and the SSE handler's Replay source the literal from this file.
  • Per-node subject template — plexsphere.node.events.<domainID>.<nodeID>, returned by SubjectFor(domainID, nodeID). The fixed prefix subjectPrefix = "plexsphere.node.events." is unexported; callers always go through SubjectFor.

validateSubjectToken (same file) rejects each token that:

  • is empty (domain id must not be empty),
  • contains the NATS subject delimiter . (collides with the hierarchy),
  • contains the wildcard * or the multi-wildcard > (would expand the publish target),
  • contains whitespace (caught via unicode.IsSpace so tab, LF, CR, and U+00A0 all surface a single error class),
  • contains a non-printable byte (caught via unicode.IsPrint so the C0/C1 control planes and DEL all hit one branch).

A rejected token surfaces ErrInvalidSubjectToken wrapped with the offending kind and value; callers branch with errors.Is. The publisher reacts by skipping the row before any bus call.

Publish pipeline diagram ​

The publish pipeline runs strictly in the order shown — the "sign before publish" invariant forbids storing or transmitting an unsigned envelope, so any failure in the resolve, canonicalise, or sign step short-circuits before the JetStream publish.

text
   ┌────────────────────────────────────────────────────────────┐
   │   Postgres   plexsphere.outbox_event                       │
   │              + plexsphere.sse_relay_cursor (per-stream)    │
   └────┬───────────────────────────────────────────────────────┘
        │ 1. RunTick: SELECT … FOR UPDATE SKIP LOCKED on cursor
        │            ReadOutboxBatchAfter(cursor, batch=64)
        v
   ┌────────────────────┐
   │  Relay (relay.go)  │  for each row:
   └────┬───────────────┘
        │ 2. resolve(domainID) ──────────────► SignerClient.PublicKey
        │                                       (gRPC mTLS)
        │      KeyID, Scope (or ErrSigningKeyNotProvisioned: skip)
        v
   ┌────────────────────────────┐
   │  Publisher.Publish         │
   │  (publisher.go)            │
   └────┬───────────────────────┘
        │ 3. allocate envelope ID = uuid.NewV7()
        │ 4. build envelope { ID, Type=node_state_updated,
        │                     Scope, KeyID, IssuedAt, Payload }
        │ 5. canonicalBytes = envelope.CanonicalBytes(env)
        │ 6. signResult = Signer.Sign(scope, keyID, canonicalBytes)
        │                              ──────► SignerClient.Sign
        │                                       (gRPC mTLS)
        │ 7. env.Signature = signResult.Signature
        │ 8. wireBody = json.Marshal(env)
        │ 9. subject = SubjectFor(domainID, nodeID)
        │       ────────────────► JetStream subject
        │                          plexsphere.node.events.<dom>.<node>
        v
   ┌────────────────────────────┐
   │  Bus.Publish(subject,      │
   │              wireBody,     │
   │              WithMsgID(ID))│ ──► JetStream stream
   └────┬───────────────────────┘     PLEXSPHERE_NODE_EVENTS
        │ 10. ack.Sequence (uint64)
        v
   ┌────────────────────────────┐
   │  Relay (relay.go)          │ 11. AdvanceSseRelayCursor on
   │                            │     successful suffix; on any
   │                            │     publish/sign error: log,
   │                            │     bump counter, halt batch
   │                            │     without partial advance.
   └────────────────────────────┘

Failure modes are typed sentinels. ErrSigningKeyNotProvisioned short-circuits at step 2 — the publisher never reaches the sign or publish call. signing.ErrProviderUnavailable and any other Signer.Sign error short-circuit at step 6. Either way the cursor does not advance; the next Tick re-publishes the failed row, and the JetStream Nats-Msg-Id dedup window (keyed on the envelope ID) suppresses any duplicate that might briefly land on the wire.

Last-Event-ID semantics ​

ParseLastEventID(header, currentLow, currentHigh) in internal/mesh/sse/replay_math.go resolves the SSE handler's Last-Event-ID header into a JetStream OptStartSeq. The four arms are non-overlapping and exhaustive:

ArmHeaderBoundsReturned startSeqReturned errorHTTP status
1emptyanycurrentHigh + 1nil200 (tail mode; consumer blocks until next publish)
2numeric ncurrentLow ≤ nn + 1nil200 (resume; folds tail-position case n > currentHigh into the resume arm so a future-sequence reconnect blocks at n+1)
3numeric nn < currentLow0ErrLastEventIDOutsideWindow410 Gone, code last_event_id_outside_replay_window
4malformedany0ErrBadLastEventID400 Bad Request

Both sentinels live in replay_math.go next to their producer:

  • ErrBadLastEventID — wraps strconv.ParseUint's error so the HTTP handler can surface a stable error code while the operator-side log carries the underlying parse detail. Trips on alphabetic input, signed integers (-7, +42), decimal/scientific notation (12.5, 1e10), hex literals, surrounding whitespace, and any base-10 integer that overflows uint64.
  • ErrLastEventIDOutsideWindow — surfaced when the parsed n falls below currentLow. The client falls back to the reconciliation pull at GET /v1/nodes/{id}/state to rebuild state without lossy replay.

Edge cases the function pins explicitly:

  • n == 0 against currentLow ≥ 1 is arm 3 (410). JetStream sequences are 1-based; a 0 bookmark cannot come from a real delivery.
  • currentLow == 0 (only when the stream is empty) with an empty header returns startSeq = 1 — the consumer blocks until the first publish.
  • A future-sequence header (n > currentHigh) with currentLow ≤ n is arm 2 — the consumer blocks at n+1 until a matching new event is published. The user-story criterion is "Reconnect with Last-Event-ID: <head+1> blocks until a matching new event is published"; the resume arm is the correct home for it.

Nonce set design ​

The NonceStore in internal/mesh/sse/nonce.go is an in-memory, bounded LRU + TTL set that admits each envelope ID exactly once per freshness window. Its design properties:

  • Insertion-time ordering, not access-time. The freshness window is a property of WHEN the nonce was first seen; an access-time LRU would let a steady stream of Seen() probes keep an aging nonce alive past TTL.
  • Capacity bound. Construction takes maxSize >= 1; Add evicts the oldest insertion-time entry when the post-expiry size is at the cap. A non-positive maxSize produces ErrInvalidNonceStoreConfig at constructor time.
  • TTL bound. Construction takes ttl >= 1ns; Add runs expire-on-touch before any duplicate-detection branch, so a TTL-aged nonce re-Add returns true.
  • Concurrency. All exported methods are safe for concurrent use across goroutines under a single sync.Mutex.
  • Container choice. Backed by container/list (doubly-linked list) plus a map[string]*list.Element for O(1) Add, O(1) Seen, and O(1) capacity eviction.
  • Prometheus gauge. sse_nonce_live_size (the constant NonceLiveSizeMetricName in internal/mesh/sse/metrics.go) is registered by New via prometheus.NewGaugeFunc. The callback reads len(map) under the store's mutex at scrape time only, so the steady-state hot path never touches Prometheus state. A nil registerer disables the gauge.

The in-memory store is single-replica-only. The composition root in cmd/plexsphere/sse_factory_prod.go refuses to boot when PLEXSPHERE_SSE_STREAM_REPLICAS > 1 is set with the in-memory NonceStore wired.

Operator knobs:

Env varDefaultEffect
PLEXSPHERE_SSE_NONCE_TTLfeature defaultFreshness window for the nonce set. Parsed as a Go time.Duration; non-positive values are rejected at boot.
PLEXSPHERE_SSE_MAX_LIVE_NONCESfeature defaultHard upper bound on live nonces. Parsed as a positive integer; non-positive values are rejected at boot.

Threat model ​

The bus mitigates four classes of attack along the publish-to-consume path. Each mitigation is implemented in a single, named place so a reader chasing a security claim does not have to assemble it from multiple files.

  1. Signature forgery. Every envelope is signed with the per-Domain Ed25519 key resolved via the SigningKeyResolver. The consumer re-runs CanonicalBytes over the received envelope and verifies the detached signature against the per-Domain public key served by SignerClient.PublicKey. A failed verification is logged with the outcome=signature_failure audit attribute and bumps sse_envelope_verification_failures_total{outcome=bad_signature}. See EventStreamAdapter in internal/mesh/sse/event_stream.go for the audit emission path and the AuditOutcomeSignatureFailure constant.
  2. Replay of a previously-valid envelope. The symmetric nonce set rejects a duplicate envelope ID inside the freshness window; the publisher's IssuedAt field anchors freshness on the signed pre-image so an attacker cannot mint a far-future timestamp under a captured signature. A duplicate increments sse_nonce_replay_total and is dropped with a structured-log line. Both sides (publish-side relay and consume-side handler) share one NonceStore at composition time so the counter aggregates the symmetric check.
  3. Tampering in transit. JetStream guarantees the wire bytes reach the consumer unmodified at the broker level; the end-to-end Ed25519 signature over CanonicalBytes makes any in-transit byte mutation observable at verification time regardless of the broker's transport. A mutated payload fails re-canonicalisation or signature verification before the consumer accepts the envelope.
  4. Signer outage. The publisher refuses to write unsigned envelopes: ErrSigningKeyNotProvisioned short-circuits the pipeline at the resolver step, BEFORE any bus call. The relay logs a structured WARN line with the Domain ID and bumps sse_relay_skip_total{reason=no_signing_key}. A generic signer error (e.g. signing.ErrProviderUnavailable) short-circuits at the sign step and bumps sse_publisher_sign_failures_total{reason=publish_error}. In either case the cursor does not advance; the next Tick replays the row once the signer recovers.

Multi-replica handoff ​

The relay supports horizontal scale via the SELECT … FOR UPDATE SKIP LOCKED cursor pattern in internal/mesh/sse/relay.go. Each Tick:

  1. Begins a Postgres transaction.
  2. Locks the per-stream cursor row (stream_name = 'PLEXSPHERE_NODE_EVENTS') in plexsphere.sse_relay_cursor under FOR UPDATE SKIP LOCKED. If another replica already owns the row, the query yields zero rows and the relay returns ErrCursorBusy — translated to a nil Tick error so the relay-loop driver does not back off on a healthy contention pattern.
  3. Reads up to batchSize outbox rows strictly after the cursor's (transaction_id, occurred_at, outbox_id) triple.
  4. Calls the publish pipeline for each row; on the longest contiguous success suffix, advances the cursor inside the same transaction.
  5. Commits.

Failure suppresses the cursor advance for any row that follows the last success, so the next Tick (on this or a sibling replica) starts exactly where this one halted. Combined with the JetStream Nats-Msg-Id dedup window (keyed on the envelope ID), the at-least-once boundary is invisible to the SSE consumer: a row that two replicas briefly attempt to publish lands on JetStream once.

A reconnecting subscriber lands on a different replica without gap or duplicate because:

  • The replay payload is owned by JetStream, not by the relay. The consumer's Replay(streamName, startSeq) call (with startSeq resolved by ParseLastEventID) reads the bus directly; the receiving plexsphere replica is a transparent passthrough.
  • The cursor row's at-least-once boundary applies only to the publish side. A reconnect's Last-Event-ID does not touch the cursor; it touches the JetStream stream's (currentLow, currentHigh) view.

The in-memory NonceStore is single-replica-only: each replica maintains its own bounded set, so a duplicate publish that two replicas race on cannot be defended against in-process across replicas. The composition root therefore refuses to boot when PLEXSPHERE_SSE_STREAM_REPLICAS > 1 is set with the in-memory store wired — operators who scale beyond one replica must wire a distributed nonce backend before scaling. The exact refusal message is pinned in cmd/plexsphere/sse_factory_prod.go and rejects the boot with a "PLEXSPHERE_SSE_STREAM_REPLICAS > 1 is incompatible with the in-memory NonceStore" error directing operators to pin replicas=1 or wire a distributed nonce store before scaling.

Operator runbook ​

For full step-by-step procedures see ../../how-to/mesh/inspect-the-event-bus.md. The pointers below are the entry points an operator chases when a subscriber complains about gaps, replays, or a stalled reconnect.

  • Inspect the JetStream stream. Use the NATS CLI inside the cluster — nats stream info PLEXSPHERE_NODE_EVENTS reports the current (first, last, age, max_age) triple. The stream name comes from the StreamName constant in internal/mesh/sse/streamname.go; per-node subjects follow plexsphere.node.events.<domainID>.<nodeID> and are listable via nats stream subjects PLEXSPHERE_NODE_EVENTS.
  • Drain the cursor row. The per-stream cursor lives in plexsphere.sse_relay_cursor. Operators inspect the row with SELECT stream_name, last_outbox_id, last_xid, updated_at FROM plexsphere.sse_relay_cursor WHERE stream_name = 'PLEXSPHERE_NODE_EVENTS';. The updated_at column is the last successful Tick instant; a stale value combined with a non-empty outbox is the signal that a publish error is halting batches.
  • Force a re-emit. Reset the cursor by deleting the cursor advance — UPDATE plexsphere.sse_relay_cursor SET last_outbox_id = '<earlier-id>', last_xid = <earlier-xid> WHERE stream_name = 'PLEXSPHERE_NODE_EVENTS';. The next Tick re-publishes everything after the new cursor; the JetStream Nats-Msg-Id dedup window (24h) suppresses any duplicate that the original publish already landed.
  • Watch the Prometheus surface.
    • sse_envelope_verification_failures_total{outcome} — consumer-side rejections labelled bad_signature, bad_nonce, or decode_error. A sustained rate is a malformed publisher, a key-rotation gap, or a replay attack.
    • sse_nonce_replay_total — NonceStore.Add returned false. A spike correlates with a replay storm; a slow drift correlates with a clock skew letting old envelopes fall back into the freshness window.
    • sse_relay_skip_total{reason} — relay rows skipped, labelled no_signing_key (rotation pipeline has not minted the per-Domain key yet).
    • sse_publisher_sign_failures_total{reason} — publish-pipeline errors labelled publish_error (today the relay's only label), with provider_unavailable and key_not_provisioned reserved for future direct-publisher call sites.
    • sse_active_streams — count of currently-open SSE handler subscriptions. Watch against per-Node connection caps to detect leaked subscriptions or a runaway reconnect loop.
    • sse_nonce_live_size — current nonce-set size. Watch against PLEXSPHERE_SSE_MAX_LIVE_NONCES to detect approaching the bounded-set limit before Add latency degrades.