GR13: a build seat's queues take named asks beside main's, each asker granted its own
Each queue gets a named worker filtered to accept.<queue>.*; an asker is granted accept.<queue>.<its seat> for the queues of the kinds it is a caller of, read from the definitions' callers lists; main's accept.<queue> is the checker's and the builder's alone. The purge of named subjects and workers (grants_named_purge) is not in this commit. Builds: design 47 G §543, §544, §600, §601, A-163, A-170, A-184 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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))
|
||||
}
|
||||
}
|
||||
@@ -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:<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
|
||||
|
||||
@@ -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.<queue>.<asker>`, 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.<queue>`, 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.<queue>.<asker>` for the queues whose
|
||||
// kinds it is a caller of, its name the seat it holds, and main's `accept.<queue>` 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
|
||||
|
||||
@@ -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{}
|
||||
|
||||
@@ -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:<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]
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user