incus_client/operations.rs
1//! Async-operation lifecycle: every mutation Incus documents as
2//! long-running returns one of these, which callers wait on via
3//! [`Client::wait_for_operation`] rather than assuming synchronous
4//! completion.
5
6use std::time::Duration;
7
8use serde::{Deserialize, Serialize};
9use uuid::Uuid;
10
11use crate::error::{Error, Result};
12use crate::transport::{Client, IncusEnvelope, Method};
13
14/// Defensive floor between re-issued long-poll calls when `timeout = None`.
15/// Under Incus's documented behavior this branch should essentially never
16/// fire - a no-timeout `.../wait` call blocks server-side until the
17/// operation completes, so a genuinely quick non-terminal response
18/// shouldn't happen - but if a daemon/proxy ever did return quickly and
19/// repeatedly, this stops the loop from hot-spinning fresh Unix-socket
20/// connections with no backoff at all.
21const WAIT_REPOLL_MIN_INTERVAL: Duration = Duration::from_millis(50);
22
23/// `#[non_exhaustive]`-equivalent: an unrecognized `class` value from a
24/// future Incus version becomes `Other(<the raw string>)` rather than
25/// failing deserialization outright (which is what a plain
26/// `#[derive(Serialize, Deserialize)]` enum would do here) - consistent
27/// with `Error`'s own `#[non_exhaustive]` forward-compatibility stance.
28/// `Serialize`/`Deserialize` are implemented by hand rather than derived so
29/// `Other` round-trips back to its original wire string exactly, instead of
30/// serializing as `{"Other": "..."}`.
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub enum OperationClass {
33 Task,
34 Websocket,
35 Token,
36 Other(String),
37}
38
39impl OperationClass {
40 fn as_wire_str(&self) -> &str {
41 match self {
42 OperationClass::Task => "task",
43 OperationClass::Websocket => "websocket",
44 OperationClass::Token => "token",
45 OperationClass::Other(raw) => raw.as_str(),
46 }
47 }
48}
49
50impl Serialize for OperationClass {
51 fn serialize<S: serde::Serializer>(
52 &self,
53 serializer: S,
54 ) -> std::result::Result<S::Ok, S::Error> {
55 serializer.serialize_str(self.as_wire_str())
56 }
57}
58
59impl<'de> Deserialize<'de> for OperationClass {
60 fn deserialize<D: serde::Deserializer<'de>>(
61 deserializer: D,
62 ) -> std::result::Result<Self, D::Error> {
63 let raw = String::deserialize(deserializer)?;
64 Ok(match raw.as_str() {
65 "task" => OperationClass::Task,
66 "websocket" => OperationClass::Websocket,
67 "token" => OperationClass::Token,
68 _ => OperationClass::Other(raw),
69 })
70 }
71}
72
73/// An Incus asynchronous operation - see
74/// <https://linuxcontainers.org/incus/docs/main/rest-api/>. `resources` and
75/// `metadata` stay untyped (`serde_json::Value`) because their shape varies
76/// per operation kind; the well-known top-level fields are fully typed.
77#[derive(Debug, Clone, Deserialize)]
78pub struct Operation {
79 pub id: Uuid,
80 pub class: OperationClass,
81 pub status: String,
82 pub status_code: u16,
83 #[serde(default)]
84 pub resources: serde_json::Value,
85 #[serde(default)]
86 pub metadata: Option<serde_json::Value>,
87 pub may_cancel: bool,
88 #[serde(default)]
89 pub err: Option<String>,
90}
91
92impl Operation {
93 fn is_terminal(&self) -> bool {
94 // Per Incus's 3-digit status_code scheme: 100-199 are in-progress
95 // states, 200-399 are positive (terminal) results, 400-599 are
96 // negative (terminal) results.
97 self.status_code >= 200
98 }
99
100 fn is_failure(&self) -> bool {
101 self.status_code >= 400
102 }
103}
104
105/// Converts an [`IncusEnvelope::Async`]'s untyped `metadata` into a typed
106/// [`Operation`]. Also accepts [`IncusEnvelope::Sync`] (used by
107/// `wait_for_operation`, whose `.../wait` endpoint returns the operation
108/// object directly as sync metadata, not wrapped in another async envelope).
109pub(crate) fn operation_from_envelope(envelope: IncusEnvelope) -> Result<Operation> {
110 let metadata = match envelope {
111 IncusEnvelope::Async { metadata, .. } => metadata,
112 IncusEnvelope::Sync { metadata, .. } => metadata,
113 };
114 Ok(serde_json::from_value(metadata)?)
115}
116
117/// Like [`operation_from_envelope`], but for endpoints Incus documents as
118/// *conditionally* sync-or-async depending on the request payload rather
119/// than always one or the other - e.g. creating a blank storage volume is
120/// synchronous, but creating one by copying another volume is async
121/// (verified against `cmd/incusd/storage_volumes.go`'s `doVolumeCreateOrCopy`
122/// on the `lxc/incus` `main` branch: it returns `response.EmptySyncResponse`
123/// when `req.Source.Name == ""`, and `operations.OperationResponse(op)`
124/// otherwise). Returns `None` for a genuinely-sync response (nothing to
125/// wait for) rather than trying to parse an `Operation` out of a body that
126/// doesn't contain one - unlike `operation_from_envelope`, which assumes
127/// every sync envelope it's given already contains the operation itself
128/// (true for `wait_for_operation`'s `.../wait` responses, not true here).
129pub(crate) fn optional_operation_from_envelope(
130 envelope: IncusEnvelope,
131) -> Result<Option<Operation>> {
132 match envelope {
133 IncusEnvelope::Sync { .. } => Ok(None),
134 IncusEnvelope::Async { metadata } => Ok(Some(serde_json::from_value(metadata)?)),
135 }
136}
137
138impl Client {
139 /// Waits for operation `id` to reach a terminal status, using Incus's
140 /// `.../wait?timeout=<seconds>` long-poll endpoint.
141 ///
142 /// - `timeout = Some(duration)`: bounds a *single* long-poll call. If
143 /// the operation is still in-progress when that window elapses, this
144 /// returns `Ok(Operation)` with the in-progress snapshot - it does
145 /// **not** re-poll, since the caller explicitly chose how long they're
146 /// willing to wait.
147 /// - `timeout = None`: waits indefinitely by transparently re-issuing
148 /// the long-poll call as many times as needed until a terminal status
149 /// is reached. Each individual call is still bounded server-side; this
150 /// just means the method as a whole doesn't return until completion.
151 ///
152 /// A terminal status in the 400-599 (failure) range returns
153 /// `Err(Error::OperationFailed { .. })`, not `Ok(Operation)` - callers
154 /// don't need to inspect `status_code` themselves to detect failure.
155 pub async fn wait_for_operation(
156 &self,
157 id: Uuid,
158 timeout: Option<Duration>,
159 ) -> Result<Operation> {
160 loop {
161 let query_value;
162 let query: &[(&str, &str)] = if let Some(duration) = timeout {
163 // Round up rather than truncate: `duration.as_secs()` alone
164 // would send `Some(Duration::from_millis(500))` as
165 // `?timeout=0`, which - depending on how Incus treats a
166 // zero-second wait - could return immediately instead of
167 // honoring something close to the caller's actual request.
168 let secs = duration.as_secs() + u64::from(duration.subsec_nanos() > 0);
169 query_value = secs.to_string();
170 &[("timeout", query_value.as_str())]
171 } else {
172 &[]
173 };
174 let path = format!("/1.0/operations/{id}/wait");
175 // Bypass the client's default per-request timeout here - this
176 // call already has its own server-side bound (the `timeout`
177 // query param above, or a genuinely unbounded long-poll when
178 // the caller passed `None`), and a legitimately slow operation
179 // (e.g. a large image import) must not fail with a client-side
180 // Error::Timeout just because it outlives the client's default,
181 // which is sized for ordinary fast requests, not long-polls.
182 let envelope = self
183 .request_with_timeout(Method::Get, &path, query, None, None, None)
184 .await?;
185 let operation = operation_from_envelope(envelope)?;
186
187 if !operation.is_terminal() {
188 if timeout.is_some() {
189 // Caller set an explicit bound on one long-poll call;
190 // honor it rather than looping past it.
191 return Ok(operation);
192 }
193 // No caller-set bound: this window elapsed without a
194 // terminal status, so re-issue the wait call and keep
195 // waiting - after a small defensive floor (see
196 // WAIT_REPOLL_MIN_INTERVAL's doc comment).
197 tokio::time::sleep(WAIT_REPOLL_MIN_INTERVAL).await;
198 continue;
199 }
200
201 if operation.is_failure() {
202 return Err(Error::OperationFailed {
203 id: operation.id,
204 status_code: operation.status_code,
205 err: operation.err,
206 });
207 }
208
209 return Ok(operation);
210 }
211 }
212
213 /// Cancels operation `op` if it's cancellable. Short-circuits with
214 /// `Error::NotCancellable` (no network call) when `op.may_cancel` is
215 /// false, since the server would reject it anyway and there's no reason
216 /// to round-trip to find that out.
217 pub async fn cancel_operation(&self, op: &Operation) -> Result<()> {
218 if !op.may_cancel {
219 return Err(Error::NotCancellable);
220 }
221 let path = format!("/1.0/operations/{}", op.id);
222 self.request(Method::Delete, &path, &[], None, None).await?;
223 Ok(())
224 }
225}
226
227#[cfg(test)]
228#[path = "operations_tests.rs"]
229mod tests;