From 427c1968abd69d55ce7edfec1f87f4ab5fff9c69 Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Mon, 24 Aug 2026 16:53:21 +0200 Subject: [PATCH 1/3] feat(sdk): consume the request counter for partition operations --- core/sdk/src/quic/quic_client.rs | 4 +- core/sdk/src/vsr.rs | 70 +++++++++++++------ .../csharp/Iggy_SDK/Vsr/ConsensusSession.cs | 10 +-- .../VsrTests/ConsensusSessionTests.cs | 19 ++++- .../Iggy_SDK_Tests/VsrTests/VsrHeaderTests.cs | 12 ++-- foreign/go/internal/vsr/envelope.go | 11 ++- foreign/go/internal/vsr/envelope_test.go | 17 +++-- foreign/go/internal/vsr/session.go | 11 +-- .../async/tcp/vsr/ConsensusSession.java | 6 +- .../async/tcp/vsr/VsrRequestEncoder.java | 8 ++- .../async/tcp/vsr/VsrRequestEncoderTest.java | 12 ++-- foreign/node/src/wire/vsr/index.ts | 7 +- foreign/node/src/wire/vsr/session.ts | 6 +- foreign/node/src/wire/vsr/vsr.test.ts | 43 ++++++++---- 14 files changed, 150 insertions(+), 86 deletions(-) diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs index 921ef5d3dc..3ab1a264d4 100644 --- a/core/sdk/src/quic/quic_client.rs +++ b/core/sdk/src/quic/quic_client.rs @@ -629,8 +629,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 4a6bda52ec..fa90e03e17 100644 --- a/core/sdk/src/vsr.rs +++ b/core/sdk/src/vsr.rs @@ -96,26 +96,22 @@ pub(crate) fn encode_request_header( 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. + // Consumes an id even though no partition-plane dedup exists + // 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.current_request_id(), session_id) + (operation, session.next_request_id(), session_id) } else { 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. - let request_checksum = if operation.is_partition() || operation == Operation::NonReplicated { + // Every replicated op stamps: the stamp is what lets a dedup layer tell a + // genuine retry from a request id reused for different bytes. + // NonReplicated ops bypass dedup, so the hash pass buys nothing there. + let request_checksum = if operation == Operation::NonReplicated { 0 } else { u128::from(calculate_checksum(payload)) @@ -347,7 +343,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; @@ -487,19 +485,25 @@ 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. + fn request_checksum_is_stamped_for_replicated_operations() { + // The stamp exists to stop a reused `request` number matching a dedup + // entry recorded for different bytes, so every replicated op carries + // it; NonReplicated ops bypass dedup and skip the hashing pass. 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(); assert_eq!( - decode_request_header(&deduped).request_checksum, + decode_request_header(&metadata).request_checksum, + u128::from(calculate_checksum(&payload)), + ); + + let partition = + encode_contiguous_request(&mut session, SEND_MESSAGES_CODE, &payload).unwrap(); + assert_eq!( + decode_request_header(&partition).request_checksum, u128::from(calculate_checksum(&payload)), ); @@ -507,6 +511,30 @@ mod tests { 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/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs b/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs index 8cdbdcc16c..c2623d8fb3 100644 --- a/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs +++ b/foreign/csharp/Iggy_SDK/Vsr/ConsensusSession.cs @@ -185,14 +185,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_Tests/VsrTests/ConsensusSessionTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/ConsensusSessionTests.cs index 51e4d7f0a1..c081181304 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/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..25862dc6ef 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) { 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..5bc5ae4f7f 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 @@ -71,7 +71,7 @@ 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. */ synchronized long nextRequestId() { if (session == null) { throw new IggyNotConnectedException("Not authenticated, call login first"); @@ -80,8 +80,8 @@ synchronized long nextRequestId() { } /** - * 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/VsrRequestEncoder.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrRequestEncoder.java index 45859f06d7..cea0aef1db 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 @@ -77,10 +77,12 @@ public ByteBuf encode(ByteBufAllocator alloc, int commandCode, ByteBuf payload) 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. + // 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.nextCorrelationId(); + requestId = session.nextRequestId(); } else { 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/node/src/wire/vsr/index.ts b/foreign/node/src/wire/vsr/index.ts index 24091ca1fe..6dcffe2dc7 100644 --- a/foreign/node/src/wire/vsr/index.ts +++ b/foreign/node/src/wire/vsr/index.ts @@ -80,9 +80,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/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); }); }); From 4691829f9d1799d3060fb94c4d69472f8340b5e6 Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Mon, 24 Aug 2026 20:32:18 +0200 Subject: [PATCH 2/3] feat(server): dedup partition writes with per-group client table slices --- core/configs/src/server_config/defaults.rs | 1 + core/configs/src/server_config/partition.rs | 51 ++ core/consensus/src/client_table.rs | 416 ++++++++++- core/consensus/src/impls.rs | 56 ++ core/consensus/src/lib.rs | 19 +- core/integration/tests/cluster/mod.rs | 1 + .../tests/cluster/partition_dedup.rs | 660 ++++++++++++++++++ core/partitions/src/iggy_partition.rs | 284 ++++++-- core/partitions/src/iggy_partitions.rs | 26 +- core/partitions/src/state_transfer.rs | 159 ++++- core/server/config.toml | 11 + core/server/src/bootstrap.rs | 5 + core/server/src/dispatch.rs | 95 ++- core/server/src/http/submit.rs | 6 +- core/server/src/partition_helpers.rs | 1 + core/shard/src/lib.rs | 222 +++++- core/shard/src/router.rs | 12 + core/simulator/src/client.rs | 13 +- core/simulator/src/lib.rs | 61 +- 19 files changed, 1948 insertions(+), 151 deletions(-) create mode 100644 core/integration/tests/cluster/partition_dedup.rs diff --git a/core/configs/src/server_config/defaults.rs b/core/configs/src/server_config/defaults.rs index f752d32659..2f375edd23 100644 --- a/core/configs/src/server_config/defaults.rs +++ b/core/configs/src/server_config/defaults.rs @@ -174,6 +174,7 @@ impl Default for PartitionConfig { let partition = &SERVER_CONFIG.partition; PartitionConfig { prepare_queue_depth: partition.prepare_queue_depth as usize, + dedup_clients_max: partition.dedup_clients_max as usize, evicted_ring_capacity: partition.evicted_ring_capacity as usize, evicted_ring_bytes_max: partition.evicted_ring_bytes_max.parse().unwrap(), transfer_served_cache_bytes_max: partition diff --git a/core/configs/src/server_config/partition.rs b/core/configs/src/server_config/partition.rs index d6e3273611..03ee5f6acb 100644 --- a/core/configs/src/server_config/partition.rs +++ b/core/configs/src/server_config/partition.rs @@ -114,6 +114,14 @@ pub const DEFAULT_EVICTED_RING_BYTES_MAX: u64 = 16 * 1024 * 1024; pub const MAX_EVICTED_RING_BYTES: u64 = 256 * 1024 * 1024; /// Capacity tunables for the per-partition consensus plane. +/// Shipped default for [`PartitionConfig::dedup_clients_max`]; pinned against +/// the runtime constant by a bootstrap assert. +pub const DEFAULT_PARTITION_DEDUP_CLIENTS_MAX: usize = 4096; + +/// Ceiling for [`PartitionConfig::dedup_clients_max`]. A per-group budget, so +/// the ceiling bounds worst-case memory at `partitions * this * ~40 bytes`. +pub const MAX_PARTITION_DEDUP_CLIENTS: usize = 1 << 16; + #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] pub struct PartitionConfig { /// Depth of a partition's prepare queue: how many uncommitted produce / @@ -123,6 +131,18 @@ pub struct PartitionConfig { /// pinned request-buffer memory by the partition count. pub prepare_queue_depth: usize, + /// Distinct clients each partition group tracks request watermarks for, + /// deduplicating retried produces and consumer-offset writes. At capacity + /// the entry whose newest commit is oldest is evicted, which costs dedup + /// coverage for that client (its next replay re-executes, exactly as it + /// would have before dedup existed) and never correctness. Must be > 0 and + /// <= [`MAX_PARTITION_DEDUP_CLIENTS`]. + /// + /// Unlike `[metadata] clients_table_max`, this budget is PER GROUP, so the + /// worst case scales with partition count: size it to the producers a + /// single partition actually sees, not the node's client total. + pub dedup_clients_max: usize, + /// Entries the evicted ring retains per multi-replica partition for /// journal repair after a peer rejoins. Larger widens the window a /// restarting peer can be served from the ring before falling back to @@ -177,6 +197,14 @@ impl Validatable for PartitionConfig { ); return Err(ConfigurationError::InvalidConfigurationValue); } + if self.dedup_clients_max == 0 || self.dedup_clients_max > MAX_PARTITION_DEDUP_CLIENTS { + eprintln!( + "{COMPONENT} partition.dedup_clients_max ({}) must be > 0 and <= \ + {MAX_PARTITION_DEDUP_CLIENTS}", + self.dedup_clients_max + ); + return Err(ConfigurationError::InvalidConfigurationValue); + } if self.evicted_ring_capacity == 0 { eprintln!("{COMPONENT} partition.evicted_ring_capacity must be > 0"); return Err(ConfigurationError::InvalidConfigurationValue); @@ -253,6 +281,29 @@ mod tests { ); } + #[test] + fn shipped_dedup_default_matches_the_runtime_constant() { + assert_eq!( + PartitionConfig::default().dedup_clients_max, + DEFAULT_PARTITION_DEDUP_CLIENTS_MAX, + "config.toml dedup_clients_max drifted from the runtime default" + ); + } + + #[test] + fn rejects_out_of_range_dedup_clients_max() { + for value in [0, MAX_PARTITION_DEDUP_CLIENTS + 1] { + let config = PartitionConfig { + dedup_clients_max: value, + ..PartitionConfig::default() + }; + assert!( + config.validate().is_err(), + "dedup_clients_max {value} must be rejected" + ); + } + } + #[test] fn rejects_zero_prepare_queue_depth() { let config = PartitionConfig { diff --git a/core/consensus/src/client_table.rs b/core/consensus/src/client_table.rs index 5d70919c3b..08a5ce3042 100644 --- a/core/consensus/src/client_table.rs +++ b/core/consensus/src/client_table.rs @@ -352,6 +352,45 @@ pub enum CommitReply { SkippedRegression { stored: u64, received: u64 }, } +/// Which of the table's mechanisms an instance runs. +/// +/// The metadata plane needs all of them. A partition group's slice needs only +/// the watermark: it has no register to mint an epoch from, no result section +/// worth caching, and one table per group rather than per node, so the +/// preallocated slot array would reserve ~384 KiB per partition before a single +/// client connects. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ClientTableMode { + /// Keep committed replies so a duplicate replays the original bytes. Off: + /// duplicates answer [`RequestStatus::AlreadyApplied`] and the caller + /// synthesizes the reply. + pub cache_replies: bool, + /// Enforce the register-minted epoch fence. Off: entries carry no epoch, + /// `check_request` ignores the presented one, and a committed request may + /// create its own entry (there is no register to do it). + pub fence_epoch: bool, + /// Allocate every slot up front. Off: slots grow to the cap on demand. + /// Slot assignment is identical either way -- both hand out the lowest free + /// index -- so eviction order and the wire encoding are unchanged. + pub preallocate_slots: bool, +} + +impl ClientTableMode { + /// Metadata plane: replies cached, epoch fenced, slots preallocated. + pub const METADATA: Self = Self { + cache_replies: true, + fence_epoch: true, + preallocate_slots: true, + }; + + /// One partition consensus group's slice: watermark only. + pub const PARTITION_SLICE: Self = Self { + cache_replies: false, + fence_epoch: false, + preallocate_slots: false, + }; +} + /// VSR client table: per-session fence epoch + request-watermark dedup. /// /// Fixed-size slot array (source of truth) + `HashMap` index (O(1) lookup). @@ -374,11 +413,11 @@ pub enum CommitReply { /// /// ## Plane /// -/// Metadata-plane today. The design spans planes (one logical table, -/// group-resident slices); partition-plane integration arrives once -/// partition prepares carry real `(session_id, request)` instead of the -/// transport id (data-plane request numbering, IGGY-137). Until then the -/// partition plane stays at-least-once with no dedup. +/// This table is the metadata plane's. The partition plane runs the same +/// watermark rule in its own per-group slices +/// (`partitions::dedup::PartitionDedupSlice`), which keep no reply ring and no +/// epoch: partition prepares carry the VSR client id and request number but no +/// session, so fencing a stale session waits on identity surviving reconnects. /// /// ## Tracking /// @@ -404,9 +443,16 @@ pub enum CommitReply { #[derive(Debug)] pub struct ClientTable { /// `None` = free slot. Deterministic iteration for eviction + serialization. + /// + /// Under [`ClientTableMode::preallocate_slots`] this is sized to + /// `clients_max` at construction; otherwise it grows to that cap on demand. slots: Vec>, /// `client_id` -> slot index. Rebuilt on decode. index: HashMap, + /// Slot ceiling. Tracked explicitly because `slots.len()` is the allocated + /// length, which only equals the cap when slots are preallocated. + clients_max: usize, + mode: ClientTableMode, } /// Whether two integrity stamps for the same request number disagree. @@ -418,17 +464,40 @@ const fn checksums_conflict(stored: u128, received: u128) -> bool { } impl ClientTable { + /// Client id that opts out of dedup entirely. Zero is reserved cluster-wide + /// and every mutating entry point refuses it. + pub const EXEMPT_CLIENT: u128 = 0; + /// `max_clients` caps slots; index pre-sized to avoid rehash storms. #[must_use] pub fn new(max_clients: usize) -> Self { - let mut slots = Vec::with_capacity(max_clients); - slots.resize_with(max_clients, || None); + Self::with_mode(max_clients, ClientTableMode::METADATA) + } + + /// `max_clients` caps slots; `mode` selects which mechanisms run. + #[must_use] + pub fn with_mode(max_clients: usize, mode: ClientTableMode) -> Self { + let (slots, index) = if mode.preallocate_slots { + let mut slots = Vec::with_capacity(max_clients); + slots.resize_with(max_clients, || None); + (slots, HashMap::with_capacity(max_clients)) + } else { + (Vec::new(), HashMap::new()) + }; Self { slots, - index: HashMap::with_capacity(max_clients), + index, + clients_max: max_clients, + mode, } } + /// The mechanisms this instance runs. + #[must_use] + pub const fn mode(&self) -> ClientTableMode { + self.mode + } + /// Resize the table to `max_clients` slots. Boot-only: reallocating a /// populated table would silently drop live sessions, so this must run /// before any client registers (the server bootstrap applies the configured @@ -441,7 +510,7 @@ impl ClientTable { self.index.is_empty(), "set_capacity must run before any client registers" ); - *self = Self::new(max_clients); + *self = Self::with_mode(max_clients, self.mode); } /// Snapshot the table for the metadata checkpoint: every occupied slot with its @@ -558,7 +627,13 @@ impl ClientTable { latest_commit, }); } - Ok(Self { slots, index }) + let clients_max = slots.len(); + Ok(Self { + slots, + index, + clients_max, + mode: ClientTableMode::METADATA, + }) } /// Check a request against the table. Epoch fence first, then the @@ -581,7 +656,10 @@ impl ClientTable { ) -> RequestStatus { assert!(client_id != 0, "client_id 0 is reserved for internal use"); // Header validation guarantees both > 0 at wire layer. - debug_assert!(epoch > 0, "check_request: epoch must be > 0"); + debug_assert!( + epoch > 0 || !self.mode.fence_epoch, + "check_request: epoch must be > 0 when fencing" + ); debug_assert!(request > 0, "check_request: request must be > 0"); // Epoch check before request: a fenced zombie must be rejected even @@ -591,17 +669,21 @@ impl ClientTable { }; let entry = self.slots[slot_idx].as_ref().expect("index/slot mismatch"); - if epoch < entry.epoch { - return RequestStatus::Fenced { - current: entry.epoch, - received: epoch, - }; - } - if epoch > entry.epoch { - return RequestStatus::EpochAhead { - current: entry.epoch, - received: epoch, - }; + // A plane with no register mints no epoch, so there is nothing to + // fence against and the presented value is ignored. + if self.mode.fence_epoch { + if epoch < entry.epoch { + return RequestStatus::Fenced { + current: entry.epoch, + received: epoch, + }; + } + if epoch > entry.epoch { + return RequestStatus::EpochAhead { + current: entry.epoch, + received: epoch, + }; + } } if request > entry.watermark { @@ -689,7 +771,7 @@ impl ClientTable { .retain(|stored| stored.header().request != REGISTER_REQUEST_ID); entry.push_latest(cached); } else { - let freed = if self.index.len() >= self.slots.len() { + let freed = if self.index.len() >= self.clients_max { self.evict_oldest() } else { None @@ -811,6 +893,118 @@ impl ClientTable { CommitReply::Cached } + /// Watermark-only dedup check for a plane that mints no epoch. + /// + /// `true` means the request is at or below the client's watermark, i.e. it + /// already committed and must be answered rather than executed again. + /// + /// [`Self::EXEMPT_CLIENT`] always reads as new: zero is reserved + /// cluster-wide, so it doubles as the marker for a request whose id carries + /// no dedup meaning (an unbound caller, or a transport that numbers + /// requests without the one-in-flight-per-group discipline the watermark + /// assumes). + /// + /// # Panics + /// If called on a table that fences epochs -- that plane must go through + /// [`Self::check_request`], which enforces the fence. + #[must_use] + pub fn is_duplicate(&self, client_id: u128, request: u64) -> bool { + assert!( + !self.mode.fence_epoch, + "is_duplicate: an epoch-fencing table must use check_request" + ); + if client_id == Self::EXEMPT_CLIENT { + return false; + } + !matches!( + self.check_request(client_id, 0, request, 0), + RequestStatus::New | RequestStatus::NoSession + ) + } + + /// Record a committed request without a reply to cache. + /// + /// The entry point for a plane that runs + /// [`ClientTableMode::PARTITION_SLICE`]: there is no register to create the + /// entry, so the first committed request creates it, and there is no result + /// section worth retaining, so a later duplicate answers + /// [`RequestStatus::AlreadyApplied`] and the caller synthesizes the reply. + /// + /// Idempotent and order-insensitive: the watermark only rises, so replaying + /// an already-folded op is a no-op and a state-transfer install followed by + /// a re-walk of the same commits converges. + /// + /// # Panics + /// If called on a table whose mode caches replies -- that plane must go + /// through [`Self::commit_reply`] so the ring stays populated. + pub fn commit_request(&mut self, client_id: u128, request: u64, commit_op: u64) { + assert!( + !self.mode.cache_replies, + "commit_request: a reply-caching table must use commit_reply" + ); + if client_id == Self::EXEMPT_CLIENT { + return; + } + + if let Some(&slot_idx) = self.index.get(&client_id) { + let entry = self.slots[slot_idx].as_mut().expect("index/slot mismatch"); + if request > entry.watermark { + entry.watermark = request; + entry.latest_commit = commit_op; + } + return; + } + + if self.index.len() >= self.clients_max { + self.evict_oldest(); + } + let Some(slot_idx) = self.first_free_slot() else { + // Only reachable at a zero cap, which config validation rejects. + return; + }; + self.index.insert(client_id, slot_idx); + self.slots[slot_idx] = Some(ClientEntry { + epoch: 0, + user_id: 0, + watermark: request, + watermark_checksum: 0, + ring: VecDeque::new(), + client_id, + latest_commit: commit_op, + }); + } + + /// Replace every entry, as a state-transfer install does. + /// + /// # Panics + /// If called on a table whose mode caches replies (those install through + /// the snapshot / wire codecs, which carry the rings). + pub fn install_watermarks(&mut self, entries: impl IntoIterator) { + assert!( + !self.mode.cache_replies, + "install_watermarks: a reply-caching table installs via decode" + ); + self.slots.clear(); + self.index.clear(); + for (client_id, watermark, latest_commit) in entries { + self.commit_request(client_id, watermark, latest_commit); + } + } + + /// Every entry as `(client, watermark, latest_commit)`, ascending by + /// client: the deterministic form a wire encoding needs. + #[must_use] + pub fn watermarks_sorted(&self) -> Vec<(u128, u64, u64)> { + let mut entries: Vec<(u128, u64, u64)> = self + .slots + .iter() + .flatten() + .map(|entry| (entry.client_id, entry.watermark, entry.latest_commit)) + .collect(); + entries.sort_unstable_by_key(|(client_id, _, _)| *client_id); + entries + } + /// Remove a client session and cached replies. /// /// **LOCAL ONLY -- does NOT replicate.** Two correct call sites: @@ -883,8 +1077,18 @@ impl ClientTable { Some(slot_idx) } - fn first_free_slot(&self) -> Option { - self.slots.iter().position(Option::is_none) + /// Lowest free slot, growing the array when slots are allocated lazily. + /// Assignment is identical to the preallocated case: both hand out the + /// lowest free index, so eviction order and the wire encoding do not + /// depend on the mode. + fn first_free_slot(&mut self) -> Option { + if let Some(index) = self.slots.iter().position(Option::is_none) { + return Some(index); + } + (self.slots.len() < self.clients_max).then(|| { + self.slots.push(None); + self.slots.len() - 1 + }) } /// Latest cached reply for a client. @@ -1200,7 +1404,7 @@ impl ClientTable { /// can exceed `[metadata] clients_table_max`. #[must_use] pub const fn capacity(&self) -> usize { - self.slots.len() + self.clients_max } } @@ -1878,6 +2082,166 @@ mod tests { // Capacity resize (boot-only) + // --- ClientTableMode::PARTITION_SLICE: watermark-only dedup --- + // + // One consensus group's slice. No register mints entries here, no reply is + // cached, and slots grow on demand, so these pin the behaviour the + // partition plane actually relies on. + + fn slice(clients_max: usize) -> ClientTable { + ClientTable::with_mode(clients_max, ClientTableMode::PARTITION_SLICE) + } + + #[test] + fn given_partition_slice_when_empty_should_admit_and_not_preallocate() { + // The reason this plane cannot use the metadata mode: one table per + // group, so preallocating the cap would reserve hundreds of KiB per + // partition before a single client connects. + let table = slice(4096); + assert_eq!(table.count(), 0); + assert_eq!(table.slots.len(), 0, "slots must grow on demand"); + assert!(!table.is_duplicate(7, 1)); + } + + #[test] + fn given_partition_slice_when_request_replayed_should_report_duplicate() { + let mut table = slice(4); + table.commit_request(7, 5, 100); + + assert!(table.is_duplicate(7, 5)); + assert!(table.is_duplicate(7, 4)); + assert!(!table.is_duplicate(7, 6)); + } + + #[test] + fn given_partition_slice_when_request_id_gaps_should_accept_the_jump() { + // One client counter feeds several groups, so a slice legitimately sees + // only a subset of the ids that client mints. + let mut table = slice(4); + table.commit_request(7, 5, 100); + table.commit_request(7, 9, 101); + + assert!(table.is_duplicate(7, 7)); + assert!(!table.is_duplicate(7, 10)); + } + + #[test] + fn given_partition_slice_when_commit_replayed_should_be_idempotent() { + let mut table = slice(4); + table.commit_request(7, 5, 100); + table.commit_request(7, 5, 100); + table.commit_request(7, 3, 99); + + assert_eq!(table.watermarks_sorted(), vec![(7, 5, 100)]); + } + + #[test] + fn given_partition_slice_when_full_should_evict_the_oldest_commit() { + let mut table = slice(2); + table.commit_request(1, 1, 10); + table.commit_request(2, 1, 20); + table.commit_request(3, 1, 30); + + assert_eq!(table.count(), 2); + let clients: Vec = table + .watermarks_sorted() + .into_iter() + .map(|(client, _, _)| client) + .collect(); + assert_eq!(clients, vec![2, 3], "oldest commit is the victim"); + } + + #[test] + fn given_partition_slice_when_entry_evicted_should_admit_its_replay_again() { + // Losing an entry costs dedup coverage, never correctness: the replay + // re-executes exactly as it would have before the slice existed. + let mut table = slice(1); + table.commit_request(1, 5, 10); + table.commit_request(2, 1, 20); + + assert!(!table.is_duplicate(1, 5)); + } + + #[test] + fn given_partition_slice_when_entry_touched_should_spare_it_from_eviction() { + let mut table = slice(2); + table.commit_request(1, 1, 10); + table.commit_request(2, 1, 20); + // Client 1 commits again, so client 2 now holds the oldest commit. + table.commit_request(1, 2, 30); + table.commit_request(3, 1, 40); + + let clients: Vec = table + .watermarks_sorted() + .into_iter() + .map(|(client, _, _)| client) + .collect(); + assert_eq!(clients, vec![1, 3]); + } + + #[test] + fn given_partition_slice_when_watermarks_installed_should_replace_not_merge() { + let mut table = slice(4); + table.commit_request(9, 3, 1); + table.install_watermarks([(1, 4, 50), (2, 7, 60)]); + + assert_eq!(table.count(), 2); + assert!( + !table.is_duplicate(9, 3), + "install replaces rather than merges" + ); + assert!(table.is_duplicate(1, 4)); + assert!(!table.is_duplicate(2, 8)); + } + + #[test] + fn given_partition_slice_when_exported_should_sort_ascending_by_client() { + let mut table = slice(8); + for (commit_op, client) in [30u128, 10, 20].into_iter().enumerate() { + table.commit_request(client, 1, commit_op as u64); + } + + let clients: Vec = table + .watermarks_sorted() + .into_iter() + .map(|(client, _, _)| client) + .collect(); + assert_eq!(clients, vec![10, 20, 30]); + } + + #[test] + fn given_partition_slice_when_cleared_should_admit_everything() { + let mut table = slice(4); + table.commit_request(7, 5, 100); + table.install_watermarks(std::iter::empty()); + + assert_eq!(table.count(), 0); + assert!(!table.is_duplicate(7, 5)); + } + + #[test] + fn given_partition_slice_when_client_is_exempt_should_record_nothing() { + // Zero is reserved cluster-wide; every mutating entry point refuses it + // rather than asserting, so it doubles as the opt-out marker. + let mut table = slice(4); + table.commit_request(ClientTable::EXEMPT_CLIENT, 5, 100); + + assert_eq!(table.count(), 0); + assert!(!table.is_duplicate(ClientTable::EXEMPT_CLIENT, 5)); + } + + #[test] + #[should_panic(expected = "an epoch-fencing table must use check_request")] + fn given_metadata_table_when_is_duplicate_called_should_panic() { + let _ = ClientTable::new(4).is_duplicate(7, 1); + } + + #[test] + #[should_panic(expected = "a reply-caching table must use commit_reply")] + fn given_metadata_table_when_commit_request_called_should_panic() { + ClientTable::new(4).commit_request(7, 1, 1); + } + // Resizing an empty table swaps its slot count in: a smaller cap then // evicts once the new bound is reached. #[test] diff --git a/core/consensus/src/impls.rs b/core/consensus/src/impls.rs index 2e4741dab2..934857ed28 100644 --- a/core/consensus/src/impls.rs +++ b/core/consensus/src/impls.rs @@ -176,6 +176,14 @@ pub const PROBE_ATTEMPTS_MAX: u32 = 5; /// When exceeded, the client with the oldest committed request is evicted. pub const CLIENTS_TABLE_MAX: usize = 8192; +/// Default live dedup entries per PARTITION consensus group. +/// +/// Far below [`CLIENTS_TABLE_MAX`] because this budget is per group rather than +/// per node: the worst case scales with partition count, so it is sized to the +/// producers one partition sees. Pinned against the +/// `[partition] dedup_clients_max` default by a bootstrap assert. +pub const PARTITION_DEDUP_CLIENTS_MAX: usize = 4096; + #[derive(Debug)] pub struct PipelineEntry { pub header: PrepareHeader, @@ -313,6 +321,22 @@ impl RequestEntry { (entry, receiver) } + /// Queued request carrying a sender the caller already owns, for a submit + /// that parked before reaching a prepare slot. Mirrors + /// [`Self::with_subscriber`], except the receiver half lives with the + /// caller rather than being minted here. + #[must_use] + pub const fn with_sender( + message: Message, + reply_sender: Option>>, + ) -> Self { + Self { + message, + received_at: 0, + reply_sender, + } + } + /// Take the reply sender for hand-off to the promoted pipeline entry. pub const fn take_reply_sender(&mut self) -> Option>> { self.reply_sender.take() @@ -656,6 +680,24 @@ impl LocalPipeline { .any(|r| r.message.header().client == client) } + /// True if either queue already holds this exact `(client, request)`. + /// + /// The partition-plane in-flight check. Narrower than + /// [`Self::has_message_from_client`] on purpose: the partition pipeline is + /// depth-`prepare_queue_depth` by design, so blocking every concurrent + /// request from one client would serialize it to one in-flight write per + /// group. Only an exact replay needs absorbing. + #[must_use] + pub fn has_message_from_client_request(&self, client: u128, request: u64) -> bool { + self.prepare_queue + .iter() + .any(|p| p.header.client == client && p.header.request == request) + || self.request_queue.iter().any(|r| { + let header = r.message.header(); + header.client == client && header.request == request + }) + } + /// Verify pipeline invariants. /// /// # Panics @@ -769,6 +811,10 @@ impl Pipeline for LocalPipeline { Self::has_message_from_client(self, client_id) } + fn has_message_from_client_request(&self, client_id: u128, request: u64) -> bool { + Self::has_message_from_client_request(self, client_id, request) + } + fn cancel_all_subscribers(&mut self) { Self::cancel_all_subscribers(self); } @@ -1685,6 +1731,16 @@ impl> VsrConsensus { self.pipeline.borrow().has_message_from_client(client_id) } + /// True iff this exact `(client, request)` is already in flight. The + /// partition plane's in-flight dedup: absorbs a replay without serializing + /// a client's pipeline depth. + #[must_use] + pub fn pipeline_has_message_from_client_request(&self, client_id: u128, request: u64) -> bool { + self.pipeline + .borrow() + .has_message_from_client_request(client_id, request) + } + /// Header of the oldest in-flight prepare. #[must_use] pub fn pipeline_head_header(&self) -> Option { diff --git a/core/consensus/src/lib.rs b/core/consensus/src/lib.rs index b88b21a4d9..0c2691eb09 100644 --- a/core/consensus/src/lib.rs +++ b/core/consensus/src/lib.rs @@ -70,13 +70,22 @@ pub trait Pipeline { fn verify(&self); /// True iff either queue carries `client_id`. Used by metadata-plane - /// preflight for in-flight dedup. Partition plane is at-least-once - /// and skips. Default `false`; falls through to slot dedup in + /// preflight for in-flight dedup; the partition plane uses the narrower + /// [`Self::has_message_from_client_request`] instead, to keep a client's + /// pipeline depth. Default `false`; falls through to slot dedup in /// `check_request`. fn has_message_from_client(&self, _client_id: u128) -> bool { false } + /// True iff either queue carries this exact `(client, request)`. The + /// partition-plane in-flight dedup check: narrow on purpose, so a client + /// keeps its pipeline depth and only an exact replay is absorbed. + /// Default `false`. + fn has_message_from_client_request(&self, _client_id: u128, _request: u64) -> bool { + false + } + /// Drop reply senders on every entry; receivers wake `Canceled`. /// View-change reset uses this to unblock awaiters while preserving /// pipeline for DVC reconciliation. @@ -161,8 +170,8 @@ where pub mod client_table; pub mod le_cursor; pub use client_table::{ - CachedReply, ClientEntrySnapshot, ClientTable, ClientTableDecodeError, ClientTableSnapshot, - ClientTableWireError, CommitReply, + CachedReply, ClientEntrySnapshot, ClientTable, ClientTableDecodeError, ClientTableMode, + ClientTableSnapshot, ClientTableWireError, CommitReply, RequestStatus, }; pub mod state_manifest; pub use state_manifest::{ @@ -176,7 +185,7 @@ pub use state_transfer::{ }; // One-shot per `PipelineEntry` for in-process commit awaiters. pub(crate) mod oneshot; -pub use oneshot::{Canceled, Receiver}; +pub use oneshot::{Canceled, Receiver, Sender, channel as oneshot_channel}; mod fatal; pub use fatal::{FatalReason, fatal}; diff --git a/core/integration/tests/cluster/mod.rs b/core/integration/tests/cluster/mod.rs index 05f42ba212..344b5f5d18 100644 --- a/core/integration/tests/cluster/mod.rs +++ b/core/integration/tests/cluster/mod.rs @@ -24,5 +24,6 @@ mod failover_client_continuity; mod metadata_checkpoint_restart; mod metadata_state_transfer; mod multi_shard_partition_convergence; +mod partition_dedup; mod partition_state_transfer; mod register_forwarding; diff --git a/core/integration/tests/cluster/partition_dedup.rs b/core/integration/tests/cluster/partition_dedup.rs new file mode 100644 index 0000000000..465ae078ca --- /dev/null +++ b/core/integration/tests/cluster/partition_dedup.rs @@ -0,0 +1,660 @@ +// 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. + +//! Spec tests for partition-plane request dedup (IGGY-274). +//! +//! Each partition consensus group keeps a slice of the VSR client table: +//! per-client request watermarks folded in at commit. A replay of an +//! already-committed `(client, request)` is answered with the empty success its +//! original earned instead of committing a second copy. +//! +//! The frames are hand-crafted on a raw TCP socket for the same reason +//! `client_table_restart` does it: the Rust SDK mints a fresh `client_id` and +//! request id per attempt, so it cannot express "the same request, twice" -- +//! which is precisely the input under test. The SDK is still used for setup and +//! for reading the log back, where it is the more honest observer. + +use bytes::{Bytes, BytesMut}; +use iggy::prelude::*; +use iggy_binary_protocol::codec::WireEncode; +use iggy_binary_protocol::consensus::{ + Command, Operation, ReplyHeader, RequestHeader, read_size_field, +}; +use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest; +use iggy_binary_protocol::requests::messages::send_messages::{RawMessage, SendMessagesEncoder}; +use iggy_binary_protocol::requests::users::LoginRegisterRequest; +use iggy_binary_protocol::{ + AckLevel, ClientVersionInfo, HEADER_SIZE, IGGY_PROTOCOL_VERSION, WireConsumer, WireIdentifier, + WireName, WirePartitioning, +}; +use integration::harness::TestHarness; +use integration::iggy_harness; +use secrecy::SecretString; +use std::mem::offset_of; +use std::net::SocketAddr; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::time::{Instant, sleep, timeout}; + +const STREAM_NAME: &str = "partition-dedup-stream"; +const TOPIC_NAME: &str = "partition-dedup-topic"; +const PARTITION_ID: u32 = 0; + +/// Fixed wire identity, so the replay frame is byte-identical to the original. +/// The SDK would randomize this. +const CLIENT_ID: u128 = 0x0DED_1234_5678; + +/// Second identity for liveness probes. The dedup watermark is a per-client +/// max, so a probe under [`CLIENT_ID`] would raise that client's watermark and +/// mask a missing transfer; the probe must not touch the identity under test. +const PROBE_CLIENT_ID: u128 = 0x0DED_9999_0001; + +const REPLY_WAIT: Duration = Duration::from_secs(10); +const COMMIT_BUDGET: Duration = Duration::from_secs(20); +const RETRY_PAUSE: Duration = Duration::from_millis(100); + +#[iggy_harness(server(system.sharding.cpu_allocation = "0..1"))] +async fn given_committed_send_when_replayed_should_absorb_without_a_second_copy( + harness: &mut TestHarness, +) { + let client = harness + .root_client_for_node(0) + .await + .expect("connect a root client"); + seed_topic(&client).await; + + let addr = harness.node(0).tcp_addr().expect("node tcp address"); + let (mut stream, session) = register(addr).await; + + let body = send_messages_body(b"only-once"); + let header = request_header(Operation::SendMessages, session, 1, body.len()); + + let original = exchange_until_committed(&mut stream, &header, &body).await; + assert_eq!(original, 0, "the original send must commit"); + + // Byte-identical replay: what a retry after a lost reply looks like. + let replayed = exchange_until_committed(&mut stream, &header, &body).await; + assert_eq!( + replayed, 0, + "an absorbed duplicate is a success, not an error" + ); + + let polled = poll_all(&client).await; + assert_eq!( + polled, 1, + "the replayed send must not append a second copy (got {polled} messages)" + ); +} + +#[iggy_harness(server(system.sharding.cpu_allocation = "0..1"))] +async fn given_committed_send_when_next_request_id_arrives_should_admit_it( + harness: &mut TestHarness, +) { + // The watermark must not wedge the client: the id above it still commits. + // Without this, "dedup works" and "the plane is broken" look identical. + let client = harness + .root_client_for_node(0) + .await + .expect("connect a root client"); + seed_topic(&client).await; + + let addr = harness.node(0).tcp_addr().expect("node tcp address"); + let (mut stream, session) = register(addr).await; + + for request in 1..=3u64 { + let body = send_messages_body(format!("message-{request}").as_bytes()); + let header = request_header(Operation::SendMessages, session, request, body.len()); + let status = exchange_until_committed(&mut stream, &header, &body).await; + assert_eq!(status, 0, "request {request} must commit"); + } + + let polled = poll_all(&client).await; + assert_eq!(polled, 3, "each distinct request id must append once"); +} + +#[iggy_harness(server(system.sharding.cpu_allocation = "0..1"))] +async fn given_gapped_request_id_when_sent_should_commit(harness: &mut TestHarness) { + // One client counter feeds every group it writes to, so a slice only ever + // sees a subset of the ids minted. Gaps must be legal, not a wedge. + let client = harness + .root_client_for_node(0) + .await + .expect("connect a root client"); + seed_topic(&client).await; + + let addr = harness.node(0).tcp_addr().expect("node tcp address"); + let (mut stream, session) = register(addr).await; + + for request in [1u64, 9, 40] { + let body = send_messages_body(format!("gap-{request}").as_bytes()); + let header = request_header(Operation::SendMessages, session, request, body.len()); + let status = exchange_until_committed(&mut stream, &header, &body).await; + assert_eq!(status, 0, "gapped request {request} must commit"); + } + + let polled = poll_all(&client).await; + assert_eq!(polled, 3, "a gapped id is new, not a duplicate"); +} + +#[iggy_harness(server(system.sharding.cpu_allocation = "0..1"))] +async fn given_committed_consumer_offset_when_replayed_should_absorb(harness: &mut TestHarness) { + // Dedup covers every replicated partition write, not just produces. A + // replayed offset store must answer success rather than committing twice. + let client = harness + .root_client_for_node(0) + .await + .expect("connect a root client"); + seed_topic(&client).await; + + let addr = harness.node(0).tcp_addr().expect("node tcp address"); + let (mut stream, session) = register(addr).await; + + // Seed a message so offset 0 is in range for the store. + let produce = send_messages_body(b"seed"); + let produce_header = request_header(Operation::SendMessages, session, 1, produce.len()); + assert_eq!( + exchange_until_committed(&mut stream, &produce_header, &produce).await, + 0, + "the seed produce must commit" + ); + + let body = store_offset_body(0); + let header = request_header(Operation::StoreConsumerOffset, session, 2, body.len()); + + let original = exchange_until_committed(&mut stream, &header, &body).await; + assert_eq!(original, 0, "the original offset store must commit"); + + let replayed = exchange_until_committed(&mut stream, &header, &body).await; + assert_eq!( + replayed, 0, + "a replayed offset store is absorbed as a success" + ); + + // The next id still gets through: the watermark must not wedge the client. + let next = store_offset_body(0); + let next_header = request_header(Operation::StoreConsumerOffset, session, 3, next.len()); + assert_eq!( + exchange_until_committed(&mut stream, &next_header, &next).await, + 0, + "the id above the watermark must still commit" + ); +} + +/// `StoreConsumerOffset` body for the raw connection's own consumer id. +fn store_offset_body(offset: u64) -> Bytes { + StoreConsumerOffsetRequest { + consumer: WireConsumer::consumer(WireIdentifier::numeric(1)), + stream_id: WireIdentifier::named(STREAM_NAME).expect("stream identifier"), + topic_id: WireIdentifier::named(TOPIC_NAME).expect("topic identifier"), + partition_id: Some(PARTITION_ID), + offset, + ack: AckLevel::Quorum, + } + .to_bytes() +} + +/// State-transfer choreography end to end: rejoin, view changes, and the +/// final commits ride slow CI runners. +const TRANSFER_BUDGET: Duration = Duration::from_secs(60); +const FINAL_COMMIT_BUDGET: Duration = Duration::from_secs(120); +const MARKER_POLL: Duration = Duration::from_millis(200); +const INSTALL_MARKER: &str = "partition state transfer installed"; + +/// Pre-stop produces fold into every replica's slice live; the rest commit +/// while node 2 is down. Total must push the evicted ring (capacity 64) past +/// the rejoiner's durable end, or repair closes the gap and no transfer runs. +const PRE_STOP_SENDS: u64 = 40; +/// The identity under test stops sending here; everything after comes from the +/// filler client. The rejoiner's tail repair re-applies the ring window (the +/// LAST ~64 commits) through the ordinary commit path, and the watermark is a +/// per-client max -- so if the tested client appeared anywhere in that window, +/// repair alone would cover every lower id and the artifact would be +/// redundant. The filler pushes the tested client's last send out of the ring, +/// leaving the transferred artifact as node 2's ONLY source for it. +const TESTED_CLIENT_SENDS: u64 = 140; +const FILLER_SENDS: u64 = 100; +const TOTAL_SENDS: u64 = TESTED_CLIENT_SENDS + FILLER_SENDS; +/// Replayed id: the tested client's watermark itself. Absorbing it requires an +/// entry for that client, which only the transferred artifact can supply. +const REPLAYED_REQUEST: u64 = TESTED_CLIENT_SENDS; + +/// Filler identity whose sends evict the tested client from the repair ring. +const FILLER_CLIENT_ID: u128 = 0x0DED_F111_E400; + +/// Sentinel status for an Eviction frame: the connection's session is gone and +/// the caller must reconnect and re-register before retrying. +const EVICTED: u32 = u32::MAX; + +#[iggy_harness( + cluster_nodes = 3, + server( + system.sharding.cpu_allocation = "0..1", + partition.evicted_ring_capacity = "64" + ) +)] +async fn given_transferred_dedup_slice_when_old_request_replays_should_absorb( + harness: &mut TestHarness, +) { + // Phase 1: node 0 is every group's view-0 primary. Produce the pre-stop + // window with node 2 live, then the rest with it stopped, so the second + // window exists on node 2 only via state transfer. + let client = harness + .root_client_for_node(0) + .await + .expect("connect a root client"); + seed_topic(&client).await; + let addr = harness.node(0).tcp_addr().expect("node 0 tcp address"); + let (mut stream, session) = register(addr).await; + raw_produce(&mut stream, session, 1..=PRE_STOP_SENDS, COMMIT_BUDGET).await; + sleep(Duration::from_secs(1)).await; + harness.stop_node(2).expect("stop node 2"); + raw_produce( + &mut stream, + session, + (PRE_STOP_SENDS + 1)..=TESTED_CLIENT_SENDS, + COMMIT_BUDGET, + ) + .await; + drop(stream); + let (mut filler, filler_session) = + register_client_with_budget(addr, FILLER_CLIENT_ID, COMMIT_BUDGET).await; + raw_produce_for( + &mut filler, + FILLER_CLIENT_ID, + filler_session, + 1..=FILLER_SENDS, + COMMIT_BUDGET, + ) + .await; + drop(filler); + drop(client); + + // Phase 2: the rejoin cannot repair past the survivors' evicted ring, so + // it converts to state transfer; the install carries the dedup section. + harness.restart_node(2).expect("restart node 2"); + await_marker(harness, 2, INSTALL_MARKER).await; + + // Phase 3: walk the primaries off node 0 and node 1 so the REPLAY is + // admitted by the transferred node. Stopping node 0 elects node 1 + // (view 1); after node 0 rejoins, stopping node 1 elects node 2 (view 2) + // with quorum {0, 2}. + harness.stop_node(0).expect("stop node 0"); + sleep(Duration::from_secs(2)).await; + harness.restart_node(0).expect("restart node 0"); + sleep(Duration::from_secs(2)).await; + harness.stop_node(1).expect("stop node 1"); + + // Phase 4, on node 2. The probe send goes FIRST and under a DIFFERENT + // client: its commit proves the view settled on node 2 and the rejoined + // node 0 is acking, and it pins the expected count -- while leaving + // CLIENT_ID's watermark exactly what the transfer installed (the watermark + // is a per-client max, so a same-client probe would mask a missing + // transfer). Sends reconnect + re-register on eviction; dedup keys on the + // client id and must hold across a re-register. + let addr = harness.node(2).tcp_addr().expect("node 2 tcp address"); + let fresh = send_reconnecting(addr, PROBE_CLIENT_ID, 1, FINAL_COMMIT_BUDGET).await; + assert_eq!( + fresh, 0, + "the probe client's send must commit on the new primary" + ); + + let replayed = send_reconnecting(addr, CLIENT_ID, REPLAYED_REQUEST, FINAL_COMMIT_BUDGET).await; + assert_eq!( + replayed, 0, + "a replay of a transferred watermark is absorbed as a success" + ); + + // The count is the discriminator: an absorbed replay leaves it at + // TOTAL_SENDS + 1; a re-execution (empty transferred slice) appends a + // second copy of the replayed payload. + let client = harness + .root_client_for_node(2) + .await + .expect("connect a root client to node 2"); + let polled = poll_up_to(&client, (TOTAL_SENDS + 16) as u32).await; + assert_eq!( + u64::from(polled), + TOTAL_SENDS + 1, + "the transferred slice must absorb the replay instead of re-executing it" + ); +} + +/// One send under `request`, surviving evictions: reconnect, re-register, and +/// retry the identical frame until it answers or the budget runs out. +async fn send_reconnecting(addr: SocketAddr, client: u128, request: u64, budget: Duration) -> u32 { + let deadline = Instant::now() + budget; + let body = send_messages_body(format!("send-{request}").as_bytes()); + loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + assert!( + remaining > Duration::ZERO, + "request {request} did not resolve within {budget:?}" + ); + let (mut stream, session) = register_client_with_budget(addr, client, remaining).await; + let header = request_header_for( + client, + Operation::SendMessages, + session, + request, + body.len(), + ); + let status = exchange_with_budget(&mut stream, &header, &body, remaining).await; + if status != EVICTED { + return status; + } + sleep(RETRY_PAUSE).await; + } +} + +/// Produce one single-message batch per request id over the lockstep raw +/// connection, waiting out each commit. +async fn raw_produce( + stream: &mut TcpStream, + session: u64, + requests: std::ops::RangeInclusive, + budget: Duration, +) { + raw_produce_for(stream, CLIENT_ID, session, requests, budget).await; +} + +async fn raw_produce_for( + stream: &mut TcpStream, + client: u128, + session: u64, + requests: std::ops::RangeInclusive, + budget: Duration, +) { + for request in requests { + let body = send_messages_body(format!("send-{request}").as_bytes()); + let header = request_header_for( + client, + Operation::SendMessages, + session, + request, + body.len(), + ); + let status = exchange_with_budget(stream, &header, &body, budget).await; + assert_eq!(status, 0, "request {request} must commit"); + } +} + +async fn await_marker(harness: &TestHarness, node: usize, marker: &str) { + let deadline = Instant::now() + TRANSFER_BUDGET; + while !harness.node(node).stdout_contains(marker) { + assert!( + Instant::now() < deadline, + "node {node} never logged {marker:?} within {TRANSFER_BUDGET:?}" + ); + sleep(MARKER_POLL).await; + } +} + +async fn seed_topic(client: &IggyClient) { + client + .create_stream(STREAM_NAME) + .await + .expect("create stream"); + let stream_id = Identifier::named(STREAM_NAME).expect("stream identifier"); + client + .create_topic( + &stream_id, + TOPIC_NAME, + &TopicCreateOptions { + partitions_count: Some(1), + message_expiry: Some(IggyExpiry::NeverExpire), + // Every commit flushes and ring-evicts, which is what marches + // the repair floor past a rejoiner and forces the transfer the + // transferred-slice spec depends on. + messages_required_to_save: Some(1), + ..TopicCreateOptions::default() + }, + ) + .await + .expect("create topic"); +} + +async fn poll_all(client: &IggyClient) -> u32 { + poll_up_to(client, 100).await +} + +async fn poll_up_to(client: &IggyClient, max: u32) -> u32 { + let stream_id = Identifier::named(STREAM_NAME).expect("stream identifier"); + let topic_id = Identifier::named(TOPIC_NAME).expect("topic identifier"); + client + .poll_messages( + &stream_id, + &topic_id, + Some(PARTITION_ID), + &Consumer::new(Identifier::numeric(1).expect("consumer identifier")), + &PollingStrategy::offset(0), + max, + false, + ) + .await + .expect("poll messages") + .messages + .len() as u32 +} + +/// Full `SendMessages` body: metadata prefix, batch header, one message. +fn send_messages_body(payload: &[u8]) -> Bytes { + let stream_id = WireIdentifier::named(STREAM_NAME).expect("stream identifier"); + let topic_id = WireIdentifier::named(TOPIC_NAME).expect("topic identifier"); + let partitioning = WirePartitioning::PartitionId(PARTITION_ID); + let messages = [RawMessage { + // A fixed id keeps the replay byte-identical; a zero would be + // server-stamped and the two frames would diverge. + id: 0x5EED, + origin_timestamp: 0, + headers: None, + payload, + }]; + let size = SendMessagesEncoder::encoded_size(&stream_id, &topic_id, &partitioning, &messages); + let mut buf = BytesMut::with_capacity(size); + SendMessagesEncoder::encode(&mut buf, &stream_id, &topic_id, &partitioning, &messages) + .expect("encode send_messages body"); + buf.freeze() +} + +fn request_header( + operation: Operation, + session: u64, + request: u64, + body_len: usize, +) -> RequestHeader { + request_header_for(CLIENT_ID, operation, session, request, body_len) +} + +fn request_header_for( + client: u128, + operation: Operation, + session: u64, + request: u64, + body_len: usize, +) -> RequestHeader { + RequestHeader { + command: Command::Request, + operation, + size: u32::try_from(HEADER_SIZE + body_len).unwrap(), + client, + session, + request, + ..Default::default() + } +} + +/// Exchange until the server stops answering transiently, returning the reply +/// status. A transient means the request was never admitted, so replaying it +/// keeps the same id -- exactly what the SDK's own retry loop does. +async fn exchange_until_committed( + stream: &mut TcpStream, + header: &RequestHeader, + body: &Bytes, +) -> u32 { + exchange_with_budget(stream, header, body, COMMIT_BUDGET).await +} + +async fn exchange_with_budget( + stream: &mut TcpStream, + header: &RequestHeader, + body: &Bytes, + budget: Duration, +) -> u32 { + let deadline = Instant::now() + budget; + loop { + let status = exchange(stream, header, body).await; + if !is_transient(status) { + return status; + } + assert!( + Instant::now() < deadline, + "request {} stayed transient for {budget:?}", + header.request + ); + sleep(RETRY_PAUSE).await; + } +} + +/// Write one frame, read one frame, return the reply status. The connection is +/// lockstep, so the reply that comes back is this request's. +async fn exchange(stream: &mut TcpStream, header: &RequestHeader, body: &Bytes) -> u32 { + stream.write_all(bytemuck::bytes_of(header)).await.unwrap(); + if !body.is_empty() { + stream.write_all(body).await.unwrap(); + } + + let mut reply_header = [0u8; HEADER_SIZE]; + timeout(REPLY_WAIT, stream.read_exact(&mut reply_header)) + .await + .expect("reply header timed out") + .expect("reply header read failed"); + + let command_offset = offset_of!(RequestHeader, command); + if reply_header[command_offset] == Command::Eviction as u8 { + // The session died (view change, epoch fence): the contract is + // reconnect + re-register, and dedup must still hold because it keys + // on the client id, not the session. + return EVICTED; + } + assert_eq!( + reply_header[command_offset], + Command::Reply as u8, + "expected a Reply frame" + ); + + let status_offset = offset_of!(ReplyHeader, status); + let status = u32::from_le_bytes( + reply_header[status_offset..status_offset + 4] + .try_into() + .unwrap(), + ); + let total_size = read_size_field(&reply_header).expect("reply size field") as usize; + if total_size > HEADER_SIZE { + let mut discard = vec![0u8; total_size - HEADER_SIZE]; + timeout(REPLY_WAIT, stream.read_exact(&mut discard)) + .await + .expect("reply body timed out") + .expect("reply body read failed"); + } + status +} + +/// Register `CLIENT_ID` as root, returning the connection and its bound +/// session. The session binds to THIS socket server-side, so every frame in a +/// test must reuse the returned stream. +async fn register(addr: SocketAddr) -> (TcpStream, u64) { + register_with_budget(addr, COMMIT_BUDGET).await +} + +/// Fresh socket per attempt: a login refused mid-election may come back as an +/// eviction that poisons the connection. +async fn register_with_budget(addr: SocketAddr, budget: Duration) -> (TcpStream, u64) { + register_client_with_budget(addr, CLIENT_ID, budget).await +} + +async fn register_client_with_budget( + addr: SocketAddr, + client: u128, + budget: Duration, +) -> (TcpStream, u64) { + let deadline = Instant::now() + budget; + loop { + let mut stream = TcpStream::connect(addr).await.unwrap(); + if let Some(session) = login_on(&mut stream, client).await { + return (stream, session); + } + assert!( + Instant::now() < deadline, + "register did not commit within {budget:?}" + ); + sleep(RETRY_PAUSE).await; + } +} + +async fn login_on(stream: &mut TcpStream, client: u128) -> Option { + let body = LoginRegisterRequest { + version_info: ClientVersionInfo { + protocol_version: IGGY_PROTOCOL_VERSION, + sdk_name: WireName::new("iggy274-raw").unwrap(), + sdk_version: WireName::new("0.0.1").unwrap(), + }, + username: WireName::new(DEFAULT_ROOT_USERNAME).unwrap(), + password: SecretString::from(DEFAULT_ROOT_PASSWORD), + client_context: None, + } + .to_bytes(); + let header = request_header_for(client, Operation::Register, 0, 0, body.len()); + + stream.write_all(bytemuck::bytes_of(&header)).await.unwrap(); + stream.write_all(&body).await.unwrap(); + + let mut reply_header = [0u8; HEADER_SIZE]; + let Ok(Ok(_)) = timeout(REPLY_WAIT, stream.read_exact(&mut reply_header)).await else { + return None; + }; + let command_offset = offset_of!(RequestHeader, command); + if reply_header[command_offset] != Command::Reply as u8 { + return None; + } + + let status_offset = offset_of!(ReplyHeader, status); + let status = u32::from_le_bytes( + reply_header[status_offset..status_offset + 4] + .try_into() + .unwrap(), + ); + let total_size = read_size_field(&reply_header).expect("login reply size") as usize; + let mut reply_body = vec![0u8; total_size - HEADER_SIZE]; + let Ok(Ok(_)) = timeout(REPLY_WAIT, stream.read_exact(&mut reply_body)).await else { + return None; + }; + if status != 0 { + return None; + } + let session_offset = offset_of!(ReplyHeader, commit); + Some(u64::from_le_bytes( + reply_header[session_offset..session_offset + 8] + .try_into() + .unwrap(), + )) +} + +fn is_transient(code: u32) -> bool { + code == IggyError::TransientNotCommitted.as_code() + || code == IggyError::TransientNotAccepted.as_code() +} diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index fb1ba35c2f..915a191cfe 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -36,9 +36,9 @@ use crate::{ PollingConsumer, }; use consensus::{ - CommitLogEvent, Consensus, PartitionDiagEvent, PipelineEntry, PlaneKind, Project, - ReplicaLogContext, RequestLogEvent, Sequencer, SimEventKind, VsrConsensus, ack_preflight, - ack_quorum_reached, build_deny_reply_from_request, build_reply_from_request, + ClientTable, ClientTableMode, CommitLogEvent, Consensus, PartitionDiagEvent, PipelineEntry, + PlaneKind, Project, ReplicaLogContext, RequestLogEvent, Sequencer, SimEventKind, VsrConsensus, + ack_preflight, ack_quorum_reached, build_deny_reply_from_request, build_reply_from_request, build_reply_message, drain_committable_prefix, emit_namespace_progress_event, emit_partition_diag, emit_sim_event, fence_old_prepare_by_commit, replicate_frozen_to_next_in_chain, replicate_preflight, restamp_prepare_view, @@ -53,7 +53,7 @@ use iggy_binary_protocol::responses::messages::{ use iggy_binary_protocol::{ AckLevel, GenericHeader, Operation, PrepareHeader, WireDecode, WireEncode, WireIdentifier, }; -use iggy_binary_protocol::{PrepareOkHeader, RoutedRequestHeader}; +use iggy_binary_protocol::{PrepareOkHeader, ReplyHeader, RoutedRequestHeader}; use iggy_common::{ ConsumerGroupId, ConsumerGroupOffsets, ConsumerKind, ConsumerOffset, ConsumerOffsets, IggyByteSize, IggyError, IggyExpiry, IggyTimestamp, PartitionStats, PollingKind, @@ -86,16 +86,21 @@ use tokio::sync::Mutex as TokioMutex; use tracing::{debug, warn}; // This struct aliases in terms of the code contained the `LocalPartition from `core/server/src/streaming/partitions/local_partition.rs`. -// -// Note: there is no per-client write dedup at the partition plane. -// `SendMessages` retries are at-least-once and may commit multiple times. -// Duplicate suppression is a consensus-layer concern: the VSR client table -// dedups by request id (at-most-once), so the data plane needs no message-id set. pub struct IggyPartition where B: MessageBus, { consensus: VsrConsensus, + /// This group's slice of the VSR client table, run in + /// [`ClientTableMode::PARTITION_SLICE`]: per-client request watermarks + /// folded in at commit, so every replica derives the same slice from the + /// same log. The mode turns off what this plane cannot use -- no reply ring + /// (`SendMessages` has no result section, so a duplicate is answered by + /// synthesizing the empty success its original earned), no epoch fence (a + /// partition group never observes a `Register`), and no preallocated slot + /// array (one table per group, where preallocating the cap would reserve + /// hundreds of KiB per partition before a client connects). + dedup: ClientTable, pub log: SegmentedLog>, /// Highest durably persisted offset. pub offset: Arc, @@ -455,6 +460,10 @@ where let single_replica = consensus.replica_count() == 1; let partition = Self { consensus, + dedup: ClientTable::with_mode( + consensus::PARTITION_DEDUP_CLIENTS_MAX, + ClientTableMode::PARTITION_SLICE, + ), log: SegmentedLog::default(), offset: Arc::new(AtomicU64::new(0)), dirty_offset: AtomicU64::new(0), @@ -549,6 +558,25 @@ where &self.consensus } + /// This group's dedup slice. Read at admission to classify a request, + /// written only from the commit path. + #[must_use] + pub const fn dedup(&self) -> &ClientTable { + &self.dedup + } + + /// Mutable slice, for the commit path and state-transfer install. + pub(crate) const fn dedup_mut(&mut self) -> &mut ClientTable { + &mut self.dedup + } + + /// Size the dedup slice to `[partition] dedup_clients_max`. Boot-time: + /// shrinking a live slice evicts its oldest-committed entries, which costs + /// dedup coverage for those clients rather than correctness. + pub fn set_dedup_clients_max(&mut self, clients_max: usize) { + self.dedup.set_capacity(clients_max.max(1)); + } + #[must_use] pub fn with_in_memory_storage( stats: Arc, @@ -1370,8 +1398,9 @@ where } /// `AckLevel::NoAck` fast path: persist, apply, send reply, no - /// replication. Single-replica durability. No reply cache: partition - /// plane is at-least-once; session lifecycle lives on metadata. + /// replication. Single-replica durability. Never recorded in the dedup + /// slice: it does not replicate, so folding it in would fork the slice + /// across replicas. Session lifecycle lives on metadata. #[allow(clippy::future_not_send)] async fn apply_consumer_offset_no_ack( &self, @@ -1379,6 +1408,7 @@ where kind: ConsumerKind, consumer_id: u32, offset: Option, + waiter: Option>>, ) { let pending = offset.map_or_else( || PendingConsumerOffsetCommit::delete(kind, consumer_id), @@ -1409,6 +1439,12 @@ where &request_header, committed_reply_body(request_header.operation), ); + // Same rule as the committed path: a submit's waiter takes the reply, + // because `header.client` is then the VSR consensus id. + if let Some(waiter) = waiter { + let _ = waiter.send(reply); + return; + } let reply_buffers = reply.into_generic().into_frozen(); if let Err(error) = self .consensus @@ -1862,16 +1898,27 @@ where /// Project a client request into a prepare. /// - /// At-least-once: no per-client dedup. `SendMessages` retry -> fresh - /// prepare, may re-commit at new offset. Consumers handle dedup - /// (message key / content / producer-id+seq). Session lifecycle + - /// eviction live on metadata plane. + /// A replay of a committed `(client, request)` is absorbed by this group's + /// dedup slice; anything above the watermark projects into a prepare. + /// Session lifecycle + eviction live on the metadata plane. /// /// # Panics /// Panics if called when this partition's consensus instance is not the /// primary, is not in normal status, or is currently syncing. #[allow(clippy::future_not_send, clippy::too_many_lines)] - pub async fn on_request(&mut self, message: Message) { + /// `reply` is the in-process channel a `PartitionSubmit` carried in. When + /// present the committed reply fires on it instead of going to the bus: + /// the connection-owning shard writes it to the socket it holds, because + /// `header.client` is the VSR consensus id and carries no home-shard + /// routing. `None` keeps the bus path (auto-commit ops, tests). + pub async fn on_request( + &mut self, + message: Message, + reply: Option>>, + ) { + // Taken by whichever arm answers: the deny paths, the NoAck fast path, + // or the pipeline entry that fires it at commit. Exactly one runs. + let mut reply = reply; self.clear_pending_consumer_offset_commits_if_view_changed(); let namespace = IggyNamespace::from_raw(message.header().group); let client_id = message.header().client; @@ -1941,6 +1988,72 @@ where _ => None, }; + // A client op landing on a non-primary (or mid-view-change) + // replica is a routing artifact -- e.g. the roster still points + // here while this group's primaryship moved after a restart. + // Answer the typed transient instead of asserting: the SDK + // replays and its leader recheck re-routes, whereas a panic + // kills the shard and a silent drop wedges the client until its + // read timeout. + if consensus.is_follower() || !consensus.is_normal() || consensus.is_transferring() { + emit_partition_diag( + tracing::Level::WARN, + &PartitionDiagEvent::new( + ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions), + "rejecting client request on non-primary partition replica", + ) + .with_operation(message.header().operation), + ); + Self::send_partition_deny_or_log( + consensus, + message.header(), + IggyError::TransientNotAccepted.as_code(), + "non-primary transient reply send failed", + reply.take(), + ) + .await; + return; + } + + // Dedup BEFORE the admission checks below: a replay of an + // already-committed delete must answer the success its original + // earned, not the typed 404 the existence check would raise now + // that the offset is gone. + // + // A replay racing its own in-flight original is absorbed here: the + // slice only knows committed ops, so it cannot yet see the copy + // still in the pipeline. Keyed on the exact `(client, request)` -- + // matching any request from the client would serialize its + // pipeline depth to one in-flight write per group. + if !is_auto_commit_client(client_id) { + if consensus.pipeline_has_message_from_client_request(client_id, request) { + Self::send_partition_deny_or_log( + consensus, + message.header(), + IggyError::TransientNotCommitted.as_code(), + "in-flight dedup transient reply send failed", + reply.take(), + ) + .await; + return; + } + if self.dedup.is_duplicate(client_id, request) { + let committed = build_reply_from_request( + &self.consensus, + message.header(), + committed_reply_body(message.header().operation), + ); + Self::answer_duplicate_or_log( + &self.consensus, + message.header(), + committed, + reply.take(), + ) + .await; + return; + } + } + if matches!(message.header().operation, Operation::DeleteConsumerOffset) && let Some((kind, consumer_id, _, _)) = consumer_offset && let Err(error) = self.ensure_consumer_offset_exists(kind, consumer_id) @@ -1965,6 +2078,7 @@ where message.header(), error.as_code(), "delete_consumer_offset deny reply send failed", + reply.take(), ) .await; return; @@ -1999,38 +2113,13 @@ where message.header(), IggyError::InvalidOffset(requested_offset).as_code(), "store_consumer_offset deny reply send failed", + reply.take(), ) .await; return; } } - // A client op landing on a non-primary (or mid-view-change) - // replica is a routing artifact -- e.g. the roster still points - // here while this group's primaryship moved after a restart. - // Answer the typed transient instead of asserting: the SDK - // replays and its leader recheck re-routes, whereas a panic - // kills the shard and a silent drop wedges the client until its - // read timeout. - if consensus.is_follower() || !consensus.is_normal() || consensus.is_transferring() { - emit_partition_diag( - tracing::Level::WARN, - &PartitionDiagEvent::new( - ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions), - "rejecting client request on non-primary partition replica", - ) - .with_operation(message.header().operation), - ); - Self::send_partition_deny_or_log( - consensus, - message.header(), - IggyError::TransientNotAccepted.as_code(), - "non-primary transient reply send failed", - ) - .await; - return; - } - // NoAck -> fast path. Quorum -> VSR pipeline. if let Some((kind, consumer_id, offset, AckLevel::NoAck)) = consumer_offset && matches!( @@ -2049,8 +2138,9 @@ where // request room -> buffer; both full -> drop+warn (client retries // via read-timeout). if consensus.pipeline_is_full() { - let push_result = - consensus.push_queued_request(consensus::RequestEntry::new(message)); + let push_result = consensus.push_queued_request( + consensus::RequestEntry::with_sender(message, reply.take()), + ); if push_result.is_err() { emit_partition_diag( tracing::Level::WARN, @@ -2065,7 +2155,14 @@ where let prepare = message.project(consensus); consensus.verify_pipeline(); - consensus.pipeline_message(PlaneKind::Partitions, &prepare); + match reply.take() { + Some(sender) => consensus.pipeline_message_with_sender( + PlaneKind::Partitions, + &prepare, + sender, + ), + None => consensus.pipeline_message(PlaneKind::Partitions, &prepare), + } Disposition::Replicate(prepare) } }; @@ -2078,17 +2175,23 @@ where consumer_id, offset, } => { - self.apply_consumer_offset_no_ack(request_header, kind, consumer_id, offset) - .await; + self.apply_consumer_offset_no_ack( + request_header, + kind, + consumer_id, + offset, + reply.take(), + ) + .await; } } } /// Promote up to `slots_freed` buffered requests into prepares post-commit. /// - /// No preflight: partition plane is at-least-once with no `ClientTable` - /// dedup. Buffered `SendMessages` retry commits at fresh offset; consumers - /// dedup by message key / content / producer-id+seq. + /// Promotion runs no preflight: the request was classified at admission + /// and the slice cannot have gained a higher watermark for it since (only + /// a commit moves it, and this entry has not committed). /// /// Per-iteration `is_primary && is_normal && !is_transferring` asserts inlined /// (closure form's `&consensus` borrow conflicts with `&mut self`). Guards @@ -3150,7 +3253,7 @@ where let committed_batch_stats = self.resolve_committed_visible_offsets(&drained); let mut messages_committed = false; - for (entry, batch_stats) in drained.into_iter().zip(committed_batch_stats) { + for (mut entry, batch_stats) in drained.into_iter().zip(committed_batch_stats) { let prepare_header = entry.header; if !self .commit_partition_entry( @@ -3180,6 +3283,19 @@ where self.consensus.advance_commit_min(prepare_header.op); + // Fold the committed request into this group's dedup slice. Runs on + // EVERY replica, not just the one that replies, so a promoted + // primary can absorb a replay of what its predecessor committed. + // Auto-commit ops carry the reserved sentinel client and no client + // ever replays them. + if !is_auto_commit_client(prepare_header.client) { + self.dedup.commit_request( + prepare_header.client, + prepare_header.request, + prepare_header.op, + ); + } + let pipeline_depth = self.consensus.pipeline_len(); let event = CommitLogEvent { replica: ReplicaLogContext::from_consensus(&self.consensus, PlaneKind::Partitions), @@ -3197,9 +3313,11 @@ where pipeline_depth, ); - // No reply cache: at-least-once means retries re-commit at new - // offsets. Only primary delivers replies; backups just advance - // commit. Session lifecycle is metadata-only. + // No reply cache: an absorbed duplicate is answered by + // synthesizing the same empty success at admission, so no committed + // bytes need keeping. Only the primary delivers replies; backups + // just advance commit and fold the slice. Session lifecycle is + // metadata-only. // // A server-generated auto-commit op (a poll's `auto_commit`, // replicated for failover) carries the reserved @@ -3214,13 +3332,19 @@ where operation => committed_reply_body(operation), }; let reply = build_reply_message(&prepare_header, &body); - let reply_buffers = reply.into_generic().into_frozen(); emit_sim_event(SimEventKind::ClientReplyEmitted, &event); - if let Err(error) = self + // An in-process waiter takes the reply instead of the bus: it + // arrived as a `PartitionSubmit`, so `header.client` is the VSR + // consensus id and carries no home-shard routing. The awaiting + // shard owns the socket. A dropped receiver is ignored -- the + // client recovers on its own read-timeout. + if let Some(sender) = entry.take_reply_sender() { + let _ = sender.send(reply); + } else if let Err(error) = self .consensus .message_bus() - .send_to_client(prepare_header.client, reply_buffers) + .send_to_client(prepare_header.client, reply.into_generic().into_frozen()) .await { tracing::error!( @@ -3439,13 +3563,51 @@ where /// body, op=0), logging a WARN under `send_fail_label` if the reply send /// fails. Callers deny on the primary, before the op enters the pipeline, /// so nothing replicates. + /// Answer an absorbed duplicate with the reply its original earned. Same + /// delivery split as [`Self::send_partition_deny_or_log`]: the submit's + /// channel when one is waiting, the bus otherwise. + async fn answer_duplicate_or_log( + consensus: &VsrConsensus, + header: &RoutedRequestHeader, + reply: Message, + waiter: Option>>, + ) { + if let Some(waiter) = waiter { + let _ = waiter.send(reply); + return; + } + if let Err(send_error) = consensus + .message_bus() + .send_to_client(header.client, reply.into_generic().into_frozen()) + .await + { + emit_partition_diag( + tracing::Level::WARN, + &PartitionDiagEvent::new( + ReplicaLogContext::from_consensus(consensus, PlaneKind::Partitions), + "duplicate reply send failed", + ) + .with_operation(header.operation) + .with_error(send_error.to_string()), + ); + } + } + + /// `waiter` is the submit's in-process channel, taken by the caller. When + /// present the deny goes there: `header.client` is then the VSR consensus + /// id, which the bus cannot route. async fn send_partition_deny_or_log( consensus: &VsrConsensus, header: &RoutedRequestHeader, status: u32, send_fail_label: &'static str, + waiter: Option>>, ) { let reply = build_deny_reply_from_request(consensus, header, status); + if let Some(waiter) = waiter { + let _ = waiter.send(reply); + return; + } if let Err(send_error) = consensus .message_bus() .send_to_client(header.client, reply.into_generic().into_frozen()) @@ -5475,7 +5637,7 @@ mod tests { let consumer_id: u32 = 5; partition - .on_request(delete_offset_request(client_id, 7, consumer_id)) + .on_request(delete_offset_request(client_id, 7, consumer_id), None) .await; { @@ -5513,7 +5675,7 @@ mod tests { ConsumerOffset::new(ConsumerKind::Consumer, consumer_id, 3, String::new()), ); partition - .on_request(delete_offset_request(client_id, 8, consumer_id)) + .on_request(delete_offset_request(client_id, 8, consumer_id), None) .await; assert_eq!( partition.consensus().pipeline_len(), @@ -6794,6 +6956,7 @@ mod tests { next_offset: 50, consumers: Vec::new(), groups: Vec::new(), + dedup: Vec::new(), }; let refused = partition .install_state_transfer(&repair_config(), 12, Vec::new(), &behind.encode(), 0) @@ -6821,6 +6984,7 @@ mod tests { next_offset: 0, consumers: Vec::new(), groups: Vec::new(), + dedup: Vec::new(), }; let accepted = partition .install_state_transfer(&repair_config(), 12, Vec::new(), &purged.encode(), 0) @@ -6860,6 +7024,7 @@ mod tests { next_offset: 50, consumers: Vec::new(), groups: Vec::new(), + dedup: Vec::new(), }; let refused = partition .install_state_transfer(&repair_config(), 12, Vec::new(), &offer.encode(), 1) @@ -6909,6 +7074,7 @@ mod tests { next_offset: 0, consumers: Vec::new(), groups: Vec::new(), + dedup: Vec::new(), }; let installed = partition .install_state_transfer(&repair_config(), 12, Vec::new(), &reset.encode(), 1) diff --git a/core/partitions/src/iggy_partitions.rs b/core/partitions/src/iggy_partitions.rs index 87887c9ee8..15f7cae6db 100644 --- a/core/partitions/src/iggy_partitions.rs +++ b/core/partitions/src/iggy_partitions.rs @@ -23,10 +23,12 @@ use crate::{IggyPartition, Partition, PollingArgs, PollingConsumer}; use ahash::AHashSet; use consensus::{Consensus, Plane, PlaneIdentity, VsrConsensus}; use iggy_binary_protocol::{ - Command, ConsensusHeader, Operation, PrepareHeader, PrepareOkHeader, RoutedRequestHeader, + Command, ConsensusHeader, Operation, PrepareHeader, PrepareOkHeader, ReplyHeader, + RoutedRequestHeader, }; use journal::superblock::{PingPongSuperblock, SuperblockStore}; use message_bus::MessageBus; +use server_common::Message; use server_common::send_messages::{ChecksumMode, convert_request_message, encrypt_batch_request}; use server_common::sharding::{IggyNamespace, LocalIdx, ShardId}; #[cfg(debug_assertions)] @@ -528,14 +530,17 @@ where } } -impl Plane> for IggyPartitions +impl IggyPartitions where B: MessageBus, SB: SuperblockStore, { - async fn on_request( + /// [`Plane::on_request`] carrying the in-process reply channel a + /// `PartitionSubmit` arrived with; `None` keeps the bus-reply path. + pub async fn on_request_with_reply( &self, message: as Consensus>::Message, + reply: Option>>, ) { let namespace = IggyNamespace::from_raw(message.header().group); if self.is_tombstoned(&namespace) { @@ -586,7 +591,20 @@ where ); return; }; - partition.on_request(message).await; + partition.on_request(message, reply).await; + } +} + +impl Plane> for IggyPartitions +where + B: MessageBus, + SB: SuperblockStore, +{ + async fn on_request( + &self, + message: as Consensus>::Message, + ) { + self.on_request_with_reply(message, None).await; } async fn on_replicate(&self, message: as Consensus>::Message) { diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs index 2e0f00c919..da8dcb5ce7 100644 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@ -50,15 +50,17 @@ use std::path::{Path, PathBuf}; use std::rc::Rc; use std::sync::atomic::Ordering; -/// Framing marker for the consumer-offsets wire artifact, "ICO1". -pub(crate) const CONSUMER_OFFSETS_MAGIC: [u8; 4] = *b"ICO1"; +/// Framing marker for the consumer-offsets wire artifact, "ICO2". Bumped with +/// the version when the dedup section was appended; the magic moves too so a +/// v1 artifact fails on the cheaper check. +pub(crate) const CONSUMER_OFFSETS_MAGIC: [u8; 4] = *b"ICO2"; /// Version byte following the magic. /// /// Any layout change bumps this, INCLUDING appended fields: the decoder /// deliberately fails closed on unknown versions and on trailing bytes, /// because a v2 field can change the meaning of fields v1 already read. -pub(crate) const CONSUMER_OFFSETS_VERSION: u8 = 1; +pub(crate) const CONSUMER_OFFSETS_VERSION: u8 = 2; /// Per-section entry ceiling for the consumer-offsets artifact. /// @@ -67,6 +69,9 @@ pub(crate) const CONSUMER_OFFSETS_VERSION: u8 = 1; /// entry ceiling. pub(crate) const CONSUMER_OFFSETS_ENTRIES_MAX: u32 = 1 << 20; +/// Wire stride of one dedup entry: client u128 + watermark u64 + commit u64. +const DEDUP_ENTRY_LEN: usize = size_of::() + 2 * size_of::(); + /// One in-flight partition state transfer on the receiving replica. /// /// Mirrors the metadata plane's session, plus `staged`: completed @@ -295,13 +300,19 @@ pub(crate) struct ConsumerOffsetsWire { pub consumers: Vec<(u32, u64)>, /// `(consumer group id, offset)`, ascending by id. pub groups: Vec<(u32, u64)>, + /// This group's dedup slice: `(client, watermark, latest_commit)`, + /// ascending by client. Carried so a replica rejoining behind the repair + /// floor can absorb a replay of what the group already committed instead + /// of re-executing it. + pub dedup: Vec<(u128, u64, u64)>, } impl ConsumerOffsetsWire { /// Encode: `magic | version u8 | purge_generation u64 | next_offset u64 | - /// consumer_count u32 | group_count u32 | {id u32, offset u64}xN | - /// {id u32, offset u64}xM | XxHash3_64 trailer`. Little-endian - /// throughout. + /// consumer_count u32 | group_count u32 | dedup_count u32 | + /// {id u32, offset u64}xN | {id u32, offset u64}xM | + /// {client u128, watermark u64, latest_commit u64}xD | XxHash3_64 + /// trailer`. Little-endian throughout. #[must_use] pub fn encode(&self) -> Vec { // Size exactly rather than guess; the reservation assert keeps the @@ -309,8 +320,9 @@ impl ConsumerOffsetsWire { let reserved = CONSUMER_OFFSETS_MAGIC.len() + size_of::() + 2 * size_of::() - + 2 * size_of::() + + 3 * size_of::() + (self.consumers.len() + self.groups.len()) * (size_of::() + size_of::()) + + self.dedup.len() * DEDUP_ENTRY_LEN + size_of::(); let mut out = Vec::with_capacity(reserved); out.extend_from_slice(&CONSUMER_OFFSETS_MAGIC); @@ -321,10 +333,17 @@ impl ConsumerOffsetsWire { out.extend_from_slice(&(self.consumers.len() as u32).to_le_bytes()); #[allow(clippy::cast_possible_truncation)] out.extend_from_slice(&(self.groups.len() as u32).to_le_bytes()); + #[allow(clippy::cast_possible_truncation)] + out.extend_from_slice(&(self.dedup.len() as u32).to_le_bytes()); for (id, offset) in self.consumers.iter().chain(self.groups.iter()) { out.extend_from_slice(&id.to_le_bytes()); out.extend_from_slice(&offset.to_le_bytes()); } + for (client, watermark, latest_commit) in &self.dedup { + out.extend_from_slice(&client.to_le_bytes()); + out.extend_from_slice(&watermark.to_le_bytes()); + out.extend_from_slice(&latest_commit.to_le_bytes()); + } debug_assert_eq!(out.len() + size_of::(), reserved, "encode reservation"); let trailer = state_artifact_checksum(&out); out.extend_from_slice(&trailer.to_le_bytes()); @@ -362,8 +381,10 @@ impl ConsumerOffsetsWire { let next_offset = cursor.u64()?; let consumer_count = cursor.u32()?; let group_count = cursor.u32()?; + let dedup_count = cursor.u32()?; let consumers = Self::decode_section(&mut cursor, "consumers", consumer_count)?; let groups = Self::decode_section(&mut cursor, "groups", group_count)?; + let dedup = Self::decode_dedup_section(&mut cursor, dedup_count)?; if !cursor.remaining().is_empty() { // Distinct from `Truncated`: extra bytes point at a NEWER // encoder, and telling the operator the artifact is short would @@ -377,9 +398,42 @@ impl ConsumerOffsetsWire { next_offset, consumers, groups, + dedup, }) } + /// Same guards as [`Self::decode_section`] at the dedup stride: peer count + /// against the ceiling, then against the bytes actually present, then + /// ascending-strict client order so the encoding stays canonical. + fn decode_dedup_section( + cursor: &mut LeCursor<'_>, + count: u32, + ) -> Result, ConsumerOffsetsWireError> { + if count > CONSUMER_OFFSETS_ENTRIES_MAX { + return Err(ConsumerOffsetsWireError::TooManyEntries { + section: "dedup", + count, + max: CONSUMER_OFFSETS_ENTRIES_MAX, + }); + } + if count as usize * DEDUP_ENTRY_LEN > cursor.remaining().len() { + return Err(ConsumerOffsetsWireError::Truncated); + } + let mut entries = Vec::with_capacity(count as usize); + let mut previous: Option = None; + for _ in 0..count { + let client = cursor.u128()?; + let watermark = cursor.u64()?; + let latest_commit = cursor.u64()?; + if previous.is_some_and(|previous| client <= previous) { + return Err(ConsumerOffsetsWireError::NonAscendingClient { client }); + } + previous = Some(client); + entries.push((client, watermark, latest_commit)); + } + Ok(entries) + } + fn decode_section( cursor: &mut LeCursor<'_>, section: &'static str, @@ -450,6 +504,11 @@ pub enum ConsumerOffsetsWireError { section: &'static str, id: u32, }, + /// Dedup clients are not strictly ascending. Same encoder bug as + /// [`Self::NonAscendingId`], on the u128-keyed section. + NonAscendingClient { + client: u128, + }, } impl From for ConsumerOffsetsWireError { @@ -490,6 +549,11 @@ impl fmt::Display for ConsumerOffsetsWireError { "consumer-offsets artifact {section} id {id} does not ascend \ (duplicate, or out of order)" ), + Self::NonAscendingClient { client } => write!( + f, + "consumer-offsets artifact dedup client {client} does not ascend \ + (duplicate, or out of order)" + ), } } } @@ -506,6 +570,7 @@ mod tests { next_offset: 43, consumers: vec![(1, 10), (7, 42)], groups: vec![(2, 5)], + dedup: vec![(11, 4, 90), (usize::MAX as u128 + 5, 9, 91)], } } @@ -525,6 +590,7 @@ mod tests { next_offset: 0, consumers: Vec::new(), groups: Vec::new(), + dedup: Vec::new(), }; let encoded = empty.encode(); assert_eq!( @@ -584,6 +650,69 @@ mod tests { ); } + #[test] + fn given_dedup_entries_when_encoded_should_round_trip() { + let wire = table(); + assert_eq!(ConsumerOffsetsWire::decode(&wire.encode()), Ok(wire)); + } + + #[test] + fn given_unordered_dedup_clients_when_decoded_should_reject() { + let unordered = ConsumerOffsetsWire { + purge_generation: 0, + next_offset: 0, + consumers: Vec::new(), + groups: Vec::new(), + dedup: vec![(9, 1, 1), (4, 2, 2)], + }; + assert_eq!( + ConsumerOffsetsWire::decode(&unordered.encode()), + Err(ConsumerOffsetsWireError::NonAscendingClient { client: 4 }) + ); + } + + #[test] + fn given_dedup_count_past_ceiling_when_decoded_should_reject_before_allocating() { + let mut bytes = Vec::new(); + bytes.extend_from_slice(&CONSUMER_OFFSETS_MAGIC); + bytes.push(CONSUMER_OFFSETS_VERSION); + bytes.extend_from_slice(&0u64.to_le_bytes()); + bytes.extend_from_slice(&0u64.to_le_bytes()); + bytes.extend_from_slice(&0u32.to_le_bytes()); + bytes.extend_from_slice(&0u32.to_le_bytes()); + bytes.extend_from_slice(&(CONSUMER_OFFSETS_ENTRIES_MAX + 1).to_le_bytes()); + let trailer = state_artifact_checksum(&bytes); + bytes.extend_from_slice(&trailer.to_le_bytes()); + assert_eq!( + ConsumerOffsetsWire::decode(&bytes), + Err(ConsumerOffsetsWireError::TooManyEntries { + section: "dedup", + count: CONSUMER_OFFSETS_ENTRIES_MAX + 1, + max: CONSUMER_OFFSETS_ENTRIES_MAX, + }) + ); + } + + #[test] + fn given_dedup_count_exceeding_bytes_when_decoded_should_reject_as_truncated() { + // Under the ceiling but past the bytes present: the stride guard is + // what stops a ~30 byte artifact reserving megabytes. + let mut bytes = Vec::new(); + bytes.extend_from_slice(&CONSUMER_OFFSETS_MAGIC); + bytes.push(CONSUMER_OFFSETS_VERSION); + bytes.extend_from_slice(&0u64.to_le_bytes()); + bytes.extend_from_slice(&0u64.to_le_bytes()); + bytes.extend_from_slice(&0u32.to_le_bytes()); + bytes.extend_from_slice(&0u32.to_le_bytes()); + bytes.extend_from_slice(&1_000u32.to_le_bytes()); + let trailer = state_artifact_checksum(&bytes); + bytes.extend_from_slice(&trailer.to_le_bytes()); + assert_eq!( + ConsumerOffsetsWire::decode(&bytes), + Err(ConsumerOffsetsWireError::Truncated) + ); + } + #[test] fn given_count_past_ceiling_when_decoded_should_reject_before_allocating() { let mut bytes = Vec::new(); @@ -593,6 +722,7 @@ mod tests { bytes.extend_from_slice(&0u64.to_le_bytes()); bytes.extend_from_slice(&(CONSUMER_OFFSETS_ENTRIES_MAX + 1).to_le_bytes()); bytes.extend_from_slice(&0u32.to_le_bytes()); + bytes.extend_from_slice(&0u32.to_le_bytes()); let trailer = state_artifact_checksum(&bytes); bytes.extend_from_slice(&trailer.to_le_bytes()); assert_eq!( @@ -612,6 +742,7 @@ mod tests { next_offset: 0, consumers: vec![(5, 1), (5, 2)], groups: Vec::new(), + dedup: Vec::new(), }; assert_eq!( ConsumerOffsetsWire::decode(&duplicate.encode()), @@ -625,6 +756,7 @@ mod tests { next_offset: 0, consumers: Vec::new(), groups: vec![(9, 1), (4, 2)], + dedup: Vec::new(), }; assert_eq!( ConsumerOffsetsWire::decode(&unordered.encode()), @@ -1704,11 +1836,13 @@ where // sealed segment while the counter stands at N, and the receiver // must resume minting at N either way. let next_offset = self.offset_frontier(); + let dedup = self.dedup().watermarks_sorted(); ConsumerOffsetsWire { purge_generation: self.applied_purge_generation, next_offset, consumers, groups, + dedup, } } @@ -2456,6 +2590,13 @@ where // file put a rejoin carrying thousands of consumers on the pump for // thousands of sequential open + write + optional fsync round trips; // the tick's superblock pre-pass sets the precedent for the width. + // The dedup slice is memory-only, so it installs here with the maps + // rather than being written anywhere. No frontier fence is needed: the + // install lifts `commit_min` to the offer's `commit_op`, so the commit + // walk that follows starts strictly above everything this artifact + // covers, and `record_commit` is idempotent besides. + self.dedup_mut() + .install_watermarks(offsets_wire.dedup.iter().copied()); let mut planned: Vec = Vec::with_capacity(offsets_wire.consumers.len() + offsets_wire.groups.len()); if let Some(dir) = self.consumer_offsets_path.clone() { @@ -2674,6 +2815,10 @@ where // promise rested on the caller clearing it first. self.segment_checksum_cache.borrow_mut().clear(); self.reuse_scan_memo.borrow_mut().take(); + // Degrade to at-least-once rather than keep watermarks that may now + // describe data this partition no longer holds: a stale entry would + // absorb a replay whose original was just unlinked. + self.dedup_mut().install_watermarks(std::iter::empty()); // Sweep EVERY segment file, not the in-memory count's worth: after // a late failure the renamed-in new chain is on disk while the diff --git a/core/server/config.toml b/core/server/config.toml index 2bac202b83..d663d397c7 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -942,6 +942,17 @@ clients_table_max = 8192 # u128 bitset, and this depth bounds that suffix. prepare_queue_depth = 32 +# Distinct clients each partition group tracks request watermarks for, so a +# retried produce or consumer-offset write is answered instead of committing a +# second time. At capacity the client whose newest commit is oldest is evicted; +# that client's next replay re-executes, exactly as it would have before dedup +# existed, so under-sizing degrades rather than breaks. Must be > 0 and <= 65536. +# +# Unlike [metadata] clients_table_max, this budget is PER GROUP, so worst-case +# memory scales with partition count. Size it to the producers one partition +# actually sees, not the node's client total. +dedup_clients_max = 4096 + # Entries the evicted ring retains per multi-replica partition for journal # repair after a peer rejoins. Larger widens the window a restarting peer can be # served from the ring before falling back to bulk sync, at the cost of pinned diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs index 587744f5dc..6dbcd8bbb5 100644 --- a/core/server/src/bootstrap.rs +++ b/core/server/src/bootstrap.rs @@ -2169,6 +2169,10 @@ const _: () = assert!( configs::partition::DEFAULT_PARTITION_PREPARE_QUEUE_DEPTH == consensus::PIPELINE_PREPARE_QUEUE_MAX ); +const _: () = assert!( + configs::partition::DEFAULT_PARTITION_DEDUP_CLIENTS_MAX + == consensus::PARTITION_DEDUP_CLIENTS_MAX +); const _: () = assert!(configs::metadata::DEFAULT_METADATA_CLIENTS_TABLE_MAX == consensus::CLIENTS_TABLE_MAX); const _: () = @@ -2665,6 +2669,7 @@ async fn load_partition( config.partition.evicted_ring_capacity, config.partition.evicted_ring_bytes_max.as_bytes_u64(), ); + partition.set_dedup_clients_max(config.partition.dedup_clients_max); partition.set_partition_dir(partition_dir.clone()); // Before the hydrate: the durable record is keyed by incarnation, so a // `purge.gen` left behind by a previous life of this namespace reads 0. diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs index 5dab8e2fd5..879009063b 100644 --- a/core/server/src/dispatch.rs +++ b/core/server/src/dispatch.rs @@ -438,9 +438,9 @@ fn build_auto_commit_request( operation: Operation::StoreConsumerOffset, size, client: AUTO_COMMIT_CLIENT_ID, - // The partition plane is sessionless (no `ClientTable` dedup); a - // nonzero session + request just satisfy the wire header - // validation. + // The reserved sentinel client is never deduped and never + // replied to; a nonzero session + request just satisfy the wire + // header validation. session: 1, request: 1, group: namespace.inner(), @@ -1161,6 +1161,7 @@ async fn handle_client_request( bound_session, transport_client_id, acting_user_id, + PartitionReplyMode::Awaited, ) .await; return; @@ -1449,13 +1450,26 @@ async fn handle_get_me( .await; } +/// Whether a partition write waits for its committed reply. +/// +/// `Awaited` attaches an in-process channel to the request, so the committed +/// reply comes back to this shard instead of the bus and is written to the +/// caller's socket here. `FireAndForget` attaches none: the reply takes the bus +/// path and is shed when nothing is listening, which is what `?ack=none` asks +/// for. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum PartitionReplyMode { + Awaited, + FireAndForget, +} + /// Route a partition data-plane op (`SendMessages` / consumer-offset writes) /// through the shard mesh by namespace: the op belongs to the partition's /// own consensus group, not the metadata group. The owning shard's -/// partitions plane runs at-least-once consensus and replies directly via -/// `send_to_client`. `header.client` therefore stays the TRANSPORT id -/// (home-shard routing bits), not the VSR session id -- partition ops are -/// sessionless ("session lifecycle is metadata-only"). +/// partitions plane dedups the request against its group's slice and answers +/// over the submit channel, so `header.client` carries the VSR consensus id +/// (the dedup key) rather than the transport id -- replies cannot be routed by +/// it and come back here instead. /// /// Callers must have authenticated the transport already: `vsr_client_id` / /// `bound_session` come from its bound VSR session. Every failure before @@ -1474,6 +1488,7 @@ pub(crate) async fn dispatch_partition_request( bound_session: u64, transport_client_id: u128, acting_user_id: Option, + reply_mode: PartitionReplyMode, ) where B: ShellBus, MJ: JournalHandle + 'static, @@ -1593,16 +1608,67 @@ pub(crate) async fn dispatch_partition_request( let request = request.transmute_header(|header, new_header: &mut RoutedRequestHeader| { *new_header = header; new_header.group = namespace; - new_header.client = transport_client_id; - // 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. + // The VSR consensus id, exactly as metadata ops are stamped: it is the + // dedup key every replica keys its slice by, and unlike the transport + // id it stays valid across nodes. Replies therefore cannot be routed + // by this field -- they ride the submit's channel back to this shard, + // which owns the socket. + new_header.client = vsr_client_id; new_header.session = bound_session; + // Header validation requires `request > 0` for non-register ops; a + // client that does not number its data-plane ops normalizes to 1 and + // is simply never deduplicated. new_header.request = new_header.request.max(1); }); - shard.dispatch(request.into_generic()); + if reply_mode == PartitionReplyMode::FireAndForget { + shard.dispatch(request.into_generic()); + return; + } + relay_partition_reply( + shard, + IggyNamespace::from_raw(namespace), + request, + transport_client_id, + header.operation, + ) + .await; +} + +/// Await a partition write's committed reply and write it to the socket this +/// shard holds. The reply cannot be routed by `RoutedRequestHeader.client` -- +/// that field carries the VSR consensus id, whose bits encode no home shard. +#[allow(clippy::future_not_send)] +async fn relay_partition_reply( + shard: &Rc>, + namespace: IggyNamespace, + request: Message, + transport_client_id: u128, + operation: Operation, +) where + B: ShellBus, + MJ: JournalHandle + 'static, + MJ::Target: Journal, Header = PrepareHeader>, + S: 'static, + SB: SuperblockStore + 'static, +{ + let Some(reply) = shard.partition_submit(namespace, request).await else { + // Unroutable, shed, or budget expiry. Deliberately silent: the outcome + // is unknown, and a synthesized failure could contradict a write that + // commits moments later. The client's read-timeout is the recovery. + return; + }; + if let Err(error) = shard + .bus + .send_to_client(transport_client_id, reply.into_frozen()) + .await + { + warn!( + transport_client_id, + operation = ?operation, + error = %error, + "failed to forward committed partition reply to its socket" + ); + } } #[allow(clippy::future_not_send, clippy::too_many_lines)] @@ -4371,6 +4437,7 @@ mod tests { SESSION, TRANSPORT, Some(DEFAULT_ROOT_USER_ID), + PartitionReplyMode::Awaited, ) .await; diff --git a/core/server/src/http/submit.rs b/core/server/src/http/submit.rs index ca763d8368..bd939a9903 100644 --- a/core/server/src/http/submit.rs +++ b/core/server/src/http/submit.rs @@ -35,8 +35,8 @@ use tracing::warn; use crate::bootstrap::ServerShard; use crate::dispatch::{ - dispatch_partition_request, resolve_delete_segments_truncate, submit_client_request_on_owner, - submit_logout_on_owner, + PartitionReplyMode, dispatch_partition_request, resolve_delete_segments_truncate, + submit_client_request_on_owner, submit_logout_on_owner, }; use crate::http::admission::admit_partition_write; use crate::http::error::{PartitionWriteError, WriteError}; @@ -407,6 +407,7 @@ pub(in crate::http) async fn partition_write_replicated( session.session, session.client_id, Some(session.user_id), + PartitionReplyMode::Awaited, ) .await; let outcome = compio::time::timeout(PARTITION_WRITE_REPLY_TIMEOUT, receiver).await; @@ -456,6 +457,7 @@ pub(in crate::http) async fn produce_unacked( session.session, session.client_id, Some(session.user_id), + PartitionReplyMode::FireAndForget, ) .await; Ok(()) diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index 5a2f49fd21..a9cb63fa7e 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -630,6 +630,7 @@ pub async fn build_partition_fresh( config.partition.evicted_ring_capacity, config.partition.evicted_ring_bytes_max.as_bytes_u64(), ); + partition.set_dedup_clients_max(config.partition.dedup_clients_max); partition.set_partition_dir(partition_dir); // Fresh dirs read generation 0; a dir surviving from a crashed process // (this "fresh" build races repair re-materialization) reads the last diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs index 1efaa598a3..31827d7eb8 100644 --- a/core/shard/src/lib.rs +++ b/core/shard/src/lib.rs @@ -379,6 +379,12 @@ pub type PartitionReadHandler = /// deadline. const PARTITION_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +/// Budget for a partition write's wait on its committed reply. Longer than a +/// read: the wait spans replication quorum plus any park-and-promote the +/// request rides through, and a view change mid-flight re-proposes under the +/// new primary. Expiry leaves the client to its own read-timeout. +const PARTITION_SUBMIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + /// Race `future` against a bus timer. /// /// `Some` if it finishes within `budget`, `None` if the timer fires first. @@ -683,6 +689,17 @@ pub enum LifecycleFrame { read: PartitionRead, reply: Sender, }, + /// Admit a partition write (`SendMessages` / consumer-offset write) on + /// the shard owning its namespace, carrying the channel its committed + /// reply travels back on. The partition plane cannot route a reply by + /// `header.client` -- that field is the VSR consensus id, whose bits + /// carry no home-shard routing -- so the reply returns to the + /// connection-owning shard, which writes it to the socket it holds. + /// See [`IggyShard::partition_submit`]. + PartitionSubmit { + request: Message, + reply: Sender>>, + }, /// Shard 0 broadcasts after a partition-shaped metadata commit; wakes /// the per-shard reconciler. No payload: reconciler re-reads target /// state. Drops covered by the periodic safety tick. @@ -1838,6 +1855,69 @@ where } } + /// Admit a partition write on the shard owning `namespace` and await its + /// committed reply. Routes through the shards table exactly like + /// [`Self::partition_read`], self-sends included, so a locally-owned + /// partition takes the same path. + /// + /// `None` = unroutable namespace, full owning-shard inbox, dropped reply + /// sender (view-change reset, park eviction), or budget expiry. The caller + /// stays silent on `None`: the client's own response read-timeout is the + /// recovery, and a synthesized failure reply could contradict a write that + /// commits moments later. + #[allow(clippy::future_not_send)] + pub async fn partition_submit( + &self, + namespace: IggyNamespace, + request: Message, + ) -> Option> { + let target = self.shards_table.shard_for(namespace).unwrap_or_else(|| { + // Same fallback as `route_typed`: a miss means "not seeded yet", + // not "unroutable", and the owning shard parks what arrives early. + crate::shards_table::calculate_shard_from_consensus_ns( + namespace.inner(), + self.shard_count, + ) + }); + let (reply_tx, reply_rx) = channel::>>(1); + let frame = ShardFrame::lifecycle(LifecycleFrame::PartitionSubmit { + request, + reply: reply_tx, + }); + let sender = self.senders.get(target as usize)?; + if let Err(error) = sender.try_send(frame) { + self.metrics.record_frame_drop( + crate::metrics::frame_drop_variant::PARTITION, + crate::coordinator::classify_try_send_err(&error), + ); + tracing::warn!( + shard = self.id, + target, + "partition_submit: inbox rejected PartitionSubmit frame: {error:?}" + ); + return None; + } + match bus_timeout(&self.bus, PARTITION_SUBMIT_TIMEOUT, reply_rx.recv()).await { + Some(Ok(reply)) => reply, + Some(Err(_)) => { + tracing::debug!( + shard = self.id, + target, + "partition_submit: reply sender dropped before commit" + ); + None + } + None => { + tracing::warn!( + shard = self.id, + target, + "partition_submit: owning shard did not reply within budget" + ); + None + } + } + } + /// Return a clone of the shard-0 coordinator handle, if attached. /// Bootstrap uses this to wire the listener accept callbacks /// (replica + client) to coordinator-driven fd-delegation instead @@ -2399,6 +2479,12 @@ struct ParkedFrame { /// the outcome from a reply rather than a timeout. passes: u32, message: Message, + /// Channel the committed reply travels back on, for a frame that arrived + /// as a [`LifecycleFrame::PartitionSubmit`]. `None` for replicated + /// prepares and for writes admitted without a waiter. Dropping the frame + /// (expiry, teardown, shutdown) drops this, which wakes the awaiting + /// dispatch with a receive error it maps to silence. + reply: Option>>>, } impl ParkedFrame { @@ -2552,7 +2638,7 @@ where let header = request.header(); (header.operation, header.group) }; - match self.park_if_unmaterialised(request, routing.0, routing.1) { + match self.park_if_unmaterialised(request, routing.0, routing.1, &mut None) { // The incarnation fence runs only here, on client traffic. // A backup denying what the primary admitted would diverge // the replicas, so replicated frames are never fenced. @@ -2583,7 +2669,7 @@ where // A tombstoned prepare still flows to the plane: replicated // traffic has no client awaiting a reply on this node, and // the plane's own tombstone guard drops it. - match self.park_if_unmaterialised(prepare, routing.0, routing.1) { + match self.park_if_unmaterialised(prepare, routing.0, routing.1, &mut None) { ParkOutcome::Deliver(prepare) | ParkOutcome::Tombstoned(prepare) => { self.on_replicate(prepare).await; // A follower learns the cluster commit point from the @@ -2932,9 +3018,10 @@ where }; let mut refused_frames: Vec = Vec::new(); let mut remaining = servable.into_iter(); - while let Some(frame) = remaining.next() { + while let Some(mut frame) = remaining.next() { let passes = frame.passes; let parked_epoch = frame.epoch; + let waiter = frame.reply.take(); // Parked frames are stored generic (the buffer holds every variant // in one Vec), so re-entering the pump costs one classify. That is // the rare path -- a post-`CreateTopic` convergence window, not the @@ -2955,20 +3042,32 @@ where continue; } }; - let Err(error) = sender.try_send(ShardFrame::consensus(self.id, bag)) else { + let Some(outgoing) = self.reparked_frame(namespace, waiter, bag) else { + continue; + }; + let Err(error) = sender.try_send(outgoing) else { continue; }; let (refused, disconnected) = match error { TrySendError::Full(frame) => (frame, false), TrySendError::Disconnected(frame) => (frame, true), }; - let ShardFrame::Consensus { message, .. } = refused else { - unreachable!("try_send returns the frame it was handed"); - }; - let refused_frame = ParkedFrame { - epoch: parked_epoch, - passes, - message: message.into_generic(), + let refused_frame = match refused { + ShardFrame::Consensus { message, .. } => ParkedFrame { + epoch: parked_epoch, + passes, + message: message.into_generic(), + reply: None, + }, + ShardFrame::Lifecycle(LifecycleFrame::PartitionSubmit { request, reply }) => { + ParkedFrame { + epoch: parked_epoch, + passes, + message: request.into_generic(), + reply: Some(reply), + } + } + _ => unreachable!("try_send returns the frame it was handed"), }; if disconnected { // Pump gone: re-parking holds the frame until process exit, and @@ -3178,6 +3277,7 @@ where message: Message, operation: Operation, namespace_raw: u64, + reply: &mut Option>>>, ) -> ParkOutcome where H: iggy_binary_protocol::ConsensusHeader, @@ -3301,6 +3401,7 @@ where epoch, passes: 0, message: message.into_generic(), + reply: reply.take(), }); drop(pending); self.parked_partition_bytes @@ -3383,6 +3484,105 @@ where true } + /// Admit a `PartitionSubmit`: same gates as the [`MessageBag::Request`] + /// arm, but every refusal answers on `reply` instead of the bus, and the + /// admitted request carries an in-process reply channel down to the + /// pipeline entry so its committed reply comes back here rather than + /// being routed by `header.client`. + #[allow(clippy::future_not_send)] + pub async fn on_partition_submit( + &self, + request: Message, + reply: Sender>>, + ) where + B: MessageBus + 'static, + MJ: JournalHandle, + ::Target: + Journal, Header = PrepareHeader>, + M: RestorableMetadataStm, + T: ShardsTable, + { + let routing = { + let header = request.header(); + (header.operation, header.group) + }; + let mut parked_reply = Some(reply); + match self.park_if_unmaterialised(request, routing.0, routing.1, &mut parked_reply) { + ParkOutcome::Deliver(request) + if !self.serves_committed_incarnation(routing.0, routing.1) => + { + Self::answer_partition_submit_transient(request.header(), parked_reply); + } + ParkOutcome::Deliver(request) => { + let (sender, receiver) = consensus::oneshot_channel(); + self.plane + .partitions() + .on_request_with_reply(request, Some(sender)) + .await; + // Await OFF the pump: the commit that fires this receiver needs + // the pump to keep draining acks, so blocking here would + // deadlock the very reply being waited on. The task holds only + // owned channel halves, never a partitions borrow. + let Some(reply) = parked_reply else { return }; + // Through the bus, not the runtime directly: the simulator + // supplies its own executor and virtual clock. + self.bus.spawn(async move { + let committed = receiver.await.ok().map(Message::into_generic); + let _ = reply.try_send(committed); + }); + } + ParkOutcome::Tombstoned(request) | ParkOutcome::Overflow(request) => { + Self::answer_partition_submit_transient(request.header(), parked_reply); + } + // Sender moved into the parked frame; it answers on drain or wakes + // the awaiter with a receive error when the frame expires. + ParkOutcome::Parked => {} + } + } + + /// Wrap a drained parked frame for its trip back through this shard's + /// inbox. A submit's waiter cannot ride the generic bag handoff, so it + /// re-enters through the lifecycle frame it arrived on and keeps its reply + /// channel attached. `None` = the frame is unusable and was dropped. + fn reparked_frame( + &self, + namespace: IggyNamespace, + waiter: Option>>>, + bag: MessageBag, + ) -> Option { + match (waiter, bag) { + (Some(reply), MessageBag::Request(request)) => { + Some(ShardFrame::lifecycle(LifecycleFrame::PartitionSubmit { + request, + reply, + })) + } + (Some(_), _) => { + tracing::error!( + shard = self.id, + namespace_raw = namespace.inner(), + "parked frame carries a reply channel but is not a client request; dropping it" + ); + None + } + (None, bag) => Some(ShardFrame::consensus(self.id, bag)), + } + } + + /// Answer a refused `PartitionSubmit` with the same transient deny the bus + /// path sends, over the submit's own channel. + fn answer_partition_submit_transient( + request_header: &RoutedRequestHeader, + reply: Option>>>, + ) { + let Some(reply) = reply else { return }; + let deny = build_deny_reply_from_request_header( + request_header, + IggyError::TransientNotAccepted.as_code(), + ); + let _ = reply.try_send(Some(deny.into_generic())); + } + #[allow(clippy::future_not_send)] pub async fn on_request(&self, request: Message) where diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs index 4340761a00..30fa7937c4 100644 --- a/core/shard/src/router.rs +++ b/core/shard/src/router.rs @@ -414,6 +414,10 @@ where async fn process_lifecycle(&self, payload: LifecycleFrame) where B: MessageBus + 'static, + MJ: JournalHandle, + ::Target: + Journal, Header = PrepareHeader>, + M: RestorableMetadataStm, { match payload { LifecycleFrame::ReplicaInboundSetup { fd, slot } => { @@ -544,6 +548,14 @@ where // times out. (self.on_partition_read)(namespace, read, reply); } + LifecycleFrame::PartitionSubmit { request, reply } => { + // Addressed to the shard owning the request's namespace (the + // sender resolved it via the shards table, same fallback as + // `route_typed`). Every refusal answers on `reply`, so the + // awaiting shard never waits out its budget on a decision + // already made. + self.on_partition_submit(request, reply).await; + } LifecycleFrame::MetadataCommitTick => { // Reconciler may not yet be wired (e.g. mid-bootstrap, or // single-shard tests that never enable the reconciler loop). diff --git a/core/simulator/src/client.rs b/core/simulator/src/client.rs index 3e7742ed1e..0ee63a844b 100644 --- a/core/simulator/src/client.rs +++ b/core/simulator/src/client.rs @@ -64,9 +64,9 @@ pub struct SimClient { /// sequence the server's `ClientTable` dedups and requires gap-free. 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 + /// range ([`PARTITION_ID_BASE`]), so a partition id never collides with a + /// metadata id even under reply duplication. Monotone per client, which is + /// what the per-group dedup watermark requires. See /// [`SimClient::request_id_for`]. partition_counter: Cell, /// Deterministic per-message id source for produced messages. The real SDK @@ -123,9 +123,10 @@ impl SimClient { /// Metadata/replicated ops advance a contiguous `1, 2, 3, …` counter: the /// `ClientTable` dedups them and rejects anything but `committed + 1`, so a /// gap opens a permanent `RequestGap` and wedges the client's metadata - /// plane. 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 + /// plane. Partition ops are deduped against a per-group watermark, which + /// accepts gaps, so they draw from a separate counter offset into a + /// disjoint range ([`PARTITION_ID_BASE`]) -- still strictly monotone per + /// client, which is all the watermark needs. 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 diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs index 8e971c7083..0d95b4c18b 100644 --- a/core/simulator/src/lib.rs +++ b/core/simulator/src/lib.rs @@ -1475,12 +1475,14 @@ mod tests { ); } - /// At-least-once failover: `SendMessages` retry on a new primary - /// re-executes. Retry reply carries a HIGHER `commit` op (re-execution - /// proof, not dedup). Duplicate payload lives at two offsets; consumers - /// dedup if they need at-most-once-per-payload. + /// Failover retry absorbed by the partition dedup slice: a `SendMessages` + /// replay of an already-committed `(client, request)` on a NEW primary is + /// answered without re-executing. The slice is folded in on every replica + /// at commit, so the promoted primary knows the watermark its predecessor + /// established -- that inheritance is what this test proves. #[test] - fn failover_retry_re_executes_under_at_least_once() { + #[allow(clippy::too_many_lines)] + fn failover_retry_absorbed_by_partition_dedup() { server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings { enabled: false, size: iggy_common::IggyByteSize::from(0u64), @@ -1523,6 +1525,15 @@ mod tests { } let original_reply = original_reply.expect("commit reply must arrive before primary crash"); let original_commit_op = original_reply.header().commit; + // Offset after exactly one committed batch: the duplicate must not + // move it. + let offset_after_original = sim.replicas[1].shards[0] + .plane + .partitions() + .get_by_ns(&ns) + .expect("partition must exist on a live replica") + .stats + .current_offset(); assert_eq!( original_reply.header().request, original_request_id, @@ -1556,7 +1567,8 @@ mod tests { "new primary must not be the crashed replica" ); - // Replay SAME request to new primary. No dedup -> re-execution. + // Replay the SAME request to the new primary: the dedup slice it + // inherited at commit must absorb it. sim.submit_request(client_id, new_primary_idx, replay_req.into_generic()); let mut retry_reply: Option> = None; @@ -1567,28 +1579,43 @@ mod tests { break; } } - let retry_reply = retry_reply.expect( - "reply must arrive after retry; new primary re-commits as \ - fresh prepare (at-least-once)", - ); + let retry_reply = retry_reply + .expect("reply must arrive after retry; the new primary absorbs it as a duplicate"); - // At-least-once: same request id (correlation), HIGHER commit op - // (re-execution). No dedup absorbs the retry. assert_eq!( retry_reply.header().request, original_request_id, "retry's reply must correlate to the request id" ); - assert!( - retry_reply.header().commit > original_commit_op, - "retry must re-execute (commit op > original={original_commit_op}, got {})", - retry_reply.header().commit - ); assert_eq!( retry_reply.header().client, client_id, "retry must echo original client_id" ); + assert_eq!( + retry_reply.header().status, + 0, + "an absorbed duplicate is a success, not an error" + ); + // The absorbed answer is synthesized at admission, so it never earns a + // new op. Re-execution would have committed past the original. + assert!( + retry_reply.header().op <= original_commit_op, + "retry must NOT re-execute (original commit={original_commit_op}, reply op={})", + retry_reply.header().op + ); + // The payload committed exactly once. + let committed = sim.replicas[usize::from(new_primary_idx)].shards[0] + .plane + .partitions() + .get_by_ns(&ns) + .expect("partition must exist on the new primary") + .stats + .current_offset(); + assert_eq!( + committed, offset_after_original, + "duplicate must not append a second copy" + ); } /// Regression: a behind backup (`commit_min < commit_max`) becoming From 5949b88496fcb060bf1b30f749d2da3198f955db Mon Sep 17 00:00:00 2001 From: Grzegorz Koszyk Date: Mon, 24 Aug 2026 21:01:10 +0200 Subject: [PATCH 3/3] fix CI --- foreign/node/src/wire/vsr/index.ts | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/foreign/node/src/wire/vsr/index.ts b/foreign/node/src/wire/vsr/index.ts index 6dcffe2dc7..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,