Add WorkQueue<T> with MPSC and linked-queue backings

[dougqh]

What

Adds the Queue<T> API surface from APMLP-1642 to utils/queue-utils (datadog.common.queue). API only — no backing implementation, no call-site migration. Opened to get the shape reviewed before either backing is built.

Seven new types:

Type	Role
Queue<T>	admission, consumption, size/dropped, lifecycle
Producer<T> / ContextualProducer<C,T>	deferred element construction
BatchProducer<T>	lossless incremental admission
RetryStrategy<T>	decides what happens to an item whose consumer threw
RetryQueue<T>	scoped capability to resubmit, only reachable inside onFailure
MaxRetries<T>	ready-made strategy for the common case

Why

The tracer hands work from many app threads to a single background consumer in several places, today via node-per-element linked queues (ConcurrentLinkedQueue/ConcurrentLinkedDeque) that are unbounded or bounded by a hand-rolled guard calling the O(n) size(). Two recurring problems:

• Build-then-drop under backpressure. Producers construct the payload before knowing there's room, so the failure path allocates hardest exactly when the agent is unreachable and the queue is jammed. tryPut reserves the slot before invoking the producer, so a rejected element is never constructed at all.
• No way to stop. APMS-20292: a dead consumer thread left a writer queue silently filling and dropping forever, recovered only by a pod restart. Hence close()/clear()/shutdown().

Design decisions worth reviewing

• Reserve-first, always. Every producer is treated as destructive. This collapses what would have been a tryPut(Producer, onRejected) / tryPut(DestructiveProducer) fork into one method and removes the onRejected callback entirely — it only existed to disambiguate "capacity check failed first" from "produced but rejected," an ambiguity that doesn't exist once reserve always precedes produce. An optimistic non-destructive fast path is deliberately deferred as a benchmark-justified escalation rather than shipped speculatively.
• reserve(int) / Reservation<T> are not public. They're only genuinely needed when caller-owned work must happen between claiming capacity and filling it. Every public admission shape already holds all its elements and can reserve-then-insert atomically and privately, so reserve/fill stays an internal technique rather than a leakable handle.
• Two unrelated booleans. RetryStrategy.onFailure returns retried-vs-gave-up. Queue.process returns "was there an item," which is the drain-loop signal and says nothing about whether the consumer succeeded. Both are documented as such because the collision is easy to misread.
• The API owns no logging. Logging, counting, or ignoring a give-up is composed by the caller inside its own RetryStrategy.
• close() and clear() stay separate, with shutdown() as the atomic combinator. Folding clear() into close() would remove the graceful-drain option — stop new work, let the consumer finish its backlog — which is what a graceful shutdown actually wants. Sequencing them by hand leaves a race window (a producer already past the closed check, an in-flight retry lease) that the combinator closes.

Open questions

1. Queue collides with java.util.Queue. Inside datadog.common.queue the new type shadows it, so any file in that package wanting the JDK type must fully qualify. Kept the ticket's name rather than renaming unilaterally — worth a decision.
2. Overload ambiguity. tryPut(T) vs tryPut(Producer<? extends T>) vs tryPut(Collection<? extends T>) are ambiguous when T is itself a producer or a collection. Tolerable, or should the producer/collection forms get distinct names?
3. RetryQueue.retry(T...) is a generic varargs on an interface method, so @SafeVarargs isn't available and call sites take an unchecked warning. A retry(T) + retry(Collection<T>) pair would avoid it.
4. Landed in utils/queue-utils, not internal-api. That module already owns Queues, already depends on JCTools, and is already shadow-included — internal-api has neither dependency. Say the word if it belongs elsewhere.

Follow-ups

Two backings are planned behind this interface: a bounded array-backed MPSC buffer (JCTools, adaptive per-site capacity, O(1) size(), reusable drain) and a ConcurrentLinkedQueue-backed one for call sites that can't take a bound yet. Then the use cases: client-side stats, PendingTrace (APMLP-1655), OkHttpSink.sendAsync (APMLP-1652), health metrics (APMLP-1565), DependencyResolverQueue (APMLP-1654).

🤖 Generated with Claude Code

[datadog-prod-us1-5]

🎯 Code Coverage (details)
• Patch Coverage: 79.26%
• Overall Coverage: 58.79% (+0.01%)

_{This comment will be updated automatically if new data arrives.
🔗 Commit SHA: e3594a5 | Docs | View more details | Give us feedback!}

[dougqh]

Update: two backings added

Both are package-private and reachable only through Queues factories — mpscQueue(capacity), mpmcQueue(capacity), unboundedMpmcQueue(). Shared admission/lifecycle/retry logic is in BaseQueue. 28 tests, all passing.

Reserve-first is real, not emulated

Worth recording, because it decides whether the central guarantee is achievable by wrapping JCTools rather than forking it. I decompiled fill(Supplier, int) in 4.0.6:

