diff --git a/ocaml/idl/datamodel_cluster_host.ml b/ocaml/idl/datamodel_cluster_host.ml
index 6ecc4bdd152..622c994c854 100644
--- a/ocaml/idl/datamodel_cluster_host.ml
+++ b/ocaml/idl/datamodel_cluster_host.ml
@@ -108,6 +108,10 @@ let t =
           ~ty:(Ref _pif) "PIF" ~default_value:(Some (VRef null_ref))
           "Reference to the PIF object"
 
+      ; field  ~qualifier:StaticRO ~lifecycle
+          ~ty:Bool "joined" ~default_value:(Some (VBool true))
+          "Whether the cluster host has joined the cluster"
+
       (* TODO: add `live` member to represent whether corosync believes that this
                cluster host actually is enabled *)
 
diff --git a/ocaml/idl/datamodel_errors.ml b/ocaml/idl/datamodel_errors.ml
index b85429a5014..e4e2cc262b1 100644
--- a/ocaml/idl/datamodel_errors.ml
+++ b/ocaml/idl/datamodel_errors.ml
@@ -1139,7 +1139,9 @@ let _ =
   error Api_errors.invalid_cluster_stack [ "cluster_stack" ]
     ~doc:"The cluster stack provided is not supported." ();
   error Api_errors.pif_not_attached_to_host [ "pif"; "host" ]
-    ~doc:"Cluster_host creation failed as the PIF provided is not attached to the host." ()
+    ~doc:"Cluster_host creation failed as the PIF provided is not attached to the host." ();
+  error Api_errors.cluster_host_not_joined [ "cluster_host" ]
+    ~doc:"Cluster_host operation failed as the cluster_host has not joined the cluster." ()
 
 
 let _ =
diff --git a/ocaml/tests/suite_alcotest.ml b/ocaml/tests/suite_alcotest.ml
index 6bbbffb7f85..23eaedcba0a 100644
--- a/ocaml/tests/suite_alcotest.ml
+++ b/ocaml/tests/suite_alcotest.ml
@@ -2,7 +2,7 @@
 let () =
   Suite_init.harness_init ();
   (* Alcotest hides the standard output of successful tests,
-     so we will probably not exceed the 4MB limit in Traivs *)
+     so we will probably not exceed the 4MB limit in Travis *)
   Debug.log_to_stdout ();
 
   Alcotest.run "Base suite"
@@ -30,6 +30,8 @@ let () =
     ; "Test_daemon_manager", Test_daemon_manager.test
     ; "Test_cluster", Test_cluster.test
     ; "Test_cluster_host", Test_cluster_host.test
+    ; "Test_clustering", Test_clustering.test
+    ; "Test_clustering_allowed_operations", Test_clustering_allowed_operations.test
     ; "Test_client", Test_client.test
     ; "Test_ca91480", Test_ca91480.test
     ; "Test_pgpu", Test_pgpu.test
@@ -43,8 +45,6 @@ let () =
     ; "Test_pvs_site", Test_pvs_site.test
     ; "Test_pvs_proxy", Test_pvs_proxy.test
     ; "Test_pvs_server", Test_pvs_server.test
-    ; "Test_clustering", Test_clustering.test
-    ; "Test_clustering_allowed_operations", Test_clustering_allowed_operations.test
     ; "Test_event", Test_event.test
     ; "Test_vm_placement", Test_vm_placement.test
     ; "Test_vm_memory_constraints", Test_vm_memory_constraints.test
diff --git a/ocaml/tests/test_cluster.ml b/ocaml/tests/test_cluster.ml
index 6bae9312c65..b4ded0d9682 100644
--- a/ocaml/tests/test_cluster.ml
+++ b/ocaml/tests/test_cluster.ml
@@ -58,9 +58,12 @@ let test_create_destroy_status () =
 let test_enable () =
   let __context = Test_common.make_test_database () in
   let cluster = create_cluster ~__context () in
-  (* simulate xapi getting restarted *)
 
-  Create_storage.maybe_reenable_cluster_host __context;
+  (* simulate xapi getting restarted *)
+  begin match Xapi_clustering.find_cluster_host ~__context ~host:Helpers.(get_localhost ~__context) with
+    | Some self -> Xapi_cluster_host.enable ~__context ~self
+    | None -> Alcotest.fail "Couldn't find freshly-created cluster_host"
+  end;
   pool_destroy ~__context ~self:cluster
 
 let test_invalid_cluster_stack () =
diff --git a/ocaml/tests/test_cluster_host.ml b/ocaml/tests/test_cluster_host.ml
index 144fdd91a70..960ac11e38a 100644
--- a/ocaml/tests/test_cluster_host.ml
+++ b/ocaml/tests/test_cluster_host.ml
@@ -64,17 +64,15 @@ let test_fix_prereq () =
   let __context = Test_common.make_test_database () in
   Context.set_test_rpc __context (pif_plug_rpc __context);
   let network = Test_common.make_network ~__context () in
-  let localhost = Helpers.get_localhost ~__context in
-  let pifref = Test_common.make_pif ~__context ~network ~host:localhost () in
-  let pif = Xapi_clustering.pif_of_host ~__context network localhost in
+  let host = Helpers.get_localhost ~__context in
+  let pifref = Test_common.make_pif ~__context ~network ~host () in
   Alcotest.check_raises
     "Should fail when checking PIF prequisites"
     Api_errors.(Server_error (pif_has_no_network_configuration, [ Ref.string_of pifref ]))
-    (fun () -> Xapi_cluster_host.fix_pif_prerequisites __context pif);
+    (fun () -> Xapi_cluster_host.fix_pif_prerequisites __context pifref);
   Db.PIF.set_IP ~__context ~self:pifref ~value:"1.1.1.1";
-  let pif = Xapi_clustering.pif_of_host ~__context network localhost in
-  Xapi_cluster_host.fix_pif_prerequisites ~__context pif;
-  let pif = Xapi_clustering.pif_of_host ~__context network localhost in
+  Xapi_cluster_host.fix_pif_prerequisites ~__context pifref;
+  let pif = Xapi_clustering.pif_of_host ~__context network host in
   Alcotest.(check unit)
     "PIF prerequisites have now been fixed"
     () (Xapi_clustering.assert_pif_prerequisites pif)
@@ -93,7 +91,7 @@ let test_create_as_necessary () =
   Alcotest.check_raises
     "create_as_necessary should fail if autojoin is set and the pool master has no cluster_host"
     Api_errors.(Server_error (internal_error,
-      [ Printf.sprintf "No cluster_host master found for cluster %s" (Ref.string_of cluster) ]))
+      [ Printf.sprintf "No cluster_host exists on master" ]))
     (fun () -> Xapi_cluster_host.create_as_necessary ~__context ~host:localhost);
   let _ = Test_common.make_cluster_host ~__context ~pIF:(fst _pif) ~host:(Helpers.get_master ~__context) ~cluster () in
   Xapi_cluster_host.create_as_necessary ~__context ~host:localhost;
