feat(sdk): consume the request counter for partition operations

core/binary_protocol/src/consensus/header.rs

@@ -287,7 +287,9 @@ pub struct RequestHeader {
     /// catch a `request` number reused for a different operation: a retry that
     /// disagrees with the stamp of the cached reply is refused rather than
     /// answered with the wrong reply. Zero means unstamped, which disables the
-    /// comparison; the wire currently sends zero.
+    /// comparison. The Rust SDK stamps the ops the table dedups; partition and
+    /// non-replicated ops, and the other SDKs, leave it zero. The server
+    /// verifies any nonzero stamp before routing.
     pub request_checksum: u128,
     pub timestamp: u64,
     pub request: u64,

core/consensus/src/client_table.rs

@@ -578,8 +578,10 @@ pub struct ClientTable {
 
 /// Whether two integrity stamps for the same request number disagree.
 ///
-/// Zero means unstamped (the wire integrity fields are zeroed today), and an
-/// unstamped side carries no evidence either way, so it never conflicts.
+/// Zero means unstamped, and an unstamped side carries no evidence either way,
+/// so it never conflicts. The Rust SDK stamps the ops this table dedups;
+/// partition ops and the other SDKs still send zero, so a conflict is only ever
+/// detectable between two stamped frames.
 const fn checksums_conflict(stored: u128, received: u128) -> bool {
     stored != 0 && received != 0 && stored != received
 }
@@ -2681,6 +2683,36 @@ mod tests {
         assert_eq!(table.get_watermark(1), Some(9));
     }
 
+    // The shape a client that spends request ids off the metadata plane
+    // produces: partition-plane ids never reach this table, so the next
+    // metadata request arrives with a gap under it. It executes, moves the
+    // watermark to itself, and its retry still replays the original reply --
+    // gaps cost the skipped ids and nothing else.
+    #[test]
+    fn check_request_dedups_a_metadata_request_that_arrives_after_a_gap() {
+        let (mut table, epoch) = table_with_client();
+        table.commit_reply(1, TEST_USER_ID, make_reply_for(1, 1, 11));
+        // Requests 2..=5 went to the partition plane, which keeps no table.
+        assert!(matches!(
+            table.check_request(1, epoch, 6, 0),
+            RequestStatus::New
+        ));
+
+        table.commit_reply(1, TEST_USER_ID, make_reply_for(1, 6, 12));
+        assert_eq!(table.get_watermark(1), Some(6));
+        match table.check_request(1, epoch, 6, 0) {
+            RequestStatus::Duplicate(cached) => {
+                assert_eq!(cached.header().request, 6);
+                assert_eq!(
+                    cached.header().commit,
+                    12,
+                    "the original reply, not a re-run"
+                );
+            }
+            other => panic!("expected the gapped request to dedup, got {other:?}"),
+        }
+    }
+
     #[test]
     fn check_request_duplicate_at_watermark() {
         let (mut table, epoch) = table_with_client();

core/integration/tests/cluster/client_table_adversarial.rs

@@ -17,8 +17,9 @@
 
 //! Adversarial specs against the VSR client table's at-most-once guarantees.
 //!
-//! Both assert the dedup contract a retrying client needs at the table's two
-//! resource edges.
+//! Each asserts the dedup contract a retrying client needs: the first two at
+//! the table's resource edges, the third across the request-id gaps a client
+//! that also produces leaves behind.
 //!
 //! 1. Capacity: a full table evicts the entry with the oldest commit. Eviction
 //!    keeps that client's request watermark (and the watermark's reply when the
@@ -34,26 +35,32 @@
 //!    fact that neither a state transfer nor a restart carries the deeper
 //!    history, are pinned by the consensus crate's unit tests; this file pins
 //!    the depth the budget buys on a live server.
+//! 3. Sequence gaps: partition sends spend request ids on a plane that keeps no
+//!    client table, so the next metadata request arrives with a gap under it.
+//!    The entry stores a watermark, not a contiguous sequence, so the gapped
+//!    request must execute once and its retry must replay that reply.
 //!
 //! The frames are hand-crafted on raw TCP sockets, same technique and frame
 //! builders as the clients-table restart tests, because the churn needs
 //! per-connection client identities and byte-level replay the SDK does not
 //! expose. The builders here are parameterized by client id, which is why the
 //! restart tests' fixed-identity helpers are not reused directly.
 
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut};
 use consensus::client_table::REPLY_RING_CAPACITY;
 use iggy::prelude::*;
 use iggy_binary_protocol::codec::{WireDecode, WireEncode};
 use iggy_binary_protocol::consensus::{
     Command, Operation, ReplyHeader, RequestHeader, read_size_field, result_code,
     result_section_len,
 };
+use iggy_binary_protocol::requests::messages::{RawMessage, SendMessagesEncoder};
 use iggy_binary_protocol::requests::streams::CreateStreamRequest;
 use iggy_binary_protocol::requests::users::LoginRegisterRequest;
 use iggy_binary_protocol::responses::users::LoginRegisterResponse;
 use iggy_binary_protocol::{
-    ClientVersionInfo, HEADER_SIZE, IGGY_PROTOCOL_VERSION, WireName, WireOptions,
+    ClientVersionInfo, HEADER_SIZE, IGGY_PROTOCOL_VERSION, WireIdentifier, WireName, WireOptions,
+    WirePartitioning,
 };
 use integration::harness::TestHarness;
 use integration::iggy_harness;
@@ -72,6 +79,13 @@ const CLIENT_A: u128 = 0xA11CE0001;
 /// eviction of `CLIENT_A`'s entry.
 const CHURN_CLIENTS: [u128; 3] = [0xB0B0001, 0xB0B0002, 0xB0B0003];
 
+/// The topic the gap spec produces into, and how many batches it sends before
+/// its next metadata request. One batch already gaps the sequence; four makes
+/// the distance the table has to tolerate unmistakable.
+const GAP_STREAM: &str = "adv-m-gap";
+const GAP_TOPIC: &str = "adv-m-gap-topic";
+const GAP_BATCHES: u64 = 4;
+
 /// Budget for one committed round-trip (covers transient replays while the
 /// single node elects itself after boot).
 const COMMIT_BUDGET: Duration = Duration::from_secs(15);
@@ -196,6 +210,65 @@ async fn given_a_retry_past_the_reply_floor_when_the_retention_budget_still_hold
     }
 }
 
