fix(watch): complete RPC builds independently

Introduce build requests with hidden completion ivars in the build system.
The watch build loop can still batch RPC build requests together, but each
request now reports success as soon as its own action builder and per-request
finalization complete instead of waiting for the entire batch.

Complete any unfinished RPC requests with the batch outcome when a build
iteration ends, so requests cancelled by --stop-on-first-error do not hang.

Update the concurrent RPC build cram test to expect the fast request to finish
before the slow target is released.

Signed-off-by: Rudi Grinberg <me@rgrinberg.com>

refactor(watch): narrow RPC request completion hook

Signed-off-by: Rudi Grinberg <me@rgrinberg.com>

refactor(watch): use input generation for RPC completion

Signed-off-by: Rudi Grinberg <me@rgrinberg.com>

fix(watch): defer diff-promotion output and finish to batch end

Per-request finalization now only saves the diff-promotion database
without promoting files, so promotion output stays ordered after
diff and error messages.

When per-request completion triggers a restart signal but all RPC
requests have already finished, treat the iteration as Done so that
the watch status line is still emitted.

Signed-off-by: Rudi Grinberg <me@rgrinberg.com>

refactor(watch): split RPC request checkpointing from completion

Checkpoint request side effects before completing RPC requests, but keep build-wide diff-promotion finalization at the end of the build iteration.

Defer request completion under --auto-promote after a checkpoint saves diff promotions, so watch mode can still restart after the final auto-promotion.

This restores ordinary auto-promote output ordering while keeping saved promotions visible to RPC promotion requests.

Signed-off-by: Rudi Grinberg <me@rgrinberg.com>

Author: rgrinberg

bin/build.ml

@@ -6,7 +6,10 @@ let action_builder_of_request request =
 ;;
 
 let run_build_system ~run_id ~request =
-  Dune_engine.Build_system.run_action_builder ~run_id (action_builder_of_request request)
+  let request =
+    action_builder_of_request request |> Dune_engine.Build_system.Build_request.create
+  in
+  Dune_engine.Build_system.run_build_requests ~run_id [ request ]
 ;;
 
 let rpc_request_action ~(common : Common.t) (kind : Dune_rpc_impl.Server.build_request) =

src/dune_engine/build_loop.ml

@@ -43,11 +43,6 @@ module Rpc_request_id = struct
   module Map = C.Map
 end
 
-type rpc_request =
-  { build : unit Action_builder.t
-  ; outcome : Build_outcome.t Fiber.Ivar.t
-  }
-
 type t =
   { mutable status : status
   ; mutable pending_reset : Memo.Invalidation.t option
@@ -58,7 +53,7 @@ type t =
     (* Incremented whenever an invalidation is recorded. This lets waiters
        distinguish input changes from other wakeups. *)
   ; mutable next_run_id : int
-  ; mutable rpc_requests : rpc_request Rpc_request_id.Map.t
+  ; mutable rpc_requests : Build_system.Build_request.t Rpc_request_id.Map.t
   ; mutable state : [ `Awaiting_init | `Init ]
   }
 
@@ -221,14 +216,20 @@ let reset_wakeup t =
   Trigger.trigger previous_wakeup
 ;;
 
+let all_rpc_requests_finished t =
+  Rpc_request_id.Map.to_list t.rpc_requests
+  |> List.for_all ~f:(fun (_, request) -> Build_system.Build_request.is_finished request)
+;;
+
 let run_current_build
       ({ watch_restart_started_at = restart_started_at
        ; input_change_generation = build_start_input_change_generation
        ; wakeup_generation = build_start_wakeup_generation
        ; _
        } as t)
       ~run_id
-      build
+      ~finish_when_rpc_requests_finished
+      run_build
   =
   let* () = reset_wakeup t in
   (match t.status with
@@ -237,10 +238,15 @@ let run_current_build
   t.status <- Building run_id;
   let+ outcome =
     Scheduler.with_current_build_cancellation (Fiber.Cancel.create ()) (fun () ->
-      Build_system.run_action_builder ?restart_started_at ~run_id build)
+      run_build ?restart_started_at ~run_id ())
   in
   let next =
     match t.status with
