feat(fabric): add the RoutingFabric seam for binding-cache invalidation (RIG-3108)

go/internal/fabric/SUBJECTS.md

@@ -1,6 +1,6 @@
 # Fabric subjects and JetStream configuration
 
-The written spec for the NATS eventing substrate: the four subject grammars, the
+The written spec for the NATS eventing substrate: the five subject grammars, the
 dead-letter subject, and the JetStream stream/consumer configuration. Frozen by
 `docs/designs/infra/runtime/compass-managed-multitenancy/design.md` §T3/§Q3;
 this file is the operational restatement that later tasks build against, and
@@ -15,6 +15,8 @@ this file is the operational restatement that later tasks build against, and
 | `compass.runner.<runner_id>.cmd` | core NATS | `RunnerCommandSubject(runnerID)` | Server → one Runner (async command push) |
 | `compass.runner.events` | core NATS, queue group `compass-runner-events` | `RunnerEventsSubject()` | Runners → exactly one Server (event fan-in) |
 | `client.<sessionID>` | core NATS | `ClientSubject(sessionID)` | Server → one live client connection |
+| `compass.routing.binding.<tenant>` | core NATS | `RoutingBindingSubject(tenant)` | Server → ALL Servers (binding-cache invalidation, no queue group) |
+| `compass.routing.binding.*` | core NATS | `RoutingBindingWildcardSubject()` | Servers → every Server (cross-tenant invalidation, **subscribe-side only**) |
 | `compass.dlq.comms` | core NATS | `DLQSubject` | fabric → operator (parked events) |
 
 `client.<sessionID>` sits outside the `compass.` root deliberately — the frozen
@@ -52,6 +54,68 @@ Postgres insert.
   `CommsSubject`, and `EventRef.valid` rejects a `*` tenant, so a wildcard
   publish is impossible rather than merely discouraged.
 
