This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 69ab058ca fix(sdk): drop the oldest diagnostic event instead of
blocking (#4501)
69ab058ca is described below
commit 69ab058caccc336f34c829a577ac679167664241
Author: Tom Planche <[email protected]>
AuthorDate: Tue Oct 13 12:13:59 2026 +0200
fix(sdk): drop the oldest diagnostic event instead of blocking (#4501)
---
core/common/src/traits/client.rs | 11 ++++++
core/sdk/src/diagnostic_events.rs | 54 ++++++++++++++++++++++++++++++
core/sdk/src/http/http_client.rs | 8 +++--
core/sdk/src/lib.rs | 1 +
core/sdk/src/quic/quic_client.rs | 7 ++--
core/sdk/src/tcp/tcp_client.rs | 35 +++++++++++++++++--
core/sdk/src/websocket/websocket_client.rs | 8 ++---
7 files changed, 112 insertions(+), 12 deletions(-)
diff --git a/core/common/src/traits/client.rs b/core/common/src/traits/client.rs
index be7afbcbc..76c48fb94 100644
--- a/core/common/src/traits/client.rs
+++ b/core/common/src/traits/client.rs
@@ -67,5 +67,16 @@ pub trait Client:
async fn shutdown(&self) -> Result<(), IggyError>;
/// Subscribe to diagnostic events.
+ ///
+ /// The event channel is bounded. When it is full, the client drops the
+ /// oldest event, so publishing never waits for a subscriber. A subscriber
+ /// that falls behind loses old events, not the newest ones.
+ ///
+ /// After a loss, the next `recv()` returns `RecvError::Overflowed` with
the
+ /// number of lost events, and the calls after it return the events that
are
+ /// left. Thus, after `Overflowed`, call `recv()` again, or read the
receiver
+ /// with `StreamExt::next()`, which skips that error.
+ ///
+ /// HTTP keeps no connection, so it publishes no events.
async fn subscribe_events(&self) -> Receiver<DiagnosticEvent>;
}
diff --git a/core/sdk/src/diagnostic_events.rs
b/core/sdk/src/diagnostic_events.rs
new file mode 100644
index 000000000..2f662181b
--- /dev/null
+++ b/core/sdk/src/diagnostic_events.rs
@@ -0,0 +1,54 @@
+// 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.
+
+//! The channel a client publishes its [`DiagnosticEvent`]s on.
+
+use async_broadcast::{Receiver, Sender, broadcast};
+use iggy_common::DiagnosticEvent;
+
+pub(crate) const DIAGNOSTIC_EVENT_CAPACITY: usize = 1000;
+
+/// Creates the channel a client publishes its diagnostic events on.
+///
+/// The client keeps the receiver so that `subscribe_events` can clone it, and
+/// that receiver never reads. A bounded channel frees a slot only once every
+/// receiver has read it, so without overflow mode the channel fills for good
+/// and the next publish waits forever, inside the connect or disconnect that
+/// sent it. In overflow mode a full channel drops its oldest event instead: a
+/// subscriber that falls behind loses old events, never the newest ones.
+pub(crate) fn diagnostic_event_channel() -> (Sender<DiagnosticEvent>,
Receiver<DiagnosticEvent>) {
+ let (mut sender, receiver) = broadcast(DIAGNOSTIC_EVENT_CAPACITY);
+ sender.set_overflow(true);
+
+ (sender, receiver)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn publishing_past_the_capacity_with_an_idle_receiver_succeeds() {
+ let (sender, _idle) = diagnostic_event_channel();
+
+ for event in 1..=DIAGNOSTIC_EVENT_CAPACITY + 1 {
+ sender
+ .try_broadcast(DiagnosticEvent::Disconnected)
+ .unwrap_or_else(|error| panic!("event {event}: {error}"));
+ }
+ }
+}
diff --git a/core/sdk/src/http/http_client.rs b/core/sdk/src/http/http_client.rs
index 2b7819ca4..3f49acf7c 100644
--- a/core/sdk/src/http/http_client.rs
+++ b/core/sdk/src/http/http_client.rs
@@ -15,10 +15,11 @@
// specific language governing permissions and limitations
// under the License.
+use crate::diagnostic_events::diagnostic_event_channel;
use crate::http::http_transport::HttpTransport;
use crate::prelude::{Client, HttpClientConfig, IggyError, NonZeroIggyDuration};
use crate::vsr::LIFECYCLE_RETRY_MAX_INTERVAL;
-use async_broadcast::{Receiver, Sender, broadcast};
+use async_broadcast::{Receiver, Sender};
use async_trait::async_trait;
use bytes::Bytes;
use iggy_common::locking::{IggyRwLock, IggyRwLockFn};
@@ -66,6 +67,9 @@ pub struct HttpClient {
access_token: IggyRwLock<String>,
/// The contexts that fence sends to an explicit partition, read from the
topic details.
pub(super) send_contexts: ConsumerGroupClientState,
+ /// HTTP keeps no connection, so no event is ever published here. The
+ /// channel exists only because the `Client` trait requires
+ /// `subscribe_events`.
events: (Sender<DiagnosticEvent>, Receiver<DiagnosticEvent>),
}
@@ -242,7 +246,7 @@ impl HttpClient {
heartbeat_interval: config.heartbeat_interval,
access_token: IggyRwLock::new(access_token),
send_contexts: ConsumerGroupClientState::new(),
- events: broadcast(1000),
+ events: diagnostic_event_channel(),
})
}
diff --git a/core/sdk/src/lib.rs b/core/sdk/src/lib.rs
index 89ae98544..721ad280b 100644
--- a/core/sdk/src/lib.rs
+++ b/core/sdk/src/lib.rs
@@ -296,6 +296,7 @@ pub mod client_provider;
pub mod client_wrappers;
pub mod clients;
pub mod consumer_ext;
+mod diagnostic_events;
pub mod http;
mod leader_aware;
mod poll_routing;
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index 61399ae93..1f1d0bc94 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use crate::diagnostic_events::diagnostic_event_channel;
use crate::leader_aware::read_transport_endpoints;
use crate::leader_aware::{
ConnectCoordinator, ConnectOwnerContext, LeaderRedirectionState,
RosterWalk,
@@ -32,7 +33,7 @@ use crate::prelude::{
IggyDuration, IggyError, IggyTimestamp, NonZeroIggyDuration,
QuicClientConfig,
};
use crate::quic::skip_server_verification::SkipServerVerification;
-use async_broadcast::{Receiver, Sender, broadcast};
+use async_broadcast::{Receiver, Sender};
use async_trait::async_trait;
use bytes::Bytes;
use iggy_binary_protocol::codes::{
@@ -179,7 +180,7 @@ impl BinaryTransport for QuicClient {
}
async fn publish_event(&self, event: DiagnosticEvent) {
- if let Err(error) = self.events.0.broadcast(event).await {
+ if let Err(error) = self.events.0.try_broadcast(event) {
error!("Failed to send a QUIC diagnostic event: {error}");
}
}
@@ -661,7 +662,7 @@ impl QuicClient {
endpoint,
connection: Arc::new(Mutex::new(None)),
state: Mutex::new(ClientState::Disconnected),
- events: broadcast(1000),
+ events: diagnostic_event_channel(),
connected_at: Mutex::new(None),
leader_redirection_state:
Mutex::new(LeaderRedirectionState::new()),
current_server_address: Mutex::new(server_address),
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index ba9517d32..b17dd5378 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use crate::diagnostic_events::diagnostic_event_channel;
use crate::leader_aware::{
ConnectCoordinator, ConnectOwnerContext, LeaderRedirectionState,
RosterWalk,
check_and_redirect_to_leader, is_same_spelling,
is_unauthenticated_metadata_probe,
@@ -30,7 +31,7 @@ use crate::tcp::tcp_connection_stream::TcpConnectionStream;
use crate::tcp::tcp_connection_stream_kind::ConnectionStreamKind;
use crate::tcp::tcp_tls_connection_stream::TcpTlsConnectionStream;
use crate::vsr::{RetainedRequest, retain_replay_header};
-use async_broadcast::{Receiver, Sender, broadcast};
+use async_broadcast::{Receiver, Sender};
use async_trait::async_trait;
use bytes::{Bytes, BytesMut};
use iggy_binary_protocol::codes::{
@@ -232,7 +233,7 @@ impl BinaryTransport for TcpClient {
}
async fn publish_event(&self, event: DiagnosticEvent) {
- if let Err(error) = self.events.0.broadcast(event).await {
+ if let Err(error) = self.events.0.try_broadcast(event) {
error!("Failed to send a TCP diagnostic event: {error}");
}
}
@@ -626,7 +627,7 @@ impl TcpClient {
client_address: Mutex::new(None),
stream: Arc::new(Mutex::new(None)),
state: Mutex::new(ClientState::Disconnected),
- events: broadcast(1000),
+ events: diagnostic_event_channel(),
connected_at: Mutex::new(None),
leader_redirection_state:
Mutex::new(LeaderRedirectionState::new()),
current_server_address: Mutex::new(server_address),
@@ -1861,6 +1862,7 @@ fn tls_server_name(server_address: &str) -> String {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::diagnostic_events::DIAGNOSTIC_EVENT_CAPACITY;
use crate::vsr::replay_after_session_reset_is_safe;
use iggy_binary_protocol::codes::{
GET_ME_CODE, JOIN_CONSUMER_GROUP_CODE, LOGOUT_USER_CODE, PING_CODE,
SEND_MESSAGES_CODE,
@@ -2877,6 +2879,33 @@ mod tests {
);
}
+ // Each failed connect publishes `Disconnected`, and nothing reads the
+ // receiver the client keeps. A server that stays down must not fill the
+ // event channel and wedge the next connect, nor every call queued behind
it.
+ #[tokio::test]
+ async fn failed_connects_past_the_event_capacity_do_not_wedge_the_client()
{
+ let client = TcpClient::create(Arc::new(TcpClientConfig {
+ server_address: dead_endpoint().await,
+ reconnection: TcpClientReconnectionConfig {
+ enabled: false,
+ ..TcpClientReconnectionConfig::default()
+ },
+ ..TcpClientConfig::default()
+ }))
+ .expect("create the client");
+
+ for attempt in 1..=DIAGNOSTIC_EVENT_CAPACITY + 1 {
+ let connect = tokio::time::timeout(Duration::from_secs(5),
TcpClient::connect(&client))
+ .await
+ .unwrap_or_else(|_| panic!("connect {attempt} hung"));
+
+ assert!(
+ matches!(connect, Err(IggyError::CannotEstablishConnection)),
+ "connect {attempt}: {connect:?}"
+ );
+ }
+ }
+
// An endpoint that accepts TCP but fails the TLS handshake is not where
// this client lives: recording it would make the next connect lead with
// it and shadow every endpoint behind it.
diff --git a/core/sdk/src/websocket/websocket_client.rs
b/core/sdk/src/websocket/websocket_client.rs
index dbc482caa..f98b55162 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use crate::diagnostic_events::diagnostic_event_channel;
use crate::leader_aware::read_transport_endpoints;
use crate::leader_aware::{
ConnectCoordinator, ConnectOwnerContext, LeaderRedirectionState,
RosterWalk,
@@ -31,7 +32,7 @@ use iggy_common::TransportProtocol;
use rustls::{ClientConfig, pki_types::pem::PemObject};
use crate::prelude::Client;
-use async_broadcast::{Receiver, Sender, broadcast};
+use async_broadcast::{Receiver, Sender};
use async_trait::async_trait;
use bytes::Bytes;
use iggy_binary_protocol::codes::{
@@ -173,7 +174,7 @@ impl BinaryTransport for WebSocketClient {
}
async fn publish_event(&self, event: DiagnosticEvent) {
- if let Err(error) = self.events.0.broadcast(event).await {
+ if let Err(error) = self.events.0.try_broadcast(event) {
error!("Failed to send a {} diagnostic event: {error}", NAME);
}
}
@@ -602,7 +603,6 @@ impl WebSocketClient {
return Err(IggyError::InvalidConfiguration);
}
- let (sender, receiver) = broadcast(1000);
let server_address = config.server_address.clone();
Ok(WebSocketClient {
poll_router: PollRouter::default(),
@@ -610,7 +610,7 @@ impl WebSocketClient {
config,
state: Mutex::new(ClientState::Disconnected),
client_address: Mutex::new(None),
- events: (sender, receiver),
+ events: diagnostic_event_channel(),
connected_at: Mutex::new(None),
leader_redirection_state:
Mutex::new(LeaderRedirectionState::new()),
current_server_address: Mutex::new(server_address),
