manager/logbroker: bound the log and subscription queues

[Leekyungun]

• relates to Memory leak of docked process moby#46068

- What I did

Bounded the log broker's two queues so that a consumer which stops draining can
no longer cause unbounded manager memory growth.

LogBroker.Start creates both queues with watch.NewQueue() and no options,
which yields a LimitQueue with limit: 0 -- documented as "limitless".
LimitQueue.Write skips its size check entirely in that case:

if eq.limit > 0 && uint64(eq.events.Len()) >= eq.limit {
    ...
}
eq.events.PushBack(event)

Each watcher's sink is an unbuffered channel (events.NewChannel(0) in
watch/sinks.go), so once a watcher stops reading, the queue keeps appending to
an unbounded container/list rather than applying backpressure or shedding
load.

ListenSubscriptions stops draining its channel whenever stream.Send blocks,
which happens when an agent's gRPC stream stalls -- an unresponsive or
partitioned node, for example. Both registerSubscription and
unregisterSubscription publish the *subscription to subscriptionQueue, so
a single stalled watcher pins every subscription that has passed through the
broker, along with its SubscriptionMessage, LogSelector,
LogSubscriptionOptions and cancel context -- even though each subscription was
unregistered correctly.

Because the backlog accumulates in a container/list rather than in blocked
goroutines, the growth is invisible to goroutine-count based monitoring and
surfaces only as unexplained manager heap growth.

This matches the profile in moby/moby#46068. We hit it in production: a
single-node swarm manager reached ~27 GB RSS with a normal goroutine count (320)
and normal fd count (143). A heap profile attributed 99.4% of the heap to the log
broker:

_Logs_SubscribeLogs_Handler                     23.33GB  (91.69% cum)
  logbroker.newSubscription (inline)             6.75GB
  logbroker.(*subscription).match.func1          4.40GB
  api.(*SubscribeLogsRequest).Unmarshal          3.83GB
  context.newCancelCtx                           2.39GB
  logbroker.(*LogBroker).ListenSubscriptions     1.73GB
  api.(*LogSubscriptionOptions).Unmarshal        1.65GB
  api.(*LogSelector).Unmarshal                   1.44GB

- How I did it

Gave both queues a limit and WithCloseOutChan(), so a consumer that falls too
far behind is disconnected instead of buffered without bound. Disconnecting is
recoverable: ListenSubscriptions returns an error, the agent reconnects, and
watchSubscriptions replays the currently registered subscriptions for that
node.

Both receive sites now handle a closed channel. Without that, closing the output
channel would turn the leak into a nil type-assertion panic on
v.(*subscription).

- How to test it

go test ./manager/logbroker/ -run TestLogBrokerSubscriptionQueueBounded -v

The added test drives subscriptions through their full lifecycle
(Run -> register -> unregister -> Stop) and uses finalizers to check
whether the runtime can reclaim them. A subscription that has been unregistered
is no longer referenced by any of the broker's bookkeeping maps, so a correctly
behaving broker must allow it to be collected regardless of what the watcher is
doing.

watcher	before	after
draining	3000/3000 reclaimed	3000/3000 reclaimed
stalled	0/3000 reclaimed	2499/3000 reclaimed

Retention after the change is bounded rather than zero, because the queue may
legitimately hold up to subscriptionQueueLimit events. Each subscription
publishes twice (register + unregister), so ~501 subscriptions are retained.
Scaling the input confirms the bound holds:

subscriptions driven through	retained
3,000	501
10,000	501
30,000	501

The test fails on master and passes with this change. Existing
manager/logbroker, watch and watch/queue tests pass; go vet and gofmt
are clean.

- Description for the changelog

Bound the swarm log broker's log and subscription queues so a stalled consumer
can no longer cause unbounded manager memory growth.

---

Open questions

I would appreciate guidance on two points.

1. The limit values are arbitrary. I chose subscriptionQueueLimit = 1000
and logQueueLimit = 10000 based on relative event frequency: subscription
events occur a couple of times per docker service logs invocation, while log
events track container output volume, so a single shared value would trip the
log queue under normal load.

That reasoning only accounts for frequency, not for per-event size. A
PublishLogsMessage carries a batch of log lines and can be far larger than a
*subscription, so 10,000 log events may well cost more memory than 1,000
subscriptions. What we actually want to bound is bytes, but watch.Queue only
offers a count-based limit. Happy to change these numbers, make them
configurable, or take a different approach.

2. Tearing the watcher down may be too blunt for logQueue. For
subscriptionQueue the consumer is an agent, and disconnecting it triggers a
reconnect that re-syncs state. For logQueue the consumer is a
docker service logs client, where falling behind on a slow link is a fairly
normal condition, and the user sees the stream terminate with
ResourceExhausted. watch.WithTimeout may be a better fit there; I kept both
queues consistent for now.

Notes

• watch.WithLimit and watch.WithTimeout exist but are currently unused by any
  production call site; all eight watch.NewQueue() callers pass no options.
  Other queues may warrant the same treatment, but I have limited this change to
  the broker, where there is a reproduction.
• The reproduction is synthetic. It models a watcher that stops draining, which
  is the state a stalled stream.Send leaves ListenSubscriptions in, but it
  does not reproduce the original production trigger (a node going offline and
  returning, as also described in Memory leak of docked process moby#46068).
• Tested on darwin/arm64 with Go 1.25.1. The test relies on finalizers running
  after runtime.GC(); it was stable across 10 consecutive runs, but it would be
  worth confirming the behaviour on CI.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 40.00000% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 14.76%. Comparing base (6e9e7b8) to head (e8b0e19).
⚠️ Report is 23 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #3267      +/-   ##
==========================================
+ Coverage   14.73%   14.76%   +0.03%     
==========================================
  Files         200      200              
  Lines       93077    93101      +24     
==========================================
+ Hits        13712    13746      +34     
+ Misses      78019    78000      -19     
- Partials     1346     1355       +9     

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.