Database Testing in OCaml: From CRUD to Connection Pool Stress Testing

A comprehensive journey through implementing enterprise-grade database testing infrastructure for an OCaml web application, featuring integration tests, transaction verification, migration rollback testing, connection pool stress tests, and query performance benchmarks.

Project: Chaufr – Personal drivers, on demand, in your own vehicle
Tech Stack: OCaml, Alcotest, Lwt, Caqti, PostgreSQL, UUID v7

What started as a simple "add some tests" task evolved into building a comprehensive database testing infrastructure that could be similar to what you'd find in enterprise-grade applications. This weekend I document the implementation of a stable test suite covering everything from basic CRUD operations to connection pool stress testing and query performance benchmarking.

⚠️ Prerequisites and Setup Requirements

Database Requirements: This testing suite requires a running PostgreSQL instance for full test execution. The tests are designed to work with real database connections rather than mocks to ensure production-like validation.

Setup Requirements:

Graceful Degradation: When a database connection is unavailable, the test suite includes graceful degradation logic that skips database-dependent tests rather than failing the entire test run. This ensures CI/CD pipelines can continue even when database services are temporarily unavailable, though full validation requires a working database connection.

Sprint Overview: The Database Testing Infrastructure Build

Testing Philosophy

My approach was guided by these principles:

Before: Basic Integration Test

let test_crud_flow _ _ () =
  match Database.Connection.init () with
  | exception ex -> Printf.printf "DB init failed, skipping..."; Lwt.return ()
  | _ -> (* basic user creation test *)

Note: This early implementation demonstrates the graceful degradation approach - when database initialization fails, tests are skipped rather than failed, allowing the test suite to continue running even when PostgreSQL is unavailable. This pattern is maintained throughout the comprehensive test suite.

After: Comprehensive Test Suite 

