hyperactor_mesh/introspect/
dto.rs

1/*
2 * Copyright (c) Meta Platforms, Inc. and affiliates.
3 * All rights reserved.
4 *
5 * This source code is licensed under the BSD-style license found in the
6 * LICENSE file in the root directory of this source tree.
7 */
8
9//! HTTP boundary DTO types for mesh-admin introspection.
10//!
11//! These types own the HTTP JSON wire contract. Domain types
12//! (`NodePayload`, `NodeProperties`, `FailureInfo`) stay clean of
13//! HTTP serialization concerns; conversion happens at the boundary
14//! via `From` / `TryFrom` impls defined here.
15//!
16//! ## Invariants
17//!
18//! - **HB-1 (typed-internal, string-external):** `NodeRef`, `ActorAddr`,
19//!   `ProcAddr`, and `SystemTime` are encoded as canonical strings in the
20//!   DTO types.
21//! - **HB-2 (round-trip):** `NodePayload → NodePayloadDto → NodePayload`
22//!   is lossless for values representable in the wire format.
23//!   Timestamps are formatted at millisecond precision
24//!   (`humantime::format_rfc3339_millis`), matching the established
25//!   HTTP contract; sub-millisecond precision is truncated at the
26//!   boundary.
27//! - **HB-3 (schema-honesty):** Schema/OpenAPI are generated from these
28//!   DTO types, so the published schema reflects the actual wire format.
29
30use std::time::SystemTime;
31
32use anyhow::Context;
33use schemars::JsonSchema;
34use serde::Deserialize;
35use serde::Serialize;
36
37use super::ActiveHandler;
38use super::Execution;
39use super::FailureInfo;
40use super::InboundOrdering;
41use super::NodePayload;
42use super::NodeProperties;
43use super::NodeRef;
44
45// DTO struct definitions
46
47/// Uniform response for any node in the mesh topology.
48///
49/// Every addressable entity (root, host, proc, actor) is represented
50/// as a `NodePayload`. The client navigates the mesh by fetching a
51/// node and following its `children` references.
52///
53/// `identity`, `children`, and `parent` are plain reference strings.
54/// `as_of` is an ISO 8601 timestamp string.
55#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
56#[schemars(title = "NodePayload")]
57pub struct NodePayloadDto {
58    /// Canonical node reference identifying this node.
59    pub identity: String,
60    /// Node-specific metadata (type, status, metrics, etc.).
61    pub properties: NodePropertiesDto,
62    /// Child node reference strings the client can URL-encode and
63    /// fetch via `GET /v1/{reference}`.
64    pub children: Vec<String>,
65    /// Parent node reference for upward navigation.
66    pub parent: Option<String>,
67    /// When this payload was captured (ISO 8601 timestamp string).
68    pub as_of: String,
69}
70
71/// Memory stats of the hosting OS process (DTO mirror of
72/// `ProcessMemoryStats`).
73#[derive(
74    Debug,
75    Clone,
76    Copy,
77    PartialEq,
78    Eq,
79    Default,
80    Serialize,
81    Deserialize,
82    JsonSchema
83)]
84#[schemars(rename = "ProcessMemoryStats")]
85pub struct ProcessMemoryStatsDto {
86    /// RSS of the hosting OS process (bytes).
87    pub process_rss_bytes: Option<u64>,
88    /// Virtual memory size of the hosting OS process (bytes).
89    pub process_vm_size_bytes: Option<u64>,
90}
91
92/// Proc-level debug/operational stats (DTO mirror of
93/// `ProcDebugStats`).
94#[derive(
95    Debug,
96    Clone,
97    Copy,
98    PartialEq,
99    Eq,
100    Default,
101    Serialize,
102    Deserialize,
103    JsonSchema
104)]
105#[schemars(rename = "ProcDebugStats")]
106pub struct ProcDebugStatsDto {
107    /// Hosting-process memory.
108    pub memory: ProcessMemoryStatsDto,
109    /// Sum of per-actor queue depths (live actors only).
110    pub actor_work_queue_depth_total: u64,
111    /// Max per-actor queue depth (live actors only).
112    pub actor_work_queue_depth_max: u64,
113    /// Maximum proc-wide queue depth since startup (PD-6, eventually consistent).
114    pub actor_work_queue_depth_high_water_mark: u64,
115    /// Milliseconds since proc-wide queue depth was last observed non-zero (PD-7, wall clock).
116    pub last_nonzero_queue_depth_age_ms: Option<u64>,
117}
118
119/// Node-specific metadata. Externally-tagged enum — the JSON
120/// key is the variant name (Root, Host, Proc, Actor, Error).
121#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
122#[schemars(rename = "NodeProperties")]
123pub enum NodePropertiesDto {
124    /// Synthetic mesh root node (not a real actor/proc).
125    Root {
126        num_hosts: usize,
127        started_at: String,
128        started_by: String,
129        system_children: Vec<String>,
130    },
131    /// A host in the mesh, represented by its `HostAgent`.
132    Host {
133        addr: String,
134        num_procs: usize,
135        system_children: Vec<String>,
136        /// Hosting-process memory stats.
137        memory: ProcessMemoryStatsDto,
138    },
139    /// Properties describing a proc running on a host.
140    Proc {
141        proc_name: String,
142        num_actors: usize,
143        system_children: Vec<String>,
144        stopped_children: Vec<String>,
145        stopped_retention_cap: usize,
146        is_poisoned: bool,
147        failed_actor_count: usize,
148        /// Runtime debug/operational stats.
149        debug: ProcDebugStatsDto,
150    },
151    /// Runtime metadata for a single actor instance.
152    Actor {
153        actor_status: String,
154        actor_type: String,
155        /// Stable per-instance Uuid::now_v7() identity assigned at
156        /// `Instance::new` (string form).
157        instance_id: String,
158        messages_processed: u64,
159        created_at: Option<String>,
160        last_message_handler: Option<String>,
161        total_processing_time_us: u64,
162        /// Accepted handler work not yet dequeued by the actor loop
163        /// (PD-5a/b in `hyperactor::proc`). Independent diagnostic from
164        /// `inbound_ordering`; no arithmetic contract -- see IO-3 in
165        /// `hyperactor::introspect`.
166        queue_depth: u64,
167        flight_recorder: Option<String>,
168        is_system: bool,
169        /// Per-session reorder-buffer state. `None` means no snapshot
170        /// callback installed (IO-1 structural absence);
171        /// `Some({enabled: false, ...})` means buffering disabled;
172        /// `Some({enabled: true, ...})` means active. See IO-1 in
173        /// `hyperactor::introspect`.
174        inbound_ordering: Option<Box<InboundOrderingDto>>,
175        failure_info: Option<FailureInfoDto>,
176        /// In-flight handler execution. `null` means the actor does not
177        /// report execution (unsupported), not idle. See EX-* in
178        /// `hyperactor_mesh::introspect`.
179        execution: Option<Box<ExecutionDto>>,
180    },
181    /// Error sentinel returned when a child reference cannot be resolved.
182    Error { code: String, message: String },
183}
184
185/// Structured failure information for failed actors.
186#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
187#[schemars(rename = "FailureInfo")]
188pub struct FailureInfoDto {
189    /// Error message describing the failure.
190    pub error_message: String,
191    /// Actor that caused the failure (root cause).
192    pub root_cause_actor: String,
193    /// Display name of the root-cause actor, if available.
194    pub root_cause_name: Option<String>,
195    /// When the failure occurred (ISO 8601 timestamp string).
196    pub occurred_at: String,
197    /// Whether this failure was propagated from a child.
198    pub is_propagated: bool,
199}
200
201/// Per-session reorder-buffer snapshot at the HTTP boundary.
202#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
203#[schemars(rename = "OrderingSessionSnapshot")]
204pub struct OrderingSessionSnapshotDto {
205    /// Session identifier (UUID string form).
206    pub session_id: String,
207    /// Session owner actor address (string form; `None` only in rare
208    /// bypass paths).
209    pub sender: Option<String>,
210    /// Highest seq released from the reorder buffer into the actor
211    /// work queue.
212    pub last_released_seq: u64,
213    /// `last_released_seq.saturating_add(1)`.
214    pub expected_next_seq: u64,
215    /// Messages held in the reorder buffer waiting for a seq gap.
216    pub buffered_count: usize,
217    /// Lowest seq currently buffered. `None` when `buffered_count == 0`.
218    pub oldest_buffered_seq: Option<u64>,
219    /// Highest seq currently buffered. `None` when `buffered_count == 0`.
220    pub newest_buffered_seq: Option<u64>,
221}
222
223/// Mesh-admin presentation of inbound ordering state.
224///
225/// When `snapshot_complete == false`, session state was unavailable.
226/// Consumers must ignore every session-derived field and retry (IO-6).
227#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
228#[schemars(rename = "InboundOrdering")]
229pub struct InboundOrderingDto {
230    /// `true` when reorder buffering is enabled for this sender.
231    pub enabled: bool,
232    /// IO-4: `true` iff `skipped_session_count == 0`.
233    pub snapshot_complete: bool,
234    /// Aggregate availability marker, not a count of sessions. Zero
235    /// means complete; nonzero means unavailable.
236    pub skipped_session_count: usize,
237    /// Exact live-session count when complete; ignore when unavailable.
238    pub known_session_count: usize,
239    /// Sessions with `buffered_count > 0` when complete; ignore when
240    /// unavailable.
241    pub returned_buffered_session_count: usize,
242    /// Sum of `buffered_count` when complete; ignore when unavailable.
243    /// Independent of `queue_depth` (IO-3).
244    pub returned_buffered_message_count: usize,
245    /// Max of `buffered_count` when complete; ignore when unavailable.
246    pub returned_max_buffered_count: usize,
247    /// Complete per-session entries when complete; ignore when unavailable.
248    pub sessions: Vec<OrderingSessionSnapshotDto>,
249}
250
251/// One handler with in-flight invocations, at the HTTP boundary.
252#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
253#[schemars(rename = "ActiveHandler")]
254pub struct ActiveHandlerDto {
255    /// Handler name (e.g. a Python endpoint method name).
256    pub name: String,
257    /// In-flight invocations of this handler.
258    pub active_count: u64,
259    /// Start time of the oldest in-flight invocation (ISO 8601 string).
260    pub oldest_since: String,
261}
262
263/// An actor's in-flight handler execution at the HTTP boundary.
264///
265/// `null` (absent) means the actor does not report execution
266/// (unsupported), not idle (EX-1). `complete == false` means the
267/// per-handler detail was momentarily unavailable on that read while
268/// `active_count` stays authoritative (EX-2).
269#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
270#[schemars(rename = "Execution")]
271pub struct ExecutionDto {
272    /// Handler invocations currently in flight (EX-2/EX-3).
273    pub active_count: u64,
274    /// Per-handler detail, oldest-first; a prefix of the N oldest when
275    /// `truncated` (EX-4).
276    pub active_handlers: Vec<ActiveHandlerDto>,
277    /// `true` iff the per-handler detail was captured on this read.
278    pub complete: bool,
279    /// `true` iff `active_handlers` is a prefix of the N oldest.
280    pub truncated: bool,
281}
282
283// Helpers
284
285fn format_time(t: &SystemTime) -> String {
286    humantime::format_rfc3339_millis(*t).to_string()
287}
288
289fn refs_to_strings(refs: &[NodeRef]) -> Vec<String> {
290    refs.iter().map(|r| r.to_string()).collect()
291}
292
293fn parse_refs(field: &str, strings: &[String]) -> anyhow::Result<Vec<NodeRef>> {
294    strings
295        .iter()
296        .enumerate()
297        .map(|(i, s)| {
298            s.parse()
299                .with_context(|| format!("failed to parse {field}[{i}]: {s:?}"))
300        })
301        .collect()
302}
303
304// Domain → DTO conversions (infallible)
305
306impl From<NodePayload> for NodePayloadDto {
307    fn from(p: NodePayload) -> Self {
308        Self {
309            identity: p.identity.to_string(),
310            properties: p.properties.into(),
311            children: refs_to_strings(&p.children),
312            parent: p.parent.as_ref().map(|r| r.to_string()),
313            as_of: format_time(&p.as_of),
314        }
315    }
316}
317
318impl From<NodeProperties> for NodePropertiesDto {
319    fn from(p: NodeProperties) -> Self {
320        match p {
321            NodeProperties::Root {
322                num_hosts,
323                started_at,
324                started_by,
325                system_children,
326            } => Self::Root {
327                num_hosts,
328                started_at: format_time(&started_at),
329                started_by,
330                system_children: refs_to_strings(&system_children),
331            },
332            NodeProperties::Host {
333                addr,
334                num_procs,
335                system_children,
336                memory,
337            } => Self::Host {
338                addr,
339                num_procs,
340                system_children: refs_to_strings(&system_children),
341                memory: ProcessMemoryStatsDto {
342                    process_rss_bytes: memory.process_rss_bytes,
343                    process_vm_size_bytes: memory.process_vm_size_bytes,
344                },
345            },
346            NodeProperties::Proc {
347                proc_name,
348                num_actors,
349                system_children,
350                stopped_children,
351                stopped_retention_cap,
352                is_poisoned,
353                failed_actor_count,
354                debug,
355            } => Self::Proc {
356                proc_name,
357                num_actors,
358                system_children: refs_to_strings(&system_children),
359                stopped_children: refs_to_strings(&stopped_children),
360                stopped_retention_cap,
361                is_poisoned,
362                failed_actor_count,
363                debug: ProcDebugStatsDto {
364                    memory: ProcessMemoryStatsDto {
365                        process_rss_bytes: debug.memory.process_rss_bytes,
366                        process_vm_size_bytes: debug.memory.process_vm_size_bytes,
367                    },
368                    actor_work_queue_depth_total: debug.actor_work_queue_depth_total,
369                    actor_work_queue_depth_max: debug.actor_work_queue_depth_max,
370                    actor_work_queue_depth_high_water_mark: debug
371                        .actor_work_queue_depth_high_water_mark,
372                    last_nonzero_queue_depth_age_ms: debug.last_nonzero_queue_depth_age_ms,
373                },
374            },
375            NodeProperties::Actor {
376                actor_status,
377                actor_type,
378                instance_id,
379                messages_processed,
380                created_at,
381                last_message_handler,
382                total_processing_time_us,
383                queue_depth,
384                flight_recorder,
385                is_system,
386                inbound_ordering,
387                failure_info,
388                execution,
389            } => Self::Actor {
390                actor_status,
391                actor_type,
392                instance_id,
393                messages_processed,
394                created_at: created_at.as_ref().map(format_time),
395                last_message_handler,
396                total_processing_time_us,
397                queue_depth,
398                flight_recorder,
399                is_system,
400                inbound_ordering: inbound_ordering
401                    .map(|io| Box::new(InboundOrderingDto::from(*io))),
402                failure_info: failure_info.map(Into::into),
403                execution: execution.map(|e| Box::new(ExecutionDto::from(*e))),
404            },
405            NodeProperties::Error { code, message } => Self::Error { code, message },
406        }
407    }
408}
409
410impl From<FailureInfo> for FailureInfoDto {
411    fn from(f: FailureInfo) -> Self {
412        Self {
413            error_message: f.error_message,
414            root_cause_actor: f.root_cause_actor.to_string(),
415            root_cause_name: f.root_cause_name,
416            occurred_at: format_time(&f.occurred_at),
417            is_propagated: f.is_propagated,
418        }
419    }
420}
421
422impl From<hyperactor::ordering::OrderingSessionSnapshot> for OrderingSessionSnapshotDto {
423    fn from(s: hyperactor::ordering::OrderingSessionSnapshot) -> Self {
424        Self {
425            session_id: s.session_id.to_string(),
426            sender: s.sender.as_ref().map(|a| a.to_string()),
427            last_released_seq: s.last_released_seq,
428            expected_next_seq: s.expected_next_seq,
429            buffered_count: s.buffered_count,
430            oldest_buffered_seq: s.oldest_buffered_seq,
431            newest_buffered_seq: s.newest_buffered_seq,
432        }
433    }
434}
435
436impl From<InboundOrdering> for InboundOrderingDto {
437    fn from(o: InboundOrdering) -> Self {
438        Self {
439            enabled: o.enabled,
440            snapshot_complete: o.snapshot_complete,
441            skipped_session_count: o.skipped_session_count,
442            known_session_count: o.known_session_count,
443            returned_buffered_session_count: o.returned_buffered_session_count,
444            returned_buffered_message_count: o.returned_buffered_message_count,
445            returned_max_buffered_count: o.returned_max_buffered_count,
446            sessions: o.sessions.into_iter().map(Into::into).collect(),
447        }
448    }
449}
450
451impl From<Execution> for ExecutionDto {
452    fn from(e: Execution) -> Self {
453        Self {
454            active_count: e.active_count,
455            active_handlers: e.active_handlers.into_iter().map(Into::into).collect(),
456            complete: e.complete,
457            truncated: e.truncated,
458        }
459    }
460}
461
462impl From<ActiveHandler> for ActiveHandlerDto {
463    fn from(h: ActiveHandler) -> Self {
464        Self {
465            name: h.name,
466            active_count: h.active_count,
467            oldest_since: format_time(&h.oldest_since),
468        }
469    }
470}
471
472// DTO → Domain conversions (fallible)
473
474impl TryFrom<NodePayloadDto> for NodePayload {
475    type Error = anyhow::Error;
476
477    fn try_from(dto: NodePayloadDto) -> Result<Self, Self::Error> {
478        Ok(Self {
479            identity: dto
480                .identity
481                .parse()
482                .with_context(|| format!("failed to parse identity: {:?}", dto.identity))?,
483            properties: dto
484                .properties
485                .try_into()
486                .context("failed to parse properties")?,
487            children: parse_refs("children", &dto.children)?,
488            parent: dto
489                .parent
490                .map(|s| {
491                    s.parse()
492                        .with_context(|| format!("failed to parse parent: {s:?}"))
493                })
494                .transpose()?,
495            as_of: humantime::parse_rfc3339(&dto.as_of)
496                .with_context(|| format!("failed to parse as_of: {:?}", dto.as_of))?,
497        })
498    }
499}
500
501impl TryFrom<NodePropertiesDto> for NodeProperties {
502    type Error = anyhow::Error;
503
504    fn try_from(
505        dto: NodePropertiesDto,
506    ) -> Result<Self, <Self as TryFrom<NodePropertiesDto>>::Error> {
507        Ok(match dto {
508            NodePropertiesDto::Root {
509                num_hosts,
510                started_at,
511                started_by,
512                system_children,
513            } => Self::Root {
514                num_hosts,
515                started_at: humantime::parse_rfc3339(&started_at)
516                    .context("failed to parse Root.started_at")?,
517                started_by,
518                system_children: parse_refs("Root.system_children", &system_children)?,
519            },
520            NodePropertiesDto::Host {
521                addr,
522                num_procs,
523                system_children,
524                memory,
525            } => Self::Host {
526                addr,
527                num_procs,
528                system_children: parse_refs("Host.system_children", &system_children)?,
529                memory: super::ProcessMemoryStats {
530                    process_rss_bytes: memory.process_rss_bytes,
531                    process_vm_size_bytes: memory.process_vm_size_bytes,
532                },
533            },
534            NodePropertiesDto::Proc {
535                proc_name,
536                num_actors,
537                system_children,
538                stopped_children,
539                stopped_retention_cap,
540                is_poisoned,
541                failed_actor_count,
542                debug,
543            } => Self::Proc {
544                proc_name,
545                num_actors,
546                system_children: parse_refs("Proc.system_children", &system_children)?,
547                stopped_children: parse_refs("Proc.stopped_children", &stopped_children)?,
548                stopped_retention_cap,
549                is_poisoned,
550                failed_actor_count,
551                debug: super::ProcDebugStats {
552                    memory: super::ProcessMemoryStats {
553                        process_rss_bytes: debug.memory.process_rss_bytes,
554                        process_vm_size_bytes: debug.memory.process_vm_size_bytes,
555                    },
556                    actor_work_queue_depth_total: debug.actor_work_queue_depth_total,
557                    actor_work_queue_depth_max: debug.actor_work_queue_depth_max,
558                    actor_work_queue_depth_high_water_mark: debug
559                        .actor_work_queue_depth_high_water_mark,
560                    last_nonzero_queue_depth_age_ms: debug.last_nonzero_queue_depth_age_ms,
561                },
562            },
563            NodePropertiesDto::Actor {
564                actor_status,
565                actor_type,
566                instance_id,
567                messages_processed,
568                created_at,
569                last_message_handler,
570                total_processing_time_us,
571                queue_depth,
572                flight_recorder,
573                is_system,
574                inbound_ordering,
575                failure_info,
576                execution,
577            } => Self::Actor {
578                actor_status,
579                actor_type,
580                instance_id,
581                messages_processed,
582                created_at: created_at
583                    .map(|s| {
584                        humantime::parse_rfc3339(&s)
585                            .with_context(|| format!("failed to parse Actor.created_at: {s:?}"))
586                    })
587                    .transpose()?,
588                last_message_handler,
589                total_processing_time_us,
590                queue_depth,
591                flight_recorder,
592                is_system,
593                inbound_ordering: inbound_ordering
594                    .map(|dto| InboundOrdering::try_from(*dto).map(Box::new))
595                    .transpose()
596                    .context("failed to parse Actor.inbound_ordering")?,
597                failure_info: failure_info
598                    .map(TryInto::try_into)
599                    .transpose()
600                    .context("failed to parse Actor.failure_info")?,
601                execution: execution
602                    .map(|dto| Execution::try_from(*dto).map(Box::new))
603                    .transpose()
604                    .context("failed to parse Actor.execution")?,
605            },
606            NodePropertiesDto::Error { code, message } => Self::Error { code, message },
607        })
608    }
609}
610
611impl TryFrom<FailureInfoDto> for FailureInfo {
612    type Error = anyhow::Error;
613
614    fn try_from(dto: FailureInfoDto) -> Result<Self, Self::Error> {
615        Ok(Self {
616            error_message: dto.error_message,
617            root_cause_actor: dto.root_cause_actor.parse().with_context(|| {
618                format!(
619                    "failed to parse FailureInfo.root_cause_actor: {:?}",
620                    dto.root_cause_actor
621                )
622            })?,
623            root_cause_name: dto.root_cause_name,
624            occurred_at: humantime::parse_rfc3339(&dto.occurred_at).with_context(|| {
625                format!(
626                    "failed to parse FailureInfo.occurred_at: {:?}",
627                    dto.occurred_at
628                )
629            })?,
630            is_propagated: dto.is_propagated,
631        })
632    }
633}
634
635impl TryFrom<ExecutionDto> for Execution {
636    type Error = anyhow::Error;
637
638    fn try_from(dto: ExecutionDto) -> Result<Self, Self::Error> {
639        Ok(Self {
640            active_count: dto.active_count,
641            active_handlers: dto
642                .active_handlers
643                .into_iter()
644                .map(ActiveHandler::try_from)
645                .collect::<Result<Vec<_>, _>>()?,
646            complete: dto.complete,
647            truncated: dto.truncated,
648        })
649    }
650}
651
652impl TryFrom<ActiveHandlerDto> for ActiveHandler {
653    type Error = anyhow::Error;
654
655    fn try_from(dto: ActiveHandlerDto) -> Result<Self, Self::Error> {
656        Ok(Self {
657            name: dto.name,
658            active_count: dto.active_count,
659            oldest_since: humantime::parse_rfc3339(&dto.oldest_since).with_context(|| {
660                format!(
661                    "failed to parse ActiveHandler.oldest_since: {:?}",
662                    dto.oldest_since
663                )
664            })?,
665        })
666    }
667}
668
669impl TryFrom<OrderingSessionSnapshotDto> for hyperactor::ordering::OrderingSessionSnapshot {
670    type Error = anyhow::Error;
671
672    fn try_from(dto: OrderingSessionSnapshotDto) -> Result<Self, Self::Error> {
673        let session_id = dto.session_id.parse().with_context(|| {
674            format!(
675                "failed to parse OrderingSessionSnapshot.session_id: {:?}",
676                dto.session_id
677            )
678        })?;
679        let sender = dto
680            .sender
681            .as_ref()
682            .map(|s| {
683                s.parse().with_context(|| {
684                    format!("failed to parse OrderingSessionSnapshot.sender: {s:?}")
685                })
686            })
687            .transpose()?;
688        Ok(Self {
689            session_id,
690            sender,
691            last_released_seq: dto.last_released_seq,
692            expected_next_seq: dto.expected_next_seq,
693            buffered_count: dto.buffered_count,
694            oldest_buffered_seq: dto.oldest_buffered_seq,
695            newest_buffered_seq: dto.newest_buffered_seq,
696        })
697    }
698}
699
700impl TryFrom<InboundOrderingDto> for InboundOrdering {
701    type Error = anyhow::Error;
702
703    fn try_from(dto: InboundOrderingDto) -> Result<Self, Self::Error> {
704        let sessions: Vec<hyperactor::ordering::OrderingSessionSnapshot> = dto
705            .sessions
706            .into_iter()
707            .enumerate()
708            .map(|(i, s)| {
709                s.try_into()
710                    .with_context(|| format!("failed to parse InboundOrdering.sessions[{i}]"))
711            })
712            .collect::<Result<_, _>>()?;
713        Ok(Self {
714            enabled: dto.enabled,
715            snapshot_complete: dto.snapshot_complete,
716            skipped_session_count: dto.skipped_session_count,
717            known_session_count: dto.known_session_count,
718            returned_buffered_session_count: dto.returned_buffered_session_count,
719            returned_buffered_message_count: dto.returned_buffered_message_count,
720            returned_max_buffered_count: dto.returned_max_buffered_count,
721            sessions,
722        })
723    }
724}
725
726#[cfg(test)]
727mod tests {
728    use super::*;
729    use crate::mesh_id::ResourceId;
730
731    // Test fixtures
732
733    fn test_proc_id() -> hyperactor::ProcAddr {
734        ResourceId::proc_addr_from_name(hyperactor::channel::ChannelAddr::Local(0), "worker")
735    }
736
737    fn test_actor_id() -> hyperactor::ActorAddr {
738        test_proc_id().actor_addr("actor")
739    }
740
741    fn test_host_actor_id() -> hyperactor::ActorAddr {
742        test_proc_id().actor_addr("host_agent")
743    }
744
745    fn test_time() -> SystemTime {
746        humantime::parse_rfc3339("2025-01-15T10:30:00.123Z").unwrap()
747    }
748
749    fn test_time_2() -> SystemTime {
750        humantime::parse_rfc3339("2025-01-15T11:00:00.456Z").unwrap()
751    }
752
753    fn make_root_payload() -> NodePayload {
754        NodePayload {
755            identity: NodeRef::Root,
756            properties: NodeProperties::Root {
757                num_hosts: 2,
758                started_at: test_time(),
759                started_by: "test_user".to_string(),
760                system_children: vec![],
761            },
762            children: vec![NodeRef::Host(test_host_actor_id())],
763            parent: None,
764            as_of: test_time(),
765        }
766    }
767
768    fn make_host_payload() -> NodePayload {
769        NodePayload {
770            identity: NodeRef::Host(test_host_actor_id()),
771            properties: NodeProperties::Host {
772                addr: "127.0.0.1:8080".to_string(),
773                num_procs: 1,
774                system_children: vec![],
775                memory: Default::default(),
776            },
777            children: vec![NodeRef::Proc(test_proc_id())],
778            parent: Some(NodeRef::Root),
779            as_of: test_time(),
780        }
781    }
782
783    fn make_proc_payload() -> NodePayload {
784        NodePayload {
785            identity: NodeRef::Proc(test_proc_id()),
786            properties: NodeProperties::Proc {
787                proc_name: "worker".to_string(),
788                num_actors: 3,
789                system_children: vec![NodeRef::Actor(test_actor_id())],
790                stopped_children: vec![],
791                stopped_retention_cap: 100,
792                is_poisoned: false,
793                failed_actor_count: 0,
794                debug: Default::default(),
795            },
796            children: vec![NodeRef::Actor(test_actor_id())],
797            parent: Some(NodeRef::Host(test_host_actor_id())),
798            as_of: test_time(),
799        }
800    }
801
802    fn test_instance_id() -> String {
803        // Stable test UUID; round-trip tests don't care about the value.
804        "01900000-0000-7000-8000-000000000001".to_string()
805    }
806
807    fn make_actor_payload_no_failure() -> NodePayload {
808        NodePayload {
809            identity: NodeRef::Actor(test_actor_id()),
810            properties: NodeProperties::Actor {
811                actor_status: "running".to_string(),
812                actor_type: "MyActor".to_string(),
813                instance_id: test_instance_id(),
814                messages_processed: 42,
815                created_at: Some(test_time()),
816                last_message_handler: Some("handle_msg".to_string()),
817                total_processing_time_us: 1500,
818                queue_depth: 0,
819                flight_recorder: None,
820                is_system: false,
821                inbound_ordering: None,
822                failure_info: None,
823                execution: None,
824            },
825            children: vec![],
826            parent: Some(NodeRef::Proc(test_proc_id())),
827            as_of: test_time(),
828        }
829    }
830
831    fn make_actor_payload_with_failure() -> NodePayload {
832        NodePayload {
833            identity: NodeRef::Actor(test_actor_id()),
834            properties: NodeProperties::Actor {
835                actor_status: "failed".to_string(),
836                actor_type: "MyActor".to_string(),
837                instance_id: test_instance_id(),
838                messages_processed: 10,
839                created_at: Some(test_time()),
840                last_message_handler: None,
841                total_processing_time_us: 500,
842                queue_depth: 0,
843                flight_recorder: Some("trace-abc".to_string()),
844                is_system: true,
845                inbound_ordering: None,
846                failure_info: Some(FailureInfo {
847                    error_message: "boom".to_string(),
848                    root_cause_actor: test_actor_id(),
849                    root_cause_name: Some("root_actor".to_string()),
850                    occurred_at: test_time_2(),
851                    is_propagated: true,
852                }),
853                execution: None,
854            },
855            children: vec![],
856            parent: Some(NodeRef::Proc(test_proc_id())),
857            as_of: test_time(),
858        }
859    }
860
861    fn make_actor_payload_minimal() -> NodePayload {
862        NodePayload {
863            identity: NodeRef::Actor(test_actor_id()),
864            properties: NodeProperties::Actor {
865                actor_status: "idle".to_string(),
866                actor_type: "MinimalActor".to_string(),
867                instance_id: test_instance_id(),
868                messages_processed: 0,
869                created_at: None,
870                last_message_handler: None,
871                total_processing_time_us: 0,
872                queue_depth: 0,
873                flight_recorder: None,
874                is_system: false,
875                inbound_ordering: None,
876                failure_info: None,
877                execution: None,
878            },
879            children: vec![],
880            parent: Some(NodeRef::Proc(test_proc_id())),
881            as_of: test_time(),
882        }
883    }
884
885    fn make_ordering_session(
886        session_id: uuid::Uuid,
887        last_released_seq: u64,
888        buffered_count: usize,
889    ) -> hyperactor::ordering::OrderingSessionSnapshot {
890        let (oldest, newest) = if buffered_count > 0 {
891            (
892                Some(last_released_seq + 2),
893                Some(last_released_seq + 1 + buffered_count as u64),
894            )
895        } else {
896            (None, None)
897        };
898        hyperactor::ordering::OrderingSessionSnapshot {
899            session_id,
900            sender: Some(test_actor_id()),
901            last_released_seq,
902            expected_next_seq: last_released_seq.saturating_add(1),
903            buffered_count,
904            oldest_buffered_seq: oldest,
905            newest_buffered_seq: newest,
906        }
907    }
908
909    fn make_actor_payload_inbound_ordering_complete() -> NodePayload {
910        NodePayload {
911            identity: NodeRef::Actor(test_actor_id()),
912            properties: NodeProperties::Actor {
913                actor_status: "running".to_string(),
914                actor_type: "MyActor".to_string(),
915                instance_id: test_instance_id(),
916                messages_processed: 17,
917                created_at: Some(test_time()),
918                last_message_handler: Some("handle_msg".to_string()),
919                total_processing_time_us: 900,
920                queue_depth: 5,
921                flight_recorder: None,
922                is_system: false,
923                inbound_ordering: Some(Box::new(InboundOrdering {
924                    enabled: true,
925                    snapshot_complete: true,
926                    skipped_session_count: 0,
927                    known_session_count: 2,
928                    returned_buffered_session_count: 1,
929                    returned_buffered_message_count: 3,
930                    returned_max_buffered_count: 3,
931                    sessions: vec![
932                        make_ordering_session(uuid::Uuid::from_u128(1), 7, 0),
933                        make_ordering_session(uuid::Uuid::from_u128(2), 1, 3),
934                    ],
935                })),
936                failure_info: None,
937                execution: None,
938            },
939            children: vec![],
940            parent: Some(NodeRef::Proc(test_proc_id())),
941            as_of: test_time(),
942        }
943    }
944
945    fn make_actor_payload_inbound_ordering_unavailable() -> NodePayload {
946        NodePayload {
947            identity: NodeRef::Actor(test_actor_id()),
948            properties: NodeProperties::Actor {
949                actor_status: "running".to_string(),
950                actor_type: "MyActor".to_string(),
951                instance_id: test_instance_id(),
952                messages_processed: 17,
953                created_at: Some(test_time()),
954                last_message_handler: Some("handle_msg".to_string()),
955                total_processing_time_us: 900,
956                queue_depth: 5,
957                flight_recorder: None,
958                is_system: false,
959                inbound_ordering: Some(Box::new(InboundOrdering {
960                    enabled: true,
961                    snapshot_complete: false,
962                    skipped_session_count: 1,
963                    known_session_count: 1,
964                    returned_buffered_session_count: 0,
965                    returned_buffered_message_count: 0,
966                    returned_max_buffered_count: 0,
967                    sessions: vec![],
968                })),
969                failure_info: None,
970                execution: None,
971            },
972            children: vec![],
973            parent: Some(NodeRef::Proc(test_proc_id())),
974            as_of: test_time(),
975        }
976    }
977
978    fn make_error_payload() -> NodePayload {
979        NodePayload {
980            identity: NodeRef::Actor(test_actor_id()),
981            properties: NodeProperties::Error {
982                code: "not_found".to_string(),
983                message: "actor not found".to_string(),
984            },
985            children: vec![],
986            parent: None,
987            as_of: test_time(),
988        }
989    }
990
991    // HB-2 (round-trip): NodePayload → NodePayloadDto → NodePayload is
992    // lossless for values representable in the wire format.
993
994    fn assert_round_trip(payload: &NodePayload) {
995        let dto: NodePayloadDto = payload.clone().into();
996        let back: NodePayload = dto.try_into().expect("round-trip conversion");
997        assert_eq!(payload, &back);
998    }
999
1000    /// HB-2: Root variant round-trips.
1001    #[test]
1002    fn test_round_trip_root() {
1003        assert_round_trip(&make_root_payload());
1004    }
1005
1006    /// HB-2: Host variant round-trips.
1007    #[test]
1008    fn test_round_trip_host() {
1009        assert_round_trip(&make_host_payload());
1010    }
1011
1012    /// HB-2: Proc variant round-trips.
1013    #[test]
1014    fn test_round_trip_proc() {
1015        assert_round_trip(&make_proc_payload());
1016    }
1017
1018    /// HB-2: Actor variant without failure round-trips.
1019    #[test]
1020    fn test_round_trip_actor_no_failure() {
1021        assert_round_trip(&make_actor_payload_no_failure());
1022    }
1023
1024    /// HB-2: Actor variant with failure round-trips.
1025    #[test]
1026    fn test_round_trip_actor_with_failure() {
1027        assert_round_trip(&make_actor_payload_with_failure());
1028    }
1029
1030    /// HB-2: Actor variant with all optional fields absent round-trips.
1031    #[test]
1032    fn test_round_trip_actor_minimal() {
1033        assert_round_trip(&make_actor_payload_minimal());
1034    }
1035
1036    /// HB-1 + HB-2 + IO-4 + IO-5: Actor with a complete inbound-ordering
1037    /// snapshot round-trips. Exercises the `Some({enabled: true,
1038    /// snapshot_complete: true, ...})` branch of IO-1 plus the IO-4
1039    /// derivation and IO-5 totality (`known_session_count ==
1040    /// sessions.len()` when no skipped).
1041    #[test]
1042    fn test_round_trip_actor_inbound_ordering_complete() {
1043        let payload = make_actor_payload_inbound_ordering_complete();
1044        // Assert the fixture itself satisfies IO-4 / IO-5 before
1045        // round-tripping, so a bug in the fixture can't pass for the
1046        // wrong reason.
1047        if let NodeProperties::Actor {
1048            inbound_ordering: Some(io),
1049            ..
1050        } = &payload.properties
1051        {
1052            assert_eq!(io.snapshot_complete, io.skipped_session_count == 0); // IO-4
1053            assert_eq!(io.known_session_count, io.sessions.len()); // IO-5
1054        } else {
1055            panic!("fixture must be Actor with Some(inbound_ordering)");
1056        }
1057        assert_round_trip(&payload);
1058    }
1059
1060    /// HB-1 + HB-2 + IO-4 + IO-6: Actor with unavailable inbound
1061    /// ordering round-trips without interpreting session-derived fields.
1062    #[test]
1063    fn test_round_trip_actor_inbound_ordering_unavailable() {
1064        let payload = make_actor_payload_inbound_ordering_unavailable();
1065        if let NodeProperties::Actor {
1066            inbound_ordering: Some(io),
1067            ..
1068        } = &payload.properties
1069        {
1070            assert!(!io.snapshot_complete);
1071            assert_eq!(io.snapshot_complete, io.skipped_session_count == 0);
1072        } else {
1073            panic!("fixture must be Actor with Some(inbound_ordering)");
1074        }
1075        assert_round_trip(&payload);
1076    }
1077
1078    /// HB-2: Error variant round-trips.
1079    #[test]
1080    fn test_round_trip_error() {
1081        assert_round_trip(&make_error_payload());
1082    }
1083
1084    // HB-1 (typed-internal, string-external): typed Rust values serialize
1085    // as canonical strings in the DTO JSON output.
1086
1087    /// HB-1: Root identity, children, parent, and timestamps serialize
1088    /// as strings; externally-tagged enum key is "Root".
1089    #[test]
1090    fn test_json_shape_root() {
1091        let dto: NodePayloadDto = make_root_payload().into();
1092        let json = serde_json::to_value(&dto).unwrap();
1093
1094        assert_eq!(json["identity"], "root");
1095        assert!(json["parent"].is_null());
1096        assert_eq!(json["as_of"], "2025-01-15T10:30:00.123Z");
1097
1098        let children = json["children"].as_array().unwrap();
1099        assert_eq!(children.len(), 1);
1100        assert_eq!(children[0], format!("host:{}", test_host_actor_id()));
1101
1102        let root = &json["properties"]["Root"];
1103        assert_eq!(root["num_hosts"], 2);
1104        assert_eq!(root["started_at"], "2025-01-15T10:30:00.123Z");
1105        assert_eq!(root["started_by"], "test_user");
1106        assert!(root["system_children"].as_array().unwrap().is_empty());
1107    }
1108
1109    /// HB-1: Actor variant with failure — ActorAddr, SystemTime, and
1110    /// nested FailureInfo fields all serialize as strings.
1111    #[test]
1112    fn test_json_shape_actor_with_failure() {
1113        let dto: NodePayloadDto = make_actor_payload_with_failure().into();
1114        let json = serde_json::to_value(&dto).unwrap();
1115
1116        assert_eq!(json["identity"], test_actor_id().to_string());
1117        assert_eq!(json["parent"], test_proc_id().to_string());
1118
1119        let actor = &json["properties"]["Actor"];
1120        assert_eq!(actor["actor_status"], "failed");
1121        assert_eq!(actor["messages_processed"], 10);
1122        assert_eq!(actor["created_at"], "2025-01-15T10:30:00.123Z");
1123        assert!(actor["last_message_handler"].is_null());
1124        assert_eq!(actor["flight_recorder"], "trace-abc");
1125        assert_eq!(actor["is_system"], true);
1126
1127        let fi = &actor["failure_info"];
1128        assert_eq!(fi["error_message"], "boom");
1129        assert_eq!(fi["root_cause_actor"], test_actor_id().to_string());
1130        assert_eq!(fi["root_cause_name"], "root_actor");
1131        assert_eq!(fi["occurred_at"], "2025-01-15T11:00:00.456Z");
1132        assert_eq!(fi["is_propagated"], true);
1133    }
1134
1135    /// HB-1: Option fields serialize as JSON null when absent.
1136    #[test]
1137    fn test_json_shape_optional_none_fields() {
1138        let dto: NodePayloadDto = make_actor_payload_minimal().into();
1139        let json = serde_json::to_value(&dto).unwrap();
1140
1141        let actor = &json["properties"]["Actor"];
1142        assert!(actor["created_at"].is_null());
1143        assert!(actor["last_message_handler"].is_null());
1144        assert!(actor["flight_recorder"].is_null());
1145        assert!(actor["failure_info"].is_null());
1146    }
1147
1148    /// HB-1: Error variant preserves code/message as plain strings.
1149    #[test]
1150    fn test_json_shape_error() {
1151        let dto: NodePayloadDto = make_error_payload().into();
1152        let json = serde_json::to_value(&dto).unwrap();
1153
1154        let err = &json["properties"]["Error"];
1155        assert_eq!(err["code"], "not_found");
1156        assert_eq!(err["message"], "actor not found");
1157    }
1158
1159    /// HB-1: Empty children vec serializes as `[]`.
1160    #[test]
1161    fn test_json_shape_empty_children() {
1162        let dto: NodePayloadDto = make_actor_payload_no_failure().into();
1163        let json = serde_json::to_value(&dto).unwrap();
1164        assert!(json["children"].as_array().unwrap().is_empty());
1165    }
1166
1167    // HB-3 (schema-honesty): published schema reflects the actual wire
1168    // format. The schemars(rename/title) attributes must produce $defs
1169    // keys and title matching the domain type names, not the Dto suffixes.
1170
1171    /// HB-3: $defs keys are "NodeProperties" and "FailureInfo", not
1172    /// "NodePropertiesDto" / "FailureInfoDto".
1173    #[test]
1174    fn test_schema_defs_keys() {
1175        let schema = schemars::schema_for!(NodePayloadDto);
1176        let json = serde_json::to_value(&schema).unwrap();
1177        let defs = json["$defs"].as_object().unwrap();
1178        assert!(
1179            defs.contains_key("NodeProperties"),
1180            "$defs must contain 'NodeProperties', got: {:?}",
1181            defs.keys().collect::<Vec<_>>()
1182        );
1183        assert!(
1184            defs.contains_key("FailureInfo"),
1185            "$defs must contain 'FailureInfo', got: {:?}",
1186            defs.keys().collect::<Vec<_>>()
1187        );
1188    }
1189
1190    /// HB-3: Top-level schema title is "NodePayload", not
1191    /// "NodePayloadDto".
1192    #[test]
1193    fn test_schema_title() {
1194        let schema = schemars::schema_for!(NodePayloadDto);
1195        let json = serde_json::to_value(&schema).unwrap();
1196        assert_eq!(json["title"], "NodePayload");
1197    }
1198}
hyperactor_mesh/introspect/dto.rs

hyperactor_mesh/introspect/
dto.rs
