Skip to content

Resilience processors

pkg/processor/resilience.go collects the failure-handling and traffic-shaping processors: Retry, Circuit Breaker, Throttle, Delay, Timeout, fixed Loop, Sample, and Load Balancer (round-robin, random, failover, sticky, weighted, topic, custom).

Processors in this file

NewRetry

Retries a processor with exponential backoff.

steps:
  - retry:
      max-attempts: 3
      delay: 500ms
      backoff-multiplier: 2.0
      steps:
        - target: http://flaky
Field Type Default Description
max-attempts int required Total attempts including the first. 3 means 1 initial + 2 retries.
delay duration required Initial wait between attempts.
backoff-multiplier float 2.0 (Go default; 0 in YAML means no backoff) Each retry multiplies the prior delay by this factor.
steps list required Inner pipeline.

Constructor: NewRetry(processor engine.Processor, maxAttempts int, initialDelay time.Duration) *Retry.

Sets HelixRedeliveryCounter (1-based) and HelixRedeliveryMaxCounter on the exchange. Honors ctx.Done() during the backoff sleep. Returns the last error wrapped as retry exhausted after N attempts: ....

NewCircuitBreaker

State machine that fails fast after a threshold of errors and tests recovery after a cooldown.

steps:
  - circuit-breaker:
      threshold: 5
      half-open-after: 30s
      steps:
        - target: http://flaky
      fallback:
        - set-body: { constant: '{"degraded": true}' }
Field Type Default Description
threshold int required Consecutive failures before opening.
half-open-after duration required Cooldown before half-open. After this time, the next call attempts the real processor.
steps list required Main processor.
fallback list optional Pipeline used while open (or when half-open fails). Without it, the breaker returns circuit breaker is open.

States:

State Behavior
Closed Normal. Tracks consecutive failures (atomic counter).
Open Fast-fail. Calls fallback (or errors) until HalfOpenAfter elapses, then transitions to HalfOpen.
HalfOpen One test call. Success → Closed; failure → Open (counter reset so next half-open starts fresh).

Constructor: NewCircuitBreaker(processor engine.Processor, threshold int, halfOpenAfter time.Duration) *CircuitBreaker. Set Fallback directly on the struct.

NewThrottle

Token-bucket rate limiter — at most Rate messages per Period.

steps:
  - throttle:
      rate: 100
      period: 1s
      steps:
        - target: http://rate-limited
Field Type Default Description
rate int required Burst capacity / steady-state rate.
period duration required Refill interval. Internally, one token refills every Period/Rate.
steps list required Inner pipeline.

Constructor: NewThrottle(processor engine.Processor, rate int, period time.Duration) *Throttle.

The bucket is preloaded full so the first Rate requests pass through immediately. After that, callers block waiting for a refill or for ctx.Done(). Call Stop() to shut down the refill goroutine.

NewDelay

Delays by a fixed duration before processing. Honors ctx.Done().

steps:
  - delay:
      duration: 2s
      steps:
        - target: http://eventually
Field Type Default Description
duration duration required How long to sleep.
steps list required Inner pipeline.

Constructor: NewDelay(processor engine.Processor, duration time.Duration) *Delay.

NewTimeout

Runs the inner processor with a derived context that cancels after Duration. If the processor doesn't finish in time, an optional fallback runs; without a fallback, the processor errors with timeout after Xs. Processor panics are caught and surfaced as errors.

steps:
  - timeout:
      duration: 5s
      steps:
        - target: http://slow
      fallback:
        - set-body: { constant: '{"error": "timeout"}' }
Field Type Default Description
duration duration required Maximum wall time.
steps list required Wrapped processor.
fallback list optional Pipeline run on timeout (with the original context, not the timed-out one).

Constructor: NewTimeout(processor engine.Processor, duration time.Duration) *Timeout. Set Fallback directly.

NewLoop

Fixed-count loop. For dynamically-computed counts, use DynamicLoop.

steps:
  - loop:
      count: 10
      steps:
        - target: kafka://retry
Field Type Default Description
count int required Iteration count.
steps list required Inner pipeline.

Sets HelixLoopIndex (0-based) and HelixLoopSize per iteration. Honors ctx.Done() between iterations. Constructor: NewLoop(processor engine.Processor, count int) *Loop.

