From 4a4c848b68948b6679172580f0b7664c1fadac6e Mon Sep 17 00:00:00 2001 From: John Gerassimou Date: Thu, 26 Feb 2026 01:10:30 -0500 Subject: [PATCH] feat: formalize P2 stress gates and ADR evidence traceability --- .github/workflows/ci.yml | 6 ++ README.md | 12 ++++ docs/adr/README.md | 1 + docs/adr/evidence-matrix.md | 17 ++++++ docs/p2-execution-board.md | 28 ++++++++++ .../BackpressureStressHarness.scala | 55 +++++++++++++++++++ .../BackpressureStressValidation.scala | 42 +++----------- .../PerformanceBaselineReport.scala | 36 ++++++++++++ .../SimpleStreamProcessorTest.scala | 15 +++-- .../StressInvariantTest.scala | 21 +++++++ 10 files changed, 194 insertions(+), 39 deletions(-) create mode 100644 docs/adr/evidence-matrix.md create mode 100644 docs/p2-execution-board.md create mode 100644 src/test/scala/SimpleStreamProcessor/BackpressureStressHarness.scala create mode 100644 src/test/scala/SimpleStreamProcessor/PerformanceBaselineReport.scala create mode 100644 src/test/scala/SimpleStreamProcessor/StressInvariantTest.scala diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fe61727..45b5a25 100644 --- a/.github/workflows/ci.yml +++ b/.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" diff --git a/README.md b/README.md index e58af25..7e308dc 100644 --- a/README.md +++ b/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 diff --git a/docs/adr/README.md b/docs/adr/README.md index 91f8306..2416a8b 100644 --- a/docs/adr/README.md +++ b/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) diff --git a/docs/adr/evidence-matrix.md b/docs/adr/evidence-matrix.md new file mode 100644 index 0000000..084d212 --- /dev/null +++ b/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"` diff --git a/docs/p2-execution-board.md b/docs/p2-execution-board.md new file mode 100644 index 0000000..a07fcec --- /dev/null +++ b/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. diff --git a/src/test/scala/SimpleStreamProcessor/BackpressureStressHarness.scala b/src/test/scala/SimpleStreamProcessor/BackpressureStressHarness.scala new file mode 100644 index 0000000..f33bd04 --- /dev/null +++ b/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}") + } +} diff --git a/src/test/scala/SimpleStreamProcessor/BackpressureStressValidation.scala b/src/test/scala/SimpleStreamProcessor/BackpressureStressValidation.scala index f9ce505..a9e9309 100644 --- a/src/test/scala/SimpleStreamProcessor/BackpressureStressValidation.scala +++ b/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) } } diff --git a/src/test/scala/SimpleStreamProcessor/PerformanceBaselineReport.scala b/src/test/scala/SimpleStreamProcessor/PerformanceBaselineReport.scala new file mode 100644 index 0000000..98a6301 --- /dev/null +++ b/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}") + } +} diff --git a/src/test/scala/SimpleStreamProcessor/SimpleStreamProcessorTest.scala b/src/test/scala/SimpleStreamProcessor/SimpleStreamProcessorTest.scala index faede7b..f35f552 100644 --- a/src/test/scala/SimpleStreamProcessor/SimpleStreamProcessorTest.scala +++ b/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") { diff --git a/src/test/scala/SimpleStreamProcessor/StressInvariantTest.scala b/src/test/scala/SimpleStreamProcessor/StressInvariantTest.scala new file mode 100644 index 0000000..ba17793 --- /dev/null +++ b/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) + } +}