Various refactorings in the RPC system

bin/build.ml

@@ -196,16 +196,18 @@ let build =
          perform the RPC call.
       *)
       let targets = Rpc.Rpc_common.prepare_targets targets in
-      Rpc.Rpc_common.run_via_rpc
-        ~common
-        ~config
-        (Rpc.Rpc_common.fire_request
-           ~name:"build"
-           ~wait:true
-           ~lock_held_by
-           builder
-           Dune_rpc_impl.Decl.build)
-        targets
+      Scheduler.go_without_rpc_server ~common ~config (fun () ->
+        let open Fiber.O in
+        let+ build_outcome =
+          Rpc.Rpc_common.fire_request
+            ~name:"build"
+            ~wait:true
+            ~lock_held_by
+            builder
+            Dune_rpc_impl.Decl.build
+            targets
+        in
+        Rpc.Rpc_common.wrap_build_outcome_exn ~print_on_success:true build_outcome)
     | Ok () ->
       let request setup =
         Target.interpret_targets (Common.root common) config setup targets

bin/diagnostics.ml

@@ -1,38 +1,24 @@
 open Import
 
-let exec () =
-  let open Fiber.O in
-  let where = Rpc.Rpc_common.active_server_exn () in
-  let module Client = Dune_rpc_client.Client in
-  let+ errors =
-    let* connect = Client.Connection.connect_exn where in
-    Dune_rpc_impl.Client.client
-      connect
-      (Dune_rpc_private.Initialize.Request.create
-         ~id:(Dune_rpc_private.Id.make (Sexp.Atom "diagnostics_cmd")))
-      ~f:(fun cli ->
-        let* decl =
-          Client.Versioned.prepare_request cli Dune_rpc_private.Public.Request.diagnostics
-        in
-        match decl with
-        | Error e -> raise (Dune_rpc_private.Version_error.E e)
-        | Ok decl -> Client.request cli decl ())
-  in
-  match errors with
-  | Ok errors ->
-    List.iter errors ~f:(fun err ->
-      Console.print_user_message (Dune_rpc.Diagnostic.to_user_message err))
-  | Error e -> Rpc.Rpc_common.raise_rpc_error e
-;;
-
 let info =
   let doc = "Fetch and return errors from the current build." in
   Cmd.info "diagnostics" ~doc
 ;;
 
 let term =
   let+ (builder : Common.Builder.t) = Common.Builder.term in
-  Rpc.Rpc_common.client_term builder exec
+  Rpc.Rpc_common.client_term builder (fun () ->
+    let open Fiber.O in
+    let+ errors =
+      Rpc.Rpc_common.fire_request
+        ~name:"diagnostics_cmd"
+        ~wait:false
+        builder
+        Dune_rpc_private.Procedures.Public.diagnostics
+        ()
+    in
+    List.iter errors ~f:(fun err ->
+      Console.print_user_message (Dune_rpc.Diagnostic.to_user_message err)))
 ;;
 
 let command = Cmd.v info term

bin/exec.ml

@@ -226,16 +226,17 @@ let build_prog_via_rpc_if_necessary ~dir ~no_rebuild builder lock_held_by prog =
             (Dune_lang.String_with_vars.make_text Loc.none (Path.to_string path))
         in
         let targets = Rpc.Rpc_common.prepare_targets [ target ] in
-        Rpc.Rpc_common.wrap_build_outcome_exn
-          ~print_on_success:false
-          (Rpc.Rpc_common.fire_request
-             ~name:"build"
-             ~wait:true
-             ~lock_held_by
-             builder
-             Dune_rpc_impl.Decl.build)
-          targets
-          ())
+        let open Fiber.O in
+        let+ build_outcome =
+          Rpc.Rpc_common.fire_request
+            ~name:"build"
+            ~wait:true
+            ~lock_held_by
+            builder
+            Dune_rpc_impl.Decl.build
+            targets
+        in
+        Rpc.Rpc_common.wrap_build_outcome_exn ~print_on_success:false build_outcome)
     in
     Path.to_absolute_filename path
   | Absolute ->

bin/fmt.ml

@@ -45,25 +45,16 @@ let run_fmt_command ~common ~config ~preview builder =
   | Error lock_held_by ->
     (* The --preview flag is being ignored by the RPC server, warn the user. *)
     if preview then Rpc.Rpc_common.warn_ignore_arguments lock_held_by;
-    let response =
-      Scheduler.go_without_rpc_server ~common ~config (fun () ->
-        Rpc.Rpc_common.fire_request
-          ~name:"format"
-          ~wait:true
-          ~warn_forwarding:false
-          ~lock_held_by
-          builder
-          Dune_rpc.Procedures.Public.format
-          ())
-    in
-    (match response with
-     | Ok () -> ()
-     | Error error ->
-       User_error.raise
-         [ Pp.paragraphf
-             "Error: %s\n%!"
-             (Dyn.to_string (Dune_rpc.Response.Error.to_dyn error))
-         ])
+    Scheduler.go_without_rpc_server
+      ~common
+      ~config
+      (Rpc.Rpc_common.fire_request
+         ~name:"format"
+         ~wait:true
+         ~warn_forwarding:false
+         ~lock_held_by
+         builder
+         Dune_rpc.Procedures.Public.format)
 ;;
 
 let command =

