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

@@ -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 {
+			if err != nil || stopped.Load() {
-				return // stopped, or the connection closed
+				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.

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)

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