Fix race condition in scan / watch setup

crates/nodelib/src/dns.rs

@@ -108,7 +108,7 @@ pub async fn list_aliases(
         "{prefix}/",
         prefix = record_key(etcd_config, from_namespace, from_hostname),
     );
-    let (kvs, _) = etcd::util::list_kvs(etcd_config, key_prefix.clone(), 0).await?;
+    let (kvs, _) = etcd::util::list_kvs(etcd_config, key_prefix.clone(), None).await?;
 
     let mut out = Vec::with_capacity(kvs.len());
     for kv in kvs {

crates/nodelib/src/etcd/util.rs

@@ -16,7 +16,7 @@ use crate::etcd::prefix;
 pub async fn list_kvs(
     config: &etcd::Config,
     key_prefix: String,
-    mut revision: i64,
+    start_revision: Option<i64>,
 ) -> Result<(Vec<KeyValue>, i64), Error> {
     let mut out = Vec::new();
 
@@ -25,6 +25,7 @@ pub async fn list_kvs(
     let range_end = prefix::range_end(&key_prefix);
     let mut key = key_prefix.as_bytes().to_vec();
 
+    let mut revision = start_revision.unwrap_or_default();
     loop {
         let response = kv_client
             .range(Request::new(RangeRequest {
@@ -37,7 +38,10 @@ pub async fn list_kvs(
             }))
             .await?
             .into_inner();
-        revision = response.header.unwrap().revision;
+
+        if revision == 0 {
+            revision = response.header.unwrap().revision;
+        }
 
         for kv in &response.kvs {
             out.push(kv.clone());

crates/nodelib/src/etcd/watcher.rs

@@ -36,17 +36,21 @@ pub async fn setup_watcher<W: Watcher + Send + Sync + 'static>(
     watcher: Arc<RwLock<W>>,
     key_prefixes: Vec<String>,
 ) -> Result<(), Error> {
-    let mut start_revision = 0;
+    if key_prefixes.is_empty() {
+        return Ok(());
+    }
+
+    let mut start_revision = None;
     for key_prefix in &key_prefixes {
         start_revision =
-            scan_initial_state(config, &watcher, key_prefix.clone(), start_revision).await?;
+            Some(scan_initial_state(config, &watcher, key_prefix.clone(), start_revision).await?);
     }
 
     tokio::spawn(watcher_task(
         config.clone(),
         watcher,
         key_prefixes,
-        start_revision,
+        start_revision.unwrap(),
     ));
 
     Ok(())
@@ -59,7 +63,7 @@ async fn scan_initial_state<W: Watcher>(
     config: &Config,
     watcher: &Arc<RwLock<W>>,
     key_prefix: String,
-    revision: i64,
+    revision: Option<i64>,
 ) -> Result<i64, Error> {
     let (kvs, revision) = etcd::util::list_kvs(config, key_prefix, revision).await?;
 

crates/nodelib/src/resources/mod.rs

@@ -51,7 +51,7 @@ pub async fn get(
 /// List all resources of the given type.
 pub async fn list(etcd_config: &etcd::Config, rtype: &str) -> Result<Vec<Resource>, Error> {
     let (kvs, _) =
-        etcd::util::list_kvs(etcd_config, prefix::resource(etcd_config, rtype), 0).await?;
+        etcd::util::list_kvs(etcd_config, prefix::resource(etcd_config, rtype), None).await?;
 
     let mut out = Vec::with_capacity(kvs.len());
     for kv in kvs {

crates/reaperd/src/node_reaper.rs

@@ -130,7 +130,7 @@ async fn get_inbox_for_node(
     node_name: &NodeName,
 ) -> Result<Vec<PodName>, Error> {
     let key_prefix = prefix::worker_inbox(etcd_config, &node_name.0);
-    let (kvs, _) = etcd::util::list_kvs(etcd_config, key_prefix.clone(), 0).await?;
+    let (kvs, _) = etcd::util::list_kvs(etcd_config, key_prefix.clone(), None).await?;
 
     let mut to_reap = Vec::with_capacity(kvs.len());
     for kv in kvs {

integration-tests/test-container-networking.nix

@@ -132,12 +132,11 @@ in
     # start apid and schedulerd on `infra`, workerd on `node1` and `node2`
     infra.systemctl("start thing-doer-apid")
     infra.systemctl("start thing-doer-schedulerd")
-    infra.wait_for_unit("thing-doer-apid")
-    infra.wait_for_unit("thing-doer-schedulerd")
-
-    # TODO: there is a race condition where concurrent workerd / schedulerd startup can lead to schedulerd missing state
     node1.systemctl("start thing-doer-workerd")
     node2.systemctl("start thing-doer-workerd")
+
+    infra.wait_for_unit("thing-doer-apid")
+    infra.wait_for_unit("thing-doer-schedulerd")
     node1.wait_for_unit("thing-doer-workerd")
     node2.wait_for_unit("thing-doer-workerd")