let database_tests =
  [
    test_case "env_and_schema_ready" `Quick (fun _ _ -> ensure_db_ready ());
    test_case "user_crud" `Quick (fun s c -> test_user_crud s c ());
    test_case "transactions" `Quick (fun s c -> test_transactions s c ());
    test_case "migrations_roundtrip" `Slow (fun s c ->
        test_migrations_roundtrip s c ());
    test_case "migrations_roundtrip_chain" `Slow (fun s c ->
        test_migrations_roundtrip_chain s c ());
    test_case "connection_pool_stress" `Slow (fun s c ->
        test_connection_pool_stress s c ());
    test_case "query_performance_benchmarks" `Slow (fun s c ->
        test_query_performance_benchmarks s c ());
  ]

Key Accomplishments

1. 🏗️ Robust Test Infrastructure Setup

Challenge: Ensure tests run consistently across environments with proper database initialization.

Solution: Comprehensive environment detection and database readiness verification.

Environment-Aware Test Setup

let () =
  (match Sys.getenv_opt "DUNE_SOURCEROOT" with
  | Some root -> Unix.chdir root
  | None -> ());
  Dotenv.load ()

Database Readiness Verification 

let ensure_db_ready () =
  let* ping = Connection.ping_database () in
  (match ping with
  | Ok _ -> ()
  | Error err ->
      Alcotest.failf "Database ping failed: %s" (Connection.error_to_string err));
  let* init_res = Migrations.init () in
  (match init_res with
  | Ok () -> ()
  | Error err ->
      Alcotest.failf "Migrations init failed: %s"
        (Migrations.migration_error_to_string err));
  let* up_res =
    Migrations.run_migrations GeneratedList.all_migrations Migrations.Up
  in
  (match up_res with
  | Ok () -> ()
  | Error err ->
      Alcotest.failf "Applying migrations failed: %s"
        (Migrations.migration_error_to_string err));
  Lwt.return_unit

Benefits Achieved:

2. 🔄 Comprehensive CRUD Testing with UUID v7

Challenge: Test complete user lifecycle with modern UUID v7 identifiers.

Solution: End-to-end CRUD testing with proper UUID validation and model mapping.

UUID-Based Test Data Generation 

let generate_uuid () =
  Uuidm.v7_non_monotonic_gen
    ~now_ms:(fun () -> Int64.of_float (Unix.gettimeofday () *. 1000.0))
    (Stdlib.Random.State.make_self_init ())

let test_user_crud _ _ () =
  let* () = ensure_db_ready () in
  let uid_gen = generate_uuid () in
  (* Use a single word name to avoid splitting issues *)
  let name = "TestUser" ^ String.sub (Uuidm.to_string (uid_gen ())) 0 8 in
  let email = "test." ^ Uuidm.to_string (uid_gen ()) ^ "@example.com" in
  let phone = Some "555-0100" in

Rigorous Assertion Testing

let* fetched = UserService.get_user_by_id uid in
(match fetched with
| Ok (Some user) ->
    Alcotest.(check string) "name matches" name user.first_name;
    Alcotest.(check string) "second_name should be empty" "" user.second_name;
    Alcotest.(check string) "email matches" email user.email;
    Alcotest.(check (option string)) "phone matches" phone user.phone
| Ok None -> Alcotest.fail "Created user not found"
| Error err ->
    Alcotest.failf "User retrieval failed: %s"
      (Queries.string_of_query_error err));

Testing Features:

3. 💰 Transaction Integrity Testing

Challenge: Verify that database transactions commit and rollback correctly.

Solution: Explicit transaction testing with both success and failure scenarios.

Transaction Commit Verification 

let test_transactions _ _ () =
  let* () = ensure_db_ready () in
  (* Commit path *)
  let uid_gen = generate_uuid () in
  let name = "Txn User " ^ Uuidm.to_string (uid_gen ()) in
  let email = "txn." ^ Uuidm.to_string (uid_gen ()) ^ "@example.com" in
  let* txn_commit =
    Connection.with_connection (fun conn ->
        let module Conn = (val conn : Caqti_lwt.CONNECTION) in
        Conn.with_transaction (fun () ->
            let* created = UserService.create_user ~name ~email ~phone:None in
            match created with
            | Error err ->
                Alcotest.failf "txn create failed: %s"
                  (Queries.string_of_query_error err)
            | Ok uid -> (
                let* got = UserService.get_user_by_id uid in
                match got with
                | Ok (Some _) -> Lwt.return (Ok ())
                | Ok None -> Alcotest.fail "txn get returned None"
                | Error err ->
                    Alcotest.failf "txn get failed: %s"
                      (Queries.string_of_query_error err))))

Rollback Verification

(* Rollback path *)
let rb_email = "rollback." ^ Uuidm.to_string (uid_gen ()) ^ "@example.com" in
let* txn_rb =
  Lwt.catch
    (fun () ->
      Connection.with_connection (fun conn ->
          let module Conn = (val conn : Caqti_lwt.CONNECTION) in
          Conn.with_transaction (fun () ->
              let* _ =
                UserService.create_user ~name:"Rollback User" ~email:rb_email
                  ~phone:None
              in
              Lwt.fail_with "force rollback")))
    (fun exn ->
      Lwt.return (Error (`Connection_error (Printexc.to_string exn))))
in
(match txn_rb with
| Error _ -> ()
| Ok _ -> Alcotest.fail "Rollback transaction unexpectedly succeeded");

Transaction Testing Features:

4. 🔄 Migration Safety Testing

Challenge: Ensure database migrations can be safely rolled back and reapplied.

Solution: Comprehensive migration round-trip testing with single and multi-step scenarios.

Single Migration Round-Trip

let test_migrations_roundtrip _ _ () =
  let* () = ensure_db_ready () in
  let migrations = GeneratedList.all_migrations in
  let* applied = Migrations.get_applied_migrations () in
  let applied =
    match applied with
    | Ok xs -> xs
    | Error err ->
        Alcotest.failf "get_applied_migrations failed: %s"
          (Migrations.migration_error_to_string err)
  in
  match List.rev applied with
  | [] -> Alcotest.fail "No applied migrations to round-trip"
  | latest :: _ -> (
      let latest_m =
        let open Database.Migrations in
        List.find (fun m -> m.id = latest.migration_id) migrations
      in
      (* down, fail the test on error *)
      let* down_res = Migrations.migrate_down latest_m in
      let* () =
        match down_res with
        | Ok () -> Lwt.return_unit
        | Error err ->
            Alcotest.failf "migrate_down failed: %s"
              (Migrations.migration_error_to_string err)
      in
      (* up again, fail the test on error *)
      let* up_res = Migrations.migrate_up latest_m in
      match up_res with
      | Ok () -> Lwt.return_unit
      | Error err ->
          Alcotest.failf "migrate_up (restore) failed: %s"
            (Migrations.migration_error_to_string err))

