diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index b60896a0..6d72de3c 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -189,6 +189,22 @@ func TestTheGrantsPartWhole(t *testing.T) { } t.Cleanup(func() { _ = osub.Unsubscribe() }) } + // D-30 (novox/hq ADR 0636 §1): the seat mesh-catalog's holder answers the seat list with no trunk seat named, so the reader + // has heard the seat and holds the bootstrapped definitions, and every module is composed. (The unit case holds a module + // while the list is unread.) + var seatAnswering atomic.Bool + seatAnswering.Store(true) + for _, subject := range []string{"mesh.seat.mesh-catalog.call.seat-definitions", "mesh.seat.mesh-catalog.call.seat-definitions.>"} { + if _, err := nc.Subscribe(subject, func(m *nats.Msg) { + if !seatAnswering.Load() { + _ = m.Respond([]byte(`{"error":"not connected yet","node":"a"}`)) + return + } + answer(m, map[string]any{"revision": 1, "seats": []any{}}) + }); err != nil { + t.Fatal(err) + } + } // A membership is never said on its subject any more (issue 1126). var said atomic.Int32 if _, err := nc.Subscribe("mesh.assignment.>", func(*nats.Msg) { said.Add(1) }); err != nil { diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 012ac6cb..e1cd993f 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -36,6 +36,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/seatlist" "git.novox.be/novox/mesh-sdk/go/views" @@ -107,6 +108,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } bus := newBusReader(ctx, js) + // The module has one durable consumer of EVENTS: one hub pulls it and hands each message to the part that hears it (the + // owners' views events, the seat mesh-catalog's verdicts and seat list events, the other seats' events the follower hears), + // started once all are registered. + hub := process.NewEventHub(logger.Printf) + defer hub.Stop() // The owners' five states by their first views (ADR 0615, ADR 0622): asked through the part's own account, then // followed on the owners' views families, which the hearing below hands them. sayCond := sayVia(ctx, js, account.Module) @@ -124,6 +130,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Objects: sealedObjects{ctx: ctx, js: js}, Logf: logger.Printf, Bus: grants.BusAddress{URL: account.URL, Fingerprint: account.Fingerprint}} f.Watch = fv + // The seat list: read over the connection and the hub, given to catalogue at every change, and a pass composes again then. + gate := &seatGate{} + gate.follow(seatlist.ProcessBus(conn.Call, hub), f.Wake, logger.Printf) + f.Part.SeatList = gate f.Part.Say = sayCond fv.Start(ctx, f.Wake) f.Asked = func(wake func()) error { @@ -142,11 +152,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // dropped carry the bindings: ADR 0595 §1, §7), each on the durable consumer its own membership names; one not // there yet is opened at the next change, and one opened is never opened twice. opened := map[string]bool{} - // The module has one durable consumer of EVENTS: one hub pulls it and hands each message to the part that hears it - // (the owners' views events, the seat mesh-catalog's verdicts, the other seats' events the follower hears), started - // once all are registered. - hub := process.NewEventHub(logger.Printf) - defer hub.Stop() registered := false f.Hear = func(handle func(string, []byte)) error { if !registered { diff --git a/modules/mesh-grants/cmd/mesh-grants/seatlist.go b/modules/mesh-grants/cmd/mesh-grants/seatlist.go new file mode 100644 index 00000000..3c477fb2 --- /dev/null +++ b/modules/mesh-grants/cmd/mesh-grants/seatlist.go @@ -0,0 +1,67 @@ +package main + +import ( + "errors" + "mesh-grants/internal/grants" + "sync/atomic" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/seatlist" +) + +// The grants part composes from the seat list it reads (novox/hq ADR 0635 §1, ADR 0636 §1, D-30): through the seat +// mesh-catalog's verb seat-definitions and its two events, from the bootstrapped definitions until the seat has answered. What +// the reader holds is given to catalogue at every change, which composes every membership and grant, and a module that names a +// seat the list does not yet is held back, as a manifest waiting for a judge is, and what runs is composed from its running +// build's manifest. The process reads over its connection and the event hub; the tools' bundle over the runtime. + +// seatListTools are the reader's two tools (ADR 0636 §7): every operator's session and the healers may read again. +var seatListAgainCallers = []string{stdio.AnyPerson, "mesh-healers"} + +// seatGate is the reader the part gates by, set once it is started; before, it acts on nothing new. +type seatGate struct { + reader atomic.Pointer[seatlist.Reader] +} + +func (g *seatGate) MayAct(seats ...string) bool { + r := g.reader.Load() + return r != nil && r.MayAct(seats...) +} + +// follow starts the reader over bus: what it holds is given to catalogue at once and at every change, and changed is called +// then (a pass composes again). +func (g *seatGate) follow(bus seatlist.Bus, changed func(), logf func(string, ...any)) *seatlist.Reader { + reader, err := seatlist.Start(seatlist.Config{Bus: bus, Bootstrap: seatlist.Bootstrapped(), + OnChange: func(seats []catalogue.Seat) { + catalogue.UseSeats(seats) + if changed != nil { + changed() + } + }}) + if err != nil { + logf("the grants part cannot follow its seat list, so it composes only the modules the bootstrapped seats name: %v", err) + } + if reader != nil { + catalogue.UseSeats(reader.Seats()) + g.reader.Store(reader) + } + return reader +} + +// tools are the two tools over whatever reader the gate holds at the call. +func (g *seatGate) tools() []stdio.Tool { + names := []string{grants.ToolSeatList, grants.ToolSeatListAgain} + shape := seatlist.Tools(&seatlist.Reader{}, names[0], names[1], seatListAgainCallers) + for i := range shape { + run := i + shape[i].Run = func(args map[string]any) (any, error) { + r := g.reader.Load() + if r == nil { + return nil, errors.New("the seat list reader has not started yet: the runtime has not served this module") + } + return seatlist.Tools(r, names[0], names[1], seatListAgainCallers)[run].Run(args) + } + } + return shape +} diff --git a/modules/mesh-grants/cmd/mesh-grants/tools.go b/modules/mesh-grants/cmd/mesh-grants/tools.go index dc973aad..9c0f81db 100644 --- a/modules/mesh-grants/cmd/mesh-grants/tools.go +++ b/modules/mesh-grants/cmd/mesh-grants/tools.go @@ -5,11 +5,13 @@ import ( "encoding/json" "fmt" "maps" + "os" "sort" "strings" "sync" "time" + "git.novox.be/novox/mesh-sdk/go/seatlist" "git.novox.be/novox/mesh-sdk/go/views" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" @@ -254,8 +256,14 @@ func recordHandAct(what, why string) error { // serveTools serves the tools to the runtime until it stops. func serveTools() error { tv := &toolViews{} - p := &grants.Part{Read: runtimeReader{views: tv}, Call: runtimeBus{}, Store: runtimeStore{}, Emit: runtimeBus{}} - tools := toolsOf(p) + gate := &seatGate{} + p := &grants.Part{Read: runtimeReader{views: tv}, Call: runtimeBus{}, Store: runtimeStore{}, Emit: runtimeBus{}, SeatList: gate} + // The reader asks through the runtime, which serves once Serve has its channel. + go func() { + <-stdio.Served() + gate.follow(seatlist.StdioBus(), nil, func(f string, a ...any) { fmt.Fprintf(os.Stderr, "mesh-grants: "+f+"\n", a...) }) + }() + tools := append(toolsOf(p), gate.tools()...) for i := range tools { run := tools[i].Run tools[i].Run = func(a map[string]any) (any, error) { tv.fresh(); return run(a) } diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 69ea7ee3..f54ed06e 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -2231,6 +2231,31 @@ func TestGrants(t *testing.T) { } }) + // D-30 (novox/hq ADR 0636 §1): until the seat list has been read, a module whose seats are not all bootstrapped ones is held back + // as a manifest waiting for a judge is, named `manifest of : it waits for the seat list`, and composed once the + // list names its seats; the bus module, which claims a bootstrapped seat, is composed throughout. + t.Run("the seat list unread holds a module whose seats are not bootstrapped", func(t *testing.T) { + b := fixture() + p := partOn(b) + gate := &seatGate{bootstrapped: []string{"mesh-broker"}} + p.SeatList = gate + p.Compose() + refused := strings.Join(p.Last().Refused, "\n") + for _, module := range []string{"mod-x", "mod-y"} { + if !strings.Contains(refused, "manifest of "+module+": it waits for the seat list") { + t.Errorf("%s is not held while the seat list is unread: %v", module, p.Last().Refused) + } + } + if strings.Contains(refused, "manifest of nats") { + t.Errorf("the bus module claims a bootstrapped seat and is held: %v", p.Last().Refused) + } + gate.read = true + p.Compose() + if r := p.Last().Refused; len(r) != 0 { + t.Errorf("held after the seat list was read: %v", r) + } + }) + // DLV-G1, 1966/AC32-AC33 (novox/hq ADR 0595 §1, §7; design 25 § 12.2): the bus module's created writes the binding into // the module's membership on every machine it is assigned, one assigned later included; recreated rewrites it; dropped // removes it; before created, on a refused, or from an emitter holding no mesh-broker, none. @@ -3280,3 +3305,25 @@ func waitFor(t *testing.T, ok func() bool) { } t.Fatal("never came") } + +// seatGate is a seat list reader as a test gives it: unread, it lets a module act whose seats are all bootstrapped ones (and +// at least one); read, any. +type seatGate struct { + bootstrapped []string + read bool +} + +func (g *seatGate) MayAct(seats ...string) bool { + if g.read { + return true + } + if len(seats) == 0 { + return false + } + for _, s := range seats { + if !slices.Contains(g.bootstrapped, s) { + return false + } + } + return true +} diff --git a/modules/mesh-grants/internal/grants/consumers.go b/modules/mesh-grants/internal/grants/consumers.go index b77fae66..48bbd2ce 100644 --- a/modules/mesh-grants/internal/grants/consumers.go +++ b/modules/mesh-grants/internal/grants/consumers.go @@ -200,7 +200,7 @@ func (p *Part) MatchConsumer(stream, name string) (any, error) { if stream == "" || name == "" { return nil, errors.New("consumer-match names a stream and a consumer") } - in := ReadInputs(p.Read, &p.cache) + in := p.inputs() recs, _ := Records(in) users, err := buscompose.Users(recs) if err != nil { diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index 0b330afa..49afd3f6 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -61,7 +61,10 @@ const ( ) // Invokes is every tool the part calls. -var Invokes = append([]string{CallUsers, CallUserCan, CallStreams, CallRootFree, CallHandAct}, firstViewTools()...) +var Invokes = append([]string{CallUsers, CallUserCan, CallStreams, CallRootFree, CallHandAct, CallSeatDefinitions}, firstViewTools()...) + +// CallSeatDefinitions reads the seat list (mesh-sdk seatlist): the verb of the seat mesh-catalog. +const CallSeatDefinitions = "seat:mesh-catalog.seat-definitions" // CallHandAct records a person's act (grants_first_view_again). const CallHandAct = "seat:mesh-hand-acts.hand-act" @@ -204,6 +207,30 @@ const BusModule = "nats" type Manifests struct { mu sync.Mutex parsed map[string]catalogue.Manifest + // gate says whether a manifest naming some seats may be composed yet (the seat list reader, ADR 0636 §1); nil gates none. + gate SeatGate +} + +// SeatGate is what the part asks of the seat list reader before it composes a module new to it: whether a module that +// claims or invokes the given seats may be acted on (seatlist.Reader.MayAct). Until the seat list has been read only a module +// whose seats are all bootstrapped ones may; the rest is held as a manifest waiting for a judge is, its running build's +// manifest composing what runs, so nothing that runs is stopped (held.go). +type SeatGate interface { + MayAct(seats ...string) bool +} + +// SeatsOf are the seats a manifest claims or invokes, which the seat list must name before the part composes the module. +func SeatsOf(m catalogue.Manifest) []string { + var seats []string + for _, c := range m.Claims { + seats = append(seats, c.Name) + } + for _, in := range m.Invokes { + if seat, _, ok := catalogue.SeatInvoke(in); ok { + seats = append(seats, seat) + } + } + return seats } func (c *Manifests) get(r Reader, digest string) (catalogue.Manifest, error) { @@ -263,6 +290,9 @@ func each(r Reader, state string, gaps *[]Gap) map[string][]byte { // ReadInputs reads what the shadow composes from. It never fails: what it cannot read is a gap. func ReadInputs(r Reader, cache *Manifests) Inputs { + cache.mu.Lock() + gate := cache.gate + cache.mu.Unlock() in := Inputs{Assigned: map[string][]string{}, Manifests: map[string]catalogue.Manifest{}, Digests: map[string]string{}, Fingerprints: map[string]string{}, NoFingerprint: map[string]string{}, Running: map[string]catalogue.Manifest{}, Values: map[string]any{}, Gaps: append([]Gap(nil), standingGaps...)} @@ -392,6 +422,9 @@ func ReadInputs(r Reader, cache *Manifests) Inputs { if err == nil { m, err = cache.get(r, v.Manifest) } + if err == nil && gate != nil && !gate.MayAct(SeatsOf(m)...) { + err = errors.New("it waits for the seat list: the list has not been read yet and the manifest names seats the bootstrapped list does not (novox/hq ADR 0636 §1)") + } if err != nil { // Every manifest, assigned or not: one refused declares no seat, so its holders compose as placed nowhere. g := Gap{Input: "manifest of " + module, Why: err.Error()} diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index a23ab3b0..7fe5eb05 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -162,6 +162,9 @@ type Part struct { // Say says a condition to the operator: the process's; nil in the tools. Say func(oc.Condition) error Now func() time.Time + // SeatList is the seat list reader, which gates a module whose seats the list has not named (inputs.go, SeatGate); nil + // gates none. + SeatList SeatGate cache Manifests mu sync.Mutex @@ -279,7 +282,7 @@ func (p *Part) Compose() Result { }() // A first view's condition the condition keeper did not take is said again at every pass until it lands. p.Views.SayUnsaid() - in := ReadInputs(p.Read, &p.cache) + in := p.inputs() p.keepBrokers(in) if bound, err := p.boundStreams(); err != nil { res.Errors = append(res.Errors, err.Error()) @@ -542,3 +545,11 @@ func names(us []User) []string { } return out } + +// inputs reads what the part composes from, a module whose seats the seat list has not named held back (SeatGate). +func (p *Part) inputs() Inputs { + p.cache.mu.Lock() + p.cache.gate = p.SeatList + p.cache.mu.Unlock() + return ReadInputs(p.Read, &p.cache) +} diff --git a/modules/mesh-grants/internal/grants/rootfree.go b/modules/mesh-grants/internal/grants/rootfree.go index a270dcc0..83bc8626 100644 --- a/modules/mesh-grants/internal/grants/rootfree.go +++ b/modules/mesh-grants/internal/grants/rootfree.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "slices" "sort" "strings" @@ -42,7 +43,11 @@ const EventAccountIssued = "mesh-vault." + buscompose.AccountIssuedEvent // Consumes is every event the part hears, as its manifest's consumes name them. // The seat mesh-broker's created, recreated and dropped are heard as every seat's event, `.` on EVENTS (the // lead's ruling, registry milestone 13:0x), never on the bus module's own stream. -var Consumes = append(slices.Clone(seatEvents), viewsEvents()...) +var Consumes = append(append(slices.Clone(seatEvents), seatListEvents...), viewsEvents()...) + +// seatListEvents are the two events of the seat mesh-catalog the seat list reader follows (mesh-sdk seatlist), heard by it on the +// module's event hub, so no handler of the hub takes them besides. +var seatListEvents = []string{meshcatalog.Seat + "." + meshcatalog.EventSeatDefinitionChanged, meshcatalog.Seat + "." + meshcatalog.EventSeatDefinitionRemoved} // seatEvents are the events of seats the part follows its passes by, as `.`. var seatEvents = []string{EventKeeperRaised, EventKeeperCleared, EventAccountIssued, meshbroker.Seat + "." + BrokerCreated, diff --git a/modules/mesh-grants/internal/grants/tools.go b/modules/mesh-grants/internal/grants/tools.go index a94e32d8..247d4081 100644 --- a/modules/mesh-grants/internal/grants/tools.go +++ b/modules/mesh-grants/internal/grants/tools.go @@ -26,11 +26,14 @@ const ( // ToolFirstViews and ToolFirstViewAgain read and ask again the part's first views (firstviews.go). ToolFirstViews = "grants_first_views" ToolFirstViewAgain = "grants_first_view_again" + // ToolSeatList and ToolSeatListAgain read and read again the seat list the part composes from (mesh-sdk seatlist.Tools). + ToolSeatList = "grants_seat_list" + ToolSeatListAgain = "grants_seat_list_again" ) // Tools is every tool, in the manifest's order. var Tools = []string{ToolGrants, ToolMembership, ToolUserList, ToolComposeNow, ToolAssert, ToolAccounts, ToolConsumerMatch, - ToolAcceptHashes, ToolFirstViews, ToolFirstViewAgain} + ToolAcceptHashes, ToolFirstViews, ToolFirstViewAgain, ToolSeatList, ToolSeatListAgain} // WritesGrants is what grants says of the part's own grants: written once the list the bus loaded carries a build's // accounts, and the one key the assigner reads (ADR 0571 clause 9, which changes ADR 0547 clause 5). @@ -105,7 +108,7 @@ func (p *Part) MembershipOf(node, module string) (any, error) { if node == "" || module == "" { return nil, errors.New("membership names a node and a module, or neither for every membership held back") } - in := ReadInputs(p.Read, &p.cache) + in := p.inputs() recs, left := Records(in) notFree := WithRootFree(&recs, p.Call) m, ok := Memberships(recs)[node+"."+module] @@ -277,7 +280,7 @@ func (p *Part) AssertOne(module string) (any, error) { held, heldErr := HeldStreams(p.Call) var nodes []string if module == buscompose.EngineModule { - nodes = ReadInputs(p.Read, &p.cache).Nodes // every node's bucket of compositions (ADR 0593 §6) + nodes = p.inputs().Nodes // every node's bucket of compositions (ADR 0593 §6) } a := Assert(module, v.Manifest, m, nodes, held, heldErr) out := map[string]any{"assertion": a} diff --git a/modules/mesh-grants/module.json b/modules/mesh-grants/module.json index 5363b8ca..e0e69f04 100644 --- a/modules/mesh-grants/module.json +++ b/modules/mesh-grants/module.json @@ -22,7 +22,9 @@ "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", "mesh-catalog.manifest-accepted", - "mesh-catalog.manifest-refused" + "mesh-catalog.manifest-refused", + "mesh-catalog.seat-definition-changed", + "mesh-catalog.seat-definition-removed" ], "uses": [ "operator-channel" @@ -37,6 +39,7 @@ "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.verdicts", + "seat:mesh-catalog.seat-definitions", "seat:mesh-hand-acts.hand-act" ], "emits": [ @@ -56,7 +59,9 @@ "grants_consumer_match", "grants_accept_hashes", "grants_first_views", - "grants_first_view_again" + "grants_first_view_again", + "grants_seat_list", + "grants_seat_list_again" ], "state": [ {