test(consumer): Tier-1 correctness hardening — real-class jcstress + deterministic crash-restart

lib/kpipe-consumer/src/jcstress/java/io/github/eschizoid/kpipe/consumer/CircuitBreakerWindowJCStressTest.java

@@ -1,7 +1,5 @@
 package io.github.eschizoid.kpipe.consumer;
 
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicLongArray;
 import org.openjdk.jcstress.annotations.Actor;
 import org.openjdk.jcstress.annotations.Arbiter;
 import org.openjdk.jcstress.annotations.Expect;
@@ -22,14 +20,10 @@
 /// equal 2 (neither record's counter increment was lost) and the failure rate must be exactly 0.5
 /// (one failure out of two), which also means failures never exceed samples.
 ///
-/// The real window lives as a `private static final` nested class inside the package-private
-/// `ConsumerHealthController`, and the host controller exposes neither its sample count nor its
-/// failure rate — only the resulting breaker state. The window therefore cannot be constructed or
-/// read back from a jcstress @State through the production surface. This @State replicates the
-/// exact slot-claim / swap / counter-adjust logic of that window on a fresh window of size 2, the
-/// same replicate-the-real-logic approach the state-transition CAS stress test uses. The two
-/// actors record one success and one failure concurrently; the arbiter then reads the running
-/// total and the derived failure rate.
+/// This drives the REAL window — `ConsumerHealthController.SlidingWindow`, made package-private for
+/// exactly this — not a copy, so a future change to its slot-claim / swap / counter-adjust logic is
+/// covered here rather than silently diverging from a replica. The two actors record one success
+/// and one failure concurrently; the arbiter reads the running total and the derived failure rate.
 ///
 /// jcstress runs the actors against fresh state under every interleaving its scheduler can
 /// produce, then evaluates the arbiter once both have finished. r1 carries the total sample count
