feat: add CQRS module and MultiPool for read/write splitting

- Add Pool.Make(Driver).Multi submodule with lock-free round-robin load
balancing using kcas, health tracking, and full pool API
- Add Cqrs.Make(Driver) functor for transparent read/write splitting:
- Automatic routing: reads to replicas, writes to primary
- Three replica selection strategies: RoundRobin, Random, LeastConnections
- Transaction pinning to primary connection
- Automatic fallback when all replicas unhealthy
- Add comprehensive unit tests (test_multi_pool.ml, test_cqrs.ml)
- Add Podman-based integration tests with PostgreSQL replication
- Add documentation (docs/cqrs.md, docs/pool.md MultiPool section)

.beads/issues.jsonl

···
 {"id":"mlecto-0b6","title":"[EPIC] Repository Module","description":"Database operations gateway. Functor over Caqti/Eio connection. CRUD operations, transactions, error handling.","status":"closed","priority":1,"issue_type":"feature","created_at":"2026-01-04T00:21:10.437260357+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:20:44.810893227+01:00","closed_at":"2026-01-04T01:20:44.810893227+01:00","close_reason":"All child tasks completed"}
 {"id":"mlecto-0hr","title":"Implement Query compilation to SQL","description":"Convert Query.t to SQL string + Caqti request. Parameter binding, SQL escaping, dialect-specific generation (Postgres first).","status":"closed","priority":1,"issue_type":"task","created_at":"2026-01-04T00:21:30.33363594+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:12:40.467657348+01:00","closed_at":"2026-01-04T01:12:40.467657348+01:00","close_reason":"Query compilation implemented in Query.to_sql - generates PostgreSQL SQL for SELECT/INSERT/UPDATE/DELETE.","dependencies":[{"issue_id":"mlecto-0hr","depends_on_id":"mlecto-7lc","type":"blocks","created_at":"2026-01-04T00:22:02.555339871+01:00","created_by":"gdiazlo"}]}
-{"id":"mlecto-1lq","title":"Create Cqrs module for read/write splitting","description":"Create a new lib/cqrs.ml module that provides transparent CQRS routing.\n\n## New File: lib/cqrs.ml\n\n### Types\n```ocaml\ntype replica_selection = RoundRobin | Random | LeastConnections\n\ntype 'conn config = {\n  primary_conninfo: string;\n  primary_max_size: int;\n  replica_conninfos: string list;  (* empty = CQRS disabled *)\n  replica_max_size_each: int;\n  replica_selection: replica_selection;\n  validate: ('conn -\u003e bool) option;\n}\n\ntype 'conn t = {\n  primary_pool: 'conn Pool.t;\n  replica_pools: 'conn Pool.t array option;  (* None when no replicas *)\n  replica_index: int Kcas.Loc.t;\n  selection: replica_selection;\n  mutable in_transaction: bool;  (* or use domain-local storage *)\n}\n\ntype intent = Read | Write\n```\n\n### Core API\n```ocaml\n(* Create CQRS-aware pool pair *)\nval create : (module Driver.S with type connection = 'conn) -\u003e 'conn config -\u003e 'conn t\n\n(* Auto-routing based on operation type *)\nval with_read : 'conn t -\u003e ('conn -\u003e 'a) -\u003e ('a, Pool.pool_error) result\nval with_write : 'conn t -\u003e ('conn -\u003e 'a) -\u003e ('a, Pool.pool_error) result\n\n(* Explicit routing *)\nval with_primary : 'conn t -\u003e ('conn -\u003e 'a) -\u003e ('a, Pool.pool_error) result\nval with_replica : 'conn t -\u003e ('conn -\u003e 'a) -\u003e ('a, Pool.pool_error) result\n\n(* Transaction - always uses primary, pins connection *)\nval transaction : 'conn t -\u003e ('conn -\u003e ('a, Error.db_error) result) -\u003e ('a, Error.db_error) result\n```\n\n### Functor Interface\n```ocaml\nmodule Make (D : Driver.S) : sig\n  type t = D.connection t\n  val create : D.connection config -\u003e t\n  (* ... rest of API *)\nend\n```\n\n## Routing Logic\n\n### Read Operations (go to replica if available)\n- Repo.get, get_opt\n- Repo.all\n- Repo.all_query, one_query, one_query_opt\n\n### Write Operations (always go to primary)\n- Repo.insert, insert_returning\n- Repo.update\n- Repo.delete\n- Repo.insert_query, update_query, delete_query\n- All transaction operations\n\n### Query-Based Auto-Detection\n```ocaml\nlet route_query query =\n  match query.Query.query_type with\n  | Query.Select -\u003e Read\n  | Query.Insert | Query.Update | Query.Delete -\u003e Write\n```\n\n## Edge Cases\n1. **No replicas configured**: All operations go to primary\n2. **All replicas down**: Fall back to primary with warning\n3. **Read after write**: Provide with_primary for consistency\n4. **Transaction context**: Pin all ops to primary connection\n\n## Acceptance Criteria\n- [ ] Cqrs.Make functor works with any Driver.S\n- [ ] Reads go to replicas when available\n- [ ] Writes always go to primary\n- [ ] Transactions pin to single primary connection\n- [ ] Graceful fallback when replicas unavailable\n- [ ] Round-robin across replicas","status":"open","priority":1,"issue_type":"task","created_at":"2026-01-05T11:13:56.804354904+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:13:56.804354904+01:00","dependencies":[{"issue_id":"mlecto-1lq","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:30.146332625+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-1lq","depends_on_id":"mlecto-2gn","type":"blocks","created_at":"2026-01-05T11:16:02.678918762+01:00","created_by":"gdiazlo"}]}
+{"id":"mlecto-1lq","title":"Create Cqrs module for read/write splitting","description":"Create a new lib/cqrs.ml module that provides transparent CQRS routing.\n\n## New File: lib/cqrs.ml\n\n### Types\n```ocaml\ntype replica_selection = RoundRobin | Random | LeastConnections\n\ntype 'conn config = {\n  primary_conninfo: string;\n  primary_max_size: int;\n  replica_conninfos: string list;  (* empty = CQRS disabled *)\n  replica_max_size_each: int;\n  replica_selection: replica_selection;\n  validate: ('conn -\u003e bool) option;\n}\n\ntype 'conn t = {\n  primary_pool: 'conn Pool.t;\n  replica_pools: 'conn Pool.t array option;  (* None when no replicas *)\n  replica_index: int Kcas.Loc.t;\n  selection: replica_selection;\n  mutable in_transaction: bool;  (* or use domain-local storage *)\n}\n\ntype intent = Read | Write\n```\n\n### Core API\n```ocaml\n(* Create CQRS-aware pool pair *)\nval create : (module Driver.S with type connection = 'conn) -\u003e 'conn config -\u003e 'conn t\n\n(* Auto-routing based on operation type *)\nval with_read : 'conn t -\u003e ('conn -\u003e 'a) -\u003e ('a, Pool.pool_error) result\nval with_write : 'conn t -\u003e ('conn -\u003e 'a) -\u003e ('a, Pool.pool_error) result\n\n(* Explicit routing *)\nval with_primary : 'conn t -\u003e ('conn -\u003e 'a) -\u003e ('a, Pool.pool_error) result\nval with_replica : 'conn t -\u003e ('conn -\u003e 'a) -\u003e ('a, Pool.pool_error) result\n\n(* Transaction - always uses primary, pins connection *)\nval transaction : 'conn t -\u003e ('conn -\u003e ('a, Error.db_error) result) -\u003e ('a, Error.db_error) result\n```\n\n### Functor Interface\n```ocaml\nmodule Make (D : Driver.S) : sig\n  type t = D.connection t\n  val create : D.connection config -\u003e t\n  (* ... rest of API *)\nend\n```\n\n## Routing Logic\n\n### Read Operations (go to replica if available)\n- Repo.get, get_opt\n- Repo.all\n- Repo.all_query, one_query, one_query_opt\n\n### Write Operations (always go to primary)\n- Repo.insert, insert_returning\n- Repo.update\n- Repo.delete\n- Repo.insert_query, update_query, delete_query\n- All transaction operations\n\n### Query-Based Auto-Detection\n```ocaml\nlet route_query query =\n  match query.Query.query_type with\n  | Query.Select -\u003e Read\n  | Query.Insert | Query.Update | Query.Delete -\u003e Write\n```\n\n## Edge Cases\n1. **No replicas configured**: All operations go to primary\n2. **All replicas down**: Fall back to primary with warning\n3. **Read after write**: Provide with_primary for consistency\n4. **Transaction context**: Pin all ops to primary connection\n\n## Acceptance Criteria\n- [ ] Cqrs.Make functor works with any Driver.S\n- [ ] Reads go to replicas when available\n- [ ] Writes always go to primary\n- [ ] Transactions pin to single primary connection\n- [ ] Graceful fallback when replicas unavailable\n- [ ] Round-robin across replicas","status":"closed","priority":1,"issue_type":"task","created_at":"2026-01-05T11:13:56.804354904+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:25:09.834278286+01:00","closed_at":"2026-01-05T11:25:09.834278286+01:00","close_reason":"Implemented Cqrs module with read/write splitting, replica selection strategies, and transaction support","dependencies":[{"issue_id":"mlecto-1lq","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:30.146332625+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-1lq","depends_on_id":"mlecto-2gn","type":"blocks","created_at":"2026-01-05T11:16:02.678918762+01:00","created_by":"gdiazlo"}]}
 {"id":"mlecto-1ud","title":"Implement Schema DSL (Mlecto.Schema)","description":"DSL for defining tables: field definitions, constraints (primary_key, not_null, unique, foreign_key, check), table_name generation.","status":"closed","priority":1,"issue_type":"task","created_at":"2026-01-04T00:21:24.765735157+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:09:24.538059462+01:00","closed_at":"2026-01-04T01:09:24.538059462+01:00","close_reason":"Schema DSL complete: table definitions, column constraints (PrimaryKey, NotNull, Unique, Default, Check, ForeignKey), foreign key actions, timestamps helper, SQL generation.","dependencies":[{"issue_id":"mlecto-1ud","depends_on_id":"mlecto-z72","type":"blocks","created_at":"2026-01-04T00:22:02.524126267+01:00","created_by":"gdiazlo"}]}
 {"id":"mlecto-1w9","title":"[EPIC] Core Type System \u0026 Schema DSL","description":"Define SQL types mapped to OCaml, schema definition DSL for tables/fields/constraints. Foundation for all other modules.","status":"closed","priority":1,"issue_type":"feature","created_at":"2026-01-04T00:21:05.917266549+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:20:44.799339065+01:00","closed_at":"2026-01-04T01:20:44.799339065+01:00","close_reason":"All child tasks completed"}
-{"id":"mlecto-2gn","title":"Implement MultiPool in pool.ml","description":"Extend the existing Pool module with multi-server support.\n\n## Changes to lib/pool.ml\n\n### New Types\n```ocaml\ntype server = {\n  conninfo: string;\n  weight: int;  (* for weighted round-robin, default 1 *)\n}\n\ntype 'conn multi_config = {\n  servers: server list;\n  max_size_per_server: int;\n  connect: string -\u003e ('conn, string) result;\n  close: 'conn -\u003e unit;\n  validate: ('conn -\u003e bool) option;\n}\n\ntype 'conn multi_t = {\n  pools: 'conn t array;\n  healthy: bool Kcas.Loc.t array;\n  next_index: int Kcas.Loc.t;\n}\n```\n\n### New Functions\n- `create_multi : 'conn multi_config -\u003e 'conn multi_t`\n- `acquire_multi : 'conn multi_t -\u003e ('conn, pool_error) result`\n- `release_multi : 'conn multi_t -\u003e 'conn -\u003e unit`\n- `with_connection_multi : 'conn multi_t -\u003e ('conn -\u003e 'a) -\u003e ('a, pool_error) result`\n- `shutdown_multi : 'conn multi_t -\u003e unit`\n- `stats_multi : 'conn multi_t -\u003e stats list`\n- `mark_unhealthy : 'conn multi_t -\u003e int -\u003e unit`\n- `mark_healthy : 'conn multi_t -\u003e int -\u003e unit`\n\n### Implementation Notes\n- Use atomic counter with Kcas.Loc for lock-free round-robin\n- Skip unhealthy servers in round-robin selection\n- When all servers unhealthy, return Pool_empty error\n- Each server gets its own pool with max_size_per_server connections\n\n## Acceptance Criteria\n- [ ] All new types defined\n- [ ] Round-robin distributes evenly across healthy servers\n- [ ] Unhealthy servers are skipped\n- [ ] Backward compatible - existing Pool API unchanged\n- [ ] Unit tests pass","status":"open","priority":1,"issue_type":"task","created_at":"2026-01-05T11:13:34.299242118+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:13:34.299242118+01:00","dependencies":[{"issue_id":"mlecto-2gn","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:25.102411986+01:00","created_by":"gdiazlo"}]}
+{"id":"mlecto-2gn","title":"Implement MultiPool in pool.ml","description":"Extend the existing Pool module with multi-server support.\n\n## Changes to lib/pool.ml\n\n### New Types\n```ocaml\ntype server = {\n  conninfo: string;\n  weight: int;  (* for weighted round-robin, default 1 *)\n}\n\ntype 'conn multi_config = {\n  servers: server list;\n  max_size_per_server: int;\n  connect: string -\u003e ('conn, string) result;\n  close: 'conn -\u003e unit;\n  validate: ('conn -\u003e bool) option;\n}\n\ntype 'conn multi_t = {\n  pools: 'conn t array;\n  healthy: bool Kcas.Loc.t array;\n  next_index: int Kcas.Loc.t;\n}\n```\n\n### New Functions\n- `create_multi : 'conn multi_config -\u003e 'conn multi_t`\n- `acquire_multi : 'conn multi_t -\u003e ('conn, pool_error) result`\n- `release_multi : 'conn multi_t -\u003e 'conn -\u003e unit`\n- `with_connection_multi : 'conn multi_t -\u003e ('conn -\u003e 'a) -\u003e ('a, pool_error) result`\n- `shutdown_multi : 'conn multi_t -\u003e unit`\n- `stats_multi : 'conn multi_t -\u003e stats list`\n- `mark_unhealthy : 'conn multi_t -\u003e int -\u003e unit`\n- `mark_healthy : 'conn multi_t -\u003e int -\u003e unit`\n\n### Implementation Notes\n- Use atomic counter with Kcas.Loc for lock-free round-robin\n- Skip unhealthy servers in round-robin selection\n- When all servers unhealthy, return Pool_empty error\n- Each server gets its own pool with max_size_per_server connections\n\n## Acceptance Criteria\n- [ ] All new types defined\n- [ ] Round-robin distributes evenly across healthy servers\n- [ ] Unhealthy servers are skipped\n- [ ] Backward compatible - existing Pool API unchanged\n- [ ] Unit tests pass","status":"closed","priority":1,"issue_type":"task","created_at":"2026-01-05T11:13:34.299242118+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:23:43.1577121+01:00","closed_at":"2026-01-05T11:23:43.1577121+01:00","close_reason":"Implemented MultiPool with round-robin, health tracking, and Make functor","dependencies":[{"issue_id":"mlecto-2gn","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:25.102411986+01:00","created_by":"gdiazlo"}]}
 {"id":"mlecto-3ow","title":"cast_assoc and put_assoc for nested changesets","status":"closed","priority":2,"issue_type":"feature","created_at":"2026-01-04T10:32:00.153765675+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T10:38:58.664932555+01:00","closed_at":"2026-01-04T10:38:58.664932555+01:00","close_reason":"Closed","dependencies":[{"issue_id":"mlecto-3ow","depends_on_id":"mlecto-d6f","type":"blocks","created_at":"2026-01-04T10:32:20.574240684+01:00","created_by":"gdiazlo"}]}