+    | Restarting_build _
+      when finish_when_rpc_requests_finished && all_rpc_requests_finished t ->
+      t.status <- Standing_by;
+      t.watch_restart_started_at <- None;
+      `Done
     | Restarting_build _ -> `Restart
     | Standing_by | Building _ ->
       t.status <- Standing_by;
@@ -260,6 +266,13 @@ let rec wait_for_wakeup_after t wakeup_generation =
     wait_for_wakeup_after t wakeup_generation)
 ;;
 
+let remove_finished_rpc_requests t =
+  Rpc_request_id.Map.to_list t.rpc_requests
+  |> List.iter ~f:(fun (id, request) ->
+    if Build_system.Build_request.is_finished request
+    then t.rpc_requests <- Rpc_request_id.Map.remove t.rpc_requests id)
+;;
+
 let cancel_rpc_requests t ~f =
   let* () = Fiber.return () in
   match
@@ -270,11 +283,17 @@ let cancel_rpc_requests t ~f =
   | to_cancel ->
     List.iter to_cancel ~f:(fun (id, _) ->
       t.rpc_requests <- Rpc_request_id.Map.remove t.rpc_requests id);
+    let to_cancel_unfinished =
+      List.filter to_cancel ~f:(fun (_, request) ->
+        not (Build_system.Build_request.is_finished request))
+    in
     let* () =
-      Fiber.parallel_iter to_cancel ~f:(fun (_, { outcome; _ }) ->
-        Fiber.Ivar.fill outcome Failure)
+      Fiber.parallel_iter to_cancel_unfinished ~f:(fun (_, request) ->
+        Build_system.Build_request.complete request Failure)
     in
-    request_rebuild_due_to_rpc_request t
+    if List.is_empty to_cancel_unfinished
+    then Fiber.return ()
+    else request_rebuild_due_to_rpc_request t
 ;;
 
 let cancel_rpc_requests_by_session t ~session_id =
@@ -285,18 +304,19 @@ let cancel_rpc_requests_by_session t ~session_id =
 let cancel_all_rpc_requests t = cancel_rpc_requests t ~f:(fun _ _ -> true)
 
 let submit_rpc_request t ~session_id ~request_id ~build =
-  let outcome = Fiber.Ivar.create () in
+  let request = Build_system.Build_request.create build in
   let id = Rpc_request_id.create ~session_id ~request_id in
   let* () = Fiber.return () in
   (match Rpc_request_id.Map.find t.rpc_requests id with
    | Some _ ->
      Code_error.raise
        "RPC build request with this id is already active"
        [ "id", Rpc_request_id.to_dyn id ]
-   | None ->
-     t.rpc_requests <- Rpc_request_id.Map.add_exn t.rpc_requests id { build; outcome });
+   | None -> t.rpc_requests <- Rpc_request_id.Map.add_exn t.rpc_requests id request);
   let* () = request_rebuild_due_to_rpc_request t in
-  Fiber.Ivar.read outcome
+  let+ outcome = Build_system.Build_request.await request in
+  t.rpc_requests <- Rpc_request_id.Map.remove t.rpc_requests id;
+  outcome
 ;;
 
 type rpc_poll_iter_result =
@@ -306,6 +326,7 @@ type rpc_poll_iter_result =
 
 let rpc_poll_iter t ~sticky_goal ~sticky_built_at =
   let rec loop () =
+    remove_finished_rpc_requests t;
     let rpc_requests = Rpc_request_id.Map.to_list t.rpc_requests in
     let sticky_goal_to_build =
       match sticky_goal with
@@ -329,11 +350,12 @@ let rpc_poll_iter t ~sticky_goal ~sticky_built_at =
       Fiber.return { wakeup_generation = t.wakeup_generation; sticky_built_at = None }
     | _ ->
       let* res, next, input_change_generation, wakeup_generation =
-        let build =
-          let builds = List.map rpc_requests ~f:(fun (_, { build; _ }) -> build) in
+        let requests =
+          let rpc_requests = List.map rpc_requests ~f:snd in
           match sticky_goal_to_build with
