helix_driver_host/
lifecycle.rs

1//! Host 侧 lifecycle correlation 合同。
2//!
3//! 该模块只保存显式、不可变的 tick 上下文和有限的父子关联；它不依赖
4//! task-local active carrier，也不把网络 exporter 的背压带回事件泵。
5
6use std::collections::{HashMap, VecDeque};
7use std::sync::atomic::{AtomicU64, Ordering};
8use std::sync::Arc;
9
10use parking_lot::Mutex;
11use tokio::sync::mpsc;
12
13use crate::owned_effect::OwnedEffect;
14use crate::trace::TraceCarrier;
15use helix_core::effect::{Correlation, TimerId};
16use helix_core::Tick;
17
18/// 每个 Tick 及其局部效果树使用的单调标识。
19pub type TickId = u64;
20
21/// lifecycle 的五个固定阶段；局部 HTTP/WS/Storage 效果均直接挂在 T1 根上。
22#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
23pub enum LifecycleStage {
24    T1,
25    T2,
26    T3,
27    T4,
28    T5,
29}
30
31impl LifecycleStage {
32    /// 返回稳定的协议标签，避免把 Rust 调试格式写入遥测。
33    pub const fn as_str(self) -> &'static str {
34        match self {
35            Self::T1 => "T1",
36            Self::T2 => "T2",
37            Self::T3 => "T3",
38            Self::T4 => "T4",
39            Self::T5 => "T5",
40        }
41    }
42
43    /// 将阶段映射到固定数组槽位，保持 O(1) 读取。
44    const fn index(self) -> usize {
45        match self {
46            Self::T1 => 0,
47            Self::T2 => 1,
48            Self::T3 => 2,
49            Self::T4 => 3,
50            Self::T5 => 4,
51        }
52    }
53}
54
55/// 可选端口能力；能力缺失只能报告 not_applicable，不能伪造网络 span。
56#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
57pub enum LifecycleCapability {
58    Http,
59    Ws,
60    Persist,
61    Effect,
62}
63
64impl LifecycleCapability {
65    /// 返回稳定的协议标签，供属性和合同测试使用。
66    pub const fn as_str(self) -> &'static str {
67        match self {
68            Self::Http => "http",
69            Self::Ws => "ws",
70            Self::Persist => "persist",
71            Self::Effect => "effect",
72        }
73    }
74
75    /// 将能力映射到固定数组槽位，保持 O(1) 读取。
76    const fn index(self) -> usize {
77        match self {
78            Self::Http => 0,
79            Self::Ws => 1,
80            Self::Persist => 2,
81            Self::Effect => 3,
82        }
83    }
84}
85
86/// lifecycle 观测状态；Skipped 与 NotApplicable 明确区分未执行和不适用。
87#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
88pub enum LifecycleStatus {
89    Pending,
90    Started,
91    Ok,
92    Error,
93    Skipped,
94    NotApplicable,
95}
96
97impl LifecycleStatus {
98    /// 返回稳定的小写状态标签，便于跨端对账。
99    pub const fn as_str(self) -> &'static str {
100        match self {
101            Self::Pending => "pending",
102            Self::Started => "started",
103            Self::Ok => "ok",
104            Self::Error => "error",
105            Self::Skipped => "skipped",
106            Self::NotApplicable => "not_applicable",
107        }
108    }
109}
110
111/// `LifecycleState` 是状态合同的兼容别名，便于调用方按 state 语义读取。
112pub type LifecycleState = LifecycleStatus;
113
114/// 一次 Tick 的不可变 correlation 上下文。
115///
116/// `with_*` 方法均返回新值；调用方可以在 T1/T2/T3/T4/T5 间传递上下文，
117/// 不需要全局 carrier 或 task-local 状态。`current_stage` 的父阶段恒为 T1，
118/// 使同一 Tick 内的本地子树不会意外嵌套到前一个效果。
119#[derive(Clone, Debug, PartialEq, Eq)]
120pub struct LifecycleContext {
121    tick_id: TickId,
122    parent_tick_id: Option<TickId>,
123    carrier: Option<TraceCarrier>,
124    span_parent: Option<TraceCarrier>,
125    current_stage: LifecycleStage,
126    stage_status: [LifecycleStatus; 5],
127    capabilities: [bool; 4],
128    capability_status: [LifecycleStatus; 4],
129}
130
131impl LifecycleContext {
132    /// 创建 T1 根上下文；网络能力默认关闭，避免缺能力时伪造 network span。
133    pub fn new(
134        tick_id: TickId,
135        parent_tick_id: Option<TickId>,
136        carrier: Option<TraceCarrier>,
137    ) -> Self {
138        Self {
139            tick_id,
140            parent_tick_id,
141            carrier,
142            span_parent: None,
143            current_stage: LifecycleStage::T1,
144            stage_status: [
145                LifecycleStatus::Started,
146                LifecycleStatus::Skipped,
147                LifecycleStatus::Skipped,
148                LifecycleStatus::Skipped,
149                LifecycleStatus::Skipped,
150            ],
151            capabilities: [false; 4],
152            capability_status: [LifecycleStatus::NotApplicable; 4],
153        }
154    }
155
156    /// 创建带父 Tick 的 T1 上下文，作为 engine 的语义化构造别名。
157    pub fn root(
158        tick_id: TickId,
159        parent_tick_id: Option<TickId>,
160        carrier: Option<TraceCarrier>,
161    ) -> Self {
162        Self::new(tick_id, parent_tick_id, carrier)
163    }
164
165    /// 返回当前 Tick 的独立 ID。
166    pub const fn tick_id(&self) -> TickId {
167        self.tick_id
168    }
169
170    /// 返回上游 Tick ID；没有可证明的上游时保持 None。
171    pub const fn parent_tick_id(&self) -> Option<TickId> {
172        self.parent_tick_id
173    }
174
175    /// 返回显式 carrier 的只读引用。
176    pub fn carrier(&self) -> Option<&TraceCarrier> {
177        self.carrier.as_ref()
178    }
179
180    /// 返回当前 Tick 内部 T1 span 的显式 parent carrier；没有 OTel 时保持 None。
181    pub fn span_parent(&self) -> Option<&TraceCarrier> {
182        self.span_parent.as_ref()
183    }
184
185    /// 返回新的上下文并绑定本 Tick 的内部 span parent，不写入全局或 task-local 状态。
186    pub fn with_span_parent(&self, span_parent: Option<TraceCarrier>) -> Self {
187        let mut next = self.clone();
188        next.span_parent = span_parent;
189        next
190    }
191
192    /// 选择异步子节点的显式 parent；内部 T1 优先，外部 W3C carrier 仅作根回退。
193    pub fn otel_parent(&self) -> Option<&TraceCarrier> {
194        self.span_parent.as_ref().or(self.carrier.as_ref())
195    }
196
197    /// 返回当前生命周期阶段。
198    pub const fn current_stage(&self) -> LifecycleStage {
199        self.current_stage
200    }
201
202    /// 返回当前阶段在 T1 根下的挂载点。
203    pub const fn parent_stage(&self) -> LifecycleStage {
204        LifecycleStage::T1
205    }
206
207    /// 读取五阶段状态，不会修改上下文。
208    pub const fn stage_status(&self, stage: LifecycleStage) -> LifecycleStatus {
209        self.stage_status[stage.index()]
210    }
211
212    /// 设置阶段状态并返回新上下文。
213    pub fn with_stage_status(&self, stage: LifecycleStage, status: LifecycleStatus) -> Self {
214        let mut next = self.clone();
215        next.stage_status[stage.index()] = status;
216        next.current_stage = stage;
217        next
218    }
219
220    /// 读取能力是否存在。
221    pub const fn has_capability(&self, capability: LifecycleCapability) -> bool {
222        self.capabilities[capability.index()]
223    }
224
225    /// 读取能力状态；未提供能力时固定返回 NotApplicable。
226    pub const fn capability_status(&self, capability: LifecycleCapability) -> LifecycleStatus {
227        self.capability_status[capability.index()]
228    }
229
230    /// 设置能力存在性；关闭能力时状态自动收敛到 NotApplicable。
231    pub fn with_capability(&self, capability: LifecycleCapability, enabled: bool) -> Self {
232        let mut next = self.clone();
233        let index = capability.index();
234        next.capabilities[index] = enabled;
235        next.capability_status[index] = if enabled {
236            LifecycleStatus::Skipped
237        } else {
238            LifecycleStatus::NotApplicable
239        };
240        next
241    }
242
243    /// 设置能力状态并返回新上下文；NotApplicable 会同时关闭该能力。
244    pub fn with_capability_status(
245        &self,
246        capability: LifecycleCapability,
247        status: LifecycleStatus,
248    ) -> Self {
249        let mut next = self.clone();
250        let index = capability.index();
251        next.capabilities[index] = status != LifecycleStatus::NotApplicable;
252        next.capability_status[index] = status;
253        next
254    }
255
256    /// 构造同一 Tick 的局部阶段上下文，所有局部节点直接挂到 T1。
257    pub fn local_stage(&self, stage: LifecycleStage) -> Self {
258        let mut next = self.clone();
259        next.current_stage = stage;
260        next
261    }
262}
263
264impl Default for LifecycleContext {
265    /// 创建 synthetic root，供无 Tick 的测试/启动 effect 使用。
266    fn default() -> Self {
267        Self::new(0, None, None)
268    }
269}
270
271/// 生命周期状态出口的结构化记录；不携带请求正文或用户数据。
272#[derive(Clone, Debug, PartialEq, Eq)]
273pub struct LifecycleObservation {
274    pub tick_id: TickId,
275    pub parent_tick_id: Option<TickId>,
276    pub stage: LifecycleStage,
277    pub capability: Option<LifecycleCapability>,
278    pub status: LifecycleStatus,
279    /// 固定诊断原因；不承载请求正文或动态错误文本。
280    pub reason: Option<&'static str>,
281}
282
283/// lifecycle 诊断出口的固定有界容量。
284pub const LIFECYCLE_TRACE_QUEUE_CAPACITY: usize = 256;
285
286#[derive(Clone, Debug, Default)]
287pub struct LifecycleTraceStats {
288    dropped: Arc<AtomicU64>,
289}
290
291impl LifecycleTraceStats {
292    /// 返回因队列满或接收端关闭而丢弃的观测数。
293    pub fn dropped_count(&self) -> u64 {
294        self.dropped.load(Ordering::Relaxed)
295    }
296}
297
298/// 不参与业务正确性的有界 lifecycle 观测 sink。
299#[derive(Clone, Debug)]
300pub struct LifecycleTraceSink {
301    tx: mpsc::Sender<LifecycleObservation>,
302    stats: LifecycleTraceStats,
303}
304
305impl LifecycleTraceSink {
306    /// 创建固定容量的生命周期观测通道。
307    pub fn channel() -> (Self, mpsc::Receiver<LifecycleObservation>) {
308        let (tx, rx) = mpsc::channel(LIFECYCLE_TRACE_QUEUE_CAPACITY);
309        (
310            Self {
311                tx,
312                stats: LifecycleTraceStats::default(),
313            },
314            rx,
315        )
316    }
317
318    /// 非阻塞投递生命周期观测；队满时丢弃并累计，不影响 engine。
319    pub fn try_emit(&self, observation: LifecycleObservation) -> bool {
320        match self.tx.try_send(observation) {
321            Ok(()) => true,
322            Err(_) => {
323                self.stats.dropped.fetch_add(1, Ordering::Relaxed);
324                false
325            }
326        }
327    }
328
329    /// 返回共享的丢弃统计。
330    pub fn stats(&self) -> LifecycleTraceStats {
331        self.stats.clone()
332    }
333}
334
335/// 父 Tick 关联表容量上限；超限仅淘汰诊断链接，不影响业务回灌。
336pub const LIFECYCLE_LINK_CAPACITY: usize = 4096;
337
338#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
339enum LinkKey {
340    Correlation(u64),
341    Timer(u64),
342}
343
344#[derive(Clone, Debug)]
345struct LifecycleLink {
346    generation: u64,
347    parent_tick_id: TickId,
348    carrier: Option<TraceCarrier>,
349    span_parent: Option<TraceCarrier>,
350}
351
352#[derive(Default)]
353struct TrackerState {
354    links: HashMap<LinkKey, LifecycleLink>,
355    order: VecDeque<(LinkKey, u64)>,
356    next_generation: u64,
357}
358
359/// engine 私有、有限容量的父 Tick 关联表。
360///
361/// 关联表只保存 correlation/timer 的标量和 bounded carrier，过载时淘汰最旧项；
362/// 它不是 active carrier，也不承担业务正确性。
363#[derive(Clone, Default)]
364pub struct LifecycleTracker {
365    next_tick_id: Arc<AtomicU64>,
366    state: Arc<Mutex<TrackerState>>,
367}
368
369impl LifecycleTracker {
370    /// 分配下一个独立 Tick ID，0 保留给 engine start 的 synthetic root。
371    pub fn next_tick_id(&self) -> TickId {
372        self.next_tick_id
373            .fetch_add(1, Ordering::Relaxed)
374            .saturating_add(1)
375    }
376
377    /// 解析 Tick 的上游父关联，并在终态 Tick 到达后释放一次性链接。
378    pub fn parent_for_tick(
379        &self,
380        tick: &Tick,
381    ) -> (Option<TickId>, Option<TraceCarrier>, Option<TraceCarrier>) {
382        let (key, terminal) = match tick {
383            Tick::PortReply { corr, .. } => (Some(LinkKey::Correlation(corr.raw())), true),
384            Tick::PortProgress { corr, .. } => (Some(LinkKey::Correlation(corr.raw())), false),
385            Tick::Timer(id) => (Some(LinkKey::Timer(id.raw())), true),
386            _ => (None, false),
387        };
388        let Some(key) = key else {
389            return (None, None, None);
390        };
391        let mut state = self.state.lock();
392        let link = if terminal {
393            state.links.remove(&key)
394        } else {
395            state.links.get(&key).cloned()
396        };
397        link.map_or((None, None, None), |link| {
398            (Some(link.parent_tick_id), link.carrier, link.span_parent)
399        })
400    }
401
402    /// 记录效果产生的未来 Tick 关联；超出容量时淘汰最旧诊断链接。
403    pub fn remember_effect(&self, effect: &OwnedEffect, context: &LifecycleContext) {
404        let (key, carrier, span_parent) = match effect {
405            OwnedEffect::Persist { corr, .. }
406            | OwnedEffect::PersistAtomic { corr, .. }
407            | OwnedEffect::Http { corr, .. }
408            | OwnedEffect::UploadFile { corr, .. }
409            | OwnedEffect::Request { corr, .. } => (
410                Some(LinkKey::Correlation(corr.raw())),
411                context.carrier.clone(),
412                context.span_parent.clone(),
413            ),
414            OwnedEffect::ScheduleTimer { id, .. } => (
415                Some(LinkKey::Timer(id.raw())),
416                context.carrier.clone(),
417                context.span_parent.clone(),
418            ),
419            OwnedEffect::CancelTimer { id } => {
420                self.forget(LinkKey::Timer(id.raw()));
421                (None, None, None)
422            }
423            _ => (None, None, None),
424        };
425        let Some(key) = key else {
426            return;
427        };
428        let mut state = self.state.lock();
429        let generation = state
430            .links
431            .get(&key)
432            .map(|link| link.generation)
433            .unwrap_or_else(|| {
434                state.next_generation = state.next_generation.wrapping_add(1);
435                let generation = state.next_generation;
436                state.order.push_back((key, generation));
437                generation
438            });
439        state.links.insert(
440            key,
441            LifecycleLink {
442                generation,
443                parent_tick_id: context.tick_id,
444                carrier,
445                span_parent,
446            },
447        );
448        compact_tracker_order_if_needed(&mut state);
449        while state.links.len() > LIFECYCLE_LINK_CAPACITY {
450            let Some((oldest, generation)) = state.order.pop_front() else {
451                break;
452            };
453            if state
454                .links
455                .get(&oldest)
456                .is_some_and(|link| link.generation == generation)
457            {
458                state.links.remove(&oldest);
459            }
460        }
461    }
462
463    /// 删除被取消的 timer 或替换前的旧关联。
464    fn forget(&self, key: LinkKey) {
465        let mut state = self.state.lock();
466        state.links.remove(&key);
467    }
468}
469
470/// 偶发压缩已消费的顺序标记，保证 stale tombstone 不会无界增长；正常热路径不扫描全表。
471fn compact_tracker_order_if_needed(state: &mut TrackerState) {
472    const COMPACTION_FACTOR: usize = 4;
473    if state.order.len() <= LIFECYCLE_LINK_CAPACITY * COMPACTION_FACTOR {
474        return;
475    }
476    state.order.retain(|(key, generation)| {
477        state
478            .links
479            .get(key)
480            .is_some_and(|link| link.generation == *generation)
481    });
482}
483
484/// 保留端口类型在该模块的可见性，方便调用方构造矩阵测试。
485pub type LifecycleCorrelation = Correlation;
486/// 保留 timer 类型在该模块的可见性，方便调用方构造矩阵测试。
487pub type LifecycleTimer = TimerId;
488
489#[cfg(test)]
490mod tests {
491    use super::*;
492    use bytes::Bytes;
493    use helix_core::effect::HttpRequest;
494
495    #[test]
496    fn context_tree_has_independent_ticks_and_t1_rooted_local_stages() {
497        let root = LifecycleContext::new(7, None, None)
498            .with_capability(LifecycleCapability::Ws, false)
499            .with_capability(LifecycleCapability::Http, true);
500        let local = root
501            .local_stage(LifecycleStage::T4)
502            .with_stage_status(LifecycleStage::T4, LifecycleStatus::Started);
503        let next = LifecycleContext::new(8, Some(7), None);
504
505        assert_eq!(root.tick_id(), 7);
506        assert_eq!(next.tick_id(), 8);
507        assert_eq!(next.parent_tick_id(), Some(7));
508        assert_eq!(local.parent_stage(), LifecycleStage::T1);
509        assert_eq!(
510            local.stage_status(LifecycleStage::T4),
511            LifecycleStatus::Started
512        );
513        assert_eq!(
514            root.capability_status(LifecycleCapability::Ws),
515            LifecycleStatus::NotApplicable
516        );
517        assert_eq!(
518            root.capability_status(LifecycleCapability::Http),
519            LifecycleStatus::Skipped
520        );
521    }
522
523    #[test]
524    fn tracker_keeps_parent_and_carrier_isolated_across_correlations() {
525        let tracker = LifecycleTracker::default();
526        let carrier = TraceCarrier::from_headers(&[(
527            "traceparent".to_string(),
528            "00-00000000000000000000000000000001-0000000000000002-01".to_string(),
529        )]);
530        let first = LifecycleContext::new(11, None, carrier);
531        let second = LifecycleContext::new(12, None, None);
532        let first_effect = OwnedEffect::Http {
533            corr: Correlation::from_raw(1),
534            req: HttpRequest {
535                method: "GET".to_string(),
536                url: "https://example.test".to_string(),
537                headers: Vec::new(),
538                body: None,
539            },
540        };
541        let second_effect = OwnedEffect::Http {
542            corr: Correlation::from_raw(2),
543            req: HttpRequest {
544                method: "GET".to_string(),
545                url: "https://example.test".to_string(),
546                headers: Vec::new(),
547                body: None,
548            },
549        };
550        tracker.remember_effect(&first_effect, &first);
551        tracker.remember_effect(&second_effect, &second);
552        let (first_parent, first_carrier, first_span_parent) =
553            tracker.parent_for_tick(&Tick::PortReply {
554                corr: Correlation::from_raw(1),
555                outcome: helix_core::tick::PortOutcome::Ok(helix_core::tick::ReplyBytes(
556                    Bytes::new(),
557                )),
558            });
559        let (second_parent, second_carrier, second_span_parent) =
560            tracker.parent_for_tick(&Tick::PortReply {
561                corr: Correlation::from_raw(2),
562                outcome: helix_core::tick::PortOutcome::Ok(helix_core::tick::ReplyBytes(
563                    Bytes::new(),
564                )),
565            });
566        assert_eq!(first_parent, Some(11));
567        assert_eq!(second_parent, Some(12));
568        assert!(first_carrier.is_some());
569        assert!(second_carrier.is_none());
570        assert!(first_span_parent.is_none());
571        assert!(second_span_parent.is_none());
572    }
573
574    #[test]
575    fn lifecycle_sink_is_non_blocking_and_bounded() {
576        let (sink, mut rx) = LifecycleTraceSink::channel();
577        let context = LifecycleContext::new(1, None, None);
578        for _ in 0..LIFECYCLE_TRACE_QUEUE_CAPACITY {
579            assert!(sink.try_emit(LifecycleObservation {
580                tick_id: context.tick_id(),
581                parent_tick_id: context.parent_tick_id(),
582                stage: LifecycleStage::T1,
583                capability: None,
584                status: LifecycleStatus::Started,
585                reason: None,
586            }));
587        }
588        assert!(!sink.try_emit(LifecycleObservation {
589            tick_id: 1,
590            parent_tick_id: None,
591            stage: LifecycleStage::T4,
592            capability: Some(LifecycleCapability::Http),
593            status: LifecycleStatus::NotApplicable,
594            reason: Some("transport_absent"),
595        }));
596        assert_eq!(sink.stats().dropped_count(), 1);
597        rx.close();
598    }
599
600    #[test]
601    fn capability_matrix_marks_http_only_ws_only_and_no_network_explicitly() {
602        let no_network = LifecycleContext::new(1, None, None);
603        let http_only = no_network.with_capability(LifecycleCapability::Http, true);
604        let ws_only = no_network.with_capability(LifecycleCapability::Ws, true);
605
606        assert_eq!(
607            no_network.capability_status(LifecycleCapability::Http),
608            LifecycleStatus::NotApplicable
609        );
610        assert_eq!(
611            no_network.capability_status(LifecycleCapability::Ws),
612            LifecycleStatus::NotApplicable
613        );
614        assert_eq!(
615            http_only.capability_status(LifecycleCapability::Http),
616            LifecycleStatus::Skipped
617        );
618        assert_eq!(
619            http_only.capability_status(LifecycleCapability::Ws),
620            LifecycleStatus::NotApplicable
621        );
622        assert_eq!(
623            ws_only.capability_status(LifecycleCapability::Ws),
624            LifecycleStatus::Skipped
625        );
626        assert_eq!(
627            ws_only.capability_status(LifecycleCapability::Http),
628            LifecycleStatus::NotApplicable
629        );
630    }
631
632    #[test]
633    fn concurrent_trackers_keep_parent_links_isolated() {
634        let tracker = LifecycleTracker::default();
635        let first_tracker = tracker.clone();
636        let second_tracker = tracker.clone();
637        let first = std::thread::spawn(move || {
638            let context = LifecycleContext::new(101, None, None);
639            let effect = OwnedEffect::Http {
640                corr: Correlation::from_raw(101),
641                req: HttpRequest {
642                    method: "GET".to_string(),
643                    url: "https://example.test/one".to_string(),
644                    headers: Vec::new(),
645                    body: None,
646                },
647            };
648            first_tracker.remember_effect(&effect, &context);
649        });
650        let second = std::thread::spawn(move || {
651            let context = LifecycleContext::new(202, None, None);
652            let effect = OwnedEffect::Http {
653                corr: Correlation::from_raw(202),
654                req: HttpRequest {
655                    method: "GET".to_string(),
656                    url: "https://example.test/two".to_string(),
657                    headers: Vec::new(),
658                    body: None,
659                },
660            };
661            second_tracker.remember_effect(&effect, &context);
662        });
663        assert!(first.join().is_ok());
664        assert!(second.join().is_ok());
665
666        let (first_parent, _, _) = tracker.parent_for_tick(&Tick::PortReply {
667            corr: Correlation::from_raw(101),
668            outcome: helix_core::tick::PortOutcome::Err(helix_core::tick::PortError::Timeout),
669        });
670        let (second_parent, _, _) = tracker.parent_for_tick(&Tick::PortReply {
671            corr: Correlation::from_raw(202),
672            outcome: helix_core::tick::PortOutcome::Err(helix_core::tick::PortError::Timeout),
673        });
674        assert_eq!(first_parent, Some(101));
675        assert_eq!(second_parent, Some(202));
676    }
677}
helix_driver_host/lifecycle.rs

helix_driver_host/
lifecycle.rs