@@ -49,7 +43,7 @@
 @State
 public class CircuitBreakerWindowJCStressTest {
 
-  private final RollingWindow window = new RollingWindow(2);
+  private final ConsumerHealthController.SlidingWindow window = new ConsumerHealthController.SlidingWindow(2);
 
   @Actor
   public void recordSuccess() {
@@ -66,46 +60,4 @@ public void observe(final JD_Result r) {
     r.r1 = window.totalSamples();
     r.r2 = window.failureRate();
   }
-
-  /// Count-based rolling window of success/failure outcomes, replicating the production window the
-  /// circuit breaker uses. Each `record` claims a unique slot via `head.getAndIncrement`, then
-  /// atomically swaps the slot, keeping the `successes` / `failures` counters in sync by
-  /// decrementing the evicted outcome and incrementing the new one so the failure rate is `O(1)`.
-  private static final class RollingWindow {
-
-    private static final long SUCCESS = 1L;
-    private static final long FAILURE = 2L;
-
-    private final AtomicLongArray slots;
-    private final AtomicLong head = new AtomicLong(0);
-    private final AtomicLong successes = new AtomicLong(0);
-    private final AtomicLong failures = new AtomicLong(0);
-
-    RollingWindow(final int windowSize) {
-      this.slots = new AtomicLongArray(windowSize);
-    }
-
-    void record(final boolean success) {
-      final long outcome = success ? SUCCESS : FAILURE;
-      final var idx = (int) (head.getAndIncrement() % slots.length());
-      final var prev = slots.getAndSet(idx, outcome);
-      if (prev == SUCCESS) successes.decrementAndGet();
-      else if (prev == FAILURE) failures.decrementAndGet();
-      if (outcome == SUCCESS) successes.incrementAndGet();
-      else failures.incrementAndGet();
-    }
-
-    long totalSamples() {
-      return successes.get() + failures.get();
-    }
-
-    double failureRate() {
-      // Snapshot both counters once — re-reading `failures` after `totalSamples` could produce a
-      // ratio above 1.0 if a failure arrived between the two reads.
-      final var f = failures.get();
-      final var s = successes.get();
-      final var total = f + s;
-      return total == 0 ? 0.0 : (double) f / total;
-    }
-  }
 }

lib/kpipe-consumer/src/jcstress/java/io/github/eschizoid/kpipe/consumer/StateTransitionCasJCStressTest.java

@@ -20,12 +20,14 @@
 /// window where both racers believe they performed the move, which would double-fire shutdown
 /// work that is meant to run exactly once.
 ///
-/// This @State reuses the real `ConsumerState` enum and replicates the exact single-read-CAS logic
-/// of the consumer's transition helper on a fresh reference initialized to `RUNNING`. The two
-/// actors run that helper concurrently. The property: exactly ONE actor observes a successful
-/// transition (one returns true, the other false), and the final state is `CLOSING`. A run where
-/// both actors return true (or both false, or the final state is anything other than `CLOSING`)
-/// means the single-read-CAS discipline failed to make the transition a single-winner operation.
+/// This @State holds a fresh `AtomicReference<ConsumerState>` initialized to `RUNNING` and drives
+/// the REAL transition logic — `KPipeConsumer.tryTransitionToClosing`, the static helper the
+/// instance method delegates to — not a copy, so the discipline is verified against the shipping
+/// code. The two actors run that helper concurrently. The property: exactly ONE actor observes a
+/// successful transition (one returns true, the other false), and the final state is `CLOSING`. A
+/// run where both actors return true (or both false, or the final state is anything other than
+/// `CLOSING`) means the single-read-CAS discipline failed to make the transition a single-winner
+/// operation.
 ///
 /// jcstress runs the actors against fresh state under every interleaving its scheduler can produce,
 /// then evaluates the arbiter once both have finished. r1/r2 carry each actor's CAS result and r3
@@ -51,22 +53,14 @@ public class StateTransitionCasJCStressTest {
 
   private final AtomicReference<ConsumerState> state = new AtomicReference<>(ConsumerState.RUNNING);
 
-  /// Single-read CAS toward CLOSING, mirroring the consumer's transition helper: read the current
-  /// state once, only proceed from an active state, then CAS against the captured value.
-  private boolean transitionToClosing() {
-    final var current = state.get();
-    if (current != ConsumerState.RUNNING && current != ConsumerState.PAUSED) return false;
-    return state.compareAndSet(current, ConsumerState.CLOSING);
-  }
-
   @Actor
   public void closer(final ZZL_Result r) {
-    r.r1 = transitionToClosing();
+    r.r1 = KPipeConsumer.tryTransitionToClosing(state);
   }
 
   @Actor
   public void selfTerminator(final ZZL_Result r) {
-    r.r2 = transitionToClosing();
+    r.r2 = KPipeConsumer.tryTransitionToClosing(state);
   }
 
   @Arbiter

lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/ConsumerHealthController.java

@@ -319,7 +319,9 @@ private void close() {
   /// Each `record` claims a unique slot via `head.getAndIncrement`, then atomically swaps the
   /// slot. Counters are kept in sync by decrementing the evicted outcome and incrementing the new
   /// one, so `failureRate()` is `O(1)` regardless of window size.
-  private static final class SlidingWindow {
+  // Package-private (not private) so the jcstress harness can drive the real window directly
+  // rather than a replica — same approach as the other package-private internals under test.
+  static final class SlidingWindow {
 
     private static final long EMPTY = 0L;
     private static final long SUCCESS = 1L;

lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/KPipeConsumer.java

@@ -1399,6 +1399,19 @@ public boolean isRunning() {
   ///
   /// @return true if the transition succeeded, false if the state was not active
   private boolean transitionToClosing() {
+    return tryTransitionToClosing(state);
+  }
+
+  /// The single-read compare-and-set that backs [#transitionToClosing], lifted to a static method
+  /// over the supplied reference so the exact discipline (read once into a local, verify the state
+  /// is active, then CAS to CLOSING) can be exercised under jcstress without constructing a full
+  /// consumer. Two threads racing this must yield exactly one winner; the loser sees CLOSING and
+  /// returns false. Package-private for that test only.
+  ///
+  /// @param state the lifecycle state reference to transition
+  /// @return true if this call performed the transition, false if the state was not active or
+  ///         another thread won the CAS
+  static boolean tryTransitionToClosing(final AtomicReference<ConsumerState> state) {
     final var current = state.get();
     if (current != ConsumerState.RUNNING && current != ConsumerState.PAUSED) return false;
     return state.compareAndSet(current, ConsumerState.CLOSING);