feat(demos): gRPC stream triage with eo-jsoniter and fs2

.github/workflows/ci.yml

@@ -114,10 +114,10 @@ jobs:
         run: sbt +update
 
       - name: Check formatting
-        run: sbt '++ ${{ matrix.scala }}' scalafmtCheckAll scalafmtSbtCheck benchmarks/scalafmtCheck demoUplcJvm/scalafmtCheckAll demoUplcJvmValidators/scalafmtCheckAll demoUplcJvmBench/scalafmtCheckAll
+        run: sbt '++ ${{ matrix.scala }}' scalafmtCheckAll scalafmtSbtCheck benchmarks/scalafmtCheck demoUplcJvm/scalafmtCheckAll demoUplcJvmValidators/scalafmtCheckAll demoUplcJvmBench/scalafmtCheckAll demoGrpcFilter/scalafmtCheckAll demoGrpcFilterBench/scalafmtCheckAll
 
       - name: Check scalafix
-        run: sbt '++ ${{ matrix.scala }}' 'scalafixAll --check' 'demoUplcJvm/scalafixAll --check' 'demoUplcJvmValidators/scalafixAll --check' 'demoUplcJvmBench/scalafixAll --check'
+        run: sbt '++ ${{ matrix.scala }}' 'scalafixAll --check' 'demoUplcJvm/scalafixAll --check' 'demoUplcJvmValidators/scalafixAll --check' 'demoUplcJvmBench/scalafixAll --check' 'demoGrpcFilter/scalafixAll --check' 'demoGrpcFilterBench/scalafixAll --check'
 
       - name: Check that workflows are up to date
         run: sbt githubWorkflowCheck
@@ -136,13 +136,16 @@ jobs:
       - name: Test UPLC → JVM demo (differential vs Scalus CEK)
         run: sbt '++ ${{ matrix.scala }}' demoUplcJvm/test demoUplcJvmBench/Jmh/compile
 
+      - name: Test gRPC filter demo (eo-jsoniter + fs2)
+        run: sbt '++ ${{ matrix.scala }}' demoGrpcFilter/test demoGrpcFilterBench/Jmh/compile
+
       - name: Make target directories
         if: github.event_name != 'pull_request' && (startsWith(github.ref, 'refs/tags/v'))
-        run: mkdir -p jsoniter/target benchmarks/target kyo/target target demos/uplc-jvm-bench/target circe/target zio/target unidocs/target avro/target laws/target tests/target generics/target demos/uplc-jvm-validators/target schemes-laws/target schemes/target core/target demos/uplc-jvm/target project/target
+        run: mkdir -p demos/grpc-filter/target jsoniter/target benchmarks/target kyo/target target demos/uplc-jvm-bench/target circe/target zio/target unidocs/target avro/target laws/target tests/target generics/target demos/uplc-jvm-validators/target schemes-laws/target schemes/target core/target demos/uplc-jvm/target demos/grpc-filter-bench/target project/target
 
       - name: Compress target directories
         if: github.event_name != 'pull_request' && (startsWith(github.ref, 'refs/tags/v'))
-        run: tar cf targets.tar jsoniter/target benchmarks/target kyo/target target demos/uplc-jvm-bench/target circe/target zio/target unidocs/target avro/target laws/target tests/target generics/target demos/uplc-jvm-validators/target schemes-laws/target schemes/target core/target demos/uplc-jvm/target project/target
+        run: tar cf targets.tar demos/grpc-filter/target jsoniter/target benchmarks/target kyo/target target demos/uplc-jvm-bench/target circe/target zio/target unidocs/target avro/target laws/target tests/target generics/target demos/uplc-jvm-validators/target schemes-laws/target schemes/target core/target demos/uplc-jvm/target demos/grpc-filter-bench/target project/target
 
       - name: Upload target directories
         if: github.event_name != 'pull_request' && (startsWith(github.ref, 'refs/tags/v'))

CLAUDE.md

