Resilience

Resilience

Handlers.Resilience runs requests through pudu-lang-resilience pipelines. A 503 is a response, not a failure, so the strategies decide from transient(): 408, 429, every 5xx, transient failures, and timeouts. retryAfter lets a handled response's retry-after header set the delay.

module Ex08Resilience

import Std.Http as Http
import Std.Io as Io
import Std.Result as Result
import PuduLangResilience.Pipeline as Pipeline
import PuduLangResilience.Registry as Registry
import PuduLangResilience.Retry as Retry
import PuduLangResilience.Timeout as Timeout
import PuduLangHttpClient as HttpClient
import PuduLangHttpClient.Client as Client
import PuduLangHttpClient.Content as Content
import PuduLangHttpClient.Factory as Factory
import PuduLangHttpClient.Factory.Builder as Builder
import PuduLangHttpClient.Handlers.Resilience as Resilience
import PuduLangHttpClient.Request as Request
import PuduLangHttpClient.Response as Response
import PuduLangHttpClient.Stub as Stub
import PuduLangHttpClient.Utils.Shared as Shared

fn flaky(calls: &Shared.Shared[Int]) -> fn(Request.Request) -> HttpClient.Outcome[Response.Response] {
  let counter = *calls
  fn(request: Request.Request) -> HttpClient.Outcome[Response.Response] {
    let attempt = Shared.change(&counter, |count: Int| (count + 1, count + 1))
    if attempt % 3 != 0 { Ok(Response.answer(&request, 503).withHeader("retry-after", "0")) } else { Ok(Response.answer(&request, 200).withText("attempt " + show(attempt))) }
  }
}

export fn main() -> Int {
  let calls = Shared.shared(0)
  let stub = Stub.create()
  Stub.when(&stub, |_request: Request.Request| true, flaky(&calls))
  let options = Resilience.standardOptions()
  let standard = match Resilience.standardWith(&Resilience.Standard{..options, retry: Retry.Options{..options.retry, delay: 10}}) {
    case Ok(built) => built
    case Err(_) => panic("standard pipeline")
  }
  let readsOnly = match Resilience.onTransient(|handles| [Retry.strategy(Retry.Options{..Retry.defaults(), shouldHandle: handles, delay: 10}), Timeout.strategy(Timeout.after(2000))]) {
    case Ok(built) => built
    case Err(_) => panic("transient pipeline")
  }
  let registry: Registry.Registry[Response.Response, HttpClient.Failure] = Registry.create()
  let _added = Registry.tryAddBuilder(&registry, "inventory", |_context: Registry.BuilderContext| [Retry.strategy(Retry.Options{..Retry.defaults(), shouldHandle: Resilience.transient(), delay: 10})])
  let factory = match Factory.build([
      Builder.defaults().withPrimary(|| Stub.send(&stub)).withBaseAddress("https://api.test/"),
      Builder.named("standard").withHandler(Resilience.handler(standard)),
      Builder.named("selective").withHandler(Resilience.selecting(|request: Request.Request| if request.method == Http.Get { readsOnly } else { Pipeline.empty() })),
      Builder.named("registry").withHandler(Resilience.fromRegistry(&registry, "inventory"))
    ]) {
    case Ok(built) => built
    case Err(invalid) => panic(Factory.explain(&invalid))
  }
  let first = Client.getText(&Factory.createClient(&factory, "standard"), "stock")
  let second = Client.getText(&Factory.createClient(&factory, "selective"), "stock")
  let posted = Result.map(Client.post(&Factory.createClient(&factory, "selective"), "stock", Content.plain("x")), |r: Response.Response| r.status.code)
  let third = Client.getText(&Factory.createClient(&factory, "registry"), "stock")
  let _a = Io.writeLine(show((first, second, posted, third)))
  if first == Ok("attempt 3") && second == Ok("attempt 6") && posted == Ok(503) && third == Ok("attempt 9") { 0 } else { 1 }
}

Check it and run it:

pudu check src/Ex08Resilience.pudu
pudu run src/Ex08Resilience.pudu

Output:

(Ok("attempt 3"), Ok("attempt 6"), Ok(503), Ok("attempt 9"))

Function	What it adds
handler(pipeline)	one pipeline for every request
selecting(choose)	a pipeline chosen per request
fromRegistry(registry, key)	the registry's pipeline for a key, rebuilt when the registry reloads it
onTransient(build)	a pipeline of strategies given the transient predicate
standard(), standardWith(options)	rate limiter (1000 permits), 30 s total timeout, three exponential retries with jitter honouring retry-after, a circuit breaker, and a 10 s attempt timeout
standardHedging(), standardHedgingWith(options)	30 s total timeout, one hedged attempt after two seconds, a circuit breaker, and a 10 s attempt timeout

A strategy's rejection — an open circuit, a full rate limiter — answers Rejected with its description; a timeout of the pipeline's answers TimedOut.

Related

• Previous: Lifetimes And Events
• Next: Logging, Propagation, And Metrics
• Reference Index