You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2026/08/05 16:44:42 UTC
(iggy) branch master updated: feat(server-ng): confirm committed sends with partition and offset (#3781)
This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new fcef8c01c feat(server-ng): confirm committed sends with partition and offset (#3781)
fcef8c01c is described below
commit fcef8c01c28aaf9cbe9295cc10e5909bb7e9cc9f
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Wed Aug 5 18:44:37 2026 +0200
feat(server-ng): confirm committed sends with partition and offset (#3781)
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 always returns the confirmation struct. The legacy server
reports none (empty reply body); until core/server is retired that
case degrades to a zeroed confirmation instead of an error, kept in
one fallback marked for removal. 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.
The PHP, Python and C++ bindings expose the confirmations as well.
All three are FFI over the Rust SDK, so the changed signature
reached them directly and PHP no longer compiled at all. Each wraps
the reply in a language-native type mirroring the Rust shape, a
response holding one confirmation per partition, so a later
multi-partition produce needs no binding change either. The
temporary zeroed-confirmation shape is marked in each binding
alongside the fallback it stands in for.
---
.github/actions/cpp-bazel/pre-merge/action.yml | 8 +-
.github/actions/php/pre-merge/action.yml | 1 +
.../actions/python-maturin/pre-merge/action.yml | 3 +
Cargo.lock | 51 ++--
Cargo.toml | 8 +-
bdd/python/uv.lock | 2 +-
core/ai/mcp/src/service/mod.rs | 5 +-
core/bench/Cargo.toml | 2 +-
core/binary_protocol/Cargo.toml | 2 +-
core/binary_protocol/src/responses/messages/mod.rs | 2 +
.../src/responses/messages/send_messages.rs | 284 +++++++++++++++++++++
core/cli/Cargo.toml | 2 +-
core/common/Cargo.toml | 2 +-
core/common/src/error/iggy_error.rs | 5 +-
core/common/src/http/messages/send_messages.rs | 58 ++++-
core/common/src/lib.rs | 4 +
core/common/src/traits/binary_impls/messages.rs | 137 +++++++++-
core/common/src/traits/binary_impls/mod.rs | 2 +
core/common/src/traits/message_client.rs | 15 +-
core/common/src/traits/mod.rs | 2 +
core/common/src/types/message/mod.rs | 4 +-
core/connectors/runtime/Cargo.toml | 2 +-
core/connectors/sinks/clickhouse_sink/Cargo.toml | 2 +-
core/connectors/sinks/delta_sink/Cargo.toml | 2 +-
core/connectors/sinks/doris_sink/Cargo.toml | 2 +-
.../connectors/sinks/elasticsearch_sink/Cargo.toml | 2 +-
core/connectors/sinks/http_sink/Cargo.toml | 2 +-
core/connectors/sinks/iceberg_sink/Cargo.toml | 2 +-
core/connectors/sinks/influxdb_sink/Cargo.toml | 2 +-
core/connectors/sinks/meilisearch_sink/Cargo.toml | 2 +-
core/connectors/sinks/mongodb_sink/Cargo.toml | 2 +-
core/connectors/sinks/postgres_sink/Cargo.toml | 2 +-
core/connectors/sinks/quickwit_sink/Cargo.toml | 2 +-
core/connectors/sinks/s3_sink/Cargo.toml | 2 +-
core/connectors/sinks/stdout_sink/Cargo.toml | 2 +-
core/connectors/sinks/surrealdb_sink/Cargo.toml | 2 +-
.../sources/elasticsearch_source/Cargo.toml | 2 +-
core/connectors/sources/influxdb_source/Cargo.toml | 2 +-
core/connectors/sources/postgres_source/Cargo.toml | 2 +-
core/connectors/sources/random_source/Cargo.toml | 2 +-
core/integration/tests/sdk/mod.rs | 1 +
core/integration/tests/sdk/producer/background.rs | 6 +-
core/integration/tests/sdk/send_confirmation.rs | 268 +++++++++++++++++++
.../server/scenarios/authentication_scenario.rs | 1 +
core/partitions/src/iggy_partition.rs | 264 +++++++++++++++----
core/sdk/Cargo.toml | 3 +-
.../src/client_wrappers/binary_message_client.rs | 3 +-
core/sdk/src/clients/binary_message.rs | 3 +-
core/sdk/src/clients/producer.rs | 167 +++++++++---
core/sdk/src/clients/producer_dispatcher.rs | 7 +-
core/sdk/src/clients/producer_error_callback.rs | 11 +-
core/sdk/src/clients/producer_sharding.rs | 11 +-
core/sdk/src/http/messages.rs | 94 ++++++-
core/sdk/src/prelude.rs | 9 +-
core/server-ng/Cargo.toml | 2 +-
core/server-ng/src/dispatch.rs | 36 ++-
core/server-ng/src/dispatch/authz.rs | 11 +-
core/server-ng/src/http/error.rs | 17 +-
core/server-ng/src/http/handlers.rs | 78 +++++-
core/server-ng/src/http/reply.rs | 146 ++++++++++-
core/server-ng/src/http/submit.rs | 11 +-
examples/python/uv.lock | 2 +-
foreign/cpp/src/client.rs | 7 +-
foreign/cpp/src/lib.rs | 27 +-
foreign/cpp/src/type_conversion.rs | 30 ++-
foreign/cpp/tests/e2e/message.cpp | 9 +-
foreign/node/package-lock.json | 4 +-
foreign/node/package.json | 2 +-
foreign/node/src/client/client.utils.test.ts | 23 +-
foreign/node/src/client/client.utils.ts | 3 -
foreign/node/src/e2e/tcp.send-message.e2e.ts | 27 +-
foreign/node/src/index.ts | 2 +
.../src/wire/message/send-messages.command.test.ts | 110 +++++++-
.../node/src/wire/message/send-messages.command.ts | 81 +++++-
foreign/node/src/wire/vsr/register.ts | 4 +-
foreign/php/iggy-php.stubs.php | 59 ++++-
foreign/php/src/client.rs | 9 +-
foreign/php/src/lib.rs | 4 +-
foreign/php/src/message_iterator.rs | 3 +
foreign/php/src/send_message.rs | 87 ++++++-
foreign/php/tests/IggySdkTest.php | 20 ++
foreign/python/Cargo.toml | 4 +-
foreign/python/apache_iggy.pyi | 68 ++++-
foreign/python/pyproject.toml | 2 +-
foreign/python/src/client.rs | 13 +-
foreign/python/src/lib.rs | 4 +-
foreign/python/src/send_message.rs | 98 ++++++-
foreign/python/tests/test_message_operations.py | 40 +++
foreign/python/uv.lock | 2 +-
scripts/ci/license-headers.sh | 7 +-
90 files changed, 2223 insertions(+), 314 deletions(-)
diff --git a/.github/actions/cpp-bazel/pre-merge/action.yml b/.github/actions/cpp-bazel/pre-merge/action.yml
index 0b8d1f3e3..adce77739 100644
--- a/.github/actions/cpp-bazel/pre-merge/action.yml
+++ b/.github/actions/cpp-bazel/pre-merge/action.yml
@@ -31,7 +31,6 @@ runs:
uses: ./.github/actions/utils/setup-cpp-with-cache
- name: Setup Rust with cache
- if: inputs.task != 'lint'
uses: ./.github/actions/utils/setup-rust-with-cache
- name: Install clang-format
@@ -60,6 +59,13 @@ runs:
printf '%s\n' "${files[@]}" | xargs clang-format-18 --style=file --dry-run --Werror
+ - name: Lint (clippy)
+ if: inputs.task == 'lint'
+ shell: bash
+ run: |
+ cd foreign/cpp
+ cargo clippy --manifest-path Cargo.toml --all-targets --all-features -- -D warnings
+
- name: Build
if: inputs.task == 'build'
shell: bash
diff --git a/.github/actions/php/pre-merge/action.yml b/.github/actions/php/pre-merge/action.yml
index 9f4b117ce..a0c817cc7 100644
--- a/.github/actions/php/pre-merge/action.yml
+++ b/.github/actions/php/pre-merge/action.yml
@@ -88,6 +88,7 @@ runs:
run: |
"${PHP}" -r 'json_decode(file_get_contents("foreign/php/composer.json"), true, 512, JSON_THROW_ON_ERROR);'
cargo fmt --manifest-path foreign/php/Cargo.toml -- --check
+ cargo clippy --manifest-path foreign/php/Cargo.toml --all-targets --all-features -- -D warnings
cargo install cargo-php --locked --version 0.1.21
cargo build --manifest-path foreign/php/Cargo.toml
extension="$(find "${CARGO_TARGET_DIR}/debug" -maxdepth 1 -name 'libiggy_php.so' -print -quit)"
diff --git a/.github/actions/python-maturin/pre-merge/action.yml b/.github/actions/python-maturin/pre-merge/action.yml
index 81d3bd50b..07eefaaa2 100644
--- a/.github/actions/python-maturin/pre-merge/action.yml
+++ b/.github/actions/python-maturin/pre-merge/action.yml
@@ -82,6 +82,9 @@ runs:
echo "Running cargo fmt for Python extension..."
cargo fmt --manifest-path Cargo.toml -- --check
+ echo "Running cargo clippy for Python extension..."
+ cargo clippy --manifest-path Cargo.toml --all-targets --all-features -- -D warnings
+
echo "Running pyrefly on SDK..."
uv run pyrefly check "$DIR_SDK"
echo "pyrefly version: $(uv run pyrefly --version)"
diff --git a/Cargo.lock b/Cargo.lock
index 7e304f58c..9111fc8d6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6623,7 +6623,7 @@ checksum = "cd62e6b5e86ea8eeeb8db1de02880a6abc01a397b2ebb64b5d74ac255318f5cb"
[[package]]
name = "iggy"
-version = "0.10.3-edge.3"
+version = "0.11.0-edge.1"
dependencies = [
"async-broadcast",
"async-dropper",
@@ -6646,6 +6646,7 @@ dependencies = [
"rustls",
"secrecy",
"serde",
+ "serde_json",
"tokio",
"tokio-rustls",
"tokio-tungstenite",
@@ -6656,7 +6657,7 @@ dependencies = [
[[package]]
name = "iggy-bench"
-version = "0.5.1-edge.1"
+version = "0.6.0-edge.1"
dependencies = [
"async-trait",
"bench-report",
@@ -6713,7 +6714,7 @@ dependencies = [
[[package]]
name = "iggy-cli"
-version = "0.13.1-edge.1"
+version = "0.14.0-edge.1"
dependencies = [
"anyhow",
"apple-native-keyring-store",
@@ -6747,7 +6748,7 @@ dependencies = [
[[package]]
name = "iggy-connectors"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"axum",
@@ -6833,7 +6834,7 @@ dependencies = [
[[package]]
name = "iggy_binary_protocol"
-version = "0.10.3-edge.3"
+version = "0.11.0-edge.1"
dependencies = [
"aligned-vec",
"bytemuck",
@@ -6845,7 +6846,7 @@ dependencies = [
[[package]]
name = "iggy_common"
-version = "0.10.3-edge.3"
+version = "0.11.0-edge.1"
dependencies = [
"aes-gcm",
"async-broadcast",
@@ -6881,7 +6882,7 @@ dependencies = [
[[package]]
name = "iggy_connector_clickhouse_sink"
-version = "0.1.0"
+version = "0.2.0-edge.1"
dependencies = [
"async-trait",
"bytes",
@@ -6899,7 +6900,7 @@ dependencies = [
[[package]]
name = "iggy_connector_delta_sink"
-version = "0.1.1-edge.1"
+version = "0.2.0-edge.1"
dependencies = [
"async-trait",
"chrono",
@@ -6916,7 +6917,7 @@ dependencies = [
[[package]]
name = "iggy_connector_doris_sink"
-version = "0.1.0"
+version = "0.2.0-edge.1"
dependencies = [
"async-trait",
"base64",
@@ -6936,7 +6937,7 @@ dependencies = [
[[package]]
name = "iggy_connector_elasticsearch_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"base64",
@@ -6954,7 +6955,7 @@ dependencies = [
[[package]]
name = "iggy_connector_elasticsearch_source"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"dashmap",
@@ -6972,7 +6973,7 @@ dependencies = [
[[package]]
name = "iggy_connector_http_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"base64",
@@ -6994,7 +6995,7 @@ dependencies = [
[[package]]
name = "iggy_connector_iceberg_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"arrow-json 57.3.1",
"async-trait",
@@ -7013,7 +7014,7 @@ dependencies = [
[[package]]
name = "iggy_connector_influxdb_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"axum",
@@ -7034,7 +7035,7 @@ dependencies = [
[[package]]
name = "iggy_connector_influxdb_source"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"ahash 0.8.12",
"async-trait",
@@ -7060,7 +7061,7 @@ dependencies = [
[[package]]
name = "iggy_connector_meilisearch_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"base64",
@@ -7078,7 +7079,7 @@ dependencies = [
[[package]]
name = "iggy_connector_mongodb_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"humantime",
@@ -7094,7 +7095,7 @@ dependencies = [
[[package]]
name = "iggy_connector_postgres_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"dashmap",
@@ -7113,7 +7114,7 @@ dependencies = [
[[package]]
name = "iggy_connector_postgres_source"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"base64",
@@ -7135,7 +7136,7 @@ dependencies = [
[[package]]
name = "iggy_connector_quickwit_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"dashmap",
@@ -7149,7 +7150,7 @@ dependencies = [
[[package]]
name = "iggy_connector_random_source"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"dashmap",
@@ -7166,7 +7167,7 @@ dependencies = [
[[package]]
name = "iggy_connector_s3_sink"
-version = "0.4.0"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"base64",
@@ -7223,7 +7224,7 @@ dependencies = [
[[package]]
name = "iggy_connector_stdout_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"dashmap",
@@ -7235,7 +7236,7 @@ dependencies = [
[[package]]
name = "iggy_connector_surrealdb_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
dependencies = [
"async-trait",
"base64",
@@ -11900,7 +11901,7 @@ dependencies = [
[[package]]
name = "server-ng"
-version = "0.9.0-edge.1"
+version = "0.9.0-edge.2"
dependencies = [
"ahash 0.8.12",
"argon2",
diff --git a/Cargo.toml b/Cargo.toml
index 4b327d8a3..f33c0eddc 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -200,10 +200,10 @@ hyper-util = { version = "0.1.20", features = ["server-auto", "service"] }
iceberg = "0.9.1"
iceberg-catalog-rest = "0.9.1"
iceberg-storage-opendal = "0.9.1"
-iggy = { path = "core/sdk", version = "0.10.3-edge.3" }
-iggy-cli = { path = "core/cli", version = "0.13.1-edge.1" }
-iggy_binary_protocol = { path = "core/binary_protocol", version = "0.10.3-edge.3" }
-iggy_common = { path = "core/common", version = "0.10.3-edge.3" }
+iggy = { path = "core/sdk", version = "0.11.0-edge.1" }
+iggy-cli = { path = "core/cli", version = "0.14.0-edge.1" }
+iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.0-edge.1" }
+iggy_common = { path = "core/common", version = "0.11.0-edge.1" }
iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.3.1-edge.1" }
indexmap = "2.14.0"
integration = { path = "core/integration" }
diff --git a/bdd/python/uv.lock b/bdd/python/uv.lock
index 846067ffc..6792069fe 100644
--- a/bdd/python/uv.lock
+++ b/bdd/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.8.1.dev4"
+version = "0.9.0.dev1"
source = { directory = "../../foreign/python" }
[package.metadata]
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/bench/Cargo.toml b/core/bench/Cargo.toml
index 110d90dd4..9842a4413 100644
--- a/core/bench/Cargo.toml
+++ b/core/bench/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-bench"
-version = "0.5.1-edge.1"
+version = "0.6.0-edge.1"
edition = "2024"
license = "Apache-2.0"
repository = "https://github.com/apache/iggy"
diff --git a/core/binary_protocol/Cargo.toml b/core/binary_protocol/Cargo.toml
index ce36c19b1..ec9b25e75 100644
--- a/core/binary_protocol/Cargo.toml
+++ b/core/binary_protocol/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_binary_protocol"
-version = "0.10.3-edge.3"
+version = "0.11.0-edge.1"
description = "Wire protocol types and codec for the Iggy binary protocol. Shared between server and SDK."
edition = "2024"
rust-version.workspace = true
diff --git a/core/binary_protocol/src/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..22a7a4624
--- /dev/null
+++ b/core/binary_protocol/src/responses/messages/send_messages.rs
@@ -0,0 +1,284 @@
+// 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)`.
+///
+/// Frozen. Entries are read at this stride out of an array whose length the
+/// peer chooses, so widening one is not a compatible change: a reader built
+/// against the old stride would take its second entry from the middle of the
+/// first and return the garbage as a successful decode. Extra data needs a new
+/// response type or a protocol-version gate. The entry count may grow freely.
+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, bounded by three properties of the send path:
+/// - Delivery is at-least-once. An earlier retry of the same batch may already
+/// have committed at a lower offset, so the value never implies uniqueness.
+/// - A batch is confirmed once it is committed in memory, not once it is
+/// fsynced. A crash-restart can stamp a later batch with an offset a client
+/// has already recorded.
+/// - The legacy server confirms nothing, so its confirmation list is empty.
+#[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 with no offsets to
+/// report. The legacy server goes further and answers a successful send with no
+/// body at all, so a caller sees an empty list either way and must handle it.
+///
+/// The server currently reports a single partition per request; the list
+/// decodes any count, so a later multi-partition send needs no wire change.
+/// Growing the count is the only compatible way to extend this response, the
+/// entry itself being frozen at 20 bytes.
+#[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.
+ // Tolerating a tail would be sound only where the extra bytes cannot
+ // precede a field the reader still indexes, which a fixed-stride array
+ // breaks at any count above one: entries past the first would be read
+ // from inside their predecessor and returned as a successful decode.
+ 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/cli/Cargo.toml b/core/cli/Cargo.toml
index 869da1fa5..b838b062c 100644
--- a/core/cli/Cargo.toml
+++ b/core/cli/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-cli"
-version = "0.13.1-edge.1"
+version = "0.14.0-edge.1"
edition = "2024"
rust-version.workspace = true
authors = ["bartosz.ciesla@gmail.com"]
diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml
index 3934e5bbd..e3a4b30d3 100644
--- a/core/common/Cargo.toml
+++ b/core/common/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_common"
-version = "0.10.3-edge.3"
+version = "0.11.0-edge.1"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
rust-version.workspace = true
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index 8d2edc9eb..fead3058d 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -17,7 +17,7 @@
use crate::Identifier;
use crate::utils::topic_size::MaxTopicSize;
-use crate::{IggyMessage, utils::byte_size::IggyByteSize};
+use crate::{IggyMessage, SendMessagesConfirmationResponse, utils::byte_size::IggyByteSize};
use std::sync::Arc;
use strum::{EnumDiscriminants, FromRepr, IntoStaticStr};
use thiserror::Error;
@@ -416,6 +416,9 @@ pub enum IggyError {
ProducerSendFailed {
cause: Box<IggyError>,
failed: Arc<Vec<IggyMessage>>,
+ /// Confirmations of the chunks that committed before `cause`; the
+ /// durable prefix of a send that was split into several requests.
+ committed: Arc<Vec<SendMessagesConfirmationResponse>>,
stream_name: String,
topic_name: String,
} = 4056,
diff --git a/core/common/src/http/messages/send_messages.rs b/core/common/src/http/messages/send_messages.rs
index 35faf87ff..48344441a 100644
--- a/core/common/src/http/messages/send_messages.rs
+++ b/core/common/src/http/messages/send_messages.rs
@@ -22,7 +22,9 @@ use crate::Validatable;
use crate::error::IggyError;
use crate::types::message::HeaderEntry;
use crate::types::message::partitioning::Partitioning;
-use crate::{IggyMessage, IggyMessagesBatch};
+use crate::{
+ IggyMessage, IggyMessagesBatch, SendMessagesConfirmationResponse, SendMessagesResponse,
+};
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
use bytes::Bytes;
use serde::de::{self, MapAccess, Visitor};
@@ -302,6 +304,60 @@ impl<'de> Deserialize<'de> for SendMessages {
}
}
+/// JSON body of a successful `POST .../messages`: one entry per partition the
+/// batch landed in. A list rather than a single confirmation, so a
+/// multi-partition produce needs no shape change.
+///
+/// Distinct from the binary `SendMessagesResponse` because wire types stay
+/// codec-only and carry no `serde` derives.
+#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SendMessagesConfirmations {
+ pub confirmations: Vec<SendMessagesConfirmation>,
+}
+
+/// One partition's commit confirmation.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct SendMessagesConfirmation {
+ pub stream_id: u32,
+ pub topic_id: u32,
+ pub partition_id: u32,
+ pub base_offset: u64,
+}
+
+impl From<SendMessagesResponse> for SendMessagesConfirmations {
+ fn from(response: SendMessagesResponse) -> Self {
+ Self {
+ confirmations: response
+ .confirmations
+ .into_iter()
+ .map(|confirmation| SendMessagesConfirmation {
+ stream_id: confirmation.stream_id,
+ topic_id: confirmation.topic_id,
+ partition_id: confirmation.partition_id,
+ base_offset: confirmation.base_offset,
+ })
+ .collect(),
+ }
+ }
+}
+
+impl From<SendMessagesConfirmations> for SendMessagesResponse {
+ fn from(body: SendMessagesConfirmations) -> Self {
+ Self {
+ confirmations: body
+ .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(),
+ }
+ }
+}
+
fn parse_message_id(value: Option<&serde_json::Value>) -> Result<u128, String> {
let value = match value {
Some(v) => v,
diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs
index 12f139d2a..dde606a38 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")]
@@ -60,6 +63,7 @@ pub use traits::client::Client;
pub use traits::cluster_client::ClusterClient;
pub use traits::consumer_group_client::ConsumerGroupClient;
pub use traits::consumer_offset_client::ConsumerOffsetClient;
+pub use traits::decode_send_confirmations;
pub use traits::message_client::MessageClient;
pub use traits::partition_client::PartitionClient;
pub use traits::partitioner::Partitioner;
diff --git a/core/common/src/traits/binary_impls/messages.rs b/core/common/src/traits/binary_impls/messages.rs
index 24c2a799d..8ffe2fd48 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,38 @@ 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 answers that way, so absence must never surface as a decode failure.
+///
+/// Absence is reported as an empty list, never as a zeroed entry. Every field
+/// of a confirmation has 0 as a legitimate value (ids are 0-based slab keys,
+/// the first batch of a partition commits at offset 0), so a synthetic entry
+/// would be indistinguishable from a real one and a caller checkpointing
+/// `base_offset` would record a commit that never happened.
+pub fn decode_send_confirmations(response: &[u8]) -> Result<SendMessagesResponse, IggyError> {
+ if response.is_empty() {
+ return Ok(SendMessagesResponse {
+ confirmations: Vec::new(),
+ });
+ }
+ super::decode_response::<SendMessagesResponse>(response)
+}
+
+/// Confirmations for a batch the server has already committed.
+///
+/// An unreadable body degrades to no confirmations instead of an error. The
+/// producer retry loop filters nothing and resends on any `Err`, so failing
+/// here would turn one committed write into as many copies as the retry budget
+/// allows, on a plane that keeps no reply cache to deduplicate them. Reporting
+/// a zeroed entry instead would be no better: the caller cannot tell it from a
+/// genuine commit at offset 0 and would checkpoint the shape mismatch.
+fn committed_send_confirmations(response: &[u8]) -> SendMessagesResponse {
+ decode_send_confirmations(response).unwrap_or_else(|_| SendMessagesResponse {
+ confirmations: Vec::new(),
+ })
+}
+
#[async_trait::async_trait]
impl<B: BinaryClient> MessageClient for B {
async fn poll_messages(
@@ -303,7 +335,7 @@ impl<B: BinaryClient> MessageClient for B {
topic_id: &Identifier,
partitioning: &Partitioning,
messages: &mut [IggyMessage],
- ) -> Result<(), IggyError> {
+ ) -> Result<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 +376,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(())
+ Ok(committed_send_confirmations(&response))
}
async fn flush_unsaved_buffer(
@@ -368,3 +401,99 @@ impl<B: BinaryClient> MessageClient for B {
Ok(())
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::{committed_send_confirmations, 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,
+ }],
+ }
+ }
+
+ /// The legacy server reports nothing at all, and nothing is what the caller
+ /// must see: no error to retry on, and no entry that reads as a commit at
+ /// offset 0.
+ #[test]
+ fn empty_body_is_no_confirmations() {
+ let decoded = decode_send_confirmations(&[]).expect("empty body must not fail");
+ assert!(decoded.confirmations.is_empty());
+ }
+
+ #[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, expected);
+ }
+
+ #[test]
+ fn zero_count_body_decodes_to_empty_list() {
+ let bytes = SendMessagesResponse {
+ confirmations: vec![],
+ }
+ .to_bytes();
+ let decoded = decode_send_confirmations(&bytes).expect("zero-count payload must decode");
+ assert!(decoded.confirmations.is_empty());
+ }
+
+ #[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}"
+ );
+ }
+ }
+
+ #[test]
+ fn committed_body_keeps_reported_confirmations() {
+ let expected = response();
+ assert_eq!(committed_send_confirmations(&expected.to_bytes()), expected);
+ }
+
+ /// The write is already durable once the reply arrives, so an unreadable
+ /// body degrades to no confirmations. Anything else either resends a
+ /// committed batch or hands the caller a fabricated offset.
+ #[test]
+ fn committed_malformed_body_is_no_confirmations() {
+ let valid = response().to_bytes();
+
+ let mut with_tail = valid.to_vec();
+ with_tail.push(0xFF);
+ let degraded = committed_send_confirmations(&with_tail);
+ assert!(degraded.confirmations.is_empty());
+
+ for length in 1..valid.len() {
+ let degraded = committed_send_confirmations(&valid[..length]);
+ assert!(
+ degraded.confirmations.is_empty(),
+ "expected no confirmations for truncation at byte {length}"
+ );
+ }
+ }
+}
diff --git a/core/common/src/traits/binary_impls/mod.rs b/core/common/src/traits/binary_impls/mod.rs
index 7a7e7e6d4..23e3c590e 100644
--- a/core/common/src/traits/binary_impls/mod.rs
+++ b/core/common/src/traits/binary_impls/mod.rs
@@ -27,6 +27,8 @@ mod system;
mod topics;
mod users;
+pub use messages::decode_send_confirmations;
+
use crate::IggyError;
use crate::http::users::defaults::{
MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH, MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH,
diff --git a/core/common/src/traits/message_client.rs b/core/common/src/traits/message_client.rs
index aa332d618..23e79c5ea 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,25 @@ 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, which may be empty: the
+ /// legacy server reports none, and a server that does report them can still
+ /// commit a batch it has no offsets to describe. Callers must handle an
+ /// empty list rather than assume a confirmation per send.
+ ///
+ /// A reported `base_offset` is where the batch's first message landed, with
+ /// two limits. Delivery is at-least-once, so an earlier retry may already
+ /// have committed the same batch at a lower offset and the value never
+ /// implies uniqueness. A batch is confirmed once it is committed in memory,
+ /// not once it is fsynced, so a crash-restart can stamp a later batch with
+ /// an offset a client has already recorded.
async fn send_messages(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
partitioning: &Partitioning,
messages: &mut [IggyMessage],
- ) -> Result<(), IggyError>;
+ ) -> Result<SendMessagesResponse, IggyError>;
/// Force flush of the `unsaved_messages` buffer to disk, optionally fsyncing the data.
#[allow(clippy::too_many_arguments)]
diff --git a/core/common/src/traits/mod.rs b/core/common/src/traits/mod.rs
index 1621fdb8a..2d8056a9b 100644
--- a/core/common/src/traits/mod.rs
+++ b/core/common/src/traits/mod.rs
@@ -34,3 +34,5 @@ pub(crate) mod system_client;
pub(crate) mod topic_client;
pub(crate) mod user_client;
pub(crate) mod validatable;
+
+pub use binary_impls::decode_send_confirmations;
diff --git a/core/common/src/types/message/mod.rs b/core/common/src/types/message/mod.rs
index 39e8375e4..ccba3df11 100644
--- a/core/common/src/types/message/mod.rs
+++ b/core/common/src/types/message/mod.rs
@@ -37,7 +37,9 @@ mod user_headers;
pub const INDEX_SIZE: usize = 16;
pub use crate::http::messages::poll_messages::PollMessages;
-pub use crate::http::messages::send_messages::SendMessages;
+pub use crate::http::messages::send_messages::{
+ SendMessages, SendMessagesConfirmation, SendMessagesConfirmations,
+};
pub use iggy_message::{IggyMessage, MAX_PAYLOAD_SIZE, MAX_USER_HEADERS_SIZE};
pub use index::IggyIndex;
pub use index_view::IggyIndexView;
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
index 648300999..953351667 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy-connectors"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Connectors runtime for Iggy message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/clickhouse_sink/Cargo.toml b/core/connectors/sinks/clickhouse_sink/Cargo.toml
index bb74cbf42..25135a7dc 100644
--- a/core/connectors/sinks/clickhouse_sink/Cargo.toml
+++ b/core/connectors/sinks/clickhouse_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_clickhouse_sink"
-version = "0.1.0"
+version = "0.2.0-edge.1"
description = "Iggy ClickHouse sink connector for streaming messages into ClickHouse"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/delta_sink/Cargo.toml b/core/connectors/sinks/delta_sink/Cargo.toml
index 72a11cf0f..16f554535 100644
--- a/core/connectors/sinks/delta_sink/Cargo.toml
+++ b/core/connectors/sinks/delta_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_delta_sink"
-version = "0.1.1-edge.1"
+version = "0.2.0-edge.1"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/doris_sink/Cargo.toml b/core/connectors/sinks/doris_sink/Cargo.toml
index 1a5a4722e..f032024b9 100644
--- a/core/connectors/sinks/doris_sink/Cargo.toml
+++ b/core/connectors/sinks/doris_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_doris_sink"
-version = "0.1.0"
+version = "0.2.0-edge.1"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/elasticsearch_sink/Cargo.toml b/core/connectors/sinks/elasticsearch_sink/Cargo.toml
index 93e848316..095590ab7 100644
--- a/core/connectors/sinks/elasticsearch_sink/Cargo.toml
+++ b/core/connectors/sinks/elasticsearch_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_elasticsearch_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy Elasticsearch sink connector"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/http_sink/Cargo.toml b/core/connectors/sinks/http_sink/Cargo.toml
index a52ec1d47..862de5d54 100644
--- a/core/connectors/sinks/http_sink/Cargo.toml
+++ b/core/connectors/sinks/http_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_http_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy HTTP sink connector for delivering stream messages to any HTTP endpoint via webhooks, REST APIs, or serverless functions."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/iceberg_sink/Cargo.toml b/core/connectors/sinks/iceberg_sink/Cargo.toml
index 1624ae0b6..c77601afb 100644
--- a/core/connectors/sinks/iceberg_sink/Cargo.toml
+++ b/core/connectors/sinks/iceberg_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_iceberg_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
edition = "2024"
license = "Apache-2.0"
keywords = ["iggy", "messaging", "streaming"]
diff --git a/core/connectors/sinks/influxdb_sink/Cargo.toml b/core/connectors/sinks/influxdb_sink/Cargo.toml
index 269a4dd14..6e5bae4df 100644
--- a/core/connectors/sinks/influxdb_sink/Cargo.toml
+++ b/core/connectors/sinks/influxdb_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_influxdb_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy InfluxDB sink connector for storing stream messages as line protocol"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/meilisearch_sink/Cargo.toml b/core/connectors/sinks/meilisearch_sink/Cargo.toml
index 44233a9dc..f45832857 100644
--- a/core/connectors/sinks/meilisearch_sink/Cargo.toml
+++ b/core/connectors/sinks/meilisearch_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_meilisearch_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy Meilisearch sink connector"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/mongodb_sink/Cargo.toml b/core/connectors/sinks/mongodb_sink/Cargo.toml
index 3c989c08c..756c0de2f 100644
--- a/core/connectors/sinks/mongodb_sink/Cargo.toml
+++ b/core/connectors/sinks/mongodb_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_mongodb_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy MongoDB sink connector for storing stream messages into MongoDB database"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/postgres_sink/Cargo.toml b/core/connectors/sinks/postgres_sink/Cargo.toml
index 7c2bd8e3d..a558022fc 100644
--- a/core/connectors/sinks/postgres_sink/Cargo.toml
+++ b/core/connectors/sinks/postgres_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_postgres_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy PostgreSQL sink connector for storing stream messages into PostgreSQL database"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/quickwit_sink/Cargo.toml b/core/connectors/sinks/quickwit_sink/Cargo.toml
index b89010baf..d4a13fdfc 100644
--- a/core/connectors/sinks/quickwit_sink/Cargo.toml
+++ b/core/connectors/sinks/quickwit_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_quickwit_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/s3_sink/Cargo.toml b/core/connectors/sinks/s3_sink/Cargo.toml
index 6f4b64606..087ec6004 100644
--- a/core/connectors/sinks/s3_sink/Cargo.toml
+++ b/core/connectors/sinks/s3_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_s3_sink"
-version = "0.4.0"
+version = "0.5.0-edge.1"
description = "Iggy S3 sink connector for writing stream messages to Amazon S3 and S3-compatible stores"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/stdout_sink/Cargo.toml b/core/connectors/sinks/stdout_sink/Cargo.toml
index b7a4f4cf7..80190ba4a 100644
--- a/core/connectors/sinks/stdout_sink/Cargo.toml
+++ b/core/connectors/sinks/stdout_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_stdout_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sinks/surrealdb_sink/Cargo.toml b/core/connectors/sinks/surrealdb_sink/Cargo.toml
index 1ad27d71d..61d0b3d97 100644
--- a/core/connectors/sinks/surrealdb_sink/Cargo.toml
+++ b/core/connectors/sinks/surrealdb_sink/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_surrealdb_sink"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy SurrealDB sink connector for writing stream messages into SurrealDB"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/elasticsearch_source/Cargo.toml b/core/connectors/sources/elasticsearch_source/Cargo.toml
index 8b4638029..5d81ad6be 100644
--- a/core/connectors/sources/elasticsearch_source/Cargo.toml
+++ b/core/connectors/sources/elasticsearch_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_elasticsearch_source"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy Elasticsearch source connector"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/influxdb_source/Cargo.toml b/core/connectors/sources/influxdb_source/Cargo.toml
index 817d9b803..65fe82957 100644
--- a/core/connectors/sources/influxdb_source/Cargo.toml
+++ b/core/connectors/sources/influxdb_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_influxdb_source"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy InfluxDB source connector for polling Flux query results"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/postgres_source/Cargo.toml b/core/connectors/sources/postgres_source/Cargo.toml
index 8860c177f..3bf59732f 100644
--- a/core/connectors/sources/postgres_source/Cargo.toml
+++ b/core/connectors/sources/postgres_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_postgres_source"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy PostgreSQL source connector supporting CDC and table polling for message streaming platform"
edition = "2024"
license = "Apache-2.0"
diff --git a/core/connectors/sources/random_source/Cargo.toml b/core/connectors/sources/random_source/Cargo.toml
index ffd6a4d93..f995ae114 100644
--- a/core/connectors/sources/random_source/Cargo.toml
+++ b/core/connectors/sources/random_source/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy_connector_random_source"
-version = "0.4.1-edge.1"
+version = "0.5.0-edge.1"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
license = "Apache-2.0"
diff --git a/core/integration/tests/sdk/mod.rs b/core/integration/tests/sdk/mod.rs
index 1182d9ea4..ade2ec45c 100644
--- a/core/integration/tests/sdk/mod.rs
+++ b/core/integration/tests/sdk/mod.rs
@@ -25,3 +25,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..167f7370e 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 response = producer.send(messages).await.unwrap();
+ assert!(
+ response.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..616ee700a
--- /dev/null
+++ b/core/integration/tests/sdk/send_confirmation.rs
@@ -0,0 +1,268 @@
+// 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 no confirmations 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");
+ 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");
+ 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");
+ 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");
+ 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 response = producer
+ .send(batch(CHUNKS * CHUNK_LENGTH))
+ .await
+ .expect("producer send");
+
+ assert_eq!(
+ response.confirmations.len(),
+ CHUNKS as usize,
+ "a direct producer concatenates the confirmation of every chunk it split the send into"
+ );
+ for (chunk, confirmation) in response.confirmations.iter().enumerate() {
+ 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_confirmations(harness: &TestHarness) {
+ let client = harness.root_client().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::balanced(),
+ &mut messages,
+ )
+ .await
+ .expect("send_messages");
+
+ assert!(
+ response.confirmations.is_empty(),
+ "the legacy server reports no offsets; a synthetic entry would be \
+ indistinguishable from a genuine commit at offset 0"
+ );
+}
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 5c6b14882..d7cc7338f 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::{
@@ -1499,9 +1502,16 @@ where
// very different risk:
// - below the sequencer: a duplicate delivery (parked-frame
// redispatch, retransmit echo) of an op this primary already
- // sequenced. Apply is keyed by `header.op` and the primary
- // never advances its sequencer post-apply, so proceeding is
- // idempotent-safe; log loudly for diagnosis.
+ // sequenced. Proceeding is safe only because the two gates above
+ // already returned for every copy this replica can still see:
+ // `fence_old_prepare_by_commit` drops the executed ops and
+ // `journal_holds_op` the resident ones, so reaching here means
+ // the journal lacks this op and has to be given it. Apply is not
+ // idempotent on its own for a produce: `append_messages`
+ // re-stamps from the local dirty counter and the journal's op
+ // index is last-write-wins, so appending an op the journal
+ // already holds would mint a second copy at fresh offsets and
+ // orphan the first. Log loudly for diagnosis.
// - above the sequencer: journaling an op the sequencer has not
// assigned yet means the next local assignment would collide
// with it. Unreachable today (view fences run first, one
@@ -2168,16 +2178,21 @@ where
}
let mut failed_commit = false;
- let committed_visible_offsets = self.resolve_committed_visible_offsets(&drained).await;
+ // Must run BEFORE the commit loop: `commit_messages` evicts the
+ // committed prefix, after which an entry survives only in the bounded
+ // repair ring - and not even there on a single replica, which keeps no
+ // ring at all. A miss degrades to a successful send carrying no
+ // confirmation, a legal answer no client can tell from a real one.
+ let committed_batch_stats = self.resolve_committed_visible_offsets(&drained);
let mut messages_committed = false;
- for entry in drained {
+ for (entry, batch_stats) in drained.into_iter().zip(committed_batch_stats) {
let prepare_header = entry.header;
if !self
.commit_partition_entry(
prepare_header,
&mut messages_committed,
- &committed_visible_offsets,
+ batch_stats,
&mut failed_commit,
config,
)
@@ -2228,10 +2243,13 @@ where
// reply. Emitting it would push an unrequested frame onto a real
// client's lockstep reply stream if the sentinel ever routed there.
if send_client_replies && !is_auto_commit_client(prepare_header.client) {
- let reply = build_reply_message(
- &prepare_header,
- &committed_reply_body(prepare_header.operation),
- );
+ let body = match prepare_header.operation {
+ Operation::SendMessages => {
+ send_messages_reply_body(prepare_header.namespace, batch_stats)
+ }
+ operation => committed_reply_body(operation),
+ };
+ let reply = build_reply_message(&prepare_header, &body);
let reply_buffers = reply.into_generic().into_frozen();
emit_sim_event(SimEventKind::ClientReplyEmitted, &event);
@@ -2269,45 +2287,47 @@ where
self.drain_request_queue_into_prepares(drained_count).await;
}
- async fn resolve_committed_visible_offsets(
+ /// Batch stats for each drained entry, positionally parallel to `drained`.
+ /// Every entry contributes exactly one slot (`None` for the operations that
+ /// carry no batch), which is what makes the pairing correct by
+ /// construction; keying on `op` instead would let a lookup miss attribute
+ /// one batch's offsets to another entry's reply.
+ fn resolve_committed_visible_offsets(
&self,
drained: &[PipelineEntry],
- ) -> HashMap<u64, CommittedBatchStats> {
- let mut committed_visible_offsets = HashMap::new();
-
- for entry in drained {
- if entry.header.operation != Operation::SendMessages {
- continue;
- }
-
- match self.committed_batch_stats_for_prepare(&entry.header).await {
- Ok(Some(batch_stats)) => {
- committed_visible_offsets.insert(entry.header.op, batch_stats);
- }
- Ok(None) => {}
- Err(error) => {
- warn!(
- target: "iggy.partitions.diag",
- plane = "partitions",
- replica_id = self.consensus.replica(),
- namespace_raw = self.namespace().inner(),
- op = entry.header.op,
- operation = ?entry.header.operation,
- %error,
- "failed to resolve committed visible offset for partition entry"
- );
+ ) -> Vec<Option<CommittedBatchStats>> {
+ drained
+ .iter()
+ .map(|entry| {
+ if entry.header.operation != Operation::SendMessages {
+ return None;
}
- }
- }
- committed_visible_offsets
+ match self.committed_batch_stats_for_prepare(&entry.header) {
+ Ok(batch_stats) => batch_stats,
+ Err(error) => {
+ warn!(
+ target: "iggy.partitions.diag",
+ plane = "partitions",
+ replica_id = self.consensus.replica(),
+ namespace_raw = self.namespace().inner(),
+ op = entry.header.op,
+ operation = ?entry.header.operation,
+ %error,
+ "failed to resolve committed visible offset for partition entry"
+ );
+ None
+ }
+ }
+ })
+ .collect()
}
async fn commit_partition_entry(
&mut self,
prepare_header: PrepareHeader,
messages_committed: &mut bool,
- committed_visible_offsets: &HashMap<u64, CommittedBatchStats>,
+ batch_stats: Option<CommittedBatchStats>,
failed_commit: &mut bool,
config: &PartitionsConfig,
) -> bool {
@@ -2331,17 +2351,18 @@ where
*messages_committed = true;
}
- if let Some(batch_stats) = committed_visible_offsets.get(&prepare_header.op) {
+ if let Some(batch_stats) = batch_stats {
+ let end_offset = batch_stats.end_offset();
// A repaired batch at or below the boot-time recovered
// durable offset was already counted (and persisted)
// before the restart; skip it. Live traffic always sits
// above the (immutable) line.
if self
.recovered_durable_offset
- .is_none_or(|durable| batch_stats.end_offset > durable)
+ .is_none_or(|durable| end_offset > durable)
{
- self.offset.store(batch_stats.end_offset, Ordering::Release);
- self.stats.set_current_offset(batch_stats.end_offset);
+ self.offset.store(end_offset, Ordering::Release);
+ self.stats.set_current_offset(end_offset);
// Advance the aggregate stats with the visible offset. Disk
// persistence is threshold-gated in `commit_messages`, which
// must not also touch these counters or committed messages
@@ -2375,13 +2396,36 @@ where
}
}
- async fn committed_batch_stats_for_prepare(
+ /// Read the committed batch's own stamps back out of the journal.
+ ///
+ /// INVARIANT: two replicas can never report a different `base_offset` for
+ /// the same batch. Backups do re-stamp from their own `dirty_offset` in
+ /// `append_messages`, so the guarantee is not "the bytes are replicated";
+ /// it rests on three mechanisms. The backup gap check drops any prepare
+ /// that is not `current_op + 1`, so every replica stamps a partition's
+ /// batches in the primary's order off the same counter.
+ /// `append_repaired_send_messages` journals a repaired prepare with its
+ /// embedded stamps instead of re-stamping, so filling a hole out of live
+ /// order cannot re-mint offsets. And that same path advances the counter
+ /// with `dirty.max(last_offset)`, so a repaired window below the recovered
+ /// durable end cannot rewind it and hand the next live batch offsets that
+ /// were already issued.
+ ///
+ /// `repair_entry` is deliberate: it never awaits, and it falls back to the
+ /// evicted ring, which the resident-only lookup does not.
+ fn committed_batch_stats_for_prepare(
&self,
prepare_header: &PrepareHeader,
) -> Result<Option<CommittedBatchStats>, IggyError> {
- let Some(entry) = self.log.journal().inner.entry(prepare_header).await else {
- return Err(IggyError::InvalidCommand);
- };
+ let entry = self
+ .log
+ .journal()
+ .inner
+ .repair_entry(prepare_header.op)
+ // A resident slot can read back empty, which the caller must treat
+ // as a miss and not as a zero-message batch.
+ .filter(|entry| !entry.is_empty())
+ .ok_or(IggyError::InvalidCommand)?;
// Trusted (no batch-hash): the entry was read back from this replica's
// own journal, where it was stamped/validated at append; only header
// stats are needed, so re-hashing the ~1 MiB blob is redundant.
@@ -2393,7 +2437,7 @@ where
}
Ok(Some(CommittedBatchStats {
- end_offset: batch.header.base_offset + u64::from(message_count) - 1,
+ base_offset: batch.header.base_offset,
message_count,
size_bytes: batch.header.total_size() as u64,
}))
@@ -3397,11 +3441,14 @@ 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.
+/// Success reply body for a committed partition op other than `SendMessages`
+/// (which confirms its offsets through [`send_messages_reply_body`]).
+///
+/// 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])
@@ -3410,16 +3457,63 @@ const fn committed_reply_body(operation: Operation) -> bytes::Bytes {
}
}
+// The confirmation payload below ships raw, with no result section ahead of it.
+// If `SendMessages` ever became result-framed, a batch with confirmations would
+// misdecode into a spurious typed error, which is loud; a batch without them
+// would decode as a clean success, which is silent.
+const _: () = assert!(!Operation::SendMessages.is_result_framed());
+
+/// One confirmation for the committed batch, or `count = 0` when its offsets
+/// could not be resolved (missing or undecodable journal entry, or an empty
+/// batch).
+///
+/// `count = 0` is a first-class answer meaning "committed, no offsets to
+/// report", not a decode problem: the SDK reads it as an empty list, exactly as
+/// it reads the legacy server's empty body. That is also why absence must stay
+/// absent - a placeholder entry would carry a valid stream/topic/partition/
+/// offset tuple and be indistinguishable from a real commit at offset 0.
+#[allow(clippy::cast_possible_truncation)]
+fn send_messages_reply_body(
+ namespace: u64,
+ batch_stats: Option<CommittedBatchStats>,
+) -> bytes::Bytes {
+ let Some(stats) = batch_stats else {
+ return bytes::Bytes::from_static(&[0, 0, 0, 0]);
+ };
+ let namespace = IggyNamespace::from_raw(namespace);
+ SendMessagesResponse {
+ confirmations: vec![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,
+ }],
+ }
+ .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 {
- end_offset: u64,
+ base_offset: u64,
message_count: u32,
size_bytes: u64,
}
+impl CommittedBatchStats {
+ /// Offset of the batch's last message. The batch carries a contiguous
+ /// offset run, and the sole constructor rejects an empty one, so the
+ /// subtraction cannot underflow.
+ fn end_offset(self) -> u64 {
+ self.base_offset + u64::from(self.message_count) - 1
+ }
+}
+
/// Fold one `SendMessages` batch's accounting into a running `JournalInfo`,
/// matching the field updates `append_messages` applies per append.
/// `current_offset` is the batch's last message offset; the batch carries a
@@ -4711,6 +4805,64 @@ mod tests {
assert_eq!(partition.consensus().commit_min(), 0);
}
+
+ fn batch_stats(base_offset: u64, message_count: u32) -> CommittedBatchStats {
+ CommittedBatchStats {
+ base_offset,
+ 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 = send_messages_reply_body(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 = send_messages_reply_body(namespace.inner(), None);
+
+ assert_eq!(&body[..], &[0, 0, 0, 0]);
+ let (response, _) = SendMessagesResponse::decode(&body).unwrap();
+ assert!(response.confirmations.is_empty());
+ }
+
+ #[test]
+ fn given_batch_stats_when_end_offset_derived_should_span_the_message_run() {
+ assert_eq!(batch_stats(9, 1).end_offset(), 9);
+ assert_eq!(batch_stats(9, 4).end_offset(), 12);
+ }
+
+ #[test]
+ fn given_result_framed_operation_when_committed_should_reply_empty_result_section() {
+ assert_eq!(
+ &committed_reply_body(Operation::StoreConsumerOffset2)[..],
+ &[0, 0, 0, 0]
+ );
+ }
+
+ #[test]
+ fn given_unframed_operation_when_committed_should_reply_empty_body() {
+ assert!(committed_reply_body(Operation::DeleteSegments).is_empty());
+ }
}
#[cfg(test)]
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index 5f8d91aa9..e5562b27f 100644
--- a/core/sdk/Cargo.toml
+++ b/core/sdk/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "iggy"
-version = "0.10.3-edge.3"
+version = "0.11.0-edge.1"
description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
edition = "2024"
rust-version.workspace = true
@@ -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..eb3eed94f 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<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..db722a006 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<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..6d93cc334 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,
+ SendMessagesConfirmationResponse, SendMessagesResponse,
};
use std::sync::Arc;
use std::sync::atomic::Ordering;
@@ -41,13 +42,42 @@ use mockall::automock;
#[cfg_attr(test, automock)]
pub trait ProducerCoreBackend: Send + Sync + 'static {
+ /// Sends `msgs`, returning the confirmations of every chunk the send was
+ /// split into, concatenated in chunk order.
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<SendMessagesResponse, IggyError>> + Send;
+}
+
+/// Reply for a send that produced no confirmation: nothing was sent, or the
+/// send is still queued on a background dispatcher.
+pub(crate) fn no_confirmations() -> SendMessagesResponse {
+ SendMessagesResponse {
+ confirmations: Vec::new(),
+ }
+}
+
+/// True when `error` can only have been raised after the server committed the
+/// batch. Resending then turns one durable write into as many copies as the
+/// retry budget allows, on a plane that keeps no reply cache to collapse them.
+///
+/// Both kinds are raised while decoding the HTTP reply body, which is reached
+/// only once the status check has accepted a 2xx, so the batch landed and just
+/// its confirmation is unreadable. The binary path degrades an unreadable body
+/// to an empty confirmation list and never raises either kind.
+///
+/// Membership is scoped to the send path and must stay conservative. An error
+/// meaning the request never arrived, or that the server rejected it before
+/// committing, has to keep retrying.
+fn implies_committed_send(error: &IggyError) -> bool {
+ matches!(
+ error,
+ IggyError::InvalidBytesResponse | IggyError::InvalidJsonResponse
+ )
}
pub struct ProducerCore {
@@ -183,7 +213,7 @@ impl ProducerCore {
topic: &Identifier,
partitioning: &Arc<Partitioning>,
messages: &mut [IggyMessage],
- ) -> Result<(), IggyError> {
+ ) -> Result<SendMessagesResponse, IggyError> {
let client = self.client.read().await;
let Some(max_retries) = self.send_retries_count else {
@@ -249,7 +279,7 @@ impl ProducerCore {
topic: &Identifier,
partitioning: &Arc<Partitioning>,
messages: &mut [IggyMessage],
- ) -> Result<(), IggyError> {
+ ) -> Result<SendMessagesResponse, IggyError> {
let mut retries = 0;
let mut timer: Option<Interval> = None;
@@ -258,8 +288,18 @@ 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) => {
+ if implies_committed_send(&error) {
+ error!(
+ "Not retrying a send to topic: {topic}, stream: {stream}: the batch \
+ committed and only its confirmation could not be read. {error}."
+ );
+ return Err(error);
+ }
+
retries += 1;
if retries > max_retries {
error!(
@@ -344,10 +384,16 @@ impl ProducerCore {
sleep(Duration::from_micros(remaining)).await;
}
- fn make_failed_error(&self, cause: IggyError, failed: Vec<IggyMessage>) -> IggyError {
+ fn make_failed_error(
+ &self,
+ cause: IggyError,
+ failed: Vec<IggyMessage>,
+ committed: Vec<SendMessagesConfirmationResponse>,
+ ) -> IggyError {
IggyError::ProducerSendFailed {
cause: Box::new(cause),
failed: Arc::new(failed),
+ committed: Arc::new(committed),
stream_name: self.stream_name.clone(),
topic_name: self.topic_name.clone(),
}
@@ -361,19 +407,19 @@ impl ProducerCoreBackend for ProducerCore {
topic: &Identifier,
mut msgs: Vec<IggyMessage>,
partitioning: Option<Arc<Partitioning>>,
- ) -> Result<(), IggyError> {
+ ) -> Result<SendMessagesResponse, IggyError> {
if msgs.is_empty() {
- return Ok(());
+ return Ok(no_confirmations());
}
if let Err(err) = self.encrypt_messages(&mut msgs) {
- return Err(self.make_failed_error(err, msgs));
+ return Err(self.make_failed_error(err, msgs, Vec::new()));
}
let part = match self.get_partitioning(stream, topic, &msgs, partitioning.clone()) {
Ok(p) => p,
Err(err) => {
- return Err(self.make_failed_error(err, msgs));
+ return Err(self.make_failed_error(err, msgs, Vec::new()));
}
};
@@ -391,30 +437,35 @@ 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));
+ match self.try_send_messages(stream, topic, &part, chunk).await {
+ Ok(response) => confirmations.extend(response.confirmations),
+ Err(err) => {
+ let failed_tail = msgs.split_off(index);
+ return Err(self.make_failed_error(err, failed_tail, confirmations));
+ }
}
self.last_sent_at
.store(IggyTimestamp::now().into(), ORDERING);
index = end;
}
+ Ok(SendMessagesResponse { confirmations })
}
// background send on
_ => {
- self.try_send_messages(stream, topic, &part, &mut msgs)
+ let response = self
+ .try_send_messages(stream, topic, &part, &mut msgs)
.await
- .map_err(|err| self.make_failed_error(err, msgs))?;
+ .map_err(|err| self.make_failed_error(err, msgs, Vec::new()))?;
self.last_sent_at
.store(IggyTimestamp::now().into(), ORDERING);
+ Ok(response)
}
}
-
- Ok(())
}
}
@@ -496,17 +547,39 @@ impl IggyProducer {
self.core.init().await
}
- pub async fn send(&self, messages: Vec<IggyMessage>) -> Result<(), IggyError> {
+ /// Sends `messages` and returns the commit confirmations of every chunk the
+ /// send was split into, concatenated in chunk order. A retried chunk
+ /// contributes only the confirmation of the attempt that finally succeeded.
+ ///
+ /// Delivery is at-least-once. An earlier retry may already have committed
+ /// the same messages at a lower offset, so `base_offset` never implies
+ /// uniqueness.
+ ///
+ /// A batch is confirmed once it is committed in memory, not once it is
+ /// fsynced. A crash-restart can stamp a later batch with an offset a client
+ /// has already recorded.
+ ///
+ /// The confirmation list is empty whenever the server sends no confirmation
+ /// payload, as the legacy server never sends one, and for a `background`
+ /// producer, which hands the messages to a dispatcher and returns before the
+ /// send happens. Branch on `confirmations.is_empty()` instead of indexing.
+ pub async fn send(
+ &self,
+ messages: Vec<IggyMessage>,
+ ) -> Result<SendMessagesResponse, IggyError> {
if messages.is_empty() {
trace!("No messages to send.");
- return Ok(());
+ return Ok(no_confirmations());
}
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(|()| no_confirmations()),
None => {
self.core
.send_internal(&stream_id, &topic_id, messages, None)
@@ -515,28 +588,30 @@ 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<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<SendMessagesResponse, IggyError> {
if messages.is_empty() {
trace!("No messages to send.");
- return Ok(());
+ return Ok(no_confirmations());
}
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(|()| no_confirmations()),
None => {
self.core
.send_internal(&stream_id, &topic_id, messages, partitioning)
@@ -545,20 +620,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<SendMessagesResponse, IggyError> {
if messages.is_empty() {
trace!("No messages to send.");
- return Ok(());
+ return Ok(no_confirmations());
}
match &self.dispatcher {
- Some(disp) => disp.dispatch(messages, stream, topic, partitioning).await,
+ Some(disp) => disp
+ .dispatch(messages, stream, topic, partitioning)
+ .await
+ .map(|()| no_confirmations()),
None => {
self.core
.send_internal(&stream, &topic, messages, partitioning)
@@ -577,3 +656,33 @@ impl IggyProducer {
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::implies_committed_send;
+ use iggy_common::IggyError;
+
+ #[test]
+ fn test_unreadable_confirmation_of_a_committed_batch_stops_retrying() {
+ assert!(implies_committed_send(&IggyError::InvalidBytesResponse));
+ assert!(implies_committed_send(&IggyError::InvalidJsonResponse));
+ }
+
+ #[test]
+ fn test_errors_reachable_before_a_commit_keep_retrying() {
+ for error in [
+ IggyError::Disconnected,
+ IggyError::EmptyResponse,
+ IggyError::Unauthenticated,
+ IggyError::Unauthorized,
+ IggyError::CannotSendMessagesDueToClientDisconnection,
+ IggyError::HttpResponseError(500, String::new()),
+ IggyError::ResourceNotFound(String::new()),
+ ] {
+ assert!(
+ !implies_committed_send(&error),
+ "{error} does not prove a commit, so it must stay retryable"
+ );
+ }
+ }
+}
diff --git a/core/sdk/src/clients/producer_dispatcher.rs b/core/sdk/src/clients/producer_dispatcher.rs
index 38950dc0f..1cfe1f751 100644
--- a/core/sdk/src/clients/producer_dispatcher.rs
+++ b/core/sdk/src/clients/producer_dispatcher.rs
@@ -207,7 +207,7 @@ mod tests {
use bytes::Bytes;
use tokio::time::sleep;
- use crate::clients::producer::MockProducerCoreBackend;
+ use crate::clients::producer::{MockProducerCoreBackend, no_confirmations};
use crate::clients::producer_error_callback::ErrorCallback;
use crate::clients::producer_sharding::Sharding;
@@ -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(no_confirmations()) }));
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(no_confirmations()) }));
let msg = ShardMessage {
stream: dummy_identifier(),
@@ -395,6 +395,7 @@ mod tests {
Err(IggyError::ProducerSendFailed {
cause: Box::new(IggyError::Error),
failed: Arc::new(vec![dummy_message(10)]),
+ committed: Arc::new(Vec::new()),
stream_name: "1".to_string(),
topic_name: "1".to_string(),
})
diff --git a/core/sdk/src/clients/producer_error_callback.rs b/core/sdk/src/clients/producer_error_callback.rs
index b376a4010..2d2bc8514 100644
--- a/core/sdk/src/clients/producer_error_callback.rs
+++ b/core/sdk/src/clients/producer_error_callback.rs
@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
-use iggy_common::{Identifier, IggyError, IggyMessage, Partitioning};
+use iggy_common::{
+ Identifier, IggyError, IggyMessage, Partitioning, SendMessagesConfirmationResponse,
+};
use std::fmt::Debug;
use std::pin::Pin;
use std::sync::Arc;
@@ -30,6 +32,9 @@ pub struct ErrorCtx {
pub topic_name: String,
pub partitioning: Option<Arc<Partitioning>>,
pub messages: Arc<Vec<IggyMessage>>,
+ /// Confirmations of the chunks that committed before the failure; `messages`
+ /// is the tail that did not.
+ pub committed: Arc<Vec<SendMessagesConfirmationResponse>>,
}
/// A trait for handling background sending errors.
@@ -42,7 +47,8 @@ pub trait ErrorCallback: Send + Sync + Debug + 'static {
/// Default implementation of [`ErrorCallback`] that logs the error using `tracing::error!`.
///
-/// Logs include stream, topic, optional partitioning, number of messages, and the cause.
+/// Logs include stream, topic, optional partitioning, number of messages, how
+/// many chunks committed before the failure, and the cause.
#[derive(Debug, Default)]
pub struct LogErrorCallback;
@@ -63,6 +69,7 @@ impl ErrorCallback for LogErrorCallback {
topic_name = ctx.topic_name,
partitioning = %partitioning,
num_messages = ctx.messages.len(),
+ committed_confirmations = ctx.committed.len(),
"Failed to send messages in background task",
);
})
diff --git a/core/sdk/src/clients/producer_sharding.rs b/core/sdk/src/clients/producer_sharding.rs
index 5a8ae0ad2..668256f83 100644
--- a/core/sdk/src/clients/producer_sharding.rs
+++ b/core/sdk/src/clients/producer_sharding.rs
@@ -249,6 +249,7 @@ impl Shard {
if let Err(error) = result {
if let IggyError::ProducerSendFailed {
failed,
+ committed,
cause,
stream_name,
topic_name,
@@ -262,6 +263,7 @@ impl Shard {
topic_name: topic_name.clone(),
partitioning: msg.inner.partitioning,
messages: failed.clone(),
+ committed: committed.clone(),
};
let _ = err_sender.send_async(ctx).await;
} else {
@@ -293,7 +295,7 @@ impl Shard {
#[cfg(test)]
mod tests {
use super::*;
- use crate::clients::producer::MockProducerCoreBackend;
+ use crate::clients::producer::{MockProducerCoreBackend, no_confirmations};
use bytes::Bytes;
use iggy_common::IggyDuration;
use std::time::Duration;
@@ -315,7 +317,7 @@ mod tests {
let mut mock = MockProducerCoreBackend::new();
mock.expect_send_internal()
.times(10)
- .returning(|_, _, _, _| Box::pin(async { Ok(()) }));
+ .returning(|_, _, _, _| Box::pin(async { Ok(no_confirmations()) }));
let bb = BackgroundConfig::builder()
.batch_length(10)
@@ -357,7 +359,7 @@ mod tests {
let mut mock = MockProducerCoreBackend::new();
mock.expect_send_internal()
.times(1)
- .returning(|_, _, _, _| Box::pin(async { Ok(()) }));
+ .returning(|_, _, _, _| Box::pin(async { Ok(no_confirmations()) }));
let bb = BackgroundConfig::builder()
.batch_length(1000)
@@ -401,7 +403,7 @@ mod tests {
let mut mock = MockProducerCoreBackend::new();
mock.expect_send_internal()
.times(1)
- .returning(|_, _, _, _| Box::pin(async { Ok(()) }));
+ .returning(|_, _, _, _| Box::pin(async { Ok(no_confirmations()) }));
let bb = BackgroundConfig::builder()
.batch_length(10)
@@ -441,6 +443,7 @@ mod tests {
let mut mock = MockProducerCoreBackend::new();
let error = IggyError::ProducerSendFailed {
failed: Arc::new(vec![dummy_message(1)]),
+ committed: Arc::new(Vec::new()),
cause: Box::new(IggyError::Error),
stream_name: "1".to_string(),
topic_name: "1".to_string(),
diff --git a/core/sdk/src/http/messages.rs b/core/sdk/src/http/messages.rs
index c6ee8869c..90cd4bdd5 100644
--- a/core/sdk/src/http/messages.rs
+++ b/core/sdk/src/http/messages.rs
@@ -19,11 +19,12 @@ 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, SendMessagesResponse,
};
use async_trait::async_trait;
use iggy_common::IggyMessagesBatch;
use iggy_common::MessageClient;
+use iggy_common::SendMessagesConfirmations;
use iggy_common::flush_unsaved_buffer::FlushUnsavedBuffer;
#[async_trait]
@@ -65,20 +66,36 @@ impl MessageClient for HttpClient {
topic_id: &Identifier,
partitioning: &Partitioning,
messages: &mut [IggyMessage],
- ) -> Result<(), IggyError> {
+ ) -> Result<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)?;
+ // The legacy server answers a successful send with 201 and no content
+ // at all. That is not JSON, and it must not read as a decode failure on
+ // a write that already committed: no body means the batch landed with
+ // no offsets reported, which is an empty list.
+ if body.is_empty() {
+ return Ok(SendMessagesResponse {
+ confirmations: Vec::new(),
+ });
+ }
+ let confirmations: SendMessagesConfirmations =
+ serde_json::from_slice(&body).map_err(|_| IggyError::InvalidJsonResponse)?;
+ Ok(SendMessagesResponse::from(confirmations))
}
async fn flush_unsaved_buffer(
@@ -118,3 +135,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::{SendMessagesConfirmations, SendMessagesResponse};
+ use crate::prelude::SendMessagesConfirmationResponse;
+
+ fn parse(json: &str) -> SendMessagesResponse {
+ let confirmations: SendMessagesConfirmations =
+ serde_json::from_str(json).expect("contract sample must parse");
+ SendMessagesResponse::from(confirmations)
+ }
+
+ #[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 d364c08ea..ee0dc06d6 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/Cargo.toml b/core/server-ng/Cargo.toml
index 6299081d7..4b66dc09d 100644
--- a/core/server-ng/Cargo.toml
+++ b/core/server-ng/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "server-ng"
-version = "0.9.0-edge.1"
+version = "0.9.0-edge.2"
edition = "2024"
license = "Apache-2.0"
publish = false
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index a477d1f88..72c238f4c 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -1091,9 +1091,10 @@ async fn handle_get_me<B, MJ, S>(
///
/// Callers must have authenticated the transport already: `vsr_client_id` /
/// `bound_session` come from its bound VSR session. Every failure before
-/// dispatch replies (empty for an unresolvable namespace, a nonzero status
-/// for denials and the exhausted routable wait) so the client fails fast
-/// instead of wedging on a silent drop.
+/// dispatch replies with a nonzero status -- unresolvable namespace,
+/// authorization denial, exhausted routable wait -- so the client fails fast
+/// instead of wedging on a silent drop or reading a status-0 frame as a
+/// committed write.
///
/// `vsr_client_id` keys the consumer-group offset fence (the member id),
/// not the transport id stamped into the partition-op header.
@@ -1120,17 +1121,26 @@ pub(crate) async fn dispatch_partition_request<B, MJ, S>(
) {
Ok(namespace) => namespace,
Err(error) => {
- // A partition op against a stream/topic that no longer
- // resolves (e.g. a consumer's trailing auto-commit racing a
- // `delete_stream`). A silent drop wedges the SDK connection
- // forever; reply empty so the client fails fast instead.
+ // A partition op against a stream/topic that no longer resolves
+ // (e.g. a consumer's trailing auto-commit racing a `delete_stream`,
+ // or an explicit partition id that skipped the client-side
+ // resolve). The op never reached the partition plane, so a status-0
+ // reply would read as a committed ack for work that never happened.
+ // A silent drop is no better: the SDK connection processes replies
+ // in lockstep and would wedge forever.
warn!(
transport_client_id,
error = %error,
operation = ?header.operation,
- "partition request with unresolved namespace; replying empty"
+ "partition request with unresolved namespace; replying denied"
);
- send_empty_partition_reply(shard, transport_client_id, &header).await;
+ send_partition_deny_reply(
+ shard,
+ transport_client_id,
+ &header,
+ IggyError::ResourceNotFound(String::new()).as_code(),
+ )
+ .await;
return;
}
};
@@ -1940,10 +1950,10 @@ async fn handle_sync_consumer_group<B, MJ, S>(
.await;
}
-/// Ack a partition op whose namespace does not resolve (deleted stream /
-/// topic or unknown consumer group) with an empty Reply. The SDK connection
-/// processes replies in lockstep, so a silent drop wedges every
-/// subsequent request on that connection.
+/// Ack a consumer-offset op whose body could not be rewritten for the
+/// partition plane with an empty Reply. The SDK connection processes replies
+/// in lockstep, so a silent drop wedges every subsequent request on that
+/// connection.
#[allow(clippy::future_not_send)]
async fn send_empty_partition_reply<B, MJ, S>(
shard: &Rc<ShellShard<B, MJ, S>>,
diff --git a/core/server-ng/src/dispatch/authz.rs b/core/server-ng/src/dispatch/authz.rs
index 57ce91add..f6f30f6ec 100644
--- a/core/server-ng/src/dispatch/authz.rs
+++ b/core/server-ng/src/dispatch/authz.rs
@@ -127,11 +127,12 @@ where
decision.err().map(|error| error.as_code())
}
-/// Reply to a denied or transiently unroutable partition op with the op's
-/// frame: empty body + nonzero `status`. Distinct from
-/// `send_empty_partition_reply` (status 0, the unresolvable-namespace ack);
-/// here the SDK peeks the status and surfaces the typed error. Same lockstep
-/// reasoning: a silent drop would wedge every later request on the connection.
+/// Reply to a partition op rejected before it reached the plane with the op's
+/// frame: empty body + nonzero `status`. The nonzero status is the whole
+/// point: the SDK peeks it and surfaces the typed error, whereas a status-0
+/// frame reads as a committed ack for work that never happened. Silence is no
+/// better, the connection decodes replies in lockstep and would wedge on every
+/// later request.
#[allow(clippy::future_not_send)]
pub(super) async fn send_partition_deny_reply<B, MJ, S>(
shard: &Rc<ShellShard<B, MJ, S>>,
diff --git a/core/server-ng/src/http/error.rs b/core/server-ng/src/http/error.rs
index 6ecb4e70d..e5f9a1844 100644
--- a/core/server-ng/src/http/error.rs
+++ b/core/server-ng/src/http/error.rs
@@ -254,10 +254,10 @@ impl IntoResponse for WriteError {
/// the `PUT`/`DELETE .../consumer-offsets` writes).
///
/// Split differently from [`WriteError`] because the partition plane replies
-/// carry no committed error code: a pre-dispatch gate failure is an empty
-/// reply distinguishable only by header (see [`classify_partition_reply`]),
-/// and an unanswered write is a distinct outcome the caller must treat as
-/// unknown rather than failed.
+/// carry no committed error code: a pre-dispatch gate failure is an
+/// empty-bodied reply that names itself only in the header (see
+/// [`classify_partition_reply`]), and an unanswered write is a distinct
+/// outcome the caller must treat as unknown rather than failed.
#[derive(Debug)]
pub(in crate::http) enum PartitionWriteError {
/// Caller-side rejection (bad identifier, oversized batch, an authorization
@@ -265,9 +265,12 @@ pub(in crate::http) enum PartitionWriteError {
/// (`ReplyHeader.status`), or a malformed reply frame, rendered through the
/// legacy `IggyError -> status` map for SDK-identical bodies.
Rejected(IggyError),
- /// The dispatch gates could not route the write: the stream, topic, or
- /// partition does not resolve (or never materialised within the routable
- /// budget). Rendered as the legacy 404 body.
+ /// Backstop for a status-0 reply carrying `op` 0: an ack with no commit
+ /// number behind it, for a write that never reached the partition plane.
+ /// Routing failures name themselves through `ReplyHeader.status`, so this
+ /// shape is left to a peer that still answers a non-committing op this
+ /// way. Rendered as the legacy 404 body: the alternative is grading a
+ /// write that never happened as a success.
NotFound,
/// The in-process reply slot could not be installed. Transient server
/// condition -> the shared 503, retryable.
diff --git a/core/server-ng/src/http/handlers.rs b/core/server-ng/src/http/handlers.rs
index c2a90eab9..5d3dfd90a 100644
--- a/core/server-ng/src/http/handlers.rs
+++ b/core/server-ng/src/http/handlers.rs
@@ -98,8 +98,8 @@ use iggy_common::wire_conversions::{
use iggy_common::{
ClientInfo, ClientInfoDetails, ClusterMetadata, Consumer, ConsumerGroup, ConsumerGroupDetails,
ConsumerOffsetInfo, Identifier, IdentityInfo, IggyError, PersonalAccessTokenInfo, PollMessages,
- PolledMessages, RawPersonalAccessToken, SendMessages, Stats, Stream, StreamDetails, TokenInfo,
- Topic, TopicDetails, UserInfo, UserInfoDetails, Validatable,
+ PolledMessages, RawPersonalAccessToken, SendMessages, SendMessagesConfirmations, Stats, Stream,
+ StreamDetails, TokenInfo, Topic, TopicDetails, UserInfo, UserInfoDetails, Validatable,
};
use metadata::impls::metadata::StreamsFrontend;
use metadata::permissioner::Permissioner;
@@ -121,7 +121,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::{
@@ -1127,8 +1127,13 @@ 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.
+///
+/// Every 201 carries a `confirmations` list, empty when the commit reported no
+/// offsets. One shape for one meaning: a caller that had to tell "no body"
+/// apart from "empty list" would be reading the same outcome two ways.
pub(in crate::http) async fn send_messages(
State(state): State<HttpState>,
identity: Authenticated,
@@ -1155,21 +1160,23 @@ pub(in crate::http) async fn send_messages(
.map_err(PartitionWriteError::Rejected)?;
match query.ack {
ProduceAck::Replicated => {
- SendWrapper::new(partition_write_replicated(
+ let (reply, header) = 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.
+ let confirmations = send_confirmations(&reply, &header)
+ .map(SendMessagesConfirmations::from)
+ .unwrap_or_default();
+ Ok((StatusCode::CREATED, durability, Json(confirmations)).into_response())
}
ProduceAck::None => {
SendWrapper::new(produce_unacked(&state, &identity.session, &body)).await?;
@@ -1580,3 +1587,46 @@ fn issue_identity(inner: &HttpInner, user_id: u32) -> Result<Json<IdentityInfo>,
}),
}))
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ use iggy_binary_protocol::responses::messages::{
+ SendMessagesConfirmationResponse, SendMessagesResponse,
+ };
+
+ /// 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 renders the envelope with an empty
+ /// list. Together with [`send_messages`] answering `Json` on every acked
+ /// produce - the unreadable-confirmation path included - that is what makes
+ /// `confirmations` present on every 201 body.
+ #[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..d039c64c7 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,24 +35,33 @@ 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, and that order is load-bearing: every pre-commit
+/// denial names itself there (a namespace that does not resolve or is not
+/// routable, a dispatch-time authorization denial, the partition primary's
+/// delete-of-missing-offset rejection), and each renders through the legacy
+/// `IggyError -> status` map, which is what keeps an unresolvable namespace a
+/// 404. Only with status 0 does `op` split the rest: a committed reply is
+/// built from its prepare header, whose op (the partition group's commit
+/// number) is always >= 1, so a 0 there is an ack with nothing committed
+/// behind it and must not grade as success.
+///
+/// Grading only, and it hands the graded header back: the committed body is
+/// read separately (see [`send_confirmations`]) because it is
+/// operation-specific - consumer-offset writes commit empty, a produce commits
+/// its confirmations - and reading it needs the header's `size`. Returning it
+/// keeps that a single parse whose failure mode is already decided here, rather
+/// than a second one whose fallback would have to invent a body.
pub(in crate::http) fn classify_partition_reply(
reply: &BusMessage,
-) -> Result<(), PartitionWriteError> {
+) -> Result<ReplyHeader, PartitionWriteError> {
let header = reply
.as_slice()
.get(..HEADER_SIZE)
@@ -68,7 +78,48 @@ pub(in crate::http) fn classify_partition_reply(
if header.op == 0 {
return Err(PartitionWriteError::NotFound);
}
- Ok(())
+ Ok(*header)
+}
+
+/// 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 an empty confirmation list rather than failing a write that already
+/// landed.
+///
+/// `header` is the one [`classify_partition_reply`] graded, so this reads the
+/// body without re-deriving its extent.
+pub(in crate::http) fn send_confirmations(
+ reply: &BusMessage,
+ header: &ReplyHeader,
+) -> Option<SendMessagesResponse> {
+ let body = partition_reply_body(reply, header);
+ 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<'a>(reply: &'a BusMessage, header: &ReplyHeader) -> &'a [u8] {
+ reply
+ .as_slice()
+ .get(HEADER_SIZE..header.size as usize)
+ .unwrap_or_default()
}
/// Classify a committed reply's leading result section and return the typed
@@ -216,6 +267,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,
@@ -343,8 +396,12 @@ mod tests {
assert!(classify_partition_reply(&reply).is_ok());
}
+ /// An ack with no commit number behind it: status 0 and `op` 0. A routing
+ /// failure names itself in `status`, so whatever emits this shape never
+ /// reached the partition plane, and grading it as success would report a
+ /// write that never happened.
#[test]
- fn gate_failure_empty_reply_classifies_as_not_found() {
+ fn status_zero_op_zero_ack_classifies_as_not_found() {
let request = build_request_message(Operation::SendMessages, 42, 7, 1, &[]);
let reply = frozen(build_empty_reply(request.header(), 42, 0, 9));
assert!(matches!(
@@ -429,6 +486,69 @@ 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))
+ }
+
+ /// Walks the same order the write path does: grade first, then read the
+ /// body with the header that grading returned.
+ fn graded_send_confirmations(body: &Bytes) -> Option<SendMessagesResponse> {
+ let reply = send_reply(body);
+ let header = classify_partition_reply(&reply).expect("committed send reply grades ok");
+ send_confirmations(&reply, &header)
+ }
+
+ #[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!(
+ graded_send_confirmations(&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!(graded_send_confirmations(&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 201 with an empty confirmation list.
+ #[test]
+ fn empty_commit_body_yields_no_confirmations() {
+ assert_eq!(graded_send_confirmations(&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!(graded_send_confirmations(&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 e50b3373e..07c45b182 100644
--- a/core/server-ng/src/http/submit.rs
+++ b/core/server-ng/src/http/submit.rs
@@ -26,8 +26,9 @@ use bytes::Bytes;
use consensus::MetadataHandle;
use futures::channel::oneshot;
use iggy_binary_protocol::consensus::Command2;
-use iggy_binary_protocol::{GenericHeader, Operation, RequestHeader};
+use iggy_binary_protocol::{GenericHeader, Operation, ReplyHeader, RequestHeader};
use iggy_common::IggyError;
+use message_bus::BusMessage;
use metadata::impls::metadata::StreamsFrontend;
use server_common::Message;
use tracing::warn;
@@ -360,12 +361,16 @@ 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 and the header grading read, so a caller
+/// that renders a committed payload can bound the body without parsing the
+/// header again; the offset writes answer 204 and drop both.
pub(in crate::http) async fn partition_write_replicated(
state: &HttpInner,
session: &HttpSession,
operation: Operation,
body: &[u8],
-) -> Result<(), PartitionWriteError> {
+) -> Result<(BusMessage, ReplyHeader), 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
@@ -409,7 +414,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(|header| (reply, header)),
// 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)),
diff --git a/examples/python/uv.lock b/examples/python/uv.lock
index a33c0238f..6d4c07bee 100644
--- a/examples/python/uv.lock
+++ b/examples/python/uv.lock
@@ -8,7 +8,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.8.1.dev4"
+version = "0.9.0.dev1"
source = { directory = "../../foreign/python" }
[package.metadata]
diff --git a/foreign/cpp/src/client.rs b/foreign/cpp/src/client.rs
index a70181200..353b76479 100644
--- a/foreign/cpp/src/client.rs
+++ b/foreign/cpp/src/client.rs
@@ -240,7 +240,7 @@ impl Client {
partitioning_kind: String,
partitioning_value: Vec<u8>,
messages: Vec<ffi::IggyMessageToSend>,
- ) -> Result<(), String> {
+ ) -> Result<ffi::SendMessagesResponse, String> {
let rust_stream_id = RustIdentifier::try_from(stream_id)
.map_err(|error| format!("Could not send messages: {error}"))?;
let rust_topic_id = RustIdentifier::try_from(topic_id)
@@ -285,7 +285,8 @@ impl Client {
.collect::<Result<Vec<_>, _>>()?;
RUNTIME.block_on(async {
- self.inner
+ let response = self
+ .inner
.send_messages(
&rust_stream_id,
&rust_topic_id,
@@ -294,7 +295,7 @@ impl Client {
)
.await
.map_err(|error| format!("Could not send messages: {error}"))?;
- Ok(())
+ Ok(ffi::SendMessagesResponse::from(response))
})
}
diff --git a/foreign/cpp/src/lib.rs b/foreign/cpp/src/lib.rs
index 0c871e9dd..3be955d58 100644
--- a/foreign/cpp/src/lib.rs
+++ b/foreign/cpp/src/lib.rs
@@ -145,6 +145,31 @@ mod ffi {
messages: Vec<IggyMessagePolled>,
}
+ /// Commit confirmation for one partition written by `send_messages`.
+ struct SendMessagesConfirmation {
+ stream_id: u32,
+ topic_id: u32,
+ partition_id: u32,
+ /// Offset assigned to the first message of the batch in this partition.
+ ///
+ /// Sends are at-least-once: an earlier retry may already have committed
+ /// the same batch at a lower offset, so this never identifies a batch
+ /// uniquely.
+ ///
+ /// A batch is confirmed once it is committed in memory, not once it is
+ /// fsynced. A crash-restart can stamp a later batch with an offset a
+ /// client has already recorded.
+ base_offset: u64,
+ }
+
+ /// Reply to `send_messages`.
+ struct SendMessagesResponse {
+ /// One entry per partition the batch was written to. Empty whenever the
+ /// server reports no offsets, which is every send against the legacy
+ /// server, so call `empty()` before indexing.
+ confirmations: Vec<SendMessagesConfirmation>,
+ }
+
struct StreamDetails {
id: u32,
created_at: u64,
@@ -461,7 +486,7 @@ mod ffi {
partitioning_kind: String,
partitioning_value: Vec<u8>,
messages: Vec<IggyMessageToSend>,
- ) -> Result<()>;
+ ) -> Result<SendMessagesResponse>;
fn flush_unsaved_buffer(
self: &Client,
stream_id: Identifier,
diff --git a/foreign/cpp/src/type_conversion.rs b/foreign/cpp/src/type_conversion.rs
index 8a47f7c40..3a8ae3ae0 100644
--- a/foreign/cpp/src/type_conversion.rs
+++ b/foreign/cpp/src/type_conversion.rs
@@ -20,8 +20,11 @@ use bytes::Bytes;
use iggy::prelude::{
ConsumerGroupDetails as RustConsumerGroupDetails, IdKind, Identifier as RustIdentifier,
IggyMessage as RustIggyMessage, Partition as RustPartition,
- PolledMessages as RustPolledMessages, Stream as RustStream, StreamDetails as RustStreamDetails,
- Topic as RustTopic, TopicDetails as RustTopicDetails, Validatable,
+ PolledMessages as RustPolledMessages,
+ SendMessagesConfirmationResponse as RustSendMessagesConfirmationResponse,
+ SendMessagesResponse as RustSendMessagesResponse, Stream as RustStream,
+ StreamDetails as RustStreamDetails, Topic as RustTopic, TopicDetails as RustTopicDetails,
+ Validatable,
};
use iggy_binary_protocol::WireUserHeaders;
use iggy_common::{
@@ -643,3 +646,26 @@ impl From<RustPolledMessages> for ffi::PolledMessages {
}
}
}
+
+impl From<RustSendMessagesConfirmationResponse> for ffi::SendMessagesConfirmation {
+ fn from(confirmation: RustSendMessagesConfirmationResponse) -> Self {
+ ffi::SendMessagesConfirmation {
+ stream_id: confirmation.stream_id,
+ topic_id: confirmation.topic_id,
+ partition_id: confirmation.partition_id,
+ base_offset: confirmation.base_offset,
+ }
+ }
+}
+
+impl From<RustSendMessagesResponse> for ffi::SendMessagesResponse {
+ fn from(response: RustSendMessagesResponse) -> Self {
+ ffi::SendMessagesResponse {
+ confirmations: response
+ .confirmations
+ .into_iter()
+ .map(ffi::SendMessagesConfirmation::from)
+ .collect(),
+ }
+ }
+}
diff --git a/foreign/cpp/tests/e2e/message.cpp b/foreign/cpp/tests/e2e/message.cpp
index 28cae33bc..e20ef6008 100644
--- a/foreign/cpp/tests/e2e/message.cpp
+++ b/foreign/cpp/tests/e2e/message.cpp
@@ -47,8 +47,13 @@ TEST_F(LowLevelE2E_Message, SendAndPollMessagesRoundTrip) {
messages.push_back(std::move(msg));
}
- ASSERT_NO_THROW(client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0),
- "partition_id", partition_id_bytes(0), std::move(messages)));
+ iggy::ffi::SendMessagesResponse sent;
+ ASSERT_NO_THROW(sent = client->send_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0),
+ "partition_id", partition_id_bytes(0), std::move(messages)));
+
+ ASSERT_TRUE(sent.confirmations.empty())
+ << "The legacy server reports no offsets, so the confirmation list must stay empty, got "
+ << sent.confirmations.size();
auto polled = client->poll_messages(make_numeric_identifier(stream.id), make_numeric_identifier(0), 0, "consumer",
make_numeric_identifier(1), "offset", 0, 100, false);
diff --git a/foreign/node/package-lock.json b/foreign/node/package-lock.json
index 1c11d0a11..7cf29b8c6 100644
--- a/foreign/node/package-lock.json
+++ b/foreign/node/package-lock.json
@@ -1,12 +1,12 @@
{
"name": "apache-iggy",
- "version": "0.8.1-edge.4",
+ "version": "0.9.0-edge.1",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "apache-iggy",
- "version": "0.8.1-edge.4",
+ "version": "0.9.0-edge.1",
"license": "Apache-2.0",
"dependencies": {
"debug": "4.4.3",
diff --git a/foreign/node/package.json b/foreign/node/package.json
index 682c6989e..62e47122a 100644
--- a/foreign/node/package.json
+++ b/foreign/node/package.json
@@ -1,7 +1,7 @@
{
"name": "apache-iggy",
"type": "module",
- "version": "0.8.1-edge.4",
+ "version": "0.9.0-edge.1",
"description": "Official Apache Iggy NodeJS SDK",
"keywords": [
"iggy",
diff --git a/foreign/node/src/client/client.utils.test.ts b/foreign/node/src/client/client.utils.test.ts
index ed9aa1df2..174f09cf5 100644
--- a/foreign/node/src/client/client.utils.test.ts
+++ b/foreign/node/src/client/client.utils.test.ts
@@ -18,10 +18,9 @@
import { describe, it } from 'node:test';
import assert from 'node:assert/strict';
-import { handleResponse, deserializeVoidResponse, deserializeStatusResponse } from './client.utils.js';
+import { handleResponse, deserializeVoidResponse } from './client.utils.js';
const SUCCESS = 0;
-const ERROR = 1;
describe('handleResponse', () => {
@@ -48,23 +47,3 @@ describe('handleResponse', () => {
});
});
-
-describe('deserializeStatusResponse', () => {
-
- it('returns true when status is SUCCESS and data is empty', () => {
- const r = { status: SUCCESS, length: 0, data: Buffer.alloc(0) };
- assert.equal(deserializeStatusResponse(r), true);
- });
-
- it('returns true when status is SUCCESS and data is non-empty (e.g. SendMessages server payload)', () => {
- // Key difference from deserializeVoidResponse: non-empty data is accepted.
- const r = { status: SUCCESS, length: 4, data: Buffer.from([1, 2, 3, 4]) };
- assert.equal(deserializeStatusResponse(r), true);
- });
-
- it('returns false when status is an error code', () => {
- const r = { status: ERROR, length: 0, data: Buffer.alloc(0) };
- assert.equal(deserializeStatusResponse(r), false);
- });
-
-});
diff --git a/foreign/node/src/client/client.utils.ts b/foreign/node/src/client/client.utils.ts
index 52862258d..bb9a92258 100644
--- a/foreign/node/src/client/client.utils.ts
+++ b/foreign/node/src/client/client.utils.ts
@@ -66,9 +66,6 @@ export const handleResponseTransform = () => new Transform({
export const deserializeVoidResponse =
(r: CommandResponse) => r.status === 0 && r.data.length === 0;
-export const deserializeStatusResponse =
- (r: CommandResponse) => r.status === 0;
-
/** Length of the command code in bytes */
const COMMAND_LENGTH = 4;
diff --git a/foreign/node/src/e2e/tcp.send-message.e2e.ts b/foreign/node/src/e2e/tcp.send-message.e2e.ts
index 7ce501675..e085e1612 100644
--- a/foreign/node/src/e2e/tcp.send-message.e2e.ts
+++ b/foreign/node/src/e2e/tcp.send-message.e2e.ts
@@ -27,6 +27,10 @@ describe('e2e -> message', async () => {
const c = getTestClient();
+ // Only the VSR lane reaches a server that reports offsets. The classic lane
+ // runs against the legacy server, which commits without confirming.
+ const vsr = process.env.IGGY_TEST_PROTOCOL === 'vsr';
+
const streamName = 'e2e-stream-934';
const topicName = 'e2e-topic-832';
const partitionId = 0;
@@ -48,7 +52,13 @@ describe('e2e -> message', async () => {
};
it('e2e -> message::send', async () => {
- assert.ok(await c.message.send(msg));
+ const { confirmations } = await c.message.send(msg);
+ if (!vsr) {
+ assert.equal(confirmations.length, 0);
+ return;
+ }
+ assert.equal(confirmations.length, 1);
+ assert.equal(confirmations[0].partitionId, partitionId);
});
it('e2e -> message::poll/last', async () => {
@@ -162,6 +172,21 @@ describe('e2e -> message', async () => {
assert.deepEqual(offset, { partitionId: 0, currentOffset: 5n, storedOffset: 2n });
});
+ it('e2e -> message::send/next batch', async () => {
+ const { confirmations } = await c.message.send({
+ ...msg,
+ messages: generateMessages(3)
+ });
+ if (!vsr) {
+ assert.equal(confirmations.length, 0);
+ return;
+ }
+ // Landing behind the already committed batch is the part no placeholder
+ // confirmation could reproduce.
+ assert.equal(confirmations.length, 1);
+ assert.equal(confirmations[0].baseOffset, BigInt(msg.messages.length));
+ });
+
it('e2e -> message::cleanup', async () => {
assert.ok(await c.stream.delete({ streamId: streamName }));
assert.ok(await c.session.logout());
diff --git a/foreign/node/src/index.ts b/foreign/node/src/index.ts
index 40266e46d..60146ae1a 100644
--- a/foreign/node/src/index.ts
+++ b/foreign/node/src/index.ts
@@ -23,6 +23,8 @@ export {
Partitioning,
HeaderValue,
HeaderKeyFactory,
+ type SendMessagesConfirmation,
+ type SendMessagesResponse,
} from "./wire/index.js";
export * from "./client/index.js";
diff --git a/foreign/node/src/wire/message/send-messages.command.test.ts b/foreign/node/src/wire/message/send-messages.command.test.ts
index c4df8e5d3..ac35b9c14 100644
--- a/foreign/node/src/wire/message/send-messages.command.test.ts
+++ b/foreign/node/src/wire/message/send-messages.command.test.ts
@@ -19,11 +19,45 @@
import { describe, it } from "node:test";
import assert from "node:assert/strict";
import { uuidv7, uuidv4 } from "uuidv7";
-import { SEND_MESSAGES, type SendMessages } from "./send-messages.command.js";
+import {
+ SEND_MESSAGES,
+ type SendMessages,
+ type SendMessagesConfirmation,
+} from "./send-messages.command.js";
import { HeaderValue, HeaderKeyFactory } from "./header.utils.js";
+import { DeserializeError } from "../error.utils.js";
const SUCCESS = 0;
-const ERROR = 1;
+
+const CONFIRMATION_SIZE = 20;
+
+const confirmation = (partitionId: number): SendMessagesConfirmation => ({
+ streamId: 1,
+ topicId: 2,
+ partitionId,
+ baseOffset: 42n,
+});
+
+const serializeConfirmations = (
+ confirmations: SendMessagesConfirmation[],
+): Buffer => {
+ const b = Buffer.allocUnsafe(4 + confirmations.length * CONFIRMATION_SIZE);
+ b.writeUInt32LE(confirmations.length, 0);
+ confirmations.forEach((c, index) => {
+ const at = 4 + index * CONFIRMATION_SIZE;
+ b.writeUInt32LE(c.streamId, at);
+ b.writeUInt32LE(c.topicId, at + 4);
+ b.writeUInt32LE(c.partitionId, at + 8);
+ b.writeBigUInt64LE(c.baseOffset, at + 12);
+ });
+ return b;
+};
+
+const response = (data: Buffer) => ({
+ status: SUCCESS,
+ length: data.length,
+ data,
+});
describe("SendMessages", () => {
describe("serialize", () => {
@@ -164,21 +198,71 @@ describe("SendMessages", () => {
describe('deserialize', () => {
- it('returns true when status is SUCCESS with empty data', () => {
- const r = { status: SUCCESS, length: 0, data: Buffer.alloc(0) };
- assert.equal(SEND_MESSAGES.deserialize(r), true);
+ it('reads one confirmation', () => {
+ const confirmations = [confirmation(3)];
+ const r = response(serializeConfirmations(confirmations));
+ assert.deepEqual(SEND_MESSAGES.deserialize(r), { confirmations });
+ });
+
+ it('reads every confirmation of a multi-partition send', () => {
+ const confirmations = [confirmation(0), confirmation(1), confirmation(2)];
+ const r = response(serializeConfirmations(confirmations));
+ assert.deepEqual(SEND_MESSAGES.deserialize(r), { confirmations });
+ });
+
+ it('reads the wire layout of a confirmation', () => {
+ const r = response(Buffer.from([
+ 0x01, 0x00, 0x00, 0x00, // count
+ 0x01, 0x00, 0x00, 0x00, // streamId
+ 0x02, 0x00, 0x00, 0x00, // topicId
+ 0x03, 0x00, 0x00, 0x00, // partitionId
+ 0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // baseOffset
+ ]));
+ assert.deepEqual(SEND_MESSAGES.deserialize(r), {
+ confirmations: [
+ { streamId: 1, topicId: 2, partitionId: 3, baseOffset: 4n }
+ ]
+ });
+ });
+
+ it('reads a committed send that reports no offsets as an empty list', () => {
+ const r = response(serializeConfirmations([]));
+ assert.deepEqual(SEND_MESSAGES.deserialize(r), { confirmations: [] });
+ });
+
+ it('reads the bodiless legacy server reply as an empty list', () => {
+ const r = response(Buffer.alloc(0));
+ assert.deepEqual(SEND_MESSAGES.deserialize(r), { confirmations: [] });
+ });
+
+ it('throws on a truncated body', () => {
+ const data = serializeConfirmations([confirmation(0), confirmation(1)]);
+ for (let i = 1; i < data.length; i += 1)
+ assert.throws(
+ () => SEND_MESSAGES.deserialize(response(data.subarray(0, i))),
+ DeserializeError,
+ `expected error for truncation at byte ${i}`
+ );
});
- it('returns true when status is SUCCESS with non-empty server payload', () => {
- // SendMessages server response includes data (e.g. partition/offset info).
- // The deserializer must accept non-empty data unlike deserializeVoidResponse.
- const r = { status: SUCCESS, length: 4, data: Buffer.from([1, 2, 3, 4]) };
- assert.equal(SEND_MESSAGES.deserialize(r), true);
+ it('throws on trailing bytes', () => {
+ const data = Buffer.concat([
+ serializeConfirmations([confirmation(1)]),
+ Buffer.from([0xFF])
+ ]);
+ assert.throws(
+ () => SEND_MESSAGES.deserialize(response(data)),
+ DeserializeError
+ );
});
- it('returns false when status is an error code', () => {
- const r = { status: ERROR, length: 0, data: Buffer.alloc(0) };
- assert.equal(SEND_MESSAGES.deserialize(r), false);
+ it('throws on a count no body could hold', () => {
+ const data = Buffer.alloc(4);
+ data.writeUInt32LE(0xFFFF_FFFF, 0);
+ assert.throws(
+ () => SEND_MESSAGES.deserialize(response(data)),
+ DeserializeError
+ );
});
});
diff --git a/foreign/node/src/wire/message/send-messages.command.ts b/foreign/node/src/wire/message/send-messages.command.ts
index a8ad20e9b..f5cc3118d 100644
--- a/foreign/node/src/wire/message/send-messages.command.ts
+++ b/foreign/node/src/wire/message/send-messages.command.ts
@@ -19,10 +19,20 @@
import { type Id } from '../identifier.utils.js';
import { serializeSendMessages, type CreateMessage } from './message.utils.js';
import type { Partitioning } from './partitioning.utils.js';
-import {deserializeStatusResponse} from '../../client/client.utils.js';
+import type { CommandResponse } from '../../client/client.type.js';
+import { DeserializeError } from '../error.utils.js';
import { wrapCommand } from '../command.utils.js';
import { COMMAND_CODE } from '../command.code.js';
+/** Size of the confirmation count prefixing the list. */
+const CONFIRMATIONS_COUNT_SIZE = 4;
+
+/**
+ * Size of one confirmation entry:
+ * `stream_id(4) + topic_id(4) + partition_id(4) + base_offset(8)`.
+ */
+const CONFIRMATION_SIZE = 20;
+
/**
* Parameters for the send messages command.
*/
@@ -37,6 +47,67 @@ export type SendMessages = {
partition?: Partitioning,
};
+/** Commit confirmation for one partition written by a send. */
+export type SendMessagesConfirmation = {
+ /** Numeric id of the stream the batch was written to */
+ streamId: number,
+ /** Numeric id of the topic the batch was written to */
+ topicId: number,
+ /** Partition the batch was written to */
+ partitionId: number,
+ /**
+ * Offset assigned to the first message of the batch in that partition.
+ *
+ * Delivery is at-least-once, so an earlier retry of the same batch may
+ * already have committed at a lower offset: this never identifies a batch
+ * uniquely. A batch is confirmed once it is committed in memory, not once it
+ * is fsynced, so a crash-restart can stamp a later batch with an offset a
+ * client has already recorded.
+ */
+ baseOffset: bigint,
+};
+
+/**
+ * Outcome of a successful send, one confirmation per written partition. The
+ * legacy server returns an empty list: it commits without reporting offsets.
+ */
+export type SendMessagesResponse = {
+ /** Commit confirmations, one per partition the batch was written to */
+ confirmations: SendMessagesConfirmation[],
+};
+
+/**
+ * Decodes the reply body of a send: `[confirmations_count:4]` then that many
+ * `[stream_id:4 topic_id:4 partition_id:4 base_offset:8]` entries.
+ *
+ * The legacy server reports a commit by sending no body at all, so absence
+ * decodes to no confirmations instead of surfacing as a decode failure.
+ */
+const deserializeSendMessages = (data: Buffer): SendMessagesResponse => {
+ if (data.length === 0) return { confirmations: [] };
+ if (data.length < CONFIRMATIONS_COUNT_SIZE)
+ throw new DeserializeError('send messages confirmation count is truncated');
+
+ const count = data.readUInt32LE(0);
+ const expected = CONFIRMATIONS_COUNT_SIZE + count * CONFIRMATION_SIZE;
+ if (expected > data.length)
+ throw new DeserializeError('send messages confirmation list is truncated');
+ if (expected !== data.length)
+ throw new DeserializeError('send messages confirmations have trailing bytes');
+
+ const confirmations = new Array<SendMessagesConfirmation>(count);
+ for (let index = 0; index < count; index += 1) {
+ const at = CONFIRMATIONS_COUNT_SIZE + index * CONFIRMATION_SIZE;
+ confirmations[index] = {
+ streamId: data.readUInt32LE(at),
+ topicId: data.readUInt32LE(at + 4),
+ partitionId: data.readUInt32LE(at + 8),
+ baseOffset: data.readBigUInt64LE(at + 12)
+ };
+ }
+ return { confirmations };
+};
+
/**
* Send messages command definition.
* Publishes messages to a topic.
@@ -48,10 +119,12 @@ export const SEND_MESSAGES = {
return serializeSendMessages(streamId, topicId, messages, partition);
},
- deserialize: deserializeStatusResponse
+ deserialize: (r: CommandResponse) => deserializeSendMessages(r.data)
};
/**
- * Executable send messages command function.
+ * Executable send messages command function. Resolves to the commit
+ * confirmations of the written partitions, empty against the legacy server.
*/
-export const sendMessages = wrapCommand<SendMessages, boolean>(SEND_MESSAGES);
+export const sendMessages =
+ wrapCommand<SendMessages, SendMessagesResponse>(SEND_MESSAGES);
diff --git a/foreign/node/src/wire/vsr/register.ts b/foreign/node/src/wire/vsr/register.ts
index ba8a198ec..11e102738 100644
--- a/foreign/node/src/wire/vsr/register.ts
+++ b/foreign/node/src/wire/vsr/register.ts
@@ -28,10 +28,10 @@ import { DeserializeError } from '../error.utils.js';
/**
* Packed protocol semver of the wire contract this port implements,
- * `pack(0, 10, 3)` per `core/binary_protocol/src/version.rs`. Bump together
+ * `pack(0, 11, 0)` per `core/binary_protocol/src/version.rs`. Bump together
* with the Rust `IGGY_PROTOCOL_VERSION` on any wire-incompatible change.
*/
-export const IGGY_PROTOCOL_VERSION = (0 << 20) | (10 << 10) | 3;
+export const IGGY_PROTOCOL_VERSION = (0 << 20) | (11 << 10) | 0;
const SDK_NAME = 'node-sdk';
diff --git a/foreign/php/iggy-php.stubs.php b/foreign/php/iggy-php.stubs.php
index 783299c24..b8cf52eaa 100644
--- a/foreign/php/iggy-php.stubs.php
+++ b/foreign/php/iggy-php.stubs.php
@@ -218,15 +218,17 @@ namespace Iggy {
public function sendBinaryRequest(int $code, string $payload): string {}
/**
- * Sends messages to a topic.
+ * Sends messages to a topic and returns the commit confirmations.
+ *
+ * The list is empty against the legacy server, which reports no offsets.
*
* @param mixed $stream
* @param mixed $topic
* @param int $partition_id
* @param array $messages
- * @return void
+ * @return \Iggy\SendMessagesResponse
*/
- public function sendMessages(mixed $stream, mixed $topic, int $partition_id, array $messages): void {}
+ public function sendMessages(mixed $stream, mixed $topic, int $partition_id, array $messages): \Iggy\SendMessagesResponse {}
}
/**
@@ -497,6 +499,57 @@ namespace Iggy {
public function __construct(string $data) {}
}
+ /**
+ * A PHP class representing where one partition's batch was committed.
+ */
+ class SendMessagesConfirmation {
+ /**
+ * The offset assigned to the first message of the batch in this partition.
+ *
+ * Delivery is at-least-once, so an earlier retry may already have committed the
+ * same batch at a lower offset. The value never implies uniqueness.
+ *
+ * A batch is confirmed once it is committed in memory, not once it is fsynced. A
+ * crash-restart can stamp a later batch with an offset a client has already
+ * recorded.
+ *
+ * The legacy server returns an empty confirmation list, so it reports no offset
+ * at all.
+ *
+ * @var int
+ */
+ public readonly int $base_offset;
+
+ public readonly int $partition_id;
+
+ public readonly int $stream_id;
+
+ public readonly int $topic_id;
+
+ public function __construct() {}
+ }
+
+ /**
+ * A PHP class representing the commit confirmations of a send.
+ */
+ class SendMessagesResponse {
+ /**
+ * One confirmation per partition the batch landed in.
+ *
+ * The list is empty when the server reports no offsets. The legacy server never
+ * reports any, and a server that does can still commit a batch it has no offsets
+ * to describe, so check for an empty array instead of indexing.
+ *
+ * The confirmations are rebuilt on each getter call; cache the result in PHP if
+ * they will be read repeatedly.
+ *
+ * @var array
+ */
+ public readonly array $confirmations;
+
+ public function __construct() {}
+ }
+
class StreamDetails {
public readonly int $id;
diff --git a/foreign/php/src/client.rs b/foreign/php/src/client.rs
index 054d35182..57e54d151 100644
--- a/foreign/php/src/client.rs
+++ b/foreign/php/src/client.rs
@@ -31,7 +31,7 @@ use crate::error::to_php_exception;
use crate::identifier::PhpIdentifier;
use crate::receive_message::{PollingStrategy, ReceiveMessage};
use crate::runtime::runtime;
-use crate::send_message::SendMessage;
+use crate::send_message::{SendMessage, SendMessagesResponse};
use crate::stream::StreamDetails;
use crate::topic::TopicDetails;
@@ -184,14 +184,16 @@ impl IggyClient {
})
}
- /// Sends messages to a topic.
+ /// Sends messages to a topic and returns the commit confirmations.
+ ///
+ /// The list is empty against the legacy server, which reports no offsets.
pub fn send_messages(
&self,
stream: PhpIdentifier,
topic: PhpIdentifier,
partition_id: u32,
messages: Vec<&SendMessage>,
- ) -> PhpResult {
+ ) -> PhpResult<SendMessagesResponse> {
let stream: Identifier = stream.try_into()?;
let topic: Identifier = topic.try_into()?;
let partitioning = Partitioning::partition_id(partition_id);
@@ -205,6 +207,7 @@ impl IggyClient {
inner
.send_messages(&stream, &topic, &partitioning, messages.as_mut())
.await
+ .map(SendMessagesResponse::from)
.map_err(to_php_exception)
})
}
diff --git a/foreign/php/src/lib.rs b/foreign/php/src/lib.rs
index 426e30955..49b944225 100644
--- a/foreign/php/src/lib.rs
+++ b/foreign/php/src/lib.rs
@@ -36,7 +36,7 @@ use crate::error::{
};
use crate::message_iterator::MessageIterator;
use crate::receive_message::{PollingStrategy, ReceiveMessage};
-use crate::send_message::SendMessage;
+use crate::send_message::{SendMessage, SendMessagesConfirmation, SendMessagesResponse};
use crate::stream::StreamDetails;
use crate::topic::TopicDetails;
@@ -58,6 +58,8 @@ pub fn get_module(module: ModuleBuilder) -> ModuleBuilder {
.class::<PollingStrategy>()
.class::<ReceiveMessage>()
.class::<SendMessage>()
+ .class::<SendMessagesConfirmation>()
+ .class::<SendMessagesResponse>()
.class::<StreamDetails>()
.class::<TopicDetails>()
}
diff --git a/foreign/php/src/message_iterator.rs b/foreign/php/src/message_iterator.rs
index da19f7418..7abed6139 100644
--- a/foreign/php/src/message_iterator.rs
+++ b/foreign/php/src/message_iterator.rs
@@ -76,6 +76,9 @@ impl MessageIterator {
self.key
}
+ // PHP's \Iterator contract fixes this method name, and PhpResult cannot
+ // satisfy Iterator::next, so the standard trait is not implementable here.
+ #[allow(clippy::should_implement_trait)]
pub fn next(&mut self) -> PhpResult {
if self.current.is_some() {
self.key += 1;
diff --git a/foreign/php/src/send_message.rs b/foreign/php/src/send_message.rs
index ab4c3748e..cabbf2b8f 100644
--- a/foreign/php/src/send_message.rs
+++ b/foreign/php/src/send_message.rs
@@ -17,7 +17,11 @@
use bytes::Bytes;
use ext_php_rs::{binary::Binary, exception::PhpResult, php_class, php_impl};
-use iggy::prelude::{IggyMessage as RustIggyMessage, IggyMessageHeader};
+use iggy::prelude::{
+ IggyMessage as RustIggyMessage, IggyMessageHeader,
+ SendMessagesConfirmationResponse as RustSendMessagesConfirmationResponse,
+ SendMessagesResponse as RustSendMessagesResponse,
+};
use crate::error::to_php_exception;
@@ -77,3 +81,84 @@ impl SendMessage {
Binary::new(self.inner.payload.to_vec())
}
}
+
+/// A PHP class representing the commit confirmations of a send.
+#[php_class]
+#[php(name = "Iggy\\SendMessagesResponse")]
+pub struct SendMessagesResponse {
+ pub(crate) inner: RustSendMessagesResponse,
+}
+
+impl From<RustSendMessagesResponse> for SendMessagesResponse {
+ fn from(inner: RustSendMessagesResponse) -> Self {
+ Self { inner }
+ }
+}
+
+#[php_impl]
+impl SendMessagesResponse {
+ /// One confirmation per partition the batch landed in.
+ ///
+ /// The list is empty when the server reports no offsets. The legacy server never
+ /// reports any, and a server that does can still commit a batch it has no offsets
+ /// to describe, so check for an empty array instead of indexing.
+ ///
+ /// The confirmations are rebuilt on each getter call; cache the result in PHP if
+ /// they will be read repeatedly.
+ #[php(getter)]
+ pub fn confirmations(&self) -> Vec<SendMessagesConfirmation> {
+ self.inner
+ .confirmations
+ .iter()
+ .cloned()
+ .map(SendMessagesConfirmation::from)
+ .collect()
+ }
+}
+
+/// A PHP class representing where one partition's batch was committed.
+#[php_class]
+#[php(name = "Iggy\\SendMessagesConfirmation")]
+pub struct SendMessagesConfirmation {
+ pub(crate) inner: RustSendMessagesConfirmationResponse,
+}
+
+impl From<RustSendMessagesConfirmationResponse> for SendMessagesConfirmation {
+ fn from(inner: RustSendMessagesConfirmationResponse) -> Self {
+ Self { inner }
+ }
+}
+
+#[php_impl]
+impl SendMessagesConfirmation {
+ #[php(getter)]
+ pub fn stream_id(&self) -> u32 {
+ self.inner.stream_id
+ }
+
+ #[php(getter)]
+ pub fn topic_id(&self) -> u32 {
+ self.inner.topic_id
+ }
+
+ #[php(getter)]
+ pub fn partition_id(&self) -> u32 {
+ self.inner.partition_id
+ }
+
+ /// The offset assigned to the first message of the batch in this partition.
+ ///
+ /// Delivery is at-least-once, so an earlier retry may already have committed the
+ /// same batch at a lower offset. The value never implies uniqueness.
+ ///
+ /// A batch is confirmed once it is committed in memory, not once it is fsynced. A
+ /// crash-restart can stamp a later batch with an offset a client has already
+ /// recorded.
+ ///
+ /// The legacy server returns an empty confirmation list, so it reports no offset
+ /// at all.
+ #[php(getter)]
+ pub fn base_offset(&self) -> u64 {
+ self.inner.base_offset
+ }
+}
diff --git a/foreign/php/tests/IggySdkTest.php b/foreign/php/tests/IggySdkTest.php
index 5a8e561f0..978a6f7fb 100644
--- a/foreign/php/tests/IggySdkTest.php
+++ b/foreign/php/tests/IggySdkTest.php
@@ -23,6 +23,7 @@ use Iggy\Client as IggyClient;
use Iggy\PollingStrategy;
use Iggy\ReceiveMessage;
use Iggy\SendMessage;
+use Iggy\SendMessagesResponse;
use PHPUnit\Framework\Attributes\TestDox;
use PHPUnit\Framework\TestCase;
@@ -221,6 +222,25 @@ final class IggySdkTest extends TestCase
}
}
+ #[TestDox('sendMessages against the legacy server reports no commit confirmations')]
+ public function testSendMessagesReportsNoConfirmationFromLegacyServer(): void
+ {
+ $client = new_client();
+ $streamName = unique_name('confirm-stream');
+ $topicName = unique_name('confirm-topic');
+ $partitionId = 0;
+
+ try {
+ create_stream_and_topic($client, $streamName, $topicName);
+
+ $response = $client->sendMessages($streamName, $topicName, $partitionId, [new SendMessage('confirm-first')]);
+ assert_instance_of(SendMessagesResponse::class, $response);
+ assert_count(0, $response->confirmations, 'the legacy server answers a send with no offsets');
+ } finally {
+ cleanup_stream_with_topics($client, $streamName, [$topicName]);
+ }
+ }
+
#[TestDox('Received messages expose payload and metadata')]
public function testMessageProperties(): void
{
diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml
index cd9c4eda9..322abf6bf 100644
--- a/foreign/python/Cargo.toml
+++ b/foreign/python/Cargo.toml
@@ -17,7 +17,7 @@
[package]
name = "apache-iggy"
-version = "0.8.1-dev4"
+version = "0.9.0-dev1"
edition = "2024"
authors = ["Iggy Committers <de...@iggy.apache.org>"]
license = "Apache-2.0"
@@ -37,7 +37,7 @@ doc = false
[dependencies]
bytes = "1.12.1"
futures = "0.3.33"
-iggy = { path = "../../core/sdk", version = "0.10.3-edge.3" }
+iggy = { path = "../../core/sdk", version = "0.11.0-edge.1" }
paste = "1"
pyo3 = "0.29.0"
pyo3-async-runtimes = { version = "0.29.0", features = [
diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index e10a5ec89..341b3885c 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi
@@ -44,6 +44,8 @@ __all__ = [
"PollingStrategy",
"ReceiveMessage",
"SendMessage",
+ "SendMessagesConfirmation",
+ "SendMessagesResponse",
"StreamDetails",
"StreamPermissions",
"Topic",
@@ -1212,10 +1214,13 @@ class IggyClient:
topic: builtins.str | builtins.int,
partitioning: builtins.int,
messages: list[SendMessage],
- ) -> collections.abc.Awaitable[None]:
+ ) -> collections.abc.Awaitable[SendMessagesResponse]:
r"""
Sends a list of messages to the specified topic.
- Returns Ok(()) on successful sending or a PyRuntimeError on failure.
+ Returns a SendMessagesResponse carrying the per-partition commit
+ confirmations, or a PyRuntimeError on failure. The confirmation list is
+ empty when the server reports no offsets, and the legacy server never
+ reports any.
"""
def poll_messages(
self,
@@ -1606,6 +1611,65 @@ class SendMessage:
directly from Python using the provided string or bytes data.
"""
+@typing.final
+class SendMessagesConfirmation:
+ r"""
+ A Python class representing the commit confirmation for one partition
+ written by a send.
+ """
+ @property
+ def stream_id(self) -> builtins.int:
+ r"""
+ Gets the unique identifier (numeric) of the stream the batch was written to.
+ """
+ @property
+ def topic_id(self) -> builtins.int:
+ r"""
+ Gets the unique identifier (numeric) of the topic the batch was written to.
+ """
+ @property
+ def partition_id(self) -> builtins.int:
+ r"""
+ Gets the identifier of the partition the batch was written to.
+ """
+ @property
+ def base_offset(self) -> builtins.int:
+ r"""
+ Gets the offset assigned to the first message of the batch in this partition.
+
+ The offset locates the batch, it does not identify it. Delivery is
+ at-least-once, so an earlier retry may already have committed these
+ messages at a lower offset.
+
+ A batch is confirmed once it is committed in memory, not once it is
+ fsynced. A crash-restart can stamp a later batch with an offset a client
+ has already recorded.
+
+ The legacy server confirms nothing, so its confirmation list is empty
+ and this value is never reached.
+ """
+
+@typing.final
+class SendMessagesResponse:
+ r"""
+ A Python class representing the outcome of a successful send.
+ """
+ @property
+ def confirmations(self) -> builtins.list[SendMessagesConfirmation]:
+ r"""
+ Gets the commit confirmations, one per partition the batch was written to.
+
+ The list is empty when the server reports no offsets, and the legacy
+ server never reports any, so branch on it being empty rather than
+ indexing into it.
+
+ A reported `base_offset` never implies uniqueness, because delivery is
+ at-least-once and an earlier retry may already have committed the same
+ messages at a lower offset. A batch is confirmed once it is committed in
+ memory, not once it is fsynced. A crash-restart can stamp a later batch
+ with an offset a client has already recorded.
+ """
+
@typing.final
class StreamDetails:
@property
diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml
index 0dd8ad144..df75ea9f0 100644
--- a/foreign/python/pyproject.toml
+++ b/foreign/python/pyproject.toml
@@ -22,7 +22,7 @@ build-backend = "maturin"
[project]
name = "apache-iggy"
requires-python = ">=3.10"
-version = "0.8.1.dev4"
+version = "0.9.0.dev1"
description = "Apache Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
readme = "README.md"
license = { file = "LICENSE" }
diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index 8d3aa1594..c7cbe3952 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs
@@ -37,7 +37,7 @@ use crate::consumer::{
use crate::identifier::PyIdentifier;
use crate::permissions::Permissions as PyPermissions;
use crate::receive_message::{PollingStrategy, ReceiveMessage};
-use crate::send_message::SendMessage;
+use crate::send_message::{SendMessage, SendMessagesResponse as PySendMessagesResponse};
use crate::stream::StreamDetails;
use crate::topic::{IggyExpiry, MaxTopicSize, Topic, TopicDetails};
use crate::user::{
@@ -898,8 +898,11 @@ impl IggyClient {
}
/// Sends a list of messages to the specified topic.
- /// Returns Ok(()) on successful sending or a PyRuntimeError on failure.
- #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[None]", imports=("collections.abc")))]
+ /// Returns a SendMessagesResponse carrying the per-partition commit
+ /// confirmations, or a PyRuntimeError on failure. The confirmation list is
+ /// empty when the server reports no offsets, and the legacy server never
+ /// reports any.
+ #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[SendMessagesResponse]", imports=("collections.abc")))]
fn send_messages<'a>(
&self,
py: Python<'a>,
@@ -926,11 +929,11 @@ impl IggyClient {
let inner = self.inner.clone();
future_into_py(py, async move {
- inner
+ let response = inner
.send_messages(&stream, &topic, &partitioning, messages.as_mut())
.await
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
- Ok(())
+ Ok(PySendMessagesResponse::from(response))
})
}
diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs
index 4f552eadc..985476ff7 100644
--- a/foreign/python/src/lib.rs
+++ b/foreign/python/src/lib.rs
@@ -34,7 +34,7 @@ use consumer::{
use permissions::{GlobalPermissions, Permissions, StreamPermissions, TopicPermissions};
use pyo3::prelude::*;
use receive_message::{PollingStrategy, ReceiveMessage};
-use send_message::SendMessage;
+use send_message::{SendMessage, SendMessagesConfirmation, SendMessagesResponse};
use stream::StreamDetails;
use topic::{IggyExpiry, MaxTopicSize, Partition, Topic, TopicDetails};
use user::{UserInfo, UserInfoDetails, UserStatus};
@@ -44,6 +44,8 @@ use user_headers::{HeaderKey, HeaderValue, UserHeaders};
#[pymodule]
fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<SendMessage>()?;
+ m.add_class::<SendMessagesResponse>()?;
+ m.add_class::<SendMessagesConfirmation>()?;
m.add_class::<ReceiveMessage>()?;
m.add_class::<IggyClient>()?;
m.add_class::<StreamDetails>()?;
diff --git a/foreign/python/src/send_message.rs b/foreign/python/src/send_message.rs
index 6c09eb313..e63b1a2f7 100644
--- a/foreign/python/src/send_message.rs
+++ b/foreign/python/src/send_message.rs
@@ -16,7 +16,11 @@
// under the License.
use bytes::Bytes;
-use iggy::prelude::{IggyMessage as RustIggyMessage, IggyMessageHeader};
+use iggy::prelude::{
+ IggyMessage as RustIggyMessage, IggyMessageHeader,
+ SendMessagesConfirmationResponse as RustSendMessagesConfirmationResponse,
+ SendMessagesResponse as RustSendMessagesResponse,
+};
use pyo3::{exceptions::PyValueError, prelude::*, types::PyBytes};
use pyo3_stub_gen::{
derive::{gen_stub_pyclass, gen_stub_pymethods},
@@ -100,3 +104,95 @@ pub enum PyMessagePayload {
Bytes(Py<PyBytes>),
}
impl_stub_type!(PyMessagePayload = String | PyBytes);
+
+/// A Python class representing the commit confirmation for one partition
+/// written by a send.
+#[pyclass]
+#[gen_stub_pyclass]
+pub struct SendMessagesConfirmation {
+ pub(crate) inner: RustSendMessagesConfirmationResponse,
+}
+
+impl From<&RustSendMessagesConfirmationResponse> for SendMessagesConfirmation {
+ fn from(confirmation: &RustSendMessagesConfirmationResponse) -> Self {
+ Self {
+ inner: confirmation.clone(),
+ }
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl SendMessagesConfirmation {
+ /// Gets the unique identifier (numeric) of the stream the batch was written to.
+ #[getter]
+ pub fn stream_id(&self) -> u32 {
+ self.inner.stream_id
+ }
+
+ /// Gets the unique identifier (numeric) of the topic the batch was written to.
+ #[getter]
+ pub fn topic_id(&self) -> u32 {
+ self.inner.topic_id
+ }
+
+ /// Gets the identifier of the partition the batch was written to.
+ #[getter]
+ pub fn partition_id(&self) -> u32 {
+ self.inner.partition_id
+ }
+
+ /// Gets the offset assigned to the first message of the batch in this partition.
+ ///
+ /// The offset locates the batch, it does not identify it. Delivery is
+ /// at-least-once, so an earlier retry may already have committed these
+ /// messages at a lower offset.
+ ///
+ /// A batch is confirmed once it is committed in memory, not once it is
+ /// fsynced. A crash-restart can stamp a later batch with an offset a client
+ /// has already recorded.
+ ///
+ /// The legacy server confirms nothing, so its confirmation list is empty
+ /// and this value is never reached.
+ #[getter]
+ pub fn base_offset(&self) -> u64 {
+ self.inner.base_offset
+ }
+}
+
+/// A Python class representing the outcome of a successful send.
+#[pyclass]
+#[gen_stub_pyclass]
+pub struct SendMessagesResponse {
+ pub(crate) inner: RustSendMessagesResponse,
+}
+
+impl From<RustSendMessagesResponse> for SendMessagesResponse {
+ fn from(response: RustSendMessagesResponse) -> Self {
+ Self { inner: response }
+ }
+}
+
+#[gen_stub_pymethods]
+#[pymethods]
+impl SendMessagesResponse {
+ /// Gets the commit confirmations, one per partition the batch was written to.
+ ///
+ /// The list is empty when the server reports no offsets, and the legacy
+ /// server never reports any, so branch on it being empty rather than
+ /// indexing into it.
+ ///
+ /// A reported `base_offset` never implies uniqueness, because delivery is
+ /// at-least-once and an earlier retry may already have committed the same
+ /// messages at a lower offset. A batch is confirmed once it is committed in
+ /// memory, not once it is fsynced. A crash-restart can stamp a later batch
+ /// with an offset a client has already recorded.
+ #[getter]
+ pub fn confirmations(&self) -> Vec<SendMessagesConfirmation> {
+ self.inner
+ .confirmations
+ .iter()
+ .map(SendMessagesConfirmation::from)
+ .collect()
+ }
+}
diff --git a/foreign/python/tests/test_message_operations.py b/foreign/python/tests/test_message_operations.py
index 6a1974bf8..f6de418e7 100644
--- a/foreign/python/tests/test_message_operations.py
+++ b/foreign/python/tests/test_message_operations.py
@@ -68,6 +68,46 @@ class TestMessageOperations:
test_messages
)
+ @pytest.mark.asyncio
+ async def test_send_messages_reports_no_confirmations_from_legacy_server(
+ self, iggy_client: IggyClient, unique_name
+ ):
+ """Test the send response confirms nothing without server-side offsets."""
+ stream_name = unique_name()
+ topic_name = unique_name()
+ partition_id = 0
+
+ await iggy_client.create_stream(stream_name)
+ await iggy_client.create_topic(
+ stream=stream_name, name=topic_name, partitions_count=1
+ )
+
+ payloads = [f"Confirmation test {i}" for i in range(3)]
+ response = await iggy_client.send_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partitioning=partition_id,
+ messages=[Message(payload) for payload in payloads],
+ )
+
+ polled_messages = await iggy_client.poll_messages(
+ stream=stream_name,
+ topic=topic_name,
+ partition_id=partition_id,
+ polling_strategy=PollingStrategy.First(),
+ count=10,
+ auto_commit=True,
+ )
+
+ assert [message.payload().decode() for message in polled_messages] == payloads
+
+ # This suite runs against the legacy server, which sends no confirmation
+ # payload at all. A genuine confirmation for this send would read as all
+ # zeros (partition 0, first batch at offset 0), so an entry here proves
+ # nothing and could only have been invented. Against a server that does
+ # report offsets, compare base_offset with polled_messages[0].offset().
+ assert response.confirmations == []
+
@pytest.mark.asyncio
async def test_send_and_poll_messages_as_bytes(
self, iggy_client: IggyClient, unique_name
diff --git a/foreign/python/uv.lock b/foreign/python/uv.lock
index 0ae74df8d..e6f5c8e76 100644
--- a/foreign/python/uv.lock
+++ b/foreign/python/uv.lock
@@ -12,7 +12,7 @@ exclude-newer-span = "P7D"
[[package]]
name = "apache-iggy"
-version = "0.8.1.dev4"
+version = "0.9.0.dev1"
source = { editable = "." }
[package.optional-dependencies]
diff --git a/scripts/ci/license-headers.sh b/scripts/ci/license-headers.sh
index 0f61e12ea..910820b55 100755
--- a/scripts/ci/license-headers.sh
+++ b/scripts/ci/license-headers.sh
@@ -202,7 +202,12 @@ find_duplicate_license_headers() {
continue
fi
- # short-circuit the symlink that may point to dir.
+ # git tracks a symlink as a blob, so it arrives here as an ordinary path.
+ # HawkEye skips symlinks too, and the target is scanned under its own path.
+ if [ -L "$path" ]; then
+ continue
+ fi
+
if [ ! -f "$path" ] || ! LC_ALL=C grep -Iq . "$path"; then
continue
fi