Keep what discovery heard when a round hears nothing, and fetch only changed announcements (issue 554)

A round in which no PING reached the laptop's console within the 750 ms window
was taken as the mesh's answer that nothing is there, and replaced the index:
every seat and module call was refused for five seconds at a time. Discovery
now waits up to its patience for a first PING, a round that heard nothing is
not kept, a runtime not heard this round is resolved from what it announced
last (said with its age), and a round that misses runtimes is logged once.
A PING carries a digest of what the instance serves, so an unchanged
announcement is not pulled again every round (some 880 kB to the laptop).

node-tools/internal/announce/announce.go

@@ -10,6 +10,8 @@
 package announce
 
 import (
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"strings"
 	"time"
@@ -35,6 +37,12 @@ var Window = 750 * time.Millisecond
 // serves is waited for until it arrives or this passes, and a service that never sends it is named.
 var Patience = 5 * time.Second
 
+// Digest is the PING answer's metadata key for a digest of what the instance serves now: a console
+// that holds the answer of that digest does not ask for it again (novox/hq issue 554). Every round
+// pulled every runtime's whole announcement — some 880 kB, every five seconds while a session works —
+// to the laptop, and the hundred-byte PING answers queued behind it.
+const Digest = "digest"
+
 // Shortened is the service metadata key a runtime sets when its announcement was cut to fit the bus.
 const Shortened = "shortened"
 
@@ -157,7 +165,14 @@ func Serve(conn *bus.Conn, s Service, current func() []Endpoint) (func(), error)
 		var v any
 		switch parts[1] {
 		case "PING":
-			v = micro.Ping{ServiceIdentity: s.identity(), Type: micro.PingResponseType}
+			id := s.identity()
+			md := map[string]string{}
+			for k, v := range id.Metadata {
+				md[k] = v
+			}
+			md[Digest] = DigestOf(Info(s, current()))
+			id.Metadata = md
+			v = micro.Ping{ServiceIdentity: id, Type: micro.PingResponseType}
 		case "INFO":
 			v = Info(s, current())
 		case "STATS":
@@ -205,6 +220,16 @@ func Serve(conn *bus.Conn, s Service, current func() []Endpoint) (func(), error)
 	}, nil
 }
 
+// DigestOf is a digest of an info_response: equal for equal announcements, whatever the moment.
+func DigestOf(i micro.Info) string {
+	body, err := json.Marshal(i)
+	if err != nil {
+		return ""
+	}
+	sum := sha256.Sum256(body)
+	return hex.EncodeToString(sum[:12])
+}
+
 // 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 {
@@ -244,6 +269,11 @@ type Heard struct {
 	Took      time.Duration
 	Bytes     int
 	Shortened string // why its descriptions are not whole, or ""
+	// Digest is the digest its PING gave for this answer, or "" when it gave none.
+	Digest string
+	// Reused is true when the instance's PING said it serves what the console already held, so its
+	// announcement was not asked for again; Took is then how long its PING took.
+	Reused bool
 }
 
 // Machine is the machine an instance runs on: its metadata's, or the one its endpoints name.
@@ -273,12 +303,24 @@ type Discovery struct {
 	Waited time.Duration
 	Heard  []Heard
 	Silent []Instance
+	// Pinged is how many PING answers arrived.
+	Pinged int
 }
 
-// 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
+// Empty is a round in which nothing arrived at all, not even a PING: no answer about the mesh, but a
+// round that failed (novox/hq issue 554).
+func (d Discovery) Empty() bool { return d.Pinged == 0 && len(d.Heard) == 0 }
+
+// Known is the announcement a console already holds for an instance at this digest, if any.
+type Known func(name, id, digest string) (Heard, bool)
+
+// Gather asks every service on the bus who it is (`$SRV.PING`), and each instance that answers what it
+// serves (`$SRV.INFO.<name>.<id>`) unless its PING's digest is one `known` holds; it waits at least the
+// window and at most the patience: until at least one instance answered PING and every instance that
+// did has said what it serves. A round in which no PING arrives within the window is not taken as the
+// mesh's answer that nothing is there (novox/hq issue 554): it waits on, up to the patience. An answer
 // that is not the protocol's is skipped.
