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

agent_effects_http/
lib.rs
