jschoubben
494d0787e2
Launched.Stop now returns once the bundle's exit is handled, bounded by StopWait, and the launcher reads its restart pauses once at Start, so the exit goroutine no longer races a test that changes them. The caller tests drop their half-second pause (novox/hq issue 476).
652 lines
20 KiB
Go
652 lines
20 KiB
Go
// Package launch starts a bundle the runtime serves and speaks MCP over stdio to it (novox/hq ADR
|
|
// 0188, ADR 0193): `initialize`, `tools/list` once, `tools/call` per call. A Go binary, a Python
|
|
// script and a Node launcher are the same thing here: an executable that answers those. The runtime
|
|
// knows no language; it starts the path it is given.
|
|
package launch
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"os/exec"
|
|
"regexp"
|
|
"strings"
|
|
"sync"
|
|
"syscall"
|
|
"time"
|
|
|
|
"golang.org/x/sys/unix"
|
|
|
|
"github.com/novox/mesh-tools/node-tools/internal/wire"
|
|
)
|
|
|
|
// Protocol is the MCP version spoken to a bundle.
|
|
const Protocol = "2025-03-26"
|
|
|
|
// How long a child has for its handshake, and a call before the caller is told it is slow.
|
|
var (
|
|
HandshakeTimeout = 10 * time.Second
|
|
CallTimeout = 30 * time.Second
|
|
)
|
|
|
|
// Tool is one tool a launched bundle listed, and how to call it.
|
|
type Tool struct {
|
|
Name string
|
|
Description string
|
|
Input json.RawMessage
|
|
// Run calls the tool for caller: the bus user the call's subject named, or "" when it named none. A
|
|
// caller is handed to the bundle in the call's `_meta` (MetaCaller), never among its arguments.
|
|
Run func(caller string, args json.RawMessage) (json.RawMessage, error)
|
|
}
|
|
|
|
// MetaCaller is where in a `tools/call`'s `_meta` the runtime names the caller it took from the subject the
|
|
// call arrived on (novox/hq issue 365): mesh-sdk's stdio.MetaCaller, which hands it to the tool as
|
|
// stdio.CallerOf, and says Unproven when it is absent.
|
|
const MetaCaller = "mesh/caller"
|
|
|
|
// callParams is a `tools/call`'s params: the tool, its arguments as the caller sent them, and — only when the
|
|
// subject named one — the caller, in `_meta`, which the runtime writes and no caller can.
|
|
func callParams(name, caller string, args json.RawMessage) map[string]any {
|
|
params := map[string]any{"name": name, "arguments": args}
|
|
if caller != "" {
|
|
params["_meta"] = map[string]any{MetaCaller: caller}
|
|
}
|
|
return params
|
|
}
|
|
|
|
// Registration is the tools a bundle listed under one name: its module's, or a seat's.
|
|
type Registration struct {
|
|
Module string
|
|
Tools []Tool
|
|
}
|
|
|
|
// Bus is what a launched bundle reaches the mesh through (novox/hq ADR 0193, ADR 0198): the runtime
|
|
// publishes, asks and subscribes on the module's behalf. Subscribe is asked each time the module's
|
|
// code subscribes; the runtime binds the module's consumer once and calls deliver for every event,
|
|
// acknowledging it on the bus when deliver returns nil.
|
|
//
|
|
// State and Watch reach the module's state (novox/hq ADR 0201): State answers `get`, `put`, `delete`
|
|
// and `keys`; Watch hands deliver the current values and then every change, and returns once the
|
|
// current values are delivered — the watch is the child's, and stops when the child does.
|
|
type Bus interface {
|
|
Publish(params json.RawMessage) error
|
|
Ask(params json.RawMessage) (json.RawMessage, error)
|
|
Subscribe(deliver func(envelope json.RawMessage) error) error
|
|
State(verb string, params json.RawMessage) (json.RawMessage, error)
|
|
Watch(params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error)
|
|
// Log answers a bundle's `append`, `read`, `last` and `info` on its module's log (novox/hq ADR 0297).
|
|
Log(verb string, params json.RawMessage) (json.RawMessage, error)
|
|
// Seat answers a bundle's seat traffic (novox/hq ADR 0259 §3): `publish`, `prove` and `record` at once;
|
|
// `take` and `answer` bind once and hand each message or proof on through handOn, to whichever child
|
|
// of the module asked last — a restarted child asks again as its code runs again.
|
|
Seat(verb string, params json.RawMessage, handOn func(method string, params any) (json.RawMessage, error)) (json.RawMessage, error)
|
|
}
|
|
|
|
// EventTimeout bounds how long a bundle has to handle one event before it is offered again.
|
|
var EventTimeout = 2 * time.Minute
|
|
|
|
// Refused is a bundle's error answer to what the runtime asked it: the bundle's own words, as opposed
|
|
// to the runtime not reaching it (not running, exited, no answer in time). For `mesh/event` it is the
|
|
// one way a bundle says it did not take an event: any answer that is not an error — `{}`, `null`, no
|
|
// result at all — takes it (mesh-sdk, "Taking an event"; novox/hq issue 276).
|
|
type Refused struct {
|
|
Method string
|
|
Message string
|
|
}
|
|
|
|
func (r *Refused) Error() string { return r.Message }
|
|
|
|
// Restart backoff: a bundle that exits is started again at once, then after growing pauses while it
|
|
// keeps exiting, back to at once once it has run a while.
|
|
var (
|
|
RestartFirst = 500 * time.Millisecond
|
|
RestartMost = 30 * time.Second
|
|
RestartSettle = time.Minute
|
|
)
|
|
|
|
// StopWait bounds how long Launched.Stop waits for its bundle's exit to be handled; a bundle that
|
|
// ignores SIGTERM longer than this is left to exit on its own (novox/hq issue 476).
|
|
var StopWait = 5 * time.Second
|
|
|
|
// Launched is a running bundle: what it registered, and how to stop it.
|
|
type Launched struct {
|
|
Registrations []Registration
|
|
Stop func()
|
|
}
|
|
|
|
// Executable says whether an entrypoint can be started: a bundle the runtime serves is executable,
|
|
// and one that is not was not built to be served (ADR 0193).
|
|
func Executable(path string) bool {
|
|
info, err := os.Stat(path)
|
|
if err != nil || info.IsDir() {
|
|
return false
|
|
}
|
|
return info.Mode().Perm()&0o111 != 0
|
|
}
|
|
|
|
type message struct {
|
|
JSONRPC string `json:"jsonrpc,omitempty"`
|
|
ID json.RawMessage `json:"id,omitempty"`
|
|
Method string `json:"method,omitempty"`
|
|
Params json.RawMessage `json:"params,omitempty"`
|
|
Result json.RawMessage `json:"result,omitempty"`
|
|
Error *struct {
|
|
Code int `json:"code"`
|
|
Message string `json:"message"`
|
|
} `json:"error,omitempty"`
|
|
}
|
|
|
|
type child struct {
|
|
cmd *exec.Cmd
|
|
stdin io.WriteCloser
|
|
writeMu sync.Mutex
|
|
mu sync.Mutex
|
|
pending map[int64]chan message
|
|
next int64
|
|
dead chan struct{}
|
|
// handled is closed once the child's exit has been handled whole — its state cleared and its
|
|
// restart decided — which Stop waits for (novox/hq issue 476).
|
|
handled chan struct{}
|
|
why error
|
|
// watches are the state watches this child asked for, stopped when it exits (ADR 0201): a child
|
|
// that starts again watches again, as its code runs again.
|
|
watches []func()
|
|
}
|
|
|
|
func (c *child) write(m any) error {
|
|
b, err := wire.Marshal(m)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
c.writeMu.Lock()
|
|
defer c.writeMu.Unlock()
|
|
_, err = c.stdin.Write(append(b, '\n'))
|
|
return err
|
|
}
|
|
|
|
var stackLine = regexp.MustCompile(`^\s+at\s`)
|
|
|
|
// Start launches one bundle and learns its tools. It fails when the child cannot be started or does
|
|
// not complete the handshake. A child that exits later is started again on its next call.
|
|
func Start(module, entry string, env []string, mesh Bus, logf func(string, ...any)) (*Launched, error) {
|
|
var mu sync.Mutex
|
|
var current *child
|
|
// The child that last subscribed is the one events are handed to: a restarted child subscribes
|
|
// again as its code is imported, and from then on the module's events are its.
|
|
var subscriber *child
|
|
// The child that last asked to take a seat's work or answer its proofs is the one they are handed to.
|
|
var seatTaker *child
|
|
handOn := func(method string, params any) (json.RawMessage, error) {
|
|
mu.Lock()
|
|
c := seatTaker
|
|
mu.Unlock()
|
|
if c == nil {
|
|
return nil, errors.New(module + "'s bundle is not running to take it")
|
|
}
|
|
return c.ask(module, method, params, EventTimeout)
|
|
}
|
|
stopped := false
|
|
deliver := func(envelope json.RawMessage) error {
|
|
mu.Lock()
|
|
c := subscriber
|
|
mu.Unlock()
|
|
if c == nil {
|
|
return errors.New(module + "'s bundle is not running to take its events")
|
|
}
|
|
// Taken unless the answer is an error; what a result says is not read (see Refused).
|
|
_, err := c.ask(module, "mesh/event", map[string]any{"envelope": envelope}, EventTimeout)
|
|
return err
|
|
}
|
|
var restart func(after time.Duration)
|
|
var startedAt time.Time
|
|
// The restart pauses are read once, here: the exit handling runs on its own goroutine and must not
|
|
// read the package's settings while a caller (a test) changes them (novox/hq issue 476).
|
|
pauseFirst, pauseMost, settle := RestartFirst, RestartMost, RestartSettle
|
|
pause := pauseFirst
|
|
|
|
start := func() (*child, error) {
|
|
cmd := exec.Command(entry)
|
|
cmd.Env = env
|
|
stdin, err := cmd.StdinPipe()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
stdout, err := cmd.StdoutPipe()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
stderr, err := cmd.StderrPipe()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := cmd.Start(); err != nil {
|
|
return nil, err
|
|
}
|
|
c := &child{cmd: cmd, stdin: stdin, pending: map[int64]chan message{}, next: 1, dead: make(chan struct{}),
|
|
handled: make(chan struct{})}
|
|
var lastSaid string
|
|
var saidMu sync.Mutex
|
|
stderrDone := make(chan struct{})
|
|
go func() {
|
|
defer close(stderrDone)
|
|
scan := bufio.NewScanner(stderr)
|
|
scan.Buffer(make([]byte, 64*1024), 1<<20)
|
|
for scan.Scan() {
|
|
line := scan.Text()
|
|
if strings.TrimSpace(line) == "" {
|
|
continue
|
|
}
|
|
logf("[%s] %s", module, line)
|
|
if !stackLine.MatchString(line) && !strings.HasPrefix(line, "Node.js v") {
|
|
saidMu.Lock()
|
|
lastSaid = strings.TrimSpace(line)
|
|
saidMu.Unlock()
|
|
}
|
|
}
|
|
}()
|
|
go func() {
|
|
defer close(c.handled)
|
|
scan := bufio.NewScanner(stdout)
|
|
scan.Buffer(make([]byte, 64*1024), 16<<20)
|
|
for scan.Scan() {
|
|
line := strings.TrimSpace(scan.Text())
|
|
if line == "" {
|
|
continue
|
|
}
|
|
var m message
|
|
if err := json.Unmarshal([]byte(line), &m); err != nil {
|
|
if len(line) > 120 {
|
|
line = line[:120]
|
|
}
|
|
logf("[mesh-tools] %s's bundle said something that is not a reply: %s", module, line)
|
|
continue
|
|
}
|
|
// The bundle asks the runtime (ADR 0193, ADR 0198): to emit, to call a tool, to hand it the
|
|
// module's events. Each on the module's behalf, answered when done; nothing else.
|
|
if m.Method != "" {
|
|
go func(m message) {
|
|
var result any = map[string]any{}
|
|
var err error
|
|
switch m.Method {
|
|
case "mesh/publish":
|
|
err = mesh.Publish(m.Params)
|
|
case "mesh/ask":
|
|
var asked json.RawMessage
|
|
if asked, err = mesh.Ask(m.Params); err == nil {
|
|
result = asked
|
|
}
|
|
case "mesh/subscribe":
|
|
mu.Lock()
|
|
subscriber = c
|
|
mu.Unlock()
|
|
err = mesh.Subscribe(deliver)
|
|
case "mesh/seat.publish", "mesh/seat.prove", "mesh/seat.record", "mesh/seat.kinds":
|
|
var answered json.RawMessage
|
|
if answered, err = mesh.Seat(strings.TrimPrefix(m.Method, "mesh/seat."), m.Params, nil); err == nil {
|
|
result = answered
|
|
}
|
|
case "mesh/seat.take", "mesh/seat.answer":
|
|
mu.Lock()
|
|
seatTaker = c
|
|
mu.Unlock()
|
|
var answered json.RawMessage
|
|
if answered, err = mesh.Seat(strings.TrimPrefix(m.Method, "mesh/seat."), m.Params, handOn); err == nil {
|
|
result = answered
|
|
}
|
|
case "mesh/state.get", "mesh/state.put", "mesh/state.delete", "mesh/state.keys",
|
|
"mesh/state.create", "mesh/state.update", "mesh/state.history":
|
|
var answered json.RawMessage
|
|
if answered, err = mesh.State(strings.TrimPrefix(m.Method, "mesh/state."), m.Params); err == nil {
|
|
result = answered
|
|
}
|
|
case "mesh/log.append", "mesh/log.read", "mesh/log.last", "mesh/log.info":
|
|
var answered json.RawMessage
|
|
if answered, err = mesh.Log(strings.TrimPrefix(m.Method, "mesh/log."), m.Params); err == nil {
|
|
result = answered
|
|
}
|
|
case "mesh/state.watch":
|
|
// Each change is asked of this child, in order, naming the watch it is for by
|
|
// the id the child asked it with — two watches of one state are two handlers;
|
|
// the watch is answered once the current values have been handed over, so a
|
|
// bundle that awaits it has the whole of the state before it goes on (ADR 0201).
|
|
var stop func()
|
|
watch := m.ID
|
|
stop, err = mesh.Watch(m.Params, func(change json.RawMessage) error {
|
|
_, err := c.ask(module, "mesh/state", map[string]any{"watch": watch, "change": change}, EventTimeout)
|
|
return err
|
|
})
|
|
if err == nil {
|
|
c.mu.Lock()
|
|
if c.why != nil {
|
|
c.mu.Unlock()
|
|
stop()
|
|
} else {
|
|
c.watches = append(c.watches, stop)
|
|
c.mu.Unlock()
|
|
}
|
|
}
|
|
default:
|
|
if len(m.ID) > 0 {
|
|
_ = c.write(map[string]any{"jsonrpc": "2.0", "id": m.ID,
|
|
"error": map[string]any{"code": -32601, "message": "the runtime answers no " + m.Method + " from a bundle"}})
|
|
}
|
|
return
|
|
}
|
|
if len(m.ID) == 0 {
|
|
return
|
|
}
|
|
if err != nil {
|
|
_ = c.write(map[string]any{"jsonrpc": "2.0", "id": m.ID,
|
|
"error": map[string]any{"code": errorCode(err), "message": err.Error()}})
|
|
return
|
|
}
|
|
_ = c.write(map[string]any{"jsonrpc": "2.0", "id": m.ID, "result": result})
|
|
}(m)
|
|
continue
|
|
}
|
|
var id int64
|
|
if json.Unmarshal(m.ID, &id) != nil {
|
|
continue
|
|
}
|
|
c.mu.Lock()
|
|
ch := c.pending[id]
|
|
delete(c.pending, id)
|
|
c.mu.Unlock()
|
|
if ch != nil {
|
|
ch <- m
|
|
}
|
|
}
|
|
<-stderrDone
|
|
err := cmd.Wait()
|
|
code := "0"
|
|
if exit := (*exec.ExitError)(nil); errors.As(err, &exit) {
|
|
if status, ok := exit.Sys().(syscall.WaitStatus); ok && status.Signaled() {
|
|
code = unix.SignalName(status.Signal())
|
|
} else {
|
|
code = fmt.Sprint(exit.ExitCode())
|
|
}
|
|
}
|
|
saidMu.Lock()
|
|
why := fmt.Sprintf("%s's bundle exited (%s)", module, code)
|
|
if lastSaid != "" {
|
|
why += ": " + lastSaid
|
|
}
|
|
saidMu.Unlock()
|
|
c.mu.Lock()
|
|
c.why = errors.New(why)
|
|
watches := c.watches
|
|
c.watches = nil
|
|
c.mu.Unlock()
|
|
for _, stop := range watches {
|
|
stop()
|
|
}
|
|
close(c.dead)
|
|
mu.Lock()
|
|
// Only a child that was serving is brought back; one that died in its own handshake was
|
|
// never accepted, and whoever started it was told why.
|
|
wasServing := current == c
|
|
if current == c {
|
|
current = nil
|
|
}
|
|
if subscriber == c {
|
|
subscriber = nil
|
|
}
|
|
if seatTaker == c {
|
|
seatTaker = nil
|
|
}
|
|
wasStopped := stopped
|
|
// Started again at once if it ran a while; after a growing pause while it keeps exiting.
|
|
wait := pauseFirst
|
|
if time.Since(startedAt) < settle {
|
|
wait = pause
|
|
if pause *= 2; pause > pauseMost {
|
|
pause = pauseMost
|
|
}
|
|
} else {
|
|
pause = pauseFirst
|
|
}
|
|
mu.Unlock()
|
|
if !wasStopped && wasServing {
|
|
// Every bundle stays up (ADR 0198): code that runs long — a handler, a provisioner —
|
|
// is not waiting for a call to bring it back, and a tool bundle back early costs nothing.
|
|
logf("[mesh-tools] %s; started again in %s", why, wait.Round(100*time.Millisecond))
|
|
restart(wait)
|
|
}
|
|
}()
|
|
if _, err := c.ask(module, "initialize", map[string]any{"protocolVersion": Protocol, "capabilities": map[string]any{},
|
|
"clientInfo": map[string]any{"name": "node-tools", "version": "1"}}, HandshakeTimeout); err != nil {
|
|
_ = cmd.Process.Kill()
|
|
return nil, err
|
|
}
|
|
_ = c.write(map[string]any{"jsonrpc": "2.0", "method": "notifications/initialized"})
|
|
return c, nil
|
|
}
|
|
|
|
asking := func(method string, params any, timeout time.Duration) (json.RawMessage, error) {
|
|
mu.Lock()
|
|
c := current
|
|
mu.Unlock()
|
|
if c == nil {
|
|
fresh, err := start()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
mu.Lock()
|
|
if current == nil {
|
|
current = fresh
|
|
startedAt = time.Now()
|
|
} else {
|
|
_ = fresh.cmd.Process.Signal(syscall.SIGTERM)
|
|
fresh = current
|
|
}
|
|
c = fresh
|
|
mu.Unlock()
|
|
}
|
|
return c.ask(module, method, params, timeout)
|
|
}
|
|
|
|
restart = func(after time.Duration) {
|
|
go func() {
|
|
time.Sleep(after)
|
|
mu.Lock()
|
|
if stopped || current != nil {
|
|
mu.Unlock()
|
|
return
|
|
}
|
|
mu.Unlock()
|
|
fresh, err := start()
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
if err != nil {
|
|
if !stopped {
|
|
wait := pause
|
|
if pause *= 2; pause > pauseMost {
|
|
pause = pauseMost
|
|
}
|
|
logf("[mesh-tools] %s's bundle did not start again: %v; trying in %s", module, err, wait)
|
|
go restart(wait)
|
|
}
|
|
return
|
|
}
|
|
if stopped {
|
|
_ = fresh.cmd.Process.Signal(syscall.SIGTERM)
|
|
return
|
|
}
|
|
if current == nil {
|
|
current = fresh
|
|
startedAt = time.Now()
|
|
} else {
|
|
_ = fresh.cmd.Process.Signal(syscall.SIGTERM)
|
|
}
|
|
}()
|
|
}
|
|
|
|
first, err := start()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
mu.Lock()
|
|
current = first
|
|
startedAt = time.Now()
|
|
mu.Unlock()
|
|
|
|
raw, err := asking("tools/list", map[string]any{}, HandshakeTimeout)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var listed struct {
|
|
Tools []struct {
|
|
Name string `json:"name"`
|
|
Description string `json:"description"`
|
|
InputSchema json.RawMessage `json:"inputSchema"`
|
|
} `json:"tools"`
|
|
}
|
|
if err := json.Unmarshal(raw, &listed); err != nil {
|
|
return nil, fmt.Errorf("%s's bundle listed its tools in a shape that is not MCP's: %w", module, err)
|
|
}
|
|
order := []string{}
|
|
groups := map[string][]Tool{}
|
|
for _, t := range listed.Tools {
|
|
under, name := module, t.Name
|
|
if dot := strings.Index(t.Name, "."); dot >= 0 {
|
|
under, name = t.Name[:dot], t.Name[dot+1:]
|
|
}
|
|
full := t.Name
|
|
input := t.InputSchema
|
|
if len(input) == 0 || string(input) == "null" {
|
|
input = json.RawMessage("{}")
|
|
}
|
|
tool := Tool{Name: name, Description: t.Description, Input: input,
|
|
Run: func(caller string, args json.RawMessage) (json.RawMessage, error) {
|
|
if len(args) == 0 || string(args) == "null" {
|
|
args = json.RawMessage("{}")
|
|
}
|
|
res, err := asking("tools/call", callParams(full, caller, args), CallTimeout)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var called struct {
|
|
Content []struct {
|
|
Type string `json:"type"`
|
|
Text string `json:"text"`
|
|
} `json:"content"`
|
|
IsError bool `json:"isError"`
|
|
}
|
|
_ = json.Unmarshal(res, &called)
|
|
text := ""
|
|
for _, c := range called.Content {
|
|
if c.Type == "text" {
|
|
text = c.Text
|
|
break
|
|
}
|
|
}
|
|
if called.IsError {
|
|
if text == "" {
|
|
text = module + "." + full + " failed"
|
|
}
|
|
return nil, errors.New(text)
|
|
}
|
|
// The bundle's answer is JSON as text; handed back as the value it encodes.
|
|
if json.Valid([]byte(text)) && text != "" {
|
|
return json.RawMessage(text), nil
|
|
}
|
|
b, _ := json.Marshal(text)
|
|
return b, nil
|
|
}}
|
|
if _, seen := groups[under]; !seen {
|
|
order = append(order, under)
|
|
}
|
|
groups[under] = append(groups[under], tool)
|
|
}
|
|
out := &Launched{Stop: func() {
|
|
mu.Lock()
|
|
stopped = true
|
|
c := current
|
|
current = nil
|
|
mu.Unlock()
|
|
if c != nil && c.cmd.Process != nil {
|
|
_ = c.cmd.Process.Signal(syscall.SIGTERM)
|
|
// Returns once the bundle's exit is handled, so nothing of it runs on after Stop
|
|
// (novox/hq issue 476); bounded, for a bundle that does not exit on SIGTERM.
|
|
select {
|
|
case <-c.handled:
|
|
case <-time.After(StopWait):
|
|
}
|
|
}
|
|
}}
|
|
for _, under := range order {
|
|
out.Registrations = append(out.Registrations, Registration{Module: under, Tools: groups[under]})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func (c *child) ask(module, method string, params any, timeout time.Duration) (json.RawMessage, error) {
|
|
c.mu.Lock()
|
|
if c.why != nil {
|
|
c.mu.Unlock()
|
|
return nil, c.why
|
|
}
|
|
id := c.next
|
|
c.next++
|
|
ch := make(chan message, 1)
|
|
c.pending[id] = ch
|
|
c.mu.Unlock()
|
|
if err := c.write(map[string]any{"jsonrpc": "2.0", "id": id, "method": method, "params": params}); err != nil {
|
|
c.mu.Lock()
|
|
delete(c.pending, id)
|
|
c.mu.Unlock()
|
|
select {
|
|
case <-c.dead:
|
|
return nil, c.deathReason()
|
|
case <-time.After(100 * time.Millisecond):
|
|
return nil, err
|
|
}
|
|
}
|
|
timer := time.NewTimer(timeout)
|
|
defer timer.Stop()
|
|
select {
|
|
case m := <-ch:
|
|
if m.Error != nil {
|
|
msg := m.Error.Message
|
|
if msg == "" {
|
|
msg = "the bundle refused the request"
|
|
}
|
|
return nil, &Refused{Method: method, Message: msg}
|
|
}
|
|
return m.Result, nil
|
|
case <-c.dead:
|
|
return nil, c.deathReason()
|
|
case <-timer.C:
|
|
c.mu.Lock()
|
|
delete(c.pending, id)
|
|
c.mu.Unlock()
|
|
return nil, fmt.Errorf("%s's bundle did not answer %s in %ds", module, method, int(timeout/time.Second))
|
|
}
|
|
}
|
|
|
|
func (c *child) deathReason() error {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
if c.why != nil {
|
|
return c.why
|
|
}
|
|
return errors.New("the bundle exited")
|
|
}
|
|
|
|
// CodeChanged is the error code a bundle is answered a lost compare-and-set with (mesh-sdk's
|
|
// stdio.CodeChanged): told by its code, never by its words (novox/hq ADR 0259).
|
|
const CodeChanged = -32010
|
|
|
|
// errorCode is the code an error is answered to a bundle with: a lost compare-and-set's own, else the
|
|
// general one.
|
|
func errorCode(err error) int {
|
|
var coded interface{ ErrorCode() int }
|
|
if errors.As(err, &coded) {
|
|
return coded.ErrorCode()
|
|
}
|
|
return -32000
|
|
}
|