-func Gather(conn *bus.Conn) (Discovery, error) {
+func Gather(conn *bus.Conn, known Known) (Discovery, error) {
 	start := time.Now()
 	d := Discovery{At: start.UTC()}
 	pings, stopPings, err := conn.Collect("$SRV.PING", nil)
@@ -286,7 +328,7 @@ func Gather(conn *bus.Conn) (Discovery, error) {
 		return d, err
 	}
 	defer stopPings()
-	infos, stopInfos, err := conn.Collect("$SRV.INFO", nil)
+	infos, askInfo, stopInfos, err := conn.Collector("$SRV.INFO")
 	if err != nil {
 		return d, err
 	}
@@ -296,8 +338,14 @@ func Gather(conn *bus.Conn) (Discovery, error) {
 	pinged := map[key]int{}
 	answered := map[key]int{}
 	where := map[key]string{}
+	digest := map[key]string{}
+	asked := map[key]bool{}
+	reused := map[key]int{} // index in d.Heard of an answer held already
 	var order []key
 	complete := func() bool {
+		if len(pinged) == 0 {
+			return false // nobody said they are there: not an answer, a round that has not arrived
+		}
 		for k, n := range pinged {
 			if answered[k] < n {
 				return false
@@ -305,6 +353,15 @@ func Gather(conn *bus.Conn) (Discovery, error) {
 		}
 		return true
 	}
+	ask := func(k key) {
+		if asked[k] {
+			return
+		}
+		asked[k] = true
+		if err := askInfo("$SRV.INFO."+k.name+"."+k.id, nil); err != nil {
+			conn.Logf("[mesh-tools] discovery could not ask %s/%s what it serves: %v", k.name, k.id, err)
+		}
+	}
 	least := time.NewTimer(Window)
 	defer least.Stop()
 	most := time.NewTimer(Patience)
@@ -317,21 +374,49 @@ func Gather(conn *bus.Conn) (Discovery, error) {
 			if json.Unmarshal(a.Data, &p) != nil || p.Type != micro.PingResponseType {
 				continue
 			}
+			d.Pinged++
 			k := key{p.Name, p.ID}
-			if _, seen := pinged[k]; !seen {
+			_, seen := pinged[k]
+			if !seen {
 				order = append(order, k)
 			}
 			pinged[k]++
 			if where[k] == "" {
 				where[k] = p.Metadata["node"]
 			}
+			if seen {
+				// Two instances under one name and id: what each serves is asked of both, never assumed.
+				if i, ok := reused[k]; ok {
+					d.Heard = append(d.Heard[:i], d.Heard[i+1:]...)
+					for kk, j := range reused {
+						if j > i {
+							reused[kk] = j - 1
+						}
+					}
+					delete(reused, k)
+					answered[k]--
+				}
+				ask(k)
+				continue
+			}
+			digest[k] = p.Metadata[Digest]
+			if h, ok := heldAt(known, k.name, k.id, digest[k]); ok {
+				h.Took, h.Reused, h.Digest = a.At.Sub(start), true, digest[k]
+				reused[k] = len(d.Heard)
+				d.Heard = append(d.Heard, h)
+				answered[k]++
+				break
+			}
+			ask(k)
 		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]})
+			k := key{i.Name, i.ID}
+			answered[k]++
+			d.Heard = append(d.Heard, Heard{Info: i, Took: a.At.Sub(start), Bytes: len(a.Data), Shortened: i.Metadata[Shortened],
+				Digest: digest[k]})
 		case <-least.C:
 			windowOver = true
 		case <-most.C:
@@ -350,6 +435,14 @@ func Gather(conn *bus.Conn) (Discovery, error) {
 	}
 }
 
+// heldAt is the answer `known` holds for an instance at this digest; never for a PING without one.
+func heldAt(known Known, name, id, digest string) (Heard, bool) {
+	if known == nil || digest == "" {
+		return Heard{}, false
+	}
+	return known(name, id, digest)
+}
+
 // Endpoints reads an info_response's endpoints back into what they announce.
 func Endpoints(i micro.Info) []Endpoint {
 	var out []Endpoint

node-tools/internal/announce/empty_round_test.go

@@ -0,0 +1,126 @@
+package announce
+
+import (
+	"encoding/json"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/micro"
+)
+
+// novox/hq issue 554: the laptop's console took a round in which not even a PING arrived within the
+// 750 ms window as the mesh's answer that nothing is there, and refused every seat and module call. A
+// PING that arrives after the window is waited for, up to the patience.
+func TestARoundWaitsForAPingThatArrivesAfterTheWindow(t *testing.T) {
+	was := Patience
+	Patience = 2500 * time.Millisecond
+	t.Cleanup(func() { Patience = was })
+
+	late := connect(t, "node-tools", "laptop")
+	s := Service{Name: "node-tools", ID: "laptop", Metadata: map[string]string{"node": "laptop"}}
+	ping, _ := json.Marshal(micro.Ping{ServiceIdentity: s.identity(), Type: micro.PingResponseType})
+	stop, err := late.Raw("$SRV.PING", func(string, []byte) []byte { time.Sleep(Window + 300*time.Millisecond); return ping })
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(stop)
+	body, _ := json.Marshal(Info(s, []Endpoint{{Kind: KindTool, Module: "slack", Tool: "slack_check", Node: "laptop", Subject: "mesh.mod.slack.tool.slack_check.laptop"}}))
+	for _, subject := range []string{"$SRV.INFO", "$SRV.INFO.node-tools.laptop"} {
+		stop, err := late.Raw(subject, func(string, []byte) []byte { return body })
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(stop)
+	}
+	late.Flush()
+
+	d, err := Gather(connect(t, "console", "laptop"), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if d.Empty() || len(d.Heard) != 1 || d.Heard[0].Machine() != "laptop" {
+		t.Fatalf("a PING after the window was not waited for: waited %s, pinged %d, heard %+v", d.Waited, d.Pinged, d.Heard)
+	}
+	if d.Waited < Window || d.Waited >= Patience {
+		t.Errorf("waited %s; the PING came after %s and patience is %s", d.Waited, Window+300*time.Millisecond, Patience)
+	}
+}
+
+// A round in which nothing at all arrives is said to be empty, after the patience — not answered as an
+// empty mesh at the window.
+func TestARoundThatHearsNothingIsEmptyAfterThePatience(t *testing.T) {
+	was := Patience
+	Patience = 1500 * time.Millisecond
+	t.Cleanup(func() { Patience = was })
+	d, err := Gather(connect(t, "console", "laptop"), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !d.Empty() || d.Waited < Patience {
+		t.Errorf("empty %v after %s (patience %s)", d.Empty(), d.Waited, Patience)
+	}
+}
+
+// What a runtime announces is fetched again only when its PING's digest changed: every round pulled
+// some 880 kB of announcements to the laptop, and the PING answers queued behind them (issue 554).
+func TestAnUnchangedAnnouncementIsNotFetchedAgain(t *testing.T) {
+	runtime := connect(t, "node-tools", "desk")
+	var endpoints atomic.Value
+	endpoints.Store([]Endpoint{{Kind: KindTool, Module: "alpha", Tool: "one", Node: "desk", Subject: "mesh.mod.alpha.tool.one.desk"}})
+	stop, err := Serve(runtime, Service{Name: "node-tools", ID: "desk", Metadata: map[string]string{"node": "desk"}},
+		func() []Endpoint { return endpoints.Load().([]Endpoint) })
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(stop)
+	runtime.Flush()
+	// Every request for what desk serves, counted where it is answered.
+	var asked atomic.Int32
+	spy := connect(t, "node-tools", "spy")
+	for _, subject := range []string{"$SRV.INFO", "$SRV.INFO.node-tools", "$SRV.INFO.node-tools.desk"} {
+		stop, err := spy.Raw(subject, func(string, []byte) []byte { asked.Add(1); return nil })
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(stop)
+	}
+	spy.Flush()
+
+	console := connect(t, "console", "laptop")
+	first, err := Gather(console, nil)
+	if err != nil || len(first.Heard) != 1 || first.Heard[0].Digest == "" || first.Heard[0].Reused {
+		t.Fatalf("first round: %v %+v", err, first.Heard)
+	}
+	if asked.Load() != 1 {
+		t.Errorf("the first round asked what desk serves %d time(s)", asked.Load())
+	}
+	held := first.Heard[0]
+	known := func(name, id, digest string) (Heard, bool) {
+		if name == held.Info.Name && id == held.Info.ID && digest == held.Digest {
+			return held, true
+		}
+		return Heard{}, false
+	}
+	asked.Store(0)
+	second, err := Gather(console, known)
+	if err != nil || len(second.Heard) != 1 || !second.Heard[0].Reused || len(Endpoints(second.Heard[0].Info)) != 1 {
+		t.Fatalf("an unchanged announcement: %v %+v", err, second.Heard)
+	}
+	if asked.Load() != 0 {
+		t.Errorf("an unchanged announcement was asked for %d time(s)", asked.Load())
+	}
+
+	endpoints.Store([]Endpoint{
+		{Kind: KindTool, Module: "alpha", Tool: "one", Node: "desk", Subject: "mesh.mod.alpha.tool.one.desk"},
+		{Kind: KindTool, Module: "alpha", Tool: "two", Node: "desk", Subject: "mesh.mod.alpha.tool.two.desk"},
+	})
+	third, err := Gather(console, known)
+	if err != nil || len(third.Heard) != 1 || third.Heard[0].Reused || len(Endpoints(third.Heard[0].Info)) != 2 ||
+		third.Heard[0].Digest == held.Digest {
+		t.Fatalf("a changed announcement was not fetched: %v %+v", err, third.Heard)
+	}
+	if asked.Load() != 1 {
+		t.Errorf("a changed announcement was asked for %d time(s)", asked.Load())
+	}
+}

node-tools/internal/announce/gather_test.go

@@ -33,18 +33,22 @@ func lagging(t *testing.T, conn *bus.Conn, s Service, lag time.Duration, endpoin
 		t.Fatal(err)
 	}
 	t.Cleanup(stop)
-	stop, err = conn.Raw("$SRV.INFO", func(string, []byte) []byte {
+	// Asked of every service, and of this instance by name and id — as discovery asks it (issue 554).
+	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)
+	for _, subject := range []string{"$SRV.INFO", "$SRV.INFO." + s.Name + "." + s.ID} {
+		stop, err := conn.Raw(subject, info)
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(stop)
+	}
 	conn.Flush()
 }
 
@@ -73,7 +77,7 @@ func TestDiscoveryWaitsForEveryRuntimeThatSaidItIsThere(t *testing.T) {
 
 	asker := connect(t, "console", "desk")
 	start := time.Now()
-	d, err := Gather(asker)
+	d, err := Gather(asker, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -109,7 +113,7 @@ func TestDiscoveryDoesNotWaitWhenEveryoneAnswered(t *testing.T) {
 	t.Cleanup(stop)
 	fast.Flush()
 	start := time.Now()
-	d, err := Gather(connect(t, "console", "desk"))
+	d, err := Gather(connect(t, "console", "desk"), nil)
 	if err != nil {
 		t.Fatal(err)
 	}