Wait for every runtime discovery hears from, and name the ones it misses

The console gathered $SRV.INFO answers for a fixed 750 ms. The laptop's runtime (48 modules, 341 endpoints, 164 kB) answers last every time: its answer crosses to the broker on another machine and back, a median of 365 ms on a quiet link and 813 ms in one of 25 rounds measured. When it missed the window the console said its modules ran nowhere ("nothing in the mesh is called slack") or only on another machine. Discovery now asks $SRV.PING alongside $SRV.INFO and waits, past the window and up to 5 s, for every instance that answered PING. A runtime that never sends what it serves, or answered before and not now, is named in every answer that might concern it instead of the module being called missing. An answer still too large after first-line descriptions drops them, and says so in its metadata. mesh_runtimes reports, per runtime, its machine, how long its answer took, its size, its modules and tools, whether it was shortened and when it was last heard, and the runtimes and machines not heard.
2026-10-05 15:36:10 +02:00
parent 8b789578c1
commit 379d19d907
9 changed files with 882 additions and 71 deletions
@@ -22,10 +22,22 @@ import (
 // Version is the announced service version (semver, as the protocol requires).
 const Version = "0.1.0"

-// Window is how long the console gathers discovery answers: every instance answers one request, and
-// how many will is what is being found out.
+// Window is how long the console gathers discovery answers at the least: every instance answers one
+// request, and how many will is what is being found out.
 var Window = 750 * time.Millisecond

+// Patience is how long the console waits, at the most, for the full answer of a service that has
+// said it is there. **A fixed window lost the largest runtime** (2026-10-05): the laptop's answer —
+// 48 modules, 341 endpoints, 164 kB — went up to the broker on another machine and back, arrived
+// last every time (a median of 365 ms, one in 25 after 813 ms on a quiet link, later still while the
+// runtime re-served), and when it missed the 750 ms window the console said its modules ran nowhere.
+// A PING answer is a hundred bytes and arrives at once, so who is there is known early; what each
+// serves is waited for until it arrives or this passes, and a service that never sends it is named.
+var Patience = 5 * time.Second
+
+// Shortened is the service metadata key a runtime sets when its announcement was cut to fit the bus.
+const Shortened = "shortened"
+
 // Kinds of endpoint.
 const (
 	KindTool = "tool" // a module's own tool
@@ -163,17 +175,10 @@ func Serve(conn *bus.Conn, s Service, current func() []Endpoint) (func(), error)
 			return nil
 		}
 		// Too large for the bus is no answer at all, and was silent: shorten every description to its
-		// first line and say so, rather than vanish.
+		// first line and say so, rather than vanish; still too large, leave descriptions out.
 		if limit := conn.MaxPayload(); limit > 0 && int64(len(body)) > limit {
 			if info, ok := v.(micro.Info); ok {
-				for i := range info.Endpoints {
-					info.Endpoints[i].Metadata["description"] = firstLine(info.Endpoints[i].Metadata["description"])
-				}
-				if shorter, err := json.Marshal(info); err == nil {
-					conn.Logf("[mesh-tools] what this runtime serves is %d bytes, beyond the bus's %d; announced with each description cut to its first line (%d bytes)",
-						len(body), limit, len(shorter))
-					body = shorter
-				}
+				body = shorten(conn.Logf, info, body, limit)
 			}
 		}
 		return body
@@ -200,22 +205,149 @@ func Serve(conn *bus.Conn, s Service, current func() []Endpoint) (func(), error)
 	}, nil
 }

