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 4a43938dd3febeea441af0572f25053161c9a027 Author: Hubert Gruszecki <[email protected]> AuthorDate: Mon Oct 5 15:18:26 2026 +0200 fix(server): prevent retirement stalls and bound protocol versions A retirement deadline could cancel an admitted WAL append and strand subsequent logins. Finish the active retirement before ending the budgeted pass and cover the cancellation window with a regression. Check full protocol version bounds and document stable release policy. Box initialization-test futures rejected by macOS Clippy. --- .../src/requests/users/login_register.rs | 7 +- core/binary_protocol/src/version.rs | 64 ++++++------ core/integration/tests/sdk/protocol_version.rs | 22 ++++ core/server/README.md | 16 ++- core/server/src/consumer_group/liveness.rs | 113 +++++++++++++++++---- core/server/src/partition_helpers.rs | 4 +- foreign/go/internal/vsr/register.go | 5 +- foreign/node/src/wire/vsr/register.ts | 5 +- 8 files changed, 173 insertions(+), 63 deletions(-) diff --git a/core/binary_protocol/src/requests/users/login_register.rs b/core/binary_protocol/src/requests/users/login_register.rs index b03ebd706..7b534311c 100644 --- a/core/binary_protocol/src/requests/users/login_register.rs +++ b/core/binary_protocol/src/requests/users/login_register.rs @@ -85,9 +85,10 @@ impl ExposeSecret<[u8; BIND_SECRET_BYTES]> for BindSecret { /// by non-`vsr` builds against the legacy `iggy-server`) is untouched. /// The server speaks VSR framing only; a non-`vsr` SDK cannot log in to /// the server. Foreign-language SDKs (C++, C#, Python, Go, Java) adopt this -/// shape, with their own `sdk_name`, when they wire VSR framing. Bump -/// [`crate::version::IGGY_PROTOCOL_VERSION`] on any wire-incompatible -/// change. +/// shape, with their own `sdk_name`, when they wire VSR framing. Maintain +/// [`crate::version::IGGY_PROTOCOL_VERSION`] and its minimum for stable +/// releases according to [`crate::version`]. Intermediate edge builds may +/// share a protocol number without compatible layouts. #[derive(Debug, Clone)] pub struct LoginRegisterRequest { pub version_info: ClientVersionInfo, diff --git a/core/binary_protocol/src/version.rs b/core/binary_protocol/src/version.rs index 0dbcfe669..adf1ba597 100644 --- a/core/binary_protocol/src/version.rs +++ b/core/binary_protocol/src/version.rs @@ -17,8 +17,9 @@ //! Binary protocol versioning. //! -//! The wire version is an explicit packed semver, independent of crate -//! releases. It is exchanged during the login/register +//! The wire version is an explicit packed semver, maintained manually for +//! stable Iggy releases and independent of server and SDK package versions. +//! It is exchanged during the login/register //! handshake: clients send [`ClientVersionInfo`] as the body prefix of both //! login-register request shapes, the server gates on //! [`is_protocol_compatible`] before touching credentials and advertises @@ -44,9 +45,8 @@ //! value = major << 20 | minor << 10 | patch //! ``` //! -//! Integer order equals semver order. Under 0.x a minor bump may break -//! the wire, so the gate is minor-scoped. Compatibility across minor versions -//! after 1.0.0 requires changing the minimum-version calculation below. +//! Integer order equals semver order. Every component participates in the +//! compatibility check; a patch increment does not imply wire compatibility. //! //! ## `ClientVersionInfo` body prefix //! @@ -67,11 +67,15 @@ //! //! ## Login gate //! -//! The server accepts a client when its packed version is -//! `>= IGGY_PROTOCOL_VERSION_MIN` and its `major.minor` is `<=` the -//! server's. The explicit minimum excludes older wire layouts, including -//! patches that predate a required field. The upper bound ignores patch; -//! newer patches of the same minor must preserve this layout. +//! The server accepts a client when its packed version is in the inclusive +//! range `IGGY_PROTOCOL_VERSION_MIN..=IGGY_PROTOCOL_VERSION`, including patch. +//! Set both bounds manually for each stable release. Keep them equal unless +//! every released protocol in a wider range is deliberately supported. +//! +//! Compatibility is defined between stable releases. Intermediate development +//! and edge builds may share a protocol number while their layouts change; +//! they require matching client and server builds. Such changes do not require +//! a protocol bump for each commit. //! //! ## Rejection frame //! @@ -95,24 +99,19 @@ const COMPONENT_BITS: u32 = 10; const COMPONENT_MAX: u32 = (1 << COMPONENT_BITS) - 1; const PATCH_MASK: u32 = COMPONENT_MAX; -/// Current binary protocol version, independent of the crate's release version. +/// Current development protocol version, independent of package versions. /// /// Version 0.11.1 requires the session-binding secret in login requests. -/// Intermediate development builds with this version may use incompatible layouts; -/// coordinated deployment requires matching clients and server binaries. +/// Finalize this number and the minimum during stable release preparation. +/// Released servers using protocol 0.11.0 accept every 0.11.x patch, so an +/// incompatible release needs a new minor version for those servers to reject it. pub const IGGY_PROTOCOL_VERSION: u32 = pack_protocol_version(0, 11, 1); /// Oldest protocol version this build accepts at login. /// Version 0.11.0 has no session-binding secret and is incompatible. /// -/// Under 0.x this makes a server minor bump a client flag-day: every -/// prior-minor client is rejected at login until MIN is widened. A rolling -/// upgrade across a minor bump must either widen MIN (when the wire stayed -/// compatible) or accept that old clients fail re-login -- decide per release. -// TODO(hubcio): past 1.0.0 follow strict semver: major bump = incompatible, -// minor/patch = compatible, so MIN derives from the current major instead -// of the current minor. Under 0.x a minor bump may break the wire, so the -// minor-scoped window is correct. +/// Keep this equal to the current protocol unless compatibility with older +/// released protocols has been verified. Decide the range per stable release. pub const IGGY_PROTOCOL_VERSION_MIN: u32 = pack_protocol_version(0, 11, 1); const _: () = assert!(IGGY_PROTOCOL_VERSION_MIN > 0 && IGGY_PROTOCOL_VERSION_MIN <= IGGY_PROTOCOL_VERSION); @@ -120,12 +119,10 @@ const _: () = /// Range check used by the server-side login gate. /// /// Packed component order preserves semver ordering, so plain integer -/// comparisons work. The upper bound ignores patch: newer patches of an -/// accepted minor must preserve its wire layout. +/// comparisons enforce both inclusive bounds, including patch versions. #[must_use] pub const fn is_protocol_compatible(client: u32) -> bool { - client >= IGGY_PROTOCOL_VERSION_MIN - && (client >> COMPONENT_BITS) <= (IGGY_PROTOCOL_VERSION >> COMPONENT_BITS) + client >= IGGY_PROTOCOL_VERSION_MIN && client <= IGGY_PROTOCOL_VERSION } /// Pack semver components: `major << 20 | minor << 10 | patch`. @@ -261,7 +258,7 @@ mod tests { fn login_without_binding_secret_is_outside_the_compatible_window() { assert!(!is_protocol_compatible(pack_protocol_version(0, 11, 0))); assert!(is_protocol_compatible(pack_protocol_version(0, 11, 1))); - assert!(is_protocol_compatible(pack_protocol_version(0, 11, 1023))); + assert!(!is_protocol_compatible(pack_protocol_version(0, 11, 1023))); assert!(!is_protocol_compatible(pack_protocol_version(0, 12, 0))); } @@ -280,15 +277,12 @@ mod tests { fn compatibility_range_boundaries() { assert!(is_protocol_compatible(IGGY_PROTOCOL_VERSION_MIN)); assert!(is_protocol_compatible(IGGY_PROTOCOL_VERSION)); - // Newer patches of the current minor preserve the wire layout. - assert!(is_protocol_compatible(IGGY_PROTOCOL_VERSION | PATCH_MASK)); - // Next minor is outside the window. - assert!(!is_protocol_compatible( - ((IGGY_PROTOCOL_VERSION >> COMPONENT_BITS) + 1) << COMPONENT_BITS - )); - if IGGY_PROTOCOL_VERSION_MIN > 0 { - assert!(!is_protocol_compatible(IGGY_PROTOCOL_VERSION_MIN - 1)); - } + assert!( + !is_protocol_compatible(IGGY_PROTOCOL_VERSION + 1), + "a newer patch must not pass the current release's protocol gate" + ); + assert!(!is_protocol_compatible(IGGY_PROTOCOL_VERSION_MIN - 1)); + assert!(!is_protocol_compatible(u32::MAX)); } #[test] diff --git a/core/integration/tests/sdk/protocol_version.rs b/core/integration/tests/sdk/protocol_version.rs index 773e31d21..ae29a24e2 100644 --- a/core/integration/tests/sdk/protocol_version.rs +++ b/core/integration/tests/sdk/protocol_version.rs @@ -68,6 +68,28 @@ async fn given_incompatible_protocol_version_when_logging_in_should_receive_evic .await; } +#[iggy_harness] +async fn given_out_of_range_version_with_no_credentials_when_logging_in_should_receive_incompatible_protocol( + harness: &TestHarness, +) { + for protocol_version in [IGGY_PROTOCOL_VERSION_MIN - 1, IGGY_PROTOCOL_VERSION + 1] { + let body = ClientVersionInfo { + protocol_version, + sdk_name: WireName::new("rust-sdk").unwrap(), + sdk_version: WireName::new("0.0.1").unwrap(), + } + .to_bytes(); + + assert_login_evicted( + harness, + &body, + EVICTION_REASON_INCOMPATIBLE_PROTOCOL, + (IGGY_PROTOCOL_VERSION, IGGY_PROTOCOL_VERSION_MIN), + ) + .await; + } +} + #[iggy_harness] async fn given_no_version_prefix_when_logging_in_should_receive_eviction(harness: &TestHarness) { // Empty body cannot hold a ClientVersionInfo prefix; the gate rejects it diff --git a/core/server/README.md b/core/server/README.md index 2c01e8320..03dbf8f3a 100644 --- a/core/server/README.md +++ b/core/server/README.md @@ -95,8 +95,20 @@ clients deploy together; mixed versions and rolling upgrades are unsupported. 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. + +This checkout uses development protocol 0.11.1. Finalize its version and minimum +before the next stable release. Released `server-0.9.0` uses protocol 0.11.0 and +accepts every 0.11.x patch. An incompatible release therefore needs a new minor +protocol version, such as 0.12.0, for that released server to reject new clients +before decoding their login. + +The protocol number is independent of server and SDK package versions. Its +current and minimum values are maintained manually for stable releases. The +client login gate checks the full inclusive range, including patch versions. +Both bounds stay equal unless compatibility with older released protocols is +explicitly supported. Edge builds require matching clients and servers, even +when their protocol numbers match. + An unsupported data directory is refused before WAL scanning or file changes. Replacing the binary does not migrate old data. Restore or migrate retained data only through a separately verified procedure; copying a format marker is not diff --git a/core/server/src/consumer_group/liveness.rs b/core/server/src/consumer_group/liveness.rs index d3f78d52b..61ba00395 100644 --- a/core/server/src/consumer_group/liveness.rs +++ b/core/server/src/consumer_group/liveness.rs @@ -20,11 +20,13 @@ use std::cell::RefCell; use std::collections::BTreeMap; +use std::pin::pin; use std::rc::Rc; use std::time::{Duration, Instant}; use bytes::BytesMut; use consensus::{Consensus, MetadataHandle}; +use futures::future::{Either, select}; use iggy_binary_protocol::requests::system::{ MAX_SESSIONS_PER_RETIREMENT, RetireSessionsRequest, SessionIdentity, }; @@ -291,26 +293,30 @@ async fn retire_sessions<B, MJ, S, SB>( 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 { + let mut deadline = pin!(shard.bus.sleep(CONSUMER_SESSION_REPORT_TIMEOUT)); + for _ in 0..MAX_LOGOUTS_PER_PASS { + let before = shard.plane.metadata().client_table.borrow().count(); + let retirement = pin!(retire_session( + shard, + liveness, + stop, + shutdown, + report_interval, + system_path, + )); + match select(retirement, deadline.as_mut()).await { + Either::Left(((), _)) => {} + Either::Right(((), retirement)) => { + // Finalization can own an admitted WAL append. Canceling it + // strands that op and blocks subsequent metadata requests. + retirement.await; break; } } - }) - .await; + if shard.plane.metadata().client_table.borrow().count() >= before { + break; + } + } } #[allow(clippy::too_many_lines)] @@ -1536,6 +1542,79 @@ mod tests { assert_eq!(tracker.retirement.as_ref().unwrap().reporters.len(), 1); } + #[compio::test] + async fn given_retirement_budget_expires_during_finalization_when_logging_in_should_make_progress() + { + const LOGIN_TIMEOUT: Duration = Duration::from_secs(5); + let (_directory, shard) = shard_with_members(2).await; + let metadata = shard.plane.metadata(); + metadata + .client_table + .borrow_mut() + .commit_capacity(3) + .unwrap(); + for client_id in [CLIENT, CLIENT + 1] { + assert!( + metadata + .client_table + .borrow_mut() + .end_session(client_id, 0, SESSION, SESSION) + ); + } + let identity = metadata + .client_table + .borrow() + .ended_sessions() + .find(|identity| identity.client_id == CLIENT) + .unwrap(); + let consensus = metadata.consensus.as_ref().unwrap(); + let revision = metadata + .mux_stm + .streams() + .read(|inner| inner.namespace_revision); + let tracker = RefCell::new(ConsumerGroupLiveness::default()); + { + let mut tracker = tracker.borrow_mut(); + tracker.observe_view(Some(consensus.view())); + tracker.reconcile_retirement(Some(identity), revision); + let progress = tracker.retirement.as_mut().unwrap(); + progress.reported = true; + progress.reporters.insert(consensus.replica()); + } + let (_stop, receiver) = shard::channel(1); + let (_signal, shutdown) = Shutdown::new(); + // Expire at the finalization's first I/O yield, after it claims an op. + shard.bus.instant_timers.set(true); + retire_sessions(&shard, &tracker, &receiver, &shutdown, TIMEOUT, "").await; + shard.bus.instant_timers.set(false); + + metadata.resume_stranded_commits().await; + let login = compio::time::timeout( + LOGIN_TIMEOUT, + metadata.submit_register_in_process(CLIENT + 2, 0, [0x6b; 32]), + ) + .await + .expect("a retirement deadline must not strand a prepare ahead of the next login") + .expect("finalization must release capacity for the new login"); + assert_eq!(login.epoch, SESSION + 2); + assert_eq!(metadata.client_table.borrow().count(), 2); + assert!(metadata.client_table.borrow().get_epoch(CLIENT).is_none()); + assert_eq!( + metadata + .client_table + .borrow() + .ended_sessions() + .next() + .map(|identity| identity.client_id), + Some(CLIENT + 1), + "the expired sweep must defer the next session's finalization" + ); + assert_eq!( + consensus.commit_min(), + consensus.sequencer().current_sequence() + ); + } + #[compio::test] async fn given_failed_partition_when_retiring_should_allow_new_sessions() { let (directory, shard) = shard_with_members(1).await; diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index 91b28d5eb..2909f21ff 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -1343,12 +1343,12 @@ mod tests { #[compio::test] async fn given_uninitialized_singleton_when_booting_should_finish_initialization() { - assert_uninitialized_partition_recovers(1).await; + Box::pin(assert_uninitialized_partition_recovers(1)).await; } #[compio::test] async fn given_uninitialized_replica_when_booting_should_finish_initialization() { - assert_uninitialized_partition_recovers(REPLICAS).await; + Box::pin(assert_uninitialized_partition_recovers(REPLICAS)).await; } async fn assert_uninitialized_partition_recovers(replica_count: u8) { diff --git a/foreign/go/internal/vsr/register.go b/foreign/go/internal/vsr/register.go index b4f8ae706..922eeccda 100644 --- a/foreign/go/internal/vsr/register.go +++ b/foreign/go/internal/vsr/register.go @@ -25,8 +25,9 @@ import ( ) // ProtocolVersion is the packed semver of the wire contract this codec -// implements, 0.11.1 per core/binary_protocol/src/version.rs. Bump it together -// with the Rust constant on any wire-incompatible change. +// implements, 0.11.1 per core/binary_protocol/src/version.rs. Update it with +// the Rust constant during stable release preparation. Edge builds sharing +// this number are not guaranteed to have compatible layouts. var ProtocolVersion = packProtocolVersion(0, 11, 1) // packProtocolVersion packs a semver into the ten-bits-per-field layout the diff --git a/foreign/node/src/wire/vsr/register.ts b/foreign/node/src/wire/vsr/register.ts index 433c29d50..afcab8620 100644 --- a/foreign/node/src/wire/vsr/register.ts +++ b/foreign/node/src/wire/vsr/register.ts @@ -30,8 +30,9 @@ export const SESSION_IDENTITY_BYTES = 32; /** * Packed protocol semver of the wire contract this port implements, - * `pack(0, 11, 1)` per `core/binary_protocol/src/version.rs`. Bump together - * with the Rust `IGGY_PROTOCOL_VERSION` on any wire-incompatible change. + * `pack(0, 11, 1)` per `core/binary_protocol/src/version.rs`. Update with the + * Rust constant during stable release preparation. Edge builds sharing this + * number are not guaranteed to have compatible layouts. */ export const IGGY_PROTOCOL_VERSION = (0 << 20) | (11 << 10) | 1;
