diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs index a9d20ed3c9..233396c182 100644 --- a/core/binary_protocol/src/consensus/header.rs +++ b/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, diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs index 67cb42e25c..0737e23bf3 100644 --- a/core/consensus/src/client_table.rs +++ b/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(); diff --git a/core/integration/tests/cluster/client_table_adversarial.rs b/core/integration/tests/cluster/client_table_adversarial.rs index 5181c8fea0..bf6c3f573c 100644 --- a/core/integration/tests/cluster/client_table_adversarial.rs +++ b/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,6 +35,10 @@ //! 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 @@ -41,7 +46,7 @@ //! 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}; @@ -49,11 +54,13 @@ 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. diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs index 28fb6529da..e69e94af6c 100644 --- a/core/sdk/src/quic/quic_client.rs +++ b/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 diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs index 2e8a210977..876457ad8a 100644 --- a/core/sdk/src/vsr.rs +++ b/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); diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs index 5e7bca80e8..44cfd93e2c 100644 --- a/core/server/src/dispatch.rs +++ b/core/server/src/dispatch.rs @@ -1598,8 +1598,9 @@ pub(crate) async fn dispatch_partition_request( // 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); }); diff --git a/core/server/src/http/session.rs b/core/server/src/http/session.rs index 2e104268d4..aa1db608c3 100644 --- a/core/server/src/http/session.rs +++ b/core/server/src/http/session.rs @@ -99,7 +99,10 @@ pub(in crate::http) struct HttpSession { /// Serializes this session's writes: the guarded value is the NEXT request /// id. A `tokio::sync::Mutex` because the write path holds it across the /// submit `.await` so each session's request numbers reach the primary in - /// order and stay gap-free for the depth-1 consensus dedup. + /// order. Ordering is what matters, not contiguity: the client table dedups + /// on a watermark (see `submit.rs`), so gaps are free but an id overtaken by + /// a larger one would arrive at or below the watermark and be refused as a + /// duplicate. pub(in crate::http) gate: Mutex, /// Next data-plane request id. A separate, gate-free counter: partition ops /// are at-least-once with no consensus dedup, so the id only correlates the diff --git a/core/simulator/src/client.rs b/core/simulator/src/client.rs index 6f19bd1077..1b1501aeed 100644 --- a/core/simulator/src/client.rs +++ b/core/simulator/src/client.rs @@ -55,24 +55,15 @@ use server_common::sharding::{IggyNamespace, METADATA_GROUP}; use server_common::{Message, iobuf::Owned}; use std::cell::Cell; -/// Partition-plane request ids are offset into the top half of the `u64` space -/// so they never collide with the small, contiguous metadata ids in the -/// auditor's `(client, request)` map. The metadata sequence would have to reach -/// `2^63` to overlap, which no run approaches. -const PARTITION_ID_BASE: u64 = 1 << 63; - // TODO: Proper client which implements the full client SDK API pub struct SimClient { client_id: u128, - /// Contiguous `1, 2, 3, …` request ids for metadata/replicated ops, the - /// sequence the server's `ClientTable` dedups and requires gap-free. + /// Monotonic `1, 2, 3, …` request ids shared by every replicated op, + /// metadata and partition alike, which is what the SDKs send. One sequence + /// also issues each id exactly once per client, so a delayed or duplicated + /// partition reply can never carry the `(client, request)` key of a live + /// metadata entry in the auditor's map. See [`SimClient::next_request_id`]. request_counter: Cell, - /// Separate id sequence for partition-plane ops, offset into a disjoint - /// range ([`PARTITION_ID_BASE`]). The partition plane has no client-table - /// dedup and treats the id as an opaque echo, so a partition id never - /// collides with a metadata id, even under reply duplication. See - /// [`SimClient::request_id_for`]. - partition_counter: Cell, /// Deterministic per-message id source for produced messages. The real SDK /// mints a random UUID for a zero message id before encoding; that mint is /// unseeded, so under the deterministic executor a produce's replicated @@ -100,7 +91,6 @@ impl SimClient { Self { client_id, request_counter: Cell::new(0), - partition_counter: Cell::new(0), message_counter: Cell::new(0), session: Cell::new(0), shell_wire: Cell::new(false), @@ -157,30 +147,21 @@ impl SimClient { self.session.set(session); } - /// Assign the wire request id for `operation`, keyed by plane. + /// Assign the wire request id for the next replicated op. /// - /// Metadata/replicated ops advance a contiguous `1, 2, 3, …` counter, matching - /// the real SDK. Gaps are admitted rather than fatal (`check_request` answers - /// `New` to anything above the watermark; there is no `RequestGap`), but the - /// dedup ring is sized for a contiguous sequence. Partition ops are at-least-once - /// with no dedup and the server - /// treats their id as an opaque echo, so they draw from a separate counter - /// offset into a disjoint range ([`PARTITION_ID_BASE`]). A partition id can - /// therefore never equal a metadata id, so a delayed or duplicated partition - /// reply is never misattributed to a metadata entry in the auditor's - /// `(client, request)` map (which would trip the group guard and drop a - /// live metadata op). This holds regardless of reply duplication, not only - /// while clients are one-in-flight. - fn request_id_for(&self, operation: Operation) -> u64 { - if operation.is_partition() { - let next = self.partition_counter.get() + 1; - self.partition_counter.set(next); - PARTITION_ID_BASE + next - } else { - let next = self.request_counter.get() + 1; - self.request_counter.set(next); - next - } + /// Every replicated op advances one counter, metadata and partition alike, + /// which is what the SDKs send: a partition op needs its own number for a + /// retry to be recognisable, and the ids it spends cost the metadata plane + /// nothing, because `ClientTable` admits anything above the watermark + /// (`client_table.rs`: "There is no `RequestGap`"). + /// + /// `NonReplicated` reads never reach here — the poll path builds its own + /// header and reads the counter without advancing it, matching the SDK, + /// since the server ignores the id for ops the table never sees. + fn next_request_id(&self) -> u64 { + let next = self.request_counter.get() + 1; + self.request_counter.set(next); + next } fn session_id(&self) -> u64 { @@ -623,10 +604,10 @@ impl SimClient { /// `count` messages from offset 0 of `group`'s partition. /// /// A `NonReplicated` read: the command code sits in the header's - /// `reserved` prefix, and the request id ECHOES the current metadata - /// counter without advancing it (matching the SDK), so a read never - /// gaps the replicated sequence the server's `ClientTable` requires - /// gap-free. Requires a bound session (polls are auth-gated). + /// `reserved` prefix, and the request id ECHOES the current counter + /// without advancing it (matching the SDK). The server ignores the id for + /// ops its `ClientTable` never sees, so burning one would buy nothing. + /// Requires a bound session (polls are auth-gated). /// /// # Panics /// Panics if the session is unbound or the request buffer is invalid. @@ -779,7 +760,7 @@ impl SimClient { request_checksum: 0, timestamp: 0, // TODO: Use actual timestamp session: self.session_id(), - request: self.request_id_for(operation), + request: self.next_request_id(), group, ..Default::default() } @@ -813,29 +794,29 @@ fn namespace_ids(ns: IggyNamespace) -> (WireIdentifier, WireIdentifier, Option = interleaved + .into_iter() + .map(|operation| client.header(operation, METADATA_GROUP, 0).request) + .collect(); - // Metadata ops advance (1, 2, 3); interleaved sends draw their own - // disjoint sequence and leave the metadata counter untouched. - assert_eq!(client.request_id_for(Operation::CreateStream), 1); - assert_eq!( - client.request_id_for(Operation::SendMessages), - PARTITION_ID_BASE + 1 - ); - assert_eq!(client.request_id_for(Operation::CreateStream), 2); - assert_eq!( - client.request_id_for(Operation::SendMessages), - PARTITION_ID_BASE + 2 - ); - assert_eq!(client.request_id_for(Operation::CreateStream), 3); + assert_eq!(ids, vec![1, 2, 3, 4, 5], "plane must not fork the sequence"); } } diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 128278899a..4e006c573d 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -1982,9 +1982,11 @@ mod tests { // `WireConsumer` discriminant so every such request was dropped unparsed (see // `ops::sample_consumer_kind`); and per-stream seeds moving from XOR salts to // [`SimSeeds`] alongside `Xoshiro256Plus` becoming `Xoshiro256PlusPlus`, which - // together remap every stream. + // together remap every stream; and partition ops drawing from the one shared + // request counter instead of a separate sequence based at `1<<63`, which + // renumbers every partition request id and so every reply header in the trace. assert_eq!( - h1, 0x1376_D480_4A3F_E6A9, + h1, 0x5C2B_6057_2DA9_908B, "workload reply hash drifted from locked baseline" ); } diff --git a/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs b/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs index 58e1bed51f..b5996d7a52 100644 --- a/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs +++ b/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs @@ -203,14 +203,8 @@ private SessionFrame ReplicatedFrameLocked(VsrOperation operation) var sessionId = _session ?? throw VsrError.Exception(VsrError.UNAUTHENTICATED, "A replicated request requires a bound consensus session."); - // Partition ops replicate in their own per-partition group with no client-table dedup, so they too - // must leave the metadata counter untouched. Only metadata operations and logout consume an id: the - // server tracks request ids for those alone, and it accepts any id above the client's watermark. - if (operation.IsPartition()) - { - return new SessionFrame(_clientId, _requestCounter, sessionId); - } - + // Partition ops consume an id too, even though no partition-plane dedup exists yet: dedup needs + // each send to carry a distinct number, and the metadata watermark tolerates the gaps. var requestId = _requestCounter; _requestCounter = checked(_requestCounter + 1); diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs b/foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs index 3de62116b2..6dce9d3af3 100644 --- a/foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs +++ b/foreign/csharp/Iggy_SDK/Vsr/VsrHeader.cs @@ -21,8 +21,10 @@ namespace Apache.Iggy.Vsr; /// /// The 256-byte consensus header, read and written by wire offset. Offsets mirror -/// core/binary_protocol/src/consensus/header.rs. Checksums stay zero: the server does not verify -/// them for client frames. +/// core/binary_protocol/src/consensus/header.rs. Checksums stay zero: the frame and body checksums +/// are not read on the client request path, and request_checksum treats zero as unstamped, which +/// opts out of the server's payload comparison. Stamping it is optional -- the Rust SDK does so for +/// deduped operations, this SDK does not yet. /// internal static class VsrHeader { diff --git a/foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs b/foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs index f46eeca25c..de5d9ed726 100644 --- a/foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs +++ b/foreign/csharp/Iggy_SDK/Vsr/VsrOperation.cs @@ -68,7 +68,6 @@ internal static class VsrOperations { private const byte InternalStart = (byte)VsrOperation.CreateTopicWithAssignments; private const byte MetadataStart = (byte)VsrOperation.CreateStream; - private const byte PartitionStart = (byte)VsrOperation.SendMessages; /// /// Non-replicated codes this build knows to leave no server-side state behind, so re-sending one after a @@ -245,16 +244,6 @@ or VsrOperation.JoinConsumerGroup or VsrOperation.LeaveConsumerGroup; } - /// - /// Data-plane operations routed by namespace to the shard owning the partition. - /// is deliberately neither metadata nor partition: the - /// server resolves it to an internal TruncatePartition, yet it still carries a packed namespace. - /// - internal static bool IsPartition(this VsrOperation operation) - { - return (byte)operation >= PartitionStart; - } - /// /// Whether a reply for this operation leads its body with the committed result section. Metadata ops /// always do; on the partition plane only the consumer-offset ops do. Register is result-framed only diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs index 3ff5256c85..aa91e558ff 100644 --- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs +++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs @@ -119,17 +119,32 @@ public void NextRequestId_BeforeBindThrows() } [Fact] - public void Resolve_DoesNotConsumeAnIdForNonReplicatedOrPartitionOps() + public void Resolve_DoesNotConsumeAnIdForNonReplicatedOps() { var session = new ConsensusSession(1); session.Resolve(VsrOperation.Register); session.Bind(10); Assert.Equal(1UL, session.Resolve(VsrOperation.NonReplicated).RequestId); - Assert.Equal(1UL, session.Resolve(VsrOperation.SendMessages).RequestId); Assert.Equal(1UL, session.RequestCounter); } + [Fact] + public void Resolve_PartitionOpsConsumeADistinctIdPerSend() + { + // Dedup identity requires each send to carry a distinct number, so + // partition ops advance the counter exactly like metadata ops and the + // two planes interleave on one sequence. + var session = new ConsensusSession(1); + session.Resolve(VsrOperation.Register); + session.Bind(10); + + Assert.Equal(1UL, session.Resolve(VsrOperation.SendMessages).RequestId); + Assert.Equal(2UL, session.Resolve(VsrOperation.SendMessages).RequestId); + Assert.Equal(3UL, session.Resolve(VsrOperation.CreateStream).RequestId); + Assert.Equal(4UL, session.RequestCounter); + } + [Fact] public void Bind_TwiceThrows() { diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs index 278037041a..4a6e54e0a9 100644 --- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs +++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs @@ -162,16 +162,18 @@ public void Encode_LogoutAdvancesTheCounter() } [Fact] - public void Encode_PartitionOpDoesNotAdvanceTheCounter() + public void Encode_PartitionOpConsumesADistinctId() { var session = BoundSession(); var payload = VsrTestPayloads.SendMessagesToPartition(2, 3, 4); - var header = Encode(session, CommandCodes.SEND_MESSAGES_CODE, payload, out _); + var first = Encode(session, CommandCodes.SEND_MESSAGES_CODE, payload, out _); + Assert.Equal((byte)VsrOperation.SendMessages, first[VsrHeader.REQUEST_OPERATION_OFFSET]); + Assert.Equal(1UL, ReadUInt64(first, VsrHeader.REQUEST_ID_OFFSET)); - Assert.Equal((byte)VsrOperation.SendMessages, header[VsrHeader.REQUEST_OPERATION_OFFSET]); - Assert.Equal(1UL, ReadUInt64(header, VsrHeader.REQUEST_ID_OFFSET)); - Assert.Equal(1UL, session.RequestCounter); + var second = Encode(session, CommandCodes.SEND_MESSAGES_CODE, payload, out _); + Assert.Equal(2UL, ReadUInt64(second, VsrHeader.REQUEST_ID_OFFSET)); + Assert.Equal(3UL, session.RequestCounter); } [Fact] diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrOperationTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrOperationTests.cs index 6888db37bd..ad191b3081 100644 --- a/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrOperationTests.cs +++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/VsrOperationTests.cs @@ -94,12 +94,9 @@ public void Classification_MatchesTheServerSideRanges() Assert.True(VsrOperation.CreateTopicWithAssignments.IsInternal()); Assert.True(VsrOperation.CreateTopicWithAssignments.IsMetadata()); Assert.True(VsrOperation.CreateStream.IsMetadata()); - Assert.False(VsrOperation.CreateStream.IsPartition()); - Assert.True(VsrOperation.SendMessages.IsPartition()); Assert.False(VsrOperation.SendMessages.IsMetadata()); - // Resolved server-side to an internal truncate, so it is neither plane despite carrying a namespace. - Assert.False(VsrOperation.DeleteSegments.IsPartition()); + // Resolved server-side to an internal truncate, so it is not metadata despite carrying a namespace. Assert.False(VsrOperation.DeleteSegments.IsMetadata()); } diff --git a/foreign/go/internal/vsr/envelope.go b/foreign/go/internal/vsr/envelope.go index 912881a2c1..75c2d1c94d 100644 --- a/foreign/go/internal/vsr/envelope.go +++ b/foreign/go/internal/vsr/envelope.go @@ -61,13 +61,10 @@ func StampRequestHeader(session *Session, code uint32, frame []byte) error { sessionID = session.SessionID() default: sessionID = session.SessionID() - if IsPartition(operation) { - // Partition operations replicate in per-partition groups that - // keep no client table, so there is nothing to deduplicate - // against: the watermark is read without being consumed and a - // partition-plane replay is at-least-once. - request = session.CurrentRequestID() - } else if request, err = session.NextRequestID(); err != nil { + // Partition operations consume an id too, even though no + // partition-plane dedup exists yet: dedup needs each send to carry a + // distinct number, and the metadata watermark tolerates the gaps. + if request, err = session.NextRequestID(); err != nil { return err } } diff --git a/foreign/go/internal/vsr/envelope_test.go b/foreign/go/internal/vsr/envelope_test.go index c497b07fc5..154f95a334 100644 --- a/foreign/go/internal/vsr/envelope_test.go +++ b/foreign/go/internal/vsr/envelope_test.go @@ -115,7 +115,7 @@ func TestEncodeRequest_SendsNonReplicatedCommandsBeforeRegister(t *testing.T) { assert.Equal(t, uint64(1), binary.LittleEndian.Uint64(header[requestOffsetRequest:])) } -func TestEncodeRequest_DoesNotAdvanceTheWatermarkOffTheMetadataPlane(t *testing.T) { +func TestEncodeRequest_DoesNotAdvanceTheWatermarkForNonReplicated(t *testing.T) { session := boundSession(t) for range 3 { @@ -123,12 +123,21 @@ func TestEncodeRequest_DoesNotAdvanceTheWatermarkOffTheMetadataPlane(t *testing. require.NoError(t, err) } assert.Equal(t, uint64(1), session.CurrentRequestID()) +} - for range 3 { - _, err := EncodeRequest(session, uint32(command.SendMessagesCode), []byte{1}) +func TestEncodeRequest_PartitionCommandConsumesTheWatermark(t *testing.T) { + // Dedup identity requires each send to carry a distinct id, so partition + // commands advance the counter exactly like metadata commands and the two + // planes interleave on one sequence. + session := boundSession(t) + + for expected := uint64(1); expected <= 3; expected++ { + frame, err := EncodeRequest(session, uint32(command.SendMessagesCode), []byte{1}) require.NoError(t, err) + header := frameHeader(t, frame) + assert.Equal(t, expected, binary.LittleEndian.Uint64(header[requestOffsetRequest:])) } - assert.Equal(t, uint64(1), session.CurrentRequestID()) + assert.Equal(t, uint64(4), session.CurrentRequestID()) } func TestEncodeRequest_AdvancesTheWatermarkPerMetadataCommand(t *testing.T) { @@ -157,6 +166,7 @@ func TestEncodeRequest_RejectsAPartitionCommandOnAnUnboundSession(t *testing.T) _, err := EncodeRequest(session, uint32(command.SendMessagesCode), []byte{1}) assert.ErrorIs(t, err, ierror.ErrUnauthenticated) + assert.Equal(t, uint64(1), session.CurrentRequestID(), "the rejection burns no id") } func TestEncodeRequest_EncodesASendMessagesFrame(t *testing.T) { diff --git a/foreign/go/internal/vsr/header.go b/foreign/go/internal/vsr/header.go index 789237efd1..93580ff839 100644 --- a/foreign/go/internal/vsr/header.go +++ b/foreign/go/internal/vsr/header.go @@ -116,7 +116,10 @@ func (c ClientID) IsZero() bool { } // RequestFields are the header fields a client fills in. The rest of the 256 -// bytes stay zero, including both checksums, which the server does not read. +// bytes stay zero. The frame and body checksums are not read on the client +// request path. request_checksum is read, but zero means unstamped and opts out +// of the server's payload comparison, so leaving it zero is legal; the Rust SDK +// stamps it for deduped ops, this SDK does not yet. type RequestFields struct { // Size is the header plus body total. Size uint32 diff --git a/foreign/go/internal/vsr/operation.go b/foreign/go/internal/vsr/operation.go index cc95c57720..a1f0140cce 100644 --- a/foreign/go/internal/vsr/operation.go +++ b/foreign/go/internal/vsr/operation.go @@ -67,9 +67,8 @@ const ( // Band boundaries. The internal band is never client-sent. const ( - internalBandStart = OperationCreateTopicWithAssignments - metadataBandStart = OperationCreateStream - partitionBandStart = OperationSendMessages + internalBandStart = OperationCreateTopicWithAssignments + metadataBandStart = OperationCreateStream ) // allOperations lists every declared discriminant in wire order. It backs both @@ -206,12 +205,6 @@ func IsMetadata(operation Operation) bool { return operation >= metadataBandStart && operation <= OperationLeaveConsumerGroup } -// IsPartition reports whether the operation is routed to the shard owning a -// partition. -func IsPartition(operation Operation) bool { - return operation >= partitionBandStart -} - // IsResultFramed reports whether the reply body leads with a committed result // section. Every metadata operation is framed; on the partition plane only the // consumer-offset operations are, because they reject with typed errors at diff --git a/foreign/go/internal/vsr/operation_test.go b/foreign/go/internal/vsr/operation_test.go index 0b7b4784f9..e064ac5250 100644 --- a/foreign/go/internal/vsr/operation_test.go +++ b/foreign/go/internal/vsr/operation_test.go @@ -158,14 +158,6 @@ func TestIsMetadata_ExcludesDeleteSegments(t *testing.T) { } } -func TestIsPartition_CoversTheDataPlaneBand(t *testing.T) { - assert.True(t, IsPartition(OperationSendMessages)) - assert.True(t, IsPartition(OperationStoreConsumerOffset)) - assert.True(t, IsPartition(OperationDeleteConsumerOffset)) - assert.False(t, IsPartition(OperationLeaveConsumerGroup)) - assert.False(t, IsPartition(OperationDeleteSegments)) -} - func TestIsResultFramed_ExcludesSendMessages(t *testing.T) { assert.False(t, IsResultFramed(OperationSendMessages), "a send confirmation is not preceded by a result section") diff --git a/foreign/go/internal/vsr/protocol_parity_test.go b/foreign/go/internal/vsr/protocol_parity_test.go index 6cddd7f167..c9a32700cb 100644 --- a/foreign/go/internal/vsr/protocol_parity_test.go +++ b/foreign/go/internal/vsr/protocol_parity_test.go @@ -428,10 +428,8 @@ func TestProtocolParity_OperationClassification(t *testing.T) { internalStart := rustValues["CreateTopicWithAssignments"] metadataStart := rustValues["CreateStream"] - partitionStart := rustValues["SendMessages"] require.NotZero(t, internalStart) require.NotZero(t, metadataStart) - require.NotZero(t, partitionStart) for name, value := range rustValues { operation := Operation(value) @@ -442,7 +440,6 @@ func TestProtocolParity_OperationClassification(t *testing.T) { assert.Equal(t, internal, IsInternal(operation), "IsInternal(%s)", name) assert.Equal(t, metadata, IsMetadata(operation), "IsMetadata(%s)", name) - assert.Equal(t, value >= partitionStart, IsPartition(operation), "IsPartition(%s)", name) assert.Equal(t, metadata || inResultFramedList, IsResultFramed(operation), "IsResultFramed(%s)", name) assert.True(t, IsKnownOperation(operation), "IsKnownOperation(%s)", name) diff --git a/foreign/go/internal/vsr/session.go b/foreign/go/internal/vsr/session.go index 78226f4a38..3226a998fb 100644 --- a/foreign/go/internal/vsr/session.go +++ b/foreign/go/internal/vsr/session.go @@ -128,14 +128,9 @@ func (s *Session) NextRequestID() (uint64, error) { } // CurrentRequestID returns the watermark without advancing it. Non-replicated -// and partition-plane requests use it because neither consults the client -// table: non-replicated requests route by transport identity, and the -// partition plane replicates in per-partition groups with no dedup table at -// all. The table accepts any id above the watermark with no contiguity -// requirement, so consuming one here would not gap anything; the invariant -// that matters is that every partition request on a session carries the id -// the next metadata operation will claim, and that a partition-plane replay -// is therefore at-least-once. +// requests use it because they never consult the client table: they route by +// transport identity, and the table accepts any id above the watermark with +// no contiguity requirement, so reading here gaps nothing. func (s *Session) CurrentRequestID() uint64 { return s.requestCounter } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java index ab79ea19ee..16498bd842 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/ConsensusSession.java @@ -52,12 +52,20 @@ public ConsensusSession() { * the whole identity re-arms with a fresh client id so the server sees a * brand-new registration. Returns the request id a Register carries, * which is always zero. + * + *

