This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch partition_superblock in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 6b5957b5516bb141e31ec87d0cc9c80b07fdc993 Merge: 1a2d35c63 0a23e161b Author: Grzegorz Koszyk <[email protected]> AuthorDate: Wed Aug 5 19:56:17 2026 +0200 Merge branch 'master' into partition_superblock .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 + .github/workflows/coverage-baseline.yml | 69 ++++- 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/coverage-baseline-affected.sh | 96 +++++++ scripts/ci/license-headers.sh | 7 +- 92 files changed, 2384 insertions(+), 318 deletions(-) diff --cc core/partitions/src/iggy_partition.rs index 7cf5f6b14,d7cc7338f..16899e94f --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@@ -5363,168 -4801,68 +5457,226 @@@ mod tests // locally durable nor repaired. partition.repair = Some(armed_session(8, 5, Some(3))); - partition.complete_repair(&repair_config()).await; + let conclusion = partition.complete_repair(&repair_config()).await; + assert_eq!( + conclusion, + RepairConclusion::InProgress, + "with the window incomplete, later frames can still lower the \ + first batch offset into connection" + ); assert_eq!(partition.consensus().commit_min(), 0); + assert!(partition.repair.is_some()); + } + /// Temp partition directory for the state-transfer fence specs below. + async fn transfer_fence_dir(label: &str) -> String { + let dir = std::env::temp_dir().join(format!( + "iggy-transfer-fence-{label}-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock after epoch") + .as_nanos(), + )); + compio::fs::create_dir_all(&dir) + .await + .expect("create temp partition dir"); + dir.to_string_lossy().into_owned() + } + + fn armed_transfer(peer: u8) -> crate::state_transfer::PartitionTransferSession { + crate::state_transfer::PartitionTransferSession { + nonce: 7, + peer, + commit_op: 12, + artifacts: Vec::new(), + target_accepted: true, + idle_ticks: 0, + } + } + + /// A purge must not leave a transfer running: its staged segments hold + /// PRE-purge data, and completing the install renames it back in durably + /// (the install takes `max(offer generation, applied)`, and this purge + /// already stamped the newer one, so the reconciler's purge gate never + /// re-fires). + #[compio::test] + async fn given_armed_transfer_when_purged_should_abandon_session_and_rearm() { + let partition_dir = transfer_fence_dir("purge-abandons").await; + let mut partition = test_partition(); + partition.set_partition_dir(partition_dir.clone()); + partition.transfer = Some(armed_transfer(1)); + partition.transfer_rearm = Some(crate::state_transfer::PendingTransferRearm { + peer: 2, + after_ticks: 5, + }); + partition.consensus().begin_state_transfer_await(); + + partition + .purge(&repair_config(), 3) + .await + .expect("purge partition"); + + assert!( + partition.transfer.is_none(), + "purge must drop the in-flight transfer session" + ); + assert!( + partition.transfer_rearm.is_none(), + "purge must cancel the scheduled re-arm" + ); + assert_eq!( + partition.consensus().state_transfer_stage(), + consensus::StateTransferStage::Idle, + "purge must release the transfer stage so a later trigger can arm" + ); + + let _ = std::fs::remove_dir_all(&partition_dir); + } + + /// An offer whose frontier sits below this replica's own offset counter is + /// refused: installing it would rewind the counter, and the next replicated + /// prepare is re-stamped from it, so this replica would persist different + /// bytes (and a different `batch_checksum`) than the rest of the group. + #[compio::test] + async fn given_offer_below_local_counter_when_installed_should_refuse_rewind() { + let partition_dir = transfer_fence_dir("rewind-refused").await; + let mut partition = test_partition(); + partition.set_partition_dir(partition_dir.clone()); + partition.should_increment_offset = true; + partition.offset.store(99, Ordering::Release); + + let behind = crate::state_transfer::ConsumerOffsetsWire { + purge_generation: 0, + next_offset: 50, + consumers: Vec::new(), + groups: Vec::new(), + }; + let refused = partition + .install_state_transfer(&repair_config(), 12, Vec::new(), &behind.encode()) + .await; + assert!( + matches!( + refused, + Err( + crate::state_transfer::PartitionInstallError::OfferRewindsDurableData { + offer_next_offset: 50, + local_next_offset: 100, + } + ) + ), + "expected a rewind refusal, got {refused:?}" + ); + + // A purge at the origin is the one legitimate rewind, and the artifact + // carries the generation that proves it: the same offer passes the fence + // once its generation advances. + let purged = crate::state_transfer::ConsumerOffsetsWire { + purge_generation: 1, + next_offset: 0, + consumers: Vec::new(), + groups: Vec::new(), + }; + let accepted = partition + .install_state_transfer(&repair_config(), 12, Vec::new(), &purged.encode()) + .await; + assert!( + !matches!( + accepted, + Err(crate::state_transfer::PartitionInstallError::OfferRewindsDurableData { .. }) + ), + "a purge-advancing offer must pass the rewind fence, got {accepted:?}" + ); + + let _ = std::fs::remove_dir_all(&partition_dir); + } + + /// Primary-by-index at view 0 with nothing committed refuses to serve: an + /// empty group is trivially "caught up", so this gate is the only thing + /// separating a real primary from a phantom whose directory vanished, whose + /// zero-segment offer at frontier 0 would make a data-holding receiver + /// unlink its chain. + #[compio::test] + async fn given_nothing_committed_when_offer_requested_should_refuse() { + let partition_dir = transfer_fence_dir("nothing-committed").await; + let mut partition = test_partition(); + partition.set_partition_dir(partition_dir.clone()); + assert_eq!(partition.consensus().commit_max(), 0); + + let refused = partition.state_transfer_offer(&repair_config()).await; + assert!( + matches!( + refused, + Err(crate::state_transfer::PartitionTransferUnavailable::NothingCommitted) + ), + "expected a NothingCommitted refusal, got {refused:?}" + ); + assert!( + refused.is_err_and(|reason| reason.transient()), + "the refusal must be transient: the requester rotates rather than \ + charging its failure count" + ); + + let _ = std::fs::remove_dir_all(&partition_dir); } + + 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 --cc core/server-ng/src/dispatch.rs index 783c531ad,72c238f4c..cbae3f257 --- a/core/server-ng/src/dispatch.rs +++ b/core/server-ng/src/dispatch.rs @@@ -1964,13 -1950,13 +1974,13 @@@ async fn handle_sync_consumer_group<B, .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>>, +async fn send_empty_partition_reply<B, MJ, S, SB>( + shard: &Rc<ShellShard<B, MJ, S, SB>>, transport_client_id: u128, request_header: &RequestHeader, ) where diff --cc core/server-ng/src/dispatch/authz.rs index 9d48d9a46,f6f30f6ec..08ed5cac1 --- a/core/server-ng/src/dispatch/authz.rs +++ b/core/server-ng/src/dispatch/authz.rs @@@ -129,14 -127,15 +129,15 @@@ wher 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>>, +pub(super) async fn send_partition_deny_reply<B, MJ, S, SB>( + shard: &Rc<ShellShard<B, MJ, S, SB>>, transport_client_id: u128, request_header: &RequestHeader, status: u32,
