Nexus BG tasks: Use loaded inventory instead of collected inventory (PR 2/2) (#9149)

jgallagher · web-flow · commit 362c4f76bbfa · 2025-10-07T21:10:43.000-04:00
(This is the second half of #9148.) As of this PR, all (I believe!) the background tasks that want to act on the latest inventory collection now act on the latest inventory collection loaded by the `inventory_loader` bg task. That task runs frequently. Fixes #8882. It's still possible two Nexuses could duel if one is acting on a newer inventory collection, but that's always true. The window for that should now be ~15 seconds, instead of waiting until the next time _our own Nexus_ collects inventory. I did not change other places in Nexus where we're reading inventory outside of background tasks. I think it would probably make sense to audit for those and make them read from this watch channel too? If that sounds right I can file an issue.
diff --git a/nexus/src/app/background/init.rs b/nexus/src/app/background/init.rs
@@ -512,13 +512,14 @@ impl BackgroundTasksInitializer {
             datastore.clone(),
             self.inventory_load_tx,
         );
+        let inventory_load_watcher = inventory_loader.watcher();
         driver.register(TaskDefinition {
             name: "inventory_loader",
             description: "loads the latest inventory collection from the DB",
             period: config.inventory.period_secs_load,
             task_impl: Box::new(inventory_loader),
             opctx: opctx.child(BTreeMap::new()),
-            watchers: vec![Box::new(inventory_collect_watcher.clone())],
+            watchers: vec![Box::new(inventory_collect_watcher)],
             activator: task_inventory_loader,
         });
 
@@ -544,7 +545,7 @@ impl BackgroundTasksInitializer {
         let blueprint_planner = blueprint_planner::BlueprintPlanner::new(
             datastore.clone(),
             reconfigurator_config_watcher.clone(),
-            inventory_collect_watcher.clone(),
+            inventory_load_watcher.clone(),
             rx_blueprint.clone(),
         );
         let rx_planner = blueprint_planner.watcher();
@@ -555,7 +556,7 @@ impl BackgroundTasksInitializer {
             task_impl: Box::new(blueprint_planner),
             opctx: opctx.child(BTreeMap::new()),
             watchers: vec![
-                Box::new(inventory_collect_watcher.clone()),
+                Box::new(inventory_load_watcher.clone()),
                 Box::new(rx_blueprint.clone()),
                 Box::new(reconfigurator_config_watcher),
             ],
@@ -620,13 +621,13 @@ impl BackgroundTasksInitializer {
             task_impl: Box::new(
                 physical_disk_adoption::PhysicalDiskAdoption::new(
                     datastore.clone(),
-                    inventory_collect_watcher.clone(),
+                    inventory_load_watcher.clone(),
                     config.physical_disk_adoption.disable,
                     rack_id,
                 ),
             ),
             opctx: opctx.child(BTreeMap::new()),
-            watchers: vec![Box::new(inventory_collect_watcher.clone())],
+            watchers: vec![Box::new(inventory_load_watcher.clone())],
             activator: task_physical_disk_adoption,
         });
 
@@ -641,10 +642,11 @@ impl BackgroundTasksInitializer {
                 blueprint_rendezvous::BlueprintRendezvous::new(
                     datastore.clone(),
                     rx_blueprint.clone(),
+                    inventory_load_watcher.clone(),
                 ),
             ),
             opctx: opctx.child(BTreeMap::new()),
-            watchers: vec![Box::new(inventory_collect_watcher.clone())],
+            watchers: vec![Box::new(inventory_load_watcher.clone())],
             activator: task_blueprint_rendezvous,
         });
 
@@ -674,6 +676,7 @@ impl BackgroundTasksInitializer {
             task_impl: Box::new(ServiceZoneNatTracker::new(
                 datastore.clone(),
                 resolver.clone(),
+                inventory_load_watcher.clone(),
             )),
             opctx: opctx.child(BTreeMap::new()),
             watchers: vec![],
diff --git a/nexus/src/app/background/tasks/blueprint_planner.rs b/nexus/src/app/background/tasks/blueprint_planner.rs
@@ -19,8 +19,8 @@ use nexus_types::deployment::BlueprintSource;
 use nexus_types::deployment::PlanningReport;
 use nexus_types::deployment::{Blueprint, BlueprintTarget};
 use nexus_types::internal_api::background::BlueprintPlannerStatus;
+use nexus_types::inventory::Collection;
 use omicron_common::api::external::LookupType;
-use omicron_uuid_kinds::CollectionUuid;
 use omicron_uuid_kinds::GenericUuid as _;
 use serde_json::json;
 use slog_error_chain::InlineErrorChain;
@@ -31,7 +31,7 @@ use tokio::sync::watch::{self, Receiver, Sender};
 pub struct BlueprintPlanner {
     datastore: Arc<DataStore>,
     rx_config: Receiver<ReconfiguratorConfigLoaderState>,
-    rx_inventory: Receiver<Option<CollectionUuid>>,
+    rx_inventory: Receiver<Option<Arc<Collection>>>,
     rx_blueprint: Receiver<Option<Arc<(BlueprintTarget, Blueprint)>>>,
     tx_blueprint: Sender<Option<Arc<(BlueprintTarget, Blueprint)>>>,
     blueprint_limit: u64,
@@ -57,7 +57,7 @@ impl BlueprintPlanner {
     pub fn new(
         datastore: Arc<DataStore>,
         rx_config: Receiver<ReconfiguratorConfigLoaderState>,
-        rx_inventory: Receiver<Option<CollectionUuid>>,
+        rx_inventory: Receiver<Option<Arc<Collection>>>,
         rx_blueprint: Receiver<Option<Arc<(BlueprintTarget, Blueprint)>>>,
     ) -> Self {
         let (tx_blueprint, _) = watch::channel(None);
@@ -114,10 +114,12 @@ impl BlueprintPlanner {
         let (target, parent) = &*loaded;
         let parent_blueprint_id = parent.id;
 
-        // Get the inventory most recently seen by the collection
-        // background task. The value is `Copy`, so with the deref
-        // we don't block the channel.
-        let Some(collection_id) = *self.rx_inventory.borrow_and_update() else {
+        // Get the inventory most recently seen by the inventory loader
+        // background task. We clone the Arc to avoid keeping the channel locked
+        // for the rest of our execution.
+        let Some(collection) =
+            self.rx_inventory.borrow_and_update().as_ref().map(Arc::clone)
+        else {
             warn!(
                 &opctx.log,
                 "blueprint planning skipped";
@@ -127,25 +129,6 @@ impl BlueprintPlanner {
                 "no inventory collection available",
             ));
         };
-        let collection = match self
-            .datastore
-            .inventory_collection_read(opctx, collection_id)
-            .await
-        {
-            Ok(collection) => collection,
-            Err(error) => {
-                error!(
-                    &opctx.log,
-                    "can't read inventory collection";
-                    "collection_id" => %collection_id,
-                    "error" => %error,
-                );
-                return BlueprintPlannerStatus::Error(format!(
-                    "can't read inventory collection {}: {}",
-                    collection_id, error
-                ));
-            }
-        };
 
         // Assemble the planning context.
         let input = match PlanningInputFromDb::assemble(
@@ -438,6 +421,7 @@ mod test {
     use crate::app::background::tasks::blueprint_execution::BlueprintExecutor;
     use crate::app::background::tasks::blueprint_load::TargetBlueprintLoader;
     use crate::app::background::tasks::inventory_collection::InventoryCollector;
+    use crate::app::background::tasks::inventory_load::InventoryLoader;
     use crate::app::{background::Activator, quiesce::NexusQuiesceHandle};
     use assert_matches::assert_matches;
     use nexus_inventory::now_db_precision;
@@ -467,10 +451,10 @@ mod test {
 
         // Spin up the blueprint loader background task.
         let (tx_loader, _) = watch::channel(None);
-        let mut loader =
+        let mut bp_loader =
             TargetBlueprintLoader::new(datastore.clone(), tx_loader);
-        let mut rx_loader = loader.watcher();
-        loader.activate(&opctx).await;
+        let mut rx_loader = bp_loader.watcher();
+        bp_loader.activate(&opctx).await;
         let (_initial_target, initial_blueprint) = &*rx_loader
             .borrow_and_update()
             .clone()
@@ -490,9 +474,14 @@ mod test {
             1,
             false,
         );
-        let rx_collector = collector.watcher();
         collector.activate(&opctx).await;
 
+        // Spin up the inventory loader background task.
+        let mut inv_loader =
+            InventoryLoader::new(datastore.clone(), watch::Sender::new(None));
+        let rx_inventory = inv_loader.watcher();
+        inv_loader.activate(&opctx).await;
+
         // Enable the planner
         let (_tx, rx_config_loader) = watch::channel(
             ReconfiguratorConfigLoaderState::Loaded(ReconfiguratorConfigView {
@@ -509,10 +498,9 @@ mod test {
         let mut planner = BlueprintPlanner::new(
             datastore.clone(),
             rx_config_loader,
-            rx_collector,
+            rx_inventory,
             rx_loader.clone(),
         );
-        let _rx_planner = planner.watcher();
 
         // On activation, the planner should run successfully and generate
         // a new target blueprint.
@@ -536,7 +524,7 @@ mod test {
         };
 
         // Load and check the new target blueprint.
-        loader.activate(&opctx).await;
+        bp_loader.activate(&opctx).await;
         let (target, blueprint) = &*rx_loader
             .borrow_and_update()
             .clone()
@@ -571,7 +559,7 @@ mod test {
             .expect("can't enable execution");
 
         // Ping the loader again so it gets the updated target.
-        loader.activate(&opctx).await;
+        bp_loader.activate(&opctx).await;
         let (target, blueprint) = &*rx_loader
             .borrow_and_update()
             .clone()
@@ -584,6 +572,7 @@ mod test {
 
         // Trigger an inventory collection.
         collector.activate(&opctx).await;
+        inv_loader.activate(&opctx).await;
 
         // Execute the plan.
         let (dummy_tx, _dummy_rx) = watch::channel(PendingMgsUpdates::new());
diff --git a/nexus/src/app/background/tasks/blueprint_rendezvous.rs b/nexus/src/app/background/tasks/blueprint_rendezvous.rs
@@ -14,9 +14,9 @@ use nexus_reconfigurator_rendezvous::reconcile_blueprint_rendezvous_tables;
 use nexus_types::{
     deployment::{Blueprint, BlueprintTarget},
     internal_api::background::BlueprintRendezvousStatus,
+    inventory::Collection,
 };
 use serde_json::json;
-use slog_error_chain::InlineErrorChain;
 use std::sync::Arc;
 use tokio::sync::watch;
 
@@ -26,6 +26,7 @@ use tokio::sync::watch;
 pub struct BlueprintRendezvous {
     datastore: Arc<DataStore>,
     rx_blueprint: watch::Receiver<Option<Arc<(BlueprintTarget, Blueprint)>>>,
+    rx_inventory: watch::Receiver<Option<Arc<Collection>>>,
 }
 
 impl BlueprintRendezvous {
@@ -34,8 +35,9 @@ impl BlueprintRendezvous {
         rx_blueprint: watch::Receiver<
             Option<Arc<(BlueprintTarget, Blueprint)>>,
         >,
+        rx_inventory: watch::Receiver<Option<Arc<Collection>>>,
     ) -> Self {
-        Self { datastore, rx_blueprint }
+        Self { datastore, rx_blueprint, rx_inventory }
     }
 
     /// Implementation for `BackgroundTask::activate` for `BlueprintRendezvous`,
@@ -56,27 +58,12 @@ impl BlueprintRendezvous {
             return json!({"error": "no blueprint" });
         };
 
-        // Get the latest inventory collection
-        let maybe_collection = match self
-            .datastore
-            .inventory_get_latest_collection(opctx)
-            .await
-        {
-            Ok(maybe_collection) => maybe_collection,
-            Err(err) => {
-                let err = InlineErrorChain::new(&err);
-                warn!(
-                    &opctx.log, "Blueprint rendezvous: skipped";
-                    "reason" => "failed to read latest inventory collection",
-                    &err,
-                );
-                return json!({ "error":
-                    format!("failed reading inventory collection: {err}"),
-                });
-            }
-        };
-
-        let Some(collection) = maybe_collection else {
+        // Get the inventory most recently seen by the inventory loader
+        // background task. We clone the Arc to avoid keeping the channel locked
+        // for the rest of our execution.
+        let Some(collection) =
+            self.rx_inventory.borrow_and_update().as_ref().map(Arc::clone)
+        else {
             warn!(
                 &opctx.log, "Blueprint rendezvous: skipped";
                 "reason" => "no inventory collection",
diff --git a/nexus/src/app/background/tasks/inventory_load.rs b/nexus/src/app/background/tasks/inventory_load.rs
@@ -54,6 +54,10 @@ impl InventoryLoader {
         Self { datastore, tx }
     }
 
+    pub fn watcher(&self) -> watch::Receiver<Option<Arc<Collection>>> {
+        self.tx.subscribe()
+    }
+
     async fn load_if_needed(&self, opctx: &OpContext) -> InventoryLoadStatus {
         // Set up a logger for this activation that includes metadata about
         // the current target.
diff --git a/nexus/src/app/background/tasks/physical_disk_adoption.rs b/nexus/src/app/background/tasks/physical_disk_adoption.rs
diff --git a/nexus/src/app/background/tasks/sync_service_zone_nat.rs b/nexus/src/app/background/tasks/sync_service_zone_nat.rs