-          | None -> Action_builder.all_unit builds
-          | Some sticky_goal -> Action_builder.all_unit (sticky_goal :: builds)
+          | None -> rpc_requests
+          | Some sticky_goal ->
+            Build_system.Build_request.create sticky_goal :: rpc_requests
         in
         let run_id = next_watch_run_id t in
         let () =
@@ -347,7 +369,19 @@ let rpc_poll_iter t ~sticky_goal ~sticky_built_at =
               Console.maybe_clear_screen ~details_hum);
             Memo.reset invalidation
         in
-        run_current_build t ~run_id build
+        let build_input_change_generation = t.input_change_generation in
+        let finish_when_rpc_requests_finished = Option.is_none sticky_goal_to_build in
+        run_current_build
+          t
+          ~run_id
+          ~finish_when_rpc_requests_finished
+          (fun ?restart_started_at ~run_id () ->
+             Build_system.run_build_requests
+               ?restart_started_at
+               ~run_id
+               ~should_complete_request:(fun () ->
+                 Int.equal t.input_change_generation build_input_change_generation)
+               requests)
       in
       (match next with
        | `Restart -> loop ()
@@ -358,15 +392,13 @@ let rpc_poll_iter t ~sticky_goal ~sticky_built_at =
              | Ok () -> Build_outcome.Success
              | Error `Already_reported -> Failure
            in
-           let+ () =
-             Fiber.sequential_iter rpc_requests ~f:(fun (id, _) ->
-               match Rpc_request_id.Map.find t.rpc_requests id with
-               | None -> Fiber.return ()
-               | Some { outcome = ivar; _ } ->
-                 t.rpc_requests <- Rpc_request_id.Map.remove t.rpc_requests id;
-                 Fiber.Ivar.fill ivar outcome)
+           let* () =
+             Fiber.sequential_iter rpc_requests ~f:(fun (id, request) ->
+               t.rpc_requests <- Rpc_request_id.Map.remove t.rpc_requests id;
+               Build_system.Build_request.complete request outcome)
            in
-           build_finish outcome
+           build_finish outcome;
+           Fiber.return ()
          in
          { wakeup_generation
          ; sticky_built_at =

src/dune_engine/build_system.ml

@@ -1121,10 +1121,19 @@ let handle_final_exns exns =
        List.iter exns ~f:Dune_util.Report_error.report)
 ;;
 
-let run ?restart_started_at ?(run_id = Run_id.Batch) f =
-  let finalize_diff_promotion () =
-    protect ~f:Diff_promotion.finalize ~finally:Diff_promotion.clear_cache
-  in
+let finalize_diff_promotion () =
+  protect ~f:Diff_promotion.finalize ~finally:Diff_promotion.clear_cache
+;;
+
+let checkpoint_build_request () =
+  let open Fiber.O in
+  Target_promotion.save ();
+  let diff_promotions_saved = Diff_promotion.save () in
+  let+ () = Scheduler.flush_file_watcher () in
+  diff_promotions_saved
+;;
+
+let run_with_error_collection ?restart_started_at ?(run_id = Run_id.Batch) collect_errors =
   let open Fiber.O in
   let f () =
     let start = Time.now () in
@@ -1144,10 +1153,7 @@ let run ?restart_started_at ?(run_id = Run_id.Batch) f =
     Fs_memo.invalidate_cached_timestamps ();
     let* () = State.reset_progress () in
     let* () = State.reset_errors () in
-    let* outcome =
-      Fiber.collect_errors (fun () ->
-        Memo.run_with_error_handler f ~handle_error_no_raise:report_early_exn)
-    in
+    let* outcome = collect_errors () in
     Dtemp.clear ();
     Sandbox.cleanup_pending_targets ();
     Target_promotion.save ();
@@ -1193,6 +1199,12 @@ let run ?restart_started_at ?(run_id = Run_id.Batch) f =
     Scheduler.with_current_build_cancellation (Fiber.Cancel.create ()) f)
 ;;
 
+let run ?restart_started_at ?run_id f =
+  run_with_error_collection ?restart_started_at ?run_id (fun () ->
+    Fiber.collect_errors (fun () ->
+      Memo.run_with_error_handler f ~handle_error_no_raise:report_early_exn))
+;;
+
 let run_exn f =
   let open Fiber.O in
   run f