Multi-Step Migration Chain Testing 

let test_migrations_roundtrip_chain _ _ () =
  let* () = ensure_db_ready () in
  let all = GeneratedList.all_migrations in
  let find_m id =
    try List.find (fun m -> m.Migrations.id = id) all
    with Not_found ->
      Alcotest.failf "Migration id not found in generated list: %s" id
  in
  let* applied = Migrations.get_applied_migrations () in
  let applied =
    match applied with
    | Ok xs -> xs
    | Error err ->
        Alcotest.failf "get_applied_migrations failed: %s"
          (Migrations.migration_error_to_string err)
  in
  let count = List.length applied in
  if count < 2 then
    Alcotest.fail "Not enough applied migrations to test multi-step round-trip";
  let steps = min 3 count in
  let rec take n = function
    | _ when n <= 0 -> []
    | [] -> []
    | x :: xs -> x :: take (n - 1) xs
  in
  let to_cycle =
    applied |> List.rev |> take steps
    |> List.map (fun a -> find_m a.Migrations.migration_id)
  in
  (* Down each selected migration from latest backwards *)
  let* () =
    Lwt_list.iter_s
      (fun m ->
        let* r = Migrations.migrate_down m in
        match r with
        | Ok () -> Lwt.return_unit
        | Error err ->
            Alcotest.failf "migrate_down %s failed: %s" m.id
              (Migrations.migration_error_to_string err))
      to_cycle
  in
  (* Up each selected migration in original order *)
  let* () =
    Lwt_list.iter_s
      (fun m ->
        let* r = Migrations.migrate_up m in
        match r with
        | Ok () -> Lwt.return_unit
        | Error err ->
            Alcotest.failf "migrate_up %s failed: %s" m.id
              (Migrations.migration_error_to_string err))
      (List.rev to_cycle)
  in
  (* Ensure the whole chain is Up at the end *)
  let* up_all =
    Migrations.run_migrations GeneratedList.all_migrations Migrations.Up
  in
  (match up_all with
  | Ok () -> ()
  | Error err ->
      Alcotest.failf "Final migrations Up failed: %s"
        (Migrations.migration_error_to_string err));

Migration Testing Features:

5. 🚀 Connection Pool Stress Testing

Challenge: Verify connection pool behavior under concurrent load.

Solution: Configurable stress testing with concurrent workers and parameterized load.

Environment-Configurable Stress Testing 

(* New: helpers to parameterize stress/bench via env *)
let getenv_int name default =
  match Sys.getenv_opt name with
  | Some v -> ( try int_of_string v with _ -> default)
  | None -> default

let getenv_float name default =
  match Sys.getenv_opt name with
  | Some v -> ( try float_of_string v with _ -> default)
  | None -> default

Concurrent Worker Implementation 

let test_connection_pool_stress _ _ () =
  let* () = ensure_db_ready () in
  let concurrency = getenv_int "DB_STRESS_CONCURRENCY" 32 in
  let iters_per_worker = getenv_int "DB_STRESS_ITERS" 20 in
  let worker _ =
    let rec loop i =
      if i = 0 then Lwt.return_unit
      else
        let* r = Connection.ping_database () in
        (match r with
        | Ok _ -> ()
        | Error err ->
            Alcotest.failf "Ping failed: %s" (Connection.error_to_string err));
        loop (i - 1)
    in
    loop iters_per_worker
  in
  let tasks = List.init concurrency worker in
  let* () = Lwt.join tasks in
  Lwt.return_unit

