fix(detectors): keep analysis whole while recording is still under way (#1212)

async-test-lib/src/main/java/se/deversity/asynctest/diagnostics/CompletableFutureObtrudeDetector.java

@@ -50,8 +50,10 @@ State newState(Object instance, String label) {
     public void recordObtrude(CompletableFuture<?> future, String label, Thread thread) {
         if (future == null || thread == null) return;
         State s = stateFor(future, label, "CompletableFuture");
-        s.obtrudeCount.incrementAndGet();
+        // The thread first, then the count: stateFor publishes the state before either is set, and
+        // analyze reads the count first, so a count it sees above zero has a thread name behind it.
         s.lastObtrudedByThread = thread.getName();
+        s.obtrudeCount.incrementAndGet();
     }
     /**
      * Analyses what has been recorded about the observation and builds the report for it.
@@ -61,9 +63,16 @@ public void recordObtrude(CompletableFuture<?> future, String label, Thread thre
     public Report analyze() {
         Report r = new Report();
         for (State s : states()) {
+            // One read of each field, count first (see recordObtrude). A state registered by a
+            // recordObtrude still under way has no count yet and nothing to report.
+            int count = s.obtrudeCount.get();
+            String lastThread = s.lastObtrudedByThread;
+            if (count == 0 || lastThread == null) {
+                continue;
+            }
             String msg = String.format(
                 "CompletableFuture '%s' obtruded %d times (last by thread '%s') — obtruding values or exceptions forces downstream pipelines to execute with outdated/inconsistent states, introducing publication races.",
-                s.label, s.obtrudeCount.get(), s.lastObtrudedByThread
+                s.label, count, lastThread
             );
             r.violations.add(msg);
             r.structuredViolations.add(new Violation(
@@ -73,8 +82,8 @@ public Report analyze() {
                 List.of(),
                 Map.of(
                     "label", s.label,
-                    "obtrudeCount", s.obtrudeCount,
-                    "lastObtrudedByThread", s.lastObtrudedByThread
+                    "obtrudeCount", count,
+                    "lastObtrudedByThread", lastThread
                 ),
                 Instant.now()
             ));

async-test-lib/src/main/java/se/deversity/asynctest/diagnostics/ThreadPoolDeadlockDetector.java

@@ -70,6 +70,8 @@ static final class PoolState {
         final int poolSize;
         final AtomicInteger activeTaskCount = new AtomicInteger(0);
         final AtomicInteger nestedSubmissionCount = new AtomicInteger(0);
+        /** This pool's share of the detector-wide deadlockRiskCount, so analyze can total what it lists. */
+        final AtomicInteger deadlockRisks = new AtomicInteger(0);
         final List<NestedSubmissionEvent> nestedSubmissions = Collections.synchronizedList(new ArrayList<>());
 
         PoolState(String name, int poolSize) {
@@ -133,6 +135,7 @@ public void recordNestedSubmission(ExecutorService pool, String name) {
 
             // Check for deadlock risk: active tasks >= pool size
             if (activeTasks >= state.poolSize) {
+                state.deadlockRisks.incrementAndGet();
                 deadlockRiskCount.incrementAndGet();
             }
         }
@@ -166,20 +169,28 @@ public ThreadPoolDeadlockReport analyze() {
             return DetectorFailurePolicy.checkedReport(this, report801);
         }
 
+        // The total comes from the pools this pass lists, not from deadlockRiskCount read after the
+        // loop: a submission recorded between the two made the report claim an issue it listed no
+        // pool for. Risks are read before the nested count, which recordNestedSubmission raises
+        // first, so a pool with a risk always makes the list.
         List<PoolDeadlockRisk> risks = new ArrayList<>();
+        int totalRisks = 0;
         for (PoolState state : states()) {
-            if (state.nestedSubmissionCount.get() > 0) {
+            int poolRisks = state.deadlockRisks.get();
+            int nested = state.nestedSubmissionCount.get();
+            if (nested > 0) {
+                totalRisks += poolRisks;
                 risks.add(new PoolDeadlockRisk(
                     state.name,
                     state.poolSize,
-                    state.nestedSubmissionCount.get(),
+                    nested,
                     state.activeTaskCount.get(),
                     new ArrayList<>(state.nestedSubmissions)
                 ));
             }
         }
 
-        ThreadPoolDeadlockReport report801 = new ThreadPoolDeadlockReport(risks, deadlockRiskCount.get());
+        ThreadPoolDeadlockReport report801 = new ThreadPoolDeadlockReport(risks, totalRisks);
         report801.fillStructuredViolations();
         return DetectorFailurePolicy.checkedReport(this, report801);
     }

async-test-lib/src/test/java/se/deversity/asynctest/diagnostics/CompletableFutureObtrudeDetectorTest.java

@@ -78,4 +78,21 @@ void doesNotKeepARecordedSubjectAlive() throws InterruptedException {
         CompletableFutureObtrudeDetector d = new CompletableFutureObtrudeDetector();
         AbstractInstanceDetectorTest.assertNotRetained(CompletableFuture<Void>::new, f -> d.recordObtrude(f, null, Thread.currentThread()));
     }
+
+    /**
+     * A recordObtrude still under way has registered the future but not yet set its thread or
+     * count. analyze() put that null thread into Map.of and threw, so a worker still recording
+     * dropped the run's whole report (DetectorAnalysisUnderConcurrentWritesTest, #1212). This
+     * builds the half-recorded state directly instead of waiting for the scheduler to.
+     */
+    @Test
+    void aFutureRegisteredButNotYetRecordedIsNotReported() {
+        CompletableFutureObtrudeDetector d = new CompletableFutureObtrudeDetector();
+        d.stateFor(new CompletableFuture<String>(), "mid-record", "CompletableFuture");
+
+        CompletableFutureObtrudeDetector.Report report = assertDoesNotThrow(d::analyze);
+
+        assertFalse(report.hasIssues());
+        assertTrue(report.structuredViolations.isEmpty());
+    }
 }

async-test-lib/src/test/java/se/deversity/asynctest/diagnostics/ThreadPoolDeadlockDetectorTest.java

@@ -261,4 +261,31 @@ void doesNotKeepARecordedSubjectAlive() throws InterruptedException {
         ThreadPoolDeadlockDetector d = new ThreadPoolDeadlockDetector();
         AbstractInstanceDetectorTest.assertNotRetained(() -> new java.util.concurrent.ThreadPoolExecutor(1, 1, 0, java.util.concurrent.TimeUnit.SECONDS, new java.util.concurrent.LinkedBlockingQueue<Runnable>()), e -> d.registerPool(e, null));
     }
+
+    /**
+     * The report's total and its pool list must come from the same pass. analyze() listed the
+     * pools, then read the detector-wide risk count; a submission recorded in between (or a pool
+     * whose weak key was cleared) left a total above zero with no pool listed, and the report
+     * claimed an issue it had no structured violation for (#1212). Dropping the pool's state
+     * while the detector-wide count stays reproduces that gap without depending on timing.
+     */
+    @Test
+    void aRiskWithNoListedPoolDoesNotMakeAnIssue() {
+        ThreadPoolDeadlockDetector detector = new ThreadPoolDeadlockDetector();
+        ExecutorService pool = Executors.newSingleThreadExecutor();
+        try {
+            detector.registerPool(pool, "single");
+            detector.recordNestedSubmission(pool, "single");
+            assertEquals(1, detector.getDeadlockRiskCount());
+            detector.clearStates();
+
+            ThreadPoolDeadlockDetector.ThreadPoolDeadlockReport report =
+                    assertDoesNotThrow(detector::analyze);
+
+            assertFalse(report.hasIssues());
+            assertEquals(report.hasIssues(), !report.structuredViolations.isEmpty());
+        } finally {
+            pool.shutdownNow();
+        }
+    }
 }

docs/CHANGELOG.md

@@ -13,6 +13,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- `CompletableFutureObtrudeDetector` and `ThreadPoolDeadlockDetector` no longer fail analysis while
+  a worker is still recording, which dropped the run's whole report (#1212). The first put a future's
+  thread name into its violation before `recordObtrude` had set it, a `NullPointerException`; a future
+  registered but not yet counted is now left out. The second listed the pools, then read its risk total
+  afterwards, so a submission recorded in between reported an issue with no structured violation; the
+  total now comes from the pools the same pass lists.
 - `AtomicNonAtomicUpdate` no longer reports a plain store whose value was never built from the
   atomic read it followed. `gauge.get()` used for a log line, then `gauge.set(queue.size())`, lost no
   update, and was reported as one (#1187). The agent now analyses each class's bytes, decides per store