• MpscArrayQueue (Unsafe, Java 8 path) — casProducerIndex at bytecode 141, Supplier.get() at 180.
• MpscVarHandleArrayQueue (Java 25+ path, what Queues selects) — casProducerIndex at 128, Supplier.get() at 167.

Both CAS-claim the slot first, and both return early without calling the supplier at all when there's no capacity. So fill(supplier, 1) == 1 is exact reserve-first admission, on both paths. The stress test asserts this directly: 8 threads × 20k producing admissions against a permanently full queue invoke the producer zero times.

A consequence the ticket didn't note

The claim is published (producer index advanced) before the element is stored. A consumer that reaches that slot waits for the element to appear. So reserve-first converts "producer allocates" into "producer allocates while holding a slot the consumer may be blocked on" — a slow producer now stalls the consumer, where before it only stalled itself. Fine for SpanSnapshot; not fine for anything that blocks or does I/O in produce(). I documented this on Queue, but it's a contract the use-case cards should be checked against — OkHttpSink (APMLP-1652) builds a Request including a full buffer copy inside what would become the producer, which is the largest candidate.

Two API defects found by the compiler

1. RetryStrategy<? super T> doesn't typecheck. A strategy over a supertype S would receive a RetryQueue<S>, and the queue can't satisfy that — it only accepts T. Changed process to take an invariant RetryStrategy<T>. The variance in the ticket's converged signature isn't merely too permissive, it's uninhabitable.
2. The overload ambiguity is real, not theoretical. process(consumer, null) doesn't compile — ambiguous between process(Consumer, RetryStrategy) and process(C, BiConsumer). Currently worked around with a cast at the internal call site, but every caller passing a literal null hits it too. This strengthens the case for renaming the context-taking forms.

Deviations from the ticket, deliberate

• The retry "lease" is not honoured. The ticket says a single-item retry "cannot fail on capacity" because the consumer still owns the slot. poll() releases the slot before the consumer runs, so there is no slot left to reuse — a retry is an ordinary re-admission and can be rejected if producers refilled the queue meanwhile (counted as a drop). Honouring the ticket would need a claim-based consumer, i.e. forking the backing rather than wrapping it, or reserving a hidden retry slot. Flagging rather than quietly weakening it.
• Batch admission is element-wise, not one atomic reservation. tryPutBatch / tryPut(Collection) / RetryQueue.retry(T...) can therefore partially admit. fill(supplier, n) would make this atomic for the MPSC backing; left out of the draft.
• shutdown() is not atomic — it sets closed, then discards. A producer already past the closed check can still land an element. Narrowing the window needs backing-level support.
• put(BatchProducer) is lossless by construction, which I think is the intended reading: an element is pulled only once a slot is claimed, so whatever doesn't fit is still held by the producer. It stops early rather than blocking, and the caller checks hasNext(). No drops are counted for stopping at capacity.

On LinkedQueue

It keeps the per-element node, so it doesn't deliver the allocation win — it exists so multi-consumer or as-yet-unbounded call sites can adopt the interface first and be re-backed later. Its size counter is what makes the bound enforceable and size() O(1), which is precisely what DependencyResolverQueue (APMLP-1654) wants in place of a hand-rolled cap plus an O(n) ConcurrentLinkedQueue.size() walk on every admission.

Still open from the original description

The static-routine bypass for hot call sites — and with it the interface-vs-final-class dispatch question — is still not addressed. With two backings live behind Queue<T>, admission call sites that see both are now genuinely bimorphic.

[dd-octo-sts]

🟢 Java Benchmark SLOs — All performance SLOs passed

Suite	Status
Startup	🟢 pass

SLO thresholds are defined here based on automatically generated metrics. A warning is raised when results are within 5% of the threshold.

PR vs. master results

Scenario	Candidate	master	Δ (95% CI of mean)
startup:insecure-bank:iast:Agent	14.05 s	14.04 s	[-0.6%; +0.8%] (no difference)
startup:insecure-bank:tracing:Agent	12.94 s	12.92 s	[-0.7%; +1.0%] (no difference)
startup:petclinic:appsec:Agent	17.34 s	16.60 s	[+0.3%; +8.6%] (maybe worse)
startup:petclinic:iast:Agent	16.81 s	17.47 s	[-8.0%; +0.5%] (no difference)
startup:petclinic:profiling:Agent	17.34 s	17.29 s	[-1.1%; +1.7%] (no difference)
startup:petclinic:sca:Agent	17.39 s	16.66 s	[+0.1%; +8.7%] (maybe worse)
startup:petclinic:tracing:Agent	16.13 s	16.59 s	[-6.9%; +1.2%] (no difference)

Commit: 6fdbe965 · CI Pipeline · Benchmarking Platform UI

---

Load and DaCapo benchmarks can be triggered manually in the GitLab pipeline. Results will appear in the Benchmarking Platform UI after completion.