-{"id":"mlecto-3wf","title":"CQRS and Multi-Server Pool Support","description":"Add CQRS (Command Query Responsibility Segregation) support and multi-server connection pooling to repodb.\n\n## Goals\n1. **CQRS**: Transparent read/write splitting where reads go to replicas and writes go to primary\n2. **Multi-Server Pool**: Round-robin load balancing across multiple database servers\n3. **Backward Compatible**: Existing single-server users see no API changes\n\n## Key Features\n- Automatic routing based on query type (SELECT vs INSERT/UPDATE/DELETE)\n- Transaction pinning to primary (all ops in transaction use same connection)\n- Lock-free round-robin using kcas\n- Health checking to skip unhealthy servers\n- Fallback to primary when all replicas down\n\n## Testing\nIntegration tests with real PostgreSQL using Podman containers.","status":"open","priority":1,"issue_type":"epic","created_at":"2026-01-05T11:13:04.912905111+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:17:02.560139947+01:00"}
+{"id":"mlecto-3wf","title":"CQRS and Multi-Server Pool Support","description":"Add CQRS (Command Query Responsibility Segregation) support and multi-server connection pooling to repodb.\n\n## Goals\n1. **CQRS**: Transparent read/write splitting where reads go to replicas and writes go to primary\n2. **Multi-Server Pool**: Round-robin load balancing across multiple database servers\n3. **Backward Compatible**: Existing single-server users see no API changes\n\n## Key Features\n- Automatic routing based on query type (SELECT vs INSERT/UPDATE/DELETE)\n- Transaction pinning to primary (all ops in transaction use same connection)\n- Lock-free round-robin using kcas\n- Health checking to skip unhealthy servers\n- Fallback to primary when all replicas down\n\n## Testing\nIntegration tests with real PostgreSQL using Podman containers.","status":"closed","priority":1,"issue_type":"epic","created_at":"2026-01-05T11:13:04.912905111+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:34:13.477207902+01:00","closed_at":"2026-01-05T11:34:13.477207902+01:00","close_reason":"All subtasks completed: MultiPool, CQRS module, unit tests, integration tests, and documentation"}
 {"id":"mlecto-4g7","title":"Constraint error mapping from DB to changeset","status":"closed","priority":2,"issue_type":"feature","created_at":"2026-01-04T10:32:00.520652974+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T10:38:58.665466447+01:00","closed_at":"2026-01-04T10:38:58.665466447+01:00","close_reason":"Closed"}
 {"id":"mlecto-5km","title":"Implement Changeset validators","description":"validate_required, validate_format (regex), validate_length (min/max/is), validate_inclusion, validate_exclusion, validate_number, validate_acceptance, validate_confirmation, validate_change (custom).","status":"closed","priority":1,"issue_type":"task","created_at":"2026-01-04T00:21:40.78012908+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:14:08.654694896+01:00","closed_at":"2026-01-04T01:14:08.654694896+01:00","close_reason":"Validators implemented in Changeset core: validate_required, validate_format, validate_length, validate_inclusion, validate_exclusion, validate_number, validate_acceptance, validate_confirmation, validate_change.","dependencies":[{"issue_id":"mlecto-5km","depends_on_id":"mlecto-m13","type":"blocks","created_at":"2026-01-04T00:22:03.958038572+01:00","created_by":"gdiazlo"}]}
 {"id":"mlecto-7lc","title":"Implement Query builder (Mlecto.Query)","description":"SELECT, INSERT, UPDATE, DELETE builders. WHERE, JOIN, ORDER BY, GROUP BY, HAVING, LIMIT, OFFSET. Composable query pipelines.","status":"closed","priority":1,"issue_type":"task","created_at":"2026-01-04T00:21:28.238632191+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:12:33.283719987+01:00","closed_at":"2026-01-04T01:12:33.283719987+01:00","close_reason":"Query builder complete: SELECT, INSERT, UPDATE, DELETE with WHERE, JOIN, ORDER BY, GROUP BY, HAVING, LIMIT, OFFSET. ON CONFLICT for upserts. RETURNING clause. Full SQL generation.","dependencies":[{"issue_id":"mlecto-7lc","depends_on_id":"mlecto-1ud","type":"blocks","created_at":"2026-01-04T00:22:02.538979501+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-7lc","depends_on_id":"mlecto-nv7","type":"blocks","created_at":"2026-01-04T00:22:02.547726588+01:00","created_by":"gdiazlo"}]}
···
 {"id":"mlecto-at2","title":"[EPIC] Migration System","description":"Versioned database migrations with up/down support. DSL for create_table, add_column, create_index, etc.","status":"closed","priority":1,"issue_type":"feature","created_at":"2026-01-04T00:21:12.288036672+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:20:44.812043932+01:00","closed_at":"2026-01-04T01:20:44.812043932+01:00","close_reason":"All child tasks completed"}
 {"id":"mlecto-c1o","title":"Real Repo execution via Caqti/Eio","status":"closed","priority":1,"issue_type":"feature","created_at":"2026-01-04T10:31:58.698847822+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T10:38:58.663072869+01:00","closed_at":"2026-01-04T10:38:58.663072869+01:00","close_reason":"Closed"}
 {"id":"mlecto-d6f","title":"Schema associations (has_many, belongs_to, has_one, many_to_many)","status":"closed","priority":1,"issue_type":"feature","created_at":"2026-01-04T10:31:59.069873954+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T10:38:58.661985306+01:00","closed_at":"2026-01-04T10:38:58.661985306+01:00","close_reason":"Closed"}
