This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/partition-routing in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b536e4b2a4a9518f89c098a68c69825a0af84208 Author: Hubert Gruszecki <[email protected]> AuthorDate: Tue Oct 13 18:27:14 2026 +0200 fix(server): preserve routing and recovery fences Pipelined requests could overtake metadata and consumer progress; HTTP reads could outlive their primary check. Retain dependency lanes through replies and validate ownership on the actual read. Keep retirement proof tied to the deleted incarnation and settle view proof only after truncation succeeds. Reject obsolete snapshot invariants and preserve completed deletes if marker creation fails. --- .../tests/cluster/partition_primary_routing.rs | 96 +++++++-- .../tests/server/pipelined_frames_vsr.rs | 233 ++++++++++++++++++++- core/metadata/src/stm/snapshot.rs | 28 ++- core/partitions/src/iggy_partition.rs | 116 +++++++++- core/partitions/src/io.rs | 80 ++++++- core/server/src/consumer_group/liveness.rs | 10 +- core/server/src/dispatch/failure.rs | 2 + core/server/src/dispatch/host.rs | 7 +- core/server/src/dispatch/mod.rs | 226 ++++++++++++++++++-- core/server/src/dispatch/partition.rs | 90 +++++++- core/server/src/http/forward.rs | 39 +++- core/server/src/http/handlers.rs | 13 +- core/server/src/http/submit.rs | 1 + core/shard/src/lib.rs | 34 ++- core/shard/src/partition_io.rs | 68 ++++++ core/shard/src/poll.rs | 46 ++-- core/shard/src/poll/completion_tests.rs | 82 ++++++-- 17 files changed, 1052 insertions(+), 119 deletions(-) diff --git a/core/integration/tests/cluster/partition_primary_routing.rs b/core/integration/tests/cluster/partition_primary_routing.rs index c455ec5e1..a9f27da0d 100644 --- a/core/integration/tests/cluster/partition_primary_routing.rs +++ b/core/integration/tests/cluster/partition_primary_routing.rs @@ -96,10 +96,6 @@ const SEND_BUDGET: Duration = Duration::from_secs(20); /// the state this test needs before it can observe anything. const PRECONDITION_BUDGET: Duration = Duration::from_secs(20); const PRECONDITION_POLL: Duration = Duration::from_millis(500); -/// How long a backup gets to relay its requests to a frozen partition primary -/// before the primary dies under them. Far below `cluster.heartbeat_timeout` -/// (5s by default), so no election ends while they are in flight. -const RELAY_IN_FLIGHT: Duration = Duration::from_secs(1); fn message(payload: &str) -> IggyMessage { IggyMessage::from_str(payload).expect("build message") @@ -538,13 +534,20 @@ async fn given_split_primaries_when_http_calls_a_partition_backup_should_serve_o StatusCode::NO_CONTENT, "store on a backup" ); - let response = http - .client - .get(&offset) - .bearer_auth(&http.token) - .send() - .await - .expect("get offset request"); + pause_node(harness, partition_primary, true); + let (response, poll_response, ()) = tokio::join!( + http.client.get(&offset).bearer_auth(&http.token).send(), + http.client.get(&poll).bearer_auth(&http.token).send(), + async { + wait_for_queued_relays(harness, partition_primary, 2).await; + pause_node(harness, partition_primary, false); + }, + ); + let poll_response = poll_response.expect("relayed poll request"); + assert_eq!(poll_response.status(), StatusCode::OK); + let relayed: PolledMessages = poll_response.json().await.expect("decode relayed poll"); + assert_eq!(relayed.messages.len(), expected.len()); + let response = response.expect("relayed get offset request"); assert_eq!(response.status(), StatusCode::OK, "get on a backup"); let info: ConsumerOffsetInfo = response.json().await.expect("decode ConsumerOffsetInfo"); assert_eq!(info.stored_offset, store.offset); @@ -690,26 +693,27 @@ async fn given_split_primaries_when_a_relayed_read_loses_its_primary_should_rese let get = |url: String| http.client.get(url).bearer_auth(&http.token).send(); // A frozen primary takes the relayed requests and never answers them. - let pid = harness - .node(partition_primary) - .pid() - .expect("the partition primary is running"); - // SAFETY: a plain syscall on a child this harness spawned and has not reaped. - assert_eq!(unsafe { libc::kill(pid as libc::pid_t, libc::SIGSTOP) }, 0); + pause_node(harness, partition_primary, true); let offset = format!("{offsets}?consumer_id={RELAY_CONSUMER_ID}&partition_id={PARTITION_ID}"); let (offset_read, read, auto_commit, ()) = tokio::join!( get(offset.clone()), get(poll(false)), get(poll(true)), async { - sleep(RELAY_IN_FLIGHT).await; + wait_for_queued_relays(harness, partition_primary, 3).await; harness .kill_node(partition_primary) .expect("kill the frozen partition primary"); }, ); - let (status, body) = status_and_body(read.expect("poll without auto_commit")).await; + let response = read.expect("poll without auto_commit"); + let response = if response.status() == StatusCode::SERVICE_UNAVAILABLE { + until_primary_resolved(|| async { get(poll(false)).await.expect("read-only poll") }).await + } else { + response + }; + let (status, body) = status_and_body(response).await; assert_eq!(status, StatusCode::OK, "poll without auto_commit: {body}"); let polled: PolledMessages = serde_json::from_str(&body).expect("decode PolledMessages"); assert_eq!(polled.messages.len(), GROUP_PAYLOADS.len()); @@ -738,6 +742,60 @@ async fn given_split_primaries_when_a_relayed_read_loses_its_primary_should_rese assert_eq!(info.stored_offset, store.offset); } +fn pause_node(harness: &TestHarness, node: usize, paused: bool) { + let pid = harness.node(node).pid().expect("the node is running"); + let signal = if paused { libc::SIGSTOP } else { libc::SIGCONT }; + // SAFETY: the signal targets a child this harness owns and has not reaped. + assert_eq!(unsafe { libc::kill(pid as libc::pid_t, signal) }, 0); +} + +/// The stopped HTTP listener cannot consume bytes. Nonempty receive queues +/// prove the relay requests reached it before it is resumed or killed. +async fn wait_for_queued_relays(harness: &TestHarness, node: usize, expected: usize) { + const TCP_ESTABLISHED: &str = "01"; + let port = harness + .node(node) + .http_addr() + .expect("node HTTP address") + .port(); + timeout(PRECONDITION_BUDGET, async { + loop { + let queued = ["/proc/net/tcp", "/proc/net/tcp6"] + .into_iter() + .map(|path| { + std::fs::read_to_string(path) + .expect("read the kernel TCP connection table") + .lines() + .skip(1) + .filter(|line| { + let fields: Vec<_> = line.split_whitespace().collect(); + fields.get(3) == Some(&TCP_ESTABLISHED) + && fields + .get(1) + .and_then(|address| address.rsplit_once(':')) + .and_then(|(_, port)| u16::from_str_radix(port, 16).ok()) + == Some(port) + && fields + .get(4) + .and_then(|queues| queues.split_once(':')) + .and_then(|(_, received)| { + usize::from_str_radix(received, 16).ok() + }) + .is_some_and(|received| received > 0) + }) + .count() + }) + .sum::<usize>(); + if queued >= expected { + break; + } + sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("every HTTP relay must reach the paused partition primary"); +} + async fn status_and_body(response: reqwest::Response) -> (StatusCode, String) { let status = response.status(); (status, response.text().await.expect("response body")) diff --git a/core/integration/tests/server/pipelined_frames_vsr.rs b/core/integration/tests/server/pipelined_frames_vsr.rs index 3f86745dc..05ada2fdc 100644 --- a/core/integration/tests/server/pipelined_frames_vsr.rs +++ b/core/integration/tests/server/pipelined_frames_vsr.rs @@ -23,12 +23,22 @@ //! each reply before it sends the next frame. use iggy::prelude::*; -use iggy_binary_protocol::codes::{GET_ME_CODE, PING_CODE}; -use iggy_binary_protocol::consensus::ReplyHeader; +use iggy_binary_protocol::codes::{GET_ME_CODE, GET_STREAM_CODE, PING_CODE, POLL_MESSAGES_CODE}; +use iggy_binary_protocol::consensus::{Operation, ReplyHeader, result_code}; +use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest; +use iggy_binary_protocol::requests::messages::PollMessagesRequest; +use iggy_binary_protocol::requests::streams::{CreateStreamRequest, GetStreamRequest}; +use iggy_binary_protocol::responses::streams::GetStreamResponse; +use iggy_binary_protocol::{ + AckLevel, HEADER_SIZE, WireConsumer, WireDecode, WireEncode, WireIdentifier, WireName, + WireOptions, WirePollingStrategy, +}; +use integration::harness::{TestBinary, TestHarness}; use integration::iggy_harness; use crate::server::raw_tcp::{ - connect, non_replicated_header, read_reply, register_root, write_frame, + connect, non_replicated_header, partition_context, read_reply, register_root, request_header, + write_frame, }; /// The header validator requires a nonzero client id; the value is otherwise @@ -74,3 +84,220 @@ fn reply_header(header: &[u8]) -> ReplyHeader { bytemuck::checked::try_pod_read_unaligned::<ReplyHeader>(header) .expect("a Reply header decodes") } + +#[iggy_harness(cluster_nodes = 3)] +async fn given_a_pending_metadata_write_when_a_read_and_ping_follow_should_order_the_read_and_serve_ping( + harness: &TestHarness, +) { + const NAME: &str = "pipelined-created-stream"; + const CREATE: u64 = 1; + const READ: u64 = 2; + const PING: u64 = 3; + let mut socket = connect(harness).await; + let session = register_root(&mut socket, CLIENT_ID).await; + for node in 1..harness.cluster_size() { + signal_node(harness, node, libc::SIGSTOP); + } + let create = CreateStreamRequest { + name: WireName::new(NAME).unwrap(), + options: WireOptions::empty(), + } + .to_bytes(); + let header = request_header( + Operation::CreateStream, + CLIENT_ID, + session, + CREATE, + create.len(), + ); + write_frame(&mut socket, &header, &create).await; + let read = GetStreamRequest { + stream_id: WireIdentifier::named(NAME).unwrap(), + } + .to_bytes(); + let mut header = non_replicated_header(CLIENT_ID, session, READ, GET_STREAM_CODE); + header.size = u32::try_from(HEADER_SIZE + read.len()).unwrap(); + write_frame(&mut socket, &header, &read).await; + write_frame( + &mut socket, + &non_replicated_header(CLIENT_ID, session, PING, PING_CODE), + &[], + ) + .await; + let first = reply_header(&read_reply(&mut socket).await.0); + for node in 1..harness.cluster_size() { + signal_node(harness, node, libc::SIGCONT); + } + assert_eq!( + first.request, PING, + "PING must bypass both the blocked write and its dependent read" + ); + assert_eq!(first.status, 0); + for _ in 0..2 { + let (header, body) = read_reply(&mut socket).await; + let reply = reply_header(&header); + assert_eq!(reply.status, 0); + match reply.request { + CREATE => assert_eq!(result_code(&body), Some(0)), + READ => assert_eq!( + GetStreamResponse::decode_from(&body) + .unwrap() + .stream + .name + .as_str(), + NAME + ), + request => panic!("unexpected reply {request}"), + } + } +} + +#[iggy_harness(cluster_nodes = 1)] +async fn given_two_pipelined_next_polls_when_auto_committing_should_return_distinct_messages( + harness: &mut TestHarness, +) { + let context = seed_poll_messages(harness).await; + let mut socket = connect(harness).await; + let session = register_root(&mut socket, CLIENT_ID).await; + let body = next_poll().to_bytes(); + for request in 1..=2 { + let mut header = non_replicated_header(CLIENT_ID, session, request, POLL_MESSAGES_CODE); + header.size = u32::try_from(HEADER_SIZE + body.len()).unwrap(); + header.partition_incarnation = context.incarnation; + header.minimum_metadata_op = context.metadata_op; + write_frame(&mut socket, &header, &body).await; + } + for offset in 0..2 { + let (header, body) = read_reply(&mut socket).await; + let reply = reply_header(&header); + assert_eq!(reply.status, 0); + assert_eq!(reply.request, offset + 1); + let polled = PolledMessages::from_bytes(body.into()).unwrap(); + assert_eq!(polled.messages.len(), 1); + assert_eq!(polled.messages[0].header.offset, offset); + } +} + +#[iggy_harness(cluster_nodes = 1)] +async fn given_a_pipelined_poll_then_store_when_both_complete_should_preserve_consumer_progress( + harness: &mut TestHarness, +) { + let context = seed_poll_messages(harness).await; + let mut socket = connect(harness).await; + let session = register_root(&mut socket, CLIENT_ID).await; + let poll = next_poll(); + let body = poll.to_bytes(); + let mut header = non_replicated_header(CLIENT_ID, session, 1, POLL_MESSAGES_CODE); + header.size = u32::try_from(HEADER_SIZE + body.len()).unwrap(); + header.partition_incarnation = context.incarnation; + header.minimum_metadata_op = context.metadata_op; + write_frame(&mut socket, &header, &body).await; + let store = StoreConsumerOffsetRequest { + consumer: poll.consumer, + stream_id: poll.stream_id, + topic_id: poll.topic_id, + partition_id: Some(0), + offset: 1, + ack: AckLevel::Quorum, + } + .to_bytes(); + let mut header = request_header( + Operation::StoreConsumerOffset, + CLIENT_ID, + session, + 2, + store.len(), + ); + header.partition_incarnation = context.incarnation; + header.minimum_metadata_op = context.metadata_op; + write_frame(&mut socket, &header, &store).await; + let (header, body) = read_reply(&mut socket).await; + let reply = reply_header(&header); + assert_eq!( + reply.request, 1, + "the poll must finish before its following store" + ); + assert_eq!(reply.status, 0); + let polled = PolledMessages::from_bytes(body.into()).unwrap(); + assert_eq!(polled.messages.len(), 1); + assert_eq!(polled.messages[0].header.offset, 0); + let (header, _) = read_reply(&mut socket).await; + let reply = reply_header(&header); + assert_eq!(reply.request, 2); + assert_eq!(reply.status, 0); + let client = harness.tcp_root_client().await.unwrap(); + let stored = client + .get_consumer_offset( + &Consumer::new(Identifier::numeric(1).unwrap()), + &Identifier::named(STREAM).unwrap(), + &Identifier::named(TOPIC).unwrap(), + Some(0), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(stored.stored_offset, 1); +} + +const STREAM: &str = "pipelined-stream"; +const TOPIC: &str = "pipelined-topic"; + +async fn seed_poll_messages(harness: &mut TestHarness) -> PartitionContext { + let client = harness.tcp_root_client().await.unwrap(); + let stream = Identifier::named(STREAM).unwrap(); + let topic = Identifier::named(TOPIC).unwrap(); + client.create_stream(STREAM).await.unwrap(); + client + .create_topic( + &stream, + TOPIC, + &TopicCreateOptions { + partitions_count: Some(1), + messages_required_to_save: Some(1), + durability: Durability::Persisted, + ..Default::default() + }, + ) + .await + .unwrap(); + let mut messages = (0..3) + .map(|value| { + IggyMessage::builder() + .payload(format!("message-{value}").into()) + .build() + .unwrap() + }) + .collect::<Vec<_>>(); + client + .send_messages( + &stream, + &topic, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .unwrap(); + drop(client); + // Recovery leaves the messages on disk, so polling must yield for I/O. + harness.restart_server().await.unwrap(); + let client = harness.tcp_root_client().await.unwrap(); + partition_context(&client, &stream, &topic, 0).await +} + +fn next_poll() -> PollMessagesRequest { + PollMessagesRequest { + consumer: WireConsumer::consumer(WireIdentifier::numeric(1)), + stream_id: WireIdentifier::named(STREAM).unwrap(), + topic_id: WireIdentifier::named(TOPIC).unwrap(), + partition_id: Some(0), + strategy: WirePollingStrategy::next(), + count: 1, + auto_commit: true, + } +} + +fn signal_node(harness: &TestHarness, node: usize, signal: libc::c_int) { + let pid = harness.node(node).pid().expect("the node is running"); + // SAFETY: the signal targets a child owned by the harness and not yet reaped. + assert_eq!(unsafe { libc::kill(pid as libc::pid_t, signal) }, 0); +} diff --git a/core/metadata/src/stm/snapshot.rs b/core/metadata/src/stm/snapshot.rs index 51935e998..dcb7cf9ac 100644 --- a/core/metadata/src/stm/snapshot.rs +++ b/core/metadata/src/stm/snapshot.rs @@ -24,9 +24,9 @@ use crate::stm::user::UsersSnapshot; /// The snapshot format version written by this build. /// -/// Each version identifies a serialized shape. Bump it when fields change, -/// including nested fields: `MessagePack` encodes structs positionally, so a -/// layout change can reinterpret bytes without a deserialization error. +/// Each version identifies a serialized shape and its invariants. Bump it when +/// fields change, including nested fields: `MessagePack` encodes structs +/// positionally, so a layout change can reinterpret bytes without an error. /// /// [`MetadataSnapshot::decode`] accepts versions from /// [`MIN_READABLE_SNAPSHOT_FORMAT_VERSION`] through this version. Older accepted @@ -56,7 +56,10 @@ use crate::stm::user::UsersSnapshot; /// /// Version 11: streams also persist pending lifecycle intents, and each /// `TopicSnapshot` the op of its last consumer group catalog change. -pub const SNAPSHOT_FORMAT_VERSION: u32 = 11; +/// +/// Version 12: pending consumer group owner generations equal their metadata +/// operation, replacing independently incremented generations. +pub const SNAPSHOT_FORMAT_VERSION: u32 = 12; /// Oldest format version [`MetadataSnapshot::decode`] still reads. /// @@ -604,7 +607,7 @@ mod tests { // operator's boot reading one field's bytes as another's. Changing either // number is the reminder to change the other. const FIELD_COUNT: u32 = 7; - const PINNED_VERSION: u32 = 11; + const PINNED_VERSION: u32 = 12; let encoded = MetadataSnapshot::new(0).encode().unwrap(); let mut cursor = encoded.as_slice(); @@ -615,7 +618,7 @@ mod tests { ); assert_eq!( SNAPSHOT_FORMAT_VERSION, PINNED_VERSION, - "SNAPSHOT_FORMAT_VERSION moved; confirm the shape moved with it" + "SNAPSHOT_FORMAT_VERSION moved; confirm the shape or invariants changed with it" ); } @@ -797,6 +800,19 @@ mod tests { } } + #[test] + fn given_independent_owner_generation_format_when_decoding_should_refuse_before_restore() { + const INDEPENDENT_OWNER_GENERATION_FORMAT: u32 = 11; + let mut snapshot = MetadataSnapshot::new(9); + snapshot.version = INDEPENDENT_OWNER_GENERATION_FORMAT; + let encoded = snapshot.encode().unwrap(); + assert!(matches!( + MetadataSnapshot::decode(&encoded), + Err(SnapshotError::UnsupportedFormatVersion { found, expected }) + if found == INDEPENDENT_OWNER_GENERATION_FORMAT && expected == SNAPSHOT_FORMAT_VERSION + )); + } + #[test] fn decode_refuses_any_other_format_version() { // Newer and older alike: there is no accepted window, so both directions are diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 00e92531d..ebec5266c 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -947,8 +947,8 @@ where self.consensus.partition_incarnation() } - /// A report covers only an applied barrier on a serving replica. WAL replay - /// or the paired checkpoint must restore this floor before admission. + /// A report covers an applied retirement barrier or terminal history fence + /// on a serving replica. Replay or a checkpoint must restore that proof. /// Retirement barriers require persistence at every durability level. #[must_use] pub fn session_retired(&self, identity: SessionIdentity) -> bool { @@ -957,8 +957,11 @@ where && self.consensus.is_normal() && !self.consensus.is_transferring() && self.consensus.pending_request(identity.client_id).is_none() - && self.required_metadata_frontier >= identity.metadata_watermark - && self.dedup.get_epoch(identity.client_id) != Some(identity.session) + && (self + .history_transition + .is_some_and(|history| history.partition_op <= self.consensus.commit_min()) + || (self.required_metadata_frontier >= identity.metadata_watermark + && self.dedup.get_epoch(identity.client_id) != Some(identity.session))) } fn pending_session_retirement(&self, client_id: u128, session: u64) -> bool { @@ -11941,9 +11944,9 @@ where Ok(()) } - /// Admission and queue promotion share this rule. Session retirement - /// carries no incarnation, and a lifecycle fence names its incarnation in - /// its body, which admission checks. + /// Retirement requests name sessions rather than histories; their prepares + /// carry this partition's incarnation. A lifecycle fence names its + /// incarnation in its body, which admission checks. fn request_history_available(&self, header: &RoutedRequestHeader) -> bool { header.operation == Operation::RetireSession || header.operation.is_partition_lifecycle() @@ -11951,7 +11954,8 @@ where } const fn prepare_history_available(&self, header: &PrepareHeader) -> bool { - header.partition_incarnation == self.created_revision() && !self.history_deleted() + header.partition_incarnation == self.created_revision() + && (matches!(header.operation, Operation::RetireSession) || !self.history_deleted()) } async fn send_prepare_ok(&self, header: &PrepareHeader) -> bool { @@ -15319,6 +15323,102 @@ mod tests { } } + #[compio::test] + async fn given_deleted_history_when_retirement_is_replicated_or_repaired_should_commit_it() { + const FENCE_OP: u64 = 1; + const RETIREMENT_OP: u64 = 2; + let config = repair_config(); + let directory = tempfile::tempdir().unwrap(); + let (mut primary, _) = recording_partition_at(0, 3); + primary.set_partition_dir(directory.path().to_string_lossy().into_owned()); + let transition = TransitionPartitionHistoryRequest { + incarnation: primary.created_revision(), + metadata_op: 10, + }; + primary + .on_request( + partition_control_request( + &primary, + Operation::TransitionPartitionHistory, + &transition.to_bytes(), + transition.metadata_op, + ), + None, + ) + .await; + primary.consensus.advance_commit_max(FENCE_OP); + primary.commit_journal(&config).await; + assert!(primary.history_deleted()); + let bytes = primary.log.journal().inner.repair_entry(FENCE_OP).unwrap(); + let fence = + Message::<PrepareHeader>::try_from(Owned::copy_from_slice(bytes.as_slice())).unwrap(); + let identity = SessionIdentity { + client_id: 1, + session: 1, + metadata_watermark: 20, + }; + assert!(primary.session_retired(identity)); + primary + .on_request(retire_session_request(primary.namespace(), identity), None) + .await; + let bytes = primary + .log + .journal() + .inner + .repair_entry(RETIREMENT_OP) + .expect("retirement must remain journaled after the terminal fence"); + let retirement = + Message::<PrepareHeader>::try_from(Owned::copy_from_slice(bytes.as_slice())).unwrap(); + + for repaired in [false, true] { + let directory = tempfile::tempdir().unwrap(); + let (mut backup, _) = recording_partition_at(1, 3); + backup.set_partition_dir(directory.path().to_string_lossy().into_owned()); + backup.on_replicate(fence.clone()).await; + backup.consensus.advance_commit_max(FENCE_OP); + backup.commit_journal(&config).await; + assert!(backup.history_deleted()); + if repaired { + backup.repair = Some(armed_session(RETIREMENT_OP, 0)); + backup.apply_repaired_prepare(retirement.clone()).await; + } else { + backup.on_replicate(retirement.clone()).await; + } + assert!( + backup.log.journal().inner.holds_op(RETIREMENT_OP), + "retirement after the delete fence must reach the backup, repaired={repaired}" + ); + backup.consensus.advance_commit_max(RETIREMENT_OP); + backup.commit_journal(&config).await; + assert_eq!(backup.consensus.commit_min(), RETIREMENT_OP); + assert!(backup.session_retired(identity)); + assert!(backup.fatal().is_none()); + } + } + + #[compio::test] + async fn given_unpublished_delete_fence_when_checking_retirement_should_wait_for_commit() { + let mut partition = test_partition(); + let identity = SessionIdentity { + client_id: 1, + session: 1, + metadata_watermark: 20, + }; + partition + .apply_history_transition( + TransitionPartitionHistoryRequest { + incarnation: partition.created_revision(), + metadata_op: 10, + }, + 1, + ) + .unwrap(); + assert!(!partition.session_retired(identity)); + partition.consensus.advance_commit_max(1); + partition.consensus.advance_commit_min(1); + assert!(partition.session_retired(identity)); + } + #[compio::test] async fn given_queued_session_retirement_when_promoted_at_a_live_incarnation_should_prepare_it() { diff --git a/core/partitions/src/io.rs b/core/partitions/src/io.rs index ecf8c814a..1e516c7e1 100644 --- a/core/partitions/src/io.rs +++ b/core/partitions/src/io.rs @@ -1271,9 +1271,8 @@ impl OffsetIoJob { #[allow(clippy::future_not_send)] pub(crate) async fn execute(mut self) -> OffsetIoResult { let value = async { - // First only as defense in depth. WAL replay of a lost store, boot - // dropping a marker beside any offset file and the transfer bound - // `emptied_at <= commit_op` carry the crash safety. + // A lost store file must not leave an older marker claiming the + // table is empty. Keep marker removal strict before writing it. if let Some(marker) = &self.retired_marker { persist_emptied_marker_with_storage(&DiskStorage, marker, 0).await?; } @@ -1337,11 +1336,13 @@ impl OffsetDeleteIoJob { #[allow(clippy::future_not_send)] pub(crate) async fn execute(self) -> Result<bool, IggyError> { let removed = delete_persisted_offset(&self.path).await?; - // After the unlink only as defense in depth. WAL replay of the delete, - // boot dropping a marker beside any offset file and the transfer bound - // `emptied_at <= commit_op` carry the crash safety. - if let Some((marker, op)) = &self.emptied_marker { - persist_emptied_marker_with_storage(&DiskStorage, marker, *op).await?; + // A missing marker after restart loses empty-table transfer proof, not + // the committed delete. Replay may find the key already absent and + // cannot be relied on to recreate the marker. + if let Some((marker, op)) = &self.emptied_marker + && let Err(error) = persist_emptied_marker_with_storage(&DiskStorage, marker, *op).await + { + warn!(%marker, op, %error, "could not persist emptied offset table marker"); } Ok(removed) } @@ -1468,6 +1469,69 @@ mod tests { } } + #[compio::test] + async fn given_failed_emptied_marker_when_offset_delete_succeeds_should_preserve_success() { + let directory = tempfile::tempdir().unwrap(); + let offset = directory.path().join("offset"); + let marker = directory.path().join("emptied"); + std::fs::write(&offset, crate::offset_storage::encode_offset_record(7)).unwrap(); + std::fs::create_dir(&marker).unwrap(); + let job = OffsetDeleteIoJob { + path: offset.to_str().unwrap().to_owned(), + emptied_marker: Some((marker.to_str().unwrap().to_owned(), 9)), + }; + + assert_eq!(job.execute().await, Ok(true)); + assert!(!offset.exists()); + assert!(marker.is_dir()); + } + + #[compio::test] + async fn given_failed_marker_clear_when_storing_should_preserve_error_without_offset() { + let directory = tempfile::tempdir().unwrap(); + let offset = directory.path().join("offset"); + let marker = directory.path().join("emptied"); + std::fs::create_dir(&marker).unwrap(); + let marker_path = marker.to_str().unwrap().to_owned(); + let job = OffsetIoJob { + path: offset.to_str().unwrap().to_owned(), + offset: 7, + fold_max: false, + persisted: true, + owner: OffsetFileOwner::Uncached, + file: None, + permit: None, + retired_marker: Some(marker_path.clone()), + }; + + assert_eq!( + job.execute().await.outcome, + Err(IggyError::CannotDeleteConsumerOffsetFile(marker_path)) + ); + assert!(!offset.exists()); + assert!(marker.is_dir()); + } + + #[compio::test] + async fn given_failed_offset_delete_when_executing_should_preserve_error_without_marker() { + let directory = tempfile::tempdir().unwrap(); + let offset = directory.path().join("offset"); + let marker = directory.path().join("emptied"); + std::fs::create_dir(&offset).unwrap(); + let path = offset.to_str().unwrap().to_owned(); + let job = OffsetDeleteIoJob { + path: path.clone(), + emptied_marker: Some((marker.to_str().unwrap().to_owned(), 9)), + }; + + assert_eq!( + job.execute().await, + Err(IggyError::CannotDeleteConsumerOffsetFile(path)) + ); + assert!(offset.is_dir()); + assert!(!marker.exists()); + } + #[compio::test] async fn given_failed_offset_scan_when_replacing_history_should_remove_known_stale_files() { let directory = tempfile::tempdir().unwrap(); diff --git a/core/server/src/consumer_group/liveness.rs b/core/server/src/consumer_group/liveness.rs index ee5c8c48c..744f859aa 100644 --- a/core/server/src/consumer_group/liveness.rs +++ b/core/server/src/consumer_group/liveness.rs @@ -455,7 +455,10 @@ async fn retire_session<B, MJ, S, SB>( let status = shard .partition_read( *namespace, - shard::PartitionRead::SessionRetired { identity }, + shard::PartitionRead::SessionRetired { + identity, + created_revision: *created_revision, + }, ) .await; let retired = match status { @@ -521,7 +524,10 @@ async fn retire_session<B, MJ, S, SB>( shard .partition_read( *namespace, - shard::PartitionRead::SessionRetired { identity } + shard::PartitionRead::SessionRetired { + identity, + created_revision: *created_revision, + } ) .await, Some(shard::PartitionReadReply::SessionRetired(true)) diff --git a/core/server/src/dispatch/failure.rs b/core/server/src/dispatch/failure.rs index 4024bb4a4..9f5b3e52d 100644 --- a/core/server/src/dispatch/failure.rs +++ b/core/server/src/dispatch/failure.rs @@ -650,6 +650,7 @@ mod tests { &Rc::new(ClientDrain::new(&server_config)), TRANSPORT, frame, + None, ) .await; @@ -896,6 +897,7 @@ mod tests { &Rc::new(ClientDrain::new(&server_config)), TRANSPORT, request.into_generic(), + None, ) .await; diff --git a/core/server/src/dispatch/host.rs b/core/server/src/dispatch/host.rs index 37aebc265..64f3cec3c 100644 --- a/core/server/src/dispatch/host.rs +++ b/core/server/src/dispatch/host.rs @@ -24,9 +24,10 @@ //! connection out. All connections installed on the shard dispatch through //! the same instance, which keeps disconnect cleanup equal across transports: //! the destination shard installs it on delegated TCP/WS/TCP-TLS/WSS -//! connections, and shard 0 also on its local QUIC connections. A drain runs -//! the frames that later frames depend on in arrival order, and every other -//! read in a task of its own (see `crate::dispatch::ClientDrain`). +//! connections, and shard 0 also on its local QUIC connections. A drain orders +//! session changes and partition admissions. Later non-PING frames wait for a +//! preceding metadata commit; independent reads run in their own tasks (see +//! `crate::dispatch::ClientDrain`). //! //! The host also owns a clone of the bus, so every adapter the bus installs, //! on client and replica connections alike, keeps the bus alive: accepted, diff --git a/core/server/src/dispatch/mod.rs b/core/server/src/dispatch/mod.rs index 3a5316e09..6e81a5ca6 100644 --- a/core/server/src/dispatch/mod.rs +++ b/core/server/src/dispatch/mod.rs @@ -47,7 +47,9 @@ use crate::dispatch::failure::{ FrameChannel, send_deny_reply, send_eviction, send_host_frame, send_pre_consensus_deny, send_unbound_deny_reply, send_undeclared_operation_deny, }; -use crate::dispatch::partition::{dispatch_partition_request, handle_delete_segments_request}; +use crate::dispatch::partition::{ + dispatch_partition_request, handle_delete_segments_request, progress_key, +}; use crate::dispatch::reads::handle_non_replicated_request; use crate::dispatch::session_ops::{handle_login_register_request, handle_logout_request}; use crate::dispatch::submit::{committed_reply_commit, submit_client_request_on_owner}; @@ -56,7 +58,7 @@ use crate::rewrite::{RewriteDeny, tcp_chain}; use crate::session_manager::{ConnectionContext, SessionManager}; use crate::shell::{ShellBus, ShellShard, ShellShardHandle}; use crate::wire::verify_request_checksum; -use ahash::AHashMap; +use ahash::{AHashMap, AHashSet}; use configs::server::ServerConfig; use consensus::MetadataHandle; use futures::future::{Either, select}; @@ -104,6 +106,7 @@ type ClientRequestQueues = Rc<RefCell<AHashMap<u128, VecDeque<Message<GenericHea /// any SDK's in-flight window, so it only ever fires under a genuine stall. const MAX_QUEUED_CLIENT_REQUESTS: usize = 1024; type ActiveClientRequests = Rc<RefCell<AHashMap<u128, Rc<ClientDrain>>>>; +type ConsumerProgress = (u64, u8, usize); // BindSession authenticates with the registered secret, never with the // public client id and epoch alone. It retains the original retry identity; @@ -253,12 +256,14 @@ impl Drop for ActiveDrainSlot { /// it spawned. /// /// The drain takes the frames of the connection in arrival order and runs the -/// prologue of each. It finishes a frame that later frames depend on before it -/// takes the next one: a login, a logout, a `BIND_SESSION`, the rewrite of a -/// metadata write, or a partition admission. A `PING` finishes there too, -/// because it never waits. Every other read, every poll and every wait for a -/// metadata commit runs in a task of its own, so a held frame does not stall -/// the frames behind it. +/// prologue of each. Login, logout, `BIND_SESSION` and partition admission +/// finish on the drain. Metadata commits run in tasks, but later non-PING +/// frames wait in the queue until the metadata lane releases, before they +/// capture session state or resolve a namespace. PING can pass that queue. +/// Polls and explicit offset writes share a lane per resolved namespace and +/// consumer until their reply. Independent requests can pass a queued conflict, +/// but cannot pass a queued session or metadata barrier. A PING still waits +/// behind a frame already suspended in admission or binding. /// /// `capacity` caps those tasks at `message_bus.client_queue_capacity`, the /// outbound queue size of a connection, so their replies never overflow that @@ -279,6 +284,7 @@ pub(in crate::dispatch) struct ClientDrain { in_flight: Cell<usize>, /// Whether a frame holds the metadata lane. metadata_write: Cell<bool>, + progress: RefCell<AHashSet<ConsumerProgress>>, /// The drain task, while it waits for a frame or for room to admit one. parked: Cell<Option<Waker>>, } @@ -293,11 +299,12 @@ enum Admission { } impl ClientDrain { - pub(in crate::dispatch) const fn new(server_config: &ServerConfig) -> Self { + pub(in crate::dispatch) fn new(server_config: &ServerConfig) -> Self { Self { capacity: server_config.message_bus.client_queue_capacity, in_flight: Cell::new(0), metadata_write: Cell::new(false), + progress: RefCell::default(), parked: Cell::new(None), } } @@ -323,6 +330,17 @@ impl ClientDrain { InFlight { drain: Rc::clone(self), metadata_write, + progress: None, + } + } + + fn admit_progress(self: &Rc<Self>, progress: ConsumerProgress) -> InFlight { + self.progress.borrow_mut().insert(progress); + self.in_flight.set(self.in_flight.get() + 1); + InFlight { + drain: Rc::clone(self), + metadata_write: false, + progress: Some(progress), } } @@ -350,9 +368,10 @@ impl ClientDrain { /// One request task counted by its [`ClientDrain`]. The task owns it, so the /// count drops when the task ends or the runtime drops it. -struct InFlight { +pub struct InFlight { drain: Rc<ClientDrain>, metadata_write: bool, + progress: Option<ConsumerProgress>, } impl Drop for InFlight { @@ -361,6 +380,9 @@ impl Drop for InFlight { if self.metadata_write { self.drain.metadata_write.set(false); } + if let Some(progress) = self.progress { + self.drain.progress.borrow_mut().remove(&progress); + } self.drain.wake(); } } @@ -382,7 +404,19 @@ async fn drain_client_requests<B, MJ, S, SB>( SB: SuperblockStore + 'static, { loop { - if let Some(message) = pop_next_client_request(&queues, client_id) { + let next = pop_next_client_request(&queues, client_id, |message| { + if is_ping(message) { + return true; + } + if drain.metadata_write.get() { + return false; + } + progress_key(&shard, message).is_none_or(|key| { + drain.in_flight.get() < drain.capacity && !drain.progress.borrow().contains(&key) + }) + }); + if let Some(message) = next { + let progress = progress_key(&shard, &message).map(|key| drain.admit_progress(key)); handle_client_request( &shard, &sessions, @@ -391,6 +425,7 @@ async fn drain_client_requests<B, MJ, S, SB>( drain, client_id, message, + progress, ) .await; } else if drain.in_flight.get() == 0 { @@ -404,6 +439,7 @@ async fn drain_client_requests<B, MJ, S, SB>( fn pop_next_client_request( queues: &ClientRequestQueues, client_id: u128, + ready: impl Fn(&Message<GenericHeader>) -> bool, ) -> Option<Message<GenericHeader>> { // Freed as soon as it drains empty. Keeping it would cost the connection // one `VecDeque` (at its peak depth, since `pop_front` never gives @@ -413,13 +449,53 @@ fn pop_next_client_request( // process exit. let mut queues = queues.borrow_mut(); let queue = queues.get_mut(&client_id)?; - let message = queue.pop_front(); + let mut waiting = false; + let mut barrier = false; + let index = queue.iter().position(|message| { + if is_ping(message) { + return true; + } + if barrier { + return false; + } + let orders_followers = message + .try_as_typed::<RequestHeader>() + .map_or(true, |request| { + let header = request.header(); + if header.operation == Operation::NonReplicated { + header.reserved[..4] == BIND_SESSION_CODE.to_le_bytes() + } else { + !header.operation.is_partition() + } + }); + if waiting && orders_followers { + barrier = true; + return false; + } + if ready(message) { + return true; + } + waiting = true; + barrier = orders_followers; + false + })?; + let message = queue.remove(index); if queue.is_empty() { queues.remove(&client_id); } message } +fn is_ping(message: &Message<GenericHeader>) -> bool { + message + .try_as_typed::<RequestHeader>() + .is_ok_and(|request| { + let header = request.header(); + header.operation == Operation::NonReplicated + && header.reserved[..4] == PING_CODE.to_le_bytes() + }) +} + /// Where the funnel routes a client request. Derived by [`classify`], which /// IS the routing: [`handle_client_request`] matches on its result. The /// variant ORDER mirrors the order of the checks inside [`classify`], and @@ -559,7 +635,11 @@ where } } -#[allow(clippy::future_not_send, clippy::too_many_lines)] +#[allow( + clippy::future_not_send, + clippy::too_many_arguments, + clippy::too_many_lines +)] async fn handle_client_request<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, sessions: &Rc<RefCell<SessionManager>>, @@ -568,6 +648,7 @@ async fn handle_client_request<B, MJ, S, SB>( drain: &Rc<ClientDrain>, transport_client_id: u128, message: Message<iggy_binary_protocol::GenericHeader>, + progress: Option<InFlight>, ) where B: ShellBus, MJ: JournalHandle + 'static, @@ -707,7 +788,10 @@ async fn handle_client_request<B, MJ, S, SB>( .await; return; } - let in_flight = drain.admit(Admission::Cap).await; + let in_flight = match progress { + Some(progress) => progress, + None => drain.admit(Admission::Cap).await, + }; let task_shard = Rc::clone(shard); let task_sessions = Rc::clone(sessions); let task_config = Arc::clone(server_config); @@ -811,6 +895,7 @@ async fn handle_client_request<B, MJ, S, SB>( user_id, consumer_session, None, + progress, ) .await; } @@ -1158,6 +1243,7 @@ mod tests { &Rc::new(ClientDrain::new(&server_config)), ROOT_TRANSPORT, held_get_me(1, UNAPPLIED_OP), + None, ) .await; settle().await; @@ -1716,6 +1802,7 @@ mod tests { &Rc::new(ClientDrain::new(&server_config)), TRANSPORT, metadata_read(), + None, ) .await; let replies = bus.client_replies.borrow(); @@ -1977,6 +2064,7 @@ mod tests { &Rc::new(ClientDrain::new(&server_config)), TRANSPORT, non_replicated_request(TRANSPORT, code), + None, ) .await; let replies = bus.client_replies.borrow(); @@ -2018,6 +2106,7 @@ mod tests { &Rc::new(ClientDrain::new(&server_config)), TRANSPORT, wire_request(Operation::CreateStream, TRANSPORT, 1, 1, &[]).into_generic(), + None, ) .await; let replies = bus.client_replies.borrow(); @@ -2054,6 +2143,7 @@ mod tests { &Rc::new(ClientDrain::new(&server_config)), TRANSPORT, non_replicated_request(TRANSPORT, PING_CODE), + None, ) .await; let replies = bus.client_replies.borrow(); @@ -2098,6 +2188,7 @@ mod tests { &Rc::new(ClientDrain::new(&server_config)), TRANSPORT, message.into_generic(), + None, ) .await; let replies = bus.client_replies.borrow(); @@ -2261,8 +2352,106 @@ mod tests { assert_eq!(reply_status(&replies[1].1), 0); } - /// At the in-flight cap the drain waits for a request task to end, and so - /// does every frame behind it. + #[compio::test] + async fn given_pending_metadata_when_read_and_ping_queue_should_answer_ping_then_wake_read() { + const READ_REQUEST: u64 = 1; + const PING_REQUEST: u64 = 2; + let config = Arc::new(ServerConfig::default()); + let (bus, shard, host) = root_host((*config).clone()); + bind_root_session(&shard, host.sessions()).await; + let drain = Rc::new(ClientDrain::new(&config)); + let metadata = drain.admit(Admission::MetadataLane).await; + let queues = Rc::new(RefCell::new(AHashMap::from_iter([( + ROOT_TRANSPORT, + VecDeque::from([ + held_get_me(READ_REQUEST, 0), + root_read(PING_CODE, PING_REQUEST).into_generic(), + ]), + )]))); + let draining = drain_client_requests( + shard, + Rc::clone(host.sessions()), + config, + 1, + queues, + &drain, + ROOT_TRANSPORT, + ); + futures::pin_mut!(draining); + assert!(futures::poll!(draining.as_mut()).is_pending()); + { + let replies = bus.client_replies.borrow(); + assert_eq!(replies.len(), 1, "the dependent read must remain queued"); + assert_eq!(reply_request(&replies[0].1), PING_REQUEST); + } + drop(metadata); + draining.await; + let replies = bus.client_replies.borrow(); + assert_eq!( + replies.len(), + 2, + "release must wake the read without another frame" + ); + assert_eq!(reply_request(&replies[1].1), READ_REQUEST); + } + + #[test] + fn given_queued_progress_conflict_when_session_barrier_follows_should_only_let_ping_pass() { + const BLOCKED: u64 = 1; + const BARRIER: u64 = 2; + const READ: u64 = 3; + const PING: u64 = 4; + for operation in [ + Operation::Logout, + Operation::DeleteTopic, + Operation::NonReplicated, + ] { + let barrier = if operation == Operation::NonReplicated { + root_read(BIND_SESSION_CODE, BARRIER).into_generic() + } else { + wire_request(operation, ROOT_CLIENT, ROOT_SESSION, BARRIER, &[]).into_generic() + }; + let queues = Rc::new(RefCell::new(AHashMap::from_iter([( + ROOT_TRANSPORT, + VecDeque::from([ + wire_request( + Operation::StoreConsumerOffset, + ROOT_CLIENT, + ROOT_SESSION, + BLOCKED, + &[], + ) + .into_generic(), + barrier, + held_get_me(READ, 0), + root_read(PING_CODE, PING).into_generic(), + ]), + )]))); + let ready = |message: &Message<GenericHeader>| { + message + .try_as_typed::<RequestHeader>() + .unwrap() + .header() + .request + != BLOCKED + }; + let ping = pop_next_client_request(&queues, ROOT_TRANSPORT, ready).unwrap(); + assert!(is_ping(&ping)); + assert!(pop_next_client_request(&queues, ROOT_TRANSPORT, ready).is_none()); + for expected in [BLOCKED, BARRIER, READ] { + let message = pop_next_client_request(&queues, ROOT_TRANSPORT, |_| true).unwrap(); + assert_eq!( + message + .try_as_typed::<RequestHeader>() + .unwrap() + .header() + .request, + expected + ); + } + } + } + #[compio::test] async fn given_a_full_in_flight_cap_when_another_frame_arrives_should_serve_it_after_a_slot_frees() { @@ -2617,6 +2806,7 @@ mod tests { &Rc::new(ClientDrain::new(&server_config)), HOME_TRANSPORT, request.into_generic(), + None, ) .await; await_replies(bus, 1).await; @@ -2699,13 +2889,13 @@ mod tests { .or_default() .push_back(non_replicated_request(CLIENT, PING_CODE)); - assert!(pop_next_client_request(&queues, CLIENT).is_some()); + assert!(pop_next_client_request(&queues, CLIENT, |_| true).is_some()); assert!( queues.borrow().is_empty(), "the entry must be freed as soon as it drains empty" ); assert!( - pop_next_client_request(&queues, CLIENT).is_none(), + pop_next_client_request(&queues, CLIENT, |_| true).is_none(), "a drained client has nothing left to pop" ); } diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs index b2f591784..08e99c4e2 100644 --- a/core/server/src/dispatch/partition.rs +++ b/core/server/src/dispatch/partition.rs @@ -30,6 +30,7 @@ use crate::dispatch::failure::{ send_non_replicated_deny, send_result_rejection, }; use crate::dispatch::submit::submit_client_request_on_owner; +use crate::dispatch::{ConsumerProgress, InFlight}; use crate::namespace::{resolve_partition_namespace, resolve_partition_request_namespace}; use crate::reply_frame::{ build_consumer_offset_body, build_polled_messages_reply, current_metadata_commit, @@ -40,16 +41,21 @@ use bytes::Bytes; use consensus::client_table::SessionAttachment; use consensus::{MetadataHandle, PartitionsHandle}; use iggy_binary_protocol::PrepareHeader; +use iggy_binary_protocol::codes::{ + GET_CONSUMER_OFFSET_CODE, POLL_MESSAGES_CODE, POLL_MESSAGES_ON_PRIMARY_CODE, +}; use iggy_binary_protocol::primitives::consumer::WireConsumer; use iggy_binary_protocol::primitives::partition_history::PartitionContext; use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy; -use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest; +use iggy_binary_protocol::requests::consumer_offsets::{ + DeleteConsumerOffsetRequest, GetConsumerOffsetRequest, StoreConsumerOffsetRequest, +}; use iggy_binary_protocol::requests::messages::PollMessagesRequest; use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest; use iggy_binary_protocol::responses::messages::poll_messages::PollMessagesResponseHeader; use iggy_binary_protocol::{ - KIND_CONSUMER, KIND_CONSUMER_GROUP, KIND_EXTERNAL_GROUP, Operation, RoutedRequestHeader, - WireDecode, WireEncode, WireIdentifier, + GenericHeader, KIND_CONSUMER, KIND_CONSUMER_GROUP, KIND_EXTERNAL_GROUP, Operation, + RequestHeader, RoutedRequestHeader, WireDecode, WireEncode, WireIdentifier, }; use iggy_common::{ConsumerKind, IggyError, PollingStrategy, RESYNC_REQUIRED_PARTITION_SENTINEL}; use journal::superblock::SuperblockStore; @@ -105,6 +111,7 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>( acting_user_id: Option<u32>, consumer_session: Option<(u128, SessionAttachment)>, resolved_namespace: Option<u64>, + progress: Option<InFlight>, ) where B: ShellBus, MJ: JournalHandle + 'static, @@ -290,6 +297,7 @@ pub async fn dispatch_partition_request<B, MJ, S, SB>( transport_client_id, &header, attachment, + progress, ) .await; } @@ -310,6 +318,7 @@ async fn relay_partition_reply<B, MJ, S, SB>( transport_client_id: u128, header: &RoutedRequestHeader, attachment: Option<ConsumerAttachment>, + progress: Option<InFlight>, ) where B: ShellBus, MJ: JournalHandle + 'static, @@ -337,6 +346,7 @@ async fn relay_partition_reply<B, MJ, S, SB>( // Through the bus, not the runtime directly: the simulator supplies its // own executor and virtual clock. shard.bus.clone().spawn(async move { + let _progress = progress; let Some(reply) = shard.await_partition_submit(ticket).await else { // Abandoned or expired. Deliberately silent: the outcome is // unknown, and a synthesized failure could contradict a write that @@ -369,6 +379,71 @@ async fn relay_partition_reply<B, MJ, S, SB>( }); } +/// The offset key shared by polls and explicit progress writes on a connection. +/// Failed resolution is left to the normal handler's authorization and error path. +pub(super) fn progress_key<B, MJ, S, SB>( + shard: &Rc<ShellShard<B, MJ, S, SB>>, + message: &Message<GenericHeader>, +) -> Option<ConsumerProgress> +where + B: ShellBus, + MJ: JournalHandle + 'static, + MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, + S: 'static, + SB: SuperblockStore + 'static, +{ + let request = message.try_as_typed::<RequestHeader>().ok()?; + let wire = match request.header().operation { + Operation::NonReplicated + if request.header().reserved[..4] == GET_CONSUMER_OFFSET_CODE.to_le_bytes() => + { + GetConsumerOffsetRequest::decode_from(request.body()).ok()? + } + Operation::NonReplicated + if request.header().reserved[..4] == POLL_MESSAGES_CODE.to_le_bytes() + || request.header().reserved[..4] + == POLL_MESSAGES_ON_PRIMARY_CODE.to_le_bytes() => + { + let poll = PollMessagesRequest::decode_from(request.body()).ok()?; + GetConsumerOffsetRequest { + consumer: poll.consumer, + stream_id: poll.stream_id, + topic_id: poll.topic_id, + partition_id: poll.partition_id, + } + } + Operation::StoreConsumerOffset => { + let store = StoreConsumerOffsetRequest::decode_from(request.body()).ok()?; + GetConsumerOffsetRequest { + consumer: store.consumer, + stream_id: store.stream_id, + topic_id: store.topic_id, + partition_id: store.partition_id, + } + } + Operation::DeleteConsumerOffset => { + let delete = DeleteConsumerOffsetRequest::decode_from(request.body()).ok()?; + GetConsumerOffsetRequest { + consumer: delete.consumer, + stream_id: delete.stream_id, + topic_id: delete.topic_id, + partition_id: delete.partition_id, + } + } + _ => return None, + }; + let (namespace, _, read) = resolve_consumer_offset_request(shard, &wire).ok()??; + let (kind, consumer_id) = match read { + PartitionRead::ConsumerOffset { consumer } => match consumer { + PollingConsumer::Consumer(id, _) => (KIND_CONSUMER, id), + PollingConsumer::ConsumerGroup(id, _) => (KIND_CONSUMER_GROUP, id), + }, + PartitionRead::ExternalGroupOffset { group_id } => (KIND_EXTERNAL_GROUP, group_id as usize), + _ => return None, + }; + Some((namespace.inner(), kind, consumer_id)) +} + pub fn capture_offset_attachment( streams: &metadata::stm::stream::Streams, namespace: IggyNamespace, @@ -720,10 +795,8 @@ pub(super) fn empty_poll_fallback(partition_id: u32) -> (Bytes, FrameChannel) { /// denies with a nonzero status instead, so an addressing typo cannot read /// back as a fresh consumer. A read that its owner does not answer in time, or /// whose owner does not hold the partition, denies with `TransientNotAccepted`. -// TODO(hubcio): plain local partition_read with no primary gate, so a -// follower answers from its own (possibly lagging) offset state. Needs the -// same is-caught-up-primary gate the auto-commit path has, or an explicit -// read-from-follower contract. +// TODO(hubcio): primary-gate binary offset reads when the active Rust clients +// route GET_CONSUMER_OFFSET to the partition primary. #[allow(clippy::future_not_send)] pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, @@ -1734,6 +1807,7 @@ mod tests { Some(DEFAULT_ROOT_USER_ID), None, Some(namespace), + None, ) .await; let _ = updater.await; @@ -1980,6 +2054,7 @@ mod tests { Some(DEFAULT_ROOT_USER_ID), None, None, + None, ) .await; let replies = bus.client_replies.borrow(); @@ -2921,6 +2996,7 @@ mod tests { Some(DEFAULT_ROOT_USER_ID), None, None, + None, ) .await; diff --git a/core/server/src/http/forward.rs b/core/server/src/http/forward.rs index 0e6c53bab..beae5ffc0 100644 --- a/core/server/src/http/forward.rs +++ b/core/server/src/http/forward.rs @@ -48,8 +48,8 @@ //! primary's pipeline: a connect-phase failure, and for a metadata relay a //! 503 whose body carries the `TransientNotAccepted` or `NotPrimary` code or //! a 307 from a stale target, for a partition relay a 503 carrying -//! `NotPrimary`. Every -//! other outcome - including a 503 carrying `TransientNotCommitted`, whose op +//! `NotPrimary`. Every other outcome - including a 503 carrying +//! `TransientNotCommitted`, whose op //! may still commit - is relayed as-is, because re-issuing it under a fresh //! session would defeat the consensus dedup and double-apply the op. //! - A partition read changes no state, so a transport failure after its send @@ -702,17 +702,25 @@ async fn attempt( classify_forwarded_reply(response, method, kind).await }; match compio::time::timeout(timeout, attempt).await { - // Elapsed: the request may be mid-commit on the primary. Outcome - // unknown, so never retried - 504, same contract as a local commit - // wait that timed out. - Err(_elapsed) => AttemptOutcome::Relay(gateway_timeout_response( - "forward_timeout", - "the primary did not answer the forwarded request in time; the outcome is unknown", - )), + Err(_elapsed) => AttemptOutcome::Relay(forward_timeout(kind)), Ok(outcome) => outcome, } } +fn forward_timeout(kind: RelayKind) -> Response { + if kind == RelayKind::Partition(PartitionEffect::Read) { + return with_retry_after(error_response( + StatusCode::SERVICE_UNAVAILABLE, + "forward_timeout", + "the primary did not answer the forwarded read in time; retry", + )); + } + gateway_timeout_response( + "forward_timeout", + "the primary did not answer the forwarded request in time; the outcome is unknown", + ) +} + async fn classify_forwarded_reply( response: cyper::Response, method: &Method, @@ -1074,6 +1082,19 @@ mod tests { use std::io::{Read, Write}; + #[test] + fn given_relay_timeout_when_request_is_read_only_should_report_retryable_unavailability() { + let response = forward_timeout(RelayKind::Partition(PartitionEffect::Read)); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + assert!(response.headers().contains_key(RETRY_AFTER)); + for kind in [ + RelayKind::Metadata, + RelayKind::Partition(PartitionEffect::Write), + ] { + assert_eq!(forward_timeout(kind).status(), StatusCode::GATEWAY_TIMEOUT); + } + } + #[test] fn linearizable_query_detected_only_on_exact_pair() { assert!(wants_linearizable(Some("consistency=linearizable"))); diff --git a/core/server/src/http/handlers.rs b/core/server/src/http/handlers.rs index 674e124a8..e8e06bbd4 100644 --- a/core/server/src/http/handlers.rs +++ b/core/server/src/http/handlers.rs @@ -1337,10 +1337,10 @@ pub(in crate::http) async fn poll_messages( .map_err(ReadError::Rejected)?; let reply = SendWrapper::new(state.shard.partition_read( namespace, - PartitionRead::Poll { + PartitionRead::UnboundPollOnPrimary { consumer, args, - metadata: Some(metadata), + metadata, }, )) .await; @@ -1445,6 +1445,15 @@ pub(in crate::http) async fn get_consumer_offset( { return Ok(response); } + let read = match read { + PartitionRead::ConsumerOffset { consumer } => { + PartitionRead::ConsumerOffsetOnPrimary { consumer } + } + PartitionRead::ExternalGroupOffset { group_id } => { + PartitionRead::ExternalGroupOffsetOnPrimary { group_id } + } + _ => return Err(ReadError::Rejected(IggyError::InvalidCommand)), + }; let reply = SendWrapper::new(state.shard.partition_read(namespace, read)).await; consumer_offset_reply(partition_id, reply).map(IntoResponse::into_response) } diff --git a/core/server/src/http/submit.rs b/core/server/src/http/submit.rs index cdb596f91..066c63054 100644 --- a/core/server/src/http/submit.rs +++ b/core/server/src/http/submit.rs @@ -483,6 +483,7 @@ async fn partition_write( Some(session.user_id), Some((session.client_id, attachment)), Some(namespace), + None, ) .await; if let Some(dispatched) = dispatched.take() { diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index f5bf10e96..7215ba796 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -365,6 +365,7 @@ pub enum PartitionRead { Primary, SessionRetired { identity: iggy_binary_protocol::requests::system::SessionIdentity, + created_revision: u64, }, /// A bound poll. A group read is served only to the partition's installed /// owner. @@ -373,6 +374,12 @@ pub enum PartitionRead { args: PollingArgs, attachment: poll::ConsumerAttachment, }, + /// An HTTP poll with no bound session, served only by the primary. + UnboundPollOnPrimary { + consumer: PollingConsumer, + args: PollingArgs, + metadata: metadata::stm::stream::PollMetadata, + }, /// A read with no owner check. In production only plain consumers use it. /// Tests and the simulator also read groups through it. Poll { @@ -383,10 +390,16 @@ pub enum PartitionRead { ConsumerOffset { consumer: PollingConsumer, }, + ConsumerOffsetOnPrimary { + consumer: PollingConsumer, + }, /// An external group's stored offset. Not a `PollingConsumer`: the kind is never polled. ExternalGroupOffset { group_id: u32, }, + ExternalGroupOffsetOnPrimary { + group_id: u32, + }, /// Cooperative-rebalance classification: the group's last-polled and /// committed offsets on this partition, so the join enrichment can tell an /// in-flight partition (committed < last-polled) from a never-polled/drained @@ -403,12 +416,27 @@ pub enum PartitionRead { } impl PartitionRead { + pub(crate) const fn requires_primary(&self) -> bool { + matches!( + self, + Self::PollOnPrimary { .. } + | Self::UnboundPollOnPrimary { .. } + | Self::ConsumerOffsetOnPrimary { .. } + | Self::ExternalGroupOffsetOnPrimary { .. } + ) + } + /// Whether the read returns a stored consumer offset, or polls from one. #[must_use] pub(crate) const fn reads_stored_offset(&self) -> bool { match self { - Self::ConsumerOffset { .. } | Self::ExternalGroupOffset { .. } => true, - Self::PollOnPrimary { args, .. } | Self::Poll { args, .. } => { + Self::ConsumerOffset { .. } + | Self::ConsumerOffsetOnPrimary { .. } + | Self::ExternalGroupOffset { .. } + | Self::ExternalGroupOffsetOnPrimary { .. } => true, + Self::PollOnPrimary { args, .. } + | Self::UnboundPollOnPrimary { args, .. } + | Self::Poll { args, .. } => { matches!(args.kind, PollingKind::Next) } Self::Primary @@ -4934,7 +4962,6 @@ where && let Some(from_op) = partition_view_divergence_from(self.id, partition, pending.as_ref()) { - partition.consensus().settle_view_log_proof(); partition.defer_view_truncation(from_op, actions, Some(header.replica)); return; } @@ -6080,7 +6107,6 @@ where partition_view_divergence_from(self.id, partition, Some(&pending)) { if partition.has_io_dispatcher() { - partition.consensus().settle_view_log_proof(); partition.defer_view_truncation(from_op, Vec::new(), None); return; } diff --git a/core/shard/src/partition_io.rs b/core/shard/src/partition_io.rs index eed6db7e9..fafd3e5b9 100644 --- a/core/shard/src/partition_io.rs +++ b/core/shard/src/partition_io.rs @@ -705,6 +705,7 @@ where } } partitions::PartitionIoStep::ViewApplied { actions, peer } => { + partition.consensus().settle_view_log_proof(); if let Some(peer) = peer { self.finish_partition_view_adoption(namespace, peer, actions) .await; @@ -4198,6 +4199,73 @@ mod tests { assert!(!partition.consensus().view_log_unproven()); } + #[compio::test] + async fn given_deferred_view_truncation_when_resuming_should_settle_proof_only_after_success() { + const APPLIED: u64 = 2; + const LOCAL_HEAD: u64 = 4; + const NEW_HEAD: u64 = 6; + const NEW_VIEW: u32 = 3; + for truncate_fails in [false, true] { + let bus = Rc::new(IggyMessageBus::new(0)); + let (owner, _sender) = test_owner(&bus, None); + let namespace = IggyNamespace::new(0, 0, 0); + let prepares = primary_prepares(&bus, namespace, NEW_HEAD).await; + let mut partition = + repairing_replica(&bus, namespace, (REPAIR_BACKUP, 3), REPAIR_PRIMARY, 0, 0); + partition.repair = None; + for prepare in prepares.iter().take(usize::try_from(LOCAL_HEAD).unwrap()) { + partition.on_replicate(prepare.clone()).await; + } + partition.consensus().advance_commit_max(APPLIED); + partition + .commit_journal(owner.plane.partitions().config()) + .await; + let partitions = owner.plane.partitions(); + partitions.remove(&namespace).unwrap(); + partitions.insert(namespace, partition); + partitions.set_io_notifier( + owner.partition_io.notifier(), + owner.partition_io.limits.bytes_max(), + ); + owner + .on_start_view(start_view_suffix( + namespace, + NEW_VIEW, + NEW_HEAD, + NEW_HEAD, + &[*prepares.last().unwrap().header()], + )) + .await; + let partition = partitions.get_mut_by_ns(&namespace).unwrap(); + assert!(partition.view_truncation_pending()); + assert!(partition.consensus().view_log_unproven()); + if truncate_fails { + partition.log.active_segment_mut().size = IggyByteSize::from(u64::MAX); + } + + for _ in 0..crate::router::COOPERATIVE_EVENT_BUDGET { + owner.service_partition_io().await; + if !partitions + .get_by_ns(&namespace) + .unwrap() + .view_truncation_pending() + { + break; + } + } + + let partition = partitions.get_mut_by_ns(&namespace).unwrap(); + assert!(!partition.view_truncation_pending()); + assert_eq!(partition.consensus().view_log_unproven(), truncate_fails); + partition.commit_journal(partitions.config()).await; + assert_eq!(partition.consensus().commit_min(), APPLIED); + assert_eq!( + partition.log.journal().inner.holds_op(APPLIED + 1), + truncate_fails + ); + } + } + #[compio::test] async fn given_commit_only_start_view_when_entries_are_acked_should_keep_the_running_suffix() { const FIRST_HEAD: u64 = 4; diff --git a/core/shard/src/poll.rs b/core/shard/src/poll.rs index e79ea47c7..86a1d3231 100644 --- a/core/shard/src/poll.rs +++ b/core/shard/src/poll.rs @@ -163,10 +163,17 @@ where } let rejection = partitions .with_partition(&namespace, |partition| { - if matches!(read, PartitionRead::PollOnPrimary { .. }) - && partition.consensus().is_normal_backup() - { - return Some(IggyError::NotPrimary); + if read.requires_primary() { + let consensus = partition.consensus(); + if consensus.is_normal_backup() { + return Some(IggyError::NotPrimary); + } + if !consensus.is_primary() + || !consensus.is_normal() + || consensus.is_transferring() + { + return Some(IggyError::TransientNotAccepted); + } } if partition.requires_state_transfer() || partition.read_history_is_changing() { return Some(IggyError::TransientNotAccepted); @@ -217,17 +224,21 @@ where if let PartitionRead::Poll { metadata: Some(metadata), .. - } = &read + } + | PartitionRead::UnboundPollOnPrimary { metadata, .. } = &read && (!metadata.is_valid(self.plane.metadata().mux_stm.streams(), namespace) || !metadata.matches_partition(self.shards_table.epoch_for(namespace))) { return Some(IggyError::TransientNotAccepted); } - (matches!(read, PartitionRead::Poll { .. }) && partition.history_deleted()) - .then_some(IggyError::HistoryUnavailable) + (matches!( + read, + PartitionRead::Poll { .. } | PartitionRead::UnboundPollOnPrimary { .. } + ) && partition.history_deleted()) + .then_some(IggyError::HistoryUnavailable) }) .unwrap_or_else(|| { - matches!(read, PartitionRead::PollOnPrimary { .. }) + read.requires_primary() .then_some(IggyError::TransientNotAccepted) }); if let Some(error) = rejection { @@ -250,9 +261,15 @@ where read => (read, None), }; let result = match read { - PartitionRead::SessionRetired { identity } => partitions + PartitionRead::SessionRetired { + identity, + created_revision, + } => partitions .with_partition(&namespace, |partition| { - PartitionReadReply::SessionRetired(partition.session_retired(identity)) + PartitionReadReply::SessionRetired( + partition.created_revision() == created_revision + && partition.session_retired(identity), + ) }) .unwrap_or(PartitionReadReply::NotFound), PartitionRead::Primary => partitions @@ -274,7 +291,8 @@ where }) .unwrap_or(PartitionReadReply::NotFound), PartitionRead::Poll { consumer, args, .. } - | PartitionRead::PollOnPrimary { consumer, args, .. } => { + | PartitionRead::PollOnPrimary { consumer, args, .. } + | PartitionRead::UnboundPollOnPrimary { consumer, args, .. } => { match partitions.build_poll_snapshot(&namespace, consumer, &args) { None => PartitionReadReply::NotFound, Some(plan) if plan.needs_off_pump_io() => { @@ -329,7 +347,8 @@ where } } } - PartitionRead::ConsumerOffset { consumer } => partitions + PartitionRead::ConsumerOffset { consumer } + | PartitionRead::ConsumerOffsetOnPrimary { consumer } => partitions .consumer_offset_read(&namespace, consumer) .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| { PartitionReadReply::ConsumerOffset { @@ -337,7 +356,8 @@ where current_offset, } }), - PartitionRead::ExternalGroupOffset { group_id } => partitions + PartitionRead::ExternalGroupOffset { group_id } + | PartitionRead::ExternalGroupOffsetOnPrimary { group_id } => partitions .external_group_offset_read(&namespace, group_id) .map_or(PartitionReadReply::NotFound, |(stored, current_offset)| { PartitionReadReply::ConsumerOffset { diff --git a/core/shard/src/poll/completion_tests.rs b/core/shard/src/poll/completion_tests.rs index 790fa013b..0b1d2b1ac 100644 --- a/core/shard/src/poll/completion_tests.rs +++ b/core/shard/src/poll/completion_tests.rs @@ -29,6 +29,7 @@ use consensus::{ encode_state_manifest, }; use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest; +use iggy_binary_protocol::requests::system::SessionIdentity; use iggy_binary_protocol::requests::topics::DeleteTopicRequest; use iggy_binary_protocol::{ AckLevel, ReplyHeader, RequestPreparesHeader, RequestStateChunkHeader, @@ -66,6 +67,45 @@ use crate::{ use iggy_binary_protocol::primitives::partition_history::ConsumerGroupOwner; use iggy_binary_protocol::requests::partitions::InstallConsumerGroupOwnerRequest; +#[compio::test] +async fn given_deleted_history_when_retirement_read_names_old_incarnation_should_refuse_replacement() + { + const ORIGINAL_REVISION: u64 = 7; + const REPLACEMENT_REVISION: u64 = 8; + let bus = Rc::new(IggyMessageBus::new(0)); + let namespace = IggyNamespace::new(0, 0, 0); + let config = partitions_config(); + let metadata = PollTestMetadata::new((Users::default(), (StreamsInner::default().into(), ()))); + let (owner, _sender) = owner_with_metadata(&bus, config.clone(), namespace, metadata); + let identity = SessionIdentity { + client_id: 1, + session: 1, + metadata_watermark: 100, + }; + for revision in [ORIGINAL_REVISION, REPLACEMENT_REVISION] { + let (mut partition, _) = + partition_with_messages_at_revision(&bus, namespace, &["message"], revision).await; + delete_history(&mut partition, &config, 10).await; + owner.plane.partitions().remove(&namespace); + owner.plane.partitions().insert(namespace, partition); + let (reply, replies) = channel(1); + owner + .on_partition_read( + namespace, + PartitionRead::SessionRetired { + identity, + created_revision: ORIGINAL_REVISION, + }, + reply, + ) + .await; + assert!(matches!( + replies.try_recv().unwrap(), + PartitionReadReply::SessionRetired(retired) if retired == (revision == ORIGINAL_REVISION) + )); + } +} + #[compio::test] #[allow(clippy::too_many_lines)] async fn given_pending_attached_poll_when_metadata_changes_should_fence_only_affected_reads() { @@ -490,23 +530,31 @@ async fn given_backup_when_polled_on_primary_should_name_the_primary_unless_tran .poll_metadata(namespace, None, BOUND_CLIENT) .expect("metadata names the partition"), }; - let (reply, replies) = channel(1); - owner - .on_partition_read( - namespace, - PartitionRead::PollOnPrimary { - consumer: PollingConsumer::Consumer(1, 0), - args: PollingArgs::new(PollingStrategy::first(), 1, true), - attachment, - }, - reply, - ) - .await; - let reply = replies.try_recv().expect("the owner answers the read"); - assert!( - matches!(&reply, PartitionReadReply::Rejected(error) if *error == expected), - "transferring {transferring}: expected {expected:?}, got {reply:?}" - ); + let metadata = attachment.metadata; + for read in [ + PartitionRead::PollOnPrimary { + consumer: PollingConsumer::Consumer(1, 0), + args: PollingArgs::new(PollingStrategy::first(), 1, true), + attachment, + }, + PartitionRead::UnboundPollOnPrimary { + consumer: PollingConsumer::Consumer(1, 0), + args: PollingArgs::new(PollingStrategy::next(), 1, false), + metadata, + }, + PartitionRead::ConsumerOffsetOnPrimary { + consumer: PollingConsumer::Consumer(1, 0), + }, + PartitionRead::ExternalGroupOffsetOnPrimary { group_id: 1 }, + ] { + let (reply, replies) = channel(1); + owner.on_partition_read(namespace, read, reply).await; + let reply = replies.try_recv().expect("the owner answers the read"); + assert!( + matches!(&reply, PartitionReadReply::Rejected(error) if *error == expected), + "transferring {transferring}: expected {expected:?}, got {reply:?}" + ); + } let attachment = ConsumerAttachment { session: table.attach_session(BOUND_CLIENT, 1, BOUND_USER).unwrap(),
