feat: add scheduler-agnostic connection pool

- Add lib/pool.ml using kcas for lock-free concurrency
- Works with any OCaml scheduler (Eio, Lwt, direct-style)
- Features: acquire/release, blocking acquire with timeout,
connection validation, pool stats, graceful shutdown
- Fix thread-safety in lib/stream.ml (use Atomic counter)
- Add comprehensive tests for Lwt and Eio integration
- Add pool integration tests to SQLite and PostgreSQL drivers
- All 181 unit tests + integration tests pass

driver-postgresql/test_integration.ml

···
     ("update_returning", `Quick, test_query_update_returning);
   ]
 
+module Pool = Repodb.Pool
+
+let pool_config () =
+  Pool.
+    {
+      max_size = 3;
+      connect =
+        (fun () ->
+          Repodb_postgresql.connect conninfo
+          |> Result.map_error Repodb_postgresql.error_message);
+      close = Repodb_postgresql.close;
+      validate =
+        Some
+          (fun conn ->
+            match Repodb_postgresql.exec conn "SELECT 1" ~params:[||] with
+            | Ok () -> true
+            | Error _ -> false);
+    }
+
+let with_pool f () =
+  let pool = Pool.create (pool_config ()) in
+  (match Pool.acquire pool with
+  | Ok conn ->
+      let _ =
+        Repodb_postgresql.exec conn "DROP TABLE IF EXISTS pool_test" ~params:[||]
+      in
+      let _ =
+        Repodb_postgresql.exec conn "DROP TABLE IF EXISTS counter" ~params:[||]
+      in
+      Pool.release pool conn
+  | Error _ -> ());
+  Fun.protect
+    ~finally:(fun () ->
+      (match Pool.acquire pool with
+      | Ok conn ->
+          let _ =
+            Repodb_postgresql.exec conn "DROP TABLE IF EXISTS pool_test"
+              ~params:[||]
+          in
+          let _ =
+            Repodb_postgresql.exec conn "DROP TABLE IF EXISTS counter"
+              ~params:[||]
+          in
+          Pool.release pool conn
+      | Error _ -> ());
+      Pool.shutdown pool)
+    (fun () -> f pool)
+
+let test_pool_acquire_release =
+  with_pool (fun pool ->
+      match Pool.acquire pool with
+      | Error e -> Alcotest.fail (Pool.error_to_string e)
+      | Ok conn ->
+          (match Repodb_postgresql.exec conn "SELECT 1" ~params:[||] with
+          | Ok () -> ()
+          | Error e -> Alcotest.fail (Repodb_postgresql.error_message e));
+          Pool.release pool conn;
+          Alcotest.(check int) "in_use after release" 0 (Pool.in_use pool))
+
+let test_pool_with_connection =
+  with_pool (fun pool ->
+      match Pool.acquire pool with
+      | Error e -> Alcotest.fail (Pool.error_to_string e)
+      | Ok conn -> (
+          let _ =
+            Repodb_postgresql.exec conn
+              "CREATE TABLE pool_test (id SERIAL PRIMARY KEY, value TEXT)"
+              ~params:[||]
+          in
+          Pool.release pool conn;
+
+          let result =
+            Pool.with_connection pool (fun conn ->
+                match
+                  Repodb_postgresql.exec conn
+                    "INSERT INTO pool_test (value) VALUES ($1)"
+                    ~params:[| Repodb.Driver.Value.text "test" |]
+                with
+                | Ok () -> "inserted"
+                | Error e -> Repodb_postgresql.error_message e)
+          in
+          match result with
+          | Ok "inserted" -> ()
+          | Ok msg -> Alcotest.fail msg
+          | Error e -> Alcotest.fail (Pool.error_to_string e)))
+
+let test_pool_max_size =
+  with_pool (fun pool ->
+      let c1 =
+        match Pool.acquire pool with
+        | Ok c -> c
+        | Error e -> failwith (Pool.error_to_string e)
+      in
+      let c2 =
+        match Pool.acquire pool with
+        | Ok c -> c
+        | Error e -> failwith (Pool.error_to_string e)
+      in
+      let c3 =
+        match Pool.acquire pool with
+        | Ok c -> c
+        | Error e -> failwith (Pool.error_to_string e)
+      in
+      Alcotest.(check int) "3 in use" 3 (Pool.in_use pool);
+      (match Pool.acquire pool with
+      | Error Pool.Pool_empty -> ()
+      | _ -> Alcotest.fail "Expected Pool_empty");
+      Pool.release pool c1;
+      Pool.release pool c2;
+      Pool.release pool c3)
+
+let test_pool_reuse =
+  with_pool (fun pool ->
+      let c1 =
+        match Pool.acquire pool with
+        | Ok c -> c
+        | Error e -> failwith (Pool.error_to_string e)
+      in
+      Pool.release pool c1;
+      Alcotest.(check int) "1 available" 1 (Pool.available pool);
+      let _c2 =
+        match Pool.acquire pool with
+        | Ok c -> c
+        | Error e -> failwith (Pool.error_to_string e)
+      in
+      Alcotest.(check int) "still 1 total" 1 (Pool.size pool))
+
+let test_pool_validation () =
+  let validation_count = Atomic.make 0 in
+  let config =
+    Pool.
+      {
+        max_size = 3;
+        connect =
+          (fun () ->
+            Repodb_postgresql.connect conninfo
+            |> Result.map_error Repodb_postgresql.error_message);
+        close = Repodb_postgresql.close;
+        validate =
+          Some
+            (fun conn ->
+              Atomic.incr validation_count;
+              match Repodb_postgresql.exec conn "SELECT 1" ~params:[||] with
+              | Ok () -> true
+              | Error _ -> false);
+      }
+  in
+  let pool = Pool.create config in
+  let c1 =
+    match Pool.acquire pool with
+    | Ok c -> c
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+  Pool.release pool c1;
+  let _c2 =
+    match Pool.acquire pool with
+    | Ok c -> c
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+  Alcotest.(check bool)
+    "validation was called" true
+    (Atomic.get validation_count > 0);
+  Pool.shutdown pool
+
+let test_pool_concurrent_domains () =
+  let config =
+    Pool.
+      {
+        max_size = 8;
+        connect =
+          (fun () ->
+            Repodb_postgresql.connect conninfo
+            |> Result.map_error Repodb_postgresql.error_message);
+        close = Repodb_postgresql.close;
+        validate = None;
+      }
+  in
+  let pool = Pool.create config in
+
+  (match Pool.acquire pool with
+  | Error e -> failwith (Pool.error_to_string e)
+  | Ok conn ->
+      let _ =
+        Repodb_postgresql.exec conn "DROP TABLE IF EXISTS counter" ~params:[||]
+      in
+      let _ =
+        Repodb_postgresql.exec conn
+          "CREATE TABLE counter (id INTEGER PRIMARY KEY, value INTEGER)"
+          ~params:[||]
+      in
+      let _ =
+        Repodb_postgresql.exec conn
+          "INSERT INTO counter (id, value) VALUES (1, 0)" ~params:[||]
+      in
+      Pool.release pool conn);
+
+  let n_domains = 4 in
+  let n_ops = 25 in
+  let completed = Atomic.make 0 in
+
+  let domain_work () =
+    for _ = 1 to n_ops do
+      match Pool.acquire pool with
+      | Ok conn ->
+          let _ =
+            Repodb_postgresql.exec conn
+              "UPDATE counter SET value = value + 1 WHERE id = 1" ~params:[||]
+          in
+          Pool.release pool conn
+      | Error e -> failwith (Pool.error_to_string e)
+    done;
+    Atomic.incr completed
+  in
+
+  let domains = List.init n_domains (fun _ -> Domain.spawn domain_work) in
+  List.iter Domain.join domains;
+
+  Alcotest.(check int) "all domains completed" n_domains (Atomic.get completed);
+
+  let final =
+    match Pool.acquire pool with
+    | Ok conn ->
+        (match
+           Repodb_postgresql.query conn "SELECT value FROM counter WHERE id = 1"
+             ~params:[||]
+         with
+          | Ok [ row ] -> Repodb.Driver.row_int row 0
+          | _ -> failwith "Query failed")
+        |> fun v ->
+        Pool.release pool conn;
+        v
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+
+  Alcotest.(check int) "all increments" (n_domains * n_ops) final;
+
+  (match Pool.acquire pool with
+  | Ok conn ->
+      let _ =
+        Repodb_postgresql.exec conn "DROP TABLE IF EXISTS counter" ~params:[||]
+      in
+      Pool.release pool conn
+  | Error _ -> ());
+  Pool.shutdown pool
+
+let pool_tests =
+  [
+    ("acquire_release", `Quick, test_pool_acquire_release);
+    ("with_connection", `Quick, test_pool_with_connection);
+    ("max_size", `Quick, test_pool_max_size);
+    ("reuse", `Quick, test_pool_reuse);
+    ("validation", `Quick, test_pool_validation);
+    ("concurrent_domains", `Slow, test_pool_concurrent_domains);
+  ]
+
 let () =
   Alcotest.run "repodb-postgresql"
     [
···
       ("Preload", preload_tests);
       ("Multi", multi_tests);
       ("Query-Repo", query_repo_tests);
+      ("Pool", pool_tests);
     ]

driver-sqlite/test_integration.ml

···
     ("complex", `Quick, test_query_complex);
   ]
 
