ledgence_orchestration_api/
discovery.rs

1//! Bounded discovery of committed task metadata using live keyset pagination.
2
3use crate::*;
4use serde::Deserializer;
5
6pub const TASK_LIST_DEFAULT_LIMIT: u32 = 50;
7pub const TASK_LIST_MAX_LIMIT: u32 = 100;
8pub const TASK_CURSOR_MAX_BYTES: usize = 8192;
9pub const TASK_PAGE_MAX_BYTES: usize = 2 * 1024 * 1024;
10const MAX_TIMESTAMP: Timestamp = 253_402_300_799_999;
11
12/// Exact metadata filters. Submission bounds are inclusive from, exclusive until.
13#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
14#[serde(deny_unknown_fields)]
15pub struct TaskFilters {
16    pub state: Option<TaskState>,
17    pub queue: Option<String>,
18    pub submitted_from: Option<Timestamp>,
19    pub submitted_until: Option<Timestamp>,
20    /// An empty string matches empty correlation metadata; absence applies no filter.
21    pub correlation_key: Option<String>,
22}
23
24impl TaskFilters {
25    pub fn validate(&self) -> Result<()> {
26        if let Some(queue) = &self.queue {
27            validate_text(queue, 128)?;
28        }
29        if let Some(key) = &self.correlation_key
30            && (key.len() > 512 || key.chars().any(char::is_control))
31        {
32            return Err(invalid("invalid correlation key filter"));
33        }
34        if [self.submitted_from, self.submitted_until]
35            .into_iter()
36            .flatten()
37            .any(|at| at > MAX_TIMESTAMP)
38            || matches!((self.submitted_from, self.submitted_until), (Some(from), Some(until)) if from >= until)
39        {
40            return Err(invalid("invalid submission time range"));
41        }
42        Ok(())
43    }
44
45    /// Whether a status satisfies every requested exact filter.
46    pub fn matches(&self, task: &TaskStatus) -> bool {
47        self.state.is_none_or(|state| state == task.state)
48            && self.queue.as_ref().is_none_or(|queue| queue == &task.queue)
49            && self
50                .submitted_from
51                .is_none_or(|from| task.submitted_at >= from)
52            && self
53                .submitted_until
54                .is_none_or(|until| task.submitted_at < until)
55            && self
56                .correlation_key
57                .as_ref()
58                .is_none_or(|key| task.correlation_key.as_ref() == Some(key))
59    }
60}
61
62/// One bounded read. Cursors bind the scope and filters, but allow a new page size.
63#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
64#[serde(deny_unknown_fields)]
65pub struct TaskListQuery {
66    #[serde(default)]
67    pub filters: TaskFilters,
68    #[serde(default = "default_limit")]
69    pub limit: u32,
70    pub cursor: Option<String>,
71}
72
73impl Default for TaskListQuery {
74    fn default() -> Self {
75        Self {
76            filters: TaskFilters::default(),
77            limit: TASK_LIST_DEFAULT_LIMIT,
78            cursor: None,
79        }
80    }
81}
82
83/// Immutable seek key, ordered by submission time and then UTF-8 task ID bytes.
84#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
85#[serde(deny_unknown_fields)]
86pub struct TaskPosition {
87    pub submitted_at: Timestamp,
88    pub task_id: String,
89}
90impl From<&TaskStatus> for TaskPosition {
91    fn from(task: &TaskStatus) -> Self {
92        Self {
93            submitted_at: task.submitted_at,
94            task_id: task.task_id.clone(),
95        }
96    }
97}
98impl TaskPosition {
99    fn validate(&self, filters: &TaskFilters) -> Result<()> {
100        validate_text(&self.task_id, 128)?;
101        if self.submitted_at > MAX_TIMESTAMP
102            || filters
103                .submitted_from
104                .is_some_and(|from| self.submitted_at < from)
105            || filters
106                .submitted_until
107                .is_some_and(|until| self.submitted_at >= until)
108        {
109            return Err(invalid("invalid cursor position"));
110        }
111        Ok(())
112    }
113}
114
115#[derive(Serialize, Deserialize)]
116#[serde(deny_unknown_fields)]
117struct Cursor {
118    version: u32,
119    scope: Scope,
120    filters: TaskFilters,
121    position: TaskPosition,
122}
123
124impl TaskListQuery {
125    /// Reject invalid bounds or a cursor from another scope or filter set.
126    pub fn validate(&self, scope: &Scope) -> Result<Option<TaskPosition>> {
127        scope.validate()?;
128        self.filters.validate()?;
129        if !(1..=TASK_LIST_MAX_LIMIT).contains(&self.limit) {
130            return Err(invalid("task list limit must be between 1 and 100"));
131        }
132        self.cursor
133            .as_deref()
134            .map(|cursor| self.decode_cursor(scope, cursor))
135            .transpose()
136    }
137
138    /// Encode the last returned position. This token is opaque, not a snapshot.
139    pub fn next_cursor(&self, scope: &Scope, position: &TaskPosition) -> Result<String> {
140        self.validate(scope)?;
141        position.validate(&self.filters)?;
142        let bytes = serde_json::to_vec(&Cursor {
143            version: 1,
144            scope: scope.clone(),
145            filters: self.filters.clone(),
146            position: position.clone(),
147        })
148        .map_err(|_| invalid("could not encode task cursor"))?;
149        const HEX: &[u8; 16] = b"0123456789abcdef";
150        let mut cursor = String::with_capacity(bytes.len() * 2);
151        for byte in bytes {
152            cursor.push(char::from(HEX[usize::from(byte >> 4)]));
153            cursor.push(char::from(HEX[usize::from(byte & 15)]));
154        }
155        if cursor.len() > TASK_CURSOR_MAX_BYTES {
156            return Err(invalid("task cursor exceeds size limit"));
157        }
158        Ok(cursor)
159    }
160
161    fn decode_cursor(&self, scope: &Scope, text: &str) -> Result<TaskPosition> {
162        if text.is_empty() || text.len() > TASK_CURSOR_MAX_BYTES || !text.len().is_multiple_of(2) {
163            return Err(invalid("invalid task cursor"));
164        }
165        let digit = |byte| match byte {
166            b'0'..=b'9' => Ok(byte - b'0'),
167            b'a'..=b'f' => Ok(byte - b'a' + 10),
168            _ => Err(invalid("invalid task cursor")),
169        };
170        let bytes = text
171            .as_bytes()
172            .as_chunks::<2>()
173            .0
174            .iter()
175            .map(|pair| Ok((digit(pair[0])? << 4) | digit(pair[1])?))
176            .collect::<Result<Vec<u8>>>()?;
177        let cursor: Cursor = decode_unique_json(&bytes, TASK_CURSOR_MAX_BYTES / 2)
178            .map_err(|_| invalid("invalid task cursor"))?;
179        if cursor.version != 1 || cursor.scope != *scope || cursor.filters != self.filters {
180            return Err(invalid("task cursor does not match scope or filters"));
181        }
182        cursor.position.validate(&self.filters)?;
183        Ok(cursor.position)
184    }
185}
186
187/// A coherent committed view for this read; later pages use later read snapshots.
188#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
189#[serde(deny_unknown_fields)]
190pub struct TaskPage {
191    pub items: Vec<TaskStatus>,
192    #[serde(deserialize_with = "required_option")]
193    pub next_cursor: Option<String>,
194}
195impl TaskPage {
196    /// Validate an adapter response, including strict descending continuation.
197    pub fn validate(&self, scope: &Scope, query: &TaskListQuery) -> Result<()> {
198        let mut previous = query.validate(scope)?;
199        let inconsistent = || ContractError::Unavailable("inconsistent task page".into());
200        if self.items.len() > query.limit as usize {
201            return Err(inconsistent());
202        }
203        let mut ids = std::collections::HashSet::with_capacity(self.items.len());
204        for task in &self.items {
205            task.validate()?;
206            let position = TaskPosition::from(task);
207            if task.scope != *scope
208                || !query.filters.matches(task)
209                || previous
210                    .as_ref()
211                    .is_some_and(|previous| position >= *previous)
212                || !ids.insert(&task.task_id)
213            {
214                return Err(inconsistent());
215            }
216            previous = Some(position);
217        }
218        if let Some(cursor) = &self.next_cursor {
219            let position = query
220                .decode_cursor(scope, cursor)
221                .map_err(|_| inconsistent())?;
222            if self.items.len() != query.limit as usize
223                || self.items.last().map(TaskPosition::from).as_ref() != Some(&position)
224            {
225                return Err(inconsistent());
226            }
227        }
228        crate::submission::check_encoded_size(self, TASK_PAGE_MAX_BYTES, "task page")
229            .map_err(|_| inconsistent())
230    }
231}
232
233fn default_limit() -> u32 {
234    TASK_LIST_DEFAULT_LIMIT
235}
236fn invalid(message: &str) -> ContractError {
237    ContractError::InvalidInput(message.into())
238}
239fn required_option<'de, D, T>(deserializer: D) -> std::result::Result<Option<T>, D::Error>
240where
241    D: Deserializer<'de>,
242    T: Deserialize<'de>,
243{
244    Option::<T>::deserialize(deserializer)
245}
246
247#[cfg(test)]
248#[path = "discovery_tests.rs"]
249mod tests;
ledgence_orchestration_api/discovery.rs

ledgence_orchestration_api/
discovery.rs
