From 809ab4cf29de1d3a214bd3f78baa1edc4b74ce97 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 8 Oct 2026 16:44:01 +0200 Subject: [PATCH] Ask a declared tool check on the NATS link the node-engine holds The queue asks a tool through a ToolBus assertion that only OverNATS satisfied; the link the engine actually holds (natsLink) had no Ask, so every tool check failed as "no link to the bus is open" while the link was up (hq issue 331). Give natsLink Ask, assert at build time every optional interface the queue expects of it, and tell a link that cannot ask apart from no link. --- internal/link/asktool_nats_test.go | 40 ++++++++++++++++++++++++++++++ internal/link/health_test.go | 31 +++++++++++++++++++++++ internal/link/hearing_nats.go | 27 ++++++++++++++++++++ internal/link/queue.go | 7 ++++-- 4 files changed, 103 insertions(+), 2 deletions(-) create mode 100644 internal/link/asktool_nats_test.go diff --git a/internal/link/asktool_nats_test.go b/internal/link/asktool_nats_test.go new file mode 100644 index 0000000..e9caa58 --- /dev/null +++ b/internal/link/asktool_nats_test.go @@ -0,0 +1,40 @@ +package link + +import ( + "context" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" +) + +// A declared tool check asked over a real bus, through the queue, on the link the node-engine holds: +// the tool's answer is read, and a tool nothing serves is said as such (novox/hq ADR 0240, to-be 48 §3). +func TestNatsAToolCheckIsAskedOnTheLink(t *testing.T) { + conn, js := aBus(t) + const node = "toolcheck" + served, err := conn.Subscribe(ToolSubject("sensors", "sensors_health", node), func(m *nats.Msg) { + _ = m.Respond([]byte(`{"result":{"healthy":true}}`)) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = served.Unsubscribe() }) + if err := conn.Flush(); err != nil { + t.Fatal(err) + } + + q := &Queue{Membership: Membership{Node: node}} + q.attach(t.Context(), &natsLink{conn: conn, js: js, node: node}) + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + healthy, why, err := q.AskTool(ctx, "sensors", "sensors_health") + if err != nil || !healthy { + t.Fatalf("a served healthy tool read as %v %q (%v)", healthy, why, err) + } + _, _, err = q.AskTool(ctx, "sensors", "nothing_serves_this") + if err == nil || !strings.Contains(err.Error(), "nothing on this machine answers") { + t.Fatalf("a tool nothing serves read as %v", err) + } +} diff --git a/internal/link/health_test.go b/internal/link/health_test.go index 3021e09..dd701ae 100644 --- a/internal/link/health_test.go +++ b/internal/link/health_test.go @@ -3,6 +3,7 @@ package link import ( "context" "encoding/json" + "strings" "testing" "time" ) @@ -70,3 +71,33 @@ func TestAToolsAnswerIsHealthyOnlyWhenItSaysSo(t *testing.T) { t.Errorf("asked on %s", ToolSubject("keycloak", "keycloak_admin_health", "anchor")) } } + +// A declared tool check is asked on the link the node-engine actually holds — the NATS link — and not +// refused as if none were open (2026-10-08: every tool check read "no link to the bus is open" while +// the link was up, because the NATS link could not Ask). With no link open, and with a link that cannot +// ask, the check fails and says which. +func TestAToolCheckIsAskedOnTheLinkTheEngineHolds(t *testing.T) { + q := &Queue{Membership: Membership{Node: "anchor"}} + if _, _, err := q.AskTool(t.Context(), "sensors", "sensors_health"); err == nil || err.Error() != "no link to the bus is open" { + t.Fatalf("with no link open: %v", err) + } + + quiet := newQuietLink() + q.attach(t.Context(), quiet) + if _, _, err := q.AskTool(t.Context(), "sensors", "sensors_health"); err == nil || err.Error() != "the link to the bus open now cannot ask a tool" { + t.Fatalf("on a link that cannot ask: %v", err) + } + q.detach(quiet) + + // The link Open returns, with no connection under it: the question must reach its Ask (and fail + // there), never stop at the queue's assertion. + held := &natsLink{node: "anchor"} + q.attach(t.Context(), held) + _, _, err := q.AskTool(t.Context(), "sensors", "sensors_health") + if err == nil { + t.Fatal("asked on no connection and was answered") + } + if strings.Contains(err.Error(), "no link") || strings.Contains(err.Error(), "cannot ask a tool") { + t.Fatalf("the NATS link was not asked: %v", err) + } +} diff --git a/internal/link/hearing_nats.go b/internal/link/hearing_nats.go index 32e8623..b2c4885 100644 --- a/internal/link/hearing_nats.go +++ b/internal/link/hearing_nats.go @@ -220,6 +220,33 @@ func (l *natsLink) Health(ctx context.Context, node string, body []byte) error { return OverNATS{Conn: l.conn, JS: l.js}.Health(ctx, node, body) } +// Ask asks a module's tool for a declared tool check (novox/hq ADR 0240, to-be 48 §3). The link the +// node-engine holds is this one, not OverNATS: without it here the queue's ToolBus assertion failed on +// every live link and each tool check read "no link to the bus is open" while the link was up. +func (l *natsLink) Ask(ctx context.Context, subject string, body []byte) ([]byte, error) { + reply, err := OverNATS{Conn: l.conn, JS: l.js}.Ask(ctx, subject, body) + switch { + case errors.Is(err, nats.ErrNoResponders): + return nil, fmt.Errorf("nothing on this machine answers %s", subject) + case err != nil: + if refused := l.refusal(subject); refused != "" { + return nil, fmt.Errorf("asking %s: %v%s", subject, err, refused) + } + return nil, fmt.Errorf("asking %s: %w", subject, err) + } + return reply, nil +} + +// What the queue asks of a live link beside Bus, each by a type assertion that fails quietly: said here +// so a link that drops one does not build. +var ( + _ Bus = (*natsLink)(nil) + _ HealthBus = (*natsLink)(nil) + _ ToolBus = (*natsLink)(nil) + _ Asker = (*natsLink)(nil) + _ Asked = (*natsLink)(nil) +) + // natsDeclaration is one declaration off the NODES stream. type natsDeclaration struct{ msg *nats.Msg } diff --git a/internal/link/queue.go b/internal/link/queue.go index 6021f5f..1ae5f5c 100644 --- a/internal/link/queue.go +++ b/internal/link/queue.go @@ -530,10 +530,13 @@ func (q *Queue) AskTool(ctx context.Context, module, tool string) (bool, string, q.mu.Lock() bus := q.bus q.mu.Unlock() - asker, ok := bus.(ToolBus) - if bus == nil || !ok { + if bus == nil { return false, "", errors.New("no link to the bus is open") } + asker, ok := bus.(ToolBus) + if !ok { + return false, "", errors.New("the link to the bus open now cannot ask a tool") + } reply, err := asker.Ask(ctx, ToolSubject(module, tool, q.Membership.Node), []byte("{}")) if err != nil { return false, "", err