Demonstrates "Commands out of sync" error with ocaml-mariadb and let

jbuild

(jbuild_version 1)

(executable
 ((name main)
  (libraries (core cohttp.lwt mariadb))))

main.ml

open Lwt.Infix
open Printf

let or_die where = function
  | Ok r -> Lwt.return r
  | Error (i, e) -> Lwt.fail_with @@ sprintf "%s: (%d) %s" where i e

let stream res =
  let next _ =
    Mdb.Res.fetch (module Mdb.Row.Map) res
    >>= function
    | Ok (Some _ as row) -> Lwt.return row
    | Ok None -> Lwt.return_none
    | Error _ -> Lwt.return_none in
  Lwt.return (Lwt_stream.from next)

let stream_next_opt s =
  Lwt.catch
    (fun () -> Lwt_stream.next s >>= Lwt.return_some)
    (function
      | Lwt_stream.Empty -> Lwt.return_none
      | exn -> Lwt.fail exn)

let execute_query conn query values yield =
  Lwt_io.printlf "Executing query %s" query >>= fun () ->
  Mdb.prepare conn query >>= or_die "prepare" >>= fun stmt ->
  Lwt_io.printlf "prepared query" >>= fun () ->
  Mdb.Stmt.execute stmt values >>= or_die "exec" >>= fun result ->
  Lwt_io.printlf "executed query, yielding" >>= fun () ->
  yield result >>= fun return -> 
  Lwt_io.printlf "yielded, closing statement" >>= fun () ->
  Mdb.Stmt.close stmt >>= or_die "stmt close" >>= fun () ->
  Lwt_io.printlf "returning value from yield" >>= fun () ->
  Lwt.return return


let test db_conn index =
  let query =
    "SELECT id FROM url WHERE id = ? ORDER BY id ASC LIMIT 1" in
  
  execute_query db_conn query [| `Int 1 |] (function
  | None -> Lwt_io.printlf "[%d] nothing found" index
  | Some result -> stream result >>= stream_next_opt >>= function
    | None -> Lwt_io.printlf "[%d] no rows returned" index
    | Some row -> Lwt_io.printlf "[%d] row found" index
  )

let main () =
  let db_connect =
    Mdb.connect ~host:"localhost" ~user:"root" ~pass:"" ~db:"example" in
  db_connect () >>= or_die "connect" >>= fun db_conn ->
  
  let mkt index = test db_conn index in
  Lwt.join [ mkt 0; mkt 1; mkt 2; ] >>= fun () ->

  Mdb.close db_conn

let () =
  Lwt_main.run @@ main ()

mdb.ml

open Lwt.Infix

module S = Mariadb.Nonblocking.Status
include Mariadb.Nonblocking.Make(struct
  module IO = struct
    type 'a future = 'a Lwt.t
    let (>>=) = (>>=)
    let return = Lwt.return
  end

  let wait mariadb status =
    let fd = Lwt_unix.of_unix_file_descr @@ Mariadb.Nonblocking.fd mariadb in
    assert (S.read status || S.write status || S.timeout status);
    let idle, _ = Lwt.task () in
    let rt =
      if S.read status then Lwt_unix.wait_read fd
      else idle in
    let wt =
      if S.write status then Lwt_unix.wait_write fd
      else idle in
    let tt =
      match S.timeout status, Mariadb.Nonblocking.timeout mariadb with
      | true, 0 -> Lwt.return ()
      | true, tmout -> Lwt_unix.timeout (float tmout)
      | false, _ -> idle in
    Lwt.catch
      (fun () ->
        Lwt.nchoose [rt; wt; tt] >>= fun _ ->
        Lwt.return @@
          S.create
            ~read:(Lwt_unix.readable fd)
            ~write:(Lwt_unix.writable fd)
            ())
      (function
      | Lwt_unix.Timeout -> Lwt.return @@ S.create ~timeout:true ()
      | e -> Lwt.fail e)
end)