Run client-side stats through WorkQueue

[dougqh]

Stacked on #12313. Trial branch — not a merge candidate as it stands. Its purpose is to give #12313 a real caller, so the API is reviewed against use rather than on its own terms.

Client-side stats is currently the only prospective adopter of WorkQueue anywhere in the tree: every other queue user (TraceProcessingWorker, SpanSamplingWorker, PendingTraceBuffer, OkHttpSink, DefaultDataStreamsMonitoring, LLMObsIntakeWorker, BuildIdCollector, both feature-flagging writers) imports only Queues or MessagePassingBlockingQueue, i.e. the raw jctools factory. So this branch is the entirety of the evidence that #12313's surface is usable.

What it exercises

WorkQueue member	exercised here
tryPut(element)	✔ control signals (STOP / REPORT / CLEAR)
tryPutBatch(source, context, producer)	✔ the publish path
process(limit, consumer)	✔ the aggregator drain loop
close / isClosed / size / dropped	✔
tryPutBatch(Collection), tryPutBatch(T...)	✘
RejectHandler overload	✘
tryReserve / Reservation	✘
RetryStrategy / RetryQueue / MaxRetries, processOrRetry / processOrHandle	✘

The unexercised rows are the honest reason this PR exists.

The commits

1. Run client-side stats through WorkQueue — replaces the hand-rolled jctools inbox.
2. Bind one producer per aggregator instead of one per trace — snapshotProducer becomes a non-capturing bound-once field, so admission allocates nothing.
3. Carry the hoisted peer tag schema as a second context — the schema is read once per trace, not once per span; this is what BiContextualProducer exists for.
4. Drain the inbox in one batched pass again — restores the old jctools drain(size()) shape.
5. Rename the drain barrier to isDrained.
6. Let the inbox hold the stopped state instead of the drainer — Drainer.stopped is deleted; STOP calls inbox.close(), and the run loop and drainer guards read that one flag. Also makes publish(List) ask isClosed() once per trace before statsExportEnabled(), so post-shutdown publishing costs one volatile read instead of a capacity's worth of snapshots built for a consumer that has already exited.
7. Split the publish loop into a decision pass and an admission pass — the one to look at. See below.

On the last commit: it works, and I don't think it's an improvement

publish(List) was one fused loop doing three jobs — filter spans, accumulate counted / forceKeep, admit. tryPutBatch means the queue owns the admission walk, and the queue stops walking when it runs out of room. Two of those jobs need every eligible span regardless of admission, and the ignored-resource case is a break that a producer returning null cannot express. So using tryPutBatch at all forces two passes; there is no lighter version.

What it gets right: ineligible spans are declined (the producer returns null), which is neither an admission nor a drop — so counted - admitted is exactly the eligible spans that did not fit, which is precisely what onStatsInboxFull() wants. A rejected-elements return could not have given that, because a place is claimed before the producer is asked, so a full queue cannot tell a genuine refusal from an element the producer would have declined. That is why #12313's transforming form returns a count.

What it costs: +34/−22 for the same three jobs. Twelve fused lines become thirty unfused ones, plus limit/position index bookkeeping, a subList ternary, a loop to report a metric that used to be one inline call, and a duplicated eligibility test in snapshotIfEligible that has to stay in agreement with the first pass with nothing enforcing it. The fusion was the compression.

Latent, not a regression: inbox.dropped() now over-counts, because an ineligible span past the fill point claims a place and fails before the producer can decline it. Nothing reads dropped() in dd-trace-core today.

So commit 7 is offered as a thing to look at, not a thing to agree with. Reverting it leaves commits 1–6, which stand on their own.

Verification

:dd-trace-core:test --tests 'datadog.trace.common.metrics.*' — 151 tests, 0 failures, no test changed by commit 7. :utils:queue-utils:test — 84 tests. spotlessCheck clean on both modules.