RSCBC-142: Only apply cluster configs that are newer than we have

Author: chvck

sdk/couchbase-core/src/agent.rs

@@ -671,11 +671,18 @@ impl Agent {
         let inner = self.inner.clone();
         tokio::spawn(async move {
             loop {
-                match watch_rx.recv().await {
-                    Ok(pc) => {
+                match watch_rx.changed().await {
+                    Ok(_) => {
+                        let pc = {
+                            // apply_config requires an owned ParsedConfig, as it takes ownership of it.
+                            // Doing the clone within a block also means we can release the lock that
+                            // borrow_and_update() takes as soon as possible.
+                            watch_rx.borrow_and_update().clone()
+                        };
                         inner.apply_config(pc).await;
                     }
                     Err(_e) => {
+                        debug!("Config watcher channel closed");
                         return;
                     }
                 }

sdk/couchbase-core/src/configfetcher.rs

@@ -10,6 +10,7 @@ use crate::memdx::hello_feature::HelloFeature;
 use crate::memdx::request::{GetClusterConfigKnownVersion, GetClusterConfigRequest};
 use crate::parsedconfig::ParsedConfig;
 use log::{debug, trace};
+use std::env;
 use std::sync::Arc;
 
 #[derive(Clone)]
@@ -56,7 +57,9 @@ impl<M: KvClientManager> ConfigFetcherMemd<M> {
 
         let config = cbconfig::parse::parse_terse_config(&resp.config, hostname)?;
 
-        trace!("Fetcher fetched new config {:?}", &config);
+        if env::var("RSCBC_DEBUG_CONFIG").is_ok() {
+            trace!("Fetcher fetched new config {:?}", &config);
+        }
 
         Ok(Some(ConfigParser::parse_terse_config(config, hostname)?))
     }

sdk/couchbase-core/src/configmanager.rs

@@ -11,13 +11,13 @@ use std::cmp::Ordering;
 use std::future::Future;
 use std::ops::Deref;
 use std::sync::atomic::AtomicBool;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, MutexGuard};
 use std::time::Duration;
-use tokio::sync::broadcast::{Receiver, Sender};
+use tokio::sync::broadcast::error::RecvError;
 use tokio::sync::{broadcast, watch, Notify};
 
 pub(crate) trait ConfigManager: Sized + Send + Sync {
-    fn watch(&self) -> Receiver<ParsedConfig>;
+    fn watch(&self) -> watch::Receiver<ParsedConfig>;
     fn reconfigure(&self, config: ConfigManagerMemdConfig) -> crate::error::Result<()>;
     fn out_of_band_version(
         &self,
@@ -41,7 +41,7 @@ pub(crate) struct ConfigManagerMemdOptions<M: KvClientManager> {
     pub polling_period: Duration,
     pub first_config: ParsedConfig,
     pub kv_client_manager: Arc<M>,
-    pub on_shutdown_rx: Receiver<()>,
+    pub on_shutdown_rx: broadcast::Receiver<()>,
 }
 
 pub(crate) struct ConfigManagerMemd<M: KvClientManager> {
@@ -55,19 +55,19 @@ pub(crate) struct ConfigManagerMemdInner<M: KvClientManager> {
     out_of_band_notify: Notify,
     performing_out_of_band_fetch: AtomicBool,
 
-    latest_config: Mutex<ParsedConfig>,
+    latest_config: Arc<Mutex<ParsedConfig>>,
     latest_version_tx: watch::Sender<ConfigVersion>,
     bucket: Mutex<Option<ParsedConfigBucket>>,
 
-    on_new_config_txs: Vec<Sender<ParsedConfig>>,
+    on_new_config_tx: watch::Sender<ParsedConfig>,
 
-    on_shutdown_rx: Receiver<()>,
-    watcher_shutdown_tx: Sender<()>,
+    on_shutdown_rx: broadcast::Receiver<()>,
+    watcher_shutdown_tx: broadcast::Sender<()>,
 }
 
 impl<M: KvClientManager + 'static> ConfigManagerMemdInner<M> {
-    pub fn watch(&self) -> Receiver<ParsedConfig> {
-        self.watcher.watch(self.watcher_shutdown_tx.subscribe())
+    pub fn watch(&self) -> watch::Receiver<ParsedConfig> {
+        self.on_new_config_tx.subscribe()
     }
 
     pub fn reconfigure(&self, config: ConfigManagerMemdConfig) -> crate::error::Result<()> {
@@ -117,34 +117,23 @@ impl<M: KvClientManager + 'static> ConfigManagerMemdInner<M> {
                     {
                         Ok(c) => c,
                         Err(e) => {
-                            // TODO: log
-                            dbg!(e);
+                            warn!("Out-of-band fetch from {endpoint} failed: {e}");
                             continue;
                         }
                     };
 
                     if let Some(parsed_config) = parsed_config {
-                        // refresh our known version in case it changed somehow.
-                        let mut latest_config = self.latest_config.lock().unwrap();
-
-                        if Self::can_update_config(&parsed_config, latest_config.deref()) {
-                            let new_latest_version = ConfigVersion {
-                                rev_epoch: parsed_config.rev_epoch,
-                                rev_id: parsed_config.rev_id,
-                            };
-                            *latest_config = parsed_config.clone();
-                            drop(latest_config);
-
-                            if let Err(e) = self.latest_version_tx.send(new_latest_version) {
-                                warn!("Failed to send new config version: {e}");
-                            }
-
+                        if let Some(cfg) = Self::handle_config(
+                            self.latest_config.lock().unwrap(),
+                            parsed_config,
+                            self.latest_version_tx.clone(),
+                        ) {
                             self.performing_out_of_band_fetch
                                 .store(false, std::sync::atomic::Ordering::SeqCst);
                             self.out_of_band_notify.notify_waiters();
 
-                            return Some(parsed_config);
-                        }
+                            return Some(cfg);
+                        };
                     }
                 }
 
@@ -164,8 +153,18 @@ impl<M: KvClientManager + 'static> ConfigManagerMemdInner<M> {
     }
 
     pub fn out_of_band_config(&self, parsed_config: ParsedConfig) -> Option<ParsedConfig> {
-        let mut latest_config = self.latest_config.lock().unwrap();
+        Self::handle_config(
+            self.latest_config.lock().unwrap(),
+            parsed_config,
+            self.latest_version_tx.clone(),
+        )
+    }
 
+    fn handle_config(
+        mut latest_config: MutexGuard<ParsedConfig>,
+        parsed_config: ParsedConfig,
+        latest_version_tx: watch::Sender<ConfigVersion>,
+    ) -> Option<ParsedConfig> {
         if Self::can_update_config(&parsed_config, latest_config.deref()) {
             let new_latest_version = ConfigVersion {
                 rev_epoch: parsed_config.rev_epoch,
@@ -174,8 +173,8 @@ impl<M: KvClientManager + 'static> ConfigManagerMemdInner<M> {
             *latest_config = parsed_config.clone();
             drop(latest_config);
 
-            if let Err(e) = self.latest_version_tx.send(new_latest_version) {
-                warn!("Failed to send new config version: {e}");
+            if let Err(e) = latest_version_tx.send(new_latest_version) {
+                warn!("Failed to update config watcher with latest version: {e}");
             }
 
             return Some(parsed_config);
@@ -184,14 +183,6 @@ impl<M: KvClientManager + 'static> ConfigManagerMemdInner<M> {
         None
     }
 
-    fn send_new_config(&self, parsed_config: ParsedConfig) {
-        for tx in self.on_new_config_txs.iter() {
-            if let Err(e) = tx.send(parsed_config.clone()) {
-                warn!("Failed to send new config: {e}");
-            }
-        }
-    }
-
     fn shutdown(&self) {
         if let Err(e) = self.watcher_shutdown_tx.send(()) {
             debug!("Failed to send shutdown signal to watcher: {e}");
@@ -200,20 +191,56 @@ impl<M: KvClientManager + 'static> ConfigManagerMemdInner<M> {
 
     fn can_update_config(new_config: &ParsedConfig, old_config: &ParsedConfig) -> bool {
         if new_config.bucket != old_config.bucket {
-            debug!("Switching config due to changed bucket type (bucket takeover)");
+            debug!(
+                "Switching config due to changed bucket type (bucket takeover) old: {:?} new: {:?}",
+                old_config.bucket, new_config.bucket
+            );
             return true;
         } else if let Some(cmp) = new_config.partial_cmp(old_config) {
             if cmp == Ordering::Less {
-                debug!("Skipping config due to new config being an older revision")
+                debug!("Skipping config due to new config being an older revision old: rev_epoch={}, rev_id={} new: rev_epoch={}, rev_id={}",
+                    old_config.rev_epoch, old_config.rev_id, new_config.rev_epoch, new_config.rev_id);
             } else if cmp == Ordering::Equal {
-                debug!("Skipping config due to matching revisions")
+                debug!("Skipping config due to matching revisions old: rev_epoch={}, rev_id={} new: rev_epoch={}, rev_id={}",
+                    old_config.rev_epoch, old_config.rev_id, new_config.rev_epoch, new_config.rev_id);
             } else {
                 return true;
             }
         }
 
         false
     }
+
+    pub fn start_watcher(&self, watcher_shutdown_rx: broadcast::Receiver<()>) {
+        let mut rx = self.watcher.watch(watcher_shutdown_rx);
+        let latest_version_tx = self.latest_version_tx.clone();
+        let guard = self.latest_config.clone();
+        let new_config_tx = self.on_new_config_tx.clone();
+
+        tokio::spawn(async move {
+            loop {
+                match rx.recv().await {
+                    Ok(cfg) => {
+                        if let Some(new_cfg) = Self::handle_config(
+                            guard.lock().unwrap(),
+                            cfg,
+                            latest_version_tx.clone(),
+                        ) {
+                            new_config_tx.send_replace(new_cfg);
+                        }
+                    }
+                    Err(e) => {
+                        if e == RecvError::Closed {
+                            debug!("Config watcher channel closed");
+                            return;
+                        } else {
+                            warn!("Config watcher channel error: {e}");
+                        }
+                    }
+                }
+            }
+        });
+    }
 }
 
 impl<M: KvClientManager + 'static> ConfigManagerMemd<M> {
@@ -243,33 +270,34 @@ impl<M: KvClientManager + 'static> ConfigManagerMemd<M> {
         let (watcher_shutdown_tx, watcher_shutdown_rx) = broadcast::channel(1);
         let bucket = opts.first_config.bucket.clone();
 
+        let (on_new_config_tx, _on_new_config_rx) =
+            watch::channel::<ParsedConfig>(opts.first_config.clone());
+
         let inner = Arc::new(ConfigManagerMemdInner {
             fetcher,
             watcher: watcher.clone(),
 
             out_of_band_notify: Notify::new(),
             performing_out_of_band_fetch: AtomicBool::new(false),
 
-            latest_config: Mutex::new(opts.first_config),
+            latest_config: Arc::new(Mutex::new(opts.first_config)),
             latest_version_tx,
             bucket: Mutex::new(bucket),
 
-            on_new_config_txs: Vec::new(),
+            on_new_config_tx,
 
             on_shutdown_rx: opts.on_shutdown_rx,
             watcher_shutdown_tx,
         });
 
-        tokio::spawn(async move {
-            watcher.watch(watcher_shutdown_rx);
-        });
+        inner.start_watcher(watcher_shutdown_rx);
 
         ConfigManagerMemd { inner }
     }
 }
 
 impl<M: KvClientManager + 'static> ConfigManager for ConfigManagerMemd<M> {
-    fn watch(&self) -> Receiver<ParsedConfig> {
+    fn watch(&self) -> watch::Receiver<ParsedConfig> {
         self.inner.watch()
     }
 

sdk/couchbase-core/src/configwatcher.rs

@@ -131,8 +131,6 @@ impl<M: KvClientManager> ConfigWatcherMemdInner<M> {
             {
                 Ok(c) => c,
                 Err(e) => {
-                    dbg!(e);
-
                     select! {
                         _ = on_shutdown_rx.recv() => {
                             return;