@@ -101,7 +99,9 @@ let test_create_as_necessary () =
   check_cluster_option "sync_required with an existing cluster_host" None result;
   let host = Test_common.make_host ~__context () in
   let result = sync_required ~__context ~host in
-  check_cluster_option "sync_required with an existing cluster_host on master but not given host" (Some cluster) result
+  check_cluster_option
+    "sync_required with an existing cluster_host on master but not given host"
+    (Some cluster) result
 
 (* CA-275728 *)
 let test_destroy_forbidden_when_sr_attached () =
diff --git a/ocaml/tests/test_clustering_allowed_operations.ml b/ocaml/tests/test_clustering_allowed_operations.ml
index d16325a48ee..a3de22d9570 100644
--- a/ocaml/tests/test_clustering_allowed_operations.ml
+++ b/ocaml/tests/test_clustering_allowed_operations.ml
@@ -19,7 +19,7 @@ let assert_true msg x = Alcotest.(check bool) msg true x
 (** cluster_create is not allowed if a cluster already exists *)
 let test_pool_cluster_create_not_allowed_when_cluster_exists () =
   let __context = make_test_database () in
-  let self = Db.Pool.get_all ~__context |> List.hd in
+  let self = Helpers.get_pool ~__context in
   let _, _ = make_cluster_and_cluster_host ~__context () in
   Xapi_pool_helpers.update_allowed_operations ~__context ~self;
   let allowed_ops = Db.Pool.get_allowed_operations ~__context ~self in
@@ -29,7 +29,7 @@ let test_pool_cluster_create_not_allowed_when_cluster_exists () =
 (** cluster_create is not allowed if any pool operations are in progress *)
 let test_pool_cluster_create_not_allowed_during_pool_ops () =
   let __context = make_test_database () in
-  let self = Db.Pool.get_all ~__context |> List.hd in
+  let self = Helpers.get_pool ~__context in
   Xapi_pool_helpers.with_pool_operation ~__context ~self ~doc:"" ~op:`ha_enable
     (fun () ->
        let allowed_ops = Db.Pool.get_allowed_operations ~__context ~self in
@@ -40,7 +40,7 @@ let test_pool_cluster_create_not_allowed_during_pool_ops () =
     operations in progress *)
 let test_pool_cluster_create_allowed () =
   let __context = make_test_database () in
-  let self = Db.Pool.get_all ~__context |> List.hd in
+  let self = Helpers.get_pool ~__context in
   Xapi_pool_helpers.update_allowed_operations ~__context ~self;
   let allowed_ops = Db.Pool.get_allowed_operations ~__context ~self in
   assert_true "Pool.allowed_operations should contain 'cluster_create'"
@@ -101,6 +101,14 @@ let test_cluster_host_ops_not_allowed_during_cluster_host_op () =
        let allowed_ops = Db.Cluster_host.get_allowed_operations ~__context ~self in
        assert_true "Cluster_host.allowed_operations should be empty" (allowed_ops = []))
 
+let with_cluster_op ~__context self op =
+  Xapi_cluster_helpers.with_cluster_operation ~__context ~self ~doc:"" ~op
+    (fun () -> ())
+
+let with_cluster_host_op ~__context self op =
+  Xapi_cluster_host_helpers.with_cluster_host_operation ~__context ~self ~doc:"" ~op
+    (fun () -> ())
+
 let test_clustering_ops_disallowed_during_rolling_upgrade () =
   let __context = Test_common.make_test_database () in
 
@@ -110,17 +118,9 @@ let test_clustering_ops_disallowed_during_rolling_upgrade () =
       (fun op ->
          Alcotest.(check unit)
            "Clustering operations should be allowed"
-           () (with_cluster_fn self op)
+           () (with_cluster_fn ~__context self op)
       ) ops
   in
-  let with_cluster_op self op =
-    Xapi_cluster_helpers.with_cluster_operation ~__context ~self ~doc:"" ~op
-      (fun () -> ())
-  in
-  let with_cluster_host_op self op =
-    Xapi_cluster_host_helpers.with_cluster_host_operation ~__context ~self ~doc:"" ~op
-      (fun () -> ())
-  in
   let cluster, cluster_host =
     Test_common.make_cluster_and_cluster_host ~__context ()
   in
@@ -151,7 +151,7 @@ let test_clustering_ops_disallowed_during_rolling_upgrade () =
        Alcotest.check_raises
          "Other than cluster_host enable/disable, no clustering operations should be allowed during RPU"
          Api_errors.(Server_error (not_supported_during_upgrade, []))
-         (fun () -> with_cluster_op cluster op)
+         (fun () -> with_cluster_op ~__context cluster op)
     ) [ `add ; `remove ; `destroy];
 
   test_clustering_ops_should_pass
@@ -161,6 +161,20 @@ let test_clustering_ops_disallowed_during_rolling_upgrade () =
 
   test_cluster_host_operations_valid ()
 
+let test_cluster_host_ops_without_join () =
+  (* Note that joined:true by default so no need to check *)
+  let __context = make_test_database () in
+  let cluster, cluster_host = make_cluster_and_cluster_host ~__context ~host:Helpers.(get_localhost ~__context) () in
+  Db.Cluster_host.set_joined ~__context ~self:cluster_host ~value:false;
+
+  List.iter
+    (fun op ->
+      Alcotest.check_raises
+        "Non-remove cluster operations invalid when not cluster_host.joined"
+        Api_errors.(Server_error (cluster_host_not_joined, [ Ref.string_of cluster_host ]))
+        (fun () -> with_cluster_host_op ~__context cluster_host op)
+    ) Xapi_cluster_host_helpers.all_cluster_host_operations
+
 let test =
   [ "test_pool_cluster_create_not_allowed_when_cluster_exists", `Quick, test_pool_cluster_create_not_allowed_when_cluster_exists
   ; "test_pool_cluster_create_not_allowed_during_pool_ops", `Quick, test_pool_cluster_create_not_allowed_during_pool_ops
@@ -171,4 +185,5 @@ let test =
   ; "test_cluster_host_enable_allowed", `Quick, test_cluster_host_enable_allowed
   ; "test_cluster_host_ops_not_allowed_during_cluster_host_op", `Quick, test_cluster_host_ops_not_allowed_during_cluster_host_op
   ; "test_clustering_ops_disallowed_during_rolling_upgrade", `Quick, test_clustering_ops_disallowed_during_rolling_upgrade
+  ; "test_cluster_host_ops_without_join", `Quick, test_cluster_host_ops_without_join
   ]
diff --git a/ocaml/tests/test_common.ml b/ocaml/tests/test_common.ml
index 26cbb33b7da..b41fdf819af 100644
--- a/ocaml/tests/test_common.ml
+++ b/ocaml/tests/test_common.ml
@@ -489,10 +489,10 @@ let make_vfs_on_pf ~__context ~pf ~num =
   make_vf num
 
 let make_cluster_host ~__context ?(ref=Ref.make ()) ?(uuid=make_uuid ())
