You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/05/22 15:34:19 UTC
(iggy) branch integration_tests updated: fix code review nits
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch integration_tests
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/integration_tests by this push:
new ee45beeb8 fix code review nits
ee45beeb8 is described below
commit ee45beeb8c88452347a48a59291f267f9d008d45
Author: Grzegorz Koszyk <g....@gmail.com>
AuthorDate: Fri May 22 17:31:34 2026 +0200
fix code review nits
---
core/binary_protocol/src/consensus/operation.rs | 15 +-
core/binary_protocol/src/dispatch.rs | 1 +
core/common/src/traits/binary_impls/users.rs | 2 +
core/consensus/src/observability.rs | 1 +
core/metadata/src/impls/metadata.rs | 145 +++++++++-
core/metadata/src/stm/user.rs | 152 ++++++++++-
core/sdk/src/quic/quic_client.rs | 111 ++++++--
core/sdk/src/tcp/tcp_client.rs | 91 +++++--
core/sdk/src/vsr.rs | 30 ++-
core/sdk/src/websocket/websocket_client.rs | 36 ++-
core/server-ng/src/bootstrap.rs | 336 ++++++++++++++++++------
core/server-ng/src/server_error.rs | 7 +
core/server-ng/src/session_manager.rs | 12 +
13 files changed, 797 insertions(+), 142 deletions(-)
diff --git a/core/binary_protocol/src/consensus/operation.rs b/core/binary_protocol/src/consensus/operation.rs
index f728f5d7e..2a6f72b3a 100644
--- a/core/binary_protocol/src/consensus/operation.rs
+++ b/core/binary_protocol/src/consensus/operation.rs
@@ -37,6 +37,11 @@ pub enum Operation {
/// legacy command code is stored outside the operation discriminant by the
/// transport/server bridge and must not enter the replicated log.
NonReplicated = 2,
+
+ /// Unregister a client session with the cluster. Carries the legacy
+ /// `LogoutUser` command through the consensus pipeline and removes the
+ /// client-table entry at commit time.
+ Logout = 3,
// Internal metadata operations (journal / replica-only)
CreateTopicWithAssignments = 64,
CreatePartitionsWithAssignments = 65,
@@ -124,7 +129,10 @@ impl Operation {
#[must_use]
#[inline]
pub const fn is_vsr_reserved(&self) -> bool {
- matches!(self, Self::Reserved | Self::Register | Self::NonReplicated)
+ matches!(
+ self,
+ Self::Reserved | Self::Register | Self::Logout | Self::NonReplicated
+ )
}
/// Data-plane operations routed to the shard owning the partition.
@@ -149,6 +157,7 @@ impl Operation {
match self {
Self::Reserved
| Self::Register
+ | Self::Logout
| Self::NonReplicated
| Self::CreateTopicWithAssignments
| Self::CreatePartitionsWithAssignments => None,
@@ -242,6 +251,7 @@ mod tests {
fn vsr_reserved_have_no_code() {
assert_eq!(Operation::Reserved.to_command_code(), None);
assert_eq!(Operation::Register.to_command_code(), None);
+ assert_eq!(Operation::Logout.to_command_code(), None);
assert_eq!(Operation::NonReplicated.to_command_code(), None);
}
@@ -249,11 +259,14 @@ mod tests {
fn vsr_reserved_classification() {
assert!(Operation::Reserved.is_vsr_reserved());
assert!(Operation::Register.is_vsr_reserved());
+ assert!(Operation::Logout.is_vsr_reserved());
assert!(Operation::NonReplicated.is_vsr_reserved());
assert!(!Operation::CreateStream.is_vsr_reserved());
assert!(!Operation::SendMessages.is_vsr_reserved());
assert!(!Operation::Register.is_metadata());
assert!(!Operation::Register.is_partition());
+ assert!(!Operation::Logout.is_metadata());
+ assert!(!Operation::Logout.is_partition());
assert_eq!(
Operation::CreateTopicWithAssignments.to_command_code(),
None
diff --git a/core/binary_protocol/src/dispatch.rs b/core/binary_protocol/src/dispatch.rs
index 9789ab64e..c8d7b307b 100644
--- a/core/binary_protocol/src/dispatch.rs
+++ b/core/binary_protocol/src/dispatch.rs
@@ -288,6 +288,7 @@ pub const fn lookup_by_operation(op: Operation) -> Option<&'static CommandMeta>
| Operation::CreatePartitionsWithAssignments
| Operation::Reserved
| Operation::Register
+ | Operation::Logout
| Operation::NonReplicated => return None,
};
Some(&COMMAND_TABLE[idx])
diff --git a/core/common/src/traits/binary_impls/users.rs b/core/common/src/traits/binary_impls/users.rs
index 81339c4f0..6ccd32016 100644
--- a/core/common/src/traits/binary_impls/users.rs
+++ b/core/common/src/traits/binary_impls/users.rs
@@ -254,6 +254,8 @@ impl<B: BinaryClient> UserClient for B {
fail_if_not_authenticated(self).await?;
self.send_raw_with_response(LOGOUT_USER_CODE, LogoutUserRequest.to_bytes())
.await?;
+ #[cfg(feature = "vsr")]
+ self.reset_vsr_session().await?;
self.set_state(ClientState::Connected).await;
self.publish_event(DiagnosticEvent::SignedOut).await;
Ok(())
diff --git a/core/consensus/src/observability.rs b/core/consensus/src/observability.rs
index a770918b3..375036af0 100644
--- a/core/consensus/src/observability.rs
+++ b/core/consensus/src/observability.rs
@@ -641,6 +641,7 @@ pub const fn operation_as_str(operation: Operation) -> &'static str {
Operation::CreatePersonalAccessToken => "create_personal_access_token",
Operation::DeletePersonalAccessToken => "delete_personal_access_token",
Operation::Register => "register",
+ Operation::Logout => "logout",
Operation::NonReplicated => "non_replicated",
Operation::SendMessages => "send_messages",
Operation::StoreConsumerOffset => "store_consumer_offset",
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 431c83ef8..c232c7fd0 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -667,6 +667,13 @@ where
in_flight,
);
reply
+ } else if prepare_header.operation == Operation::Logout {
+ // Logout unregisters the VSR client session on every replica.
+ let reply = build_reply_message(&prepare_header, &bytes::Bytes::new());
+ self.client_table
+ .borrow_mut()
+ .remove_client(prepare_header.client);
+ reply
} else {
// Normal op: apply SM, commit_reply.
let response = self.mux_stm.update(prepare).unwrap_or_else(|err| {
@@ -748,7 +755,7 @@ where
Command2::Request | Command2::Prepare | Command2::PrepareOk
));
let op = message.header().operation();
- op.is_metadata() || op == Operation::Register
+ op.is_metadata() || matches!(op, Operation::Register | Operation::Logout)
}
}
@@ -890,6 +897,109 @@ where
}
}
+ /// Submit `Logout` from in-process, await commit.
+ ///
+ /// # Returns
+ /// Commit op for the logout. If the client session is already absent, this
+ /// is idempotent and returns the current metadata commit.
+ ///
+ /// # Errors
+ /// Returns a consensus submission error when this node cannot accept the
+ /// logout prepare, the metadata pipeline is saturated, or the pending
+ /// request is canceled before commit.
+ ///
+ /// # Panics
+ /// Panics when called with the reserved client id `0`, on a non-consensus
+ /// metadata shard, or if the prepare gate flips between validation and
+ /// local dispatch.
+ #[allow(clippy::future_not_send)]
+ pub async fn submit_logout_in_process(
+ &self,
+ client_id: u128,
+ session: u64,
+ request: u64,
+ ) -> Result<u64, RegisterSubmitError> {
+ assert!(client_id != 0, "client_id 0 is reserved for internal use");
+ let consensus = self
+ .consensus
+ .as_ref()
+ .expect("submit_logout_in_process: consensus only exists on shard 0");
+
+ if self.client_table.borrow().get_session(client_id).is_none() {
+ return Ok(consensus.commit_min());
+ }
+
+ if !is_caught_up_primary(consensus) {
+ return Err(
+ if consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing() {
+ RegisterSubmitError::NotCaughtUp
+ } else {
+ RegisterSubmitError::NotPrimary
+ },
+ );
+ }
+
+ if consensus
+ .pipeline()
+ .borrow()
+ .has_message_from_client(client_id)
+ {
+ return Err(RegisterSubmitError::InProgress);
+ }
+
+ if consensus.pipeline().borrow().is_full() {
+ return Err(RegisterSubmitError::PipelineFull);
+ }
+
+ let request = build_logout_request_message(consensus, client_id, session, request);
+ debug_assert!(
+ {
+ use iggy_binary_protocol::ConsensusHeader;
+ request.header().validate().is_ok()
+ },
+ "build_logout_request_message produced a header that fails validate()"
+ );
+ let prepare = self
+ .prepare_request(request)
+ .expect("Operation::Logout is client-allowed; prepare projection cannot fail");
+
+ consensus.verify_pipeline();
+ let receiver = consensus.pipeline_message_with_subscriber(PlaneKind::Metadata, &prepare);
+ debug_assert!(
+ is_caught_up_primary(consensus),
+ "submit_logout_in_process: gate flipped between check and dispatch"
+ );
+ self.on_replicate(prepare).await;
+ let mut loopback = Vec::new();
+ consensus.drain_loopback_into(&mut loopback);
+ for message in loopback {
+ match message.header().command {
+ Command2::PrepareOk => match message.try_into_typed::<PrepareOkHeader>() {
+ Ok(prepare_ok) => self.on_ack(prepare_ok).await,
+ Err(error) => warn!(
+ error = %error,
+ "dropping malformed PrepareOk from metadata loopback queue"
+ ),
+ },
+ command => warn!(
+ ?command,
+ "dropping unexpected message from metadata loopback queue"
+ ),
+ }
+ }
+
+ match receiver.await {
+ Ok(reply) => Ok(reply.header().commit),
+ Err(Canceled) => {
+ if self.client_table.borrow().get_session(client_id).is_none() {
+ Ok(consensus.commit_min())
+ } else {
+ Err(RegisterSubmitError::Canceled)
+ }
+ }
+ }
+ }
+
pub fn remove_client_session(&self, client_id: u128) -> bool {
self.client_table.borrow_mut().remove_client(client_id)
}
@@ -1155,6 +1265,8 @@ where
self.client_table
.borrow_mut()
.commit_register(header.client, reply, in_flight);
+ } else if header.operation == Operation::Logout {
+ self.client_table.borrow_mut().remove_client(header.client);
} else {
// Normal op: apply SM, commit_reply.
let response = self.mux_stm.update(prepare).unwrap_or_else(|err| {
@@ -1261,6 +1373,37 @@ where
msg
}
+fn build_logout_request_message<B, P>(
+ consensus: &VsrConsensus<B, P>,
+ client_id: u128,
+ session: u64,
+ request: u64,
+) -> Message<RequestHeader>
+where
+ B: MessageBus,
+ P: Pipeline<Entry = PipelineEntry>,
+{
+ let header_size = size_of::<RequestHeader>();
+ let mut msg = Message::<RequestHeader>::new(header_size);
+ let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
+ &mut msg.as_mut_slice()[..header_size],
+ )
+ .expect("zeroed bytes are a valid RequestHeader");
+ *header = RequestHeader {
+ command: Command2::Request,
+ operation: Operation::Logout,
+ size: u32::try_from(header_size).expect("RequestHeader size fits u32"),
+ cluster: consensus.cluster(),
+ view: consensus.view(),
+ release: 0,
+ client: client_id,
+ session,
+ request,
+ ..RequestHeader::default()
+ };
+ msg
+}
+
fn build_prepare_message<B, P>(
consensus: &VsrConsensus<B, P>,
request: &RequestHeader,
diff --git a/core/metadata/src/stm/user.rs b/core/metadata/src/stm/user.rs
index d9f34763a..d4be3827b 100644
--- a/core/metadata/src/stm/user.rs
+++ b/core/metadata/src/stm/user.rs
@@ -21,10 +21,9 @@ use crate::stm::snapshot::Snapshotable;
use crate::{collect_handlers, define_state, impl_fill_restore};
use ahash::AHashMap;
use bytes::Bytes;
+use bytes::{BufMut, BytesMut};
+use iggy_binary_protocol::codec::{WireDecode, WireEncode, read_u32_le, read_u64_le};
use iggy_binary_protocol::primitives::permissions::{WireGlobalPermissions, WirePermissions};
-use iggy_binary_protocol::requests::personal_access_tokens::{
- CreatePersonalAccessTokenRequest, DeletePersonalAccessTokenRequest,
-};
use iggy_binary_protocol::requests::users::{
ChangePasswordRequest, CreateUserRequest, DeleteUserRequest, UpdatePermissionsRequest,
UpdateUserRequest,
@@ -170,6 +169,70 @@ impl Users {
}
}
+/// Replicated create-PAT command enriched with the authenticated user id.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CreatePersonalAccessTokenRequest {
+ pub user_id: UserId,
+ pub name: WireName,
+ pub expiry: u64,
+}
+
+impl WireEncode for CreatePersonalAccessTokenRequest {
+ fn encoded_size(&self) -> usize {
+ 4 + self.name.encoded_size() + 8
+ }
+
+ fn encode(&self, buf: &mut BytesMut) {
+ buf.put_u32_le(self.user_id);
+ self.name.encode(buf);
+ buf.put_u64_le(self.expiry);
+ }
+}
+
+impl WireDecode for CreatePersonalAccessTokenRequest {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), iggy_binary_protocol::WireError> {
+ let user_id = read_u32_le(buf, 0)?;
+ let (name, mut pos) = WireName::decode(&buf[4..])?;
+ pos += 4;
+ let expiry = read_u64_le(buf, pos)?;
+ pos += 8;
+ Ok((
+ Self {
+ user_id,
+ name,
+ expiry,
+ },
+ pos,
+ ))
+ }
+}
+
+/// Replicated delete-PAT command enriched with the authenticated user id.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DeletePersonalAccessTokenRequest {
+ pub user_id: UserId,
+ pub name: WireName,
+}
+
+impl WireEncode for DeletePersonalAccessTokenRequest {
+ fn encoded_size(&self) -> usize {
+ 4 + self.name.encoded_size()
+ }
+
+ fn encode(&self, buf: &mut BytesMut) {
+ buf.put_u32_le(self.user_id);
+ self.name.encode(buf);
+ }
+}
+
+impl WireDecode for DeletePersonalAccessTokenRequest {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), iggy_binary_protocol::WireError> {
+ let user_id = read_u32_le(buf, 0)?;
+ let (name, consumed) = WireName::decode(&buf[4..])?;
+ Ok((Self { user_id, name }, 4 + consumed))
+ }
+}
+
// TODO(hubcio): Serialize proper reply (e.g. assigned user ID) instead of empty Bytes.
impl StateHandler for CreateUserRequest {
type State = UsersInner;
@@ -299,10 +362,11 @@ impl StateHandler for UpdatePermissionsRequest {
impl StateHandler for CreatePersonalAccessTokenRequest {
type State = UsersInner;
fn apply(&self, state: &mut UsersInner) -> Bytes {
- // TODO: Stub until protocol gets adjusted.
- let user_id = 0;
let expiry = IggyExpiry::from(self.expiry);
- let user_tokens = state.personal_access_tokens.entry(user_id).or_default();
+ let user_tokens = state
+ .personal_access_tokens
+ .entry(self.user_id)
+ .or_default();
let name_arc: Arc<str> = Arc::from(self.name.as_str());
if user_tokens.contains_key(&name_arc) {
return Bytes::new();
@@ -315,13 +379,17 @@ impl StateHandler for CreatePersonalAccessTokenRequest {
return Bytes::new();
}
- let (pat, _) =
- PersonalAccessToken::new(user_id, self.name.as_ref(), IggyTimestamp::now(), expiry);
+ let (pat, _) = PersonalAccessToken::new(
+ self.user_id,
+ self.name.as_ref(),
+ IggyTimestamp::now(),
+ expiry,
+ );
let token_hash = Arc::clone(&pat.token);
user_tokens.insert(name_arc, pat);
state
.personal_access_token_index
- .insert(token_hash, (user_id, Arc::from(self.name.as_str())));
+ .insert(token_hash, (self.user_id, Arc::from(self.name.as_str())));
Bytes::new()
}
}
@@ -329,10 +397,7 @@ impl StateHandler for CreatePersonalAccessTokenRequest {
impl StateHandler for DeletePersonalAccessTokenRequest {
type State = UsersInner;
fn apply(&self, state: &mut UsersInner) -> Bytes {
- // TODO: Stub until protocol gets adjusted.
- let user_id = 0;
-
- if let Some(user_tokens) = state.personal_access_tokens.get_mut(&user_id) {
+ if let Some(user_tokens) = state.personal_access_tokens.get_mut(&self.user_id) {
let name_arc: Arc<str> = Arc::from(self.name.as_str());
if let Some(pat) = user_tokens.remove(&name_arc) {
state.personal_access_token_index.remove(&pat.token);
@@ -565,3 +630,64 @@ impl Snapshotable for Users {
}
impl_fill_restore!(Users, users);
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn create_pat_request_roundtrip_preserves_user_id() {
+ let request = CreatePersonalAccessTokenRequest {
+ user_id: 7,
+ name: WireName::new("api-token").unwrap(),
+ expiry: 3600,
+ };
+
+ let bytes = request.to_bytes();
+ let (decoded, consumed) = CreatePersonalAccessTokenRequest::decode(&bytes).unwrap();
+
+ assert_eq!(consumed, bytes.len());
+ assert_eq!(decoded, request);
+ }
+
+ #[test]
+ fn delete_pat_request_roundtrip_preserves_user_id() {
+ let request = DeletePersonalAccessTokenRequest {
+ user_id: 11,
+ name: WireName::new("staging").unwrap(),
+ };
+
+ let bytes = request.to_bytes();
+ let (decoded, consumed) = DeletePersonalAccessTokenRequest::decode(&bytes).unwrap();
+
+ assert_eq!(consumed, bytes.len());
+ assert_eq!(decoded, request);
+ }
+
+ #[test]
+ fn pat_apply_uses_authenticated_user_id() {
+ let mut users = UsersInner::new();
+ let create = CreatePersonalAccessTokenRequest {
+ user_id: 5,
+ name: WireName::new("deploy").unwrap(),
+ expiry: 0,
+ };
+ create.apply(&mut users);
+
+ assert!(users.personal_access_tokens.contains_key(&5));
+ assert!(!users.personal_access_tokens.contains_key(&0));
+ let token = users.personal_access_tokens[&5]
+ .get("deploy")
+ .expect("token should be stored under the authenticated user");
+ assert_eq!(token.user_id, 5);
+
+ let delete = DeletePersonalAccessTokenRequest {
+ user_id: 5,
+ name: WireName::new("deploy").unwrap(),
+ };
+ delete.apply(&mut users);
+
+ assert!(users.personal_access_tokens[&5].is_empty());
+ assert!(users.personal_access_token_index.is_empty());
+ }
+}
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index 1f2eb1080..c089173f8 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -27,6 +27,8 @@ use crate::quic::skip_server_verification::SkipServerVerification;
use async_broadcast::{Receiver, Sender, broadcast};
use async_trait::async_trait;
use bytes::Bytes;
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::codes::{LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE};
use iggy_common::{
ClientState, ConnectionString, ConnectionStringUtils, Credentials, DiagnosticEvent,
QuicConnectionStringOptions, TransportProtocol, validate_server_address,
@@ -62,6 +64,8 @@ pub struct QuicClient {
pub(crate) current_server_address: Mutex<String>,
#[cfg(feature = "vsr")]
consensus_session: Arc<Mutex<ConsensusSession>>,
+ #[cfg(feature = "vsr")]
+ skip_auto_login_once: Mutex<bool>,
}
unsafe impl Send for QuicClient {}
@@ -138,12 +142,23 @@ impl BinaryTransport for QuicClient {
}
self.disconnect().await?;
+ #[cfg(feature = "vsr")]
+ let skip_auto_login = is_login_register_code(code);
+ #[cfg(feature = "vsr")]
+ if skip_auto_login {
+ *self.skip_auto_login_once.lock().await = true;
+ }
let server_address = self.current_server_address.lock().await.to_string();
info!(
"Reconnecting to the server: {}, by client: {}",
server_address, self.config.client_address
);
- self.connect().await?;
+ let reconnect = self.connect().await;
+ #[cfg(feature = "vsr")]
+ if skip_auto_login && reconnect.is_err() {
+ *self.skip_auto_login_once.lock().await = false;
+ }
+ reconnect?;
self.send_raw(code, payload).await
}
@@ -244,6 +259,8 @@ impl QuicClient {
current_server_address: Mutex::new(server_address),
#[cfg(feature = "vsr")]
consensus_session: Arc::new(Mutex::new(ConsensusSession::new())),
+ #[cfg(feature = "vsr")]
+ skip_auto_login_once: Mutex::new(false),
})
}
@@ -424,6 +441,12 @@ impl QuicClient {
self.connected_at.lock().await.replace(now);
self.publish_event(DiagnosticEvent::Connected).await;
+ #[cfg(feature = "vsr")]
+ let skip_auto_login = {
+ let mut guard = self.skip_auto_login_once.lock().await;
+ std::mem::take(&mut *guard)
+ };
+
// Handle auto-login
let should_redirect = match &self.config.auto_login {
AutoLogin::Disabled => {
@@ -431,32 +454,67 @@ impl QuicClient {
false
}
AutoLogin::Enabled(credentials) => {
- info!(
- "{NAME} client: {} is signing in...",
- self.config.client_address
- );
- self.set_state(ClientState::Authenticating).await;
- match credentials {
- Credentials::UsernamePassword(username, password) => {
- self.login_user(username, password.expose_secret()).await?;
- self.publish_event(DiagnosticEvent::SignedIn).await;
- info!(
- "{NAME} client: {} has signed in with the user credentials, username: {username}",
- self.config.client_address
- );
- }
- Credentials::PersonalAccessToken(token) => {
- self.login_with_personal_access_token(token.expose_secret())
- .await?;
- self.publish_event(DiagnosticEvent::SignedIn).await;
- info!(
- "{NAME} client: {} has signed in with a personal access token.",
- self.config.client_address
- );
+ #[cfg(feature = "vsr")]
+ if skip_auto_login {
+ info!("Skipping automatic sign-in for a retried login/register request.");
+ false
+ } else {
+ info!(
+ "{NAME} client: {} is signing in...",
+ self.config.client_address
+ );
+ self.set_state(ClientState::Authenticating).await;
+ match credentials {
+ Credentials::UsernamePassword(username, password) => {
+ self.login_user(username, password.expose_secret()).await?;
+ self.publish_event(DiagnosticEvent::SignedIn).await;
+ info!(
+ "{NAME} client: {} has signed in with the user credentials, username: {username}",
+ self.config.client_address
+ );
+ }
+ Credentials::PersonalAccessToken(token) => {
+ self.login_with_personal_access_token(token.expose_secret())
+ .await?;
+ self.publish_event(DiagnosticEvent::SignedIn).await;
+ info!(
+ "{NAME} client: {} has signed in with a personal access token.",
+ self.config.client_address
+ );
+ }
}
+
+ self.handle_leader_redirection().await?
}
+ #[cfg(not(feature = "vsr"))]
+ {
+ info!(
+ "{NAME} client: {} is signing in...",
+ self.config.client_address
+ );
+ self.set_state(ClientState::Authenticating).await;
+ match credentials {
+ Credentials::UsernamePassword(username, password) => {
+ self.login_user(username, password.expose_secret()).await?;
+ self.publish_event(DiagnosticEvent::SignedIn).await;
+ info!(
+ "{NAME} client: {} has signed in with the user credentials, username: {username}",
+ self.config.client_address
+ );
+ }
+ Credentials::PersonalAccessToken(token) => {
+ self.login_with_personal_access_token(token.expose_secret())
+ .await?;
+ self.publish_event(DiagnosticEvent::SignedIn).await;
+ info!(
+ "{NAME} client: {} has signed in with a personal access token.",
+ self.config.client_address
+ );
+ }
+ }
- self.handle_leader_redirection().await?
+ self.handle_leader_redirection().await?
+ }
}
};
@@ -652,6 +710,11 @@ impl QuicClient {
}
}
+#[cfg(feature = "vsr")]
+const fn is_login_register_code(code: u32) -> bool {
+ matches!(code, LOGIN_REGISTER_CODE | LOGIN_REGISTER_WITH_PAT_CODE)
+}
+
fn configure(config: &QuicClientConfig) -> Result<ClientConfig, IggyError> {
let max_concurrent_bidi_streams = VarInt::try_from(config.max_concurrent_bidi_streams);
if max_concurrent_bidi_streams.is_err() {
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index c424b4451..060aa54e8 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -27,6 +27,8 @@ use crate::tcp::tcp_tls_connection_stream::TcpTlsConnectionStream;
use async_broadcast::{Receiver, Sender, broadcast};
use async_trait::async_trait;
use bytes::{BufMut, Bytes, BytesMut};
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::codes::{LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE};
#[cfg(not(feature = "vsr"))]
use iggy_common::IggyErrorDiscriminants;
use iggy_common::{
@@ -65,6 +67,8 @@ pub struct TcpClient {
pub(crate) current_server_address: Mutex<String>,
#[cfg(feature = "vsr")]
consensus_session: Arc<Mutex<ConsensusSession>>,
+ #[cfg(feature = "vsr")]
+ skip_auto_login_once: Mutex<bool>,
}
impl Default for TcpClient {
@@ -140,6 +144,13 @@ impl BinaryTransport for TcpClient {
self.disconnect().await?;
+ #[cfg(feature = "vsr")]
+ let skip_auto_login = is_login_register_code(code);
+ #[cfg(feature = "vsr")]
+ if skip_auto_login {
+ *self.skip_auto_login_once.lock().await = true;
+ }
+
{
let client_address = self.get_client_address_value().await;
let server_address = self.current_server_address.lock().await.clone();
@@ -149,7 +160,12 @@ impl BinaryTransport for TcpClient {
);
}
- self.connect().await?;
+ let reconnect = self.connect().await;
+ #[cfg(feature = "vsr")]
+ if skip_auto_login && reconnect.is_err() {
+ *self.skip_auto_login_once.lock().await = false;
+ }
+ reconnect?;
self.send_raw(code, payload).await
}
@@ -243,6 +259,8 @@ impl TcpClient {
current_server_address: Mutex::new(server_address),
#[cfg(feature = "vsr")]
consensus_session: Arc::new(Mutex::new(ConsensusSession::new())),
+ #[cfg(feature = "vsr")]
+ skip_auto_login_once: Mutex::new(false),
})
}
@@ -468,6 +486,12 @@ impl TcpClient {
self.set_state(ClientState::Connected).await;
self.connected_at.lock().await.replace(now);
self.publish_event(DiagnosticEvent::Connected).await;
+ #[cfg(feature = "vsr")]
+ let skip_auto_login = {
+ let mut guard = self.skip_auto_login_once.lock().await;
+ std::mem::take(&mut *guard)
+ };
+
// Handle auto-login
let should_redirect = match &self.config.auto_login {
AutoLogin::Disabled => {
@@ -475,25 +499,53 @@ impl TcpClient {
false
}
AutoLogin::Enabled(credentials) => {
- info!("{NAME} client: {client_address} is signing in...");
- self.set_state(ClientState::Authenticating).await;
- match credentials {
- Credentials::UsernamePassword(username, password) => {
- self.login_user(username, password.expose_secret()).await?;
- info!(
- "{NAME} client: {client_address} has signed in with the user credentials, username: {username}",
- );
- }
- Credentials::PersonalAccessToken(token) => {
- self.login_with_personal_access_token(token.expose_secret())
- .await?;
- info!(
- "{NAME} client: {client_address} has signed in with a personal access token.",
- );
+ #[cfg(feature = "vsr")]
+ if skip_auto_login {
+ info!("Skipping automatic sign-in for a retried login/register request.");
+ false
+ } else {
+ info!("{NAME} client: {client_address} is signing in...");
+ self.set_state(ClientState::Authenticating).await;
+ match credentials {
+ Credentials::UsernamePassword(username, password) => {
+ self.login_user(username, password.expose_secret()).await?;
+ info!(
+ "{NAME} client: {client_address} has signed in with the user credentials, username: {username}",
+ );
+ }
+ Credentials::PersonalAccessToken(token) => {
+ self.login_with_personal_access_token(token.expose_secret())
+ .await?;
+ info!(
+ "{NAME} client: {client_address} has signed in with a personal access token.",
+ );
+ }
}
+
+ self.handle_leader_redirection().await?
}
+ #[cfg(not(feature = "vsr"))]
+ {
+ info!("{NAME} client: {client_address} is signing in...");
+ self.set_state(ClientState::Authenticating).await;
+ match credentials {
+ Credentials::UsernamePassword(username, password) => {
+ self.login_user(username, password.expose_secret()).await?;
+ info!(
+ "{NAME} client: {client_address} has signed in with the user credentials, username: {username}",
+ );
+ }
+ Credentials::PersonalAccessToken(token) => {
+ self.login_with_personal_access_token(token.expose_secret())
+ .await?;
+ info!(
+ "{NAME} client: {client_address} has signed in with a personal access token.",
+ );
+ }
+ }
- self.handle_leader_redirection().await?
+ self.handle_leader_redirection().await?
+ }
}
};
@@ -717,6 +769,11 @@ impl TcpClient {
}
}
+#[cfg(feature = "vsr")]
+const fn is_login_register_code(code: u32) -> bool {
+ matches!(code, LOGIN_REGISTER_CODE | LOGIN_REGISTER_WITH_PAT_CODE)
+}
+
/// Unit tests for TcpClient.
/// Currently only tests for "from_connection_string()" are implemented.
/// TODO: Add complete unit tests for TcpClient.
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index 0bbc27ceb..4089b69ee 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -21,7 +21,7 @@ use bytes::{BufMut, Bytes, BytesMut};
use iggy_binary_protocol::codec::WireDecode;
use iggy_binary_protocol::codes::{
DELETE_CONSUMER_OFFSET_2_CODE, DELETE_CONSUMER_OFFSET_CODE, DELETE_SEGMENTS_CODE,
- LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE, SEND_MESSAGES_CODE,
+ LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE, LOGOUT_USER_CODE, SEND_MESSAGES_CODE,
STORE_CONSUMER_OFFSET_2_CODE, STORE_CONSUMER_OFFSET_CODE,
};
use iggy_binary_protocol::consensus::{
@@ -39,6 +39,16 @@ use iggy_common::{IggyError, IggyTimestamp};
const NON_REPLICATED_CODE_RANGE: std::ops::Range<usize> = 0..4;
+// TODO(vsr): transparent retry-after-disconnect can weaken at-most-once semantics
+// for replicated writes. The transports currently disconnect, reconnect, and encode
+// the retry from the current ConsensusSession. If disconnect created a fresh VSR
+// client/session, the retried request gets a new (client_id, request_id) tuple, so
+// server-side deduplication cannot match a mutation that may already have committed
+// before the transport failure. TigerBeetle avoids session resume and relies on
+// idempotency for requests retried from a new client session. For Iggy, either stop
+// transparent retries for replicated mutations after VSR session reset, add explicit
+// session resume/rebind semantics, or add a protocol-level idempotency key that is
+// independent of (client_id, request_id).
pub(crate) fn encode_contiguous_request(
session: &mut ConsensusSession,
code: u32,
@@ -92,6 +102,10 @@ pub(crate) fn encode_request_header(
}
fn operation_for_code(code: u32) -> Result<Operation, IggyError> {
+ if code == LOGOUT_USER_CODE {
+ return Ok(Operation::Logout);
+ }
+
if let Some(operation) = Operation::from_command_code(code) {
return Ok(operation);
}
@@ -138,6 +152,7 @@ fn namespace_for_request(
// Control-plane requests do not target a concrete stream/topic/partition shard.
// They are encoded with namespace 0 and routed through metadata/shard 0.
if operation == Operation::Register
+ || operation == Operation::Logout
|| operation == Operation::NonReplicated
|| operation.is_metadata()
{
@@ -316,7 +331,7 @@ mod tests {
}
#[test]
- fn logout_uses_non_replicated_operation() {
+ fn logout_uses_replicated_logout_operation() {
let mut session = ConsensusSession::with_client_id(42);
session.bind(99);
let bytes =
@@ -324,15 +339,8 @@ mod tests {
let request = decode_request(&bytes);
let header = request.header();
- assert_eq!(header.operation, Operation::NonReplicated);
- assert_eq!(
- u32::from_le_bytes(
- header.reserved[NON_REPLICATED_CODE_RANGE]
- .try_into()
- .unwrap()
- ),
- LOGOUT_USER_CODE
- );
+ assert_eq!(header.operation, Operation::Logout);
+ assert_eq!(header.request, 1);
assert_eq!(header.session, 99);
assert_eq!(header.namespace, 0);
}
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
index b902f02b4..574016afe 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -28,6 +28,8 @@ use crate::prelude::Client;
use async_broadcast::{Receiver, Sender, broadcast};
use async_trait::async_trait;
use bytes::{BufMut, Bytes, BytesMut};
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::codes::{LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE};
#[cfg(not(feature = "vsr"))]
use iggy_common::IggyErrorDiscriminants;
use iggy_common::{
@@ -65,6 +67,8 @@ pub struct WebSocketClient {
pub(crate) current_server_address: Mutex<String>,
#[cfg(feature = "vsr")]
consensus_session: Arc<Mutex<ConsensusSession>>,
+ #[cfg(feature = "vsr")]
+ skip_auto_login_once: Mutex<bool>,
}
impl Default for WebSocketClient {
@@ -142,6 +146,13 @@ impl BinaryTransport for WebSocketClient {
self.disconnect().await?;
+ #[cfg(feature = "vsr")]
+ let skip_auto_login = is_login_register_code(code);
+ #[cfg(feature = "vsr")]
+ if skip_auto_login {
+ *self.skip_auto_login_once.lock().await = true;
+ }
+
{
let client_address = self.get_client_address_value().await;
info!(
@@ -150,7 +161,12 @@ impl BinaryTransport for WebSocketClient {
);
}
- self.connect().await?;
+ let reconnect = self.connect().await;
+ #[cfg(feature = "vsr")]
+ if skip_auto_login && reconnect.is_err() {
+ *self.skip_auto_login_once.lock().await = false;
+ }
+ reconnect?;
self.send_raw(code, payload).await
}
@@ -203,6 +219,8 @@ impl WebSocketClient {
current_server_address: Mutex::new(server_address),
#[cfg(feature = "vsr")]
consensus_session: Arc::new(Mutex::new(ConsensusSession::new())),
+ #[cfg(feature = "vsr")]
+ skip_auto_login_once: Mutex::new(false),
})
}
@@ -521,12 +539,23 @@ impl WebSocketClient {
async fn auto_login(&self) -> Result<(), IggyError> {
let client_address = self.get_client_address_value().await;
+ #[cfg(feature = "vsr")]
+ let skip_auto_login = {
+ let mut guard = self.skip_auto_login_once.lock().await;
+ std::mem::take(&mut *guard)
+ };
+
match &self.config.auto_login {
AutoLogin::Disabled => {
info!("{NAME} client: {client_address} - automatic sign-in is disabled.");
Ok(())
}
AutoLogin::Enabled(credentials) => {
+ #[cfg(feature = "vsr")]
+ if skip_auto_login {
+ info!("Skipping automatic sign-in for a retried login/register request.");
+ return Ok(());
+ }
info!("{NAME} client: {client_address} is signing in...");
self.set_state(ClientState::Authenticating).await;
match credentials {
@@ -725,6 +754,11 @@ impl WebSocketClient {
}
}
+#[cfg(feature = "vsr")]
+const fn is_login_register_code(code: u32) -> bool {
+ matches!(code, LOGIN_REGISTER_CODE | LOGIN_REGISTER_WITH_PAT_CODE)
+}
+
#[cfg(test)]
mod tests {
use super::*;
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index e24fbc6f6..616160e0e 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -21,18 +21,22 @@ use crate::config_writer::write_current_config;
use crate::login_register::LoginRegisterError;
use crate::server_error::ServerNgError;
use crate::session_manager::SessionManager;
-use bytes::{BufMut, Bytes, BytesMut};
+use bytes::Bytes;
use configs::server_ng::ServerNgConfig;
use consensus::{LocalPipeline, MetadataHandle, PartitionsHandle, Sequencer, VsrConsensus};
use iggy_binary_protocol::codes::{
GET_CLUSTER_METADATA_CODE, GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE,
- GET_TOPICS_CODE, LOGOUT_USER_CODE, PING_CODE, POLL_MESSAGES_CODE,
+ GET_TOPICS_CODE, PING_CODE, POLL_MESSAGES_CODE,
};
use iggy_binary_protocol::requests::consumer_offsets::{
DeleteConsumerOffset2Request, DeleteConsumerOffsetRequest, StoreConsumerOffset2Request,
StoreConsumerOffsetRequest,
};
use iggy_binary_protocol::requests::messages::{PollMessagesRequest, SendMessagesHeader};
+use iggy_binary_protocol::requests::personal_access_tokens::{
+ CreatePersonalAccessTokenRequest as WireCreatePersonalAccessTokenRequest,
+ DeletePersonalAccessTokenRequest as WireDeletePersonalAccessTokenRequest,
+};
use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest;
use iggy_binary_protocol::requests::streams::{GetStreamRequest, GetStreamsRequest};
use iggy_binary_protocol::requests::topics::{GetTopicRequest, GetTopicsRequest};
@@ -48,7 +52,6 @@ use iggy_binary_protocol::responses::system::get_cluster_metadata::{
use iggy_binary_protocol::responses::system::get_stats::StatsResponse;
use iggy_binary_protocol::responses::topics::get_topic::{GetTopicResponse, PartitionResponse};
use iggy_binary_protocol::responses::topics::get_topics::GetTopicsResponse;
-use iggy_binary_protocol::responses::users::LoginRegisterResponse;
use iggy_binary_protocol::{
Command2, GenericHeader, Message, Operation, ReplyHeader, RequestHeader, WireDecode,
WireEncode, WireIdentifier, WireName, WirePartitioning,
@@ -83,6 +86,10 @@ use metadata::stm::consumer_group::ConsumerGroups;
use metadata::stm::snapshot::Snapshot;
use metadata::stm::stream::{Partition, Streams};
use metadata::stm::user::Users;
+use metadata::stm::user::{
+ CreatePersonalAccessTokenRequest as ReplicatedCreatePersonalAccessTokenRequest,
+ DeletePersonalAccessTokenRequest as ReplicatedDeletePersonalAccessTokenRequest,
+};
use partitions::{
IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, PartitionsConfig, Segment,
};
@@ -94,6 +101,7 @@ use server::streaming::partitions::storage::{load_consumer_group_offsets, load_c
use server::streaming::segments::storage::create_segment_storage;
use server::streaming::users::user::User as LegacyUser;
use server::streaming::utils::crypto;
+use server::{IGGY_ROOT_PASSWORD_ENV, IGGY_ROOT_USERNAME_ENV};
use shard::builder::IggyShardBuilder;
use shard::shards_table::PapayaShardsTable;
use shard::{
@@ -101,6 +109,7 @@ use shard::{
};
use std::cell::{Cell, RefCell};
use std::collections::{HashMap, HashSet, VecDeque};
+use std::env;
use std::future::Future;
use std::net::{IpAddr, SocketAddr};
use std::path::{Path, PathBuf};
@@ -255,6 +264,7 @@ pub async fn bootstrap(
let recovered = recover::<ServerNgMuxStateMachine>(Path::new(&config.system.path))
.await
.map_err(ServerNgError::MetadataRecovery)?;
+ validate_cluster_root_bootstrap(config, &recovered.mux_stm)?;
ensure_default_root_user(&recovered.mux_stm);
let restored_op = recovered.last_applied_op.unwrap_or_else(|| {
recovered
@@ -1310,8 +1320,7 @@ fn make_client_request_handler(shard: &Rc<ServerNgShard>) -> RequestHandler {
{
shard_for_disconnect
.plane
- .inner()
- .0
+ .metadata()
.remove_client_session(vsr_client_id);
}
}));
@@ -1461,7 +1470,7 @@ async fn handle_client_request(
let header = *request.header();
if header.operation == Operation::NonReplicated {
- handle_non_replicated_request(shard, sessions, transport_client_id, request).await;
+ handle_non_replicated_request(shard, transport_client_id, request).await;
return;
}
@@ -1470,6 +1479,11 @@ async fn handle_client_request(
return;
}
+ if header.operation == Operation::Logout {
+ handle_logout_request(shard, sessions, transport_client_id, request).await;
+ return;
+ }
+
let bound = sessions.borrow().get_session(transport_client_id);
let resolved_namespace = if header.operation.is_partition() {
match resolve_partition_request_namespace(shard, header.operation, request_body(&request)) {
@@ -1498,13 +1512,24 @@ async fn handle_client_request(
new_header.session = bound_session;
}
});
+ let request = match maybe_rewrite_pat_request(sessions, transport_client_id, request) {
+ Ok(request) => request,
+ Err(error) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ operation = ?header.operation,
+ "dropping request with invalid PAT replication context"
+ );
+ return;
+ }
+ };
shard.dispatch(request.into_generic());
}
#[allow(clippy::future_not_send)]
async fn handle_non_replicated_request(
shard: &Rc<ServerNgShard>,
- sessions: &Rc<RefCell<SessionManager>>,
transport_client_id: u128,
request: Message<RequestHeader>,
) {
@@ -1513,12 +1538,11 @@ async fn handle_non_replicated_request(
match code {
PING_CODE => {
let commit = current_metadata_commit(shard);
- let reply = build_login_register_reply(
+ let reply = build_empty_reply(
request.header(),
request.header().client,
request.header().session,
commit,
- &Bytes::new(),
);
if let Err(error) = shard
.bus
@@ -1532,41 +1556,14 @@ async fn handle_non_replicated_request(
);
}
}
- LOGOUT_USER_CODE => {
- if let Some(vsr_client_id) =
- sessions.borrow_mut().remove_connection(transport_client_id)
- {
- shard.plane.metadata().remove_client_session(vsr_client_id);
- }
- let commit = current_metadata_commit(shard);
- let reply = build_login_register_reply(
- request.header(),
- request.header().client,
- request.header().session,
- commit,
- &Bytes::new(),
- );
- if let Err(error) = shard
- .bus
- .send_to_client(transport_client_id, reply.into_generic().into_frozen())
- .await
- {
- warn!(
- transport_client_id,
- error = %error,
- "failed to send non-replicated logout reply"
- );
- }
- }
_ => match build_non_replicated_response(shard, code, request_body(&request)) {
- Ok(body) => {
+ Ok(response) => {
let commit = current_metadata_commit(shard);
- let reply = build_login_register_reply(
+ let reply = response.into_reply(
request.header(),
request.header().client,
request.header().session,
commit,
- &body,
);
if let Err(error) = shard
.bus
@@ -1593,6 +1590,51 @@ async fn handle_non_replicated_request(
}
}
+#[allow(clippy::future_not_send)]
+async fn handle_logout_request(
+ shard: &Rc<ServerNgShard>,
+ sessions: &Rc<RefCell<SessionManager>>,
+ transport_client_id: u128,
+ request: Message<RequestHeader>,
+) {
+ let Some((vsr_client_id, session)) = sessions.borrow().get_session(transport_client_id) else {
+ warn!(
+ transport_client_id,
+ "dropping logout for unbound VSR session"
+ );
+ return;
+ };
+
+ let request_id = request.header().request;
+ let commit = match shard
+ .plane
+ .metadata()
+ .submit_logout_in_process(vsr_client_id, session, request_id)
+ .await
+ {
+ Ok(commit) => commit,
+ Err(error) => {
+ warn!(transport_client_id, error = %error, "logout/unregister failed");
+ return;
+ }
+ };
+
+ sessions.borrow_mut().remove_connection(transport_client_id);
+
+ let reply = build_empty_reply(request.header(), vsr_client_id, session, commit);
+ if let Err(error) = shard
+ .bus
+ .send_to_client(transport_client_id, reply.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to send logout reply"
+ );
+ }
+}
+
fn ensure_transport_connection(
shard: &Rc<ServerNgShard>,
sessions: &Rc<RefCell<SessionManager>>,
@@ -1606,6 +1648,66 @@ fn ensure_transport_connection(
.ensure_connection(transport_client_id, meta.peer_addr);
}
+fn maybe_rewrite_pat_request(
+ sessions: &Rc<RefCell<SessionManager>>,
+ transport_client_id: u128,
+ request: Message<RequestHeader>,
+) -> Result<Message<RequestHeader>, IggyError> {
+ let operation = request.header().operation;
+ let user_id = match operation {
+ Operation::CreatePersonalAccessToken | Operation::DeletePersonalAccessToken => sessions
+ .borrow()
+ .get_user_id(transport_client_id)
+ .ok_or(IggyError::Unauthenticated)?,
+ _ => return Ok(request),
+ };
+
+ let body = request_body(&request);
+ let rewritten = match operation {
+ Operation::CreatePersonalAccessToken => {
+ let wire = WireCreatePersonalAccessTokenRequest::decode_from(body)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ ReplicatedCreatePersonalAccessTokenRequest {
+ user_id,
+ name: wire.name,
+ expiry: wire.expiry,
+ }
+ .to_bytes()
+ }
+ Operation::DeletePersonalAccessToken => {
+ let wire = WireDeletePersonalAccessTokenRequest::decode_from(body)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ ReplicatedDeletePersonalAccessTokenRequest {
+ user_id,
+ name: wire.name,
+ }
+ .to_bytes()
+ }
+ _ => unreachable!(),
+ };
+
+ rewrite_request_body(&request, &rewritten)
+}
+
+fn rewrite_request_body(
+ request: &Message<RequestHeader>,
+ body: &Bytes,
+) -> Result<Message<RequestHeader>, IggyError> {
+ let total_size = std::mem::size_of::<RequestHeader>()
+ .checked_add(body.len())
+ .ok_or(IggyError::InvalidConfiguration)?;
+ let size = u32::try_from(total_size).map_err(|_| IggyError::InvalidConfiguration)?;
+ let mut rewritten = Message::<RequestHeader>::new(total_size);
+ let header = bytemuck::checked::try_from_bytes_mut::<RequestHeader>(
+ &mut rewritten.as_mut_slice()[..std::mem::size_of::<RequestHeader>()],
+ )
+ .expect("zeroed bytes are a valid request header");
+ *header = *request.header();
+ header.size = size;
+ rewritten.as_mut_slice()[std::mem::size_of::<RequestHeader>()..].copy_from_slice(body);
+ Ok(rewritten)
+}
+
#[allow(clippy::future_not_send)]
async fn handle_login_register_request(
shard: &Rc<ServerNgShard>,
@@ -1799,40 +1901,54 @@ fn build_non_replicated_response(
shard: &Rc<ServerNgShard>,
code: u32,
body: &[u8],
-) -> Result<Bytes, IggyError> {
+) -> Result<NonReplicatedResponse, IggyError> {
match code {
- GET_CLUSTER_METADATA_CODE => Ok(build_cluster_metadata_response().to_bytes()),
- GET_STATS_CODE => Ok(build_stats_response(shard)?.to_bytes()),
+ GET_CLUSTER_METADATA_CODE => Ok(NonReplicatedResponse::Bytes(
+ build_cluster_metadata_response().to_bytes(),
+ )),
+ GET_STATS_CODE => Ok(NonReplicatedResponse::Bytes(
+ build_stats_response(shard)?.to_bytes(),
+ )),
GET_STREAM_CODE => {
let request =
GetStreamRequest::decode_from(body).map_err(|_| IggyError::InvalidCommand)?;
- build_get_stream_response(shard, &request.stream_id)
- .map(|response| response.map_or_else(Bytes::new, |response| response.to_bytes()))
+ build_get_stream_response(shard, &request.stream_id).map(|response| {
+ response.map_or(NonReplicatedResponse::Empty, |response| {
+ NonReplicatedResponse::Bytes(response.to_bytes())
+ })
+ })
}
GET_STREAMS_CODE => {
let _ = GetStreamsRequest::decode_from(body).map_err(|_| IggyError::InvalidCommand)?;
- Ok(build_get_streams_response(shard)?.to_bytes())
+ Ok(NonReplicatedResponse::Bytes(
+ build_get_streams_response(shard)?.to_bytes(),
+ ))
}
GET_TOPIC_CODE => {
let request =
GetTopicRequest::decode_from(body).map_err(|_| IggyError::InvalidCommand)?;
- build_get_topic_response(shard, &request.stream_id, &request.topic_id)
- .map(|response| response.map_or_else(Bytes::new, |response| response.to_bytes()))
+ build_get_topic_response(shard, &request.stream_id, &request.topic_id).map(|response| {
+ response.map_or(NonReplicatedResponse::Empty, |response| {
+ NonReplicatedResponse::Bytes(response.to_bytes())
+ })
+ })
}
GET_TOPICS_CODE => {
let request =
GetTopicsRequest::decode_from(body).map_err(|_| IggyError::InvalidCommand)?;
- Ok(build_get_topics_response(shard, &request.stream_id)?.to_bytes())
+ Ok(NonReplicatedResponse::Bytes(
+ build_get_topics_response(shard, &request.stream_id)?.to_bytes(),
+ ))
}
POLL_MESSAGES_CODE => {
let request =
PollMessagesRequest::decode_from(body).map_err(|_| IggyError::InvalidCommand)?;
- Ok(empty_polled_messages_response(
- request.partition_id.unwrap_or(0),
- ))
+ Ok(NonReplicatedResponse::EmptyPolledMessages {
+ partition_id: request.partition_id.unwrap_or(0),
+ })
}
_ => match iggy_binary_protocol::dispatch::lookup_command(code) {
- Some(meta) if !meta.is_replicated() => Ok(Bytes::new()),
+ Some(meta) if !meta.is_replicated() => Ok(NonReplicatedResponse::Empty),
Some(_) => Err(IggyError::FeatureUnavailable),
None => Err(IggyError::InvalidCommand),
},
@@ -2078,14 +2194,6 @@ fn partition_response(
})
}
-fn empty_polled_messages_response(partition_id: u32) -> Bytes {
- let mut response = BytesMut::with_capacity(16);
- response.put_u32_le(partition_id);
- response.put_u64_le(0);
- response.put_u32_le(0);
- response.freeze()
-}
-
fn usize_to_u32(value: usize) -> Result<u32, IggyError> {
u32::try_from(value).map_err(|_| IggyError::InvalidIdentifier)
}
@@ -2160,10 +2268,9 @@ async fn complete_login_register(
.map(|(_, session)| session)
};
if let Some(session) = existing_session {
- let response = LoginRegisterResponse { user_id, session }.to_bytes();
let commit = current_metadata_commit(shard);
let reply =
- build_login_register_reply(request_header, vsr_client_id, session, commit, &response);
+ build_login_register_reply(request_header, vsr_client_id, session, commit, user_id);
let _ = shard
.bus
.send_to_client(transport_client_id, reply.into_generic().into_frozen())
@@ -2180,8 +2287,7 @@ async fn complete_login_register(
let session = match shard
.plane
- .inner()
- .0
+ .metadata()
.submit_register_in_process(vsr_client_id)
.await
{
@@ -2202,10 +2308,8 @@ async fn complete_login_register(
}
}
- let response = LoginRegisterResponse { user_id, session }.to_bytes();
let commit = current_metadata_commit(shard);
- let reply =
- build_login_register_reply(request_header, vsr_client_id, session, commit, &response);
+ let reply = build_login_register_reply(request_header, vsr_client_id, session, commit, user_id);
if let Err(error) = shard
.bus
.send_to_client(transport_client_id, reply.into_generic().into_frozen())
@@ -2232,18 +2336,107 @@ fn ensure_default_root_user(mux_stm: &ServerNgMuxStateMachine) {
mux_stm.users().ensure_root_user(&username, &password);
}
+fn validate_cluster_root_bootstrap(
+ config: &ServerNgConfig,
+ mux_stm: &ServerNgMuxStateMachine,
+) -> Result<(), ServerNgError> {
+ if !config.cluster.enabled || !mux_stm.users().read(|users| users.items.is_empty()) {
+ return Ok(());
+ }
+
+ if env::var(IGGY_ROOT_USERNAME_ENV).is_ok() && env::var(IGGY_ROOT_PASSWORD_ENV).is_ok() {
+ return Ok(());
+ }
+
+ Err(ServerNgError::ClusterRootCredentialsRequired {
+ username_env: IGGY_ROOT_USERNAME_ENV,
+ password_env: IGGY_ROOT_PASSWORD_ENV,
+ })
+}
+
+enum NonReplicatedResponse {
+ Empty,
+ EmptyPolledMessages { partition_id: u32 },
+ Bytes(Bytes),
+}
+
+impl NonReplicatedResponse {
+ fn into_reply(
+ self,
+ request_header: &RequestHeader,
+ client_id: u128,
+ session: u64,
+ commit: u64,
+ ) -> Message<ReplyHeader> {
+ match self {
+ Self::Empty => build_empty_reply(request_header, client_id, session, commit),
+ Self::EmptyPolledMessages { partition_id } => {
+ build_reply_with_body(request_header, client_id, session, commit, 16, |body| {
+ body[..4].copy_from_slice(&partition_id.to_le_bytes());
+ body[4..12].copy_from_slice(&0u64.to_le_bytes());
+ body[12..16].copy_from_slice(&0u32.to_le_bytes());
+ })
+ }
+ Self::Bytes(body) => {
+ build_reply_from_bytes(request_header, client_id, session, commit, &body)
+ }
+ }
+ }
+}
+
+fn build_empty_reply(
+ request_header: &RequestHeader,
+ client_id: u128,
+ session: u64,
+ commit: u64,
+) -> Message<ReplyHeader> {
+ build_reply_with_body(request_header, client_id, session, commit, 0, |_| {})
+}
+
fn build_login_register_reply(
+ request_header: &RequestHeader,
+ client_id: u128,
+ session: u64,
+ commit: u64,
+ user_id: u32,
+) -> Message<ReplyHeader> {
+ build_reply_with_body(request_header, client_id, session, commit, 12, |body| {
+ body[..4].copy_from_slice(&user_id.to_le_bytes());
+ body[4..12].copy_from_slice(&session.to_le_bytes());
+ })
+}
+
+fn build_reply_from_bytes(
request_header: &RequestHeader,
client_id: u128,
session: u64,
commit: u64,
body: &Bytes,
) -> Message<ReplyHeader> {
- let total_size = std::mem::size_of::<ReplyHeader>() + body.len();
+ build_reply_with_body(
+ request_header,
+ client_id,
+ session,
+ commit,
+ body.len(),
+ |out| out.copy_from_slice(body),
+ )
+}
+
+fn build_reply_with_body(
+ request_header: &RequestHeader,
+ client_id: u128,
+ session: u64,
+ commit: u64,
+ body_len: usize,
+ write_body: impl FnOnce(&mut [u8]),
+) -> Message<ReplyHeader> {
+ let header_len = std::mem::size_of::<ReplyHeader>();
+ let total_size = header_len + body_len;
let mut reply = Message::<ReplyHeader>::new(total_size);
let header_size = u32::try_from(total_size).expect("reply size must fit into u32");
let header = bytemuck::checked::try_from_bytes_mut::<ReplyHeader>(
- &mut reply.as_mut_slice()[..std::mem::size_of::<ReplyHeader>()],
+ &mut reply.as_mut_slice()[..header_len],
)
.expect("zeroed bytes are valid");
*header = ReplyHeader {
@@ -2263,12 +2456,7 @@ fn build_login_register_reply(
namespace: request_header.namespace,
..Default::default()
};
- if !body.is_empty() {
- // TODO(vsr): encode reply bodies directly into the Message body region
- // instead of serializing to Bytes first and copying them into the
- // aligned message backing here.
- reply.as_mut_slice()[std::mem::size_of::<ReplyHeader>()..total_size].copy_from_slice(body);
- }
+ write_body(&mut reply.as_mut_slice()[header_len..total_size]);
reply
}
diff --git a/core/server-ng/src/server_error.rs b/core/server-ng/src/server_error.rs
index 534f69e1c..769a6ec03 100644
--- a/core/server-ng/src/server_error.rs
+++ b/core/server-ng/src/server_error.rs
@@ -55,6 +55,13 @@ pub enum ServerNgError {
MissingReplicaId,
#[error("cluster node for replica {replica_id} is missing tcp_replica port")]
ClusterReplicaPortMissing { replica_id: u8 },
+ #[error(
+ "cluster bootstrap with empty metadata requires both {username_env} and {password_env} to be set before server-ng can create the root user deterministically"
+ )]
+ ClusterRootCredentialsRequired {
+ username_env: &'static str,
+ password_env: &'static str,
+ },
#[error(
"recovered segment for stream {stream_id}, topic {topic_id}, partition {partition_id} at start_offset {start_offset} has message/index divergence (messages_size={messages_size_bytes}, indexed_size={indexed_size_bytes}, end_offset={end_offset}); recovery aborted before opening listeners. Restore the partition from a healthy replica or snapshot, or move the segment aside for offline repair before restarting."
)]
diff --git a/core/server-ng/src/session_manager.rs b/core/server-ng/src/session_manager.rs
index 5573d3788..1e783141a 100644
--- a/core/server-ng/src/session_manager.rs
+++ b/core/server-ng/src/session_manager.rs
@@ -227,6 +227,18 @@ impl SessionManager {
}
}
+ /// Look up the authenticated user id for a connection.
+ #[must_use]
+ pub fn get_user_id(&self, connection_id: u128) -> Option<u32> {
+ let conn = self.connections.get(&connection_id)?;
+ match conn.state {
+ ConnectionState::Authenticated { user_id } | ConnectionState::Bound { user_id, .. } => {
+ Some(user_id)
+ }
+ ConnectionState::Connected => None,
+ }
+ }
+
/// Look up the connection ID for a client (for routing consensus replies).
#[must_use]
pub fn connection_for_client(&self, client_id: u128) -> Option<u128> {