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 shared 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 whose load-bearing ports are all nil. 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 cursor store in internal/mesh/sse/repo, the Postgres nonce store, the signing-key rotation watcher, and the consumer-side EventStreamAdapter. It then populates EventStream, SignatureVerifier, Publisher, and the three run hooks: the relay loop, the rotation watcher, and the nonce sweep. With the full security cone (EventStream, 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 four 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 handling runs ahead of both skip-and-advance gates, so the replica holding the tick calls Reset(DomainID) on the resolver the moment it reads a signing_key_rotated row, even though that row is never published to a Node. The reset is inline rather than deferred to the replica's own watcher, so a later row of the same batch cannot be signed with the key that just retired. Gating the invalidation on the allowed set would leave a retired key trusted until the process restarts.

Running on several replicas ​

The slice serves any number of API replicas. Four things make that hold:

  • The relay's resume position. plexsphere.sse_relay_cursor carries the whole (last_xid, last_occurred_at, last_outbox_id) triple the outbox read seeks on, so whichever replica wins the next tick continues strictly after the last published row. The relay runs on every replica; the FOR UPDATE SKIP LOCKED lock on the cursor row is what serialises the ticks, which is why it is not a leader-elected loop.
  • The envelope nonce set. plexsphere.sse_nonces is the set every replica admits against, through one gate, the EventStreamAdapter. A reconnect that lands on a sibling replica is refused the nonce the first delivery already spent.
  • The signing-key rotation signal. A watcher on every replica pages the signing_key_rotated outbox rows at the relay's cadence and drops the cached key of each rotated Domain, so a key opened through another replica's signer is trusted here too. Between ticks the verify path does not depend on the watcher at all: it resolves the key the envelope names rather than the key this replica believes is active, which is what keeps a replica that is still signing under the outgoing half of an open rotation readable by one that has already moved on. The publisher refreshes once on ErrKeyRetired, on the exact failure a stale cache produces.
  • The nonce hygiene pass. sse-nonce-sweep runs on the background leader and deletes, in bounded passes, the rows older than PLEXSPHERE_SSE_NONCE_TTL. It carries no readiness probe: the admit measures seen_at against the window on every call, so an unswept row costs storage, not correctness. Both ends of that window come from the database's now(), which is the only clock every replica shares.

PLEXSPHERE_SSE_STREAM_REPLICAS is the JetStream stream's replication factor and nothing else. It defaults to 1 and has no bearing on how many API replicas may serve the bus.

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 leaves every load-bearing port nil and keeps the 501 stub; a non-empty NATS URL dials JetStream and assembles the full publish → relay → consume graph.
PLEXSPHERE_SSE_STREAM_REPLICAS1The JetStream stream's replication factor, passed to EnsureNodeEventsStream. It sizes the stream's RAFT group, not the API Deployment.
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 knob PLEXSPHERE_SSE_NONCE_TTL is documented under Nonce set design. It sets both the freshness window an admitted nonce stays remembered for and the cadence the sweep runs at, so the table holds roughly one window of rows.

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.
  • 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 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 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 API Deployment in that fixture runs two Pods, and three steps assert the bus across them. sse-two-replicas-deliver-once commits three dispatches in ONE database transaction and reads each Pod's IP directly: every dispatch arrives exactly once, on the Pod that did not relay it. sse-nonce-shared-across-replicas resumes with a Last-Event-ID against the sibling Pod and proves the frame is dropped there, because the nonce the first read spent is a row both Pods see. sse-rotation-reaches-both-replicas opens a signing-key rotation through plexctl mid-run and proves the replica that never saw the rotate call signs its next envelope with the incoming key. 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 Postgres nonce set with its TTL window and leader-elected sweep, 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 and plexsphere.sse_nonces migrations, the depguard allow-list, the OpenAPI surface byte-equality drift gate, the in-process EventStreamAdapter, the per-replica signing-key rotation watcher, and the consumer-side capture harness contract. The relay loop, stream provisioning, and the RelayStore in internal/mesh/sse/repo 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 Postgres table plexsphere.sse_nonces, admitted through NonceStore in internal/mesh/sse/repo/nonce_store.go, which lets each envelope ID through exactly once per freshness window. Add(ctx, nonce) → (bool, error): true admits, false is a duplicate the consumer must drop, and an error means the set could not answer, which the consumer treats as a refusal. Every replica admits against the same table.
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, last_occurred_at, updated_at). The three position columns are the whole (transaction_id, occurred_at, id) triple the outbox read seeks strictly after, so the resume position outlives the replica that reached it.
Stream Replica FactorThe JetStream stream's RAFT replication factor, read from PLEXSPHERE_SSE_STREAM_REPLICAS at boot and passed to EnsureNodeEventsStream. It sizes the stream's replica group inside NATS and says nothing about how many API replicas serve the bus.
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 the SigningKeyRotationWatcher in internal/mesh/sse/signing_key_rotation_watcher.go, one per API replica, which pages these rows on the relay's cadence and calls Reset(DomainID) once per rotated Domain to invalidate the per-Domain public-key cache. Its paging position is held in memory and seeks to the tail of the rotation history when the store is built, so a restarted replica watches for rotations opened from that point rather than re-walking rows that name Domains it has cached no key for; the partial index outbox_events_signing_key_rotated_idx serves both that seek and the forward pages. The replica holding the relay tick also resets inline as it reads the row. 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 nonceKey := env.ID.String() site in internal/mesh/sse/event_stream.go, which is the one place the bus admits a nonce.

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 nonce set is the Postgres table plexsphere.sse_nonces, one row per admitted envelope nonce:

