The node-engine's bus watch never holds the client back: it keeps the newest update per key while its reader is busy (issue 1966)

Live fault 2026-11-08: node-engines on two machines logged only "nats: slow consumer, messages dropped" and applied nothing. The client's watch sends into its channel from inside the subscription's callback; KVWatch forwarded that channel to its reader blocking, and the reader (composing for minutes) stopped draining, so the subscription's pending limit overflowed. Quick fix: a larger buffer on the watch's channel; it only moves the limit. Thorough fix (chosen): the forwarder always drains the client's channel and, while its reader is busy, coalesces the pending updates by key (newest value and revision, first-seen order, Caught kept in its place), so no reader's pace can back the client up, whatever the burst. A slow consumer error now names its subscription and what it holds pending. Claude-Session: https://claude.ai/code/session_01NcbgfFfQ4Z9UE5jK32PmE5
2026-11-08 00:26:33 +01:00
parent 70e70eda5a
commit fe9533c186
3 changed files with 140 additions and 24 deletions
@@ -257,11 +257,11 @@ func (l *natsLink) HearMoveSecrets(handle MoveSecretFunc) {
 }

 // asyncError is said by the client as before, and a subscription the bus refused is handed to serve (Refused).
-func (l *natsLink) asyncError(_ *nats.Conn, _ *nats.Subscription, err error) {
+func (l *natsLink) asyncError(_ *nats.Conn, sub *nats.Subscription, err error) {
 	if err == nil {
 		return
 	}
-	fmt.Fprintf(os.Stderr, "%v\n", err)
+	fmt.Fprintf(os.Stderr, "%v%s\n", err, subscriptionDetail(sub))
 	if subject, ok := RefusedSubscription(err); ok && l.refused != nil {
 		select {
 		case l.refused <- subject:
@@ -368,3 +368,16 @@ func RefusedSubscription(err error) (string, bool) {

 // User is the bus user this link authenticated as.
 func (l *natsLink) User() string { return l.user }
+
+// subscriptionDetail names the subscription an asynchronous error came from and how much it holds undelivered, so a slow
+// consumer says which one it is (issue 1966).
+func subscriptionDetail(sub *nats.Subscription) string {
+	if sub == nil {
+		return ""
+	}
+	msgs, bytes, err := sub.Pending()
+	if err != nil {
+		return fmt.Sprintf(" (subscription %s)", sub.Subject)
+	}
+	return fmt.Sprintf(" (subscription %s: %d messages, %d bytes pending)", sub.Subject, msgs, bytes)
+}
@@ -52,32 +52,67 @@ func (b OverNATS) KVWatch(ctx context.Context, bucket, pattern string) (<-chan K
 	go func() {
 		defer close(out)
 		defer func() { _ = w.Stop() }()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case e, ok := <-w.Updates():
-				if !ok {
-					return
-				}
-				var u KVUpdate
-				if e == nil {
-					u.Caught = true
-				} else {
-					u = KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(),
-						Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge}
-				}
-				select {
-				case out <- u:
-				case <-ctx.Done():
-					return
-				}
-			}
-		}
+		forwardCoalescing(ctx, w.Updates(), entryUpdate, out)
 	}()
 	return out, nil
 }

+// entryUpdate is the update a client's entry is; the nil entry the client sends after the last key there was is Caught.
+func entryUpdate(e nats.KeyValueEntry) KVUpdate {
+	if e == nil {
+		return KVUpdate{Caught: true}
+	}
+	return KVUpdate{Key: e.Key(), Value: e.Value(), Revision: e.Revision(),
+		Deleted: e.Operation() == nats.KeyValueDelete || e.Operation() == nats.KeyValuePurge}
+}
+
+// forwardCoalescing hands in's items, as updates, to out, and never lets out hold in back: the client's watch sends into
+// its channel from inside the subscription's callback, so a reader that is not draining (a node-engine composing for
+// minutes) overflows the subscription's pending limit and the bus drops messages (issue 1966). While out is not taken,
+// the pending updates are kept by key (the newest value and revision of each, in the order the keys were first seen) and
+// the Caught marker keeps its place among the keys queued before it. It returns when ctx ends, or when in is closed and
+// what is pending is handed over.
+func forwardCoalescing[T any](ctx context.Context, in <-chan T, conv func(T) KVUpdate, out chan<- KVUpdate) {
+	var queue []*KVUpdate
+	byKey := map[string]*KVUpdate{}
+	add := func(u KVUpdate) {
+		if !u.Caught {
+			if q, ok := byKey[u.Key]; ok {
+				*q = u
+				return
+			}
+		}
+		q := &u
+		queue = append(queue, q)
+		if !u.Caught {
+			byKey[u.Key] = q
+		}
+	}
+	source := in
+	for source != nil || len(queue) > 0 {
+		var send chan<- KVUpdate
+		var head KVUpdate
+		if len(queue) > 0 {
+			send, head = out, *queue[0]
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case x, ok := <-source:
+			if !ok {
+				source = nil
+				continue
+			}
+			add(conv(x))
+		case send <- head:
+			if !head.Caught {
+				delete(byKey, head.Key)
+			}
+			queue = queue[1:]
+		}
+	}
+}
+
 func (l *natsLink) KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error) {
 	return OverNATS{Conn: l.conn, JS: l.js}.KVWatch(ctx, bucket, pattern)
 }
@@ -0,0 +1,68 @@
+package link
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+func TestForwardCoalescing(t *testing.T) {
+	u := func(k string, rev uint64) KVUpdate { return KVUpdate{Key: k, Value: []byte{byte(rev)}, Revision: rev} }
+	caught := KVUpdate{Caught: true}
+	tests := []struct {
+		name string
+		in   []KVUpdate
+		want []KVUpdate
+	}{
+		{"newest per key, first-seen order", []KVUpdate{u("a", 1), u("b", 2), u("a", 3), u("b", 4), u("a", 5)}, []KVUpdate{u("a", 5), u("b", 4)}},
+		{"caught keeps its place", []KVUpdate{u("a", 1), u("b", 2), caught, u("c", 3), u("a", 4)}, []KVUpdate{u("a", 4), u("b", 2), caught, u("c", 3)}},
+		{"delete is the newest", []KVUpdate{u("a", 1), {Key: "a", Revision: 2, Deleted: true}}, []KVUpdate{{Key: "a", Revision: 2, Deleted: true}}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			in := make(chan KVUpdate)  // unbuffered: every send must be taken at once, as the client's callback needs
+			out := make(chan KVUpdate) // unbuffered and not read yet
+			done := make(chan struct{})
+			go func() { forwardCoalescing(ctx, in, func(x KVUpdate) KVUpdate { return x }, out); close(done) }()
+			for _, x := range tt.in {
+				select {
+				case in <- x:
+				case <-time.After(5 * time.Second):
+					t.Fatal("the producer was held back by a reader that does not read")
+				}
+			}
+			close(in)
+			var got []KVUpdate
+			for x := range out2(out, done) {
+				got = append(got, x)
+			}
+			if len(got) != len(tt.want) {
+				t.Fatalf("got %d updates %v, want %d %v", len(got), got, len(tt.want), tt.want)
+			}
+			for i := range got {
+				if got[i].Key != tt.want[i].Key || got[i].Revision != tt.want[i].Revision || got[i].Caught != tt.want[i].Caught || got[i].Deleted != tt.want[i].Deleted {
+					t.Fatalf("update %d is %+v, want %+v", i, got[i], tt.want[i])
+				}
+			}
+		})
+	}
+}
+
+// out2 reads out until the forwarder has ended.
+func out2(out <-chan KVUpdate, done <-chan struct{}) <-chan KVUpdate {
+	r := make(chan KVUpdate)
+	go func() {
+		defer close(r)
+		for {
+			select {
+			case x := <-out:
+				r <- x
+			case <-done:
+				return
+			}
+		}
+	}()
+	return r
+}