diff --git a/modules/mesh-grants/internal/buscompose/named_asks_test.go b/modules/mesh-grants/internal/buscompose/named_asks_test.go new file mode 100644 index 00000000..066edd62 --- /dev/null +++ b/modules/mesh-grants/internal/buscompose/named_asks_test.go @@ -0,0 +1,69 @@ +package buscompose + +import "testing" + +// Design 47 G §543, §544, §600: a build seat's ask names its asker, each queue has a named worker beside main's, an asker +// is granted the queues of its kinds under its own seat's name alone, and main's subject is the checker's and the builder's. +func TestNamedAsks(t *testing.T) { + build := Seat{Name: "node-build-go", Scope: "node", Accepts: []string{"delivery", "check", "recheck"}, PulledByKind: true, + AcceptCallers: map[string][]string{ + "delivery": {"mesh-builds"}, + "check": {"mesh-pipeline-runner", "mesh-catalog", "mesh-delivery-checker"}, + "recheck": {"mesh-delivery-checker"}, + }} + asker := func(module string, held ...string) Principal { + p := Principal{Kind: KindModule, Node: "one", Module: module, Uses: []Seat{build}} + for _, h := range held { + p.Holds = append(p.Holds, Seat{Name: h}) + } + return p + } + const base = "mesh.seat.node-build-go.accept." + for _, c := range []struct { + who Principal + want map[string]bool + }{ + {asker("runner-module", "mesh-pipeline-runner"), map[string]bool{ + base + "check.mesh-pipeline-runner": true, base + "delivery.mesh-pipeline-runner": false, + base + "check.mesh-catalog": false, base + "check": false, base + "delivery": false}}, + {asker("an-indexer", "mesh-catalog"), map[string]bool{ + base + "check.mesh-catalog": true, base + "check.mesh-pipeline-runner": false, base + "recheck.mesh-catalog": false}}, + {asker("the-builder", "mesh-builds"), map[string]bool{ + base + "delivery.mesh-builds": true, base + "delivery": true, base + "check": true, base + "check.mesh-builds": false}}, + {asker("the-checker", "mesh-delivery-checker"), map[string]bool{ + base + "check.mesh-delivery-checker": true, base + "recheck.mesh-delivery-checker": true, base + "recheck": true, + base + "delivery.mesh-delivery-checker": false}}, + {asker("a-stranger"), map[string]bool{base + "check": false, base + "delivery": false, base + "check.a-stranger": false}}, + } { + perms, err := PermissionsFor(c.who) + if err != nil { + t.Fatal(err) + } + for subject, want := range c.want { + if got := MayPublish(perms, subject); got != want { + t.Errorf("%s may publish %s: %v, want %v", c.who.Module, subject, got, want) + } + } + } + // Each queue's workers: main's and the named one, filtered to subjects that overlap nothing. + held := Seat{Name: "node-build-go", Scope: "node", Accepts: []string{"delivery", "check"}, PulledByKind: true} + tr := SeatTrafficOf("go-module", []Seat{held}, nil, nil) + filters := map[string]string{} + for _, w := range tr.Workers { + filters[w.Consumer] = w.Filter + } + for consumer, filter := range map[string]string{ + "SEAT_NODE_BUILD_GO_CHECK_worker": base + "check", + "SEAT_NODE_BUILD_GO_CHECK_named_worker": base + "check.*", + "SEAT_NODE_BUILD_GO_DELIVERY_worker": base + "delivery", + "SEAT_NODE_BUILD_GO_DELIVERY_named_worker": base + "delivery.*", + } { + if filters[consumer] != filter { + t.Errorf("worker %s filters %q, want %q", consumer, filters[consumer], filter) + } + } + _, consumers := SeatTrafficObjects([]Principal{{Kind: KindModule, Node: "one", Module: "go-module", Holds: []Seat{held}}}) + if len(consumers) != 4 { + t.Errorf("the work queue's consumers: %d, want 4 (a main and a named worker for each of two queues)", len(consumers)) + } +} diff --git a/modules/mesh-grants/internal/buscompose/nats.go b/modules/mesh-grants/internal/buscompose/nats.go index 12fd9c8d..9d9942b7 100644 --- a/modules/mesh-grants/internal/buscompose/nats.go +++ b/modules/mesh-grants/internal/buscompose/nats.go @@ -103,6 +103,10 @@ type Seat struct { // PulledByKind says each accepted kind is a worker of its own on the seat's queue, filtered to that kind, which // every holder pulls in the order Accepts lists them: the build seats' (novox/hq ADR 0525 §2, ADR 0536 §4). PulledByKind bool + // AcceptCallers are, for a seat pulled by kind, the seats that may ask each of its queues, by queue: the `seat:` + // entries of the callers of the accepted kinds the definition puts on that queue (design 47 G §543, §600). The asker + // in a named ask's subject is the asking seat's name (A-163). Empty where the seat list is unread. + AcceptCallers map[string][]string // HeldByTheEngine says every node's engine holds the seat on its own account (catalogue.Seat.HeldByTheEngine, novox/hq // ADR 0563 §1): each of its events names the engine's machine as its last token (machineEvent). HeldByTheEngine bool diff --git a/modules/mesh-grants/internal/buscompose/seattraffic.go b/modules/mesh-grants/internal/buscompose/seattraffic.go index 5e369c6b..0cb21b5f 100644 --- a/modules/mesh-grants/internal/buscompose/seattraffic.go +++ b/modules/mesh-grants/internal/buscompose/seattraffic.go @@ -35,6 +35,16 @@ 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} +// NamedWorkerName is the worker of a pulled-by-kind seat's named asks, `accept..`, beside the queue's own (WorkerName): +// two filters that overlap nothing, so the work queue takes both (design 47 G §544). +func NamedWorkerName(seat, queue string) string { + return "SEAT_" + upperSnake(seat) + "_" + upperSnake(queue) + "_named_worker" +} + +// MainAskers are the seats that may publish a build seat's ask on main's subject `accept.`, which names no asker: the +// checker and the builder, until main's subject goes (design 47 G §543, §600, A-184). +var MainAskers = []string{"mesh-delivery-checker", "mesh-builds"} + // 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 == "" { @@ -86,6 +96,9 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic { 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}) + // Its named asks have a worker of their own (§544); a holder of G pulls it, one before G knows only main's. + t.Workers = append(t.Workers, Worker{Stream: seatStreamName(s.Name), Consumer: NamedWorkerName(s.Name, a), + Filter: "mesh.seat." + s.Name + ".accept." + a + ".*"}) } } else if len(s.Accepts) > 0 { stream := seatStreamName(s.Name) @@ -121,7 +134,25 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic { if !s.isNewTraffic() { continue } + if s.PulledByKind { + // **An ask names its asker** (§543, §600): each asker is granted `accept..` for the queues whose + // kinds it is a caller of, its name the seat it holds, and main's `accept.` is the checker's and the + // builder's alone. + for _, q := range s.Accepts { + for _, h := range holds { + if namesVerb(s.AcceptCallers[q], h.Name) { + t.Publish = append(t.Publish, seatSubject(s, "accept", q+"."+h.Name)) + } + if namesVerb(MainAskers, h.Name) { + t.Publish = append(t.Publish, seatSubject(s, "accept", q)) + } + } + } + } for _, a := range s.Accepts { + if s.PulledByKind { + break + } switch { case namesVerb(s.ByCaller, a): t.Publish = append(t.Publish, seatSubject(s, "accept", a+"."+module)) @@ -205,7 +236,7 @@ func (s Seat) isNewTraffic() bool { // 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 { + if s.Kinded || s.PulledByKind { return nil } var out []string diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index 251515b1..0308ae7a 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -190,6 +190,8 @@ type Inputs struct { // EventSubjects are, by seat, the subjects of its events that carry parts, by the event's local name, as the definitions // give them: where a node-scoped seat's event names its machine (buscompose.Seat.EventSubjects). EventSubjects map[string]map[string]string + // AcceptCallers are, by pulled-by-kind seat and queue, the seats that may ask it (acceptCallersOf). + AcceptCallers map[string]map[string][]string // Defined are the events a seat definition of the seat list defines, by local name, and mesh-sdk's finding events: the // events a `*` emitter before them reaches each declaring module's own stream for (design 47 G §386, §474, §644). Defined map[string]bool @@ -370,6 +372,7 @@ func ReadInputs(r Reader, cache *Manifests) Inputs { in.Kept = buscompose.KeptEventsOf(d.Definitions()) in.Chunked = buscompose.ChunkVerbsOf(d.Definitions()) in.EventSubjects = eventSubjectsOf(d.Definitions()) + in.AcceptCallers = acceptCallersOf(d.Definitions()) in.Defined = definedEventsOf(d.Definitions()) } nodes := map[string]bool{} diff --git a/modules/mesh-grants/internal/grants/kept.go b/modules/mesh-grants/internal/grants/kept.go index 42afb8bb..3994d901 100644 --- a/modules/mesh-grants/internal/grants/kept.go +++ b/modules/mesh-grants/internal/grants/kept.go @@ -135,3 +135,45 @@ func withEventSubjects(subjects map[string]map[string]string, d *buscompose.Decl } } } + +// acceptCallersOf is, by pulled-by-kind seat and then by queue, the seats that may ask the queue: the `seat:` entries +// of the callers of every accepted kind the definition puts on it (a kind's `stream` is its queue, else its name is). Never a +// name of the grants part's own: it reads the definitions' lists (design 47 G §543, §600, A-163). +func acceptCallersOf(defs map[string]seatdef.Definition) map[string]map[string][]string { + out := map[string]map[string][]string{} + for _, d := range defs { + if !d.PulledByKind { + continue + } + for _, a := range d.Accepts { + if a.Callers == nil || a.Callers.IsText() { + continue + } + queue := a.Name + if a.Stream != "" { + queue = a.Stream + } + for _, e := range a.Callers.List { + if seat, ok := strings.CutPrefix(e, seatdef.CallerSeatPrefix); ok && !slices.Contains(out[d.Seat][queue], seat) { + if out[d.Seat] == nil { + out[d.Seat] = map[string][]string{} + } + out[d.Seat][queue] = append(out[d.Seat][queue], seat) + } + } + } + } + return out +} + +// withAcceptCallers gives each pulled-by-kind seat a declaration holds or uses the askers of its queues (Seat.AcceptCallers). +func withAcceptCallers(askers map[string]map[string][]string, d *buscompose.Declared) { + if len(askers) == 0 { + return + } + for _, list := range []*[]buscompose.Seat{&d.Holds, &d.Uses, &d.Watches} { + for i := range *list { + (*list)[i].AcceptCallers = askers[(*list)[i].Name] + } + } +} diff --git a/modules/mesh-grants/internal/grants/named_asks_test.go b/modules/mesh-grants/internal/grants/named_asks_test.go new file mode 100644 index 00000000..bc58743c --- /dev/null +++ b/modules/mesh-grants/internal/grants/named_asks_test.go @@ -0,0 +1,36 @@ +package grants + +import ( + "slices" + "testing" + + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +// The askers of a build seat's queue are read from the definition's callers lists (design 47 G §543, §600, A-163): a kind's +// `stream` is its queue, a text callers names none, and only `seat:` entries are askers. +func TestAcceptCallersOf(t *testing.T) { + msg := func(name, stream string, callers ...string) seatdef.Message { + m := seatdef.Message{Name: name, Stream: stream} + if callers != nil { + m.Callers = &seatdef.Callers{List: callers} + } + return m + } + got := acceptCallersOf(map[string]seatdef.Definition{ + "node-build-go": {Seat: "node-build-go", PulledByKind: true, Accepts: []seatdef.Message{ + msg("check", "", "seat:mesh-catalog", "operator"), + msg("step", "check", "seat:mesh-pipeline-runner"), + msg("delivery", "", "seat:mesh-builds"), + {Name: "recheck", Callers: &seatdef.Callers{Text: "the checker"}}, + }}, + "other": {Seat: "other", Accepts: []seatdef.Message{msg("go", "", "seat:x")}}, + }) + if len(got) != 1 { + t.Fatalf("seats: %v, want node-build-go alone", got) + } + q := got["node-build-go"] + if !slices.Equal(q["check"], []string{"mesh-catalog", "mesh-pipeline-runner"}) || !slices.Equal(q["delivery"], []string{"mesh-builds"}) || len(q) != 2 { + t.Errorf("queues: %v", q) + } +} diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index 1442daf8..5fce46a5 100644 --- a/modules/mesh-grants/internal/grants/records.go +++ b/modules/mesh-grants/internal/grants/records.go @@ -106,6 +106,7 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) { // so a grant and its binding are written together, never one before the other (issue 1966, the kind of issue 249). d.Bindings = in.Bindings[m.Module] withEventSubjects(in.EventSubjects, &d) + withAcceptCallers(in.AcceptCallers, &d) return d } d := declare(m) @@ -137,6 +138,7 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) { out.Engine = &buscompose.Declared{Module: m.Module, Consumes: m.Consumes, Watches: watches, Invokes: m.Invokes, State: bucketsFor(m, in.Nodes), Logs: logsOf(m), Streams: streamsOf(m)} withEventSubjects(in.EventSubjects, out.Engine) + withAcceptCallers(in.AcceptCallers, out.Engine) } return out, gaps }