internal/link/hearing_nats.go

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

internal/link/pinned_test.go

@@ -61,8 +61,8 @@ func server(t *testing.T) (address string, fingerprint string) {
 	return listener.Addr().String(), Fingerprint(der)
 }
 
-// The right broker is accepted; and a malformed pin is caught before connecting, rather than at the handshake, so a mistyped
-// token fails while a person is looking at it (two tests merged).
+// The right broker is accepted, a different one refused, and a malformed pin is caught before connecting, rather than at
+// the handshake, so a mistyped token fails while a person is looking at it (three tests merged).
 func TestTheRightBrokerIsAcceptedAndAMalformedPinRefused(t *testing.T) {
 	address, pin := server(t)
 	conn, err := dialPinned(address, pin, 5*time.Second)
@@ -76,15 +76,11 @@ func TestTheRightBrokerIsAcceptedAndAMalformedPinRefused(t *testing.T) {
 			t.Errorf("%q was accepted as a fingerprint", bad)
 		}
 	}
-}
 
-func TestADifferentBrokerIsRefused(t *testing.T) {
 	// The case the pin exists for: something else answering at that address. Since the host
 	// applies whatever the link delivers, connecting to the wrong mesh is the whole machine.
-	address, _ := server(t)
 	_, other := server(t)
-
-	_, err := dialPinned(address, other, 5*time.Second)
+	_, err = dialPinned(address, other, 5*time.Second)
 	if err == nil {
 		t.Fatal("a broker presenting a different certificate was accepted")
 	}

internal/link/watch.go

@@ -23,8 +23,11 @@ type KVUpdate struct {
 // KVBus is a link that can watch a bucket's keys: the node agent follows its wants and its compositions with it, and
 // never polls.
 type KVBus interface {
-	// KVWatch yields the keys of the bucket matching the pattern now, a Caught update, then every change, until ctx ends
-	// or the link does (the channel is then closed). A bucket the node's account may not read is an error. The pattern
+	// KVWatch yields the keys of the bucket matching the pattern now, a Caught update, then the changes, until ctx ends
+	// or the link does (the channel is then closed). It never holds the bus client back: while the reader does not take,
+	// a key's changes fold into its newest (a delete included), so a reader sees each key's latest state, not every
+	// change, and a key still queued from before Caught is handed before Caught with its newest value; across keys the
+	// order is the order each key was first queued. A bucket the node's account may not read is an error. The pattern
 	// may be several, separated by spaces (WatchPatterns), watched as one: a node's assignment in either form of its
 	// keys (research 072 row 37).
 	KVWatch(ctx context.Context, bucket, pattern string) (<-chan KVUpdate, error)
@@ -52,32 +55,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)
 }

internal/link/watch_coalesce_test.go

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