From f59e2522a86b008228ab251ed13f00807620c6b0 Mon Sep 17 00:00:00 2001 From: Daniel McCoy Stephenson Date: Tue, 10 Nov 2026 19:53:57 -0700 Subject: [PATCH] Pin coalescing under reports from many threads at once report() runs on the host's threads, but no test called it from more than one. Eight threads now report one event 500 times each, and the first report plus the count close() sends add up to all 4000. Eight threads also open five windows each at the same moment, and every event is sent once and counted once, with nothing dropped below the queue and window limits. Co-Authored-By: Claude Opus 5.5 --- .../stephenson/trace/TraceClientTest.java | 97 +++++++++++++++++++ 1 file changed, 97 insertions(+) diff --git a/src/test/java/software/stephenson/trace/TraceClientTest.java b/src/test/java/software/stephenson/trace/TraceClientTest.java index 15f2545..770a602 100644 --- a/src/test/java/software/stephenson/trace/TraceClientTest.java +++ b/src/test/java/software/stephenson/trace/TraceClientTest.java @@ -29,6 +29,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; +import java.util.function.IntConsumer; import java.util.logging.Handler; import java.util.logging.Level; import java.util.logging.LogRecord; @@ -2334,6 +2335,102 @@ void coalesce_keysOnNameValueAndTagsButNotTagOrder() throws Exception { "an event with a value keeps it; the count is the tag: " + bodies); } + /** Starts {@code threads} threads that each run {@code work} with their index, all at once, and waits for them. */ + private static void runAtOnce(int threads, IntConsumer work) throws InterruptedException { + CountDownLatch start = new CountDownLatch(1); + List started = new ArrayList<>(); + for (int t = 0; t < threads; t++) { + final int index = t; + Thread thread = new Thread(() -> { + try { + start.await(); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + return; + } + work.accept(index); + }); + thread.start(); + started.add(thread); + } + start.countDown(); + for (Thread thread : started) { + thread.join(10_000); + assertFalse(thread.isAlive(), "a reporting thread did not finish"); + } + } + + @Test + void coalesce_reportsFromManyThreadsAtOnceAreEachCountedExactlyOnce() throws Exception { + // Arrange + // report() runs on the host's threads -- a plugin's async tasks + // beside the main thread -- so the window it shares must neither lose + // a repeat nor count one twice. + TraceClient client = TraceClient.builder(baseUrl(), "MyPlugin", "1.2.3").key("k").build(); + Map power = Collections.singletonMap("name", "power"); + int threads = 8; + int each = 500; + + // Act + // The real window, a minute: everything after the first report is + // counted until close() sends the count. + runAtOnce(threads, index -> { + for (int i = 0; i < each; i++) { + client.report("command", null, power); + } + }); + client.close(); + + // Assert + List bodies = received.stream().map(r -> r.body).collect(Collectors.toList()); + assertEquals(2, bodies.size(), "one first report and one count: " + bodies); + assertEquals("{\"application\":\"MyPlugin\",\"name\":\"command\",\"tags\":{\"name\":\"power\",\"version\":\"1.2.3\"}}", + bodies.get(0)); + int repeats = threads * each - 1; + assertEquals("{\"application\":\"MyPlugin\",\"name\":\"command\",\"value\":" + (double) repeats + "," + + "\"tags\":{\"name\":\"power\",\"version\":\"1.2.3\",\"coalesced\":\"" + repeats + "\"}}", + bodies.get(1)); + } + + @Test + void coalesce_distinctEventsFromManyThreadsAtOnceAreEachSentOnceAndCountedOnce() throws Exception { + // Arrange + // Every thread opens windows of its own at the same moment, so the + // windows map and the queue count are changed from all of them at + // once. Well under QUEUE_CAPACITY and MAX_COALESCED_EVENTS, so + // nothing may be dropped. + TraceClient client = TraceClient.builder(baseUrl(), "MyPlugin", "1.2.3").key("k").build(); + int threads = 8; + int events = 5; + assertTrue(2 * threads * events <= TraceClient.QUEUE_CAPACITY); + assertTrue(threads * events <= TraceClient.MAX_COALESCED_EVENTS); + + // Act + runAtOnce(threads, index -> { + for (int repeat = 0; repeat < 3; repeat++) { + for (int e = 0; e < events; e++) { + client.report("command", null, Collections.singletonMap("name", index + "-" + e)); + } + } + }); + client.close(); + + // Assert + List bodies = received.stream().map(r -> r.body).collect(Collectors.toList()); + assertEquals(2 * threads * events, bodies.size(), "a first report and a count per event: " + bodies); + for (int t = 0; t < threads; t++) { + for (int e = 0; e < events; e++) { + String tags = "\"tags\":{\"name\":\"" + t + "-" + e + "\",\"version\":\"1.2.3\""; + assertEquals(1, bodies.stream().filter(b -> b.equals( + "{\"application\":\"MyPlugin\",\"name\":\"command\"," + tags + "}}")).count(), + "one first report of " + t + "-" + e + ": " + bodies); + assertEquals(1, bodies.stream().filter(b -> b.equals( + "{\"application\":\"MyPlugin\",\"name\":\"command\",\"value\":2.0," + tags + ",\"coalesced\":\"2\"}}")).count(), + "one count of 2 for " + t + "-" + e + ": " + bodies); + } + } + } + @Test void coalesce_neverHoldsBackStartup() throws Exception { // Arrange