From 21c86c819c0bdeb45f3308b40cf994468540dd1a Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:33:32 +0100 Subject: [PATCH] Follow the runtime's watches through mesh-sdk's KeptState.Follow, not four loops of their own (issue 774) mesh-provisioner, mesh-config, mesh-issues and mesh-healers each wrote a watch-again loop (`watched`). mesh-sdk 61b67d61 adds KeptState.Follow, which asks again with backoff and says each give-up, and its watches now answer a change at once and run the handler beside the runtime (the provisioner's "names no watch of this bundle"). The four modules use it; their loops are gone. Built and tested against the mesh-sdk group head through a go.work that is not committed: their mesh-sdk pins move to that head with the group. --- modules/mesh-config/cmd/mesh-config/main.go | 12 ++-------- modules/mesh-healers/cmd/mesh-healers/main.go | 18 +++------------ modules/mesh-issues/cmd/mesh-issues/main.go | 10 ++------- .../cmd/mesh-provisioner/main.go | 22 +++++-------------- 4 files changed, 12 insertions(+), 50 deletions(-) diff --git a/modules/mesh-config/cmd/mesh-config/main.go b/modules/mesh-config/cmd/mesh-config/main.go index c0bba92e..c0bff6a7 100644 --- a/modules/mesh-config/cmd/mesh-config/main.go +++ b/modules/mesh-config/cmd/mesh-config/main.go @@ -134,16 +134,8 @@ func resolving(c *Config) { // every change of what a module declares where (issue 1953); the builds themselves are read, not watched. for _, read := range []string{PlacementRead, WantsRead, ModulesRead, PredefinedRead, IndexerPredefinedRead, CandidatesRead, MirrorDeclaredRead, TakeoverState} { - go func(read string) { - for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) - if err := stdio.State(read).Watch("", poke); err != nil { - logf("[mesh-config] %s is not watched yet (%v); asking again in %s", read, err, wait) - continue - } - return - } - }(read) + // Asked again until the runtime takes it, each give-up said (mesh-sdk's Follow). + go stdio.State(read).Follow("", poke) } tick := time.NewTicker(5 * time.Minute) defer tick.Stop() diff --git a/modules/mesh-healers/cmd/mesh-healers/main.go b/modules/mesh-healers/cmd/mesh-healers/main.go index 6e48bfba..fb1e50b5 100644 --- a/modules/mesh-healers/cmd/mesh-healers/main.go +++ b/modules/mesh-healers/cmd/mesh-healers/main.go @@ -52,18 +52,6 @@ func (runtimeStore) Delete(state, key string) error { return stdio.State(state). func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } -// watched asks a watch of the runtime again until it is taken: the runtime answers only once Serve runs. -func watched(state, key string, handle func(stdio.StateChange) error) { - for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) - if err := stdio.State(state).Watch(key, handle); err != nil { - logf("[%s] %s is not watched yet (%v); asking again in %s", Module, state, err, wait) - continue - } - return - } -} - func main() { h := &Healers{Store: runtimeStore{}, Registry: Registry(), Now: time.Now, Logf: logf, Ask: func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }, @@ -71,16 +59,16 @@ func main() { // The lease first: until it is read the healers write would-acts only. Once it is first read, a settle that ended // while no instance watched is judged (Rearm); the waits still running are keys on the bus, heard on their expiry. var started sync.Once - go watched(LeaseRead, LeaseKey, func(c stdio.StateChange) error { + go stdio.State(LeaseRead).Follow(LeaseKey, func(c stdio.StateChange) error { h.LeaseHeard(c.Value, c.Op == "put", nil) started.Do(func() { go h.Rearm() }) return nil }) - go watched(WakesState, "", func(c stdio.StateChange) error { + go stdio.State(WakesState).Follow("", func(c stdio.StateChange) error { h.WakeHeard(c.Key, c.Op) return nil }) - go watched(HandActsRead, "", func(c stdio.StateChange) error { + go stdio.State(HandActsRead).Follow("", func(c stdio.StateChange) error { h.HandActHeard(c.Key, c.Op, c.Value) return nil }) diff --git a/modules/mesh-issues/cmd/mesh-issues/main.go b/modules/mesh-issues/cmd/mesh-issues/main.go index ce7a0a4a..08654a2d 100644 --- a/modules/mesh-issues/cmd/mesh-issues/main.go +++ b/modules/mesh-issues/cmd/mesh-issues/main.go @@ -132,14 +132,8 @@ func main() { // reads the bucket, and a write carries the issue it changes first, so no write reaches the log before its // issue's carry. func start(t *Tracker) { - for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) - if err := stdio.State(IssuesState).Watch("", t.Watched); err != nil { - logf("[mesh-issues] the bucket is not watched yet (%v); asking again in %s", err, wait) - continue - } - break - } + // Asked again until the runtime takes it, each give-up said (mesh-sdk's Follow). + stdio.State(IssuesState).Follow("", t.Watched) for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { s, err := t.Start() if err == nil { diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go index cca4a622..95d795be 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go @@ -103,34 +103,22 @@ func (w *work) run(h *Holder) { } } -// watched asks a watch of the runtime again until it is taken: the runtime answers only once Serve runs. -func watched(state string, handle func(stdio.StateChange) error) { - for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - time.Sleep(wait) - if err := stdio.State(state).Watch("", handle); err != nil { - logf("[%s] %s is not watched yet (%v); asking again in %s", Module, state, err, wait) - continue - } - return - } -} - func main() { s := runtimeStore{} h := &Holder{Store: s, Compose: FromBus(), Asks: runtimeAsks{}, Now: time.Now, Logf: logf} w := &work{pending: map[string]bool{}, wake: make(chan struct{}, 1)} go w.run(h) go starting(h) - go watched(PlacementRead, func(c stdio.StateChange) error { w.add(NodesOfPlacement(c.Key)...); return nil }) - go watched(WantsRead, func(c stdio.StateChange) error { w.add(NodesOfWant(c.Key)...); return nil }) - go watched(ModulesRead, func(c stdio.StateChange) error { + go stdio.State(PlacementRead).Follow("", func(c stdio.StateChange) error { w.add(NodesOfPlacement(c.Key)...); return nil }) + go stdio.State(WantsRead).Follow("", func(c stdio.StateChange) error { w.add(NodesOfWant(c.Key)...); return nil }) + go stdio.State(ModulesRead).Follow("", func(c stdio.StateChange) error { if module, err := delivery.ParseModulesKey(c.Key); err == nil { w.add(h.NodesOfModule(module)...) } return nil }) for _, provider := range BoundReads { - go watched(boundRead(provider), func(c stdio.StateChange) error { + go stdio.State(boundRead(provider)).Follow("", func(c stdio.StateChange) error { if said, err := h.Rejudge(c.Key); err != nil { logf("[%s] %s is not judged again: %v", Module, c.Key, err) } else if said != "" { @@ -139,7 +127,7 @@ func main() { return nil }) } - go watched(Registry, func(c stdio.StateChange) error { + go stdio.State(Registry).Follow("", func(c stdio.StateChange) error { if c.Op != "put" { if _, _, err := Publish(s); err != nil { logf("[%s] the registry is not written back: %v", Module, err)