krabka_client_streams/runtime/iqv2/
result.rs

1//! `IQv2` results. Each partition gives a `QueryResult<R>`, and the code
2//! aggregates them into a `StateQueryResult<R>`.
3
4use std::collections::BTreeMap;
5
6use super::request::Position;
7
8/// Why a partition's query gave no result. This mirrors the subset of the JVM
9/// `FailureReason` that krabka can produce locally.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub enum FailureReason {
12    /// The store kind does not support this query variant.
13    UnknownQueryType,
14    /// The partition's `Position` did not meet the requested bound.
15    NotUpToBound,
16    /// The store exists in the topology but not on this partition's task.
17    NotPresent,
18    /// The partition is a standby or is restoring, and the caller asked an
19    /// active-only query.
20    NotActive,
21    /// The store name is not in the topology.
22    DoesNotExist,
23    /// An internal failure, such as a result-type or key-type mismatch across
24    /// the boundary.
25    StoreException,
26}
27
28/// One partition's outcome.
29pub enum QueryResult<R> {
30    Success {
31        result: R,
32        position: Position,
33    },
34    Failure {
35        reason: FailureReason,
36        message: String,
37    },
38}
39
40impl<R> QueryResult<R> {
41    #[must_use]
42    pub fn is_success(&self) -> bool {
43        matches!(self, QueryResult::Success { .. })
44    }
45    #[must_use]
46    pub fn result(&self) -> Option<&R> {
47        match self {
48            QueryResult::Success { result, .. } => Some(result),
49            QueryResult::Failure { .. } => None,
50        }
51    }
52    #[must_use]
53    pub fn into_result(self) -> Option<R> {
54        match self {
55            QueryResult::Success { result, .. } => Some(result),
56            QueryResult::Failure { .. } => None,
57        }
58    }
59    #[must_use]
60    pub fn position(&self) -> Option<&Position> {
61        match self {
62            QueryResult::Success { position, .. } => Some(position),
63            QueryResult::Failure { .. } => None,
64        }
65    }
66    #[must_use]
67    pub fn failure_reason(&self) -> Option<FailureReason> {
68        match self {
69            QueryResult::Failure { reason, .. } => Some(*reason),
70            QueryResult::Success { .. } => None,
71        }
72    }
73    #[must_use]
74    pub fn failure_message(&self) -> Option<&str> {
75        match self {
76            QueryResult::Failure { message, .. } => Some(message),
77            QueryResult::Success { .. } => None,
78        }
79    }
80}
81
82/// All local partitions' outcomes for one query.
83pub struct StateQueryResult<R> {
84    partition_results: BTreeMap<i32, QueryResult<R>>,
85}
86
87impl<R> StateQueryResult<R> {
88    #[must_use]
89    pub(crate) fn new(partition_results: BTreeMap<i32, QueryResult<R>>) -> Self {
90        Self { partition_results }
91    }
92    #[must_use]
93    pub fn partition_results(&self) -> &BTreeMap<i32, QueryResult<R>> {
94        &self.partition_results
95    }
96    /// The single partition's result, but only when exactly one partition
97    /// responded.
98    #[must_use]
99    pub fn only_partition_result(&self) -> Option<&QueryResult<R>> {
100        if self.partition_results.len() == 1 {
101            self.partition_results.values().next()
102        } else {
103            None
104        }
105    }
106    /// The global-store result. It is always `None` in slice 3a, which does not
107    /// cover global stores.
108    #[must_use]
109    pub fn global_result(&self) -> Option<&QueryResult<R>> {
110        None
111    }
112}
113
114#[cfg(test)]
115mod tests {
116    use super::*;
117
118    #[test]
119    fn success_accessors_expose_result_and_position() {
120        let qr: QueryResult<i64> = QueryResult::Success {
121            result: 42,
122            position: Position::default(),
123        };
124        assert!(qr.is_success());
125        assert_eq!(qr.result(), Some(&42));
126        assert_eq!(qr.position(), Some(&Position::default()));
127        assert_eq!(qr.failure_reason(), None);
128        assert_eq!(qr.failure_message(), None);
129        assert_eq!(qr.into_result(), Some(42));
130    }
131
132    #[test]
133    fn failure_accessors_expose_reason_and_message() {
134        let qr: QueryResult<i64> = QueryResult::Failure {
135            reason: FailureReason::NotPresent,
136            message: "store not on this task".to_string(),
137        };
138        assert!(!qr.is_success());
139        assert_eq!(qr.result(), None);
140        assert_eq!(qr.position(), None);
141        assert_eq!(qr.failure_reason(), Some(FailureReason::NotPresent));
142        assert_eq!(qr.failure_message(), Some("store not on this task"));
143        assert_eq!(qr.into_result(), None);
144    }
145
146    #[test]
147    fn state_query_result_multi_partition_has_no_only_or_global() {
148        let mut map: BTreeMap<i32, QueryResult<Option<i64>>> = BTreeMap::new();
149        map.insert(
150            0,
151            QueryResult::Success {
152                result: Some(1),
153                position: Position::default(),
154            },
155        );
156        map.insert(
157            1,
158            QueryResult::Success {
159                result: Some(2),
160                position: Position::default(),
161            },
162        );
163        let sqr = StateQueryResult::new(map);
164        assert_eq!(sqr.partition_results().len(), 2);
165        assert!(sqr.only_partition_result().is_none());
166        assert!(sqr.global_result().is_none());
167    }
168
169    #[test]
170    fn state_query_result_single_partition_has_only() {
171        let mut map: BTreeMap<i32, QueryResult<Option<i64>>> = BTreeMap::new();
172        map.insert(
173            3,
174            QueryResult::Success {
175                result: Some(9),
176                position: Position::default(),
177            },
178        );
179        let sqr = StateQueryResult::new(map);
180        assert_eq!(sqr.partition_results().len(), 1);
181        let only = sqr.only_partition_result().expect("exactly one partition");
182        assert_eq!(only.result(), Some(&Some(9)));
183    }
184}
krabka_client_streams/runtime/iqv2/result.rs

krabka_client_streams/runtime/iqv2/
result.rs
