fix(cluster): recover writes after a fast partition primary rejoin

Cargo.toml

@@ -206,10 +206,10 @@ hyper-util = { version = "0.1.20", features = ["server-auto", "service"] }
 iceberg = "0.9.1"
 iceberg-catalog-rest = "0.9.1"
 iceberg-storage-opendal = "0.9.1"
-iggy = { path = "core/sdk", version = "0.11.0-edge.4" }
-iggy-cli = { path = "core/cli", version = "0.14.0-edge.4" }
-iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.4" }
-iggy_common = { path = "core/common", version = "0.11.0-edge.4" }
+iggy = { path = "core/sdk", version = "0.11.0-edge.5" }
+iggy-cli = { path = "core/cli", version = "0.14.0-edge.5" }
+iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.5" }
+iggy_common = { path = "core/common", version = "0.11.0-edge.5" }
 iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.4.0-edge.3" }
 indexmap = "2.14.0"
 integration = { path = "core/integration" }

bdd/rust/tests/steps/streams.rs

@@ -18,6 +18,11 @@
 use crate::common::global_context::GlobalContext;
 use cucumber::{given, then, when};
 use iggy::prelude::{Identifier, StreamClient, StreamUpdateOptions};
+use std::time::Duration;
+use tokio::time::{Instant, sleep};
+
+const METADATA_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(2);
+const METADATA_CONVERGENCE_POLL: Duration = Duration::from_millis(10);
 
 #[given("I have no streams in the system")]
 pub async fn given_no_streams(world: &mut GlobalContext) {
@@ -123,11 +128,18 @@ pub async fn when_delete_stream_by_numeric_id(world: &mut GlobalContext) {
 
 #[then("getting the stream by its numeric ID should return no stream")]
 pub async fn then_get_stream_returns_no_stream(world: &mut GlobalContext) {
-    get_stream_by_numeric_id(world).await;
-    assert!(
-        world.last_stream_name.is_none(),
-        "Deleted stream should not be returned"
-    );
+    let deadline = Instant::now() + METADATA_CONVERGENCE_TIMEOUT;
+    loop {
+        get_stream_by_numeric_id(world).await;
+        if world.last_stream_name.is_none() {
+            return;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "Deleted stream should not be returned after {METADATA_CONVERGENCE_TIMEOUT:?}"
+        );
+        sleep(METADATA_CONVERGENCE_POLL).await;
+    }
 }
 
 async fn create_stream(world: &mut GlobalContext, stream_name: &str) {

core/ai/mcp/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy-mcp"
-version = "0.5.0-edge.3"
+version = "0.5.0-edge.4"
 description = "MCP Server for Iggy message streaming platform"
 edition = "2024"
 license = "Apache-2.0"

core/bench/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy-bench"
-version = "0.6.0-edge.4"
+version = "0.6.0-edge.5"
 edition = "2024"
 license = "Apache-2.0"
 repository = "https://github.com/apache/iggy"

core/binary_protocol/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy_binary_protocol"
-version = "0.11.0-edge.4"
+version = "0.11.0-edge.5"
 description = "Wire protocol types and codec for the Iggy binary protocol. Shared between server and SDK."
 edition = "2024"
 rust-version.workspace = true

core/cli/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy-cli"
-version = "0.14.0-edge.4"
+version = "0.14.0-edge.5"
 edition = "2024"
 rust-version.workspace = true
 authors = ["bartosz.ciesla@gmail.com"]

core/common/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy_common"
-version = "0.11.0-edge.4"
+version = "0.11.0-edge.5"
 description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
 edition = "2024"
 rust-version.workspace = true

core/connectors/runtime/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy-connectors"
-version = "0.5.0-edge.4"
+version = "0.5.0-edge.5"
 description = "Connectors runtime for Iggy message streaming platform"
 edition = "2024"
 license = "Apache-2.0"

core/integration/src/harness/handle/client_builder.rs

@@ -38,12 +38,12 @@ use crate::harness::config::{AutoLoginConfig, TlsConfig};
 use crate::harness::error::TestBinaryError;
 use iggy::http::http_client::HttpClient;
 use iggy::prelude::{
-    Client, HttpClientConfig, IggyClient, QuicClientConfig, TcpClient, TcpClientConfig, UserClient,
-    WebSocketClientConfig,
+    Client, HttpClientConfig, IggyClient, IggyDuration, QuicClientConfig, TcpClient,
+    TcpClientConfig, UserClient, WebSocketClientConfig,
 };
 use iggy::quic::quic_client::QuicClient;
 use iggy::websocket::websocket_client::WebSocketClient;
-use iggy_common::TransportProtocol;
+use iggy_common::{AutoLogin, Credentials, TransportProtocol};
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -69,7 +69,9 @@ pub struct ClientBuilder {
     transport: TransportProtocol,
     connection: ServerConnection,
     auto_login: Option<AutoLoginConfig>,
+    reconnecting_login: bool,
     tcp_nodelay: bool,
+    reestablish_after: Option<IggyDuration>,
     encryptor: Option<Arc<iggy_common::EncryptorKind>>,
 }
 
@@ -79,7 +81,9 @@ impl ClientBuilder {
             transport,
             connection,
             auto_login: None,
+            reconnecting_login: false,
             tcp_nodelay: false,
+            reestablish_after: None,
             encryptor: None,
         }
     }
@@ -90,6 +94,14 @@ impl ClientBuilder {
         self
     }
 
+    /// Configure the binary transport itself to restore the root session
+    /// after reconnecting instead of running a one-time harness login.
+    pub fn with_reconnecting_root_login(mut self) -> Self {
+        self.auto_login = Some(AutoLoginConfig::root());
+        self.reconnecting_login = true;
+        self
+    }
+
     /// Enable automatic login with custom credentials after connection.
     pub fn with_login(mut self, username: impl Into<String>, password: impl Into<String>) -> Self {
         self.auto_login = Some(AutoLoginConfig::new(username, password));
@@ -102,6 +114,13 @@ impl ClientBuilder {
         self
     }
 
+    /// Override how long a binary transport prefers its previous endpoint
+    /// before rotating through the cluster roster after a disconnect.
+    pub fn with_reestablish_after(mut self, reestablish_after: IggyDuration) -> Self {
+        self.reestablish_after = Some(reestablish_after);
+        self
+    }
+
     /// Set the client-side encryptor for encrypting/decrypting message payloads and headers.
     pub fn with_encryptor(mut self, encryptor: Arc<iggy_common::EncryptorKind>) -> Self {
         self.encryptor = Some(encryptor);
@@ -117,7 +136,9 @@ impl ClientBuilder {
             TransportProtocol::WebSocket => self.create_websocket_client().await?,
         };
 
-        if let Some(ref login) = self.auto_login {
+        if let Some(ref login) = self.auto_login
+            && (!self.reconnecting_login || self.transport == TransportProtocol::Http)
+        {
             client
                 .login_user(&login.username, &login.password)
                 .await
@@ -142,7 +163,7 @@ impl ClientBuilder {
         let tls_enabled = self.connection.tls.is_some();
         let tls_validate = self.connection.tls.as_ref().is_some_and(|t| !t.self_signed);
 
-        let config = TcpClientConfig {
+        let mut config = TcpClientConfig {
             server_address: addr.to_string(),
             nodelay: self.tcp_nodelay,
             tls_enabled,
@@ -153,8 +174,12 @@ impl ClientBuilder {
                 .as_ref()
                 .map(|p| p.to_string_lossy().to_string()),
             tls_validate_certificate: tls_validate,
+            auto_login: self.binary_auto_login(),
             ..TcpClientConfig::default()
         };
+        if let Some(reestablish_after) = self.reestablish_after {
+            config.reconnection.reestablish_after = reestablish_after;
+        }
 
         let client =
             TcpClient::create(Arc::new(config)).map_err(|e| TestBinaryError::ClientCreation {
@@ -213,11 +238,15 @@ impl ClientBuilder {
                 message: "QUIC transport not available".to_string(),
             })?;
 
-        let config = QuicClientConfig {
+        let mut config = QuicClientConfig {
             server_address: addr.to_string(),
             max_idle_timeout: 2_000_000,
+            auto_login: self.binary_auto_login(),
             ..QuicClientConfig::default()
         };
+        if let Some(reestablish_after) = self.reestablish_after {
+            config.reconnection.reestablish_after = reestablish_after;
+        }
 
         let client =
             QuicClient::create(Arc::new(config)).map_err(|e| TestBinaryError::ClientCreation {
@@ -256,7 +285,7 @@ impl ClientBuilder {
             .as_ref()
             .is_some_and(|t| !t.self_signed);
 
-        let config = WebSocketClientConfig {
+        let mut config = WebSocketClientConfig {
             server_address: addr.to_string(),
             tls_enabled,
             tls_domain: "localhost".to_string(),
@@ -266,8 +295,12 @@ impl ClientBuilder {
                 .as_ref()
                 .map(|p| p.to_string_lossy().to_string()),
             tls_validate_certificate: tls_validate,
+            auto_login: self.binary_auto_login(),
             ..WebSocketClientConfig::default()
         };
+        if let Some(reestablish_after) = self.reestablish_after {
+            config.reconnection.reestablish_after = reestablish_after;
+        }
 
         let client = WebSocketClient::create(Arc::new(config)).map_err(|e| {
             TestBinaryError::ClientCreation {
@@ -292,6 +325,20 @@ impl ClientBuilder {
         ))
     }
 
+    fn binary_auto_login(&self) -> AutoLogin {
+        if !self.reconnecting_login {
+            return AutoLogin::Disabled;
+        }
+        self.auto_login
+            .as_ref()
+            .map_or(AutoLogin::Disabled, |login| {
+                AutoLogin::Enabled(Credentials::UsernamePassword(
+                    login.username.clone(),
+                    login.password.clone().into(),
+                ))
+            })
+    }
+
     fn get_address_string(&self) -> String {
         match self.transport {
             TransportProtocol::Tcp => self