+### Binding invalidation: `compass.routing.binding.<tenant>`
+
+The routing plane (§T4). The hub's in-memory binding maps become instance-local
+caches over durable truth, and `RoutingFabric` carries the invalidations that
+keep them honest: `PublishBindingChange(ctx, tenant, change)` on the concrete
+per-tenant subject, `SubscribeBindingChanges(ctx, fn)` on the tenant wildcard.
+
+- **Core NATS, and deliberately droppable.** §T4: "Postgres is the arbiter on
+  any cache miss or conflict; core NATS at-most-once suffices because a dropped
+  invalidation degrades to a cache-miss re-read." A stream here would buy
+  durability for a message whose whole content is "go ask Postgres", and would
+  add a second store of state Postgres already owns. So there is no ack, no
+  retry and **no DLQ** on this plane: a malformed payload is logged and dropped,
+  which is the one visible behavioural difference from the comms path.
+- **No queue group — every instance caches.** Each Server holds its own binding
+  cache, so each must receive every invalidation. A queue group would hand each
+  change to exactly one Server and leave the others serving a stale binding,
+  silently: nothing on a core-NATS plane surfaces an undelivered message.
+  Contrast `compass.runner.events`, which queue-groups on purpose — a Runner
+  event is *work*, and work must be done once.
+- **The literal `binding` precedes the tenant token, and that is a correctness
+  requirement.** The comms stream captures `compass.*.comms.*`, which matches
+  any four-token subject whose **third** token is `comms`. Under the other
+  ordering, `compass.routing.<tenant>.binding`, a tenant literally named `comms`
+  yields `compass.routing.comms.binding` — captured by that wildcard. A stream
+  is an ordinary subscriber in the account sublist, so the capture is *additive*
+  and the hubs would still get the message; the harm is that a best-effort
+  core-NATS invalidation would ALSO be persisted into a durable stream specified
+  to hold only `EventRef`s, burning its storage and `MaxAge` budget for traffic
+  no comms consumer can use (each consumer's `FilterSubject` fixes a concrete
+  kind, and no `EventKind` is `binding`). Pinning `binding` to token 3 makes
+  that unrepresentable for *every* tenant value, because this grammar's token 3
+  is never variable.
+- **The subscribe is tenant-wildcard, publish is concrete.** A Server's cache
+  spans every tenant it has resolved a session for, so one subscription is what
+  the cache needs and tenant creation stays a Postgres insert (the same argument
+  `SubscribeKind` makes for the delivery consumer). `PublishBindingChange`
+  derives its subject via `RoutingBindingSubject`, and `BindingChange.valid`
+  rejects a `*` tenant, so a wildcard publish is impossible rather than merely
+  discouraged.
+- **Payload: a reference, never a copy.** `BindingChange` is JSON
+  `{tenant, session_id, op}` — never the resolved instance. Carrying the
+  resolution would let a reordered or duplicated delivery install an older
+  binding over a newer one, and the cache would then need a version to
+  arbitrate. Carrying only the identity makes every delivery idempotent and
+  every *drop* recoverable by re-reading Postgres. `op` exists so an unbind
+  needs no read at all: the correct action is "drop the entry", and querying
+  Postgres to confirm an absence is a read whose answer is already in the
+  message.
+- **`op` is an OPEN set.** `bound` and `unbound` are the values this version
+  publishes, but a receiver must handle any other value by invalidating and
+  re-reading. The receive path carries an unrecognized `op` through rather than
+  dropping it: the binding genuinely changed, and this plane has no ack, no
+  retry and no dead-letter subject, so a drop would leave a stale entry with
+  nothing to reveal it. The publish path stays strict, because a caller minting
+  an unknown `op` is a bug with a stack.
+- **The tenant rides in both the subject and the payload.** The read side is
+  wildcard and its callback receives no subject, so the payload's tenant is the
+  receiver's only scope. `PublishBindingChange` requires the two to be equal, so
+  they cannot disagree on the wire — the same cross-tenant guard `Publish` gets
+  by deriving its subject from the ref.
+
 ### Token validation: reject, never sanitize
 
 NATS reserves `.` (token separator), `*` and `>` (wildcards), and rejects

go/internal/fabric/bindingchange.go

