1use std::borrow::Cow;
5use std::fmt::Display;
6use std::sync::Arc;
7use std::time::{Duration, Instant};
8
9use bytes::Bytes;
10use serde::de::DeserializeOwned;
11use url::Url;
12
13use crate::auth::{AuthStrategy, BearerAuth, CookieAuth, StaticTokenProvider, TokenProvider};
14use crate::cache::{CachedResponse, FileCache, ResponseCache, cache_key};
15use crate::config::Config;
16use crate::error::{Error, ErrorCode, MAX_ERROR_BODY_BYTES, retry_after_seconds};
17use crate::http::header::{
18 ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE, COOKIE, IF_NONE_MATCH, USER_AGENT,
19};
20use crate::http::{
21 Body, HeaderMap, HeaderValue, HttpClient, Method, Request, Response as HttpResponse, StatusCode,
22};
23use crate::observability::{
24 Hooks, NoopHooks, OperationInfo, OperationState, RequestInfo, RequestResult,
25};
26use crate::operation::{DEFAULT_RETRY_ON, Operation, RetryPolicy};
27use crate::pagination::Page;
28use crate::route::Route;
29use crate::security::{is_same_origin, require_secure_endpoint};
30use crate::version::default_user_agent;
31
32pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
34pub const DEFAULT_MAX_ATTEMPTS: u32 = 3;
38pub const DEFAULT_BASE_DELAY: Duration = Duration::from_secs(1);
40pub const DEFAULT_MAX_DELAY: Duration = Duration::from_secs(30);
45pub const DEFAULT_MAX_RETRY_AFTER: Duration = Duration::from_secs(60);
49pub const DEFAULT_MAX_JITTER: Duration = Duration::from_millis(100);
51pub const DEFAULT_MAX_PAGES: usize = 10_000;
53pub const DEFAULT_MAX_RESPONSE_BODY_BYTES: usize = 10 << 20;
55
56const MAX_REDIRECTS: usize = 10;
58
59#[derive(Clone)]
65pub struct Client {
66 pub(crate) shared: Arc<Shared>,
67}
68
69pub(crate) struct Shared {
70 pub(crate) config: Config,
71 pub(crate) base_url: Url,
72 pub(crate) http: Arc<dyn HttpClient>,
73 pub(crate) auth: Arc<dyn AuthStrategy>,
74 pub(crate) user_agent: String,
75 pub(crate) max_attempts: u32,
76 pub(crate) base_delay: Duration,
77 pub(crate) max_delay: Duration,
78 pub(crate) max_retry_after: Duration,
79 pub(crate) max_jitter: Duration,
80 pub(crate) max_pages: usize,
81 pub(crate) max_response_body_bytes: usize,
82 pub(crate) cache: Option<Arc<dyn ResponseCache>>,
83 pub(crate) hooks: Arc<dyn Hooks>,
84}
85
86#[derive(Clone)]
89pub struct AccountClient {
90 client: Client,
91 account_id: String,
92}
93
94impl std::fmt::Debug for Client {
95 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
96 f.debug_struct("Client")
97 .field("base_url", &self.shared.base_url.as_str())
98 .field("user_agent", &self.shared.user_agent)
99 .finish_non_exhaustive()
100 }
101}
102
103impl std::fmt::Debug for AccountClient {
104 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
105 f.debug_struct("AccountClient")
106 .field("account_id", &self.account_id)
107 .field("client", &self.client)
108 .finish()
109 }
110}
111
112impl AccountClient {
113 pub fn client(&self) -> &Client {
115 &self.client
116 }
117
118 pub fn account_id(&self) -> &str {
120 &self.account_id
121 }
122
123 pub fn scope(&self) -> Scope<'_> {
125 Scope {
126 client: &self.client,
127 account_id: Some(&self.account_id),
128 }
129 }
130}
131
132#[derive(Clone, Copy)]
134pub struct Scope<'a> {
135 client: &'a Client,
136 account_id: Option<&'a str>,
137}
138
139impl<'a> Scope<'a> {
140 pub fn client(&self) -> &'a Client {
142 self.client
143 }
144
145 pub fn account_id(&self) -> Option<&'a str> {
147 self.account_id
148 }
149
150 pub fn operation(
153 &self,
154 route: &'static Route,
155 params: &[&dyn Display],
156 ) -> Result<Operation, Error> {
157 if route.account_scoped && self.account_id.is_none() {
158 return Err(Error::usage_with_hint(
159 format!("{} needs an account", route.id),
160 "reach it through Client::for_account",
161 ));
162 }
163 let account_id = if route.account_scoped {
164 self.account_id
165 } else {
166 None
167 };
168 Operation::for_route(route, account_id, params)
169 }
170}
171
172#[derive(Clone)]
174#[non_exhaustive]
175pub struct Response {
176 pub status: StatusCode,
178 pub headers: HeaderMap,
180 pub body: Bytes,
182 pub url: Url,
184 pub from_cache: bool,
187}
188
189impl std::fmt::Debug for Response {
193 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
194 f.debug_struct("Response")
195 .field("status", &self.status)
196 .field("headers", &crate::security::redact_headers(&self.headers))
197 .field("body_len", &self.body.len())
198 .field("url", &self.url.as_str())
199 .field("from_cache", &self.from_cache)
200 .finish()
201 }
202}
203
204impl Response {
205 pub fn json<T: DeserializeOwned>(&self) -> Result<T, Error> {
208 let status = self.status.as_u16();
209 let error = if self.body.is_empty() {
210 Error::api(status, "empty response body")
211 } else {
212 match serde_json::from_slice(&self.body) {
213 Ok(value) => return Ok(value),
214 Err(error) => Error::api(status, "unexpected JSON")
215 .with_hint(error.to_string())
216 .with_source(error),
217 }
218 };
219 Err(match self.header("x-request-id") {
220 Some(request_id) => error.with_request_id(request_id),
221 None => error,
222 })
223 }
224
225 pub fn header(&self, name: &str) -> Option<&str> {
227 self.headers.get(name).and_then(|value| value.to_str().ok())
228 }
229}
230
231#[derive(Debug, Clone, Default, PartialEq, Eq)]
233pub struct RequestOptions {
234 pub no_retry: bool,
236 pub idempotent: bool,
238}
239
240impl RequestOptions {
241 pub fn new() -> RequestOptions {
243 RequestOptions::default()
244 }
245
246 pub fn no_retry(mut self) -> RequestOptions {
248 self.no_retry = true;
249 self
250 }
251
252 pub fn idempotent(mut self) -> RequestOptions {
254 self.idempotent = true;
255 self
256 }
257
258 pub(crate) fn apply(&self, mut operation: Operation) -> Operation {
259 if self.idempotent {
260 operation.idempotent(true);
261 }
262 if self.no_retry {
263 operation.no_retry();
264 }
265 operation
266 }
267}
268
269pub struct ClientBuilder {
271 config: Config,
272 auth: Option<Arc<dyn AuthStrategy>>,
273 http: Option<Arc<dyn HttpClient>>,
274 user_agent: String,
275 timeout: Duration,
276 max_attempts: u32,
277 base_delay: Duration,
278 max_delay: Duration,
279 max_retry_after: Duration,
280 max_jitter: Duration,
281 max_pages: usize,
282 max_response_body_bytes: usize,
283 cache: Option<Arc<dyn ResponseCache>>,
284 pub(crate) hooks: Arc<dyn Hooks>,
285}
286
287impl ClientBuilder {
288 pub fn new(config: Config) -> ClientBuilder {
290 ClientBuilder {
291 config,
292 auth: None,
293 http: None,
294 user_agent: default_user_agent(),
295 timeout: DEFAULT_TIMEOUT,
296 max_attempts: DEFAULT_MAX_ATTEMPTS,
297 base_delay: DEFAULT_BASE_DELAY,
298 max_delay: DEFAULT_MAX_DELAY,
299 max_retry_after: DEFAULT_MAX_RETRY_AFTER,
300 max_jitter: DEFAULT_MAX_JITTER,
301 max_pages: DEFAULT_MAX_PAGES,
302 max_response_body_bytes: DEFAULT_MAX_RESPONSE_BODY_BYTES,
303 cache: None,
304 hooks: Arc::new(NoopHooks),
305 }
306 }
307
308 pub fn token_provider(self, provider: impl TokenProvider + 'static) -> ClientBuilder {
310 self.auth_strategy(BearerAuth::new(provider))
311 }
312
313 pub fn access_token(self, token: impl Into<crate::types::SensitiveString>) -> ClientBuilder {
315 self.token_provider(StaticTokenProvider::new(token))
316 }
317
318 pub fn session_token(self, token: impl Into<crate::types::SensitiveString>) -> ClientBuilder {
321 self.auth_strategy(CookieAuth::new(StaticTokenProvider::new(token)))
322 }
323
324 pub fn auth_strategy(mut self, strategy: impl AuthStrategy + 'static) -> ClientBuilder {
326 self.auth = Some(Arc::new(strategy));
327 self
328 }
329
330 pub fn http_client(mut self, http: impl HttpClient + 'static) -> ClientBuilder {
334 self.http = Some(Arc::new(http));
335 self
336 }
337
338 pub fn user_agent(mut self, user_agent: impl Into<String>) -> ClientBuilder {
340 self.user_agent = user_agent.into();
341 self
342 }
343
344 pub fn timeout(mut self, timeout: Duration) -> ClientBuilder {
347 self.timeout = timeout;
348 self
349 }
350
351 pub fn max_attempts(mut self, max_attempts: u32) -> ClientBuilder {
355 self.max_attempts = max_attempts.max(1);
356 self
357 }
358
359 pub fn base_delay(mut self, base_delay: Duration) -> ClientBuilder {
361 self.base_delay = base_delay;
362 self
363 }
364
365 pub fn max_delay(mut self, max_delay: Duration) -> ClientBuilder {
368 self.max_delay = max_delay;
369 self
370 }
371
372 pub fn max_retry_after(mut self, max_retry_after: Duration) -> ClientBuilder {
374 self.max_retry_after = max_retry_after;
375 self
376 }
377
378 pub fn max_jitter(mut self, max_jitter: Duration) -> ClientBuilder {
380 self.max_jitter = max_jitter;
381 self
382 }
383
384 pub fn max_pages(mut self, max_pages: usize) -> ClientBuilder {
386 self.max_pages = max_pages;
387 self
388 }
389
390 pub fn max_response_body_bytes(mut self, bytes: usize) -> ClientBuilder {
393 self.max_response_body_bytes = bytes;
394 self
395 }
396
397 pub fn cache(mut self, cache: impl ResponseCache + 'static) -> ClientBuilder {
400 self.cache = Some(Arc::new(cache));
401 self
402 }
403
404 pub fn hooks(mut self, hooks: impl Hooks + 'static) -> ClientBuilder {
407 self.hooks = Arc::new(hooks);
408 self
409 }
410
411 pub fn build(self) -> Result<Client, Error> {
414 let base_url = parse_base_url(&self.config.base_url)?;
415 let auth = self
416 .auth
417 .ok_or_else(|| Error::usage("a token provider or auth strategy is required"))?;
418 if self.timeout.is_zero() {
419 return Err(Error::usage("timeout must be greater than zero"));
420 }
421 if self.max_pages == 0 {
422 return Err(Error::usage("max pages must be greater than zero"));
423 }
424 let http = match self.http {
425 Some(http) => http,
426 None => shipped_http_client(self.timeout)?,
427 };
428 let cache =
429 match (self.cache, self.config.cache_enabled) {
430 (Some(cache), _) => Some(cache),
431 (None, true) => Some(Arc::new(FileCache::new(self.config.cache_dir.clone()))
432 as Arc<dyn ResponseCache>),
433 (None, false) => None,
434 };
435 let max_response_body_bytes = match self.max_response_body_bytes {
436 0 => DEFAULT_MAX_RESPONSE_BODY_BYTES,
437 bytes => bytes,
438 };
439 let shared = Shared {
440 config: self.config,
441 base_url,
442 http,
443 auth,
444 user_agent: self.user_agent,
445 max_attempts: self.max_attempts,
446 base_delay: self.base_delay.min(self.max_delay),
447 max_delay: self.max_delay,
448 max_retry_after: self.max_retry_after,
449 max_jitter: self.max_jitter,
450 max_pages: self.max_pages,
451 max_response_body_bytes,
452 cache,
453 hooks: self.hooks,
454 };
455 Ok(Client {
456 shared: Arc::new(shared),
457 })
458 }
459}
460
461impl Client {
462 pub fn builder(config: Config) -> ClientBuilder {
464 ClientBuilder::new(config)
465 }
466
467 #[cfg(feature = "reqwest")]
471 #[cfg_attr(docsrs, doc(cfg(feature = "reqwest")))]
472 pub fn new(config: Config, provider: impl TokenProvider + 'static) -> Result<Client, Error> {
473 Client::builder(config).token_provider(provider).build()
474 }
475
476 pub fn config(&self) -> &Config {
478 &self.shared.config
479 }
480
481 pub fn base_url(&self) -> &Url {
483 &self.shared.base_url
484 }
485
486 pub fn max_pages(&self) -> usize {
488 self.shared.max_pages
489 }
490
491 pub fn for_account(&self, account_id: impl Into<String>) -> Result<AccountClient, Error> {
494 let account_id = account_id.into();
495 if account_id.is_empty()
496 || account_id == "."
497 || account_id == ".."
498 || account_id.contains(['/', '?', '#', '%'])
499 || account_id
500 .chars()
501 .any(|c| c.is_whitespace() || c.is_control())
502 {
503 return Err(Error::usage(format!("invalid account id {account_id:?}")));
504 }
505 Ok(AccountClient {
506 client: self.clone(),
507 account_id,
508 })
509 }
510
511 pub fn for_configured_account(&self) -> Result<AccountClient, Error> {
513 match &self.shared.config.account {
514 Some(account_id) => self.for_account(account_id.clone()),
515 None => Err(Error::usage_with_hint(
516 "no account configured",
517 "set FIZZY_ACCOUNT or Config::account, or call Client::for_account",
518 )),
519 }
520 }
521
522 pub fn scope(&self) -> Scope<'_> {
524 Scope {
525 client: self,
526 account_id: None,
527 }
528 }
529
530 pub fn operation(
534 &self,
535 route: &'static Route,
536 params: &[&dyn Display],
537 ) -> Result<Operation, Error> {
538 self.scope().operation(route, params)
539 }
540
541 pub fn request(&self, method: Method, path: impl Into<String>) -> Operation {
544 Operation::raw(method, path.into())
545 }
546
547 pub async fn send<T: DeserializeOwned>(&self, operation: Operation) -> Result<T, Error> {
549 self.execute(operation).await?.json()
550 }
551
552 pub async fn send_unit(&self, operation: Operation) -> Result<(), Error> {
554 self.execute(operation).await.map(|_| ())
555 }
556
557 pub async fn send_page<T: DeserializeOwned>(
559 &self,
560 operation: Operation,
561 ) -> Result<Page<T>, Error> {
562 let info = operation.info.clone();
563 let retry = Some(self.policy_for(&operation));
564 let response = self.execute(operation).await?;
565 Ok(Page::new(response.json()?, &response, info, retry))
566 }
567
568 pub async fn execute(&self, operation: Operation) -> Result<Response, Error> {
572 let work = self.dispatch(&operation);
573 #[cfg(feature = "tracing")]
574 let work = {
575 use tracing::Instrument;
576 let span = tracing::info_span!(
577 "fizzy.operation",
578 operation = %operation.info.operation,
579 service = %operation.info.service,
580 http.status = tracing::field::Empty,
581 request_id = tracing::field::Empty,
582 );
583 async move {
584 let outcome = work.await;
585 let span = tracing::Span::current();
586 match &outcome {
587 Ok(response) => {
588 span.record("http.status", response.status.as_u16());
589 if let Some(id) = response.header("x-request-id") {
590 span.record("request_id", id);
591 }
592 }
593 Err(error) => {
594 if let Some(status) = error.http_status() {
595 span.record("http.status", status);
596 }
597 if let Some(id) = error.request_id() {
598 span.record("request_id", id);
599 }
600 }
601 }
602 outcome
603 }
604 .instrument(span)
605 };
606 self.instrument(&operation, work).await
607 }
608
609 async fn instrument<T>(
617 &self,
618 operation: &Operation,
619 work: impl Future<Output = Result<T, Error>>,
620 ) -> Result<T, Error> {
621 let hooks = &self.shared.hooks;
622 hooks.on_operation_gate(&operation.info).await?;
623
624 let mut running = Running {
625 hooks,
626 info: &operation.info,
627 state: Some(hooks.on_operation_start(&operation.info)),
628 started: Instant::now(),
629 };
630 let outcome = work.await;
631 running.finished(outcome.as_ref().map(|_| ()));
632 outcome
633 }
634
635 async fn dispatch(&self, operation: &Operation) -> Result<Response, Error> {
638 let url = self.url_for(operation)?;
639 let answered = self.attempt(operation, &url).await?;
640 let status = answered.status;
641 let finished = self
642 .finish(
643 operation,
644 &url,
645 answered.url,
646 status,
647 answered.headers,
648 answered.body,
649 answered.cached,
650 )
651 .await;
652 self.shared.hooks.on_request_end(
653 &answered.info,
654 &RequestResult {
655 status: Some(status),
656 duration: answered.duration,
657 error: finished.as_ref().err(),
658 from_cache: finished.as_ref().is_ok_and(|response| response.from_cache),
659 retryable: answered.retryable,
660 retry_after: answered.retry_after,
661 },
662 );
663 finished
664 }
665
666 pub(crate) fn policy_for(&self, operation: &Operation) -> RetryPolicy {
670 let shared = &self.shared;
671 if !operation.idempotent {
672 return RetryPolicy::none();
673 }
674 let policy = operation.retry.clone().unwrap_or_else(|| RetryPolicy {
675 attempts: shared.max_attempts,
676 base_delay: shared.base_delay,
677 retry_on: Cow::Borrowed(DEFAULT_RETRY_ON),
678 });
679 RetryPolicy {
680 attempts: policy.attempts.min(shared.max_attempts).max(1),
681 base_delay: policy.base_delay.min(shared.max_delay),
682 retry_on: policy.retry_on,
683 }
684 }
685
686 async fn attempt(&self, operation: &Operation, url: &Url) -> Result<Answered, Error> {
689 let policy = self.policy_for(operation);
690 let mut backoff = Backoff {
691 attempt: 1,
692 delay: policy.base_delay,
693 };
694 let mut cached = None;
697
698 loop {
699 let once = self.attempt_once(operation, url, &policy, &mut backoff, &mut cached);
700 if let Some(answered) = once.await? {
701 return Ok(answered);
702 }
703 }
704 }
705
706 async fn attempt_once(
709 &self,
710 operation: &Operation,
711 url: &Url,
712 policy: &RetryPolicy,
713 backoff: &mut Backoff,
714 cached: &mut Option<(String, CachedResponse)>,
715 ) -> Result<Option<Answered>, Error> {
716 let hooks = &self.shared.hooks;
717 let attempt = backoff.attempt;
718 let request = self.prepare(operation, url, cached).await?;
719 let info = RequestInfo {
720 method: operation.method.clone(),
721 url: url.clone(),
722 attempt,
723 };
724 hooks.on_request_start(&info);
725 let started = Instant::now();
726 let sent = self.transmit(operation, url.clone(), request).await;
727 let duration = started.elapsed();
728
729 let (final_url, response) = match sent {
730 Err(error) => {
731 let again = error.is_retryable() && attempt < policy.attempts;
732 hooks.on_request_end(
733 &info,
734 &RequestResult::failed(None, duration, &error, error.is_retryable(), None),
735 );
736 return if again {
737 self.resend(backoff, &info, operation, &error, None, "request failed")
738 .await;
739 Ok(None)
740 } else {
741 Err(error)
742 };
743 }
744 Ok(sent) => sent,
745 };
746
747 let status = response.status();
748 let retryable = policy.retry_on.contains(&status.as_u16());
749 let retry_after = retry_after_asked(status, response.headers());
750 let wait = match retry_after {
751 Some(seconds) if seconds > 0 => Some(Duration::from_secs(seconds)),
752 _ => None,
753 };
754 let too_long = wait.is_some_and(|wait| wait > self.shared.max_retry_after);
755 if retryable && attempt < policy.attempts && !too_long {
756 let cause = Error::from_response(status, &operation.method, response.headers(), &[]);
757 hooks.on_request_end(
758 &info,
759 &RequestResult::failed(Some(status), duration, &cause, retryable, retry_after),
760 );
761 self.resend(backoff, &info, operation, &cause, wait, "retryable status")
762 .await;
763 return Ok(None);
764 }
765
766 let (parts, body) = response.into_parts();
767 match self.read_answer(operation, url, status, body).await {
768 Ok(body) => Ok(Some(Answered {
769 url: final_url,
770 status,
771 headers: parts.headers,
772 body,
773 cached: cached.take(),
774 info,
775 duration,
776 retryable,
777 retry_after,
778 })),
779 Err(error) => {
780 let (error, retryable) = unread(operation, status, &parts.headers, error);
781 hooks.on_request_end(
782 &info,
783 &RequestResult::failed(Some(status), duration, &error, retryable, retry_after),
784 );
785 if retryable && attempt < policy.attempts {
786 self.resend(backoff, &info, operation, &error, None, "body broke off")
787 .await;
788 Ok(None)
789 } else {
790 Err(error)
791 }
792 }
793 }
794 }
795
796 async fn resend(
799 &self,
800 backoff: &mut Backoff,
801 info: &RequestInfo,
802 operation: &Operation,
803 error: &Error,
804 wait: Option<Duration>,
805 why: &str,
806 ) {
807 crate::trace::debug(&operation.id, backoff.attempt, &format!("{why}, retrying"));
808 self.shared.hooks.on_retry(info, backoff.attempt + 1, error);
809 self.wait(wait.unwrap_or(backoff.delay)).await;
810 backoff.delay = self.next_delay(backoff.delay);
811 backoff.attempt += 1;
812 }
813
814 async fn read_answer(
817 &self,
818 operation: &Operation,
819 url: &Url,
820 status: StatusCode,
821 body: Body,
822 ) -> Result<Bytes, Error> {
823 if status.is_success() {
824 let bound = self.shared.max_response_body_bytes;
825 read_body(body, bound, &operation.method, url.path()).await
826 } else {
827 body.prefix(MAX_ERROR_BODY_BYTES).await
828 }
829 }
830
831 pub(crate) fn url_for(&self, operation: &Operation) -> Result<Url, Error> {
835 let mut url = match &operation.url {
836 Some(url) => url.clone(),
837 None => self
838 .shared
839 .base_url
840 .join(operation.path.trim_start_matches('/'))?,
841 };
842 if !operation.query.is_empty() {
843 url.query_pairs_mut().extend_pairs(&operation.query);
844 }
845 require_secure_endpoint(&url)?;
846 if !url.username().is_empty() || url.password().is_some() {
847 return Err(Error::usage(format!(
848 "{} names a URL carrying credentials; use an access token or a session token",
849 operation.id
850 )));
851 }
852 if !is_same_origin(&url, &self.shared.base_url) {
853 return Err(Error::usage(format!(
854 "{} resolves off the Fizzy origin {}, onto {}",
855 operation.id,
856 self.shared.base_url.origin().ascii_serialization(),
857 url.origin().ascii_serialization()
858 )));
859 }
860 Ok(url)
861 }
862
863 async fn prepare(
867 &self,
868 operation: &Operation,
869 url: &Url,
870 cached: &mut Option<(String, CachedResponse)>,
871 ) -> Result<Request<Bytes>, Error> {
872 let mut request = Request::builder()
873 .method(operation.method.clone())
874 .uri(url.as_str())
875 .body(Bytes::new())
876 .map_err(Error::from_std)?;
877 let headers = request.headers_mut();
878 headers.insert(USER_AGENT, header_value(&self.shared.user_agent)?);
879 headers.insert(ACCEPT, HeaderValue::from_static("application/json"));
880 for (name, value) in &operation.headers {
881 headers.append(name.clone(), value.clone());
882 }
883 if let Some(body) = &operation.body {
884 headers.insert(CONTENT_TYPE, header_value(&body.content_type)?);
885 *request.body_mut() = body.bytes.clone();
886 }
887 self.shared.auth.authenticate(&mut request).await?;
888
889 let key = match self.cacheable(operation) {
890 None => None,
891 Some(cache) => match credential(request.headers()) {
892 None => None,
893 Some(credential) => {
894 let key = cache_key(url.as_str(), &credential);
895 if cached.as_ref().is_none_or(|(held, _)| *held != key) {
896 *cached = self.look_up(cache, &key).await;
897 }
898 Some(key)
899 }
900 },
901 };
902 if key.is_none() {
903 *cached = None;
904 }
905 if let Some((_, entry)) = cached.as_ref()
906 && !entry.etag.is_empty()
907 {
908 let validator = header_value(&entry.etag)?;
909 request.headers_mut().insert(IF_NONE_MATCH, validator);
910 }
911 Ok(request)
912 }
913
914 async fn look_up(
918 &self,
919 cache: &Arc<dyn ResponseCache>,
920 key: &str,
921 ) -> Option<(String, CachedResponse)> {
922 match cache_get(cache, key).await {
923 Some(entry) if entry.body.len() <= self.shared.max_response_body_bytes => {
924 Some((key.to_string(), entry))
925 }
926 Some(_) => {
927 cache_invalidate(cache, key).await;
928 None
929 }
930 None => Some((
931 key.to_string(),
932 CachedResponse {
933 etag: String::new(),
934 body: Bytes::new(),
935 },
936 )),
937 }
938 }
939
940 fn cacheable(&self, operation: &Operation) -> Option<&Arc<dyn ResponseCache>> {
944 if operation.no_cache || operation.method != Method::GET {
945 None
946 } else {
947 self.shared.cache.as_ref()
948 }
949 }
950
951 async fn transmit(
961 &self,
962 operation: &Operation,
963 mut url: Url,
964 mut request: Request<Bytes>,
965 ) -> Result<(Url, HttpResponse<Body>), Error> {
966 let mut hops = 0;
967 loop {
968 let outgoing = (
969 request.method().clone(),
970 request.headers().clone(),
971 request.body().clone(),
972 );
973 let response = self.shared.http.send(request).await?;
974 match redirect_target(&url, &response) {
975 None => return Ok((url, response)),
976 Some(_) if hops == MAX_REDIRECTS => {
977 return Err(Error::new(
978 ErrorCode::Network,
979 format!(
980 "{} redirected more than {MAX_REDIRECTS} times",
981 operation.id
982 ),
983 )
984 .retryable());
985 }
986 Some(next) => {
987 require_secure_endpoint(&next)?;
988 if !next.username().is_empty() || next.password().is_some() {
989 return Err(Error::usage(format!(
990 "{} redirected to a URL carrying credentials",
991 operation.id
992 )));
993 }
994 if !is_same_origin(&next, &self.shared.base_url) {
995 return Err(Error::usage(format!(
996 "{} redirected off the Fizzy origin to {}",
997 operation.id,
998 next.origin().ascii_serialization()
999 )));
1000 }
1001 request = redirected(outgoing, response.status(), &next)?;
1002 url = next;
1003 hops += 1;
1004 }
1005 }
1006 }
1007 }
1008
1009 #[allow(clippy::too_many_arguments)]
1010 async fn finish(
1011 &self,
1012 operation: &Operation,
1013 url: &Url,
1014 final_url: Url,
1015 status: StatusCode,
1016 headers: HeaderMap,
1017 body: Bytes,
1018 cached: Option<(String, CachedResponse)>,
1019 ) -> Result<Response, Error> {
1020 let cached = if final_url == *url { cached } else { None };
1023
1024 if status == StatusCode::NOT_MODIFIED {
1025 return match cached {
1026 Some((_, entry)) if !entry.etag.is_empty() => Ok(Response {
1027 status: StatusCode::OK,
1028 headers,
1029 body: entry.body,
1030 url: final_url,
1031 from_cache: true,
1032 }),
1033 _ => Err(Error::api(
1034 304,
1035 "304 received but no cached response available",
1036 )),
1037 };
1038 }
1039
1040 if status.is_success() {
1041 if let (Some((key, _)), Some(cache)) = (cached, self.cacheable(operation))
1042 && let Some(etag) = headers.get("etag").and_then(|value| value.to_str().ok())
1043 {
1044 cache_set(
1045 cache,
1046 &key,
1047 CachedResponse {
1048 etag: etag.to_string(),
1049 body: body.clone(),
1050 },
1051 )
1052 .await;
1053 }
1054 Ok(Response {
1055 status,
1056 headers,
1057 body,
1058 url: final_url,
1059 from_cache: false,
1060 })
1061 } else {
1062 Err(Error::from_response(
1063 status,
1064 &operation.method,
1065 &headers,
1066 &body,
1067 ))
1068 }
1069 }
1070
1071 async fn wait(&self, delay: Duration) {
1072 let jitter = match u64::try_from(self.shared.max_jitter.as_millis()).unwrap_or(u64::MAX) {
1073 0 => Duration::ZERO,
1074 millis => Duration::from_millis(rand::random_range(0..millis)),
1075 };
1076 tokio::time::sleep(delay + jitter).await;
1077 }
1078
1079 fn next_delay(&self, delay: Duration) -> Duration {
1080 (delay * 2).min(self.shared.max_delay)
1081 }
1082}
1083
1084struct Running<'a> {
1088 hooks: &'a Arc<dyn Hooks>,
1089 info: &'a OperationInfo,
1090 state: Option<OperationState>,
1091 started: Instant,
1092}
1093
1094impl Running<'_> {
1095 fn finished(&mut self, outcome: Result<(), &Error>) {
1096 if let Some(state) = self.state.take() {
1097 self.hooks
1098 .on_operation_end(self.info, state, outcome, self.started.elapsed());
1099 }
1100 }
1101}
1102
1103impl Drop for Running<'_> {
1104 fn drop(&mut self) {
1105 if self.state.is_some() {
1106 self.finished(Err(&Error::cancelled()));
1107 }
1108 }
1109}
1110
1111struct Answered {
1115 url: Url,
1116 status: StatusCode,
1117 headers: HeaderMap,
1118 body: Bytes,
1119 cached: Option<(String, CachedResponse)>,
1120 info: RequestInfo,
1121 duration: Duration,
1122 retryable: bool,
1123 retry_after: Option<u64>,
1124}
1125
1126fn credential(headers: &HeaderMap) -> Option<String> {
1128 headers
1129 .get(AUTHORIZATION)
1130 .or_else(|| headers.get(COOKIE))
1131 .and_then(|value| value.to_str().ok())
1132 .map(str::to_string)
1133}
1134
1135async fn cache_get(cache: &Arc<dyn ResponseCache>, key: &str) -> Option<CachedResponse> {
1140 let cache = cache.clone();
1141 let key = key.to_string();
1142 tokio::task::spawn_blocking(move || cache.get(&key))
1143 .await
1144 .ok()
1145 .flatten()
1146}
1147
1148async fn cache_set(cache: &Arc<dyn ResponseCache>, key: &str, response: CachedResponse) {
1149 let cache = cache.clone();
1150 let key = key.to_string();
1151 let _ = tokio::task::spawn_blocking(move || cache.set(&key, response)).await;
1152}
1153
1154async fn cache_invalidate(cache: &Arc<dyn ResponseCache>, key: &str) {
1155 let cache = cache.clone();
1156 let key = key.to_string();
1157 let _ = tokio::task::spawn_blocking(move || cache.invalidate(&key)).await;
1158}
1159
1160#[cfg(feature = "reqwest")]
1161fn shipped_http_client(timeout: Duration) -> Result<Arc<dyn HttpClient>, Error> {
1162 Ok(Arc::new(crate::http::ReqwestClient::with_timeout(timeout)?))
1163}
1164
1165#[cfg(not(feature = "reqwest"))]
1166fn shipped_http_client(_timeout: Duration) -> Result<Arc<dyn HttpClient>, Error> {
1167 Err(Error::usage(
1168 "no HTTP client: supply one with ClientBuilder::http_client, or enable the reqwest feature",
1169 ))
1170}
1171
1172fn redirect_target(url: &Url, response: &HttpResponse<Body>) -> Option<Url> {
1175 let status = response.status();
1176 if status.is_redirection() && status != StatusCode::NOT_MODIFIED {
1177 response
1178 .headers()
1179 .get("location")
1180 .and_then(|value| value.to_str().ok())
1181 .and_then(|location| url.join(location).ok())
1182 } else {
1183 None
1184 }
1185}
1186
1187fn redirected(
1190 (method, mut headers, body): (Method, HeaderMap, Bytes),
1191 status: StatusCode,
1192 next: &Url,
1193) -> Result<Request<Bytes>, Error> {
1194 headers.remove(IF_NONE_MATCH);
1195 let keeps_method = method == Method::GET
1196 || method == Method::HEAD
1197 || status == StatusCode::TEMPORARY_REDIRECT
1198 || status == StatusCode::PERMANENT_REDIRECT;
1199 let (method, body) = if keeps_method {
1200 (method, body)
1201 } else {
1202 headers.remove(CONTENT_TYPE);
1203 headers.remove(CONTENT_LENGTH);
1204 (Method::GET, Bytes::new())
1205 };
1206 let mut request = Request::builder()
1207 .method(method)
1208 .uri(next.as_str())
1209 .body(body)
1210 .map_err(Error::from_std)?;
1211 *request.headers_mut() = headers;
1212 Ok(request)
1213}
1214
1215fn parse_base_url(base_url: &str) -> Result<Url, Error> {
1216 let mut url = Url::parse(base_url)
1217 .map_err(|error| Error::usage(format!("base URL {base_url}: {error}")))?;
1218 require_secure_endpoint(&url)?;
1219 if !url.username().is_empty() || url.password().is_some() {
1220 return Err(Error::usage(
1221 "base URL must not carry credentials; use an access token or a session token",
1222 ));
1223 }
1224 if !url.path().ends_with('/') {
1225 url.set_path(&format!("{}/", url.path()));
1226 }
1227 Ok(url)
1228}
1229
1230fn retry_after_asked(status: StatusCode, headers: &HeaderMap) -> Option<u64> {
1232 if status == StatusCode::TOO_MANY_REQUESTS || status == StatusCode::SERVICE_UNAVAILABLE {
1233 retry_after_seconds(headers)
1234 } else {
1235 None
1236 }
1237}
1238
1239fn header_value(value: &str) -> Result<HeaderValue, Error> {
1240 HeaderValue::from_str(value)
1241 .map_err(|_| Error::usage(format!("{value:?} is not a valid header value")))
1242}
1243
1244fn unread(
1250 operation: &Operation,
1251 status: StatusCode,
1252 headers: &HeaderMap,
1253 error: Error,
1254) -> (Error, bool) {
1255 let retryable = status.is_success() && error.is_retryable();
1256 let error = if status.is_success() {
1257 error
1258 } else {
1259 Error::from_response(status, &operation.method, headers, &[]).refusing(error)
1260 };
1261 (error, retryable)
1262}
1263
1264struct Backoff {
1266 attempt: u32,
1267 delay: Duration,
1268}
1269
1270pub(crate) async fn read_body(
1271 body: Body,
1272 limit: usize,
1273 method: &Method,
1274 path: &str,
1275) -> Result<Bytes, Error> {
1276 body.collect(limit, || Error::response_too_large(limit, method, path))
1277 .await
1278}
1279
1280#[cfg(test)]
1281#[allow(clippy::unwrap_used)]
1282mod tests {
1283 use std::sync::Mutex;
1284
1285 use async_trait::async_trait;
1286 use serde_json::Value;
1287
1288 use super::*;
1289 use crate::auth::StaticTokenProvider;
1290
1291 struct Canned {
1295 answer: Box<Answer>,
1296 sent: Mutex<Vec<(Method, String, HeaderMap)>>,
1297 }
1298
1299 type Answer = dyn Fn(&Request<Bytes>) -> HttpResponse<Body> + Send + Sync;
1300
1301 impl Canned {
1302 fn new(
1303 answer: impl Fn(&Request<Bytes>) -> HttpResponse<Body> + Send + Sync + 'static,
1304 ) -> Arc<Canned> {
1305 Arc::new(Canned {
1306 answer: Box::new(answer),
1307 sent: Mutex::new(Vec::new()),
1308 })
1309 }
1310
1311 fn sent(&self) -> Vec<(Method, String, HeaderMap)> {
1312 self.sent.lock().unwrap().clone()
1313 }
1314 }
1315
1316 #[async_trait]
1317 impl HttpClient for Arc<Canned> {
1318 async fn send(&self, request: Request<Bytes>) -> Result<HttpResponse<Body>, Error> {
1319 self.sent.lock().unwrap().push((
1320 request.method().clone(),
1321 request.uri().to_string(),
1322 request.headers().clone(),
1323 ));
1324 Ok((self.answer)(&request))
1325 }
1326 }
1327
1328 fn answer(status: u16, body: &'static str) -> HttpResponse<Body> {
1329 let mut response = HttpResponse::new(Body::from(body));
1330 *response.status_mut() = StatusCode::from_u16(status).unwrap();
1331 response
1332 }
1333
1334 fn redirect(location: &str) -> HttpResponse<Body> {
1335 let mut response = answer(302, "");
1336 response
1337 .headers_mut()
1338 .insert("location", HeaderValue::from_str(location).unwrap());
1339 response
1340 }
1341
1342 fn client_over(http: Arc<Canned>) -> Client {
1343 Client::builder(Config::default().with_base_url("https://fizzy.test"))
1344 .token_provider(StaticTokenProvider::new("secret"))
1345 .http_client(http)
1346 .max_attempts(1)
1347 .build()
1348 .unwrap()
1349 }
1350
1351 #[tokio::test]
1352 async fn a_request_goes_out_on_the_supplied_http_client_with_credentials() {
1353 let http = Canned::new(|_| answer(200, r#"{"ok":true}"#));
1354 let client = client_over(http.clone());
1355
1356 let body: Value = client
1357 .send(client.request(Method::GET, "/my/identity.json"))
1358 .await
1359 .unwrap();
1360
1361 assert_eq!(body, serde_json::json!({ "ok": true }));
1362 let sent = http.sent();
1363 assert_eq!(sent.len(), 1);
1364 assert_eq!(sent[0].0, Method::GET);
1365 assert_eq!(sent[0].1, "https://fizzy.test/my/identity.json");
1366 assert_eq!(sent[0].2[AUTHORIZATION], "Bearer secret");
1367 }
1368
1369 #[tokio::test]
1370 async fn a_redirect_on_the_same_origin_is_followed_with_credentials() {
1371 let http = Canned::new(|request| {
1372 if request.uri().path() == "/old.json" {
1373 redirect("/new.json")
1374 } else {
1375 answer(200, r#"{"moved":true}"#)
1376 }
1377 });
1378 let client = client_over(http.clone());
1379
1380 let response = client
1381 .execute(client.request(Method::GET, "/old.json"))
1382 .await
1383 .unwrap();
1384
1385 assert_eq!(response.url.as_str(), "https://fizzy.test/new.json");
1386 assert_eq!(response.body, r#"{"moved":true}"#);
1387 let sent = http.sent();
1388 assert_eq!(sent.len(), 2);
1389 assert_eq!(sent[1].2[AUTHORIZATION], "Bearer secret");
1390 }
1391
1392 #[tokio::test]
1393 async fn a_redirect_off_the_origin_is_refused_with_nothing_sent_there() {
1394 let http = Canned::new(|_| redirect("https://storage.test/blobs/1"));
1395 let client = client_over(http.clone());
1396
1397 let error = client.get("/blobs/1").await.unwrap_err();
1398
1399 assert_eq!(error.code(), ErrorCode::Usage);
1400 assert_eq!(http.sent().len(), 1);
1401 }
1402
1403 fn broken_body(status: u16) -> HttpResponse<Body> {
1404 let chunks = futures_util::stream::iter([
1405 Ok(Bytes::from_static(b"{\"ok\":")),
1406 Err(Error::new(ErrorCode::Network, "cut off").retryable()),
1407 ]);
1408 let mut response = HttpResponse::new(Body::from_stream(chunks, None));
1409 *response.status_mut() = StatusCode::from_u16(status).unwrap();
1410 response
1411 }
1412
1413 #[tokio::test]
1414 async fn a_body_that_breaks_off_is_asked_for_again() {
1415 let calls = Arc::new(Mutex::new(0));
1416 let seen = calls.clone();
1417 let http = Canned::new(move |_| {
1418 let mut calls = seen.lock().unwrap();
1419 *calls += 1;
1420 if *calls == 1 {
1421 broken_body(200)
1422 } else {
1423 answer(200, r#"{"ok":true}"#)
1424 }
1425 });
1426 let client = Client::builder(Config::default().with_base_url("https://fizzy.test"))
1427 .token_provider(StaticTokenProvider::new("secret"))
1428 .http_client(http.clone())
1429 .max_attempts(2)
1430 .max_jitter(Duration::ZERO)
1431 .base_delay(Duration::from_millis(1))
1432 .build()
1433 .unwrap();
1434
1435 let response = client.get("/x.json").await.unwrap();
1436
1437 assert_eq!(response.body, r#"{"ok":true}"#);
1438 assert_eq!(http.sent().len(), 2);
1439 }
1440
1441 #[tokio::test]
1442 async fn a_failure_whose_body_breaks_off_keeps_its_status_and_is_not_resent() {
1443 let http = Canned::new(|_| broken_body(422));
1444 let client = Client::builder(Config::default().with_base_url("https://fizzy.test"))
1445 .token_provider(StaticTokenProvider::new("secret"))
1446 .http_client(http.clone())
1447 .max_attempts(2)
1448 .max_jitter(Duration::ZERO)
1449 .base_delay(Duration::from_millis(1))
1450 .build()
1451 .unwrap();
1452
1453 let error = client.get("/x.json").await.unwrap_err();
1454
1455 assert_eq!(error.code(), ErrorCode::Validation);
1456 assert_eq!(error.http_status(), Some(422));
1457 assert_eq!(http.sent().len(), 1);
1458 }
1459
1460 #[derive(Default)]
1462 struct Retryability(Mutex<Vec<bool>>);
1463
1464 impl Hooks for Retryability {
1465 fn on_request_end(&self, _info: &RequestInfo, result: &RequestResult<'_>) {
1466 self.0.lock().unwrap().push(result.retryable);
1467 }
1468 }
1469
1470 #[tokio::test]
1471 async fn a_body_that_breaks_off_on_the_last_attempt_is_still_reported_resendable() {
1472 let http = Canned::new(|_| broken_body(200));
1473 let seen = Arc::new(Retryability::default());
1474 let client = Client::builder(Config::default().with_base_url("https://fizzy.test"))
1475 .token_provider(StaticTokenProvider::new("secret"))
1476 .http_client(http.clone())
1477 .hooks(seen.clone())
1478 .max_attempts(2)
1479 .max_jitter(Duration::ZERO)
1480 .base_delay(Duration::from_millis(1))
1481 .build()
1482 .unwrap();
1483
1484 let error = client.get("/x.json").await.unwrap_err();
1485
1486 assert_eq!(error.code(), ErrorCode::Network);
1487 assert_eq!(http.sent().len(), 2);
1488 assert_eq!(*seen.0.lock().unwrap(), [true, true]);
1489 }
1490
1491 #[tokio::test]
1492 async fn a_redirect_carrying_credentials_is_refused_without_echoing_them() {
1493 let http = Canned::new(|_| redirect("https://user:s3cret@fizzy.test/new.json"));
1494 let client = client_over(http.clone());
1495
1496 let error = client.get("/old.json").await.unwrap_err();
1497
1498 assert_eq!(error.code(), ErrorCode::Usage);
1499 assert!(!error.to_string().contains("s3cret"));
1500 assert_eq!(http.sent().len(), 1);
1501 }
1502
1503 #[tokio::test]
1504 async fn a_path_that_resolves_off_the_origin_is_refused_before_anything_is_sent() {
1505 let http = Canned::new(|_| answer(200, "{}"));
1506 let client = client_over(http.clone());
1507
1508 for path in [
1509 "http://evil.test/x",
1510 "HTTP://evil.test/x",
1511 "https://evil.test/x",
1512 ] {
1513 let error = client.get(path).await.unwrap_err();
1514 assert_eq!(error.code(), ErrorCode::Usage, "{path}");
1515 let error = client
1516 .execute(client.request(Method::GET, path))
1517 .await
1518 .unwrap_err();
1519 assert_eq!(error.code(), ErrorCode::Usage, "{path}");
1520 }
1521 assert!(http.sent().is_empty());
1522 }
1523
1524 #[tokio::test]
1525 async fn a_url_carrying_userinfo_is_refused_without_echoing_it() {
1526 let http = Canned::new(|_| answer(200, "{}"));
1527 let client = client_over(http.clone());
1528
1529 let error = client
1530 .get("https://user:s3cret@fizzy.test/x")
1531 .await
1532 .unwrap_err();
1533
1534 assert_eq!(error.code(), ErrorCode::Usage);
1535 assert!(!error.to_string().contains("s3cret"));
1536 assert!(http.sent().is_empty());
1537 }
1538
1539 #[tokio::test]
1540 async fn a_refused_redirect_is_not_retried() {
1541 let http = Canned::new(|_| redirect("http://evil.test/"));
1542 let client = Client::builder(Config::default().with_base_url("https://fizzy.test"))
1543 .token_provider(StaticTokenProvider::new("secret"))
1544 .http_client(http.clone())
1545 .max_jitter(Duration::ZERO)
1546 .base_delay(Duration::from_millis(1))
1547 .build()
1548 .unwrap();
1549
1550 let error = client.get("/anything").await.unwrap_err();
1551
1552 assert_eq!(error.code(), ErrorCode::Usage);
1553 assert_eq!(http.sent().len(), 1);
1554 }
1555
1556 #[test]
1557 fn a_response_and_an_operation_print_without_their_secrets() {
1558 let mut headers = HeaderMap::new();
1559 headers.insert(
1560 "set-cookie",
1561 HeaderValue::from_static("session_token=s3cret; HttpOnly"),
1562 );
1563 let response = Response {
1564 status: StatusCode::OK,
1565 headers,
1566 body: Bytes::from_static(br#"{"email_address":"jane@example.com"}"#),
1567 url: Url::parse("https://fizzy.test/x").unwrap(),
1568 from_cache: false,
1569 };
1570 let printed = format!("{response:?}");
1571 assert!(!printed.contains("s3cret"));
1572 assert!(!printed.contains("jane@example.com"));
1573 assert!(printed.contains("[REDACTED]"));
1574
1575 let mut operation = Operation::raw(Method::POST, "/session.json".into());
1576 operation
1577 .json(&serde_json::json!({"email_address": "jane@example.com"}))
1578 .unwrap();
1579 let printed = format!("{operation:?}");
1580 assert!(!printed.contains("jane@example.com"));
1581 assert!(printed.contains("len"));
1582 }
1583
1584 #[tokio::test]
1585 async fn a_redirect_to_plain_http_elsewhere_is_refused() {
1586 let http = Canned::new(|_| redirect("http://evil.test/"));
1587 let client = client_over(http.clone());
1588
1589 let error = client.get("/anything").await.unwrap_err();
1590
1591 assert_eq!(error.code(), ErrorCode::Usage);
1592 assert_eq!(http.sent().len(), 1);
1593 }
1594
1595 #[tokio::test]
1596 async fn a_redirect_loop_is_given_up_on() {
1597 let http = Canned::new(|_| redirect("/again"));
1598 let client = client_over(http.clone());
1599
1600 let error = client.get("/again").await.unwrap_err();
1601
1602 assert_eq!(error.code(), ErrorCode::Network);
1603 assert_eq!(http.sent().len(), MAX_REDIRECTS + 1);
1604 }
1605
1606 #[test]
1607 fn base_url_must_be_https_or_local() {
1608 assert!(parse_base_url("https://fizzy.do").is_ok());
1609 assert!(parse_base_url("http://127.0.0.1:3000").is_ok());
1610 assert_eq!(
1611 parse_base_url("http://evil.example.com")
1612 .unwrap_err()
1613 .code(),
1614 ErrorCode::Usage
1615 );
1616 assert_eq!(
1617 parse_base_url("https://user:secret@fizzy.do")
1618 .unwrap_err()
1619 .code(),
1620 ErrorCode::Usage
1621 );
1622 }
1623
1624 #[test]
1625 fn an_account_id_has_to_fit_in_a_path() {
1626 let http = Canned::new(|_| answer(200, "[]"));
1627 let client = client_over(http);
1628 assert!(client.for_account("999").is_ok());
1629 assert!(client.for_account("").is_err());
1630 assert!(client.for_account("a/b").is_err());
1631 }
1632}