hubcio commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4261453759
##########
core/message_bus/src/transports/quic.rs:
##########
@@ -326,12 +316,202 @@ 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();
Review Comment:
Deferred to multi-leader PR6. Profile QUIC waiter/map/timer overhead before
changing it; any deadline sweep must preserve per-request expiry, cancellation
cleanup and the bounded shutdown drain fixed in f90c6c858.
##########
core/simulator/src/workload/mod.rs:
##########
@@ -621,12 +758,34 @@ impl Workload {
// Effect-follows-actual: drive the shadow off the committed outcome,
never
// the targeted one. Success mutates; a nonzero code is a committed
no-op
// whose `predicted_effect` is `Effect::None`. Keeps the shadow correct
- // under at-least-once re-execution and races.
- let effect = ops::predicted_effect(&entry.input, &classified);
+ // under at-least-once re-execution and races. A partition-plane
success
+ // sampled across a fixture replacement can belong to the deleted
+ // incarnation, whose state the shadow dropped with it.
+ let current = header.operation.is_metadata()
+ || self
+ .shadow
Review Comment:
The pending-delete shadow bug is fixed in 0dd5939c2, with tests for both
refused and successful deletion. Simulator poll actions and consumer-liveness
execution are deferred to multi-leader PR6 acceptance coverage. That follow-up
must exercise duplicate-delivery and session-retirement failures, including
crash/recovery runs.
--
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]
