From ef069516580838e3d5fa0078920b5be4883be86c Mon Sep 17 00:00:00 2001 From: mintaka Date: Sat, 5 Sep 2026 14:32:15 -0400 Subject: [PATCH 1/3] feat(fabric): add tenant-wildcard comms subscribe (RIG-3107) The T3 delivery consumer is a per-Server singleton serving every tenant, so it must receive one event kind across all tenants. Each event publishes to a concrete `compass..comms.`, so the singleton needs a tenant-wildcard subscribe. - `CommsWildcardSubject(kind)` builds `compass.*.comms.`: the tenant token is the literal `*`, the kind stays concrete and `ValidSubjectToken`-checked so a wildcard kind can never widen the subject to all kinds. - `(*Fabric).SubscribeKind(ctx, kind, fn)` subscribes on that subject via one durable queue-group consumer, sharing the exact ack/park/teardown machinery of `Subscribe` (both now route through a private `subscribeSubject` helper). `Subscribe` keeps its strict concrete-only grammar; the wildcard is reachable only through `SubscribeKind`'s own validated builder. - Publish is untouched and still cannot target a wildcard: it derives its subject from the ref, and `EventRef.valid` rejects a `*` tenant. - DLQ provenance + per-message logs record `msg.Subject()` (the concrete delivered subject), so a parked message on the wildcard consumer keeps its tenant. - design.md's frozen `EventFabric` interface + SUBJECTS.md document the new read-side seam; the wildcard and concrete consumers are independent durables, so a later migration must retire the concrete subscribes rather than run both. Co-authored-by: Matt Wilkinson --- .../compass-managed-multitenancy/design.md | 10 +- go/internal/fabric/SUBJECTS.md | 39 ++- go/internal/fabric/event_fabric.go | 75 ++++-- go/internal/fabric/event_fabric_test.go | 254 +++++++++++++++++- go/internal/fabric/fabric.go | 5 + go/internal/fabric/fabric_test.go | 3 + go/internal/fabric/subjects.go | 29 +- go/internal/fabric/subjects_test.go | 80 ++++++ 8 files changed, 470 insertions(+), 25 deletions(-) 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..10340208 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,22 @@ 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 } 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 +225,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 +262,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..a569496b 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,125 @@ 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") + } +} diff --git a/go/internal/fabric/fabric.go b/go/internal/fabric/fabric.go index 94e670a0..c8165d01 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 diff --git a/go/internal/fabric/fabric_test.go b/go/internal/fabric/fabric_test.go index 57cf220f..306b63fc 100644 --- a/go/internal/fabric/fabric_test.go +++ b/go/internal/fabric/fabric_test.go @@ -263,6 +263,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) } 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 From 34415b27af759f5bd1f5ff093e5fcfcfff971136 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 6 Sep 2026 19:07:29 -0400 Subject: [PATCH 2/3] fix(fabric): cross-check a delivered ref's tenant against its subject (RIG-3107) The subscribe path decoded an EventRef and handed it to the subscriber without comparing the ref's tenant to the subject the message arrived on, so the tenant-isolation invariant was enforced on the write side only. Publish derives the required subject from the ref and refuses a mismatch, but Publish is not the only writer the stream can have: COMPASS_COMMS is shared and the server carries no per-tenant authorization yet, so any client that reaches the client port can put arbitrary bytes on any comms subject. That mattered because EventRef's contract directs a subscriber to re-read its row under ref.Tenant WITHOUT consulting the subject. A ref naming tenant-a delivered on tenant-b's subject therefore drove a tenant-b-scoped subscriber to read a tenant-a row. SubscribeKind makes it the primary path rather than a latent gap: its consumer spans every tenant, which leaves the payload's tenant field as the only scope a delivery carries. handleEvent now rebuilds the wanted subject from the ref and parks on a mismatch. A mismatch is as unprocessable as an undecodable payload, so it parks rather than retrying, and the DLQ entry keeps the concrete delivered subject so an operator can see which tenant was targeted. Also wires nats.ErrorHandler, without which the Runner plane's documented dropped-and-reported semantic was only dropped: nats.go reports a full channel subscription through the async error callback and the defaults install none, so the fabric's one lossy path dropped events with no log line, no metric and no error. --- go/internal/fabric/event_fabric.go | 23 +++++++ go/internal/fabric/event_fabric_test.go | 84 +++++++++++++++++++++++++ go/internal/fabric/fabric.go | 26 ++++++++ 3 files changed, 133 insertions(+) diff --git a/go/internal/fabric/event_fabric.go b/go/internal/fabric/event_fabric.go index 10340208..38659db0 100644 --- a/go/internal/fabric/event_fabric.go +++ b/go/internal/fabric/event_fabric.go @@ -195,6 +195,29 @@ func (f *Fabric) handleEvent(ctx context.Context, msg jetstream.Msg, fn func(Eve 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, msg, err) return diff --git a/go/internal/fabric/event_fabric_test.go b/go/internal/fabric/event_fabric_test.go index a569496b..86bedd88 100644 --- a/go/internal/fabric/event_fabric_test.go +++ b/go/internal/fabric/event_fabric_test.go @@ -1154,3 +1154,87 @@ func TestSubscribeKindRejectsBadInput(t *testing.T) { 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) + } +} diff --git a/go/internal/fabric/fabric.go b/go/internal/fabric/fabric.go index c8165d01..97a5fec3 100644 --- a/go/internal/fabric/fabric.go +++ b/go/internal/fabric/fabric.go @@ -262,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...) @@ -277,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 From fa9f5f42308d426d166ad464c631fd6ab724fe84 Mon Sep 17 00:00:00 2001 From: mintaka Date: Sun, 6 Sep 2026 20:41:43 -0400 Subject: [PATCH 3/3] test(fabric): pin the subject cross-check symmetry and the slow-consumer report (RIG-3107) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two guarantees from the review loop shipped without tests, and both fail invisibly. The subject cross-check parks a ref whose tenant disagrees with its delivered subject, deciding that by rebuilding the subject with the same CommsSubject call Publish uses. That the two agree was emergent rather than tested: 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. The forged-ref test covers the negative case and the wildcard tests only exercise SubscribeKind, so the highest-traffic path had no defense. TestLegitimateRefsSurviveTheSubjectCrossCheck publishes across the token space ValidSubjectToken admits — dashes, underscores, mixed case, a uuid — asserts each is delivered through a concrete Subscribe, and asserts nothing legitimate reaches the DLQ. The async error handler is the Runner plane's only report of its only lossy path, and nats.go installs none by default while its own fallback writes to raw stderr rather than the fabric's logger. Dropping or shadowing the option would silently return the plane to losing events invisibly with no test failing. TestAsyncErrorHandlerReportsSlowConsumers invokes the installed handler directly — nats.go dispatches it on its own goroutine holding no lock, so calling it is faithful and carries no timing dependence — covering both the nil-subscription branch that the whole drain path takes and the branch that must name the subject. TestCallerErrorHandlerWins pins the Config.Options precedence, which is correct to allow here because the handler is log-only, unlike the ClosedHandler that Close deliberately avoids depending on. --- go/internal/fabric/event_fabric_test.go | 74 ++++++++++++++++++++ go/internal/fabric/fabric_test.go | 91 +++++++++++++++++++++++++ 2 files changed, 165 insertions(+) diff --git a/go/internal/fabric/event_fabric_test.go b/go/internal/fabric/event_fabric_test.go index 86bedd88..467687c8 100644 --- a/go/internal/fabric/event_fabric_test.go +++ b/go/internal/fabric/event_fabric_test.go @@ -1238,3 +1238,77 @@ func TestForgedTenantRefIsParked(t *testing.T) { 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_test.go b/go/internal/fabric/fabric_test.go index 306b63fc..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" @@ -377,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") + } +}