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 4ae2a2cf9e369dba11c7075c2fa05d29da7d1a05 Author: Hubert Gruszecki <[email protected]> AuthorDate: Sat Oct 3 19:44:07 2026 +0200 fix(partitions): preserve transferred retry checkpoints Delayed persistence completion could delete the receipt checkpoint installed by state transfer. Reclaim against the published frontier. Share immutable replies on bus paths and avoid redundant retained-file barriers while preserving recovery truncation durability. Correct SDK session diagnostics and retry docs, and check Java protocol parity. --- core/common/src/error/iggy_error.rs | 2 + core/consensus/src/client_table.rs | 49 ++-- core/journal/src/partition_journal/segments.rs | 63 ++--- core/message_bus/src/replica/handshake.rs | 9 +- core/partitions/src/iggy_partition.rs | 265 ++++++++++++++------- core/sdk/src/clients/producer.rs | 8 +- core/sdk/src/session.rs | 16 +- core/simulator/src/storage/tests.rs | 49 ++++ .../client/async/tcp/vsr/VsrLoginCodecTest.java | 23 ++ 9 files changed, 335 insertions(+), 149 deletions(-) diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs index 1628ecb3a..9a076b43c 100644 --- a/core/common/src/error/iggy_error.rs +++ b/core/common/src/error/iggy_error.rs @@ -575,6 +575,8 @@ pub enum IggyError { IncompatibleProtocolVersion(u32, u32, u32) = 14003, #[error("VSR session mismatch: requested {0}, server bound {1}")] SessionMismatch(u64, u64) = 14004, + #[error("VSR request IDs exhausted; start a new logical session")] + RequestIdExhausted = 14005, } impl IggyError { diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs index 776fb75f8..865a61283 100644 --- a/core/consensus/src/client_table.rs +++ b/core/consensus/src/client_table.rs @@ -73,19 +73,16 @@ impl CachedReply { } } -impl CachedReply { - /// Freeze owned buffer in place; no alloc. Subsequent `Clone`s are Arc bumps. - /// - /// `pub(crate)` so [`Self::header`]'s validity invariant cannot be - /// bypassed by an unvalidated buffer from outside the crate. - pub(crate) fn from_message(msg: Message<ReplyHeader>) -> Self { +impl From<Message<ReplyHeader>> for CachedReply { + fn from(message: Message<ReplyHeader>) -> Self { Self { - bytes: msg.into_generic().into_frozen(), + bytes: message.into_generic().into_frozen(), } } +} - /// Raw reply bytes for checkpoint serialization, round-tripped through - /// [`Self::from_message`] on decode. +impl CachedReply { + /// Raw reply bytes for checkpoint serialization, validated on decode. fn as_bytes(&self) -> &[u8] { self.bytes.as_slice() } @@ -856,7 +853,7 @@ impl ClientTable { } let latest_commit = reply.header().commit; let mut ring = VecDeque::with_capacity(REPLY_RING_CAPACITY); - ring.push_back(CachedReply::from_message(reply)); + ring.push_back(CachedReply::from(reply)); // Two entries claiming one slot would silently drop the first, leaving it // indexed but pointing at another client's state. if slots[slot_idx].is_some() { @@ -1023,7 +1020,7 @@ impl ClientTable { })?; let epoch = reply.header().commit; let mut ring = VecDeque::with_capacity(REPLY_RING_CAPACITY); - ring.push_back(CachedReply::from_message(reply)); + ring.push_back(CachedReply::from(reply)); self.slots[slot] = Some(ClientEntry { bind_verifier, ended_op: None, @@ -1091,7 +1088,7 @@ impl ClientTable { } // Freeze once; later dedup-hit clones Arc-bump. - let cached = CachedReply::from_message(reply); + let cached = CachedReply::from(reply); if new_request == entry.watermark { // Same request re-committed (WAL replay shape): replace in // place, never push a stale twin - two cached replies for one @@ -1183,7 +1180,7 @@ impl ClientTable { } } - /// Retain a partition receipt with its session and immutable request identity. + /// Retain a partition receipt and return a shared handle to its first committed result. /// /// # Errors /// Refuses invalid or inconsistent receipts and exhausted protection capacity. @@ -1195,7 +1192,7 @@ impl ClientTable { user_id: u32, session: u64, reply: Message<ReplyHeader>, - ) -> Result<(), ClientTableWireError> { + ) -> Result<CachedReply, ClientTableWireError> { let header = *reply.header(); if header.client == 0 || session == 0 @@ -1222,10 +1219,12 @@ impl ClientTable { let slot = self.index[&header.client]; let entry = self.slots[slot].as_mut().expect("index/slot mismatch"); entry.epoch = session; - if entry.find_cached(header.request).is_none() { - entry.push_latest(CachedReply::from_message(reply)); + if let Some(cached) = entry.find_cached(header.request) { + return Ok(cached.clone()); } - Ok(()) + let cached = CachedReply::from(reply); + entry.push_latest(cached.clone()); + Ok(cached) } /// Fold an AUTO offset request into the partition watermark without a receipt. @@ -1374,7 +1373,7 @@ impl ClientTable { { return Err(ClientTableWireError::InvalidReply); } - restored.ring.push_back(CachedReply::from_message(reply)); + restored.ring.push_back(CachedReply::from(reply)); } replacement.slots.push(Some(restored)); } @@ -1746,7 +1745,7 @@ impl ClientTable { .map_err(|_| ClientTableWireError::InvalidReply)? .try_into_typed::<ReplyHeader>() .map_err(|_| ClientTableWireError::InvalidReply)?; - ring.push_back(CachedReply::from_message(message)); + ring.push_back(CachedReply::from(message)); } let latest_commit = ring .back() @@ -3157,9 +3156,19 @@ mod tests { let watermark = receipt_watermark(7, 5, 11); let reply = Message::<ReplyHeader>::try_from(Owned::copy_from_slice(&watermark.reply)).unwrap(); - table + let committed = table .commit_partition_reply(TEST_USER_ID, watermark.session, reply.clone()) .unwrap(); + let mut changed = reply.clone(); + changed.as_mut_slice()[size_of::<ReplyHeader>()] ^= 1; + let replayed = table + .commit_partition_reply(TEST_USER_ID, watermark.session, changed) + .unwrap(); + assert!( + committed + .into_wire_bytes() + .shares_allocation(&replayed.into_wire_bytes()) + ); assert!( table .commit_partition_reply(TEST_USER_ID + 1, watermark.session, reply.clone()) diff --git a/core/journal/src/partition_journal/segments.rs b/core/journal/src/partition_journal/segments.rs index 962c13f2b..44067c49d 100644 --- a/core/journal/src/partition_journal/segments.rs +++ b/core/journal/src/partition_journal/segments.rs @@ -596,6 +596,7 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { "discarding unpublished segment bytes beyond the durable WAL frontier" ); file.truncate(cursor.position.length).await?; + self.segment_files_dirty = true; if self.preallocate_segments { file.preallocate(&cursor.path(&self.directory), segments.max_size); } @@ -682,41 +683,41 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { if self.segment_files.contains_key(&key) { return Ok(()); } - let parent = self.segment_directory()?.to_path_buf(); let retained = segment_path(&self.directory, generation, start_offset); + if self.storage.exists(&retained).await? { + let file = self.storage.open(&retained, OpenMode::ReadWrite).await?; + self.segment_files.insert(key, file); + return Ok(()); + } + let parent = self.segment_directory()?.to_path_buf(); let public = parent.join(format!("{start_offset:020}.log")); - let file = if self.storage.exists(&retained).await? { - self.storage.open(&retained, OpenMode::ReadWrite).await? + if length == 0 + && self.storage.exists(&public).await? + && self + .storage + .open(&public, OpenMode::Read) + .await? + .length() + .await? + > 0 + { + // A purged inode can still be retained by older prepares. + self.remove_segment_name(&public).await?; + } + // Segment roll can create the public name during any await. Both + // creators must open that inode without truncation, then retain it. + let mode = if length == 0 { + OpenMode::CreateOrOpen } else { - if length == 0 - && self.storage.exists(&public).await? - && self - .storage - .open(&public, OpenMode::Read) - .await? - .length() - .await? - > 0 - { - // A purged inode can still be retained by older prepares. - self.remove_segment_name(&public).await?; - } - // Segment roll can create the public name during any await. Both - // creators must open that inode without truncation, then retain it. - let mode = if length == 0 { - OpenMode::CreateOrOpen - } else { - OpenMode::ReadWrite - }; - let file = self.storage.open(&public, mode).await?; - self.storage.hard_link(&public, &retained).await?; - if length == 0 - && let Some(size) = preallocate_size - { - file.preallocate(&retained, size); - } - file + OpenMode::ReadWrite }; + let file = self.storage.open(&public, mode).await?; + self.storage.hard_link(&public, &retained).await?; + if length == 0 + && let Some(size) = preallocate_size + { + file.preallocate(&retained, size); + } self.segment_files_dirty = true; self.segment_links_dirty = true; self.segment_files.insert(key, file); diff --git a/core/message_bus/src/replica/handshake.rs b/core/message_bus/src/replica/handshake.rs index b3beffc25..c14003e6a 100644 --- a/core/message_bus/src/replica/handshake.rs +++ b/core/message_bus/src/replica/handshake.rs @@ -153,7 +153,6 @@ pub async fn acceptor_handshake<S: AsyncRead + AsyncWrite>( let peer_id = header.replica; let has_nonce = auth::has_nonce(&header.reserved_command); // Both authenticated and unauthenticated peers read the build-gate response. - let nackable = true; if header.command != Command::ReplicaHello { return reject( @@ -162,7 +161,7 @@ pub async fn acceptor_handshake<S: AsyncRead + AsyncWrite>( self_id, peer_id, HandshakeStatus::UnknownCommand, - nackable, + true, ) .await; } @@ -173,7 +172,7 @@ pub async fn acceptor_handshake<S: AsyncRead + AsyncWrite>( self_id, peer_id, HandshakeStatus::ClusterMismatch, - nackable, + true, ) .await; } @@ -186,7 +185,7 @@ pub async fn acceptor_handshake<S: AsyncRead + AsyncWrite>( self_id, peer_id, HandshakeStatus::IncompatibleBuild, - nackable, + true, ) .await; } @@ -200,7 +199,7 @@ pub async fn acceptor_handshake<S: AsyncRead + AsyncWrite>( self_id, peer_id, HandshakeStatus::DirectionalRule, - nackable, + true, ) .await; } diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index b292ebca8..bb5f1d746 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -43,6 +43,7 @@ use crate::{ PollingArgs, PollingConsumer, }; use consensus::Pipeline; +use consensus::client_table::CachedReply; use consensus::{ AutoCommitRequestContext, ClientTable, ClientTableMode, CommitLogEvent, Consensus, PartitionDiagEvent, PipelineEntry, PlaneKind, Project, ReplicaLogContext, RequestLogEvent, @@ -1071,13 +1072,12 @@ where self.drive_persistence().await; if self.fatal.is_none() && let Some(through_op) = self.pending_retry_checkpoint - && self - .persistence - .as_ref() - .is_some_and(|persistence| persistence.checkpoint_op() >= through_op) + && let Some(persistence) = &self.persistence + && persistence.checkpoint_op() >= through_op { + let checkpoint_op = persistence.checkpoint_op(); self.pending_retry_checkpoint = None; - if let Err(error) = self.reclaim_retry_checkpoints(through_op).await { + if let Err(error) = self.reclaim_retry_checkpoints(checkpoint_op).await { warn!(%error, namespace_raw = self.namespace().inner(), "cannot reclaim obsolete receipt checkpoints"); } } @@ -4563,11 +4563,10 @@ where ) { consensus::client_table::RequestStatus::New => {} consensus::client_table::RequestStatus::Duplicate(cached) => { - let committed = cached.into_message(); Self::deliver_reply_or_log( &self.consensus, message.header(), - committed, + cached, reply.take(), "duplicate reply send failed", ) @@ -6624,23 +6623,25 @@ where }; let reply = build_reply_message(&prepare_header, &body); if auto_commit { - Some(reply) + Some(CachedReply::from(reply)) } else { - let recipient_reply = send_client_replies.then(|| reply.clone()); - if let Err(error) = self.dedup.commit_partition_reply( + let cached = match self.dedup.commit_partition_reply( prepare_header.user_id, prepare_header.session, reply, ) { - error!(namespace_raw, op = prepare_header.op, %error, "cannot retain committed partition receipt"); - self.fatal = Some(FatalCommit { - namespace_raw, - op: prepare_header.op, - operation: Some(prepare_header.operation), - }); - return; - } - recipient_reply + Ok(cached) => cached, + Err(error) => { + error!(namespace_raw, op = prepare_header.op, %error, "cannot retain committed partition receipt"); + self.fatal = Some(FatalCommit { + namespace_raw, + op: prepare_header.op, + operation: Some(prepare_header.operation), + }); + return; + } + }; + send_client_replies.then_some(cached) } }; self.consensus.advance_commit_min(prepare_header.op); @@ -6672,7 +6673,7 @@ where // client's lockstep reply stream if the sentinel ever routed there. if auto_commit { if let Some(sender) = reply_sender { - let _ = sender.send(reply); + let _ = sender.send(reply.into_message()); } } else { emit_sim_event(SimEventKind::ClientReplyEmitted, &event); @@ -6690,11 +6691,11 @@ where // waiter was cancelled by the view change and the client is // already on its read-timeout. if let Some(sender) = reply_sender { - let _ = sender.send(reply); + let _ = sender.send(reply.into_message()); } else if let Err(error) = self .consensus .message_bus() - .send_to_client(prepare_header.client, reply.into_generic().into_frozen()) + .send_to_client(prepare_header.client, reply.into_wire_bytes()) .await { tracing::debug!( @@ -7072,7 +7073,11 @@ where return; } let reply = build_deny_reply_from_request(consensus, header, status); - Self::deliver_reply_or_log(consensus, header, reply, waiter, send_fail_label).await; + if let Some(waiter) = waiter { + let _ = waiter.send(reply); + return; + } + Self::deliver_reply_or_log(consensus, header, reply.into(), None, send_fail_label).await; } /// Deliver an admission-time reply (a deny, or an absorbed duplicate's @@ -7082,17 +7087,17 @@ where async fn deliver_reply_or_log( consensus: &VsrConsensus<B>, header: &RoutedRequestHeader, - reply: Message<ReplyHeader>, + reply: CachedReply, waiter: Option<consensus::Sender<Message<ReplyHeader>>>, send_fail_label: &'static str, ) { if let Some(waiter) = waiter { - let _ = waiter.send(reply); + let _ = waiter.send(reply.into_message()); return; } if let Err(send_error) = consensus .message_bus() - .send_to_client(header.client, reply.into_generic().into_frozen()) + .send_to_client(header.client, reply.into_wire_bytes()) .await { emit_partition_diag( @@ -9749,6 +9754,17 @@ mod tests { assert_eq!(partition.consensus.pipeline_len(), 0); assert_eq!(partition.stats.messages_count_inconsistent(), 2); assert_eq!(replies.borrow().len(), 2); + let consensus::client_table::RequestStatus::Duplicate(cached) = partition + .dedup + .check_partition_request(1, 0, 1, 2, Operation::SendMessages) + else { + panic!("the committed reply must remain replayable"); + }; + assert!( + replies.borrow()[1] + .1 + .shares_allocation(&cached.into_wire_bytes()) + ); } } @@ -10734,64 +10750,134 @@ mod tests { } #[compio::test] - async fn checkpoint_submission_does_not_wait_for_publication_or_reclaim_early() { - let root = tempfile::tempdir().unwrap(); - let mut config = repair_config(); - config.path_layout.streams_root = root.path().to_string_lossy().into_owned(); - let directory = std::path::PathBuf::from(config.get_partition_path(1, 1, 0)); - std::fs::create_dir_all(&directory).unwrap(); - let mut partition = partition_at_view(0, 0); - partition.set_partition_dir(directory.to_string_lossy().into_owned()); - partition.runtime_options.durability = iggy_common::Durability::Persisted; - partition.open_persistence().await.unwrap(); - partition.log.retire_front().unwrap(); - partition.install_empty_segment(&config, 0).await.unwrap(); - let prepare = checksummed_segment_prepare(1, 0, 0, b"checkpoint"); - let header = *prepare.header(); - let persistence = Rc::clone(partition.persistence.as_ref().unwrap()); - persistence - .append(prepare.clone().into_frozen(), true) - .unwrap(); - assert!(persistence.start()); - Rc::clone(&persistence).run().await; - partition - .append_repaired_send_messages(prepare) - .await - .unwrap(); - partition.consensus.restore_commit_state(0, header.op); - partition.commit_journal(&config).await; - assert!(partition.fatal().is_none()); - assert_eq!(partition.consensus.commit_min(), header.op); - let completion = Rc::new(std::cell::Cell::new(None)); - let captured = Rc::clone(&completion); - persistence.set_notifier(Rc::new(move |value| captured.set(Some(value)))); - let previous = directory.join("prepares-0/receipts-0.checkpoint"); - std::fs::write(&previous, b"obsolete receipt").unwrap(); - persistence.mark_purge(0, 0); - assert!(persistence.start()); - persistence.request_checkpoint(); + async fn checkpoint_receipts_survive_deferred_completion_and_state_transfer() { + for install_before_completion in [false, true] { + let root = tempfile::tempdir().unwrap(); + let mut config = repair_config(); + config.path_layout.streams_root = root.path().to_string_lossy().into_owned(); + let directory = std::path::PathBuf::from(config.get_partition_path(1, 1, 0)); + std::fs::create_dir_all(&directory).unwrap(); + let mut partition = partition_at_view(0, 0); + partition.set_partition_dir(directory.to_string_lossy().into_owned()); + let consumers = directory.join("offsets/consumers"); + let groups = directory.join("offsets/groups"); + std::fs::create_dir_all(&consumers).unwrap(); + std::fs::create_dir_all(&groups).unwrap(); + partition.consumer_offsets_path = Some(consumers.to_string_lossy().into_owned()); + partition.consumer_group_offsets_path = Some(groups.to_string_lossy().into_owned()); + partition.runtime_options.durability = iggy_common::Durability::Persisted; + partition.open_persistence().await.unwrap(); + partition.log.retire_front().unwrap(); + partition.install_empty_segment(&config, 0).await.unwrap(); + let prepare = checksummed_segment_prepare(1, 0, 0, b"checkpoint"); + let header = *prepare.header(); + let persistence = Rc::clone(partition.persistence.as_ref().unwrap()); + persistence + .append(prepare.clone().into_frozen(), true) + .unwrap(); + assert!(persistence.start()); + Rc::clone(&persistence).run().await; + partition + .append_repaired_send_messages(prepare.clone()) + .await + .unwrap(); + partition.consensus.restore_commit_state(0, header.op); + partition.commit_journal(&config).await; + assert!(partition.fatal().is_none()); + assert_eq!(partition.consensus.commit_min(), header.op); + let completion = Rc::new(std::cell::Cell::new(None)); + let captured = Rc::clone(&completion); + persistence.set_notifier(Rc::new(move |value| captured.set(Some(value)))); + let previous = directory.join("prepares-0/receipts-0.checkpoint"); + std::fs::write(&previous, b"obsolete receipt").unwrap(); + persistence.mark_purge(0, 0); + assert!(persistence.start()); + persistence.request_checkpoint(); - partition.checkpoint_persistence(&config).await; + partition.checkpoint_persistence(&config).await; - assert!(partition.fatal().is_none()); - assert_eq!(persistence.checkpoint_op(), 0); - assert!( - previous.exists(), - "unpublished checkpoints cannot reclaim receipts" - ); - assert!(directory.join("prepares-0/receipts-1.checkpoint").exists()); - Rc::clone(&persistence).run().await; - assert!( - persistence.failure().is_none(), - "{:?}", - persistence.failure() - ); - assert_eq!(persistence.checkpoint_op(), header.op); - partition - .on_persistence_completed(completion.get().unwrap()) - .await; - assert!(!previous.exists()); - assert!(directory.join("prepares-0/receipts-1.checkpoint").exists()); + assert!(partition.fatal().is_none()); + assert_eq!(persistence.checkpoint_op(), 0); + assert!( + previous.exists(), + "unpublished checkpoints cannot reclaim receipts" + ); + assert!(directory.join("prepares-0/receipts-1.checkpoint").exists()); + Rc::clone(&persistence).run().await; + assert!( + persistence.failure().is_none(), + "{:?}", + persistence.failure() + ); + assert_eq!(persistence.checkpoint_op(), header.op); + let checkpoint_op = if install_before_completion { + let transferred = + checksummed_segment_prepare(2, header.checksum, 1, b"transferred"); + let mut segment_body = prepare.as_slice()[size_of::<PrepareHeader>()..].to_vec(); + segment_body + .extend_from_slice(&transferred.as_slice()[size_of::<PrepareHeader>()..]); + let artifact = consensus::StateArtifact::for_bytes( + consensus::state_manifest::artifact_kind::SEGMENT_LOG, + 0, + &segment_body, + ); + let staged = partition + .spill_transfer_segment(&artifact, segment_body) + .await + .unwrap(); + let offsets = crate::state_transfer::ConsumerOffsetsWire { + dedup_capacity: partition.dedup.capacity(), + required_metadata_frontier: 0, + prepare_checksum: Some(transferred.header().checksum), + checkpoint_prepare: transferred.as_slice().to_vec(), + purge_generation: 0, + next_offset: 2, + consumers: Vec::new(), + groups: Vec::new(), + dedup: partition.dedup.watermarks_sorted(), + }; + partition + .install_state_transfer(&config, 2, vec![staged], &offsets.encode(), 0) + .await + .unwrap(); + 2 + } else { + header.op + }; + let current = directory.join(format!("prepares-0/receipts-{checkpoint_op}.checkpoint")); + assert!(current.exists()); + let completion = completion.get().unwrap(); + assert!(persistence.accepts_completion(completion)); + partition.on_persistence_completed(completion).await; + assert!(!previous.exists()); + assert!( + current.exists(), + "completion must retain the published checkpoint" + ); + drop(partition); + drop(persistence); + let mut reopened = partition_at_view(0, 0); + reopened.set_partition_dir(directory.to_string_lossy().into_owned()); + let (recovered, _) = crate::persistence::PartitionPersistence::open_with_storage( + &directory.join("prepares-0"), + reopened.namespace().inner(), + reopened.created_revision, + journal::durable_storage::DiskStorage, + ) + .await + .unwrap(); + assert_eq!(recovered.checkpoint_op(), checkpoint_op); + reopened + .restore_retry_checkpoint(recovered.checkpoint_op()) + .await + .unwrap(); + assert!(matches!( + reopened + .dedup + .check_request(1, 1, 1, Operation::SendMessages), + consensus::client_table::RequestStatus::Duplicate(_) + )); + } } #[compio::test] @@ -12712,6 +12798,7 @@ mod tests { request.as_mut_slice()[size_of::<RoutedRequestHeader>()..] .copy_from_slice(&build_segment_record(partition.namespace(), 0)); let header = *request.header(); + let mut cached_reply = None; if expected_status == 0 { let prepare = PrepareHeader { client: header.client, @@ -12734,10 +12821,13 @@ mod tests { }), ), ); - partition - .dedup - .commit_partition_reply(header.user_id, header.session, reply) - .unwrap(); + cached_reply = Some( + partition + .dedup + .commit_partition_reply(header.user_id, header.session, reply) + .unwrap() + .into_wire_bytes(), + ); } else if expected_status == IggyError::TransientNotCommitted.as_code() { let prepare = request @@ -12778,6 +12868,9 @@ mod tests { &replies[0].1.as_slice()[..size_of::<ReplyHeader>()], ); assert_eq!(reply.status, expected_status); + if let Some(cached) = cached_reply { + assert!(replies[0].1.shares_allocation(&cached)); + } assert_eq!(partition.consensus().pipeline_len(), pipeline_len); assert_eq!(partition.persistence.as_ref().unwrap().head(), 0); } diff --git a/core/sdk/src/clients/producer.rs b/core/sdk/src/clients/producer.rs index c0fdefe3f..01ef9a6a4 100644 --- a/core/sdk/src/clients/producer.rs +++ b/core/sdk/src/clients/producer.rs @@ -682,7 +682,8 @@ unsafe impl Sync for IggyProducer {} /// | Situation | Retried | What the caller ends up with | /// | --- | --- | --- | /// | the client is not signed in, so nothing may be sent yet | yes | the send goes ahead as soon as the client is signed in, or fails with [`IggyError::CannotSendMessagesDueToClientDisconnection`] once the retry budget is spent | -/// | the server explicitly refused admission ([`IggyError::TransientNotAccepted`]) | yes, as a new request | the confirmation of the admitted attempt, or the final refusal | +/// | the binary transport reports that the server refused admission ([`IggyError::TransientNotAccepted`]) | yes, as a new request | the confirmation of the admitted attempt, or the final refusal | +/// | an HTTP request fails, including a server admission refusal | no producer retry | the HTTP error; HTTP transport retries are configured separately below | /// | another request error or an ambiguous failure | no | the cause and unconfirmed batch; the caller decides whether to resend | /// | the batch committed, but its confirmation could not be read ([`IggyError::InvalidBytesResponse`] or [`IggyError::InvalidJsonResponse`], raised on the HTTP transport only) | no | the write did happen and retrying would duplicate it on purpose | /// | the request aged out of the dedup window ([`IggyError::RequestTooOld`]) | no | its outcome is unknown, and the server refuses the resend while it retains the client entry | @@ -697,8 +698,9 @@ unsafe impl Sync for IggyProducer {} /// /// The HTTP transport also resends a failed POST on its own, up to [`HttpClientConfig::retries`] /// times (3 by default), when the server answers with a 5xx, 408 or 429 status or the connection -/// fails. Each resend is a new write, so these retries multiply the producer attempts, even with -/// `send_retries(None)`. A 504 or a connection lost mid-request leaves the outcome unknown, so such +/// fails. HTTP admission refusals remain [`IggyError::HttpResponseError`] and do not trigger +/// producer retries. Each transport resend is a new write, even with `send_retries(None)`. +/// A 504 or a connection lost mid-request leaves the outcome unknown, so such /// a resend can duplicate the batch. /// /// Where a failure surfaces is the main difference between the two modes. Either way it names the diff --git a/core/sdk/src/session.rs b/core/sdk/src/session.rs index ae170e115..ecd22d3d7 100644 --- a/core/sdk/src/session.rs +++ b/core/sdk/src/session.rs @@ -93,9 +93,14 @@ impl ConsensusSession { /// Accept the original registration, including its replay after reconnect. pub fn bind(&mut self, session: u64) -> Result<(), IggyError> { - if session == 0 || self.session.is_some_and(|bound| bound != session) { + if session == 0 { return Err(IggyError::InvalidSession(session)); } + if let Some(bound) = self.session + && bound != session + { + return Err(IggyError::SessionMismatch(bound, session)); + } self.session = Some(session); Ok(()) } @@ -123,7 +128,7 @@ impl ConsensusSession { self.request_counter = self .request_counter .checked_add(1) - .ok_or(IggyError::InvalidSession(self.session.unwrap_or(0)))?; + .ok_or(IggyError::RequestIdExhausted)?; Ok(id) } @@ -196,7 +201,10 @@ mod tests { let mut session = ConsensusSession::with_client_id(7); assert_eq!(session.bind(0), Err(IggyError::InvalidSession(0))); session.bind(42).unwrap(); - assert_eq!(session.bind(43), Err(IggyError::InvalidSession(43))); + assert!(matches!( + session.bind(43), + Err(IggyError::SessionMismatch(42, 43)) + )); assert_eq!(session.session(), Some(42)); } @@ -214,7 +222,7 @@ mod tests { session.request_counter = u64::MAX; assert_eq!( session.next_request_id(), - Err(IggyError::InvalidSession(42)) + Err(IggyError::RequestIdExhausted) ); assert_eq!(session.current_request_id(), u64::MAX); } diff --git a/core/simulator/src/storage/tests.rs b/core/simulator/src/storage/tests.rs index 434066d13..68640ebe8 100644 --- a/core/simulator/src/storage/tests.rs +++ b/core/simulator/src/storage/tests.rs @@ -1783,6 +1783,55 @@ async fn owned_segment_baseline( (storage, journal) } +#[test] +fn reopening_retained_segments_syncs_only_recovery_mutations() { + block_on(async { + for buffered_tail in [false, true] { + let (storage, mut journal) = owned_segment_baseline(buffered_tail).await; + journal.cleanup_obsolete().await; + drop(journal); + storage.clear_trace(); + let recovered = + PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7, storage.clone()) + .await + .unwrap(); + let trace = storage.trace(); + let file_syncs = trace + .iter() + .filter(|operation| **operation == StorageOperation::FileSync) + .count(); + assert_eq!(file_syncs, 1 + usize::from(buffered_tail), "{trace:?}"); + assert_eq!( + trace + .iter() + .filter(|operation| **operation == StorageOperation::DirectorySync) + .count(), + 3, + "retained names need no new publication barrier: {trace:?}" + ); + let expected_head = if buffered_tail { 1 } else { 2 }; + assert_eq!(recovered.head(), expected_head); + drop(recovered); + storage.crash(Crash::PowerLoss); + let public = Path::new("/partition/00000000000000000000.log"); + let file = storage.open(public, OpenMode::Read).await.unwrap(); + assert_eq!( + file.length().await.unwrap(), + expected_head * OWNED_BATCH_BYTES as u64 + ); + let recovered = + PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7, storage) + .await + .unwrap(); + assert_eq!(recovered.head(), expected_head); + assert_eq!( + recovered.prepares().await.unwrap().len() as u64, + expected_head + ); + } + }); +} + #[test] fn sealed_tail_recovery_preserves_the_public_name_at_every_crash_boundary() { block_on(async { diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodecTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodecTest.java index d629618ac..87a620d4a 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodecTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodecTest.java @@ -30,14 +30,37 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EmptySource; import org.junit.jupiter.params.provider.NullSource; +import java.io.IOException; import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; import java.util.Arrays; +import java.util.regex.Matcher; +import java.util.regex.Pattern; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; class VsrLoginCodecTest { + @Test + void protocolVersionMatchesRustWireVersion() throws IOException { + Path versionPath = Path.of("core/binary_protocol/src/version.rs"); + Path root = Path.of("").toAbsolutePath(); + while (root != null && !Files.isRegularFile(root.resolve(versionPath))) { + root = root.getParent(); + } + assertThat(root).as("repository containing %s", versionPath).isNotNull(); + Matcher version = Pattern.compile( + "pub const IGGY_PROTOCOL_VERSION:\\s*u32\\s*=\\s*pack_protocol_version\\((\\d+),\\s*(\\d+),\\s*(\\d+)\\)") + .matcher(Files.readString(root.resolve(versionPath))); + assertThat(version.find()).as("Rust wire protocol version declaration").isTrue(); + int major = Integer.parseInt(version.group(1)); + int minor = Integer.parseInt(version.group(2)); + int patch = Integer.parseInt(version.group(3)); + assertThat(VsrLoginCodec.PROTOCOL_VERSION).isEqualTo((major << 20) | (minor << 10) | patch); + } + @Test void sdkVersionFieldEncodesVersionAsUtf8() { assertThat(VsrLoginCodec.sdkVersionField("0.9.0-SNAPSHOT"))
