fix(managed): restart resubmitted terminal targets #109
No files matched your search
@@ -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)]
|
#[derive(Default)]
|
||||||
struct OutcomeHistory {
|
struct OutcomeHistory {
|
||||||
limit: usize,
|
limit: usize,
|
||||||
@@ -996,7 +1000,11 @@ impl ManagedClientTask {
|
|||||||
}
|
}
|
||||||
let mut retirements = Vec::new();
|
let mut retirements = Vec::new();
|
||||||
for (index, runtime) in self.targets.iter().enumerate() {
|
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;
|
continue;
|
||||||
}
|
}
|
||||||
let reason = if prepared
|
let reason = if prepared
|
||||||
@@ -1020,7 +1028,7 @@ impl ManagedClientTask {
|
|||||||
if self
|
if self
|
||||||
.targets
|
.targets
|
||||||
.iter()
|
.iter()
|
||||||
.any(|runtime| runtime.desired && runtime.config == target)
|
.any(|runtime| runtime_satisfies_target(runtime, &target))
|
||||||
{
|
{
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ enum ServerBehavior {
|
|||||||
Echo,
|
Echo,
|
||||||
NoTest,
|
NoTest,
|
||||||
PeerClose,
|
PeerClose,
|
||||||
|
InvalidOpen,
|
||||||
DelayedEcho(Duration),
|
DelayedEcho(Duration),
|
||||||
DeferredBurst(usize),
|
DeferredBurst(usize),
|
||||||
NoisyOpen(usize),
|
NoisyOpen(usize),
|
||||||
@@ -153,6 +154,10 @@ fn start_server_with_gates(
|
|||||||
if let Some(gate) = &open_gate {
|
if let Some(gate) = &open_gate {
|
||||||
gate.arrive_and_wait();
|
gate.arrive_and_wait();
|
||||||
}
|
}
|
||||||
|
if matches!(behavior, ServerBehavior::InvalidOpen) {
|
||||||
|
socket.send_to(&[0_u8], peer).unwrap();
|
||||||
|
return;
|
||||||
|
}
|
||||||
if let Some(interval) = interval {
|
if let Some(interval) = interval {
|
||||||
negotiated.interval_ns = i64::try_from(interval.as_nanos()).unwrap();
|
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();
|
let acknowledgement = runtime.block_on(receipt).unwrap();
|
||||||
assert_eq!(acknowledgement.status.targets[0].target, instance);
|
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| {
|
drive_task_until(&runtime, &mut task, |task| {
|
||||||
task.targets[0].counters.packets_sent > sent
|
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]
|
#[test]
|
||||||
fn dynamic_retiring_active_is_ineligible_before_target_cleanup() {
|
fn dynamic_retiring_active_is_ineligible_before_target_cleanup() {
|
||||||
let server = start_server(ServerBehavior::Echo, None);
|
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")],
|
vec![ManagedTargetConfig::new("same", "127.0.0.1:9")],
|
||||||
)
|
)
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let acknowledgement = pending
|
let mut changed_auth = ManagedTargetConfig::new("same", "127.0.0.1:9");
|
||||||
.apply_targets(vec![ManagedTargetConfig::new("same", "127.0.0.1:10")])
|
changed_auth.auth = Some(ClientAuthConfig {
|
||||||
.unwrap();
|
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.len(), 1);
|
||||||
assert_eq!(acknowledgement.status.targets[0].target.generation, 2);
|
assert_eq!(acknowledgement.status.targets[0].target.generation, 2);
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
|
|||||||
Reference in new issue
Block a user