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/// Rollups marked `returned_*` are computed over `sessions` only and
226/// are LOWER BOUNDS when `snapshot_complete == false` (IO-6).
227/// `known_session_count` is the only rollup that totals returned +
228/// skipped sessions.
229#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
230#[schemars(rename = "InboundOrdering")]
231pub struct InboundOrderingDto {
232    /// `true` when reorder buffering is enabled for this sender.
233    pub enabled: bool,
234    /// IO-4: `true` iff `skipped_session_count == 0`.
235    pub snapshot_complete: bool,
236    /// Sessions held by a concurrent send at snapshot time.
237    pub skipped_session_count: usize,
238    /// IO-5: `sessions.len() + skipped_session_count`. The only rollup
239    /// that totals returned + skipped sessions.
240    pub known_session_count: usize,
241    /// IO-6: sessions with `buffered_count > 0` AMONG RETURNED
242    /// sessions. Lower bound if `!snapshot_complete`.
243    pub returned_buffered_session_count: usize,
244    /// IO-6: sum of `buffered_count` OVER RETURNED sessions.
245    /// Reorder-buffer scope only (independent of `queue_depth`; see
246    /// IO-3). Lower bound if `!snapshot_complete`.
247    pub returned_buffered_message_count: usize,
248    /// IO-6: max of `buffered_count` OVER RETURNED sessions. Lower
249    /// bound if `!snapshot_complete`.
250    pub returned_max_buffered_count: usize,
251    /// Per-session entries, sorted by `session_id`.
252    pub sessions: Vec<OrderingSessionSnapshotDto>,
253}
254
255/// One handler with in-flight invocations, at the HTTP boundary.
256#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
257#[schemars(rename = "ActiveHandler")]
258pub struct ActiveHandlerDto {
259    /// Handler name (e.g. a Python endpoint method name).
260    pub name: String,
261    /// In-flight invocations of this handler.
262    pub active_count: u64,
263    /// Start time of the oldest in-flight invocation (ISO 8601 string).
264    pub oldest_since: String,
265}
266
267/// An actor's in-flight handler execution at the HTTP boundary.
268///
269/// `null` (absent) means the actor does not report execution
270/// (unsupported), not idle (EX-1). `complete == false` means the
271/// per-handler detail was momentarily unavailable on that read while
272/// `active_count` stays authoritative (EX-2).
273#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
274#[schemars(rename = "Execution")]
275pub struct ExecutionDto {
276    /// Handler invocations currently in flight (EX-2/EX-3).
277    pub active_count: u64,
278    /// Per-handler detail, oldest-first; a prefix of the N oldest when
279    /// `truncated` (EX-4).
280    pub active_handlers: Vec<ActiveHandlerDto>,
281    /// `true` iff the per-handler detail was captured on this read.
282    pub complete: bool,
283    /// `true` iff `active_handlers` is a prefix of the N oldest.
284    pub truncated: bool,
285}
286
287// Helpers
288
289fn format_time(t: &SystemTime) -> String {
290    humantime::format_rfc3339_millis(*t).to_string()
291}
292
293fn refs_to_strings(refs: &[NodeRef]) -> Vec<String> {
294    refs.iter().map(|r| r.to_string()).collect()
295}
296
297fn parse_refs(field: &str, strings: &[String]) -> anyhow::Result<Vec<NodeRef>> {
298    strings
299        .iter()
300        .enumerate()
301        .map(|(i, s)| {
302            s.parse()
303                .with_context(|| format!("failed to parse {field}[{i}]: {s:?}"))
304        })
305        .collect()
306}
307
308// Domain → DTO conversions (infallible)
309
310impl From<NodePayload> for NodePayloadDto {
311    fn from(p: NodePayload) -> Self {
312        Self {
313            identity: p.identity.to_string(),
314            properties: p.properties.into(),
315            children: refs_to_strings(&p.children),
316            parent: p.parent.as_ref().map(|r| r.to_string()),
317            as_of: format_time(&p.as_of),
318        }
319    }
320}
321
322impl From<NodeProperties> for NodePropertiesDto {
323    fn from(p: NodeProperties) -> Self {
324        match p {
325            NodeProperties::Root {
326                num_hosts,
327                started_at,
328                started_by,
329                system_children,
330            } => Self::Root {
331                num_hosts,
332                started_at: format_time(&started_at),
333                started_by,
334                system_children: refs_to_strings(&system_children),
335            },
336            NodeProperties::Host {
337                addr,
338                num_procs,
339                system_children,
340                memory,
341            } => Self::Host {
342                addr,
343                num_procs,
344                system_children: refs_to_strings(&system_children),
345                memory: ProcessMemoryStatsDto {
346                    process_rss_bytes: memory.process_rss_bytes,
347                    process_vm_size_bytes: memory.process_vm_size_bytes,
348                },
349            },
350            NodeProperties::Proc {
351                proc_name,
352                num_actors,
353                system_children,
354                stopped_children,
355                stopped_retention_cap,
356                is_poisoned,
357                failed_actor_count,
358                debug,
359            } => Self::Proc {
360                proc_name,
361                num_actors,
362                system_children: refs_to_strings(&system_children),
363                stopped_children: refs_to_strings(&stopped_children),
364                stopped_retention_cap,
365                is_poisoned,
366                failed_actor_count,
367                debug: ProcDebugStatsDto {
368                    memory: ProcessMemoryStatsDto {
369                        process_rss_bytes: debug.memory.process_rss_bytes,
370                        process_vm_size_bytes: debug.memory.process_vm_size_bytes,
371                    },
372                    actor_work_queue_depth_total: debug.actor_work_queue_depth_total,
373                    actor_work_queue_depth_max: debug.actor_work_queue_depth_max,
374                    actor_work_queue_depth_high_water_mark: debug
375                        .actor_work_queue_depth_high_water_mark,
376                    last_nonzero_queue_depth_age_ms: debug.last_nonzero_queue_depth_age_ms,
377                },
378            },
379            NodeProperties::Actor {
380                actor_status,
381                actor_type,
382                instance_id,
383                messages_processed,
384                created_at,
385                last_message_handler,
386                total_processing_time_us,
387                queue_depth,
388                flight_recorder,
389                is_system,
390                inbound_ordering,
391                failure_info,
392                execution,
393            } => Self::Actor {
394                actor_status,
395                actor_type,
396                instance_id,
397                messages_processed,
398                created_at: created_at.as_ref().map(format_time),
399                last_message_handler,
400                total_processing_time_us,
401                queue_depth,
402                flight_recorder,
403                is_system,
404                inbound_ordering: inbound_ordering
405                    .map(|io| Box::new(InboundOrderingDto::from(*io))),
406                failure_info: failure_info.map(Into::into),
407                execution: execution.map(|e| Box::new(ExecutionDto::from(*e))),
408            },
409            NodeProperties::Error { code, message } => Self::Error { code, message },
410        }
411    }
412}
413
414impl From<FailureInfo> for FailureInfoDto {
415    fn from(f: FailureInfo) -> Self {
416        Self {
417            error_message: f.error_message,
418            root_cause_actor: f.root_cause_actor.to_string(),
419            root_cause_name: f.root_cause_name,
420            occurred_at: format_time(&f.occurred_at),
421            is_propagated: f.is_propagated,
422        }
423    }
424}
425
426impl From<hyperactor::ordering::OrderingSessionSnapshot> for OrderingSessionSnapshotDto {
427    fn from(s: hyperactor::ordering::OrderingSessionSnapshot) -> Self {
428        Self {
429            session_id: s.session_id.to_string(),
430            sender: s.sender.as_ref().map(|a| a.to_string()),
431            last_released_seq: s.last_released_seq,
432            expected_next_seq: s.expected_next_seq,
433            buffered_count: s.buffered_count,
434            oldest_buffered_seq: s.oldest_buffered_seq,
435            newest_buffered_seq: s.newest_buffered_seq,
436        }
437    }
438}
439
440impl From<InboundOrdering> for InboundOrderingDto {
441    fn from(o: InboundOrdering) -> Self {
442        Self {
443            enabled: o.enabled,
444            snapshot_complete: o.snapshot_complete,
445            skipped_session_count: o.skipped_session_count,
446            known_session_count: o.known_session_count,
447            returned_buffered_session_count: o.returned_buffered_session_count,
448            returned_buffered_message_count: o.returned_buffered_message_count,
449            returned_max_buffered_count: o.returned_max_buffered_count,
450            sessions: o.sessions.into_iter().map(Into::into).collect(),
451        }
452    }
453}
454
455impl From<Execution> for ExecutionDto {
456    fn from(e: Execution) -> Self {
457        Self {
458            active_count: e.active_count,
459            active_handlers: e.active_handlers.into_iter().map(Into::into).collect(),
460            complete: e.complete,
461            truncated: e.truncated,
462        }
463    }
464}
465
466impl From<ActiveHandler> for ActiveHandlerDto {
467    fn from(h: ActiveHandler) -> Self {
468        Self {
469            name: h.name,
470            active_count: h.active_count,
471            oldest_since: format_time(&h.oldest_since),
472        }
473    }
474}
475
476// DTO → Domain conversions (fallible)
477
478impl TryFrom<NodePayloadDto> for NodePayload {
479    type Error = anyhow::Error;
480
481    fn try_from(dto: NodePayloadDto) -> Result<Self, Self::Error> {
482        Ok(Self {
483            identity: dto
484                .identity
485                .parse()
486                .with_context(|| format!("failed to parse identity: {:?}", dto.identity))?,
487            properties: dto
488                .properties
489                .try_into()
490                .context("failed to parse properties")?,
491            children: parse_refs("children", &dto.children)?,
492            parent: dto
493                .parent
494                .map(|s| {
495                    s.parse()
496                        .with_context(|| format!("failed to parse parent: {s:?}"))
497                })
498                .transpose()?,
499            as_of: humantime::parse_rfc3339(&dto.as_of)
500                .with_context(|| format!("failed to parse as_of: {:?}", dto.as_of))?,
501        })
502    }
503}
504
505impl TryFrom<NodePropertiesDto> for NodeProperties {
506    type Error = anyhow::Error;
507
508    fn try_from(
509        dto: NodePropertiesDto,
510    ) -> Result<Self, <Self as TryFrom<NodePropertiesDto>>::Error> {
511        Ok(match dto {
512            NodePropertiesDto::Root {
513                num_hosts,
514                started_at,
515                started_by,
516                system_children,
517            } => Self::Root {
518                num_hosts,
519                started_at: humantime::parse_rfc3339(&started_at)
520                    .context("failed to parse Root.started_at")?,
521                started_by,
522                system_children: parse_refs("Root.system_children", &system_children)?,
523            },
524            NodePropertiesDto::Host {
525                addr,
526                num_procs,
527                system_children,
528                memory,
529            } => Self::Host {
530                addr,
531                num_procs,
532                system_children: parse_refs("Host.system_children", &system_children)?,
533                memory: super::ProcessMemoryStats {
534                    process_rss_bytes: memory.process_rss_bytes,
535                    process_vm_size_bytes: memory.process_vm_size_bytes,
536                },
537            },
538            NodePropertiesDto::Proc {
539                proc_name,
540                num_actors,
541                system_children,
542                stopped_children,
543                stopped_retention_cap,
544                is_poisoned,
545                failed_actor_count,
546                debug,
547            } => Self::Proc {
548                proc_name,
549                num_actors,
550                system_children: parse_refs("Proc.system_children", &system_children)?,
551                stopped_children: parse_refs("Proc.stopped_children", &stopped_children)?,
552                stopped_retention_cap,
553                is_poisoned,
554                failed_actor_count,
555                debug: super::ProcDebugStats {
556                    memory: super::ProcessMemoryStats {
557                        process_rss_bytes: debug.memory.process_rss_bytes,
558                        process_vm_size_bytes: debug.memory.process_vm_size_bytes,
559                    },
560                    actor_work_queue_depth_total: debug.actor_work_queue_depth_total,
561                    actor_work_queue_depth_max: debug.actor_work_queue_depth_max,
562                    actor_work_queue_depth_high_water_mark: debug
563                        .actor_work_queue_depth_high_water_mark,
564                    last_nonzero_queue_depth_age_ms: debug.last_nonzero_queue_depth_age_ms,
565                },
566            },
567            NodePropertiesDto::Actor {
568                actor_status,
569                actor_type,
570                instance_id,
571                messages_processed,
572                created_at,
573                last_message_handler,
574                total_processing_time_us,
575                queue_depth,
576                flight_recorder,
577                is_system,
578                inbound_ordering,
579                failure_info,
580                execution,
581            } => Self::Actor {
582                actor_status,
583                actor_type,
584                instance_id,
585                messages_processed,
586                created_at: created_at
587                    .map(|s| {
588                        humantime::parse_rfc3339(&s)
589                            .with_context(|| format!("failed to parse Actor.created_at: {s:?}"))
590                    })
591                    .transpose()?,
592                last_message_handler,
593                total_processing_time_us,
594                queue_depth,
595                flight_recorder,
596                is_system,
597                inbound_ordering: inbound_ordering
598                    .map(|dto| InboundOrdering::try_from(*dto).map(Box::new))
599                    .transpose()
600                    .context("failed to parse Actor.inbound_ordering")?,
601                failure_info: failure_info
602                    .map(TryInto::try_into)
603                    .transpose()
604                    .context("failed to parse Actor.failure_info")?,
605                execution: execution
606                    .map(|dto| Execution::try_from(*dto).map(Box::new))
607                    .transpose()
608                    .context("failed to parse Actor.execution")?,
609            },
610            NodePropertiesDto::Error { code, message } => Self::Error { code, message },
611        })
612    }
613}
614
615impl TryFrom<FailureInfoDto> for FailureInfo {
616    type Error = anyhow::Error;
617
618    fn try_from(dto: FailureInfoDto) -> Result<Self, Self::Error> {
619        Ok(Self {
620            error_message: dto.error_message,
621            root_cause_actor: dto.root_cause_actor.parse().with_context(|| {
622                format!(
623                    "failed to parse FailureInfo.root_cause_actor: {:?}",
624                    dto.root_cause_actor
625                )
626            })?,
627            root_cause_name: dto.root_cause_name,
628            occurred_at: humantime::parse_rfc3339(&dto.occurred_at).with_context(|| {
629                format!(
630                    "failed to parse FailureInfo.occurred_at: {:?}",
631                    dto.occurred_at
632                )
633            })?,
634            is_propagated: dto.is_propagated,
635        })
636    }
637}
638
639impl TryFrom<ExecutionDto> for Execution {
640    type Error = anyhow::Error;
641
642    fn try_from(dto: ExecutionDto) -> Result<Self, Self::Error> {
643        Ok(Self {
644            active_count: dto.active_count,
645            active_handlers: dto
646                .active_handlers
647                .into_iter()
648                .map(ActiveHandler::try_from)
649                .collect::<Result<Vec<_>, _>>()?,
650            complete: dto.complete,
651            truncated: dto.truncated,
652        })
653    }
654}
655
656impl TryFrom<ActiveHandlerDto> for ActiveHandler {
657    type Error = anyhow::Error;
658
659    fn try_from(dto: ActiveHandlerDto) -> Result<Self, Self::Error> {
660        Ok(Self {
661            name: dto.name,
662            active_count: dto.active_count,
663            oldest_since: humantime::parse_rfc3339(&dto.oldest_since).with_context(|| {
664                format!(
665                    "failed to parse ActiveHandler.oldest_since: {:?}",
666                    dto.oldest_since
667                )
668            })?,
669        })
670    }
671}
672
673impl TryFrom<OrderingSessionSnapshotDto> for hyperactor::ordering::OrderingSessionSnapshot {
674    type Error = anyhow::Error;
675
676    fn try_from(dto: OrderingSessionSnapshotDto) -> Result<Self, Self::Error> {
677        let session_id = dto.session_id.parse().with_context(|| {
678            format!(
679                "failed to parse OrderingSessionSnapshot.session_id: {:?}",
680                dto.session_id
681            )
682        })?;
683        let sender = dto
684            .sender
685            .as_ref()
686            .map(|s| {
687                s.parse().with_context(|| {
688                    format!("failed to parse OrderingSessionSnapshot.sender: {s:?}")
689                })
690            })
691            .transpose()?;
692        Ok(Self {
693            session_id,
694            sender,
695            last_released_seq: dto.last_released_seq,
696            expected_next_seq: dto.expected_next_seq,
697            buffered_count: dto.buffered_count,
698            oldest_buffered_seq: dto.oldest_buffered_seq,
699            newest_buffered_seq: dto.newest_buffered_seq,
700        })
701    }
702}
703
704impl TryFrom<InboundOrderingDto> for InboundOrdering {
705    type Error = anyhow::Error;
706
707    fn try_from(dto: InboundOrderingDto) -> Result<Self, Self::Error> {
708        let sessions: Vec<hyperactor::ordering::OrderingSessionSnapshot> = dto
709            .sessions
710            .into_iter()
711            .enumerate()
712            .map(|(i, s)| {
713                s.try_into()
714                    .with_context(|| format!("failed to parse InboundOrdering.sessions[{i}]"))
715            })
716            .collect::<Result<_, _>>()?;
717        Ok(Self {
718            enabled: dto.enabled,
719            snapshot_complete: dto.snapshot_complete,
720            skipped_session_count: dto.skipped_session_count,
721            known_session_count: dto.known_session_count,
722            returned_buffered_session_count: dto.returned_buffered_session_count,
723            returned_buffered_message_count: dto.returned_buffered_message_count,
724            returned_max_buffered_count: dto.returned_max_buffered_count,
725            sessions,
726        })
727    }
728}
729
730#[cfg(test)]
731mod tests {
732    use super::*;
733    use crate::mesh_id::ResourceId;
734
735    // Test fixtures
736
737    fn test_proc_id() -> hyperactor::ProcAddr {
738        ResourceId::proc_addr_from_name(hyperactor::channel::ChannelAddr::Local(0), "worker")
739    }
740
741    fn test_actor_id() -> hyperactor::ActorAddr {
742        test_proc_id().actor_addr("actor")
743    }
744
745    fn test_host_actor_id() -> hyperactor::ActorAddr {
746        test_proc_id().actor_addr("host_agent")
747    }
748
749    fn test_time() -> SystemTime {
750        humantime::parse_rfc3339("2025-01-15T10:30:00.123Z").unwrap()
751    }
752
753    fn test_time_2() -> SystemTime {
754        humantime::parse_rfc3339("2025-01-15T11:00:00.456Z").unwrap()
755    }
756
757    fn make_root_payload() -> NodePayload {
758        NodePayload {
759            identity: NodeRef::Root,
760            properties: NodeProperties::Root {
761                num_hosts: 2,
762                started_at: test_time(),
763                started_by: "test_user".to_string(),
764                system_children: vec![],
765            },
766            children: vec![NodeRef::Host(test_host_actor_id())],
767            parent: None,
768            as_of: test_time(),
769        }
770    }
771
772    fn make_host_payload() -> NodePayload {
773        NodePayload {
774            identity: NodeRef::Host(test_host_actor_id()),
775            properties: NodeProperties::Host {
776                addr: "127.0.0.1:8080".to_string(),
777                num_procs: 1,
778                system_children: vec![],
779                memory: Default::default(),
780            },
781            children: vec![NodeRef::Proc(test_proc_id())],
782            parent: Some(NodeRef::Root),
783            as_of: test_time(),
784        }
785    }
786
787    fn make_proc_payload() -> NodePayload {
788        NodePayload {
789            identity: NodeRef::Proc(test_proc_id()),
790            properties: NodeProperties::Proc {
791                proc_name: "worker".to_string(),
792                num_actors: 3,
793                system_children: vec![NodeRef::Actor(test_actor_id())],
794                stopped_children: vec![],
795                stopped_retention_cap: 100,
796                is_poisoned: false,
797                failed_actor_count: 0,
798                debug: Default::default(),
799            },
800            children: vec![NodeRef::Actor(test_actor_id())],
801            parent: Some(NodeRef::Host(test_host_actor_id())),
802            as_of: test_time(),
803        }
804    }
805
806    fn test_instance_id() -> String {
807        // Stable test UUID; round-trip tests don't care about the value.
808        "01900000-0000-7000-8000-000000000001".to_string()
809    }
810
811    fn make_actor_payload_no_failure() -> NodePayload {
812        NodePayload {
813            identity: NodeRef::Actor(test_actor_id()),
814            properties: NodeProperties::Actor {
815                actor_status: "running".to_string(),
816                actor_type: "MyActor".to_string(),
817                instance_id: test_instance_id(),
818                messages_processed: 42,
819                created_at: Some(test_time()),
820                last_message_handler: Some("handle_msg".to_string()),
821                total_processing_time_us: 1500,
822                queue_depth: 0,
823                flight_recorder: None,
824                is_system: false,
825                inbound_ordering: None,
826                failure_info: None,
827                execution: None,
828            },
829            children: vec![],
830            parent: Some(NodeRef::Proc(test_proc_id())),
831            as_of: test_time(),
832        }
833    }
834
835    fn make_actor_payload_with_failure() -> NodePayload {
836        NodePayload {
837            identity: NodeRef::Actor(test_actor_id()),
838            properties: NodeProperties::Actor {
839                actor_status: "failed".to_string(),
840                actor_type: "MyActor".to_string(),
841                instance_id: test_instance_id(),
842                messages_processed: 10,
843                created_at: Some(test_time()),
844                last_message_handler: None,
845                total_processing_time_us: 500,
846                queue_depth: 0,
847                flight_recorder: Some("trace-abc".to_string()),
848                is_system: true,
849                inbound_ordering: None,
850                failure_info: Some(FailureInfo {
851                    error_message: "boom".to_string(),
852                    root_cause_actor: test_actor_id(),
853                    root_cause_name: Some("root_actor".to_string()),
854                    occurred_at: test_time_2(),
855                    is_propagated: true,
856                }),
857                execution: None,
858            },
859            children: vec![],
860            parent: Some(NodeRef::Proc(test_proc_id())),
861            as_of: test_time(),
862        }
863    }
864
865    fn make_actor_payload_minimal() -> NodePayload {
866        NodePayload {
867            identity: NodeRef::Actor(test_actor_id()),
868            properties: NodeProperties::Actor {
869                actor_status: "idle".to_string(),
870                actor_type: "MinimalActor".to_string(),
871                instance_id: test_instance_id(),
872                messages_processed: 0,
873                created_at: None,
874                last_message_handler: None,
875                total_processing_time_us: 0,
876                queue_depth: 0,
877                flight_recorder: None,
878                is_system: false,
879                inbound_ordering: None,
880                failure_info: None,
881                execution: None,
882            },
883            children: vec![],
884            parent: Some(NodeRef::Proc(test_proc_id())),
885            as_of: test_time(),
886        }
887    }
888
889    fn make_ordering_session(
890        session_id: uuid::Uuid,
891        last_released_seq: u64,
892        buffered_count: usize,
893    ) -> hyperactor::ordering::OrderingSessionSnapshot {
894        let (oldest, newest) = if buffered_count > 0 {
895            (
896                Some(last_released_seq + 2),
897                Some(last_released_seq + 1 + buffered_count as u64),
898            )
899        } else {
900            (None, None)
901        };
902        hyperactor::ordering::OrderingSessionSnapshot {
903            session_id,
904            sender: Some(test_actor_id()),
905            last_released_seq,
906            expected_next_seq: last_released_seq.saturating_add(1),
907            buffered_count,
908            oldest_buffered_seq: oldest,
909            newest_buffered_seq: newest,
910        }
911    }
912
913    fn make_actor_payload_inbound_ordering_complete() -> NodePayload {
914        NodePayload {
915            identity: NodeRef::Actor(test_actor_id()),
916            properties: NodeProperties::Actor {
917                actor_status: "running".to_string(),
918                actor_type: "MyActor".to_string(),
919                instance_id: test_instance_id(),
920                messages_processed: 17,
921                created_at: Some(test_time()),
922                last_message_handler: Some("handle_msg".to_string()),
923                total_processing_time_us: 900,
924                queue_depth: 5,
925                flight_recorder: None,
926                is_system: false,
927                inbound_ordering: Some(Box::new(InboundOrdering {
928                    enabled: true,
929                    snapshot_complete: true,
930                    skipped_session_count: 0,
931                    known_session_count: 2,
932                    returned_buffered_session_count: 1,
933                    returned_buffered_message_count: 3,
934                    returned_max_buffered_count: 3,
935                    sessions: vec![
936                        make_ordering_session(uuid::Uuid::from_u128(1), 7, 0),
937                        make_ordering_session(uuid::Uuid::from_u128(2), 1, 3),
938                    ],
939                })),
940                failure_info: None,
941                execution: None,
942            },
943            children: vec![],
944            parent: Some(NodeRef::Proc(test_proc_id())),
945            as_of: test_time(),
946        }
947    }
948
949    fn make_actor_payload_inbound_ordering_partial() -> NodePayload {
950        NodePayload {
951            identity: NodeRef::Actor(test_actor_id()),
952            properties: NodeProperties::Actor {
953                actor_status: "running".to_string(),
954                actor_type: "MyActor".to_string(),
955                instance_id: test_instance_id(),
956                messages_processed: 17,
957                created_at: Some(test_time()),
958                last_message_handler: Some("handle_msg".to_string()),
959                total_processing_time_us: 900,
960                queue_depth: 5,
961                flight_recorder: None,
962                is_system: false,
963                inbound_ordering: Some(Box::new(InboundOrdering {
964                    enabled: true,
965                    snapshot_complete: false,
966                    skipped_session_count: 2,
967                    // IO-5: 1 returned + 2 skipped = 3.
968                    known_session_count: 3,
969                    // IO-6: rollups over the one returned session only.
970                    returned_buffered_session_count: 1,
971                    returned_buffered_message_count: 4,
972                    returned_max_buffered_count: 4,
973                    sessions: vec![make_ordering_session(uuid::Uuid::from_u128(7), 0, 4)],
974                })),
975                failure_info: None,
976                execution: None,
977            },
978            children: vec![],
979            parent: Some(NodeRef::Proc(test_proc_id())),
980            as_of: test_time(),
981        }
982    }
983
984    fn make_error_payload() -> NodePayload {
985        NodePayload {
986            identity: NodeRef::Actor(test_actor_id()),
987            properties: NodeProperties::Error {
988                code: "not_found".to_string(),
989                message: "actor not found".to_string(),
990            },
991            children: vec![],
992            parent: None,
993            as_of: test_time(),
994        }
995    }
996
997    // HB-2 (round-trip): NodePayload → NodePayloadDto → NodePayload is
998    // lossless for values representable in the wire format.
999
1000    fn assert_round_trip(payload: &NodePayload) {
1001        let dto: NodePayloadDto = payload.clone().into();
1002        let back: NodePayload = dto.try_into().expect("round-trip conversion");
1003        assert_eq!(payload, &back);
1004    }
1005
1006    /// HB-2: Root variant round-trips.
1007    #[test]
1008    fn test_round_trip_root() {
1009        assert_round_trip(&make_root_payload());
1010    }
1011
1012    /// HB-2: Host variant round-trips.
1013    #[test]
1014    fn test_round_trip_host() {
1015        assert_round_trip(&make_host_payload());
1016    }
1017
1018    /// HB-2: Proc variant round-trips.
1019    #[test]
1020    fn test_round_trip_proc() {
1021        assert_round_trip(&make_proc_payload());
1022    }
1023
1024    /// HB-2: Actor variant without failure round-trips.
1025    #[test]
1026    fn test_round_trip_actor_no_failure() {
1027        assert_round_trip(&make_actor_payload_no_failure());
1028    }
1029
1030    /// HB-2: Actor variant with failure round-trips.
1031    #[test]
1032    fn test_round_trip_actor_with_failure() {
1033        assert_round_trip(&make_actor_payload_with_failure());
1034    }
1035
1036    /// HB-2: Actor variant with all optional fields absent round-trips.
1037    #[test]
1038    fn test_round_trip_actor_minimal() {
1039        assert_round_trip(&make_actor_payload_minimal());
1040    }
1041
1042    /// HB-1 + HB-2 + IO-4 + IO-5: Actor with a complete inbound-ordering
1043    /// snapshot round-trips. Exercises the `Some({enabled: true,
1044    /// snapshot_complete: true, ...})` branch of IO-1 plus the IO-4
1045    /// derivation and IO-5 totality (`known_session_count ==
1046    /// sessions.len()` when no skipped).
1047    #[test]
1048    fn test_round_trip_actor_inbound_ordering_complete() {
1049        let payload = make_actor_payload_inbound_ordering_complete();
1050        // Assert the fixture itself satisfies IO-4 / IO-5 before
1051        // round-tripping, so a bug in the fixture can't pass for the
1052        // wrong reason.
1053        if let NodeProperties::Actor {
1054            inbound_ordering: Some(io),
1055            ..
1056        } = &payload.properties
1057        {
1058            assert_eq!(io.snapshot_complete, io.skipped_session_count == 0); // IO-4
1059            assert_eq!(
1060                io.known_session_count,
1061                io.sessions.len() + io.skipped_session_count
1062            ); // IO-5
1063        } else {
1064            panic!("fixture must be Actor with Some(inbound_ordering)");
1065        }
1066        assert_round_trip(&payload);
1067    }
1068
1069    /// HB-1 + HB-2 + IO-4 + IO-5 + IO-6: Actor with a PARTIAL
1070    /// inbound-ordering snapshot round-trips and the rollups reflect
1071    /// returned sessions only. `skipped_session_count > 0` forces
1072    /// IO-4's `snapshot_complete == false`, IO-5's
1073    /// `known_session_count == returned + skipped`, and IO-6's
1074    /// returned-only rollups (NOT computed over the skipped sessions
1075    /// the snapshot doesn't carry).
1076    #[test]
1077    fn test_round_trip_actor_inbound_ordering_partial() {
1078        let payload = make_actor_payload_inbound_ordering_partial();
1079        if let NodeProperties::Actor {
1080            inbound_ordering: Some(io),
1081            ..
1082        } = &payload.properties
1083        {
1084            // IO-4: false because skipped > 0.
1085            assert!(!io.snapshot_complete);
1086            assert_eq!(io.snapshot_complete, io.skipped_session_count == 0);
1087            // IO-5: total includes skipped.
1088            assert_eq!(
1089                io.known_session_count,
1090                io.sessions.len() + io.skipped_session_count
1091            );
1092            // IO-6: rollups over RETURNED sessions only -- NOT over
1093            // returned + skipped.
1094            assert_eq!(
1095                io.returned_buffered_session_count,
1096                io.sessions.iter().filter(|s| s.buffered_count > 0).count()
1097            );
1098            assert_eq!(
1099                io.returned_buffered_message_count,
1100                io.sessions.iter().map(|s| s.buffered_count).sum::<usize>()
1101            );
1102            assert_eq!(
1103                io.returned_max_buffered_count,
1104                io.sessions
1105                    .iter()
1106                    .map(|s| s.buffered_count)
1107                    .max()
1108                    .unwrap_or(0)
1109            );
1110        } else {
1111            panic!("fixture must be Actor with Some(inbound_ordering)");
1112        }
1113        assert_round_trip(&payload);
1114    }
1115
1116    /// HB-2: Error variant round-trips.
1117    #[test]
1118    fn test_round_trip_error() {
1119        assert_round_trip(&make_error_payload());
1120    }
1121
1122    // HB-1 (typed-internal, string-external): typed Rust values serialize
1123    // as canonical strings in the DTO JSON output.
1124
1125    /// HB-1: Root identity, children, parent, and timestamps serialize
1126    /// as strings; externally-tagged enum key is "Root".
1127    #[test]
1128    fn test_json_shape_root() {
1129        let dto: NodePayloadDto = make_root_payload().into();
1130        let json = serde_json::to_value(&dto).unwrap();
1131
1132        assert_eq!(json["identity"], "root");
1133        assert!(json["parent"].is_null());
1134        assert_eq!(json["as_of"], "2025-01-15T10:30:00.123Z");
1135
1136        let children = json["children"].as_array().unwrap();
1137        assert_eq!(children.len(), 1);
1138        assert_eq!(children[0], format!("host:{}", test_host_actor_id()));
1139
1140        let root = &json["properties"]["Root"];
1141        assert_eq!(root["num_hosts"], 2);
1142        assert_eq!(root["started_at"], "2025-01-15T10:30:00.123Z");
1143        assert_eq!(root["started_by"], "test_user");
1144        assert!(root["system_children"].as_array().unwrap().is_empty());
1145    }
1146
1147    /// HB-1: Actor variant with failure — ActorAddr, SystemTime, and
1148    /// nested FailureInfo fields all serialize as strings.
1149    #[test]
1150    fn test_json_shape_actor_with_failure() {
1151        let dto: NodePayloadDto = make_actor_payload_with_failure().into();
1152        let json = serde_json::to_value(&dto).unwrap();
1153
1154        assert_eq!(json["identity"], test_actor_id().to_string());
1155        assert_eq!(json["parent"], test_proc_id().to_string());
1156
1157        let actor = &json["properties"]["Actor"];
1158        assert_eq!(actor["actor_status"], "failed");
1159        assert_eq!(actor["messages_processed"], 10);
1160        assert_eq!(actor["created_at"], "2025-01-15T10:30:00.123Z");
1161        assert!(actor["last_message_handler"].is_null());
1162        assert_eq!(actor["flight_recorder"], "trace-abc");
1163        assert_eq!(actor["is_system"], true);
1164
1165        let fi = &actor["failure_info"];
1166        assert_eq!(fi["error_message"], "boom");
1167        assert_eq!(fi["root_cause_actor"], test_actor_id().to_string());
1168        assert_eq!(fi["root_cause_name"], "root_actor");
1169        assert_eq!(fi["occurred_at"], "2025-01-15T11:00:00.456Z");
1170        assert_eq!(fi["is_propagated"], true);
1171    }
1172
1173    /// HB-1: Option fields serialize as JSON null when absent.
1174    #[test]
1175    fn test_json_shape_optional_none_fields() {
1176        let dto: NodePayloadDto = make_actor_payload_minimal().into();
1177        let json = serde_json::to_value(&dto).unwrap();
1178
1179        let actor = &json["properties"]["Actor"];
1180        assert!(actor["created_at"].is_null());
1181        assert!(actor["last_message_handler"].is_null());
1182        assert!(actor["flight_recorder"].is_null());
1183        assert!(actor["failure_info"].is_null());
1184    }
1185
1186    /// HB-1: Error variant preserves code/message as plain strings.
1187    #[test]
1188    fn test_json_shape_error() {
1189        let dto: NodePayloadDto = make_error_payload().into();
1190        let json = serde_json::to_value(&dto).unwrap();
1191
1192        let err = &json["properties"]["Error"];
1193        assert_eq!(err["code"], "not_found");
1194        assert_eq!(err["message"], "actor not found");
1195    }
1196
1197    /// HB-1: Empty children vec serializes as `[]`.
1198    #[test]
1199    fn test_json_shape_empty_children() {
1200        let dto: NodePayloadDto = make_actor_payload_no_failure().into();
1201        let json = serde_json::to_value(&dto).unwrap();
1202        assert!(json["children"].as_array().unwrap().is_empty());
1203    }
1204
1205    // HB-3 (schema-honesty): published schema reflects the actual wire
1206    // format. The schemars(rename/title) attributes must produce $defs
1207    // keys and title matching the domain type names, not the Dto suffixes.
1208
1209    /// HB-3: $defs keys are "NodeProperties" and "FailureInfo", not
1210    /// "NodePropertiesDto" / "FailureInfoDto".
1211    #[test]
1212    fn test_schema_defs_keys() {
1213        let schema = schemars::schema_for!(NodePayloadDto);
1214        let json = serde_json::to_value(&schema).unwrap();
1215        let defs = json["$defs"].as_object().unwrap();
1216        assert!(
1217            defs.contains_key("NodeProperties"),
1218            "$defs must contain 'NodeProperties', got: {:?}",
1219            defs.keys().collect::<Vec<_>>()
1220        );
1221        assert!(
1222            defs.contains_key("FailureInfo"),
1223            "$defs must contain 'FailureInfo', got: {:?}",
1224            defs.keys().collect::<Vec<_>>()
1225        );
1226    }
1227
1228    /// HB-3: Top-level schema title is "NodePayload", not
1229    /// "NodePayloadDto".
1230    #[test]
1231    fn test_schema_title() {
1232        let schema = schemars::schema_for!(NodePayloadDto);
1233        let json = serde_json::to_value(&schema).unwrap();
1234        assert_eq!(json["title"], "NodePayload");
1235    }
1236}
hyperactor_mesh/introspect/dto.rs

hyperactor_mesh/introspect/
dto.rs
