bootstore: keep NetworkConfig in a watch channel (#10162)

This is a baby step towards addressing #9902:

* Internally, make the bootstore keep its `Option<NetworkConfig>` in a
watch channel. This allows us to remove the `get_network_config()`
method entirely; caller can instead subscribe to the watch channel and
peek at its contents any time.
* Rework the sled-agent "retry forever until success" loop that waits
for a successfully-parsed `RackNetworkConfig` to retry based on
`.changed()` notifications from the channel instead of constantly
polling the bootstore task. (This wasn't really because that was a
problem or anything, just that it had to change because of the previous
bullet.)

Author: jgallagher

bootstore/src/schemes/v0/peer.rs

@@ -22,7 +22,7 @@ use std::net::{SocketAddr, SocketAddrV6};
 use std::time::Duration;
 use thiserror::Error;
 use tokio::net::{TcpListener, TcpStream};
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::{mpsc, oneshot, watch};
 use tokio::time::{Instant, MissedTickBehavior, interval};
 
 #[derive(Debug, Clone)]
@@ -106,15 +106,13 @@ pub enum NodeApiRequest {
         config: NetworkConfig,
         responder: oneshot::Sender<Result<(), NodeRequestError>>,
     },
-
-    /// Retrieve the current network config
-    GetNetworkConfig { responder: oneshot::Sender<Option<NetworkConfig>> },
 }
 
 /// A handle for interacting with a `Node` task
 #[derive(Debug, Clone)]
 pub struct NodeHandle {
     tx: mpsc::Sender<NodeApiRequest>,
+    network_config_rx: watch::Receiver<Option<NetworkConfig>>,
 }
 
 impl NodeHandle {
@@ -223,17 +221,11 @@ impl NodeHandle {
         rx.await?
     }
 
-    /// Retrieve the current network config
-    pub async fn get_network_config(
+    /// Subscribe to the watch channel containing the network config
+    pub fn network_config_subscribe(
         &self,
-    ) -> Result<Option<NetworkConfig>, NodeRequestError> {
-        let (tx, rx) = oneshot::channel();
-        self.tx
-            .send(NodeApiRequest::GetNetworkConfig { responder: tx })
-            .await
-            .map_err(|_| NodeRequestError::Send)?;
-        let res = rx.await?;
-        Ok(res)
+    ) -> watch::Receiver<Option<NetworkConfig>> {
+        self.network_config_rx.clone()
     }
 }
 
@@ -256,7 +248,7 @@ pub struct Status {
 /// via control of an underlying  `Fsm`.
 pub struct Node {
     fsm_ledger_generation: u64,
-    network_config: Option<NetworkConfig>,
+    network_config: watch::Sender<Option<NetworkConfig>>,
     config: Config,
     fsm: Fsm,
     peers: BTreeSet<SocketAddrV6>,
@@ -339,11 +331,13 @@ impl Node {
             config.clone().into(),
         )
         .await;
-        let network_config = NetworkConfig::load(
-            &log,
-            config.network_config_ledger_paths.clone(),
-        )
-        .await;
+        let (network_config, network_config_rx) = watch::channel(
+            NetworkConfig::load(
+                &log,
+                config.network_config_ledger_paths.clone(),
+            )
+            .await,
+        );
 
         (
             Node {
@@ -364,14 +358,14 @@ impl Node {
                 conn_rx,
                 conn_tx,
             },
-            NodeHandle { tx },
+            NodeHandle { tx, network_config_rx },
         )
     }
 
     /// Run the main loop of the peer
     ///
     /// This should be spawned into its own tokio task
-    pub async fn run(&mut self) {
+    pub async fn run(mut self) {
         // select among timer tick/received messages
         let mut interval = interval(self.config.time_per_tick);
         interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
@@ -515,6 +509,7 @@ impl Node {
                     fsm_ledger_generation: self.fsm_ledger_generation,
                     network_config_ledger_generation: self
                         .network_config
+                        .borrow()
                         .as_ref()
                         .map(|c| c.generation),
                     fsm_state: self.fsm.state().name(),
@@ -552,8 +547,11 @@ impl Node {
                 }
             }
             NodeApiRequest::UpdateNetworkConfig { config, responder } => {
-                let current_gen =
-                    self.network_config.as_ref().map_or(0, |c| c.generation);
+                let current_gen = self
+                    .network_config
+                    .borrow()
+                    .as_ref()
+                    .map_or(0, |c| c.generation);
                 info!(
                     self.log,
                     concat!(
@@ -608,7 +606,9 @@ impl Node {
                         },
                     ));
                 } else {
-                    self.network_config = Some(config.clone());
+                    self.network_config.send_modify(|c| {
+                        *c = Some(config.clone());
+                    });
                     NetworkConfig::save(
                         &self.log,
                         self.config.network_config_ledger_paths.clone(),
@@ -622,9 +622,6 @@ impl Node {
                     let _ = responder.send(Ok(()));
                 }
             }
-            NodeApiRequest::GetNetworkConfig { responder } => {
-                let _ = responder.send(self.network_config.clone());
-            }
         }
     }
 
@@ -637,7 +634,7 @@ impl Node {
     ) {
         // We only call this method when there has been an update. Otherwise we
         // have an invariant violation due to programmer error and should panic.
-        let network_config = self.network_config.as_ref().unwrap();
+        let network_config = self.network_config.borrow().clone().unwrap();
         info!(
             self.log,
             "Broadcasting network config with generation {}",
@@ -835,13 +832,10 @@ impl Node {
                     addr,
                     unique_id: accepted_handle.unique_id,
                 };
-                if let Some(network_config) = self.network_config.as_ref() {
-                    self.send_network_config(
-                        network_config.clone(),
-                        &peer_id,
-                        &handle,
-                    )
-                    .await;
+                let maybe_network_config = self.network_config.borrow().clone();
+                if let Some(network_config) = maybe_network_config {
+                    self.send_network_config(network_config, &peer_id, &handle)
+                        .await;
                 }
 
                 self.established_connections.insert(peer_id.clone(), handle);
@@ -864,9 +858,11 @@ impl Node {
                         return;
                     }
 
-                    if let Some(network_config) = self.network_config.as_ref() {
+                    let maybe_network_config =
+                        self.network_config.borrow().clone();
+                    if let Some(network_config) = maybe_network_config {
                         self.send_network_config(
-                            network_config.clone(),
+                            network_config,
                             &peer_id,
                             &handle,
                         )
@@ -941,8 +937,11 @@ impl Node {
                 self.accepted_connections.remove(&addr);
             }
             ConnToMainMsgInner::ReceivedNetworkConfig { from, config } => {
-                let current_gen =
-                    self.network_config.as_ref().map_or(0, |c| c.generation);
+                let current_gen = self
+                    .network_config
+                    .borrow()
+                    .as_ref()
+                    .map_or(0, |c| c.generation);
                 let generation = config.generation;
                 info!(
                     self.log,
@@ -955,7 +954,9 @@ impl Node {
                     current_gen
                 );
                 if generation > current_gen {
-                    self.network_config = Some(config.clone());
+                    self.network_config.send_modify(|c| {
+                        *c = Some(config.clone());
+                    });
                     NetworkConfig::save(
                         &self.log,
                         self.config.network_config_ledger_paths.clone(),
@@ -1082,7 +1083,9 @@ mod tests {
             TestNode { config, log, node_handles: None }
         }
 
-        async fn start_node(&mut self) {
+        async fn start_node(
+            &mut self,
+        ) -> watch::Receiver<Option<NetworkConfig>> {
             // Node must have previously been shutdown (or never started)
             assert!(
                 self.node_handles.is_none(),
@@ -1094,7 +1097,7 @@ mod tests {
             self.config.addr.set_port(0);
 
             // (Re-)create node with existing config and its persistent state (if any)
-            let (mut node, handle) =
+            let (node, handle) =
                 Node::new(self.config.clone(), &self.log).await;
             let jh = tokio::spawn(async move {
                 node.run().await;
@@ -1113,7 +1116,9 @@ mod tests {
                 .port();
             self.config.addr.set_port(port);
 
+            let network_config_rx = handle.network_config_subscribe();
             self.node_handles = Some((handle, jh));
+            network_config_rx
         }
 
         async fn shutdown_node(&mut self) {
@@ -1184,11 +1189,15 @@ mod tests {
         }
 
         /// (Re-)start the given node and update peer addresses for everyone
-        async fn start_node(&mut self, i: usize) {
+        async fn start_node(
+            &mut self,
+            i: usize,
+        ) -> watch::Receiver<Option<NetworkConfig>> {
             let node = &mut self.nodes[i];
-            node.start_node().await;
+            let network_config_rx = node.start_node().await;
             self.addrs.insert(node.config.addr);
             self.load_all_peer_addresses().await;
+            network_config_rx
         }
 
         // Stop the given node and update peer addresses for everyone
@@ -1472,13 +1481,13 @@ mod tests {
     async fn network_config() {
         // Create and start test nodes
         let mut nodes = TestNodes::setup(initial_members());
-        nodes.start_node(0).await;
-        nodes.start_node(1).await;
-        nodes.start_node(2).await;
+        let node0_rx = nodes.start_node(0).await;
+        let mut node1_rx = nodes.start_node(1).await;
+        let mut node2_rx = nodes.start_node(2).await;
 
         // Ensure there is no network config at any of the nodes
-        for node in nodes.iter() {
-            assert_eq!(None, node.get_network_config().await.unwrap());
+        for rx in [&node0_rx, &node1_rx, &node2_rx] {
+            assert_eq!(None, rx.borrow().as_ref());
         }
 
         // Update the network config at node0 and ensure it has taken effect
@@ -1487,10 +1496,7 @@ mod tests {
             blob: b"Some network data".to_vec(),
         };
         nodes[0].update_network_config(network_config.clone()).await.unwrap();
-        assert_eq!(
-            Some(&network_config),
-            nodes[0].get_network_config().await.unwrap().as_ref()
-        );
+        assert_eq!(Some(&network_config), node0_rx.borrow().as_ref(),);
 
         // Poll node1 and node2 until the network config update shows up
         // Timeout after 5 seconds
@@ -1499,19 +1505,25 @@ mod tests {
         let mut node1_done = false;
         let mut node2_done = false;
         while !(node1_done && node2_done) {
-            let timeout = POLL_TIMEOUT.saturating_sub(Instant::now() - start);
+            let timeout = POLL_TIMEOUT.saturating_sub(start.elapsed());
             tokio::select! {
                 _ = sleep(timeout) => {
                     panic!("Network config not replicated");
                 }
-                res = nodes[1].get_network_config(), if !node1_done => {
-                    if res.unwrap().as_ref() == Some(&network_config) {
+                _ = node1_rx.changed(), if !node1_done => {
+                    if node1_rx
+                        .borrow_and_update()
+                        .as_ref() == Some(&network_config)
+                    {
                         node1_done = true;
                         continue;
                     }
                 }
-                res = nodes[2].get_network_config(), if !node2_done => {
-                    if res.unwrap().as_ref() == Some(&network_config) {
+                _ = node2_rx.changed(), if !node2_done => {
+                    if node2_rx
+                        .borrow_and_update()
+                        .as_ref() == Some(&network_config)
+                    {
                         node2_done = true;
                         continue;
                     }
@@ -1526,30 +1538,29 @@ mod tests {
         // Poll the learner to ensure it gets the network config
         // Note that the learner doesn't even need to learn its share
         // for network config replication to work.
+        let mut learner_rx = nodes[LEARNER].network_config_subscribe();
         let start = Instant::now();
-        let mut done = false;
-        while !done {
-            let timeout = POLL_TIMEOUT.saturating_sub(Instant::now() - start);
+        loop {
+            if learner_rx.borrow_and_update().as_ref() == Some(&network_config)
+            {
+                break;
+            }
+            let timeout = POLL_TIMEOUT.saturating_sub(start.elapsed());
             tokio::select! {
                 _ = sleep(timeout) => {
                     panic!("Network config not replicated");
                 }
-                res = nodes[LEARNER].get_network_config() => {
-                    if res.unwrap().as_ref() == Some(&network_config) {
-                        done = true;
-                    }
+                _ = learner_rx.changed() => {
+                    continue;
                 }
             }
         }
 
         // Stop node0, bring it back online and ensure it still sees the config
         // at generation 1
         nodes.shutdown_node(0).await;
-        nodes.start_node(0).await;
-        assert_eq!(
-            Some(&network_config),
-            nodes[0].get_network_config().await.unwrap().as_ref()
-        );
+        let node0_rx = nodes.start_node(0).await;
+        assert_eq!(Some(&network_config), node0_rx.borrow().as_ref(),);
 
         // Stop node0 again, update network config via node1, bring node0 back online,
         // and ensure all nodes see the latest configuration.
@@ -1559,25 +1570,22 @@ mod tests {
             blob: b"Some more network data".to_vec(),
         };
         nodes[1].update_network_config(new_config.clone()).await.unwrap();
-        assert_eq!(
-            Some(&new_config),
-            nodes[1].get_network_config().await.unwrap().as_ref()
-        );
-        nodes.start_node(0).await;
+        assert_eq!(Some(&new_config), node1_rx.borrow().as_ref(),);
+        let node0_rx = nodes.start_node(0).await;
         let start = Instant::now();
         // These should all resolve instantly, so no real need for a select,
         // which is getting tedious.
         // We also want to repeatedly loop until all consistently have the same version
         // to give some assurance that the old version from node0 doesn't replicate
         'outer: loop {
-            if Instant::now() - start > POLL_TIMEOUT {
+            if start.elapsed() > POLL_TIMEOUT {
                 panic!("network config not replicated");
             }
-            for node in nodes.iter() {
-                if node.get_network_config().await.unwrap().as_ref()
-                    != Some(&new_config)
-                {
-                    // We need to try again
+            for rx in [&node0_rx, &node1_rx, &node2_rx] {
+                if rx.borrow().as_ref() != Some(&new_config) {
+                    // We need to try again; sleep to yield back to the runtime
+                    // and give it a chance to propagate changes to us.
+                    tokio::time::sleep(Duration::from_millis(10)).await;
                     continue 'outer;
                 }
             }