diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 47c0aa6e..45ff87bb 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -5312,6 +5312,66 @@ fn ticket_notification_content(ticket_id: &str, current_state: &str) -> String { ) } +#[derive(Debug, Clone, Serialize, PartialEq, Eq)] +struct TicketNotificationDeliveryWarning { + level: &'static str, + event: &'static str, + ticket_id: String, + recipient_runtime_id: String, + recipient_worker_id: String, + error_category: &'static str, +} + +impl TicketNotificationDeliveryWarning { + fn new(ticket_id: &str, recipient: &RuntimeWorkerRef, error_category: &'static str) -> Self { + Self { + level: "warning", + event: "ticket_notification_delivery_failed", + ticket_id: ticket_id.to_string(), + recipient_runtime_id: recipient.runtime_id.clone(), + recipient_worker_id: recipient.worker_id.clone(), + error_category, + } + } +} + +#[cfg(test)] +static TICKET_NOTIFICATION_DELIVERY_WARNING_CAPTURE: Mutex> = + Mutex::new(Vec::new()); + +fn ticket_notification_delivery_error_category( + result: &std::result::Result, +) -> Option<&'static str> { + match result { + Ok(result) => match result.state { + WorkerOperationState::Accepted => None, + WorkerOperationState::Rejected => Some("runtime_rejected"), + WorkerOperationState::Unsupported => Some("runtime_unsupported"), + }, + Err(RuntimeRegistryError::InvalidIdentifier { .. }) => Some("invalid_identifier"), + Err(RuntimeRegistryError::UnknownRuntime(_)) => Some("unknown_runtime"), + Err(RuntimeRegistryError::UnknownHost(_)) => Some("unknown_host"), + Err(RuntimeRegistryError::UnknownWorker { .. }) => Some("unknown_worker"), + Err(RuntimeRegistryError::RuntimeOperationFailed { .. }) => { + Some("runtime_operation_failed") + } + } +} + +fn emit_ticket_notification_delivery_warning(warning: TicketNotificationDeliveryWarning) { + let serialized = serde_json::to_string(&warning) + .expect("Ticket notification delivery warnings serialize from bounded string fields"); + eprintln!( + "{} yoi-server {serialized}", + Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true) + ); + #[cfg(test)] + TICKET_NOTIFICATION_DELIVERY_WARNING_CAPTURE + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .push(warning); +} + fn notify_ticket_recipients( api: &WorkspaceApi, workspace_id: &str, @@ -5343,7 +5403,7 @@ fn notify_ticket_recipients( if source.as_ref().is_some_and(|source| source == &recipient) { continue; } - let _ = api.runtime.send_input( + let result = api.runtime.send_input( &recipient, WorkerInputRequest { kind: WorkerInputKind::Notify, @@ -5351,6 +5411,13 @@ fn notify_ticket_recipients( segments: None, }, ); + if let Some(error_category) = ticket_notification_delivery_error_category(&result) { + emit_ticket_notification_delivery_warning(TicketNotificationDeliveryWarning::new( + ticket_id, + &recipient, + error_category, + )); + } } } @@ -15727,7 +15794,7 @@ mod tests { } #[tokio::test] - async fn queued_ticket_mutation_succeeds_without_orchestrator() { + async fn queued_ticket_mutation_stays_committed_when_notification_recipient_is_missing() { let dir = tempfile::tempdir().unwrap(); let api = test_api(dir.path()).await; let source = api @@ -15766,6 +15833,43 @@ mod tests { let mut input = ticket::NewTicket::new("Queued notification"); input.workflow_state = Some(TicketWorkflowState::Queued); let ticket_ref = backend.create(input).unwrap(); + let missing_recipient = + RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, "missing-notification-recipient"); + api.store + .upsert_worker_registry(&WorkerRegistryRecord { + workspace_id: TEST_WORKSPACE_ID.to_string(), + worker: missing_recipient.clone(), + display_name: "Missing notification recipient".to_string(), + profile: Some("builtin:coder".to_string()), + retention_state: "normal".to_string(), + transcript_ref: None, + session_ref: None, + summary_ref: None, + diagnostics_ref: None, + created_at: TEST_CREATED_AT.to_string(), + updated_at: TEST_CREATED_AT.to_string(), + }) + .unwrap(); + api.store + .set_current_ticket_worker_assignment( + &TicketWorkerAssignmentRecord { + workspace_id: TEST_WORKSPACE_ID.to_string(), + ticket_id: ticket_ref.id.clone(), + assignment_id: "missing-recipient-assignment".to_string(), + worker: missing_recipient.clone(), + assigned_by: "test-user".to_string(), + assigned_at: TEST_CREATED_AT.to_string(), + }, + None, + "missing-recipient-assignment-event", + "missing-recipient-assignment-operation", + false, + ) + .unwrap(); + TICKET_NOTIFICATION_DELIVERY_WARNING_CAPTURE + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .retain(|warning| warning.ticket_id != ticket_ref.id); let mut headers = HeaderMap::new(); headers.insert( "x-yoi-runtime-id", @@ -15788,6 +15892,28 @@ mod tests { ) .await .unwrap(); + + assert_eq!( + api.authority.ticket(&ticket_ref.id).unwrap().state, + TicketWorkflowState::Queued.as_str() + ); + let warnings = TICKET_NOTIFICATION_DELIVERY_WARNING_CAPTURE + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .iter() + .filter(|warning| warning.ticket_id == ticket_ref.id) + .cloned() + .collect::>(); + assert_eq!(warnings.len(), 1); + let warning = &warnings[0]; + assert_eq!(warning.level, "warning"); + assert_eq!(warning.event, "ticket_notification_delivery_failed"); + assert_eq!(warning.recipient_runtime_id, missing_recipient.runtime_id); + assert_eq!(warning.recipient_worker_id, missing_recipient.worker_id); + assert_eq!(warning.error_category, "unknown_worker"); + let serialized = serde_json::to_string(warning).unwrap(); + assert!(!serialized.contains("queued update")); + assert!(!serialized.contains("Ticket notification:")); } #[tokio::test]