-    ?(cluster=Ref.null) ?(host=Ref.null) ?(pIF=Ref.null) ?(enabled=true)
+    ?(cluster=Ref.null) ?(host=Ref.null) ?(pIF=Ref.null) ?(enabled=true) ?(joined=true)
     ?(allowed_operations=[]) ?(current_operations=[]) ?(other_config=[]) () =
   Db.Cluster_host.create ~__context ~ref ~uuid ~cluster ~host ~pIF ~enabled
-    ~allowed_operations ~current_operations ~other_config;
+    ~allowed_operations ~current_operations ~other_config ~joined;
   ref
 
 let make_cluster_and_cluster_host ~__context ?(ref=Ref.make ()) ?(uuid=make_uuid ())
diff --git a/ocaml/xapi-consts/api_errors.ml b/ocaml/xapi-consts/api_errors.ml
index 71b14fc9688..aecbd380c9a 100644
--- a/ocaml/xapi-consts/api_errors.ml
+++ b/ocaml/xapi-consts/api_errors.ml
@@ -614,3 +614,4 @@ let cluster_force_destroy_failed = "CLUSTER_FORCE_DESTROY_FAILED"
 let cluster_stack_in_use = "CLUSTER_STACK_IN_USE"
 let invalid_cluster_stack = "INVALID_CLUSTER_STACK"
 let pif_not_attached_to_host = "PIF_NOT_ATTACHED_TO_HOST"
+let cluster_host_not_joined = "CLUSTER_HOST_NOT_JOINED"
diff --git a/ocaml/xapi/create_storage.ml b/ocaml/xapi/create_storage.ml
index bfabc8d1bbd..becb42f5976 100644
--- a/ocaml/xapi/create_storage.ml
+++ b/ocaml/xapi/create_storage.ml
@@ -41,17 +41,10 @@ let plug_all_pbds __context =
     my_pbds;
   !result
 
-let maybe_reenable_cluster_host __context =
-  let host = Helpers.get_localhost __context in
-  match Xapi_clustering.find_cluster_host ~__context ~host with
-  | Some self ->
-     Xapi_cluster_host.enable ~__context ~self
-  | None -> ()
 
 let plug_unplugged_pbds __context =
   (* If the plug is to succeed for SM's requiring a cluster stack
    * we have to enable the cluster stack too if we have one *)
-  log_and_ignore_exn(fun () -> maybe_reenable_cluster_host __context);
   let my_pbds = Helpers.get_my_pbds __context in
   List.iter
     (fun (self, pbd_record) ->
diff --git a/ocaml/xapi/records.ml b/ocaml/xapi/records.ml
index cc4487da39d..d7265d1f065 100644
--- a/ocaml/xapi/records.ml
+++ b/ocaml/xapi/records.ml
@@ -2144,6 +2144,9 @@ let cluster_host_record rpc session_id cluster_host =
       ; make_field ~name:"enabled"
           ~get:(fun () -> (x ()).API.cluster_host_enabled |> string_of_bool)
           ()
+      ; make_field ~name:"joined"
+          ~get:(fun () -> (x ()).API.cluster_host_joined |> string_of_bool)
+          ()
       ; make_field ~name:"allowed-operations"
           ~get:(fun () -> String.concat "; " (List.map Record_util.cluster_host_operation_to_string (x ()).API.cluster_host_allowed_operations))
           ~get_set:(fun () -> List.map Record_util.cluster_host_operation_to_string (x ()).API.cluster_host_allowed_operations)
diff --git a/ocaml/xapi/xapi.ml b/ocaml/xapi/xapi.ml
index 97b52cda6f7..c49a5ae4367 100644
--- a/ocaml/xapi/xapi.ml
+++ b/ocaml/xapi/xapi.ml
@@ -910,8 +910,6 @@ let server_init() =
           (* CA-22417: bring up all non-bond slaves so that the SM backends can use storage NIC IP addresses (if the routing
              	 table happens to be right) *)
           "Best-effort bring up of physical and sriov NICs", [ Startup.NoExnRaising ], Xapi_pif.start_of_day_best_effort_bring_up;
-          "Create any necessary cluster_host objects", [ Startup.NoExnRaising ], (fun () -> Xapi_cluster_host.create_as_necessary __context (Helpers.get_localhost ~__context));
-          "resync cluster host state", [], (fun () -> Xapi_cluster_host.resync_host ~__context ~host:Helpers.(get_localhost ~__context));
           "updating the vswitch controller", [], (fun () -> Helpers.update_vswitch_controller ~__context ~host:(Helpers.get_localhost ~__context));
           "initialising storage", [ Startup.NoExnRaising ],
           (fun () -> Helpers.call_api_functions ~__context Create_storage.create_storage_localhost);
@@ -943,20 +941,43 @@ let server_init() =
 
         let wait_management_interface () =
           let management_if = Xapi_inventory.lookup Xapi_inventory._management_interface in
-          if management_if <> "" then (
+          if management_if <> "" then begin
             debug "Waiting forever for the management interface to gain an IP address";
             let ip = wait_for_management_ip_address ~__context in
-            debug "Management interface got IP address: %s; attempting to re-plug any unplugged PBDs" ip;
+            debug "Management interface got IP address: %s, attempting to re-plug unplugged PBDs" ip;
+            (* This may fail without the clustering IP, which is why we attempt
+               another replug in maybe_wait_for_clustering_ip *)
             Helpers.call_api_functions ~__context (fun rpc session_id ->
                 Create_storage.plug_unplugged_pbds __context)
-          )
+          end
+        in
+
+        let maybe_wait_for_clustering_ip () =
+          let host = Helpers.get_localhost ~__context in
+          match Xapi_clustering.find_cluster_host ~__context ~host with
+          | Some self -> begin
+              debug "Waiting forever for cluster_host to gain an IP address";
+              let ip = Xapi_mgmt_iface.(wait_for_clustering_ip ~__context ~self) in
+              debug "Got clustering IP %s, resyncing cluster_host %s" ip (Ref.string_of self);
+              Xapi_cluster_host.resync_host ~__context ~host;
+              debug "Attempting to re-plug remaining unplugged PBDs";
+              Helpers.call_api_functions ~__context (fun rpc session_id ->
+                  Create_storage.plug_unplugged_pbds __context)
+            end
+          | None -> ()
         in
 
         Startup.run ~__context [
           "fetching database backup", [ Startup.OnlySlave; Startup.NoExnRaising ],
           (fun () -> Pool_db_backup.fetch_database_backup ~master_address:(Pool_role.get_master_address())
               ~pool_secret:!Xapi_globs.pool_secret ~force:None);
-          "wait management interface to come up", [ Startup.NoExnRaising ], wait_management_interface;
+          "wait management interface to come up, re-plug unplugged PBDs", [ Startup.NoExnRaising ], wait_management_interface;
+
+          (* CA-290237, CA-290473: Create cluster objects after network objects and management IP initialised *)
+          "Create any necessary cluster_host objects", [ Startup.NoExnRaising ],
+          (fun () -> log_and_ignore_exn (fun () -> Xapi_cluster_host.create_as_necessary __context (Helpers.get_localhost ~__context)));
+          "wait for clustering IP if any, re-plug remaining unplugged PBDs", [ Startup.OnThread ],
+          (fun () -> log_and_ignore_exn (fun () -> maybe_wait_for_clustering_ip () ));
           "considering sending a master transition alert", [ Startup.NoExnRaising; Startup.OnlyMaster ],
           Xapi_pool_transition.consider_sending_alert __context;
           "Cancelling in-progress storage migrations", [], (fun () -> Storage_migrate.killall ~dbg:"xapi init");
@@ -972,7 +993,7 @@ let server_init() =
         ];
 
         debug "startup: startup sequence finished");