[dougqh]

Renamed: Queue<T> → WorkQueue<T>, factories split out

Set apart from the raw JCTools factory, and the java.util.Queue collision (open question 1) is gone — no caller has to qualify an import now.

Queue            -> WorkQueue          (+ WorkQueues factory)
BaseQueue        -> BaseWorkQueue
MpscBoundedQueue -> MpscWorkQueue
LinkedQueue      -> LinkedWorkQueue

The two factory surfaces are now separate classes, because they answer different questions for the caller:

	returns	caller's job
WorkQueues.createMpscQueue(n)	WorkQueue<T>	hand work over; backing is hidden and re-backable
Queues.mpscArrayQueue(n)	MessagePassingQueue<T>	drive the raw queue yourself

Queues keeps only the raw factories and is otherwise untouched, so its eleven existing callers — which include OkHttpSink and ClientStatsAggregator, two of the ticket's own use cases — are unaffected until they migrate deliberately.

Usage now reads:

WorkQueue<SpanSnapshot> inbox = WorkQueues.createMpscQueue(1024);
inbox.tryPut(ctx, SNAPSHOT);
inbox.process(this::publish, new MaxRetries<>(3));

28 tests still green.

Open questions, updated

1. Queue collides with java.util.Queue — resolved by the rename.
2. Overload ambiguity still stands, and is still real rather than theoretical: process(consumer, null) does not compile. Worth deciding whether the context-taking forms get distinct names.
3. RetryQueue.retry(T...) unchecked warning — unchanged. (Note it kept its name: it is the retry capability, not a WorkQueue.)
4. Module placement — settled: utils/queue-utils, alongside Queues.
5. Static-routine bypass / dispatch — still open, and now the sharper question of the two, since two backings sit behind WorkQueue<T>.

[dougqh]

Dropped BatchProducer and put() — deferred to the SCA follow-on

Removed from this PR. The reasoning, so the follow-on doesn't relitigate it:

Why it goes now. Across all five use cases on APMLP-1642 — client-side stats, PendingTrace, OkHttpSink, health metrics, DependencyResolverQueue — every admission is a single element. Batch admission had no caller, so it was a public type, a method, and a lossless-admission contract that every future backing would have to honour, carried on spec.

Why it comes back with SCA. SCA's partition-on-failure (the #11977 Reachability case) is the genuine caller. Landing it there means its real access pattern picks the shape, instead of us guessing between pull, push, and reservation.

The name was wrong too, which is what started this. BatchProducer sat in a family with Producer.produce() and ContextualProducer.produce(ctx) but had hasNext()/next() — so it read as "produces batches" while actually being pulled one element at a time, advertising the opposite of the mechanism. It was also structurally java.util.Iterator<T> with no remove(), which has had a throwing default since Java 8.

Agreed shape for when it returns

Not a callback taking the WorkQueue itself. Handing out the full interface exposes close(), shutdown(), clear() and process() to arbitrary caller code, and it lets the filler stash the reference past the call. It also inverts loop ownership: the filler learns about capacity only from tryPut returning false, by which point the eager tryPut(T) form has already built the element — reintroducing build-then-drop inside the one method meant to prevent it.

Instead, a scoped admission-only capability in the manner of RetryQueue — obtainable only inside the call, dead on return:

interface Admission<T> {                 // no lifecycle, no consumption
  boolean tryPut(T element);
  <C> boolean tryPut(C context, ContextualProducer<? super C, ? extends T> producer);
}

Note this is close to the reserve(int) / Reservation<T> that the ticket deliberately kept private — "only genuinely needed when caller-owned work must happen between claiming capacity and filling it." SCA bisection may be exactly that case, in which case the follow-on is really the argument for making a bounded, scoped form of reservation public. Worth deciding there with the use case in hand.

24 tests, all passing. tryPutBatch and tryPut(Collection) are still present and also have no caller today; happy to trim those too, though they cost no new type.

[dougqh]

To Claude - I think I prefer the present tense Retry to Retried

[dougqh]

For Claude - is this a location where batch claiming could work?
There would need to a limit on batch size, so it doesn't created starvation. But at least here, there's some safety since the rejects are returned to be used again.

I suppose the alternative is to registered the elements in a simple Generator, but we don't have Generator support yet.

[dougqh]

Should we estimate the reject count based on availability cap and number of elements that we're trying to submit?

[dougqh]

Same question about estimating reject count?

[dougqh]

For Claude, I tend to want to make methods final for both code cleanest and performance reasons. Is that possible here?

[dougqh]

For Claude - We should probably call tryPutBatch, too

[dougqh]

I think we should probably also have...
boolean process(Consumer, ExceptionHandler)

interface ExceptionHandler {
void handle(Throwable t);
}

I mostly just want that for the case where someone wants to log, etc without retrying.