This is an automated email from the ASF dual-hosted git repository. spetz pushed a commit to branch vsr_raw_request_rust_python in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 094d8fbc85942478fd05845266173c93b3dd6451 Author: spetz <[email protected]> AuthorDate: Mon Jul 27 22:09:56 2026 +0200 feat(sdk): support VSR raw requests in Rust and Python --- .../actions/python-maturin/pre-merge/action.yml | 48 +++- .github/config/components.yml | 2 +- Cargo.lock | 6 +- Cargo.toml | 6 +- bdd/python/uv.lock | 2 +- core/binary_protocol/Cargo.toml | 2 +- core/common/Cargo.toml | 2 +- core/common/src/lib.rs | 1 + core/common/src/traits/binary_transport.rs | 13 +- core/common/src/types/binary_request_kind.rs | 80 +++++++ core/common/src/types/mod.rs | 1 + core/integration/tests/sdk/raw.rs | 185 ++++++++++++--- core/sdk/Cargo.toml | 2 +- core/sdk/src/binary/mod.rs | 23 ++ core/sdk/src/clients/client.rs | 79 +++++-- core/sdk/src/prelude.rs | 12 +- core/sdk/src/quic/quic_client.rs | 166 ++++++++----- core/sdk/src/tcp/tcp_client.rs | 178 +++++++++----- core/sdk/src/vsr.rs | 263 ++++++++++++++++++--- core/sdk/src/websocket/websocket_client.rs | 165 ++++++++----- examples/python/uv.lock | 2 +- foreign/python/Cargo.toml | 7 +- foreign/python/README.md | 30 +++ foreign/python/apache_iggy.pyi | 40 ++++ foreign/python/pyproject.toml | 3 +- foreign/python/src/binary_request_kind.rs | 52 ++++ foreign/python/src/client.rs | 39 ++- foreign/python/src/lib.rs | 3 + foreign/python/tests/test_raw_command.py | 62 ++++- foreign/python/tests/test_raw_command_vsr.py | 70 ++++++ foreign/python/uv.lock | 2 +- 31 files changed, 1262 insertions(+), 284 deletions(-) diff --git a/.github/actions/python-maturin/pre-merge/action.yml b/.github/actions/python-maturin/pre-merge/action.yml index 81d3bd50b..e93c4fc76 100644 --- a/.github/actions/python-maturin/pre-merge/action.yml +++ b/.github/actions/python-maturin/pre-merge/action.yml @@ -20,7 +20,7 @@ description: Python pre-merge testing with maturin github iggy actions inputs: task: - description: "Task to run (lint, test, build)" + description: "Task to run (lint, test, test-vsr, build)" required: true runs: @@ -117,12 +117,31 @@ runs: ls -la dist/ shell: bash + - name: Build VSR Python extension and server + if: inputs.task == 'test-vsr' + run: | + cd foreign/python + uv run --no-sync maturin develop --features vsr + cd ../.. + cargo build --locked --bin iggy-server-ng --features vsr + shell: bash + - name: Start Iggy server if: inputs.task == 'test' id: iggy uses: ./.github/actions/utils/server-start continue-on-error: true + - name: Start VSR Iggy server + if: inputs.task == 'test-vsr' + id: iggy-vsr + uses: ./.github/actions/utils/server-start + with: + mode: bin + bin: ${{ github.workspace }}/target/debug/iggy-server-ng + log-file: ${{ runner.temp }}/iggy-server-vsr.log + pid-file: ${{ runner.temp }}/iggy-server-vsr.pid + - name: Run Python integration tests if: inputs.task == 'test' && steps.iggy.outcome == 'success' run: | @@ -144,6 +163,20 @@ runs: exit ${TEST_EXIT_CODE:-0} shell: bash + - name: Run Python VSR raw request tests + if: inputs.task == 'test-vsr' + run: | + cd foreign/python + mkdir -p ../../reports + IGGY_TEST_PROTOCOL=vsr \ + IGGY_SERVER_HOST=127.0.0.1 \ + IGGY_SERVER_TCP_PORT=8090 \ + uv run --no-sync pytest tests/test_raw_command_vsr.py -v \ + --junitxml=../../reports/python-vsr-junit.xml \ + --tb=short \ + --capture=no + shell: bash + - name: Run Python unit tests only (fallback) if: inputs.task == 'test' && steps.iggy.outcome != 'success' run: | @@ -168,6 +201,13 @@ runs: pid-file: ${{ steps.iggy.outputs.pid_file }} log-file: ${{ steps.iggy.outputs.log_file }} + - name: Stop VSR Iggy server + if: always() && inputs.task == 'test-vsr' + uses: ./.github/actions/utils/server-stop + with: + pid-file: ${{ steps.iggy-vsr.outputs.pid_file }} + log-file: ${{ steps.iggy-vsr.outputs.log_file }} + - name: Generate coverage report if: inputs.task == 'test' run: | @@ -189,13 +229,15 @@ runs: shell: bash - name: Upload test artifacts - if: always() && inputs.task == 'test' + if: always() && (inputs.task == 'test' || inputs.task == 'test-vsr') uses: actions/upload-artifact@v7 with: - name: python-test-results-${{ github.run_id }}-${{ github.run_attempt }} + name: python-${{ inputs.task }}-results-${{ github.run_id }}-${{ github.run_attempt }} path: | reports/python-junit.xml + reports/python-vsr-junit.xml reports/python-coverage.lcov foreign/python/dist/*.whl + ${{ runner.temp }}/iggy-server-vsr.log retention-days: 7 if-no-files-found: ignore diff --git a/.github/config/components.yml b/.github/config/components.yml index 2d30025cb..631025aca 100644 --- a/.github/config/components.yml +++ b/.github/config/components.yml @@ -232,7 +232,7 @@ components: - "ci-infrastructure" # CI changes trigger full regression paths: - "foreign/python/**" - tasks: ["lint", "test", "build"] + tasks: ["lint", "test", "test-vsr", "build"] sdk-php: depends_on: diff --git a/Cargo.lock b/Cargo.lock index 7591cdbf3..50939ed39 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6611,7 +6611,7 @@ checksum = "cd62e6b5e86ea8eeeb8db1de02880a6abc01a397b2ebb64b5d74ac255318f5cb" [[package]] name = "iggy" -version = "0.10.3-edge.2" +version = "0.10.3-edge.3" dependencies = [ "async-broadcast", "async-dropper", @@ -6821,7 +6821,7 @@ dependencies = [ [[package]] name = "iggy_binary_protocol" -version = "0.10.3-edge.2" +version = "0.10.3-edge.3" dependencies = [ "aligned-vec", "bytemuck", @@ -6833,7 +6833,7 @@ dependencies = [ [[package]] name = "iggy_common" -version = "0.10.3-edge.2" +version = "0.10.3-edge.3" dependencies = [ "aes-gcm", "async-broadcast", diff --git a/Cargo.toml b/Cargo.toml index b03900ab1..153461028 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -196,10 +196,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.2" } +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.2" } -iggy_common = { path = "core/common", version = "0.10.3-edge.2" } +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_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 ed686c9bd..37cc01eb7 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.dev3" +version = "0.8.1.dev4" source = { directory = "../../foreign/python" } [package.metadata] diff --git a/core/binary_protocol/Cargo.toml b/core/binary_protocol/Cargo.toml index 1155dcdd0..ce36c19b1 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.2" +version = "0.10.3-edge.3" 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/common/Cargo.toml b/core/common/Cargo.toml index d5c80224d..3934e5bbd 100644 --- a/core/common/Cargo.toml +++ b/core/common/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy_common" -version = "0.10.3-edge.2" +version = "0.10.3-edge.3" 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/lib.rs b/core/common/src/lib.rs index 12f139d2a..e4964a3b2 100644 --- a/core/common/src/lib.rs +++ b/core/common/src/lib.rs @@ -72,6 +72,7 @@ pub use traits::topic_client::TopicClient; pub use traits::user_client::UserClient; pub use traits::validatable::Validatable; pub use types::args::*; +pub use types::binary_request_kind::BinaryRequestKind; pub use types::client::client_info::*; pub use types::client_state::ClientState; pub use types::cluster::*; diff --git a/core/common/src/traits/binary_transport.rs b/core/common/src/traits/binary_transport.rs index 3dacff9ad..80f1b7182 100644 --- a/core/common/src/traits/binary_transport.rs +++ b/core/common/src/traits/binary_transport.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use crate::{ClientState, DiagnosticEvent, IggyDuration, IggyError}; +use crate::{BinaryRequestKind, ClientState, DiagnosticEvent, IggyDuration, IggyError}; use async_trait::async_trait; use bytes::Bytes; #[cfg(feature = "vsr")] @@ -28,7 +28,18 @@ pub trait BinaryTransport { /// Sets the state of the client. async fn set_state(&self, state: ClientState); async fn publish_event(&self, event: DiagnosticEvent); + /// Send a standard command. The protocol tables classify `code`, so the + /// caller declares no execution model; every typed SDK method lands here. async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError>; + /// Send a raw command whose `code` the SDK need not know, with the caller + /// declaring how it executes. A code the tables do know keeps its own + /// class and a conflicting `kind` is rejected. + async fn send_raw_with_kind( + &self, + kind: BinaryRequestKind, + code: u32, + payload: Bytes, + ) -> Result<Bytes, IggyError>; fn get_heartbeat_interval(&self) -> IggyDuration; /// Per-transport consumer-group + partitioning cache used to resolve diff --git a/core/common/src/types/binary_request_kind.rs b/core/common/src/types/binary_request_kind.rs new file mode 100644 index 000000000..cfd1e49d2 --- /dev/null +++ b/core/common/src/types/binary_request_kind.rs @@ -0,0 +1,80 @@ +// 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 strum::{Display, EnumString}; + +/// How a raw binary request executes on the server. +/// +/// A vendor command code is unknown to the SDK, so the caller has to say whether +/// it runs outside consensus or is replicated through it. The declaration is +/// never authoritative: a code the protocol tables already know keeps its own +/// class, and the server independently rejects a declaration it disagrees with. +/// +/// Classic framing carries `[length][code][payload]` with no operation field, so +/// the kind is inert there and both variants encode identical bytes. It only +/// selects a wire path under VSR. +/// +/// The string form is the cross-SDK spelling the shared BDD scenarios and the +/// PHP binding use. +#[derive(Debug, Copy, Clone, PartialEq, Eq, Display, EnumString)] +#[strum(serialize_all = "snake_case")] +pub enum BinaryRequestKind { + /// Runs on the receiving node only, outside consensus. Reads, pings, and + /// vendor commands that own no replicated state. + NonReplicated, + /// Replicated through consensus before it takes effect. + /// + /// Only the standard replicated commands are supported today. A vendor code + /// declared `Replicated` yields [`crate::IggyError::FeatureUnavailable`]: + /// the protocol has no deterministic handler registry, replicated state + /// ownership, or snapshot contract for one yet. + Replicated, +} + +#[cfg(test)] +mod tests { + use std::str::FromStr; + + use super::BinaryRequestKind; + + #[test] + fn given_display_output_when_parsed_should_round_trip() { + for kind in [ + BinaryRequestKind::NonReplicated, + BinaryRequestKind::Replicated, + ] { + assert_eq!( + BinaryRequestKind::from_str(&kind.to_string()).unwrap(), + kind + ); + } + } + + #[test] + fn given_cross_sdk_names_when_parsed_should_match_the_other_bindings() { + assert_eq!( + BinaryRequestKind::NonReplicated.to_string(), + "non_replicated" + ); + assert_eq!(BinaryRequestKind::Replicated.to_string(), "replicated"); + } + + #[test] + fn given_unknown_text_when_parsed_should_fail() { + assert!(BinaryRequestKind::from_str("auto").is_err()); + } +} diff --git a/core/common/src/types/mod.rs b/core/common/src/types/mod.rs index f1a2f37af..a8e2edfab 100644 --- a/core/common/src/types/mod.rs +++ b/core/common/src/types/mod.rs @@ -16,6 +16,7 @@ // under the License. pub(crate) mod args; +pub(crate) mod binary_request_kind; pub(crate) mod client; pub(crate) mod client_state; pub(crate) mod cluster; diff --git a/core/integration/tests/sdk/raw.rs b/core/integration/tests/sdk/raw.rs index cffeec954..89accc9a6 100644 --- a/core/integration/tests/sdk/raw.rs +++ b/core/integration/tests/sdk/raw.rs @@ -22,6 +22,15 @@ use iggy_binary_protocol::codes::{GET_STATS_CODE, LOGIN_USER_CODE, PING_CODE}; use iggy_binary_protocol::requests::system::{GetStatsRequest, PingRequest}; use integration::iggy_harness; +/// A code no server registers a handler for, well past every range the protocol +/// assigns, so it stays a vendor code as the command table grows. +const VENDOR_CODE: u32 = 60_001; + +const KINDS: [BinaryRequestKind; 2] = [ + BinaryRequestKind::NonReplicated, + BinaryRequestKind::Replicated, +]; + #[cfg(not(feature = "vsr"))] #[iggy_harness(test_client_transport = [Tcp, Quic, Http, WebSocket])] async fn given_authenticated_client_when_sending_raw_request_should_round_trip( @@ -63,39 +72,24 @@ async fn assert_raw_round_trip(client: &IggyClient) { let error = client .send_binary_request(PING_CODE, PingRequest.to_bytes()) .await - .expect_err("binary command must be unavailable on HTTP"); + .expect_err("legacy binary command must be unavailable on HTTP"); assert_eq!(error, IggyError::FeatureUnavailable); - } - _ => { - let response = client - .send_binary_request(PING_CODE, PingRequest.to_bytes()) - .await - .expect("binary ping request should succeed"); - assert!(response.is_empty()); - - let stats = client - .send_binary_request(GET_STATS_CODE, GetStatsRequest.to_bytes()) - .await - .expect("authenticated binary command should return a body"); - assert!(!stats.is_empty()); - - // Rejected before the wire (guards the VSR consensus-session panic - // when a login code is sent raw on a bound client). - let error = client - .send_binary_request(LOGIN_USER_CODE, Bytes::new()) - .await - .expect_err("session-control codes must be rejected by the raw path"); - assert_eq!(error, IggyError::InvalidCommand); - // VSR encoder is closed-world: unknown code rejected at encode time. - #[cfg(feature = "vsr")] - { + for kind in KINDS { let error = client - .send_binary_request(60_000, Bytes::new()) + .send_binary_request_with_kind(kind, PING_CODE, PingRequest.to_bytes()) .await - .expect_err("unknown code must be rejected under VSR"); - assert_eq!(error, IggyError::InvalidCommand); + .expect_err("binary command must be unavailable on HTTP"); + assert_eq!(error, IggyError::FeatureUnavailable); } + } + _ => { + assert_standard_codes_round_trip(client).await; + assert_session_control_codes_are_rejected(client).await; + assert_legacy_vendor_code_stays_closed_world(client).await; + assert_vendor_code_reaches_the_server(client).await; + assert_replicated_declaration_is_honored(client).await; + assert_conflicting_declaration_is_honored(client).await; let error = client .send_http_request(HttpMethod::Get, "/ping", None) @@ -105,3 +99,138 @@ async fn assert_raw_round_trip(client: &IggyClient) { } } } + +async fn assert_standard_codes_round_trip(client: &IggyClient) { + let response = client + .send_binary_request(PING_CODE, PingRequest.to_bytes()) + .await + .expect("legacy binary ping request should succeed"); + assert!(response.is_empty()); + + let stats = client + .send_binary_request(GET_STATS_CODE, GetStatsRequest.to_bytes()) + .await + .expect("legacy authenticated binary command should return a body"); + assert!(!stats.is_empty()); + + let response = client + .send_binary_request_with_kind( + BinaryRequestKind::NonReplicated, + PING_CODE, + PingRequest.to_bytes(), + ) + .await + .expect("binary ping request should succeed"); + assert!(response.is_empty()); + + let stats = client + .send_binary_request_with_kind( + BinaryRequestKind::NonReplicated, + GET_STATS_CODE, + GetStatsRequest.to_bytes(), + ) + .await + .expect("authenticated binary command should return a body"); + assert!(!stats.is_empty()); +} + +/// Rejected before the wire whatever the declaration, which also guards the VSR +/// consensus-session panic a raw login on a bound client used to trigger. +async fn assert_session_control_codes_are_rejected(client: &IggyClient) { + let error = client + .send_binary_request(LOGIN_USER_CODE, Bytes::new()) + .await + .expect_err("legacy session-control codes must be rejected"); + assert_eq!(error, IggyError::InvalidCommand); + + for kind in KINDS { + let error = client + .send_binary_request_with_kind(kind, LOGIN_USER_CODE, Bytes::new()) + .await + .expect_err("session-control codes must be rejected by the raw path"); + assert_eq!(error, IggyError::InvalidCommand); + } +} + +/// The compatibility API keeps its previous VSR closed-world behavior while +/// classic framing still lets the server judge an arbitrary code. +async fn assert_legacy_vendor_code_stays_closed_world(client: &IggyClient) { + let error = client + .send_binary_request(VENDOR_CODE, Bytes::from_static(b"vendor-body")) + .await + .expect_err("no server or VSR table knows the vendor code"); + assert_eq!(error, IggyError::InvalidCommand); + + client + .ping() + .await + .expect("legacy raw rejection must not poison the connection"); +} + +/// A non-replicated vendor code leaves the SDK, so the rejection comes from the +/// server rather than the encoder, and the connection survives it. +async fn assert_vendor_code_reaches_the_server(client: &IggyClient) { + let error = client + .send_binary_request_with_kind( + BinaryRequestKind::NonReplicated, + VENDOR_CODE, + Bytes::from_static(b"vendor-body"), + ) + .await + .expect_err("no server registers a handler for the vendor code"); + assert_eq!(error, IggyError::InvalidCommand); + + client + .ping() + .await + .expect("connection must stay usable after a request-level rejection"); +} + +/// Under VSR a replicated vendor code has no deterministic handler registry, so +/// it fails in the encoder. Classic framing has no operation field, so the same +/// call reaches the server and is rejected there. +async fn assert_replicated_declaration_is_honored(client: &IggyClient) { + let error = client + .send_binary_request_with_kind( + BinaryRequestKind::Replicated, + VENDOR_CODE, + Bytes::from_static(b"vendor-body"), + ) + .await + .expect_err("a replicated vendor code has no handler on either protocol"); + + #[cfg(feature = "vsr")] + assert_eq!(error, IggyError::FeatureUnavailable); + #[cfg(not(feature = "vsr"))] + assert_eq!(error, IggyError::InvalidCommand); + + client + .ping() + .await + .expect("connection must stay usable after a rejected declaration"); +} + +/// A declaration that disagrees with a standard code cannot redirect it. Under +/// VSR the encoder rejects the pair; classic framing ignores the declaration and +/// emits the same bytes it would for the matching one. +async fn assert_conflicting_declaration_is_honored(client: &IggyClient) { + let result = client + .send_binary_request_with_kind( + BinaryRequestKind::Replicated, + GET_STATS_CODE, + GetStatsRequest.to_bytes(), + ) + .await; + + #[cfg(feature = "vsr")] + assert_eq!( + result.expect_err("a read declared replicated must be rejected"), + IggyError::InvalidCommand + ); + #[cfg(not(feature = "vsr"))] + assert!( + !result + .expect("classic framing ignores the declaration") + .is_empty() + ); +} diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml index d308dc363..5f8d91aa9 100644 --- a/core/sdk/Cargo.toml +++ b/core/sdk/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "iggy" -version = "0.10.3-edge.2" +version = "0.10.3-edge.3" 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/sdk/src/binary/mod.rs b/core/sdk/src/binary/mod.rs index d892b1910..5224b404b 100644 --- a/core/sdk/src/binary/mod.rs +++ b/core/sdk/src/binary/mod.rs @@ -15,4 +15,27 @@ // specific language governing permissions and limitations // under the License. +use iggy_binary_protocol::codes::{ + LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE, LOGIN_USER_CODE, + LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, LOGOUT_USER_CODE, +}; +use iggy_common::IggyError; pub use iggy_common::{BinaryClient, BinaryTransport}; + +/// Auth/session codes rejected by the raw binary path. Must go through the +/// typed `login_user` / `logout_user` methods to keep session state correct. +pub(crate) const SESSION_CONTROL_CODES: [u32; 5] = [ + LOGIN_USER_CODE, + LOGOUT_USER_CODE, + LOGIN_REGISTER_CODE, + LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, + LOGIN_REGISTER_WITH_PAT_CODE, +]; + +pub(crate) fn validate_binary_request_code(code: u32) -> Result<(), IggyError> { + if SESSION_CONTROL_CODES.contains(&code) { + Err(IggyError::InvalidCommand) + } else { + Ok(()) + } +} diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs index 98beb9830..c2cdb65e0 100644 --- a/core/sdk/src/clients/client.rs +++ b/core/sdk/src/clients/client.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use crate::binary::validate_binary_request_code; use crate::client_wrappers::client_wrapper::ClientWrapper; use crate::client_wrappers::connection_info::ConnectionInfo; use crate::clients::client_builder::IggyClientBuilder; @@ -30,13 +31,9 @@ use crate::websocket::websocket_client::WebSocketClient; use async_broadcast::Receiver; use async_trait::async_trait; use bytes::Bytes; -use iggy_binary_protocol::codes::{ - LOGIN_REGISTER_CODE, LOGIN_REGISTER_WITH_PAT_CODE, LOGIN_USER_CODE, - LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, LOGOUT_USER_CODE, -}; use iggy_common::Consumer; use iggy_common::locking::{IggyRwLock, IggyRwLockFn}; -use iggy_common::{BinaryTransport, Client, HttpMethod, SystemClient}; +use iggy_common::{BinaryRequestKind, BinaryTransport, Client, HttpMethod, SystemClient}; use iggy_common::{ConnectionStringUtils, DiagnosticEvent, Partitioner, TransportProtocol}; use std::fmt::Debug; use std::sync::Arc; @@ -45,16 +42,6 @@ use tokio::time::sleep; use tracing::log::warn; use tracing::{debug, error, info}; -/// Auth/session codes rejected by the raw binary path. Must go through the -/// typed `login_user` / `logout_user` methods to keep session state correct. -const SESSION_CONTROL_CODES: [u32; 5] = [ - LOGIN_USER_CODE, - LOGOUT_USER_CODE, - LOGIN_REGISTER_CODE, - LOGIN_WITH_PERSONAL_ACCESS_TOKEN_CODE, - LOGIN_REGISTER_WITH_PAT_CODE, -]; - /// The main client struct which implements all the `Client` traits and wraps the underlying low-level client for the specific transport. /// /// It also provides the additional builders for the standalone consumer, consumer group, and producer. @@ -204,13 +191,8 @@ impl IggyClient { /// Login and logout codes are rejected with `InvalidCommand`. Use the /// `login_user` / `logout_user` methods so SDK session state stays correct. /// - /// Custom codes only work on the classic protocol. Under `vsr` the encoder - /// is closed-world: an unknown code yields `InvalidCommand`, a replicated - /// code with no mapping yields `UnknownReplicatedCommand`. pub async fn send_binary_request(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> { - if SESSION_CONTROL_CODES.contains(&code) { - return Err(IggyError::InvalidCommand); - } + validate_binary_request_code(code)?; match &*self.client.read().await { ClientWrapper::Tcp(client) => client.send_raw_with_response(code, payload).await, ClientWrapper::Quic(client) => client.send_raw_with_response(code, payload).await, @@ -219,6 +201,38 @@ impl IggyClient { } } + /// Send a raw binary command with an explicit replication declaration. + /// + /// A standard command keeps the class defined by the protocol tables, and + /// a conflicting declaration is rejected with `InvalidCommand`. Classic + /// framing carries no operation field, so the declaration does not change + /// the encoded bytes. Under `vsr`, an unknown + /// [`BinaryRequestKind::NonReplicated`] code uses + /// `Operation::NonReplicated`; an unknown + /// [`BinaryRequestKind::Replicated`] code yields `FeatureUnavailable` + /// until the protocol has a replicated extension registry. + /// + /// Login and logout codes are rejected with `InvalidCommand`. Use the + /// typed session methods so SDK state stays synchronized with the server. + pub async fn send_binary_request_with_kind( + &self, + kind: BinaryRequestKind, + code: u32, + payload: Bytes, + ) -> Result<Bytes, IggyError> { + // Validate before dispatch so HTTP returns InvalidCommand for session + // control instead of its blanket FeatureUnavailable. + validate_binary_request_code(code)?; + match &*self.client.read().await { + ClientWrapper::Tcp(client) => client.send_raw_with_kind(kind, code, payload).await, + ClientWrapper::Quic(client) => client.send_raw_with_kind(kind, code, payload).await, + ClientWrapper::WebSocket(client) => { + client.send_raw_with_kind(kind, code, payload).await + } + ClientWrapper::Http(_) | ClientWrapper::Iggy(_) => Err(IggyError::FeatureUnavailable), + } + } + /// Invoke an arbitrary HTTP endpoint and return the raw response body. HTTP /// transport only; binary transports yield `FeatureUnavailable`. pub async fn send_http_request( @@ -289,6 +303,7 @@ impl Client for IggyClient { #[cfg(test)] mod tests { use super::*; + use crate::binary::SESSION_CONTROL_CODES; #[test] fn should_fail_with_empty_connection_string() { @@ -482,6 +497,11 @@ mod tests { IggyClient::from_connection_string("iggy+http://user:[email protected]:1234";).unwrap(); let result = client.send_binary_request(0, Bytes::new()).await; assert!(matches!(result, Err(IggyError::FeatureUnavailable))); + + let result = client + .send_binary_request_with_kind(BinaryRequestKind::NonReplicated, 0, Bytes::new()) + .await; + assert!(matches!(result, Err(IggyError::FeatureUnavailable))); } #[tokio::test] @@ -491,8 +511,23 @@ mod tests { let result = client.send_binary_request(code, Bytes::new()).await; assert!( matches!(result, Err(IggyError::InvalidCommand)), - "code {code} must be rejected before reaching the transport" + "legacy raw code {code} must be rejected before reaching the transport" ); } + + for kind in [ + BinaryRequestKind::NonReplicated, + BinaryRequestKind::Replicated, + ] { + for code in SESSION_CONTROL_CODES { + let result = client + .send_binary_request_with_kind(kind, code, Bytes::new()) + .await; + assert!( + matches!(result, Err(IggyError::InvalidCommand)), + "code {code} declared {kind} must be rejected before reaching the transport" + ); + } + } } } diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs index 34f8d1484..1aa6ea93b 100644 --- a/core/sdk/src/prelude.rs +++ b/core/sdk/src/prelude.rs @@ -48,12 +48,12 @@ pub use crate::stream_builder::{IggyStream, IggyStreamConfig}; pub use crate::tcp::tcp_client::TcpClient; pub use crate::websocket::websocket_client::WebSocketClient; pub use iggy_common::{ - Aes256GcmEncryptor, Args, ArgsOptional, AutoLogin, CacheMetrics, CacheMetricsKey, ClientError, - ClientInfoDetails, ClusterMetadata, ClusterNode, ClusterNodeRole, ClusterNodeStatus, - CompressionAlgorithm, Consumer, ConsumerGroup, ConsumerGroupDetails, ConsumerGroupMember, - ConsumerKind, EncryptorKind, GlobalPermissions, HeaderKey, HeaderKind, HeaderValue, - HttpClientConfig, HttpClientConfigBuilder, HttpMethod, IdKind, Identifier, IdentityInfo, - IggyByteSize, IggyDuration, IggyError, IggyExpiry, IggyIndexView, IggyMessage, + Aes256GcmEncryptor, Args, ArgsOptional, AutoLogin, BinaryRequestKind, CacheMetrics, + CacheMetricsKey, ClientError, ClientInfoDetails, ClusterMetadata, ClusterNode, ClusterNodeRole, + ClusterNodeStatus, CompressionAlgorithm, Consumer, ConsumerGroup, ConsumerGroupDetails, + ConsumerGroupMember, ConsumerKind, EncryptorKind, GlobalPermissions, HeaderKey, HeaderKind, + HeaderValue, HttpClientConfig, HttpClientConfigBuilder, HttpMethod, IdKind, Identifier, + IdentityInfo, IggyByteSize, IggyDuration, IggyError, IggyExpiry, IggyIndexView, IggyMessage, IggyMessageHeader, IggyMessageHeaderView, IggyMessageView, IggyMessageViewIterator, IggyTimestamp, MaxTopicSize, Partition, Partitioner, Partitioning, Permissions, PersonalAccessTokenExpiry, PollMessages, PolledMessages, PollingKind, PollingStrategy, diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs index 9a731f608..75e4ee2d5 100644 --- a/core/sdk/src/quic/quic_client.rs +++ b/core/sdk/src/quic/quic_client.rs @@ -15,13 +15,16 @@ // specific language governing permissions and limitations // under the License. +use crate::binary::validate_binary_request_code; use crate::leader_aware::{LeaderRedirectionState, check_and_redirect_to_leader}; use crate::prelude::AutoLogin; #[cfg(feature = "vsr")] use crate::session::ConsensusSession; #[cfg(feature = "vsr")] use iggy_common::VsrSessionControl as _; -use iggy_common::{BinaryClient, BinaryTransport, Client, PersonalAccessTokenClient, UserClient}; +use iggy_common::{ + BinaryClient, BinaryRequestKind, BinaryTransport, Client, PersonalAccessTokenClient, UserClient, +}; use crate::prelude::{IggyDuration, IggyError, IggyTimestamp, QuicClientConfig}; use crate::quic::skip_server_verification::SkipServerVerification; @@ -137,59 +140,17 @@ impl BinaryTransport for QuicClient { } async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> { - let result = self.send_raw(code, payload.clone()).await; - if result.is_ok() { - return result; - } - - let error = result.unwrap_err(); - if !matches!( - error, - IggyError::Disconnected - | IggyError::EmptyResponse - | IggyError::Unauthenticated - | IggyError::StaleClient - | IggyError::NotConnected - | IggyError::CannotEstablishConnection - | IggyError::QuicError - ) { - return Err(error); - } - - if !self.config.reconnection.enabled { - return Err(IggyError::Disconnected); - } - - #[cfg(feature = "vsr")] - if matches!(self.config.auto_login, AutoLogin::Disabled) && !is_login_register_code(code) { - // Without auto-login a reconnect cannot re-establish the session, so - // non-login requests are not recovered here - their transient replay - // happens on the live connection inside `send_raw`. Login/register - // is the exception: the server stays deliberately silent on a - // transient register failure and relies on the client replaying via - // a reconnect with a fresh session. - return Err(error); - } + self.send_with_reconnect(None, code, payload).await + } - self.disconnect().await?; - #[cfg(feature = "vsr")] - let skip_auto_login = is_login_register_code(code); - #[cfg(feature = "vsr")] - if skip_auto_login { - *self.skip_auto_login_once.lock().await = true; - } - let server_address = self.current_server_address.lock().await.to_string(); - info!( - "Reconnecting to the server: {}, by client: {}", - server_address, self.config.client_address - ); - let reconnect = self.connect().await; - #[cfg(feature = "vsr")] - if skip_auto_login && reconnect.is_err() { - *self.skip_auto_login_once.lock().await = false; - } - reconnect?; - self.send_raw(code, payload).await + async fn send_raw_with_kind( + &self, + kind: BinaryRequestKind, + code: u32, + payload: Bytes, + ) -> Result<Bytes, IggyError> { + validate_binary_request_code(code)?; + self.send_with_reconnect(Some(kind), code, payload).await } fn get_heartbeat_interval(&self) -> IggyDuration { @@ -679,7 +640,76 @@ impl QuicClient { Ok(()) } - async fn send_raw(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> { + /// Single reconnect-and-replay path shared by the typed and the raw entry + /// points, so `kind` survives the retry attempt unchanged. + async fn send_with_reconnect( + &self, + kind: Option<BinaryRequestKind>, + code: u32, + payload: Bytes, + ) -> Result<Bytes, IggyError> { + let result = self.send_raw(kind, code, payload.clone()).await; + if result.is_ok() { + return result; + } + + let error = result.unwrap_err(); + if !matches!( + error, + IggyError::Disconnected + | IggyError::EmptyResponse + | IggyError::Unauthenticated + | IggyError::StaleClient + | IggyError::NotConnected + | IggyError::CannotEstablishConnection + | IggyError::QuicError + ) { + return Err(error); + } + + if !self.config.reconnection.enabled { + return Err(IggyError::Disconnected); + } + + #[cfg(feature = "vsr")] + if matches!(self.config.auto_login, AutoLogin::Disabled) && !is_login_register_code(code) { + // Without auto-login a reconnect cannot re-establish the session, so + // non-login requests are not recovered here - their transient replay + // happens on the live connection inside `send_raw`. Login/register + // is the exception: the server stays deliberately silent on a + // transient register failure and relies on the client replaying via + // a reconnect with a fresh session. + return Err(error); + } + + self.disconnect().await?; + #[cfg(feature = "vsr")] + let skip_auto_login = is_login_register_code(code); + #[cfg(feature = "vsr")] + if skip_auto_login { + *self.skip_auto_login_once.lock().await = true; + } + let server_address = self.current_server_address.lock().await.to_string(); + info!( + "Reconnecting to the server: {}, by client: {}", + server_address, self.config.client_address + ); + let reconnect = self.connect().await; + #[cfg(feature = "vsr")] + if skip_auto_login && reconnect.is_err() { + *self.skip_auto_login_once.lock().await = false; + } + reconnect?; + self.send_raw(kind, code, payload).await + } + + #[cfg_attr(not(feature = "vsr"), expect(unused_variables))] + async fn send_raw( + &self, + kind: Option<BinaryRequestKind>, + code: u32, + payload: Bytes, + ) -> Result<Bytes, IggyError> { match self.get_state().await { ClientState::Shutdown => { trace!("Cannot send data. Client is shutdown."); @@ -720,7 +750,12 @@ impl QuicClient { let mut consensus_session = consensus_session .lock() .expect("consensus session mutex poisoned"); - crate::vsr::encode_request_header(&mut consensus_session, code, &payload)? + crate::vsr::encode_request_header( + &mut consensus_session, + kind, + code, + &payload, + )? }; trace!("Sending a QUIC VSR request of size {request_size} with code: {code}"); // Same-connection transient resend, gated on the EXPLICIT @@ -925,6 +960,27 @@ mod tests { assert!(quic_client.is_err()); } + #[tokio::test] + async fn raw_binary_transport_rejects_session_control_before_io() { + let client = + QuicClient::from_connection_string("iggy+quic://iggy:[email protected]:8080").unwrap(); + + for kind in [ + BinaryRequestKind::NonReplicated, + BinaryRequestKind::Replicated, + ] { + let error = client + .send_raw_with_kind( + kind, + iggy_binary_protocol::codes::LOGIN_USER_CODE, + Bytes::new(), + ) + .await + .unwrap_err(); + assert!(matches!(error, IggyError::InvalidCommand)); + } + } + #[tokio::test] async fn should_fail_without_username() { let connection_string_prefix = "iggy+"; diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs index f827b9f55..34da743e7 100644 --- a/core/sdk/src/tcp/tcp_client.rs +++ b/core/sdk/src/tcp/tcp_client.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use crate::binary::validate_binary_request_code; use crate::leader_aware::{LeaderRedirectionState, check_and_redirect_to_leader}; use crate::prelude::Client; use crate::prelude::TcpClientConfig; @@ -38,7 +39,9 @@ use iggy_common::{ AutoLogin, ClientState, ConnectionString, ConnectionStringUtils, Credentials, DiagnosticEvent, IggyDuration, IggyError, IggyTimestamp, TcpConnectionStringOptions, TransportProtocol, }; -use iggy_common::{BinaryClient, BinaryTransport, PersonalAccessTokenClient, UserClient}; +use iggy_common::{ + BinaryClient, BinaryRequestKind, BinaryTransport, PersonalAccessTokenClient, UserClient, +}; use rustls::pki_types::{CertificateDer, ServerName, pem::PemObject}; use secrecy::ExposeSecret; use std::net::SocketAddr; @@ -148,64 +151,17 @@ impl BinaryTransport for TcpClient { } async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> { - let result = self.send_raw(code, payload.clone()).await; - if result.is_ok() { - return result; - } - - let error = result.unwrap_err(); - if !matches!( - error, - IggyError::Disconnected - | IggyError::EmptyResponse - | IggyError::Unauthenticated - | IggyError::StaleClient - | IggyError::NotConnected - | IggyError::CannotEstablishConnection - | IggyError::TcpError - ) { - return Err(error); - } - - if !self.config.reconnection.enabled { - return Err(IggyError::Disconnected); - } - - #[cfg(feature = "vsr")] - if matches!(self.config.auto_login, AutoLogin::Disabled) && !is_login_register_code(code) { - // Without auto-login a reconnect cannot re-establish the session, - // so non-login requests fail fast. Login/register itself is the - // exception: the server stays deliberately silent on transient - // register failures (server-ng `surface_login_failure`) and - // relies on the client timing out and replaying the request. - return Err(error); - } - - self.disconnect().await?; - - #[cfg(feature = "vsr")] - let skip_auto_login = is_login_register_code(code); - #[cfg(feature = "vsr")] - if skip_auto_login { - *self.skip_auto_login_once.lock().await = true; - } - - { - let client_address = self.get_client_address_value().await; - let server_address = self.current_server_address.lock().await.clone(); - info!( - "Reconnecting to the server: {} by client: {client_address}...", - server_address - ); - } + self.send_with_reconnect(None, code, payload).await + } - let reconnect = self.connect().await; - #[cfg(feature = "vsr")] - if skip_auto_login && reconnect.is_err() { - *self.skip_auto_login_once.lock().await = false; - } - reconnect?; - self.send_raw(code, payload).await + async fn send_raw_with_kind( + &self, + kind: BinaryRequestKind, + code: u32, + payload: Bytes, + ) -> Result<Bytes, IggyError> { + validate_binary_request_code(code)?; + self.send_with_reconnect(Some(kind), code, payload).await } fn get_heartbeat_interval(&self) -> IggyDuration { @@ -708,7 +664,81 @@ impl TcpClient { Ok(()) } - async fn send_raw(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> { + /// Single reconnect-and-replay path shared by the typed and the raw entry + /// points, so `kind` survives the retry attempt unchanged. + async fn send_with_reconnect( + &self, + kind: Option<BinaryRequestKind>, + code: u32, + payload: Bytes, + ) -> Result<Bytes, IggyError> { + let result = self.send_raw(kind, code, payload.clone()).await; + if result.is_ok() { + return result; + } + + let error = result.unwrap_err(); + if !matches!( + error, + IggyError::Disconnected + | IggyError::EmptyResponse + | IggyError::Unauthenticated + | IggyError::StaleClient + | IggyError::NotConnected + | IggyError::CannotEstablishConnection + | IggyError::TcpError + ) { + return Err(error); + } + + if !self.config.reconnection.enabled { + return Err(IggyError::Disconnected); + } + + #[cfg(feature = "vsr")] + if matches!(self.config.auto_login, AutoLogin::Disabled) && !is_login_register_code(code) { + // Without auto-login a reconnect cannot re-establish the session, + // so non-login requests fail fast. Login/register itself is the + // exception: the server stays deliberately silent on transient + // register failures (server-ng `surface_login_failure`) and + // relies on the client timing out and replaying the request. + return Err(error); + } + + self.disconnect().await?; + + #[cfg(feature = "vsr")] + let skip_auto_login = is_login_register_code(code); + #[cfg(feature = "vsr")] + if skip_auto_login { + *self.skip_auto_login_once.lock().await = true; + } + + { + let client_address = self.get_client_address_value().await; + let server_address = self.current_server_address.lock().await.clone(); + info!( + "Reconnecting to the server: {} by client: {client_address}...", + server_address + ); + } + + let reconnect = self.connect().await; + #[cfg(feature = "vsr")] + if skip_auto_login && reconnect.is_err() { + *self.skip_auto_login_once.lock().await = false; + } + reconnect?; + self.send_raw(kind, code, payload).await + } + + #[cfg_attr(not(feature = "vsr"), expect(unused_variables))] + async fn send_raw( + &self, + kind: Option<BinaryRequestKind>, + code: u32, + payload: Bytes, + ) -> Result<Bytes, IggyError> { match self.get_state().await { ClientState::Shutdown => { trace!("Cannot send data. Client is shutdown."); @@ -744,6 +774,7 @@ impl TcpClient { }; let (header, result) = self .send_raw_vsr_attempt( + kind, code, payload.clone(), preencoded, @@ -858,6 +889,7 @@ impl TcpClient { #[cfg(feature = "vsr")] async fn send_raw_vsr_attempt( &self, + kind: Option<BinaryRequestKind>, code: u32, payload: Bytes, preencoded: Option<iggy_binary_protocol::consensus::RequestHeader>, @@ -892,7 +924,12 @@ impl TcpClient { let mut consensus_session = consensus_session .lock() .expect("consensus session mutex poisoned"); - crate::vsr::encode_request_header(&mut consensus_session, code, &payload) + crate::vsr::encode_request_header( + &mut consensus_session, + kind, + code, + &payload, + ) }; match encoded { Ok((header, request_size)) => { @@ -1065,6 +1102,27 @@ mod tests { assert!(tcp_client.is_err()); } + #[tokio::test] + async fn raw_binary_transport_rejects_session_control_before_io() { + let client = + TcpClient::from_connection_string("iggy+tcp://iggy:[email protected]:8090").unwrap(); + + for kind in [ + BinaryRequestKind::NonReplicated, + BinaryRequestKind::Replicated, + ] { + let error = client + .send_raw_with_kind( + kind, + iggy_binary_protocol::codes::LOGIN_USER_CODE, + Bytes::new(), + ) + .await + .unwrap_err(); + assert!(matches!(error, IggyError::InvalidCommand)); + } + } + #[test] fn should_fail_without_username() { let connection_string_prefix = "iggy+"; diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs index d1e9ab859..0aae763df 100644 --- a/core/sdk/src/vsr.rs +++ b/core/sdk/src/vsr.rs @@ -38,7 +38,7 @@ use iggy_binary_protocol::requests::consumer_offsets::{ use iggy_binary_protocol::requests::messages::SendMessagesHeader; use iggy_binary_protocol::requests::segments::DeleteSegmentsRequest; use iggy_binary_protocol::{WireIdentifier, WirePartitioning}; -use iggy_common::{IggyError, eviction_reason_to_error}; +use iggy_common::{BinaryRequestKind, IggyError, eviction_reason_to_error}; const NON_REPLICATED_CODE_RANGE: std::ops::Range<usize> = 0..4; @@ -54,18 +54,23 @@ const NON_REPLICATED_CODE_RANGE: std::ops::Range<usize> = 0..4; // independent of (client_id, request_id). pub(crate) fn encode_contiguous_request( session: &mut ConsensusSession, + kind: Option<BinaryRequestKind>, code: u32, payload: &Bytes, ) -> Result<Bytes, IggyError> { - let (header, total_size) = encode_request_header(session, code, payload)?; + let (header, total_size) = encode_request_header(session, kind, code, payload)?; let mut request = BytesMut::with_capacity(total_size); request.put_slice(bytemuck::bytes_of(&header)); request.put_slice(payload); Ok(request.freeze()) } +/// `kind` is the caller's replication declaration for a raw request. `None` +/// marks a standard typed command, which the protocol tables classify on their +/// own. pub(crate) fn encode_request_header( session: &mut ConsensusSession, + kind: Option<BinaryRequestKind>, code: u32, payload: &Bytes, ) -> Result<(RequestHeader, usize), IggyError> { @@ -79,7 +84,7 @@ pub(crate) fn encode_request_header( (Operation::Register, session.begin_register(), 0) } _ => { - let operation = operation_for_code(code)?; + let operation = operation_for_code(code, kind)?; // NonReplicated ops (ping, reads) bypass server-side dedup -- // `ClientTable` only tracks request_ids for replicated ops. If // they consumed the monotonic counter, the next replicated @@ -144,19 +149,57 @@ pub(crate) fn encode_request_header( Ok((header, total_size)) } -fn operation_for_code(code: u32) -> Result<Operation, IggyError> { +/// Pick the header operation for `code`, honoring the caller's declaration +/// only where the protocol tables have nothing to say. +/// +/// Standard codes resolve first so a caller cannot redirect a shipped command +/// into the other execution model, then an unknown code falls back to the +/// declaration. Without a declaration an unknown code is a bug in a typed SDK +/// method rather than a vendor extension, so it stays `InvalidCommand`. +fn operation_for_code(code: u32, kind: Option<BinaryRequestKind>) -> Result<Operation, IggyError> { if code == LOGOUT_USER_CODE { - return Ok(Operation::Logout); + return accept_declaration(Operation::Logout, kind); } if let Some(operation) = Operation::from_command_code(code) { - return Ok(operation); + return accept_declaration(operation, kind); } match iggy_binary_protocol::dispatch::lookup_command(code) { - Some(meta) if !meta.is_replicated() => Ok(Operation::NonReplicated), + Some(meta) if !meta.is_replicated() => accept_declaration(Operation::NonReplicated, kind), Some(_) => Err(IggyError::UnknownReplicatedCommand(code)), - None => Err(IggyError::InvalidCommand), + None => match kind { + Some(BinaryRequestKind::NonReplicated) => Ok(Operation::NonReplicated), + // A replicated vendor command needs a deterministic server-side + // handler registry, replicated state ownership, and a snapshot + // contract. None of that exists, and a half-supported frame would + // be worse than no frame. + Some(BinaryRequestKind::Replicated) => Err(IggyError::FeatureUnavailable), + None => Err(IggyError::InvalidCommand), + }, + } +} + +/// Reject a declaration that disagrees with the class `operation` already +/// carries. Partition operations decode their namespace from a payload the SDK +/// must understand, so they are replicated but never reachable from a raw +/// request: `namespace_for_request` refuses an unrecognised code. +fn accept_declaration( + operation: Operation, + kind: Option<BinaryRequestKind>, +) -> Result<Operation, IggyError> { + let Some(declared) = kind else { + return Ok(operation); + }; + let standard = if operation == Operation::NonReplicated { + BinaryRequestKind::NonReplicated + } else { + BinaryRequestKind::Replicated + }; + if declared == standard { + Ok(operation) + } else { + Err(IggyError::InvalidCommand) } } @@ -460,6 +503,7 @@ mod tests { use iggy_binary_protocol::codes::{ CREATE_STREAM_CODE, GET_STREAM_CODE, LOGOUT_USER_CODE, PING_CODE, }; + use iggy_binary_protocol::dispatch::COMMAND_TABLE; use iggy_binary_protocol::requests::messages::SendMessagesHeader; use iggy_binary_protocol::requests::streams::CreateStreamRequest; use iggy_binary_protocol::requests::users::LoginRegisterRequest; @@ -467,10 +511,29 @@ mod tests { use iggy_binary_protocol::{ClientVersionInfo, WireEncode, WireName}; use secrecy::SecretString; + /// Outside every range the protocol assigns, so it stays unknown as the + /// command table grows. + const VENDOR_CODE: u32 = 60_001; + fn decode_request_header(bytes: &Bytes) -> RequestHeader { *bytemuck::checked::try_from_bytes::<RequestHeader>(&bytes[..HEADER_SIZE]).unwrap() } + fn read_non_replicated_code(header: &RequestHeader) -> u32 { + u32::from_le_bytes( + header.reserved[NON_REPLICATED_CODE_RANGE] + .try_into() + .unwrap(), + ) + } + + fn bound_session() -> ConsensusSession { + let mut session = ConsensusSession::with_client_id(42); + let _ = session.register_request_id(); + session.bind(99); + session + } + #[test] fn register_request_uses_zero_request_and_session() { let mut session = ConsensusSession::with_client_id(7); @@ -486,7 +549,7 @@ mod tests { }; let bytes = - encode_contiguous_request(&mut session, LOGIN_REGISTER_CODE, &request.to_bytes()) + encode_contiguous_request(&mut session, None, LOGIN_REGISTER_CODE, &request.to_bytes()) .unwrap(); let header = decode_request_header(&bytes); @@ -513,13 +576,14 @@ mod tests { }; let mut session = ConsensusSession::with_client_id(7); - encode_contiguous_request(&mut session, LOGIN_REGISTER_CODE, &request.to_bytes()).unwrap(); + encode_contiguous_request(&mut session, None, LOGIN_REGISTER_CODE, &request.to_bytes()) + .unwrap(); session.bind(42); // A second login on the same bound session must encode a fresh Register // (request 0, session 0), not panic in the one-shot register guard. let bytes = - encode_contiguous_request(&mut session, LOGIN_REGISTER_CODE, &request.to_bytes()) + encode_contiguous_request(&mut session, None, LOGIN_REGISTER_CODE, &request.to_bytes()) .unwrap(); let header = decode_request_header(&bytes); assert_eq!(header.operation, Operation::Register); @@ -618,8 +682,10 @@ mod tests { } .to_bytes(); - let first = encode_contiguous_request(&mut session, CREATE_STREAM_CODE, &payload).unwrap(); - let second = encode_contiguous_request(&mut session, CREATE_STREAM_CODE, &payload).unwrap(); + let first = + encode_contiguous_request(&mut session, None, CREATE_STREAM_CODE, &payload).unwrap(); + let second = + encode_contiguous_request(&mut session, None, CREATE_STREAM_CODE, &payload).unwrap(); assert_eq!(decode_request_header(&first).request, 1); assert_eq!(decode_request_header(&second).request, 2); @@ -631,18 +697,12 @@ mod tests { fn ping_uses_non_replicated_operation() { let mut session = ConsensusSession::with_client_id(42); session.bind(99); - let bytes = encode_contiguous_request(&mut session, PING_CODE, &Bytes::new()).unwrap(); + let bytes = + encode_contiguous_request(&mut session, None, PING_CODE, &Bytes::new()).unwrap(); let header = decode_request_header(&bytes); assert_eq!(header.operation, Operation::NonReplicated); - assert_eq!( - u32::from_le_bytes( - header.reserved[NON_REPLICATED_CODE_RANGE] - .try_into() - .unwrap() - ), - PING_CODE - ); + assert_eq!(read_non_replicated_code(&header), PING_CODE); assert_eq!(header.session, 99); assert_eq!(header.namespace, 0); } @@ -652,7 +712,7 @@ mod tests { let mut session = ConsensusSession::with_client_id(42); session.bind(99); let bytes = - encode_contiguous_request(&mut session, LOGOUT_USER_CODE, &Bytes::new()).unwrap(); + encode_contiguous_request(&mut session, None, LOGOUT_USER_CODE, &Bytes::new()).unwrap(); let header = decode_request_header(&bytes); assert_eq!(header.operation, Operation::Logout); @@ -668,18 +728,11 @@ mod tests { let mut session = ConsensusSession::with_client_id(42); session.bind(99); let bytes = - encode_contiguous_request(&mut session, GET_STREAM_CODE, &Bytes::new()).unwrap(); + encode_contiguous_request(&mut session, None, GET_STREAM_CODE, &Bytes::new()).unwrap(); let header = decode_request_header(&bytes); assert_eq!(header.operation, Operation::NonReplicated); - assert_eq!( - u32::from_le_bytes( - header.reserved[NON_REPLICATED_CODE_RANGE] - .try_into() - .unwrap() - ), - GET_STREAM_CODE - ); + assert_eq!(read_non_replicated_code(&header), GET_STREAM_CODE); assert_eq!(header.session, 99); } @@ -836,4 +889,150 @@ mod tests { let out = split_metadata_result(Operation::SendMessages, body.clone()).unwrap(); assert_eq!(out, body); } + + #[test] + fn vendor_non_replicated_code_encodes_with_the_code_in_reserved() { + let mut session = bound_session(); + let payload = Bytes::from_static(b"vendor-body"); + + let bytes = encode_contiguous_request( + &mut session, + Some(BinaryRequestKind::NonReplicated), + VENDOR_CODE, + &payload, + ) + .unwrap(); + let header = decode_request_header(&bytes); + + assert_eq!(header.operation, Operation::NonReplicated); + assert_eq!(read_non_replicated_code(&header), VENDOR_CODE); + assert_eq!(header.namespace, 0); + assert_eq!(header.session, 99); + assert_eq!(&bytes[HEADER_SIZE..], &payload[..]); + assert_eq!(header.size as usize, HEADER_SIZE + payload.len()); + } + + #[test] + fn vendor_non_replicated_code_does_not_advance_the_request_identifier() { + let mut session = bound_session(); + let payload = CreateStreamRequest { + name: WireName::new("stream").unwrap(), + } + .to_bytes(); + + for _ in 0..3 { + encode_contiguous_request( + &mut session, + Some(BinaryRequestKind::NonReplicated), + VENDOR_CODE, + &Bytes::new(), + ) + .unwrap(); + } + // The next replicated request must still claim id 1: a gap makes the + // primary's `request_preflight` drop it as a `RequestGap`. + let replicated = + encode_contiguous_request(&mut session, None, CREATE_STREAM_CODE, &payload).unwrap(); + + assert_eq!(decode_request_header(&replicated).request, 1); + } + + #[test] + fn vendor_replicated_code_is_unavailable_until_the_registry_exists() { + let mut session = bound_session(); + let error = encode_contiguous_request( + &mut session, + Some(BinaryRequestKind::Replicated), + VENDOR_CODE, + &Bytes::new(), + ) + .unwrap_err(); + + assert!(matches!(error, IggyError::FeatureUnavailable)); + } + + #[test] + fn vendor_code_without_a_declaration_stays_closed_world() { + let mut session = bound_session(); + let error = + encode_contiguous_request(&mut session, None, VENDOR_CODE, &Bytes::new()).unwrap_err(); + + assert!(matches!(error, IggyError::InvalidCommand)); + } + + #[test] + fn every_standard_code_accepts_its_own_class_and_rejects_the_other() { + for meta in COMMAND_TABLE { + let Ok(operation) = operation_for_code(meta.code, None) else { + // A replicated table entry with no `Operation` mapping stays + // unknown, and no declaration may rescue it. + for kind in [ + BinaryRequestKind::NonReplicated, + BinaryRequestKind::Replicated, + ] { + assert!( + matches!( + operation_for_code(meta.code, Some(kind)), + Err(IggyError::UnknownReplicatedCommand(_)) + ), + "{} must stay unknown when declared {kind}", + meta.name + ); + } + continue; + }; + let (matching, conflicting) = if operation == Operation::NonReplicated { + ( + BinaryRequestKind::NonReplicated, + BinaryRequestKind::Replicated, + ) + } else { + ( + BinaryRequestKind::Replicated, + BinaryRequestKind::NonReplicated, + ) + }; + + assert_eq!( + operation_for_code(meta.code, Some(matching)).unwrap(), + operation, + "{} must keep {operation:?} when declared {matching}", + meta.name + ); + assert!( + matches!( + operation_for_code(meta.code, Some(conflicting)), + Err(IggyError::InvalidCommand) + ), + "{} must not be redirected by a {conflicting} declaration", + meta.name + ); + } + } + + #[test] + fn logout_keeps_its_replicated_class_despite_the_table_entry() { + // The command table lists logout as non-replicated because classic + // framing runs it locally, but VSR commits it through `Operation::Logout`. + assert!( + matches!( + operation_for_code(LOGOUT_USER_CODE, Some(BinaryRequestKind::NonReplicated)), + Err(IggyError::InvalidCommand) + ), + "logout must not be reachable as a non-replicated raw request" + ); + assert_eq!( + operation_for_code(LOGOUT_USER_CODE, Some(BinaryRequestKind::Replicated)).unwrap(), + Operation::Logout + ); + } + + #[test] + fn vendor_replicated_code_cannot_reach_partition_routing() { + // Partition namespaces are decoded from a payload the SDK must + // understand, so no raw request can select one. + let error = + namespace_for_request(VENDOR_CODE, &Bytes::new(), Operation::SendMessages).unwrap_err(); + assert!(matches!(error, IggyError::FeatureUnavailable)); + } } diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs index 537d68de2..3e76e8646 100644 --- a/core/sdk/src/websocket/websocket_client.rs +++ b/core/sdk/src/websocket/websocket_client.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +use crate::binary::validate_binary_request_code; use crate::leader_aware::{LeaderRedirectionState, check_and_redirect_to_leader}; #[cfg(feature = "vsr")] use crate::session::ConsensusSession; @@ -39,7 +40,9 @@ use iggy_common::{ AutoLogin, ClientState, ConnectionString, Credentials, DiagnosticEvent, IggyDuration, IggyError, IggyTimestamp, WebSocketClientConfig, WebSocketConnectionStringOptions, }; -use iggy_common::{BinaryClient, BinaryTransport, PersonalAccessTokenClient, UserClient}; +use iggy_common::{ + BinaryClient, BinaryRequestKind, BinaryTransport, PersonalAccessTokenClient, UserClient, +}; use secrecy::ExposeSecret; use std::net::SocketAddr; use std::sync::Arc; @@ -136,61 +139,17 @@ impl BinaryTransport for WebSocketClient { } async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> { - let result = self.send_raw(code, payload.clone()).await; - if result.is_ok() { - return result; - } - - let error = result.unwrap_err(); - if !matches!( - error, - IggyError::Disconnected - | IggyError::EmptyResponse - | IggyError::Unauthenticated - | IggyError::StaleClient - | IggyError::NotConnected - | IggyError::CannotEstablishConnection - | IggyError::TcpError - | IggyError::ConnectionClosed - | IggyError::WebSocketSendError - | IggyError::WebSocketReceiveError - ) { - return Err(error); - } - - if !self.config.reconnection.enabled { - return Err(IggyError::Disconnected); - } - - #[cfg(feature = "vsr")] - if matches!(self.config.auto_login, AutoLogin::Disabled) { - return Err(error); - } - - self.disconnect().await?; - - #[cfg(feature = "vsr")] - let skip_auto_login = is_login_register_code(code); - #[cfg(feature = "vsr")] - if skip_auto_login { - *self.skip_auto_login_once.lock().await = true; - } - - { - let client_address = self.get_client_address_value().await; - info!( - "Reconnecting to the server: {} by client: {client_address}...", - self.config.server_address - ); - } + self.send_with_reconnect(None, code, payload).await + } - let reconnect = self.connect().await; - #[cfg(feature = "vsr")] - if skip_auto_login && reconnect.is_err() { - *self.skip_auto_login_once.lock().await = false; - } - reconnect?; - self.send_raw(code, payload).await + async fn send_raw_with_kind( + &self, + kind: BinaryRequestKind, + code: u32, + payload: Bytes, + ) -> Result<Bytes, IggyError> { + validate_binary_request_code(code)?; + self.send_with_reconnect(Some(kind), code, payload).await } fn get_heartbeat_interval(&self) -> IggyDuration { @@ -683,7 +642,78 @@ impl WebSocketClient { Ok(()) } - async fn send_raw(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError> { + /// Single reconnect-and-replay path shared by the typed and the raw entry + /// points, so `kind` survives the retry attempt unchanged. + async fn send_with_reconnect( + &self, + kind: Option<BinaryRequestKind>, + code: u32, + payload: Bytes, + ) -> Result<Bytes, IggyError> { + let result = self.send_raw(kind, code, payload.clone()).await; + if result.is_ok() { + return result; + } + + let error = result.unwrap_err(); + if !matches!( + error, + IggyError::Disconnected + | IggyError::EmptyResponse + | IggyError::Unauthenticated + | IggyError::StaleClient + | IggyError::NotConnected + | IggyError::CannotEstablishConnection + | IggyError::TcpError + | IggyError::ConnectionClosed + | IggyError::WebSocketSendError + | IggyError::WebSocketReceiveError + ) { + return Err(error); + } + + if !self.config.reconnection.enabled { + return Err(IggyError::Disconnected); + } + + #[cfg(feature = "vsr")] + if matches!(self.config.auto_login, AutoLogin::Disabled) { + return Err(error); + } + + self.disconnect().await?; + + #[cfg(feature = "vsr")] + let skip_auto_login = is_login_register_code(code); + #[cfg(feature = "vsr")] + if skip_auto_login { + *self.skip_auto_login_once.lock().await = true; + } + + { + let client_address = self.get_client_address_value().await; + info!( + "Reconnecting to the server: {} by client: {client_address}...", + self.config.server_address + ); + } + + let reconnect = self.connect().await; + #[cfg(feature = "vsr")] + if skip_auto_login && reconnect.is_err() { + *self.skip_auto_login_once.lock().await = false; + } + reconnect?; + self.send_raw(kind, code, payload).await + } + + #[cfg_attr(not(feature = "vsr"), expect(unused_variables))] + async fn send_raw( + &self, + kind: Option<BinaryRequestKind>, + code: u32, + payload: Bytes, + ) -> Result<Bytes, IggyError> { match self.get_state().await { ClientState::Shutdown => { trace!("Cannot send data. Client is shutdown."); @@ -720,7 +750,7 @@ impl WebSocketClient { .consensus_session .lock() .expect("consensus session mutex poisoned"); - crate::vsr::encode_contiguous_request(&mut consensus_session, code, &payload)? + crate::vsr::encode_contiguous_request(&mut consensus_session, kind, code, &payload)? }; trace!( "Sending {NAME} VSR request of size {} with code: {code}", @@ -875,6 +905,27 @@ mod tests { use super::*; use std::str::FromStr; + #[tokio::test] + async fn raw_binary_transport_rejects_session_control_before_io() { + let client = + WebSocketClient::from_connection_string("iggy+ws://iggy:[email protected]:8092").unwrap(); + + for kind in [ + BinaryRequestKind::NonReplicated, + BinaryRequestKind::Replicated, + ] { + let error = client + .send_raw_with_kind( + kind, + iggy_binary_protocol::codes::LOGIN_USER_CODE, + Bytes::new(), + ) + .await + .unwrap_err(); + assert!(matches!(error, IggyError::InvalidCommand)); + } + } + #[test] fn should_be_created_with_default_config() { let client = WebSocketClient::default(); diff --git a/examples/python/uv.lock b/examples/python/uv.lock index 639a0f563..a33c0238f 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.dev3" +version = "0.8.1.dev4" source = { directory = "../../foreign/python" } [package.metadata] diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml index 2dec15c20..c6e10b396 100644 --- a/foreign/python/Cargo.toml +++ b/foreign/python/Cargo.toml @@ -17,7 +17,7 @@ [package] name = "apache-iggy" -version = "0.8.1-dev3" +version = "0.8.1-dev4" edition = "2024" authors = ["Iggy Committers <[email protected]>"] license = "Apache-2.0" @@ -34,10 +34,13 @@ name = "stub_gen" path = "src/bin/stub_gen.rs" doc = false +[features] +vsr = ["iggy/vsr"] + [dependencies] bytes = "1.12.1" futures = "0.3.33" -iggy = { path = "../../core/sdk", version = "0.10.3-edge.2" } +iggy = { path = "../../core/sdk", version = "0.10.3-edge.3" } pyo3 = "0.29.0" pyo3-async-runtimes = { version = "0.29.0", features = [ "attributes", diff --git a/foreign/python/README.md b/foreign/python/README.md index f01754f6d..caf4740d0 100644 --- a/foreign/python/README.md +++ b/foreign/python/README.md @@ -58,6 +58,36 @@ maturin develop pytest tests/ -v # Run tests (requires iggy-server running) ``` +### VSR builds + +The Python extension uses the Rust SDK's compile-time framing mode. The default +build speaks classic framing. Build a VSR extension explicitly: + +```bash +uv run maturin develop --features vsr +``` + +A built wheel contains one framing mode; it cannot switch between classic and +VSR at runtime. + +Raw requests retain the compatibility method and provide an explicit extension +path: + +```python +from apache_iggy import BinaryRequestKind + +response = await client.send_binary_request(code, payload) +response = await client.send_binary_request_with_kind( + BinaryRequestKind.NonReplicated, + code, + payload, +) +``` + +With VSR framing, standard command tables remain authoritative. Unknown +non-replicated codes are forwarded for the server to judge. Unknown replicated +codes are unavailable until the protocol has a replicated extension registry. + ## Examples Refer to the [examples/python/](https://github.com/apache/iggy/tree/master/examples/python) directory for usage examples. diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi index 5b980c2e0..1a432ae46 100644 --- a/foreign/python/apache_iggy.pyi +++ b/foreign/python/apache_iggy.pyi @@ -29,6 +29,7 @@ __all__ = [ "AutoCommit", "AutoCommitAfter", "AutoCommitWhen", + "BinaryRequestKind", "ConsumerGroup", "ConsumerGroupDetails", "ConsumerGroupMember", @@ -744,6 +745,27 @@ class IggyClient: Raises: PyRuntimeError: If the command cannot be sent or the server returns an error. """ + def send_binary_request_with_kind( + self, kind: BinaryRequestKind, code: builtins.int, payload: builtins.bytes + ) -> collections.abc.Awaitable[bytes]: + r""" + Send a raw command with an explicit replication declaration. + + The declaration is inert on classic framing. With a VSR build, standard + command tables remain authoritative, unknown non-replicated codes are + forwarded, and unknown replicated codes are unavailable. + + Args: + kind: How the command executes, as `BinaryRequestKind`. + code: Command code as `int`. + payload: Request payload as `bytes`. + + Returns: + An awaitable that resolves to the raw response `bytes`. + + Raises: + PyRuntimeError: If the command cannot be sent or the server returns an error. + """ @typing.final class IggyConsumer: @@ -1018,6 +1040,24 @@ class UserInfoDetails: The username of the user. """ [email protected] +class BinaryRequestKind(enum.Enum): + r""" + How a raw binary request executes on the server. + """ + + NonReplicated = ... + r""" + Runs on the receiving node only, outside consensus. + """ + Replicated = ... + r""" + Replicated through consensus before it takes effect. Inert on classic + framing, where both kinds encode identical bytes; under `vsr` an + unknown code declared this way is rejected until the server grows a + replicated extension registry. + """ + @typing.final class UserStatus(enum.Enum): r""" diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml index d7f878885..d062ec1e7 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.dev3" +version = "0.8.1.dev4" 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" } @@ -141,6 +141,7 @@ addopts = ["-v", "--strict-markers", "--tb=short", "--disable-warnings"] markers = [ "integration: Integration tests that require a running server", "unit: Fast unit tests", + "vsr: Integration tests that require a VSR-built extension and server", ] filterwarnings = [ "ignore::DeprecationWarning", diff --git a/foreign/python/src/binary_request_kind.rs b/foreign/python/src/binary_request_kind.rs new file mode 100644 index 000000000..796fe140a --- /dev/null +++ b/foreign/python/src/binary_request_kind.rs @@ -0,0 +1,52 @@ +// 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 iggy::prelude::BinaryRequestKind as RustBinaryRequestKind; +use pyo3::prelude::*; +use pyo3_stub_gen::derive::gen_stub_pyclass_enum; + +/// How a raw binary request executes on the server. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[gen_stub_pyclass_enum] +#[pyclass(eq, from_py_object)] +pub enum BinaryRequestKind { + /// Runs on the receiving node only, outside consensus. + NonReplicated, + /// Replicated through consensus before it takes effect. Inert on classic + /// framing, where both kinds encode identical bytes; under `vsr` an + /// unknown code declared this way is rejected until the server grows a + /// replicated extension registry. + Replicated, +} + +impl From<BinaryRequestKind> for RustBinaryRequestKind { + fn from(kind: BinaryRequestKind) -> Self { + match kind { + BinaryRequestKind::NonReplicated => RustBinaryRequestKind::NonReplicated, + BinaryRequestKind::Replicated => RustBinaryRequestKind::Replicated, + } + } +} + +impl From<RustBinaryRequestKind> for BinaryRequestKind { + fn from(kind: RustBinaryRequestKind) -> Self { + match kind { + RustBinaryRequestKind::NonReplicated => BinaryRequestKind::NonReplicated, + RustBinaryRequestKind::Replicated => BinaryRequestKind::Replicated, + } + } +} diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs index b860f19b5..27f2d4f71 100644 --- a/foreign/python/src/client.rs +++ b/foreign/python/src/client.rs @@ -17,7 +17,8 @@ use bytes::Bytes; use iggy::prelude::{ - Consumer as RustConsumer, IggyClient as RustIggyClient, IggyMessage as RustMessage, + BinaryRequestKind as RustBinaryRequestKind, Consumer as RustConsumer, + IggyClient as RustIggyClient, IggyMessage as RustMessage, PollingStrategy as RustPollingStrategy, *, }; use pyo3::PyRef; @@ -29,6 +30,7 @@ use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; use std::str::FromStr; use std::sync::Arc; +use crate::binary_request_kind::BinaryRequestKind; use crate::consumer::{ AutoCommit, ConsumerGroup as PyConsumerGroup, ConsumerGroupDetails as PyConsumerGroupDetails, IggyConsumer, py_delta_to_iggy_duration, @@ -999,6 +1001,41 @@ impl IggyClient { Ok(Python::attach(|py| PyBytes::new(py, &response).unbind())) }) } + + /// Send a raw command with an explicit replication declaration. + /// + /// The declaration is inert on classic framing. With a VSR build, standard + /// command tables remain authoritative, unknown non-replicated codes are + /// forwarded, and unknown replicated codes are unavailable. + /// + /// Args: + /// kind: How the command executes, as `BinaryRequestKind`. + /// code: Command code as `int`. + /// payload: Request payload as `bytes`. + /// + /// Returns: + /// An awaitable that resolves to the raw response `bytes`. + /// + /// Raises: + /// PyRuntimeError: If the command cannot be sent or the server returns an error. + #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[bytes]", imports=("collections.abc")))] + fn send_binary_request_with_kind<'a>( + &self, + py: Python<'a>, + kind: BinaryRequestKind, + code: u32, + #[gen_stub(override_type(type_repr = "builtins.bytes"))] payload: Vec<u8>, + ) -> PyResult<Bound<'a, PyAny>> { + let inner = self.inner.clone(); + let kind = RustBinaryRequestKind::from(kind); + future_into_py(py, async move { + let response = inner + .send_binary_request_with_kind(kind, code, Bytes::from(payload)) + .await + .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?; + Ok(Python::attach(|py| PyBytes::new(py, &response).unbind())) + }) + } } define_stub_info_gatherer!(stub_info); diff --git a/foreign/python/src/lib.rs b/foreign/python/src/lib.rs index f831fd703..be458ced6 100644 --- a/foreign/python/src/lib.rs +++ b/foreign/python/src/lib.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +mod binary_request_kind; pub mod client; mod consumer; mod identifier; @@ -24,6 +25,7 @@ mod stream; mod topic; mod user; +use binary_request_kind::BinaryRequestKind; use client::IggyClient; use consumer::{ AutoCommit, AutoCommitAfter, AutoCommitWhen, ConsumerGroup, ConsumerGroupDetails, @@ -57,5 +59,6 @@ fn apache_iggy(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::<UserStatus>()?; m.add_class::<UserInfo>()?; m.add_class::<UserInfoDetails>()?; + m.add_class::<BinaryRequestKind>()?; Ok(()) } diff --git a/foreign/python/tests/test_raw_command.py b/foreign/python/tests/test_raw_command.py index 3313b6b2e..571a3c9ec 100644 --- a/foreign/python/tests/test_raw_command.py +++ b/foreign/python/tests/test_raw_command.py @@ -15,9 +15,17 @@ # specific language governing permissions and limitations # under the License. +from typing import cast + import pytest -from apache_iggy import IggyClient +from apache_iggy import BinaryRequestKind, IggyClient + +KINDS = [BinaryRequestKind.NonReplicated, BinaryRequestKind.Replicated] + +# No server registers a handler for this code, and it sits past every range the +# protocol assigns. +VENDOR_CODE = 60_001 @pytest.mark.asyncio @@ -34,6 +42,31 @@ async def test_raw_get_stats_returns_non_empty_response(iggy_client: IggyClient) assert response [email protected] +async def test_raw_replicated_declaration_is_ignored_on_classic_framing( + iggy_client: IggyClient, +): + # The kind is inert on classic framing, so a replicated declaration on a + # standard command still succeeds. + response = await iggy_client.send_binary_request_with_kind( + BinaryRequestKind.Replicated, 10, b"" + ) + + assert response + + [email protected] +async def test_raw_undefined_kind_is_rejected_before_sending( + iggy_client: IggyClient, +): + # The cast defeats the static signature on purpose: the runtime boundary + # itself must reject a value that is not a BinaryRequestKind. + with pytest.raises(TypeError): + await iggy_client.send_binary_request_with_kind( + cast(BinaryRequestKind, "auto"), 1, b"" + ) + + @pytest.mark.asyncio @pytest.mark.parametrize("code", [38, 39, 40, 44, 45]) async def test_raw_session_control_code_is_rejected(iggy_client: IggyClient, code: int): @@ -42,6 +75,29 @@ async def test_raw_session_control_code_is_rejected(iggy_client: IggyClient, cod @pytest.mark.asyncio -async def test_raw_unknown_code_is_rejected_by_server(iggy_client: IggyClient): [email protected]("kind", KINDS) [email protected]("code", [38, 39, 40, 44, 45]) +async def test_raw_session_control_code_with_kind_is_rejected( + iggy_client: IggyClient, kind: BinaryRequestKind, code: int +): + with pytest.raises(RuntimeError, match="(?i)invalid command"): + await iggy_client.send_binary_request_with_kind(kind, code, b"") + + [email protected] +async def test_raw_vendor_code_is_rejected_by_server(iggy_client: IggyClient): with pytest.raises(RuntimeError, match="(?i)invalid command"): - await iggy_client.send_binary_request(60_000, b"") + await iggy_client.send_binary_request_with_kind( + BinaryRequestKind.NonReplicated, VENDOR_CODE, b"" + ) + + # The rejection is request-level, so the connection stays usable. + assert await iggy_client.send_binary_request(1, b"") == b"" + + [email protected] +async def test_raw_replicated_vendor_code_has_no_handler(iggy_client: IggyClient): + with pytest.raises(RuntimeError): + await iggy_client.send_binary_request_with_kind( + BinaryRequestKind.Replicated, VENDOR_CODE, b"" + ) diff --git a/foreign/python/tests/test_raw_command_vsr.py b/foreign/python/tests/test_raw_command_vsr.py new file mode 100644 index 000000000..7436f345f --- /dev/null +++ b/foreign/python/tests/test_raw_command_vsr.py @@ -0,0 +1,70 @@ +# 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. + +import os + +import pytest + +from apache_iggy import BinaryRequestKind, IggyClient + +VENDOR_CODE = 60_001 + +pytestmark = [ + pytest.mark.asyncio, + pytest.mark.vsr, + pytest.mark.skipif( + os.getenv("IGGY_TEST_PROTOCOL") != "vsr", + reason="requires a VSR-built Python extension and VSR server", + ), +] + + +async def test_vsr_raw_request_kinds_preserve_session_sequence( + iggy_client: IggyClient, unique_name +): + assert await iggy_client.send_binary_request(1, b"") == b"" + assert ( + await iggy_client.send_binary_request_with_kind( + BinaryRequestKind.NonReplicated, 1, b"" + ) + == b"" + ) + + for _ in range(3): + with pytest.raises(RuntimeError, match="(?i)invalid command"): + await iggy_client.send_binary_request_with_kind( + BinaryRequestKind.NonReplicated, VENDOR_CODE, b"vendor-body" + ) + + with pytest.raises(RuntimeError, match="(?i)feature.*unavailable"): + await iggy_client.send_binary_request_with_kind( + BinaryRequestKind.Replicated, VENDOR_CODE, b"vendor-body" + ) + + with pytest.raises(RuntimeError, match="(?i)invalid command"): + await iggy_client.send_binary_request_with_kind( + BinaryRequestKind.Replicated, 10, b"" + ) + + with pytest.raises(RuntimeError, match="(?i)invalid command"): + await iggy_client.send_binary_request_with_kind( + BinaryRequestKind.NonReplicated, 38, b"" + ) + + # A metadata mutation must still use request ID 1. Advancing the counter + # for any non-replicated request would create a gap and stall this call. + await iggy_client.create_stream(unique_name("raw-vsr-")) diff --git a/foreign/python/uv.lock b/foreign/python/uv.lock index 55960c278..72fc9918d 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.dev3" +version = "0.8.1.dev4" source = { editable = "." } [package.optional-dependencies]
