Author SHA1 Message Date
jschoubben bd555e6014 No client names the holder of the seat mesh-provisioner: each reader keeps its own constant, held to its manifest (ADR 0625 section 4, issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 19:36:21 +01:00
jschoubben c3754e48c8 No client names a holder: each reader keeps its own constant, held to its manifest (ADR 0625 section 4, issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 18:37:28 +01:00
jschoubben a00c780352 Require every mesh package of the group at the floor version (ADR 0573 sections 2 and 4, issue 1966)
A member's go.mod names the oldest version a module accepts; the check builds the group's own heads whatever it names. go.sum lines of mesh packages are left out.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 16:56:39 +00:00
jschoubben 5bbe669cfb Merge b6/registry into the registry milestone's join: the console reads through the assigner's client, with the seat name mesh-module-assigner (issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 15:39:17 +01:00
jschoubben c55b1bcf38 First views: the integration test reads the views families of the manifest's consumes, not the seat list's events (T-3; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 15:37:17 +01:00
jschoubben 6965119bce Tool runner: the grants part's membership bucket and key are membership's, which the SDK keeps (ADR 0635 §4; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 15:36:18 +01:00
jschoubben 387f547346 Tool runner: the console, its test bed and the provider read placement, the catalogue's keys and the provisions through the seats' clients, not go/delivery, go/provider or go/provisions (T-1, ADR 0634 §4; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 15:31:22 +01:00
jschoubben fce9189579 First views: a finding's times are times, as the conditions client types them (issue 1966) 2026-11-05 15:26:49 +01:00
jschoubben a2402c4873 Merge branch 'b6/registry' into feat/registry-milestone 2026-11-05 14:54:25 +01:00
jschoubben f917fc60c1 Tool runner: waits for the first read of the seat list on the reader's Ready, to a bound, and serves with the bootstrapped seats past it (B9c, ADR 0636 §1; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 14:49:52 +01:00
jschoubben 0a6fd0cb5c Merge b6/registry into the registry milestone's join (issue 1966) 2026-11-05 14:39:37 +01:00
jschoubben a918352ff1 Tool runner: describes the seats from the seat list it reads through the seat mesh-catalog; the first views' finding in the conditions seat's layout (T-1, T-3, ADR 0635 §1, ADR 0636 §4; issue 1966) 2026-11-05 14:29:02 +01:00
jschoubben d5e4e4e8ab Console: the address text and the tests name the seat mesh-module-assigner (ADR 0634 §9; T-2; issue 1966) 2026-11-05 14:17:14 +01:00
jschoubben 07f6078050 Pin mesh-sdk 05b3e2e1b94f and mesh-contracts ae5266e77639, the group's final heads (issue 1966)
mesh/delivery ready: it is merged on the Approve
mesh/check checking 07f6078050ec on 84ede95c4af7 | novox/mesh-tools#61 in group feat/registry-milestone
mesh/delivery-group group feat/registry-milestone checking: 8 of 8 member(s) ready
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 18:47:39 +01:00
jschoubben e40cb291f8 Pin mesh-sdk e5d89c5eedb5 and mesh-contracts 347a3053580a, the group's heads after review round 2 (issue 1966)
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 18:26:33 +01:00
jschoubben e0d7122567 TLS-5 integration test: the owners' own views streams, the runner's consumers on them and a membership naming them, events on views.Emit's subject
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 18:09:09 +01:00
jschoubben c637da2eb7 Pin mesh-sdk c54648e74346 and mesh-contracts ec9529022c44, the group's heads (issue 1966)
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 17:46:08 +01:00
jschoubben 7390652f33 One guarded handler slot for a bundle's event and heartbeat handlers, so a reload never races the consumer's loop
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 16:28:08 +01:00
jschoubben db3b5fe63d F-tools: the runner's views consumer is proven on the real bus; the heartbeat is asked of the bus only when a bundle wants it, with no race; a seat record says when the bus wrote it; the beat test counts ticks, not sleeps
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 16:26:50 +01:00
jschoubben 76e025dbbc J2: a state read carries the time the bus wrote the value, as at (stdio.StateEntry.At)
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 14:31:36 +01:00
jschoubben f87c9720ed J2: the tool runner tells a subscribed bundle how its module's consumer stands, as mesh/heartbeat (stdio.OnHeartbeat), at each idle beat and once Missed (ADR 0615, ADR 0621, issue 1966)
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 14:10:25 +01:00
jschoubben d71cd5cef6 TLS-5, TLS-6: the views package is mesh-sdk's (go/views), and mesh_first_view_again is recorded through mesh-hand-acts before it is done (ADR 0615 §6, ADR 0565 §10, issue 1966)
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 12:14:14 +01:00
jschoubben 31247dff6b TLS-5, TLS-6: the tool runner holds the indexer's manifests and modules and the assigner's placement and seat-holders from their first views and views events, never their state, and the console gains mesh_first_views and mesh_first_view_again (ADR 0615, ADR 0621, issue 1966)
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 12:06:20 +01:00
jschoubben 8b1ab467a9 TLS-1, TLS-2, TLS-3: the runner declares its own secret ledgers, refuses a bundle tool named as a ledger repair, and states the capabilities of its calls in its PING (issue 1966)
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 11:40:03 +01:00
50 changed files with 1711 additions and 221 deletions
+9 -1
View File
@@ -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)
}
+73
View File
@@ -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
}
@@ -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()
}
})
}
}
+14 -5
View File
@@ -3,20 +3,29 @@ module github.com/novox/mesh-tools/node-tools
go 1.26.0
require (
git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/provisioning/go/seats/meshprovisioner v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner 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
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
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
)
+8 -10
View File
@@ -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=
+5
View File
@@ -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"
+95 -7
View File
@@ -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
// `<node>.<module>` (novox/hq ADR 0571 clause 7 and 10, mesh-sdk buscompose.GrantsMembershipsBucket), and nowhere else:
// `<node>.<module>` (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() {
+1 -1
View File
@@ -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()
+94
View File
@@ -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)
}
}
+8
View File
@@ -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 }
}
+1 -1
View File
@@ -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) {
+7 -1
View File
@@ -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
+5 -2
View File
@@ -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)
}
+3 -1
View File
@@ -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).
@@ -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 {
+2 -2
View File
@@ -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")
+82 -61
View File
@@ -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/placement/go/clients/meshmoduleassignerclient"
"git.novox.be/novox/mesh-contracts/placement/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: `<seat>.<verb>` for a seat held once for the mesh (e.g. `mesh-placement.placement`); " +
const grammar = "Addresses: `<seat>.<verb>` for a seat held once for the mesh (e.g. `mesh-module-assigner.placement`); " +
"`<node>/<seat>.<verb>` for a seat every machine holds (e.g. `ace/node-packet-filter.rules`); " +
"`<node>/<module>.<tool>` for a module on one machine (e.g. `novox/postgres.postgres_list_databases`); " +
"and `<module>.<tool>` 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 placement 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
@@ -137,8 +151,8 @@ type index struct {
// 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 `<seat>` and `<seat>.<node>`, 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 `<seat>` and `<seat>.<node>`, 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):
@@ -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
@@ -258,9 +272,9 @@ func indexOn(conn *bus.Conn, held announce.Known, before func(announce.Discovery
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) }()
go func() { defer wg.Done(); recorded, known, recordsWhy = recordsOn(src) }()
// 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) }()
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 placement 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
// placement and seat-holders, the indexer's modules and manifests, by `<owner>.<state>` (firstview.PlacementRead 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
@@ -388,38 +391,38 @@ 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
// and every machine placement names (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) {
placements, err := src.State(firstview.PlacementRead)
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.PlacementRead+" (where modules are placed could not be read, so what is missing cannot "+
"be said: "+err.Error()+")")
}
machines = delivery.MachineNames(placements)
machines = assigner.MachineNames(placements)
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)
assembled, err := assigner.PlacementsOf(placements)
if err != nil {
why = append(why, delivery.PlacementRead+" (not every key is a placement: "+err.Error()+")")
why = append(why, firstview.PlacementRead+" (not every key is a placement: "+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,24 +434,39 @@ 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)
@@ -456,19 +474,19 @@ func manifestOf(conn *bus.Conn, digest string) (manifestFacts, error) {
}
// 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 {
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
}
@@ -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)
}
+14 -14
View File
@@ -71,9 +71,9 @@ 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 {
@@ -82,8 +82,8 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) {
t.Cleanup(stopNodes)
// What the mesh holds, as the indexer and placement 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>/node-shelf.list") ||
if isErr || !strings.Contains(overview, "mesh-module-assigner.nodes") || !strings.Contains(overview, "<node>/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"`) {
@@ -168,7 +168,7 @@ func TestTheMeshsToolsAreFoundByAddress(t *testing.T) {
for address, want := range map[string]string{
"beta.three": "keeps state on each machine it runs on, so a call names the machine: write <node>/beta.three — it runs on desk",
"node-shelf.list": "held once per machine: write <node>/node-shelf.list — it is held on desk",
"desk/mesh-placement.nodes": "held once for the whole mesh",
"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",
} {
@@ -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)
+2 -2
View File
@@ -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
}
+2 -2
View File
@@ -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)
}
}
@@ -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)
+65
View File
@@ -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")
@@ -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: "placement", State: "placement", 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, "placement") {
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))
}
}
}
@@ -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.placement", "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)
}
}
+11 -8
View File
@@ -6,8 +6,8 @@ package console
import (
"context"
"encoding/json"
"git.novox.be/novox/mesh-contracts/placement/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
@@ -69,13 +72,13 @@ const MethodSeats = "mesh/seats"
// seatsAnswer is MethodSeats' answer: mesh-sdk's seat definitions, with each seat's holders by placement.
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 {
@@ -91,8 +94,8 @@ func (s *Surface) seatsAnswer() map[string]any {
}
// 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
}
@@ -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 <module>.<tool>."
}
return "The tools of a Novox mesh, reached as " + s.who + ", found by address rather than listed " +
@@ -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 {
@@ -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)
}
}
}
}
+16 -12
View File
@@ -8,11 +8,13 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
assigner "git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
seatplacement "git.novox.be/novox/mesh-contracts/placement/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"
)
@@ -60,14 +62,16 @@ func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T)
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 placement (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(seatplacement.SeatHolders{V: assigner.Version, Seat: "mesh-wants", Scope: "mesh",
Holders: []seatplacement.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)
}
+6 -6
View File
@@ -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
}
@@ -57,13 +57,13 @@ 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",
"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-placement": "anchor/node-service-manager.restart",
"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",
@@ -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" {
@@ -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())) }
@@ -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
}
+3 -3
View File
@@ -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)
}
+230
View File
@@ -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
// placement 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 `<owner>.<tool>`).
const (
Indexer = "mesh-delivery-indexer"
Assigner = "mesh-delivery-assigner"
)
// The names the console reads the four states by (`<owner>.<state>`), as the manifest's `consumes` names their views
// families (no seat's client names a holder, novox/hq ADR 0625 §4; TestTheFirstViewsAreTheManifestsViews).
const (
PlacementRead = Assigner + ".placement"
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, "placement", "placement"},
{Assigner, "seat-holders", "seat-holders"},
{Indexer, "manifests", "manifests"},
{Indexer, "modules", "modules"},
}
// Name is the name a state is read by, `<owner>.<state>`: PlacementRead, 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, `<owner>.<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 (`<owner>.<state>`), 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 `<owner>.<state>`, 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"
@@ -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.<owner>.event.views.<state>`, 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 + ".placement"); 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" }
@@ -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{"placement": {"anchor": `{"p":1}`}, "seat-holders": {"seat": `{"h":1}`},
"manifests": {"d1": `{"m":1}`}, "modules": {"web": `{"w":1}`}},
rev: map[string]uint64{"placement": 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.placement"); 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("placement")
o.say(t, "placement", "laptop", `{"p":2}`, 5, false)
o.say(t, "placement", "laptop", `{"p":0}`, 5, false)
o.say(t, "placement", "anchor", "", 6, false)
// (a deleted mark carries no value: Deleted is the encoding's, set below)
if got, _ := v.Entry("mesh-delivery-assigner.placement", "laptop"); string(got) != `{"p":2}` {
t.Errorf("the next revision is not applied: %q", got)
}
if n := asked("placement"); 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, "placement", "x", `{"p":9}`, 9, false)
eventually(t, "a gap did not ask the first view again", func() bool { return asked("placement") > before })
// The status says owner, tool, revision and the asks; the hand ask asks every view again.
var placement views.Status
for _, s := range v.Status() {
if s.State == "placement" {
placement = s
}
}
if placement.Owner != "mesh-delivery-assigner" || placement.Tool != "placement" || placement.Asks < 2 || placement.Answered.IsZero() {
t.Errorf("the status of placement is %+v", placement)
}
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{PlacementRead: 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)
}
}
@@ -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)
}
}
+16 -3
View File
@@ -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 {
+38 -23
View File
@@ -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/placement/go/clients/meshmoduleassignerclient"
seatplacement "git.novox.be/novox/mesh-contracts/placement/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 `<node>.<module>` (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,14 +153,26 @@ 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"
// PlacementBucket and SeatHoldersBucket are the assigner's states placement and seat-holders.
PlacementBucket = assignerHolder + "_placement"
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.
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 + ".placement", Bucket: PlacementBucket},
{Name: indexer + ".modules", Bucket: ModulesBucket},
{Name: indexer + ".manifests", Bucket: ManifestsBucket},
{Name: assignerHolder + ".seat-holders", Bucket: SeatHoldersBucket},
}
}
@@ -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, placement := kv(ModulesBucket), kv(ManifestsBucket), kv(PlacementBucket)
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,12 +210,12 @@ 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 := seatplacement.Placement{V: assigner.Version, Node: node}
for _, module := range placed {
p.Modules = append(p.Modules, delivery.PlacedModule{Module: module})
p.Modules = append(p.Modules, seatplacement.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.PlacementValues(p)
if err != nil {
t.Fatal(err)
}
+11 -7
View File
@@ -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,
+11 -11
View File
@@ -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] {
+27 -4
View File
@@ -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)
}
}
+28
View File
@@ -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.
@@ -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")
}
}
@@ -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)
}
+15
View File
@@ -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"
@@ -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()
}
}
})
}
}
}
+2 -2
View File
@@ -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)
+86 -11
View File
@@ -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
@@ -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)
+13 -7
View File
@@ -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.placement",
"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"
]
}