mesh-tools/node-tools/internal/console/runtimes_test.go

package console

import (
	"encoding/json"
	"fmt"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"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"
)

// slowRuntime answers PING at once and INFO after `lag`, or never with a negative lag.
func slowRuntime(t *testing.T, conn *bus.Conn, node string, lag time.Duration, endpoints []announce.Endpoint) {
	t.Helper()
	id := micro.ServiceIdentity{Name: "node-tools", ID: node, Version: announce.Version, Metadata: map[string]string{"node": node}}
	ping, _ := json.Marshal(micro.Ping{ServiceIdentity: id, Type: micro.PingResponseType})
	stop, err := conn.Raw("$SRV.PING", func(string, []byte) []byte { return ping })
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(stop)
	// 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
	}
	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()
}

// The laptop's symptom (2026-10-05), reproduced: its runtime's one large answer reached the console
// after the 750 ms window, and the console said its modules ran nowhere — "nothing in the mesh is
// called slack" — or ran only on another machine. Now a runtime that said it is there is waited for,
// one that never says what it serves is named wherever an address it might answer is asked for, and
// mesh_runtimes shows who answered, how fast and how much.
func TestARuntimeThatAnswersLateIsFoundAndOneThatNeverDoesIsNamed(t *testing.T) {
	keptWas, patienceWas := IndexKept, announce.Patience
	// Kept for the test: one round of discovery answers every question but mesh_runtimes, which asks anew.
	IndexKept, announce.Patience = time.Minute, 2500*time.Millisecond
	t.Cleanup(func() { IndexKept, announce.Patience = keptWas, patienceWas })
	mesh := mt.New(t)
	// Who holds which seat, by placement (ADR 0571 clause 4), as the seat mesh-placement's holder composes it.
	placedKV := mesh.Bucket(t, delivery.SeatHoldersBucket)
	for key, sh := range delivery.SeatHoldersOf(map[string]delivery.Placement{"bench": {Node: "bench",
		Holds: []delivery.Holding{{Claim: "mesh-wants", Scope: "mesh", Module: "mesh-delivery-assigner"}}}}) {
		raw, _ := json.Marshal(sh)
		if _, err := placedKV.Put(key, raw); err != nil {
			t.Fatal(err)
		}
	}
	consoleMembership := mt.MembershipOf("node-tools", "laptop-console", false, nil)
	consoleMembership.State = []bus.StateIssued{{Name: delivery.SeatHoldersRead, Bucket: delivery.SeatHoldersBucket}}
	mesh.Issue(t, consoleMembership)

	laptop := connect(t, "node-tools", "laptop")
	stopTool, err := laptop.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)
	slowRuntime(t, laptop, "laptop", announce.Window+400*time.Millisecond, []announce.Endpoint{
		{Kind: announce.KindTool, Module: "slack", Tool: "slack_check", Node: "laptop", Description: "Is Slack started once?",
			Subject: "mesh.mod.slack.tool.slack_check.laptop"},
		{Kind: announce.KindTool, Module: "nextcloud-client", Tool: "nextcloud_check", Node: "laptop",
			Subject: "mesh.mod.nextcloud-client.tool.nextcloud_check.laptop"},
	})
	sleeper := connect(t, "node-tools", "sleeper")
	slowRuntime(t, sleeper, "sleeper", -1, nil)
	desktop := connect(t, "node-tools", "desktop")
	slowRuntime(t, desktop, "desktop", 0, []announce.Endpoint{
		{Kind: announce.KindTool, Module: "nextcloud-client", Tool: "nextcloud_check", Node: "desktop",
			Subject: "mesh.mod.nextcloud-client.tool.nextcloud_check.desktop"},
	})

	controller := connect(t, "mesh-controller", "bench")
	var toolsAsked atomic.Int32
	for verb, output := range map[string]string{
		"nodes": `[{"name":"laptop"},{"name":"sleeper"},{"name":"desktop"},{"name":"bench"}]`,
		"modules": `[{"module":"slack","on":["laptop"],"tools":true},{"module":"nap","on":["sleeper"],"tools":true},` +
			`{"module":"nextcloud-client","on":["laptop","desktop","sleeper"],"tools":true}]`,
		// No part keeps a central record of tools (ADR 0571 clause 5): a seat verb's `replaces` is read from its contract
		// in mesh-sdk, so the controller's `tools` is served here only to see that it is never asked.
		"tools": `{"seats":[]}`,
	} {
		output, verb := output, verb
		stop, err := controller.HandleSubject("mesh.seat.mesh-controller.tool."+verb, func(json.RawMessage) (any, error) {
			if verb == "tools" {
				toolsAsked.Add(1)
			}
			return map[string]any{"output": output, "ok": true}, nil
		})
		if err != nil {
			t.Fatal(err)
		}
		t.Cleanup(stop)
	}
	controller.Flush()

	console := connect(t, "node-tools", "laptop-console")
	console.Follow("node-tools")
	mt.Until(t, func() error {
		if console.Membership("node-tools") == nil {
			return fmt.Errorf("the console's membership is not issued yet")
		}
		return nil
	})
	up, err := Serve(NewSurface(console, "laptop.node-tools"), "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	defer up.Close()
	endpoint := "http://" + up.Address + "/mcp"

	// The late answer is waited for: the laptop's module is found and called.
	if got, isErr := call(t, endpoint, "mesh_call", map[string]any{"address": "laptop/slack.slack_check"}); isErr || !strings.Contains(got, `"ok": true`) {
		t.Errorf("a late runtime's module: %s", got)
	}
	if got, isErr := call(t, endpoint, "mesh_call", map[string]any{"address": "laptop/nextcloud-client.nextcloud_check"}); isErr && strings.Contains(got, "does not run on laptop") {
		t.Errorf("a late runtime's module was said to run elsewhere: %s", got)
	}
	machine, _ := call(t, endpoint, "mesh_machine", map[string]any{"node": "laptop"})
	if !strings.Contains(machine, "laptop/slack.slack_check") {
		t.Errorf("the late runtime's machine lists nothing: %s", machine)
	}

	// The seats' contract and holders, for a client on this machine with no seat definitions (the TypeScript `mesh`).
	seats := post(t, endpoint, `{"jsonrpc":"2.0","id":9,"method":"mesh/seats"}`)
	if raw, _ := json.Marshal(seats["result"]); !strings.Contains(string(raw), `"holders":[{"module":"mesh-delivery-assigner","node":"bench"}],"scope":"mesh","seat":"mesh-wants"`) ||
		!strings.Contains(string(raw), `"replaces":["journalctl"]`) {
		t.Errorf("mesh/seats answered %s", raw)
	}
	if n := toolsAsked.Load(); n != 0 {
		t.Errorf("the console asked the controller's tools %d times", n)
	}

	// The silent one is named, never called missing.
	for address, want := range map[string]string{
		// A seat whose holder announced nothing is named with its holder by placement, never called missing.
		"mesh-wants.read":                          "the seat mesh-wants is held by mesh-delivery-assigner on bench by placement",
		"sleeper/nap.doze":                         "nap is assigned to sleeper in the mesh's records, but nothing that answered discovery serves it: discovery did not hear in full from node-tools on sleeper (it answered PING",
		"sleeper/nextcloud-client.nextcloud_check": "nextcloud-client on sleeper did not answer discovery: discovery did not hear in full from node-tools on sleeper",
	} {
		got, isErr := call(t, endpoint, "mesh_call", map[string]any{"address": address})
		if !isErr || !strings.Contains(got, want) || strings.Contains(got, "nothing in the mesh is called") || strings.Contains(got, "does not run on") {
			t.Errorf("%s: %s\nwant %q", address, got, want)
		}
	}
	machine, _ = call(t, endpoint, "mesh_machine", map[string]any{"node": "sleeper"})
	if !strings.Contains(machine, `"incomplete"`) || !strings.Contains(machine, `"modules": []`) {
		t.Errorf("a machine whose runtime did not answer: %s", machine)
	}
	overview, _ := call(t, endpoint, "mesh_overview", nil)
	if !strings.Contains(overview, "the runtime node-tools on sleeper") {
		t.Errorf("overview does not name the silent runtime: %s", overview)
	}

	// mesh_runtimes: who answered, how fast, how much; who did not.
	got, isErr := call(t, endpoint, "mesh_runtimes", nil)
	if isErr {
		t.Fatal(got)
	}
	var report struct {
		Answered []map[string]any `json:"answered"`
		NotHeard []map[string]any `json:"not heard"`
	}
	if err := json.Unmarshal([]byte(got), &report); err != nil {
		t.Fatalf("%v: %s", err, got)
	}
	var lap map[string]any
	for _, r := range report.Answered {
		if r["machine"] == "laptop" {
			lap = r
		}
	}
	if lap == nil {
		t.Fatalf("the laptop's runtime is not listed as answering: %s", got)
	}
	took, _ := time.ParseDuration(lap["took"].(string))
	if took < announce.Window || lap["bytes"].(float64) <= 0 || lap["modules"].(float64) != 2 || lap["tools"].(float64) != 2 ||
		lap["shortened"] != false || lap["last heard"] == nil {
		t.Errorf("the laptop's runtime: %v", lap)
	}
	if !strings.Contains(got, `"machine": "sleeper"`) || !strings.Contains(got, "answered PING") {
		t.Errorf("the silent runtime is not listed: %s", got)
	}
	silent := false
	for _, r := range report.NotHeard {
		silent = silent || r["machine"] == "sleeper"
	}
	if !silent {
		t.Errorf("sleeper is not among the not heard: %s", got)
	}
}

