test(plugin): signal crashes after host registration

Drive each worker exit from the prepared generation so scheduling delays cannot turn a post-registration crash into a registration failure.

Co-authored-by: Codex <noreply@openai.com>

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());

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();

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;