diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/diagnostics.rs b/backend/crates/gateway-plugin/runtime/tests/generation/diagnostics.rs index 368f33667..df4f96971 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/diagnostics.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/diagnostics.rs @@ -129,18 +129,15 @@ async fn preparing_diagnostics_never_wait_for_the_serialized_prepare_io() { #[tokio::test] async fn diagnostics_report_a_published_process_fault_without_plugin_details() { let (cache, store, runtime) = super::setup().await; + let controls = tempfile::tempdir().unwrap(); + let exit = controls.path().join("exit"); let mut snapshot = store.snapshot.lock().unwrap().clone(); - snapshot.instances[0].configuration = serde_json::json!({"exit_after_ready_ms":20}); + snapshot.instances[0].configuration = serde_json::json!({"exit_after_ready_signals":[exit]}); let active = PluginPreparation::prepare(&runtime, snapshot.clone()) .await .expect("candidate starts before the controlled exit"); - tokio::time::timeout(std::time::Duration::from_secs(2), async { - while active.is_ready() { - tokio::time::sleep(std::time::Duration::from_millis(10)).await; - } - }) - .await - .expect("worker exit observed"); + std::fs::write(exit, b"exit").unwrap(); + wait_until_unready(&active).await; let diagnostics = gateway_admin::ports::plugins::PluginRuntimeDiagnostics::runtime_diagnostics( &runtime, @@ -171,17 +168,20 @@ async fn diagnostics_report_a_published_process_fault_without_plugin_details() { #[tokio::test] async fn a_published_process_crash_preserves_the_serving_snapshot_and_other_plugin_status() { let (cache, store, runtime) = super::setup().await; + let controls = tempfile::tempdir().unwrap(); + let exit = controls.path().join("exit"); let snapshot = { let mut snapshot = store.snapshot.lock().unwrap(); let mut failed = snapshot.instances[0].clone(); failed.id = "crashing-plugin".into(); - failed.configuration = serde_json::json!({"exit_after_ready_ms":100}); + failed.configuration = serde_json::json!({"exit_after_ready_signals":[exit]}); snapshot.instances.push(failed); snapshot.clone() }; let generation = ExtensionPreparationPort::prepare(&runtime, ConfigRevision::new(1).unwrap()) .await .unwrap(); + std::fs::write(exit, b"exit").unwrap(); wait_until_unready(&generation).await; assert!(generation.can_serve()); assert!(!generation.is_ready()); diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/instance.rs b/backend/crates/gateway-plugin/runtime/tests/generation/instance.rs index 2801156e3..92bb61259 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/instance.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/instance.rs @@ -94,29 +94,32 @@ async fn registration_with_a_changed_contribution_id_is_rejected() { async fn restart_circuit_caps_short_lived_crashes_and_a_new_revision_recovers() { let restart_circuit = gateway_plugin_runtime::PluginRestartCircuitConfig { maximum_failures: NonZeroUsize::new(3).unwrap(), - stability_window: Duration::from_secs(1), + // 本用例只验证连续崩溃的预算,不让宿主调度延迟触发稳定运行后的重置 + stability_window: Duration::MAX, }; let (cache, store, runtime) = super::setup_with_restart_circuit(restart_circuit).await; let controls = tempfile::tempdir().unwrap(); let marker = controls.path().join("starts.jsonl"); + let exits = [1, 2, 3].map(|start| controls.path().join(format!("exit-{start}"))); let mut snapshot = store.snapshot.lock().unwrap().clone(); snapshot.instances[0].configuration = serde_json::json!({ "startup_marker": marker, - "exit_after_ready_ms": 20, + "exit_after_ready_signals": exits, }); let mut active = PluginPreparation::prepare(&runtime, snapshot.clone()) .await .expect("first incarnation"); - for expected_start in 2..=3 { + for (index, exit) in exits.iter().enumerate() { + assert_eq!(startup_count(&marker), index + 1); + // Ready 早于注册完成;只有宿主拿到这一代后,才让对应的进程退出 + std::fs::write(exit, b"exit").unwrap(); wait_until_unready(&active).await; - let replacement = PluginPreparation::prepare(&runtime, snapshot.clone()) - .await - .expect("failure budget still permits a restart"); - drop(active); - active = replacement; - assert_eq!(startup_count(&marker), expected_start); + if index + 1 < exits.len() { + active = PluginPreparation::prepare(&runtime, snapshot.clone()) + .await + .expect("failure budget still permits a restart"); + } } - wait_until_unready(&active).await; assert!( PluginPreparation::prepare(&runtime, snapshot.clone()) .await @@ -434,7 +437,6 @@ async fn abandoned_candidate_keeps_shared_process_failure_in_restart_budget() { let mut snapshot = store.snapshot.lock().unwrap().clone(); snapshot.instances[0].configuration = serde_json::json!({ "startup_marker":marker, - "exit_after_ready_delays_ms":[null,null], "exit_after_ready_signals":[old_exit,null], }); let mut other = snapshot.instances[0].clone(); diff --git a/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs b/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs index aaaf857e5..dd63aa976 100644 --- a/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/support/worker/mod.rs @@ -1546,11 +1546,6 @@ async fn main() { resumptions: Mutex::new(BTreeMap::new()), streams: Mutex::new(BTreeMap::new()), }); - let exit_after_ready = peer.configuration["exit_after_ready_delays_ms"] - .as_array() - .and_then(|delays| delays.get(startup)) - .and_then(Value::as_u64) - .or_else(|| peer.configuration["exit_after_ready_ms"].as_u64()); let exit_after_ready_signal = peer.configuration["exit_after_ready_signals"] .as_array() .and_then(|signals| signals.get(startup)) @@ -1571,12 +1566,7 @@ async fn main() { vec![], ) .await; - if let Some(delay) = exit_after_ready { - tokio::spawn(async move { - tokio::time::sleep(Duration::from_millis(delay)).await; - std::process::exit(7); - }); - } else if let Some(signal) = exit_after_ready_signal { + if let Some(signal) = exit_after_ready_signal { tokio::spawn(async move { while !std::path::Path::new(&signal).exists() { tokio::time::sleep(Duration::from_millis(5)).await;