+module Pool = Repodb.Pool
+
+let pool_db_path = "/tmp/repodb_pool_sqlite_test.db"
+
+let pool_config () =
+  Pool.
+    {
+      max_size = 3;
+      connect =
+        (fun () ->
+          match Repodb_sqlite.connect pool_db_path with
+          | Error e -> Error (Repodb_sqlite.error_message e)
+          | Ok conn ->
+              let _ =
+                Repodb_sqlite.exec conn "PRAGMA journal_mode=WAL" ~params:[||]
+              in
+              let _ =
+                Repodb_sqlite.exec conn "PRAGMA busy_timeout=5000" ~params:[||]
+              in
+              Ok conn);
+      close = Repodb_sqlite.close;
+      validate =
+        Some
+          (fun conn ->
+            match Repodb_sqlite.exec conn "SELECT 1" ~params:[||] with
+            | Ok () -> true
+            | Error _ -> false);
+    }
+
+let with_pool f () =
+  (try Sys.remove pool_db_path with Sys_error _ -> ());
+  let pool = Pool.create (pool_config ()) in
+  Fun.protect
+    ~finally:(fun () ->
+      Pool.shutdown pool;
+      try Sys.remove pool_db_path with Sys_error _ -> ())
+    (fun () -> f pool)
+
+let test_pool_acquire_release =
+  with_pool (fun pool ->
+      match Pool.acquire pool with
+      | Error e -> Alcotest.fail (Pool.error_to_string e)
+      | Ok conn ->
+          (match Repodb_sqlite.exec conn "SELECT 1" ~params:[||] with
+          | Ok () -> ()
+          | Error e -> Alcotest.fail (Repodb_sqlite.error_message e));
+          Pool.release pool conn;
+          Alcotest.(check int) "in_use after release" 0 (Pool.in_use pool))
+
+let test_pool_with_connection =
+  with_pool (fun pool ->
+      match Pool.acquire pool with
+      | Error e -> Alcotest.fail (Pool.error_to_string e)
+      | Ok conn -> (
+          let _ =
+            Repodb_sqlite.exec conn
+              "CREATE TABLE pool_test (id INTEGER PRIMARY KEY, value TEXT)"
+              ~params:[||]
+          in
+          Pool.release pool conn;
+
+          let result =
+            Pool.with_connection pool (fun conn ->
+                match
+                  Repodb_sqlite.exec conn
+                    "INSERT INTO pool_test (value) VALUES (?)"
+                    ~params:[| Repodb.Driver.Value.text "test" |]
+                with
+                | Ok () -> "inserted"
+                | Error e -> Repodb_sqlite.error_message e)
+          in
+          match result with
+          | Ok "inserted" -> ()
+          | Ok msg -> Alcotest.fail msg
+          | Error e -> Alcotest.fail (Pool.error_to_string e)))
+
+let test_pool_max_size =
+  with_pool (fun pool ->
+      let c1 =
+        match Pool.acquire pool with
+        | Ok c -> c
+        | Error e -> failwith (Pool.error_to_string e)
+      in
+      let c2 =
+        match Pool.acquire pool with
+        | Ok c -> c
+        | Error e -> failwith (Pool.error_to_string e)
+      in
+      let c3 =
+        match Pool.acquire pool with
+        | Ok c -> c
+        | Error e -> failwith (Pool.error_to_string e)
+      in
+      Alcotest.(check int) "3 in use" 3 (Pool.in_use pool);
+      (match Pool.acquire pool with
+      | Error Pool.Pool_empty -> ()
+      | _ -> Alcotest.fail "Expected Pool_empty");
+      Pool.release pool c1;
+      Pool.release pool c2;
+      Pool.release pool c3)
+
+let test_pool_reuse =
+  with_pool (fun pool ->
+      let c1 =
+        match Pool.acquire pool with
+        | Ok c -> c
+        | Error e -> failwith (Pool.error_to_string e)
+      in
+      Pool.release pool c1;
+      Alcotest.(check int) "1 available" 1 (Pool.available pool);
+      let _c2 =
+        match Pool.acquire pool with
+        | Ok c -> c
+        | Error e -> failwith (Pool.error_to_string e)
+      in
+      Alcotest.(check int) "still 1 total" 1 (Pool.size pool))
+
+let test_pool_validation () =
+  (try Sys.remove pool_db_path with Sys_error _ -> ());
+  let validation_count = Atomic.make 0 in
+  let config =
+    Pool.
+      {
+        max_size = 3;
+        connect =
+          (fun () ->
+            match Repodb_sqlite.connect pool_db_path with
+            | Error e -> Error (Repodb_sqlite.error_message e)
+            | Ok conn ->
+                let _ =
+                  Repodb_sqlite.exec conn "PRAGMA journal_mode=WAL" ~params:[||]
+                in
+                Ok conn);
+        close = Repodb_sqlite.close;
+        validate =
+          Some
+            (fun conn ->
+              Atomic.incr validation_count;
+              match Repodb_sqlite.exec conn "SELECT 1" ~params:[||] with
+              | Ok () -> true
+              | Error _ -> false);
+      }
+  in
+  let pool = Pool.create config in
+  let c1 =
+    match Pool.acquire pool with
+    | Ok c -> c
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+  Pool.release pool c1;
+  let _c2 =
+    match Pool.acquire pool with
+    | Ok c -> c
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+  Alcotest.(check bool)
+    "validation was called" true
+    (Atomic.get validation_count > 0);
+  Pool.shutdown pool;
+  try Sys.remove pool_db_path with Sys_error _ -> ()
+
+let test_pool_concurrent_domains () =
+  (try Sys.remove pool_db_path with Sys_error _ -> ());
+  let config =
+    Pool.
+      {
+        max_size = 8;
+        connect =
+          (fun () ->
+            match Repodb_sqlite.connect pool_db_path with
+            | Error e -> Error (Repodb_sqlite.error_message e)
+            | Ok conn ->
+                let _ =
+                  Repodb_sqlite.exec conn "PRAGMA journal_mode=WAL" ~params:[||]
+                in
+                let _ =
+                  Repodb_sqlite.exec conn "PRAGMA busy_timeout=5000" ~params:[||]
+                in
+                Ok conn);
+        close = Repodb_sqlite.close;
+        validate = None;
+      }
+  in
+  let pool = Pool.create config in
+
+  (match Pool.acquire pool with
+  | Error e -> failwith (Pool.error_to_string e)
+  | Ok conn ->
+      let _ =
+        Repodb_sqlite.exec conn
+          "CREATE TABLE counter (id INTEGER PRIMARY KEY, value INTEGER)"
+          ~params:[||]
+      in
+      let _ =
+        Repodb_sqlite.exec conn "INSERT INTO counter (id, value) VALUES (1, 0)"
+          ~params:[||]
+      in
+      Pool.release pool conn);
+
+  let n_domains = 4 in
+  let n_ops = 25 in
+  let completed = Atomic.make 0 in
+
+  let domain_work () =
+    for _ = 1 to n_ops do
+      match Pool.acquire pool with
+      | Ok conn ->
+          let _ =
+            Repodb_sqlite.exec conn
+              "UPDATE counter SET value = value + 1 WHERE id = 1" ~params:[||]
+          in
+          Pool.release pool conn
+      | Error e -> failwith (Pool.error_to_string e)
+    done;
+    Atomic.incr completed
+  in
+
+  let domains = List.init n_domains (fun _ -> Domain.spawn domain_work) in
+  List.iter Domain.join domains;
+
+  Alcotest.(check int) "all domains completed" n_domains (Atomic.get completed);
+
+  let final =
+    match Pool.acquire pool with
+    | Ok conn ->
+        (match
+           Repodb_sqlite.query conn "SELECT value FROM counter WHERE id = 1"
+             ~params:[||]
+         with
+          | Ok [ row ] -> Repodb.Driver.row_int row 0
+          | _ -> failwith "Query failed")
+        |> fun v ->
+        Pool.release pool conn;
+        v
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+
+  Alcotest.(check int) "all increments" (n_domains * n_ops) final;
+  Pool.shutdown pool;
+  try Sys.remove pool_db_path with Sys_error _ -> ()
+
+let pool_tests =
+  [
+    ("acquire_release", `Quick, test_pool_acquire_release);
+    ("with_connection", `Quick, test_pool_with_connection);
+    ("max_size", `Quick, test_pool_max_size);
+    ("reuse", `Quick, test_pool_reuse);
+    ("validation", `Quick, test_pool_validation);
+    ("concurrent_domains", `Quick, test_pool_concurrent_domains);
+  ]
+
 let () =
   Alcotest.run "repodb-sqlite"
     [
···
       ("Preload", preload_tests);
       ("Multi", multi_tests);
       ("Query-Repo", query_repo_tests);
+      ("Pool", pool_tests);
     ]