@@ -0,0 +1,145 @@
+package fabric
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+// BindingOp discriminates what happened to a session binding. It exists so a
+// receiver can act on the change WITHOUT a Postgres read in the one case where
+// the correct action needs no row: an unbind can only ever mean "drop the cache
+// entry", and reading Postgres to confirm an absence is a query whose answer is
+// already in the message.
+//
+// A bind still re-reads. The value is deliberately not carried (see
+// BindingChange), so "bound" says only that the binding for this session
+// changed — the receiver invalidates and lets the next resolution re-read.
+//
+// THE SET IS OPEN, and a receiver MUST handle a value that is not one of the
+// constants below. The wire is shared with publishers that may be newer than
+// the process reading it, and the receive path deliberately carries an
+// unrecognized op through rather than dropping the message: an unknown op still
+// means this session's binding genuinely changed, and this plane has no ack, no
+// retry and no dead-letter subject, so a drop would leave the entry stale with
+// nothing to reveal it. Treat any unrecognized op as "invalidate and re-read",
+// which is the conservative action and stays correct for every op this type can
+// grow — a switch with no default is the bug this warning exists to prevent.
+type BindingOp string
+
+// The binding operations known to this version. Snake-case-safe single words:
+// they are JSON values, not subject tokens, but keeping them token-legal means
+// a later grammar that wants the op in the subject does not have to rename
+// them. See BindingOp: a receiver may observe a value outside this set.
+const (
+	// BindingBound means the session now resolves to a (possibly different)
+	// instance. The receiver invalidates its entry; the next resolution reads
+	// the durable truth.
+	BindingBound BindingOp = "bound"
+	// BindingUnbound means the session has no binding any more. The receiver
+	// can drop its entry outright — there is no row to re-read.
+	BindingUnbound BindingOp = "unbound"
+)
+
+// BindingChange is a compact reference to a committed session-binding change —
+// tenant, session id, and which way it went — and NEVER a copy of the binding
+// itself (§Global Constraints: "Postgres is the sole durability source of
+// truth"). A receiver treats it as "your cached binding for this session is
+// stale", exactly as a subscriber treats an EventRef as "re-read this row".
+//
+// That reference discipline is what makes this plane safe to run at-most-once
+// (§T4). If the message carried the resolved instance, a reordered or duplicated
+// delivery could install an older binding over a newer one, and the cache would
+// need a version to arbitrate — a second store of state Postgres already owns.
+// Carrying only the identity means every delivery is idempotent, and a DROPPED
+// delivery degrades to a cache miss that re-reads Postgres, which is the
+// arbiter.
+type BindingChange struct {
+	// Tenant is the owning tenant id, and also the tenant token of the subject
+	// the change rides.
+	//
+	// It is carried in the payload as well as the subject because
+	// SubscribeBindingChanges is a TENANT-WILDCARD subscribe whose callback
+	// receives the decoded change and no subject (RoutingFabric) — so without
+	// this field a receiver watching every tenant could not scope the cache
+	// entry it is being told to drop. Publish requires it to equal the tenant
+	// it publishes under, so the two can never disagree on the wire.
+	Tenant string `json:"tenant"`
+	// SessionID names the session whose binding changed. It is the cache key
+	// the receiver invalidates.
+	SessionID string `json:"session_id"`
+	// Op says which way the binding went, so an unbind needs no read.
+	Op BindingOp `json:"op"`
+}
+
+// encode marshals the change for the wire as JSON, for the same reasons
+// EventRef.encode does: the payload is three short strings, so the size
+// difference against a packed encoding is noise, while `nats sub
+// 'compass.routing.binding.*'` on a live system stays readable and a later
+// additive field is forward-compatible with an older subscriber.
+func (b BindingChange) encode() ([]byte, error) {
+	data, err := json.Marshal(b)
+	if err != nil {
+		return nil, fmt.Errorf("fabric: encoding binding change %s/%s: %w", b.Tenant, b.SessionID, err)
+	}
+	return data, nil
+}
+
+// valid reports whether the change is publishable. Checked publish-side, like
+// EventRef.valid, because this plane has no ack, no retry and no DLQ: a change
+// that decodes but names no session is invisible on the wire — the receiver
+// would invalidate cache key "" and see nothing wrong — so the only place it can
+// be caught with the caller's stack is at its origin.
+//
+// Tenant must be a valid subject token because it IS a subject token.
+func (b BindingChange) valid() error {
+	if err := ValidSubjectToken("tenant", b.Tenant); err != nil {
+		return err
+	}
+	if b.SessionID == "" {
+		return fmt.Errorf("fabric: binding change for tenant %q has an empty session id", b.Tenant)
+	}
+	switch b.Op {
+	case BindingBound, BindingUnbound:
+	default:
+		return fmt.Errorf("fabric: binding change %s/%s has op %q, want %q or %q",
+			b.Tenant, b.SessionID, b.Op, BindingBound, BindingUnbound)
+	}
+	return nil
+}
+
+// decodeBindingChange parses a wire payload back into a BindingChange,
+// rejecting one that decodes but names nothing: a change naming no session
+// would invalidate cache key "" and read as "nothing changed".
+//
+// Deliberately NOT as strict as the publish side on Op. The publish side
+// rejects an unknown op because a caller minting one is a bug with a stack;
+// a RECEIVER seeing one is talking to a newer publisher, and dropping that
+// message would leave a genuinely-changed binding cached as stale — with no
+// ack, no retry and no DLQ to reveal it. So an unrecognized op is carried
+// through and the receiver treats it as "invalidate and re-read", which is the
+// conservative action and stays correct for every op this type can grow.
+// decodeEventRef takes the same position on an unknown EventKind.
+//
+// Tenant is still checked as a subject token, because handleBindingChange
+// rebuilds the subject from it to cross-check the delivery.
+//
+// A rejection here is LOGGED AND DROPPED by the caller, never retried: core
+// NATS has no ack, so there is nothing to Nak and nowhere to park. That is the
+// visible difference from the JetStream path, where an undecodable payload goes
+// to DLQSubject.
+func decodeBindingChange(data []byte) (BindingChange, error) {
+	var b BindingChange
+	if err := json.Unmarshal(data, &b); err != nil {
+		return BindingChange{}, fmt.Errorf("fabric: decoding binding change from %d bytes: %w", len(data), err)
+	}
+	if err := ValidSubjectToken("tenant", b.Tenant); err != nil {
+		return BindingChange{}, err
+	}
+	if b.SessionID == "" {
+		return BindingChange{}, fmt.Errorf("fabric: binding change for tenant %q has an empty session id", b.Tenant)
+	}
+	if b.Op == "" {
+		return BindingChange{}, fmt.Errorf("fabric: binding change %s/%s has no op", b.Tenant, b.SessionID)
+	}
+	return b, nil
+}

