fix(client): replace deferred open stop with immediate cancellation
Rust CI / cargo fmt (push) Successful in 16s
Rust CI / cargo check (x86_64-unknown-linux-musl) (push) Failing after 11s
Rust CI / cargo check (x86_64-unknown-linux-gnu) (push) Successful in 35s
Rust CI / cargo test (x86_64-unknown-linux-musl) (push) Failing after 8s
Rust CI / cargo clippy (push) Successful in 35s
Netem Interop / netem interop (push) Failing after 17s
Supply chain / cargo-deny advisories (push) Failing after 4s
Supply chain / cargo-deny bans licenses sources (push) Failing after 3s
Rust CI / cargo doc (push) Successful in 30s
Supply chain / actionlint (push) Successful in 4s
Fuzz Smoke / cargo fuzz smoke (push) Failing after 34s
Rust CI / cargo test (x86_64-unknown-linux-gnu) (push) Successful in 1m16s

This commit is contained in:
Nils Andreas Svee committed 2026-10-21 23:50:14 +07:00
1 parent d2a7acdec4
commit 52f2a07490
2 files changed
+26 -78

No files matched your search

+10 -23
View File
@@ -93,8 +93,6 @@ pub(crate) struct AsyncOpenState {
deadline: Option<Instant>, deadline: Option<Instant>,
deadline_timer: Option<Pin<Box<time::Sleep>>>, deadline_timer: Option<Pin<Box<time::Sleep>>>,
request_submitted: bool, request_submitted: bool,
stop_after_current_attempt: bool,
stopped_after_current_attempt: bool,
buffer: [u8; MAX_OPEN_PACKET_SIZE], buffer: [u8; MAX_OPEN_PACKET_SIZE],
cleanup: Option<AsyncOpenCleanup>, cleanup: Option<AsyncOpenCleanup>,
} }
@@ -106,8 +104,6 @@ impl AsyncOpenState {
deadline: None, deadline: None,
deadline_timer: None, deadline_timer: None,
request_submitted: false, request_submitted: false,
stop_after_current_attempt: false,
stopped_after_current_attempt: false,
buffer: [0_u8; MAX_OPEN_PACKET_SIZE], buffer: [0_u8; MAX_OPEN_PACKET_SIZE],
cleanup: None, cleanup: None,
} }
@@ -128,21 +124,12 @@ impl AsyncOpenState {
self.deadline = None; self.deadline = None;
self.deadline_timer = None; self.deadline_timer = None;
self.request_submitted = false; self.request_submitted = false;
self.stopped_after_current_attempt = self.stop_after_current_attempt;
} }
pub(crate) fn has_in_flight_work(&self) -> bool { pub(crate) fn has_in_flight_work(&self) -> bool {
self.request_submitted || self.cleanup.is_some() self.request_submitted || self.cleanup.is_some()
} }
pub(crate) fn request_stop_after_current_attempt(&mut self) {
self.stop_after_current_attempt = true;
}
pub(crate) fn stopped_after_current_attempt(&self) -> bool {
self.stopped_after_current_attempt
}
fn deadline(&self) -> Instant { fn deadline(&self) -> Instant {
self.deadline self.deadline
.expect("an open attempt has an active deadline") .expect("an open attempt has an active deadline")
@@ -324,10 +311,6 @@ impl AsyncClient {
if state.attempt >= attempt_count { if state.attempt >= attempt_count {
return Poll::Ready(Err(ClientError::OpenTimeout)); return Poll::Ready(Err(ClientError::OpenTimeout));
} }
if state.stop_after_current_attempt && state.deadline.is_none() {
state.stopped_after_current_attempt = true;
return Poll::Ready(Err(ClientError::OpenTimeout));
}
if state.deadline.is_none() { if state.deadline.is_none() {
let timeout = self.machine.config().open.timeouts[state.attempt]; let timeout = self.machine.config().open.timeouts[state.attempt];
if let Err(error) = state.start_attempt(timeout) { if let Err(error) = state.start_attempt(timeout) {
@@ -368,9 +351,6 @@ impl AsyncClient {
Poll::Ready(Err(error)) => return Poll::Ready(Err(error)), Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
Poll::Ready(Ok(false)) => { Poll::Ready(Ok(false)) => {
state.finish_attempt(); state.finish_attempt();
if state.stopped_after_current_attempt() {
return Poll::Ready(Err(ClientError::OpenTimeout));
}
continue; continue;
} }
Poll::Ready(Ok(true)) => {} Poll::Ready(Ok(true)) => {}
@@ -396,9 +376,6 @@ impl AsyncClient {
}; };
if datagram.received_at.mono > state.deadline() { if datagram.received_at.mono > state.deadline() {
state.finish_attempt(); state.finish_attempt();
if state.stopped_after_current_attempt() {
return Poll::Ready(Err(ClientError::OpenTimeout));
}
continue; continue;
} }
@@ -778,6 +755,16 @@ impl AsyncClient {
.map_err(ClientError::Socket) .map_err(ClientError::Socket)
} }
/// Cancel opening without waiting for a reply or socket writability.
/// Retained post-token cleanup gets one best-effort nonblocking send.
pub(crate) fn cancel_open(&self, state: &mut AsyncOpenState) -> Option<ClientError> {
let cleanup = state.cleanup.take()?;
if let Some(packet) = cleanup.packet() {
let _ = self.socket.try_send(packet);
}
Some(cleanup.into_primary())
}
fn poll_open_cleanup( fn poll_open_cleanup(
&self, &self,
state: &mut AsyncOpenState, state: &mut AsyncOpenState,
+16 -55
View File
@@ -409,12 +409,12 @@ impl ManagedClientTask {
let retirement = retirement let retirement = retirement
.clone() .clone()
.unwrap_or(ManagedTargetEndReason::Stopped); .unwrap_or(ManagedTargetEndReason::Stopped);
if !open.has_in_flight_work() { let cleanup_failure = client.cancel_open(open).map(|error| {
self.targets[index].sync_live_packets_sent(); classify_client_error(ManagedTargetFailurePhase::Opening, &error)
self.finish_target(index, retirement, None); });
return false; self.targets[index].sync_live_packets_sent();
} self.finish_target(index, retirement, cleanup_failure);
open.request_stop_after_current_attempt(); return false;
} }
match client.poll_open(open, cx) { match client.poll_open(open, cx) {
Poll::Pending => false, Poll::Pending => false,
@@ -438,64 +438,25 @@ impl ManagedClientTask {
index, index,
vec![ClientEvent::SessionStarted(started)], vec![ClientEvent::SessionStarted(started)],
); );
if stopping { let previous = self.targets[index].take_phase();
self.targets[index].sync_live_packets_sent(); self.targets[index].restore_phase(TargetPhase::Active {
self.begin_drain( client: previous.into_client(),
index, schedule,
retirement send_waiting: false,
.clone() });
.unwrap_or(ManagedTargetEndReason::Stopped), self.publish_target_phase(index, ManagedTargetLifecycle::Opening);
now, true
)
} else {
let previous = self.targets[index].take_phase();
self.targets[index].restore_phase(TargetPhase::Active {
client: previous.into_client(),
schedule,
send_waiting: false,
});
self.publish_target_phase(index, ManagedTargetLifecycle::Opening);
true
}
} }
Poll::Ready(Ok(OpenOutcome::NoTestCompleted(completed))) => { Poll::Ready(Ok(OpenOutcome::NoTestCompleted(completed))) => {
self.publish_client_events( self.publish_client_events(
index, index,
vec![ClientEvent::NoTestCompleted(completed)], vec![ClientEvent::NoTestCompleted(completed)],
); );
let end_reason = if stopping { self.finish_target(index, ManagedTargetEndReason::NoTestComplete, None);
retirement
.clone()
.unwrap_or(ManagedTargetEndReason::Stopped)
} else {
ManagedTargetEndReason::NoTestComplete
};
self.finish_target(index, end_reason, None);
false false
} }
Poll::Ready(Err(error)) => { Poll::Ready(Err(error)) => {
if stopping { self.fail_target(index, ManagedTargetFailurePhase::Opening, error);
let TargetPhase::Opening { open, .. } = self.targets[index].phase()
else {
unreachable!()
};
let cleanup_failure =
(!open.stopped_after_current_attempt()).then(|| {
classify_client_error(
ManagedTargetFailurePhase::Opening,
&error,
)
});
self.finish_target(
index,
retirement
.clone()
.unwrap_or(ManagedTargetEndReason::Stopped),
cleanup_failure,
);
} else {
self.fail_target(index, ManagedTargetFailurePhase::Opening, error);
}
false false
} }
} }