-// Gather asks every service on the bus what it serves and answers what came back within the window.
-// An answer that is not an info_response is skipped.
-func Gather(conn *bus.Conn) ([]micro.Info, error) {
-	raw, err := conn.Gather("$SRV.INFO", nil, Window)
-	if err != nil {
-		return nil, err
+// shorten is an info_response cut to fit the bus: descriptions to their first line, then none, the
+// service's metadata saying which (so discovery can tell a person what it was given).
+func shorten(logf func(string, ...any), info micro.Info, body []byte, limit int64) []byte {
+	md := map[string]string{}
+	for k, v := range info.Metadata {
+		md[k] = v
 	}
-	var out []micro.Info
-	for _, b := range raw {
-		var i micro.Info
-		if json.Unmarshal(b, &i) != nil || i.Type != micro.InfoResponseType {
-			continue
+	info.Metadata = md
+	for _, step := range []string{"descriptions cut to their first line", "descriptions left out"} {
+		for i := range info.Endpoints {
+			e := info.Endpoints[i].Metadata
+			if step == "descriptions left out" {
+				delete(e, "description")
+			} else {
+				e["description"] = firstLine(e["description"])
+			}
+		}
+		md[Shortened] = step
+		shorter, err := json.Marshal(info)
+		if err != nil {
+			return body
+		}
+		logf("[mesh-tools] what this runtime serves is %d bytes, beyond the bus's %d; announced with %s (%d bytes)",
+			len(body), limit, step, len(shorter))
+		if int64(len(shorter)) <= limit {
+			return shorter
+		}
+		body = shorter
+	}
+	return body
+}
+
+// Heard is one service instance's answer to discovery: what it serves, how long it took to arrive,
+// and how large it was.
+type Heard struct {
+	Info      micro.Info
+	Took      time.Duration
+	Bytes     int
+	Shortened string // why its descriptions are not whole, or ""
+}
+
+// Machine is the machine an instance runs on: its metadata's, or the one its endpoints name.
+func (h Heard) Machine() string {
+	if n := h.Info.Metadata["node"]; n != "" {
+		return n
+	}
+	for _, e := range h.Info.Endpoints {
+		if n := e.Metadata["node"]; n != "" {
+			return n
+		}
+	}
+	return ""
+}
+
+// Instance names one service instance.
+type Instance struct {
+	Name    string
+	ID      string
+	Machine string
+}
+
+// Discovery is what one round of discovery heard, and who said it was there but did not say what it
+// serves in time.
+type Discovery struct {
+	At     time.Time
+	Waited time.Duration
+	Heard  []Heard
+	Silent []Instance
+}
+
+// Gather asks every service on the bus who it is and what it serves, and waits at least the window
+// and at most the patience: until every instance that answered PING has answered INFO. An answer
+// that is not the protocol's is skipped.
+func Gather(conn *bus.Conn) (Discovery, error) {
+	start := time.Now()
+	d := Discovery{At: start.UTC()}
+	pings, stopPings, err := conn.Collect("$SRV.PING", nil)
+	if err != nil {
+		return d, err
+	}
+	defer stopPings()
+	infos, stopInfos, err := conn.Collect("$SRV.INFO", nil)
+	if err != nil {
+		return d, err
+	}
+	defer stopInfos()
+
+	type key struct{ name, id string }
+	pinged := map[key]int{}
+	answered := map[key]int{}
+	where := map[key]string{}
+	var order []key
+	complete := func() bool {
+		for k, n := range pinged {
+			if answered[k] < n {
+				return false
+			}
+		}
+		return true
+	}
+	least := time.NewTimer(Window)
+	defer least.Stop()
+	most := time.NewTimer(Patience)
+	defer most.Stop()
+	windowOver := false
+	for {
+		select {
+		case a := <-pings:
+			var p micro.Ping
+			if json.Unmarshal(a.Data, &p) != nil || p.Type != micro.PingResponseType {
+				continue
+			}
+			k := key{p.Name, p.ID}
+			if _, seen := pinged[k]; !seen {
+				order = append(order, k)
+			}
+			pinged[k]++
+			if where[k] == "" {
+				where[k] = p.Metadata["node"]
+			}
+		case a := <-infos:
+			var i micro.Info
+			if json.Unmarshal(a.Data, &i) != nil || i.Type != micro.InfoResponseType {
+				continue
+			}
+			answered[key{i.Name, i.ID}]++
+			d.Heard = append(d.Heard, Heard{Info: i, Took: a.At.Sub(start), Bytes: len(a.Data), Shortened: i.Metadata[Shortened]})
+		case <-least.C:
+			windowOver = true
+		case <-most.C:
+			d.Waited = time.Since(start)
+			for _, k := range order {
+				if answered[k] < pinged[k] {
+					d.Silent = append(d.Silent, Instance{Name: k.name, ID: k.id, Machine: where[k]})
+				}
+			}
+			return d, nil
+		}
+		if windowOver && complete() {
+			d.Waited = time.Since(start)
+			return d, nil
 		}
-		out = append(out, i)
 	}
-	return out, nil
 }

 // Endpoints reads an info_response's endpoints back into what they announce.
@@ -0,0 +1,156 @@
+package announce
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/micro"
+
+	"github.com/novox/mesh-tools/node-tools/internal/bus"
+	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
+)
+
+func connect(t *testing.T, module, node string) *bus.Conn {
+	t.Helper()
+	c, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: module, Node: node})
+	if err != nil {
+		t.Fatal(err)
+	}
+	c.Logf = func(string, ...any) {}
+	t.Cleanup(c.Close)
+	return c
+}
+
+// lagging answers PING at once and INFO after `lag` — or never, with a negative lag: a runtime whose
+// one large answer is slow to cross the bus, as the laptop's was (2026-10-05).
+func lagging(t *testing.T, conn *bus.Conn, s Service, lag time.Duration, endpoints []Endpoint) {
+	t.Helper()
+	ping, _ := json.Marshal(micro.Ping{ServiceIdentity: s.identity(), Type: micro.PingResponseType})
+	stop, err := conn.Raw("$SRV.PING", func(string, []byte) []byte { return ping })
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(stop)
+	stop, err = conn.Raw("$SRV.INFO", func(string, []byte) []byte {
+		if lag < 0 {
+			return nil
+		}
+		time.Sleep(lag)
+		body, _ := json.Marshal(Info(s, endpoints))
+		return body
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(stop)
+	conn.Flush()
+}
+
+// A runtime whose answer arrives after the window is waited for, because it said it was there; one
+// that never says what it serves is named, not dropped. With the fixed 750 ms window the laptop's
+// modules were said to run nowhere whenever its answer was late (2026-10-05).
+func TestDiscoveryWaitsForEveryRuntimeThatSaidItIsThere(t *testing.T) {
+	was := Patience
+	Patience = 2500 * time.Millisecond
+	t.Cleanup(func() { Patience = was })
+
+	fast := connect(t, "node-tools", "desk")
+	stopFast, err := Serve(fast, Service{Name: "node-tools", ID: "desk", Metadata: map[string]string{"node": "desk"}}, func() []Endpoint {
+		return []Endpoint{{Kind: KindTool, Module: "alpha", Tool: "one", Node: "desk", Subject: "mesh.mod.alpha.tool.one.desk"}}
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(stopFast)
+	fast.Flush()
+	slow := connect(t, "node-tools", "laptop")
+	lagging(t, slow, Service{Name: "node-tools", ID: "laptop", Metadata: map[string]string{"node": "laptop"}}, Window+400*time.Millisecond,
+		[]Endpoint{{Kind: KindTool, Module: "slack", Tool: "slack_check", Node: "laptop", Subject: "mesh.mod.slack.tool.slack_check.laptop"}})
+	mute := connect(t, "node-tools", "sleeper")
+	lagging(t, mute, Service{Name: "node-tools", ID: "sleeper", Metadata: map[string]string{"node": "sleeper"}}, -1, nil)
+
+	asker := connect(t, "console", "desk")
+	start := time.Now()
+	d, err := Gather(asker)
+	if err != nil {
+		t.Fatal(err)
+	}
+	took := time.Since(start)
+
+	heard := map[string]Heard{}
+	for _, h := range d.Heard {
+		heard[h.Info.ID] = h
+	}
+	if h, ok := heard["laptop"]; !ok {
+		t.Fatalf("the slow runtime was not heard: %+v", d.Heard)
+	} else if h.Took <= Window || h.Machine() != "laptop" || h.Bytes == 0 {
+		t.Errorf("the slow runtime's answer: took %s (window %s), machine %q, %d bytes", h.Took, Window, h.Machine(), h.Bytes)
+	}
+	if _, ok := heard["desk"]; !ok {
+		t.Errorf("the fast runtime was not heard: %+v", d.Heard)
+	}
+	if len(d.Silent) != 1 || d.Silent[0].ID != "sleeper" || d.Silent[0].Machine != "sleeper" {
+		t.Errorf("the runtime that never said what it serves is not named: %+v", d.Silent)
+	}
+	if took < Patience || took > Patience+time.Second {
+		t.Errorf("waited %s with a silent runtime; patience is %s", took, Patience)
+	}
+}
+
+// With nobody slow, discovery takes the window and no longer.
+func TestDiscoveryDoesNotWaitWhenEveryoneAnswered(t *testing.T) {
+	fast := connect(t, "node-tools", "desk")
+	stop, err := Serve(fast, Service{Name: "node-tools", ID: "desk"}, func() []Endpoint { return nil })
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(stop)
+	fast.Flush()
+	start := time.Now()
+	d, err := Gather(connect(t, "console", "desk"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if took := time.Since(start); took > Window+500*time.Millisecond || len(d.Silent) != 0 || len(d.Heard) == 0 {
+		t.Errorf("took %s, heard %d, silent %+v", took, len(d.Heard), d.Silent)
+	}
+}
+
+// Too large even with first lines, descriptions are left out — and the answer says which, without
+// touching the service's own metadata.
+func TestAnAnswerTooLargeIsShortenedUntilItFits(t *testing.T) {
+	long := strings.Repeat("A sentence that is long. ", 20)
+	var endpoints []Endpoint
+	for i := 0; i < 50; i++ {
+		endpoints = append(endpoints, Endpoint{Kind: KindTool, Module: "m", Tool: "t" + strings.Repeat("x", i), Node: "laptop",
+			Description: long, Subject: "mesh.mod.m.tool.t"})
+	}
+	own := map[string]string{"node": "laptop"}
+	info := Info(Service{Name: "node-tools", ID: "laptop", Metadata: own}, endpoints)
+	full, _ := json.Marshal(info)
+	firstLines := 0
+	for range endpoints {
+		firstLines += len(firstLine(long))
+	}
+	limit := int64(len(full) - len(long)*len(endpoints) + firstLines/2) // first lines alone do not fit
+	var said []string
+	got := shorten(func(f string, a ...any) { said = append(said, f) }, info, full, limit)
+	if int64(len(got)) > limit {
+		t.Fatalf("still %d bytes, beyond %d", len(got), limit)
+	}
+	var back micro.Info
+	if err := json.Unmarshal(got, &back); err != nil {
+		t.Fatal(err)
+	}
+	if back.Metadata[Shortened] != "descriptions left out" || back.Metadata["node"] != "laptop" {
+		t.Errorf("metadata: %v", back.Metadata)
+	}
+	if _, has := own[Shortened]; has {
+		t.Error("the service's own metadata was changed")
+	}
+	if len(said) != 2 {
+		t.Errorf("each step is said: %v", said)
+	}
+}