diff --git a/README.md b/README.md index 0c40988..b419c2e 100644 --- a/README.md +++ b/README.md @@ -72,7 +72,7 @@ of the words agents use for the mesh's (`logs` finds the journal). Best first: a words in a name or a replaced command, then words in a description; seats before modules on a tie. Discovery asks the bus (ADR 0197): every runtime answers the NATS services protocol's `$SRV.PING` and -`$SRV.INFO` with what it serves at that moment, and the console reads where the placement seat's records +`$SRV.INFO` with what it serves at that moment, and the console reads where the assigner seat's records place each module. The console waits at least 750 ms, and up to 5 s for every runtime that answered PING to send what it serves — a large runtime's answer crossing the bus to a broker on another machine can arrive well after the rest. A runtime that said it is there and did not say what it serves, or diff --git a/node-tools/cmd/node-tools/main.go b/node-tools/cmd/node-tools/main.go index bf55106..7da745a 100644 --- a/node-tools/cmd/node-tools/main.go +++ b/node-tools/cmd/node-tools/main.go @@ -14,6 +14,7 @@ package main import ( "bytes" + "context" "encoding/json" "errors" "fmt" @@ -27,6 +28,7 @@ import ( "github.com/novox/mesh-tools/node-tools/internal/alive" "github.com/novox/mesh-tools/node-tools/internal/bus" "github.com/novox/mesh-tools/node-tools/internal/console" + "github.com/novox/mesh-tools/node-tools/internal/firstview" "github.com/novox/mesh-tools/node-tools/internal/launch" "github.com/novox/mesh-tools/node-tools/internal/provide" "github.com/novox/mesh-tools/node-tools/internal/runtime" @@ -48,6 +50,8 @@ func main() { until := time.Now().Add(startWait) cred := credential(until) conn := connectPatiently(cred, until) + // And the seat list it describes the seats by, read through the seat mesh-catalog (novox/hq ADR 0636 §4). + _ = startSeatList(conn) served, err := runtime.ServedModulesFrom(os.Getenv(runtime.ToolModules), cred.Module) if err != nil { @@ -86,12 +90,16 @@ func main() { } var up *console.Listening if listen != "" { + // The indexer's and the assigner's states the console indexes the mesh by come to it as first views, asked of the owners' + // own tools and followed on their views families, never read from their state (novox/hq ADR 0615 §1). + source, stopViews := firstview.Start(context.Background(), conn, log.Printf) + defer stopViews() node := cred.Node if node == "" { node = "?" } who := node + "." + cred.Module - up, err = console.Serve(console.NewSurface(conn, who), listen) + up, err = console.Serve(console.NewSurface(conn, source, who), listen) if err != nil { log.Fatalf("node-tools: %v", err) } diff --git a/node-tools/cmd/node-tools/seatlist.go b/node-tools/cmd/node-tools/seatlist.go new file mode 100644 index 0000000..5d93ec3 --- /dev/null +++ b/node-tools/cmd/node-tools/seatlist.go @@ -0,0 +1,73 @@ +package main + +import ( + "encoding/json" + "fmt" + "log" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/seatdef" + "git.novox.be/novox/mesh-sdk/go/seatlist" + "github.com/novox/mesh-tools/node-tools/internal/bus" +) + +// The tool runner reads the seat list as every part does (novox/hq ADR 0635 §1, ADR 0636): through the seat mesh-catalog, +// from the bootstrapped definitions until it has been read, and following the seat's events. What it holds is given to +// catalogue at every change; the console describes the seats from it, so the descriptions are the mesh's own and not a +// list compiled into the runner. + +// busOfTheRunner is the reader's bus over the runner's connection: a verb of the seat asked on its subject, an event heard +// on the seat's event subject. +type busOfTheRunner struct{ conn *bus.Conn } + +func (b busOfTheRunner) Ask(key string, body any) (json.RawMessage, error) { + seat, verb, _ := strings.Cut(strings.TrimPrefix(key, "seat:"), ".") + answered, err := b.conn.Ask(key, body, "mesh.seat."+seat+".tool."+verb) + if err != nil { + return nil, err + } + return answered.Result, nil +} + +func (b busOfTheRunner) Subscribe(pattern string, handle func(string, json.RawMessage) error) error { + seat, event, _ := strings.Cut(pattern, ".") + _, err := b.conn.Raw(seatdef.EventSubject(seat, event), func(_ string, data []byte) []byte { + _ = handle(pattern, data) + return nil // an event is answered by nobody + }) + return err +} + +// seatListWithin is how long the runner waits for the first read of the seat list before it serves; the read goes on +// after it, and the descriptions are the bootstrapped ones until it ends (ADR 0636 §1). +const seatListWithin = 5 * time.Second + +// startSeatList starts the runner's reader and gives catalogue its bootstrapped seats at once, then waits for the first +// read to end (the reader's Ready) for the bound, so the runner's first descriptions are of the list in the common case. +func startSeatList(conn *bus.Conn) *seatlist.Reader { + timer := time.NewTimer(seatListWithin) + defer timer.Stop() + return startSeatListOn(busOfTheRunner{conn}, timer.C, log.Printf) +} + +// startSeatListOn is startSeatList over any bus, with the bound as a channel the caller gives. +func startSeatListOn(b seatlist.Bus, bound <-chan time.Time, logf func(string, ...any)) *seatlist.Reader { + reader, err := seatlist.Start(seatlist.Config{Bus: b, Bootstrap: seatlist.Bootstrapped(), + OnChange: func(seats []catalogue.Seat) { catalogue.UseSeats(seats) }}) + if err != nil { + logf("node-tools: %v", fmt.Errorf("the tool runner cannot follow the seat list: %w", err)) + } + if reader == nil { + return nil + } + catalogue.UseSeats(reader.Seats()) + select { + case <-reader.Ready(): + catalogue.UseSeats(reader.Seats()) + case <-bound: + logf("node-tools: the seat list is not read yet; the runner serves with the bootstrapped seats and reads on") + } + return reader +} diff --git a/node-tools/cmd/node-tools/seatlist_test.go b/node-tools/cmd/node-tools/seatlist_test.go new file mode 100644 index 0000000..fda2420 --- /dev/null +++ b/node-tools/cmd/node-tools/seatlist_test.go @@ -0,0 +1,69 @@ +package main + +import ( + "encoding/json" + "errors" + "strings" + "sync/atomic" + "testing" + "time" +) + +// busOfAStandIn is a seat mesh-catalog that answers after release, or never (a refusal). +type busOfAStandIn struct { + release chan struct{} + asked atomic.Int32 +} + +func (b *busOfAStandIn) Ask(string, any) (json.RawMessage, error) { + b.asked.Add(1) + <-b.release + return nil, errors.New("the seat does not answer") +} + +func (b *busOfAStandIn) Subscribe(string, func(string, json.RawMessage) error) error { return nil } + +// ADR 0636 §1: the runner waits for the first read of the seat list on the reader's Ready, but only to a bound that the +// test gives as a channel; past it, it serves with the bootstrapped seats and says so, and nothing sleeps. +func TestTheRunnerWaitsForTheFirstReadOfTheSeatListOnlyToItsBound(t *testing.T) { + for _, row := range []struct { + name string + release bool // the seat ends its first read (here by refusing) before the bound + wantSaid string + wantReady bool + }{ + {"the first read ends before the bound", true, "", true}, + {"the bound passes with the seat silent", false, "the seat list is not read yet", false}, + } { + t.Run(row.name, func(t *testing.T) { + b := &busOfAStandIn{release: make(chan struct{})} + bound := make(chan time.Time, 1) + var said []string + if row.release { + close(b.release) + } else { + bound <- time.Time{} + } + r := startSeatListOn(b, bound, func(f string, a ...any) { said = append(said, f) }) + if r == nil { + t.Fatal("no reader") + } + ended := false + select { + case <-r.Ready(): + ended = true + default: + } + if ended != row.wantReady { + t.Errorf("first read ended = %v, want %v", ended, row.wantReady) + } + if got := strings.Join(said, ";"); !strings.Contains(got, row.wantSaid) || (row.wantSaid == "" && got != "") { + t.Errorf("said %q, want %q", got, row.wantSaid) + } + if !row.release { + close(b.release) + <-r.Ready() + } + }) + } +} diff --git a/node-tools/go.mod b/node-tools/go.mod index d88a8a1..54e4d8b 100644 --- a/node-tools/go.mod +++ b/node-tools/go.mod @@ -3,20 +3,27 @@ module github.com/novox/mesh-tools/node-tools go 1.26.0 require ( + git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 github.com/nats-io/nats.go v1.54.0 golang.org/x/sys v0.48.0 ) require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 - github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 + github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op // indirect github.com/google/go-tpm v0.9.8 // indirect - github.com/klauspost/compress v1.20.0 // indirect + github.com/klauspost/compress v1.20.1 // indirect github.com/minio/highwayhash v1.0.4 // indirect - github.com/nats-io/jwt/v2 v2.8.1 // indirect + github.com/nats-io/jwt/v2 v2.8.2 // indirect github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/crypto v0.57.0 // indirect - golang.org/x/time v0.15.0 // indirect + golang.org/x/time v0.16.0 // indirect ) diff --git a/node-tools/go.sum b/node-tools/go.sum index 7679a5d..8ca93df 100644 --- a/node-tools/go.sum +++ b/node-tools/go.sum @@ -1,15 +1,13 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 h1:XEbIVImp6yGjA91rp0iZBoDdL5KfamjOuDLwWvqh0hc= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= -github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= +github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E= +github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY= -github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= -github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ= +github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4= github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ= -github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU= -github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg= +github.com/nats-io/jwt/v2 v2.8.2 h1:XXRgB60MSTnqsRwejQurVDs/hcv2dkt+86GjI+I/bMc= +github.com/nats-io/jwt/v2 v2.8.2/go.mod h1:Ag/56sq9OblL4JgdYufDd16Egb17Kr/8WwwuO/forVc= github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I= github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24= github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= @@ -23,5 +21,5 @@ golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZ golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= -golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= -golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE= +golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s= diff --git a/node-tools/internal/announce/announce.go b/node-tools/internal/announce/announce.go index 6c941ea..8771120 100644 --- a/node-tools/internal/announce/announce.go +++ b/node-tools/internal/announce/announce.go @@ -43,6 +43,11 @@ var Patience = 5 * time.Second // to the laptop, and the hundred-byte PING answers queued behind it. const Digest = "digest" +// Capabilities is the service metadata key a runtime states what it can do under, a comma-separated list of capabilities +// (runtime.CallCapabilities): its machine's node-engine asks the runtime's PING and takes them into the node's record, so a +// module needing one is placed only where the runtime states it. +const Capabilities = "capabilities" + // Shortened is the service metadata key a runtime sets when its announcement was cut to fit the bus. const Shortened = "shortened" diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 23f0cf2..ec1648b 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -18,12 +18,12 @@ import ( "encoding/json" "errors" "fmt" - "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/membership" "log" "regexp" "strings" "sync" + "sync/atomic" "time" "github.com/nats-io/nats.go" @@ -354,7 +354,7 @@ func (c *Conn) Credential() Credential { return c.cred } // Follow reads a module's membership on this machine once and follows it live, so its tools are // served where the mesh issued them (ADR 0175). It is read from the grants part's state memberships, its key -// `.` (novox/hq ADR 0571 clause 7 and 10, mesh-sdk buscompose.GrantsMembershipsBucket), and nowhere else: +// `.` (novox/hq ADR 0571 clause 7 and 10, mesh-sdk membership.GrantsBucket), and nowhere else: // the old assignments stream is not read and its subject is not heard. func (c *Conn) Follow(module string) { c.mu.Lock() @@ -433,10 +433,10 @@ func (c *Conn) watchGrants(module string, say bool) bool { c.Logf(format, a...) } } - key := buscompose.MembershipKey(c.node, module) + key := membership.GrantsKey(c.node, module) // First the key's direct get, which the account is granted alone: a bus that refuses it, or a state not there yet, // answers at once, where binding the bucket would wait out its bound on a refusal. - bucket := buscompose.GrantsMembershipsBucket + bucket := membership.GrantsBucket probe, err := c.nc.Request("$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+"."+key, nil, GrantsPatience) if err != nil || probe.Header.Get("Status") == "503" { why := "it did not answer" @@ -456,7 +456,7 @@ func (c *Conn) watchGrants(module string, say bool) bool { kv, err := js.KeyValue(ctx, bucket) if err != nil { logf("[mesh-tools] the grants part's memberships (%s) cannot be read (%v); %s has no membership "+ - "until it can be", buscompose.GrantsMembershipsBucket, err, module) + "until it can be", membership.GrantsBucket, err, module) return false } w, err := kv.Watch(context.Background(), key) @@ -1077,21 +1077,109 @@ var EventHeartbeat = 10 * time.Second // returns an error, terminated when it is not an event at all. The consumers are the mesh's to // create; this binds to them and never makes one. Runs until stopped. func (c *Conn) ConsumeAs(module string, deliver func(Envelope) error) (func(), error) { + return c.ConsumeAsBeating(module, deliver, nil) +} + +// IdleHeartbeat is how often the runtime tells a bundle that its module's consumer is bound and the bus answers, +// while no event arrives (stdio.OnHeartbeat), so a subscription's silence is told from a quiet source. +var IdleHeartbeat = 15 * time.Second + +// Beat is the runtime's word on a module's consumer between events: it is bound and the bus answers (At), or the +// bus's heartbeat stopped arriving or the bus ended the loop (Missed, with Why). Its JSON is the stdio Heartbeat. +type Beat struct { + At time.Time `json:"at"` + Missed bool `json:"missed,omitempty"` + Why string `json:"why,omitempty"` +} + +// ConsumeAsBeating is ConsumeAs that also says, to beat, how the module's consumer on EVENTS stands: a Beat at each +// IdleHeartbeat in which no event arrived and the bus answered about the consumer, and one Missed (said once) when +// the bus's own heartbeat to the loop stopped arriving, the bus ended the loop or no longer answers. A Beat that is +// not Missed follows the first answer after one that was. A nil beat says nothing. +func (c *Conn) ConsumeAsBeating(module string, deliver func(Envelope) error, beat func(Beat)) (func(), error) { + return c.ConsumeAsBeatingWhen(module, deliver, beat, nil) +} + +// tickHook, when set before a consume starts, is called after each idle heartbeat interval was handled (tests only, +// through SetTickHook), so a test counts beats instead of sleeping. A consume reads it once, when it starts. +var tickHook func() + +// ConsumeAsBeatingWhen is ConsumeAsBeating that asks the bus about the consumer, and says anything, only while wanted +// answers true (nil is always): a module whose children asked for no heartbeat costs the bus no consumer-info request. +func (c *Conn) ConsumeAsBeatingWhen(module string, deliver func(Envelope) error, beat func(Beat), wanted func() bool) (func(), error) { + asked := func() bool { return beat != nil && (wanted == nil || wanted()) } durable := ConsumerOf(c.node, module) js, err := jetstream.New(c.nc) if err != nil { return nil, err } - h := newHearing(c, js, module, deliver) + var lastEvent atomic.Int64 + var missed atomic.Bool + say := func(why string) { + if asked() && missed.CompareAndSwap(false, true) { + beat(Beat{At: time.Now().UTC(), Missed: true, Why: why}) + } + } + h := newHearing(c, js, module, func(env Envelope) error { + lastEvent.Store(time.Now().UnixNano()) + missed.Store(false) // an event arrived: the consumer is heard, whatever was said before + return deliver(env) + }) // Bound by name, never created and never matched against a subject: the consumer and its // filters are the mesh's (design 29 §3), exactly as the module's own runtime bound it. - events, _, doing, err := h.bind(EventsStream, durable, nil) + events, _, doing, err := h.bind(EventsStream, durable, func(err error) { + if errors.Is(err, jetstream.ErrNoHeartbeat) { + say("the bus's heartbeat to the consumer " + durable + " stopped arriving") + } + }) if err != nil { return nil, fmt.Errorf("%s %s's consumer %s: %w", doing, module, durable, err) } // And every consumer on a module's stream its membership names (novox/hq ADR 0363 §8), followed as the // membership changes. A membership naming none binds nothing more: the module hears EVENTS alone. h.start() + if beat != nil { + every := IdleHeartbeat + hook := tickHook + tickOnce := func() { + select { + case <-events.Closed(): + say("the bus ended reading the consumer " + durable) + return + default: + } + if !asked() { + return + } + ctx, cancel := context.WithTimeout(context.Background(), every) + _, err := js.Consumer(ctx, EventsStream, durable) + cancel() + if err != nil { + say("the bus does not answer about the consumer " + durable + ": " + err.Error()) + return + } + if time.Since(time.Unix(0, lastEvent.Load())) < every { + return // an event arrived in this interval: it was the word + } + missed.Store(false) + beat(Beat{At: time.Now().UTC()}) + } + go func() { + tick := time.NewTicker(every) + defer tick.Stop() + for { + select { + case <-h.done: + return + case <-tick.C: + } + tickOnce() + if hook != nil { + hook() + } + } + }() + } var once sync.Once return func() { once.Do(func() { diff --git a/node-tools/internal/bus/caller_bus_test.go b/node-tools/internal/bus/caller_bus_test.go index 502a3dd..c46049d 100644 --- a/node-tools/internal/bus/caller_bus_test.go +++ b/node-tools/internal/bus/caller_bus_test.go @@ -16,7 +16,7 @@ import ( mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) -// A bus whose users are granted as the controller grants them (mesh-placement internal/broker): each named +// A bus whose users are granted as the controller grants them (mesh-module-assigner internal/broker): each named // user publishes the subjects listed for it. func grantedBus(t *testing.T, publish map[string][]string) string { t.Helper() diff --git a/node-tools/internal/bus/consume_test.go b/node-tools/internal/bus/consume_test.go index 4c29d44..e5cc2e5 100644 --- a/node-tools/internal/bus/consume_test.go +++ b/node-tools/internal/bus/consume_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "sync" + "sync/atomic" "testing" "time" @@ -60,3 +61,96 @@ func TestASlowHandlerAndABurstAreEachHandedOverOnce(t *testing.T) { } } } + +// The runtime tells a bundle how its module's consumer stands between events (stdio.OnHeartbeat): a word at each +// idle heartbeat while the bus answers, once Missed when the bus ended the loop, and a word again when it is bound again. +func TestAConsumerSaysItIsBoundWhileIdleAndOnceWhenTheBusEndedIt(t *testing.T) { + mesh := mt.New(t) + restore := bus.IdleHeartbeat + bus.IdleHeartbeat = 200 * time.Millisecond + t.Cleanup(func() { bus.IdleHeartbeat = restore }) + ticks := make(chan struct{}, 64) + t.Cleanup(bus.SetTickHook(func() { + select { + case ticks <- struct{}{}: + default: + } + })) + mesh.Consumer(t, "anchor", "quiet", []string{"mesh.mod.alpha.event.>"}, time.Second) + c, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + c.Logf = func(string, ...any) {} + t.Cleanup(c.Close) + + var mu sync.Mutex + var beats []bus.Beat + stop, err := c.ConsumeAsBeating("quiet", func(bus.Envelope) error { return nil }, func(b bus.Beat) { + mu.Lock() + beats = append(beats, b) + mu.Unlock() + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(stop) + // A consume nobody asked a heartbeat of says nothing and asks the bus nothing, however many intervals pass. + var unaskedBeats atomic.Int32 + unasked, err := c.ConsumeAsBeatingWhen("quiet", func(bus.Envelope) error { return nil }, + func(bus.Beat) { unaskedBeats.Add(1) }, func() bool { return false }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(unasked) + last := func() (bus.Beat, int) { + mu.Lock() + defer mu.Unlock() + if len(beats) == 0 { + return bus.Beat{}, 0 + } + return beats[len(beats)-1], len(beats) + } + mt.Until(t, func() error { + if b, n := last(); n == 0 || b.Missed || b.At.IsZero() { + return fmt.Errorf("no word that the consumer is bound yet: %d beats", n) + } + return nil + }) + mesh.RemoveConsumer(t, bus.EventsStream, "anchor", "quiet") + mt.Until(t, func() error { + if b, _ := last(); !b.Missed || b.Why == "" { + return fmt.Errorf("the consumer is gone from the bus and no Missed word came: %+v", b) + } + return nil + }) + mu.Lock() + missed := 0 + for _, b := range beats { + if b.Missed { + missed++ + } + } + mu.Unlock() + for n := 0; n < 3; n++ { // three more intervals pass, counted by the loop's own tick + select { + case <-ticks: + case <-time.After(10 * time.Second): + t.Fatal("the heartbeat loop handled no more intervals") + } + } + mu.Lock() + defer mu.Unlock() + after := 0 + for _, b := range beats { + if b.Missed { + after++ + } + } + if unaskedBeats.Load() != 0 { + t.Errorf("a consume with no heartbeat asked said %d words", unaskedBeats.Load()) + } + if after != missed || missed != 1 { + t.Errorf("Missed is said once, not at every beat: %d then %d", missed, after) + } +} diff --git a/node-tools/internal/bus/export_test.go b/node-tools/internal/bus/export_test.go new file mode 100644 index 0000000..92420cd --- /dev/null +++ b/node-tools/internal/bus/export_test.go @@ -0,0 +1,8 @@ +package bus + +// SetTickHook makes the next consumes call f after each idle heartbeat interval was handled; the restore puts it back. +func SetTickHook(f func()) (restore func()) { + old := tickHook + tickHook = f + return func() { tickHook = old } +} diff --git a/node-tools/internal/bus/per_node.go b/node-tools/internal/bus/per_node.go index ba64d35..dae0580 100644 --- a/node-tools/internal/bus/per_node.go +++ b/node-tools/internal/bus/per_node.go @@ -33,7 +33,7 @@ func (c *Conn) perNodeGet(s StateIssued, key string) (*StateEntry, error) { if err != nil || e == nil { return nil, err } - return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision()}, nil + return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision(), At: e.Created().UTC()}, nil } func (c *Conn) perNodeKeys(s StateIssued) ([]string, error) { diff --git a/node-tools/internal/bus/seat.go b/node-tools/internal/bus/seat.go index 5a84212..1ef7fc1 100644 --- a/node-tools/internal/bus/seat.go +++ b/node-tools/internal/bus/seat.go @@ -370,7 +370,13 @@ func (c *Conn) SeatRecord(module, bucket, key string) (*StateEntry, error) { } var revision uint64 fmt.Sscan(reply.Header.Get("Nats-Sequence"), &revision) - return &StateEntry{Key: module + "." + key, Value: valueOf(reply.Data), Revision: revision}, nil + var at time.Time + if stamp := reply.Header.Get("Nats-Time-Stamp"); stamp != "" { + if t, err := time.Parse(time.RFC3339Nano, stamp); err == nil { + at = t.UTC() + } + } + return &StateEntry{Key: module + "." + key, Value: valueOf(reply.Data), Revision: revision, At: at}, nil } // StateCreate writes one key only when it has no value, and answers its revision; a key that has one is diff --git a/node-tools/internal/bus/seat_test.go b/node-tools/internal/bus/seat_test.go index fe49a19..c8dbd0e 100644 --- a/node-tools/internal/bus/seat_test.go +++ b/node-tools/internal/bus/seat_test.go @@ -98,7 +98,7 @@ func TestAModuleSubmitsAndSaysOnlyUnderItsOwnNameAndKind(t *testing.T) { t.Fatalf("an asker could not ask under its own name: %v", err) } for module, subject := range map[string]string{ - "mesh-delivery": "mesh.seat.operator-channel.accept.ask.mesh-placement", + "mesh-delivery": "mesh.seat.operator-channel.accept.ask.mesh-module-assigner", "telegram": "mesh.seat.intake.event.choice.desktop", "messenger": "mesh.seat.intake.event.choice.telegram", } { @@ -206,13 +206,16 @@ func TestAnAskerReadsItsOwnRecordAndNoOther(t *testing.T) { if _, err := kv.Put("mesh-delivery.a1", []byte(`{"state":"open"}`)); err != nil { t.Fatal(err) } - if _, err := kv.Put("mesh-placement.a1", []byte(`{"state":"open"}`)); err != nil { + if _, err := kv.Put("mesh-module-assigner.a1", []byte(`{"state":"open"}`)); err != nil { t.Fatal(err) } got, err := conn.SeatRecord("mesh-delivery", "messenger_asks", "a1") if err != nil || got == nil || string(got.Value) != `{"state":"open"}` { t.Fatalf("the asker could not read its own ask: %v %v", got, err) } + if got.At.IsZero() || time.Since(got.At) > time.Minute { + t.Errorf("the record does not say when the bus wrote it: at %v", got.At) + } if missing, err := conn.SeatRecord("mesh-delivery", "messenger_asks", "a2"); err != nil || missing != nil { t.Fatalf("a missing ask read as %v, %v", missing, err) } diff --git a/node-tools/internal/bus/state.go b/node-tools/internal/bus/state.go index 935ddff..c3cade3 100644 --- a/node-tools/internal/bus/state.go +++ b/node-tools/internal/bus/state.go @@ -39,6 +39,8 @@ type StateEntry struct { Key string `json:"key"` Value json.RawMessage `json:"value"` Revision uint64 `json:"revision"` + // At is when the bus wrote this value (the entry's created time), as stdio.StateEntry.At reads it. + At time.Time `json:"at,omitzero"` } // StateChange is one change a watch delivers: a key put, deleted, or expired (its own time to live, or the @@ -134,7 +136,7 @@ func (c *Conn) StateGet(module, name, key string) (*StateEntry, error) { if err != nil { return nil, err } - return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision()}, nil + return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision(), At: e.Created().UTC()}, nil } // StateKept is one value a bucket still keeps for a key: a put, or a delete (a purge reads as one). @@ -543,7 +545,7 @@ var credentialEndings = []string{"password", "passwd", "secret", "token", "crede // credentialField is the first field anywhere in a JSON value whose name says it is a credential, or "". The one shape a // credential-shaped name may hold is a provider reference, an object of exactly `node` and `module`, both strings: a -// placement's binding of the provision "secret" to the vault (issue 1126) names who provides it, never its value. Any +// the assignment's binding of the provision "secret" to the vault (issue 1126) names who provides it, never its value. Any // other shape under such a name is refused. func credentialField(value json.RawMessage) string { var v any diff --git a/node-tools/internal/bus/state_pernode_test.go b/node-tools/internal/bus/state_pernode_test.go index de04809..82baf52 100644 --- a/node-tools/internal/bus/state_pernode_test.go +++ b/node-tools/internal/bus/state_pernode_test.go @@ -10,6 +10,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/novox/mesh-tools/node-tools/internal/bus" mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" @@ -47,7 +48,7 @@ func TestAPerNodeStateIsServedAcrossEveryNodesBucket(t *testing.T) { return nil }) - if e, err := conn.StateGet("mesh-provisioner", "mesh-host.composed", "b.m2.f2"); err != nil || e == nil || e.Key != "b.m2.f2" { + if e, err := conn.StateGet("mesh-provisioner", "mesh-host.composed", "b.m2.f2"); err != nil || e == nil || e.Key != "b.m2.f2" || e.At.IsZero() || time.Since(e.At) > time.Minute { t.Fatalf("get from the second node's bucket: %+v, %v", e, err) } if e, err := conn.StateGet("mesh-provisioner", "mesh-host.composed", "q.m.f"); err != nil || e != nil { diff --git a/node-tools/internal/bus/state_test.go b/node-tools/internal/bus/state_test.go index 6ce8393..243d017 100644 --- a/node-tools/internal/bus/state_test.go +++ b/node-tools/internal/bus/state_test.go @@ -15,7 +15,7 @@ func TestACredentialNamedFieldIsFoundAnywhereInAValue(t *testing.T) { `{"clientSecret":"s"}`: "clientSecret", `{"maxTokens":4096,"name":"a","generation":3}`: "", `"just a string"`: "", - // Issue 1126: a placement binds the provision "secret" to the vault; only an exact provider reference passes. + // Issue 1126: an assignment binds the provision "secret" to the vault; only an exact provider reference passes. `{"bindings":{"secret":{"node":"anchor","module":"mesh-vault"}}}`: "", `{"secret":{"value":"s"}}`: "secret", `{"secret":{"node":"anchor","module":"mesh-vault","value":"s"}}`: "secret", diff --git a/node-tools/internal/bus/state_watch_test.go b/node-tools/internal/bus/state_watch_test.go index cb69445..42b16da 100644 --- a/node-tools/internal/bus/state_watch_test.go +++ b/node-tools/internal/bus/state_watch_test.go @@ -33,7 +33,7 @@ func TestAWatchHoldsWhileItsCurrentValuesKeepArriving(t *testing.T) { for i := 0; i < older; i++ { keys = append(keys, fmt.Sprintf("novox/mesh-catalog.%d.%040x.status=mesh/check", 400+i, i)) } - for _, member := range []string{"novox/mesh-sdk.76.34f210e5", "novox/mesh-placement.453.c06f75d0", "novox/mesh-delivery.33.28ee22ef"} { + for _, member := range []string{"novox/mesh-sdk.76.34f210e5", "novox/mesh-module-assigner.453.c06f75d0", "novox/mesh-delivery.33.28ee22ef"} { keys = append(keys, member+strings.Repeat("0", 32)+".status=mesh/check") } for _, k := range keys { @@ -151,7 +151,7 @@ func TestAMembershipTheGrantsPartWroteWrappedIsServed(t *testing.T) { if _, err := conn.StatePut("gitea", "repos", "novox.hq", []byte(`{"v":1}`)); err != nil { t.Fatalf("the state the membership issues is refused: %v", err) } - if e, err := conn.StateGet("gitea", "repos", "novox.hq"); err != nil || e == nil || string(e.Value) != `{"v":1}` { + if e, err := conn.StateGet("gitea", "repos", "novox.hq"); err != nil || e == nil || string(e.Value) != `{"v":1}` || e.At.IsZero() || time.Since(e.At) > time.Minute { t.Errorf("the state reads back %+v, %v", e, err) } conn.Follow("pages") diff --git a/node-tools/internal/console/address.go b/node-tools/internal/console/address.go index e3f8aaf..7b884dd 100644 --- a/node-tools/internal/console/address.go +++ b/node-tools/internal/console/address.go @@ -17,8 +17,10 @@ package console // session — so a tool that arrived a minute ago is found without the client reconnecting. import ( + assigner "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" + "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" + meshcatalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-sdk/go/catalogue" - "git.novox.be/novox/mesh-sdk/go/delivery" "context" "encoding/json" @@ -30,6 +32,7 @@ import ( "github.com/novox/mesh-tools/node-tools/internal/announce" "github.com/novox/mesh-tools/node-tools/internal/bus" + "github.com/novox/mesh-tools/node-tools/internal/firstview" ) // IndexKept is how long what the mesh answered is kept before it is asked again: long enough that @@ -38,19 +41,21 @@ var IndexKept = 5 * time.Second // The seven tools the console announces. Names of the API's kind — letters, digits, `_`, `-`. const ( - verbOverview = "mesh_overview" - verbMachine = "mesh_machine" - verbSearch = "mesh_search" - verbDescribe = "mesh_describe" - verbCall = "mesh_call" - verbRuntimes = "mesh_runtimes" - verbWait = "mesh_wait" + verbOverview = "mesh_overview" + verbMachine = "mesh_machine" + verbSearch = "mesh_search" + verbDescribe = "mesh_describe" + verbCall = "mesh_call" + verbRuntimes = "mesh_runtimes" + verbFirstViews = "mesh_first_views" + verbFirstAgain = "mesh_first_view_again" + verbWait = "mesh_wait" ) // searchCap is how many matches a search answers before it says how many more there were. const searchCap = 25 -const grammar = "Addresses: `.` for a seat held once for the mesh (e.g. `mesh-placement.placement`); " + +const grammar = "Addresses: `.` for a seat held once for the mesh (e.g. `mesh-module-assigner.assignments`); " + "`/.` for a seat every machine holds (e.g. `ace/node-packet-filter.rules`); " + "`/.` for a module on one machine (e.g. `novox/postgres.postgres_list_databases`); " + "and `.` also for a module whose instances are interchangeable." @@ -89,13 +94,22 @@ func discovery() []map[string]any { "arrive, its size in bytes, how many modules and tools it announced, whether it was shortened to fit the " + "bus, and when it was last heard — and every runtime or machine expected that was not heard. Read-only; " + "for when an address is said to be missing or on another machine."}, + {"name": verbFirstViews, "inputSchema": obj(map[string]any{}), + "description": "The tool runner's first views of the four states it reads of the indexer and the assigner (the " + + "assigner's assignments and seat-holders, the indexer's manifests and modules): for each, the owner, the tool " + + "asked, the revision held and when it was made, how many keys, when it was asked and answered, how often " + + "it was asked and how often by hand, whether its condition is raised, and why it was last asked. Read-only."}, + {"name": verbFirstAgain, "inputSchema": obj(map[string]any{ + "why": map[string]any{"type": "string", "description": "why it is asked again, kept in the log"}, + }), "description": "Ask the tool runner's four first views again, now, and say what each answers. A hand act, recorded in " + + "the runner's log with its reason; for when a picture is doubted, or after an owner's state was made again."}, waitTool(), } } func isDiscovery(name string) bool { switch name { - case verbOverview, verbMachine, verbSearch, verbDescribe, verbCall, verbRuntimes, verbWait: + case verbOverview, verbMachine, verbSearch, verbDescribe, verbCall, verbRuntimes, verbFirstViews, verbFirstAgain, verbWait: return true } return false @@ -131,14 +145,14 @@ type index struct { NotAnswering []string Listing *Listing // the flat catalogue the call path resolves subjects with Discovery announce.Discovery - // Recorded is where placement places each module that declares tools. + // Recorded is where the assigner places each module that declares tools. Recorded map[string][]string // Unheard is every runtime whose answer this index lacks: it said it was there and did not say // what it serves in time, or it answered before and not now. An address it might answer is never // called missing while it is here (2026-10-05). Unheard []unheard - // Placed is the state seat-holders as the seat mesh-placement's holder wrote it (novox/hq ADR 0571 clause 4), by its - // keys `` and `.`, read with delivery.HoldersOf; PlacedWhy says why it could not be read. + // Placed is the state seat-holders as the seat mesh-module-assigner's holder wrote it (novox/hq ADR 0571 clause 4), by its + // keys `` and `.`, read with assigner.HoldersOf; PlacedWhy says why it could not be read. Placed map[string][]byte PlacedWhy string // Replaces is what each verb and tool replaces (novox/hq ADR 0245): @@ -231,7 +245,7 @@ func findTool(tools []Tool, name string) *Tool { return nil } -// recordedModule is a module as the bus holds it: where the placement seat places it, and whether its manifest declares +// recordedModule is a module as the bus holds it: where the assigner seat places it, and whether its manifest declares // tools — what should announce itself, and where (novox/hq ADR 0571). type recordedModule struct { Module string `json:"module"` @@ -248,7 +262,7 @@ type recordedModule struct { // `held` is what the console holds already, so an instance whose announcement is unchanged is not // asked for it again; `before` is the last answer of every runtime this round did not hear, which the // index holds instead, marked with its age (novox/hq issue 554). -func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery) []earlier) (*index, error) { +func indexOn(conn *bus.Conn, src Source, held announce.Known, before func(announce.Discovery) []earlier) (*index, error) { var wg sync.WaitGroup var placed map[string][]byte var placedErr error @@ -257,10 +271,10 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery var recordsWhy []string wg.Add(2) // Which modules there are and where they are placed, and which machines the mesh has, from the bus (ADR 0571 - // clauses 4 and 5): the indexer's modules and manifests and the placement seat's placements, never any other record. - go func() { defer wg.Done(); recorded, known, recordsWhy = recordsOn(conn) }() - // Who holds which seat on which node, from placement (ADR 0571 clause 4): no other call. - go func() { defer wg.Done(); placed, placedErr = seatHoldersOn(conn) }() + // clauses 4 and 5): the indexer's modules and manifests and the assigner seat's assignments, never any other record. + go func() { defer wg.Done(); recorded, known, recordsWhy = recordsOn(src) }() + // Who holds which seat on which node, from the assigner (ADR 0571 clause 4): no other call. + go func() { defer wg.Done(); placed, placedErr = seatHoldersOn(src) }() d, err := announce.Gather(conn, held) wg.Wait() if err != nil { @@ -310,7 +324,7 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery // What should have answered: every assignment of a module that declares tools. Silence is named; // a module with no tools is never a name here. - x.Replaces = replacesOf(catalogue.DefaultSeats(), recorded) + x.Replaces = replacesOf(catalogue.Seats(), recorded) for _, m := range recorded { if !m.Tools { continue @@ -326,7 +340,7 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery } l.NotAnswering = append(l.NotAnswering, recordsWhy...) if x.PlacedWhy != "" { - l.NotAnswering = append(l.NotAnswering, delivery.SeatHoldersRead+" (who holds which seat by placement could not be read: "+ + l.NotAnswering = append(l.NotAnswering, firstview.SeatHoldersRead+" (who holds which seat by assignment could not be read: "+ x.PlacedWhy+")") } for _, u := range x.Unheard { @@ -345,37 +359,26 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery return x, nil } -// stateOn reads a state the runtime's membership reads, whole: its current values by key, a deleted key absent. -func stateOn(conn *bus.Conn, name string) (map[string][]byte, error) { - var mu sync.Mutex - keys := map[string][]byte{} - stop, err := conn.StateWatch(conn.Credential().Module, name, "", func(c bus.StateChange) error { - if c.Current && c.Op == "put" { - mu.Lock() - keys[c.Key] = c.Value - mu.Unlock() - } - return nil - }) - if err != nil { - return nil, err - } - stop() - mu.Lock() - defer mu.Unlock() - return keys, nil +// Source is where the console reads the four states of the indexer and the assigner it indexes the mesh by: the assigner's +// assignments and seat-holders, the indexer's modules and manifests, by `.` (firstview.AssignmentsRead and the +// rest). The tool runner's is its first views (firstview.Views), never the owners' state (novox/hq ADR 0615 §1). +type Source interface { + // State is every current value of a state by key; an error says why it cannot be told yet. + State(name string) (map[string][]byte, error) + // Entry is one value of a state, nil when it holds none. + Entry(name, key string) ([]byte, error) } -// seatHoldersOn reads the state seat-holders whole (novox/hq ADR 0571 clause 4), as delivery.SeatHoldersRead. -func seatHoldersOn(conn *bus.Conn) (map[string][]byte, error) { - return stateOn(conn, delivery.SeatHoldersRead) +// seatHoldersOn reads the state seat-holders whole (novox/hq ADR 0571 clause 4), as firstview.SeatHoldersRead. +func seatHoldersOn(src Source) (map[string][]byte, error) { + return src.State(firstview.SeatHoldersRead) } // The indexer's states the console reads (novox/hq ADR 0571 clauses 4 and 5): every module, and each manifest by its // digest, by the names a reader's manifest declares. var ( - modulesRead = delivery.IndexerModule + ".modules" - manifestsRead = delivery.IndexerModule + ".manifests" + modulesRead = firstview.ModulesRead + manifestsRead = firstview.ManifestsRead ) // manifestFacts is what the console takes of a manifest: the tools it declares and what each replaces. A manifest's @@ -387,39 +390,39 @@ type manifestFacts struct { var manifestsKept sync.Map // digest → manifestFacts -// recordsOn is every module the indexer holds, with where the placement seat places it and whether it declares tools, -// and every machine placement names (delivery.ModuleNames, delivery.MachineNames). What cannot be read is said, one line +// recordsOn is every module the indexer holds, with where the assigner seat places it and whether it declares tools, +// and every machine the assignments name (moduleNames, assigner.MachineNames). What cannot be read is said, one line // per state, for the listing's not answering. -func recordsOn(conn *bus.Conn) (recorded []recordedModule, machines []string, why []string) { - placements, err := stateOn(conn, delivery.PlacementRead) +func recordsOn(src Source) (recorded []recordedModule, machines []string, why []string) { + held, err := src.State(firstview.AssignmentsRead) if err != nil { - why = append(why, delivery.PlacementRead+" (where modules are placed could not be read, so what is missing cannot "+ + why = append(why, firstview.AssignmentsRead+" (where modules are placed could not be read, so what is missing cannot "+ "be said: "+err.Error()+")") } - machines = delivery.MachineNames(placements) + machines = assigner.MachineNames(held) on := map[string][]string{} - // The placement's keys in either form, a node key and its module keys or the form before (research 072 row 37). - assembled, err := delivery.PlacementsOf(placements) + // The assignments' keys in either form, a node key and its module keys or the form before (research 072 row 37). + assembled, err := assigner.AssignmentsOf(held) if err != nil { - why = append(why, delivery.PlacementRead+" (not every key is a placement: "+err.Error()+")") + why = append(why, firstview.AssignmentsRead+" (not every key is an assignment: "+err.Error()+")") } for _, node := range machines { for _, m := range assembled[node].Modules { on[m.Module] = append(on[m.Module], node) } } - modules, err := stateOn(conn, modulesRead) + modules, err := src.State(modulesRead) if err != nil { return nil, machines, append(why, modulesRead+" (which modules there are could not be read: "+err.Error()+")") } - names, err := delivery.ModuleNames(modules) + names, err := moduleNames(modules) if err != nil { return nil, machines, append(why, modulesRead+" ("+err.Error()+")") } for _, name := range names { r := recordedModule{Module: name, On: on[name]} - if m, err := delivery.DecodeModule(modules[name]); err == nil && m.Manifest != "" { - facts, err := manifestOf(conn, m.Manifest) + if m, err := meshcatalogclient.DecodeModule(modules[name]); err == nil && m.Manifest != "" { + facts, err := manifestOf(src, m.Manifest) if err != nil { why = append(why, name+" (its manifest could not be read, so whether it answers tools is not known: "+err.Error()+")") } else { @@ -431,44 +434,59 @@ func recordsOn(conn *bus.Conn) (recorded []recordedModule, machines []string, wh return recorded, machines, why } +// moduleNames is every module the indexer holds, sorted, from its state `modules` as a reader holds its keys (key to +// value; a deleted key absent). A key that is not a module's is refused, naming it. +func moduleNames(keys map[string][]byte) ([]string, error) { + out := make([]string, 0, len(keys)) + for k := range keys { + module, err := meshcatalogclient.ParseModulesKey(k) + if err != nil { + return nil, err + } + out = append(out, module) + } + sort.Strings(out) + return out, nil +} + // manifestOf is what the console takes of the manifest of a digest, read from the indexer's manifests once. -func manifestOf(conn *bus.Conn, digest string) (manifestFacts, error) { +func manifestOf(src Source, digest string) (manifestFacts, error) { if kept, ok := manifestsKept.Load(digest); ok { return kept.(manifestFacts), nil } - entry, err := conn.StateGet(conn.Credential().Module, manifestsRead, digest) + entry, err := src.Entry(manifestsRead, digest) if err != nil { return manifestFacts{}, err } if entry == nil { return manifestFacts{}, fmt.Errorf("the indexer holds no manifest %s", digest) } - m, err := delivery.DecodeManifest(entry.Value) + m, err := meshcatalogclient.DecodeManifest(entry) if err != nil { return manifestFacts{}, err } var facts manifestFacts - if err := json.Unmarshal([]byte(m.JSON), &facts); err != nil { + if err := json.Unmarshal([]byte(m.Manifest), &facts); err != nil { return manifestFacts{}, err } manifestsKept.Store(digest, facts) return facts, nil } -// placedHolders is who holds seat (on node, for a seat held per machine) by placement, or none. -func (x *index) placedHolders(seat, node string) []delivery.SeatHolder { +// placedHolders is who holds seat (on node, for a seat held per machine) by assignment, or none. +func (x *index) placedHolders(seat, node string) []meshmoduleassigner.SeatHoldersHoldersItem { if len(x.Placed) == 0 { return nil } - hs, err := delivery.HoldersOf(x.Placed, seat, node) + hs, err := assigner.HoldersOf(x.Placed, seat, node) if err != nil || len(hs) > 0 || node != "" { return hs } // A seat held per machine, asked without one: every node's key of it. - var all []delivery.SeatHolder + var all []meshmoduleassigner.SeatHoldersHoldersItem for key := range x.Placed { if s, n, ok := strings.Cut(key, "."); ok && s == seat { - if more, err := delivery.HoldersOf(x.Placed, seat, n); err == nil { + if more, err := assigner.HoldersOf(x.Placed, seat, n); err == nil { all = append(all, more...) } } @@ -501,7 +519,7 @@ func (s *Surface) indexAsked(fresh bool) (*index, error) { return x, nil } s.mu.Unlock() - x, err := indexOn(s.conn, s.held, s.before) + x, err := indexOn(s.conn, s.src, s.held, s.before) if err != nil { return nil, err } @@ -707,7 +725,7 @@ func resolve(x *index, address string) (target, error) { for _, h := range hs { held = append(held, h.Module+" on "+h.Node) } - return target{}, fmt.Errorf("the seat %s is held by %s by placement, but no holder of it announced itself on the bus. "+ + return target{}, fmt.Errorf("the seat %s is held by %s by assignment, but no holder of it announced itself on the bus. "+ "Ask again in a moment; mesh_runtimes shows which runtimes answered", prefix, strings.Join(held, ", ")) } if len(x.Unheard) > 0 { @@ -926,6 +944,9 @@ func (s *Surface) discover(ctx context.Context, name string, args map[string]any if name == verbRuntimes { return s.runtimesAnswer() } + if name == verbFirstViews || name == verbFirstAgain { + return s.firstViewsAnswer(ctx, name, args) + } if name == verbWait { return s.waitAnswer(ctx, args) } diff --git a/node-tools/internal/console/address_test.go b/node-tools/internal/console/address_test.go index fec01f4..d27fb1d 100644 --- a/node-tools/internal/console/address_test.go +++ b/node-tools/internal/console/address_test.go @@ -71,19 +71,19 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { } // The controller: its own seat's verbs announced on the bus like every runtime's; its records are asked by no one. - controller := connect(t, "mesh-placement", "bench") + controller := connect(t, "mesh-module-assigner", "bench") // Its verb nodes, called through the console as any seat's verb is. - stopNodes, err := controller.HandleSubject("mesh.seat.mesh-placement.tool.nodes", func(json.RawMessage) (any, error) { + stopNodes, err := controller.HandleSubject("mesh.seat.mesh-module-assigner.tool.nodes", func(json.RawMessage) (any, error) { return map[string]any{"ok": true, "output": `[{"name":"bench","mode":"converged"},{"name":"desk","mode":"converged"}]`}, nil }) if err != nil { t.Fatal(err) } t.Cleanup(stopNodes) - // What the mesh holds, as the indexer and placement write it (ADR 0571 clauses 4, 5): never the controller's records. + // What the mesh holds, as the indexer and the assigner write it (ADR 0571 clauses 4, 5): never the controller's records. tools := map[string][]string{"alpha": {"one", "two"}, "beta": {"three"}, "gamma": {"given"}, "delta": nil, - "epsilon": {"seven"}, "mesh-placement": {"nodes"}} - mesh.Placed(t, tools, map[string][]string{"desk": {"alpha", "beta", "delta"}, "bench": {"epsilon", "mesh-placement"}}) + "epsilon": {"seven"}, "mesh-module-assigner": {"nodes"}} + mesh.Placed(t, tools, map[string][]string{"desk": {"alpha", "beta", "delta"}, "bench": {"epsilon", "mesh-module-assigner"}}) console := mt.MembershipOf("node-tools", "desk", false, nil) console.State = mt.ConsoleReads() mesh.Issue(t, console) @@ -93,11 +93,11 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { } return nil }) - stopAnn, err := announce.Serve(controller, announce.Service{Name: "mesh-placement", ID: "bench"}, func() []announce.Endpoint { - return []announce.Endpoint{{Kind: announce.KindSeat, Module: "mesh-placement", Seat: "mesh-placement", Scope: "mesh", + stopAnn, err := announce.Serve(controller, announce.Service{Name: "mesh-module-assigner", ID: "bench"}, func() []announce.Endpoint { + return []announce.Endpoint{{Kind: announce.KindSeat, Module: "mesh-module-assigner", Seat: "mesh-module-assigner", Scope: "mesh", // As the live controller announces: a mesh seat's holder names no machine. Tool: "nodes", Node: "", Description: "Every machine the mesh knows.", Schema: json.RawMessage(`{}`), - Subject: "mesh.seat.mesh-placement.tool.nodes"}} + Subject: "mesh.seat.mesh-module-assigner.tool.nodes"}} }) if err != nil { t.Fatal(err) @@ -106,7 +106,7 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { controller.Flush() watcher.Flush() - up, err := Serve(NewSurface(nodeTools, "desk.node-tools"), "127.0.0.1:0") + up, err := Serve(NewSurface(nodeTools, busSource{nodeTools}, "desk.node-tools"), "127.0.0.1:0") if err != nil { t.Fatal(err) } @@ -125,19 +125,19 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { } } } - if got := strings.Join(names, ","); got != "mesh_overview,mesh_machine,mesh_search,mesh_describe,mesh_call,mesh_runtimes,mesh_wait" { + if got := strings.Join(names, ","); got != "mesh_overview,mesh_machine,mesh_search,mesh_describe,mesh_call,mesh_runtimes,mesh_first_views,mesh_first_view_again,mesh_wait" { t.Errorf("announced %s", got) } // The overview names the mesh's seats, the machines' seats and the machines. overview, isErr := call(t, endpoint, "mesh_overview", nil) - if isErr || !strings.Contains(overview, "mesh-placement.nodes") || !strings.Contains(overview, "/node-shelf.list") || + if isErr || !strings.Contains(overview, "mesh-module-assigner.nodes") || !strings.Contains(overview, "/node-shelf.list") || !strings.Contains(overview, `"bench"`) || !strings.Contains(overview, `"desk"`) { t.Errorf("overview: %s", overview) } // Silence is named only where tools should have answered: epsilon declares tools on bench and // nothing there announced it; delta declares none and is never a name. - if strings.Contains(overview, "mesh-placement on bench") { + if strings.Contains(overview, "mesh-module-assigner on bench") { t.Errorf("a mesh seat's holder that announced no machine is called silent on its machine:\n%s", overview) } if !strings.Contains(overview, "epsilon on bench") || strings.Contains(overview, "delta") { @@ -150,7 +150,7 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { } // A mesh seat, a node seat, an assignment and an interchangeable module, each by address. - if got, isErr := call(t, endpoint, "mesh_call", map[string]any{"address": "mesh-placement.nodes"}); isErr || !strings.Contains(got, "converged") { + if got, isErr := call(t, endpoint, "mesh_call", map[string]any{"address": "mesh-module-assigner.nodes"}); isErr || !strings.Contains(got, "converged") { t.Errorf("mesh seat: %s", got) } if got, isErr := call(t, endpoint, "mesh_call", map[string]any{"address": "desk/node-shelf.list"}); isErr || !strings.Contains(got, `"a"`) { @@ -166,11 +166,11 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { // Refused, by name, where the address does not say enough or says the wrong thing. for address, want := range map[string]string{ - "beta.three": "keeps state on each machine it runs on, so a call names the machine: write /beta.three — it runs on desk", - "node-shelf.list": "held once per machine: write /node-shelf.list — it is held on desk", - "desk/mesh-placement.nodes": "held once for the whole mesh", - "bench/beta.three": "beta does not run on bench; it runs on desk", - "nonsense": "is not an address", + "beta.three": "keeps state on each machine it runs on, so a call names the machine: write /beta.three — it runs on desk", + "node-shelf.list": "held once per machine: write /node-shelf.list — it is held on desk", + "desk/mesh-module-assigner.nodes": "held once for the whole mesh", + "bench/beta.three": "beta does not run on bench; it runs on desk", + "nonsense": "is not an address", } { got, isErr := call(t, endpoint, "mesh_call", map[string]any{"address": address}) if !isErr || !strings.Contains(got, want) { @@ -201,7 +201,7 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) { t.Fatalf("gamma was found before it served: %s", got) } mesh.Issue(t, mt.MembershipOf("gamma", "desk", false, nil)) - mesh.Placed(t, tools, map[string][]string{"desk": {"alpha", "beta", "delta", "gamma"}, "bench": {"epsilon", "mesh-placement"}}) + mesh.Placed(t, tools, map[string][]string{"desk": {"alpha", "beta", "delta", "gamma"}, "bench": {"epsilon", "mesh-module-assigner"}}) late := connect(t, "node-tools", "desk") stopLate, err := runtime.Run(late, []runtime.Served{{Module: "gamma", Entrypoints: []string{mt.Fixture("env-gamma.serve.mjs")}}}, nil, (&mt.Logs{}).Logf) diff --git a/node-tools/internal/console/client.go b/node-tools/internal/console/client.go index 751731b..dc1ef15 100644 --- a/node-tools/internal/console/client.go +++ b/node-tools/internal/console/client.go @@ -63,8 +63,8 @@ func toolKey(name string, seats Seats) string { // toolsOn is the flat catalogue (MESH_CONSOLE_FLAT=1): what announced itself on the bus, every // tool and seat verb, and the assignments with tools that did not (novox/hq ADR 0197). -func toolsOn(conn *bus.Conn) (*Listing, error) { - x, err := indexOn(conn, nil, nil) +func toolsOn(conn *bus.Conn, src Source) (*Listing, error) { + x, err := indexOn(conn, src, nil, nil) if err != nil { return nil, err } diff --git a/node-tools/internal/console/console_test.go b/node-tools/internal/console/console_test.go index 2ad8e70..174d844 100644 --- a/node-tools/internal/console/console_test.go +++ b/node-tools/internal/console/console_test.go @@ -57,7 +57,7 @@ func TestTheConsoleListsAndCallsOverHTTP(t *testing.T) { } defer stop() - // What the mesh holds, as the indexer and placement write it (ADR 0571): ghost declares tools on desk and nothing + // What the mesh holds, as the indexer and the assigner write it (ADR 0571): ghost declares tools on desk and nothing // announces it. mesh.Placed(t, map[string][]string{"alpha": {"one", "two"}, "beta": {"three"}, "ghost": {"boo"}}, map[string][]string{"desk": {"alpha", "beta", "ghost"}}) @@ -71,7 +71,7 @@ func TestTheConsoleListsAndCallsOverHTTP(t *testing.T) { return nil }) - flat := NewSurface(nodeTools, "desk.node-tools") + flat := NewSurface(nodeTools, busSource{nodeTools}, "desk.node-tools") flat.Flat = true // the whole catalogue, as before ADR 0195: still reachable, no longer announced up, err := Serve(flat, "127.0.0.1:0") if err != nil { @@ -132,7 +132,7 @@ func TestTheConsoleListsAndCallsOverHTTP(t *testing.T) { } func TestTheConsoleListensOnLoopbackAndNowhereElse(t *testing.T) { - if _, err := Serve(NewSurface(nil, "x"), "0.0.0.0:0"); err == nil || !strings.Contains(err.Error(), "loopback and nowhere else") { + if _, err := Serve(NewSurface(nil, nil, "x"), "0.0.0.0:0"); err == nil || !strings.Contains(err.Error(), "loopback and nowhere else") { t.Errorf("a non-loopback console was not refused: %v", err) } } diff --git a/node-tools/internal/console/empty_round_test.go b/node-tools/internal/console/empty_round_test.go index 474dafa..bff7b5b 100644 --- a/node-tools/internal/console/empty_round_test.go +++ b/node-tools/internal/console/empty_round_test.go @@ -72,7 +72,7 @@ func TestACallAfterARoundThatHeardNothingUsesWhatWasHeardBefore(t *testing.T) { } return n } - surface := NewSurface(console, "laptop.node-tools") + surface := NewSurface(console, busSource{console}, "laptop.node-tools") up, err := Serve(surface, "127.0.0.1:0") if err != nil { t.Fatal(err) diff --git a/node-tools/internal/console/firstviews.go b/node-tools/internal/console/firstviews.go new file mode 100644 index 0000000..0bb995c --- /dev/null +++ b/node-tools/internal/console/firstviews.go @@ -0,0 +1,65 @@ +package console + +import ( + "context" + "errors" + "log" + "strings" + + "git.novox.be/novox/mesh-sdk/go/views" +) + +// FirstViewer is the tool runner's four first views as the console shows them (novox/hq ADR 0615 §6; issue 1966 TLS-6): a +// Source that also is one is asked by mesh_first_views and mesh_first_view_again. +type FirstViewer interface { + Status() []views.Status + Again(ctx context.Context, why string) []error +} + +// firstViewsAnswer is mesh_first_views and mesh_first_view_again. +func (s *Surface) firstViewsAnswer(ctx context.Context, name string, args map[string]any) map[string]any { + fv, ok := s.src.(FirstViewer) + if !ok { + return failure("this tool runner holds no first views: it reads the indexer's and the assigner's states through " + + "another source") + } + if name == verbFirstViews { + return answerText(map[string]any{"first views": fv.Status()}) + } + why, _ := args["why"].(string) + if why = strings.TrimSpace(why); why == "" { + why = "asked by hand with no reason given" + } + // A hand act, recorded through the seat mesh-hand-acts (novox/hq ADR 0615 §6, ADR 0565 §10) before it is done, and in the + // runner's log. One that cannot be recorded is said, and the ask is not made. + log.Printf("[mesh-tools] hand act: %s asked the four first views again: %s", s.who, why) + if err := s.recordAct(map[string]any{"what": "asked the tool runner's four first views again", "why": why, + "cause": "first-view-again", "verb": verbFirstAgain}); err != nil { + return failure("the hand act was not recorded through mesh-hand-acts, so the first views were not asked again: " + err.Error()) + } + out := map[string]any{"asked again": why, "recorded": "mesh-hand-acts.hand-act"} + if errs := fv.Again(ctx, why); len(errs) > 0 { + var msgs []string + for _, e := range errs { + msgs = append(msgs, e.Error()) + } + out["not answered"] = msgs + } + out["first views"] = fv.Status() + return answerText(out) +} + +// recordAct records one act done by hand through the seat mesh-hand-acts (its verb hand-act); Surface.Record replaces the call +// in a test. +func (s *Surface) recordAct(args map[string]any) error { + if s.Record != nil { + return s.Record(args) + } + if s.conn == nil { + return errNoBus + } + _, err := s.conn.Ask("mesh-hand-acts.hand-act", args, "") + return err +} + +var errNoBus = errors.New("no bus connection") diff --git a/node-tools/internal/console/firstviews_test.go b/node-tools/internal/console/firstviews_test.go new file mode 100644 index 0000000..1af9b3b --- /dev/null +++ b/node-tools/internal/console/firstviews_test.go @@ -0,0 +1,59 @@ +package console + +import ( + "context" + "encoding/json" + "errors" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/views" +) + +// viewer is a Source that also holds first views. +type viewer struct { + busSource + asked []string +} + +func (v *viewer) Status() []views.Status { + return []views.Status{{Owner: "mesh-delivery-assigner", Tool: "assignments", State: "assignments", Revision: 4, Asks: 1}} +} + +func (v *viewer) Again(_ context.Context, why string) []error { + v.asked = append(v.asked, why) + return nil +} + +// issue 1966 TLS-6: the console's mesh_first_views shows each first view, mesh_first_view_again asks them again with its +// reason, and a source holding none refuses both in words. +func TestTheConsoleShowsAndAsksAgainTheRunnersFirstViews(t *testing.T) { + text := func(r map[string]any) string { + c, _ := json.Marshal(r["content"]) + return string(c) + } + v := &viewer{} + var recorded []map[string]any + s := &Surface{src: v, who: "desk.node-tools", Record: func(a map[string]any) error { recorded = append(recorded, a); return nil }} + if got := text(s.discover(context.Background(), "mesh_first_views", nil)); !strings.Contains(got, "mesh-delivery-assigner") || !strings.Contains(got, "assignments") { + t.Errorf("mesh_first_views shows %s", got) + } + got := s.discover(context.Background(), "mesh_first_view_again", map[string]any{"why": "doubted"}) + if len(v.asked) != 1 || v.asked[0] != "doubted" || !strings.Contains(text(got), "asked again") { + t.Errorf("mesh_first_view_again asked %v and answered %s", v.asked, text(got)) + } + if len(recorded) != 1 || recorded[0]["why"] != "doubted" || recorded[0]["cause"] != "first-view-again" { + t.Errorf("the hand act recorded is %v", recorded) + } + // An act that cannot be recorded is not done. + refusing := &Surface{src: v, who: "desk.node-tools", Record: func(map[string]any) error { return errors.New("no answer") }} + if r := refusing.discover(context.Background(), "mesh_first_view_again", map[string]any{"why": "x"}); r["isError"] != true || len(v.asked) != 1 { + t.Errorf("an unrecorded hand act was done: %s (asked %v)", text(r), v.asked) + } + for _, name := range []string{"mesh_first_views", "mesh_first_view_again"} { + none := &Surface{src: busSource{}, who: "desk.node-tools"} + if r := none.discover(context.Background(), name, nil); r["isError"] != true || !strings.Contains(text(r), "holds no first views") { + t.Errorf("%s without first views answers %s", name, text(r)) + } + } +} diff --git a/node-tools/internal/console/manifest_waits_test.go b/node-tools/internal/console/manifest_waits_test.go index 3d9386d..3511ec8 100644 --- a/node-tools/internal/console/manifest_waits_test.go +++ b/node-tools/internal/console/manifest_waits_test.go @@ -35,7 +35,30 @@ func TestTheManifestNamesTheFirstFamiliesOneByOne(t *testing.T) { t.Errorf("a family named by a wildcard: %s", w) } } - if len(m.Consumes) > 0 { - t.Errorf("the tool runner consumes %v: a wait is no consumer (ADR 0367 §2)", m.Consumes) + // A wait is no consumer (ADR 0367 §2): what it consumes is the four views families of its first views (ADR 0615 §5, issue + // 1966 TLS-5) and the two events of the seat that serves the seat list (ADR 0636 §4), and nothing else. + wantConsumes := []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.seat-holders", + "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules"} + if !reflect.DeepEqual(m.Consumes, wantConsumes) { + t.Errorf("the tool runner consumes %v, want %v", m.Consumes, wantConsumes) + } +} + +// issue 1966 TLS-1: the tool runner declares its own secret `ledgers`, each module's ledger credential, in the directory it +// keeps its broker credential in, which is where the runner reads it (runtime.LedgersFileOf). +func TestTheManifestDeclaresTheRunnersOwnSecretLedgers(t *testing.T) { + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + var m struct { + OwnSecrets map[string]string `json:"own-secrets"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + want := map[string]string{"broker": "${dir:mesh-state}/broker", "ledgers": "${dir:mesh-state}/ledgers"} + if !reflect.DeepEqual(m.OwnSecrets, want) { + t.Errorf("own secrets %v, want %v", m.OwnSecrets, want) } } diff --git a/node-tools/internal/console/mcp.go b/node-tools/internal/console/mcp.go index 66a196a..1874356 100644 --- a/node-tools/internal/console/mcp.go +++ b/node-tools/internal/console/mcp.go @@ -6,8 +6,8 @@ package console import ( "context" "encoding/json" + "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "git.novox.be/novox/mesh-sdk/go/catalogue" - "git.novox.be/novox/mesh-sdk/go/delivery" "os" "strings" "sync" @@ -47,6 +47,7 @@ type RPCError struct { // Surface answers MCP requests over one bus connection, as one account. type Surface struct { conn *bus.Conn + src Source who string mu sync.Mutex known *Listing @@ -60,6 +61,8 @@ type Surface struct { // Flat announces the whole catalogue, as the console did before ADR 0195: for a person reading // it or a client that wants it. Off by default; MESH_CONSOLE_FLAT=1 turns it on. Flat bool + // Record, when set, records a hand act in place of the seat mesh-hand-acts' verb (a test's). + Record func(args map[string]any) error } // MethodSeats is the console's JSON-RPC method beside MCP's that answers every seat's verbs and holders, in the shape @@ -67,15 +70,15 @@ type Surface struct { // [{module, node}]}], holders_why}`. const MethodSeats = "mesh/seats" -// seatsAnswer is MethodSeats' answer: mesh-sdk's seat definitions, with each seat's holders by placement. +// seatsAnswer is MethodSeats' answer: mesh-sdk's seat definitions, with each seat's holders by assignment. func (s *Surface) seatsAnswer() map[string]any { - placed, err := seatHoldersOn(s.conn) + placed, err := seatHoldersOn(s.src) x := &index{Placed: placed} seats := []map[string]any{} - for _, st := range catalogue.DefaultSeats() { + for _, st := range catalogue.Seats() { holders := x.placedHolders(st.Name, "") if holders == nil { - holders = []delivery.SeatHolder{} + holders = []meshmoduleassigner.SeatHoldersHoldersItem{} } tools := st.Serves if tools == nil { @@ -85,14 +88,14 @@ func (s *Surface) seatsAnswer() map[string]any { } out := map[string]any{"seats": seats} if err != nil { - out["holders_why"] = "who holds which seat by placement could not be read: " + err.Error() + out["holders_why"] = "who holds which seat by assignment could not be read: " + err.Error() } return out } // NewSurface is the surface over a connection, as `who`. -func NewSurface(conn *bus.Conn, who string) *Surface { - return &Surface{conn: conn, who: who, Flat: os.Getenv("MESH_CONSOLE_FLAT") == "1"} +func NewSurface(conn *bus.Conn, src Source, who string) *Surface { + return &Surface{conn: conn, src: src, who: who, Flat: os.Getenv("MESH_CONSOLE_FLAT") == "1"} } func (s *Surface) listing() (*Listing, error) { @@ -103,7 +106,7 @@ func (s *Surface) listing() (*Listing, error) { return l, nil } s.mu.Unlock() - l, err := toolsOn(s.conn) + l, err := toolsOn(s.conn, s.src) if err != nil { return nil, err } @@ -157,7 +160,7 @@ func (s *Surface) HandleContext(ctx context.Context, r Request) *Reply { return answer(r.ID, map[string]any{}) case MethodSeats: // The seats' contract and who holds each, for a client on this machine that has no seat definitions of its own - // (the TypeScript `mesh`): read in-process from mesh-sdk and from placement, never asked of anything else + // (the TypeScript `mesh`): read in-process from mesh-sdk and from the assigner, never asked of anything else // (novox/hq ADR 0571 clauses 4 and 5). return answer(r.ID, s.seatsAnswer()) case "tools/list": @@ -338,7 +341,7 @@ func (s *Surface) instructions() string { "that serves it; what may be called was fixed when this account was issued, so a " + "refusal means the account, not the tool. The list is what the running modules " + "answered, plus every role's tools from the mesh's records — the mesh's own verbs " + - "(mesh-delivery.deliveries, mesh-placement.propose …) among them; a module that did not answer " + + "(mesh-delivery.deliveries, mesh-module-assigner.propose …) among them; a module that did not answer " + "is named in the list's _meta and can still be called by .." } return "The tools of a Novox mesh, reached as " + s.who + ", found by address rather than listed " + diff --git a/node-tools/internal/console/node_argument_test.go b/node-tools/internal/console/node_argument_test.go index 2f00e51..0ea14b1 100644 --- a/node-tools/internal/console/node_argument_test.go +++ b/node-tools/internal/console/node_argument_test.go @@ -11,9 +11,9 @@ import ( // 2026-10-05 the controller's push was described as taking nothing, and a push naming one machine // ran as a push of every machine behind. func TestAMeshSeatsNodeIsItsOwnArgument(t *testing.T) { - push := Tool{Module: "mesh-placement", Name: "push", Seat: true, Scope: "mesh", + push := Tool{Module: "mesh-module-assigner", Name: "push", Seat: true, Scope: "mesh", Input: json.RawMessage(`{"type":"object","properties":{"node":{"type":"string"},"behind":{"type":"string"}}}`)} - mesh := target{Address: "mesh-placement.push", Tool: push, Seat: true} + mesh := target{Address: "mesh-module-assigner.push", Tool: push, Seat: true} props := schemaAsPassed(mesh)["properties"].(map[string]any) if _, has := props["node"]; !has { diff --git a/node-tools/internal/console/refusal_test.go b/node-tools/internal/console/refusal_test.go index 288cb12..68e350c 100644 --- a/node-tools/internal/console/refusal_test.go +++ b/node-tools/internal/console/refusal_test.go @@ -40,5 +40,11 @@ func TestATimeoutIsNotAFailure(t *testing.T) { if strings.Contains(text, "controller") { t.Errorf("%s names the controller: %s", name, text) } + // The seats were renamed (ADR 0634 §9): no hint names the seat that placed modules or enrolled a machine by its old name. + for _, old := range []string{"mesh-placement", "mesh-enrolment"} { + if strings.Contains(text, old) { + t.Errorf("%s names the seat %s, which is now mesh-module-assigner or mesh-node-enroller: %s", name, old, text) + } + } } } diff --git a/node-tools/internal/console/runtimes_test.go b/node-tools/internal/console/runtimes_test.go index 5287279..eb466e1 100644 --- a/node-tools/internal/console/runtimes_test.go +++ b/node-tools/internal/console/runtimes_test.go @@ -8,11 +8,13 @@ import ( "testing" "time" - "git.novox.be/novox/mesh-sdk/go/delivery" + assigner "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" + seatassigner "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "github.com/nats-io/nats.go/micro" "github.com/novox/mesh-tools/node-tools/internal/announce" "github.com/novox/mesh-tools/node-tools/internal/bus" + "github.com/novox/mesh-tools/node-tools/internal/meshtest" mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) @@ -56,18 +58,20 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T) IndexKept, announce.Patience = time.Minute, 2500*time.Millisecond t.Cleanup(func() { IndexKept, announce.Patience = keptWas, patienceWas }) mesh := mt.New(t) - // What the mesh holds, as the indexer and placement write it (ADR 0571): never the controller's records. + // What the mesh holds, as the indexer and the assigner write it (ADR 0571): never the controller's records. mesh.Placed(t, map[string][]string{"slack": {"slack_check"}, "nap": {"doze"}, "nextcloud-client": {"nextcloud_check"}}, map[string][]string{"laptop": {"slack", "nextcloud-client"}, "sleeper": {"nap", "nextcloud-client"}, "desktop": {"nextcloud-client"}, "bench": nil}) - // Who holds which seat, by placement (ADR 0571 clause 4), as the seat mesh-placement's holder composes it. - placedKV := mesh.Bucket(t, delivery.SeatHoldersBucket) - for key, sh := range delivery.SeatHoldersOf(map[string]delivery.Placement{"bench": {Node: "bench", - Holds: []delivery.Holding{{Claim: "mesh-wants", Scope: "mesh", Module: "mesh-delivery-assigner"}}}}) { - raw, _ := json.Marshal(sh) - if _, err := placedKV.Put(key, raw); err != nil { - t.Fatal(err) - } + // Who holds which seat, by assignment (ADR 0571 clause 4), as the seat mesh-module-assigner's holder composes it. + placedKV := mesh.Bucket(t, meshtest.SeatHoldersBucket) + holdersKey, err := assigner.SeatHoldersKey("mesh-wants", "") + if err != nil { + t.Fatal(err) + } + raw, _ := json.Marshal(seatassigner.SeatHolders{V: assigner.Version, Seat: "mesh-wants", Scope: "mesh", + Holders: []seatassigner.SeatHoldersHoldersItem{{Module: "mesh-delivery-assigner", Node: "bench"}}}) + if _, err := placedKV.Put(holdersKey, raw); err != nil { + t.Fatal(err) } consoleMembership := mt.MembershipOf("node-tools", "laptop-console", false, nil) consoleMembership.State = mt.ConsoleReads() @@ -95,7 +99,7 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T) Subject: "mesh.mod.nextcloud-client.tool.nextcloud_check.desktop"}, }) - controller := connect(t, "mesh-placement", "bench") + controller := connect(t, "mesh-module-assigner", "bench") var toolsAsked atomic.Int32 for verb, output := range map[string]string{ // The controller's records are served here only to see that none is asked (ADR 0571 clauses 4 and 5). @@ -104,7 +108,7 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T) "tools": `{"seats":[]}`, } { output, verb := output, verb - stop, err := controller.HandleSubject("mesh.seat.mesh-placement.tool."+verb, func(json.RawMessage) (any, error) { + stop, err := controller.HandleSubject("mesh.seat.mesh-module-assigner.tool."+verb, func(json.RawMessage) (any, error) { toolsAsked.Add(1) return map[string]any{"output": output, "ok": true}, nil }) @@ -123,7 +127,7 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T) } return nil }) - up, err := Serve(NewSurface(console, "laptop.node-tools"), "127.0.0.1:0") + up, err := Serve(NewSurface(console, busSource{console}, "laptop.node-tools"), "127.0.0.1:0") if err != nil { t.Fatal(err) } @@ -154,8 +158,8 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T) // The silent one is named, never called missing. for address, want := range map[string]string{ - // A seat whose holder announced nothing is named with its holder by placement, never called missing. - "mesh-wants.read": "the seat mesh-wants is held by mesh-delivery-assigner on bench by placement", + // A seat whose holder announced nothing is named with its holder by assignment, never called missing. + "mesh-wants.read": "the seat mesh-wants is held by mesh-delivery-assigner on bench by assignment", "sleeper/nap.doze": "nap is assigned to sleeper in the mesh's records, but nothing that answered discovery serves it: discovery did not hear in full from node-tools on sleeper (it answered PING", "sleeper/nextcloud-client.nextcloud_check": "nextcloud-client on sleeper did not answer discovery: discovery did not hear in full from node-tools on sleeper", } { diff --git a/node-tools/internal/console/search_test.go b/node-tools/internal/console/search_test.go index c624a07..247be24 100644 --- a/node-tools/internal/console/search_test.go +++ b/node-tools/internal/console/search_test.go @@ -33,8 +33,8 @@ func meshToSearch(t *testing.T) *index { "node": "What one machine reported it can do, what it is assigned, and why.", "nodes": "Every machine the mesh knows, with whether it is converged or adopted.", } { - endpoints = append(endpoints, announce.Endpoint{Kind: announce.KindSeat, Module: "mesh-placement", Seat: "mesh-placement", - Tool: verb, Scope: "mesh", Description: does, Subject: "mesh.seat.mesh-placement.tool." + verb}) + endpoints = append(endpoints, announce.Endpoint{Kind: announce.KindSeat, Module: "mesh-module-assigner", Seat: "mesh-module-assigner", + Tool: verb, Scope: "mesh", Description: does, Subject: "mesh.seat.mesh-module-assigner.tool." + verb}) } for tool, does := range map[string]string{ "docker_list": "Every container on this machine: name, image, state, ports.", @@ -46,7 +46,7 @@ func meshToSearch(t *testing.T) *index { Description: does, Subject: "mesh.mod.docker.tool." + tool + ".anchor"}) } x, _, _ := indexOfEndpoints(endpoints) - x.Replaces = replacesOf(catalogue.DefaultSeats(), []recordedModule{{Module: "docker", On: []string{"anchor"}, Tools: true, + x.Replaces = replacesOf(catalogue.Seats(), []recordedModule{{Module: "docker", On: []string{"anchor"}, Tools: true, Replaces: map[string][]string{"docker_list": {"docker ps"}, "docker_logs": {"docker logs"}, "docker_restart": {"docker restart"}}}}) return x } @@ -56,21 +56,21 @@ func meshToSearch(t *testing.T) *index { func TestSearchFindsTheToolThatReplacesACommandFirst(t *testing.T) { x := meshToSearch(t) for query, want := range map[string]string{ - "journalctl": "anchor/node-service-manager.journal", - "journalctl -u mesh-placement -n 50": "anchor/node-service-manager.journal", - "sudo journalctl -fu sshd": "anchor/node-service-manager.journal", - "/usr/bin/journalctl": "anchor/node-service-manager.journal", - "logs": "anchor/node-service-manager.journal", - "systemctl status": "anchor/node-service-manager.status", - "systemctl status nats.service": "anchor/node-service-manager.status", - "systemctl restart mesh-placement": "anchor/node-service-manager.restart", - "systemctl --failed": "anchor/node-service-manager.failed", - "docker ps": "anchor/docker.docker_list", - "docker ps -a": "anchor/docker.docker_list", - "docker logs": "anchor/docker.docker_logs", - "docker logs --tail 100 nats": "anchor/docker.docker_logs", - "docker restart nats": "anchor/docker.docker_restart", - "container logs": "anchor/docker.docker_logs", + "journalctl": "anchor/node-service-manager.journal", + "journalctl -u mesh-module-assigner -n 50": "anchor/node-service-manager.journal", + "sudo journalctl -fu sshd": "anchor/node-service-manager.journal", + "/usr/bin/journalctl": "anchor/node-service-manager.journal", + "logs": "anchor/node-service-manager.journal", + "systemctl status": "anchor/node-service-manager.status", + "systemctl status nats.service": "anchor/node-service-manager.status", + "systemctl restart mesh-module-assigner": "anchor/node-service-manager.restart", + "systemctl --failed": "anchor/node-service-manager.failed", + "docker ps": "anchor/docker.docker_list", + "docker ps -a": "anchor/docker.docker_list", + "docker logs": "anchor/docker.docker_logs", + "docker logs --tail 100 nats": "anchor/docker.docker_logs", + "docker restart nats": "anchor/docker.docker_restart", + "container logs": "anchor/docker.docker_logs", } { hits := x.search(query) if len(hits) == 0 { @@ -111,7 +111,7 @@ func TestASearchByCommandSaysWhatItReplaces(t *testing.T) { // A synonym is a word, never part of one: `logs` matches through `log`, and finds no `login`. func TestSearchStillMatchesEveryWord(t *testing.T) { x := meshToSearch(t) - if hits := x.search("machine converged"); len(hits) != 1 || hits[0].Address != "mesh-placement.nodes" { + if hits := x.search("machine converged"); len(hits) != 1 || hits[0].Address != "mesh-module-assigner.nodes" { t.Errorf("words in a description: %+v", hits) } if hits := x.search("journalctl kubernetes"); len(hits) != 1 || hits[0].Address != "anchor/node-service-manager.journal" { diff --git a/node-tools/internal/console/seats_for_tests_test.go b/node-tools/internal/console/seats_for_tests_test.go new file mode 100644 index 0000000..834bc56 --- /dev/null +++ b/node-tools/internal/console/seats_for_tests_test.go @@ -0,0 +1,10 @@ +package console + +import ( + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/seatlist/seatlisttest" +) + +// The seats the tests describe are given to catalogue as the tool runner gives the ones it read from the bus +// (novox/hq ADR 0635 §1); catalogue compiles none. +func init() { catalogue.UseSeats(catalogue.SeatsFromDefinitions(seatlisttest.Definitions())) } diff --git a/node-tools/internal/console/source_test.go b/node-tools/internal/console/source_test.go new file mode 100644 index 0000000..a3fed70 --- /dev/null +++ b/node-tools/internal/console/source_test.go @@ -0,0 +1,39 @@ +package console + +import ( + "sync" + + "github.com/novox/mesh-tools/node-tools/internal/bus" +) + +// busSource is a Source over a bus connection's own states, which a test's fixture mesh fills: the console's tests stand a +// membership that reads the four states beside it, where the tool runner itself holds first views (firstview.Views). +type busSource struct{ conn *bus.Conn } + +func (b busSource) State(name string) (map[string][]byte, error) { + var mu sync.Mutex + keys := map[string][]byte{} + stop, err := b.conn.StateWatch(b.conn.Credential().Module, name, "", func(c bus.StateChange) error { + if c.Current && c.Op == "put" { + mu.Lock() + keys[c.Key] = c.Value + mu.Unlock() + } + return nil + }) + if err != nil { + return nil, err + } + stop() + mu.Lock() + defer mu.Unlock() + return keys, nil +} + +func (b busSource) Entry(name, key string) ([]byte, error) { + entry, err := b.conn.StateGet(b.conn.Credential().Module, name, key) + if err != nil || entry == nil { + return nil, err + } + return entry.Value, nil +} diff --git a/node-tools/internal/console/wait_test.go b/node-tools/internal/console/wait_test.go index 40c5a0c..862517f 100644 --- a/node-tools/internal/console/wait_test.go +++ b/node-tools/internal/console/wait_test.go @@ -33,7 +33,7 @@ func waitingRuntime(t *testing.T) (*Surface, *mt.Mesh) { } return nil }) - return NewSurface(conn, "desk.node-tools"), mesh + return NewSurface(conn, busSource{conn}, "desk.node-tools"), mesh } // callWait is one tools/call of mesh_wait, answered as an agent host reads it. @@ -176,7 +176,7 @@ func TestMeshWaitRefusesAnEventOutsideTheGrantedFamilies(t *testing.T) { // A tool runner issued no families waits on nothing, and says so. func TestMeshWaitWithNoFamiliesIssuedWaitsOnNothing(t *testing.T) { mt.New(t) - s := NewSurface(connect(t, "node-tools", "desk"), "desk.node-tools") + s := NewSurface(connect(t, "node-tools", "desk"), nil, "desk.node-tools") text, isErr := callWait(s, map[string]any{"subjects": []any{"mesh-knowledge.reviewed"}, "timeout": 1}) if !isErr || !strings.Contains(text, "no event families") { t.Errorf("a runtime issued no families: %s", text) @@ -194,7 +194,7 @@ func TestMeshWaitIsAnnouncedAmongSeven(t *testing.T) { } } sort.Strings(names) - want := []string{"mesh_call", "mesh_describe", "mesh_machine", "mesh_overview", "mesh_runtimes", "mesh_search", "mesh_wait"} + want := []string{"mesh_call", "mesh_describe", "mesh_first_view_again", "mesh_first_views", "mesh_machine", "mesh_overview", "mesh_runtimes", "mesh_search", "mesh_wait"} if !reflect.DeepEqual(names, want) { t.Errorf("announced %v, want %v", names, want) } diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go new file mode 100644 index 0000000..4b93e15 --- /dev/null +++ b/node-tools/internal/firstview/firstview.go @@ -0,0 +1,230 @@ +// Package firstview is the tool runner's picture of the four states it reads of the indexer and the assigner: the assigner's +// assignments and seat-holders, the indexer's manifests and modules (novox/hq ADR 0615 §1, §3; ADR 0621 §2; issue 1966 TLS-5). +// +// **It never reads the owners' state.** At start it asks each owner's first-view tool until answered (the owner answers in a +// runtime of its own, never through this one: ADR 0621 §1), then follows the owner's views family, one event per changed key; +// a gap in the revisions, or an owner whose state was made again, asks the first view once more, said in the log. Past two +// minutes without an answer it writes a finding in its own state `findings`, which the condition keeper raises and clears. +package firstview + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "sync" + "time" + + conditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + + "git.novox.be/novox/mesh-sdk/go/views" + + "github.com/novox/mesh-tools/node-tools/internal/bus" +) + +// The owners and their first-view tools (the tools' addresses are `.`). +const ( + Indexer = "mesh-delivery-indexer" + Assigner = "mesh-delivery-assigner" +) + +// The names the console reads the four states by (`.`), as the manifest's `consumes` names their views +// families (no seat's client names a holder, novox/hq ADR 0625 §4; TestTheFirstViewsAreTheManifestsViews). +const ( + AssignmentsRead = Assigner + ".assignments" + SeatHoldersRead = Assigner + ".seat-holders" + ModulesRead = Indexer + ".modules" + ManifestsRead = Indexer + ".manifests" +) + +// Spec is one first view the runner holds. +type Spec struct{ Owner, Tool, State string } + +// Specs are the four. +var Specs = []Spec{ + {Assigner, "assignments", "assignments"}, + {Assigner, "seat-holders", "seat-holders"}, + {Indexer, "manifests", "manifests"}, + {Indexer, "modules", "modules"}, +} + +// Name is the name a state is read by, `.`: AssignmentsRead, SeatHoldersRead and the indexer's. +func (s Spec) Name() string { return s.Owner + "." + s.State } + +// Views is the four followers. +type Views struct { + logf func(string, ...any) + followers map[string]*views.Follower // by Spec.Name() + byState map[string]*views.Follower // by the state an event names + ctx context.Context + cancel context.CancelFunc + wg sync.WaitGroup + deps Deps +} + +// FindingsState is the runner's own state the findings are written in. +const FindingsState = "findings" + +// Deps is what the views reach the bus through: Start makes them from the runner's connection, a test from fakes. +type Deps struct { + // Module is the runner's own name, the observer of its findings. + Module string + // Ask calls an owner's tool, `.`. + Ask func(address string, body any) (json.RawMessage, error) + // Consume hands each event of the owners' views families to deliver, until the stop is called. + Consume func(deliver func(bus.Envelope) error) (stop func(), err error) + // PutFinding and DeleteFinding keep and remove a finding in the runner's own state. + PutFinding func(key string, value json.RawMessage) error + DeleteFinding func(key string) error + // Bound and Wait are the follower's (zero: two minutes, five seconds). + Bound, Wait time.Duration +} + +// Start asks the four first views, each until answered, and follows the owners' views families on the runner's own +// consumers. It returns at once: the picture fills as the owners answer. +func Start(parent context.Context, conn *bus.Conn, logf func(string, ...any)) (*Views, func()) { + return StartWith(parent, DepsOf(conn), logf) +} + +// DepsOf is the views' reach through the runner's own connection: its tool calls, its consumer and its findings state. +func DepsOf(conn *bus.Conn) Deps { + me := conn.Module() + return Deps{Module: me, + Ask: func(address string, body any) (json.RawMessage, error) { + got, err := conn.Ask(address, body, "") + return got.Result, err + }, + Consume: func(deliver func(bus.Envelope) error) (func(), error) { return conn.ConsumeAs(me, deliver) }, + PutFinding: func(key string, value json.RawMessage) error { + _, err := conn.StatePut(me, FindingsState, key, value) + return err + }, + DeleteFinding: func(key string) error { return conn.StateDelete(me, FindingsState, key) }, + } +} + +// StartWith is Start over deps. +func StartWith(parent context.Context, d Deps, logf func(string, ...any)) (*Views, func()) { + ctx, cancel := context.WithCancel(parent) + v := &Views{logf: logf, ctx: ctx, cancel: cancel, followers: map[string]*views.Follower{}, byState: map[string]*views.Follower{}, deps: d} + for _, s := range Specs { + s := s + call := func(tool string, body any) (json.RawMessage, error) { return d.Ask(s.Owner+"."+tool, body) } + f := &views.Follower{Owner: s.Owner, Tool: s.Tool, State: s.State, Ask: views.AskVia(call, s.Tool), Logf: logf, + Bound: d.Bound, Wait: d.Wait, + Raise: func(st views.Status) { v.finding(st, true) }, + Clear: func(st views.Status) { v.finding(st, false) }} + v.followers[s.Name()] = f + v.byState[s.State] = f + v.wg.Add(1) + go func() { defer v.wg.Done(); _ = f.Start(ctx) }() + } + stop, err := d.Consume(v.hear) + if err != nil { + logf("[mesh-tools] the owners' views are not followed (%v): the pictures are the first views alone until the runner starts again", err) + stop = func() {} + } + return v, func() { stop(); cancel(); v.wg.Wait() } +} + +// hear applies one views event; a gap asks the first view once more. +func (v *Views) hear(env bus.Envelope) error { + ev, err := views.DecodeEvent(env.Body) + if err != nil { + return nil // not a views event: nothing to hold, nothing to retry + } + f := v.byState[ev.State] + if f == nil || f.Owner != ev.Owner { + return nil + } + if gap, why := f.Apply(ev); gap { + v.logf("[mesh-tools] a gap in %s's %s (%s): its first view %s is asked once more", f.Owner, f.State, why, f.Tool) + go func() { _ = f.Again(v.ctx, "a gap: "+why) }() + } + return nil +} + +// finding writes or removes the runner's finding of a first view unanswered past its bound, in its own state. +func (v *Views) finding(st views.Status, raised bool) { + key, value, err := conditionOf(views.UnansweredFinding(v.deps.Module, st, time.Now())) + if err != nil { + v.logf("[mesh-tools] the finding of the first view %s is not made: %v", st.Tool, err) + return + } + if raised { + err = v.deps.PutFinding(key, value) + } else { + err = v.deps.DeleteFinding(key) + } + if err != nil { + v.logf("[mesh-tools] the finding of the first view %s is not kept: %v", st.Tool, err) + } +} + +// State is the picture of the state spec names (`.`), every current value by key; an error says the owner has +// not answered yet, so nothing is claimed of it. +func (v *Views) State(name string) (map[string][]byte, error) { + f := v.followers[name] + if f == nil { + return nil, fmt.Errorf("the tool runner holds no first view of %s", name) + } + select { + case <-f.Ready(): + default: + return nil, fmt.Errorf("%s's first view %s has not been answered yet", f.Owner, f.Tool) + } + out := map[string][]byte{} + for k, raw := range f.Values() { + out[k] = raw + } + return out, nil +} + +// Entry is one value of the picture, nil when the owner holds none. +func (v *Views) Entry(name, key string) ([]byte, error) { + all, err := v.State(name) + if err != nil { + return nil, err + } + return all[key], nil +} + +// Status is each first view as it stands, by `.`, sorted. +func (v *Views) Status() []views.Status { + var out []views.Status + for _, f := range v.followers { + out = append(out, f.Status()) + } + sort.Slice(out, func(i, j int) bool { return out[i].Owner+out[i].State < out[j].Owner+out[j].State }) + return out +} + +// Again asks every first view once more, by hand, for why. +func (v *Views) Again(ctx context.Context, why string) []error { + var errs []error + for _, f := range v.followers { + if err := f.Again(ctx, why); err != nil { + errs = append(errs, err) + } + } + return errs +} + +// conditionOf is a finding in the layout of the seat mesh-conditions' state findings (its contract, ADR 0634 §1), about a +// module, a warning. +func conditionOf(f views.Finding) (string, []byte, error) { + key := conditions.Key(scopeModule, f.ID, f.Kind) + v := condseat.FindingsByKeyValue{V: 1, Key: key, Kind: f.Kind, Observer: f.Observer, Severity: conditions.Warning, + Subject: condseat.Subject{Scope: scopeModule, ID: f.ID}, Summary: f.Summary, Headline: f.Headline, + Explanation: f.Explanation, Source: f.Observer + "." + f.Kind, At: f.At.UTC(), + Evidence: []condseat.Evidence{{At: f.At.UTC(), Said: f.Summary}}} + if err := conditions.CheckFinding(v); err != nil { + return "", nil, err + } + raw, err := json.Marshal(v) + return key, raw, err +} + +// scopeModule is the scope of a finding about a module. +const scopeModule = "module" diff --git a/node-tools/internal/firstview/firstview_integration_test.go b/node-tools/internal/firstview/firstview_integration_test.go new file mode 100644 index 0000000..baa369d --- /dev/null +++ b/node-tools/internal/firstview/firstview_integration_test.go @@ -0,0 +1,158 @@ +//go:build integration + +package firstview + +import ( + "context" + "encoding/json" + "os" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// issue 1966 TLS-5 [i] (ADR 0615 §2, §4, §6): against the real bus, with the owners down and then up, the runner's own +// consumers on the owners' own views streams carry the manifest's four `consumes`: the first views are asked until answered, +// the unanswered ones are findings that the answer clears, and each owner's next event reaches its picture. +func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThenUp(t *testing.T) { + mesh := mt.New(t) + url := mt.URL(t) + + // The owners publish their views on their own streams (views.Emit: `mesh.stream..event.views.`, on + // assigner-views and indexer-views), never on EVENTS. The consumers on them are the mesh's, one filter per family the + // manifest's `consumes` names, and the runner's membership names them as the mesh issues it (ADR 0363 §8). + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + var manifest struct { + Consumes []string `json:"consumes"` + } + if err := json.Unmarshal(raw, &manifest); err != nil { + t.Fatal(err) + } + streamOf := map[string]string{Assigner: catalogue.StreamBusName("assigner-views"), Indexer: catalogue.StreamBusName("indexer-views")} + filters := map[string][]string{} + families := 0 + for _, c := range manifest.Consumes { + emitter, event, isView := strings.Cut(c, ".views.") + if !isView { // the seat mesh-catalog's events: the seat list's, on EVENTS (cmd/node-tools/seatlist.go) + continue + } + stream := streamOf[emitter] + if stream == "" { + t.Fatalf("the manifest consumes %s, whose views stream the test does not know", c) + } + filters[stream] = append(filters[stream], process.ModuleStreamSubject(emitter, views.EventPrefix+event)) + families++ + } + if families != len(Specs) { + t.Fatalf("the manifest consumes %d families, the runner holds %d first views", families, len(Specs)) + } + runner := mt.MembershipOf("node-tools", "anchor", false, nil) + for _, owner := range []string{Assigner, Indexer} { + stream := streamOf[owner] + mesh.ModuleStream(t, owner, stream) + mesh.ConsumerOn(t, stream, "anchor", "node-tools", filters[stream], 2*time.Second) + runner.Hears = append(runner.Hears, bus.HeardIssued{Stream: stream, Consumer: "anchor_node-tools"}) + } + mesh.Consumer(t, "anchor", "node-tools", []string{"mesh.mod.mesh-knowledge.event.reviewed"}, 2*time.Second) // its `waits`, on EVENTS + mesh.Issue(t, runner) + + conn, err := bus.Connect(bus.Credential{URL: url, Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + conn.Logf = func(string, ...any) {} + t.Cleanup(conn.Close) + conn.Follow("node-tools") + mt.Until(t, func() error { + if conn.Membership("node-tools") == nil { + return errFindings(-1) + } + return nil + }) + + var mu sync.Mutex + findings := map[string]bool{} + deps := DepsOf(conn) + deps.Bound, deps.Wait = 100*time.Millisecond, 20*time.Millisecond + deps.PutFinding = func(key string, _ json.RawMessage) error { mu.Lock(); findings[key] = true; mu.Unlock(); return nil } + deps.DeleteFinding = func(key string) error { mu.Lock(); delete(findings, key); mu.Unlock(); return nil } + v, stop := StartWith(context.Background(), deps, func(string, ...any) {}) + t.Cleanup(stop) + + // The owners are down: no tool answers, nothing is claimed, and past the bound each view is a finding. + if _, err := v.State(Assigner + ".assignments"); err == nil { + t.Fatal("a state its owner has not answered is claimed") + } + mt.Until(t, func() error { + mu.Lock() + defer mu.Unlock() + if len(findings) != len(Specs) { + return errFindings(len(findings)) + } + return nil + }) + + // The owners come up and answer their first views. + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + for _, s := range Specs { + s := s + if _, err := nc.Subscribe("mesh.mod."+s.Owner+".tool."+s.Tool, func(m *nats.Msg) { + answer, _ := json.Marshal(views.Answer{V: 1, Owner: s.Owner, State: s.State, Revision: 4, Created: time.Unix(1000, 0).UTC(), + Values: map[string]json.RawMessage{"first": json.RawMessage(`{"n":1}`)}}) + body, _ := json.Marshal(map[string]json.RawMessage{"result": answer}) + _ = m.Respond(body) + }); err != nil { + t.Fatal(err) + } + } + _ = nc.Flush() + for _, s := range Specs { + name := s.Name() + mt.Until(t, func() error { _, err := v.Entry(name, "first"); return err }) + } + mt.Until(t, func() error { + mu.Lock() + defer mu.Unlock() + if len(findings) != 0 { + return errFindings(len(findings)) + } + return nil + }) + + // Each owner's next event arrives on the runner's consumer: all four families are bound. + for _, s := range Specs { + ev, _ := json.Marshal(views.Event{V: 1, Owner: s.Owner, State: s.State, Key: "next", Revision: 5, Created: time.Unix(1000, 0).UTC(), + Value: json.RawMessage(`{"n":2}`)}) + mesh.Emit(t, process.ModuleStreamSubject(s.Owner, views.EventPrefix+s.State), json.RawMessage(ev)) + } + for _, s := range Specs { + name := s.Name() + mt.Until(t, func() error { + got, err := v.Entry(name, "next") + if err != nil || string(got) != `{"n":2}` { + return errFindings(-1) + } + return nil + }) + } +} + +type errFindings int + +func (e errFindings) Error() string { return "not yet" } diff --git a/node-tools/internal/firstview/firstview_test.go b/node-tools/internal/firstview/firstview_test.go new file mode 100644 index 0000000..11c299d --- /dev/null +++ b/node-tools/internal/firstview/firstview_test.go @@ -0,0 +1,197 @@ +package firstview + +import ( + "context" + "encoding/json" + "errors" + "os" + "slices" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/views" + + "github.com/novox/mesh-tools/node-tools/internal/bus" +) + +// owners is the four owners' first-view tools as a test answers them: a state's values at a revision, or nothing while down. +type owners struct { + mu sync.Mutex + down bool + state map[string]map[string]string // tool → key → value + rev map[string]uint64 + asked map[string]int + deliver func(bus.Envelope) error + finds map[string]bool +} + +func newOwners() *owners { + return &owners{state: map[string]map[string]string{"assignments": {"anchor": `{"p":1}`}, "seat-holders": {"seat": `{"h":1}`}, + "manifests": {"d1": `{"m":1}`}, "modules": {"web": `{"w":1}`}}, + rev: map[string]uint64{"assignments": 4, "seat-holders": 2, "manifests": 7, "modules": 3}, asked: map[string]int{}, finds: map[string]bool{}} +} + +func (o *owners) ask(address string, _ any) (json.RawMessage, error) { + o.mu.Lock() + defer o.mu.Unlock() + tool := address[strings.LastIndex(address, ".")+1:] + o.asked[tool]++ + if o.down { + return nil, errors.New("503 no responders") + } + vals := map[string]json.RawMessage{} + for k, v := range o.state[tool] { + vals[k] = json.RawMessage(v) + } + return json.Marshal(views.Answer{V: 1, Owner: "o", State: tool, Revision: o.rev[tool], Created: time.Unix(1000, 0).UTC(), Values: vals}) +} + +func (o *owners) say(t *testing.T, state, key, value string, revision uint64, gap bool) { + t.Helper() + ev := views.Event{V: 1, Owner: owner(state), State: state, Key: key, Revision: revision, Created: time.Unix(1000, 0).UTC(), Gap: gap} + if value != "" { + ev.Value = json.RawMessage(value) + } + raw, _ := json.Marshal(ev) + if err := o.deliver(bus.Envelope{Key: ev.Owner + ".views." + state, Body: raw}); err != nil { + t.Fatal(err) + } +} + +func owner(state string) string { + for _, s := range Specs { + if s.State == state { + return s.Owner + } + } + return "" +} + +func eventually(t *testing.T, what string, ok func() bool) { + t.Helper() + for end := time.Now().Add(3 * time.Second); !ok(); time.Sleep(5 * time.Millisecond) { + if time.Now().After(end) { + t.Fatal(what) + } + } +} + +// issue 1966 TLS-5 (ADR 0615 §2, §4, §5, §6): the runner holds the four states from the owners' first views and their +// events alone; nothing is claimed of a state not yet answered; a gap asks the first view once more; an owner that stays down +// past the bound is a finding, cleared by the answer; the status and the hand ask show them. +func TestTheRunnerHoldsTheFourStatesFromTheOwnersFirstViewsAndTheirEvents(t *testing.T) { + o := newOwners() + o.down = true + var logs []string + var lmu sync.Mutex + deps := Deps{Module: "node-tools", Ask: o.ask, Bound: 30 * time.Millisecond, Wait: 10 * time.Millisecond, + Consume: func(deliver func(bus.Envelope) error) (func(), error) { o.deliver = deliver; return func() {}, nil }, + PutFinding: func(key string, _ json.RawMessage) error { + o.mu.Lock() + o.finds[key] = true + o.mu.Unlock() + return nil + }, + DeleteFinding: func(key string) error { + o.mu.Lock() + delete(o.finds, key) + o.mu.Unlock() + return nil + }} + v, stop := StartWith(context.Background(), deps, func(f string, a ...any) { lmu.Lock(); logs = append(logs, f); lmu.Unlock() }) + defer stop() + + // Down: nothing is claimed, and past the bound each unanswered view is a finding. + if _, err := v.State("mesh-delivery-assigner.assignments"); err == nil || !strings.Contains(err.Error(), "not been answered") { + t.Fatalf("a state its owner has not answered is claimed: %v", err) + } + eventually(t, "no finding past the bound", func() bool { o.mu.Lock(); defer o.mu.Unlock(); return len(o.finds) == 4 }) + + // Up: answered, the findings are cleared, and the pictures hold the owners' values. + o.mu.Lock() + o.down = false + o.mu.Unlock() + eventually(t, "not answered", func() bool { _, err := v.State("mesh-delivery-indexer.manifests"); return err == nil }) + eventually(t, "findings not cleared", func() bool { o.mu.Lock(); defer o.mu.Unlock(); return len(o.finds) == 0 }) + if got, _ := v.Entry("mesh-delivery-indexer.modules", "web"); string(got) != `{"w":1}` { + t.Errorf("the modules picture holds %q", got) + } + if got, _ := v.Entry("mesh-delivery-indexer.modules", "none"); got != nil { + t.Errorf("an absent key is %q", got) + } + + // Events: the next revision is applied; an old one passed over; a deletion removes; none of it asks again. + asked := func(tool string) int { o.mu.Lock(); defer o.mu.Unlock(); return o.asked[tool] } + before := asked("assignments") + o.say(t, "assignments", "laptop", `{"p":2}`, 5, false) + o.say(t, "assignments", "laptop", `{"p":0}`, 5, false) + o.say(t, "assignments", "anchor", "", 6, false) + // (a deleted mark carries no value: Deleted is the encoding's, set below) + if got, _ := v.Entry("mesh-delivery-assigner.assignments", "laptop"); string(got) != `{"p":2}` { + t.Errorf("the next revision is not applied: %q", got) + } + if n := asked("assignments"); n != before { + t.Errorf("an event in order asked the first view again (%d asks, was %d)", n, before) + } + + // A gap (revision 9 where 7 was next) asks the first view once more, said in the log. + o.say(t, "assignments", "x", `{"p":9}`, 9, false) + eventually(t, "a gap did not ask the first view again", func() bool { return asked("assignments") > before }) + + // The status says owner, tool, revision and the asks; the hand ask asks every view again. + var assignments views.Status + for _, s := range v.Status() { + if s.State == "assignments" { + assignments = s + } + } + if assignments.Owner != "mesh-delivery-assigner" || assignments.Tool != "assignments" || assignments.Asks < 2 || assignments.Answered.IsZero() { + t.Errorf("the status of assignments is %+v", assignments) + } + manifests := asked("manifests") + if errs := v.Again(context.Background(), "doubted"); len(errs) != 0 { + t.Fatalf("the hand ask failed: %v", errs) + } + if asked("manifests") != manifests+1 { + t.Errorf("the hand ask did not ask the manifests again") + } + for _, st := range v.Status() { + if st.State == "manifests" && st.ByHand != 1 { + t.Errorf("the hand ask is not counted: %+v", st) + } + } + lmu.Lock() + defer lmu.Unlock() + if !strings.Contains(strings.Join(logs, "\n"), "a gap in") { + t.Errorf("the gap was not said in the log: %v", logs) + } +} + +// The four first views are the manifest's `consumes` families, and the console reads each by the name here. +func TestTheFirstViewsAreTheManifestsViews(t *testing.T) { + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + var m struct { + Consumes []string `json:"consumes"` + } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + want := map[string]bool{AssignmentsRead: true, SeatHoldersRead: true, ModulesRead: true, ManifestsRead: true} + for _, s := range Specs { + if !want[s.Name()] { + t.Errorf("the first view %s is read by no name here", s.Name()) + } + delete(want, s.Name()) + if view := s.Owner + ".views." + s.Tool; !slices.Contains(m.Consumes, view) { + t.Errorf("the first view %s has no family in the manifest's consumes %v", view, m.Consumes) + } + } + if len(want) > 0 { + t.Errorf("no first view reads %v", want) + } +} diff --git a/node-tools/internal/launch/event_answer_test.go b/node-tools/internal/launch/event_answer_test.go index 0824263..01fde7c 100644 --- a/node-tools/internal/launch/event_answer_test.go +++ b/node-tools/internal/launch/event_answer_test.go @@ -19,6 +19,9 @@ while IFS= read -r line; do *'"method":"tools/list"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id" printf '{"jsonrpc":"2.0","id":"s1","method":"mesh/subscribe","params":{"pattern":"#"}}\n' ;; + *'"method":"mesh/heartbeat"'*) + params=$(printf '%s' "$line" | sed -n 's/.*"params":\(.*\)}$/\1/p') + printf '{"jsonrpc":"2.0","id":"h1","method":"mesh/publish","params":%s}\n' "$params" ;; *'"method":"mesh/event"'*) case "$ANSWER" in error) printf '{"jsonrpc":"2.0","id":%s,"error":{"code":-32000,"message":"Unexpected end of JSON input"}}\n' "$id" ;; @@ -30,11 +33,18 @@ done ` type subscribing struct { - mu sync.Mutex - deliver func(json.RawMessage) error + mu sync.Mutex + deliver func(json.RawMessage) error + beat func(json.RawMessage) + published []json.RawMessage } -func (b *subscribing) Publish(json.RawMessage) error { return nil } +func (b *subscribing) Publish(p json.RawMessage) error { + b.mu.Lock() + b.published = append(b.published, p) + b.mu.Unlock() + return nil +} func (b *subscribing) Ask(json.RawMessage) (json.RawMessage, error) { return nil, nil } func (b *subscribing) State(string, json.RawMessage) (json.RawMessage, error) { return nil, nil } func (b *subscribing) Log(string, json.RawMessage) (json.RawMessage, error) { return nil, nil } @@ -53,10 +63,11 @@ func (b *subscribing) BusStateWatch(string, json.RawMessage, func(json.RawMessag return func() {}, nil } func (b *subscribing) Begin(string) (string, func()) { return "", func() {} } -func (b *subscribing) Subscribe(d func(json.RawMessage) error) error { +func (b *subscribing) Subscribe(d func(json.RawMessage) error, beat func(json.RawMessage)) error { b.mu.Lock() defer b.mu.Unlock() b.deliver = d + b.beat = beat return nil } @@ -104,3 +115,45 @@ func TestAnEventIsTakenByAnyAnswerThatIsNotAnError(t *testing.T) { func (b *subscribing) LogWatch(json.RawMessage, func(json.RawMessage) error) (func(), error) { return func() {}, nil } + +// The runtime's word on the module's consumer between events reaches the child that subscribed as `mesh/heartbeat`, +// unchanged (stdio.OnHeartbeat reads it in the bundle). +func TestTheWordOnTheConsumerBetweenEventsReachesTheBundleThatSubscribed(t *testing.T) { + entry := filepath.Join(t.TempDir(), "bundle") + if err := os.WriteFile(entry, []byte(answeringBundle), 0o755); err != nil { + t.Fatal(err) + } + b := &subscribing{} + l, err := Start("plex", entry, append(os.Environ(), "ANSWER=result"), b, t.Logf) + if err != nil { + t.Fatal(err) + } + t.Cleanup(l.Stop) + var beat func(json.RawMessage) + for i := 0; beat == nil; i++ { + b.mu.Lock() + beat = b.beat + b.mu.Unlock() + if i > 100 { + t.Fatal("the bundle never subscribed") + } + time.Sleep(20 * time.Millisecond) + } + word := json.RawMessage(`{"at":"2026-11-04T13:00:00Z","missed":true,"why":"the bus ended it"}`) + beat(word) + for i := 0; ; i++ { + b.mu.Lock() + got := b.published + b.mu.Unlock() + if len(got) == 1 { + if string(got[0]) != string(word) { + t.Fatalf("the bundle was told %s, not %s", got[0], word) + } + return + } + if i > 100 { + t.Fatal("the bundle was told nothing") + } + time.Sleep(20 * time.Millisecond) + } +} diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index f663c82..0044ea5 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -80,7 +80,9 @@ type Registration struct { // Bus is what a launched bundle reaches the mesh through (novox/hq ADR 0193, ADR 0198): the runtime // publishes, asks and subscribes on the module's behalf. Subscribe is asked each time the module's // code subscribes; the runtime binds the module's consumer once and calls deliver for every event, -// acknowledging it on the bus when deliver returns nil. +// acknowledging it on the bus when deliver returns nil. beat is the runtime's word on that consumer between +// events (a bus.Beat as JSON), which the launch hands the child that subscribed as `mesh/heartbeat`, so a bundle +// can tell a silent source from a quiet one (stdio.OnHeartbeat). // // State and Watch reach the module's state (novox/hq ADR 0201): State answers `get`, `put`, `delete` // and `keys`; Watch hands deliver the current values and then every change, and returns once the @@ -88,7 +90,7 @@ type Registration struct { type Bus interface { Publish(params json.RawMessage) error Ask(params json.RawMessage) (json.RawMessage, error) - Subscribe(deliver func(envelope json.RawMessage) error) error + Subscribe(deliver func(envelope json.RawMessage) error, beat func(heartbeat json.RawMessage)) error State(verb string, params json.RawMessage) (json.RawMessage, error) Watch(params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error) // Log answers a bundle's `append`, `read`, `last` and `info` on its module's log (novox/hq ADR 0297). @@ -261,6 +263,17 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an _, err := c.ask(module, "mesh/event", map[string]any{"envelope": envelope}, EventTimeout) return err } + // The heartbeat of the module's consumer goes to the child that last subscribed, as a notification: the + // bundle answers nothing, and a child not running misses it, as it misses the events it did not take. + beat := func(heartbeat json.RawMessage) { + mu.Lock() + c := subscriber + mu.Unlock() + if c == nil { + return + } + _ = c.write(map[string]any{"jsonrpc": "2.0", "method": "mesh/heartbeat", "params": heartbeat}) + } var restart func(after time.Duration) var startedAt time.Time // The restart pauses are read once, here: the exit handling runs on its own goroutine and must not @@ -344,7 +357,7 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an mu.Lock() subscriber = c mu.Unlock() - err = mesh.Subscribe(deliver) + err = mesh.Subscribe(deliver, beat) case "mesh/seat.publish", "mesh/seat.prove", "mesh/seat.record", "mesh/seat.kinds", "mesh/seat.pull", "mesh/seat.settle": var answered json.RawMessage if answered, err = mesh.Seat(strings.TrimPrefix(m.Method, "mesh/seat."), m.Params, nil); err == nil { diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index 9ca7cbf..2f021e1 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -13,9 +13,6 @@ package meshtest import ( "context" "encoding/json" - "git.novox.be/novox/mesh-sdk/go/buscompose" - "git.novox.be/novox/mesh-sdk/go/delivery" - "git.novox.be/novox/mesh-sdk/go/membership" "os" "path/filepath" "runtime" @@ -24,6 +21,12 @@ import ( "testing" "time" + assigner "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" + seatassigner "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" + meshcatalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + meshcatalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/membership" + "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -88,8 +91,8 @@ func New(t *testing.T) *Mesh { js, _ := nc.JetStream() _ = js.DeleteStream("EVENTS") // The grants part's state memberships, where a runtime reads what is issued it (ADR 0571 clause 10). - _ = js.DeleteKeyValue(buscompose.GrantsMembershipsBucket) - if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: buscompose.GrantsMembershipsBucket, History: 1}); err != nil { + _ = js.DeleteKeyValue(membership.GrantsBucket) + if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: membership.GrantsBucket, History: 1}); err != nil { t.Fatal(err) } if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil { @@ -100,7 +103,7 @@ func New(t *testing.T) *Mesh { } // Issue writes a membership where the grants part keeps it, its state memberships, key `.` (novox/hq ADR -// 0571 clause 7 and 10; mesh-sdk buscompose.GrantsMembershipsBucket), the one place a runtime reads it. It writes with +// 0571 clause 7 and 10; mesh-sdk membership.GrantsBucket), the one place a runtime reads it. It writes with // mesh-sdk's encoder, the grants part's own, never json of this package's type: a value built here by hand agreed with // the reader's mistake and hid that the part wrote another shape (issue 774). func (m *Mesh) Issue(t *testing.T, mem bus.Membership) { @@ -120,13 +123,13 @@ func (m *Mesh) IssueInGrantsWrapped(t *testing.T, mem bus.Membership, published func (m *Mesh) putGrants(t *testing.T, mem bus.Membership, body []byte) { t.Helper() - kv, err := m.js.KeyValue(buscompose.GrantsMembershipsBucket) + kv, err := m.js.KeyValue(membership.GrantsBucket) if err != nil { - if kv, err = m.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: buscompose.GrantsMembershipsBucket, History: 1}); err != nil { + if kv, err = m.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: membership.GrantsBucket, History: 1}); err != nil { t.Fatal(err) } } - if _, err := kv.Put(buscompose.MembershipKey(mem.Node, mem.Module), body); err != nil { + if _, err := kv.Put(membership.GrantsKey(mem.Node, mem.Module), body); err != nil { t.Fatal(err) } } @@ -134,7 +137,7 @@ func (m *Mesh) putGrants(t *testing.T, mem bus.Membership, body []byte) { // DropGrants deletes the grants part's state memberships, as a bus that has not made it yet holds none; Issue makes it again. func (m *Mesh) DropGrants(t *testing.T) { t.Helper() - if err := m.js.DeleteKeyValue(buscompose.GrantsMembershipsBucket); err != nil { + if err := m.js.DeleteKeyValue(membership.GrantsBucket); err != nil { t.Fatal(err) } } @@ -150,19 +153,31 @@ func sdkMembership(t *testing.T, mem bus.Membership) membership.Membership { return out } +// The indexer's two states the console reads, and their buckets (the test bed's own names: the indexer's client names +// no holder). +const ( + indexer = "mesh-delivery-indexer" + assignerHolder = "mesh-delivery-assigner" + // AssignmentsBucket and SeatHoldersBucket are the assigner's states assignments and seat-holders. + AssignmentsBucket = assignerHolder + "_assignments" + SeatHoldersBucket = assignerHolder + "_seat-holders" + ModulesBucket = indexer + "_modules" + ManifestsBucket = indexer + "_manifests" +) + // ConsoleReads are the states a console's runtime reads to say what the mesh holds (novox/hq ADR 0571 clauses 4, 5): -// placement, the indexer's modules and manifests, and the seat holders. +// assignments, the indexer's modules and manifests, and the seat holders. func ConsoleReads() []bus.StateIssued { return []bus.StateIssued{ - {Name: delivery.PlacementRead, Bucket: delivery.PlacementBucket}, - {Name: delivery.IndexerModule + ".modules", Bucket: delivery.ModulesBucket}, - {Name: delivery.IndexerModule + ".manifests", Bucket: delivery.ManifestsBucket}, - {Name: delivery.SeatHoldersRead, Bucket: delivery.SeatHoldersBucket}, + {Name: assignerHolder + ".assignments", Bucket: AssignmentsBucket}, + {Name: indexer + ".modules", Bucket: ModulesBucket}, + {Name: indexer + ".manifests", Bucket: ManifestsBucket}, + {Name: assignerHolder + ".seat-holders", Bucket: SeatHoldersBucket}, } } // Placed writes what the mesh holds as its parts write it, with mesh-sdk's encoders: each module the indexer holds with -// its manifest (tools, the tools it declares; none for a module without), and each node's placement of modules. +// its manifest (tools, the tools it declares; none for a module without), and each node's assignment of modules. func (m *Mesh) Placed(t *testing.T, tools map[string][]string, on map[string][]string) { t.Helper() kv := func(bucket string) nats.KeyValue { @@ -173,20 +188,20 @@ func (m *Mesh) Placed(t *testing.T, tools map[string][]string, on map[string][]s } return b } - modules, manifests, placement := kv(delivery.ModulesBucket), kv(delivery.ManifestsBucket), kv(delivery.PlacementBucket) - kv(delivery.SeatHoldersBucket) + modules, manifests, assigned := kv(ModulesBucket), kv(ManifestsBucket), kv(AssignmentsBucket) + kv(SeatHoldersBucket) for module, declared := range tools { manifest, _ := json.Marshal(map[string]any{"module": module, "tools": declared}) - digest := delivery.ManifestDigest(manifest) - raw, err := delivery.EncodeManifest(delivery.Manifest{Module: module, JSON: string(manifest)}) + digest := meshcatalogclient.ManifestDigest(manifest) + raw, err := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: module, Manifest: string(manifest), V: 1}) if err != nil { t.Fatal(err) } if _, err := manifests.Put(digest, raw); err != nil { t.Fatal(err) } - raw, err = delivery.EncodeModule(delivery.Module{None: "not built here", Manifest: digest, Repository: "novox/" + module, - Commit: strings.Repeat("a", 40)}) + raw, err = meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{None: "not built here", Manifest: digest, Repository: "novox/" + module, + Commit: strings.Repeat("a", 40), V: 1}) if err != nil { t.Fatal(err) } @@ -195,17 +210,17 @@ func (m *Mesh) Placed(t *testing.T, tools map[string][]string, on map[string][]s } } for node, placed := range on { - p := delivery.Placement{V: delivery.Version, Node: node} + p := seatassigner.Assignments{V: assigner.Version, Node: node} for _, module := range placed { - p.Modules = append(p.Modules, delivery.PlacedModule{Module: module}) + p.Modules = append(p.Modules, seatassigner.PlacedModule{Module: module}) } // As the seat writes it: its node key and a key per module (research 072 row 37). - values, err := delivery.PlacementValues(p) + values, err := assigner.AssignmentsValues(p) if err != nil { t.Fatal(err) } for key, raw := range values { - if _, err := placement.Put(key, raw); err != nil { + if _, err := assigned.Put(key, raw); err != nil { t.Fatal(err) } } diff --git a/node-tools/internal/provide/provide.go b/node-tools/internal/provide/provide.go index c7a400c..7e331a8 100644 --- a/node-tools/internal/provide/provide.go +++ b/node-tools/internal/provide/provide.go @@ -15,8 +15,8 @@ import ( "fmt" "time" - "git.novox.be/novox/mesh-sdk/go/provider" - "git.novox.be/novox/mesh-sdk/go/provisions" + provisions "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient" + "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient/provider" "github.com/novox/mesh-tools/node-tools/internal/bus" ) @@ -38,6 +38,10 @@ func Offers() map[string]provider.Offer { }} } +// Provisioner is the module holding the seat mesh-provisioner, whose provisions and plans the helper reads: named from this +// module's manifest (`reads`; TestTheOfferServesThePortTheManifestServes), since no client names a holder (novox/hq ADR 0625 §4). +const Provisioner = "mesh-provisioner" + // busOf is the helper's bus over the runtime's own state: the provider's bucket is its state `bound`, and the // provisions and plans of the seat mesh-provisioner are the states it reads, by mesh-sdk's names (novox/hq ADR 0570, // issue 1954 L6-T1). The old assignments are read no more (mesh-sdk 692dbf9, L6-P3). @@ -47,10 +51,10 @@ func (b busOf) state(bucket string) (string, error) { switch bucket { case provider.BucketOf(Module): return "bound", nil - case provisions.ProvisionsBucket: - return provisions.ProvisionsRead, nil - case provisions.PlansBucket: - return provisions.PlansRead, nil + case provisions.ProvisionsBucketOf(Provisioner): + return provisions.ProvisionsReadOf(Provisioner), nil + case provisions.PlansBucketOf(Provisioner): + return provisions.PlansReadOf(Provisioner), nil } return "", fmt.Errorf("%s is not a bucket this provider keeps or reads", bucket) } @@ -97,7 +101,7 @@ func (b busOf) Delete(_ context.Context, bucket, key string) error { func New(conn *bus.Conn, node string, every time.Duration, logf func(string, ...any)) *provider.Provider { return provider.New(provider.Options{ Node: node, - Module: Module, + Module: Module, Provisioner: Provisioner, Offers: Offers(), Bus: busOf{conn: conn}, Every: every, diff --git a/node-tools/internal/provide/provide_test.go b/node-tools/internal/provide/provide_test.go index 47957c9..eb35808 100644 --- a/node-tools/internal/provide/provide_test.go +++ b/node-tools/internal/provide/provide_test.go @@ -7,9 +7,9 @@ import ( "testing" "time" + provisions "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient" + "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient/provider" "git.novox.be/novox/mesh-sdk/go/catalogue" - "git.novox.be/novox/mesh-sdk/go/provider" - "git.novox.be/novox/mesh-sdk/go/provisions" "github.com/novox/mesh-tools/node-tools/internal/bus" mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" @@ -45,14 +45,14 @@ func fill(t *testing.T, template string, entry map[string]any) string { func TestTheRuntimeAnswersAHandedConsumersMcpEndpointThroughTheHelper(t *testing.T) { mesh := mt.New(t) bound := mesh.Bucket(t, "node-tools_bound") - plans := mesh.Bucket(t, provisions.PlansBucket) - provs := mesh.Bucket(t, provisions.ProvisionsBucket) + plans := mesh.Bucket(t, provisions.PlansBucketOf(Provisioner)) + provs := mesh.Bucket(t, provisions.ProvisionsBucketOf(Provisioner)) mem := mt.MembershipOf("node-tools", "anchor", false, nil) mem.State = []bus.StateIssued{ {Name: "bound", Bucket: "node-tools_bound", Writes: true}, - {Name: provisions.PlansRead, Bucket: provisions.PlansBucket}, - {Name: provisions.ProvisionsRead, Bucket: provisions.ProvisionsBucket}, + {Name: provisions.PlansReadOf(Provisioner), Bucket: provisions.PlansBucketOf(Provisioner)}, + {Name: provisions.ProvisionsReadOf(Provisioner), Bucket: provisions.ProvisionsBucketOf(Provisioner)}, } mesh.Issue(t, mem) @@ -160,8 +160,8 @@ func TestTheOfferServesThePortTheManifestServes(t *testing.T) { } } for _, r := range m.Reads { - if r == "mesh-placement.assignments" { - t.Error("the manifest still reads mesh-placement.assignments, which the sdk's helper no longer reads (issue 1954)") + if r == "mesh-module-assigner.assignments" { + t.Error("the manifest still reads mesh-module-assigner.assignments, which the sdk's helper no longer reads (issue 1954)") } } for _, a := range m.Build.Artifacts { @@ -176,9 +176,9 @@ func TestTheOfferServesThePortTheManifestServes(t *testing.T) { declared[n] = true } for bucket, want := range map[string]string{ - provider.BucketOf(Module): "bound", - provisions.ProvisionsBucket: "mesh-provisioner.provisions", - provisions.PlansBucket: "mesh-provisioner.plans", + provider.BucketOf(Module): "bound", + provisions.ProvisionsBucketOf(Provisioner): "mesh-provisioner.provisions", + provisions.PlansBucketOf(Provisioner): "mesh-provisioner.plans", } { got, err := (busOf{}).state(bucket) if err != nil || got != want || !declared[got] { diff --git a/node-tools/internal/runtime/announce_test.go b/node-tools/internal/runtime/announce_test.go index ab49ada..c263b7e 100644 --- a/node-tools/internal/runtime/announce_test.go +++ b/node-tools/internal/runtime/announce_test.go @@ -10,6 +10,7 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/micro" + "github.com/novox/mesh-tools/node-tools/internal/announce" "github.com/novox/mesh-tools/node-tools/internal/bus" mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) @@ -100,10 +101,6 @@ func TestTheRuntimeAnnouncesWhatItServesInTheServicesProtocol(t *testing.T) { if msg, err := nc.Request("$SRV.PING", nil, 2*time.Second); err != nil || !strings.Contains(string(msg.Data), micro.PingResponseType) { t.Errorf("ping: %v %v", msg, err) } - if _, err := nc.Request("$SRV.INFO.somebody-else", nil, 300*time.Millisecond); err == nil { - t.Error("answered a request for another service") - } - // alpha is re-issued a plain subject: the announcement says so, and says it is interchangeable. mesh.Issue(t, mt.MembershipOf("alpha", "anchor", true, nil)) var after string @@ -120,3 +117,29 @@ func TestTheRuntimeAnnouncesWhatItServesInTheServicesProtocol(t *testing.T) { } _ = bus.Served{} } + +// issue 1966 TLS-3 (ENG-2): the runner states the capabilities of the calls it answers in its PING, on its own instance, which +// is what its machine's node-engine asks and takes into the node's record; no bundle is needed for it. +func TestTheRunnerStatesTheCapabilitiesOfItsCallsInItsPing(t *testing.T) { + mesh := mt.New(t) + mesh.Issue(t, mt.MembershipOf("alpha", "anchor", false, nil)) + nodeTools := connect(t, "node-tools", "anchor") + stop, err := Run(nodeTools, nil, nil, (&mt.Logs{}).Logf) + if err != nil { + t.Fatal(err) + } + defer stop() + nc, err := nats.Connect(mt.URL(t)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + msg, err := nc.Request("$SRV.PING.node-tools.anchor", nil, 2*time.Second) + if err != nil { + t.Fatalf("ping of the instance: %v", err) + } + var ping micro.Ping + if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Metadata[announce.Capabilities] != "bus-state-calls,ledger-calls" { + t.Errorf("the runner states %q (%v), want its bus-state and ledger calls", ping.Metadata[announce.Capabilities], err) + } +} diff --git a/node-tools/internal/runtime/busstate.go b/node-tools/internal/runtime/busstate.go index bd72a15..3d293d4 100644 --- a/node-tools/internal/runtime/busstate.go +++ b/node-tools/internal/runtime/busstate.go @@ -3,10 +3,38 @@ package runtime import ( "encoding/json" "fmt" + "sort" "git.novox.be/novox/mesh-sdk/go/buscalls" ) +// The capabilities the tool runner states of the calls it answers (novox/hq ADR 0595 §9, ADR 0597; design 25 § 12.4): each is +// a family of the methods a bundle asks it, named so a manifest's `capabilities` can ask for it, and stated in its announcement +// (announce.Capabilities) for its machine's node-engine to take into the node's record. A module whose manifest asks for one is +// held off a machine whose runner does not state it. +const ( + CapBusStateCalls = "bus-state-calls" + CapLedgerCalls = "ledger-calls" +) + +// callFamilies are the methods each capability stands for, the ones BusState and BusStateWatch answer. +var callFamilies = map[string][]string{ + CapBusStateCalls: {buscalls.ObjectsPut, buscalls.ObjectsGet, buscalls.ObjectsList, buscalls.ObjectsDelete, buscalls.ObjectsInfo, + buscalls.ObjectsWatch, buscalls.StreamRead, buscalls.StreamFill, buscalls.WorkQueueSize, buscalls.WorkQueueDead}, + CapLedgerCalls: {buscalls.LedgerApply, buscalls.LedgerGet, buscalls.LedgerHistory, buscalls.LedgerWatch, buscalls.LedgerQuery, + buscalls.LedgerRebuild, buscalls.LedgerReindex, buscalls.LedgerSnapshot, buscalls.LedgerLag}, +} + +// CallCapabilities is every capability of the calls this runner answers, sorted. +func CallCapabilities() []string { + out := make([]string, 0, len(callFamilies)) + for c := range callFamilies { + out = append(out, c) + } + sort.Strings(out) + return out +} + // BusState answers a bundle's calls on its module's bus state in the new form (mesh-sdk buscalls; novox/hq ADR 0595 // §9, design 25 § 12.5; issue 1966): each as the module whose bundle asks, never a module the request names, so a // bundle reaches its own module's bindings alone. diff --git a/node-tools/internal/runtime/events_test.go b/node-tools/internal/runtime/events_test.go index bea321b..3f98d4e 100644 --- a/node-tools/internal/runtime/events_test.go +++ b/node-tools/internal/runtime/events_test.go @@ -1,11 +1,13 @@ package runtime import ( + "encoding/json" "fmt" "os" "path/filepath" "strings" "sync" + "sync/atomic" "testing" "time" @@ -157,3 +159,37 @@ func TestALongRunningBundleThatExitsIsStartedAgain(t *testing.T) { func fmtSprintf(format string, args ...any) string { return fmt.Sprintf(format, args...) } func errorf(format string, args ...any) error { return fmt.Errorf(format, args...) } + +// A bundle loaded again sets its event and heartbeat handlers while the consumer's loop hands a word or an event on: no data race (-race), +// and the heartbeat is asked of the bus only while some bundle holds a handler. +func TestAHeartbeatHandlerSetAgainWhileWordsAreHandedOnIsNotARace(t *testing.T) { + m := &moduleEvents{} + all := &consumers{logf: t.Logf, of: map[string]*moduleEvents{"mod": m}} + slot := &beatSlot{} + m.beats = []*beatSlot{slot} + if m.beatAsked() { + t.Fatal("a heartbeat is asked while no bundle gave a handler") + } + took := &deliverSlot{} + took.set(func(json.RawMessage) error { return nil }) + m.delivers = []*deliverSlot{took} + var got atomic.Int32 + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 200; i++ { + slot.set(func(json.RawMessage) { got.Add(1) }) + took.set(func(json.RawMessage) error { got.Add(1); return nil }) + } + }() + for i := 0; i < 200; i++ { + all.handBeat(m, bus.Beat{At: time.Now()}) + if err := all.handOut("mod", m, bus.Envelope{Key: "a.b", Body: json.RawMessage(`{}`)}); err != nil { + t.Fatal(err) + } + } + <-done + if !m.beatAsked() { + t.Error("a bundle gave a handler and no heartbeat is asked") + } +} diff --git a/node-tools/internal/runtime/ledger_credentials_test.go b/node-tools/internal/runtime/ledger_credentials_test.go index c96afd5..0377d5c 100644 --- a/node-tools/internal/runtime/ledger_credentials_test.go +++ b/node-tools/internal/runtime/ledger_credentials_test.go @@ -63,7 +63,7 @@ func TestALedgerIsWrittenOnlyOnItsModulesLedgerConnection(t *testing.T) { t.Fatal(err) } } - for _, b := range []string{"m_records", buscompose.GrantsMembershipsBucket} { + for _, b := range []string{"m_records", membership.GrantsBucket} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil { t.Fatal(err) } @@ -76,8 +76,8 @@ func TestALedgerIsWrittenOnlyOnItsModulesLedgerConnection(t *testing.T) { raw, _ := json.Marshal(mem) var sdk membership.Membership _ = json.Unmarshal(raw, &sdk) - grants, _ := js.KeyValue(ctx, buscompose.GrantsMembershipsBucket) - if _, err := grants.Put(ctx, buscompose.MembershipKey("anchor", "m"), membership.EncodeGrantsMembership(sdk)); err != nil { + grants, _ := js.KeyValue(ctx, membership.GrantsBucket) + if _, err := grants.Put(ctx, membership.GrantsKey("anchor", "m"), membership.EncodeGrantsMembership(sdk)); err != nil { t.Fatal(err) } diff --git a/node-tools/internal/runtime/ledgers.go b/node-tools/internal/runtime/ledgers.go index ab2be1e..114cd54 100644 --- a/node-tools/internal/runtime/ledgers.go +++ b/node-tools/internal/runtime/ledgers.go @@ -213,6 +213,21 @@ const ( RepairLag = "ledger_lag" ) +// repairClash is why a bundle's tools of a module cannot be served beside the runner's own ledger repairs: a tool named as +// one of them, which the runner serves for every module owning a ledger, would be taken for it or hide it. It names the +// tool and the repair it clashes with; empty when none does. +func repairClash(module string, tools []launch.Tool) string { + for _, t := range tools { + for _, repair := range []string{RepairRebuild, RepairReindex, RepairSnapshot, RepairLag} { + if t.Name == repair { + return fmt.Sprintf("%s's bundle serves a tool named %s, which is also the ledger repair the runner serves for %s "+ + "(novox/hq ADR 0597 §10): rename the bundle's tool", module, t.Name, module) + } + } + } + return "" +} + // The effects a tool carries (novox/hq ADR 0387 §1). const ( EffectReads = "reads" diff --git a/node-tools/internal/runtime/ledgers_test.go b/node-tools/internal/runtime/ledgers_test.go index b688711..6d530fd 100644 --- a/node-tools/internal/runtime/ledgers_test.go +++ b/node-tools/internal/runtime/ledgers_test.go @@ -15,6 +15,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-tools/node-tools/internal/bus" + "github.com/novox/mesh-tools/node-tools/internal/launch" mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) @@ -179,3 +180,45 @@ func TestAnApplyIsStampedWithTheCallerTheRuntimeProved(t *testing.T) { t.Fatal("a call that ended was still taken as in flight") } } + +// issue 1966 TLS-2: a bundle's tool named as one of the runner's ledger repairs is refused at the load, naming the tool and the +// repair; any other name is served. +func TestABundleToolNamedAsALedgerRepairIsRefused(t *testing.T) { + for _, c := range []struct{ name, tool, want string }{ + {"rebuild", RepairRebuild, "m's bundle serves a tool named ledger_rebuild, which is also the ledger repair the runner serves for m"}, + {"reindex", RepairReindex, "ledger_reindex"}, + {"snapshot", RepairSnapshot, "ledger_snapshot"}, + {"lag", RepairLag, "ledger_lag"}, + {"another name", "ledger_status", ""}, + } { + t.Run(c.name, func(t *testing.T) { + got := repairClash("m", []launch.Tool{{Name: "first"}, {Name: c.tool}}) + if (c.want == "") != (got == "") || !strings.Contains(got, c.want) { + t.Errorf("clash %q, want one containing %q", got, c.want) + } + }) + } +} + +// issue 1966 TLS-3: every method of a capability the runner states is a call it answers, and the two states its calls under +// no other name; none answers "the runtime answers no ...". +func TestEveryCallOfAStatedCapabilityIsAnswered(t *testing.T) { + m, _, _ := aLedger(t) + if got := strings.Join(CallCapabilities(), ","); got != "bus-state-calls,ledger-calls" { + t.Fatalf("the runner states %s", got) + } + for capability, methods := range callFamilies { + for _, method := range methods { + t.Run(capability+" "+method, func(t *testing.T) { + _, err := m.BusState(method, json.RawMessage(`"not a request"`)) + if err != nil && strings.Contains(err.Error(), "the runtime answers no") { + if stop, werr := m.BusStateWatch(method, json.RawMessage(`"not a request"`), func(json.RawMessage) error { return nil }); werr != nil && strings.Contains(werr.Error(), "the runtime answers no") { + t.Errorf("%s is stated under %s and answered by neither BusState nor BusStateWatch", method, capability) + } else if stop != nil { + stop() + } + } + }) + } + } +} diff --git a/node-tools/internal/runtime/reload_test.go b/node-tools/internal/runtime/reload_test.go index 52599df..7375d77 100644 --- a/node-tools/internal/runtime/reload_test.go +++ b/node-tools/internal/runtime/reload_test.go @@ -343,7 +343,7 @@ func TestAnEventHeardDuringAReloadIsOfferedAgainNotAPanic(t *testing.T) { c := &consumers{logf: func(string, ...any) {}, of: map[string]*moduleEvents{}} var got []string before := func(json.RawMessage) error { got = append(got, "before"); return nil } - old := &moduleEvents{delivers: []*func(json.RawMessage) error{&before}} + old := &moduleEvents{delivers: []*deliverSlot{{f: before}}} c.of["ledger"] = old c.dropModule("ledger") env := bus.Envelope{Key: "alpha.happened", Body: json.RawMessage(`{"n":1}`)} @@ -359,7 +359,7 @@ func TestAnEventHeardDuringAReloadIsOfferedAgainNotAPanic(t *testing.T) { t.Error("an event heard while the module is loaded again was taken, not offered again") } after := func(json.RawMessage) error { got = append(got, "after"); return nil } - fresh := &moduleEvents{delivers: []*func(json.RawMessage) error{&after}} + fresh := &moduleEvents{delivers: []*deliverSlot{{f: after}}} c.of["ledger"] = fresh if err := hand(fresh); err != nil { t.Errorf("the bundle loaded again was not handed the event: %v", err) diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index cd76059..de633ed 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -308,7 +308,7 @@ func Run(conn *bus.Conn, served []Served, envs map[string]map[string]string, log announced, err := announce.Serve(conn, announce.Service{ Name: conn.Module(), ID: instanceOf(conn), Description: "the mesh's tool runtime on " + node + ": every assigned module's tools and the seats they hold", - Metadata: map[string]string{"node": node}, + Metadata: map[string]string{"node": node, announce.Capabilities: strings.Join(CallCapabilities(), ",")}, }, func() []announce.Endpoint { return endpointsOf(conn, r.own(), r.registrations(), r.modules) }) if err != nil { logf("[mesh-tools] cannot announce what it serves: %v", err) @@ -396,6 +396,19 @@ func (r *runner) load(s Served) *loadedModule { fail(err.Error()) continue } + // A tool named as one of the runner's ledger repairs would be taken for it, or hide it: refused at the load, naming both + // (ADR 0597 §10). + clashed := "" + for _, reg := range child.Registrations { + if clashed == "" && r.isServed[reg.Module] { + clashed = repairClash(reg.Module, reg.Tools) + } + } + if clashed != "" { + child.Stop() + fail(clashed) + continue + } l.children = append(l.children, child.Stop) for _, reg := range child.Registrations { got := registration{module: reg.Module, owner: module, tools: reg.Tools} @@ -761,7 +774,42 @@ type consumers struct { type moduleEvents struct { stop func() - delivers []*func(json.RawMessage) error + delivers []*deliverSlot + beats []*beatSlot +} + +// slot is where a bundle's handler is kept (its event handler, its heartbeat handler); a bundle loaded again sets it +// while the consumer's loop reads it, so both go through its own lock. +type slot[F any] struct { + mu sync.Mutex + f F +} + +func (s *slot[F]) get() F { + s.mu.Lock() + defer s.mu.Unlock() + return s.f +} + +func (s *slot[F]) set(f F) { + s.mu.Lock() + defer s.mu.Unlock() + s.f = f +} + +type ( + deliverSlot = slot[func(json.RawMessage) error] + beatSlot = slot[func(json.RawMessage)] +) + +// beatAsked says whether any bundle of the module asked for the heartbeat; the caller holds c.mu. +func (m *moduleEvents) beatAsked() bool { + for _, t := range m.beats { + if t.get() != nil { + return true + } + } + return false } // stopAll unbinds every module's consumer. @@ -808,7 +856,8 @@ type moduleBus struct { all *consumers module string mu sync.Mutex - deliver *func(json.RawMessage) error + deliver *deliverSlot + beat *beatSlot } // Publish emits an event as the module (ADR 0193). @@ -846,15 +895,18 @@ func (b *moduleBus) Ask(params json.RawMessage) (json.RawMessage, error) { // Subscribe hands this bundle the module's events. The consumer is bound once per module; each of // the module's children that subscribed is handed every event, and the event is acknowledged only // when all of them took it — one consumer split between two readers would give each half. -func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error) error { +func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error, beat func(json.RawMessage)) error { b.mu.Lock() if b.deliver == nil { - d := deliver - b.deliver = &d - } else { - *b.deliver = deliver + b.deliver = &deliverSlot{} } + b.deliver.set(deliver) mine := b.deliver + if b.beat == nil { + b.beat = &beatSlot{} + } + b.beat.set(beat) + mineBeat := b.beat b.mu.Unlock() c := b.all @@ -873,12 +925,19 @@ func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error) error { } if !listed { m.delivers = append(m.delivers, mine) + m.beats = append(m.beats, mineBeat) } if m.stop != nil { return nil } module := b.module - stop, err := c.conn.ConsumeAs(module, func(env bus.Envelope) error { return c.handOut(module, m, env) }) + // The bus is asked about the consumer only while a bundle of the module asked for the heartbeat. + stop, err := c.conn.ConsumeAsBeatingWhen(module, func(env bus.Envelope) error { return c.handOut(module, m, env) }, + func(b bus.Beat) { c.handBeat(m, b) }, func() bool { + c.mu.Lock() + defer c.mu.Unlock() + return m.beatAsked() + }) if err != nil { return err } @@ -887,6 +946,22 @@ func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error) error { return nil } +// handBeat hands the word on a module's consumer to every child of the module that subscribed, as an event is. +func (c *consumers) handBeat(m *moduleEvents, b bus.Beat) { + raw, err := json.Marshal(b) + if err != nil { + return + } + c.mu.Lock() + targets := append([]*beatSlot(nil), m.beats...) + c.mu.Unlock() + for _, t := range targets { + if f := t.get(); f != nil { + f(raw) + } + } +} + // errModuleReloading refuses an event heard by a module's consumer that was stopped for a reload. var errModuleReloading = errors.New("the module's bundle is being loaded again; offered again") @@ -904,10 +979,10 @@ func (c *consumers) handOut(module string, m *moduleEvents, env bus.Envelope) er c.mu.Unlock() return errModuleReloading } - targets := append([]*func(json.RawMessage) error(nil), m.delivers...) + targets := append([]*deliverSlot(nil), m.delivers...) c.mu.Unlock() for _, d := range targets { - if err := (*d)(raw); err != nil { + if err := d.get()(raw); err != nil { c.logf("[mesh-tools] %s did not take %s%s: %s; offered again in %s", module, env.Key, eventRef(env), whyNotTaken(err), bus.NakDelay) return err diff --git a/node-tools/internal/runtime/seat_traffic_test.go b/node-tools/internal/runtime/seat_traffic_test.go index 00a4723..9857a32 100644 --- a/node-tools/internal/runtime/seat_traffic_test.go +++ b/node-tools/internal/runtime/seat_traffic_test.go @@ -70,7 +70,7 @@ func TestABundlesSeatTrafficGoesThroughItsModulesBus(t *testing.T) { if !strings.Contains(string(first), `"duplicate":false`) || !strings.Contains(string(again), `"duplicate":true`) { t.Errorf("the same id twice: %s, then %s", first, again) } - if _, err := asking.Seat("publish", json.RawMessage(`{"subject":"mesh.seat.operator-channel.accept.ask.mesh-placement","body":{}}`), nil); err == nil { + if _, err := asking.Seat("publish", json.RawMessage(`{"subject":"mesh.seat.operator-channel.accept.ask.mesh-module-assigner","body":{}}`), nil); err == nil { t.Error("a module submitted under another's name") } got, err := asking.Seat("record", json.RawMessage(`{"key":"d-1"}`), nil) diff --git a/node-tools/module.json b/node-tools/module.json index e9d1a43..f490158 100644 --- a/node-tools/module.json +++ b/node-tools/module.json @@ -29,7 +29,8 @@ } }, "own-secrets": { - "broker": "${dir:mesh-state}/broker" + "broker": "${dir:mesh-state}/broker", + "ledgers": "${dir:mesh-state}/ledgers" }, "listens": [ { @@ -66,14 +67,19 @@ ] }, "state": [ - "bound" + "bound", + "findings" + ], + "consumes": [ + "mesh-catalog.seat-definition-changed", + "mesh-catalog.seat-definition-removed", + "mesh-delivery-assigner.views.assignments", + "mesh-delivery-assigner.views.seat-holders", + "mesh-delivery-indexer.views.manifests", + "mesh-delivery-indexer.views.modules" ], "reads": [ "mesh-provisioner.provisions", - "mesh-provisioner.plans", - "mesh-delivery-assigner.seat-holders", - "mesh-delivery-assigner.placement", - "mesh-delivery-indexer.modules", - "mesh-delivery-indexer.manifests" + "mesh-provisioner.plans" ] }