-{"id":"mlecto-dg4","title":"Add unit tests for MultiPool","description":"Add comprehensive unit tests for the MultiPool functionality.\n\n## New File: test/test_multi_pool.ml\n\n### Test Cases\n\n1. **Round-Robin Distribution**\n   - Create multi-pool with 3 servers\n   - Acquire N connections, verify even distribution\n   - Assert each server gets ~N/3 connections\n\n2. **Health Check Skipping**\n   - Create multi-pool with 3 servers\n   - Mark one server unhealthy\n   - Verify connections only go to healthy servers\n   - Mark all unhealthy -\u003e expect Pool_empty error\n\n3. **Mark Healthy/Unhealthy**\n   - Test mark_unhealthy and mark_healthy functions\n   - Verify atomic updates work correctly\n\n4. **with_connection_multi**\n   - Test automatic release on success\n   - Test automatic release on exception\n\n5. **shutdown_multi**\n   - Verify all pools are closed\n   - Verify subsequent acquire returns Pool_closed\n\n6. **stats_multi**\n   - Verify stats aggregation across all pools\n\n7. **Edge Cases**\n   - Single server multi-pool (degenerate case)\n   - Empty server list (should error)\n   - All servers fail to connect\n\n## Mock Setup\nUse similar mock pattern to existing test_pool.ml:\n```ocaml\ntype mock_conn = { id : int; server_idx : int; mutable closed : bool }\n\nlet mock_multi_config ?(servers = 3) ?(max_size = 2) () = ...\n```\n\n## Acceptance Criteria\n- [ ] All test cases implemented\n- [ ] Tests pass with `dune test`\n- [ ] Edge cases covered","status":"open","priority":2,"issue_type":"task","created_at":"2026-01-05T11:14:18.326292543+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:14:18.326292543+01:00","dependencies":[{"issue_id":"mlecto-dg4","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:35.191030964+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-dg4","depends_on_id":"mlecto-2gn","type":"blocks","created_at":"2026-01-05T11:16:07.722230492+01:00","created_by":"gdiazlo"}]}
-{"id":"mlecto-e6g","title":"Set up Podman-based PostgreSQL integration tests","description":"Create integration test infrastructure using Podman to run real PostgreSQL instances.\n\n## Goal\nTest CQRS and MultiPool against real PostgreSQL primary + replicas.\n\n## Infrastructure\n\n### Podman Compose File: test/integration/docker-compose.yml\n```yaml\nversion: '3.8'\nservices:\n  primary:\n    image: postgres:16\n    environment:\n      POSTGRES_USER: repodb\n      POSTGRES_PASSWORD: repodb\n      POSTGRES_DB: repodb_test\n    ports:\n      - '5432:5432'\n    command: \u003e\n      postgres\n      -c wal_level=replica\n      -c max_wal_senders=3\n      -c max_replication_slots=3\n      -c hot_standby=on\n    volumes:\n      - primary_data:/var/lib/postgresql/data\n    healthcheck:\n      test: ['CMD-SHELL', 'pg_isready -U repodb']\n      interval: 5s\n      timeout: 5s\n      retries: 5\n\n  replica1:\n    image: postgres:16\n    environment:\n      POSTGRES_USER: repodb\n      POSTGRES_PASSWORD: repodb\n      POSTGRES_DB: repodb_test\n    ports:\n      - '5433:5432'\n    depends_on:\n      primary:\n        condition: service_healthy\n    command: \u003e\n      bash -c \"\n        until pg_basebackup -h primary -D /var/lib/postgresql/data -U repodb -Fp -Xs -P -R; do\n          sleep 1\n        done\n        chmod 700 /var/lib/postgresql/data\n        postgres\n      \"\n    volumes:\n      - replica1_data:/var/lib/postgresql/data\n\n  replica2:\n    image: postgres:16\n    environment:\n      POSTGRES_USER: repodb\n      POSTGRES_PASSWORD: repodb\n      POSTGRES_DB: repodb_test\n    ports:\n      - '5434:5432'\n    depends_on:\n      primary:\n        condition: service_healthy\n    command: \u003e\n      bash -c \"\n        until pg_basebackup -h primary -D /var/lib/postgresql/data -U repodb -Fp -Xs -P -R; do\n          sleep 1\n        done\n        chmod 700 /var/lib/postgresql/data\n        postgres\n      \"\n    volumes:\n      - replica2_data:/var/lib/postgresql/data\n\nvolumes:\n  primary_data:\n  replica1_data:\n  replica2_data:\n```\n\n### Test Runner Script: test/integration/run_tests.sh\n```bash\n#!/bin/bash\nset -e\n\n# Start PostgreSQL cluster\npodman-compose -f test/integration/docker-compose.yml up -d\n\n# Wait for replicas to catch up\nsleep 10\n\n# Run integration tests\ndune exec test/integration/test_cqrs_integration.exe\n\n# Cleanup\npodman-compose -f test/integration/docker-compose.yml down -v\n```\n\n### Integration Test File: test/integration/test_cqrs_integration.ml\nTest cases:\n1. Write to primary, read from replica (verify replication)\n2. Transaction isolation (all ops on primary)\n3. Replica lag handling\n4. Connection pooling under load\n5. Failover when replica dies\n\n## Acceptance Criteria\n- [ ] Podman compose file creates working PG cluster\n- [ ] Replication works between primary and replicas\n- [ ] Integration tests pass\n- [ ] Cleanup script removes all containers/volumes","status":"open","priority":2,"issue_type":"task","created_at":"2026-01-05T11:14:57.307024006+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:14:57.307024006+01:00","dependencies":[{"issue_id":"mlecto-e6g","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:45.282902638+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-e6g","depends_on_id":"mlecto-2gn","type":"blocks","created_at":"2026-01-05T11:16:17.810402811+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-e6g","depends_on_id":"mlecto-1lq","type":"blocks","created_at":"2026-01-05T11:16:22.852706838+01:00","created_by":"gdiazlo"}]}
+{"id":"mlecto-dg4","title":"Add unit tests for MultiPool","description":"Add comprehensive unit tests for the MultiPool functionality.\n\n## New File: test/test_multi_pool.ml\n\n### Test Cases\n\n1. **Round-Robin Distribution**\n   - Create multi-pool with 3 servers\n   - Acquire N connections, verify even distribution\n   - Assert each server gets ~N/3 connections\n\n2. **Health Check Skipping**\n   - Create multi-pool with 3 servers\n   - Mark one server unhealthy\n   - Verify connections only go to healthy servers\n   - Mark all unhealthy -\u003e expect Pool_empty error\n\n3. **Mark Healthy/Unhealthy**\n   - Test mark_unhealthy and mark_healthy functions\n   - Verify atomic updates work correctly\n\n4. **with_connection_multi**\n   - Test automatic release on success\n   - Test automatic release on exception\n\n5. **shutdown_multi**\n   - Verify all pools are closed\n   - Verify subsequent acquire returns Pool_closed\n\n6. **stats_multi**\n   - Verify stats aggregation across all pools\n\n7. **Edge Cases**\n   - Single server multi-pool (degenerate case)\n   - Empty server list (should error)\n   - All servers fail to connect\n\n## Mock Setup\nUse similar mock pattern to existing test_pool.ml:\n```ocaml\ntype mock_conn = { id : int; server_idx : int; mutable closed : bool }\n\nlet mock_multi_config ?(servers = 3) ?(max_size = 2) () = ...\n```\n\n## Acceptance Criteria\n- [ ] All test cases implemented\n- [ ] Tests pass with `dune test`\n- [ ] Edge cases covered","status":"closed","priority":2,"issue_type":"task","created_at":"2026-01-05T11:14:18.326292543+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:29:05.161622523+01:00","closed_at":"2026-01-05T11:29:05.161622523+01:00","close_reason":"Added 14 unit tests for MultiPool covering round-robin, health checks, and edge cases","dependencies":[{"issue_id":"mlecto-dg4","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:35.191030964+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-dg4","depends_on_id":"mlecto-2gn","type":"blocks","created_at":"2026-01-05T11:16:07.722230492+01:00","created_by":"gdiazlo"}]}
+{"id":"mlecto-e6g","title":"Set up Podman-based PostgreSQL integration tests","description":"Create integration test infrastructure using Podman to run real PostgreSQL instances.\n\n## Goal\nTest CQRS and MultiPool against real PostgreSQL primary + replicas.\n\n## Infrastructure\n\n### Podman Compose File: test/integration/docker-compose.yml\n```yaml\nversion: '3.8'\nservices:\n  primary:\n    image: postgres:16\n    environment:\n      POSTGRES_USER: repodb\n      POSTGRES_PASSWORD: repodb\n      POSTGRES_DB: repodb_test\n    ports:\n      - '5432:5432'\n    command: \u003e\n      postgres\n      -c wal_level=replica\n      -c max_wal_senders=3\n      -c max_replication_slots=3\n      -c hot_standby=on\n    volumes:\n      - primary_data:/var/lib/postgresql/data\n    healthcheck:\n      test: ['CMD-SHELL', 'pg_isready -U repodb']\n      interval: 5s\n      timeout: 5s\n      retries: 5\n\n  replica1:\n    image: postgres:16\n    environment:\n      POSTGRES_USER: repodb\n      POSTGRES_PASSWORD: repodb\n      POSTGRES_DB: repodb_test\n    ports:\n      - '5433:5432'\n    depends_on:\n      primary:\n        condition: service_healthy\n    command: \u003e\n      bash -c \"\n        until pg_basebackup -h primary -D /var/lib/postgresql/data -U repodb -Fp -Xs -P -R; do\n          sleep 1\n        done\n        chmod 700 /var/lib/postgresql/data\n        postgres\n      \"\n    volumes:\n      - replica1_data:/var/lib/postgresql/data\n\n  replica2:\n    image: postgres:16\n    environment:\n      POSTGRES_USER: repodb\n      POSTGRES_PASSWORD: repodb\n      POSTGRES_DB: repodb_test\n    ports:\n      - '5434:5432'\n    depends_on:\n      primary:\n        condition: service_healthy\n    command: \u003e\n      bash -c \"\n        until pg_basebackup -h primary -D /var/lib/postgresql/data -U repodb -Fp -Xs -P -R; do\n          sleep 1\n        done\n        chmod 700 /var/lib/postgresql/data\n        postgres\n      \"\n    volumes:\n      - replica2_data:/var/lib/postgresql/data\n\nvolumes:\n  primary_data:\n  replica1_data:\n  replica2_data:\n```\n\n### Test Runner Script: test/integration/run_tests.sh\n```bash\n#!/bin/bash\nset -e\n\n# Start PostgreSQL cluster\npodman-compose -f test/integration/docker-compose.yml up -d\n\n# Wait for replicas to catch up\nsleep 10\n\n# Run integration tests\ndune exec test/integration/test_cqrs_integration.exe\n\n# Cleanup\npodman-compose -f test/integration/docker-compose.yml down -v\n```\n\n### Integration Test File: test/integration/test_cqrs_integration.ml\nTest cases:\n1. Write to primary, read from replica (verify replication)\n2. Transaction isolation (all ops on primary)\n3. Replica lag handling\n4. Connection pooling under load\n5. Failover when replica dies\n\n## Acceptance Criteria\n- [ ] Podman compose file creates working PG cluster\n- [ ] Replication works between primary and replicas\n- [ ] Integration tests pass\n- [ ] Cleanup script removes all containers/volumes","status":"closed","priority":2,"issue_type":"task","created_at":"2026-01-05T11:14:57.307024006+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:32:01.121669503+01:00","closed_at":"2026-01-05T11:32:01.121669503+01:00","close_reason":"Created Podman compose setup with primary+2 replicas, init script, test runner, and integration test suite","dependencies":[{"issue_id":"mlecto-e6g","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:45.282902638+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-e6g","depends_on_id":"mlecto-2gn","type":"blocks","created_at":"2026-01-05T11:16:17.810402811+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-e6g","depends_on_id":"mlecto-1lq","type":"blocks","created_at":"2026-01-05T11:16:22.852706838+01:00","created_by":"gdiazlo"}]}
 {"id":"mlecto-flv","title":"Implement Multi module (Mlecto.Multi)","description":"Chain operations: Multi.new |\u003e Multi.insert |\u003e Multi.update |\u003e Multi.run. Named operations, access previous results, atomic execution.","status":"closed","priority":2,"issue_type":"task","created_at":"2026-01-04T00:21:50.583520239+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:20:32.435485703+01:00","closed_at":"2026-01-04T01:20:32.435485703+01:00","close_reason":"Multi with operation chaining, named results, merge, validate, execute_sync","dependencies":[{"issue_id":"mlecto-flv","depends_on_id":"mlecto-8ki","type":"blocks","created_at":"2026-01-04T00:22:07.51964609+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-flv","depends_on_id":"mlecto-hwj","type":"blocks","created_at":"2026-01-04T00:22:07.527438964+01:00","created_by":"gdiazlo"}]}
 {"id":"mlecto-gfo","title":"[EPIC] Changeset System","description":"Casting, validation, and error accumulation. Separate validation from persistence with composable validators.","status":"closed","priority":1,"issue_type":"feature","created_at":"2026-01-04T00:21:07.363157981+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:20:44.808207655+01:00","closed_at":"2026-01-04T01:20:44.808207655+01:00","close_reason":"All child tasks completed"}
 {"id":"mlecto-gxh","title":"Implement Changeset constraints","description":"unique_constraint, foreign_key_constraint, check_constraint, exclusion_constraint. Convert DB errors to changeset errors.","status":"closed","priority":1,"issue_type":"task","created_at":"2026-01-04T00:21:42.297745588+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:14:08.663238895+01:00","closed_at":"2026-01-04T01:14:08.663238895+01:00","close_reason":"Constraints implemented in Changeset core: unique_constraint, foreign_key_constraint, check_constraint.","dependencies":[{"issue_id":"mlecto-gxh","depends_on_id":"mlecto-m13","type":"blocks","created_at":"2026-01-04T00:22:03.965402454+01:00","created_by":"gdiazlo"}]}
 {"id":"mlecto-hw6","title":"[EPIC] mlecto - Ecto-like database toolkit for OCaml","status":"closed","priority":1,"issue_type":"feature","created_at":"2026-01-04T00:20:55.991644266+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:20:49.927971061+01:00","closed_at":"2026-01-04T01:20:49.927971061+01:00","close_reason":"Core mlecto library complete: Type, Schema, Expr, Query, Changeset, Repo, Migration, Multi modules implemented"}
 {"id":"mlecto-hwj","title":"Implement Repo transactions","description":"transaction/1 function wrapping Caqti transactions. Automatic rollback on error. Nested transaction support (savepoints).","status":"closed","priority":1,"issue_type":"task","created_at":"2026-01-04T00:21:46.455864854+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:19:20.791036484+01:00","closed_at":"2026-01-04T01:19:20.791036484+01:00","close_reason":"Transaction state machine with BEGIN/COMMIT/ROLLBACK and nested savepoint support","dependencies":[{"issue_id":"mlecto-hwj","depends_on_id":"mlecto-8ki","type":"blocks","created_at":"2026-01-04T00:22:05.217260647+01:00","created_by":"gdiazlo"}]}
-{"id":"mlecto-hwq","title":"Write CQRS documentation","description":"Create documentation for the CQRS and MultiPool features.\n\n## New File: docs/cqrs.md\n\n### Sections\n\n1. **Overview**\n   - What is CQRS\n   - When to use it\n   - How repodb implements it\n\n2. **Quick Start**\n   ```ocaml\n   (* Basic CQRS setup *)\n   module Cqrs = Cqrs.Make(Repodb_postgresql)\n\n   let cqrs = Cqrs.create {\n     primary_conninfo = \"postgresql://primary/mydb\";\n     primary_max_size = 10;\n     replica_conninfos = [\n       \"postgresql://replica1/mydb\";\n       \"postgresql://replica2/mydb\";\n     ];\n     replica_max_size_each = 5;\n     replica_selection = RoundRobin;\n     validate = None;\n   }\n   ```\n\n3. **Read Operations**\n   - Which operations are routed to replicas\n   - How to force read from primary\n\n4. **Write Operations**\n   - Always go to primary\n   - Transaction handling\n\n5. **Explicit Routing**\n   - with_primary for read-after-write consistency\n   - with_replica for specific use cases\n\n6. **Configuration Options**\n   - Full config reference\n   - Replica selection strategies\n\n7. **Best Practices**\n   - When to use CQRS\n   - Handling replica lag\n   - Monitoring\n\n## Update: docs/pool.md\n\nAdd section on MultiPool:\n- Multi-server configuration\n- Round-robin behavior\n- Health checking\n\n## Acceptance Criteria\n- [ ] docs/cqrs.md complete with examples\n- [ ] docs/pool.md updated for MultiPool\n- [ ] README.md mentions new features\n- [ ] Examples compile and work","status":"open","priority":3,"issue_type":"task","created_at":"2026-01-05T11:15:13.481713538+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:15:13.481713538+01:00","dependencies":[{"issue_id":"mlecto-hwq","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:50.325885062+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-hwq","depends_on_id":"mlecto-1lq","type":"blocks","created_at":"2026-01-05T11:16:27.893913307+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-hwq","depends_on_id":"mlecto-2gn","type":"blocks","created_at":"2026-01-05T11:16:32.936486931+01:00","created_by":"gdiazlo"}]}
+{"id":"mlecto-hwq","title":"Write CQRS documentation","description":"Create documentation for the CQRS and MultiPool features.\n\n## New File: docs/cqrs.md\n\n### Sections\n\n1. **Overview**\n   - What is CQRS\n   - When to use it\n   - How repodb implements it\n\n2. **Quick Start**\n   ```ocaml\n   (* Basic CQRS setup *)\n   module Cqrs = Cqrs.Make(Repodb_postgresql)\n\n   let cqrs = Cqrs.create {\n     primary_conninfo = \"postgresql://primary/mydb\";\n     primary_max_size = 10;\n     replica_conninfos = [\n       \"postgresql://replica1/mydb\";\n       \"postgresql://replica2/mydb\";\n     ];\n     replica_max_size_each = 5;\n     replica_selection = RoundRobin;\n     validate = None;\n   }\n   ```\n\n3. **Read Operations**\n   - Which operations are routed to replicas\n   - How to force read from primary\n\n4. **Write Operations**\n   - Always go to primary\n   - Transaction handling\n\n5. **Explicit Routing**\n   - with_primary for read-after-write consistency\n   - with_replica for specific use cases\n\n6. **Configuration Options**\n   - Full config reference\n   - Replica selection strategies\n\n7. **Best Practices**\n   - When to use CQRS\n   - Handling replica lag\n   - Monitoring\n\n## Update: docs/pool.md\n\nAdd section on MultiPool:\n- Multi-server configuration\n- Round-robin behavior\n- Health checking\n\n## Acceptance Criteria\n- [ ] docs/cqrs.md complete with examples\n- [ ] docs/pool.md updated for MultiPool\n- [ ] README.md mentions new features\n- [ ] Examples compile and work","status":"closed","priority":3,"issue_type":"task","created_at":"2026-01-05T11:15:13.481713538+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:34:08.431613146+01:00","closed_at":"2026-01-05T11:34:08.431613146+01:00","close_reason":"Created docs/cqrs.md, updated docs/pool.md with MultiPool section, updated README.md","dependencies":[{"issue_id":"mlecto-hwq","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:50.325885062+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-hwq","depends_on_id":"mlecto-1lq","type":"blocks","created_at":"2026-01-05T11:16:27.893913307+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-hwq","depends_on_id":"mlecto-2gn","type":"blocks","created_at":"2026-01-05T11:16:32.936486931+01:00","created_by":"gdiazlo"}]}
 {"id":"mlecto-l6s","title":"[EPIC] Query DSL","description":"Type-safe SQL query builder using GADTs and phantom types. Composable queries as first-class values.","status":"closed","priority":1,"issue_type":"feature","created_at":"2026-01-04T00:21:08.537372834+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:20:44.809745142+01:00","closed_at":"2026-01-04T01:20:44.809745142+01:00","close_reason":"All child tasks completed"}