ColumnTypeMeaning
noncetext, primary keyThe envelope ID. The primary key is what makes the admission a single statement: the insert IS the admission and a duplicate IS the conflict.
seen_attimestamptzThe instant of the admission. The freshness window is measured against it, and sse_nonces_seen_at_idx serves the sweep that reads it.

NonceStore in internal/mesh/sse/repo/nonce_store.go admits each nonce with one statement:

sql
INSERT INTO plexsphere.sse_nonces (nonce, seen_at)
VALUES ($1, now())
ON CONFLICT (nonce) DO UPDATE
SET seen_at = EXCLUDED.seen_at
WHERE sse_nonces.seen_at < now() - make_interval(secs => $2)
RETURNING nonce;

A returned row is an admission. No row means the nonce was already seen inside the freshness window, which the caller reads off pgx.ErrNoRows without a second statement. The WHERE arm is what re-admits a nonce whose seen_at has aged past the window: an envelope carrying it is refused for its age by the freshness gate before its nonce is ever looked up, so remembering it longer buys nothing.

Both ends of the window come from now(), the database's clock, and the caller supplies only the window's length. The set is shared, so a window each replica measured against its own wall clock would not be one window: a replica running more than the TTL behind would write a seen_at every sibling already reads as stale and re-admit the duplicates the set exists to refuse, and a sweep on a replica running that far ahead would compute a cutoff later than every real row and empty the table on each pass. Neither failure announces itself.

Design properties:

  • One gate, one owner. The admit runs at exactly one layer, the EventStreamAdapter. It owns the audit record for a rejection and the sse_nonce_replay_total increment, so a reader chasing the replay claim finds it in one file.
  • Shared across replicas. Every replica admits against the same table, so a reconnect that lands elsewhere is refused the nonce the first delivery spent.
  • Fail closed. An admit the store cannot answer is returned to the caller as an error, never folded into an admission. A set that cannot answer says nothing about whether the envelope in hand is a replay, and admitting on error would turn a Postgres outage into an open replay window.
  • TTL bound, swept on the leader. NewNonceStore rejects a non-positive window with ErrInvalidNonceTTL. The sse-nonce-sweep loop deletes the rows older than the window on the background leader at the window's own cadence, so the table holds roughly one window of rows. The sweep carries no readiness probe, because an unswept row costs storage and not correctness.
  • Swept in bounded passes. Each DELETE carries a LIMIT, and the pass repeats until one comes back short of it or a per-tick cap is reached. The table gains a row per envelope delivered across the whole fleet, so an unbounded DELETE over a backlog would eventually exceed the worker pool's statement timeout — and every pass after that would face a larger set and fail sooner, while the table grew without a limit. A bounded pass always completes.

There is no gauge for the set's size: the sweep is what bounds the table, so there is no capacity a gauge could be read against. An operator who wants the live count reads it directly:

sql
SELECT count(*) FROM plexsphere.sse_nonces
 WHERE seen_at > now() - interval '5 minutes';

Substitute the deployment's PLEXSPHERE_SSE_NONCE_TTL for the interval. The table carries no Domain column, so the count is fleet-wide rather than per Domain.

Operator knob:

Env varDefaultEffect
PLEXSPHERE_SSE_NONCE_TTLfeature defaultFreshness window an admitted nonce stays remembered for, the age past which the consumer refuses an envelope outright, and the cadence sse-nonce-sweep runs at. Parsed as a Go time.Duration; non-positive values are rejected at boot. See Freshness window and replay horizon — it is also the furthest back a Last-Event-ID resume can reach.

Freshness window and replay horizon ​

One duration does three jobs, and they are the same job seen from three sides:

  • an admitted nonce is remembered for it,
  • an envelope older than it is refused with outcome=stale,
  • the sweep reclaims rows past it.

They have to be one duration. The set may forget a nonce only because an envelope carrying it would be refused for its age anyway; widen the retention without widening the refusal and a captured envelope replayed after the sweep is admitted as a first sight and verifies, because the signature is over immutable canonical bytes.

The consequence is the part to plan around: the window is also the horizon a Last-Event-ID resume can reach back over. JetStream retains envelopes for DefaultStreamMaxAge (24h), but a node that reconnects after longer than the window is refused the envelopes published while it was away, exactly as an attacker replaying them would be — the consumer cannot tell the two apart, which is why the refusal exists.

That is workable because the bus is not the authoritative channel for mesh state: GET /v1/nodes/{id}/state is, and a node that has been away long enough to fall outside the window reconciles there. A deployment that wants a longer catch-up horizon raises PLEXSPHERE_SSE_NONCE_TTL, and pays for it in nonce rows — the table then holds one window of the whole fleet's delivery rate.

