This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch send_messages_confirmation in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 991d84cd0456b4b0979f7f0ba5aecb110a377628 Author: Hubert Gruszecki <[email protected]> AuthorDate: Thu Jul 30 11:33:36 2026 +0200 feat(server-ng): confirm committed sends with partition and offset A committed SendMessages answered with an empty body, so producers could not learn which partition a balanced or keyed batch landed in nor the offsets it was assigned. server-ng now builds a confirmation payload at commit time on the partition group primary: a count-prefixed list of entries carrying stream, topic and partition ids plus the base offset of the batch, derived from the journaled batch header so any primary draining the commit reports identical offsets. The list shape lets a future multi-partition produce grow the count without a wire change; an undecodable journal entry degrades to an empty list rather than an empty body. The SDK surfaces it as an Option: servers answering with an empty body (legacy, older server-ng) report None instead of a decode error. IggyProducer returns one confirmation per chunk it split a send into; the background producer keeps returning nothing since its shard merges batches and no 1:1 mapping exists. The HTTP endpoint mirrors the tuple as JSON on the 201 reply. --- Cargo.lock | 1 + core/ai/mcp/src/service/mod.rs | 5 +- core/binary_protocol/src/responses/messages/mod.rs | 2 + .../src/responses/messages/send_messages.rs | 264 ++++++++++++++++++++ core/common/src/lib.rs | 3 + core/common/src/traits/binary_impls/messages.rs | 92 ++++++- core/common/src/traits/message_client.rs | 8 +- core/integration/tests/sdk/mod.rs | 1 + core/integration/tests/sdk/producer/background.rs | 6 +- core/integration/tests/sdk/send_confirmation.rs | 272 +++++++++++++++++++++ .../server/scenarios/authentication_scenario.rs | 1 + core/partitions/src/iggy_partition.rs | 139 +++++++++-- core/sdk/Cargo.toml | 1 + .../src/client_wrappers/binary_message_client.rs | 3 +- core/sdk/src/clients/binary_message.rs | 3 +- core/sdk/src/clients/producer.rs | 90 +++++-- core/sdk/src/clients/producer_dispatcher.rs | 4 +- core/sdk/src/clients/producer_sharding.rs | 6 +- core/sdk/src/http/messages.rs | 122 +++++++-- core/sdk/src/prelude.rs | 9 +- core/server-ng/src/http/handlers.rs | 109 ++++++++- core/server-ng/src/http/reply.rs | 123 +++++++++- core/server-ng/src/http/submit.rs | 8 +- 23 files changed, 1176 insertions(+), 96 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 50939ed39..f06b25cd2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6634,6 +6634,7 @@ dependencies = [ "rustls", "secrecy", "serde", + "serde_json", "tokio", "tokio-rustls", "tokio-tungstenite", diff --git a/core/ai/mcp/src/service/mod.rs b/core/ai/mcp/src/service/mod.rs index 8b382adc2..1653b02fb 100644 --- a/core/ai/mcp/src/service/mod.rs +++ b/core/ai/mcp/src/service/mod.rs @@ -412,7 +412,10 @@ impl IggyService { &partitioning, &mut messages, ) - .await, + .await + // Wire responses carry no serde on purpose; the MCP reply stays + // the unit acknowledgement it always was. + .map(|_| ()), ) } diff --git a/core/binary_protocol/src/responses/messages/mod.rs b/core/binary_protocol/src/responses/messages/mod.rs index c272f3424..dbb981f3f 100644 --- a/core/binary_protocol/src/responses/messages/mod.rs +++ b/core/binary_protocol/src/responses/messages/mod.rs @@ -16,5 +16,7 @@ // under the License. pub mod poll_messages; +pub mod send_messages; pub use poll_messages::{PollMessagesResponse, PollMessagesResponseHeader}; +pub use send_messages::{SendMessagesConfirmationResponse, SendMessagesResponse}; diff --git a/core/binary_protocol/src/responses/messages/send_messages.rs b/core/binary_protocol/src/responses/messages/send_messages.rs new file mode 100644 index 000000000..f24f07fb3 --- /dev/null +++ b/core/binary_protocol/src/responses/messages/send_messages.rs @@ -0,0 +1,264 @@ +// 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. + +use crate::codec::{WireDecode, WireEncode, capped_capacity, read_u32_le, read_u64_le}; +use crate::error::WireError; +use bytes::{BufMut, BytesMut}; +use std::borrow::Cow; + +/// Size of one confirmation entry: +/// `stream_id(4) + topic_id(4) + partition_id(4) + base_offset(8)`. +const CONFIRMATION_SIZE: usize = 20; + +/// Commit confirmation for one partition written by a `SendMessages` request. +/// +/// Wire format: +/// ```text +/// [stream_id:4][topic_id:4][partition_id:4][base_offset:8] +/// ``` +/// +/// `base_offset` is the offset assigned to the first message of the batch in +/// that partition. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SendMessagesConfirmationResponse { + pub stream_id: u32, + pub topic_id: u32, + pub partition_id: u32, + pub base_offset: u64, +} + +impl WireEncode for SendMessagesConfirmationResponse { + fn encoded_size(&self) -> usize { + CONFIRMATION_SIZE + } + + fn encode(&self, buf: &mut BytesMut) { + buf.put_u32_le(self.stream_id); + buf.put_u32_le(self.topic_id); + buf.put_u32_le(self.partition_id); + buf.put_u64_le(self.base_offset); + } +} + +impl WireDecode for SendMessagesConfirmationResponse { + fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> { + let stream_id = read_u32_le(buf, 0)?; + let topic_id = read_u32_le(buf, 4)?; + let partition_id = read_u32_le(buf, 8)?; + let base_offset = read_u64_le(buf, 12)?; + Ok(( + Self { + stream_id, + topic_id, + partition_id, + base_offset, + }, + CONFIRMATION_SIZE, + )) + } +} + +/// `SendMessages` response. +/// +/// Wire format: +/// ```text +/// [confirmations_count:4][SendMessagesConfirmationResponse]* +/// ``` +/// +/// `confirmations_count == 0` means the batch committed but no offsets are +/// available. The server currently reports a single partition per request; +/// the list decodes any count so a later multi-partition send needs no wire +/// change. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SendMessagesResponse { + pub confirmations: Vec<SendMessagesConfirmationResponse>, +} + +impl WireEncode for SendMessagesResponse { + fn encoded_size(&self) -> usize { + 4 + self.confirmations.len() * CONFIRMATION_SIZE + } + + #[allow(clippy::cast_possible_truncation)] + fn encode(&self, buf: &mut BytesMut) { + buf.put_u32_le(self.confirmations.len() as u32); + for confirmation in &self.confirmations { + confirmation.encode(buf); + } + } +} + +impl WireDecode for SendMessagesResponse { + fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> { + let confirmations_count = read_u32_le(buf, 0)?; + let remaining = buf.len().saturating_sub(4); + let mut confirmations = Vec::with_capacity(capped_capacity( + confirmations_count as usize, + remaining, + CONFIRMATION_SIZE, + )); + let mut offset = 4; + for _ in 0..confirmations_count { + let (confirmation, consumed) = + SendMessagesConfirmationResponse::decode(&buf[offset..])?; + offset += consumed; + confirmations.push(confirmation); + } + // The payload is the whole frame body, never a prefix of a larger + // value, so leftover bytes mean a shape this build cannot read. + if offset != buf.len() { + return Err(WireError::Validation(Cow::Owned(format!( + "send_messages response has {} trailing bytes", + buf.len() - offset + )))); + } + Ok((Self { confirmations }, offset)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn confirmation(partition_id: u32) -> SendMessagesConfirmationResponse { + SendMessagesConfirmationResponse { + stream_id: 1, + topic_id: 2, + partition_id, + base_offset: 42, + } + } + + #[test] + fn roundtrip_single_confirmation() { + let response = SendMessagesResponse { + confirmations: vec![confirmation(3)], + }; + let bytes = response.to_bytes(); + assert_eq!(bytes.len(), 4 + CONFIRMATION_SIZE); + let (decoded, consumed) = SendMessagesResponse::decode(&bytes).unwrap(); + assert_eq!(consumed, bytes.len()); + assert_eq!(decoded, response); + } + + #[test] + fn roundtrip_multiple_confirmations() { + let response = SendMessagesResponse { + confirmations: vec![confirmation(0), confirmation(1), confirmation(2)], + }; + let bytes = response.to_bytes(); + let (decoded, consumed) = SendMessagesResponse::decode(&bytes).unwrap(); + assert_eq!(consumed, bytes.len()); + assert_eq!(decoded, response); + } + + #[test] + fn roundtrip_zero_confirmations() { + let response = SendMessagesResponse { + confirmations: vec![], + }; + let bytes = response.to_bytes(); + assert_eq!(&bytes[..], &[0x00, 0x00, 0x00, 0x00]); + let (decoded, consumed) = SendMessagesResponse::decode(&bytes).unwrap(); + assert_eq!(consumed, 4); + assert_eq!(decoded, response); + } + + #[test] + fn wire_compat_confirmation_layout() { + let response = SendMessagesResponse { + confirmations: vec![SendMessagesConfirmationResponse { + stream_id: 1, + topic_id: 2, + partition_id: 3, + base_offset: 4, + }], + }; + assert_eq!( + &response.to_bytes()[..], + &[ + 0x01, 0x00, 0x00, 0x00, // count + 0x01, 0x00, 0x00, 0x00, // stream_id + 0x02, 0x00, 0x00, 0x00, // topic_id + 0x03, 0x00, 0x00, 0x00, // partition_id + 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // base_offset + ] + ); + } + + #[test] + fn confirmation_truncated_returns_error() { + let bytes = confirmation(1).to_bytes(); + for i in 0..bytes.len() { + assert!( + SendMessagesConfirmationResponse::decode(&bytes[..i]).is_err(), + "expected error for truncation at byte {i}" + ); + } + } + + #[test] + fn truncated_returns_error() { + let response = SendMessagesResponse { + confirmations: vec![confirmation(0), confirmation(1)], + }; + let bytes = response.to_bytes(); + for i in 0..bytes.len() { + assert!( + SendMessagesResponse::decode(&bytes[..i]).is_err(), + "expected error for truncation at byte {i}" + ); + } + } + + #[test] + fn trailing_bytes_rejected() { + let response = SendMessagesResponse { + confirmations: vec![confirmation(1)], + }; + let mut bytes = response.to_bytes().to_vec(); + bytes.push(0xFF); + assert!(matches!( + SendMessagesResponse::decode(&bytes), + Err(WireError::Validation(_)) + )); + } + + #[test] + fn trailing_bytes_after_zero_confirmations_rejected() { + let bytes = [0x00, 0x00, 0x00, 0x00, 0x00]; + assert!(matches!( + SendMessagesResponse::decode(&bytes), + Err(WireError::Validation(_)) + )); + } + + #[test] + fn empty_input_returns_error() { + assert!(matches!( + SendMessagesResponse::decode(&[]), + Err(WireError::UnexpectedEof { .. }) + )); + } + + #[test] + fn bogus_confirmations_count_does_not_oom() { + let mut buf = BytesMut::new(); + buf.put_u32_le(u32::MAX); + assert!(SendMessagesResponse::decode(&buf).is_err()); + } +} diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs index 12f139d2a..ad6048941 100644 --- a/core/common/src/lib.rs +++ b/core/common/src/lib.rs @@ -52,6 +52,9 @@ pub use http::streams::*; pub use http::system::*; pub use http::topics::*; pub use http::users::*; +pub use iggy_binary_protocol::responses::messages::{ + SendMessagesConfirmationResponse, SendMessagesResponse, +}; pub use traits::binary_client::BinaryClient; pub use traits::binary_transport::BinaryTransport; #[cfg(feature = "vsr")] diff --git a/core/common/src/traits/binary_impls/messages.rs b/core/common/src/traits/binary_impls/messages.rs index 24c2a799d..0eda65a4e 100644 --- a/core/common/src/traits/binary_impls/messages.rs +++ b/core/common/src/traits/binary_impls/messages.rs @@ -22,7 +22,7 @@ use crate::wire_conversions::{ }; use crate::{ Consumer, Identifier, IggyError, IggyMessage, MessageClient, Partitioning, PolledMessages, - PollingStrategy, + PollingStrategy, SendMessagesResponse, }; #[cfg(feature = "vsr")] use crate::{ConsumerKind, PartitioningKind, TopicClient, calculate_32}; @@ -253,6 +253,17 @@ async fn poll_group_messages<B: BinaryClient>( Ok(PolledMessages::empty()) } +/// Map a raw `SendMessages` reply body to its confirmation payload. An empty +/// body means the batch was accepted but no offsets were reported: the legacy +/// server and `server-ng` builds predating the confirmation reply answer that +/// way, so absence must never surface as a decode failure. +fn decode_send_confirmations(response: &[u8]) -> Result<Option<SendMessagesResponse>, IggyError> { + if response.is_empty() { + return Ok(None); + } + super::decode_response::<SendMessagesResponse>(response).map(Some) +} + #[async_trait::async_trait] impl<B: BinaryClient> MessageClient for B { async fn poll_messages( @@ -303,7 +314,7 @@ impl<B: BinaryClient> MessageClient for B { topic_id: &Identifier, partitioning: &Partitioning, messages: &mut [IggyMessage], - ) -> Result<(), IggyError> { + ) -> Result<Option<SendMessagesResponse>, IggyError> { fail_if_not_authenticated(self).await?; // VSR: resolve Balanced/MessagesKey to an explicit partition client-side. // An explicit `PartitionId` needs no resolution, so borrow the input @@ -344,9 +355,10 @@ impl<B: BinaryClient> MessageClient for B { &wire_partitioning, &raw_messages, ); - self.send_raw_with_response(SEND_MESSAGES_CODE, buf.freeze()) + let response = self + .send_raw_with_response(SEND_MESSAGES_CODE, buf.freeze()) .await?; - Ok(()) + decode_send_confirmations(&response) } async fn flush_unsaved_buffer( @@ -368,3 +380,75 @@ impl<B: BinaryClient> MessageClient for B { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::decode_send_confirmations; + use crate::{IggyError, SendMessagesConfirmationResponse, SendMessagesResponse}; + use iggy_binary_protocol::codec::WireEncode; + + fn response() -> SendMessagesResponse { + SendMessagesResponse { + confirmations: vec![SendMessagesConfirmationResponse { + stream_id: 1, + topic_id: 2, + partition_id: 3, + base_offset: 42, + }], + } + } + + #[test] + fn empty_body_is_absent_confirmation() { + assert!(matches!(decode_send_confirmations(&[]), Ok(None))); + } + + #[test] + fn populated_body_decodes() { + let expected = response(); + let bytes = expected.to_bytes(); + let decoded = decode_send_confirmations(&bytes).expect("valid payload must decode"); + assert_eq!(decoded, Some(expected)); + } + + /// A zero-count payload is a present-but-empty confirmation list, which is + /// not the same thing as a server that reports nothing at all. + #[test] + fn zero_count_body_decodes_to_present_empty_list() { + let bytes = SendMessagesResponse { + confirmations: vec![], + } + .to_bytes(); + let decoded = decode_send_confirmations(&bytes).expect("zero-count payload must decode"); + assert_eq!( + decoded, + Some(SendMessagesResponse { + confirmations: vec![] + }) + ); + } + + #[test] + fn trailing_bytes_are_rejected() { + let mut bytes = response().to_bytes().to_vec(); + bytes.push(0xFF); + assert!(matches!( + decode_send_confirmations(&bytes), + Err(IggyError::InvalidFormat) + )); + } + + #[test] + fn truncated_body_is_rejected() { + let bytes = response().to_bytes(); + for length in 1..bytes.len() { + assert!( + matches!( + decode_send_confirmations(&bytes[..length]), + Err(IggyError::InvalidFormat) + ), + "expected error for truncation at byte {length}" + ); + } + } +} diff --git a/core/common/src/traits/message_client.rs b/core/common/src/traits/message_client.rs index aa332d618..f69237fbf 100644 --- a/core/common/src/traits/message_client.rs +++ b/core/common/src/traits/message_client.rs @@ -17,6 +17,7 @@ use crate::{ Consumer, Identifier, IggyError, IggyMessage, Partitioning, PolledMessages, PollingStrategy, + SendMessagesResponse, }; use async_trait::async_trait; @@ -43,13 +44,18 @@ pub trait MessageClient { /// Send messages using specified partitioning strategy to the given stream and topic by unique IDs or names. /// /// Authentication is required, and the permission to send the messages. + /// + /// Returns the per-partition commit confirmations when the server reports + /// them. `None` means the send succeeded but the server sent no + /// confirmation payload: servers predating the confirmation reply answer + /// with an empty body, so absence is never an error. async fn send_messages( &self, stream_id: &Identifier, topic_id: &Identifier, partitioning: &Partitioning, messages: &mut [IggyMessage], - ) -> Result<(), IggyError>; + ) -> Result<Option<SendMessagesResponse>, IggyError>; /// Force flush of the `unsaved_messages` buffer to disk, optionally fsyncing the data. #[allow(clippy::too_many_arguments)] diff --git a/core/integration/tests/sdk/mod.rs b/core/integration/tests/sdk/mod.rs index 7d7248031..d554d37d2 100644 --- a/core/integration/tests/sdk/mod.rs +++ b/core/integration/tests/sdk/mod.rs @@ -29,3 +29,4 @@ mod producer; #[cfg(feature = "vsr")] mod protocol_version; mod raw; +mod send_confirmation; diff --git a/core/integration/tests/sdk/producer/background.rs b/core/integration/tests/sdk/producer/background.rs index c78d94357..8c5aba7ca 100644 --- a/core/integration/tests/sdk/producer/background.rs +++ b/core/integration/tests/sdk/producer/background.rs @@ -55,7 +55,11 @@ async fn background_send_receive_ok(harness: &TestHarness) { .background(BackgroundConfig::builder().build()) .build(); - producer.send(messages).await.unwrap(); + let confirmations = producer.send(messages).await.unwrap(); + assert!( + confirmations.is_empty(), + "a background producer returns before the send happens, so no confirmation can reach it" + ); sleep(Duration::from_millis(500)).await; producer.shutdown().await; diff --git a/core/integration/tests/sdk/send_confirmation.rs b/core/integration/tests/sdk/send_confirmation.rs new file mode 100644 index 000000000..dfd9965ab --- /dev/null +++ b/core/integration/tests/sdk/send_confirmation.rs @@ -0,0 +1,272 @@ +// 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. + +//! Commit confirmations for `SendMessages`: which partition a batch landed in +//! and at which offset. server-ng answers a committed send with a confirmation +//! payload; the legacy server answers with an empty body, which the SDK reports +//! as `None` rather than as a decode failure. + +use iggy::prelude::*; +use integration::iggy_harness; + +const STREAM_NAME: &str = "confirmation-stream"; +const TOPIC_NAME: &str = "confirmation-topic"; +const MESSAGES_COUNT: u32 = 10; +const PARTITIONS_COUNT: u32 = 3; + +// server-ng partition ids are 0-based (CreateTopic assigns them from 0). +#[cfg(feature = "vsr")] +const PARTITION_ID: u32 = 0; +/// Chunking for the direct producer: `CHUNKS * CHUNK_LENGTH` messages exceed +/// one request, so the send is split and every chunk confirms separately. +#[cfg(feature = "vsr")] +const CHUNK_LENGTH: u32 = 4; +#[cfg(feature = "vsr")] +const CHUNKS: u32 = 3; + +fn batch(count: u32) -> Vec<IggyMessage> { + (0..count) + .map(|i| { + IggyMessage::builder() + .id(u128::from(i + 1)) + .payload(format!("payload-{i}").into()) + .build() + .expect("message build") + }) + .collect() +} + +/// Returns the numeric ids the server assigned to the created stream and topic. +async fn create_stream_and_topic(client: &IggyClient, partitions_count: u32) -> (u32, u32) { + let stream = client + .create_stream(STREAM_NAME) + .await + .expect("create_stream"); + let topic = client + .create_topic( + &Identifier::named(STREAM_NAME).unwrap(), + TOPIC_NAME, + partitions_count, + CompressionAlgorithm::default(), + None, + IggyExpiry::NeverExpire, + MaxTopicSize::ServerDefault, + ) + .await + .expect("create_topic"); + (stream.id, topic.id) +} + +#[cfg(feature = "vsr")] +fn sole_confirmation(response: &SendMessagesResponse) -> &SendMessagesConfirmationResponse { + assert_eq!( + response.confirmations.len(), + 1, + "a send routed to one partition confirms exactly one partition" + ); + &response.confirmations[0] +} + +/// Each transport carries the reply body on its own path, so the full +/// confirmation shape is pinned on all three. +#[cfg(feature = "vsr")] +#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])] +async fn given_explicit_partition_when_sending_two_batches_should_confirm_advancing_base_offset( + harness: &TestHarness, +) { + let client = harness.new_client().await.unwrap(); + client + .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) + .await + .unwrap(); + + let (stream_id, topic_id) = create_stream_and_topic(&client, 1).await; + let stream = Identifier::named(STREAM_NAME).unwrap(); + let topic = Identifier::named(TOPIC_NAME).unwrap(); + let partitioning = Partitioning::partition_id(PARTITION_ID); + + let mut messages = batch(MESSAGES_COUNT); + let response = client + .send_messages(&stream, &topic, &partitioning, &mut messages) + .await + .expect("send_messages") + .expect("server-ng must confirm a committed send"); + let first = sole_confirmation(&response); + + assert_eq!(first.stream_id, stream_id, "confirmed stream id"); + assert_eq!(first.topic_id, topic_id, "confirmed topic id"); + assert_eq!(first.partition_id, PARTITION_ID, "confirmed partition id"); + assert_eq!(first.base_offset, 0, "the first batch starts at offset 0"); + + let mut messages = batch(MESSAGES_COUNT); + let response = client + .send_messages(&stream, &topic, &partitioning, &mut messages) + .await + .expect("send_messages") + .expect("server-ng must confirm a committed send"); + let second = sole_confirmation(&response); + + assert_eq!(second.partition_id, PARTITION_ID, "same partition"); + assert_eq!( + second.base_offset, + u64::from(MESSAGES_COUNT), + "the next batch starts where the previous one ended" + ); + + client.logout_user().await.unwrap(); +} + +#[cfg(feature = "vsr")] +#[iggy_harness] +async fn given_balanced_partitioning_when_sending_should_confirm_a_partition_of_the_topic( + harness: &TestHarness, +) { + let client = harness.new_client().await.unwrap(); + client + .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) + .await + .unwrap(); + + let (stream_id, topic_id) = create_stream_and_topic(&client, PARTITIONS_COUNT).await; + let mut messages = batch(MESSAGES_COUNT); + let response = client + .send_messages( + &Identifier::named(STREAM_NAME).unwrap(), + &Identifier::named(TOPIC_NAME).unwrap(), + &Partitioning::balanced(), + &mut messages, + ) + .await + .expect("send_messages") + .expect("server-ng must confirm a committed send"); + let confirmation = sole_confirmation(&response); + + assert_eq!(confirmation.stream_id, stream_id, "confirmed stream id"); + assert_eq!(confirmation.topic_id, topic_id, "confirmed topic id"); + assert!( + confirmation.partition_id < PARTITIONS_COUNT, + "balanced routing must land in one of the topic's {PARTITIONS_COUNT} partitions, got {}", + confirmation.partition_id + ); + assert_eq!( + confirmation.base_offset, 0, + "the chosen partition was empty before the send" + ); + + client.logout_user().await.unwrap(); +} + +#[cfg(feature = "vsr")] +#[iggy_harness] +async fn given_messages_key_partitioning_when_sending_should_confirm_a_partition_of_the_topic( + harness: &TestHarness, +) { + let client = harness.new_client().await.unwrap(); + client + .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) + .await + .unwrap(); + + create_stream_and_topic(&client, PARTITIONS_COUNT).await; + let mut messages = batch(MESSAGES_COUNT); + let response = client + .send_messages( + &Identifier::named(STREAM_NAME).unwrap(), + &Identifier::named(TOPIC_NAME).unwrap(), + &Partitioning::messages_key_str("confirmation-key").unwrap(), + &mut messages, + ) + .await + .expect("send_messages") + .expect("server-ng must confirm a committed send"); + let confirmation = sole_confirmation(&response); + + assert!( + confirmation.partition_id < PARTITIONS_COUNT, + "keyed routing must land in one of the topic's {PARTITIONS_COUNT} partitions, got {}", + confirmation.partition_id + ); + + client.logout_user().await.unwrap(); +} + +#[cfg(feature = "vsr")] +#[iggy_harness] +async fn given_direct_producer_when_send_splits_into_chunks_should_confirm_every_chunk( + harness: &TestHarness, +) { + let client = harness.new_client().await.unwrap(); + client + .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD) + .await + .unwrap(); + + let (stream_id, topic_id) = create_stream_and_topic(&client, 1).await; + let producer = client + .producer(STREAM_NAME, TOPIC_NAME) + .unwrap() + .partitioning(Partitioning::partition_id(PARTITION_ID)) + .direct(DirectConfig::builder().batch_length(CHUNK_LENGTH).build()) + .build(); + + let confirmations = producer + .send(batch(CHUNKS * CHUNK_LENGTH)) + .await + .expect("producer send"); + + assert_eq!( + confirmations.len(), + CHUNKS as usize, + "a direct producer confirms every chunk it split the send into" + ); + for (chunk, response) in confirmations.iter().enumerate() { + let confirmation = sole_confirmation(response); + assert_eq!(confirmation.stream_id, stream_id, "confirmed stream id"); + assert_eq!(confirmation.topic_id, topic_id, "confirmed topic id"); + assert_eq!(confirmation.partition_id, PARTITION_ID, "pinned partition"); + assert_eq!( + confirmation.base_offset, + chunk as u64 * u64::from(CHUNK_LENGTH), + "chunk {chunk} must start where the previous chunk ended" + ); + } + + client.logout_user().await.unwrap(); +} + +#[cfg(not(feature = "vsr"))] +#[iggy_harness] +async fn given_legacy_server_when_sending_should_report_no_confirmation(harness: &TestHarness) { + let client = harness.root_client().await.unwrap(); + + create_stream_and_topic(&client, PARTITIONS_COUNT).await; + let mut messages = batch(MESSAGES_COUNT); + let confirmation = client + .send_messages( + &Identifier::named(STREAM_NAME).unwrap(), + &Identifier::named(TOPIC_NAME).unwrap(), + &Partitioning::balanced(), + &mut messages, + ) + .await + .expect("send_messages"); + + assert!( + confirmation.is_none(), + "a server that replies with an empty body reports no confirmation" + ); +} diff --git a/core/integration/tests/server/scenarios/authentication_scenario.rs b/core/integration/tests/server/scenarios/authentication_scenario.rs index 2c4ede107..e39d6cd6a 100644 --- a/core/integration/tests/server/scenarios/authentication_scenario.rs +++ b/core/integration/tests/server/scenarios/authentication_scenario.rs @@ -265,6 +265,7 @@ async fn test_all_commands_require_auth(client: &IggyClient) { &mut msgs, ) .await + .map(|_| ()) } POLL_MESSAGES_CODE => client .poll_messages( diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 8d2c3a4bf..a2a6527d3 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -43,8 +43,11 @@ use iggy_binary_protocol::requests::consumer_offsets::{ DeleteConsumerOffset2Request, DeleteConsumerOffsetRequest, StoreConsumerOffset2Request, StoreConsumerOffsetRequest, }; +use iggy_binary_protocol::responses::messages::{ + SendMessagesConfirmationResponse, SendMessagesResponse, +}; use iggy_binary_protocol::{ - AckLevel, GenericHeader, Operation, PrepareHeader, WireDecode, WireIdentifier, + AckLevel, GenericHeader, Operation, PrepareHeader, WireDecode, WireEncode, WireIdentifier, }; use iggy_binary_protocol::{PrepareOkHeader, RequestHeader}; use iggy_common::{ @@ -646,7 +649,9 @@ where let reply = build_reply_from_request( &self.consensus, &request_header, - committed_reply_body(request_header.operation), + // Consumer-offset ops only: `SendMessages` never reaches the + // `NoAck` fast path, so there are no batch offsets to confirm. + committed_reply_body(request_header.operation, request_header.namespace, None), ); let reply_buffers = reply.into_generic().into_frozen(); if let Err(error) = self @@ -2163,7 +2168,11 @@ where if send_client_replies && !is_auto_commit_client(prepare_header.client) { let reply = build_reply_message( &prepare_header, - &committed_reply_body(prepare_header.operation), + &committed_reply_body( + prepare_header.operation, + prepare_header.namespace, + committed_visible_offsets.get(&prepare_header.op), + ), ); let reply_buffers = reply.into_generic().into_frozen(); emit_sim_event(SimEventKind::ClientReplyEmitted, &event); @@ -2323,6 +2332,7 @@ where } Ok(Some(CommittedBatchStats { + base_offset: batch.header.base_offset, end_offset: batch.header.base_offset + u64::from(message_count) - 1, message_count, size_bytes: batch.header.total_size() as u64, @@ -3324,24 +3334,61 @@ fn peek_operation(entry: &Frozen<4096>) -> Operation { .operation } -/// Success reply body for a committed partition op. Result-framed ops -/// (`Operation::is_result_framed`; on this plane the consumer-offset ops, -/// whose rejections ship typed errors) must carry an explicit empty result -/// section (`[count = 0]`) so the SDK's framed decode does not misread the -/// payload; every other partition op replies with an empty body. -const fn committed_reply_body(operation: Operation) -> bytes::Bytes { - if operation.is_result_framed() { - bytes::Bytes::from_static(&[0, 0, 0, 0]) - } else { - bytes::Bytes::new() +/// Success reply body for a committed partition op. +/// +/// `SendMessages` carries a [`SendMessagesResponse`] confirming where the batch +/// landed. It is not result-framed, so its body is the payload itself; a +/// `batch_stats` of `None` still ships a well-formed `count = 0` payload rather +/// than an empty body, which the SDK cannot decode. +/// +/// Result-framed ops (`Operation::is_result_framed`; on this plane the +/// consumer-offset ops, whose rejections ship typed errors) must carry an +/// explicit empty result section (`[count = 0]`) so the SDK's framed decode +/// does not misread the payload; every other partition op replies with an +/// empty body. +fn committed_reply_body( + operation: Operation, + namespace: u64, + batch_stats: Option<&CommittedBatchStats>, +) -> bytes::Bytes { + match operation { + Operation::SendMessages => send_messages_reply_body(namespace, batch_stats), + _ if operation.is_result_framed() => bytes::Bytes::from_static(&[0, 0, 0, 0]), + _ => bytes::Bytes::new(), + } +} + +/// One confirmation for the committed batch, or `count = 0` when its offsets +/// could not be resolved (undecodable journal entry, or an empty batch). +#[allow(clippy::cast_possible_truncation)] +fn send_messages_reply_body( + namespace: u64, + batch_stats: Option<&CommittedBatchStats>, +) -> bytes::Bytes { + let namespace = IggyNamespace::from_raw(namespace); + SendMessagesResponse { + confirmations: batch_stats + .map(|stats| SendMessagesConfirmationResponse { + // `IggyNamespace` packs the ids into 12/12/20 bits, so each + // component fits a `u32` by construction. + stream_id: namespace.stream_id() as u32, + topic_id: namespace.topic_id() as u32, + partition_id: namespace.partition_id() as u32, + base_offset: stats.base_offset, + }) + .into_iter() + .collect(), } + .to_bytes() } /// Committed-batch accounting surfaced at commit time so the aggregate stats /// (`messages_count`, `size_bytes`) advance with the visible offset rather than -/// waiting on the threshold-gated disk persist. +/// waiting on the threshold-gated disk persist, and so the `SendMessages` reply +/// can confirm where the batch landed. #[derive(Clone, Copy)] struct CommittedBatchStats { + base_offset: u64, end_offset: u64, message_count: u32, size_bytes: u64, @@ -4115,6 +4162,70 @@ mod tests { assert_eq!(partition.consensus().commit_min(), 0); } + + fn batch_stats(base_offset: u64, message_count: u32) -> CommittedBatchStats { + CommittedBatchStats { + base_offset, + end_offset: base_offset + u64::from(message_count) - 1, + message_count, + size_bytes: 128, + } + } + + #[test] + fn given_send_messages_when_offsets_resolved_should_confirm_base_offset() { + let namespace = IggyNamespace::new(3, 7, 5); + let stats = batch_stats(42, 3); + + let body = committed_reply_body(Operation::SendMessages, namespace.inner(), Some(&stats)); + let (response, consumed) = SendMessagesResponse::decode(&body).unwrap(); + + assert_eq!(consumed, body.len()); + assert_eq!( + response.confirmations, + vec![SendMessagesConfirmationResponse { + stream_id: 3, + topic_id: 7, + partition_id: 5, + base_offset: 42, + }] + ); + } + + #[test] + fn given_send_messages_when_offsets_unavailable_should_reply_zero_confirmations() { + let namespace = IggyNamespace::new(1, 1, 0); + + let body = committed_reply_body(Operation::SendMessages, namespace.inner(), None); + + assert_eq!(&body[..], &[0, 0, 0, 0]); + let (response, _) = SendMessagesResponse::decode(&body).unwrap(); + assert!(response.confirmations.is_empty()); + } + + #[test] + fn given_result_framed_operation_when_committed_should_reply_empty_result_section() { + let namespace = IggyNamespace::new(1, 1, 0); + + // Batch stats belong to a `SendMessages` prepare and never leak here. + assert_eq!( + &committed_reply_body( + Operation::StoreConsumerOffset2, + namespace.inner(), + Some(&batch_stats(9, 1)) + )[..], + &[0, 0, 0, 0] + ); + } + + #[test] + fn given_unframed_operation_when_committed_should_reply_empty_body() { + let namespace = IggyNamespace::new(1, 1, 0); + + assert!( + committed_reply_body(Operation::DeleteSegments, namespace.inner(), None).is_empty() + ); + } } #[cfg(test)] diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml index 5f8d91aa9..b83fb9d2a 100644 --- a/core/sdk/Cargo.toml +++ b/core/sdk/Cargo.toml @@ -53,6 +53,7 @@ reqwest-tracing = { workspace = true } rustls = { workspace = true } secrecy = { workspace = true } serde = { workspace = true } +serde_json = { workspace = true } tokio = { workspace = true } tokio-rustls = { workspace = true } tokio-tungstenite = { workspace = true } diff --git a/core/sdk/src/client_wrappers/binary_message_client.rs b/core/sdk/src/client_wrappers/binary_message_client.rs index 91fb303fe..9c8796953 100644 --- a/core/sdk/src/client_wrappers/binary_message_client.rs +++ b/core/sdk/src/client_wrappers/binary_message_client.rs @@ -20,6 +20,7 @@ use async_trait::async_trait; use iggy_common::MessageClient; use iggy_common::{ Consumer, Identifier, IggyError, IggyMessage, Partitioning, PolledMessages, PollingStrategy, + SendMessagesResponse, }; #[async_trait] @@ -109,7 +110,7 @@ impl MessageClient for ClientWrapper { topic_id: &Identifier, partitioning: &Partitioning, messages: &mut [IggyMessage], - ) -> Result<(), IggyError> { + ) -> Result<Option<SendMessagesResponse>, IggyError> { match self { ClientWrapper::Iggy(client) => { client diff --git a/core/sdk/src/clients/binary_message.rs b/core/sdk/src/clients/binary_message.rs index 7105e63c9..350a7b561 100644 --- a/core/sdk/src/clients/binary_message.rs +++ b/core/sdk/src/clients/binary_message.rs @@ -22,6 +22,7 @@ use iggy_common::MessageClient; use iggy_common::locking::IggyRwLockFn; use iggy_common::{ Consumer, Identifier, IggyError, IggyMessage, Partitioning, PolledMessages, PollingStrategy, + SendMessagesResponse, }; #[async_trait] @@ -78,7 +79,7 @@ impl MessageClient for IggyClient { topic_id: &Identifier, partitioning: &Partitioning, messages: &mut [IggyMessage], - ) -> Result<(), IggyError> { + ) -> Result<Option<SendMessagesResponse>, IggyError> { if messages.is_empty() { return Err(IggyError::InvalidMessagesCount); } diff --git a/core/sdk/src/clients/producer.rs b/core/sdk/src/clients/producer.rs index 452e361e9..a438d8fe5 100644 --- a/core/sdk/src/clients/producer.rs +++ b/core/sdk/src/clients/producer.rs @@ -28,6 +28,7 @@ use iggy_common::{Client, MessageClient, StreamClient, TopicClient}; use iggy_common::{ CompressionAlgorithm, DiagnosticEvent, EncryptorKind, IdKind, Identifier, IggyDuration, IggyError, IggyExpiry, IggyMessage, IggyTimestamp, MaxTopicSize, Partitioner, Partitioning, + SendMessagesResponse, }; use std::sync::Arc; use std::sync::atomic::Ordering; @@ -41,13 +42,16 @@ use mockall::automock; #[cfg_attr(test, automock)] pub trait ProducerCoreBackend: Send + Sync + 'static { + /// Sends `msgs`, returning one confirmation per chunk the server confirmed. + /// The vector is shorter than the chunk count when the server reports no + /// offsets, and empty when it never does. fn send_internal( &self, stream: &Identifier, topic: &Identifier, msgs: Vec<IggyMessage>, partitioning: Option<Arc<Partitioning>>, - ) -> impl Future<Output = Result<(), IggyError>> + Send; + ) -> impl Future<Output = Result<Vec<SendMessagesResponse>, IggyError>> + Send; } pub struct ProducerCore { @@ -183,7 +187,7 @@ impl ProducerCore { topic: &Identifier, partitioning: &Arc<Partitioning>, messages: &mut [IggyMessage], - ) -> Result<(), IggyError> { + ) -> Result<Option<SendMessagesResponse>, IggyError> { let client = self.client.read().await; let Some(max_retries) = self.send_retries_count else { @@ -249,7 +253,7 @@ impl ProducerCore { topic: &Identifier, partitioning: &Arc<Partitioning>, messages: &mut [IggyMessage], - ) -> Result<(), IggyError> { + ) -> Result<Option<SendMessagesResponse>, IggyError> { let mut retries = 0; let mut timer: Option<Interval> = None; @@ -258,7 +262,9 @@ impl ProducerCore { .send_messages(stream, topic, partitioning, messages) .await { - Ok(_) => return Ok(()), + // Only the attempt that finally succeeds yields a confirmation; + // failed attempts have none to report. + Ok(confirmation) => return Ok(confirmation), Err(error) => { retries += 1; if retries > max_retries { @@ -361,9 +367,9 @@ impl ProducerCoreBackend for ProducerCore { topic: &Identifier, mut msgs: Vec<IggyMessage>, partitioning: Option<Arc<Partitioning>>, - ) -> Result<(), IggyError> { + ) -> Result<Vec<SendMessagesResponse>, IggyError> { if msgs.is_empty() { - return Ok(()); + return Ok(Vec::new()); } if let Err(err) = self.encrypt_messages(&mut msgs) { @@ -391,30 +397,37 @@ impl ProducerCoreBackend for ProducerCore { cfg.batch_length as usize }; let mut index = 0; + let mut confirmations = Vec::with_capacity(msgs.len().div_ceil(max)); while index < msgs.len() { let end = (index + max).min(msgs.len()); let chunk = &mut msgs[index..end]; - if let Err(err) = self.try_send_messages(stream, topic, &part, chunk).await { - let failed_tail = msgs.split_off(index); - return Err(self.make_failed_error(err, failed_tail)); + let sent = self.try_send_messages(stream, topic, &part, chunk).await; + match sent { + // `None` (server reported no offsets) contributes nothing. + Ok(confirmation) => confirmations.extend(confirmation), + Err(err) => { + let failed_tail = msgs.split_off(index); + return Err(self.make_failed_error(err, failed_tail)); + } } self.last_sent_at .store(IggyTimestamp::now().into(), ORDERING); index = end; } + Ok(confirmations) } // background send on _ => { - self.try_send_messages(stream, topic, &part, &mut msgs) + let confirmation = self + .try_send_messages(stream, topic, &part, &mut msgs) .await .map_err(|err| self.make_failed_error(err, msgs))?; self.last_sent_at .store(IggyTimestamp::now().into(), ORDERING); + Ok(confirmation.into_iter().collect()) } } - - Ok(()) } } @@ -496,17 +509,35 @@ impl IggyProducer { self.core.init().await } - pub async fn send(&self, messages: Vec<IggyMessage>) -> Result<(), IggyError> { + /// Sends `messages`, returning one commit confirmation per chunk the send + /// was split into, in chunk order. + /// + /// A chunk contributes an entry only when the server reported offsets for + /// it, so the vector is shorter than the chunk count against a server that + /// answers without a confirmation payload, and empty against one that never + /// does. A retried chunk contributes only the confirmation of the attempt + /// that finally succeeded. + /// + /// A `background` producer always returns an empty vector: it hands the + /// messages to a dispatcher and returns before the send happens, so no + /// confirmation can reach this caller. + pub async fn send( + &self, + messages: Vec<IggyMessage>, + ) -> Result<Vec<SendMessagesResponse>, IggyError> { if messages.is_empty() { trace!("No messages to send."); - return Ok(()); + return Ok(Vec::new()); } let stream_id = self.core.stream_id.clone(); let topic_id = self.core.topic_id.clone(); match &self.dispatcher { - Some(disp) => disp.dispatch(messages, stream_id, topic_id, None).await, + Some(disp) => disp + .dispatch(messages, stream_id, topic_id, None) + .await + .map(|()| Vec::new()), None => { self.core .send_internal(&stream_id, &topic_id, messages, None) @@ -515,28 +546,33 @@ impl IggyProducer { } } - pub async fn send_one(&self, message: IggyMessage) -> Result<(), IggyError> { + /// See [`IggyProducer::send`] for the confirmation semantics. + pub async fn send_one( + &self, + message: IggyMessage, + ) -> Result<Vec<SendMessagesResponse>, IggyError> { self.send(vec![message]).await } + /// See [`IggyProducer::send`] for the confirmation semantics. pub async fn send_with_partitioning( &self, messages: Vec<IggyMessage>, partitioning: Option<Arc<Partitioning>>, - ) -> Result<(), IggyError> { + ) -> Result<Vec<SendMessagesResponse>, IggyError> { if messages.is_empty() { trace!("No messages to send."); - return Ok(()); + return Ok(Vec::new()); } let stream_id = self.core.stream_id.clone(); let topic_id = self.core.topic_id.clone(); match &self.dispatcher { - Some(disp) => { - disp.dispatch(messages, stream_id, topic_id, partitioning) - .await - } + Some(disp) => disp + .dispatch(messages, stream_id, topic_id, partitioning) + .await + .map(|()| Vec::new()), None => { self.core .send_internal(&stream_id, &topic_id, messages, partitioning) @@ -545,20 +581,24 @@ impl IggyProducer { } } + /// See [`IggyProducer::send`] for the confirmation semantics. pub async fn send_to( &self, stream: Arc<Identifier>, topic: Arc<Identifier>, messages: Vec<IggyMessage>, partitioning: Option<Arc<Partitioning>>, - ) -> Result<(), IggyError> { + ) -> Result<Vec<SendMessagesResponse>, IggyError> { if messages.is_empty() { trace!("No messages to send."); - return Ok(()); + return Ok(Vec::new()); } match &self.dispatcher { - Some(disp) => disp.dispatch(messages, stream, topic, partitioning).await, + Some(disp) => disp + .dispatch(messages, stream, topic, partitioning) + .await + .map(|()| Vec::new()), None => { self.core .send_internal(&stream, &topic, messages, partitioning) diff --git a/core/sdk/src/clients/producer_dispatcher.rs b/core/sdk/src/clients/producer_dispatcher.rs index 38950dc0f..4205defc8 100644 --- a/core/sdk/src/clients/producer_dispatcher.rs +++ b/core/sdk/src/clients/producer_dispatcher.rs @@ -229,7 +229,7 @@ mod tests { let mut mock = MockProducerCoreBackend::new(); mock.expect_send_internal() .times(1) - .returning(|_, _, _, _| Box::pin(async { Ok(()) })); + .returning(|_, _, _, _| Box::pin(async { Ok(Vec::new()) })); let msg = dummy_message(5); let config = BackgroundConfig::builder() @@ -305,7 +305,7 @@ mod tests { let mut mock = MockProducerCoreBackend::new(); mock.expect_send_internal() .times(1) - .returning(|_, _, _, _| Box::pin(async { Ok(()) })); + .returning(|_, _, _, _| Box::pin(async { Ok(Vec::new()) })); let msg = ShardMessage { stream: dummy_identifier(), diff --git a/core/sdk/src/clients/producer_sharding.rs b/core/sdk/src/clients/producer_sharding.rs index 5a8ae0ad2..ee9afbc3f 100644 --- a/core/sdk/src/clients/producer_sharding.rs +++ b/core/sdk/src/clients/producer_sharding.rs @@ -315,7 +315,7 @@ mod tests { let mut mock = MockProducerCoreBackend::new(); mock.expect_send_internal() .times(10) - .returning(|_, _, _, _| Box::pin(async { Ok(()) })); + .returning(|_, _, _, _| Box::pin(async { Ok(Vec::new()) })); let bb = BackgroundConfig::builder() .batch_length(10) @@ -357,7 +357,7 @@ mod tests { let mut mock = MockProducerCoreBackend::new(); mock.expect_send_internal() .times(1) - .returning(|_, _, _, _| Box::pin(async { Ok(()) })); + .returning(|_, _, _, _| Box::pin(async { Ok(Vec::new()) })); let bb = BackgroundConfig::builder() .batch_length(1000) @@ -401,7 +401,7 @@ mod tests { let mut mock = MockProducerCoreBackend::new(); mock.expect_send_internal() .times(1) - .returning(|_, _, _, _| Box::pin(async { Ok(()) })); + .returning(|_, _, _, _| Box::pin(async { Ok(Vec::new()) })); let bb = BackgroundConfig::builder() .batch_length(10) diff --git a/core/sdk/src/http/messages.rs b/core/sdk/src/http/messages.rs index c6ee8869c..8bb85d7ca 100644 --- a/core/sdk/src/http/messages.rs +++ b/core/sdk/src/http/messages.rs @@ -19,12 +19,13 @@ use crate::http::http_client::HttpClient; use crate::http::http_transport::HttpTransport; use crate::prelude::{ Consumer, Identifier, IggyError, IggyMessage, Partitioning, PollMessages, PolledMessages, - PollingStrategy, SendMessages, + PollingStrategy, SendMessages, SendMessagesConfirmationResponse, SendMessagesResponse, }; use async_trait::async_trait; use iggy_common::IggyMessagesBatch; use iggy_common::MessageClient; use iggy_common::flush_unsaved_buffer::FlushUnsavedBuffer; +use serde::Deserialize; #[async_trait] impl MessageClient for HttpClient { @@ -65,20 +66,32 @@ impl MessageClient for HttpClient { topic_id: &Identifier, partitioning: &Partitioning, messages: &mut [IggyMessage], - ) -> Result<(), IggyError> { + ) -> Result<Option<SendMessagesResponse>, IggyError> { let batch = IggyMessagesBatch::from(&*messages); - self.post( - &get_path(&stream_id.as_cow_str(), &topic_id.as_cow_str()), - &SendMessages { - metadata_length: 0, // this field is used only for TCP/QUIC - stream_id: stream_id.clone(), - topic_id: topic_id.clone(), - partitioning: partitioning.clone(), - batch, - }, - ) - .await?; - Ok(()) + let response = self + .post( + &get_path(&stream_id.as_cow_str(), &topic_id.as_cow_str()), + &SendMessages { + metadata_length: 0, // this field is used only for TCP/QUIC + stream_id: stream_id.clone(), + topic_id: topic_id.clone(), + partitioning: partitioning.clone(), + batch, + }, + ) + .await?; + let body = response + .bytes() + .await + .map_err(|_| IggyError::InvalidBytesResponse)?; + // Servers predating the confirmation reply answer a successful send with + // no content, so an absent body is not a failure. + if body.is_empty() { + return Ok(None); + } + let confirmations: SendMessagesConfirmationsDto = + serde_json::from_slice(&body).map_err(|_| IggyError::InvalidJsonResponse)?; + Ok(Some(SendMessagesResponse::from(confirmations))) } async fn flush_unsaved_buffer( @@ -106,6 +119,38 @@ impl MessageClient for HttpClient { } } +/// JSON shape of a successful `SendMessages` reply. Mirrors the binary +/// `SendMessagesResponse`, which carries no `serde` derives because wire types +/// stay codec-only. +#[derive(Deserialize)] +struct SendMessagesConfirmationsDto { + confirmations: Vec<SendMessagesConfirmationDto>, +} + +#[derive(Deserialize)] +struct SendMessagesConfirmationDto { + stream_id: u32, + topic_id: u32, + partition_id: u32, + base_offset: u64, +} + +impl From<SendMessagesConfirmationsDto> for SendMessagesResponse { + fn from(dto: SendMessagesConfirmationsDto) -> Self { + let confirmations = dto + .confirmations + .into_iter() + .map(|confirmation| SendMessagesConfirmationResponse { + stream_id: confirmation.stream_id, + topic_id: confirmation.topic_id, + partition_id: confirmation.partition_id, + base_offset: confirmation.base_offset, + }) + .collect(); + Self { confirmations } + } +} + fn get_path(stream_id: &str, topic_id: &str) -> String { format!("streams/{stream_id}/topics/{topic_id}/messages") } @@ -118,3 +163,52 @@ fn get_path_flush_unsaved_buffer( ) -> String { format!("streams/{stream_id}/topics/{topic_id}/messages/flush/{partition_id}/fsync={fsync}") } + +#[cfg(test)] +mod tests { + use super::{SendMessagesConfirmationsDto, SendMessagesResponse}; + use crate::prelude::SendMessagesConfirmationResponse; + + fn parse(json: &str) -> SendMessagesResponse { + let dto: SendMessagesConfirmationsDto = + serde_json::from_str(json).expect("contract sample must parse"); + SendMessagesResponse::from(dto) + } + + #[test] + fn confirmation_converts_all_fields() { + let response = parse( + r#"{"confirmations":[{"stream_id":1,"topic_id":2,"partition_id":3,"base_offset":42}]}"#, + ); + assert_eq!( + response.confirmations, + vec![SendMessagesConfirmationResponse { + stream_id: 1, + topic_id: 2, + partition_id: 3, + base_offset: 42, + }] + ); + } + + #[test] + fn empty_list_converts_to_empty_confirmations() { + let response = parse(r#"{"confirmations":[]}"#); + assert!(response.confirmations.is_empty()); + } + + #[test] + fn preserves_order_of_multiple_confirmations() { + let response = parse( + r#"{"confirmations":[ + {"stream_id":1,"topic_id":2,"partition_id":7,"base_offset":10}, + {"stream_id":1,"topic_id":2,"partition_id":3,"base_offset":20}]}"#, + ); + let partitions: Vec<u32> = response + .confirmations + .iter() + .map(|confirmation| confirmation.partition_id) + .collect(); + assert_eq!(partitions, vec![7, 3]); + } +} diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs index 34f8d1484..6ae9f05c3 100644 --- a/core/sdk/src/prelude.rs +++ b/core/sdk/src/prelude.rs @@ -58,10 +58,11 @@ pub use iggy_common::{ IggyTimestamp, MaxTopicSize, Partition, Partitioner, Partitioning, Permissions, PersonalAccessTokenExpiry, PollMessages, PolledMessages, PollingKind, PollingStrategy, QuicClientConfig, QuicClientConfigBuilder, QuicClientReconnectionConfig, SendMessages, - Sizeable, SnapshotCompression, Stats, Stream, StreamDetails, StreamPermissions, - SystemSnapshotType, TcpClientConfig, TcpClientConfigBuilder, TcpClientReconnectionConfig, - Topic, TopicDetails, TopicPermissions, TransportEndpoints, TransportProtocol, UserId, UserInfo, - UserInfoDetails, UserStatus, Validatable, WebSocketClientConfig, WebSocketClientConfigBuilder, + SendMessagesConfirmationResponse, SendMessagesResponse, Sizeable, SnapshotCompression, Stats, + Stream, StreamDetails, StreamPermissions, SystemSnapshotType, TcpClientConfig, + TcpClientConfigBuilder, TcpClientReconnectionConfig, Topic, TopicDetails, TopicPermissions, + TransportEndpoints, TransportProtocol, UserId, UserInfo, UserInfoDetails, UserStatus, + Validatable, WebSocketClientConfig, WebSocketClientConfigBuilder, WebSocketClientReconnectionConfig, defaults, locking, }; pub use iggy_common::{ diff --git a/core/server-ng/src/http/handlers.rs b/core/server-ng/src/http/handlers.rs index f97f3a53b..c8df70e4a 100644 --- a/core/server-ng/src/http/handlers.rs +++ b/core/server-ng/src/http/handlers.rs @@ -63,6 +63,7 @@ use iggy_binary_protocol::responses::clients::get_client::ClientDetailsResponse; use iggy_binary_protocol::responses::clients::get_clients::GetClientsResponse; use iggy_binary_protocol::responses::consumer_groups::get_consumer_group::ConsumerGroupDetailsResponse; use iggy_binary_protocol::responses::consumer_groups::get_consumer_groups::GetConsumerGroupsResponse; +use iggy_binary_protocol::responses::messages::SendMessagesResponse; use iggy_binary_protocol::responses::personal_access_tokens::GetPersonalAccessTokensResponse; use iggy_binary_protocol::responses::streams::get_stream::GetStreamResponse; use iggy_binary_protocol::responses::streams::get_streams::GetStreamsResponse; @@ -105,7 +106,7 @@ use metadata::impls::metadata::StreamsFrontend; use metadata::permissioner::Permissioner; use secrecy::ExposeSecret; use send_wrapper::SendWrapper; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use shard::{PartitionRead, PartitionReadReply}; use crate::auth::{verify_login_credentials, verify_pat_credentials}; @@ -121,7 +122,7 @@ use crate::http::reads::{ }; use crate::http::reply::{ committed_payload, decode_consumer_group_details, decode_raw_pat_token, decode_stream_details, - decode_topic_details, decode_user_details, login_error_to_iggy, + decode_topic_details, decode_user_details, login_error_to_iggy, send_confirmations, }; use crate::http::state::{HttpInner, HttpState}; use crate::http::submit::{ @@ -1116,6 +1117,41 @@ pub(in crate::http) async fn get_consumer_offset( } } +/// `POST .../messages` response body: one entry per partition the batch landed +/// in. A list, not a single confirmation, so a future multi-partition produce +/// needs no shape change. Local DTOs because the wire types carry no serde by +/// design. +#[derive(Debug, Serialize)] +struct SendMessagesConfirmations { + confirmations: Vec<PartitionConfirmation>, +} + +/// One partition's commit confirmation. +#[derive(Debug, Serialize)] +struct PartitionConfirmation { + stream_id: u32, + topic_id: u32, + partition_id: u32, + base_offset: u64, +} + +impl From<SendMessagesResponse> for SendMessagesConfirmations { + fn from(response: SendMessagesResponse) -> Self { + Self { + confirmations: response + .confirmations + .into_iter() + .map(|confirmation| PartitionConfirmation { + stream_id: confirmation.stream_id, + topic_id: confirmation.topic_id, + partition_id: confirmation.partition_id, + base_offset: confirmation.base_offset, + }) + .collect(), + } + } +} + /// `POST /streams/{stream_id}/topics/{topic_id}/messages`: produce a batch of /// messages to a topic. The JSON body is the same `SendMessages` shape the /// legacy server accepts (partitioning + base64 messages); stream and topic @@ -1126,8 +1162,9 @@ pub(in crate::http) async fn get_consumer_offset( /// on one credential are legal), and the committed reply comes back through /// the session's in-process reply slot rather than a submit return value. /// The default answers 201 + `Iggy-Durability: replicated-memory` only -/// after the quorum commit; `?ack=none` answers 202 + `Iggy-Durability: -/// none` immediately after dispatch. +/// after the quorum commit, with the commit's per-partition confirmations as +/// the body; `?ack=none` answers 202 + `Iggy-Durability: none` immediately +/// after dispatch and can carry no confirmation, having awaited none. pub(in crate::http) async fn send_messages( State(state): State<HttpState>, identity: Authenticated, @@ -1154,21 +1191,28 @@ pub(in crate::http) async fn send_messages( .map_err(PartitionWriteError::Rejected)?; match query.ack { ProduceAck::Replicated => { - SendWrapper::new(partition_write_replicated( + let reply = SendWrapper::new(partition_write_replicated( &state, &identity.session, Operation::SendMessages, &body, )) .await?; - Ok(( - StatusCode::CREATED, - [( - DURABILITY_HEADER, - HeaderValue::from_static(DURABILITY_REPLICATED_MEMORY), - )], - ) - .into_response()) + let durability = [( + DURABILITY_HEADER, + HeaderValue::from_static(DURABILITY_REPLICATED_MEMORY), + )]; + // An unreadable confirmation still answers 201: the batch committed, + // only its offsets did not survive the reply. + match send_confirmations(&reply) { + Some(response) => Ok(( + StatusCode::CREATED, + durability, + Json(SendMessagesConfirmations::from(response)), + ) + .into_response()), + None => Ok((StatusCode::CREATED, durability).into_response()), + } } ProduceAck::None => { SendWrapper::new(produce_unacked(&state, &identity.session, &body)).await?; @@ -1579,3 +1623,42 @@ fn issue_identity(inner: &HttpInner, user_id: u32) -> Result<Json<IdentityInfo>, }), })) } + +#[cfg(test)] +mod tests { + use super::*; + + use iggy_binary_protocol::responses::messages::SendMessagesConfirmationResponse; + + /// Pins the produce response contract the SDKs decode: `snake_case` field + /// names and a numeric `base_offset`. + #[test] + fn confirmation_renders_the_pinned_json_shape() { + let response = SendMessagesResponse { + confirmations: vec![SendMessagesConfirmationResponse { + stream_id: 3, + topic_id: 5, + partition_id: 7, + base_offset: 41, + }], + }; + let json = serde_json::to_string(&SendMessagesConfirmations::from(response)) + .expect("confirmations serialize"); + assert_eq!( + json, + r#"{"confirmations":[{"stream_id":3,"topic_id":5,"partition_id":7,"base_offset":41}]}"# + ); + } + + /// A commit that reports no offsets still renders the envelope, so the + /// field is always present for a caller that indexes into it. + #[test] + fn empty_confirmations_render_an_empty_list() { + let response = SendMessagesResponse { + confirmations: Vec::new(), + }; + let json = serde_json::to_string(&SendMessagesConfirmations::from(response)) + .expect("confirmations serialize"); + assert_eq!(json, r#"{"confirmations":[]}"#); + } +} diff --git a/core/server-ng/src/http/reply.rs b/core/server-ng/src/http/reply.rs index a9708983a..cccbfd295 100644 --- a/core/server-ng/src/http/reply.rs +++ b/core/server-ng/src/http/reply.rs @@ -23,6 +23,7 @@ use iggy_binary_protocol::consensus::{ Command2, EvictionHeader, HEADER_SIZE, result_code, result_section_len, }; use iggy_binary_protocol::responses::consumer_groups::get_consumer_group::ConsumerGroupDetailsResponse; +use iggy_binary_protocol::responses::messages::SendMessagesResponse; use iggy_binary_protocol::responses::personal_access_tokens::RawPersonalAccessTokenResponse; use iggy_binary_protocol::responses::streams::get_stream::GetStreamResponse; use iggy_binary_protocol::responses::topics::get_topic::GetTopicResponse; @@ -34,21 +35,26 @@ use iggy_common::{ }; use message_bus::BusMessage; use server_common::Message; +use tracing::warn; use crate::http::error::{PartitionWriteError, WriteError}; use crate::login_register::LoginRegisterError; /// Discriminate a partition write reply. Partition replies carry no result -/// section (success and denial are both empty-bodied), so the discriminators -/// live in the header. `status` is read first: a nonzero value is the typed -/// pre-commit denial (the partition primary's delete-of-missing-offset -/// rejection, or a dispatch-time authorization denial) and renders through -/// the legacy `IggyError -> status` map. With status 0, the reply's `op` -/// splits the rest: a committed reply is built from its prepare header, whose -/// op (the partition group's commit number) is always >= 1, while the -/// pre-dispatch gate failures reply through `build_empty_reply` with 0 in -/// that field. The gate reply cannot name which entity was missing, hence the -/// generic legacy 404 body. +/// section - a denial is empty-bodied and a committed body, where there is one, +/// is the bare typed payload - so the discriminators live in the header. +/// `status` is read first: a nonzero value is the typed pre-commit denial (the +/// partition primary's delete-of-missing-offset rejection, or a dispatch-time +/// authorization denial) and renders through the legacy `IggyError -> status` +/// map. With status 0, the reply's `op` splits the rest: a committed reply is +/// built from its prepare header, whose op (the partition group's commit +/// number) is always >= 1, while the pre-dispatch gate failures reply through +/// `build_empty_reply` with 0 in that field. The gate reply cannot name which +/// entity was missing, hence the generic legacy 404 body. +/// +/// Grading only. The committed body is read separately (see +/// [`send_confirmations`]) because it is operation-specific: consumer-offset +/// writes commit empty, a produce commits its confirmations. pub(in crate::http) fn classify_partition_reply( reply: &BusMessage, ) -> Result<(), PartitionWriteError> { @@ -71,6 +77,43 @@ pub(in crate::http) fn classify_partition_reply( Ok(()) } +/// The per-partition commit confirmations carried by a graded `SendMessages` +/// reply, or `None` when the reply has no readable confirmation. +/// +/// `None` is a normal outcome, never a failure: a peer whose partition plane +/// does not stamp the payload commits with an empty body, and a body this build +/// cannot decode is a shape mismatch. Neither unmakes the commit, so both fall +/// back to the payload-less success response rather than failing a write that +/// already landed. +pub(in crate::http) fn send_confirmations(reply: &BusMessage) -> Option<SendMessagesResponse> { + let body = partition_reply_body(reply); + if body.is_empty() { + return None; + } + match SendMessagesResponse::decode_from(body) { + Ok(confirmations) => Some(confirmations), + Err(error) => { + warn!( + ?error, + "server-ng HTTP: undecodable send_messages commit confirmation" + ); + None + } + } +} + +/// A partition reply's body past the header, bounded by the header's `size` +/// rather than by the buffer length: `size` is the frame's authoritative +/// extent, and the typed decoders reject trailing bytes. +fn partition_reply_body(reply: &BusMessage) -> &[u8] { + let size = reply + .as_slice() + .get(..HEADER_SIZE) + .and_then(|bytes| bytemuck::checked::try_from_bytes::<ReplyHeader>(bytes).ok()) + .map_or(0, |header| header.size as usize); + reply.as_slice().get(HEADER_SIZE..size).unwrap_or_default() +} + /// Classify a committed reply's leading result section and return the typed /// payload slice on success. Mirrors the SDK's `split_metadata_result`: /// `Some(0)` is success and the payload follows the result section; a nonzero @@ -216,6 +259,8 @@ mod tests { use bytes::Bytes; use iggy_binary_protocol::Operation; use iggy_binary_protocol::PrepareHeader; + use iggy_binary_protocol::WireEncode; + use iggy_binary_protocol::responses::messages::SendMessagesConfirmationResponse; use crate::responses::{ NonReplicatedResponse, build_deny_reply, build_empty_reply, build_reply_from_bytes, @@ -429,6 +474,64 @@ mod tests { )); } + fn send_reply(body: &Bytes) -> BusMessage { + let prepare = PrepareHeader { + command: Command2::Prepare, + operation: Operation::SendMessages, + client: 42, + op: 1, + request: 1, + ..Default::default() + }; + frozen(consensus::build_reply_message(&prepare, body)) + } + + #[test] + fn committed_send_reply_yields_its_confirmations() { + let response = SendMessagesResponse { + confirmations: vec![SendMessagesConfirmationResponse { + stream_id: 3, + topic_id: 5, + partition_id: 7, + base_offset: 41, + }], + }; + assert_eq!( + send_confirmations(&send_reply(&response.to_bytes())), + Some(response) + ); + } + + /// A zero-count body is a committed batch that reports no offsets, which is + /// distinct from the empty body below: it decodes, so the response carries + /// an empty confirmation list rather than falling back to no body at all. + #[test] + fn zero_count_commit_body_yields_empty_confirmations() { + let empty = SendMessagesResponse { + confirmations: Vec::new(), + }; + assert_eq!( + send_confirmations(&send_reply(&empty.to_bytes())), + Some(empty) + ); + } + + /// A peer whose partition plane does not stamp the payload commits with an + /// empty body. The write still landed, so this is `None`, not an error, and + /// the handler answers its payload-less 201. + #[test] + fn empty_commit_body_yields_no_confirmations() { + assert_eq!(send_confirmations(&send_reply(&Bytes::new())), None); + } + + /// Same fallback for a body this build cannot read: a shape mismatch must + /// not fail a produce that already committed. + #[test] + fn undecodable_commit_body_yields_no_confirmations() { + let truncated = Bytes::from_static(&[1, 0, 0, 0, 9]); + assert_eq!(send_confirmations(&send_reply(&truncated)), None); + } + /// The partition primary's typed pre-commit deny (delete of a missing /// consumer offset) rides `ReplyHeader.status` and must classify as the /// mapped `IggyError`, not as the generic op-0 not-found. diff --git a/core/server-ng/src/http/submit.rs b/core/server-ng/src/http/submit.rs index 82a715bac..a5122d1b6 100644 --- a/core/server-ng/src/http/submit.rs +++ b/core/server-ng/src/http/submit.rs @@ -27,6 +27,7 @@ use futures::channel::oneshot; use iggy_binary_protocol::consensus::Command2; use iggy_binary_protocol::{GenericHeader, Operation, RequestHeader}; use iggy_common::IggyError; +use message_bus::BusMessage; use server_common::Message; use tracing::warn; @@ -341,12 +342,15 @@ pub(in crate::http) async fn logout_session(state: &HttpInner, session: &Rc<Http /// which fires an installed slot, so one slot catches every exit. The slot /// guard borrows the registry, which is why this whole future runs inside /// the caller's `SendWrapper` on shard 0. +/// +/// Hands back the graded reply frame so a caller that renders a committed +/// payload can read it; the offset writes answer 204 and drop it. pub(in crate::http) async fn partition_write_replicated( state: &HttpInner, session: &HttpSession, operation: Operation, body: &[u8], -) -> Result<(), PartitionWriteError> { +) -> Result<BusMessage, PartitionWriteError> { // Admission sits here rather than before body decode: axum's extractors // already buffered and deserialized the body (bounded by the router-wide // `DefaultBodyLimit`) before the handler ran, so the caps gate what is @@ -390,7 +394,7 @@ pub(in crate::http) async fn partition_write_replicated( // reply after a timeout sheds at the bus instead of leaking a waiter. drop(guard); match outcome { - Ok(Ok(reply)) => classify_partition_reply(&reply), + Ok(Ok(reply)) => classify_partition_reply(&reply).map(|()| reply), // Cancelled (reply target torn down by session eviction mid-wait) or // elapsed: same caller contract either way - outcome unknown, 504. Ok(Err(_)) | Err(_) => Err(PartitionWriteError::Timeout(operation)),
