cratestack_axum/idempotency/service.rs
1//! `IdempotencyService` — the tower `Service` that owns the per-request
2//! state machine (admit → run → complete/release).
3//!
4//! Since ADR 0015 slice 1 the *decision* is not made here: this is the
5//! HTTP adapter that assembles a `cratestack_exec::OpInput` out of things
6//! only a transport knows — the `Idempotency-Key` header, the method,
7//! the path + query, the content-type, the buffered body — and asks
8//! `OpExecutor::admit`. Everything transport-shaped stayed put on
9//! purpose; nothing about the bytes on the wire changed.
10
11use std::sync::Arc;
12
13use axum::body::Body;
14use axum::extract::Request;
15use axum::response::Response;
16use cratestack_core::CratestackError;
17use cratestack_exec::{OpAdmission, OpExecutor};
18use http::header;
19use tower::Service;
20
21use crate::middleware_error::middleware_error_response;
22
23use super::finish::finish_response;
24use super::hash::{hash_request, is_idempotent_target_method};
25use super::parse::parse_idempotency_key;
26use super::reserve::{Reservation, admit_or_response};
27use super::store::MAX_BODY_BYTES;
28
29#[derive(Clone)]
30pub struct IdempotencyService<S> {
31 pub(super) inner: S,
32 pub(super) executor: OpExecutor,
33 pub(super) principal_fingerprint:
34 Arc<dyn Fn(&Request) -> Result<String, CratestackError> + Send + Sync>,
35 pub(super) op_resolver: Arc<dyn Fn(&Request) -> OpAdmission + Send + Sync>,
36}
37
38impl<S> Service<Request> for IdempotencyService<S>
39where
40 S: Service<Request, Response = Response, Error = std::convert::Infallible>
41 + Clone
42 + Send
43 + 'static,
44 S::Future: Send + 'static,
45{
46 type Response = Response;
47 type Error = std::convert::Infallible;
48 type Future =
49 std::pin::Pin<Box<dyn std::future::Future<Output = Result<Response, Self::Error>> + Send>>;
50
51 fn poll_ready(
52 &mut self,
53 cx: &mut std::task::Context<'_>,
54 ) -> std::task::Poll<Result<(), Self::Error>> {
55 self.inner.poll_ready(cx)
56 }
57
58 fn call(&mut self, req: Request) -> Self::Future {
59 let mut inner = self.inner.clone();
60 let executor = self.executor.clone();
61 let principal_fp = self.principal_fingerprint.clone();
62 let op_resolver = self.op_resolver.clone();
63 Box::pin(async move {
64 let method = req.method().clone();
65 if !is_idempotent_target_method(&method) {
66 return inner.call(req).await;
67 }
68 let key = match parse_idempotency_key(req.headers()) {
69 Ok(Some(k)) => k,
70 Ok(None) => return inner.call(req).await,
71 Err(error) => {
72 return Ok(middleware_error_response(
73 req.headers(),
74 req.uri().path(),
75 error,
76 ));
77 }
78 };
79 // Resolve the op BEFORE anything else touches the request:
80 // both resolvers need only the method and the path (plus
81 // `MatchedPath` from the extensions), none of which survives
82 // `into_parts` below. With no resolver installed this is
83 // `OpAdmission::unresolved()`, whose `idempotent_by_default`
84 // is `false` — so the short-circuit below never fires and the
85 // path through this function is bit-for-bit the pre-ADR-0015
86 // one.
87 let op = (op_resolver)(&req);
88 if op.idempotent_by_default {
89 // The op does not participate: `admit` would answer
90 // `Bypass`, and everything between here and there exists
91 // only to build that answer's argument. Returning now,
92 // rather than after buffering, makes a bypassed request
93 // behave exactly like one that sent no `Idempotency-Key`
94 // at all — otherwise a `@no_idempotency` POST paid the
95 // 2 MiB request cap to compute a fingerprint nobody reads
96 // (and was refused at 2 MiB + 1 for carrying a header its
97 // own schema says does nothing), and could be refused
98 // outright by the principal fingerprint (cratestack#416)
99 // for want of a namespace it will never use.
100 return inner.call(req).await;
101 }
102 let principal = match (principal_fp)(&req) {
103 Ok(principal) => principal,
104 Err(error) => {
105 return Ok(middleware_error_response(
106 req.headers(),
107 req.uri().path(),
108 error,
109 ));
110 }
111 };
112 // Hash the full path + query string. Skipping the query
113 // makes `POST /transfer?dry_run=true` collide with
114 // `POST /transfer?dry_run=false` under the same key, so a
115 // dry-run preview would replay the live execution's
116 // response (or vice versa). Banks routinely encode
117 // operation modifiers like `?confirm=true` or
118 // `?settlement=instant` in the query string — those must
119 // produce distinct idempotency hashes.
120 let path = req
121 .uri()
122 .path_and_query()
123 .map(|pq| pq.as_str().to_owned())
124 .unwrap_or_else(|| req.uri().path().to_owned());
125 let content_type = req
126 .headers()
127 .get(header::CONTENT_TYPE)
128 .and_then(|v| v.to_str().ok())
129 .map(|s| s.to_owned());
130
131 // Buffer the request body so we can both hash it and replay
132 // it into the inner handler.
133 let (parts, body) = req.into_parts();
134 // Kept alive past the `Request::from_parts` below because
135 // every error response this middleware emits after that point
136 // still has to negotiate its content type against the
137 // *request's* headers (cratestack#846). One `HeaderMap` clone
138 // per idempotent request; the alternative — reconstructing a
139 // headers-shaped value at four call sites — costs more in
140 // clarity than this does in allocation.
141 let error_headers = parts.headers.clone();
142 let error_path = parts.uri.path().to_owned();
143 let bytes = match axum::body::to_bytes(body, MAX_BODY_BYTES).await {
144 Ok(b) => b,
145 Err(_) => {
146 return Ok(middleware_error_response(
147 &error_headers,
148 &error_path,
149 CratestackError::BadRequest(
150 "request body exceeds idempotency buffer limit".to_owned(),
151 ),
152 ));
153 }
154 };
155 let hash = hash_request(&method, &path, content_type.as_deref(), &bytes);
156
157 // Atomic reservation: exactly one caller gets `Reserved`,
158 // and only then do we let the handler run. Concurrent
159 // callers with the same key + same hash see `InFlight`;
160 // different-body conflicts see `Conflict`. This is the
161 // banking-grade duplicate-execution guarantee that the
162 // previous fetch-then-put pattern could not provide. The
163 // decision itself is L3's (ADR 0015 slice 1); the four
164 // outcomes and their responses are unchanged.
165 let token = match admit_or_response(
166 &executor,
167 op,
168 &principal,
169 &key,
170 hash,
171 &error_headers,
172 &error_path,
173 )
174 .await
175 {
176 Reservation::Held(token) => Some(token),
177 // The op opted out (`@no_idempotency`, or a read): run
178 // it, but there is no reservation to complete or release
179 // afterwards.
180 Reservation::Bypass => None,
181 Reservation::Finished(response) => return Ok(response),
182 };
183
184 // Run the handler, then dispose of its response
185 // (release / forward / persist) in `finish_response`.
186 let req2 = Request::from_parts(parts, Body::from(bytes));
187 let response_result = inner.call(req2).await;
188 Ok(finish_response(
189 &executor,
190 token,
191 &principal,
192 &key,
193 response_result,
194 &error_headers,
195 &error_path,
196 )
197 .await)
198 })
199 }
200}