This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch integration_tests_basics in repository https://gitbox.apache.org/repos/asf/iggy.git
commit f4f2b561ebf63a188b79ea96285881618bc373a2 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Wed May 27 17:28:12 2026 +0200 metadata crud responses --- core/harness_derive/src/codegen.rs | 21 +++ core/integration/tests/mod.rs | 10 ++ core/integration/tests/sdk/hello_world.rs | 46 +++++- core/metadata/src/impls/metadata.rs | 136 +++++++++++++---- core/metadata/src/stm/consumer_group.rs | 20 ++- core/metadata/src/stm/stream.rs | 58 +++++++- core/metadata/src/stm/user.rs | 17 ++- core/sdk/src/vsr.rs | 14 +- core/server-ng/src/bootstrap.rs | 238 +++++++++++++++++++++++++++--- core/shard/src/lib.rs | 20 ++- 10 files changed, 516 insertions(+), 64 deletions(-) diff --git a/core/harness_derive/src/codegen.rs b/core/harness_derive/src/codegen.rs index 4acbd8c25..f4a68c4e1 100644 --- a/core/harness_derive/src/codegen.rs +++ b/core/harness_derive/src/codegen.rs @@ -125,6 +125,18 @@ fn generate_variants(attrs: &IggyTestAttrs) -> Vec<TestVariant> { variants } +/// HTTP is not served by the next-gen (VSR) server, so HTTP transport variants +/// must not compile into the test binary under `--features vsr`. The proc macro +/// cannot read the consuming crate's feature flags at expansion time, so it +/// emits this `cfg` gate on HTTP variants; the `integration` crate resolves it. +fn vsr_transport_cfg(transport: Transport) -> TokenStream { + if matches!(transport, Transport::Http) { + quote!(#[cfg(not(feature = "vsr"))]) + } else { + quote!() + } +} + /// Generate test code from attributes and input function. pub fn generate_tests(attrs: &IggyTestAttrs, input: &ItemFn) -> syn::Result<TokenStream> { let fn_name = &input.sig.ident; @@ -189,8 +201,10 @@ fn generate_single_test( let fixture_seed = generate_fixture_seed(params); let start_and_seed = generate_start_and_seed(attrs, fixture_seed); let harness_param_bindings = generate_harness_param_bindings(params); + let vsr_cfg = vsr_transport_cfg(variant.transport); Ok(quote! { + #vsr_cfg #(#other_attrs)* #[::tokio::test] #[::serial_test::parallel] @@ -231,8 +245,10 @@ fn generate_test_module( let test_name = format_ident!("{}", variant.suffix()); let harness_setup = generate_harness_setup(variant, has_fixtures, attrs); let start_and_seed = generate_start_and_seed(attrs, fixture_seed.clone()); + let vsr_cfg = vsr_transport_cfg(variant.transport); test_fns.push(quote! { + #vsr_cfg #(#other_attrs)* #[::tokio::test] #[::serial_test::parallel] @@ -302,8 +318,10 @@ fn generate_impl_functions_for_test_matrix( if variants.len() == 1 { let variant = &variants[0]; let harness_setup = generate_harness_setup(variant, has_fixtures, attrs); + let vsr_cfg = vsr_transport_cfg(variant.transport); return Ok(quote! { + #vsr_cfg #(#other_attrs)* #[::tokio::test] #[::serial_test::parallel] @@ -328,8 +346,10 @@ fn generate_impl_functions_for_test_matrix( for variant in variants { let impl_name = format_ident!("__impl_{}", variant.suffix()); let harness_setup = generate_harness_setup(variant, has_fixtures, attrs); + let vsr_cfg = vsr_transport_cfg(variant.transport); impl_fns.push(quote! { + #vsr_cfg async fn #impl_name(#(#param_names: #param_types),*) { #fixture_setup #fixture_envs @@ -346,6 +366,7 @@ fn generate_impl_functions_for_test_matrix( let test_name = format_ident!("{}", variant.suffix()); test_fn_calls.push(quote! { + #vsr_cfg #(#other_attrs)* #[::tokio::test] #[::serial_test::parallel] diff --git a/core/integration/tests/mod.rs b/core/integration/tests/mod.rs index afec3d483..3cc92db3e 100644 --- a/core/integration/tests/mod.rs +++ b/core/integration/tests/mod.rs @@ -39,6 +39,16 @@ mod cluster; mod config_provider; #[cfg(not(feature = "vsr"))] mod connectors; +// TODO(vsr): enable the `data_integrity` suite under the `vsr` feature once +// the full VSR path is done. Tier 1 (`verify_user_login_after_restart`, +// `verify_no_plaintext_credentials_on_disk`) is server-side ready (create_user +// reply body, get_users/get_user via `frontend()`, change_password, cross-shard +// committed-reply routing) -- proven by `sdk::hello_world::replicated_*` under +// vsr. Blocked on: (1) these tests use `[Tcp, Http, Quic, WebSocket]` but only +// Tcp works under vsr today (Http does no VSR framing; WebSocket/Quic hit the +// compio `buffer.rs:83` bug); (2) the module also carries +// `verify_after_server_restart` + `verify_consumer_group_partition_assignment`, +// which need send/poll messages + consumer groups (out of Tier 1 scope). #[cfg(not(feature = "vsr"))] mod data_integrity; #[cfg(not(feature = "vsr"))] diff --git a/core/integration/tests/sdk/hello_world.rs b/core/integration/tests/sdk/hello_world.rs index 1e1787c19..905f00ed4 100644 --- a/core/integration/tests/sdk/hello_world.rs +++ b/core/integration/tests/sdk/hello_world.rs @@ -63,9 +63,53 @@ async fn replicated_create_stream_round_trip(harness: &TestHarness) { .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) .await .unwrap(); - client + let stream = client .create_stream("vsr-smoke") .await .expect("create_stream must commit through VSR"); + let topic = client + .create_topic( + &stream.id.try_into().unwrap(), + "vsr-topic", + 1, + CompressionAlgorithm::None, + None, + IggyExpiry::NeverExpire, + MaxTopicSize::ServerDefault, + ) + .await + .expect("create_topic must commit through VSR (CreateTopicWithAssignments transform)"); + client + .create_consumer_group( + &stream.id.try_into().unwrap(), + &topic.id.try_into().unwrap(), + "vsr-group", + ) + .await + .expect("create_consumer_group must commit through VSR"); + client.logout_user().await.unwrap(); +} + +/// VSR raw-PAT return path. The token is minted non-deterministically on the +/// home shard (never replicated), so the committed reply body is empty and the +/// home shard must inject the raw token before answering the client. A blank +/// token here means that injection regressed. +#[cfg(feature = "vsr")] +#[iggy_harness(test_client_transport = [Tcp])] +async fn replicated_create_pat_returns_raw_token(harness: &TestHarness) { + use iggy::prelude::*; + let client = harness.new_client().await.unwrap(); + client + .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) + .await + .unwrap(); + let raw = client + .create_personal_access_token("vsr-pat", PersonalAccessTokenExpiry::NeverExpire) + .await + .expect("create_personal_access_token must commit through VSR"); + assert!( + !raw.token.is_empty(), + "home shard must return the minted raw token, not the empty committed body" + ); client.logout_user().await.unwrap(); } diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs index 32a33380f..557e114a7 100644 --- a/core/metadata/src/impls/metadata.rs +++ b/core/metadata/src/impls/metadata.rs @@ -36,7 +36,7 @@ use iggy_binary_protocol::requests::topics::CreateTopicRequest as WireCreateTopi use iggy_binary_protocol::requests::topics::CreateTopicWithAssignmentsRequest as PersistedCreateTopicRequest; use iggy_binary_protocol::{ Command2, ConsensusHeader, GenericHeader, Operation, PrepareHeader, PrepareOkHeader, - RequestHeader, WireDecode, WireEncode, + ReplyHeader, RequestHeader, WireDecode, WireEncode, }; use iggy_common::IggyError; use iggy_common::variadic; @@ -622,15 +622,6 @@ where async fn on_ack(&self, message: <VsrConsensus<B> as Consensus>::Message<PrepareOkHeader>) { let consensus = self.consensus.as_ref().unwrap(); let header = message.header(); - eprintln!( - "DBG on_ack op={} from_replica={} self={} view={} cmin={} cmax={}", - header.op, - header.replica, - consensus.replica(), - consensus.view(), - consensus.commit_min(), - consensus.commit_max(), - ); if let Err(reason) = ack_preflight(consensus) { warn!( @@ -663,7 +654,8 @@ where } } - if ack_quorum_reached(consensus, PlaneKind::Metadata, header) { + let quorum = ack_quorum_reached(consensus, PlaneKind::Metadata, header); + if quorum { let journal = self.journal.as_ref().unwrap(); debug!( @@ -884,15 +876,6 @@ where return Ok(session); } - eprintln!( - "DBG submit_register entry client={client_id} primary={} normal={} syncing={} cmin={} cmax={} view={}", - consensus.is_primary(), - consensus.is_normal(), - consensus.is_syncing(), - consensus.commit_min(), - consensus.commit_max(), - consensus.view(), - ); // Status + catch-up gate (see doc). Split variants for telemetry: // NotPrimary (try peer) vs NotCaughtUp (retry). Caller policy same. if !is_caught_up_primary(consensus) { @@ -955,7 +938,6 @@ where is_caught_up_primary(consensus), "submit_register_in_process: gate flipped between check and dispatch" ); - eprintln!("DBG submit_register dispatched prepare client={client_id}, awaiting commit"); self.on_replicate(prepare).await; debug_assert!( consensus.view() == view_snapshot && consensus.commit_min() == commit_min_snapshot, @@ -963,12 +945,6 @@ where ); let mut loopback = Vec::new(); consensus.drain_loopback_into(&mut loopback); - eprintln!( - "DBG submit_register after on_replicate client={client_id} loopback={} cmin={} cmax={}", - loopback.len(), - consensus.commit_min(), - consensus.commit_max(), - ); for message in loopback { match message.header().command { Command2::PrepareOk => match message.try_into_typed::<PrepareOkHeader>() { @@ -1112,6 +1088,105 @@ where } } + /// Submit a replicated client request from in-process and await the + /// committed reply. + /// + /// A peer (home) shard relays a client's replicated request here (shard + /// 0 owns the metadata consensus group) and awaits the full committed + /// reply over the pipeline subscriber. The home shard then writes the + /// reply to the originating socket -- it holds the connection and the + /// `vsr -> transport` mapping that this side cannot reconstruct. + /// + /// Mirrors [`Self::submit_register_in_process`] but: (1) uses + /// `request_preflight` (dedup / session check) instead of the register + /// gate, (2) returns the committed `Message<ReplyHeader>` (body = state + /// machine output) rather than just the commit op. + /// + /// # Errors + /// `NotPrimary` / `NotCaughtUp` when this node cannot accept the + /// prepare, `InProgress` / `PipelineFull` on pipeline pressure, + /// `Canceled` when preflight absorbed the request (dedup / eviction / + /// gap) or the pending prepare was canceled before commit. + /// + /// # Panics + /// On a shard without consensus (only shard 0 owns the metadata + /// consensus group); callers must route here only on shard 0. + #[allow(clippy::future_not_send)] + pub async fn submit_request_in_process( + &self, + message: Message<RequestHeader>, + ) -> Result<Message<ReplyHeader>, RegisterSubmitError> { + let request_header = *message.header(); + let client_id = request_header.client; + let session = request_header.session; + let request = request_header.request; + + let consensus = self + .consensus + .as_ref() + .expect("submit_request_in_process: consensus only exists on shard 0"); + + if !is_caught_up_primary(consensus) { + return Err( + if consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing() { + RegisterSubmitError::NotCaughtUp + } else { + RegisterSubmitError::NotPrimary + }, + ); + } + + // Dedup / session / eviction. `false` = absorbed (duplicate cached + // reply already resent, or evicted, or gap). Surface as Canceled so + // the home shard stays silent and the SDK replays. + if !request_preflight(consensus, &self.client_table, client_id, session, request).await { + return Err(RegisterSubmitError::Canceled); + } + + if consensus.pipeline().borrow().is_full() { + return Err(RegisterSubmitError::PipelineFull); + } + + let prepare = self + .prepare_request(message) + .map_err(|_| RegisterSubmitError::Canceled)?; + + consensus.verify_pipeline(); + let view_snapshot = consensus.view(); + let commit_min_snapshot = consensus.commit_min(); + let receiver = consensus.pipeline_message_with_subscriber(PlaneKind::Metadata, &prepare); + debug_assert!( + is_caught_up_primary(consensus), + "submit_request_in_process: gate flipped between check and dispatch" + ); + self.on_replicate(prepare).await; + debug_assert!( + consensus.view() == view_snapshot && consensus.commit_min() == commit_min_snapshot, + "submit_request_in_process: view/commit_min advanced across on_replicate 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" + ), + } + } + + receiver + .await + .map_err(|Canceled| RegisterSubmitError::Canceled) + } + pub fn remove_client_session(&self, client_id: u128) -> bool { self.client_table.borrow_mut().remove_client(client_id) } @@ -1480,6 +1555,11 @@ where client: client_id, session: 0, request: 0, + // Route through the metadata consensus group. The chain-forwarded + // prepare is re-routed on each peer by namespace; a `0` here would + // hash to a non-zero shard with no metadata consensus and be + // silently dropped (see `shard::router::route_typed`). + namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE, ..RequestHeader::default() }; msg @@ -1511,6 +1591,8 @@ where client: client_id, session, request, + // Metadata consensus group (see `build_register_request_message`). + namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE, ..RequestHeader::default() }; msg diff --git a/core/metadata/src/stm/consumer_group.rs b/core/metadata/src/stm/consumer_group.rs index aaf50a4f2..b922fbeb5 100644 --- a/core/metadata/src/stm/consumer_group.rs +++ b/core/metadata/src/stm/consumer_group.rs @@ -22,9 +22,12 @@ use bytes::Bytes; use ahash::AHashMap; use iggy_binary_protocol::WireIdentifier; +use iggy_binary_protocol::codec::WireEncode; use iggy_binary_protocol::requests::consumer_groups::{ CreateConsumerGroupRequest, DeleteConsumerGroupRequest, }; +use iggy_binary_protocol::responses::consumer_groups::consumer_group_response::ConsumerGroupResponse; +use iggy_binary_protocol::responses::consumer_groups::get_consumer_group::ConsumerGroupDetailsResponse; use serde::{Deserialize, Serialize}; use slab::Slab; use std::sync::Arc; @@ -164,9 +167,9 @@ impl ConsumerGroups { } } -// TODO(hubcio): Serialize proper reply (e.g. assigned group ID) instead of empty Bytes. impl StateHandler for CreateConsumerGroupRequest { type State = ConsumerGroupsInner; + #[allow(clippy::cast_possible_truncation)] fn apply( &self, state: &mut ConsumerGroupsInner, @@ -199,7 +202,20 @@ impl StateHandler for CreateConsumerGroupRequest { let key = (Arc::from(s.as_str()), Arc::from(t.as_str())); state.topic_name_index.entry(key).or_default().push(id); } - Bytes::new() + + // Reply body: the SDK `create_consumer_group` decodes a + // `ConsumerGroupDetailsResponse`. A freshly created group has no + // assigned partitions and no joined members yet. + ConsumerGroupDetailsResponse { + group: ConsumerGroupResponse { + id: id as u32, + partitions_count: 0, + members_count: 0, + name: self.name.clone(), + }, + members: Vec::new(), + } + .to_bytes() } } diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs index 42013fb95..a3e3b231a 100644 --- a/core/metadata/src/stm/stream.rs +++ b/core/metadata/src/stm/stream.rs @@ -21,6 +21,7 @@ use crate::{collect_handlers, define_state, impl_fill_restore}; use ahash::AHashMap; use bytes::Bytes; use iggy_binary_protocol::WireIdentifier; +use iggy_binary_protocol::codec::WireEncode; use iggy_binary_protocol::requests::partitions::{ CreatePartitionsWithAssignmentsRequest, DeletePartitionsRequest, }; @@ -30,6 +31,9 @@ use iggy_binary_protocol::requests::streams::{ use iggy_binary_protocol::requests::topics::{ CreateTopicWithAssignmentsRequest, DeleteTopicRequest, PurgeTopicRequest, UpdateTopicRequest, }; +use iggy_binary_protocol::responses::streams::StreamResponse; +use iggy_binary_protocol::responses::streams::get_stream::{GetStreamResponse, TopicHeader}; +use iggy_binary_protocol::responses::topics::get_topic::{GetTopicResponse, PartitionResponse}; use iggy_common::{ CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize, StreamStats, TopicStats, }; @@ -349,9 +353,9 @@ impl Streams { } } -// TODO(hubcio): Serialize proper reply (e.g. assigned stream ID) instead of empty Bytes. impl StateHandler for CreateStreamRequest { type State = StreamsInner; + #[allow(clippy::cast_possible_truncation)] fn apply(&self, state: &mut StreamsInner, timestamp: IggyTimestamp) -> Bytes { let name_arc: Arc<str> = Arc::from(self.name.as_str()); if state.index.contains_key(&name_arc) { @@ -372,7 +376,22 @@ impl StateHandler for CreateStreamRequest { stream.id = id; } state.index.insert(name_arc, id); - Bytes::new() + + // Reply body: a freshly created stream has no topics. The SDK + // `create_stream` decodes a `GetStreamResponse`. Serialization is + // local to this state machine (it owns the committed shape). + GetStreamResponse { + stream: StreamResponse { + id: id as u32, + created_at: timestamp.as_micros(), + topics_count: 0, + size_bytes: 0, + messages_count: 0, + name: self.name.clone(), + }, + topics: Vec::new(), + } + .to_bytes() } } @@ -425,9 +444,9 @@ impl StateHandler for PurgeStreamRequest { } } -// TODO(hubcio): Serialize proper reply (e.g. assigned topic ID) instead of empty Bytes. impl StateHandler for CreateTopicWithAssignmentsRequest { type State = StreamsInner; + #[allow(clippy::cast_possible_truncation)] fn apply(&self, state: &mut StreamsInner, timestamp: IggyTimestamp) -> Bytes { let Some(stream_id) = state.resolve_stream_id(&self.request.stream_id) else { return Bytes::new(); @@ -478,7 +497,38 @@ impl StateHandler for CreateTopicWithAssignmentsRequest { } stream.topic_index.insert(name_arc, topic_id); - Bytes::new() + + // Reply body: the SDK `create_topic` decodes a `GetTopicResponse`. A + // freshly created topic has empty/zeroed segment stats; each assigned + // partition is reported at offset 0. + let partitions = self + .partitions + .iter() + .map(|partition| PartitionResponse { + id: partition.partition_id, + created_at: timestamp.as_micros(), + segments_count: 0, + current_offset: 0, + size_bytes: 0, + messages_count: 0, + }) + .collect(); + GetTopicResponse { + topic: TopicHeader { + id: topic_id as u32, + created_at: timestamp.as_micros(), + partitions_count: self.partitions.len() as u32, + message_expiry: self.request.message_expiry, + compression_algorithm: self.request.compression_algorithm, + max_topic_size: self.request.max_topic_size, + replication_factor, + size_bytes: 0, + messages_count: 0, + name: self.request.name.clone(), + }, + partitions, + } + .to_bytes() } } diff --git a/core/metadata/src/stm/user.rs b/core/metadata/src/stm/user.rs index 3aa0c7bb7..c0b05a4b5 100644 --- a/core/metadata/src/stm/user.rs +++ b/core/metadata/src/stm/user.rs @@ -28,6 +28,8 @@ use iggy_binary_protocol::requests::users::{ ChangePasswordRequest, CreateUserRequest, DeleteUserRequest, UpdatePermissionsRequest, UpdateUserRequest, }; +use iggy_binary_protocol::responses::users::get_user::UserDetailsResponse; +use iggy_binary_protocol::responses::users::user_response::UserResponse; use iggy_binary_protocol::{WireIdentifier, WireName}; use iggy_common::{ GlobalPermissions, IggyExpiry, IggyTimestamp, Permissions, PersonalAccessToken, @@ -271,7 +273,6 @@ impl WireDecode for DeletePersonalAccessTokenRequest { } } -// TODO(hubcio): Serialize proper reply (e.g. assigned user ID) instead of empty Bytes. impl StateHandler for CreateUserRequest { type State = UsersInner; #[allow(clippy::cast_possible_truncation)] @@ -305,7 +306,19 @@ impl StateHandler for CreateUserRequest { state .personal_access_tokens .insert(id as UserId, AHashMap::default()); - Bytes::new() + + // Reply body: the SDK `create_user` decodes a `UserDetailsResponse`. + // Serialization local to this state machine. + UserDetailsResponse { + user: UserResponse { + id: id as u32, + created_at: timestamp.as_micros(), + status: self.status, + username: self.username.clone(), + }, + permissions: self.permissions.clone(), + } + .to_bytes() } } diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs index 318a00b41..b0481e9fa 100644 --- a/core/sdk/src/vsr.rs +++ b/core/sdk/src/vsr.rs @@ -77,7 +77,19 @@ pub(crate) fn encode_request_header( _ => { let operation = operation_for_code(code)?; let session_id = session.session().ok_or(IggyError::Unauthenticated)?; - (operation, session.next_request_id(), session_id) + // NonReplicated ops (ping, reads) bypass server-side dedup -- + // `ClientTable` only tracks request_ids for replicated ops. If + // they consumed the monotonic counter, the next replicated + // request would skip an id and the primary's `request_preflight` + // would see a `RequestGap` and silently drop it. Read the + // current id without advancing; the server ignores it for + // NonReplicated. + let request_id = if operation == Operation::NonReplicated { + session.current_request_id() + } else { + session.next_request_id() + }; + (operation, request_id, session_id) } }; let namespace = namespace_for_request(code, payload, operation)?; diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index 0df29dcb0..3e2f5e72e 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -33,7 +33,7 @@ use consensus::{LocalPipeline, MetadataHandle, PartitionsHandle, Sequencer, VsrC use crossfire::{AsyncRxTrait, AsyncTxTrait}; use iggy_binary_protocol::codes::{ GET_CLUSTER_METADATA_CODE, GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE, - GET_TOPICS_CODE, PING_CODE, POLL_MESSAGES_CODE, + GET_TOPICS_CODE, GET_USER_CODE, GET_USERS_CODE, PING_CODE, POLL_MESSAGES_CODE, }; use iggy_binary_protocol::requests::consumer_offsets::{ DeleteConsumerOffset2Request, DeleteConsumerOffsetRequest, StoreConsumerOffset2Request, @@ -47,7 +47,10 @@ use iggy_binary_protocol::requests::personal_access_tokens::{ use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest; use iggy_binary_protocol::requests::streams::{GetStreamRequest, GetStreamsRequest}; use iggy_binary_protocol::requests::topics::{GetTopicRequest, GetTopicsRequest}; -use iggy_binary_protocol::requests::users::{LoginRegisterRequest, LoginRegisterWithPatRequest}; +use iggy_binary_protocol::requests::users::{ + GetUserRequest, LoginRegisterRequest, LoginRegisterWithPatRequest, +}; +use iggy_binary_protocol::responses::personal_access_tokens::RawPersonalAccessTokenResponse; use iggy_binary_protocol::responses::streams::StreamResponse; use iggy_binary_protocol::responses::streams::get_stream::{ GetStreamResponse, TopicHeader as StreamTopicHeader, @@ -59,6 +62,9 @@ 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::get_user::UserDetailsResponse; +use iggy_binary_protocol::responses::users::get_users::GetUsersResponse; +use iggy_binary_protocol::responses::users::user_response::UserResponse; use iggy_binary_protocol::{ Command2, EvictionReason, GenericHeader, Operation, ReplyHeader, RequestHeader, WireDecode, WireEncode, WireIdentifier, WireName, WirePartitioning, @@ -2194,7 +2200,9 @@ fn make_deferred_client_request_handler( /// 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. -fn make_metadata_submit_handler(shard_handle: &ServerNgShardHandle) -> shard::MetadataSubmitHandler { +fn make_metadata_submit_handler( + shard_handle: &ServerNgShardHandle, +) -> shard::MetadataSubmitHandler { let shard_handle = Rc::clone(shard_handle); Rc::new(move |submit| { let shard_handle = Rc::clone(&shard_handle); @@ -2229,6 +2237,22 @@ fn make_metadata_submit_handler(shard_handle: &ServerNgShardHandle) -> shard::Me .ok(); let _ = reply.try_send(commit); } + shard::MetadataSubmit::ClientRequest { request, reply } => { + let committed = match request.try_into_typed::<RequestHeader>() { + Ok(typed) => shard + .plane + .metadata() + .submit_request_in_process(typed) + .await + .ok() + .map(server_common::Message::into_generic), + Err(error) => { + warn!(?error, "ClientRequest submit: undecodable request header"); + None + } + }; + let _ = reply.try_send(committed); + } } }) .detach(); @@ -2412,19 +2436,64 @@ 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) => { + let (request, raw_pat_token) = + match maybe_rewrite_pat_request(sessions, transport_client_id, request) { + Ok(rewritten) => rewritten, + Err(error) => { + warn!( + transport_client_id, + error = %error, + operation = ?header.operation, + "dropping request with invalid PAT replication context" + ); + return; + } + }; + let request_header = *request.header(); + // Replicated request: run consensus on the metadata owner (shard 0) and + // bring the committed reply back here. This shard owns the connection, + // so it writes the reply to the socket via the transport client id -- + // shard 0 can't route by the consensus client id (no home-shard bits). + match submit_client_request_on_owner(shard, request).await { + Some(reply) => { + // The raw PAT token never enters consensus (it is non-deterministic + // and secret), so the committed reply body is empty. Substitute the + // raw-token response here, on the minting client's home shard, using + // the confirmed commit position from the committed reply. + let reply = match build_raw_pat_reply(&request_header, reply, raw_pat_token) { + Ok(reply) => reply, + Err(error) => { + warn!( + transport_client_id, + error = %error, + "failed to build raw PAT reply" + ); + return; + } + }; + if let Err(error) = shard + .bus + .send_to_client(transport_client_id, reply.into_frozen()) + .await + { + warn!( + transport_client_id, + error = %error, + operation = ?header.operation, + "failed to deliver committed reply to client" + ); + } + } + None => { + // Transient submit failure (not primary / not caught up / dedup + // absorbed). Stay silent; the SDK read-timeout replays. warn!( transport_client_id, - error = %error, operation = ?header.operation, - "dropping request with invalid PAT replication context" + "replicated request not committed (transient); client will replay" ); - return; } - }; - shard.dispatch(request.into_generic()); + } } #[allow(clippy::future_not_send)] @@ -2550,6 +2619,37 @@ async fn submit_logout_on_owner( } } +/// Submit a replicated client request to the metadata owner (shard 0) and +/// return the committed reply. +/// +/// The metadata consensus group lives on shard 0, but the connection lives +/// on the home shard (this shard). Run consensus where it belongs and bring +/// the committed reply back here so the caller can write it to the +/// originating socket -- shard 0 cannot route the reply by the consensus +/// `client` id (it's the VSR id, not the transport/home-shard-encoding id). +/// `None` = transient submit failure (SDK read-timeout replays). +#[allow(clippy::future_not_send)] +async fn submit_client_request_on_owner( + shard: &Rc<ServerNgShard>, + request: Message<RequestHeader>, +) -> Option<Message<GenericHeader>> { + if shard.id == 0 { + return shard + .plane + .metadata() + .submit_request_in_process(request) + .await + .ok() + .map(server_common::Message::into_generic); + } + let (reply, rx) = shard::channel::<Option<Message<GenericHeader>>>(1); + shard.forward_metadata_submit(shard::MetadataSubmit::ClientRequest { + request: request.into_generic(), + reply, + }); + rx.recv().await.ok().flatten() +} + #[allow(clippy::future_not_send)] async fn handle_logout_request( shard: &Rc<ServerNgShard>, @@ -2607,17 +2707,18 @@ fn maybe_rewrite_pat_request( sessions: &Rc<RefCell<SessionManager>>, transport_client_id: u128, request: Message<RequestHeader>, -) -> Result<Message<RequestHeader>, IggyError> { +) -> Result<(Message<RequestHeader>, Option<String>), 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), + _ => return Ok((request, None)), }; let body = request_body(&request); + let mut raw_token = None; let rewritten = match operation { Operation::CreatePersonalAccessToken => { let wire = WireCreatePersonalAccessTokenRequest::decode_from(body) @@ -2625,8 +2726,10 @@ fn maybe_rewrite_pat_request( // Primary mints the raw token + hash here and ships the hash // through consensus. Replicas decode the hash directly. Doing // this inside `CreatePersonalAccessTokenRequest::apply` would - // call `ring::rand` per-replica and diverge state. - let token_hash = mint_pat_token_hash(); + // call `ring::rand` per-replica and diverge state. The raw token + // is returned to this client only (see `handle_client_request`). + let (raw, token_hash) = mint_pat_raw_and_hash(); + raw_token = Some(raw); ReplicatedCreatePersonalAccessTokenRequest { user_id, name: wire.name, @@ -2647,20 +2750,20 @@ fn maybe_rewrite_pat_request( _ => unreachable!(), }; - rewrite_request_body(&request, &rewritten) + Ok((rewrite_request_body(&request, &rewritten)?, raw_token)) } -/// Mints a fresh PAT raw token and returns its hex-encoded SHA-256 hash -/// (64 bytes ASCII) for replication. The raw token is currently dropped -- -/// see the TODO on `CreatePersonalAccessTokenRequest::apply` for the missing -/// return-to-client path. -fn mint_pat_token_hash() -> [u8; 64] { - let (_raw, hash) = iggy_common::PersonalAccessToken::mint_raw_and_hash(); +/// Mints a fresh PAT and returns the raw token plus its hex-encoded SHA-256 +/// hash (64 bytes ASCII). Only the hash is replicated; the raw token is +/// returned to the minting client by the home shard (it cannot be reproduced +/// by the deterministic `apply` running on every replica). +fn mint_pat_raw_and_hash() -> (String, [u8; 64]) { + let (raw, hash) = iggy_common::PersonalAccessToken::mint_raw_and_hash(); let bytes = hash.as_bytes(); let mut out = [0u8; 64]; let len = bytes.len().min(64); out[..len].copy_from_slice(&bytes[..len]); - out + (raw, out) } fn rewrite_request_body( @@ -2983,6 +3086,18 @@ fn build_non_replicated_response( build_get_topics_response(shard, &request.stream_id)?.to_bytes(), )) } + GET_USERS_CODE => Ok(NonReplicatedResponse::Bytes( + build_get_users_response(shard)?.to_bytes(), + )), + GET_USER_CODE => { + let request = + GetUserRequest::decode_from(body).map_err(|_| IggyError::InvalidCommand)?; + build_get_user_response(shard, &request.user_id).map(|response| { + response.map_or(NonReplicatedResponse::Empty, |response| { + NonReplicatedResponse::Bytes(response.to_bytes()) + }) + }) + } POLL_MESSAGES_CODE => { let request = PollMessagesRequest::decode_from(body).map_err(|_| IggyError::InvalidCommand)?; @@ -3118,6 +3233,53 @@ fn build_get_streams_response(shard: &Rc<ServerNgShard>) -> Result<GetStreamsRes }) } +#[allow(clippy::cast_possible_truncation)] +fn user_response(user: &metadata::stm::user::User) -> Result<UserResponse, IggyError> { + Ok(UserResponse { + id: user.id, + created_at: user.created_at.as_micros(), + status: user.status.as_code(), + username: WireName::new(user.username.as_ref()).map_err(|_| IggyError::InvalidFormat)?, + }) +} + +fn build_get_users_response(shard: &Rc<ServerNgShard>) -> Result<GetUsersResponse, IggyError> { + shard.plane.metadata().mux_stm.users().read(|users| { + users + .items + .iter() + .map(|(_, user)| user_response(user)) + .collect::<Result<Vec<_>, _>>() + .map(|users| GetUsersResponse { users }) + }) +} + +fn build_get_user_response( + shard: &Rc<ServerNgShard>, + user_id: &WireIdentifier, +) -> Result<Option<UserDetailsResponse>, IggyError> { + shard.plane.metadata().mux_stm.users().read(|users| { + let resolved = match user_id { + WireIdentifier::Numeric(id) => { + let id = *id as usize; + users.items.contains(id).then_some(id) + } + WireIdentifier::String(name) => users.index.get(name.as_str()).map(|&id| id as usize), + }; + let Some(id) = resolved else { + return Ok(None); + }; + let user = users.items.get(id).ok_or(IggyError::InvalidIdentifier)?; + Ok(Some(UserDetailsResponse { + user: user_response(user)?, + permissions: user + .permissions + .as_ref() + .map(|p| iggy_common::wire_conversions::permissions_to_wire(p)), + })) + }) +} + fn build_get_topic_response( shard: &Rc<ServerNgShard>, stream_id: &WireIdentifier, @@ -3466,6 +3628,36 @@ fn build_reply_from_bytes( ) } +/// If a raw PAT token was minted (`CreatePersonalAccessToken`), replace the +/// committed reply -- whose body is empty because the raw token never entered +/// consensus -- with a `RawPersonalAccessTokenResponse`, reusing the confirmed +/// commit position from the committed reply. Otherwise the committed reply +/// passes through unchanged. +fn build_raw_pat_reply( + request_header: &RequestHeader, + committed: Message<GenericHeader>, + raw_token: Option<String>, +) -> Result<Message<GenericHeader>, IggyError> { + let Some(raw) = raw_token else { + return Ok(committed); + }; + let header_len = std::mem::size_of::<ReplyHeader>(); + let committed_header = + bytemuck::checked::try_from_bytes::<ReplyHeader>(&committed.as_slice()[..header_len]) + .map_err(|_| IggyError::InvalidFormat)?; + let commit = committed_header.commit; + let token = WireName::new(raw.as_str()).map_err(|_| IggyError::InvalidFormat)?; + let body = RawPersonalAccessTokenResponse { token }.to_bytes(); + let reply = build_reply_from_bytes( + request_header, + request_header.client, + request_header.session, + commit, + &body, + ); + Ok(reply.into_generic()) +} + fn build_reply_with_body( request_header: &RequestHeader, client_id: u128, diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 72c90f21d..cc1fbd09b 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -120,12 +120,24 @@ pub enum MetadataSubmit { request: u64, reply: Sender<Option<u64>>, }, + /// A peer (home) shard relays a client's replicated request to shard 0 + /// and awaits the committed reply over `reply` (`None` on a transient + /// submit failure). The home shard then writes the reply to the + /// originating socket -- it owns the connection and the + /// `vsr -> transport` mapping, which shard 0 cannot reconstruct from the + /// consensus client id. + ClientRequest { + request: Message<GenericHeader>, + reply: Sender<Option<Message<GenericHeader>>>, + }, } -/// Handler shard 0 runs for an inbound [`MetadataSubmit`]. server-ng wires -/// it to `submit_register_in_process` / `submit_logout_in_process` and -/// sends the result back over the frame's `reply` sender. `None` on a peer -/// shard (no consensus): a peer must never receive this frame. +/// Handler shard 0 runs for an inbound [`MetadataSubmit`]. +/// +/// server-ng wires it to `submit_register_in_process` / +/// `submit_logout_in_process` / `submit_request_in_process` and sends the +/// result back over the frame's `reply` sender. A peer shard (no consensus) +/// must never receive this frame. pub type MetadataSubmitHandler = Rc<dyn Fn(MetadataSubmit)>; /// Create a bounded inter-shard channel whose sender is tagged with the
