refactor(client): make terminal target outcomes authoritative
This commit is contained in:
4 files changed
+82
-25
No files matched your search
@@ -141,6 +141,12 @@ pub struct ManagedTargetStatus {
|
|||||||
pub target: TargetInstance,
|
pub target: TargetInstance,
|
||||||
pub desired: bool,
|
pub desired: bool,
|
||||||
pub lifecycle: ManagedTargetLifecycle,
|
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<Arc<ManagedTargetOutcome>>,
|
||||||
pub server_addr: Arc<str>,
|
pub server_addr: Arc<str>,
|
||||||
pub remote: Option<SocketAddr>,
|
pub remote: Option<SocketAddr>,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -369,7 +369,10 @@ enum TargetPhase {
|
|||||||
primary_end: ManagedTargetEndReason,
|
primary_end: ManagedTargetEndReason,
|
||||||
cleanup_failure: Option<ManagedTargetFailure>,
|
cleanup_failure: Option<ManagedTargetFailure>,
|
||||||
},
|
},
|
||||||
Terminal,
|
/// Terminal state always owns its finalized authoritative outcome.
|
||||||
|
Terminal {
|
||||||
|
outcome: Arc<ManagedTargetOutcome>,
|
||||||
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
impl TargetPhase {
|
impl TargetPhase {
|
||||||
@@ -381,7 +384,7 @@ impl TargetPhase {
|
|||||||
Self::Active { .. } => ManagedTargetLifecycle::Active,
|
Self::Active { .. } => ManagedTargetLifecycle::Active,
|
||||||
Self::Draining { .. } => ManagedTargetLifecycle::Draining,
|
Self::Draining { .. } => ManagedTargetLifecycle::Draining,
|
||||||
Self::Closing { .. } => ManagedTargetLifecycle::Closing,
|
Self::Closing { .. } => ManagedTargetLifecycle::Closing,
|
||||||
Self::Terminal => ManagedTargetLifecycle::Terminal,
|
Self::Terminal { .. } => ManagedTargetLifecycle::Terminal,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -589,9 +592,9 @@ struct UpdatePlan {
|
|||||||
|
|
||||||
fn synchronously_retireable(phase: &TargetPhase) -> bool {
|
fn synchronously_retireable(phase: &TargetPhase) -> bool {
|
||||||
match phase {
|
match phase {
|
||||||
TargetPhase::Pending { .. } | TargetPhase::Connecting { .. } | TargetPhase::Terminal => {
|
TargetPhase::Pending { .. }
|
||||||
true
|
| TargetPhase::Connecting { .. }
|
||||||
}
|
| TargetPhase::Terminal { .. } => true,
|
||||||
TargetPhase::Opening { open, .. } => !open.has_in_flight_work(),
|
TargetPhase::Opening { open, .. } => !open.has_in_flight_work(),
|
||||||
TargetPhase::Active { .. } | TargetPhase::Draining { .. } | TargetPhase::Closing { .. } => {
|
TargetPhase::Active { .. } | TargetPhase::Draining { .. } | TargetPhase::Closing { .. } => {
|
||||||
false
|
false
|
||||||
@@ -601,7 +604,7 @@ fn synchronously_retireable(phase: &TargetPhase) -> bool {
|
|||||||
|
|
||||||
fn runtime_satisfies_target(runtime: &TargetRuntime, target: &ManagedTargetConfig) -> bool {
|
fn runtime_satisfies_target(runtime: &TargetRuntime, target: &ManagedTargetConfig) -> bool {
|
||||||
runtime.membership.is_desired()
|
runtime.membership.is_desired()
|
||||||
&& !matches!(runtime.phase(), TargetPhase::Terminal)
|
&& !matches!(runtime.phase(), TargetPhase::Terminal { .. })
|
||||||
&& runtime.config == *target
|
&& runtime.config == *target
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -766,6 +769,10 @@ impl ManagedClientTask {
|
|||||||
target: target.instance.clone(),
|
target: target.instance.clone(),
|
||||||
desired: target.membership.is_desired(),
|
desired: target.membership.is_desired(),
|
||||||
lifecycle,
|
lifecycle,
|
||||||
|
outcome: match target.phase() {
|
||||||
|
TargetPhase::Terminal { outcome } => Some(Arc::clone(outcome)),
|
||||||
|
_ => None,
|
||||||
|
},
|
||||||
server_addr: Arc::clone(&target.server_addr),
|
server_addr: Arc::clone(&target.server_addr),
|
||||||
remote: target.remote,
|
remote: target.remote,
|
||||||
}
|
}
|
||||||
@@ -841,13 +848,17 @@ impl ManagedClientTask {
|
|||||||
end_reason: ManagedTargetEndReason,
|
end_reason: ManagedTargetEndReason,
|
||||||
cleanup_failure: Option<ManagedTargetFailure>,
|
cleanup_failure: Option<ManagedTargetFailure>,
|
||||||
) {
|
) {
|
||||||
self.install_target_phase(index, TargetPhase::Terminal);
|
let outcome = Arc::new(self.targets[index].outcome(end_reason, cleanup_failure));
|
||||||
let outcome = self.targets[index].outcome(end_reason, cleanup_failure);
|
// Finalize history before the phase publisher installs Terminal, updates
|
||||||
self.history.record(outcome.clone());
|
// pacing and publishes one complete status followed by the lifecycle event.
|
||||||
self.replace_status();
|
self.history.record(outcome.as_ref().clone());
|
||||||
self.publish_event(ManagedEvent::TargetFinished {
|
self.install_target_phase(
|
||||||
outcome: Arc::new(outcome),
|
index,
|
||||||
});
|
TargetPhase::Terminal {
|
||||||
|
outcome: Arc::clone(&outcome),
|
||||||
|
},
|
||||||
|
);
|
||||||
|
self.publish_event(ManagedEvent::TargetFinished { outcome });
|
||||||
}
|
}
|
||||||
|
|
||||||
fn fail_target(&mut self, index: usize, phase: ManagedTargetFailurePhase, error: ClientError) {
|
fn fail_target(&mut self, index: usize, phase: ManagedTargetFailurePhase, error: ClientError) {
|
||||||
@@ -955,12 +966,10 @@ impl ManagedClientTask {
|
|||||||
target.membership = TargetMembership::Withdrawn(retirement.reason);
|
target.membership = TargetMembership::Withdrawn(retirement.reason);
|
||||||
if retirement.synchronous {
|
if retirement.synchronous {
|
||||||
synchronous.insert(retirement.index);
|
synchronous.insert(retirement.index);
|
||||||
if !matches!(target.phase(), TargetPhase::Terminal) {
|
if !matches!(target.phase(), TargetPhase::Terminal { .. }) {
|
||||||
let outcome = target.outcome(retirement.reason.into(), None);
|
let outcome = Arc::new(target.outcome(retirement.reason.into(), None));
|
||||||
self.history.record(outcome.clone());
|
self.history.record(outcome.as_ref().clone());
|
||||||
finished.push(ManagedEvent::TargetFinished {
|
finished.push(ManagedEvent::TargetFinished { outcome });
|
||||||
outcome: Arc::new(outcome),
|
|
||||||
});
|
|
||||||
}
|
}
|
||||||
} else if was_paced_active {
|
} else if was_paced_active {
|
||||||
// Removal is directional: it may discard an elapsed gate but never
|
// Removal is directional: it may discard an elapsed gate but never
|
||||||
@@ -1130,12 +1139,14 @@ impl ManagedClientTask {
|
|||||||
|
|
||||||
fn prune_undesired_terminal(&mut self) {
|
fn prune_undesired_terminal(&mut self) {
|
||||||
if !self.targets.iter().any(|target| {
|
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;
|
return;
|
||||||
}
|
}
|
||||||
self.targets.retain(|target| {
|
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.rebase_target_cursors();
|
||||||
self.replace_status();
|
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 {
|
fn all_targets_terminal(&self) -> bool {
|
||||||
self.targets
|
self.targets
|
||||||
.iter()
|
.iter()
|
||||||
.all(|target| matches!(target.phase(), TargetPhase::Terminal))
|
.all(|target| matches!(target.phase(), TargetPhase::Terminal { .. }))
|
||||||
}
|
}
|
||||||
|
|
||||||
fn next_deadline(&self) -> Option<Instant> {
|
fn next_deadline(&self) -> Option<Instant> {
|
||||||
@@ -2398,6 +2409,7 @@ fn build_task(
|
|||||||
target: target.instance.clone(),
|
target: target.instance.clone(),
|
||||||
desired: target.membership.is_desired(),
|
desired: target.membership.is_desired(),
|
||||||
lifecycle: ManagedTargetLifecycle::Pending,
|
lifecycle: ManagedTargetLifecycle::Pending,
|
||||||
|
outcome: None,
|
||||||
server_addr: Arc::clone(&target.server_addr),
|
server_addr: Arc::clone(&target.server_addr),
|
||||||
remote: None,
|
remote: None,
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
mod support;
|
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_client::{managed::*, ClientConfig, ClientEvent, OpenPolicy, SessionRequest};
|
||||||
use irtt_server::ServerConfig;
|
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
|
assert!(status.targets.iter().all(|target| target.target != first
|
||||||
|| target.lifecycle == ManagedTargetLifecycle::Terminal));
|
|| 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;
|
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
|
assert!(status.targets.iter().all(|target| target.target != second
|
||||||
|| target.lifecycle == ManagedTargetLifecycle::Terminal));
|
|| 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 } => {
|
ManagedEvent::Completed { outcome } => {
|
||||||
let status = handle.status();
|
let status = handle.status();
|
||||||
|
|||||||
@@ -454,6 +454,12 @@ fn event_loss_independence() {
|
|||||||
let mut config = config(ManagedPacing::Staggered);
|
let mut config = config(ManagedPacing::Staggered);
|
||||||
config.event_capacity = 1;
|
config.event_capacity = 1;
|
||||||
let (task, handle) = ManagedClient::task(config, vec![target("one", server.addr)]).unwrap();
|
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 mut events = handle.subscribe().unwrap();
|
||||||
let outcome = runtime().block_on(task);
|
let outcome = runtime().block_on(task);
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
@@ -461,7 +467,31 @@ fn event_loss_independence() {
|
|||||||
Err(broadcast::error::TryRecvError::Lagged(_))
|
Err(broadcast::error::TryRecvError::Lagged(_))
|
||||||
));
|
));
|
||||||
assert_eq!(outcome.successful_target_outcomes, 1);
|
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();
|
server.finish();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user