Grants part: composes from the seat list it reads (reader over its connection and the event hub, catalogue and the pass follow it, a module whose seats the unread list does not name is held as a manifest waiting for a judge is, the two runtime tools, the manifest invokes and consumes) (D-30; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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) }
|
||||
|
||||
@@ -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 <module>: 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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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()}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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, `<seat>.<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 `<seat>.<event>`.
|
||||
var seatEvents = []string{EventKeeperRaised, EventKeeperCleared, EventAccountIssued, meshbroker.Seat + "." + BrokerCreated,
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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": [
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user