This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new d0f3cd13 Fix/rust telemetry session defects (#1404)
d0f3cd13 is described below
commit d0f3cd13a1b815a7da6b1cfd02ad6bb4701c6af7
Author: zhaohai <[email protected]>
AuthorDate: Sat Oct 24 04:18:35 2026 +0800
Fix/rust telemetry session defects (#1404)
* fix(rust): stop rebuilding telemetry streams on shadow sessions and keep
the session map bounded
The telemetry stream was rebuilt on throw-away shadow sessions during the
periodic
settings sync. Because a shadow session is dropped as soon as the sync
cycle ends, the
rebuilt stream was dropped with it: the registered session kept its dead
handle, so every
cycle produced a fresh one-off stream (a few milliseconds each) instead of
restoring the
long-lived connection. Combined with a `session_map` that only ever grew,
dead endpoints
accumulated and failed on every settings report cycle, which showed up as
recurring
errors and gaps in message consumption.
Changes, following the Java client as the reference:
* Sync settings on the registered sessions
(`SessionManager::sync_settings`) so a rebuilt
stream actually sticks, mirroring `ClientImpl#syncSettings` which reuses
one
`ClientSession` instance. Sessions whose stream cannot be restored are
evicted, mirroring
`ClientImpl#removeClientSession`.
* Evict a session when its heartbeat RPC fails outright (`evict_session`),
so the next
access rebuilds it against a fresh connection.
* Reap dead and idle sessions periodically
(`clear_dead_and_idle_sessions`), mirroring
`ClientManagerImpl#clearIdleRpcClients`; activity is recorded on every
request and
shared with shadow sessions through an atomic counter.
* Enable HTTP/2 keep-alive on the channel (300s interval, 30s timeout, ping
while idle),
matching `RpcClientImpl`. Without it a rolling deployment left a
half-open connection
that was never detected.
* Fix the receive deadline, which was set to exactly `long_polling_timeout`
while the
server is allowed to hold the response for that long. It now adds the
request timeout on
top, as documented on `ClientOption::set_long_polling_timeout` and as the
Java client
does (`awaitDuration.plus(tolerance)`). This is why every poll failed once
`long_polling_timeout` was set below a few seconds.
* Apply the settings pushed by the server (`PushConsumer::apply_settings`),
which were
previously discarded: long polling timeout, receive batch size and fifo
now follow the
server, mirroring `PushSubscriptionSettings#sync`.
Tests: six new unit tests cover session eviction, idle/dead session
reaping, settings sync
acting on the registered session, eviction of unrecoverable sessions, and
the receive
deadline contract. `cargo test` (95 lib + 3 integration + doc tests) and
`cargo clippy --all-features -- -D warnings` are green.
* fix(rust): propagate session errors instead of panicking when the server
is unreachable
`send_message`, `recall_message`, `receive_message`, `ack_message` and
`change_invisible_duration` acquired their session with
`get_session_with_endpoints(...).await.unwrap()`. Whenever the session
could not be
obtained, the client panicked instead of reporting a failure.
That is reachable whenever the session is not cached yet and the peer is
down, which is
exactly what happens around a rolling deployment: the first call after the
server went
away took the whole process down, and the caller never got a chance to
retry. It also
amplified the consumption gaps seen with the telemetry defects, since a
dead endpoint
turned a transient outage into a crash.
The session error is now returned to the caller with `?`, so callers can
retry or fall
back. Verified against a local RocketMQ 5.5.1 cluster: with the proxy
killed mid-run, the
producer now reports the connection failure and recovers on its own once
the proxy is back,
instead of panicking.
* fix(rust): recover the telemetry stream proactively instead of waiting
for periodic syncs
A rolling deployment or failover of the server used to interrupt
consumption for 42-70 seconds, because the client only noticed the dead
telemetry stream at the next 30s settings sync, kept pushing settings
into the fire-and-forget channel of the dying stream, and never
re-registered when the new server instance answered unary RPCs with
`cannot find client` / `settings is null`.
Mirroring the Java client (`ClientSessionImpl#renewRequestObserver`,
`ClientImpl#recoverTransport`), this introduces an error-driven
recovery path:
* the telemetry stream task notifies the client layer through a
stream-closed channel the moment the server closes the stream;
* the client schedules a reconnect that retries on the registered
session every second (deliberately not exponential: a growing
back-off keeps missing the moment the new instance becomes ready);
* `SessionManager#reconnect_session` re-establishes the stream on the
registered session, keeping it in the map on failure so that an idle
client keeps a driver for recovery; rebuilding the stream also
re-registers the client with the new instance;
* a heartbeat business error now triggers the same reconnect (the
heartbeat interval drops from 30s to 10s, aligned with
`HEART_BEAT_PERIOD`), and a session is only evicted after 2
consecutive transport failures (`HEART_BEAT_FAILURE_THRESHOLD`);
* the long-polling timeout moved behind a shared handle: settings
pushed by the server now take effect on the receive deadline and on
the request body of the simple consumer too, instead of being
discarded (`SimpleSubscriptionSettings#sync`).
Verified against a live cluster: after `kill -9` of the proxy and a
restart, the telemetry stream is re-established within one retry cycle
of the new instance becoming ready; the remaining ~20s of the measured
interruption window is the restarted proxy rebuilding its topic routes
from broker heartbeats, which no client can skip.
* test(rust): tolerate the client recovering faster than the readiness probe
The client can re-establish its stream before the external driver has
detected that the proxy is up and written the marker file, so poll for
the marker briefly instead of failing outright.
* fix(rust): bound the reconnect loop, rebuild evicted sessions, fix MSRV
pins and keep-alive interval
Follow-up on the proactive reconnect, addressing real-cluster findings:
* A session that was evicted or reaped meanwhile made `reconnect_session`
return `Ok(())`, which the reconnect retry loop treated as "stream
re-established" and exited without any stream. A missing session is now
rebuilt from the access endpoints instead (`Ok(bool)` tells the loop
whether the stream was really restored). The lock is released before that
rebuild: `get_or_create_session` takes the same map lock, and holding it
across the rebuild dead-locked the tokio mutex.
* The retry loop is bounded: the delay doubles from 1s up to a 30s cap and
the loop gives up after 8 attempts (~2 minutes), leaving a permanently
offline endpoint to the periodic settings sync, the heartbeat and the next
business traffic instead of hammering it once a second forever. Every
attempt is logged with its index and next delay.
* The periodic topic-route sync drops from 30s to 10s: when a broker
endpoint drifts, the stale-route window (measured 13-29s) is decided by
that timer, not by the telemetry reconnect.
* The HTTP/2 keep-alive interval drops from 300s to 60s: the proxy permits
client pings every 30s by default (`ProxyConfig#
grpcServerPermitKeepAliveTimeMillis = 30000`, overriding the bare
grpc-java
5-minute default), so a dead peer is now detected within ~90s instead of
~330s. The Java client does not configure keep-alive at all, so the
previous "aligned with Java" comment was wrong.
* MSRV 1.74 broke on fresh dependency resolution (as the CI build does every
run): pin `hyper-util <0.1.16` (edition 2024), `powerfmt <0.2.1`
(rustc 1.79) and `tokio <1.47` (rustc 1.85).
Verified on a live cluster: after `kill -9` of the proxy the reconnect loop
logged attempts 1/8 (1s), 2/8 (2s), 3/8 (4s) ... and re-established the
stream on attempt 5; the first successful receive happened 79ms after the
new instance became ready.
* fix(rust): iterate the session map through values() to satisfy clippy 1.99
The current stable clippy promotes `for_kv_map` to a style error, so
`cargo clippy --all-features -- -D warnings` fails on the two session-map
loops that destructure an unused key. Iterating with `values()` /
`values_mut()` expresses the same intent without the discarded binding.
* fix(rust): reconnect the requested peer instead of the access endpoint
`reconnect_session` rebuilt a missing session from `option.access_url()`
while
reporting `Ok(true)` for the peer that was asked for. When the two differ
(broker endpoint drift), the reconnect retry loop logged "telemetry stream
for
peer ... re-established" and exited although that peer had no stream at all.
Session-map keys always come from `Endpoints::endpoint_url`, so the peer url
round-trips back into a valid endpoint and the rebuild can target it
directly.
The new test took no part in the `MTX` serialization that the other
`MockSession::new_context()` tests use, so mockall's process-wide context
was
swapped under concurrently running tests and the suite failed at random
(`MockSession::new: No matching expectation found`). `client_get_session`
had
the same gap. Both now take the lock.
---
rust/Cargo.toml | 9 +-
rust/src/client.rs | 686 ++++++++++++++++++++++++++++++++++++---
rust/src/conf.rs | 15 +
rust/src/push_consumer.rs | 39 +++
rust/src/session.rs | 197 ++++++++++-
rust/src/simple_consumer.rs | 43 ++-
rust/tests/rolling_restart_it.rs | 210 ++++++++++++
7 files changed, 1151 insertions(+), 48 deletions(-)
diff --git a/rust/Cargo.toml b/rust/Cargo.toml
index 14968994..3f1e105d 100644
--- a/rust/Cargo.toml
+++ b/rust/Cargo.toml
@@ -32,7 +32,8 @@ description = "Rust client for Apache RocketMQ"
keywords = ["rocketmq", "api", "client", "sdk", "grpc"]
[dependencies]
-tokio = { version = "1", features = ["full"] }
+# Pin tokio to versions compatible with Rust 1.74; 1.47+ raised the MSRV
beyond it.
+tokio = { version = ">=1.20, <1.47", features = ["full"] }
tokio-rustls = { version = "0.24.0", features = ["default",
"dangerous_configuration"] }
tokio-stream = "0.1.12"
async-trait = "0.1.68"
@@ -71,6 +72,12 @@ tokio-util = { version = "=0.7.10", features = ["rt"] }
indexmap = ">=2.0, <2.8"
# Explicitly pin hashbrown to versions compatible with Rust 1.74
hashbrown = ">=0.12, <0.16"
+# Pin hyper-util (pulled in via tonic -> hyper-timeout) to versions that still
build with
+# Rust 1.74; 0.1.16+ switched to edition 2024 and broke the MSRV build.
+hyper-util = ">=0.1.2, <0.1.16"
+# Pin powerfmt (pulled in via time 0.3 -> deranged/time-macros) to versions
compatible
+# with Rust 1.74; 0.2.1+ requires rustc 1.79.
+powerfmt = ">=0.2.0, <0.2.1"
[build-dependencies]
tonic-build = "0.10.2"
diff --git a/rust/src/client.rs b/rust/src/client.rs
index 8243d75e..93820223 100644
--- a/rust/src/client.rs
+++ b/rust/src/client.rs
@@ -22,7 +22,7 @@ use std::{collections::HashMap, sync::atomic::AtomicUsize,
sync::Arc};
use mockall::automock;
use mockall_double::double;
use once_cell::sync::Lazy;
-use parking_lot::Mutex;
+use parking_lot::{Mutex, RwLock};
use prost_types::Duration;
use tokio::select;
use tokio::sync::{mpsc, oneshot};
@@ -42,9 +42,9 @@ use crate::pb::{
RecallMessageResponse, ReceiveMessageRequest, Resource,
SendMessageRequest, Status,
TelemetryCommand,
};
-use crate::session::RPCClient;
#[double]
use crate::session::Session;
+use crate::session::{RPCClient, SharedLongPollingTimeout};
use crate::util::{handle_receive_message_status, handle_response_status,
select_message_queue};
#[derive(Debug)]
@@ -57,6 +57,10 @@ pub(crate) struct Client {
settings: TelemetryCommand,
telemetry_command_tx: Option<mpsc::Sender<pb::telemetry_command::Command>>,
shutdown_tx: Option<oneshot::Sender<()>>,
+ /// Long-polling timeout, shared with every session so that a
server-pushed settings
+ /// update takes effect on the receive deadline and request body without
recreating
+ /// anything.
+ long_polling_timeout: SharedLongPollingTimeout,
}
#[derive(Debug, Clone)]
@@ -78,6 +82,53 @@ const OPERATION_SEND_MESSAGE: &str = "client.send_message";
const OPERATION_RECEIVE_MESSAGE: &str = "client.receive_message";
const OPERATION_ACK_MESSAGE: &str = "client.ack_message";
+/// Maximum duration a session may stay idle before it is dropped from the
session map.
+///
+/// A session is re-created on demand, so reaping idle ones only costs a
re-connect for
+/// endpoints that are used again. Mirrors
`ClientManagerImpl#RPC_CLIENT_MAX_IDLE_DURATION`
+/// of the Java client.
+const SESSION_MAX_IDLE_DURATION: std::time::Duration =
std::time::Duration::from_secs(30 * 60);
+
+/// Interval of the housekeeping task that drops dead and idle sessions.
+///
+/// It is deliberately shorter than the settings/heartbeat intervals so that a
dead endpoint
+/// is removed before the next report cycle tries to use it again.
+const SESSION_CLEANUP_INTERVAL: std::time::Duration =
std::time::Duration::from_secs(30);
+
+/// Period between heartbeats. Kept in line with the Java client
+/// (`ClientImpl#HEART_BEAT_PERIOD`).
+const HEART_BEAT_INTERVAL: std::time::Duration =
std::time::Duration::from_secs(10);
+
+/// Number of consecutive heartbeat transport failures before the session is
evicted.
+/// Kept in line with the Java client
(`ClientImpl#HEART_BEAT_FAILURE_THRESHOLD`), which
+/// tolerates a transient blip instead of tearing the connection down at the
first failure.
+const HEART_BEAT_FAILURE_THRESHOLD: u32 = 2;
+
+/// Initial delay before the first reconnect attempt after the telemetry
stream was
+/// observed dead. Kept in line with the Java client
+/// (`ClientSessionImpl#REQUEST_OBSERVER_RENEW_BACKOFF_DELAY`).
+const RECONNECT_BACKOFF_DELAY: std::time::Duration =
std::time::Duration::from_secs(1);
+
+/// Upper bound of the delay between reconnect attempts. The delay doubles on
every
+/// failure, so a permanently offline endpoint is probed at most every 30s
instead of being
+/// hammered once a second, while a peer that comes back is still picked up
quickly.
+const RECONNECT_RETRY_MAX_DELAY: std::time::Duration =
std::time::Duration::from_secs(30);
+
+/// Number of reconnect attempts after a stream-close notification before
giving up on the
+/// peer (about two minutes in total with the doubling back-off). After that
the endpoint is
+/// considered gone for now: recovery is left to the periodic settings sync,
the heartbeat
+/// and the next business traffic, so a permanently offline endpoint is not
probed forever.
+const RECONNECT_MAX_ATTEMPTS: u32 = 8;
+
+/// Interval of the periodic topic-route sync.
+///
+/// Deliberately shorter than the 30s the Java client uses: when a broker
endpoint drifts
+/// (failover, migration), the client keeps sending to the stale, dead
endpoints until the
+/// route data is refreshed, and this timer — not the telemetry reconnect —
decides how long
+/// that window lasts. 10s halves-to-thirds the measured 13-29s interruption
of the 30s
+/// variant.
+const ROUTE_SYNC_INTERVAL: std::time::Duration =
std::time::Duration::from_secs(10);
+
#[automock]
impl Client {
pub(crate) fn new(
@@ -87,7 +138,9 @@ impl Client {
let id = Self::generate_client_id();
let endpoints = Endpoints::from_url(option.access_url())
.map_err(|e| e.with_operation(OPERATION_CLIENT_NEW))?;
- let session_manager = SessionManager::new(id.clone(), &option);
+ let long_polling_timeout =
Arc::new(RwLock::new(*option.long_polling_timeout()));
+ let session_manager =
+ SessionManager::new(id.clone(), &option,
Arc::clone(&long_polling_timeout));
let route_manager =
TopicRouteManager::new(option.get_namespace().to_string(),
endpoints.clone());
Ok(Client {
@@ -99,6 +152,7 @@ impl Client {
settings,
telemetry_command_tx: None,
shutdown_tx: None,
+ long_polling_timeout,
})
}
@@ -173,6 +227,7 @@ impl Client {
settings: new_settings,
telemetry_command_tx: None, // Will be set during start()
shutdown_tx: None, // Intentionally None to avoid
duplicate shutdown
+ long_polling_timeout: Arc::clone(&self.long_polling_timeout),
}
}
@@ -189,7 +244,16 @@ impl Client {
let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
self.shutdown_tx = Some(shutdown_tx);
- self.telemetry_command_tx = Some(telemetry_command_tx);
+ self.telemetry_command_tx = Some(telemetry_command_tx.clone());
+
+ // Channel the sessions notify through when the server closes a
telemetry stream.
+ // Registered before the first session is created, so that the stream
task captures
+ // the sender while it is spawned.
+ let (stream_closed_tx, mut stream_closed_rx) =
mpsc::channel::<String>(16);
+ session_manager.set_stream_closed_channel(stream_closed_tx);
+ // Also captured so that reconnect_session can rebuild a session that
was evicted
+ // or reaped meanwhile.
+ session_manager.set_telemetry_command_channel(telemetry_command_tx);
let mut rpc_client = self
.get_session()
@@ -201,11 +265,19 @@ impl Client {
let settings = self.settings.clone();
tokio::spawn(async move {
rpc_client.is_started();
+ let mut heartbeat_interval =
tokio::time::interval(HEART_BEAT_INTERVAL);
let seconds_30 = std::time::Duration::from_secs(30);
- let mut heartbeat_interval = tokio::time::interval(seconds_30);
let mut sync_settings_interval =
tokio::time::interval_at(Instant::now() + seconds_30,
seconds_30);
- let mut sync_route_timer = tokio::time::interval(seconds_30);
+ let mut sync_route_timer =
tokio::time::interval(ROUTE_SYNC_INTERVAL);
+ let mut session_cleanup_timer = tokio::time::interval_at(
+ Instant::now() + SESSION_CLEANUP_INTERVAL,
+ SESSION_CLEANUP_INTERVAL,
+ );
+ // Consecutive heartbeat transport failures per peer; a session is
only evicted
+ // after the failure threshold is reached, so a transient blip
does not tear a
+ // working connection down.
+ let mut heartbeat_failures: HashMap<String, u32> = HashMap::new();
loop {
select! {
_ = heartbeat_interval.tick() => {
@@ -226,6 +298,16 @@ impl Client {
"send heartbeat failed: failed to send
heartbeat rpc: {}",
e
);
+ // The endpoint did not answer at all. Evict
it only after
+ // the failure threshold is reached so that
the next access
+ // rebuilds it against a fresh connection,
instead of
+ // keeping a dead endpoint around forever.
+ let failures =
heartbeat_failures.entry(peer.clone()).or_insert(0);
+ *failures += 1;
+ if *failures >= HEART_BEAT_FAILURE_THRESHOLD {
+ session_manager.evict_session(&peer).await;
+ heartbeat_failures.remove(&peer);
+ }
continue;
}
let result =
@@ -235,27 +317,99 @@ impl Client {
"send heartbeat failed: server return
error: {}",
e
);
+ heartbeat_failures.remove(&peer);
+ // A business error such as `cannot find
client` means the
+ // server no longer knows this client, which
happens when a
+ // failover moved the traffic to another
instance. Rebuild
+ // the telemetry stream: the telemetry RPC
itself
+ // re-registers the client with the new
instance.
+ match session_manager.reconnect_session(&peer,
&settings).await {
+ Ok(_) => {}
+ Err(re) => {
+ error!(
+ "reconnect session after heartbeat
error failed, peer={}, error={}",
+ peer, re
+ );
+ }
+ }
continue;
}
+ heartbeat_failures.remove(&peer);
debug!("send heartbeat to server success,
peer={}",peer);
}
},
_ = sync_settings_interval.tick() => {
- let sessions =
session_manager.get_all_sessions().await;
- if let Err(e) = sessions {
- error!("sync settings failed: failed to get
sessions: {}", e);
- continue;
- }
- for mut session in sessions.unwrap() {
- let peer = session.peer().to_string();
- let result =
session.update_settings(settings.clone()).await;
- if let Err(e) = result {
- error!("sync settings failed: failed to call
rpc: {}", e);
- continue;
+ // Sync on the registered sessions, not on shadow
copies: rebuilding a
+ // stream on a throw-away copy loses it immediately,
which is what left
+ // the client with a new one-off stream on every cycle.
+ session_manager.sync_settings(&settings).await;
+ },
+ _ = session_cleanup_timer.tick() => {
+ // Keep the session map from growing without bound:
drop sessions whose
+ // long-lived connection is gone, or that nobody used
for a long time.
+ session_manager.clear_dead_and_idle_sessions().await;
+ },
+ peer = stream_closed_rx.recv() => {
+ match peer {
+ Some(peer) => {
+ warn!(
+ "telemetry stream for peer {} was closed,
scheduling reconnect",
+ peer
+ );
+ // Retry in the background: the first attempt
follows the
+ // initial back-off delay, then the delay
doubles up to a
+ // cap, so a permanently offline endpoint is
not hammered
+ // once a second. After the attempt budget is
exhausted the
+ // endpoint is left to the periodic settings
sync, the
+ // heartbeat and the next business traffic.
+ let session_manager = session_manager.clone();
+ let settings = settings.clone();
+ tokio::spawn(async move {
+ let mut delay = RECONNECT_BACKOFF_DELAY;
+ for attempt in 1..=RECONNECT_MAX_ATTEMPTS {
+ tokio::time::sleep(delay).await;
+ match session_manager
+ .reconnect_session(&peer,
&settings)
+ .await
+ {
+ Ok(true) => {
+ info!(
+ "telemetry stream for peer
{} re-established after {} attempt(s)",
+ peer, attempt
+ );
+ return;
+ }
+ Ok(false) => {
+ // Nothing could be driven (no
access
+ // endpoints configured);
retrying without
+ // a session to rebuild on is
pointless.
+ warn!(
+ "session for peer {} could
not be rebuilt, it will be recreated on the next access",
+ peer
+ );
+ return;
+ }
+ Err(e) => {
+ error!(
+ "reconnect to peer {}
failed (attempt {}/{}): {}, next attempt in {:?}",
+ peer, attempt,
RECONNECT_MAX_ATTEMPTS, e, delay
+ );
+ delay = (delay *
2).min(RECONNECT_RETRY_MAX_DELAY);
+ }
+ }
+ }
+ error!(
+ "gave up reconnecting to peer {} after
{} attempts; it will be rebuilt by the periodic settings sync or the next
traffic",
+ peer, RECONNECT_MAX_ATTEMPTS
+ );
+ });
+ }
+ None => {
+ // Unreachable while the session manager is
alive: it holds
+ // a sender of this channel.
+ error!("stream closed channel was closed
unexpectedly");
}
- debug!("sync settings success, peer = {}", peer);
}
-
},
_ = sync_route_timer.tick() => {
let result = route_manager.sync_route_data(&mut
rpc_client).await;
@@ -325,6 +479,12 @@ impl Client {
&self.id
}
+ /// Handle of the shared long-polling timeout, so that consumers can apply
a
+ /// server-pushed settings update to it from their telemetry loop.
+ pub(crate) fn long_polling_timeout_handle(&self) ->
SharedLongPollingTimeout {
+ Arc::clone(&self.long_polling_timeout)
+ }
+
/// Get the client type
#[allow(dead_code)]
pub(crate) fn get_client_type(&self) -> ClientType {
@@ -473,11 +633,8 @@ impl Client {
endpoints: &Endpoints,
messages: Vec<Message>,
) -> Result<Vec<SendReceipt>, ClientError> {
- self.send_message_inner(
- self.get_session_with_endpoints(endpoints).await.unwrap(),
- messages,
- )
- .await
+ let rpc_client = self.get_session_with_endpoints(endpoints).await?;
+ self.send_message_inner(rpc_client, messages).await
}
pub(crate) async fn recall_message(
@@ -485,11 +642,8 @@ impl Client {
endpoints: &Endpoints,
request: RecallMessageRequest,
) -> Result<RecallMessageResponse, ClientError> {
- self.recall_message_inner(
- self.get_session_with_endpoints(endpoints).await.unwrap(),
- request,
- )
- .await
+ let rpc_client = self.get_session_with_endpoints(endpoints).await?;
+ self.recall_message_inner(rpc_client, request).await
}
pub(crate) async fn send_message_inner<T: RPCClient + 'static>(
@@ -525,8 +679,9 @@ impl Client {
batch_size: i32,
invisible_duration: Duration,
) -> Result<Vec<Message>, ClientError> {
+ let rpc_client = self.get_session_with_endpoints(endpoints).await?;
self.receive_message_inner(
- self.get_session_with_endpoints(endpoints).await.unwrap(),
+ rpc_client,
message_queue,
expression,
batch_size,
@@ -553,8 +708,10 @@ impl Client {
batch_size,
invisible_duration: Some(invisible_duration),
auto_renew: false,
+ // Read from the shared handle: the server may have pushed a new
value through
+ // the telemetry stream since this client was created.
long_polling_timeout: Some(
-
Duration::try_from(*self.option.long_polling_timeout()).unwrap(),
+ Duration::try_from(*self.long_polling_timeout.read()).unwrap(),
),
attempt_id: None,
};
@@ -596,11 +753,12 @@ impl Client {
&self,
ack_entry: &T,
) -> Result<AckMessageResultEntry, ClientError> {
+ let rpc_client = self
+ .get_session_with_endpoints(ack_entry.endpoints())
+ .await?;
let result = self
.ack_message_inner(
- self.get_session_with_endpoints(ack_entry.endpoints())
- .await
- .unwrap(),
+ rpc_client,
ack_entry.topic(),
vec![pb::AckMessageEntry {
message_id: ack_entry.message_id(),
@@ -639,11 +797,12 @@ impl Client {
ack_entry: &T,
invisible_duration: Duration,
) -> Result<String, ClientError> {
+ let rpc_client = self
+ .get_session_with_endpoints(ack_entry.endpoints())
+ .await?;
let result = self
.change_invisible_duration_inner(
- self.get_session_with_endpoints(ack_entry.endpoints())
- .await
- .unwrap(),
+ rpc_client,
ack_entry.topic(),
ack_entry.receipt_handle(),
invisible_duration,
@@ -867,19 +1026,50 @@ pub(crate) struct SessionManager {
client_id: String,
option: ClientOption,
session_map: tokio::sync::Mutex<HashMap<String, Session>>,
+ /// Long-polling timeout shared with every session, see
[`SharedLongPollingTimeout`].
+ long_polling_timeout: SharedLongPollingTimeout,
+ /// Sender of the channel stream-end notifications are sent through;
handed to every
+ /// session when it is created.
+ stream_closed_tx: Mutex<Option<mpsc::Sender<String>>>,
+ /// Sender of the client's telemetry command channel, captured on start so
that
+ /// `reconnect_session` can rebuild a missing session from scratch.
+ telemetry_command_tx:
Mutex<Option<mpsc::Sender<pb::telemetry_command::Command>>>,
}
#[automock]
impl SessionManager {
- pub(crate) fn new(client_id: String, option: &ClientOption) -> Self {
+ pub(crate) fn new(
+ client_id: String,
+ option: &ClientOption,
+ long_polling_timeout: SharedLongPollingTimeout,
+ ) -> Self {
let session_map = tokio::sync::Mutex::new(HashMap::new());
SessionManager {
client_id,
option: option.clone(),
session_map,
+ long_polling_timeout,
+ stream_closed_tx: Mutex::new(None),
+ telemetry_command_tx: Mutex::new(None),
}
}
+ /// Register the channel telemetry stream-end notifications are sent
through.
+ ///
+ /// Must be called before the first session is created.
+ pub(crate) fn set_stream_closed_channel(&self, tx: mpsc::Sender<String>) {
+ *self.stream_closed_tx.lock() = Some(tx);
+ }
+
+ /// Capture the client's telemetry command channel so that
`reconnect_session` can
+ /// rebuild a missing session from scratch. Called by `Client::start`.
+ pub(crate) fn set_telemetry_command_channel(
+ &self,
+ tx: mpsc::Sender<pb::telemetry_command::Command>,
+ ) {
+ *self.telemetry_command_tx.lock() = Some(tx);
+ }
+
pub(crate) async fn get_or_create_session(
&self,
endpoints: &Endpoints,
@@ -891,7 +1081,16 @@ impl SessionManager {
if session_map.contains_key(&endpoint_url) {
Ok(session_map.get(&endpoint_url).unwrap().shadow_session())
} else {
- let mut session = Session::new(endpoints, self.client_id.clone(),
&self.option).await?;
+ let mut session = Session::new(
+ endpoints,
+ self.client_id.clone(),
+ &self.option,
+ Arc::clone(&self.long_polling_timeout),
+ )
+ .await?;
+ if let Some(tx) = self.stream_closed_tx.lock().clone() {
+ session.set_stream_closed_tx(tx);
+ }
session.start(settings, telemetry_command_tx).await?;
let shadow_session = session.shadow_session();
session_map.insert(endpoint_url.clone(), session);
@@ -901,15 +1100,168 @@ impl SessionManager {
pub(crate) async fn get_all_sessions(&self) -> Result<Vec<Session>,
ClientError> {
let session_map = self.session_map.lock().await;
let mut sessions = Vec::new();
- for (_, session) in session_map.iter() {
+ for session in session_map.values() {
sessions.push(session.shadow_session());
}
Ok(sessions)
}
+ /// Drop the session bound to `endpoint_url`, if it is still the one
registered under
+ /// that key.
+ ///
+ /// A session removed here is torn down and a later call to
+ /// [`Self::get_or_create_session`] rebuilds it from scratch, which is the
only way to
+ /// recover from a server that closed the connection for good.
+ ///
+ /// Mirrors `ClientImpl#removeClientSession` of the Java client.
+ pub(crate) async fn evict_session(&self, endpoint_url: &str) -> bool {
+ let mut session_map = self.session_map.lock().await;
+ match session_map.remove(endpoint_url) {
+ Some(mut session) => {
+ session.shutdown();
+ info!("evicted session for endpoint {}", endpoint_url);
+ true
+ }
+ None => false,
+ }
+ }
+
+ /// Re-establish the telemetry stream of the session bound to
`endpoint_url`.
+ ///
+ /// Rebuilding the stream also re-registers the client with the server
instance that
+ /// answers, which is what recovers from a failover or rolling deployment
where the
+ /// server no longer knows this client (`cannot find client`, `settings is
null`).
+ ///
+ /// The session is taken out of the map for the duration of the RPC so
that other
+ /// callers are not blocked on the network round trip. On failure it is
put back so
+ /// that the reconnect retry loop keeps driving recovery on the same
session even when
+ /// the client is idle.
+ ///
+ /// Returns `Ok(true)` when the stream of `endpoint_url` was
re-established. A session
+ /// that is missing from the map (evicted or reaped meanwhile) is rebuilt
against that
+ /// same endpoint, so an idle client keeps a driver for recovery even when
no traffic
+ /// refers to the endpoint anymore. `Ok(false)` means nothing could be
driven (no
+ /// telemetry command channel captured yet).
+ pub(crate) async fn reconnect_session(
+ &self,
+ endpoint_url: &str,
+ settings: &TelemetryCommand,
+ ) -> Result<bool, ClientError> {
+ // Take the session out of the map and release the lock before any
await: the
+ // rebuild path below calls get_or_create_session, which locks the
same map again
+ // (tokio::sync::Mutex is not reentrant).
+ let session_opt = {
+ let mut session_map = self.session_map.lock().await;
+ session_map.remove(endpoint_url)
+ };
+ let Some(mut session) = session_opt else {
+ // The session was evicted or reaped meanwhile. Rebuild it against
the very
+ // endpoint that was asked for: rebuilding the access session
instead would
+ // leave this peer without a stream while still reporting
`Ok(true)`, which
+ // makes the reconnect retry loop log "re-established" and stop.
Every key of
+ // `session_map` comes from `Endpoints::endpoint_url`, so it
round-trips back
+ // into a valid endpoint here.
+ let telemetry_command_tx =
self.telemetry_command_tx.lock().clone();
+ let Some(telemetry_command_tx) = telemetry_command_tx else {
+ return Ok(false);
+ };
+ let endpoints = Endpoints::from_url(endpoint_url)?;
+ return self
+ .get_or_create_session(&endpoints, settings.clone(),
telemetry_command_tx)
+ .await
+ .map(|_| true);
+ };
+ match session.reconnect(settings.clone()).await {
+ Ok(()) => {
+ self.session_map
+ .lock()
+ .await
+ .insert(endpoint_url.to_string(), session);
+ Ok(true)
+ }
+ Err(e) => {
+ // Keep the session registered so that the reconnect retry
loop keeps
+ // retrying on it; dropping it here would leave an idle client
without any
+ // driver to re-establish the stream until the next traffic
arrives.
+ self.session_map
+ .lock()
+ .await
+ .insert(endpoint_url.to_string(), session);
+ Err(e)
+ }
+ }
+ }
+
+ /// Push the client settings to every session that currently holds a
telemetry stream.
+ ///
+ /// This deliberately operates on the *registered* sessions rather than on
throw-away
+ /// shadow copies: a shadow that rebuilds its telemetry stream would drop
that stream
+ /// again as soon as it goes out of scope, which used to leave the
registered session
+ /// permanently without a stream (one short-lived stream per sync
interval, and a dead
+ /// endpoint that fails on every cycle). Mirrors `ClientImpl#syncSettings`
of the Java
+ /// client, where the same `ClientSession` instance is reused.
+ ///
+ /// Sessions whose stream cannot be restored are evicted so that
`session_map` shrinks
+ /// instead of accumulating dead endpoints.
+ pub(crate) async fn sync_settings(&self, settings: &TelemetryCommand) {
+ let mut session_map = self.session_map.lock().await;
+ let mut broken: Vec<String> = Vec::new();
+ for (endpoint_url, session) in session_map.iter_mut() {
+ match session.update_settings(settings.clone()).await {
+ Ok(()) => {
+ debug!("sync settings success, peer = {}", endpoint_url);
+ }
+ Err(e) => {
+ error!(
+ "sync settings failed, peer = {}, error = {}",
+ endpoint_url, e
+ );
+ broken.push(endpoint_url.clone());
+ }
+ }
+ }
+ for endpoint_url in broken {
+ if let Some(mut session) = session_map.remove(&endpoint_url) {
+ session.shutdown();
+ warn!(
+ "evicted session after settings sync failure, peer = {}",
+ endpoint_url
+ );
+ }
+ }
+ }
+
+ /// Forget the sessions that are no longer usable.
+ ///
+ /// Two kinds of session are dropped, both of which used to accumulate
forever because
+ /// `session_map` only ever grew:
+ ///
+ /// * sessions whose telemetry stream is gone, i.e. the long-lived
connection was lost
+ /// (for example after a rolling deployment of the server);
+ /// * sessions that have been idle for longer than
[`SESSION_MAX_IDLE_DURATION`],
+ /// mirroring `ClientManagerImpl#clearIdleRpcClients` of the Java client.
+ ///
+ /// Every dropped session is torn down so that the next access rebuilds it.
+ pub(crate) async fn clear_dead_and_idle_sessions(&self) {
+ let mut session_map = self.session_map.lock().await;
+ let stale: Vec<String> = session_map
+ .iter()
+ .filter(|(_, session)| {
+ !session.is_healthy() || session.idle_duration() >
SESSION_MAX_IDLE_DURATION
+ })
+ .map(|(url, _)| url.clone())
+ .collect();
+ for url in stale {
+ if let Some(mut session) = session_map.remove(&url) {
+ session.shutdown();
+ info!("evicted stale session for endpoint {}", url);
+ }
+ }
+ }
+
pub(crate) async fn shutdown(&self) {
let mut session_map = self.session_map.lock().await;
- for (_, session) in session_map.iter_mut() {
+ for session in session_map.values_mut() {
session.shutdown();
}
session_map.clear();
@@ -957,9 +1309,21 @@ pub(crate) mod tests {
group: Some("group".to_string()),
..Default::default()
},
+ Arc::new(RwLock::new(std::time::Duration::from_secs(40))),
)
}
+ fn mock_session(started: bool) -> Session {
+ let mut session = MockSession::default();
+ session.expect_is_started().return_const(started);
+ session.expect_is_healthy().return_const(started);
+ session
+ .expect_idle_duration()
+ .return_const(std::time::Duration::ZERO);
+ session.expect_shutdown().returning(|| ());
+ session
+ }
+
fn new_client_for_test() -> Client {
Client {
option: ClientOption {
@@ -973,6 +1337,7 @@ pub(crate) mod tests {
settings: TelemetryCommand::default(),
telemetry_command_tx: None,
shutdown_tx: None,
+ long_polling_timeout:
Arc::new(RwLock::new(std::time::Duration::from_secs(40))),
}
}
@@ -987,6 +1352,7 @@ pub(crate) mod tests {
settings: TelemetryCommand::default(),
telemetry_command_tx: Some(tx),
shutdown_tx: None,
+ long_polling_timeout:
Arc::new(RwLock::new(std::time::Duration::from_secs(40))),
}
}
@@ -1013,9 +1379,10 @@ pub(crate) mod tests {
async fn client_get_session() {
let _m = MTX.lock();
let context = MockSession::new_context();
- context.expect().returning(|_, _, _| {
+ context.expect().returning(|_, _, _, _| {
let mut session = MockSession::default();
session.expect_start().returning(|_, _| Ok(()));
+ session.expect_set_stream_closed_tx().returning(|_| ());
session
.expect_shadow_session()
.returning(|| MockSession::default());
@@ -1466,9 +1833,10 @@ pub(crate) mod tests {
let _m = MTX.lock();
let context = MockSession::new_context();
- context.expect().returning(|_, _, _| {
+ context.expect().returning(|_, _, _, _| {
let mut session = MockSession::default();
session.expect_start().returning(|_, _| Ok(()));
+ session.expect_set_stream_closed_tx().returning(|_| ());
session
.expect_shadow_session()
.returning(|| MockSession::default());
@@ -1502,9 +1870,10 @@ pub(crate) mod tests {
let _m = MTX.lock();
let context = MockSession::new_context();
- context.expect().returning(|_, _, _| {
+ context.expect().returning(|_, _, _, _| {
let mut session = MockSession::default();
session.expect_start().returning(|_, _| Ok(()));
+ session.expect_set_stream_closed_tx().returning(|_| ());
session
.expect_shadow_session()
.returning(|| MockSession::default());
@@ -1548,4 +1917,233 @@ pub(crate) mod tests {
let cloned_client = original_client.clone_for_lite_consumer();
assert!(cloned_client.is_lite_consumer());
}
+
+ #[tokio::test]
+ async fn test_evict_session() {
+ let session_manager = new_session_manager();
+ session_manager
+ .session_map
+ .lock()
+ .await
+ .insert("endpoint_a".to_string(), mock_session(true));
+
+ assert!(session_manager.evict_session("endpoint_a").await);
+ assert!(session_manager.session_map.lock().await.is_empty());
+
+ // Evicting an unknown endpoint is a no-op.
+ assert!(!session_manager.evict_session("endpoint_a").await);
+ }
+
+ #[tokio::test]
+ async fn test_clear_dead_and_idle_sessions() {
+ let session_manager = new_session_manager();
+ {
+ let mut session_map = session_manager.session_map.lock().await;
+ // Healthy session: kept.
+ session_map.insert("healthy".to_string(), mock_session(true));
+ // Lost its telemetry stream: dropped.
+ session_map.insert("dead".to_string(), mock_session(false));
+ }
+
+ session_manager.clear_dead_and_idle_sessions().await;
+
+ let session_map = session_manager.session_map.lock().await;
+ assert_eq!(session_map.len(), 1);
+ assert!(session_map.contains_key("healthy"));
+ }
+
+ #[tokio::test]
+ async fn test_clear_idle_sessions() {
+ let session_manager = new_session_manager();
+ {
+ let mut session_map = session_manager.session_map.lock().await;
+ let mut session = MockSession::default();
+ session.expect_is_started().return_const(true);
+ session.expect_is_healthy().return_const(true);
+ session
+ .expect_idle_duration()
+ .return_const(SESSION_MAX_IDLE_DURATION +
std::time::Duration::from_secs(1));
+ session.expect_shutdown().returning(|| ());
+ session_map.insert("idle".to_string(), session);
+ }
+
+ session_manager.clear_dead_and_idle_sessions().await;
+
+ assert!(session_manager.session_map.lock().await.is_empty());
+ }
+
+ #[tokio::test]
+ async fn test_sync_settings_rebuilds_stream_on_registered_session() {
+ let session_manager = new_session_manager();
+ let rebuilt = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+ {
+ let mut session_map = session_manager.session_map.lock().await;
+ let mut session = MockSession::default();
+ // The stream is gone, so the settings sync must re-establish it...
+ session.expect_is_started().return_const(false);
+ session.expect_is_healthy().return_const(false);
+ let counter = Arc::clone(&rebuilt);
+ session.expect_update_settings().returning(move |_| {
+ counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+ Ok(())
+ });
+ session.expect_shutdown().returning(|| ());
+ session_map.insert("endpoint_a".to_string(), session);
+ }
+
+ session_manager
+ .sync_settings(&TelemetryCommand::default())
+ .await;
+
+ // ... and it must have been rebuilt exactly once, on the registered
session rather
+ // than on a throw-away shadow copy.
+ assert_eq!(
+ rebuilt.load(std::sync::atomic::Ordering::SeqCst),
+ 1,
+ "settings sync should rebuild the stream of the registered session
once"
+ );
+ // A session whose stream was restored successfully stays registered.
+ assert_eq!(session_manager.session_map.lock().await.len(), 1);
+ }
+
+ #[tokio::test]
+ async fn test_sync_settings_evicts_unrecoverable_session() {
+ let session_manager = new_session_manager();
+ {
+ let mut session_map = session_manager.session_map.lock().await;
+ let mut session = MockSession::default();
+ session.expect_is_started().return_const(false);
+ session.expect_is_healthy().return_const(false);
+ session.expect_update_settings().returning(|_| {
+ Err(ClientError::new(
+ ErrorKind::ServerUnavailable,
+ "server unavailable",
+ "test.sync_settings",
+ ))
+ });
+ session.expect_shutdown().returning(|| ());
+ session_map.insert("endpoint_a".to_string(), session);
+ }
+
+ session_manager
+ .sync_settings(&TelemetryCommand::default())
+ .await;
+
+ // A dead endpoint must not be kept around: it would fail again on
every cycle.
+ assert!(session_manager.session_map.lock().await.is_empty());
+ }
+
+ #[tokio::test]
+ async fn test_reconnect_session_rebuilds_stream_on_registered_session() {
+ let session_manager = new_session_manager();
+ let reconnected = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+ {
+ let mut session_map = session_manager.session_map.lock().await;
+ let mut session = MockSession::default();
+ let counter = Arc::clone(&reconnected);
+ session.expect_reconnect().returning(move |_| {
+ counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+ Ok(())
+ });
+ session_map.insert("endpoint_a".to_string(), session);
+ }
+
+ session_manager
+ .reconnect_session("endpoint_a", &TelemetryCommand::default())
+ .await
+ .unwrap();
+
+ // The stream was rebuilt on the registered session, which stayed in
the map.
+ assert_eq!(
+ reconnected.load(std::sync::atomic::Ordering::SeqCst),
+ 1,
+ "reconnect should re-establish the stream of the registered
session"
+ );
+ assert_eq!(session_manager.session_map.lock().await.len(), 1);
+ assert!(
+ session_manager
+ .reconnect_session("endpoint_a", &TelemetryCommand::default())
+ .await
+ .unwrap(),
+ "a re-established stream must be reported as restored"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_reconnect_session_keeps_session_that_cannot_be_restored() {
+ let session_manager = new_session_manager();
+ {
+ let mut session_map = session_manager.session_map.lock().await;
+ let mut session = MockSession::default();
+ session.expect_reconnect().returning(|_| {
+ Err(ClientError::new(
+ ErrorKind::ServerUnavailable,
+ "server unavailable",
+ "test.reconnect_session",
+ ))
+ });
+ session_map.insert("endpoint_a".to_string(), session);
+ }
+
+ assert!(session_manager
+ .reconnect_session("endpoint_a", &TelemetryCommand::default())
+ .await
+ .is_err());
+
+ // The session must stay registered: the reconnect retry loop keeps
driving
+ // recovery on it, which is the only driver an idle client has while
the server is
+ // unavailable.
+ assert_eq!(session_manager.session_map.lock().await.len(), 1);
+ }
+
+ #[tokio::test]
+ async fn test_reconnect_session_rebuilds_missing_session() {
+ let _m = MTX.lock();
+
+ let session_manager = new_session_manager();
+ let (tx, _rx) = mpsc::channel(16);
+ session_manager.set_telemetry_command_channel(tx);
+ let created: Arc<parking_lot::Mutex<Vec<String>>> =
+ Arc::new(parking_lot::Mutex::new(Vec::new()));
+ let context = MockSession::new_context();
+ let recorded = Arc::clone(&created);
+ context.expect().returning(move |endpoints, _, _, _| {
+ recorded.lock().push(endpoints.endpoint_url().to_string());
+ let mut session = MockSession::default();
+ session.expect_start().returning(|_, _| Ok(()));
+ session.expect_set_stream_closed_tx().returning(|_| ());
+ session
+ .expect_shadow_session()
+ .returning(|| MockSession::default());
+ Ok(session)
+ });
+
+ // A session that was evicted or reaped meanwhile must be rebuilt
against the peer
+ // that is being reconnected, not against the access endpoint.
Rebuilding the access
+ // session used to report Ok(true) for a peer that had no stream left,
so the
+ // reconnect retry loop stopped as if it had restored the connection.
The peer here
+ // deliberately differs from the default access_url (localhost:8081).
+ let peer = "127.0.0.1:9999";
+ let restored = session_manager
+ .reconnect_session(peer, &TelemetryCommand::default())
+ .await
+ .unwrap();
+ assert!(
+ restored,
+ "the rebuilt peer stream must be reported as restored"
+ );
+ assert_eq!(created.lock().as_slice(), &[peer.to_string()]);
+ let keys: Vec<String> = session_manager
+ .session_map
+ .lock()
+ .await
+ .keys()
+ .cloned()
+ .collect();
+ assert_eq!(
+ keys,
+ vec![peer.to_string()],
+ "the session must be registered under the reconnected peer"
+ );
+ }
}
diff --git a/rust/src/conf.rs b/rust/src/conf.rs
index 697f41fc..78463575 100644
--- a/rust/src/conf.rs
+++ b/rust/src/conf.rs
@@ -312,6 +312,14 @@ impl PushConsumerOption {
&self.long_polling_timeout
}
+ /// Set the await duration during long polling.
+ ///
+ /// Also used to apply the value pushed by the server, so that a client
configured with
+ /// a short `long_polling_timeout` follows the server after the first
settings sync.
+ pub(crate) fn set_long_polling_timeout(&mut self, long_polling_timeout:
Duration) {
+ self.long_polling_timeout = long_polling_timeout;
+ }
+
pub fn subscription_expressions(&self) -> &HashMap<String,
FilterExpression> {
&self.subscription_expressions
}
@@ -355,6 +363,13 @@ impl PushConsumerOption {
self.batch_size
}
+ /// Set the reception batch size.
+ ///
+ /// Also used to apply the value pushed by the server.
+ pub(crate) fn set_batch_size(&mut self, batch_size: i32) {
+ self.batch_size = batch_size;
+ }
+
pub fn consumer_worker_count_each_queue(&self) -> usize {
self.consumer_worker_count_each_queue
}
diff --git a/rust/src/push_consumer.rs b/rust/src/push_consumer.rs
index d16c8d88..33ed6ee9 100644
--- a/rust/src/push_consumer.rs
+++ b/rust/src/push_consumer.rs
@@ -209,6 +209,12 @@ impl PushConsumer {
select! {
command = telemetry_command_rx.recv() => {
if let Some(command) = command {
+ if let
pb::telemetry_command::Command::Settings(ref settings) = command {
+ // Apply what the server pushed, otherwise
these settings are
+ // silently dropped and the client keeps
polling with the
+ // configured values.
+ Self::apply_settings(settings, &option);
+ }
let remote_backoff_policy =
BackOffRetryPolicy::try_from(command);
if let Ok(remote_backoff_policy) =
remote_backoff_policy {
retry_policy.lock().replace(remote_backoff_policy);
@@ -322,6 +328,9 @@ impl PushConsumer {
select! {
command = command_rx.recv() => {
if let Some(command) = command {
+ if let
pb::telemetry_command::Command::Settings(ref settings) = command {
+ Self::apply_settings(settings, &option);
+ }
let remote_backoff_policy =
BackOffRetryPolicy::try_from(command);
if let Ok(remote_backoff_policy) =
remote_backoff_policy {
retry_policy.lock().replace(remote_backoff_policy);
@@ -397,6 +406,36 @@ impl PushConsumer {
Ok(())
}
+ /// Apply the settings pushed by the server.
+ ///
+ /// Only subscription settings are applied; the back-off policy is handled
separately by
+ /// [`BackOffRetryPolicy::try_from`]. Mirrors
`PushSubscriptionSettings#sync` of the Java
+ /// client, where the same values are also taken from the server-pushed
settings.
+ fn apply_settings(settings: &pb::Settings, option:
&Arc<RwLock<PushConsumerOption>>) {
+ let Some(pb::settings::PubSub::Subscription(ref subscription)) =
settings.pub_sub else {
+ warn!(
+ "received settings that do not match the client type,
pub_sub={:?}",
+ settings.pub_sub
+ );
+ return;
+ };
+
+ let mut option = option.write();
+ if subscription.receive_batch_size.unwrap_or_default() > 0 {
+ option.set_batch_size(subscription.receive_batch_size.unwrap());
+ }
+ if let Some(ref long_polling_timeout) =
subscription.long_polling_timeout {
+ let seconds = long_polling_timeout.seconds;
+ let nanos = long_polling_timeout.nanos;
+ if seconds > 0 || nanos > 0 {
+ option.set_long_polling_timeout(Duration::new(seconds as u64,
nanos as u32));
+ }
+ }
+ if let Some(fifo) = subscription.fifo {
+ option.set_fifo(fifo);
+ }
+ }
+
async fn process_assignments(
rpc_client: &Session,
option: &PushConsumerOption,
diff --git a/rust/src/session.rs b/rust/src/session.rs
index 55e9d4a1..73163859 100644
--- a/rust/src/session.rs
+++ b/rust/src/session.rs
@@ -15,6 +15,9 @@
* limitations under the License.
*/
+use std::sync::Arc;
+use std::time::Instant;
+
use async_trait::async_trait;
use mockall::{automock, mock};
use ring::hmac;
@@ -47,6 +50,14 @@ use tracing::{debug, error, info};
const OPERATION_START: &str = "session.start";
const OPERATION_UPDATE_SETTINGS: &str = "session.update_settings";
+/// The long-polling timeout, shared between the client layer and every
session.
+///
+/// The server can push a new value at any time through the telemetry stream,
so it has to
+/// live behind a lock instead of being baked into the immutable
[`ClientOption`] clone each
+/// session keeps. Both the gRPC deadline of a receive request and the request
body itself
+/// read from here.
+pub(crate) type SharedLongPollingTimeout =
Arc<parking_lot::RwLock<std::time::Duration>>;
+
const OPERATION_QUERY_ROUTE: &str = "rpc.query_route";
const OPERATION_HEARTBEAT: &str = "rpc.heartbeat";
const OPERATION_SEND_MESSAGE: &str = "rpc.send_message";
@@ -59,6 +70,20 @@ const OPERATION_NOTIFY_CLIENT_TERMINATION: &str =
"rpc.notify_client_termination
const OPERATION_QUERY_ASSIGNMENT: &str = "rpc.query_assignment";
const OPERATION_FORWARD_TO_DEADLETTER_QUEUE: &str =
"rpc.forward_to_deadletter_queue";
+/// Interval between HTTP/2 keep-alive pings.
+///
+/// The RocketMQ proxy allows client pings every 30 seconds by default
+/// (`ProxyConfig#grpcServerPermitKeepAliveTimeMillis = 30000`; the bare
grpc-java default is
+/// 5 minutes, but the proxy overrides it), so the interval must stay above
that or the
+/// server answers with `GOAWAY(ENHANCE_YOUR_CALM)`. 60s doubles the permitted
minimum,
+/// which detects a dead peer within ~90s instead of the ~330s a 300s interval
would take.
+const KEEP_ALIVE_INTERVAL: std::time::Duration =
std::time::Duration::from_secs(60);
+
+/// How long to wait for the keep-alive ping acknowledgement before
considering the
+/// connection dead and reconnecting.
+/// Kept in line with the Java client
(`RpcClientImpl.KEEP_ALIVE_TIMEOUT_MILLIS`).
+const KEEP_ALIVE_TIMEOUT: std::time::Duration =
std::time::Duration::from_secs(30);
+
#[async_trait]
#[automock]
pub(crate) trait RPCClient {
@@ -121,6 +146,26 @@ pub(crate) struct Session {
telemetry_tx: Option<mpsc::Sender<TelemetryCommand>>,
telemetry_command_tx: Option<mpsc::Sender<Command>>,
shutdown_tx: Option<oneshot::Sender<()>>,
+ /// Milliseconds elapsed since [`Instant`] `SESSION_ACTIVITY_ORIGIN` at
the time of the
+ /// last request on this session, shared with every shadow session so that
activity is
+ /// recorded on the registered session as well.
+ last_activity_millis: Arc<std::sync::atomic::AtomicU64>,
+ /// Long-polling timeout shared with the client layer, see
+ /// [`SharedLongPollingTimeout`].
+ long_polling_timeout: SharedLongPollingTimeout,
+ /// Notifies the client layer when the telemetry stream ends, so that it
can schedule a
+ /// proactive reconnect instead of waiting for the next periodic settings
sync. Mirrors
+ /// `ClientSessionImpl#renewRequestObserver` of the Java client, which
renews the stream
+ /// as soon as it is observed dead.
+ stream_closed_tx: Option<mpsc::Sender<String>>,
+}
+
+/// Origin the session activity timestamps are measured from.
+static SESSION_ACTIVITY_ORIGIN: once_cell::sync::Lazy<Instant> =
+ once_cell::sync::Lazy::new(Instant::now);
+
+fn now_millis() -> u64 {
+ SESSION_ACTIVITY_ORIGIN.elapsed().as_millis() as u64
}
impl Session {
@@ -138,6 +183,9 @@ impl Session {
telemetry_tx: self.telemetry_tx.clone(),
telemetry_command_tx: self.telemetry_command_tx.clone(),
shutdown_tx: None,
+ last_activity_millis: Arc::clone(&self.last_activity_millis),
+ long_polling_timeout: Arc::clone(&self.long_polling_timeout),
+ stream_closed_tx: self.stream_closed_tx.clone(),
}
}
@@ -145,6 +193,7 @@ impl Session {
endpoints: &Endpoints,
client_id: String,
option: &ClientOption,
+ long_polling_timeout: SharedLongPollingTimeout,
) -> Result<Self, ClientError> {
let peer = endpoints.endpoint_url().to_owned();
@@ -188,6 +237,9 @@ impl Session {
telemetry_tx: None,
telemetry_command_tx: None,
shutdown_tx: None,
+ last_activity_millis:
Arc::new(std::sync::atomic::AtomicU64::new(now_millis())),
+ long_polling_timeout,
+ stream_closed_tx: None,
})
}
@@ -223,7 +275,21 @@ impl Session {
// .with_context("peer", &peer_addr)
// })?
.connect_timeout(option.timeout)
- .tcp_nodelay(true);
+ .tcp_nodelay(true)
+ // Keep the long-lived HTTP/2 connection alive so that a rolling
deployment of the
+ // server does not silently break the telemetry stream, and so
that a dead peer is
+ // detected within `KEEP_ALIVE_TIMEOUT` instead of leaving a
half-open connection.
+ //
+ // NOTE: `KEEP_ALIVE_INTERVAL` must not fall below the minimum
ping interval the
+ // server permits. The RocketMQ proxy defaults to 30s
+ // (`ProxyConfig#grpcServerPermitKeepAliveTimeMillis`), so 60s
keeps a safety
+ // margin; lowering it further makes the server answer with
+ // GOAWAY(ENHANCE_YOUR_CALM) and drop the connection.
+ .http2_keep_alive_interval(KEEP_ALIVE_INTERVAL)
+ .keep_alive_timeout(KEEP_ALIVE_TIMEOUT)
+ // Ping even while no RPC is in flight: the telemetry stream is
long-lived but is
+ // idle between settings syncs, so without this the connection is
never probed.
+ .keep_alive_while_idle(true);
Ok(endpoint)
}
@@ -232,6 +298,8 @@ impl Session {
}
fn sign<T>(&self, message: T) -> tonic::Request<T> {
+ // Any request proves the session is in use, even when it goes through
a shadow copy.
+ self.touch();
let mut request = tonic::Request::new(message);
let metadata = request.metadata_mut();
self.sign_without_timeout(metadata);
@@ -340,6 +408,8 @@ impl Session {
self.telemetry_tx = Some(tx);
let telemetry_command_tx =
self.telemetry_command_tx.as_ref().unwrap().clone();
+ let stream_closed_tx = self.stream_closed_tx.clone();
+ let peer = self.peer().to_string();
tokio::spawn(async move {
let mut stream = response.into_inner();
loop {
@@ -354,10 +424,19 @@ impl Session {
}
Ok(None) => {
info!("telemetry command stream closed by
server");
+ if let Some(tx) = stream_closed_tx {
+ // Tell the client layer so that it
re-establishes the
+ // stream right away instead of waiting
for the next
+ // periodic settings sync.
+ let _ = tx.send(peer.clone()).await;
+ }
break;
}
Err(e) => {
error!("telemetry response error: {:?}", e);
+ if let Some(tx) = stream_closed_tx {
+ let _ = tx.send(peer.clone()).await;
+ }
break;
}
}
@@ -381,10 +460,72 @@ impl Session {
}
}
+ /// Register the channel the stream-end notification is sent through.
+ ///
+ /// Must be called before [`Self::start`], because the telemetry stream
task captures
+ /// the sender when it is spawned.
+ pub(crate) fn set_stream_closed_tx(&mut self, tx: mpsc::Sender<String>) {
+ self.stream_closed_tx = Some(tx);
+ }
+
+ /// Unconditionally re-establish the telemetry stream on this session.
+ ///
+ /// Unlike [`Self::update_settings`], which only rebuilds the stream when
it observes a
+ /// closed sender, this is called by the client layer when the server
closed the stream
+ /// (or when a heartbeat error reveals the server no longer knows this
client):
+ /// re-establishing the stream also re-registers the client with the
(possibly new)
+ /// server instance.
+ pub(crate) async fn reconnect(
+ &mut self,
+ settings: TelemetryCommand,
+ ) -> Result<(), ClientError> {
+ info!("re-establishing telemetry stream, peer={}", self.peer());
+ self.establish_telemetry_stream(settings).await
+ }
+
pub(crate) fn is_started(&self) -> bool {
self.telemetry_tx.is_some() &&
!self.telemetry_tx.as_ref().unwrap().is_closed()
}
+ /// Whether the session still owns a live telemetry stream.
+ ///
+ /// A session whose telemetry stream is gone can no longer receive
server-push
+ /// commands (settings, reconnect, unsubscribe, ...), so it must not be
kept in the
+ /// session map: keeping it makes every settings sync fail on a dead
endpoint and
+ /// blocks the long-lived connection from ever being re-established.
+ pub(crate) fn is_healthy(&self) -> bool {
+ self.is_started()
+ }
+
+ /// Record activity on this session so that it is not reaped while still
in use.
+ pub(crate) fn touch(&self) {
+ self.last_activity_millis
+ .store(now_millis(), std::sync::atomic::Ordering::Relaxed);
+ }
+
+ /// How long this session has been idle.
+ pub(crate) fn idle_duration(&self) -> std::time::Duration {
+ let last = self
+ .last_activity_millis
+ .load(std::sync::atomic::Ordering::Relaxed);
+ std::time::Duration::from_millis(now_millis().saturating_sub(last))
+ }
+
+ /// Deadline applied to a receive request.
+ ///
+ /// The request itself carries `long_polling_timeout`, so the server may
hold the
+ /// response for exactly that long before answering. The gRPC deadline
therefore has to
+ /// leave room for the request tolerance on top of it, which is what
+ /// `ClientOption::set_long_polling_timeout` documents and what the Java
client does
+ /// (`awaitDuration.plus(tolerance)`).
+ ///
+ /// Using `long_polling_timeout` alone races the server response: the
client gives up
+ /// right as the answer arrives, so every poll fails once
`long_polling_timeout` is short
+ /// (a few seconds or less) even though the server behaves correctly.
+ pub(crate) fn receive_deadline(&self) -> std::time::Duration {
+ *self.long_polling_timeout.read() + *self.option.timeout()
+ }
+
pub(crate) async fn update_settings(
&mut self,
settings: TelemetryCommand,
@@ -496,7 +637,7 @@ impl RPCClient for Session {
) -> Result<Vec<ReceiveMessageResponse>, ClientError> {
let batch_size = request.batch_size;
let mut request = self.sign(request);
- request.set_timeout(*self.option.long_polling_timeout());
+ request.set_timeout(self.receive_deadline());
let mut stream = self
.stub
.receive_message(request)
@@ -669,9 +810,18 @@ mock! {
endpoints: &Endpoints,
client_id: String,
option: &ClientOption,
+ long_polling_timeout: SharedLongPollingTimeout,
) -> Result<Self, ClientError>;
pub(crate) fn peer(&self) -> &str;
pub(crate) fn is_started(&self) -> bool;
+ pub(crate) fn is_healthy(&self) -> bool;
+ pub(crate) fn touch(&self);
+ pub(crate) fn idle_duration(&self) -> std::time::Duration;
+ pub(crate) fn set_stream_closed_tx(&mut self, tx:
mpsc::Sender<String>);
+ pub(crate) async fn reconnect(
+ &mut self,
+ settings: TelemetryCommand,
+ ) -> Result<(), ClientError>;
pub(crate) async fn update_settings(
&mut self,
settings: TelemetryCommand,
@@ -753,6 +903,7 @@ mod tests {
&Endpoints::from_url(&format!("localhost:{}",
server.address().port())).unwrap(),
"test_client".to_string(),
&ClientOption::default(),
+
Arc::new(parking_lot::RwLock::new(std::time::Duration::from_secs(40))),
)
.await;
debug!("session: {:?}", session);
@@ -765,6 +916,7 @@ mod tests {
&Endpoints::from_url("127.0.0.1:8080,127.0.0.1:8081").unwrap(),
"test_client".to_string(),
&ClientOption::default(),
+
Arc::new(parking_lot::RwLock::new(std::time::Duration::from_secs(40))),
)
.await;
debug!("session: {:?}", session);
@@ -775,4 +927,45 @@ mod tests {
async fn session_start() {
// TODO: add grpc bi-stream test
}
+
+ #[tokio::test]
+ async fn receive_deadline_leaves_room_for_the_server_response() {
+ let mut option = ClientOption::default();
+ // A short long polling timeout used to make every poll fail, because
the gRPC
+ // deadline was exactly the time the server is allowed to hold the
response.
+ option.long_polling_timeout = std::time::Duration::from_secs(3);
+ option.timeout = std::time::Duration::from_secs(3);
+
+ let server = RocketMQMockServer::start_default().await;
+ // The deadline is driven by the shared handle, which the server can
update through
+ // the telemetry stream at any time.
+ let long_polling_timeout =
+
Arc::new(parking_lot::RwLock::new(std::time::Duration::from_secs(3)));
+ let session = Session::new(
+ &Endpoints::from_url(&format!("localhost:{}",
server.address().port())).unwrap(),
+ "test_client".to_string(),
+ &option,
+ Arc::clone(&long_polling_timeout),
+ )
+ .await
+ .unwrap();
+
+ assert_eq!(
+ session.receive_deadline(),
+ std::time::Duration::from_secs(6),
+ "the deadline must be long_polling_timeout + timeout, as
documented"
+ );
+ assert!(
+ session.receive_deadline() > *option.long_polling_timeout(),
+ "the deadline must be longer than the server side long polling"
+ );
+
+ // A server-pushed update must be picked up without recreating the
session.
+ *long_polling_timeout.write() = std::time::Duration::from_secs(10);
+ assert_eq!(
+ session.receive_deadline(),
+ std::time::Duration::from_secs(13),
+ "the deadline must follow the shared long_polling_timeout"
+ );
+ }
}
diff --git a/rust/src/simple_consumer.rs b/rust/src/simple_consumer.rs
index c78bed81..0584b414 100644
--- a/rust/src/simple_consumer.rs
+++ b/rust/src/simple_consumer.rs
@@ -28,6 +28,8 @@ use crate::conf::{ClientOption, SimpleConsumerOption};
use crate::error::{ClientError, ErrorKind};
use crate::model::common::{ClientType, FilterExpression};
use crate::model::message::{AckMessageEntry, MessageView};
+use crate::pb::telemetry_command::Command;
+use crate::session::SharedLongPollingTimeout;
use crate::util::{
build_endpoints_by_message_queue, build_simple_consumer_settings,
select_message_queue,
};
@@ -98,11 +100,21 @@ impl SimpleConsumer {
}
let (shutdown_tx, mut shutdown_rx) = oneshot::channel();
self.shutdown_tx = Some(shutdown_tx);
+ let long_polling_timeout = self.client.long_polling_timeout_handle();
tokio::spawn(async move {
loop {
select! {
command = telemetry_command_rx.recv() => {
- warn!("command {:?} cannot be handled in simple
consumer.", command);
+ match command {
+ Some(Command::Settings(settings)) => {
+ Self::apply_server_settings(&settings,
&long_polling_timeout);
+ }
+ Some(command) => {
+ warn!("command {:?} cannot be handled in
simple consumer.", command);
+ }
+ // The client was dropped; there is nothing left
to handle.
+ None => break,
+ }
}
_ = &mut shutdown_rx => {
@@ -125,6 +137,32 @@ impl SimpleConsumer {
self.client.shutdown().await
}
+ /// Apply the settings pushed by the server.
+ ///
+ /// The simple consumer only consumes the long-polling timeout for now: it
drives both
+ /// the gRPC deadline of a receive request and the `long_polling_timeout`
field of the
+ /// request body itself. Mirrors `SimpleSubscriptionSettings#sync` of the
Java client.
+ fn apply_server_settings(
+ settings: &crate::pb::Settings,
+ long_polling_timeout: &SharedLongPollingTimeout,
+ ) {
+ if let Some(crate::pb::settings::PubSub::Subscription(ref
subscription)) = settings.pub_sub
+ {
+ if let Some(ref timeout) = subscription.long_polling_timeout {
+ let seconds = timeout.seconds;
+ let nanos = timeout.nanos;
+ if seconds > 0 || nanos > 0 {
+ let timeout = Duration::new(seconds as u64, nanos as u32);
+ *long_polling_timeout.write() = timeout;
+ info!(
+ "simple consumer applied server-pushed
long_polling_timeout: {:?}",
+ timeout
+ );
+ }
+ }
+ }
+ }
+
/// receive messages from the specified topic
///
/// # Arguments
@@ -232,6 +270,9 @@ mod tests {
}))
});
client.expect_start().returning(|_| Ok(()));
+ client
+ .expect_long_polling_timeout_handle()
+ .returning(||
Arc::new(parking_lot::RwLock::new(Duration::from_secs(40))));
client
.expect_client_id()
.return_const("fake_id".to_string());
diff --git a/rust/tests/rolling_restart_it.rs b/rust/tests/rolling_restart_it.rs
new file mode 100644
index 00000000..8c9f6c57
--- /dev/null
+++ b/rust/tests/rolling_restart_it.rs
@@ -0,0 +1,210 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+//! Live integration test: the telemetry stream must be re-established
proactively after a
+//! rolling restart of the proxy, instead of degrading until the next periodic
settings
+//! sync.
+//!
+//! Protocol:
+//! 1. the test starts a producer (one message per second) and a simple
consumer;
+//! 2. an external driver touches the signal file, kills the proxy and
restarts it; when the
+//! new proxy listens again, the driver writes the recovery timestamp
(seconds since the
+//! UNIX epoch) into the proxy-up file;
+//! 3. the test measures the client-side recovery: from the proxy being up
again to the
+//! first successful receive. With a proactive reconnect this is bounded by
the retry
+//! delay (about a second); the old behaviour degraded for 42-70 seconds
because the
+//! client only noticed the dead stream at the next 30s settings sync and
never
+//! re-registered.
+
+use std::collections::HashSet;
+use std::time::Duration;
+
+use rocketmq::conf::{ClientOption, ProducerOption, SimpleConsumerOption};
+use rocketmq::model::common::{FilterExpression, FilterType};
+use rocketmq::model::message::MessageBuilder;
+use rocketmq::{Producer, SimpleConsumer};
+
+const TOPIC: &str = "rust-rolling-topic";
+const GROUP: &str = "rust-rolling-group";
+const SIGNAL_FILE: &str = "/tmp/rmq_rust_live/rolling_signal";
+const PROXY_UP_FILE: &str = "/tmp/rmq_rust_live/proxy_up";
+/// The proxy gRPC server is up and the stream must be re-established well
within this
+/// budget. The proactive reconnect retries every second; the remaining budget
is the time
+/// the restarted proxy itself needs to rebuild the topic routes from the
broker heartbeats
+/// (about 20s in practice), which no client can skip.
+const CLIENT_RECOVERY_BUDGET: Duration = Duration::from_secs(30);
+const MAX_TEST_DURATION: Duration = Duration::from_secs(180);
+
+fn file_mtime(path: &str) -> Option<std::time::SystemTime> {
+ std::fs::metadata(path).and_then(|m| m.modified()).ok()
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn simple_consumer_recovers_quickly_after_rolling_restart() {
+ // This test needs a live cluster and an external driver that kills and
restarts the
+ // proxy; opt in with ROCKETMQ_ROLLING_IT=1 (see the module docs for the
protocol).
+ if std::env::var("ROCKETMQ_ROLLING_IT").as_deref() != Ok("1") {
+ println!("skipping: set ROCKETMQ_ROLLING_IT=1 and follow the module
docs to run");
+ return;
+ }
+ tracing_subscriber::fmt()
+ .with_max_level(tracing::Level::INFO)
+ .init();
+ let _ = std::fs::remove_file(SIGNAL_FILE);
+ let _ = std::fs::remove_file(PROXY_UP_FILE);
+
+ let mut consumer_option = SimpleConsumerOption::default();
+ consumer_option.set_topics(vec![TOPIC]);
+ consumer_option.set_consumer_group(GROUP);
+
+ let mut consumer_client_option = ClientOption::default();
+ consumer_client_option.set_access_url("localhost:8081");
+ consumer_client_option.set_enable_tls(false);
+
+ let mut consumer = SimpleConsumer::new(consumer_option,
consumer_client_option).unwrap();
+ consumer.start().await.unwrap();
+
+ let mut producer_option = ProducerOption::default();
+ producer_option.set_topics(vec![TOPIC]);
+ let mut producer_client_option = ClientOption::default();
+ producer_client_option.set_access_url("localhost:8081");
+ producer_client_option.set_enable_tls(false);
+ let mut producer = Producer::new(producer_option,
producer_client_option).unwrap();
+ producer.start().await.unwrap();
+ let producer = std::sync::Arc::new(producer);
+
+ // Producer task: one message per second, retrying on failures. Send
failures are
+ // printed only a few times: when the broker disk is full every send fails
and the log
+ // would otherwise drown the timeline.
+ let producer_for_task = std::sync::Arc::clone(&producer);
+ let producer_handle = tokio::spawn(async move {
+ let producer = producer_for_task;
+ let start = std::time::Instant::now();
+ let mut seq = 0u64;
+ let mut send_failures = 0u64;
+ while start.elapsed() < MAX_TEST_DURATION {
+ let body = format!("rolling-{}", seq);
+ let message = MessageBuilder::builder()
+ .set_topic(TOPIC)
+ .set_body(body.clone().into_bytes())
+ .build()
+ .unwrap();
+ match producer.send(message).await {
+ Ok(_) => {
+ seq += 1;
+ }
+ Err(e) => {
+ send_failures += 1;
+ if send_failures <= 3 {
+ println!("send failed (seq={}): {:?}", body, e);
+ }
+ }
+ }
+ tokio::time::sleep(Duration::from_secs(1)).await;
+ }
+ (seq, send_failures)
+ });
+
+ // Consumer loop: every receive outcome is recorded; the first successful
receive after
+ // the signal marks the recovery.
+ let t0 = std::time::Instant::now();
+ let mut received: HashSet<String> = HashSet::new();
+ let mut receive_failures = 0u64;
+ let mut signal_seen = false;
+ let mut recovery: Option<Duration> = None;
+
+ while t0.elapsed() < MAX_TEST_DURATION {
+ if !signal_seen && std::path::Path::new(SIGNAL_FILE).exists() {
+ signal_seen = true;
+ println!("signal file observed at +{:?}", t0.elapsed());
+ }
+
+ match consumer
+ .receive(TOPIC, &FilterExpression::new(FilterType::Tag, "*"))
+ .await
+ {
+ Ok(messages) => {
+ for message in messages {
+ let body =
String::from_utf8_lossy(message.body()).to_string();
+ if body.starts_with("rolling-") &&
received.insert(body.clone()) {
+ println!("received {} at +{:?}", body, t0.elapsed());
+ }
+ }
+ if signal_seen {
+ // The client can recover faster than the driver detects
the proxy
+ // being up, so the marker file may not exist yet for a
moment.
+ let proxy_up = loop {
+ if let Some(mtime) = file_mtime(PROXY_UP_FILE) {
+ break mtime;
+ }
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ };
+ recovery = Some(
+ std::time::SystemTime::now()
+ .duration_since(proxy_up)
+ .expect("proxy-up file must be in the past"),
+ );
+ println!(
+ "RECOVERED at +{:?}: first successful receive {:?}
after the proxy came back up",
+ t0.elapsed(),
+ recovery.unwrap()
+ );
+ break;
+ }
+ // No sleep only during the warm-up phase; afterwards keep the
loop calm.
+ if !signal_seen {
+ tokio::time::sleep(Duration::from_millis(500)).await;
+ }
+ }
+ Err(e) => {
+ receive_failures += 1;
+ if receive_failures <= 5 || receive_failures % 50 == 0 {
+ println!("receive failed at +{:?}: {:?}", t0.elapsed(), e);
+ }
+ tokio::time::sleep(Duration::from_millis(200)).await;
+ }
+ }
+ }
+
+ let (produced, send_failures) = producer_handle.await.unwrap();
+ if let Ok(producer) = std::sync::Arc::try_unwrap(producer) {
+ let _ = producer.shutdown().await;
+ }
+ consumer.shutdown().await.unwrap();
+
+ println!(
+ "produced {} messages ({} send failures), received {} unique messages,
{} receive failures",
+ produced,
+ send_failures,
+ received.len(),
+ receive_failures
+ );
+
+ let recovery = recovery.unwrap_or_else(|| {
+ panic!(
+ "the consumer never recovered within {:?}",
+ MAX_TEST_DURATION
+ );
+ });
+ println!("client-side recovery time: {:?}", recovery);
+ assert!(
+ recovery < CLIENT_RECOVERY_BUDGET,
+ "the client must re-establish its telemetry stream within {:?} after
the proxy came back, took {:?}",
+ CLIENT_RECOVERY_BUDGET,
+ recovery
+ );
+}