Stress Testing Features:

6. 📊 Query Performance Benchmarking

Challenge: Establish performance baselines and catch regressions.

Solution: Comprehensive benchmarking infrastructure with timing and thresholds.

Timing Infrastructure

(* Helper: A Time function for benchmarking *)
let time_function f =
  let start = Unix.gettimeofday () in
  let* result = f () in
  let duration = Unix.gettimeofday () -. start in
  Lwt.return (result, duration)

Ping Benchmark Implementation

let ping_benchmark () =
  let open Lwt.Syntax in
  let iterations = getenv_int "DB_PING_ITERATIONS" 50 in
  let threshold_ms = getenv_float "DB_PING_THRESHOLD_MS" 300.0 in

  let rec ping_loop acc n =
    if n <= 0 then Lwt.return acc
    else
      let* _, duration =
        time_function (fun () -> Connection.ping_database ())
      in
      ping_loop (duration :: acc) (n - 1)
  in

  let* durations = ping_loop [] iterations in
  let total_time = List.fold_left ( +. ) 0.0 durations in
  let avg_time_ms = total_time /. float_of_int iterations *. 1000.0 in

  Printf.printf "[bench] ping x%d total=%.3fs avg=%.2fms (threshold=%.1fms)\n"
    iterations total_time avg_time_ms threshold_ms;

  if avg_time_ms <= threshold_ms then
    Printf.printf "ASSERT avg ping <= threshold\n"
  else
    Alcotest.failf "Ping benchmark failed: avg %.2fms > threshold %.1fms"
      avg_time_ms threshold_ms;

  Lwt.return_unit

End-to-End Query Benchmarking

let test_query_performance_benchmarks _ _ () =
  let open Lwt.Syntax in
  (* First, create a test user for benchmarking *)
  let* user_result =
    Services.User_service.create_user ~name:"Benchmark User"
      ~email:"benchmark@example.com" ~phone:None
  in

  match user_result with
  | Error err ->
      Alcotest.failf "Failed to create benchmark user: %s"
        (Database.Queries.string_of_query_error err)
  | Ok user_id ->
      let* () = ping_benchmark () in

      let lookup_loop () =
        let* result =
          Services.User_service.get_user_by_email "benchmark@example.com"
        in
        match result with
        | Error err ->
            Alcotest.failf "get_user_by_email failed: %s"
              (Database.Queries.string_of_query_error err)
        | Ok (Some _user) -> Lwt.return_unit
        | Ok None ->
            Alcotest.fail "get_user_by_email returned None for existing user"
      in

      let rec run_benchmarks n =
        if n <= 0 then Lwt.return_unit
        else
          let* () = lookup_loop () in
          run_benchmarks (n - 1)
      in

      (* Run the benchmarks *)
      let* () = run_benchmarks 50 in

      (* Clean up: delete the test user after successful benchmarking *)
      let* cleanup_result = Services.User_service.delete_user user_id in
      (match cleanup_result with
      | Ok _ -> Printf.printf "[cleanup] Deleted benchmark user successfully\n"
      | Error err ->
          Printf.printf
            "[cleanup] Warning: Failed to delete benchmark user: %s\n"
            (Database.Queries.string_of_query_error err));

      Lwt.return_unit

Performance Testing Features:

Technical Deep Dive

Test Dependencies Evolution

Before: Minimal Dependencies

