Class: AceMQ::AMQP::RetryPolicy

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/retry_policy.rb

Overview

When to try again, and when to stop.

The arithmetic here is part of the cross-language contract rather than a local choice: the same policy must produce the same delays in Java, Go, .NET, Python and Ruby, because the same message can be retried by a consumer written in any of them. Jitter is the exception — it is random by definition — so #schedule exposes the delays without it, which is what to read when deciding whether a policy is the one you meant.

Durations are seconds, as floats, because that is what Ruby's own sleep and Time arithmetic speak.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(max_attempts: 1, initial_delay: 0.0, multiplier: 2.0, max_delay: 0.0, jitter_factor: 0.0, max_message_age: 0.0) ⇒ RetryPolicy

Returns a new instance of RetryPolicy.

Parameters:

  • max_attempts (Integer) (defaults to: 1) —

    total deliveries including the first

  • initial_delay (Float) (defaults to: 0.0) —

    seconds before the second attempt

  • multiplier (Float) (defaults to: 2.0) —

    what the delay is multiplied by each time

  • max_delay (Float) (defaults to: 0.0) —

    the ceiling in seconds, or 0 for none

  • jitter_factor (Float) (defaults to: 0.0) —

    how far a delay may move either side, 0 to 1

  • max_message_age (Float) (defaults to: 0.0) —

    give up on anything older, or 0 for never



40
41
42
43
44
45
46
47
48
49
 
# File 'lib/acemq/amqp/retry_policy.rb', line 40

def initialize(max_attempts: 1, initial_delay: 0.0, multiplier: 2.0,
               max_delay: 0.0, jitter_factor: 0.0, max_message_age: 0.0)
  @max_attempts = max_attempts
  @initial_delay = initial_delay.to_f
  @multiplier = multiplier.to_f
  @max_delay = max_delay.to_f
  @jitter_factor = jitter_factor.to_f
  @max_message_age = max_message_age.to_f
  freeze
end

Instance Attribute Details

#initial_delay ⇒ Object (readonly)

Returns the value of attribute initial_delay.



31
32
33
 
# File 'lib/acemq/amqp/retry_policy.rb', line 31

def initial_delay
  @initial_delay
end

#jitter_factor ⇒ Object (readonly)

Returns the value of attribute jitter_factor.



31
32
33
 
# File 'lib/acemq/amqp/retry_policy.rb', line 31

def jitter_factor
  @jitter_factor
end

#max_attempts ⇒ Object (readonly)

Returns the value of attribute max_attempts.



31
32
33
 
# File 'lib/acemq/amqp/retry_policy.rb', line 31

def max_attempts
  @max_attempts
end

#max_delay ⇒ Object (readonly)

Returns the value of attribute max_delay.



31
32
33
 
# File 'lib/acemq/amqp/retry_policy.rb', line 31

def max_delay
  @max_delay
end

#max_message_age ⇒ Object (readonly)

Returns the value of attribute max_message_age.



31
32
33
 
# File 'lib/acemq/amqp/retry_policy.rb', line 31

def max_message_age
  @max_message_age
end

#multiplier ⇒ Object (readonly)

Returns the value of attribute multiplier.



31
32
33
 
# File 'lib/acemq/amqp/retry_policy.rb', line 31

def multiplier
  @multiplier
end

Class Method Details

.exponential(max_attempts, initial_delay, max_delay = 0.0) ⇒ Object

Doubling delays with 20% jitter, which is the sane default.

Parameters:

  • max_attempts (Integer) —

    total deliveries including the first

  • initial_delay (Float) —

    seconds before the second attempt

  • max_delay (Float) (defaults to: 0.0) —

    the ceiling, or 0 for none



61
62
63
64
 
# File 'lib/acemq/amqp/retry_policy.rb', line 61

def self.exponential(max_attempts, initial_delay, max_delay = 0.0)
  new(max_attempts: max_attempts, initial_delay: initial_delay,
      multiplier: 2.0, max_delay: max_delay, jitter_factor: 0.2)
end

.fixed(max_attempts, delay) ⇒ Object

The same wait every time, with no jitter.



67
68
69
 
# File 'lib/acemq/amqp/retry_policy.rb', line 67

def self.fixed(max_attempts, delay)
  new(max_attempts: max_attempts, initial_delay: delay, multiplier: 1.0)
end

.none ⇒ Object

One delivery, no second chance.



52
53
54
 
# File 'lib/acemq/amqp/retry_policy.rb', line 52

def self.none
  new(max_attempts: 1)
end

Instance Method Details

#give_up_after(age) ⇒ Object

A copy that abandons a message older than age seconds.

The honest limit when a queue has been paused: attempts say nothing about how long a message has been waiting.



75
76
77
 
# File 'lib/acemq/amqp/retry_policy.rb', line 75

def give_up_after(age)
  copy(max_message_age: age)
end

#jittered(delay) ⇒ Float

The same delay, moved either side by the jitter factor.

Both directions, so a fleet of consumers that failed together does not come back together. One-sided jitter only ever delays, which turns a thundering herd into a slower thundering herd.

Public because the retry engine applies it separately from working the delay out: a wait that happens in the broker needs no jitter at all, since each message's time-to-live starts when it enters the rung, so a fleet that failed over ten seconds is released over ten seconds.

Parameters:

  • delay (Float) —

    seconds

Returns:

  • (Float) 


134
135
136
137
138
 
# File 'lib/acemq/amqp/retry_policy.rb', line 134

def jittered(delay)
  return delay unless jitter_factor.positive? && delay.positive?

  delay * (1 + (((Kernel.rand * 2) - 1) * jitter_factor))
end

#next_delay(attempt, message_age = 0.0, jitter: true) ⇒ Float?

How long to wait before the next attempt, or nil to give up.

jitter: false gives the delay the schedule says, which is what to ask for when the number has to line up with something: a retry that waits in the broker waits in a queue named after its delay, and a jittered number names no queue. The retry engine asks for it that way, decides where the wait happens, and only jitters the waits it performs itself.

Parameters:

  • attempt (Integer) —

    the attempt that has just failed, from 1

  • message_age (Float) (defaults to: 0.0) —

    how old the message is, in seconds

  • jitter (Boolean) (defaults to: true) —

    whether to spread the delay

Returns:

  • (Float, nil) —

    the delay in seconds, or nil for no further attempt



96
97
98
99
100
101
102
 
# File 'lib/acemq/amqp/retry_policy.rb', line 96

def next_delay(attempt, message_age = 0.0, jitter: true)
  return nil if attempt >= max_attempts
  return nil if max_message_age.positive? && message_age >= max_message_age

  delay = backoff(attempt)
  [jitter ? jittered(delay) : delay, 0.0].max
end

#schedule ⇒ Array<Float>

The delays this policy would use, without jitter.

What to look at when deciding whether a policy is the one you meant: four numbers are easier to argue with than three parameters.

Returns:

  • (Array<Float>) 


110
111
112
113
114
115
116
117
118
119
 
# File 'lib/acemq/amqp/retry_policy.rb', line 110

def schedule
  delays = []
  delay = initial_delay
  (1...max_attempts).each do
    capped = max_delay.positive? && delay > max_delay ? max_delay : delay
    delays << capped
    delay *= multiplier
  end
  delays
end

#with_jitter(factor) ⇒ Object

A copy using a different jitter factor, between 0 and 1.



80
81
82
 
# File 'lib/acemq/amqp/retry_policy.rb', line 80

def with_jitter(factor)
  copy(jitter_factor: factor)
end