numinnex commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4260669737
##########
core/server/src/dispatch/mod.rs:
##########
@@ -557,25 +688,48 @@ async fn handle_client_request<B, MJ, S, SB>(
// (`submit_register_local_or_forward`). Every other non-replicated
// code MUST go through Register first, which binds the acting user
// the per-op authz gates resolve.
- handle_non_replicated_request(
- shard,
- sessions,
- server_config,
- transport_client_id,
- request,
- ConnectionContext {
- bound,
- user_id,
- address: client_address,
- metadata_watermark,
- },
- )
- .await;
+ let context = ConnectionContext {
+ bound,
+ user_id,
+ address: client_address,
+ metadata_watermark,
+ };
+ // `ClientDrain` says which frames stay on the drain.
+ if matches!(non_replicated_code(header), PING_CODE |
BIND_SESSION_CODE) {
+ handle_non_replicated_request(
+ shard,
+ sessions,
+ server_config,
+ transport_client_id,
+ request,
+ context,
+ )
+ .await;
+ return;
+ }
+ let in_flight = drain.admit(Admission::Cap).await;
+ let task_shard = Rc::clone(shard);
+ let task_sessions = Rc::clone(sessions);
+ let task_config = Arc::clone(server_config);
+ shard.bus.spawn(async move {
Review Comment:
**warning**: every non-PING read is spawned per frame, and every in-tree SDK
is lockstep, so each poll pays a new drain + task. Measured vs master:
standalone poll -14..-17%, consumer under concurrent produce -24..-48%. It also
lets two pipelined `Next`/auto-commit disk-tier polls on one connection plan
the same offset and deliver twice (and a promoted poll can overwrite a later
StoreConsumerOffset).
Suggest: take `InFlight` first, run the read inline and promote it to a task
only when another frame arrives (PING still overtakes a held read); a one-slot
poll lane keyed by the resolved (namespace, consumer), released after the
reply, that offset store/delete also wait on.
##########
core/metadata/src/impls/metadata.rs:
##########
@@ -3107,20 +3099,22 @@ where
.as_ref()
.expect("submit_request_in_process: consensus only exists on shard
0");
- // Not-primary is transient: the same request replayed against the
- // current primary commits fine. Reply with the explicit transient
- // frame (relayed to the socket by the home shard) so the client
- // replays immediately rather than waiting out its read-timeout.
- // `TransientNotAccepted` specifically: the request never entered the
- // pipeline here, so the client may re-issue it ANYWHERE -- including
- // under a fresh session after failing over to the current leader --
- // without double-apply risk. (`TransientNotCommitted` conversely means
- // the outcome is unknown and only a same-session replay is safe.)
+ // Reply before admission so the client can retry without a read
timeout.
+ // This refusal covers only this attempt; earlier copies may still
commit.
+ // Retrying with their session and request id preserves deduplication.
//
// No catch-up gate: a non-register op admitted mid-commit-window
// simply pipelines behind the in-flight batch (the wire path has
// always done this); the register-specific invariant is guarded in
// `submit_register_in_process` / `register_preflight`.
+ if consensus.is_normal_backup() {
+ return Ok(build_result_rejection_reply(
Review Comment:
**critical**: backups now answer `NotPrimary` (89) here and in
`iggy_partition.rs:6755` where they answered 58. No foreign SDK knows 89, and
their retry / roster-walk predicates only match 58 (node `client.socket.ts:59`,
go `tcp_core.go:746`, java `AsyncTcpConnection`, C# `VsrConnection.cs:150`), so
in a cluster these requests fail terminally. Single-node e2e won't catch it.
Please add 89 everywhere and treat it exactly like 58, keeping the rule that
a refusal after an uncertain attempt doesn't prove non-commit.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -11670,12 +11919,11 @@ where
fn request_history_available(&self, header: &RoutedRequestHeader) -> bool {
header.operation == Operation::RetireSession
|| header.operation.is_partition_lifecycle()
- || (header.partition_incarnation == self.created_revision &&
!self.history_deleted())
+ || (header.partition_incarnation == self.created_revision() &&
!self.history_deleted())
}
- fn prepare_history_available(&self, header: &PrepareHeader) -> bool {
- header.operation == Operation::RetireSession
- || (header.partition_incarnation == self.created_revision &&
!self.history_deleted())
+ const fn prepare_history_available(&self, header: &PrepareHeader) -> bool {
Review Comment:
**warning**: the `RetireSession` exemption was dropped here but kept in
`request_history_available` above, so the primary prepares a retirement that
every backup with the delete fence applied discards (`on_replicate`,
`apply_repaired_prepare`). No quorum, the op sits at the pipeline head and the
liveness sweep never completes; if a lagging backup acks, the third replica
can't repair it.
Suggest `incarnation == created_revision() && (op == RetireSession ||
!history_deleted())`, fix the stale doc above, and let `SessionRetired`
(`shard/src/poll.rs:253`) answer true once the fence applied, carrying
`created_revision` so a recreated incarnation never answers for the old one.
##########
core/binary_protocol/src/responses/system/get_cluster_metadata.rs:
##########
@@ -60,6 +64,7 @@ impl WireEncode for ClusterNodeResponse {
#[allow(clippy::cast_possible_truncation)]
fn encode(&self, buf: &mut BytesMut) {
+ buf.put_u8(self.replica_id);
Review Comment:
**critical**: the leading `replica_id` byte shifts every node for old
decoders (node `cluster.utils.ts:56`, go `cluster_node.go:91`, java
`BytesDeserializer.java:560`, C# `BinaryMapper.cs:1428`), breaking
GetClusterMetadata and the poll/offset routing replies that embed a node. Nodes
have no length prefix, so there is no compatible server-side form.
This also turns bdd-go/java/node/csharp red (they depend on rust-server and
call GetClusterMetadata). Please update the decoders, plus `replicaId: 0` in
node `tcp.cluster.e2e.ts` expectedMeta.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -659,6 +819,7 @@ async fn handle_client_request<B, MJ, S, SB>(
// stamps the consensus client / session / group over the header,
// and a pre-consensus deny still has to echo what the client sent.
let header = *header;
+ let in_flight = drain.admit(Admission::MetadataLane).await;
Review Comment:
**warning**: the metadata write commits in a spawned task, and nothing holds
back the frames behind it on the same connection. A pipelined CreateTopic then
SendMessages resolves the namespace from the local STM before the topic commits
and gets a permanent `TopicIdNotFound`; a write then GetTopic misses the write
(the read's watermark is captured before this one is recorded at :886).
Reachable today via the Java async client.
Suggest parking every later non-PING frame (before it takes a slot) while
the metadata lane is held.
##########
core/server/src/dispatch/host.rs:
##########
@@ -19,27 +19,28 @@
//! mode build a shard on.
//!
//! One [`ServerHost`] per shard owns everything the request plane shares
-//! across its transports: the per-client FIFO queues and their drain slots,
-//! the [`SessionManager`], and the bus connection-lost hook that logs a
-//! dropped connection out. All connections installed on the shard dispatch
-//! through the same instance, which preserves ordering and disconnect
-//! cleanup across transports: the destination shard installs it on delegated
-//! TCP/WS/TCP-TLS/WSS connections, and shard 0 also on its local QUIC
-//! connections.
+//! across its transports: the per-client FIFO queues and their drains, the
+//! [`SessionManager`], and the bus connection-lost hook that logs a dropped
+//! connection out. All connections installed on the shard dispatch through
+//! the same instance, which keeps disconnect cleanup equal across transports:
+//! the destination shard installs it on delegated TCP/WS/TCP-TLS/WSS
+//! connections, and shard 0 also on its local QUIC connections. A drain runs
+//! the frames that later frames depend on in arrival order, and every other
Review Comment:
nit: the commit wait of a metadata write is not in the set of frames the
drain orders (see :822 in `mod.rs`), so "the frames that later frames depend
on" overstates the contract. Please state the actual ordering guarantee here
and in `mod.rs:256-262`.
##########
core/server/src/http/handlers.rs:
##########
@@ -1314,12 +1313,28 @@ pub(in crate::http) async fn poll_messages(
.streams()
.poll_metadata(namespace, None, HTTP_READ_CLIENT_ID)
.ok_or(ReadError::Rejected(IggyError::TransientNotAccepted))?;
- let context =
metadata.context(state.shard.plane.metadata().applied_frontier().get());
- if let Some(requested) = query.strategy.context {
- // TCP parity: a group poll already got the re-sync sentinel at resolve
- // above, so this poll has no group and names group 0, as TCP does.
- fence_poll_context(requested, context, 0,
partition_id).map_err(ReadError::Rejected)?;
+ let local =
metadata.context(state.shard.plane.metadata().applied_frontier().get());
+ let requested = request
+ .route()
+ .map(|route| route.context)
+ .or(query.strategy.context)
+ .unwrap_or(local);
+ let effect = if query.auto_commit {
+ PartitionEffect::Write
+ } else {
+ PartitionEffect::Read
+ };
+ if let Some(response) = SendWrapper::new(route_partition(
Review Comment:
**warning**: every HTTP partition request (here and :1437, :1558, :1657,
:1723) first does a cross-shard `PartitionRead::Primary` probe via
`route_partition`, even on the primary and on the hop target; plain GETs used
to run locally. HTTP produce benched about -10%.
Suggest running locally first with a primary-gated read, and relaying only
on a refusal that proves non-admission with deadline left. A missing reply or
spent deadline must be the final answer (never re-run), and the session-gate
timeout should get its own error rather than look like "not primary".
##########
core/server/src/http/handlers.rs:
##########
@@ -1314,12 +1313,28 @@ pub(in crate::http) async fn poll_messages(
.streams()
.poll_metadata(namespace, None, HTTP_READ_CLIENT_ID)
.ok_or(ReadError::Rejected(IggyError::TransientNotAccepted))?;
- let context =
metadata.context(state.shard.plane.metadata().applied_frontier().get());
- if let Some(requested) = query.strategy.context {
- // TCP parity: a group poll already got the re-sync sentinel at resolve
- // above, so this poll has no group and names group 0, as TCP does.
- fence_poll_context(requested, context, 0,
partition_id).map_err(ReadError::Rejected)?;
+ let local =
metadata.context(state.shard.plane.metadata().applied_frontier().get());
+ let requested = request
+ .route()
+ .map(|route| route.context)
+ .or(query.strategy.context)
+ .unwrap_or(local);
+ let effect = if query.auto_commit {
+ PartitionEffect::Write
+ } else {
+ PartitionEffect::Read
+ };
+ if let Some(response) = SendWrapper::new(route_partition(
+ &state, &request, namespace, requested, effect,
+ ))
+ .await
+ {
+ return Ok(response);
}
+ // `poll_wire_request` refused a group poll above, so this poll has no
+ // member and names client 0, the HTTP read client.
+ fence_poll_context(requested, local, HTTP_READ_CLIENT_ID, partition_id)
+ .map_err(ReadError::Rejected)?;
let reply = SendWrapper::new(state.shard.partition_read(
namespace,
PartitionRead::Poll {
Review Comment:
nit: `PartitionRead::Poll` never refuses on a backup, and the primary check
in `route_partition` is a separate cross-shard message, so a view change in
between serves this on a backup (stale stored offset for `Next`; auto-commit
only refused later in `complete_poll`). Small window today, but a local-first
change would make it the main path. An OnPrimary-gated read variant closes it.
##########
core/server/src/http/forward.rs:
##########
@@ -241,19 +361,108 @@ pub(in crate::http) async fn forward_to_primary(
SendWrapper::new(forward_or_pass(state, request, next)).await
}
-/// Route-layer fallback for acknowledged partition writes over HTTP.
-///
-/// HTTP has no persistent leader-aware connection to retarget. Execute on the
-/// contacted node first, then retry across the configured HTTP nodes within
-/// the deadline after a typed `TransientNotAccepted` denial. That
-/// denial proves the write never entered a partition pipeline. Every ambiguous
-/// outcome is returned without replay.
-pub(in crate::http) async fn forward_partition_write(
+/// Route layer of the partition routes: buffer the request once and attach it
+/// as a [`PartitionRequest`], so its handler can relay it to the partition
+/// primary with [`route_partition`]. No attempt runs here.
+pub(in crate::http) async fn forward_partition_request(
State(state): State<HttpState>,
request: Request,
next: Next,
) -> Response {
- SendWrapper::new(forward_partition_or_pass(state, request, next)).await
+ SendWrapper::new(attach_partition_request(state, request, next)).await
+}
+
+/// Decide where a partition request runs once its partition and context are
+/// fixed. `None` serves it on this node. `Some` is the final answer: the reply
+/// of the partition primary, or a refusal that proves nothing was admitted.
+///
+/// Until the deadline, every hop waits for a view with a primary, because a
+/// new partition starts on each replica at its own pace. A hop target never
+/// relays: it refuses with `NotPrimary` when another replica is the primary,
+/// so a stale view never chains hops. The contacted node starts another turn
+/// only after such a proven non-admission, a failed connect, or a read that
+/// failed after its send.
+pub(in crate::http) async fn route_partition(
+ state: &HttpInner,
+ request: &PartitionRequest,
+ namespace: IggyNamespace,
+ context: PartitionContext,
+ effect: PartitionEffect,
+) -> Option<Response> {
+ if request.hop == PartitionHop::Unrouted {
+ return None;
+ }
+ let Ok(partition_id) = u32::try_from(namespace.partition_id()) else {
+ return
Some(CustomError::from(IggyError::InvalidCommand).into_response());
+ };
+ let mut guard = None;
+ loop {
+ if let Some(primary) = partition_primary(state, namespace).await {
Review Comment:
nit: after the hop target answers `NotPrimary`, the retry re-resolves the
primary only from this node's own view, so a lagging node keeps relaying to the
same stale primary until the 10s deadline. Carrying the refusing node's
view/primary as a hint (like `RouteCache::not_primary`) would fix it; fine as a
follow-up.
##########
core/binary_protocol/src/responses/messages/partition_route.rs:
##########
@@ -0,0 +1,157 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::WireError;
+use crate::codec::{WireDecode, WireEncode, read_u8, read_u32_le, read_u64_le};
+use crate::primitives::partition_history::PartitionContext;
+use bytes::{BufMut, BytesMut};
+
+/// One replica's route to a partition, the reply to `GetPartitionRoute` (105).
+///
+/// The incarnation and the partition count come from metadata at
+/// `metadata_op` or later. The view and its primary come from this replica's
+/// copy of the partition group, so another replica can answer a higher view.
+///
+/// Wire format, exactly 25 bytes, little endian:
+/// ```text
+///
[incarnation:8][metadata_op:8][view:4][partitions_count:4][primary_replica:1]
+/// ```
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct PartitionRouteResponse {
Review Comment:
**critical**: 105 keeps its code but the reply grows from the 24-byte
`PartitionContext` to this 25-byte route. Node, Go, Java and C# still decode
the old shape: Go rejects every explicit-partition send
(`contracts/partitions.go:36` wants 24), Node/C# check only `<` and misread
`metadata_op` as `view|partitions_count<<32`, so sends spin on
`TransientNotAccepted` until the deadline.
Please update all four SDKs (rename to GetPartitionRoute, decode exactly 25
bytes into {incarnation, owner 0, metadata_op}, use `!=` not `<`, regenerate
fixtures).
##########
core/server/src/dispatch/mod.rs:
##########
@@ -240,13 +249,130 @@ impl Drop for ActiveDrainSlot {
}
}
+/// The state that the drain task of one connection shares with the requests
+/// it spawned.
+///
+/// The drain takes the frames of the connection in arrival order and runs the
+/// prologue of each. It finishes a frame that later frames depend on before it
+/// takes the next one: a login, a logout, a `BIND_SESSION`, the rewrite of a
+/// metadata write, or a partition admission. A `PING` finishes there too,
Review Comment:
nit: "A `PING` finishes there too, because it never waits" holds only when
the PING is at the queue head. A PING behind a frame parked in `admit`,
`wait_for_partition_routable` or an attached relay waits too, and
`touch_connection` never runs for it, so a long metadata stall can evict a
pipelining client that keeps pinging. Answer or touch PING on enqueue, or
narrow the doc.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -240,13 +249,130 @@ impl Drop for ActiveDrainSlot {
}
}
+/// The state that the drain task of one connection shares with the requests
+/// it spawned.
+///
+/// The drain takes the frames of the connection in arrival order and runs the
+/// prologue of each. It finishes a frame that later frames depend on before it
+/// takes the next one: a login, a logout, a `BIND_SESSION`, the rewrite of a
+/// metadata write, or a partition admission. A `PING` finishes there too,
+/// because it never waits. Every other read, every poll and every wait for a
+/// metadata commit runs in a task of its own, so a held frame does not stall
+/// the frames behind it.
+///
+/// `capacity` caps those tasks at `message_bus.client_queue_capacity`, the
+/// outbound queue size of a connection, so their replies never overflow that
+/// queue. A metadata write, a segment delete, a login and a logout also take
+/// the metadata lane, which holds one of them at a time. The metadata client
+/// table keeps one pending write per client and refuses a lower request id
+/// after a higher one committed, so a second write in flight would fail with
+/// `TransientNotAccepted` or `RequestTooOld`. The lane also runs the rewrite
of
+/// a write after the previous write replied, because the rewrite reads state
+/// that the previous write can change.
+///
+/// At the cap or the lane the drain waits, and the frames behind it queue up
+/// to [`MAX_QUEUED_CLIENT_REQUESTS`]. While tasks run, the drain outlives its
+/// queue, so the cap and the lane still count them when the next frame
+/// arrives.
+pub(in crate::dispatch) struct ClientDrain {
+ capacity: usize,
+ in_flight: Cell<usize>,
+ /// Whether a frame holds the metadata lane.
+ metadata_write: Cell<bool>,
+ /// The drain task, while it waits for a frame or for room to admit one.
+ parked: Cell<Option<Waker>>,
+}
+
+/// What a frame waits for before [`ClientDrain::admit`] counts it.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Admission {
+ /// Room under the in-flight cap.
+ Cap,
+ /// Room under the cap and a free metadata lane.
+ MetadataLane,
+}
+
+impl ClientDrain {
+ pub(in crate::dispatch) const fn new(server_config: &ServerConfig) -> Self
{
+ Self {
+ capacity: server_config.message_bus.client_queue_capacity,
Review Comment:
**warning**: spawned reads are capped by `client_queue_capacity` (256), so
one pipelining connection can hold 256 concurrent polls and take a large share
of the shard's `poll_completion_capacity` (1024, counts ops not bytes). A small
dedicated read cap (around 8) would bound it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
