Drive circuit-breaker probes for validator notification streams with a timer

linera-client/src/client_options.rs

@@ -130,7 +130,7 @@ pub struct Options {
     #[arg(
         long = "notification-circuit-breaker-initial-probe-interval-ms",
         default_value = "300000",
-        value_parser = util::parse_millis
+        value_parser = util::parse_millis_nonzero
     )]
     pub notification_circuit_breaker_initial_probe_interval: Duration,
 
@@ -139,7 +139,7 @@ pub struct Options {
     #[arg(
         long = "notification-circuit-breaker-max-probe-interval-ms",
         default_value = "3600000",
-        value_parser = util::parse_millis
+        value_parser = util::parse_millis_nonzero
     )]
     pub notification_circuit_breaker_max_probe_interval: Duration,
 

linera-client/src/util.rs

@@ -23,6 +23,16 @@ pub fn parse_millis(s: &str) -> Result<Duration, ParseIntError> {
     Ok(Duration::from_millis(s.parse()?))
 }
 
+/// Parses a number of milliseconds into a `Duration`, rejecting zero.
+///
+/// For a delay, zero usually means "no throttle". For anything a timer sleeps on it means
+/// "spin as fast as the CPU allows", so the flags that feed one are parsed through here.
+pub fn parse_millis_nonzero(s: &str) -> Result<Duration, anyhow::Error> {
+    let millis = s.parse::<u64>()?;
+    anyhow::ensure!(millis > 0, "must be greater than zero");
+    Ok(Duration::from_millis(millis))
+}
+
 /// Converts a `Duration` to `Option<Duration>`, treating zero as `None`.
 pub fn non_zero_duration(d: Duration) -> Option<Duration> {
     if d.is_zero() {

linera-core/src/client/mod.rs

@@ -59,7 +59,8 @@ use crate::{
     remote_node::RemoteNode,
     updater::{communicate_with_quorum, CommunicateAction, RemoteNodeUpdater},
     worker::{Notification, ProcessableCertificate, Reason, WorkerError, WorkerState},
-    ChainWorkerConfig, ProcessConfirmedBlockMode, CHAIN_INFO_MAX_RECEIVED_LOG_ENTRIES,
+    ChainWorkerConfig, MaybeSendBoxFuture, ProcessConfirmedBlockMode,
+    CHAIN_INFO_MAX_RECEIVED_LOG_ENTRIES,
 };
 
 /// The client for interacting with a single chain.
@@ -388,15 +389,9 @@ pub struct Client<Env: Environment> {
     options: chain_client::Options,
 }
 
-/// Boxed future returned by `receive_sender_certificate`. It is `Send` off the `web`
-/// target (where futures must be `Send`) and `?Send` on `web` (single-threaded, where
-/// the validator node is not `Sync`).
-#[cfg(not(web))]
-type ReceiveSenderCertificateFuture<'a> =
-    std::pin::Pin<Box<dyn Future<Output = Result<(), chain_client::Error>> + Send + 'a>>;
-#[cfg(web)]
-type ReceiveSenderCertificateFuture<'a> =
-    std::pin::Pin<Box<dyn Future<Output = Result<(), chain_client::Error>> + 'a>>;
+/// Boxed future returned by `receive_sender_certificate`. `?Send` on `web` because the
+/// validator node is not `Sync` there.
+type ReceiveSenderCertificateFuture<'a> = MaybeSendBoxFuture<'a, Result<(), chain_client::Error>>;
 
 impl<Env: Environment> Client<Env> {
     /// Creates a new `Client` with a new cache and notifiers.
@@ -761,14 +756,8 @@ impl<Env: Environment> Client<Env> {
         // Download remaining certificates from the remote node using a pipelined
         // sliding window. A background task downloads up to `max_concurrent_batch_downloads`
         // batches concurrently and sends them through a channel for sequential processing.
-        #[cfg(not(web))]
-        type CertificateBatchFuture = std::pin::Pin<
-            Box<dyn Future<Output = Result<Vec<ConfirmedBlockCertificate>, NodeError>> + Send>,
-        >;
-        #[cfg(web)]
-        type CertificateBatchFuture = std::pin::Pin<
-            Box<dyn Future<Output = Result<Vec<ConfirmedBlockCertificate>, NodeError>>>,
-        >;
+        type CertificateBatchFuture =
+            crate::MaybeSendBoxFuture<'static, Result<Vec<ConfirmedBlockCertificate>, NodeError>>;
 
         let max_concurrent = self.options.max_concurrent_batch_downloads;
         let batch_size = self.options.certificate_download_batch_size;

linera-core/src/lib.rs

@@ -29,6 +29,16 @@ mod local_node;
 pub mod node;
 /// Utilities for notifying subscribers about chain events.
 pub mod notifier;
+
+/// A boxed future that is `Send` on native targets and not on web, where there is a
+/// single thread. Keeps the `Send` policy for boxed futures in one place.
+#[cfg(not(web))]
+pub(crate) use futures::future::BoxFuture as MaybeSendBoxFuture;
+/// A boxed future that is `Send` on native targets and not on web, where there is a
+/// single thread. Keeps the `Send` policy for boxed futures in one place.
+#[cfg(web)]
+pub(crate) use futures::future::LocalBoxFuture as MaybeSendBoxFuture;
+
 pub mod proof;
 /// A validator node paired with the validator's public key.
 pub mod remote_node;

linera-core/src/proof/notifications.rs

@@ -129,9 +129,9 @@ pub trait NotificationChannelIsLossy {}
 /// same future calls `Client::synchronize_chain_state_from` against that validator before yielding
 /// the stream, precisely because, in the code's own words, "we may have missed notifications since
 /// the last time we synchronized". Since `update_notification_streams` is re-run on every
-/// `Reason::NewBlock`, and a dropped connection is re-established through the same path, a gap in
-/// the stream is closed by *state synchronization* rather than by recovering the messages that fell
-/// in it.
+/// `Reason::NewBlock`, on the death of any validator stream, and on the circuit breaker's probe
+/// timer, and a dropped connection is re-established through the same path, a gap in the stream is
+/// closed by *state synchronization* rather than by recovering the messages that fell in it.
 ///
 /// *Coalescing.* No handler applies the change it was told about; each brings the chain up to date.
 /// In `linera_client::chain_listener`, `Reason::NewIncomingBundle` and `Reason::NewEvents` both