feldera · blp · Jul 22, 2026 · Jul 22, 2026 · Jul 22, 2026 · gz
diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs
@@ -1455,7 +1455,7 @@ async fn approve(
 async fn status_handler(
     state: WebData<ServerState>,
 ) -> Result<ExtendedRuntimeStatus, ExtendedRuntimeStatusError> {
-    get_status(&state)
+    get_status(&state, true)
 }
 
 #[allow(clippy::result_large_err)]
@@ -1500,24 +1500,24 @@ fn terminated_status(
 }
 
 #[allow(clippy::result_large_err)]
-fn get_status(state: &ServerState) -> Result<ExtendedRuntimeStatus, ExtendedRuntimeStatusError> {
+fn get_status(
+    state: &ServerState,
+    with_storage_status_details: bool,
+) -> Result<ExtendedRuntimeStatus, ExtendedRuntimeStatusError> {
     // Runtime desired status
     let runtime_desired_status = state.desired_status();
 
     // Storage status details
-    let storage_status_details = match &state.storage {
-        Some(backend) => match Checkpointer::read_checkpoints(&**backend) {
-            Ok(list_checkpoints) => Some(StorageStatusDetails {
-                checkpoints: list_checkpoints.into(),
-            }),
-            Err(e) => {
-                error!(
+    let storage_status_details = if with_storage_status_details
+        && let Some(backend) = &state.storage
+    {
+        Checkpointer::read_checkpoints(&**backend).inspect_err(|e| error!(
                     "Unable to read checkpoints; storage status details are not provided. Error: {e}"
-                );
-                None
-            }
-        },
-        None => None,
+                )).ok().map(|list_checkpoints| StorageStatusDetails {
+                checkpoints: list_checkpoints.into(),
+            })
+    } else {
+        None
     };
 
     // Current status
@@ -2774,10 +2774,10 @@ async fn coordination_activate_handler(
 async fn coordination_status(state: WebData<ServerState>) -> Result<HttpResponse, PipelineError> {
     let stream = unfold((state, None), |(state, prev)| async move {
         let status = match prev {
-            None => get_status(&state),
+            None => get_status(&state, false),
             Some(prev) => loop {
                 let notify = state.desired_status_change.notified();
-                let status = get_status(&state);
+                let status = get_status(&state, false);
                 if status != prev {
                     break status;
                 }

diff --git a/crates/dbsp/src/circuit/checkpointer.rs b/crates/dbsp/src/circuit/checkpointer.rs
@@ -442,7 +442,8 @@ impl Checkpointer {
     pub fn read_checkpoints(
         backend: &dyn StorageBackend,
     ) -> Result<VecDeque<CheckpointMetadata>, Error> {
-        match backend.read_json(&StoragePath::from(CHECKPOINT_FILE_NAME)) {
+        let file_name = StoragePath::from(CHECKPOINT_FILE_NAME);
+        match backend.read_json(&file_name) {
             Ok(checkpoints) => Ok(checkpoints),
             Err(error) if error.kind() == ErrorKind::NotFound => {
                 let mut orphan_uuid_dirs: Vec<String> = Vec::new();
@@ -464,6 +465,11 @@ impl Checkpointer {
                         path: Some(CHECKPOINT_FILE_NAME.to_string()),
                     }));
                 }
+
+                // Write an empty checkpoint file to save the cost of listing
+                // all the files next time.
+                backend.write_json(&file_name, &VecDeque::<CheckpointMetadata>::new())?;
+
                 Ok(VecDeque::new())
             }
             Err(error) => Err(error)?,

diff --git a/crates/dbsp/src/operator/communication/exchange.rs b/crates/dbsp/src/operator/communication/exchange.rs
@@ -458,31 +458,26 @@ impl ExchangeClient {
         remote_workers: Range<usize>,
         channel: ExchangeChannel,
     ) {
+        let description = format!("{remote_address} for exchange with workers {remote_workers:?}");
         loop {
-            match TcpStream::connect(remote_address).await {
-                Ok(stream) => {
-                    if let Err(error) =
-                        Self::run_connection(stream, message_type, &remote_workers, &channel).await
-                    {
-                        warn!("connection to {remote_address} dropped ({error}), waiting to retry")
-                    }
-                }
-                Err(error) => {
-                    info!("connection to {remote_address} failed ({error}), waiting to retry")
-                }
+            let mut n_failures = 0u64;
+            let stream = loop {
+                let error = match TcpStream::connect(remote_address).await {
+                    Ok(stream) => break stream,
+                    Err(error) => error,
+                };
+                info!("connection to {description} failed ({error}), waiting to retry");
+                n_failures += 1;
+                sleep(backoff_time()).await;
+            };
+            if n_failures > 0 {
+                info!("connected to {description} after {n_failures} failures")
             }
-
-            fn sleep_time() -> Duration {
-                #[cfg(test)]
-                {
-                    use rand::Rng;
-                    return Duration::from_micros(rand::thread_rng().gen_range(0..1000));
-                }
-
-                #[cfg(not(test))]
-                Duration::from_millis(1000)
+            if let Err(error) =
+                Self::run_connection(stream, message_type, &remote_workers, &channel).await
+            {
+                warn!("connection to {description} dropped ({error}), waiting to retry")
             }
-            sleep(sleep_time()).await;
         }
     }
 
@@ -1967,6 +1962,17 @@ fn inject_fault(_kind: impl Display) -> bool {
     false
 }
 
+#[cfg(test)]
+fn backoff_time() -> Duration {
+    use rand::Rng;
+    Duration::from_micros(rand::thread_rng().gen_range(0..1000))
+}
+
+#[cfg(not(test))]
+fn backoff_time() -> Duration {
+    Duration::from_millis(1000)
+}
+
 #[cfg(test)]
 mod tests {
     use feldera_storage::tokio::TOKIO;