1 # Hairpin: an HTTP client with a base URL, default headers, a pool, cookies, mTLS, redirects, and proven retries. Source: https://github.com/paymog/bend-kit/tree/main/hairpin 2 import Base 3 import bend-kit-http@0.23.0.1/http.bend as Http 4 import bend-kit-time@0.1.0.0/time.bend as Time 5 # By hash, as http imports them: bend-kit-url@0.4.1.0, -json@0.5.0.1, -bytes@0.3.0.0. 6 import 0x1f2d80f53f971b16c6de6a65cb1918ae/url.bend as Url 7 import 0x584fc27920487ceab242392391418d7f/json.bend as Json 8 import 0x49814d83de8f70993a43e1002be29ecd/bytes.bend as Bytes 9 import ./retry.bend as Retry 10 11 # base: "" for none. headers: lowercase names. ms: the step timeout. policy: attempts, backoff, and deadline of each request. 12 type Opts is Data: 13 Opts{base: String, headers: Map<&2, List<&2, String>>, ms: U32, mode: Http.Mode, policy: Retry.Policy, cert: Maybe<&2, Http.Cert>} 14 15 # breaker: the circuit breaker of every request on this client. 16 type Client is Type: 17 Client{opts: Opts, pool: Http.Pool, jar: Maybe<&2, Http.Jar>, breaker: Retry.Breaker} 18 19 # ErrHttp: the last attempt failed. ErrDenied: no attempt started, for why. 20 type Err is Data: 21 ErrHttp{error: Http.Err} 22 ErrDenied{why: Retry.Deny} 23 24 # One attempt, the first included; backoff from 500 ms, doubled to at most 30 s; no deadline. 25 def policy() -> Retry.Policy: 26 Retry.Policy{1n, 500n, 30000, None{}} 27 28 # A pool of up to cap idle sockets per origin, no base, no default headers, 30 s steps, follow redirects, 29 # no retries, and no circuit breaker. 30 def new.with(+cap: U32) -> Client: 31 Client{Opts{"", Http.empty(), 30000, Http.ModeFollow{}, policy(), None{}}, Http.pool.new.with(cap), None{}, Retry.BreakerOff{}} 32 33 def new() -> Client: 34 new.with(8) 35 36 def base(c: Client, url: String) -> Client: 37 Client{Opts{b, h, ms, mode, p, cert}, pool, j, br} = c 38 Client{Opts{url, h, ms, mode, p, cert}, pool, j, br} 39 40 # A default header. It replaces an earlier default of the same name; a request header of that name replaces it. 41 def header(c: Client, k: String, v: String) -> Client: 42 Client{Opts{b, h, ms, mode, p, cert}, pool, j, br} = c 43 Client{Opts{b, Http.set(h, String.to_lower(k), v), ms, mode, p, cert}, pool, j, br} 44 45 def timeout(c: Client, ms: U32) -> Client: 46 Client{Opts{b, h, old, mode, p, cert}, pool, j, br} = c 47 Client{Opts{b, h, ms, mode, p, cert}, pool, j, br} 48 49 def redirect(c: Client, mode: Http.Mode) -> Client: 50 Client{Opts{b, h, ms, old, p, cert}, pool, j, br} = c 51 Client{Opts{b, h, ms, mode, p, cert}, pool, j, br} 52 53 # Up to n retries after the first attempt: n + 1 attempts in all. 54 def retry(c: Client, n: U32) -> Client: 55 Client{Opts{b, h, ms, mode, Retry.Policy{old, base, cap, d}, cert}, pool, j, br} = c 56 Client{Opts{b, h, ms, mode, Retry.Policy{1n+U32.to_nat(n), base, cap, d}, cert}, pool, j, br} 57 58 # No attempt of a request starts ms or more after the request starts. 59 def deadline(c: Client, ms: U32) -> Client: 60 Client{Opts{b, h, t, mode, Retry.Policy{a, base, cap, old}, cert}, pool, j, br} = c 61 Client{Opts{b, h, t, mode, Retry.Policy{a, base, cap, Some{U32.to_nat(ms)}}, cert}, pool, j, br} 62 63 # A circuit breaker in front of every request, starting closed. 64 def circuit(c: Client, circ: Retry.Circuit) -> Client: 65 Client{o, pool, j, old} = c 66 Client{o, pool, j, Retry.Breaker.new(circ)} 67 68 # The breaker, for inspection. 69 def breaker.of(c: Client) -> Client & Retry.Breaker: 70 Client{o, pool, j, +br} = c 71 (Client{o, pool, j, br}, br) 72 73 # Paths to a PEM client chain and its private key, presented to every HTTPS origin. 74 def cert(c: Client, chain: String, key: String) -> Client: 75 Client{Opts{b, h, ms, mode, p, old}, pool, j, br} = c 76 Client{Opts{b, h, ms, mode, p, Some{Http.Cert{chain, key}}}, pool, j, br} 77 78 # Every request stores and sends cookies through jar. Its clock moves to Time.now() before each try. 79 def jar(c: Client, jar: Http.Jar) -> Client: 80 Client{o, pool, old, br} = c 81 Client{o, pool, Some{jar}, br} 82 83 # The jar, if the client has one. 84 def jar.of(c: Client) -> Client & Maybe<&2, Http.Jar>: 85 Client{o, pool, +j, br} = c 86 (Client{o, pool, j, br}, j) 87 88 def close(c: Client) -> IO(Unit): 89 Client{o, pool, j, br} = c 90 Http.pool.close(pool) 91 92 def ms.of(r: U32 & U32 & U32) -> Nat: 93 (hi, lo, ns) = r 94 Nat.add(Nat.mul(Nat.add(Nat.mul(U32.to_nat(hi), 1n+4294967295n), U32.to_nat(lo)), 1000n), U32.to_nat((ns / 1000000 : U32))) 95 96 # Milliseconds on the monotonic clock that Retry budgets and breakers use. 97 def now.ms() -> IO(Nat): 98 do IO: 99 r : U32 & U32 & U32 <- Time.mono.raw() 100 return ms.of(r) 101 102 def target.pick(none: Bool, b: String, ref: String) -> Maybe<&2, String>: 103 match none: 104 case True{}: 105 Some{ref} 106 case False{}: 107 Http.resolve(b, ref) 108 109 # ref against the base (RFC 3986 ยง5.2); with no base, ref as given. None when the base is not absolute. 110 def target(+b: String, ref: String) -> Maybe<&2, String>: 111 target.pick(String.is_empty(b), b, ref) 112 113 def target.abs(m: Maybe<&2, String>) -> Maybe<&2, Url.Abs>: 114 match m: 115 case None{}: 116 None{} 117 case Some{s}: 118 Url.absolute(s) 119 120 # The defaults under the request headers, names lowercased; a request header replaces a default of its name. 121 # Host, Connection, Content-Length, and Transfer-Encoding are the client's own, so both sides drop them. 122 def merge(defaults: Map<&2, List<&2, String>>, headers: Map<&2, List<&2, String>>) -> Map<&2, List<&2, String>>: 123 Http.req.put(Map.to_list(&2, List<&2, String>, headers), Http.req.put(Map.to_list(&2, List<&2, String>, defaults), Http.empty())) 124 125 def jar.now(j: Maybe<&2, Http.Jar>) -> IO(Maybe<&2, Http.Jar>): 126 match j: 127 case None{}: 128 IO.pure(Maybe<&2, Http.Jar>, None{}) 129 case Some{jar}: 130 do IO>: 131 now : Time.Instant <- Time.now() 132 return Some{Http.jar.at(jar, now)} 133 134 def once.final(+enc: String, x: Http.Pool & Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, Http.Jar>) -> 135 IO(Http.Pool & Maybe<&2, Http.Jar> & Result<&1, &1, Http.Err, Http.Res>): 136 (p, r, j) = x 137 do IO & Result<&1, &1, Http.Err, Http.Res>>: 138 d : Result<&1, &1, Http.Err, Http.Res> <- Http.fetch.final(enc, r) 139 return (p, j, d) 140 141 # One try: the redirect loop on the pool, with the jar and the certificate, then the body decoded. 142 def once(+ms: U32, +mode: Http.Mode, +cert: Maybe<&2, Http.Cert>, +enc: String, p: Http.Pool, j: Maybe<&2, Http.Jar>, 143 +method: String, +u: Maybe<&2, Url.Abs>, +headers: Map<&2, List<&2, String>>, body: Bytes.Bytes) -> 144 IO(Http.Pool & Maybe<&2, Http.Jar> & Result<&1, &1, Http.Err, Http.Res>): 145 do IO & Result<&1, &1, Http.Err, Http.Res>>: 146 now : Maybe<&2, Http.Jar> <- jar.now(j) 147 x : Http.Pool & Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, Http.Jar> <- 148 Http.pool.hops(mode, ms, Http.Hops{p, Http.fetch.start(method, headers, body, u, enc), cert, now}) 149 once.final(enc, x) 150 151 def hint.ms(m: Maybe<&2, U32>) -> Maybe<&2, Nat>: 152 match m: 153 case None{}: 154 None{} 155 case Some{ms}: 156 Some{U32.to_nat(ms)} 157 158 # m is Http.retry.judge's verdict for an idempotent request: Some{Retry-After} when another try may help. 159 # A non-idempotent request with that verdict is Unsafe: the breaker counts it, but it never repeats. 160 def outcome(+idem: Bool, m: Maybe<&2, String>, now: Time.Instant) -> Retry.Outcome: 161 match m: 162 case None{}: 163 Retry.Settled{} 164 case Some{+v}: 165 Bool.pick(Retry.Outcome, idem, Retry.Transient{hint.ms(Http.retry.after(v, now))}, Retry.Unsafe{}) 166 167 def lift(r: Result<&1, &1, Http.Err, Http.Res>) -> Result<&1, &1, Err, Http.Res>: 168 match r: 169 case Fail{e}: 170 Fail{ErrHttp{e}} 171 case Done{res}: 172 Done{res} 173 174 # The live state of one request between attempts. last: the result of the last attempt, None before the first. 175 type Step is Type: 176 Step{pool: Http.Pool, jar: Maybe<&2, Http.Jar>, breaker: Retry.Breaker, budget: Retry.Budget, body: Bytes.Bytes, 177 last: Maybe<&1, Result<&1, &1, Err, Http.Res>>, plan: Retry.Plan} 178 179 # The clock, read only when a decision depends on it; 0 otherwise, which no such decision reads. 180 def clock(need: Bool) -> IO(Nat): 181 match need: 182 case True{}: 183 now.ms() 184 case False{}: 185 IO.pure(Nat, 0n) 186 187 def breaker.on(br: Retry.Breaker) -> Bool: 188 match br: 189 case Retry.BreakerOff{}: 190 False{} 191 case Retry.BreakerOn{c, s}: 192 True{} 193 194 def transient(o: Retry.Outcome) -> Bool: 195 match o: 196 case Retry.Transient{after}: 197 True{} 198 case Retry.Settled{}: 199 False{} 200 case Retry.Unsafe{}: 201 False{} 202 203 def some(m: Maybe<&2, Nat>) -> Bool: 204 match m: 205 case None{}: 206 False{} 207 case Some{t}: 208 True{} 209 210 def deadline.some(b: Retry.Budget) -> Bool: 211 Retry.Budget{l, d} = b 212 some(d) 213 214 # The wall clock dates a Retry-After, so it is read only for a transient failure. 215 def outcome.io(+idem: Bool, m: Maybe<&2, String>) -> IO(Retry.Outcome): 216 match m: 217 case None{}: 218 IO.pure(Retry.Outcome, Retry.Settled{}) 219 case Some{v}: 220 do IO: 221 wall : Time.Instant <- Time.now() 222 return outcome(idem, Some{v}, wall) 223 224 # Jitter for the backoff, drawn only when a retry may follow. 225 def jitter.io(need: Bool) -> IO(Nat): 226 match need: 227 case True{}: 228 do IO: 229 rnd : Result<&1, &1, U32 & String, U32> <- IO.random_u32() 230 return U32.to_nat(Http.retry.jitter(rnd)) 231 case False{}: 232 IO.pure(Nat, 0n) 233 234 def settle.at(+now: Nat, +rnd: Nat, +p: Retry.Policy, +n: Nat, +o: Retry.Outcome, br: Retry.Breaker, +b: Retry.Budget, 235 body: Bytes.Bytes, pool: Http.Pool, jar: Maybe<&2, Http.Jar>, r: Result<&1, &1, Http.Err, Http.Res>) -> Step: 236 Step{pool, jar, Retry.Breaker.done(br, now, Retry.failed(o)), b, body, Some{lift(r)}, Retry.plan(o, p, b, n, now, rnd)} 237 238 # The attempt has ended: tell the breaker, and plan the next step. 239 def settle.outcome(+p: Retry.Policy, +n: Nat, +o: Retry.Outcome, +br: Retry.Breaker, b: Retry.Budget, body: Bytes.Bytes, 240 pool: Http.Pool, jar: Maybe<&2, Http.Jar>, r: Result<&1, &1, Http.Err, Http.Res>) -> IO(Step): 241 do IO: 242 now : Nat <- clock(Bool.or(breaker.on(br), transient(o))) 243 rnd : Nat <- jitter.io(transient(o)) 244 return settle.at(now, rnd, p, n, o, br, b, body, pool, jar, r) 245 246 def settle.judged(+p: Retry.Policy, +idem: Bool, +n: Nat, br: Retry.Breaker, b: Retry.Budget, body: Bytes.Bytes, 247 pool: Http.Pool, jar: Maybe<&2, Http.Jar>, j: Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>) -> IO(Step): 248 (r, m) = j 249 do IO: 250 o : Retry.Outcome <- outcome.io(idem, m) 251 settle.outcome(p, n, o, br, b, body, pool, jar, r) 252 253 # The default judge: Http.retry.judge, as if the request were idempotent; outcome then applies idem. 254 def judge(r: Result<&1, &1, Http.Err, Http.Res>) -> Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>: 255 Http.retry.judge(True{}, r) 256 257 def settle(~judge: Result<&1, &1, Http.Err, Http.Res> -> Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>, 258 +p: Retry.Policy, +idem: Bool, +n: Nat, br: Retry.Breaker, b: Retry.Budget, body: Bytes.Bytes, 259 x: Http.Pool & Maybe<&2, Http.Jar> & Result<&1, &1, Http.Err, Http.Res>) -> IO(Step): 260 (pool, jar, r) = x 261 settle.judged(p, idem, n, br, b, body, pool, jar, judge(r)) 262 263 def attempt.run(~judge: Result<&1, &1, Http.Err, Http.Res> -> Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>, 264 +o: Opts, +enc: String, +idem: Bool, +method: String, +u: Maybe<&2, Url.Abs>, 265 +headers: Map<&2, List<&2, String>>, +n: Nat, pool: Http.Pool, jar: Maybe<&2, Http.Jar>, br: Retry.Breaker, 266 b: Retry.Budget, r: Bytes.Bytes & Bytes.Bytes) -> IO(Step): 267 Opts{base, h, +ms, +mode, +p, +cert} = o 268 (body, spare) = r 269 do IO: 270 x : Http.Pool & Maybe<&2, Http.Jar> & Result<&1, &1, Http.Err, Http.Res> <- once(ms, mode, cert, enc, pool, jar, method, u, headers, spare) 271 settle(~judge, p, idem, n, br, b, body, x) 272 273 def denied(why: Retry.Deny, last: Maybe<&1, Result<&1, &1, Err, Http.Res>>) -> Maybe<&1, Result<&1, &1, Err, Http.Res>>: 274 match last: 275 case None{}: 276 Some{Fail{ErrDenied{why}}} 277 case Some{r}: 278 Some{r} 279 280 def attempt.gate(~judge: Result<&1, &1, Http.Err, Http.Res> -> Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>, 281 x: Retry.Breaker & Retry.Budget & Retry.Gate, +o: Opts, +enc: String, +idem: Bool, +method: String, 282 +u: Maybe<&2, Url.Abs>, +headers: Map<&2, List<&2, String>>, +n: Nat, pool: Http.Pool, jar: Maybe<&2, Http.Jar>, 283 body: Bytes.Bytes, last: Maybe<&1, Result<&1, &1, Err, Http.Res>>) -> IO(Step): 284 (br, b, g) = x 285 match g: 286 case Retry.GateNo{why}: 287 IO.pure(Step, Step{pool, jar, br, b, body, denied(why, last), Retry.PlanStop{}}) 288 case Retry.GateGo{}: 289 attempt.run(~judge, o, enc, idem, method, u, headers, n, pool, jar, br, b, Bytes.slice(body, 0, 4294967295)) 290 291 # Ask the gate; when it admits, make attempt n (0 first) of this layer. With no deadline and no breaker, 292 # admission does not read the clock. 293 def attempt(~judge: Result<&1, &1, Http.Err, Http.Res> -> Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>, 294 +o: Opts, +enc: String, +idem: Bool, +method: String, +u: Maybe<&2, Url.Abs>, 295 +headers: Map<&2, List<&2, String>>, +n: Nat, s: Step) -> IO(Step): 296 Step{pool, jar, +br, +b, body, last, plan} = s 297 do IO: 298 now : Nat <- clock(Bool.or(breaker.on(br), deadline.some(b))) 299 attempt.gate(~judge, Retry.admit(br, b, now), o, enc, idem, method, u, headers, n, pool, jar, body, last) 300 301 def pause(ms: Nat) -> IO(Unit): 302 match ms: 303 case 0n: 304 IO.pure(Unit, Unit{}) 305 case 1n+k: 306 IO.sleep(U32.from_nat(1n+k)) 307 308 # left: attempts this layer may still make. n: attempts made so far. Each turn follows the plan of the last step. 309 def go(~judge: Result<&1, &1, Http.Err, Http.Res> -> Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>, 310 left: Nat, +n: Nat, +o: Opts, +enc: String, +idem: Bool, +method: String, +u: Maybe<&2, Url.Abs>, 311 +headers: Map<&2, List<&2, String>>, s: Step) -> IO(Step): 312 match left: 313 case 0n: 314 IO.pure(Step, s) 315 case 1n+rest: 316 Step{pool, jar, br, b, body, last, plan} = s 317 match plan: 318 case Retry.PlanStop{}: 319 IO.pure(Step, Step{pool, jar, br, b, body, last, Retry.PlanStop{}}) 320 case Retry.PlanWait{ms}: 321 do IO: 322 pause(ms) 323 next : Step <- attempt(~judge, o, enc, idem, method, u, headers, n, Step{pool, jar, br, b, body, last, Retry.PlanStop{}}) 324 go(~judge, rest, 1n+n, o, enc, idem, method, u, headers, next) 325 326 def result.of(m: Maybe<&1, Result<&1, &1, Err, Http.Res>>) -> Result<&1, &1, Err, Http.Res>: 327 match m: 328 case None{}: 329 Fail{ErrDenied{Retry.DenySpent{}}} 330 case Some{r}: 331 r 332 333 def request.done(+o: Opts, s: Step) -> Client & Retry.Budget & Result<&1, &1, Err, Http.Res>: 334 Step{pool, jar, br, b, body, last, plan} = s 335 (Client{o, pool, jar, br}, b, result.of(last)) 336 337 def attempts(p: Retry.Policy) -> Nat: 338 Retry.Policy{a, base, cap, d} = p 339 a 340 341 # The budget of one plain request: the policy's attempts, and its deadline from now. The clock is read only for a deadline. 342 def budget(c: Client) -> IO(Client & Retry.Budget): 343 Client{+o, pool, jar, br} = c 344 Opts{base, h, ms, mode, Retry.Policy{+a, bs, cap, +d}, cert} = o 345 do IO: 346 now : Nat <- clock(some(d)) 347 return (Client{o, pool, jar, br}, Retry.Budget.new(a, now, d)) 348 349 # One request inside a caller's retry layer, with every choice explicit. Every attempt takes from b, the 350 # caller's shared budget, and obeys its deadline; the client's own policy caps the attempts of this call. 351 # The budget left comes back beside the result: pass it to the next call, so nested layers never reset or 352 # multiply it. The client's own deadline does not apply here; b's does. 353 # judge: Some{Retry-After} when another attempt may help. idem: the request is safe to repeat, so a judged 354 # failure is retried; else the breaker counts it and it never repeats. enc: the Accept-Encoding codings. 355 def request.as(~judge: Result<&1, &1, Http.Err, Http.Res> -> Result<&1, &1, Http.Err, Http.Res> & Maybe<&2, String>, 356 c: Client, b: Retry.Budget, +idem: Bool, +enc: String, +method: String, url: String, 357 headers: Map<&2, List<&2, String>>, body: Bytes.Bytes) -> IO(Client & Retry.Budget & Result<&1, &1, Err, Http.Res>): 358 Client{+o, pool, jar, br} = c 359 Opts{+base, +h, ms, mode, +p, cert} = o 360 +u = target.abs(target(base, url)) 361 +hs = merge(h, headers) 362 do IO>: 363 s : Step <- go(~judge, attempts(p), 0n, o, enc, idem, method, u, hs, Step{pool, jar, br, b, body, None{}, Retry.PlanWait{0n}}) 364 return request.done(o, s) 365 366 # request.as with judge, the idempotency of method, and every coding Http decodes. 367 def request.in(c: Client, b: Retry.Budget, +method: String, url: String, headers: Map<&2, List<&2, String>>, 368 body: Bytes.Bytes) -> IO(Client & Retry.Budget & Result<&1, &1, Err, Http.Res>): 369 do IO>: 370 enc : String <- Http.codings() 371 request.as(~judge, c, b, Http.pool.idempotent(method), enc, method, url, headers, body) 372 373 def request.drop(x: Client & Retry.Budget & Result<&1, &1, Err, Http.Res>) -> Client & Result<&1, &1, Err, Http.Res>: 374 (c, b, r) = x 375 (c, r) 376 377 def request.budget(+method: String, url: String, headers: Map<&2, List<&2, String>>, body: Bytes.Bytes, 378 x: Client & Retry.Budget) -> IO(Client & Retry.Budget & Result<&1, &1, Err, Http.Res>): 379 (c, b) = x 380 request.in(c, b, method, url, headers, body) 381 382 # url resolves against the base; headers go over the defaults. Redirects follow the client's mode, at most 20. 383 # Each request gets a fresh budget from budget(c). Before each attempt the deadline, the 384 # budget, and the breaker must admit it. GET, HEAD, OPTIONS, TRACE, PUT, and DELETE try again after a connect 385 # error, a timeout, or 408, 429, 500, 502, 503, or 504, as Http.retry.judge says, on the same pool. 386 # When the attempts run out, the last result comes back; when none started, ErrDenied says why. 387 def request(c: Client, +method: String, url: String, headers: Map<&2, List<&2, String>>, body: Bytes.Bytes) -> 388 IO(Client & Result<&1, &1, Err, Http.Res>): 389 do IO>: 390 y : Client & Retry.Budget <- budget(c) 391 x : Client & Retry.Budget & Result<&1, &1, Err, Http.Res> <- request.budget(method, url, headers, body, y) 392 return request.drop(x) 393 394 def get(c: Client, url: String) -> IO(Client & Result<&1, &1, Err, Http.Res>): 395 request(c, "GET", url, Http.empty(), Bytes.new(0)) 396 397 # POST v as compact JSON, with content-type: application/json. 398 def post.json(c: Client, url: String, v: Json.Val) -> IO(Client & Result<&1, &1, Err, Http.Res>): 399 request(c, "POST", url, Http.set(Http.empty(), "content-type", "application/json"), Json.encode.bytes(v))