This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/durable-protocol-and-sessions in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 18a87ce3ad86e5c71819474579e6cd5ca17f7588 Author: Hubert Gruszecki <[email protected]> AuthorDate: Sat Oct 3 18:27:47 2026 +0200 fix(consensus): reject invalid bindings and trim retry overhead Poll routes accepted malformed or mismatched bind replies, and the Node protocol guard missed patch drift. Validate bindings before polling and report both session identities on a resume mismatch. Avoid capacity scans when queue bounds prove space and retain only retirement entries during commit. Preserve receipt fences and HTTP ordering without adding reservation counters. Remove a redundant journal sync and correct session and protocol documentation. --- core/common/src/traits/binary_transport.rs | 2 +- core/consensus/src/impls.rs | 59 +++++++++++ core/journal/src/partition_journal/segments.rs | 1 - core/partitions/src/iggy_partition.rs | 114 +++++++++++++++------ core/sdk/src/tcp/tcp_client.rs | 52 ++++++++++ core/server/src/http/submit.rs | 2 + foreign/go/README.md | 4 +- foreign/go/client/tcp/tcp_poll_routing.go | 12 ++- foreign/go/client/tcp/tcp_poll_routing_test.go | 44 +++++++- foreign/go/client/tcp/tcp_testing_test.go | 5 + .../iggy/client/async/tcp/vsr/VsrLoginCodec.java | 9 +- foreign/node/README.md | 15 +-- foreign/node/scripts/check-vsr-protocol.mjs | 6 +- 13 files changed, 275 insertions(+), 50 deletions(-) diff --git a/core/common/src/traits/binary_transport.rs b/core/common/src/traits/binary_transport.rs index 1c37e879a..b4cd8c42d 100644 --- a/core/common/src/traits/binary_transport.rs +++ b/core/common/src/traits/binary_transport.rs @@ -100,7 +100,7 @@ pub trait VsrSessionControl: vsr_session_sealed::Sealed + BinaryTransport { let bound = LoginRegisterResponse::decode_from(&response).map_err(|_| IggyError::InvalidFormat)?; if bound.session != identity.session { - return Err(IggyError::InvalidSession(bound.session)); + return Err(IggyError::SessionMismatch(identity.session, bound.session)); } self.bind_vsr_session(bound.session).await?; self.set_state(ClientState::Authenticated).await; diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index c48e127c5..974496254 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -460,6 +460,10 @@ where } let capacity = self.retry_capacity(table); let pipeline = self.pipeline.borrow(); + // Queue lengths bound distinct reservations without scanning either queue. + if pipeline.len() + pipeline.request_queue_len() < capacity.saturating_sub(table.count()) { + return true; + } let pending: HashSet<_> = pipeline .pending_client_ids() .filter(|candidate| { @@ -4509,7 +4513,9 @@ mod fresh_group_start_tests { #[cfg(test)] mod request_queue_tests { + use super::test_bus::NoopBus; use super::*; + use crate::client_table::{ClientTable, ClientTableMode}; use iggy_binary_protocol::{Command, Operation}; use iggy_common::ConsumerKind; use server_common::poll::AutoCommitReservationToken; @@ -4534,6 +4540,59 @@ mod request_queue_tests { msg } + #[test] + fn given_pending_clients_when_checking_capacity_should_reserve_distinct_unprotected_clients() { + const CAPACITY: usize = 3; + let mut table = ClientTable::with_mode(CAPACITY, ClientTableMode::PartitionSlice); + table.commit_capacity(CAPACITY).unwrap(); + table.commit_request(1, 1, 1, 1).unwrap(); + let consensus = VsrConsensus::new(1, 0, 1, METADATA_GROUP, NoopBus, LocalPipeline::new()); + assert!(consensus.has_retry_capacity(&table, 2)); + consensus.with_pipeline_mut(|pipeline| { + pipeline.push(PipelineEntry::new(PrepareHeader { + client: 2, + op: 1, + ..PrepareHeader::default() + })); + for client in [0, message_bus::AUTO_COMMIT_CLIENT_ID, 1, 2] { + pipeline + .push_request(RequestEntry::new(make_request(client, 1))) + .unwrap(); + } + }); + assert!( + consensus.has_retry_capacity(&table, 3), + "only client 2 reserves a new slot" + ); + consensus.with_pipeline_mut(|pipeline| { + pipeline + .push_request(RequestEntry::new(make_request(3, 1))) + .unwrap(); + }); + assert!( + !consensus.has_retry_capacity(&table, 4), + "both queues must reserve capacity" + ); + assert!( + consensus.has_retry_capacity(&table, 1), + "committed protection is reusable" + ); + assert!( + consensus.has_retry_capacity(&table, 2), + "a pending client already owns its reservation" + ); + consensus.with_pipeline_mut(LocalPipeline::clear_request_queue); + assert!( + consensus.has_retry_capacity(&table, 3), + "queue reset must release its reservations" + ); + assert_eq!( + consensus.pipeline_len(), + 1, + "the prepare reservation must survive queue reset" + ); + } + #[test] fn queued_contexts_keep_each_reservation_until_their_own_removal() { let consumer_id = 7; diff --git a/core/journal/src/partition_journal/segments.rs b/core/journal/src/partition_journal/segments.rs index 42ddedb4f..962c13f2b 100644 --- a/core/journal/src/partition_journal/segments.rs +++ b/core/journal/src/partition_journal/segments.rs @@ -182,7 +182,6 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { self.preallocate_segments.then_some(segments.max_size), ) .await?; - self.sync_segment_files().await?; self.publish(state).await?; self.state = state; self.retain_active_segment_file(); diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index a52d10437..b292ebca8 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -6425,30 +6425,23 @@ where } let mut failed_commit = false; - // Must run BEFORE the commit loop: `commit_messages` evicts the - // committed prefix, after which an entry survives only in the bounded - // repair ring - and not even there on a single replica, which keeps no - // ring at all. A miss degrades to a successful send carrying no - // confirmation, a legal answer no client can tell from a real one. + // Capture receipt and retirement data before commit_messages evicts the + // journal prefix. Missing data must fence the partition before apply; + // an incomplete receipt cannot safely answer a retry. let committed_batch_stats = self.resolve_committed_batch_stats(&drained); - let retirements: Vec<_> = drained - .iter() - .map(|entry| { - if entry.header.operation != Operation::RetireSession { - return None; - } - let prepare = self.log.journal().inner.repair_entry(entry.header.op)?; - SessionIdentity::decode_from( - prepare - .as_slice() - .get(iggy_binary_protocol::HEADER_SIZE..)?, - ) - .ok() - }) - .collect(); - for ((entry, batch_stats), retirement) in - drained.iter().zip(&committed_batch_stats).zip(&retirements) - { + let mut retirements = Vec::new(); + for (entry, batch_stats) in drained.iter().zip(&committed_batch_stats) { + let retirement = (entry.header.operation == Operation::RetireSession) + .then(|| { + let prepare = self.log.journal().inner.repair_entry(entry.header.op)?; + SessionIdentity::decode_from( + prepare + .as_slice() + .get(iggy_binary_protocol::HEADER_SIZE..)?, + ) + .ok() + }) + .flatten(); if self .dedup .commit_capacity(entry.header.retry_capacity as usize) @@ -6468,6 +6461,9 @@ where }); return; } + if let Some(identity) = retirement { + retirements.push((entry.header.op, identity)); + } } // Keep the threshold decision used before draining. An append during an // offset write or lock wait must not require an unchecked body mid-walk. @@ -6600,13 +6596,10 @@ where } } - for ((mut entry, batch_stats), retirement) in drained - .into_iter() - .zip(committed_batch_stats) - .zip(retirements) - { + let mut retirements = retirements.into_iter().peekable(); + for (mut entry, batch_stats) in drained.into_iter().zip(committed_batch_stats) { let prepare_header = entry.header; - if let Some(identity) = retirement { + if let Some((_, identity)) = retirements.next_if(|(op, _)| *op == prepare_header.op) { self.dedup .forget_session(identity.client_id, identity.session); self.required_metadata_frontier = self @@ -10310,6 +10303,69 @@ mod tests { assert!(partition.fatal.is_none(), "{:?}", partition.fatal); } + #[compio::test] + async fn given_interleaved_sends_and_retirements_when_committing_should_preserve_live_receipts() + { + const CLIENTS: u128 = 3; + const ENDING_FRONTIER: u64 = 10; + let (mut partition, replies) = recording_partition_at(0, 3); + for client_id in 1..=CLIENTS { + let mut send = checksumless_send_request(partition.namespace(), 1).transmute_header( + |header, next: &mut RoutedRequestHeader| { + *next = header; + next.client = client_id; + }, + ); + send.as_mut_slice()[size_of::<RoutedRequestHeader>()..] + .copy_from_slice(&build_segment_record(partition.namespace(), 0)); + partition.on_request(send, None).await; + if client_id == CLIENTS { + continue; + } + let identity = SessionIdentity { + client_id, + session: 1, + metadata_watermark: ENDING_FRONTIER + u64::try_from(client_id).unwrap(), + }; + let body = identity.to_bytes(); + let size = size_of::<RoutedRequestHeader>() + body.len(); + let mut retire = Message::<RoutedRequestHeader>::new(size).transmute_header( + |_, header: &mut RoutedRequestHeader| { + header.command = Command::Request; + header.operation = Operation::RetireSession; + header.cluster = TEST_CLUSTER; + header.client = message_bus::AUTO_COMMIT_CLIENT_ID; + header.session = 1; + header.request = identity.metadata_watermark; + header.metadata_watermark = identity.metadata_watermark; + header.group = partition.consensus.group(); + header.size = u32::try_from(size).unwrap(); + }, + ); + retire.as_mut_slice()[size_of::<RoutedRequestHeader>()..].copy_from_slice(&body); + partition.on_request(retire, None).await; + } + let committed_op = u64::try_from(CLIENTS * 2 - 1).unwrap(); + assert_eq!( + partition.consensus.pipeline_len(), + usize::try_from(committed_op).unwrap() + ); + partition.consensus.advance_commit_max(committed_op); + partition.commit_journal(&repair_config()).await; + assert!(partition.fatal.is_none(), "{:?}", partition.fatal); + assert_eq!(partition.consensus.commit_min(), committed_op); + assert_eq!(partition.dedup.count(), 1); + assert!(partition.dedup.contains(CLIENTS)); + assert_eq!(replies.borrow().len(), usize::try_from(CLIENTS).unwrap()); + for client_id in 1..CLIENTS { + assert!(partition.session_retired(SessionIdentity { + client_id, + session: 1, + metadata_watermark: ENDING_FRONTIER + u64::try_from(client_id).unwrap(), + })); + } + } + #[compio::test] async fn certified_empty_wal_without_superblock_preserves_its_log_view() { const CREATED_VIEW: u32 = 2; diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs index 596767f86..a61f92450 100644 --- a/core/sdk/src/tcp/tcp_client.rs +++ b/core/sdk/src/tcp/tcp_client.rs @@ -1803,6 +1803,58 @@ mod tests { stream.write_all(body).await.unwrap(); } + #[tokio::test] + async fn a_mismatched_resume_reply_should_preserve_the_original_session() { + const TEST_BUDGET: std::time::Duration = std::time::Duration::from_secs(5); + let (listener, address) = live_endpoint().await; + let client = TcpClient::create(Arc::new(TcpClientConfig { + server_address: address, + ..TcpClientConfig::default() + })) + .unwrap(); + Client::connect(&client).await.unwrap(); + client.bind_vsr_session(1).await.unwrap(); + let original = client.session_identity().await.unwrap(); + let state = client.get_state().await; + let mut events = client.subscribe_events().await; + let peer = tokio::spawn(async move { + let (mut connection, _) = listener.accept().await.unwrap(); + let (header, body) = read_test_request(&mut connection).await; + assert_eq!( + BindSessionRequest::decode_from(&body).unwrap().identity, + original + ); + let response = LoginRegisterResponse { + user_id: SESSION_USER_ID, + session: original.session + 1, + server_protocol_version: IGGY_PROTOCOL_VERSION, + server_version: WireName::new("test").unwrap(), + } + .to_bytes(); + answer_test_request(&mut connection, &header, 0, &response).await; + }); + let result = tokio::time::timeout(TEST_BUDGET, client.resume_vsr_session()) + .await + .unwrap(); + assert_eq!( + result, + Err(IggyError::SessionMismatch( + original.session, + original.session + 1 + )) + ); + assert_eq!(client.session_identity().await.unwrap(), original); + assert_eq!(client.get_state().await, state); + while let Ok(event) = events.try_recv() { + assert!(!matches!(event, DiagnosticEvent::SignedIn)); + } + tokio::time::timeout(TEST_BUDGET, peer) + .await + .unwrap() + .unwrap(); + Client::shutdown(&client).await.unwrap(); + } + #[tokio::test] async fn a_lost_reply_resumes_and_replays_only_the_original_session() { const TEST_BUDGET: std::time::Duration = std::time::Duration::from_secs(5); diff --git a/core/server/src/http/submit.rs b/core/server/src/http/submit.rs index aa39fb81e..dd2342c4b 100644 --- a/core/server/src/http/submit.rs +++ b/core/server/src/http/submit.rs @@ -375,6 +375,8 @@ async fn partition_write( ) .map_err(PartitionWriteError::Rejected)?; let gate = session.partition_gate(namespace); + // The group retains one receipt per session. Keep its lane occupied through + // the reply wait, including when NoAck has already returned after dispatch. let _partition_guard = compio::time::timeout( deadline.saturating_duration_since(Instant::now()), gate.lock(), diff --git a/foreign/go/README.md b/foreign/go/README.md index 8e6a42037..aeeb1afd5 100644 --- a/foreign/go/README.md +++ b/foreign/go/README.md @@ -61,8 +61,8 @@ but does not by itself prove that new messages were appended. In a cluster, auto-commit polls use persistent connections to partition primaries while the coordinator keeps the consumer's group membership. -Servers must support primary poll routing and consumer-session attachment -(binary commands 14, 103 and 104). Pause binary auto-commit consumers for the +Servers must support primary poll routing and shared-session binding +(binary commands 15 (`BindSession`), 103 and 104). Pause binary auto-commit consumers for the whole upgrade: upgrade every server first, then the SDKs, and restart consumers so they rejoin their groups. Older SDKs can lose membership when a backup refuses an offset commit; the new SDK does not fall back to legacy polling. diff --git a/foreign/go/client/tcp/tcp_poll_routing.go b/foreign/go/client/tcp/tcp_poll_routing.go index 697c3b51b..d6daad84a 100644 --- a/foreign/go/client/tcp/tcp_poll_routing.go +++ b/foreign/go/client/tcp/tcp_poll_routing.go @@ -298,7 +298,8 @@ func (c *IggyTcpClient) pollOnRoute(ctx context.Context, key string, payload []b if err != nil { return nil, err } - if _, state, err := slot.client.sendPollRequest(exchangeCtx, uint32(command.BindSessionCode), binding); err != nil { + response, state, err := slot.client.sendPollRequest(exchangeCtx, uint32(command.BindSessionCode), binding) + if err != nil { // An attach cannot advance an offset, even if its reply is lost. state.written = false err = slot.exchangeError(ctx, state, err) @@ -307,6 +308,15 @@ func (c *IggyTcpClient) pollOnRoute(ctx context.Context, key string, payload []b } return nil, err } + bound, err := vsr.DecodeLoginRegister(response) + if err != nil { + c.polls.dropConnection(route.endpoint, slot) + return nil, err + } + if bound.Session != route.parent.session { + c.polls.dropConnection(route.endpoint, slot) + return nil, ierror.SessionMismatch{Requested: route.parent.session, Bound: bound.Session} + } slot.parent = route.parent slot.attached = true } diff --git a/foreign/go/client/tcp/tcp_poll_routing_test.go b/foreign/go/client/tcp/tcp_poll_routing_test.go index ce6953d18..1f04987f7 100644 --- a/foreign/go/client/tcp/tcp_poll_routing_test.go +++ b/foreign/go/client/tcp/tcp_poll_routing_test.go @@ -61,7 +61,7 @@ func newPrimaryPollFixture(t *testing.T, t.Error("an auxiliary connection must bind the parent session without Register") return statusReplyFrame(vsr.OperationRegister, uint32(ierror.ErrUnauthenticated.Code()), nil) case read.code() == uint32(command.BindSessionCode): - return replyFrame(vsr.OperationNonReplicated, nil) + return bindReplyFrame(7, read.sessionID()) case read.code() == uint32(command.PollMessagesOnPrimaryCode): partition := routedPollPartition(t, read) require.Equal(t, primary, int(partition), "poll must reach its own partition primary") @@ -488,7 +488,7 @@ func TestPrimaryPoll_InternalBudgetDoesNotReturnCallerDeadline(t *testing.T) { polls := 0 serve(primaryConn, func(_ int, read request) []byte { if read.code() == uint32(command.BindSessionCode) { - return replyFrame(vsr.OperationNonReplicated, nil) + return bindReplyFrame(7, read.sessionID()) } polls++ if outcome == "refused" { @@ -694,6 +694,46 @@ func TestPrimaryPoll_StaleAttachmentReconnectsDataWithoutRejoining(t *testing.T) "attachment failure is retried before any auto-commit poll is admitted") } +func TestPrimaryPoll_InvalidBindingDoesNotPollAndRetiresConnection(t *testing.T) { + for _, test := range []struct { + name string + reply func(request) []byte + err error + }{ + { + name: "mismatched session", + reply: func(read request) []byte { + return bindReplyFrame(7, read.sessionID()+1) + }, + err: ierror.ErrSessionMismatch, + }, + { + name: "missing response body", + reply: func(request) []byte { + return replyFrame(vsr.OperationNonReplicated, nil) + }, + err: vsr.ErrTruncatedRegisterReply, + }, + } { + t.Run(test.name, func(t *testing.T) { + fixture := newPrimaryPollFixture(t, func(_, connection int, read request) ([]byte, bool) { + if connection == 0 && read.code() == uint32(command.BindSessionCode) { + return test.reply(read), true + } + return nil, false + }, nil) + _, err := pollPrimaryPartition(context.Background(), fixture.client, 0) + require.ErrorIs(t, err, test.err) + assert.Zero(t, requestCount(fixture.primaries[0].recorded(), command.PollMessagesOnPrimaryCode), + "an invalid binding must not admit an auto-commit poll") + _, err = pollPrimaryPartition(context.Background(), fixture.client, 0) + require.NoError(t, err) + assert.Equal(t, 2, fixture.primaries[0].connections(), "the invalid binding must close its connection") + assert.Equal(t, 1, fixture.coordinator.connections(), "the coordinator session must survive") + }) + } +} + func TestPrimaryPoll_PlainConsumerRoutesAndNonAutoCommitStaysOnCoordinator(t *testing.T) { fixture := newPrimaryPollFixture(t, nil, nil) stream, topic, group := groupConsumer(t) diff --git a/foreign/go/client/tcp/tcp_testing_test.go b/foreign/go/client/tcp/tcp_testing_test.go index 8d754cd65..f759720f3 100644 --- a/foreign/go/client/tcp/tcp_testing_test.go +++ b/foreign/go/client/tcp/tcp_testing_test.go @@ -141,6 +141,11 @@ func registerReplyFrame(userID uint32, session uint64) []byte { return replyFrame(vsr.OperationRegister, body) } +func bindReplyFrame(userID uint32, session uint64) []byte { + registered := registerReplyFrame(userID, session) + return replyFrame(vsr.OperationNonReplicated, registered[vsr.HeaderSize+len(resultSection()):]) +} + // clusterMetadataFrame builds a cluster roster where the node at leaderIndex // is the healthy leader. func clusterMetadataFrame(t *testing.T, leaderIndex int, addresses ...string) []byte { diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java index 373ac2227..ba80e077f 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java @@ -38,10 +38,11 @@ import java.util.Arrays; final class VsrLoginCodec { /** - * Packed semver of the {@code iggy_binary_protocol} crate this codec - * targets: {@code major << 20 | minor << 10 | patch}, 10 bits per field. - * Keep in sync with {@code core/binary_protocol/Cargo.toml}; the server - * accepts any client whose major.minor is not newer than its own. + * Packed wire protocol version this codec targets: + * {@code major << 20 | minor << 10 | patch}, 10 bits per field. + * Keep in sync with {@code core/binary_protocol/src/version.rs}, independently + * of the crate release. The server enforces its minimum wire version and rejects + * newer major.minor versions. */ static final int PROTOCOL_VERSION = (11 << 10) | 1; // 0.11.1 diff --git a/foreign/node/README.md b/foreign/node/README.md index 9d1779968..f302b1adf 100644 --- a/foreign/node/README.md +++ b/foreign/node/README.md @@ -58,11 +58,12 @@ partition at admission. VSR works over TCP and TLS. It restricts `Client` to one pooled connection because authentication, request sequencing, and consumer-group assignments belong to one consensus session. Configurations requesting more than one pooled connection fail before a socket is opened. VSR authentication translates the existing password and personal-access-token -login APIs into the register handshake required by the consensus protocol. A -disconnect or eviction invalidates the session, and later work must register a -new session. Transient not-committed responses retry the exact encoded request -within one bounded deadline. A disconnected mutation is never replayed under a -new session. +login APIs into the register handshake required by the consensus protocol. +Transport loss clears local authentication but retains the logical session and +its bind proof. Login with the same credentials first tries to bind that session; +a confirmed ended session requires a new registration. Logout forgets the local +session. Transient not-committed responses retry the exact encoded request within +one bounded deadline. A disconnected mutation is never replayed under a new session. The client pings every `heartbeatInterval` milliseconds, 5000 by default, which keeps an idle session alive when the server's `[heartbeat]` eviction is enabled. @@ -89,7 +90,7 @@ try { } ``` -The client includes its npm package version and the binary protocol crate +The client includes its npm package version and the binary wire protocol version in VSR registration. An incompatible server rejects registration with a protocol-version error instead of accepting a mismatched wire contract. @@ -148,7 +149,7 @@ Durations accept the same expressions as the Rust SDK, for example `500ms`, Cluster auto-commit polling over TCP/TLS keeps group membership on the coordinator and uses separate connections to partition primaries. It requires server support -for binary commands 14, 103 and 104. Pause binary auto-commit consumers for the +for binary commands 15 (`BindSession`), 103 and 104. Pause binary auto-commit consumers for the whole upgrade: upgrade every server first, then the SDKs, and restart consumers so they rejoin their groups. Older SDKs can lose membership when a backup refuses an offset commit; the new SDK does not fall back to legacy polling. diff --git a/foreign/node/scripts/check-vsr-protocol.mjs b/foreign/node/scripts/check-vsr-protocol.mjs index f01d7abee..dbf4758a6 100644 --- a/foreign/node/scripts/check-vsr-protocol.mjs +++ b/foreign/node/scripts/check-vsr-protocol.mjs @@ -302,9 +302,9 @@ const nodePackedVersion = nodeRegister.match( ); assert.ok(nodePackedVersion, 'Node packed protocol version source changed'); assert.deepEqual( - [Number(nodePackedVersion[1]), Number(nodePackedVersion[2])], - [Number(protocolVersion[1]), Number(protocolVersion[2])], - 'Node protocol major.minor differs from the Rust wire version' + [Number(nodePackedVersion[1]), Number(nodePackedVersion[2]), Number(nodePackedVersion[3])], + [Number(protocolVersion[1]), Number(protocolVersion[2]), Number(protocolVersion[3])], + 'Node protocol major.minor.patch differs from the Rust wire version' ); console.log('Node VSR protocol mirror matches Rust sources');