// A runtime heard once and silent now is named in the answers about its machine.
func TestARuntimeHeardBeforeAndNotNowIsNamed(t *testing.T) {
	s := &Surface{}
	before := &index{Modules: map[string]*moduleInfo{}, Recorded: map[string][]string{}, Discovery: announce.Discovery{At: time.Now().UTC(),
		Heard: []announce.Heard{{Info: micro.Info{ServiceIdentity: micro.ServiceIdentity{Name: "node-tools", ID: "laptop",
			Metadata: map[string]string{"node": "laptop"}}}, Took: 300 * time.Millisecond, Bytes: 1000}}}}
	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(), 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)
	}
	_, err := resolve(now, "laptop/nextcloud-client.nextcloud_check")
	if err == nil || !strings.Contains(err.Error(), "nextcloud-client on laptop did not answer discovery") ||
		!strings.Contains(err.Error(), "answered discovery") {
		t.Errorf("%v", err)
	}
	_, err = resolve(now, "laptop/slack.slack_check")
	if err == nil || strings.Contains(err.Error(), "nothing in the mesh is called") || !strings.Contains(err.Error(), "slack may be theirs") {
		t.Errorf("%v", err)
	}
	// Nobody unheard: the plain answers stand.
	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)
	}
	if _, err := resolve(quiet, "laptop/slack.slack_check"); err == nil || !strings.Contains(err.Error(), "nothing in the mesh is called slack") {
		t.Errorf("%v", err)
	}
}

// A runtime restarted answers under a new instance id; its old instance is gone, not unheard.
func TestARestartedRuntimeIsNotNamedUnheard(t *testing.T) {
	heard := func(id string) *index {
		return &index{Modules: map[string]*moduleInfo{}, Recorded: map[string][]string{}, Discovery: announce.Discovery{At: time.Now().UTC(),
			Heard: []announce.Heard{{Info: micro.Info{ServiceIdentity: micro.ServiceIdentity{Name: "node-tools", ID: id,
				Metadata: map[string]string{"node": "laptop"}}}}}}}
	}
	s := &Surface{}
	s.remember(heard("before-restart"))
	after := heard("after-restart")
	s.remember(after)
	if len(after.Unheard) != 0 {
		t.Errorf("a restarted runtime was named unheard: %+v", after.Unheard)
	}
	if len(s.runtimes) != 1 {
		t.Errorf("the old instance is still kept: %d runtimes", len(s.runtimes))
	}
}