feat(server)!: auth-gate cluster metadata, forward register to primary (#3872)
Unauthenticated clients could read the full cluster roster through
the binary GetClusterMetadata pre-auth carve-out, leaking private
network topology. Plain gating would strand any client that dialed
a backup: Register commits only on the primary, and without the
pre-auth read there is no leader discovery left, so login replays
transient denies until the read budget dies.
Gate the read on every roster shape (PING stays the only pre-auth
code) and let the dialed backup complete the login instead:
credentials are verified locally against replicated user state,
only the verified identity (client id, user id) crosses the
replica interconnect as a sealed ForwardRegister frame, the
primary commits the Register, and the backup binds the session
and replies on its own connection. No credentials travel on the
wire, no reply routing lands on the primary, and the HTTP
forward's fail-closed semantics (an unreplicated PAT is denied at
the backup) carry over; a forward that never left the backup maps
to the retryable class, not a client-side 503. An unbound reader
is denied with a plain Unauthenticated Reply on the status
channel, never a session-terminal Eviction, so a foreign SDK's
stray probe cannot tear down the connection its login is about to
use. The Rust and Node SDKs drop their pre-login leader probe;
the authenticated post-login redirect is the sole leader
settlement, which Node gains here with a short poll through an
election. The C++ e2e expectation of the pre-auth read inverts to
match.
Kill-primary testing of the forward exposed a pre-existing
view-start bug: the metadata parked-view arm dispatched
RebuildPipeline behind the superblock persist, so a login admitted
during that fsync minted an op into the still-empty pipeline and
the deferred rebuild panicked the pump ("sequence must be
sequential"), leaving a deaf primary holding its quorum slot while
the survivor spun elections forever. Dispatch the local actions
before the persist, as every sibling site already does. The drain
path now surfaces the pump's panic verdict instead of claiming a
clean exit, the harness fails any test whose server wrote a panic
to stderr, a simulator test pins the interleaving
deterministically, and forward failures no longer masquerade as a
view-change cancel.diff --git a/Cargo.lock b/Cargo.lock
index 54c50c7..7e44d56 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1950,7 +1950,7 @@
[[package]]
name = "bench-dashboard-frontend"
-version = "0.7.1-edge.1"
+version = "0.8.0-edge.1"
dependencies = [
"bench-dashboard-shared",
"bench-report",
@@ -1980,7 +1980,7 @@
[[package]]
name = "bench-report"
-version = "0.3.1-edge.1"
+version = "0.4.0-edge.1"
dependencies = [
"charming",
"colored",
@@ -6623,7 +6623,7 @@
[[package]]
name = "iggy"
-version = "0.11.0-edge.1"
+version = "0.11.0-edge.2"
dependencies = [
"async-broadcast",
"async-dropper",
@@ -6657,7 +6657,7 @@
[[package]]
name = "iggy-bench"
-version = "0.6.0-edge.1"
+version = "0.6.0-edge.2"
dependencies = [
"async-trait",
"bench-report",
@@ -6686,7 +6686,7 @@
[[package]]
name = "iggy-bench-dashboard-server"
-version = "0.7.1-edge.1"
+version = "0.8.0-edge.1"
dependencies = [
"actix-cors",
"actix-files",
@@ -6714,7 +6714,7 @@
[[package]]
name = "iggy-cli"
-version = "0.14.0-edge.1"
+version = "0.14.0-edge.2"
dependencies = [
"anyhow",
"apple-native-keyring-store",
@@ -6748,7 +6748,7 @@
[[package]]
name = "iggy-connectors"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"axum",
@@ -6800,7 +6800,7 @@
[[package]]
name = "iggy-mcp"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"axum",
"axum-server",
@@ -6834,7 +6834,7 @@
[[package]]
name = "iggy_binary_protocol"
-version = "0.11.0-edge.1"
+version = "0.11.0-edge.2"
dependencies = [
"aligned-vec",
"bytemuck",
@@ -6847,7 +6847,7 @@
[[package]]
name = "iggy_common"
-version = "0.11.0-edge.1"
+version = "0.11.0-edge.2"
dependencies = [
"aes-gcm",
"async-broadcast",
@@ -6883,7 +6883,7 @@
[[package]]
name = "iggy_connector_clickhouse_sink"
-version = "0.2.0-edge.1"
+version = "0.2.0-edge.2"
dependencies = [
"async-trait",
"bytes",
@@ -6901,7 +6901,7 @@
[[package]]
name = "iggy_connector_delta_sink"
-version = "0.2.0-edge.1"
+version = "0.2.0-edge.2"
dependencies = [
"async-trait",
"chrono",
@@ -6918,7 +6918,7 @@
[[package]]
name = "iggy_connector_doris_sink"
-version = "0.2.0-edge.1"
+version = "0.2.0-edge.2"
dependencies = [
"async-trait",
"base64",
@@ -6938,7 +6938,7 @@
[[package]]
name = "iggy_connector_elasticsearch_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"base64",
@@ -6956,7 +6956,7 @@
[[package]]
name = "iggy_connector_elasticsearch_source"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"dashmap",
@@ -6974,7 +6974,7 @@
[[package]]
name = "iggy_connector_http_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"base64",
@@ -6996,7 +6996,7 @@
[[package]]
name = "iggy_connector_iceberg_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"arrow-json 57.3.1",
"async-trait",
@@ -7015,7 +7015,7 @@
[[package]]
name = "iggy_connector_influxdb_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"axum",
@@ -7036,7 +7036,7 @@
[[package]]
name = "iggy_connector_influxdb_source"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"ahash 0.8.12",
"async-trait",
@@ -7062,7 +7062,7 @@
[[package]]
name = "iggy_connector_meilisearch_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"base64",
@@ -7080,7 +7080,7 @@
[[package]]
name = "iggy_connector_mongodb_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"humantime",
@@ -7096,7 +7096,7 @@
[[package]]
name = "iggy_connector_postgres_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"dashmap",
@@ -7115,7 +7115,7 @@
[[package]]
name = "iggy_connector_postgres_source"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"base64",
@@ -7137,7 +7137,7 @@
[[package]]
name = "iggy_connector_quickwit_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"dashmap",
@@ -7151,7 +7151,7 @@
[[package]]
name = "iggy_connector_random_source"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"dashmap",
@@ -7168,7 +7168,7 @@
[[package]]
name = "iggy_connector_s3_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"base64",
@@ -7189,7 +7189,7 @@
[[package]]
name = "iggy_connector_sdk"
-version = "0.3.1-edge.1"
+version = "0.4.0-edge.1"
dependencies = [
"anyhow",
"apache-avro",
@@ -7225,7 +7225,7 @@
[[package]]
name = "iggy_connector_stdout_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"dashmap",
@@ -7237,7 +7237,7 @@
[[package]]
name = "iggy_connector_surrealdb_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
dependencies = [
"async-trait",
"base64",
@@ -11839,7 +11839,7 @@
[[package]]
name = "server"
-version = "0.9.0-edge.2"
+version = "0.9.0-edge.3"
dependencies = [
"ahash 0.8.12",
"argon2",
diff --git a/Cargo.toml b/Cargo.toml
index 4e4d18e..31e37ba 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -199,11 +199,11 @@
iceberg = "0.9.1"
iceberg-catalog-rest = "0.9.1"
iceberg-storage-opendal = "0.9.1"
-iggy = { path = "core/sdk", version = "0.11.0-edge.1" }
-iggy-cli = { path = "core/cli", version = "0.14.0-edge.1" }
-iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.1" }
-iggy_common = { path = "core/common", version = "0.11.0-edge.1" }
-iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.3.1-edge.1" }
+iggy = { path = "core/sdk", version = "0.11.0-edge.2" }
+iggy-cli = { path = "core/cli", version = "0.14.0-edge.2" }
+iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.2" }
+iggy_common = { path = "core/common", version = "0.11.0-edge.2" }
+iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.4.0-edge.1" }
indexmap = "2.14.0"
integration = { path = "core/integration" }
ipnet = "2.12.0"
diff --git a/bdd/python/uv.lock b/bdd/python/uv.lock
index b88cd86..21a0ddc 100644
--- a/bdd/python/uv.lock
+++ b/bdd/python/uv.lock
@@ -8,7 +8,7 @@
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev1"
+version = "0.9.0.dev2"
source = { directory = "../../foreign/python" }
[package.metadata]
diff --git a/core/ai/mcp/Cargo.toml b/core/ai/mcp/Cargo.toml
index 9cc785d..e58068e 100644
--- a/core/ai/mcp/Cargo.toml
+++ b/core/ai/mcp/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-mcp"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "MCP Server for Iggy message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/bench/Cargo.toml b/core/bench/Cargo.toml
index 9842a44..1ac1b9b 100644
--- a/core/bench/Cargo.toml
+++ b/core/bench/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-bench"
-version = "0.6.0-edge.1"
+version = "0.6.0-edge.2"
edition = "2024"
license = "Apache-2.0"
repository = "https://github.com/apache/iggy"
diff --git a/core/bench/dashboard/frontend/Cargo.toml b/core/bench/dashboard/frontend/Cargo.toml
index 1e0675d..7c9be35 100644
--- a/core/bench/dashboard/frontend/Cargo.toml
+++ b/core/bench/dashboard/frontend/Cargo.toml
@@ -18,7 +18,7 @@
[package]
name = "bench-dashboard-frontend"
license = "Apache-2.0"
-version = "0.7.1-edge.1"
+version = "0.8.0-edge.1"
edition = "2024"
publish = false
diff --git a/core/bench/dashboard/server/Cargo.toml b/core/bench/dashboard/server/Cargo.toml
index fd2d51d..dc61bf6 100644
--- a/core/bench/dashboard/server/Cargo.toml
+++ b/core/bench/dashboard/server/Cargo.toml
@@ -18,7 +18,7 @@
[package]
name = "iggy-bench-dashboard-server"
license = "Apache-2.0"
-version = "0.7.1-edge.1"
+version = "0.8.0-edge.1"
edition = "2024"
publish = false
diff --git a/core/bench/report/Cargo.toml b/core/bench/report/Cargo.toml
index ae4e933..a4eb23e 100644
--- a/core/bench/report/Cargo.toml
+++ b/core/bench/report/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "bench-report"
-version = "0.3.1-edge.1"
+version = "0.4.0-edge.1"
edition = "2024"
description = "Benchmark report and chart generation library for iggy-bench binary and iggy-benchmarks-dashboard web app"
license = "Apache-2.0"
diff --git a/core/binary_protocol/Cargo.toml b/core/binary_protocol/Cargo.toml
index 4fcdab1..0d8b74d 100644
--- a/core/binary_protocol/Cargo.toml
+++ b/core/binary_protocol/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_binary_protocol"
-version = "0.11.0-edge.1"
+version = "0.11.0-edge.2"
description = "Wire protocol types and codec for the Iggy binary protocol. Shared between server and SDK."
edition = "2024"
rust-version.workspace = true
diff --git a/core/binary_protocol/src/consensus/command.rs b/core/binary_protocol/src/consensus/command.rs
index 8e60655..f358c3c 100644
--- a/core/binary_protocol/src/consensus/command.rs
+++ b/core/binary_protocol/src/consensus/command.rs
@@ -70,6 +70,19 @@
StateTransferTarget = 23,
RequestStateChunk = 24,
StateChunk = 25,
+
+ // Register forwarding: a client that dialed a backup authenticates
+ // there, and only the consensus proposal travels to the primary. The
+ // backup forwards the verified identity and parks the login on the
+ // matching `ForwardRegisterResult`.
+ ForwardRegister = 26,
+ ForwardRegisterResult = 27,
+
+ // Logout forwarding: a session bound on a backup asks the primary to
+ // commit its replicated teardown, then the backup answers the client on
+ // the connection it owns.
+ ForwardLogout = 28,
+ ForwardLogoutResult = 29,
}
// SAFETY: Command2 is #[repr(u8)] with no padding bytes.
@@ -80,7 +93,7 @@
type Bits = u8;
fn is_valid_bit_pattern(bits: &u8) -> bool {
- *bits <= Self::StateChunk as u8
+ *bits <= Self::ForwardLogoutResult as u8
}
}
@@ -101,8 +114,8 @@
#[test]
fn replica_auth_commands_are_valid_bit_patterns() {
- // Locks the is_valid_bit_pattern bump: 14..=25 parse, 26 still rejects.
- for command in 14u8..=25 {
+ // Locks the is_valid_bit_pattern bump: 14..=29 parse, 30 still rejects.
+ for command in 14u8..=29 {
let mut buf: AVec<u8, ConstAlign<16>> = AVec::new(16);
buf.resize(256, 0);
buf[60] = command;
@@ -110,7 +123,7 @@
}
let mut buf: AVec<u8, ConstAlign<16>> = AVec::new(16);
buf.resize(256, 0);
- buf[60] = 26;
+ buf[60] = 30;
assert!(bytemuck::checked::try_from_bytes::<GenericHeader>(&buf).is_err());
}
}
diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs
index 4f8fffe..9dd4d7d 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -32,6 +32,12 @@
//! mixed pair is dropped, so no view change reaches a quorum. Nothing enforces this,
//! because there is nothing left to enforce it with; this note is the declaration.
//!
+//! The forwarding commands at discriminants 26 through 29 are same-release
+//! replica messages. A build whose command bit-pattern predates them drops the
+//! frames as unparsable and never sends a result, so backup-dialed session
+//! operations wait for their forward timeout in a mixed fleet. These commands are
+//! another reason the release cannot be rolled node by node.
+//!
//! `Prepare`, `Request`, `Reply`, and `Eviction` are unaffected. Prepares keep
//! `checksum` as their view-independent identity, and the three client-facing
//! headers are sealed on neither side, so SDKs are untouched.
@@ -2234,13 +2240,464 @@
}
}
+// ForwardRegisterHeader - backup shard 0 -> primary shard 0
+
+/// A backup relays a login it has already authenticated.
+///
+/// Credential verification runs on the node the client dialed, against the
+/// replicated users table, so neither the client's frame nor its credentials
+/// travel: this header carries the VERIFIED identity and nothing else. The
+/// backup keeps the session bind, the reply build, and the connection.
+///
+/// The trust boundary is the replica interconnect's network placement: by
+/// default the replica port trusts any peer that reaches it (the PSK
+/// handshake and TLS ship disabled and are what upgrade the boundary), and
+/// the seal is an unkeyed integrity check, not a MAC. Trusting `user_id`
+/// here adds no capability the port did not already expose, since that same
+/// peer could inject a `Request` + `Register` directly. Clients cannot
+/// reach this command, because every client frame is typed through
+/// [`RequestHeader`], whose `validate` rejects any command but
+/// [`Command2::Request`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct ForwardRegisterHeader {
+ pub checksum: u128,
+ pub checksum_body: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub view: u32,
+ pub release: u32,
+ pub command: Command2,
+ pub replica: u8,
+ pub reserved_frame: [u8; 66],
+
+ /// The login frame's consensus client id, proposed verbatim by the primary.
+ pub client: u128,
+ /// Correlation minted by the backup, echoed verbatim in the result. Unique
+ /// per in-flight forward on the originating node, across its restarts as
+ /// well: an answer that outlives the login it was minted for must not match
+ /// the one holding that nonce next.
+ pub nonce: u128,
+ /// The acting user the backup authenticated. The trust payload: the primary
+ /// proposes the register under this id without re-verifying credentials.
+ pub user_id: u32,
+ pub reserved: [u8; 92],
+}
+const _: () = {
+ assert!(size_of::<ForwardRegisterHeader>() == HEADER_SIZE);
+ assert!(
+ offset_of!(ForwardRegisterHeader, client)
+ == offset_of!(ForwardRegisterHeader, reserved_frame) + size_of::<[u8; 66]>()
+ );
+ assert!(offset_of!(ForwardRegisterHeader, nonce) == 144);
+ assert!(offset_of!(ForwardRegisterHeader, user_id) == 160);
+ assert!(offset_of!(ForwardRegisterHeader, reserved) + size_of::<[u8; 92]>() == HEADER_SIZE);
+};
+
+impl ConsensusHeader for ForwardRegisterHeader {
+ // The seal covers `user_id`, which is the whole reason this frame is
+ // believed: a flipped bit there would commit a register under another user.
+ const FRAME_SEALED: bool = true;
+
+ const COMMAND: Command2 = Command2::ForwardRegister;
+
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+
+ fn set_checksum(&mut self, checksum: u128) {
+ self.checksum = checksum;
+ }
+ fn operation(&self) -> Operation {
+ Operation::Reserved
+ }
+ fn command(&self) -> Command2 {
+ self.command
+ }
+ fn size(&self) -> u32 {
+ self.size
+ }
+
+ fn validate(&self) -> Result<(), ConsensusError> {
+ if self.command != Command2::ForwardRegister {
+ return Err(ConsensusError::InvalidCommand {
+ expected: Command2::ForwardRegister,
+ found: self.command,
+ });
+ }
+ validate_forward_register_frame(self.size, self.client, self.nonce, &self.reserved)
+ }
+}
+
+// ForwardRegisterResultHeader - primary shard 0 -> backup shard 0
+
+/// The primary's verdict on a [`ForwardRegisterHeader`].
+///
+/// Header-only: the backup owns the client connection and builds the wire
+/// reply itself, so only the committed bind (or the refusal) crosses the
+/// interconnect.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct ForwardRegisterResultHeader {
+ pub checksum: u128,
+ pub checksum_body: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub view: u32,
+ pub release: u32,
+ pub command: Command2,
+ pub replica: u8,
+ pub reserved_frame: [u8; 66],
+
+ /// Echo of [`ForwardRegisterHeader::nonce`]; with `client`, the backup's
+ /// routing key.
+ pub nonce: u128,
+ /// Echo of [`ForwardRegisterHeader::client`]. Half of the routing key, not a
+ /// diagnostic: the backup drops an answer whose client disagrees with the
+ /// login it parked under `nonce`.
+ pub client: u128,
+ /// The committed register's op number, which fences the session. Zero
+ /// unless `outcome` is [`ForwardRegisterOutcome::Ok`].
+ pub epoch: u64,
+ /// The client-table entry's highest committed request number, for a caller
+ /// that must resume numbering. Zero unless `outcome` is
+ /// [`ForwardRegisterOutcome::Ok`].
+ pub watermark: u64,
+ pub reserved: [u8; 79],
+ pub outcome: ForwardRegisterOutcome,
+}
+const _: () = {
+ assert!(size_of::<ForwardRegisterResultHeader>() == HEADER_SIZE);
+ assert!(
+ offset_of!(ForwardRegisterResultHeader, nonce)
+ == offset_of!(ForwardRegisterResultHeader, reserved_frame) + size_of::<[u8; 66]>()
+ );
+ assert!(offset_of!(ForwardRegisterResultHeader, client) == 144);
+ assert!(offset_of!(ForwardRegisterResultHeader, epoch) == 160);
+ assert!(offset_of!(ForwardRegisterResultHeader, watermark) == 168);
+ assert!(
+ offset_of!(ForwardRegisterResultHeader, outcome) + size_of::<ForwardRegisterOutcome>()
+ == HEADER_SIZE
+ );
+};
+
+/// Wire verdict on a forwarded register.
+///
+/// Mirrors the server-side submit error one-for-one so the backup can surface
+/// the primary's exact answer: every variant but
+/// [`Self::ClientIdOwnedByAnotherUser`] is transient and replayable.
+///
+/// **Wire-version pinned**, like [`EvictionReason`]: reordering or reusing a
+/// discriminant silently reinterprets a live cluster's frames.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, NoUninit, CheckedBitPattern)]
+#[repr(u8)]
+pub enum ForwardRegisterOutcome {
+ /// Committed; `epoch` and `watermark` carry the bind.
+ Ok = 0,
+ NotPrimary = 1,
+ /// Reserved: the register submit parks instead of bouncing when the
+ /// primary is not caught up, so nothing produces this today. Wire-pinned,
+ /// so it must keep its discriminant either way.
+ NotCaughtUp = 2,
+ PipelineFull = 3,
+ InProgress = 4,
+ Canceled = 5,
+ /// Terminal: the presented client id belongs to another user.
+ ClientIdOwnedByAnotherUser = 6,
+}
+
+impl ConsensusHeader for ForwardRegisterResultHeader {
+ const FRAME_SEALED: bool = true;
+
+ const COMMAND: Command2 = Command2::ForwardRegisterResult;
+
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+
+ fn set_checksum(&mut self, checksum: u128) {
+ self.checksum = checksum;
+ }
+ fn operation(&self) -> Operation {
+ Operation::Reserved
+ }
+ fn command(&self) -> Command2 {
+ self.command
+ }
+ fn size(&self) -> u32 {
+ self.size
+ }
+
+ fn validate(&self) -> Result<(), ConsensusError> {
+ if self.command != Command2::ForwardRegisterResult {
+ return Err(ConsensusError::InvalidCommand {
+ expected: Command2::ForwardRegisterResult,
+ found: self.command,
+ });
+ }
+ validate_forward_register_frame(self.size, self.client, self.nonce, &self.reserved)?;
+ if self.outcome != ForwardRegisterOutcome::Ok && (self.epoch != 0 || self.watermark != 0) {
+ return Err(ConsensusError::InvalidField(
+ "forward register result bind must be zero on failure".to_string(),
+ ));
+ }
+ Ok(())
+ }
+}
+
+/// Shared field rules of the two register-forwarding headers.
+///
+/// Reserved bytes are strict-zero rather than ignored: this pair only ever
+/// travels between replicas of one release (see the wire-compatibility note at
+/// the top of this module), so there is no forward-compatibility to preserve
+/// and a nonzero byte can only mean a builder bug or a mangled frame.
+fn validate_forward_register_frame(
+ size: u32,
+ client: u128,
+ nonce: u128,
+ reserved: &[u8],
+) -> Result<(), ConsensusError> {
+ validate_forward_frame(size, client, nonce, reserved)
+}
+
+// ForwardLogoutHeader - backup shard 0 -> primary shard 0
+
+/// A backup asks the metadata primary to tear down a session it owns.
+///
+/// The client-facing Logout request stays on the backup. Only the replicated
+/// session identity and request number cross the replica interconnect, and the
+/// primary's epoch guard makes a delayed forward harmless after a rebind.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct ForwardLogoutHeader {
+ pub checksum: u128,
+ pub checksum_body: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub view: u32,
+ pub release: u32,
+ pub command: Command2,
+ pub replica: u8,
+ pub reserved_frame: [u8; 66],
+
+ /// The consensus client id whose table entry should be removed.
+ pub client: u128,
+ /// Correlation minted by the backup and echoed in the result.
+ pub nonce: u128,
+ /// The exact register epoch being logged out.
+ pub session: u64,
+ /// The client's request number, or the server's synthetic disconnect id.
+ pub request: u64,
+ pub reserved: [u8; 80],
+}
+const _: () = {
+ assert!(size_of::<ForwardLogoutHeader>() == HEADER_SIZE);
+ assert!(
+ offset_of!(ForwardLogoutHeader, client)
+ == offset_of!(ForwardLogoutHeader, reserved_frame) + size_of::<[u8; 66]>()
+ );
+ assert!(offset_of!(ForwardLogoutHeader, nonce) == 144);
+ assert!(offset_of!(ForwardLogoutHeader, session) == 160);
+ assert!(offset_of!(ForwardLogoutHeader, request) == 168);
+ assert!(offset_of!(ForwardLogoutHeader, reserved) + size_of::<[u8; 80]>() == HEADER_SIZE);
+};
+
+impl ConsensusHeader for ForwardLogoutHeader {
+ const FRAME_SEALED: bool = true;
+ const COMMAND: Command2 = Command2::ForwardLogout;
+
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+
+ fn set_checksum(&mut self, checksum: u128) {
+ self.checksum = checksum;
+ }
+
+ fn operation(&self) -> Operation {
+ Operation::Reserved
+ }
+
+ fn command(&self) -> Command2 {
+ self.command
+ }
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+
+ fn validate(&self) -> Result<(), ConsensusError> {
+ if self.command != Command2::ForwardLogout {
+ return Err(ConsensusError::InvalidCommand {
+ expected: Command2::ForwardLogout,
+ found: self.command,
+ });
+ }
+ validate_forward_logout_frame(
+ self.size,
+ self.client,
+ self.nonce,
+ self.session,
+ self.request,
+ &self.reserved,
+ )
+ }
+}
+
+// ForwardLogoutResultHeader - primary shard 0 -> backup shard 0
+
+/// The metadata primary's verdict on a [`ForwardLogoutHeader`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct ForwardLogoutResultHeader {
+ pub checksum: u128,
+ pub checksum_body: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub view: u32,
+ pub release: u32,
+ pub command: Command2,
+ pub replica: u8,
+ pub reserved_frame: [u8; 66],
+
+ /// Echo of [`ForwardLogoutHeader::nonce`].
+ pub nonce: u128,
+ /// Echo of [`ForwardLogoutHeader::client`].
+ pub client: u128,
+ /// Committed Logout op, or the current commit for an idempotent no-op.
+ /// Zero unless `outcome` is [`ForwardLogoutOutcome::Ok`].
+ pub commit: u64,
+ pub reserved: [u8; 87],
+ pub outcome: ForwardLogoutOutcome,
+}
+const _: () = {
+ assert!(size_of::<ForwardLogoutResultHeader>() == HEADER_SIZE);
+ assert!(
+ offset_of!(ForwardLogoutResultHeader, nonce)
+ == offset_of!(ForwardLogoutResultHeader, reserved_frame) + size_of::<[u8; 66]>()
+ );
+ assert!(offset_of!(ForwardLogoutResultHeader, client) == 144);
+ assert!(offset_of!(ForwardLogoutResultHeader, commit) == 160);
+ assert!(
+ offset_of!(ForwardLogoutResultHeader, outcome) + size_of::<ForwardLogoutOutcome>()
+ == HEADER_SIZE
+ );
+};
+
+/// Wire verdict on a forwarded logout.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, NoUninit, CheckedBitPattern)]
+#[repr(u8)]
+pub enum ForwardLogoutOutcome {
+ Ok = 0,
+ NotPrimary = 1,
+ PipelineFull = 2,
+ InProgress = 3,
+ Canceled = 4,
+}
+
+impl ConsensusHeader for ForwardLogoutResultHeader {
+ const FRAME_SEALED: bool = true;
+ const COMMAND: Command2 = Command2::ForwardLogoutResult;
+
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+
+ fn set_checksum(&mut self, checksum: u128) {
+ self.checksum = checksum;
+ }
+
+ fn operation(&self) -> Operation {
+ Operation::Reserved
+ }
+
+ fn command(&self) -> Command2 {
+ self.command
+ }
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+
+ fn validate(&self) -> Result<(), ConsensusError> {
+ if self.command != Command2::ForwardLogoutResult {
+ return Err(ConsensusError::InvalidCommand {
+ expected: Command2::ForwardLogoutResult,
+ found: self.command,
+ });
+ }
+ validate_forward_frame(self.size, self.client, self.nonce, &self.reserved)?;
+ if self.outcome != ForwardLogoutOutcome::Ok && self.commit != 0 {
+ return Err(ConsensusError::InvalidField(
+ "forward logout result commit must be zero on failure".to_string(),
+ ));
+ }
+ Ok(())
+ }
+}
+
+fn validate_forward_logout_frame(
+ size: u32,
+ client: u128,
+ nonce: u128,
+ session: u64,
+ request: u64,
+ reserved: &[u8],
+) -> Result<(), ConsensusError> {
+ validate_forward_frame(size, client, nonce, reserved)?;
+ if session == 0 {
+ return Err(ConsensusError::InvalidField(
+ "forward logout session must be non-zero".to_string(),
+ ));
+ }
+ if request == 0 {
+ return Err(ConsensusError::InvalidField(
+ "forward logout request must be non-zero".to_string(),
+ ));
+ }
+ Ok(())
+}
+
+#[allow(clippy::cast_possible_truncation)]
+fn validate_forward_frame(
+ size: u32,
+ client: u128,
+ nonce: u128,
+ reserved: &[u8],
+) -> Result<(), ConsensusError> {
+ if size as usize != HEADER_SIZE {
+ return Err(ConsensusError::InvalidSize {
+ expected: HEADER_SIZE as u32,
+ found: size,
+ });
+ }
+ if client == 0 {
+ return Err(ConsensusError::InvalidField(
+ "forward client must be non-zero".to_string(),
+ ));
+ }
+ if nonce == 0 {
+ return Err(ConsensusError::InvalidField(
+ "forward nonce must be non-zero".to_string(),
+ ));
+ }
+ if reserved.iter().any(|&byte| byte != 0) {
+ return Err(ConsensusError::InvalidField(
+ "forward reserved bytes must be zero".to_string(),
+ ));
+ }
+ Ok(())
+}
+
// Tests
#[cfg(test)]
mod tests {
use super::{
Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader,
- EvictionHeader, EvictionReason, GenericHeader, HEADER_SIZE, Operation, PrepareHeader,
+ EvictionHeader, EvictionReason, ForwardLogoutHeader, ForwardLogoutOutcome,
+ ForwardLogoutResultHeader, ForwardRegisterHeader, ForwardRegisterOutcome,
+ ForwardRegisterResultHeader, GenericHeader, HEADER_SIZE, Operation, PrepareHeader,
PrepareOkHeader, RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader, RequestHeader,
RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
RequestStateTransferHeader, RoutedRequestHeader, StartViewChangeHeader, StartViewHeader,
@@ -2316,6 +2773,10 @@
StateTransferTargetHeader,
RequestStateChunkHeader,
StateChunkHeader,
+ ForwardRegisterHeader,
+ ForwardRegisterResultHeader,
+ ForwardLogoutHeader,
+ ForwardLogoutResultHeader,
);
}
@@ -2683,4 +3144,318 @@
fn eviction_header_is_256_bytes() {
assert_eq!(size_of::<EvictionHeader>(), 256);
}
+
+ /// A sealed, well-formed `ForwardRegister` ready to be mutated per test.
+ fn forward_register() -> ForwardRegisterHeader {
+ let mut header = ForwardRegisterHeader {
+ checksum: 0,
+ checksum_body: 0,
+ cluster: 7,
+ size: u32::try_from(HEADER_SIZE).expect("HEADER_SIZE fits u32"),
+ view: 3,
+ release: 0,
+ command: Command2::ForwardRegister,
+ replica: 2,
+ reserved_frame: [0; 66],
+ client: 0xCAFE,
+ nonce: 0xF00D,
+ user_id: 41,
+ reserved: [0; 92],
+ };
+ header.seal();
+ header
+ }
+
+ /// A sealed, well-formed `ForwardRegisterResult` ready to be mutated.
+ fn forward_register_result(outcome: ForwardRegisterOutcome) -> ForwardRegisterResultHeader {
+ let mut header = ForwardRegisterResultHeader {
+ checksum: 0,
+ checksum_body: 0,
+ cluster: 7,
+ size: u32::try_from(HEADER_SIZE).expect("HEADER_SIZE fits u32"),
+ view: 3,
+ release: 0,
+ command: Command2::ForwardRegisterResult,
+ replica: 0,
+ reserved_frame: [0; 66],
+ nonce: 0xF00D,
+ client: 0xCAFE,
+ epoch: u64::from(outcome == ForwardRegisterOutcome::Ok) * 91,
+ watermark: u64::from(outcome == ForwardRegisterOutcome::Ok) * 12,
+ reserved: [0; 79],
+ outcome,
+ };
+ header.seal();
+ header
+ }
+
+ #[test]
+ fn forward_register_round_trips_through_generic_bytes() {
+ for (command, bytes) in [
+ (
+ Command2::ForwardRegister,
+ bytemuck::bytes_of(&forward_register()).to_vec(),
+ ),
+ (
+ Command2::ForwardRegisterResult,
+ bytemuck::bytes_of(&forward_register_result(ForwardRegisterOutcome::Ok)).to_vec(),
+ ),
+ ] {
+ // 16-byte alignment: `Vec<u8>` requests align=1 and fails Miri.
+ let mut buf = aligned_zeroed(HEADER_SIZE);
+ buf.copy_from_slice(&bytes);
+ let generic = bytemuck::checked::try_from_bytes::<GenericHeader>(&buf)
+ .expect("a forward-register frame is a valid generic header");
+ assert_eq!(generic.command, command);
+ assert_eq!(generic.size as usize, HEADER_SIZE);
+ }
+
+ let buf = {
+ let mut buf = aligned_zeroed(HEADER_SIZE);
+ buf.copy_from_slice(bytemuck::bytes_of(&forward_register()));
+ buf
+ };
+ let typed = bytemuck::checked::try_from_bytes::<ForwardRegisterHeader>(&buf)
+ .expect("round-trips into its own type");
+ assert_eq!(typed.verify_frame(), Ok(()));
+ assert_eq!(typed.validate(), Ok(()));
+ assert_eq!(typed.user_id, 41);
+ assert_eq!(typed.nonce, 0xF00D);
+
+ let buf = {
+ let mut buf = aligned_zeroed(HEADER_SIZE);
+ buf.copy_from_slice(bytemuck::bytes_of(&forward_register_result(
+ ForwardRegisterOutcome::ClientIdOwnedByAnotherUser,
+ )));
+ buf
+ };
+ let typed = bytemuck::checked::try_from_bytes::<ForwardRegisterResultHeader>(&buf)
+ .expect("round-trips into its own type");
+ assert_eq!(typed.verify_frame(), Ok(()));
+ assert_eq!(typed.validate(), Ok(()));
+ assert_eq!(
+ typed.outcome,
+ ForwardRegisterOutcome::ClientIdOwnedByAnotherUser
+ );
+ }
+
+ // The seal is the whole trust story for `user_id`: the primary proposes a
+ // register under it without re-verifying credentials, so a flipped bit
+ // must not reach the proposal.
+ #[test]
+ fn forward_register_tampered_user_id_fails_the_seal() {
+ const USER_ID_OFFSET: usize = std::mem::offset_of!(ForwardRegisterHeader, user_id);
+
+ assert!(matches!(
+ tamper(forward_register(), USER_ID_OFFSET),
+ Err(ConsensusError::FrameChecksumMismatch { .. })
+ ));
+ }
+
+ #[test]
+ fn forward_register_validate_rejects_malformed_frames() {
+ let mut zero_client = forward_register();
+ zero_client.client = 0;
+ assert!(matches!(
+ zero_client.validate(),
+ Err(ConsensusError::InvalidField(_))
+ ));
+
+ let mut zero_nonce = forward_register();
+ zero_nonce.nonce = 0;
+ assert!(matches!(
+ zero_nonce.validate(),
+ Err(ConsensusError::InvalidField(_))
+ ));
+
+ let mut dirty_reserved = forward_register();
+ dirty_reserved.reserved[0] = 1;
+ assert!(matches!(
+ dirty_reserved.validate(),
+ Err(ConsensusError::InvalidField(_))
+ ));
+
+ let mut wrong_size = forward_register();
+ wrong_size.size = 512;
+ assert!(matches!(
+ wrong_size.validate(),
+ Err(ConsensusError::InvalidSize { .. })
+ ));
+
+ let mut result_zero_nonce = forward_register_result(ForwardRegisterOutcome::Ok);
+ result_zero_nonce.nonce = 0;
+ assert!(matches!(
+ result_zero_nonce.validate(),
+ Err(ConsensusError::InvalidField(_))
+ ));
+
+ let mut failed_with_bind = forward_register_result(ForwardRegisterOutcome::NotPrimary);
+ failed_with_bind.epoch = 91;
+ failed_with_bind.watermark = 12;
+ assert!(matches!(
+ failed_with_bind.validate(),
+ Err(ConsensusError::InvalidField(_))
+ ));
+ failed_with_bind.epoch = 0;
+ failed_with_bind.watermark = 0;
+ assert_eq!(failed_with_bind.validate(), Ok(()));
+ }
+
+ // An unknown outcome is rejected by the bit-pattern check, one layer
+ // before `validate` ever runs.
+ #[test]
+ fn forward_register_result_rejects_unknown_outcome() {
+ const OUTCOME_OFFSET: usize = std::mem::offset_of!(ForwardRegisterResultHeader, outcome);
+
+ let mut buf = aligned_zeroed(HEADER_SIZE);
+ buf.copy_from_slice(bytemuck::bytes_of(&forward_register_result(
+ ForwardRegisterOutcome::Ok,
+ )));
+ buf[OUTCOME_OFFSET] = 7;
+ assert!(bytemuck::checked::try_from_bytes::<ForwardRegisterResultHeader>(&buf).is_err());
+ }
+
+ // Wire-discriminant pin: reordering reinterprets a live cluster's frames.
+ #[test]
+ fn forward_register_outcome_discriminants_pinned() {
+ assert_eq!(ForwardRegisterOutcome::Ok as u8, 0);
+ assert_eq!(ForwardRegisterOutcome::NotPrimary as u8, 1);
+ assert_eq!(ForwardRegisterOutcome::NotCaughtUp as u8, 2);
+ assert_eq!(ForwardRegisterOutcome::PipelineFull as u8, 3);
+ assert_eq!(ForwardRegisterOutcome::InProgress as u8, 4);
+ assert_eq!(ForwardRegisterOutcome::Canceled as u8, 5);
+ assert_eq!(ForwardRegisterOutcome::ClientIdOwnedByAnotherUser as u8, 6);
+ }
+
+ fn forward_logout() -> ForwardLogoutHeader {
+ let mut header = ForwardLogoutHeader {
+ checksum: 0,
+ checksum_body: 0,
+ cluster: 7,
+ size: u32::try_from(HEADER_SIZE).expect("HEADER_SIZE fits u32"),
+ view: 3,
+ release: 0,
+ command: Command2::ForwardLogout,
+ replica: 2,
+ reserved_frame: [0; 66],
+ client: 0xCAFE,
+ nonce: 0xF00D,
+ session: 91,
+ request: 12,
+ reserved: [0; 80],
+ };
+ header.seal();
+ header
+ }
+
+ fn forward_logout_result(outcome: ForwardLogoutOutcome) -> ForwardLogoutResultHeader {
+ let mut header = ForwardLogoutResultHeader {
+ checksum: 0,
+ checksum_body: 0,
+ cluster: 7,
+ size: u32::try_from(HEADER_SIZE).expect("HEADER_SIZE fits u32"),
+ view: 3,
+ release: 0,
+ command: Command2::ForwardLogoutResult,
+ replica: 0,
+ reserved_frame: [0; 66],
+ nonce: 0xF00D,
+ client: 0xCAFE,
+ commit: if outcome == ForwardLogoutOutcome::Ok {
+ 92
+ } else {
+ 0
+ },
+ reserved: [0; 87],
+ outcome,
+ };
+ header.seal();
+ header
+ }
+
+ #[test]
+ fn forward_logout_headers_round_trip_and_validate() {
+ let forward = forward_logout();
+ assert_eq!(forward.verify_frame(), Ok(()));
+ assert_eq!(forward.validate(), Ok(()));
+
+ for outcome in [
+ ForwardLogoutOutcome::Ok,
+ ForwardLogoutOutcome::NotPrimary,
+ ForwardLogoutOutcome::PipelineFull,
+ ForwardLogoutOutcome::InProgress,
+ ForwardLogoutOutcome::Canceled,
+ ] {
+ let result = forward_logout_result(outcome);
+ assert_eq!(result.verify_frame(), Ok(()));
+ assert_eq!(result.validate(), Ok(()));
+ let mut bytes = aligned_zeroed(HEADER_SIZE);
+ bytes.copy_from_slice(bytemuck::bytes_of(&result));
+ let generic = bytemuck::checked::try_from_bytes::<GenericHeader>(&bytes)
+ .expect("forward logout result is a valid generic header");
+ assert_eq!(generic.command, Command2::ForwardLogoutResult);
+ }
+ }
+
+ #[test]
+ fn forward_logout_validate_rejects_malformed_frames() {
+ let mut header = forward_logout();
+ header.client = 0;
+ assert!(matches!(
+ header.validate(),
+ Err(ConsensusError::InvalidField(_))
+ ));
+ header = forward_logout();
+ header.nonce = 0;
+ assert!(matches!(
+ header.validate(),
+ Err(ConsensusError::InvalidField(_))
+ ));
+ header = forward_logout();
+ header.session = 0;
+ assert!(matches!(
+ header.validate(),
+ Err(ConsensusError::InvalidField(_))
+ ));
+ header = forward_logout();
+ header.request = 0;
+ assert!(matches!(
+ header.validate(),
+ Err(ConsensusError::InvalidField(_))
+ ));
+ header = forward_logout();
+ header.reserved[0] = 1;
+ assert!(matches!(
+ header.validate(),
+ Err(ConsensusError::InvalidField(_))
+ ));
+
+ let mut result = forward_logout_result(ForwardLogoutOutcome::NotPrimary);
+ result.commit = 92;
+ assert!(matches!(
+ result.validate(),
+ Err(ConsensusError::InvalidField(_))
+ ));
+ }
+
+ #[test]
+ fn forward_logout_result_rejects_unknown_outcome() {
+ const OUTCOME_OFFSET: usize = std::mem::offset_of!(ForwardLogoutResultHeader, outcome);
+
+ let mut bytes = aligned_zeroed(HEADER_SIZE);
+ bytes.copy_from_slice(bytemuck::bytes_of(&forward_logout_result(
+ ForwardLogoutOutcome::Ok,
+ )));
+ bytes[OUTCOME_OFFSET] = 5;
+ assert!(bytemuck::checked::try_from_bytes::<ForwardLogoutResultHeader>(&bytes).is_err());
+ }
+
+ #[test]
+ fn forward_logout_outcome_discriminants_pinned() {
+ assert_eq!(ForwardLogoutOutcome::Ok as u8, 0);
+ assert_eq!(ForwardLogoutOutcome::NotPrimary as u8, 1);
+ assert_eq!(ForwardLogoutOutcome::PipelineFull as u8, 2);
+ assert_eq!(ForwardLogoutOutcome::InProgress as u8, 3);
+ assert_eq!(ForwardLogoutOutcome::Canceled as u8, 4);
+ }
}
diff --git a/core/binary_protocol/src/consensus/mod.rs b/core/binary_protocol/src/consensus/mod.rs
index 7738256..bac436e 100644
--- a/core/binary_protocol/src/consensus/mod.rs
+++ b/core/binary_protocol/src/consensus/mod.rs
@@ -46,7 +46,9 @@
pub use error::ConsensusError;
pub use header::{
CHECKSUM_UNSEALED, CommitHeader, ConsensusHeader, DVC_HEADERS_MAX, DoViewChangeHeader,
- EvictionHeader, EvictionReason, GenericHeader, HEADER_SIZE, PrepareHeader, PrepareOkHeader,
+ EvictionHeader, EvictionReason, ForwardLogoutHeader, ForwardLogoutOutcome,
+ ForwardLogoutResultHeader, ForwardRegisterHeader, ForwardRegisterOutcome,
+ ForwardRegisterResultHeader, GenericHeader, HEADER_SIZE, PrepareHeader, PrepareOkHeader,
RESERVED_COMMAND_LEN, RepairPrepareHeader, RepairRangeReplyHeader, ReplyHeader, RequestHeader,
RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
RequestStateTransferHeader, RoutedRequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader,
diff --git a/core/binary_protocol/src/lib.rs b/core/binary_protocol/src/lib.rs
index 3510976..0825d0f 100644
--- a/core/binary_protocol/src/lib.rs
+++ b/core/binary_protocol/src/lib.rs
@@ -72,13 +72,14 @@
pub use codec::{WireDecode, WireEncode};
pub use consensus::{
CHECKSUM_UNSEALED, Command2, CommitHeader, ConsensusError, ConsensusHeader, DVC_HEADERS_MAX,
- DoViewChangeHeader, EvictionHeader, EvictionReason, GenericHeader, HEADER_SIZE, Operation,
- PrepareHeader, PrepareOkHeader, RESERVED_COMMAND_LEN, RepairPrepareHeader,
- RepairRangeReplyHeader, ReplyHeader, RequestHeader, RequestPreparesHeader,
- RequestStartViewHeader, RequestStateChunkHeader, RequestStateTransferHeader,
- RoutedRequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader,
- StateChunkHeader, StateTransferTargetHeader, frame_body, frame_checksum_bytes, read_size_field,
- result_code, result_section_len,
+ DoViewChangeHeader, EvictionHeader, EvictionReason, ForwardLogoutHeader, ForwardLogoutOutcome,
+ ForwardLogoutResultHeader, ForwardRegisterHeader, ForwardRegisterOutcome,
+ ForwardRegisterResultHeader, GenericHeader, HEADER_SIZE, Operation, PrepareHeader,
+ PrepareOkHeader, RESERVED_COMMAND_LEN, RepairPrepareHeader, RepairRangeReplyHeader,
+ ReplyHeader, RequestHeader, RequestPreparesHeader, RequestStartViewHeader,
+ RequestStateChunkHeader, RequestStateTransferHeader, RoutedRequestHeader, SIZE_FIELD_OFFSET,
+ StartViewChangeHeader, StartViewHeader, StateChunkHeader, StateTransferTargetHeader,
+ frame_body, frame_checksum_bytes, read_size_field, result_code, result_section_len,
};
pub use dispatch::{COMMAND_TABLE, CommandMeta, lookup_by_operation, lookup_command};
pub use error::WireError;
diff --git a/core/cli/Cargo.toml b/core/cli/Cargo.toml
index 9bb2474..2a6841d 100644
--- a/core/cli/Cargo.toml
+++ b/core/cli/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-cli"
-version = "0.14.0-edge.1"
+version = "0.14.0-edge.2"
edition = "2024"
rust-version.workspace = true
authors = ["bartosz.ciesla@gmail.com"]
diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml
index 2e41e25..a521aad 100644
--- a/core/common/Cargo.toml
+++ b/core/common/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_common"
-version = "0.11.0-edge.1"
+version = "0.11.0-edge.2"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
rust-version.workspace = true
diff --git a/core/common/src/traits/cluster_client.rs b/core/common/src/traits/cluster_client.rs
index 0cf09af..0d67e88 100644
--- a/core/common/src/traits/cluster_client.rs
+++ b/core/common/src/traits/cluster_client.rs
@@ -23,7 +23,6 @@
pub trait ClusterClient {
/// Get the metadata of the cluster including node information, roles, and status.
///
- /// Served pre-auth so an unauthenticated client can locate the cluster
- /// leader before signing in; the server applies its own policy.
+ /// Requires authentication; leader placement happens after login.
async fn get_cluster_metadata(&self) -> Result<ClusterMetadata, IggyError>;
}
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
index c9b7f5a..8f73799 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-connectors"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Connectors runtime for Iggy message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sdk/Cargo.toml b/core/connectors/sdk/Cargo.toml
index fcd2b6f..74b6e7d 100644
--- a/core/connectors/sdk/Cargo.toml
+++ b/core/connectors/sdk/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_sdk"
-version = "0.3.1-edge.1"
+version = "0.4.0-edge.1"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/clickhouse_sink/Cargo.toml b/core/connectors/sinks/clickhouse_sink/Cargo.toml
index 25135a7..c2ae22f 100644
--- a/core/connectors/sinks/clickhouse_sink/Cargo.toml
+++ b/core/connectors/sinks/clickhouse_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_clickhouse_sink"
-version = "0.2.0-edge.1"
+version = "0.2.0-edge.2"
description = "Iggy ClickHouse sink connector for streaming messages into ClickHouse"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/delta_sink/Cargo.toml b/core/connectors/sinks/delta_sink/Cargo.toml
index 16f5545..6a5ec4b 100644
--- a/core/connectors/sinks/delta_sink/Cargo.toml
+++ b/core/connectors/sinks/delta_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_delta_sink"
-version = "0.2.0-edge.1"
+version = "0.2.0-edge.2"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/doris_sink/Cargo.toml b/core/connectors/sinks/doris_sink/Cargo.toml
index f032024..815cea2 100644
--- a/core/connectors/sinks/doris_sink/Cargo.toml
+++ b/core/connectors/sinks/doris_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_doris_sink"
-version = "0.2.0-edge.1"
+version = "0.2.0-edge.2"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/elasticsearch_sink/Cargo.toml b/core/connectors/sinks/elasticsearch_sink/Cargo.toml
index 095590a..0103476 100644
--- a/core/connectors/sinks/elasticsearch_sink/Cargo.toml
+++ b/core/connectors/sinks/elasticsearch_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_elasticsearch_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy Elasticsearch sink connector"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/http_sink/Cargo.toml b/core/connectors/sinks/http_sink/Cargo.toml
index 862de5d..ece790c 100644
--- a/core/connectors/sinks/http_sink/Cargo.toml
+++ b/core/connectors/sinks/http_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_http_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy HTTP sink connector for delivering stream messages to any HTTP endpoint via webhooks, REST APIs, or serverless functions."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/iceberg_sink/Cargo.toml b/core/connectors/sinks/iceberg_sink/Cargo.toml
index c77601a..73df016 100644
--- a/core/connectors/sinks/iceberg_sink/Cargo.toml
+++ b/core/connectors/sinks/iceberg_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_iceberg_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
edition = "2024"
license = "Apache-2.0"
keywords = ["iggy", "messaging", "streaming"]
diff --git a/core/connectors/sinks/influxdb_sink/Cargo.toml b/core/connectors/sinks/influxdb_sink/Cargo.toml
index 6e5bae4..4e56e00 100644
--- a/core/connectors/sinks/influxdb_sink/Cargo.toml
+++ b/core/connectors/sinks/influxdb_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_influxdb_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy InfluxDB sink connector for storing stream messages as line protocol"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/meilisearch_sink/Cargo.toml b/core/connectors/sinks/meilisearch_sink/Cargo.toml
index f458328..bbc771a 100644
--- a/core/connectors/sinks/meilisearch_sink/Cargo.toml
+++ b/core/connectors/sinks/meilisearch_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_meilisearch_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy Meilisearch sink connector"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/mongodb_sink/Cargo.toml b/core/connectors/sinks/mongodb_sink/Cargo.toml
index 756c0de..2ec58fd 100644
--- a/core/connectors/sinks/mongodb_sink/Cargo.toml
+++ b/core/connectors/sinks/mongodb_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_mongodb_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy MongoDB sink connector for storing stream messages into MongoDB database"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/postgres_sink/Cargo.toml b/core/connectors/sinks/postgres_sink/Cargo.toml
index a558022..e17653b 100644
--- a/core/connectors/sinks/postgres_sink/Cargo.toml
+++ b/core/connectors/sinks/postgres_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_postgres_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy PostgreSQL sink connector for storing stream messages into PostgreSQL database"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/quickwit_sink/Cargo.toml b/core/connectors/sinks/quickwit_sink/Cargo.toml
index d4a13fd..71fc393 100644
--- a/core/connectors/sinks/quickwit_sink/Cargo.toml
+++ b/core/connectors/sinks/quickwit_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_quickwit_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/s3_sink/Cargo.toml b/core/connectors/sinks/s3_sink/Cargo.toml
index 087ec60..484d311 100644
--- a/core/connectors/sinks/s3_sink/Cargo.toml
+++ b/core/connectors/sinks/s3_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_s3_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy S3 sink connector for writing stream messages to Amazon S3 and S3-compatible stores"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/stdout_sink/Cargo.toml b/core/connectors/sinks/stdout_sink/Cargo.toml
index 80190ba..3a87301 100644
--- a/core/connectors/sinks/stdout_sink/Cargo.toml
+++ b/core/connectors/sinks/stdout_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_stdout_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/surrealdb_sink/Cargo.toml b/core/connectors/sinks/surrealdb_sink/Cargo.toml
index 61d0b3d..76b3017 100644
--- a/core/connectors/sinks/surrealdb_sink/Cargo.toml
+++ b/core/connectors/sinks/surrealdb_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_surrealdb_sink"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy SurrealDB sink connector for writing stream messages into SurrealDB"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/elasticsearch_source/Cargo.toml b/core/connectors/sources/elasticsearch_source/Cargo.toml
index 5d81ad6..17b5f80 100644
--- a/core/connectors/sources/elasticsearch_source/Cargo.toml
+++ b/core/connectors/sources/elasticsearch_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_elasticsearch_source"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy Elasticsearch source connector"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/influxdb_source/Cargo.toml b/core/connectors/sources/influxdb_source/Cargo.toml
index 65fe829..173a4c0 100644
--- a/core/connectors/sources/influxdb_source/Cargo.toml
+++ b/core/connectors/sources/influxdb_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_influxdb_source"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy InfluxDB source connector for polling Flux query results"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/postgres_source/Cargo.toml b/core/connectors/sources/postgres_source/Cargo.toml
index 3bf5973..6ebe69b 100644
--- a/core/connectors/sources/postgres_source/Cargo.toml
+++ b/core/connectors/sources/postgres_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_postgres_source"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy PostgreSQL source connector supporting CDC and table polling for message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/random_source/Cargo.toml b/core/connectors/sources/random_source/Cargo.toml
index f995ae1..93ac415 100644
--- a/core/connectors/sources/random_source/Cargo.toml
+++ b/core/connectors/sources/random_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_random_source"
-version = "0.5.0-edge.1"
+version = "0.5.0-edge.2"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/integration/src/harness/handle/common.rs b/core/integration/src/harness/handle/common.rs
index f42e1bd..1017aa4 100644
--- a/core/integration/src/harness/handle/common.rs
+++ b/core/integration/src/harness/handle/common.rs
@@ -44,6 +44,36 @@
(stdout, stderr)
}
+/// The binary's stderr content when it contains a panic report.
+///
+/// A dead async task leaves the process alive and every assertion green, so
+/// without this check a test can pass over a server whose consensus pump
+/// panicked; the report also drowns at the tail of the full log dump. Stderr
+/// is quiet in normal operation, so returning it whole keeps the decisive
+/// lines together with any abort context around them.
+pub fn stderr_panic_report(stderr_path: &Option<PathBuf>) -> Option<String> {
+ let stderr = stderr_path
+ .as_ref()
+ .and_then(|p| fs::read_to_string(p).ok())?;
+ let mut panics = stderr.lines().filter(|line| line.contains("panicked at"));
+ if panics.any(|line| !is_blocking_pool_dispatch(line)) {
+ Some(stderr.trim().to_string())
+ } else {
+ None
+ }
+}
+
+/// Whether a panic came from dispatching work to compio's blocking pool.
+///
+/// The runtime catches these and hands them to the awaiting caller as an
+/// error, so they cannot leave a dead task behind, which is the only thing
+/// the check above exists for. Segment preallocation dispatches `fallocate`
+/// to that pool while the shard proactors run with it disabled, so every
+/// segment creation prints one and then falls back to buffered allocation.
+fn is_blocking_pool_dispatch(panic_line: &str) -> bool {
+ panic_line.contains("asyncify.rs")
+}
+
/// Dump logs to stderr if we're panicking (for Drop impls).
pub fn dump_logs_on_panic(
binary_name: &str,
diff --git a/core/integration/src/harness/handle/server.rs b/core/integration/src/harness/handle/server.rs
index 9e23d07..5965bf1 100644
--- a/core/integration/src/harness/handle/server.rs
+++ b/core/integration/src/harness/handle/server.rs
@@ -1027,6 +1027,23 @@
// without waiting, so the freed slot could be reused while this server
// still holds its ports.
let _ = self.stop();
+ if let Some(report) = super::common::stderr_panic_report(&self.stderr_path) {
+ if std::thread::panicking() {
+ // Ahead of the full dump, which buries these lines under the
+ // complete stdout of every node.
+ eprintln!("Iggy server panicked:\n{report}");
+ } else {
+ // A dead task leaves the process alive and the test green;
+ // failing here is the only thing that surfaces it. The panic
+ // unwinds out of this `Drop` before the dump below runs, so
+ // print this node's logs first.
+ let (stdout, stderr) =
+ super::common::collect_logs(&self.stdout_path, &self.stderr_path);
+ eprintln!("Iggy server stdout:\n{stdout}");
+ eprintln!("Iggy server stderr:\n{stderr}");
+ panic!("Iggy server panicked:\n{report}");
+ }
+ }
super::common::dump_logs_on_panic("Iggy server", &self.stdout_path, &self.stderr_path);
}
}
diff --git a/core/integration/tests/cluster/client_table_restart.rs b/core/integration/tests/cluster/client_table_restart.rs
index d2ea0ea..b6fd026 100644
--- a/core/integration/tests/cluster/client_table_restart.rs
+++ b/core/integration/tests/cluster/client_table_restart.rs
@@ -376,6 +376,7 @@
let deadline = Instant::now() + RESUME_BUDGET;
let mut last_failure = "the listener never came back".to_string();
let mut attempt = 0usize;
+ let mut authenticated_attempts = Vec::new();
while Instant::now() < deadline {
let addr = addrs[attempt % addrs.len()];
attempt += 1;
@@ -438,6 +439,10 @@
);
}
}
+ // Backups can forward Register even though they cannot serve the
+ // following metadata write. Keep each authenticated socket alive so
+ // its disconnect cleanup cannot race the next rebind with a Logout.
+ authenticated_attempts.push(stream);
sleep(RETRY_PAUSE).await;
}
panic!(
diff --git a/core/integration/tests/cluster/mod.rs b/core/integration/tests/cluster/mod.rs
index 673d9a7..bff0c5c 100644
--- a/core/integration/tests/cluster/mod.rs
+++ b/core/integration/tests/cluster/mod.rs
@@ -20,3 +20,4 @@
mod metadata_state_transfer;
mod multi_shard_partition_convergence;
mod partition_state_transfer;
+mod register_forwarding;
diff --git a/core/integration/tests/cluster/register_forwarding.rs b/core/integration/tests/cluster/register_forwarding.rs
new file mode 100644
index 0000000..4811956
--- /dev/null
+++ b/core/integration/tests/cluster/register_forwarding.rs
@@ -0,0 +1,382 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Spec tests for logging in at a node that is not the metadata primary.
+//!
+//! A client may dial any node. Credentials verify against replicated user
+//! state, so the whole login except the consensus proposal already works on a
+//! backup; the backup forwards the verified identity to the primary over the
+//! replica interconnect and binds the session itself once the register
+//! commits. Nothing about the client's frame or its credentials travels.
+//!
+//! These tests speak to the backup through the raw transport
+//! (`TcpClient::login_user`) rather than through `IggyClient`, whose
+//! `login_user` redirects to the leader after a successful sign-in and would
+//! hide which node actually served the register.
+
+use iggy::prelude::*;
+use integration::harness::TestHarness;
+use integration::iggy_harness;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::time::{Instant, sleep};
+
+/// Budget for a login that may have to replay past a transient refusal (a
+/// primary still catching up, a view settling) and past a full re-election.
+///
+/// Must exceed TWO of the SDK's `RESPONSE_READ_TIMEOUT` (30s): on a read
+/// timeout `send_raw_with_response` reconnects once and re-enters `send_raw`
+/// with a fresh deadline, so a single `login_user` against an unresponsive
+/// node can burn 2 x 30s before returning. A budget at or below that ceiling
+/// admits exactly ONE attempt and makes the retry loops below unreachable.
+const LOGIN_BUDGET: Duration = Duration::from_secs(90);
+const LOGIN_RETRY_INTERVAL: Duration = Duration::from_millis(250);
+
+/// The only partition of the single-partition topic the produce test creates.
+/// Partition ids are 0-based.
+const PARTITION_ID: u32 = 0;
+
+/// Budget for a PAT to replicate from the node that minted it. The backup
+/// verifies the token against its own replicated copy and fails closed until
+/// it arrives, which is the deliberate parity with the HTTP forward.
+const REPLICATION_BUDGET: Duration = Duration::from_secs(15);
+
+/// A connected, NOT signed-in client on `address`.
+///
+/// `AutoLogin::Disabled` on purpose: `connect()` must not sign in or settle
+/// leadership, so the login below is the first and only thing the node is
+/// asked to do.
+async fn connect_without_login(address: SocketAddr) -> TcpClient {
+ let config = TcpClientConfig {
+ server_address: address.to_string(),
+ nodelay: true,
+ ..TcpClientConfig::default()
+ };
+ let client = TcpClient::create(Arc::new(config)).expect("build a tcp client");
+ Client::connect(&client)
+ .await
+ .expect("connect without signing in");
+ client
+}
+
+/// The TCP port the roster marks as the metadata primary's.
+async fn leader_tcp_port(harness: &TestHarness) -> u16 {
+ let client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("a root client (redirecting to the leader if node 0 is not it)");
+ let metadata = client
+ .get_cluster_metadata()
+ .await
+ .expect("get cluster metadata");
+ metadata
+ .nodes
+ .iter()
+ .find(|node| node.role == ClusterNodeRole::Leader)
+ .unwrap_or_else(|| panic!("the cluster must have elected a leader, got {metadata}"))
+ .endpoints
+ .tcp
+}
+
+/// The TCP address of a node that is not the metadata primary.
+async fn backup_address(harness: &TestHarness) -> SocketAddr {
+ let leader_port = leader_tcp_port(harness).await;
+ (0..harness.cluster_size())
+ .map(|index| {
+ harness
+ .node(index)
+ .tcp_addr()
+ .expect("every node must expose a TCP address")
+ })
+ .find(|address| address.port() != leader_port)
+ .expect("a multi-node cluster has a node that does not lead")
+}
+
+/// Sign in, replaying transient refusals until `budget` runs out.
+///
+/// A login is transient whenever the cluster cannot commit right now (an
+/// election in flight, a primary still catching up), and the SDK's contract
+/// for that answer is to replay.
+async fn login_root_within(
+ client: &TcpClient,
+ budget: Duration,
+) -> Result<IdentityInfo, IggyError> {
+ let deadline = Instant::now() + budget;
+ loop {
+ match client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ {
+ Ok(identity) => return Ok(identity),
+ Err(error) if Instant::now() >= deadline => return Err(error),
+ Err(_) => sleep(LOGIN_RETRY_INTERVAL).await,
+ }
+ }
+}
+
+#[iggy_harness(cluster_nodes = 3, server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_a_backup_when_a_client_signs_in_should_bind_the_session_there(
+ harness: &TestHarness,
+) {
+ let address = backup_address(harness).await;
+ let client = connect_without_login(address).await;
+
+ let identity = login_root_within(&client, LOGIN_BUDGET)
+ .await
+ .expect("a backup must complete a login by forwarding the register");
+ assert_eq!(identity.user_id, 0, "root user should have id 0");
+
+ // The session is bound on THIS node, not merely committed somewhere: an
+ // authenticated read is served locally and would be evicted otherwise.
+ client
+ .get_me()
+ .await
+ .expect("the backup must serve an authenticated read on the session it bound");
+}
+
+#[iggy_harness(cluster_nodes = 3, server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_a_backup_bound_session_when_the_client_logs_out_should_remove_it_cluster_wide(
+ harness: &TestHarness,
+) {
+ let address = backup_address(harness).await;
+ let client = connect_without_login(address).await;
+
+ login_root_within(&client, LOGIN_BUDGET)
+ .await
+ .expect("a backup must complete the login before logout");
+ client
+ .logout_user()
+ .await
+ .expect("the backup must forward Logout to the metadata primary");
+
+ assert_eq!(
+ client
+ .get_me()
+ .await
+ .expect_err("the local session must be unbound after logout")
+ .as_code(),
+ IggyError::Unauthenticated.as_code(),
+ );
+}
+
+#[iggy_harness(cluster_nodes = 3, server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_a_backup_when_a_pat_login_arrives_should_bind_the_session_there(
+ harness: &TestHarness,
+) {
+ const TOKEN_NAME: &str = "backup-login-pat";
+
+ let leader_client = harness
+ .root_client_for_node(0)
+ .await
+ .expect("a root client at the leader");
+ let raw_pat = leader_client
+ .create_personal_access_token(TOKEN_NAME, PersonalAccessTokenExpiry::NeverExpire)
+ .await
+ .expect("mint a PAT at the leader");
+
+ let address = backup_address(harness).await;
+ let client = connect_without_login(address).await;
+
+ // The backup verifies the token against its own replicated copy, so it
+ // refuses until the mint has replicated. Fail-closed by design; poll
+ // through the window rather than asserting on its width.
+ let deadline = Instant::now() + REPLICATION_BUDGET;
+ let identity = loop {
+ match client
+ .login_with_personal_access_token(&raw_pat.token)
+ .await
+ {
+ Ok(identity) => break identity,
+ Err(error) => {
+ assert!(
+ Instant::now() < deadline,
+ "a replicated PAT must eventually authenticate at a backup, got {error}"
+ );
+ sleep(LOGIN_RETRY_INTERVAL).await;
+ }
+ }
+ };
+ assert_eq!(identity.user_id, 0, "the PAT authenticates as root");
+
+ client
+ .get_me()
+ .await
+ .expect("the backup must serve an authenticated read on the session it bound");
+}
+
+#[iggy_harness(cluster_nodes = 3, server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_a_backup_when_credentials_are_wrong_should_refuse_terminally(harness: &TestHarness) {
+ let address = backup_address(harness).await;
+ let client = connect_without_login(address).await;
+
+ // Terminal, and decided locally: the credential check never leaves the
+ // backup, so a wrong password is refused before anything is forwarded.
+ let error = client
+ .login_user(DEFAULT_ROOT_USERNAME, "definitely-not-the-root-password")
+ .await
+ .expect_err("wrong credentials must be refused");
+ assert_eq!(
+ error.as_code(),
+ IggyError::InvalidCredentials.as_code(),
+ "a backup must answer wrong credentials with the terminal error, got {error}"
+ );
+}
+
+/// Losing a replica must not cost the forward its answer: the primary is
+/// still there, so the surviving backup keeps completing logins.
+///
+/// Deliberately kills a FOLLOWER rather than the primary: killing the
+/// primary tests re-election, not this feature. The sibling test below
+/// covers the primary-kill path.
+#[iggy_harness(cluster_nodes = 3, server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_a_degraded_cluster_when_a_client_signs_in_at_a_backup_should_succeed(
+ harness: &mut TestHarness,
+) {
+ let leader_port = leader_tcp_port(harness).await;
+ let backups: Vec<usize> = (0..harness.cluster_size())
+ .filter(|index| {
+ harness
+ .node(*index)
+ .tcp_addr()
+ .is_some_and(|address| address.port() != leader_port)
+ })
+ .collect();
+ let (killed, survivor) = (backups[0], backups[1]);
+ let survivor_address = harness
+ .node(survivor)
+ .tcp_addr()
+ .expect("the surviving backup must expose a TCP address");
+
+ harness.stop_node(killed).expect("stop one backup");
+
+ let client = connect_without_login(survivor_address).await;
+ let identity = login_root_within(&client, LOGIN_BUDGET)
+ .await
+ .expect("a backup must still forward its register with one replica down");
+ assert_eq!(identity.user_id, 0, "root user should have id 0");
+ client
+ .get_me()
+ .await
+ .expect("the backup must serve an authenticated read on the session it bound");
+}
+
+/// Kill the metadata PRIMARY, then sign in at a survivor: the two survivors
+/// must elect a new primary and complete the forwarded register. Used to
+/// starve on roughly half the runs before the view-start pipeline fix
+/// (a register admitted during the superblock persist panicked the pump);
+/// the deterministic interleaving is pinned by the simulator gate.
+#[iggy_harness(cluster_nodes = 3, server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_a_killed_primary_when_a_client_signs_in_at_a_survivor_should_succeed(
+ harness: &mut TestHarness,
+) {
+ let leader_port = leader_tcp_port(harness).await;
+ let leader = (0..harness.cluster_size())
+ .find(|index| {
+ harness
+ .node(*index)
+ .tcp_addr()
+ .is_some_and(|address| address.port() == leader_port)
+ })
+ .expect("the leader must be one of the roster nodes");
+ let survivor = (0..harness.cluster_size())
+ .find(|index| *index != leader)
+ .expect("a 3-node cluster has a survivor");
+ let survivor_address = harness
+ .node(survivor)
+ .tcp_addr()
+ .expect("the survivor must expose a TCP address");
+
+ harness.stop_node(leader).expect("stop the primary");
+
+ let client = connect_without_login(survivor_address).await;
+ let identity = login_root_within(&client, LOGIN_BUDGET)
+ .await
+ .expect("survivors must re-elect and complete the login");
+ assert_eq!(identity.user_id, 0, "root user should have id 0");
+}
+
+#[iggy_harness(cluster_nodes = 3, server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_a_backup_when_auto_login_dials_it_should_settle_on_the_leader(
+ harness: &TestHarness,
+) {
+ const STREAM_NAME: &str = "backup-login-stream";
+ const TOPIC_NAME: &str = "backup-login-topic";
+ const PAYLOAD: &str = "produced after a backup-dialed login";
+
+ let address = backup_address(harness).await;
+ let client = IggyClient::create(
+ ClientWrapper::Tcp(connect_without_login(address).await),
+ None,
+ None,
+ );
+
+ // `IggyClient::login_user` redirects to the leader after the backup has
+ // served the sign-in, which is what makes replicated writes work below.
+ client
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ .expect("a backup-dialed login must succeed and settle on the leader");
+
+ client
+ .create_stream(STREAM_NAME)
+ .await
+ .expect("create stream after a backup-dialed login");
+ let stream_id = Identifier::named(STREAM_NAME).expect("stream identifier");
+ client
+ .create_topic(
+ &stream_id,
+ TOPIC_NAME,
+ 1,
+ CompressionAlgorithm::None,
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .expect("create topic after a backup-dialed login");
+
+ let topic_id = Identifier::named(TOPIC_NAME).expect("topic identifier");
+ let mut messages = vec![IggyMessage::from(PAYLOAD)];
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(PARTITION_ID),
+ &mut messages,
+ )
+ .await
+ .expect("produce after a backup-dialed login");
+
+ let polled = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(PARTITION_ID),
+ &Consumer::default(),
+ &PollingStrategy::offset(0),
+ 1,
+ false,
+ )
+ .await
+ .expect("poll after a backup-dialed login");
+ assert_eq!(
+ polled.messages.len(),
+ 1,
+ "the produced message must poll back"
+ );
+}
diff --git a/core/integration/tests/server/legacy_login_vsr.rs b/core/integration/tests/server/legacy_login_vsr.rs
index 9e129e9..06ae510 100644
--- a/core/integration/tests/server/legacy_login_vsr.rs
+++ b/core/integration/tests/server/legacy_login_vsr.rs
@@ -20,9 +20,9 @@
//! `LOGIN_WITH_PERSONAL_ACCESS_TOKEN` (44) codes -- which the vsr SDK never
//! emits (its typed login methods send the register codes, its raw path
//! rejects session-control codes) -- must be rejected with a typed
-//! `MalformedLogin` eviction, instead of the misleading `NoSession` eviction
-//! the pre-auth guard would send unbound, or the silent empty-ok reply the
-//! bound non-replicated path would send.
+//! `MalformedLogin` eviction, instead of the generic `Unauthenticated` deny
+//! reply the pre-auth guard would send unbound, or the silent empty-ok reply
+//! the bound non-replicated path would send.
//! Since the SDK cannot send these codes, the frames are hand-crafted on a raw
//! TCP socket: a header-only non-replicated frame carrying the code in the
//! reserved command slot.
diff --git a/core/integration/tests/server/scenarios/authentication_scenario.rs b/core/integration/tests/server/scenarios/authentication_scenario.rs
index 0932e55..c5b3ddf 100644
--- a/core/integration/tests/server/scenarios/authentication_scenario.rs
+++ b/core/integration/tests/server/scenarios/authentication_scenario.rs
@@ -114,7 +114,7 @@
let name = entry.name;
// ================================================================
- // SKIPPED COMMANDS (11 total)
+ // SKIPPED COMMANDS
// ================================================================
// No auth required
if matches!(
@@ -127,12 +127,6 @@
) {
continue;
}
- // The server serves `GetClusterMetadata` pre-auth so a client can
- // locate the cluster leader before signing in; the legacy server
- // still auth-gates it.
- if code == GET_CLUSTER_METADATA_CODE {
- continue;
- }
// Stateful - not supported on HTTP. `SYNC_CONSUMER_GROUP` is
// SDK-internal (issued during poll partition resolution), with no
// top-level client method to invoke unauthenticated here; its auth
@@ -170,6 +164,7 @@
GET_ME_CODE => client.get_me().await.map(|_| ()),
GET_CLIENT_CODE => client.get_client(1).await.map(|_| ()),
GET_CLIENTS_CODE => client.get_clients().await.map(|_| ()),
+
GET_CLUSTER_METADATA_CODE => client.get_cluster_metadata().await.map(|_| ()),
// Users
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index c4bb620..73f1e31 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -493,6 +493,12 @@
/// the pipeline). The caller retries; the SDK read-timeout replay reaches
/// the new primary.
Canceled,
+ /// The node this view names primary is not reachable from this shard, so
+ /// a forwarded session operation never left. Nothing was proposed.
+ PrimaryUnreachable,
+ /// A forwarded session operation left but no verdict came back within the
+ /// forward timeout. The proposal's outcome is unknown.
+ ForwardTimedOut,
/// The presented `client_id` already has a table entry owned by a
/// DIFFERENT user. TERMINAL, unlike every sibling: retrying cannot help,
/// and admitting it would run the caller's replicated ops under the
@@ -511,6 +517,8 @@
impl MetadataSubmitError {
/// Whether a retry (here, or against another replica) could succeed.
/// Every variant is transient by contract except the ownership refusal.
+ /// Deliberately a deny-list: a new variant is transient by default, so
+ /// adding one cannot silently surface a terminal error to clients.
#[must_use]
pub const fn is_transient(&self) -> bool {
!matches!(self, Self::ClientIdOwnedByAnotherUser)
@@ -525,6 +533,10 @@
Self::PipelineFull => f.write_str("metadata prepare queue is full"),
Self::InProgress => f.write_str("another in-flight prepare from this client"),
Self::Canceled => f.write_str("view change canceled the pending prepare"),
+ Self::PrimaryUnreachable => f.write_str("no route to the metadata primary"),
+ Self::ForwardTimedOut => {
+ f.write_str("the metadata primary did not answer the forwarded register")
+ }
Self::ClientIdOwnedByAnotherUser => {
f.write_str("client id already registered to a different user")
}
@@ -1821,9 +1833,10 @@
/// # Errors
/// [`MetadataSubmitError`]. All transient except
/// `ClientIdOwnedByAnotherUser`, which is terminal: `NotPrimary`,
- /// `NotCaughtUp`, `PipelineFull`, `InProgress`, `Canceled`. `Canceled`
- /// dominates on view change; the new primary inherits via
- /// `commit_journal` and the SDK retries.
+ /// `PipelineFull`, `InProgress`, `Canceled`. Never `NotCaughtUp`: a
+ /// not-caught-up primary parks the register in the request queue instead
+ /// of bouncing it. `Canceled` dominates on view change; the new primary
+ /// inherits via `commit_journal` and the SDK retries.
///
/// # Panics
/// On `client_id == 0` or shard without consensus.
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index 398680d..ea76718 100644
--- a/core/sdk/Cargo.toml
+++ b/core/sdk/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy"
-version = "0.11.0-edge.1"
+version = "0.11.0-edge.2"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
rust-version.workspace = true
diff --git a/core/sdk/src/leader_aware.rs b/core/sdk/src/leader_aware.rs
index aa25ad9..35ffce9 100644
--- a/core/sdk/src/leader_aware.rs
+++ b/core/sdk/src/leader_aware.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+use iggy_binary_protocol::codes::GET_CLUSTER_METADATA_CODE;
use iggy_common::ClusterClient;
use iggy_common::{
ClusterMetadata, ClusterNodeRole, ClusterNodeStatus, IggyError, TransportProtocol,
@@ -26,6 +27,17 @@
/// Maximum number of leader redirections to prevent infinite loops
const MAX_LEADER_REDIRECTS: u8 = 3;
+/// An auth-gated `get_cluster_metadata` read denied before sign-in.
+///
+/// The read is public API, so an unauthenticated caller can reach the gate on
+/// any transport. Reconnecting cannot repair it: `connect()` would re-issue
+/// the same unauthenticated read forever, so every transport must fail such a
+/// request fast instead of entering its reconnect path. One definition, since
+/// a transport missing this rule livelocks its reconnect loop.
+pub(crate) fn is_unauthenticated_metadata_probe(code: u32, error: &IggyError) -> bool {
+ code == GET_CLUSTER_METADATA_CODE && matches!(error, IggyError::Unauthenticated)
+}
+
/// Check if we need to redirect to leader and return the leader address if redirection is needed
pub async fn check_and_redirect_to_leader<C: ClusterClient>(
client: &C,
@@ -62,6 +74,16 @@
}
}
}
+ // The read is auth-gated everywhere, and this check runs after
+ // sign-in, so an unauthenticated answer means the session died
+ // between the two. Proceed on the current node and let the
+ // caller's next request surface the eviction.
+ Err(IggyError::Unauthenticated) => {
+ debug!(
+ "Cluster metadata answered Unauthenticated; the session is gone, connection will continue on server node {current_address}"
+ );
+ return Ok(None);
+ }
Err(e) => {
warn!(
"Failed to get cluster metadata: {}, connection will continue on server node {}",
@@ -208,6 +230,22 @@
use super::*;
#[test]
+ fn only_unauthenticated_cluster_metadata_is_a_pre_login_probe() {
+ assert!(is_unauthenticated_metadata_probe(
+ GET_CLUSTER_METADATA_CODE,
+ &IggyError::Unauthenticated,
+ ));
+ assert!(!is_unauthenticated_metadata_probe(
+ GET_CLUSTER_METADATA_CODE,
+ &IggyError::Disconnected,
+ ));
+ assert!(!is_unauthenticated_metadata_probe(
+ GET_CLUSTER_METADATA_CODE + 1,
+ &IggyError::Unauthenticated,
+ ));
+ }
+
+ #[test]
fn test_is_same_address() {
assert!(is_same_address("127.0.0.1:8090", "127.0.0.1:8090"));
assert!(is_same_address("localhost:8090", "127.0.0.1:8090"));
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index 7bf52ee..921ef5d 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use crate::leader_aware::{LeaderRedirectionState, check_and_redirect_to_leader};
+use crate::leader_aware::{
+ LeaderRedirectionState, check_and_redirect_to_leader, is_unauthenticated_metadata_probe,
+};
use crate::prelude::AutoLogin;
use crate::session::ConsensusSession;
use iggy_common::VsrSessionControl as _;
@@ -144,6 +146,10 @@
return Err(error);
}
+ if is_unauthenticated_metadata_probe(code, &error) {
+ return Err(error);
+ }
+
if !self.config.reconnection.enabled {
return Err(IggyError::Disconnected);
}
@@ -443,26 +449,17 @@
let should_redirect = match &self.config.auto_login {
AutoLogin::Disabled => {
info!("Automatic sign-in is disabled.");
- // Leadership still matters without auto-login: the caller
- // signs in manually, and a login against a non-leader
- // replays for its whole read timeout. `GetClusterMetadata`
- // is sessionless and pre-auth, so the check works on the
- // unauthenticated connection.
- self.handle_leader_redirection().await?
+ // Only `IggyClient` redirects after a manual sign-in, so
+ // a raw transport can stay on a backup, and nothing on
+ // the send path redirects either: its replicated writes
+ // replay on the live connection, then surface the
+ // transient failure to the caller.
+ false
}
AutoLogin::Enabled(credentials) => {
if skip_auto_login {
info!("Skipping automatic sign-in for a retried login/register request.");
false
- } else if self.handle_leader_redirection().await? {
- // Check leadership BEFORE signing in: register/login are
- // consensus ops a backup answers with
- // `TransientNotCommitted`, so signing in against a
- // non-leader replays for the whole read timeout instead
- // of failing over. `GetClusterMetadata` is sessionless
- // and pre-auth, so it works on the unauthenticated
- // connection.
- true
} else {
info!(
"{NAME} client: {} is signing in...",
@@ -489,6 +486,11 @@
}
}
+ // The sole leader settlement, and it runs
+ // authenticated. Any node completes a login now -- a
+ // backup forwards the register to the primary -- so
+ // this decides where later ops land, not whether
+ // sign-in works.
self.handle_leader_redirection().await?
}
}
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index d68e3d6..2c09e3f 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use crate::leader_aware::{LeaderRedirectionState, check_and_redirect_to_leader};
+use crate::leader_aware::{
+ LeaderRedirectionState, check_and_redirect_to_leader, is_unauthenticated_metadata_probe,
+};
use crate::prelude::Client;
use crate::prelude::TcpClientConfig;
use crate::session::ConsensusSession;
@@ -149,6 +151,10 @@
return Err(error);
}
+ if is_unauthenticated_metadata_probe(code, &error) {
+ return Err(error);
+ }
+
if !self.config.reconnection.enabled {
return Err(IggyError::Disconnected);
}
@@ -482,26 +488,17 @@
let should_redirect = match &self.config.auto_login {
AutoLogin::Disabled => {
info!("Automatic sign-in is disabled.");
- // Leadership still matters without auto-login: the caller
- // signs in manually, and a login against a non-leader
- // replays for its whole read timeout. `GetClusterMetadata`
- // is sessionless and pre-auth, so the check works on the
- // unauthenticated connection.
- self.handle_leader_redirection().await?
+ // Only `IggyClient` redirects after a manual sign-in, so
+ // a raw transport can stay on a backup: its first
+ // replicated write gets `TransientNotAccepted`, the
+ // redirect drops the session, and the retry fails
+ // `Unauthenticated` until the caller signs in again.
+ false
}
AutoLogin::Enabled(credentials) => {
if skip_auto_login {
info!("Skipping automatic sign-in for a retried login/register request.");
false
- } else if self.handle_leader_redirection().await? {
- // Check leadership BEFORE signing in: register/login are
- // consensus ops a backup answers with
- // `TransientNotCommitted`, so signing in against a
- // non-leader replays for the whole read timeout instead
- // of failing over. `GetClusterMetadata` is sessionless
- // and pre-auth, so it works on the unauthenticated
- // connection.
- true
} else {
info!("{NAME} client: {client_address} is signing in...");
self.set_state(ClientState::Authenticating).await;
@@ -521,6 +518,11 @@
}
}
+ // The sole leader settlement, and it runs
+ // authenticated. Any node completes a login now -- a
+ // backup forwards the register to the primary -- so
+ // this decides where later ops land, not whether
+ // sign-in works.
self.handle_leader_redirection().await?
}
}
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
index 514ab6d..c439a38 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use crate::leader_aware::{LeaderRedirectionState, check_and_redirect_to_leader};
+use crate::leader_aware::{
+ LeaderRedirectionState, check_and_redirect_to_leader, is_unauthenticated_metadata_probe,
+};
use crate::session::ConsensusSession;
use crate::websocket::websocket_connection_stream::WebSocketConnectionStream;
use crate::websocket::websocket_stream_kind::WebSocketStreamKind;
@@ -141,11 +143,15 @@
return Err(error);
}
+ if is_unauthenticated_metadata_probe(code, &error) {
+ return Err(error);
+ }
+
if !self.config.reconnection.enabled {
return Err(IggyError::Disconnected);
}
- if matches!(self.config.auto_login, AutoLogin::Disabled) {
+ if matches!(self.config.auto_login, AutoLogin::Disabled) && !is_login_register_code(code) {
return Err(error);
}
@@ -507,21 +513,17 @@
async fn check_and_maybe_redirect(&self) -> Result<bool, IggyError> {
match &self.config.auto_login {
- // Leadership still matters without auto-login: the caller signs in
- // manually, and a login against a non-leader replays for its whole
- // read timeout. `GetClusterMetadata` is sessionless and pre-auth,
- // so the check works on the unauthenticated connection.
- AutoLogin::Disabled => self.handle_leader_redirection().await,
+ // Only `IggyClient` redirects after a manual sign-in, so a raw
+ // transport can stay on a backup, and nothing on the send path
+ // redirects either: its replicated writes replay on the live
+ // connection, then surface the transient failure to the caller.
+ AutoLogin::Disabled => Ok(false),
AutoLogin::Enabled(_) => {
- // Check leadership BEFORE signing in: register/login are
- // consensus ops a backup answers with a transient frame, so
- // signing in against a non-leader replays for the whole read
- // timeout instead of failing over. `GetClusterMetadata` is
- // sessionless and pre-auth.
- if self.handle_leader_redirection().await? {
- return Ok(true);
- }
self.auto_login().await?;
+ // The sole leader settlement, and it runs authenticated. Any
+ // node completes a login now -- a backup forwards the register
+ // to the primary -- so this decides where later ops land, not
+ // whether sign-in works.
self.handle_leader_redirection().await
}
}
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index e36b0d1..31c8efd 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "server"
-version = "0.9.0-edge.2"
+version = "0.9.0-edge.3"
edition = "2024"
license = "Apache-2.0"
publish = false
diff --git a/core/server/src/auth.rs b/core/server/src/auth.rs
index a491eb9..7af2e8c 100644
--- a/core/server/src/auth.rs
+++ b/core/server/src/auth.rs
@@ -19,8 +19,8 @@
//!
//! Verifies password + PAT credentials locally, then runs the consensus
//! `Register` proposal on the metadata owner; terminal failures are
-//! surfaced as typed `Eviction` frames, transient ones as
-//! `TransientNotAccepted` replay hints.
+//! surfaced as typed `Eviction` frames, transient ones as result-framed
+//! replay hints.
use crate::bootstrap::{ShellBus, ShellShard};
use crate::dispatch::{send_login_eviction, submit_register_on_owner};
@@ -36,6 +36,7 @@
use iggy_common::{IggyError, IggyTimestamp, PersonalAccessToken, UserStatus};
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
+use metadata::MetadataSubmitError;
use metadata::impls::metadata::StreamsFrontend;
use server_common::Message;
use server_common::crypto;
@@ -211,7 +212,9 @@
sessions
.borrow_mut()
.record_sdk_info(transport_client_id, sdk_info);
- let commit = current_metadata_commit(shard);
+ // A lagging backup's commit_max can sit below the epoch this session
+ // already bound; never advertise a commit behind the session itself.
+ let commit = current_metadata_commit(shard).max(session);
let reply =
build_login_register_reply(request_header, vsr_client_id, session, commit, user_id);
let _ = shard
@@ -256,7 +259,10 @@
}
}
- let commit = current_metadata_commit(shard);
+ // `session` IS the register's commit op, and on a backup that forwarded
+ // the proposal the local applied commit still lags it. Reporting the
+ // lower number would make one frame contradict itself.
+ let commit = current_metadata_commit(shard).max(session);
let reply = build_login_register_reply(request_header, vsr_client_id, session, commit, user_id);
let send_result = shard
.bus
@@ -278,10 +284,10 @@
///
/// A transient consensus failure ([`LoginRegisterError::is_terminal`] is
/// `false`) means the cluster could not commit *right now* (a freshly booted
-/// primary still catching up, or a cross-shard submit canceled). Staying
-/// silent lets the SDK read-timeout replay once the primary is caught up;
-/// replying empty would surface as a hard `InvalidFormat` decode failure and
-/// break the replay.
+/// primary still catching up, or a cross-shard submit canceled). Those get a
+/// result-framed replay hint instead of silence, so the SDK replays at once
+/// rather than waiting out its read-timeout; replying empty would surface as
+/// a hard `InvalidFormat` decode failure and break the replay.
///
/// Terminal auth errors (`InvalidCredentials` / `InvalidToken` /
/// `UserInactive` / `Session`) fast-fail with a typed `Eviction` frame so the
@@ -309,24 +315,28 @@
)
.await;
} else {
- // Transient consensus failure (not-caught-up / not-primary / pipeline
- // full): send the explicit `TransientNotAccepted` frame instead of
- // staying silent, so the SDK replays the login immediately rather than
- // waiting out its read-timeout. Same contract as a transient metadata
- // request -- nothing committed, so the replayed Register is idempotent.
- send_login_transient_reply(shard, transport_client_id, request_header).await;
+ // Which code the hint carries is what tells the client whether the
+ // replay may move to another node: see `transient_login_code`.
+ send_login_transient_reply(
+ shard,
+ transport_client_id,
+ request_header,
+ transient_login_code(error),
+ )
+ .await;
}
}
-/// Result-framed `TransientNotAccepted` Reply on a transient (non-terminal)
-/// failed Register. The SDK decodes the nonzero result code and replays the
-/// same login on the same connection. Only call for transient errors -- see
+/// Result-framed transient Reply on a non-terminal failed Register. The SDK
+/// decodes the nonzero result code and replays the same login on the same
+/// connection. Only call for transient errors -- see
/// [`surface_login_failure`].
#[allow(clippy::future_not_send)]
async fn send_login_transient_reply<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
request_header: &RoutedRequestHeader,
+ code: IggyError,
) where
B: ShellBus,
MJ: JournalHandle + 'static,
@@ -335,14 +345,7 @@
SB: SuperblockStore + 'static,
{
let commit = current_metadata_commit(shard);
- // `TransientNotAccepted`: a login/register replay is safe under any
- // session (a duplicate register mints a fresh session and the server
- // evicts the abandoned one), so the client may fail over freely.
- let reply = build_result_rejection_reply(
- request_header,
- commit,
- IggyError::TransientNotAccepted.as_code(),
- );
+ let reply = build_result_rejection_reply(request_header, commit, code.as_code());
if let Err(error) = shard
.bus
.send_to_client(transport_client_id, reply.into_generic().into_frozen())
@@ -356,6 +359,30 @@
}
}
+/// Wire code for a transient (non-terminal) login/register failure.
+///
+/// `TransientNotAccepted` asserts nothing was committed: the register never
+/// entered a pipeline (not primary / not caught up / pipeline full) or never
+/// left this node (primary unreachable). The client may re-issue it anywhere,
+/// including under a fresh identity after failing over to another node.
+///
+/// A forward timeout, an in-progress proposal, or a canceled proposal has an
+/// UNKNOWN outcome, so none can ride that assertion. `TransientNotCommitted`
+/// pins the replay to this connection and its client id, where a register that
+/// did commit rebinds its own client-table entry. Re-issuing under a freshly
+/// minted id would instead orphan that entry until capacity eviction reclaims
+/// it.
+const fn transient_login_code(error: &LoginRegisterError) -> IggyError {
+ match error {
+ LoginRegisterError::Transient(
+ MetadataSubmitError::ForwardTimedOut
+ | MetadataSubmitError::InProgress
+ | MetadataSubmitError::Canceled,
+ ) => IggyError::TransientNotCommitted,
+ _ => IggyError::TransientNotAccepted,
+ }
+}
+
/// Wire reason for a terminal login/register failure. Session-level
/// rejections (including the non-retryable submit refusal, where the
/// presented client id belongs to another user) collapse to
@@ -402,4 +429,29 @@
);
}
}
+
+ #[test]
+ fn unknown_register_outcomes_pin_the_client_identity() {
+ for error in [
+ MetadataSubmitError::ForwardTimedOut,
+ MetadataSubmitError::InProgress,
+ MetadataSubmitError::Canceled,
+ ] {
+ assert_eq!(
+ transient_login_code(&LoginRegisterError::Transient(error)),
+ IggyError::TransientNotCommitted,
+ );
+ }
+ for error in [
+ MetadataSubmitError::NotPrimary,
+ MetadataSubmitError::NotCaughtUp,
+ MetadataSubmitError::PipelineFull,
+ MetadataSubmitError::PrimaryUnreachable,
+ ] {
+ assert_eq!(
+ transient_login_code(&LoginRegisterError::Transient(error)),
+ IggyError::TransientNotAccepted,
+ );
+ }
+ }
}
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 15e2905..685e040 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -31,6 +31,7 @@
use crate::segment_recovery::{RecoveredSegment, load_persisted_segments};
use crate::server_error::{ServerError, ShardJoinFailure, ShardJoinFailureKind};
use crate::session_manager::SessionManager;
+use compio::runtime::ResumeUnwind;
use configs::server::{ServerConfig, ServerSystemConfig};
use configs::sharding::{
INBOX_CAPACITY_MAX, SHUTDOWN_DRAIN_TIMEOUT_MAX, SHUTDOWN_POLL_INTERVAL_MAX,
@@ -103,6 +104,7 @@
use std::collections::HashMap;
use std::env;
use std::net::{IpAddr, SocketAddr};
+use std::panic;
use std::path::{Path, PathBuf};
use std::rc::{Rc, Weak};
use std::sync::Arc;
@@ -1412,7 +1414,9 @@
if let Some(tx) = &segment_cleaner_stop {
let _ = tx.try_send(());
}
- await_pump_drain(pump_handle.take(), config, shard_id).await;
+ // The bind failure is the primary fault; the drain verdict only
+ // matters for the log it emits.
+ let _ = await_pump_drain(pump_handle.take(), config, shard_id).await;
return Err(error);
}
@@ -1439,7 +1443,7 @@
let _ = tx.try_send(());
}
- await_pump_drain(pump_handle.take(), config, shard_id).await;
+ await_pump_drain(pump_handle.take(), config, shard_id).await?;
info!(shard = shard_id, "server shard exited cleanly");
Ok(())
@@ -1449,25 +1453,58 @@
/// post-loop work includes the final flush of every committed journal to
/// segment storage, and returning first drops the compio runtime, which
/// cancels that flush at its next await point.
+///
+/// `Err` means the pump was already dead (a panic, or an exit outside the
+/// stop protocol), so its final flush never ran and the shard must not
+/// report a clean exit. The verdict is the inner `JoinError`; the timeout
+/// wrapper alone cannot see it, and a shard that swallows it prints
+/// "exited cleanly" over a corpse.
async fn await_pump_drain(
pump_handle: Option<compio::runtime::JoinHandle<()>>,
config: &ServerConfig,
shard_id: u16,
-) {
+) -> Result<(), ServerError> {
let Some(pump_handle) = pump_handle else {
- return;
+ return Ok(());
};
let drain_budget = config.system.sharding.shutdown_drain_timeout.get_duration();
- if compio::time::timeout(drain_budget, pump_handle)
- .await
- .is_err()
- {
- warn!(
+ let Ok(join_result) = compio::time::timeout(drain_budget, pump_handle).await else {
+ error!(
shard = shard_id,
+ timeout = ?drain_budget,
"message pump did not drain within the shutdown budget; \
committed journal tail may not have flushed"
);
- }
+ return Err(ServerError::ShardPumpDrainTimedOut {
+ shard_id,
+ timeout: drain_budget,
+ });
+ };
+ // `JoinError` renders a panic as the bare "Task has panicked" and the
+ // type is not re-exported, so the payload -- the only part with
+ // diagnostic value -- is lifted by re-raising into an immediate catch.
+ // The panic hook already ran when the task died; `resume_unwind` does
+ // not run it again, so nothing is printed twice and the message finally
+ // reaches the tracing sink too.
+ let reason = match panic::catch_unwind(panic::AssertUnwindSafe(|| join_result.resume_unwind()))
+ {
+ Ok(Some(())) => return Ok(()),
+ Ok(None) => "task was cancelled".to_string(),
+ Err(payload) => payload
+ .downcast_ref::<&str>()
+ .map(|message| (*message).to_string())
+ .or_else(|| payload.downcast_ref::<String>().cloned())
+ .map_or_else(
+ || "task panicked".to_string(),
+ |message| format!("task panicked: {message}"),
+ ),
+ };
+ error!(
+ shard = shard_id,
+ "message pump died instead of draining ({reason}); \
+ committed journal tail may not have flushed"
+ );
+ Err(ServerError::ShardPumpDied { shard_id, reason })
}
/// Block until shard 0 broadcasts the metadata factory bundle, or the
@@ -1921,7 +1958,7 @@
// Same wiring path as the simulator's shell mode: one per-shard
// SessionManager shared by the client-request handler (binds sessions)
// and the get_clients handler (reads them). It also carries this shard's
- // cluster roster for the pre-auth GetClusterMetadata read.
+ // cluster roster for the GetClusterMetadata read.
let ShellHandlers {
on_replica_message,
on_client_request,
@@ -4378,6 +4415,28 @@
}
#[compio::test]
+ async fn pump_drain_timeout_is_not_reported_as_clean() {
+ let mut config = ServerConfig::default();
+ let timeout = Duration::from_millis(1);
+ Arc::get_mut(&mut config.system)
+ .expect("a fresh ServerConfig owns its system config")
+ .sharding
+ .shutdown_drain_timeout = iggy_common::IggyDuration::new(timeout);
+ let pump = compio::runtime::spawn(std::future::pending::<()>());
+
+ let error = await_pump_drain(Some(pump), &config, 7)
+ .await
+ .expect_err("a live pump past the drain budget is not a clean exit");
+ assert!(matches!(
+ error,
+ ServerError::ShardPumpDrainTimedOut {
+ shard_id: 7,
+ timeout: actual,
+ } if actual == timeout
+ ));
+ }
+
+ #[compio::test]
async fn signal_bootstrap_complete_aborts_when_owner_drops_rx() {
// Shard 0 aborted before draining and dropped its receiver; a peer's
// signal must surface the disconnect instead of stranding.
diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs
index aa198b8..807c198 100644
--- a/core/server/src/dispatch.rs
+++ b/core/server/src/dispatch.rs
@@ -35,7 +35,7 @@
};
use crate::dispatch::authz::{
authorize_default_read, authorize_partition_op, authorize_partition_read, authorize_uid,
- send_non_replicated_deny, send_partition_deny_reply,
+ send_deny_reply, send_non_replicated_deny, send_unbound_deny_reply,
};
use crate::login_register::LoginRegisterError;
use crate::pat::maybe_rewrite_pat_request;
@@ -84,9 +84,11 @@
use iggy_binary_protocol::responses::consumer_groups::SyncConsumerGroupResponse;
use iggy_binary_protocol::responses::system::get_snapshot::GetSnapshotResponse;
use iggy_binary_protocol::{
- AckLevel, ClientVersionInfo, Command2, EvictionReason, GenericHeader, HEADER_SIZE,
- KIND_CONSUMER_GROUP, MAX_PARTITIONS_PER_REQUEST, Operation, ProtocolVersion, RequestHeader,
- RoutedRequestHeader, WireDecode, WireEncode, WireIdentifier, is_protocol_compatible,
+ AckLevel, ClientVersionInfo, Command2, ConsensusHeader, EvictionReason, ForwardLogoutHeader,
+ ForwardLogoutOutcome, ForwardLogoutResultHeader, ForwardRegisterHeader, ForwardRegisterOutcome,
+ ForwardRegisterResultHeader, GenericHeader, HEADER_SIZE, KIND_CONSUMER_GROUP,
+ MAX_PARTITIONS_PER_REQUEST, Operation, ProtocolVersion, RequestHeader, RoutedRequestHeader,
+ WireDecode, WireEncode, WireIdentifier, is_protocol_compatible,
};
use iggy_common::{
IggyError, MaxTopicSize, PollingStrategy, SnapshotCompression, SystemSnapshotType,
@@ -116,6 +118,7 @@
use std::net::IpAddr;
use std::rc::Rc;
use std::sync::Arc;
+use std::time::Duration;
use tracing::{debug, warn};
pub(crate) type ClientRequestQueues = Rc<RefCell<HashMap<u128, VecDeque<Message<GenericHeader>>>>>;
@@ -524,8 +527,8 @@
/// shard has verified credentials and owns the session locally, and asks
/// shard 0 (the metadata consensus owner) to run only the consensus
/// proposal. Spawns a task so the awaiting peer is woken once the op
-/// commits; replies `None` on transient submit failure so the peer never
-/// blocks forever.
+/// commits. Submit failures are returned verbatim so the peer can preserve
+/// unknown-outcome retry semantics.
pub(crate) fn make_metadata_submit_handler<B, MJ, S, SB>(
shard_handle: &ShellShardHandle<B, MJ, S, SB>,
) -> shard::MetadataSubmitHandler
@@ -549,26 +552,52 @@
user_id,
reply,
} => {
- let bound = shard
- .plane
- .metadata()
- .submit_register_in_process(vsr_client_id, user_id)
- .await;
+ let bound =
+ submit_register_local_or_forward(&shard, vsr_client_id, user_id).await;
let _ = reply.try_send(bound);
}
+ shard::MetadataSubmit::ForwardedRegister {
+ vsr_client_id,
+ user_id,
+ nonce,
+ origin_replica,
+ } => {
+ answer_forwarded_register(
+ &shard,
+ vsr_client_id,
+ user_id,
+ nonce,
+ origin_replica,
+ )
+ .await;
+ }
+ shard::MetadataSubmit::ForwardedLogout {
+ vsr_client_id,
+ session,
+ request,
+ nonce,
+ origin_replica,
+ } => {
+ answer_forwarded_logout(
+ &shard,
+ vsr_client_id,
+ session,
+ request,
+ nonce,
+ origin_replica,
+ )
+ .await;
+ }
shard::MetadataSubmit::Logout {
vsr_client_id,
session,
request,
reply,
} => {
- let commit = shard
- .plane
- .metadata()
- .submit_logout_in_process(vsr_client_id, session, request)
- .await
- .ok();
- let _ = reply.try_send(commit);
+ let outcome =
+ submit_logout_local_or_forward(&shard, vsr_client_id, session, request)
+ .await;
+ let _ = reply.try_send(outcome);
}
shard::MetadataSubmit::ClientRequest { request, reply } => {
let committed = match request.try_into_typed::<RoutedRequestHeader>() {
@@ -878,25 +907,13 @@
request = request.header().request,
"dropping client request whose body does not match its own checksum"
);
- let commit = current_metadata_commit(shard);
- let reply = build_deny_reply(
- request.header(),
+ send_deny_reply(
+ shard,
transport_client_id,
- 0,
- commit,
+ request.header(),
error.as_code(),
- );
- if let Err(send_error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %send_error,
- "failed to send request-checksum deny reply"
- );
- }
+ )
+ .await;
return;
}
@@ -910,23 +927,23 @@
let header = *request.header();
if header.operation == Operation::NonReplicated {
- // Auth bypass guard: only `PING` and `GET_CLUSTER_METADATA` are
- // legitimately pre-auth (liveness probe + connection bootstrap
- // metadata). Pre-auth metadata is a narrow topology oracle: it leaks
- // only the advertised-address mapping for networks the caller can
- // already send packets from, and the HTTP mirror of the same read
- // sits behind `Identity`. Every other non-replicated code (`GET_STREAM*`,
- // `GET_TOPIC*`, `GET_STATS`, `POLL_MESSAGES`) reads live state and
- // MUST go through Register first, which binds the acting user the
- // per-op authz gates resolve.
+ // Auth bypass guard: `PING`, the liveness probe, is the only pre-auth
+ // code, on every roster shape. `GET_CLUSTER_METADATA` describes the
+ // private replica network and is not something an unauthenticated
+ // caller gets to read; a client that dialed a backup no longer needs
+ // it to find the leader, because the backup authenticates the login
+ // locally and forwards only the consensus proposal
+ // (`submit_register_local_or_forward`). Every other non-replicated
+ // code MUST go through Register first, which binds the acting user
+ // the per-op authz gates resolve.
let nr_code = u32::from_le_bytes(request.header().reserved[..4].try_into().unwrap());
// Legacy (pre-register) login codes. The server authenticates only via
// the Register handshake (LOGIN_REGISTER / LOGIN_REGISTER_WITH_PAT,
// Operation::Register); the vsr SDK funnels both logins there and never
// emits these. Reject them uniformly with a typed MalformedLogin (the
// SDK maps it to InvalidFormat) before the session gate, so a legacy or
- // foreign client fails fast instead of getting the misleading
- // NoSession eviction the pre-auth guard would send unbound, or the
+ // foreign client fails fast instead of getting the generic
+ // Unauthenticated deny the pre-auth guard would send unbound, or the
// silent empty-ok Reply the bound non-replicated path would send.
if matches!(
nr_code,
@@ -946,14 +963,35 @@
.await;
return;
}
- let allowed_pre_auth = matches!(nr_code, PING_CODE | GET_CLUSTER_METADATA_CODE);
+ let allowed_pre_auth = nr_code == PING_CODE;
if !allowed_pre_auth && sessions.borrow().get_session(transport_client_id).is_none() {
- warn!(
+ // Foreign SDKs still probe `GET_CLUSTER_METADATA` before login
+ // until they are fixed, so that rejection is routine traffic and
+ // logs at debug rather than warn.
+ if nr_code == GET_CLUSTER_METADATA_CODE {
+ debug!(
+ transport_client_id,
+ "denying pre-auth cluster-metadata read with Unauthenticated"
+ );
+ } else {
+ warn!(
+ transport_client_id,
+ code = nr_code,
+ "denying pre-auth non-replicated read with Unauthenticated"
+ );
+ }
+ // A plain deny Reply, not an Eviction: there is no session to
+ // evict, and an Eviction is session-terminal by wire contract,
+ // so SDKs would tear down the very connection their login is
+ // about to use. The status channel carries the error the same
+ // way the request-checksum denial above does.
+ send_unbound_deny_reply(
+ shard,
transport_client_id,
- code = nr_code,
- "rejecting pre-auth non-replicated read with Eviction(NoSession)"
- );
- send_unauthenticated_eviction(shard, transport_client_id).await;
+ request.header(),
+ IggyError::Unauthenticated.as_code(),
+ )
+ .await;
return;
}
handle_non_replicated_request(shard, sessions, system_config, transport_client_id, request)
@@ -977,12 +1015,14 @@
// circuit, the rewrite below overwrites `header.client` with
// `transport_client_id` and dispatches; the request_preflight then
// rejects with `NoSession`/`Fenced` and the failure disappears
- // silently, wedging the SDK until the socket timeout. Reject with
- // the same typed `Eviction(NoSession)` the pre-auth read guard
- // sends: the session is gone, so the client must register again. An
- // empty status-0 Reply is not safe here, because SendMessages is the
- // one replicated operation without a result section, and its decoder
- // would read the empty body as a successful send.
+ // silently, wedging the SDK until the socket timeout. A typed
+ // `Eviction(NoSession)` is right here, unlike the pre-auth read
+ // guard above: a replicated request implies the client believes it
+ // has a session, and that session is gone, so it must register
+ // again. An empty status-0 Reply is not safe here, because
+ // SendMessages is the one replicated operation without a result
+ // section, and its decoder would read the empty body as a
+ // successful send.
warn!(
transport_client_id,
operation = ?header.operation,
@@ -1273,7 +1313,7 @@
operation = ?header.operation,
"partition request with unresolved namespace; replying denied"
);
- send_partition_deny_reply(
+ send_deny_reply(
shard,
transport_client_id,
&header,
@@ -1311,7 +1351,7 @@
operation = ?header.operation,
"partition request denied by authorization; replying with status"
);
- send_partition_deny_reply(shard, transport_client_id, &header, status).await;
+ send_deny_reply(shard, transport_client_id, &header, status).await;
return;
}
// Convergence wait: a CreateTopic commit returns to the client before the
@@ -1333,7 +1373,7 @@
operation = ?header.operation,
"partition request not routable within budget; replying transient"
);
- send_partition_deny_reply(
+ send_deny_reply(
shard,
transport_client_id,
&header,
@@ -1393,8 +1433,7 @@
let code = u32::from_le_bytes(request.header().reserved[CODE_RANGE].try_into().unwrap());
// Acting user and peer address for the read gates below, resolved in one
// connection lookup. `user_id` is `None` only on the pre-auth path
- // (PING / GET_CLUSTER_METADATA), which serves ungated codes; the gated
- // arms fail closed on it.
+ // (PING), which serves ungated codes; the gated arms fail closed on it.
let (user_id, client_address) = sessions.borrow().read_context(transport_client_id);
match code {
PING_CODE => {
@@ -1723,7 +1762,10 @@
}
}
-/// Reject a pre-auth request with a typed `Eviction(NoSession)` frame.
+/// Reject a replicated request from an unbound transport with a typed
+/// `Eviction(NoSession)` frame: the session the client believes it has is
+/// gone, so it must register again. Pre-auth non-replicated reads get a
+/// deny Reply instead (no session exists, so nothing is evicted).
///
/// The SDK's reply decoder maps eviction reasons to typed errors
/// (`NoSession` -> `Unauthenticated`), so clients fail fast with the same
@@ -2386,15 +2428,463 @@
Ok(mapped)
}
+/// Answer a backup's forwarded `Register` from the node it named primary.
+///
+/// Proposes in process, never through [`submit_register_local_or_forward`]:
+/// that is what bounds a forward at one hop. A node that has since lost
+/// primaryship answers `NotPrimary`, and the origin's client replays against
+/// whichever node it names next.
+#[allow(clippy::future_not_send)]
+async fn answer_forwarded_register<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ vsr_client_id: u128,
+ user_id: u32,
+ nonce: u128,
+ origin_replica: u8,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let Some((cluster, view, replica)) = shard
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .map(|consensus| (consensus.cluster(), consensus.view(), consensus.replica()))
+ else {
+ warn!("ForwardedRegister submit reached a shard without metadata consensus");
+ return;
+ };
+ let bound = shard
+ .plane
+ .metadata()
+ .submit_register_in_process(vsr_client_id, user_id)
+ .await;
+ // `view` predates the await above, which parks with no deadline, so the
+ // sealed value can be stale by send time. The origin routes the result by
+ // `(nonce, client)` alone; this field must never become a freshness fence.
+ let result =
+ build_forward_register_result_message(cluster, view, replica, vsr_client_id, nonce, &bound);
+ if let Err(error) = shard
+ .bus
+ .send_to_replica(origin_replica, result.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ origin_replica,
+ error = %error,
+ "failed to answer a forwarded register"
+ );
+ }
+}
+
+/// How long a login waits for the primary's verdict on a forwarded register.
+///
+/// Expiry does NOT prove the peer or the frame was lost. The primary answers
+/// only once the proposal resolves, and its own submit parks with no deadline:
+/// a primary that is not caught up, or whose pipeline is full, absorbs the
+/// register into its request queue and answers when that drains. So a slow but
+/// healthy primary commits the register after this node has stopped waiting,
+/// which is why expiry surfaces as `TransientNotCommitted` rather than the
+/// not-accepted flavor.
+///
+/// The budget stays well under the SDK's response-read timeout on purpose: the
+/// client only replays a login while it is still reading, so a longer wait
+/// here turns a transient into a torn-down socket.
+const FORWARD_SUBMIT_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Run the `Register` proposal for a login this node has already
+/// authenticated, wherever the metadata primary currently is. Shard 0 only.
+///
+/// A client may dial any node in the cluster. Credentials verify against the
+/// replicated users table, which every node holds, so the whole login except
+/// the consensus proposal already works on a backup. Only the verified
+/// identity crosses the replica interconnect -- never the client's frame and
+/// never its credentials -- and the session bind, the reply, and the
+/// connection all stay on the node the client dialed.
+///
+/// The hop does not move any credential decision:
+/// - `verify_login_credentials` reads the backup's applied replicated user
+/// state.
+/// - `verify_pat_credentials` reads the same state, so a PAT minted on the
+/// primary that has not replicated here yet is refused until it does.
+/// Fail-closed on purpose, the same parity the HTTP forward keeps: it too
+/// answers 401 until replication catches up rather than relaying an
+/// unverified bearer.
+/// - `ClientIdOwnedByAnotherUser` stays a decision of the caught-up primary
+/// and round-trips as a terminal refusal.
+///
+/// Verification is point-in-time on the backup. A password change, PAT
+/// revocation, or user deactivation committed on the primary but not yet
+/// applied on the backup can therefore admit a login during the backup's apply
+/// lag. The forward cannot complete while the backup is partitioned from the
+/// primary, which bounds this to a connected replica's replication lag. This
+/// is the same stale-read window as the existing HTTP forward.
+///
+/// The session binds here before this node applies the commit locally. That
+/// is the window a primary-side login already has against every other node's
+/// apply lag, not a new one.
+#[allow(clippy::future_not_send)]
+async fn submit_register_local_or_forward<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ vsr_client_id: u128,
+ user_id: u32,
+) -> Result<BoundSession, MetadataSubmitError>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let Some(consensus) = shard.plane.metadata().consensus.as_ref() else {
+ return Err(MetadataSubmitError::NotPrimary);
+ };
+ let (cluster, view, self_replica) =
+ (consensus.cluster(), consensus.view(), consensus.replica());
+ let target = consensus.primary_index(view);
+ // Forward only as a healthy backup. Everything else answers locally: the
+ // in-process submit proposes when this node is the serving primary and
+ // re-derives `NotPrimary` otherwise -- mid view change there is nobody to
+ // forward to (the node the view names has not finished taking over, the
+ // SDK replays once it settles), and the view's own primary under state
+ // transfer has nowhere to forward to and nothing to commit yet.
+ if target == self_replica || !consensus.is_normal() {
+ return shard
+ .plane
+ .metadata()
+ .submit_register_in_process(vsr_client_id, user_id)
+ .await;
+ }
+
+ let nonce = shard.next_forward_nonce(self_replica);
+ let (reply, outcome) = shard::channel::<ForwardRegisterResultHeader>(1);
+ shard.park_register_forward(nonce, vsr_client_id, reply);
+ let forward =
+ build_forward_register_message(cluster, view, self_replica, vsr_client_id, nonce, user_id);
+ if let Err(error) = shard
+ .bus
+ .send_to_replica(target, forward.into_generic().into_frozen())
+ .await
+ {
+ shard.cancel_register_forward(nonce, vsr_client_id);
+ warn!(
+ target,
+ error = %error,
+ "failed to forward register to the metadata primary"
+ );
+ return Err(MetadataSubmitError::PrimaryUnreachable);
+ }
+
+ match shard::bus_timeout(&shard.bus, FORWARD_SUBMIT_TIMEOUT, outcome.recv()).await {
+ Some(Ok(result)) => forward_register_result(&result),
+ // Shard-0 teardown dropped the sender without answering.
+ Some(Err(_)) => Err(MetadataSubmitError::Canceled),
+ None => {
+ shard.cancel_register_forward(nonce, vsr_client_id);
+ warn!(target, "forwarded register timed out");
+ Err(MetadataSubmitError::ForwardTimedOut)
+ }
+ }
+}
+
+/// The primary's verdict, back in the vocabulary the login path speaks.
+const fn forward_register_result(
+ result: &ForwardRegisterResultHeader,
+) -> Result<BoundSession, MetadataSubmitError> {
+ match result.outcome {
+ ForwardRegisterOutcome::Ok => Ok(BoundSession {
+ epoch: result.epoch,
+ watermark: result.watermark,
+ }),
+ ForwardRegisterOutcome::NotPrimary => Err(MetadataSubmitError::NotPrimary),
+ ForwardRegisterOutcome::NotCaughtUp => Err(MetadataSubmitError::NotCaughtUp),
+ ForwardRegisterOutcome::PipelineFull => Err(MetadataSubmitError::PipelineFull),
+ ForwardRegisterOutcome::InProgress => Err(MetadataSubmitError::InProgress),
+ ForwardRegisterOutcome::Canceled => Err(MetadataSubmitError::Canceled),
+ ForwardRegisterOutcome::ClientIdOwnedByAnotherUser => {
+ Err(MetadataSubmitError::ClientIdOwnedByAnotherUser)
+ }
+ }
+}
+
+/// Inverse of [`forward_register_result`], for the answering primary.
+const fn forward_register_outcome(
+ bound: &Result<BoundSession, MetadataSubmitError>,
+) -> (BoundSession, ForwardRegisterOutcome) {
+ let zero = BoundSession {
+ epoch: 0,
+ watermark: 0,
+ };
+ match bound {
+ Ok(bound) => (*bound, ForwardRegisterOutcome::Ok),
+ Err(MetadataSubmitError::NotPrimary) => (zero, ForwardRegisterOutcome::NotPrimary),
+ Err(MetadataSubmitError::NotCaughtUp) => (zero, ForwardRegisterOutcome::NotCaughtUp),
+ Err(MetadataSubmitError::PipelineFull) => (zero, ForwardRegisterOutcome::PipelineFull),
+ Err(MetadataSubmitError::InProgress) => (zero, ForwardRegisterOutcome::InProgress),
+ Err(MetadataSubmitError::ClientIdOwnedByAnotherUser) => {
+ (zero, ForwardRegisterOutcome::ClientIdOwnedByAnotherUser)
+ }
+ // `MetadataSubmitError` is `#[non_exhaustive]`. Every variant but the
+ // ownership refusal is transient by contract, and `Canceled` is the
+ // transient answer that claims nothing beyond "retry".
+ Err(_) => (zero, ForwardRegisterOutcome::Canceled),
+ }
+}
+
+#[allow(clippy::cast_possible_truncation)]
+fn build_forward_register_message(
+ cluster: u128,
+ view: u32,
+ replica: u8,
+ client: u128,
+ nonce: u128,
+ user_id: u32,
+) -> Message<ForwardRegisterHeader> {
+ Message::<ForwardRegisterHeader>::new(HEADER_SIZE).transmute_header(
+ |_, header: &mut ForwardRegisterHeader| {
+ header.command = Command2::ForwardRegister;
+ header.cluster = cluster;
+ header.view = view;
+ header.replica = replica;
+ header.client = client;
+ header.nonce = nonce;
+ header.user_id = user_id;
+ header.size = HEADER_SIZE as u32;
+ header.seal();
+ },
+ )
+}
+
+#[allow(clippy::cast_possible_truncation)]
+fn build_forward_register_result_message(
+ cluster: u128,
+ view: u32,
+ replica: u8,
+ client: u128,
+ nonce: u128,
+ bound: &Result<BoundSession, MetadataSubmitError>,
+) -> Message<ForwardRegisterResultHeader> {
+ let (session, outcome) = forward_register_outcome(bound);
+ Message::<ForwardRegisterResultHeader>::new(HEADER_SIZE).transmute_header(
+ |_, header: &mut ForwardRegisterResultHeader| {
+ header.command = Command2::ForwardRegisterResult;
+ header.cluster = cluster;
+ header.view = view;
+ header.replica = replica;
+ header.client = client;
+ header.nonce = nonce;
+ header.epoch = session.epoch;
+ header.watermark = session.watermark;
+ header.outcome = outcome;
+ header.size = HEADER_SIZE as u32;
+ header.seal();
+ },
+ )
+}
+
+/// Answer a backup's forwarded Logout from the node it named primary.
+#[allow(clippy::future_not_send)]
+async fn answer_forwarded_logout<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ vsr_client_id: u128,
+ session: u64,
+ request: u64,
+ nonce: u128,
+ origin_replica: u8,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let Some((cluster, view, replica)) = shard
+ .plane
+ .metadata()
+ .consensus
+ .as_ref()
+ .map(|consensus| (consensus.cluster(), consensus.view(), consensus.replica()))
+ else {
+ warn!("ForwardedLogout submit reached a shard without metadata consensus");
+ return;
+ };
+ let outcome = shard
+ .plane
+ .metadata()
+ .submit_logout_in_process(vsr_client_id, session, request)
+ .await;
+ let result =
+ build_forward_logout_result_message(cluster, view, replica, vsr_client_id, nonce, &outcome);
+ if let Err(error) = shard
+ .bus
+ .send_to_replica(origin_replica, result.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ origin_replica,
+ error = %error,
+ "failed to answer a forwarded logout"
+ );
+ }
+}
+
+/// Commit a Logout locally when this node is primary, otherwise forward it
+/// once to the primary named by the current normal view.
+#[allow(clippy::future_not_send)]
+async fn submit_logout_local_or_forward<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ vsr_client_id: u128,
+ session: u64,
+ request: u64,
+) -> Result<u64, MetadataSubmitError>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let Some(consensus) = shard.plane.metadata().consensus.as_ref() else {
+ return Err(MetadataSubmitError::NotPrimary);
+ };
+ let (cluster, view, self_replica) =
+ (consensus.cluster(), consensus.view(), consensus.replica());
+ let target = consensus.primary_index(view);
+ if target == self_replica || !consensus.is_normal() {
+ return shard
+ .plane
+ .metadata()
+ .submit_logout_in_process(vsr_client_id, session, request)
+ .await;
+ }
+
+ let nonce = shard.next_forward_nonce(self_replica);
+ let (reply, outcome) = shard::channel::<ForwardLogoutResultHeader>(1);
+ shard.park_logout_forward(nonce, vsr_client_id, reply);
+ let forward = build_forward_logout_message(
+ cluster,
+ view,
+ self_replica,
+ vsr_client_id,
+ nonce,
+ session,
+ request,
+ );
+ if let Err(error) = shard
+ .bus
+ .send_to_replica(target, forward.into_generic().into_frozen())
+ .await
+ {
+ shard.cancel_logout_forward(nonce, vsr_client_id);
+ warn!(
+ target,
+ error = %error,
+ "failed to forward logout to the metadata primary"
+ );
+ return Err(MetadataSubmitError::PrimaryUnreachable);
+ }
+
+ match shard::bus_timeout(&shard.bus, FORWARD_SUBMIT_TIMEOUT, outcome.recv()).await {
+ Some(Ok(result)) => forward_logout_result(&result),
+ Some(Err(_)) => Err(MetadataSubmitError::Canceled),
+ None => {
+ shard.cancel_logout_forward(nonce, vsr_client_id);
+ warn!(target, "forwarded logout timed out");
+ Err(MetadataSubmitError::ForwardTimedOut)
+ }
+ }
+}
+
+const fn forward_logout_result(
+ result: &ForwardLogoutResultHeader,
+) -> Result<u64, MetadataSubmitError> {
+ match result.outcome {
+ ForwardLogoutOutcome::Ok => Ok(result.commit),
+ ForwardLogoutOutcome::NotPrimary => Err(MetadataSubmitError::NotPrimary),
+ ForwardLogoutOutcome::PipelineFull => Err(MetadataSubmitError::PipelineFull),
+ ForwardLogoutOutcome::InProgress => Err(MetadataSubmitError::InProgress),
+ ForwardLogoutOutcome::Canceled => Err(MetadataSubmitError::Canceled),
+ }
+}
+
+const fn forward_logout_outcome(
+ outcome: &Result<u64, MetadataSubmitError>,
+) -> (u64, ForwardLogoutOutcome) {
+ match outcome {
+ Ok(commit) => (*commit, ForwardLogoutOutcome::Ok),
+ Err(MetadataSubmitError::NotPrimary) => (0, ForwardLogoutOutcome::NotPrimary),
+ Err(MetadataSubmitError::PipelineFull) => (0, ForwardLogoutOutcome::PipelineFull),
+ Err(MetadataSubmitError::InProgress) => (0, ForwardLogoutOutcome::InProgress),
+ Err(_) => (0, ForwardLogoutOutcome::Canceled),
+ }
+}
+
+#[allow(clippy::cast_possible_truncation, clippy::too_many_arguments)]
+fn build_forward_logout_message(
+ cluster: u128,
+ view: u32,
+ replica: u8,
+ client: u128,
+ nonce: u128,
+ session: u64,
+ request: u64,
+) -> Message<ForwardLogoutHeader> {
+ Message::<ForwardLogoutHeader>::new(HEADER_SIZE).transmute_header(
+ |_, header: &mut ForwardLogoutHeader| {
+ header.command = Command2::ForwardLogout;
+ header.cluster = cluster;
+ header.view = view;
+ header.replica = replica;
+ header.client = client;
+ header.nonce = nonce;
+ header.session = session;
+ header.request = request;
+ header.size = HEADER_SIZE as u32;
+ header.seal();
+ },
+ )
+}
+
+#[allow(clippy::cast_possible_truncation)]
+fn build_forward_logout_result_message(
+ cluster: u128,
+ view: u32,
+ replica: u8,
+ client: u128,
+ nonce: u128,
+ result: &Result<u64, MetadataSubmitError>,
+) -> Message<ForwardLogoutResultHeader> {
+ let (commit, outcome) = forward_logout_outcome(result);
+ Message::<ForwardLogoutResultHeader>::new(HEADER_SIZE).transmute_header(
+ |_, header: &mut ForwardLogoutResultHeader| {
+ header.command = Command2::ForwardLogoutResult;
+ header.cluster = cluster;
+ header.view = view;
+ header.replica = replica;
+ header.client = client;
+ header.nonce = nonce;
+ header.commit = commit;
+ header.outcome = outcome;
+ header.size = HEADER_SIZE as u32;
+ header.seal();
+ },
+ )
+}
+
/// Run the consensus `Register` proposal on the metadata owner (shard 0)
/// and return the committed session.
///
/// Credential verification and session binding stay on the calling (home)
/// shard -- only this consensus step must execute where the metadata
-/// consensus group lives. On shard 0 it calls in-process directly; on a
-/// peer it forwards a [`shard::MetadataSubmit`] to shard 0 and awaits the
-/// committed op. A dropped reply (shard-0 inbox full / shutdown) maps to a
-/// transient `Canceled`, which the caller wraps so the SDK replays.
+/// consensus group lives. On shard 0 it goes straight to
+/// [`submit_register_local_or_forward`]; on a peer it forwards a
+/// [`shard::MetadataSubmit`] to shard 0 and awaits the committed op. A dropped
+/// reply (shard-0 inbox full / shutdown) maps to a transient `Canceled`, which
+/// the caller wraps so the SDK replays.
#[allow(clippy::future_not_send)]
pub(crate) async fn submit_register_on_owner<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
@@ -2409,11 +2899,7 @@
SB: SuperblockStore + 'static,
{
if shard.id == 0 {
- return shard
- .plane
- .metadata()
- .submit_register_in_process(vsr_client_id, user_id)
- .await;
+ return submit_register_local_or_forward(shard, vsr_client_id, user_id).await;
}
let (reply, rx) = shard::channel::<Result<BoundSession, MetadataSubmitError>>(1);
shard.forward_metadata_submit(shard::MetadataSubmit::Register {
@@ -2444,23 +2930,18 @@
SB: SuperblockStore + 'static,
{
if shard.id == 0 {
- return shard
- .plane
- .metadata()
- .submit_logout_in_process(vsr_client_id, session, request)
- .await;
+ return submit_logout_local_or_forward(shard, vsr_client_id, session, request).await;
}
- let (reply, rx) = shard::channel::<Option<u64>>(1);
+ let (reply, rx) = shard::channel::<Result<u64, MetadataSubmitError>>(1);
shard.forward_metadata_submit(shard::MetadataSubmit::Logout {
vsr_client_id,
session,
request,
reply,
});
- match rx.recv().await {
- Ok(Some(commit)) => Ok(commit),
- _ => Err(MetadataSubmitError::Canceled),
- }
+ rx.recv()
+ .await
+ .map_or(Err(MetadataSubmitError::Canceled), |outcome| outcome)
}
/// Handle a client `DeleteSegments`: resolve the requested count to an offset
@@ -2850,7 +3331,7 @@
vsr_client_id,
session,
commit,
- IggyError::TransientNotAccepted.as_code(),
+ transient_logout_code(&error).as_code(),
);
if let Err(send_error) = shard
.bus
@@ -2883,6 +3364,18 @@
}
}
+/// Preserve the client identity when a Logout may already have entered the
+/// primary's pipeline. Moving an unknown-outcome replay to another connection
+/// could race a later Register and obscure whether the old epoch was removed.
+const fn transient_logout_code(error: &MetadataSubmitError) -> IggyError {
+ match error {
+ MetadataSubmitError::ForwardTimedOut
+ | MetadataSubmitError::InProgress
+ | MetadataSubmitError::Canceled => IggyError::TransientNotCommitted,
+ _ => IggyError::TransientNotAccepted,
+ }
+}
+
fn ensure_transport_connection<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
sessions: &Rc<RefCell<SessionManager>>,
@@ -3165,7 +3658,7 @@
IggyShard, LifecycleFrame, PartitionConsensusConfig, ReconcileOp, ReplicaTopology,
ShardFrame, ShardIdentity, shard_channel,
};
- use std::cell::RefCell;
+ use std::cell::{Cell, RefCell};
use std::future::Future;
use std::mem::size_of;
use std::rc::Rc;
@@ -3174,13 +3667,36 @@
type TestShard = IggyShard<SpyBus, PrepareJournal, IggySnapshot, TestMux, PapayaShardsTable>;
/// `(target client id, reply frame bytes)` per `send_to_client` call.
type RecordedReplies = Rc<RefCell<Vec<(u128, Vec<u8>)>>>;
+ /// `(target replica id, frame bytes)` per `send_to_replica` call.
+ type RecordedReplicaSends = Rc<RefCell<Vec<(u8, Vec<u8>)>>>;
- /// Records every client-bound reply (target id + frame bytes) instead of
- /// writing to a socket; everything else is a no-op. The two `ShellBus`
- /// halves are stubbed.
+ /// Records every client-bound reply and replica-bound frame (target +
+ /// bytes) instead of writing to a socket; everything else is a no-op. The
+ /// two `ShellBus` halves are stubbed.
#[derive(Debug, Clone, Default)]
struct SpyBus {
client_replies: RecordedReplies,
+ replica_sends: RecordedReplicaSends,
+ /// Resolve [`MessageBus::sleep`] immediately instead of arming a real
+ /// timer. The register forward is the only path here that races a
+ /// timer, and its budget is five seconds -- too long to wait for in a
+ /// unit test, and too long to shorten in production for one.
+ instant_timers: Rc<Cell<bool>>,
+ }
+
+ impl SpyBus {
+ /// Decode the single frame this bus sent to a replica.
+ fn sole_replica_send<H: iggy_binary_protocol::ConsensusHeader>(&self) -> (u8, H) {
+ let sends = self.replica_sends.borrow();
+ assert_eq!(sends.len(), 1, "expected exactly one replica-bound frame");
+ let (target, frame) = &sends[0];
+ let mut aligned = server_common::iobuf::Owned::<MESSAGE_ALIGN>::zeroed(frame.len());
+ aligned.as_mut_slice().copy_from_slice(frame);
+ let header =
+ *bytemuck::checked::try_from_bytes::<H>(&aligned.as_slice()[..size_of::<H>()])
+ .expect("replica frame decodes into the expected header");
+ (*target, header)
+ }
}
#[allow(clippy::future_not_send)]
@@ -3198,11 +3714,19 @@
}
async fn send_to_replica(
&self,
- _replica: u8,
- _data: Frozen<MESSAGE_ALIGN>,
+ replica: u8,
+ data: Frozen<MESSAGE_ALIGN>,
) -> Result<(), SendError> {
+ self.replica_sends
+ .borrow_mut()
+ .push((replica, data.as_slice().to_vec()));
Ok(())
}
+ async fn sleep(&self, duration: std::time::Duration) {
+ if !self.instant_timers.get() {
+ compio::time::sleep(duration).await;
+ }
+ }
fn set_connection_lost_fn(&self, _f: ConnectionLostFn) {}
fn set_replica_forward_fn(&self, _f: ReplicaForwardFn) {}
fn set_client_forward_fn(&self, _f: ClientForwardFn) {}
@@ -3246,6 +3770,57 @@
fn set_client_connection_lost_fn(&self, _f: ClientConnectionLostFn) {}
}
+ /// Consensus incarnations standing for two successive boots of one node, as
+ /// far apart as the random draw at bootstrap makes them.
+ const FIRST_BOOT: u128 = 0x5EED_0001;
+ const SECOND_BOOT: u128 = 0x9E37_79B9_7F4A_7C15;
+
+ /// Shard 0 carrying a metadata consensus group of `replica_count`
+ /// replicas in which this node is `replica`. No journal: every test using
+ /// it either never proposes, or is a backup that cannot.
+ ///
+ /// `incarnation` stands for one boot of this node: the shard seeds its
+ /// forward-nonce counter from it, so passing a different value models a
+ /// restart.
+ fn test_shard(bus: &SpyBus, replica: u8, replica_count: u8, incarnation: u128) -> TestShard {
+ let consensus = VsrConsensus::new(
+ 1,
+ replica,
+ replica_count,
+ server_common::sharding::METADATA_GROUP,
+ bus.clone(),
+ LocalPipeline::new(),
+ );
+ consensus.set_incarnation(incarnation);
+ consensus.init();
+ let metadata: IggyMetadata<_, PrepareJournal, IggySnapshot, TestMux> =
+ IggyMetadata::new(Some(consensus), None, None, None, TestMux::default(), None);
+ let partitions = IggyPartitions::new(
+ ShardId::new(0),
+ PartitionsConfig {
+ messages_required_to_save: 1,
+ size_of_messages_required_to_save: iggy_common::IggyByteSize::from(1024_u64),
+ enforce_fsync: false,
+ validate_checksum: true,
+ segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
+ preallocate_segments: false,
+ encryptor: None,
+ },
+ );
+ TestShard::without_inbox(
+ ShardIdentity::new(0, "dispatch-test".to_string()),
+ bus.clone(),
+ metadata,
+ partitions,
+ PapayaShardsTable::new(),
+ PartitionConsensusConfig::new(
+ 1,
+ ReplicaTopology::new(replica, replica_count),
+ bus.clone(),
+ ),
+ )
+ }
+
/// Minimal committed `Register` reply for `ClientTable::commit_register`
/// (reads only `client` and `commit`).
fn register_reply(client: u128, session: u64) -> Message<ReplyHeader> {
@@ -3768,6 +4343,493 @@
);
}
+ /// A backup's login: it forwards the register it authenticated to the
+ /// view's primary and completes on the primary's verdict, with the whole
+ /// round trip going through the real shard ingest arm.
+ #[compio::test]
+ async fn backup_forwards_register_and_completes_on_the_primary_verdict() {
+ const CLIENT: u128 = 0xCAFE;
+ const USER: u32 = 7;
+ const EPOCH: u64 = 41;
+ const WATERMARK: u64 = 9;
+
+ let bus = SpyBus::default();
+ // Replica 1 of 3, view 0: `primary_index(0)` is replica 0.
+ let shard = Rc::new(test_shard(&bus, 1, 3, FIRST_BOOT));
+ let login = {
+ let shard = Rc::clone(&shard);
+ compio::runtime::spawn(async move {
+ submit_register_local_or_forward(&shard, CLIENT, USER).await
+ })
+ };
+ await_forward(&bus).await;
+ let (target, forward) = bus.sole_replica_send::<ForwardRegisterHeader>();
+ assert_eq!(target, 0, "forward must address the view's primary");
+ assert_eq!(forward.command, Command2::ForwardRegister);
+ assert_eq!(forward.client, CLIENT);
+ assert_eq!(
+ forward.user_id, USER,
+ "the forwarded identity is the payload"
+ );
+ assert_eq!(forward.replica, 1, "the origin names itself for the answer");
+ assert_ne!(forward.nonce, 0);
+ assert_eq!(forward.verify_frame(), Ok(()), "the frame must be sealed");
+ assert_eq!(forward.validate(), Ok(()));
+
+ shard
+ .on_message(forward_register_result(
+ &forward,
+ ForwardRegisterOutcome::Ok,
+ EPOCH,
+ WATERMARK,
+ ))
+ .await;
+ assert_eq!(
+ login.await.expect("the login task ran to completion"),
+ Ok(BoundSession {
+ epoch: EPOCH,
+ watermark: WATERMARK,
+ })
+ );
+ }
+
+ #[compio::test]
+ async fn backup_forwards_logout_and_completes_on_the_primary_verdict() {
+ const CLIENT: u128 = 0xCAFE;
+ const SESSION: u64 = 41;
+ const REQUEST: u64 = 9;
+ const COMMIT: u64 = 42;
+
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 1, 3, FIRST_BOOT));
+ let logout = {
+ let shard = Rc::clone(&shard);
+ compio::runtime::spawn(async move {
+ submit_logout_local_or_forward(&shard, CLIENT, SESSION, REQUEST).await
+ })
+ };
+ await_forward(&bus).await;
+ let (target, forward) = bus.sole_replica_send::<ForwardLogoutHeader>();
+ assert_eq!(target, 0, "forward must address the view's primary");
+ assert_eq!(forward.command, Command2::ForwardLogout);
+ assert_eq!(forward.client, CLIENT);
+ assert_eq!(forward.session, SESSION);
+ assert_eq!(forward.request, REQUEST);
+ assert_eq!(forward.replica, 1);
+ assert_ne!(forward.nonce, 0);
+ assert_eq!(forward.verify_frame(), Ok(()));
+ assert_eq!(forward.validate(), Ok(()));
+
+ shard
+ .on_message(forward_logout_result_message(&forward, &Ok(COMMIT)))
+ .await;
+ assert_eq!(
+ logout.await.expect("the logout task ran to completion"),
+ Ok(COMMIT)
+ );
+ }
+
+ #[compio::test]
+ async fn unanswered_logout_forward_times_out_and_clears_the_waiter() {
+ let bus = SpyBus::default();
+ bus.instant_timers.set(true);
+ let shard = Rc::new(test_shard(&bus, 1, 3, FIRST_BOOT));
+
+ let outcome = submit_logout_local_or_forward(&shard, 0xCAFE, 41, 9).await;
+ assert_eq!(outcome, Err(MetadataSubmitError::ForwardTimedOut));
+
+ let (_, forward) = bus.sole_replica_send::<ForwardLogoutHeader>();
+ shard
+ .on_message(forward_logout_result_message(&forward, &Ok(42)))
+ .await;
+ }
+
+ #[test]
+ fn unknown_logout_outcomes_pin_the_session() {
+ for error in [
+ MetadataSubmitError::ForwardTimedOut,
+ MetadataSubmitError::InProgress,
+ MetadataSubmitError::Canceled,
+ ] {
+ assert_eq!(
+ transient_logout_code(&error),
+ IggyError::TransientNotCommitted
+ );
+ }
+ for error in [
+ MetadataSubmitError::NotPrimary,
+ MetadataSubmitError::PipelineFull,
+ MetadataSubmitError::PrimaryUnreachable,
+ ] {
+ assert_eq!(
+ transient_logout_code(&error),
+ IggyError::TransientNotAccepted
+ );
+ }
+ }
+
+ /// The ownership refusal is the one terminal verdict, and it has to stay
+ /// terminal across the hop or the SDK replays a login that cannot succeed.
+ #[compio::test]
+ async fn forwarded_register_keeps_the_ownership_refusal_terminal() {
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 1, 3, FIRST_BOOT));
+ let login = {
+ let shard = Rc::clone(&shard);
+ compio::runtime::spawn(async move {
+ submit_register_local_or_forward(&shard, 0xCAFE, 7).await
+ })
+ };
+ await_forward(&bus).await;
+ let (_, forward) = bus.sole_replica_send::<ForwardRegisterHeader>();
+ shard
+ .on_message(forward_register_result(
+ &forward,
+ ForwardRegisterOutcome::ClientIdOwnedByAnotherUser,
+ 0,
+ 0,
+ ))
+ .await;
+ let error = login
+ .await
+ .expect("the login task ran to completion")
+ .expect_err("the refusal must surface");
+ assert_eq!(error, MetadataSubmitError::ClientIdOwnedByAnotherUser);
+ assert!(!error.is_transient(), "the refusal must stay terminal");
+ }
+
+ /// A primary that never answers must not strand the login or leak its
+ /// parked entry; the client gets a transient failure and replays.
+ #[compio::test]
+ async fn unanswered_forward_times_out_and_clears_the_parked_login() {
+ let bus = SpyBus::default();
+ bus.instant_timers.set(true);
+ let shard = Rc::new(test_shard(&bus, 1, 3, FIRST_BOOT));
+
+ let outcome = submit_register_local_or_forward(&shard, 0xCAFE, 7).await;
+ assert_eq!(outcome, Err(MetadataSubmitError::ForwardTimedOut));
+ assert!(
+ outcome.unwrap_err().is_transient(),
+ "a lost answer is replayable"
+ );
+
+ // The parked entry is gone: the answer that arrives late finds nothing
+ // and is dropped rather than completing a login nobody is waiting on.
+ let (_, forward) = bus.sole_replica_send::<ForwardRegisterHeader>();
+ shard
+ .on_message(forward_register_result(
+ &forward,
+ ForwardRegisterOutcome::Ok,
+ 41,
+ 0,
+ ))
+ .await;
+ }
+
+ /// The reply frame is where an unknown outcome has to be told apart from a
+ /// refusal: a forward that timed out may still commit, so the client must
+ /// replay under the same client id instead of failing over under a fresh
+ /// one. A verdict that refused the register carries no such doubt.
+ #[compio::test]
+ async fn transient_login_reply_marks_a_timed_out_forward_not_committed() {
+ const TRANSPORT: u128 = 91;
+ const VSR_CLIENT: u128 = 0xCAFE;
+ const RESULT_OFFSET: usize = size_of::<ReplyHeader>() + 8;
+
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 1, 3, FIRST_BOOT));
+ let request = request_message(Operation::Register, VSR_CLIENT, 0, 0, &[]);
+
+ for (submit_error, expected) in [
+ (
+ MetadataSubmitError::ForwardTimedOut,
+ IggyError::TransientNotCommitted,
+ ),
+ (
+ MetadataSubmitError::NotPrimary,
+ IggyError::TransientNotAccepted,
+ ),
+ ] {
+ let error = LoginRegisterError::Transient(submit_error);
+ surface_login_failure(&shard, TRANSPORT, request.header(), &error).await;
+
+ let replies = bus.client_replies.borrow();
+ assert_eq!(replies.len(), 1, "a transient login must answer a frame");
+ let (client, frame) = &replies[0];
+ assert_eq!(*client, TRANSPORT, "reply must target the transport id");
+ let result =
+ u32::from_le_bytes(frame[RESULT_OFFSET..RESULT_OFFSET + 4].try_into().unwrap());
+ assert_eq!(result, expected.as_code(), "{error} must reply {expected}");
+ drop(replies);
+ bus.client_replies.borrow_mut().clear();
+ }
+ }
+
+ /// A restart must not re-mint the nonce sequence of the boot before it. The
+ /// nonce is never persisted, and an answer to a pre-restart forward can
+ /// still be in flight: routed by a repeated nonce it would confirm a login
+ /// the cluster never committed, with another client's epoch.
+ #[compio::test]
+ async fn a_restart_moves_the_forward_nonce_sequence() {
+ assert_ne!(
+ first_forward_nonce(FIRST_BOOT).await,
+ first_forward_nonce(SECOND_BOOT).await,
+ "each boot must start its nonce sequence somewhere the other did not"
+ );
+ }
+
+ /// Seeding the counter from the incarnation means it can start one step
+ /// short of wrapping, and a zero nonce is a frame every replica rejects.
+ #[compio::test]
+ async fn wrapping_forward_nonce_counter_skips_zero() {
+ let nonce = first_forward_nonce(u128::from(u64::MAX)).await;
+ assert_ne!(
+ nonce & u128::from(u64::MAX),
+ 0,
+ "a counter that wrapped must not contribute a zero nonce half"
+ );
+ }
+
+ /// An answer echoing a client the nonce was never parked for must neither
+ /// complete that login nor evict it, since a repeated nonce is exactly what
+ /// a late cross-boot answer carries.
+ #[compio::test]
+ async fn forward_result_for_another_client_leaves_the_login_parked() {
+ const CLIENT: u128 = 0xCAFE;
+ const EPOCH: u64 = 41;
+ const WATERMARK: u64 = 9;
+ const FOREIGN_EPOCH: u64 = 77;
+
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 1, 3, FIRST_BOOT));
+ let login = {
+ let shard = Rc::clone(&shard);
+ compio::runtime::spawn(async move {
+ submit_register_local_or_forward(&shard, CLIENT, 7).await
+ })
+ };
+ await_forward(&bus).await;
+ let (_, forward) = bus.sole_replica_send::<ForwardRegisterHeader>();
+
+ let mut foreign = forward;
+ foreign.client = CLIENT + 1;
+ shard
+ .on_message(forward_register_result(
+ &foreign,
+ ForwardRegisterOutcome::Ok,
+ FOREIGN_EPOCH,
+ 0,
+ ))
+ .await;
+ shard
+ .on_message(forward_register_result(
+ &forward,
+ ForwardRegisterOutcome::Ok,
+ EPOCH,
+ WATERMARK,
+ ))
+ .await;
+ assert_eq!(
+ login.await.expect("the login task ran to completion"),
+ Ok(BoundSession {
+ epoch: EPOCH,
+ watermark: WATERMARK,
+ }),
+ "the login must bind the epoch addressed to it, and must still be \
+ parked to receive it"
+ );
+ }
+
+ /// A node that is primary itself never forwards -- that is what bounds a
+ /// forward at one hop.
+ #[compio::test]
+ async fn primary_proposes_locally_instead_of_forwarding() {
+ let bus = SpyBus::default();
+ // Replica 0 of 3, view 0: this node IS the primary.
+ let shard = Rc::new(test_shard(&bus, 0, 3, FIRST_BOOT));
+
+ // No journal on the test shard, so the proposal cannot commit; what
+ // matters is that nothing left over the interconnect.
+ let _ = compio::time::timeout(
+ Duration::from_millis(50),
+ submit_register_local_or_forward(&shard, 0xCAFE, 7),
+ )
+ .await;
+ assert!(
+ bus.replica_sends.borrow().is_empty(),
+ "a primary must propose in process"
+ );
+ }
+
+ /// The nonce a shard booted at `incarnation` stamps on its first forward.
+ /// Nobody answers, so the login abandons on the instant timer; the frame it
+ /// left on the bus is what the caller is after.
+ async fn first_forward_nonce(incarnation: u128) -> u128 {
+ let bus = SpyBus::default();
+ bus.instant_timers.set(true);
+ let shard = Rc::new(test_shard(&bus, 1, 3, incarnation));
+ let outcome = submit_register_local_or_forward(&shard, 0xCAFE, 7).await;
+ assert_eq!(outcome, Err(MetadataSubmitError::ForwardTimedOut));
+ bus.sole_replica_send::<ForwardRegisterHeader>().1.nonce
+ }
+
+ /// Let a spawned login run until it has parked on the primary's answer.
+ async fn await_forward(bus: &SpyBus) {
+ for _ in 0..1000 {
+ if !bus.replica_sends.borrow().is_empty() {
+ return;
+ }
+ compio::time::sleep(Duration::from_millis(1)).await;
+ }
+ panic!("the login never forwarded a register");
+ }
+
+ /// A sealed `ForwardRegisterResult` addressed to `forward`'s nonce.
+ fn forward_register_result(
+ forward: &ForwardRegisterHeader,
+ outcome: ForwardRegisterOutcome,
+ epoch: u64,
+ watermark: u64,
+ ) -> Message<GenericHeader> {
+ let bound = match outcome {
+ ForwardRegisterOutcome::Ok => Ok(BoundSession { epoch, watermark }),
+ ForwardRegisterOutcome::ClientIdOwnedByAnotherUser => {
+ Err(MetadataSubmitError::ClientIdOwnedByAnotherUser)
+ }
+ _ => Err(MetadataSubmitError::NotPrimary),
+ };
+ build_forward_register_result_message(
+ forward.cluster,
+ forward.view,
+ 0,
+ forward.client,
+ forward.nonce,
+ &bound,
+ )
+ .into_generic()
+ }
+
+ fn forward_logout_result_message(
+ forward: &ForwardLogoutHeader,
+ outcome: &Result<u64, MetadataSubmitError>,
+ ) -> Message<GenericHeader> {
+ build_forward_logout_result_message(
+ forward.cluster,
+ forward.view,
+ 0,
+ forward.client,
+ forward.nonce,
+ outcome,
+ )
+ .into_generic()
+ }
+
+ /// The `GET_CLUSTER_METADATA` auth gate holds on every roster shape: it
+ /// describes the private replica network, and a client that dialed a
+ /// backup reaches the cluster by logging in there (the backup forwards
+ /// the register), not by reading the topology first.
+ ///
+ /// The denial must be a plain Reply on the status channel, not an
+ /// Eviction: no session exists yet, and a session-terminal frame makes
+ /// SDKs drop the connection their login is about to use.
+ #[compio::test]
+ async fn pre_auth_cluster_metadata_denied_on_every_roster() {
+ use configs::cluster::{ClusterNodeConfig, TransportPorts};
+ use iggy_binary_protocol::codes::GET_CLUSTER_METADATA_CODE;
+ use iggy_binary_protocol::{GenericHeader, ReplyHeader};
+
+ const TRANSPORT: u128 = 91;
+ const COMMAND_OFFSET: usize = std::mem::offset_of!(GenericHeader, command);
+ const STATUS_OFFSET: usize = std::mem::offset_of!(ReplyHeader, status);
+ const OP_OFFSET: usize = std::mem::offset_of!(ReplyHeader, op);
+ const COMMIT_OFFSET: usize = std::mem::offset_of!(ReplyHeader, commit);
+
+ fn metadata_read() -> Message<GenericHeader> {
+ let header_size = size_of::<RequestHeader>();
+ let mut message = Message::<RequestHeader>::new(header_size);
+ {
+ let header = bytemuck::checked::from_bytes_mut::<RequestHeader>(
+ &mut message.as_mut_slice()[..header_size],
+ );
+ *header = RequestHeader {
+ command: Command2::Request,
+ operation: Operation::NonReplicated,
+ size: u32::try_from(header_size).expect("header fits u32"),
+ client: TRANSPORT,
+ ..Default::default()
+ };
+ header.reserved[..4].copy_from_slice(&GET_CLUSTER_METADATA_CODE.to_le_bytes());
+ }
+ message.into_generic()
+ }
+
+ fn roster_node(name: &str) -> ClusterNodeConfig {
+ ClusterNodeConfig {
+ name: name.to_owned(),
+ ip: "127.0.0.1".to_owned(),
+ advertised_address: None,
+ advertised_addresses: Vec::new(),
+ replica_id: 0,
+ ports: TransportPorts::default(),
+ }
+ }
+
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 0, 1, FIRST_BOOT));
+ let sessions = Rc::new(RefCell::new(SessionManager::new()));
+ let system_config = Arc::new(ServerSystemConfig::default());
+
+ let multi_node = Rc::new(ClusterRoster {
+ enabled: true,
+ name: "test-cluster".to_owned(),
+ nodes: vec![roster_node("node-0").into(), roster_node("node-1").into()],
+ self_ip: "127.0.0.1".to_owned(),
+ self_ports: TransportPorts::default(),
+ metadata_view: Arc::new(std::sync::atomic::AtomicU64::new(
+ crate::cluster_meta::METADATA_VIEW_UNKNOWN,
+ )),
+ });
+ // Default roster is disabled / single node; the installed one is a
+ // real cluster. Neither serves an unbound caller.
+ for roster in [None, Some(multi_node)] {
+ if let Some(roster) = roster {
+ sessions.borrow_mut().set_cluster_roster(roster);
+ }
+ handle_client_request(
+ &shard,
+ &sessions,
+ &system_config,
+ 1,
+ TRANSPORT,
+ metadata_read(),
+ )
+ .await;
+ let replies = bus.client_replies.borrow();
+ assert_eq!(replies.len(), 1, "gated read must still produce a frame");
+ let (client, frame) = &replies[0];
+ assert_eq!(*client, TRANSPORT);
+ assert_eq!(
+ frame[COMMAND_OFFSET],
+ Command2::Reply as u8,
+ "an unbound cluster-metadata read must be denied with a Reply, not evicted"
+ );
+ let status =
+ u32::from_le_bytes(frame[STATUS_OFFSET..STATUS_OFFSET + 4].try_into().unwrap());
+ assert_eq!(
+ status,
+ IggyError::Unauthenticated.as_code(),
+ "deny reply status must be Unauthenticated"
+ );
+ let op = u64::from_le_bytes(frame[OP_OFFSET..OP_OFFSET + 8].try_into().unwrap());
+ assert_eq!(op, 0, "pre-auth deny carries no session, so op must be 0");
+ let commit =
+ u64::from_le_bytes(frame[COMMIT_OFFSET..COMMIT_OFFSET + 8].try_into().unwrap());
+ assert_eq!(commit, 0, "pre-auth deny must not disclose commit activity");
+ drop(replies);
+ bus.client_replies.borrow_mut().clear();
+ }
+ }
+
#[test]
fn create_topic_bounds_deny_pre_consensus() {
let config = ServerSystemConfig::default();
diff --git a/core/server/src/dispatch/authz.rs b/core/server/src/dispatch/authz.rs
index afd8cc0..7812915 100644
--- a/core/server/src/dispatch/authz.rs
+++ b/core/server/src/dispatch/authz.rs
@@ -29,9 +29,9 @@
use consensus::MetadataHandle;
use iggy_binary_protocol::codes::{
- GET_CONSUMER_GROUP_CODE, GET_CONSUMER_GROUPS_CODE, GET_PERSONAL_ACCESS_TOKENS_CODE,
- GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE, GET_TOPICS_CODE,
- GET_USER_CODE, GET_USERS_CODE,
+ GET_CLUSTER_METADATA_CODE, GET_CONSUMER_GROUP_CODE, GET_CONSUMER_GROUPS_CODE,
+ GET_PERSONAL_ACCESS_TOKENS_CODE, GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE,
+ GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE, GET_USERS_CODE,
};
use iggy_binary_protocol::requests::consumer_groups::{
GetConsumerGroupRequest, GetConsumerGroupsRequest,
@@ -131,14 +131,14 @@
decision.err().map(|error| error.as_code())
}
-/// Reply to a partition op rejected before it reached the plane with the op's
-/// frame: empty body + nonzero `status`. The nonzero status is the whole
+/// Reply to a request rejected before it reached its plane with the request's
+/// own frame: empty body + nonzero `status`. The nonzero status is the whole
/// point: the SDK peeks it and surfaces the typed error, whereas a status-0
/// frame reads as a committed ack for work that never happened. Silence is no
/// better, the connection decodes replies in lockstep and would wedge on every
/// later request.
#[allow(clippy::future_not_send)]
-pub(super) async fn send_partition_deny_reply<B, MJ, S, SB>(
+pub(super) async fn send_deny_reply<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
transport_client_id: u128,
request_header: &RoutedRequestHeader,
@@ -162,7 +162,39 @@
status,
error = %error,
operation = ?request_header.operation,
- "failed to surface partition authz denial"
+ "failed to surface request denial"
+ );
+ }
+}
+
+/// Deny a request from an unbound transport without disclosing the metadata
+/// commit frontier. The status is the only field a pre-authenticated caller
+/// needs, while the live commit would expose cluster write activity.
+#[allow(clippy::future_not_send)]
+pub(super) async fn send_unbound_deny_reply<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ transport_client_id: u128,
+ request_header: &RoutedRequestHeader,
+ status: u32,
+) where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<MJ::Storage, Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let reply = build_deny_reply(request_header, transport_client_id, 0, 0, status);
+ if let Err(error) = shard
+ .bus
+ .send_to_client(transport_client_id, reply.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ status,
+ error = %error,
+ operation = ?request_header.operation,
+ "failed to surface unbound request denial"
);
}
}
@@ -228,10 +260,8 @@
/// notfound-before-permission ordering holds. `Err` denies with that code.
/// Unscoped rules gate directly; identifier-scoped rules resolve (stream[,
/// topic]) against committed state first. The PAT list is self-scoped, so
-/// authentication is its whole rule. `GET_CLUSTER_METADATA` is deliberately
-/// pre-auth (bootstrap / leader discovery; the dispatch allowlist admits it
-/// unauthenticated) and, like every other code the builder serves, is ungated
-/// here.
+/// authentication is its whole rule, and `GET_CLUSTER_METADATA` -- which
+/// describes the private replica network -- is gated the same way.
pub(super) fn authorize_default_read<B, MJ, S, SB>(
shard: &Rc<ShellShard<B, MJ, S, SB>>,
code: u32,
@@ -254,6 +284,10 @@
// Self-scoped: lists only the caller's own tokens, so there is no
// permissioner rule to run (legacy runs none either).
GET_PERSONAL_ACCESS_TOKENS_CODE => user_id.map(|_| ()).ok_or(IggyError::Unauthenticated),
+ // Defence in depth: `handle_client_request` already denies an unbound
+ // transport with an `Unauthenticated` Reply before it reaches the
+ // builder, so this arm only ever fires if that gate is bypassed.
+ GET_CLUSTER_METADATA_CODE => user_id.map(|_| ()).ok_or(IggyError::Unauthenticated),
GET_STREAMS_CODE => authorize_uid(shard, user_id, Permissioner::get_streams),
GET_STREAM_CODE => gate_stream_scoped::<GetStreamRequest, _, _, _, _>(
shard,
diff --git a/core/server/src/http/state.rs b/core/server/src/http/state.rs
index 0c8a2cc..8ea1bb4 100644
--- a/core/server/src/http/state.rs
+++ b/core/server/src/http/state.rs
@@ -326,31 +326,7 @@
.map_err(|_| AuthError::SessionUnavailable)?
.map_err(|error| {
warn!(?error, "server HTTP: VSR Register submit failed");
- match error {
- // The Register never entered the pipeline, so re-issuing
- // it anywhere is safe; the transient-not-accepted body
- // tells a forwarding peer to retry against the current
- // primary. `Canceled` / `InProgress` mean a prepare may
- // still commit cluster-wide, so they stay the plain
- // unavailable 503.
- MetadataSubmitError::NotPrimary
- | MetadataSubmitError::NotCaughtUp
- | MetadataSubmitError::PipelineFull => AuthError::SessionNotAccepted,
- // Terminal, and the only variant here that is: retrying
- // anywhere cannot make the id free. Kept off the 503 path
- // so the caller's HTTP stack does not auto-retry forever.
- MetadataSubmitError::ClientIdOwnedByAnotherUser => {
- AuthError::SessionIdOwnedByAnotherUser
- }
- // `InProgress` / `Canceled` mean a prepare may still
- // commit cluster-wide, so the outcome is unknown rather
- // than terminal. A future variant lands here too: 503 is
- // the safe default, since it never asserts a refusal the
- // server did not make.
- MetadataSubmitError::InProgress | MetadataSubmitError::Canceled | _ => {
- AuthError::SessionUnavailable
- }
- }
+ register_submit_auth_error(&error)
})?;
// A fresh mint must land on a fresh entry, so a watermark it did not
// write means the id was already registered to this same user (see
@@ -423,6 +399,24 @@
}
}
+const fn register_submit_auth_error(error: &MetadataSubmitError) -> AuthError {
+ match error {
+ // These outcomes prove the Register never entered a pipeline, so a
+ // forwarding peer may safely retry against a re-resolved primary.
+ MetadataSubmitError::NotPrimary
+ | MetadataSubmitError::NotCaughtUp
+ | MetadataSubmitError::PipelineFull
+ | MetadataSubmitError::PrimaryUnreachable => AuthError::SessionNotAccepted,
+ MetadataSubmitError::ClientIdOwnedByAnotherUser => AuthError::SessionIdOwnedByAnotherUser,
+ // The proposal may still commit. Unknown future outcomes fail closed
+ // into the same client-only retry class.
+ MetadataSubmitError::InProgress
+ | MetadataSubmitError::Canceled
+ | MetadataSubmitError::ForwardTimedOut
+ | _ => AuthError::SessionUnavailable,
+ }
+}
+
/// Set the [`VIEW_HEADER`] to the current VSR view on a successful or redirect
/// `response`. Omits the header on error responses and when this node has no
/// live consensus: a missing header is unambiguous, whereas a fabricated view
@@ -443,3 +437,39 @@
}
response
}
+
+#[cfg(test)]
+mod tests {
+ use super::register_submit_auth_error;
+ use crate::http::error::AuthError;
+ use metadata::MetadataSubmitError;
+
+ #[test]
+ fn register_submit_errors_preserve_known_and_unknown_outcomes() {
+ for error in [
+ MetadataSubmitError::NotPrimary,
+ MetadataSubmitError::NotCaughtUp,
+ MetadataSubmitError::PipelineFull,
+ MetadataSubmitError::PrimaryUnreachable,
+ ] {
+ assert!(matches!(
+ register_submit_auth_error(&error),
+ AuthError::SessionNotAccepted
+ ));
+ }
+ assert!(matches!(
+ register_submit_auth_error(&MetadataSubmitError::ClientIdOwnedByAnotherUser),
+ AuthError::SessionIdOwnedByAnotherUser
+ ));
+ for error in [
+ MetadataSubmitError::InProgress,
+ MetadataSubmitError::Canceled,
+ MetadataSubmitError::ForwardTimedOut,
+ ] {
+ assert!(matches!(
+ register_submit_auth_error(&error),
+ AuthError::SessionUnavailable
+ ));
+ }
+ }
+}
diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs
index 8528193..ae8b238 100644
--- a/core/server/src/server_error.rs
+++ b/core/server/src/server_error.rs
@@ -72,6 +72,19 @@
message_bus::OWNER_NONE - 1
)]
ShardsCountOverflow { count: usize },
+ #[error(
+ "shard {shard_id} message pump died instead of draining ({reason}); \
+ committed journal tail may not have flushed"
+ )]
+ ShardPumpDied { shard_id: u16, reason: String },
+ #[error(
+ "shard {shard_id} message pump did not drain within {timeout:?}. \
+ Committed journal tail may not have flushed"
+ )]
+ ShardPumpDrainTimedOut {
+ shard_id: u16,
+ timeout: std::time::Duration,
+ },
#[error("system.sharding.inbox_capacity must be in 1..={max}; got {value}")]
InvalidInboxCapacity { value: usize, max: usize },
#[error("system.sharding.shutdown_drain_timeout must be in (0, {max:?}]; got {value:?}")]
diff --git a/core/server/src/session_manager.rs b/core/server/src/session_manager.rs
index 11fca2a..3a896b0 100644
--- a/core/server/src/session_manager.rs
+++ b/core/server/src/session_manager.rs
@@ -100,7 +100,7 @@
/// a consensus reply arrives and needs routing to the right connection.
client_to_connection: HashMap<u128, u128>,
/// This shard's copy of the configured cluster roster, served by the
- /// pre-auth `GetClusterMetadata` read. Lives here because it is the
+ /// `GetClusterMetadata` read. Lives here because it is the
/// per-shard context already threaded to the non-replicated read path;
/// installed once at bootstrap, disabled until then.
cluster_roster: Rc<ClusterRoster>,
diff --git a/core/server_common/src/consensus_message.rs b/core/server_common/src/consensus_message.rs
index 66f973a..f390181 100644
--- a/core/server_common/src/consensus_message.rs
+++ b/core/server_common/src/consensus_message.rs
@@ -17,11 +17,13 @@
use crate::iobuf::{Frozen, Owned};
use iggy_binary_protocol::{
- Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader, GenericHeader,
- Operation, PrepareHeader, PrepareOkHeader, RepairPrepareHeader, RepairRangeReplyHeader,
- RequestHeader, RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
- RequestStateTransferHeader, RoutedRequestHeader, StartViewChangeHeader, StartViewHeader,
- StateChunkHeader, StateTransferTargetHeader,
+ Command2, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader,
+ ForwardLogoutHeader, ForwardLogoutResultHeader, ForwardRegisterHeader,
+ ForwardRegisterResultHeader, GenericHeader, Operation, PrepareHeader, PrepareOkHeader,
+ RepairPrepareHeader, RepairRangeReplyHeader, RequestHeader, RequestPreparesHeader,
+ RequestStartViewHeader, RequestStateChunkHeader, RequestStateTransferHeader,
+ RoutedRequestHeader, StartViewChangeHeader, StartViewHeader, StateChunkHeader,
+ StateTransferTargetHeader,
};
use smallvec::SmallVec;
use std::{
@@ -552,6 +554,15 @@
RequestStateChunk(Message<RequestStateChunkHeader>),
/// Artifact bytes ride the body (`size` spans header + payload).
StateChunk(Message<StateChunkHeader>),
+ /// A backup relays a login it authenticated locally to the primary, which
+ /// owns the `Register` proposal.
+ ForwardRegister(Message<ForwardRegisterHeader>),
+ /// The primary's verdict, routed back to the parked login by nonce.
+ ForwardRegisterResult(Message<ForwardRegisterResultHeader>),
+ /// A backup asks the primary to commit a session teardown.
+ ForwardLogout(Message<ForwardLogoutHeader>),
+ /// The primary's verdict, routed back to the parked logout by nonce.
+ ForwardLogoutResult(Message<ForwardLogoutResultHeader>),
}
impl MessageBag {
@@ -573,6 +584,10 @@
Self::StateTransferTarget(message) => message.header().command,
Self::RequestStateChunk(message) => message.header().command,
Self::StateChunk(message) => message.header().command,
+ Self::ForwardRegister(message) => message.header().command,
+ Self::ForwardRegisterResult(message) => message.header().command,
+ Self::ForwardLogout(message) => message.header().command,
+ Self::ForwardLogoutResult(message) => message.header().command,
}
}
@@ -594,6 +609,10 @@
Self::StateTransferTarget(message) => message.header().size(),
Self::RequestStateChunk(message) => message.header().size(),
Self::StateChunk(message) => message.header().size(),
+ Self::ForwardRegister(message) => message.header().size(),
+ Self::ForwardRegisterResult(message) => message.header().size(),
+ Self::ForwardLogout(message) => message.header().size(),
+ Self::ForwardLogoutResult(message) => message.header().size(),
}
}
@@ -615,6 +634,10 @@
Self::StateTransferTarget(message) => message.header().operation(),
Self::RequestStateChunk(message) => message.header().operation(),
Self::StateChunk(message) => message.header().operation(),
+ Self::ForwardRegister(message) => message.header().operation(),
+ Self::ForwardRegisterResult(message) => message.header().operation(),
+ Self::ForwardLogout(message) => message.header().operation(),
+ Self::ForwardLogoutResult(message) => message.header().operation(),
}
}
}
@@ -674,6 +697,18 @@
Command2::StateChunk => Ok(Self::StateChunk(
value.try_into_typed::<StateChunkHeader>()?,
)),
+ Command2::ForwardRegister => Ok(Self::ForwardRegister(
+ value.try_into_typed::<ForwardRegisterHeader>()?,
+ )),
+ Command2::ForwardRegisterResult => Ok(Self::ForwardRegisterResult(
+ value.try_into_typed::<ForwardRegisterResultHeader>()?,
+ )),
+ Command2::ForwardLogout => Ok(Self::ForwardLogout(
+ value.try_into_typed::<ForwardLogoutHeader>()?,
+ )),
+ Command2::ForwardLogoutResult => Ok(Self::ForwardLogoutResult(
+ value.try_into_typed::<ForwardLogoutResultHeader>()?,
+ )),
// Reply / Eviction are server-to-client frames; they do not
// appear on the inbound dispatch path.
Command2::Reply | Command2::Eviction => {
@@ -691,7 +726,8 @@
mod tests {
use super::*;
use iggy_binary_protocol::{
- HEADER_SIZE, Operation, ReplyHeader, RequestHeader, frame_checksum_bytes,
+ ForwardLogoutHeader, ForwardLogoutOutcome, ForwardLogoutResultHeader, HEADER_SIZE,
+ Operation, ReplyHeader, RequestHeader, frame_checksum_bytes,
};
use smallvec::smallvec;
@@ -830,6 +866,42 @@
}
}
+ #[test]
+ fn forward_logout_commands_round_trip_into_bag() {
+ let forward = Message::<ForwardLogoutHeader>::new(HEADER_SIZE).transmute_header(
+ |_, header: &mut ForwardLogoutHeader| {
+ header.command = Command2::ForwardLogout;
+ header.size = HEADER_SIZE as u32;
+ header.client = 7;
+ header.nonce = 8;
+ header.session = 9;
+ header.request = 10;
+ header.seal();
+ },
+ );
+ let result = Message::<ForwardLogoutResultHeader>::new(HEADER_SIZE).transmute_header(
+ |_, header: &mut ForwardLogoutResultHeader| {
+ header.command = Command2::ForwardLogoutResult;
+ header.size = HEADER_SIZE as u32;
+ header.client = 7;
+ header.nonce = 8;
+ header.commit = 11;
+ header.outcome = ForwardLogoutOutcome::Ok;
+ header.seal();
+ },
+ );
+
+ let forward = MessageBag::try_from(forward.into_generic()).expect("parse ForwardLogout");
+ let result =
+ MessageBag::try_from(result.into_generic()).expect("parse ForwardLogoutResult");
+ assert!(matches!(forward, MessageBag::ForwardLogout(_)));
+ assert!(matches!(result, MessageBag::ForwardLogoutResult(_)));
+ assert_eq!(forward.command(), Command2::ForwardLogout);
+ assert_eq!(result.command(), Command2::ForwardLogoutResult);
+ assert_eq!(forward.operation(), Operation::Reserved);
+ assert_eq!(result.size(), HEADER_SIZE as u32);
+ }
+
// Construction via Message::new (zeroed)
#[test]
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index d898e6a..efdc368 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -39,11 +39,12 @@
use crossfire::TrySendError;
use futures::FutureExt;
use iggy_binary_protocol::{
- CHECKSUM_UNSEALED, Command2, CommitHeader, ConsensusHeader, DoViewChangeHeader, GenericHeader,
- Operation, PrepareHeader, PrepareOkHeader, RepairPrepareHeader, RepairRangeReplyHeader,
- RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
- RequestStateTransferHeader, RoutedRequestHeader, StartViewChangeHeader, StartViewHeader,
- StateChunkHeader, StateTransferTargetHeader,
+ CHECKSUM_UNSEALED, Command2, CommitHeader, ConsensusHeader, DoViewChangeHeader,
+ ForwardLogoutHeader, ForwardLogoutResultHeader, ForwardRegisterHeader,
+ ForwardRegisterResultHeader, GenericHeader, Operation, PrepareHeader, PrepareOkHeader,
+ RepairPrepareHeader, RepairRangeReplyHeader, RequestPreparesHeader, RequestStartViewHeader,
+ RequestStateChunkHeader, RequestStateTransferHeader, RoutedRequestHeader,
+ StartViewChangeHeader, StartViewHeader, StateChunkHeader, StateTransferTargetHeader,
};
#[cfg(any(test, feature = "simulator"))]
use iggy_common::PartitionStats;
@@ -173,7 +174,9 @@
/// and awaits the outcome over `reply`. `Register` carries the submit error
/// verbatim because one variant
/// (`MetadataSubmitError::ClientIdOwnedByAnotherUser`) is terminal and must
-/// not be retried; the remaining variants are transient by contract.
+/// not be retried. The remaining variants are transient by contract, and
+/// Logout preserves them so its caller can distinguish an unknown outcome
+/// from a request that never entered the primary pipeline.
pub enum MetadataSubmit {
Register {
vsr_client_id: u128,
@@ -184,11 +187,37 @@
/// client a retry storm of full password verifications.
reply: Sender<Result<BoundSession, MetadataSubmitError>>,
},
+ /// A backup node authenticated a login and asks this node -- which it
+ /// believes is the primary -- to run only the `Register` proposal. The
+ /// verdict travels back over the replica interconnect as a
+ /// `ForwardRegisterResult`, not over a channel: the awaiting login lives
+ /// in another process.
+ ///
+ /// Handled by proposing IN PROCESS, never by forwarding again. That is
+ /// what bounds a forward at one hop: a node that has since lost
+ /// primaryship answers `NotPrimary`, and the client's SDK replays.
+ ForwardedRegister {
+ vsr_client_id: u128,
+ user_id: u32,
+ /// Correlation the origin minted; echoed verbatim in the result.
+ nonce: u128,
+ /// Replica the result frame goes back to.
+ origin_replica: u8,
+ },
+ /// A backup owns a bound client connection and asks the metadata primary
+ /// to commit its Logout. The result returns over the replica interconnect.
+ ForwardedLogout {
+ vsr_client_id: u128,
+ session: u64,
+ request: u64,
+ nonce: u128,
+ origin_replica: u8,
+ },
Logout {
vsr_client_id: u128,
session: u64,
request: u64,
- reply: Sender<Option<u64>>,
+ reply: Sender<Result<u64, MetadataSubmitError>>,
},
/// A peer (home) shard relays a client's replicated request to shard 0
/// and awaits the committed reply over `reply` (`None` on a transient
@@ -350,13 +379,14 @@
/// deadline.
const PARTITION_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
-/// Race `future` against a bus timer: `Some` if it finishes within `budget`,
-/// `None` if the timer fires first. Uses [`MessageBus::sleep`] (virtual under
-/// the simulator, wall-clock in production) rather than `compio::time::timeout`,
-/// which panics outside a compio runtime and so cannot run under the
-/// deterministic executor.
+/// Race `future` against a bus timer.
+///
+/// `Some` if it finishes within `budget`, `None` if the timer fires first.
+/// Uses [`MessageBus::sleep`] (virtual under the simulator, wall-clock in
+/// production) rather than `compio::time::timeout`, which panics outside a
+/// compio runtime and so cannot run under the deterministic executor.
#[allow(clippy::future_not_send)]
-async fn bus_timeout<B, F>(bus: &B, budget: std::time::Duration, future: F) -> Option<F::Output>
+pub async fn bus_timeout<B, F>(bus: &B, budget: std::time::Duration, future: F) -> Option<F::Output>
where
B: MessageBus,
F: Future,
@@ -502,6 +532,23 @@
Ok(())
}
+/// Starting point for [`IggyShard::next_forward_nonce`]: the low half
+/// of this boot's consensus incarnation.
+///
+/// Forward nonces are node-local and never persisted, so a counter that starts
+/// at zero every boot re-mints the exact sequence the previous boot used. A
+/// forward answer still in flight across a restart would then match a nonce a
+/// DIFFERENT login now holds and confirm a login that never committed. The
+/// incarnation is fresh per boot, which moves the whole sequence.
+///
+/// Zero on shards owning no metadata consensus (they never forward) and
+/// wherever nothing set an incarnation, which degenerates to the unseeded
+/// sequence: no worse than before, and the shards that take it are test ones.
+#[allow(clippy::cast_possible_truncation)]
+fn forward_nonce_seed<B: MessageBus>(consensus: Option<&VsrConsensus<B>>) -> u64 {
+ consensus.map_or(0, VsrConsensus::incarnation) as u64
+}
+
/// Lifecycle frame variants.
///
/// Connection setup and cross-shard forwards: every frame the inter-shard
@@ -1180,6 +1227,28 @@
/// See [`ServedSegmentCache`].
served_segment_cache: RefCell<ServedSegmentCache>,
+ /// Logins this node forwarded to the primary and is still waiting on, keyed
+ /// by the `(nonce, client)` pair stamped into the `ForwardRegister` frame.
+ /// Shard 0 only, since that is where the forward is issued and where the
+ /// result routes back to. Entries are removed at exactly three points --
+ /// result delivery, forward timeout, and a failed send -- so an abandoned
+ /// login cannot leak one.
+ ///
+ /// The client id is part of the key rather than payload the ingest compares:
+ /// an answer echoing a client the nonce was never parked for is then exactly
+ /// as unroutable as one carrying an unknown nonce, and the miss leaves the
+ /// legitimate entry parked instead of evicting it.
+ register_forwards: RefCell<HashMap<(u128, u128), Sender<ForwardRegisterResultHeader>>>,
+
+ /// Logouts this node forwarded to the primary and is still waiting on.
+ logout_forwards: RefCell<HashMap<(u128, u128), Sender<ForwardLogoutResultHeader>>>,
+
+ /// Monotonic source of forwarding nonces. Node-local: the
+ /// nonce only has to distinguish this node's own in-flight forwards, across
+ /// its restarts as well as within one boot. Seeded by
+ /// [`forward_nonce_seed`].
+ forward_nonce: Cell<u64>,
+
/// Handler for inbound [`MetadataSubmit`] frames. Only shard 0 receives
/// these (it owns the metadata consensus group); peers send them here
/// via [`Self::forward_metadata_submit`]. Defaults to a no-op for the
@@ -1406,6 +1475,7 @@
u32::try_from(senders.len()).map_err(|_| ShardCtorError::ShardCountOverflow {
count: senders.len(),
})?;
+ let nonce_seed = forward_nonce_seed(metadata.consensus.as_ref());
let plane = MuxPlane::new(variadic!(metadata, partitions));
let ShardIdentity { id, name } = identity;
Ok(Self {
@@ -1436,6 +1506,9 @@
state_transfer_offers: RefCell::new(HashMap::new()),
partition_offer_builds: RefCell::new(HashMap::new()),
served_segment_cache: RefCell::new(ServedSegmentCache::default()),
+ register_forwards: RefCell::new(HashMap::new()),
+ logout_forwards: RefCell::new(HashMap::new()),
+ forward_nonce: Cell::new(nonce_seed),
served_segment_cache_bytes_max: Cell::new(SERVED_SEGMENT_CACHE_BYTES_DEFAULT),
partition_artifact_len_max: Cell::new(PARTITION_ARTIFACT_LEN_DEFAULT),
repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX),
@@ -1479,6 +1552,56 @@
self.bus_max_message_size.set(max_message_size);
}
+ /// Mint a fresh, never-zero nonce for a register or logout forward.
+ ///
+ /// `replica` rides the high half, which separates the nonce spaces of
+ /// different NODES: a result frame that somehow arrives from the wrong node
+ /// cannot collide with a live entry. Successive boots of THIS node are
+ /// separated by the counter's incarnation seed instead.
+ ///
+ /// The counter skips zero on wrap: both forwarding headers reject a zero
+ /// nonce in `validate`, so a wrapped counter would have the origin build a
+ /// frame the primary drops.
+ pub fn next_forward_nonce(&self, replica: u8) -> u128 {
+ let counter = self.forward_nonce.get().wrapping_add(1).max(1);
+ self.forward_nonce.set(counter);
+ (u128::from(replica) << 64) | u128::from(counter)
+ }
+
+ /// Park a forwarded login under `(nonce, client)` until the primary answers.
+ pub fn park_register_forward(
+ &self,
+ nonce: u128,
+ client: u128,
+ reply: Sender<ForwardRegisterResultHeader>,
+ ) {
+ self.register_forwards
+ .borrow_mut()
+ .insert((nonce, client), reply);
+ }
+
+ /// Drop a parked login (timeout, or a forward that never left the node).
+ pub fn cancel_register_forward(&self, nonce: u128, client: u128) {
+ self.register_forwards.borrow_mut().remove(&(nonce, client));
+ }
+
+ /// Park a forwarded logout under `(nonce, client)` until the primary answers.
+ pub fn park_logout_forward(
+ &self,
+ nonce: u128,
+ client: u128,
+ reply: Sender<ForwardLogoutResultHeader>,
+ ) {
+ self.logout_forwards
+ .borrow_mut()
+ .insert((nonce, client), reply);
+ }
+
+ /// Drop a parked logout after timeout or a failed send.
+ pub fn cancel_logout_forward(&self, nonce: u128, client: u128) {
+ self.logout_forwards.borrow_mut().remove(&(nonce, client));
+ }
+
/// Hand a metadata consensus submit (login/logout) to shard 0.
///
/// Sends a [`LifecycleFrame::MetadataSubmit`] into shard 0's inbox. The
@@ -1654,6 +1777,7 @@
// with the current type setup; revisit when crossfire grows an
// unbounded variant or we replace it.
let (_tx, inbox) = channel(1);
+ let nonce_seed = forward_nonce_seed(metadata.consensus.as_ref());
let plane = MuxPlane::new(variadic!(metadata, partitions));
let ShardIdentity { id, name } = identity;
Self {
@@ -1690,6 +1814,9 @@
state_transfer_offers: RefCell::new(HashMap::new()),
partition_offer_builds: RefCell::new(HashMap::new()),
served_segment_cache: RefCell::new(ServedSegmentCache::default()),
+ register_forwards: RefCell::new(HashMap::new()),
+ logout_forwards: RefCell::new(HashMap::new()),
+ forward_nonce: Cell::new(nonce_seed),
served_segment_cache_bytes_max: Cell::new(SERVED_SEGMENT_CACHE_BYTES_DEFAULT),
partition_artifact_len_max: Cell::new(PARTITION_ARTIFACT_LEN_DEFAULT),
repair_chunk_max: Cell::new(REPAIR_CHUNK_MAX),
@@ -2369,12 +2496,89 @@
}
Ok(MessageBag::RequestStateChunk(ref msg)) => self.on_request_state_chunk(msg).await,
Ok(MessageBag::StateChunk(ref msg)) => self.on_state_chunk(msg).await,
+ // A forwarded proposal must leave the pump because its commit is
+ // driven by this same pump. The metadata-submit handler spawns it.
+ Ok(MessageBag::ForwardRegister(ref msg)) => self.on_forward_register(*msg.header()),
+ Ok(MessageBag::ForwardRegisterResult(ref msg)) => {
+ self.on_forward_register_result(*msg.header());
+ }
+ Ok(MessageBag::ForwardLogout(ref msg)) => self.on_forward_logout(*msg.header()),
+ Ok(MessageBag::ForwardLogoutResult(ref msg)) => {
+ self.on_forward_logout_result(*msg.header());
+ }
Err(e) => {
tracing::warn!(shard = self.id, error = %e, "dropping unparsable consensus frame");
}
}
}
+ fn on_forward_register(&self, header: ForwardRegisterHeader) {
+ if !self.peer_is_known(header.replica, "ForwardRegister") {
+ return;
+ }
+ debug_assert_eq!(
+ self.id, 0,
+ "ForwardRegister routes to the metadata consensus owner"
+ );
+ (self.on_metadata_submit)(MetadataSubmit::ForwardedRegister {
+ vsr_client_id: header.client,
+ user_id: header.user_id,
+ nonce: header.nonce,
+ origin_replica: header.replica,
+ });
+ }
+
+ fn on_forward_register_result(&self, header: ForwardRegisterResultHeader) {
+ let waiter = self
+ .register_forwards
+ .borrow_mut()
+ .remove(&(header.nonce, header.client));
+ if let Some(waiter) = waiter {
+ let _ = waiter.try_send(header);
+ } else {
+ tracing::debug!(
+ shard = self.id,
+ nonce = header.nonce,
+ client = header.client,
+ "dropping forward-register result with no parked login"
+ );
+ }
+ }
+
+ fn on_forward_logout(&self, header: ForwardLogoutHeader) {
+ if !self.peer_is_known(header.replica, "ForwardLogout") {
+ return;
+ }
+ debug_assert_eq!(
+ self.id, 0,
+ "ForwardLogout routes to the metadata consensus owner"
+ );
+ (self.on_metadata_submit)(MetadataSubmit::ForwardedLogout {
+ vsr_client_id: header.client,
+ session: header.session,
+ request: header.request,
+ nonce: header.nonce,
+ origin_replica: header.replica,
+ });
+ }
+
+ fn on_forward_logout_result(&self, header: ForwardLogoutResultHeader) {
+ let waiter = self
+ .logout_forwards
+ .borrow_mut()
+ .remove(&(header.nonce, header.client));
+ if let Some(waiter) = waiter {
+ let _ = waiter.try_send(header);
+ } else {
+ tracing::debug!(
+ shard = self.id,
+ nonce = header.nonce,
+ client = header.client,
+ "dropping forward-logout result with no parked request"
+ );
+ }
+ }
+
/// Does the partition materialised under `namespace_raw` belong to the
/// incarnation the committed metadata denotes?
///
@@ -4913,6 +5117,7 @@
let Some(missing_op) = missing else {
let actions = consensus.start_pending_view(PlaneKind::Metadata);
+ let (local_actions, wire_actions) = split_local_actions(actions);
tracing::info!(
shard = self.id,
view = consensus.view(),
@@ -4920,10 +5125,17 @@
commit_max = pending.commit_max,
"merged log is locally serveable; starting the view"
);
+ // Locals run BEFORE the persist await. `start_pending_view` has
+ // already flipped this replica into a Normal primary, so yielding
+ // with a still-empty pipeline lets a concurrent client submit mint
+ // the next op below the inherited suffix, and the later rebuild
+ // then pushes out of sequence. The locals must also survive a
+ // failed persist, which fences only the wire sends.
+ dispatch_vsr_actions(consensus, metadata.journal.as_ref(), &local_actions).await;
if metadata.persist_superblock_if_needed(consensus).await {
- dispatch_vsr_actions(consensus, metadata.journal.as_ref(), &actions).await;
+ dispatch_vsr_actions(consensus, metadata.journal.as_ref(), &wire_actions).await;
}
- if actions
+ if local_actions
.iter()
.any(|action| matches!(action, VsrAction::CommitJournal))
&& !consensus.is_transferring()
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index b683af2..3a375b1 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -104,6 +104,25 @@
let h = *m.header();
(h.operation(), h.group, m.into_generic())
}
+ // Register forwarding is a metadata-plane errand, and the metadata
+ // consensus group lives on shard 0 on every node; the headers carry no
+ // group field because there is nothing else they could address.
+ MessageBag::ForwardRegister(m) => {
+ let h = *m.header();
+ (h.operation(), METADATA_GROUP, m.into_generic())
+ }
+ MessageBag::ForwardRegisterResult(m) => {
+ let h = *m.header();
+ (h.operation(), METADATA_GROUP, m.into_generic())
+ }
+ MessageBag::ForwardLogout(m) => {
+ let h = *m.header();
+ (h.operation(), METADATA_GROUP, m.into_generic())
+ }
+ MessageBag::ForwardLogoutResult(m) => {
+ let h = *m.header();
+ (h.operation(), METADATA_GROUP, m.into_generic())
+ }
}
}
@@ -254,7 +273,10 @@
/// records the drop in `frame_drops_total`, under `variant=partition` for a
/// partition-plane operation and `variant=consensus` otherwise -- the two
/// have different recovery stories, so folding them into one label hides
- /// which one is bleeding. VSR retransmit recovers consensus drops. A
+ /// which one is bleeding. VSR retransmit recovers consensus drops, except
+ /// the four register/logout forwarding frames, which no retransmit covers: a
+ /// dropped forward or its result surfaces as the origin's forward timeout
+ /// plus the SDK's session-operation replay. A
/// `target` past the end of `senders` (a stored `u16` from `shard_for`, not
/// a trusted index) is dropped with `reason=unroutable` rather than
/// panicking. Metadata frames always pass `target = 0` here, since
diff --git a/core/simulator/src/deps.rs b/core/simulator/src/deps.rs
index 2502b8e..f44ab3c 100644
--- a/core/simulator/src/deps.rs
+++ b/core/simulator/src/deps.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::executor::TimerHandle;
+use crate::executor::{TimerHandle, yield_once};
use clock::Clock;
use iggy_binary_protocol::PrepareHeader;
use iggy_common::{IggyTimestamp, variadic};
@@ -385,6 +385,12 @@
/// shard withholds the view-scoped send. Proves the split-brain gate, that a
/// replica never sends in a view it has not durably recorded.
fail_writes: Cell<bool>,
+ /// Fault injection: when set, every [`SuperblockStore::write`] suspends once
+ /// before completing. A real superblock persist is an fsync-wide suspension
+ /// point; the default in-memory write completes on first poll, so nothing can
+ /// interleave with a persist and every schedule-sensitive bug behind one is
+ /// invisible to the simulator. The yield restores the window.
+ yield_writes: Cell<bool>,
}
impl SimSuperblock {
@@ -400,6 +406,12 @@
pub fn set_fail_writes(&self) {
self.fail_writes.set(true);
}
+
+ /// Make every subsequent write suspend once before completing, so tasks that
+ /// are ready at persist time interleave with it. See [`Self::yield_writes`].
+ pub fn set_yield_writes(&self) {
+ self.yield_writes.set(true);
+ }
}
#[allow(clippy::future_not_send)]
@@ -408,6 +420,9 @@
if self.fail_writes.get() {
return Err(std::io::Error::other("sim superblock write fault"));
}
+ if self.yield_writes.get() {
+ yield_once().await;
+ }
*self.latest.borrow_mut() = Some(payload.to_vec());
Ok(())
}
diff --git a/core/simulator/src/executor/mod.rs b/core/simulator/src/executor/mod.rs
index e44e8a6..8fdfc89 100644
--- a/core/simulator/src/executor/mod.rs
+++ b/core/simulator/src/executor/mod.rs
@@ -53,6 +53,30 @@
/// streams keep scheduling draws from perturbing network or workload traces.
pub const EXECUTOR_SEED_SALT: u64 = 0x5A1A_F0E5_FACE_0002;
+/// Ready on the second poll; the first re-queues the task behind every other
+/// ready task, exactly like a wake arriving mid-await. Lets a task give the
+/// executor a turn without parking on a timer, which would only wake it at
+/// the next step.
+pub(crate) struct YieldOnce(bool);
+
+impl Future for YieldOnce {
+ type Output = ();
+
+ fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<()> {
+ if self.0 {
+ Poll::Ready(())
+ } else {
+ self.0 = true;
+ context.waker().wake_by_ref();
+ Poll::Pending
+ }
+ }
+}
+
+pub(crate) const fn yield_once() -> YieldOnce {
+ YieldOnce(false)
+}
+
/// A type-erased, `!Send` task future: what the executor stores and what
/// [`PendingSpawns`] stages.
type BoxedTask = Pin<Box<dyn Future<Output = ()>>>;
@@ -471,29 +495,6 @@
use std::cell::Cell;
use std::rc::Rc;
- /// Completes on the second poll; self-wakes on the first.
- struct YieldOnce {
- yielded: bool,
- }
-
- impl Future for YieldOnce {
- type Output = ();
-
- fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
- if self.yielded {
- Poll::Ready(())
- } else {
- self.yielded = true;
- cx.waker().wake_by_ref();
- Poll::Pending
- }
- }
- }
-
- fn yield_once() -> YieldOnce {
- YieldOnce { yielded: false }
- }
-
#[test]
fn spawn_and_complete() {
let mut executor = DetExecutor::new(1);
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index d4c2ba0..a119587 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -3115,8 +3115,10 @@
//! tests cover the sequencer-truncation path directly.
use super::*;
+ use crate::executor::yield_once;
use consensus::{Sequencer, Status};
use journal::Journal;
+ use message_bus::MessageBus;
/// Whether a replica's shard-0 metadata consensus is a settled primary in a
/// view past the one that crashed.
@@ -3255,4 +3257,131 @@
"op {committed} must be repaired back into the new primary's journal"
);
}
+
+ /// A client submit landing inside the new primary's view-start superblock
+ /// persist must not corrupt the pipeline.
+ ///
+ /// `start_pending_view` flips the replica into a Normal primary
+ /// synchronously and defers the rebuild of the inherited uncommitted
+ /// suffix; the persist then suspends the pump. A register admitted in that
+ /// window used to mint the next op into the still-empty pipeline, and the
+ /// deferred rebuild panicked pushing the inherited op beneath it
+ /// ("sequence must be sequential"); the same empty pipeline also blinded
+ /// the register dedup, admitting an inherited in-flight register twice.
+ /// The suspension is real on disk-backed stores (an fsync) and is restored
+ /// here with `set_yield_writes`.
+ #[test]
+ fn given_a_register_inside_the_view_start_persist_when_the_pipeline_rebuilds_should_commit_once()
+ {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolConfigOther {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+
+ let replica_count: u8 = 3;
+ let settled_client: u128 = 1;
+ let straggler_client: u128 = 2;
+ let network_opts = packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 2,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::new(
+ replica_count as usize,
+ [settled_client, straggler_client].into_iter(),
+ network_opts,
+ );
+
+ let client = SimClient::new(settled_client);
+ sim.register_client_with_primary(&client);
+ for _ in 0..100 {
+ sim.step();
+ }
+ let (baseline_head, baseline_commit) = metadata_progress(&sim, 1);
+ assert_eq!(
+ baseline_head, baseline_commit,
+ "the cluster must be quiescent before the straggler is staged"
+ );
+
+ // Stage the inherited suffix: the straggler's register reaches the
+ // next primary's journal, then the old primary dies before the commit
+ // makes it back.
+ let straggler = SimClient::new(straggler_client);
+ sim.submit_request(straggler_client, 0, straggler.register().into_generic());
+ let mut staged = None;
+ for _ in 0..200 {
+ sim.step();
+ let (head, commit_max) = metadata_progress(&sim, 1);
+ if head > baseline_head && commit_max < head {
+ staged = Some(head);
+ break;
+ }
+ }
+ let staged =
+ staged.expect("the register must reach the next primary's journal before it commits");
+
+ // Both survivors' next persists suspend once, opening the window a
+ // real fsync has.
+ sim.replicas[1].superblock.set_yield_writes();
+ sim.replicas[2].superblock.set_yield_writes();
+ sim.replica_crash(0);
+
+ // The straggler's retry loop, as the server runs it: `dispatch` spawns
+ // the in-process submit on its own task, which is what can interleave
+ // with the parked pump. The sim's wire path processes requests inside
+ // the pump itself, so the window is only reachable from a spawned
+ // task. A plain once-per-step retry is never ready inside the drain
+ // where the pump flips to primary and suspends on the persist, so
+ // each tick wake spends a small budget of yield-separated attempts:
+ // the yields land the retry between the pump's polls, one of which is
+ // the suspended view-start persist.
+ let registered = std::rc::Rc::new(std::cell::Cell::new(false));
+ let submit_shard = std::rc::Rc::clone(&sim.replicas[1].shards[0]);
+ let submit_flag = std::rc::Rc::clone(®istered);
+ sim.executor.spawn(async move {
+ loop {
+ for _ in 0..32 {
+ match submit_shard
+ .plane
+ .metadata()
+ .submit_register_in_process(straggler_client, 0)
+ .await
+ {
+ Ok(_) => {
+ submit_flag.set(true);
+ return;
+ }
+ Err(error) if error.is_transient() => yield_once().await,
+ Err(_) => return,
+ }
+ }
+ submit_shard
+ .bus
+ .sleep(std::time::Duration::from_millis(10))
+ .await;
+ }
+ });
+
+ for _ in 0..1500 {
+ sim.step();
+ if registered.get() {
+ break;
+ }
+ }
+ assert!(
+ registered.get(),
+ "the straggler's login must complete after the failover"
+ );
+
+ let primary = (1..replica_count)
+ .find(|&replica| is_new_metadata_primary(&sim, replica))
+ .expect("a metadata primary must be elected after the old one crashes");
+ let (_, commit_max) = metadata_progress(&sim, primary);
+ assert!(
+ commit_max >= staged,
+ "the inherited op ({staged}) must commit under the new primary \
+ (commit_max = {commit_max})"
+ );
+ }
}
diff --git a/examples/python/uv.lock b/examples/python/uv.lock
index 45dc80e..3ce31e1 100644
--- a/examples/python/uv.lock
+++ b/examples/python/uv.lock
@@ -8,7 +8,7 @@
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev1"
+version = "0.9.0.dev2"
source = { directory = "../../foreign/python" }
[package.metadata]
diff --git a/foreign/cpp/tests/e2e/client.cpp b/foreign/cpp/tests/e2e/client.cpp
index 53c1a96..692481e 100644
--- a/foreign/cpp/tests/e2e/client.cpp
+++ b/foreign/cpp/tests/e2e/client.cpp
@@ -1563,20 +1563,22 @@
EXPECT_TRUE(found_after);
}
-TEST_F(LowLevelE2E_Client, GetClusterMetadataBeforeLoginSucceeds) {
- RecordProperty(
- "description",
- "Serves get_cluster_metadata to a connected but unauthenticated client, and rejects it without a connection.");
+TEST_F(LowLevelE2E_Client, GetClusterMetadataBeforeLoginThrows) {
+ RecordProperty("description",
+ "Rejects get_cluster_metadata before connect, after connect but before login, and after disconnect, "
+ "and serves it once authenticated.");
iggy::ffi::Client *client = GetLoggedOutClient();
ASSERT_THROW(client->get_cluster_metadata(), std::exception);
ASSERT_NO_THROW(client->connect());
- // By design pre-login on the VSR server: an SDK must read the roster to
- // find the primary before it can authenticate (redirect bootstrap).
+ // The roster is private, so the read is auth-gated. No pre-login read is
+ // needed: a client that dialed a backup logs in there and the server
+ // forwards the register to the primary.
+ ASSERT_THROW(client->get_cluster_metadata(), std::exception);
+ ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
iggy::ffi::ClusterMetadata metadata{};
ASSERT_NO_THROW({ metadata = client->get_cluster_metadata(); });
- ASSERT_EQ(metadata.nodes.size(), 1u);
- ASSERT_NO_THROW(client->login_user("iggy", "iggy"));
+ ASSERT_GE(metadata.nodes.size(), 1u);
ASSERT_NO_THROW(client->disconnect());
ASSERT_THROW(client->get_cluster_metadata(), std::exception);
}
diff --git a/foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/LeaderRedirectionSteps.cs b/foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/LeaderRedirectionSteps.cs
index 6987a45..a02934d 100644
--- a/foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/LeaderRedirectionSteps.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/LeaderRedirectionSteps.cs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+using System.Net;
using Apache.Iggy.Configuration;
using Apache.Iggy.Contracts;
using Apache.Iggy.Enums;
@@ -222,8 +223,10 @@
var clientA = GetClient("A");
var clientB = GetClient("B");
- clientA.GetCurrentAddress().ShouldBe(clientB.GetCurrentAddress(),
- "Both clients should be connected to the same server");
+ // A client that never redirected still holds the address it was given (possibly a
+ // hostname), while a redirected one holds the roster address the cluster published.
+ // Both may name the same node, so they are compared once resolved.
+ await AssertSameEndpointAsync(clientA.GetCurrentAddress(), clientB.GetCurrentAddress());
await clientA.PingAsync();
await clientB.PingAsync();
@@ -240,6 +243,24 @@
// ---------- Helpers ----------
+ private static async Task AssertSameEndpointAsync(string left, string right)
+ {
+ static (string Host, int Port) Split(string address)
+ {
+ var separator = address.LastIndexOf(':');
+ return (address[..separator], int.Parse(address[(separator + 1)..]));
+ }
+
+ var (leftHost, leftPort) = Split(left);
+ var (rightHost, rightPort) = Split(right);
+ leftPort.ShouldBe(rightPort, $"Both clients should use the same port, got {left} and {right}");
+
+ var leftIps = await Dns.GetHostAddressesAsync(leftHost);
+ var rightIps = await Dns.GetHostAddressesAsync(rightHost);
+ leftIps.Intersect(rightIps).ShouldNotBeEmpty(
+ $"Both clients should be connected to the same server, got {left} and {right}");
+ }
+
private string ResolveAddressForRole(string role) => role.ToLowerInvariant() switch
{
"leader" => _context.LeaderTcpUrl,
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/ClusterRedirectionTests.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/ClusterRedirectionTests.cs
index 006946f..879b45f 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/ClusterRedirectionTests.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/ClusterRedirectionTests.cs
@@ -17,6 +17,7 @@
using Apache.Iggy.Configuration;
using Apache.Iggy.Contracts;
+using Apache.Iggy.Contracts.Auth;
using Apache.Iggy.Enums;
using Apache.Iggy.Factory;
using Apache.Iggy.Tests.Integrations.Fixtures;
@@ -116,10 +117,25 @@
AutoLoginSettings = new AutoLoginSettings { Enabled = false }
});
await client.ConnectAsync();
- var authResponse = await client.LoginWithPersonalAccessTokenAsync(pat.Token);
+ // The follower verifies the token against its own replicated copy, so it refuses until the mint has
+ // replicated. Fail-closed by design; poll through the window rather than asserting on its width.
+ var replicationDeadline = DateTimeOffset.UtcNow.AddSeconds(10);
+ AuthResponse? authResponse;
+ while (true)
+ {
+ try
+ {
+ authResponse = await client.LoginWithPersonalAccessTokenAsync(pat.Token);
+ break;
+ }
+ catch (Exception) when (DateTimeOffset.UtcNow < replicationDeadline)
+ {
+ await Task.Delay(TimeSpan.FromMilliseconds(250));
+ }
+ }
authResponse.ShouldNotBeNull();
- authResponse.UserId.ShouldBeGreaterThanOrEqualTo(0);
+ authResponse!.UserId.ShouldBeGreaterThanOrEqualTo(0);
var address = client.GetCurrentAddress();
address.ShouldNotBeNullOrEmpty();
diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/IggyTlsConnectionTests.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/IggyTlsConnectionTests.cs
index 2255bbd..a89da78 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/IggyTlsConnectionTests.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/IggyTlsConnectionTests.cs
@@ -67,10 +67,11 @@
ReconnectionSettings = new ReconnectionSettings { Enabled = false }
});
- // The VSR register handshake runs inside ConnectAsync and dies against the TLS listener, so the
- // client never reaches the connected state.
+ // Nothing travels at connect time (the roster read is auth-gated away), so the plaintext client
+ // only dies against the TLS listener on its first request: the server drops the connection
+ // without answering a single byte.
await client.ConnectAsync();
- await Should.ThrowAsync<NotConnectedException>(client.LoginUserAsync("iggy", "iggy"));
+ await Should.ThrowAsync<IggyZeroBytesException>(client.LoginUserAsync("iggy", "iggy"));
}
[Test]
diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
index 81898c0..f70a3b2 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.Vsr.cs
@@ -389,7 +389,11 @@
}
var leaderAddress = ServerAddress.HostPort(currentLeaderNode.Ip, currentLeaderNode.Endpoints.Tcp);
- if (ServerAddress.IsSame(leaderAddress, _currentAddress))
+ // Compare against the endpoint the socket resolved, not the configured string: a client
+ // configured with a hostname is otherwise never "on" the leader the roster names by IP,
+ // and every login would reconnect it to the node it is already talking to.
+ var connectedAddress = _currentRemoteAddress.Length > 0 ? _currentRemoteAddress : _currentAddress;
+ if (ServerAddress.IsSame(leaderAddress, connectedAddress))
{
Interlocked.Exchange(ref _leaderRedirectCount, 0);
return false;
diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
index 146c93e..c68ad62 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
@@ -17,6 +17,7 @@
using System.Buffers;
using System.Buffers.Binary;
+using System.Net;
using System.Net.Security;
using System.Net.Sockets;
using System.Runtime.CompilerServices;
@@ -64,6 +65,12 @@
private readonly ILogger<TcpMessageStream> _logger;
private readonly SemaphoreSlim _sendingSemaphore;
private string _currentAddress = string.Empty;
+
+ // The address the socket actually connected to, as an IP the roster can be compared against.
+ // _currentAddress keeps whatever the caller configured (possibly a hostname the roster never
+ // mentions), so leader comparisons made against it would move a client that is already on the
+ // leader. Written only by the connect loop.
+ private string _currentRemoteAddress = string.Empty;
private X509Certificate2Collection _customCaStore = [];
private volatile bool _disposed;
private int _isConnecting;
@@ -1053,6 +1060,10 @@
await socket.ConnectAsync(host, port, token);
+ _currentRemoteAddress = socket.RemoteEndPoint is IPEndPoint remote
+ ? ServerAddress.HostPort(remote.Address.ToString(), (ushort)remote.Port)
+ : string.Empty;
+
socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true);
socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime, 5);
@@ -1077,12 +1088,9 @@
socket = null;
- if (await RedirectAsync(token))
- {
- await BackoffOrThrowAsync();
- continue;
- }
-
+ // No pre-login roster read: the server auth-gates cluster metadata, so leadership settles after
+ // a sign-in binds a session. A login dialed at a backup still succeeds because the server
+ // forwards the register to the primary.
if (autoLogin && _configuration.AutoLoginSettings.Enabled && !ConsumeSkipAutoLogin())
{
_logger.LogInformation("Auto login enabled. Trying to login with credentials: {Username}",
diff --git a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
index 79a6390..4708ee7 100644
--- a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
+++ b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
@@ -27,7 +27,7 @@
<TargetFrameworks>net8.0;net10.0</TargetFrameworks>
<AssemblyName>Apache.Iggy</AssemblyName>
<RootNamespace>Apache.Iggy</RootNamespace>
- <Version>0.9.0-edge.1</Version>
+ <Version>0.9.0-edge.2</Version>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
</PropertyGroup>
diff --git a/foreign/go/client/tcp/tcp_connect_test.go b/foreign/go/client/tcp/tcp_connect_test.go
index 9e7cf56..822958b 100644
--- a/foreign/go/client/tcp/tcp_connect_test.go
+++ b/foreign/go/client/tcp/tcp_connect_test.go
@@ -180,7 +180,7 @@
}
}
-func TestConnect_DiscoversTheLeaderAndSignsIn(t *testing.T) {
+func TestConnect_SignsInAndThenSettlesLeadership(t *testing.T) {
var server *testListener
server = listenVSR(t, nil, singleNodeHandler(t, func() string { return server.address() }))
@@ -190,14 +190,15 @@
recorded := server.recorded()
require.Len(t, recorded, 2)
- assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[0].code(),
- "leadership is settled before the sign-in")
- assert.Equal(t, uint64(0), recorded[0].sessionID(), "metadata works unauthenticated")
+ assert.Equal(t, vsr.OperationRegister, recorded[0].operation(),
+ "the roster read is auth-gated, so the sign-in comes first")
+ assert.Zero(t, recorded[0].code(), "only a non-replicated frame carries the code")
+ assert.Zero(t, recorded[0].requestID(), "a register is always request zero")
+ assert.Zero(t, recorded[0].sessionID())
- assert.Equal(t, vsr.OperationRegister, recorded[1].operation())
- assert.Zero(t, recorded[1].code(), "only a non-replicated frame carries the code")
- assert.Zero(t, recorded[1].requestID(), "a register is always request zero")
- assert.Zero(t, recorded[1].sessionID())
+ assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[1].code(),
+ "leadership is settled after the sign-in")
+ assert.Equal(t, uint64(128), recorded[1].sessionID(), "the roster read is authenticated")
assert.True(t, client.session.Bound())
assert.Equal(t, uint64(128), client.session.SessionID())
@@ -214,47 +215,96 @@
recorded := server.recorded()
require.Len(t, recorded, 2)
- assert.Equal(t, vsr.OperationRegister, recorded[1].operation())
- assert.Contains(t, string(recorded[1].payload), "token-value")
- assert.Contains(t, string(recorded[1].payload), vsr.SDKName)
+ assert.Equal(t, vsr.OperationRegister, recorded[0].operation())
+ assert.Contains(t, string(recorded[0].payload), "token-value")
+ assert.Contains(t, string(recorded[0].payload), vsr.SDKName)
}
-func TestConnect_ChecksLeadershipEvenWithoutAutoLogin(t *testing.T) {
+func TestConnect_SendsNothingWithoutAutoLogin(t *testing.T) {
var server *testListener
server = listenVSR(t, nil, singleNodeHandler(t, func() string { return server.address() }))
client := newDialingClient(t, server.address())
require.NoError(t, client.Connect(context.Background()))
- recorded := server.recorded()
- require.Len(t, recorded, 1)
- assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[0].code())
+ // The roster read is auth-gated, so an unauthenticated connection has
+ // nothing to settle leadership with; the caller's sign-in does it later.
+ assert.Empty(t, server.recorded())
assert.False(t, client.session.Bound(), "no sign-in happened")
}
-func TestConnect_RedirectsToTheLeaderBeforeSigningIn(t *testing.T) {
+func TestConnect_RedirectsToTheLeaderAfterSigningIn(t *testing.T) {
var leader *testListener
leader = listenVSR(t, nil, singleNodeHandler(t, func() string { return leader.address() }))
- // The follower reports the leader elsewhere, so the client reconnects
- // before it registers anywhere.
+ // The follower completes the sign-in (the server forwards the register to
+ // the primary) and only then reports the leader elsewhere.
follower := listenVSR(t, nil, func(_, _ int, read request) []byte {
- if read.code() == uint32(command.GetClusterMetadataCode) {
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
return clusterMetadataFrame(t, 1, "127.0.0.1:1", leader.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, 512)
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
}
- return replyFrame(vsr.OperationNonReplicated, nil)
})
client := newDialingClient(t, follower.address(),
WithAutoLogin(NewUsernamePasswordCredentials("iggy", "iggy")))
require.NoError(t, client.Connect(context.Background()))
- assert.Len(t, follower.recorded(), 1, "the follower only answered the roster")
+ onFollower := follower.recorded()
+ require.Len(t, onFollower, 2, "the follower answered the sign-in and the roster")
+ assert.Equal(t, vsr.OperationRegister, onFollower[0].operation())
+ assert.Equal(t, uint32(command.GetClusterMetadataCode), onFollower[1].code())
assert.Equal(t, leader.address(), client.currentServerAddress)
onLeader := leader.recorded()
- require.Len(t, onLeader, 2)
- assert.Equal(t, vsr.OperationRegister, onLeader[1].operation())
+ require.Len(t, onLeader, 2, "the sign-in replay is followed by a leader re-check")
+ assert.Equal(t, vsr.OperationRegister, onLeader[0].operation())
+ assert.Equal(t, uint32(command.GetClusterMetadataCode), onLeader[1].code())
+ assert.True(t, client.session.Bound())
+ assert.Equal(t, uint64(128), client.session.SessionID(),
+ "the leader's session superseded the follower's")
+}
+
+func TestConnect_RechecksLeadershipAfterTheRedirectedSignIn(t *testing.T) {
+ var leader *testListener
+ leader = listenVSR(t, nil, singleNodeHandler(t, func() string { return leader.address() }))
+
+ var intermediate *testListener
+ intermediate = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ return clusterMetadataFrame(t, 1, intermediate.address(), leader.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, 256)
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ }
+ })
+
+ var follower *testListener
+ follower = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ return clusterMetadataFrame(t, 1, follower.address(), intermediate.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, 512)
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ }
+ })
+
+ client := newDialingClient(t, follower.address(),
+ WithAutoLogin(NewUsernamePasswordCredentials("iggy", "iggy")))
+ require.NoError(t, client.Connect(context.Background()))
+
+ assert.Equal(t, leader.address(), client.currentServerAddress)
+ assert.Equal(t, 1, follower.connections())
+ assert.Equal(t, 1, intermediate.connections())
+ assert.Equal(t, 1, leader.connections())
assert.True(t, client.session.Bound())
}
@@ -296,9 +346,9 @@
require.Len(t, recorded, 6)
assert.Equal(t, uint32(command.PingCode), recorded[2].code(), "the dropped attempt")
- assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[3].code())
- assert.Equal(t, vsr.OperationRegister, recorded[4].operation(),
+ assert.Equal(t, vsr.OperationRegister, recorded[3].operation(),
"the replay signs in again before it repeats the request")
+ assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[4].code())
assert.Equal(t, uint32(command.PingCode), recorded[5].code())
assert.NotEqual(t, recorded[2].clientID(), recorded[5].clientID(),
@@ -377,10 +427,14 @@
assert.Equal(t, follower.address(), client.currentServerAddress)
assert.True(t, client.session.Bound())
recorded := follower.recorded()
- require.Len(t, recorded, 4)
+ require.Len(t, recorded, 5)
+ assert.Equal(t, vsr.OperationRegister, recorded[0].operation(),
+ "the dialed follower completed the initial sign-in")
assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[1].code())
- assert.Equal(t, vsr.OperationRegister, recorded[2].operation())
- assert.Equal(t, uint32(command.PingCode), recorded[3].code())
+ assert.Equal(t, vsr.OperationRegister, recorded[2].operation(),
+ "the reconnect signed in again on the surviving node")
+ assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[3].code())
+ assert.Equal(t, uint32(command.PingCode), recorded[4].code())
}
func TestConnect_DropsTheConnectionWhenAutomaticSignInFails(t *testing.T) {
@@ -445,12 +499,8 @@
}
func TestExchange_FailsFastWhenAutoLoginIsOff(t *testing.T) {
- var server *testListener
- server = listenVSR(t, nil, func(_, index int, read request) []byte {
- if read.code() == uint32(command.GetClusterMetadataCode) {
- return clusterMetadataFrame(t, 0, server.address())
- }
- if index == 1 {
+ server := listenVSR(t, nil, func(_, index int, read request) []byte {
+ if index == 0 {
return nil
}
return replyFrame(vsr.OperationNonReplicated, nil)
@@ -490,7 +540,7 @@
require.NoError(t, err)
recorded := server.recorded()
- require.Len(t, recorded, 4)
+ require.Len(t, recorded, 5)
assert.Equal(t, vsr.OperationLogout, recorded[2].operation(),
"a re-login logs the live session out first")
assert.Equal(t, vsr.OperationRegister, recorded[3].operation())
@@ -567,7 +617,8 @@
recorded := server.recorded()
require.Len(t, recorded, 3)
- assert.Equal(t, vsr.OperationRegister, recorded[1].operation())
+ assert.Equal(t, vsr.OperationRegister, recorded[0].operation())
+ assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[1].code())
assert.Equal(t, uint32(command.PingCode), recorded[2].code())
assert.True(t, client.session.Bound())
}
diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go
index e8ec0e9..fe662af 100644
--- a/foreign/go/client/tcp/tcp_core.go
+++ b/foreign/go/client/tcp/tcp_core.go
@@ -447,9 +447,10 @@
return append(buf, body...), nil
}
-// connectScoped marks the context of a request the connect flow itself
-// issues. exchange must not reconnect such a request: Connect is already on
-// the stack, and re-entering it has no depth bound.
+// connectScoped marks the context of a request that must not enter the
+// reconnect path: the sign-in flow holding the register lock is on the stack
+// (possibly under Connect), and the reconnect's automatic sign-in would
+// deadlock on that lock or recurse Connect without a bound.
type connectScoped struct{}
// localPreconditionError marks a request that failed before its frame was
@@ -587,9 +588,10 @@
deadline := time.Now().Add(responseReadTimeout)
stamped := false
for {
- // A sign-in owns the whole budget on this connection. The connect flow
- // already pointed it at the leader, and failing over from under the
- // handshake would recurse back into it.
+ // A sign-in owns the whole budget on this connection: any node
+ // completes it (a backup forwards the register to the primary), and
+ // failing over from under the handshake would recurse back into the
+ // sign-in flow.
transientDeadline := deadline
if !isRegisterCode(code) {
if failover := time.Now().Add(failoverCheckInterval); failover.Before(deadline) {
@@ -1014,39 +1016,15 @@
return addresses
}
-// establishSession points the connection at the leader and signs in when
-// auto-login is configured.
+// establishSession signs in when auto-login is configured.
//
-// Leadership is settled before the sign-in, and it is settled even when
-// auto-login is off: register is a consensus operation that a backup answers
-// transiently, so signing in against a follower replays for the whole request
-// budget instead of failing over. Cluster metadata is sessionless and works on
-// the unauthenticated connection.
+// Leader settlement runs after the sign-in (see settleOnLeader): the roster
+// read is auth-gated, so it only works once a login binds a session, and a
+// login dialed at a backup still succeeds because the server forwards the
+// register to the primary. Without auto-login the connection can stay on a
+// backup: once the caller signs in, the first replicated request fails over
+// through the transient-deny path.
func (c *IggyTcpClient) establishSession(ctx context.Context, skipAutoLogin bool) error {
- // The metadata request runs while Connect is on the stack. Reconnecting
- // it would re-enter Connect and recurse without a bound, so its failure
- // unwinds to this Connect instead. The sign-in below keeps the reconnect
- // path: its replay is the documented recovery for a transient register
- // failure, and the replayed sign-in suppresses the automatic one, so the
- // depth is bounded at one nested Connect.
- redirect, err := c.HandleLeaderRedirection(context.WithValue(ctx, connectScoped{}, struct{}{}))
- if err != nil {
- return err
- }
- if redirect {
- if skipAutoLogin {
- // The suppression belongs to the sign-in replay, not to this
- // connection attempt. Connect already consumed the flag, so it is
- // re-armed for the post-redirect Connect; otherwise that nested
- // Connect signs in automatically and the replayed login then
- // commits a second Register.
- c.mtx.Lock()
- c.skipAutoLoginOnce = true
- c.mtx.Unlock()
- }
- return c.Connect(ctx)
- }
-
if !c.config.autoLogin.enabled {
c.logger.Info("Automatic sign-in is disabled.")
return nil
@@ -1058,10 +1036,10 @@
credentials := c.config.autoLogin.credentials
if credentials.personalAccessToken != "" {
- _, err = c.LoginWithPersonalAccessToken(ctx, credentials.personalAccessToken)
+ _, err := c.LoginWithPersonalAccessToken(ctx, credentials.personalAccessToken)
return err
}
- _, err = c.LoginUser(ctx, credentials.username, credentials.password)
+ _, err := c.LoginUser(ctx, credentials.username, credentials.password)
return err
}
diff --git a/foreign/go/client/tcp/tcp_core_review_test.go b/foreign/go/client/tcp/tcp_core_review_test.go
index 15261d6..f9deff4 100644
--- a/foreign/go/client/tcp/tcp_core_review_test.go
+++ b/foreign/go/client/tcp/tcp_core_review_test.go
@@ -130,27 +130,19 @@
require.NoError(t, client.Ping(context.Background()))
}
-func TestConnect_RedirectDuringAReplayedLoginDoesNotAutoSignIn(t *testing.T) {
- var leader *testListener
- leader = listenVSR(t, nil, singleNodeHandler(t, func() string { return leader.address() }))
- follower := listenVSR(t, nil, func(_, _ int, read request) []byte {
- if read.code() == uint32(command.GetClusterMetadataCode) {
- return clusterMetadataFrame(t, 1, "127.0.0.1:1", leader.address())
- }
- return replyFrame(vsr.OperationNonReplicated, nil)
- })
+func TestConnect_SuppressedSignInSendsNothing(t *testing.T) {
+ var server *testListener
+ server = listenVSR(t, nil, singleNodeHandler(t, func() string { return server.address() }))
- client := newDialingClient(t, follower.address(),
+ client := newDialingClient(t, server.address(),
WithAutoLogin(NewUsernamePasswordCredentials("iggy", "iggy")))
// The state a replayed login leaves behind before its reconnect.
client.skipAutoLoginOnce = true
require.NoError(t, client.Connect(context.Background()))
- for _, read := range leader.recorded() {
- assert.NotEqual(t, vsr.OperationRegister, read.operation(),
- "the redirected Connect must keep suppressing the automatic sign-in")
- }
+ assert.Empty(t, server.recorded(),
+ "the replayed login owns the sign-in; Connect must not preempt it")
assert.False(t, client.skipAutoLoginOnce, "the suppression is consumed exactly once")
}
diff --git a/foreign/go/client/tcp/tcp_session_management.go b/foreign/go/client/tcp/tcp_session_management.go
index 3ca38e4..d409f7d 100644
--- a/foreign/go/client/tcp/tcp_session_management.go
+++ b/foreign/go/client/tcp/tcp_session_management.go
@@ -44,13 +44,8 @@
return c.register(ctx, uint32(command.LoginRegisterWithPATCode), body)
}
-// register runs the sign-in handshake and binds the session the server
-// assigned. Leadership is already settled by the connect flow, so no
-// redirection happens here.
-//
-// A failed sign-in never writes the session state: a server-side reject leaves
-// the existing session untouched, and a connection that dies mid-attempt is
-// already reset by invalidateConnLocked.
+// register runs the sign-in handshake, binds the session the server assigned,
+// and settles the connection on the cluster leader.
func (c *IggyTcpClient) register(ctx context.Context, code uint32, body []byte) (*iggcon.IdentityInfo, error) {
// One sign-in at a time. BeginRegister runs inside the exchange lock but
// Bind runs after it, so two interleaved sign-ins would let the second
@@ -66,6 +61,28 @@
return nil, err
}
+ identity, err := c.signIn(ctx, code, body)
+ if err != nil {
+ return nil, err
+ }
+
+ settled, err := c.settleOnLeader(ctx, code, body)
+ if err != nil {
+ return nil, err
+ }
+ if settled != nil {
+ return settled, nil
+ }
+ return identity, nil
+}
+
+// signIn runs one sign-in exchange on the current connection and binds the
+// session the server assigned.
+//
+// A failed sign-in never writes the session state: a server-side reject leaves
+// the existing session untouched, and a connection that dies mid-attempt is
+// already reset by invalidateConnLocked.
+func (c *IggyTcpClient) signIn(ctx context.Context, code uint32, body []byte) (*iggcon.IdentityInfo, error) {
bp := acquireRequestBuf()
defer releaseRequestBuf(bp)
frame := append(reserveHeader(*bp), body...)
@@ -103,6 +120,49 @@
return &iggcon.IdentityInfo{UserId: registered.UserID}, nil
}
+// settleOnLeader moves a freshly signed-in session to the cluster leader.
+//
+// Only the leader accepts replicated commands, and the roster read is
+// auth-gated, so the topology cannot be inspected before a login binds a
+// session. A login dialed at a backup still succeeds (the server forwards the
+// register to the primary); this settlement decides where later requests
+// land, not whether the sign-in works. The redirect drops the fresh session
+// along with the socket, so the sign-in is replayed on the leader and its
+// identity supersedes the dialed node's. Leadership can move between the
+// roster read and the replay, so each freshly bound hop rechecks the roster
+// under the shared redirect budget.
+//
+// Returns nil when the client stays where it is.
+func (c *IggyTcpClient) settleOnLeader(ctx context.Context, code uint32, body []byte) (*iggcon.IdentityInfo, error) {
+ var settled *iggcon.IdentityInfo
+ for {
+ // The roster read runs while register holds the sign-in lock, so it must
+ // not enter the reconnect path: the reconnect's automatic sign-in would
+ // deadlock on that lock. The connect scope fails it fast instead.
+ redirect, err := c.HandleLeaderRedirection(
+ context.WithValue(ctx, connectScoped{}, struct{}{}))
+ if err != nil || !redirect {
+ return settled, err
+ }
+
+ // The replayed sign-in below owns the session; the redirected Connect
+ // must not sign in on its own, or the replay commits a second Register.
+ c.mtx.Lock()
+ c.skipAutoLoginOnce = true
+ c.mtx.Unlock()
+ if err := c.Connect(ctx); err != nil {
+ c.mtx.Lock()
+ c.skipAutoLoginOnce = false
+ c.mtx.Unlock()
+ return nil, err
+ }
+ settled, err = c.signIn(ctx, code, body)
+ if err != nil {
+ return nil, err
+ }
+ }
+}
+
// endBoundSession logs out a live session before a re-login, so the server
// drops its client-table entry instead of leaving it to be fenced.
func (c *IggyTcpClient) endBoundSession(ctx context.Context) error {
diff --git a/foreign/go/contracts/version.go b/foreign/go/contracts/version.go
index ba39f3a..03840a8 100644
--- a/foreign/go/contracts/version.go
+++ b/foreign/go/contracts/version.go
@@ -17,4 +17,4 @@
package iggcon
-const Version = "0.9.0-edge.1"
+const Version = "0.9.0-edge.2"
diff --git a/foreign/go/internal/util/leader_aware.go b/foreign/go/internal/util/leader_aware.go
index ce4e86c..b840322 100644
--- a/foreign/go/internal/util/leader_aware.go
+++ b/foreign/go/internal/util/leader_aware.go
@@ -25,12 +25,40 @@
"net"
"strconv"
"strings"
+ "time"
iggcon "github.com/apache/iggy/foreign/go/contracts"
)
+// How long a transiently leaderless cluster is polled for an elected leader
+// before proceeding on the current node anyway. Vars, not consts, so tests
+// can shrink the budget.
+var (
+ leaderlessWaitBudget = 5 * time.Second
+ leaderlessPollInterval = 250 * time.Millisecond
+)
+
+// leaderOutcome is one leader-check verdict from a cluster-metadata snapshot.
+type leaderOutcome int
+
+const (
+ // The current node is the leader (or the cluster is single-node).
+ leaderIsCurrent leaderOutcome = iota
+ // A healthy leader exists elsewhere; reconnect to it.
+ redirectToLeader
+ // No healthy leader is marked (e.g. mid-election).
+ noLeader
+)
+
// CheckAndRedirectToLeader queries the client for cluster metadata and returns
// an address to redirect to (empty string means no redirection needed).
+//
+// A cluster can be transiently leaderless: a restarted node cedes the
+// primaryship its stale view assigns it, and until the peers' election
+// completes the roster reports no leader. That window is roughly one
+// heartbeat timeout; poll through it instead of proceeding leaderless (a
+// replicated request against a non-primary replays for its whole read
+// timeout).
func CheckAndRedirectToLeader(
ctx context.Context,
c iggcon.Client,
@@ -40,30 +68,59 @@
) (string, []string, error) {
logger.Debug("Checking cluster metadata for leader detection")
- meta, err := c.GetClusterMetadata(ctx)
- if err != nil {
- if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+ deadline := time.Now().Add(leaderlessWaitBudget)
+ for {
+ meta, err := c.GetClusterMetadata(ctx)
+ if err != nil {
+ if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+ return "", nil, err
+ }
+ // The read is auth-gated, so an Unauthenticated answer means the
+ // session died between the sign-in and this check. Like any other
+ // failure it keeps the current node; the caller's next request
+ // surfaces the eviction.
+ logger.Warn(
+ "Failed to get cluster metadata, connection will continue on server node",
+ "error", err,
+ "current_address", currentAddress,
+ )
+ return "", nil, nil
+ }
+
+ logger.Debug(
+ "Got cluster metadata",
+ "nodes", len(meta.Nodes),
+ "cluster", meta.Name,
+ )
+ addresses, err := clusterAddresses(meta, transport)
+ if err != nil {
return "", nil, err
}
- logger.Warn(
- "Failed to get cluster metadata, connection will continue on server node",
- "error", err,
- "current_address", currentAddress,
- )
- return "", nil, nil
+ outcome, leader, err := processClusterMetadata(meta, currentAddress, transport, logger)
+ if err != nil {
+ return "", nil, err
+ }
+ switch outcome {
+ case redirectToLeader:
+ return leader, addresses, nil
+ case leaderIsCurrent:
+ return "", addresses, nil
+ case noLeader:
+ if time.Now().After(deadline) {
+ logger.Warn(
+ "No active leader found in cluster metadata within the wait budget, connection will continue on server node",
+ "wait_budget", leaderlessWaitBudget,
+ "current_address", currentAddress,
+ )
+ return "", addresses, nil
+ }
+ select {
+ case <-ctx.Done():
+ return "", nil, ctx.Err()
+ case <-time.After(leaderlessPollInterval):
+ }
+ }
}
-
- logger.Debug(
- "Got cluster metadata",
- "nodes", len(meta.Nodes),
- "cluster", meta.Name,
- )
- addresses, err := clusterAddresses(meta, transport)
- if err != nil {
- return "", nil, err
- }
- leader, err := processClusterMetadata(meta, currentAddress, transport, logger)
- return leader, addresses, err
}
func clusterAddresses(metadata *iggcon.ClusterMetadata, transport iggcon.Protocol) ([]string, error) {
@@ -82,13 +139,13 @@
return addresses, nil
}
-func processClusterMetadata(metadata *iggcon.ClusterMetadata, currentAddress string, transport iggcon.Protocol, logger *slog.Logger) (string, error) {
+func processClusterMetadata(metadata *iggcon.ClusterMetadata, currentAddress string, transport iggcon.Protocol, logger *slog.Logger) (leaderOutcome, string, error) {
if len(metadata.Nodes) == 1 {
logger.Debug(
"Single-node cluster detected, no leader redirection needed",
"node", metadata.Nodes[0].Name,
)
- return "", nil
+ return leaderIsCurrent, "", nil
}
var leader *iggcon.ClusterNode
@@ -101,16 +158,12 @@
}
if leader == nil {
- logger.Warn(
- "No active leader found in cluster metadata, connection will continue on server node",
- "current_address", currentAddress,
- )
- return "", nil
+ return noLeader, "", nil
}
leaderAddress, err := clusterNodeAddress(leader, transport)
if err != nil {
- return "", err
+ return leaderIsCurrent, "", err
}
logger.Debug(
"Found leader node",
@@ -125,11 +178,11 @@
"current_address", currentAddress,
"leader_address", leaderAddress,
)
- return leaderAddress, nil
+ return redirectToLeader, leaderAddress, nil
}
logger.Debug("Already connected to leader", "current_address", currentAddress)
- return "", nil
+ return leaderIsCurrent, "", nil
}
func clusterNodeAddress(node *iggcon.ClusterNode, transport iggcon.Protocol) (string, error) {
diff --git a/foreign/go/internal/util/leader_aware_test.go b/foreign/go/internal/util/leader_aware_test.go
index 7624b16..3fa8e10 100644
--- a/foreign/go/internal/util/leader_aware_test.go
+++ b/foreign/go/internal/util/leader_aware_test.go
@@ -21,6 +21,7 @@
"context"
"log/slog"
"testing"
+ "time"
"github.com/apache/iggy/foreign/go/contracts"
ierror "github.com/apache/iggy/foreign/go/errors"
@@ -97,7 +98,18 @@
assert.Equal(t, []string{"10.0.0.1:8090"}, addresses)
}
+// shrinkLeaderlessBudget makes the leaderless poll converge fast in tests.
+func shrinkLeaderlessBudget(t *testing.T, budget, interval time.Duration) {
+ t.Helper()
+ previousBudget, previousInterval := leaderlessWaitBudget, leaderlessPollInterval
+ leaderlessWaitBudget, leaderlessPollInterval = budget, interval
+ t.Cleanup(func() {
+ leaderlessWaitBudget, leaderlessPollInterval = previousBudget, previousInterval
+ })
+}
+
func TestCheckAndRedirectToLeader_StaysPutWhenNoHealthyLeaderExists(t *testing.T) {
+ shrinkLeaderlessBudget(t, 0, time.Millisecond)
client := &metadataClient{metadata: &iggcon.ClusterMetadata{
Nodes: []iggcon.ClusterNode{
clusterNode("10.0.0.1", 8090, iggcon.RoleLeader, iggcon.Unreachable),
@@ -112,6 +124,38 @@
"only healthy nodes are reconnect candidates")
}
+// electingClient answers a leaderless roster until the election settles.
+type electingClient struct {
+ iggcon.Client
+ reads int
+ settleAt int
+}
+
+func (c *electingClient) GetClusterMetadata(context.Context) (*iggcon.ClusterMetadata, error) {
+ c.reads++
+ role := iggcon.RoleFollower
+ if c.reads >= c.settleAt {
+ role = iggcon.RoleLeader
+ }
+ return &iggcon.ClusterMetadata{
+ Nodes: []iggcon.ClusterNode{
+ clusterNode("10.0.0.1", 8090, role, iggcon.Healthy),
+ clusterNode("10.0.0.2", 8090, iggcon.RoleFollower, iggcon.Healthy),
+ },
+ }, nil
+}
+
+func TestCheckAndRedirectToLeader_PollsThroughALeaderlessElection(t *testing.T) {
+ shrinkLeaderlessBudget(t, time.Second, time.Millisecond)
+ client := &electingClient{settleAt: 3}
+
+ leader, _, err := checkRedirect(t, client, "10.0.0.2:8090")
+ require.NoError(t, err)
+ assert.Equal(t, "10.0.0.1:8090", leader,
+ "the poll rides through the election instead of settling leaderless")
+ assert.Equal(t, 3, client.reads)
+}
+
func TestCheckAndRedirectToLeader_SwallowsAMetadataFailure(t *testing.T) {
client := &metadataClient{err: ierror.ErrDisconnected}
diff --git a/foreign/go/tests/e2e_test.go b/foreign/go/tests/e2e_test.go
index 761c871..93a7c02 100644
--- a/foreign/go/tests/e2e_test.go
+++ b/foreign/go/tests/e2e_test.go
@@ -24,6 +24,7 @@
"github.com/apache/iggy/foreign/go/client/tcp"
iggcon "github.com/apache/iggy/foreign/go/contracts"
+ ierror "github.com/apache/iggy/foreign/go/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -259,15 +260,17 @@
assert.NotNil(t, streams)
}
-func TestE2E_PingWorksBeforeSigningIn(t *testing.T) {
+func TestE2E_OnlyPingWorksBeforeSigningIn(t *testing.T) {
connected := newClient(t)
require.NoError(t, connected.Ping(context.Background()))
+ // Ping is the only command the server answers on an unbound connection.
+ // The cluster roster is auth-gated so that an unauthenticated reader cannot
+ // enumerate the private network topology.
metadata, err := connected.GetClusterMetadata(context.Background())
- require.NoError(t, err, "cluster metadata is readable before the sign-in")
- require.NotNil(t, metadata)
- assert.NotEmpty(t, metadata.Nodes)
+ require.ErrorIs(t, err, ierror.ErrUnauthenticated)
+ assert.Nil(t, metadata, "no roster leaks to an unauthenticated reader")
}
func TestE2E_LogoutEndsTheSession(t *testing.T) {
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java
index 04f46ec..e2673d9 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java
@@ -624,20 +624,17 @@
}
/**
- * Serializes pre-login leader discovery and Register across concurrent
- * logins. A queued login waits for the entire in-flight transaction, then
- * checks the roster from the connection that transaction published. Each
- * transaction has a fresh redirection budget, so hitting the cap affects
- * only that login. Metadata and retargeting failures retain best-effort
- * behavior and let Register run against the current target.
+ * Serializes Register and authenticated leader settlement across concurrent
+ * logins. A queued login waits for the entire in-flight transaction. Each
+ * redirect drops the bound session, so the login is replayed before the next
+ * roster read. Each transaction has a fresh redirection budget.
*/
CompletableFuture<IdentityInfo> loginOnLeader(Supplier<CompletableFuture<IdentityInfo>> loginAttempt) {
CompletableFuture<Void> gate = new CompletableFuture<>();
CompletableFuture<Void> previous = loginChain.getAndSet(gate);
LeaderRedirectionState redirectionState = new LeaderRedirectionState();
- CompletableFuture<IdentityInfo> transaction = previous.thenCompose(
- ignored -> redirectToLeader(redirectionState))
- .thenCompose(ignored -> loginAttempt.get());
+ CompletableFuture<IdentityInfo> transaction =
+ previous.thenCompose(ignored -> loginAndSettleOnLeader(loginAttempt, redirectionState));
CompletableFuture<IdentityInfo> callerFuture = new CompletableFuture<>();
transaction.whenComplete((identity, error) -> {
gate.complete(null);
@@ -650,11 +647,48 @@
return callerFuture;
}
+ private CompletableFuture<IdentityInfo> loginAndSettleOnLeader(
+ Supplier<CompletableFuture<IdentityInfo>> loginAttempt, LeaderRedirectionState redirectionState) {
+ return loginAttempt.get().thenCompose(identity -> {
+ ConnectionInfo currentTarget = connectionInfo;
+ return findLeaderElsewhere(currentTarget).thenCompose(leaderTarget -> {
+ if (leaderTarget.isEmpty()) {
+ return CompletableFuture.completedFuture(identity);
+ }
+ if (!redirectionState.canRedirect()) {
+ log.warn(
+ "Maximum leader redirections ({}) reached, connection will continue on server node {}",
+ LeaderAwareness.MAX_LEADER_REDIRECTS,
+ currentTarget.serverAddress());
+ return CompletableFuture.completedFuture(identity);
+ }
+ return retarget(leaderTarget.get())
+ .handle((ignored, error) -> {
+ if (error != null) {
+ log.warn(
+ "Failed to reconnect to leader at {}: {}, connection will continue"
+ + " on server node {}",
+ leaderTarget.get().serverAddress(),
+ error.getMessage(),
+ currentTarget.serverAddress());
+ return CompletableFuture.completedFuture(identity);
+ }
+ redirectionState.recordRedirect();
+ return loginAndSettleOnLeader(loginAttempt, redirectionState);
+ })
+ .thenCompose(Function.identity());
+ });
+ });
+ }
+
/**
- * One authentication-independent discovery hop. When the roster names a
- * healthy leader elsewhere, reconnect to it and re-check from the new
- * node, since mid-election metadata can point at a node that is itself not
- * the leader. Register is sent only after this bounded process settles.
+ * One discovery hop for transient failover. When the roster names a healthy leader elsewhere,
+ * reconnect to it and re-check from the new node, since mid-election
+ * metadata can point at a node that is itself not the leader. Register is
+ * sent only after this bounded process settles. Reading the roster needs a
+ * bound session, so a connection that has none fails the fetch locally and
+ * stays where it is. Login settlement uses {@link #loginAndSettleOnLeader}
+ * so every redirected connection binds before its next roster read.
*/
private CompletableFuture<Void> redirectToLeader(LeaderRedirectionState redirectionState) {
ConnectionInfo currentTarget = connectionInfo;
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
index c421947..102e236 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java
@@ -620,9 +620,9 @@
}
/**
- * Ping and cluster metadata are the only sessionless bootstrap commands.
- * Cluster metadata must be available before Register so a VSR client can
- * select the leader; every other non-login command requires a bound
+ * Ping is the only command the server answers without a bound session. The
+ * cluster roster is auth-gated as well, so an unauthenticated caller cannot
+ * enumerate the topology and leader selection can only run on a bound
* session.
*/
private static boolean requiresAuthentication(int commandCode) {
@@ -630,8 +630,7 @@
}
private static boolean isAllowedBeforeAuthentication(int commandCode) {
- return commandCode == CommandCode.System.PING.getValue()
- || commandCode == CommandCode.System.GET_CLUSTER_METADATA.getValue();
+ return commandCode == CommandCode.System.PING.getValue();
}
private void sendFrame(
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncConnectionPoolAuthTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncConnectionPoolAuthTest.java
index 5a5c4e5..17a7507 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncConnectionPoolAuthTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/AsyncConnectionPoolAuthTest.java
@@ -78,15 +78,16 @@
}
@Test
- @DisplayName("should allow only bootstrap commands before login")
- void shouldAllowOnlyBootstrapCommandsBeforeLogin() throws Exception {
+ @DisplayName("should allow only ping before login")
+ void shouldAllowOnlyPingBeforeLogin() throws Exception {
// when
var ping = client.system().ping().get(5, TimeUnit.SECONDS);
- var metadata = client.system().getClusterMetadata().get(5, TimeUnit.SECONDS);
// then
assertThat(ping).isEqualTo("pong");
- assertThat(metadata).isNotNull();
+ assertThatThrownBy(() -> client.system().getClusterMetadata().get(5, TimeUnit.SECONDS))
+ .isInstanceOf(ExecutionException.class)
+ .hasCauseInstanceOf(IggyNotConnectedException.class);
assertThatThrownBy(() -> client.streams().getStreams().get(5, TimeUnit.SECONDS))
.isInstanceOf(ExecutionException.class)
.hasCauseInstanceOf(IggyNotConnectedException.class);
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientLoginRoutingTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientLoginRoutingTest.java
index b039741..7afaf91 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientLoginRoutingTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientLoginRoutingTest.java
@@ -41,7 +41,7 @@
private static final IdentityInfo SECOND_IDENTITY = new IdentityInfo(2L, Optional.empty());
@Test
- void shouldLoginOnceAfterAllLeaderRedirects() {
+ void shouldLoginAfterEveryLeaderRedirect() {
var client = new TestClient();
client.enqueueLeader(new ConnectionInfo("leader-a", 8091));
client.enqueueLeader(new ConnectionInfo("leader-b", 8092));
@@ -56,10 +56,11 @@
.join();
assertThat(identity).isEqualTo(FIRST_IDENTITY);
- assertThat(attempts).hasValue(1);
+ assertThat(attempts).hasValue(3);
assertThat(client.retargets)
.containsExactly(new ConnectionInfo("leader-a", 8091), new ConnectionInfo("leader-b", 8092));
- assertThat(client.events).containsExactly("discover", "retarget", "discover", "retarget", "discover", "login");
+ assertThat(client.events)
+ .containsExactly("login", "discover", "retarget", "login", "discover", "retarget", "login", "discover");
}
@Test
@@ -78,14 +79,14 @@
return CompletableFuture.completedFuture(SECOND_IDENTITY);
});
- assertThat(client.events).containsExactly("discover", "first-login");
+ assertThat(client.events).containsExactly("first-login");
assertThat(secondLogin).isNotDone();
firstAttempt.complete(FIRST_IDENTITY);
assertThat(firstLogin.join()).isEqualTo(FIRST_IDENTITY);
assertThat(secondLogin.join()).isEqualTo(SECOND_IDENTITY);
- assertThat(client.events).containsExactly("discover", "first-login", "discover", "second-login");
+ assertThat(client.events).containsExactly("first-login", "discover", "second-login", "discover");
}
@Test
@@ -105,7 +106,7 @@
assertThatThrownBy(failedLogin::join)
.isInstanceOf(CompletionException.class)
.hasCause(routingFailure);
- assertThat(failedAttemptCount).hasValue(0);
+ assertThat(failedAttemptCount).hasValue(1);
assertThat(nextLogin.join()).isEqualTo(SECOND_IDENTITY);
}
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java
index 442dfd5..409e831 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncTcpConnectionConcurrencyTest.java
@@ -61,7 +61,6 @@
private static final int OPERATION_LOGOUT = 3;
private static final int OPERATION_SEND_MESSAGES = 160;
private static final int PING_CODE = 1;
- private static final int GET_CLUSTER_METADATA_CODE = 12;
private static final int LOGIN_CODE = 38;
private static final int LOGOUT_CODE = 39;
private static final int SEND_MESSAGES_CODE = 101;
@@ -245,9 +244,9 @@
assertNoRequest(input, socket);
writeResponse(output, register, registerBody());
- Request metadataDuringRegister = readRequest(input);
- assertThat(metadataDuringRegister.commandCode()).isEqualTo(GET_CLUSTER_METADATA_CODE);
- writeResponse(output, metadataDuringRegister, new byte[0]);
+ Request pingDuringRegister = readRequest(input);
+ assertThat(pingDuringRegister.commandCode()).isEqualTo(PING_CODE);
+ writeResponse(output, pingDuringRegister, new byte[0]);
Request logout = readRequest(input);
assertThat(logout.operation()).isEqualTo(OPERATION_LOGOUT);
@@ -256,9 +255,9 @@
assertNoRequest(input, socket);
writeResponse(output, logout, new byte[0]);
- Request metadataDuringLogout = readRequest(input);
- assertThat(metadataDuringLogout.commandCode()).isEqualTo(GET_CLUSTER_METADATA_CODE);
- writeResponse(output, metadataDuringLogout, new byte[0]);
+ Request pingDuringLogout = readRequest(input);
+ assertThat(pingDuringLogout.commandCode()).isEqualTo(PING_CODE);
+ writeResponse(output, pingDuringLogout, new byte[0]);
} catch (InterruptedException error) {
Thread.currentThread().interrupt();
registerRead.completeExceptionally(error);
@@ -276,19 +275,19 @@
connection.connect().get(5, TimeUnit.SECONDS);
CompletableFuture<ByteBuf> login = connection.send(LOGIN_CODE, loginPayload());
registerRead.get(5, TimeUnit.SECONDS);
- CompletableFuture<ByteBuf> metadataDuringRegister =
- connection.send(GET_CLUSTER_METADATA_CODE, Unpooled.EMPTY_BUFFER);
+ // Ping needs no session, so the probe waits only for the lease
+ // and never for lazy authentication.
+ CompletableFuture<ByteBuf> pingDuringRegister = connection.send(PING_CODE, Unpooled.EMPTY_BUFFER);
registerConcurrentSendStarted.complete(null);
login.get(5, TimeUnit.SECONDS).release();
- metadataDuringRegister.get(5, TimeUnit.SECONDS).release();
+ pingDuringRegister.get(5, TimeUnit.SECONDS).release();
CompletableFuture<ByteBuf> logout = connection.send(LOGOUT_CODE, Unpooled.EMPTY_BUFFER);
logoutRead.get(5, TimeUnit.SECONDS);
- CompletableFuture<ByteBuf> metadataDuringLogout =
- connection.send(GET_CLUSTER_METADATA_CODE, Unpooled.EMPTY_BUFFER);
+ CompletableFuture<ByteBuf> pingDuringLogout = connection.send(PING_CODE, Unpooled.EMPTY_BUFFER);
logoutConcurrentSendStarted.complete(null);
logout.get(5, TimeUnit.SECONDS).release();
- metadataDuringLogout.get(5, TimeUnit.SECONDS).release();
+ pingDuringLogout.get(5, TimeUnit.SECONDS).release();
} finally {
connection.close().get(5, TimeUnit.SECONDS);
}
diff --git a/foreign/node/package-lock.json b/foreign/node/package-lock.json
index 3070ba2..517e639 100644
--- a/foreign/node/package-lock.json
+++ b/foreign/node/package-lock.json
@@ -1,12 +1,12 @@
{
"name": "apache-iggy",
- "version": "0.10.0-edge.1",
+ "version": "0.10.0-edge.2",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "apache-iggy",
- "version": "0.10.0-edge.1",
+ "version": "0.10.0-edge.2",
"license": "Apache-2.0",
"dependencies": {
"debug": "4.4.3",
diff --git a/foreign/node/package.json b/foreign/node/package.json
index dc3a9e6..5696c58 100644
--- a/foreign/node/package.json
+++ b/foreign/node/package.json
@@ -1,7 +1,7 @@
{
"name": "apache-iggy",
"type": "module",
- "version": "0.10.0-edge.1",
+ "version": "0.10.0-edge.2",
"description": "Official Apache Iggy NodeJS SDK",
"keywords": [
"iggy",
diff --git a/foreign/node/src/client/client.socket.test.ts b/foreign/node/src/client/client.socket.test.ts
index 33e841f..2f72cb6 100644
--- a/foreign/node/src/client/client.socket.test.ts
+++ b/foreign/node/src/client/client.socket.test.ts
@@ -209,6 +209,19 @@
reconnect: { enabled: false, interval: 100, maxRetries: 1 }
});
+/** Shrinks the leaderless poll so a test observes it without waiting on it. */
+const compressLeaderlessPoll = (
+ client: CommandResponseStream,
+ budget: number
+): void => {
+ const settlement = client as unknown as {
+ leaderlessWaitBudget: number,
+ leaderlessPollInterval: number
+ };
+ settlement.leaderlessWaitBudget = budget;
+ settlement.leaderlessPollInterval = 1;
+};
+
describe('VSR client socket', () => {
it('exchanges VSR frames over TLS', async () => {
const server = await startVsrServer(
@@ -252,13 +265,18 @@
(frame) => frame.readUInt8(REQUEST_OFFSET.operation)
);
assert.deepEqual(operations, [
- Operation.NonReplicated,
Operation.Register,
+ Operation.NonReplicated,
Operation.NonReplicated
]);
- const register = server.frames[1];
+ const register = server.frames[0];
assert.equal(register.readBigUInt64LE(REQUEST_OFFSET.request), 0n);
assert.equal(register.readBigUInt64LE(REQUEST_OFFSET.session), 0n);
+ const settlement = server.frames[1];
+ assert.equal(
+ settlement.readUInt32LE(REQUEST_OFFSET.reserved),
+ COMMAND_CODE.GetClusterMetadata
+ );
const request = server.frames[2];
assert.equal(
request.readBigUInt64LE(REQUEST_OFFSET.session),
@@ -371,29 +389,31 @@
}
});
- it('redirects a direct login to the advertised leader before registering',
+ it('redirects a login to the advertised leader and registers there',
async () => {
const leader = await startVsrServer(
(frame, socket) => singleNodeHandler(leader.port)(frame, socket)
);
const follower = await startVsrServer((frame, socket) => {
- const code = frame.readUInt32LE(REQUEST_OFFSET.reserved);
- if (code === COMMAND_CODE.GetClusterMetadata) {
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ if (frame.readUInt32LE(REQUEST_OFFSET.reserved) ===
+ COMMAND_CODE.GetClusterMetadata) {
socket.write(replyFrame(
Operation.NonReplicated,
twoNodeMetadataBody(follower.port, leader.port)
));
return;
}
- socket.write(replyFrame(
- frame.readUInt8(REQUEST_OFFSET.operation),
- Buffer.alloc(0),
- 3
- ));
+ socket.write(replyFrame(operation, Buffer.alloc(0), 58));
});
const client = new CommandResponseStream(vsrConfig(follower.port));
try {
- const response = await client.sendCommand(COMMAND_CODE.LoginUser,
+ const response = await client.sendCommand(
+ COMMAND_CODE.LoginUser,
Buffer.concat([
Buffer.from([4]),
Buffer.from('iggy'),
@@ -405,14 +425,26 @@
const followerOperations = follower.frames.map(
(frame) => frame.readUInt8(REQUEST_OFFSET.operation)
);
- assert.deepEqual(followerOperations, [Operation.NonReplicated]);
+ assert.deepEqual(followerOperations, [
+ Operation.Register,
+ Operation.NonReplicated
+ ]);
+
+ await client.sendCommand(60_018, Buffer.alloc(0));
+
const leaderOperations = leader.frames.map(
(frame) => frame.readUInt8(REQUEST_OFFSET.operation)
);
assert.deepEqual(leaderOperations, [
+ Operation.Register,
Operation.NonReplicated,
- Operation.Register
+ Operation.NonReplicated
]);
+ assert.equal(
+ leader.frames[2].readUInt32LE(REQUEST_OFFSET.reserved),
+ 60_018
+ );
+ assert.equal(follower.frames.length, 2);
} finally {
client.destroy();
await leader.close();
@@ -420,6 +452,204 @@
}
});
+ it('rechecks leadership after a redirected login', async () => {
+ const leader = await startVsrServer(
+ (frame, socket) => singleNodeHandler(leader.port)(frame, socket)
+ );
+ const intermediate = await startVsrServer((frame, socket) => {
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ socket.write(replyFrame(
+ Operation.NonReplicated,
+ twoNodeMetadataBody(intermediate.port, leader.port)
+ ));
+ });
+ const follower = await startVsrServer((frame, socket) => {
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ socket.write(replyFrame(
+ Operation.NonReplicated,
+ twoNodeMetadataBody(follower.port, intermediate.port)
+ ));
+ });
+ const client = new CommandResponseStream(vsrConfig(follower.port));
+ try {
+ await client.authenticate(vsrConfig(follower.port).credentials);
+ await client.sendCommand(60_019, Buffer.alloc(0));
+
+ assert.deepEqual(
+ follower.frames.map((frame) => frame.readUInt8(REQUEST_OFFSET.operation)),
+ [Operation.Register, Operation.NonReplicated]
+ );
+ assert.deepEqual(
+ intermediate.frames.map(
+ (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
+ ),
+ [Operation.Register, Operation.NonReplicated]
+ );
+ assert.equal(
+ leader.frames[2].readUInt32LE(REQUEST_OFFSET.reserved),
+ 60_019
+ );
+ } finally {
+ client.destroy();
+ await follower.close();
+ await intermediate.close();
+ await leader.close();
+ }
+ });
+
+ it('keeps a single-node login on its node', async () => {
+ const server = await startVsrServer(
+ (frame, socket) => singleNodeHandler(server.port)(frame, socket)
+ );
+ const client = new CommandResponseStream(vsrConfig(server.port));
+ try {
+ await client.authenticate(vsrConfig(server.port).credentials);
+
+ const operations = server.frames.map(
+ (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
+ );
+ assert.deepEqual(operations, [
+ Operation.Register,
+ Operation.NonReplicated
+ ]);
+ assert.equal(
+ server.frames[1].readUInt32LE(REQUEST_OFFSET.reserved),
+ COMMAND_CODE.GetClusterMetadata
+ );
+ assert.equal(client.isAuthenticated, true);
+ } finally {
+ client.destroy();
+ await server.close();
+ }
+ });
+
+ it('polls a leaderless roster before redirecting to the elected leader',
+ async () => {
+ const leader = await startVsrServer(
+ (frame, socket) => singleNodeHandler(leader.port)(frame, socket)
+ );
+ let rosterReads = 0;
+ const follower = await startVsrServer((frame, socket) => {
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ rosterReads += 1;
+ // The first answer is mid-election: neither node holds the leader role.
+ socket.write(replyFrame(
+ Operation.NonReplicated,
+ rosterReads === 1
+ ? twoNodeMetadataBody(follower.port, leader.port, 1)
+ : twoNodeMetadataBody(follower.port, leader.port)
+ ));
+ });
+ const client = new CommandResponseStream(vsrConfig(follower.port));
+ compressLeaderlessPoll(client, 1_000);
+ try {
+ await client.authenticate(vsrConfig(follower.port).credentials);
+
+ assert.equal(rosterReads, 2);
+ const leaderOperations = leader.frames.map(
+ (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
+ );
+ assert.deepEqual(leaderOperations, [
+ Operation.Register,
+ Operation.NonReplicated
+ ]);
+ assert.equal(client.isAuthenticated, true);
+ } finally {
+ client.destroy();
+ await leader.close();
+ await follower.close();
+ }
+ });
+
+ it('keeps a login alive when the session dies mid-poll', async () => {
+ let rosterReads = 0;
+ const server = await startVsrServer((frame, socket) => {
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ rosterReads += 1;
+ // The roster stays leaderless and the session dies in the same breath.
+ // Polling on without a session would re-enter authentication, which
+ // awaits the login being settled, and the login would never return.
+ socket.write(Buffer.concat([
+ replyFrame(
+ Operation.NonReplicated,
+ twoNodeMetadataBody(server.port, server.port, 1)
+ ),
+ evictionFrame(EvictionReason.NoSession)
+ ]));
+ });
+ const client = new CommandResponseStream(vsrConfig(server.port));
+ compressLeaderlessPoll(client, 1_000);
+ try {
+ const outcome = await Promise.race([
+ client.authenticate(vsrConfig(server.port).credentials)
+ .then(() => 'authenticated'),
+ // Unreferenced so a passing run is not held open by the stall timer.
+ new Promise((resolve) => {
+ setTimeout(() => resolve('stalled'), 500).unref();
+ })
+ ]);
+
+ assert.equal(outcome, 'authenticated');
+ assert.equal(rosterReads, 1);
+ } finally {
+ client.destroy();
+ await server.close();
+ }
+ });
+
+ it('keeps a login on its node when no leader appears in the budget',
+ async () => {
+ const unavailable = await startVsrServer(() => {});
+ const unavailablePort = unavailable.port;
+ await unavailable.close();
+ // The roster marks its leader-role node unhealthy on a dead port, so a
+ // redirect would fail the dial instead of passing unnoticed.
+ const server = await startVsrServer((frame, socket) => {
+ const operation = frame.readUInt8(REQUEST_OFFSET.operation);
+ if (operation === Operation.Register) {
+ socket.write(replyFrame(Operation.Register, registerReplyBody()));
+ return;
+ }
+ socket.write(replyFrame(
+ Operation.NonReplicated,
+ twoNodeMetadataBody(server.port, unavailablePort, 0, 1)
+ ));
+ });
+ const client = new CommandResponseStream(vsrConfig(server.port));
+ compressLeaderlessPoll(client, 0);
+ try {
+ await client.authenticate(vsrConfig(server.port).credentials);
+
+ const operations = server.frames.map(
+ (frame) => frame.readUInt8(REQUEST_OFFSET.operation)
+ );
+ assert.deepEqual(operations, [
+ Operation.Register,
+ Operation.NonReplicated
+ ]);
+ assert.equal(client.isAuthenticated, true);
+ } finally {
+ client.destroy();
+ await server.close();
+ }
+ });
+
it('rejects instead of hanging while a connection attempt is unresolved', async () => {
const server = await startVsrServer(() => {});
const port = server.port;
@@ -672,25 +902,6 @@
}
);
- it('rejects a cluster without a healthy leader', async () => {
- const server = await startVsrServer((frame, socket) => {
- socket.write(replyFrame(
- frame.readUInt8(REQUEST_OFFSET.operation),
- twoNodeMetadataBody(server.port, server.port, 1)
- ));
- });
- const client = new CommandResponseStream(vsrConfig(server.port));
- try {
- await assert.rejects(
- () => client.sendCommand(60_010, Buffer.alloc(0)),
- /VSR cluster has no healthy leader/
- );
- } finally {
- client.destroy();
- await server.close();
- }
- });
-
it('shares token authentication between concurrent callers', async () => {
const server = await startVsrServer(
(frame, socket) => singleNodeHandler(server.port)(frame, socket)
diff --git a/foreign/node/src/client/client.socket.ts b/foreign/node/src/client/client.socket.ts
index 9bbafa4..a3aabcf 100644
--- a/foreign/node/src/client/client.socket.ts
+++ b/foreign/node/src/client/client.socket.ts
@@ -40,6 +40,9 @@
const VSR_RESPONSE_TIMEOUT_MS = 30_000;
const VSR_RETRY_INTERVAL_MS = 50;
+const LEADERLESS_WAIT_BUDGET_MS = 5_000;
+const LEADERLESS_POLL_INTERVAL_MS = 250;
+const MAX_LEADER_REDIRECTS = 3;
const TRANSIENT_NOT_COMMITTED = 57;
const TRANSIENT_NOT_ACCEPTED = 58;
@@ -95,6 +98,12 @@
private vsrSession: VsrSession;
/** Shared authentication attempt for concurrent callers */
private authenticationPromise?: Promise<boolean>;
+ /** Whether a login is already being moved to the leader */
+ private settlingLeader: boolean;
+ /** How long a leaderless roster is polled before settling in place */
+ private leaderlessWaitBudget: number;
+ /** Delay between roster reads while the cluster elects */
+ private leaderlessPollInterval: number;
/** Calls that have acquired this stream but have not fully settled */
private pendingSubmissions: number;
/** Whether the stream is currently processing a command */
@@ -123,6 +132,9 @@
this._execQueue = [];
this.vsrSession = new VsrSession();
this.authenticationPromise = undefined;
+ this.settlingLeader = false;
+ this.leaderlessWaitBudget = LEADERLESS_WAIT_BUDGET_MS;
+ this.leaderlessPollInterval = LEADERLESS_POLL_INTERVAL_MS;
this.pendingSubmissions = 0;
this.heartbeatInFlight = false;
this._init();
@@ -150,7 +162,7 @@
/**
* Sends a command to the server.
- * Automatically handles connection and authentication if needed.
+ * Automatically handles connection, authentication and leader settlement.
*
* @param command - Command code to send
* @param payload - Command payload buffer
@@ -172,26 +184,33 @@
if (!this.connection.connected)
await this.connection.connect()
- if (isLoginCommand(command))
- await this._ensureVsrLeader();
-
if (!this.isAuthenticated && !this.isUnloggedCommand(command))
await this.authenticate(this.options.credentials);
- return await new Promise((resolve, reject) => {
- const job = {
- command,
- payload,
- handleResponse,
- resolve,
- reject
- };
- if (last)
- this._execQueue.push(job);
- else
- this._execQueue.unshift(job);
- this._processQueue();
- });
+ const response = await new Promise<CommandResponse>(
+ (resolve, reject) => {
+ const job = {
+ command,
+ payload,
+ handleResponse,
+ resolve,
+ reject
+ };
+ if (last)
+ this._execQueue.push(job);
+ else
+ this._execQueue.unshift(job);
+ this._processQueue();
+ });
+ if (!isLoginCommand(command) || this.settlingLeader)
+ return response;
+ this.settlingLeader = true;
+ try {
+ const settled = await this._settleOnLeader(command, payload);
+ return settled ?? response;
+ } finally {
+ this.settlingLeader = false;
+ }
} finally {
this.pendingSubmissions -= 1;
this._emitFinishQueue();
@@ -385,38 +404,101 @@
});
}
+ // `GetClusterMetadata` is deliberately absent: the server auth-gates it,
+ // so the client authenticates before reading the topology. A login dialed
+ // at a backup still succeeds because the server forwards the register to
+ // the primary.
private isUnloggedCommand(command: number): boolean {
- return UNLOGGED_COMMAND_CODE.includes(command) ||
- command === COMMAND_CODE.GetClusterMetadata;
+ return UNLOGGED_COMMAND_CODE.includes(command);
}
- private async _ensureVsrLeader(): Promise<void> {
- for (let attempt = 0; attempt < 3; attempt += 1) {
- // Queue the metadata fetch instead of writing directly: a bare write
- // would race an in-flight exchange and both would wake on the same
- // response event.
- const response = await this.sendCommand(
- GET_CLUSTER_METADATA.code,
- GET_CLUSTER_METADATA.serialize(),
+ /**
+ * Moves a freshly authenticated session to the cluster leader.
+ *
+ * Only the leader accepts replicated commands, and the roster read is
+ * auth-gated, so the topology cannot be inspected before a login binds a
+ * session. The redirect drops that session along with the socket, so the
+ * login is replayed on the leader and its answer supersedes the one from the
+ * node the client dialed. Leadership can move between the roster read and
+ * the replay, so each freshly bound hop rechecks the roster under a bounded
+ * redirect budget.
+ *
+ * @returns The leader's login response, or undefined when the client stays
+ */
+ private async _settleOnLeader(
+ loginCommand: number,
+ loginPayload: Buffer
+ ): Promise<CommandResponse | undefined> {
+ let settledResponse: CommandResponse | undefined;
+ for (let redirects = 0; redirects < MAX_LEADER_REDIRECTS; redirects += 1) {
+ const leader = await this._readLeaderEndpoint();
+ if (!leader || this.connection.isConnectedTo(leader.host, leader.port))
+ return settledResponse;
+ await this.connection.redirect(leader.host, leader.port);
+ settledResponse = await this.sendCommand(
+ loginCommand,
+ loginPayload,
{ last: false }
);
- const metadata = GET_CLUSTER_METADATA.deserialize(response);
- if (metadata.nodes.length <= 1)
- return;
- const leader = metadata.nodes.find(
- (node) => node.role === 'Leader' && node.status === 'Healthy'
- );
- if (!leader) {
- await delay(100);
- continue;
- }
- if (!this.connection.isConnectedTo(leader.ip, leader.endpoints.tcp)) {
- await this.connection.redirect(leader.ip, leader.endpoints.tcp);
- continue;
- }
- return;
}
- throw new Error('VSR cluster has no healthy leader');
+ debug(
+ `leader settlement reached its ${MAX_LEADER_REDIRECTS}-hop budget, ` +
+ 'staying on the current node'
+ );
+ return settledResponse;
+ }
+
+ /**
+ * Reads the cluster roster and picks the endpoint to settle on.
+ *
+ * Best effort: an unreadable roster, `Unauthenticated` included (the session
+ * died between the login and this read), keeps the client on its current
+ * node instead of failing a login that already succeeded.
+ */
+ private async _readLeaderEndpoint():
+ Promise<{ host: string, port: number } | undefined> {
+ // A cluster can be transiently leaderless: a restarted node cedes the
+ // primaryship its stale view assigns it, and the roster reports no leader
+ // until the peers' election completes. That window is roughly one heartbeat
+ // timeout, so poll through it rather than settling on a replica that denies
+ // every replicated command for its whole retry budget.
+ const deadline = Date.now() + this.leaderlessWaitBudget;
+ while (true) {
+ // Reading without a session would re-enter authentication, which awaits
+ // the very login this settlement runs inside of. The session can also die
+ // between polls, so this holds for every pass, not just the first.
+ if (!this.isAuthenticated)
+ return undefined;
+ try {
+ // Queue the metadata fetch instead of writing directly: a bare write
+ // would race an in-flight exchange and both would wake on the same
+ // response event.
+ const response = await this.sendCommand(
+ GET_CLUSTER_METADATA.code,
+ GET_CLUSTER_METADATA.serialize(),
+ { last: false }
+ );
+ const metadata = GET_CLUSTER_METADATA.deserialize(response);
+ if (metadata.nodes.length <= 1)
+ return undefined;
+ const leader = metadata.nodes.find(
+ (node) => node.role === 'Leader' && node.status === 'Healthy'
+ );
+ if (leader)
+ return { host: leader.ip, port: leader.endpoints.tcp };
+ } catch (error) {
+ debug('cluster metadata is unreadable, staying on this node', error);
+ return undefined;
+ }
+ if (Date.now() >= deadline) {
+ debug(
+ 'cluster metadata named no healthy leader within ' +
+ `${this.leaderlessWaitBudget} ms, staying on this node`
+ );
+ return undefined;
+ }
+ await delay(this.leaderlessPollInterval);
+ }
}
/**
diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml
index 94ae86a..56f8891 100644
--- a/foreign/python/Cargo.toml
+++ b/foreign/python/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "apache-iggy"
-version = "0.9.0-dev1"
+version = "0.9.0-dev2"
edition = "2024"
authors = ["Iggy Committers <dev@iggy.apache.org>"]
license = "Apache-2.0"
@@ -37,7 +37,7 @@
[dependencies]
bytes = "1.12.1"
futures = "0.3.33"
-iggy = { path = "../../core/sdk", version = "0.11.0-edge.1" }
+iggy = { path = "../../core/sdk", version = "0.11.0-edge.2" }
paste = "1"
pyo3 = "0.29.0"
pyo3-async-runtimes = { version = "0.29.0", features = [
diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml
index ef2e5d5..c3fe391 100644
--- a/foreign/python/pyproject.toml
+++ b/foreign/python/pyproject.toml
@@ -22,7 +22,7 @@
[project]
name = "apache-iggy"
requires-python = ">=3.10"
-version = "0.9.0.dev1"
+version = "0.9.0.dev2"
description = "Apache Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
readme = "README.md"
license = { file = "LICENSE" }
diff --git a/foreign/python/uv.lock b/foreign/python/uv.lock
index 83e1c98..c77af3e 100644
--- a/foreign/python/uv.lock
+++ b/foreign/python/uv.lock
@@ -12,7 +12,7 @@
[[package]]
name = "apache-iggy"
-version = "0.9.0.dev1"
+version = "0.9.0.dev2"
source = { editable = "." }
[package.optional-dependencies]
diff --git a/web/package-lock.json b/web/package-lock.json
index b7f18ae..71d94fa 100644
--- a/web/package-lock.json
+++ b/web/package-lock.json
@@ -1,12 +1,12 @@
{
"name": "iggy-web-ui",
- "version": "0.3.1-edge.1",
+ "version": "0.4.0-edge.1",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "iggy-web-ui",
- "version": "0.3.1-edge.1",
+ "version": "0.4.0-edge.1",
"dependencies": {
"@floating-ui/dom": "^1.8.0",
"@tailwindcss/postcss": "^4.3.3",
diff --git a/web/package.json b/web/package.json
index ff680f4..3b278f3 100644
--- a/web/package.json
+++ b/web/package.json
@@ -1,6 +1,6 @@
{
"name": "iggy-web-ui",
- "version": "0.3.1-edge.1",
+ "version": "0.4.0-edge.1",
"private": true,
"scripts": {
"dev": "vite dev --port 3050",
