From 0211844e6ad31523bc41b4a557db5f8b7dd65ead Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 24 Oct 2026 07:42:08 +0200 Subject: [PATCH] Bound a state watch's hand-over by its progress, not its whole set, so a large state's newest keys are handed over (issue 1726) The gitea module's watch of mesh-delivery-checker.owed failed at every attempt since 03:05 on novox: about 200 owed keys, each a round trip to the bundle, outlast a 10 s bound for the whole set. The bundle dropped its handler, the oldest keys were written again every 40 s and the newest, handed over last, never were: group fix/1937-done-key-blobs's pass never reached mesh/check. StateWatch now fails only when no current value is handed over within StateTimeout. --- node-tools/internal/bus/state.go | 43 +++++- node-tools/internal/bus/state_watch_test.go | 144 ++++++++++++++++++++ 2 files changed, 180 insertions(+), 7 deletions(-) create mode 100644 node-tools/internal/bus/state_watch_test.go diff --git a/node-tools/internal/bus/state.go b/node-tools/internal/bus/state.go index e0125bd..e2db7b3 100644 --- a/node-tools/internal/bus/state.go +++ b/node-tools/internal/bus/state.go @@ -261,7 +261,12 @@ func (c *Conn) writable(module, name string) (StateIssued, error) { // StateWatch hands deliver the current value of every key matching the pattern — none that is // deleted — and then every change, in order (ADR 0201). It returns once the current values are -// delivered; deliver is called from one goroutine, one change at a time, and an error from it is said +// delivered. **StateTimeout bounds the time without progress, not the whole set** (novox/hq issue +// 1726): each current value is a round trip to the bundle, so a state of a few hundred keys outlasts +// any fixed bound for all of them; the watch then failed at every attempt, the bundle dropped its +// handler, and the newest keys, handed over last, were never taken (the checker's owed, whose newest +// verdicts never reached mesh/check). It fails only when no current value arrives, or one is not +// taken, within StateTimeout; deliver is called from one goroutine, one change at a time, and an error from it is said // and the watch goes on: state is not a queue, and the next change, or the next start, reads it again. // The pattern is a key, with `*` for one name and `**` for the rest; empty is every key. func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange) error) (stop func(), err error) { @@ -306,6 +311,14 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange }) } current := make(chan struct{}) + // handed says a current value was handed over or passed by: progress, which restarts the bound. + handed := make(chan struct{}, 1) + progress := func() { + select { + case handed <- struct{}{}: + default: + } + } go func() { initial := true for e := range w.Updates() { @@ -326,6 +339,7 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange // A deletion among the current values is a key that is not there: not handed over // (measured, research 024 — the server sends its marker among the initial values). if initial { + progress() continue } change.Op = "delete" @@ -334,17 +348,32 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange c.Logf("[mesh-tools] %s did not take %s %s of %s: %v; its next change, or its next start, reads it again", module, change.Op, change.Key, name, err) } + if initial { + progress() + } } if initial { close(current) } }() - select { - case <-current: - return stop, nil - case <-time.After(StateTimeout): - stop() - return nil, fmt.Errorf("the current values of %s did not arrive in %s", name, StateTimeout) + bound := time.NewTimer(StateTimeout) + defer bound.Stop() + for { + select { + case <-current: + return stop, nil + case <-handed: + if !bound.Stop() { + select { + case <-bound.C: + default: + } + } + bound.Reset(StateTimeout) + case <-bound.C: + stop() + return nil, fmt.Errorf("the current values of %s did not arrive in %s: none was handed over in that time", name, StateTimeout) + } } } diff --git a/node-tools/internal/bus/state_watch_test.go b/node-tools/internal/bus/state_watch_test.go new file mode 100644 index 0000000..7e0e842 --- /dev/null +++ b/node-tools/internal/bus/state_watch_test.go @@ -0,0 +1,144 @@ +//go:build integration + +package bus_test + +import ( + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// **A watch whose current values take longer than the bound to hand over still holds** (novox/hq issue +// 1726). The gitea module watches the checker's owed: one key per pull request head, each a round trip +// to the bundle. With a few hundred keys the hand-over outlasted StateTimeout at every attempt, the +// watch was refused, and the newest keys, handed over last, never reached mesh/check: a group's passing +// verdict was never written on its members' heads. The bound is on progress: a watch fails only when +// no current value is handed over within it. +func TestAWatchHoldsWhileItsCurrentValuesKeepArriving(t *testing.T) { + was := bus.StateTimeout + bus.StateTimeout = 400 * time.Millisecond + t.Cleanup(func() { bus.StateTimeout = was }) + + conn, kv := owedWatcher(t) + // A group of three pull requests judged last, after many older heads: their keys come last. + const older = 30 + var keys []string + for i := 0; i < older; i++ { + keys = append(keys, fmt.Sprintf("novox/mesh-catalog.%d.%040x.status=mesh/check", 400+i, i)) + } + for _, member := range []string{"novox/mesh-sdk.76.34f210e5", "novox/mesh-controller.453.c06f75d0", "novox/mesh-delivery.33.28ee22ef"} { + keys = append(keys, member+strings.Repeat("0", 32)+".status=mesh/check") + } + for _, k := range keys { + if _, err := kv.Put(k, []byte(`{"v":1,"act":"status","context":"mesh/check","state":"success"}`)); err != nil { + t.Fatal(err) + } + } + + var mu sync.Mutex + var current, live []string + // Each value takes 40 ms to be taken: 33 values take about 1.3 s, three times the bound. + stop, err := conn.StateWatch("gitea", "mesh-delivery-checker.owed", "", func(ch bus.StateChange) error { + if ch.Current { + time.Sleep(40 * time.Millisecond) + } + mu.Lock() + defer mu.Unlock() + if ch.Current { + current = append(current, ch.Key) + } else { + live = append(live, ch.Key) + } + return nil + }) + if err != nil { + t.Fatalf("a watch whose current values kept arriving was refused: %v", err) + } + t.Cleanup(stop) + mu.Lock() + got := append([]string(nil), current...) + mu.Unlock() + if len(got) != len(keys) { + t.Fatalf("the watch returned after %d of %d current values", len(got), len(keys)) + } + for i, k := range keys { + if got[i] != k { + t.Fatalf("current value %d is %s, want %s", i, got[i], k) + } + } + + // And it holds: a verdict written afterwards reaches it as a change. + next := "novox/mesh-sdk.77." + strings.Repeat("a", 40) + ".status=mesh/check" + if _, err := kv.Put(next, []byte(`{"v":1,"act":"status","context":"mesh/check","state":"pending"}`)); err != nil { + t.Fatal(err) + } + mt.Until(t, func() error { + mu.Lock() + defer mu.Unlock() + if len(live) != 1 || live[0] != next { + return fmt.Errorf("live changes so far: %v", live) + } + return nil + }) +} + +// A watch whose bundle takes nothing for longer than the bound is still refused, so a bundle that hangs +// is said, not waited on for ever. +func TestAWatchIsRefusedWhenNoCurrentValueIsTakenWithinTheBound(t *testing.T) { + was := bus.StateTimeout + bus.StateTimeout = 300 * time.Millisecond + t.Cleanup(func() { bus.StateTimeout = was }) + + conn, kv := owedWatcher(t) + for i := 0; i < 3; i++ { + if _, err := kv.Put(fmt.Sprintf("novox/hq.%d.%040x.status=mesh/check", i+1, i), []byte(`{"v":1}`)); err != nil { + t.Fatal(err) + } + } + release := make(chan struct{}) + t.Cleanup(func() { close(release) }) + began := time.Now() + _, err := conn.StateWatch("gitea", "mesh-delivery-checker.owed", "", func(bus.StateChange) error { + <-release + return nil + }) + if err == nil || !strings.Contains(err.Error(), "did not arrive in") { + t.Fatalf("a watch whose first value was never taken: want refused, got %v", err) + } + if took := time.Since(began); took > 3*time.Second { + t.Fatalf("the refusal took %s, far past the bound", took) + } +} + +// owedWatcher is the gitea module's view of the checker's owed: the bucket, and a runtime connection +// that serves gitea with the owed issued for reading. +func owedWatcher(t *testing.T) (*bus.Conn, interface { + Put(string, []byte) (uint64, error) +}) { + t.Helper() + mesh := mt.New(t) + kv := mesh.Bucket(t, "mesh-delivery-checker_owed") + gitea := mt.MembershipOf("gitea", "anchor", false, nil) + gitea.State = []bus.StateIssued{{Name: "mesh-delivery-checker.owed", Bucket: "mesh-delivery-checker_owed"}} + mesh.Issue(t, gitea) + conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + conn.Logf = func(string, ...any) {} + t.Cleanup(conn.Close) + conn.Follow("gitea") + mt.Until(t, func() error { + if conn.Membership("gitea") == nil { + return errors.New("gitea not issued yet") + } + return nil + }) + return conn, kv +}