helix_driver_host/engine/
types.rs

1use std::collections::HashMap;
2use std::sync::atomic::{AtomicU64, Ordering};
3use std::sync::Arc;
4
5use tokio::sync::{mpsc, oneshot};
6
7use crate::metrics::AsyncMetricSink;
8use crate::trace::TraceHooks;
9use helix_core::effect::TransportId;
10use helix_core::PortError;
11
12/// `TransportId → 出站帧发送器` 路由表（泛型 over FrameSender 实现）。
13///
14/// host 装配阶段填入（或经 `transport_rx` 注册通道在连接成功后投递入表）；`Effect::Send`
15/// 分支按 id 查表调 `FrameSender::send`（`&self`）。表内 `Arc<Fs>` 无外层锁，连接任务只在
16/// 成功后把独占句柄注册进表，避免连接期锁跨 `await`。
17pub type TransportTable<Tr> = HashMap<TransportId, Arc<Tr>>;
18
19/// 平台连接任务提交的动态 sender 注册；engine 插入路由表后才回 ack。
20pub struct TransportRegistration<Fs> {
21    pub(super) id: TransportId,
22    pub(super) sender: Arc<Fs>,
23    pub(super) registered_tx: oneshot::Sender<()>,
24}
25
26pub async fn register_transport<Fs>(
27    registration_tx: &mpsc::UnboundedSender<TransportRegistration<Fs>>,
28    id: TransportId,
29    sender: Arc<Fs>,
30) -> Result<(), PortError> {
31    let (registered_tx, registered_rx) = oneshot::channel();
32    registration_tx
33        .send(TransportRegistration {
34            id,
35            sender,
36            registered_tx,
37        })
38        .map_err(|_| PortError::Transport("transport registration channel closed".into()))?;
39    registered_rx
40        .await
41        .map_err(|_| PortError::Transport("transport registration was not acknowledged".into()))
42}
43
44#[derive(Clone, Copy, Debug, PartialEq, Eq)]
45pub enum TransportLifecycleEvent {
46    Disconnected {
47        transport_id: TransportId,
48        reason: &'static str,
49    },
50}
51
52#[derive(Clone, Debug, PartialEq, Eq)]
53pub struct TransportTraceEvent {
54    pub transport_id: TransportId,
55    pub name: &'static str,
56    pub action: &'static str,
57    pub attempt: Option<u32>,
58    pub delay_ms: Option<u64>,
59    pub next_delay_ms: Option<u64>,
60    pub reason: Option<&'static str>,
61    pub error_class: Option<String>,
62}
63
64/// Transport 生命周期遥测的固定有界队列容量。
65///
66/// 这是可丢诊断面，不参与 transport/control 正确性；过载时丢最新事件并累计计数，
67/// 绝不把 exporter 背压传进 engine 热路径。
68pub const TRANSPORT_TRACE_QUEUE_CAPACITY: usize = 256;
69
70#[derive(Clone, Debug, Default)]
71pub struct TransportTraceStats {
72    dropped: Arc<AtomicU64>,
73}
74
75impl TransportTraceStats {
76    pub fn dropped_count(&self) -> u64 {
77        self.dropped.load(Ordering::Relaxed)
78    }
79}
80
81/// 有界、同步非阻塞的 transport trace 出口。
82#[derive(Clone, Debug)]
83pub struct TransportTraceSink {
84    tx: mpsc::Sender<TransportTraceEvent>,
85    stats: TransportTraceStats,
86}
87
88impl TransportTraceSink {
89    pub fn channel() -> (Self, mpsc::Receiver<TransportTraceEvent>) {
90        let (tx, rx) = mpsc::channel(TRANSPORT_TRACE_QUEUE_CAPACITY);
91        let stats = TransportTraceStats::default();
92        (Self { tx, stats }, rx)
93    }
94
95    /// 队满或接收端关闭时立即返回，丢本条并累计，不 await、不阻塞。
96    pub fn try_emit(&self, event: TransportTraceEvent) -> bool {
97        match self.tx.try_send(event) {
98            Ok(()) => true,
99            Err(_) => {
100                self.stats.dropped.fetch_add(1, Ordering::Relaxed);
101                false
102            }
103        }
104    }
105
106    pub fn stats(&self) -> TransportTraceStats {
107        self.stats.clone()
108    }
109}
110
111/// 事件泵依赖聚合（泛型 ports + 配置）。
112///
113/// native / ffi 各自构造此结构（注入自家 EventSink egress），调 `run_engine_loop`。
114pub struct EngineDeps<S, H, U, E, C> {
115    pub storage: Arc<S>,
116    pub http: Arc<H>,
117    pub uploader: Arc<U>,
118    pub event_sink: Arc<E>,
119    pub clock: C,
120    pub trace: TraceHooks,
121    /// Host 边界共享异步指标出口；热路径只执行有界 `try_record`。
122    pub metrics: Arc<dyn AsyncMetricSink>,
123    /// Http effect 并发上限（Http BoundedSpawner 的 worker 数 N）。`0` → `.max(1)` 兜底 1。
124    pub max_http_inflight: usize,
125    pub transport_lifecycle_tx: Option<mpsc::UnboundedSender<TransportLifecycleEvent>>,
126    pub transport_trace_tx: Option<TransportTraceSink>,
127}
helix_driver_host/engine/types.rs

helix_driver_host/engine/
types.rs
