From 63951398f001f2c54aaaa00aeb1509c9e3da3fd0 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 1 Oct 2026 12:26:30 +0900 Subject: [PATCH] fix durable job notification receipts --- crates/protocol/src/lib.rs | 38 ++++ crates/tui/src/app.rs | 5 +- crates/worker-runtime/src/execution.rs | 30 ++++ crates/worker-runtime/src/interaction.rs | 5 +- crates/worker-runtime/src/runtime.rs | 80 +++++++-- crates/worker-runtime/src/worker_backend.rs | 88 ++++++++- crates/worker/src/controller.rs | 190 +++++++++++--------- crates/workspace-server/src/backend_job.rs | 43 +++++ crates/workspace-server/src/server.rs | 33 ++-- docs/design/durable-operations.md | 15 +- 10 files changed, 408 insertions(+), 119 deletions(-) diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 05c9f37ae..53512b526 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -1005,6 +1005,18 @@ pub enum Event { submission_request_id: String, message: String, }, + /// Durable notification acceptance. The request identity is persisted with + /// the notification receipt before this event is emitted. Repeating the + /// same request id and exact payload returns the same acknowledgement + /// without appending the notification twice. + NotificationAccepted { + notification_request_id: String, + }, + /// Correlated notification rejection before durable acceptance. + NotificationRejected { + notification_request_id: String, + message: String, + }, /// Revisioned FIFO replacement following enqueue, activation, cancel, or clear. PendingSubmissionsChanged { pending: PendingSubmissionsSnapshot, @@ -2809,6 +2821,32 @@ mod tests { } } + #[test] + fn notification_receipt_events_roundtrip_with_request_identity() { + for event in [ + Event::NotificationAccepted { + notification_request_id: "notification-1".into(), + }, + Event::NotificationRejected { + notification_request_id: "notification-1".into(), + message: "invalid notification".into(), + }, + ] { + let json = serde_json::to_string(&event).unwrap(); + let decoded: Event = serde_json::from_str(&json).unwrap(); + match decoded { + Event::NotificationAccepted { + notification_request_id, + } + | Event::NotificationRejected { + notification_request_id, + .. + } => assert_eq!(notification_request_id, "notification-1"), + other => panic!("expected notification receipt, got {other:?}"), + } + } + } + #[test] fn event_error_format() { let event = Event::Error { diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index 3ab3e5123..91b382945 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -1143,8 +1143,9 @@ impl App { } match event { - Event::SubmissionAccepted { .. } => {} - Event::SubmissionRejected { message, .. } => self.push_error(message), + Event::SubmissionAccepted { .. } | Event::NotificationAccepted { .. } => {} + Event::SubmissionRejected { message, .. } + | Event::NotificationRejected { message, .. } => self.push_error(message), Event::PendingSubmissionsChanged { pending } => { self.pending_submissions = pending; } diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 01b95521b..82881d2d5 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -66,6 +66,12 @@ pub struct WorkerSubmissionAck { pub disposition: protocol::SubmissionDisposition, } +/// Evidence that a notification request reached the durable Worker session boundary. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkerNotificationAck { + pub notification_request_id: String, +} + /// Typed execution result class. Results are transient operation outcomes and /// are not persisted as Worker lifecycle authority. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -78,6 +84,8 @@ pub struct WorkerExecutionResult { pub message: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub submission: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub notification: Option, } /// Backend result class for a Worker execution operation. @@ -99,6 +107,7 @@ impl WorkerExecutionResult { worker_state: None, message: None, submission: None, + notification: None, } } @@ -123,6 +132,23 @@ impl WorkerExecutionResult { submission_id: submission_id.into(), disposition, }), + notification: None, + } + } + + pub fn accepted_notification( + operation: WorkerExecutionOperation, + notification_request_id: impl Into, + ) -> Self { + Self { + operation, + outcome: WorkerExecutionOutcome::Accepted, + worker_state: None, + message: None, + submission: None, + notification: Some(WorkerNotificationAck { + notification_request_id: notification_request_id.into(), + }), } } @@ -133,6 +159,7 @@ impl WorkerExecutionResult { worker_state: None, message: Some(message.into()), submission: None, + notification: None, } } @@ -143,6 +170,7 @@ impl WorkerExecutionResult { worker_state: None, message: Some(message.into()), submission: None, + notification: None, } } @@ -153,6 +181,7 @@ impl WorkerExecutionResult { worker_state: None, message: Some(message.into()), submission: None, + notification: None, } } @@ -163,6 +192,7 @@ impl WorkerExecutionResult { worker_state: None, message: Some(message.into()), submission: None, + notification: None, } } diff --git a/crates/worker-runtime/src/interaction.rs b/crates/worker-runtime/src/interaction.rs index 52e0168ea..e3324e998 100644 --- a/crates/worker-runtime/src/interaction.rs +++ b/crates/worker-runtime/src/interaction.rs @@ -94,6 +94,9 @@ pub struct WorkerInteractionAck { pub worker_ref: WorkerRef, pub status: WorkerStatus, /// Present for User Submit and absent for non-Submit interactions. - #[serde(skip_serializing_if = "Option::is_none")] + #[serde(default, skip_serializing_if = "Option::is_none")] pub submission: Option, + /// Present only after a Notify request reaches the Worker's durable receipt boundary. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub notification: Option, } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 949e75ff8..579f57573 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -2104,6 +2104,12 @@ impl Runtime { .clone() .expect("User input request id must be assigned before dispatch") }); + let expected_notification_id = (input.kind == WorkerInputKind::Notify).then(|| { + input + .submission_request_id + .clone() + .expect("Notify request id must be assigned before dispatch") + }); self.ensure_worker_execution(worker_ref)?; let (backend, handle) = { let state = self.lock()?; @@ -2160,7 +2166,28 @@ impl Runtime { }); } + if let Some(expected_notification_id) = expected_notification_id + && dispatch_result + .notification + .as_ref() + .is_none_or(|ack| ack.notification_request_id != expected_notification_id) + { + let result = WorkerExecutionResult::rejected( + WorkerExecutionOperation::Input, + "execution backend did not acknowledge the durable Runtime notification request id", + ); + self.record_execution_result(worker_ref, result.clone())?; + return Err(RuntimeError::WorkerExecutionRejected { + worker_id: worker_ref.worker_id.clone(), + operation: result.operation, + outcome: result.outcome, + message: result.message_or_default(), + result, + }); + } + let submission = dispatch_result.submission.clone(); + let notification = dispatch_result.notification.clone(); let mut state = self.lock()?; state.ensure_running()?; let worker = state.worker_mut(worker_ref)?; @@ -2180,6 +2207,7 @@ impl Runtime { worker_ref: worker_ref.clone(), status, submission, + notification, }) } @@ -6451,7 +6479,8 @@ mod tests { _handle: &WorkerExecutionHandle, input: WorkerInput, ) -> WorkerExecutionResult { - let submission_id = input.submission_request_id.clone(); + let request_id = input.submission_request_id.clone(); + let input_kind = input.kind.clone(); self.dispatched_inputs.lock().unwrap().push(input); let mut result = self .dispatch_result @@ -6459,20 +6488,28 @@ mod tests { .unwrap() .clone() .unwrap_or_else(|| { - WorkerExecutionResult::accepted_submission( - WorkerExecutionOperation::Input, - "request-test", - "test-submission", - protocol::SubmissionDisposition::Started, - ) + if input_kind == WorkerInputKind::Notify { + WorkerExecutionResult::accepted_notification( + WorkerExecutionOperation::Input, + request_id + .clone() + .expect("Runtime assigns Notify request identity before dispatch"), + ) + } else { + WorkerExecutionResult::accepted_submission( + WorkerExecutionOperation::Input, + "request-test", + "test-submission", + protocol::SubmissionDisposition::Started, + ) + } }); if !self .preserve_submission_acknowledgement_id .load(Ordering::SeqCst) - && let (Some(ack), Some(submission_id)) = - (result.submission.as_mut(), submission_id) + && let (Some(ack), Some(request_id)) = (result.submission.as_mut(), request_id) { - ack.submission_request_id = submission_id; + ack.submission_request_id = request_id; } result } @@ -7911,6 +7948,29 @@ mod tests { assert!(runtime.list_workers().unwrap().is_empty()); } + #[test] + fn notify_rejects_backend_enqueue_without_durable_worker_receipt() { + let (runtime, backend) = runtime_and_backend(); + backend.set_dispatch_result(WorkerExecutionResult::accepted( + WorkerExecutionOperation::Input, + )); + let detail = runtime + .create_worker(task_request("notify receipt")) + .unwrap(); + let mut input = WorkerInput::notify("do not acknowledge enqueue alone"); + input.submission_request_id = Some("notification-request".to_string()); + + let error = runtime.send_input(&detail.worker_ref, input).unwrap_err(); + + assert!(matches!( + error, + RuntimeError::WorkerExecutionRejected { + outcome: crate::execution::WorkerExecutionOutcome::Rejected, + .. + } + )); + } + #[test] fn connected_backend_busy_dispatch_is_typed_and_not_transcribed() { let (runtime, backend) = runtime_and_backend(); diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 8f62224a7..381a972a2 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -45,6 +45,11 @@ use protocol::{Event, Method, Segment, WorkerCommandEnvelope}; static NEXT_INTERNAL_COMMAND_ID: AtomicU64 = AtomicU64::new(1); +enum NotificationAcceptance { + Accepted, + Rejected(String), +} + fn rollback_materialized_spawn_workdirs( materializer: &dyn WorkingDirectoryMaterializer, bindings: &BTreeMap, @@ -1883,6 +1888,75 @@ where .unwrap_or_else(|message| WorkerExecutionResult::errored(operation, message)) } + fn send_notification_and_wait_for_acceptance( + &self, + operation: WorkerExecutionOperation, + worker: WorkerHandle, + method: Method, + notification_request_id: String, + ) -> WorkerExecutionResult { + let request_id = notification_request_id.clone(); + let result = self.run_cancellable_on_adapter_runtime(USER_INPUT_TASK_TIMEOUT, async move { + // Subscribe before enqueueing so a fast durable acceptance cannot + // race the Runtime acknowledgement. + let mut events = worker.subscribe(); + worker + .send(method) + .await + .map_err(|err| format!("failed to send Worker method: {err}"))?; + + tokio::time::timeout(USER_INPUT_COMMIT_TIMEOUT, async move { + loop { + match events.recv().await { + Ok(Event::NotificationAccepted { + notification_request_id, + }) if notification_request_id == request_id => { + return Ok(NotificationAcceptance::Accepted); + } + Ok(Event::NotificationRejected { + notification_request_id, + message, + }) if notification_request_id == request_id => { + return Ok(NotificationAcceptance::Rejected(message)); + } + Ok(Event::Shutdown) => { + return Err( + "worker shut down before notification was durably accepted" + .to_string(), + ); + } + Ok(_) => {} + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + return Err(format!( + "worker notification acknowledgement lagged by {skipped} protocol event(s)" + )); + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + return Err( + "worker event stream closed before notification was durably accepted" + .to_string(), + ); + } + } + } + }) + .await + .map_err(|_| { + "timed out waiting for durable Worker notification acceptance".to_string() + })? + }); + + match result { + Ok(NotificationAcceptance::Accepted) => { + WorkerExecutionResult::accepted_notification(operation, notification_request_id) + } + Ok(NotificationAcceptance::Rejected(message)) => { + WorkerExecutionResult::rejected(operation, message) + } + Err(message) => WorkerExecutionResult::errored(operation, message), + } + } + fn retain_uncertain_unconnected_controller( &self, worker_ref: &crate::identity::WorkerRef, @@ -2879,16 +2953,17 @@ where let notification_request_id = input .submission_request_id .unwrap_or_else(protocol::new_submission_request_id); - return self.send_method( + return self.send_notification_and_wait_for_acceptance( WorkerExecutionOperation::Input, worker, Method::NotifyTracked { notification_request_id: notification_request_id.clone(), message: input.content, source: protocol::AuthenticatedInputSource::Backend { - operation_id: notification_request_id, + operation_id: notification_request_id.clone(), }, }, + notification_request_id, ); } @@ -5108,7 +5183,7 @@ mod tests { } #[test] - fn running_worker_accepts_notify_without_a_submit_receipt() { + fn running_worker_waits_for_durable_notify_receipt() { let client = MockClient::sequential(vec![MockResponse::Hang(vec![])]); let runtime_base = tempfile::tempdir().unwrap(); let cwd = tempfile::tempdir().unwrap(); @@ -5143,6 +5218,13 @@ mod tests { .expect("Running Worker must accept Notify without a Submit receipt"); assert!(acknowledgement.submission.is_none()); + assert_eq!( + acknowledgement + .notification + .as_ref() + .map(|ack| ack.notification_request_id.as_str()), + Some("notification-request") + ); wait_for_adapter_state(&backend, &detail.worker_ref, WorkerStatus::Running); } diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 42c4efc25..11a8e3042 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1489,6 +1489,40 @@ where Ok(workdir_for_view) } +fn durably_accept_notification( + notification_request_id: String, + message: String, + source_namespace: String, + provenance: session_store::LoggedSessionHistoryOrigin, + pending_submissions: &PendingSubmissionHandle, + working_event_tx: &broadcast::Sender, +) -> bool +where + St: session_store::Store + Clone, +{ + let request_id = notification_request_id.clone(); + match pending_submissions.accept_notification_from_source( + notification_request_id, + message, + source_namespace, + provenance, + ) { + Ok(_) => { + let _ = working_event_tx.send(Event::NotificationAccepted { + notification_request_id: request_id, + }); + true + } + Err(error) => { + let _ = working_event_tx.send(Event::NotificationRejected { + notification_request_id: request_id, + message: error.to_string(), + }); + false + } + } +} + fn durably_accept_method_while_busy( method: Method, pending_submissions: &PendingSubmissionHandle, @@ -1561,17 +1595,14 @@ where notification_request_id, message, } => { - if let Err(error) = pending_submissions.accept_notification_from_source( + durably_accept_notification( notification_request_id, message, pending_submissions.direct_client_namespace(), session_store::LoggedSessionHistoryOrigin::LegacyUnknown, - ) { - let _ = working_event_tx.send(Event::Error { - code: ErrorCode::InvalidRequest, - message: error.to_string(), - }); - } + pending_submissions, + working_event_tx, + ); None } Method::NotifyTracked { @@ -1581,17 +1612,14 @@ where } => { let (source_namespace, provenance) = resolved_input_source(pending_submissions, &source); - if let Err(error) = pending_submissions.accept_notification_from_source( + durably_accept_notification( notification_request_id, message, source_namespace, provenance, - ) { - let _ = working_event_tx.send(Event::Error { - code: ErrorCode::InvalidRequest, - message: error.to_string(), - }); - } + pending_submissions, + working_event_tx, + ); None } method => Some(method), @@ -1619,9 +1647,16 @@ fn reject_method_while_attention_locked( }); true } - Method::Notify { .. } | Method::NotifyTracked { .. } => { - let _ = working_event_tx.send(Event::Error { - code: ErrorCode::InvalidRequest, + Method::Notify { + notification_request_id, + .. + } + | Method::NotifyTracked { + notification_request_id, + .. + } => { + let _ = working_event_tx.send(Event::NotificationRejected { + notification_request_id: notification_request_id.clone(), message: message.to_owned(), }); true @@ -2031,21 +2066,14 @@ async fn controller_loop( notification_request_id, message, } => { - let source_namespace = pending_submissions.direct_client_namespace(); - match pending_submissions.accept_notification_from_source( + durably_accept_notification( notification_request_id, message, - source_namespace.clone(), + pending_submissions.direct_client_namespace(), session_store::LoggedSessionHistoryOrigin::LegacyUnknown, - ) { - Ok(_) => {} - Err(error) => { - let _ = working_event_tx.send(Event::Error { - code: ErrorCode::InvalidRequest, - message: error.to_string(), - }); - } - } + &pending_submissions, + &working_event_tx, + ); } Method::NotifyTracked { @@ -2055,20 +2083,14 @@ async fn controller_loop( } => { let (source_namespace, provenance) = resolved_input_source(&pending_submissions, &source); - match pending_submissions.accept_notification_from_source( + durably_accept_notification( notification_request_id, message, - source_namespace.clone(), + source_namespace, provenance, - ) { - Ok(_) => {} - Err(error) => { - let _ = working_event_tx.send(Event::Error { - code: ErrorCode::InvalidRequest, - message: error.to_string(), - }); - } - } + &pending_submissions, + &working_event_tx, + ); } Method::ListPendingSubmissions => { @@ -3113,22 +3135,15 @@ where notification_request_id, message, }) => { - let source_namespace = pending_submissions.direct_client_namespace(); - match pending_submissions.accept_notification_from_source( + if durably_accept_notification( notification_request_id, message, - source_namespace.clone(), + pending_submissions.direct_client_namespace(), session_store::LoggedSessionHistoryOrigin::LegacyUnknown, + pending_submissions, + working_event_tx, ) { - Ok(_) => { - stage_pending_notifications(&pending_submissions, notify_buffer); - } - Err(error) => { - let _ = working_event_tx.send(Event::Error { - code: ErrorCode::InvalidRequest, - message: error.to_string(), - }); - } + stage_pending_notifications(&pending_submissions, notify_buffer); } let _ = working_event_tx.send(Event::PendingSubmissionsChanged { pending: pending_submissions.snapshot(), @@ -3141,21 +3156,15 @@ where }) => { let (source_namespace, provenance) = resolved_input_source(pending_submissions, &source); - match pending_submissions.accept_notification_from_source( + if durably_accept_notification( notification_request_id, message, - source_namespace.clone(), + source_namespace, provenance, + pending_submissions, + working_event_tx, ) { - Ok(_) => { - stage_pending_notifications(&pending_submissions, notify_buffer); - } - Err(error) => { - let _ = working_event_tx.send(Event::Error { - code: ErrorCode::InvalidRequest, - message: error.to_string(), - }); - } + stage_pending_notifications(&pending_submissions, notify_buffer); } let _ = working_event_tx.send(Event::PendingSubmissionsChanged { pending: pending_submissions.snapshot(), @@ -3622,29 +3631,46 @@ mod tests { } #[test] - fn busy_notification_is_durable_before_shutdown_can_observe_it() { + fn busy_notification_receipt_follows_durable_idempotent_acceptance() { let dir = tempfile::tempdir().unwrap(); let pending = PendingSubmissionHandle::for_test(dir.path()); - let (event_tx, _event_rx) = broadcast::channel(4); + let (event_tx, mut event_rx) = broadcast::channel(4); + let method = Method::NotifyTracked { + notification_request_id: "completion-1".into(), + message: "subworker completed".into(), + source: protocol::AuthenticatedInputSource::UntrustedWire, + }; - assert!( - durably_accept_method_while_busy( - Method::NotifyTracked { - notification_request_id: "completion-1".into(), - message: "subworker completed".into(), - source: protocol::AuthenticatedInputSource::UntrustedWire, - }, - &pending, - &event_tx, - ) - .is_none() + for _ in 0..2 { + assert!( + durably_accept_method_while_busy(method.clone(), &pending, &event_tx).is_none() + ); + let entries = pending.persisted_entries_for_test(); + assert!(entries.iter().any(|entry| matches!( + entry, + LogEntry::Extension { domain, .. } + if domain == "worker.pending_activations.v1" + ))); + assert!(matches!( + event_rx.try_recv().unwrap(), + Event::NotificationAccepted { + notification_request_id, + } if notification_request_id == "completion-1" + )); + } + assert_eq!(pending.snapshot().notification_count, 1); + assert_eq!( + pending + .persisted_entries_for_test() + .iter() + .filter(|entry| matches!( + entry, + LogEntry::Extension { domain, .. } + if domain == "worker.pending_activations.v1" + )) + .count(), + 1 ); - let entries = pending.persisted_entries_for_test(); - assert!(entries.iter().any(|entry| matches!( - entry, - LogEntry::Extension { domain, .. } - if domain == "worker.pending_activations.v1" - ))); } #[test] diff --git a/crates/workspace-server/src/backend_job.rs b/crates/workspace-server/src/backend_job.rs index 996f01780..c1c83e50c 100644 --- a/crates/workspace-server/src/backend_job.rs +++ b/crates/workspace-server/src/backend_job.rs @@ -355,6 +355,27 @@ pub fn allocation_key(attempt_id: &str) -> String { format!("backend-job:{attempt_id}") } +/// Fixed-size Runtime request identity for the exact Job attempt. Caller-selected +/// Job ids may be longer than the Worker protocol request-id bound, so never +/// embed them directly in tracked Runtime identities. +pub fn runtime_run_id(attempt_id: &str) -> String { + tracked_request_id("run", attempt_id.as_bytes()) +} + +/// Fixed-size durable delivery identity, also reused as the Worker's tracked +/// notification request id so restart replay converges on one receipt. +pub fn delivery_id(job_id: &str, attempt_id: &str) -> String { + let mut identity = Vec::with_capacity(job_id.len() + attempt_id.len() + 1); + identity.extend_from_slice(job_id.as_bytes()); + identity.push(0); + identity.extend_from_slice(attempt_id.as_bytes()); + tracked_request_id("notification", &identity) +} + +fn tracked_request_id(kind: &str, identity: &[u8]) -> String { + format!("backend-job-{kind}:{}", sha256(identity)) +} + pub fn result_digest(result: &Value, max_bytes: u32) -> Result<(String, String)> { let encoded = serde_json::to_string(result) .map_err(|error| Error::InvalidInput(format!("serialize Backend Job result: {error}")))?; @@ -448,6 +469,28 @@ mod tests { } } + #[test] + fn tracked_runtime_ids_are_fixed_size_for_maximum_job_ids() { + let job_id = "j".repeat(256); + let attempt_id = attempt_id(&job_id, ABSOLUTE_MAX_ATTEMPTS); + let run_id = runtime_run_id(&attempt_id); + let notification_id = delivery_id(&job_id, &attempt_id); + + assert!(run_id.len() <= 128, "run id was {} bytes", run_id.len()); + assert!( + notification_id.len() <= 128, + "notification id was {} bytes", + notification_id.len() + ); + assert_eq!(run_id, runtime_run_id(&attempt_id)); + assert_eq!(notification_id, delivery_id(&job_id, &attempt_id)); + assert_ne!(run_id, runtime_run_id(&format!("{attempt_id}-other"))); + assert_ne!( + notification_id, + delivery_id(&job_id, &format!("{attempt_id}-other")) + ); + } + #[test] fn result_is_bounded_before_digesting() { let value = serde_json::json!({"ok": true}); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index dd6f8aa06..e54510319 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -3394,7 +3394,7 @@ impl WorkspaceApi { } }; let bound_at = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true); - let runtime_run_id = format!("backend-job-run:{}", reservation.attempt.attempt_id); + let runtime_run_id = crate::backend_job::runtime_run_id(&reservation.attempt.attempt_id); let (commit, newly_bound) = self.store.bind_backend_job_attempt_worker( &self.config.workspace_id, &request.job_id, @@ -3671,9 +3671,9 @@ impl WorkspaceApi { acceptance: &BackendJobResultAcceptance, target: &RuntimeWorkerRef, ) { - let delivery_id = format!( - "{}:{}:notification", - acceptance.job.request.job_id, acceptance.attempt.attempt_id + let delivery_id = crate::backend_job::delivery_id( + &acceptance.job.request.job_id, + &acceptance.attempt.attempt_id, ); let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true); let Ok((delivery, claimed)) = self.store.reserve_backend_job_delivery( @@ -33863,6 +33863,7 @@ mod tests { .cloned() .expect("execution context"); let submission_request_id = input.submission_request_id.clone(); + let input_kind = input.kind.clone(); let content = input.content.clone(); std::thread::spawn(move || { std::thread::sleep(std::time::Duration::from_millis(25)); @@ -33870,7 +33871,13 @@ mod tests { text: format!("server companion echoed: {content}"), }); }); - if let Some(submission_request_id) = submission_request_id { + if input_kind == worker_runtime::interaction::WorkerInputKind::Notify { + worker_runtime::execution::WorkerExecutionResult::accepted_notification( + worker_runtime::execution::WorkerExecutionOperation::Input, + submission_request_id.expect("Notify test input has a Runtime request id"), + ) + .with_worker_state(protocol::WorkerStateSnapshot::initial()) + } else if let Some(submission_request_id) = submission_request_id { worker_runtime::execution::WorkerExecutionResult::accepted_submission( worker_runtime::execution::WorkerExecutionOperation::Input, submission_request_id, @@ -34121,7 +34128,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let (api, execution) = test_api_with_recording_backend(temp.path()).await; let request = BackendJobRequest { - job_id: "runner-fast-result-1".to_string(), + job_id: "r".repeat(256), purpose: "fast_result_contract_check".to_string(), input_revision: "revision-1".to_string(), input_ref: "test://runner/fast-result/1".to_string(), @@ -34156,10 +34163,8 @@ mod tests { let dispatched = api.dispatch_backend_job(&request).unwrap(); assert_eq!(*acceptance_observed.lock().unwrap(), Some(true)); assert_eq!(dispatched.job.state, BackendJobState::Completed); - let expected_run_id = format!( - "backend-job-run:{}", - crate::backend_job::attempt_id(&request.job_id, 1) - ); + let expected_run_id = + crate::backend_job::runtime_run_id(&crate::backend_job::attempt_id(&request.job_id, 1)); assert_eq!( dispatched.attempt.runtime_run_id.as_deref(), Some(expected_run_id.as_str()) @@ -34207,7 +34212,7 @@ mod tests { .unwrap() .worker; let request = BackendJobRequest { - job_id: "runner-delivery-recovery-1".to_string(), + job_id: "n".repeat(256), purpose: "delivery_recovery_contract_check".to_string(), input_revision: "revision-1".to_string(), input_ref: "test://runner/delivery-recovery/1".to_string(), @@ -34237,10 +34242,8 @@ mod tests { &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), ) .unwrap(); - let delivery_id = format!( - "{}:{}:notification", - request.job_id, acceptance.attempt.attempt_id - ); + let delivery_id = + crate::backend_job::delivery_id(&request.job_id, &acceptance.attempt.attempt_id); let (_, claimed) = api .store .reserve_backend_job_delivery( diff --git a/docs/design/durable-operations.md b/docs/design/durable-operations.md index 0fd602e4b..08616d3de 100644 --- a/docs/design/durable-operations.md +++ b/docs/design/durable-operations.md @@ -319,10 +319,12 @@ The durable split is deliberate: `(Workspace, Runtime, Worker)` identity, a precommitted tracked Runtime submission request identity, a deadline, and its bounded outcome; - `backend_job_deliveries` records best-effort notification separately from the - result. Its durable delivery ID is also the Runtime's tracked notification - request ID, so restart recovery can replay an ambiguous `sending` claim - without duplicating Worker context. A missing or stopped recipient never - discards a result or reruns model work. + result. Its fixed-size collision-resistant durable delivery ID is also the + Runtime's tracked notification request ID. Runtime reports acceptance only + after the Worker has persisted the correlated notification receipt, so restart + recovery can replay an ambiguous `sending` claim without duplicating Worker + context. A missing or stopped recipient never discards a result or reruns + model work. Exact dispatch retry reuses the deterministic attempt and Worker allocation key. Only explicit re-evaluation of a definitively failed attempt increments the @@ -334,8 +336,9 @@ state, recursively create Workers or Jobs, or inherit caller Workdir authority. Success is accepted only by the structured result boundary. Before Runtime input can execute, the attempt is fenced by Workspace, Job ID, attempt ID, input -revision, bound Runtime Worker identity, and the deterministic tracked submission -request identity. Ordinary Worker input and Console mutation are rejected for +revision, bound Runtime Worker identity, and a fixed-size collision-resistant +tracked submission request identity derived from the attempt. Ordinary Worker +input and Console mutation are rejected for Job-owned Workers, so no later run in the same Session can exercise the result capability. Result JSON bytes are bounded before persistence. Worker final prose, Idle, Stopped, or a successful model call are not success evidence. Identical