dune-project

···
   (ptime (>= 1.1))
   (re (>= 1.11))
   (uuidm (>= 0.9))
+  (kcas (>= 0.7))
+  (kcas_data (>= 0.7))
   (alcotest (and (>= 1.7) :with-test))
-  (yojson (and (>= 2.0) :with-test)))
+  (yojson (and (>= 2.0) :with-test))
+  (lwt (and (>= 5.6) :with-test))
+  (eio (and (>= 1.0) :with-test))
+  (eio_main (and (>= 1.0) :with-test)))
  (tags
   (database postgresql sqlite ecto orm query-builder migrations)))
 

lib/dune

···
 (library
  (name repodb)
  (public_name repodb)
- (libraries ptime re uuidm)
+ (libraries ptime re uuidm unix kcas kcas_data)
  (modules
   error
   driver
···
   repo
   migration
   multi
-  stream))
+  stream
+  pool))

lib/pool.ml

···
+(** Lock-free connection pool using kcas for thread-safe operations.
+
+    Works with any concurrency library (Eio, Lwt, direct-style) because kcas
+    provides lock-free data structures independent of any specific scheduler.
+
+    {[
+      let config =
+        Pool.
+          {
+            max_size = 10;
+            connect = (fun () -> Repodb_postgresql.connect conninfo);
+            close = Repodb_postgresql.close;
+            validate = None;
+          }
+      in
+      let pool = Pool.create config in
+      Pool.with_connection pool (fun conn -> Repo.all conn ~table:users ~decode)
+    ]} *)
+
+open Kcas
+open Kcas_data
+
+type pool_error =
+  | Pool_empty
+  | Pool_closed
+  | Pool_timeout
+  | Connection_error of string
+
+type 'conn config = {
+  max_size : int;
+  connect : unit -> ('conn, string) result;
+  close : 'conn -> unit;
+  validate : ('conn -> bool) option;
+}
+
+type 'conn pooled = { conn : 'conn; created_at : float }
+
+type 'conn t = {
+  config : 'conn config;
+  available : 'conn pooled Queue.t;
+  total_count : int Loc.t;
+  in_use_count : int Loc.t;
+  closed : bool Loc.t;
+}
+
+let create config =
+  {
+    config;
+    available = Queue.create ();
+    total_count = Loc.make 0;
+    in_use_count = Loc.make 0;
+    closed = Loc.make false;
+  }
+
+let size t = Loc.get t.total_count
+let in_use t = Loc.get t.in_use_count
+let available t = Queue.length t.available
+let is_closed t = Loc.get t.closed
+
+let validate_connection t pooled =
+  match t.config.validate with
+  | None -> true
+  | Some validate -> validate pooled.conn
+
+let close_pooled t pooled =
+  t.config.close pooled.conn;
+  Loc.decr t.total_count
+
+let try_create_connection t =
+  let tx ~xt =
+    if Xt.get ~xt t.closed then `Closed
+    else
+      let total = Xt.get ~xt t.total_count in
+      if total >= t.config.max_size then `At_capacity
+      else begin
+        Xt.set ~xt t.total_count (total + 1);
+        Xt.set ~xt t.in_use_count (Xt.get ~xt t.in_use_count + 1);
+        `Can_create
+      end
+  in
+  match Xt.commit { tx } with
+  | `Closed -> Error Pool_closed
+  | `At_capacity -> Error Pool_empty
+  | `Can_create -> (
+      match t.config.connect () with
+      | Ok conn ->
+          let pooled = { conn; created_at = Unix.gettimeofday () } in
+          Ok pooled
+      | Error msg ->
+          Loc.decr t.total_count;
+          Loc.decr t.in_use_count;
+          Error (Connection_error msg))
+
+let rec acquire_from_pool t =
+  if Loc.get t.closed then Error Pool_closed
+  else
+    match Queue.take_opt t.available with
+    | None -> Error Pool_empty
+    | Some pooled ->
+        if validate_connection t pooled then begin
+          Loc.incr t.in_use_count;
+          Ok pooled.conn
+        end
+        else begin
+          close_pooled t pooled;
+          acquire_from_pool t
+        end
+
+let acquire t =
+  match acquire_from_pool t with
+  | Ok conn -> Ok conn
+  | Error Pool_empty -> (
+      match try_create_connection t with
+      | Ok pooled -> Ok pooled.conn
+      | Error e -> Error e)
+  | Error e -> Error e
+
+let rec acquire_blocking ?timeoutf t =
+  if Loc.get t.closed then Error Pool_closed
+  else
+    match acquire t with
+    | Ok conn -> Ok conn
+    | Error Pool_empty -> (
+        match timeoutf with
+        | Some timeout -> (
+            match Queue.take_blocking ~timeoutf:timeout t.available with
+            | exception Kcas.Timeout.Timeout -> Error Pool_timeout
+            | pooled ->
+                if validate_connection t pooled then begin
+                  Loc.incr t.in_use_count;
+                  Ok pooled.conn
+                end
+                else begin
+                  close_pooled t pooled;
+                  acquire_blocking ?timeoutf t
+                end)
+        | None ->
+            let pooled = Queue.take_blocking t.available in
+            if validate_connection t pooled then begin
+              Loc.incr t.in_use_count;
+              Ok pooled.conn
+            end
+            else begin
+              close_pooled t pooled;
+              acquire_blocking ?timeoutf t
+            end)
+    | Error e -> Error e
+
+let release t conn =
+  if Loc.get t.closed then t.config.close conn
+  else begin
+    Loc.decr t.in_use_count;
+    let pooled = { conn; created_at = Unix.gettimeofday () } in
+    Queue.add pooled t.available
+  end
+
+let with_connection t f =
+  match acquire t with
+  | Error e -> Error e
+  | Ok conn -> (
+      match f conn with
+      | result ->
+          release t conn;
+          Ok result
+      | exception exn ->
+          release t conn;
+          raise exn)
+
+let with_connection_blocking ?timeoutf t f =
+  match acquire_blocking ?timeoutf t with
+  | Error e -> Error e
+  | Ok conn -> (
+      match f conn with
+      | result ->
+          release t conn;
+          Ok result
+      | exception exn ->
+          release t conn;
+          raise exn)
+
+let drain t =
+  let rec loop () =
+    match Queue.take_opt t.available with
+    | None -> ()
+    | Some pooled ->
+        close_pooled t pooled;
+        loop ()
+  in
+  loop ()
+
+let shutdown t =
+  Loc.set t.closed true;
+  drain t
+
+type stats = { total : int; available : int; in_use : int; closed : bool }
+
+let stats t =
+  {
+    total = size t;
+    available = available t;
+    in_use = in_use t;
+    closed = is_closed t;
+  }
+
+let error_to_string = function
+  | Pool_empty -> "Pool empty: no connections available"
+  | Pool_closed -> "Pool closed"
+  | Pool_timeout -> "Timeout waiting for connection"
+  | Connection_error msg -> Printf.sprintf "Connection error: %s" msg
+
+module Make (D : Driver.S) = struct
+  type nonrec t = D.connection t
+  type conn = D.connection
+
+  let create ~max_size ~conninfo ?validate () =
+    let config =
+      {
+        max_size;
+        connect =
+          (fun () -> D.connect conninfo |> Result.map_error D.error_message);
+        close = D.close;
+        validate;
+      }
+    in
+    create config
+
+  let acquire = acquire
+  let acquire_blocking = acquire_blocking
+  let release = release
+  let with_connection = with_connection
+  let with_connection_blocking = with_connection_blocking
+  let drain = drain
+  let shutdown = shutdown
+  let stats = stats
+  let size = size
+  let available = available
+  let in_use = in_use
+  let is_closed = is_closed
+end

