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 b841581cd083bba24d1ecd747c9788923e73d769 Author: Hubert Gruszecki <[email protected]> AuthorDate: Mon Oct 5 14:26:04 2026 +0200 fix(server): preserve session fences across recovery and retirement Uninitialized partitions could remain offline after restart, while failed partitions stalled retirement and exhausted session capacity. Deleted or disabled users could also retain authenticated bindings. Persist incarnation evidence and retirement fences outside partition directories, batch retirement barriers, and preserve progress across truncation. Revoke sessions and memberships during commit and replay. Use protocol, release and storage identity for peer admission, and correct the NoAck and capacity contracts. Require storage format 3 and metadata snapshot format 8. Cover the reviewed failures with regressions and a three-replica restart test. --- .../src/requests/system/bind_session.rs | 8 +- core/binary_protocol/src/requests/system/mod.rs | 2 + .../src/requests/system/retire_sessions.rs | 117 ++++++ core/configs/src/server_config/metadata.rs | 12 +- core/configs/src/server_config/partition.rs | 11 +- core/consensus/src/client_table.rs | 21 + core/integration/tests/cluster/crash_durability.rs | 56 +++ core/message_bus/src/replica/handshake.rs | 93 +++-- core/metadata/src/impls/metadata.rs | 155 +++++++- core/metadata/src/impls/recovery.rs | 139 ++++++- core/metadata/src/stm/result.rs | 10 +- core/metadata/src/stm/snapshot.rs | 21 +- core/metadata/src/stm/stream.rs | 11 + core/metadata/src/stm/user.rs | 10 +- core/partitions/src/iggy_partition.rs | 149 +++++-- core/partitions/src/iggy_partitions.rs | 30 +- core/partitions/src/lib.rs | 3 +- core/partitions/src/partition_storage.rs | 6 + core/server/README.md | 40 +- core/server/config.toml | 23 +- core/server/src/boot/mod.rs | 13 +- core/server/src/boot/recovery.rs | 1 - core/server/src/consumer_group.rs | 10 +- core/server/src/consumer_group/lease.rs | 5 +- core/server/src/consumer_group/liveness.rs | 384 ++++++++++++++++-- core/server/src/dispatch/mod.rs | 188 ++++++++- core/server/src/http/admission.rs | 11 +- core/server/src/partition_helpers.rs | 436 ++++++++++++++++++--- core/server/src/partition_reconciler.rs | 83 +++- core/server/src/server_error.rs | 15 +- core/shard/src/lib.rs | 12 +- core/shard/src/poll.rs | 14 + 32 files changed, 1871 insertions(+), 218 deletions(-) diff --git a/core/binary_protocol/src/requests/system/bind_session.rs b/core/binary_protocol/src/requests/system/bind_session.rs index 16208074c..c4933c8d3 100644 --- a/core/binary_protocol/src/requests/system/bind_session.rs +++ b/core/binary_protocol/src/requests/system/bind_session.rs @@ -30,9 +30,13 @@ pub struct SessionIdentity { pub metadata_watermark: u64, } +impl SessionIdentity { + pub const ENCODED_SIZE: usize = 32; +} + impl WireEncode for SessionIdentity { fn encoded_size(&self) -> usize { - 32 + Self::ENCODED_SIZE } fn encode(&self, buf: &mut BytesMut) { @@ -50,7 +54,7 @@ impl WireDecode for SessionIdentity { session: read_u64_le(buf, 16)?, metadata_watermark: read_u64_le(buf, 24)?, }, - 32, + Self::ENCODED_SIZE, )) } } diff --git a/core/binary_protocol/src/requests/system/mod.rs b/core/binary_protocol/src/requests/system/mod.rs index bf3b8a218..5d457658f 100644 --- a/core/binary_protocol/src/requests/system/mod.rs +++ b/core/binary_protocol/src/requests/system/mod.rs @@ -24,6 +24,7 @@ pub mod get_me; pub mod get_snapshot; pub mod get_stats; pub mod ping; +pub mod retire_sessions; pub use bind_session::{BindSessionRequest, SessionIdentity}; pub use describe_options::{ @@ -36,3 +37,4 @@ pub use get_me::GetMeRequest; pub use get_snapshot::GetSnapshotRequest; pub use get_stats::GetStatsRequest; pub use ping::PingRequest; +pub use retire_sessions::{MAX_SESSIONS_PER_RETIREMENT, RetireSessionsRequest}; diff --git a/core/binary_protocol/src/requests/system/retire_sessions.rs b/core/binary_protocol/src/requests/system/retire_sessions.rs new file mode 100644 index 000000000..6ab770a4e --- /dev/null +++ b/core/binary_protocol/src/requests/system/retire_sessions.rs @@ -0,0 +1,117 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use bytes::BytesMut; + +use crate::requests::system::SessionIdentity; +use crate::{WireDecode, WireEncode, WireError}; + +pub const MAX_SESSIONS_PER_RETIREMENT: usize = 128; + +/// Server-originated partition barrier. Every identity is already ended in +/// committed metadata. Admission excludes pending writes for the whole batch. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RetireSessionsRequest { + pub identities: Vec<SessionIdentity>, +} + +impl WireEncode for RetireSessionsRequest { + fn encoded_size(&self) -> usize { + self.identities.len() * SessionIdentity::ENCODED_SIZE + } + + fn encode(&self, buf: &mut BytesMut) { + for identity in &self.identities { + identity.encode(buf); + } + } +} + +impl WireDecode for RetireSessionsRequest { + fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> { + let (chunks, remainder) = buf.as_chunks::<{ SessionIdentity::ENCODED_SIZE }>(); + if chunks.is_empty() || chunks.len() > MAX_SESSIONS_PER_RETIREMENT || !remainder.is_empty() + { + return Err(WireError::Validation( + "invalid session retirement batch length".into(), + )); + } + let identities = chunks + .iter() + .map(|bytes| { + let identity = SessionIdentity::decode_from(bytes)?; + if identity.client_id == 0 + || identity.session == 0 + || identity.metadata_watermark < identity.session + { + return Err(WireError::Validation( + "invalid session retirement identity".into(), + )); + } + Ok(identity) + }) + .collect::<Result<Vec<_>, _>>()?; + Ok((Self { identities }, buf.len())) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn retirement_batch_roundtrips_and_rejects_invalid_boundaries() { + let identity = SessionIdentity { + client_id: 1, + session: 2, + metadata_watermark: 3, + }; + for count in [1, MAX_SESSIONS_PER_RETIREMENT] { + let request = RetireSessionsRequest { + identities: vec![identity; count], + }; + let bytes = request.to_bytes(); + assert_eq!(RetireSessionsRequest::decode_from(&bytes).unwrap(), request); + assert!(RetireSessionsRequest::decode_from(&bytes[..bytes.len() - 1]).is_err()); + } + assert!(RetireSessionsRequest::decode_from(&[]).is_err()); + assert!( + RetireSessionsRequest::decode_from( + &RetireSessionsRequest { + identities: vec![identity; MAX_SESSIONS_PER_RETIREMENT + 1], + } + .to_bytes() + ) + .is_err() + ); + assert!( + RetireSessionsRequest::decode_from( + &RetireSessionsRequest { + identities: vec![ + identity, + SessionIdentity { + session: 0, + ..identity + } + ], + } + .to_bytes() + ) + .is_err() + ); + } +} diff --git a/core/configs/src/server_config/metadata.rs b/core/configs/src/server_config/metadata.rs index cd24781b4..1fad9c91f 100644 --- a/core/configs/src/server_config/metadata.rs +++ b/core/configs/src/server_config/metadata.rs @@ -108,11 +108,13 @@ pub struct MetadataConfig { /// restore the previous value to recover. pub journal_slots: usize, - /// Slot count of the VSR client table: how many distinct clients - /// (TCP/QUIC/WS virtual clients and HTTP sessions together) hold live - /// session state before the oldest-committed entry is evicted. The - /// HTTP session cap tracks this at half, so raising it lifts - /// both. + /// Initial capacity of the metadata session registry, shared by binary + /// clients and HTTP sessions. Live and unretired sessions occupy slots; + /// a full registry refuses new sessions instead of evicting retry protection. + /// The first committed operation fixes this capacity. Changing configuration + /// does not resize an existing registry, including after restart or transfer. + /// The HTTP cache cap uses half the configured value, but cannot raise the + /// committed registry limit. pub clients_table_max: usize, /// Cap on the partitions of all streams and topics. A CreateTopic or diff --git a/core/configs/src/server_config/partition.rs b/core/configs/src/server_config/partition.rs index 00db29e40..b57d3be41 100644 --- a/core/configs/src/server_config/partition.rs +++ b/core/configs/src/server_config/partition.rs @@ -204,11 +204,12 @@ pub struct PartitionConfig { /// pinned request-buffer memory by the partition count. pub prepare_queue_depth: usize, - /// Distinct clients each partition group tracks request watermarks for, - /// deduplicating retried produces and consumer-offset writes. At capacity - /// the entry whose newest commit is oldest is evicted, which costs dedup - /// coverage for that client (its next replay re-executes, exactly as it - /// would have before dedup existed) and never correctness. Must be > 0 and + /// Initial capacity for session-qualified retry protection in each partition + /// group. Live and unretired sessions occupy slots. A full table refuses new + /// writers with `TransientNotAccepted`; existing protection is never evicted. + /// Ended sessions release slots only after ordered retirement. The first + /// committed operation fixes this limit, so changing configuration does not + /// resize an existing group after restart or transfer. Must be > 0 and /// <= [`PARTITION_DEDUP_CLIENTS_CEILING`]. /// /// Unlike `[metadata] clients_table_max`, this budget is PER GROUP, so the diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs index a16959c6b..dabf85f73 100644 --- a/core/consensus/src/client_table.rs +++ b/core/consensus/src/client_table.rs @@ -688,6 +688,27 @@ impl ClientTable { false } + /// Revoke a principal's bindings while retaining receipts until ordered retirement. + pub fn end_user_sessions( + &mut self, + user_id: u32, + ended_op: u64, + ) -> Vec<iggy_binary_protocol::requests::system::SessionIdentity> { + let mut ended = Vec::new(); + for entry in self.slots.iter_mut().flatten() { + if entry.user_id == user_id && entry.ended_op.is_none() { + entry.ended_op = Some(ended_op); + entry.attachment = None; + ended.push(iggy_binary_protocol::requests::system::SessionIdentity { + client_id: entry.client_id, + session: entry.epoch, + metadata_watermark: ended_op, + }); + } + } + ended + } + /// Retain the exact Logout result before ending its session. /// /// # Errors diff --git a/core/integration/tests/cluster/crash_durability.rs b/core/integration/tests/cluster/crash_durability.rs index a1eaa9c83..a9d13bf37 100644 --- a/core/integration/tests/cluster/crash_durability.rs +++ b/core/integration/tests/cluster/crash_durability.rs @@ -775,6 +775,62 @@ async fn given_all_replicas_checkpointed_when_restarted_should_elect_and_extend_ verify_transferred_quorum(harness, &stream, &topic).await; } +#[iggy_harness(cluster_nodes = 3)] +async fn given_uninitialized_partitions_when_all_replicas_restart_should_elect_and_accept_writes( + harness: &mut TestHarness, +) { + let client = harness.tcp_root_client().await.unwrap(); + create_stream_and_topic(&client, true).await; + let stream = Identifier::named(STREAM_NAME).unwrap(); + let topic = Identifier::named(TOPIC_NAME).unwrap(); + let details = client.get_topic(&stream, &topic).await.unwrap().unwrap(); + let stream_id = client.get_stream(&stream).await.unwrap().unwrap().id; + let deadline = tokio::time::Instant::now() + CONVERGE_TIMEOUT; + for node in 0..3 { + let follower = harness + .node(node) + .tcp_client() + .unwrap() + .with_root_login() + .connect() + .await + .unwrap(); + loop { + if matches!(follower.get_topic(&stream, &topic).await, Ok(Some(_))) { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "all replicas must commit the topic metadata" + ); + sleep(POLL_INTERVAL).await; + } + } + harness.kill_cluster().unwrap(); + // No partition writes were submitted. Recreate the crash cut after the + // metadata commit and before any local initialization was published. + for node in 0..3 { + let data = harness.node(node).data_path(); + for path in [ + data.join(format!( + "streams/{stream_id}/topics/{}/partitions/0", + details.id + )), + data.join("partition-initialization"), + ] { + if path.exists() { + std::fs::remove_dir_all(path).unwrap(); + } + } + } + harness.restart_cluster().await.unwrap(); + let client = wait_until_cluster_serves(harness, &[0, 1, 2], CONVERGE_TIMEOUT).await; + let acked = produce_acked(&client, "initialized-after-restart", 1).await; + wait_for_acked_readable(&client, &acked, CONVERGE_TIMEOUT) + .await + .unwrap(); +} + async fn verify_checkpoint_quarantine( harness: &mut TestHarness, stream_id: u32, diff --git a/core/message_bus/src/replica/handshake.rs b/core/message_bus/src/replica/handshake.rs index 786a3b4d9..89db00473 100644 --- a/core/message_bus/src/replica/handshake.rs +++ b/core/message_bus/src/replica/handshake.rs @@ -69,11 +69,9 @@ use iggy_binary_protocol::{Command, HEADER_SIZE}; use iggy_common::IggyError; use rustls::pki_types::ServerName; use std::collections::HashMap; -use std::io::Read; use std::mem::size_of; use std::rc::Rc; use std::sync::Arc; -use std::sync::OnceLock; use tracing::{debug, warn}; /// Identity and auth parameters the owning shard needs to run a @@ -535,32 +533,73 @@ fn with_identity( }) } -/// Exact release, protocol, and executable identity for coordinated deployment. -/// # Errors -/// Returns an error if the running executable cannot be opened or read. -pub fn binary_identity(release: &str) -> std::io::Result<[u8; auth::IDENTITY_LEN]> { - static ARTIFACT: OnceLock<std::io::Result<[u8; auth::IDENTITY_LEN]>> = OnceLock::new(); - const READ_BYTES: usize = 32 * 1024; - let artifact = ARTIFACT.get_or_init(|| { - let mut file = std::fs::File::open(std::env::current_exe()?)?; - let mut hasher = blake3::Hasher::new(); - let mut bytes = vec![0; READ_BYTES]; - loop { - let count = file.read(&mut bytes)?; - if count == 0 { - break; - } - hasher.update(&bytes[..count]); - } - Ok(*hasher.finalize().as_bytes()) - }); - let artifact = artifact - .as_ref() - .map_err(|error| std::io::Error::new(error.kind(), error.to_string()))?; - let mut hasher = blake3::Hasher::new_derive_key("apache.iggy replica identity v1"); +/// Protocol, release and storage identity, independent of executable packaging. +#[must_use] +pub fn binary_identity(release: &str, storage_format: &[u8]) -> [u8; auth::IDENTITY_LEN] { + let mut hasher = blake3::Hasher::new_derive_key("apache.iggy replica identity v2"); hasher.update(&iggy_binary_protocol::IGGY_PROTOCOL_VERSION.to_le_bytes()); hasher.update(&(release.len() as u64).to_le_bytes()); hasher.update(release.as_bytes()); - hasher.update(artifact); - Ok(*hasher.finalize().as_bytes()) + hasher.update(&(storage_format.len() as u64).to_le_bytes()); + hasher.update(storage_format); + *hasher.finalize().as_bytes() +} + +#[cfg(all(test, target_os = "linux"))] +mod tests { + use super::binary_identity; + use std::io::Write; + use std::process::Command; + + #[test] + fn given_identical_release_when_executable_bytes_differ_should_keep_peer_identity() { + const OUTPUT_ENV: &str = "IGGY_TEST_PEER_IDENTITY_OUTPUT"; + const RELEASE: &str = "test-release"; + let identity = binary_identity(RELEASE, b"test-storage"); + if let Some(output) = std::env::var_os(OUTPUT_ENV) { + std::fs::write(output, identity).unwrap(); + return; + } + + let directory = tempfile::tempdir().unwrap(); + let executable = directory.path().join("identity-probe"); + let output = directory.path().join("identity"); + std::fs::copy(std::env::current_exe().unwrap(), &executable).unwrap(); + // Trailing bytes change the artifact digest without changing the ELF program. + std::fs::OpenOptions::new() + .append(true) + .open(&executable) + .unwrap() + .write_all(b"different packaging") + .unwrap(); + let probe = Command::new(executable) + .args([ + "--exact", + "replica::handshake::tests::given_identical_release_when_executable_bytes_differ_should_keep_peer_identity", + ]) + .env(OUTPUT_ENV, &output) + .output() + .unwrap(); + assert!( + probe.status.success(), + "identity probe failed: {}", + String::from_utf8_lossy(&probe.stderr) + ); + assert_eq!( + std::fs::read(output).unwrap(), + identity, + "packaging bytes must not make compatible replicas reject each other" + ); + } + + #[test] + fn given_different_release_or_storage_when_identifying_peers_should_reject_mixing() { + let identity = binary_identity("release-1", b"storage-1"); + assert_ne!(identity, binary_identity("release-2", b"storage-1")); + assert_ne!(identity, binary_identity("release-1", b"storage-2")); + assert_ne!( + binary_identity("release", b"-storage"), + binary_identity("release-", b"storage") + ); + } } diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index b559bfe78..abc106522 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -732,6 +732,17 @@ pub fn apply_committed_prepare<M>( header.op ); }); + if table_mutations_allowed && let Some(user_id) = apply.revoked_user { + for identity in client_table + .borrow_mut() + .end_user_sessions(user_id, header.op) + { + mux_stm.streams().remove_consumer_group_member( + identity.client_id, + iggy_common::IggyTimestamp::from(header.timestamp), + ); + } + } fire_notifier(header.operation); let reply = build_reply_message_with(&header, apply.reply_body_len(), |dst| { apply.write_reply_body(dst); @@ -3147,6 +3158,20 @@ where prepare_header.op ); }); + if self.client_table_mutation_allowed(prepare_header.op) + && let Some(user_id) = apply.revoked_user + { + for identity in self + .client_table + .borrow_mut() + .end_user_sessions(user_id, prepare_header.op) + { + self.mux_stm.streams().remove_consumer_group_member( + identity.client_id, + iggy_common::IggyTimestamp::from(prepare_header.timestamp), + ); + } + } // Post-commit notifier (e.g. partition reconciler // wake-up). Filtering by operation is the // recipient's responsibility. @@ -4480,7 +4505,7 @@ mod tests { use iggy_binary_protocol::requests::consumer_groups::CreateConsumerGroupRequest; use iggy_binary_protocol::requests::streams::CreateStreamRequest; use iggy_binary_protocol::requests::topics::CreateTopicRequest; - use iggy_binary_protocol::requests::users::CreateUserRequest; + use iggy_binary_protocol::requests::users::{CreateUserRequest, DeleteUserRequest}; use iggy_common::{IggyTimestamp, UserStatus, variadic}; use journal::prepare_journal::PrepareJournal; use message_bus::{ @@ -6691,6 +6716,134 @@ mod tests { ); } + #[compio::test] + async fn given_user_revocation_when_committed_should_release_group_memberships() { + const CLIENT: u128 = 1; + const USER: u32 = 7; + let (_directory, metadata) = metadata_with_group_member(CLIENT).await; + metadata.mux_stm.users().ensure_root_user("root", "hash"); + let consensus = metadata.consensus.as_ref().unwrap(); + let request = RoutedRequestHeader::default(); + for user_id in 1..=USER { + let create = CreateUserRequest { + username: WireName::new(format!("user-{user_id}")).unwrap(), + password: "hash".to_owned(), + status: UserStatus::Active.as_code(), + permissions: None, + options: WireOptions::empty(), + }; + metadata + .mux_stm + .update(build_prepare_message( + consensus, + &request, + Operation::CreateUser, + &create.to_bytes(), + )) + .unwrap(); + } + let deletion = consensus::seal_prepare_capacity( + build_prepare_message( + consensus, + &request, + Operation::DeleteUser, + &DeleteUserRequest { + user_id: WireIdentifier::numeric(USER), + } + .to_bytes(), + ), + metadata.client_table_capacity(), + ); + apply_committed_prepare( + &*metadata.mux_stm, + &metadata.client_table, + true, + |_| {}, + deletion, + ); + assert_eq!(metadata.client_table.borrow().ended_sessions().count(), 1); + assert!( + metadata + .mux_stm + .streams() + .consumer_group_memberships(CLIENT) + .is_empty(), + "revoked sessions must release their consumer assignments" + ); + } + + #[test] + fn given_rejected_or_older_user_deletion_when_applying_should_preserve_newer_sessions() { + const CLIENT: u128 = 1; + const SESSION: u64 = 100; + const USER: u32 = 1; + for (frontier, acting_user) in [(SESSION, 0), (0, u32::MAX)] { + let metadata = metadata_plane(); + metadata.mux_stm.users().ensure_root_user("root", "hash"); + let consensus = metadata.consensus.as_ref().unwrap(); + let request = RoutedRequestHeader::default(); + let create = CreateUserRequest { + username: WireName::new("original").unwrap(), + password: "hash".to_owned(), + status: UserStatus::Active.as_code(), + permissions: None, + options: WireOptions::empty(), + }; + metadata + .mux_stm + .update(build_prepare_message( + consensus, + &request, + Operation::CreateUser, + &create.to_bytes(), + )) + .unwrap(); + metadata + .client_table + .borrow_mut() + .commit_register(CLIENT, USER, [0x5a; 32], register_reply(CLIENT, SESSION)) + .unwrap(); + metadata.client_table_frontier.set(frontier); + let request = RoutedRequestHeader { + user_id: acting_user, + ..request + }; + let deletion = build_prepare_message( + consensus, + &request, + Operation::DeleteUser, + &DeleteUserRequest { + user_id: WireIdentifier::numeric(USER), + } + .to_bytes(), + ) + .transmute_header(|old, header: &mut PrepareHeader| { + *header = old; + header.retry_capacity = u32::try_from(metadata.client_table_capacity()).unwrap(); + }); + apply_committed_prepare( + &*metadata.mux_stm, + &metadata.client_table, + metadata.client_table_mutation_allowed(deletion.header().op), + |_| {}, + consensus::seal_prepare_checksum(deletion), + ); + assert_eq!( + metadata.client_table.borrow().get_epoch(CLIENT), + Some(SESSION) + ); + assert_eq!(metadata.client_table.borrow().ended_sessions().count(), 0); + assert_eq!( + metadata + .mux_stm + .users() + .read(|users| users.items.get(USER as usize).is_some()), + acting_user != 0, + "only the authorized deletion updates the state machine" + ); + } + } + #[allow(clippy::future_not_send)] async fn metadata_with_group_member( client_id: u128, diff --git a/core/metadata/src/impls/recovery.rs b/core/metadata/src/impls/recovery.rs index 948fa02bb..ec2f2444e 100644 --- a/core/metadata/src/impls/recovery.rs +++ b/core/metadata/src/impls/recovery.rs @@ -345,6 +345,10 @@ pub struct RecoveredMetadata<M> { /// `clients_table_max` sizes a fresh registry. Recovery preserves the /// checkpoint's committed capacity. /// +/// `on_replayed_session` receives registration, logout, and user-revocation +/// effects. For revocation, `client` and `session` identify each affected +/// session; the operation, op, and timestamp still name the committed user mutation. +/// /// `identity` is what this replica expects its own durable superblock to carry, and /// a mismatch refuses boot ([`RecoveryError::SuperblockIdentityMismatch`]). Its /// `replica_count == 1` also marks a single-replica cluster: the quorum is 1/1, so @@ -676,6 +680,18 @@ where // WAL replay must recompute authorization denials identically to the // primary/backup commit paths, so it goes through the same gate. let reply = mux_stm.gated_update(entry)?; + if let Some(user_id) = reply.revoked_user { + for identity in client_table.end_user_sessions(user_id, header.op) { + on_replayed_session( + &mux_stm, + &PrepareHeader { + client: identity.client_id, + session: identity.session, + ..*header + }, + ); + } + } // Every client commit must restore its original receipt before admission. if header.client != consensus::client_table::RESERVED_CLIENT_ID && header.operation != Operation::CompleteConsumerGroupRevocation @@ -837,10 +853,17 @@ fn verify_checkpoint_pairing( #[allow(clippy::cast_possible_truncation)] mod tests { use super::*; - use crate::impls::metadata::checkpoint_checksum; + use crate::impls::metadata::{StreamsFrontend, checkpoint_checksum}; use crate::stm::snapshot::SNAPSHOT_FORMAT_VERSION; + use crate::stm::stream::Streams; + use crate::stm::user::Users; use consensus::CLIENTS_TABLE_MAX; use iggy_binary_protocol::consensus::{Command, Operation}; + use iggy_binary_protocol::requests::users::{ + CreateUserRequest, DeleteUserRequest, UpdateUserRequest, + }; + use iggy_binary_protocol::{WireEncode, WireIdentifier, WireName, WireOptions}; + use iggy_common::UserStatus; use journal::Journal; use server_common::iobuf::Owned; use tempfile::tempdir; @@ -1576,6 +1599,120 @@ mod tests { assert_eq!(recovered.last_applied_op, Some(2)); } + #[compio::test] + async fn given_revoked_user_when_replaying_wal_should_restore_ended_sessions() { + const CLIENT: u128 = 0x1337; + const ROOT_CLIENT: u128 = CLIENT + 1; + const USER: u32 = 1; + type UserStm = MuxStateMachine<iggy_common::variadic!(Users, Streams)>; + for operation in [Operation::DeleteUser, Operation::UpdateUser] { + let directory = tempdir().unwrap(); + let metadata_dir = directory.path().join("metadata"); + std::fs::create_dir_all(&metadata_dir).unwrap(); + let create = CreateUserRequest { + username: WireName::new("user").unwrap(), + password: "hash".to_owned(), + status: UserStatus::Active.as_code(), + permissions: None, + options: WireOptions::empty(), + } + .to_bytes(); + let revocation = if operation == Operation::DeleteUser { + DeleteUserRequest { + user_id: WireIdentifier::numeric(USER), + } + .to_bytes() + } else { + UpdateUserRequest { + user_id: WireIdentifier::numeric(USER), + username: None, + status: Some(UserStatus::Inactive.as_code()), + options: WireOptions::empty(), + } + .to_bytes() + }; + let mut parent = 0; + { + let journal = PrepareJournal::open(&metadata_dir.join("journal.wal"), 0) + .await + .unwrap(); + for (op, operation, body, client, user_id) in [ + ( + 1, + Operation::Register, + bytes::Bytes::from_static(&[0x5a; 32]), + ROOT_CLIENT, + 0, + ), + (2, Operation::CreateUser, create, ROOT_CLIENT, 0), + ( + 3, + Operation::Register, + bytes::Bytes::from_static(&[0x5a; 32]), + CLIENT, + USER, + ), + (4, operation, revocation, ROOT_CLIENT, 0), + ] { + let mut prepare = make_prepare(op, body.len()); + prepare.as_mut_slice()[HEADER_SIZE..].copy_from_slice(&body); + let prepare = consensus::seal_prepare_checksum(prepare.transmute_header( + |old, header: &mut PrepareHeader| { + *header = old; + header.operation = operation; + header.client = client; + header.user_id = user_id; + header.session = 1; + header.request = if operation == Operation::Register { + 0 + } else { + op + }; + header.parent = parent; + header.checksum_body = + u128::from(iggy_common::calculate_checksum(&body)); + }, + )); + parent = prepare.header().checksum; + journal.append(prepare).await.unwrap(); + } + journal.storage_ref().fsync().await.unwrap(); + } + let revoked = std::cell::RefCell::new(Vec::new()); + let recovered = recover::<UserStm>( + directory.path(), + SOLO, + journal::prepare_journal::DEFAULT_SLOT_COUNT, + CLIENTS_TABLE_MAX, + |stm| { + stm.users().ensure_root_user("root", "hash"); + }, + |_, header| { + if header.operation == operation { + revoked + .borrow_mut() + .push((header.client, header.session, header.op)); + } + }, + ) + .await + .unwrap(); + assert_eq!(recovered.last_applied_op, Some(4)); + assert_eq!(recovered.client_table.get_epoch(CLIENT), None); + assert_eq!(recovered.client_table.get_epoch(ROOT_CLIENT), Some(1)); + assert_eq!(*revoked.borrow(), vec![(CLIENT, 3, 4)]); + let ended: Vec<_> = recovered.client_table.ended_sessions().collect(); + assert_eq!( + ended, + vec![iggy_binary_protocol::requests::system::SessionIdentity { + client_id: CLIENT, + session: 3, + metadata_watermark: 4, + }] + ); + } + } + #[test] fn snapshot_persist_load_roundtrip() { let dir = tempdir().unwrap(); diff --git a/core/metadata/src/stm/result.rs b/core/metadata/src/stm/result.rs index cca06dc54..2a42fad19 100644 --- a/core/metadata/src/stm/result.rs +++ b/core/metadata/src/stm/result.rs @@ -47,13 +47,20 @@ pub use iggy_binary_protocol::consensus::result_code; pub struct ApplyReply { pub code: u32, pub body: Bytes, + /// Local apply effect, recomputed from the committed operation on every replay. + /// It is not part of the client reply or persisted receipt. + pub(crate) revoked_user: Option<u32>, } impl ApplyReply { /// Successful apply carrying the typed reply body. #[must_use] pub const fn ok(body: Bytes) -> Self { - Self { code: 0, body } + Self { + code: 0, + body, + revoked_user: None, + } } /// Committed business rejection: empty body, `code` is the result @@ -63,6 +70,7 @@ impl ApplyReply { Self { code: code.into(), body: Bytes::new(), + revoked_user: None, } } diff --git a/core/metadata/src/stm/snapshot.rs b/core/metadata/src/stm/snapshot.rs index 982208695..ce1712fa9 100644 --- a/core/metadata/src/stm/snapshot.rs +++ b/core/metadata/src/stm/snapshot.rs @@ -49,7 +49,9 @@ use crate::stm::user::UsersSnapshot; /// field. /// /// Version 7: session protection no longer stores a payload fingerprint. -pub const SNAPSHOT_FORMAT_VERSION: u32 = 7; +/// +/// Version 8: streams persist the revision of their partition incarnations. +pub const SNAPSHOT_FORMAT_VERSION: u32 = 8; /// Oldest format version [`MetadataSnapshot::decode`] still reads. /// @@ -566,7 +568,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 = 7; + const PINNED_VERSION: u32 = 8; let encoded = MetadataSnapshot::new(0).encode().unwrap(); let mut cursor = encoded.as_slice(); @@ -590,6 +592,7 @@ mod tests { // actually grows need their own pins. const TOPIC_FIELD_COUNT: u32 = 11; const STREAM_FIELD_COUNT: u32 = 6; + const STREAMS_FIELD_COUNT: u32 = 3; const USER_FIELD_COUNT: u32 = 7; // Version 4 appended `fences`; it is defaulted on read, which is what // keeps version 3 readable, so a further append here needs the same @@ -669,6 +672,18 @@ mod tests { "StreamSnapshot's field count changed; bump SNAPSHOT_FORMAT_VERSION with it" ); + let streams = StreamsSnapshot { + items: vec![(0, stream)], + revision: 1, + namespace_revision: 1, + }; + let encoded = rmp_serde::to_vec(&streams).unwrap(); + assert_eq!( + rmp::decode::read_array_len(&mut encoded.as_slice()).unwrap(), + STREAMS_FIELD_COUNT, + "StreamsSnapshot's field count changed; bump SNAPSHOT_FORMAT_VERSION with it" + ); + let user = crate::stm::user::UserSnapshot { id: 0, username: String::new(), @@ -803,6 +818,7 @@ mod tests { let mut snapshot = MetadataSnapshot::new(100); snapshot.streams = Some(StreamsSnapshot { revision: 0, + namespace_revision: 0, items: vec![( 0, StreamSnapshot { @@ -922,6 +938,7 @@ mod tests { let streams_snap = StreamsSnapshot { revision: 0, + namespace_revision: 0, items: vec![ ( 0, diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs index f4c275f33..330c2a0bd 100644 --- a/core/metadata/src/stm/stream.rs +++ b/core/metadata/src/stm/stream.rs @@ -853,6 +853,9 @@ define_state! { // it onto each new Partition::created_revision. Deterministic across // replicas: same ops, same order. pub revision: u64, + // Retirement proofs survive truncation and purge, but not a changed + // set of partition incarnations. + pub namespace_revision: u64, // Total pending cooperative revocations across all groups, recomputed // once per commit by `post_apply`. The consensus tick reads it O(1) // every 10ms instead of walking every stream/topic/group/member to @@ -2163,6 +2166,7 @@ impl StateHandler for DeleteStreamRequest { state.items.remove(stream_id); state.index.remove(&name); state.revision = state.revision.wrapping_add(1); + state.namespace_revision = state.revision; // The dropped stream may have held groups with pending revocations. state.recompute_consumer_group_metadata(); ApplyReply::ok(Bytes::new()) @@ -2274,6 +2278,7 @@ impl StateHandler for CreateTopicWithAssignmentsRequest { // monotonic revision and stamp every new partition with it. let new_revision = state.revision.wrapping_add(1); state.revision = new_revision; + state.namespace_revision = state.revision; // Share one `Arc<TopicStats>` across both left-right buffers via the // registry, parented to the stream's shared `Arc<StreamStats>`. The id @@ -2493,6 +2498,7 @@ impl StateHandler for DeleteTopicRequest { .stats_registry .remove_topic(stream_id, topic_id, &partition_ids); state.revision = state.revision.wrapping_add(1); + state.namespace_revision = state.revision; // The dropped topic may have held groups with pending revocations. state.recompute_consumer_group_metadata(); ApplyReply::ok(Bytes::new()) @@ -2612,6 +2618,7 @@ impl StateHandler for CreatePartitionsWithAssignmentsRequest { let new_revision = state.revision.wrapping_add(1); state.revision = new_revision; + state.namespace_revision = state.revision; let Some(stream) = state.items.get_mut(stream_id) else { return ApplyReply::err(CreatePartitionsResult::StreamNotFound); @@ -2681,6 +2688,7 @@ impl StateHandler for DeletePartitionsRequest { .stats_registry .remove_partitions(stream_id, topic_id, &removed_ids); state.revision = state.revision.wrapping_add(1); + state.namespace_revision = state.revision; } ApplyReply::ok(Bytes::new()) } @@ -2699,6 +2707,7 @@ pub struct StreamsSnapshot { /// `#[serde(default)]` so older snapshots restore at revision 0. #[serde(default)] pub revision: u64, + pub namespace_revision: u64, } impl Snapshotable for Streams { @@ -2777,6 +2786,7 @@ impl Snapshotable for Streams { StreamsSnapshot { items, revision: inner.revision, + namespace_revision: inner.namespace_revision, } }) } @@ -2921,6 +2931,7 @@ impl StreamsInner { index, items, revision: snapshot.revision, + namespace_revision: snapshot.namespace_revision, // Recomputed from the restored groups just below. pending_revocations_count: 0, consumer_group_members: AHashMap::new(), diff --git a/core/metadata/src/stm/user.rs b/core/metadata/src/stm/user.rs index 288b696c0..64a7652bc 100644 --- a/core/metadata/src/stm/user.rs +++ b/core/metadata/src/stm/user.rs @@ -549,7 +549,10 @@ impl StateHandler for UpdateUserRequest { // Patch, never replace: keys the client did not send keep their // current value, so a client that predates a key cannot erase it. user.options.extend(updated_options); - ApplyReply::ok(Bytes::new()) + ApplyReply { + revoked_user: (user.status != UserStatus::Active).then_some(user_id as UserId), + ..ApplyReply::ok(Bytes::new()) + } } } @@ -590,7 +593,10 @@ impl StateHandler for DeleteUserRequest { .permissioner .delete_permissions_for_user(user_id as UserId); } - ApplyReply::ok(Bytes::new()) + ApplyReply { + revoked_user: Some(user_id as UserId), + ..ApplyReply::ok(Bytes::new()) + } } } diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index e7d88ab86..de5a2e1f7 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -59,7 +59,7 @@ use iggy_binary_protocol::primitives::consumer::WireConsumer; use iggy_binary_protocol::requests::consumer_offsets::{ DeleteConsumerOffsetRequest, StoreConsumerOffsetRequest, }; -use iggy_binary_protocol::requests::system::SessionIdentity; +use iggy_binary_protocol::requests::system::{RetireSessionsRequest, SessionIdentity}; use iggy_binary_protocol::responses::messages::{ SendMessagesConfirmationResponse, SendMessagesResponse, }; @@ -719,6 +719,11 @@ where self.required_metadata_frontier = self.required_metadata_frontier.max(created_revision); } + #[must_use] + pub const fn created_revision(&self) -> u64 { + self.created_revision + } + /// A report covers only an applied barrier on a serving replica. WAL replay /// or the paired checkpoint must restore this floor before admission. #[must_use] @@ -734,8 +739,11 @@ where fn pending_session_retirement(&self, client_id: u128, session: u64) -> bool { let matches_identity = |body: &[u8]| { - SessionIdentity::decode_from(body).is_ok_and(|identity| { - identity.client_id == client_id && identity.session == session + RetireSessionsRequest::decode_from(body).is_ok_and(|request| { + request + .identities + .iter() + .any(|identity| identity.client_id == client_id && identity.session == session) }) }; self.consensus.with_pipeline(|pipeline| { @@ -4524,7 +4532,7 @@ where // Retirement and the retiring identity must never occupy the same // pipeline. Otherwise a later commit could recreate its receipt. let retirement_conflict = if message.header().operation == Operation::RetireSession { - let Ok(identity) = SessionIdentity::decode_from(message.body()) else { + let Ok(retirement) = RetireSessionsRequest::decode_from(message.body()) else { Self::send_partition_deny_or_log( consensus, message.header(), @@ -4535,8 +4543,10 @@ where .await; return; }; - consensus.pending_request(identity.client_id).is_some() - || self.pending_session_retirement(identity.client_id, identity.session) + retirement.identities.iter().any(|identity| { + consensus.pending_request(identity.client_id).is_some() + || self.pending_session_retirement(identity.client_id, identity.session) + }) } else { !is_auto_commit_client(client_id) && self.pending_session_retirement(client_id, message.header().session) @@ -5716,12 +5726,12 @@ where match header.operation { Operation::RetireSession => { - let identity = SessionIdentity::decode_from(message.body()) + let retirement = RetireSessionsRequest::decode_from(message.body()) .map_err(|_| IggyError::InvalidCommand)?; - if identity.client_id == 0 - || is_auto_commit_client(identity.client_id) - || identity.session == 0 - || identity.metadata_watermark < identity.session + if retirement + .identities + .iter() + .any(|identity| is_auto_commit_client(identity.client_id)) { return Err(IggyError::InvalidCommand); } @@ -6493,10 +6503,12 @@ where let committed_batch_stats = self.resolve_committed_batch_stats(&drained); let mut retirements = Vec::new(); for (entry, batch_stats) in drained.iter().zip(&committed_batch_stats) { + let configured_capacity = + (!self.dedup.capacity_committed()).then(|| self.dedup.capacity()); let retirement = (entry.header.operation == Operation::RetireSession) .then(|| { let prepare = self.log.journal().inner.repair_entry(entry.header.op)?; - SessionIdentity::decode_from( + RetireSessionsRequest::decode_from( prepare .as_slice() .get(iggy_binary_protocol::HEADER_SIZE..)?, @@ -6523,6 +6535,16 @@ where }); return; } + if let Some(configured_capacity) = configured_capacity + && configured_capacity != self.dedup.capacity() + { + warn!( + namespace_raw, + configured_capacity, + committed_capacity = self.dedup.capacity(), + "partition retry capacity is fixed by committed state; ignoring local configuration" + ); + } if let Some(identity) = retirement { retirements.push((entry.header.op, identity)); } @@ -6661,12 +6683,14 @@ where let mut retirements = retirements.into_iter().peekable(); for (mut entry, batch_stats) in drained.into_iter().zip(committed_batch_stats) { let prepare_header = entry.header; - if let Some((_, identity)) = retirements.next_if(|(op, _)| *op == prepare_header.op) { - self.dedup - .forget_session(identity.client_id, identity.session); - self.required_metadata_frontier = self - .required_metadata_frontier - .max(identity.metadata_watermark); + if let Some((_, retirement)) = retirements.next_if(|(op, _)| *op == prepare_header.op) { + for identity in retirement.identities { + self.dedup + .forget_session(identity.client_id, identity.session); + self.required_metadata_frontier = self + .required_metadata_frontier + .max(identity.metadata_watermark); + } } // Fold the committed request into this group's dedup slice. Runs on // EVERY replica, not just the one that replies, so a promoted @@ -9390,7 +9414,7 @@ mod tests { use crate::iggy_index::{IGGY_INDEX_SIZE, IggyIndex, IggyIndexCache}; use crate::iggy_index_reader::IggyIndexReader; use crate::poll_plan::{DiskReadOutcome, SealedSegmentHandle}; - use bytes::Bytes; + use bytes::{Bytes, BytesMut}; use compio::io::AsyncWriteAtExt; use consensus::LocalPipeline; use iggy_binary_protocol::batch::BATCH_MESSAGE_HEADER_SIZE; @@ -10391,7 +10415,22 @@ mod tests { namespace: IggyNamespace, identity: SessionIdentity, ) -> Message<RoutedRequestHeader> { - let body = identity.to_bytes(); + retire_sessions_request(namespace, &[identity]) + } + + fn retire_sessions_request( + namespace: IggyNamespace, + identities: &[SessionIdentity], + ) -> Message<RoutedRequestHeader> { + let mut body = BytesMut::new(); + for identity in identities { + identity.encode(&mut body); + } + let watermark = identities + .iter() + .map(|identity| identity.metadata_watermark) + .max() + .unwrap(); let size = size_of::<RoutedRequestHeader>() + body.len(); let mut request = Message::<RoutedRequestHeader>::new(size).transmute_header( |_, header: &mut RoutedRequestHeader| { @@ -10400,8 +10439,8 @@ mod tests { header.cluster = TEST_CLUSTER; header.client = message_bus::AUTO_COMMIT_CLIENT_ID; header.session = 1; - header.request = identity.metadata_watermark; - header.metadata_watermark = identity.metadata_watermark; + header.request = watermark; + header.metadata_watermark = watermark; header.group = namespace.inner(); header.size = u32::try_from(size).unwrap(); }, @@ -10410,6 +10449,59 @@ mod tests { request } + #[compio::test] + async fn given_ended_sessions_when_retiring_batch_should_use_one_partition_commit() { + const LIVE_CLIENT: u128 = 3; + let (mut partition, _) = recording_partition_at(0, 3); + for client_id in 1..=LIVE_CLIENT { + let mut send = checksumless_send_request(partition.namespace(), 1).transmute_header( + |header, next: &mut RoutedRequestHeader| { + *next = header; + next.client = client_id; + }, + ); + send.as_mut_slice()[size_of::<RoutedRequestHeader>()..] + .copy_from_slice(&build_segment_record(partition.namespace(), 0)); + partition.on_request(send, None).await; + } + partition.consensus.advance_commit_max(3); + partition.commit_journal(&repair_config()).await; + let identities = [ + SessionIdentity { + client_id: 1, + session: 1, + metadata_watermark: 10, + }, + SessionIdentity { + client_id: 2, + session: 1, + metadata_watermark: 11, + }, + ]; + partition + .on_request( + retire_sessions_request(partition.namespace(), &identities), + None, + ) + .await; + assert_eq!(partition.consensus.sequencer().current_sequence(), 4); + partition.consensus.advance_commit_max(4); + partition.commit_journal(&repair_config()).await; + assert!(partition.fatal().is_none()); + assert_eq!( + partition.dedup.count(), + 1, + "one barrier must retire both ended sessions" + ); + assert!( + partition.dedup.contains(LIVE_CLIENT), + "active sessions keep their retry receipts" + ); + for identity in identities { + assert!(partition.session_retired(identity)); + } + } + #[compio::test] async fn given_interleaved_sends_and_retirements_when_committing_should_preserve_live_receipts() { @@ -10561,9 +10653,13 @@ mod tests { session: 1, metadata_watermark: 10, }; + let second_identity = SessionIdentity { + client_id: 3, + ..identity + }; partition .on_request( - retire_session_request(partition.namespace(), identity), + retire_sessions_request(partition.namespace(), &[identity, second_identity]), None, ) .await; @@ -10573,7 +10669,12 @@ mod tests { .with_pipeline(LocalPipeline::request_queue_len), 1 ); - let mut target = checksumless_send_request(partition.namespace(), 1); + let mut target = checksumless_send_request(partition.namespace(), 1).transmute_header( + |old, header: &mut RoutedRequestHeader| { + *header = old; + header.client = second_identity.client_id; + }, + ); target.as_mut_slice()[size_of::<RoutedRequestHeader>()..] .copy_from_slice(&build_segment_record(partition.namespace(), 0)); partition.on_request(target, None).await; diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs index 17f089964..22b818fa9 100644 --- a/core/partitions/src/iggy_partitions.rs +++ b/core/partitions/src/iggy_partitions.rs @@ -21,7 +21,7 @@ use crate::poll_plan::{PollPlan, PollReadResult}; use crate::types::PartitionsConfig; use crate::{IggyPartition, Partition, PollingArgs, PollingConsumer}; use crate::{PollCompletion, PollReplication}; -use ahash::AHashSet; +use ahash::AHashMap; use consensus::{ Consensus, Plane, PlaneIdentity, VsrConsensus, build_deny_reply_from_request_header, }; @@ -108,7 +108,7 @@ where /// `partitions` / `namespace_to_local` does NOT hold here. Compio's /// per-shard runtime is single-threaded, so runtime borrow checks /// suffice; callers must not hold a borrow across `.await`. - tombstoned: RefCell<AHashSet<IggyNamespace>>, + tombstoned: RefCell<AHashMap<IggyNamespace, Option<u64>>>, consumer_group_offsets_reconcile_epoch: Rc<Cell<u64>>, persistence_notifier: RefCell<Option<crate::PersistenceNotifier>>, /// Debug-only tripwire: counts live [`Self::with_partition`] borrows so @@ -133,7 +133,7 @@ where config, partitions: UnsafeCell::new(Vec::new()), namespace_to_local: UnsafeCell::new(BTreeMap::new()), - tombstoned: RefCell::new(AHashSet::new()), + tombstoned: RefCell::new(AHashMap::new()), consumer_group_offsets_reconcile_epoch: Rc::new(Cell::new(0)), persistence_notifier: RefCell::new(None), #[cfg(debug_assertions)] @@ -149,7 +149,7 @@ where partitions: UnsafeCell::new(Vec::with_capacity(capacity)), // BTreeMap has no capacity hint; the Vec above absorbs the sizing. namespace_to_local: UnsafeCell::new(BTreeMap::new()), - tombstoned: RefCell::new(AHashSet::new()), + tombstoned: RefCell::new(AHashMap::new()), consumer_group_offsets_reconcile_epoch: Rc::new(Cell::new(0)), persistence_notifier: RefCell::new(None), #[cfg(debug_assertions)] @@ -447,21 +447,37 @@ where } pub fn is_tombstoned(&self, namespace: &IggyNamespace) -> bool { - self.tombstoned.borrow().contains(namespace) + self.tombstoned.borrow().contains_key(namespace) } /// Snapshot every tombstoned namespace, including ones never /// materialised: a boot-time damage verdict fences a namespace before /// any partition exists, so it appears in no other view of this map. pub fn tombstoned_namespaces(&self) -> Vec<IggyNamespace> { - self.tombstoned.borrow().iter().copied().collect() + self.tombstoned.borrow().keys().copied().collect() } /// Mark a namespace as tombstoned. Callable from any task on the /// shard's runtime (reconciler sets the fence synchronously before /// awaiting disk delete). pub fn tombstone(&self, namespace: IggyNamespace) { - self.tombstoned.borrow_mut().insert(namespace); + self.tombstoned + .borrow_mut() + .entry(namespace) + .or_insert(None); + } + + /// A permanent recovery failure, unlike a temporary teardown tombstone. + /// Retirement must persist this incarnation's fence before reporting success. + pub fn fence(&self, namespace: IggyNamespace, created_revision: u64) { + self.tombstoned + .borrow_mut() + .insert(namespace, Some(created_revision)); + } + + #[must_use] + pub fn failed_revision(&self, namespace: &IggyNamespace) -> Option<u64> { + self.tombstoned.borrow().get(namespace).copied().flatten() } /// Clear a namespace tombstone. Pump-side hook called from diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs index 2519f5915..9ff71a20b 100644 --- a/core/partitions/src/lib.rs +++ b/core/partitions/src/lib.rs @@ -75,7 +75,8 @@ pub use offset_storage::delete_persisted_offset; pub use partition_storage::{ CREATED_REVISION_FILE, configure_consumer_offsets, configure_consumer_offsets_with_storage, create_partition_file_hierarchy, delete_partitions_from_disk, ensure_initial_segment, - hydrate_partition_log, read_created_revision, write_created_revision, + hydrate_partition_log, read_created_revision, read_revision_record, write_created_revision, + write_revision_record, }; pub use poll_plan::{PollPlan, PollReadResult}; pub use segment::Segment; diff --git a/core/partitions/src/partition_storage.rs b/core/partitions/src/partition_storage.rs index b64cca237..6902dcd7f 100644 --- a/core/partitions/src/partition_storage.rs +++ b/core/partitions/src/partition_storage.rs @@ -123,6 +123,9 @@ pub async fn read_created_revision(partition_dir: &str) -> std::io::Result<Optio /// Durably replace the 8-byte little-endian record `name` in `directory`: write /// a temporary sibling, sync it, rename it over the record, sync the directory. +/// +/// # Errors +/// Returns the underlying open, write, rename, or synchronization error. pub async fn write_revision_record( directory: &str, name: &str, @@ -148,6 +151,9 @@ pub async fn write_revision_record( /// Read the 8-byte little-endian record `name` in `directory`, `None` when it /// is absent. +/// +/// # Errors +/// Returns an I/O error or `InvalidData` if the record is not exactly eight bytes. pub async fn read_revision_record(directory: &str, name: &str) -> std::io::Result<Option<u64>> { let path = Path::new(directory).join(name); match compio::fs::read(&path) diff --git a/core/server/README.md b/core/server/README.md index 300dee6d7..2c01e8320 100644 --- a/core/server/README.md +++ b/core/server/README.md @@ -92,9 +92,9 @@ completion contracts. This release changes protocol and storage formats. Servers and compatible clients deploy together; mixed versions and rolling upgrades are unsupported. -Peers verify protocol, release and executable identity before admission. -Deploy the same executable artifact to every replica; stripped or independently -rebuilt binaries have different identities even when their release matches. +Peers verify protocol, release and storage-format identity before admission. +Executable packaging does not affect that identity, so stripping or rebuilding +the same compatible release does not by itself prevent a replica from joining. Protocol 0.11.1 is the coordinated release boundary. Intermediate development builds advertising that version are not a compatibility guarantee. An unsupported data directory is refused before WAL scanning or file changes. @@ -155,18 +155,40 @@ There is no automatic in-place migration for below-floor log certificates. Do no view numbers or delete superblocks or WAL directories to bypass the refusal: an empty history could then replace committed data during a view change. -A recovered partition with no durable prepare-WAL frontier is also fenced as -missing history. This includes a crash before its first frontier was published: -the remaining files cannot prove that the partition never accepted a write. -If every replica is in this state, none can supply the history needed to elect -a partition primary. Restarting the replicas does not clear the fence. +Before a partition can serve, initialization publishes its incarnation in +`partition-initialization/<namespace>/created.revision` under the system data +directory. This atomic record remains outside the partition directory, including +after that directory is lost. Preserve it with metadata and partition backups. +Storage format `IGGY-DURABLE-SESSIONS-3` requires this initialization contract; +older data directories are refused before mutation. + +Retirement also writes `retirement.fence` in that external namespace directory +before counting a permanently failed partition as retired. The fence names the +failed incarnation and keeps it offline on subsequent boots, even if its +partition directory is replaced. Healthy partitions and new logins can continue. +If the fence cannot be persisted, the session remains retained and retirement +retries. Temporary teardown tombstones and state transfer do not qualify as +permanent failures. + +Partition retirement barriers batch up to 128 ended sessions per commit. Each +session still requires retirement reports covering every allocated partition +from a common replica quorum before metadata releases its registry slot. Segment +deletion and purge do not invalidate those reports; changes to partition +incarnations do. Metadata snapshot format 8 persists that separate revision. + +A committed partition that has never initialized on this replica can finish +initialization after a crash, including when all replicas stopped before their +first WAL frontiers were published. An initialized partition with no durable +prepare-WAL frontier remains fenced as missing history. A singleton refuses to +serve it; a replicated partition requires history from a healthy peer. If every +replica lost initialized history, restarting does not clear the fence. Preserve the files and restore a verified consistent backup or use a validated recovery procedure. If the partition is independently known to be empty, or its data may be discarded, delete and recreate it through the metadata API. This creates a new partition incarnation and removes the old data. Missing files alone are not evidence that discarding that data is safe. Do not create a WAL -frontier or remove the recovery fence manually. +frontier or remove initialization records or recovery fences manually. ## Systemd integration diff --git a/core/server/config.toml b/core/server/config.toml index e89778a4b..6caaaf5bf 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -858,11 +858,13 @@ prepare_queue_depth = 32 # restore the previous value to recover. journal_slots = 1024 -# Slot count of the VSR client table: how many distinct clients (TCP/QUIC/WS -# virtual clients and HTTP sessions together) hold live session state at once. -# When full, the client whose last commit is oldest is evicted and its next -# request re-registers. The HTTP session cap tracks this at half, so raising -# it lifts both. Must be between 2 and 65536. +# Initial capacity of the metadata session registry, shared by binary clients +# and HTTP sessions. Live and unretired sessions occupy slots. A full registry +# refuses new sessions; it never evicts retry protection. The first committed +# operation fixes the capacity, so changing this value does not resize an +# existing registry after restart or transfer. The HTTP cache cap uses half +# the configured value, but cannot raise the committed registry limit. +# Must be between 2 and 65536. clients_table_max = 8192 # Cap on the partitions of all streams and topics. A request to create a @@ -968,11 +970,12 @@ validate_checksum = true # u128 bitset, and this depth bounds that suffix. prepare_queue_depth = 32 -# Distinct clients each partition group tracks request watermarks for, so a -# retried produce or consumer-offset write is answered instead of committing a -# second time. At capacity the client whose newest commit is oldest is evicted; -# that client's next replay re-executes, exactly as it would have before dedup -# existed, so under-sizing degrades rather than breaks. Must be > 0 and <= 65536. +# Initial capacity for session-qualified retry protection in each partition. +# Live and unretired sessions occupy slots. A full table refuses new writers +# with TransientNotAccepted; it never evicts existing protection. Ended sessions +# release slots only after ordered retirement. The first committed operation +# fixes the capacity, so changing this value does not resize an existing group +# after restart or transfer. Must be > 0 and <= 65536. # # Unlike [metadata] clients_table_max, this budget is PER GROUP, so worst-case # memory scales with partition count. Size it to the producers one partition diff --git a/core/server/src/boot/mod.rs b/core/server/src/boot/mod.rs index f864ed422..f3d81704e 100644 --- a/core/server/src/boot/mod.rs +++ b/core/server/src/boot/mod.rs @@ -90,7 +90,7 @@ use tracing::{error, info, warn}; const STORAGE_FORMAT_FILE: &str = "storage-format"; const STORAGE_FORMAT_TEMP_FILE: &str = "storage-format.tmp"; -const STORAGE_FORMAT: &[u8] = b"IGGY-DURABLE-SESSIONS-2\n"; +const STORAGE_FORMAT: &[u8] = b"IGGY-DURABLE-SESSIONS-3\n"; /// Load the server configuration from the active config provider. /// @@ -298,9 +298,6 @@ pub fn bootstrap( install_default_crypto_provider(); validate_root_credentials_env(&config)?; warm_dummy_password_hash(); - // Hash the executable before shard runtimes start doing asynchronous work. - message_bus::replica::handshake::binary_identity(crate::VERSION) - .map_err(ServerError::BinaryIdentity)?; // The sync GetStats read path has no access to server config, so capture // the data directory here for its disk-usage reporting. crate::sysinfo_probe::init_stats_data_path(config.get_system_path().into()); @@ -533,8 +530,10 @@ async fn shard_main( // shard's bus needs the handshake identity (the handshake itself // runs on the owning shard, not on shard 0). bus.set_replica_handshake_ctx(ReplicaHandshakeCtx { - binary_identity: message_bus::replica::handshake::binary_identity(crate::VERSION) - .map_err(ServerError::BinaryIdentity)?, + binary_identity: message_bus::replica::handshake::binary_identity( + crate::VERSION, + STORAGE_FORMAT, + ), cluster_id: topology.cluster_id, self_id: topology.self_replica_id, replica_count: topology.replica_count, @@ -1040,6 +1039,7 @@ async fn shard_main( let cleaner_shard = Rc::clone(&shard); let interval = config.consumer_group.heartbeat_interval.get_duration(); let timeout = config.consumer_group.session_timeout.get_duration(); + let system_path = config.get_system_path(); let handle = compio::runtime::spawn(async move { crate::consumer_group::liveness::run( cleaner_shard, @@ -1047,6 +1047,7 @@ async fn shard_main( stop_rx, interval, timeout, + system_path, ) .await; }); diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs index fcb986944..aa0ea2d76 100644 --- a/core/server/src/boot/recovery.rs +++ b/core/server/src/boot/recovery.rs @@ -215,7 +215,6 @@ pub(in crate::boot) async fn build_shard_for_thread( topology.cluster_id, topology.self_replica_id, topology.replica_count, - false, Rc::clone(&bus), &partitions, ) diff --git a/core/server/src/consumer_group.rs b/core/server/src/consumer_group.rs index c0ab103a1..f7b45b3a1 100644 --- a/core/server/src/consumer_group.rs +++ b/core/server/src/consumer_group.rs @@ -699,7 +699,7 @@ mod tests { /// Create a group and routes for two partitions, with no members or local /// partitions. Tests install partition state and apply joins explicitly. fn group_shard() -> Rc<TestShard> { - let shard = partition_read_shard(); + let shard = partition_read_shard(3); let mux = &shard.plane.metadata().mux_stm; mux.update(prepare_message( Operation::CreateStream, @@ -758,7 +758,7 @@ mod tests { /// Build a shard with its own inbox so tests can serve group progress reads /// and clears through the production message pump. - fn partition_read_shard() -> Rc<TestShard> { + pub(super) fn partition_read_shard(replica_count: u8) -> Rc<TestShard> { // These tests do not dispatch disk polls, so keep that lane minimal. const POLL_COMPLETION_CAPACITY: usize = 1; @@ -766,7 +766,7 @@ mod tests { let consensus = VsrConsensus::new( 1, 0, - 3, + replica_count, server_common::sharding::METADATA_GROUP, bus.clone(), LocalPipeline::new(), @@ -800,7 +800,7 @@ mod tests { replies, POLL_COMPLETION_CAPACITY, PapayaShardsTable::new(), - PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 3), bus), + PartitionConsensusConfig::new(1, ReplicaTopology::new(0, replica_count), bus), None, ShardMetrics::for_shard(), ) @@ -911,7 +911,7 @@ mod tests { /// Poll the shard's message pump alongside the operation until it finishes. /// This serves progress reads and also executes any requested stale clears. - async fn run_with_partition_message_pump<T>( + pub(super) async fn run_with_partition_message_pump<T>( shard: &Rc<TestShard>, operation: impl Future<Output = T>, ) -> T { diff --git a/core/server/src/consumer_group/lease.rs b/core/server/src/consumer_group/lease.rs index 8f3b2882d..ccce26e0b 100644 --- a/core/server/src/consumer_group/lease.rs +++ b/core/server/src/consumer_group/lease.rs @@ -77,6 +77,8 @@ impl RetirementProgress { } } +type RetirementNamespaces = Rc<[(IggyNamespace, u64)]>; + /// Bounded by the session registry, independent of heartbeat volume. #[derive(Default)] pub struct ConsumerGroupLiveness { @@ -88,7 +90,8 @@ pub struct ConsumerGroupLiveness { pub(super) report_offset: usize, pub(super) report_incomplete: bool, pub(super) retirement: Option<RetirementProgress>, - pub(super) retirement_namespaces: Option<(u64, Rc<[IggyNamespace]>)>, + pub(super) retirement_namespaces: Option<(u64, RetirementNamespaces)>, + pub(super) retirement_fences: BTreeMap<IggyNamespace, u64>, } impl ConsumerGroupLiveness { diff --git a/core/server/src/consumer_group/liveness.rs b/core/server/src/consumer_group/liveness.rs index 44b867754..d3f78d52b 100644 --- a/core/server/src/consumer_group/liveness.rs +++ b/core/server/src/consumer_group/liveness.rs @@ -25,6 +25,9 @@ use std::time::{Duration, Instant}; use bytes::BytesMut; use consensus::{Consensus, MetadataHandle}; +use iggy_binary_protocol::requests::system::{ + MAX_SESSIONS_PER_RETIREMENT, RetireSessionsRequest, SessionIdentity, +}; use iggy_binary_protocol::{ Command, ConsensusHeader, ConsumerSession, ConsumerSessionHeartbeatHeader, HEADER_SIZE, MAX_CONSUMER_SESSIONS_PER_HEARTBEAT, PrepareHeader, WireEncode, @@ -60,6 +63,7 @@ pub async fn run( stop: Receiver<()>, interval: Duration, timeout: Duration, + system_path: String, ) { let shutdown = shard.bus.token(); let mut next_report = Instant::now() + interval; @@ -83,8 +87,16 @@ pub async fn run( } if Instant::now() >= next_report { loop { - match report_and_expire(&shard, &liveness, &stop, &shutdown, interval, timeout) - .await + match report_and_expire( + &shard, + &liveness, + &stop, + &shutdown, + interval, + timeout, + &system_path, + ) + .await { Pass::Stopped => break 'running, Pass::HitCap => {} @@ -93,7 +105,7 @@ pub async fn run( } next_report = Instant::now() + interval; } else { - retire_sessions(&shard, &liveness, &stop, &shutdown, interval).await; + retire_sessions(&shard, &liveness, &stop, &shutdown, interval, &system_path).await; } } } @@ -105,6 +117,7 @@ async fn report_and_expire<B, MJ, S, SB>( shutdown: &ShutdownToken, report_interval: Duration, timeout: Duration, + system_path: &str, ) -> Pass where B: ShellBus, @@ -156,12 +169,28 @@ where // Even an empty gather must report incompleteness to the primary. let pass = report_sessions(shard, liveness, &sessions, incomplete, stop, shutdown).await; if pass != Pass::Stopped { - retire_sessions(shard, liveness, stop, shutdown, report_interval).await; + retire_sessions( + shard, + liveness, + stop, + shutdown, + report_interval, + system_path, + ) + .await; } return pass; } - retire_sessions(shard, liveness, stop, shutdown, report_interval).await; + retire_sessions( + shard, + liveness, + stop, + shutdown, + report_interval, + system_path, + ) + .await; expire_sessions(shard, liveness, stop, shutdown, timeout, now).await } @@ -248,13 +277,50 @@ where .unwrap_or(Pass::Drained) } -#[allow(clippy::too_many_lines)] async fn retire_sessions<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, liveness: &RefCell<ConsumerGroupLiveness>, stop: &Receiver<()>, shutdown: &ShutdownToken, report_interval: Duration, + system_path: &str, +) where + B: ShellBus, + MJ: JournalHandle + 'static, + MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, + S: 'static, + SB: SuperblockStore + 'static, +{ + // Already-batched partition barriers can cover several identities. Drain + // their metadata finalizations within the same heartbeat budget. + let _ = shard::bus_timeout(&shard.bus, CONSUMER_SESSION_REPORT_TIMEOUT, async { + for _ in 0..MAX_LOGOUTS_PER_PASS { + let before = shard.plane.metadata().client_table.borrow().count(); + retire_session( + shard, + liveness, + stop, + shutdown, + report_interval, + system_path, + ) + .await; + if shard.plane.metadata().client_table.borrow().count() >= before { + break; + } + } + }) + .await; +} + +#[allow(clippy::too_many_lines)] +async fn retire_session<B, MJ, S, SB>( + shard: &Rc<ShellShard<B, MJ, S, SB>>, + liveness: &RefCell<ConsumerGroupLiveness>, + stop: &Receiver<()>, + shutdown: &ShutdownToken, + report_interval: Duration, + system_path: &str, ) where B: ShellBus, MJ: JournalHandle + 'static, @@ -283,7 +349,10 @@ async fn retire_sessions<B, MJ, S, SB>( liveness.borrow_mut().reconcile_retirement(None, 0); return; }; - let revision = metadata.mux_stm.streams().read(|inner| inner.revision); + let revision = metadata + .mux_stm + .streams() + .read(|inner| inner.namespace_revision); liveness .borrow_mut() .reconcile_retirement(Some(identity), revision); @@ -313,6 +382,18 @@ async fn retire_sessions<B, MJ, S, SB>( (progress.cursor, progress.reported) }; if !reported { + let mut identities: Vec<_> = metadata.client_table.borrow().ended_sessions().collect(); + identities + .sort_unstable_by_key(|identity| (identity.metadata_watermark, identity.client_id)); + let batch_limit = MAX_SESSIONS_PER_RETIREMENT.min( + shard.bus_max_message_size().saturating_sub(HEADER_SIZE) + / SessionIdentity::ENCODED_SIZE, + ); + if batch_limit == 0 { + return; + } + identities.truncate(batch_limit); + let body = RetireSessionsRequest { identities }.to_bytes(); let namespaces = { let mut tracker = liveness.borrow_mut(); if tracker @@ -320,6 +401,7 @@ async fn retire_sessions<B, MJ, S, SB>( .as_ref() .is_none_or(|(cached_revision, _)| *cached_revision != revision) { + tracker.retirement_fences.clear(); tracker.retirement_namespaces = Some(( revision, metadata.mux_stm.streams().read(|inner| { @@ -327,10 +409,13 @@ async fn retire_sessions<B, MJ, S, SB>( for (_, stream) in &inner.items { for (topic_id, topic) in &stream.topics { for partition in &topic.partitions { - namespaces.push(server_common::sharding::IggyNamespace::new( - stream.id, - topic_id, - partition.id, + namespaces.push(( + server_common::sharding::IggyNamespace::new( + stream.id, + topic_id, + partition.id, + ), + partition.created_revision, )); } } @@ -347,7 +432,7 @@ async fn retire_sessions<B, MJ, S, SB>( // Repeated barriers are idempotent, so an unfinished sweep can resume // without delaying activity reports past their existing budget. let completed = shard::bus_timeout(&shard.bus, CONSUMER_SESSION_REPORT_TIMEOUT, async { - for namespace in namespaces + for (namespace, created_revision) in namespaces .iter() .skip(cursor) .take(MAX_RETIREMENTS_PER_PASS) @@ -358,17 +443,47 @@ async fn retire_sessions<B, MJ, S, SB>( if consensus.view() != view || !consensus.is_normal() { return false; } - let retired = matches!( - shard - .partition_read( - *namespace, - shard::PartitionRead::SessionRetired { identity } - ) - .await, - Some(shard::PartitionReadReply::SessionRetired(true)) - ); + let status = shard + .partition_read( + *namespace, + shard::PartitionRead::SessionRetired { identity }, + ) + .await; + let retired = match status { + Some(shard::PartitionReadReply::SessionRetired(true)) => true, + Some(shard::PartitionReadReply::SessionRetirementFailed { + created_revision: failed_revision, + }) if failed_revision == *created_revision => { + if liveness.borrow().retirement_fences.get(namespace) + == Some(created_revision) + { + true + } else { + match crate::partition_helpers::record_partition_retirement_fence( + system_path, + *namespace, + *created_revision, + ) + .await + { + Ok(()) => { + liveness + .borrow_mut() + .retirement_fences + .insert(*namespace, *created_revision); + true + } + Err(error) => { + tracing::warn!(%error, namespace_raw = namespace.inner(), + "retirement waits for durable partition fence"); + false + } + } + } + } + _ => false, + }; if !retired { - let body = identity.to_bytes(); let size = HEADER_SIZE + body.len(); let mut request = Message::<iggy_binary_protocol::RoutedRequestHeader>::new(size) @@ -436,7 +551,11 @@ async fn retire_sessions<B, MJ, S, SB>( if consensus.view() != view || !consensus.is_normal() || consensus.is_transferring() - || metadata.mux_stm.streams().read(|inner| inner.revision) != revision + || metadata + .mux_stm + .streams() + .read(|inner| inner.namespace_revision) + != revision { return; } @@ -611,7 +730,7 @@ mod tests { SpyBus, TestShard, prepare_message, register_reply, test_shard, }; use consensus::client_table::ClientTable; - use consensus::{LocalPipeline, Sequencer, ViewRestore, VsrConsensus}; + use consensus::{LocalPipeline, PartitionsHandle, Sequencer, ViewRestore, VsrConsensus}; use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment; use iggy_binary_protocol::requests::consumer_groups::CreateConsumerGroupRequest; use iggy_binary_protocol::requests::streams::CreateStreamRequest; @@ -623,6 +742,7 @@ mod tests { use message_bus::lifecycle::Shutdown; use metadata::stm::StateMachine; use metadata::stm::consumer_group::JoinConsumerGroupRequest; + use metadata::stm::stream::TruncatePartitionRequest; use server_common::MessageBag; use server_common::sharding::METADATA_GROUP; @@ -819,7 +939,7 @@ mod tests { let tracker = RefCell::new(ConsumerGroupLiveness::default()); let (_stop, receiver) = shard::channel(1); let (_signal, shutdown) = Shutdown::new(); - report_and_expire(&shard, &tracker, &receiver, &shutdown, TIMEOUT, TIMEOUT).await; + report_and_expire(&shard, &tracker, &receiver, &shutdown, TIMEOUT, TIMEOUT, "").await; assert_eq!(tracker.borrow().leases[&CLIENT].session, Some(SESSION)); let expired_at = Instant::now() + TIMEOUT * 2; assert_eq!( @@ -848,7 +968,8 @@ mod tests { let (_signal, shutdown) = Shutdown::new(); let tracker = RefCell::new(ConsumerGroupLiveness::default()); assert_eq!( - report_and_expire(&shard, &tracker, &receiver, &shutdown, TIMEOUT, TIMEOUT).await, + report_and_expire(&shard, &tracker, &receiver, &shutdown, TIMEOUT, TIMEOUT, "") + .await, Pass::Drained ); if replica == 0 { @@ -1018,7 +1139,9 @@ mod tests { async fn shard_with_members(count: usize) -> (tempfile::TempDir, Rc<TestShard>) { let dir = tempfile::tempdir().unwrap(); - let mut shard = test_shard(&SpyBus::default(), 0, 1, 1); + let mut shard = Rc::try_unwrap(super::super::tests::partition_read_shard(1)) + .ok() + .unwrap(); let metadata = shard.plane.metadata_mut(); metadata.journal = Some( PrepareJournal::open(&dir.path().join("journal.wal"), 0) @@ -1331,7 +1454,10 @@ mod tests { .ended_sessions() .next() .unwrap(); - let revision = metadata.mux_stm.streams().read(|inner| inner.revision); + let revision = metadata + .mux_stm + .streams() + .read(|inner| inner.namespace_revision); let tracker = RefCell::new(ConsumerGroupLiveness::default()); { let mut tracker = tracker.borrow_mut(); @@ -1345,7 +1471,7 @@ mod tests { stop.try_send(()).unwrap(); let (_signal, shutdown) = Shutdown::new(); let sweep = Box::pin(retire_sessions( - &shard, &tracker, &receiver, &shutdown, TIMEOUT, + &shard, &tracker, &receiver, &shutdown, TIMEOUT, "", )); assert!( futures::poll!(sweep).is_ready(), @@ -1410,6 +1536,206 @@ mod tests { assert_eq!(tracker.retirement.as_ref().unwrap().reporters.len(), 1); } + #[compio::test] + async fn given_failed_partition_when_retiring_should_allow_new_sessions() { + let (directory, shard) = shard_with_members(1).await; + let metadata = shard.plane.metadata(); + assert!( + metadata + .client_table + .borrow_mut() + .end_session(CLIENT, 0, SESSION, SESSION) + ); + shard + .plane + .partitions() + .fence(server_common::sharding::IggyNamespace::new(0, 0, 0), 1); + let tracker = RefCell::new(ConsumerGroupLiveness::default()); + let (_stop, receiver) = shard::channel(1); + let (_signal, shutdown) = Shutdown::new(); + super::super::tests::run_with_partition_message_pump( + &shard, + retire_sessions( + &shard, + &tracker, + &receiver, + &shutdown, + TIMEOUT, + directory.path().to_str().unwrap(), + ), + ) + .await; + assert_eq!( + metadata.client_table.borrow().count(), + 0, + "one failed partition must not fill the session registry permanently" + ); + assert_eq!( + crate::partition_helpers::partition_retirement_fence( + directory.path().to_str().unwrap(), + server_common::sharding::IggyNamespace::new(0, 0, 0), + ) + .await + .unwrap(), + Some(1) + ); + metadata + .client_table + .borrow_mut() + .commit_register( + CLIENT + 1, + 0, + [0x6b; 32], + register_reply(CLIENT + 1, SESSION + 2), + ) + .expect("retirement must release capacity for a new login"); + } + + #[compio::test] + async fn given_failed_fence_write_when_retiring_should_retain_session_until_retry() { + let (directory, shard) = shard_with_members(1).await; + let system_path = directory.path().to_str().unwrap(); + let metadata = shard.plane.metadata(); + assert!( + metadata + .client_table + .borrow_mut() + .end_session(CLIENT, 0, SESSION, SESSION) + ); + let namespace = server_common::sharding::IggyNamespace::new(0, 0, 0); + shard.plane.partitions().fence(namespace, 1); + let obstruction = directory.path().join("partition-initialization"); + std::fs::write(&obstruction, b"not a directory").unwrap(); + let tracker = RefCell::new(ConsumerGroupLiveness::default()); + let (_stop, receiver) = shard::channel(1); + let (_signal, shutdown) = Shutdown::new(); + super::super::tests::run_with_partition_message_pump( + &shard, + retire_sessions(&shard, &tracker, &receiver, &shutdown, TIMEOUT, system_path), + ) + .await; + assert_eq!( + metadata.client_table.borrow().count(), + 1, + "an unpersisted fence is not a retirement proof" + ); + std::fs::remove_file(obstruction).unwrap(); + super::super::tests::run_with_partition_message_pump( + &shard, + retire_sessions(&shard, &tracker, &receiver, &shutdown, TIMEOUT, system_path), + ) + .await; + assert_eq!(metadata.client_table.borrow().count(), 0); + } + + #[compio::test] + async fn given_teardown_or_stale_failure_when_retiring_should_not_fence_current_incarnation() { + for failed_revision in [None, Some(0)] { + let (directory, shard) = shard_with_members(1).await; + let metadata = shard.plane.metadata(); + assert!( + metadata + .client_table + .borrow_mut() + .end_session(CLIENT, 0, SESSION, SESSION) + ); + let namespace = server_common::sharding::IggyNamespace::new(0, 0, 0); + if let Some(revision) = failed_revision { + shard.plane.partitions().fence(namespace, revision); + } else { + shard.plane.partitions().tombstone(namespace); + } + let tracker = RefCell::new(ConsumerGroupLiveness::default()); + let (_stop, receiver) = shard::channel(1); + let (_signal, shutdown) = Shutdown::new(); + super::super::tests::run_with_partition_message_pump( + &shard, + retire_sessions( + &shard, + &tracker, + &receiver, + &shutdown, + TIMEOUT, + directory.path().to_str().unwrap(), + ), + ) + .await; + assert_eq!(metadata.client_table.borrow().count(), 1); + assert_eq!( + crate::partition_helpers::partition_retirement_fence( + directory.path().to_str().unwrap(), + namespace, + ) + .await + .unwrap(), + None + ); + } + } + + #[compio::test] + async fn given_segment_deletion_when_retiring_should_keep_sweep_progress() { + let (_directory, shard) = shard_with_members(1).await; + let metadata = shard.plane.metadata(); + assert!( + metadata + .client_table + .borrow_mut() + .end_session(CLIENT, 0, SESSION, SESSION) + ); + let identity = metadata + .client_table + .borrow() + .ended_sessions() + .next() + .unwrap(); + let mut tracker = ConsumerGroupLiveness::default(); + tracker.reconcile_retirement( + Some(identity), + metadata + .mux_stm + .streams() + .read(|inner| inner.namespace_revision), + ); + let progress = tracker.retirement.as_mut().unwrap(); + progress.cursor = 1; + progress.reporters.insert(1); + let truncate = TruncatePartitionRequest { + stream_id: WireIdentifier::numeric(0), + topic_id: WireIdentifier::numeric(0), + partition_id: 0, + up_to_offset: 0, + }; + for _ in 0..3 { + assert_eq!( + metadata + .mux_stm + .update(prepare_message( + Operation::TruncatePartition, + CLIENT, + 1, + &truncate.to_bytes() + )) + .unwrap() + .code, + 0 + ); + tracker.reconcile_retirement( + Some(identity), + metadata + .mux_stm + .streams() + .read(|inner| inner.namespace_revision), + ); + let progress = tracker.retirement.as_ref().unwrap(); + assert_eq!( + progress.cursor, 1, + "segment deletion must not restart session retirement" + ); + assert!(progress.reporters.contains(&1)); + } + } + #[test] fn bounded_heartbeat_round_trips_through_replica_routing() { let sessions = vec![ diff --git a/core/server/src/dispatch/mod.rs b/core/server/src/dispatch/mod.rs index 91099ba7a..21733e720 100644 --- a/core/server/src/dispatch/mod.rs +++ b/core/server/src/dispatch/mod.rs @@ -830,16 +830,20 @@ mod tests { use crate::cluster_meta::ClusterRoster; use crate::dispatch::host::ServerHost; use crate::dispatch::test_support::{ - FIRST_BOOT, SpyBus, TestMux, TestShard, prepare_message, register_reply, test_shard, + FIRST_BOOT, SpyBus, TestMux, TestShard, prepare_message, register_reply, request_message, + test_shard, }; + use consensus::Sequencer; use consensus::client_table::bind_verifier; use iggy_binary_protocol::Command; use iggy_binary_protocol::codes::{ GET_CLUSTER_METADATA_CODE, GET_CONSUMER_OFFSET_CODE, POLL_MESSAGES_CODE, }; - use iggy_binary_protocol::requests::users::UpdateUserRequest; + use iggy_binary_protocol::requests::users::{ + CreateUserRequest, DeleteUserRequest, UpdateUserRequest, + }; use iggy_binary_protocol::{EvictionHeader, ReplyHeader}; - use iggy_binary_protocol::{WireEncode, WireIdentifier, WireOptions}; + use iggy_binary_protocol::{WireEncode, WireIdentifier, WireName, WireOptions}; use journal::prepare_journal::PrepareJournal; use message_bus::installer::conn_info::ClientTransportKind; use metadata::IggyMetadata; @@ -931,6 +935,184 @@ mod tests { } } + #[compio::test] + async fn given_deleted_user_when_id_reused_should_reject_retained_session() { + for primary in [false, true] { + assert_revoked_user_cannot_resume(Operation::DeleteUser, primary).await; + } + } + + #[compio::test] + async fn given_deactivated_user_when_reactivated_should_reject_retained_session() { + for primary in [false, true] { + assert_revoked_user_cannot_resume(Operation::UpdateUser, primary).await; + } + } + + #[allow(clippy::too_many_lines)] + async fn assert_revoked_user_cannot_resume(operation: Operation, primary: bool) { + const CLIENT: u128 = 10; + const TRANSPORT: u128 = 20; + const SESSION: u64 = 2; + const USER: u32 = 1; + const ROOT_CLIENT: u128 = CLIENT + 1; + const SECRET: [u8; 32] = [0x5a; 32]; + let bus = SpyBus::default(); + let directory = tempfile::tempdir().unwrap(); + let mut shard = test_shard(&bus, 0, 1, FIRST_BOOT); + shard.plane.metadata_mut().journal = Some( + PrepareJournal::open(&directory.path().join("journal.wal"), 0) + .await + .unwrap(), + ); + let shard = Rc::new(shard); + let metadata = shard.plane.metadata(); + metadata.mux_stm.users().ensure_root_user("iggy", "hash"); + let create = CreateUserRequest { + username: WireName::new("original").unwrap(), + password: "hash".to_owned(), + status: UserStatus::Active.as_code(), + permissions: None, + options: WireOptions::empty(), + }; + let created = metadata + .mux_stm + .update(prepare_message( + Operation::CreateUser, + 0, + 1, + &create.to_bytes(), + )) + .unwrap(); + assert_eq!(created.code, 0); + metadata + .client_table + .borrow_mut() + .commit_register( + CLIENT, + USER, + bind_verifier(CLIENT, USER, &SECRET), + register_reply(CLIENT, SESSION), + ) + .unwrap(); + metadata.applied_frontier().advance(SESSION); + let sessions = Rc::new(RefCell::new(SessionManager::new())); + sessions.borrow_mut().ensure_connection( + TRANSPORT, + "127.0.0.1:34567".parse().unwrap(), + ClientTransportKind::Tcp, + ); + let identity = SessionIdentity { + client_id: CLIENT, + session: SESSION, + metadata_watermark: SESSION, + }; + complete_session_binding( + &shard, + &sessions, + TRANSPORT, + identity, + BindSecret::new(Box::new(SECRET)), + ) + .await + .unwrap(); + let body = if operation == Operation::DeleteUser { + DeleteUserRequest { + user_id: WireIdentifier::numeric(USER), + } + .to_bytes() + } else { + UpdateUserRequest { + user_id: WireIdentifier::numeric(USER), + username: None, + status: Some(UserStatus::Inactive.as_code()), + options: WireOptions::empty(), + } + .to_bytes() + }; + if primary { + metadata + .client_table + .borrow_mut() + .commit_register(ROOT_CLIENT, 0, SECRET, register_reply(ROOT_CLIENT, 1)) + .unwrap(); + let consensus = metadata.consensus.as_ref().unwrap(); + consensus.restore_commit_state(SESSION, SESSION); + consensus.sequencer().set_sequence(SESSION); + let request = request_message(operation, ROOT_CLIENT, 1, 1, &body).transmute_header( + |old, header: &mut RoutedRequestHeader| { + *header = old; + header.metadata_watermark = SESSION; + }, + ); + metadata.submit_request_in_process(request).await.unwrap(); + } else { + let revocation = prepare_message(operation, 0, 3, &body).transmute_header( + |old, header: &mut iggy_binary_protocol::PrepareHeader| { + *header = old; + header.op = 3; + }, + ); + metadata::apply_committed_prepare( + &*metadata.mux_stm, + &metadata.client_table, + true, + |_| {}, + consensus::seal_prepare_checksum(revocation), + ); + } + let old_binding = sessions.borrow().get_session(TRANSPORT); + let restore = if operation == Operation::DeleteUser { + prepare_message( + Operation::CreateUser, + 0, + 4, + &CreateUserRequest { + username: WireName::new("replacement").unwrap(), + ..create + } + .to_bytes(), + ) + } else { + prepare_message( + Operation::UpdateUser, + 0, + 4, + &UpdateUserRequest { + user_id: WireIdentifier::numeric(USER), + username: None, + status: Some(UserStatus::Active.as_code()), + options: WireOptions::empty(), + } + .to_bytes(), + ) + }; + assert_eq!(metadata.mux_stm.update(restore).unwrap().code, 0); + assert!(metadata.mux_stm.users().read(|users| { + users + .items + .get(USER as usize) + .is_some_and(|user| user.status == UserStatus::Active) + })); + let outcome = complete_session_binding( + &shard, + &sessions, + TRANSPORT, + identity, + BindSecret::new(Box::new(SECRET)), + ) + .await; + assert!( + matches!(outcome, Err(IggyError::Unauthenticated)), + "committed {operation:?} must reject the old secret after restoring user id {USER}: {outcome:?}" + ); + assert!( + old_binding.is_none(), + "committed {operation:?} must invalidate an already authenticated connection" + ); + assert_eq!(metadata.client_table.borrow().ended_sessions().count(), 1); + } + /// A test shard wired to its own lanes (the held sender feeds them), /// for the reply-lane pump tests below. fn reply_lane_test_shard(name: &str) -> (SpyBus, shard::TaggedSender, Rc<TestShard>) { diff --git a/core/server/src/http/admission.rs b/core/server/src/http/admission.rs index e88f79a00..7ca5d23e5 100644 --- a/core/server/src/http/admission.rs +++ b/core/server/src/http/admission.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -//! In-flight admission for awaited partition writes: the per-session / +//! In-flight admission for partition writes, including detached `NoAck` replies: the per-session / //! shard-global budget guard. Coupled to the router body cap because the shard //! inbox is one shared bounded channel, so admission is correctness-adjacent. @@ -37,10 +37,11 @@ const MAX_IN_FLIGHT_WRITES_PER_SESSION: u32 = 32; /// consensus. The budget therefore bounds both starvation terms: budget x /// `max_request_size` bounds the worst-case buffered bytes, and budget x /// per-request CPU bounds how far admitted HTTP work can delay the consensus -/// pump. `?ack=none` produces are admitted through the same caps: they install -/// no reply slot and never await a commit, but they still park inside dispatch -/// for the routable-wait budget while pinning their buffered body, so leaving -/// them uncapped would bypass both terms. A session that saturates its own cap +/// pump. `?ack=none` returns to the caller after dispatch, while a detached task +/// retains the reply slot, partition gate and admission guard until a reply or +/// deadline. Its 202 response does not acknowledge admission or commit. Holding +/// the budget through that wait prevents `NoAck` traffic from bypassing these +/// limits. A session that saturates its own cap /// reads its own 429 before it can spill onto the shared budget. const MAX_IN_FLIGHT_WRITES_GLOBAL: u32 = 128; diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index f0715e28a..91b28d5eb 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -60,6 +60,9 @@ use std::sync::Arc; use std::sync::atomic::Ordering; use tracing::{error, warn}; +const PARTITION_INITIALIZATION_DIR: &str = "partition-initialization"; +const RETIREMENT_FENCE_FILE: &str = "retirement.fence"; + /// Open the durable superblock for one partition's consensus group and read /// back the last recorded VSR state. /// @@ -179,10 +182,46 @@ pub async fn load_partition_or_fence( cluster_id: u128, self_replica_id: u8, replica_count: u8, - allow_bootstrap: bool, bus: Rc<IggyMessageBus>, partitions: &IggyPartitions<Rc<IggyMessageBus>, PingPongSuperblock>, ) -> Result<Option<IggyPartition<Rc<IggyMessageBus>>>, ServerError> { + let created_revision = partition_metadata.created_revision; + match partition_retirement_fence(&config.get_system_path(), namespace).await { + Ok(Some(recorded)) if recorded > created_revision => return Ok(None), + Ok(Some(recorded)) if recorded == created_revision => { + warn!( + namespace_raw = namespace.inner(), + created_revision, "partition remains offline under its durable retirement fence" + ); + partitions.fence(namespace, created_revision); + return Ok(None); + } + Err(ref error @ ServerError::PartitionRetirementFenceIo { ref source, .. }) + if source.kind() == std::io::ErrorKind::InvalidData => + { + error!(%error, namespace_raw = namespace.inner(), "unreadable retirement fence; refusing partition"); + partitions.fence(namespace, created_revision); + return Ok(None); + } + Err(error) => return Err(error), + Ok(_) => {} + } + let initialization_dir = partition_initialization_path(&config.get_system_path(), namespace); + match partitions::read_created_revision(&initialization_dir).await { + Ok(Some(recorded)) if recorded > created_revision => return Ok(None), + Err(source) if source.kind() == std::io::ErrorKind::InvalidData => { + error!(%source, namespace_raw = namespace.inner(), "unreadable initialization record; refusing partition"); + partitions.fence(namespace, created_revision); + return Ok(None); + } + Err(source) => { + return Err(ServerError::PartitionCreatedRevisionIo { + dir: PathBuf::from(initialization_dir), + source, + }); + } + Ok(_) => {} + } let stream_id = namespace.stream_id(); let topic_id = namespace.topic_id(); let directory = config.get_partition_path(stream_id, topic_id, namespace.partition_id()); @@ -192,7 +231,6 @@ pub async fn load_partition_or_fence( dir: PathBuf::from(&directory), source, })?; - let created_revision = partition_metadata.created_revision; let revision_io = |source| ServerError::PartitionCreatedRevisionIo { dir: PathBuf::from(&directory), source, @@ -269,7 +307,7 @@ pub async fn load_partition_or_fence( "cannot tell which incarnation this partition directory belongs to; \ tombstoning the partition instead of serving it" ); - partitions.tombstone(namespace); + partitions.fence(namespace, created_revision); return Ok(None); } Err(source) => return Err(revision_io(source)), @@ -303,7 +341,6 @@ pub async fn load_partition_or_fence( cluster_id, self_replica_id, replica_count, - allow_bootstrap, Rc::clone(&bus), )) .await @@ -388,7 +425,7 @@ pub async fn load_partition_or_fence( segment files in place and tombstoning it instead of serving it \ empty" ); - partitions.tombstone(namespace); + partitions.fence(namespace, created_revision); return Ok(None); } if replica_count > 1 { @@ -437,7 +474,7 @@ pub async fn load_partition_or_fence( "failed to quarantine the refused segment files; leaving this \ partition tombstoned rather than rebuilding over them" ); - partitions.tombstone(namespace); + partitions.fence(namespace, created_revision); return Ok(None); } } @@ -451,7 +488,6 @@ pub async fn load_partition_or_fence( cluster_id, self_replica_id, replica_count, - false, partition_metadata.created_view, Rc::clone(&bus), )) @@ -495,7 +531,7 @@ pub async fn load_partition_or_fence( partition instead of serving it" ); partition_stats.zero_out_all(); - partitions.tombstone(namespace); + partitions.fence(namespace, created_revision); Ok(None) } Err(error) => Err(error), @@ -513,12 +549,14 @@ async fn load_partition( cluster_id: u128, self_replica_id: u8, replica_count: u8, - allow_bootstrap: bool, bus: Rc<IggyMessageBus>, ) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerError> { let stream_id = namespace.stream_id(); let topic_id = namespace.topic_id(); let partition_id = namespace.partition_id(); + let first_initialization = + partition_needs_initialization(config, namespace, partition_metadata.created_revision) + .await?; // (view, log_view) come from the group's durable superblock when present; // a present but unverifiable record already refused boot inside // `open_partition_superblock`. @@ -573,7 +611,7 @@ async fn load_partition( join, }, ); - consensus.set_recovery_election_allowed(allow_bootstrap); + consensus.set_recovery_election_allowed(first_initialization); // No prepare-timestamp floor is restored here: the partition consensus // journal is non-durable today, so there is no persisted head to observe @@ -608,6 +646,7 @@ async fn load_partition( let recovered_persistence = { let directory = wal_directory.clone(); if replica_count == 1 + && !first_initialization && !journal::PartitionPrepareJournal::has_published_frontier(&directory) .await .map_err(&wal_error)? @@ -617,7 +656,7 @@ async fn load_partition( "existing singleton partition is missing its prepare WAL frontier", ))); } - if replica_count > 1 && !allow_bootstrap { + if replica_count > 1 && !first_initialization { PartitionPersistence::fence_missing_history( &directory, partition_metadata.created_revision, @@ -714,6 +753,13 @@ async fn load_partition( .await .map_err(|error| ServerError::Iggy(Box::new(error)))?; } + if let Err(error) = + record_partition_initialization(config, namespace, partition_metadata.created_revision) + .await + { + partition.stats.zero_out_all(); + return Err(error); + } Ok(partition) } @@ -877,21 +923,44 @@ pub async fn build_partition_fresh( cluster_id: u128, self_replica_id: u8, replica_count: u8, - allow_bootstrap: bool, created_view: u32, bus: Rc<IggyMessageBus>, ) -> Result<IggyPartition<Rc<IggyMessageBus>>, ServerError> { + if partition_retirement_fence(&config.get_system_path(), namespace) + .await? + .is_some_and(|recorded| recorded >= created_revision) + { + return Err(ServerError::PartitionRetirementFenced { + namespace_raw: namespace.inner(), + created_revision, + }); + } let stream_id = namespace.stream_id(); let topic_id = namespace.topic_id(); let partition_id = namespace.partition_id(); + let first_initialization = + partition_needs_initialization(config, namespace, created_revision).await?; - // Sampled BEFORE the hierarchy create: a pre-existing partition directory - // is the marker of a prior life (the .log inside may legitimately be - // empty -- committed-but-unflushed data dies with the journal), while a - // genuinely fresh create finds nothing. let partition_dir = config.get_partition_path(stream_id, topic_id, partition_id); - let restarted = - replica_count > 1 && (!allow_bootstrap || std::fs::metadata(&partition_dir).is_ok()); + let restarted = replica_count > 1 && !first_initialization; + if replica_count == 1 && !first_initialization { + let directory = Path::new(&partition_dir).join(format!("prepares-{created_revision}")); + if !journal::PartitionPrepareJournal::has_published_frontier(&directory) + .await + .map_err(|source| ServerError::PartitionPrepareWalIo { + dir: directory.clone(), + source, + })? + { + return Err(ServerError::PartitionPrepareWalIo { + dir: directory, + source: std::io::Error::new( + std::io::ErrorKind::InvalidData, + "initialized singleton partition is missing its prepare WAL frontier", + ), + }); + } + } create_partition_file_hierarchy(stream_id, topic_id, partition_id, partitions_config) .await .map_err(|source| { @@ -928,18 +997,9 @@ pub async fn build_partition_fresh( ) .await?; - // A partition directory that already exists here is a rebuild over a - // fenced chain (`load_partition_or_fence` quarantined the refused segment - // files and kept the superblock; every other prior life is hydrated by - // that loader before reaching this builder), not a fresh create: this - // replica's group state died with the process, so claiming view-0 - // primaryship would heartbeat - // commit_min=0 at peers that hold the committed log (racing their - // election). Join as a quorum-invisible backup and probe for the - // current view instead; journal repair re-materializes the data from a - // peer, byte-identical by the deterministic-roll/replicated-ciphertext - // design. A truly fresh create keeps the plain init: every group needs - // its view-0 primary to exist. + // The external initialization record, including after directory loss, + // distinguishes a prior participant from a fresh group member. Prior + // participants probe before counting toward quorums. let durable_view = recovered_state .as_ref() .map(|state| (state.view, state.log_view)); @@ -990,7 +1050,7 @@ pub async fn build_partition_fresh( let mut partition = IggyPartition::new(stats, consensus); partition .consensus() - .set_recovery_election_allowed(allow_bootstrap); + .set_recovery_election_allowed(first_initialization); partition.set_runtime_options(runtime_options); partition.set_superblock(superblock, recovered_state.as_ref()); // Surface the evicted-ring ceilings from config onto the fresh journal. @@ -1092,9 +1152,109 @@ pub async fn build_partition_fresh( } open_partition_persistence(&mut partition, config, None).await?; + if let Err(error) = record_partition_initialization(config, namespace, created_revision).await { + partition.stats.zero_out_all(); + return Err(error); + } Ok(partition) } +fn partition_initialization_path(system_path: &str, namespace: IggyNamespace) -> String { + format!( + "{system_path}/{PARTITION_INITIALIZATION_DIR}/{}", + namespace.inner() + ) +} + +pub async fn partition_retirement_fence( + system_path: &str, + namespace: IggyNamespace, +) -> Result<Option<u64>, ServerError> { + let directory = partition_initialization_path(system_path, namespace); + partitions::read_revision_record(&directory, RETIREMENT_FENCE_FILE) + .await + .map_err(|source| ServerError::PartitionRetirementFenceIo { + dir: PathBuf::from(directory), + source, + }) +} + +/// A failed incarnation may stop retaining session receipts only once it can +/// no longer rejoin with those receipts missing after a restart. +pub async fn record_partition_retirement_fence( + system_path: &str, + namespace: IggyNamespace, + created_revision: u64, +) -> Result<(), ServerError> { + if partition_retirement_fence(system_path, namespace) + .await? + .is_some_and(|recorded| recorded >= created_revision) + { + let directory = partition_initialization_path(system_path, namespace); + return persist_partition_hierarchy(&directory, system_path).await; + } + let directory = partition_initialization_path(system_path, namespace); + create_dir_all(&directory) + .await + .map_err(|source| ServerError::PartitionRetirementFenceIo { + dir: PathBuf::from(&directory), + source, + })?; + persist_partition_hierarchy(&directory, system_path).await?; + partitions::write_revision_record(&directory, RETIREMENT_FENCE_FILE, created_revision) + .await + .map_err(|source| ServerError::PartitionRetirementFenceIo { + dir: PathBuf::from(directory), + source, + }) +} + +async fn partition_needs_initialization( + config: &ServerConfig, + namespace: IggyNamespace, + created_revision: u64, +) -> Result<bool, ServerError> { + let directory = partition_initialization_path(&config.get_system_path(), namespace); + let recorded = partitions::read_created_revision(&directory) + .await + .map_err(|source| ServerError::PartitionCreatedRevisionIo { + dir: PathBuf::from(&directory), + source, + })?; + if recorded.is_some_and(|revision| revision > created_revision) { + return Err(ServerError::PartitionCreatedRevisionIo { + dir: PathBuf::from(directory), + source: std::io::Error::new( + std::io::ErrorKind::InvalidData, + "partition initialization belongs to a newer metadata incarnation", + ), + }); + } + Ok(recorded != Some(created_revision)) +} + +/// Publish outside the partition directory before this incarnation can serve. +/// An absent directory can then be distinguished from lost initialized history. +async fn record_partition_initialization( + config: &ServerConfig, + namespace: IggyNamespace, + created_revision: u64, +) -> Result<(), ServerError> { + if !partition_needs_initialization(config, namespace, created_revision).await? { + let directory = partition_initialization_path(&config.get_system_path(), namespace); + return persist_partition_hierarchy(&directory, &config.get_system_path()).await; + } + let directory = partition_initialization_path(&config.get_system_path(), namespace); + create_dir_all(&directory) + .await + .map_err(|source| ServerError::PartitionCreatedRevisionIo { + dir: PathBuf::from(&directory), + source, + })?; + persist_partition_hierarchy(&directory, &config.get_system_path()).await?; + record_created_revision(&directory, created_revision).await +} + /// Record which incarnation `partition_dir` belongs to; see /// [`partitions::CREATED_REVISION_FILE`]. async fn record_created_revision( @@ -1181,6 +1341,132 @@ mod tests { const OFFSETS_VERSION: u8 = 3; const RETRY_CHECKPOINT_MAGIC: &[u8; 4] = b"IRP2"; + #[compio::test] + async fn given_uninitialized_singleton_when_booting_should_finish_initialization() { + assert_uninitialized_partition_recovers(1).await; + } + + #[compio::test] + async fn given_uninitialized_replica_when_booting_should_finish_initialization() { + assert_uninitialized_partition_recovers(REPLICAS).await; + } + + async fn assert_uninitialized_partition_recovers(replica_count: u8) { + const REVISION: u64 = 7; + let namespace = IggyNamespace::new(1, 1, 0); + for interrupted in [false, true] { + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + let partitions = solo_partitions(&config); + let directory = config.get_partition_path(1, 1, 0); + if interrupted { + create_partition_file_hierarchy(1, 1, 0, partitions.config()) + .await + .unwrap(); + record_created_revision(&directory, REVISION).await.unwrap(); + open_partition_superblock( + &directory, + ReplicaIdentity { + cluster: CLUSTER, + replica_id: 0, + replica_count, + }, + 0, + ) + .await + .unwrap(); + } + let recovered = load_partition_or_fence( + &config, + namespace, + Arc::new(PartitionStats::default()), + &Partition::new(0, namespace.inner(), IggyTimestamp::now(), REVISION, 0), + TopicRuntimeOptions::default(), + CLUSTER, + 0, + replica_count, + Rc::new(IggyMessageBus::new(0)), + &partitions, + ) + .await + .unwrap() + .expect("a partition that never initialized must finish its committed creation"); + assert!( + !recovered.requires_state_transfer(), + "first initialization has no missing history to transfer; interrupted={interrupted}" + ); + assert!( + journal::PartitionPrepareJournal::has_published_frontier( + &Path::new(&directory).join(format!("prepares-{REVISION}")) + ) + .await + .unwrap() + ); + } + } + + #[compio::test] + async fn given_initialized_partition_when_directory_lost_should_preserve_recovery_fence() { + const REVISION: u64 = 7; + let namespace = IggyNamespace::new(1, 1, 0); + for replica_count in [1, REPLICAS] { + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + let partitions = solo_partitions(&config); + let initialized = build_partition_fresh( + &config, + partitions.config(), + namespace, + Arc::new(PartitionStats::default()), + REVISION, + TopicRuntimeOptions::default(), + CLUSTER, + 0, + replica_count, + 0, + Rc::new(IggyMessageBus::new(0)), + ) + .await + .unwrap(); + drop(initialized); + std::fs::remove_dir_all(config.get_partition_path(1, 1, 0)).unwrap(); + let recovered = load_partition_or_fence( + &config, + namespace, + Arc::new(PartitionStats::default()), + &Partition::new(0, namespace.inner(), IggyTimestamp::now(), REVISION, 0), + TopicRuntimeOptions::default(), + CLUSTER, + 0, + replica_count, + Rc::new(IggyMessageBus::new(0)), + &partitions, + ) + .await + .unwrap(); + if replica_count == 1 { + assert!( + recovered.is_none(), + "a singleton must not hide lost history" + ); + } else { + assert!(recovered.unwrap().requires_state_transfer()); + } + } + } + + #[compio::test] + async fn given_initialized_singleton_when_rebuilding_lost_directory_should_refuse() { + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + drop(build_solo_partition(&config).await.unwrap()); + std::fs::remove_dir_all(config.get_partition_path(1, 1, 0)).unwrap(); + assert!( + build_solo_partition(&config).await.is_err(), + "reconciliation must not bypass the initialized partition's missing-history fence" + ); + } + #[compio::test] async fn given_uncertified_view_when_loading_or_rebuilding_should_persist_creation_floor() { const CREATED_VIEW: u32 = 2; @@ -1205,7 +1491,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - true, CREATED_VIEW, Rc::new(IggyMessageBus::new(0)), ) @@ -1239,7 +1524,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - false, Rc::new(IggyMessageBus::new(0)), ) .await @@ -1254,7 +1538,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - true, CREATED_VIEW, Rc::new(IggyMessageBus::new(0)), ) @@ -1308,7 +1591,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - false, Rc::new(IggyMessageBus::new(0)), &partitions, ) @@ -1336,7 +1618,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - true, CREATED_VIEW, Rc::new(IggyMessageBus::new(0)), ) @@ -1378,7 +1659,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - false, Rc::new(IggyMessageBus::new(0)), ) .await; @@ -1399,7 +1679,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - false, Rc::new(IggyMessageBus::new(0)), &partitions, ) @@ -1497,6 +1776,75 @@ mod tests { ); } + #[compio::test] + async fn given_retirement_fence_when_restarting_should_require_a_new_incarnation() { + const REVISION: u64 = 7; + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + let namespace = IggyNamespace::new(1, 1, 0); + drop(build_incarnation(&config, REVISION, 0).await); + record_partition_retirement_fence(&config.get_system_path(), namespace, REVISION) + .await + .unwrap(); + for directory_lost in [false, true] { + if directory_lost { + std::fs::remove_dir_all(config.get_partition_path(1, 1, 0)).unwrap(); + } + let partitions = solo_partitions(&config); + assert!( + load_incarnation(&config, &partitions, REVISION, 0) + .await + .is_none() + ); + assert_eq!(partitions.failed_revision(&namespace), Some(REVISION)); + assert!(matches!( + build_partition_fresh( + &config, + &solo_partitions_config(&config), + namespace, + Arc::new(PartitionStats::default()), + REVISION, + TopicRuntimeOptions::default(), + CLUSTER, + REPLICA, + REPLICAS, + 0, + Rc::new(IggyMessageBus::new(0)), + ) + .await, + Err(ServerError::PartitionRetirementFenced { .. }) + )); + } + drop(build_incarnation(&config, REVISION + 1, 0).await); + assert!( + load_incarnation(&config, &solo_partitions(&config), REVISION + 1, 0) + .await + .is_some() + ); + } + + #[compio::test] + async fn given_corrupt_retirement_fence_when_restarting_should_refuse_partition() { + const REVISION: u64 = 7; + let root = tempfile::tempdir().unwrap(); + let config = solo_config(&root); + let namespace = IggyNamespace::new(1, 1, 0); + drop(build_incarnation(&config, REVISION, 0).await); + let directory = partition_initialization_path(&config.get_system_path(), namespace); + std::fs::write(Path::new(&directory).join(RETIREMENT_FENCE_FILE), b"torn").unwrap(); + let partitions = solo_partitions(&config); + assert!( + load_incarnation(&config, &partitions, REVISION, 0) + .await + .is_none() + ); + assert!( + record_partition_retirement_fence(&config.get_system_path(), namespace, REVISION) + .await + .is_err() + ); + } + #[compio::test] async fn given_no_recorded_revision_when_loading_should_record_the_committed_one() { const CREATED_REVISION: u64 = 4; @@ -1612,7 +1960,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - true, CREATED_VIEW, Rc::new(IggyMessageBus::new(0)), ) @@ -1677,7 +2024,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - false, Rc::new(IggyMessageBus::new(0)), ) .await @@ -1756,7 +2102,6 @@ mod tests { CLUSTER, 0, 1, - false, Rc::new(IggyMessageBus::new(0)), &partitions, ) @@ -1808,7 +2153,6 @@ mod tests { CLUSTER, 0, 1, - false, Rc::new(IggyMessageBus::new(0)), &partitions, ) @@ -1854,7 +2198,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - true, 0, Rc::new(IggyMessageBus::new(0)), ) @@ -1933,7 +2276,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - false, Rc::new(IggyMessageBus::new(0)), ) .await @@ -1977,7 +2319,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - true, 0, Rc::new(IggyMessageBus::new(0)), ) @@ -2013,7 +2354,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - false, Rc::new(IggyMessageBus::new(0)), &partitions, ) @@ -2102,7 +2442,6 @@ mod tests { CLUSTER, 0, 1, - true, 0, Rc::new(IggyMessageBus::new(0)), ) @@ -2147,7 +2486,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - true, created_view, Rc::new(IggyMessageBus::new(0)), ) @@ -2194,7 +2532,6 @@ mod tests { CLUSTER, REPLICA, REPLICAS, - false, Rc::new(IggyMessageBus::new(0)), partitions, ) @@ -2343,7 +2680,6 @@ mod tests { CLUSTER, 0, 1, - false, Rc::new(IggyMessageBus::new(0)), &partitions, ) diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index 389310409..c191e15e6 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -659,11 +659,15 @@ async fn reconcile_additions( // delete that failed leaves the tombstone standing with no // `ConfirmRemove` behind it -- the same permanent fence the in-map // branch above escapes, told apart by the same signal. - if ctx.has_pending_delete_failure(ns) { + if ctx.has_pending_delete_failure(ns) + || partitions + .failed_revision(&ns) + .is_some_and(|failed| failed < epoch) + { trace!( shard = shard_id, ns_raw = ns.inner(), - "additions: ns tombstoned before materialisation with a failed disk delete; re-driving teardown" + "additions: removing fenced files of a superseded or incompletely deleted incarnation" ); tear_down_owned_partition(ctx, ns, counters).await; continue; @@ -777,13 +781,6 @@ async fn reconcile_additions( ctx.cluster_id, ctx.self_replica_id, ctx.replica_count, - created_revision - > ctx - .shard - .plane - .metadata() - .applied_frontier() - .recovered_revision(), Rc::clone(&ctx.shard.bus), partitions, ) @@ -799,13 +796,6 @@ async fn reconcile_additions( ctx.cluster_id, ctx.self_replica_id, ctx.replica_count, - created_revision - > ctx - .shard - .plane - .metadata() - .applied_frontier() - .recovered_revision(), created_view, Rc::clone(&ctx.shard.bus), ) @@ -2328,7 +2318,6 @@ mod tests { CLUSTER_ID, 0, 1, - true, created_view(&ctx, ns), Rc::clone(&ctx.shard.bus), ) @@ -2360,7 +2349,6 @@ mod tests { CLUSTER_ID, 0, 1, - true, created_view(&ctx, ns), Rc::clone(&ctx.shard.bus), ) @@ -3362,6 +3350,15 @@ mod tests { let partition_root = ctx.config.get_partition_path(0, 0, 0); assert!(std::path::Path::new(&partition_root).exists()); + seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0); + seed_topic( + &shard.plane.metadata().mux_stm, + 4, + 0, + "topic-wedge", + vec![assignment(0, 1)], + ); + // Reconstruct the post-failed-teardown state: tombstone set + // shards_table row gone + a `FailureCause::Delete` record, but the // partition still in the map and its directory still on disk (the @@ -3568,6 +3565,47 @@ mod tests { assert!(!partitions.is_tombstoned(&ns)); } + #[compio::test] + async fn given_retirement_fence_when_topic_recreated_between_passes_should_rebuild() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let mux = TestMux::default(); + seed_stream(&mux, 1, "stream"); + seed_topic(&mux, 2, 0, "old", vec![assignment(0, 1)]); + let shard = build_test_shard(0, &config, mux); + let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config)); + let namespace = IggyNamespace::new(0, 0, 0); + let old_revision = shard + .plane + .metadata() + .mux_stm + .streams() + .read(|inner| inner.revision); + shard.plane.partitions().fence(namespace, old_revision); + crate::partition_helpers::record_partition_retirement_fence( + &ctx.config.get_system_path(), + namespace, + old_revision, + ) + .await + .unwrap(); + seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0); + seed_topic( + &shard.plane.metadata().mux_stm, + 4, + 0, + "new", + vec![assignment(0, 1)], + ); + reconcile_pass(&ctx).await; + reconcile_pass(&ctx).await; + assert!(!shard.plane.partitions().is_tombstoned(&namespace)); + assert!( + shard.plane.partitions().contains(&namespace), + "a fence for the deleted incarnation must not strand its replacement" + ); + } + /// The sibling guard: while the namespace is STILL in the committed /// target, the fenced-ghost sweep must not touch it -- only an operator /// delete authorises destroying the bytes the fence guards. (The @@ -3675,6 +3713,15 @@ mod tests { let partitions = shard.plane.partitions(); assert!(partitions.contains(&ns)); + seed_delete_topic(&shard.plane.metadata().mux_stm, 3, 0, 0); + seed_topic( + &shard.plane.metadata().mux_stm, + 4, + 0, + "topic-defer-skip", + vec![assignment(0, 1)], + ); + // Post-successful-teardown, pre-drain state: fenced, unlinked, and a // `ConfirmRemove` queued but not yet applied. `ns` is still in the // committed target, so the additions pass can only defer the rebuild. diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs index 519e58e6b..e19c1532b 100644 --- a/core/server/src/server_error.rs +++ b/core/server/src/server_error.rs @@ -37,8 +37,6 @@ pub enum ServerError { #[source] source: std::io::Error, }, - #[error("cannot fingerprint the server artifact: {0}")] - BinaryIdentity(#[source] std::io::Error), #[error(transparent)] Iggy(Box<iggy_common::IggyError>), #[error("failed to load server config")] @@ -188,6 +186,19 @@ pub enum ServerError { #[source] source: std::io::Error, }, + #[error("failed to read or write partition retirement fence at {dir}: {source}")] + PartitionRetirementFenceIo { + dir: PathBuf, + #[source] + source: std::io::Error, + }, + #[error( + "partition {namespace_raw} incarnation {created_revision} is durably fenced for session retirement" + )] + PartitionRetirementFenced { + namespace_raw: u64, + created_revision: u64, + }, // Quarantines the one partition rather than treating the group as fresh or // reading through to a superseded view: mirrors the metadata plane's // `RecoveryError::SuperblockUnreadable` policy, minus the boot refusal, diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 0342f2f70..77952c7e4 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -388,6 +388,11 @@ pub enum PartitionRead { pub enum PartitionReadReply { Primary(u8), SessionRetired(bool), + /// Permanent failure of this incarnation. It is not a retirement proof + /// until the caller durably fences the incarnation against future serving. + SessionRetirementFailed { + created_revision: u64, + }, Poll { fragments: PollFragments, current_offset: u64, @@ -2108,7 +2113,12 @@ where namespace: IggyNamespace, read: PartitionRead, ) -> Option<PartitionReadReply> { - let Some(target) = self.shards_table.shard_for(namespace) else { + let target = self.shards_table.shard_for(namespace).or_else(|| { + matches!(read, PartitionRead::SessionRetired { .. }).then(|| { + crate::shards_table::calculate_shard_assignment(&namespace, self.shard_count) + }) + }); + let Some(target) = target else { tracing::warn!( shard = self.id, namespace_raw = namespace.inner(), diff --git a/core/shard/src/poll.rs b/core/shard/src/poll.rs index a19d88b08..316697292 100644 --- a/core/shard/src/poll.rs +++ b/core/shard/src/poll.rs @@ -143,6 +143,20 @@ where reply: Sender<PartitionReadReply>, ) { let partitions = self.plane.partitions(); + if matches!(read, PartitionRead::SessionRetired { .. }) { + let failed_revision = partitions.failed_revision(&namespace).or_else(|| { + partitions + .with_partition(&namespace, |partition| { + partition.fatal().map(|_| partition.created_revision()) + }) + .flatten() + }); + if let Some(created_revision) = failed_revision { + let _ = reply + .try_send(PartitionReadReply::SessionRetirementFailed { created_revision }); + return; + } + } let rejected = partitions .with_partition(&namespace, |partition| { if partition.requires_state_transfer() {
