From 1d799c4ca0a5ca8c74bca942c4ba722129f1791d Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sat, 15 Aug 2026 22:50:19 +0700 Subject: [PATCH] fix(managed): restart resubmitted terminal targets --- crates/irtt-client/src/managed/task.rs | 12 +++- crates/irtt-client/src/managed/tests.rs | 86 ++++++++++++++++++++++++- 2 files changed, 93 insertions(+), 5 deletions(-) diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index da2787b..b200968 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -531,6 +531,10 @@ fn synchronously_retireable(state: &TargetState) -> bool { } } +fn runtime_satisfies_target(runtime: &TargetRuntime, target: &ManagedTargetConfig) -> bool { + runtime.desired && !matches!(runtime.state, TargetState::Terminal) && runtime.config == *target +} + #[derive(Default)] struct OutcomeHistory { limit: usize, @@ -996,7 +1000,11 @@ impl ManagedClientTask { } let mut retirements = Vec::new(); for (index, runtime) in self.targets.iter().enumerate() { - if !runtime.desired || prepared.iter().any(|(target, _)| target == &runtime.config) { + if !runtime.desired + || prepared + .iter() + .any(|(target, _)| runtime_satisfies_target(runtime, target)) + { continue; } let reason = if prepared @@ -1020,7 +1028,7 @@ impl ManagedClientTask { if self .targets .iter() - .any(|runtime| runtime.desired && runtime.config == target) + .any(|runtime| runtime_satisfies_target(runtime, &target)) { continue; } diff --git a/crates/irtt-client/src/managed/tests.rs b/crates/irtt-client/src/managed/tests.rs index 8f1ac6d..6c13daa 100644 --- a/crates/irtt-client/src/managed/tests.rs +++ b/crates/irtt-client/src/managed/tests.rs @@ -28,6 +28,7 @@ enum ServerBehavior { Echo, NoTest, PeerClose, + InvalidOpen, DelayedEcho(Duration), DeferredBurst(usize), NoisyOpen(usize), @@ -153,6 +154,10 @@ fn start_server_with_gates( if let Some(gate) = &open_gate { gate.arrive_and_wait(); } + if matches!(behavior, ServerBehavior::InvalidOpen) { + socket.send_to(&[0_u8], peer).unwrap(); + return; + } if let Some(interval) = interval { negotiated.interval_ns = i64::try_from(interval.as_nanos()).unwrap(); } @@ -645,6 +650,10 @@ fn dynamic_identical_open_target_preserves_session_without_pending_event() { }); let acknowledgement = runtime.block_on(receipt).unwrap(); assert_eq!(acknowledgement.status.targets[0].target, instance); + assert_eq!(acknowledgement.status.targets.len(), 1); + assert!(!observations.lock().unwrap().iter().any(|(event, _)| { + matches!(event, ManagedEvent::TargetFinished { outcome } if outcome.target == instance) + })); drive_task_until(&runtime, &mut task, |task| { task.targets[0].counters.packets_sent > sent }); @@ -664,6 +673,75 @@ fn dynamic_identical_open_target_preserves_session_without_pending_event() { ); } +#[test] +fn dynamic_identical_terminal_target_restarts_without_duplicate_outcome_at_live_limit() { + let server = start_server(ServerBehavior::InvalidOpen, None); + let mut managed = config(ManagedPacing::Staggered); + managed.completion = ManagedCompletionPolicy::ExplicitStop; + managed.max_live_target_generations = 1; + let configured = target("same", server.addr); + let (mut task, handle) = ManagedClient::task(managed, vec![configured.clone()]).unwrap(); + let observations = Arc::new(Mutex::new(Vec::new())); + task.event_observations = Some(Arc::clone(&observations)); + let runtime = runtime(); + let mut task = Box::pin(task); + + drive_task_until(&runtime, &mut task, |task| { + matches!(task.targets[0].state, TargetState::Terminal) + }); + let old = task.targets[0].instance.clone(); + assert!(task.targets[0].desired); + assert_eq!(handle.status().desired_target_count, 1); + assert_eq!(handle.status().terminal_target_count, 1); + let original_outcome = task.history.recent.back().unwrap().clone(); + assert_eq!(original_outcome.target, old); + assert!(matches!( + original_outcome.end_reason, + ManagedTargetEndReason::Failed(_) + )); + + let receipt = handle.update_targets(vec![configured]).unwrap(); + drive_task_until(&runtime, &mut task, |task| { + task.applied_command_sequence == 1 + }); + let acknowledgement = runtime.block_on(receipt).unwrap(); + assert_eq!(acknowledgement.sequence, 1); + assert_eq!(acknowledgement.status.desired_target_count, 1); + assert_eq!(acknowledgement.status.terminal_target_count, 0); + assert_eq!(acknowledgement.status.targets.len(), 1); + let new = acknowledgement.status.targets[0].target.clone(); + assert!(new.generation > old.generation); + assert_eq!( + acknowledgement.status.targets[0].lifecycle, + ManagedTargetLifecycle::Pending + ); + assert_eq!(task.targets.len(), 1); + assert_eq!(task.targets[0].instance, new); + assert!(task.targets[0].desired); + assert!(!matches!(task.targets[0].state, TargetState::Terminal)); + assert!(observations.lock().unwrap().iter().any(|(event, _)| { + matches!(event, ManagedEvent::TargetStateChanged { target, lifecycle: ManagedTargetLifecycle::Pending } if *target == new) + })); + assert_eq!(task.history.recent.len(), 1); + assert_eq!(task.history.recent[0], original_outcome); + assert_eq!( + observations + .lock() + .unwrap() + .iter() + .filter(|(event, _)| { + matches!(event, ManagedEvent::TargetFinished { outcome } if outcome.target == old) + }) + .count(), + 1 + ); + + let stop = handle.stop(); + runtime.block_on(task); + runtime.block_on(stop); + server.finish(); +} + #[test] fn dynamic_retiring_active_is_ineligible_before_target_cleanup() { let server = start_server(ServerBehavior::Echo, None); @@ -718,9 +796,11 @@ fn dynamic_replacement_counts_sync_and_async_live_generations() { vec![ManagedTargetConfig::new("same", "127.0.0.1:9")], ) .unwrap(); - let acknowledgement = pending - .apply_targets(vec![ManagedTargetConfig::new("same", "127.0.0.1:10")]) - .unwrap(); + let mut changed_auth = ManagedTargetConfig::new("same", "127.0.0.1:9"); + changed_auth.auth = Some(ClientAuthConfig { + hmac_key: Some(vec![1, 2, 3]), + }); + let acknowledgement = pending.apply_targets(vec![changed_auth]).unwrap(); assert_eq!(acknowledgement.status.targets.len(), 1); assert_eq!(acknowledgement.status.targets[0].target.generation, 2); assert!(matches!(