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 882c61cb8e4d2d9423033a92eecbd7335e9ad391 Author: Hubert Gruszecki <[email protected]> AuthorDate: Sat Oct 3 05:26:50 2026 +0200 fix(server): restore startup and buffered recovery Empty storage and Replicated writes were rejected despite existing startup and SDK contracts. Restore automatic metadata initialization and scope crash-safe retries to durable writes. Complete metadata directory loss still requires verified restore. Orderly shutdown must publish buffered WAL writes, and recovery must checkpoint them before advancing reserved offsets. Retain repair bodies across the gap and establish segment links before sends so descriptor exhaustion cannot block the shutdown barrier. Align replica, session and simulator fixtures, standalone formatting and isolated Miri with the supported contracts. --- .github/actions/rust/pre-merge/action.yml | 6 +- core/common/src/types/streaming_stats.rs | 4 +- core/consensus/src/impls.rs | 24 +-- core/consensus/src/metadata_helpers.rs | 22 ++- .../tests/cluster/crash_recovery_corruption.rs | 30 ++- .../tests/server/descriptor_exhaustion_vsr.rs | 62 +++--- core/integration/tests/server/http_vsr.rs | 46 ++++- ...er_group_new_messages_after_restart_scenario.rs | 1 + core/journal/src/partition_journal.rs | 70 +++++++ core/journal/src/partition_journal/segments.rs | 58 ++++++ core/message_bus/src/connection_cap.rs | 2 +- core/message_bus/tests/head_of_line.rs | 47 ++--- core/partitions/src/iggy_partition.rs | 77 +++++--- core/partitions/src/partition_storage.rs | 10 +- core/partitions/src/persistence.rs | 31 +++ core/partitions/tests/consumer_offset_history.rs | 10 +- core/sdk/src/clients/producer.rs | 12 +- core/sdk/src/clients/producer_builder.rs | 4 +- core/server/README.md | 27 ++- core/server/src/boot/mod.rs | 49 +++-- core/server/src/boot/recovery.rs | 4 +- core/server/src/boot/threads.rs | 2 - core/server/src/dispatch/host.rs | 9 +- core/server/src/dispatch/mod.rs | 60 ++++++ core/server/src/dispatch/reads.rs | 59 +----- core/server/src/dispatch/session_ops.rs | 33 ++-- core/server/src/main.rs | 2 +- core/server/src/partition_helpers.rs | 54 ++---- core/server_common/src/poll.rs | 2 +- core/simulator/src/lib.rs | 8 +- core/simulator/src/storage/tests.rs | 208 +++++++++++---------- .../Fixtures/IggyServerFixture.cs | 4 +- .../Iggy_SDK.Tests.Integration/HeartbeatTests.cs | 5 +- foreign/csharp/README.md | 7 +- foreign/python/src/client.rs | 5 +- foreign/swift/Tools/golden-vectors/src/main.rs | 10 +- justfile | 6 +- 37 files changed, 673 insertions(+), 397 deletions(-) diff --git a/.github/actions/rust/pre-merge/action.yml b/.github/actions/rust/pre-merge/action.yml index 41b211300..dc87fca6f 100644 --- a/.github/actions/rust/pre-merge/action.yml +++ b/.github/actions/rust/pre-merge/action.yml @@ -417,8 +417,8 @@ runs: - name: Install nightly Rust + miri component if: inputs.task == 'miri' run: | - rustup toolchain install nightly-2026-04-21 --component miri --profile minimal --no-self-update - cargo +nightly-2026-04-21 miri setup + rustup toolchain install nightly-2026-10-02 --component miri --profile minimal --no-self-update + cargo +nightly-2026-10-02 miri setup shell: bash - name: Run miri @@ -426,7 +426,7 @@ runs: env: MIRIFLAGS: "-Zmiri-tree-borrows -Zmiri-strict-provenance" run: | - cargo +nightly-2026-04-21 miri test --locked \ + cargo +nightly-2026-10-02 miri test --locked \ -p iggy_binary_protocol \ -p consensus shell: bash diff --git a/core/common/src/types/streaming_stats.rs b/core/common/src/types/streaming_stats.rs index 5d64c632c..a98e87a05 100644 --- a/core/common/src/types/streaming_stats.rs +++ b/core/common/src/types/streaming_stats.rs @@ -110,7 +110,7 @@ static PARTITION_SEGMENTS_COUNT: UnderflowSite = UnderflowSite::new("partition", /// Subtract `amount`, clamping at zero instead of wrapping. Returns what was /// actually taken, which is what the caller passes on to its parent. /// -/// `fetch_update` rather than a load followed by a subtract: the counters are +/// `try_update` rather than a load followed by a subtract: the counters are /// written from the metadata shard and from whichever shard owns the partition, /// so a separate load leaves a window where the clamp reads one value and /// subtracts from another. @@ -118,7 +118,7 @@ macro_rules! clamped_sub { ($name:ident, $counter:ty, $amount:ty) => { fn $name(counter: &$counter, amount: $amount, site: &'static UnderflowSite) -> $amount { let previous = counter - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| { + .try_update(Ordering::AcqRel, Ordering::Acquire, |current| { // `None` on an already-zero counter: no store, and the // `Err` it returns carries that same zero, so the clamp // reports identically either way. That is the steady shape diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 0d220ba3b..509c8f8a7 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -75,6 +75,18 @@ impl ConsensusClock { } } +#[cfg(test)] +pub struct FixedClock(pub u64); + +#[cfg(test)] +impl Clock for FixedClock { + type Realtime = IggyTimestamp; + + fn realtime(&self) -> Self::Realtime { + IggyTimestamp::from(self.0) + } +} + impl std::fmt::Debug for ConsensusClock { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.write_str("ConsensusClock") @@ -4918,18 +4930,6 @@ mod timestamp_clamp_tests { use super::*; use crate::LocalPipeline; - /// Clock frozen at a fixed instant, standing in for a lagging wall - /// clock on a freshly elected primary. - struct FixedClock(u64); - - impl clock::Clock for FixedClock { - type Realtime = IggyTimestamp; - - fn realtime(&self) -> Self::Realtime { - IggyTimestamp::from(self.0) - } - } - use crate::test_bus::{NoopBus, make_start_view}; #[test] diff --git a/core/consensus/src/metadata_helpers.rs b/core/consensus/src/metadata_helpers.rs index 094da7d29..b0699de6c 100644 --- a/core/consensus/src/metadata_helpers.rs +++ b/core/consensus/src/metadata_helpers.rs @@ -445,8 +445,11 @@ pub async fn send_eviction_to_client<B, P>( #[cfg(test)] mod tests { + use std::rc::Rc; + use super::*; use crate::client_table::{REGISTER_REQUEST_ID, REPLY_RING_RETENTION_BYTES}; + use crate::impls::{ConsensusClock, FixedClock}; use crate::{CLIENTS_TABLE_MAX, LocalPipeline, RequestEntry}; use iggy_binary_protocol::{ Command, Operation, PrepareHeader, ReplyHeader, RoutedRequestHeader, @@ -585,7 +588,16 @@ mod tests { const SESSION: u64 = 10; const REQUEST: u64 = 1; const REQUEST_CHECKSUM: u128 = 0xAA; - let consensus = VsrConsensus::new(1, 0, 3, 0, ClientSpyBus::new(), LocalPipeline::new()); + const CLOCK_MICROS: u64 = 1_000; + let consensus = VsrConsensus::with_clock( + 1, + 0, + 3, + 0, + ClientSpyBus::new(), + LocalPipeline::new(), + ConsensusClock::new(Rc::new(FixedClock(CLOCK_MICROS))), + ); consensus.init(); let client_table = fresh_client_table(); let mut message = Message::<RoutedRequestHeader>::new(HEADER_SIZE); @@ -643,13 +655,9 @@ mod tests { )); } assert_eq!(consensus.request_queue_len(), 1); + let queued_request = consensus.pop_queued_request().unwrap(); assert_eq!( - consensus - .pop_queued_request() - .unwrap() - .message - .header() - .request_checksum, + queued_request.message.header().request_checksum, REQUEST_CHECKSUM ); consensus.with_pipeline_mut(|pipeline| { diff --git a/core/integration/tests/cluster/crash_recovery_corruption.rs b/core/integration/tests/cluster/crash_recovery_corruption.rs index 24515bd21..f608bf0bb 100644 --- a/core/integration/tests/cluster/crash_recovery_corruption.rs +++ b/core/integration/tests/cluster/crash_recovery_corruption.rs @@ -78,12 +78,9 @@ const MIN_INDEX_ENTRIES: usize = 4; /// Infix of the directory the refusal path renames a partition's segment files /// into (`partitions::state_transfer::quarantine_partition_files`). const FENCED_DIR_MARKER: &str = ".fenced."; -/// Boot log line recovery emits when the log cannot back the last entry of an -/// index (`partitions::segment_recovery::recover_segment_bounds`): the positive -/// evidence that path ran, as opposed to the clean anchored walk or a refusal. -/// Distinct from the line the self-contradicting-index check emits, which ends -/// "rebuilding it from the log". -const INDEX_REBUILD_MARKER: &str = "discarding the index and rebuilding it from a byte-0 walk"; +/// WAL-owned recovery discards unpublished segment bytes before rebuilding the index. +const UNPUBLISHED_TAIL_MARKER: &str = + "discarding unpublished segment bytes beyond the durable WAL frontier"; const PARTITION_WAL_REFUSAL_MARKER: &str = "prepare WAL at"; async fn create_stream_and_topic(client: &IggyClient, durability: Durability) { @@ -677,12 +674,8 @@ async fn given_an_index_ahead_of_a_truncated_log_when_a_node_recovers_should_pre ) }); - // Read the index BEFORE the log, both right after boot: catch-up grows - // the two files together, so a log still at the cut proves the index was - // read before any append landed, and the rebuild is then the only shape it - // may have. Once the log has grown the refilled tail re-mints entries over - // the same bytes, and nothing on disk tells the two apart; the boot log - // marker below is the evidence that survives that. + // WAL recovery discards unpublished bytes; peer repair can immediately refill them. + // Read the index first so a log still at the cut proves it cannot describe later bytes. // // The rebuild's stride is its own, so the entry COUNT is not the spec. // What is: the index describes only bytes the walk proved, which is the @@ -716,9 +709,8 @@ async fn given_an_index_ahead_of_a_truncated_log_when_a_node_recovers_should_pre let fenced = fenced_segment_paths(&backup_data); assert!( fenced.is_empty(), - "recovery must rebuild the index from the log, keeping the {} \ - surviving batches in service; instead the chain was refused and fenced aside: {fenced:?}", - surviving.len() + "the unpublished tail must be discarded for peer repair, \ + without quarantining the segment chain: {fenced:?}" ); // `fenced_segment_paths` walks past unreadable directories, so an empty // result alone could be vacuous: the files must still be where boot found @@ -740,9 +732,11 @@ async fn given_an_index_ahead_of_a_truncated_log_when_a_node_recovers_should_pre "boot must absorb an index that outruns its log, not refuse the chain" ); assert!( - harness.node(backup).stdout_contains(INDEX_REBUILD_MARKER), - "boot must log the byte-0 rebuild ({INDEX_REBUILD_MARKER:?}); recovery took \ - another path" + harness + .node(backup) + .stdout_contains(UNPUBLISHED_TAIL_MARKER), + "boot must discard the unpublished tail using the durable WAL frontier \ + ({UNPUBLISHED_TAIL_MARKER:?}); recovery took another path" ); } } diff --git a/core/integration/tests/server/descriptor_exhaustion_vsr.rs b/core/integration/tests/server/descriptor_exhaustion_vsr.rs index a86abf5f4..a6c0d49a5 100644 --- a/core/integration/tests/server/descriptor_exhaustion_vsr.rs +++ b/core/integration/tests/server/descriptor_exhaustion_vsr.rs @@ -17,19 +17,20 @@ //! A partition write that finds no free file descriptor stops the server with //! exit status 4, after the shutdown flush. The tests run the server under a -//! small `RLIMIT_NOFILE`, keep a few messages in memory only, fill the +//! small `RLIMIT_NOFILE`, leave a few messages outside the durable WAL frontier, fill the //! descriptor table with idle HTTP sockets, and write consumer offsets until //! an offset file cannot be opened. A restart under the normal limit then -//! serves the messages that only the flush could have written. +//! serves the messages that only the flush could have published durably. use std::net::SocketAddr; -use std::path::{Path, PathBuf}; +use std::path::PathBuf; use std::time::Duration; use bytes::Bytes; use iggy::prelude::*; use integration::harness::{ServerHandle, TestBinary, TestHarness}; use integration::iggy_harness; +use journal::partition_journal::{FRONTIER_FILE_NAME, PARTITION_WAL_BLOCK_SIZE}; use tokio::net::TcpStream; use tokio::time::{sleep, timeout}; @@ -40,7 +41,7 @@ const PARTITION_ID: u32 = 0; /// the rest quickly. More sockets than this cannot fit, so opening this many /// fills the table whatever the boot used. const OPEN_FILES_LIMIT: u64 = 128; -/// Far below `messages_required_to_save`, so only a flush writes them. +/// Far below `messages_required_to_save`, so their acknowledgement does not flush. const PAYLOADS: [&str; 3] = ["first", "second", "third"]; const CONNECT_TIMEOUT: Duration = Duration::from_secs(2); /// Each attempt uses a new consumer, so each one opens a new offset file. @@ -50,6 +51,8 @@ const WRITE_TIMEOUT: Duration = Duration::from_secs(5); /// that frees up goes to a waiting socket between two writes. const WRITE_PAUSE: Duration = Duration::from_millis(500); const EXIT_TIMEOUT: Duration = Duration::from_secs(60); +/// The head field of a frontier slot in `journal::partition_journal::JournalState`. +const WAL_HEAD_OFFSET: usize = 72; /// How a test writes the offset of a new consumer. #[derive(Clone, Copy)] @@ -132,11 +135,22 @@ async fn stop_on_offset_write_without_descriptor(harness: &mut TestHarness, writ ) .await .expect("send messages"); - let segment = segment_log(harness.server(), stream.id, topic.id); - assert_eq!( - file_len(&segment), - 0, - "the messages must be in memory only before the fault" + let frontier = partition_frontier(harness.server(), stream.id, topic.id); + let before = std::fs::read(&frontier).expect("read the WAL frontier after sending"); + assert_eq!(before.len(), 2 * PARTITION_WAL_BLOCK_SIZE); + assert!( + before + .as_chunks::<PARTITION_WAL_BLOCK_SIZE>() + .0 + .iter() + .all(|slot| { + u64::from_le_bytes( + slot[WAL_HEAD_OFFSET..WAL_HEAD_OFFSET + size_of::<u64>()] + .try_into() + .expect("WAL head field"), + ) == 0 + }), + "Replicated acknowledgements must leave the messages outside the durable WAL frontier" ); let held = fill_descriptor_table(harness.server().http_addr().expect("HTTP listener")).await; @@ -187,8 +201,8 @@ async fn stop_on_offset_write_without_descriptor(harness: &mut TestHarness, writ "a write that ran out of descriptors must stop the server with exit status 4, got {status}" ); assert!( - file_len(&segment) > 0, - "the shutdown flush must write the messages to their segment before the exit" + std::fs::read(&frontier).expect("read the WAL frontier after shutdown") != before, + "the shutdown flush must publish the buffered messages before the exit" ); assert_first_exhaustion_opened_an_offset_file(harness.server()); @@ -258,14 +272,20 @@ fn assert_first_exhaustion_opened_an_offset_file(server: &ServerHandle) { ); } -fn segment_log(server: &ServerHandle, stream_id: u32, topic_id: u32) -> PathBuf { - server.data_path().join(format!( - "streams/{stream_id}/topics/{topic_id}/partitions/{PARTITION_ID}/00000000000000000000.log" - )) -} - -fn file_len(path: &Path) -> u64 { - std::fs::metadata(path) - .unwrap_or_else(|error| panic!("read {}: {error}", path.display())) - .len() +fn partition_frontier(server: &ServerHandle, stream_id: u32, topic_id: u32) -> PathBuf { + let partition = server.data_path().join(format!( + "streams/{stream_id}/topics/{topic_id}/partitions/{PARTITION_ID}" + )); + let directories: Vec<PathBuf> = std::fs::read_dir(&partition) + .expect("read the partition directory") + .map(|entry| entry.expect("read a partition entry").path()) + .filter(|path| { + path.is_dir() + && path + .file_name() + .is_some_and(|name| name.to_string_lossy().starts_with("prepares-")) + }) + .collect(); + assert_eq!(directories.len(), 1, "one current partition incarnation"); + directories[0].join(FRONTIER_FILE_NAME) } diff --git a/core/integration/tests/server/http_vsr.rs b/core/integration/tests/server/http_vsr.rs index 84b0ae746..5ba4ffd41 100644 --- a/core/integration/tests/server/http_vsr.rs +++ b/core/integration/tests/server/http_vsr.rs @@ -45,6 +45,7 @@ const PARTITION_ID: u32 = 0; const CONSUMER_ID: u32 = 1; const DURABILITY_HEADER: &str = "iggy-durability"; +const DURABILITY_REPLICATED: &str = "replicated"; const DURABILITY_PERSISTED: &str = "persisted"; const DURABILITY_NONE: &str = "none"; @@ -1108,7 +1109,8 @@ async fn given_independent_durability_policies_when_producing_should_attest_mess let policies = iggy_common::TopicRuntimeOptions::from_resource_options(&details.options); assert_eq!(policies.durability, message_policy); assert_eq!(policies.consumer_offset_durability, offset_policy); - for query in ["", "?ack=replicated", "?ack=none"] { + let queries = ["", "?ack=replicated", "?ack=none"]; + for query in queries { let message = IggyMessage::builder() .payload("policy".into()) .build() @@ -1125,14 +1127,44 @@ async fn given_independent_durability_policies_when_producing_should_attest_mess if query == "?ack=none" { assert_eq!(response.status(), StatusCode::ACCEPTED); assert_eq!(durability(&response), DURABILITY_NONE); - } else if message_policy.is_persisted() { - assert_eq!(response.status(), StatusCode::CREATED); - assert_eq!(durability(&response), DURABILITY_PERSISTED); } else { - assert_eq!(response.status(), StatusCode::BAD_REQUEST); - let error: serde_json::Value = response.json().await.unwrap(); - assert_eq!(error["id"], IggyError::DurabilityRequired.as_code()); + assert_eq!(response.status(), StatusCode::CREATED); + assert_eq!( + durability(&response), + if message_policy.is_persisted() { + DURABILITY_PERSISTED + } else { + DURABILITY_REPLICATED + } + ); + } + } + let deadline = Instant::now() + ASYNC_COMMIT_TIMEOUT; + loop { + if let Some(polled) = http + .try_poll( + "durability-http", + &topic, + PARTITION_ID, + 0, + queries.len() as u32, + ) + .await + && polled.messages.len() == queries.len() + { + assert!( + polled + .messages + .iter() + .all(|message| message.payload == "policy") + ); + break; } + assert!( + Instant::now() < deadline, + "writes with {message_policy}/{offset_policy} did not become pollable" + ); + sleep(ASYNC_COMMIT_RETRY_INTERVAL).await; } } } diff --git a/core/integration/tests/server/scenarios/consumer_group_new_messages_after_restart_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_new_messages_after_restart_scenario.rs index ad4c2012f..4bbceb890 100644 --- a/core/integration/tests/server/scenarios/consumer_group_new_messages_after_restart_scenario.rs +++ b/core/integration/tests/server/scenarios/consumer_group_new_messages_after_restart_scenario.rs @@ -124,6 +124,7 @@ async fn execute_scenario(harness: &TestHarness, client: &IggyClient) { ); // 6. Disconnect the consumer and client (simulating runtime restart) + consumer.shutdown().await.unwrap(); drop(consumer); runtime_client.disconnect().await.unwrap(); drop(runtime_client); diff --git a/core/journal/src/partition_journal.rs b/core/journal/src/partition_journal.rs index b2e5c5da7..5f0f54e8b 100644 --- a/core/journal/src/partition_journal.rs +++ b/core/journal/src/partition_journal.rs @@ -3143,6 +3143,76 @@ mod tests { } } + #[compio::test] + async fn given_checkpointed_history_when_reanchoring_a_reserved_gap_should_preserve_repair_bodies() + { + const BODY_BYTES: usize = 4096; + const RESERVED_OFFSET: u64 = 8; + let partition = tempdir().unwrap(); + let directory = partition.path().join("prepares-7"); + let mut journal = PartitionPrepareJournal::open(&directory, 42, 7) + .await + .unwrap(); + journal + .enable_segment_storage(SegmentPosition::default(), (4 * BODY_BYTES) as u64) + .await + .unwrap(); + let first = segment_prepare(1, 0, 0, BODY_BYTES); + journal.append(first.clone().into_frozen()).await.unwrap(); + let frontier = std::fs::read(directory.join(FRONTIER_FILE_NAME)).unwrap(); + assert!( + journal + .reanchor_segment_storage(RESERVED_OFFSET) + .await + .is_err() + ); + assert_eq!( + std::fs::read(directory.join(FRONTIER_FILE_NAME)).unwrap(), + frontier, + "uncheckpointed history must not be replaced by an empty append point" + ); + journal.checkpoint(1).await.unwrap(); + assert!(journal.reanchor_segment_storage(0).await.is_err()); + journal + .reanchor_segment_storage(RESERVED_OFFSET) + .await + .unwrap(); + drop(journal); + + let mut journal = PartitionPrepareJournal::open(&directory, 42, 7) + .await + .unwrap(); + assert_eq!(journal.head(), 1); + assert_eq!(journal.checkpoint_op(), 1); + assert_eq!( + journal.segment_checkpoint(), + Some(SegmentPosition { + start_offset: RESERVED_OFFSET, + length: 0, + next_offset: RESERVED_OFFSET, + }) + ); + assert_eq!( + journal.prepares().await.unwrap()[0].as_slice(), + first.as_slice() + ); + let second = segment_prepare(2, first.header().checksum, RESERVED_OFFSET, BODY_BYTES); + journal.append(second.clone().into_frozen()).await.unwrap(); + drop(journal); + + let journal = PartitionPrepareJournal::open(&directory, 42, 7) + .await + .unwrap(); + let recovered = journal.prepares().await.unwrap(); + assert_eq!(recovered.len(), 2); + assert_eq!(recovered[0].as_slice(), first.as_slice()); + assert_eq!(recovered[1].as_slice(), second.as_slice()); + assert_eq!( + std::fs::read(partition.path().join(format!("{RESERVED_OFFSET:020}.log"))).unwrap(), + second.as_slice()[size_of::<PrepareHeader>()..] + ); + } + #[compio::test] async fn owned_segments_rollback_physical_tails_without_advancing_the_checkpoint() { const BODY_BYTES: usize = 8192; diff --git a/core/journal/src/partition_journal/segments.rs b/core/journal/src/partition_journal/segments.rs index 84ca7e57b..42ddedb4f 100644 --- a/core/journal/src/partition_journal/segments.rs +++ b/core/journal/src/partition_journal/segments.rs @@ -143,6 +143,53 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { } } + /// Move the physical append point over a reserved offset gap. + /// The complete preceding chain must already be durably checkpointed. + /// + /// # Errors + /// Returns an error for an uncheckpointed tail, a regressing offset, or a failed barrier. + pub async fn reanchor_segment_storage(&mut self, next_offset: u64) -> io::Result<()> { + self.ensure_healthy()?; + let mut segments = self + .state + .segment_storage + .ok_or_else(|| invalid("segment storage is not enabled"))?; + if self.state.checkpoint != self.state.head || self.durable_head != self.state.head { + return Err(invalid("cannot reanchor an uncheckpointed segment tail")); + } + if next_offset < segments.tail.position.next_offset { + return Err(invalid( + "reserved segment offset regresses the physical tail", + )); + } + if next_offset == segments.tail.position.next_offset { + return Ok(()); + } + segments.reset_position(SegmentPosition { + start_offset: next_offset, + length: 0, + next_offset, + })?; + let state = JournalState { + segment_storage: Some(segments), + ..self.state + }; + self.poisoned = true; + self.open_segment_file( + segments.tail.generation, + next_offset, + 0, + self.preallocate_segments.then_some(segments.max_size), + ) + .await?; + self.sync_segment_files().await?; + self.publish(state).await?; + self.state = state; + self.retain_active_segment_file(); + self.poisoned = false; + Ok(()) + } + pub fn segment_reference(&self, header: &PrepareHeader) -> Option<SegmentReference> { if !self.contains(header) { return None; @@ -541,6 +588,14 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { )); } if length > cursor.position.length { + tracing::warn!( + group = self.state.group, + incarnation = self.state.incarnation, + start_offset = cursor.position.start_offset, + durable_bytes = cursor.position.length, + discarded_bytes = length - cursor.position.length, + "discarding unpublished segment bytes beyond the durable WAL frontier" + ); file.truncate(cursor.position.length).await?; if self.preallocate_segments { file.preallocate(&cursor.path(&self.directory), segments.max_size); @@ -666,6 +721,9 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> { self.segment_files_dirty = true; self.segment_links_dirty = true; self.segment_files.insert(key, file); + // Establish the inode and links before buffered appends can be acknowledged. + // A later shutdown barrier must not need a new directory descriptor. + self.sync_segment_files().await?; Ok(()) } diff --git a/core/message_bus/src/connection_cap.rs b/core/message_bus/src/connection_cap.rs index a912581c2..524b805a9 100644 --- a/core/message_bus/src/connection_cap.rs +++ b/core/message_bus/src/connection_cap.rs @@ -69,7 +69,7 @@ impl ConnectionCap { pub fn try_acquire(&self) -> Option<ConnectionPermit> { if let Some(max) = self.max.map(NonZeroUsize::get) { let admitted = - LIVE_CONNECTIONS.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |live| { + LIVE_CONNECTIONS.try_update(Ordering::Relaxed, Ordering::Relaxed, |live| { (live < max).then_some(live + 1) }); if let Err(live) = admitted { diff --git a/core/message_bus/tests/head_of_line.rs b/core/message_bus/tests/head_of_line.rs index e5bec8bb0..455cbd733 100644 --- a/core/message_bus/tests/head_of_line.rs +++ b/core/message_bus/tests/head_of_line.rs @@ -18,8 +18,8 @@ //! Verify that a slow peer cannot stall sends to other peers. //! //! Setup: a sender bus has two replica connections. Peer A reads normally; -//! peer B's listener accepts but then drops the read half (so the sender's -//! kernel TCP send buffer eventually fills, blocking peer B's writer task). +//! peer B completes the replica handshake, then stops reading (so the sender's +//! kernel TCP send buffer eventually fills, blocking that peer's writer task). //! With the per-peer queue model, sends to peer A must remain O(microseconds) //! regardless of how blocked peer B is. @@ -32,6 +32,8 @@ use common::{ use compio::net::TcpListener; use iggy_binary_protocol::{Command, HEADER_SIZE}; use message_bus::connector::start as start_connector; +use message_bus::replica::auth::ChannelBinding; +use message_bus::replica::handshake::{ReplicaHandshakeCtx, acceptor_handshake}; use message_bus::replica::listener::{MessageHandler, bind, run}; use message_bus::{IggyMessageBus, MessageBus, SendError}; use std::cell::Cell; @@ -39,6 +41,7 @@ use std::rc::Rc; use std::time::{Duration, Instant}; const CLUSTER: u128 = 0xF00D; +const PEER_B_ID: u8 = 2; #[compio::test] async fn slow_peer_does_not_block_other_peers() { @@ -58,30 +61,28 @@ async fn slow_peer_does_not_block_other_peers() { }); bus_a.track_background(la_handle); - // Peer B: raw TCP listener that accepts connections but never reads - // from them. This is how we force head-of-line blocking: bus0's dial - // completes (TCP connect + Ping write into the peer's kernel recv - // buffer), but subsequent Prepare frames pile up. Once the kernel send - // buffer on bus0's side and the per-peer queue are both full, - // send_to_replica(2) returns Backpressure. + // Peer B joins through the normal handshake, then stops reading so + // subsequent Prepare frames backpressure its independent writer. let lb = TcpListener::bind(loopback()).await.unwrap(); let addr_b = lb.local_addr().unwrap(); + let peer_b_ctx = ReplicaHandshakeCtx { + self_id: PEER_B_ID, + ..bus_a.replica_handshake_ctx().unwrap().clone() + }; let held_streams: Rc<std::cell::RefCell<Vec<compio::net::TcpStream>>> = Rc::new(std::cell::RefCell::new(Vec::new())); let held_streams_clone = held_streams.clone(); let accept_b_handle = compio::runtime::spawn(async move { while let Ok((mut stream, _)) = lb.accept().await { - // Complete the two-way mesh handshake (consume the dialer's - // Ping, ack it) so the dial installs; only then go mute. The - // head-of-line scenario needs an installed peer that stops - // reading, not a peer that never joins. - if message_bus::framing::read_message(&mut stream, HEADER_SIZE) - .await - .is_ok() - { - let ack = header_only(Command::Ping, CLUSTER, 2); - let _ = message_bus::framing::write_message(&mut stream, ack).await; - } + let peer_id = acceptor_handshake( + &mut stream, + &peer_b_ctx, + ChannelBinding::Plaintext, + HEADER_SIZE, + ) + .await + .expect("slow peer must complete replica admission"); + assert_eq!(peer_id, 0); held_streams_clone.borrow_mut().push(stream); } }); @@ -94,14 +95,14 @@ async fn slow_peer_does_not_block_other_peers() { start_connector( &bus0, 0, - vec![(1, addr_a), (2, addr_b)], + vec![(1, addr_a), (PEER_B_ID, addr_b)], dial_delegate, bus0.config().reconnect_period, ) .await; let deadline = Instant::now() + Duration::from_secs(2); - while !(bus0.replicas().contains(1) && bus0.replicas().contains(2)) { + while !(bus0.replicas().contains(1) && bus0.replicas().contains(PEER_B_ID)) { assert!(Instant::now() < deadline, "both replicas must connect"); compio::time::sleep(Duration::from_millis(5)).await; } @@ -119,7 +120,7 @@ async fn slow_peer_does_not_block_other_peers() { let mut b_saturated = false; for _ in 0..100_000 { match bus0 - .send_to_replica(2, header_only(Command::Prepare, 0, 0).into_frozen()) + .send_to_replica(PEER_B_ID, header_only(Command::Prepare, 0, 0).into_frozen()) .await { Ok(()) => {} @@ -147,7 +148,7 @@ async fn slow_peer_does_not_block_other_peers() { let send_b_start = Instant::now(); let send_b_result = bus0 - .send_to_replica(2, header_only(Command::Prepare, 0, 0).into_frozen()) + .send_to_replica(PEER_B_ID, header_only(Command::Prepare, 0, 0).into_frozen()) .await; let send_b_elapsed = send_b_start.elapsed(); let b_backpressured = matches!(send_b_result, Err(SendError::Backpressure)); diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 09c46a4b2..521b358cb 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -4605,25 +4605,6 @@ where return; } } - let durable = match message.header().operation { - Operation::SendMessages => self.durability().is_persisted(), - Operation::StoreConsumerOffset | Operation::DeleteConsumerOffset => { - self.consumer_offset_durability().is_persisted() - && consumer_offset.is_some_and(|(_, _, _, ack)| ack == AckLevel::Quorum) - } - _ => false, - }; - if !durable { - Self::send_partition_deny_or_log( - consensus, - message.header(), - IggyError::DurabilityRequired.as_code(), - "durable retry contract unavailable", - reply.take(), - ) - .await; - return; - } if !consensus.has_retry_capacity(&self.dedup, client_id) { Self::send_partition_deny_or_log( consensus, @@ -6031,6 +6012,8 @@ where .as_ref() .filter(|persistence| persistence.segment_checkpoint().is_some()) { + // Recovery only keeps segment bytes covered by the published WAL frontier. + persistence.sync(); self.start_persistence(); // Shutdown and transfer must finish the flush, but never drain while // holding the lock used to append new messages. @@ -7437,7 +7420,10 @@ where Ok(()) } - async fn rotate_segment(&mut self, config: &PartitionsConfig) -> Result<(), IggyError> { + pub(crate) async fn rotate_segment( + &mut self, + config: &PartitionsConfig, + ) -> Result<(), IggyError> { let start_offset = self.log.active_segment().end_offset + 1; self.rotate_segment_at(config, start_offset).await } @@ -7811,6 +7797,23 @@ where &mut self, config: &PartitionsConfig, ) -> Result<(), IggyError> { + if let Some(persistence) = self.persistence.as_ref().map(Rc::clone) { + // Retained WAL bodies still belong to the old chain. Materialize + // and checkpoint them before moving the append point past it. + while self.consensus.commit_min() < self.consensus.commit_max() { + let before = self.consensus.commit_min(); + self.commit_journal(config).await; + if self.consensus.commit_min() == before { + return Err(IggyError::CannotSyncFile); + } + } + self.flush_committed_messages(config).await?; + persistence.request_checkpoint(); + self.checkpoint_persistence(config).await; + if self.fatal.is_some() || persistence.checkpoint_op() != persistence.head() { + return Err(IggyError::CannotSyncFile); + } + } // Where the next append will land: the counter, or an armed mint floor // above it. The floor is the whole reason a hole can appear, so // anchoring to the counter alone would leave the chain as unprepared. @@ -7944,6 +7947,18 @@ where } Some(_) => {} } + if let Some(persistence) = self.persistence.as_ref().map(Rc::clone) + && persistence.segment_checkpoint().is_some_and(|checkpoint| { + checkpoint.next_offset < self.log.active_segment().start_offset + }) + { + persistence.reanchor_segments(self.log.active_segment().start_offset); + self.start_persistence(); + persistence.drain_with_timeout().await.map_err(|error| { + warn!(%error, "cannot reanchor the partition WAL at its reserved offset"); + IggyError::CannotSyncFile + })?; + } Ok(()) } @@ -12524,6 +12539,12 @@ mod tests { assert_eq!(partition.log.journal().info.messages_count, 0); } } + if let Some(persistence) = &persistence { + assert_eq!( + persistence.durable_op(), + if durability.is_persisted() { 3 } else { 0 } + ); + } partition.flush_committed_messages(&config).await.unwrap(); let mut actual = Vec::new(); for segment in partition.log.segments() { @@ -12555,10 +12576,7 @@ mod tests { assert_eq!(partition.log.journal().info.messages_count, 0); if let Some(persistence) = persistence { assert!(persistence.segment_checkpoint().is_some()); - assert_eq!( - persistence.durable_op(), - if durability.is_persisted() { 3 } else { 0 } - ); + assert_eq!(persistence.durable_op(), 3); } } } @@ -15528,7 +15546,7 @@ mod tests { } #[compio::test] - async fn given_no_ack_explicit_store_when_submitted_should_refuse_without_mutation() { + async fn given_no_ack_explicit_store_when_submitted_should_apply_without_replication() { let dir = tempfile::tempdir().unwrap(); let (mut partition, sent) = recording_partition(); partition.set_partition_dir(dir.path().to_string_lossy().into_owned()); @@ -15540,10 +15558,13 @@ mod tests { None, ) .await; - assert!(partition.consumer_offsets.pin().is_empty()); + assert_eq!( + partition.get_consumer_offset(PollingConsumer::Consumer(7, 0)), + Some(0) + ); assert_eq!( partition.durable_consumer_offset_count(ConsumerKind::Consumer), - 0 + 1 ); assert_eq!(partition.consensus().pipeline_len(), 0); assert_eq!(partition.persistence.as_ref().unwrap().head(), 0); @@ -15551,7 +15572,7 @@ mod tests { let header = bytemuck::checked::from_bytes::<ReplyHeader>( &sent[0].1.as_slice()[..size_of::<ReplyHeader>()], ); - assert_eq!(header.status, IggyError::DurabilityRequired.as_code()); + assert_eq!(header.status, 0); assert_eq!(header.op, 0); } diff --git a/core/partitions/src/partition_storage.rs b/core/partitions/src/partition_storage.rs index b4cec0149..b64cca237 100644 --- a/core/partitions/src/partition_storage.rs +++ b/core/partitions/src/partition_storage.rs @@ -428,9 +428,8 @@ fn seed_recovered_offset<B: MessageBus>( /// Provision an initial segment + writers for a partition that has none. /// -/// No-op when `partition.log.has_segments()` already returns `true` -/// (recovery hydrated existing segments), so callers can invoke this -/// unconditionally. +/// With WAL-owned messages, restore the deterministic empty tail when the +/// recovered last segment already reached its size limit. /// /// # Errors /// @@ -443,6 +442,11 @@ pub async fn ensure_initial_segment<B: MessageBus>( wal_owned_messages: bool, ) -> Result<(), PartitionRecoveryError> { if partition.log.has_segments() { + if wal_owned_messages + && partition.log.active_segment().size >= partition.log.active_segment().max_size + { + partition.rotate_segment(config).await?; + } return Ok(()); } let stream_id = namespace.stream_id(); diff --git a/core/partitions/src/persistence.rs b/core/partitions/src/persistence.rs index b458eb947..385187a8f 100644 --- a/core/partitions/src/persistence.rs +++ b/core/partitions/src/persistence.rs @@ -441,6 +441,10 @@ enum Mutation<S: DurableStorage> { initial: SegmentPosition, max_size: u64, }, + ReanchorSegments { + epoch: u64, + next_offset: u64, + }, CertifyView { epoch: u64, view: u32, @@ -458,6 +462,9 @@ enum Mutation<S: DurableStorage> { durable: bool, retained_bytes: u64, }, + Sync { + epoch: u64, + }, Truncate { epoch: u64, from_op: u64, @@ -711,6 +718,16 @@ impl<S: DurableStorage> PartitionPersistence<S> { self.segment_checkpoint.get() } + /// Queue a reserved offset boundary after checkpointing the preceding chain. + pub fn reanchor_segments(&self, next_offset: u64) { + self.queue + .borrow_mut() + .push_back(Mutation::ReanchorSegments { + epoch: self.epoch.get(), + next_offset, + }); + } + pub fn enable_segment_storage(&self, initial: SegmentPosition, max_size: u64) { self.queue.borrow_mut().push_back(Mutation::EnableSegments { epoch: self.epoch.get(), @@ -1152,6 +1169,13 @@ impl<S: DurableStorage> PartitionPersistence<S> { .map_err(|_| io::Error::new(io::ErrorKind::TimedOut, "partition WAL drain timed out"))? } + /// Queue a barrier for preceding buffered writes; drain before claiming durability. + pub fn sync(&self) { + self.queue.borrow_mut().push_back(Mutation::Sync { + epoch: self.epoch.get(), + }); + } + pub fn start(&self) -> bool { let start = !self.retired.get() && self.failure.borrow().is_none() @@ -1209,7 +1233,9 @@ impl<S: DurableStorage> PartitionPersistence<S> { .. } => (*epoch, *retained_bytes), Mutation::CertifyView { epoch, .. } + | Mutation::Sync { epoch } | Mutation::EnableSegments { epoch, .. } + | Mutation::ReanchorSegments { epoch, .. } | Mutation::Purge { epoch, .. } | Mutation::Truncate { epoch, .. } | Mutation::Checkpoint { epoch, .. } @@ -1221,6 +1247,7 @@ impl<S: DurableStorage> PartitionPersistence<S> { let rebuild_references = matches!( mutation, Mutation::EnableSegments { .. } + | Mutation::ReanchorSegments { .. } | Mutation::Purge { .. } | Mutation::Truncate { .. } | Mutation::Checkpoint { .. } @@ -1311,6 +1338,9 @@ impl<S: DurableStorage> PartitionPersistence<S> { Mutation::EnableSegments { initial, max_size, .. } => journal.enable_segment_storage(initial, max_size).await, + Mutation::ReanchorSegments { next_offset, .. } => { + journal.reanchor_segment_storage(next_offset).await + } Mutation::CertifyView { view, op, checksum, .. } => journal.certify_log_view(view, op, checksum).await, @@ -1323,6 +1353,7 @@ impl<S: DurableStorage> PartitionPersistence<S> { self.append_batch(journal, prepare, durable, epoch, retained_bytes) .await } + Mutation::Sync { .. } => journal.sync().await, Mutation::Truncate { from_op, .. } => journal.truncate_from(from_op).await, Mutation::Checkpoint { through_op, diff --git a/core/partitions/tests/consumer_offset_history.rs b/core/partitions/tests/consumer_offset_history.rs index 4d0aa4d77..2f3dcafc9 100644 --- a/core/partitions/tests/consumer_offset_history.rs +++ b/core/partitions/tests/consumer_offset_history.rs @@ -226,11 +226,11 @@ async fn given_many_obsolete_offsets_when_purged_should_finish_rejections_and_re .on_request(append(2, b"old pending prepare"), None) .await; let mut receivers = Vec::new(); - for request_id in 1..=6 { + for request_id in 1_u64..=6 { let (sender, receiver) = oneshot_channel(); partition .on_request( - store(42, request_id, ConsumerKind::Consumer, 0), + store(42 + u128::from(request_id), 1, ConsumerKind::Consumer, 0), Some(sender), ) .await; @@ -284,10 +284,10 @@ async fn given_queued_delete_when_replacement_checkpoint_commits_should_preserve .await; partition.on_request(append(3, REPLACEMENT), None).await; let mut obsolete_stores = Vec::new(); - for request_id in 2..=5 { + for request_id in 2_u64..=5 { let (sender, receiver) = oneshot_channel(); partition - .on_request(store(42, request_id, kind, 0), Some(sender)) + .on_request(store(42 + u128::from(request_id), 1, kind, 0), Some(sender)) .await; obsolete_stores.push(receiver); } @@ -453,7 +453,7 @@ fn append(request: u64, payload: &[u8]) -> Message<RoutedRequestHeader> { .encode_request(RoutedRequestHeader { command: Command::Request, operation: Operation::SendMessages, - client: 1, + client: u128::from(request), session: 1, request, group: namespace().inner(), diff --git a/core/sdk/src/clients/producer.rs b/core/sdk/src/clients/producer.rs index f9dae5741..c0fdefe3f 100644 --- a/core/sdk/src/clients/producer.rs +++ b/core/sdk/src/clients/producer.rs @@ -653,8 +653,8 @@ unsafe impl Sync for IggyProducer {} /// `base_offset` assigned to the first message in that chunk. The list can be empty when the /// server supplies no offsets, and a background producer always returns an empty list. /// -/// Confirmed sends require persisted durability and recoverable storage on a quorum. -/// A weaker topic policy returns [`IggyError::DurabilityRequired`] before admission. +/// Confirmed sends follow the topic's durability policy. Persisted durability also +/// provides recoverable storage and crash-safe retry receipts on a quorum. /// /// # Retrying and what a failure means /// @@ -728,7 +728,7 @@ unsafe impl Sync for IggyProducer {} /// | [`partitioning()`] | [`Partitioning::balanced()`] | which partition a batch lands in | /// | [`partitioner()`] | none | computing the partition on the client instead | /// | [`send_retries()`] | three retries, one-second interval after the immediate first retry | retrying a failed request | -/// | [`topic_durability()`](crate::clients::producer_builder::IggyProducerBuilder::topic_durability) | `Durability::Replicated` | producer sends require `Durability::Persisted` | +/// | [`topic_durability()`](crate::clients::producer_builder::IggyProducerBuilder::topic_durability) | `Durability::Replicated` | crash-safe retry receipts require `Durability::Persisted` | /// | [`create_stream_if_not_exists()`] | on | creating the stream during [`init()`](Self::init) | /// | [`create_topic_if_not_exists()`] | on, one partition, server defaults for expiry and max size | creating the topic during [`init()`](Self::init) | /// | [`encryptor()`] | inherited from the client | encrypting payloads and user headers | @@ -883,8 +883,7 @@ impl IggyProducer { /// [`IggyProducerBuilder::create_topic_if_not_exists`](crate::clients::producer_builder::IggyProducerBuilder::create_topic_if_not_exists). /// Message durability comes from /// [`IggyProducerBuilder::topic_durability`](crate::clients::producer_builder::IggyProducerBuilder::topic_durability). - /// This producer's sends require `Durability::Persisted`; weaker settings - /// return `IggyError::DurabilityRequired`. Other settings come from + /// `Durability::Persisted` also provides crash-safe retry receipts. Other settings come from /// [`TopicCreateOptions::default`]. /// /// # Errors @@ -917,7 +916,8 @@ impl IggyProducer { /// An offset is a position, not an identity. An application resend may /// commit the same messages at another offset, see /// [Retrying and what a failure means](IggyProducer#retrying-and-what-a-failure-means). - /// Confirmed sends require persisted durability and recoverable storage on a quorum. + /// Confirmed sends follow the topic's durability policy; Persisted also waits + /// for recoverable storage on a quorum. /// /// # How long the call takes /// diff --git a/core/sdk/src/clients/producer_builder.rs b/core/sdk/src/clients/producer_builder.rs index 6cc176790..ea6003a62 100644 --- a/core/sdk/src/clients/producer_builder.rs +++ b/core/sdk/src/clients/producer_builder.rs @@ -179,8 +179,8 @@ impl IggyProducerBuilder { } /// Sets message durability for an automatically created topic. - /// Defaults to Replicated. Producer sends require `Durability::Persisted` - /// and otherwise fail with `IggyError::DurabilityRequired`. + /// Defaults to Replicated. `Durability::Persisted` also provides stable + /// storage and crash-safe retry receipts. pub fn topic_durability(self, durability: Durability) -> Self { Self { topic_durability: durability, diff --git a/core/server/README.md b/core/server/README.md index a8b154b72..4243b88e9 100644 --- a/core/server/README.md +++ b/core/server/README.md @@ -10,10 +10,15 @@ Clients connect over TCP (custom binary protocol), QUIC, WebSocket, or the HTTP cargo run --bin iggy-server --release ``` -Initialize a new data directory once with `--fresh`. This deletes any existing -data in the configured system path, so omit it on subsequent starts. An empty -replacement disk in cluster mode must recover through a live quorum. It cannot -initialize a new history merely because its peers are unreachable. +Missing or empty storage is initialized automatically. Existing incompatible +storage is refused without modification. `--fresh` deletes existing data in +the configured system path and must be an explicit reset, never a restart +default. A new cluster starts automatically when its metadata WAL and +superblock are absent. A node with retained metadata rejoins through recovery. +Loss of a replica's entire metadata directory is outside the automatic recovery +guarantee, including loss on every replica. Empty metadata cannot distinguish a +new cluster from a previously used one. Restore a verified consistent backup +through a validated recovery procedure before restarting erased metadata disks. The Docker image `apache/iggy:latest` ships the server together with the CLI; the `edge` tag tracks the latest development build. @@ -32,8 +37,7 @@ To run one node of a cluster, pass its replica ID from the `cluster.nodes` roste cargo run --bin iggy-server --release -- --replica-id 0 ``` -`--replica-id` selects the configured replica. `--fresh` explicitly initializes -its storage. Other settings come from configuration. +`--replica-id` selects the configured replica. Other settings come from configuration. ## Configuration @@ -58,10 +62,13 @@ External group offsets belong to groups managed outside Iggy, such as a Kafka gateway. They require no Iggy group membership and can exceed the partition's message-offset range. -Durability defaults remain `Replicated`. Explicit sends require `Persisted`; -explicit offset writes require `Persisted` and `Quorum`. Weaker settings return -`DurabilityRequired` (86) before mutation. Internal auto-commit polls keep their -existing durability policy. +Durability defaults remain `Replicated`, and ordinary SDK sends and explicit +offset writes support the configured policy. Crash-safe send retries require +`Persisted`; crash-safe explicit offset retries require `Persisted` and +`Quorum`. Weaker policies can lose data and receipts on a crash. A retained +receipt still replays its original result, but a lost receipt cannot prevent +another execution. NoAck and internal auto-commit polls retain their weaker +completion contracts. ## Upgrade recovery diff --git a/core/server/src/boot/mod.rs b/core/server/src/boot/mod.rs index 0acd49822..e424c7f95 100644 --- a/core/server/src/boot/mod.rs +++ b/core/server/src/boot/mod.rs @@ -123,9 +123,6 @@ pub async fn prepare_runtime_dirs( } else { validate_storage_format(&system_path)? }; - if !fresh && !format_present && !config.cluster.enabled { - return Err(ServerError::UnsupportedStorage { path: system_path }); - } if fresh { wipe_system_path(config).await?; } @@ -222,9 +219,9 @@ async fn wipe_system_path(config: &ServerConfig) -> Result<(), ServerError> { if config.cluster.enabled { warn!( path = %resolved.display(), - "--fresh wipes only this replica, which then refills from the cluster by \ - state transfer; wiping a quorum at once destroys committed data, and a \ - service unit file carrying --fresh re-transfers everything on every restart" + "--fresh wipes only this replica; recovery after erasing its entire metadata \ + directory is outside the automatic recovery guarantee. Wiping a quorum \ + destroys committed data. Do not keep --fresh in a restart service command" ); } @@ -276,7 +273,6 @@ async fn wipe_system_path(config: &ServerConfig) -> Result<(), ServerError> { pub fn bootstrap( config: ServerConfig, current_replica_id: Option<u8>, - fresh: bool, ) -> Result<ShardHandles, ServerError> { // One process-wide rustls provider, installed before any shard thread // exists. rustls is compiled with both `ring` and `aws-lc-rs`, so a @@ -422,7 +418,6 @@ pub fn bootstrap( shard_id, total_shards, current_replica_id, - fresh, assignment, senders_for_shard, inbox, @@ -498,7 +493,6 @@ async fn shard_main( shard_id: u16, total_shards: u16, replica_id: Option<u8>, - fresh: bool, senders: Vec<TaggedSender>, inbox: ShardReceiver<ShardFrame>, reply_inbox: ShardReceiver<ShardFrame>, @@ -662,8 +656,7 @@ async fn shard_main( // it. Reuse that superblock rather than re-opening it, which would fork the // ping-pong sequence counter. Consensus recovers its true (view, log_view) // from `recovered_state` instead of inferring a stale view from the WAL. - let consensus = - restore_metadata_consensus(&owner, &topology, config, Rc::clone(&bus), fresh); + let consensus = restore_metadata_consensus(&owner, &topology, config, Rc::clone(&bus)); let superblock = Rc::new(owner.superblock); ( Some(consensus), @@ -1158,6 +1151,40 @@ const fn operation_triggers_partition_reconcile(op: Operation) -> bool { mod tests { use super::*; + #[compio::test] + async fn given_missing_or_empty_storage_when_booting_should_initialize_without_wiping_on_restart() + { + let mut logging = Logging::new(crate::VERSION); + logging.early_init(); + for exists in [false, true] { + let directory = tempfile::tempdir().unwrap(); + let data_path = directory.path().join("data"); + if exists { + std::fs::create_dir(&data_path).unwrap(); + } + let mut config = ServerConfig { + path: data_path.to_string_lossy().into_owned(), + ..ServerConfig::default() + }; + config.logging.file_enabled = false; + config.telemetry.enabled = false; + prepare_runtime_dirs(&config, &mut logging, false) + .await + .unwrap(); + assert_eq!( + std::fs::read(data_path.join(STORAGE_FORMAT_FILE)).unwrap(), + STORAGE_FORMAT + ); + let sentinel = data_path.join("existing-data"); + let bytes = b"preserve existing data"; + std::fs::write(&sentinel, bytes).unwrap(); + prepare_runtime_dirs(&config, &mut logging, false) + .await + .unwrap(); + assert_eq!(std::fs::read(sentinel).unwrap(), bytes); + } + } + #[test] fn incompatible_storage_is_refused_without_changing_existing_files() { let directory = tempfile::tempdir().unwrap(); diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs index 572849fb6..fcb986944 100644 --- a/core/server/src/boot/recovery.rs +++ b/core/server/src/boot/recovery.rs @@ -504,7 +504,6 @@ pub(in crate::boot) fn restore_metadata_consensus( topology: &TcpTopology, config: &ServerConfig, bus: Rc<IggyMessageBus>, - fresh: bool, ) -> VsrConsensus<Rc<IggyMessageBus>> { let journal = &owner.journal; let replica_count = topology.replica_count; @@ -547,7 +546,7 @@ pub(in crate::boot) fn restore_metadata_consensus( // finds, then journal repair fills the tail. If the probe exhausts // instead -- full-cluster bootstrap, nobody live to fetch from -- the // election fallback clears the stage and this local recovery stands. - let join = if replica_count > 1 && (!fresh || restored_op > 0 || recovered_state.is_some()) { + let join = if replica_count > 1 && (restored_op > 0 || recovered_state.is_some()) { JoinMode::ProbeAsBackup { await_state_transfer: true, } @@ -591,7 +590,6 @@ pub(in crate::boot) fn restore_metadata_consensus( join, }, ); - consensus.set_recovery_election_allowed(fresh || restored_op > 0); consensus.sequencer().set_sequence(restored_op); // A SOLO replica's durable journal head IS its commit point: quorum is // 1-of-1, so an entry commits the instant it is durable, and the acks diff --git a/core/server/src/boot/threads.rs b/core/server/src/boot/threads.rs index 119ace3e1..12bfec460 100644 --- a/core/server/src/boot/threads.rs +++ b/core/server/src/boot/threads.rs @@ -645,7 +645,6 @@ pub(in crate::boot) fn run_shard_thread( shard_id: u16, total_shards: u16, replica_id: Option<u8>, - fresh: bool, assignment: ShardInfo, senders: Vec<TaggedSender>, inbox: ShardReceiver<ShardFrame>, @@ -697,7 +696,6 @@ pub(in crate::boot) fn run_shard_thread( shard_id, total_shards, replica_id, - fresh, senders, inbox, reply_inbox, diff --git a/core/server/src/dispatch/host.rs b/core/server/src/dispatch/host.rs index 01b2069fa..8232fabba 100644 --- a/core/server/src/dispatch/host.rs +++ b/core/server/src/dispatch/host.rs @@ -50,6 +50,7 @@ use shard::{ListClientsReply, MetadataSubmit, ShardHost}; use std::cell::RefCell; use std::rc::Rc; use std::sync::Arc; +use std::time::Duration; /// The shard host the server runs; see the module docs. /// @@ -187,12 +188,14 @@ where let _ = reply.try_send( self.sessions .borrow() - .iter_consumer_sessions( + .iter_consumer_sessions(if self.server_config.heartbeat.enabled { self.server_config .consumer_group .session_timeout - .get_duration(), - ) + .get_duration() + } else { + Duration::MAX + }) .collect(), ); } diff --git a/core/server/src/dispatch/mod.rs b/core/server/src/dispatch/mod.rs index 397d31753..8b2863326 100644 --- a/core/server/src/dispatch/mod.rs +++ b/core/server/src/dispatch/mod.rs @@ -59,19 +59,25 @@ use crate::shell::{ShellBus, ShellShard, ShellShardHandle}; use crate::wire::verify_request_checksum; use ahash::{AHashMap, AHashSet}; use configs::server::ServerConfig; +use consensus::MetadataHandle; +use futures::future::{Either, select}; use iggy_binary_protocol::PrepareHeader; use iggy_binary_protocol::codes::{ LOGIN_USER_CODE, LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, PING_CODE, }; +use iggy_binary_protocol::requests::system::SessionIdentity; +use iggy_binary_protocol::requests::users::login_register::BindSecret; use iggy_binary_protocol::{ EvictionReason, GenericHeader, Operation, RequestHeader, RoutedRequestHeader, }; use iggy_common::IggyError; use journal::superblock::SuperblockStore; use journal::{Journal, JournalHandle}; +use secrecy::ExposeSecret; use server_common::Message; use std::cell::RefCell; use std::collections::VecDeque; +use std::pin::pin; use std::rc::Rc; use std::sync::Arc; use tracing::{debug, warn}; @@ -787,6 +793,60 @@ where .and_then(std::rc::Weak::upgrade) } +#[allow(clippy::future_not_send)] +async fn complete_session_binding<B, MJ, S, SB>( + shard: &Rc<ShellShard<B, MJ, S, SB>>, + sessions: &Rc<RefCell<SessionManager>>, + transport_client_id: u128, + identity: SessionIdentity, + secret: BindSecret, +) -> Result<(u32, u64), IggyError> +where + B: ShellBus, + MJ: JournalHandle + 'static, + MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, + S: 'static, + SB: SuperblockStore + 'static, +{ + let watermark = identity.metadata_watermark.max(identity.session); + reads::await_metadata_read_frontier(shard, watermark).await?; + let (user_id, session, attachment) = if shard.id == 0 { + shard + .plane + .metadata() + .client_table + .borrow_mut() + .bind_session(identity.client_id, identity.session, secret.expose_secret())? + } else { + let (reply, receiver) = shard::channel(1); + shard.forward_metadata_submit(shard::MetadataSubmit::BindSession { + vsr_client_id: identity.client_id, + session: identity.session, + secret, + reply, + }); + let outcome = pin!(receiver.recv()); + let deadline = pin!( + shard + .bus + .sleep(shard.plane.metadata().applied_frontier().read_budget()) + ); + match select(outcome, deadline).await { + Either::Left((result, _)) => result.map_err(|_| IggyError::TransientNotAccepted)??, + Either::Right(_) => return Err(IggyError::TransientNotAccepted), + } + }; + sessions.borrow_mut().bind_authenticated_connection( + transport_client_id, + identity.client_id, + session, + user_id, + attachment, + watermark, + )?; + Ok((user_id, session)) +} + #[cfg(test)] mod tests { use super::*; diff --git a/core/server/src/dispatch/reads.rs b/core/server/src/dispatch/reads.rs index 7dc088486..6db759553 100644 --- a/core/server/src/dispatch/reads.rs +++ b/core/server/src/dispatch/reads.rs @@ -27,6 +27,7 @@ use crate::cluster_meta::ClusterRoster; use crate::dispatch::authz::{authorize_default_read, authorize_partition_read, authorize_uid}; +use crate::dispatch::complete_session_binding; use crate::dispatch::failure::{ FrameChannel, send_host_frame, send_non_replicated_bytes, send_non_replicated_deny, }; @@ -62,7 +63,6 @@ use iggy_binary_protocol::requests::messages::PollMessagesRequest; use iggy_binary_protocol::requests::system::get_client::GetClientRequest; use iggy_binary_protocol::requests::system::get_snapshot::GetSnapshotRequest; use iggy_binary_protocol::requests::system::{BindSessionRequest, SessionIdentity}; -use iggy_binary_protocol::requests::users::login_register::BindSecret; use iggy_binary_protocol::responses::clients::client_response::ConsumerGroupInfoResponse; use iggy_binary_protocol::responses::clients::get_client::ClientDetailsResponse; use iggy_binary_protocol::responses::clients::get_clients::GetClientsResponse; @@ -78,7 +78,6 @@ use message_bus::framing::MAX_MESSAGE_SIZE; use metadata::AppliedFrontier; use metadata::impls::metadata::StreamsFrontend; use metadata::permissioner::Permissioner; -use secrecy::ExposeSecret; use server_common::Message; use shard::{PartitionRead, PartitionReadReply}; use std::cell::RefCell; @@ -291,7 +290,7 @@ pub async fn hold_for_frontier( /// The wait ends on the commit that closes the gap, not on a poll: the budget /// timer is the only timer armed, so a read that resumes costs one wake. #[allow(clippy::future_not_send)] -async fn await_metadata_read_frontier<B, MJ, S, SB>( +pub(super) async fn await_metadata_read_frontier<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, watermark: u64, ) -> Result<FrontierWait, IggyError> @@ -649,60 +648,6 @@ where .to_bytes()) } -#[allow(clippy::future_not_send)] -pub(in crate::dispatch) async fn complete_session_binding<B, MJ, S, SB>( - shard: &Rc<ShellShard<B, MJ, S, SB>>, - sessions: &Rc<RefCell<SessionManager>>, - transport_client_id: u128, - identity: SessionIdentity, - secret: BindSecret, -) -> Result<(u32, u64), IggyError> -where - B: ShellBus, - MJ: JournalHandle + 'static, - MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, - S: 'static, - SB: SuperblockStore + 'static, -{ - let watermark = identity.metadata_watermark.max(identity.session); - await_metadata_read_frontier(shard, watermark).await?; - let (user_id, session, attachment) = if shard.id == 0 { - shard - .plane - .metadata() - .client_table - .borrow_mut() - .bind_session(identity.client_id, identity.session, secret.expose_secret())? - } else { - let (reply, receiver) = shard::channel(1); - shard.forward_metadata_submit(shard::MetadataSubmit::BindSession { - vsr_client_id: identity.client_id, - session: identity.session, - secret, - reply, - }); - let outcome = pin!(receiver.recv()); - let deadline = pin!( - shard - .bus - .sleep(shard.plane.metadata().applied_frontier().read_budget()) - ); - match select(outcome, deadline).await { - Either::Left((result, _)) => result.map_err(|_| IggyError::TransientNotAccepted)??, - Either::Right(_) => return Err(IggyError::TransientNotAccepted), - } - }; - sessions.borrow_mut().bind_authenticated_connection( - transport_client_id, - identity.client_id, - session, - user_id, - attachment, - watermark, - )?; - Ok((user_id, session)) -} - #[allow(clippy::future_not_send, clippy::too_many_arguments)] async fn consumer_routing<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, diff --git a/core/server/src/dispatch/session_ops.rs b/core/server/src/dispatch/session_ops.rs index 9e8de7792..469298a0a 100644 --- a/core/server/src/dispatch/session_ops.rs +++ b/core/server/src/dispatch/session_ops.rs @@ -25,18 +25,17 @@ //! of that replica protocol on this page. Terminal failures surface as typed //! `Eviction` frames, transient ones as result-framed replay hints. //! -//! Deliberate asymmetry (the two session stores): the per-shard -//! `SessionManager` owns transport sessions (connection -> user binding, -//! heartbeats, SDK info) and is never replicated; the consensus `ClientTable` -//! owns replicated VSR sessions and their dedup watermarks. Login binds the -//! two together, so logout and eviction must release BOTH -- every teardown -//! path below pairs `remove_connection` with a replicated `Logout`. +//! The per-shard `SessionManager` owns local connection bindings, heartbeats, +//! and SDK info; the consensus `ClientTable` owns replicated logical sessions +//! and retry protection. Disconnect and heartbeat eviction remove only the +//! local binding. Explicit logout or committed lease expiry retires the shared +//! session and its consumer-group membership. +use crate::dispatch::complete_session_binding; use crate::dispatch::failure::{ FrameChannel, send_eviction, send_host_frame, send_result_rejection, }; use crate::dispatch::login_error::LoginRegisterError; -use crate::dispatch::reads::complete_session_binding; use crate::reply_frame::{ build_deny_reply, build_empty_reply, build_login_register_reply, current_metadata_commit, }; @@ -366,9 +365,9 @@ const fn eviction_reason_for(error: &LoginRegisterError) -> EvictionReason { /// Per-shard heartbeat verifier: evict connections that have not pinged within /// `1.2 x interval`. Mirrors the legacy `verify_heartbeats` periodic task. -/// Eviction reuses the disconnect path (drops the client from its consumer -/// groups + rebalances via the replicated `Logout`) and sends a session- -/// terminal `Eviction(StaleClient)` so the client fails fast and can reconnect. +/// Eviction removes the connection binding and sends `Eviction(StaleClient)`. +/// The shared logical session and its group membership remain until committed +/// lease expiry or explicit logout. #[allow(clippy::future_not_send)] pub async fn run_heartbeat_verifier<B, MJ, S, SB>( shard: Rc<ShellShard<B, MJ, S, SB>>, @@ -403,14 +402,8 @@ pub async fn run_heartbeat_verifier<B, MJ, S, SB>( .borrow() .collect_stale(max_age, std::time::Instant::now()); for transport_client_id in stale { - // The heartbeat verifier exists to release a dead client's - // consumer-group membership (so the group rebalances off it). A - // connection that holds no membership has nothing for the eviction - // to clean up; reaping it would only drop a still-usable session - // (e.g. an idle admin connection that polls between long gaps), - // which the legacy server tolerates. The real transport-disconnect - // path still reaps it on socket close. So only evict a stale - // connection that is actually a group member. + // Preserve idle admin connections. Group-member connections get an + // eviction notice; membership ends through committed lease expiry. let is_group_member = sessions .borrow() .bound_client_id(transport_client_id) @@ -430,9 +423,7 @@ pub async fn run_heartbeat_verifier<B, MJ, S, SB>( } } -/// Evict one stale connection: drop its session (releasing consumer-group -/// membership through a replicated `Logout`) and notify the client with a -/// session-terminal `Eviction(StaleClient)`. +/// Remove one stale connection binding and notify it with `Eviction(StaleClient)`. #[allow(clippy::future_not_send)] async fn evict_stale_client<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, diff --git a/core/server/src/main.rs b/core/server/src/main.rs index f823f6320..75f8303b1 100644 --- a/core/server/src/main.rs +++ b/core/server/src/main.rs @@ -100,7 +100,7 @@ fn main() -> Result<(), ServerError> { let config = bootstrap_result?; drop(bootstrap_runtime); - let shards = bootstrap(config, args.replica_id, args.fresh)?; + let shards = bootstrap(config, args.replica_id)?; if let Err(error) = shards.install_ctrlc_handler() { // Without a working SIGINT handler the server has no way to // observe an operator Ctrl-C and the shutdown flag would never diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index 669a5c4dd..43e069790 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -686,42 +686,43 @@ async fn load_partition( .await?; partition.created_at = partition_metadata.created_at; - restore_partition_offsets( - &mut partition, - partitions_config, - recovered_state.as_ref(), - segment_checkpoint, - ) - .await?; + restore_partition_offsets(&mut partition, recovered_state.as_ref(), segment_checkpoint); let current_offset = partition.offset.load(Ordering::Acquire); configure_consumer_offsets(&mut partition, partitions_config, namespace, current_offset) .await?; - ensure_initial_segment( - &mut partition, - partitions_config, - namespace, - segment_checkpoint.is_some(), - ) - .await?; + if !partition.log.has_segments() { + ensure_initial_segment( + &mut partition, + partitions_config, + namespace, + segment_checkpoint.is_some(), + ) + .await?; + } open_partition_persistence(&mut partition, config, recovered_persistence).await?; + ensure_initial_segment(&mut partition, partitions_config, namespace, true).await?; + if replica_count == 1 && !partition.durability().is_persisted() { + partition + .reanchor_to_offset_frontier(partitions_config) + .await + .map_err(|error| ServerError::Iggy(Box::new(error)))?; + } Ok(partition) } /// Restore the offset counter of a recovered partition from what boot could -/// prove about its offset space, then put the next append point where the -/// recovery walk can read it back. +/// prove about its offset space. /// /// Three carriers, weakest last: the sized segments' end offset, an empty /// chain's file name (a state-transfer install at the group frontier), and the /// superblock's durable frontier as a lower bound over both. -async fn restore_partition_offsets( +fn restore_partition_offsets( partition: &mut IggyPartition<Rc<IggyMessageBus>>, - partitions_config: &PartitionsConfig, recovered_state: Option<&VsrState>, segment_checkpoint: Option<SegmentPosition>, -) -> Result<(), ServerError> { +) { let sized_end = partition .log .segments() @@ -778,21 +779,6 @@ async fn restore_partition_offsets( // gone (an all-GC'd origin's install, a crash inside the swap window), and // taking the max means real recovered data always wins. partition.restore_offset_frontier(recovered_state); - // Minting from the reservation leaves a hole between the recovered chain - // and the new append point, and the recovery walk REFUSES a hole inside a - // segment (tombstoning the partition on the solo arm), so put it on a - // segment boundary instead. - // - // Solo only, in step with the reservation itself: a replicated group's - // segment boundaries must be a function of the batches alone or the - // reconciler's offset-keyed segment GC never converges. - if partition.consensus().replica_count() == 1 && !partition.durability().is_persisted() { - partition - .reanchor_to_offset_frontier(partitions_config) - .await - .map_err(|error| ServerError::Iggy(Box::new(error)))?; - } - Ok(()) } /// Recover this partition's persisted segment chain, stamping each segment diff --git a/core/server_common/src/poll.rs b/core/server_common/src/poll.rs index ec1006982..a06aef467 100644 --- a/core/server_common/src/poll.rs +++ b/core/server_common/src/poll.rs @@ -44,7 +44,7 @@ impl PollHistoryId { fn allocate(counter: &AtomicU64) -> Self { // The counter provides uniqueness, not publication of partition state. let id = counter - .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |next| { + .try_update(Ordering::Relaxed, Ordering::Relaxed, |next| { next.checked_add(1) }) .expect("poll history ID counter exhausted"); diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index db376537a..c6e6f4c6c 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -2790,11 +2790,11 @@ mod tests { // together remap every stream; and partition ops drawing from the one shared // request counter instead of a separate sequence based at `1<<63`, which // renumbers every partition request id and so every reply header in the trace. - // Multi-replica consumer-offset requests carrying `NoAck` now enter - // VSR, so their scheduling and committed replies contribute to the - // deterministic trace instead of taking the primary-local fast path. + // NoAck explicit offsets use the primary-local path. Ordinary Replicated + // writes remain admitted, so their policy-scoped replies contribute to + // the deterministic trace. assert_eq!( - h1, 0x8DAA_C1BA_2EE3_F490, + h1, 0x31C2_ADA9_9411_FCD4, "workload reply hash drifted from locked baseline" ); } diff --git a/core/simulator/src/storage/tests.rs b/core/simulator/src/storage/tests.rs index 2aada613c..434066d13 100644 --- a/core/simulator/src/storage/tests.rs +++ b/core/simulator/src/storage/tests.rs @@ -1396,6 +1396,19 @@ async fn storage_for_partition() -> SimStorage { storage } +async fn owned_journal_for_partition() -> (SimStorage, PartitionPrepareJournal<SimStorage>) { + let storage = storage_for_partition().await; + let mut journal = + PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7, storage.clone()) + .await + .unwrap(); + journal + .enable_segment_storage(SegmentPosition::default(), OWNED_BATCH_BYTES as u64) + .await + .unwrap(); + (storage, journal) +} + async fn queued_batch(count: u64) -> (SimStorage, Rc<PartitionPersistence<SimStorage>>) { let storage = storage_for_partition().await; let (persistence, _) = @@ -1512,105 +1525,104 @@ async fn segment_roll_during_wal_open(operation: StorageOperation) { } #[test] -fn buffered_owned_segments_rotate_without_barriers_and_persist_offset_predecessors() { +fn buffered_owned_rotations_publish_names_and_persist_offset_predecessors() { block_on(async { - let storage = storage_for_partition().await; - let mut journal = - PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7, storage.clone()) - .await - .unwrap(); - journal - .enable_segment_storage(SegmentPosition::default(), OWNED_BATCH_BYTES as u64) - .await - .unwrap(); - storage.clear_trace(); - let written_before: usize = storage.state.borrow().written_bytes.values().sum(); - let mut parent = 0; - let mut prepares = Vec::new(); - for offset in 0..3 { - let prepare = owned_prepare(offset + 1, parent, offset); - parent = prepare.header().checksum; - journal - .append_buffered(prepare.clone().into_frozen()) - .await - .unwrap(); - prepares.push(prepare); - } - assert!( - !storage.trace().iter().any(|operation| matches!( - operation, - StorageOperation::FileSync | StorageOperation::DirectorySync - )), - "replicated bodies must not require a barrier, including across append groups and rotations" - ); - assert_eq!(journal.durable_op(), 0); - assert_eq!(journal.size_bytes(), (3 * PARTITION_WAL_BLOCK_SIZE) as u64); - assert_eq!( - journal.retained_bytes(), - 3 * journal::partition_journal::record_length(prepares[0].as_slice().len()).unwrap() - as u64 - ); - let written_after: usize = storage.state.borrow().written_bytes.values().sum(); - assert_eq!( - written_after - written_before, - 3 * (OWNED_BATCH_BYTES + PARTITION_WAL_BLOCK_SIZE), - "each append writes one body and one metadata WAL record" - ); - let offset = offset_prepare(4, parent); - journal.append(offset.clone().into_frozen()).await.unwrap(); - prepares.push(offset); - for prepare in &prepares[..3] { - let reference = journal.segment_reference(prepare.header()).unwrap(); - let public = Path::new(DIRECTORY).join(format!("{:020}.log", reference.start_offset)); - let retained = Path::new(WAL).join(format!( - "segment-{}-{}.log", - reference.generation, reference.start_offset - )); - let state = storage.state.borrow(); - let inode = state.lookup(&public).unwrap(); - assert_eq!(inode, state.lookup(&retained).unwrap()); - assert_eq!(state.written_bytes[&inode], OWNED_BATCH_BYTES); - } - drop(journal); - storage.crash(Crash::PowerLoss); - let mut recovered = - PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7, storage.clone()) - .await - .unwrap(); - assert_eq!(recovered.durable_op(), 4); - assert_eq!( - recovered.segment_checkpoint(), - Some(SegmentPosition::default()) - ); - let actual = recovered.prepares().await.unwrap(); - assert_eq!(actual.len(), prepares.len()); - for (actual, expected) in actual.iter().zip(&prepares) { - assert_eq!(actual.as_slice(), expected.as_slice()); - } - recovered.checkpoint(2).await.unwrap(); - let retained_path = Path::new(DIRECTORY).join(format!("{:020}.log", 1)); - let reader = storage.open(&retained_path, OpenMode::Read).await.unwrap(); - for offset in 0..2 { - storage - .remove_file(&Path::new(DIRECTORY).join(format!("{offset:020}.log"))) - .await - .unwrap(); - } - storage.sync_directory(Path::new(DIRECTORY)).await.unwrap(); - assert_eq!( - reader.read(0, OWNED_BATCH_BYTES).await.unwrap(), - prepares[1].as_slice()[size_of::<PrepareHeader>()..] - ); - drop(recovered); - storage.crash(Crash::PowerLoss); - let recovered = PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7, storage) - .await - .unwrap(); - assert_eq!(recovered.checkpoint_op(), 2); - let actual = recovered.prepares().await.unwrap(); - assert_eq!(actual.len(), prepares.len() - 1); - for (actual, expected) in actual.iter().zip(&prepares[1..]) { - assert_eq!(actual.as_slice(), expected.as_slice()); + let checkpoint = Some(SegmentPosition::default()); + for persist_offset in [false, true] { + let (storage, mut journal) = owned_journal_for_partition().await; + let written_before: usize = storage.state.borrow().written_bytes.values().sum(); + let mut parent = 0; + let mut prepares = Vec::new(); + for offset in 0..3 { + let prepare = owned_prepare(offset + 1, parent, offset); + parent = prepare.header().checksum; + journal + .append_buffered(prepare.clone().into_frozen()) + .await + .unwrap(); + prepares.push(prepare); + } + assert_eq!(journal.durable_op(), 0); + assert_eq!(journal.size_bytes(), (3 * PARTITION_WAL_BLOCK_SIZE) as u64); + assert_eq!( + journal.retained_bytes(), + 3 * journal::partition_journal::record_length(prepares[0].as_slice().len()).unwrap() + as u64 + ); + let written_after: usize = storage.state.borrow().written_bytes.values().sum(); + assert_eq!( + written_after - written_before, + 3 * (OWNED_BATCH_BYTES + PARTITION_WAL_BLOCK_SIZE), + "each append writes one body and one metadata WAL record" + ); + let mut segment_names = (0..prepares.len()) + .map(|offset| Path::new(DIRECTORY).join(format!("{offset:020}.log"))) + .collect::<Vec<_>>(); + if persist_offset { + let offset = offset_prepare(4, parent); + journal.append(offset.clone().into_frozen()).await.unwrap(); + prepares.push(offset); + for prepare in &prepares[..3] { + let reference = journal.segment_reference(prepare.header()).unwrap(); + let public = + Path::new(DIRECTORY).join(format!("{:020}.log", reference.start_offset)); + let retained = Path::new(WAL).join(format!( + "segment-{}-{}.log", + reference.generation, reference.start_offset + )); + let state = storage.state.borrow(); + let inode = state.lookup(&public).unwrap(); + assert_eq!(inode, state.lookup(&retained).unwrap()); + assert_eq!(state.written_bytes[&inode], OWNED_BATCH_BYTES); + segment_names.push(retained); + } + } + drop(journal); + storage.crash(Crash::PowerLoss); + for name in segment_names { + assert!(storage.exists(&name).await.unwrap()); + } + let mut recovered = + PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7, storage.clone()) + .await + .unwrap(); + assert_eq!(recovered.durable_op(), if persist_offset { 4 } else { 0 }); + assert_eq!(recovered.segment_checkpoint(), checkpoint); + let actual = recovered.prepares().await.unwrap(); + if !persist_offset { + assert!(actual.is_empty(), "buffered WAL persisted"); + continue; + } + assert_eq!(actual.len(), prepares.len()); + for (actual, expected) in actual.iter().zip(&prepares) { + assert_eq!(actual.as_slice(), expected.as_slice()); + } + recovered.checkpoint(2).await.unwrap(); + let retained_path = Path::new(DIRECTORY).join(format!("{:020}.log", 1)); + let reader = storage.open(&retained_path, OpenMode::Read).await.unwrap(); + for offset in 0..2 { + storage + .remove_file(&Path::new(DIRECTORY).join(format!("{offset:020}.log"))) + .await + .unwrap(); + } + storage.sync_directory(Path::new(DIRECTORY)).await.unwrap(); + assert_eq!( + reader.read(0, OWNED_BATCH_BYTES).await.unwrap(), + prepares[1].as_slice()[size_of::<PrepareHeader>()..] + ); + drop(recovered); + storage.crash(Crash::PowerLoss); + let recovered = + PartitionPrepareJournal::open_with_storage(Path::new(WAL), 42, 7, storage) + .await + .unwrap(); + assert_eq!(recovered.checkpoint_op(), 2); + let actual = recovered.prepares().await.unwrap(); + assert_eq!(actual.len(), prepares.len() - 1); + for (actual, expected) in actual.iter().zip(&prepares[1..]) { + assert_eq!(actual.as_slice(), expected.as_slice()); + } } }); } diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs index 4d3138eab..158c1b966 100644 --- a/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs +++ b/foreign/csharp/Iggy_SDK.Tests.Integration/Fixtures/IggyServerFixture.cs @@ -63,6 +63,7 @@ public class IggyServerFixture : IAsyncInitializer, IAsyncDisposable /// be evicted once idle past 1.2 intervals. /// </summary> public static readonly TimeSpan ServerHeartbeatInterval = TimeSpan.FromSeconds(2); + public static readonly TimeSpan SessionTimeout = TimeSpan.FromSeconds(30); /// <summary> /// Extra environment variables for every node, layered over the cluster's base configuration. @@ -72,7 +73,8 @@ public class IggyServerFixture : IAsyncInitializer, IAsyncDisposable protected virtual Dictionary<string, string> EnvironmentVariables => new() { { "IGGY_HEARTBEAT_ENABLED", "true" }, - { "IGGY_HEARTBEAT_INTERVAL", $"{ServerHeartbeatInterval.TotalSeconds}s" } + { "IGGY_HEARTBEAT_INTERVAL", $"{ServerHeartbeatInterval.TotalSeconds}s" }, + { "IGGY_CONSUMER_GROUP_SESSION_TIMEOUT", $"{SessionTimeout.TotalSeconds}s" } }; /// <summary> diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/HeartbeatTests.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/HeartbeatTests.cs index c6b724579..52ca8aaf6 100644 --- a/foreign/csharp/Iggy_SDK.Tests.Integration/HeartbeatTests.cs +++ b/foreign/csharp/Iggy_SDK.Tests.Integration/HeartbeatTests.cs @@ -34,8 +34,9 @@ public class HeartbeatTests private const string TopicName = "heartbeat-topic"; private const string GroupName = "heartbeat-group"; - // Comfortably past the server's stale threshold of 1.2 intervals, with room for the verifier's own tick. - private static readonly TimeSpan IdleFor = IggyServerFixture.ServerHeartbeatInterval * 3; + // Transport eviction removes the binding; committed lease expiry removes membership. + private static readonly TimeSpan IdleFor = IggyServerFixture.SessionTimeout + + IggyServerFixture.ServerHeartbeatInterval * 3; [ClassDataSource<IggyServerFixture>(Shared = SharedType.PerAssembly)] public required IggyServerFixture Fixture { get; init; } diff --git a/foreign/csharp/README.md b/foreign/csharp/README.md index 6b0ff9a0d..48083dee9 100644 --- a/foreign/csharp/README.md +++ b/foreign/csharp/README.md @@ -34,9 +34,10 @@ a binary replacement cannot upgrade it. See the [server recovery guide](../../co Register creates a shared logical session; other TCP connections authenticate with BindSession (15), using the parent's identity and proof. A disconnect removes its binding without ending the session. Logout or committed expiry ends it; a new session must not replay an unresolved mutation from the old one. Command 14 is retired. -Explicit sends require Persisted message durability; explicit offset mutations require Persisted offset durability -and Quorum acknowledgement. Defaults remain Replicated; acknowledged weaker explicit mutations return -DurabilityRequired (86). HTTP NoAck's 202 confirms dispatch, not partition admission or commit, and can hide that refusal. +Ordinary SDK sends and explicit offset mutations support the configured durability, including the Replicated default. +Crash-safe send retries require Persisted message durability; crash-safe explicit offset retries require Persisted +offset durability and Quorum acknowledgement. Weaker policies can lose data and receipts on a crash. HTTP NoAck's +202 confirms dispatch, not partition admission or commit, and supplies no recoverable caller receipt. The following upgrade guidance describes the 0.9.0 protocol, not source builds of this checkout. Cluster auto-commit polling over TCP/TLS keeps group membership on the coordinator diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index 322c61e30..a404d2c89 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -1420,8 +1420,9 @@ impl IggyClient { >, send_retries: Option<i64>, send_retry_interval: RetryInterval, - #[gen_stub(override_type(type_repr = "Durability | None"))] - topic_durability: Option<&Bound<'_, PyAny>>, + #[gen_stub(override_type(type_repr = "Durability | None"))] topic_durability: Option< + &Bound<'_, PyAny>, + >, ) -> PyResult<Bound<'a, PyAny>> { let mode = mode.unwrap_or_default(); diff --git a/foreign/swift/Tools/golden-vectors/src/main.rs b/foreign/swift/Tools/golden-vectors/src/main.rs index cd5ded37f..20df2f8ed 100644 --- a/foreign/swift/Tools/golden-vectors/src/main.rs +++ b/foreign/swift/Tools/golden-vectors/src/main.rs @@ -31,8 +31,6 @@ use iggy_binary_protocol::batch::{ BATCH_HEADER_SIZE, BATCH_MESSAGE_HEADER_SIZE, BatchHeader, calculate_batch_checksum, }; use iggy_binary_protocol::codec::WireEncode; -use iggy_binary_protocol::requests::system::{BindSessionRequest, SessionIdentity}; -use iggy_binary_protocol::requests::users::login_register::BindSecret; use iggy_binary_protocol::codes::*; use iggy_binary_protocol::consensus::{ Command, EvictionHeader, EvictionReason, HEADER_SIZE, Operation, ReplyHeader, RequestHeader, @@ -54,10 +52,12 @@ use iggy_binary_protocol::requests::personal_access_tokens::*; use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest; use iggy_binary_protocol::requests::streams::*; use iggy_binary_protocol::requests::system::*; +use iggy_binary_protocol::requests::system::{BindSessionRequest, SessionIdentity}; use iggy_binary_protocol::requests::topics::{ CreateTopicRequest, DeleteTopicRequest, GetTopicRequest, GetTopicsRequest, PurgeTopicRequest, UpdateTopicRequest, }; +use iggy_binary_protocol::requests::users::login_register::BindSecret; use iggy_binary_protocol::requests::users::*; use iggy_binary_protocol::responses::clients::{ ClientDetailsResponse, ClientResponse, ConsumerGroupInfoResponse, GetClientsResponse, @@ -740,7 +740,11 @@ fn main() { "request.bind_session", &BindSessionRequest { version_info: version_info(), - identity: SessionIdentity { client_id: 1, session: 100, metadata_watermark: 120 }, + identity: SessionIdentity { + client_id: 1, + session: 100, + metadata_watermark: 120, + }, bind_secret: BindSecret::new(Box::new([0x5a; 32])), }, ); diff --git a/justfile b/justfile index 7e18398c8..f8c3244a2 100644 --- a/justfile +++ b/justfile @@ -75,14 +75,14 @@ clean-test-containers: # Run Miri (UB detector) on the unsafe-heavy crates that don't pull # tokio/compio. Mirrors the `miri` task in CI. Pinned to the same nightly # as `.github/actions/rust/pre-merge/action.yml` so local runs don't drift -# from CI on the next nightly bump — keep these two dates in sync. +# from CI on the next nightly bump - keep these two dates in sync. # Requires: # -# rustup toolchain install nightly-2026-04-21 --component miri +# rustup toolchain install nightly-2026-10-02 --component miri # miri: MIRIFLAGS="-Zmiri-tree-borrows -Zmiri-strict-provenance" \ - cargo +nightly-2026-04-21 miri test -p iggy_binary_protocol -p consensus + cargo +nightly-2026-10-02 miri test -p iggy_binary_protocol -p consensus server *ARGS: cargo run --bin iggy-server {{ARGS}}
