1use std::sync::OnceLock;
51
52use enum_as_inner::EnumAsInner;
53use serde::Deserialize;
54use serde::Serialize;
55use thiserror::Error;
56
57use crate::ActorAddr;
58use crate::Addr;
59use crate::Client;
60use crate::EndpointLocation;
61use crate::Message;
63use crate::Proc;
64use crate::mailbox::DeliveryFailure;
65use crate::mailbox::MailboxSender;
66use crate::mailbox::MailboxSenderError;
67use crate::mailbox::MessageEnvelope;
68use crate::mailbox::PortHandle;
69use crate::mailbox::PortReceiver;
70use crate::mailbox::TransportFailure;
71use crate::mailbox::TransportFailureReason;
72use crate::mailbox::UndeliverableMailboxSender;
73use crate::mailbox::UndeliverableReason;
74use crate::mailbox::headers::OPERATION_ADVERB;
75use crate::mailbox::headers::OPERATION_ENDPOINT;
76use crate::mailbox::headers::RUST_MESSAGE_TYPE;
77
78#[derive(Debug, Serialize, Deserialize, Clone, typeuri::Named)]
80pub struct DeliveryFailureReport {
81 pub sender: ActorAddr,
83 pub dest: EndpointLocation,
85 pub message_type: Option<String>,
87 pub delivery_failures: Vec<DeliveryFailure>,
91}
92
93impl DeliveryFailureReport {
94 pub fn new(
96 sender: ActorAddr,
97 dest: EndpointLocation,
98 message_type: Option<String>,
99 failure: DeliveryFailure,
100 ) -> Self {
101 Self {
102 sender,
103 dest,
104 message_type,
105 delivery_failures: vec![failure],
106 }
107 }
108
109 pub(crate) fn from_send_error<M: Message>(
111 sender: ActorAddr,
112 dest: EndpointLocation,
113 error: &MailboxSenderError,
114 ) -> Self {
115 let failure = match &dest {
116 EndpointLocation::Port(port) => {
117 super::serialized_send_error_delivery_failure(port, error)
118 }
119 EndpointLocation::Actor(actor) => {
120 DeliveryFailure::new(UndeliverableReason::Transport(TransportFailure::new(
121 actor.clone(),
122 TransportFailureReason::LinkUnavailable(error.to_string()),
123 )))
124 }
125 EndpointLocation::Local { actor, .. } => {
126 DeliveryFailure::new(UndeliverableReason::Transport(TransportFailure::new(
127 actor.clone(),
128 TransportFailureReason::LinkUnavailable(error.to_string()),
129 )))
130 }
131 };
132 Self {
133 sender,
134 dest,
135 message_type: Some(std::any::type_name::<M>().to_string()),
136 delivery_failures: vec![failure],
137 }
138 }
139
140 pub(crate) fn link_unavailable<M: Message>(
142 sender: ActorAddr,
143 dest: EndpointLocation,
144 error: impl Into<String>,
145 ) -> Self {
146 let failure = DeliveryFailure::new(UndeliverableReason::Transport(TransportFailure::new(
147 delivery_failure_target(&dest),
148 TransportFailureReason::LinkUnavailable(error.into()),
149 )));
150 Self::new(
151 sender,
152 dest,
153 Some(std::any::type_name::<M>().to_string()),
154 failure,
155 )
156 }
157
158 pub fn root_delivery_failure(&self) -> Option<&DeliveryFailure> {
160 self.delivery_failures.first()
161 }
162
163 pub fn error_msg(&self) -> Option<String> {
165 if self.delivery_failures.is_empty() {
166 return None;
167 }
168
169 Some(
170 self.delivery_failures
171 .iter()
172 .map(DeliveryFailure::render_bounded)
173 .collect::<Vec<_>>()
174 .join("; "),
175 )
176 }
177}
178
179fn delivery_failure_target(dest: &EndpointLocation) -> Addr {
180 match dest {
181 EndpointLocation::Actor(actor) => actor.clone().into(),
182 EndpointLocation::Port(port) => port.clone().into(),
183 EndpointLocation::Local { actor, .. } => actor.clone().into(),
184 }
185}
186
187#[expect(
189 clippy::large_enum_variant,
190 reason = "returned messages stay inline so callers can recover the original payload without extra allocation"
191)]
192#[derive(Debug, EnumAsInner, Serialize, Deserialize, Clone, typeuri::Named)]
193pub enum Undeliverable<M: Message> {
194 Returned(M),
196 Report(DeliveryFailureReport),
198}
199
200impl<M: Message> Undeliverable<M> {
201 pub fn message(message: M) -> Self {
203 Self::Returned(message)
204 }
205
206 pub fn as_message(&self) -> Option<&M> {
208 match self {
209 Self::Returned(message) => Some(message),
210 Self::Report(_) => None,
211 }
212 }
213
214 pub fn as_message_mut(&mut self) -> Option<&mut M> {
216 match self {
217 Self::Returned(message) => Some(message),
218 Self::Report(_) => None,
219 }
220 }
221
222 #[expect(
225 clippy::result_large_err,
226 reason = "preserve the old helper shape while callers migrate to explicit variants"
227 )]
228 pub fn into_message(self) -> Result<M, Self> {
229 match self {
230 Self::Returned(message) => Ok(message),
231 report @ Self::Report(_) => Err(report),
232 }
233 }
234
235 pub fn report(report: DeliveryFailureReport) -> Self {
238 Self::Report(report)
239 }
240}
241
242impl Undeliverable<MessageEnvelope> {
243 pub fn root_delivery_failure(&self) -> Option<&DeliveryFailure> {
246 match self {
247 Self::Returned(envelope) => envelope.root_delivery_failure(),
248 Self::Report(report) => report.root_delivery_failure(),
249 }
250 }
251
252 pub fn into_error(self) -> UndeliverableMessageError {
254 match self {
255 Self::Returned(envelope) => UndeliverableMessageError::DeliveryFailure { envelope },
256 Self::Report(report) => UndeliverableMessageError::Report { report },
257 }
258 }
259}
260
261pub(crate) fn new_undeliverable_port() -> (
263 PortHandle<Undeliverable<MessageEnvelope>>,
264 PortReceiver<Undeliverable<MessageEnvelope>>,
265) {
266 let proc = Proc::isolated();
267 crate::mailbox::Mailbox::new(proc.proc_addr().actor_addr("undeliverable"))
268 .open_port::<Undeliverable<MessageEnvelope>>()
269}
270
271static MONITORED_RETURN_HANDLE: OnceLock<PortHandle<Undeliverable<MessageEnvelope>>> =
275 OnceLock::new();
276pub fn monitored_return_handle() -> PortHandle<Undeliverable<MessageEnvelope>> {
281 let return_handle = MONITORED_RETURN_HANDLE.get_or_init(|| {
282 let (return_handle, mut rx) = new_undeliverable_port();
283 let (h, _) = new_undeliverable_port();
286 crate::init::get_runtime().spawn(async move {
287 while let Ok(undeliverable) = rx.recv().await {
288 match undeliverable {
289 Undeliverable::Returned(mut envelope) => {
290 envelope.push_delivery_failure(DeliveryFailure::new(
291 UndeliverableReason::Transport(TransportFailure::new(
292 envelope.dest().clone(),
293 TransportFailureReason::LinkUnavailable(
294 "message returned to undeliverable port".to_string(),
295 ),
296 )),
297 ));
298 super::UndeliverableMailboxSender
299 .post(envelope, h.clone());
300 }
301 Undeliverable::Report(report) => {
302 tracing::error!(
303 sender = %report.sender,
304 dest = %report.dest,
305 message_type = report.message_type.as_deref().unwrap_or("unknown"),
306 error = %report.error_msg().unwrap_or_default(),
307 "undeliverable message report returned to undeliverable port"
308 );
309 }
310 }
311 }
312 });
313 return_handle
314 });
315
316 return_handle.clone()
317}
318
319#[track_caller]
323pub fn custom_monitored_return_handle(caller: &str) -> PortHandle<Undeliverable<MessageEnvelope>> {
324 let caller = caller.to_owned();
325 let (return_handle, mut rx) = new_undeliverable_port();
326 tokio::task::spawn(async move {
327 while let Ok(undeliverable) = rx.recv().await {
328 match undeliverable {
329 Undeliverable::Returned(mut envelope) => {
330 envelope.push_delivery_failure(DeliveryFailure::new(
331 UndeliverableReason::Transport(TransportFailure::new(
332 envelope.dest().clone(),
333 TransportFailureReason::LinkUnavailable(
334 "message returned to undeliverable port".to_string(),
335 ),
336 )),
337 ));
338 tracing::error!("{caller} took back an undeliverable message: {}", envelope);
339 }
340 Undeliverable::Report(report) => {
341 tracing::error!(
342 sender = %report.sender,
343 dest = %report.dest,
344 message_type = report.message_type.as_deref().unwrap_or("unknown"),
345 error = %report.error_msg().unwrap_or_default(),
346 "{caller} took back an undeliverable message report"
347 );
348 }
349 }
350 }
351 });
352 return_handle
353}
354
355pub(crate) fn return_undeliverable(
357 return_handle: PortHandle<Undeliverable<MessageEnvelope>>,
358 envelope: MessageEnvelope,
359) {
360 if envelope.return_undeliverable() {
361 static CLIENT: OnceLock<Client> = OnceLock::new();
363 let client = CLIENT.get_or_init(|| Proc::global().client("global_return_client"));
364 let envelope_copy = envelope.clone();
365 if return_handle
366 .try_post(client, Undeliverable::message(envelope))
367 .is_err()
368 {
369 UndeliverableMailboxSender.post(envelope_copy, return_handle)
370 }
371 }
372}
373
374#[derive(Debug, Error)]
375pub enum UndeliverableMessageError {
377 DeliveryFailure {
379 envelope: MessageEnvelope,
381 },
382
383 ReturnFailure {
386 envelope: MessageEnvelope,
388 },
389
390 Report {
392 report: DeliveryFailureReport,
394 },
395}
396
397fn undeliverable_prefix(error: &UndeliverableMessageError) -> String {
406 let envelope = match error {
407 UndeliverableMessageError::DeliveryFailure { envelope }
408 | UndeliverableMessageError::ReturnFailure { envelope } => envelope,
409 UndeliverableMessageError::Report { report } => {
410 return format!("undeliverable message report to {}", report.dest);
411 }
412 };
413 if let Some(endpoint) = envelope.headers().get(OPERATION_ENDPOINT) {
414 let adverb = envelope
415 .headers()
416 .get(OPERATION_ADVERB)
417 .unwrap_or_else(|| "?".to_string());
418 return format!("undeliverable message for {} ({})", endpoint, adverb);
419 }
420 match error {
421 UndeliverableMessageError::DeliveryFailure { .. } => {
422 format!("undeliverable message to {}", envelope.dest())
423 }
424 UndeliverableMessageError::ReturnFailure { .. } => {
425 format!(
426 "undeliverable return to original sender {}",
427 envelope.sender()
428 )
429 }
430 UndeliverableMessageError::Report { report } => {
431 format!("undeliverable message report to {}", report.dest)
432 }
433 }
434}
435
436impl std::fmt::Display for UndeliverableMessageError {
437 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
438 let (envelope, description, sender_label, dest_label) = match self {
445 UndeliverableMessageError::DeliveryFailure { envelope } => (
446 envelope,
447 "delivery of message from sender to dest failed",
448 "sender",
449 "dest",
450 ),
451 UndeliverableMessageError::ReturnFailure { envelope } => (
452 envelope,
453 "returning undeliverable message to original sender failed",
454 "original sender",
455 "original dest",
456 ),
457 UndeliverableMessageError::Report { report } => {
458 writeln!(f, "{}:", undeliverable_prefix(self))?;
459 writeln!(
460 f,
461 "\tdescription: delivery failed and the original payload is unavailable"
462 )?;
463 writeln!(f, "\tsender: {}", report.sender)?;
464 writeln!(f, "\tdest: {}", report.dest)?;
465 writeln!(
466 f,
467 "\tmessage type: {}",
468 report.message_type.as_deref().unwrap_or("unknown")
469 )?;
470 writeln!(
471 f,
472 "\terror: {}",
473 report.error_msg().unwrap_or("<none>".to_string())
474 )?;
475 return Ok(());
476 }
477 };
478
479 writeln!(f, "{}:", undeliverable_prefix(self))?;
480 writeln!(f, "\tdescription: {}", description)?;
481 writeln!(f, "\t{}: {}", sender_label, envelope.sender())?;
482 writeln!(f, "\t{}: {}", dest_label, envelope.dest())?;
483 let message_type = envelope
487 .data()
488 .typename()
489 .map(|s| s.to_string())
490 .or_else(|| envelope.headers().get(RUST_MESSAGE_TYPE))
491 .unwrap_or_else(|| "unknown".to_string());
492 writeln!(f, "\tmessage type: {}", message_type)?;
493 writeln!(f, "\tdata_len: {}", envelope.data().len())?;
494 writeln!(
495 f,
496 "\terror: {}",
497 envelope.error_msg().unwrap_or("<none>".to_string())
498 )
499 }
500}
501
502#[cfg(test)]
503mod tests {
504 use hyperactor_config::Flattrs;
505
506 use super::*;
507 use crate::mailbox::InvalidReference;
508 use crate::mailbox::InvalidReferenceReason;
509 use crate::mailbox::MessageEnvelope;
510 use crate::testing::ids::test_actor_id;
511 use crate::testing::ids::test_port_id;
512
513 fn make_envelope(payload: &str, headers: Flattrs) -> MessageEnvelope {
514 let sender = test_actor_id("ue_proc", "ue_sender");
515 let dest = test_port_id("ue_dest_proc", "ue_dest", 42);
516 let data = wirevalue::Any::serialize(&payload.to_string()).unwrap();
517 MessageEnvelope::new(sender, dest, data, headers)
518 }
519
520 #[test]
524 fn test_ue1_delivery_failure_bounded() {
525 let payload: String = std::iter::repeat_n('x', 10_000).collect();
526 let mut headers = Flattrs::new();
527 headers.set(OPERATION_ENDPOINT, "training.buffer.sample()".to_string());
528 let envelope = make_envelope(&payload, headers);
529 let rendered = format!(
530 "{}",
531 UndeliverableMessageError::DeliveryFailure { envelope }
532 );
533
534 assert!(
535 rendered.contains("message type:"),
536 "UE-1: message type field must be present, got:\n{rendered}"
537 );
538 assert!(
539 rendered.contains("data_len:"),
540 "UE-1: data_len field must be present, got:\n{rendered}"
541 );
542 assert!(
543 rendered.contains("sender:"),
544 "UE-2: sender field must be preserved, got:\n{rendered}"
545 );
546 assert!(
547 rendered.contains("dest:"),
548 "UE-2: dest field must be preserved, got:\n{rendered}"
549 );
550 assert!(
551 rendered.contains("error:"),
552 "UE-2: error field must be preserved, got:\n{rendered}"
553 );
554 assert!(
556 !rendered.contains("\theaders: "),
557 "UE-1: raw headers dump leaked, got:\n{rendered}"
558 );
559 assert!(
560 !rendered.contains("\tdata: "),
561 "UE-1: raw data dump leaked, got:\n{rendered}"
562 );
563 assert!(
565 !rendered.contains(&payload),
566 "UE-1: payload body leaked into rendered text"
567 );
568 }
569
570 #[test]
573 fn test_ue1_return_failure_bounded() {
574 let payload: String = std::iter::repeat_n('y', 10_000).collect();
575 let envelope = make_envelope(&payload, Flattrs::new());
576 let rendered = format!("{}", UndeliverableMessageError::ReturnFailure { envelope });
577
578 assert!(
579 rendered.contains("data_len:"),
580 "UE-1: data_len field must be present, got:\n{rendered}"
581 );
582 assert!(
583 !rendered.contains("\theaders: "),
584 "UE-1: raw headers dump leaked, got:\n{rendered}"
585 );
586 assert!(
587 !rendered.contains("\tdata: "),
588 "UE-1: raw data dump leaked, got:\n{rendered}"
589 );
590 assert!(
591 !rendered.contains(&payload),
592 "UE-1: payload body leaked into rendered text"
593 );
594 }
595
596 #[test]
597 fn test_delivery_failure_display_uses_structured_failure() {
598 let mut envelope = make_envelope("payload", Flattrs::new());
599 let dest = envelope.dest().clone();
600 envelope.push_delivery_failure(DeliveryFailure::new(InvalidReference::new(
601 dest,
602 InvalidReferenceReason::PortNeverAllocated,
603 )));
604
605 let rendered = format!(
606 "{}",
607 UndeliverableMessageError::DeliveryFailure { envelope }
608 );
609
610 assert!(
611 rendered.contains("\terror: delivery failure: invalid reference"),
612 "structured delivery failure should render in error field, got:\n{rendered}"
613 );
614 assert!(
615 rendered.contains("port never allocated"),
616 "structured reason should render in error field, got:\n{rendered}"
617 );
618 }
619
620 #[test]
625 fn test_ue3_operation_endpoint_names_top_line() {
626 let mut headers = Flattrs::new();
627 headers.set(OPERATION_ENDPOINT, "training.buffer.sample()".to_string());
628 headers.set(OPERATION_ADVERB, "call_one".to_string());
629 let envelope = make_envelope("payload", headers);
630 let rendered = format!(
631 "{}",
632 UndeliverableMessageError::DeliveryFailure { envelope }
633 );
634
635 let expected_line = "undeliverable message for training.buffer.sample() (call_one):";
636 assert!(
637 rendered.starts_with(expected_line),
638 "UE-3/UE-4: expected top line `{expected_line}`, got:\n{rendered}"
639 );
640 assert!(
644 !rendered.contains("undeliverable reply"),
645 "UE-4: must not claim reply-kind from header presence alone, got:\n{rendered}"
646 );
647 assert!(
648 !rendered.contains("undeliverable send"),
649 "UE-4: must not claim send-kind from header presence alone, got:\n{rendered}"
650 );
651 }
652
653 #[test]
657 fn test_ue3_delivery_failure_no_context_names_destination() {
658 let envelope = make_envelope("payload", Flattrs::new());
659 let dest_str = envelope.dest().to_string();
660 let rendered = format!(
661 "{}",
662 UndeliverableMessageError::DeliveryFailure { envelope }
663 );
664
665 let expected_prefix = format!("undeliverable message to {}", dest_str);
666 assert!(
667 rendered.starts_with(&expected_prefix),
668 "UE-3: delivery failure no context → destination prefix `{expected_prefix}`, got:\n{rendered}"
669 );
670 assert!(
672 !rendered.contains("undeliverable message error"),
673 "UE-3: neutral fallback must not be re-introduced, got:\n{rendered}"
674 );
675 }
676
677 #[test]
683 fn test_ue3_return_failure_no_context_names_original_sender() {
684 let envelope = make_envelope("payload", Flattrs::new());
685 let sender_str = envelope.sender().to_string();
686 let dest_str = envelope.dest().to_string();
687 let rendered = format!("{}", UndeliverableMessageError::ReturnFailure { envelope });
688
689 let expected_prefix = format!("undeliverable return to original sender {}", sender_str);
690 assert!(
691 rendered.starts_with(&expected_prefix),
692 "UE-3: return failure no context → original-sender prefix `{expected_prefix}`, got:\n{rendered}"
693 );
694 assert!(
698 !rendered.starts_with(&format!("undeliverable message to {}", dest_str)),
699 "UE-3: return failure must not headline the original destination, got:\n{rendered}"
700 );
701 assert!(
703 !rendered.contains("undeliverable message error"),
704 "UE-3: neutral fallback must not be re-introduced, got:\n{rendered}"
705 );
706 }
707
708 #[test]
712 fn test_ue5_message_type_falls_back_to_rust_message_type() {
713 let sender = test_actor_id("ue_proc", "ue_sender");
717 let dest = test_port_id("ue_dest_proc", "ue_dest", 42);
718 let mut headers = Flattrs::new();
719 headers.set(RUST_MESSAGE_TYPE, "my::Foo".to_string());
720 let envelope = MessageEnvelope::new(sender, dest, wirevalue::Any::new_broken(), headers);
721 assert!(
722 envelope.data().typename().is_none(),
723 "test fixture invariant: broken Any must have no typename()"
724 );
725
726 let rendered = format!(
727 "{}",
728 UndeliverableMessageError::DeliveryFailure { envelope }
729 );
730
731 assert!(
732 rendered.contains("\tmessage type: my::Foo\n"),
733 "UE-5: must surface RUST_MESSAGE_TYPE when typename() is absent, got:\n{rendered}"
734 );
735 assert!(
736 !rendered.contains("\tmessage type: unknown"),
737 "UE-5: must not render \"unknown\" when RUST_MESSAGE_TYPE is present, got:\n{rendered}"
738 );
739 }
740
741 #[test]
745 fn test_ue5_unknown_when_typename_and_rust_message_type_both_absent() {
746 let sender = test_actor_id("ue_proc", "ue_sender");
747 let dest = test_port_id("ue_dest_proc", "ue_dest", 42);
748 let envelope =
749 MessageEnvelope::new(sender, dest, wirevalue::Any::new_broken(), Flattrs::new());
750 assert!(
751 envelope.data().typename().is_none(),
752 "test fixture invariant: broken Any must have no typename()"
753 );
754
755 let rendered = format!(
756 "{}",
757 UndeliverableMessageError::DeliveryFailure { envelope }
758 );
759
760 assert!(
761 rendered.contains("\tmessage type: unknown\n"),
762 "UE-5: with no typename and no RUST_MESSAGE_TYPE, must render \"unknown\", got:\n{rendered}"
763 );
764 }
765}