Class: Delayed::Limit

Inherits:
ActiveRecord::Base show all
Defined in:
app/models/delayed/limit.rb

Overview

A database-backed concurrency limiter/optimizer designed for use with Delayed::Job. Given a 'purpose' (stringy identifier), a limit, and a time interval, it will attempt to maximize throughput without exceeding the limit ("traffic shaping"). The wait_timeout parameter can also be lowered (from its default of 5.seconds) in order to shed load more proactively ("traffic enforcement").

Because the algorithm relies on database-specific timestamp arithmetic and an upserting RETURNING clause, only PostgreSQL and SQLite (3.35+) are supported. (See .supported?.)

Wrap the work you want to limit in a block, and the limiter will either yield to the block (within the configured wait_timeout) or raise LimitExceededError immediately (if the wait_timeout would be exceeded):

Delayed::Limit.within_limit(:emails, max: 100, per: 1.minute) do
deliver_email!
end

Currently, the limiter does not support "burst" capacity. A limit of 60 req/minute will behave identically to a limit of 1 req/second, because the limiter only allows a single call per "drain interval" (per / max). When combined with the wait_timeout, this degree of smoothing is acceptable for background job processing (and traffic shaping in general), but may be revisited in the future to support more types of workloads.

To share a limit across multiple calls, register a named 'purpose' in advance (e.g. in an initializer) and then reference it by name later:

Delayed::Limit.register!(:emails, max: 100, per: 1.minute)

Delayed::Limit.within_limit(:emails, wait_timeout: 10.seconds) do
deliver_email!
end

It is not recommended to register limits dynamically at runtime, because registered limits are cached indefinitely in memory and are not thread-safe.

Defined Under Namespace

Classes: LimitExceededError, UnsupportedDatabaseError

Constant Summary collapse

MINIMUM_SQLITE_VERSION =

SQLite gained support for the RETURNING clause in version 3.35.

Gem::Version.new('3.35.0')
SECONDS_PER_DAY = 
86_400.0

Class Method Summary collapse

Methods inherited from ActiveRecord::Base

#to_yaml_properties, yaml_new

Class Method Details

.limits ⇒ Object

Used only for limits registered in advance (via .register!):



57
58
59
 
# File 'app/models/delayed/limit.rb', line 57

def limits
  @limits ||= {}.freeze
end

.register!(purpose, max:, per:) ⇒ Object

Register a limit policy for a given 'purpose' (stringy/symbol identifier). This is optional and should not be used for dynamic purpose names or limits, for memory and thread-safety reasons.

Re-registering a purpose is a no-op if the config matches exactly, and raises ArgumentError otherwise.



80
81
82
83
84
85
86
87
88
 
# File 'app/models/delayed/limit.rb', line 80

def register!(purpose, max:, per:)
  config = { max: max, per: per }.freeze

  if limits.key?(purpose.to_sym) && limits.fetch(purpose.to_sym) != config
    raise ArgumentError, "Limit policy '#{purpose}' is already registered and does not match #{config.inspect}"
  end

  @limits = limits.merge(purpose.to_sym => config).freeze
end

.supported? ⇒ Boolean

The algorithm requires an upserting RETURNING clause and timestamp arithmetic, so only certain database adapters/versions are supported:

Returns:

  • (Boolean) 


63
64
65
66
67
68
69
70
71
72
 
# File 'app/models/delayed/limit.rb', line 63

def supported?
  case connection.adapter_name
  when 'PostgreSQL', 'PostGIS'
    true
  when 'SQLite'
    Gem::Version.new(connection.select_value('SELECT sqlite_version()')) >= MINIMUM_SQLITE_VERSION
  else
    false
  end
end

.within_limit(purpose, max: nil, per: nil, wait_timeout: 5.seconds) ⇒ Object

This method implements a leaky bucket algorithm (or, more specifically, a Generic Cell Rate Algorithm) to enforce a per-'purpose' work limit.

It will wait up to wait_timeout for the caller to come within the configured limit before yielding to the caller, and will raise LimitExceededError if the wait time would exceed that timeout (shedding the caller proactively rather than sleeping).

In Generic Cell Rate Algorithm (GCRA) terms:

  • TAT (theoretical arrival time) -> drained_at
  • T (emission interval) -> drain_interval
  • t0 (time of request) -> the database's current time
  • τ (bucket capacity) -> 1 call (implicitly)


103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'app/models/delayed/limit.rb', line 103

def within_limit(purpose, max: nil, per: nil, wait_timeout: 5.seconds)
  config = limits[purpose.to_sym]
  if config && (max || per)
    raise ArgumentError, "Limit policy '#{purpose}' is already registered (overriding 'max'/'per' is not supported)"
  end

  config ||= { max: max, per: per }.compact

  # The drain_interval reflects the per-call rate at which the bucket
  # empties, calculated as the overall interval (per) divided by the
  # maximum number of calls allowed in that interval (max).
  #
  # e.g. for a target of 100 req/min, the drain_interval would be 0.6 sec/req.
  drain_interval = config.fetch(:per).seconds / config.fetch(:max).to_d

  # Attempt to reserve capacity via an uncached database query:
  limit = connection.uncached do
    find_by_sql(reserve_sql(purpose, drain_interval, wait_timeout.seconds)).first
  end

  # If 'limit' is nil, it means the WHERE clause prevented us from
  # reserving capacity in the bucket. (This happens if the configured
  # `wait_timeout` would be exceeded.) We assume the caller will back off
  # and retry later, so we avoid wasting bucket capacity on a no-op.
  if limit.nil?
    ActiveSupport::Notifications.instrument('delayed.limit.exceeded', purpose: purpose)
    raise LimitExceededError, "Concurrency limit exceeded for '#{purpose}'"
  end

  # If we successfully reserved capacity within the `wait_timeout`, it
  # means that we've been told by the query how long to sleep in order to
  # comply with the configured rate.
  #
  # (For best results, we MUST make a best attempt to sleep for the
  # returned 'wait' duration before proceeding.)
  wait = limit.wait.to_f
  sleep(wait) if wait.positive?

  ActiveSupport::Notifications.instrument('delayed.limit.within_limit', purpose: purpose)
  yield
end