-    wait_to_die()
+    wait_to_die ()
   with
   | Sys.Break -> cleanup_handler 0
   | (Unix.Unix_error (e,s1,s2)) as exn ->
diff --git a/ocaml/xapi/xapi_cluster.ml b/ocaml/xapi/xapi_cluster.ml
index c29ddba4562..85d98efaa07 100644
--- a/ocaml/xapi/xapi_cluster.ml
+++ b/ocaml/xapi/xapi_cluster.ml
@@ -30,7 +30,6 @@ let create ~__context ~pIF ~cluster_stack ~pool_auto_join ~token_timeout ~token_
   (* Currently we only support corosync. If we support more cluster stacks, this
    * should be replaced by a general function that checks the given cluster_stack *)
   Pool_features.assert_enabled ~__context ~f:Features.Corosync;
-  (* TODO: take network lock *)
   with_clustering_lock (fun () ->
       let dbg = Context.string_of_task __context in
       validate_params ~token_timeout ~token_timeout_coefficient;
@@ -64,7 +63,7 @@ let create ~__context ~pIF ~cluster_stack ~pool_auto_join ~token_timeout ~token_
           ~pool_auto_join ~token_timeout ~token_timeout_coefficient ~current_operations:[] ~allowed_operations:[] ~cluster_config:[]
           ~other_config:[];
         Db.Cluster_host.create ~__context ~ref:cluster_host_ref ~uuid:cluster_host_uuid ~cluster:cluster_ref ~host ~enabled:true ~pIF
-          ~current_operations:[] ~allowed_operations:[] ~other_config:[];
+          ~current_operations:[] ~allowed_operations:[] ~other_config:[] ~joined:true;
         Xapi_cluster_host_helpers.update_allowed_operations ~__context ~self:cluster_host_ref;
         D.debug "Created Cluster: %s and Cluster_host: %s" (Ref.string_of cluster_ref) (Ref.string_of cluster_host_ref);
         set_ha_cluster_stack ~__context;
@@ -142,6 +141,7 @@ let pool_force_destroy ~__context ~self =
   let slave_cluster_hosts =
     Db.Cluster.get_cluster_hosts ~__context ~self |> filter_on_option master_cluster_host
   in
+  debug "Destroying cluster_hosts in pool";
   (* First try to destroy each cluster_host - if we can do so safely then do *)
   List.iter
     (fun cluster_host ->
@@ -175,14 +175,14 @@ let pool_force_destroy ~__context ~self =
     [] all_remaining_cluster_hosts
     in
 
-    begin
-    match exns with
-    | [] -> D.debug "Cluster.force_destroy was successful"
-    | e :: _ -> raise Api_errors.(Server_error (cluster_force_destroy_failed, [Ref.string_of self]))
+    begin match exns with
+      | [] -> D.debug "Successfully destroyed all cluster_hosts in pool, now destroying cluster %s" (Ref.string_of self)
+      | e :: _ -> raise Api_errors.(Server_error (cluster_force_destroy_failed, [Ref.string_of self]))
     end;
 
     Helpers.call_api_functions ~__context (fun rpc session_id ->
-        Client.Client.Cluster.destroy ~rpc ~session_id ~self)
+      Client.Client.Cluster.destroy ~rpc ~session_id ~self);
+    debug "Cluster_host.force_destroy was successful"
 
 (* Helper function; concurrency checks are done in implementation of Cluster.destroy and Cluster_host.destroy *)
 let pool_destroy ~__context ~self =
@@ -212,6 +212,7 @@ let pool_resync ~__context ~(self : API.ref_Cluster) =
   List.iter
     (fun host -> log_and_ignore_exn
         (fun () ->
+           Xapi_cluster_host.create_as_necessary ~__context ~host;
            Xapi_cluster_host.resync_host ~__context ~host;
            if is_clustering_disabled_on_host ~__context host
            then raise Api_errors.(Server_error (no_compatible_cluster_host, [Ref.string_of host]))
diff --git a/ocaml/xapi/xapi_cluster_helpers.ml b/ocaml/xapi/xapi_cluster_helpers.ml
index 43a17fbd966..dea945ba5bf 100644
--- a/ocaml/xapi/xapi_cluster_helpers.ml
+++ b/ocaml/xapi/xapi_cluster_helpers.ml
@@ -45,23 +45,29 @@ let get_operation_error ~__context ~self ~op =
   let ref_str = Ref.string_of self in
 
   let current_error = None in
-
   let check c f = match c with | Some e -> Some e | None -> f () in
 
+  let assert_allowed_during_rpu __context = function
+    | `add | `remove | `destroy when Helpers.rolling_upgrade_in_progress ~__context ->
+      Some (Api_errors.not_supported_during_upgrade, [])
+    | _ -> None
+  in
+
   (* if other operations are in progress, check that the new operation is allowed concurrently with them *)
   let current_error = check current_error (fun () ->
       let current_ops = cr.Db_actions.cluster_current_operations in
-      if (current_ops <> []) && not (is_allowed_concurrently ~op ~current_ops)
-      then report_concurrent_operations_error ~current_ops ~ref_str
-      else None)
+      match current_ops with
+      | _::_ when not (is_allowed_concurrently ~op ~current_ops) ->
+        report_concurrent_operations_error ~current_ops ~ref_str
+      | _ ->
+        check
+          (assert_allowed_during_rpu __context op)
+          (fun () -> None)
+    )
   in
   current_error
 
 let assert_operation_valid ~__context ~self ~op =
-  begin match (Helpers.rolling_upgrade_in_progress ~__context), op with
-    | true, (`add | `remove | `destroy) -> raise Api_errors.(Server_error (not_supported_during_upgrade, []))
-    | _, _                              -> ()
-  end;
   match get_operation_error ~__context ~self ~op with
   | None       -> ()
   | Some (a,b) -> raise (Api_errors.Server_error (a,b))
diff --git a/ocaml/xapi/xapi_cluster_host.ml b/ocaml/xapi/xapi_cluster_host.ml
index 8dfca63beaa..7f074e97706 100644
--- a/ocaml/xapi/xapi_cluster_host.ml
+++ b/ocaml/xapi/xapi_cluster_host.ml
@@ -22,119 +22,110 @@ open D
 (* We can't fix _all_ of the prerequisites, as we can't automatically
    create an IP address. So what we do here is to at least plug the
    thing in and ensure it has disallow unplug set. *)
-let fix_pif_prerequisites ~__context (pif_ref,pif_rec) =
+let fix_pif_prerequisites ~__context (self : API.ref_PIF) =
   (* The following is to raise an exception if there's no IP. This
      avoids making any changes to the PIF if there's something we
      simply can't fix. *)
-  ignore(ip_of_pif (pif_ref,pif_rec));
-  if not pif_rec.API.pIF_currently_attached then
+  let pif_rec self = Db.PIF.get_record ~__context ~self in
+  ip_of_pif (self,pif_rec self) |> ignore;
+  if not (pif_rec self).API.pIF_currently_attached then
     Helpers.call_api_functions ~__context (fun rpc session_id ->
-      Client.Client.PIF.plug ~rpc ~session_id ~self:pif_ref);
-  if not pif_rec.API.pIF_disallow_unplug then begin
-    debug "Setting disallow_unplug on cluster PIF";
-    Db.PIF.set_disallow_unplug ~__context ~self:pif_ref ~value:true
+      Client.Client.PIF.plug ~rpc ~session_id ~self);
+  if not (pif_rec self).API.pIF_disallow_unplug then begin
+    debug "Setting disallow_unplug on cluster PIF %s" (Ref.string_of self);
+    Db.PIF.set_disallow_unplug ~__context ~self ~value:true
   end
 
-let sync_required ~__context ~host =
-  let clusters = Db.Cluster.get_all_records ~__context in
-  match clusters with
-  | [] -> None
-  | [cluster_ref, cluster_rec] -> begin
-      let expr = Db_filter_types.(And (Eq (Field "host", Literal (Ref.string_of host)),
-                                       Eq (Field "cluster", Literal (Ref.string_of cluster_ref)))) in
-      let my_cluster_hosts = Db.Cluster_host.get_internal_records_where ~__context ~expr in
-      match my_cluster_hosts with
-      | [(_ref,_rec)] -> None
-      | [] ->
-        if cluster_rec.API.cluster_pool_auto_join
-        then Some cluster_ref
-        else None
-      | _ -> raise Api_errors.(Server_error (internal_error, [ "Host cannot be associated with more than one cluster_host"; Ref.string_of host ]))
-    end
-  | _ -> raise Api_errors.(Server_error (internal_error, ["Cannot have more than one Cluster object per pool currently"]))
-
 let call_api_function_with_alert ~__context ~msg ~cls ~obj_uuid ~body
   ~(api_func : (Rpc.call -> Rpc.response) -> API.ref_session -> unit) =
-    Helpers.call_api_functions ~__context
-      (fun rpc session_id ->
+    Helpers.call_api_functions ~__context (fun rpc session_id ->
         try
           api_func rpc session_id
         with err ->
+          Backtrace.is_important err;
           let body = Printf.sprintf "Error: %s\nMessage: %s" ExnHelper.(string_of_exn err) body in
           Xapi_alert.add ~msg ~cls ~obj_uuid ~body;
           raise err
       )
 
-let create_as_necessary ~__context ~host =
-  match sync_required ~__context ~host with
-  | Some cluster_ref -> (* assume pool autojoin set *)
-    let network = get_network_internal ~__context ~self:cluster_ref in
-    let (pifref,pifrec) = Xapi_clustering.pif_of_host ~__context network host in
-    fix_pif_prerequisites ~__context (pifref,pifrec);
-
-    (* Best effort Cluster_host.create *)
-    let body = Printf.sprintf "Unable to create cluster host on %s."
-        (Db.Host.get_name_label ~__context ~self:host) in
-    let obj_uuid = Db.Host.get_uuid ~__context ~self:host in
-    call_api_function_with_alert ~__context
-      ~msg:Api_messages.cluster_host_creation_failed
-      ~cls:`Host ~obj_uuid ~body
-      ~api_func:(fun rpc session_id -> ignore
-          (Client.Client.Cluster_host.create rpc session_id cluster_ref host pifref))
-  | None -> ()
-
-let resync_host ~__context ~host =
-  create_as_necessary ~__context ~host;
-  match (find_cluster_host ~__context ~host) with
-  | None              -> () (* no clusters exist *)
-  | Some cluster_host ->    (* cluster_host and cluster exist *)
-    (* Cluster_host.enable unconditionally invokes the low-level enable operations and is idempotent. *)
-    if Db.Cluster_host.get_enabled ~__context ~self:cluster_host then begin
-      (* RPU reformats partition, losing service status, never re-enables clusterd *)
-      debug "Cluster_host %s is enabled, starting up xapi-clusterd" (Ref.string_of cluster_host);
-      Xapi_clustering.Daemon.enable ~__context;
-
-      (* Best-effort Cluster_host.enable *)
-      let body = Printf.sprintf "Unable to create cluster host on %s."
-          (Db.Host.get_name_label ~__context ~self:host) in
-      let obj_uuid = Db.Host.get_uuid ~__context ~self:host in
-      call_api_function_with_alert ~__context
-        ~msg:Api_messages.cluster_host_enable_failed
-        ~cls:`Host ~obj_uuid ~body
-        ~api_func:(fun rpc session_id -> ignore
-            (Client.Client.Cluster_host.enable rpc session_id cluster_host))
-    end
-
-let create ~__context ~cluster ~host ~pif =
-  (* TODO: take network lock *)
+(* Create xapi db object for cluster_host, resync_host calls clusterd *)
+let create_internal ~__context ~cluster ~host ~pIF : API.ref_Cluster_host =
   with_clustering_lock (fun () ->
       assert_operation_host_target_is_localhost ~__context ~host;
+      assert_pif_attached_to ~host ~pIF ~__context;
       assert_cluster_host_can_be_created ~__context ~host;
-      assert_pif_attached_to ~host ~pif ~__context;
       let ref = Ref.make () in
-      let dbg = Context.string_of_task __context in
       let uuid = Uuidm.to_string (Uuidm.create `V4) in
+      Db.Cluster_host.create ~__context ~ref ~uuid ~cluster ~host ~pIF ~enabled:false
+        ~current_operations:[] ~allowed_operations:[] ~other_config:[] ~joined:false;
+      ref
+  )
+
+(* Helper function atomically enables clusterd and joins the cluster_host *)
+let join_internal ~__context ~self =
+  with_clustering_lock (fun () ->
+
+      let pIF = Db.Cluster_host.get_PIF ~__context ~self in
+      fix_pif_prerequisites ~__context pIF;
+
+      let dbg = Context.string_of_task __context in
+      let cluster = Db.Cluster_host.get_cluster ~__context ~self in
       let cluster_token = Db.Cluster.get_cluster_token ~__context ~self:cluster in
-      let pifref,pifrec = pif,(Db.PIF.get_record ~__context ~self:pif) in
-      assert_pif_prerequisites (pifref,pifrec);
-      let ip = ip_of_pif (pifref,pifrec) in
-      let ip_list = List.map (fun cluster_host ->
-        let pref = Db.Cluster_host.get_PIF ~__context ~self:cluster_host in
-        let prec = Db.PIF.get_record ~__context ~self:pref in
-          ip_of_pif (pref,prec)
+      let ip = ip_of_pif (pIF, Db.PIF.get_record ~__context ~self:pIF) in
+      let ip_list = List.map (fun self ->
+        let p_ref = Db.Cluster_host.get_PIF ~__context ~self in
+        let p_rec = Db.PIF.get_record ~__context ~self:p_ref in
+          ip_of_pif (p_ref,p_rec)
         ) (Db.Cluster.get_cluster_hosts ~__context ~self:cluster) in
+
+      debug "Enabling clusterd and joining cluster_host %s" (Ref.string_of self);
       Xapi_clustering.Daemon.enable ~__context;
-      let result = Cluster_client.LocalClient.join (rpc ~__context) dbg cluster_token ip ip_list in
+      let result =
+        Cluster_client.LocalClient.join (rpc ~__context) dbg cluster_token ip ip_list
+      in
       match result with
       | Result.Ok () ->
-        Db.Cluster_host.create ~__context ~ref ~uuid ~cluster ~host ~pIF:pifref ~enabled:true
-          ~current_operations:[] ~allowed_operations:[] ~other_config:[];
-        debug "Cluster_host.create was successful; cluster_host: %s" (Ref.string_of ref);
-        ref
+        debug "Cluster join create was successful for cluster_host %s" (Ref.string_of self);
+        Db.Cluster_host.set_joined ~__context ~self ~value:true;
+        Db.Cluster_host.set_enabled ~__context ~self ~value:true;
+        debug "Cluster_host %s joined and enabled" (Ref.string_of self)
       | Result.Error error ->
-        warn "Error occurred during Cluster_host.create";
+        warn "Error occurred when joining cluster_host %s" (Ref.string_of self);
         handle_error error
-    )
+  )
+
+(* Enable cluster_host in client layer via clusterd *)
+let resync_host ~__context ~host =
+  match find_cluster_host ~__context ~host with
+  | None      -> () (* no clusters exist *)
+  | Some self ->    (* cluster_host and cluster exist *)
+    let body = Printf.sprintf "Unable to create cluster host on %s."
+        (Db.Host.get_name_label ~__context ~self:host) in
+    let obj_uuid = Db.Host.get_uuid ~__context ~self:host in
+
+    call_api_function_with_alert ~__context
+      ~msg:Api_messages.cluster_host_enable_failed
+      ~cls:`Host ~obj_uuid ~body
+      ~api_func:(fun rpc session_id ->
+          (* If we have just joined, enable will prevent concurrent clustering ops *)
+          if not (Db.Cluster_host.get_joined ~__context ~self)
+          then join_internal ~__context ~self
+          else
+            if Db.Cluster_host.get_enabled ~__context ~self then begin
+              (* [enable] unconditionally invokes low-level enable operations and is idempotent.
+                 RPU reformats partition, losing service status, never re-enables clusterd *)
+              debug "Cluster_host %s is enabled, starting up xapi-clusterd" (Ref.string_of self);
+              Xapi_clustering.Daemon.enable ~__context;
+
+              (* Note that join_internal and enable both use the clustering lock *)
+              Client.Client.Cluster_host.enable rpc session_id self end
+          )
+
+(* API call split into separate functions to create in db and enable in client layer *)
+let create ~__context ~cluster ~host ~pif =
+  let cluster_host : API.ref_Cluster_host = create_internal ~__context ~cluster ~host ~pIF:pif in
+  resync_host ~__context ~host;
+  cluster_host
 
 let force_destroy ~__context ~self =
   let dbg = Context.string_of_task __context in
@@ -246,3 +237,30 @@ let disable_clustering ~__context =
      info "Disabling cluster_host %s" (Ref.string_of self);
      disable ~__context ~self
 
+let sync_required ~__context ~host =
+  let clusters = Db.Cluster.get_all_records ~__context in
+  match clusters with
+  | [] -> None
+  | [cluster_ref, cluster_rec] -> begin
+      let expr = Db_filter_types.(And (Eq (Field "host", Literal (Ref.string_of host)),
+                                       Eq (Field "cluster", Literal (Ref.string_of cluster_ref)))) in
+      let my_cluster_hosts = Db.Cluster_host.get_internal_records_where ~__context ~expr in
+      match my_cluster_hosts with
+      | [(_ref,_rec)] -> None
+      | [] ->
+        if cluster_rec.API.cluster_pool_auto_join
+        then Some cluster_ref
+        else None
+      | _ -> raise Api_errors.(Server_error (internal_error, [ "Host cannot be associated with more than one cluster_host"; Ref.string_of host ]))
+    end
+  | _ -> raise Api_errors.(Server_error (internal_error, ["Cannot have more than one Cluster object per pool currently"]))
+
+(* If cluster found without local cluster_host, create one in db *)
+let create_as_necessary ~__context ~host =
+  match sync_required ~__context ~host with
+  | Some cluster -> (* assume pool autojoin set *)
+    let network = get_network_internal ~__context ~self:cluster in
+    let (pIF,pifrec) = Xapi_clustering.pif_of_host ~__context network host in
+    create_internal ~__context ~cluster ~host ~pIF |> ignore
+  | None -> ()
+
diff --git a/ocaml/xapi/xapi_cluster_host.mli b/ocaml/xapi/xapi_cluster_host.mli
index b783e547ce0..ff0b2e2b2e7 100644
--- a/ocaml/xapi/xapi_cluster_host.mli
+++ b/ocaml/xapi/xapi_cluster_host.mli
@@ -18,15 +18,15 @@
 (******************************************************************************)
 (** {2 Internal helper functions} *)
 
-val fix_pif_prerequisites : __context:Context.t -> (API.ref_PIF * API.pIF_t) ->
-  unit
+val fix_pif_prerequisites : __context:Context.t -> API.ref_PIF ->
+    unit
 (* [fix_pif_prerequisites ~__context (pif_ref,pif_rec)] will fix those
    prerequisites that are fixable automatically. It won't be able to fix a
    missing IP address, but it will plug the PIF if it's not attached and it
    will set disallow_unplug once the PIF is plugged *)
 
 val sync_required : __context:Context.t -> host:API.ref_host ->
-  API.ref_Cluster option
+    API.ref_Cluster option
 (** [sync_required ~__context ~host] returns an option type indicating whether
     any action is required to sync the cluster. This will only be the case if
     the cluster object has [pool_auto_join] set and no corresponding
@@ -35,7 +35,7 @@ val sync_required : __context:Context.t -> host:API.ref_host ->
 
 val create_as_necessary : __context:Context.t -> host:API.ref_host -> unit
 (** [create_as_necessary ~__context ~host] calls [sync_required], and if any
-    Cluster_host objects are required it will create them *)
+    Cluster_host objects are required it will create them in the database *)
 
 (******************************************************************************)
 (** {2 External API calls} *)
@@ -43,7 +43,9 @@ val create_as_necessary : __context:Context.t -> host:API.ref_host -> unit
 val create : __context:Context.t -> cluster:API.ref_Cluster -> host:API.ref_host -> pif:API.ref_PIF
   -> API.ref_Cluster_host
 (** [create ~__context ~cluster ~host] is implementation of the XenAPI call
-    'Cluster_host.create'. It is the Cluster_host object constructor *)
+    'Cluster_host.create'. It is the Cluster_host object constructor, and creates
+    a cluster_host in the DB before calling [resync_host ~__context ~host], which
+    either joins the host to the cluster or enables the cluster host *)
 
 val force_destroy : __context:Context.t -> self:API.ref_Cluster_host -> unit
 (** [force_destroy ~__context ~self] is the implementation of the XenAPI call
@@ -52,7 +54,9 @@ val force_destroy : __context:Context.t -> self:API.ref_Cluster_host -> unit
 
 val destroy : __context:Context.t -> self:API.ref_Cluster_host -> unit
 (** [destroy ~__context ~self] is the implementation of the XenAPI call
-    'Cluster_host.destroy'. It is the Cluster_host destructor *)
+    'Cluster_host.destroy'. It is the Cluster_host destructor
+    Note that this is the only Cluster_host call that is still valid if the
+    clustering daemon is disabled, all others require it enabled *)
 
 val enable : __context:Context.t -> self:API.ref_Cluster_host -> unit
 (** [enable ~__context ~self] is the implementation of the XenAPI call
@@ -71,13 +75,14 @@ val disable_clustering : __context:Context.t -> unit
     and logs its actions. *)
 
 val resync_host : __context:Context.t -> host:API.ref_host -> unit
-(** [resync_host ~__context ~host] checks for any clusters on the host. If one
-    exists but is not associated with a cluster_host, it creates one. If the
-    database indicates the cluster_host is enabled, host_resync enables it
-    in the Client layer too. Otherwise, nothing happens. *)
+(** [resync_host ~__context ~host] checks for the existence of a cluster_host.
+    If one exists but hasn't joined the cluster, xapi asks xapi-clusterd to add
+    the host to the cluster, otherwise it enables the cluster host.
+    If no cluster_host is found, nothing happens.
+    If a failure occurs, Xapi sends an alert to XenCenter *)
 
 val forget : __context:Context.t -> self:API.ref_Cluster_host -> unit
 (** [forget ~__context ~self] marks the cluster host as permanently removed
- *  from the cluster. This will only succeed if the rest of the hosts are online,
- *  so in the case of failure the cluster's pending_forget list will be updated.
- *  If you declare all your dead hosts as dead one by one the last one should succeed *)
+    from the cluster. This will only succeed if the rest of the hosts are online,
+    so in the case of failure the cluster's pending_forget list will be updated.
+    If you declare all your dead hosts as dead one by one the last one should succeed *)
diff --git a/ocaml/xapi/xapi_cluster_host_helpers.ml b/ocaml/xapi/xapi_cluster_host_helpers.ml
index 79f9205525f..adb5c000994 100644
--- a/ocaml/xapi/xapi_cluster_host_helpers.ml
+++ b/ocaml/xapi/xapi_cluster_host_helpers.ml
@@ -40,18 +40,29 @@ let report_concurrent_operations_error ~current_ops ~ref_str =
 let get_operation_error ~__context ~self ~op =
   let chr = Db.Cluster_host.get_record_internal ~__context ~self in
   let ref_str = Ref.string_of self in
-  (* let cluster = Db.Cluster_host.get_cluster ~__context ~self in *)
 
   let current_error = None in
 
   let check c f = match c with | Some e -> Some e | None -> f () in
 
+  let assert_joined_cluster ~__context ~self = function
+    | (`enable | `disable) when not (Db.Cluster_host.get_joined ~__context ~self) ->
+      (* Cannot enable nor disable without joining cluster *)
+      Some (Api_errors.cluster_host_not_joined, [ ref_str ])
+    | _ -> None
+  in
+
   (* if other operations are in progress, check that the new operation is allowed concurrently with them *)
   let current_error = check current_error (fun () ->
       let current_ops = chr.Db_actions.cluster_host_current_operations in
-      if current_ops <> [] && not (is_allowed_concurrently ~op ~current_ops)
-      then report_concurrent_operations_error ~current_ops ~ref_str
-      else None) in
+      match current_ops with
+      | _::_ when not (is_allowed_concurrently ~op ~current_ops) ->
+        report_concurrent_operations_error ~current_ops ~ref_str
+      | _ ->
+        check
+          (assert_joined_cluster ~__context ~self op)
+          (fun () -> None) (* replace this function if adding new checks *)
+  ) in
 
   current_error
 
diff --git a/ocaml/xapi/xapi_clustering.ml b/ocaml/xapi/xapi_clustering.ml
index e567e7dad28..4677837bb6e 100644
--- a/ocaml/xapi/xapi_clustering.ml
+++ b/ocaml/xapi/xapi_clustering.ml
@@ -76,9 +76,9 @@ let assert_pif_prerequisites pif =
   ignore (ip_of_pif pif);
   debug "Got IP %s for PIF %s" record.API.pIF_IP (Ref.string_of pif_ref)
 
-let assert_pif_attached_to ~__context ~host ~pif =
-  if not (List.mem pif (Db.Host.get_PIFs ~__context ~self:host)) then
-    raise Api_errors.(Server_error (pif_not_attached_to_host, [Ref.string_of pif; Ref.string_of host]))
+let assert_pif_attached_to ~__context ~host ~pIF =
+  if not (List.mem pIF (Db.Host.get_PIFs ~__context ~self:host)) then
+    raise Api_errors.(Server_error (pif_not_attached_to_host, [Ref.string_of pIF; Ref.string_of host]))
 
 let handle_error = function
   | InternalError message -> raise Api_errors.(Server_error (internal_error, [ message ]))
@@ -127,15 +127,13 @@ let find_cluster_host ~__context ~host =
     raise Api_errors.(Server_error(internal_error, [msg; (Ref.string_of host)]))
   | _ -> None
 
+let get_master_pif ~__context =
+  match find_cluster_host ~__context ~host:Helpers.(get_master ~__context) with
+  | Some self -> Db.Cluster_host.get_PIF ~__context ~self
+  | None -> raise Api_errors.(Server_error (internal_error, [ "No cluster_host exists on master" ]))
+
 let get_network_internal ~__context ~self =
-  let cluster_host = match find_cluster_host ~__context ~host:Helpers.(get_master ~__context) with
-  | Some cluster_host -> cluster_host
-  | None ->
-    raise Api_errors.(Server_error (internal_error,
-    [ Printf.sprintf "No cluster_host master found for cluster %s" (Ref.string_of self) ]))
-  in
-  let master_pif = Db.Cluster_host.get_PIF ~__context ~self:cluster_host in
-  let cluster_network = Db.PIF.get_network ~__context ~self:master_pif in
+  let cluster_network = Db.PIF.get_network ~__context ~self:(get_master_pif ~__context) in
   if List.exists
     (fun cluster_host ->
       let pif = Db.Cluster_host.get_PIF ~__context ~self:cluster_host in
@@ -233,7 +231,7 @@ let rpc ~__context =
   match Context.get_test_clusterd_rpc __context with
   | Some rpc -> rpc
   | None ->
-     Cluster_client.rpc (fun () -> failwith "Can only communicate with xapi-clusterd through message-switch")
+    Cluster_client.rpc (fun () -> failwith "Can only communicate with xapi-clusterd through message-switch")
 
 let is_clustering_disabled_on_host ~__context host =
   match find_cluster_host ~__context ~host with
@@ -243,6 +241,6 @@ let is_clustering_disabled_on_host ~__context host =
 let compute_corosync_max_host_failures ~__context =
   let all_hosts = Db.Host.get_all ~__context in
   let nhosts = List.length (all_hosts) in
-  let disabled_hosts = List.length (List.filter (fun host -> is_clustering_disabled_on_host ~__context host = true ) all_hosts) in
+  let disabled_hosts = List.length (List.filter (fun host -> is_clustering_disabled_on_host ~__context host) all_hosts) in
   let corosync_ha_max_hosts = ((nhosts - disabled_hosts  - 1) / 2) + disabled_hosts in
   corosync_ha_max_hosts
diff --git a/ocaml/xapi/xapi_mgmt_iface.ml b/ocaml/xapi/xapi_mgmt_iface.ml
index 57fc6f38b0c..7f12cdfd7cd 100644
--- a/ocaml/xapi/xapi_mgmt_iface.ml
+++ b/ocaml/xapi/xapi_mgmt_iface.ml
@@ -157,20 +157,35 @@ let enable_himn ~__context ~addr =
 let rebind ~__context =
   run ~__context ~mgmt_enabled:!listening_all
 
-let management_ip_mutex = Mutex.create ()
-let management_ip_cond = Condition.create ()
+let ip_mutex = Mutex.create ()
+let ip_cond = Condition.create ()
 
 let wait_for_management_ip ~__context =
   let ip = ref (match Helpers.get_management_ip_addr ~__context with Some x -> x | None -> "") in
   let is_connected = ref (Helpers.get_management_iface_is_connected ~__context) in
-  Mutex.execute management_ip_mutex
-    (fun () -> begin while !ip = "" && !is_connected = false do
-          Condition.wait management_ip_cond management_ip_mutex;
+  Mutex.execute ip_mutex
+    (fun () -> while !ip = "" && !is_connected = false do
+          Condition.wait ip_cond ip_mutex;
           ip := (match Helpers.get_management_ip_addr ~__context with Some x -> x | None -> "");
           is_connected := (Helpers.get_management_iface_is_connected ~__context)
-        done; end);
+        done);
   !ip
 
+let has_carrier ~__context ~self =
+  let metrics = Db.PIF.get_metrics ~__context ~self in
+  Db.PIF_metrics.get_carrier ~__context ~self:metrics
+
+(* CA-280237: Called in startup sequence after creating cluster_hosts *)
+let wait_for_clustering_ip ~__context ~(self : API.ref_Cluster_host) =
+  let pIF = Db.Cluster_host.get_PIF ~__context ~self in
+  let iP = ref (Db.PIF.get_IP ~__context ~self:pIF) in
+  Mutex.execute ip_mutex (* Don't return until PIF is plugged AND has a valid IP *)
+    (fun () -> while !iP="" || not (has_carrier ~__context ~self:pIF) do
+        Condition.wait ip_cond ip_mutex;
+        iP := Db.PIF.get_IP ~__context ~self:pIF
+      done);
+  !iP
+
 let on_dom0_networking_change ~__context =
   debug "Checking to see if hostname or management IP has changed";
   (* Need to update:
@@ -207,6 +222,6 @@ let on_dom0_networking_change ~__context =
   end;
   Helpers.update_domain_zero_name ~__context localhost new_hostname;
   debug "Signalling anyone waiting for the management IP address to change";
-  Mutex.execute management_ip_mutex
-    (fun () -> Condition.broadcast management_ip_cond)
+  Mutex.execute ip_mutex
+    (fun () -> Condition.broadcast ip_cond)
 
diff --git a/ocaml/xapi/xapi_mgmt_iface.mli b/ocaml/xapi/xapi_mgmt_iface.mli
index 65ea31c2556..18a34925a85 100644
--- a/ocaml/xapi/xapi_mgmt_iface.mli
+++ b/ocaml/xapi/xapi_mgmt_iface.mli
@@ -21,6 +21,9 @@ val himn_addr : string option ref
 (** Block until an IP address appears on the management interface *)
 val wait_for_management_ip : __context:Context.t -> string
 
+(** Block until an IP address appears on the given cluster host PIF *)
+val wait_for_clustering_ip : __context:Context.t -> self:API.ref_Cluster_host -> string
+
 (** Called anywhere we suspect dom0's networking (hostname, IP address) has been changed
     underneath us (eg by dhclient) *)
 val on_dom0_networking_change : __context:Context.t -> unit
diff --git a/ocaml/xapi/xapi_pbd.ml b/ocaml/xapi/xapi_pbd.ml
index e22ea92799d..03621f296af 100644
--- a/ocaml/xapi/xapi_pbd.ml
+++ b/ocaml/xapi/xapi_pbd.ml
@@ -112,18 +112,6 @@ let check_sharing_constraint ~__context ~sr =
                                         [ Ref.string_of sr; Ref.string_of (Db.PBD.get_host ~__context ~self:(List.hd others)) ]))
   end
 
-(** If the SR requires some cluster stacks, we resync every compatible Cluster *)
-let resync_cluster_stack_for_sr_type ~__context ~sr_sm_type ~host =
-  let required_cluster_stacks = Xapi_clustering.get_required_cluster_stacks ~__context ~sr_sm_type in
-  (* This is empty if the SR requires no cluster stack *)
-  match (Xapi_clustering.find_cluster_host ~__context ~host) with
-  | None -> ()
-  | Some cluster_host ->
-  (* check cluster_host associated with both the host and a cluster with a matching cluster_stack *)
-    let self  = Db.Cluster_host.get_cluster ~__context ~self:cluster_host in
-    if List.mem (Db.Cluster.get_cluster_stack ~__context ~self) required_cluster_stacks
-    then Xapi_cluster_host.resync_host ~__context ~host
-
 module C = Storage_interface.Client(struct let rpc = Storage_access.rpc end)
 
 let plug ~__context ~self =
@@ -139,7 +127,6 @@ let plug ~__context ~self =
     (* This must NOT be done while holding the lock, because the functions that
        eventually get called also grab the clustering lock. We can call this
        unconditionally because the operations it calls should be idempotent. *)
-    log_and_ignore_exn (fun () -> resync_cluster_stack_for_sr_type ~__context ~sr_sm_type ~host);
     Xapi_clustering.with_clustering_lock_if_needed ~__context ~sr_sm_type (fun () ->
         Xapi_clustering.assert_cluster_host_is_enabled_for_matching_sms ~__context ~host ~sr_sm_type;
         check_sharing_constraint ~__context ~sr;