diff --git a/merge-check.sh b/merge-check.sh index de57f12..873b60a 100644 --- a/merge-check.sh +++ b/merge-check.sh @@ -31,8 +31,9 @@ if ! command -v node >/dev/null 2>&1 || [ ! -d node_modules/@novox/mesh-sdk ]; t echo "NOT TESTED HERE: internal/runtime and internal/console launch bundles that need node and @novox/mesh-sdk" packages=$(printf '%s\n' "$packages" | grep -v -e '/internal/runtime$' -e '/internal/console$') # What of the console launches no bundle runs anyway: how an address resolves (novox/hq issue 287), - # and how a search ranks (novox/hq ADR 0245). - bundleless='^(TestASeatAndAModuleOfOneNameAreEachReached|TestSearch.*|TestASearchByCommandSaysWhatItReplaces)$' + # and how a search ranks (novox/hq ADR 0245), and what discovery heard and how a round that heard nothing is + # answered (novox/hq issue 554). + bundleless='^(TestASeatAndAModuleOfOneNameAreEachReached|TestSearch.*|TestASearchByCommandSaysWhatItReplaces|TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed|TestARuntimeHeardBeforeAndNotNowIsNamed|TestACallAfterARoundThatHeardNothingUsesWhatWasHeardBefore|TestAnAddressKnownOnlyFromAnEarlierAnswerSaysHowOld)$' # And what of the runtime launches no bundle: how a bundle's seat traffic reaches the bus, what its # environment holds, what a runtime on a module's own account serves (novox/hq ADR 0259), and that a write # by compare-and-set loses no update under contention (novox/hq research 042), and a module's log verbs diff --git a/node-tools/internal/announce/announce.go b/node-tools/internal/announce/announce.go index f013cad..6c941ea 100644 --- a/node-tools/internal/announce/announce.go +++ b/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..`) 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 diff --git a/node-tools/internal/announce/empty_round_test.go b/node-tools/internal/announce/empty_round_test.go new file mode 100644 index 0000000..9a62780 --- /dev/null +++ b/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()) + } +} diff --git a/node-tools/internal/announce/gather_test.go b/node-tools/internal/announce/gather_test.go index 27d836f..c5d8492 100644 --- a/node-tools/internal/announce/gather_test.go +++ b/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) } diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index ee80667..c1dd0e6 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -775,6 +775,21 @@ type Arrival struct { // that said it is there, not for a fixed moment (2026-10-05). Answers beyond what the caller has // taken are kept up to the channel's room; an empty answer is no answer. func (c *Conn) Collect(subject string, body []byte) (<-chan Arrival, func(), error) { + out, ask, stop, err := c.Collector(subject) + if err != nil { + return nil, func() {}, err + } + if err := ask(subject, body); err != nil { + stop() + return nil, func() {}, err + } + return out, stop, nil +} + +// Collector is one inbox for many requests: every answer to every request `ask` publishes arrives on +// the one channel, until stopped — for discovery, which asks each service that said it is there for +// what it serves, as each says so (novox/hq issue 554). `what` names the requests in a log line. +func (c *Conn) Collector(what string) (<-chan Arrival, func(subject string, body []byte) error, func(), error) { inbox := c.nc.NewRespInbox() out := make(chan Arrival, 1024) var mu sync.Mutex @@ -792,11 +807,11 @@ func (c *Conn) Collect(subject string, body []byte) (<-chan Arrival, func(), err select { case out <- a: default: - c.Logf("[mesh-tools] more answers to %s than are kept; one (%d bytes) dropped", subject, len(msg.Data)) + c.Logf("[mesh-tools] more answers to %s than are kept; one (%d bytes) dropped", what, len(msg.Data)) } }) if err != nil { - return nil, func() {}, err + return nil, nil, func() {}, err } stop := func() { _ = sub.Unsubscribe() @@ -804,11 +819,8 @@ func (c *Conn) Collect(subject string, body []byte) (<-chan Arrival, func(), err stopped = true mu.Unlock() } - if err := c.nc.PublishRequest(subject, inbox, body); err != nil { - stop() - return nil, func() {}, err - } - return out, stop, nil + ask := func(subject string, body []byte) error { return c.nc.PublishRequest(subject, inbox, body) } + return out, ask, stop, nil } // MaxPayload is the largest message the bus carries, as the server told this connection. diff --git a/node-tools/internal/console/address.go b/node-tools/internal/console/address.go index f216ddc..73ef7c7 100644 --- a/node-tools/internal/console/address.go +++ b/node-tools/internal/console/address.go @@ -134,6 +134,42 @@ type index struct { // Replaces is what each verb and tool replaces, from the controller's records (novox/hq ADR 0245): // `seat:.` and `.`. Replaces map[string][]string + // Earlier is every runtime not heard this round whose last answer, within RuntimeForgotten, this + // index holds instead (novox/hq issue 554), by runtimeKey: when it was heard. + Earlier map[string]time.Time + // Stale is every module on a machine (`mod:@`) and every seat's holder + // (`seat:@`) this index knows only from an earlier answer: from whom, and when. + Stale map[string]staleFrom +} + +// staleFrom is where an address known only from an earlier round of discovery was heard. +type staleFrom struct { + Runtime string + Machine string + At time.Time +} + +// say is the sentence a call or a description carries for it. +func (f staleFrom) say() string { + who := f.Runtime + if f.Machine != "" { + who += " on " + f.Machine + } + return fmt.Sprintf("found from what %s announced %s ago: it did not answer the latest round of discovery", + who, time.Since(f.At).Round(time.Second)) +} + +// earlier is a runtime's last answer, used for a round that did not hear it. +type earlier struct { + Heard announce.Heard + At time.Time +} + +func endpointKey(e announce.Endpoint) string { + if e.Kind == announce.KindSeat { + return "seat:" + e.Seat + "@" + e.Node + } + return "mod:" + e.Module + "@" + e.Node } // unheard is one runtime discovery did not hear in full, and why. @@ -237,7 +273,11 @@ type recordedMachine struct { // indexOn asks the mesh what it holds (novox/hq ADR 0197): what answers, from every runtime's own // announcement on the bus — one `$SRV.INFO` request — and what should, from the controller's records // read as JSON. Nothing is inferred from a roster and nothing is parsed from print. -func indexOn(conn *bus.Conn) (*index, error) { +// +// `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) { var wg sync.WaitGroup var nodesOut, modulesOut string var toolsAnswer json.RawMessage @@ -252,19 +292,44 @@ func indexOn(conn *bus.Conn) (*index, error) { toolsAnswer = got.Result } }() - d, err := announce.Gather(conn) + d, err := announce.Gather(conn, held) wg.Wait() if err != nil { return nil, err } var endpoints []announce.Endpoint + fresh := map[string]bool{} for _, h := range d.Heard { - endpoints = append(endpoints, announce.Endpoints(h.Info)...) + for _, e := range announce.Endpoints(h.Info) { + fresh[endpointKey(e)] = true + endpoints = append(endpoints, e) + } + } + // **A runtime not heard this round is not gone** (novox/hq issue 554): a round in which no answer + // crossed the laptop's link in time replaced a full index with an empty one, and every seat and + // module call was refused. What it announced last, within RuntimeForgotten, stands — said with its age. + stale := map[string]staleFrom{} + earlierAt := map[string]time.Time{} + if before != nil { + for _, b := range before(d) { + earlierAt[runtimeKey(b.Heard.Info.Name, b.Heard.Info.ID)] = b.At + for _, e := range announce.Endpoints(b.Heard.Info) { + k := endpointKey(e) + if fresh[k] { + continue + } + if _, have := stale[k]; !have { + stale[k] = staleFrom{Runtime: b.Heard.Info.Name, Machine: b.Heard.Machine(), At: b.At} + } + endpoints = append(endpoints, e) + } + } } x, announced, machines := indexOfEndpoints(endpoints) l := x.Listing x.Discovery = d + x.Stale, x.Earlier = stale, earlierAt for _, s := range d.Silent { x.Unheard = append(x.Unheard, unheard{Runtime: s.Name, Machine: s.Machine, Why: fmt.Sprintf("it answered PING; what it serves did not arrive within %s", announce.Patience)}) @@ -325,6 +390,10 @@ func containsHolder(hs []holder, h holder) bool { func (s *Surface) index() (*index, error) { return s.indexAsked(false) } // indexAsked is what the mesh answered, asked again when `fresh` or when the kept answer is old. +// +// **A round that heard nothing is not kept** (novox/hq issue 554): the index it gives holds what every +// runtime announced last, marked with its age, and the next question asks again rather than living +// with it for IndexKept. func (s *Surface) indexAsked(fresh bool) (*index, error) { s.mu.Lock() if !fresh && s.idx != nil && time.Since(s.idxAt) <= IndexKept { @@ -333,13 +402,17 @@ func (s *Surface) indexAsked(fresh bool) (*index, error) { return x, nil } s.mu.Unlock() - x, err := indexOn(s.conn) + x, err := indexOn(s.conn, s.held, s.before) if err != nil { return nil, err } s.mu.Lock() s.remember(x) - s.idx, s.idxAt = x, time.Now() + s.sayRound(x) + s.idx = x + if !x.Discovery.Empty() { + s.idxAt = time.Now() + } s.mu.Unlock() return x, nil } @@ -450,6 +523,26 @@ type target struct { Node string Tool Tool Seat bool + // Stale is set when the target is known only from an earlier round of discovery (novox/hq issue 554). + Stale *staleFrom +} + +// staleOf is where a target was heard, when every place it answers is known only from an earlier round. +func (x *index) staleOf(keys []string) *staleFrom { + if len(keys) == 0 || len(x.Stale) == 0 { + return nil + } + var first *staleFrom + for _, k := range keys { + f, ok := x.Stale[k] + if !ok { + return nil + } + if first == nil { + first = &f + } + } + return first } // resolve turns an address into exactly one target, or says why it cannot. @@ -487,12 +580,17 @@ func resolve(x *index, address string) (target, error) { return target{}, fmt.Errorf("%s is held once per machine: write /%s — it is held on %s", prefix, rest, orNobody(nodesOf(s.Holders))) } - return target{Address: node + "/" + rest, Key: "seat:" + rest + "@" + node, Name: rest, Node: node, Tool: *verb, Seat: true}, nil + return target{Address: node + "/" + rest, Key: "seat:" + rest + "@" + node, Name: rest, Node: node, Tool: *verb, Seat: true, + Stale: x.staleOf([]string{"seat:" + prefix + "@" + node})}, nil } if node != "" { return target{}, fmt.Errorf("%s is held once for the whole mesh: write %s, without a machine", prefix, rest) } - return target{Address: rest, Key: "seat:" + rest, Name: rest, Tool: *verb, Seat: true}, nil + var holders []string + for _, h := range s.Holders { + holders = append(holders, "seat:"+prefix+"@"+h.Node) + } + return target{Address: rest, Key: "seat:" + rest, Name: rest, Tool: *verb, Seat: true, Stale: x.staleOf(holders)}, nil } m := x.Modules[prefix] @@ -509,6 +607,11 @@ func resolve(x *index, address string) (target, error) { return target{}, fmt.Errorf("nothing that answered discovery is called %s, but not every runtime answered: %s — "+ "%s may be theirs. Ask again in a moment; mesh_runtimes shows which runtimes answered", prefix, sayUnheard(x.Unheard), prefix) } + if x.Discovery.Empty() { + return target{}, fmt.Errorf("nothing answered discovery at all within %s, not even PING, and this console holds no "+ + "earlier answer: whether anything is called %s is not known. Ask again in a moment; mesh_runtimes shows which "+ + "runtimes answered", x.Discovery.Waited.Round(time.Millisecond), prefix) + } return target{}, fmt.Errorf("nothing in the mesh is called %s: no seat, and no module that answers tools. "+ "mesh_search finds a tool by words", prefix) } @@ -521,7 +624,11 @@ func resolve(x *index, address string) (target, error) { return target{}, fmt.Errorf("%s keeps state on each machine it runs on, so a call names the machine: "+ "write /%s — it runs on %s", prefix, rest, orNobody(m.On)) } - return target{Address: rest, Key: rest, Name: rest, Tool: *tool}, nil + var on []string + for _, n := range m.On { + on = append(on, "mod:"+prefix+"@"+n) + } + return target{Address: rest, Key: rest, Name: rest, Tool: *tool, Stale: x.staleOf(on)}, nil } if len(m.On) > 0 && !contains(m.On, node) { if us := x.unheardOn(node); len(us) > 0 || contains(x.Recorded[prefix], node) { @@ -534,7 +641,8 @@ func resolve(x *index, address string) (target, error) { } return target{}, fmt.Errorf("%s does not run on %s; it runs on %s", prefix, node, orNobody(m.On)) } - return target{Address: node + "/" + rest, Key: rest + "@" + node, Name: rest, Node: node, Tool: *tool}, nil + return target{Address: node + "/" + rest, Key: rest + "@" + node, Name: rest, Node: node, Tool: *tool, + Stale: x.staleOf([]string{"mod:" + prefix + "@" + node})}, nil } // unheardOnAny is what is unheard on any of these machines. @@ -861,6 +969,9 @@ func (s *Surface) discover(name string, args map[string]any) map[string]any { if r := x.Replaces[key]; len(r) > 0 { described["instead_of"] = r } + if t.Stale != nil { + described["found"] = t.Stale.say() + } return answerText(described) case verbCall: @@ -880,12 +991,19 @@ func (s *Surface) discover(name string, args map[string]any) map[string]any { got, err = callTool(s.conn, t.Key, callArgs, nil, x.Listing) } if err != nil { - return failure(whyItFailed(t.Address, err)) + why := whyItFailed(t.Address, err) + if t.Stale != nil { + why += " (" + t.Stale.say() + ")" + } + return failure(why) } content := []map[string]any{{"type": "text", "text": pretty(got.Result)}} if got.Node != "" { content = append(content, map[string]any{"type": "text", "text": "answered by " + got.Node}) } + if t.Stale != nil { + content = append(content, map[string]any{"type": "text", "text": t.Stale.say()}) + } return map[string]any{"content": content} } return failure("no discovery verb " + name) diff --git a/node-tools/internal/console/client.go b/node-tools/internal/console/client.go index 2cc3e08..1b80b87 100644 --- a/node-tools/internal/console/client.go +++ b/node-tools/internal/console/client.go @@ -62,7 +62,7 @@ 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) + x, err := indexOn(conn, nil, nil) if err != nil { return nil, err } diff --git a/node-tools/internal/console/empty_round_test.go b/node-tools/internal/console/empty_round_test.go new file mode 100644 index 0000000..474dafa --- /dev/null +++ b/node-tools/internal/console/empty_round_test.go @@ -0,0 +1,151 @@ +package console + +import ( + "encoding/json" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/novox/mesh-tools/node-tools/internal/announce" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// novox/hq issue 554, reproduced: the laptop's console heard every runtime, then a round in which +// nothing answered discovery in time — and every seat and module call was refused ("it answered +// discovery 2s ago, not now") for the five seconds that round was kept, again and again. Now a call +// is resolved against what the runtime announced last, says how old that is, and reaches the tool; the +// empty round is logged once, is not kept, and the next question asks again. +func TestACallAfterARoundThatHeardNothingUsesWhatWasHeardBefore(t *testing.T) { + keptWas, patienceWas := IndexKept, announce.Patience + IndexKept, announce.Patience = 0, 1200*time.Millisecond + t.Cleanup(func() { IndexKept, announce.Patience = keptWas, patienceWas }) + mt.New(t) + + // The tools answer on their own connection; discovery is answered on another, which goes quiet. + tools := connect(t, "node-tools", "laptop") + stopTool, err := tools.HandleSubject("mesh.mod.slack.tool.slack_check.laptop", func(json.RawMessage) (any, error) { + return map[string]any{"ok": true}, nil + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(stopTool) + stopSeat, err := tools.HandleSubject("mesh.seat.issue-tracker.tool.list", func(json.RawMessage) (any, error) { + return map[string]any{"issues": []int{554}}, nil + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(stopSeat) + tools.Flush() + endpoints := []announce.Endpoint{ + {Kind: announce.KindTool, Module: "slack", Tool: "slack_check", Node: "laptop", Subject: "mesh.mod.slack.tool.slack_check.laptop"}, + {Kind: announce.KindSeat, Module: "issue-tracker", Seat: "issue-tracker", Tool: "list", Scope: "mesh", Node: "laptop", + Schema: json.RawMessage(`{}`), Subject: "mesh.seat.issue-tracker.tool.list"}, + } + announcing := connect(t, "node-tools", "laptop") + stopAnnounce, err := announce.Serve(announcing, announce.Service{Name: "node-tools", ID: "laptop", + Metadata: map[string]string{"node": "laptop"}}, func() []announce.Endpoint { return endpoints }) + if err != nil { + t.Fatal(err) + } + announcing.Flush() + + console := connect(t, "node-tools", "laptop-console") + var mu sync.Mutex + var logged []string + console.Logf = func(f string, a ...any) { + mu.Lock() + defer mu.Unlock() + logged = append(logged, fmt.Sprintf(f, a...)) + } + said := func(words string) int { + mu.Lock() + defer mu.Unlock() + n := 0 + for _, l := range logged { + if strings.Contains(l, words) { + n++ + } + } + return n + } + surface := NewSurface(console, "laptop.node-tools") + up, err := Serve(surface, "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer up.Close() + endpoint := "http://" + up.Address + "/mcp" + + if got, isErr := call(t, endpoint, "mesh_call", map[string]any{"address": "laptop/slack.slack_check"}); isErr || strings.Contains(got, "announced") { + t.Fatalf("while discovery hears the runtime: %s", got) + } + + // Discovery goes quiet: no PING arrives at all. + stopAnnounce() + announcing.Flush() + for _, address := range []string{"laptop/slack.slack_check", "issue-tracker.list"} { + got, isErr := call(t, endpoint, "mesh_call", map[string]any{"address": address}) + if isErr || !strings.Contains(got, "found from what node-tools on laptop announced") || + !strings.Contains(got, "ago: it did not answer the latest round of discovery") { + t.Errorf("%s after a round that heard nothing: %s", address, got) + } + } + if got, isErr := call(t, endpoint, "mesh_describe", map[string]any{"address": "issue-tracker.list"}); isErr || !strings.Contains(got, `"found"`) { + t.Errorf("describe after a round that heard nothing: %s", got) + } + if n := said("discovery heard nothing at all"); n != 1 { + t.Errorf("the empty round was logged %d time(s), want once: %v", n, logged) + } + // A runtime really gone: the call is sent on what it announced, and fails with the bus's own refusal, + // saying where the address was found. + stopTool() + tools.Flush() + if got, isErr := call(t, endpoint, "mesh_call", map[string]any{"address": "laptop/slack.slack_check"}); !isErr || + !strings.Contains(got, "nothing serves") || !strings.Contains(got, "found from what node-tools on laptop announced") { + t.Errorf("a gone runtime's tool: %s", got) + } + surface.mu.Lock() + kept := surface.idx != nil && !surface.idx.Discovery.Empty() && time.Since(surface.idxAt) < time.Second + surface.mu.Unlock() + if kept { + t.Error("a round that heard nothing was kept as the answer") + } + + // Discovery answers again: the calls are plain again, and that is logged once. + stopAgain, err := announce.Serve(announcing, announce.Service{Name: "node-tools", ID: "laptop", + Metadata: map[string]string{"node": "laptop"}}, func() []announce.Endpoint { return endpoints }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(stopAgain) + announcing.Flush() + if got, isErr := call(t, endpoint, "mesh_call", map[string]any{"address": "issue-tracker.list"}); isErr || strings.Contains(got, "announced") { + t.Errorf("once discovery hears the runtime again: %s", got) + } + if n := said("discovery hears every runtime again"); n != 1 { + t.Errorf("the recovery was logged %d time(s): %v", n, logged) + } +} + +// An address known only from an earlier answer says how old it is; one heard fresh on its machine is +// plain, even where the same module is known only from an earlier answer on another. +func TestAnAddressKnownOnlyFromAnEarlierAnswerSaysHowOld(t *testing.T) { + at := time.Now().Add(-20 * time.Second) + x := &index{Modules: map[string]*moduleInfo{"slack": {Module: "slack", On: []string{"laptop"}, + Tools: []Tool{{Module: "slack", Name: "slack_check"}}}}, Recorded: map[string][]string{}, + Stale: map[string]staleFrom{"mod:slack@laptop": {Runtime: "node-tools", Machine: "laptop", At: at}}, + Discovery: announce.Discovery{Pinged: 0}} + got, err := resolve(x, "laptop/slack.slack_check") + if err != nil || got.Stale == nil || !strings.Contains(got.Stale.say(), "announced 20s ago") { + t.Fatalf("%+v %v", got, err) + } + // A module heard fresh on one machine and only earlier on another: the fresh one is plain. + x.Modules["slack"].On = []string{"desk", "laptop"} + if got, err := resolve(x, "desk/slack.slack_check"); err != nil || got.Stale != nil { + t.Errorf("a fresh machine's module marked stale: %+v %v", got, err) + } +} diff --git a/node-tools/internal/console/mcp.go b/node-tools/internal/console/mcp.go index 0183e83..062d52c 100644 --- a/node-tools/internal/console/mcp.go +++ b/node-tools/internal/console/mcp.go @@ -52,6 +52,8 @@ type Surface struct { idxAt time.Time // runtimes is every runtime discovery has heard, as it last heard it (mesh_runtimes). runtimes map[string]*runtimeSeen + // roundSaid is what the last round logged was missing, so a round is logged once (novox/hq issue 554). + roundSaid string // 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 diff --git a/node-tools/internal/console/runtimes.go b/node-tools/internal/console/runtimes.go index 855b28f..6335a71 100644 --- a/node-tools/internal/console/runtimes.go +++ b/node-tools/internal/console/runtimes.go @@ -7,7 +7,9 @@ package console import ( "fmt" + "log" "sort" + "strings" "time" "github.com/novox/mesh-tools/node-tools/internal/announce" @@ -31,6 +33,11 @@ type runtimeSeen struct { Shortened string Answered bool // in the latest round Why string // why not, when it did not + // Reused is true when its PING said its announcement was unchanged, so it was not fetched again. + Reused bool + // Last is its last whole answer: what a round that does not hear it uses instead, and what an + // unchanged digest stands for (novox/hq issue 554). + Last announce.Heard } func runtimeKey(name, id string) string { return name + "/" + id } @@ -50,7 +57,108 @@ func seenOf(h announce.Heard, at time.Time) *runtimeSeen { } } return &runtimeSeen{Name: h.Info.Name, ID: h.Info.ID, Machine: h.Machine(), LastHeard: at, Took: h.Took, - Bytes: h.Bytes, Modules: len(modules), Tools: len(tools), SeatVerbs: len(verbs), Shortened: h.Shortened, Answered: true} + Bytes: h.Bytes, Modules: len(modules), Tools: len(tools), SeatVerbs: len(verbs), Shortened: h.Shortened, Answered: true, + Reused: h.Reused, Last: h} +} + +// held is the answer this console holds for an instance at this digest (announce.Known): an instance +// whose announcement is unchanged is not asked for it again (novox/hq issue 554). +func (s *Surface) held(name, id, digest string) (announce.Heard, bool) { + s.mu.Lock() + defer s.mu.Unlock() + r := s.runtimes[runtimeKey(name, id)] + if r == nil || r.Last.Digest == "" || r.Last.Digest != digest || r.Last.Info.Name == "" { + return announce.Heard{}, false + } + h := r.Last + h.Took, h.Reused = 0, false + return h, true +} + +// before is the last answer of every runtime this round did not hear, heard within RuntimeForgotten +// and not since restarted under another instance (novox/hq issue 554). +func (s *Surface) before(d announce.Discovery) []earlier { + heard := map[string]bool{} + running := map[string]bool{} + for _, h := range d.Heard { + heard[runtimeKey(h.Info.Name, h.Info.ID)] = true + if m := h.Machine(); m != "" { + running[h.Info.Name+"@"+m] = true + } + } + s.mu.Lock() + defer s.mu.Unlock() + var out []earlier + for k, r := range s.runtimes { + if heard[k] || r.Last.Info.Name == "" || r.LastHeard.IsZero() || time.Since(r.LastHeard) > RuntimeForgotten { + continue + } + if r.Machine != "" && running[r.Name+"@"+r.Machine] { + continue // restarted: the new instance answered + } + out = append(out, earlier{Heard: r.Last, At: r.LastHeard}) + } + sort.Slice(out, func(i, j int) bool { + return runtimeKey(out[i].Heard.Info.Name, out[i].Heard.Info.ID) < runtimeKey(out[j].Heard.Info.Name, out[j].Heard.Info.ID) + }) + return out +} + +// sayRound logs a round of discovery that did not hear every runtime — once, when what it missed +// changes, and once when every runtime is heard again — so the next case leaves evidence in the +// journal (novox/hq issue 554). Called with s.mu held, after remember. +func (s *Surface) sayRound(x *index) { + d := x.Discovery + var missed []string + for _, u := range x.Unheard { + who := u.Runtime + if u.Machine != "" { + who += " on " + u.Machine + } + missed = append(missed, who) + } + sort.Strings(missed) + shape := strings.Join(missed, ", ") + if d.Empty() { + shape = "empty|" + shape + } + if shape == s.roundSaid { + return + } + s.roundSaid = shape + reused := 0 + for _, h := range d.Heard { + if h.Reused { + reused++ + } + } + logf := log.Printf + if s.conn != nil && s.conn.Logf != nil { + logf = s.conn.Logf + } + window := fmt.Sprintf("waited %s (window %s, patience %s)", d.Waited.Round(time.Millisecond), announce.Window, announce.Patience) + held := "" + if len(x.Earlier) > 0 { + held = fmt.Sprintf("; what %d of them announced earlier is used, with its age", len(x.Earlier)) + } + switch { + case d.Empty(): + logf("[mesh-tools] discovery heard nothing at all: no PING arrived; %s; not heard: %s%s; asking again at the next question", + window, orNone(missed), held) + case len(missed) > 0: + logf("[mesh-tools] discovery did not hear every runtime: %d PING answer(s), %d announcement(s) (%d unchanged, not fetched); %s; not heard: %s%s", + d.Pinged, len(d.Heard), reused, window, shape, held) + default: + logf("[mesh-tools] discovery hears every runtime again: %d PING answer(s), %d announcement(s) (%d unchanged, not fetched); %s", + d.Pinged, len(d.Heard), reused, window) + } +} + +func orNone(xs []string) string { + if len(xs) == 0 { + return "none heard before either" + } + return strings.Join(xs, ", ") } // remember records a round of discovery and adds to the index every runtime heard lately that did not @@ -101,14 +209,25 @@ func (s *Surface) remember(x *index) { continue } r.Answered = false + used := "" + if at, ok := x.Earlier[k]; ok { + used = fmt.Sprintf("; what it announced %s ago is used", time.Since(at).Round(time.Second)) + } if why, ok := silent[k]; ok { r.Why = why + if used != "" { + for i := range x.Unheard { + if x.Unheard[i].Runtime == r.Name && x.Unheard[i].Machine == r.Machine && x.Unheard[i].Why == why { + x.Unheard[i].Why += used + } + } + } continue // already in the index's unheard } r.Why = "it did not answer this round of discovery" if time.Since(r.LastHeard) <= RuntimeForgotten { x.Unheard = append(x.Unheard, unheard{Runtime: r.Name, Machine: r.Machine, - Why: fmt.Sprintf("it answered discovery %s ago, not now", time.Since(r.LastHeard).Round(time.Second))}) + Why: fmt.Sprintf("it answered discovery %s ago, not now%s", time.Since(r.LastHeard).Round(time.Second), used)}) } } for _, u := range x.Discovery.Silent { @@ -157,6 +276,9 @@ func (s *Surface) runtimesAnswer() map[string]any { if r.Shortened != "" { entry["shortened, how"] = r.Shortened } + if r.Reused && r.Answered { + entry["announcement"] = "unchanged (its PING's digest), so not fetched again; took is its PING's" + } } if r.Answered { onMachine[r.Machine] = true @@ -184,7 +306,7 @@ func (s *Surface) runtimesAnswer() map[string]any { return answerText(map[string]any{ "asked": x.Discovery.At.Format(time.RFC3339), "waited": x.Discovery.Waited.Round(time.Millisecond).String(), - "window": fmt.Sprintf("at least %s; up to %s for a runtime that answered PING", announce.Window, announce.Patience), + "window": fmt.Sprintf("at least %s; up to %s for a runtime that answered PING, or until one does", announce.Window, announce.Patience), "answered": answered, "not heard": notHeard, }) diff --git a/node-tools/internal/console/runtimes_test.go b/node-tools/internal/console/runtimes_test.go index c82d48a..81bedf5 100644 --- a/node-tools/internal/console/runtimes_test.go +++ b/node-tools/internal/console/runtimes_test.go @@ -23,18 +23,22 @@ func slowRuntime(t *testing.T, conn *bus.Conn, node string, lag time.Duration, e 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(announce.Info(announce.Service{Name: id.Name, ID: id.ID, Metadata: id.Metadata}, endpoints)) return body - }) - if err != nil { - t.Fatal(err) } - t.Cleanup(stop) + for _, subject := range []string{"$SRV.INFO", "$SRV.INFO." + id.Name + "." + id.ID} { + stop, err := conn.Raw(subject, info) + if err != nil { + t.Fatal(err) + } + t.Cleanup(stop) + } conn.Flush() } @@ -175,7 +179,7 @@ func TestARuntimeHeardBeforeAndNotNowIsNamed(t *testing.T) { s.remember(before) now := &index{Modules: map[string]*moduleInfo{"nextcloud-client": {Module: "nextcloud-client", On: []string{"desktop"}, Tools: []Tool{{Module: "nextcloud-client", Name: "nextcloud_check"}}}}, Recorded: map[string][]string{}, - Discovery: announce.Discovery{At: time.Now().UTC()}} + Discovery: announce.Discovery{At: time.Now().UTC(), Pinged: 1}} // desktop's runtime answered s.remember(now) if len(now.Unheard) != 1 || now.Unheard[0].Machine != "laptop" { t.Fatalf("unheard: %+v", now.Unheard) @@ -190,7 +194,7 @@ func TestARuntimeHeardBeforeAndNotNowIsNamed(t *testing.T) { t.Errorf("%v", err) } // Nobody unheard: the plain answers stand. - quiet := &index{Modules: now.Modules, Recorded: map[string][]string{}} + quiet := &index{Modules: now.Modules, Recorded: map[string][]string{}, Discovery: now.Discovery} if _, err := resolve(quiet, "laptop/nextcloud-client.nextcloud_check"); err == nil || !strings.Contains(err.Error(), "does not run on laptop; it runs on desktop") { t.Errorf("%v", err) }