package broker import ( "fmt" "sort" "strings" ) // A module's own stream on the bus (novox/hq ADR 0363). // // A module declares a stream as it declares a log: named locally, created by the controller from the // catalogue on every raise and every send, never removed by the mesh. Its events are published on // `mesh.stream..event.`, a subject space the shared EVENTS stream (`mesh.mod.*.event.>`) // does not take — the bus refuses two streams claiming one subject, so a module's stream cannot share // EVENTS' — and a consumer hears it through a durable consumer of its own on that stream, named as on // EVENTS. Pure; the controller's raise asserts what is composed here. // moduleStreamPrefix starts every subject a module's own stream takes. const moduleStreamPrefix = "mesh.stream." // ModuleStreamSubject is where a module's event on its own stream lands: `mesh.stream..event.`. // The token before `.event.` is the emitter, as on EVENTS, so a runtime reads the same key from it. func ModuleStreamSubject(module, event string) string { return moduleStreamPrefix + module + ".event." + event } // OnAModuleStream says a subject is one a module's own stream takes. func OnAModuleStream(subject string) bool { return strings.HasPrefix(subject, moduleStreamPrefix) } // WirePattern is a local event pattern as the bus spells it: `**` is `>`. func WirePattern(local string) string { parts := strings.Split(local, ".") for i, p := range parts { if p == "**" { parts[i] = ">" } } return strings.Join(parts, ".") } // A ModuleStream is one module's declared stream as the bus holds it. type ModuleStream struct { Module string // Name is its name on the bus, `MAIL`. Name string // Events are the events it carries, as the bus spells them (WirePattern). Events []string MaxAgeDays int MaxMiB int // Restricted says a trusted config value of the module grants who hears it (ADR 0363 §4). Restricted bool } // Subjects are the subjects the stream takes. func (s ModuleStream) Subjects() []string { var out []string for _, e := range s.Events { out = append(out, ModuleStreamSubject(s.Module, e)) } sort.Strings(out) return out } // Stream is the stream the controller asserts: kept by limits, its age and size the module's, dropping // its oldest when full — a consumer that was away catches up within the age, and the module's own record // answers for anything older. func (s ModuleStream) Stream() Stream { why := fmt.Sprintf("%s's own stream (novox/hq ADR 0363): its events, kept %d days and at most %d MiB so a "+ "consumer that was away catches up", s.Module, s.MaxAgeDays, s.MaxMiB) if s.Restricted { why += "; heard only by the consumers its trusted config value grants" } return Stream{ Name: s.Name, Subjects: s.Subjects(), Retention: RetentionLimits, MaxAge: s.MaxAgeDays * 24 * 60 * 60, MaxBytes: int64(s.MaxMiB) * 1024 * 1024, Why: why, // Said on the bus, so a dead letter kept from it ages out with it (novox/hq ADR 0376 §3). Restricted: s.Restricted, } } // ModuleStreams is every declared module stream as the controller asserts it, in name order. func ModuleStreams(declared []ModuleStream) []Stream { sorted := append([]ModuleStream(nil), declared...) sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name }) out := make([]Stream, 0, len(sorted)) for _, s := range sorted { out = append(out, s.Stream()) } return out } // Heard is a module's stream another module hears, and the subjects it hears of it: its `consumes` // within the stream and, for a restricted stream, within what the provider's trusted config value grants it // (ADR 0363 §5). Composed only with at least one subject. type Heard struct { Stream string Emitter string Filters []string // Restricted says the stream is restricted by its provider's trusted config value (ADR 0363 §4): then it // is heard only by a module on an account of its own (ADR 0376 §1, ComposedHears). Restricted bool } // ComposedHears is what of a module's hearing the controller composes (novox/hq ADR 0376 §1): every stream it // hears, but a restricted one only for a module that runs in a runtime of its own on a bus account of its own // (`runs-as`, ADR 0259). **The tool runner never hears a restricted stream**: a module it carries reaches the bus // through the tool runner's account, which every agent session on that machine reaches the bus as too, so a // consumer, its grants or its bind word composed there would hand the stream to any session. A module that // declares no `runs-as` and keeps its own container credential gets none either: only `runs-as` is an account // the mesh keeps from agents. func ComposedHears(d Declared) []Heard { if d.RunsAs != "" { return d.Hears } var out []Heard for _, h := range d.Hears { if !h.Restricted { out = append(out, h) } } return out } // Withheld is a restricted stream a grant gives a module that the controller composes nothing for, because the // module has no account of its own (novox/hq ADR 0376 §2): said in `doctor` and in the controller's log, never // silent. type Withheld struct { Node, Module, Stream, Emitter string } // Said is a withheld hearing in the operator's words: the module, the stream and the remedy. func (w Withheld) Said() string { return fmt.Sprintf("%s on %s is granted part of %s's restricted stream %s, and hears none of it: it is carried "+ "by the machine's tool runner, whose bus account every agent session there shares, so nothing is composed "+ "for it. Declare runs-as in %s's manifest, so it runs as an account of its own (novox/hq ADR 0376)", w.Module, w.Node, w.Emitter, w.Stream, w.Module) } // WithheldOf is every granted hearing of a restricted stream composed for nothing, by machine and module. func WithheldOf(r Records) []Withheld { var out []Withheld for _, node := range sortedCopy(r.Nodes) { for _, d := range r.Assigned[node] { if d.RunsAs != "" { continue } for _, h := range d.Hears { if h.Restricted && len(h.Filters) > 0 { out = append(out, Withheld{Node: node, Module: d.Module, Stream: h.Stream, Emitter: h.Emitter}) } } } } return out } // heardConsumer is the durable consumer a module hears a module's stream through: named as its consumer // on EVENTS, on that stream, with the subjects it was granted as its filters. func heardConsumer(p Principal, h Heard) Consumer { filters := append([]string(nil), h.Filters...) sort.Strings(filters) return Consumer{ Name: consumerDurable(p), Stream: h.Stream, Filters: filters, AckWaitSeconds: 30, MaxDeliver: 5, Why: "what " + p.Module + " consumes of " + h.Emitter + "'s stream " + h.Stream + " and is granted " + "(novox/hq ADR 0363); after max-deliver it gives an event up, and the controller keeps it in " + "DEAD_LETTERS until a person delivers it again or drops it", } } // heardGrants are what a principal publishes to read and acknowledge the consumers it hears module streams // through: asking about each, pulling from it, acknowledging it — its own name on each stream, no other's. func heardGrants(p Principal, hears []Heard) []string { var out []string durable := consumerDurable(p) for _, h := range hears { if !safeSubject.MatchString(h.Stream) || len(h.Filters) == 0 { continue } out = append(out, "$JS.API.CONSUMER.INFO."+h.Stream+"."+durable, "$JS.API.CONSUMER.MSG.NEXT."+h.Stream+"."+durable, "$JS.ACK."+h.Stream+"."+durable+".>") } return out } // ownStreamGrants are the subjects a module publishes its own streams' events on, and no other's. func ownStreamGrants(module string, streams []ModuleStream) []string { var out []string for _, s := range streams { if s.Module != module { continue } out = append(out, s.Subjects()...) } return out } // StreamIssued is one of a module's own streams as its membership names it (ADR 0363 §2): the events it // carries, as the bus spells them, and the subject an event of them is published on, `{event}` the event. type StreamIssued struct { Stream string `json:"stream"` Events []string `json:"events"` Subject string `json:"subject"` } // HeardIssued is one module stream an assignment hears, as its membership names it (ADR 0363 §8): the // stream and the durable consumer on it the runtime binds. type HeardIssued struct { Stream string `json:"stream"` Consumer string `json:"consumer"` } // streamsIssuedFor is every stream of its own a module's membership names. func streamsIssuedFor(d Declared) []StreamIssued { var out []StreamIssued for _, s := range d.Streams { if s.Module != d.Module || !safeSubject.MatchString(d.Module) { continue } out = append(out, StreamIssued{Stream: s.Name, Events: append([]string(nil), s.Events...), Subject: ModuleStreamSubject(d.Module, "{event}")}) } return out } // HeardIssuedFor is every module stream an assignment hears, as its membership names it. func HeardIssuedFor(node string, d Declared) []HeardIssued { var out []HeardIssued // A restricted stream's bind word only for a module of its own account (novox/hq ADR 0376 §1). for _, h := range ComposedHears(d) { if len(h.Filters) == 0 { continue } out = append(out, HeardIssued{Stream: h.Stream, Consumer: node + "_" + d.Module}) } sort.Slice(out, func(i, j int) bool { return out[i].Stream < out[j].Stream }) return out } // Pruner is the part of a JetStream connection that removes a module's consumer on a module's stream it // may no longer hear (ADR 0363 §6). Optional: what only records what it is asked to assert has none. type Pruner interface { ConsumerNames(stream string) ([]string, error) RemoveConsumer(stream, name string) error } // PruneHeard removes, from each module stream, every durable consumer the composition no longer asks for, // and answers what it removed as `/`. A consumer that loses its grant loses what it reads // through, whatever its runtime still holds; a stream that does not exist yet has nothing to remove. func PruneHeard(p Pruner, streams []ModuleStream, wanted []Consumer) ([]string, error) { keep := map[string]bool{} for _, c := range wanted { keep[c.Stream+"/"+c.Name] = true } var removed []string for _, s := range streams { names, err := p.ConsumerNames(s.Name) if err != nil { return removed, fmt.Errorf("listing the consumers on %s: %w", s.Name, err) } for _, n := range names { if keep[s.Name+"/"+n] { continue } if err := p.RemoveConsumer(s.Name, n); err != nil { return removed, fmt.Errorf("removing %s's consumer on %s, which is no longer granted: %w", n, s.Name, err) } removed = append(removed, s.Name+"/"+n) } } return removed, nil } // EnsureHeard asserts a consumer on a module's stream (ADR 0363 §5, §6). **On a restricted stream, a consumer // the bus will not bring to match is removed**: a pull consumer delivers to whoever pulls it, so one left with // its wider filters after a narrowing was refused would keep handing over what the grant no longer names, // whatever the user list now says. The next assertion makes it again with the granted filters. The refusal is // returned either way, saying what was done, so the send says it and raises it. func EnsureHeard(r Raiser, c Consumer, restricted bool) error { err := r.EnsureConsumer(c) if err == nil || !restricted { return err } p, prunes := r.(Pruner) if !prunes { return fmt.Errorf("%w; it could not be removed here, and may still deliver more than it is granted", err) } if rerr := p.RemoveConsumer(c.Stream, c.Name); rerr != nil { return fmt.Errorf("%w; REMOVING IT FAILED TOO, so it may still deliver more than it is granted: %v", err, rerr) } return fmt.Errorf("%w; REMOVED, so it delivers nothing it is no longer granted: the next assertion makes it "+ "again with its granted filters (novox/hq ADR 0363 §6)", err) }