Resilience patterns: survive a slow or failing dependency

// Outgoing copies the time left on ctx into the request, so the next service stops when the
// caller stops waiting.
func Outgoing(ctx context.Context, req *http.Request) error {
  dl, ok := ctx.Deadline()
  if !ok {
    return nil
  }
  left := time.Until(dl)
  if left <= 0 {
    return fmt.Errorf("deadline passed before the call: %w", context.DeadlineExceeded)
  }
  req.Header.Set(TimeoutHeader, strconv.FormatInt(left.Milliseconds(), 10))
  return nil
}

// Incoming gives each request a context that ends when the caller's time runs out. A request
// that arrives with no time left gets 504 and costs no work.
func Incoming(next http.Handler) http.Handler {
  return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    v := r.Header.Get(TimeoutHeader)
    if v == "" {
      next.ServeHTTP(w, r)
      return
    }
    ms, err := strconv.ParseInt(v, 10, 64)
    if err != nil {
      http.Error(w, "bad "+TimeoutHeader, http.StatusBadRequest)
      return
    }
    if ms <= 0 {
      http.Error(w, "no time left", http.StatusGatewayTimeout)
      return
    }
    ctx, cancel := context.WithTimeout(r.Context(), time.Duration(ms)*time.Millisecond)
    defer cancel()
    next.ServeHTTP(w, r.WithContext(ctx))
  })
}

// Backoff is capped exponential backoff.
type Backoff struct {
  Base   time.Duration
  Cap    time.Duration
  Jitter Jitter
}

// Delay is the wait before retry number `retry` (1 for the first retry).
func (b Backoff) Delay(retry int, r *rand.Rand) time.Duration {
  ceiling := b.Base
  for i := 1; i < retry && ceiling < b.Cap; i++ {
    ceiling *= 2
  }
  ceiling = min(ceiling, b.Cap)
  if b.Jitter == FullJitter {
    return time.Duration(r.Int64N(int64(ceiling)))
  }
  return ceiling
}

// RetryBudget caps retries at a share of first attempts. Every request adds Ratio tokens, up to
// Max. A retry spends one token. With no token left, the caller fails fast instead of retrying.
type RetryBudget struct {
  mu     sync.Mutex
  ratio  float64
  max    float64
  tokens float64
}

// NewRetryBudget returns a full budget.
func NewRetryBudget(ratio, max float64) *RetryBudget {
  return &RetryBudget{ratio: ratio, max: max, tokens: max}
}

// OnRequest is called once for every new request, before its first attempt.
func (b *RetryBudget) OnRequest() {
  b.mu.Lock()
  defer b.mu.Unlock()
  b.tokens = min(b.max, b.tokens+b.ratio)
}

// TryRetry spends a token and reports whether a retry is allowed.
func (b *RetryBudget) TryRetry() bool {
  b.mu.Lock()
  defer b.mu.Unlock()
  if b.tokens < 1 {
    return false
  }
  b.tokens--
  return true
}

// Bulkhead is a fixed pool of slots for calls to one dependency. When the dependency is slow,
// its calls fill its own pool and fail fast. Calls to other dependencies use other pools, so
// they keep working.
type Bulkhead struct{ slots chan struct{} }

// NewBulkhead returns a pool of n slots.
func NewBulkhead(n int) *Bulkhead { return &Bulkhead{slots: make(chan struct{}, n)} }

// Do runs fn in a free slot, or returns ErrFull at once. It never waits for a slot.
func (b *Bulkhead) Do(ctx context.Context, fn func(context.Context) error) error {
  select {
  case b.slots <- struct{}{}:
  default:
    return ErrFull
  }
  defer func() { <-b.slots }()
  return fn(ctx)
}

// Push admits an entry, or reports false when the queue is full: the server rejects the
// request at once, and the caller can go elsewhere.
func (q *Queue) Push(e Entry, now time.Time) bool {
  if q.policy.MaxLen > 0 && q.Len() >= q.policy.MaxLen {
    q.dropExpiredFront(now)
    if q.Len() >= q.policy.MaxLen {
      return false
    }
  }
  q.items = append(q.items, e)
  return true
}

// Pop returns the next entry to serve. Under overload it serves the newest entry, whose caller
// still waits, instead of the oldest, whose caller has probably left.
func (q *Queue) Pop(now time.Time) (Entry, bool) {
  for q.Len() > 0 {
    var e Entry
    if q.policy.LIFOAbove > 0 && q.Len() > q.policy.LIFOAbove {
      e = q.items[len(q.items)-1]
      q.items = q.items[:len(q.items)-1]
    } else {
      e = q.items[q.head]
      q.head++
    }
    if q.policy.DropExpired && !now.Before(e.Deadline) {
      q.Dropped++
      continue
    }
    q.compact()
    return e, true
  }
  q.compact()
  return Entry{}, false
}

// Hedge calls fn. If no answer arrives within `after`, it calls fn a second time and returns
// whichever answer succeeds first; the other call is cancelled. Set `after` near the p95
// latency, so only about 5% of requests send a second copy. Use it only for reads and other
// idempotent calls.
func Hedge[T any](ctx context.Context, after time.Duration, fn func(context.Context) (T, error)) (T, error) {
  ctx, cancel := context.WithCancel(ctx)
  defer cancel() // stops the call that lost

  type result struct {
    v   T
    err error
  }
  results := make(chan result, 2)
  call := func() {
    v, err := fn(ctx)
    results <- result{v, err}
  }
  go call()
  timer := time.NewTimer(after)
  defer timer.Stop()

  sent, failed := 1, 0
  var last error
  for {
    select {
    case <-timer.C:
      if sent == 1 {
        sent++
        go call()
      }
    case r := <-results:
      if r.err == nil {
        return r.v, nil
      }
      failed++
      last = r.err
      if failed == sent { // a hedge is not a retry: an error before the hedge ends the call
        var zero T
        return zero, last
      }
    case <-ctx.Done():
      var zero T
      return zero, ctx.Err()
    }
  }
}