lib/stream.ml

···
     D.query_fold conn sql ~params ~init:[] ~f:(fun acc row -> f row :: acc)
     |> Result.map List.rev
 
-  let cursor_counter = ref 0
+  let cursor_counter = Atomic.make 0
 
   let generate_cursor_name () =
-    incr cursor_counter;
-    Printf.sprintf "repodb_cursor_%d" !cursor_counter
+    let n = Atomic.fetch_and_add cursor_counter 1 in
+    Printf.sprintf "repodb_cursor_%d" n
 
   let cursor_fold conn ~config sql ~params ~init ~f =
     let cursor_name = generate_cursor_name () in

repodb.opam

···
   "ptime" {>= "1.1"}
   "re" {>= "1.11"}
   "uuidm" {>= "0.9"}
+  "kcas" {>= "0.7"}
+  "kcas_data" {>= "0.7"}
   "alcotest" {>= "1.7" & with-test}
   "yojson" {>= "2.0" & with-test}
+  "lwt" {>= "5.6" & with-test}
+  "eio" {>= "1.0" & with-test}
+  "eio_main" {>= "1.0" & with-test}
   "odoc" {with-doc}
 ]
 build: [

test/dune

···
   test_migration
   test_multi
   test_stream
+  test_pool
   test_repodb))
+
+(test
+ (name test_pool_lwt)
+ (libraries repodb repodb-sqlite lwt lwt.unix)
+ (modules test_pool_lwt))
+
+(test
+ (name test_pool_eio)
+ (libraries repodb repodb-sqlite eio eio_main)
+ (modules test_pool_eio))

test/test_pool.ml