go/internal/fabric/doc.go

@@ -1,16 +1,17 @@
 // Package fabric is the NATS eventing substrate under Compass's async layer:
-// one client, one connection per party, two planes.
+// one client, one connection per party, three planes.
 //
 // The design record is
-// docs/designs/infra/runtime/compass-managed-multitenancy/design.md (§T3, §Q3).
-// The two seams it freezes are [EventFabric] (comms/delivery event fan-out) and
-// [RunnerFabric] (Server→Runner command push and Runner→Server event fan-in);
-// [Fabric] implements both over a single [nats.Conn], so each Runner and each
-// Server holds exactly one fabric connection.
+// docs/designs/infra/runtime/compass-managed-multitenancy/design.md (§T3, §Q3,
+// §T4). The two seams §T3 freezes are [EventFabric] (comms/delivery event
+// fan-out) and [RunnerFabric] (Server→Runner command push and Runner→Server
+// event fan-in); [RoutingFabric] (§T4 binding-cache invalidation) is the third.
+// [Fabric] implements all three over a single [nats.Conn], so each Runner and
+// each Server holds exactly one fabric connection.
 //
 // # The plane split
 //
-// The two seams ride deliberately different NATS semantics:
+// The seams ride deliberately different NATS semantics:
 //
 //   - [EventFabric] rides JetStream — durable at-least-once fan-out with
 //     explicit per-message acks, publish-side dedup, a bounded delivery-attempt
@@ -20,6 +21,12 @@
 //     offline Runner is not a lost write: the delivery-cursor sweep in Postgres
 //     recovers it ("a fabric outage degrades to sweep-recovered delivery"), so
 //     paying for a stream here would buy nothing and add a second store.
+//   - [RoutingFabric] rides core NATS too, but fans out rather than
+//     queue-grouping: every Server caches bindings, so every Server must see
+//     every invalidation, and a dropped one degrades to a cache-miss re-read
+//     against Postgres. That fan-out-and-droppable pair is why it is a third
+//     seam and not a method on [EventFabric], whose read paths are all durable
+//     queue-group consumers claiming each event for exactly one instance.
 //
 // # Postgres is the only truth
 //
@@ -38,6 +45,12 @@
 // rather than dropped, and a subscriber callback that panics is caught, retried
 // up to Config.MaxDeliver times, then parked.
 //
+// Those last two are JetStream properties. On [RoutingFabric] there is nowhere
+// to park — core NATS has no ack, so an undecodable payload and a panicking
+// callback are both logged and DROPPED, and the lost invalidation is recovered
+// by the next cache-miss re-read against Postgres. Fail-closed there means the
+// receiver keeps serving from Postgres, not that the message is retried.
+//
 // The subject grammar and the full JetStream stream/consumer/DLQ configuration
 // are specified in SUBJECTS.md beside this file — that document, not this
 // package's defaults, is what later tasks build against.