The request counter is deliberately not rewound. This SDK multiplexes + * a single pinned channel and correlates replies by (operation, request + * id), so a send still in flight when a re-login re-arms would share its + * key with the first send of the new session: the correlation map would + * refuse the second one and a late reply for the first could be handed to + * it. A re-arm registers a fresh client id, which the server admits at + * watermark zero and which accepts any id above it, so carrying the + * counter forward costs nothing on the wire. */ synchronized long beginRegister() { if (registerConsumed || session != null) { regenerateClientId(); session = null; - requestCounter = 1; } registerConsumed = true; return 0; @@ -71,17 +79,31 @@ synchronized void bind(long sessionEpoch) { this.session = sessionEpoch; } - /** Replicated metadata ops consume the monotonic VSR dedup counter. */ + /** + * Replicated ops (metadata and partition) consume the monotonic VSR dedup + * counter. The wire field is a u64 but Java has no unsigned long, so the + * counter is refused at {@link Long#MAX_VALUE} rather than wrapping + * negative and sending an id below the server's watermark. + * + *

Exhaustion is terminal for this instance. {@link #beginRegister()} + * deliberately carries the counter across a re-login to keep pending-reply + * correlation keys unique, so reconnecting cannot rewind it; only a new + * client instance starts a fresh sequence. + */ synchronized long nextRequestId() { if (session == null) { throw new IggyNotConnectedException("Not authenticated, call login first"); } + if (requestCounter == Long.MAX_VALUE) { + throw new IllegalStateException( + "VSR request counter exhausted, create a fresh client instance (reconnecting preserves the counter)"); + } return requestCounter++; } /** - * Partition and non-replicated ops use an independent sequence for reply - * correlation, so they do not create gaps in the metadata dedup sequence. + * Non-replicated ops use an independent sequence for reply correlation, + * so they do not create gaps in the dedup sequence. */ synchronized long nextCorrelationId() { return correlationCounter++; diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrHeaders.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrHeaders.java index 35b933ef55..c49244cb06 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrHeaders.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrHeaders.java @@ -26,7 +26,11 @@ * Byte offsets and readers for the 256-byte consensus headers, mirroring the * {@code #[repr(C)]} layouts in * {@code core/binary_protocol/src/consensus/header.rs}. All fields are - * little-endian; the checksum fields stay zero by protocol contract. + * little-endian. The checksum fields stay zero by choice, not by protocol + * requirement: the frame and body checksums are not read on the client request + * path, and {@code request_checksum} treats zero as unstamped, which opts out of + * the server's payload comparison. The Rust SDK stamps it for deduped + * operations; this SDK does not yet. */ public final class VsrHeaders { diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrOperation.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrOperation.java index 794dd6f286..5ada71abbb 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrOperation.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrOperation.java @@ -69,7 +69,6 @@ public final class VsrOperation { private static final int INTERNAL_START = 64; private static final int METADATA_START = 128; - private static final int PARTITION_START = 160; /** * Replicated command code to operation, from the server's @@ -143,10 +142,6 @@ static boolean isMetadata(int operation) { return operation >= METADATA_START && operation <= LEAVE_CONSUMER_GROUP; } - static boolean isPartition(int operation) { - return operation >= PARTITION_START; - } - /** * Whether a reply body for this operation starts with a committed result * section ({@code [count:u32][{index,result} x count]}). diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoder.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoder.java index 45859f06d7..4bfcbeed81 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoder.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoder.java @@ -76,12 +76,11 @@ public ByteBuf encode(ByteBufAllocator alloc, int commandCode, ByteBuf payload) // sessionless before login. requestId = session.nextCorrelationId(); sessionId = session.sessionOrZero(); - } else if (VsrOperation.isPartition(operation)) { - // Partition ops replicate in their own group without client - // table dedup, so use the independent correlation sequence. - sessionId = session.boundSession(); - requestId = session.nextCorrelationId(); } else { + // Partition ops consume the dedup counter too, even though no + // partition-plane dedup exists yet: dedup needs each send to + // carry a distinct number, and the metadata watermark + // tolerates the gaps. sessionId = session.boundSession(); requestId = session.nextRequestId(); } diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoderTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoderTest.java index 426aa0b501..92c89243ba 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoderTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoderTest.java @@ -120,9 +120,10 @@ void shouldAdvanceRequestIdsForReplicatedCommands() { } @Test - void shouldCorrelatePartitionOpsWithoutAdvancingTheDedupRequestId() { - // Partition ops replicate in their own group with no client-table dedup, so - // they take a correlation id and leave the dedup counter where it was. + void shouldConsumeTheDedupRequestIdForPartitionOps() { + // Dedup identity requires each send to carry a distinct number, so + // partition ops advance the dedup counter exactly like metadata ops + // and the two planes interleave on one sequence. session.beginRegister(); session.bind(42); @@ -132,13 +133,16 @@ void shouldCorrelatePartitionOpsWithoutAdvancingTheDedupRequestId() { ByteBuf second = encoder.encode(alloc, SEND_MESSAGES_CODE, secondPayload); firstPayload.release(); secondPayload.release(); + ByteBuf metadata = encoder.encode(alloc, CREATE_STREAM_CODE, Unpooled.EMPTY_BUFFER); try { assertThat(first.getLongLE(VsrHeaders.REQUEST_ID_OFFSET)).isEqualTo(1); assertThat(second.getLongLE(VsrHeaders.REQUEST_ID_OFFSET)).isEqualTo(2); - assertThat(session.currentRequestId()).isEqualTo(1); + assertThat(metadata.getLongLE(VsrHeaders.REQUEST_ID_OFFSET)).isEqualTo(3); + assertThat(session.currentRequestId()).isEqualTo(4); } finally { first.release(); second.release(); + metadata.release(); } } diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java index 4a9e3bc2d6..51c4d3790d 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrResponseHandlerTest.java @@ -28,8 +28,10 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; +import java.nio.charset.StandardCharsets; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; @@ -37,6 +39,9 @@ class VsrResponseHandlerTest { + private static final int LOGIN_USER_CODE = 38; + private static final int SEND_MESSAGES_CODE = 101; + private final ConsensusSession session = new ConsensusSession(); private final AtomicInteger evictions = new AtomicInteger(); private final AtomicInteger lastEvictionReason = new AtomicInteger(); @@ -251,6 +256,47 @@ void shouldCorrelateRepliesArrivingInReverseOrder() throws Exception { } } + @Test + void shouldCorrelateASendInFlightAcrossReLogin() throws Exception { + // A re-login re-arms the session while an earlier send is still + // pending. Replies correlate by (operation, request id), so the first + // send of the new session must not claim the in-flight one's key: + // registering it would be refused and a late reply for the older send + // would be handed to the newer one. + VsrRequestEncoder encoder = new VsrRequestEncoder(session); + session.beginRegister(); + session.bind(42); + + CompletableFuture inFlight = new CompletableFuture<>(); + long beforeLoginId = registerEncodedSend(encoder, inFlight); + + ByteBuf loginPayload = loginUserPayload(); + encoder.encode(channel.alloc(), LOGIN_USER_CODE, loginPayload).release(); + loginPayload.release(); + session.bind(43); + + CompletableFuture afterLogin = new CompletableFuture<>(); + long afterLoginId = registerEncodedSend(encoder, afterLogin); + + assertThat(afterLoginId).isNotEqualTo(beforeLoginId); + assertThat(channel.isActive()).isTrue(); + + channel.writeInbound( + replyFrame(VsrOperation.SEND_MESSAGES, afterLoginId, Unpooled.wrappedBuffer(new byte[] {2}))); + channel.writeInbound( + replyFrame(VsrOperation.SEND_MESSAGES, beforeLoginId, Unpooled.wrappedBuffer(new byte[] {1}))); + + ByteBuf inFlightResponse = inFlight.get(); + ByteBuf afterLoginResponse = afterLogin.get(); + try { + assertThat(inFlightResponse.readByte()).isEqualTo((byte) 1); + assertThat(afterLoginResponse.readByte()).isEqualTo((byte) 2); + } finally { + inFlightResponse.release(); + afterLoginResponse.release(); + } + } + @Test void shouldCorrelateRepliesForServerRewrittenOperations() throws Exception { int[][] rewrittenOperations = { @@ -296,6 +342,32 @@ private CompletableFuture enqueue(int operation, long requestId) { return future; } + /** + * Encodes a partition send off the live session and registers it the way + * the connection does, returning the request id the encoder minted. + */ + private long registerEncodedSend(VsrRequestEncoder encoder, CompletableFuture future) { + ByteBuf frame = encoder.encode(channel.alloc(), SEND_MESSAGES_CODE, Unpooled.EMPTY_BUFFER); + try { + handler.registerRequest( + channel, frame, future, System.nanoTime() + TimeUnit.MINUTES.toNanos(1), SEND_MESSAGES_CODE); + return VsrHeaders.readRequestId(frame); + } finally { + frame.release(); + } + } + + private static ByteBuf loginUserPayload() { + ByteBuf payload = Unpooled.buffer(); + payload.writeByte(4); + payload.writeBytes("iggy".getBytes(StandardCharsets.UTF_8)); + payload.writeByte(4); + payload.writeBytes("iggy".getBytes(StandardCharsets.UTF_8)); + payload.writeIntLE(0); + payload.writeIntLE(0); + return payload; + } + private static ByteBuf emptyFrame() { ByteBuf frame = Unpooled.buffer(VsrHeaders.HEADER_SIZE); frame.writeZero(VsrHeaders.HEADER_SIZE); diff --git a/foreign/node/scripts/check-vsr-protocol.mjs b/foreign/node/scripts/check-vsr-protocol.mjs index dcbe060fb3..41426e5bbc 100644 --- a/foreign/node/scripts/check-vsr-protocol.mjs +++ b/foreign/node/scripts/check-vsr-protocol.mjs @@ -252,7 +252,6 @@ const operationModule = await import( ); const internalStart = rustOperations.get('CreateTopicWithAssignments'); const metadataStart = rustOperations.get('CreateStream'); -const partitionStart = rustOperations.get('SendMessages'); const rustMetadataNames = new Set( [...(rustOperation.match( /fn is_metadata[\s\S]*?matches!\(\s*self,([\s\S]*?)\)\s*\n\s*\}/ @@ -276,11 +275,6 @@ for (const [name, value] of rustOperations) { metadata, `Node isMetadata(${name}) differs from Rust is_metadata` ); - assert.equal( - operationModule.isPartition(value), - value >= partitionStart, - `Node isPartition(${name}) differs from Rust is_partition` - ); assert.equal( operationModule.isResultFramed(value), metadata || rustResultFramedNames.has(name), diff --git a/foreign/node/src/wire/vsr/header.ts b/foreign/node/src/wire/vsr/header.ts index d54182c5ba..e7128a1159 100644 --- a/foreign/node/src/wire/vsr/header.ts +++ b/foreign/node/src/wire/vsr/header.ts @@ -113,8 +113,10 @@ const U64_MASK = 0xFFFFFFFFFFFFFFFFn; /** * Encodes a 256-byte request header. Only the six fields the server reads - * are written; the checksums stay zero, matching the Rust SDK's contract - * with the VSR server. + * are written. The checksums stay zero: the frame and body checksums are not + * read on the client request path, and `request_checksum` treats zero as + * unstamped, which opts out of the server's payload comparison. Stamping it is + * optional -- the Rust SDK does for deduped ops, this SDK does not yet. */ export const encodeRequestHeader = (fields: RequestHeaderFields): Buffer => { const header = Buffer.alloc(HEADER_SIZE); diff --git a/foreign/node/src/wire/vsr/index.ts b/foreign/node/src/wire/vsr/index.ts index 24091ca1fe..0fdee6f365 100644 --- a/foreign/node/src/wire/vsr/index.ts +++ b/foreign/node/src/wire/vsr/index.ts @@ -20,11 +20,7 @@ import type { CommandResponse } from '../../client/client.type.js'; import { COMMAND_CODE } from '../command.code.js'; import { responseError } from '../error.utils.js'; import { HEADER_SIZE, encodeRequestHeader } from './header.js'; -import { - Operation, - isPartition, - operationForCode, -} from './operation.js'; +import { Operation, operationForCode } from './operation.js'; import { deserializeLoginRegister, serializeLoginRegister, @@ -80,9 +76,10 @@ export class VsrSession { } else { if (this.state.session === null) throw responseError(command, UNAUTHENTICATED); - request = isPartition(operation) - ? this.state.currentRequestId() - : this.state.nextRequestId(); + // Partition ops consume an id too, even though no partition-plane dedup + // exists yet: dedup needs each send to carry a distinct number, and the + // metadata watermark tolerates the gaps. + request = this.state.nextRequestId(); session = this.state.session; } diff --git a/foreign/node/src/wire/vsr/operation.test.ts b/foreign/node/src/wire/vsr/operation.test.ts index daac920474..4ce539ffe9 100644 --- a/foreign/node/src/wire/vsr/operation.test.ts +++ b/foreign/node/src/wire/vsr/operation.test.ts @@ -22,7 +22,6 @@ import { isInternal, isKnownOperation, isMetadata, - isPartition, isResultFramed, Operation, operationForCode @@ -87,8 +86,6 @@ describe('VSR operation classification', () => { assert.equal(isMetadata(Operation.LeaveConsumerGroup), true); assert.equal(isMetadata(Operation.DeleteSegments), false); assert.equal(isMetadata(150), false); - assert.equal(isPartition(Operation.SendMessages), true); - assert.equal(isPartition(159), false); assert.equal(isResultFramed(Operation.StoreConsumerOffset), true); assert.equal(isResultFramed(Operation.DeleteConsumerOffset), true); assert.equal(isResultFramed(Operation.SendMessages), false); diff --git a/foreign/node/src/wire/vsr/operation.ts b/foreign/node/src/wire/vsr/operation.ts index 2781668885..030d9f0fef 100644 --- a/foreign/node/src/wire/vsr/operation.ts +++ b/foreign/node/src/wire/vsr/operation.ts @@ -63,7 +63,6 @@ export const Operation = { const INTERNAL_START = 64; const METADATA_START = 128; -const PARTITION_START = 160; /** * Replicated command code to `Operation` mapping, the client half of the @@ -126,10 +125,6 @@ export const isMetadata = (operation: number): boolean => { operation <= Operation.LeaveConsumerGroup; }; -/** Partition band is a bare range, mirroring `Operation::is_partition`. */ -export const isPartition = (operation: number): boolean => - operation >= PARTITION_START; - /** Whether a reply body leads with a committed result section. */ export const isResultFramed = (operation: number): boolean => isMetadata(operation) || diff --git a/foreign/node/src/wire/vsr/session.ts b/foreign/node/src/wire/vsr/session.ts index f61cbc3f0b..5f381fbbbf 100644 --- a/foreign/node/src/wire/vsr/session.ts +++ b/foreign/node/src/wire/vsr/session.ts @@ -24,9 +24,9 @@ const MAX_U64 = 0xFFFF_FFFF_FFFF_FFFFn; * * Each client instance generates an ephemeral random `clientId` (u128). * After a Register commits, the server assigns a `session` number (commit op - * number). Replicated metadata requests advance a monotonic request watermark. - * Non-replicated and partition-plane requests reuse the current value because - * the server only applies request sequencing to replicated metadata. + * number). Every replicated request (metadata and partition) advances a + * monotonic request watermark; non-replicated requests reuse the current + * value because they bypass server-side request sequencing. */ export class ConsensusSession { private _clientId: bigint; diff --git a/foreign/node/src/wire/vsr/vsr.test.ts b/foreign/node/src/wire/vsr/vsr.test.ts index 60e9c19088..cbc9feb7e4 100644 --- a/foreign/node/src/wire/vsr/vsr.test.ts +++ b/foreign/node/src/wire/vsr/vsr.test.ts @@ -79,7 +79,7 @@ describe('VSR custom request framing', () => { ); }); - it('does not advance for non-replicated or partition operations', () => { + it('does not advance for non-replicated operations', () => { const session = new VsrSession(7n); session.bind(42n); const custom = session.encode( @@ -88,26 +88,43 @@ describe('VSR custom request framing', () => { ); assert.equal(custom.readBigUInt64LE(REQUEST_OFFSET.request), 1n); - const partition = session.encode( - COMMAND_CODE.SendMessages, - serializeSendMessages( - 1, - 2, - [{ payload: 'x' }], - Partitioning.PartitionId(3) - ) + const metadata = session.encode( + COMMAND_CODE.CreateStream, + Buffer.alloc(0) ); + assert.equal(metadata.readBigUInt64LE(REQUEST_OFFSET.request), 1n); + assert.equal(metadata.length, HEADER_SIZE); + }); + + it('partition operations consume a distinct id per send', () => { + // Dedup identity requires each send to carry a distinct number, so + // partition ops advance the counter exactly like metadata ops and the + // two planes interleave on one sequence. + const session = new VsrSession(7n); + session.bind(42n); + const sendMessages = () => + session.encode( + COMMAND_CODE.SendMessages, + serializeSendMessages( + 1, + 2, + [{ payload: 'x' }], + Partitioning.PartitionId(3) + ) + ); + + const first = sendMessages(); assert.equal( - partition.readUInt8(REQUEST_OFFSET.operation), + first.readUInt8(REQUEST_OFFSET.operation), Operation.SendMessages ); - assert.equal(partition.readBigUInt64LE(REQUEST_OFFSET.request), 1n); + assert.equal(first.readBigUInt64LE(REQUEST_OFFSET.request), 1n); + assert.equal(sendMessages().readBigUInt64LE(REQUEST_OFFSET.request), 2n); const metadata = session.encode( COMMAND_CODE.CreateStream, Buffer.alloc(0) ); - assert.equal(metadata.readBigUInt64LE(REQUEST_OFFSET.request), 1n); - assert.equal(metadata.length, HEADER_SIZE); + assert.equal(metadata.readBigUInt64LE(REQUEST_OFFSET.request), 3n); }); });