···
+open Repodb
+
+type mock_conn = { id : int; mutable closed : bool }
+
+let conn_counter = Atomic.make 0
+
+let mock_config ?(max_size = 3) ?(fail_connect = false) ?(validate = None) () =
+  Pool.
+    {
+      max_size;
+      connect =
+        (fun () ->
+          if fail_connect then Error "Connection failed"
+          else
+            let id = Atomic.fetch_and_add conn_counter 1 in
+            Ok { id; closed = false });
+      close = (fun conn -> conn.closed <- true);
+      validate;
+    }
+
+let test_create_pool () =
+  let pool = Pool.create (mock_config ()) in
+  Alcotest.(check int) "initial size" 0 (Pool.size pool);
+  Alcotest.(check int) "initial available" 0 (Pool.available pool);
+  Alcotest.(check int) "initial in_use" 0 (Pool.in_use pool);
+  Alcotest.(check bool) "not closed" false (Pool.is_closed pool)
+
+let test_acquire_creates_connection () =
+  let pool = Pool.create (mock_config ()) in
+  match Pool.acquire pool with
+  | Ok conn ->
+      Alcotest.(check int) "total is 1" 1 (Pool.size pool);
+      Alcotest.(check int) "in_use is 1" 1 (Pool.in_use pool);
+      Alcotest.(check bool) "conn not closed" false conn.closed;
+      Pool.release pool conn
+  | Error _ -> Alcotest.fail "Expected successful acquire"
+
+let test_release_returns_to_pool () =
+  let pool = Pool.create (mock_config ()) in
+  match Pool.acquire pool with
+  | Ok conn ->
+      Pool.release pool conn;
+      Alcotest.(check int) "available is 1" 1 (Pool.available pool);
+      Alcotest.(check int) "in_use is 0" 0 (Pool.in_use pool)
+  | Error _ -> Alcotest.fail "Expected successful acquire"
+
+let test_acquire_reuses_connection () =
+  let pool = Pool.create (mock_config ()) in
+  let conn1 =
+    match Pool.acquire pool with Ok c -> c | Error _ -> Alcotest.fail "fail"
+  in
+  let id1 = conn1.id in
+  Pool.release pool conn1;
+  let conn2 =
+    match Pool.acquire pool with Ok c -> c | Error _ -> Alcotest.fail "fail"
+  in
+  Alcotest.(check int) "same connection reused" id1 conn2.id;
+  Alcotest.(check int) "still only 1 total" 1 (Pool.size pool);
+  Pool.release pool conn2
+
+let test_max_size_enforced () =
+  let pool = Pool.create (mock_config ~max_size:2 ()) in
+  let c1 =
+    match Pool.acquire pool with Ok c -> c | Error _ -> Alcotest.fail "fail"
+  in
+  let c2 =
+    match Pool.acquire pool with Ok c -> c | Error _ -> Alcotest.fail "fail"
+  in
+  (match Pool.acquire pool with
+  | Error Pool.Pool_empty -> ()
+  | _ -> Alcotest.fail "Expected Pool_empty");
+  Pool.release pool c1;
+  Pool.release pool c2
+
+let test_with_connection () =
+  let pool = Pool.create (mock_config ()) in
+  let result =
+    Pool.with_connection pool (fun conn ->
+        Alcotest.(check int) "in_use during work" 1 (Pool.in_use pool);
+        conn.id + 100)
+  in
+  match result with
+  | Ok n ->
+      Alcotest.(check bool) "got result" true (n >= 100);
+      Alcotest.(check int) "released after" 0 (Pool.in_use pool)
+  | Error _ -> Alcotest.fail "Expected successful with_connection"
+
+let test_with_connection_releases_on_exception () =
+  let pool = Pool.create (mock_config ()) in
+  (try
+     let _ =
+       Pool.with_connection pool (fun _ ->
+           Alcotest.(check int) "in_use" 1 (Pool.in_use pool);
+           failwith "boom")
+     in
+     ()
+   with Failure _ -> ());
+  Alcotest.(check int) "released after exception" 0 (Pool.in_use pool)
+
+let test_connection_error () =
+  let pool = Pool.create (mock_config ~fail_connect:true ()) in
+  match Pool.acquire pool with
+  | Error (Pool.Connection_error msg) ->
+      Alcotest.(check string) "error message" "Connection failed" msg
+  | _ -> Alcotest.fail "Expected Connection_error"
+
+let test_validation () =
+  let call_count = ref 0 in
+  let validate conn =
+    incr call_count;
+    not conn.closed
+  in
+  let pool = Pool.create (mock_config ~validate:(Some validate) ()) in
+  let conn =
+    match Pool.acquire pool with Ok c -> c | Error _ -> Alcotest.fail "fail"
+  in
+  Pool.release pool conn;
+  let _ =
+    match Pool.acquire pool with Ok c -> c | Error _ -> Alcotest.fail "fail"
+  in
+  Alcotest.(check bool) "validate called" true (!call_count > 0)
+
+let test_shutdown () =
+  let pool = Pool.create (mock_config ()) in
+  let conn =
+    match Pool.acquire pool with Ok c -> c | Error _ -> Alcotest.fail "fail"
+  in
+  Pool.release pool conn;
+  Pool.shutdown pool;
+  Alcotest.(check bool) "pool closed" true (Pool.is_closed pool);
+  match Pool.acquire pool with
+  | Error Pool.Pool_closed -> ()
+  | _ -> Alcotest.fail "Expected Pool_closed"
+
+let test_drain () =
+  let pool = Pool.create (mock_config ()) in
+  let c1 =
+    match Pool.acquire pool with Ok c -> c | Error _ -> Alcotest.fail "fail"
+  in
+  let c2 =
+    match Pool.acquire pool with Ok c -> c | Error _ -> Alcotest.fail "fail"
+  in
+  Pool.release pool c1;
+  Pool.release pool c2;
+  Alcotest.(check int) "2 available before drain" 2 (Pool.available pool);
+  Pool.drain pool;
+  Alcotest.(check int) "0 available after drain" 0 (Pool.available pool);
+  Alcotest.(check int) "0 total after drain" 0 (Pool.size pool)
+
+let test_stats () =
+  let pool = Pool.create (mock_config ~max_size:5 ()) in
+  let conn =
+    match Pool.acquire pool with Ok c -> c | Error _ -> Alcotest.fail "fail"
+  in
+  let stats = Pool.stats pool in
+  Alcotest.(check int) "stats.total" 1 stats.total;
+  Alcotest.(check int) "stats.in_use" 1 stats.in_use;
+  Alcotest.(check int) "stats.available" 0 stats.available;
+  Alcotest.(check bool) "stats.closed" false stats.closed;
+  Pool.release pool conn
+
+let test_error_to_string () =
+  Alcotest.(check string)
+    "Pool_empty" "Pool empty: no connections available"
+    (Pool.error_to_string Pool.Pool_empty);
+  Alcotest.(check string)
+    "Pool_closed" "Pool closed"
+    (Pool.error_to_string Pool.Pool_closed);
+  Alcotest.(check string)
+    "Pool_timeout" "Timeout waiting for connection"
+    (Pool.error_to_string Pool.Pool_timeout);
+  Alcotest.(check string)
+    "Connection_error" "Connection error: oops"
+    (Pool.error_to_string (Pool.Connection_error "oops"))
+
+let tests =
+  [
+    ("create pool", `Quick, test_create_pool);
+    ("acquire creates connection", `Quick, test_acquire_creates_connection);
+    ("release returns to pool", `Quick, test_release_returns_to_pool);
+    ("acquire reuses connection", `Quick, test_acquire_reuses_connection);
+    ("max size enforced", `Quick, test_max_size_enforced);
+    ("with_connection", `Quick, test_with_connection);
+    ( "with_connection releases on exception",
+      `Quick,
+      test_with_connection_releases_on_exception );
+    ("connection error", `Quick, test_connection_error);
+    ("validation", `Quick, test_validation);
+    ("shutdown", `Quick, test_shutdown);
+    ("drain", `Quick, test_drain);
+    ("stats", `Quick, test_stats);
+    ("error_to_string", `Quick, test_error_to_string);
+  ]

test/test_pool_eio.ml

···
+(** Example: Using repodb Pool with Eio
+
+    This demonstrates how to use the lock-free connection pool with Eio's
+    direct-style effects. The pool uses kcas internally, which integrates
+    naturally with Eio's scheduler.
+
+    Key patterns shown:
+    - Creating a pool for use with Eio fibers
+    - Running blocking DB operations in a thread pool
+    - Concurrent database access from multiple Eio fibers
+    - Using domains for true parallelism *)
+
+module Pool = Repodb.Pool
+module Driver = Repodb_sqlite
+
+let db_path = "/tmp/repodb_pool_eio_test.db"
+
+let pool_config () =
+  Pool.
+    {
+      max_size = 5;
+      connect =
+        (fun () ->
+          match Driver.connect db_path with
+          | Error e -> Error (Driver.error_message e)
+          | Ok conn ->
+              let _ = Driver.exec conn "PRAGMA journal_mode=WAL" ~params:[||] in
+              let _ =
+                Driver.exec conn "PRAGMA busy_timeout=5000" ~params:[||]
+              in
+              Ok conn);
+      close = Driver.close;
+      validate = None;
+    }
+
+let setup_schema conn =
+  let exec sql =
+    match Driver.exec conn sql ~params:[||] with
+    | Ok () -> ()
+    | Error e -> failwith (Driver.error_message e)
+  in
+  exec
+    "CREATE TABLE IF NOT EXISTS counter (id INTEGER PRIMARY KEY, value INTEGER)";
+  exec "DELETE FROM counter";
+  exec "INSERT INTO counter (id, value) VALUES (1, 0)"
+
+let increment conn =
+  match
+    Driver.exec conn "UPDATE counter SET value = value + 1 WHERE id = 1"
+      ~params:[||]
+  with
+  | Ok () -> ()
+  | Error e -> failwith (Driver.error_message e)
+
+let get_value conn =
+  match
+    Driver.query conn "SELECT value FROM counter WHERE id = 1" ~params:[||]
+  with
+  | Ok [ row ] -> Repodb.Driver.row_int row 0
+  | Ok _ -> failwith "Expected one row"
+  | Error e -> failwith (Driver.error_message e)
+
+let test_concurrent_fibers ~executor pool =
+  Eio.traceln "=== Test: Concurrent fibers ===";
+
+  (match Pool.acquire pool with
+  | Ok conn ->
+      setup_schema conn;
+      Pool.release pool conn
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let n_fibers = 50 in
+  let n_increments = 10 in
+
+  Eio.Fiber.all
+    (List.init n_fibers (fun i () ->
+         for _ = 1 to n_increments do
+           Eio.Executor_pool.submit_exn executor ~weight:1.0 (fun () ->
+               match Pool.acquire pool with
+               | Ok conn ->
+                   increment conn;
+                   Pool.release pool conn
+               | Error e -> failwith (Pool.error_to_string e))
+         done;
+         if (i + 1) mod 10 = 0 then Eio.traceln "  Fiber %d completed" (i + 1)));
+
+  let final =
+    match Pool.acquire pool with
+    | Ok conn ->
+        let v = get_value conn in
+        Pool.release pool conn;
+        v
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+
+  let expected = n_fibers * n_increments in
+  Eio.traceln "  Final counter: %d (expected %d)" final expected;
+  assert (final = expected);
+  Eio.traceln "  PASSED"
+
+let test_blocking_acquire ~clock pool =
+  Eio.traceln "=== Test: Blocking acquire ===";
+
+  let acquired = ref [] in
+  for _ = 1 to 5 do
+    match Pool.acquire pool with
+    | Ok conn -> acquired := conn :: !acquired
+    | Error e -> failwith (Pool.error_to_string e)
+  done;
+  Eio.traceln "  Acquired all 5 connections";
+
+  (match Pool.acquire pool with
+  | Error Pool.Pool_empty -> Eio.traceln "  Pool correctly reports empty"
+  | _ -> failwith "Expected Pool_empty");
+
+  let got_connection = Atomic.make false in
+
+  let rec poll_acquire timeout =
+    if timeout <= 0.0 then None
+    else
+      match Pool.acquire pool with
+      | Ok conn -> Some conn
+      | Error Pool.Pool_empty ->
+          Eio.Time.sleep clock 0.01;
+          poll_acquire (timeout -. 0.01)
+      | Error e -> failwith (Pool.error_to_string e)
+  in
+
+  Eio.Fiber.both
+    (fun () ->
+      Eio.Time.sleep clock 0.05;
+      Eio.traceln "  Releasing one connection...";
+      match !acquired with
+      | conn :: rest ->
+          Pool.release pool conn;
+          acquired := rest
+      | [] -> ())
+    (fun () ->
+      Eio.traceln "  Waiting for connection (polling)...";
+      match poll_acquire 1.0 with
+      | Some conn ->
+          Eio.traceln "  Got connection!";
+          Atomic.set got_connection true;
+          Pool.release pool conn
+      | None -> failwith "Timeout waiting for connection");
+
+  assert (Atomic.get got_connection);
+  List.iter (Pool.release pool) !acquired;
+  Eio.traceln "  PASSED"
+
+let test_with_connection pool =
+  Eio.traceln "=== Test: with_connection pattern ===";
+
+  (match Pool.acquire pool with
+  | Ok conn ->
+      setup_schema conn;
+      Pool.release pool conn
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let result =
+    Pool.with_connection pool (fun conn ->
+        increment conn;
+        increment conn;
+        get_value conn)
+  in
+
+  (match result with
+  | Ok value -> Eio.traceln "  Counter after 2 increments: %d" value
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let stats = Pool.stats pool in
+  Eio.traceln "  Stats: total=%d in_use=%d available=%d" stats.total
+    stats.in_use stats.available;
+  assert (stats.in_use = 0);
+  Eio.traceln "  PASSED"
+
+let test_multi_domain ~executor pool =
+  Eio.traceln "=== Test: Multi-domain parallelism ===";
+
+  (match Pool.acquire pool with
+  | Ok conn ->
+      setup_schema conn;
+      Pool.release pool conn
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let n_domains = 4 in
+  let n_ops = 25 in
+
+  let domain_work domain_id =
+    for _ = 1 to n_ops do
+      match Pool.acquire pool with
+      | Ok conn ->
+          increment conn;
+          Pool.release pool conn
+      | Error e -> failwith (Pool.error_to_string e)
+    done;
+    Eio.traceln "  Domain %d completed %d operations" domain_id n_ops
+  in
+
+  Eio.Fiber.all
+    (List.init n_domains (fun i () ->
+         Eio.Executor_pool.submit_exn executor ~weight:1.0 (fun () ->
+             domain_work i)));
+
+  let final =
+    match Pool.acquire pool with
+    | Ok conn ->
+        let v = get_value conn in
+        Pool.release pool conn;
+        v
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+
+  let expected = n_domains * n_ops in
+  Eio.traceln "  Final counter: %d (expected %d)" final expected;
+  assert (final = expected);
+  Eio.traceln "  PASSED"
+
+let test_pool_stats pool =
+  Eio.traceln "=== Test: Pool statistics ===";
+
+  let print_stats label =
+    let s = Pool.stats pool in
+    Eio.traceln "  [%s] total=%d in_use=%d available=%d" label s.total s.in_use
+      s.available
+  in
+
+  Pool.drain pool;
+  print_stats "after drain";
+
+  let c1 =
+    match Pool.acquire pool with
+    | Ok c -> c
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+  print_stats "after 1st acquire";
+
+  let c2 =
+    match Pool.acquire pool with
+    | Ok c -> c
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+  print_stats "after 2nd acquire";
+
+  Pool.release pool c1;
+  print_stats "after 1st release";
+
+  Pool.release pool c2;
+  print_stats "after 2nd release";
+
+  Eio.traceln "  PASSED"
+
+let setup_streaming_data conn =
+  let exec sql =
+    match Driver.exec conn sql ~params:[||] with
+    | Ok () -> ()
+    | Error e -> failwith (Driver.error_message e)
+  in
+  exec
+    "CREATE TABLE IF NOT EXISTS numbers (id INTEGER PRIMARY KEY, value INTEGER)";
+  exec "DELETE FROM numbers";
+  for i = 1 to 1000 do
+    let _ =
+      Driver.exec conn
+        (Printf.sprintf "INSERT INTO numbers (value) VALUES (%d)" i)
+        ~params:[||]
+    in
+    ()
+  done
+
+let test_streaming_fold ~executor pool =
+  Eio.traceln "=== Test: Streaming fold with Pool ===";
+
+  (match Pool.acquire pool with
+  | Ok conn ->
+      setup_streaming_data conn;
+      Pool.release pool conn
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let n_fibers = 10 in
+  let results = Array.make n_fibers 0 in
+
+  Eio.Fiber.all
+    (List.init n_fibers (fun id () ->
+         Eio.Executor_pool.submit_exn executor ~weight:1.0 (fun () ->
+             match Pool.acquire pool with
+             | Ok conn ->
+                 let sum =
+                   match
+                     Driver.query_fold conn "SELECT value FROM numbers"
+                       ~params:[||] ~init:0 ~f:(fun acc row ->
+                         acc + Repodb.Driver.row_int row 0)
+                   with
+                   | Ok s -> s
+                   | Error e -> failwith (Driver.error_message e)
+                 in
+                 results.(id) <- sum;
+                 Pool.release pool conn
+             | Error e -> failwith (Pool.error_to_string e))));
+
+  let expected_sum = 1000 * 1001 / 2 in
+  Array.iteri
+    (fun i sum ->
+      if sum <> expected_sum then
+        failwith
+          (Printf.sprintf "Fiber %d got sum %d, expected %d" i sum expected_sum))
+    results;
+
+  Eio.traceln "  All %d fibers computed correct sum: %d" n_fibers expected_sum;
+  Eio.traceln "  PASSED"
+
+let test_streaming_iter ~executor pool =
+  Eio.traceln "=== Test: Streaming iter with Pool ===";
+
+  (match Pool.acquire pool with
+  | Ok conn ->
+      setup_streaming_data conn;
+      Pool.release pool conn
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let count = Atomic.make 0 in
+  let sum = Atomic.make 0 in
+  let n_fibers = 5 in
+
+  Eio.Fiber.all
+    (List.init n_fibers (fun _id () ->
+         Eio.Executor_pool.submit_exn executor ~weight:1.0 (fun () ->
+             match Pool.acquire pool with
+             | Ok conn ->
+                 (match
+                    Driver.query_iter conn
+                      "SELECT value FROM numbers WHERE value <= 100" ~params:[||]
+                      ~f:(fun row ->
+                        Atomic.incr count;
+                        let _ =
+                          Atomic.fetch_and_add sum (Repodb.Driver.row_int row 0)
+                        in
+                        ())
+                  with
+                 | Ok () -> ()
+                 | Error e -> failwith (Driver.error_message e));
+                 Pool.release pool conn
+             | Error e -> failwith (Pool.error_to_string e))));
+
+  let expected_sum = n_fibers * (100 * 101) / 2 in
+  let actual_sum = Atomic.get sum in
+  Eio.traceln "  Total rows processed: %d" (Atomic.get count);
+  Eio.traceln "  Combined sum: %d (expected %d)" actual_sum expected_sum;
+
+  if actual_sum <> expected_sum then
+    failwith
+      (Printf.sprintf "Sum mismatch: got %d, expected %d" actual_sum
+         expected_sum);
+
+  Eio.traceln "  PASSED"
+
+let test_streaming_concurrent_domains ~executor pool =
+  Eio.traceln "=== Test: Streaming across domains ===";
+
+  (match Pool.acquire pool with
+  | Ok conn ->
+      setup_streaming_data conn;
+      Pool.release pool conn
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let n_domains = 4 in
+  let results = Array.make n_domains 0 in
+
+  Eio.Fiber.all
+    (List.init n_domains (fun id () ->
+         Eio.Executor_pool.submit_exn executor ~weight:1.0 (fun () ->
+             match Pool.acquire pool with
+             | Ok conn ->
+                 let sum =
+                   match
+                     Driver.query_fold conn
+                       "SELECT value FROM numbers WHERE value > 500"
+                       ~params:[||] ~init:0 ~f:(fun acc row ->
+                         acc + Repodb.Driver.row_int row 0)
+                   with
+                   | Ok s -> s
+                   | Error e -> failwith (Driver.error_message e)
+                 in
+                 results.(id) <- sum;
+                 Pool.release pool conn
+             | Error e -> failwith (Pool.error_to_string e))));
+
+  let expected_sum =
+    List.init 500 (fun i -> 501 + i) |> List.fold_left ( + ) 0
+  in
+  Array.iteri
+    (fun i sum ->
+      if sum <> expected_sum then
+        failwith
+          (Printf.sprintf "Domain %d got sum %d, expected %d" i sum expected_sum))
+    results;
+
+  Eio.traceln "  All %d domains computed correct sum: %d" n_domains expected_sum;
+  Eio.traceln "  PASSED"
+
+let () =
+  Eio.traceln "========================================";
+  Eio.traceln "repodb Pool + Eio Integration Tests";
+  Eio.traceln "========================================";
+
+  (try Sys.remove db_path with Sys_error _ -> ());
+
+  Eio_main.run @@ fun env ->
+  let clock = Eio.Stdenv.clock env in
+  let domain_mgr = Eio.Stdenv.domain_mgr env in
+
+  Eio.Switch.run @@ fun sw ->
+  let executor = Eio.Executor_pool.create ~sw domain_mgr ~domain_count:4 in
+  let pool = Pool.create (pool_config ()) in
+
+  test_concurrent_fibers ~executor pool;
+  test_blocking_acquire ~clock pool;
+  test_with_connection pool;
+  test_multi_domain ~executor pool;
+  test_pool_stats pool;
+  test_streaming_fold ~executor pool;
+  test_streaming_iter ~executor pool;
+  test_streaming_concurrent_domains ~executor pool;
+
+  Pool.shutdown pool;
+
+  (try Sys.remove db_path with Sys_error _ -> ());
+
+  Eio.traceln "========================================";
+  Eio.traceln "All Eio pool tests PASSED!";
+  Eio.traceln "========================================"

test/test_pool_lwt.ml

···
+(** Example: Using repodb Pool with Lwt
+
+    This demonstrates how to use the lock-free connection pool with Lwt's
+    cooperative threading model. The pool uses kcas internally, which is
+    scheduler-agnostic, so it works seamlessly with Lwt.
+
+    Key patterns shown:
+    - Creating a pool with Lwt
+    - Using Lwt_preemptive.detach for blocking pool operations
+    - Concurrent database access from multiple Lwt threads
+    - Pool exhaustion and blocking acquire with timeout *)
+
+open Lwt.Infix
+module Pool = Repodb.Pool
+module Driver = Repodb_sqlite
+
+let db_path = "/tmp/repodb_pool_lwt_test.db"
+
+let pool_config () =
+  Pool.
+    {
+      max_size = 5;
+      connect =
+        (fun () ->
+          match Driver.connect db_path with
+          | Error e -> Error (Driver.error_message e)
+          | Ok conn ->
+              let _ = Driver.exec conn "PRAGMA journal_mode=WAL" ~params:[||] in
+              let _ =
+                Driver.exec conn "PRAGMA busy_timeout=5000" ~params:[||]
+              in
+              Ok conn);
+      close = Driver.close;
+      validate = None;
+    }
+
+let setup_schema conn =
+  let exec sql =
+    match Driver.exec conn sql ~params:[||] with
+    | Ok () -> ()
+    | Error e -> failwith (Driver.error_message e)
+  in
+  exec "PRAGMA journal_mode=WAL";
+  exec "PRAGMA busy_timeout=5000";
+  exec
+    "CREATE TABLE IF NOT EXISTS counter (id INTEGER PRIMARY KEY, value INTEGER)";
+  exec "DELETE FROM counter";
+  exec "INSERT INTO counter (id, value) VALUES (1, 0)"
+
+let increment conn =
+  match
+    Driver.exec conn "UPDATE counter SET value = value + 1 WHERE id = 1"
+      ~params:[||]
+  with
+  | Ok () -> ()
+  | Error e -> failwith (Driver.error_message e)
+
+let get_value conn =
+  match
+    Driver.query conn "SELECT value FROM counter WHERE id = 1" ~params:[||]
+  with
+  | Ok [ row ] -> Repodb.Driver.row_int row 0
+  | Ok _ -> failwith "Expected one row"
+  | Error e -> failwith (Driver.error_message e)
+
+let test_concurrent_increments () =
+  Printf.printf "\n=== Test: Concurrent increments with Lwt ===\n%!";
+
+  let pool = Pool.create (pool_config ()) in
+
+  (match Pool.acquire pool with
+  | Ok conn ->
+      setup_schema conn;
+      Pool.release pool conn
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let n_tasks = 50 in
+  let n_increments = 10 in
+
+  let worker id =
+    let rec loop remaining =
+      if remaining <= 0 then Lwt.return_unit
+      else
+        Lwt_preemptive.detach
+          (fun () ->
+            match Pool.acquire pool with
+            | Ok conn ->
+                increment conn;
+                Pool.release pool conn
+            | Error e -> failwith (Pool.error_to_string e))
+          ()
+        >>= fun () ->
+        if remaining mod 5 = 0 then
+          Printf.printf "  Worker %d: %d increments remaining\n%!" id remaining;
+        loop (remaining - 1)
+    in
+    loop n_increments
+  in
+
+  Lwt_main.run (Lwt.join (List.init n_tasks worker));
+
+  let final =
+    match Pool.acquire pool with
+    | Ok conn ->
+        let v = get_value conn in
+        Pool.release pool conn;
+        v
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+
+  let expected = n_tasks * n_increments in
+  Printf.printf "  Final counter value: %d (expected %d)\n%!" final expected;
+  assert (final = expected);
+
+  Pool.shutdown pool;
+  Printf.printf "  PASSED\n%!"
+
+let test_blocking_acquire () =
+  Printf.printf "\n=== Test: Blocking acquire with Lwt ===\n%!";
+
+  let pool = Pool.create (pool_config ()) in
+  let acquired = ref [] in
+
+  for _ = 1 to 5 do
+    match Pool.acquire pool with
+    | Ok conn -> acquired := conn :: !acquired
+    | Error e -> failwith (Pool.error_to_string e)
+  done;
+  Printf.printf "  Acquired all 5 connections\n%!";
+
+  (match Pool.acquire pool with
+  | Error Pool.Pool_empty -> Printf.printf "  Pool correctly reports empty\n%!"
+  | _ -> failwith "Expected Pool_empty");
+
+  let success = ref false in
+
+  let release_task =
+    Lwt_unix.sleep 0.05 >>= fun () ->
+    Printf.printf "  Releasing one connection...\n%!";
+    (match !acquired with
+    | conn :: rest ->
+        Pool.release pool conn;
+        acquired := rest
+    | [] -> ());
+    Lwt.return_unit
+  in
+
+  let rec poll_acquire timeout =
+    if timeout <= 0.0 then Lwt.return_none
+    else
+      match Pool.acquire pool with
+      | Ok conn -> Lwt.return_some conn
+      | Error Pool.Pool_empty ->
+          Lwt_unix.sleep 0.01 >>= fun () -> poll_acquire (timeout -. 0.01)
+      | Error e -> failwith (Pool.error_to_string e)
+  in
+
+  let acquire_task =
+    Printf.printf "  Waiting for connection (polling)...\n%!";
+    poll_acquire 1.0 >>= function
+    | Some conn ->
+        Printf.printf "  Got connection after wait!\n%!";
+        success := true;
+        Pool.release pool conn;
+        Lwt.return_unit
+    | None -> failwith "Timeout waiting for connection"
+  in
+
+  Lwt_main.run (Lwt.join [ release_task; acquire_task ]);
+  assert !success;
+
+  List.iter (Pool.release pool) !acquired;
+  Pool.shutdown pool;
+  Printf.printf "  PASSED\n%!"
+
+let test_with_connection_pattern () =
+  Printf.printf "\n=== Test: with_connection pattern ===\n%!";
+
+  let pool = Pool.create (pool_config ()) in
+
+  (match Pool.acquire pool with
+  | Ok conn ->
+      setup_schema conn;
+      Pool.release pool conn
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let result =
+    Pool.with_connection pool (fun conn ->
+        increment conn;
+        increment conn;
+        get_value conn)
+  in
+
+  (match result with
+  | Ok value -> Printf.printf "  Counter after 2 increments: %d\n%!" value
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let stats = Pool.stats pool in
+  Printf.printf "  Pool stats: total=%d, in_use=%d, available=%d\n%!"
+    stats.total stats.in_use stats.available;
+  assert (stats.in_use = 0);
+
+  Pool.shutdown pool;
+  Printf.printf "  PASSED\n%!"
+
+let test_pool_stats () =
+  Printf.printf "\n=== Test: Pool statistics ===\n%!";
+
+  let pool = Pool.create (pool_config ()) in
+
+  let print_stats label =
+    let s = Pool.stats pool in
+    Printf.printf "  [%s] total=%d in_use=%d available=%d\n%!" label s.total
+      s.in_use s.available
+  in
+
+  print_stats "initial";
+  assert ((Pool.stats pool).total = 0);
+
+  let c1 =
+    match Pool.acquire pool with
+    | Ok c -> c
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+  print_stats "after 1st acquire";
+  assert ((Pool.stats pool).in_use = 1);
+
+  let c2 =
+    match Pool.acquire pool with
+    | Ok c -> c
+    | Error e -> failwith (Pool.error_to_string e)
+  in
+  print_stats "after 2nd acquire";
+  assert ((Pool.stats pool).in_use = 2);
+
+  Pool.release pool c1;
+  print_stats "after 1st release";
+  assert ((Pool.stats pool).in_use = 1);
+  assert ((Pool.stats pool).available = 1);
+
+  Pool.release pool c2;
+  print_stats "after 2nd release";
+  assert ((Pool.stats pool).in_use = 0);
+
+  Pool.shutdown pool;
+  Printf.printf "  PASSED\n%!"
+
+let setup_streaming_data conn =
+  let exec sql =
+    match Driver.exec conn sql ~params:[||] with
+    | Ok () -> ()
+    | Error e -> failwith (Driver.error_message e)
+  in
+  exec
+    "CREATE TABLE IF NOT EXISTS numbers (id INTEGER PRIMARY KEY, value INTEGER)";
+  exec "DELETE FROM numbers";
+  for i = 1 to 1000 do
+    let _ =
+      Driver.exec conn
+        (Printf.sprintf "INSERT INTO numbers (value) VALUES (%d)" i)
+        ~params:[||]
+    in
+    ()
+  done
+
+let test_streaming_with_pool () =
+  Printf.printf "\n=== Test: Streaming with Pool (Lwt) ===\n%!";
+
+  let pool = Pool.create (pool_config ()) in
+
+  (match Pool.acquire pool with
+  | Ok conn ->
+      setup_streaming_data conn;
+      Pool.release pool conn
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let n_workers = 10 in
+  let results = Array.make n_workers 0 in
+
+  let worker id =
+    Lwt_preemptive.detach
+      (fun () ->
+        match Pool.acquire pool with
+        | Ok conn ->
+            let sum =
+              match
+                Driver.query_fold conn "SELECT value FROM numbers" ~params:[||]
+                  ~init:0 ~f:(fun acc row -> acc + Repodb.Driver.row_int row 0)
+              with
+              | Ok s -> s
+              | Error e -> failwith (Driver.error_message e)
+            in
+            results.(id) <- sum;
+            Pool.release pool conn
+        | Error e -> failwith (Pool.error_to_string e))
+      ()
+  in
+
+  Lwt_main.run (Lwt.join (List.init n_workers worker));
+
+  let expected_sum = 1000 * 1001 / 2 in
+  Array.iteri
+    (fun i sum ->
+      if sum <> expected_sum then
+        failwith
+          (Printf.sprintf "Worker %d got sum %d, expected %d" i sum expected_sum))
+    results;
+
+  Printf.printf "  All %d workers computed correct sum: %d\n%!" n_workers
+    expected_sum;
+
+  Pool.shutdown pool;
+  Printf.printf "  PASSED\n%!"
+
+let test_streaming_iter_with_pool () =
+  Printf.printf "\n=== Test: Streaming iter with Pool (Lwt) ===\n%!";
+
+  let pool = Pool.create (pool_config ()) in
+
+  (match Pool.acquire pool with
+  | Ok conn ->
+      setup_streaming_data conn;
+      Pool.release pool conn
+  | Error e -> failwith (Pool.error_to_string e));
+
+  let count = Atomic.make 0 in
+  let sum = Atomic.make 0 in
+
+  let n_workers = 5 in
+
+  let worker _id =
+    Lwt_preemptive.detach
+      (fun () ->
+        match Pool.acquire pool with
+        | Ok conn ->
+            (match
+               Driver.query_iter conn
+                 "SELECT value FROM numbers WHERE value <= 100" ~params:[||]
+                 ~f:(fun row ->
+                   Atomic.incr count;
+                   let _ =
+                     Atomic.fetch_and_add sum (Repodb.Driver.row_int row 0)
+                   in
+                   ())
+             with
+            | Ok () -> ()
+            | Error e -> failwith (Driver.error_message e));
+            Pool.release pool conn
+        | Error e -> failwith (Pool.error_to_string e))
+      ()
+  in
+
+  Lwt_main.run (Lwt.join (List.init n_workers worker));
+
+  let expected_sum = n_workers * (100 * 101) / 2 in
+  let actual_sum = Atomic.get sum in
+  Printf.printf "  Total rows processed: %d\n%!" (Atomic.get count);
+  Printf.printf "  Combined sum: %d (expected %d)\n%!" actual_sum expected_sum;
+
+  if actual_sum <> expected_sum then
+    failwith
+      (Printf.sprintf "Sum mismatch: got %d, expected %d" actual_sum
+         expected_sum);
+
+  Pool.shutdown pool;
+  Printf.printf "  PASSED\n%!"
+
+let () =
+  Printf.printf "========================================\n%!";
+  Printf.printf "repodb Pool + Lwt Integration Tests\n%!";
+  Printf.printf "========================================\n%!";
+
+  (try Sys.remove db_path with Sys_error _ -> ());
+
+  test_concurrent_increments ();
+  test_blocking_acquire ();
+  test_with_connection_pattern ();
+  test_pool_stats ();
+  test_streaming_with_pool ();
+  test_streaming_iter_with_pool ();
+
+  (try Sys.remove db_path with Sys_error _ -> ());
+
+  Printf.printf "\n========================================\n%!";
+  Printf.printf "All Lwt pool tests PASSED!\n%!";
+  Printf.printf "========================================\n%!"

test/test_repodb.ml

···
       ("Migration", Test_migration.tests);
       ("Multi", Test_multi.tests);
       ("Stream", Test_stream.tests);
+      ("Pool", Test_pool.tests);
     ]