jschoubben
9b01c94a9e
An asker started before its grant was composed was refused once and stayed deaf until it started again, silently. DeadLetterHearing tries in the background and again on an event (Again: a reconnect, an ask waited for while deaf), and tells each change of Hearing so the module's health tool can say it. DeadLetterReads drops the flow control a pull consumer does not use and deletes one consumer by name.
309 lines
12 KiB
Go
309 lines
12 KiB
Go
package buscompose
|
|
|
|
import (
|
|
"fmt"
|
|
"sort"
|
|
)
|
|
|
|
// What a module may say and take on the seats it holds, uses and watches, beyond tools (novox/hq ADR
|
|
// 0259 §3, to-be 46 §10).
|
|
//
|
|
// Three rules are added to the ones a seat always had, each a subject whose last token names who may
|
|
// publish it, granted to that publisher alone — the way a node seat's event about a machine carries the
|
|
// machine (ADR 0219):
|
|
//
|
|
// - **A verb named by its caller** (`by-caller`). A user of the seat submits that accept, and hears that
|
|
// event, under its own module's name and no other: `accept.ask.<module>`, `event.decided.<module>`. The
|
|
// holder takes every caller's accept and says the event to any caller. So an ask's asker is a fact the
|
|
// server enforces, and a warrant reaches only the asker it is for.
|
|
// - **A kinded bench** (ADR 0234 §2). A holder claims one kind and takes its own kind's accepts, says its
|
|
// own kind's events and asks its own kind's proofs, and nothing of another kind; a user submits to any
|
|
// kind. Each kind has its own worker on the seat's queue, so a holder that is away keeps its work and
|
|
// holds up no other kind.
|
|
// - **A proof** (`proofs`): core request and reply on `mesh.seat.<seat>.proof.<verb>.<kind>`. No stream's
|
|
// subjects cover it, so what travels there — a code typed by the operator — is never persisted. A kinded
|
|
// holder asks with its own kind; the modules that watch the seat answer.
|
|
//
|
|
// And two reads: **a holder's records**, a bucket the seat names, read by each user under its own name only
|
|
// (`$KV.<bucket>.<module>.>`), so an asker reads the state of its own asks and no other asker's; and on a seat pulled by
|
|
// kind, **what its workers gave up on**, the copies DEAD_LETTERS keeps of the seat's stream (DeadLetterReads, issue 734).
|
|
|
|
// BuildAskMaxDeliver bounds the deliveries of a build seat's ask (issue 1907): mesh-sdk delivery.AskMaxDeliver, written
|
|
// here as the broker mirrors the catalogue rather than importing it.
|
|
const BuildAskMaxDeliver = 5
|
|
|
|
// KindedBenches are the seats that may be kinded (ADR 0234 §2): making another is a decision, recorded.
|
|
var KindedBenches = map[string]bool{"channel": true, "intake": true}
|
|
|
|
// WorkerName is the worker a seat's holders pull from: one for the seat, or one per kind on a kinded bench.
|
|
func WorkerName(seat, kind string) string {
|
|
if kind == "" {
|
|
return "SEAT_" + upperSnake(seat) + "_worker"
|
|
}
|
|
return "SEAT_" + upperSnake(seat) + "_" + upperSnake(kind) + "_worker"
|
|
}
|
|
|
|
func namesVerb(list []string, s string) bool {
|
|
for _, x := range list {
|
|
if x == s {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// SeatTrafficOf derives one module's seat traffic from the seats it holds, uses and watches. Only seats
|
|
// carrying one of the rules above are read: every other seat is composed as it always was.
|
|
func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic {
|
|
var t SeatTraffic
|
|
for _, s := range holds {
|
|
if !s.isNewTraffic() {
|
|
continue
|
|
}
|
|
kind := ""
|
|
if s.Kinded {
|
|
kind = s.Kind
|
|
if kind == "" || !safeSubject.MatchString(kind) {
|
|
// A kinded claim without a usable kind is refused at registration; here it is granted
|
|
// nothing, which is the same answer at the last place it could be asked.
|
|
continue
|
|
}
|
|
}
|
|
if len(s.Accepts) > 0 && s.PulledByKind {
|
|
// One worker per accepted kind, which every holder pulls in the order of Accepts (ADR 0525 §4, ADR 0536 §4):
|
|
// `SEAT_<SEAT>_<KIND>_worker`, filtered to `accept.<kind>` (mesh-sdk delivery.AskWorker).
|
|
for _, a := range s.Accepts {
|
|
t.Workers = append(t.Workers, Worker{Stream: seatStreamName(s.Name), Consumer: WorkerName(s.Name, a),
|
|
Filter: "mesh.seat." + s.Name + ".accept." + a})
|
|
}
|
|
} else if len(s.Accepts) > 0 {
|
|
stream := seatStreamName(s.Name)
|
|
filter := "mesh.seat." + s.Name + ".accept.>"
|
|
if kind != "" {
|
|
filter = "mesh.seat." + s.Name + ".accept.*." + kind
|
|
}
|
|
t.Workers = append(t.Workers, Worker{Stream: stream, Consumer: WorkerName(s.Name, kind), Filter: filter})
|
|
}
|
|
for _, a := range s.Accepts {
|
|
switch {
|
|
case kind != "":
|
|
t.Subscribe = append(t.Subscribe, seatSubject(s, "accept", a+"."+kind))
|
|
case namesVerb(s.ByCaller, a):
|
|
t.Subscribe = append(t.Subscribe, seatSubject(s, "accept", a+".*"))
|
|
}
|
|
}
|
|
for _, e := range s.Emits {
|
|
switch {
|
|
case kind != "":
|
|
t.Publish = append(t.Publish, seatSubject(s, "event", e+"."+kind))
|
|
case namesVerb(s.ByCaller, e):
|
|
t.Publish = append(t.Publish, seatSubject(s, "event", e+".*"))
|
|
}
|
|
}
|
|
if kind != "" {
|
|
for _, v := range s.Proofs {
|
|
t.Publish = append(t.Publish, seatSubject(s, "proof", v+"."+kind))
|
|
}
|
|
}
|
|
}
|
|
for _, s := range uses {
|
|
if !s.isNewTraffic() {
|
|
continue
|
|
}
|
|
for _, a := range s.Accepts {
|
|
switch {
|
|
case namesVerb(s.ByCaller, a):
|
|
t.Publish = append(t.Publish, seatSubject(s, "accept", a+"."+module))
|
|
case s.Kinded && module == s.DeclaredBy:
|
|
// Work for a kind is put on its queue by the bench's own router, and by no other user.
|
|
t.Publish = append(t.Publish, seatSubject(s, "accept", a+".*"))
|
|
}
|
|
}
|
|
for _, e := range s.Emits {
|
|
if namesVerb(s.ByCaller, e) {
|
|
t.Subscribe = append(t.Subscribe, seatSubject(s, "event", e+"."+module))
|
|
}
|
|
}
|
|
for _, b := range s.Records {
|
|
if !safeSubject.MatchString(b) {
|
|
continue
|
|
}
|
|
t.Records = append(t.Records, "$JS.API.DIRECT.GET.KV_"+b+".$KV."+b+"."+module+".>")
|
|
}
|
|
if s.PulledByKind && len(s.Accepts) > 0 && safeSubject.MatchString(s.Name) {
|
|
t.Records = append(t.Records, DeadLetterReads(seatStreamName(s.Name))...)
|
|
}
|
|
}
|
|
for _, w := range watches {
|
|
if w.Kinded {
|
|
for _, e := range w.Emits {
|
|
t.Subscribe = append(t.Subscribe, seatSubject(w, "event", e+".*"))
|
|
}
|
|
if module == w.DeclaredBy {
|
|
// A code is answered by the bench's own router, and by no other watcher.
|
|
for _, v := range w.Proofs {
|
|
t.Answers = append(t.Answers, seatSubject(w, "proof", v+".*"))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
t.Publish = unique(t.Publish)
|
|
t.Subscribe = unique(t.Subscribe)
|
|
t.Answers = unique(t.Answers)
|
|
t.Records = unique(t.Records)
|
|
sort.Slice(t.Workers, func(i, j int) bool { return t.Workers[i].Consumer < t.Workers[j].Consumer })
|
|
return t
|
|
}
|
|
|
|
// DeadLetterReads is what a user of a seat pulled by kind is granted to hear which of the asks it put the seat's workers
|
|
// gave up on (novox/hq issue 734): an ordered consumer on DEAD_LETTERS filtered to the copies the controller keeps of that
|
|
// seat's stream (DeadLetterSubject(stream, *)), pulled, and its deletion. A copy carries the ask's own message id
|
|
// (Mesh-Dead-Msg-Id), so the asker ends the step it waits for instead of waiting for ever; it reads the seat's dead
|
|
// letters and no other stream's, and changes none.
|
|
func DeadLetterReads(stream string) []string {
|
|
return []string{
|
|
"$JS.API.STREAM.INFO." + DeadLettersStream,
|
|
"$JS.API.CONSUMER.CREATE." + DeadLettersStream + ".*." + DeadLetterSubject(stream, ">"),
|
|
"$JS.API.CONSUMER.MSG.NEXT." + DeadLettersStream + ".*",
|
|
"$JS.API.CONSUMER.DELETE." + DeadLettersStream + ".*",
|
|
}
|
|
}
|
|
|
|
// seatTrafficGrants is the bus permissions seat traffic needs: the subjects themselves, and the JetStream API a
|
|
// worker is pulled and acknowledged through and a record is read through.
|
|
func seatTrafficGrants(t SeatTraffic) (pub, sub []string) {
|
|
pub = append(pub, t.Publish...)
|
|
sub = append(sub, t.Subscribe...)
|
|
sub = append(sub, t.Answers...)
|
|
for _, w := range t.Workers {
|
|
pub = append(pub,
|
|
"$JS.API.CONSUMER.INFO."+w.Stream+"."+w.Consumer,
|
|
"$JS.API.CONSUMER.MSG.NEXT."+w.Stream+"."+w.Consumer,
|
|
"$JS.ACK."+w.Stream+"."+w.Consumer+".>")
|
|
}
|
|
pub = append(pub, t.Records...)
|
|
return pub, sub
|
|
}
|
|
|
|
// isNewTraffic says whether a seat carries any of the rules above, so a seat that carries none is
|
|
// composed exactly as before them.
|
|
func (s Seat) isNewTraffic() bool {
|
|
return s.Kinded || len(s.ByCaller) > 0 || len(s.Proofs) > 0 || len(s.Records) > 0 || s.PulledByKind
|
|
}
|
|
|
|
// plainVerbs is a seat's accepts or emits with those the rules above compose taken out: a verb named by
|
|
// its caller and every verb of a kinded bench are composed by SeatTrafficOf and nowhere else.
|
|
func plainVerbs(s Seat, verbs []string) []string {
|
|
if s.Kinded {
|
|
return nil
|
|
}
|
|
var out []string
|
|
for _, v := range verbs {
|
|
if !namesVerb(s.ByCaller, v) {
|
|
out = append(out, v)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// SeatTrafficObjects is the work queues and workers the seat traffic of every composed user implies
|
|
// (novox/hq ADR 0259 §3): a queue for each seat a holder takes work from, and each holder's worker on it —
|
|
// one per kind on a kinded bench, filtered to that kind, so the kinds never take each other's work. Only
|
|
// seats carrying the rules above; the mesh's own seats' queues are RaiseSeats'.
|
|
func SeatTrafficObjects(users []Principal) ([]Stream, []Consumer) {
|
|
streams := map[string]Stream{}
|
|
consumers := map[string]Consumer{}
|
|
add := func(module string, holds []Seat) {
|
|
for _, w := range SeatTrafficOf(module, holds, nil, nil).Workers {
|
|
seat, byKind := "", false
|
|
for _, s := range holds {
|
|
if seatStreamName(s.Name) == w.Stream {
|
|
seat, byKind = s.Name, s.PulledByKind
|
|
}
|
|
}
|
|
streams[w.Stream] = Stream{
|
|
Name: w.Stream,
|
|
Subjects: []string{"mesh.seat." + seat + ".accept.>"},
|
|
Retention: RetentionWorkQueue,
|
|
MaxAge: 7 * 24 * 60 * 60,
|
|
Why: "work submitted to the " + seat + " seat; its holders take it, each kind its own, and it queues while nobody does",
|
|
}
|
|
c := Consumer{
|
|
Name: w.Consumer,
|
|
Stream: w.Stream,
|
|
Filters: []string{w.Filter},
|
|
AckWaitSeconds: 60,
|
|
// No bound on redelivery: a channel away for a day keeps its work, offered again later and
|
|
// later by its holder's runtime (novox/hq ADR 0259; the correctness review of 2026-10-08).
|
|
MaxDeliver: 0,
|
|
Why: module + " holds " + seat + "; it pulls one ask at a time and acknowledges once it has " +
|
|
"recorded it, so a crash redelivers rather than loses",
|
|
}
|
|
if byKind {
|
|
// A build seat's ask is offered a bounded number of times (issue 1907): the holder that takes it past
|
|
// mesh-sdk delivery.AskRuns ends it failed saying it is poison, and the bound stops a run that kills
|
|
// every node it reaches even when that holder dies too.
|
|
c.MaxDeliver = BuildAskMaxDeliver
|
|
c.Why = module + " holds " + seat + "; each holder pulls an ask when it has a free slot, and an ask that " +
|
|
"no run ends is offered at most " + fmt.Sprint(BuildAskMaxDeliver) + " times, its step failed as poison"
|
|
}
|
|
consumers[w.Consumer] = c
|
|
}
|
|
}
|
|
for _, p := range users {
|
|
switch p.Kind {
|
|
case KindModule:
|
|
add(p.Module, p.Holds)
|
|
case KindNodeTools:
|
|
for _, d := range p.Carries {
|
|
add(d.Module, d.Holds)
|
|
}
|
|
}
|
|
}
|
|
var ss []Stream
|
|
for _, s := range streams {
|
|
ss = append(ss, s)
|
|
}
|
|
sort.Slice(ss, func(i, j int) bool { return ss[i].Name < ss[j].Name })
|
|
var cs []Consumer
|
|
for _, c := range consumers {
|
|
cs = append(cs, c)
|
|
}
|
|
sort.Slice(cs, func(i, j int) bool { return cs[i].Name < cs[j].Name })
|
|
return ss, cs
|
|
}
|
|
|
|
// trustedTraffic is why a module's seat traffic is the trusted holder's (novox/hq ADR 0259 §8), or "": it
|
|
// says a seat's event to one caller each (a warrant), or holds a kind of a kinded bench that proves its sender.
|
|
func trustedTraffic(d Declared) string {
|
|
for _, s := range d.Holds {
|
|
for _, e := range s.Emits {
|
|
if namesVerb(s.ByCaller, e) {
|
|
return "it says " + s.Name + "'s " + e + " to one caller each"
|
|
}
|
|
}
|
|
if s.Kinded && namesVerb(s.Capabilities, "verified-sender") {
|
|
return "it holds " + s.Name + " of kind " + s.Kind + ", which proves its sender"
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// TrafficQueues is the work queue of every seat naming its caller or its kind that accepts work, held or not
|
|
// (novox/hq ADR 0259 §3): what is submitted before a holder is assigned waits for it.
|
|
func TrafficQueues(seats []Seat) []Stream {
|
|
var out []Stream
|
|
seen := map[string]bool{}
|
|
for _, s := range seats {
|
|
if len(s.Accepts) == 0 || !s.isNewTraffic() || seen[s.Name] {
|
|
continue
|
|
}
|
|
seen[s.Name] = true
|
|
out = append(out, Stream{Name: seatStreamName(s.Name), Subjects: []string{"mesh.seat." + s.Name + ".accept.>"},
|
|
Retention: RetentionWorkQueue, MaxAge: 7 * 24 * 60 * 60,
|
|
Why: "work submitted to the " + s.Name + " seat; its holders take it, each kind its own, and it queues while nobody does"})
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
|
|
return out
|
|
}
|