Merge pull request 'Bound a state watch's hand-over by its progress, not its whole set, so the gitea module binds the checker's owed statuses again (issue 1726)' (#41) from fix/1726-group-check-status into main

Reviewed-on: #41
2026-10-24 10:35:43 +00:00
parent 0e985f9bf1 0211844e6a
commit 77b05089c8
2 changed files with 180 additions and 7 deletions
@@ -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)
+		}
 	}
 }

@@ -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
+}