agent_effects_http/
lib.rs

1//! HTTP effects for [`agent-effects`](https://crates.io/crates/agent-effects).
2//!
3//! [`HttpEffect`] builds an effect's action from an HTTP request: it sends
4//! the effect's stable idempotency key in a header, and classifies every
5//! failure so the runtime knows whether the request may have applied.
6//!
7//! ```no_run
8//! use agent_effects::{EffectKind, Runtime};
9//! use agent_effects_http::HttpEffect;
10//! use agent_effects_memory::MemoryStore;
11//!
12//! #[derive(serde::Serialize)]
13//! struct Charge { amount: u64, currency: &'static str }
14//! #[derive(serde::Serialize, serde::Deserialize)]
15//! struct Payment { id: String }
16//!
17//! # async fn demo() -> Result<(), agent_effects::RuntimeError> {
18//! let client = reqwest::Client::new();
19//! let runtime = Runtime::new(MemoryStore::new());
20//! let outcome = runtime
21//!     .effect("payment.charge", "order-42")
22//!     .kind(EffectKind::IrreversibleWrite)
23//!     .remote_idempotency(true) // the provider honours Idempotency-Key
24//!     .run(
25//!         HttpEffect::post(&client, "https://api.example.com/v1/charges")
26//!             .json(&Charge { amount: 4200, currency: "eur" })
27//!             .idempotency_key_header()
28//!             .send_json::<Payment>(),
29//!     )
30//!     .await?;
31//! # Ok(())
32//! # }
33//! ```
34//!
35//! The crate enables no reqwest features: TLS and the rest come from your
36//! own `reqwest` dependency and the [`reqwest::Client`] you pass in.
37//!
38//! # How failures are classified
39//!
40//! A failure is only definite when the request provably never reached the
41//! server, or the server's answer says it did not apply. Everything in
42//! between is ambiguous, and the runtime then verifies, re-runs only if
43//! safe, or escalates.
44//!
45//! | Outcome | Class |
46//! |---|---|
47//! | connect, DNS or TLS failure | `Ambiguous`; `Transient` (never sent) only with [`HttpEffect::client_follows_no_redirects`] |
48//! | invalid request (bad URL or header) | `Validation` (never sent) |
49//! | timeout or connection lost after sending | `Ambiguous` |
50//! | 2xx with an unreadable or unparseable body | `Ambiguous` (it applied, the answer was lost) |
51//! | 3xx (seen only when the client does not follow redirects) | `Ambiguous` |
52//! | 408, 425, 503 | `Transient` |
53//! | 429, or 503 with `Retry-After` | `RateLimited`, honouring `Retry-After` |
54//! | 400, 422 | `Validation` |
55//! | 401 | `Authentication` |
56//! | 403 | `Authorization` |
57//! | 409, 500, 502, 504, other 5xx | `Ambiguous` |
58//! | 501, other 4xx | `Permanent` |
59//!
60//! Override per request with [`HttpEffect::classify_status`].
61//!
62//! A client that follows redirects (reqwest's default) can fail to connect
63//! to a redirect's target after the original request applied, and reqwest
64//! reports that exactly like a failure to reach the original server. So a
65//! connection failure only proves the request was never sent when the
66//! client does not follow redirects: build the client with
67//! `.redirect(reqwest::redirect::Policy::none())` and say so with
68//! [`HttpEffect::client_follows_no_redirects`].
69
70use std::future::Future;
71use std::pin::Pin;
72use std::sync::Arc;
73use std::time::{Duration, SystemTime};
74
75use agent_effects::{EffectContext, EffectFailure, FailureClass, Verification};
76use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue, RETRY_AFTER};
77use reqwest::{Client, Method, StatusCode};
78use serde::de::DeserializeOwned;
79use serde::{Deserialize, Serialize};
80
81/// The conventional idempotency header.
82pub const IDEMPOTENCY_KEY: &str = "Idempotency-Key";
83
84/// At most this much of an error response's body is kept in the message.
85const ERROR_BODY_LIMIT: usize = 512;
86
87/// What a response status means.
88#[derive(Clone, Copy, Debug, PartialEq, Eq)]
89pub enum StatusClass {
90    /// The request succeeded; read the body.
91    Success,
92    /// The request failed with this class.
93    Failure(FailureClass),
94}
95
96/// The default meaning of `status`; see the [crate docs](crate).
97pub fn classify_status(status: StatusCode, headers: &HeaderMap) -> StatusClass {
98    let failure = match status.as_u16() {
99        200..=299 => return StatusClass::Success,
100        // A redirect the client did not follow: whether the request
101        // applied before it is unknown.
102        300..=399 => FailureClass::Ambiguous,
103        408 | 425 => FailureClass::Transient,
104        429 => FailureClass::RateLimited {
105            retry_after: retry_after(headers),
106        },
107        503 => match retry_after(headers) {
108            Some(after) => FailureClass::RateLimited {
109                retry_after: Some(after),
110            },
111            None => FailureClass::Transient,
112        },
113        400 | 422 => FailureClass::Validation,
114        401 => FailureClass::Authentication,
115        403 => FailureClass::Authorization,
116        409 | 500 | 502 | 504 => FailureClass::Ambiguous,
117        501 => FailureClass::Permanent,
118        500..=599 => FailureClass::Ambiguous,
119        _ => FailureClass::Permanent,
120    };
121    StatusClass::Failure(failure)
122}
123
124/// `Retry-After` as a delay: seconds, or an HTTP date.
125fn retry_after(headers: &HeaderMap) -> Option<Duration> {
126    let value = headers.get(RETRY_AFTER)?.to_str().ok()?.trim();
127    if let Ok(seconds) = value.parse::<u64>() {
128        return Some(Duration::from_secs(seconds));
129    }
130    let at = httpdate::parse_http_date(value).ok()?;
131    Some(at.duration_since(SystemTime::now()).unwrap_or_default())
132}
133
134/// The class of a request that failed without a response, from a client
135/// that may follow redirects: everything but an invalid request is
136/// ambiguous, a connection failure included (see the [crate docs](crate)).
137pub fn classify_error(error: &reqwest::Error) -> EffectFailure {
138    let message = error.to_string();
139    if error.is_builder() {
140        EffectFailure::validation(message)
141    } else {
142        // Timed out or lost after sending, the response broke midway, or a
143        // redirect's target was unreachable after the request applied.
144        EffectFailure::ambiguous(message)
145    }
146}
147
148/// Like [`classify_error`], for a client that does not follow redirects: a
149/// connection failure (including a connect timeout) means nothing reached
150/// the server, so the request was never sent.
151pub fn classify_error_without_redirects(error: &reqwest::Error) -> EffectFailure {
152    if error.is_connect() {
153        EffectFailure::ambiguous(error.to_string()).request_sent(false)
154    } else {
155        classify_error(error)
156    }
157}
158
159/// The future an action or verification built here returns.
160type Call<T> = Pin<Box<dyn Future<Output = Result<T, EffectFailure>> + Send>>;
161
162type StatusOverride = Arc<dyn Fn(StatusCode, &HeaderMap) -> Option<StatusClass> + Send + Sync>;
163
164/// An HTTP request as an effect's action. Build it, then turn it into an
165/// action with [`Self::send_json`] or [`Self::send`].
166#[derive(Clone)]
167#[must_use]
168pub struct HttpEffect {
169    client: Client,
170    method: Method,
171    url: String,
172    headers: HeaderMap,
173    body: Option<Vec<u8>>,
174    idempotency_header: Option<HeaderName>,
175    timeout: Option<Duration>,
176    classify: Option<StatusOverride>,
177    /// The caller declared that the client does not follow redirects.
178    no_redirects: bool,
179    /// A builder mistake, reported as a `Validation` failure when run.
180    invalid: Option<String>,
181}
182
183impl HttpEffect {
184    /// A request with `method` to `url`.
185    pub fn new(client: &Client, method: Method, url: impl Into<String>) -> Self {
186        Self {
187            client: client.clone(),
188            method,
189            url: url.into(),
190            headers: HeaderMap::new(),
191            body: None,
192            idempotency_header: None,
193            timeout: None,
194            classify: None,
195            no_redirects: false,
196            invalid: None,
197        }
198    }
199
200    /// A `GET` request.
201    pub fn get(client: &Client, url: impl Into<String>) -> Self {
202        Self::new(client, Method::GET, url)
203    }
204
205    /// A `POST` request.
206    pub fn post(client: &Client, url: impl Into<String>) -> Self {
207        Self::new(client, Method::POST, url)
208    }
209
210    /// A `PUT` request.
211    pub fn put(client: &Client, url: impl Into<String>) -> Self {
212        Self::new(client, Method::PUT, url)
213    }
214
215    /// A `PATCH` request.
216    pub fn patch(client: &Client, url: impl Into<String>) -> Self {
217        Self::new(client, Method::PATCH, url)
218    }
219
220    /// A `DELETE` request.
221    pub fn delete(client: &Client, url: impl Into<String>) -> Self {
222        Self::new(client, Method::DELETE, url)
223    }
224
225    /// Adds a header. Credentials belong here or in the client, not in the
226    /// effect's input.
227    pub fn header(mut self, name: &str, value: &str) -> Self {
228        match (HeaderName::try_from(name), HeaderValue::try_from(value)) {
229            (Ok(name), Ok(value)) => {
230                self.headers.append(name, value);
231            }
232            _ => self.invalid = Some(format!("invalid header `{name}`")),
233        }
234        self
235    }
236
237    /// Sends `body` as JSON.
238    pub fn json<T: Serialize + ?Sized>(mut self, body: &T) -> Self {
239        match serde_json::to_vec(body) {
240            Ok(bytes) => {
241                self.body = Some(bytes);
242                self.headers
243                    .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
244            }
245            Err(e) => self.invalid = Some(format!("body is not JSON-serializable: {e}")),
246        }
247        self
248    }
249
250    /// Sends `body` as is.
251    pub fn body(mut self, body: impl Into<Vec<u8>>) -> Self {
252        self.body = Some(body.into());
253        self
254    }
255
256    /// Sends the effect's idempotency key in header `name` on every attempt.
257    /// The key is the same across attempts, so a server that honours it
258    /// applies the request once. Pair with
259    /// [`EffectBuilder::remote_idempotency`](agent_effects::EffectBuilder::remote_idempotency).
260    pub fn idempotency_header(mut self, name: &str) -> Self {
261        match HeaderName::try_from(name) {
262            Ok(name) => self.idempotency_header = Some(name),
263            Err(_) => self.invalid = Some(format!("invalid header `{name}`")),
264        }
265        self
266    }
267
268    /// Sends the idempotency key as `Idempotency-Key`.
269    pub fn idempotency_key_header(self) -> Self {
270        self.idempotency_header(IDEMPOTENCY_KEY)
271    }
272
273    /// Gives up on one request after `timeout`: an ambiguous failure, since
274    /// it may have been sent.
275    pub fn timeout(mut self, timeout: Duration) -> Self {
276        self.timeout = Some(timeout);
277        self
278    }
279
280    /// Declares that the client does not follow redirects: it was built
281    /// with `.redirect(reqwest::redirect::Policy::none())`. Only then does a
282    /// connection failure prove the request was never sent, so that it is
283    /// retried as `Transient`, even for an irreversible write. Without it,
284    /// a connection failure is ambiguous. This does not change the client:
285    /// declaring it for a client that does follow redirects can duplicate
286    /// effects.
287    pub fn client_follows_no_redirects(mut self) -> Self {
288        self.no_redirects = true;
289        self
290    }
291
292    /// Overrides how statuses are classified: return `Some` to decide,
293    /// `None` to fall back to [`classify_status`]. For APIs with their own
294    /// conventions, e.g. a 409 that means "already done".
295    pub fn classify_status(
296        mut self,
297        classify: impl Fn(StatusCode, &HeaderMap) -> Option<StatusClass> + Send + Sync + 'static,
298    ) -> Self {
299        self.classify = Some(Arc::new(classify));
300        self
301    }
302
303    /// The action: sends the request and parses a successful response as
304    /// JSON `T`. An empty body reads as JSON `null`, e.g. for `T = ()`.
305    pub fn send_json<T: DeserializeOwned + Send + 'static>(
306        self,
307    ) -> impl Fn(EffectContext) -> Call<T> + Send + Sync + 'static {
308        let request = Arc::new(self);
309        move |ctx| {
310            let request = Arc::clone(&request);
311            Box::pin(async move {
312                let body = request.execute(&ctx).await?;
313                parse_json(&body).map_err(|e| {
314                    EffectFailure::ambiguous(format!(
315                        "the request succeeded but its response could not be read: {e}"
316                    ))
317                })
318            })
319        }
320    }
321
322    /// The action: sends the request and returns the successful response's
323    /// status and body as text.
324    pub fn send(self) -> impl Fn(EffectContext) -> Call<HttpResponse> + Send + Sync + 'static {
325        let request = Arc::new(self);
326        move |ctx| {
327            let request = Arc::clone(&request);
328            Box::pin(async move {
329                let (status, body) = request.execute_with_status(&ctx).await?;
330                Ok(HttpResponse {
331                    status,
332                    body: String::from_utf8_lossy(&body).into_owned(),
333                })
334            })
335        }
336    }
337
338    /// A verification: a lookup whose 2xx response, parsed as JSON `T`,
339    /// confirms the effect, and whose 404 or 410 (only those) says it did not
340    /// apply. Any other status or error is a failed check, which the runtime
341    /// treats as inconclusive: a broken lookup must never be read as "not
342    /// applied". Pass it to `.verify(..)` if the lookup reads its own
343    /// writes, else to `.verify_eventually(settle, ..)`.
344    pub fn verify_json<T: DeserializeOwned + Send + 'static>(
345        self,
346    ) -> impl Fn(EffectContext) -> Call<Verification<T>> + Send + Sync + 'static {
347        let request = Arc::new(self);
348        move |ctx| {
349            let request = Arc::clone(&request);
350            Box::pin(async move {
351                let response = request.send_once(&ctx).await?;
352                let status = response.status();
353                // Only "not found" proves the effect did not apply; every
354                // other failure is a check that could not tell.
355                if matches!(status.as_u16(), 404 | 410) {
356                    return Ok(Verification::NotApplied);
357                }
358                let body = request.success_body(response).await?;
359                parse_json(&body)
360                    .map(Verification::Confirmed)
361                    .map_err(|e| EffectFailure::ambiguous(format!("unreadable lookup: {e}")))
362            })
363        }
364    }
365
366    async fn execute(&self, ctx: &EffectContext) -> Result<Vec<u8>, EffectFailure> {
367        self.execute_with_status(ctx).await.map(|(_, body)| body)
368    }
369
370    /// Sends one attempt; returns the status and body of a success.
371    async fn execute_with_status(
372        &self,
373        ctx: &EffectContext,
374    ) -> Result<(u16, Vec<u8>), EffectFailure> {
375        let response = self.send_once(ctx).await?;
376        let status = response.status().as_u16();
377        Ok((status, self.success_body(response).await?))
378    }
379
380    /// Sends the request once. Fails only if no response arrived.
381    async fn send_once(&self, ctx: &EffectContext) -> Result<reqwest::Response, EffectFailure> {
382        if let Some(problem) = &self.invalid {
383            return Err(EffectFailure::validation(problem.clone()));
384        }
385        let mut request = self
386            .client
387            .request(self.method.clone(), &self.url)
388            .headers(self.headers.clone());
389        if let Some(name) = &self.idempotency_header {
390            let key = ctx.idempotency_key().to_string();
391            let value = HeaderValue::try_from(key).map_err(EffectFailure::validation)?;
392            request = request.header(name.clone(), value);
393        }
394        if let Some(body) = &self.body {
395            request = request.body(body.clone());
396        }
397        if let Some(timeout) = self.timeout {
398            request = request.timeout(timeout);
399        }
400        request.send().await.map_err(|e| {
401            if self.no_redirects {
402                classify_error_without_redirects(&e)
403            } else {
404                classify_error(&e)
405            }
406        })
407    }
408
409    /// The body of a successful response, or the classified failure.
410    async fn success_body(&self, response: reqwest::Response) -> Result<Vec<u8>, EffectFailure> {
411        let status = response.status();
412        let verdict = self
413            .classify
414            .as_ref()
415            .and_then(|classify| classify(status, response.headers()))
416            .unwrap_or_else(|| classify_status(status, response.headers()));
417        match verdict {
418            StatusClass::Success => {
419                // It applied; losing the body now makes the answer unknown.
420                let body = response.bytes().await.map_err(|e| {
421                    EffectFailure::ambiguous(format!(
422                        "HTTP {status}, but reading the response failed: {e}"
423                    ))
424                })?;
425                Ok(body.to_vec())
426            }
427            StatusClass::Failure(class) => {
428                let body = response.bytes().await.unwrap_or_default();
429                let mut snippet = String::from_utf8_lossy(&body).into_owned();
430                if snippet.len() > ERROR_BODY_LIMIT {
431                    let cut = (0..=ERROR_BODY_LIMIT)
432                        .rev()
433                        .find(|&i| snippet.is_char_boundary(i))
434                        .unwrap_or(0);
435                    snippet.truncate(cut);
436                    snippet.push('…');
437                }
438                Err(EffectFailure::new(
439                    class,
440                    format!("HTTP {status}: {snippet}"),
441                ))
442            }
443        }
444    }
445}
446
447/// A successful response, as stored and replayed.
448#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
449pub struct HttpResponse {
450    /// The status code.
451    pub status: u16,
452    /// The body, as text (invalid UTF-8 replaced).
453    pub body: String,
454}
455
456fn parse_json<T: DeserializeOwned>(body: &[u8]) -> Result<T, serde_json::Error> {
457    if body.iter().all(u8::is_ascii_whitespace) {
458        serde_json::from_slice(b"null")
459    } else {
460        serde_json::from_slice(body)
461    }
462}
463
464#[cfg(test)]
465mod tests {
466    use super::*;
467
468    fn class(status: u16, headers: &[(&str, &str)]) -> StatusClass {
469        let mut map = HeaderMap::new();
470        for (name, value) in headers {
471            map.insert(
472                HeaderName::try_from(*name).unwrap(),
473                HeaderValue::try_from(*value).unwrap(),
474            );
475        }
476        classify_status(StatusCode::from_u16(status).unwrap(), &map)
477    }
478
479    #[test]
480    fn statuses_map_to_failure_classes() {
481        use FailureClass as F;
482        use StatusClass::{Failure, Success};
483        let table = [
484            (200, Success),
485            (201, Success),
486            (204, Success),
487            (301, Failure(F::Ambiguous)),
488            (303, Failure(F::Ambiguous)),
489            (308, Failure(F::Ambiguous)),
490            (400, Failure(F::Validation)),
491            (401, Failure(F::Authentication)),
492            (403, Failure(F::Authorization)),
493            (404, Failure(F::Permanent)),
494            (408, Failure(F::Transient)),
495            (409, Failure(F::Ambiguous)),
496            (418, Failure(F::Permanent)),
497            (422, Failure(F::Validation)),
498            (425, Failure(F::Transient)),
499            (429, Failure(F::RateLimited { retry_after: None })),
500            (500, Failure(F::Ambiguous)),
501            (501, Failure(F::Permanent)),
502            (502, Failure(F::Ambiguous)),
503            (503, Failure(F::Transient)),
504            (504, Failure(F::Ambiguous)),
505            (599, Failure(F::Ambiguous)),
506        ];
507        for (status, expected) in table {
508            assert_eq!(class(status, &[]), expected, "HTTP {status}");
509        }
510    }
511
512    #[test]
513    fn retry_after_is_honoured_in_seconds_and_as_a_date() {
514        let seven = Some(Duration::from_secs(7));
515        assert_eq!(
516            class(429, &[("retry-after", "7")]),
517            StatusClass::Failure(FailureClass::RateLimited { retry_after: seven })
518        );
519        assert_eq!(
520            class(503, &[("retry-after", "7")]),
521            StatusClass::Failure(FailureClass::RateLimited { retry_after: seven })
522        );
523        let in_a_minute = httpdate::fmt_http_date(SystemTime::now() + Duration::from_secs(60));
524        let StatusClass::Failure(FailureClass::RateLimited {
525            retry_after: Some(after),
526        }) = class(429, &[("retry-after", &in_a_minute)])
527        else {
528            panic!("expected a rate limit");
529        };
530        assert!(after > Duration::from_secs(55) && after <= Duration::from_secs(60));
531    }
532
533    #[test]
534    fn an_empty_body_reads_as_null() {
535        let unit: () = parse_json(b"").unwrap();
536        assert_eq!(unit, ());
537        let none: Option<u8> = parse_json(b"  \n").unwrap();
538        assert_eq!(none, None);
539    }
540}
agent_effects_http/lib.rs

agent_effects_http/
lib.rs
