Files
mesh-controller/internal/link/ask.go
T
jschoubben 47205403d3
mesh/merge-gate pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without …
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery superseded: a newer head of the same pull request
Wait out a forge not served for a moment when registering a build (novox/hq issue 486)
A registration asked the forge while node-tools re-served after a tool-runner
rollout, heard "nothing serves gitea.gitea_api", and failed a good build's
plan. The bus's not-served answer is now link.ErrNothingServes (and a timeout
link.ErrNoAnswer), and the registration asks again for up to 90 s on those
alone, never on a refusal; past the bound it fails saying how long it waited.
2026-10-12 11:23:03 +02:00

72 lines
2.8 KiB
Go

package link
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/nats-io/nats.go"
)
// Answer is what a module's tool replies: one of the two, never both.
type Answer struct {
Result json.RawMessage `json:"result,omitempty"`
Error string `json:"error,omitempty"`
}
// ErrNoAnswer is a tool asked that did not answer within the asker's wait: nothing refused, and asking again
// may be answered. With ErrNothingServes, what an asker may wait out (novox/hq issue 486).
var ErrNoAnswer = errors.New("no answer within the wait")
// saidAs is an error said in its own words that is still the error it is a case of, for errors.Is.
type saidAs struct {
text string
is error
}
func (e saidAs) Error() string { return e.text }
func (e saidAs) Unwrap() error { return e.is }
// Ask calls one of a module's tools over the broker and waits for its answer.
//
// **The control plane is the way in** (novox/hq 04-ISSUES/049, ADR 0095). A module's broker
// account is scoped to what it emits, consumes and serves, and a tool call needs a reply queue the
// caller creates and a publish to the serving module's request key — which no module's scope
// grants, and should not. The control plane already holds a connection that may, so a person or
// an agent asks through it, and every question passes one process where an audit belongs.
//
// The answer comes back on the asker's own inbox, which only the asker may read; the serving
// module answers there and nowhere else. The bus refuses a request nothing serves at once, so a
// module that is down or a tool that does not exist is said now rather than after the whole wait.
func Ask(ctx context.Context, bus Bus, module, tool string, args json.RawMessage,
timeout time.Duration) (Answer, error) {
if len(args) == 0 {
args = json.RawMessage(`{}`)
}
key := module + "." + tool
reply, err := bus.AskTool(ctx, module, tool, args, timeout)
if err != nil {
if errors.Is(err, nats.ErrNoResponders) {
return Answer{}, saidAs{fmt.Sprintf(
"nothing serves %s: no runtime has bound %q on the bus. The module is not "+
"assigned, its runtime is not up, or it serves no such tool — `status` says "+
"whether the machine carrying it has applied", module, key), ErrNothingServes}
}
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) {
return Answer{}, saidAs{fmt.Sprintf(
"%s did not answer within %s. Its runtime serves %q when it is up and has bound "+
"the bus — `status` says whether the machine carrying it has applied",
module, timeout, key), ErrNoAnswer}
}
return Answer{}, fmt.Errorf("cannot ask %s: %w", key, err)
}
var answer Answer
if err := json.Unmarshal(reply, &answer); err != nil {
return Answer{}, fmt.Errorf("%s answered with something unreadable: %w", key, err)
}
return answer, nil
}