NewSample

Percentage-based sampler. Lets Rate% of messages through, drops the rest.

steps:
  - sample:
      rate: 10
      steps:
        - target: log://sampled
Field Type Default Description
rate int required Percent 1-100.
steps list required Pipeline run for sampled messages.

Constructor: NewSample(processor engine.Processor, rate int) *Sample. Implementation uses an atomic counter and count % 100 < rate — deterministic by call count, not random. For probabilistic sampling use SampleProcessor in eip_extra.go.

NewLoadBalancer

Distributes exchanges across N processors using one of seven strategies.

steps:
  - load-balancer:
      strategy: round-robin   # one of: round-robin, random, sticky, failover, weighted, topic
      uris:
        - http://backend-1
        - http://backend-2
        - http://backend-3
      weights: [3, 1, 1]           # for weighted only
      sticky-header: X-Session-Id  # for sticky only
Field Type Default Description
strategy string roundrobin One of roundrobin (or empty), random, sticky, failover, weighted, topic. The Custom strategy is Go-only.
uris list required Target URIs.
weights list of int optional Used by weighted only. Index-aligned with uris.
sticky-header string "" (exchange ID) Header whose value is hashed (FNV-1a). Used by sticky only.

Strategy semantics:

  • RoundRobin — atomic counter mod N. Even distribution.
  • Random — rand.Intn(N). No state.
  • Failover — try each in order; first success wins. All-failure returns all load balancer targets failed.
  • Sticky — hash a key to pick an index. Key is the exchange ID by default, or the value of StickyHeader. Same key always → same processor.
  • Weighted — cumulative-weight binary search. Heavier indices get proportionally more traffic. Falls back to equal weights if none configured.
  • Topic — reads TopicHeader (default HelixLoadBalancerTopic). If the value parses as an in-range int, used as a direct index; otherwise hashed (FNV-1a).
  • Custom — calls CustomFunc(ex, processors) → engine.Processor. Used by Go callers to implement bespoke routing.

Constructors:

  • NewLoadBalancer(strategy LoadBalanceStrategy, processors ...engine.Processor) *LoadBalancer
  • NewWeightedLoadBalancer(weights []int, processors ...engine.Processor) *LoadBalancer
  • NewStickyLoadBalancer(headerName string, processors ...engine.Processor) *LoadBalancer

Pitfalls

  • Retry is wrapped processing, not endpoint redelivery. The whole inner pipeline runs from scratch each attempt — non-idempotent side effects (writes, sends) run again.
  • Throttle.Stop() is idempotent (uses closed-channel guard). Forgetting to call it leaks the refill goroutine; usually harmless in long-running daemons but matters in tests.
  • The simple Sample here is deterministic — every 100 messages, exactly Rate of them pass. For true probabilistic sampling, use SampleProcessor with Rate as a 0.0-1.0 float.
  • Load-balancer Failover tries targets sequentially per call. There's no fast-fail or marking of unhealthy nodes; pair with CircuitBreaker on each target if you need that.
  • Sticky load-balancing depends on hash distribution. With few keys or N close to bucket count, distribution can be skewed.

Examples

Retry + timeout + circuit-breaker stack:

workflows:
  - id: protected-call
    trigger: { uri: direct://call }
    steps:
      - circuit-breaker:
          threshold: 10
          half-open-after: 30s
          steps:
            - timeout:
                duration: 5s
                steps:
                  - retry:
                      max-attempts: 3
                      delay: 200ms
                      backoff-multiplier: 2.0
                      steps:
                        - target: http://flaky-service
          fallback:
            - set-body: { constant: '{"degraded": true}' }

Weighted load-balanced fan-out:

workflows:
  - id: weighted-fan-out
    trigger: { uri: kafka://incoming }
    steps:
      - load-balancer:
          strategy: weighted
          weights: [3, 1, 1]
          uris:
            - http://big-backend
            - http://small-backend-a
            - http://small-backend-b

Source

pkg/processor/resilience.go

See also

  • error_handling — DeadLetterChannel with engine-level redelivery (the route-level equivalent of retry).
  • loop_dynamic — counted loop with expression-driven count.
  • eip_extra — SampleProcessor with rate/period modes.