11 Commits
Author SHA1 Message Date
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 failed (7): ace: novox/mesh-catalog#612 does not merge into the trunk: files in conflict: module…; and 6 more | 07f6078050ec on 84ede95c4af7
mesh/delivery-group group feat/registry-milestone rejected: composed together: the checker's verdict at the members' heads (3a038b9f8b4d): ace
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
38 changed files with 1334 additions and 111 deletions
+7 -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"
@@ -86,12 +88,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)
}
+7 -5
View File
@@ -7,16 +7,18 @@ require (
golang.org/x/sys v0.48.0
)
require git.novox.be/novox/mesh-contracts/go v0.0.0-20261104174344-ae5266e77639 // indirect
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.1.14-0.20261104174359-05b3e2e1b94f
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
)
+12 -10
View File
@@ -1,15 +1,17 @@
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=
git.novox.be/novox/mesh-contracts/go v0.0.0-20261104174344-ae5266e77639 h1:Zt7CfWt2S2W/fVgWNyW2u2+xwz051LXvGxVTNIyfhdM=
git.novox.be/novox/mesh-contracts/go v0.0.0-20261104174344-ae5266e77639/go.mod h1:0dD1XEJBJ0G5UYDQxSZI84nuujb6m3tOpJl4C8rQVvg=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f h1:jDsfDMd48MXY9Zershc3eRxnhmx2sSpUfQkLq5jAIds=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f/go.mod h1:08OomDXdwVN8/87nyDtv5R3rH/6LoP8SFFXz30bTYCM=
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 +25,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"
+91 -2
View File
@@ -24,6 +24,7 @@ import (
"regexp"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/nats-io/nats.go"
@@ -1077,21 +1078,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() {
+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
+3
View File
@@ -213,6 +213,9 @@ func TestAnAskerReadsItsOwnRecordAndNoOther(t *testing.T) {
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 {
+1 -1
View File
@@ -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")
+43 -40
View File
@@ -38,13 +38,15 @@ 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.
@@ -89,13 +91,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
@@ -248,7 +259,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 +269,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 {
@@ -345,30 +356,19 @@ 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>` (delivery.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)
func seatHoldersOn(src Source) (map[string][]byte, error) {
return src.State(delivery.SeatHoldersRead)
}
// The indexer's states the console reads (novox/hq ADR 0571 clauses 4 and 5): every module, and each manifest by its
@@ -390,8 +390,8 @@ 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
// 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(delivery.PlacementRead)
if err != nil {
why = append(why, delivery.PlacementRead+" (where modules are placed could not be read, so what is missing cannot "+
"be said: "+err.Error()+")")
@@ -408,7 +408,7 @@ func recordsOn(conn *bus.Conn) (recorded []recordedModule, machines []string, wh
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()+")")
}
@@ -419,7 +419,7 @@ func recordsOn(conn *bus.Conn) (recorded []recordedModule, machines []string, wh
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)
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 {
@@ -432,18 +432,18 @@ func recordsOn(conn *bus.Conn) (recorded []recordedModule, machines []string, wh
}
// 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 := delivery.DecodeManifest(entry)
if err != nil {
return manifestFacts{}, err
}
@@ -501,7 +501,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 +926,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)
}
+2 -2
View File
@@ -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,7 +125,7 @@ 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)
}
+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 nothing else.
wantConsumes := []string{"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)
}
}
+7 -4
View File
@@ -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,7 +72,7 @@ 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() {
@@ -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
}
+1 -1
View File
@@ -123,7 +123,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)
}
@@ -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)
}
+200
View File
@@ -0,0 +1,200 @@
// 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"
"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"
)
// 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>`: delivery.PlacementRead, delivery.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 := 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
}
@@ -0,0 +1,155 @@
//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, _ := strings.Cut(c, ".views.")
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,168 @@
package firstview
import (
"context"
"encoding/json"
"errors"
"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)
}
}
@@ -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 {
+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")
}
}
+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
+11 -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,17 @@
]
},
"state": [
"bound"
"bound",
"findings"
],
"consumes": [
"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"
]
}