solari/
http.rs

1//! Shared HTTP transport for SDK ⇆ Gateway REST. Owns auth headers, error
2//! mapping, retries/backoff, idempotency keys, and timeouts. Mirrors `http.ts`.
3//!
4//! Retry policy: idempotent requests (GET, DELETE, or any request carrying an
5//! Idempotency-Key) retry on network errors, HTTP 5xx, and bodies flagged
6//! `retryable`, with exponential backoff + jitter. 429 is NOT retried (org at
7//! its session cap). Non-idempotent writes are never silently retried.
8
9use std::sync::atomic::{AtomicU64, Ordering};
10use std::time::{Duration, SystemTime, UNIX_EPOCH};
11
12use reqwest::Method;
13use serde::de::DeserializeOwned;
14use serde_json::Value;
15
16use crate::error::{map_gateway_error, SolariError};
17use crate::types::GatewayErrorBody;
18
19/// JS `encodeURIComponent`: leaves `A-Za-z0-9 - _ . ! ~ * ' ( )` unescaped,
20/// percent-encodes every other byte as `%XX` over its UTF-8 bytes.
21///
22/// USE THIS FOR EVERY ID INTERPOLATED INTO A PATH. Session ids are
23/// `<poolId>:<vmId>:<orgId>.<sig>`, and this crate previously built paths with a
24/// bare `format!("/sandboxes/{sandbox_id}")` — no escaping at all. That happened
25/// to work (colons are legal path characters, and the id alphabet is otherwise
26/// URL-safe), but it left the id's bytes at the mercy of whatever the gateway
27/// hands back, and it made Rust's URLs differ byte-for-byte from the TypeScript
28/// reference, which the cross-language wire contract test cannot look away from.
29pub(crate) fn encode_uri_component(s: &str) -> String {
30    let mut out = String::with_capacity(s.len());
31    for &b in s.as_bytes() {
32        let c = b as char;
33        if c.is_ascii_alphanumeric()
34            || matches!(c, '-' | '_' | '.' | '!' | '~' | '*' | '\'' | '(' | ')')
35        {
36            out.push(c);
37        } else {
38            out.push('%');
39            out.push_str(&format!("{b:02X}"));
40        }
41    }
42    out
43}
44
45/// Options for constructing an [`HttpTransport`].
46#[derive(Debug, Clone)]
47pub struct HttpOptions {
48    pub api_key: String,
49    pub base_url: String,
50    /// Max retry attempts for idempotent requests. Default 5.
51    pub max_retries: u32,
52    /// Per-request timeout. Default 300s.
53    pub request_timeout: Duration,
54    /// Fixed retry delay. When `None`, uses exponential backoff + jitter
55    /// (mainly for tests: pass `Some(0)` to make retries instant/deterministic).
56    pub retry_delay_ms: Option<u64>,
57}
58
59impl HttpOptions {
60    pub fn new(api_key: impl Into<String>, base_url: impl Into<String>) -> Self {
61        HttpOptions {
62            api_key: api_key.into(),
63            base_url: base_url.into(),
64            max_retries: 5,
65            request_timeout: Duration::from_millis(300_000),
66            retry_delay_ms: None,
67        }
68    }
69}
70
71pub struct HttpTransport {
72    api_key: String,
73    base_url: String,
74    client: reqwest::Client,
75    max_retries: u32,
76    retry_delay_ms: Option<u64>,
77}
78
79impl HttpTransport {
80    pub fn new(opts: HttpOptions) -> Result<Self, SolariError> {
81        if opts.api_key.is_empty() {
82            return Err(SolariError::Other("HttpTransport requires an apiKey".into()));
83        }
84        if opts.base_url.is_empty() {
85            return Err(SolariError::Other("HttpTransport requires a baseUrl".into()));
86        }
87        let base_url = opts.base_url.trim_end_matches('/').to_string();
88        let client = reqwest::Client::builder()
89            .timeout(opts.request_timeout)
90            .build()
91            .map_err(|e| SolariError::Other(format!("failed to build HTTP client: {e}")))?;
92        Ok(HttpTransport {
93            api_key: opts.api_key,
94            base_url,
95            client,
96            max_retries: opts.max_retries,
97            retry_delay_ms: opts.retry_delay_ms,
98        })
99    }
100
101    pub fn api_key(&self) -> &str {
102        &self.api_key
103    }
104
105    /// `Authorization` header for handles + WS upgrades.
106    pub fn auth_headers(&self) -> Vec<(String, String)> {
107        vec![("Authorization".into(), format!("Bearer {}", self.api_key))]
108    }
109
110    /// ws:// (or wss://) origin derived from the gateway base URL.
111    pub fn ws_origin(&self) -> String {
112        // Mirrors TS `.replace(/^http/, "ws")`: http→ws, https→wss.
113        self.base_url.replacen("http", "ws", 1)
114    }
115
116    pub async fn request<T: DeserializeOwned>(
117        &self,
118        method: &str,
119        path: &str,
120        body: Option<Value>,
121        idempotency_key: Option<String>,
122    ) -> Result<T, SolariError> {
123        let idempotent =
124            method == "GET" || method == "DELETE" || idempotency_key.is_some();
125        let m = Method::from_bytes(method.as_bytes())
126            .map_err(|e| SolariError::Other(format!("bad method {method}: {e}")))?;
127        let url = format!("{}{}", self.base_url, path);
128        let body_str = body.as_ref().map(|b| serde_json::to_string(b).unwrap());
129
130        let mut attempt: u32 = 0;
131        loop {
132            let mut rb = self
133                .client
134                .request(m.clone(), &url)
135                .header("Authorization", format!("Bearer {}", self.api_key))
136                .header("Accept", "application/json");
137            if let Some(k) = &idempotency_key {
138                rb = rb.header("Idempotency-Key", k);
139            }
140            if let Some(b) = &body_str {
141                rb = rb.header("Content-Type", "application/json").body(b.clone());
142            } else if matches!(method, "POST" | "PUT" | "PATCH") {
143                // reqwest sends NO Content-Length for a body-less POST, and the Google front end answers 411 Length
144                // Required (pause/resume/start failed against the real gateway). Say "empty body" explicitly.
145                rb = rb.header("Content-Length", "0");
146            }
147
148            let resp = match rb.send().await {
149                Ok(r) => r,
150                Err(err) => {
151                    if idempotent && attempt < self.max_retries {
152                        tokio::time::sleep(self.backoff(attempt)).await;
153                        attempt += 1;
154                        continue;
155                    }
156                    return Err(SolariError::connection(format!(
157                        "{method} {path} failed: {err}"
158                    )));
159                }
160            };
161
162            let status = resp.status().as_u16();
163            if !(200..300).contains(&status) {
164                let text = resp.text().await.unwrap_or_default();
165                let err_body: Option<GatewayErrorBody> =
166                    if text.is_empty() { None } else { serde_json::from_str(&text).ok() };
167                // 🚨 ALLOWLIST, NOT `>= 500` — a 5xx is not automatically transient.
168                // The gateway's seven 501 sites (every /volumes route, permanently
169                // unavailable on GCP) were each retried 5x with backoff; the curve is
170                // visible in prod LB logs (04:47:11, +24s, +47s — doubling, which
171                // separates a retry loop from an evenly-spaced poll).
172                // 🪤 507 is named here rather than left to the `retryable` hint: the
173                // hint would cover it today, but that ties this SDK's correctness to
174                // which gateway build the caller reaches. 500 excluded (an unhandled
175                // error is not known-transient); 429 excluded as before.
176                let retryable = matches!(status, 502 | 503 | 504 | 507)
177                    || err_body.as_ref().and_then(|b| b.retryable).unwrap_or(false);
178                if idempotent && retryable && attempt < self.max_retries {
179                    tokio::time::sleep(self.backoff(attempt)).await;
180                    attempt += 1;
181                    continue;
182                }
183                return Err(map_gateway_error(status, err_body));
184            }
185
186            let text = resp.text().await.unwrap_or_default();
187            if text.is_empty() {
188                return serde_json::from_str("null")
189                    .map_err(|e| SolariError::Other(format!("empty body decode: {e}")));
190            }
191            return serde_json::from_str(&text)
192                .map_err(|e| SolariError::Other(format!("bad response body: {e}")));
193        }
194    }
195
196    fn backoff(&self, attempt: u32) -> Duration {
197        if let Some(ms) = self.retry_delay_ms {
198            return Duration::from_millis(ms);
199        }
200        // 150·2^n capped at 8s, plus rand(0..250)ms jitter. Mirrors http.ts.
201        let base = (150u64.saturating_mul(1u64 << attempt.min(20))).min(8000);
202        Duration::from_millis(base + jitter_ms())
203    }
204}
205
206/// A fresh idempotency key. Not a real UUID (no `uuid` dep) but unique + opaque,
207/// which is all the gateway requires. Format resembles a v4 UUID.
208pub fn new_idempotency_key() -> String {
209    static COUNTER: AtomicU64 = AtomicU64::new(0);
210    let n = COUNTER.fetch_add(1, Ordering::Relaxed);
211    let t = SystemTime::now()
212        .duration_since(UNIX_EPOCH)
213        .map(|d| d.as_nanos() as u64)
214        .unwrap_or(0);
215    let a = t ^ n.rotate_left(17) ^ 0x9E37_79B9_7F4A_7C15;
216    let b = t.rotate_left(31) ^ n.wrapping_mul(0x2545_F491_4F6C_DD1D);
217    format!(
218        "{:08x}-{:04x}-4{:03x}-{:04x}-{:012x}",
219        (a >> 32) as u32,
220        (a >> 16) as u16,
221        (a & 0x0fff) as u16,
222        ((b >> 48) as u16 & 0x3fff) | 0x8000,
223        b & 0xffff_ffff_ffff
224    )
225}
226
227fn jitter_ms() -> u64 {
228    let t = SystemTime::now()
229        .duration_since(UNIX_EPOCH)
230        .map(|d| d.subsec_nanos() as u64)
231        .unwrap_or(0);
232    (t ^ t.rotate_left(13)) % 250
233}
solari/http.rs

solari/
http.rs
