fix(cluster): prevent spurious heartbeat elections (#4194)
diff --git a/core/configs/src/server_config/message_bus.rs b/core/configs/src/server_config/message_bus.rs
index b37c17d..8ce3ea5 100644
--- a/core/configs/src/server_config/message_bus.rs
+++ b/core/configs/src/server_config/message_bus.rs
@@ -99,7 +99,8 @@
pub client_queue_capacity: usize,
/// Interval between outbound reconnect attempts to peers with
- /// `peer_id > self_id`.
+ /// `peer_id > self_id`. Also bounds each dial, so a peer that drops
+ /// SYNs cannot stall the sweep.
#[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub reconnect_period: IggyDuration,
diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs
index 64e2882..2f2fa03 100644
--- a/core/consensus/src/impls.rs
+++ b/core/consensus/src/impls.rs
@@ -1198,9 +1198,9 @@
timeouts: RefCell<TimeoutManager>,
- /// Monotonic timestamp from the most recent accepted commit heartbeat.
- /// Old/replayed commit messages with a lower timestamp are ignored.
- heartbeat_timestamp: Cell<u64>,
+ /// View and counter of the last emitted or accepted heartbeat. A new
+ /// primary's counter can be lower, so freshness is scoped to its view.
+ heartbeat_watermark: Cell<(u32, u64)>,
/// Time source for [`Self::next_monotonic_timestamp`]; see
/// [`ConsensusClock`].
@@ -1489,7 +1489,7 @@
sent_own_start_view_change: Cell::new(false),
sent_own_do_view_change: Cell::new(false),
timeouts: RefCell::new(TimeoutManager::new(timeout_seed)),
- heartbeat_timestamp: Cell::new(0),
+ heartbeat_watermark: Cell::new((0, 0)),
clock,
}
}
@@ -2837,8 +2837,8 @@
// After view change the new primary may have commit_min < commit_max
// until commit_journal catches up. Send commit_min (what we've
// actually applied) so backups don't advance past us.
- let ts = self.heartbeat_timestamp.get() + 1;
- self.heartbeat_timestamp.set(ts);
+ let ts = self.heartbeat_watermark.get().1 + 1;
+ self.heartbeat_watermark.set((self.view.get(), ts));
vec![VsrAction::SendCommit {
view: self.view.get(),
@@ -3495,8 +3495,8 @@
/// so it doesn't start a spurious view change. Returns `true` if
/// `commit_max` advanced, signalling the caller to run `commit_journal`.
///
- /// Only accepts heartbeats with a strictly newer monotonic timestamp
- /// to prevent old/replayed messages from suppressing view changes.
+ /// Only refreshes liveness for a newer heartbeat within its view, so
+ /// old/replayed messages cannot suppress view changes.
///
/// # Panics
/// If `header.group` does not match this replica's namespace.
@@ -3550,10 +3550,9 @@
return CommitOutcome::Accepted;
}
- // Only accept heartbeats with a strictly newer timestamp to prevent
- // old/replayed commit messages from resetting the timeout.
- if self.heartbeat_timestamp.get() < header.timestamp_monotonic {
- self.heartbeat_timestamp.set(header.timestamp_monotonic);
+ let heartbeat = (header.view, header.timestamp_monotonic);
+ if self.heartbeat_watermark.get() < heartbeat {
+ self.heartbeat_watermark.set(heartbeat);
self.timeouts
.borrow_mut()
.reset(TimeoutKind::NormalHeartbeat);
@@ -4593,6 +4592,7 @@
#[cfg(test)]
pub mod test_bus {
use super::{Command, METADATA_GROUP, Message, StartViewHeader};
+ use iggy_binary_protocol::{CommitHeader, ConsensusHeader};
use message_bus::{BusMessage, MessageBus};
use server_common::MESSAGE_ALIGN;
use server_common::iobuf::Frozen;
@@ -4629,6 +4629,26 @@
msg
}
+ /// A `Commit` heartbeat from `replica` in `view` stamped with `counter`.
+ ///
+ /// # Panics
+ /// Never: the header size fits in `u32`.
+ #[must_use]
+ pub fn make_commit(view: u32, replica: u8, counter: u64) -> Message<CommitHeader> {
+ Message::<CommitHeader>::new(std::mem::size_of::<CommitHeader>()).transmute_header(
+ |_, header: &mut CommitHeader| {
+ header.command = Command::Commit;
+ header.cluster = 1;
+ header.replica = replica;
+ header.view = view;
+ header.group = METADATA_GROUP;
+ header.timestamp_monotonic = counter;
+ header.size = u32::try_from(std::mem::size_of::<CommitHeader>()).unwrap();
+ header.seal();
+ },
+ )
+ }
+
/// A [`MessageBus`] that accepts everything and remembers nothing.
pub struct NoopBus;
@@ -5254,6 +5274,93 @@
);
}
+ #[test]
+ fn given_an_old_primary_when_the_new_primary_heartbeats_should_remain_in_the_new_view() {
+ for enters_election in [false, true] {
+ let consensus =
+ VsrConsensus::new(1, 0, 3, METADATA_GROUP, StageNoopBus, LocalPipeline::new());
+ consensus.init();
+ for _ in 0..TimeoutManager::NORMAL_HEARTBEAT_TICKS {
+ let _ = consensus.tick(PlaneKind::Metadata);
+ }
+ assert_eq!(
+ consensus.heartbeat_watermark.get(),
+ (
+ 0,
+ TimeoutManager::NORMAL_HEARTBEAT_TICKS / TimeoutManager::COMMIT_MESSAGE_TICKS
+ ),
+ "the old primary must have emitted a full timeout's worth of heartbeats"
+ );
+
+ if enters_election {
+ let _ = consensus.start_election(
+ PlaneKind::Metadata,
+ ViewChangeReason::NormalHeartbeatTimeout,
+ );
+ }
+ let start_view = test_bus::make_start_view(1, 0, 0, 1, 0);
+ let _ = consensus.handle_start_view(PlaneKind::Metadata, start_view.header(), &[]);
+
+ for tick in 1..=TimeoutManager::NORMAL_HEARTBEAT_TICKS * 2 {
+ if tick % TimeoutManager::COMMIT_MESSAGE_TICKS == 0 {
+ let heartbeat =
+ test_bus::make_commit(1, 1, tick / TimeoutManager::COMMIT_MESSAGE_TICKS);
+ consensus.handle_commit(heartbeat.header());
+ }
+ let _ = consensus.tick(PlaneKind::Metadata);
+ assert_eq!(
+ consensus.view(),
+ 1,
+ "fresh heartbeats must keep the adopted view alive at tick {tick}, \
+ enters_election={enters_election}"
+ );
+ }
+
+ for _ in 0..TimeoutManager::NORMAL_HEARTBEAT_TICKS {
+ let _ = consensus.tick(PlaneKind::Metadata);
+ }
+ assert_eq!(
+ consensus.view(),
+ 2,
+ "losing the new primary must still trigger an election"
+ );
+ }
+ }
+
+ #[test]
+ fn given_an_adopted_view_when_heartbeats_are_stale_or_foreign_should_time_out() {
+ let consensus =
+ VsrConsensus::new(1, 0, 3, METADATA_GROUP, StageNoopBus, LocalPipeline::new());
+ consensus.init();
+ let start_view = test_bus::make_start_view(1, 0, 0, 1, 0);
+ let _ = consensus.handle_start_view(PlaneKind::Metadata, start_view.header(), &[]);
+ let heartbeat = test_bus::make_commit(1, 1, 1);
+ consensus.handle_commit(heartbeat.header());
+ // Re-adopting the same view restarts the timer but must keep the
+ // watermark; the replayed heartbeat below proves it.
+ let _ = consensus.handle_start_view(PlaneKind::Metadata, start_view.header(), &[]);
+
+ let invalid_heartbeats = [
+ heartbeat,
+ test_bus::make_commit(1, 1, 0),
+ test_bus::make_commit(0, 0, u64::MAX),
+ test_bus::make_commit(1, 2, u64::MAX),
+ ];
+ for tick in 1..=TimeoutManager::NORMAL_HEARTBEAT_TICKS {
+ if tick % TimeoutManager::COMMIT_MESSAGE_TICKS == 0 {
+ for heartbeat in &invalid_heartbeats {
+ consensus.handle_commit(heartbeat.header());
+ }
+ }
+ let _ = consensus.tick(PlaneKind::Metadata);
+ }
+ assert_eq!(
+ consensus.view(),
+ 2,
+ "replayed, older-view, and non-primary heartbeats must not suppress an election"
+ );
+ }
+
// `observe_newer_view` is monotone: an older stamp cannot lower it.
#[test]
fn observe_newer_view_keeps_the_max() {
diff --git a/core/message_bus/src/connector.rs b/core/message_bus/src/connector.rs
index e1771f4..b7b4d60 100644
--- a/core/message_bus/src/connector.rs
+++ b/core/message_bus/src/connector.rs
@@ -44,15 +44,6 @@
use std::time::Duration;
use tracing::{debug, info};
-/// Default reconnect sweep period.
-///
-/// Equivalent to `MessageBusConfig::default().reconnect_period`; exposed
-/// as a named const for test / bench ergonomics. Kept in sync with the
-/// `MessageBusConfig::default` impl. Remove once the configs-crate
-/// migration lands and bootstrap always reads the period from
-/// `ServerConfig`.
-pub const DEFAULT_RECONNECT_PERIOD: Duration = Duration::from_secs(5);
-
/// Dial every peer with `peer_id > self_id` once, then launch a periodic
/// sweep in the background. The periodic task handle is tracked on the bus
/// so graceful shutdown can await it.
@@ -64,7 +55,7 @@
on_dialed: DialedReplicaFn,
reconnect_period: Duration,
) {
- connect_all(bus, self_id, &peers, &on_dialed).await;
+ connect_all(bus, self_id, &peers, &on_dialed, reconnect_period).await;
let handler = on_dialed.clone();
let token = bus.token();
@@ -89,6 +80,7 @@
self_id: u8,
peers: &[(u8, SocketAddr)],
on_dialed: &DialedReplicaFn,
+ connect_timeout: Duration,
) {
let dials = peers.iter().filter_map(|&(peer_id, addr)| {
if peer_id <= self_id {
@@ -116,7 +108,7 @@
);
return None;
}
- Some(connect_one(peer_id, addr, on_dialed))
+ Some(connect_one(peer_id, addr, on_dialed, connect_timeout))
});
// Dial concurrently so one unreachable peer's connect latency does not
// stall the rest. The futures share one task, so the `on_dialed`
@@ -136,7 +128,7 @@
token: ShutdownToken,
) {
while token.sleep_or_shutdown(period).await {
- connect_all(bus, self_id, &peers, &on_dialed).await;
+ connect_all(bus, self_id, &peers, &on_dialed, period).await;
}
debug!("replica reconnect periodic task exiting");
}
@@ -146,14 +138,28 @@
/// Connect failures are logged and swallowed; VSR tolerates missing
/// peers and the periodic sweep retries. The handshake (and its
/// `handshake_grace` bound) runs on the owning shard after delegation.
+///
+/// The dial is bounded by `connect_timeout`: the sweep joins every dial,
+/// so a peer that drops SYNs would otherwise hold the whole sweep for the
+/// kernel connect timeout (about two minutes on Linux) and starve the
+/// retries the heartbeat timeout depends on.
#[allow(clippy::future_not_send)]
-async fn connect_one(peer_id: u8, addr: SocketAddr, on_dialed: &DialedReplicaFn) {
- let stream = match TcpStream::connect(addr).await {
- Ok(s) => s,
- Err(e) => {
+async fn connect_one(
+ peer_id: u8,
+ addr: SocketAddr,
+ on_dialed: &DialedReplicaFn,
+ connect_timeout: Duration,
+) {
+ let stream = match compio::time::timeout(connect_timeout, TcpStream::connect(addr)).await {
+ Ok(Ok(s)) => s,
+ Ok(Err(e)) => {
debug!(replica = peer_id, %addr, "connect failed: {e}");
return;
}
+ Err(_) => {
+ debug!(replica = peer_id, %addr, ?connect_timeout, "connect timed out");
+ return;
+ }
};
if let Err(e) = stream.set_nodelay(true) {
debug!(replica = peer_id, %addr, "set_nodelay failed: {e}");
diff --git a/core/message_bus/tests/backpressure.rs b/core/message_bus/tests/backpressure.rs
index a329d50..d109c6b 100644
--- a/core/message_bus/tests/backpressure.rs
+++ b/core/message_bus/tests/backpressure.rs
@@ -29,7 +29,7 @@
set_replica_ctx,
};
use iggy_binary_protocol::Command;
-use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
+use message_bus::connector::start as start_connector;
use message_bus::replica::listener::{MessageHandler, bind, run};
use message_bus::{IggyMessageBus, MessageBus, SendError};
use std::rc::Rc;
@@ -66,7 +66,14 @@
// bus0 dials bus1.
let on_message0: MessageHandler = Rc::new(|_, _| {});
let dial_0 = install_dialed_replicas_locally(bus0.clone(), on_message0);
- start_connector(&bus0, 0, vec![(1, addr1)], dial_0, DEFAULT_RECONNECT_PERIOD).await;
+ start_connector(
+ &bus0,
+ 0,
+ vec![(1, addr1)],
+ dial_0,
+ bus0.config().reconnect_period,
+ )
+ .await;
// Wait for the connection to register.
let deadline = std::time::Instant::now() + Duration::from_secs(2);
diff --git a/core/message_bus/tests/connection_lost_notify.rs b/core/message_bus/tests/connection_lost_notify.rs
index e3203dc..bc430e9 100644
--- a/core/message_bus/tests/connection_lost_notify.rs
+++ b/core/message_bus/tests/connection_lost_notify.rs
@@ -26,7 +26,7 @@
install_dialed_replicas_locally, install_replicas_locally, loopback, set_replica_ctx,
};
use message_bus::IggyMessageBus;
-use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
+use message_bus::connector::start as start_connector;
use message_bus::replica::listener::{MessageHandler, bind, run};
use std::cell::Cell;
use std::rc::Rc;
@@ -70,7 +70,7 @@
0,
vec![(1u8, addr1)],
dial_0,
- DEFAULT_RECONNECT_PERIOD,
+ bus0.config().reconnect_period,
)
.await;
diff --git a/core/message_bus/tests/directional_connection.rs b/core/message_bus/tests/directional_connection.rs
index 4825f0f..a999cf4 100644
--- a/core/message_bus/tests/directional_connection.rs
+++ b/core/message_bus/tests/directional_connection.rs
@@ -26,7 +26,7 @@
install_dialed_replicas_locally, install_replicas_locally, loopback, set_replica_ctx,
};
use message_bus::IggyMessageBus;
-use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
+use message_bus::connector::start as start_connector;
use message_bus::replica::listener::{MessageHandler, bind, run};
use std::rc::Rc;
use std::time::Duration;
@@ -64,9 +64,16 @@
let peers = vec![(0u8, addr0), (1u8, addr1)];
let dial_0 = install_dialed_replicas_locally(bus0.clone(), on_message.clone());
- start_connector(&bus0, 0, peers.clone(), dial_0, DEFAULT_RECONNECT_PERIOD).await;
+ start_connector(
+ &bus0,
+ 0,
+ peers.clone(),
+ dial_0,
+ bus0.config().reconnect_period,
+ )
+ .await;
let dial_1 = install_dialed_replicas_locally(bus1.clone(), on_message.clone());
- start_connector(&bus1, 1, peers, dial_1, DEFAULT_RECONNECT_PERIOD).await;
+ start_connector(&bus1, 1, peers, dial_1, bus1.config().reconnect_period).await;
// Wait for the directional connection to settle.
let deadline = std::time::Instant::now() + Duration::from_secs(2);
diff --git a/core/message_bus/tests/head_of_line.rs b/core/message_bus/tests/head_of_line.rs
index d632f6b..e5bec8b 100644
--- a/core/message_bus/tests/head_of_line.rs
+++ b/core/message_bus/tests/head_of_line.rs
@@ -31,7 +31,7 @@
};
use compio::net::TcpListener;
use iggy_binary_protocol::{Command, HEADER_SIZE};
-use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
+use message_bus::connector::start as start_connector;
use message_bus::replica::listener::{MessageHandler, bind, run};
use message_bus::{IggyMessageBus, MessageBus, SendError};
use std::cell::Cell;
@@ -96,7 +96,7 @@
0,
vec![(1, addr_a), (2, addr_b)],
dial_delegate,
- DEFAULT_RECONNECT_PERIOD,
+ bus0.config().reconnect_period,
)
.await;
diff --git a/core/message_bus/tests/replica_roundtrip.rs b/core/message_bus/tests/replica_roundtrip.rs
index 0b2ed65..3c7b491 100644
--- a/core/message_bus/tests/replica_roundtrip.rs
+++ b/core/message_bus/tests/replica_roundtrip.rs
@@ -28,7 +28,7 @@
set_replica_ctx,
};
use iggy_binary_protocol::Command;
-use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
+use message_bus::connector::start as start_connector;
use message_bus::replica::listener::{MessageHandler, bind, run};
use message_bus::{IggyMessageBus, MessageBus};
use std::rc::Rc;
@@ -70,7 +70,7 @@
0,
vec![(1, addr1)],
dial_delegate_0,
- DEFAULT_RECONNECT_PERIOD,
+ bus0.config().reconnect_period,
)
.await;
diff --git a/core/message_bus/tests/replica_tls_bench.rs b/core/message_bus/tests/replica_tls_bench.rs
index f3f9f30..31ca8fa 100644
--- a/core/message_bus/tests/replica_tls_bench.rs
+++ b/core/message_bus/tests/replica_tls_bench.rs
@@ -38,7 +38,7 @@
self_signed_replica_tls_ctx, set_replica_ctx, set_replica_ctx_with_tls,
};
use iggy_binary_protocol::{Command, GenericHeader};
-use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
+use message_bus::connector::start as start_connector;
use message_bus::replica::listener::{MessageHandler, bind, run};
use message_bus::{IggyMessageBus, MessageBus, SendError};
use server_common::Message;
@@ -108,7 +108,7 @@
0,
vec![(1, addr1)],
dial_delegate_0,
- DEFAULT_RECONNECT_PERIOD,
+ bus0.config().reconnect_period,
)
.await;
wait_until(|| bus0.replicas().contains(1), Duration::from_secs(5)).await;
diff --git a/core/message_bus/tests/replica_tls_mitm.rs b/core/message_bus/tests/replica_tls_mitm.rs
index f3bfc29..ba0714a 100644
--- a/core/message_bus/tests/replica_tls_mitm.rs
+++ b/core/message_bus/tests/replica_tls_mitm.rs
@@ -36,7 +36,7 @@
use compio::net::{TcpListener, TcpStream};
use futures::AsyncReadExt;
use message_bus::IggyMessageBus;
-use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
+use message_bus::connector::start as start_connector;
use message_bus::replica::auth::ReplicaAuth;
use message_bus::replica::handshake::ReplicaTlsCtx;
use message_bus::replica::listener::{MessageHandler, bind, run};
@@ -100,7 +100,7 @@
0,
vec![(1, relay_addr)],
dial_delegate_0,
- DEFAULT_RECONNECT_PERIOD,
+ bus0.config().reconnect_period,
)
.await;
diff --git a/core/message_bus/tests/replica_tls_roundtrip.rs b/core/message_bus/tests/replica_tls_roundtrip.rs
index 8cf1c0b..cb8aece 100644
--- a/core/message_bus/tests/replica_tls_roundtrip.rs
+++ b/core/message_bus/tests/replica_tls_roundtrip.rs
@@ -29,7 +29,7 @@
self_signed_replica_tls_ctx, set_replica_ctx, set_replica_ctx_with_tls,
};
use iggy_binary_protocol::Command;
-use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
+use message_bus::connector::start as start_connector;
use message_bus::replica::auth::ReplicaAuth;
use message_bus::replica::listener::{MessageHandler, bind, run};
use message_bus::{IggyMessageBus, MessageBus};
@@ -89,7 +89,7 @@
0,
vec![(1, addr1)],
dial_delegate_0,
- DEFAULT_RECONNECT_PERIOD,
+ bus0.config().reconnect_period,
)
.await;
@@ -143,7 +143,7 @@
0,
vec![(1, addr1)],
dial_delegate_0,
- DEFAULT_RECONNECT_PERIOD,
+ bus0.config().reconnect_period,
)
.await;
@@ -196,7 +196,7 @@
0,
vec![(1, addr1)],
dial_delegate_0,
- DEFAULT_RECONNECT_PERIOD,
+ bus0.config().reconnect_period,
)
.await;
diff --git a/core/message_bus/tests/shard_zero_gating.rs b/core/message_bus/tests/shard_zero_gating.rs
index 34108d5..fbb703a 100644
--- a/core/message_bus/tests/shard_zero_gating.rs
+++ b/core/message_bus/tests/shard_zero_gating.rs
@@ -28,7 +28,6 @@
};
use iggy_common::IggyError;
use message_bus::client_listener::RequestHandler;
-use message_bus::connector::DEFAULT_RECONNECT_PERIOD;
use message_bus::replica::io::{QuicServerCredentials, start_on_shard_zero};
use message_bus::replica::listener::{MessageHandler, bind, run};
use message_bus::transports::tls::self_signed_for_loopback;
@@ -86,7 +85,7 @@
None,
None,
None,
- DEFAULT_RECONNECT_PERIOD,
+ bus_zero.config().reconnect_period,
)
.await
.expect("start_on_shard_zero must succeed on shard 0");
@@ -149,7 +148,7 @@
None,
None,
None,
- DEFAULT_RECONNECT_PERIOD,
+ bus_one.config().reconnect_period,
)
.await
.expect("start_on_shard_zero must succeed on non-zero shard (no-op)");
@@ -227,7 +226,7 @@
Some(accepted_quic),
Some(accepted_tls),
Some(accepted_wss),
- DEFAULT_RECONNECT_PERIOD,
+ bus_zero.config().reconnect_period,
)
.await
.expect("start_on_shard_zero must succeed");
@@ -297,7 +296,7 @@
None,
Some(accepted_tls),
None,
- DEFAULT_RECONNECT_PERIOD,
+ bus_zero.config().reconnect_period,
)
.await
.expect_err("partial TCP-TLS trio must reject");
@@ -339,7 +338,7 @@
None,
None,
Some(accepted_wss),
- DEFAULT_RECONNECT_PERIOD,
+ bus_zero.config().reconnect_period,
)
.await
.expect_err("partial WSS trio must reject");
diff --git a/core/message_bus/tests/tcp_tls_client_roundtrip.rs b/core/message_bus/tests/tcp_tls_client_roundtrip.rs
index b3154f9..8bfcefa 100644
--- a/core/message_bus/tests/tcp_tls_client_roundtrip.rs
+++ b/core/message_bus/tests/tcp_tls_client_roundtrip.rs
@@ -27,7 +27,6 @@
use iggy_binary_protocol::GenericHeader;
use message_bus::BusMessage;
use message_bus::client_listener::RequestHandler;
-use message_bus::connector::DEFAULT_RECONNECT_PERIOD;
use message_bus::replica::io::start_on_shard_zero;
use message_bus::replica::listener::MessageHandler;
use message_bus::transports::tcp_tls::TcpTlsTransportConn;
@@ -92,7 +91,7 @@
None,
Some(accepted_tls),
None,
- DEFAULT_RECONNECT_PERIOD,
+ bus.config().reconnect_period,
)
.await
.expect("start_on_shard_zero must succeed")
diff --git a/core/message_bus/tests/vectored_batch.rs b/core/message_bus/tests/vectored_batch.rs
index fc692c3..c271297 100644
--- a/core/message_bus/tests/vectored_batch.rs
+++ b/core/message_bus/tests/vectored_batch.rs
@@ -25,7 +25,7 @@
set_replica_ctx,
};
use iggy_binary_protocol::Command;
-use message_bus::connector::{DEFAULT_RECONNECT_PERIOD, start as start_connector};
+use message_bus::connector::start as start_connector;
use message_bus::replica::listener::{MessageHandler, bind, run};
use message_bus::{IggyMessageBus, MessageBus};
use std::cell::Cell;
@@ -61,7 +61,14 @@
set_replica_ctx(&bus0, CLUSTER, 0, 2, None);
let on_reply: MessageHandler = Rc::new(|_, _| {});
let dial_0 = install_dialed_replicas_locally(bus0.clone(), on_reply);
- start_connector(&bus0, 0, vec![(1, addr1)], dial_0, DEFAULT_RECONNECT_PERIOD).await;
+ start_connector(
+ &bus0,
+ 0,
+ vec![(1, addr1)],
+ dial_0,
+ bus0.config().reconnect_period,
+ )
+ .await;
// Wait for connect.
let deadline = std::time::Instant::now() + Duration::from_secs(2);
diff --git a/core/message_bus/tests/wss_client_roundtrip.rs b/core/message_bus/tests/wss_client_roundtrip.rs
index a1e2950..b3d3f78 100644
--- a/core/message_bus/tests/wss_client_roundtrip.rs
+++ b/core/message_bus/tests/wss_client_roundtrip.rs
@@ -27,7 +27,6 @@
use iggy_binary_protocol::GenericHeader;
use message_bus::BusMessage;
use message_bus::client_listener::RequestHandler;
-use message_bus::connector::DEFAULT_RECONNECT_PERIOD;
use message_bus::replica::io::start_on_shard_zero;
use message_bus::replica::listener::MessageHandler;
use message_bus::transports::tls::{install_default_crypto_provider, self_signed_for_loopback};
@@ -92,7 +91,7 @@
None,
None,
Some(accepted_wss),
- DEFAULT_RECONNECT_PERIOD,
+ bus.config().reconnect_period,
)
.await
.expect("start_on_shard_zero must succeed")
diff --git a/core/server/config.toml b/core/server/config.toml
index 2143beb..b32988b 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -1066,7 +1066,8 @@
client_queue_capacity = 256
# Interval between outbound reconnect attempts to peers with peer_id > self_id.
-reconnect_period = "5 s"
+# Also bounds each dial, so a peer that drops SYNs cannot stall the sweep.
+reconnect_period = "1 s"
# Timeout for per-peer close drain (flush writer, tear down reader)
# before force-cancellation.
