hubcio commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4263738636
##########
core/server/src/dispatch/mod.rs:
##########
@@ -557,25 +769,51 @@ 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 = match progress {
+ Some(progress) => progress,
+ None => 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:
Deferred to PR6's dispatch performance work. A poll still awaits its
partition owner and possibly disk I/O; running the whole poll inline holds the
connection drain during that wait. The held-owner regression covers both poll
codes and requires PING to complete before the poll reply. The blanket-inline
negative control fails this test: no PING reply arrives while the owner reply
is held. PR6 must remove scheduling overhead while preserving that progress,
in-flight bounds, cancellation/unknown-outcome behavior and Next/auto-commit
ordering, then measure lockstep and pipelined workloads against a reproducible
baseline. I have not independently reproduced the reported benchmark
percentages.
##########
core/message_bus/src/transports/quic.rs:
##########
@@ -326,12 +316,205 @@ fn reply_route(frame: &BusMessage) -> ReplyRoute {
if !is_reply {
return ReplyRoute::Unkeyed;
}
+ read_u64(bytes, offset_of!(ReplyHeader,
request)).map_or(ReplyRoute::Unkeyed, ReplyRoute::Keyed)
+}
+
+/// The request id of a client frame, read as raw bytes like [`reply_route`],
+/// so a frame with an undeclared operation still has one.
+fn request_id(request: &Message<GenericHeader>) -> Option<u64> {
+ read_u64(request.as_slice(), offset_of!(RequestHeader, request))
+}
+
+fn read_u64(bytes: &[u8], offset: usize) -> Option<u64> {
bytes
- .get(..HEADER_SIZE)
- .and_then(|header|
bytemuck::checked::try_from_bytes::<ReplyHeader>(header).ok())
- .map_or(ReplyRoute::Unkeyed, |header| {
- ReplyRoute::Keyed(header.request)
- })
+ .get(offset..offset + size_of::<u64>())
+ .and_then(|field| field.try_into().ok())
+ .map(u64::from_le_bytes)
+}
+
+/// Hands the outbound frames of one connection to the streams that wait
+/// for them.
+///
+/// A stream registers its waiter before it forwards its request, so a reply
+/// never comes before the stream it answers.
+struct ReplyRouter {
+ /// Waiting streams by request id, oldest first.
+ waiters: HashMap<u64, VecDeque<oneshot::Sender<BusMessage>>>,
+ /// Frames that came while no stream waited for them, oldest first.
+ unclaimed: VecDeque<BusMessage>,
+ /// When `unclaimed` holds this many frames, it drops its oldest one.
+ unclaimed_max: usize,
+}
+
+impl ReplyRouter {
+ fn new(unclaimed_max: usize) -> Self {
+ Self {
+ waiters: HashMap::new(),
+ unclaimed: VecDeque::new(),
+ unclaimed_max,
+ }
+ }
+
+ /// Register a stream that waits for the reply to `request_id`.
+ ///
+ /// The stream first takes the oldest unclaimed frame that has no request
+ /// id or has its own. The SDK replays an explicit transient with the same
+ /// request id on a new stream, so a late reply of the first attempt
+ /// answers the replay. The replies for other ids ahead of that frame are
+ /// dropped, because their streams already ended.
+ fn register(&mut self, request_id: u64) -> oneshot::Receiver<BusMessage> {
+ let (waiter, reply) = oneshot::channel();
+ while let Some(frame) = self.unclaimed.pop_front() {
+ if !matches!(reply_route(&frame), ReplyRoute::Keyed(id) if id !=
request_id) {
+ // `reply` is alive, so the send cannot fail.
+ let _ = waiter.send(frame);
+ return reply;
+ }
+ }
+ self.waiters
+ .entry(request_id)
+ .or_default()
+ .push_back(waiter);
+ reply
+ }
+
+ /// Hand a reply to the oldest waiter of its request id, and a frame
+ /// without one to every waiter, because an eviction ends every request
+ /// of the connection. A frame no stream takes stays unclaimed.
+ fn route(&mut self, frame: BusMessage) {
+ let unclaimed = match reply_route(&frame) {
+ ReplyRoute::Keyed(request_id) => self.deliver(request_id, frame),
+ ReplyRoute::Unkeyed => self.broadcast(frame),
+ };
+ if let Some(frame) = unclaimed {
+ if self.unclaimed.len() == self.unclaimed_max {
+ self.unclaimed.pop_front();
+ }
+ self.unclaimed.push_back(frame);
+ }
+ }
+
+ /// Drop the waiters of `request_id` whose streams ended without a reply.
+ fn release(&mut self, request_id: u64) {
+ if let Some(waiters) = self.waiters.get_mut(&request_id) {
+ waiters.retain(|waiter| !waiter.is_canceled());
+ if waiters.is_empty() {
+ self.waiters.remove(&request_id);
+ }
+ }
+ }
+
+ /// Returns the frame if no waiter took it.
+ fn deliver(&mut self, request_id: u64, mut frame: BusMessage) ->
Option<BusMessage> {
+ let Some(waiters) = self.waiters.get_mut(&request_id) else {
+ return Some(frame);
+ };
+ let unclaimed = loop {
+ let Some(waiter) = waiters.pop_front() else {
+ break Some(frame);
+ };
+ match waiter.send(frame) {
+ Ok(()) => break None,
+ Err(returned) => frame = returned,
+ }
+ };
+ if waiters.is_empty() {
+ self.waiters.remove(&request_id);
+ }
+ unclaimed
+ }
+
+ /// Returns the frame if no waiter took it.
+ fn broadcast(&mut self, frame: BusMessage) -> Option<BusMessage> {
+ let mut delivered = false;
+ for waiter in self.waiters.drain().flat_map(|(_, waiters)| waiters) {
+ delivered |= waiter.send(frame.clone()).is_ok();
+ }
+ (!delivered).then_some(frame)
+ }
+}
+
+/// What the stream futures of [`QuicTransportConn::run`] borrow from it.
+struct StreamContext<'a> {
+ in_tx: &'a Sender<Message<GenericHeader>>,
+ router: &'a RefCell<ReplyRouter>,
+ max_message_size: usize,
+ reply_backstop: Duration,
+ label: &'static str,
+ peer: &'a str,
+}
+
+impl StreamContext<'_> {
+ /// Serve one request stream. Breaks when the inbound queue is gone,
+ /// because the connection then has no dispatch.
+ #[allow(clippy::future_not_send)]
+ async fn serve(&self, mut send: SendStream, mut recv: RecvStream) ->
ControlFlow<()> {
+ let Self {
+ in_tx,
+ router,
+ max_message_size,
+ reply_backstop,
+ label,
+ peer,
+ } = *self;
+ // The SDK `finish()`-es after its request, so the stream carries one
+ // frame.
+ let request = match framing::read_message(&mut recv,
max_message_size).await {
+ Ok(request) => request,
+ Err(e) => {
+ debug!(%label, %peer, error = ?e, "quic: bidi read error");
+ return ControlFlow::Continue(());
+ }
+ };
+ let Some(request_id) = request_id(&request) else {
+ debug!(%label, %peer, "quic: frame shorter than a request header");
+ return ControlFlow::Continue(());
+ };
+ let mut reply = router.borrow_mut().register(request_id);
+ if in_tx.send(request).await.is_err() {
+ debug!(%label, %peer, "quic: inbound queue dropped");
+ return ControlFlow::Break(());
+ }
+ let replied = {
+ let mut stopped = pin!(send.stopped().fuse());
+ let mut backstop =
pin!(compio::time::sleep(reply_backstop).fuse());
+ futures::select_biased! {
+ frame = reply => frame.ok(),
+ _ = stopped => {
+ debug!(%label, %peer, request_id, "quic: client stopped
the stream before its reply");
+ None
+ }
+ () = backstop => {
+ warn!(%label, %peer, request_id, "quic: no reply within
the backstop; resetting the stream");
+ None
+ }
+ }
+ };
+ let Some(frame) = replied else {
+ drop(reply);
+ router.borrow_mut().release(request_id);
+ if let Err(e) = send.reset(VarInt::from_u32(QUIC_NO_REPLY)) {
+ debug!(%label, %peer, error = ?e, "quic: stream reset failed");
+ }
+ return ControlFlow::Continue(());
+ };
+ for fragment in frame.into_fragments() {
+ let BufResult(result, _frozen) = send.write_all(fragment).await;
+ if let Err(e) = result {
+ debug!(%label, %peer, error = ?e, "quic: write failed");
+ return ControlFlow::Continue(());
+ }
+ }
+ // `finish()` flushes pending data + a FIN, which tells the SDK the
+ // reply is complete.
+ if let Err(e) = send.finish() {
+ debug!(%label, %peer, error = ?e, "quic: send.finish() failed");
+ }
+ // Keep the reply alive through graceful shutdown until the peer has
+ // acknowledged it. Closing the connection after FIN alone can discard
it.
+ let _ = send.stopped().await;
Review Comment:
Deferred to PR6's QUIC bookkeeping work. Waiting only when shutdown has
already been observed misses replies whose FIN was sent just before shutdown
but is still unacknowledged when the connection closes. The optimization needs
bounded tracking of those replies and proof that it preserves graceful reply
draining and the close_grace deadline; the current ACK wait stays until that
replacement is verified.
##########
core/server/src/http/handlers.rs:
##########
@@ -1256,25 +1253,27 @@ pub(in crate::http) async fn delete_segments(
/// [`poll_wire_request`]) since HTTP has no persistent connection and cannot
observe
/// group's partition assignment (that can be subject to change).
///
-/// A non-replicated read served in band: the same resolution the TCP dispatch
-/// runs ([`resolve_poll_request`]), then a mesh read on the owning shard and a
-/// re-encode of the stored batches into the legacy wire body, decoded here by
-/// the SDK's own [`PolledMessages::from_bytes`] so the JSON is field-identical
-/// to a TCP poll. `auto_commit` rides [`resolve_poll_request`]'s args and is
-/// honored by the owning shard's poll plan; no extra HTTP work.
+/// A non-replicated read served by the partition primary: the same resolution
+/// the TCP dispatch runs ([`resolve_poll_request`]), then a mesh read on the
+/// owning shard and a re-encode of the stored batches into the legacy wire
+/// body, decoded here by the SDK's own [`PolledMessages::from_bytes`] so the
+/// JSON is field-identical to a TCP poll. `auto_commit` rides
+/// [`resolve_poll_request`]'s args and is honored by the owning shard's poll
+/// plan. Another node relays the poll to the primary ([`route_partition`]).
pub(in crate::http) async fn poll_messages(
State(state): State<HttpState>,
identity: Identity,
Path((stream_id, topic_id)): Path<(String, String)>,
Query(query): Query<PollMessages>,
Query(consistency): Query<ConsistencyQuery>,
-) -> Result<Json<PolledMessages>, ReadError> {
+ Extension(request): Extension<Arc<PartitionRequest>>,
+) -> Result<Response, ReadError> {
let stream_id =
Identifier::from_str_value(&stream_id).map_err(ReadError::Rejected)?;
let topic_id =
Identifier::from_str_value(&topic_id).map_err(ReadError::Rejected)?;
SendWrapper::new(gate_local_read(
&state,
&identity,
- consistency.consistency,
+ partition_read_consistency(&request, consistency.consistency),
Review Comment:
Deferred to PR6's HTTP routing performance work. The contacted node
currently enforces the requested metadata consistency before partition routing.
A partition-primary relay proves partition read authority, but not current
namespace/RBAC metadata on a lagging contacted node, so removing this redirect
alone weakens the linearizable-read contract. Acceptance for a one-relay path:
preserve namespace/incarnation resolution, authorization under revocation,
metadata read-your-writes floors and partition view-change safety when the
contacted node is neither primary.
##########
core/simulator/src/workload/state_checker.rs:
##########
@@ -265,6 +381,53 @@ impl StateChecker {
}
}
+/// Membership is metadata state; the partition's ordered install barrier is
+/// what fences a departed member's later offset writes.
+fn check_partition_owner(
+ owners: &mut BTreeMap<u64, ConsumerGroupOwner>,
+ header: &PrepareHeader,
+ body: &[u8],
+ seed: u64,
+) {
+ if header.operation == Operation::InstallConsumerGroupOwner {
+ let install = InstallConsumerGroupOwnerRequest::decode_from(body)
+ .expect("a committed owner installation decodes");
+ owners.insert(install.group_id, install.owner);
+ return;
+ }
+ if is_auto_commit_client(header.client) {
+ return;
+ }
+ let (request, generation) =
+ split_prepared_offset(body).expect("a committed offset has its owner
token");
+ let (request, _) =
+ GetConsumerOffsetRequest::decode(request).expect("a committed offset
prefix decodes");
+ if request.consumer.kind != KIND_CONSUMER_GROUP {
+ return;
+ }
+ let group_id = u64::from(
+ request
+ .consumer
+ .id
+ .as_u32()
+ .expect("a prepared group id is numeric"),
+ );
+ // The repair ring may have evicted the install before this checker
started.
+ if let Some(owner) = owners.get(&group_id) {
Review Comment:
Added a captured stale-owner store action and a deterministic leave/rejoin
scenario that retains the same client/session while replaying the old
generation. It records 29 submissions and 29 owner-fence refusals, plus a
strict offset-shadow comparison. Removing only the generation comparison makes
the committed-log oracle fail with the same client/session and generation 3
versus installed generation 7; restoring it passes. The separate production
session-only mutant proof remains deferred to PR6's session/recovery workload:
a direct oracle test rejects stale sessions, but earlier authentication can
reject those requests before they reach the owner check, so that unit test does
not prove workload coverage. PR6 must exercise that fence independently and
show the session-only mutant fails; the generation scenario does not close that
gap.
##########
foreign/go/client/tcp/tcp_poll_routing.go:
##########
@@ -200,33 +198,90 @@ func (c *IggyTcpClient) sendRouted(caller
context.Context, code command.Code, ke
c.polls.mu.Unlock()
if heartbeatDue {
if _, _, err := c.sendPollRequest(ctx,
uint32(command.PingCode), nil); err != nil {
Review Comment:
Deferred to PR5's managed replica-link lifecycle, with RTT/allocation/lock
measurements in PR6. Moving keepalive off the request path needs one session
owner with cancellation on close/logout/reset, bounded reconnect and liveness
when the metadata socket is unavailable but replica activity can still reach
the authority. That lifecycle must replace the current owner coherently; key
reuse and lock consolidation will be measured alongside it.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PollRouter.java:
##########
@@ -89,33 +95,105 @@ final class PollRouter {
}
CompletableFuture<ByteBuf> poll(ByteBuf payload,
Optional<PartitionContext> context) {
+ return partitionRequest(CommandCode.Messages.POLL.getValue(), payload,
context);
+ }
+
+ CompletableFuture<ByteBuf> partitionRequest(int command, ByteBuf payload,
Optional<PartitionContext> context) {
byte[] request = takeBytes(payload);
- return send(new Poll(
- CommandCode.Messages.GET_POLL_ROUTING,
- request,
- CommandCode.Messages.POLL_ON_PRIMARY.getValue(),
- request,
- request.length - POLL_PARAMETERS_BYTES,
- pollTimeout,
- context));
- }
-
- /**
- * Sends a store or delete offset request to the partition primary. The
route is
- * looked up by the consumer, stream, topic and partition in the first
- * {@code targetLength} bytes, as in Rust's {@code write_offset}.
- */
- CompletableFuture<ByteBuf> writeOffset(
- int command, ByteBuf payload, int targetLength,
Optional<PartitionContext> context) {
+ try {
+ byte[] query = partitionQuery(command, request);
+ return send(new Poll(query, command, request, pollTimeout,
context));
+ } catch (RuntimeException error) {
+ return CompletableFuture.failedFuture(error);
+ }
+ }
+
+ CompletableFuture<ByteBuf> sendMessages(ByteBuf payload, ByteBuf query) {
byte[] request = takeBytes(payload);
- return send(new Poll(
- CommandCode.ConsumerOffset.GET_ROUTING,
- Arrays.copyOf(request, targetLength),
- command,
- request,
- targetLength,
- pollTimeout,
- context));
+ byte[] target = takeBytes(query);
+ return send(new Poll(target, CommandCode.Messages.SEND.getValue(),
request, pollTimeout, Optional.empty()));
+ }
+
+ CompletableFuture<Long> partitionCount(StreamId stream, TopicId topic) {
+ ByteBuf query = BytesSerializer.toBytes(stream);
+ ByteBuf topicBytes = BytesSerializer.toBytes(topic);
+ try {
+ query.writeBytes(topicBytes);
+ query.writeIntLE(0);
+ } finally {
+ topicBytes.release();
+ }
+ byte[] target = takeBytes(query);
+ String topicKey = ByteBufUtil.hexDump(target, 0, target.length -
Integer.BYTES);
+ synchronized (this) {
+ Long count = partitionCounts.get(topicKey);
+ if (count != null) {
+ return CompletableFuture.completedFuture(count);
+ }
+ }
+ Poll lookup = new Poll(
+ target, CommandCode.Messages.GET_PARTITION_ROUTE.getValue(),
target, pollTimeout, Optional.empty());
+ return send(lookup).thenApply(response -> {
+ response.release();
+ return lookup.partitionsCount;
+ });
+ }
+
+ synchronized void clear() {
+ routes.clear();
+ partitionCounts.clear();
+ }
+
+ synchronized void observeTopology(ClusterMetadata metadata) {
+ replicas.clear();
+ rosterSize = metadata.nodes().size();
+ for (var node : metadata.nodes()) {
+ if (node.endpoints().tcp() != 0) {
+ replicas.put(
+ node.replicaId(),
+ new ConnectionInfo(node.ip(), node.endpoints().tcp()));
+ }
+ }
+ }
+
+ private static byte[] partitionQuery(int command, byte[] request) {
+ ByteBuf body = Unpooled.wrappedBuffer(request);
+ try {
+ int consumerKind = body.readUnsignedByte();
+ skipIdentifier(body);
+ int streamStart = body.readerIndex();
+ skipIdentifier(body);
+ skipIdentifier(body);
+ int partitionStart = body.readerIndex();
+ boolean explicit = body.readUnsignedByte() != 0;
+ long partition = body.readUnsignedIntLE();
+ if (!explicit) {
+ if (command == CommandCode.ConsumerOffset.STORE.getValue()
+ || command ==
CommandCode.ConsumerOffset.DELETE.getValue()
+ || (command == CommandCode.Messages.POLL.getValue() &&
consumerKind == 2)) {
+ throw new IggyInvalidArgumentException("A group poll or
offset mutation requires a partition id");
+ }
+ partition = 0;
+ body.setByte(partitionStart, 1);
+ body.setIntLE(partitionStart + 1, 0);
+ }
+ ByteBuf query = Unpooled.buffer(partitionStart - streamStart +
Integer.BYTES);
+ query.writeBytes(request, streamStart, partitionStart -
streamStart).writeIntLE((int) partition);
+ return takeBytes(query);
+ } catch (IndexOutOfBoundsException error) {
+ throw new IggyInvalidArgumentException("Truncated partition
target");
+ } finally {
+ body.release();
+ }
+ }
+
+ private static void skipIdentifier(ByteBuf body) {
+ int kind = body.readUnsignedByte();
+ int length = body.readUnsignedByte();
+ if ((kind == 1 && length != Integer.BYTES) || (kind != 1 && kind != 2)
|| length == 0) {
+ throw new IggyInvalidArgumentException("Invalid partition target
identifier");
+ }
+ body.skipBytes(length);
}
private static byte[] takeBytes(ByteBuf payload) {
Review Comment:
Deferred Java pipelining to PR5's shared lanes and reply mux. The current
slot owns one exchange; removing serialization alone lets one timeout close
another operation and bypasses auto-commit/explicit-offset ordering. Acceptance
includes reordered replies, cancellation and late-reply isolation, fixed
request identity/history and bounded retained bytes. ByteBuf ownership changes
and copy/allocation measurements remain PR6 follow-ups, with releases proven on
queue rejection, timeout, cancellation, retries and late replies under Netty
paranoid leak detection.
--
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]