+/// A metadata request that lands above a gap the partition plane opened must
+/// still deduplicate.
+///
+/// Every replicated request on a session spends an id, but only the metadata
+/// plane keeps a client table, so a session that produces reaches its next
+/// metadata request several ids above the watermark. The entry records the
+/// highest committed request rather than a contiguous run, so the gapped
+/// request executes once and the retry of its exact frame is answered from the
+/// cache. A committed duplicate-name rejection is the proof of re-execution.
+#[iggy_harness(cluster_nodes = 1)]
+async fn given_partition_batches_spent_request_ids_when_a_metadata_request_is_retried_should_replay_from_cache(
+    harness: &mut TestHarness,
+) {
+    // The topic the batches target is set up over the SDK on its own session,
+    // so the raw session below spends ids only on what this spec is about.
+    let setup = harness.tcp_root_client().await.unwrap();
+    setup
+        .create_stream(GAP_STREAM)
+        .await
+        .expect("create stream");
+    setup
+        .create_topic(
+            &Identifier::named(GAP_STREAM).unwrap(),
+            GAP_TOPIC,
+            &TopicCreateOptions {
+                partitions_count: Some(1),
+                ..TopicCreateOptions::default()
+            },
+        )
+        .await
+        .expect("create topic");
+    drop(setup);
+
+    let addr = tcp_addr(harness);
+    let (mut stream, session) = register(addr, CLIENT_A).await;
+
+    for request in 1..=GAP_BATCHES {
+        let batch = send_messages_payload(u128::from(request));
+        commit_batch(&mut stream, CLIENT_A, session, request, &batch).await;
+    }
+
+    let payload = create_stream_payload("adv-m-after-gap");
+    let gapped_request = GAP_BATCHES + 1;
+    let committed = commit_request(&mut stream, CLIENT_A, session, gapped_request, &payload).await;
+    let replay = replay_request(&mut stream, CLIENT_A, session, gapped_request, &payload).await;
+
+    match replay {
+        Verdict::Success(replayed) => {
+            assert_replayed_from_cache(&committed, &replayed, gapped_request);
+        }
+        other => panic!(
+            "a metadata request above a partition-plane gap lost its dedup: request \
+             {gapped_request} committed after {GAP_BATCHES} batches spent the ids below it, \
+             so its retry must be answered from the cache rather than re-executed; the \
+             watermark is a high-water mark, not a contiguity check; got {other:?}"
+        ),
+    }
+}
+
 fn tcp_addr(harness: &TestHarness) -> SocketAddr {
     harness
         .server()
@@ -211,6 +284,31 @@ fn create_stream_payload(name: &str) -> Bytes {
     .to_bytes()
 }
 
+/// One canonical batch for the gap spec's partition, in the shape
+/// `SendMessagesEncoder` writes and admission verifies. `message_id` keeps
+/// successive batches distinct on the wire.
+fn send_messages_payload(message_id: u128) -> Bytes {
+    let stream_id = WireIdentifier::named(GAP_STREAM).unwrap();
+    let topic_id = WireIdentifier::named(GAP_TOPIC).unwrap();
+    let partitioning = WirePartitioning::PartitionId(0);
+    let messages = [RawMessage {
+        id: message_id,
+        origin_timestamp: 0,
+        headers: None,
+        payload: b"adv-m-gap-batch",
+    }];
+
+    let mut buf = BytesMut::with_capacity(SendMessagesEncoder::encoded_size(
+        &stream_id,
+        &topic_id,
+        &partitioning,
+        &messages,
+    ));
+    SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages)
+        .expect("send batch encodes");
+    buf.freeze()
+}
+
 fn request_header(client: u128, session: u64, request: u64, body_len: usize) -> RequestHeader {
     RequestHeader {
         command: Command::Request,
@@ -300,6 +398,40 @@ async fn commit_request(
     }
 }
 
+/// Send one batch on the partition plane and require it committed. The reply
+/// is not result-framed and carries no body, so a zero status is the whole
+/// verdict; what this spec needs from it is only the request id it spends.
+async fn commit_batch(
+    stream: &mut TcpStream,
+    client: u128,
+    session: u64,
+    request: u64,
+    body: &Bytes,
+) {
+    let header = RequestHeader {
+        command: Command::Request,
+        operation: Operation::SendMessages,
+        size: u32::try_from(HEADER_SIZE + body.len()).unwrap(),
+        client,
+        session,
+        request,
+        ..Default::default()
+    };
+    let deadline = Instant::now() + COMMIT_BUDGET;
+    loop {
+        match exchange(stream, &header, body).await {
+            Exchange::Reply { status: 0, .. } => return,
+            Exchange::Reply { status, .. } if is_transient(status) && Instant::now() < deadline => {
+                sleep(RETRY_PAUSE).await;
+            }
+            other => panic!(
+                "batch at request {request} did not commit: {:?}",
+                other.verdict()
+            ),
+        }
+    }
+}
+
 /// Replay `request` and return the first non-transient verdict. Unlike
 /// `commit_request` this never panics on a committed rejection: the rejection
 /// IS the observation the red specs are after.

core/sdk/src/quic/quic_client.rs

@@ -919,8 +919,8 @@ impl QuicClient {
                 // construction, so replaying the SAME request header on a fresh
                 // bidi cannot double-commit), and it no longer abandons a bidi
                 // whose op is still committing. Silence therefore is NOT a
-                // retry signal: partition ops share one request id and have no
-                // reply cache, so resending a silently-unanswered request whose
+                // retry signal: the partition plane has no dedup or reply
+                // cache, so resending a silently-unanswered request whose
                 // first attempt was buffered and later commits would commit it
                 // twice (duplicate `SendMessages`, or a succeeded delete coming
                 // back as terminal `ConsumerOffsetNotFound`). A silent deadline

core/sdk/src/vsr.rs

@@ -94,26 +94,22 @@ pub(crate) fn encode_request_header(
                     session.current_request_id(),
                     session.session().unwrap_or(0),
                 )
-            } else if operation.is_partition() {
-                // Partition ops replicate in their own per-partition group,
-                // which is at-least-once with no `ClientTable` dedup -- the
-                // metadata table never records their request ids, so there
-                // is nothing for a consumed id to deduplicate against. Every
-                // partition request on a session therefore carries the id
-                // the next metadata op will claim, and a partition-plane
-                // replay is at-least-once.
-                let session_id = session.session().ok_or(IggyError::Unauthenticated)?;
-                (operation, session.current_request_id(), session_id)
             } else {
+                // Partition ops consume an id too, even though nothing dedups
+                // them yet: dedup needs each send to carry a distinct number,
+                // and the metadata watermark tolerates the resulting gaps
+                // (`client_table.rs`: "There is no `RequestGap`").
                 let session_id = session.session().ok_or(IggyError::Unauthenticated)?;
                 (operation, session.next_request_id(), session_id)
             }
         }
     };