go/internal/fabric/fabric.go

@@ -44,6 +44,42 @@ type RunnerFabric interface {
 	Events(ctx context.Context) (<-chan RunnerEvent, error)
 }
 
+// RoutingFabric is the binding-invalidation seam (§T4). It rides core NATS —
+// best-effort at-most-once, deliberately: the hub's binding maps become
+// instance-local caches over durable truth, and "Postgres is the arbiter on any
+// cache miss or conflict; core NATS at-most-once suffices because a dropped
+// invalidation degrades to a cache-miss re-read".
+//
+// A THIRD seam rather than a method on EventFabric because the two delivery
+// contracts are opposites. EventFabric promises durable at-least-once fan-out
+// and every one of its read paths is a durable queue-group consumer, so each
+// event is claimed by exactly one instance. An invalidation must reach EVERY
+// instance that might hold the stale entry, and must be droppable. Neither
+// property can be expressed on the other seam without breaking it for its own
+// callers.
+//
+// So: no ack, no retry, and NO DLQ on this plane. A malformed received payload
+// is logged and dropped — there is nothing to Nak and nowhere to park — which
+// is the one visible behavioural difference from the JetStream path, where an
+// undecodable payload lands on DLQSubject.
+type RoutingFabric interface {
+	PublishBindingChange(ctx context.Context, tenant string, b BindingChange) error
+	// SubscribeBindingChanges is a PLAIN core-NATS subscribe with NO queue
+	// group, and that is the whole point of the seam. Every Server caches
+	// bindings independently, so every Server must receive every invalidation;
+	// joining a queue group would hand each invalidation to exactly one
+	// instance and leave the rest serving a stale binding — silently, since
+	// there is no error and no missing ack to notice. Contrast
+	// RunnerEventsQueue, which queue-groups on purpose: a Runner event is WORK,
+	// and work must be done once.
+	//
+	// fn may be handed a BindingOp outside the declared constants, published
+	// by a newer process; treat any unrecognized op as invalidate-and-re-read
+	// (see BindingOp). A switch with no default silently discards a real
+	// change on a plane with no ack, no retry and no dead-letter subject.
+	SubscribeBindingChanges(ctx context.Context, fn func(BindingChange)) (Unsubscribe, error)
+}
+
 // Config configures a Fabric. Only URL is required; every other field has a
 // documented default from stream.go, so the common case is
 // fabric.New(fabric.Config{URL: natsURL}).
@@ -178,8 +214,8 @@ func (c Config) logger() *slog.Logger {
 	return slog.Default()
 }
 
-// Fabric is the one NATS client: it implements both EventFabric and
-// RunnerFabric over a single connection, because the record gives each party
+// Fabric is the one NATS client: it implements EventFabric, RunnerFabric and
+// RoutingFabric over a single connection, because the record gives each party
 // exactly one ("Each Runner holds ONE fabric connection … each Server
 // likewise"). Safe for concurrent use.
 type Fabric struct {
@@ -215,11 +251,12 @@ type Fabric struct {
 	teardown chan struct{}
 }
 
-// Compile-time proof Fabric satisfies both frozen seams. Cheap here, and it
-// fails the build rather than a consumer's wiring if a signature drifts.
+// Compile-time proof Fabric satisfies every seam. Cheap here, and it fails the
+// build rather than a consumer's wiring if a signature drifts.
 var (
-	_ EventFabric  = (*Fabric)(nil)
-	_ RunnerFabric = (*Fabric)(nil)
+	_ EventFabric   = (*Fabric)(nil)
+	_ RunnerFabric  = (*Fabric)(nil)
+	_ RoutingFabric = (*Fabric)(nil)
 )
 
 // New connects to NATS and returns the fabric. It does not create the JetStream