Reduce writer lock contention + unify the writer/medium queue

[FZambia]

Reworks the internal unbounded queue behind the per-connection message
writer and the per-channel channel medium:

1. Reduce lock contention on the hot enqueue path, especially when many
   goroutines (concurrent broadcasts/subscriptions) produce into one
   connection's queue.
2. Unify the two near-identical queues (internal/queue.Queue and
   channel_medium.go's publicationQueue) into one generic queue.Queue[T],
   removing ~150 lines of duplicate code.
3. Fix a goroutine leak in channelMedium.close() (a writer goroutine could
   park forever on an empty queue if the queue wasn't closed).

The queue stays an unbounded, dynamically grown/shrunk ring buffer tuned for
MPSC (many producers, exactly one consumer — how both callers use it).

What changed

internal/queue/queue.go

• Queue → Queue[T any] with a sizeFn func(T) int for byte accounting.
  Item/ItemSize stay for the writer; the medium parameterises it with
  queuedPublication + queuedPublicationSize.
• Lock is a plain sync.Mutex (master used sync.RWMutex). For this
  workload — short critical sections hit by many producers — a sync.Mutex is
  measurably lighter than RWMutex, and it's the main source of the
  high-contention wins below.
• Blocking Wait uses sync.Cond (on that mutex). An earlier iteration of
  this branch used a channel doorbell; profiling showed channel send/recv take
  the runtime hchan lock on every park/unpark and that dominates when the single
  consumer parks frequently — sync.Cond is cheaper, so it was restored.
• cnt/size are plain int64 under the mutex, not atomics (an atomic RMW
  per op bounces its cache line across all producer cores). closed/cp stay
  atomic (written only on close/resize → free lock-free reads for Closed/Cap).
• Add/AddMany return the post-insert byte size, so the writer's
  MaxQueueSize check avoids a second synchronised Size() call (master took
  two locks per enqueue: Add + Size).
• sizeFn is computed outside the critical section in Add.
• waiting flag gates cond.Signal so producers skip the wakeup when the
  consumer isn't parked. No lost wakeup (the flag is set/read under the same
  mutex the consumer uses to check cnt).

writer.go

• enqueue/enqueueMany use the size returned by Add/AddMany.
• timerScheduledFlag atomic.Bool: in timer mode, concurrent producers skip
  the writer mutex while a flush is already pending.
• flush reduced from up to 3 Len() calls to 1.

channel_medium.go

• Uses queue.Queue[queuedPublication]; the bespoke publicationQueue is gone.

Tests / benchmarks

• All queue/writer/medium tests updated; added coverage for close/wake,
  idempotent close, the returned-size contract, and non-Item instantiations.
• Old BenchmarkWriteMerge/BenchmarkWriteMergeDisabled replaced by one
  self-contained suite, BenchmarkWriterQueue (writer_queue_bench_test.go),
  covering every writer path: all flush modes (immediate / merged / delay-batched
  / timer), every maxMessagesInFrame regime (1, N, unlimited), both enqueue
  entry points, the shrink path, and 1 vs 8 producers. Uses only the writer API
  so the identical file runs on master for comparison.

Correctness / safety

MPSC is a hard invariant (single consumer). Close/CloseRemaining use
cond.Broadcast. No lost wakeups (waiting-flag reasoning + close/wake race
tests). Idempotent Close. Whole suite passes -race (-count ≥ 5), go vet
and gofmt clean.

Benchmarks

Methodology — the benchmark models a real writer, which matters enormously:

• Both builds compiled to test binaries and run interleaved (alternating
  master/branch, 12 rounds, -benchtime=300ms) — sequential runs drift far too
  much to compare (±30% on master's own numbers).
• Realistic small initial capacity (16); the ring grows/shrinks under load —
  resize is measured, not disabled.
• The consumer models the network write: one write(2) syscall per frame to
  /dev/null. A real consumer spends its time in that syscall outside the
  queue lock, so it cannot monopolise the lock.
• Producers model the per-message encode (marshaling) they do before
  enqueuing. Real broadcast producers are not infinitely-fast enqueue loops;
  this pacing keeps the consumer able to keep up, so the queue stays shallow
  (frames of tens–~128 messages, never thousands) — exactly how a healthy
  connection behaves. (A connection slow enough to back the queue up to thousands
  is disconnected as slow via MaxQueueSize.)

ns/op is per produced message. Apple M4.

                                         master      branch     vs master
WriterQueue/Simple/producers=1            509.2n      458.2n      -10.01%
WriterQueue/Simple/producers=8           2591.5n      816.6n      -68.49%
WriterQueue/Batched16/producers=1         319.3n      311.6n       -2.43%
WriterQueue/Batched16/producers=8         374.2n      281.6n      -24.76%
WriterQueue/Batched64/producers=1         324.0n      312.1n       -3.69%
WriterQueue/Batched64/producers=8         297.3n      286.9n        ~
WriterQueue/Unlimited/producers=1         328.1n      310.0n       -5.50%
WriterQueue/Unlimited/producers=8         300.6n      278.7n        ~
WriterQueue/DelayBatch16/producers=1      276.0n      278.3n        ~
WriterQueue/DelayBatch16/producers=8      375.4n      254.4n      -32.21%
WriterQueue/Timer16/producers=1           309.4n      309.8n        ~
WriterQueue/Timer16/producers=8           295.9n      253.9n      -14.19%
WriterQueue/TimerUnlimited/producers=1    259.3n      256.3n       -1.16%
WriterQueue/TimerUnlimited/producers=8    254.8n      211.8n      -16.86%
WriterQueue/TimerShrink16/producers=1     290.3n      284.8n       -1.93%
WriterQueue/TimerShrink16/producers=8     263.9n      222.0n      -15.88%
WriterQueue/BatchedMany8/producers=1      278.5n      284.2n        ~
WriterQueue/BatchedMany8/producers=8      315.2n      149.8n      -52.46%
WriterQueue/TimerMany8/producers=1        288.0n      285.3n        ~
WriterQueue/TimerMany8/producers=8        184.0n      166.9n        ~
geomean                                   335.9n      281.6n      -16.16%

Reading the results

• No regressions. Every case is a win or statistically neutral. Geomean
  −16.16%, with the big wins under concurrent producers (the broadcast fan-in
  this targets): Simple/8p −68%, BatchedMany8/8p −52%, DelayBatch16/8p −32%, Batched16/8p −25%, TimerUnlimited/8p −17%, TimerShrink16/8p −16%.
• Single-producer cases are slightly-better-to-neutral (less lock contention to
  remove; the encode cost dominates per-message time there, which is realistic).

How we got here (so the design choices are documented, not guessed)

pprof plus controlled experiments pinned every choice:

• sync.Mutex is the source of the wins — switching the queue to RWMutex
  (master's type) erased them.
• sync.Cond beats a channel doorbell for this MPSC use — pprof showed
  channel send/recv hammering the runtime hchan lock on every park/unpark.
• The generic sizeFn is not a factor — a parallel byte-size ring removed
  the per-item indirect call from the drain and changed nothing (reverted).
• An unrealistic benchmark was manufacturing regressions. A free (no-op)
  transport made the consumer a lock-monopolising spin, and infinitely-fast
  pure-loop producers forced the consumer to drain thousands of items under
  the lock — neither happens in production. Modeling the per-frame write syscall
  and the per-message producer encode (which keeps the queue shallow) made
  the apparent large/unlimited-frame regressions disappear entirely.

Net

A clear win across the realistic operating range — biggest under concurrent
broadcast fan-in, neutral where there's little contention to remove — with no
regressions.

[codecov]

Codecov Report

❌ Patch coverage is 90.21739% with 18 lines in your changes missing coverage. Please review.
✅ Project coverage is 84.17%. Comparing base (101d1ac) to head (2d380ed).
⚠️ Report is 26 commits behind head on master.

Files with missing lines	Patch %	Lines
internal/queue/queue.go	91.77%	8 Missing and 5 partials ⚠️
writer.go	76.19%	3 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #576      +/-   ##
==========================================
- Coverage   84.64%   84.17%   -0.48%     
==========================================
  Files          58       58              
  Lines       14996    14955      -41     
==========================================
- Hits        12693    12588     -105     
- Misses       1644     1703      +59     
- Partials      659      664       +5     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.