diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index 8869484b4..c2fb52bf7 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -523,6 +523,7 @@ pub(crate) struct PersistedWorkerRecord { pub(crate) config_bundle: Option, pub(crate) created_at_ms: Option, pub(crate) status: WorkerStatus, + pub(crate) last_finished_submission_request_id: Option, pub(crate) execution_state: PersistedWorkerExecutionState, pub(crate) workspace_id: Option, pub(crate) workdir_attachments: Vec, @@ -1824,6 +1825,8 @@ struct WorkerAggregateRecord { created_at_ms: Option, status: WorkerStatus, #[serde(default, skip_serializing_if = "Option::is_none")] + last_finished_submission_request_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] workspace_id: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] workdir_attachments: Vec, @@ -1874,6 +1877,7 @@ impl WorkerAggregateRecord { config_bundle: worker.config_bundle.clone(), created_at_ms: worker.created_at_ms, status: worker.status, + last_finished_submission_request_id: worker.last_finished_submission_request_id.clone(), workspace_id: worker.workspace_id.clone(), workdir_attachments: worker.workdir_attachments.clone(), logical_workdir_attachments: worker.logical_workdir_attachments.clone(), @@ -1893,6 +1897,7 @@ impl WorkerAggregateRecord { } fn validate(self, path: &Path) -> Result { + let last_finished_submission_request_id = self.last_finished_submission_request_id; let identity = WorkerIdentityRecord { schema_version: self.schema_version, worker_ref: self.worker_ref, @@ -1984,7 +1989,9 @@ impl WorkerAggregateRecord { } WorkerExecutionStateRecord::Unavailable => PersistedWorkerExecutionState::Unavailable, }; - Ok(identity.into_persisted(execution_state)) + let mut persisted = identity.into_persisted(execution_state); + persisted.last_finished_submission_request_id = last_finished_submission_request_id; + Ok(persisted) } } @@ -2164,6 +2171,7 @@ impl WorkerIdentityRecord { config_bundle: self.config_bundle, created_at_ms: self.created_at_ms, status: self.status, + last_finished_submission_request_id: None, execution_state, workspace_id: self.workspace_id, workdir_attachments, diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index b1683d286..a2a937e6f 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1095,6 +1095,7 @@ impl Runtime { worker_id: worker_id.clone(), status: WorkerStatus::Stopped, worker_state: None, + last_finished_submission_request_id: None, workspace_id: scope.map(|scope| scope.workspace_id.clone()), profile: durable_request.profile.clone(), display_name: durable_request.display_name.clone(), @@ -3419,6 +3420,16 @@ impl Runtime { candidate.pending_stop = None; candidate.restore_candidate_context = None; candidate.status = status; + let mut worker_state = worker_state; + if worker_state.last_finished_submission_request_id.is_none() { + // The restored controller starts with fresh live state, but this + // historical fence proves that a previously accepted Submit Run + // already returned. Preserve it across process reconstruction so + // durable consumers can finish post-restart cleanup without + // waiting for an old request to run again. + worker_state.last_finished_submission_request_id = + candidate.last_finished_submission_request_id.clone(); + } let _ = candidate.apply_worker_state(&worker_state); candidate.restore_intent = restore_intent_for_status(candidate.status); candidate.workdir_attachments = workdir_attachments; @@ -4029,6 +4040,7 @@ impl RuntimeState { worker_id: worker.worker_id, status: worker.status, worker_state: None, + last_finished_submission_request_id: worker.last_finished_submission_request_id, workspace_id: worker.workspace_id, profile: worker.profile, display_name: worker.display_name, @@ -4926,6 +4938,7 @@ struct WorkerRecord { worker_id: WorkerId, status: WorkerStatus, worker_state: Option, + last_finished_submission_request_id: Option, workspace_id: Option, profile: ProfileSelector, display_name: Option, @@ -4947,6 +4960,9 @@ struct WorkerRecord { impl WorkerRecord { fn apply_worker_state(&mut self, incoming: &protocol::WorkerStateSnapshot) { + if let Some(request_id) = &incoming.last_finished_submission_request_id { + self.last_finished_submission_request_id = Some(request_id.clone()); + } self.worker_state = Some(incoming.clone()); } @@ -5055,6 +5071,7 @@ impl WorkerRecord { config_bundle: self.config_bundle.clone(), created_at_ms: self.created_at_ms, status: self.status, + last_finished_submission_request_id: self.last_finished_submission_request_id.clone(), execution_state, workspace_id: self.workspace_id.clone(), workdir_attachments: self.workdir_attachments.clone(), @@ -9231,6 +9248,75 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[cfg(feature = "fs-store")] + #[test] + fn fs_store_preserves_finished_submission_fence_across_active_restore() { + let root = fs_store_root("finished-submission-restore"); + let options = crate::fs_store::FsRuntimeStoreOptions { + root: root.clone(), + runtime_id: "test-runtime".to_string(), + display_name: None, + }; + let execution = Arc::new(TestExecutionBackend::default()); + let runtime = + Runtime::with_fs_store_and_execution_backend(options.clone(), execution.clone()) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let worker = runtime + .create_worker(task_request("persist finished submission fence")) + .unwrap(); + execution.set_dispatch_result( + WorkerExecutionResult::accepted_submission( + WorkerExecutionOperation::Input, + "job-attempt-request", + "submission-id", + protocol::SubmissionDisposition::Started, + ) + .with_worker_state(protocol::WorkerStateSnapshot { + state: protocol::WorkerState::Idle, + // Production Submit Run does not consume a lifecycle command + // id, so this intentionally remains zero. + last_command_id: 0, + last_finished_submission_request_id: Some("job-attempt-request".to_string()), + }), + ); + let mut input = WorkerInput::user("run durable Backend Job attempt"); + input.submission_request_id = Some("job-attempt-request".to_string()); + runtime.send_input(&worker.worker_ref, input).unwrap(); + + let aggregate_path = root + .join("workers") + .join(worker.worker_id.to_string()) + .join("worker.json"); + let aggregate: serde_json::Value = + serde_json::from_slice(&std::fs::read(&aggregate_path).unwrap()).unwrap(); + assert_eq!( + aggregate["last_finished_submission_request_id"], + serde_json::json!("job-attempt-request") + ); + drop(runtime); + + let restored = Runtime::with_fs_store_and_execution_backend( + options, + Arc::new(TestExecutionBackend::default()), + ) + .unwrap(); + let restored_state = restored + .worker_detail(&worker.worker_ref) + .unwrap() + .worker_state + .expect("automatic restore publishes fresh Worker state"); + assert_eq!(restored_state.last_command_id, 0); + assert_eq!( + restored_state + .last_finished_submission_request_id + .as_deref(), + Some("job-attempt-request") + ); + + let _ = std::fs::remove_dir_all(root); + } + #[cfg(feature = "fs-store")] #[test] fn fs_store_explicitly_retries_rejected_automatic_restore() {