diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs index dbed5080663..27e65a2db92 100644 --- 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, ) -> Result { - 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 { +fn get_status( + state: &ServerState, + with_storage_status_details: bool, +) -> Result { // 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) -> Result { 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 index 36f0167d042..27ba423cc5a 100644 --- 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, 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 = 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::::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 index 9326da34ab3..8de048eefa1 100644 --- 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, 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;