From 2df0fa3d7920a7047206b3bf61970d4896ffd53d Mon Sep 17 00:00:00 2001 From: Charles Cote Date: Fri, 30 Oct 2026 21:02:51 +0100 Subject: [PATCH 1/7] Demonstrate long-running trace ownership loss --- .../core/LongRunningTracesTrackerTest.java | 15 +++ formal-methods/long-running-traces/README.md | 42 ++++++++ .../long-running-traces/TrackerOwnership.cfg | 9 ++ .../long-running-traces/TrackerOwnership.tla | 97 +++++++++++++++++++ 4 files changed, 163 insertions(+) create mode 100644 formal-methods/long-running-traces/README.md create mode 100644 formal-methods/long-running-traces/TrackerOwnership.cfg create mode 100644 formal-methods/long-running-traces/TrackerOwnership.tla diff --git a/dd-trace-core/src/test/java/datadog/trace/core/LongRunningTracesTrackerTest.java b/dd-trace-core/src/test/java/datadog/trace/core/LongRunningTracesTrackerTest.java index 27c8156e27f..7a69dbd65cc 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/LongRunningTracesTrackerTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/LongRunningTracesTrackerTest.java @@ -1,6 +1,7 @@ package datadog.trace.core; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.clearInvocations; @@ -128,6 +129,20 @@ void maxTrackedTracesIsEnforced() { assertEquals(1, tracker.getDropped()); } + @Test + void fullTrackerLeavesOverflowTraceForBufferFallback() { + for (int i = 0; i < MAX_TRACKED_TRACES; i++) { + tracker.add(newTraceToTrack()); + } + + PendingTrace overflowTrace = newTraceToTrack(); + + assertFalse( + tracker.add(overflowTrace), + "a trace rejected at capacity must remain available to the buffer's fallback path"); + assertEquals(LongRunningTracesTracker.NOT_TRACKED, overflowTrace.getLongRunningTrackedState()); + } + @Test void expiredTraces() { PendingTrace trace = newTraceToTrack(); diff --git a/formal-methods/long-running-traces/README.md b/formal-methods/long-running-traces/README.md new file mode 100644 index 00000000000..8256cfa316b --- /dev/null +++ b/formal-methods/long-running-traces/README.md @@ -0,0 +1,42 @@ +# Long-running trace ownership model + +`TrackerOwnership.tla` models the ownership hand-off between +`PendingTraceBuffer.Worker` and `LongRunningTracesTracker`. + +The safety invariant is: + +> Every live pending trace is owned by the queue, the worker, the tracker, or +> the buffer fallback path. + +The current implementation violates this invariant when the tracker is at +capacity: + +1. the worker removes the trace from the queue; +2. `LongRunningTracesTracker.add` changes its state from `TO_TRACK` to + `TRACKED`; +3. `addTrace` observes a full tracker and does not retain the trace; +4. `add` still returns `true`, so the worker skips the fallback path and + relinquishes the trace. + +Run the model from the repository root: + +```shell +mkdir -p /tmp/dd-trace-java-tlc/tracker-ownership +java -XX:+UseParallelGC -jar /tmp/tla2tools-1.7.4.jar \ + -config formal-methods/long-running-traces/TrackerOwnership.cfg \ + -metadir /tmp/dd-trace-java-tlc/tracker-ownership \ + formal-methods/long-running-traces/TrackerOwnership.tla +``` + +TLC is expected to report `Invariant LiveTraceHasOwner is violated` with a +four-state counterexample. + +The corresponding Java regression is +`LongRunningTracesTrackerTest.fullTrackerLeavesOverflowTraceForBufferFallback`. +On the unfixed implementation it fails because `add` returns `true` and leaves +the rejected trace in `TRACKED` state. + +The practical impact is that, under long-running tracker saturation, finished +spans belonging to a trace with an unfinished span or continuation can remain +unreported until that outstanding work finishes. If it never finishes, the +trace no longer has any component capable of writing it. diff --git a/formal-methods/long-running-traces/TrackerOwnership.cfg b/formal-methods/long-running-traces/TrackerOwnership.cfg new file mode 100644 index 00000000000..2c8a426aef7 --- /dev/null +++ b/formal-methods/long-running-traces/TrackerOwnership.cfg @@ -0,0 +1,9 @@ +SPECIFICATION Spec + +CONSTANT Capacity = 1 + +INVARIANTS + TypeOK + LiveTraceHasOwner + +CHECK_DEADLOCK FALSE diff --git a/formal-methods/long-running-traces/TrackerOwnership.tla b/formal-methods/long-running-traces/TrackerOwnership.tla new file mode 100644 index 00000000000..f22dd385b7c --- /dev/null +++ b/formal-methods/long-running-traces/TrackerOwnership.tla @@ -0,0 +1,97 @@ +-------------------------- MODULE TrackerOwnership -------------------------- +EXTENDS Naturals + +(** + * Ownership model for the hand-off from PendingTraceBuffer.Worker to + * LongRunningTracesTracker. A live trace must always have an owner that can + * eventually write it: the queue, worker, tracker, or fallback buffer path. + *) + +CONSTANT Capacity + +ASSUME Capacity \in Nat \ {0} + +VARIABLES + queueOwns, + workerOwns, + trackerOwns, + fallbackOwns, + trackerSize, + traceState, + addResult, + pc + +vars == + <> + +Init == + /\ queueOwns = TRUE + /\ workerOwns = FALSE + /\ trackerOwns = FALSE + /\ fallbackOwns = FALSE + /\ trackerSize = Capacity + /\ traceState = "TO_TRACK" + /\ addResult = FALSE + /\ pc = "Queued" + +Dequeue == + /\ pc = "Queued" + /\ queueOwns' = FALSE + /\ workerOwns' = TRUE + /\ pc' = "Adding" + /\ UNCHANGED + <> + +(** + * Current LongRunningTracesTracker.add behavior: the state transition succeeds + * and add returns true even when addTrace drops the trace at capacity. + *) +TrackerAdd == + /\ pc = "Adding" + /\ traceState = "TO_TRACK" + /\ traceState' = "TRACKED" + /\ addResult' = TRUE + /\ IF trackerSize = Capacity + THEN /\ UNCHANGED <> + ELSE /\ trackerOwns' = TRUE + /\ trackerSize' = trackerSize + 1 + /\ pc' = "AfterAdd" + /\ UNCHANGED <> + +TrackerAccepted == + /\ pc = "AfterAdd" + /\ addResult + /\ workerOwns' = FALSE + /\ pc' = "Done" + /\ UNCHANGED + <> + +TrackerRejected == + /\ pc = "AfterAdd" + /\ ~addResult + /\ workerOwns' = FALSE + /\ fallbackOwns' = TRUE + /\ pc' = "Done" + /\ UNCHANGED + <> + +Next == Dequeue \/ TrackerAdd \/ TrackerAccepted \/ TrackerRejected + +Spec == Init /\ [][Next]_vars + +TypeOK == + /\ queueOwns \in BOOLEAN + /\ workerOwns \in BOOLEAN + /\ trackerOwns \in BOOLEAN + /\ fallbackOwns \in BOOLEAN + /\ trackerSize \in Nat + /\ traceState \in {"TO_TRACK", "TRACKED", "NOT_TRACKED"} + /\ addResult \in BOOLEAN + /\ pc \in {"Queued", "Adding", "AfterAdd", "Done"} + +LiveTraceHasOwner == + queueOwns \/ workerOwns \/ trackerOwns \/ fallbackOwns + +============================================================================= From f0c63233d8108528be1d08dcf94a51e9b7b11a85 Mon Sep 17 00:00:00 2001 From: Charles Cote Date: Fri, 30 Oct 2026 21:30:34 +0100 Subject: [PATCH 2/7] Demonstrate additional tracer lifecycle failures --- .../writer/TraceProcessingWorkerTest.java | 60 +++++++ .../trace/core/PendingTraceBufferTest.java | 165 ++++++++++++++++++ .../flush-dump-marker/FlushDumpMarker.cfg | 7 + .../flush-dump-marker/FlushDumpMarker.tla | 66 +++++++ formal-methods/flush-dump-marker/README.md | 30 ++++ .../FlushDrainOwnership.cfg | 7 + .../FlushDrainOwnership.tla | 58 ++++++ formal-methods/pending-trace-flush/README.md | 35 ++++ formal-methods/span-sampling-flush/README.md | 30 ++++ .../span-sampling-flush/SpanSamplingFlush.cfg | 7 + .../span-sampling-flush/SpanSamplingFlush.tla | 79 +++++++++ 11 files changed, 544 insertions(+) create mode 100644 formal-methods/flush-dump-marker/FlushDumpMarker.cfg create mode 100644 formal-methods/flush-dump-marker/FlushDumpMarker.tla create mode 100644 formal-methods/flush-dump-marker/README.md create mode 100644 formal-methods/pending-trace-flush/FlushDrainOwnership.cfg create mode 100644 formal-methods/pending-trace-flush/FlushDrainOwnership.tla create mode 100644 formal-methods/pending-trace-flush/README.md create mode 100644 formal-methods/span-sampling-flush/README.md create mode 100644 formal-methods/span-sampling-flush/SpanSamplingFlush.cfg create mode 100644 formal-methods/span-sampling-flush/SpanSamplingFlush.tla diff --git a/dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerTest.java b/dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerTest.java index 86dbfa61b22..3c19adca4b8 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerTest.java @@ -11,6 +11,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import datadog.trace.api.sampling.PrioritySampling; import datadog.trace.bootstrap.instrumentation.api.SpanPostProcessor; import datadog.trace.common.sampling.SingleSpanSampler; import datadog.trace.common.writer.ddagent.PrioritizationStrategy.PublishResult; @@ -25,6 +26,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -148,6 +150,64 @@ void testAFlushShouldClearThePrimaryQueue() { } } + @Test + void flushWaitsForInFlightSingleSpanSampling() throws Exception { + CountDownLatch samplingStarted = new CountDownLatch(1); + CountDownLatch allowSampling = new CountDownLatch(1); + SingleSpanSampler blockingSampler = + new SingleSpanSampler() { + @Override + public > boolean setSamplingPriority(T span) { + samplingStarted.countDown(); + try { + allowSampling.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return true; + } + }; + AtomicInteger serializedTraces = new AtomicInteger(); + PayloadDispatcher dispatcher = mock(PayloadDispatcher.class); + doAnswer( + invocation -> { + serializedTraces.incrementAndGet(); + return null; + }) + .when(dispatcher) + .addTrace(any()); + + try (TraceProcessingWorker worker = + new TraceProcessingWorker( + 10, + mock(HealthMetrics.class), + dispatcher, + () -> false, + FAST_LANE, + 100, + TimeUnit.SECONDS, + blockingSampler)) { + try { + worker.start(); + DDSpan span = mock(DDSpan.class); + when(span.samplingPriority()).thenReturn((int) PrioritySampling.USER_DROP); + + assertEquals( + PublishResult.ENQUEUED_FOR_SINGLE_SPAN_SAMPLING, + worker.publish(span, PrioritySampling.USER_DROP, Collections.singletonList(span))); + assertTrue(samplingStarted.await(3, TimeUnit.SECONDS)); + + assertTrue(worker.flush(3, TimeUnit.SECONDS)); + assertEquals( + 1, + serializedTraces.get(), + "flush returned before an accepted single-span-sampling candidate was serialized"); + } finally { + allowSampling.countDown(); + } + } + } + // ------------------------------------------------------------------------- // Test 4: should report failure if serialization fails // ------------------------------------------------------------------------- diff --git a/dd-trace-core/src/test/java/datadog/trace/core/PendingTraceBufferTest.java b/dd-trace-core/src/test/java/datadog/trace/core/PendingTraceBufferTest.java index 926262c1b35..8e529453278 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/PendingTraceBufferTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/PendingTraceBufferTest.java @@ -45,6 +45,9 @@ import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.zip.ZipEntry; @@ -379,6 +382,168 @@ public boolean writeOnBufferFull() { assertEquals(3, counter.get()); } + @Test + void flushDrainClearsEnqueuedStateBeforeWriting() throws Exception { + CountDownLatch workerBlocked = new CountDownLatch(1); + CountDownLatch allowWorker = new CountDownLatch(1); + CountDownLatch traceWrites = new CountDownLatch(2); + doAnswer( + invocation -> { + traceWrites.countDown(); + return null; + }) + .when(tracer) + .write(any()); + PendingTraceBuffer.Element blocker = + new PendingTraceBuffer.Element() { + @Override + public long oldestFinishedTime() { + workerBlocked.countDown(); + try { + allowWorker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return 0; + } + + @Override + public boolean lastReferencedNanosAgo(long nanos) { + return false; + } + + @Override + public void write() {} + + @Override + public DDSpan getRootSpan() { + return null; + } + + @Override + public boolean setEnqueued(boolean enqueued) { + return true; + } + + @Override + public boolean writeOnBufferFull() { + return true; + } + }; + + buffer.start(); + bufferSpy.enqueue(blocker); + assertTrue(workerBlocked.await(3, TimeUnit.SECONDS)); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future flush = executor.submit(buffer::flush); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3); + while (delayingBuffer.getQueue().isEmpty() && System.nanoTime() < deadline) { + Thread.yield(); + } + assertEquals(1, delayingBuffer.getQueue().size(), "flush marker was not queued"); + + PendingTrace trace = factory.create(DDTraceId.ONE); + DDSpan parent = newSpanOf(trace); + DDSpan child = newSpanOf(parent); + parent.finish(); + + assertEquals(1, trace.getIsEnqueued()); + allowWorker.countDown(); + flush.get(3, TimeUnit.SECONDS); + + assertTrue(trace.isRootSpanWritten()); + assertEquals( + 0, + trace.getIsEnqueued(), + "a trace written by the flush drain must be eligible for re-enqueue"); + + child.finish(); + assertTrue(traceWrites.await(3, TimeUnit.SECONDS)); + assertEquals(0, trace.size()); + verify(tracer, times(2)).write(any()); + } finally { + allowWorker.countDown(); + executor.shutdownNow(); + } + } + + @Test + void flushDrainDoesNotConsumeTracerDumpMarker() throws Exception { + CountDownLatch workerBlocked = new CountDownLatch(1); + CountDownLatch allowWorker = new CountDownLatch(1); + PendingTraceBuffer.Element blocker = + new PendingTraceBuffer.Element() { + @Override + public long oldestFinishedTime() { + workerBlocked.countDown(); + try { + allowWorker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return 0; + } + + @Override + public boolean lastReferencedNanosAgo(long nanos) { + return false; + } + + @Override + public void write() {} + + @Override + public DDSpan getRootSpan() { + return null; + } + + @Override + public boolean setEnqueued(boolean enqueued) { + return true; + } + + @Override + public boolean writeOnBufferFull() { + return true; + } + }; + + buffer.start(); + bufferSpy.enqueue(blocker); + assertTrue(workerBlocked.await(3, TimeUnit.SECONDS)); + + ExecutorService executor = Executors.newFixedThreadPool(2); + Future flare = null; + try { + Future flush = executor.submit(buffer::flush); + awaitQueueSize(1); + + flare = executor.submit(TracerFlare::prepareForFlare); + awaitQueueSize(2); + + allowWorker.countDown(); + flush.get(3, TimeUnit.SECONDS); + flare.get(3, TimeUnit.SECONDS); + } finally { + allowWorker.countDown(); + buffer.close(); + if (flare != null) { + flare.get(3, TimeUnit.SECONDS); + } + executor.shutdownNow(); + } + } + + private void awaitQueueSize(int expectedSize) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3); + while (delayingBuffer.getQueue().size() != expectedSize && System.nanoTime() < deadline) { + Thread.yield(); + } + assertEquals(expectedSize, delayingBuffer.getQueue().size()); + } + @Test void samePendingTraceIsNotEnqueuedMultipleTimes() { when(tracer.getPartialFlushMinSpans()).thenReturn(10000); diff --git a/formal-methods/flush-dump-marker/FlushDumpMarker.cfg b/formal-methods/flush-dump-marker/FlushDumpMarker.cfg new file mode 100644 index 00000000000..6d49d947f00 --- /dev/null +++ b/formal-methods/flush-dump-marker/FlushDumpMarker.cfg @@ -0,0 +1,7 @@ +SPECIFICATION Spec + +INVARIANTS + TypeOK + WaitingDumpHasOwner + +CHECK_DEADLOCK FALSE diff --git a/formal-methods/flush-dump-marker/FlushDumpMarker.tla b/formal-methods/flush-dump-marker/FlushDumpMarker.tla new file mode 100644 index 00000000000..e32c23a0b02 --- /dev/null +++ b/formal-methods/flush-dump-marker/FlushDumpMarker.tla @@ -0,0 +1,66 @@ +--------------------------- MODULE FlushDumpMarker --------------------------- + +(** + * Models a tracer-flare dump command queued behind a flush command while the + * PendingTraceBuffer worker is busy. Processing the flush drains the dump + * command through CommandElement.write(), which is a no-op, so the waiter + * loses both the queued command and its acknowledgement. + *) + +VARIABLES + flushQueued, + dumpQueued, + flushAcknowledged, + dumpAcknowledged, + dumpWaiting, + pc + +vars == + <> + +Init == + /\ flushQueued = FALSE + /\ dumpQueued = FALSE + /\ flushAcknowledged = FALSE + /\ dumpAcknowledged = FALSE + /\ dumpWaiting = FALSE + /\ pc = "WorkerBlocked" + +QueueFlush == + /\ pc = "WorkerBlocked" + /\ flushQueued' = TRUE + /\ pc' = "FlushQueued" + /\ UNCHANGED <> + +QueueDump == + /\ pc = "FlushQueued" + /\ dumpQueued' = TRUE + /\ dumpWaiting' = TRUE + /\ pc' = "DumpQueued" + /\ UNCHANGED <> + +(** Current behavior: WriteDrain consumes the dump marker without acknowledging it. *) +FlushDrainsDump == + /\ pc = "DumpQueued" + /\ flushQueued' = FALSE + /\ dumpQueued' = FALSE + /\ flushAcknowledged' = TRUE + /\ pc' = "DumpMarkerLost" + /\ UNCHANGED <> + +Next == QueueFlush \/ QueueDump \/ FlushDrainsDump + +Spec == Init /\ [][Next]_vars + +TypeOK == + /\ flushQueued \in BOOLEAN + /\ dumpQueued \in BOOLEAN + /\ flushAcknowledged \in BOOLEAN + /\ dumpAcknowledged \in BOOLEAN + /\ dumpWaiting \in BOOLEAN + /\ pc \in {"WorkerBlocked", "FlushQueued", "DumpQueued", "DumpMarkerLost"} + +WaitingDumpHasOwner == + dumpWaiting => (dumpQueued \/ dumpAcknowledged) + +============================================================================= diff --git a/formal-methods/flush-dump-marker/README.md b/formal-methods/flush-dump-marker/README.md new file mode 100644 index 00000000000..6225dea08c1 --- /dev/null +++ b/formal-methods/flush-dump-marker/README.md @@ -0,0 +1,30 @@ +# Flush/dump command ownership model + +This model checks the ownership of a tracer-flare dump command when it is +queued behind a `PendingTraceBuffer` flush command. + +Both commands share the pending-trace queue. When the worker processes the +flush marker, `WriteDrain` drains every following element by calling +`Element.write()`. A dump marker is a `CommandElement`, whose `write()` method +is a no-op. The marker therefore disappears without incrementing +`dumpCounter`, leaving `TracerDump.prepareForFlare()` waiting indefinitely. + +Run from the repository root: + +```shell +mkdir -p /tmp/dd-trace-java-tlc/flush-dump-marker +java -XX:+UseParallelGC -jar /tmp/tla2tools-1.7.4.jar \ + -config formal-methods/flush-dump-marker/FlushDumpMarker.cfg \ + -metadir /tmp/dd-trace-java-tlc/flush-dump-marker \ + formal-methods/flush-dump-marker/FlushDumpMarker.tla +``` + +The current implementation produces a four-state `WaitingDumpHasOwner` +invariant violation. + +The corresponding deterministic Java regression is +`PendingTraceBufferTest.flushDrainDoesNotConsumeTracerDumpMarker`. It blocks +the worker, queues a flush and then a flare dump, and releases the worker. The +flush completes, but flare preparation remains stuck because its marker was +silently consumed. On the unfixed implementation it fails with a +`TimeoutException` while waiting for flare preparation to complete. diff --git a/formal-methods/pending-trace-flush/FlushDrainOwnership.cfg b/formal-methods/pending-trace-flush/FlushDrainOwnership.cfg new file mode 100644 index 00000000000..1a9de9301b9 --- /dev/null +++ b/formal-methods/pending-trace-flush/FlushDrainOwnership.cfg @@ -0,0 +1,7 @@ +SPECIFICATION Spec + +INVARIANTS + TypeOK + PendingChildHasWriter + +CHECK_DEADLOCK FALSE diff --git a/formal-methods/pending-trace-flush/FlushDrainOwnership.tla b/formal-methods/pending-trace-flush/FlushDrainOwnership.tla new file mode 100644 index 00000000000..f26934823b6 --- /dev/null +++ b/formal-methods/pending-trace-flush/FlushDrainOwnership.tla @@ -0,0 +1,58 @@ +------------------------- MODULE FlushDrainOwnership ------------------------- +EXTENDS Naturals + +(** + * Models a PendingTrace queued behind FLUSH_ELEMENT while the worker is busy. + * The flush drain writes the trace but, unlike normal dequeue, does not clear + * its isEnqueued flag. A later child finish then cannot requeue the trace. + *) + +VARIABLES + queueOwns, + enqueued, + rootWritten, + lateChildPending, + pc + +vars == <> + +Init == + /\ queueOwns = TRUE + /\ enqueued = TRUE + /\ rootWritten = FALSE + /\ lateChildPending = FALSE + /\ pc = "QueuedBehindFlush" + +(** Current WriteDrain.accept behavior: write without setEnqueued(false). *) +FlushDrainWrites == + /\ pc = "QueuedBehindFlush" + /\ queueOwns' = FALSE + /\ rootWritten' = TRUE + /\ pc' = "Flushed" + /\ UNCHANGED <> + +LateChildFinishes == + /\ pc = "Flushed" + /\ lateChildPending' = TRUE + /\ IF enqueued + THEN /\ UNCHANGED <> + ELSE /\ queueOwns' = TRUE + /\ enqueued' = TRUE + /\ pc' = "Done" + /\ UNCHANGED rootWritten + +Next == FlushDrainWrites \/ LateChildFinishes + +Spec == Init /\ [][Next]_vars + +TypeOK == + /\ queueOwns \in BOOLEAN + /\ enqueued \in BOOLEAN + /\ rootWritten \in BOOLEAN + /\ lateChildPending \in BOOLEAN + /\ pc \in {"QueuedBehindFlush", "Flushed", "Done"} + +PendingChildHasWriter == + lateChildPending => queueOwns + +============================================================================= diff --git a/formal-methods/pending-trace-flush/README.md b/formal-methods/pending-trace-flush/README.md new file mode 100644 index 00000000000..24464ece5c8 --- /dev/null +++ b/formal-methods/pending-trace-flush/README.md @@ -0,0 +1,35 @@ +# Pending trace flush-drain ownership model + +This model checks the queue ownership flag when `PendingTraceBuffer` bulk-drains +elements after a flush marker. + +The normal worker path clears `isEnqueued` before writing a trace. The bulk +`WriteDrain` path writes without clearing it. If a child span finishes after +that flush, the trace believes it is already queued and cannot be re-enqueued. + +Run from the repository root: + +```shell +mkdir -p /tmp/dd-trace-java-tlc/flush-drain-ownership +java -XX:+UseParallelGC -jar /tmp/tla2tools-1.7.4.jar \ + -config formal-methods/pending-trace-flush/FlushDrainOwnership.cfg \ + -metadir /tmp/dd-trace-java-tlc/flush-drain-ownership \ + formal-methods/pending-trace-flush/FlushDrainOwnership.tla +``` + +The current implementation produces a three-state `PendingChildHasWriter` +invariant violation. + +The corresponding deterministic Java regression is +`PendingTraceBufferTest.flushDrainClearsEnqueuedStateBeforeWriting`. It blocks +the worker, puts a flush marker ahead of a pending trace, and then releases the +worker so `WriteDrain` handles the trace. On the unfixed implementation it +fails with: + +```text +a trace written by the flush drain must be eligible for re-enqueue +expected: <0> but was: <1> +``` + +The stale flag means a child that finishes after `CoreTracer.flush()` cannot +place its trace back into the pending buffer, so the child span is never sent. diff --git a/formal-methods/span-sampling-flush/README.md b/formal-methods/span-sampling-flush/README.md new file mode 100644 index 00000000000..8a3d76f2689 --- /dev/null +++ b/formal-methods/span-sampling-flush/README.md @@ -0,0 +1,30 @@ +# Single-span-sampling flush model + +This model checks whether `TraceProcessingWorker.flush()` covers traces that +were already accepted into its upstream single-span-sampling queue. + +The sampling worker moves sampled spans into the primary serialization queue. +However, `flush()` inserts its barrier directly into that primary queue. The +serializer can therefore acknowledge the flush while an older accepted trace +is still being sampled. `RemoteWriter.close()` trusts that acknowledgement and +then interrupts the sampling worker, so shutdown can lose sampled spans. + +Run from the repository root: + +```shell +mkdir -p /tmp/dd-trace-java-tlc/span-sampling-flush +java -XX:+UseParallelGC -jar /tmp/tla2tools-1.7.4.jar \ + -config formal-methods/span-sampling-flush/SpanSamplingFlush.cfg \ + -metadir /tmp/dd-trace-java-tlc/span-sampling-flush \ + formal-methods/span-sampling-flush/SpanSamplingFlush.tla +``` + +The current implementation violates `FlushCoversAcceptedCandidate` after the +serializer acknowledges a flush while the accepted trace remains in the +sampling stage. + +The corresponding deterministic Java regression is +`TraceProcessingWorkerTest.flushWaitsForInFlightSingleSpanSampling`. It blocks +the sampler after the trace has been accepted, calls `flush()`, and observes +that the call returns successfully before the trace reaches the payload +dispatcher. diff --git a/formal-methods/span-sampling-flush/SpanSamplingFlush.cfg b/formal-methods/span-sampling-flush/SpanSamplingFlush.cfg new file mode 100644 index 00000000000..e6162fb02f9 --- /dev/null +++ b/formal-methods/span-sampling-flush/SpanSamplingFlush.cfg @@ -0,0 +1,7 @@ +SPECIFICATION Spec + +INVARIANTS + TypeOK + FlushCoversAcceptedCandidate + +CHECK_DEADLOCK FALSE diff --git a/formal-methods/span-sampling-flush/SpanSamplingFlush.tla b/formal-methods/span-sampling-flush/SpanSamplingFlush.tla new file mode 100644 index 00000000000..03b02aafe65 --- /dev/null +++ b/formal-methods/span-sampling-flush/SpanSamplingFlush.tla @@ -0,0 +1,79 @@ +-------------------------- MODULE SpanSamplingFlush -------------------------- + +(** + * Models an accepted trace moving through the single-span-sampling worker + * before it reaches the serializer. TraceProcessingWorker.flush places its + * barrier directly in the serializer queue, so that barrier can complete + * while an older accepted trace is still in the upstream sampling stage. + *) + +VARIABLES + samplingQueued, + samplingInFlight, + serializerQueued, + serialized, + flushRequested, + flushAcknowledged, + pc + +vars == + <> + +Init == + /\ samplingQueued = FALSE + /\ samplingInFlight = FALSE + /\ serializerQueued = FALSE + /\ serialized = FALSE + /\ flushRequested = FALSE + /\ flushAcknowledged = FALSE + /\ pc = "Ready" + +AcceptCandidate == + /\ pc = "Ready" + /\ samplingQueued' = TRUE + /\ pc' = "SamplingQueued" + /\ UNCHANGED <> + +StartSampling == + /\ pc = "SamplingQueued" + /\ samplingQueued' = FALSE + /\ samplingInFlight' = TRUE + /\ pc' = "SamplingInFlight" + /\ UNCHANGED <> + +RequestFlush == + /\ pc = "SamplingInFlight" + /\ flushRequested' = TRUE + /\ pc' = "FlushRequested" + /\ UNCHANGED <> + +(** Current behavior: the serializer barrier does not wait for the sampling stage. *) +AcknowledgeSerializerFlush == + /\ pc = "FlushRequested" + /\ flushAcknowledged' = TRUE + /\ pc' = "FlushReturned" + /\ UNCHANGED <> + +Next == + AcceptCandidate \/ StartSampling \/ RequestFlush \/ AcknowledgeSerializerFlush + +Spec == Init /\ [][Next]_vars + +TypeOK == + /\ samplingQueued \in BOOLEAN + /\ samplingInFlight \in BOOLEAN + /\ serializerQueued \in BOOLEAN + /\ serialized \in BOOLEAN + /\ flushRequested \in BOOLEAN + /\ flushAcknowledged \in BOOLEAN + /\ pc \in {"Ready", "SamplingQueued", "SamplingInFlight", + "FlushRequested", "FlushReturned"} + +FlushCoversAcceptedCandidate == + flushAcknowledged => serialized + +============================================================================= From dc65bcc51df3c475fe6ffdd7e5991fdd4978f889 Mon Sep 17 00:00:00 2001 From: Charles Cote Date: Sat, 31 Oct 2026 18:13:48 +0100 Subject: [PATCH 3/7] Add Tier-1 formal verification suite --- .../DefaultDataStreamsMonitoringTest.java | 48 ++++++++++ ...efaultDataStreamsMonitoringTestBridge.java | 27 ++++++ formal-methods/README.md | 24 +++++ .../ContinuationLifecycle.cfg | 13 +++ .../ContinuationLifecycle.tla | 81 +++++++++++++++++ .../continuation-lifecycle/README.md | 9 ++ formal-methods/contracts/BarrierContracts.tla | 14 +++ .../contracts/OwnershipContracts.tla | 22 +++++ .../data-streams-close/DataStreamsClose.cfg | 7 ++ .../data-streams-close/DataStreamsClose.tla | 46 ++++++++++ formal-methods/data-streams-close/README.md | 9 ++ formal-methods/models.tsv | 9 ++ .../ListenerDistribution.cfg | 7 ++ .../ListenerDistribution.tla | 60 +++++++++++++ .../remote-config-listeners/README.md | 9 ++ formal-methods/run-models.sh | 78 ++++++++++++++++ formal-methods/telemetry-retry/README.md | 9 ++ .../TelemetryRetryOwnership.cfg | 8 ++ .../TelemetryRetryOwnership.tla | 90 +++++++++++++++++++ formal-methods/tier1/CONTRACTS.md | 30 +++++++ .../state/ProductStateSpecification.java | 22 +++++ 21 files changed, 622 insertions(+) create mode 100644 formal-methods/README.md create mode 100644 formal-methods/continuation-lifecycle/ContinuationLifecycle.cfg create mode 100644 formal-methods/continuation-lifecycle/ContinuationLifecycle.tla create mode 100644 formal-methods/continuation-lifecycle/README.md create mode 100644 formal-methods/contracts/BarrierContracts.tla create mode 100644 formal-methods/contracts/OwnershipContracts.tla create mode 100644 formal-methods/data-streams-close/DataStreamsClose.cfg create mode 100644 formal-methods/data-streams-close/DataStreamsClose.tla create mode 100644 formal-methods/data-streams-close/README.md create mode 100644 formal-methods/models.tsv create mode 100644 formal-methods/remote-config-listeners/ListenerDistribution.cfg create mode 100644 formal-methods/remote-config-listeners/ListenerDistribution.tla create mode 100644 formal-methods/remote-config-listeners/README.md create mode 100755 formal-methods/run-models.sh create mode 100644 formal-methods/telemetry-retry/README.md create mode 100644 formal-methods/telemetry-retry/TelemetryRetryOwnership.cfg create mode 100644 formal-methods/telemetry-retry/TelemetryRetryOwnership.tla create mode 100644 formal-methods/tier1/CONTRACTS.md diff --git a/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTest.java b/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTest.java index 9438632a088..f60ab8064eb 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTest.java @@ -1,10 +1,14 @@ package datadog.trace.core.datastreams; import static datadog.trace.core.datastreams.DefaultDataStreamsMonitoring.FEATURE_CHECK_INTERVAL_NANOS; +import static datadog.trace.core.datastreams.DefaultDataStreamsMonitoringTestBridge.getInboxSize; import static datadog.trace.core.datastreams.DefaultDataStreamsMonitoringTestBridge.getThreadState; +import static datadog.trace.core.datastreams.DefaultDataStreamsMonitoringTestBridge.interruptThread; import static datadog.trace.core.datastreams.DefaultDataStreamsMonitoringTestBridge.isInboxEmpty; +import static datadog.trace.core.datastreams.DefaultDataStreamsMonitoringTestBridge.isThreadAlive; import static java.util.concurrent.TimeUnit.SECONDS; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -43,6 +47,7 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Set; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -1474,6 +1479,49 @@ void kafkaConfigsFlushedOnClose() throws InterruptedException { payloadWriter.close(); } + @Test + void closeStopsWorkerWhenInboxIsFull() throws InterruptedException { + CountDownLatch workerStarting = new CountDownLatch(1); + CountDownLatch allowWorker = new CountDownLatch(1); + DDAgentFeaturesDiscovery features = stubFeatures(true); + when(features.getDataStreamsEndpoint()).thenReturn(null); + org.mockito.Mockito.doAnswer( + invocation -> { + workerStarting.countDown(); + allowWorker.await(); + return null; + }) + .when(features) + .discoverIfOutdated(); + DefaultDataStreamsMonitoring dataStreams = + new DefaultDataStreamsMonitoring( + mock(Sink.class), + features, + new ControllableTimeSource(), + () -> stubTraceConfig(true), + mock(DatastreamsPayloadWriter.class), + DEFAULT_BUCKET_DURATION_NANOS); + + try { + dataStreams.start(); + assertTrue(workerStarting.await(3, SECONDS)); + for (int i = 0; i < 1024; i++) { + dataStreams.trackBacklog(DataStreamsTags.EMPTY, i); + } + assertEquals(1024, getInboxSize(dataStreams)); + + dataStreams.close(); + + assertFalse( + isThreadAlive(dataStreams), + "close returned after losing the poison pill offered to a full inbox"); + } finally { + allowWorker.countDown(); + interruptThread(dataStreams); + dataStreams.close(); + } + } + @Test void kafkaConfigReportEqualsAndHashCodeWorkCorrectly() { Map sameConfig = new HashMap<>(); diff --git a/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTestBridge.java b/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTestBridge.java index 37c204757fd..9399e4dd4e2 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTestBridge.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/datastreams/DefaultDataStreamsMonitoringTestBridge.java @@ -37,4 +37,31 @@ static Thread.State getThreadState(DefaultDataStreamsMonitoring monitoring) { throw new IllegalStateException(e); } } + + static int getInboxSize(DefaultDataStreamsMonitoring monitoring) { + try { + MessagePassingQueue inbox = (MessagePassingQueue) INBOX_FIELD.get(monitoring); + return inbox.size(); + } catch (IllegalAccessException e) { + throw new IllegalStateException(e); + } + } + + static boolean isThreadAlive(DefaultDataStreamsMonitoring monitoring) { + try { + Thread thread = (Thread) THREAD_FIELD.get(monitoring); + return thread.isAlive(); + } catch (IllegalAccessException e) { + throw new IllegalStateException(e); + } + } + + static void interruptThread(DefaultDataStreamsMonitoring monitoring) { + try { + Thread thread = (Thread) THREAD_FIELD.get(monitoring); + thread.interrupt(); + } catch (IllegalAccessException e) { + throw new IllegalStateException(e); + } + } } diff --git a/formal-methods/README.md b/formal-methods/README.md new file mode 100644 index 00000000000..a1eb0d30f66 --- /dev/null +++ b/formal-methods/README.md @@ -0,0 +1,24 @@ +# Formal methods + +This directory contains small executable specifications for lifecycle and +concurrency contracts in the tracer. Models intentionally describe one +handoff or protocol at a time rather than attempting to represent the whole +agent. + +## Run the Tier-1 suite + +TLC requires Java 11 or newer. Set `TLA2TOOLS_JAR` to a TLA+ tools jar, or put +version 1.7.4 at `/tmp/tla2tools-1.7.4.jar`, then run: + +```shell +formal-methods/run-models.sh +``` + +The runner reads `formal-methods/models.tsv`. A model can be expected to pass +or to produce a specific, documented invariant violation. Expected violations +are diagnostic specifications for confirmed bugs and do not make the runner +fail. An unexpected pass, unexpected violation, parse failure, or wrong +invariant makes the runner return non-zero. + +See `formal-methods/tier1/CONTRACTS.md` for the coverage map and the Java code +represented by each model. diff --git a/formal-methods/continuation-lifecycle/ContinuationLifecycle.cfg b/formal-methods/continuation-lifecycle/ContinuationLifecycle.cfg new file mode 100644 index 00000000000..d19b3045017 --- /dev/null +++ b/formal-methods/continuation-lifecycle/ContinuationLifecycle.cfg @@ -0,0 +1,13 @@ +CONSTANTS + Threads = {t1, t2} + MaxResumes = 2 + +SPECIFICATION Spec + +INVARIANTS + TypeOK + FinalizedAtMostOnce + TerminalIsQuiescent + FinalizedIffTerminal + +CHECK_DEADLOCK FALSE diff --git a/formal-methods/continuation-lifecycle/ContinuationLifecycle.tla b/formal-methods/continuation-lifecycle/ContinuationLifecycle.tla new file mode 100644 index 00000000000..fb20bc11536 --- /dev/null +++ b/formal-methods/continuation-lifecycle/ContinuationLifecycle.tla @@ -0,0 +1,81 @@ +------------------------ MODULE ContinuationLifecycle ------------------------ +EXTENDS Integers, Naturals, FiniteSets + +(** + * Abstract lifecycle shared by ScopeContinuation and + * ThreadLocalContextManager.ContextContinuationImpl. + *) + +CONSTANTS Threads, MaxResumes + +ASSUME /\ Threads # {} + /\ MaxResumes \in Nat \ {0} + +ScopeTokens == Threads \X (1..MaxResumes) + +VARIABLES held, terminal, openScopes, finalizations + +vars == <> + +Init == + /\ held = FALSE + /\ terminal = FALSE + /\ openScopes = {} + /\ finalizations = 0 + +Hold == + /\ ~terminal + /\ ~held + /\ openScopes = {} + /\ held' = TRUE + /\ UNCHANGED <> + +Resume(thread, slot) == + LET token == <> IN + /\ ~terminal + /\ token \notin openScopes + /\ openScopes' = openScopes \cup {token} + /\ UNCHANGED <> + +Close(thread, slot) == + LET token == <> + remaining == openScopes \ {token} + IN + /\ token \in openScopes + /\ openScopes' = remaining + /\ IF ~held /\ remaining = {} + THEN /\ terminal' = TRUE + /\ finalizations' = finalizations + 1 + ELSE /\ UNCHANGED <> + /\ UNCHANGED held + +Release == + /\ ~terminal + /\ (held \/ openScopes = {}) + /\ held' = FALSE + /\ IF openScopes = {} + THEN /\ terminal' = TRUE + /\ finalizations' = finalizations + 1 + ELSE /\ UNCHANGED <> + /\ UNCHANGED openScopes + +Next == + Hold \/ Release \/ + \E thread \in Threads, slot \in 1..MaxResumes: + Resume(thread, slot) \/ Close(thread, slot) + +Spec == Init /\ [][Next]_vars + +TypeOK == + /\ held \in BOOLEAN + /\ terminal \in BOOLEAN + /\ openScopes \subseteq ScopeTokens + /\ finalizations \in Nat + +FinalizedAtMostOnce == finalizations <= 1 + +TerminalIsQuiescent == terminal => (openScopes = {} /\ ~held) + +FinalizedIffTerminal == (finalizations = 1) <=> terminal + +============================================================================= diff --git a/formal-methods/continuation-lifecycle/README.md b/formal-methods/continuation-lifecycle/README.md new file mode 100644 index 00000000000..c7fc7b161fb --- /dev/null +++ b/formal-methods/continuation-lifecycle/README.md @@ -0,0 +1,9 @@ +# Continuation lifecycle + +Models the abstract hold, resume, scope-close, and release protocol shared by +the context continuation implementations. Within the configured bounds, TLC +checks that finalization occurs at most once and only after all holds and +resumed scopes are gone. + +This is a passing executable contract. A CAS-level refinement can be added +without changing the higher-level invariants. diff --git a/formal-methods/contracts/BarrierContracts.tla b/formal-methods/contracts/BarrierContracts.tla new file mode 100644 index 00000000000..dc129f2a885 --- /dev/null +++ b/formal-methods/contracts/BarrierContracts.tla @@ -0,0 +1,14 @@ +------------------------- MODULE BarrierContracts -------------------------- + +(** Reusable predicates for flush, close, and command acknowledgement. *) + +FlushCoversAccepted(accepted, delivered, explicitlyDropped, acknowledged) == + acknowledged => (~accepted \/ delivered \/ explicitlyDropped) + +CloseStopsWorkers(closeReturned, workersRunning) == + closeReturned => ~workersRunning + +WaitingCommandHasOwner(waiting, queued, processing, acknowledged, cancelled) == + waiting => (queued \/ processing \/ acknowledged \/ cancelled) + +============================================================================= diff --git a/formal-methods/contracts/OwnershipContracts.tla b/formal-methods/contracts/OwnershipContracts.tla new file mode 100644 index 00000000000..6d29f50d9b5 --- /dev/null +++ b/formal-methods/contracts/OwnershipContracts.tla @@ -0,0 +1,22 @@ +------------------------ MODULE OwnershipContracts ------------------------- +EXTENDS FiniteSets, Naturals + +(** Reusable predicates for work that moves between asynchronous components. *) + +ActiveOwnerCount(queued, processing, retained) == + Cardinality({owner \in {"queued", "processing", "retained"} : + CASE owner = "queued" -> queued + [] owner = "processing" -> processing + [] owner = "retained" -> retained}) + +HasOwnerOrTerminal(queued, processing, retained, delivered, dropped) == + queued \/ processing \/ retained \/ delivered \/ dropped + +AtMostOneActiveOwner(queued, processing, retained) == + ActiveOwnerCount(queued, processing, retained) <= 1 + +TerminalIsExclusive(queued, processing, retained, delivered, dropped) == + (delivered \/ dropped) => + (~queued /\ ~processing /\ ~retained /\ ~(delivered /\ dropped)) + +============================================================================= diff --git a/formal-methods/data-streams-close/DataStreamsClose.cfg b/formal-methods/data-streams-close/DataStreamsClose.cfg new file mode 100644 index 00000000000..fa2eb6581b3 --- /dev/null +++ b/formal-methods/data-streams-close/DataStreamsClose.cfg @@ -0,0 +1,7 @@ +SPECIFICATION Spec + +INVARIANTS + TypeOK + CloseReturnsOnlyAfterWorkerStops + +CHECK_DEADLOCK FALSE diff --git a/formal-methods/data-streams-close/DataStreamsClose.tla b/formal-methods/data-streams-close/DataStreamsClose.tla new file mode 100644 index 00000000000..5c58c1f7ed3 --- /dev/null +++ b/formal-methods/data-streams-close/DataStreamsClose.tla @@ -0,0 +1,46 @@ +-------------------------- MODULE DataStreamsClose -------------------------- + +(** + * Models DefaultDataStreamsMonitoring.close with a full bounded inbox. The + * non-blocking offer of POISON_PILL fails, join times out, and close returns + * while the worker remains alive without a future stop signal. + *) + +VARIABLES inboxFull, poisonQueued, workerAlive, closeReturned, pc + +vars == <> + +Init == + /\ inboxFull = TRUE + /\ poisonQueued = FALSE + /\ workerAlive = TRUE + /\ closeReturned = FALSE + /\ pc = "WorkerBlocked" + +OfferPoison == + /\ pc = "WorkerBlocked" + /\ poisonQueued' = IF inboxFull THEN FALSE ELSE TRUE + /\ pc' = "Joining" + /\ UNCHANGED <> + +JoinTimesOut == + /\ pc = "Joining" + /\ ~poisonQueued + /\ closeReturned' = TRUE + /\ pc' = "Done" + /\ UNCHANGED <> + +Next == OfferPoison \/ JoinTimesOut + +Spec == Init /\ [][Next]_vars + +TypeOK == + /\ inboxFull \in BOOLEAN + /\ poisonQueued \in BOOLEAN + /\ workerAlive \in BOOLEAN + /\ closeReturned \in BOOLEAN + /\ pc \in {"WorkerBlocked", "Joining", "Done"} + +CloseReturnsOnlyAfterWorkerStops == closeReturned => ~workerAlive + +============================================================================= diff --git a/formal-methods/data-streams-close/README.md b/formal-methods/data-streams-close/README.md new file mode 100644 index 00000000000..442a0a9f71d --- /dev/null +++ b/formal-methods/data-streams-close/README.md @@ -0,0 +1,9 @@ +# Data-streams close signalling + +Checks that `DefaultDataStreamsMonitoring.close()` cannot return while its +worker remains alive. With a full 1024-element inbox, the non-blocking poison +offer fails and the worker has no stop signal. TLC produces a three-state +counterexample. + +The corresponding regression is +`DefaultDataStreamsMonitoringTest.closeStopsWorkerWhenInboxIsFull`. diff --git a/formal-methods/models.tsv b/formal-methods/models.tsv new file mode 100644 index 00000000000..edcbd3a60fe --- /dev/null +++ b/formal-methods/models.tsv @@ -0,0 +1,9 @@ +# id expectation directory module invariant +long-running-tracker violation long-running-traces TrackerOwnership LiveTraceHasOwner +pending-trace-flush violation pending-trace-flush FlushDrainOwnership PendingChildHasWriter +flush-dump-marker violation flush-dump-marker FlushDumpMarker WaitingDumpHasOwner +span-sampling-flush violation span-sampling-flush SpanSamplingFlush FlushCoversAcceptedCandidate +continuation-lifecycle pass continuation-lifecycle ContinuationLifecycle FinalizedAtMostOnce +remote-config-listeners violation remote-config-listeners ListenerDistribution CachedTargetIsFullyDistributed +telemetry-retry pass telemetry-retry TelemetryRetryOwnership AcceptedEventHasOwnerOrTerminal +data-streams-close violation data-streams-close DataStreamsClose CloseReturnsOnlyAfterWorkerStops diff --git a/formal-methods/remote-config-listeners/ListenerDistribution.cfg b/formal-methods/remote-config-listeners/ListenerDistribution.cfg new file mode 100644 index 00000000000..6bf5b47c718 --- /dev/null +++ b/formal-methods/remote-config-listeners/ListenerDistribution.cfg @@ -0,0 +1,7 @@ +SPECIFICATION Spec + +INVARIANTS + TypeOK + CachedTargetIsFullyDistributed + +CHECK_DEADLOCK FALSE diff --git a/formal-methods/remote-config-listeners/ListenerDistribution.tla b/formal-methods/remote-config-listeners/ListenerDistribution.tla new file mode 100644 index 00000000000..66ceda17f6e --- /dev/null +++ b/formal-methods/remote-config-listeners/ListenerDistribution.tla @@ -0,0 +1,60 @@ +------------------------ MODULE ListenerDistribution ------------------------ + +(** + * Models ProductState.callListenerApplyTarget when the first product listener + * throws a generic exception. The exception handler caches the target and the + * next identical response is treated as unchanged, so later healthy listeners + * never receive it. + *) + +VARIABLES + firstListenerCalled, + healthyListenerCalled, + targetCached, + retryPending, + pc + +vars == + <> + +Init == + /\ firstListenerCalled = FALSE + /\ healthyListenerCalled = FALSE + /\ targetCached = FALSE + /\ retryPending = TRUE + /\ pc = "ResponseReceived" + +FirstListenerThrows == + /\ pc = "ResponseReceived" + /\ firstListenerCalled' = TRUE + /\ pc' = "HandlingFailure" + /\ UNCHANGED <> + +(** Current generic-exception path calls updateConfigState and caches the hash. *) +CacheFailedTarget == + /\ pc = "HandlingFailure" + /\ targetCached' = TRUE + /\ retryPending' = FALSE + /\ pc' = "Cached" + /\ UNCHANGED <> + +ReplaySameResponse == + /\ pc = "Cached" + /\ pc' = "Done" + /\ UNCHANGED <> + +Next == FirstListenerThrows \/ CacheFailedTarget \/ ReplaySameResponse + +Spec == Init /\ [][Next]_vars + +TypeOK == + /\ firstListenerCalled \in BOOLEAN + /\ healthyListenerCalled \in BOOLEAN + /\ targetCached \in BOOLEAN + /\ retryPending \in BOOLEAN + /\ pc \in {"ResponseReceived", "HandlingFailure", "Cached", "Done"} + +CachedTargetIsFullyDistributed == + targetCached => (healthyListenerCalled \/ retryPending) + +============================================================================= diff --git a/formal-methods/remote-config-listeners/README.md b/formal-methods/remote-config-listeners/README.md new file mode 100644 index 00000000000..a81e0d27222 --- /dev/null +++ b/formal-methods/remote-config-listeners/README.md @@ -0,0 +1,9 @@ +# Remote-config listener distribution + +Checks that caching a target means every healthy listener received it, or that +a retry remains pending. The current generic-exception path catches around the +entire listener loop, caches the failed target, and suppresses retries for an +identical response. TLC produces a three-state counterexample. + +The corresponding regression is +`ProductStateSpecification.listenerFailureDoesNotPreventLaterListenersFromReceivingConfiguration`. diff --git a/formal-methods/run-models.sh b/formal-methods/run-models.sh new file mode 100755 index 00000000000..06fbf369c87 --- /dev/null +++ b/formal-methods/run-models.sh @@ -0,0 +1,78 @@ +#!/usr/bin/env bash + +set -u + +repo_root=$(cd "$(dirname "$0")/.." && pwd) +manifest="$repo_root/formal-methods/models.tsv" +tla_jar=${TLA2TOOLS_JAR:-/tmp/tla2tools-1.7.4.jar} +metadata_root=${TLC_METADATA_ROOT:-/tmp/dd-trace-java-tlc/tier1} +failures=0 + +if [[ ! -f "$tla_jar" ]]; then + echo "TLA+ tools jar not found: $tla_jar" >&2 + echo "Set TLA2TOOLS_JAR or install version 1.7.4 at /tmp/tla2tools-1.7.4.jar." >&2 + exit 2 +fi + +contract_output=$(mktemp "${TMPDIR:-/tmp}/dd-trace-tla-contracts.XXXXXX") +( + cd "$repo_root/formal-methods/contracts" || exit 1 + java -cp "$tla_jar" tla2sany.SANY OwnershipContracts.tla BarrierContracts.tla +) >"$contract_output" 2>&1 +contract_status=$? + +if [[ $contract_status -ne 0 ]] || grep -Eq "Semantic errors:|Fatal errors:|Parse Error|Lexical error" "$contract_output"; then + echo "FAIL shared-contracts (parse or semantic error)" >&2 + sed -n '1,160p' "$contract_output" >&2 + failures=$((failures + 1)) +else + echo "PASS shared-contracts" +fi +rm -f "$contract_output" + +while IFS=$'\t' read -r id expectation directory module invariant; do + if [[ -z "$id" || "$id" == \#* ]]; then + continue + fi + + model_dir="$repo_root/formal-methods/$directory" + output_file=$(mktemp "${TMPDIR:-/tmp}/dd-trace-tlc-${id}.XXXXXX") + metadata_dir="$metadata_root/$id" + mkdir -p "$metadata_dir" + + java -XX:+UseParallelGC -jar "$tla_jar" \ + -config "$model_dir/$module.cfg" \ + -metadir "$metadata_dir" \ + "$model_dir/$module.tla" >"$output_file" 2>&1 + status=$? + + if [[ "$expectation" == "pass" ]]; then + if [[ $status -eq 0 ]] && grep -q "Model checking completed. No error has been found." "$output_file"; then + echo "PASS $id" + else + echo "FAIL $id (expected model-check success)" >&2 + sed -n '1,160p' "$output_file" >&2 + failures=$((failures + 1)) + fi + elif [[ "$expectation" == "violation" ]]; then + if grep -q "Invariant $invariant is violated" "$output_file"; then + echo "XFAIL $id ($invariant)" + else + echo "FAIL $id (expected violation of $invariant)" >&2 + sed -n '1,160p' "$output_file" >&2 + failures=$((failures + 1)) + fi + else + echo "FAIL $id (unknown expectation: $expectation)" >&2 + failures=$((failures + 1)) + fi + + rm -f "$output_file" +done < "$manifest" + +if [[ $failures -ne 0 ]]; then + echo "$failures model expectation(s) failed." >&2 + exit 1 +fi + +echo "All Tier-1 model expectations matched." diff --git a/formal-methods/telemetry-retry/README.md b/formal-methods/telemetry-retry/README.md new file mode 100644 index 00000000000..47f56060a13 --- /dev/null +++ b/formal-methods/telemetry-retry/README.md @@ -0,0 +1,9 @@ +# Telemetry retry ownership + +Models one event through the primary queue, request construction, retry +buffer, and the current bounded retry policy. It checks that accepted telemetry +always has an owner or an explicit terminal outcome, and that delivery and +policy-drop are mutually exclusive. + +This model passes. The `policyDropped` state documents the existing behavior +where events are discarded after the buffered retry also fails. diff --git a/formal-methods/telemetry-retry/TelemetryRetryOwnership.cfg b/formal-methods/telemetry-retry/TelemetryRetryOwnership.cfg new file mode 100644 index 00000000000..0b80451cc8b --- /dev/null +++ b/formal-methods/telemetry-retry/TelemetryRetryOwnership.cfg @@ -0,0 +1,8 @@ +SPECIFICATION Spec + +INVARIANTS + TypeOK + AcceptedEventHasOwnerOrTerminal + TerminalStateIsExclusive + +CHECK_DEADLOCK FALSE diff --git a/formal-methods/telemetry-retry/TelemetryRetryOwnership.tla b/formal-methods/telemetry-retry/TelemetryRetryOwnership.tla new file mode 100644 index 00000000000..68ecc038a7e --- /dev/null +++ b/formal-methods/telemetry-retry/TelemetryRetryOwnership.tla @@ -0,0 +1,90 @@ +---------------------- MODULE TelemetryRetryOwnership ----------------------- +EXTENDS Naturals + +(** + * Models TelemetryService moving one accepted event from its source queue to + * BufferedEvents for one retry. A second failed attempt is an explicit bounded + * retry-policy drop, matching the current implementation. + *) + +VARIABLES queued, requestOwns, buffered, delivered, policyDropped, attempts, pc + +vars == <> + +Init == + /\ queued = TRUE + /\ requestOwns = FALSE + /\ buffered = FALSE + /\ delivered = FALSE + /\ policyDropped = FALSE + /\ attempts = 0 + /\ pc = "Accepted" + +BuildFirstRequest == + /\ pc = "Accepted" + /\ queued' = FALSE + /\ requestOwns' = TRUE + /\ buffered' = TRUE + /\ attempts' = 1 + /\ pc' = "FirstInFlight" + /\ UNCHANGED <> + +FirstSucceeds == + /\ pc = "FirstInFlight" + /\ requestOwns' = FALSE + /\ buffered' = FALSE + /\ delivered' = TRUE + /\ pc' = "Done" + /\ UNCHANGED <> + +FirstFails == + /\ pc = "FirstInFlight" + /\ requestOwns' = FALSE + /\ pc' = "RetryBuffered" + /\ UNCHANGED <> + +BuildRetry == + /\ pc = "RetryBuffered" + /\ buffered' = FALSE + /\ requestOwns' = TRUE + /\ attempts' = 2 + /\ pc' = "RetryInFlight" + /\ UNCHANGED <> + +RetrySucceeds == + /\ pc = "RetryInFlight" + /\ requestOwns' = FALSE + /\ delivered' = TRUE + /\ pc' = "Done" + /\ UNCHANGED <> + +RetryFails == + /\ pc = "RetryInFlight" + /\ requestOwns' = FALSE + /\ policyDropped' = TRUE + /\ pc' = "Done" + /\ UNCHANGED <> + +Next == + BuildFirstRequest \/ FirstSucceeds \/ FirstFails \/ BuildRetry \/ + RetrySucceeds \/ RetryFails + +Spec == Init /\ [][Next]_vars + +TypeOK == + /\ queued \in BOOLEAN + /\ requestOwns \in BOOLEAN + /\ buffered \in BOOLEAN + /\ delivered \in BOOLEAN + /\ policyDropped \in BOOLEAN + /\ attempts \in 0..2 + /\ pc \in {"Accepted", "FirstInFlight", "RetryBuffered", "RetryInFlight", "Done"} + +AcceptedEventHasOwnerOrTerminal == + queued \/ requestOwns \/ buffered \/ delivered \/ policyDropped + +TerminalStateIsExclusive == + (delivered \/ policyDropped) => + (~queued /\ ~requestOwns /\ ~buffered /\ ~(delivered /\ policyDropped)) + +============================================================================= diff --git a/formal-methods/tier1/CONTRACTS.md b/formal-methods/tier1/CONTRACTS.md new file mode 100644 index 00000000000..c1dce007c3c --- /dev/null +++ b/formal-methods/tier1/CONTRACTS.md @@ -0,0 +1,30 @@ +# Tier-1 contract coverage + +The Tier-1 suite focuses on ownership, acknowledgement, flush, and close +boundaries where failures can silently lose trace data or leave agent workers +alive. + +| Area | Contract | Model | Current result | Java evidence | +|---|---|---|---|---| +| Pending traces | Every live trace has an owner | `TrackerOwnership` | Known violation | `LongRunningTracesTrackerTest.fullTrackerLeavesOverflowTraceForBufferFallback` | +| Pending traces | A late child retains a writer after flush | `FlushDrainOwnership` | Known violation | `PendingTraceBufferTest.flushDrainClearsEnqueuedStateBeforeWriting` | +| Buffer commands | A waiting command remains owned or acknowledged | `FlushDumpMarker` | Known violation | `PendingTraceBufferTest.flushDrainDoesNotConsumeTracerDumpMarker` | +| Writer pipeline | Flush covers accepted upstream sampling work | `SpanSamplingFlush` | Known violation | `TraceProcessingWorkerTest.flushWaitsForInFlightSingleSpanSampling` | +| Context | Continuations finalize once and only when quiescent | `ContinuationLifecycle` | Passes bounded model | Existing context lifecycle tests | +| Remote config | Cached targets were distributed to every healthy listener | `ListenerDistribution` | Known violation | `ProductStateSpecification.listenerFailureDoesNotPreventLaterListenersFromReceivingConfiguration` | +| Telemetry | Accepted events remain queued, retry-buffered, delivered, or policy-dropped | `TelemetryRetryOwnership` | Passes bounded model | Existing `TelemetryServiceTest` retry tests | +| Data streams | `close()` cannot return with its worker alive | `DataStreamsClose` | Known violation | `DefaultDataStreamsMonitoringTest.closeStopsWorkerWhenInboxIsFull` | + +## Review rule + +Every new asynchronous Tier-1 handoff should identify: + +1. the owner before and after the handoff; +2. receiver rejection and capacity behavior; +3. the barrier semantics for flush and close; +4. the terminal states: delivered, explicitly dropped, or cancelled; +5. a deterministic Java schedule for every counterexample. + +The shared predicates in `formal-methods/contracts` provide the vocabulary; +individual models remain small so their counterexamples map directly to Java +methods and tests. diff --git a/remote-config/remote-config-core/src/test/java/datadog/remoteconfig/state/ProductStateSpecification.java b/remote-config/remote-config-core/src/test/java/datadog/remoteconfig/state/ProductStateSpecification.java index 360c2d0a99c..b5e420d0477 100644 --- a/remote-config/remote-config-core/src/test/java/datadog/remoteconfig/state/ProductStateSpecification.java +++ b/remote-config/remote-config-core/src/test/java/datadog/remoteconfig/state/ProductStateSpecification.java @@ -8,6 +8,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -190,6 +191,27 @@ void testErrorHandlingDuringApply() throws Exception { verify(listener).commit(hinter); } + @Test + void listenerFailureDoesNotPreventLaterListenersFromReceivingConfiguration() throws Exception { + ProductState productState = new ProductState(Product.ASM_DATA); + ProductListener failingListener = mock(ProductListener.class); + ProductListener healthyListener = mock(ProductListener.class); + productState.addProductListener(failingListener); + productState.addProductListener(healthyListener); + + RemoteConfigResponse response = + buildResponse(targets("org/ASM_DATA/config1/foo", new TargetSpec(1, 8, "hash1"))); + ParsedConfigKey key = ParsedConfigKey.parse("org/ASM_DATA/config1/foo"); + doThrow(new RuntimeException("Listener error")) + .when(failingListener) + .accept(any(), any(), any()); + + assertTrue(productState.apply(response, Collections.singletonList(key), hinter)); + assertFalse(productState.apply(response, Collections.singletonList(key), hinter)); + + verify(healthyListener, times(1)).accept(any(), any(), any()); + } + @Test void testReportableExceptionIsRecorded() throws Exception { // a ProductState From 0843620e49a51b8444d19f22446d0c30782afb68 Mon Sep 17 00:00:00 2001 From: Charles Cote Date: Sat, 31 Oct 2026 18:28:11 +0100 Subject: [PATCH 4/7] Add deterministic writer state-machine testing --- ...TraceProcessingWorkerStateMachineTest.java | 179 ++++++++++++++++++ formal-methods/README.md | 7 + formal-methods/tier1/CONTRACTS.md | 4 + formal-methods/tier1/STATE_MACHINE_TESTING.md | 28 +++ 4 files changed, 218 insertions(+) create mode 100644 dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerStateMachineTest.java create mode 100644 formal-methods/tier1/STATE_MACHINE_TESTING.md diff --git a/dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerStateMachineTest.java b/dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerStateMachineTest.java new file mode 100644 index 00000000000..6c8c8f57663 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerStateMachineTest.java @@ -0,0 +1,179 @@ +package datadog.trace.common.writer; + +import static datadog.trace.api.sampling.PrioritySampling.USER_DROP; +import static datadog.trace.api.sampling.PrioritySampling.USER_KEEP; +import static datadog.trace.common.writer.ddagent.Prioritization.FAST_LANE; +import static datadog.trace.common.writer.ddagent.PrioritizationStrategy.PublishResult.ENQUEUED_FOR_SERIALIZATION; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; + +import datadog.trace.common.writer.ddagent.PrioritizationStrategy.PublishResult; +import datadog.trace.core.DDSpan; +import datadog.trace.core.monitor.HealthMetrics; +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.IdentityHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; + +class TraceProcessingWorkerStateMachineTest { + + private enum Command { + PUBLISH_PRIMARY, + PUBLISH_SECONDARY + } + + @Test + void generatedCommandSequencesRespectFlushBarrier() throws Exception { + Counterexample counterexample = null; + + for (List commands : generatedPublishPrefixes(4)) { + Counterexample candidate = execute(commands); + if (candidate != null) { + counterexample = candidate; + break; + } + } + + assertNull( + counterexample, + "flush acknowledged before all previously accepted traces were serialized; minimal schedule: " + + counterexample); + } + + private static Counterexample execute(List commands) throws Exception { + Object observationsLock = new Object(); + Map, Integer> traceIds = new IdentityHashMap<>(); + Set serialized = new LinkedHashSet<>(); + AtomicReference> serializedAtFlush = new AtomicReference<>(); + PayloadDispatcher dispatcher = mock(PayloadDispatcher.class); + doAnswer( + invocation -> { + List trace = invocation.getArgument(0); + synchronized (observationsLock) { + serialized.add(traceIds.get(trace)); + } + return null; + }) + .when(dispatcher) + .addTrace(any()); + doAnswer( + invocation -> { + synchronized (observationsLock) { + serializedAtFlush.set(new LinkedHashSet<>(serialized)); + } + return null; + }) + .when(dispatcher) + .flush(); + + Set accepted = new LinkedHashSet<>(); + ExecutorService flushExecutor = Executors.newSingleThreadExecutor(); + try (TraceProcessingWorker worker = + new TraceProcessingWorker( + 16, + mock(HealthMetrics.class), + dispatcher, + () -> false, + FAST_LANE, + 100, + SECONDS, + null)) { + int primaryEvents = 0; + for (int id = 0; id < commands.size(); id++) { + Command command = commands.get(id); + List trace = Collections.singletonList(mock(DDSpan.class)); + traceIds.put(trace, id); + int priority; + if (command == Command.PUBLISH_PRIMARY) { + priority = USER_KEEP; + primaryEvents++; + } else { + priority = USER_DROP; + } + + PublishResult result = worker.publish(trace.get(0), priority, trace); + if (result == ENQUEUED_FOR_SERIALIZATION) { + accepted.add(id); + } + } + + Future flush = flushExecutor.submit(() -> worker.flush(3, SECONDS)); + awaitPrimaryQueueSize(worker, primaryEvents + 1); + worker.start(); + + assertTrue(flush.get(3, SECONDS), "generated flush did not complete"); + Set observed = serializedAtFlush.get(); + if (!accepted.equals(observed)) { + return new Counterexample(commands, accepted, observed); + } + return null; + } finally { + flushExecutor.shutdownNow(); + } + } + + private static void awaitPrimaryQueueSize(TraceProcessingWorker worker, int expectedSize) { + long deadline = System.nanoTime() + SECONDS.toNanos(3); + while (worker.getPrimaryQueue().size() != expectedSize && System.nanoTime() < deadline) { + Thread.yield(); + } + assertEquals(expectedSize, worker.getPrimaryQueue().size(), "flush marker was not enqueued"); + } + + private static List> generatedPublishPrefixes(int maximumLength) { + List> sequences = new ArrayList<>(); + for (int length = 1; length <= maximumLength; length++) { + generateSequences(length, new ArrayList<>(), sequences); + } + return sequences; + } + + private static void generateSequences( + int remaining, List prefix, List> sequences) { + if (remaining == 0) { + sequences.add(new ArrayList<>(prefix)); + return; + } + for (Command command : EnumSet.allOf(Command.class)) { + prefix.add(command); + generateSequences(remaining - 1, prefix, sequences); + prefix.remove(prefix.size() - 1); + } + } + + private static final class Counterexample { + private final List commands; + private final Set accepted; + private final Set serializedAtFlush; + + private Counterexample( + List commands, Set accepted, Set serializedAtFlush) { + this.commands = new ArrayList<>(commands); + this.accepted = new LinkedHashSet<>(accepted); + this.serializedAtFlush = new LinkedHashSet<>(serializedAtFlush); + } + + @Override + public String toString() { + return commands + + " -> FLUSH; accepted=" + + accepted + + ", serializedAtFlush=" + + serializedAtFlush; + } + } +} diff --git a/formal-methods/README.md b/formal-methods/README.md index a1eb0d30f66..c9c7c72f543 100644 --- a/formal-methods/README.md +++ b/formal-methods/README.md @@ -22,3 +22,10 @@ invariant makes the runner return non-zero. See `formal-methods/tier1/CONTRACTS.md` for the coverage map and the Java code represented by each model. + +## Deterministic Java state machines + +The Tier-1 suite also generates bounded command sequences against selected Java +implementations and compares their observations with a small reference model. +See `formal-methods/tier1/STATE_MACHINE_TESTING.md` for the scheduling and +counterexample-reduction rules. diff --git a/formal-methods/tier1/CONTRACTS.md b/formal-methods/tier1/CONTRACTS.md index c1dce007c3c..bb0b3e89b9b 100644 --- a/formal-methods/tier1/CONTRACTS.md +++ b/formal-methods/tier1/CONTRACTS.md @@ -10,11 +10,15 @@ alive. | Pending traces | A late child retains a writer after flush | `FlushDrainOwnership` | Known violation | `PendingTraceBufferTest.flushDrainClearsEnqueuedStateBeforeWriting` | | Buffer commands | A waiting command remains owned or acknowledged | `FlushDumpMarker` | Known violation | `PendingTraceBufferTest.flushDrainDoesNotConsumeTracerDumpMarker` | | Writer pipeline | Flush covers accepted upstream sampling work | `SpanSamplingFlush` | Known violation | `TraceProcessingWorkerTest.flushWaitsForInFlightSingleSpanSampling` | +| Writer pipeline | Flush covers accepted secondary-queue work | Generated reference model | Known violation | `TraceProcessingWorkerStateMachineTest.generatedCommandSequencesRespectFlushBarrier` | | Context | Continuations finalize once and only when quiescent | `ContinuationLifecycle` | Passes bounded model | Existing context lifecycle tests | | Remote config | Cached targets were distributed to every healthy listener | `ListenerDistribution` | Known violation | `ProductStateSpecification.listenerFailureDoesNotPreventLaterListenersFromReceivingConfiguration` | | Telemetry | Accepted events remain queued, retry-buffered, delivered, or policy-dropped | `TelemetryRetryOwnership` | Passes bounded model | Existing `TelemetryServiceTest` retry tests | | Data streams | `close()` cannot return with its worker alive | `DataStreamsClose` | Known violation | `DefaultDataStreamsMonitoringTest.closeStopsWorkerWhenInboxIsFull` | +The writer pipeline also has a generated Java state-machine harness. See +`STATE_MACHINE_TESTING.md` for its reference model and minimized schedules. + ## Review rule Every new asynchronous Tier-1 handoff should identify: diff --git a/formal-methods/tier1/STATE_MACHINE_TESTING.md b/formal-methods/tier1/STATE_MACHINE_TESTING.md new file mode 100644 index 00000000000..37cf7a199a0 --- /dev/null +++ b/formal-methods/tier1/STATE_MACHINE_TESTING.md @@ -0,0 +1,28 @@ +# Deterministic state-machine testing + +The writer state-machine harness complements the TLA+ models by executing +generated command sequences against the Java implementation. A small reference +model records every trace accepted before a flush. The implementation must have +serialized the same set when the flush is acknowledged. + +`TraceProcessingWorkerStateMachineTest` generates publish prefixes in increasing +length and lexicographic order. For each prefix it deliberately queues the flush +marker before starting the serializer. This removes timing from the schedule and +makes the first mismatch the shortest reproducible counterexample. + +The current minimal counterexample is: + +```text +PUBLISH_SECONDARY -> FLUSH +``` + +The secondary trace is accepted first, but the flush marker is placed in the +primary queue. The serializer consumes and acknowledges the primary marker +before visiting the secondary queue, so the flush barrier does not cover all +previously accepted work. + +Future state machines should preserve three properties: + +1. generated schedules are deterministic and bounded; +2. the oracle is an independent reference model, not internal queue state; +3. failures report a replayable, minimal command sequence. From 49aa8bc58678020a8782848c11eeeeafa4bc6206 Mon Sep 17 00:00:00 2001 From: Charles Cote Date: Sat, 31 Oct 2026 18:40:58 +0100 Subject: [PATCH 5/7] Expand Tier-1 state-machine fault exploration --- ...TraceProcessingWorkerStateMachineTest.java | 253 ++++++++++++++---- formal-methods/tier1/CONTRACTS.md | 4 +- formal-methods/tier1/STATE_MACHINE_TESTING.md | 29 +- .../ProductStateListenerStateMachineTest.java | 178 ++++++++++++ 4 files changed, 405 insertions(+), 59 deletions(-) create mode 100644 remote-config/remote-config-core/src/test/java/datadog/remoteconfig/state/ProductStateListenerStateMachineTest.java diff --git a/dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerStateMachineTest.java b/dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerStateMachineTest.java index 6c8c8f57663..881b6faa806 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerStateMachineTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/writer/TraceProcessingWorkerStateMachineTest.java @@ -3,10 +3,12 @@ import static datadog.trace.api.sampling.PrioritySampling.USER_DROP; import static datadog.trace.api.sampling.PrioritySampling.USER_KEEP; import static datadog.trace.common.writer.ddagent.Prioritization.FAST_LANE; +import static datadog.trace.common.writer.ddagent.PrioritizationStrategy.PublishResult.DROPPED_BUFFER_OVERFLOW; +import static datadog.trace.common.writer.ddagent.PrioritizationStrategy.PublishResult.DROPPED_BUFFER_OVERFLOW_SAMPLED_OUT; import static datadog.trace.common.writer.ddagent.PrioritizationStrategy.PublishResult.ENQUEUED_FOR_SERIALIZATION; import static java.util.concurrent.TimeUnit.SECONDS; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.doAnswer; @@ -17,6 +19,7 @@ import datadog.trace.core.monitor.HealthMetrics; import java.util.ArrayList; import java.util.Collections; +import java.util.EnumMap; import java.util.EnumSet; import java.util.IdentityHashMap; import java.util.LinkedHashSet; @@ -26,39 +29,79 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; -import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; class TraceProcessingWorkerStateMachineTest { private enum Command { PUBLISH_PRIMARY, - PUBLISH_SECONDARY + PUBLISH_SECONDARY, + FLUSH + } + + private enum Finding { + SECONDARY_QUEUE_BYPASSES_FLUSH, + PRIMARY_QUEUE_BYPASSES_FLUSH, + MIXED_ACCEPTED_WORK_BYPASSES_FLUSH, + WORK_CROSSES_FLUSH_FROM_THE_FUTURE, + FLUSH_DID_NOT_COMPLETE, + FLUSH_OBSERVATION_MISSING } @Test - void generatedCommandSequencesRespectFlushBarrier() throws Exception { - Counterexample counterexample = null; - - for (List commands : generatedPublishPrefixes(4)) { - Counterexample candidate = execute(commands); - if (candidate != null) { - counterexample = candidate; - break; + void generatedCommandSequencesRespectFlushBarriers() throws Exception { + Map distinctFindings = new EnumMap<>(Finding.class); + + for (List commands : generatedCommandSequences(4)) { + for (Counterexample counterexample : executeFlushStateMachine(commands)) { + distinctFindings.putIfAbsent(counterexample.finding, counterexample); } } - assertNull( - counterexample, - "flush acknowledged before all previously accepted traces were serialized; minimal schedule: " - + counterexample); + assertTrue( + distinctFindings.isEmpty(), + () -> + "flush state machine found distinct counterexamples (shortest schedule per cause): " + + distinctFindings.values()); + } + + @Test + void generatedSaturationSequencesReportExplicitDrops() { + for (List commands : generatedPublishSequences(5)) { + Map acceptedByLane = new EnumMap<>(Command.class); + acceptedByLane.put(Command.PUBLISH_PRIMARY, 0); + acceptedByLane.put(Command.PUBLISH_SECONDARY, 0); + + try (TraceProcessingWorker worker = + new TraceProcessingWorker( + 2, + mock(HealthMetrics.class), + mock(PayloadDispatcher.class), + () -> false, + FAST_LANE, + 100, + SECONDS, + null)) { + for (Command command : commands) { + int accepted = acceptedByLane.get(command); + PublishResult expected = expectedSaturationResult(command, accepted); + PublishResult actual = + publish(worker, command, Collections.singletonList(mock(DDSpan.class))); + assertEquals(expected, actual, "schedule=" + commands); + if (actual == ENQUEUED_FOR_SERIALIZATION) { + acceptedByLane.put(command, accepted + 1); + } + } + } + } } - private static Counterexample execute(List commands) throws Exception { + private static List executeFlushStateMachine(List commands) + throws Exception { Object observationsLock = new Object(); Map, Integer> traceIds = new IdentityHashMap<>(); Set serialized = new LinkedHashSet<>(); - AtomicReference> serializedAtFlush = new AtomicReference<>(); + List> serializedAtFlush = new ArrayList<>(); PayloadDispatcher dispatcher = mock(PayloadDispatcher.class); doAnswer( invocation -> { @@ -73,7 +116,7 @@ private static Counterexample execute(List commands) throws Exception { doAnswer( invocation -> { synchronized (observationsLock) { - serializedAtFlush.set(new LinkedHashSet<>(serialized)); + serializedAtFlush.add(new LinkedHashSet<>(serialized)); } return null; }) @@ -81,7 +124,12 @@ private static Counterexample execute(List commands) throws Exception { .flush(); Set accepted = new LinkedHashSet<>(); - ExecutorService flushExecutor = Executors.newSingleThreadExecutor(); + Set primaryIds = new LinkedHashSet<>(); + Set secondaryIds = new LinkedHashSet<>(); + List> expectedAtFlush = new ArrayList<>(); + List> flushes = new ArrayList<>(); + ExecutorService flushExecutor = Executors.newCachedThreadPool(); + List counterexamples = new ArrayList<>(); try (TraceProcessingWorker worker = new TraceProcessingWorker( 16, @@ -92,38 +140,114 @@ private static Counterexample execute(List commands) throws Exception { 100, SECONDS, null)) { - int primaryEvents = 0; - for (int id = 0; id < commands.size(); id++) { - Command command = commands.get(id); - List trace = Collections.singletonList(mock(DDSpan.class)); - traceIds.put(trace, id); - int priority; - if (command == Command.PUBLISH_PRIMARY) { - priority = USER_KEEP; - primaryEvents++; - } else { - priority = USER_DROP; + int primaryQueueSize = 0; + int nextTraceId = 0; + for (Command command : commands) { + if (command == Command.FLUSH) { + expectedAtFlush.add(new LinkedHashSet<>(accepted)); + flushes.add(flushExecutor.submit(() -> worker.flush(3, SECONDS))); + primaryQueueSize++; + awaitPrimaryQueueSize(worker, primaryQueueSize); + continue; } - PublishResult result = worker.publish(trace.get(0), priority, trace); + int id = nextTraceId++; + List trace = Collections.singletonList(mock(DDSpan.class)); + traceIds.put(trace, id); + PublishResult result = publish(worker, command, trace); if (result == ENQUEUED_FOR_SERIALIZATION) { accepted.add(id); + if (command == Command.PUBLISH_PRIMARY) { + primaryIds.add(id); + primaryQueueSize++; + } else { + secondaryIds.add(id); + } } } - Future flush = flushExecutor.submit(() -> worker.flush(3, SECONDS)); - awaitPrimaryQueueSize(worker, primaryEvents + 1); + assertFalse(flushes.isEmpty(), "generated schedule must contain a flush"); worker.start(); + for (int barrier = 0; barrier < flushes.size(); barrier++) { + if (!flushes.get(barrier).get(3, SECONDS)) { + counterexamples.add( + new Counterexample( + Finding.FLUSH_DID_NOT_COMPLETE, + commands, + barrier, + expectedAtFlush.get(barrier), + Collections.emptySet())); + } + } - assertTrue(flush.get(3, SECONDS), "generated flush did not complete"); - Set observed = serializedAtFlush.get(); - if (!accepted.equals(observed)) { - return new Counterexample(commands, accepted, observed); + List> observations; + synchronized (observationsLock) { + observations = new ArrayList<>(serializedAtFlush); + } + for (int barrier = 0; barrier < expectedAtFlush.size(); barrier++) { + Set expected = expectedAtFlush.get(barrier); + if (barrier >= observations.size()) { + counterexamples.add( + new Counterexample( + Finding.FLUSH_OBSERVATION_MISSING, + commands, + barrier, + expected, + Collections.emptySet())); + continue; + } + Set observed = observations.get(barrier); + if (!expected.equals(observed)) { + counterexamples.add( + new Counterexample( + classify(expected, observed, primaryIds, secondaryIds), + commands, + barrier, + expected, + observed)); + } } - return null; } finally { flushExecutor.shutdownNow(); } + return counterexamples; + } + + private static Finding classify( + Set expected, + Set observed, + Set primaryIds, + Set secondaryIds) { + Set unexpected = new LinkedHashSet<>(observed); + unexpected.removeAll(expected); + if (!unexpected.isEmpty()) { + return Finding.WORK_CROSSES_FLUSH_FROM_THE_FUTURE; + } + + Set missing = new LinkedHashSet<>(expected); + missing.removeAll(observed); + if (secondaryIds.containsAll(missing)) { + return Finding.SECONDARY_QUEUE_BYPASSES_FLUSH; + } + if (primaryIds.containsAll(missing)) { + return Finding.PRIMARY_QUEUE_BYPASSES_FLUSH; + } + return Finding.MIXED_ACCEPTED_WORK_BYPASSES_FLUSH; + } + + private static PublishResult publish( + TraceProcessingWorker worker, Command command, List trace) { + int priority = command == Command.PUBLISH_PRIMARY ? USER_KEEP : USER_DROP; + return worker.publish(trace.get(0), priority, trace); + } + + private static PublishResult expectedSaturationResult(Command command, int accepted) { + if (accepted < 2) { + return ENQUEUED_FOR_SERIALIZATION; + } + return command == Command.PUBLISH_PRIMARY + ? DROPPED_BUFFER_OVERFLOW + : DROPPED_BUFFER_OVERFLOW_SAMPLED_OUT; } private static void awaitPrimaryQueueSize(TraceProcessingWorker worker, int expectedSize) { @@ -134,46 +258,69 @@ private static void awaitPrimaryQueueSize(TraceProcessingWorker worker, int expe assertEquals(expectedSize, worker.getPrimaryQueue().size(), "flush marker was not enqueued"); } - private static List> generatedPublishPrefixes(int maximumLength) { + private static List> generatedCommandSequences(int maximumLength) { + List> sequences = generateSequences(maximumLength, EnumSet.allOf(Command.class)); + sequences.removeIf(sequence -> !sequence.contains(Command.FLUSH)); + return sequences; + } + + private static List> generatedPublishSequences(int maximumLength) { + return generateSequences( + maximumLength, EnumSet.of(Command.PUBLISH_PRIMARY, Command.PUBLISH_SECONDARY)); + } + + private static List> generateSequences(int maximumLength, Set alphabet) { List> sequences = new ArrayList<>(); for (int length = 1; length <= maximumLength; length++) { - generateSequences(length, new ArrayList<>(), sequences); + generateSequences(length, alphabet, new ArrayList<>(), sequences); } return sequences; } private static void generateSequences( - int remaining, List prefix, List> sequences) { + int remaining, Set alphabet, List prefix, List> sequences) { if (remaining == 0) { sequences.add(new ArrayList<>(prefix)); return; } - for (Command command : EnumSet.allOf(Command.class)) { + for (Command command : alphabet) { prefix.add(command); - generateSequences(remaining - 1, prefix, sequences); + generateSequences(remaining - 1, alphabet, prefix, sequences); prefix.remove(prefix.size() - 1); } } private static final class Counterexample { + private final Finding finding; private final List commands; - private final Set accepted; - private final Set serializedAtFlush; + private final int barrier; + private final Set expected; + private final Set observed; private Counterexample( - List commands, Set accepted, Set serializedAtFlush) { + Finding finding, + List commands, + int barrier, + Set expected, + Set observed) { + this.finding = finding; this.commands = new ArrayList<>(commands); - this.accepted = new LinkedHashSet<>(accepted); - this.serializedAtFlush = new LinkedHashSet<>(serializedAtFlush); + this.barrier = barrier; + this.expected = new LinkedHashSet<>(expected); + this.observed = new LinkedHashSet<>(observed); } @Override public String toString() { - return commands - + " -> FLUSH; accepted=" - + accepted - + ", serializedAtFlush=" - + serializedAtFlush; + return finding + + ": " + + commands + + "; barrier=" + + barrier + + ", expected=" + + expected + + ", observed=" + + observed; } } } diff --git a/formal-methods/tier1/CONTRACTS.md b/formal-methods/tier1/CONTRACTS.md index bb0b3e89b9b..1bab4a81fb5 100644 --- a/formal-methods/tier1/CONTRACTS.md +++ b/formal-methods/tier1/CONTRACTS.md @@ -10,9 +10,11 @@ alive. | Pending traces | A late child retains a writer after flush | `FlushDrainOwnership` | Known violation | `PendingTraceBufferTest.flushDrainClearsEnqueuedStateBeforeWriting` | | Buffer commands | A waiting command remains owned or acknowledged | `FlushDumpMarker` | Known violation | `PendingTraceBufferTest.flushDrainDoesNotConsumeTracerDumpMarker` | | Writer pipeline | Flush covers accepted upstream sampling work | `SpanSamplingFlush` | Known violation | `TraceProcessingWorkerTest.flushWaitsForInFlightSingleSpanSampling` | -| Writer pipeline | Flush covers accepted secondary-queue work | Generated reference model | Known violation | `TraceProcessingWorkerStateMachineTest.generatedCommandSequencesRespectFlushBarrier` | +| Writer pipeline | Flush covers accepted secondary-queue work | Generated reference model | Known violation | `TraceProcessingWorkerStateMachineTest.generatedCommandSequencesRespectFlushBarriers` | | Context | Continuations finalize once and only when quiescent | `ContinuationLifecycle` | Passes bounded model | Existing context lifecycle tests | | Remote config | Cached targets were distributed to every healthy listener | `ListenerDistribution` | Known violation | `ProductStateSpecification.listenerFailureDoesNotPreventLaterListenersFromReceivingConfiguration` | +| Remote config | Removal reaches every healthy listener and remains retryable | Generated listener state machine | Known violation | `ProductStateListenerStateMachineTest.generatedListenerFailuresRemainRetryableForHealthyListeners` | +| Remote config | Commit reaches every healthy listener and remains retryable | Generated listener state machine | Known violation | `ProductStateListenerStateMachineTest.generatedListenerFailuresRemainRetryableForHealthyListeners` | | Telemetry | Accepted events remain queued, retry-buffered, delivered, or policy-dropped | `TelemetryRetryOwnership` | Passes bounded model | Existing `TelemetryServiceTest` retry tests | | Data streams | `close()` cannot return with its worker alive | `DataStreamsClose` | Known violation | `DefaultDataStreamsMonitoringTest.closeStopsWorkerWhenInboxIsFull` | diff --git a/formal-methods/tier1/STATE_MACHINE_TESTING.md b/formal-methods/tier1/STATE_MACHINE_TESTING.md index 37cf7a199a0..43e730e10ae 100644 --- a/formal-methods/tier1/STATE_MACHINE_TESTING.md +++ b/formal-methods/tier1/STATE_MACHINE_TESTING.md @@ -5,10 +5,11 @@ generated command sequences against the Java implementation. A small reference model records every trace accepted before a flush. The implementation must have serialized the same set when the flush is acknowledged. -`TraceProcessingWorkerStateMachineTest` generates publish prefixes in increasing -length and lexicographic order. For each prefix it deliberately queues the flush -marker before starting the serializer. This removes timing from the schedule and -makes the first mismatch the shortest reproducible counterexample. +`TraceProcessingWorkerStateMachineTest` generates command sequences in increasing +length and lexicographic order. For each sequence it deliberately queues flush +markers before starting the serializer. This removes timing from the schedule. +It explores the full bound, groups failures by causal signature, and retains the +first (therefore shortest) schedule for each distinct problem. The current minimal counterexample is: @@ -25,4 +26,22 @@ Future state machines should preserve three properties: 1. generated schedules are deterministic and bounded; 2. the oracle is an independent reference model, not internal queue state; -3. failures report a replayable, minimal command sequence. +3. failures report a replayable, minimal command sequence without stopping + exploration after a previously known defect. + +The saturation state machine is independent of flush ordering. It fills the +primary and secondary lanes without starting consumers and checks that every +rejected publish is reported as an explicit lane-specific terminal drop. + +## Remote-config listener fan-out + +`ProductStateListenerStateMachineTest` injects a failure into each listener +phase (`accept`, `remove`, and `commit`) ahead of a healthy listener. It then +replays the same remote-config state. The reference model requires the healthy +listener to observe the complete phase and requires incomplete delivery to +remain retryable. + +All three injected phases currently violate the model. In each case the healthy +listener misses the failed phase and replay reports no change, so delivery is +never retried. The `remove` and `commit` failures are distinct additions to the +previously known `accept` failure. diff --git a/remote-config/remote-config-core/src/test/java/datadog/remoteconfig/state/ProductStateListenerStateMachineTest.java b/remote-config/remote-config-core/src/test/java/datadog/remoteconfig/state/ProductStateListenerStateMachineTest.java new file mode 100644 index 00000000000..f77c38687f4 --- /dev/null +++ b/remote-config/remote-config-core/src/test/java/datadog/remoteconfig/state/ProductStateListenerStateMachineTest.java @@ -0,0 +1,178 @@ +package datadog.remoteconfig.state; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.Collections.emptyList; +import static java.util.Collections.singletonList; +import static java.util.Collections.singletonMap; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import datadog.remoteconfig.PollingRateHinter; +import datadog.remoteconfig.Product; +import datadog.remoteconfig.tuf.RemoteConfigResponse; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.EnumMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class ProductStateListenerStateMachineTest { + + private static final String CONFIG_PATH = "org/ASM_DATA/config1/foo"; + private static final ParsedConfigKey CONFIG_KEY = ParsedConfigKey.parse(CONFIG_PATH); + + private enum FailurePhase { + ACCEPT, + REMOVE, + COMMIT + } + + @Test + void generatedListenerFailuresRemainRetryableForHealthyListeners() { + Map distinctFindings = new EnumMap<>(FailurePhase.class); + + for (FailurePhase failurePhase : FailurePhase.values()) { + Counterexample counterexample = execute(failurePhase); + if (counterexample != null) { + distinctFindings.put(failurePhase, counterexample); + } + } + + assertTrue( + distinctFindings.isEmpty(), + () -> + "listener state machine found fan-out failures that replay cannot repair: " + + distinctFindings.values()); + } + + private static Counterexample execute(FailurePhase failurePhase) { + PollingRateHinter hinter = mock(PollingRateHinter.class); + ProductState productState = new ProductState(Product.ASM_DATA); + productState.addProductListener(new FailingListener(failurePhase)); + RecordingListener healthyListener = new RecordingListener(); + productState.addProductListener(healthyListener); + + RemoteConfigResponse present = responseWithConfig(); + List expected; + boolean replayChanged; + if (failurePhase == FailurePhase.REMOVE) { + productState.apply(present, singletonList(CONFIG_KEY), hinter); + healthyListener.operations.clear(); + + RemoteConfigResponse absent = responseWithoutConfig(); + productState.apply(absent, emptyList(), hinter); + replayChanged = productState.apply(absent, emptyList(), hinter); + expected = Arrays.asList("remove", "commit"); + } else { + productState.apply(present, singletonList(CONFIG_KEY), hinter); + replayChanged = productState.apply(present, singletonList(CONFIG_KEY), hinter); + expected = Arrays.asList("accept", "commit"); + } + + if (!expected.equals(healthyListener.operations)) { + return new Counterexample(failurePhase, expected, healthyListener.operations, replayChanged); + } + return null; + } + + private static RemoteConfigResponse responseWithConfig() { + RemoteConfigResponse response = mock(RemoteConfigResponse.class); + RemoteConfigResponse.Targets.ConfigTarget target = + new RemoteConfigResponse.Targets.ConfigTarget(); + target.hashes = singletonMap("sha256", "hash1"); + target.length = 8; + RemoteConfigResponse.Targets.ConfigTarget.ConfigTargetCustom custom = + new RemoteConfigResponse.Targets.ConfigTarget.ConfigTargetCustom(); + custom.version = 1; + target.custom = custom; + when(response.getTarget(CONFIG_PATH)).thenReturn(target); + when(response.getFileContents(CONFIG_PATH)).thenReturn("content1".getBytes(UTF_8)); + return response; + } + + private static RemoteConfigResponse responseWithoutConfig() { + RemoteConfigResponse response = mock(RemoteConfigResponse.class); + when(response.getTarget(any())).thenReturn(null); + return response; + } + + private static final class FailingListener implements ProductListener { + private final FailurePhase failurePhase; + + private FailingListener(FailurePhase failurePhase) { + this.failurePhase = failurePhase; + } + + @Override + public void accept(ConfigKey configKey, byte[] content, PollingRateHinter hinter) { + failAt(FailurePhase.ACCEPT); + } + + @Override + public void remove(ConfigKey configKey, PollingRateHinter hinter) { + failAt(FailurePhase.REMOVE); + } + + @Override + public void commit(PollingRateHinter hinter) { + failAt(FailurePhase.COMMIT); + } + + private void failAt(FailurePhase currentPhase) { + if (failurePhase == currentPhase) { + throw new IllegalStateException("injected " + currentPhase + " failure"); + } + } + } + + private static final class RecordingListener implements ProductListener { + private final List operations = new ArrayList<>(); + + @Override + public void accept(ConfigKey configKey, byte[] content, PollingRateHinter hinter) { + operations.add("accept"); + } + + @Override + public void remove(ConfigKey configKey, PollingRateHinter hinter) { + operations.add("remove"); + } + + @Override + public void commit(PollingRateHinter hinter) { + operations.add("commit"); + } + } + + private static final class Counterexample { + private final FailurePhase failurePhase; + private final List expected; + private final List observed; + private final boolean replayChanged; + + private Counterexample( + FailurePhase failurePhase, + List expected, + List observed, + boolean replayChanged) { + this.failurePhase = failurePhase; + this.expected = new ArrayList<>(expected); + this.observed = new ArrayList<>(observed); + this.replayChanged = replayChanged; + } + + @Override + public String toString() { + return failurePhase + + ": expected=" + + expected + + ", observed=" + + observed + + ", replayChanged=" + + replayChanged; + } + } +} From c70d6fb42e3d4a09de9c404b846fdbd5eb51e0d0 Mon Sep 17 00:00:00 2001 From: Charles Cote Date: Sat, 31 Oct 2026 18:59:52 +0100 Subject: [PATCH 6/7] Add telemetry state machine and JCStress probes --- .gitignore | 2 + dd-trace-core/build.gradle | 27 +++ dd-trace-core/gradle.lockfile | 61 ++--- ...eContinuationMultipleResumeStressTest.java | 66 +++++ ...peContinuationReleaseResumeStressTest.java | 57 +++++ formal-methods/tier1/CONTRACTS.md | 2 + formal-methods/tier1/JCSTRESS.md | 22 ++ formal-methods/tier1/STATE_MACHINE_TESTING.md | 17 ++ .../TelemetryServiceStateMachineTest.java | 225 ++++++++++++++++++ 9 files changed, 450 insertions(+), 29 deletions(-) create mode 100644 dd-trace-core/src/jcstress/java/datadog/trace/core/scopemanager/ScopeContinuationMultipleResumeStressTest.java create mode 100644 dd-trace-core/src/jcstress/java/datadog/trace/core/scopemanager/ScopeContinuationReleaseResumeStressTest.java create mode 100644 formal-methods/tier1/JCSTRESS.md create mode 100644 telemetry/src/test/java/datadog/telemetry/TelemetryServiceStateMachineTest.java diff --git a/.gitignore b/.gitignore index 54ae092deba..74b07801b6d 100644 --- a/.gitignore +++ b/.gitignore @@ -76,6 +76,8 @@ replay_pid* **/src/*/generated/* !dd-java-agent/benchmark/releases/*.jar **/errors/*.log +**/jcstress-results-*.bin.gz +/dd-trace-core/results/ # Magic for local JMC built /vendor/jmc-libs diff --git a/dd-trace-core/build.gradle b/dd-trace-core/build.gradle index f1b42962a50..4ac96e17d2f 100644 --- a/dd-trace-core/build.gradle +++ b/dd-trace-core/build.gradle @@ -69,6 +69,30 @@ excludedClassesCoverage += [ addTestSuite('traceAgentTest') +sourceSets { + jcstress { + java.srcDir 'src/jcstress/java' + compileClasspath += sourceSets.main.output + runtimeClasspath += output + compileClasspath + } +} + +configurations { + jcstressImplementation.extendsFrom implementation + jcstressRuntimeClasspath { + resolutionStrategy.force 'com.datadoghq:dd-instrument-java:0.0.5' + } +} + +tasks.register('jcstress', JavaExec) { + group = 'verification' + description = 'Runs Tier-1 JCStress concurrency probes.' + dependsOn tasks.named('jcstressClasses') + classpath = sourceSets.jcstress.runtimeClasspath + mainClass = 'org.openjdk.jcstress.Main' + args '-t', 'datadog.trace.core.scopemanager.*', '-f', '1', '-iters', '5', '-time', '200' +} + tasks.named('forkedTest', Test) { // Needed for FootprintForkedTest on Java 17 jvmArgs += ['-Djol.magicFieldOffset=true'] @@ -118,6 +142,9 @@ dependencies { testImplementation group: 'org.msgpack', name: 'jackson-dataformat-msgpack', version: '0.8.20' testImplementation group: 'org.openjdk.jol', name: 'jol-core', version: '0.17' testImplementation group: 'commons-codec', name: 'commons-codec', version: '1.3' + + jcstressImplementation group: 'org.openjdk.jcstress', name: 'jcstress-core', version: '0.16' + jcstressAnnotationProcessor group: 'org.openjdk.jcstress', name: 'jcstress-core', version: '0.16' testImplementation group: 'com.amazonaws', name: 'aws-lambda-java-events', version:'3.11.0' testImplementation group: 'com.google.protobuf', name: 'protobuf-java', version: '3.14.0' testImplementation libs.jnr.unixsocket diff --git a/dd-trace-core/gradle.lockfile b/dd-trace-core/gradle.lockfile index 6c8603df7d4..058c5086125 100644 --- a/dd-trace-core/gradle.lockfile +++ b/dd-trace-core/gradle.lockfile @@ -2,18 +2,18 @@ # Manual edits can break the build and are not advised. # This file is expected to be part of source control. # To regenerate this file, run: ./gradlew :dd-trace-core:dependencies --write-locks -cafe.cryptography:curve25519-elisabeth:0.1.0=jmhRuntimeClasspath,runtimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath -cafe.cryptography:ed25519-elisabeth:0.1.0=jmhRuntimeClasspath,runtimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath +cafe.cryptography:curve25519-elisabeth:0.1.0=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath +cafe.cryptography:ed25519-elisabeth:0.1.0=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath ch.qos.logback:logback-classic:1.2.13=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath ch.qos.logback:logback-core:1.2.13=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.amazonaws:aws-lambda-java-events:3.11.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.blogspot.mydailyjava:weak-lock-free:0.17=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.datadoghq.okhttp3:okhttp:3.12.15=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.datadoghq.okio:okio:1.17.6=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.datadoghq:dd-instrument-java:0.0.5=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.datadoghq:dd-javac-plugin-client:0.2.2=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.datadoghq:java-dogstatsd-client:4.4.5=jmhRuntimeClasspath,runtimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath -com.datadoghq:sketches-java:0.8.3=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath +com.datadoghq.okhttp3:okhttp:3.12.15=compileClasspath,jcstressCompileClasspath,jcstressRuntimeClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.datadoghq.okio:okio:1.17.6=compileClasspath,jcstressCompileClasspath,jcstressRuntimeClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.datadoghq:dd-instrument-java:0.0.5=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.datadoghq:dd-javac-plugin-client:0.2.2=compileClasspath,jcstressCompileClasspath,jcstressRuntimeClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.datadoghq:java-dogstatsd-client:4.4.5=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath +com.datadoghq:sketches-java:0.8.3=compileClasspath,jcstressRuntimeClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath com.fasterxml.jackson.core:jackson-annotations:2.10.3=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.fasterxml.jackson.core:jackson-core:2.9.9=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.fasterxml.jackson.core:jackson-databind:2.9.9.3=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath @@ -21,14 +21,14 @@ com.github.docker-java:docker-java-api:3.4.2=jmhRuntimeClasspath,testCompileClas com.github.docker-java:docker-java-transport-zerodep:3.4.2=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.github.docker-java:docker-java-transport:3.4.2=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.github.javaparser:javaparser-core:3.25.6=codenarc -com.github.jnr:jffi:1.3.15=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.github.jnr:jnr-a64asm:1.0.0=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.github.jnr:jnr-constants:0.10.4=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.github.jnr:jnr-enxio:0.32.20=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.github.jnr:jnr-ffi:2.2.19=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.github.jnr:jnr-posix:3.1.22=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.github.jnr:jnr-unixsocket:0.38.25=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.github.jnr:jnr-x86asm:1.0.2=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.github.jnr:jffi:1.3.15=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.github.jnr:jnr-a64asm:1.0.0=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.github.jnr:jnr-constants:0.10.4=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.github.jnr:jnr-enxio:0.32.20=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.github.jnr:jnr-ffi:2.2.19=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.github.jnr:jnr-posix:3.1.22=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.github.jnr:jnr-unixsocket:0.38.25=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.github.jnr:jnr-x86asm:1.0.2=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.github.spotbugs:spotbugs-annotations:4.10.4=compileClasspath,jmhCompileClasspath,spotbugs com.github.spotbugs:spotbugs:4.10.4=spotbugs com.github.stephenc.jcip:jcip-annotations:1.0-1=spotbugs @@ -48,12 +48,12 @@ com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava=jmhR com.google.j2objc:j2objc-annotations:2.8=testAnnotationProcessor,traceAgentTestAnnotationProcessor com.google.j2objc:j2objc-annotations:3.1=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.google.protobuf:protobuf-java:3.14.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.google.re2j:re2j:1.8=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.squareup.moshi:moshi:1.11.0=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.google.re2j:re2j:1.8=compileClasspath,jcstressCompileClasspath,jcstressRuntimeClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.squareup.moshi:moshi:1.11.0=compileClasspath,jcstressCompileClasspath,jcstressRuntimeClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.squareup.okhttp3:logging-interceptor:3.12.12=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.squareup.okhttp3:mockwebserver:3.12.12=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.squareup.okhttp3:okhttp:3.12.12=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -com.squareup.okio:okio:1.17.5=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +com.squareup.okio:okio:1.17.5=compileClasspath,jcstressCompileClasspath,jcstressRuntimeClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath com.thoughtworks.qdox:qdox:1.12.1=codenarc commons-codec:commons-codec:1.3=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath commons-fileupload:commons-fileupload:1.5=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath @@ -68,8 +68,10 @@ joda-time:joda-time:2.6=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClas junit:junit:4.13.2=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath net.bytebuddy:byte-buddy-agent:1.18.12=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath net.bytebuddy:byte-buddy:1.18.12=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -net.java.dev.jna:jna-platform:5.8.0=jmhRuntimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath +net.java.dev.jna:jna-platform:5.8.0=jcstressAnnotationProcessor,jcstressCompileClasspath,jcstressRuntimeClasspath,jmhRuntimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath net.java.dev.jna:jna:5.13.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +net.java.dev.jna:jna:5.8.0=jcstressAnnotationProcessor,jcstressCompileClasspath,jcstressRuntimeClasspath +net.sf.jopt-simple:jopt-simple:4.6=jcstressAnnotationProcessor,jcstressCompileClasspath,jcstressRuntimeClasspath net.sf.jopt-simple:jopt-simple:5.0.4=jmh,jmhCompileClasspath,jmhRuntimeClasspath net.sf.saxon:Saxon-HE:12.10=spotbugs org.apache.ant:ant-antlr:1.10.14=codenarc @@ -101,8 +103,8 @@ org.jacoco:org.jacoco.agent:0.8.15=jacocoAgent,jacocoAnt org.jacoco:org.jacoco.ant:0.8.15=jacocoAnt org.jacoco:org.jacoco.core:0.8.15=jacocoAnt org.jacoco:org.jacoco.report:0.8.15=jacocoAnt -org.jctools:jctools-core-jdk11:4.0.6=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -org.jctools:jctools-core:4.0.6=compileClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +org.jctools:jctools-core-jdk11:4.0.6=compileClasspath,jcstressCompileClasspath,jcstressRuntimeClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +org.jctools:jctools-core:4.0.6=compileClasspath,jcstressCompileClasspath,jcstressRuntimeClasspath,jmhCompileClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.jetbrains:annotations:17.0.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.jspecify:jspecify:1.0.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.junit.jupiter:junit-jupiter-api:5.14.1=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath @@ -121,6 +123,7 @@ org.mockito:mockito-junit-jupiter:4.4.0=jmhRuntimeClasspath,testCompileClasspath org.msgpack:jackson-dataformat-msgpack:0.8.20=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.msgpack:msgpack-core:0.8.20=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.objenesis:objenesis:3.3=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +org.openjdk.jcstress:jcstress-core:0.16=jcstressAnnotationProcessor,jcstressCompileClasspath,jcstressRuntimeClasspath org.openjdk.jmh:jmh-core:1.37=jmh,jmhCompileClasspath,jmhRuntimeClasspath org.openjdk.jmh:jmh-generator-asm:1.37=jmh,jmhCompileClasspath,jmhRuntimeClasspath org.openjdk.jmh:jmh-generator-bytecode:1.37=jmh,jmhCompileClasspath,jmhRuntimeClasspath @@ -128,26 +131,26 @@ org.openjdk.jmh:jmh-generator-reflection:1.37=jmh,jmhCompileClasspath,jmhRuntime org.openjdk.jol:jol-core:0.17=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.opentest4j:opentest4j:1.3.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.ow2.asm:asm-analysis:9.10.1=spotbugs -org.ow2.asm:asm-analysis:9.7.1=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +org.ow2.asm:asm-analysis:9.7.1=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.ow2.asm:asm-commons:9.10.1=jacocoAnt,spotbugs -org.ow2.asm:asm-commons:9.7.1=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +org.ow2.asm:asm-commons:9.7.1=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.ow2.asm:asm-tree:9.10.1=jacocoAnt,spotbugs -org.ow2.asm:asm-tree:9.7.1=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +org.ow2.asm:asm-tree:9.7.1=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.ow2.asm:asm-util:9.10.1=spotbugs -org.ow2.asm:asm-util:9.7.1=jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath +org.ow2.asm:asm-util:9.7.1=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.ow2.asm:asm:9.0=jmh,jmhCompileClasspath org.ow2.asm:asm:9.10.1=jacocoAnt,jmhRuntimeClasspath,spotbugs,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -org.ow2.asm:asm:9.7.1=runtimeClasspath +org.ow2.asm:asm:9.7.1=jcstressRuntimeClasspath,runtimeClasspath org.rnorth.duct-tape:duct-tape:1.0.8=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.slf4j:jcl-over-slf4j:1.7.30=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.slf4j:jul-to-slf4j:1.7.30=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.slf4j:log4j-over-slf4j:1.7.30=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath -org.slf4j:slf4j-api:1.7.30=compileClasspath,jmhCompileClasspath,runtimeClasspath +org.slf4j:slf4j-api:1.7.30=compileClasspath,jcstressCompileClasspath,jcstressRuntimeClasspath,jmhCompileClasspath,runtimeClasspath org.slf4j:slf4j-api:1.7.36=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.slf4j:slf4j-api:2.0.17=spotbugsSlf4j org.slf4j:slf4j-api:2.0.18=spotbugs org.slf4j:slf4j-simple:2.0.17=spotbugsSlf4j -org.snakeyaml:snakeyaml-engine:2.9=jmhRuntimeClasspath,runtimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath +org.snakeyaml:snakeyaml-engine:2.9=jcstressRuntimeClasspath,jmhRuntimeClasspath,runtimeClasspath,testRuntimeClasspath,traceAgentTestRuntimeClasspath org.spockframework:spock-bom:2.4-groovy-3.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.spockframework:spock-core:2.4-groovy-3.0=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath org.tabletest:tabletest-junit:1.2.2=jmhRuntimeClasspath,testCompileClasspath,testRuntimeClasspath,traceAgentTestCompileClasspath,traceAgentTestRuntimeClasspath diff --git a/dd-trace-core/src/jcstress/java/datadog/trace/core/scopemanager/ScopeContinuationMultipleResumeStressTest.java b/dd-trace-core/src/jcstress/java/datadog/trace/core/scopemanager/ScopeContinuationMultipleResumeStressTest.java new file mode 100644 index 00000000000..3fbbd8147a6 --- /dev/null +++ b/dd-trace-core/src/jcstress/java/datadog/trace/core/scopemanager/ScopeContinuationMultipleResumeStressTest.java @@ -0,0 +1,66 @@ +package datadog.trace.core.scopemanager; + +import static org.openjdk.jcstress.annotations.Expect.ACCEPTABLE; +import static org.openjdk.jcstress.annotations.Expect.FORBIDDEN; + +import datadog.context.Context; +import datadog.context.ContextContinuation; +import datadog.context.ContextScope; +import datadog.trace.bootstrap.instrumentation.api.AgentTraceCollector; +import java.util.concurrent.atomic.AtomicInteger; +import org.openjdk.jcstress.annotations.Actor; +import org.openjdk.jcstress.annotations.Arbiter; +import org.openjdk.jcstress.annotations.JCStressTest; +import org.openjdk.jcstress.annotations.Outcome; +import org.openjdk.jcstress.annotations.State; +import org.openjdk.jcstress.infra.results.I_Result; + +@JCStressTest +@Outcome(id = "1", expect = ACCEPTABLE, desc = "All resumed scopes converge on one finalization.") +@Outcome(expect = FORBIDDEN, desc = "A resumed scope leaked or finalized the continuation twice.") +@State +public class ScopeContinuationMultipleResumeStressTest { + private final RecordingCollector collector = new RecordingCollector(); + private final ScopeContinuation continuation = + new ScopeContinuation( + new ContinuableScopeManager(0, false), Context.root(), (byte) 0, collector) + .register() + .hold(); + + @Actor + public void firstResume() { + resumeAndClose(); + } + + @Actor + public void secondResume() { + resumeAndClose(); + } + + @Actor + public void release() { + continuation.release(); + } + + @Arbiter + public void observe(I_Result result) { + result.r1 = collector.removals.get(); + } + + private void resumeAndClose() { + ContextScope scope = continuation.resume(); + scope.close(); + } + + private static final class RecordingCollector implements AgentTraceCollector { + private final AtomicInteger removals = new AtomicInteger(); + + @Override + public void registerContinuation(ContextContinuation continuation) {} + + @Override + public void removeContinuation(ContextContinuation continuation) { + removals.incrementAndGet(); + } + } +} diff --git a/dd-trace-core/src/jcstress/java/datadog/trace/core/scopemanager/ScopeContinuationReleaseResumeStressTest.java b/dd-trace-core/src/jcstress/java/datadog/trace/core/scopemanager/ScopeContinuationReleaseResumeStressTest.java new file mode 100644 index 00000000000..d1c1bd45056 --- /dev/null +++ b/dd-trace-core/src/jcstress/java/datadog/trace/core/scopemanager/ScopeContinuationReleaseResumeStressTest.java @@ -0,0 +1,57 @@ +package datadog.trace.core.scopemanager; + +import static org.openjdk.jcstress.annotations.Expect.ACCEPTABLE; +import static org.openjdk.jcstress.annotations.Expect.FORBIDDEN; + +import datadog.context.Context; +import datadog.context.ContextContinuation; +import datadog.context.ContextScope; +import datadog.trace.bootstrap.instrumentation.api.AgentTraceCollector; +import java.util.concurrent.atomic.AtomicInteger; +import org.openjdk.jcstress.annotations.Actor; +import org.openjdk.jcstress.annotations.Arbiter; +import org.openjdk.jcstress.annotations.JCStressTest; +import org.openjdk.jcstress.annotations.Outcome; +import org.openjdk.jcstress.annotations.State; +import org.openjdk.jcstress.infra.results.I_Result; + +@JCStressTest +@Outcome(id = "1", expect = ACCEPTABLE, desc = "The continuation is finalized exactly once.") +@Outcome(expect = FORBIDDEN, desc = "The continuation leaked or was finalized more than once.") +@State +public class ScopeContinuationReleaseResumeStressTest { + private final RecordingCollector collector = new RecordingCollector(); + private final ScopeContinuation continuation = + new ScopeContinuation( + new ContinuableScopeManager(0, false), Context.root(), (byte) 0, collector) + .register() + .hold(); + + @Actor + public void release() { + continuation.release(); + } + + @Actor + public void resumeAndClose() { + ContextScope scope = continuation.resume(); + scope.close(); + } + + @Arbiter + public void observe(I_Result result) { + result.r1 = collector.removals.get(); + } + + private static final class RecordingCollector implements AgentTraceCollector { + private final AtomicInteger removals = new AtomicInteger(); + + @Override + public void registerContinuation(ContextContinuation continuation) {} + + @Override + public void removeContinuation(ContextContinuation continuation) { + removals.incrementAndGet(); + } + } +} diff --git a/formal-methods/tier1/CONTRACTS.md b/formal-methods/tier1/CONTRACTS.md index 1bab4a81fb5..b425ac59ccb 100644 --- a/formal-methods/tier1/CONTRACTS.md +++ b/formal-methods/tier1/CONTRACTS.md @@ -12,10 +12,12 @@ alive. | Writer pipeline | Flush covers accepted upstream sampling work | `SpanSamplingFlush` | Known violation | `TraceProcessingWorkerTest.flushWaitsForInFlightSingleSpanSampling` | | Writer pipeline | Flush covers accepted secondary-queue work | Generated reference model | Known violation | `TraceProcessingWorkerStateMachineTest.generatedCommandSequencesRespectFlushBarriers` | | Context | Continuations finalize once and only when quiescent | `ContinuationLifecycle` | Passes bounded model | Existing context lifecycle tests | +| Context | Concurrent resume/release finalizes exactly once | JCStress | Passes 156 JVM/compiler configurations | `ScopeContinuationReleaseResumeStressTest`, `ScopeContinuationMultipleResumeStressTest` | | Remote config | Cached targets were distributed to every healthy listener | `ListenerDistribution` | Known violation | `ProductStateSpecification.listenerFailureDoesNotPreventLaterListenersFromReceivingConfiguration` | | Remote config | Removal reaches every healthy listener and remains retryable | Generated listener state machine | Known violation | `ProductStateListenerStateMachineTest.generatedListenerFailuresRemainRetryableForHealthyListeners` | | Remote config | Commit reaches every healthy listener and remains retryable | Generated listener state machine | Known violation | `ProductStateListenerStateMachineTest.generatedListenerFailuresRemainRetryableForHealthyListeners` | | Telemetry | Accepted events remain queued, retry-buffered, delivered, or policy-dropped | `TelemetryRetryOwnership` | Passes bounded model | Existing `TelemetryServiceTest` retry tests | +| Telemetry | An empty retry cannot starve newly queued events | Generated retry state machine | Known violation | `TelemetryServiceStateMachineTest.generatedRetrySequencesPreserveAcceptedEventOwnership` | | Data streams | `close()` cannot return with its worker alive | `DataStreamsClose` | Known violation | `DefaultDataStreamsMonitoringTest.closeStopsWorkerWhenInboxIsFull` | The writer pipeline also has a generated Java state-machine harness. See diff --git a/formal-methods/tier1/JCSTRESS.md b/formal-methods/tier1/JCSTRESS.md new file mode 100644 index 00000000000..c4819f88940 --- /dev/null +++ b/formal-methods/tier1/JCSTRESS.md @@ -0,0 +1,22 @@ +# JCStress concurrency probes + +The `dd-trace-core` module has a dedicated `jcstress` source set and task for +Tier-1 Java Memory Model probes: + +```shell +./gradlew :dd-trace-core:jcstress +``` + +The task uses JCStress 0.16 and currently exercises two `ScopeContinuation` +races: + +- `release` against `resume` plus scope close; +- two concurrent `resume` plus scope close operations against `release`. + +Both probes require exactly one call to +`AgentTraceCollector.removeContinuation`. Zero is a continuation leak and a +value greater than one is double finalization. + +The initial local run covered 156 JVM/compiler configurations. All passed with +no forbidden outcomes. This is positive evidence for the continuation counter +protocol, not proof for unrelated scope-manager behavior. diff --git a/formal-methods/tier1/STATE_MACHINE_TESTING.md b/formal-methods/tier1/STATE_MACHINE_TESTING.md index 43e730e10ae..d04aa0fc5cf 100644 --- a/formal-methods/tier1/STATE_MACHINE_TESTING.md +++ b/formal-methods/tier1/STATE_MACHINE_TESTING.md @@ -45,3 +45,20 @@ All three injected phases currently violate the model. In each case the healthy listener misses the failed phase and replay reports no change, so delivery is never retried. The `remove` and `commit` failures are distinct additions to the previously known `accept` failure. + +## Telemetry retry ownership + +`TelemetryServiceStateMachineTest` generates event additions and successful or +failed sends. It extracts event identifiers from the real serialized request +body and compares them with an independent queued/retry-buffered/delivered/drop +model. + +The current minimal counterexample is: + +```text +SEND_FAILURE -> ADD -> SEND_SUCCESS +``` + +The first empty failed request creates an empty retry buffer. The next send +retries an empty heartbeat instead of taking the newly queued event, then +returns `false` even though that event remains queued. diff --git a/telemetry/src/test/java/datadog/telemetry/TelemetryServiceStateMachineTest.java b/telemetry/src/test/java/datadog/telemetry/TelemetryServiceStateMachineTest.java new file mode 100644 index 00000000000..1a761867376 --- /dev/null +++ b/telemetry/src/test/java/datadog/telemetry/TelemetryServiceStateMachineTest.java @@ -0,0 +1,225 @@ +package datadog.telemetry; + +import static datadog.telemetry.TelemetryClient.Result.FAILURE; +import static datadog.telemetry.TelemetryClient.Result.SUCCESS; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.telemetry.api.Integration; +import java.io.IOException; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.EnumMap; +import java.util.EnumSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Queue; +import java.util.Set; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import okhttp3.Request; +import okio.Buffer; +import org.junit.jupiter.api.Test; + +class TelemetryServiceStateMachineTest { + + private static final Pattern EVENT_ID = Pattern.compile("\\\"name\\\":\\\"event-(\\d+)\\\""); + + private enum Command { + ADD, + SEND_SUCCESS, + SEND_FAILURE + } + + private enum Finding { + REQUEST_OWNERSHIP_MISMATCH, + MORE_DATA_SIGNAL_MISMATCH + } + + @Test + void generatedRetrySequencesPreserveAcceptedEventOwnership() { + Map distinctFindings = new EnumMap<>(Finding.class); + + for (List commands : generatedSequences(5)) { + Counterexample counterexample = execute(commands); + if (counterexample != null) { + distinctFindings.putIfAbsent(counterexample.finding, counterexample); + } + } + + assertTrue( + distinctFindings.isEmpty(), + () -> + "telemetry state machine found distinct counterexamples (shortest schedule per cause): " + + distinctFindings.values()); + } + + private static Counterexample execute(List commands) { + RecordingRouter router = new RecordingRouter(); + TelemetryService service = new TelemetryService(router, 1_000_000, false); + Set queued = new LinkedHashSet<>(); + Set retryBuffered = null; + int nextEventId = 0; + + for (int index = 0; index < commands.size(); index++) { + Command command = commands.get(index); + if (command == Command.ADD) { + int id = nextEventId++; + service.addIntegration(new Integration("event-" + id, true)); + queued.add(id); + continue; + } + + boolean retrying = retryBuffered != null; + Set expectedRequest = new LinkedHashSet<>(retrying ? retryBuffered : queued); + TelemetryClient.Result result = command == Command.SEND_SUCCESS ? SUCCESS : FAILURE; + router.nextResult(result); + boolean actualMoreData = service.sendTelemetryEvents(); + Set actualRequest = router.lastRequestEventIds(); + + if (!expectedRequest.equals(actualRequest)) { + return new Counterexample( + Finding.REQUEST_OWNERSHIP_MISMATCH, + commands, + index, + expectedRequest, + actualRequest, + !queued.isEmpty(), + actualMoreData); + } + + if (retrying) { + retryBuffered = null; + } else { + queued.removeAll(expectedRequest); + if (result == FAILURE && !expectedRequest.isEmpty()) { + retryBuffered = expectedRequest; + } + } + + boolean expectedMoreData = result == SUCCESS && !queued.isEmpty(); + if (expectedMoreData != actualMoreData) { + return new Counterexample( + Finding.MORE_DATA_SIGNAL_MISMATCH, + commands, + index, + expectedRequest, + actualRequest, + expectedMoreData, + actualMoreData); + } + } + return null; + } + + private static List> generatedSequences(int maximumLength) { + List> sequences = new ArrayList<>(); + for (int length = 1; length <= maximumLength; length++) { + generate(length, new ArrayList<>(), sequences); + } + sequences.removeIf( + sequence -> + !sequence.contains(Command.ADD) + || (!sequence.contains(Command.SEND_SUCCESS) + && !sequence.contains(Command.SEND_FAILURE))); + return sequences; + } + + private static void generate(int remaining, List prefix, List> sequences) { + if (remaining == 0) { + sequences.add(new ArrayList<>(prefix)); + return; + } + for (Command command : EnumSet.allOf(Command.class)) { + prefix.add(command); + generate(remaining - 1, prefix, sequences); + prefix.remove(prefix.size() - 1); + } + } + + private static final class RecordingRouter extends TelemetryRouter { + private final Queue results = new ArrayDeque<>(); + private Set lastRequestEventIds; + + private RecordingRouter() { + super(null, null, null, false); + } + + private void nextResult(TelemetryClient.Result result) { + results.add(result); + } + + @Override + public TelemetryClient.Result sendRequest(TelemetryRequest request) { + lastRequestEventIds = eventIds(request); + return results.remove(); + } + + private Set lastRequestEventIds() { + return lastRequestEventIds; + } + + private static Set eventIds(TelemetryRequest telemetryRequest) { + try { + Request.Builder requestBuilder = telemetryRequest.httpRequest(); + requestBuilder.url("https://example.com"); + Request request = requestBuilder.build(); + Buffer buffer = new Buffer(); + request.body().writeTo(buffer); + String body = buffer.readUtf8(); + Set ids = new LinkedHashSet<>(); + Matcher matcher = EVENT_ID.matcher(body); + while (matcher.find()) { + ids.add(Integer.parseInt(matcher.group(1))); + } + return ids; + } catch (IOException e) { + throw new IllegalStateException(e); + } + } + } + + private static final class Counterexample { + private final Finding finding; + private final List commands; + private final int commandIndex; + private final Set expectedRequest; + private final Set actualRequest; + private final boolean expectedMoreData; + private final boolean actualMoreData; + + private Counterexample( + Finding finding, + List commands, + int commandIndex, + Set expectedRequest, + Set actualRequest, + boolean expectedMoreData, + boolean actualMoreData) { + this.finding = finding; + this.commands = new ArrayList<>(commands); + this.commandIndex = commandIndex; + this.expectedRequest = new LinkedHashSet<>(expectedRequest); + this.actualRequest = new LinkedHashSet<>(actualRequest); + this.expectedMoreData = expectedMoreData; + this.actualMoreData = actualMoreData; + } + + @Override + public String toString() { + return finding + + ": " + + commands + + "; commandIndex=" + + commandIndex + + ", expectedRequest=" + + expectedRequest + + ", actualRequest=" + + actualRequest + + ", expectedMoreData=" + + expectedMoreData + + ", actualMoreData=" + + actualMoreData; + } + } +} From 54bea739ba4c1ca49c42c5cd5323d935e418ebb1 Mon Sep 17 00:00:00 2001 From: Charles Cote Date: Sat, 31 Oct 2026 19:11:46 +0100 Subject: [PATCH 7/7] Expose concurrent pending trace flush race --- dd-trace-core/build.gradle | 2 +- ...gTraceBufferConcurrentFlushStressTest.java | 83 +++++++++++++++++++ formal-methods/tier1/CONTRACTS.md | 1 + formal-methods/tier1/JCSTRESS.md | 28 +++++-- 4 files changed, 106 insertions(+), 8 deletions(-) create mode 100644 dd-trace-core/src/jcstress/java/datadog/trace/core/PendingTraceBufferConcurrentFlushStressTest.java diff --git a/dd-trace-core/build.gradle b/dd-trace-core/build.gradle index 4ac96e17d2f..64dc4888ece 100644 --- a/dd-trace-core/build.gradle +++ b/dd-trace-core/build.gradle @@ -90,7 +90,7 @@ tasks.register('jcstress', JavaExec) { dependsOn tasks.named('jcstressClasses') classpath = sourceSets.jcstress.runtimeClasspath mainClass = 'org.openjdk.jcstress.Main' - args '-t', 'datadog.trace.core.scopemanager.*', '-f', '1', '-iters', '5', '-time', '200' + args '-t', 'datadog.trace.core.*', '-f', '1', '-iters', '5', '-time', '200' } tasks.named('forkedTest', Test) { diff --git a/dd-trace-core/src/jcstress/java/datadog/trace/core/PendingTraceBufferConcurrentFlushStressTest.java b/dd-trace-core/src/jcstress/java/datadog/trace/core/PendingTraceBufferConcurrentFlushStressTest.java new file mode 100644 index 00000000000..deb0c19feef --- /dev/null +++ b/dd-trace-core/src/jcstress/java/datadog/trace/core/PendingTraceBufferConcurrentFlushStressTest.java @@ -0,0 +1,83 @@ +package datadog.trace.core; + +import static org.openjdk.jcstress.annotations.Expect.ACCEPTABLE; +import static org.openjdk.jcstress.annotations.Expect.FORBIDDEN; + +import datadog.trace.api.Config; +import datadog.trace.api.time.SystemTimeSource; +import datadog.trace.core.monitor.HealthMetrics; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.openjdk.jcstress.annotations.Actor; +import org.openjdk.jcstress.annotations.Arbiter; +import org.openjdk.jcstress.annotations.JCStressTest; +import org.openjdk.jcstress.annotations.Outcome; +import org.openjdk.jcstress.annotations.State; +import org.openjdk.jcstress.infra.results.I_Result; + +@JCStressTest +@Outcome(id = "1", expect = ACCEPTABLE, desc = "The caller's flush wrote its accepted element.") +@Outcome(id = "0", expect = FORBIDDEN, desc = "A concurrent marker released flush too early.") +@State +public class PendingTraceBufferConcurrentFlushStressTest { + private final PendingTraceBuffer.DelayingPendingTraceBuffer buffer = + new PendingTraceBuffer.DelayingPendingTraceBuffer( + 16, SystemTimeSource.INSTANCE, Config.get(), null, HealthMetrics.NO_OP); + private final RecordingElement element = new RecordingElement(); + + public PendingTraceBufferConcurrentFlushStressTest() { + buffer.getWorker().start(); + } + + @Actor + public void competingFlush() { + buffer.flush(); + } + + @Actor + public void enqueueThenFlush(I_Result result) { + buffer.enqueue(element); + buffer.flush(); + result.r1 = element.writes.get(); + } + + @Arbiter + public void close() { + buffer.close(); + } + + private static final class RecordingElement implements PendingTraceBuffer.Element { + private final AtomicBoolean enqueued = new AtomicBoolean(); + private final AtomicInteger writes = new AtomicInteger(); + + @Override + public long oldestFinishedTime() { + return Long.MAX_VALUE; + } + + @Override + public boolean lastReferencedNanosAgo(long nanos) { + return false; + } + + @Override + public void write() { + writes.incrementAndGet(); + } + + @Override + public DDSpan getRootSpan() { + return null; + } + + @Override + public boolean setEnqueued(boolean newValue) { + return enqueued.compareAndSet(!newValue, newValue); + } + + @Override + public boolean writeOnBufferFull() { + return true; + } + } +} diff --git a/formal-methods/tier1/CONTRACTS.md b/formal-methods/tier1/CONTRACTS.md index b425ac59ccb..ee33d77dcf4 100644 --- a/formal-methods/tier1/CONTRACTS.md +++ b/formal-methods/tier1/CONTRACTS.md @@ -8,6 +8,7 @@ alive. |---|---|---|---|---| | Pending traces | Every live trace has an owner | `TrackerOwnership` | Known violation | `LongRunningTracesTrackerTest.fullTrackerLeavesOverflowTraceForBufferFallback` | | Pending traces | A late child retains a writer after flush | `FlushDrainOwnership` | Known violation | `PendingTraceBufferTest.flushDrainClearsEnqueuedStateBeforeWriting` | +| Pending traces | Each concurrent flush waits for its own accepted work | JCStress | Known violation: 6 forbidden outcomes in 913 samples | `PendingTraceBufferConcurrentFlushStressTest` | | Buffer commands | A waiting command remains owned or acknowledged | `FlushDumpMarker` | Known violation | `PendingTraceBufferTest.flushDrainDoesNotConsumeTracerDumpMarker` | | Writer pipeline | Flush covers accepted upstream sampling work | `SpanSamplingFlush` | Known violation | `TraceProcessingWorkerTest.flushWaitsForInFlightSingleSpanSampling` | | Writer pipeline | Flush covers accepted secondary-queue work | Generated reference model | Known violation | `TraceProcessingWorkerStateMachineTest.generatedCommandSequencesRespectFlushBarriers` | diff --git a/formal-methods/tier1/JCSTRESS.md b/formal-methods/tier1/JCSTRESS.md index c4819f88940..e3cb32e4f1f 100644 --- a/formal-methods/tier1/JCSTRESS.md +++ b/formal-methods/tier1/JCSTRESS.md @@ -7,16 +7,30 @@ Tier-1 Java Memory Model probes: ./gradlew :dd-trace-core:jcstress ``` -The task uses JCStress 0.16 and currently exercises two `ScopeContinuation` -races: +The task uses JCStress 0.16 and currently exercises these races: - `release` against `resume` plus scope close; -- two concurrent `resume` plus scope close operations against `release`. +- two concurrent `resume` plus scope close operations against `release`; +- two concurrent pending-trace flushes, one of which owns an accepted element. -Both probes require exactly one call to +The continuation probes require exactly one call to `AgentTraceCollector.removeContinuation`. Zero is a continuation leak and a value greater than one is double finalization. -The initial local run covered 156 JVM/compiler configurations. All passed with -no forbidden outcomes. This is positive evidence for the continuation counter -protocol, not proof for unrelated scope-manager behavior. +The initial continuation-only run covered 156 JVM/compiler configurations. All +passed with no forbidden outcomes. This is positive evidence for the +continuation counter protocol, not proof for unrelated scope-manager behavior. + +The pending-trace probe found a forbidden outcome in 6 of 913 samples across +five JVM/compiler configurations. One actor enqueues an element and invokes +`flush()` while another actor invokes `flush()` concurrently. The first +processed marker increments the shared `flushCounter`; both callers can observe +that increment and return even though the second caller's accepted element has +not been written. The observed result was: + +```text +writes observed when owning flush returned = 0 +``` + +This is a distinct concurrent-barrier failure: each caller waits for any newer +counter value rather than an acknowledgement tied to its own marker.