alien_worker_protocol/
batch.rs

1//! Validation shared by batch transports and application hosts.
2
3use std::collections::{HashMap, HashSet};
4
5use crate::control::{queue_delivery_outcome::Disposition, QueueBatch, QueueBatchResult};
6
7/// Validate every outcome before making any acknowledgment decision.
8/// Returns dispositions in delivery order, regardless of result order.
9/// Invalid input or results cause the entire batch to remain unacknowledged.
10pub fn validate_outcomes(
11    batch: &QueueBatch,
12    result: &QueueBatchResult,
13) -> Option<Vec<Disposition>> {
14    if batch.deliveries.is_empty() || batch.deliveries.len() != result.outcomes.len() {
15        return None;
16    }
17    let mut identities = HashSet::with_capacity(batch.deliveries.len());
18    for delivery in &batch.deliveries {
19        if delivery.delivery_id.is_empty()
20            || delivery.message.is_none()
21            || !identities.insert(delivery.delivery_id.as_str())
22        {
23            return None;
24        }
25    }
26    let mut outcomes = HashMap::with_capacity(result.outcomes.len());
27    for outcome in &result.outcomes {
28        let disposition = Disposition::try_from(outcome.disposition).ok()?;
29        if disposition == Disposition::Unspecified
30            || !identities.contains(outcome.delivery_id.as_str())
31            || outcomes
32                .insert(outcome.delivery_id.as_str(), disposition)
33                .is_some()
34        {
35            return None;
36        }
37    }
38    batch
39        .deliveries
40        .iter()
41        .map(|delivery| outcomes.get(delivery.delivery_id.as_str()).copied())
42        .collect()
43}
44
45#[cfg(test)]
46mod tests {
47    use super::*;
48    use crate::control::{QueueDelivery, QueueDeliveryOutcome, QueueMessage};
49
50    fn batch() -> QueueBatch {
51        QueueBatch {
52            source: "jobs".into(),
53            deliveries: ["attempt-a", "attempt-b"]
54                .into_iter()
55                .map(|id| QueueDelivery {
56                    delivery_id: id.into(),
57                    message: Some(QueueMessage {
58                        id: "same-provider-id".into(),
59                        ..Default::default()
60                    }),
61                })
62                .collect(),
63        }
64    }
65
66    fn result() -> QueueBatchResult {
67        QueueBatchResult {
68            outcomes: vec![
69                QueueDeliveryOutcome {
70                    delivery_id: "attempt-b".into(),
71                    disposition: Disposition::Retry as i32,
72                },
73                QueueDeliveryOutcome {
74                    delivery_id: "attempt-a".into(),
75                    disposition: Disposition::Success as i32,
76                },
77            ],
78        }
79    }
80
81    #[test]
82    fn correlates_attempts_independently_of_provider_ids_and_result_order() {
83        assert_eq!(
84            validate_outcomes(&batch(), &result()),
85            Some(vec![Disposition::Success, Disposition::Retry])
86        );
87    }
88
89    #[test]
90    fn rejects_missing_duplicate_unknown_and_unspecified_outcomes() {
91        let mut missing = result();
92        missing.outcomes.pop();
93        assert_eq!(validate_outcomes(&batch(), &missing), None);
94        let mut duplicate = result();
95        duplicate.outcomes[0].delivery_id = "attempt-a".into();
96        assert_eq!(validate_outcomes(&batch(), &duplicate), None);
97        let mut unknown = result();
98        unknown.outcomes[0].delivery_id = "attempt-c".into();
99        assert_eq!(validate_outcomes(&batch(), &unknown), None);
100        for invalid in [0, 99] {
101            let mut unspecified = result();
102            unspecified.outcomes[0].disposition = invalid;
103            assert_eq!(validate_outcomes(&batch(), &unspecified), None);
104        }
105        let mut ambiguous = batch();
106        ambiguous.deliveries[0].delivery_id = "attempt-b".into();
107        assert_eq!(validate_outcomes(&ambiguous, &result()), None);
108    }
109}
alien_worker_protocol/batch.rs

alien_worker_protocol/
batch.rs
