harn_vm/triggers/
streaming.rs

1use std::collections::{BTreeMap, VecDeque};
2use std::time::Duration;
3
4use serde::{Deserialize, Serialize};
5use serde_json::{json, Value as JsonValue};
6use time::OffsetDateTime;
7
8use crate::event_log::{AnyEventLog, EventLog, LogEvent, Topic};
9use crate::triggers::test_util::clock;
10use crate::triggers::{
11    Dispatcher, ProviderPayload, SignatureStatus, TraceId, TriggerEvent, TriggerEventId,
12};
13
14use super::{DispatchError, DispatchOutcome, TRIGGERS_LIFECYCLE_TOPIC, TRIGGER_DLQ_TOPIC};
15
16pub const TRIGGER_STREAM_WINDOWS_TOPIC: &str = "trigger.stream.windows";
17pub const TRIGGER_STREAM_STATUS_TOPIC: &str = "trigger.stream.status";
18pub const TRIGGER_STREAM_GATE_TOPIC: &str = "trigger.stream.gates";
19
20#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
21#[serde(rename_all = "snake_case")]
22pub enum StreamWindowMode {
23    Fixed,
24    Tumbling,
25    Sliding,
26}
27
28#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
29pub struct StreamWindowConfig {
30    pub mode: StreamWindowMode,
31    pub size: usize,
32    #[serde(default = "default_window_step")]
33    pub step: usize,
34}
35
36impl StreamWindowConfig {
37    pub fn fixed(size: usize) -> Self {
38        Self {
39            mode: StreamWindowMode::Fixed,
40            size,
41            step: size.max(1),
42        }
43    }
44
45    pub fn tumbling(size: usize) -> Self {
46        Self {
47            mode: StreamWindowMode::Tumbling,
48            size,
49            step: size.max(1),
50        }
51    }
52
53    pub fn sliding(size: usize, step: usize) -> Self {
54        Self {
55            mode: StreamWindowMode::Sliding,
56            size,
57            step: step.max(1),
58        }
59    }
60
61    fn validate(&self) -> Result<(), DispatchError> {
62        if self.size == 0 {
63            return Err(DispatchError::Local(
64                "stream window size must be positive".to_string(),
65            ));
66        }
67        if self.step == 0 {
68            return Err(DispatchError::Local(
69                "stream window step must be positive".to_string(),
70            ));
71        }
72        // A sliding window drains `step` events after each emit, but events
73        // are admitted one at a time so the buffer never exceeds `size`.
74        // A step larger than size would therefore silently collapse to
75        // tumbling (min(step, size) == size), never skipping the intended
76        // gap events. Gap-sampling is not supported by this drain design,
77        // so reject it rather than emit misleading windows.
78        if self.mode == StreamWindowMode::Sliding && self.step > self.size {
79            return Err(DispatchError::Local(format!(
80                "sliding stream window step ({}) must not exceed size ({}); \
81                 a hop larger than the window is not supported",
82                self.step, self.size
83            )));
84        }
85        Ok(())
86    }
87
88    fn drain_after_emit(&self, pending: &mut VecDeque<TriggerEvent>) {
89        match self.mode {
90            StreamWindowMode::Fixed | StreamWindowMode::Tumbling => pending.clear(),
91            StreamWindowMode::Sliding => {
92                let remove = self.step.min(pending.len());
93                pending.drain(..remove);
94            }
95        }
96    }
97}
98
99#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
100#[serde(rename_all = "snake_case")]
101pub enum StreamOverflowPolicy {
102    DropNewest,
103    DropOldest,
104    DeadLetterNewest,
105}
106
107#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
108pub struct StreamBackpressureConfig {
109    pub max_pending_events: usize,
110    pub overflow: StreamOverflowPolicy,
111}
112
113impl Default for StreamBackpressureConfig {
114    fn default() -> Self {
115        Self {
116            max_pending_events: 1024,
117            overflow: StreamOverflowPolicy::DeadLetterNewest,
118        }
119    }
120}
121
122#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
123pub struct StreamThrottleConfig {
124    pub max: usize,
125    #[serde(with = "duration_millis")]
126    pub period: Duration,
127}
128
129#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
130pub struct StreamFlowConfig {
131    pub debounce_by_dedupe_key: bool,
132    pub throttle: Option<StreamThrottleConfig>,
133}
134
135#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
136pub struct StreamGateConfig {
137    pub gate_id: String,
138    pub cache_key: String,
139    pub replay_of_event_id: Option<String>,
140}
141
142#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
143pub struct StreamTriggerConfig {
144    pub stream_id: String,
145    pub window: StreamWindowConfig,
146    #[serde(default)]
147    pub backpressure: StreamBackpressureConfig,
148    #[serde(default)]
149    pub flow: StreamFlowConfig,
150    #[serde(default, skip_serializing_if = "Option::is_none")]
151    pub gate: Option<StreamGateConfig>,
152}
153
154impl StreamTriggerConfig {
155    pub fn validate(&self) -> Result<(), DispatchError> {
156        if self.stream_id.trim().is_empty() {
157            return Err(DispatchError::Local(
158                "stream id must be a non-empty string".to_string(),
159            ));
160        }
161        self.window.validate()?;
162        if self.backpressure.max_pending_events == 0 {
163            return Err(DispatchError::Local(
164                "stream max_pending_events must be positive".to_string(),
165            ));
166        }
167        if let Some(throttle) = self.flow.throttle.as_ref() {
168            if throttle.max == 0 || throttle.period.is_zero() {
169                return Err(DispatchError::Local(
170                    "stream throttle requires positive max and period".to_string(),
171                ));
172            }
173        }
174        Ok(())
175    }
176}
177
178#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
179pub struct StreamWindowEnvelope {
180    pub stream_id: String,
181    pub window_id: String,
182    pub event_ids: Vec<String>,
183    pub dedupe_keys: Vec<String>,
184    pub first_occurred_at: Option<OffsetDateTime>,
185    pub last_occurred_at: Option<OffsetDateTime>,
186    pub replay_of_event_id: Option<String>,
187}
188
189#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
190pub struct StreamStatusSnapshot {
191    pub stream_id: String,
192    pub pending_events: usize,
193    pub admitted_events: u64,
194    pub dropped_events: u64,
195    pub dead_lettered_events: u64,
196    pub emitted_windows: u64,
197    pub gate_passed: u64,
198    pub gate_blocked: u64,
199}
200
201#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
202pub struct StreamGateRecord {
203    pub gate_id: String,
204    pub cache_key: String,
205    pub window_id: String,
206    pub result: bool,
207    pub cached: bool,
208    pub reason: Option<String>,
209    pub replay_of_event_id: Option<String>,
210}
211
212#[derive(Clone, Debug, PartialEq, Eq)]
213pub enum StreamGateOutcome {
214    Pass { reason: Option<String> },
215    Block { reason: Option<String> },
216}
217
218impl StreamGateOutcome {
219    fn result(&self) -> bool {
220        matches!(self, Self::Pass { .. })
221    }
222
223    fn reason(&self) -> Option<String> {
224        match self {
225            Self::Pass { reason } | Self::Block { reason } => reason.clone(),
226        }
227    }
228}
229
230#[derive(Debug, Default)]
231struct StreamRuntimeState {
232    pending: VecDeque<TriggerEvent>,
233    throttle_hits: VecDeque<OffsetDateTime>,
234    status: StreamStatusSnapshot,
235}
236
237pub struct StreamTriggerRuntime {
238    config: StreamTriggerConfig,
239    event_log: std::sync::Arc<AnyEventLog>,
240    dispatcher: Dispatcher,
241    state: StreamRuntimeState,
242}
243
244impl StreamTriggerRuntime {
245    pub fn new(
246        config: StreamTriggerConfig,
247        event_log: std::sync::Arc<AnyEventLog>,
248        dispatcher: Dispatcher,
249    ) -> Result<Self, DispatchError> {
250        config.validate()?;
251        let status = StreamStatusSnapshot {
252            stream_id: config.stream_id.clone(),
253            ..Default::default()
254        };
255        Ok(Self {
256            config,
257            event_log,
258            dispatcher,
259            state: StreamRuntimeState {
260                status,
261                ..Default::default()
262            },
263        })
264    }
265
266    pub fn snapshot(&self) -> StreamStatusSnapshot {
267        let mut snapshot = self.state.status.clone();
268        snapshot.pending_events = self.state.pending.len();
269        snapshot
270    }
271
272    pub async fn push_event(
273        &mut self,
274        event: TriggerEvent,
275    ) -> Result<Vec<DispatchOutcome>, DispatchError> {
276        self.push_event_with_gate(event, |_| StreamGateOutcome::Pass { reason: None })
277            .await
278    }
279
280    pub async fn push_event_with_gate(
281        &mut self,
282        event: TriggerEvent,
283        gate: impl Fn(&StreamWindowEnvelope) -> StreamGateOutcome,
284    ) -> Result<Vec<DispatchOutcome>, DispatchError> {
285        self.append_status_event(
286            "stream_event_received",
287            json!({
288                "stream_id": self.config.stream_id,
289                "event_id": event.id.0,
290                "dedupe_key": event.dedupe_key,
291                "provider": event.provider.as_str(),
292                "kind": event.kind,
293            }),
294        )
295        .await?;
296
297        if !self.apply_throttle(&event).await? {
298            return Ok(Vec::new());
299        }
300
301        self.admit_event(event).await?;
302        let windows = self.emit_ready_windows().await?;
303        let mut outcomes = Vec::new();
304        for (envelope, window_event) in windows {
305            if !self.evaluate_gate(&envelope, &gate).await? {
306                continue;
307            }
308            outcomes.extend(self.dispatcher.dispatch_event(window_event).await?);
309        }
310        Ok(outcomes)
311    }
312
313    async fn admit_event(&mut self, event: TriggerEvent) -> Result<(), DispatchError> {
314        if self.config.flow.debounce_by_dedupe_key {
315            let before = self.state.pending.len();
316            self.state
317                .pending
318                .retain(|pending| pending.dedupe_key != event.dedupe_key);
319            if self.state.pending.len() != before {
320                self.append_status_event(
321                    "stream_event_debounced",
322                    json!({
323                        "stream_id": self.config.stream_id,
324                        "dedupe_key": event.dedupe_key,
325                    }),
326                )
327                .await?;
328            }
329        }
330
331        if self.state.pending.len() >= self.config.backpressure.max_pending_events {
332            match self.config.backpressure.overflow {
333                StreamOverflowPolicy::DropNewest => {
334                    self.state.status.dropped_events =
335                        self.state.status.dropped_events.saturating_add(1);
336                    self.append_status_event(
337                        "stream_event_dropped",
338                        json!({
339                            "stream_id": self.config.stream_id,
340                            "event_id": event.id.0,
341                            "reason": "backpressure_drop_newest",
342                            "max_pending_events": self.config.backpressure.max_pending_events,
343                        }),
344                    )
345                    .await?;
346                    return Ok(());
347                }
348                StreamOverflowPolicy::DropOldest => {
349                    if let Some(dropped) = self.state.pending.pop_front() {
350                        self.state.status.dropped_events =
351                            self.state.status.dropped_events.saturating_add(1);
352                        self.append_status_event(
353                            "stream_event_dropped",
354                            json!({
355                                "stream_id": self.config.stream_id,
356                                "event_id": dropped.id.0,
357                                "reason": "backpressure_drop_oldest",
358                                "max_pending_events": self.config.backpressure.max_pending_events,
359                            }),
360                        )
361                        .await?;
362                    }
363                }
364                StreamOverflowPolicy::DeadLetterNewest => {
365                    self.dead_letter_event(&event, "backpressure_queue_full")
366                        .await?;
367                    return Ok(());
368                }
369            }
370        }
371
372        self.state.status.admitted_events = self.state.status.admitted_events.saturating_add(1);
373        self.state.pending.push_back(event);
374        Ok(())
375    }
376
377    async fn emit_ready_windows(
378        &mut self,
379    ) -> Result<Vec<(StreamWindowEnvelope, TriggerEvent)>, DispatchError> {
380        let mut windows = Vec::new();
381        while self.state.pending.len() >= self.config.window.size {
382            let members = self
383                .state
384                .pending
385                .iter()
386                .take(self.config.window.size)
387                .cloned()
388                .collect::<Vec<_>>();
389            let envelope = self.window_envelope(&members);
390            let event = self.window_event(&members, &envelope)?;
391            self.append_window_event(&envelope).await?;
392            self.state.status.emitted_windows = self.state.status.emitted_windows.saturating_add(1);
393            windows.push((envelope, event));
394            self.config.window.drain_after_emit(&mut self.state.pending);
395        }
396        Ok(windows)
397    }
398
399    fn window_envelope(&self, members: &[TriggerEvent]) -> StreamWindowEnvelope {
400        let event_ids = members
401            .iter()
402            .map(|event| event.id.0.clone())
403            .collect::<Vec<_>>();
404        let dedupe_keys = members
405            .iter()
406            .map(|event| event.dedupe_key.clone())
407            .collect::<Vec<_>>();
408        let first_occurred_at = members
409            .iter()
410            .filter_map(|event| event.occurred_at)
411            .min()
412            .or_else(|| members.first().map(|event| event.received_at));
413        let last_occurred_at = members
414            .iter()
415            .filter_map(|event| event.occurred_at)
416            .max()
417            .or_else(|| members.last().map(|event| event.received_at));
418        let source = dedupe_keys.join("+");
419        StreamWindowEnvelope {
420            stream_id: self.config.stream_id.clone(),
421            window_id: format!("stream_window:{}:{source}", self.config.stream_id),
422            event_ids,
423            dedupe_keys,
424            first_occurred_at,
425            last_occurred_at,
426            replay_of_event_id: self
427                .config
428                .gate
429                .as_ref()
430                .and_then(|gate| gate.replay_of_event_id.clone()),
431        }
432    }
433
434    fn window_event(
435        &self,
436        members: &[TriggerEvent],
437        envelope: &StreamWindowEnvelope,
438    ) -> Result<TriggerEvent, DispatchError> {
439        let first = members
440            .first()
441            .ok_or_else(|| DispatchError::Local("stream window cannot be empty".to_string()))?;
442        let mut headers = first.headers.clone();
443        headers.insert("harn_stream_id".to_string(), envelope.stream_id.clone());
444        headers.insert(
445            "harn_stream_window_id".to_string(),
446            envelope.window_id.clone(),
447        );
448        headers.insert(
449            "harn_stream_source_event_ids".to_string(),
450            envelope.event_ids.join(","),
451        );
452        let batch = members
453            .iter()
454            .map(|event| serde_json::to_value(event).map_err(|error| error.to_string()))
455            .collect::<Result<Vec<_>, _>>()
456            .map_err(DispatchError::Serde)?;
457        let mut window_event = first.clone();
458        window_event.id = TriggerEventId::new();
459        window_event.kind = format!("{}.window", first.kind);
460        window_event.dedupe_key = envelope.window_id.clone();
461        window_event.trace_id = TraceId::new();
462        window_event.headers = headers;
463        window_event.batch = Some(batch);
464        Ok(window_event)
465    }
466
467    async fn evaluate_gate(
468        &mut self,
469        envelope: &StreamWindowEnvelope,
470        gate: &impl Fn(&StreamWindowEnvelope) -> StreamGateOutcome,
471    ) -> Result<bool, DispatchError> {
472        let Some(config) = self.config.gate.clone() else {
473            return Ok(true);
474        };
475        let cache_key = format!("{}:{}", config.cache_key, envelope.dedupe_keys.join("+"));
476        if let Some(record) = self.read_gate_record(&config.gate_id, &cache_key).await? {
477            self.append_gate_record(StreamGateRecord {
478                cached: true,
479                ..record.clone()
480            })
481            .await?;
482            if record.result {
483                self.state.status.gate_passed = self.state.status.gate_passed.saturating_add(1);
484            } else {
485                self.state.status.gate_blocked = self.state.status.gate_blocked.saturating_add(1);
486            }
487            return Ok(record.result);
488        }
489
490        let outcome = gate(envelope);
491        let record = StreamGateRecord {
492            gate_id: config.gate_id,
493            cache_key,
494            window_id: envelope.window_id.clone(),
495            result: outcome.result(),
496            cached: false,
497            reason: outcome.reason(),
498            replay_of_event_id: config.replay_of_event_id,
499        };
500        self.append_gate_record(record.clone()).await?;
501        if record.result {
502            self.state.status.gate_passed = self.state.status.gate_passed.saturating_add(1);
503        } else {
504            self.state.status.gate_blocked = self.state.status.gate_blocked.saturating_add(1);
505            self.append_status_event(
506                "stream_window_gate_blocked",
507                json!({
508                    "stream_id": self.config.stream_id,
509                    "window_id": envelope.window_id,
510                    "gate_id": record.gate_id,
511                    "reason": record.reason,
512                }),
513            )
514            .await?;
515        }
516        Ok(record.result)
517    }
518
519    async fn apply_throttle(&mut self, event: &TriggerEvent) -> Result<bool, DispatchError> {
520        let Some(throttle) = self.config.flow.throttle.clone() else {
521            return Ok(true);
522        };
523        let now = event.occurred_at.unwrap_or(event.received_at);
524        trim_hits(&mut self.state.throttle_hits, now, throttle.period);
525        if self.state.throttle_hits.len() >= throttle.max {
526            self.state.status.dropped_events = self.state.status.dropped_events.saturating_add(1);
527            self.append_status_event(
528                "stream_event_throttled",
529                json!({
530                    "stream_id": self.config.stream_id,
531                    "event_id": event.id.0,
532                    "max": throttle.max,
533                    "period_ms": throttle.period.as_millis(),
534                }),
535            )
536            .await?;
537            return Ok(false);
538        }
539        self.state.throttle_hits.push_back(now);
540        Ok(true)
541    }
542
543    async fn dead_letter_event(
544        &mut self,
545        event: &TriggerEvent,
546        reason: &str,
547    ) -> Result<(), DispatchError> {
548        self.state.status.dead_lettered_events =
549            self.state.status.dead_lettered_events.saturating_add(1);
550        self.append_topic_event(
551            TRIGGER_DLQ_TOPIC,
552            "stream_dead_lettered",
553            json!({
554                "stream_id": self.config.stream_id,
555                "event": event,
556                "reason": reason,
557                "pending_events": self.state.pending.len(),
558                "max_pending_events": self.config.backpressure.max_pending_events,
559            }),
560        )
561        .await?;
562        self.append_status_event(
563            "stream_event_dead_lettered",
564            json!({
565                "stream_id": self.config.stream_id,
566                "event_id": event.id.0,
567                "reason": reason,
568            }),
569        )
570        .await
571    }
572
573    async fn append_window_event(
574        &self,
575        envelope: &StreamWindowEnvelope,
576    ) -> Result<(), DispatchError> {
577        self.append_topic_event(
578            TRIGGER_STREAM_WINDOWS_TOPIC,
579            "stream_window_emitted",
580            serde_json::to_value(envelope)
581                .map_err(|error| DispatchError::Serde(error.to_string()))?,
582        )
583        .await?;
584        self.append_topic_event(
585            TRIGGERS_LIFECYCLE_TOPIC,
586            "StreamWindowEmitted",
587            json!({
588                "stream_id": envelope.stream_id,
589                "window_id": envelope.window_id,
590                "event_ids": envelope.event_ids,
591                "replay_of_event_id": envelope.replay_of_event_id,
592            }),
593        )
594        .await
595    }
596
597    async fn append_gate_record(&self, record: StreamGateRecord) -> Result<(), DispatchError> {
598        self.append_topic_event(
599            TRIGGER_STREAM_GATE_TOPIC,
600            "stream_gate_decision",
601            serde_json::to_value(record)
602                .map_err(|error| DispatchError::Serde(error.to_string()))?,
603        )
604        .await
605    }
606
607    async fn read_gate_record(
608        &self,
609        gate_id: &str,
610        cache_key: &str,
611    ) -> Result<Option<StreamGateRecord>, DispatchError> {
612        let topic =
613            Topic::new(TRIGGER_STREAM_GATE_TOPIC).expect("static stream gate topic is valid");
614        let records = self
615            .event_log
616            .read_range(&topic, None, usize::MAX)
617            .await
618            .map_err(DispatchError::from)?;
619        Ok(records
620            .into_iter()
621            .rev()
622            .filter(|(_, event)| event.kind == "stream_gate_decision")
623            .filter_map(|(_, event)| serde_json::from_value::<StreamGateRecord>(event.payload).ok())
624            .find(|record| record.gate_id == gate_id && record.cache_key == cache_key))
625    }
626
627    async fn append_status_event(
628        &self,
629        kind: &str,
630        mut payload: JsonValue,
631    ) -> Result<(), DispatchError> {
632        if let JsonValue::Object(ref mut object) = payload {
633            object.insert("status".to_string(), json!(self.snapshot()));
634        }
635        self.append_topic_event(TRIGGER_STREAM_STATUS_TOPIC, kind, payload)
636            .await
637    }
638
639    async fn append_topic_event(
640        &self,
641        topic: &str,
642        kind: &str,
643        payload: JsonValue,
644    ) -> Result<(), DispatchError> {
645        self.event_log
646            .append(
647                &Topic::new(topic).expect("static stream trigger topic is valid"),
648                LogEvent::new(kind, payload),
649            )
650            .await
651            .map(|_| ())
652            .map_err(DispatchError::from)
653    }
654}
655
656pub fn stream_window_summary(event: &TriggerEvent) -> Option<StreamWindowEnvelope> {
657    let batch = event.batch.as_ref()?;
658    let event_ids = batch
659        .iter()
660        .filter_map(|member| member.get("id"))
661        .filter_map(JsonValue::as_str)
662        .map(ToString::to_string)
663        .collect::<Vec<_>>();
664    Some(StreamWindowEnvelope {
665        stream_id: event
666            .headers
667            .get("harn_stream_id")
668            .cloned()
669            .unwrap_or_default(),
670        window_id: event
671            .headers
672            .get("harn_stream_window_id")
673            .cloned()
674            .unwrap_or_else(|| event.dedupe_key.clone()),
675        event_ids,
676        dedupe_keys: batch
677            .iter()
678            .filter_map(|member| member.get("dedupe_key"))
679            .filter_map(JsonValue::as_str)
680            .map(ToString::to_string)
681            .collect(),
682        first_occurred_at: None,
683        last_occurred_at: None,
684        replay_of_event_id: None,
685    })
686}
687
688pub fn stream_fixture_event(
689    provider: impl Into<String>,
690    kind: impl Into<String>,
691    stream: impl Into<String>,
692    offset: u64,
693    payload: JsonValue,
694) -> TriggerEvent {
695    let provider = provider.into();
696    let stream = stream.into();
697    let kind = kind.into();
698    let provider_id = crate::triggers::ProviderId::from(provider);
699    let raw = json!({
700        "event": kind,
701        "stream": stream,
702        "offset": offset,
703        "value": payload,
704    });
705    TriggerEvent::new(
706        provider_id.clone(),
707        kind.clone(),
708        Some(clock::now_utc()),
709        format!("{stream}:{offset}"),
710        None,
711        BTreeMap::new(),
712        ProviderPayload::normalize(&provider_id, &kind, &BTreeMap::new(), raw).unwrap_or_else(
713            |_| {
714                ProviderPayload::Known(crate::triggers::event::KnownProviderPayload::Webhook(
715                    crate::triggers::GenericWebhookPayload {
716                        source: Some("stream_fixture".to_string()),
717                        content_type: Some("application/json".to_string()),
718                        raw: json!({}),
719                    },
720                ))
721            },
722        ),
723        SignatureStatus::Unsigned,
724    )
725}
726
727fn trim_hits(hits: &mut VecDeque<OffsetDateTime>, now: OffsetDateTime, period: Duration) {
728    let period = time::Duration::try_from(period).unwrap_or_default();
729    while hits.front().is_some_and(|hit| now - *hit >= period) {
730        hits.pop_front();
731    }
732}
733
734fn default_window_step() -> usize {
735    1
736}
737
738mod duration_millis {
739    use std::time::Duration;
740
741    use serde::{Deserialize, Deserializer, Serializer};
742
743    pub fn serialize<S>(duration: &Duration, serializer: S) -> Result<S::Ok, S::Error>
744    where
745        S: Serializer,
746    {
747        serializer.serialize_u64(duration.as_millis() as u64)
748    }
749
750    pub fn deserialize<'de, D>(deserializer: D) -> Result<Duration, D::Error>
751    where
752        D: Deserializer<'de>,
753    {
754        Ok(Duration::from_millis(u64::deserialize(deserializer)?))
755    }
756}
757
758#[cfg(test)]
759mod tests {
760    use std::cell::Cell;
761    use std::rc::Rc;
762    use std::sync::Arc;
763
764    use serde_json::json;
765
766    use crate::event_log::{install_memory_for_current_thread, EventLog};
767    use crate::stdlib::register_vm_stdlib;
768    use crate::triggers::dispatcher::RetryPolicy;
769    use crate::triggers::TriggerRetryConfig;
770    use crate::triggers::{
771        clear_trigger_registry, install_manifest_triggers, ProviderId, TriggerBindingSource,
772        TriggerBindingSpec, TriggerHandlerSpec,
773    };
774    use crate::trust_graph::AutonomyTier;
775    use crate::vm::Vm;
776
777    use super::*;
778
779    fn install_test_event_log() -> Arc<AnyEventLog> {
780        install_memory_for_current_thread(
781            crate::runtime_limits::RuntimeLimits::DEFAULT.default_event_log_queue_depth,
782        )
783    }
784
785    async fn read_topic(
786        log: Arc<AnyEventLog>,
787        topic: &str,
788    ) -> Vec<(u64, crate::event_log::LogEvent)> {
789        log.read_range(&Topic::new(topic).unwrap(), None, usize::MAX)
790            .await
791            .unwrap()
792    }
793
794    async fn stream_dispatcher_fixture(
795        source: &str,
796    ) -> (tempfile::TempDir, Arc<AnyEventLog>, Dispatcher) {
797        crate::reset_thread_local_state();
798        clear_trigger_registry();
799        let dir = tempfile::tempdir().expect("tempdir");
800        let log = install_test_event_log();
801        let lib_path = dir.path().join("lib.harn");
802        std::fs::write(&lib_path, source).expect("write harn source");
803
804        let mut vm = Vm::new();
805        register_vm_stdlib(&mut vm);
806        vm.set_source_dir(dir.path());
807        let exports = vm
808            .load_module_exports(&lib_path)
809            .await
810            .expect("load stream handler");
811        let handler = exports.get("local_fn").expect("local_fn export").clone();
812        install_manifest_triggers(vec![TriggerBindingSpec {
813            id: "stream-window-handler".to_string(),
814            source: TriggerBindingSource::Manifest,
815            kind: "stream".to_string(),
816            provider: ProviderId::from("kafka"),
817            autonomy_tier: AutonomyTier::ActAuto,
818            handler: TriggerHandlerSpec::Local {
819                raw: "local_fn".to_string(),
820                closure: handler,
821            },
822            dispatch_priority: crate::triggers::WorkerQueuePriority::Normal,
823            when: None,
824            when_budget: None,
825            retry: TriggerRetryConfig::new(1, RetryPolicy::Linear { delay_ms: 1 }),
826            match_events: vec!["issues.opened.window".to_string()],
827            dedupe_key: Some("event.dedupe_key".to_string()),
828            dedupe_retention_days: crate::triggers::DEFAULT_INBOX_RETENTION_DAYS,
829            filter: None,
830            daily_cost_usd: None,
831            hourly_cost_usd: None,
832            max_autonomous_decisions_per_hour: None,
833            max_autonomous_decisions_per_day: None,
834            on_budget_exhausted: crate::triggers::TriggerBudgetExhaustionStrategy::False,
835            max_concurrent: None,
836            flow_control: crate::triggers::TriggerFlowControlConfig::default(),
837            aggregation: None,
838            manifest_path: None,
839            package_name: Some("workspace".to_string()),
840            definition_fingerprint: "stream-window-handler:v1".to_string(),
841        }])
842        .await
843        .expect("install stream trigger");
844
845        (dir, log.clone(), Dispatcher::with_event_log(vm, log))
846    }
847
848    #[tokio::test(flavor = "current_thread")]
849    async fn chat_stream_windows_gate_and_dispatch_deterministically() {
850        let local = tokio::task::LocalSet::new();
851        local
852            .run_until(async {
853                let (_dir, log, dispatcher) = stream_dispatcher_fixture(
854                    r#"
855import "std/triggers"
856
857pub fn local_fn(event: TriggerEvent) -> int {
858  return len(event.batch ?? [])
859}
860"#,
861                )
862                .await;
863
864                let mut runtime = StreamTriggerRuntime::new(
865                    StreamTriggerConfig {
866                        stream_id: "chat:triage".to_string(),
867                        window: StreamWindowConfig::tumbling(2),
868                        backpressure: StreamBackpressureConfig::default(),
869                        flow: StreamFlowConfig::default(),
870                        gate: Some(StreamGateConfig {
871                            gate_id: "chat-llm-gate".to_string(),
872                            cache_key: "chat:v1".to_string(),
873                            replay_of_event_id: None,
874                        }),
875                    },
876                    log.clone(),
877                    dispatcher,
878                )
879                .unwrap();
880
881                let calls = Rc::new(Cell::new(0));
882                for offset in 0..2 {
883                    let calls = calls.clone();
884                    let outcomes = runtime
885                        .push_event_with_gate(
886                            stream_fixture_event(
887                                "kafka",
888                                "issues.opened",
889                                "chat",
890                                offset,
891                                json!({"text": format!("message {offset}")}),
892                            ),
893                            move |_| {
894                                calls.set(calls.get() + 1);
895                                StreamGateOutcome::Pass {
896                                    reason: Some("fixture-positive".to_string()),
897                                }
898                            },
899                        )
900                        .await
901                        .unwrap();
902                    if offset == 1 {
903                        assert_eq!(outcomes.len(), 1);
904                        assert_eq!(outcomes[0].result, Some(json!(2)));
905                    } else {
906                        assert!(outcomes.is_empty());
907                    }
908                }
909                assert_eq!(calls.get(), 1);
910
911                let gate_events = read_topic(log.clone(), TRIGGER_STREAM_GATE_TOPIC).await;
912                assert_eq!(
913                    gate_events
914                        .iter()
915                        .filter(|(_, event)| event.kind == "stream_gate_decision")
916                        .count(),
917                    1
918                );
919
920                let first_window_id = gate_events[0].1.payload["window_id"]
921                    .as_str()
922                    .unwrap()
923                    .to_string();
924                runtime.config.gate.as_mut().unwrap().replay_of_event_id =
925                    Some(first_window_id.clone());
926
927                for offset in 0..2 {
928                    let calls = calls.clone();
929                    runtime
930                        .push_event_with_gate(
931                            stream_fixture_event(
932                                "kafka",
933                                "issues.opened",
934                                "chat",
935                                offset,
936                                json!({"text": format!("message {offset}")}),
937                            ),
938                            move |_| {
939                                calls.set(calls.get() + 1);
940                                StreamGateOutcome::Block {
941                                    reason: Some("should-not-run".to_string()),
942                                }
943                            },
944                        )
945                        .await
946                        .unwrap();
947                }
948                assert_eq!(calls.get(), 1);
949                let gate_events = read_topic(log.clone(), TRIGGER_STREAM_GATE_TOPIC).await;
950                assert!(gate_events.iter().any(|(_, event)| {
951                    event.kind == "stream_gate_decision"
952                        && event.payload["cached"] == json!(true)
953                        && event.payload["window_id"] == json!(first_window_id)
954                }));
955            })
956            .await;
957    }
958
959    #[tokio::test(flavor = "current_thread")]
960    async fn issue_feed_backpressure_dead_letters_overflow() {
961        let local = tokio::task::LocalSet::new();
962        local
963            .run_until(async {
964                let dir = tempfile::tempdir().unwrap();
965                let log = install_test_event_log();
966                let mut vm = Vm::new();
967                register_vm_stdlib(&mut vm);
968                vm.set_source_dir(dir.path());
969                let dispatcher = Dispatcher::with_event_log(vm, log.clone());
970                let mut runtime = StreamTriggerRuntime::new(
971                    StreamTriggerConfig {
972                        stream_id: "issue-feed".to_string(),
973                        window: StreamWindowConfig::fixed(3),
974                        backpressure: StreamBackpressureConfig {
975                            max_pending_events: 1,
976                            overflow: StreamOverflowPolicy::DeadLetterNewest,
977                        },
978                        flow: StreamFlowConfig::default(),
979                        gate: None,
980                    },
981                    log.clone(),
982                    dispatcher,
983                )
984                .unwrap();
985
986                runtime
987                    .push_event(stream_fixture_event(
988                        "kafka",
989                        "issues.opened",
990                        "issues",
991                        1,
992                        json!({"number": 1}),
993                    ))
994                    .await
995                    .unwrap();
996                runtime
997                    .push_event(stream_fixture_event(
998                        "kafka",
999                        "issues.opened",
1000                        "issues",
1001                        2,
1002                        json!({"number": 2}),
1003                    ))
1004                    .await
1005                    .unwrap();
1006
1007                let snapshot = runtime.snapshot();
1008                assert_eq!(snapshot.pending_events, 1);
1009                assert_eq!(snapshot.dead_lettered_events, 1);
1010
1011                let dlq = read_topic(log.clone(), TRIGGER_DLQ_TOPIC).await;
1012                assert!(dlq.iter().any(|(_, event)| {
1013                    event.kind == "stream_dead_lettered"
1014                        && event.payload["reason"] == json!("backpressure_queue_full")
1015                }));
1016            })
1017            .await;
1018    }
1019
1020    #[tokio::test(flavor = "current_thread")]
1021    async fn sliding_windows_emit_overlapping_batches() {
1022        let local = tokio::task::LocalSet::new();
1023        local
1024            .run_until(async {
1025                let (_dir, log, dispatcher) = stream_dispatcher_fixture(
1026                    r#"
1027import "std/triggers"
1028
1029pub fn local_fn(event: TriggerEvent) -> int {
1030  return len(event.batch ?? [])
1031}
1032"#,
1033                )
1034                .await;
1035                let mut runtime = StreamTriggerRuntime::new(
1036                    StreamTriggerConfig {
1037                        stream_id: "sliding".to_string(),
1038                        window: StreamWindowConfig::sliding(2, 1),
1039                        backpressure: StreamBackpressureConfig::default(),
1040                        flow: StreamFlowConfig::default(),
1041                        gate: None,
1042                    },
1043                    log.clone(),
1044                    dispatcher,
1045                )
1046                .unwrap();
1047
1048                let mut dispatched = 0;
1049                for offset in 1..=3 {
1050                    dispatched += runtime
1051                        .push_event(stream_fixture_event(
1052                            "kafka",
1053                            "issues.opened",
1054                            "sliding",
1055                            offset,
1056                            json!({"offset": offset}),
1057                        ))
1058                        .await
1059                        .unwrap()
1060                        .len();
1061                }
1062
1063                assert_eq!(dispatched, 2);
1064                assert_eq!(runtime.snapshot().emitted_windows, 2);
1065                let windows = read_topic(log, TRIGGER_STREAM_WINDOWS_TOPIC).await;
1066                assert_eq!(windows.len(), 2);
1067            })
1068            .await;
1069    }
1070
1071    #[test]
1072    fn sliding_window_rejects_step_greater_than_size() {
1073        // sliding(2, 5): a hop larger than the window would silently
1074        // collapse to tumbling instead of gap-sampling, so it must be
1075        // rejected at validation time.
1076        let error = StreamWindowConfig::sliding(2, 5)
1077            .validate()
1078            .expect_err("step > size must be rejected");
1079        match error {
1080            DispatchError::Local(message) => {
1081                assert!(
1082                    message.contains("must not exceed size"),
1083                    "unexpected message: {message}"
1084                );
1085            }
1086            other => panic!("expected DispatchError::Local, got {other:?}"),
1087        }
1088    }
1089
1090    #[test]
1091    fn sliding_window_allows_step_up_to_size() {
1092        // Overlap (step < size) and the tumbling-equivalent (step == size)
1093        // remain valid; only step > size is rejected.
1094        StreamWindowConfig::sliding(3, 1)
1095            .validate()
1096            .expect("overlap sliding window is valid");
1097        StreamWindowConfig::sliding(3, 2)
1098            .validate()
1099            .expect("partial-overlap sliding window is valid");
1100        StreamWindowConfig::sliding(2, 2)
1101            .validate()
1102            .expect("step == size sliding window is valid");
1103    }
1104
1105    #[tokio::test(flavor = "current_thread")]
1106    async fn sliding_window_step_equal_size_emits_non_overlapping_batches() {
1107        let local = tokio::task::LocalSet::new();
1108        local
1109            .run_until(async {
1110                let (_dir, log, dispatcher) = stream_dispatcher_fixture(
1111                    r#"
1112import "std/triggers"
1113
1114pub fn local_fn(event: TriggerEvent) -> int {
1115  return len(event.batch ?? [])
1116}
1117"#,
1118                )
1119                .await;
1120                // step == size drains the whole window after each emit, so
1121                // windows tile without overlap (tumbling-equivalent).
1122                let mut runtime = StreamTriggerRuntime::new(
1123                    StreamTriggerConfig {
1124                        stream_id: "sliding-tumbling".to_string(),
1125                        window: StreamWindowConfig::sliding(2, 2),
1126                        backpressure: StreamBackpressureConfig::default(),
1127                        flow: StreamFlowConfig::default(),
1128                        gate: None,
1129                    },
1130                    log.clone(),
1131                    dispatcher,
1132                )
1133                .unwrap();
1134
1135                let mut dispatched = 0;
1136                for offset in 1..=4 {
1137                    dispatched += runtime
1138                        .push_event(stream_fixture_event(
1139                            "kafka",
1140                            "issues.opened",
1141                            "sliding-tumbling",
1142                            offset,
1143                            json!({"offset": offset}),
1144                        ))
1145                        .await
1146                        .unwrap()
1147                        .len();
1148                }
1149
1150                // 4 events, size 2, step 2 -> exactly 2 non-overlapping windows.
1151                assert_eq!(dispatched, 2);
1152                assert_eq!(runtime.snapshot().emitted_windows, 2);
1153                let windows = read_topic(log, TRIGGER_STREAM_WINDOWS_TOPIC).await;
1154                assert_eq!(windows.len(), 2);
1155            })
1156            .await;
1157    }
1158}
harn_vm/triggers/streaming.rs

harn_vm/triggers/
streaming.rs
