package buscompose import ( "strings" ) // Retention is how a stream decides what to keep, which is the whole of what distinguishes the // mesh's four relationships on the wire (design 29 §4). type Retention string const ( // RetentionWorkQueue: a message is removed once a consumer acknowledges it. Exactly one // worker does the work, and a worker that dies has its message redelivered. RetentionWorkQueue Retention = "workqueue" // RetentionLastPerSubject: only the newest message on each subject survives. This is the // state shape — a node that was away gets exactly the current declaration and nothing older. RetentionLastPerSubject Retention = "last_per_subject" // RetentionLimits: kept until it ages or the stream fills. Events, where a subscriber that // was down catches up and nobody is obliged to act. RetentionLimits Retention = "limits" ) // A Stream is one of the mesh's own, as the controller asserts it. type Stream struct { Name string Subjects []string Retention Retention // MaxAge in seconds, zero for unbounded. Per stream — JetStream has no per-subject age, // which is why differing retention between modules would mean a stream each. MaxAge int // MaxMsgsPerSubject caps each subject independently, so one noisy emitter cannot push // another's events out of a shared stream. Verified: with a cap of 3, ten messages on one // subject and one on another leave four in the stream, not three. MaxMsgsPerSubject int // Why is carried into the assertion so an operator reading the server's own state finds the // reason there, rather than only in a repository they may not have. Why string // Direct lets a client read a subject's last message without a consumer, which is how a // runtime reads its own membership with no JetStream API beyond one request (ADR 0160). Direct bool // MaxBytes bounds the stream's size, zero for unbounded. With DiscardNew a full stream refuses // what comes next rather than dropping what it holds: the publisher is told, and says so. MaxBytes int64 DiscardNew bool // DuplicatesSeconds is the window in which a message id published twice is kept once; zero for the // server's default (two minutes). DuplicatesSeconds int // AllowMsgTTL lets a message carry a time to live of its own (`Nats-TTL`), for what must not outlive the // stream it came from (novox/hq ADR 0376 §3, §5). AllowMsgTTL bool // Restricted marks a module's stream restricted by a grant (novox/hq ADR 0363 §4), in the stream's own // metadata, so whoever keeps one of its messages reads it from the bus (ADR 0376 §3). Restricted bool // WorkQueue marks a module stream declared retention: work-queue (novox/hq ADR 0436 §6), in the // stream's own metadata, so a dead letter kept from it is read as drop-only from the bus, by the // declaration and not by the bus's work-queue retention alone — which SEAT_* and other mesh-internal // streams share without being one of ADR 0436's, and whose dead letters are still delivered again // (ADR 0264 §4). WorkQueue bool } // RestrictedMetadata is the key a restricted stream's metadata says it under. const RestrictedMetadata = "mesh-restricted" // WorkQueueMetadata is the key a module stream declared retention: work-queue (novox/hq ADR 0436 §6) says // it under: never set by a mesh-internal work-queue stream (a seat's, a check's), only by Stream.WorkQueue. const WorkQueueMetadata = "mesh-work-queue" // What a durable consumer gave up on is kept (novox/hq issue 330, design 25 §3). // // **The server says it and keeps it; the controller copies it.** A consumer that handed a message over // as often as it may stops offering it and publishes `$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES` with // the stream and the message's sequence. DeadLetterNoticesStream captures those advisories as the // server publishes them, so one said while no controller listens is still there when one starts. The // controller's consumer on it fetches the given-up message by its sequence, while the source stream // still holds it, and keeps a copy in DeadLettersStream under DeadLetterSubject, with the consumer, // the subject, how often it was handed over and when it was given up. It stays there until a person // delivers it again or drops it, with why; a condition is open for as long as it does. const ( DeadLetterNoticesStream = "DEAD_LETTER_NOTICES" DeadLettersStream = "DEAD_LETTERS" // MaxDeliveriesAdvisories is the subject the server says a given-up message on. MaxDeliveriesAdvisories = "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>" deadLetterPrefix = "mesh.events.dead." againPrefix = "mesh.again." // DeadLettersBytes bounds the kept copies. Full, the stream refuses the next copy, which is said // as the consumer's condition; it never drops one it holds. DeadLettersBytes = 256 << 20 ) // DeadLetterSubject is where a message one consumer gave up on is kept: `mesh.events.dead..`. func DeadLetterSubject(stream, consumer string) string { return deadLetterPrefix + stream + "." + consumer } // DeadLetterOf is the stream and consumer a kept message's subject names; false for any other subject. func DeadLetterOf(subject string) (stream, consumer string, ok bool) { rest, found := strings.CutPrefix(subject, deadLetterPrefix) if !found { return "", "", false } stream, consumer, ok = strings.Cut(rest, ".") return stream, consumer, ok && stream != "" && consumer != "" && !strings.Contains(consumer, ".") } // MeshStreams is the foundation set, in the order a person reads it. // // **EVENTS filters on the `event` token**, which is the reason that token exists. A module's // namespace carries both its events and its tool calls; a filter of `mesh.mod.*.>` would persist // every tool invocation in the mesh, and a tool call must never be persisted (design 25 §3 keeps // tools on core NATS, where a lost call is a timeout the caller already handles). func MeshStreams() []Stream { return []Stream{ // CONTROL, the ended controller's work queue of what nodes reported, is no longer the mesh's (novox/hq ADR 0590 // clause 6): not declared here, which would make it again; the bus module's tools remove it. { Name: "NODES", Subjects: []string{"mesh.node.*.declare"}, Retention: RetentionLastPerSubject, Why: "one declaration per node, always the newest; a node that sees sequence n refuses " + "n-1 by construction (issue 107)", }, // The assignments stream (ADR 0160) is no longer the mesh's: a runtime reads its membership from the grants // part's state memberships (ADR 0571 clause 7), and the bus module's tools delete the stream at the lease write // (research 072 row 33), so it is not declared here, which would make it again. { Name: EventsStream, // A seat's own events ride here too: they are 1:many like any event, and the // `event` token keeps them clear of both the seat's work queue (`accept`) and its // tools (`tool`), which must not be persisted. // And an event given up on, delivered again to the one consumer that gave it up // (novox/hq issue 330): under `mesh.again..`, which only that consumer filters. Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>", againPrefix + ">"}, Retention: RetentionLimits, MaxAge: 7 * 24 * 60 * 60, MaxMsgsPerSubject: 10000, // A restricted dead letter delivered again keeps what is left of its stream's age (ADR 0376 §5). AllowMsgTTL: true, Why: "a subscriber that was down catches up; tool traffic under the same prefix is " + "excluded by the event token; per-subject caps keep a noisy emitter from " + "evicting a quiet one without splitting the stream", }, { Name: DeadLetterNoticesStream, Subjects: []string{MaxDeliveriesAdvisories}, Retention: RetentionWorkQueue, MaxAge: 7 * 24 * 60 * 60, Why: "the server's word that a consumer gave up on a message, kept until the controller has " + "copied the message into DEAD_LETTERS (novox/hq issue 330); a week, the longest the source " + "streams keep what they are about", }, { Name: DeadLettersStream, Subjects: []string{deadLetterPrefix + ">"}, Retention: RetentionLimits, MaxBytes: DeadLettersBytes, DiscardNew: true, DuplicatesSeconds: 24 * 60 * 60, AllowMsgTTL: true, Why: "every message a consumer gave up on, with its consumer, subject, deliveries and when, kept " + "until a person delivers it again or drops it with why (novox/hq issue 330); no age but a restricted " + "stream's message, which ages out with its stream (ADR 0376), and full it refuses the next copy " + "rather than drop one it holds", }, { Name: ProvAppliedStream, Subjects: []string{ProvAppliedSubjects}, Retention: RetentionLastPerSubject, Direct: true, MaxAge: 7 * 24 * 60 * 60, Why: "each machine's provisioner client's newest word on each module's files, applied, waiting or " + "failed with the reason (novox/hq ADR 0349 §4): what a walk's end and a gate wait for (ADR 0362, " + "ADR 0349 §15), read by subject; a week, as the events are", }, } } // ProvAppliedStream keeps what each machine's provisioner client last said of each module's files, newest per // subject (novox/hq ADR 0349 §4, ADR 0362, design 57 P6): `prov.applied..` lies outside `mesh.*`, so // no other stream captures it, and a report said while no controller listened is still there to be read. const ProvAppliedStream = "PROV_APPLIED" // ProvAppliedSubjects is every machine's client's reports: `prov.applied..`, a module's name dots and // all (ProvisionerAppliedSubject). const ProvAppliedSubjects = "prov.applied.>" // The ended controller's consumers' names (novox/hq ADR 0590 clause 6): kept only so its leftovers are named for their // removal (ControllerLeftoverConsumers) and the watchman can read whether `controller` on DEAD_LETTER_NOTICES still // stands. const ( ControllerName = "controller" MergesConsumerName = "controller-merges" ) // EventsStream holds every module's and every role's events, a build's log among them. const EventsStream = "EVENTS"