jschoubben
47205403d3
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.
72 lines
2.8 KiB
Go
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
|
|
}
|