@@ -30,12 +30,14 @@ Test-only: `org.typelevel:discipline-specs2_3:2.0.0`.
 | `benchmarks` | `benchmarks/` | — (not published) | JMH benchmarks vs Monocle (not part of root aggregate) |
 | `demoUplcJvm` | `demos/uplc-jvm/` | — (not published) | Demo: UPLC (Cardano Plutus Core) → JVM bytecode as a chain of LAWFUL monomorphic eo optics, `CborScript ⇀ TypedProgram ≅ JvmClass ⇀ JvmClassScript` (stages `toTyped`, `lowering`, `compiling` — the last is `UplcJvm.compile` / `decompile`; callers use `UplcJvm.toTyped` / `toJvmClass` / `toBytecode`, all from the script, all `MendTearPrism`s). Two PRISMS around one iso: Scalus decodes ANY declared Plutus Core version, but the ledger takes only those Plutus V3 has at PV11 (`Semantics.versions` = 1.0.0, 1.1.0), so `toTyped` misses on any other; a class can exceed the JVM's 64 KB method / 65 535-constant limits, so `compiling` misses (`UplcJvm.compile` gives the limit as a `Left` — the one catch of ASM's `MethodTooLargeException` / `ClassTooLargeException`). `lowering`'s way back types the embedded script WITHOUT the version check (only a typed program is lowered). Lawful because every stage keeps its script — `TypedProgram` its source, the class the script as Base64 constants — so `reverseGet` walks back exactly; `OpticSpec` runs cats-eo-laws' `PrismTests` on `toTyped` and `toBytecode` (JvmClass has no equality, so no `IsoTests` for `lowering`); tests reach stages through `Oracle.typed` / `jvmClass` / `classFile`, and `Oracle.compare` checks the round trips on every program. The four types hand out COPIES only (`bytes`, `tree`), so nothing changes a value after the compiler made or checked it. Compile time is STACK SAFE (`OpticSpec` compiles 100 000-deep programs on a 1 MB stack): decode, `Typer` and `Lower` are walks of `Step`s (`Step.scala`) — plain calls while shallow, `cats.Eval` on the heap from `Walk.Deep` (128) calls deep — so EVERY recursive call must go through `Walk.nested`. Trampolining every node (plain `Eval`) made translation ~2× slower. Steps run once each, in order, so `Lower` writes instructions as a step is made. Never sequence effectful `Eval`s with cats' `traverse_` / `void`: cats treats `Eval` as pure and drops a step whose result is unused. Scalus' flat decoder/encoder still recurse (overflow at 2 000–20 000 deep on 1 MB). `CborScript` / `JvmClassScript` are opaque over Scalus' `ByteString` (value equality, so the laws can use `==`), `JvmClass` opaque over ASM's `ClassNode` (made only by `Lower` and `UplcJvm.decompile`; `.tree` to inspect); `CborScript.parse` is the checked entry for untrusted bytes. Compiled with `-opt -opt-inline:**,!scala.**`: inlining from the Scala library made the short validations 12–27% slower (it grows `Rt.force` past what the JIT inlines), everything else measured neutral to ~5% faster. Keep generated code free of unreachable instructions (the typer drops what follows a failure): dead code leaves constant-pool entries that break the class file's read/write fixpoint. TypedProgram = typed Scala-like expression over Scalus' own `Term`s (hybrid typing: types are Scalus `DefaultUni`s, builtin signatures its `TypeScheme`s, instantiated at argument types; unboxed `BigInt`/`boolean`/`ByteString`/`String` where provable, boxed Scalus `Constant`s elsewhere), JvmClass written by asm-commons' `InstructionAdapter` (no own IR — and not a library above ASM: those cost translation speed), class file by ASM `ClassWriter` (computes frames). Saturated builtins are direct JVM calls of `scalus.uplc.builtin.Builtins` (or lazy-`Data` natives in `Rt`); the rest, and builtins used as values, run Scalus' CEK meanings. Reuse Scalus code before adding demo-local code: failures are `MachineError`, `Exec.run` returns `eval.Result`. Correctness is DIFFERENTIAL against the Scalus 1.3.0 CEK machine (Plutus V3 @ PV11, variant E — `Semantics.scala`): vendored Plutus conformance suite (`src/test/resources/plutus-conformance`, regenerate with `scripts/refresh-conformance.sh`, never hand-edit) + validators. Result/error only — no budget metering. Not in the root aggregate (Scalus pulls JNI crypto); CI runs `demoUplcJvm/test` as its own step. See its README |
 | `demoUplcJvmValidators` | `demos/uplc-jvm-validators/` | — (not published) | On-chain validators compiled to UPLC by `scalus-plugin` (isolated so the plugin never sees the compiler's sources), plus the shared off-chain fixtures (`Contexts`, `Scenarios`); test-scope dep of `demoUplcJvm` |
-| `demoUplcJvmBench` | `demos/uplc-jvm-bench/` | — (not published) | JMH for the demo: CEK machine vs generated class per validator, Fibonacci scaling, translation cost. NOT part of `benchmarks/`, the nightly sweep or `BENCHMARKS.md`; its numbers live on `site/docs/demos/uplc-jvm.md`. CI only compiles it (`demoUplcJvmBench/Jmh/compile`) |
+| `demoUplcJvmBench` | `demos/uplc-jvm-bench/` | — (not published) | JMH for the demo: CEK machine vs generated class per validator, Fibonacci scaling, translation cost. NOT part of `benchmarks/`, the nightly sweep or `BENCHMARKS.md`; its numbers live on `site/docs/cookbook/uplc-jvm.md` (the cookbook's "Blockchain smart contract compilation demo" page). CI only compiles it (`demoUplcJvmBench/Jmh/compile`) |
+| `demoGrpcFilter` | `demos/grpc-filter/` | — (not published) | Demo: an fs2 filter on a gRPC stream read as raw bytes off a TCP socket — Length-Prefixed-Message frames (`GrpcFrames.splitFrames` / `joinFrames` pipes) of JSON `Issue`s (64 fields; the 61st, `detail`, a 3-case enum nested 3 objects deep). It classifies and filters every message with eo-jsoniter prisms WITHOUT decoding it (`Wire.detail` = `JsoniterPrism[Issue].field(_.detail).raw` slices the detail's bytes into the opaque `DetailJson`; typed per-case leaf prisms, `JsoniterPrism[Detail.Incident].field(_.impact)…`, read it) and forwards kept frames byte for byte. `Policy.keep` is written ONCE against `CanGetOption` and runs on bytes (`Wire` givens) and on a decoded `Detail` (`Model` givens: eo-generics `prism`/`lens` chains, checked with `OptionalTests`); the differential tests compare the two on every message. Leaf prisms don't check the kind — the policy classifies first. `Wire.kind` is the one string path (`"type"` is the codec's tag, not a model field) and the one construction-time throw; `Issue.decode` the one catch of jsoniter-scala's throwing decode. `GrpcFrames.Malformed` fails the stream (compressed / oversized / truncated frames): no HTTP/2, no compression. Sockets are read 256 KiB at a time (`Net.reads`): fs2's 8 KiB `reads` was the loopback bottleneck. `Issues` = ScalaCheck generators DERIVED from the model by scalacheck-derived (only leaves hand-written: text, JSON-safe doubles, the opaque `Module` / `Votes`); `Issues(seed)` samples at `Issues.Size` (7, ~4 KB/message) with `Seed.slide` — consecutive `Seed.next` seeds draw overlapping, correlated values; derived sums shrink the size by one per level and fail at 0, so properties start at `minSize = 2`. Not in the root aggregate; CI runs `demoGrpcFilter/test` |
+| `demoGrpcFilterBench` | `demos/grpc-filter-bench/` | — (not published) | JMH for that demo: one message judged (full decode vs optics, with `detail` late as written or moved first) and the in-memory filter pipe. Numbers live on `site/docs/cookbook/grpc-filter.md` (the cookbook's "Allocation-free stream triage demo" page). CI only compiles it (`demoGrpcFilterBench/Jmh/compile`) |
 
 The root project aggregates `core`, `laws`, `tests`, `generics`, `schemes`,
 `schemesLaws`, `circe`, `avro`, `jsoniter`, `zio`, and `kyo`. `sbt compile` and `sbt test`
-cover those; benchmarks and the `demoUplcJvm` demo must be invoked explicitly
-(see below; `sbt demoUplcJvm/test`).
+cover those; benchmarks and the demos must be invoked explicitly
+(see below; `sbt demoUplcJvm/test`, `sbt demoGrpcFilter/test`).
 
 Published modules compile with Scala 3's bytecode optimizer (`scala3LibrarySettings` in
 `build.sbt`): `-opt`, plus `-opt-inline:dev.constructive.eo.**` — the inliner may copy code
@@ -76,6 +78,8 @@ sbt test                              # run all test suites (core, tests, generi
 sbt "~compile"                        # watch compile
 scalafmt                              # format in place per .scalafmt.conf
 scalafmt --check                      # CI-style verify
+                                      # (`project.git = true`: untracked files are skipped —
+                                      #  `git add` new files before formatting them)
 scalafix --rules RemoveUnused         # example rewrite
 scala -e 'println(1 + 2)'             # quick REPL eval
 ```

README.md

@@ -138,7 +138,7 @@ for cats typeclasses.
 - jsoniter-scala integration: <https://eo.constructive.dev/integrations/jsoniter.html>
 - ZIO integration: <https://eo.constructive.dev/integrations/zio.html>
 - Kyo integration (Records, kyo-schema bridge): <https://eo.constructive.dev/integrations/kyo.html>
-- Cookbook (recipes): <https://eo.constructive.dev/cookbook.html>
+- Cookbook (recipes): <https://eo.constructive.dev/cookbook/index.html>
 - Demo — a UPLC (Cardano Plutus Core) → JVM bytecode compiler built as one
   eo `Iso`, checked against the Scalus CEK machine:
   [`demos/uplc-jvm`](./demos/uplc-jvm/README.md)

build.sbt

@@ -68,6 +68,8 @@ ThisBuild / githubWorkflowBuildPreamble ++= Seq(
       "demoUplcJvm/scalafmtCheckAll",
       "demoUplcJvmValidators/scalafmtCheckAll",
       "demoUplcJvmBench/scalafmtCheckAll",
+      "demoGrpcFilter/scalafmtCheckAll",
+      "demoGrpcFilterBench/scalafmtCheckAll",
     ),
     name = Some("Check formatting"),
   ),
@@ -77,6 +79,8 @@ ThisBuild / githubWorkflowBuildPreamble ++= Seq(
       "demoUplcJvm/scalafixAll --check",
       "demoUplcJvmValidators/scalafixAll --check",
       "demoUplcJvmBench/scalafixAll --check",
+      "demoGrpcFilter/scalafixAll --check",
+      "demoGrpcFilterBench/scalafixAll --check",
     ),
     name = Some("Check scalafix"),
   ),
@@ -89,6 +93,10 @@ ThisBuild / githubWorkflowBuild += WorkflowStep.Sbt(
   List("demoUplcJvm/test", "demoUplcJvmBench/Jmh/compile"),
   name = Some("Test UPLC → JVM demo (differential vs Scalus CEK)"),
 )
+ThisBuild / githubWorkflowBuild += WorkflowStep.Sbt(
+  List("demoGrpcFilter/test", "demoGrpcFilterBench/Jmh/compile"),
+  name = Some("Test gRPC filter demo (eo-jsoniter + fs2)"),
+)
 
 // -------------------------------------------------------------------
 // Scalafix wiring. SemanticDB exports are required by the semantic
@@ -986,7 +994,7 @@ lazy val docs: Project = project
     mdocExtraArguments ++= Seq(
       "--scalac-options",
       // 120s: the 60s budget still tripped intermittently inside cold-JVM
-      // pre-commit hook sessions (cookbook.md avro fences on a loaded box).
+      // pre-commit hook sessions (the cookbook's avro fences on a loaded box).
       // The `s` suffix is required — kindlings' DurationPattern rejects a bare
       // integer and silently reverts to the 5s default (see the ThisBuild
       // scalacOptions comment above).
@@ -1225,6 +1233,57 @@ lazy val demoUplcJvmBench: Project = project
     publish / skip := true,
   )
 
+// -------------------------------------------------------------------
+// Demo: triage a gRPC stream of JSON messages straight off a socket with
+// eo-jsoniter optics and fs2 — every message is classified and filtered by
+// reading only the few leaves the policy needs, never decoding the 64-field
+// message, and kept messages are forwarded byte for byte. Unpublished and OUT
+// of the root aggregate (CI runs it as its own step); run with
+// `sbt demoGrpcFilter/test` or `sbt demoGrpcFilter/run`.
+// -------------------------------------------------------------------
+lazy val Fs2Version = "3.14.0"
+lazy val fs2Io = "co.fs2" %% "fs2-io" % Fs2Version
+// Arbitrary / Cogen derivation from Scala 3 mirrors: the demo's issues are drawn from generators
+// derived from the model.
+lazy val scalacheckDerived = "io.github.martinhh" %% "scalacheck-derived" % "0.10.0"
+
+lazy val demoGrpcFilter: Project = project
+  .in(file("demos/grpc-filter"))
+  .dependsOn(
+    LocalProject("core"),
+    LocalProject("generics"),
+    LocalProject("jsoniterIntegration"),
+    // OptionalTests: the model's optics are checked against the library's own rule set.
+    LocalProject("laws") % Test,
+  )
+  .settings(commonSettings *)
+  .settings(
+    name := "cats-eo-demo-grpc-filter",
+    publish / skip := true,
+    libraryDependencies += fs2Io,
+    // The message model's codecs: the source encodes whole messages, the baseline decodes them.
+    libraryDependencies += jsoniterMacros,
+    // The stream's issues come from ScalaCheck generators derived from the model, in main code:
+    // the demo's upstream service and the benchmarks draw from them too.
+    libraryDependencies += scalacheck,
+    libraryDependencies += scalacheckDerived,
+    libraryDependencies += discipline % Test,
+    Compile / run / fork := true,
+  )
+
+// JMH for the gRPC filter demo: full decode against eo-jsoniter triage, per message and per
+// stream. Results are reported on the demo's docs page, like the UPLC demo's.
+// Run with `sbt "demoGrpcFilterBench/Jmh/run -prof gc"`.
+lazy val demoGrpcFilterBench: Project = project
+  .in(file("demos/grpc-filter-bench"))
+  .enablePlugins(JmhPlugin)
+  .dependsOn(LocalProject("demoGrpcFilter"))
+  .settings(commonSettings *)
+  .settings(
+    name := "cats-eo-demo-grpc-filter-bench",
+    publish / skip := true,
+  )
+
 // Single source of truth for the JMH invocation (the config that the per-class
 // annotations pin — see JmhDefaults' Phase-4 note for why that preamble can't be
 // DRY'd). Append a filter: `sbt "bench .*OrderAvroBench.*"`.

demos/grpc-filter-bench/src/main/scala/dev/constructive/eo/demo/grpc/bench/StreamBench.scala

@@ -0,0 +1,57 @@
+package dev.constructive.eo.demo.grpc
+package bench
+
+import scala.compiletime.uninitialized
+
+import cats.effect.unsafe.implicits.global
+import cats.effect.{IO, Ref}
+import com.github.plokhotnyuk.jsoniter_scala.core.writeToArray
+import fs2.{Chunk, Stream}
+import java.util.concurrent.TimeUnit
+import org.openjdk.jmh.annotations.*
+
+/** The whole filter, per message, in memory: gRPC frames arriving in 256 KiB pieces (as the demo
+  * reads its socket) are cut into messages, judged, tallied, and the kept ones framed again. No
+  * sockets, so what differs between the two `judge`s is what each costs in a running pipeline.
+  */
+@State(Scope.Benchmark)
+@BenchmarkMode(Array(Mode.AverageTime))
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Fork(3)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 5, time = 2)
+@OperationsPerInvocation(StreamBench.Messages)
+class StreamBench:
+
+  @Param(Array("optics", "decode"))
+  var judge: String = uninitialized
+
+  private var verdict: Array[Byte] => Verdict = uninitialized
+
+  private val pieces: Vector[Chunk[Byte]] =
+    val framed = Stream
+      .emits(Issues(seed = 7).take(StreamBench.Messages).map(writeToArray(_)).toSeq)
+      .through(GrpcFrames.joinFrames)
+      .compile
+      .to(Array)
+    framed.grouped(256 * 1024).map(Chunk.array(_)).toVector
+
+  @Setup def prepare(): Unit =
+    verdict =
+      if judge == "optics" then Triage.optics(Policy.default) else Triage.decoding(Policy.default)
+
+  @Benchmark def filter: Long =
+    Ref
+      .of[IO, Tally](Tally.empty)
+      .flatMap(tally =>
+        Stream
+          .emits(pieces)
+          .flatMap(Stream.chunk)
+          .through(Triage.filter(verdict, tally))
+          .compile
+          .count
+      )
+      .unsafeRunSync()
+
+object StreamBench:
+  final val Messages = 1024