Add a direct backend for plain blocking code

.github/workflows/ci.yml

@@ -89,7 +89,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        backend: [Zio, Ce, Ox, Pekko, Kyo]
+        backend: [Zio, Ce, Ox, Pekko, Kyo, Direct]
     steps:
       - name: Checkout current branch
         uses: actions/checkout@v6

README.md

@@ -5,7 +5,7 @@
 
 Sage is a Redis and Valkey client for Scala 3. It implements the Redis protocol directly in Scala instead of wrapping a Java client.
 
-- Choose the artifact for [ZIO](https://zio.dev), [Cats Effect](https://typelevel.org/cats-effect/), [Kyo](https://getkyo.io), [Ox](https://ox.softwaremill.com), or [Pekko](https://pekko.apache.org). Each artifact uses its ecosystem's native types.
+- Use the direct artifact for plain blocking code with no effect library dependency, or choose the artifact for [Ox](https://ox.softwaremill.com), [ZIO](https://zio.dev), [Cats Effect](https://typelevel.org/cats-effect/), [Kyo](https://getkyo.io), or [Pekko](https://pekko.apache.org), which uses that ecosystem's native types.
 - The core implements RESP3, commands, and codecs in Scala 3.
 - Redis 8+ and Valkey 8+ support includes auto-pipelining, transactions, cluster routing, sharded pub/sub, client-side caching, rate limiting, distributed locks, and TLS.
 

benchmarks/README.md

@@ -25,6 +25,7 @@ Each backend runs in a separate JVM and `projectMatrix` cell. The Sage backends
 | `sage-ox` / `lettuce` / `rediscala` / `jedis` | `benchmarksOx` | `lettuce` is the asynchronous, auto-pipelined Java baseline. `jedis` is the synchronous RESP3 Java client and uses one pooled connection per concurrency lane. |
 | `sage-pekko` | `benchmarksPekko` | no native Pekko (Future) competitor exists. |
 | `sage-kyo` | `benchmarksKyo` | No native Kyo competitor exists. |
+| `sage-direct` | `benchmarksDirect` | Runs each concurrency lane on a virtual thread. Compare it with `jedis` and `lettuce` from the Ox cell. |
 
 ## Running
 

benchmarks/direct/src/main/scala/sage/benchmarks/Clients.scala

@@ -0,0 +1,33 @@
+package sage.benchmarks
+
+import java.util.concurrent.Executors
+
+import scala.util.Using
+
+import sage.backend.*
+import sage.client.{Endpoint, SageConfig, Topology}
+
+/**
+  * The direct benchmark cell contains only Sage. The Ox cell already measures Jedis and Lettuce from blocking code on virtual threads.
+  */
+object Clients {
+  def build(host: String, port: Int, name: String): BenchClient = name match {
+    case "sage-direct" => new SageDirectBench(host, port)
+    case other         => throw new IllegalArgumentException(s"unknown client: $other")
+  }
+}
+
+/**
+  * Runs each lane on its own virtual thread, as a framework that serves each request on a virtual thread would.
+  */
+final class SageDirectBench(host: String, port: Int) extends SageBench[[A] =>> A] {
+
+  protected val client: SageClient = SageClient.connect(SageConfig(topology = Topology.Standalone(Endpoint(host, port))))
+
+  protected def run[A](result: A): Unit = ()
+
+  protected def inLanes[A](work: Payloads.Workload)(perKey: String => A): Any =
+    Using.resource(Executors.newVirtualThreadPerTaskExecutor()) { executor =>
+      work.lanes.map(lane => executor.submit[Unit](() => lane.foreach(perKey))).foreach(_.get())
+    }
+}

benchmarks/direct/src/main/scala/sage/benchmarks/DirectBenchmarks.scala

@@ -0,0 +1,13 @@
+package sage.benchmarks
+
+import org.openjdk.jmh.annotations.Param
+
+class ThroughputBench extends ThroughputBenchBase {
+
+  @Param(Array("sage-direct")) var client: String = "sage-direct"
+}
+
+class CollectionBench extends CollectionBenchBase {
+
+  @Param(Array("sage-direct")) var client: String = "sage-direct"
+}

benchmarks/merge-results.sh

@@ -1,11 +1,11 @@
 #!/usr/bin/env bash
-# Merge the per-cell JMH JSONs written by `sbt benchAll` into one all.json covering every client (5 sage backends + zio-redis + redis4cats +
+# Merge the per-cell JMH JSONs written by `sbt benchAll` into one all.json covering every client (6 sage backends + zio-redis + redis4cats +
 # lettuce + rediscala + jedis). Cells use the same benchmark class and method names and a unique `client` parameter. The resulting all.json
 # can be compared as one run or uploaded to https://jmh.morethan.io.
 set -euo pipefail
 cd "$(dirname "$0")/results"
 
-cells=(zio.json ce.json ox.json pekko.json kyo.json)
+cells=(zio.json ce.json ox.json pekko.json kyo.json direct.json)
 present=()
 for f in "${cells[@]}"; do [ -s "$f" ] && present+=("$f"); done
 if [ ${#present[@]} -eq 0 ]; then

benchmarks/run.sh

@@ -14,7 +14,7 @@ root="$(cd "$(dirname "$0")/.." && pwd)"
 cd "$root"
 
 args="$*"
-cells=(benchmarksZio:zio benchmarksCe:ce benchmarksOx:ox benchmarksPekko:pekko benchmarksKyo:kyo)
+cells=(benchmarksZio:zio benchmarksCe:ce benchmarksOx:ox benchmarksPekko:pekko benchmarksKyo:kyo benchmarksDirect:direct)
 failed=0
 
 rm -f benchmarks/results/*.json

build.sbt

@@ -32,6 +32,10 @@ val PekkoLib = CompatBackendAxis.external("pekko", "Pekko", "-pekko", Set("jvm")
 // suffixes match the removed built-in, which keeps cell ids and artifact names unchanged.
 val CeLib = CompatBackendAxis.local("ce", "Ce", "-ce", Set("jvm"))
 
+// The direct backend binds `kyo.compat` to plain blocking code on JDK virtual threads, implemented locally in `sage-compat-direct`. Its
+// artifacts depend only on Sage modules and the Scala library.
+val DirectLib = CompatBackendAxis.local("direct", "Direct", "-direct", Set("jvm"))
+
 inThisBuild(
   List(
     scalaVersion     := scala3Version,
@@ -50,30 +54,34 @@ name := "sage"
 addCommandAlias(
   "fmt",
   "all scalafmtSbt scalafmt test:scalafmt " +
-    "benchmarksZio/scalafmt benchmarksCe/scalafmt benchmarksOx/scalafmt benchmarksPekko/scalafmt benchmarksKyo/scalafmt"
+    "benchmarksZio/scalafmt benchmarksCe/scalafmt benchmarksOx/scalafmt benchmarksPekko/scalafmt benchmarksKyo/scalafmt " +
+    "benchmarksDirect/scalafmt"
 )
 addCommandAlias(
   "check",
   "all scalafmtSbtCheck scalafmtCheck test:scalafmtCheck " +
-    "benchmarksZio/scalafmtCheck benchmarksCe/scalafmtCheck benchmarksOx/scalafmtCheck benchmarksPekko/scalafmtCheck benchmarksKyo/scalafmtCheck"
+    "benchmarksZio/scalafmtCheck benchmarksCe/scalafmtCheck benchmarksOx/scalafmtCheck benchmarksPekko/scalafmtCheck benchmarksKyo/scalafmtCheck " +
+    "benchmarksDirect/scalafmtCheck"
 )
 
 addCommandAlias(
   "testUnit",
-  "all core/test opentelemetry/test clientZio/test clientCe/test clientOx/test clientPekko/test clientKyo/test " +
+  "all core/test opentelemetry/test clientZio/test clientCe/test clientOx/test clientPekko/test clientKyo/test clientDirect/test " +
     "clientFuture/Test/compile integrationTestsFuture/Test/compile integrationTestsPekko/Test/compile " +
-    "benchmarksZio/compile benchmarksCe/compile benchmarksOx/compile benchmarksPekko/compile benchmarksKyo/compile " +
+    "benchmarksZio/compile benchmarksCe/compile benchmarksOx/compile benchmarksPekko/compile benchmarksKyo/compile benchmarksDirect/compile " +
     "examplesZio/Compile/compile examplesCe/Compile/compile examplesOx/Compile/compile examplesPekko/Compile/compile " +
-    "examplesKyo/Compile/compile examplesFuture/Compile/compile; " +
-    // The conformance suite has wall-clock parallelism checks. Run it after the parallel block so five forked test cells do not starve it.
-    "ceConformanceCe/test"
+    "examplesKyo/Compile/compile examplesDirect/Compile/compile examplesFuture/Compile/compile; " +
+    // The conformance suite has wall-clock parallelism checks. Run it after the parallel block so six forked test cells do not starve it.
+    "ceConformanceCe/test; directConformanceDirect/test"
 )
 addCommandAlias("conformanceCe", "ceConformanceCe/test")
+addCommandAlias("conformanceDirect", "directConformanceDirect/test")
 addCommandAlias("itZio", "integrationTestsZio/test")
 addCommandAlias("itCe", "integrationTestsCe/test")
 addCommandAlias("itOx", "integrationTestsOx/test")
 addCommandAlias("itPekko", "integrationTestsPekko/test")
 addCommandAlias("itKyo", "integrationTestsKyo/test")
+addCommandAlias("itDirect", "integrationTestsDirect/test")
 
 addCommandAlias("exampleKyo", "examplesKyo/run")
 
@@ -86,7 +94,7 @@ addCommandAlias(
 
 addCommandAlias(
   "docAll",
-  "all core/doc opentelemetry/doc compatCe/doc clientZio/doc clientCe/doc clientOx/doc clientPekko/doc clientKyo/doc"
+  "all core/doc opentelemetry/doc compatCe/doc compatDirect/doc clientZio/doc clientCe/doc clientOx/doc clientPekko/doc clientKyo/doc clientDirect/doc"
 )
 
 lazy val root = project
@@ -96,7 +104,7 @@ lazy val root = project
   // on-demand benchmark runs. Use benchAll or benchmarks<Cell>/Jmh/run to run them directly.
   .aggregate(
     core.projectRefs ++ client.projectRefs ++ opentelemetry.projectRefs ++ integrationTests.projectRefs ++ examples.projectRefs ++
-      Seq[ProjectReference](compatCe): _*
+      Seq[ProjectReference](compatCe, compatDirect): _*
   )
 
 // Pure sans-IO core: RESP3 protocol, command model, codecs. Zero external dependencies.
@@ -156,6 +164,20 @@ lazy val ceConformance = (projectMatrix in file("sage-compat-ce/.conformance"))
   .bindLocally(CeLib, compatCe)
   .compatConformance()
 
+// A dependency-free binding of the `kyo.compat` package for blocking code on virtual threads. The `direct` cells of the client matrices bind
+// to this project through `bindLocally`.
+lazy val compatDirect = project
+  .in(file("sage-compat-direct"))
+  .settings(name := "sage-compat-direct")
+  .settings(commonSettings)
+
+// Run the bundled `kyo.compat` conformance suite against the direct binding, as `ceConformance` does for the Cats Effect binding.
+lazy val directConformance = (projectMatrix in file("sage-compat-direct/.conformance"))
+  .settings(publish / skip := true)
+  .compatLibrary(DirectLib)(VirtualAxis.jvm)(Seq(scala3Version))
+  .bindLocally(DirectLib, compatDirect)
+  .compatConformance()
+
 // Runtime written once against kyo-compat, cross-published per backend. JDK 21+.
 lazy val client = (projectMatrix in file("sage-client"))
   .dependsOn(core)
@@ -184,8 +206,9 @@ lazy val client = (projectMatrix in file("sage-client"))
       else Seq.empty
     }
   )
-  .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib)(VirtualAxis.jvm)(Seq(scala3Version))
+  .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib, DirectLib)(VirtualAxis.jvm)(Seq(scala3Version))
   .bindLocally(CeLib, compatCe)
+  .bindLocally(DirectLib, compatDirect)
 
 // Run each backend's smoke suite against a real server. Run the shared command, security, cluster, master-replica, rate-limit, and lock suites
 // on one backend because those features use the same implementation in every backend.
@@ -218,10 +241,11 @@ lazy val integrationTests = (projectMatrix in file("integration-tests"))
       Tests.Filter(name => !isAnchor && (isDesignated || !onceOnly.exists(name.startsWith)))
     }
   )
-  .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib)(VirtualAxis.jvm)(Seq(scala3Version))
+  .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib, DirectLib)(VirtualAxis.jvm)(Seq(scala3Version))
   .bindLocally(CeLib, compatCe)
+  .bindLocally(DirectLib, compatDirect)
 
-// Build runnable, unpublished examples for ZIO, Cats Effect, Ox, Kyo, and Pekko in separate cells so each uses its native client artifact. The
+// Build runnable, unpublished examples for ZIO, Cats Effect, Ox, Kyo, Pekko, and the direct client in separate cells so each uses its native client artifact. The
 // root project compiles them in CI through testUnit. The Future cell compiles examples/shared. Run an example against a local server with a
 // command such as `sbt examplesZio/run`. See examples/README.md.
 lazy val examples = (projectMatrix in file("examples"))
@@ -232,12 +256,13 @@ lazy val examples = (projectMatrix in file("examples"))
   // These runnable samples are not published. Skip API documentation because third-party parent types such as Cats Effect's IOApp contain
   // links that cannot be resolved here.
   .settings(Compile / doc / sources := Seq.empty)
-  .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib)(VirtualAxis.jvm)(Seq(scala3Version))
+  .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib, DirectLib)(VirtualAxis.jvm)(Seq(scala3Version))
   .bindLocally(CeLib, compatCe)
+  .bindLocally(DirectLib, compatDirect)
 
 // Runtime end-to-end benchmark harness (JMH), used only during development. Each backend has a separate cell. The ZIO cell includes
-// zio-redis, the Cats Effect cell includes redis4cats, and the Ox cell includes Lettuce, Rediscala, and Jedis. The Kyo and Pekko cells include
-// only Sage. See benchmarks/README.md.
+// zio-redis, the Cats Effect cell includes redis4cats, and the Ox cell includes Lettuce, Rediscala, and Jedis. The Kyo, Pekko, and direct
+// cells include only Sage. See benchmarks/README.md.
 lazy val benchmarks = (projectMatrix in file("benchmarks"))
   .dependsOn(client)
   .enablePlugins(JmhPlugin)
@@ -262,8 +287,9 @@ lazy val benchmarks = (projectMatrix in file("benchmarks"))
       else Seq.empty
     }
   )
-  .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib)(VirtualAxis.jvm)(Seq(scala3Version))
+  .compatLibrary(KyoLib, ZioLib, CeLib, OxLib, PekkoLib, DirectLib)(VirtualAxis.jvm)(Seq(scala3Version))
   .bindLocally(CeLib, compatCe)
+  .bindLocally(DirectLib, compatDirect)
 
 // The runtime benchmark harness: runs every backend cell's JMH suite against its own self-provisioned Redis (the future anchor cell is skipped),
 // each writing JMH JSON to benchmarks/results/<cell>.json. benchmarks/merge-results.sh merges them into one all.json covering every client
@@ -274,7 +300,8 @@ addCommandAlias(
     ";benchmarksCe/Jmh/run -rf json -rff benchmarks/results/ce.json " +
     ";benchmarksOx/Jmh/run -rf json -rff benchmarks/results/ox.json " +
     ";benchmarksPekko/Jmh/run -rf json -rff benchmarks/results/pekko.json " +
-    ";benchmarksKyo/Jmh/run -rf json -rff benchmarks/results/kyo.json"
+    ";benchmarksKyo/Jmh/run -rf json -rff benchmarks/results/kyo.json " +
+    ";benchmarksDirect/Jmh/run -rf json -rff benchmarks/results/direct.json"
 )
 
 lazy val commonSettings = Def.settings(

docs/client-side-caching.md

@@ -6,7 +6,7 @@ Caching is opt-in per call. An ordinary `get` always goes to the server; only `c
 
 ::: code-group
 
-```scala [Ox]
+```scala [Direct · Ox]
 client.set("cached:key", "v1")
 // first fetches and caches; second is a local hit
 val v1 = client.cached(Commands.get[String, String]("cached:key"), 1.minute)

docs/commands.md

@@ -8,7 +8,7 @@ A `Command[Out]` describes a server command without running it. Client methods c
 
 ::: code-group
 
-```scala [Ox]
+```scala [Direct · Ox]
 // per-command sugar
 val greeting = client.get[String]("greeting")
 

docs/distributed-locks.md

@@ -14,7 +14,7 @@ locks.withLock("account:42", waitTimeout = 2.seconds) {
 }
 ```
 
-The body returns your backend's native effect. With Ox, it is a direct-style block. Construct the operation inside the block: an already-started `Future` may run before acquisition. Keep all protected work inside the returned effect. Wait for every child task before the body completes because detached work can continue after Sage releases the lock.
+The body returns your backend's native effect. With the direct client and Ox, it is a plain block of code. Construct the operation inside the block: an already-started `Future` may run before acquisition. Keep all protected work inside the returned effect. Wait for every child task before the body completes because detached work can continue after Sage releases the lock.
 
 ## Waiting for a lock
 
@@ -55,7 +55,7 @@ Lock checks share `dedicatedPool` with blocking commands and transactions. A ful
 
 If replicas do not acknowledge acquisition in time, Sage raises `SageException.TimedOut` without starting the body. Acquisition can also raise `SageException.LockLost` if the granting master changes role or the lease expires before the body starts. Sage retries temporary renewal failures while the usable lease remains. If it cannot recover, or loses ownership, the operation fails with `SageException.LockLost` and attempts to cancel the body. Server and connection errors propagate through your backend's normal failure channel. Cleanup preserves an existing body failure or cancellation.
 
-Cancellation is cooperative and cannot undo completed work. Uninterruptible work can delay completion. Pekko uses `Future`, so its body may continue after lock loss even though Sage stops renewing the lock. A timed-out or cancelled acquisition can leave the lock held until its lease expires.
+Cancellation is cooperative and cannot undo completed work. Uninterruptible work can delay completion. Pekko uses `Future`, so its body may continue after lock loss even though Sage stops renewing the lock. With the direct client, the body runs on the calling thread, so it sees that thread's thread-local values, such as a framework's transaction context. Sage interrupts that thread to stop the body. A timed-out or cancelled acquisition can leave the lock held until its lease expires.
 
 ## Limits
 

docs/error-handling.md

@@ -1,6 +1,6 @@
 # Error handling
 
-Every Sage failure is a `SageException` in a sealed hierarchy that you can match exhaustively. ZIO and Kyo include `SageException` in the error type as `IO[SageException, *]` and `Abort[SageException]`. Cats Effect raises the exception in `IO`, Ox throws it, and Pekko returns a failed `scala.concurrent.Future`.
+Every Sage failure is a `SageException` in a sealed hierarchy that you can match exhaustively. The direct client and Ox throw it. ZIO and Kyo include `SageException` in the error type as `IO[SageException, *]` and `Abort[SageException]`. Cats Effect raises the exception in `IO`, and Pekko returns a failed `scala.concurrent.Future`.
 
 ## The hierarchy
 
@@ -73,8 +73,9 @@ Sage does not retry `-CLUSTERDOWN` for [commands that run on every master](/conf
 
 Each ecosystem reports the same `SageException` through its normal failure channel. ZIO and Kyo also include it in the error type. In those backends, another exception is a defect, such as a ZIO die or Kyo `Panic`.
 
+- The direct client throws the exception from the blocking call. Handle it with `try` and `catch`.
+- Ox throws the exception in direct style. Handle it with `try` and `catch`.
 - ZIO returns a failed `IO[SageException, *]`. Recover with `catchAll` or `catchSome`.
 - Cats Effect raises the exception in `IO`. Recover with `handleErrorWith` or `recoverWith` and match on `SageException`.
 - Kyo returns an `Abort[SageException]`. Handle it with the `Abort` combinators.
-- Ox throws the exception in direct style. Handle it with `try` and `catch`.
 - Pekko returns a failed `scala.concurrent.Future`. Recover with `recover` or `recoverWith`.

docs/faq.md

@@ -10,10 +10,11 @@ Run [the benchmarks](https://github.com/ghostdogpr/sage/tree/main/benchmarks) ag
 
 One per Scala stack, all sharing the same runtime:
 
+- Plain blocking code, with no effect library: `sage-client-direct`
+- Ox: `sage-client-ox`
 - ZIO: `sage-client-zio`
 - Cats Effect: `sage-client-ce`
 - Kyo: `sage-client-kyo`
-- Ox: `sage-client-ox`
 - Pekko: `sage-client-pekko`
 
 `sage-core` comes in transitively, so you depend on the backend artifact only. See [Getting started](/getting-started).