From 66a11267c60428d3e65e0c0ab77d8f59e852d8d0 Mon Sep 17 00:00:00 2001 From: mintaka Date: Fri, 4 Sep 2026 12:44:15 -0400 Subject: [PATCH] feat(fabric): add NATS EventFabric + RunnerFabric seams (RIG-3107) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First code slice of the T3 NATS fabric (frozen in RIG-2861): a new `go/internal/fabric` package providing the two transport seams the comms and runner planes cut over to, over a single NATS connection. Stacked on the go-1.26 floor bump (its own PR) that the nats-server test dependency requires. **EventFabric** — durable comms fan-out on JetStream. `Publish(ctx, subject, EventRef)` / `Subscribe(ctx, subject, func(EventRef))`. Carries a compact `EventRef` (tenant + kind + row id) as an at-least-once reference, never the payload, so JetStream stays a transport rather than a second store; subscribers re-read Postgres by row id. MsgID = sha256 of the length-prefixed ref for dedup within the Duplicates window; app-level DLQ via `Term()` at the delivery budget. **RunnerFabric** — best-effort runner control on core NATS. `SendCommand(ctx, runnerID, *SessionsResponse)` / `Events(ctx) (<-chan RunnerEvent, error)`; offline runners are recovered by the delivery cursor sweep, not by broker durability. One `Fabric` implements both seams over one `nats.Conn`. Lazy stream ensure (the frozen `New` carries no context, so the topology call derives its ctx from the first Publish/Subscribe). Per-subject durable consumer named `comms-`+sha256(subject) — injective, unlike a `.`→`_` substitution which would collapse two subjects onto one shared consumer. Close registers a `StatusChanged(nats.CLOSED)` listener before draining so it observes connection teardown directly rather than through a clobberable ClosedHandler. Full subject grammar + JetStream config in `SUBJECTS.md` beside the package. Also ignores the golangci-lint `-review` cache dir (`/go/.golangci-cache-review/`) alongside the existing `.golangci-cache` entry, so the review lane's per-workspace cache is never committed. Subjects: `compass..comms.`, `compass.runner..cmd`, `compass.runner.events` (queue-grouped), `client.`. Spec-impact: none. Refs RIG-3107 Co-authored-by: Matt Wilkinson --- .gitignore | 6 +- go/internal/fabric/SUBJECTS.md | 132 ++++ go/internal/fabric/doc.go | 44 ++ go/internal/fabric/event_fabric.go | 268 +++++++ go/internal/fabric/event_fabric_test.go | 870 +++++++++++++++++++++++ go/internal/fabric/eventref.go | 111 +++ go/internal/fabric/eventref_test.go | 160 +++++ go/internal/fabric/fabric.go | 372 ++++++++++ go/internal/fabric/fabric_test.go | 376 ++++++++++ go/internal/fabric/runner_fabric.go | 199 ++++++ go/internal/fabric/runner_fabric_test.go | 383 ++++++++++ go/internal/fabric/stream.go | 146 ++++ go/internal/fabric/subjects.go | 136 ++++ go/internal/fabric/subjects_test.go | 219 ++++++ 14 files changed, 3420 insertions(+), 2 deletions(-) create mode 100644 go/internal/fabric/SUBJECTS.md create mode 100644 go/internal/fabric/doc.go create mode 100644 go/internal/fabric/event_fabric.go create mode 100644 go/internal/fabric/event_fabric_test.go create mode 100644 go/internal/fabric/eventref.go create mode 100644 go/internal/fabric/eventref_test.go create mode 100644 go/internal/fabric/fabric.go create mode 100644 go/internal/fabric/fabric_test.go create mode 100644 go/internal/fabric/runner_fabric.go create mode 100644 go/internal/fabric/runner_fabric_test.go create mode 100644 go/internal/fabric/stream.go create mode 100644 go/internal/fabric/subjects.go create mode 100644 go/internal/fabric/subjects_test.go diff --git a/.gitignore b/.gitignore index 34ac2ea7d..963823bb0 100644 --- a/.gitignore +++ b/.gitignore @@ -2,9 +2,11 @@ .moon/cache .moon/docker -# golangci-lint per-workspace cache (compass-go:lint sets GOLANGCI_LINT_CACHE -# to $projectRoot/.golangci-cache to avoid shared-cache contention). +# golangci-lint per-workspace caches (compass-go:lint sets GOLANGCI_LINT_CACHE +# to $projectRoot/.golangci-cache; the review lane uses a -review suffix). Both +# avoid shared-cache contention and must never be committed. /go/.golangci-cache/ +/go/.golangci-cache-review/ # Rust /target diff --git a/go/internal/fabric/SUBJECTS.md b/go/internal/fabric/SUBJECTS.md new file mode 100644 index 000000000..8e0079dd2 --- /dev/null +++ b/go/internal/fabric/SUBJECTS.md @@ -0,0 +1,132 @@ +# Fabric subjects and JetStream configuration + +The written spec for the NATS eventing substrate: the four subject grammars, the +dead-letter subject, and the JetStream stream/consumer configuration. Frozen by +`docs/designs/infra/runtime/compass-managed-multitenancy/design.md` §T3/§Q3; +this file is the operational restatement that later tasks build against, and +`package fabric` is its only implementation. + +## Subject grammar + +| Grammar | Plane | Builder | Direction | +| --- | --- | --- | --- | +| `compass..comms.` | JetStream | `CommsSubject(tenant, kind)` | Server → Servers (comms/delivery fan-out) | +| `compass.runner..cmd` | core NATS | `RunnerCommandSubject(runnerID)` | Server → one Runner (async command push) | +| `compass.runner.events` | core NATS, queue group `compass-runner-events` | `RunnerEventsSubject()` | Runners → exactly one Server (event fan-in) | +| `client.` | core NATS | `ClientSubject(sessionID)` | Server → one live client connection | +| `compass.dlq.comms` | core NATS | `DLQSubject` | fabric → operator (parked events) | + +`client.` sits outside the `compass.` root deliberately — the frozen +grammar names it that way, and it must not be captured by the comms stream's +subject wildcard. + +### Token validation: reject, never sanitize + +NATS reserves `.` (token separator), `*` and `>` (wildcards), and rejects +whitespace in a subject token. Tenant ids, runner ids and session ids are opaque +to the fabric, so `ValidSubjectToken` **refuses** a token carrying any of those +rather than escaping it. Escaping would need an unambiguous inverse the grammar +does not have, and a silently-rewritten token would publish a tenant's events to +a subject nobody is subscribed to — a routing bug masquerading as a quiet +success. An id with a reserved character is a bug where the id is minted. + +### Event kinds + +The `` token is one of the seven comms kinds (`EventKind` in +`eventref.go`), all snake_case so each is a legal single token: +`account_changed`, `channel_group_changed`, `channel_changed`, +`agent_workspace_changed`, `message_posted`, `message_updated`, +`topic_upserted`. + +### Payload: a reference, never a copy + +Every comms subject carries a JSON-encoded `EventRef` — `{tenant, kind, row_id}` +— and never the changed row. Postgres is the sole durability truth; the +subscriber re-reads the row the ref names. That is what makes a replay or a +double delivery idempotent and a drop recoverable from the delivery cursor. + +## JetStream: stream `COMPASS_COMMS` + +Created idempotently with `CreateOrUpdateStream`, so a restart, a second Server +and a config change converge instead of racing. + +| Field | Value | Why | +| --- | --- | --- | +| `Name` | `COMPASS_COMMS` | One stream for every tenant; the consumer's subject filter isolates tenants, so tenant creation stays a Postgres insert rather than a JetStream admin op. | +| `Subjects` | `compass.*.comms.*` | Exactly the four-token comms grammar, tenant and kind wildcarded. | +| `Retention` | limits | A message ages out on `MaxAge` rather than vanishing on ack, so a second consumer group and a bounded replay stay possible. | +| `Storage` | file | Durability across a NATS restart (§Q3). | +| `Replicas` | 1 (3 clustered) | Single-node NATS is R1 by construction; §Q3 specifies R3 when clustered. Postgres is the recovery truth either way. | +| `Discard` | old | At the age/size limit, drop the oldest rather than refusing new publishes — a refused publish would fail a live comms write for the sake of a transport's backlog. | +| `MaxAge` | 24h | Bounds the replay window. A subscriber further behind than this recovers by cursor sweep, not replay. | +| `Duplicates` | 2m | Publish-side dedup window (see MsgID below). | + +### `sync_interval: 100ms` — set on the server, not the stream + +§Q3 specifies `sync_interval: 100ms` for a bounded fsync window (the December +2025 Jepsen analysis documented ~14% acknowledged-write loss under NATS +defaults). **It is not a stream field.** In nats-server it is a file-store +option, and `jetstream.StreamConfig` in nats.go v1.53.1 exposes no equivalent — +so it cannot be set from this package. It is configured on the NATS process: + +- stack/server config: `jetstream { store_dir: "…", sync_interval: "100ms" }` +- Go-embedded or in-process (this package's tests): + `server.Options{SyncInterval: 100 * time.Millisecond}` + +The stack's NATS service config owns the deployment value; `testServer` in +`fabric_test.go` sets the `server.Options` field so the tests run at the +record's value rather than the server default. + +### MsgID / dedup + +`Publish` sets `Nats-Msg-Id` to `sha256(len:tenant | len:kind | len:row_id)` +(`EventRef.msgID`). Deterministic in the ref's three fields, so two Servers +publishing the same logical change — or one retrying a publish whose ack was +lost — collapse to one stored message within the `Duplicates` window. Hashed and +length-prefixed rather than concatenated so no field boundary is ambiguous and +a long row id does not widen the header. + +## JetStream: the per-subject consumer + +One durable pull consumer per subscribed subject, created with +`CreateOrUpdateConsumer`. + +| Field | Value | Why | +| --- | --- | --- | +| `Durable` | `comms-` + sha256(subject) as untruncated hex (e.g. `comms-f48b3059…e25211` for `compass.tenant-a.comms.message_posted`) | Consumer names cannot contain `.`, and subject tokens may legally contain `_` (every snake_case EventKind does), so a `.` → `_` substitution is NOT injective: two distinct subjects would collapse onto one shared durable consumer and the second Subscribe would silently re-point the first's `FilterSubject` (a cross-tenant mis-delivery). Hashing is injective by construction; 70 chars is far inside JetStream's 255-char limit, and truncating would reintroduce the collision surface. Durable and shared, so every Server instance on that subject draws from one consumer: each event is claimed by exactly one instance (§Q3 queue groups), and a restart resumes rather than replaying. The consumer's `Description`/`FilterSubject` still carry the readable subject for operators. | +| `FilterSubject` | the subscribed subject | The tenant/kind isolation the single stream relies on. | +| `AckPolicy` | explicit | §Q3: explicit per-message acks. | +| `AckWait` | 30s | Redelivery backstop for a subscriber that hangs or dies mid-callback; a callback that *fails* is Nak'd for immediate redelivery instead. | +| `MaxDeliver` | 5 | Finite budget of total delivery **attempts**, not retries: `MaxDeliver=1` parks on the first failure with no retry at all. Enforced twice by design — the app-level check parks at the budget, and the consumer's server-side `MaxDeliver` is the backstop for a delivery whose metadata is unreadable. Both derive from the SAME fabric `Config`, and the consumer is shared, so every Server instance on a subject must run one config (RIG-2861: one stack config) or the shared consumer's server-side budget flip-flops with whichever instance last ran `CreateOrUpdateConsumer`. | +| `Replicas` | matches the stream | — | + +Delivery semantics per message: + +1. Decode the `EventRef`. Undecodable → **park immediately** (no number of + redeliveries changes the bytes). +2. Run the subscriber callback under a panic guard. A panic becomes a failure — + it neither takes the process down nor acks an event nobody handled. +3. Success → `Ack()`. A *failed ack* after successful handling is logged, never + parked: it costs one redelivery, which the subscriber's Postgres re-read makes + idempotent. +4. Failure → read `Metadata().NumDelivered`, which counts **attempts**. Below + `MaxDeliver` → `Nak()` for immediate redelivery. At `MaxDeliver` → park. + +## Dead-letter: `compass.dlq.comms` + +JetStream has no native DLQ, so the fabric implements the app-level pattern: on +park, republish the raw payload to `compass.dlq.comms` and then +`TermWithReason` the message so the server stops redelivering it. + +- The republish goes over **core NATS**, not JetStream. The DLQ is a diagnostic + tap, not a recovery path — recovery always terminates in the Postgres row — and + 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. + +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 +restarts. diff --git a/go/internal/fabric/doc.go b/go/internal/fabric/doc.go new file mode 100644 index 000000000..0cf447ca4 --- /dev/null +++ b/go/internal/fabric/doc.go @@ -0,0 +1,44 @@ +// Package fabric is the NATS eventing substrate under Compass's async layer: +// one client, one connection per party, two planes. +// +// The design record is +// docs/designs/infra/runtime/compass-managed-multitenancy/design.md (§T3, §Q3). +// The two seams it freezes are [EventFabric] (comms/delivery event fan-out) and +// [RunnerFabric] (Server→Runner command push and Runner→Server event fan-in); +// [Fabric] implements both over a single [nats.Conn], so each Runner and each +// Server holds exactly one fabric connection. +// +// # The plane split +// +// The two seams ride deliberately different NATS semantics: +// +// - [EventFabric] rides JetStream — durable at-least-once fan-out with +// explicit per-message acks, publish-side dedup, a bounded delivery-attempt +// count and a dead-letter subject. Comms events must survive a subscriber +// restart, so they need a stream. +// - [RunnerFabric] rides core NATS — best-effort at-most-once. A command to an +// offline Runner is not a lost write: the delivery-cursor sweep in Postgres +// recovers it ("a fabric outage degrades to sweep-recovered delivery"), so +// paying for a stream here would buy nothing and add a second store. +// +// # Postgres is the only truth +// +// JetStream is a transport, never a second store. That is why an [EventRef] is a +// compact reference — tenant, kind, row id — and never a payload copy: a +// subscriber re-reads the row from Postgres, so a dropped, replayed or +// double-delivered event reconciles against the message row and the +// per-(agent, channel) delivery cursor. Consumer state here is disposable by +// construction. +// +// # Fail-closed +// +// Every error surfaces wrapped; nothing is swallowed and nothing panics. A +// subject built from an invalid token is refused rather than silently corrupted +// (see [ValidSubjectToken]), an undecodable [EventRef] is parked on the DLQ +// rather than dropped, and a subscriber callback that panics is caught, retried +// up to Config.MaxDeliver times, then parked. +// +// The subject grammar and the full JetStream stream/consumer/DLQ configuration +// are specified in SUBJECTS.md beside this file — that document, not this +// package's defaults, is what later tasks build against. +package fabric diff --git a/go/internal/fabric/event_fabric.go b/go/internal/fabric/event_fabric.go new file mode 100644 index 000000000..9caf00d72 --- /dev/null +++ b/go/internal/fabric/event_fabric.go @@ -0,0 +1,268 @@ +package fabric + +import ( + "context" + "fmt" + "strings" + "sync" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// Publish sends ref to subject on JetStream, returning only once the server has +// acked it into the stream — so a Publish that returns nil means the event is +// stored, and one that returns an error is genuinely unpublished and the caller +// can leave its cursor unadvanced. +// +// WithMsgID(ref.msgID()) gives publish-side dedup: two Servers publishing the +// same logical change, or a retry of a publish whose ack was lost, collapse to +// one stored message inside the stream's duplicate window. +func (f *Fabric) Publish(ctx context.Context, subject string, ref EventRef) error { + if err := f.checkOpen(); err != nil { + return err + } + if err := ref.valid(); err != nil { + return err + } + // The subject must be exactly the one the ref itself names. A head-only + // check would let a caller publish tenant-a's ref on tenant-b's subject: + // the subscriber that claims it is scoped to tenant-b but the ref tells it + // to re-read a tenant-a row, which is the cross-tenant read the EventRef + // invariant (eventref.go) exists to prevent. Deriving the wanted subject + // from the ref also subsumes the whole-subject grammar check. + want, err := CommsSubject(ref.Tenant, ref.Kind) + if err != nil { + return err + } + if subject != want { + return fmt.Errorf("fabric: event ref %s/%s is for subject %q but was published on %q", ref.Tenant, ref.Kind, want, subject) + } + if _, err := f.ensureStream(ctx); err != nil { + return err + } + data, err := ref.encode() + if err != nil { + return err + } + if _, err := f.js.Publish(ctx, subject, data, jetstream.WithMsgID(ref.msgID())); err != nil { + return fmt.Errorf("fabric: publishing %s/%s to %q: %w", ref.Kind, ref.RowID, subject, err) + } + return nil +} + +// Subscribe drives fn for every event on subject until the returned Unsubscribe +// is called, ctx is done, or the Fabric is closed — whichever comes first. Every +// one of those three paths DRAINS the consumer, so an event this process had +// already claimed is processed and acked rather than discarded. +// +// The consumer is a DURABLE pull consumer named from the subject, so every +// Server instance on that subject shares one consumer: each event is claimed by +// exactly one instance (§Q3's queue-group semantics), and a restart resumes from +// the consumer's position instead of replaying or skipping. Because that +// consumer is shared, every instance subscribing to a subject must run the same +// fabric Config — see Config.MaxDeliver. +// +// Acking is explicit and follows fn: fn returning normally acks, and fn +// panicking is recovered and treated as a failure (a panic in one subscriber +// must not take down the process — and must not silently ack an unprocessed +// event either). A failure Naks for immediate redelivery until NumDelivered +// reaches MaxDeliver — total ATTEMPTS, not retries — at which point the message +// is parked on DLQSubject and Term'd. An undecodable payload is parked +// immediately: redelivering it can never succeed. +func (f *Fabric) Subscribe(ctx context.Context, subject string, fn func(EventRef)) (Unsubscribe, error) { + if err := f.checkOpen(); err != nil { + return nil, err + } + if fn == nil { + return nil, fmt.Errorf("fabric: Subscribe(%q) requires a callback", subject) + } + if err := validCommsSubject(subject); err != nil { + return nil, err + } + stream, err := f.ensureStream(ctx) + if err != nil { + return nil, err + } + // The consumer is shared and durable, so consumerConfig's values come from + // a Config every instance on this subject must agree on (see + // Config.MaxDeliver). + cons, err := stream.CreateOrUpdateConsumer(ctx, f.cfg.consumerConfig(subject)) + if err != nil { + return nil, fmt.Errorf("fabric: creating consumer for %q on %s: %w", subject, f.cfg.streamName(), err) + } + + cc, err := cons.Consume(func(msg jetstream.Msg) { + f.handleEvent(ctx, subject, 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 + // consumer wedged for good. + f.log.WarnContext(ctx, "fabric: consume error", "subject", subject, "error", err) + })) + if err != nil { + return nil, fmt.Errorf("fabric: consuming %q: %w", subject, err) + } + + // One teardown path, reached from the caller's Unsubscribe, from ctx being + // done, or from the fabric closing, and run at most once — so the watchdog + // goroutine always exits and the consumer is never torn down twice. + // + // Drain, not Stop: Stop DISCARDS whatever the pull consumer has already + // buffered (up to its prefetch), and on a durable shared consumer those + // messages are claimed-not-acked, so they only come back to anyone after + // AckWait — a silent multi-second stall for events this process had already + // accepted. Drain runs them through fn and acks them first. That is the + // right choice for all three teardown paths: the durability contract says a + // claimed event is not silently dropped, and a cancelled context on + // shutdown does not change that. + var once sync.Once + done := make(chan struct{}) + stop := func() { + once.Do(func() { + cc.Drain() + close(done) + }) + } + go func() { + select { + case <-ctx.Done(): + stop() + case <-f.teardown: + // Close alone must tear this down: nats.go does not close a + // ConsumeContext's buffer when the connection closes, so without + // this case a Close with an uncancelled ctx leaks this goroutine + // and the consumer with it. + stop() + case <-done: + } + }() + return stop, nil +} + +// 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)) { + ref, decodeErr := decodeEventRef(msg.Data()) + if decodeErr != nil { + // Unparseable: no number of redeliveries changes the bytes. + f.park(ctx, subject, msg, decodeErr) + return + } + if err := invoke(fn, ref); err != nil { + f.retryOrPark(ctx, subject, 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) + } +} + +// invoke calls fn, converting a panic into an error. A subscriber callback is +// consumer code running on the fabric's goroutine: letting it panic would take +// the process down, and recovering without failing the message would ack an +// event nobody processed. +func invoke(fn func(EventRef), ref EventRef) (err error) { + defer func() { + if r := recover(); r != nil { + err = fmt.Errorf("fabric: subscriber panicked handling %s/%s: %v", ref.Kind, ref.RowID, r) + } + }() + fn(ref) + return nil +} + +// retryOrPark Naks a failed delivery for another attempt, or parks it once the +// 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) { + 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)) + return + } + if md.NumDelivered >= f.cfg.deliveryBudget() { + f.park(ctx, subject, 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) + 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) + } +} + +// park implements the dead-letter pattern JetStream has no native support for: +// republish the raw payload to DLQSubject, then Term the message so the server +// stops redelivering it. +// +// The payload goes out verbatim over CORE NATS, with the original subject and +// the reason in headers. Core rather than JetStream because the DLQ is a +// diagnostic tap, not a recovery path — recovery is always the Postgres row — +// and because a DLQ publish that itself needed a stream would need its own DLQ. +// +// Term is issued even if the DLQ publish fails: leaving a poison message +// redelivering forever is the worse failure, and the reason is logged either +// way. +// +// 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) { + f.log.ErrorContext(ctx, "fabric: parking event on the dlq", + "subject", subject, "dlq_subject", DLQSubject, "error", cause) + + dlq := nats.NewMsg(DLQSubject) + dlq.Data = msg.Data() + dlq.Header.Set(dlqHeaderSubject, 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) + } + 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) + } +} + +// DLQ message headers, so a consumer of DLQSubject knows what the payload was +// and why it parked without parsing a log line. +const ( + dlqHeaderSubject = "Compass-Original-Subject" + dlqHeaderReason = "Compass-Park-Reason" +) + +// maxParkReason bounds the reason written to the DLQ header and the +TERM ack +// body. Both are wire-protocol fields under the server's max-payload ceiling, +// and jetstream's TermWithReason applies no sanitization of its own, so an +// unbounded or newline-bearing reason could make the park itself fail — the +// worst possible place to fail, since the alternative is a poison message +// redelivering forever. +const maxParkReason = 256 + +// sanitizeReason strips CR/LF, which would corrupt the +TERM ack line and the +// header, and bounds the length. A subscriber panic embeds an arbitrary consumer +// value in the cause, so neither property can be assumed. The full, untruncated +// cause still reaches the ErrorContext log. +// +// Truncation is on a byte boundary and can split a trailing rune; this is a +// diagnostic string read by an operator, not something that is parsed, so a +// mangled last character is an acceptable price for a hard byte bound. +func sanitizeReason(s string) string { + s = strings.NewReplacer("\r", " ", "\n", " ").Replace(s) + if len(s) > maxParkReason { + s = s[:maxParkReason] + } + return s +} diff --git a/go/internal/fabric/event_fabric_test.go b/go/internal/fabric/event_fabric_test.go new file mode 100644 index 000000000..1499b3b2d --- /dev/null +++ b/go/internal/fabric/event_fabric_test.go @@ -0,0 +1,870 @@ +package fabric + +import ( + "context" + "errors" + "fmt" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/nats-io/nats.go" +) + +// TestEventFabricRoundTrip defends the core promise of the event plane: an +// EventRef published on a comms subject reaches a subscriber on that subject +// with tenant, kind and row id intact through encode → JetStream → decode. If +// any field were lost the subscriber would re-read the wrong row, or no row. +func TestEventFabricRoundTrip(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + subject, err := CommsSubject("t1", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + got := make(chan EventRef, 1) + unsub, err := f.Subscribe(ctx, subject, func(r EventRef) { got <- r }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsub() + + want := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-1"} + if err := f.Publish(ctx, subject, want); err != nil { + t.Fatalf("Publish: %v", err) + } + if delivered := recvRef(t, got); delivered != want { + t.Fatalf("delivered %+v, want %+v", delivered, want) + } +} + +// TestEventFabricDedupsIdenticalPublishes defends the WithMsgID dedup the +// record calls for: two Servers publishing the same logical change — or one +// retrying a publish whose ack was lost — must deliver ONCE. Without it every +// retry would drive the subscriber's handler a second time. +// +// The second delivery is proven absent by a positive gate, not a sleep: a third +// publish with a DIFFERENT ref must arrive, and the duplicate must not have +// arrived before it. Ordering within one stream subject is the stream's, so if +// the duplicate were going to be delivered it would be delivered first. +func TestEventFabricDedupsIdenticalPublishes(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + subject, err := CommsSubject("t1", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + got := make(chan EventRef, 4) + unsub, err := f.Subscribe(ctx, subject, func(r EventRef) { got <- r }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsub() + + dup := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-dup"} + sentinel := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-sentinel"} + for _, ref := range []EventRef{dup, dup, sentinel} { + if err := f.Publish(ctx, subject, ref); err != nil { + t.Fatalf("Publish %s: %v", ref.RowID, err) + } + } + + if first := recvRef(t, got); first != dup { + t.Fatalf("first delivery = %+v, want %+v", first, dup) + } + // The sentinel arriving second is what proves the duplicate was deduped: + // the stream preserves per-subject order, so a stored duplicate would be + // delivered ahead of it. + if second := recvRef(t, got); second != sentinel { + t.Fatalf("second delivery = %+v, want the sentinel %+v (the duplicate was not deduped)", second, sentinel) + } +} + +// TestEventFabricFiltersBySubject defends tenant isolation on a single shared +// stream. The COMPASS_COMMS stream captures every tenant's every kind, so the +// consumer's FilterSubject is the ONLY thing keeping tenant t2's events out of +// tenant t1's subscriber — a cross-tenant leak if it drifted. +func TestEventFabricFiltersBySubject(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + mine, err := CommsSubject("t1", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + theirs, err := CommsSubject("t2", KindAccountChanged) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + + got := make(chan EventRef, 4) + unsub, err := f.Subscribe(ctx, mine, func(r EventRef) { got <- r }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsub() + + // The other tenant's event goes first. If the filter leaked it would be + // delivered before the sentinel, since both are already stored. + if err := f.Publish(ctx, theirs, EventRef{Tenant: "t2", Kind: KindAccountChanged, RowID: "acct-1"}); err != nil { + t.Fatalf("Publish to the other tenant: %v", err) + } + want := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-1"} + if err := f.Publish(ctx, mine, want); err != nil { + t.Fatalf("Publish: %v", err) + } + if delivered := recvRef(t, got); delivered != want { + t.Fatalf("delivered %+v, want %+v — tenant t2's event leaked through the subject filter", delivered, want) + } +} + +// 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 +// durable consumer means silently dropped for every other instance too. +// +// Absence of delivery is proven positively: after unsubscribing, a fresh +// subscriber on the same subject receives the event the stale callback must not +// have seen. +func TestUnsubscribeStopsDelivery(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + subject, err := CommsSubject("t1", KindChannelChanged) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + + var stale atomic.Int64 + first := make(chan EventRef, 1) + unsub, err := f.Subscribe(ctx, subject, func(r EventRef) { + stale.Add(1) + select { + case first <- r: + default: + } + }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + + // Prove the subscription is live before tearing it down, so the assertion + // below distinguishes "stopped" from "never started". + warmup := EventRef{Tenant: "t1", Kind: KindChannelChanged, RowID: "ch-warmup"} + if err := f.Publish(ctx, subject, warmup); err != nil { + t.Fatalf("Publish warmup: %v", err) + } + if got := recvRef(t, first); got != warmup { + t.Fatalf("warmup delivered %+v, want %+v", got, warmup) + } + before := stale.Load() + + unsub() + // Idempotent by contract: a second call must not double-Stop. + unsub() + + // A second subscriber on the same subject picks up where the consumer left + // off; its delivery is the gate. + second := make(chan EventRef, 1) + unsub2, err := f.Subscribe(ctx, subject, func(r EventRef) { second <- r }) + if err != nil { + t.Fatalf("second Subscribe: %v", err) + } + defer unsub2() + + want := EventRef{Tenant: "t1", Kind: KindChannelChanged, RowID: "ch-after"} + if err := f.Publish(ctx, subject, want); err != nil { + t.Fatalf("Publish after unsubscribe: %v", err) + } + if got := recvRef(t, second); got != want { + t.Fatalf("second subscriber delivered %+v, want %+v", got, want) + } + if after := stale.Load(); after != before { + t.Fatalf("the unsubscribed callback ran %d more time(s) after Unsubscribe", after-before) + } +} + +// TestSubscribeStopsWhenContextIsDone defends the other teardown path: a +// Subscribe whose ctx is cancelled must stop consuming without the caller +// calling Unsubscribe. Otherwise a server shutdown that cancels its root +// context would leave every subscription's goroutine running. +func TestSubscribeStopsWhenContextIsDone(t *testing.T) { + t.Parallel() + f := newFabric(t, Config{}) + + subject, err := CommsSubject("t1", KindTopicUpserted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + + // Rooted at context.Background() because this is a test root. + subCtx, cancel := context.WithCancel(context.Background()) + live := make(chan EventRef, 1) + unsub, err := f.Subscribe(subCtx, subject, func(r EventRef) { live <- r }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsub() + + pubCtx := testCtx(t) + warmup := EventRef{Tenant: "t1", Kind: KindTopicUpserted, RowID: "topic-warmup"} + if err := f.Publish(pubCtx, subject, warmup); err != nil { + t.Fatalf("Publish warmup: %v", err) + } + if got := recvRef(t, live); got != warmup { + t.Fatalf("warmup delivered %+v, want %+v", got, warmup) + } + + cancel() + + // Gate on the replacement subscription receiving, exactly as the + // Unsubscribe test does. + after := make(chan EventRef, 1) + unsub2, err := f.Subscribe(pubCtx, subject, func(r EventRef) { after <- r }) + if err != nil { + t.Fatalf("second Subscribe: %v", err) + } + defer unsub2() + + want := EventRef{Tenant: "t1", Kind: KindTopicUpserted, RowID: "topic-after"} + if err := f.Publish(pubCtx, subject, want); err != nil { + t.Fatalf("Publish after cancel: %v", err) + } + if got := recvRef(t, after); got != want { + t.Fatalf("delivered %+v, want %+v", got, want) + } + select { + case leaked := <-live: + t.Fatalf("the cancelled subscription delivered %+v; its consume context was not stopped", leaked) + default: + } +} + +// TestPublishRejectsInvalidInput defends the fail-closed publish path. Each of +// these would otherwise become a stored, undecodable, or mis-subjected message +// that only surfaces later on the DLQ — far from the caller that caused it. +func TestPublishRejectsInvalidInput(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + subject, err := CommsSubject("t1", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + for _, tc := range []struct { + name string + subject string + ref EventRef + }{ + {"no tenant", subject, EventRef{Kind: KindMessagePosted, RowID: "m1"}}, + {"no kind", subject, EventRef{Tenant: "t1", RowID: "m1"}}, + {"no row id", subject, EventRef{Tenant: "t1", Kind: KindMessagePosted}}, + {"empty subject", "", EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "m1"}}, + {"wildcard subject root", "*.t1.comms.message_posted", EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "m1"}}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if err := f.Publish(ctx, tc.subject, tc.ref); err == nil { + t.Fatalf("Publish(%q, %+v): want an error", tc.subject, tc.ref) + } + }) + } +} + +// TestSubscribeRequiresCallback defends against the nil-callback footgun: a +// Subscribe with no handler would create a durable consumer that acks every +// event and hands it to nobody — a silent, permanent event sink on a shared +// consumer. +func TestSubscribeRequiresCallback(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + subject, err := CommsSubject("t1", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + if _, err := f.Subscribe(ctx, subject, nil); err == nil { + t.Fatal("Subscribe with a nil callback: want an error") + } +} + +// TestPoisonMessageParksOnDLQ defends the record's "max_deliver + dead-letter +// subject so a poison message parks instead of redelivering forever". A +// callback that always fails must be retried up to MaxDeliver and then parked — +// not retried in perpetuity, which would wedge the shared durable consumer's +// ack window and stall every subsequent event on that subject. +// +// The DLQ arrival is the gate; a raw core-NATS subscription reads it, and the +// park headers must identify the original subject so an operator needs no log +// correlation. +func TestPoisonMessageParksOnDLQ(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + // MaxDeliver 2 with an immediate Nak keeps the test fast without weakening + // the invariant: the budget must be finite and the park must happen at it. + f := newFabric(t, Config{URL: url, MaxDeliver: 2, Log: quietLogger(t)}) + + // A raw connection reads the DLQ, proving the park is observable to a plain + // core-NATS consumer (the DLQ is a diagnostic tap, not a stream). + 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) + } + + subject, err := CommsSubject("t1", KindMessageUpdated) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + + var attempts atomic.Int64 + // The callback panics: the fabric must treat a subscriber panic as a + // failure (neither crashing the process nor acking an unhandled event), so + // this exercises the panic guard and the retry budget together. + unsub, err := f.Subscribe(ctx, subject, func(EventRef) { + attempts.Add(1) + panic("subscriber is broken") + }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsub() + + poison := EventRef{Tenant: "t1", Kind: KindMessageUpdated, RowID: "msg-poison"} + if err := f.Publish(ctx, subject, 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) + } + if got := msg.Header.Get(dlqHeaderSubject); got != subject { + t.Errorf("park header %s = %q, want %q", dlqHeaderSubject, got, subject) + } + if msg.Header.Get(dlqHeaderReason) == "" { + t.Errorf("park header %s is empty; an operator reading the dlq has no reason", dlqHeaderReason) + } + if got := attempts.Load(); got != 2 { + t.Errorf("callback ran %d time(s), want exactly MaxDeliver=2 attempts before parking", got) + } +} + +// 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 — +// which is only true if the failing message is Term'd rather than left pending. +func TestSubscriberPanicDoesNotBlockOtherEvents(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{MaxDeliver: 1, Log: quietLogger(t)}) + + subject, err := CommsSubject("t1", KindAgentWorkspaceChanged) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + + good := make(chan EventRef, 1) + unsub, err := f.Subscribe(ctx, subject, func(r EventRef) { + if r.RowID == "poison" { + panic("subscriber is broken") + } + good <- r + }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsub() + + if err := f.Publish(ctx, subject, EventRef{Tenant: "t1", Kind: KindAgentWorkspaceChanged, RowID: "poison"}); err != nil { + t.Fatalf("Publish poison: %v", err) + } + want := EventRef{Tenant: "t1", Kind: KindAgentWorkspaceChanged, RowID: "ws-ok"} + if err := f.Publish(ctx, subject, want); err != nil { + t.Fatalf("Publish good: %v", err) + } + if got := recvRef(t, good); got != want { + t.Fatalf("delivered %+v, want %+v", got, want) + } +} + +// TestUndecodablePayloadParksImmediately defends the decode-failure path's +// distinct policy: bytes that cannot parse will never parse, so redelivering +// them burns the budget for nothing. It must park on the FIRST delivery. +// +// The malformed payload is published raw through JetStream on a comms subject, +// which is what a rolling deploy of an incompatible publisher would look like. +func TestUndecodablePayloadParksImmediately(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url, MaxDeliver: 5, 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) + } + + subject, err := CommsSubject("t1", KindChannelGroupChanged) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + + var calls atomic.Int64 + unsub, err := f.Subscribe(ctx, subject, func(EventRef) { calls.Add(1) }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsub() + + // Publishing through the fabric's own JetStream context, but with bytes the + // codec cannot read. + if _, err := f.js.Publish(ctx, subject, []byte("{not an event ref")); err != nil { + t.Fatalf("publishing a malformed payload: %v", err) + } + + msg, err := dlq.NextMsgWithContext(ctx) + if err != nil { + t.Fatalf("waiting for the parked message on %q: %v", DLQSubject, err) + } + if got, want := string(msg.Data), "{not an event ref"; got != want { + t.Errorf("parked payload = %q, want the original bytes %q", got, want) + } + if got := calls.Load(); got != 0 { + t.Errorf("callback ran %d time(s) for an undecodable payload, want 0", got) + } +} + +// TestSubscribeIsIdempotentAcrossInstances defends the durable-consumer design: +// two Fabrics on the same subject share ONE durable consumer, so each event is +// claimed by exactly one of them (§Q3's queue-group semantics). Two independent +// consumers would double-handle every comms event across a two-Server +// deployment. +func TestSubscribeIsIdempotentAcrossInstances(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + a := newFabric(t, Config{URL: url}) + b := newFabric(t, Config{URL: url}) + + subject, err := CommsSubject("t1", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + + var ( + mu sync.Mutex + seen []EventRef + total = make(chan struct{}, 8) + ) + record := func(r EventRef) { + mu.Lock() + seen = append(seen, r) + mu.Unlock() + total <- struct{}{} + } + unsubA, err := a.Subscribe(ctx, subject, record) + if err != nil { + t.Fatalf("Subscribe on a: %v", err) + } + defer unsubA() + unsubB, err := b.Subscribe(ctx, subject, record) + if err != nil { + t.Fatalf("Subscribe on b: %v", err) + } + defer unsubB() + + want := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-shared"} + if err := a.Publish(ctx, subject, want); err != nil { + t.Fatalf("Publish: %v", err) + } + + // One claim is the invariant. Gate on the first, then assert no second + // arrives before a sentinel published afterwards is claimed — the same + // positive-gate pattern as the dedup test. + select { + case <-total: + case <-time.After(gate): + t.Fatalf("no instance claimed the event within %s", gate) + } + sentinel := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-sentinel"} + if err := a.Publish(ctx, subject, sentinel); err != nil { + t.Fatalf("Publish sentinel: %v", err) + } + select { + case <-total: + case <-time.After(gate): + t.Fatalf("no instance claimed the sentinel within %s", gate) + } + + mu.Lock() + defer mu.Unlock() + if len(seen) != 2 { + t.Fatalf("claims = %+v (%d), want exactly 2 — a shared durable consumer must not double-deliver", seen, len(seen)) + } + if seen[0] != want || seen[1] != sentinel { + t.Fatalf("claims = %+v, want [%+v %+v]", seen, want, sentinel) + } +} + +// TestEnsureStreamIsIdempotent defends CreateOrUpdateStream over CreateStream: +// a container restart, a second Server, and a config change must all converge +// on the same stream rather than one of them failing with "stream already +// exists". +func TestEnsureStreamIsIdempotent(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + a := newFabric(t, Config{URL: url}) + b := newFabric(t, Config{URL: url}) + + for i, f := range []*Fabric{a, b, a, b} { + s, err := f.ensureStream(ctx) + if err != nil { + t.Fatalf("ensureStream #%d: %v", i, err) + } + if got := s.CachedInfo().Config.Name; got != DefaultStreamName { + t.Fatalf("ensureStream #%d returned stream %q, want %q", i, got, DefaultStreamName) + } + } +} + +// TestEnsureStreamErrorIsNotCached defends the retryability of a failed +// topology call: caching a failure would poison the Fabric for its whole +// lifetime, so a NATS blip during startup would permanently disable publishing +// on a process that is otherwise healthy and reconnected. +func TestEnsureStreamErrorIsNotCached(t *testing.T) { + t.Parallel() + f := newFabric(t, Config{}) + + // An already-cancelled context fails the topology call without touching the + // server. Rooted at Background because this is a test root. + dead, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := f.ensureStream(dead); err == nil { + t.Fatal("ensureStream with a cancelled context: want an error") + } + if f.stream != nil { + t.Fatal("a failed ensureStream must not cache a stream handle") + } + if _, err := f.ensureStream(testCtx(t)); err != nil { + t.Fatalf("ensureStream must be retryable after a failure: %v", err) + } +} + +// TestInvokeConvertsPanicToError defends the guard in isolation: a subscriber +// callback runs on the fabric's goroutine, so an unrecovered panic there would +// take the whole server down. It must become an error the delivery path can act +// on. +func TestInvokeConvertsPanicToError(t *testing.T) { + t.Parallel() + ref := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "m1"} + + if err := invoke(func(EventRef) {}, ref); err != nil { + t.Fatalf("a callback that returns normally must not error: %v", err) + } + + err := invoke(func(EventRef) { panic(errors.New("boom")) }, ref) + if err == nil { + t.Fatal("a panicking callback must yield an error, not a nil (which would ack an unhandled event)") + } + if got := err.Error(); !strings.Contains(got, "m1") || !strings.Contains(got, "boom") { + t.Errorf("error %q should name the event and the cause", got) + } +} + +// TestPublishRejectsCrossTenantRef defends the EventRef↔subject binding, and it +// is a cross-tenant leak that it closes: a head-only subject guard accepted +// tenant-a's ref published on tenant-b's subject, so the instance that claimed +// it — scoped to tenant-b — would be told to re-read a tenant-a row. +// +// Absence of the bad event is proven positively: the same subscriber receives a +// later, well-formed publish on that subject, and the rejected ref must not +// have arrived ahead of it (per-subject stream order means a stored one would). +func TestPublishRejectsCrossTenantRef(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + theirs, err := CommsSubject("tenant-b", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + + got := make(chan EventRef, 2) + unsub, err := f.Subscribe(ctx, theirs, func(r EventRef) { got <- r }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsub() + + crossTenant := EventRef{Tenant: "tenant-a", Kind: KindMessagePosted, RowID: "r1"} + err = f.Publish(ctx, theirs, crossTenant) + if err == nil { + t.Fatal("Publish of a tenant-a ref on tenant-b's subject: want an error, got nil") + } + // The message must name both sides, or an operator cannot tell which half + // of the mismatch is the bug. + if msg := err.Error(); !strings.Contains(msg, "tenant-a") || !strings.Contains(msg, "tenant-b") { + t.Errorf("error %q must name both the ref's tenant and the subject's", msg) + } + + want := EventRef{Tenant: "tenant-b", Kind: KindMessagePosted, RowID: "r2"} + if err := f.Publish(ctx, theirs, want); err != nil { + t.Fatalf("Publish of a matching ref: %v", err) + } + if delivered := recvRef(t, got); delivered != want { + t.Fatalf("delivered %+v, want %+v — the cross-tenant ref was stored and delivered", delivered, want) + } +} + +// TestParkReasonIsSanitizedAndBounded defends the park path against the reason +// string it does not control. A subscriber panic embeds an arbitrary consumer +// value in the cause, and that cause goes onto the wire twice — a NATS header +// and the +TERM ack body — neither of which TermWithReason sanitizes or bounds. +// An unbounded or CR/LF-bearing reason could make the park itself fail, which is +// the worst place to fail: the poison message would keep redelivering. +func TestParkReasonIsSanitizedAndBounded(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) + } + + subject, err := CommsSubject("t1", KindMessageUpdated) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + + // A hostile panic value: CR/LF to corrupt the ack line and the header, and + // multiple KB to blow any size bound. This is test code deliberately + // driving the package's documented panic guard, as the DLQ tests above do. + hostile := "line-one\r\nline-two " + strings.Repeat("x", 4096) + unsub, err := f.Subscribe(ctx, subject, func(EventRef) { panic(hostile) }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsub() + + poison := EventRef{Tenant: "t1", Kind: KindMessageUpdated, RowID: "msg-hostile"} + if err := f.Publish(ctx, subject, poison); err != nil { + t.Fatalf("Publish: %v", err) + } + + // The park still happening is half the invariant: a reason the wire rejects + // would take the DLQ publish down with it. + msg, err := dlq.NextMsgWithContext(ctx) + if err != nil { + t.Fatalf("waiting for the parked message on %q: %v", DLQSubject, err) + } + reason := msg.Header.Get(dlqHeaderReason) + if reason == "" { + t.Fatalf("park header %s is empty; an operator reading the dlq has no reason", dlqHeaderReason) + } + if strings.ContainsAny(reason, "\r\n") { + t.Errorf("park reason %q contains CR/LF; it must be stripped before the header and the +TERM body", reason) + } + if len(reason) > maxParkReason { + t.Errorf("park reason is %d bytes, want at most maxParkReason=%d", len(reason), maxParkReason) + } + // Bounding must not empty it out: the surviving prefix is what an operator + // reads. + if !strings.Contains(reason, "line-one") { + t.Errorf("park reason %q lost the head of the cause; truncation must keep the prefix", reason) + } +} + +// TestUnsubscribeDrainsBufferedEvents defends the durability contract across +// teardown: the pull consumer prefetches, so at Unsubscribe there are events +// this instance has already CLAIMED but not yet run. Stopping discards them, and +// on a shared durable consumer a claimed-not-acked event only returns to anyone +// after AckWait (~30s) — a silent stall for events the fabric had accepted. +// Draining runs them through the callback and acks them instead. +// +// The buffer is established positively rather than by sleeping: the callback +// blocks on the first event, and the test waits for the server to report all n +// as delivered-unacked before tearing down, so all n are provably in this +// client's hands. +func TestUnsubscribeDrainsBufferedEvents(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + subject, err := CommsSubject("t1", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + + const n = 5 + published := make([]EventRef, 0, n) + for i := range n { + ref := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: fmt.Sprintf("msg-%d", i)} + if err := f.Publish(ctx, subject, ref); err != nil { + t.Fatalf("Publish %s: %v", ref.RowID, err) + } + published = append(published, ref) + } + + var ( + got = make(chan EventRef, n) + release = make(chan struct{}) + gateOne sync.Once + ) + unsub, err := f.Subscribe(ctx, subject, func(r EventRef) { + got <- r + // Only the first delivery blocks; that is enough to let the rest pile + // up in the consumer's buffer, which is what teardown must not discard. + gateOne.Do(func() { + select { + case <-release: + case <-time.After(gate): + } + }) + }) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsub() + + // Wait for the server to have handed every event to this client. Polling + // the consumer, not sleeping: the assertion below is only meaningful once + // the buffer it defends actually exists. + stream, err := f.ensureStream(ctx) + if err != nil { + t.Fatalf("ensureStream: %v", err) + } + cons, err := stream.Consumer(ctx, durableName(subject)) + if err != nil { + t.Fatalf("Consumer(%q): %v", durableName(subject), err) + } + pollUntil(t, "all events claimed by the subscriber", func() bool { + info, err := cons.Info(ctx) + if err != nil { + t.Fatalf("consumer Info: %v", err) + } + return info.NumAckPending == n + }) + + // Tear down with the buffer full, THEN let the blocked callback go: a + // discarding teardown loses every buffered event, a draining one runs them. + unsub() + close(release) + + seen := make(map[string]int, n) + for range n { + seen[recvRef(t, got).RowID]++ + } + for _, ref := range published { + if seen[ref.RowID] != 1 { + t.Errorf("%s was delivered %d time(s), want exactly 1 — Unsubscribe discarded a claimed event", ref.RowID, seen[ref.RowID]) + } + } +} + +// TestSubscribeWatchdogExitsOnClose defends Close as a complete shutdown of the +// event plane. The watchdog selects on the subscribe context, and nats.go closes +// neither a ConsumeContext's buffer nor its own subscription channel when the +// connection closes — so with an UNCANCELLED context (a Server whose root +// context outlives its fabric, the common shape) Close left the watchdog parked +// forever, holding a consumer on a dead connection. +// +// The context here is deliberately never cancelled: Close alone must do it. +func TestSubscribeWatchdogExitsOnClose(t *testing.T) { + // Deliberately NOT parallel: this counts goroutines process-wide, and a + // sibling test's live Subscribe is indistinguishable from a leak of this + // one's. Go runs non-parallel tests while the parallel ones are paused. + f := newFabric(t, Config{}) + + subject, err := CommsSubject("t-watchdog", KindMessagePosted) + if err != nil { + 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" + baseline := countGoroutinesWith(t, marker) + + // Rooted at context.Background() because this is a test root, and an + // uncancelled context is the whole point of the test. + unsub, err := f.Subscribe(context.Background(), subject, func(EventRef) {}) + if err != nil { + t.Fatalf("Subscribe: %v", err) + } + defer unsub() + + // Prove the watchdog started, so the assertion below distinguishes "exited" + // from "never ran". + pollUntil(t, "the subscribe watchdog to start", func() bool { + return countGoroutinesWith(t, marker) > baseline + }) + + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + pollUntil(t, "the subscribe watchdog to exit after Close", func() bool { + return countGoroutinesWith(t, marker) <= baseline + }) +} + +// countGoroutinesWith counts live goroutines whose stack mentions marker. Used +// instead of a goleak dependency, which this module does not carry: the whole +// assertion is one count, and adding a test-only dep for it is not worth it. +func countGoroutinesWith(t *testing.T, marker string) int { + t.Helper() + buf := make([]byte, 1<<16) + for { + n := runtime.Stack(buf, true) + // A full buffer means the dump was truncated and a goroutine may have + // been cut off mid-frame, so the count would be wrong. + if n < len(buf) { + return strings.Count(string(buf[:n]), marker) + } + buf = make([]byte, 2*len(buf)) + } +} diff --git a/go/internal/fabric/eventref.go b/go/internal/fabric/eventref.go new file mode 100644 index 000000000..bbfca07f5 --- /dev/null +++ b/go/internal/fabric/eventref.go @@ -0,0 +1,111 @@ +package fabric + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "strconv" +) + +// EventKind names a class of comms event. It is the last token of a comms +// subject, so its value must be a valid single subject token (see +// ValidSubjectToken) — every constant below is snake_case for that reason. +type EventKind string + +// The comms event kinds. These mirror the write-through publish sites in +// go/internal/comms: each one says "a row of this class changed", never what +// changed — the subscriber re-reads the row (see EventRef). +const ( + KindAccountChanged EventKind = "account_changed" + KindChannelGroupChanged EventKind = "channel_group_changed" + KindChannelChanged EventKind = "channel_changed" + KindAgentWorkspaceChanged EventKind = "agent_workspace_changed" + KindMessagePosted EventKind = "message_posted" + KindMessageUpdated EventKind = "message_updated" + KindTopicUpserted EventKind = "topic_upserted" +) + +// EventRef is a compact reference to a committed Postgres row — event kind, row +// id, tenant — and NEVER a copy of the row's payload (§Global Constraints: +// "Postgres is the sole durability source of truth"). A subscriber receiving one +// re-reads the row it names. +// +// That is what keeps JetStream a transport rather than a second store: a +// replayed or double-delivered EventRef re-reads the same row and is idempotent, +// and a dropped one degrades to a delivery-cursor sweep. It also bounds the wire +// payload to a few dozen bytes regardless of how large the underlying row is. +type EventRef struct { + // Tenant is the owning tenant id (store.TenantID's underlying type is + // string). It is also the tenant token of the subject the ref rides, so the + // subscriber can scope its re-read without trusting the subject. + Tenant string `json:"tenant"` + // Kind is the class of event. + Kind EventKind `json:"kind"` + // RowID is the primary-row id the subscriber re-reads from Postgres. + RowID string `json:"row_id"` +} + +// encode marshals the ref for the wire as JSON. JSON rather than proto or a +// packed encoding on purpose: the payload is three short strings, so the size +// difference is noise, while a `nats sub` on a live subject stays readable and a +// later additive field is forward-compatible with an older subscriber. +func (r EventRef) encode() ([]byte, error) { + b, err := json.Marshal(r) + if err != nil { + return nil, fmt.Errorf("fabric: encoding event ref %s/%s: %w", r.Kind, r.RowID, err) + } + return b, nil +} + +// msgID derives the JetStream deduplication id for the ref. Deterministic in +// (tenant, kind, row id) so two Servers publishing the same logical change — or +// one Server retrying a publish whose ack was lost — collapse to one stored +// message inside the stream's duplicate window. +// +// Hashed rather than concatenated because the three fields are opaque: a raw +// "tenant|kind|rowid" join could collide across differently-split ids, and a +// long row id would push the Nats-Msg-Id header wide for no benefit. +func (r EventRef) msgID() string { + h := sha256.New() + // Length-prefixed so no field boundary is ambiguous. hash.Hash.Write is + // documented never to return an error, so there is none to handle. + for _, f := range [...]string{r.Tenant, string(r.Kind), r.RowID} { + h.Write([]byte(strconv.Itoa(len(f)))) + h.Write([]byte{':'}) + h.Write([]byte(f)) + } + return hex.EncodeToString(h.Sum(nil)) +} + +// valid reports whether the ref is publishable: every field is required, and +// tenant and kind must be valid subject tokens because they are subject tokens. +// Checked publish-side so a malformed ref fails at its origin, with the caller's +// stack, instead of becoming an undecodable DLQ entry later. +func (r EventRef) valid() error { + if err := ValidSubjectToken("tenant", r.Tenant); err != nil { + return err + } + if err := ValidSubjectToken("event kind", string(r.Kind)); err != nil { + return err + } + if r.RowID == "" { + return fmt.Errorf("fabric: event ref %s/%s has an empty row id", r.Tenant, r.Kind) + } + return nil +} + +// decodeEventRef parses a wire payload back into an EventRef, rejecting one +// missing any required field. A ref that decodes but names nothing is worse than +// a decode error: the subscriber would re-read row "" and see the miss as +// "nothing changed". +func decodeEventRef(b []byte) (EventRef, error) { + var r EventRef + if err := json.Unmarshal(b, &r); err != nil { + return EventRef{}, fmt.Errorf("fabric: decoding event ref from %d bytes: %w", len(b), err) + } + if r.Tenant == "" || r.Kind == "" || r.RowID == "" { + return EventRef{}, fmt.Errorf("fabric: decoded event ref is incomplete (tenant=%q kind=%q row_id=%q)", r.Tenant, r.Kind, r.RowID) + } + return r, nil +} diff --git a/go/internal/fabric/eventref_test.go b/go/internal/fabric/eventref_test.go new file mode 100644 index 000000000..7094d3eb7 --- /dev/null +++ b/go/internal/fabric/eventref_test.go @@ -0,0 +1,160 @@ +package fabric + +import ( + "strings" + "testing" +) + +// TestEventRefCodecRoundTrip defends the wire contract: what a publisher encodes +// is exactly what a subscriber decodes. Every field matters — a lost tenant +// re-reads the wrong tenant's row, a lost kind mis-routes, a lost row id reads +// nothing. +func TestEventRefCodecRoundTrip(t *testing.T) { + t.Parallel() + for _, want := range []EventRef{ + {Tenant: "t1", Kind: KindMessagePosted, RowID: "msg-1"}, + {Tenant: "0d8f1a2b-3c4d", Kind: KindTopicUpserted, RowID: "topic/with-slash"}, + {Tenant: "unicode-ténant", Kind: KindChannelChanged, RowID: "ch-✓"}, + } { + t.Run(want.RowID, func(t *testing.T) { + t.Parallel() + b, err := want.encode() + if err != nil { + t.Fatalf("encode: %v", err) + } + got, err := decodeEventRef(b) + if err != nil { + t.Fatalf("decodeEventRef: %v", err) + } + if got != want { + t.Fatalf("round trip = %+v, want %+v", got, want) + } + }) + } +} + +// TestEventRefCarriesNoPayload defends the load-bearing global constraint: +// EventRef is a reference, never a copy. A field added later that carried row +// content would make JetStream a second store of committed state. Asserted +// structurally — the encoding has exactly the three reference fields. +func TestEventRefCarriesNoPayload(t *testing.T) { + t.Parallel() + b, err := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "m1"}.encode() + if err != nil { + t.Fatalf("encode: %v", err) + } + got := string(b) + if want := `{"tenant":"t1","kind":"message_posted","row_id":"m1"}`; got != want { + t.Fatalf("encoded = %s, want %s (an extra field means a payload crept onto the wire)", got, want) + } +} + +// TestDecodeEventRefRejectsIncomplete defends the fail-closed decode. A ref that +// parses but names nothing is worse than a parse error: the subscriber would +// re-read row "" and read the miss as "nothing changed" — a silently dropped +// event. +func TestDecodeEventRefRejectsIncomplete(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + body string + }{ + {"empty bytes", ``}, + {"not json", `not json at all`}, + {"empty object", `{}`}, + {"missing tenant", `{"kind":"message_posted","row_id":"m1"}`}, + {"missing kind", `{"tenant":"t1","row_id":"m1"}`}, + {"missing row id", `{"tenant":"t1","kind":"message_posted"}`}, + {"blank row id", `{"tenant":"t1","kind":"message_posted","row_id":""}`}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got, err := decodeEventRef([]byte(tc.body)); err == nil { + t.Fatalf("decodeEventRef(%q) = %+v, want an error", tc.body, got) + } + }) + } +} + +// TestDecodeEventRefIgnoresUnknownFields defends forward compatibility, which is +// why the codec is JSON: an older subscriber must keep working against a +// publisher that added a field, rather than parking every event on the DLQ +// during a rolling deploy. +func TestDecodeEventRefIgnoresUnknownFields(t *testing.T) { + t.Parallel() + body := `{"tenant":"t1","kind":"message_posted","row_id":"m1","future_field":42}` + got, err := decodeEventRef([]byte(body)) + if err != nil { + t.Fatalf("decodeEventRef with an unknown field: %v", err) + } + want := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "m1"} + if got != want { + t.Fatalf("decoded = %+v, want %+v", got, want) + } +} + +// TestMsgIDIsDeterministic defends the dedup key's whole purpose: two publishers +// of the same logical change must derive the SAME id (or JetStream stores two +// copies and the subscriber handles it twice), and two different changes must +// derive DIFFERENT ids (or the second is silently deduped away — a lost event). +func TestMsgIDIsDeterministic(t *testing.T) { + t.Parallel() + base := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "m1"} + if a, b := base.msgID(), base.msgID(); a != b { + t.Fatalf("msgID is not stable: %q vs %q", a, b) + } + same := EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "m1"} + if base.msgID() != same.msgID() { + t.Fatal("two refs with identical fields must derive one msg id, or dedup never fires") + } + + distinct := map[string]EventRef{ + "base": base, + "other tenant": {Tenant: "t2", Kind: KindMessagePosted, RowID: "m1"}, + "other kind": {Tenant: "t1", Kind: KindMessageUpdated, RowID: "m1"}, + "other row": {Tenant: "t1", Kind: KindMessagePosted, RowID: "m2"}, + // The field-boundary case a naive concatenation gets wrong: "t1|a" and + // "t|1a" would join to the same string. + "boundary a": {Tenant: "t1", Kind: KindMessagePosted, RowID: "am1"}, + "boundary b": {Tenant: "t", Kind: KindMessagePosted, RowID: "1am1"}, + } + seen := make(map[string]string, len(distinct)) + for name, ref := range distinct { + id := ref.msgID() + if prev, dup := seen[id]; dup { + t.Fatalf("msg id collision: %q and %q derive %q; the second event would be deduped away", prev, name, id) + } + seen[id] = name + } +} + +// TestEventRefValid defends publish-side validation: a malformed ref must fail +// at its origin with the caller's stack, not become an undecodable DLQ entry +// discovered by an operator later. +func TestEventRefValid(t *testing.T) { + t.Parallel() + if err := (EventRef{Tenant: "t1", Kind: KindMessagePosted, RowID: "m1"}).valid(); err != nil { + t.Fatalf("a complete ref must be valid: %v", err) + } + for _, tc := range []struct { + name string + ref EventRef + }{ + {"no tenant", EventRef{Kind: KindMessagePosted, RowID: "m1"}}, + {"no kind", EventRef{Tenant: "t1", RowID: "m1"}}, + {"no row id", EventRef{Tenant: "t1", Kind: KindMessagePosted}}, + {"tenant with a dot", EventRef{Tenant: "t.1", Kind: KindMessagePosted, RowID: "m1"}}, + {"kind with a wildcard", EventRef{Tenant: "t1", Kind: "message*", RowID: "m1"}}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + err := tc.ref.valid() + if err == nil { + t.Fatalf("%+v: want an error", tc.ref) + } + if !strings.Contains(err.Error(), "fabric:") { + t.Errorf("error %q should be attributed to the fabric package", err) + } + }) + } +} diff --git a/go/internal/fabric/fabric.go b/go/internal/fabric/fabric.go new file mode 100644 index 000000000..94e670a0b --- /dev/null +++ b/go/internal/fabric/fabric.go @@ -0,0 +1,372 @@ +package fabric + +import ( + "context" + "errors" + "fmt" + "log/slog" + "sync" + "time" + + compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// Unsubscribe tears down a Subscribe. Idempotent: calling it twice is safe, so a +// deferred Unsubscribe beside an explicit one is not a bug. +type Unsubscribe func() + +// EventFabric is the comms/delivery event seam (frozen, §T3). It carries +// compact references on JetStream — durable at-least-once fan-out — never +// payloads; see EventRef. +type EventFabric interface { + Publish(ctx context.Context, subject string, ref EventRef) error + Subscribe(ctx context.Context, subject string, fn func(EventRef)) (Unsubscribe, error) +} + +// RunnerFabric is the Server↔Runner async seam (frozen, §T3): per-Runner +// command push out, queue-grouped event fan-in back. It rides core NATS +// (best-effort); a command to an offline Runner is recovered by the +// delivery-cursor sweep, not by a stream. +// +// The typed request/reply legs (enrollment, the unary Relay* calls, +// FetchSecrets, the bulk FetchAgentConfig pull) are NOT here — they stay on the +// Connect/gRPC edge (§Q3, OQ-5 Variant B), which keeps deadline propagation and +// typed proto errors. +type RunnerFabric interface { + SendCommand(ctx context.Context, runnerID string, cmd *compassv1internal.SessionsResponse) error + Events(ctx context.Context) (<-chan RunnerEvent, error) +} + +// Config configures a Fabric. Only URL is required; every other field has a +// documented default from stream.go, so the common case is +// fabric.New(fabric.Config{URL: natsURL}). +type Config struct { + // URL is the NATS connection string. A single node and a cluster differ + // only here (§Q3: scaling NATS is never an application mode) — a + // comma-separated seed list is a cluster. + URL string + + // Name labels this connection in NATS monitoring (`nats server report + // connections`). Defaults to "compass". + Name string + + // Options are extra nats.Options appended after the fabric's own, so a + // caller can add credentials or TLS without this package growing a field per + // auth mechanism. A later option wins over an earlier one. + // + // The fabric reserves the CONNECTION LIFECYCLE for its own shutdown + // coordination: Close observes the connection's status directly rather than + // through a nats.ClosedHandler, so a caller may add its own ClosedHandler + // (a monitoring hook, say) without disarming Close's drain wait. Replacing + // the fabric's DisconnectErrHandler/ReconnectHandler is likewise safe — + // those are log-only, and a caller that replaces them loses the fabric's + // outage diagnostics, nothing more. + Options []nats.Option + + // StreamName overrides DefaultStreamName. + StreamName string + // MaxDeliver overrides DefaultMaxDeliver. It is the total number of + // delivery ATTEMPTS, not retries: MaxDeliver=1 parks on the first failure + // with no retry at all, and MaxDeliver=5 means the fifth failing attempt + // parks. + // + // The budget is enforced twice, by design (§Q3): the app-level check in + // retryOrPark parks at the budget, and the consumer's server-side + // MaxDeliver is the backstop for the case where the app-level check cannot + // run (unreadable metadata — which already parks). Both derive from THIS + // field, so every Server instance subscribing to a given subject MUST run + // the same fabric Config: the durable consumer is shared, so a divergent + // MaxDeliver would make the shared consumer's server-side budget + // flip-flop with whichever instance last ran CreateOrUpdateConsumer. One + // stack config for the whole deployment (RIG-2861 topology) is what + // guarantees it; the fabric deliberately does not probe for drift. + MaxDeliver int + // AckWait overrides DefaultAckWait. + AckWait time.Duration + // DuplicateWindow overrides DefaultDuplicateWindow. + DuplicateWindow time.Duration + // MaxAge overrides DefaultMaxAge. + MaxAge time.Duration + // Replicas overrides DefaultReplicas; set 3 against a clustered NATS. + Replicas int + + // RunnerEventBuffer is the capacity of the channel Events returns. A + // consumer slower than the Runner event rate makes the core-NATS + // subscription's own buffer the backstop — NATS drops for a slow consumer + // rather than blocking the connection, which is the intended best-effort + // semantic. Defaults to 256. + RunnerEventBuffer int + + // Log carries the diagnostics that cannot be returned to a caller — a + // dropped Runner event, a parked comms event, an async consume error. A nil + // Log falls back to slog.Default (the house convention). + Log *slog.Logger +} + +func (c Config) streamName() string { + if c.StreamName != "" { + return c.StreamName + } + return DefaultStreamName +} + +func (c Config) maxDeliver() int { + if c.MaxDeliver > 0 { + return c.MaxDeliver + } + return DefaultMaxDeliver +} + +// deliveryBudget is maxDeliver as the unsigned count JetStream reports in +// MsgMetadata.NumDelivered. maxDeliver is an int only because ConsumerConfig +// takes one; the guard makes the widening provably wrap-free rather than +// relying on the accessor's own invariant holding forever. +func (c Config) deliveryBudget() uint64 { + n := c.maxDeliver() + if n <= 0 { + return DefaultMaxDeliver + } + return uint64(n) +} + +func (c Config) ackWait() time.Duration { + if c.AckWait > 0 { + return c.AckWait + } + return DefaultAckWait +} + +func (c Config) duplicateWindow() time.Duration { + if c.DuplicateWindow > 0 { + return c.DuplicateWindow + } + return DefaultDuplicateWindow +} + +func (c Config) maxAge() time.Duration { + if c.MaxAge > 0 { + return c.MaxAge + } + return DefaultMaxAge +} + +func (c Config) replicas() int { + if c.Replicas > 0 { + return c.Replicas + } + return DefaultReplicas +} + +func (c Config) runnerEventBuffer() int { + if c.RunnerEventBuffer > 0 { + return c.RunnerEventBuffer + } + return defaultRunnerEventBuffer +} + +func (c Config) logger() *slog.Logger { + if c.Log != nil { + return c.Log + } + return slog.Default() +} + +// Fabric is the one NATS client: it implements both EventFabric and +// RunnerFabric over a single connection, because the record gives each party +// exactly one ("Each Runner holds ONE fabric connection … each Server +// likewise"). Safe for concurrent use. +type Fabric struct { + cfg Config + log *slog.Logger + + nc *nats.Conn + js jetstream.JetStream + + // streamMu guards the lazily-ensured stream handle (see ensureStream). + streamMu sync.Mutex + stream jetstream.Stream + + // closeOnce makes Close idempotent, and closed gates new work so a + // Subscribe racing a Close cannot register a consumer on a dying + // connection. + closeOnce sync.Once + closeErr error + closedMu sync.RWMutex + closed bool + + // teardown is closed by Close at the START of shutdown, before the drain. + // It is the signal the runner-events pump and every Subscribe watchdog + // select on, so a Close with an uncancelled subscribe/Events context still + // tears them down — nats.go closes neither a ChanSubscription's channel nor + // a ConsumeContext's buffer on connection close, so relying on those to + // unblock is a leaked goroutine and, for Events, a channel that never + // closes under a ranging consumer. + // + // Distinct from the closed/closedMu fail-closed gate above and not a + // substitute for it: that gate answers "may I start new work", this one + // answers "stop the work already running". + teardown chan struct{} +} + +// Compile-time proof Fabric satisfies both frozen seams. Cheap here, and it +// fails the build rather than a consumer's wiring if a signature drifts. +var ( + _ EventFabric = (*Fabric)(nil) + _ RunnerFabric = (*Fabric)(nil) +) + +// New connects to NATS and returns the fabric. It does not create the JetStream +// stream — that happens lazily on the first Publish/Subscribe, because the +// frozen signature carries no context and rooting a fresh one here would sever +// the caller's cancellation chain. +func New(cfg Config) (*Fabric, error) { + if cfg.URL == "" { + return nil, errors.New("fabric: Config.URL is required (the NATS connection string)") + } + name := cfg.Name + if name == "" { + name = "compass" + } + log := cfg.logger() + + // f is captured by the connection handlers below so they can tell an + // unplanned disconnect (a real outage the operator must see) from the drain + // Close performs (expected, and noise if logged). It is assigned before + // nats.Connect can fire either. + f := &Fabric{ + cfg: cfg, + log: log, + teardown: make(chan struct{}), + } + + // Reconnect forever: NATS being briefly unreachable is an outage to ride + // out, not a reason to abandon the connection — the record's degrade path is + // "fabric outage degrades to sweep-recovered delivery", which requires the + // client to come back on its own. + opts := append([]nats.Option{ + nats.Name(name), + nats.MaxReconnects(-1), + nats.DisconnectErrHandler(func(_ *nats.Conn, err error) { + if f.checkOpen() != nil { + return // Close's own drain. + } + log.Warn("fabric: nats disconnected; delivery degrades to the cursor sweep until reconnect", "error", err) + }), + nats.ReconnectHandler(func(nc *nats.Conn) { + log.Info("fabric: nats reconnected", "url", nc.ConnectedUrl()) + }), + }, cfg.Options...) + + nc, err := nats.Connect(cfg.URL, opts...) + if err != nil { + return nil, fmt.Errorf("fabric: connecting to nats at %q: %w", cfg.URL, err) + } + js, err := jetstream.New(nc) + if err != nil { + nc.Close() + return nil, fmt.Errorf("fabric: creating jetstream context: %w", err) + } + f.nc, f.js = nc, js + return f, nil +} + +// 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 +// work this fabric started — the runner-events pump and every live Subscribe — +// even when their contexts are still uncancelled, because nats.go leaves both a +// ChanSubscription's channel and a ConsumeContext's buffer open on connection +// close and neither would ever unblock on its own. +// +// Close returns only once the drain has completed (bounded by closeTimeout), so +// a caller that returns from Close can rely on the flush having happened. +// Idempotent — the first call's result is returned to every caller. +func (f *Fabric) Close() error { + f.closeOnce.Do(func() { + f.closedMu.Lock() + f.closed = true + f.closedMu.Unlock() + // Before the drain: the pump and the subscribe watchdogs must stop + // consuming and hand their consumers back while the connection is + // still usable, so the drain has something coherent to flush. + close(f.teardown) + + // Register the CLOSED listener BEFORE Drain: StatusChanged reports only + // future transitions and does not replay one that already fired, so a + // listener installed after Drain could miss the close entirely. Using + // the connection's own status (not a nats.ClosedHandler option) means a + // caller's Config.Options cannot overwrite the drain-completion signal. + closed := f.nc.StatusChanged(nats.CLOSED) + defer f.nc.RemoveStatusListener(closed) + + if err := f.nc.Drain(); err != nil { + f.closeErr = fmt.Errorf("fabric: draining nats connection: %w", err) + // Drain refused (already closed, or the connection is gone); close + // outright so the socket and its goroutines are not leaked. That + // also drives the connection to CLOSED, so there is nothing left + // worth waiting for on this path. + f.nc.Close() + return + } + // Drain is asynchronous: it returns as soon as the connection enters + // DRAINING. Wait for CLOSED so Close does not return mid-flush. Guard + // the already-closed case first: if the connection reached CLOSED + // between Drain and here, the transition has already fired and the + // listener will never see it. + if f.nc.IsClosed() { + return + } + select { + case <-closed: + case <-time.After(closeTimeout): + // The drain WAS initiated and continues in the background; the + // bound only exists so a wedged server cannot hang shutdown + // forever, which would be a worse failure than an unconfirmed + // flush. + f.log.Warn("fabric: nats drain did not complete within the close timeout; shutting down anyway", + "timeout", closeTimeout) + } + }) + return f.closeErr +} + +// closeTimeout bounds how long Close waits for the asynchronous drain to reach +// a closed connection. Generous, because the wait is what makes Close's +// no-lost-publish contract real; bounded, because a wedged or vanished server +// must not hang a process's shutdown. +const closeTimeout = 10 * time.Second + +// flushTimeout bounds a flush whose caller's context carries no deadline. A +// flush is a round-trip to the server, and nats.go refuses a context without +// one; a Publish/Events call whose ctx is a process-lifetime context still +// needs the write confirmed in bounded time rather than hanging until shutdown. +const flushTimeout = 10 * time.Second + +// flush round-trips the connection so a preceding core-NATS publish or +// subscribe is known to have reached the server. It derives from the caller's +// ctx — never re-roots — adding a deadline only when ctx has none, so +// cancellation still propagates. +func (f *Fabric) flush(ctx context.Context) error { + if _, ok := ctx.Deadline(); !ok { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, flushTimeout) + defer cancel() + } + return f.nc.FlushWithContext(ctx) +} + +// errClosed is returned by any operation attempted after Close. +var errClosed = errors.New("fabric: closed") + +// checkOpen refuses work on a closed fabric. Fail-closed: a Publish that +// silently no-ops after shutdown would look like a delivered event. +func (f *Fabric) checkOpen() error { + f.closedMu.RLock() + defer f.closedMu.RUnlock() + if f.closed { + return errClosed + } + return nil +} diff --git a/go/internal/fabric/fabric_test.go b/go/internal/fabric/fabric_test.go new file mode 100644 index 000000000..57cf220fe --- /dev/null +++ b/go/internal/fabric/fabric_test.go @@ -0,0 +1,376 @@ +package fabric + +import ( + "context" + "errors" + "log/slog" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + natsserver "github.com/nats-io/nats-server/v2/test" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// gate is the bound on every positive async assertion in this package. Long +// enough that a loaded CI box does not fail a correct implementation, short +// enough that a genuine hang fails the test rather than the suite's deadline. +// It is a FAILURE bound, never a wait: no test sleeps for delivery, and no test +// retries. +const gate = 10 * time.Second + +// testServer runs an in-process NATS with JetStream for one test, and returns +// its client URL. Hermetic by construction: Port -1 picks a free port, so +// parallel tests never collide, and StoreDir under t.TempDir means the stream's +// file storage is discarded with the test. +// +// SyncInterval is the record's `sync_interval: 100ms`. It is a SERVER option, +// not a stream one (see SUBJECTS.md), so this is the only place in the repo the +// tests can exercise the record's value rather than the server default. +func testServer(t *testing.T) string { + t.Helper() + srv := natsserver.RunServer(&server.Options{ + Port: -1, + JetStream: true, + StoreDir: t.TempDir(), + SyncInterval: 100 * time.Millisecond, + NoLog: true, + NoSigs: true, + }) + t.Cleanup(srv.Shutdown) + return srv.ClientURL() +} + +// newFabric returns a Fabric against a fresh in-process server, closed at test +// end. cfg.URL is filled in; every other field the caller sets is honored. +func newFabric(t *testing.T, cfg Config) *Fabric { + t.Helper() + if cfg.URL == "" { + cfg.URL = testServer(t) + } + f, err := New(cfg) + if err != nil { + t.Fatalf("New: %v", err) + } + t.Cleanup(func() { + if err := f.Close(); err != nil { + t.Errorf("Close: %v", err) + } + }) + return f +} + +// testCtx is a context bounded by the gate, so a wedged test fails at the gate +// with its own message instead of at the package deadline. Rooted at +// context.Background() because this is a test root — the only place besides +// main() where minting a root context is correct. +func testCtx(t *testing.T) context.Context { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), gate) + t.Cleanup(cancel) + return ctx +} + +// recvRef takes the next EventRef from ch, failing at the gate. +func recvRef(t *testing.T, ch <-chan EventRef) EventRef { + t.Helper() + select { + case r := <-ch: + return r + case <-time.After(gate): + t.Fatalf("no event within %s", gate) + return EventRef{} + } +} + +// pollUntil blocks until cond reports true, failing at the gate. For the two +// states in this package that expose no event to block on — a goroutine count +// read out of runtime.Stack, and a JetStream consumer's server-side +// NumAckPending — there is no channel or WaitGroup to receive from, so a +// bounded poll is the only gate available. It is still a FAILURE bound and not +// a wait: it returns the moment cond holds, and a genuine hang fails the test +// with what it was waiting for. +func pollUntil(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.After(gate) + // A ticker rather than a sleep so the loop is driven by a channel the + // deadline can race, and the failure is the gate's rather than the tick's. + tick := time.NewTicker(time.Millisecond) + defer tick.Stop() + for { + if cond() { + return + } + select { + case <-tick.C: + case <-deadline: + t.Fatalf("waited %s for %s", gate, what) + return + } + } +} + +// TestCloseWaitsForDrain defends Close's documented promise that "a shutdown +// does not lose an already-published event". nc.Drain is ASYNCHRONOUS — it +// returns as soon as the connection enters DRAINING — so a Close that returned +// straight after it returned while the flush was still in flight, and the +// contract was a hope rather than a fact. +// +// Deterministic, no sleep: the connection reporting CLOSED is a state Close must +// already have observed by the time it returns. +func TestCloseWaitsForDrain(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f, err := New(Config{URL: testServer(t)}) + if err != nil { + t.Fatalf("New: %v", err) + } + + subject, err := CommsSubject("t-drain", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + if err := f.Publish(ctx, subject, EventRef{Tenant: "t-drain", Kind: KindMessagePosted, RowID: "m1"}); err != nil { + t.Fatalf("Publish: %v", err) + } + + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + // Immediately, with no intervening wait: if Close returned mid-drain the + // connection would still be DRAINING here. + if !f.nc.IsClosed() { + t.Fatal("Close returned while the connection was still draining; the flush it promises was not confirmed") + } +} + +// TestCloseIsPromptWithCallerClosedHandler defends Close's drain wait against a +// caller's own nats.ClosedHandler. Config.Options are appended AFTER the +// fabric's, and a nats.Option is just a mutator on nats.Options — so when the +// drain-completion signal came from a fabric-installed ClosedHandler, a caller +// adding one (a supported use: a monitoring hook) silently overwrote it, and +// every Close burned the full closeTimeout and then logged a false "drain did +// not complete" warning. Close now watches the connection's own status, which +// no option can clobber. +func TestCloseIsPromptWithCallerClosedHandler(t *testing.T) { + t.Parallel() + var callerRan atomic.Bool + f, err := New(Config{ + URL: testServer(t), + Options: []nats.Option{ + nats.ClosedHandler(func(_ *nats.Conn) { callerRan.Store(true) }), + }, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + start := time.Now() + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + // Close must return well under closeTimeout (10s): if it depended on the + // clobbered handler it would burn the full timeout. A generous ceiling + // keeps the test non-flaky while still failing the 10s-stall bug. + if elapsed := time.Since(start); elapsed > 2*time.Second { + t.Fatalf("Close took %s; expected prompt return (caller ClosedHandler disarmed the drain signal?)", elapsed) + } + // The caller's handler is NOT synchronous with Close: nc.close notifies + // status listeners (which is what Close now waits on) before it pushes + // ClosedCB onto the connection's async-callback dispatcher, so the handler + // lands just after Close returns. pollUntil is the suite's FAILURE bound, + // not a wait — a suppressed handler still fails, it just takes `gate`. + pollUntil(t, "the caller's own ClosedHandler to run (the fabric must not suppress it)", callerRan.Load) +} + +// TestNewRequiresURL defends the fail-closed constructor: a Fabric with no +// connection string must not come back as a usable object that fails later at +// an arbitrary Publish. +func TestNewRequiresURL(t *testing.T) { + t.Parallel() + if _, err := New(Config{}); err == nil { + t.Fatal("New with no URL: want an error, got nil") + } +} + +// TestNewUnreachableURL defends that New actually connects. If it deferred the +// dial, a misconfigured NATS URL would surface as a mysterious publish failure +// at the first comms write instead of at startup. +func TestNewUnreachableURL(t *testing.T) { + t.Parallel() + // Port 1 on the loopback with no listener: connection refused, fast. + if _, err := New(Config{URL: "nats://127.0.0.1:1", Options: []nats.Option{nats.Timeout(2 * time.Second)}}); err == nil { + t.Fatal("New against an unreachable server: want an error, got nil") + } +} + +// TestFabricImplementsBothSeamsOverOneConnection defends the record's +// one-connection-per-party contract. That *Fabric satisfies both interfaces is +// a compile-time assertion in fabric.go; what a test can add is that the two +// seams are the SAME object over ONE nats.Conn — if EventFabric and +// RunnerFabric were ever split into two clients, each Runner and Server would +// hold two connections, which the record forbids. +func TestFabricImplementsBothSeamsOverOneConnection(t *testing.T) { + t.Parallel() + f := newFabric(t, Config{}) + var ( + ef EventFabric = f + rf RunnerFabric = f + ) + efFabric, ok := ef.(*Fabric) + if !ok { + t.Fatalf("EventFabric is backed by %T, want *Fabric", ef) + } + rfFabric, ok := rf.(*Fabric) + if !ok { + t.Fatalf("RunnerFabric is backed by %T, want *Fabric", rf) + } + if efFabric != rfFabric { + t.Fatal("the two seams must be one object, or a party holds two fabrics") + } + if efFabric.nc != rfFabric.nc { + t.Fatal("the two seams must share one nats connection") + } +} + +// TestCloseIsIdempotentAndFailsClosed defends two things at once: Close can be +// called twice (a deferred Close beside an explicit one is not a bug), and +// post-Close work is refused rather than silently no-oping — a Publish that +// quietly succeeded after shutdown would look like a delivered event. +func TestCloseIsIdempotentAndFailsClosed(t *testing.T) { + t.Parallel() + f, err := New(Config{URL: testServer(t)}) + if err != nil { + t.Fatalf("New: %v", err) + } + if err := f.Close(); err != nil { + t.Fatalf("first Close: %v", err) + } + if err := f.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } + + ctx := testCtx(t) + subject, err := CommsSubject("t-closed", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + if err := f.Publish(ctx, subject, EventRef{Tenant: "t-closed", Kind: KindMessagePosted, RowID: "m1"}); !errors.Is(err, errClosed) { + t.Fatalf("Publish after Close: want errClosed, got %v", err) + } + 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.SendCommand(ctx, "r1", nil); !errors.Is(err, errClosed) { + t.Fatalf("SendCommand after Close: want errClosed, got %v", err) + } + if _, err := f.Events(ctx); !errors.Is(err, errClosed) { + t.Fatalf("Events after Close: want errClosed, got %v", err) + } +} + +// TestDurableNameHasNoDots defends the JetStream naming constraint the +// subject→durable mapping exists for: a consumer name containing a "." is +// rejected by the server, so a subject passed through verbatim would fail every +// Subscribe. +func TestDurableNameHasNoDots(t *testing.T) { + t.Parallel() + subject, err := CommsSubject("tenant-a", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + got := durableName(subject) + if strings.ContainsAny(got, ".*> \t") { + t.Fatalf("durableName(%q) = %q: contains a character JetStream forbids in a consumer name", subject, got) + } + // sha256("compass.tenant-a.comms.message_posted"), hex, behind the + // greppable "comms-" prefix — see durableName's injectivity note. + if want := "comms-f48b30590555bc3c1a67cfe133311032c5d8d7115e7d3b70fc89ae6156e25211"; got != want { + t.Fatalf("durableName(%q) = %q, want %q", subject, got, want) + } +} + +// TestDurableNameIsInjectiveAcrossUnderscoreSplits defends the property the old +// "."→"_" substitution only claimed to have. ValidSubjectToken permits "_", so +// two distinct, individually-valid comms subjects collapsed to one durable +// name — and because Subscribe uses CreateOrUpdateConsumer, the second +// subscriber would silently re-point the first's shared consumer FilterSubject: +// cross-tenant mis-delivery. +func TestDurableNameIsInjectiveAcrossUnderscoreSplits(t *testing.T) { + t.Parallel() + // tenant "a" + kind "b_comms_c" vs tenant "a_comms_b" + kind "c". + a := "compass.a.comms.b_comms_c" + b := "compass.a_comms_b.comms.c" + if durableName(a) == durableName(b) { + t.Fatalf("durableName not injective: %q and %q both map to %q", a, b, durableName(a)) + } +} + +// TestStreamConfigMatchesTheRecord defends the stream values §Q3 fixes: file +// storage (not memory — the whole point of putting comms on JetStream), the +// exact comms wildcard (a wider one would capture client.* traffic; a narrower +// one would silently drop a tenant), and limits retention (so a bounded replay +// stays possible). +func TestStreamConfigMatchesTheRecord(t *testing.T) { + t.Parallel() + cfg := Config{URL: "nats://ignored"}.streamConfig() + if cfg.Name != DefaultStreamName { + t.Errorf("stream name = %q, want %q", cfg.Name, DefaultStreamName) + } + if len(cfg.Subjects) != 1 || cfg.Subjects[0] != "compass.*.comms.*" { + t.Errorf("stream subjects = %v, want [compass.*.comms.*]", cfg.Subjects) + } + if cfg.Storage != jetstream.FileStorage { + t.Errorf("stream storage = %v, want file storage (durability is the reason comms rides JetStream)", cfg.Storage) + } + if cfg.Retention != jetstream.LimitsPolicy { + t.Errorf("stream retention = %v, want limits", cfg.Retention) + } + if cfg.Duplicates <= 0 { + t.Error("stream duplicate window must be positive, or WithMsgID dedup is a no-op") + } +} + +// TestConsumerConfigMatchesTheRecord defends the consumer values §Q3 fixes: +// explicit acks and a FINITE MaxDeliver. An unlimited MaxDeliver (the server +// default) would redeliver a poison message forever and the DLQ would never be +// reached. +func TestConsumerConfigMatchesTheRecord(t *testing.T) { + t.Parallel() + subject, err := CommsSubject("t1", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + cfg := Config{}.consumerConfig(subject) + if cfg.AckPolicy != jetstream.AckExplicitPolicy { + t.Errorf("ack policy = %v, want explicit", cfg.AckPolicy) + } + if cfg.MaxDeliver <= 0 { + t.Errorf("max deliver = %d, want a finite budget so a poison message parks", cfg.MaxDeliver) + } + if cfg.FilterSubject != subject { + t.Errorf("filter subject = %q, want %q", cfg.FilterSubject, subject) + } + if cfg.Durable == "" { + t.Error("consumer must be durable so instances share one consumer and a restart resumes") + } +} + +// quietLogger routes the fabric's diagnostics into the test log instead of +// stderr. Tests that deliberately drive a failure path (a poison message, an +// undecodable payload) are EXPECTED to log at error level; sending that to +// t.Log keeps a passing run clean while preserving the output on a failure. +func quietLogger(t *testing.T) *slog.Logger { + t.Helper() + return slog.New(slog.NewTextHandler(testWriter{t}, &slog.HandlerOptions{Level: slog.LevelDebug})) +} + +// testWriter adapts *testing.T to io.Writer for quietLogger. +type testWriter struct{ t *testing.T } + +func (w testWriter) Write(p []byte) (int, error) { + w.t.Logf("fabric log: %s", strings.TrimRight(string(p), "\n")) + return len(p), nil +} diff --git a/go/internal/fabric/runner_fabric.go b/go/internal/fabric/runner_fabric.go new file mode 100644 index 000000000..324b46334 --- /dev/null +++ b/go/internal/fabric/runner_fabric.go @@ -0,0 +1,199 @@ +package fabric + +import ( + "context" + "errors" + "fmt" + + compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" + "github.com/nats-io/nats.go" + "google.golang.org/protobuf/proto" +) + +// defaultRunnerEventBuffer is the default capacity of the channel Events +// returns. Deep enough to absorb a burst while a consumer is mid-write, shallow +// enough that a genuinely stalled consumer shows up as NATS slow-consumer drops +// (the intended best-effort semantic) rather than as unbounded memory growth. +const defaultRunnerEventBuffer = 256 + +// RunnerEvent is one Runner→Server agent event as it arrives off the fan-in +// subject. +// +// It carries the decoded PublishEventsRequest whole rather than flattening its +// fields: the hub's existing write-through path already consumes exactly that +// proto (RunnerSeq for gap detection, SessionId for routing, Frame for +// classification, IdempotencyKey for at-most-once commit), so re-projecting it +// into local fields would only create a shape that has to be un-projected again +// at the seam — and would silently drop any field a later proto revision adds. +type RunnerEvent struct { + // RunnerID identifies the publishing Runner. The fan-in subject is shared + // (compass.runner.events, queue-grouped), so unlike the per-Runner command + // subject it does not encode the sender; the Runner stamps it in a header. + // Empty if the publisher set no header — a consumer is EXPECTED to treat an + // unattributed event as it treats an unknown frame: logged and counted, + // never silently trusted. The fabric itself does not enforce that; it + // reports the header verbatim. + // + // This value is publisher-asserted, NOT authenticated attribution: any + // publisher on the shared fan-in subject can stamp any id. A consumer must + // not treat it as identity — §Q2 puts the trust model on the Server's own + // resolution of the Runner, and this header is only a routing/diagnostic + // hint on top of it. + RunnerID string + // Event is the decoded envelope, never nil for an event read off the + // channel. + Event *compassv1internal.PublishEventsRequest +} + +// RunnerIDHeader is the header the Runner stamps its id into when publishing on +// the shared fan-in subject. A header rather than a proto field because the wire +// (PublishEventsRequest) is frozen and carries no runner id — it was a per-stream +// property back when the Runner had one stream, and pub/sub has no stream to +// carry it. +const RunnerIDHeader = "Compass-Runner-Id" + +// SendCommand pushes cmd to one Runner's command subject over core NATS. +// +// Core NATS, not JetStream, by design (§Q3): commands are best-effort. A command +// published while a Runner is offline is dropped, and that is correct — the +// delivery-cursor sweep in Postgres re-drives it on reconnect, so a stream here +// would add a second store of state Postgres already owns. +// +// Publish is fire-and-forget, so a nil return means "handed to the NATS client", +// not "the Runner got it". The flush against the caller's ctx below is what +// makes the error real: it surfaces a connection that cannot actually take the +// write, which is the failure a caller can act on. +func (f *Fabric) SendCommand(ctx context.Context, runnerID string, cmd *compassv1internal.SessionsResponse) error { + if err := f.checkOpen(); err != nil { + return err + } + if cmd == nil { + return fmt.Errorf("fabric: SendCommand to runner %q requires a command", runnerID) + } + subject, err := RunnerCommandSubject(runnerID) + if err != nil { + return err + } + data, err := proto.Marshal(cmd) + if err != nil { + return fmt.Errorf("fabric: marshaling command %q for runner %q: %w", cmd.GetRequestId(), runnerID, err) + } + if err := f.nc.Publish(subject, data); err != nil { + return fmt.Errorf("fabric: publishing command %q to %q: %w", cmd.GetRequestId(), subject, err) + } + if err := f.flush(ctx); err != nil { + return fmt.Errorf("fabric: flushing command %q to %q: %w", cmd.GetRequestId(), subject, err) + } + return nil +} + +// Events returns a channel of Runner events, closed once ctx is done or the +// Fabric is closed — whichever happens first, so a consumer ranging over it +// terminates on either shutdown path rather than only on its own cancellation. +// +// The subscription joins RunnerEventsQueue, so with several Servers connected +// each event is delivered to exactly one of them — the fan-in half of the Q2 +// topology, and what lets any Server handle any Runner's events without sticky +// connections. +// +// The channel is closed exactly once, after the subscription is drained: a +// receiver ranging over it sees every event NATS had already delivered and then +// a clean close, never a send on a closed channel. +func (f *Fabric) Events(ctx context.Context) (<-chan RunnerEvent, error) { + if err := f.checkOpen(); err != nil { + return nil, err + } + subject := RunnerEventsSubject() + + // A channel-backed subscription rather than a callback one: NATS owns the + // buffering and applies its slow-consumer policy to it, so a stalled + // receiver is dropped-and-reported instead of blocking the connection's + // dispatcher. + raw := make(chan *nats.Msg, f.cfg.runnerEventBuffer()) + sub, err := f.nc.QueueSubscribeSyncWithChan(subject, RunnerEventsQueue, raw) + if err != nil { + return nil, fmt.Errorf("fabric: queue-subscribing %q (group %q): %w", subject, RunnerEventsQueue, err) + } + // Flush so Events returns only once the server has registered the interest. + // Without it a caller that subscribes and then triggers a Runner would race + // its own first event — core NATS drops a message with no interest yet, and + // the drop would be invisible. + if err := f.flush(ctx); err != nil { + if derr := sub.Unsubscribe(); derr != nil { + f.log.WarnContext(ctx, "fabric: unsubscribing after a failed flush", "subject", subject, "error", derr) + } + return nil, fmt.Errorf("fabric: establishing the runner-events subscription on %q: %w", subject, err) + } + + out := make(chan RunnerEvent, f.cfg.runnerEventBuffer()) + go f.pumpRunnerEvents(ctx, sub, raw, out) + return out, nil +} + +// pumpRunnerEvents decodes raw messages onto out until ctx is done or the +// fabric is closed, then drains the subscription and closes out. It owns both +// the subscription's and the channel's lifetime, so there is exactly one place +// either is torn down. +// +// f.teardown is a load-bearing case in BOTH selects below, not a belt-and-braces +// duplicate of ctx.Done(): nats.go does not close a ChanSubscription's channel +// when the connection closes, so a Close with an uncancelled ctx would otherwise +// leave this goroutine parked on raw forever — and out never closed under a +// consumer ranging over it. +func (f *Fabric) pumpRunnerEvents(ctx context.Context, sub *nats.Subscription, raw <-chan *nats.Msg, out chan<- RunnerEvent) { + defer close(out) + defer func() { + // Drain rather than Unsubscribe: it lets NATS deliver what it has + // already accepted for this subject before removing the interest, + // mirroring Close's connection-level drain. + // + // An already-closed connection is the expected shutdown outcome, not a + // failure: Close closes f.teardown before nc.Drain(), and the + // connection-level drain reclaims every subscription itself. If that + // finishes before this descheduled goroutine reaches the defer, + // sub.Drain() returns ErrConnectionClosed for a subscription that WAS + // drained — so warning on it would be a false alarm on a clean path. + if err := sub.Drain(); err != nil && !errors.Is(err, nats.ErrConnectionClosed) { + f.log.WarnContext(ctx, "fabric: draining the runner-events subscription failed", + "subject", sub.Subject, "error", err) + } + }() + + for { + select { + case <-ctx.Done(): + return + case <-f.teardown: + return + case msg, ok := <-raw: + if !ok { + // The subscription's channel closed under us (connection gone). + return + } + ev, err := decodeRunnerEvent(msg) + if err != nil { + // Best-effort plane: one undecodable event must not tear down + // the fan-in for every other Runner. Surfaced, never silent. + f.log.ErrorContext(ctx, "fabric: dropping an undecodable runner event", + "subject", msg.Subject, "error", err) + continue + } + select { + case out <- ev: + case <-ctx.Done(): + return + case <-f.teardown: + return + } + } + } +} + +// decodeRunnerEvent unmarshals one fan-in message into a RunnerEvent. +func decodeRunnerEvent(msg *nats.Msg) (RunnerEvent, error) { + var req compassv1internal.PublishEventsRequest + if err := proto.Unmarshal(msg.Data, &req); err != nil { + return RunnerEvent{}, fmt.Errorf("fabric: unmarshaling a runner event from %d bytes: %w", len(msg.Data), err) + } + return RunnerEvent{RunnerID: msg.Header.Get(RunnerIDHeader), Event: &req}, nil +} diff --git a/go/internal/fabric/runner_fabric_test.go b/go/internal/fabric/runner_fabric_test.go new file mode 100644 index 000000000..e4623e5d3 --- /dev/null +++ b/go/internal/fabric/runner_fabric_test.go @@ -0,0 +1,383 @@ +package fabric + +import ( + "context" + "testing" + "time" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" + "github.com/nats-io/nats.go" + "google.golang.org/protobuf/proto" +) + +// recvEvent takes the next RunnerEvent from ch, failing at the gate. +func recvEvent(t *testing.T, ch <-chan RunnerEvent) RunnerEvent { + t.Helper() + select { + case ev, ok := <-ch: + if !ok { + t.Fatal("the runner-events channel closed before delivering an event") + } + return ev + case <-time.After(gate): + t.Fatalf("no runner event within %s", gate) + return RunnerEvent{} + } +} + +// TestSendCommandPublishesADecodableCommand defends the command plane's wire +// contract: what a Server sends must proto-unmarshal on the Runner side, on the +// Runner's OWN subject. A drift in either the encoding or the subject means the +// Runner never sees its commands — and, being best-effort core NATS, sees no +// error either. +func TestSendCommandPublishesADecodableCommand(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url}) + + // A raw core-NATS subscriber stands in for the Runner, so the test asserts + // the wire and not the fabric's own decoder. + runner, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(runner.Close) + + subject, err := RunnerCommandSubject("runner-7") + if err != nil { + t.Fatalf("RunnerCommandSubject: %v", err) + } + sub, err := runner.SubscribeSync(subject) + if err != nil { + t.Fatalf("SubscribeSync(%q): %v", subject, err) + } + if err := runner.FlushWithContext(ctx); err != nil { + t.Fatalf("flushing the runner subscription: %v", err) + } + + want := &compassv1internal.SessionsResponse{ + RequestId: "req-1", + Command: &compassv1internal.SessionsResponse_Stop{ + Stop: &compassv1.StopAgentSessionRequest{}, + }, + } + if err := f.SendCommand(ctx, "runner-7", want); err != nil { + t.Fatalf("SendCommand: %v", err) + } + + msg, err := sub.NextMsgWithContext(ctx) + if err != nil { + t.Fatalf("waiting for the command on %q: %v", subject, err) + } + var got compassv1internal.SessionsResponse + if err := proto.Unmarshal(msg.Data, &got); err != nil { + t.Fatalf("the published command must proto-unmarshal: %v", err) + } + if got.GetRequestId() != want.GetRequestId() { + t.Errorf("request id = %q, want %q", got.GetRequestId(), want.GetRequestId()) + } + if got.GetStop() == nil { + t.Error("the command oneof must survive the wire; GetStop() is nil") + } +} + +// TestSendCommandIsPerRunner defends the addressing itself: a command for one +// Runner must not land on another's subject. With several Runners on one NATS +// this is the only thing keeping commands from being executed by the wrong +// Runner. +func TestSendCommandIsPerRunner(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url}) + + other, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(other.Close) + + mineSubject, err := RunnerCommandSubject("runner-mine") + if err != nil { + t.Fatalf("RunnerCommandSubject: %v", err) + } + theirsSubject, err := RunnerCommandSubject("runner-theirs") + if err != nil { + t.Fatalf("RunnerCommandSubject: %v", err) + } + mine, err := other.SubscribeSync(mineSubject) + if err != nil { + t.Fatalf("SubscribeSync(%q): %v", mineSubject, err) + } + theirs, err := other.SubscribeSync(theirsSubject) + if err != nil { + t.Fatalf("SubscribeSync(%q): %v", theirsSubject, err) + } + if err := other.FlushWithContext(ctx); err != nil { + t.Fatalf("flushing subscriptions: %v", err) + } + + if err := f.SendCommand(ctx, "runner-mine", &compassv1internal.SessionsResponse{RequestId: "req-mine"}); err != nil { + t.Fatalf("SendCommand: %v", err) + } + if _, err := mine.NextMsgWithContext(ctx); err != nil { + t.Fatalf("the addressed runner did not receive its command: %v", err) + } + // Core NATS delivers in order on one connection, and the flush above + // established both interests before the publish, so if the command had + // fanned out it would already be queued here. + if n, _, err := theirs.Pending(); err != nil { + t.Fatalf("reading the other runner's pending count: %v", err) + } else if n != 0 { + t.Fatalf("the other runner has %d pending message(s); commands must not fan out", n) + } +} + +// TestSendCommandRejectsInvalidInput defends the fail-closed command path. A nil +// command would publish an empty message the Runner cannot classify, and an +// invalid runner id would build a corrupted subject — both silent on a +// best-effort plane, so both must fail at the caller. +func TestSendCommandRejectsInvalidInput(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + f := newFabric(t, Config{}) + + if err := f.SendCommand(ctx, "runner-1", nil); err == nil { + t.Error("SendCommand with a nil command: want an error") + } + for _, id := range []string{"", "runner.1", "runner*", "runner>", "runner 1"} { + if err := f.SendCommand(ctx, id, &compassv1internal.SessionsResponse{RequestId: "r"}); err == nil { + t.Errorf("SendCommand to runner id %q: want an error", id) + } + } +} + +// TestEventsYieldsDecodedRunnerEvents defends the fan-in wire contract in the +// other direction: a raw PublishEventsRequest published by a Runner must arrive +// as a RunnerEvent with every field intact. RunnerSeq is load-bearing (a gap is +// how the Server detects in-transit loss), and IdempotencyKey is what makes the +// durable commit at-most-once — losing either silently breaks a guarantee +// upstream. +func TestEventsYieldsDecodedRunnerEvents(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url}) + + events, err := f.Events(ctx) + if err != nil { + t.Fatalf("Events: %v", err) + } + + runner, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(runner.Close) + + want := &compassv1internal.PublishEventsRequest{ + RunnerSeq: 42, + SessionId: "sess-1", + IdempotencyKey: "idem-1", + } + body, err := proto.Marshal(want) + if err != nil { + t.Fatalf("proto.Marshal: %v", err) + } + msg := nats.NewMsg(RunnerEventsSubject()) + msg.Data = body + msg.Header.Set(RunnerIDHeader, "runner-7") + if err := runner.PublishMsg(msg); err != nil { + t.Fatalf("publishing a runner event: %v", err) + } + + got := recvEvent(t, events) + if got.RunnerID != "runner-7" { + t.Errorf("runner id = %q, want %q (the fan-in subject is shared, so the header is the only attribution)", got.RunnerID, "runner-7") + } + if got.Event == nil { + t.Fatal("RunnerEvent.Event must never be nil for an event read off the channel") + } + if got.Event.GetRunnerSeq() != want.GetRunnerSeq() { + t.Errorf("runner seq = %d, want %d (gap detection depends on it)", got.Event.GetRunnerSeq(), want.GetRunnerSeq()) + } + if got.Event.GetSessionId() != want.GetSessionId() { + t.Errorf("session id = %q, want %q", got.Event.GetSessionId(), want.GetSessionId()) + } + if got.Event.GetIdempotencyKey() != want.GetIdempotencyKey() { + t.Errorf("idempotency key = %q, want %q (at-most-once commit depends on it)", got.Event.GetIdempotencyKey(), want.GetIdempotencyKey()) + } +} + +// TestEventsClosesTheChannelOnContextDone defends the lifetime contract: a +// receiver ranging over the channel must see a clean close when the caller's +// context ends, not a goroutine that lives on holding a subscription. A +// never-closed channel is how a shutdown hangs. +func TestEventsClosesTheChannelOnContextDone(t *testing.T) { + t.Parallel() + f := newFabric(t, Config{}) + + // Rooted at context.Background() because this is a test root. + ctx, cancel := context.WithCancel(context.Background()) + events, err := f.Events(ctx) + if err != nil { + t.Fatalf("Events: %v", err) + } + + cancel() + select { + case _, ok := <-events: + if ok { + t.Fatal("want a closed channel after cancel, got an event") + } + case <-time.After(gate): + t.Fatalf("the runner-events channel was not closed within %s of cancel", gate) + } +} + +// TestEventsChannelClosesOnClose is the other half of the lifetime contract, +// and the one the ctx test above cannot see: with an UNCANCELLED context — +// a Server whose root context outlives the fabric it closes, which is the +// ordinary shutdown shape — Close alone must close the channel. +// +// It could not, before: the pump selected only on ctx.Done() and the raw +// subscription channel, and nats.go closes neither a ChanSubscription's channel +// nor its buffer when the connection closes. So a consumer ranging over Events +// blocked forever at shutdown, with the pump goroutine leaked behind it — a hung +// process, not a slow one. +func TestEventsChannelClosesOnClose(t *testing.T) { + t.Parallel() + f, err := New(Config{URL: testServer(t)}) + if err != nil { + t.Fatalf("New: %v", err) + } + + // Rooted at context.Background() because this is a test root, and its never + // being cancelled is the invariant under test: Close is the only teardown. + events, err := f.Events(context.Background()) + if err != nil { + t.Fatalf("Events: %v", err) + } + + if err := f.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + select { + case _, ok := <-events: + if ok { + t.Fatal("want a closed channel after Close, got an event") + } + case <-time.After(gate): + t.Fatalf("the runner-events channel was not closed within %s of Close; a ranging consumer would hang forever", gate) + } +} + +// TestEventsSkipsUndecodableMessages defends the best-effort plane's resilience: +// one malformed publish must not tear down the fan-in for every other Runner. +// The next valid event arriving is the gate. +func TestEventsSkipsUndecodableMessages(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + f := newFabric(t, Config{URL: url, Log: quietLogger(t)}) + + events, err := f.Events(ctx) + if err != nil { + t.Fatalf("Events: %v", err) + } + + runner, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(runner.Close) + + // Bytes that are not a valid PublishEventsRequest: a wire type of 7 is + // invalid in protobuf, so this cannot be read as an empty message. + if err := runner.Publish(RunnerEventsSubject(), []byte{0xFF, 0xFF, 0xFF}); err != nil { + t.Fatalf("publishing a malformed event: %v", err) + } + body, err := proto.Marshal(&compassv1internal.PublishEventsRequest{RunnerSeq: 7, SessionId: "sess-ok"}) + if err != nil { + t.Fatalf("proto.Marshal: %v", err) + } + if err := runner.Publish(RunnerEventsSubject(), body); err != nil { + t.Fatalf("publishing a valid event: %v", err) + } + + got := recvEvent(t, events) + if got.Event.GetSessionId() != "sess-ok" { + t.Fatalf("session id = %q, want %q — the malformed message was not skipped cleanly", got.Event.GetSessionId(), "sess-ok") + } +} + +// TestEventsQueueGroupDeliversOnce defends the queue-group semantics the record +// requires for fan-in: with two Servers subscribed, exactly ONE handles each +// Runner event. Plain subscriptions would have both Servers process every event +// — a double write-through for every agent frame. +func TestEventsQueueGroupDeliversOnce(t *testing.T) { + t.Parallel() + ctx := testCtx(t) + url := testServer(t) + a := newFabric(t, Config{URL: url}) + b := newFabric(t, Config{URL: url}) + + eventsA, err := a.Events(ctx) + if err != nil { + t.Fatalf("Events on a: %v", err) + } + eventsB, err := b.Events(ctx) + if err != nil { + t.Fatalf("Events on b: %v", err) + } + + runner, err := nats.Connect(url) + if err != nil { + t.Fatalf("nats.Connect: %v", err) + } + t.Cleanup(runner.Close) + + const n = 20 + for i := range n { + body, err := proto.Marshal(&compassv1internal.PublishEventsRequest{RunnerSeq: uint64(i), SessionId: "sess-1"}) + if err != nil { + t.Fatalf("proto.Marshal: %v", err) + } + if err := runner.Publish(RunnerEventsSubject(), body); err != nil { + t.Fatalf("publishing event %d: %v", i, err) + } + } + + // Gate on the exact expected count: n events published, n claims total + // across both instances. A plain (non-queue) subscription would yield 2n + // and this loop would see the extras. + seen := make(map[uint64]int, n) + for range n { + var ev RunnerEvent + select { + case ev = <-eventsA: + case ev = <-eventsB: + case <-time.After(gate): + t.Fatalf("only %d of %d events claimed within %s", len(seen), n, gate) + } + seen[ev.Event.GetRunnerSeq()]++ + } + if len(seen) != n { + t.Fatalf("claimed %d distinct sequences, want %d", len(seen), n) + } + for seq, count := range seen { + if count != 1 { + t.Errorf("sequence %d was claimed %d times, want exactly 1", seq, count) + } + } + // No duplicate is still queued behind the n claims. + select { + case ev := <-eventsA: + t.Fatalf("instance a claimed a duplicate of sequence %d", ev.Event.GetRunnerSeq()) + case ev := <-eventsB: + t.Fatalf("instance b claimed a duplicate of sequence %d", ev.Event.GetRunnerSeq()) + default: + } +} diff --git a/go/internal/fabric/stream.go b/go/internal/fabric/stream.go new file mode 100644 index 000000000..bf3439b9e --- /dev/null +++ b/go/internal/fabric/stream.go @@ -0,0 +1,146 @@ +package fabric + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +// JetStream topology defaults for the comms event plane. Every one is +// overridable through Config; these are the values the deployment runs unless a +// stack config says otherwise, and SUBJECTS.md is their written spec. +const ( + // DefaultStreamName is the single stream capturing every tenant's comms + // events. One stream rather than one per tenant: subject filtering on the + // consumer already isolates tenants, and a stream per tenant would make + // tenant creation a JetStream admin operation. + DefaultStreamName = "COMPASS_COMMS" + + // DefaultMaxDeliver bounds delivery attempts per message before the fabric + // parks it on DLQSubject. Finite by requirement (§Q3: "max_deliver with a + // dead-letter subject so a poison message parks instead of redelivering + // forever"); 5 is enough to ride out a transient subscriber fault and small + // enough that a genuine poison message parks in seconds rather than hours. + DefaultMaxDeliver = 5 + + // DefaultAckWait is how long the server waits for an explicit ack before + // redelivering. It only governs a subscriber that hangs or dies mid-callback + // — a callback that fails is Nak'd, which redelivers immediately. + DefaultAckWait = 30 * time.Second + + // DefaultDuplicateWindow is the publish-side dedup window. Two Servers + // publishing the same logical change, or one retrying a publish whose ack + // was lost, collapse to one stored message inside it (EventRef.msgID is the + // key). Sized to comfortably exceed any plausible publish retry. + DefaultDuplicateWindow = 2 * time.Minute + + // DefaultMaxAge bounds the stream's replay window. JetStream is a transport + // whose state is disposable (§Global Constraints), so retaining events past + // the point where the Postgres delivery cursor is the only sane recovery + // path buys nothing but disk. A subscriber further behind than this + // recovers by cursor sweep, not by replay. + DefaultMaxAge = 24 * time.Hour + + // DefaultReplicas is the stream/consumer replica count. A single-node NATS + // is R1 by construction; a clustered deployment sets 3 (§Q3: "file storage + // with R3 replication when NATS runs clustered"). Postgres is the recovery + // truth either way, so this is a durability optimization, not a correctness + // requirement. + DefaultReplicas = 1 +) + +// streamConfig is the COMPASS_COMMS stream configuration. +// +// # Where sync_interval lives +// +// The record specifies `sync_interval: 100ms` for this stream — a bounded fsync +// window, because JetStream's default sync behavior can lose acknowledged writes +// under power failure. It is deliberately NOT set here: sync_interval is a +// nats-server FILE STORE option, not a per-stream one, so jetstream.StreamConfig +// (nats.go v1.53.1) exposes no field for it. It is set on the NATS process: +// +// - server config: jetstream { sync_interval: "100ms" } +// - Go embedded/test: server.Options.SyncInterval = 100 * time.Millisecond +// +// The stack's NATS service config is where the deployment sets it (the sibling +// nats-container task); testServer in this package sets the server.Options field +// so the tests exercise the record's value rather than the server default. +func (c Config) streamConfig() jetstream.StreamConfig { + return jetstream.StreamConfig{ + Name: c.streamName(), + Description: "Compass comms events (EventRef references; Postgres is the truth)", + Subjects: []string{commsStreamSubjects}, + // Limits retention (the default): a message ages out on MaxAge rather + // than being removed once acked, so a second consumer group and a + // bounded replay both stay possible. + Retention: jetstream.LimitsPolicy, + Storage: jetstream.FileStorage, + Replicas: c.replicas(), + Discard: jetstream.DiscardOld, + MaxAge: c.maxAge(), + Duplicates: c.duplicateWindow(), + } +} + +// consumerConfig is the durable pull-consumer configuration for one subscribed +// subject. Durable and named deterministically from the subject so every Server +// instance subscribing to that subject shares one consumer — which is what gives +// the delivery plane its queue-group semantics (§Q3: each event is claimed by +// exactly one worker) and what lets a restarted Server resume where it stopped. +func (c Config) consumerConfig(subject string) jetstream.ConsumerConfig { + return jetstream.ConsumerConfig{ + Durable: durableName(subject), + Description: "comms EventRef fan-out for " + subject, + FilterSubject: subject, + AckPolicy: jetstream.AckExplicitPolicy, + AckWait: c.ackWait(), + MaxDeliver: c.maxDeliver(), + Replicas: c.replicas(), + } +} + +// durableName derives a JetStream durable consumer name for a subject. +// Consumer names cannot contain "." (nor whitespace, "*", ">", or a path +// separator), and a subject's tokens can legally contain "_" (ValidSubjectToken +// permits it — every snake_case EventKind uses it), so a "."→"_" substitution +// is NOT injective: compass.a.comms.b_comms_c and compass.a_comms_b.comms.c +// would collapse to one name and the second Subscribe would silently re-point +// the first's shared consumer FilterSubject (a cross-tenant mis-delivery). +// Hashing the subject is injective by construction; the "comms-" prefix keeps +// the name greppable, and the consumer's Description/FilterSubject still carry +// the readable subject for operators. Untruncated: "comms-" + 64 hex is 70 +// chars, far inside JetStream's 255-char limit, and truncating would reintroduce +// the collision surface this exists to remove. +func durableName(subject string) string { + sum := sha256.Sum256([]byte(subject)) + return "comms-" + hex.EncodeToString(sum[:]) +} + +// ensureStream creates or updates the comms stream, idempotently, and caches it. +// Called lazily from the first Publish/Subscribe rather than from New because the +// frozen New(cfg Config) signature carries no context — deriving the topology +// call from the first caller's ctx keeps the cancellation chain intact instead of +// rooting a fresh one. +// +// CreateOrUpdateStream (not CreateStream) so a container restart, a second +// Server, and a config change all converge on the same stream rather than +// racing or failing. +func (f *Fabric) ensureStream(ctx context.Context) (jetstream.Stream, error) { + f.streamMu.Lock() + defer f.streamMu.Unlock() + if f.stream != nil { + return f.stream, nil + } + s, err := f.js.CreateOrUpdateStream(ctx, f.cfg.streamConfig()) + if err != nil { + // Not cached: a failed ensure must be retryable on the next call, never + // poison the fabric for its lifetime. + return nil, fmt.Errorf("fabric: ensuring stream %s: %w", f.cfg.streamName(), err) + } + f.stream = s + return s, nil +} diff --git a/go/internal/fabric/subjects.go b/go/internal/fabric/subjects.go new file mode 100644 index 000000000..07224eab2 --- /dev/null +++ b/go/internal/fabric/subjects.go @@ -0,0 +1,136 @@ +package fabric + +import ( + "fmt" + "strings" +) + +// The frozen subject grammar (§T3). The fabric owns the grammar: callers build a +// subject through the helpers below and hand the string to Publish / Subscribe / +// SendCommand, so no consumer ever concatenates a subject itself. +const ( + // subjectPrefix roots every Compass-owned subject. + subjectPrefix = "compass" + + // 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.*" + + // RunnerEventsQueue is the queue group every Server's RunnerFabric.Events + // subscription joins, so one Runner event is handled by exactly one Server + // instance (§Q3, "delivery queue groups": the three-hop model's hop 2). + RunnerEventsQueue = "compass-runner-events" + + // DLQSubject is where a comms event that exhausted its delivery attempts is + // parked. JetStream has no native dead-letter queue, so the fabric + // implements the app-level pattern: publish to this subject, then Term() the + // message so the server stops redelivering it. Parking is a diagnostic, not + // a recovery path — recovery is always the Postgres row. + DLQSubject = subjectPrefix + ".dlq.comms" +) + +// CommsSubject builds the comms-event subject for a tenant and event kind: +// compass..comms.. It returns an error if tenant or kind is not a +// valid single subject token — a tenant id containing a "." is an upstream bug, +// and silently corrupting the subject would cross-wire tenants (see +// ValidSubjectToken). +func CommsSubject(tenant string, kind EventKind) (string, error) { + if err := ValidSubjectToken("tenant", tenant); err != nil { + return "", err + } + if err := ValidSubjectToken("event kind", string(kind)); err != nil { + return "", err + } + return subjectPrefix + "." + tenant + ".comms." + string(kind), nil +} + +// validCommsSubject checks a whole comms subject against the frozen grammar: +// exactly compass..comms., with both variable tokens valid. +// +// Subscribe takes a subject string, not a (tenant, kind) pair, so this is the +// only place the grammar can be enforced on that path — and a subject that is +// merely rooted at "compass" would otherwise reach CreateOrUpdateConsumer with +// a FilterSubject the COMPASS_COMMS stream does not capture, producing a +// consumer that silently never delivers. Publish gets the same guarantee for +// free by deriving the subject from the ref. +func validCommsSubject(subject string) error { + tokens := strings.Split(subject, ".") + if len(tokens) != commsSubjectTokens { + return fmt.Errorf("fabric: subject %q has %d tokens, want %d (compass..comms.)", + subject, len(tokens), commsSubjectTokens) + } + if tokens[0] != subjectPrefix { + return fmt.Errorf("fabric: subject %q is not rooted at %q", subject, subjectPrefix) + } + if tokens[2] != commsToken { + return fmt.Errorf("fabric: subject %q has %q where the grammar requires %q", subject, tokens[2], commsToken) + } + if err := ValidSubjectToken("tenant", tokens[1]); err != nil { + return err + } + return ValidSubjectToken("event kind", tokens[3]) +} + +// The shape validCommsSubject enforces: compass..comms., the same +// four tokens CommsSubject builds and commsStreamSubjects captures. +const ( + commsSubjectTokens = 4 + commsToken = "comms" +) + +// RunnerCommandSubject builds a Runner's command subject: +// compass.runner..cmd. Every Server publishes async commands for that +// Runner here; the Runner core-NATS-subscribes to it from enrollment (§Q2 — +// Runners are subject-addressable, not connection-owned). +func RunnerCommandSubject(runnerID string) (string, error) { + if err := ValidSubjectToken("runner id", runnerID); err != nil { + return "", err + } + return subjectPrefix + ".runner." + runnerID + ".cmd", nil +} + +// RunnerEventsSubject is the Runner→Server event fan-in subject: +// compass.runner.events. It takes no token and cannot fail. Subscribers join +// RunnerEventsQueue so exactly one Server instance claims each event. +func RunnerEventsSubject() string { + return subjectPrefix + ".runner.events" +} + +// ClientSubject builds the per-connection delivery subject for a live client: +// client.. Deliberately outside the "compass." root — the frozen +// grammar names it "client." (§T3), and it is not a +// COMPASS_COMMS-stream subject. Unused in this task; the delivery edge lands on +// it later. +func ClientSubject(sessionID string) (string, error) { + if err := ValidSubjectToken("session id", sessionID); err != nil { + return "", err + } + return "client." + sessionID, nil +} + +// ValidSubjectToken reports whether s is usable as a single NATS subject token, +// naming what was wrong. NATS reserves "." (token separator), "*" and ">" +// (wildcards), and rejects whitespace; an empty token would collapse two +// separators into an off-by-one subject. +// +// Tenant ids, runner ids and session ids are opaque to the fabric, so the +// grammar cannot encode-and-escape them without becoming ambiguous on the way +// back out. The choice is therefore to REJECT rather than sanitize: an id +// carrying a reserved character is a bug at the identifier's source, and turning +// it into a different-but-valid subject would route a tenant's events to the +// wrong subject rather than surfacing the bug. +func ValidSubjectToken(what, s string) error { + if s == "" { + return fmt.Errorf("fabric: %s is empty; a subject token cannot be empty", what) + } + if i := strings.IndexAny(s, ".*>"); i >= 0 { + return fmt.Errorf("fabric: %s %q contains reserved NATS subject character %q at %d", what, s, s[i], i) + } + if i := strings.IndexFunc(s, func(r rune) bool { + return r == ' ' || r == '\t' || r == '\r' || r == '\n' + }); i >= 0 { + return fmt.Errorf("fabric: %s %q contains whitespace at %d; a subject token cannot contain whitespace", what, s, i) + } + return nil +} diff --git a/go/internal/fabric/subjects_test.go b/go/internal/fabric/subjects_test.go new file mode 100644 index 000000000..fb41ac3be --- /dev/null +++ b/go/internal/fabric/subjects_test.go @@ -0,0 +1,219 @@ +package fabric + +import ( + "strings" + "testing" +) + +// TestSubjectBuilders defends the frozen grammar itself: every subject a +// consumer will ever publish or subscribe to comes from these four builders, so +// a drift in any one of them silently re-wires a plane. The exact strings are +// asserted, not a pattern — the grammar is the contract other tasks build +// against (SUBJECTS.md). +func TestSubjectBuilders(t *testing.T) { + t.Parallel() + + t.Run("comms", func(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + tenant string + kind EventKind + want string + }{ + {"message posted", "t1", KindMessagePosted, "compass.t1.comms.message_posted"}, + {"account changed", "acme", KindAccountChanged, "compass.acme.comms.account_changed"}, + {"uuid tenant", "0d8f1a2b-3c4d", KindTopicUpserted, "compass.0d8f1a2b-3c4d.comms.topic_upserted"}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + got, err := CommsSubject(tc.tenant, tc.kind) + if err != nil { + t.Fatalf("CommsSubject(%q, %q): %v", tc.tenant, tc.kind, err) + } + if got != tc.want { + t.Fatalf("CommsSubject(%q, %q) = %q, want %q", tc.tenant, tc.kind, got, tc.want) + } + }) + } + }) + + t.Run("runner command", func(t *testing.T) { + t.Parallel() + got, err := RunnerCommandSubject("runner-7") + if err != nil { + t.Fatalf("RunnerCommandSubject: %v", err) + } + if want := "compass.runner.runner-7.cmd"; got != want { + t.Fatalf("RunnerCommandSubject = %q, want %q", got, want) + } + }) + + t.Run("runner events", func(t *testing.T) { + t.Parallel() + if got, want := RunnerEventsSubject(), "compass.runner.events"; got != want { + t.Fatalf("RunnerEventsSubject = %q, want %q", got, want) + } + if RunnerEventsQueue == "" { + t.Fatal("RunnerEventsQueue must be set, or every Server handles every Runner event") + } + }) + + t.Run("client", func(t *testing.T) { + t.Parallel() + got, err := ClientSubject("sess-42") + if err != nil { + t.Fatalf("ClientSubject: %v", err) + } + if want := "client.sess-42"; got != want { + t.Fatalf("ClientSubject = %q, want %q", got, want) + } + }) +} + +// TestClientSubjectIsOutsideTheCommsStream defends a real cross-plane hazard: +// the COMPASS_COMMS stream captures compass.*.comms.*, and a client subject that +// happened to land inside that wildcard would push every per-connection +// delivery into a durable stream — a second store of state Postgres owns. +func TestClientSubjectIsOutsideTheCommsStream(t *testing.T) { + t.Parallel() + got, err := ClientSubject("sess-42") + if err != nil { + t.Fatalf("ClientSubject: %v", err) + } + if strings.HasPrefix(got, subjectPrefix+".") { + t.Fatalf("ClientSubject = %q: must not sit under the %q root captured by the comms stream", got, subjectPrefix) + } +} + +// TestSubjectBuildersRejectInvalidTokens defends the reject-never-sanitize +// choice. Each of these tokens would, if silently accepted, produce a +// well-formed but WRONG subject: a "." adds a token (routing a tenant's events +// somewhere nobody is subscribed), and a "*" or ">" turns a publish target into +// a wildcard. The builder must refuse. +func TestSubjectBuildersRejectInvalidTokens(t *testing.T) { + t.Parallel() + bad := []struct { + name string + token string + }{ + {"empty", ""}, + {"dot", "ten.ant"}, + {"star", "ten*ant"}, + {"gt", "ten>ant"}, + {"space", "ten ant"}, + {"tab", "ten\tant"}, + {"newline", "ten\nant"}, + } + for _, tc := range bad { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if s, err := CommsSubject(tc.token, KindMessagePosted); err == nil { + t.Errorf("CommsSubject(%q, …) = %q, want an error", tc.token, s) + } + if s, err := RunnerCommandSubject(tc.token); err == nil { + t.Errorf("RunnerCommandSubject(%q) = %q, want an error", tc.token, s) + } + if s, err := ClientSubject(tc.token); err == nil { + t.Errorf("ClientSubject(%q) = %q, want an error", tc.token, s) + } + }) + } +} + +// TestCommsSubjectRejectsInvalidKind defends the kind token too — it is as much +// a subject token as the tenant is, and an unvalidated kind is how a +// caller-supplied string escapes into the grammar. +func TestCommsSubjectRejectsInvalidKind(t *testing.T) { + t.Parallel() + for _, kind := range []EventKind{"", "message.posted", "message>posted", "message posted"} { + if s, err := CommsSubject("t1", kind); err == nil { + t.Errorf("CommsSubject(t1, %q) = %q, want an error", kind, s) + } + } +} + +// 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 +// publish for that kind at runtime, not at compile time. +func TestEventKindsAreValidSubjectTokens(t *testing.T) { + t.Parallel() + kinds := []EventKind{ + KindAccountChanged, KindChannelGroupChanged, KindChannelChanged, + KindAgentWorkspaceChanged, KindMessagePosted, KindMessageUpdated, + KindTopicUpserted, + } + if len(kinds) != 7 { + t.Fatalf("expected the 7 frozen comms kinds, listed %d", len(kinds)) + } + seen := make(map[EventKind]bool, len(kinds)) + for _, k := range kinds { + if err := ValidSubjectToken("event kind", string(k)); err != nil { + t.Errorf("kind %q is not a usable subject token: %v", k, err) + } + if seen[k] { + t.Errorf("kind %q is duplicated; two event classes would share a subject", k) + } + seen[k] = true + } +} + +// TestDLQSubjectIsOutsideTheCommsStream defends against the pathological loop: +// if the dead-letter subject were captured by the comms stream, parking a poison +// message would store it back in the stream it came from. +func TestDLQSubjectIsOutsideTheCommsStream(t *testing.T) { + t.Parallel() + // The stream wildcard is compass.*.comms.* — four tokens with "comms" + // third. The DLQ subject is three tokens, so it cannot match; assert the + // shape that guarantees it. + if got, want := strings.Count(DLQSubject, "."), 2; got != want { + t.Fatalf("DLQSubject = %q has %d separators, want %d (a 4-token dlq subject could be captured by %q)", + DLQSubject, got, want, commsStreamSubjects) + } +} + +// TestValidCommsSubjectRejectsMalformed defends the whole-subject guard that +// replaced the head-only one on the Subscribe path. Every case below is rooted +// at "compass", so a first-token check passed all of them — and a subject the +// COMPASS_COMMS stream's compass.*.comms.* wildcard does not capture produces a +// consumer that is created successfully and then silently never delivers, which +// is the worst failure shape available: no error anywhere. +func TestValidCommsSubjectRejectsMalformed(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + subject string + }{ + {"wildcard tokens", "compass.*.comms.*"}, + {"empty tenant", "compass..comms.x"}, + {"trailing token", "compass.t1.comms.message_posted.extra"}, + {"too few tokens", "compass.t1.comms"}, + {"wrong root", "client.t1.comms.message_posted"}, + {"wrong plane token", "compass.t1.runner.message_posted"}, + {"empty", ""}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if err := validCommsSubject(tc.subject); err == nil { + t.Errorf("validCommsSubject(%q) = nil, want an error", tc.subject) + } + }) + } + + // The positive gate: the grammar the builder itself produces must pass, or + // the guard would reject every legitimate Subscribe. + t.Run("valid", func(t *testing.T) { + t.Parallel() + if err := validCommsSubject("compass.t1.comms.message_posted"); err != nil { + t.Errorf("validCommsSubject on a well-formed subject: %v", err) + } + built, err := CommsSubject("t1", KindMessagePosted) + if err != nil { + t.Fatalf("CommsSubject: %v", err) + } + if err := validCommsSubject(built); err != nil { + t.Errorf("validCommsSubject rejected the subject CommsSubject built (%q): %v", built, err) + } + }) +}