From 316a5a2fc31793cdcb08329b5a07fb79a2e404f2 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:22:01 +0100 Subject: [PATCH] A watch given up or stopped hands over nothing more; arrival of current values restarts its bound (issue 774) Quick: check the watch context before each deliver in StateWatch. Thorough (chosen, same file kind): also LogWatch drops a message fetched as it stopped, and the bound restarts on arrival, not only after the bundle took a value. --- node-tools/internal/bus/log.go | 8 ++- node-tools/internal/bus/state.go | 14 +++++ node-tools/internal/bus/state_watch_test.go | 70 +++++++++++++++++++++ 3 files changed, 89 insertions(+), 3 deletions(-) diff --git a/node-tools/internal/bus/log.go b/node-tools/internal/bus/log.go index 423d0c7..f7a49af 100644 --- a/node-tools/internal/bus/log.go +++ b/node-tools/internal/bus/log.go @@ -9,6 +9,7 @@ import ( "sort" "strconv" "strings" + "sync/atomic" "time" "github.com/nats-io/nats.go" @@ -381,18 +382,19 @@ func (c *Conn) LogWatch(module, name, keys string, from uint64, since time.Time, return nil, fmt.Errorf("the watch of %s: %w", name, err) } } + var stopped atomic.Bool go func() { for { m, err := it.Next() - if err != nil { - return // stopped, or the connection closed + if err != nil || stopped.Load() { + return // stopped, or the connection closed; a message fetched as it stopped is not handed over } if err := hand(m); err != nil { c.Logf("[mesh-tools] %s did not take entry of %s: %v; it is in the log, a read finds it", module, name, err) } } }() - return it.Stop, nil + return func() { stopped.Store(true); it.Stop() }, nil } // LogLast is a key's last entry, or nil when it has none. diff --git a/node-tools/internal/bus/state.go b/node-tools/internal/bus/state.go index 7185c08..1f80af5 100644 --- a/node-tools/internal/bus/state.go +++ b/node-tools/internal/bus/state.go @@ -423,6 +423,12 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange go func() { initial := true for e := range w.Updates() { + // A watch given up, or stopped, hands over nothing more, though values stay buffered in the + // watcher: they would reach a watch id the bundle no longer holds (issue 774: 1257 orphan + // changes after 51 failed watches). + if watching.Err() != nil { + break + } if e == nil { // The end of what was there when the watch began. if initial { @@ -431,6 +437,11 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange } continue } + if initial { + // Arrival from the bus is progress before the bundle is asked: the bound measures the current + // values coming in, and restarts when the bundle has taken one. + progress() + } change := StateChange{State: name, Key: e.Key(), Revision: e.Revision(), Current: initial} switch e.Operation() { case jetstream.KeyValuePut: @@ -451,6 +462,9 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange change.Op = "expire" } } + if watching.Err() != nil { + break + } if err := deliver(change); err != nil { 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) diff --git a/node-tools/internal/bus/state_watch_test.go b/node-tools/internal/bus/state_watch_test.go index 5ca378e..81dbc43 100644 --- a/node-tools/internal/bus/state_watch_test.go +++ b/node-tools/internal/bus/state_watch_test.go @@ -7,6 +7,7 @@ import ( "fmt" "strings" "sync" + "sync/atomic" "testing" "time" @@ -192,3 +193,72 @@ func owedWatcher(t *testing.T) (*bus.Conn, interface { }) return conn, kv } + +// **A watch given up, or stopped, hands over nothing more** (issue 774). The runtime's watch stopped its watcher +// when the current values did not arrive within the bound, but the watcher still held values, and its goroutine +// went on handing them over: to a watch id the bundle no longer held (on the provisioner, 51 failed watches were +// followed by 1257 orphan changes). Each row blocks the bundle on its first value, gives the watch up, lets the +// bundle go and counts what it is still handed. +func TestAWatchGivenUpHandsOverNothingMore(t *testing.T) { + for _, tc := range []struct { + name string + refused bool // the bound gives it up, or the caller stops it + }{ + {"refused at the bound", true}, + {"stopped by its caller", false}, + } { + t.Run(tc.name, func(t *testing.T) { + was := bus.StateTimeout + bus.StateTimeout = 300 * time.Millisecond + t.Cleanup(func() { bus.StateTimeout = was }) + conn, kv := owedWatcher(t) + put := func(i int) { + if _, err := kv.Put(fmt.Sprintf("novox/hq.%d.%040x.status=mesh/check", i+1, i), []byte(`{"v":1}`)); err != nil { + t.Fatal(err) + } + } + if tc.refused { + for i := 0; i < 4; i++ { + put(i) + } + } + var calls atomic.Int32 + first := make(chan struct{}, 1) + release := make(chan struct{}) + deliver := func(ch bus.StateChange) error { + if calls.Add(1) == 1 { + first <- struct{}{} + <-release + } + return nil + } + var stop func() + if tc.refused { + if _, err := conn.StateWatch("gitea", "mesh-delivery-checker.owed", "", deliver); err == nil { + t.Fatal("the watch whose bundle took nothing was not refused") + } + } else { + var err error + if stop, err = conn.StateWatch("gitea", "mesh-delivery-checker.owed", "", deliver); err != nil { + t.Fatal(err) + } + for i := 0; i < 4; i++ { + put(i) + } + select { + case <-first: + case <-time.After(5 * time.Second): + t.Fatal("the bundle was handed no change") + } + stop() + } + // Let the watcher's server side deliver what it held, then free the bundle. + time.Sleep(300 * time.Millisecond) + close(release) + time.Sleep(500 * time.Millisecond) + if n := calls.Load(); n != 1 { + t.Fatalf("the bundle was handed %d changes of a watch given up, want 1 (the one in hand)", n) + } + }) + } +}