diff --git a/docs/designs/infra/runtime/compass-managed-multitenancy/design.md b/docs/designs/infra/runtime/compass-managed-multitenancy/design.md index f26b0de7..dedc1eb8 100644 --- a/docs/designs/infra/runtime/compass-managed-multitenancy/design.md +++ b/docs/designs/infra/runtime/compass-managed-multitenancy/design.md @@ -770,9 +770,12 @@ the async command-push and event fan-in ride `RunnerFabric`. (`go/internal/runnerhub/hub.go:925-938`), and `github.com/nats-io/nats.go`. Produces: `package fabric` with - `type EventFabric interface { Publish(ctx context.Context, subject string, ref EventRef) error; Subscribe(ctx context.Context, subject string, fn func(EventRef)) (Unsubscribe, error) }` + `type EventFabric interface { Publish(ctx context.Context, subject string, ref EventRef) error; Subscribe(ctx context.Context, subject string, fn func(EventRef)) (Unsubscribe, error); SubscribeKind(ctx context.Context, kind EventKind, fn func(EventRef)) (Unsubscribe, error) }` where `EventRef` is a compact reference (event kind + row id + tenant), - never a payload copy — subscribers re-read Postgres; + never a payload copy — subscribers re-read Postgres; `SubscribeKind` is the + tenant-wildcard read side (`compass.*.comms.`, one durable queue-group + consumer across every tenant) the per-Server delivery singleton needs, while + `Publish` stays per-tenant and concrete; `type RunnerFabric interface { SendCommand(ctx context.Context, runnerID string, cmd *compassv1internal.SessionsResponse) error; Events(ctx context.Context) (<-chan RunnerEvent, error) }`; `fabric.New(cfg Config) (*Fabric, error)` where `Config` carries the NATS connection (`nats.Connect(url, opts...)`) — one implementation, one client, @@ -782,7 +785,8 @@ the async command-push and event fan-in ride `RunnerFabric`. Also produces: the JetStream delivery stream (durable at-least-once fan-out, `sync_interval: 100ms`, explicit acks, `max_deliver` + DLQ subject); and the subject-naming doc - (`compass..comms.`, `compass.runner..cmd`, + (`compass..comms.` plus its subscribe-side + `compass.*.comms.`, `compass.runner..cmd`, `compass.runner.events` queue-grouped, `client.` per-connection delivery) as a supporting file beside this record. Gate instrumentation: the delivery-backlog OTel emitter (scale-out gate signal) rides this task. diff --git a/go/internal/fabric/SUBJECTS.md b/go/internal/fabric/SUBJECTS.md index 8e0079dd..1f447317 100644 --- a/go/internal/fabric/SUBJECTS.md +++ b/go/internal/fabric/SUBJECTS.md @@ -11,6 +11,7 @@ this file is the operational restatement that later tasks build against, and | Grammar | Plane | Builder | Direction | | --- | --- | --- | --- | | `compass..comms.` | JetStream | `CommsSubject(tenant, kind)` | Server → Servers (comms/delivery fan-out) | +| `compass.*.comms.` | JetStream | `CommsWildcardSubject(kind)` | Servers → one delivery consumer (cross-tenant fan-in, **subscribe-side only**) | | `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 | @@ -20,6 +21,37 @@ this file is the operational restatement that later tasks build against, and grammar names it that way, and it must not be captured by the comms stream's subject wildcard. +### The tenant-wildcard subscribe: `compass.*.comms.` + +Publish is always per-tenant and concrete. The read side has a second entry +point, `EventFabric.SubscribeKind(ctx, kind, fn)`, which subscribes on +`compass.*.comms.` — one kind, every tenant. The T3 delivery consumer is +a per-Server **singleton** serving all tenants, so a per-tenant subscribe would +need one consumer per tenant created at tenant-creation time; the wildcard gives +it one durable queue-group consumer instead, and tenant creation stays a +Postgres insert. + +- **The wildcard is on the tenant token only.** The kind stays concrete and is + validated by `ValidSubjectToken`. A wildcard kind would put all seven comms + kinds on the delivery consumer, waking it (and its Postgres re-read) for every + unrelated write. +- **No stream-config change.** `Subjects` is already `compass.*.comms.*`, which + captures this subject by construction; JetStream accepts a wildcard + `FilterSubject` on a durable consumer. +- **Its own durable consumer.** `Durable` is `comms-` + sha256(subject), so the + wildcard subject hashes to a name distinct from every concrete-tenant + consumer. Shared and durable as usual: each matching event is claimed by + exactly one Server instance. Wildcard and concrete consumers on the same kind + are independent durables, so an event matching both is delivered once to each; + a migration introducing `SubscribeKind` must retire the concrete subscribes + rather than double-handle events. +- **`Subscribe` stays concrete-only.** `validCommsSubject` still rejects a `*` + token, so the wildcard is reachable only through `SubscribeKind`'s own + validated builder — a caller cannot hand-write a cross-tenant subject. +- **Publish cannot target it.** `Publish` derives its subject from the ref via + `CommsSubject`, and `EventRef.valid` rejects a `*` tenant, so a wildcard + publish is impossible rather than merely discouraged. + ### Token validation: reject, never sanitize NATS reserves `.` (token separator), `*` and `>` (wildcards), and rejects @@ -123,9 +155,10 @@ park, republish the raw payload to `compass.dlq.comms` and then a DLQ publish that needed a stream would need a DLQ of its own. - `Term` is issued **even if the DLQ publish fails**, with both failures logged: a poison message redelivering forever is the worse outcome. -- Headers on the parked message: `Compass-Original-Subject` (the subject it was - delivered on) and `Compass-Park-Reason` (the error), so an operator reading the - DLQ needs no log correlation. +- Headers on the parked message: `Compass-Original-Subject` (the concrete + subject the message was delivered on, even for a wildcard (`SubscribeKind`) + consumer, so it always names the tenant) and `Compass-Park-Reason` (the + error), so an operator reading the DLQ needs no log correlation. The attempt count comes from the message's server-side metadata rather than any local counter, which is what makes the budget hold across Server instances and diff --git a/go/internal/fabric/event_fabric.go b/go/internal/fabric/event_fabric.go index 9caf00d7..38659db0 100644 --- a/go/internal/fabric/event_fabric.go +++ b/go/internal/fabric/event_fabric.go @@ -80,6 +80,51 @@ func (f *Fabric) Subscribe(ctx context.Context, subject string, fn func(EventRef if err := validCommsSubject(subject); err != nil { return nil, err } + return f.subscribeSubject(ctx, subject, fn) +} + +// SubscribeKind drives fn for every event of one kind ACROSS EVERY TENANT, +// until the returned Unsubscribe is called, ctx is done, or the Fabric is +// closed. It is the delivery plane's cross-tenant fan-in path: the delivery +// consumer is a per-Server singleton serving all tenants, while each event is +// published on a concrete compass..comms., so the one consumer +// subscribes on the tenant-wildcard subject CommsWildcardSubject builds. +// +// Identical in every other respect to Subscribe — one DURABLE queue-group +// consumer (durableName hashes the wildcard subject to its own name, distinct +// from any concrete-tenant consumer, so each matching event is claimed by +// exactly one Server instance), the same explicit ack / Nak-to-MaxDeliver / +// park-on-DLQSubject semantics, and the same drain on all three teardown +// paths. Wildcard and concrete consumers are independent durables; see +// SUBJECTS.md's "Its own durable consumer" property when migrating callers. +// +// The wildcard is on the TENANT token only: kind is concrete and validated, so +// a SubscribeKind(KindMessagePosted) receives message_posted for every tenant +// and nothing else. Subscribe keeps its strict concrete-subject grammar — a +// wildcard subject cannot be reached through it. +func (f *Fabric) SubscribeKind(ctx context.Context, kind EventKind, fn func(EventRef)) (Unsubscribe, error) { + if err := f.checkOpen(); err != nil { + return nil, err + } + if fn == nil { + return nil, fmt.Errorf("fabric: SubscribeKind(%q) requires a callback", kind) + } + subject, err := CommsWildcardSubject(kind) + if err != nil { + return nil, err + } + return f.subscribeSubject(ctx, subject, fn) +} + +// subscribeSubject is the shared body of Subscribe and SubscribeKind: it +// registers the durable consumer on an ALREADY-VALIDATED subject and wires its +// teardown. Split out so each public entry point owns its own subject +// validation — Subscribe's strict concrete-only grammar, SubscribeKind's +// tenant-wildcard builder — and neither can reach the other's. +// +// It performs no validation of its own: subject must come from +// validCommsSubject or CommsWildcardSubject. +func (f *Fabric) subscribeSubject(ctx context.Context, subject string, fn func(EventRef)) (Unsubscribe, error) { stream, err := f.ensureStream(ctx) if err != nil { return nil, err @@ -93,7 +138,7 @@ func (f *Fabric) Subscribe(ctx context.Context, subject string, fn func(EventRef } cc, err := cons.Consume(func(msg jetstream.Msg) { - f.handleEvent(ctx, subject, msg, fn) + f.handleEvent(ctx, msg, fn) }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { // Transient pull errors are the library's to retry; surfacing them is // the only thing this side can do, and swallowing them would hide a @@ -143,22 +188,45 @@ func (f *Fabric) Subscribe(ctx context.Context, subject string, fn func(EventRef // handleEvent runs one delivery: decode, invoke fn under a panic guard, then ack // or park. Split out of Subscribe so the ack/park decision is readable on its // own. -func (f *Fabric) handleEvent(ctx context.Context, subject string, msg jetstream.Msg, fn func(EventRef)) { +func (f *Fabric) handleEvent(ctx context.Context, msg jetstream.Msg, fn func(EventRef)) { ref, decodeErr := decodeEventRef(msg.Data()) if decodeErr != nil { // Unparseable: no number of redeliveries changes the bytes. - f.park(ctx, subject, msg, decodeErr) + f.park(ctx, msg, decodeErr) + return + } + // The ref's tenant must match the subject it arrived on. Publish enforces + // this from the write side, but Publish is not the only writer the stream + // can have: COMPASS_COMMS is shared and the server carries no per-tenant + // authorization yet (OQ-3), so a client reaching the client port can put + // arbitrary bytes on any comms subject. Without this check a ref naming + // tenant-a delivered on tenant-b's subject would hand a tenant-b-scoped + // subscriber a tenant-a row id, and EventRef's contract directs that + // subscriber to re-read under ref.Tenant WITHOUT consulting the subject + // (eventref.go) — so the payload would be the only tenant discriminator. + // SubscribeKind makes that the primary path: its consumer spans every + // tenant, leaving ref.Tenant as the sole scope for a delivery. + // + // A mismatch is as unprocessable as an undecodable payload — no redelivery + // changes it — so it parks rather than retries. + want, subjErr := CommsSubject(ref.Tenant, ref.Kind) + if subjErr != nil { + f.park(ctx, msg, subjErr) + return + } + if got := msg.Subject(); got != want { + f.park(ctx, msg, fmt.Errorf("fabric: event ref %s/%s names subject %q but was delivered on %q", ref.Tenant, ref.Kind, want, got)) return } if err := invoke(fn, ref); err != nil { - f.retryOrPark(ctx, subject, msg, err) + f.retryOrPark(ctx, msg, err) return } if err := msg.Ack(); err != nil { // The event WAS processed; a lost ack costs a redelivery, which the // subscriber's Postgres re-read makes idempotent. Log, never park. f.log.WarnContext(ctx, "fabric: acking delivered event failed; it will be redelivered", - "subject", subject, "kind", string(ref.Kind), "row_id", ref.RowID, "error", err) + "subject", msg.Subject(), "kind", string(ref.Kind), "row_id", ref.RowID, "error", err) } } @@ -180,25 +248,25 @@ func invoke(fn func(EventRef), ref EventRef) (err error) { // attempt budget is spent. Reading NumDelivered from the message metadata (not a // local counter) is what makes the budget hold across Server instances and // restarts — the count is the server's. -func (f *Fabric) retryOrPark(ctx context.Context, subject string, msg jetstream.Msg, cause error) { +func (f *Fabric) retryOrPark(ctx context.Context, msg jetstream.Msg, cause error) { md, err := msg.Metadata() if err != nil { // No metadata means no attempt count, so the budget cannot be enforced; // park rather than risk redelivering a poison message forever. - f.park(ctx, subject, msg, fmt.Errorf("%w (and its metadata was unreadable: %w)", cause, err)) + f.park(ctx, msg, fmt.Errorf("%w (and its metadata was unreadable: %w)", cause, err)) return } if md.NumDelivered >= f.cfg.deliveryBudget() { - f.park(ctx, subject, msg, fmt.Errorf("%w (after %d delivery attempts)", cause, md.NumDelivered)) + f.park(ctx, msg, fmt.Errorf("%w (after %d delivery attempts)", cause, md.NumDelivered)) return } f.log.WarnContext(ctx, "fabric: event handling failed; redelivering", - "subject", subject, "attempt", md.NumDelivered, "max_deliver", f.cfg.maxDeliver(), "error", cause) + "subject", msg.Subject(), "attempt", md.NumDelivered, "max_deliver", f.cfg.maxDeliver(), "error", cause) if err := msg.Nak(); err != nil { // AckWait still expires and redelivers, so this is a latency cost, not // a lost event. f.log.WarnContext(ctx, "fabric: nak failed; redelivery waits for ack_wait", - "subject", subject, "error", err) + "subject", msg.Subject(), "error", err) } } @@ -217,22 +285,22 @@ func (f *Fabric) retryOrPark(ctx context.Context, subject string, msg jetstream. // // The reason on the wire is sanitized and bounded (see sanitizeReason); the // full cause goes to the log, which has no wire limit. -func (f *Fabric) park(ctx context.Context, subject string, msg jetstream.Msg, cause error) { +func (f *Fabric) park(ctx context.Context, msg jetstream.Msg, cause error) { f.log.ErrorContext(ctx, "fabric: parking event on the dlq", - "subject", subject, "dlq_subject", DLQSubject, "error", cause) + "subject", msg.Subject(), "dlq_subject", DLQSubject, "error", cause) dlq := nats.NewMsg(DLQSubject) dlq.Data = msg.Data() - dlq.Header.Set(dlqHeaderSubject, subject) + dlq.Header.Set(dlqHeaderSubject, msg.Subject()) reason := sanitizeReason(cause.Error()) dlq.Header.Set(dlqHeaderReason, reason) if err := f.nc.PublishMsg(dlq); err != nil { f.log.ErrorContext(ctx, "fabric: publishing to the dlq failed; terminating the message anyway", - "subject", subject, "error", err) + "subject", msg.Subject(), "error", err) } if err := msg.TermWithReason(reason); err != nil { f.log.ErrorContext(ctx, "fabric: terminating a parked message failed; it may redeliver until max_deliver", - "subject", subject, "error", err) + "subject", msg.Subject(), "error", err) } } diff --git a/go/internal/fabric/event_fabric_test.go b/go/internal/fabric/event_fabric_test.go index 40e87019..467687c8 100644 --- a/go/internal/fabric/event_fabric_test.go +++ b/go/internal/fabric/event_fabric_test.go @@ -126,6 +126,61 @@ func TestEventFabricFiltersBySubject(t *testing.T) { } } +// TestEventFabricConcreteAndWildcardConsumersCoexist proves that concrete and +// tenant-wildcard subscriptions are independent durables on one fabric. The +// concrete filter must exclude t2, while the wildcard receives both tenants. +func TestEventFabricConcreteAndWildcardConsumersCoexist(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + concreteSubject, err := CommsSubject("t1", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + wildcardSubject, err := CommsWildcardSubject(KindMessagePosted) + if err != nil { + t.Fatalf("CommsWildcardSubject: %v", err) + } + if durableName(wildcardSubject) == durableName(concreteSubject) { + t.Fatalf("wildcard and concrete durable names collide: %q", durableName(wildcardSubject)) + } + concrete := make(chan EventRef, 4) + wildcard := make(chan EventRef, 4) + unsubConcrete, err := f.Subscribe(ctx, concreteSubject, func(r EventRef) { concrete <- r }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsubConcrete() + unsubWildcard, err := f.SubscribeKind(ctx, KindMessagePosted, func(r EventRef) { wildcard <- r }) + if err != nil { + t.Fatalf("SubscribeKind: %v", err) + } + defer unsubWildcard() + t1 := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-t1"} + t2 := EventRef{Tenant: "t2", Kind: KindMessagePosted, RowID: "msg-t2"} + sentinel := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-sentinel"} + for _, ref := range []EventRef{t1, t2, sentinel} { + subject, subjectErr := CommsSubject(ref.Tenant, ref.Kind) + if subjectErr != nil { + t.Fatalf("CommsSubject(%s): %v", ref.Tenant, subjectErr) + } + if err := f.Publish(ctx, subject, ref); err != nil { + t.Fatalf("Publish %s: %v", ref.RowID, err) + } + } + if got := recvRef(t, concrete); got != t1 { + t.Fatalf("concrete first delivery = %+v, want %+v (t2 leaked)", got, t1) + } + if got := recvRef(t, concrete); got != sentinel { + t.Fatalf("concrete second delivery = %+v, want sentinel %+v (filter leaked)", got, sentinel) + } + first, second := recvRef(t, wildcard), recvRef(t, wildcard) + seen := map[string]bool{first.RowID: true, second.RowID: true} + if !seen[t1.RowID] || !seen[t2.RowID] || len(seen) != 2 { + t.Fatalf("wildcard deliveries = %v, want t1 and t2", seen) + } +} + // TestUnsubscribeStopsDelivery defends that Unsubscribe actually stops the // consume context. A leaked consumer would keep draining the shared durable // consumer after its owner is gone — events claimed by nobody, which on a @@ -375,6 +430,75 @@ func TestPoisonMessageParksOnDLQ(t *testing.T) { } } +// TestWildcardConsumerParksWithConcreteSubject defends the DLQ's provenance for a +// wildcard consumer. park writes msg.Subject() — the CONCRETE delivered subject — +// not the consumer's filter, so a message parked by a SubscribeKind consumer still +// names its tenant. With the filter subject the header would read +// compass.*.comms. and an operator reading the DLQ could not tell which +// tenant the poison event belonged to. +func TestWildcardConsumerParksWithConcreteSubject(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url, MaxDeliver: 2, Log: quietLogger(t)}) + + raw, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(raw.Close) + dlq, err := raw.SubscribeSync(DLQSubject) + if err != nil { + t.Fatalf("SubscribeSync(%q): %v", DLQSubject, err) + } + if err := raw.FlushWithContext(ctx); err != nil { + t.Fatalf("flushing the dlq subscription: %v", err) + } + + concrete, err := CommsSubject("t1", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(EventRef) { + panic("subscriber is broken") + }) + if err != nil { + t.Fatalf("SubscribeKind: %v", err) + } + defer unsub() + + poison := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-poison"} + if err := f.Publish(ctx, concrete, poison); err != nil { + t.Fatalf("Publish: %v", err) + } + + msg, err := dlq.NextMsgWithContext(ctx) + if err != nil { + t.Fatalf("waiting for the parked message on %q: %v", DLQSubject, err) + } + parked, err := decodeEventRef(msg.Data) + if err != nil { + t.Fatalf("the parked payload must be the original event: %v", err) + } + if parked != poison { + t.Fatalf("parked %+v, want %+v", parked, poison) + } + got := msg.Header.Get(dlqHeaderSubject) + if got != concrete { + t.Errorf("park header %s = %q, want concrete subject %q", dlqHeaderSubject, got, concrete) + } + wildcard, err := CommsWildcardSubject(KindMessagePosted) + if err != nil { + t.Fatalf("CommsWildcardSubject: %v", err) + } + if got == wildcard { + t.Errorf("park header %s used wildcard subject %q", dlqHeaderSubject, wildcard) + } + if msg.Header.Get(dlqHeaderReason) == "" { + t.Errorf("park header %s is empty; an operator reading the dlq has no reason", dlqHeaderReason) + } +} + // TestSubscriberPanicDoesNotBlockOtherEvents defends the panic guard's // consequence for throughput: one broken event must not wedge the subject. The // poison event exhausts its budget and parks, and the next event is delivered — @@ -863,9 +987,11 @@ func TestSubscribeWatchdogExitsOnClose(t *testing.T) { t.Fatalf("CommsSubject: %v", err) } - // Every goroutine spawned inside Subscribe carries this in its stack, so a - // residual watchdog shows up as a count that never falls back to baseline. - const marker = "fabric.(*Fabric).Subscribe.func" + // Every goroutine spawned on the subscribe path carries this in its stack, + // so a residual watchdog shows up as a count that never falls back to + // baseline. The watchdog lives in subscribeSubject, the body Subscribe and + // SubscribeKind share, so this marker covers both entry points. + const marker = "fabric.(*Fabric).subscribeSubject.func" baseline := countGoroutinesWith(t, marker) // Rooted at context.Background() because this is a test root, and an @@ -906,3 +1032,283 @@ func countGoroutinesWith(t *testing.T, marker string) int { buf = make([]byte, 2*len(buf)) } } + +// TestSubscribeKindReceivesEveryTenant is the load-bearing test for the +// tenant-wildcard subscribe. The T3 delivery consumer is a per-Server +// singleton serving every tenant, while each event is published on its own +// concrete compass..comms.; if the wildcard captured only some +// tenants, delivery for the rest would silently stop and only the cursor sweep +// would recover it. One SubscribeKind must see BOTH tenants' events with the +// tenant field intact — intact because the subscriber re-reads Postgres under +// that tenant, so a lost or wrong tenant is a cross-tenant read. +func TestSubscribeKindReceivesEveryTenant(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + got := make(chan EventRef, 4) + unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(r EventRef) { got <- r }) + if err != nil { + t.Fatalf("SubscribeKind: %v", err) + } + defer unsub() + + want := map[string]EventRef{ + "t1": {Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-t1"}, + "t2": {Tenant: "t2", Kind: KindMessagePosted, RowID: "msg-t2"}, + } + for tenant, ref := range want { + subject, err := CommsSubject(tenant, KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject(%q): %v", tenant, err) + } + if err := f.Publish(ctx, subject, ref); err != nil { + t.Fatalf("Publish for %q: %v", tenant, err) + } + } + + // Two distinct stream subjects, so their relative delivery order is not + // guaranteed; collect both and compare as a set. + seen := make(map[string]EventRef, len(want)) + for range want { + ref := recvRef(t, got) + if _, dup := seen[ref.Tenant]; dup { + t.Fatalf("tenant %q delivered twice; got %+v", ref.Tenant, ref) + } + seen[ref.Tenant] = ref + } + for tenant, wantRef := range want { + gotRef, ok := seen[tenant] + if !ok { + t.Fatalf("tenant %q never reached the wildcard subscriber (got %+v)", tenant, seen) + } + if gotRef != wantRef { + t.Fatalf("tenant %q delivered %+v, want %+v", tenant, gotRef, wantRef) + } + } +} + +// TestSubscribeKindIsolatesKinds defends the half of the subject that is NOT +// wildcarded. The stream captures compass.*.comms.*, so the consumer's +// FilterSubject is the only thing keeping the other six kinds out — and a +// delivery consumer woken for every topic_upsert would do a Postgres re-read +// per unrelated write. +// +// Absence is proven by a positive gate, not a sleep: the foreign-kind event is +// published and acked into the stream FIRST, so if the kind filter leaked it +// would already be stored and deliverable when the message_posted sentinel +// arrives. +func TestSubscribeKindIsolatesKinds(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + got := make(chan EventRef, 4) + unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(r EventRef) { got <- r }) + if err != nil { + t.Fatalf("SubscribeKind: %v", err) + } + defer unsub() + + other := EventRef{Tenant: "t1", Kind: KindTopicUpserted, RowID: "topic-1"} + otherSubject, err := CommsSubject(other.Tenant, other.Kind) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + if err := f.Publish(ctx, otherSubject, other); err != nil { + t.Fatalf("Publish the foreign kind: %v", err) + } + + sentinel := EventRef{Tenant: "t2", Kind: KindMessagePosted, RowID: "msg-sentinel"} + sentinelSubject, err := CommsSubject(sentinel.Tenant, sentinel.Kind) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + if err := f.Publish(ctx, sentinelSubject, sentinel); err != nil { + t.Fatalf("Publish the sentinel: %v", err) + } + + if delivered := recvRef(t, got); delivered != sentinel { + t.Fatalf("delivered %+v, want the sentinel %+v — the wildcard leaked a %s event", + delivered, sentinel, other.Kind) + } +} + +// TestSubscribeKindRejectsBadInput defends the wildcard entry point's own +// guards. An invalid kind must fail at the builder rather than reach +// CreateOrUpdateConsumer, and a nil callback must be refused rather than +// panicking on the first delivery — the same contract Subscribe has. +func TestSubscribeKindRejectsBadInput(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + if _, err := f.SubscribeKind(ctx, KindMessagePosted, nil); err == nil { + t.Error("SubscribeKind with a nil callback = nil error, want a refusal") + } + if _, err := f.SubscribeKind(ctx, EventKind("bad.kind"), func(EventRef) {}); err == nil { + t.Error("SubscribeKind with a reserved-character kind = nil error, want a refusal") + } + // A wildcard kind would put all seven comms kinds on one consumer. + if _, err := f.SubscribeKind(ctx, EventKind("*"), func(EventRef) {}); err == nil { + t.Error("SubscribeKind with a wildcard kind = nil error, want a refusal") + } +} + +// TestForgedTenantRefIsParked defends the read side of the tenant invariant. +// Publish refuses a ref whose tenant disagrees with its subject, but Publish is +// not the only writer the shared stream can have: the server carries no +// per-tenant authorization yet, so this test bypasses Publish with a raw client +// exactly as a rogue publisher would. The subscriber must never see the ref — +// EventRef's contract tells it to re-read under ref.Tenant WITHOUT consulting +// the subject, so a delivered mismatch is a cross-tenant read. +// +// SubscribeKind is the entry point under test because its consumer spans every +// tenant, which leaves the payload's tenant field as the only scope a delivery +// carries. +func TestForgedTenantRefIsParked(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url, MaxDeliver: 1, Log: quietLogger(t)}) + + raw, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(raw.Close) + dlq, err := raw.SubscribeSync(DLQSubject) + if err != nil { + t.Fatalf("SubscribeSync(%q): %v", DLQSubject, err) + } + if err := raw.FlushWithContext(ctx); err != nil { + t.Fatalf("flushing the dlq subscription: %v", err) + } + + delivered := make(chan EventRef, 2) + unsub, err := f.SubscribeKind(ctx, KindMessagePosted, func(ref EventRef) { delivered <- ref }) + if err != nil { + t.Fatalf("SubscribeKind: %v", err) + } + defer unsub() + + // tenant-victim's subject carrying tenant-attacker's ref. Publish would + // reject this pairing outright, so it goes on the wire raw. + victimSubject, err := CommsSubject("tenant-victim", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + forged, err := EventRef{Tenant: "tenant-attacker", Kind: KindMessagePosted, RowID: "msg-forged"}.encode() + if err != nil { + t.Fatalf("encoding the forged ref: %v", err) + } + if err := raw.Publish(victimSubject, forged); err != nil { + t.Fatalf("raw Publish(%q): %v", victimSubject, err) + } + + // Positive gate on the negative assertion: a legitimate ref published + // AFTER the forgery must arrive, so the forgery has demonstrably had its + // chance to be delivered rather than merely not arriving yet. + goodSubject, err := CommsSubject("tenant-victim", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + good := EventRef{Tenant: "tenant-victim", Kind: KindMessagePosted, RowID: "msg-legit"} + if err := f.Publish(ctx, goodSubject, good); err != nil { + t.Fatalf("Publish: %v", err) + } + + got := <-delivered + if got.RowID != "msg-legit" { + t.Fatalf("first delivery = %s/%s, want the legitimate tenant-victim/msg-legit — the forged ref reached the subscriber", got.Tenant, got.RowID) + } + select { + case extra := <-delivered: + t.Fatalf("a second event reached the subscriber: %s/%s, want only the legitimate one", extra.Tenant, extra.RowID) + default: + } + + // The forgery must be parked, not silently dropped: an operator needs the + // concrete subject it was delivered on to know which tenant was targeted. + msg, err := dlq.NextMsgWithContext(ctx) + if err != nil { + t.Fatalf("waiting for the parked forgery on %q: %v", DLQSubject, err) + } + if subj := msg.Header.Get(dlqHeaderSubject); subj != victimSubject { + t.Errorf("parked subject header = %q, want the concrete delivered subject %q", subj, victimSubject) + } +} + +// TestLegitimateRefsSurviveTheSubjectCrossCheck pins the symmetry the cross-check +// depends on. handleEvent parks a ref whose tenant disagrees with its delivered +// subject, and it decides that by rebuilding the subject with the same +// CommsSubject call Publish uses. That the two agree is currently emergent — two +// call sites deriving one string from the same two fields — rather than tested, +// and a one-sided change (normalizing, lowercasing or trimming the tenant token +// on one side only) would park every legitimate delivery on every tenant: a +// silent fleet-wide denial of delivery whose only signal is a filling DLQ. +// +// So this asserts the positive direction across the token space +// ValidSubjectToken actually admits, on a CONCRETE Subscribe — the +// highest-traffic path, and the one the forged-ref and wildcard tests do not +// cover. +func TestLegitimateRefsSurviveTheSubjectCrossCheck(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url, MaxDeliver: 1, Log: quietLogger(t)}) + + raw, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(raw.Close) + dlq, err := raw.SubscribeSync(DLQSubject) + if err != nil { + t.Fatalf("SubscribeSync(%q): %v", DLQSubject, err) + } + if err := raw.FlushWithContext(ctx); err != nil { + t.Fatalf("flushing the dlq subscription: %v", err) + } + + // Every form ValidSubjectToken admits: it rejects rather than sanitizes, so + // case and separators other than . * > are all legal and must round-trip + // byte-for-byte through publish and delivery. + tenants := []string{"t1", "tenant-with-dashes", "tenant_with_underscores", "MixedCaseTenant", "0f9aaa31-048f-459b-b235-99fcb6e50690"} + + for _, tenant := range tenants { + subject, err := CommsSubject(tenant, KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject(%q): %v", tenant, err) + } + delivered := make(chan EventRef, 1) + unsub, err := f.Subscribe(ctx, subject, func(ref EventRef) { delivered <- ref }) + if err != nil { + t.Fatalf("Subscribe(%q): %v", subject, err) + } + ref := EventRef{Tenant: tenant, Kind: KindMessagePosted, RowID: "msg-" + tenant} + if err := f.Publish(ctx, subject, ref); err != nil { + t.Fatalf("Publish(%q): %v", subject, err) + } + select { + case got := <-delivered: + if got.Tenant != tenant || got.RowID != ref.RowID { + t.Errorf("delivered %s/%s, want %s/%s", got.Tenant, got.RowID, tenant, ref.RowID) + } + case <-ctx.Done(): + t.Fatalf("tenant %q: a legitimate event was never delivered — the subject cross-check is rejecting well-formed traffic", tenant) + } + unsub() + } + + // Nothing legitimate may reach the DLQ. Every delivery above is a completed + // round-trip through the cross-check, so a park would already have been + // published — this is a check on a settled state, not a poll, and the short + // deadline only bounds the "nothing arrived" case. + dlqCtx, cancel := context.WithTimeout(ctx, time.Second) + defer cancel() + if msg, err := dlq.NextMsgWithContext(dlqCtx); err == nil { + t.Errorf("a legitimate event was parked on %q: subject header %q, reason %q", + DLQSubject, msg.Header.Get(dlqHeaderSubject), msg.Header.Get(dlqHeaderReason)) + } +} diff --git a/go/internal/fabric/fabric.go b/go/internal/fabric/fabric.go index 94e670a0..97a5fec3 100644 --- a/go/internal/fabric/fabric.go +++ b/go/internal/fabric/fabric.go @@ -23,6 +23,11 @@ type Unsubscribe func() type EventFabric interface { Publish(ctx context.Context, subject string, ref EventRef) error Subscribe(ctx context.Context, subject string, fn func(EventRef)) (Unsubscribe, error) + // SubscribeKind is the tenant-wildcard read side: one durable queue-group + // consumer receiving one kind across EVERY tenant, which is what the + // per-Server delivery singleton needs (§T3). Publish stays per-tenant and + // concrete. + SubscribeKind(ctx context.Context, kind EventKind, fn func(EventRef)) (Unsubscribe, error) } // RunnerFabric is the Server↔Runner async seam (frozen, §T3): per-Runner @@ -257,6 +262,20 @@ func New(cfg Config) (*Fabric, error) { nats.ReconnectHandler(func(nc *nats.Conn) { log.Info("fabric: nats reconnected", "url", nc.ConnectedUrl()) }), + // The Runner plane's safety argument is that a stalled receiver is + // dropped AND reported, with the cursor sweep recovering what was + // dropped (see RunnerEventBuffer). nats.go reports a full channel + // subscription through the async error callback, and the default + // options install none — so without this the one lossy path in the + // fabric drops events with no log line, no metric and no error. + nats.ErrorHandler(func(_ *nats.Conn, sub *nats.Subscription, err error) { + if sub == nil { + log.Warn("fabric: nats async error", "error", err) + return + } + log.Warn("fabric: nats async error; a slow consumer drops events until it keeps up", + "subject", sub.Subject, "dropped", subDropped(sub), "error", err) + }), }, cfg.Options...) nc, err := nats.Connect(cfg.URL, opts...) @@ -272,6 +291,18 @@ func New(cfg Config) (*Fabric, error) { return f, nil } +// subDropped reports a subscription's dropped-message count for logging. +// Subscription.Dropped returns an error once the subscription is invalid, which +// is exactly the moment an error handler may run — so a failed read degrades to +// -1 rather than losing the log line that names the slow consumer. +func subDropped(sub *nats.Subscription) int { + n, err := sub.Dropped() + if err != nil { + return -1 + } + return n +} + // Close drains and closes the connection: Drain flushes pending publishes and // lets in-flight subscription callbacks finish before the socket goes away, so a // shutdown does not lose an already-published event. It also tears down the diff --git a/go/internal/fabric/fabric_test.go b/go/internal/fabric/fabric_test.go index 57cf220f..e99d9267 100644 --- a/go/internal/fabric/fabric_test.go +++ b/go/internal/fabric/fabric_test.go @@ -5,6 +5,7 @@ import ( "errors" "log/slog" "strings" + "sync" "sync/atomic" "testing" "time" @@ -263,6 +264,9 @@ func TestCloseIsIdempotentAndFailsClosed(t *testing.T) { if _, err := f.Subscribe(ctx, subject, func(EventRef) {}); !errors.Is(err, errClosed) { t.Fatalf("Subscribe after Close: want errClosed, got %v", err) } + if _, err := f.SubscribeKind(ctx, KindMessagePosted, func(EventRef) {}); !errors.Is(err, errClosed) { + t.Fatalf("SubscribeKind after Close: want errClosed, got %v", err) + } if err := f.SendCommand(ctx, "r1", nil); !errors.Is(err, errClosed) { t.Fatalf("SendCommand after Close: want errClosed, got %v", err) } @@ -374,3 +378,93 @@ func (w testWriter) Write(p []byte) (int, error) { w.t.Logf("fabric log: %s", strings.TrimRight(string(p), "\n")) return len(p), nil } + +// capturingLogger records log output so a test can assert on it. quietLogger +// routes to t.Logf, which is right for noise but gives a test nothing to read. +type capturingLogger struct { + mu sync.Mutex + buf strings.Builder +} + +func (c *capturingLogger) Write(p []byte) (int, error) { + c.mu.Lock() + defer c.mu.Unlock() + return c.buf.Write(p) +} + +func (c *capturingLogger) String() string { + c.mu.Lock() + defer c.mu.Unlock() + return c.buf.String() +} + +// TestAsyncErrorHandlerReportsSlowConsumers pins the Runner plane's only report +// of its only lossy path. A full channel subscription is dropped-and-reported by +// contract, and the reporting half exists solely because New installs an async +// error handler — nats.go's defaults install none, and its own fallback writes +// to raw stderr rather than the fabric's logger. If that option is ever dropped, +// reordered behind a caller option, or broken, the runner plane silently returns +// to losing events invisibly, so the guarantee needs a test rather than trust. +// +// The handler is invoked directly rather than by simulating a stalled receiver: +// nats.go dispatches it on its own goroutine with no lock held, so calling it is +// faithful and carries no timing dependence. +func TestAsyncErrorHandlerReportsSlowConsumers(t *testing.T) { + t.Parallel() + logs := &capturingLogger{} + log := slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug})) + f := newFabric(t, Config{Log: log}) + + handler := f.nc.ErrorHandler() + if handler == nil { + t.Fatal("nc.ErrorHandler() = nil, want the fabric's handler — a slow consumer would drop events with no report") + } + + // The nil-subscription branch. Every connection-level error and the whole + // drain path pass nil, so this branch runs during teardown and would panic + // without its guard. + handler(f.nc, nil, nats.ErrSlowConsumer) + if out := logs.String(); !strings.Contains(out, nats.ErrSlowConsumer.Error()) { + t.Errorf("nil-subscription branch logged %q, want it to name the error", out) + } + + // The subscription branch: the subject must appear so an operator knows + // WHICH consumer is falling behind. + sub, err := f.nc.SubscribeSync("compass.test.slow") + if err != nil { + t.Fatalf("SubscribeSync: %v", err) + } + defer func() { + if err := sub.Unsubscribe(); err != nil { + t.Errorf("Unsubscribe: %v", err) + } + }() + handler(f.nc, sub, nats.ErrSlowConsumer) + if out := logs.String(); !strings.Contains(out, "compass.test.slow") { + t.Errorf("subscription branch logged %q, want it to name the subject", out) + } +} + +// TestCallerErrorHandlerWins defends the documented Config.Options precedence for +// the async error handler specifically. It is log-only, so a caller overriding it +// is legitimate — unlike the ClosedHandler, which Close deliberately avoids +// depending on so a caller cannot disarm the drain signal. +func TestCallerErrorHandlerWins(t *testing.T) { + t.Parallel() + var called atomic.Bool + f := newFabric(t, Config{ + Log: quietLogger(t), + Options: []nats.Option{ + nats.ErrorHandler(func(*nats.Conn, *nats.Subscription, error) { called.Store(true) }), + }, + }) + + handler := f.nc.ErrorHandler() + if handler == nil { + t.Fatal("nc.ErrorHandler() = nil, want the caller's handler") + } + handler(f.nc, nil, nats.ErrSlowConsumer) + if !called.Load() { + t.Error("the caller's ErrorHandler was not invoked; Config.Options must win over the fabric's default") + } +} diff --git a/go/internal/fabric/subjects.go b/go/internal/fabric/subjects.go index 07224eab..733cc705 100644 --- a/go/internal/fabric/subjects.go +++ b/go/internal/fabric/subjects.go @@ -15,7 +15,7 @@ const ( // commsStreamSubjects is the single wildcard the COMPASS_COMMS JetStream // stream captures: every tenant's every comms kind. It matches exactly what // CommsSubject builds — four tokens, tenant and kind wildcarded. - commsStreamSubjects = subjectPrefix + ".*.comms.*" + commsStreamSubjects = subjectPrefix + "." + wildcardToken + ".comms." + wildcardToken // RunnerEventsQueue is the queue group every Server's RunnerFabric.Events // subscription joins, so one Runner event is handled by exactly one Server @@ -45,6 +45,28 @@ func CommsSubject(tenant string, kind EventKind) (string, error) { return subjectPrefix + "." + tenant + ".comms." + string(kind), nil } +// CommsWildcardSubject builds the TENANT-WILDCARD comms subject for one event +// kind: compass.*.comms.. It is the delivery plane's cross-tenant +// fan-in subject — the delivery consumer is a per-Server singleton serving +// every tenant, and each message publishes to a concrete +// compass..comms., so catching every tenant needs one consumer +// whose FilterSubject wildcards the tenant token (§T3). +// +// Only the TENANT token is wildcarded. The kind stays a concrete, validated +// token — a wildcard kind would capture all seven comms kinds on one consumer, +// which delivery must not do — so this returns an error if kind is not a valid +// single subject token (see ValidSubjectToken). +// +// Subscribe-side only. Publish derives its subject from the ref via +// CommsSubject, and EventRef.valid rejects a "*" tenant, so no publish can +// ever target this subject. +func CommsWildcardSubject(kind EventKind) (string, error) { + if err := ValidSubjectToken("event kind", string(kind)); err != nil { + return "", err + } + return subjectPrefix + "." + wildcardToken + ".comms." + string(kind), nil +} + // validCommsSubject checks a whole comms subject against the frozen grammar: // exactly compass..comms., with both variable tokens valid. // @@ -77,6 +99,11 @@ func validCommsSubject(subject string) error { const ( commsSubjectTokens = 4 commsToken = "comms" + + // wildcardToken is NATS's single-token wildcard, used by + // CommsWildcardSubject for the tenant token only. Spelled once here so the + // wildcard subject and commsStreamSubjects cannot drift apart. + wildcardToken = "*" ) // RunnerCommandSubject builds a Runner's command subject: diff --git a/go/internal/fabric/subjects_test.go b/go/internal/fabric/subjects_test.go index fb41ac3b..41b1e6a5 100644 --- a/go/internal/fabric/subjects_test.go +++ b/go/internal/fabric/subjects_test.go @@ -133,6 +133,86 @@ func TestCommsSubjectRejectsInvalidKind(t *testing.T) { } } +// TestCommsWildcardSubject defends the delivery plane's cross-tenant fan-in +// subject. The wildcard must sit on the TENANT token and nowhere else: a +// wildcard kind would put all seven comms kinds on one delivery consumer, and +// a subject outside the stream's compass.*.comms.* capture would build a +// consumer that is created successfully and then silently never delivers. +func TestCommsWildcardSubject(t *testing.T) { + t.Parallel() + + t.Run("builds the tenant-wildcard subject", func(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + kind EventKind + want string + }{ + {KindMessagePosted, "compass.*.comms.message_posted"}, + {KindTopicUpserted, "compass.*.comms.topic_upserted"}, + } { + got, err := CommsWildcardSubject(tc.kind) + if err != nil { + t.Fatalf("CommsWildcardSubject(%q): %v", tc.kind, err) + } + if got != tc.want { + t.Fatalf("CommsWildcardSubject(%q) = %q, want %q", tc.kind, got, tc.want) + } + } + }) + + // The kind token is NOT wildcarded, and it is not exempt from the grammar + // either: it is the one caller-supplied token on this path, so an + // unvalidated kind is how a reserved character escapes into the subject. + t.Run("rejects an invalid kind", func(t *testing.T) { + t.Parallel() + for _, kind := range []EventKind{"", "bad.kind", "message>posted", "*", "message posted"} { + if s, err := CommsWildcardSubject(kind); err == nil { + t.Errorf("CommsWildcardSubject(%q) = %q, want an error", kind, s) + } + } + }) + + // Subscribe's strict grammar must stay strict: the wildcard path has its + // own validated builder precisely so validCommsSubject never has to accept + // a "*" tenant, which would also let a concrete-subject caller subscribe + // across tenants by hand. + t.Run("is not reachable through the concrete grammar", func(t *testing.T) { + t.Parallel() + wildcard, err := CommsWildcardSubject(KindMessagePosted) + if err != nil { + t.Fatalf("CommsWildcardSubject: %v", err) + } + if err := validCommsSubject(wildcard); err == nil { + t.Fatalf("validCommsSubject(%q) = nil; Subscribe must stay concrete-only", wildcard) + } + }) + + // The stream captures compass.*.comms.* — if the wildcard subject fell + // outside it the delivery consumer's FilterSubject would match nothing. + t.Run("is captured by the comms stream", func(t *testing.T) { + t.Parallel() + wildcard, err := CommsWildcardSubject(KindMessagePosted) + if err != nil { + t.Fatalf("CommsWildcardSubject: %v", err) + } + streamTokens := strings.Split(commsStreamSubjects, ".") + got := strings.Split(wildcard, ".") + if len(got) != len(streamTokens) { + t.Fatalf("CommsWildcardSubject = %q has %d tokens, want %d to match %q", + wildcard, len(got), len(streamTokens), commsStreamSubjects) + } + for i, want := range streamTokens { + if want == wildcardToken { + continue // the stream wildcards this position; anything matches. + } + if got[i] != want { + t.Fatalf("CommsWildcardSubject = %q: token %d is %q, want %q to be captured by %q", + wildcard, i, got[i], want, commsStreamSubjects) + } + } + }) +} + // TestEventKindsAreValidSubjectTokens defends the closed set of kinds against // the grammar: a kind constant is used verbatim as a subject token, so one // introduced with a "." or an uppercase-with-space spelling would break every