From 1dc5d8cb3d652821244af54264c3bf4404f1efdc Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:15:46 +0100 Subject: [PATCH] Run a watch's handler beside the runtime, and give a refused watch up loudly (issue 774) The provisioner's watches of gitea.bound, mesh-vault.bound and postgres.bound failed every minute for an hour on the control-node ("the current values ... did not arrive in 10s") and 1257 changes were refused as "names no watch of this bundle". Both sides were at fault. The SDK answered a change only once its handler returned, so a slow handler (Rejudge waits on the lock its composition holds) counted against the runtime's 10 s bound on the current values; and when the runtime gave up, the SDK forgot the id at once. The runtime (mesh-tools node-tools internal/bus/state.go StateWatch) stops its watcher on that bound but still drains and delivers the values already buffered, to an id nobody holds. Quick fix: a longer runtime bound. Thorough fix here: each state and provision watch has a follower that takes a change in order and answers at once, its handler on its own goroutine; Watch still returns only once the current values were handled, and a handler's error is said on standard error. A watch the runtime refuses is said given up and kept by id, so a change still handed to it is refused saying so. KeptState.Follow asks again with backoff until taken, the loop four modules wrote for themselves. A log watch keeps its answer after the handler: that is its back-pressure. --- go/follower.go | 135 +++++++++++++++++++++++++++++++++++++++++++++++ go/provision.go | 7 ++- go/state.go | 12 ++++- go/state_test.go | 58 ++++++++++++++++++++ go/stdio.go | 49 ++++++++--------- 5 files changed, 230 insertions(+), 31 deletions(-) create mode 100644 go/follower.go diff --git a/go/follower.go b/go/follower.go new file mode 100644 index 0000000..17a9407 --- /dev/null +++ b/go/follower.go @@ -0,0 +1,135 @@ +package stdio + +import ( + "fmt" + "os" + "sync" + "time" +) + +// A watch's handler runs beside the runtime, never in its way (issue 774, the sweep of 2026-10-30). +// +// The runtime hands a watch's current values over one at a time and gives up on the watch when none was handed over +// within its bound (mesh-tools node-tools internal/bus/state.go StateWatch, 10 s). The bundle answered each change only +// once its handler returned, so a slow handler (mesh-provisioner's Rejudge waits on the lock its composition holds) +// counted against the bus's bound: the watch was given up, the SDK forgot its id, and the values the runtime still +// delivered to it were refused as "names no watch of this bundle", 1257 times in an hour on the control-node. +// +// So each watch has a follower: a change is taken into it in the order the runtime hands it over and answered at once, +// and the handler runs on the follower's own goroutine, in that order. Watch still returns only once every current +// value was handled. A handler's error is said on the bundle's standard error, naming the watch and the change, since +// the runtime no longer hears it. A log watch keeps its answer after the handler: its runtime sends the next entry only +// after the answer, which is the back-pressure a replay of a long log needs. + +// follower runs one watch's handler on its own goroutine, in order. +type follower[T any] struct { + what string + handle func(T) error + describe func(T) string + mu sync.Mutex + items []followed[T] + wake chan struct{} +} + +// followed is one change, or with no change a mark that closes done once everything before it was handled. +type followed[T any] struct { + change T + mark bool + done chan struct{} +} + +func newFollower[T any](what string, handle func(T) error, describe func(T) string) *follower[T] { + f := &follower[T]{what: what, handle: handle, describe: describe, wake: make(chan struct{}, 1)} + go f.run() + return f +} + +// take queues a change; it never waits on the handler. +func (f *follower[T]) take(change T) { + f.mu.Lock() + f.items = append(f.items, followed[T]{change: change}) + f.mu.Unlock() + f.poke() +} + +// handled is closed once every change taken before it was handled. +func (f *follower[T]) handled() <-chan struct{} { + done := make(chan struct{}) + f.mu.Lock() + f.items = append(f.items, followed[T]{mark: true, done: done}) + f.mu.Unlock() + f.poke() + return done +} + +func (f *follower[T]) poke() { + select { + case f.wake <- struct{}{}: + default: + } +} + +func (f *follower[T]) run() { + for range f.wake { + for { + f.mu.Lock() + if len(f.items) == 0 { + f.mu.Unlock() + break + } + next := f.items[0] + f.items = f.items[1:] + f.mu.Unlock() + if next.mark { + close(next.done) + continue + } + if err := f.handle(next.change); err != nil { + sayf("%s: the handler did not take %s: %v; the next change hands it again", f.what, f.describe(next.change), err) + } + } + } +} + +// sayf says something the runtime cannot say for the bundle, on its standard error, which its journal keeps. +func sayf(format string, args ...any) { + fmt.Fprintf(os.Stderr, "[mesh-sdk] "+format+"\n", args...) +} + +// gaveUpMost bounds what a bundle remembers of the watches it gave up. +const gaveUpMost = 256 + +// gaveUp keeps why a watch was given up, so a change the runtime still hands to it is refused saying so, never as a +// watch nobody asked for. Called with c.mu held. +func (c *channel) gaveUp(id, why string) { + if c.givenUp == nil || len(c.givenUp) >= gaveUpMost { + c.givenUp = map[string]string{} + } + c.givenUp[id] = why +} + +// noWatch is the refusal of a change for watch id, which this bundle never asked or gave up. +func (c *channel) noWatch(kind, id string) string { + c.mu.Lock() + why, given := c.givenUp[id] + c.mu.Unlock() + if given { + sayf("%s: the runtime still hands over changes for the watch %s, which this bundle gave up: %s", kind, id, why) + return fmt.Sprintf("%s: this bundle gave the watch %s up (%s); stop handing it over", kind, id, why) + } + return kind + " names no watch of this bundle" +} + +// Follow watches key of the state as Watch does and asks again until the watch is taken, saying each give-up and +// when it asks again (2 s, doubled to a minute): the loop every module wrote for itself, in one place. It returns once +// the watch is taken; call it beside the code that needs it. +func (s KeptState) Follow(key string, handle func(StateChange) error) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := s.Watch(key, handle) + if err == nil { + return + } + sayf("the state %s is not watched yet (%v); asking again in %s", s.name, err, wait) + time.Sleep(wait) + } +} diff --git a/go/provision.go b/go/provision.go index e4d65c7..be289e2 100644 --- a/go/provision.go +++ b/go/provision.go @@ -3,6 +3,7 @@ package stdio import ( "encoding/json" "errors" + "fmt" "time" ) @@ -42,15 +43,19 @@ func provisionWatch(kind, key string, handle func(ProvisionChange) error) error return err } id := c.newID() + f := newFollower("the watch "+id+" of the "+kind+" "+key, handle, func(ch ProvisionChange) string { return ch.Kind + " " + ch.Key }) c.mu.Lock() - c.provisioned[id] = handle + c.provisioned[id] = f c.mu.Unlock() if _, err := c.askAs(id, "mesh/provision.watch", map[string]any{"kind": kind, "key": key}, provisionTimeout); err != nil { c.mu.Lock() delete(c.provisioned, id) + c.gaveUp(id, fmt.Sprintf("the %s %s at %s: %v", kind, key, time.Now().Format(time.RFC3339), err)) c.mu.Unlock() + sayf("the watch %s of the %s %s is given up: %v", id, kind, key, err) return err } + <-f.handled() return nil } diff --git a/go/state.go b/go/state.go index ad49de2..a085570 100644 --- a/go/state.go +++ b/go/state.go @@ -213,21 +213,29 @@ func (s KeptState) Keys() ([]string, error) { // Watch hands handle every current value — none that is deleted — and then every change, in order, // and returns once the current values have all been handled. key narrows it: a key, with `*` for one -// name and `**` for the rest; empty is every key. The watch lasts as long as the process. +// name and `**` for the rest; empty is every key. The watch lasts as long as the process. handle runs beside the +// runtime (follower.go): its time never counts against the runtime's bound on the current values, and an error it +// returns is said on standard error. A watch the runtime refuses is given up, said, and a change the runtime still +// hands to it is refused saying so; Follow asks again until it is taken. func (s KeptState) Watch(key string, handle func(StateChange) error) error { c, err := live() if err != nil { return err } id := c.newID() + f := newFollower("the watch "+id+" of the state "+s.name, handle, func(ch StateChange) string { return ch.Op + " " + ch.Key }) c.mu.Lock() - c.watches[id] = handle + c.watches[id] = f c.mu.Unlock() if _, err := c.askAs(id, "mesh/state.watch", map[string]any{"state": s.name, "key": key}, stateTimeout); err != nil { c.mu.Lock() delete(c.watches, id) + c.gaveUp(id, fmt.Sprintf("the state %s at %s: %v", s.name, time.Now().Format(time.RFC3339), err)) c.mu.Unlock() + sayf("the watch %s of the state %s is given up: %v", id, s.name, err) return err } + // The runtime hands every current value over before it answers the watch: each is in the follower by now. + <-f.handled() return nil } diff --git a/go/state_test.go b/go/state_test.go index dc80765..cac843b 100644 --- a/go/state_test.go +++ b/go/state_test.go @@ -20,6 +20,7 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { outR, outW := io.Pipe() seen := make(chan string, 16) watched := make(chan error, 1) + slowWatched, release := make(chan error, 1), make(chan struct{}) go func() { _ = ServeOn("", []Tool{ {Name: "start", Run: func(map[string]any) (any, error) { @@ -31,6 +32,19 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { }() return "started", nil }}, + {Name: "slow", Run: func(map[string]any) (any, error) { + go func() { + slowWatched <- State("slow").Watch("", func(c StateChange) error { + <-release + seen <- "slow " + c.Key + return errors.New("the handler's own error") + }) + }() + return "started", nil + }}, + {Name: "refused", Run: func(map[string]any) (any, error) { + return nil, State("gone").Watch("", func(StateChange) error { return nil }) + }}, {Name: "remember", Run: func(a map[string]any) (any, error) { return State("servers").Put(a["key"].(string), a["value"]) }}, @@ -111,6 +125,50 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { t.Fatalf("a change for no watch was answered %v", r) } + // Issue 774 (the provisioner's "names no watch of this bundle"): a change is answered as soon as it is taken, so a + // slow handler never counts against the runtime's bound on the current values; Watch still returns only once the + // current values were handled, and a handler's error stays the bundle's to say. + send(map[string]any{"jsonrpc": "2.0", "id": 60, "method": "tools/call", "params": map[string]any{"name": "slow"}}) + var slowAsk map[string]any + for i := 0; i < 2; i++ { + if m := next(); m["method"] == "mesh/state.watch" { + slowAsk = m + } + } + for i, key := range []string{"a", "b"} { + send(map[string]any{"jsonrpc": "2.0", "id": 61 + i, "method": "mesh/state", "params": map[string]any{"watch": slowAsk["id"], + "change": map[string]any{"state": "slow", "key": key, "op": "put", "value": 1, "current": true}}}) + if r := next(); r["id"] != float64(61+i) || r["error"] != nil { + t.Fatalf("a change to a slow handler was answered %v", r) + } + } + send(map[string]any{"jsonrpc": "2.0", "id": slowAsk["id"], "result": map[string]any{}}) + select { + case err := <-slowWatched: + t.Fatalf("the watch returned before its current values were handled: %v", err) + case <-time.After(50 * time.Millisecond): + } + close(release) + if a, b := <-seen, <-seen; a != "slow a" || b != "slow b" { + t.Fatalf("handled out of order: %q, %q", a, b) + } + if err := <-slowWatched; err != nil { + t.Fatal(err) + } + // A watch the runtime refuses is given up, and a change the runtime still hands to it is refused saying so. + send(map[string]any{"jsonrpc": "2.0", "id": 70, "method": "tools/call", "params": map[string]any{"name": "refused"}}) + goneAsk := next() + send(map[string]any{"jsonrpc": "2.0", "id": goneAsk["id"], "error": map[string]any{"code": -32000, + "message": "the current values of gone did not arrive in 10s"}}) + if r := next(); r["id"] != float64(70) { + t.Fatalf("the refused watch answered %v", r) + } + send(map[string]any{"jsonrpc": "2.0", "id": 71, "method": "mesh/state", "params": map[string]any{"watch": goneAsk["id"], + "change": map[string]any{"state": "gone", "key": "k", "op": "put"}}}) + if r := next(); r["id"] != float64(71) || !strings.Contains(fmt.Sprint(r), "gave the watch") || !strings.Contains(fmt.Sprint(r), "did not arrive in 10s") { + t.Fatalf("a change for a given-up watch was answered %v", r) + } + // Put: asked of the runtime as the module, the revision is the tool's answer. send(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": map[string]any{"name": "remember", "arguments": map[string]any{"key": "all.two", "value": map[string]any{"url": "http://two"}}}}) diff --git a/go/stdio.go b/go/stdio.go index 82568f6..71ab95b 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -69,9 +69,11 @@ type channel struct { // subscribed numbers the listeners, so one whose subscribe failed is the one removed. subscribed int // watches are this bundle's state watches by the id each was asked with (ADR 0201). - watches map[string]func(StateChange) error + watches map[string]*follower[StateChange] // provisioned are this bundle's watches of its values, by the id each was asked with (provision.go). - provisioned map[string]func(ProvisionChange) error + provisioned map[string]*follower[ProvisionChange] + // givenUp are the watches this bundle gave up, by id, each with why (follower.go). + givenUp map[string]string // logWatches are this bundle's watches of a log, by the id each was asked with (log.go). logWatches map[string]func(LogEntry) error // takers and provers are this bundle's seat work and proof handlers, by worker and subject (ADR 0259). @@ -183,8 +185,8 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { for _, t := range tools { byName[t.Name] = t } - c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}, watches: map[string]func(StateChange) error{}, - provisioned: map[string]func(ProvisionChange) error{}, logWatches: map[string]func(LogEntry) error{}, + c := &channel{enc: json.NewEncoder(out), waiting: map[string]chan message{}, watches: map[string]*follower[StateChange]{}, + provisioned: map[string]*follower[ProvisionChange]{}, logWatches: map[string]func(LogEntry) error{}, takers: map[string]func(Work) error{}, provers: map[string]func(string, json.RawMessage) (any, error){}} currentMu.Lock() current = c @@ -296,27 +298,23 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { c.say(map[string]any{"id": id, "result": map[string]any{}}) }(id, *p.Envelope, matching) case "mesh/state": - // One change to a state this bundle watches (ADR 0201), named by the watch's id, answered - // once its handler has taken it. Beside the loop, like a tool call. + // One change to a state this bundle watches (ADR 0201), named by the watch's id: taken into the watch's + // follower in the order it came and answered at once, so a slow handler never counts against the runtime's + // bound on the current values (follower.go, issue 774). var p struct { Watch string `json:"watch"` Change *StateChange `json:"change"` } _ = json.Unmarshal(m.Params, &p) c.mu.Lock() - handle := c.watches[p.Watch] + f := c.watches[p.Watch] c.mu.Unlock() - if handle == nil || p.Change == nil { - c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/state names no watch of this bundle"}}) + if f == nil || p.Change == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": c.noWatch("mesh/state", p.Watch)}}) continue } - go func(id any, change StateChange) { - if err := handle(change); err != nil { - c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) - return - } - c.say(map[string]any{"id": id, "result": map[string]any{}}) - }(id, *p.Change) + f.take(*p.Change) + go c.say(map[string]any{"id": id, "result": map[string]any{}}) case "mesh/log": // One entry of a log this bundle watches (log.go), named by the watch's id, answered once its handler has // taken it, in order: the runtime sends the next only after this answer. @@ -340,27 +338,22 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { c.say(map[string]any{"id": id, "result": map[string]any{}}) }(id, *p.Entry) case "mesh/provision": - // One change to a value of this module's that this bundle watches (provision.go), named by the - // watch's id, answered once its handler has taken it. Beside the loop, like a tool call. + // One change to a value of this module's that this bundle watches (provision.go), named by the watch's id: + // taken into the watch's follower and answered at once, as a state's change is (follower.go, issue 774). var p struct { Watch string `json:"watch"` Change *ProvisionChange `json:"change"` } _ = json.Unmarshal(m.Params, &p) c.mu.Lock() - handle := c.provisioned[p.Watch] + f := c.provisioned[p.Watch] c.mu.Unlock() - if handle == nil || p.Change == nil { - c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": "mesh/provision names no watch of this bundle"}}) + if f == nil || p.Change == nil { + c.say(map[string]any{"id": id, "error": map[string]any{"code": -32602, "message": c.noWatch("mesh/provision", p.Watch)}}) continue } - go func(id any, change ProvisionChange) { - if err := handle(change); err != nil { - c.say(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}}) - return - } - c.say(map[string]any{"id": id, "result": map[string]any{}}) - }(id, *p.Change) + f.take(*p.Change) + go c.say(map[string]any{"id": id, "result": map[string]any{}}) case "mesh/work": // One piece of a seat's work this bundle takes (novox/hq ADR 0259): taken by an answer that is // not an error, offered again after one. Beside the loop, like a tool call.