bin/promotion.ml

@@ -62,16 +62,18 @@ module Apply = struct
         let+ () = Fiber.return () in
         Diff_promotion.promote_files_registered_in_last_run files_to_promote)
     | Error lock_held_by ->
-      Rpc.Rpc_common.run_via_rpc
-        ~common
-        ~config
-        (Rpc.Rpc_common.fire_request
-           ~name:"promote_many"
-           ~wait:true
-           ~lock_held_by
-           builder
-           Dune_rpc_private.Procedures.Public.promote_many)
-        files_to_promote
+      Scheduler.go_without_rpc_server ~common ~config (fun () ->
+        let open Fiber.O in
+        let+ build_outcome =
+          Rpc.Rpc_common.fire_request
+            ~name:"promote_many"
+            ~wait:true
+            ~lock_held_by
+            builder
+            Dune_rpc_private.Procedures.Public.promote_many
+            files_to_promote
+        in
+        Rpc.Rpc_common.wrap_build_outcome_exn ~print_on_success:true build_outcome)
   ;;
 
   let command = Cmd.v info term

bin/rpc/rpc_build.ml

@@ -12,10 +12,8 @@ let term =
     Rpc_common.fire_request ~name:"build" ~wait builder Dune_rpc_impl.Decl.build targets
   in
   match response with
-  | Error (error : Dune_rpc.Response.Error.t) ->
-    Printf.eprintf "Error: %s\n%!" (Dyn.to_string (Dune_rpc.Response.Error.to_dyn error))
-  | Ok Success -> print_endline "Success"
-  | Ok (Failure _) -> print_endline "Failure"
+  | Success -> print_endline "Success"
+  | Failure _ -> print_endline "Failure"
 ;;
 
 let info =

bin/rpc/rpc_common.ml

@@ -24,14 +24,26 @@ let raise_rpc_error (e : Rpc_error.t) =
     ]
 ;;
 
-let request_exn client request n =
+let request_exn client request arg =
   let open Fiber.O in
   let* decl =
     Client.Versioned.prepare_request client (Dune_rpc.Decl.Request.witness request)
   in
   match decl with
+  | Ok decl -> Client.request client decl arg
+  | Error e -> raise (Dune_rpc.Version_error.E e)
+;;
+
+let notify_exn client notification arg =
+  let open Fiber.O in
+  let* res =
+    Client.Versioned.prepare_notification
+      client
+      (Dune_rpc.Decl.Notification.witness notification)
+  in
+  match res with
+  | Ok decl -> Client.notification client decl arg
   | Error e -> raise (Dune_rpc.Version_error.E e)
-  | Ok decl -> Client.request client decl n
 ;;
 
 let client_term builder f =
@@ -58,9 +70,9 @@ let establish_connection_exn () =
 ;;
 
 let establish_connection_with_retry () =
-  let open Fiber.O in
   let pause_between_retries_s = 0.2 in
   let rec loop () =
+    let open Fiber.O in
     establish_connection ()
     >>= function
     | Ok x -> Fiber.return x
@@ -92,6 +104,17 @@ let warn_ignore_arguments lock_held_by =
     ]
 ;;
 
+let should_warn ~warn_forwarding builder =
+  warn_forwarding && not (Common.Builder.equal builder Common.Builder.default)
+;;
+
+let send_request ~f connection name =
+  Dune_rpc_impl.Client.client
+    connection
+    (Dune_rpc.Initialize.Request.create ~id:(Dune_rpc.Id.make (Sexp.Atom name)))
+    ~f
+;;
+
 let fire_request
       ~name
       ~wait
@@ -103,23 +126,35 @@ let fire_request
   =
   let open Fiber.O in
   let* connection = establish_client_session ~wait in
-  if warn_forwarding && not (Common.Builder.equal builder Common.Builder.default)
-  then warn_ignore_arguments lock_held_by;
-  Dune_rpc_impl.Client.client
-    connection
-    (Dune_rpc.Initialize.Request.create ~id:(Dune_rpc.Id.make (Sexp.Atom name)))
-    ~f:(fun client -> request_exn client request arg)
+  if should_warn ~warn_forwarding builder then warn_ignore_arguments lock_held_by;
+  send_request connection name ~f:(fun client ->
+    let+ res = request_exn client request arg in
+    match res with
+    | Ok result -> result
+    | Error e -> raise_rpc_error e)
 ;;
 
-let wrap_build_outcome_exn ~print_on_success f args () =
+let fire_notification
+      ~name
+      ~wait
+      ?(warn_forwarding = true)
+      ?(lock_held_by = Dune_util.Global_lock.Lock_held_by.Unknown)
+      builder
+      notification
+      arg
+  =
   let open Fiber.O in