Nothing pushes the node to that pull, and this is the sharp edge to plan around. The truncated resume still answers 200 OK; the refused envelopes are dropped one by one behind an otherwise healthy stream, so the node has no way to tell a quiet Domain from a backlog it will never receive, and it reconciles only on its own cadence. The response that would tell it — 410 Gone with last_event_id_outside_replay_window — needs the resume position's age, which means the Bounds() seam tracked in the event-bus roadmap; the sentinel and the client-side fallback already exist and wait on it. Until then the signal is server-side only: the consumer raises one outcome=resume_window_expired error line per affected subscription, naming the node and the resume position, and bumps sse_envelope_verification_failures_total{outcome=stale} per refused envelope.

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 public key SignerClient.PublicKey returns for the key the envelope NAMES in its KeyID field, refusing one the signer reports as retired. KeyID is inside the signed pre-image, so it cannot be altered without invalidating the signature, and it is the only field that says which key the publisher used — resolving whichever key happens to be active on the verifying replica drops a whole relay batch whenever the two ends disagree, which is exactly what a rotation overlap produces. A failed verification is logged with the outcome=signature_failure audit attribute and bumps sse_envelope_verification_failures_total{outcome=bad_signature}. The by-id lookup is memoised — the resolved key AND the refusal — for PinnedKeyCacheTTL, evicted early by the rotation watcher's Reset. Without it a rotation overlap costs one signer RPC, and one KMS provider call behind it, for every envelope on every open subscription; and an envelope naming a key id that does not exist is refused only AFTER the lookup, so an uncached refusal would let anyone able to publish on the node-events subject spend one RPC per message before a signature is ever checked. The TTL is what bounds the memoised verdict, because a rotation is announced when it OPENS and never when it CLOSES. 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. Two gates, one window. The nonce set rejects a duplicate envelope ID inside the freshness window, and the freshness gate rejects any envelope whose IssuedAt is older than that window (or further into the future than the skew two replicas may plausibly disagree by). IssuedAt is inside the signed pre-image, so a captured envelope cannot be re-dated, and the gate is what lets the sweep forget a row: without it, a replay of the same bytes one window later would find no row, be admitted as a first sight, and verify — the signature is over immutable canonical bytes, so re-publishing an hours-old dispatch would be enough. A duplicate increments sse_nonce_replay_total and is dropped with a structured-log line; a stale envelope is dropped with outcome=stale. Both checks run at exactly one layer, the EventStreamAdapter, and the set behind the first is the shared table plexsphere.sse_nonces, so a replay is refused whichever replica served the first delivery and whichever serves the second. The admit runs LAST, after the addressing, freshness and signature gates, so an envelope that fails any of them never writes to the shared set — an injected message carrying a captured envelope ID cannot burn that ID's one admission and suppress the genuine envelope. A set that cannot answer is retried a bounded number of times and then refuses the envelope and closes the stream rather than admitting it, so an outage does not open the window.
  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. All three members are columns on the cursor row, so the position the read seeks after is the one the last successful advance persisted, not one the replica reconstructed from memory.
  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 nonce set is shared, so the consume side holds across replicas the same way. Two replicas racing the same nonce serialise on the row, and PostgreSQL evaluates the conflict arm against the latest committed version, so exactly one of them sees the admission and the other reads a duplicate. A subscriber that reconnects to a sibling replica with a Last-Event-ID covering envelopes it already received is therefore refused those envelopes rather than handed them twice.

The signing-key rotation signal is shared through the outbox rather than through memory. Each replica runs its own SigningKeyRotationWatcher over the signing_key_rotated rows, so a key one replica's signer opened is trusted on every replica within a tick, and neither the relay nor the handler depends on having held the tick that read the row.

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, last_occurred_at, 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. The other three columns are the resume position, and any replica may have written them.
  • 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>, last_occurred_at = '<earlier-occurred-at>' WHERE stream_name = 'PLEXSPHERE_NODE_EVENTS';. Set all three position columns together: the read seeks on the triple, so an update that moves only two of them lands the relay somewhere neither the operator nor the previous cursor named. 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, decode_error, or stale. A sustained rate is a malformed publisher, a key-rotation gap, or a replay attack. outcome=stale is the freshness gate and has two quite different causes: spread across replicas it is a node resuming from outside the window (see Freshness window and replay horizon); concentrated on ONE replica that refuses everything from every publisher it is that replica's clock, which the gate reads locally while the nonce window it is paired with reads the database clock.
    • sse_nonce_replay_total — the admit against plexsphere.sse_nonces reported a duplicate. 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_nonce_sweep_behind_total — one sse-nonce-sweep tick spent every pass it is allowed and the last one still came back full, so more rows expired than the tick can reclaim. A sustained rate means the table is growing without bound; raise the reclaim ceiling or lower PLEXSPHERE_SSE_NONCE_TTL. This counter is the table's only growth signal — there is deliberately no size gauge.
    • 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.