@@ -1201,12 +1213,84 @@ let run_exn f =
   | Error `Already_reported -> raise Dune_util.Report_error.Already_reported
 ;;
 
-let run_action_builder ?restart_started_at ?run_id request =
-  run ?restart_started_at ?run_id (fun () ->
-    let+ (), (_ : Dep.Fact.t Dep.Map.t) =
-      Action_builder.evaluate_and_collect_facts request
-    in
-    ())
+let evaluate_action_builder request =
+  let+ (), (_ : Dep.Fact.t Dep.Map.t) =
+    Action_builder.evaluate_and_collect_facts request
+  in
+  ()
+;;
+
+module Build_request = struct
+  type t =
+    { build : unit Action_builder.t
+    ; outcome : Build_outcome.t Fiber.Ivar.t
+    }
+
+  let create build = { build; outcome = Fiber.Ivar.create () }
+  let await t = Fiber.Ivar.read t.outcome
+  let is_finished t = Option.is_some (Fiber.Ivar.peek t.outcome)
+
+  let complete t outcome =
+    Fiber.of_thunk (fun () ->
+      match Fiber.Ivar.peek t.outcome with
+      | Some _ -> Fiber.return ()
+      | None -> Fiber.Ivar.fill t.outcome outcome)
+  ;;
+
+  let build t = t.build
+end
+
+let run_build_requests
+      ?restart_started_at
+      ?run_id
+      ?(should_complete_request = fun () -> true)
+      requests
+  =
+  let open Fiber.O in
+  let run_action_builder_fiber request =
+    Fiber.collect_errors (fun () ->
+      Memo.run_with_error_handler
+        (fun () -> evaluate_action_builder request)
+        ~handle_error_no_raise:report_early_exn)
+  in
+  let diff_promotions_saved = ref false in
+  let checkpoint_request () =
+    let+ saved = checkpoint_build_request () in
+    if saved then diff_promotions_saved := true
+  in
+  let should_defer_completion () =
+    match !Clflags.promote with
+    | Some Automatically -> !diff_promotions_saved
+    | Some Never | None -> false
+  in
+  let finish_request request outcome =
+    let* () = checkpoint_request () in
+    if should_complete_request () && not (should_defer_completion ())
+    then Build_request.complete request outcome
+    else Fiber.return ()
+  in
+  let run_request request =
+    Build_request.build request
+    |> run_action_builder_fiber
+    >>= function
+    | Ok () ->
+      let+ () = finish_request request Build_outcome.Success in
+      Ok ()
+    | Error exns when List.for_all exns ~f:caused_by_cancellation ->
+      Fiber.return (Error exns)
+    | Error exns ->
+      let+ () = finish_request request Build_outcome.Failure in
+      Error exns
+  in
+  run_with_error_collection ?restart_started_at ?run_id (fun () ->
+    let+ results = Fiber.parallel_map requests ~f:run_request in
+    match
+      List.concat_map results ~f:(function
+        | Ok () -> []
+        | Error exns -> exns)
+    with
+    | [] -> Ok ()
+    | exns -> Error exns)
 ;;
 
 let build_file p =

src/dune_engine/build_system.mli

@@ -60,10 +60,20 @@ val run
 (** A variant of [run] that raises an [Already_reported] exception on error. *)
 val run_exn : (unit -> 'a Memo.t) -> 'a Fiber.t
 
-val run_action_builder
+module Build_request : sig
+  type t
+
+  val create : unit Action_builder.t -> t
+  val await : t -> Build_outcome.t Fiber.t
+  val is_finished : t -> bool
+  val complete : t -> Build_outcome.t -> unit Fiber.t
+end
+
+val run_build_requests
   :  ?restart_started_at:Time.t
   -> ?run_id:Run_id.t
-  -> unit Action_builder.t
+  -> ?should_complete_request:(unit -> bool)
+  -> Build_request.t list
   -> (unit, [ `Already_reported ]) result Fiber.t
 
 (** {2 Misc} *)