You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/10/19 12:20:42 UTC
(iggy) 01/04: fix(sdk): resume polling after a replica detaches its session
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/sdk-partition-routing
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 03ab789d9d55ad283be09fa141acc8434947341c
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Oct 19 14:19:18 2026 +0200
fix(sdk): resume polling after a replica detaches its session
A restored replica can refuse a live session's reads with status 40.
Rebind that link and retry refused reads within the existing deadline,
preserving mutation refusals and session identity.
---
core/sdk/src/driver.rs | 90 ++++++++++++++++++++++++++++++++++++++++++++------
1 file changed, 80 insertions(+), 10 deletions(-)
diff --git a/core/sdk/src/driver.rs b/core/sdk/src/driver.rs
index cb8a7a39c..cc5b40513 100644
--- a/core/sdk/src/driver.rs
+++ b/core/sdk/src/driver.rs
@@ -354,14 +354,22 @@ impl OperationDriver {
Ok(reply) => match decode_reply_for(&request, &reply.header, reply.body) {
Ok(outcome) => {
self.session.observe_metadata_reply(&reply.header);
- close_if_detached(&hop.link, &outcome.result);
+ let bound = self.session.epoch().bound;
+ close_if_detached(&hop.link, &outcome.result, bound);
if outcome.committed && call.uncertain {
return match outcome.result {
Ok(body) => self.signed_in(call, &hop, body),
Err(error) => Err(Next::Committed(error)),
};
}
- if call.uncertain
+ if bound
+ && request.operation == Operation::NonReplicated
+ && !is_sign_in(call.code)
+ && outcome.result == Err(IggyError::Unauthenticated)
+ {
+ // A detached link refused the read before admission.
+ Err(IggyError::TransientNotAccepted)
+ } else if call.uncertain
&& !matches!(
request.operation,
Operation::NonReplicated | Operation::Register | Operation::Logout
@@ -704,7 +712,11 @@ impl OperationDriver {
let frame = ticket.freeze(code, body, self.read_context(), None)?;
let reply = link.send(frame, deadline, ticket.sent_flag()).await?;
let result = read_reply(&self.session, reply);
- close_if_detached(link, &result);
+ let bound = self.session.epoch().bound;
+ close_if_detached(link, &result, bound);
+ if bound && result == Err(IggyError::Unauthenticated) {
+ return Err(IggyError::TransientNotAccepted);
+ }
result
}
@@ -946,13 +958,12 @@ fn retries_lookup(error: &IggyError) -> bool {
) || lost_link(error)
}
-/// A reply with `StaleClient` means that the server no longer attaches
-/// `link` to the session, for example because the session ended. The slot
-/// of a closed link binds the session again, as after an eviction. That
-/// bind resumes a session that the server still knows, and drops an ended
-/// one, so the slot signs in again.
-fn close_if_detached(link: &Link, result: &Result<Bytes, IggyError>) {
- if matches!(result, Err(IggyError::StaleClient)) {
+/// State transfer can detach a link while its session remains live.
+/// Rebinding distinguishes that case from an ended session that needs sign-in.
+fn close_if_detached(link: &Link, result: &Result<Bytes, IggyError>, bound: bool) {
+ if matches!(result, Err(IggyError::StaleClient))
+ || (bound && matches!(result, Err(IggyError::Unauthenticated)))
+ {
link.close();
}
}
@@ -2167,6 +2178,65 @@ mod tests {
}
}
+ #[tokio::test(start_paused = true)]
+ async fn given_a_detached_replica_when_polling_should_rebind_and_continue_the_same_session() {
+ for (auto_commit, strategy) in [
+ (false, WirePollingStrategy::first()),
+ (true, WirePollingStrategy::first()),
+ (false, WirePollingStrategy::next()),
+ ] {
+ let fixture = Fixture::new().await;
+ let fence = fixture.driver.session.fence();
+ let binds = fixture.cluster.count(BIND_SESSION_CODE);
+ fixture.cluster.script(
+ &endpoint(PRIMARY),
+ POLL_MESSAGES_CODE,
+ [Answer::Refuse(IggyError::Unauthenticated)],
+ );
+
+ fixture
+ .driver
+ .poll(&fixture.poll(auto_commit, strategy), None)
+ .await
+ .unwrap();
+
+ eventually("the detached replica binds the same session", || {
+ fixture.cluster.count(BIND_SESSION_CODE) > binds
+ })
+ .await;
+ assert_eq!(fixture.cluster.count(POLL_MESSAGES_CODE), 2);
+ assert_eq!(fixture.driver.session.fence(), fence);
+ assert_eq!(fixture.cluster.count(LOGIN_REGISTER_CODE), 1);
+ }
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_a_detached_replica_when_looking_up_a_route_should_try_another_ready_replica() {
+ let fixture = Fixture::new().await;
+ let fence = fixture.driver.session.fence();
+ let binds = fixture.cluster.count(BIND_SESSION_CODE);
+ fixture.cluster.script(
+ &endpoint(0),
+ GET_PARTITION_ROUTE_CODE,
+ [Answer::Refuse(IggyError::Unauthenticated)],
+ );
+
+ fixture
+ .driver
+ .poll(&fixture.poll(false, WirePollingStrategy::first()), None)
+ .await
+ .unwrap();
+
+ eventually("the detached lookup link binds again", || {
+ fixture.cluster.count(BIND_SESSION_CODE) > binds
+ })
+ .await;
+ assert_eq!(fixture.cluster.count(POLL_MESSAGES_CODE), 1);
+ assert!(fixture.cluster.count(GET_PARTITION_ROUTE_CODE) > 1);
+ assert_eq!(fixture.driver.session.fence(), fence);
+ assert_eq!(fixture.cluster.count(LOGIN_REGISTER_CODE), 1);
+ }
+
#[tokio::test(start_paused = true)]
async fn given_a_sent_mutation_when_its_reply_denies_it_as_unauthenticated_should_free_the_lane()
{