From 2cd8a7d60d4d8376ce5d447fd2260643b4014c52 Mon Sep 17 00:00:00 2001 From: Nils Andreas Svee Date: Sun, 4 Oct 2026 21:48:20 +0700 Subject: [PATCH] refactor(client): make terminal target outcomes authoritative --- crates/irtt-client/src/managed/model.rs | 6 ++ crates/irtt-client/src/managed/task.rs | 58 +++++++++++-------- .../tests/managed_behavior.rs | 11 +++- .../tests/managed_lifecycle.rs | 32 +++++++++- 4 files changed, 82 insertions(+), 25 deletions(-) diff --git a/crates/irtt-client/src/managed/model.rs b/crates/irtt-client/src/managed/model.rs index 17e824a..4056717 100644 --- a/crates/irtt-client/src/managed/model.rs +++ b/crates/irtt-client/src/managed/model.rs @@ -141,6 +141,12 @@ pub struct ManagedTargetStatus { pub target: TargetInstance, pub desired: bool, pub lifecycle: ManagedTargetLifecycle, + /// Authoritative completed accounting for this target generation. + /// + /// Present if and only if `lifecycle` is [`ManagedTargetLifecycle::Terminal`]. + /// This shares the terminal phase's outcome with [`ManagedEvent::TargetFinished`], + /// so terminal details remain durable even when presentation events are lost. + pub outcome: Option>, pub server_addr: Arc, pub remote: Option, } diff --git a/crates/irtt-client/src/managed/task.rs b/crates/irtt-client/src/managed/task.rs index ee84970..98e40e9 100644 --- a/crates/irtt-client/src/managed/task.rs +++ b/crates/irtt-client/src/managed/task.rs @@ -369,7 +369,10 @@ enum TargetPhase { primary_end: ManagedTargetEndReason, cleanup_failure: Option, }, - Terminal, + /// Terminal state always owns its finalized authoritative outcome. + Terminal { + outcome: Arc, + }, } impl TargetPhase { @@ -381,7 +384,7 @@ impl TargetPhase { Self::Active { .. } => ManagedTargetLifecycle::Active, Self::Draining { .. } => ManagedTargetLifecycle::Draining, Self::Closing { .. } => ManagedTargetLifecycle::Closing, - Self::Terminal => ManagedTargetLifecycle::Terminal, + Self::Terminal { .. } => ManagedTargetLifecycle::Terminal, } } @@ -589,9 +592,9 @@ struct UpdatePlan { fn synchronously_retireable(phase: &TargetPhase) -> bool { match phase { - TargetPhase::Pending { .. } | TargetPhase::Connecting { .. } | TargetPhase::Terminal => { - true - } + TargetPhase::Pending { .. } + | TargetPhase::Connecting { .. } + | TargetPhase::Terminal { .. } => true, TargetPhase::Opening { open, .. } => !open.has_in_flight_work(), TargetPhase::Active { .. } | TargetPhase::Draining { .. } | TargetPhase::Closing { .. } => { false @@ -601,7 +604,7 @@ fn synchronously_retireable(phase: &TargetPhase) -> bool { fn runtime_satisfies_target(runtime: &TargetRuntime, target: &ManagedTargetConfig) -> bool { runtime.membership.is_desired() - && !matches!(runtime.phase(), TargetPhase::Terminal) + && !matches!(runtime.phase(), TargetPhase::Terminal { .. }) && runtime.config == *target } @@ -766,6 +769,10 @@ impl ManagedClientTask { target: target.instance.clone(), desired: target.membership.is_desired(), lifecycle, + outcome: match target.phase() { + TargetPhase::Terminal { outcome } => Some(Arc::clone(outcome)), + _ => None, + }, server_addr: Arc::clone(&target.server_addr), remote: target.remote, } @@ -841,13 +848,17 @@ impl ManagedClientTask { end_reason: ManagedTargetEndReason, cleanup_failure: Option, ) { - self.install_target_phase(index, TargetPhase::Terminal); - let outcome = self.targets[index].outcome(end_reason, cleanup_failure); - self.history.record(outcome.clone()); - self.replace_status(); - self.publish_event(ManagedEvent::TargetFinished { - outcome: Arc::new(outcome), - }); + let outcome = Arc::new(self.targets[index].outcome(end_reason, cleanup_failure)); + // Finalize history before the phase publisher installs Terminal, updates + // pacing and publishes one complete status followed by the lifecycle event. + self.history.record(outcome.as_ref().clone()); + self.install_target_phase( + index, + TargetPhase::Terminal { + outcome: Arc::clone(&outcome), + }, + ); + self.publish_event(ManagedEvent::TargetFinished { outcome }); } fn fail_target(&mut self, index: usize, phase: ManagedTargetFailurePhase, error: ClientError) { @@ -955,12 +966,10 @@ impl ManagedClientTask { target.membership = TargetMembership::Withdrawn(retirement.reason); if retirement.synchronous { synchronous.insert(retirement.index); - if !matches!(target.phase(), TargetPhase::Terminal) { - let outcome = target.outcome(retirement.reason.into(), None); - self.history.record(outcome.clone()); - finished.push(ManagedEvent::TargetFinished { - outcome: Arc::new(outcome), - }); + if !matches!(target.phase(), TargetPhase::Terminal { .. }) { + let outcome = Arc::new(target.outcome(retirement.reason.into(), None)); + self.history.record(outcome.as_ref().clone()); + finished.push(ManagedEvent::TargetFinished { outcome }); } } else if was_paced_active { // Removal is directional: it may discard an elapsed gate but never @@ -1130,12 +1139,14 @@ impl ManagedClientTask { fn prune_undesired_terminal(&mut self) { if !self.targets.iter().any(|target| { - !target.membership.is_desired() && matches!(target.phase(), TargetPhase::Terminal) + !target.membership.is_desired() + && matches!(target.phase(), TargetPhase::Terminal { .. }) }) { return; } self.targets.retain(|target| { - target.membership.is_desired() || !matches!(target.phase(), TargetPhase::Terminal) + target.membership.is_desired() + || !matches!(target.phase(), TargetPhase::Terminal { .. }) }); self.rebase_target_cursors(); self.replace_status(); @@ -1505,7 +1516,7 @@ impl ManagedClientTask { } } } - TargetPhase::Terminal => false, + TargetPhase::Terminal { .. } => false, } } @@ -2039,7 +2050,7 @@ impl ManagedClientTask { fn all_targets_terminal(&self) -> bool { self.targets .iter() - .all(|target| matches!(target.phase(), TargetPhase::Terminal)) + .all(|target| matches!(target.phase(), TargetPhase::Terminal { .. })) } fn next_deadline(&self) -> Option { @@ -2398,6 +2409,7 @@ fn build_task( target: target.instance.clone(), desired: target.membership.is_desired(), lifecycle: ManagedTargetLifecycle::Pending, + outcome: None, server_addr: Arc::clone(&target.server_addr), remote: None, }) diff --git a/crates/irtt-integration-tests/tests/managed_behavior.rs b/crates/irtt-integration-tests/tests/managed_behavior.rs index 9785da5..e58be0d 100644 --- a/crates/irtt-integration-tests/tests/managed_behavior.rs +++ b/crates/irtt-integration-tests/tests/managed_behavior.rs @@ -1,6 +1,6 @@ mod support; -use std::{collections::HashMap, time::Duration}; +use std::{collections::HashMap, sync::Arc, time::Duration}; use irtt_client::{managed::*, ClientConfig, ClientEvent, OpenPolicy, SessionRequest}; use irtt_server::ServerConfig; @@ -186,6 +186,9 @@ async fn target_updates_preserve_identical_sessions_and_retire_removed_generatio ); assert!(status.targets.iter().all(|target| target.target != first || target.lifecycle == ManagedTargetLifecycle::Terminal)); + if let Some(target) = status.targets.iter().find(|target| target.target == first) { + assert!(Arc::ptr_eq(target.outcome.as_ref().unwrap(), &outcome)); + } break; } } @@ -224,6 +227,12 @@ async fn target_updates_preserve_identical_sessions_and_retire_removed_generatio ); assert!(status.targets.iter().all(|target| target.target != second || target.lifecycle == ManagedTargetLifecycle::Terminal)); + let target = status + .targets + .iter() + .find(|target| target.target == second) + .unwrap(); + assert!(Arc::ptr_eq(target.outcome.as_ref().unwrap(), &outcome)); } ManagedEvent::Completed { outcome } => { let status = handle.status(); diff --git a/crates/irtt-integration-tests/tests/managed_lifecycle.rs b/crates/irtt-integration-tests/tests/managed_lifecycle.rs index 6fcee0e..061db3e 100644 --- a/crates/irtt-integration-tests/tests/managed_lifecycle.rs +++ b/crates/irtt-integration-tests/tests/managed_lifecycle.rs @@ -454,6 +454,12 @@ fn event_loss_independence() { let mut config = config(ManagedPacing::Staggered); config.event_capacity = 1; let (task, handle) = ManagedClient::task(config, vec![target("one", server.addr)]).unwrap(); + let initial = handle.status(); + assert_eq!( + initial.targets[0].lifecycle, + ManagedTargetLifecycle::Pending + ); + assert!(initial.targets[0].outcome.is_none()); let mut events = handle.subscribe().unwrap(); let outcome = runtime().block_on(task); assert!(matches!( @@ -461,7 +467,31 @@ fn event_loss_independence() { Err(broadcast::error::TryRecvError::Lagged(_)) )); assert_eq!(outcome.successful_target_outcomes, 1); - assert_eq!(handle.status().lifecycle, ManagedLifecycle::Completed); + let status = handle.status(); + assert_eq!(status.lifecycle, ManagedLifecycle::Completed); + assert_eq!(status.targets.len(), 1); + let target = &status.targets[0]; + assert!(target.desired); + assert_eq!(target.lifecycle, ManagedTargetLifecycle::Terminal); + let terminal = target + .outcome + .as_ref() + .expect("terminal details are durable"); + assert_eq!(terminal.target, target.target); + assert_eq!(status.terminal_target_count, 1); + assert_eq!(status.total_target_outcomes, 1); + assert_eq!(status.successful_target_outcomes, 1); + assert_eq!(status.failed_target_outcomes, 0); + assert_eq!(status.peer_closed_target_outcomes, 0); + assert_eq!(status.discarded_target_outcomes, 0); + assert_eq!( + status.recent_target_outcomes.as_ref(), + &[terminal.as_ref().clone()] + ); + assert_eq!( + status.recent_target_outcomes, + outcome.recent_target_outcomes + ); server.finish(); }