-{"id":"mlecto-lg9","title":"Add unit tests for Cqrs module","description":"Add comprehensive unit tests for the CQRS routing functionality.\n\n## New File: test/test_cqrs.ml\n\n### Test Cases\n\n1. **Read Routing**\n   - Configure CQRS with primary + replicas\n   - Execute with_read -\u003e verify goes to replica\n   - Execute Repo.all -\u003e verify goes to replica\n\n2. **Write Routing**\n   - Execute with_write -\u003e verify goes to primary\n   - Execute Repo.insert -\u003e verify goes to primary\n\n3. **Query-Based Routing**\n   - SELECT query -\u003e replica\n   - INSERT query -\u003e primary\n   - UPDATE query -\u003e primary\n   - DELETE query -\u003e primary\n\n4. **Transaction Pinning**\n   - Start transaction -\u003e all ops go to primary\n   - Even read operations inside transaction go to primary\n   - Same connection used throughout transaction\n\n5. **No Replicas Mode**\n   - Configure with empty replica list\n   - All operations go to primary\n   - No errors, works like single-server pool\n\n6. **Replica Fallback**\n   - All replicas unhealthy\n   - Reads fall back to primary\n   - Warning logged (if logging added)\n\n7. **with_primary / with_replica**\n   - Explicit routing works as expected\n   - with_primary always uses primary even for reads\n   - with_replica errors if no replicas configured\n\n8. **Round-Robin Across Replicas**\n   - Multiple replicas configured\n   - Reads distributed evenly\n\n## Mock Setup\n```ocaml\n(* Track which pool each operation went to *)\ntype routing_tracker = {\n  mutable primary_calls: int;\n  mutable replica_calls: int list;  (* per-replica counts *)\n}\n```\n\n## Acceptance Criteria\n- [ ] All test cases implemented\n- [ ] Tests pass with `dune test`\n- [ ] Transaction isolation tested\n- [ ] Fallback behavior tested","status":"open","priority":2,"issue_type":"task","created_at":"2026-01-05T11:14:37.578081717+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:14:37.578081717+01:00","dependencies":[{"issue_id":"mlecto-lg9","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:40.237975143+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-lg9","depends_on_id":"mlecto-1lq","type":"blocks","created_at":"2026-01-05T11:16:12.765564629+01:00","created_by":"gdiazlo"}]}
+{"id":"mlecto-lg9","title":"Add unit tests for Cqrs module","description":"Add comprehensive unit tests for the CQRS routing functionality.\n\n## New File: test/test_cqrs.ml\n\n### Test Cases\n\n1. **Read Routing**\n   - Configure CQRS with primary + replicas\n   - Execute with_read -\u003e verify goes to replica\n   - Execute Repo.all -\u003e verify goes to replica\n\n2. **Write Routing**\n   - Execute with_write -\u003e verify goes to primary\n   - Execute Repo.insert -\u003e verify goes to primary\n\n3. **Query-Based Routing**\n   - SELECT query -\u003e replica\n   - INSERT query -\u003e primary\n   - UPDATE query -\u003e primary\n   - DELETE query -\u003e primary\n\n4. **Transaction Pinning**\n   - Start transaction -\u003e all ops go to primary\n   - Even read operations inside transaction go to primary\n   - Same connection used throughout transaction\n\n5. **No Replicas Mode**\n   - Configure with empty replica list\n   - All operations go to primary\n   - No errors, works like single-server pool\n\n6. **Replica Fallback**\n   - All replicas unhealthy\n   - Reads fall back to primary\n   - Warning logged (if logging added)\n\n7. **with_primary / with_replica**\n   - Explicit routing works as expected\n   - with_primary always uses primary even for reads\n   - with_replica errors if no replicas configured\n\n8. **Round-Robin Across Replicas**\n   - Multiple replicas configured\n   - Reads distributed evenly\n\n## Mock Setup\n```ocaml\n(* Track which pool each operation went to *)\ntype routing_tracker = {\n  mutable primary_calls: int;\n  mutable replica_calls: int list;  (* per-replica counts *)\n}\n```\n\n## Acceptance Criteria\n- [ ] All test cases implemented\n- [ ] Tests pass with `dune test`\n- [ ] Transaction isolation tested\n- [ ] Fallback behavior tested","status":"closed","priority":2,"issue_type":"task","created_at":"2026-01-05T11:14:37.578081717+01:00","created_by":"gdiazlo","updated_at":"2026-01-05T11:29:10.203623799+01:00","closed_at":"2026-01-05T11:29:10.203623799+01:00","close_reason":"Added 13 unit tests for Cqrs covering routing, replicas, fallback, and transactions","dependencies":[{"issue_id":"mlecto-lg9","depends_on_id":"mlecto-3wf","type":"blocks","created_at":"2026-01-05T11:15:40.237975143+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-lg9","depends_on_id":"mlecto-1lq","type":"blocks","created_at":"2026-01-05T11:16:12.765564629+01:00","created_by":"gdiazlo"}]}
 {"id":"mlecto-m13","title":"Implement Changeset core (Mlecto.Changeset)","description":"Changeset record type: data, changes, errors, valid?, params, required. Cast function for external params. Change function for internal data.","status":"closed","priority":1,"issue_type":"task","created_at":"2026-01-04T00:21:38.875652686+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:13:59.358336893+01:00","closed_at":"2026-01-04T01:13:59.358336893+01:00","close_reason":"Changeset core complete: type-safe field handling, validations (required, format, length, inclusion, exclusion, number, acceptance, confirmation, custom), constraints (unique, foreign_key, check), error handling.","dependencies":[{"issue_id":"mlecto-m13","depends_on_id":"mlecto-z72","type":"blocks","created_at":"2026-01-04T00:22:03.941717052+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-m13","depends_on_id":"mlecto-1ud","type":"blocks","created_at":"2026-01-04T00:22:03.95059939+01:00","created_by":"gdiazlo"}]}
 {"id":"mlecto-nv7","title":"Implement Expression types (Mlecto.Expr)","description":"GADTs for SQL expressions: literals, columns, operators (+, -, =, \u003c\u003e, AND, OR), function calls, casts. Type-safe expression composition.","status":"closed","priority":1,"issue_type":"task","created_at":"2026-01-04T00:21:26.662064082+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:11:24.632729849+01:00","closed_at":"2026-01-04T01:11:24.632729849+01:00","close_reason":"Expression types complete: GADT for all SQL expressions, operators (comparison, logical, arithmetic), functions (aggregate, string, date, math), CASE, BETWEEN, IN, CAST, SQL generation.","dependencies":[{"issue_id":"mlecto-nv7","depends_on_id":"mlecto-z72","type":"blocks","created_at":"2026-01-04T00:22:02.531940181+01:00","created_by":"gdiazlo"}]}
 {"id":"mlecto-pvs","title":"Implement Migration runner","description":"schema_migrations table, version tracking, up/down execution, migrate/rollback commands, migration status reporting.","status":"closed","priority":1,"issue_type":"task","created_at":"2026-01-04T00:21:48.916746145+01:00","created_by":"gdiazlo","updated_at":"2026-01-04T01:18:41.847908163+01:00","closed_at":"2026-01-04T01:18:41.847908163+01:00","close_reason":"Migration runner with version tracking, plan_migrate, plan_rollback, format_status","dependencies":[{"issue_id":"mlecto-pvs","depends_on_id":"mlecto-uek","type":"blocks","created_at":"2026-01-04T00:22:06.667807096+01:00","created_by":"gdiazlo"},{"issue_id":"mlecto-pvs","depends_on_id":"mlecto-8ki","type":"blocks","created_at":"2026-01-04T00:22:06.6757586+01:00","created_by":"gdiazlo"}]}

README.md

···
 - **Multi** - Compose multiple operations in a single transaction
 - **Migrations** - Version your database schema
 - **Connection Pool** - Scheduler-agnostic, lock-free connection pooling
+- **CQRS** - Transparent read/write splitting with replica support
+- **Multi-Server Pool** - Round-robin load balancing across database servers
 
 ## Quick Example
 
···
 - [Transactions](docs/transactions.md) - Multi and transaction handling
 - [Migrations](docs/migrations.md) - Database schema versioning
 - [Connection Pool](docs/pool.md) - Scheduler-agnostic connection pooling
+- [CQRS](docs/cqrs.md) - Read/write splitting with replicas
 
 ## Packages
 

docs/cqrs.md

···
+# CQRS (Command Query Responsibility Segregation)
+
+repodb supports CQRS through transparent read/write splitting. Reads are automatically routed to replica servers while writes go to the primary, all without requiring application code changes.
+
+## When to Use CQRS
+
+CQRS is beneficial when:
+- Read traffic significantly exceeds write traffic
+- You want to scale reads horizontally with replicas
+- You want to avoid database load balancers like pgbouncer
+- Your application can tolerate slight replica lag for reads
+
+## Quick Start
+
+```ocaml
+open Repodb
+
+module Cqrs = Cqrs.Make(Repodb_postgresql)
+module Repo = Repo.Make(Repodb_postgresql)
+
+let cqrs = Cqrs.create {
+  primary_conninfo = "host=primary dbname=myapp user=app";
+  primary_max_size = 10;
+  replica_conninfos = [
+    "host=replica1 dbname=myapp user=app";
+    "host=replica2 dbname=myapp user=app";
+  ];
+  replica_max_size_each = 5;
+  replica_selection = RoundRobin;
+  validate = None;
+}
+
+(* Reads automatically go to replicas *)
+let users = Cqrs.with_read cqrs (fun conn ->
+  Repo.all conn ~table:users_table ~decode:decode_user
+)
+
+(* Writes always go to primary *)
+let () = Cqrs.with_write cqrs (fun conn ->
+  Repo.insert conn ~table:users_table ~columns ~values
+)
+```
+
+## Read Operations
+
+Use `with_read` for operations that only query data:
+
+```ocaml
+(* Routed to a replica *)
+let result = Cqrs.with_read cqrs (fun conn ->
+  Repo.all_query conn query ~decode
+)
+```
+
+When replicas are configured, reads are distributed using the configured selection strategy. If all replicas are unhealthy, reads automatically fall back to the primary.
+
+### Forcing Read from Primary
+
+For read-after-write consistency, use `with_primary`:
+
+```ocaml
+(* Insert a user *)
+let user_id = Cqrs.with_write cqrs (fun conn ->
+  Repo.insert_returning conn ~table ~columns ~values ~decode
+)
+
+(* Read immediately after write - use primary to avoid replica lag *)
+let user = Cqrs.with_primary cqrs (fun conn ->
+  Repo.get conn ~table ~id:user_id ~decode
+)
+```
+
+## Write Operations
+
+Writes always go to the primary server:
+
+```ocaml
+let () = Cqrs.with_write cqrs (fun conn ->
+  Repo.update conn ~table ~columns ~values ~where_column ~where_value
+)
+```
+
+## Transactions
+
+Transactions are always executed on the primary with a pinned connection:
+
+```ocaml
+let result = Cqrs.transaction cqrs (fun conn ->
+  (* All operations use the same primary connection *)
+  match Repo.insert conn ~table:orders ~columns ~values with
+  | Ok () ->
+      Repo.update conn ~table:inventory ~columns:["stock"] 
+        ~values:[Value.int (stock - 1)] ~where_column:"id" ~where_value:(Value.int item_id)
+  | Error e -> Error e
+)
+```
+
+## Replica Selection Strategies
+
+### RoundRobin (Default)
+
+Distributes requests evenly across healthy replicas:
+
+```ocaml
+replica_selection = RoundRobin;
+```
+
+### Random
+
+Selects a random healthy replica for each request:
+
+```ocaml
+replica_selection = Random;
+```
+
+### LeastConnections
+
+Routes to the replica with the fewest active connections:
+
+```ocaml
+replica_selection = LeastConnections;
+```
+
+## Health Management
+
+Mark replicas as unhealthy when they fail:
+
+```ocaml
+(* Mark replica 0 as unhealthy *)
+Cqrs.mark_replica_unhealthy cqrs 0;
+
+(* Later, mark it healthy again *)
+Cqrs.mark_replica_healthy cqrs 0;
+```
+
+When all replicas are marked unhealthy, reads automatically fall back to the primary.
+
+## Configuration Reference
+
+```ocaml
+type 'conn config = {
+  primary_conninfo : string;       (* Primary server connection string *)
+  primary_max_size : int;          (* Max connections to primary *)
+  replica_conninfos : string list; (* Replica connection strings, empty = no CQRS *)
+  replica_max_size_each : int;     (* Max connections per replica *)
+  replica_selection : replica_selection; (* RoundRobin | Random | LeastConnections *)
+  validate : ('conn -> bool) option;     (* Optional connection validator *)
+}
+```
+
+## Monitoring
+
+Check CQRS status with `stats`:
+
+```ocaml
+let s = Cqrs.stats cqrs in
+Printf.printf "Primary: total=%d in_use=%d available=%d\n"
+  s.primary.total s.primary.in_use s.primary.available;
+Printf.printf "Healthy replicas: %d\n" s.healthy_replica_count;
+```
+
+## Disabling CQRS
+
+To disable CQRS, pass an empty replica list:
+
+```ocaml
+let cqrs = Cqrs.create {
+  primary_conninfo = "host=localhost dbname=myapp";
+  primary_max_size = 10;
+  replica_conninfos = [];  (* No replicas = all traffic goes to primary *)
+  replica_max_size_each = 0;
+  replica_selection = RoundRobin;
+  validate = None;
+}
+```
+
+## Shutdown
+
+Always shut down the CQRS instance when done:
+
+```ocaml
+Cqrs.shutdown cqrs
+```
+
+## Best Practices
+
+1. **Size pools appropriately**: Primary pool should handle write load; replica pools handle read load
+2. **Handle replica lag**: Use `with_primary` for reads that need immediate consistency
+3. **Monitor health**: Track unhealthy replica counts and investigate failures
+4. **Connection validation**: Use `validate` to detect stale connections
+
+## Next Steps
+
+- [Connection Pool](pool.md) - Lower-level pool management and MultiPool
+- [Repo](repo.md) - Database operations
+- [Transactions](transactions.md) - Transaction handling

