feat: formalize P2 stress gates and ADR evidence traceability

.github/workflows/ci.yml

@@ -26,5 +26,11 @@ jobs:
       - name: Run unit tests
         run: sbt -Dsbt.supershell=false test
 
+      - name: Run stress invariant suite
+        run: sbt -Dsbt.supershell=false "Test / testOnly SimpleStreamProcessor.StressInvariantTest"
+
       - name: Run backpressure stress assertions
         run: sbt -Dsbt.supershell=false "Test / runMain SimpleStreamProcessor.BackpressureStressValidation 10000"
+
+      - name: Run performance baseline smoke
+        run: sbt -Dsbt.supershell=false "Test / runMain SimpleStreamProcessor.PerformanceBaselineReport 5000 4 16"

README.md

@@ -44,6 +44,18 @@ Run calibrated backpressure stress validation (30s target):
 sbt -Dsbt.supershell=false "Test / runMain SimpleStreamProcessor.BackpressureStressValidation 30000"
 ```
 
+Run deterministic stress invariants suite:
+
+```bash
+sbt -Dsbt.supershell=false "Test / testOnly SimpleStreamProcessor.StressInvariantTest"
+```
+
+Run performance baseline snapshot:
+
+```bash
+sbt -Dsbt.supershell=false "Test / runMain SimpleStreamProcessor.PerformanceBaselineReport"
+```
+
 ## Current Scope Notes
 
 This implementation is single-process and in-memory. It is designed to validate pipeline semantics and concurrency

docs/adr/README.md

@@ -36,3 +36,4 @@ These ADRs define the baseline contracts for stream semantics and implementation
 - [Event-Time Example](event-time-example.md)
 - [Acceptance Criteria](acceptance-criteria.md)
 - [Metric Schema](metric-schema.md)
+- [Evidence Matrix](evidence-matrix.md)

docs/adr/evidence-matrix.md

@@ -0,0 +1,17 @@
+# ADR Evidence Matrix
+
+| ADR | Invariant Area | Evidence |
+| --- | --- | --- |
+| 0001 | Operator taxonomy exists across stateless/stateful/boundary/terminal operators | `src/main/scala/SimpleStreamProcessor/Node.scala`, `src/test/scala/SimpleStreamProcessor/SimpleStreamProcessorTest.scala` |
+| 0002 | Ordered `parMap`, bounded parallelism, invalid parallelism fail-fast | `src/main/scala/SimpleStreamProcessor/Stream.scala`, `src/test/scala/SimpleStreamProcessor/SimpleStreamProcessorTest.scala` (`Stream parMap preserves input order`, `Stream parMap fails fast on invalid parallelism`) |
+| 0003 | Fail-fast error propagation and recovery operators | `src/main/scala/SimpleStreamProcessor/Stream.scala`, `src/main/scala/SimpleStreamProcessor/Node.scala`, tests (`Stream recover converts failures into values`, `Node recoverWith allows stream fallback`) |
+| 0004 | Bounded async boundary and backpressure invariants | `src/main/scala/SimpleStreamProcessor/Node.scala`, `src/test/scala/SimpleStreamProcessor/SimpleStreamProcessorTest.scala` (`Async boundary queue depth stays within configured capacity`) and `src/test/scala/SimpleStreamProcessor/StressInvariantTest.scala` |
+| 0005 | Managed source/sink close semantics and precedence | `src/main/scala/SimpleStreamProcessor/Node.scala`, tests (`Managed sink preserves processing failure and suppresses close failure`, `Managed source preserves processing failure and suppresses close failure`) |
+| 0006 | Count windows and event-time window closure semantics | `src/main/scala/SimpleStreamProcessor/Node.scala`, tests (`Count windows split stream into fixed-size batches`, `Watermarks are emitted and event-time windows close`) |
+| 0007 | Late-event drop + watermark regression policy | `src/main/scala/SimpleStreamProcessor/Node.scala`, tests (`Event-time windows drop late records and ignore regressing watermarks`) |
+
+## Stress and Baseline Commands
+
+- `sbt -Dsbt.supershell=false "Test / testOnly SimpleStreamProcessor.StressInvariantTest"`
+- `sbt -Dsbt.supershell=false "Test / runMain SimpleStreamProcessor.BackpressureStressValidation 10000"`
+- `sbt -Dsbt.supershell=false "Test / runMain SimpleStreamProcessor.PerformanceBaselineReport"`

docs/p2-execution-board.md

@@ -0,0 +1,28 @@
+# P2 Execution Board
+
+## Objective
+
+Close the final stabilization scope before `v0.2.0` by hardening stress invariants, performance visibility, and ADR evidence traceability.
+
+## Work Items
+
+1. Stress invariants in CI
+   - Add a reusable stress harness API.
+   - Add deterministic stress invariant tests (bounded queue depth, completion under target, no boundary thread leaks).
+   - CI must run these tests explicitly.
+
+2. Performance baseline reporting
+   - Add a baseline runner that prints throughput/latency snapshots for the canonical pipeline.
+   - Keep output machine-readable enough for future regression parsing.
+
+3. ADR evidence traceability
+   - Publish ADR evidence matrix mapping ADR invariants to concrete test names/paths.
+   - Ensure matrix references both unit and stress validations.
+
+## Acceptance Criteria
+
+- `sbt -Dsbt.supershell=false test` is green.
+- `sbt -Dsbt.supershell=false "Test / testOnly SimpleStreamProcessor.StressInvariantTest"` is green.
+- `sbt -Dsbt.supershell=false "Test / runMain SimpleStreamProcessor.PerformanceBaselineReport"` runs successfully.
+- CI workflow includes explicit stress invariant job step.
+- `docs/adr/evidence-matrix.md` exists and covers ADR-0001 through ADR-0007.

src/test/scala/SimpleStreamProcessor/BackpressureStressHarness.scala

@@ -0,0 +1,55 @@
+package SimpleStreamProcessor
+
+import scala.concurrent.ExecutionContext
+
+object BackpressureStressHarness {
+  case class Result(
+    elapsedMs: Long,
+    sum: Long,
+    targetDurationMs: Long,
+    observedPerElementMs: Double,
+    elementCount: Int,
+    capacity: Int,
+    metrics: Metrics.Snapshot
+  )
+
+  def run(targetDurationMs: Long, capacity: Int = 8)(implicit executionContext: ExecutionContext): Result = {
+    val sleepPerElementMs = 1
+    val calibrationSamples = 200
+
+    val calibrationStart = System.nanoTime()
+    (1 to calibrationSamples).foreach(_ => Thread.sleep(sleepPerElementMs.toLong))
+    val observedPerElementMs = ((System.nanoTime() - calibrationStart) / 1000000.0) / calibrationSamples
+    val elementCount = math.max(1000, (targetDurationMs / observedPerElementMs).toInt + 300)
+
+    Metrics.reset()
+    val startedAt = System.nanoTime()
+
+    val sum = Source[Int](Stream.fromList((1 to elementCount).toList))
+      .asyncBoundary(capacity)
+      .map { value =>
+        Thread.sleep(sleepPerElementMs.toLong)
+        value
+      }
+      .toSink((acc: Int, i: Int) => acc + i, 0)
+      .run(Stream.Empty)
+
+    val elapsedMs = (System.nanoTime() - startedAt) / 1000000
+    val snapshot = Metrics.snapshot()
+
+    Result(
+      elapsedMs = elapsedMs,
+      sum = sum.toLong,
+      targetDurationMs = targetDurationMs,
+      observedPerElementMs = observedPerElementMs,
+      elementCount = elementCount,
+      capacity = capacity,
+      metrics = snapshot
+    )
+  }
+
+  def assertInvariants(result: Result): Unit = {
+    require(result.elapsedMs >= result.targetDurationMs, s"Expected >=${result.targetDurationMs} ms elapsed, got ${result.elapsedMs}")
+    require(result.metrics.boundaryQueueDepthMax <= result.capacity, s"Queue depth exceeded capacity: ${result.metrics.boundaryQueueDepthMax} > ${result.capacity}")
+  }
+}

src/test/scala/SimpleStreamProcessor/BackpressureStressValidation.scala

@@ -7,40 +7,16 @@ object BackpressureStressValidation {
     implicit val executionContext: ExecutionContext = ExecutionContext.global
 
     val targetDurationMs = args.headOption.map(_.toLong).getOrElse(5000L)
-    val capacity = 8
-    val sleepPerElementMs = 1
-    val calibrationSamples = 200
+    val result = BackpressureStressHarness.run(targetDurationMs = targetDurationMs, capacity = 8)
 
-    val calibrationStart = System.nanoTime()
-    (1 to calibrationSamples).foreach(_ => Thread.sleep(sleepPerElementMs.toLong))
-    val observedPerElementMs = ((System.nanoTime() - calibrationStart) / 1000000.0) / calibrationSamples
-    val elementCount = math.max(1000, (targetDurationMs / observedPerElementMs).toInt + 300)
+    println(s"elapsed_ms=${result.elapsedMs}")
+    println(s"sum=${result.sum}")
+    println(s"boundary_queue_depth_max=${result.metrics.boundaryQueueDepthMax}")
+    println(s"boundary_producer_blocked_ms=${result.metrics.boundaryProducerBlockedMs}")
+    println(s"target_duration_ms=${result.targetDurationMs}")
+    println(f"observed_per_element_ms=${result.observedPerElementMs}%.3f")
+    println(s"element_count=${result.elementCount}")
 
-    Metrics.reset()
-
-    val startedAt = System.nanoTime()
-
-    val result = Source[Int](Stream.fromList((1 to elementCount).toList))
-      .asyncBoundary(capacity)
-      .map { value =>
-        Thread.sleep(sleepPerElementMs.toLong)
-        value
-      }
-      .toSink((acc: Int, i: Int) => acc + i, 0)
-      .run(Stream.Empty)
-
-    val elapsedMs = (System.nanoTime() - startedAt) / 1000000
-    val snapshot = Metrics.snapshot()
-
-    println(s"elapsed_ms=$elapsedMs")
-    println(s"sum=$result")
-    println(s"boundary_queue_depth_max=${snapshot.boundaryQueueDepthMax}")
-    println(s"boundary_producer_blocked_ms=${snapshot.boundaryProducerBlockedMs}")
-    println(s"target_duration_ms=$targetDurationMs")
-    println(f"observed_per_element_ms=$observedPerElementMs%.3f")
-    println(s"element_count=$elementCount")
-
-    require(elapsedMs >= targetDurationMs, s"Expected >=$targetDurationMs ms elapsed, got $elapsedMs")
-    require(snapshot.boundaryQueueDepthMax <= capacity, s"Queue depth exceeded capacity: ${snapshot.boundaryQueueDepthMax} > $capacity")
+    BackpressureStressHarness.assertInvariants(result)
   }
 }

src/test/scala/SimpleStreamProcessor/PerformanceBaselineReport.scala

@@ -0,0 +1,36 @@
+package SimpleStreamProcessor
+
+import scala.concurrent.ExecutionContext
+
+object PerformanceBaselineReport {
+  def main(args: Array[String]): Unit = {
+    implicit val executionContext: ExecutionContext = ExecutionContext.global
+
+    val elementCount = args.headOption.map(_.toInt).getOrElse(20000)
+    val parallelism = args.lift(1).map(_.toInt).getOrElse(4)
+    val boundarySize = args.lift(2).map(_.toInt).getOrElse(16)
+
+    Metrics.reset()
+    val startedNs = System.nanoTime()
+
+    val sum = Source[Int](Stream.fromList((1 to elementCount).toList))
+      .parMap(parallelism)(_ * 2)
+      .asyncBoundary(boundarySize)
+      .toSink((acc: Int, i: Int) => acc + i, 0)
+      .run(Stream.Empty)
+
+    val elapsedNs = System.nanoTime() - startedNs
+    val elapsedMs = elapsedNs / 1000000.0
+    val throughputPerSec = (elementCount.toDouble / elapsedNs) * 1000000000.0
+    val snapshot = Metrics.snapshot()
+
+    println(s"element_count=$elementCount")
+    println(s"parallelism=$parallelism")
+    println(s"boundary_size=$boundarySize")
+    println(f"elapsed_ms=$elapsedMs%.3f")
+    println(f"throughput_per_sec=$throughputPerSec%.2f")
+    println(s"sum=$sum")
+    println(s"boundary_queue_depth_max=${snapshot.boundaryQueueDepthMax}")
+    println(s"boundary_producer_blocked_ms=${snapshot.boundaryProducerBlockedMs}")
+  }
+}

src/test/scala/SimpleStreamProcessor/SimpleStreamProcessorTest.scala

@@ -308,19 +308,22 @@ class SimpleStreamProcessorTest extends AnyFunSuite with BeforeAndAfterEach {
   }
 
   test("Async boundary queue depth stays within configured capacity") {
+    implicit val executionContext: ExecutionContext = ExecutionContext.global
     val capacity = 4
-    val result = Source[Int](Stream.fromList((1 to 300).toList))
+    val handle = Source[Int](Stream.fromList((1 to 300).toList))
       .asyncBoundary(capacity)
       .map { i =>
         Thread.sleep(1)
         i
       }
-      .run(Stream.Empty)
-      .toList
+      .runToListAsync(Stream.Empty)
+
+    val outcome = Await.result(handle.outcome, 2.seconds)
+    val metrics = handle.metricsSnapshot()
 
-    assert(result.size == 300)
-    assert(Metrics.snapshot().boundaryQueueDepthMax <= capacity)
-    assert(Metrics.snapshot().boundaryProducerBlockedMs >= 0)
+    assert(outcome == ExecutionCompleted((1 to 300).toList))
+    assert(metrics.boundaryQueueDepthMax <= capacity)
+    assert(metrics.boundaryProducerBlockedMs >= 0)
   }
 
   test("Managed sink close failure is recorded") {

src/test/scala/SimpleStreamProcessor/StressInvariantTest.scala

@@ -0,0 +1,21 @@
+package SimpleStreamProcessor
+
+import org.scalatest.funsuite.AnyFunSuite
+
+import scala.concurrent.ExecutionContext
+
+class StressInvariantTest extends AnyFunSuite {
+
+  test("Backpressure harness invariants hold across repeated runs") {
+    implicit val executionContext: ExecutionContext = ExecutionContext.global
+
+    val first = BackpressureStressHarness.run(targetDurationMs = 3000, capacity = 8)
+    val second = BackpressureStressHarness.run(targetDurationMs = 3000, capacity = 8)
+
+    BackpressureStressHarness.assertInvariants(first)
+    BackpressureStressHarness.assertInvariants(second)
+
+    assert(first.metrics.boundaryQueueDepthMax <= 8)
+    assert(second.metrics.boundaryQueueDepthMax <= 8)
+  }
+}