-  let+ response = f args in
-  match response with
-  | Error (error : Rpc_error.t) -> raise_rpc_error error
-  | Ok Dune_rpc.Build_outcome_with_diagnostics.Success ->
+  let* connection = establish_client_session ~wait in
+  if should_warn ~warn_forwarding builder then warn_ignore_arguments lock_held_by;
+  send_request connection name ~f:(fun client -> notify_exn client notification arg)
+;;
+
+let wrap_build_outcome_exn ~print_on_success build_outcome =
+  match build_outcome with
+  | Dune_rpc.Build_outcome_with_diagnostics.Success ->
     if print_on_success
     then Console.print [ Pp.text "Success" |> Pp.tag User_message.Style.Success ]
-  | Ok (Failure errors) ->
+  | Failure errors ->
     let error_msg =
       match List.length errors with
       | 0 ->
@@ -133,10 +168,3 @@ let wrap_build_outcome_exn ~print_on_success f args () =
       Console.print_user_message main);
     User_error.raise [ error_msg |> Pp.tag User_message.Style.Error ]
 ;;
-
-let run_via_rpc ~common ~config f args =
-  Scheduler.go_without_rpc_server
-    ~common
-    ~config
-    (wrap_build_outcome_exn ~print_on_success:true f args)
-;;

bin/rpc/rpc_common.mli

@@ -7,14 +7,6 @@ val active_server_exn : unit -> Dune_rpc.Where.t
 (** Raise an RPC response error. *)
 val raise_rpc_error : Dune_rpc.Response.Error.t -> 'a
 
-(** Make a request and raise an exception if the preparation for the request
-    fails in any way. Returns an [Error] if the response errors. *)
-val request_exn
-  :  Dune_rpc_client.Client.t
-  -> ('a, 'b) Dune_rpc.Decl.request
-  -> 'a
-  -> ('b, Dune_rpc.Response.Error.t) result Fiber.t
-
 (** Cmdliner term for a generic RPC client. *)
 val client_term : Common.Builder.t -> (unit -> 'a Fiber.t) -> 'a
 
@@ -38,26 +30,27 @@ val fire_request
   -> Common.Builder.t
   -> ('a, 'b) Dune_rpc.Decl.request
   -> 'a
-  -> ('b, Dune_rpc.Response.Error.t) result Fiber.t
+  -> 'b Fiber.t
+
+(** Send a notification to the RPC server. If [wait], it will poll forever until a server is listening.
+    Should be scheduled by a scheduler that does not come with a RPC server on its own.
+
+    [warn_forwarding] defaults to true, warns the user that since a RPC server is running, some arguments are ignored.
+    [lock_held_by] defaults to [Unknown], is only used to allow error messages to print the PID. *)
+val fire_notification
+  :  name:string
+  -> wait:bool
+  -> ?warn_forwarding:bool
+  -> ?lock_held_by:Dune_util.Global_lock.Lock_held_by.t
+  -> Common.Builder.t
+  -> 'a Dune_rpc.Decl.notification
+  -> 'a
+  -> unit Fiber.t
 
 val wrap_build_outcome_exn
   :  print_on_success:bool
-  -> ('a
-      -> (Dune_rpc.Build_outcome_with_diagnostics.t, Dune_rpc.Response.Error.t) result
-           Fiber.t)
-  -> 'a
+  -> Dune_rpc.Build_outcome_with_diagnostics.t
   -> unit
-  -> unit Fiber.t
 
 (** Warn the user that since a RPC server is running, some arguments are ignored. *)
 val warn_ignore_arguments : Dune_util.Global_lock.Lock_held_by.t -> unit
-
-(**  Schedule a fiber to run via RPC, wrapping any errors. *)
-val run_via_rpc
-  :  common:Common.t
-  -> config:Dune_config_file.Dune_config.t
-  -> ('a
-      -> (Dune_rpc.Build_outcome_with_diagnostics.t, Dune_rpc.Response.Error.t) result
-           Fiber.t)
-  -> 'a
-  -> unit

bin/rpc/rpc_ping.ml

@@ -15,15 +15,15 @@ let term =
   Rpc_common.client_term builder
   @@ fun () ->
   let open Fiber.O in
-  Rpc_common.fire_request
-    ~name:"ping_cmd"
-    ~wait
-    builder
-    Dune_rpc_private.Procedures.Public.ping
-    ()
-  >>| function
-  | Ok () -> Console.print [ Pp.text "Server appears to be responding normally" ]
-  | Error e -> Rpc_common.raise_rpc_error e
+  let+ () =
+    Rpc_common.fire_request
+      ~name:"ping_cmd"
+      ~wait
+      builder
+      Dune_rpc_private.Procedures.Public.ping
+      ()
+  in
+  Console.print [ Pp.text "Server appears to be responding normally" ]
 ;;
 
 let cmd = Cmd.v info term