fix: runtime retain restore facts when Stop supersedes publication

2026-10-26 11:50:54 +09:00
parent ceb01f6eaa
commit 9e3f87ac52
18 changed files with 314 additions and 77 deletions
@@ -694,6 +694,9 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
                        server_api::WorkerRestoreState::RolledBack => {
                            runtime_api::WorkerRestoreState::RolledBack
                        }
+                        server_api::WorkerRestoreState::Superseded => {
+                            runtime_api::WorkerRestoreState::Superseded
+                        }
                        server_api::WorkerRestoreState::ReconciliationRequired => {
                            runtime_api::WorkerRestoreState::ReconciliationRequired
                        }
@@ -744,6 +747,9 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
                server_api::WorkerRestoreState::RolledBack => {
                    runtime_api::WorkerRestoreState::RolledBack
                }
+                server_api::WorkerRestoreState::Superseded => {
+                    runtime_api::WorkerRestoreState::Superseded
+                }
                server_api::WorkerRestoreState::ReconciliationRequired => {
                    runtime_api::WorkerRestoreState::ReconciliationRequired
                }
@@ -225,6 +225,28 @@ impl RuntimeWorkerRestoreResult {
        }
    }

+    fn superseded_by_stop(pending: Option<PendingWorkerRestore>) -> Self {
+        let (code, message) = match pending {
+            Some(pending) if pending.commit == RestoreCommitFact::Committed => (
+                "worker_restore_committed_then_stopped",
+                "Restore committed, then its execution was stopped; prior durable effects are not rolled back",
+            ),
+            Some(pending) if pending.commit == RestoreCommitFact::LegacyUnknown => (
+                "worker_restore_unknown_decision_then_stopped",
+                "Restore execution was stopped; its earlier commit decision remains unknown",
+            ),
+            Some(pending) if pending.source_repair_possible => (
+                "worker_restore_source_repair_then_stopped",
+                "Uncommitted Restore was stopped; prior source repair effects may remain",
+            ),
+            _ => (
+                "worker_restore_superseded_by_stop",
+                "Worker Restore was superseded by confirmed Stop",
+            ),
+        };
+        Self::failed(WorkerRestoreState::Superseded, code, message)
+    }
+
    fn into_legacy_result(self) -> Result<WorkerDetail, RuntimeError> {
        match (self.state, self.worker) {
            (WorkerRestoreState::Accepted, Some(worker)) => Ok(worker),
@@ -3684,11 +3706,7 @@ impl Runtime {
            worker.restore_guard.rotate();
            worker
                .restore_guard
-                .finish(RuntimeWorkerRestoreResult::failed(
-                    WorkerRestoreState::RolledBack,
-                    "worker_restore_superseded_by_stop",
-                    "Worker Restore was superseded by Stop",
-                ));
+                .finish(RuntimeWorkerRestoreResult::superseded_by_stop(None));
            state.persist_worker_record(&worker)?;
            let ack = WorkerLifecycleAck {
                worker_ref: worker_ref.clone(),
@@ -4250,6 +4268,7 @@ impl Runtime {
            WorkerRestoreState::Accepted => return Ok(()),
            WorkerRestoreState::Rejected
            | WorkerRestoreState::RolledBack
+            | WorkerRestoreState::Superseded
            | WorkerRestoreState::ReconciliationRequired => {}
        }
        let mut state = self.lock()?;
@@ -4363,6 +4382,8 @@ impl Runtime {
                worker_ref.worker_id
            )));
        }
+        let restore_receipt =
+            RuntimeWorkerRestoreResult::superseded_by_stop(candidate.pending_restore);
        candidate.pending_stop = None;
        candidate.pending_restore = None;
        if let Some(context) = candidate.restore_candidate_context.take() {
@@ -4373,13 +4394,7 @@ impl Runtime {
        candidate.worker_state = None;
        candidate.restore_intent = WorkerRestoreIntent::Explicit;

-        candidate
-            .restore_guard
-            .finish(RuntimeWorkerRestoreResult::failed(
-                WorkerRestoreState::RolledBack,
-                "worker_restore_superseded_by_stop",
-                "Worker Restore was superseded by Stop",
-            ));
+        candidate.restore_guard.finish(restore_receipt);
        candidate.internal_workers.clear();
        state.persist_worker_record(&candidate)?;
        let ack = WorkerLifecycleAck {
@@ -15138,6 +15153,104 @@ mod tests {
        }
    }

+    #[cfg(feature = "fs-store")]
+    #[test]
+    fn stop_retains_restore_decision_evidence_in_superseded_receipt_across_restart() {
+        // Component scope: real Runtime admission and durable store; execution
+        // is substituted. Controller shutdown races are covered separately.
+        for (commit, repaired, code) in [
+            (
+                RestoreCommitFact::NotCommitted,
+                false,
+                "worker_restore_superseded_by_stop",
+            ),
+            (
+                RestoreCommitFact::NotCommitted,
+                true,
+                "worker_restore_source_repair_then_stopped",
+            ),
+            (
+                RestoreCommitFact::Committed,
+                false,
+                "worker_restore_committed_then_stopped",
+            ),
+            (
+                RestoreCommitFact::LegacyUnknown,
+                true,
+                "worker_restore_unknown_decision_then_stopped",
+            ),
+        ] {
+            let dir = tempfile::tempdir().unwrap();
+            let options = FsRuntimeStoreOptions::new(dir.path().join("runtime"))
+                .with_runtime_id("stop-restore-evidence");
+            let backend = Arc::new(TestExecutionBackend::default());
+            let runtime =
+                Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone())
+                    .unwrap();
+            runtime.store_config_bundle(test_bundle()).unwrap();
+            let worker = runtime
+                .create_worker(task_request("stop restore evidence"))
+                .unwrap();
+            runtime.stop_worker(&worker.worker_ref, None).unwrap();
+            let request = runtime.test_restore_request(&worker.worker_ref);
+            *backend.restore_result.lock().unwrap() = Some(WorkerExecutionSpawnResult::Errored(
+                WorkerExecutionResult::errored(
+                    WorkerExecutionOperation::Restore,
+                    "publication not confirmed",
+                ),
+            ));
+            assert_eq!(
+                runtime
+                    .restore_worker_operation(&worker.worker_ref, request.clone())
+                    .unwrap()
+                    .state,
+                WorkerRestoreState::ReconciliationRequired
+            );
+            {
+                // Fixture represents each canonical checkpoint. Stop must keep
+                // its classification, rather than infer rollback from cleanup.
+                let mut state = runtime.lock().unwrap();
+                let record = state.worker_mut(&worker.worker_ref).unwrap();
+                let pending = record.pending_restore.as_mut().unwrap();
+                pending.commit = commit;
+                pending.source_repair_possible = repaired;
+                state.persist_worker(&worker.worker_id).unwrap();
+            }
+            runtime.stop_worker(&worker.worker_ref, None).unwrap();
+            let receipt = runtime
+                .restore_worker_operation(&worker.worker_ref, request.clone())
+                .unwrap();
+            assert_eq!(receipt.state, WorkerRestoreState::Superseded);
+            assert_eq!(receipt.reason_code.as_deref(), Some(code));
+            assert!(receipt.worker.is_none());
+            drop(runtime);
+            let restarted_backend = Arc::new(TestExecutionBackend::default());
+            let restarted =
+                Runtime::with_fs_store_and_execution_backend(options, restarted_backend.clone())
+                    .unwrap();
+            assert_eq!(
+                restarted
+                    .restore_worker_operation(&worker.worker_ref, request.clone())
+                    .unwrap(),
+                receipt
+            );
+            assert_eq!(*restarted_backend.restore_count.lock().unwrap(), 0);
+            assert_eq!(
+                restarted.worker_detail(&worker.worker_ref).unwrap().status,
+                WorkerStatus::Stopped
+            );
+            let mut input = WorkerInput::user("must not revive");
+            input.kind = WorkerInputKind::UserIfIdle;
+            assert!(restarted.send_input(&worker.worker_ref, input).is_err());
+            let mut stale = request;
+            stale.request_id = "new-intent-with-old-token".into();
+            assert!(matches!(
+                restarted.restore_worker_operation(&worker.worker_ref, stale),
+                Err(RuntimeError::RestoreObservationConflict { .. })
+            ));
+        }
+    }
+
    #[cfg(feature = "fs-store")]
    #[test]
    fn restore_decision_reread_preserves_commit_across_uncertainty_stop_and_restart() {
@@ -330,6 +330,22 @@ impl Runtime {
            Err(std::sync::TryLockError::WouldBlock) => return Ok(Step::Contended),
            Err(_) => return Err(RuntimeError::StatePoisoned),
        };
+        let stop_is_owned = {
+            let mut state = self.lock()?;
+            state.ensure_running()?;
+            state.refresh_uncertain_worker_write(worker_ref)?;
+            let worker = state.worker(worker_ref)?;
+            worker.pending_stop.is_some()
+                && worker
+                    .pending_restore
+                    .is_some_and(|pending| pending.operation_id == operation)
+        };
+        if stop_is_owned {
+            // Stop supersedes publication, not ownership of late resources.
+            // Retain this same continuation until its cleanup is confirmed.
+            self.reconcile_pending_stop_under_lock(worker_ref)?;
+            return Ok(Step::Finished);
+        }
        let (backend, pending) = {
            let mut state = self.lock()?;
            state.ensure_running()?;
@@ -94,6 +94,43 @@ mod publication {
        (release, received)
    }

+    #[test]
+    #[serial_test::serial(worker_allocation)]
+    fn stop_timeout_retains_autonomous_late_cleanup_and_blocks_remove_until_confirmed() {
+        let mut fixture = Fixture::no_model();
+        Arc::get_mut(&mut fixture.backend).unwrap().spawn_restore_timeout = Duration::from_millis(250);
+        let runtime = stopped_runtime(&fixture);
+        let pause = runtime.test_pause_restore_continuation();
+        let worker_ref = fixture.request.worker_ref.clone();
+        let scope = fixture.request.workspace_scope.as_ref().unwrap();
+        let request = runtime.test_restore_request(&worker_ref);
+        let before = fixture.canonical_files();
+        let (release, prepared) = prepared_gate(&fixture);
+        let result = runtime.restore_worker_operation_scoped(scope, &worker_ref, request.clone()).unwrap();
+        prepared.recv_timeout(WAIT).unwrap();
+        assert_eq!(result.state, server_api::WorkerRestoreState::ReconciliationRequired);
+        let operation = durable_operation(&runtime, &worker_ref).unwrap();
+        assert!(runtime.stop_worker_scoped(scope, &worker_ref, None).is_err());
+        assert!(runtime.delete_worker_scoped(scope, &worker_ref).is_err());
+        assert!(fixture.backend.workers.lock().unwrap().is_empty());
+        release.notify_one();
+        drop(pause);
+        runtime.test_wait_restore_continuation(&worker_ref, operation.operation_id, WAIT);
+        assert_eq!(runtime.worker_detail(&worker_ref).unwrap().status, crate::catalog::WorkerStatus::Stopped);
+        assert!(durable_operation(&runtime, &worker_ref).is_none());
+        assert!(fixture.backend.pending_restore(&worker_ref).unwrap().is_none());
+        assert!(fixture.backend.workers.lock().unwrap().is_empty());
+        assert!(fixture.backend.factory.observation_hub.get(&worker_ref).is_none());
+        assert!(fixture.backend.clone().attach_worker_protocol(&worker_ref).is_err());
+        let receipt = runtime.restore_worker_operation_scoped(scope, &worker_ref, request).unwrap();
+        assert_eq!(receipt.state, server_api::WorkerRestoreState::Superseded);
+        assert_eq!(receipt.reason_code.as_deref(), Some("worker_restore_superseded_by_stop"));
+        assert_eq!(fixture.canonical_files(), before);
+        fixture.assert_released();
+        fixture.assert_quiet();
+        assert!(runtime.delete_worker_scoped(scope, &worker_ref).unwrap().deleted);
+    }
+
    #[test]
    #[serial_test::serial(worker_allocation)]
    fn autonomous_late_prepare_and_publication_settle_without_another_request() {