Latency.pudu

Pudu61 lines2.8 KB

Copy pathGitHub ↗

1/** @Chaos.Latency.Strategy — delays a callback before it runs */2module PuduLangResilience.Chaos.Latency34import Std.Concurrent.Cancel as Cancel5import PuduLangResilience.Chaos as Chaos6import PuduLangResilience.Clock as Clock7import PuduLangResilience.Constants.Events as Events8import PuduLangResilience.Context as Context9import PuduLangResilience as Resilience10import PuduLangResilience.Strategy as Strategy11import PuduLangResilience.Telemetry as Telemetry1213/** @Chaos.Latency.Injected — the delay injected into one execution */14export type Injected = { context: Context.Context, latency: Int }1516/** @Chaos.Latency.Options — how long to delay and how often */17export type Options = {18  name: Option[Str],19  injection: Chaos.Injection,20  latency: Int,21  generator: Option[fn(Chaos.Arguments) -> Int],22  onInjected: Option[fn(Injected) -> ()]23}2425/// Thirty seconds at the default injection.26export fn defaults() -> Options { Options{name: None, injection: Chaos.defaults(), latency: 30000, generator: None, onInjected: None} }2728/// Injects `millis` of delay at the given injection settings.29export fn injecting(millis: Int, injection: Chaos.Injection) -> Options { Options{..defaults(), injection: injection, latency: millis} }3031/// Every reason the options cannot be followed.32export fn validate(options: &Options) -> Array[Str] {33  var problems = Chaos.validate(&options.injection)34  if options.latency < 0 { problems = problems.push("latency must not be negative") }35  problems36}3738/// A latency injection strategy following the options. A delay of zero or less injects39/// nothing; a token firing during the delay answers `Cancelled` without running the callback.40export fn strategy[T, E](options: Options) -> Strategy.Strategy[T, E] {41  Strategy.Strategy {42    name: options.name,43    kind: "ChaosLatency",44    summary: "injects " + show(options.latency) + " ms at rate " + show(options.injection.rate),45    problems: validate(&options),46    attach: Strategy.detached,47    execute: fn(runtime: Strategy.Runtime, context: Context.Context, callback: Strategy.Callback[T, E]) -> Resilience.Outcome[T, E] {48      if !Chaos.shouldInject(&options.injection, &context, &runtime.randomizer) { return callback(context) }49      let latency = match options.generator {50        case Some(generate) => generate(Chaos.arguments(&context, &runtime.randomizer))51        case None => options.latency52      }53      if latency <= 0 { return callback(context) }54      Telemetry.report(&runtime.telemetry, &context, Events.ON_LATENCY, Telemetry.Information, None, Telemetry.LatencyInjected(latency))55      if let Err(why) = Clock.sleep(&runtime.clock, latency, &context.token) { return Err(Resilience.Cancelled(Cancel.explain(&why))) }56      if let Some(notify) = options.onInjected { notify(Injected{context: context, latency: latency}) }57      callback(context)58    }59  }60}61