Closing merge file_descr when aborting

CHANGES.md

@@ -8,6 +8,8 @@
 - Attempt to recover from `log_async` invariant violations during an explicit
   sync operation, rather than failing immediately. (#329)
 
+- Proper cleaning of merge file descriptors when aborting a merge (#326)
+
 ## Changed
 
 - Release overly defensive warnings occuring when pre-fetching the disk. (#322)

src/index.ml

@@ -768,7 +768,9 @@ struct
           | `Aborted -> `Aborted)
     in
     match merge_result with
-    | `Aborted -> (`Aborted, Mtime.Span.zero)
+    | `Aborted ->
+        IO.clear ~generation ~reopen:false merge;
+        (`Aborted, Mtime.Span.zero)
     | `Index_io io ->
         let fan_out = Fan.finalize fan_out in
         let index = { io; fan_out } in

test/unix/common.ml

@@ -188,3 +188,44 @@ let check_disjoint index htbl =
           Alcotest.failf "Found value %a when checking for the absence of %a"
             (Repr.pp Value.t) v' pp_binding (k, v))
     htbl
+
+let get_open_fd root =
+  let ( >>? ) x f = match x with `Ok x -> f x | `Skip err -> `Skip err in
+  let pid = string_of_int (Unix.getpid ()) in
+  let name = Filename.concat root "empty" in
+  let fd_file = "index_fd_tmp" in
+  let lsof_command = "lsof -a -s -p " ^ pid ^ " > " ^ fd_file in
+  (match Sys.os_type with
+  | "Unix" -> `Ok ()
+  | _ -> `Skip "non-UNIX operating system")
+  >>? fun () ->
+  (match Unix.system lsof_command with
+  | Unix.WEXITED 0 -> `Ok ()
+  | Unix.WEXITED _ ->
+      `Skip "failing `lsof` command. Is `lsof` installed on your system?"
+  | Unix.WSIGNALED _ | Unix.WSTOPPED _ -> `Skip "`lsof` command was interrupted")
+  >>? fun () ->
+  let lines = ref [] in
+  let extract_fd line =
+    try
+      let pos = Re.Str.search_forward (Re.Str.regexp name) line 0 in
+      let fd = Re.Str.string_after line pos in
+      lines := fd :: !lines
+    with Not_found -> ()
+  in
+  let ic = open_in fd_file in
+  (try
+     while true do
+       extract_fd (input_line ic)
+     done
+   with End_of_file -> close_in ic);
+  `Ok !lines
+
+let partition sub l =
+  List.partition
+    (fun line ->
+      try
+        ignore (Re.Str.search_forward (Re.Str.regexp sub) line 0);
+        true
+      with Not_found -> false)
+    l

test/unix/common.mli

@@ -106,3 +106,5 @@ val check_completed :
 
 val check_equivalence : Index.t -> (Key.t, Value.t) Hashtbl.t -> unit
 val check_disjoint : Index.t -> (Key.t, Value.t) Hashtbl.t -> unit
+val get_open_fd : string -> [> `Ok of string list | `Skip of string ]
+val partition : string -> string list -> string list * string list

test/unix/force_merge.ml

@@ -33,63 +33,31 @@ let test_one_entry r k v =
       Alcotest.failf "Inserted value is not present anymore: %s." k
 
 let test_fd () =
-  let ( >>? ) x f = match x with `Ok x -> f x | err -> err in
-  (* TODO: fix these tests to take the correct directory name
-           (and not break when given the wrong one) *)
-  let name = "/tmp/empty" in
-  (* construct an index at a known location *)
-  let pid = string_of_int (Unix.getpid ()) in
-  let fd_file = "tmp" in
-  let lsof_command = "lsof -a -s -p " ^ pid ^ " > " ^ fd_file in
-  let result =
-    (match Sys.os_type with
-    | "Unix" -> `Ok ()
-    | _ -> `Skip "non-UNIX operating system")
-    >>? fun () ->
-    (match Unix.system lsof_command with
-    | Unix.WEXITED 0 -> `Ok ()
-    | Unix.WEXITED _ ->
-        `Skip "failing `lsof` command. Is `lsof` installed on your system?"
-    | Unix.WSIGNALED _ | Unix.WSTOPPED _ ->
-        Alcotest.fail "`lsof` command was interrupted")
-    >>? fun () ->
-    let lines = ref [] in
-    let extract_fd line =
-      try
-        let pos = Re.Str.search_forward (Re.Str.regexp name) line 0 in
-        let fd = Re.Str.string_after line pos in
-        lines := fd :: !lines
-      with Not_found -> ()
-    in
-    let ic = open_in fd_file in
-    let lines =
-      (try
-         while true do
-           extract_fd (input_line ic)
-         done
-       with End_of_file -> close_in ic);
-      !lines
-    in
-    let contains sub s =
-      try
-        ignore (Re.Str.search_forward (Re.Str.regexp sub) s 0);
-        true
-      with Not_found -> false
-    in
-    let data, rs = List.partition (contains "data") lines in
-    if List.length data > 2 then
-      Alcotest.fail "Too many file descriptors opened for data files";
-    let log, rs = List.partition (contains "log") rs in
-    if List.length log > 2 then
-      Alcotest.fail "Too many file descriptors opened for log files";
-    let lock, rs = List.partition (contains "lock") rs in
-    if List.length lock > 2 then
-      Alcotest.fail "Too many file descriptors opened for lock files";
-    if List.length rs > 0 then Alcotest.fail "Unknown file descriptors opened";
-    `Ok ()
-  in
-  match result with
-  | `Ok () -> ()
+  match Common.get_open_fd root with
+  | `Ok lines -> (
+      let contains sub s =
+        try
+          ignore (Re.Str.search_forward (Re.Str.regexp sub) s 0);
+          true
+        with Not_found -> false
+      in
+      let result =
+        let data, rs = List.partition (contains "data") lines in
+        if List.length data > 2 then
+          Alcotest.fail "Too many file descriptors opened for data files";
+        let log, rs = List.partition (contains "log") rs in
+        if List.length log > 2 then
+          Alcotest.fail "Too many file descriptors opened for log files";
+        let lock, rs = List.partition (contains "lock") rs in
+        if List.length lock > 2 then
+          Alcotest.fail "Too many file descriptors opened for lock files";
+        if List.length rs > 0 then
+          Alcotest.fail "Unknown file descriptors opened";
+        `Ok ()
+      in
+      match result with
+      | `Ok () -> ()
+      | `Skip err -> Log.warn (fun m -> m "`test_fd` was skipped: %s" err))
   | `Skip err -> Log.warn (fun m -> m "`test_fd` was skipped: %s" err)
 
 let readonly_s () =

test/unix/main.ml

@@ -756,7 +756,13 @@ module Close = struct
         Alcotest.failf
           "Asynchronous exception occurred during try_merge ~force:true: %s"
           (Printexc.to_string exn)
-    | Ok `Aborted -> ()
+    | Ok `Aborted -> (
+        match Common.get_open_fd root with
+        | `Ok ofd ->
+            let merge, _ = Common.partition "merge" ofd in
+            if List.length merge > 0 then
+              Alcotest.fail "Too many file descriptors opened for merge files"
+        | `Skip err -> Log.warn (fun m -> m "`aborted_fd` was skipped: %s" err))
 
   let tests =
     [