diff --git a/go/internal/fabric/SUBJECTS.md b/go/internal/fabric/SUBJECTS.md index 1f447317..9696870d 100644 --- a/go/internal/fabric/SUBJECTS.md +++ b/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..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.` | core NATS | `ClientSubject(sessionID)` | Server → one live client connection | +| `compass.routing.binding.` | 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.` 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.` + +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..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 diff --git a/go/internal/fabric/bindingchange.go b/go/internal/fabric/bindingchange.go new file mode 100644 index 00000000..1cf8a9b5 --- /dev/null +++ b/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 +} diff --git a/go/internal/fabric/doc.go b/go/internal/fabric/doc.go index 0cf447ca..170148b3 100644 --- a/go/internal/fabric/doc.go +++ b/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. diff --git a/go/internal/fabric/fabric.go b/go/internal/fabric/fabric.go index 97a5fec3..d83f59a5 100644 --- a/go/internal/fabric/fabric.go +++ b/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 diff --git a/go/internal/fabric/fabric_test.go b/go/internal/fabric/fabric_test.go index e99d9267..2d3fe8fe 100644 --- a/go/internal/fabric/fabric_test.go +++ b/go/internal/fabric/fabric_test.go @@ -207,18 +207,24 @@ func TestNewUnreachableURL(t *testing.T) { } } -// TestFabricImplementsBothSeamsOverOneConnection defends the record's -// one-connection-per-party contract. That *Fabric satisfies both interfaces is -// a compile-time assertion in fabric.go; what a test can add is that the two -// seams are the SAME object over ONE nats.Conn — if EventFabric and -// RunnerFabric were ever split into two clients, each Runner and Server would -// hold two connections, which the record forbids. -func TestFabricImplementsBothSeamsOverOneConnection(t *testing.T) { +// TestFabricImplementsEverySeamOverOneConnection defends the record's +// one-connection-per-party contract. That *Fabric satisfies all three +// interfaces is a compile-time assertion in fabric.go; what a test can add is +// that the seams are the SAME object over ONE nats.Conn — if EventFabric, +// RunnerFabric and RoutingFabric were ever split into separate clients, each +// Runner and Server would hold several connections, which the record forbids. +// +// RoutingFabric is included because it is the seam most likely to be split off: +// it rides core NATS rather than JetStream, so "give the invalidation plane its +// own connection" reads as a reasonable change right up until every Server +// holds two. +func TestFabricImplementsEverySeamOverOneConnection(t *testing.T) { t.Parallel() f := newFabric(t, Config{}) var ( - ef EventFabric = f - rf RunnerFabric = f + ef EventFabric = f + rf RunnerFabric = f + rtf RoutingFabric = f ) efFabric, ok := ef.(*Fabric) if !ok { @@ -228,11 +234,15 @@ func TestFabricImplementsBothSeamsOverOneConnection(t *testing.T) { if !ok { t.Fatalf("RunnerFabric is backed by %T, want *Fabric", rf) } - if efFabric != rfFabric { - t.Fatal("the two seams must be one object, or a party holds two fabrics") + rtfFabric, ok := rtf.(*Fabric) + if !ok { + t.Fatalf("RoutingFabric is backed by %T, want *Fabric", rtf) + } + if efFabric != rfFabric || efFabric != rtfFabric { + t.Fatal("the three seams must be one object, or a party holds several fabrics") } - if efFabric.nc != rfFabric.nc { - t.Fatal("the two seams must share one nats connection") + if efFabric.nc != rfFabric.nc || efFabric.nc != rtfFabric.nc { + t.Fatal("the three seams must share one nats connection") } } diff --git a/go/internal/fabric/routing_fabric.go b/go/internal/fabric/routing_fabric.go new file mode 100644 index 00000000..22d216c7 --- /dev/null +++ b/go/internal/fabric/routing_fabric.go @@ -0,0 +1,209 @@ +package fabric + +import ( + "context" + "errors" + "fmt" + "sync" + + "github.com/nats-io/nats.go" +) + +// PublishBindingChange announces on core NATS that one session's binding +// changed, so every Server holding a cached entry for it invalidates. +// +// Core NATS, not JetStream, by design (§T4): "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" — a second store of state Postgres already owns — and would drag in +// the queue-group consumer semantics that are exactly wrong for this plane (see +// RoutingFabric). +// +// tenant is the subject's tenant token, and b.Tenant must equal it. The +// redundancy is deliberate: the read side subscribes tenant-wildcard and its +// callback never sees the subject, so the payload's tenant is the receiver's +// only scope — checking them equal here is what stops a caller from publishing +// tenant-a's change on tenant-b's subject, which would tell a receiver to +// invalidate the wrong tenant's cache entry. +// +// nc.Publish is fire-and-forget, so a nil return means "handed to the NATS +// client", not "every Server got it" — nothing on this plane can promise the +// latter. The flush against the caller's ctx is what makes the error real: it +// surfaces a connection that cannot take the write, which is the failure a +// caller can act on. The same reasoning as SendCommand's. +func (f *Fabric) PublishBindingChange(ctx context.Context, tenant string, b BindingChange) error { + if err := f.checkOpen(); err != nil { + return err + } + if err := b.valid(); err != nil { + return err + } + subject, err := RoutingBindingSubject(tenant) + if err != nil { + return err + } + if b.Tenant != tenant { + return fmt.Errorf("fabric: binding change for tenant %q was published under tenant %q", b.Tenant, tenant) + } + data, err := b.encode() + if err != nil { + return err + } + if err := f.nc.Publish(subject, data); err != nil { + return fmt.Errorf("fabric: publishing binding change %s/%s to %q: %w", b.Tenant, b.SessionID, subject, err) + } + if err := f.flush(ctx); err != nil { + return fmt.Errorf("fabric: flushing binding change %s/%s to %q: %w", b.Tenant, b.SessionID, subject, err) + } + return nil +} + +// SubscribeBindingChanges drives fn for every binding change on every tenant +// until the returned Unsubscribe is called, ctx is done, or the Fabric is +// closed — whichever comes first. +// +// NO QUEUE GROUP, and that is the contract, not an omission: each Server keeps +// its own binding cache, so each must see every invalidation. A queue group +// would deliver each change to exactly one Server and leave the others serving +// a stale binding with no error, no gap and no missing ack to reveal it. +// RunnerFabric.Events does join a queue group (RunnerEventsQueue) for the +// opposite reason: a Runner event is work, and work must be done once. +// +// The subscribe is tenant-wildcard (RoutingBindingWildcardSubject) because 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. +// +// A malformed payload is logged and DROPPED — this plane has no ack to withhold, +// no Nak to send and no DLQ to park on, so surviving is the only option that +// keeps invalidation flowing for every other session. A panicking fn is +// recovered for the same reason: fn runs on the NATS dispatcher goroutine, so +// letting it panic would take the process down. +func (f *Fabric) SubscribeBindingChanges(ctx context.Context, fn func(BindingChange)) (Unsubscribe, error) { + if err := f.checkOpen(); err != nil { + return nil, err + } + if fn == nil { + return nil, errors.New("fabric: SubscribeBindingChanges requires a callback") + } + subject := RoutingBindingWildcardSubject() + + // nc.Subscribe, NOT nc.QueueSubscribe. See the seam's contract: a queue + // group here silently breaks cache coherence. + sub, err := f.nc.Subscribe(subject, func(msg *nats.Msg) { + f.handleBindingChange(ctx, msg, fn) + }) + if err != nil { + return nil, fmt.Errorf("fabric: subscribing %q: %w", subject, err) + } + // Flush so this returns only once the server has registered the interest. + // Load-bearing on core NATS in a way it is not on JetStream: the server + // DROPS a message with no matching interest, so a caller that subscribed + // and then triggered a binding change would race its own first + // invalidation and never learn it was lost. Same argument as Events'. + if err := f.flush(ctx); err != nil { + if derr := sub.Unsubscribe(); derr != nil { + f.log.WarnContext(ctx, "fabric: unsubscribing after a failed flush", "subject", subject, "error", derr) + } + return nil, fmt.Errorf("fabric: establishing the binding-change subscription on %q: %w", subject, err) + } + + // One teardown path, reached from the caller's Unsubscribe, from ctx being + // done, or from the fabric closing, and run at most once — so the watchdog + // goroutine always exits and the subscription is never torn down twice. + // + // f.teardown is load-bearing, not a duplicate of ctx.Done(): a Close with + // an uncancelled ctx (a Server whose root context outlives the fabric — + // the ordinary shutdown shape) would otherwise leak this goroutine. + // + // Drain rather than Unsubscribe, mirroring the runner-events pump: it lets + // NATS deliver what it has already accepted for this subject before the + // interest goes away, so an invalidation this process already had in hand + // still reaches the cache. An already-closed connection is the expected + // shutdown outcome, not a failure — Close closes f.teardown before + // nc.Drain(), and the connection-level drain reclaims every subscription + // itself, so a later sub.Drain() reporting ErrConnectionClosed describes a + // subscription that WAS drained. + var once sync.Once + done := make(chan struct{}) + stop := func() { + once.Do(func() { + if err := sub.Drain(); err != nil && !errors.Is(err, nats.ErrConnectionClosed) { + f.log.WarnContext(ctx, "fabric: draining the binding-change subscription failed", + "subject", subject, "error", err) + } + close(done) + }) + } + go func() { + select { + case <-ctx.Done(): + stop() + case <-f.teardown: + stop() + case <-done: + } + }() + return stop, nil +} + +// handleBindingChange runs one delivery: decode, cross-check the payload's +// tenant against the subject it arrived on, then invoke fn under a panic guard. +// Split out of SubscribeBindingChanges so the drop decision is readable on its +// own — and so it is visibly the WHOLE of it: there is no ack, no +// retry-or-park branch and no DLQ republish after this point, because core NATS +// offers nowhere to put a message it could not process. +func (f *Fabric) handleBindingChange(ctx context.Context, msg *nats.Msg, fn func(BindingChange)) { + b, err := decodeBindingChange(msg.Data) + if err != nil { + // Best-effort plane: one malformed publish must not stop invalidation + // for every other session. Surfaced, never silent — the log line is the + // only trace this message will ever leave. + f.log.ErrorContext(ctx, "fabric: dropping an undecodable binding change", + "subject", msg.Subject, "error", err) + return + } + // The payload's tenant must match the subject's, exactly as handleEvent + // checks a comms ref. The read side subscribes tenant-wildcard and fn never + // sees the subject, so b.Tenant is the receiver's ONLY scope — and until + // OQ-3 lands per-tenant NATS authorization, any client that can reach the + // client port can publish any subject. Without this, a publish of + // {tenant: victim} on the attacker's own subject would hand a subscriber a + // victim-scoped change, and an "unbound" op is a stale NEGATIVE the + // receiver drops outright rather than a re-read that would self-correct. + // + // Dropped, not parked: this plane has no dead-letter subject, and a + // mismatch is as unprocessable as an undecodable payload. + want, subjErr := RoutingBindingSubject(b.Tenant) + if subjErr != nil { + f.log.ErrorContext(ctx, "fabric: dropping a binding change whose tenant is not a valid subject token", + "subject", msg.Subject, "error", subjErr) + return + } + if msg.Subject != want { + f.log.ErrorContext(ctx, "fabric: dropping a binding change delivered on a foreign tenant's subject", + "subject", msg.Subject, "claimed_tenant", b.Tenant, "claimed_subject", want, "session_id", b.SessionID) + return + } + if err := invokeBindingChange(fn, b); err != nil { + // A failed callback is a DROP here, not a redelivery: the cache entry + // stays stale until the next change or a cache miss re-reads Postgres, + // which is the arbiter. Logged, because a subscriber panicking is a bug + // even though this plane tolerates it. + f.log.ErrorContext(ctx, "fabric: a binding-change subscriber failed; the invalidation is dropped", + "subject", msg.Subject, "session_id", b.SessionID, "error", err) + } +} + +// invokeBindingChange calls fn, converting a panic into an error. fn is consumer +// code running on the fabric's dispatcher goroutine, so letting it panic would +// take the process down over one bad cache entry. +func invokeBindingChange(fn func(BindingChange), b BindingChange) (err error) { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("fabric: binding-change subscriber panicked handling %s/%s: %v", b.Tenant, b.SessionID, r) + } + }() + fn(b) + return nil +} diff --git a/go/internal/fabric/routing_fabric_test.go b/go/internal/fabric/routing_fabric_test.go new file mode 100644 index 00000000..999975c3 --- /dev/null +++ b/go/internal/fabric/routing_fabric_test.go @@ -0,0 +1,775 @@ +package fabric + +import ( + "context" + "errors" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/nats-io/nats.go" +) + +// bindingWatchdogMarker names every goroutine spawned on the +// SubscribeBindingChanges path. The routing seam has its OWN copy of the +// watchdog pattern rather than reusing subscribeSubject, so the sibling +// event-plane goroutine tests provably do not cover it: their marker is +// "fabric.(*Fabric).subscribeSubject.func", which never appears in this seam's +// stacks. A residual watchdog shows up as a count that never falls back to +// baseline. +const bindingWatchdogMarker = "fabric.(*Fabric).SubscribeBindingChanges.func" + +// recvBindingChange takes the next BindingChange from ch, failing at the gate. +func recvBindingChange(t *testing.T, ch <-chan BindingChange) BindingChange { + t.Helper() + select { + case b := <-ch: + return b + case <-time.After(gate): + t.Fatalf("no binding change within %s", gate) + return BindingChange{} + } +} + +// TestBindingChangesFanOutToEverySubscriber is the reason this seam exists at +// all. Every Server keeps its own binding cache, so an invalidation must reach +// ALL of them; if SubscribeBindingChanges ever joins a queue group, the server +// hands each change to exactly one instance and the rest go on serving a stale +// binding — with no error, no gap and no missing ack to reveal it. This is the +// only test that fails when that happens. +// +// Two Fabrics against one NATS, because two connections is what two Servers +// actually are — and a queue group partitions across connections, so a +// single-connection test would be a weaker guard. +func TestBindingChangesFanOutToEverySubscriber(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + first := newFabric(t, Config{URL: url}) + second := newFabric(t, Config{URL: url}) + + firstCh := make(chan BindingChange, 1) + unsubFirst, err := first.SubscribeBindingChanges(ctx, func(b BindingChange) { firstCh <- b }) + if err != nil { + t.Fatalf("SubscribeBindingChanges (first server): %v", err) + } + defer unsubFirst() + + secondCh := make(chan BindingChange, 1) + unsubSecond, err := second.SubscribeBindingChanges(ctx, func(b BindingChange) { secondCh <- b }) + if err != nil { + t.Fatalf("SubscribeBindingChanges (second server): %v", err) + } + defer unsubSecond() + + want := BindingChange{Tenant: "tenant-a", SessionID: "sess-1", Op: BindingBound} + if err := first.PublishBindingChange(ctx, "tenant-a", want); err != nil { + t.Fatalf("PublishBindingChange: %v", err) + } + + // Both are awaited concurrently rather than one after the other, and the + // gate is shared: under a queue group the server picks ONE subscriber, and + // which one is not deterministic — so a sequential receive would fail on + // whichever channel happened to lose, with a bare timeout that does not + // name the broken invariant. Counting arrivals fails the same way every + // time, and says what went wrong. + got := map[string]BindingChange{} + deadline := time.After(gate) + for range 2 { + select { + case b := <-firstCh: + got["the publishing server"] = b + case b := <-secondCh: + got["the second server"] = b + case <-deadline: + t.Fatalf("only %d of 2 subscribers received the change within %s (received: %v); every Server caches bindings independently, so every Server must be invalidated — a queue group delivers to exactly one and leaves the rest serving a stale binding", + len(got), gate, got) + } + } + for who, b := range got { + if b != want { + t.Errorf("%s received %+v, want %+v", who, b, want) + } + } +} + +// TestBindingSubjectIsNotCapturedByTheCommsStream defends the token order, on +// the degenerate tenant value that order exists for. compass.*.comms.* — the +// COMPASS_COMMS stream's capture — matches any four-token subject whose THIRD +// token is "comms", so the rejected grammar compass.routing..binding +// would put a tenant literally named "comms" INSIDE the JetStream comms stream: +// compass.routing.comms.binding. A stream is an ordinary subscriber in the +// account sublist, so that capture is ADDITIVE and the hubs would still receive +// the message; the harm is that a best-effort invalidation would ALSO be +// persisted into a durable stream specified to hold only EventRefs, burning its +// storage and MaxAge budget on traffic no comms consumer can use. That is why +// the assertion below is that the stream stayed EMPTY. +// +// Both directions are asserted, because a shared NATS makes cross-plane leakage +// possible either way: a binding change must not enter the stream, and a comms +// publish must not reach a binding subscriber. +func TestBindingSubjectIsNotCapturedByTheCommsStream(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url}) + + // The stream must exist BEFORE the publish, or "the stream is empty" would + // prove nothing. + stream, err := f.ensureStream(ctx) + if err != nil { + t.Fatalf("ensureStream: %v", err) + } + + // A raw core-NATS subscriber on the binding wildcard, so the isolation + // assertion is about the wire and not the fabric's own dispatch. + raw, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(raw.Close) + bindings, err := raw.SubscribeSync(RoutingBindingWildcardSubject()) + if err != nil { + t.Fatalf("SubscribeSync(%q): %v", RoutingBindingWildcardSubject(), err) + } + if err := raw.FlushWithContext(ctx); err != nil { + t.Fatalf("flushing the binding subscription: %v", err) + } + + // "comms" as a tenant id: legal (ValidSubjectToken permits it) and exactly + // the value the token order defends against. + const tenant = "comms" + change := BindingChange{Tenant: tenant, SessionID: "sess-collide", Op: BindingUnbound} + if err := f.PublishBindingChange(ctx, tenant, change); err != nil { + t.Fatalf("PublishBindingChange: %v", err) + } + if _, err := bindings.NextMsgWithContext(ctx); err != nil { + t.Fatalf("the binding subscriber did not receive its own plane's change: %v", err) + } + + // PublishBindingChange flushed, so the server has already decided whether + // the subject matched the stream — nothing is in flight. + info, err := stream.Info(ctx) + if err != nil { + t.Fatalf("stream Info: %v", err) + } + if info.State.Msgs != 0 { + subject, serr := RoutingBindingSubject(tenant) + if serr != nil { + t.Fatalf("RoutingBindingSubject(%q): %v", tenant, serr) + } + t.Fatalf("%s holds %d message(s) after a binding change on %q; the core-NATS invalidation plane was captured by %q", + f.cfg.streamName(), info.State.Msgs, subject, commsStreamSubjects) + } + + // The other direction: a real comms publish for the same degenerate tenant + // must not reach a binding subscriber. + commsSubject, err := CommsSubject(tenant, KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + if err := f.Publish(ctx, commsSubject, EventRef{Tenant: tenant, Kind: KindMessagePosted, RowID: "m1"}); err != nil { + t.Fatalf("Publish: %v", err) + } + // Publish is acked by the stream, and the binding subscription's interest + // was established before it, so a leaked message would already be queued. + if n, _, err := bindings.Pending(); err != nil { + t.Fatalf("reading the binding subscription's pending count: %v", err) + } else if n != 0 { + t.Fatalf("the binding subscriber has %d pending message(s) after a comms publish on %q; the planes must not cross", n, commsSubject) + } + // Positive control for the count above: this comms publish MUST be in the + // stream. Without it, "Msgs == 0" would also pass if the stream simply + // never counted anything on this server, and the isolation assertion would + // be vacuous. + info, err = stream.Info(ctx) + if err != nil { + t.Fatalf("stream Info after the comms publish: %v", err) + } + if info.State.Msgs != 1 { + t.Fatalf("%s holds %d message(s) after a comms publish on %q, want 1; the earlier empty-stream assertion proves nothing if the stream never captures", + f.cfg.streamName(), info.State.Msgs, commsSubject) + } +} + +// TestBindingChangeIsTenantScoped defends the subject's tenant token. A Server +// resolving sessions for one tenant only — or an operator tapping one tenant's +// subject — must not see another tenant's invalidations, which would drop cache +// entries by a key from the wrong namespace. +func TestBindingChangeIsTenantScoped(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url}) + + raw, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(raw.Close) + + subjectA, err := RoutingBindingSubject("tenant-a") + if err != nil { + t.Fatalf("RoutingBindingSubject: %v", err) + } + onlyA, err := raw.SubscribeSync(subjectA) + if err != nil { + t.Fatalf("SubscribeSync(%q): %v", subjectA, err) + } + // A wildcard subscriber alongside it, so the test gates on a POSITIVE + // delivery rather than on the absence of one: tenant-b's change arriving + // here is proof the publish happened at all. + everyTenant, err := raw.SubscribeSync(RoutingBindingWildcardSubject()) + if err != nil { + t.Fatalf("SubscribeSync(%q): %v", RoutingBindingWildcardSubject(), err) + } + if err := raw.FlushWithContext(ctx); err != nil { + t.Fatalf("flushing subscriptions: %v", err) + } + + changeB := BindingChange{Tenant: "tenant-b", SessionID: "sess-b", Op: BindingBound} + if err := f.PublishBindingChange(ctx, "tenant-b", changeB); err != nil { + t.Fatalf("PublishBindingChange: %v", err) + } + if _, err := everyTenant.NextMsgWithContext(ctx); err != nil { + t.Fatalf("the wildcard subscriber did not receive tenant-b's change: %v", err) + } + if n, _, err := onlyA.Pending(); err != nil { + t.Fatalf("reading tenant-a's pending count: %v", err) + } else if n != 0 { + t.Fatalf("tenant-a's subscriber has %d pending message(s); a binding change must not cross tenants", n) + } +} + +// TestBindingChangeRejectsInvalidInput defends the fail-closed publish path at +// both gates: the subject builder and PublishBindingChange itself. On a +// best-effort plane every one of these failures is otherwise silent — a +// corrupted subject publishes to a subject nobody watches, and a change naming +// no session invalidates cache key "". +func TestBindingChangeRejectsInvalidInput(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + for _, tenant := range []string{"", "tenant.a", "tenant*", "tenant>", "tenant a", "tenant\tb", "tenant\nc"} { + if _, err := RoutingBindingSubject(tenant); err == nil { + t.Errorf("RoutingBindingSubject(%q): want an error", tenant) + } + change := BindingChange{Tenant: tenant, SessionID: "sess-1", Op: BindingBound} + if err := f.PublishBindingChange(ctx, tenant, change); err == nil { + t.Errorf("PublishBindingChange for tenant %q: want an error", tenant) + } + } + + // A well-formed tenant with a payload that names nothing, or names a + // nonsense op, must still fail. + for _, change := range []BindingChange{ + {Tenant: "tenant-a", SessionID: "", Op: BindingBound}, + {Tenant: "tenant-a", SessionID: "sess-1", Op: ""}, + {Tenant: "tenant-a", SessionID: "sess-1", Op: "rebound"}, + } { + if err := f.PublishBindingChange(ctx, "tenant-a", change); err == nil { + t.Errorf("PublishBindingChange(%+v): want an error", change) + } + } + + // A payload whose tenant disagrees with the subject it is published on: + // the read side is tenant-wildcard and its callback never sees the subject, + // so the payload's tenant is the receiver's only scope. + crossed := BindingChange{Tenant: "tenant-b", SessionID: "sess-1", Op: BindingBound} + if err := f.PublishBindingChange(ctx, "tenant-a", crossed); err == nil { + t.Error("PublishBindingChange with a payload tenant that differs from the subject tenant: want an error") + } + + if _, err := f.SubscribeBindingChanges(ctx, nil); err == nil { + t.Error("SubscribeBindingChanges with a nil callback: want an error") + } +} + +// TestMalformedBindingChangeIsDropped defends the plane's resilience where +// TestPoisonMessageParksOnDLQ defends JetStream's. There is no ack to withhold, +// no Nak to send and no DLQ to park on here, so one unparseable publish must be +// logged and dropped while the subscription keeps running — the next valid +// change arriving is the gate. +// +// The garbage goes out over a plain nats.Connect, because the fabric's own +// publish path cannot produce it: BindingChange.valid rejects it first. Which is +// the point — COMPASS's NATS carries no per-tenant authorization yet (OQ-3), so +// the fabric is not the only writer that can reach this subject. +func TestMalformedBindingChangeIsDropped(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url, Log: quietLogger(t)}) + + got := make(chan BindingChange, 1) + unsub, err := f.SubscribeBindingChanges(ctx, func(b BindingChange) { got <- b }) + if err != nil { + t.Fatalf("SubscribeBindingChanges: %v", err) + } + defer unsub() + + writer, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(writer.Close) + + subject, err := RoutingBindingSubject("tenant-a") + if err != nil { + t.Fatalf("RoutingBindingSubject: %v", err) + } + // Two shapes of bad payload: bytes that are not JSON at all, and JSON that + // decodes but names no session. The second is the dangerous one — it would + // otherwise invalidate cache key "" and look like a successful delivery. + for _, bad := range [][]byte{ + []byte("{not json"), + []byte(`{"tenant":"tenant-a","session_id":"","op":"bound"}`), + } { + if err := writer.Publish(subject, bad); err != nil { + t.Fatalf("publishing a malformed binding change: %v", err) + } + } + // Flush the writer BEFORE the valid publish. Core NATS preserves order per + // publisher only, and these are two connections — without this the + // malformed bytes could still be in flight when the valid change is + // delivered, so the assertion below would pass without the drop path having + // run at all, and would keep passing if the rejection were deleted. + if err := writer.FlushWithContext(ctx); err != nil { + t.Fatalf("flushing the malformed publishes: %v", err) + } + + want := BindingChange{Tenant: "tenant-a", SessionID: "sess-after", Op: BindingUnbound} + if err := f.PublishBindingChange(ctx, "tenant-a", want); err != nil { + t.Fatalf("PublishBindingChange: %v", err) + } + // Both malformed payloads are now known to the server, so if either were + // surfaced instead of dropped it would be queued AHEAD of the valid change + // and this assertion would name it. + if received := recvBindingChange(t, got); received != want { + t.Fatalf("received %+v, want %+v; a malformed payload must be dropped, not surfaced", received, want) + } +} + +// TestRoutingFabricFailsClosedAfterClose defends the fail-closed gate on the +// third seam, the same way TestCloseIsIdempotentAndFailsClosed does for the +// other two. A PublishBindingChange that silently no-oped after shutdown would +// look like a delivered invalidation, and on a plane with no ack there is +// nothing else that would ever contradict it. +func TestRoutingFabricFailsClosedAfterClose(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f, err := New(Config{URL: testServer(t)}) + if err != nil { + t.Fatalf("New: %v", err) + } + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + change := BindingChange{Tenant: "tenant-a", SessionID: "sess-1", Op: BindingBound} + if err := f.PublishBindingChange(ctx, "tenant-a", change); !errors.Is(err, errClosed) { + t.Errorf("PublishBindingChange after Close: want errClosed, got %v", err) + } + if _, err := f.SubscribeBindingChanges(ctx, func(BindingChange) {}); !errors.Is(err, errClosed) { + t.Errorf("SubscribeBindingChanges after Close: want errClosed, got %v", err) + } +} + +// TestForgedTenantBindingChangeIsDropped defends the read side of the tenant +// invariant, and it is a cross-tenant cache attack that it closes. +// PublishBindingChange refuses a change whose payload tenant disagrees with the +// subject it publishes under, but the fabric is NOT the only writer that can +// reach this subject: COMPASS's NATS carries no per-tenant authorization yet +// (OQ-3), so any client that can reach the client port can publish any subject. +// This test bypasses the fabric with a raw connection exactly as a rogue +// publisher would. +// +// SubscribeBindingChanges is tenant-WILDCARD and its callback receives the +// decoded change and no subject, so b.Tenant is the receiver's only scope — a +// delivered mismatch is an instruction to drop another tenant's cache entry. +// +// The forged op is "unbound" deliberately: that is the dangerous one. A forged +// "bound" degrades to a spurious re-read that Postgres self-corrects, while an +// "unbound" is a stale NEGATIVE the receiver may drop outright without ever +// consulting the arbiter. +// +// Dropped, not parked: core NATS has no ack to withhold and no DLQ, which is +// the one difference from TestForgedTenantRefIsParked on the JetStream seam. +func TestForgedTenantBindingChangeIsDropped(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url, Log: quietLogger(t)}) + + delivered := make(chan BindingChange, 2) + unsub, err := f.SubscribeBindingChanges(ctx, func(b BindingChange) { delivered <- b }) + if err != nil { + t.Fatalf("SubscribeBindingChanges: %v", err) + } + defer unsub() + + raw, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(raw.Close) + + // tenant-attacker's own subject carrying a tenant-victim payload. The + // attacker needs no access to the victim's subject at all, which is why the + // subject alone cannot be the scope. + attackerSubject, err := RoutingBindingSubject("tenant-attacker") + if err != nil { + t.Fatalf("RoutingBindingSubject: %v", err) + } + forged, err := BindingChange{Tenant: "tenant-victim", SessionID: "sess-forged", Op: BindingUnbound}.encode() + if err != nil { + t.Fatalf("encoding the forged change: %v", err) + } + if err := raw.Publish(attackerSubject, forged); err != nil { + t.Fatalf("raw Publish(%q): %v", attackerSubject, err) + } + // Flush the ATTACKER's connection before the legitimate publish. Core NATS + // preserves order per publisher and these are two publishers, so without + // this the forgery could still be in flight when the assertion below runs + // and "it did not arrive" would mean "not yet". The flush round-trips a + // PING, so once it returns the server has processed the forged publish and + // already made its delivery decision for this subscription. + if err := raw.FlushWithContext(ctx); err != nil { + t.Fatalf("flushing the forged publish: %v", err) + } + + // The positive gate on a negative assertion, and it doubles as proof the + // drop did not kill the subscription: a mismatch is unprocessable, but it + // must cost only its own message. + good := BindingChange{Tenant: "tenant-attacker", SessionID: "sess-legit", Op: BindingBound} + if err := f.PublishBindingChange(ctx, "tenant-attacker", good); err != nil { + t.Fatalf("PublishBindingChange: %v", err) + } + + if got := recvBindingChange(t, delivered); got != good { + t.Fatalf("first delivery = %+v, want the legitimate %+v — a change claiming tenant %q was delivered from %q, and the receiver would drop that tenant's cache entry", + got, good, "tenant-victim", attackerSubject) + } + select { + case extra := <-delivered: + t.Fatalf("a second change reached the subscriber: %+v, want only the legitimate one", extra) + default: + } +} + +// TestUnrecognizedBindingOpIsDelivered defends the read side's DELIBERATE +// laxity about Op, which is not an oversight in decodeBindingChange but the +// forward-compatibility contract. A subscriber running older code than the +// publisher must still invalidate: dropping an op it does not recognize would +// leave a genuinely-changed binding cached as stale, and on a plane with no +// ack, no retry and no DLQ nothing would ever contradict it. "Invalidate and +// re-read" stays correct for every op this type can grow, so it is the only +// safe response to an unknown one. +// +// An EMPTY op is still rejected, and that is the line: absent is not unknown. +// An empty op is a payload that names no operation at all — the same class of +// nothing as an empty session id — and it is what a zero-valued or truncated +// publish produces rather than a newer peer. +// +// Both payloads go out raw, because the fabric's own publish path cannot +// produce either: BindingChange.valid rejects them first (see +// TestPublishBindingChangeRejectsAnUnrecognizedOp for that side). +func TestUnrecognizedBindingOpIsDelivered(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url, Log: quietLogger(t)}) + + delivered := make(chan BindingChange, 2) + unsub, err := f.SubscribeBindingChanges(ctx, func(b BindingChange) { delivered <- b }) + if err != nil { + t.Fatalf("SubscribeBindingChanges: %v", err) + } + defer unsub() + + raw, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(raw.Close) + + subject, err := RoutingBindingSubject("tenant-a") + if err != nil { + t.Fatalf("RoutingBindingSubject: %v", err) + } + // Order matters and is guaranteed: both publishes share one connection, so + // core NATS preserves their order. The empty-op payload goes FIRST, so if + // it were wrongly delivered it would arrive ahead of the one this test + // expects and the assertion names it rather than timing out. + for _, payload := range []string{ + `{"tenant":"tenant-a","session_id":"sess-empty-op","op":""}`, + `{"tenant":"tenant-a","session_id":"sess-newer","op":"resumed"}`, + } { + if err := raw.Publish(subject, []byte(payload)); err != nil { + t.Fatalf("raw Publish(%q): %v", subject, err) + } + } + + // "resumed" is a value no BindingOp constant defines today, which is the + // whole point: this is what a newer publisher looks like from here. + want := BindingChange{Tenant: "tenant-a", SessionID: "sess-newer", Op: "resumed"} + if got := recvBindingChange(t, delivered); got != want { + t.Fatalf("first delivery = %+v, want %+v — an op this subscriber does not recognize must be carried through, and an empty one must not be", + got, want) + } + select { + case extra := <-delivered: + t.Fatalf("a second change reached the subscriber: %+v; the empty-op payload must be dropped", extra) + default: + } +} + +// TestPublishBindingChangeRejectsAnUnrecognizedOp pins the OTHER half of the +// asymmetry TestUnrecognizedBindingOpIsDelivered describes, and the asymmetry +// itself is the contract: lax on receive, strict on publish. A caller minting +// an op the type does not define is a bug in this process, and this is the only +// point on the whole plane where such a bug can be reported to someone holding +// a stack — after the publish there is no ack, no retry and no DLQ. +// +// So the strict closed-set check must survive on the publish side even though +// the receive side deliberately dropped it. +func TestPublishBindingChangeRejectsAnUnrecognizedOp(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + // The same value the receive side carries through, so the two tests read as + // the pair they are. + unknown := BindingChange{Tenant: "tenant-a", SessionID: "sess-newer", Op: "resumed"} + if err := f.PublishBindingChange(ctx, "tenant-a", unknown); err == nil { + t.Error("PublishBindingChange with an op no BindingOp constant defines: want an error; the closed-set check must stay on the publish side") + } + // The two ops that ARE defined must pass, or the check above would be + // satisfied by a publish path that rejects everything. + for _, op := range []BindingOp{BindingBound, BindingUnbound} { + change := BindingChange{Tenant: "tenant-a", SessionID: "sess-ok", Op: op} + if err := f.PublishBindingChange(ctx, "tenant-a", change); err != nil { + t.Errorf("PublishBindingChange with op %q: %v", op, err) + } + } +} + +// TestSubscribeBindingChangesWatchdogExitsOnClose defends Close as a complete +// shutdown of the ROUTING plane. This seam has its own watchdog goroutine +// rather than reusing subscribeSubject, so the event plane's +// TestSubscribeWatchdogExitsOnClose provably does not cover it — a different +// function, a different marker, a separately-written select. +// +// The context is deliberately never cancelled, which is the case f.teardown +// exists for: a Server whose root context outlives its fabric is the ORDINARY +// shutdown shape, and a watchdog selecting on ctx.Done() alone would park +// forever holding a subscription on a dead connection. Close must release it +// with no help from the caller's context. +func TestSubscribeBindingChangesWatchdogExitsOnClose(t *testing.T) { + // Deliberately NOT parallel: this counts goroutines process-wide, and a + // sibling test's live SubscribeBindingChanges is indistinguishable from a + // leak of this one's. Go runs non-parallel tests while the parallel ones + // are paused. + // + // New directly rather than newFabric: this test closes the fabric itself, + // and newFabric's cleanup would then fail the test on the second Close. + f, err := New(Config{URL: testServer(t)}) + if err != nil { + t.Fatalf("New: %v", err) + } + + baseline := countGoroutinesWith(t, bindingWatchdogMarker) + + // Rooted at context.Background() because this is a test root, and an + // UNCANCELLED context is the whole point of the test. + unsub, err := f.SubscribeBindingChanges(context.Background(), func(BindingChange) {}) + if err != nil { + t.Fatalf("SubscribeBindingChanges: %v", err) + } + defer unsub() + + // Prove the watchdog started, so the assertion below distinguishes "exited" + // from "never ran". + pollUntil(t, "the binding-change watchdog to start", func() bool { + return countGoroutinesWith(t, bindingWatchdogMarker) > baseline + }) + + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + pollUntil(t, "the binding-change watchdog to exit after Close", func() bool { + return countGoroutinesWith(t, bindingWatchdogMarker) <= baseline + }) +} + +// TestSubscribeBindingChangesStopsWhenContextIsDone defends the second of the +// three teardown paths into the same sync.Once: a subscription whose ctx is +// cancelled must tear itself down with no Unsubscribe call and no Close. +// Otherwise a Server shutdown that cancels its root context would leave a +// watchdog per subscription running against a connection it no longer owns. +func TestSubscribeBindingChangesStopsWhenContextIsDone(t *testing.T) { + // Not parallel, for the same process-wide goroutine count as the Close + // test above. + f := newFabric(t, Config{}) + + baseline := countGoroutinesWith(t, bindingWatchdogMarker) + + // Rooted at context.Background() because this is a test root; the fabric + // stays open, so cancelling this context is the only teardown in play. + subCtx, cancel := context.WithCancel(context.Background()) + unsub, err := f.SubscribeBindingChanges(subCtx, func(BindingChange) {}) + if err != nil { + cancel() + t.Fatalf("SubscribeBindingChanges: %v", err) + } + defer unsub() + + pollUntil(t, "the binding-change watchdog to start", func() bool { + return countGoroutinesWith(t, bindingWatchdogMarker) > baseline + }) + + cancel() + pollUntil(t, "the binding-change watchdog to exit after its context was cancelled", func() bool { + return countGoroutinesWith(t, bindingWatchdogMarker) <= baseline + }) +} + +// TestUnsubscribeBindingChangesStopsDelivery defends the third path — the +// caller's own Unsubscribe — and the sync.Once that makes all three safe. +// +// Two things are asserted, and the second is why the Once is there. A leaked +// subscription on this plane is quiet in a way a JetStream one is not: there is +// no consumer to hold and no ack to withhold, so a stale callback simply keeps +// running against cache entries its owner has abandoned. And a SECOND unsub() +// must be safe: the caller's defer commonly runs after ctx was already +// cancelled or the fabric closed, so a double teardown is the ordinary case, +// not an abuse — without the Once it would Drain a subscription twice and close +// an already-closed channel, which panics on the caller's goroutine. +// +// Absence of delivery is proven POSITIVELY: a second subscription on the same +// wildcard receives the change the stale callback must not have seen, so the +// test cannot pass merely because nothing was published. +func TestUnsubscribeBindingChangesStopsDelivery(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + var stale atomic.Int64 + first := make(chan BindingChange, 1) + unsub, err := f.SubscribeBindingChanges(ctx, func(b BindingChange) { + stale.Add(1) + select { + case first <- b: + default: + } + }) + if err != nil { + t.Fatalf("SubscribeBindingChanges: %v", err) + } + + // Prove the subscription is live before tearing it down, so the assertion + // below distinguishes "stopped" from "never started". + warmup := BindingChange{Tenant: "tenant-a", SessionID: "sess-warmup", Op: BindingBound} + if err := f.PublishBindingChange(ctx, "tenant-a", warmup); err != nil { + t.Fatalf("PublishBindingChange warmup: %v", err) + } + if got := recvBindingChange(t, first); got != warmup { + t.Fatalf("warmup delivered %+v, want %+v", got, warmup) + } + before := stale.Load() + + unsub() + // Idempotent by contract: a second call must not double-Drain or re-close. + unsub() + + second := make(chan BindingChange, 1) + unsub2, err := f.SubscribeBindingChanges(ctx, func(b BindingChange) { second <- b }) + if err != nil { + t.Fatalf("second SubscribeBindingChanges: %v", err) + } + defer unsub2() + + want := BindingChange{Tenant: "tenant-a", SessionID: "sess-after", Op: BindingUnbound} + if err := f.PublishBindingChange(ctx, "tenant-a", want); err != nil { + t.Fatalf("PublishBindingChange after unsubscribe: %v", err) + } + if got := recvBindingChange(t, second); got != want { + t.Fatalf("second subscriber delivered %+v, want %+v", got, want) + } + // The live subscription received, so the unsubscribed one has demonstrably + // had its chance; a delivery here would be a leak, not a race. + if after := stale.Load(); after != before { + t.Fatalf("the unsubscribed callback ran %d more time(s) after Unsubscribe", after-before) + } +} + +// TestInvokeBindingChangeConvertsPanicToError defends the guard in isolation. +// fn is consumer code that runs on the NATS DISPATCHER goroutine, so an +// unrecovered panic there is not a failed delivery — it is the process, and +// every other plane sharing this connection, gone over one bad cache entry. +// The panic must become an error the delivery path can log and drop. +func TestInvokeBindingChangeConvertsPanicToError(t *testing.T) { + t.Parallel() + b := BindingChange{Tenant: "tenant-a", SessionID: "sess-1", Op: BindingBound} + + if err := invokeBindingChange(func(BindingChange) {}, b); err != nil { + t.Fatalf("a callback that returns normally must not error: %v", err) + } + + err := invokeBindingChange(func(BindingChange) { panic(errors.New("boom")) }, b) + if err == nil { + t.Fatal("a panicking callback must yield an error, not a nil — the delivery path would otherwise log nothing and the panic would be invisible") + } + // The log line is the ONLY trace this message will ever leave: no ack, no + // DLQ, no redelivery. So it has to name the session whose invalidation was + // lost and the cause. + if got := err.Error(); !strings.Contains(got, "sess-1") || !strings.Contains(got, "boom") { + t.Errorf("error %q should name the session and the cause", got) + } +} + +// TestBindingSubscriberPanicDoesNotStopTheSubscription defends the guard's +// consequence for the plane: one broken subscriber must cost exactly its own +// message. The panicking callback runs on the dispatcher goroutine shared by +// every delivery on this subscription, so without the recover the first bad +// change would take down the process — and with a recover but no continuation +// it would still wedge the subscription and stop invalidating every OTHER +// session, which on a cache plane means serving stale bindings indefinitely. +// +// The gate is the next valid change arriving, which is the only observation +// that separates "recovered and carried on" from "recovered and died quietly". +func TestBindingSubscriberPanicDoesNotStopTheSubscription(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{Log: quietLogger(t)}) + + good := make(chan BindingChange, 1) + unsub, err := f.SubscribeBindingChanges(ctx, func(b BindingChange) { + if b.SessionID == "sess-poison" { + panic("binding-change subscriber is broken") + } + good <- b + }) + if err != nil { + t.Fatalf("SubscribeBindingChanges: %v", err) + } + defer unsub() + + poison := BindingChange{Tenant: "tenant-a", SessionID: "sess-poison", Op: BindingUnbound} + if err := f.PublishBindingChange(ctx, "tenant-a", poison); err != nil { + t.Fatalf("PublishBindingChange poison: %v", err) + } + // One publisher, one connection: core NATS preserves order, so the poison + // change is dispatched before this one. The good change arriving therefore + // proves the dispatcher survived the panic rather than proving it raced + // ahead of it. + want := BindingChange{Tenant: "tenant-a", SessionID: "sess-ok", Op: BindingBound} + if err := f.PublishBindingChange(ctx, "tenant-a", want); err != nil { + t.Fatalf("PublishBindingChange good: %v", err) + } + if got := recvBindingChange(t, good); got != want { + t.Fatalf("delivered %+v, want %+v — a panicking subscriber must not stop invalidation for every other session", got, want) + } +} diff --git a/go/internal/fabric/subjects.go b/go/internal/fabric/subjects.go index 733cc705..28491a8a 100644 --- a/go/internal/fabric/subjects.go +++ b/go/internal/fabric/subjects.go @@ -22,6 +22,28 @@ const ( // instance (§Q3, "delivery queue groups": the three-hop model's hop 2). RunnerEventsQueue = "compass-runner-events" + // routingBindingPrefix roots the binding-invalidation plane (§T4): + // compass.routing.binding., built by RoutingBindingSubject. + // + // The literal "binding" precedes the tenant token DELIBERATELY, and the + // token order is a correctness requirement rather than a style choice. The + // COMPASS_COMMS stream captures commsStreamSubjects — compass.*.comms.* — + // which matches any four-token subject whose THIRD token is "comms". Under + // the other order, compass.routing..binding, a tenant literally + // named "comms" yields compass.routing.comms.binding, which that wildcard + // captures. + // + // A stream is an ordinary subscriber in the account's sublist, so that + // capture is ADDITIVE, not exclusive: the hubs would still receive the + // message. The harm is that a best-effort core-NATS invalidation would ALSO + // be persisted into a durable stream specified to carry only EventRefs, + // burning its file storage and MaxAge budget with traffic that no comms + // consumer can use (every 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 the comms wildcard + // requires token 3 == "comms" and this grammar's token 3 is never variable. + routingBindingPrefix = subjectPrefix + ".routing.binding" + // DLQSubject is where a comms event that exhausted its delivery attempts is // parked. JetStream has no native dead-letter queue, so the fabric // implements the app-level pattern: publish to this subject, then Term() the @@ -124,6 +146,37 @@ func RunnerEventsSubject() string { return subjectPrefix + ".runner.events" } +// RoutingBindingSubject builds the binding-invalidation subject for one tenant: +// compass.routing.binding.. It returns an error if tenant is not a +// valid single subject token, for the same reason CommsSubject does — a tenant +// id carrying a reserved character is an upstream bug, and rewriting it would +// publish an invalidation nobody is subscribed to (see ValidSubjectToken). +// +// The tenant is the LAST token, not the third: see routingBindingPrefix for why +// that ordering is what keeps the subject out of the COMPASS_COMMS stream's +// compass.*.comms.* capture for every possible tenant value. +func RoutingBindingSubject(tenant string) (string, error) { + if err := ValidSubjectToken("tenant", tenant); err != nil { + return "", err + } + return routingBindingPrefix + "." + tenant, nil +} + +// RoutingBindingWildcardSubject is the TENANT-WILDCARD binding-invalidation +// subject: compass.routing.binding.*. It takes no token and cannot fail. +// +// Subscribe-side only, and it is the only subject the routing plane's read side +// uses: a Server's binding cache holds entries for every tenant it has resolved +// a session for, so one subscription across all tenants is what the cache +// needs, and a per-tenant subscribe would make tenant creation a NATS operation +// (the same argument CommsWildcardSubject makes for the delivery consumer). +// Publish derives its subject from the change's tenant via +// RoutingBindingSubject, and BindingChange.valid rejects a "*" tenant, so no +// publish can ever target this subject. +func RoutingBindingWildcardSubject() string { + return routingBindingPrefix + "." + wildcardToken +} + // ClientSubject builds the per-connection delivery subject for a live client: // client.. Deliberately outside the "compass." root — the frozen // grammar names it "client." (§T3), and it is not a diff --git a/go/internal/fabric/subjects_test.go b/go/internal/fabric/subjects_test.go index 41b1e6a5..f8a62b52 100644 --- a/go/internal/fabric/subjects_test.go +++ b/go/internal/fabric/subjects_test.go @@ -6,10 +6,10 @@ import ( ) // TestSubjectBuilders defends the frozen grammar itself: every subject a -// consumer will ever publish or subscribe to comes from these four builders, so -// a drift in any one of them silently re-wires a plane. The exact strings are -// asserted, not a pattern — the grammar is the contract other tasks build -// against (SUBJECTS.md). +// consumer will ever publish or subscribe to comes from a builder in +// subjects.go, so a drift in any one of them silently re-wires a plane. The +// exact strings are asserted, not a pattern — the grammar is the contract other +// tasks build against (SUBJECTS.md). func TestSubjectBuilders(t *testing.T) { t.Parallel() @@ -59,6 +59,27 @@ func TestSubjectBuilders(t *testing.T) { } }) + // The routing plane's two builders. Both the publisher + // (PublishBindingChange) and the subscriber (SubscribeBindingChanges) + // derive their subject from the one routingBindingPrefix constant, so every + // integration test on this seam is a round-trip through the same string and + // stays green after a typo in it — while the wire grammar silently diverges + // from SUBJECTS.md and from any other process on the bus. Pinning the + // literals is the only thing that catches that. + t.Run("routing binding", func(t *testing.T) { + t.Parallel() + got, err := RoutingBindingSubject("t1") + if err != nil { + t.Fatalf("RoutingBindingSubject: %v", err) + } + if want := "compass.routing.binding.t1"; got != want { + t.Fatalf("RoutingBindingSubject = %q, want %q", got, want) + } + if got, want := RoutingBindingWildcardSubject(), "compass.routing.binding.*"; got != want { + t.Fatalf("RoutingBindingWildcardSubject = %q, want %q", got, want) + } + }) + t.Run("client", func(t *testing.T) { t.Parallel() got, err := ClientSubject("sess-42") @@ -253,6 +274,38 @@ func TestDLQSubjectIsOutsideTheCommsStream(t *testing.T) { } } +// TestBindingSubjectPutsTheLiteralBeforeTheTenant defends the token ORDER of +// the routing plane, structurally and without a live server. The COMPASS_COMMS +// stream captures compass.*.comms.* — any four-token subject whose token 2 is +// "comms" — so the rejected grammar compass.routing..binding would put +// a tenant literally named "comms" inside that stream. Keeping the literal +// "binding" at the VARIABLE-FREE index 2 makes capture impossible for every +// tenant value rather than for the ones a test happens to try. +// +// TestBindingSubjectIsNotCapturedByTheCommsStream proves the same thing through +// a real server on the one degenerate tenant; this pins the order itself, so a +// reordered grammar fails here even if the integration test's tenant no longer +// happens to be the colliding one. +func TestBindingSubjectPutsTheLiteralBeforeTheTenant(t *testing.T) { + t.Parallel() + subject, err := RoutingBindingSubject("comms") + if err != nil { + t.Fatalf("RoutingBindingSubject: %v", err) + } + tokens := strings.Split(subject, ".") + streamTokens := strings.Split(commsStreamSubjects, ".") + if len(tokens) != len(streamTokens) { + t.Fatalf("RoutingBindingSubject(%q) = %q has %d tokens, want %d — this test compares it against %q token-for-token", + "comms", subject, len(tokens), len(streamTokens), commsStreamSubjects) + } + // Index 2 is where the stream demands the literal "comms", and it is the + // only index of this grammar that carries no caller-supplied value. + if got, want := tokens[2], "binding"; got != want { + t.Fatalf("RoutingBindingSubject(%q) = %q: token 2 is %q, want the literal %q — a variable token there lets a tenant named %q be captured by %q", + "comms", subject, got, want, streamTokens[2], commsStreamSubjects) + } +} + // TestValidCommsSubjectRejectsMalformed defends the whole-subject guard that // replaced the head-only one on the Subscribe path. Every case below is rooted // at "compass", so a first-token check passed all of them — and a subject the