hubcio commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4270143419
##########
core/simulator/src/workload/oracle.rs:
##########
@@ -449,14 +500,15 @@ pub fn assert_converged(sim: &Simulator, workload: &mut
Workload) -> Convergence
"committed metadata prefixes agree across every live replica"
);
- let Some(leader) = metadata_leader(sim, &live) else {
+ let Some(leader) = sim.metadata_leader().map(usize::from) else {
panic!("no metadata leader live at quiesce (seed={seed:#x})");
};
// Settlement treats "no live replica hosts this group" as settled, so
losing
// every instance of a LIVE partition reads as converged. The metadata read
// separates that from a group whose stream the workload deleted.
let namespaces_checked = assert_live_namespaces_have_a_primary(sim,
workload, leader, seed);
Review Comment:
Deferred to PR6 as recovery release blockers. B5 now retains the
RebuildPipeline floor fix (`max(commit_min, merged.commit_max) + 1`),
quorum-loss creation/delete wedge (#4476 family, including seed 24), and the
baseline metadata fork campaigns. Closure requires separate baseline/candidate
reproductions with full settings and acknowledged-history/recovery proof. The
passing PR4 seed 7016 does not close these findings; the baseline attribution
remains your evidence, not a campaign I reran.
##########
core/simulator/src/workload/oracle.rs:
##########
@@ -449,14 +500,15 @@ pub fn assert_converged(sim: &Simulator, workload: &mut
Workload) -> Convergence
"committed metadata prefixes agree across every live replica"
);
- let Some(leader) = metadata_leader(sim, &live) else {
+ let Some(leader) = sim.metadata_leader().map(usize::from) else {
panic!("no metadata leader live at quiesce (seed={seed:#x})");
};
// Settlement treats "no live replica hosts this group" as settled, so
losing
// every instance of a LIVE partition reads as converged. The metadata read
// separates that from a group whose stream the workload deleted.
let namespaces_checked = assert_live_namespaces_have_a_primary(sim,
workload, leader, seed);
+ assert_hosted_incarnations_named(sim, workload, &live, seed);
// Safety direction: no live replica may have COMMITTED more of a group
than
Review Comment:
Deferred to PR6. The plan and consolidated simulator acceptance now require
per-namespace applied-data/readback digests through apply, rebuild and transfer
installation, plus a negative control where prepare identities agree but
applied bytes differ. Simulator partition transfer must be implemented or its
unsupported-directory runs explicitly classified, including seed 1078 with its
full campaign settings. Such a classified run does not count as passing
recovery evidence. The PR description limits PR4's convergence claim to prepare
identities and offset tables.
##########
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) {
Review Comment:
Deferred to PR6 with the QUIC reply-tracking work. An unclaimed Eviction
that predates a new stream must not be assigned to that stream. Acceptance
requires late-event isolation together with session-terminal Eviction behavior,
bounded tracking, late acknowledgements and draining within close_grace. The B7
and PR6 plan entries retain these conditions; PR4 does not change Eviction into
a nonterminal event.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -557,25 +814,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. The recorded proposal is to poll inline, select against
frame arrival, then promote the same pending future, InFlight permit and
progress-lane guard when another frame arrives. The existing PING negative
control rejects blanket inline serialization; it does not disprove this
proposal. B7 requires lockstep and pipelined benchmarks plus preserved
cancellation, bounded admission, unknown outcomes and Next/auto-commit
ordering. The reported performance figures remain attributed measurements, not
results I reproduced.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -377,6 +381,74 @@ async fn relay_partition_reply<B, MJ, S, SB>(
});
}
+/// The offset key shared by stored-offset reads and 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
+ if request.header().reserved[..4] ==
POLL_MESSAGES_CODE.to_le_bytes()
+ || request.header().reserved[..4]
+ == POLL_MESSAGES_ON_PRIMARY_CODE.to_le_bytes() =>
+ {
+ let poll = PollMessagesRequest::decode_from(request.body()).ok()?;
Review Comment:
Deferred to PR6. B7 now explicitly requires removing the duplicate full
PollMessages decode/allocation for non-Next, non-auto-commit polls, using the
flag/strategy fields or passing the decoded request onward. Closure requires
allocation/throughput measurements and unchanged Next/auto-commit progress
ordering.
--
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]