docs/pool.md

···
       Ok conn
 ```
 
+## Multi-Server Pool
+
+For load balancing across multiple database servers without an external proxy, use `Pool.Make(Driver).Multi`:
+
+### Setup
+
+```ocaml
+module Pool = Pool.Make(Repodb_postgresql)
+
+let multi = Pool.Multi.create_sized
+  ~servers:[
+    "host=server1 dbname=myapp";
+    "host=server2 dbname=myapp";
+    "host=server3 dbname=myapp";
+  ]
+  ~max_size_per_server:5
+  ()
+```
+
+### Usage
+
+```ocaml
+(* Connections distributed via round-robin *)
+let result = Pool.Multi.with_connection multi (fun conn ->
+  Repo.all conn ~table:users ~decode
+)
+```
+
+### Health Management
+
+```ocaml
+(* Mark a server as unhealthy (skipped in round-robin) *)
+Pool.Multi.mark_unhealthy multi 1;
+
+(* Mark healthy again *)
+Pool.Multi.mark_healthy multi 0;
+```
+
+### Statistics
+
+```ocaml
+let stats = Pool.Multi.stats multi in
+Printf.printf "Total servers: %d\n" stats.total_servers;
+Printf.printf "Healthy servers: %d\n" stats.healthy_servers;
+Printf.printf "Total connections: %d\n" stats.aggregate.total;
+```
+
+### Shutdown
+
+```ocaml
+Pool.Multi.shutdown multi
+```
+
+For read/write splitting with replicas, see [CQRS](cqrs.md).
+
 ## Next Steps
 
+- [CQRS](cqrs.md) - Read/write splitting with replicas
 - [Repo](repo.md) - Database operations
 - [Transactions](transactions.md) - Multi and transaction handling
 - [Queries](queries.md) - Building complex queries

lib/cqrs.ml

···
+open Kcas
+
+type replica_selection = RoundRobin | Random | LeastConnections
+
+type 'conn config = {
+  primary_conninfo : string;
+  primary_max_size : int;
+  replica_conninfos : string list;
+  replica_max_size_each : int;
+  replica_selection : replica_selection;
+  validate : ('conn -> bool) option;
+}
+
+type 'conn t = {
+  primary_pool : 'conn Pool.t;
+  replica_pools : 'conn Pool.t array option;
+  replica_healthy : bool Loc.t array option;
+  replica_index : int Loc.t;
+  selection : replica_selection;
+}
+
+type intent = Read | Write
+
+let route_query_type = function
+  | Query.Select -> Read
+  | Query.Insert | Query.Update | Query.Delete -> Write
+
+let has_replicas t = Option.is_some t.replica_pools
+
+let count_healthy_replicas t =
+  match t.replica_healthy with
+  | None -> 0
+  | Some healthy ->
+      Array.fold_left
+        (fun acc h -> if Loc.get h then acc + 1 else acc)
+        0 healthy
+
+let next_replica_round_robin t =
+  match (t.replica_pools, t.replica_healthy) with
+  | None, _ | _, None -> None
+  | Some pools, Some healthy ->
+      let n = Array.length pools in
+      let start = Loc.fetch_and_add t.replica_index 1 mod n in
+      let rec find_healthy attempts idx =
+        if attempts >= n then None
+        else if Loc.get healthy.(idx) then Some pools.(idx)
+        else find_healthy (attempts + 1) ((idx + 1) mod n)
+      in
+      find_healthy 0 start
+
+let next_replica_random t =
+  match (t.replica_pools, t.replica_healthy) with
+  | None, _ | _, None -> None
+  | Some pools, Some healthy ->
+      let healthy_indices =
+        Array.to_list (Array.mapi (fun i h -> (i, h)) healthy)
+        |> List.filter (fun (_, h) -> Loc.get h)
+        |> List.map fst
+      in
+      if healthy_indices = [] then None
+      else
+        let idx =
+          List.nth healthy_indices (Random.int (List.length healthy_indices))
+        in
+        Some pools.(idx)
+
+let next_replica_least_connections t =
+  match (t.replica_pools, t.replica_healthy) with
+  | None, _ | _, None -> None
+  | Some pools, Some healthy ->
+      let n = Array.length pools in
+      let rec find_min best_idx best_count idx =
+        if idx >= n then if best_idx < 0 then None else Some pools.(best_idx)
+        else if Loc.get healthy.(idx) then
+          let count = Pool.in_use pools.(idx) in
+          if best_idx < 0 || count < best_count then find_min idx count (idx + 1)
+          else find_min best_idx best_count (idx + 1)
+        else find_min best_idx best_count (idx + 1)
+      in
+      find_min (-1) max_int 0
+
+let select_replica t =
+  match t.selection with
+  | RoundRobin -> next_replica_round_robin t
+  | Random -> next_replica_random t
+  | LeastConnections -> next_replica_least_connections t
+
+let mark_replica_unhealthy t idx =
+  match t.replica_healthy with
+  | None -> ()
+  | Some healthy ->
+      if idx >= 0 && idx < Array.length healthy then Loc.set healthy.(idx) false
+
+let mark_replica_healthy t idx =
+  match t.replica_healthy with
+  | None -> ()
+  | Some healthy ->
+      if idx >= 0 && idx < Array.length healthy then Loc.set healthy.(idx) true
+
+let acquire_read t =
+  match select_replica t with
+  | Some pool -> (
+      match Pool.acquire pool with
+      | Ok conn -> Ok conn
+      | Error Pool.Pool_empty | Error Pool.Pool_closed ->
+          Pool.acquire t.primary_pool
+      | Error e -> Error e)
+  | None -> Pool.acquire t.primary_pool
+
+let acquire_write t = Pool.acquire t.primary_pool
+
+let release_read t conn =
+  match t.replica_pools with
+  | Some pools ->
+      let released = ref false in
+      Array.iter
+        (fun pool ->
+          if not !released then
+            try
+              Pool.release pool conn;
+              released := true
+            with _ -> ())
+        pools;
+      if not !released then Pool.release t.primary_pool conn
+  | None -> Pool.release t.primary_pool conn
+
+let release_write t conn = Pool.release t.primary_pool conn
+
+let with_read t f =
+  match acquire_read t with
+  | Error e -> Error e
+  | Ok conn -> (
+      match f conn with
+      | result ->
+          release_read t conn;
+          Ok result
+      | exception exn ->
+          release_read t conn;
+          raise exn)
+
+let with_write t f =
+  match acquire_write t with
+  | Error e -> Error e
+  | Ok conn -> (
+      match f conn with
+      | result ->
+          release_write t conn;
+          Ok result
+      | exception exn ->
+          release_write t conn;
+          raise exn)
+
+let with_primary t f = with_write t f
+
+let with_replica t f =
+  match t.replica_pools with
+  | None -> Error Pool.Pool_empty
+  | Some _ -> with_read t f
+
+type stats = {
+  primary : Pool.stats;
+  replicas : Pool.stats list option;
+  healthy_replica_count : int;
+}
+
+let stats t =
+  let replicas =
+    Option.map
+      (fun pools -> Array.to_list (Array.map Pool.stats pools))
+      t.replica_pools
+  in
+  {
+    primary = Pool.stats t.primary_pool;
+    replicas;
+    healthy_replica_count = count_healthy_replicas t;
+  }
+
+let shutdown t =
+  Pool.shutdown t.primary_pool;
+  Option.iter (Array.iter Pool.shutdown) t.replica_pools
+
+module Make (D : Driver.S) = struct
+  type nonrec t = D.connection t
+
+  let create config =
+    let primary_pool =
+      let pool_config =
+        Pool.
+          {
+            max_size = config.primary_max_size;
+            connect =
+              (fun () ->
+                D.connect config.primary_conninfo
+                |> Result.map_error D.error_message);
+            close = D.close;
+            validate = config.validate;
+          }
+      in
+      Pool.create pool_config
+    in
+    let replica_pools, replica_healthy =
+      if config.replica_conninfos = [] then (None, None)
+      else
+        let pools =
+          Array.of_list
+            (List.map
+               (fun conninfo ->
+                 let pool_config =
+                   Pool.
+                     {
+                       max_size = config.replica_max_size_each;
+                       connect =
+                         (fun () ->
+                           D.connect conninfo
+                           |> Result.map_error D.error_message);
+                       close = D.close;
+                       validate = config.validate;
+                     }
+                 in
+                 Pool.create pool_config)
+               config.replica_conninfos)
+        in
+        let healthy =
+          Array.init (Array.length pools) (fun _ -> Loc.make true)
+        in
+        (Some pools, Some healthy)
+    in
+    {
+      primary_pool;
+      replica_pools;
+      replica_healthy;
+      replica_index = Loc.make 0;
+      selection = config.replica_selection;
+    }
+
+  let with_read = with_read
+  let with_write = with_write
+  let with_primary = with_primary
+  let with_replica = with_replica
+
+  let transaction t f =
+    match Pool.acquire t.primary_pool with
+    | Error e -> Error (Error.Query_failed (Pool.error_to_string e))
+    | Ok conn -> (
+        match D.exec conn "BEGIN" ~params:[||] with
+        | Error e ->
+            Pool.release t.primary_pool conn;
+            Error (Error.Query_failed (D.error_message e))
+        | Ok () -> (
+            match f conn with
+            | Ok v -> (
+                match D.exec conn "COMMIT" ~params:[||] with
+                | Ok () ->
+                    Pool.release t.primary_pool conn;
+                    Ok v
+                | Error e ->
+                    Pool.release t.primary_pool conn;
+                    Error (Error.Query_failed (D.error_message e)))
+            | Error err ->
+                let _ = D.exec conn "ROLLBACK" ~params:[||] in
+                Pool.release t.primary_pool conn;
+                Error err
+            | exception exn ->
+                let _ = D.exec conn "ROLLBACK" ~params:[||] in
+                Pool.release t.primary_pool conn;
+                raise exn))
+
+  let has_replicas = has_replicas
+  let count_healthy_replicas = count_healthy_replicas
+  let mark_replica_healthy = mark_replica_healthy
+  let mark_replica_unhealthy = mark_replica_unhealthy
+  let stats = stats
+  let shutdown = shutdown
+  let route_query_type = route_query_type
+end

lib/dune

···
   migration
   multi
   stream
-  pool))
+  pool
+  cqrs))

lib/pool.ml

···
   | Pool_timeout -> "Timeout waiting for connection"
   | Connection_error msg -> Printf.sprintf "Connection error: %s" msg
 
+(** Multi-server pool with round-robin load balancing.
+
+    Distributes connections across multiple database servers without requiring
+    an external load balancer. Each server maintains its own connection pool.
+
+    {[
+      let config =
+        Pool.
+          {
+            servers = [ "host1"; "host2"; "host3" ];
+            max_size_per_server = 5;
+            connect = (fun conninfo -> Repodb_postgresql.connect conninfo);
+            close = Repodb_postgresql.close;
+            validate = None;
+          }
+      in
+      let multi = Pool.create_multi config in
+      Pool.with_connection_multi multi (fun conn ->
+          Repo.all conn ~table:users ~decode)
+    ]} *)
+
+type 'conn multi_config = {
+  servers : string list;
+  max_size_per_server : int;
+  connect : string -> ('conn, string) result;
+  close : 'conn -> unit;
+  validate : ('conn -> bool) option;
+}
+
+type 'conn multi_t = {
+  pools : 'conn t array;
+  pool_conninfos : string array;
+  conn_to_pool : ('conn, int) Stdlib.Hashtbl.t;
+  healthy : bool Loc.t array;
+  next_index : int Loc.t;
+  closed : bool Loc.t;
+}
+
+let create_multi config =
+  if config.servers = [] then
+    invalid_arg "Pool.create_multi: servers list cannot be empty";
+  let n = List.length config.servers in
+  let pools =
+    Array.of_list
+      (List.map
+         (fun conninfo ->
+           let pool_config =
+             {
+               max_size = config.max_size_per_server;
+               connect = (fun () -> config.connect conninfo);
+               close = config.close;
+               validate = config.validate;
+             }
+           in
+           create pool_config)
+         config.servers)
+  in
+  let healthy = Array.init n (fun _ -> Loc.make true) in
+  {
+    pools;
+    pool_conninfos = Array.of_list config.servers;
+    conn_to_pool = Stdlib.Hashtbl.create (n * config.max_size_per_server);
+    healthy;
+    next_index = Loc.make 0;
+    closed = Loc.make false;
+  }
+
+let multi_size multi =
+  Array.fold_left (fun acc pool -> acc + size pool) 0 multi.pools
+
+let multi_in_use multi =
+  Array.fold_left (fun acc pool -> acc + in_use pool) 0 multi.pools
+
+let multi_available multi =
+  Array.fold_left (fun acc pool -> acc + available pool) 0 multi.pools
+
+let multi_is_closed multi = Loc.get multi.closed
+let multi_server_count multi = Array.length multi.pools
+
+let multi_is_healthy multi idx =
+  if idx < 0 || idx >= Array.length multi.healthy then false
+  else Loc.get multi.healthy.(idx)
+
+let mark_unhealthy multi idx =
+  if idx >= 0 && idx < Array.length multi.healthy then
+    Loc.set multi.healthy.(idx) false
+
+let mark_healthy multi idx =
+  if idx >= 0 && idx < Array.length multi.healthy then
+    Loc.set multi.healthy.(idx) true
+
+let count_healthy multi =
+  Array.fold_left
+    (fun acc h -> if Loc.get h then acc + 1 else acc)
+    0 multi.healthy
+
+(** Find next healthy server using round-robin *)
+let next_healthy_index multi =
+  let n = Array.length multi.pools in
+  let start = Loc.fetch_and_add multi.next_index 1 mod n in
+  let rec find_healthy attempts idx =
+    if attempts >= n then None
+    else if Loc.get multi.healthy.(idx) then Some idx
+    else find_healthy (attempts + 1) ((idx + 1) mod n)
+  in
+  find_healthy 0 start
+
+let acquire_multi multi =
+  if Loc.get multi.closed then Error Pool_closed
+  else
+    match next_healthy_index multi with
+    | None -> Error Pool_empty (* All servers unhealthy *)
+    | Some idx -> (
+        match acquire multi.pools.(idx) with
+        | Ok conn ->
+            Stdlib.Hashtbl.replace multi.conn_to_pool conn idx;
+            Ok conn
+        | Error Pool_empty ->
+            (* This server's pool is exhausted, try others *)
+            let n = Array.length multi.pools in
+            let rec try_others attempts current_idx =
+              if attempts >= n then Error Pool_empty
+              else if current_idx <> idx && Loc.get multi.healthy.(current_idx)
+              then
+                match acquire multi.pools.(current_idx) with
+                | Ok conn ->
+                    Stdlib.Hashtbl.replace multi.conn_to_pool conn current_idx;
+                    Ok conn
+                | Error Pool_empty ->
+                    try_others (attempts + 1) ((current_idx + 1) mod n)
+                | Error e -> Error e
+              else try_others (attempts + 1) ((current_idx + 1) mod n)
+            in
+            try_others 1 ((idx + 1) mod n)
+        | Error e -> Error e)
+
+let release_multi multi conn =
+  match Stdlib.Hashtbl.find_opt multi.conn_to_pool conn with
+  | Some idx ->
+      Stdlib.Hashtbl.remove multi.conn_to_pool conn;
+      if Loc.get multi.closed then multi.pools.(idx).config.close conn
+      else release multi.pools.(idx) conn
+  | None ->
+      (* Connection not tracked, close it directly using first pool's config *)
+      if Array.length multi.pools > 0 then multi.pools.(0).config.close conn
+
+let with_connection_multi multi f =
+  match acquire_multi multi with
+  | Error e -> Error e
+  | Ok conn -> (
+      match f conn with
+      | result ->
+          release_multi multi conn;
+          Ok result
+      | exception exn ->
+          release_multi multi conn;
+          raise exn)
+
+let rec acquire_multi_blocking ?timeoutf multi =
+  if Loc.get multi.closed then Error Pool_closed
+  else
+    match acquire_multi multi with
+    | Ok conn -> Ok conn
+    | Error Pool_empty -> (
+        (* All pools exhausted, wait on a healthy one *)
+        match next_healthy_index multi with
+        | None -> Error Pool_empty
+        | Some idx -> (
+            match acquire_blocking ?timeoutf multi.pools.(idx) with
+            | Ok conn ->
+                Stdlib.Hashtbl.replace multi.conn_to_pool conn idx;
+                Ok conn
+            | Error Pool_timeout -> Error Pool_timeout
+            | Error Pool_closed -> Error Pool_closed
+            | Error Pool_empty -> acquire_multi_blocking ?timeoutf multi
+            | Error e -> Error e))
+    | Error e -> Error e
+
+let with_connection_multi_blocking ?timeoutf multi f =
+  match acquire_multi_blocking ?timeoutf multi with
+  | Error e -> Error e
+  | Ok conn -> (
+      match f conn with
+      | result ->
+          release_multi multi conn;
+          Ok result
+      | exception exn ->
+          release_multi multi conn;
+          raise exn)
+
+let drain_multi multi = Array.iter drain multi.pools
+
+let shutdown_multi multi =
+  Loc.set multi.closed true;
+  Array.iter shutdown multi.pools;
+  Stdlib.Hashtbl.clear multi.conn_to_pool
+
+type multi_stats = {
+  total_servers : int;
+  healthy_servers : int;
+  per_server : (string * stats) list;
+  aggregate : stats;
+}
+
+let stats_multi multi =
+  let per_server =
+    Array.to_list
+      (Array.mapi
+         (fun i pool -> (multi.pool_conninfos.(i), stats pool))
+         multi.pools)
+  in
+  let aggregate =
+    {
+      total = multi_size multi;
+      available = multi_available multi;
+      in_use = multi_in_use multi;
+      closed = multi_is_closed multi;
+    }
+  in
+  {
+    total_servers = Array.length multi.pools;
+    healthy_servers = count_healthy multi;
+    per_server;
+    aggregate;
+  }
+
 module Make (D : Driver.S) = struct
   type nonrec t = D.connection t
   type conn = D.connection
···
   let available = available
   let in_use = in_use
   let is_closed = is_closed
+
+  module Multi = struct
+    type t = D.connection multi_t
+
+    let create ~servers ?validate () =
+      let cfg =
+        {
+          servers;
+          max_size_per_server = 10;
+          connect =
+            (fun conninfo ->
+              D.connect conninfo |> Result.map_error D.error_message);
+          close = D.close;
+          validate;
+        }
+      in
+      create_multi cfg
+
+    let create_sized ~servers ~max_size_per_server ?validate () =
+      let cfg =
+        {
+          servers;
+          max_size_per_server;
+          connect =
+            (fun conninfo ->
+              D.connect conninfo |> Result.map_error D.error_message);
+          close = D.close;
+          validate;
+        }
+      in
+      create_multi cfg
+
+    let acquire = acquire_multi
+    let acquire_blocking = acquire_multi_blocking
+    let release = release_multi
+    let with_connection = with_connection_multi
+    let with_connection_blocking = with_connection_multi_blocking
+    let drain = drain_multi
+    let shutdown = shutdown_multi
+    let stats = stats_multi
+    let size = multi_size
+    let available = multi_available
+    let in_use = multi_in_use
+    let is_closed = multi_is_closed
+    let server_count = multi_server_count
+    let is_healthy = multi_is_healthy
+    let mark_healthy = mark_healthy
+    let mark_unhealthy = mark_unhealthy
+  end
 end

repodb-postgresql.opam

···
 # This file is generated by dune, edit dune-project instead
 opam-version: "2.0"
-version: "0.2.0"
+version: "0.2.1"
 synopsis: "PostgreSQL driver for repodb"
 description:
   "PostgreSQL driver implementation for repodb using the postgresql-ocaml library."

repodb-sqlite.opam

···
 # This file is generated by dune, edit dune-project instead
 opam-version: "2.0"
-version: "0.2.0"
+version: "0.2.1"
 synopsis: "SQLite driver for repodb"
 description:
   "SQLite driver implementation for repodb using the sqlite3-ocaml library."

repodb.opam

···
 # This file is generated by dune, edit dune-project instead
 opam-version: "2.0"
-version: "0.2.0"
+version: "0.2.1"
 synopsis: "Ecto-like database toolkit for OCaml"
 description:
   "repodb is a database toolkit inspired by Elixir's Ecto. It provides type-safe query building, schema definitions, changesets for data validation, and database migrations."

test/dune

···
   test_multi
   test_stream
   test_pool
+  test_multi_pool
+  test_cqrs
   test_repodb))
 
 (test

test/integration/docker-compose.yml

···
+version: '3.8'
+
+services:
+  primary:
+    image: postgres:16
+    container_name: repodb_primary
+    environment:
+      POSTGRES_USER: repodb
+      POSTGRES_PASSWORD: repodb
+      POSTGRES_DB: repodb_test
+    ports:
+      - '5432:5432'
+    command: >
+      postgres
+      -c wal_level=replica
+      -c max_wal_senders=3
+      -c max_replication_slots=3
+      -c hot_standby=on
+      -c listen_addresses='*'
+    volumes:
+      - ./init-primary.sh:/docker-entrypoint-initdb.d/init-primary.sh
+    healthcheck:
+      test: ['CMD-SHELL', 'pg_isready -U repodb -d repodb_test']
+      interval: 5s
+      timeout: 5s
+      retries: 10
+
+  replica1:
+    image: postgres:16
+    container_name: repodb_replica1
+    environment:
+      POSTGRES_USER: repodb
+      POSTGRES_PASSWORD: repodb
+      PGUSER: repodb
+      PGPASSWORD: repodb
+    ports:
+      - '5433:5432'
+    depends_on:
+      primary:
+        condition: service_healthy
+    entrypoint: /bin/bash
+    command:
+      - -c
+      - |
+        rm -rf /var/lib/postgresql/data/*
+        until PGPASSWORD=repodb pg_basebackup -h primary -D /var/lib/postgresql/data -U repodb -Fp -Xs -P -R; do
+          echo "Waiting for primary..."
+          sleep 2
+        done
+        chmod 700 /var/lib/postgresql/data
+        exec postgres
+    healthcheck:
+      test: ['CMD-SHELL', 'pg_isready -U repodb']
+      interval: 5s
+      timeout: 5s
+      retries: 10
+
+  replica2:
+    image: postgres:16
+    container_name: repodb_replica2
+    environment:
+      POSTGRES_USER: repodb
+      POSTGRES_PASSWORD: repodb
+      PGUSER: repodb
+      PGPASSWORD: repodb
+    ports:
+      - '5434:5432'
+    depends_on:
+      primary:
+        condition: service_healthy
+    entrypoint: /bin/bash
+    command:
+      - -c
+      - |
+        rm -rf /var/lib/postgresql/data/*
+        until PGPASSWORD=repodb pg_basebackup -h primary -D /var/lib/postgresql/data -U repodb -Fp -Xs -P -R; do
+          echo "Waiting for primary..."
+          sleep 2
+        done
+        chmod 700 /var/lib/postgresql/data
+        exec postgres
+    healthcheck:
+      test: ['CMD-SHELL', 'pg_isready -U repodb']
+      interval: 5s
+      timeout: 5s
+      retries: 10

test/integration/dune

···
+(executable
+ (name test_cqrs_integration)
+ (libraries repodb repodb-postgresql unix))

test/integration/init-primary.sh

···
+#!/bin/bash
+set -e
+
+# Configure pg_hba.conf to allow replication connections
+echo "host replication repodb 0.0.0.0/0 md5" >> "$PGDATA/pg_hba.conf"
+echo "host all repodb 0.0.0.0/0 md5" >> "$PGDATA/pg_hba.conf"
+
+# Create test table
+psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" <<-EOSQL
+    CREATE TABLE IF NOT EXISTS test_users (
+        id SERIAL PRIMARY KEY,
+        name VARCHAR(100) NOT NULL,
+        email VARCHAR(255) NOT NULL,
+        created_at TIMESTAMP DEFAULT NOW()
+    );
+EOSQL

test/integration/run_tests.sh

···
+#!/bin/bash
+set -e
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
+
+cleanup() {
+    echo "Cleaning up..."
+    cd "$SCRIPT_DIR"
+    podman-compose down -v 2>/dev/null || docker-compose down -v 2>/dev/null || true
+}
+
+trap cleanup EXIT
+
+echo "=== Starting PostgreSQL cluster ==="
+cd "$SCRIPT_DIR"
+chmod +x init-primary.sh
+
+if command -v podman-compose &> /dev/null; then
+    COMPOSE_CMD="podman-compose"
+elif command -v docker-compose &> /dev/null; then
+    COMPOSE_CMD="docker-compose"
+else
+    echo "Error: Neither podman-compose nor docker-compose found"
+    exit 1
+fi
+
+$COMPOSE_CMD down -v 2>/dev/null || true
+$COMPOSE_CMD up -d
+
+echo "=== Waiting for primary to be ready ==="
+for i in {1..30}; do
+    if $COMPOSE_CMD exec -T primary pg_isready -U repodb -d repodb_test 2>/dev/null; then
+        echo "Primary is ready"
+        break
+    fi
+    echo "Waiting for primary... ($i/30)"
+    sleep 2
+done
+
+echo "=== Waiting for replicas to sync ==="
+sleep 15
+
+echo "=== Verifying replication ==="
+$COMPOSE_CMD exec -T primary psql -U repodb -d repodb_test -c "SELECT client_addr, state FROM pg_stat_replication;" || true
+
+echo "=== Running integration tests ==="
+cd "$PROJECT_ROOT"
+dune exec test/integration/test_cqrs_integration.exe
+
+echo "=== Tests completed successfully ==="

test/integration/test_cqrs_integration.ml

···
+open Repodb
+module PgCqrs = Cqrs.Make (Repodb_postgresql)
+module Repo = Repo.Make (Repodb_postgresql)
+
+let primary_conninfo =
+  "host=localhost port=5432 dbname=repodb_test user=repodb password=repodb"
+
+let replica1_conninfo =
+  "host=localhost port=5433 dbname=repodb_test user=repodb password=repodb"
+
+let replica2_conninfo =
+  "host=localhost port=5434 dbname=repodb_test user=repodb password=repodb"
+
+let _users_table = Schema.table "test_users"
+
+type user = { id : int; name : string; email : string [@warning "-69"] }
+
+let decode_user row =
+  {
+    id = Driver.row_int row 0;
+    name = Driver.row_text row 1;
+    email = Driver.row_text row 2;
+  }
+
+let test_write_to_primary_read_from_replica () =
+  Printf.printf "=== Test: Write to primary, read from replica ===\n%!";
+  let cqrs =
+    PgCqrs.create
+      {
+        primary_conninfo;
+        primary_max_size = 5;
+        replica_conninfos = [ replica1_conninfo; replica2_conninfo ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  let test_name = Printf.sprintf "test_user_%d" (Random.int 10000) in
+  let test_email = Printf.sprintf "[e-mail]" test_name in
+
+  let insert_result =
+    PgCqrs.with_write cqrs (fun conn ->
+        let sql =
+          Printf.sprintf
+            "INSERT INTO test_users (name, email) VALUES ('%s', '%s') \
+             RETURNING id"
+            test_name test_email
+        in
+        Repodb_postgresql.query_one conn sql ~params:[||])
+  in
+  (match insert_result with
+  | Ok (Ok (Some row)) ->
+      let id = Driver.row_int row 0 in
+      Printf.printf "  Inserted user with id=%d\n%!" id
+  | _ -> failwith "Insert failed");
+
+  Unix.sleepf 2.0;
+
+  let read_result =
+    PgCqrs.with_read cqrs (fun conn ->
+        let sql =
+          Printf.sprintf
+            "SELECT id, name, email FROM test_users WHERE name = '%s'" test_name
+        in
+        Repodb_postgresql.query_one conn sql ~params:[||])
+  in
+  (match read_result with
+  | Ok (Ok (Some row)) ->
+      let user = decode_user row in
+      Printf.printf "  Read user from replica: id=%d, name=%s\n%!" user.id
+        user.name;
+      assert (user.name = test_name)
+  | _ -> failwith "Read from replica failed");
+
+  PgCqrs.shutdown cqrs;
+  Printf.printf "  PASSED\n%!"
+
+let test_transaction_uses_primary () =
+  Printf.printf "=== Test: Transaction uses primary ===\n%!";
+  let cqrs =
+    PgCqrs.create
+      {
+        primary_conninfo;
+        primary_max_size = 5;
+        replica_conninfos = [ replica1_conninfo ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  let test_name = Printf.sprintf "tx_user_%d" (Random.int 10000) in
+
+  let tx_result =
+    PgCqrs.transaction cqrs (fun conn ->
+        let insert_sql =
+          Printf.sprintf
+            "INSERT INTO test_users (name, email) VALUES ('%s', '[e-mail]')"
+            test_name test_name
+        in
+        match Repodb_postgresql.exec conn insert_sql ~params:[||] with
+        | Ok () -> (
+            let select_sql =
+              Printf.sprintf
+                "SELECT id, name, email FROM test_users WHERE name = '%s'"
+                test_name
+            in
+            match Repodb_postgresql.query_one conn select_sql ~params:[||] with
+            | Ok (Some row) ->
+                let user = decode_user row in
+                Ok user
+            | Ok None -> Error Error.Not_found
+            | Error e ->
+                Error (Error.Query_failed (Repodb_postgresql.error_message e)))
+        | Error e ->
+            Error (Error.Query_failed (Repodb_postgresql.error_message e)))
+  in
+  (match tx_result with
+  | Ok user ->
+      Printf.printf "  Transaction completed: id=%d, name=%s\n%!" user.id
+        user.name;
+      assert (user.name = test_name)
+  | Error e ->
+      Printf.printf "  Error: %s\n%!" (Error.show_db_error e);
+      failwith "Transaction failed");
+
+  PgCqrs.shutdown cqrs;
+  Printf.printf "  PASSED\n%!"
+
+module TestPool = Pool.Make (Repodb_postgresql)
+
+let test_multi_pool_round_robin () =
+  Printf.printf "=== Test: MultiPool round-robin ===\n%!";
+  let multi =
+    TestPool.Multi.create_sized
+      ~servers:[ primary_conninfo; replica1_conninfo; replica2_conninfo ]
+      ~max_size_per_server:3 ()
+  in
+  let successful_queries = ref 0 in
+  for _ = 1 to 9 do
+    match
+      TestPool.Multi.with_connection multi (fun conn ->
+          match Repodb_postgresql.query_one conn "SELECT 1" ~params:[||] with
+          | Ok _ -> incr successful_queries
+          | Error _ -> ())
+    with
+    | Ok () -> ()
+    | Error e -> Printf.printf "  Warning: %s\n%!" (Pool.error_to_string e)
+  done;
+  Printf.printf "  Successful queries: %d/9\n%!" !successful_queries;
+  assert (!successful_queries >= 6);
+  TestPool.Multi.shutdown multi;
+  Printf.printf "  PASSED\n%!"
+
+let test_cqrs_fallback_when_replica_down () =
+  Printf.printf "=== Test: CQRS fallback when replicas marked unhealthy ===\n%!";
+  let cqrs =
+    PgCqrs.create
+      {
+        primary_conninfo;
+        primary_max_size = 5;
+        replica_conninfos = [ replica1_conninfo; replica2_conninfo ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  PgCqrs.mark_replica_unhealthy cqrs 0;
+  PgCqrs.mark_replica_unhealthy cqrs 1;
+
+  let read_result =
+    PgCqrs.with_read cqrs (fun conn ->
+        Repodb_postgresql.query_one conn "SELECT 1" ~params:[||])
+  in
+  (match read_result with
+  | Ok (Ok _) -> Printf.printf "  Read succeeded via fallback to primary\n%!"
+  | _ -> failwith "Fallback read failed");
+
+  PgCqrs.shutdown cqrs;
+  Printf.printf "  PASSED\n%!"
+
+let test_concurrent_load () =
+  Printf.printf "=== Test: Concurrent load ===\n%!";
+  let cqrs =
+    PgCqrs.create
+      {
+        primary_conninfo;
+        primary_max_size = 10;
+        replica_conninfos = [ replica1_conninfo; replica2_conninfo ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  let success_count = Atomic.make 0 in
+  let error_count = Atomic.make 0 in
+  let num_domains = 4 in
+  let ops_per_domain = 25 in
+
+  let domains =
+    Array.init num_domains (fun i ->
+        Domain.spawn (fun () ->
+            for j = 1 to ops_per_domain do
+              let op = ((i * ops_per_domain) + j) mod 3 in
+              if op = 0 then
+                match
+                  PgCqrs.with_write cqrs (fun conn ->
+                      Repodb_postgresql.exec conn
+                        (Printf.sprintf
+                           "INSERT INTO test_users (name, email) VALUES \
+                            ('load_%d_%d', '[e-mail]')"
+                           i j)
+                        ~params:[||])
+                with
+                | Ok (Ok ()) -> Atomic.incr success_count
+                | _ -> Atomic.incr error_count
+              else
+                match
+                  PgCqrs.with_read cqrs (fun conn ->
+                      Repodb_postgresql.query_one conn
+                        "SELECT COUNT(*) FROM test_users" ~params:[||])
+                with
+                | Ok (Ok _) -> Atomic.incr success_count
+                | _ -> Atomic.incr error_count
+            done))
+  in
+  Array.iter Domain.join domains;
+
+  let total_success = Atomic.get success_count in
+  let total_errors = Atomic.get error_count in
+  Printf.printf "  Successful: %d, Errors: %d\n%!" total_success total_errors;
+  assert (total_success > num_domains * ops_per_domain * 8 / 10);
+
+  PgCqrs.shutdown cqrs;
+  Printf.printf "  PASSED\n%!"
+
+let () =
+  Random.self_init ();
+  Printf.printf "\n========================================\n";
+  Printf.printf "repodb CQRS Integration Tests\n";
+  Printf.printf "========================================\n\n%!";
+
+  test_write_to_primary_read_from_replica ();
+  test_transaction_uses_primary ();
+  test_multi_pool_round_robin ();
+  test_cqrs_fallback_when_replica_down ();
+  test_concurrent_load ();
+
+  Printf.printf "\n========================================\n";
+  Printf.printf "All integration tests PASSED!\n";
+  Printf.printf "========================================\n%!"

test/test_cqrs.ml

···
+open Repodb
+
+type mock_conn = { id : int; mutable closed : bool; server : string }
+
+let conn_counter = Atomic.make 0
+
+module MockDriver : Driver.S with type connection = mock_conn = struct
+  type connection = mock_conn
+  type error = string
+
+  let dialect = Driver.PostgreSQL
+  let error_message e = e
+
+  let connect conninfo =
+    let id = Atomic.fetch_and_add conn_counter 1 in
+    Ok { id; closed = false; server = conninfo }
+
+  let close conn = conn.closed <- true
+
+  let with_connection conninfo f =
+    match connect conninfo with
+    | Error e -> Error e
+    | Ok conn ->
+        let result = f conn in
+        close conn;
+        result
+
+  let exec _conn _sql ~params:_ = Ok ()
+  let query _conn _sql ~params:_ = Ok []
+  let query_one _conn _sql ~params:_ = Ok None
+  let query_fold _conn _sql ~params:_ ~init ~f:_ = Ok init
+  let query_iter _conn _sql ~params:_ ~f:_ = Ok ()
+  let transaction conn f = f conn
+  let placeholder n = Printf.sprintf "$%d" n
+  let returning_supported = true
+  let upsert_syntax = `PostgreSQL
+  let last_insert_id _conn = Error "Use RETURNING"
+end
+
+module TestCqrs = Cqrs.Make (MockDriver)
+
+let test_create_no_replicas () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  Alcotest.(check bool) "no replicas" false (TestCqrs.has_replicas cqrs);
+  Alcotest.(check int) "healthy count" 0 (TestCqrs.count_healthy_replicas cqrs);
+  TestCqrs.shutdown cqrs
+
+let test_create_with_replicas () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [ "replica1"; "replica2" ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  Alcotest.(check bool) "has replicas" true (TestCqrs.has_replicas cqrs);
+  Alcotest.(check int) "healthy count" 2 (TestCqrs.count_healthy_replicas cqrs);
+  TestCqrs.shutdown cqrs
+
+let test_with_read_uses_replica () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [ "replica1" ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  let server_used = ref "" in
+  let result =
+    TestCqrs.with_read cqrs (fun conn ->
+        server_used := conn.server;
+        42)
+  in
+  (match result with Ok 42 -> () | _ -> Alcotest.fail "Expected Ok 42");
+  Alcotest.(check string) "read used replica" "replica1" !server_used;
+  TestCqrs.shutdown cqrs
+
+let test_with_write_uses_primary () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [ "replica1" ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  let server_used = ref "" in
+  let result =
+    TestCqrs.with_write cqrs (fun conn ->
+        server_used := conn.server;
+        42)
+  in
+  (match result with Ok 42 -> () | _ -> Alcotest.fail "Expected Ok 42");
+  Alcotest.(check string) "write used primary" "primary" !server_used;
+  TestCqrs.shutdown cqrs
+
+let test_with_primary_uses_primary () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [ "replica1" ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  let server_used = ref "" in
+  let result =
+    TestCqrs.with_primary cqrs (fun conn ->
+        server_used := conn.server;
+        42)
+  in
+  (match result with Ok 42 -> () | _ -> Alcotest.fail "Expected Ok 42");
+  Alcotest.(check string) "with_primary used primary" "primary" !server_used;
+  TestCqrs.shutdown cqrs
+
+let test_no_replicas_read_falls_back_to_primary () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  let server_used = ref "" in
+  let result =
+    TestCqrs.with_read cqrs (fun conn ->
+        server_used := conn.server;
+        42)
+  in
+  (match result with Ok 42 -> () | _ -> Alcotest.fail "Expected Ok 42");
+  Alcotest.(check string) "read fallback to primary" "primary" !server_used;
+  TestCqrs.shutdown cqrs
+
+let test_round_robin_replicas () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [ "r1"; "r2"; "r3" ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  let servers = ref [] in
+  for _ = 1 to 6 do
+    let _ =
+      TestCqrs.with_read cqrs (fun conn ->
+          servers := conn.server :: !servers;
+          ())
+    in
+    ()
+  done;
+  let counts =
+    [ "r1"; "r2"; "r3" ]
+    |> List.map (fun s -> List.length (List.filter (( = ) s) !servers))
+  in
+  Alcotest.(check int) "r1 count" 2 (List.nth counts 0);
+  Alcotest.(check int) "r2 count" 2 (List.nth counts 1);
+  Alcotest.(check int) "r3 count" 2 (List.nth counts 2);
+  TestCqrs.shutdown cqrs
+
+let test_unhealthy_replica_skipped () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [ "r1"; "r2" ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  TestCqrs.mark_replica_unhealthy cqrs 0;
+  let servers = ref [] in
+  for _ = 1 to 3 do
+    let _ =
+      TestCqrs.with_read cqrs (fun conn ->
+          servers := conn.server :: !servers;
+          ())
+    in
+    ()
+  done;
+  let r1_count = List.length (List.filter (( = ) "r1") !servers) in
+  let r2_count = List.length (List.filter (( = ) "r2") !servers) in
+  Alcotest.(check int) "r1 should be skipped" 0 r1_count;
+  Alcotest.(check int) "r2 should handle all" 3 r2_count;
+  TestCqrs.shutdown cqrs
+
+let test_all_replicas_unhealthy_fallback () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [ "r1"; "r2" ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  TestCqrs.mark_replica_unhealthy cqrs 0;
+  TestCqrs.mark_replica_unhealthy cqrs 1;
+  let server_used = ref "" in
+  let result =
+    TestCqrs.with_read cqrs (fun conn ->
+        server_used := conn.server;
+        42)
+  in
+  (match result with Ok 42 -> () | _ -> Alcotest.fail "Expected Ok 42");
+  Alcotest.(check string) "fallback to primary" "primary" !server_used;
+  TestCqrs.shutdown cqrs
+
+let test_transaction_uses_primary () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [ "r1" ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  let server_used = ref "" in
+  let result =
+    TestCqrs.transaction cqrs (fun conn ->
+        server_used := conn.server;
+        Ok 42)
+  in
+  (match result with Ok 42 -> () | _ -> Alcotest.fail "Expected Ok 42");
+  Alcotest.(check string) "transaction used primary" "primary" !server_used;
+  TestCqrs.shutdown cqrs
+
+let test_stats () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [ "r1"; "r2" ];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  let stats = TestCqrs.stats cqrs in
+  Alcotest.(check int) "healthy count" 2 stats.healthy_replica_count;
+  (match stats.replicas with
+  | Some replicas ->
+      Alcotest.(check int) "replica pools" 2 (List.length replicas)
+  | None -> Alcotest.fail "Expected replicas");
+  TestCqrs.shutdown cqrs
+
+let test_route_query_type () =
+  Alcotest.(check bool)
+    "SELECT is Read" true
+    (TestCqrs.route_query_type Query.Select = Cqrs.Read);
+  Alcotest.(check bool)
+    "INSERT is Write" true
+    (TestCqrs.route_query_type Query.Insert = Cqrs.Write);
+  Alcotest.(check bool)
+    "UPDATE is Write" true
+    (TestCqrs.route_query_type Query.Update = Cqrs.Write);
+  Alcotest.(check bool)
+    "DELETE is Write" true
+    (TestCqrs.route_query_type Query.Delete = Cqrs.Write)
+
+let test_with_replica_no_replicas () =
+  let cqrs =
+    TestCqrs.create
+      {
+        primary_conninfo = "primary";
+        primary_max_size = 5;
+        replica_conninfos = [];
+        replica_max_size_each = 5;
+        replica_selection = Cqrs.RoundRobin;
+        validate = None;
+      }
+  in
+  let result = TestCqrs.with_replica cqrs (fun _ -> 42) in
+  (match result with
+  | Error Pool.Pool_empty -> ()
+  | _ -> Alcotest.fail "Expected Pool_empty when no replicas");
+  TestCqrs.shutdown cqrs
+
+let tests =
+  [
+    ("create no replicas", `Quick, test_create_no_replicas);
+    ("create with replicas", `Quick, test_create_with_replicas);
+    ("with_read uses replica", `Quick, test_with_read_uses_replica);
+    ("with_write uses primary", `Quick, test_with_write_uses_primary);
+    ("with_primary uses primary", `Quick, test_with_primary_uses_primary);
+    ( "no replicas read fallback",
+      `Quick,
+      test_no_replicas_read_falls_back_to_primary );
+    ("round robin replicas", `Quick, test_round_robin_replicas);
+    ("unhealthy replica skipped", `Quick, test_unhealthy_replica_skipped);
+    ( "all replicas unhealthy fallback",
+      `Quick,
+      test_all_replicas_unhealthy_fallback );
+    ("transaction uses primary", `Quick, test_transaction_uses_primary);
+    ("stats", `Quick, test_stats);
+    ("route_query_type", `Quick, test_route_query_type);
+    ("with_replica no replicas", `Quick, test_with_replica_no_replicas);
+  ]

test/test_multi_pool.ml

···
+open Repodb
+
+type mock_conn = { id : int; server_idx : int; mutable closed : bool }
+
+let conn_counter = Atomic.make 0
+
+let find_index_in_list elem lst =
+  let rec aux i = function
+    | [] -> None
+    | x :: xs -> if x = elem then Some i else aux (i + 1) xs
+  in
+  aux 0 lst
+
+let mock_multi_config ?(servers = [ "s1"; "s2"; "s3" ]) ?(max_size = 2)
+    ?(fail_servers = []) () =
+  Pool.
+    {
+      servers;
+      max_size_per_server = max_size;
+      connect =
+        (fun conninfo ->
+          let server_idx =
+            match find_index_in_list conninfo servers with
+            | Some i -> i
+            | None -> -1
+          in
+          if List.mem server_idx fail_servers then
+            Error (Printf.sprintf "Failed to connect to %s" conninfo)
+          else
+            let id = Atomic.fetch_and_add conn_counter 1 in
+            Ok { id; server_idx; closed = false });
+      close = (fun conn -> conn.closed <- true);
+      validate = None;
+    }
+
+let test_create_multi () =
+  let multi = Pool.create_multi (mock_multi_config ()) in
+  Alcotest.(check int) "initial size" 0 (Pool.multi_size multi);
+  Alcotest.(check int) "initial available" 0 (Pool.multi_available multi);
+  Alcotest.(check int) "initial in_use" 0 (Pool.multi_in_use multi);
+  Alcotest.(check bool) "not closed" false (Pool.multi_is_closed multi);
+  Alcotest.(check int) "server count" 3 (Pool.multi_server_count multi)
+
+let test_create_multi_empty_servers () =
+  try
+    let _ = Pool.create_multi (mock_multi_config ~servers:[] ()) in
+    Alcotest.fail "Expected Invalid_argument"
+  with Invalid_argument _ -> ()
+
+let test_round_robin_distribution () =
+  let multi = Pool.create_multi (mock_multi_config ~max_size:10 ()) in
+  let counts = [| 0; 0; 0 |] in
+  for _ = 1 to 9 do
+    match Pool.acquire_multi multi with
+    | Ok conn -> counts.(conn.server_idx) <- counts.(conn.server_idx) + 1
+    | Error _ -> Alcotest.fail "Unexpected acquire error"
+  done;
+  Alcotest.(check int) "server 0 count" 3 counts.(0);
+  Alcotest.(check int) "server 1 count" 3 counts.(1);
+  Alcotest.(check int) "server 2 count" 3 counts.(2);
+  Pool.shutdown_multi multi
+
+let test_health_check_skipping () =
+  let multi = Pool.create_multi (mock_multi_config ~max_size:10 ()) in
+  Pool.mark_unhealthy multi 1;
+  let counts = [| 0; 0; 0 |] in
+  for _ = 1 to 6 do
+    match Pool.acquire_multi multi with
+    | Ok conn -> counts.(conn.server_idx) <- counts.(conn.server_idx) + 1
+    | Error _ -> Alcotest.fail "Unexpected acquire error"
+  done;
+  Alcotest.(check int) "server 1 should be skipped" 0 counts.(1);
+  Alcotest.(check bool)
+    "server 0 or 2 got connections" true
+    (counts.(0) > 0 || counts.(2) > 0);
+  Pool.shutdown_multi multi
+
+let test_all_unhealthy () =
+  let multi = Pool.create_multi (mock_multi_config ()) in
+  Pool.mark_unhealthy multi 0;
+  Pool.mark_unhealthy multi 1;
+  Pool.mark_unhealthy multi 2;
+  (match Pool.acquire_multi multi with
+  | Error Pool.Pool_empty -> ()
+  | _ -> Alcotest.fail "Expected Pool_empty when all servers unhealthy");
+  Pool.shutdown_multi multi
+
+let test_mark_healthy_recovery () =
+  let multi = Pool.create_multi (mock_multi_config ~max_size:10 ()) in
+  Pool.mark_unhealthy multi 0;
+  Pool.mark_unhealthy multi 1;
+  Pool.mark_unhealthy multi 2;
+  (match Pool.acquire_multi multi with
+  | Error Pool.Pool_empty -> ()
+  | _ -> Alcotest.fail "Expected Pool_empty");
+  Pool.mark_healthy multi 1;
+  (match Pool.acquire_multi multi with
+  | Ok conn -> Alcotest.(check int) "should use server 1" 1 conn.server_idx
+  | Error _ -> Alcotest.fail "Expected successful acquire after marking healthy");
+  Pool.shutdown_multi multi
+
+let test_with_connection_multi () =
+  let multi = Pool.create_multi (mock_multi_config ()) in
+  let result =
+    Pool.with_connection_multi multi (fun conn ->
+        Alcotest.(check int) "in_use during work" 1 (Pool.multi_in_use multi);
+        conn.id + 100)
+  in
+  (match result with
+  | Ok n ->
+      Alcotest.(check bool) "got result" true (n >= 100);
+      Alcotest.(check int) "released after" 0 (Pool.multi_in_use multi)
+  | Error _ -> Alcotest.fail "Expected successful with_connection_multi");
+  Pool.shutdown_multi multi
+
+let test_with_connection_multi_exception () =
+  let multi = Pool.create_multi (mock_multi_config ()) in
+  (try
+     let _ =
+       Pool.with_connection_multi multi (fun _ ->
+           Alcotest.(check int) "in_use" 1 (Pool.multi_in_use multi);
+           failwith "boom")
+     in
+     ()
+   with Failure _ -> ());
+  Alcotest.(check int) "released after exception" 0 (Pool.multi_in_use multi);
+  Pool.shutdown_multi multi
+
+let test_shutdown_multi () =
+  let multi = Pool.create_multi (mock_multi_config ()) in
+  let conn =
+    match Pool.acquire_multi multi with
+    | Ok c -> c
+    | Error _ -> Alcotest.fail "Expected successful acquire"
+  in
+  Pool.release_multi multi conn;
+  Pool.shutdown_multi multi;
+  Alcotest.(check bool) "pool closed" true (Pool.multi_is_closed multi);
+  match Pool.acquire_multi multi with
+  | Error Pool.Pool_closed -> ()
+  | _ -> Alcotest.fail "Expected Pool_closed after shutdown"
+
+let test_stats_multi () =
+  let multi = Pool.create_multi (mock_multi_config ~max_size:5 ()) in
+  let conn1 =
+    match Pool.acquire_multi multi with
+    | Ok c -> c
+    | Error _ -> Alcotest.fail "Expected successful acquire"
+  in
+  let conn2 =
+    match Pool.acquire_multi multi with
+    | Ok c -> c
+    | Error _ -> Alcotest.fail "Expected successful acquire"
+  in
+  let stats = Pool.stats_multi multi in
+  Alcotest.(check int) "total servers" 3 stats.total_servers;
+  Alcotest.(check int) "healthy servers" 3 stats.healthy_servers;
+  Alcotest.(check int) "aggregate in_use" 2 stats.aggregate.in_use;
+  Alcotest.(check int) "aggregate total" 2 stats.aggregate.total;
+  Pool.release_multi multi conn1;
+  Pool.release_multi multi conn2;
+  Pool.shutdown_multi multi
+
+let test_single_server_multi () =
+  let multi = Pool.create_multi (mock_multi_config ~servers:[ "single" ] ()) in
+  let conn =
+    match Pool.acquire_multi multi with
+    | Ok c -> c
+    | Error _ -> Alcotest.fail "Expected successful acquire"
+  in
+  Alcotest.(check int) "single server idx" 0 conn.server_idx;
+  Pool.release_multi multi conn;
+  Pool.shutdown_multi multi
+
+let test_connection_reuse () =
+  let multi =
+    Pool.create_multi (mock_multi_config ~servers:[ "s1" ] ~max_size:2 ())
+  in
+  let conn1 =
+    match Pool.acquire_multi multi with
+    | Ok c -> c
+    | Error _ -> Alcotest.fail "Expected successful acquire"
+  in
+  let id1 = conn1.id in
+  Pool.release_multi multi conn1;
+  let conn2 =
+    match Pool.acquire_multi multi with
+    | Ok c -> c
+    | Error _ -> Alcotest.fail "Expected successful acquire"
+  in
+  Alcotest.(check int) "same connection reused" id1 conn2.id;
+  Pool.release_multi multi conn2;
+  Pool.shutdown_multi multi
+
+let test_max_size_per_server () =
+  let multi =
+    Pool.create_multi (mock_multi_config ~servers:[ "s1" ] ~max_size:2 ())
+  in
+  let c1 =
+    match Pool.acquire_multi multi with
+    | Ok c -> c
+    | Error _ -> Alcotest.fail "Expected 1st acquire"
+  in
+  let c2 =
+    match Pool.acquire_multi multi with
+    | Ok c -> c
+    | Error _ -> Alcotest.fail "Expected 2nd acquire"
+  in
+  (match Pool.acquire_multi multi with
+  | Error Pool.Pool_empty -> ()
+  | _ -> Alcotest.fail "Expected Pool_empty at capacity");
+  Pool.release_multi multi c1;
+  Pool.release_multi multi c2;
+  Pool.shutdown_multi multi
+
+let test_fallback_on_pool_exhaustion () =
+  let multi = Pool.create_multi (mock_multi_config ~max_size:1 ()) in
+  let conns = ref [] in
+  for _ = 1 to 3 do
+    match Pool.acquire_multi multi with
+    | Ok c -> conns := c :: !conns
+    | Error _ -> Alcotest.fail "Expected acquire"
+  done;
+  Alcotest.(check int) "3 connections acquired" 3 (List.length !conns);
+  let servers_used =
+    List.sort_uniq compare (List.map (fun c -> c.server_idx) !conns)
+  in
+  Alcotest.(check int) "all 3 servers used" 3 (List.length servers_used);
+  List.iter (Pool.release_multi multi) !conns;
+  Pool.shutdown_multi multi
+
+let tests =
+  [
+    ("create multi", `Quick, test_create_multi);
+    ("create multi empty servers", `Quick, test_create_multi_empty_servers);
+    ("round robin distribution", `Quick, test_round_robin_distribution);
+    ("health check skipping", `Quick, test_health_check_skipping);
+    ("all unhealthy", `Quick, test_all_unhealthy);
+    ("mark healthy recovery", `Quick, test_mark_healthy_recovery);
+    ("with_connection_multi", `Quick, test_with_connection_multi);
+    ( "with_connection_multi exception",
+      `Quick,
+      test_with_connection_multi_exception );
+    ("shutdown_multi", `Quick, test_shutdown_multi);
+    ("stats_multi", `Quick, test_stats_multi);
+    ("single server multi", `Quick, test_single_server_multi);
+    ("connection reuse", `Quick, test_connection_reuse);
+    ("max size per server", `Quick, test_max_size_per_server);
+    ("fallback on pool exhaustion", `Quick, test_fallback_on_pool_exhaustion);
+  ]

test/test_repodb.ml

···
       ("Multi", Test_multi.tests);
       ("Stream", Test_stream.tests);
       ("Pool", Test_pool.tests);
+      ("MultiPool", Test_multi_pool.tests);
+      ("Cqrs", Test_cqrs.tests);
     ]