diff --git a/internal/link/hearing_nats.go b/internal/link/hearing_nats.go index 93209a9..e9f9fba 100644 --- a/internal/link/hearing_nats.go +++ b/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) +} diff --git a/internal/link/watch.go b/internal/link/watch.go index 948a675..621ef6a 100644 --- a/internal/link/watch.go +++ b/internal/link/watch.go @@ -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) } diff --git a/internal/link/watch_coalesce_test.go b/internal/link/watch_coalesce_test.go new file mode 100644 index 0000000..2678be9 --- /dev/null +++ b/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 +}