(test
 (name test_database)
 (libraries
  chaufr.server.models
  chaufr.server.database
  chaufr.server.repository
  chaufr.server.services
  lwt
  uuidm
  caqti
  caqti-lwt
  yojson
  ppx_deriving_yojson
  ppx_deriving.show
  alcotest
  alcotest-lwt)

After: Development-Ready Dependencies 

(test
 (name test_database)
 (libraries
  chaufr.server.models
  chaufr.server.database
  chaufr.server.repository
  chaufr.server.services
  chaufr.generated.migrations      # Migration testing support
  lwt
  uuidm
  caqti
  caqti-lwt
  caqti-driver-postgresql          # Direct PostgreSQL driver
  yojson
  simple_dotenv                    # Environment configuration
  ppx_deriving_yojson
  ppx_deriving.show
  alcotest
  alcotest-lwt)

Service Layer Testing Integration

The testing infrastructure integrates seamlessly with the existing service layer:

User Service Integration

module UserService = Services.User_service
module GeneratedList = Util_database_generated.Generated_list

(* Direct service layer testing *)
let* created = UserService.create_user ~name ~email ~phone in
let uid =
  match created with
  | Ok uid -> uid
  | Error err ->
      Alcotest.failf "User creation failed: %s"
        (Queries.string_of_query_error err)

Error Handling Integration

(match fetched with
| Ok (Some user) ->
    Alcotest.(check string) "name matches" name user.first_name;
    (* ... more assertions *)
| Ok None -> Alcotest.fail "Created user not found"
| Error err ->
    Alcotest.failf "User retrieval failed: %s"
      (Queries.string_of_query_error err));

Performance and Scalability Results

Benchmark Configuration

Parameter Environment Variable Default Purpose
Stress Concurrency DB_STRESS_CONCURRENCY 32 Concurrent workers for pool testing
Stress Iterations DB_STRESS_ITERS 20 Operations per worker
Ping Iterations DB_PING_ITERATIONS 50 Ping benchmark sample size
Ping Threshold DB_PING_THRESHOLD_MS 300.0 Maximum acceptable ping time

Development Deployment Considerations

CI/CD Integration

The test suite is designed for CI/CD environments:

Environment Variable Configuration 

# Development-like stress testing
export DB_STRESS_CONCURRENCY=64
export DB_STRESS_ITERS=100
export DB_PING_ITERATIONS=200
export DB_PING_THRESHOLD_MS=100.0

# Run tests
dune exec -- test/test_database.exe

Database State Management

Monitoring and Alerting Integration

The benchmarking infrastructure provides foundation for:

  1. Performance Regression Detection

    • Threshold-based assertions catch slowdowns
    • Statistical analysis identifies trends
    • Configurable thresholds per environment

  2. Connection Pool Health

    • Stress testing validates pool behavior
    • Concurrent operation verification
    • Resource exhaustion detection

  3. Migration Safety Verification

    • Round-trip testing ensures reversibility
    • Multi-step chain validation
    • Automatic state restoration

Lessons Learned and Best Practices

1. Environment Configuration Strategy

✅ What Worked:

⚠️ Challenges Faced:

2. Real Database vs Mocking

✅ What Worked:

⚠️ Lessons Learned:

3. Test Performance Architecture

✅ Best Practices Established:

⚠️ Performance Considerations:

Testing Infrastructure Readiness Checklist

✅ Completed This Sprint

Implementation Impact

Code Quality Improvements

  1. Confidence in Database Operations

    • Comprehensive test coverage eliminates deployment anxiety
    • Migration testing prevents production rollback issues
    • Connection pool validation ensures scalability

  2. Performance Regression Prevention

    • Benchmark baselines catch performance degradation
    • Threshold-based assertions provide early warning
    • Statistical analysis identifies trends

  3. Operational Excellence

    • Real database testing mirrors production behavior
    • Environment configuration supports multiple deployment stages
    • Comprehensive error handling improves debugging

Development Velocity

The testing infrastructure provides:

Conclusion

Building a production-grade database testing infrastructure in OCaml demonstrates the ecosystem's maturity for enterprise applications. The combination of Alcotest, Lwt, and Caqti provides a robust foundation for comprehensive database testing.

Key achievements:

This testing infrastructure serves as both a safety net for current development and a foundation for future scaling. The investment in comprehensive testing pays dividends through reduced production issues, faster development cycles, and increased confidence in database operations.

Next Steps

  1. CI/CD Integration - Automated test execution in deployment pipeline
  2. Observability Enhancement - OpenTelemetry integration for test metrics
  3. Load Testing Expansion - Extended stress testing scenarios
  4. Documentation - Comprehensive testing guides and runbooks

The OCaml ecosystem provides excellent tools for this critical infrastructure layer.

Hey, this site is part of ring.muhokama.fun!