onlyne_client/runtime/
intent.rs

1use anyhow::{Context, Result};
2use chrono::{DateTime, Utc};
3use onlyne_proto::{ClientOp, Envelope, ErrorCode, MsgKind, Receipt, Report, ResBody, new_op_id};
4use onlyne_store::{ClientStore, IntentRow};
5use rusqlite::Connection;
6use serde_json::Value;
7use std::time::Duration;
8
9/// Refusals that end an intent.
10///
11/// `Invalid` stays out: the server answers it for several conditions a retry
12/// clears, and exhaustion keeps the row visible as a fault, so a transient
13/// refusal costs retries rather than the queued message (plan §6 line 289).
14pub const PERMANENT_ERRORS: &[ErrorCode] = &[
15    ErrorCode::AclDenied,
16    ErrorCode::Conflict,
17    ErrorCode::Forbidden,
18    ErrorCode::UnknownRole,
19    ErrorCode::NotAdmin,
20    ErrorCode::BadFrame,
21    ErrorCode::FrameTooLarge,
22    ErrorCode::ProtocolVersion,
23];
24
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum IntentState {
27    Pending,
28    Retrying,
29    Accepted,
30    Exhausted,
31}
32
33impl IntentState {
34    pub fn as_str(self) -> &'static str {
35        match self {
36            Self::Pending => "pending",
37            Self::Retrying => "retrying",
38            Self::Accepted => "accepted",
39            Self::Exhausted => "exhausted",
40        }
41    }
42}
43
44#[derive(Debug, Clone, PartialEq)]
45pub enum IntentResult {
46    Accepted(Option<Receipt>),
47    Retryable(ErrorCode, String),
48    Dropped(ErrorCode, String),
49    Exhausted,
50}
51
52/// Give one outbound envelope the `op_id` its intent row is keyed by.
53///
54/// The proto requires the key for every kind but `Note` (`Envelope::validate`
55/// in `onlyne-proto`), so a plugin's note legitimately arrives without one
56/// while the queue still keys every row by an id. Minting it here keeps the
57/// stamped envelope and the row's key one value: what the row replays is what
58/// it stored. A non-note envelope keeps the key it brought, which is what
59/// makes a re-delivered task dedup on its original id.
60pub fn stamp_op_id(envelope: &mut Envelope) -> String {
61    if let Some(op_id) = &envelope.op_id {
62        return op_id.clone();
63    }
64    let op_id = new_op_id();
65    envelope.op_id = Some(op_id.clone());
66    op_id
67}
68
69#[derive(Clone)]
70pub struct IntentMachine {
71    pub store: ClientStore,
72    pub attempts: u32,
73    pub backoff_ms: Vec<u64>,
74}
75
76impl IntentMachine {
77    pub fn new(store: ClientStore, attempts: u32, backoff_ms: Vec<u64>) -> Self {
78        Self {
79            store,
80            attempts,
81            backoff_ms,
82        }
83    }
84
85    /// Queue one envelope under the id its row is keyed by.
86    ///
87    /// A kind that may arrive without a key (a note) gets a fresh one from
88    /// [`stamp_op_id`], and the stamped envelope is what the row stores and
89    /// replays, so the row's `op_id` and its `env_json` never disagree.
90    pub fn enqueue(&self, envelope: &Envelope) -> Result<bool> {
91        let mut stamped = envelope.clone();
92        let op_id = stamp_op_id(&mut stamped);
93        Ok(self
94            .store
95            .enqueue_intent(&op_id, &serde_json::to_value(&stamped)?)?)
96    }
97
98    pub fn enqueue_value(&self, op_id: &str, envelope: &Value) -> Result<bool> {
99        Ok(self.store.enqueue_intent(op_id, envelope)?)
100    }
101
102    pub fn pending(&self) -> Result<Vec<IntentRow>> {
103        Ok(self.store.flush_order()?)
104    }
105
106    pub fn next_delay(&self, attempt: u32) -> Duration {
107        let idx = attempt.saturating_sub(1) as usize;
108        Duration::from_millis(
109            self.backoff_ms
110                .get(idx)
111                .copied()
112                .or_else(|| self.backoff_ms.last().copied())
113                .unwrap_or(1_000),
114        )
115    }
116
117    pub fn attempt(&self, row: &IntentRow, response: Option<&ResBody>) -> Result<IntentResult> {
118        let op_id = row.op_id.as_str();
119        let Some(body) = response else {
120            let next = row.attempt.saturating_add(1) as u32;
121            if next >= self.attempts {
122                self.store
123                    .exhaust_intent(op_id, "intent attempts exhausted")?;
124                self.record_exhausted(&row.env_json, row.attempt, "intent attempts exhausted")?;
125                return Ok(IntentResult::Exhausted);
126            }
127            let due = Utc::now() + self.next_delay(next);
128            self.store
129                .bump_intent(op_id, due, "connection unavailable")?;
130            return Ok(IntentResult::Retryable(
131                ErrorCode::Internal,
132                "connection unavailable".into(),
133            ));
134        };
135        self.apply_response(row, body)
136    }
137
138    /// Hold an intent whose send never reached the server.
139    ///
140    /// The link being down says nothing about the message, so the queue keeps
141    /// the row and the attempt counter stays where it was (plan §6 line 289).
142    pub fn defer(&self, row: &IntentRow, reason: &str) -> Result<IntentResult> {
143        let due = Utc::now() + self.next_delay(row.attempt.max(0) as u32 + 1);
144        self.store.defer_intent(&row.op_id, due, reason)?;
145        Ok(IntentResult::Retryable(
146            ErrorCode::Internal,
147            reason.to_string(),
148        ))
149    }
150
151    fn apply_response(
152        &self,
153        row: &IntentRow,
154        body: &onlyne_proto::ResBody,
155    ) -> Result<IntentResult> {
156        if body.ok {
157            let receipt = body
158                .data
159                .as_ref()
160                .and_then(|v| serde_json::from_value::<Receipt>(v.clone()).ok());
161            self.store
162                .accept_intent(&row.op_id, &body.data.clone().unwrap_or(Value::Null))?;
163            return Ok(IntentResult::Accepted(receipt));
164        }
165        let error = body
166            .error
167            .as_ref()
168            .context("error response missing payload")?;
169        if error.code == ErrorCode::Duplicate {
170            let receipt = body
171                .data
172                .as_ref()
173                .and_then(|v| serde_json::from_value::<Receipt>(v.clone()).ok());
174            self.store
175                .accept_intent(&row.op_id, &body.data.clone().unwrap_or(Value::Null))?;
176            return Ok(IntentResult::Accepted(receipt));
177        }
178        // A frame answered before the server session finished its `hello` names a
179        // window of the connection, so the row waits for the handshake instead of
180        // leaving the queue (plan §7 line 310's refusal).
181        if error.message == onlyne_proto::HELLO_REQUIRED_MESSAGE {
182            return self.defer(row, "connection not authenticated");
183        }
184        if PERMANENT_ERRORS.contains(&error.code) {
185            self.delete_intent(&row.op_id)?;
186            return Ok(IntentResult::Dropped(error.code, error.message.clone()));
187        }
188        let next = row.attempt.saturating_add(1) as u32;
189        if next >= self.attempts {
190            self.store.exhaust_intent(&row.op_id, &error.message)?;
191            self.record_exhausted(&row.env_json, row.attempt, &error.message)?;
192            Ok(IntentResult::Exhausted)
193        } else {
194            let due = Utc::now() + self.next_delay(next);
195            self.store.bump_intent(&row.op_id, due, &error.message)?;
196            Ok(IntentResult::Retryable(error.code, error.message.clone()))
197        }
198    }
199
200    fn delete_intent(&self, op_id: &str) -> Result<()> {
201        let conn = Connection::open(self.store.path())?;
202        conn.execute("DELETE FROM intents WHERE op_id = ?", [op_id])?;
203        Ok(())
204    }
205
206    fn record_exhausted(&self, env: &Value, attempt: i64, reason: &str) -> Result<()> {
207        let task = env
208            .get("causality")
209            .and_then(|v| v.get("task"))
210            .and_then(Value::as_str)
211            .unwrap_or("");
212        let _ =
213            onlyne_session::record_fault(&self.store, task, "intent_exhausted", "intent", reason)?;
214        let _ = attempt;
215        let report = Report::Fault {
216            task_id: Some(task.to_string()),
217            session_id: None,
218            generation: None,
219            seq: None,
220            kind: "intent_exhausted".into(),
221            reason: reason.into(),
222            desired: None,
223            observed: None,
224        };
225        let _ = self.store.append_event(
226            "report_fault",
227            &serde_json::to_value(&report).unwrap_or(Value::Null),
228        );
229        Ok(())
230    }
231}
232
233pub fn op_for_intent(row: &IntentRow) -> Result<ClientOp> {
234    if let Ok(op) = serde_json::from_value::<ClientOp>(row.env_json.clone()) {
235        return Ok(op);
236    }
237    let envelope: Envelope = serde_json::from_value(row.env_json.clone())?;
238    Ok(ClientOp::Send(Box::new(envelope)))
239}
240
241pub fn due(row: &IntentRow) -> Result<DateTime<Utc>> {
242    Ok(DateTime::parse_from_rfc3339(&row.next_attempt_at)?.with_timezone(&Utc))
243}
244
245/// The session whose completion intent one queue row carries, when the row is
246/// one.
247///
248/// Two shapes and no others, and the restriction is a soundness rule rather
249/// than tidiness. A receipt closes the drain the completion is riding on, and
250/// the reducer only admits one from a drain that is open — so feeding a receipt
251/// for an accepted op that merely *names* a task would close the completion's
252/// drain on the strength of something else entirely. An ack settles a delivery,
253/// a ready or heartbeat report publishes a projection, and a note wakes an
254/// agent: none of them is the completion, and none of them may end it. The two
255/// that are, are the `Completion` envelope the settlement sends and the
256/// `complete` report the residual reconcile sends when it holds no envelope.
257///
258/// The session is keyed by the task the intent answers, which is what a
259/// client-held session's own row is keyed by. A row whose payload names no task
260/// — a note, or an ack — answers `None`, and the caller leaves the reducer
261/// alone rather than guessing a session.
262pub fn completion_task_id(row: &IntentRow) -> Option<String> {
263    if let Ok(envelope) = serde_json::from_value::<Envelope>(row.env_json.clone()) {
264        if envelope.kind != MsgKind::Completion {
265            return None;
266        }
267        return envelope.task_id().map(str::to_string);
268    }
269    match serde_json::from_value::<ClientOp>(row.env_json.clone()) {
270        Ok(ClientOp::Report(Report::Complete { task_id, .. })) => Some(task_id),
271        _ => None,
272    }
273}
274
275pub fn permanent_error(code: ErrorCode) -> bool {
276    PERMANENT_ERRORS.contains(&code)
277}
onlyne_client/runtime/intent.rs

onlyne_client/runtime/
intent.rs