-    // Stamped only for ops the server's `ClientTable` dedups. Partition ops are
-    // at-least-once with no reply cache to poison, and theirs are the large payloads,
-    // already covered client-side by `batch_checksum` over the same bytes.
-    // NonReplicated ops bypass dedup too.
+    // Stamped only for the ops the server's `ClientTable` dedups, and only by this
+    // SDK: the others leave the field zero, which the server reads as unstamped.
+    // Partition ops are the large payloads and already carry `batch_checksum` over
+    // the same bytes, and nothing dedups them, so hashing here would only buy the
+    // server a second full-payload pass in `verify_request_checksum`. NonReplicated
+    // ops bypass dedup too.
     let request_checksum = if operation.is_partition() || operation == Operation::NonReplicated {
         0
     } else {
@@ -362,7 +358,9 @@ fn read_window_field(header_bytes: &[u8; HEADER_SIZE], offset: usize) -> u32 {
 mod tests {
     use super::*;
     use crate::session::ConsensusSession;
-    use iggy_binary_protocol::codes::{CREATE_STREAM_CODE, GET_STREAM_CODE, PING_CODE};
+    use iggy_binary_protocol::codes::{
+        CREATE_STREAM_CODE, GET_STREAM_CODE, PING_CODE, SEND_MESSAGES_CODE,
+    };
     use iggy_binary_protocol::requests::streams::CreateStreamRequest;
     use iggy_binary_protocol::requests::users::LoginRegisterRequest;
     use iggy_binary_protocol::version::IGGY_PROTOCOL_VERSION;
@@ -503,25 +501,57 @@ mod tests {
 
     #[test]
     fn request_checksum_is_stamped_only_for_deduped_operations() {
-        // The stamp exists to stop a reused `request` number returning the wrong
-        // cached reply, so it is worth its hashing pass only where `ClientTable`
-        // dedups. Partition payloads are the large ones and carry `batch_checksum`
-        // over the same bytes already; hashing them again is pure cost.
+        // The stamp exists to stop a reused `request` number matching a dedup
+        // entry recorded for different bytes, so it is worth its hashing pass only
+        // where `ClientTable` dedups. Partition ops are the large payloads and
+        // already carry `batch_checksum` over the same bytes; NonReplicated ops
+        // bypass dedup. Neither stamps.
         let mut session = ConsensusSession::with_client_id(42);
         session.bind(99);
         let payload = Bytes::from_static(b"payload");
 
-        let deduped =
+        let metadata =
             encode_contiguous_request(&mut session, CREATE_STREAM_CODE, &payload).unwrap();
+        // Hashed against the framed body rather than the `payload` the encoder was
+        // handed, so a slice mistake between what is stamped and what is sent fails
+        // here instead of reaching the server's `verify_request_checksum`.
         assert_eq!(
-            decode_request_header(&deduped).request_checksum,
-            u128::from(calculate_checksum(&payload)),
+            decode_request_header(&metadata).request_checksum,
+            u128::from(calculate_checksum(&metadata[HEADER_SIZE..])),
         );
 
+        let partition =
+            encode_contiguous_request(&mut session, SEND_MESSAGES_CODE, &payload).unwrap();
+        assert_eq!(decode_request_header(&partition).request_checksum, 0);
+
         let ping = encode_contiguous_request(&mut session, PING_CODE, &Bytes::new()).unwrap();
         assert_eq!(decode_request_header(&ping).request_checksum, 0);
     }
 
+    #[test]
+    fn partition_request_consumes_the_request_counter() {
+        // Dedup identity requires each send to carry a distinct id, so
+        // partition ops advance the counter exactly like metadata ops and
+        // the two planes interleave on one sequence.
+        let mut session = ConsensusSession::with_client_id(42);
+        session.bind(99);
+        let payload = Bytes::from_static(b"batch");
+
+        let first = encode_contiguous_request(&mut session, SEND_MESSAGES_CODE, &payload).unwrap();
+        let second = encode_contiguous_request(&mut session, SEND_MESSAGES_CODE, &payload).unwrap();
+        assert_eq!(decode_request_header(&first).request, 1);
+        assert_eq!(decode_request_header(&second).request, 2);
+
+        let metadata_payload = CreateStreamRequest {
+            name: WireName::new("stream").unwrap(),
+            options: WireOptions::empty(),
+        }
+        .to_bytes();
+        let metadata =
+            encode_contiguous_request(&mut session, CREATE_STREAM_CODE, &metadata_payload).unwrap();
+        assert_eq!(decode_request_header(&metadata).request, 3);
+    }
+
     #[test]
     fn ping_uses_non_replicated_operation() {
         let mut session = ConsensusSession::with_client_id(42);

core/server/src/dispatch.rs

@@ -1598,8 +1598,9 @@ pub(crate) async fn dispatch_partition_request<B, MJ, S, SB>(
         // Header validation requires `session > 0 && request > 0` for
         // non-register ops. The partition plane itself is sessionless
         // (at-least-once, no `ClientTable` dedup), so the bound VSR
-        // session merely satisfies validation, and a zero request id
-        // (the SDK does not number data-plane ops) is normalized.
+        // session merely satisfies validation. Current SDKs do number
+        // partition ops, but older and internal callers may still send
+        // zero, so a zero id is normalized to the compatibility value 1.
         new_header.session = bound_session;
         new_header.request = new_header.request.max(1);
     });