a3s_code_core/flow_graph/
decision.rs

1use super::decision_ledger::{
2    FlowDecisionClaimOutcome, FlowDecisionLedger, MemoryFlowDecisionLedger,
3};
4use crate::execution_identity::{
5    ExecutionClaimV1, ExecutionResultOutcomeV1, ExecutionResultReceiptV1,
6};
7use a3s_flow::RetryPolicy;
8use async_trait::async_trait;
9use serde::{Deserialize, Serialize};
10use serde_json::Value;
11use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
12use std::sync::Arc;
13use std::time::{Duration, Instant};
14use thiserror::Error;
15use tokio::sync::Mutex;
16
17#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
18#[serde(tag = "type", rename_all = "snake_case")]
19pub enum FlowDecision {
20    ScheduleStep { step: FlowDecisionStep },
21    ScheduleSteps { steps: Vec<FlowDecisionStep> },
22    Complete { output: Value },
23    Fail { error: String },
24}
25
26#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
27pub struct FlowDecisionStep {
28    pub step_id: String,
29    pub step_name: String,
30    pub input: Value,
31    #[serde(default)]
32    pub retry: RetryPolicy,
33}
34
35/// A graph proposal submitted through a host-owned Flow boundary.
36#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
37pub struct FlowDecisionRequest {
38    pub decision_id: String,
39    pub run_id: String,
40    pub authority_branch_id: String,
41    pub causation_event_id: String,
42    pub decision: FlowDecision,
43}
44
45#[async_trait]
46pub trait FlowDecisionSink: Send + Sync {
47    /// Submit using `request.decision_id` as the downstream idempotency key.
48    /// Implementations must deduplicate that key because an expired lease can
49    /// be reclaimed after a process crashes between submission and receipt.
50    async fn submit(
51        &self,
52        request: &FlowDecisionRequest,
53    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>>;
54}
55
56#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
57#[serde(rename_all = "snake_case")]
58pub enum FlowDecisionHealthStatus {
59    Healthy,
60    Degraded,
61}
62
63#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
64pub struct FlowDecisionHealthSnapshot {
65    pub status: FlowDecisionHealthStatus,
66    pub attempted: u64,
67    pub claimed: u64,
68    pub takeovers: u64,
69    pub completed: u64,
70    pub duplicates: u64,
71    pub busy: u64,
72    pub conflicts: u64,
73    pub rejected: u64,
74    pub lease_renewals: u64,
75    pub lease_lost: u64,
76    pub sink_failures: u64,
77    pub ledger_failures: u64,
78    pub failures: u64,
79    pub cancellations: u64,
80    pub in_flight: u64,
81    pub average_dispatch_micros: u64,
82    pub max_dispatch_micros: u64,
83    pub last_success_at_ms: Option<u64>,
84    pub last_failure_at_ms: Option<u64>,
85}
86
87#[derive(Default)]
88struct FlowDecisionMetrics {
89    attempted: AtomicU64,
90    claimed: AtomicU64,
91    takeovers: AtomicU64,
92    completed: AtomicU64,
93    duplicates: AtomicU64,
94    busy: AtomicU64,
95    conflicts: AtomicU64,
96    rejected: AtomicU64,
97    lease_renewals: AtomicU64,
98    lease_lost: AtomicU64,
99    sink_failures: AtomicU64,
100    ledger_failures: AtomicU64,
101    failures: AtomicU64,
102    cancellations: AtomicU64,
103    in_flight: AtomicU64,
104    total_dispatch_micros: AtomicU64,
105    max_dispatch_micros: AtomicU64,
106    last_success_at_ms: AtomicU64,
107    last_failure_at_ms: AtomicU64,
108    degraded: AtomicBool,
109}
110
111/// Enforces production-branch authority and idempotent successful submission.
112pub struct FlowDecisionDispatcher {
113    production_branch_id: String,
114    sink: Arc<dyn FlowDecisionSink>,
115    ledger: Arc<dyn FlowDecisionLedger>,
116    owner_id: String,
117    lease_ms: u64,
118    dispatch_lock: Mutex<()>,
119    metrics: Arc<FlowDecisionMetrics>,
120}
121
122impl FlowDecisionDispatcher {
123    pub fn new(production_branch_id: impl Into<String>, sink: Arc<dyn FlowDecisionSink>) -> Self {
124        Self::with_ledger(
125            production_branch_id,
126            sink,
127            Arc::new(MemoryFlowDecisionLedger::new()),
128        )
129    }
130
131    pub fn with_ledger(
132        production_branch_id: impl Into<String>,
133        sink: Arc<dyn FlowDecisionSink>,
134        ledger: Arc<dyn FlowDecisionLedger>,
135    ) -> Self {
136        Self {
137            production_branch_id: production_branch_id.into(),
138            sink,
139            ledger,
140            owner_id: format!("decision-dispatcher-{}", uuid::Uuid::new_v4()),
141            lease_ms: 30_000,
142            dispatch_lock: Mutex::new(()),
143            metrics: Arc::new(FlowDecisionMetrics::default()),
144        }
145    }
146
147    pub fn with_lease_ms(mut self, lease_ms: u64) -> Self {
148        self.lease_ms = lease_ms.max(1);
149        self
150    }
151
152    pub fn health(&self) -> FlowDecisionHealthSnapshot {
153        let attempted = self.metrics.attempted.load(Ordering::Relaxed);
154        let total_micros = self.metrics.total_dispatch_micros.load(Ordering::Relaxed);
155        let last_success_at_ms = nonzero(self.metrics.last_success_at_ms.load(Ordering::Relaxed));
156        let last_failure_at_ms = nonzero(self.metrics.last_failure_at_ms.load(Ordering::Relaxed));
157        FlowDecisionHealthSnapshot {
158            status: if self.metrics.degraded.load(Ordering::Relaxed) {
159                FlowDecisionHealthStatus::Degraded
160            } else {
161                FlowDecisionHealthStatus::Healthy
162            },
163            attempted,
164            claimed: self.metrics.claimed.load(Ordering::Relaxed),
165            takeovers: self.metrics.takeovers.load(Ordering::Relaxed),
166            completed: self.metrics.completed.load(Ordering::Relaxed),
167            duplicates: self.metrics.duplicates.load(Ordering::Relaxed),
168            busy: self.metrics.busy.load(Ordering::Relaxed),
169            conflicts: self.metrics.conflicts.load(Ordering::Relaxed),
170            rejected: self.metrics.rejected.load(Ordering::Relaxed),
171            lease_renewals: self.metrics.lease_renewals.load(Ordering::Relaxed),
172            lease_lost: self.metrics.lease_lost.load(Ordering::Relaxed),
173            sink_failures: self.metrics.sink_failures.load(Ordering::Relaxed),
174            ledger_failures: self.metrics.ledger_failures.load(Ordering::Relaxed),
175            failures: self.metrics.failures.load(Ordering::Relaxed),
176            cancellations: self.metrics.cancellations.load(Ordering::Relaxed),
177            in_flight: self.metrics.in_flight.load(Ordering::Relaxed),
178            average_dispatch_micros: total_micros.checked_div(attempted).unwrap_or(0),
179            max_dispatch_micros: self.metrics.max_dispatch_micros.load(Ordering::Relaxed),
180            last_success_at_ms,
181            last_failure_at_ms,
182        }
183    }
184
185    /// Returns `true` only when the sink accepted a new decision.
186    pub async fn dispatch(
187        &self,
188        request: &FlowDecisionRequest,
189    ) -> Result<bool, FlowDecisionDispatchError> {
190        let mut in_flight = DecisionInFlight::new(Arc::clone(&self.metrics));
191        let result = self.dispatch_inner(request).await;
192        in_flight.finish();
193        self.record_result(&result);
194        result
195    }
196
197    async fn dispatch_inner(
198        &self,
199        request: &FlowDecisionRequest,
200    ) -> Result<bool, FlowDecisionDispatchError> {
201        if request.authority_branch_id != self.production_branch_id {
202            return Err(FlowDecisionDispatchError::UnauthorizedBranch {
203                expected: self.production_branch_id.clone(),
204                actual: request.authority_branch_id.clone(),
205            });
206        }
207        if request.decision_id.trim().is_empty() || request.causation_event_id.trim().is_empty() {
208            return Err(FlowDecisionDispatchError::InvalidIdentity);
209        }
210        let _dispatch_guard = self.dispatch_lock.lock().await;
211        let request_hash = request_hash(request)?;
212        let execution_identity = request_identity(request)?;
213        let claim = ExecutionClaimV1::new(
214            execution_identity,
215            &request.decision_id,
216            &request_hash,
217            &self.owner_id,
218        )
219        .map_err(|error| FlowDecisionDispatchError::Ledger(error.to_string()))?;
220        let result_receipt = decision_result_receipt(&claim, request)?;
221        tracing::trace!(
222            decision_id = request.decision_id.as_str(),
223            identity = claim.identity().key(),
224            "Flow decision execution identity bound to claim ledger"
225        );
226        match self
227            .ledger
228            .claim_with_identity(
229                claim.record_id(),
230                claim.ledger_key(),
231                claim.identity(),
232                claim.owner_id(),
233                now_ms(),
234                self.lease_ms,
235            )
236            .await
237            .map_err(|error| FlowDecisionDispatchError::Ledger(error.to_string()))?
238        {
239            FlowDecisionClaimOutcome::Completed => return Ok(false),
240            FlowDecisionClaimOutcome::Busy {
241                lease_expires_at_ms,
242            } => {
243                return Err(FlowDecisionDispatchError::Busy {
244                    lease_expires_at_ms,
245                })
246            }
247            FlowDecisionClaimOutcome::Conflict => {
248                return Err(FlowDecisionDispatchError::DecisionIdConflict(
249                    request.decision_id.clone(),
250                ))
251            }
252            FlowDecisionClaimOutcome::Claimed { attempt } => {
253                self.metrics.claimed.fetch_add(1, Ordering::Relaxed);
254                if attempt > 1 {
255                    self.metrics.takeovers.fetch_add(1, Ordering::Relaxed);
256                }
257            }
258        }
259        let heartbeat_period = Duration::from_millis((self.lease_ms / 3).max(1));
260        let first_heartbeat = tokio::time::Instant::now() + heartbeat_period;
261        let mut heartbeat = tokio::time::interval_at(first_heartbeat, heartbeat_period);
262        let submission = self.sink.submit(request);
263        tokio::pin!(submission);
264        let submission_result = loop {
265            tokio::select! {
266                result = &mut submission => break result,
267                _ = heartbeat.tick() => {
268                    let renewed = self.ledger
269                        .renew_with_identity(
270                            claim.record_id(),
271                            claim.ledger_key(),
272                            claim.identity(),
273                            claim.owner_id(),
274                            now_ms(),
275                            self.lease_ms,
276                        )
277                        .await
278                        .map_err(|error| FlowDecisionDispatchError::Ledger(error.to_string()))?;
279                    if !renewed {
280                        return Err(FlowDecisionDispatchError::LeaseLost(
281                            request.decision_id.clone(),
282                        ));
283                    }
284                    self.metrics.lease_renewals.fetch_add(1, Ordering::Relaxed);
285                }
286            }
287        };
288        if let Err(error) = submission_result {
289            if let Err(release_error) = self
290                .ledger
291                .release_with_identity(
292                    claim.record_id(),
293                    claim.ledger_key(),
294                    claim.identity(),
295                    claim.owner_id(),
296                )
297                .await
298            {
299                tracing::warn!(decision_id = request.decision_id, error = %release_error, "failed to release Flow decision claim");
300            }
301            return Err(FlowDecisionDispatchError::Sink(error.to_string()));
302        }
303        self.ledger
304            .complete_with_receipt(
305                claim.record_id(),
306                claim.ledger_key(),
307                claim.identity(),
308                claim.owner_id(),
309                &result_receipt,
310                now_ms(),
311            )
312            .await
313            .map_err(|error| FlowDecisionDispatchError::Ledger(error.to_string()))?;
314        Ok(true)
315    }
316
317    fn record_result(&self, result: &Result<bool, FlowDecisionDispatchError>) {
318        match result {
319            Ok(true) => {
320                self.metrics.completed.fetch_add(1, Ordering::Relaxed);
321                self.metrics.degraded.store(false, Ordering::Relaxed);
322                self.metrics
323                    .last_success_at_ms
324                    .store(now_ms(), Ordering::Relaxed);
325            }
326            Ok(false) => {
327                self.metrics.duplicates.fetch_add(1, Ordering::Relaxed);
328                self.metrics.degraded.store(false, Ordering::Relaxed);
329                self.metrics
330                    .last_success_at_ms
331                    .store(now_ms(), Ordering::Relaxed);
332            }
333            Err(FlowDecisionDispatchError::Busy { .. }) => {
334                self.metrics.busy.fetch_add(1, Ordering::Relaxed);
335            }
336            Err(FlowDecisionDispatchError::DecisionIdConflict(_)) => {
337                self.metrics.conflicts.fetch_add(1, Ordering::Relaxed);
338            }
339            Err(
340                FlowDecisionDispatchError::UnauthorizedBranch { .. }
341                | FlowDecisionDispatchError::InvalidIdentity,
342            ) => {
343                self.metrics.rejected.fetch_add(1, Ordering::Relaxed);
344            }
345            Err(FlowDecisionDispatchError::LeaseLost(_)) => {
346                self.metrics.lease_lost.fetch_add(1, Ordering::Relaxed);
347                self.record_failure();
348            }
349            Err(FlowDecisionDispatchError::Ledger(_)) => {
350                self.metrics.ledger_failures.fetch_add(1, Ordering::Relaxed);
351                self.record_failure();
352            }
353            Err(FlowDecisionDispatchError::Sink(_)) => {
354                self.metrics.sink_failures.fetch_add(1, Ordering::Relaxed);
355                self.record_failure();
356            }
357        }
358    }
359
360    fn record_failure(&self) {
361        self.metrics.failures.fetch_add(1, Ordering::Relaxed);
362        self.metrics.degraded.store(true, Ordering::Relaxed);
363        self.metrics
364            .last_failure_at_ms
365            .store(now_ms(), Ordering::Relaxed);
366    }
367}
368
369struct DecisionInFlight {
370    metrics: Arc<FlowDecisionMetrics>,
371    started: Instant,
372    completed: bool,
373}
374
375impl DecisionInFlight {
376    fn new(metrics: Arc<FlowDecisionMetrics>) -> Self {
377        metrics.attempted.fetch_add(1, Ordering::Relaxed);
378        metrics.in_flight.fetch_add(1, Ordering::Relaxed);
379        Self {
380            metrics,
381            started: Instant::now(),
382            completed: false,
383        }
384    }
385
386    fn finish(&mut self) {
387        self.record_elapsed();
388        self.metrics.in_flight.fetch_sub(1, Ordering::Relaxed);
389        self.completed = true;
390    }
391
392    fn record_elapsed(&self) {
393        let elapsed = self.started.elapsed().as_micros().min(u128::from(u64::MAX)) as u64;
394        self.metrics
395            .total_dispatch_micros
396            .fetch_add(elapsed, Ordering::Relaxed);
397        self.metrics
398            .max_dispatch_micros
399            .fetch_max(elapsed, Ordering::Relaxed);
400    }
401}
402
403impl Drop for DecisionInFlight {
404    fn drop(&mut self) {
405        if self.completed {
406            return;
407        }
408        self.record_elapsed();
409        self.metrics.in_flight.fetch_sub(1, Ordering::Relaxed);
410        self.metrics.failures.fetch_add(1, Ordering::Relaxed);
411        self.metrics.cancellations.fetch_add(1, Ordering::Relaxed);
412        self.metrics.degraded.store(true, Ordering::Relaxed);
413        self.metrics
414            .last_failure_at_ms
415            .store(now_ms(), Ordering::Relaxed);
416    }
417}
418
419#[derive(Debug, Error, PartialEq, Eq)]
420pub enum FlowDecisionDispatchError {
421    #[error("graph branch is not authorized for production Flow decisions: expected `{expected}`, got `{actual}`")]
422    UnauthorizedBranch { expected: String, actual: String },
423    #[error("decision_id and causation_event_id must be non-empty")]
424    InvalidIdentity,
425    #[error("Flow decision `{0}` reuses an existing decision id with different content")]
426    DecisionIdConflict(String),
427    #[error("Flow decision is owned by another dispatcher until {lease_expires_at_ms}")]
428    Busy { lease_expires_at_ms: u64 },
429    #[error("Flow decision `{0}` lost its lease while the sink was running")]
430    LeaseLost(String),
431    #[error("Flow decision ledger failed: {0}")]
432    Ledger(String),
433    #[error("Flow decision sink failed: {0}")]
434    Sink(String),
435}
436
437fn request_hash(request: &FlowDecisionRequest) -> Result<String, FlowDecisionDispatchError> {
438    serde_json::to_vec(request)
439        .map(sha256::digest)
440        .map_err(|error| FlowDecisionDispatchError::Ledger(error.to_string()))
441}
442
443fn request_identity(
444    request: &FlowDecisionRequest,
445) -> Result<crate::execution_identity::ExecutionIdentityV1, FlowDecisionDispatchError> {
446    crate::execution_identity::ExecutionIdentityV1::derive(
447        crate::execution_identity::FLOW_DECISION_IDENTITY_DOMAIN_V1,
448        request,
449    )
450    .map_err(|error| FlowDecisionDispatchError::Ledger(error.to_string()))
451}
452
453fn decision_result_receipt(
454    claim: &ExecutionClaimV1,
455    request: &FlowDecisionRequest,
456) -> Result<ExecutionResultReceiptV1, FlowDecisionDispatchError> {
457    let result = serde_json::to_vec(&request.decision)
458        .map_err(|error| FlowDecisionDispatchError::Ledger(error.to_string()))?;
459    let evidence_digest = crate::evaluation::digest_bytes(
460        "a3s.code.flow-decision.evidence.v1",
461        request.causation_event_id.as_bytes(),
462    );
463    let result_digest =
464        crate::evaluation::digest_bytes("a3s.code.flow-decision.result.v1", &result);
465    let result_bytes = u64::try_from(result.len())
466        .map_err(|_| FlowDecisionDispatchError::Ledger("decision result is too large".into()))?;
467    claim
468        .result_receipt(
469            evidence_digest,
470            ExecutionResultOutcomeV1::Succeeded,
471            Some(result_digest),
472            result_bytes,
473        )
474        .map_err(|error| FlowDecisionDispatchError::Ledger(error.to_string()))
475}
476
477fn now_ms() -> u64 {
478    use std::time::{SystemTime, UNIX_EPOCH};
479    SystemTime::now()
480        .duration_since(UNIX_EPOCH)
481        .unwrap_or_default()
482        .as_millis()
483        .min(u128::from(u64::MAX)) as u64
484}
485
486fn nonzero(value: u64) -> Option<u64> {
487    (value != 0).then_some(value)
488}
489
490#[cfg(test)]
491mod tests {
492    use super::*;
493    use crate::{FileFlowDecisionLedger, FlowDecisionLedger};
494    use std::sync::atomic::{AtomicUsize, Ordering};
495
496    #[derive(Default)]
497    struct RecordingSink(Mutex<Vec<String>>);
498
499    #[async_trait]
500    impl FlowDecisionSink for RecordingSink {
501        async fn submit(
502            &self,
503            request: &FlowDecisionRequest,
504        ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
505            self.0.lock().await.push(request.decision_id.clone());
506            Ok(())
507        }
508    }
509
510    fn request(branch: &str) -> FlowDecisionRequest {
511        FlowDecisionRequest {
512            decision_id: "decision-1".into(),
513            run_id: "run-1".into(),
514            authority_branch_id: branch.into(),
515            causation_event_id: "event-1".into(),
516            decision: FlowDecision::Complete {
517                output: Value::Null,
518            },
519        }
520    }
521
522    #[test]
523    fn flow_claim_keeps_legacy_ledger_key_while_binding_typed_identity() {
524        let request = request("production");
525        let legacy = request_hash(&request).unwrap();
526        let identity = request_identity(&request).unwrap();
527        let claim = crate::execution_identity::ExecutionClaimV1::new(
528            identity.clone(),
529            &request.decision_id,
530            &legacy,
531            "owner-1",
532        )
533        .unwrap();
534
535        assert_eq!(legacy.len(), 64);
536        assert!(!legacy.starts_with("sha256:"));
537        assert_eq!(claim.ledger_key(), legacy);
538        assert_ne!(claim.identity().key(), legacy);
539        claim.identity().validate_for(&request).unwrap();
540    }
541
542    #[tokio::test]
543    async fn submits_once_and_rejects_fork_branches() {
544        let sink = Arc::new(RecordingSink::default());
545        let ledger = Arc::new(MemoryFlowDecisionLedger::new());
546        let dispatcher =
547            FlowDecisionDispatcher::with_ledger("production", sink.clone(), ledger.clone());
548        let accepted = request("production");
549        assert!(dispatcher.dispatch(&accepted).await.unwrap());
550        assert!(!dispatcher.dispatch(&request("production")).await.unwrap());
551        assert_eq!(sink.0.lock().await.as_slice(), ["decision-1"]);
552        let receipt = ledger
553            .completed_receipt("decision-1")
554            .await
555            .unwrap()
556            .expect("accepted decision stores a terminal result receipt");
557        assert_eq!(receipt.identity, request_identity(&accepted).unwrap());
558        receipt.validate().unwrap();
559        assert!(matches!(
560            dispatcher.dispatch(&request("fork")).await,
561            Err(FlowDecisionDispatchError::UnauthorizedBranch { .. })
562        ));
563        let health = dispatcher.health();
564        assert_eq!(health.status, FlowDecisionHealthStatus::Healthy);
565        assert_eq!(health.attempted, 3);
566        assert_eq!(health.claimed, 1);
567        assert_eq!(health.completed, 1);
568        assert_eq!(health.duplicates, 1);
569        assert_eq!(health.rejected, 1);
570        assert_eq!(health.in_flight, 0);
571        assert!(health.last_success_at_ms.is_some());
572    }
573
574    #[tokio::test]
575    async fn completed_receipt_survives_dispatcher_restart() {
576        let sink = Arc::new(RecordingSink::default());
577        let ledger = Arc::new(MemoryFlowDecisionLedger::new());
578        let first = FlowDecisionDispatcher::with_ledger("production", sink.clone(), ledger.clone());
579        assert!(first.dispatch(&request("production")).await.unwrap());
580        let restarted = FlowDecisionDispatcher::with_ledger("production", sink.clone(), ledger);
581        assert!(!restarted.dispatch(&request("production")).await.unwrap());
582        assert_eq!(sink.0.lock().await.len(), 1);
583    }
584
585    #[tokio::test]
586    async fn decision_id_cannot_be_reused_with_different_content() {
587        let sink = Arc::new(RecordingSink::default());
588        let ledger = Arc::new(MemoryFlowDecisionLedger::new());
589        let dispatcher = FlowDecisionDispatcher::with_ledger("production", sink, ledger);
590        dispatcher.dispatch(&request("production")).await.unwrap();
591        let mut conflicting = request("production");
592        conflicting.decision = FlowDecision::Fail {
593            error: "different".into(),
594        };
595        assert!(matches!(
596            dispatcher.dispatch(&conflicting).await,
597            Err(FlowDecisionDispatchError::DecisionIdConflict(_))
598        ));
599        assert_eq!(dispatcher.health().conflicts, 1);
600    }
601
602    struct FailingOnceSink(AtomicUsize);
603
604    #[async_trait]
605    impl FlowDecisionSink for FailingOnceSink {
606        async fn submit(
607            &self,
608            _request: &FlowDecisionRequest,
609        ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
610            if self.0.fetch_add(1, Ordering::SeqCst) == 0 {
611                return Err(std::io::Error::other("transient").into());
612            }
613            Ok(())
614        }
615    }
616
617    #[tokio::test]
618    async fn sink_failure_releases_claim_for_retry() {
619        let sink = Arc::new(FailingOnceSink(AtomicUsize::new(0)));
620        let dispatcher = FlowDecisionDispatcher::with_ledger(
621            "production",
622            sink.clone(),
623            Arc::new(MemoryFlowDecisionLedger::new()),
624        );
625        assert!(matches!(
626            dispatcher.dispatch(&request("production")).await,
627            Err(FlowDecisionDispatchError::Sink(_))
628        ));
629        assert!(dispatcher.dispatch(&request("production")).await.unwrap());
630        assert_eq!(sink.0.load(Ordering::SeqCst), 2);
631        let health = dispatcher.health();
632        assert_eq!(health.status, FlowDecisionHealthStatus::Healthy);
633        assert_eq!(health.sink_failures, 1);
634        assert_eq!(health.failures, 1);
635        assert_eq!(health.completed, 1);
636    }
637
638    #[tokio::test]
639    async fn independent_file_ledgers_serialize_claim_and_allow_expired_takeover() {
640        let directory = tempfile::tempdir().unwrap();
641        let left = FileFlowDecisionLedger::new(directory.path());
642        let right = FileFlowDecisionLedger::new(directory.path());
643        let (left_claim, right_claim) = tokio::join!(
644            left.claim("decision", "hash", "left", 100, 50),
645            right.claim("decision", "hash", "right", 100, 50),
646        );
647        let claims = [left_claim.unwrap(), right_claim.unwrap()];
648        assert_eq!(
649            claims
650                .iter()
651                .filter(|claim| matches!(claim, FlowDecisionClaimOutcome::Claimed { .. }))
652                .count(),
653            1
654        );
655        assert_eq!(
656            claims
657                .iter()
658                .filter(|claim| matches!(claim, FlowDecisionClaimOutcome::Busy { .. }))
659                .count(),
660            1
661        );
662        assert_eq!(
663            right
664                .claim("decision", "hash", "takeover", 151, 50)
665                .await
666                .unwrap(),
667            FlowDecisionClaimOutcome::Claimed { attempt: 2 }
668        );
669        right
670            .complete("decision", "hash", "takeover", 160)
671            .await
672            .unwrap();
673        assert_eq!(right.prune_completed(161).await.unwrap(), 1);
674        assert_eq!(
675            left.claim("decision", "hash", "new", 162, 50)
676                .await
677                .unwrap(),
678            FlowDecisionClaimOutcome::Claimed { attempt: 1 }
679        );
680    }
681
682    #[tokio::test]
683    async fn lease_renewal_requires_current_owner_and_preserves_attempt() {
684        let ledger = MemoryFlowDecisionLedger::new();
685        assert_eq!(
686            ledger
687                .claim("decision", "hash", "owner", 100, 30)
688                .await
689                .unwrap(),
690            FlowDecisionClaimOutcome::Claimed { attempt: 1 }
691        );
692        assert!(!ledger
693            .renew("decision", "hash", "other", 120, 30)
694            .await
695            .unwrap());
696        assert!(ledger
697            .renew("decision", "hash", "owner", 120, 30)
698            .await
699            .unwrap());
700        assert_eq!(
701            ledger
702                .claim("decision", "hash", "other", 145, 30)
703                .await
704                .unwrap(),
705            FlowDecisionClaimOutcome::Busy {
706                lease_expires_at_ms: 150
707            }
708        );
709        assert_eq!(
710            ledger
711                .claim("decision", "hash", "other", 151, 30)
712                .await
713                .unwrap(),
714            FlowDecisionClaimOutcome::Claimed { attempt: 2 }
715        );
716        assert!(!ledger
717            .renew("decision", "hash", "owner", 152, 30)
718            .await
719            .unwrap());
720    }
721
722    #[tokio::test]
723    async fn file_ledger_persists_renewed_lease_across_instances() {
724        let directory = tempfile::tempdir().unwrap();
725        let owner = FileFlowDecisionLedger::new(directory.path());
726        let competitor = FileFlowDecisionLedger::new(directory.path());
727        assert_eq!(
728            owner
729                .claim("decision", "hash", "owner", 100, 20)
730                .await
731                .unwrap(),
732            FlowDecisionClaimOutcome::Claimed { attempt: 1 }
733        );
734        assert!(owner
735            .renew("decision", "hash", "owner", 115, 30)
736            .await
737            .unwrap());
738        assert_eq!(
739            competitor
740                .claim("decision", "hash", "competitor", 125, 20)
741                .await
742                .unwrap(),
743            FlowDecisionClaimOutcome::Busy {
744                lease_expires_at_ms: 145
745            }
746        );
747    }
748
749    struct SlowCountingSink(AtomicUsize);
750
751    #[async_trait]
752    impl FlowDecisionSink for SlowCountingSink {
753        async fn submit(
754            &self,
755            _request: &FlowDecisionRequest,
756        ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
757            self.0.fetch_add(1, Ordering::SeqCst);
758            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
759            Ok(())
760        }
761    }
762
763    #[tokio::test]
764    async fn competing_file_backed_dispatchers_submit_to_sink_once() {
765        let directory = tempfile::tempdir().unwrap();
766        let sink = Arc::new(SlowCountingSink(AtomicUsize::new(0)));
767        let left = FlowDecisionDispatcher::with_ledger(
768            "production",
769            sink.clone(),
770            Arc::new(FileFlowDecisionLedger::new(directory.path())),
771        );
772        let right = FlowDecisionDispatcher::with_ledger(
773            "production",
774            sink.clone(),
775            Arc::new(FileFlowDecisionLedger::new(directory.path())),
776        );
777        let request = request("production");
778        let (left_result, right_result) =
779            tokio::join!(left.dispatch(&request), right.dispatch(&request));
780        let results = [left_result, right_result];
781        assert_eq!(
782            results
783                .iter()
784                .filter(|result| matches!(result, Ok(true)))
785                .count(),
786            1
787        );
788        assert_eq!(
789            results
790                .iter()
791                .filter(|result| matches!(result, Err(FlowDecisionDispatchError::Busy { .. })))
792                .count(),
793            1
794        );
795        assert_eq!(sink.0.load(Ordering::SeqCst), 1);
796    }
797
798    struct RenewalNotifyingLedger {
799        inner: MemoryFlowDecisionLedger,
800        renewed: Arc<tokio::sync::Notify>,
801    }
802
803    #[async_trait]
804    impl FlowDecisionLedger for RenewalNotifyingLedger {
805        async fn claim(
806            &self,
807            decision_id: &str,
808            request_hash: &str,
809            owner_id: &str,
810            now_ms: u64,
811            lease_ms: u64,
812        ) -> anyhow::Result<FlowDecisionClaimOutcome> {
813            self.inner
814                .claim(decision_id, request_hash, owner_id, now_ms, lease_ms)
815                .await
816        }
817
818        async fn renew(
819            &self,
820            decision_id: &str,
821            request_hash: &str,
822            owner_id: &str,
823            now_ms: u64,
824            lease_ms: u64,
825        ) -> anyhow::Result<bool> {
826            let renewed = self
827                .inner
828                .renew(decision_id, request_hash, owner_id, now_ms, lease_ms)
829                .await?;
830            if renewed {
831                self.renewed.notify_one();
832            }
833            Ok(renewed)
834        }
835
836        async fn complete(
837            &self,
838            decision_id: &str,
839            request_hash: &str,
840            owner_id: &str,
841            completed_at_ms: u64,
842        ) -> anyhow::Result<()> {
843            self.inner
844                .complete(decision_id, request_hash, owner_id, completed_at_ms)
845                .await
846        }
847
848        async fn release(
849            &self,
850            decision_id: &str,
851            request_hash: &str,
852            owner_id: &str,
853        ) -> anyhow::Result<()> {
854            self.inner
855                .release(decision_id, request_hash, owner_id)
856                .await
857        }
858    }
859
860    struct ControlledSink {
861        calls: AtomicUsize,
862        finish: Arc<tokio::sync::Notify>,
863    }
864
865    #[async_trait]
866    impl FlowDecisionSink for ControlledSink {
867        async fn submit(
868            &self,
869            _request: &FlowDecisionRequest,
870        ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
871            self.calls.fetch_add(1, Ordering::SeqCst);
872            self.finish.notified().await;
873            Ok(())
874        }
875    }
876
877    #[tokio::test]
878    async fn heartbeat_prevents_takeover_while_sink_is_running() {
879        let renewed = Arc::new(tokio::sync::Notify::new());
880        let finish = Arc::new(tokio::sync::Notify::new());
881        let sink = Arc::new(ControlledSink {
882            calls: AtomicUsize::new(0),
883            finish: finish.clone(),
884        });
885        let ledger = Arc::new(RenewalNotifyingLedger {
886            inner: MemoryFlowDecisionLedger::new(),
887            renewed: renewed.clone(),
888        });
889        let left = Arc::new(
890            FlowDecisionDispatcher::with_ledger("production", sink.clone(), ledger.clone())
891                .with_lease_ms(30),
892        );
893        let right = FlowDecisionDispatcher::with_ledger("production", sink.clone(), ledger)
894            .with_lease_ms(30);
895        let request = request("production");
896        let left_dispatcher = left.clone();
897        let left_request = request.clone();
898        let left_task = tokio::spawn(async move { left_dispatcher.dispatch(&left_request).await });
899        tokio::time::timeout(std::time::Duration::from_secs(2), renewed.notified())
900            .await
901            .expect("dispatcher must renew its live claim");
902        let right_result = right.dispatch(&request).await;
903        assert!(matches!(
904            right_result,
905            Err(FlowDecisionDispatchError::Busy { .. })
906        ));
907        finish.notify_one();
908        assert!(left_task.await.unwrap().unwrap());
909        assert_eq!(sink.calls.load(Ordering::SeqCst), 1);
910        let left_health = left.health();
911        assert_eq!(left_health.completed, 1);
912        assert!(left_health.lease_renewals > 0);
913        assert_eq!(right.health().busy, 1);
914    }
915
916    #[tokio::test]
917    async fn dispatcher_health_counts_expired_claim_takeover() {
918        let ledger = Arc::new(MemoryFlowDecisionLedger::new());
919        let request = request("production");
920        let hash = request_hash(&request).unwrap();
921        ledger
922            .claim(
923                &request.decision_id,
924                &hash,
925                "expired-owner",
926                now_ms().saturating_sub(10),
927                1,
928            )
929            .await
930            .unwrap();
931        let dispatcher = FlowDecisionDispatcher::with_ledger(
932            "production",
933            Arc::new(RecordingSink::default()),
934            ledger,
935        );
936        assert!(dispatcher.dispatch(&request).await.unwrap());
937        let health = dispatcher.health();
938        assert_eq!(health.claimed, 1);
939        assert_eq!(health.takeovers, 1);
940        assert_eq!(health.completed, 1);
941    }
942
943    #[derive(Default)]
944    struct LeaseLosingLedger;
945
946    #[async_trait]
947    impl FlowDecisionLedger for LeaseLosingLedger {
948        async fn claim(
949            &self,
950            _decision_id: &str,
951            _request_hash: &str,
952            _owner_id: &str,
953            _now_ms: u64,
954            _lease_ms: u64,
955        ) -> anyhow::Result<FlowDecisionClaimOutcome> {
956            Ok(FlowDecisionClaimOutcome::Claimed { attempt: 1 })
957        }
958
959        async fn renew(
960            &self,
961            _decision_id: &str,
962            _request_hash: &str,
963            _owner_id: &str,
964            _now_ms: u64,
965            _lease_ms: u64,
966        ) -> anyhow::Result<bool> {
967            Ok(false)
968        }
969
970        async fn complete(
971            &self,
972            _decision_id: &str,
973            _request_hash: &str,
974            _owner_id: &str,
975            _completed_at_ms: u64,
976        ) -> anyhow::Result<()> {
977            Ok(())
978        }
979
980        async fn release(
981            &self,
982            _decision_id: &str,
983            _request_hash: &str,
984            _owner_id: &str,
985        ) -> anyhow::Result<()> {
986            Ok(())
987        }
988    }
989
990    #[derive(Default)]
991    struct ClaimFailingLedger;
992
993    #[async_trait]
994    impl FlowDecisionLedger for ClaimFailingLedger {
995        async fn claim(
996            &self,
997            _decision_id: &str,
998            _request_hash: &str,
999            _owner_id: &str,
1000            _now_ms: u64,
1001            _lease_ms: u64,
1002        ) -> anyhow::Result<FlowDecisionClaimOutcome> {
1003            anyhow::bail!("ledger unavailable")
1004        }
1005
1006        async fn renew(
1007            &self,
1008            _decision_id: &str,
1009            _request_hash: &str,
1010            _owner_id: &str,
1011            _now_ms: u64,
1012            _lease_ms: u64,
1013        ) -> anyhow::Result<bool> {
1014            Ok(false)
1015        }
1016
1017        async fn complete(
1018            &self,
1019            _decision_id: &str,
1020            _request_hash: &str,
1021            _owner_id: &str,
1022            _completed_at_ms: u64,
1023        ) -> anyhow::Result<()> {
1024            Ok(())
1025        }
1026
1027        async fn release(
1028            &self,
1029            _decision_id: &str,
1030            _request_hash: &str,
1031            _owner_id: &str,
1032        ) -> anyhow::Result<()> {
1033            Ok(())
1034        }
1035    }
1036
1037    #[tokio::test]
1038    async fn dispatcher_health_records_ledger_failure() {
1039        let dispatcher = FlowDecisionDispatcher::with_ledger(
1040            "production",
1041            Arc::new(RecordingSink::default()),
1042            Arc::new(ClaimFailingLedger),
1043        );
1044        assert!(matches!(
1045            dispatcher.dispatch(&request("production")).await,
1046            Err(FlowDecisionDispatchError::Ledger(_))
1047        ));
1048        let health = dispatcher.health();
1049        assert_eq!(health.status, FlowDecisionHealthStatus::Degraded);
1050        assert_eq!(health.ledger_failures, 1);
1051        assert_eq!(health.in_flight, 0);
1052    }
1053
1054    struct CancellableSink {
1055        started: AtomicUsize,
1056        completed: AtomicUsize,
1057    }
1058
1059    #[async_trait]
1060    impl FlowDecisionSink for CancellableSink {
1061        async fn submit(
1062            &self,
1063            _request: &FlowDecisionRequest,
1064        ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
1065            self.started.fetch_add(1, Ordering::SeqCst);
1066            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1067            self.completed.fetch_add(1, Ordering::SeqCst);
1068            Ok(())
1069        }
1070    }
1071
1072    #[tokio::test]
1073    async fn lost_lease_cancels_in_flight_sink_future() {
1074        let sink = Arc::new(CancellableSink {
1075            started: AtomicUsize::new(0),
1076            completed: AtomicUsize::new(0),
1077        });
1078        let dispatcher = FlowDecisionDispatcher::with_ledger(
1079            "production",
1080            sink.clone(),
1081            Arc::new(LeaseLosingLedger),
1082        )
1083        .with_lease_ms(9);
1084        assert!(matches!(
1085            dispatcher.dispatch(&request("production")).await,
1086            Err(FlowDecisionDispatchError::LeaseLost(_))
1087        ));
1088        tokio::time::sleep(std::time::Duration::from_millis(60)).await;
1089        assert_eq!(sink.started.load(Ordering::SeqCst), 1);
1090        assert_eq!(sink.completed.load(Ordering::SeqCst), 0);
1091        let health = dispatcher.health();
1092        assert_eq!(health.status, FlowDecisionHealthStatus::Degraded);
1093        assert_eq!(health.lease_lost, 1);
1094        assert_eq!(health.in_flight, 0);
1095        assert!(health.last_failure_at_ms.is_some());
1096    }
1097
1098    struct BlockingSink(Arc<tokio::sync::Notify>);
1099
1100    #[async_trait]
1101    impl FlowDecisionSink for BlockingSink {
1102        async fn submit(
1103            &self,
1104            _request: &FlowDecisionRequest,
1105        ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
1106            self.0.notify_one();
1107            std::future::pending().await
1108        }
1109    }
1110
1111    #[tokio::test]
1112    async fn cancelled_dispatch_does_not_leak_in_flight_health() {
1113        let started = Arc::new(tokio::sync::Notify::new());
1114        let dispatcher = Arc::new(FlowDecisionDispatcher::new(
1115            "production",
1116            Arc::new(BlockingSink(started.clone())),
1117        ));
1118        let task_dispatcher = dispatcher.clone();
1119        let pending =
1120            tokio::spawn(async move { task_dispatcher.dispatch(&request("production")).await });
1121        started.notified().await;
1122        assert_eq!(dispatcher.health().in_flight, 1);
1123        pending.abort();
1124        assert!(pending.await.unwrap_err().is_cancelled());
1125        let health = dispatcher.health();
1126        assert_eq!(health.status, FlowDecisionHealthStatus::Degraded);
1127        assert_eq!(health.attempted, 1);
1128        assert_eq!(health.cancellations, 1);
1129        assert_eq!(health.in_flight, 0);
1130    }
1131}
a3s_code_core/flow_graph/decision.rs

a3s_code_core/flow_graph/
decision.rs
