numinnex commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4263189782
##########
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:
**warning**: re the PR6 deferral: this still spawns a task for every
non-PING read where master runs it inline, which measured -17% poll and -48%
consumer under concurrent produce for every lockstep SDK. Merging ships that
via the edge builds.
The new progress lane already makes an inline poll order-safe, and the doc
at :259-266 already accepts PING waiting behind a bounded frame. An inline-only
`POLL_MESSAGES*` fast path recovered 97% in the earlier measurement. Please
land it here rather than in PR6.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -377,6 +379,71 @@ async fn relay_partition_reply<B, MJ, S, SB>(
});
}
+/// The offset key shared by polls and explicit progress writes on a
connection.
+/// Failed resolution is left to the normal handler's authorization and error
path.
+pub(super) fn progress_key<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ message: &Message<GenericHeader>,
+) -> Option<ConsumerProgress>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let request = message.try_as_typed::<RequestHeader>().ok()?;
+ let wire = match request.header().operation {
+ Operation::NonReplicated
+ if request.header().reserved[..4] ==
GET_CONSUMER_OFFSET_CODE.to_le_bytes() =>
+ {
+ GetConsumerOffsetRequest::decode_from(request.body()).ok()?
+ }
+ Operation::NonReplicated
Review Comment:
nit: the lane key covers every poll and `GetConsumerOffset`, so
Offset/Timestamp polls without auto_commit also serialize behind a store.
Only polls that read or commit the stored offset (`Next` or `auto_commit`)
need it. Keep store<->store serialized, though: a transient-refused store
retried after a later accepted one would move the offset backwards.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -240,13 +252,149 @@ 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. Login, logout, `BIND_SESSION` and partition admission
+/// finish on the drain. Metadata commits run in tasks, but later non-PING
+/// frames wait in the queue until the metadata lane releases, before they
+/// capture session state or resolve a namespace. PING can pass that queue.
Review Comment:
nit: the doc says later frames wait "before they ... resolve a namespace",
but the Partition arm drops `context.metadata_watermark` (:864-900) and
resolves from the local STM. On a backup-homed or peer-shard connection, a
pipelined JoinConsumerGroup -> StoreConsumerOffset can get
`ConsumerGroupPartitionNotOwned` (namespace.rs:278-283).
Either hold the partition frame on the session watermark or narrow the doc.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -283,13 +449,53 @@ fn pop_next_client_request(
// process exit.
let mut queues = queues.borrow_mut();
let queue = queues.get_mut(&client_id)?;
- let message = queue.pop_front();
+ let mut waiting = false;
+ let mut barrier = false;
+ let index = queue.iter().position(|message| {
Review Comment:
nit: every wake re-runs `progress_key` (body decode, `WireName` allocs, STM
resolve) for each queued keyed frame, so N pipelined polls on one key cost
O(N²) up to the 1024-frame queue. Together with the handler's own pass a poll
now decodes its key 3 times.
Compute the key once at enqueue and store it with the frame.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -9919,6 +10175,8 @@ where
self.publish_consumer_offset_effect(*effect);
self.apply_consumer_offset_commit(effect.pending);
}
+ self.durable_consumer_offsets
+ .set_emptied_at(pending.emptied_at);
Review Comment:
nit (repeat, no reply): `emptied_at` is still written back unconditionally
from the `select_commit` snapshot. A no-ack store that publishes in between
gets overwritten with the stale value. That's safe (it only refuses more), but
memory, disk and the offer diverge.
Write it only when this run's plan set `emptied_at`.
##########
core/server/src/http/forward.rs:
##########
@@ -42,24 +42,32 @@
//! - The primary re-authenticates and re-authorizes the forwarded request
//! through its ordinary extractor stack; the forward marker header is a loop
//! guard only and never a trust input.
-//! - A forwarded request is retried only when it provably never entered the
-//! primary's pipeline: a connect-phase failure, a 503 whose body carries the
-//! `TransientNotAccepted` code, or a 307 from a stale target. Every other
-//! outcome - including a 503 carrying `TransientNotCommitted`, whose op may
-//! still commit - is relayed as-is, because re-issuing it under a fresh
+//! - The route header, like the marker, carries only values a caller can send
+//! itself, so the hop target authenticates and authorizes the request again.
+//! - A forwarded write is retried only when it provably never entered the
+//! primary's pipeline: a connect-phase failure, and for a metadata relay a
+//! 503 whose body carries the `TransientNotAccepted` or `NotPrimary` code or
+//! a 307 from a stale target, for a partition relay a 503 carrying
+//! `NotPrimary`. Every other outcome - including a 503 carrying
+//! `TransientNotCommitted`, whose op
Review Comment:
nit (partial): the reflow still leaves a short line,
"`TransientNotCommitted`, whose op". Please rewrap the paragraph.
##########
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:
nit (repeat, no reply): with `?consistency=linearizable`, a node that is
neither the metadata nor the partition primary first 307s to the metadata
primary, which then relays to the partition primary. That is two hops where the
partition relay alone would do.
On `PartitionHop::Contacted`, the relay to the partition primary should
satisfy linearizable without the 307.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -345,6 +346,7 @@ async fn relay_partition_reply<B, MJ, S, SB>(
// Through the bus, not the runtime directly: the simulator supplies its
// own executor and virtual clock.
shard.bus.clone().spawn(async move {
+ let _progress = progress;
Review Comment:
nit: when `await_partition_submit` expires (30 s) the lane is released here,
but the store can still commit later and overwrite a newer auto-commit for the
same consumer. The outcome is already unknown to the client, so a doc line on
the lane contract is enough.
##########
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:
nit: this check never fires in fuzz. Two of us mutated
`matches_consumer_owner` to always accept, and separately to skip only the
generation compare or only the session compare. 24 fuzz seeds produced logs
byte-identical to base, and all 228 sim tests still pass with
`owners_compared=0`. The workload stores from a random consumer id
(`ops/store_consumer_offset.rs:64`) and never from a stale owner.
Add a stale-owner store action plus a counter of owner-fence refusals that
must be > 0. The PR body claim "Simulator checks cover replacement histories"
should be softened until then.
##########
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:
nit: `send.stopped().await` now runs on every stream in steady state,
costing a per-request insert into the connection's waiter map and keeping the
serve future alive for up to max_ack_delay. Only shutdown needs it.
Await `stopped()` only once shutdown has been observed.
--
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]
