From 2091f072dc3803170edfc0d38d3f6bcad52c9168 Mon Sep 17 00:00:00 2001 From: smudge Date: Mon, 13 Jul 2026 17:03:52 -0400 Subject: [PATCH 1/5] feat: add delayed_limits table migration --- db/migrate/10_create_delayed_limits.rb | 30 ++++++++++++++++++++++++++ spec/helper.rb | 7 ++++-- 2 files changed, 35 insertions(+), 2 deletions(-) create mode 100644 db/migrate/10_create_delayed_limits.rb diff --git a/db/migrate/10_create_delayed_limits.rb b/db/migrate/10_create_delayed_limits.rb new file mode 100644 index 0000000..a163f77 --- /dev/null +++ b/db/migrate/10_create_delayed_limits.rb @@ -0,0 +1,30 @@ +class CreateDelayedLimits < ActiveRecord::Migration[6.0] + # The `delayed_limits` table backs the `Delayed::Limit` concurrency limiter. + # (See `Delayed::Limit` for details.) + # + # You can delete this migration if you do not intend to use `Delayed::Limit`, + # but it is safe to leave the table in place. + def up + create_table :delayed_limits, primary_key: :purpose, id: :string do |t| + t.datetime :drained_at, null: false + end + + return unless connection.adapter_name == 'PostgreSQL' + + # As this is a small, extremely high-churn table, we make it UNLOGGED (the + # limiter state need not survive a crash) and tune fillfactor/autovacuum to + # favor in-page [HOT updates](https://www.postgresql.org/docs/current/storage-hot.html). + execute <<~SQL + ALTER TABLE delayed_limits SET UNLOGGED; + ALTER TABLE delayed_limits SET ( + fillfactor = 33, + autovacuum_vacuum_scale_factor = 0, + autovacuum_vacuum_threshold = 30 + ); + SQL + end + + def down + drop_table :delayed_limits + end +end diff --git a/spec/helper.rb b/spec/helper.rb index b34ed4f..44914b2 100644 --- a/spec/helper.rb +++ b/spec/helper.rb @@ -73,8 +73,10 @@ def current_adapter ActiveRecord::Schema.define do if ActiveRecord::VERSION::MAJOR >= 7 drop_table :delayed_jobs, if_exists: true - elsif ActiveRecord::Base.connection.table_exists?(:delayed_jobs) - drop_table :delayed_jobs + drop_table :delayed_limits, if_exists: true + else + drop_table :delayed_jobs if ActiveRecord::Base.connection.table_exists?(:delayed_jobs) + drop_table :delayed_limits if ActiveRecord::Base.connection.table_exists?(:delayed_limits) end # Let's prove reversibility when we set up our test DB: @@ -94,6 +96,7 @@ def run_migration(klass) run_migration(RemoveLegacyIndex) run_migration(AddRunAtAndNameNotNullCheck) run_migration(ValidateRunAtAndNameNotNull) + run_migration(CreateDelayedLimits) # Test that these index migrations can be re-applied idempotently. # (In case identical indexes had been manually applied previously.) From e063bc2e45c29e35d380ba081d8aefc80dd84735 Mon Sep 17 00:00:00 2001 From: smudge Date: Mon, 13 Jul 2026 17:06:41 -0400 Subject: [PATCH 2/5] feat: add Delayed::Limit GCRA rate limiter --- app/models/delayed/limit.rb | 179 ++++++++++++++++++++++++++++++++++++ lib/delayed.rb | 1 + spec/delayed/limit_spec.rb | 174 +++++++++++++++++++++++++++++++++++ 3 files changed, 354 insertions(+) create mode 100644 app/models/delayed/limit.rb create mode 100644 spec/delayed/limit_spec.rb diff --git a/app/models/delayed/limit.rb b/app/models/delayed/limit.rb new file mode 100644 index 0000000..539b128 --- /dev/null +++ b/app/models/delayed/limit.rb @@ -0,0 +1,179 @@ +# frozen_string_literal: true + +module Delayed + # 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. + class Limit < ActiveRecord::Base + self.table_name = 'delayed_limits' + + # SQLite gained support for the `RETURNING` clause in version 3.35. + MINIMUM_SQLITE_VERSION = Gem::Version.new('3.35.0') + SECONDS_PER_DAY = 86_400.0 + + # Raised when limit adherence would require exceeding the `wait_timeout`. + class LimitExceededError < StandardError; end + + class << self + # Used only for limits registered in advance (via `.register!`): + def limits + @limits ||= {}.freeze + end + + # The algorithm requires an upserting `RETURNING` clause and timestamp + # arithmetic, so only certain database adapters/versions are supported: + 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 + + # 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. + def register!(purpose, max:, per:) + raise ArgumentError, "Limit policy '#{purpose}' already registered" if limits.key?(purpose.to_sym) + + @limits = limits.merge(purpose.to_sym => { max: max, per: per }.freeze).freeze + end + + # 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) + # + # IMPORTANT: For best results, the calling code MUST make its best attempt + # to sleep for the returned 'wait' duration before proceeding. (Sleeping is + # what shapes traffic to a smooth rate; it's best-effort, subject to + # GIL / OS scheduling variability.) + 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. + wait = limit.wait.to_f + sleep(wait) if wait.positive? + + ActiveSupport::Notifications.instrument('delayed.limit.within_limit', purpose: purpose) + yield + end + + private + + # We reserve capacity by pushing a 'drained_at' timestamp forward by the + # drain_interval, returning how long the caller must wait to avoid filling + # the bucket beyond its capacity. (The WHERE clause significantly reduces + # contention on the row when the bucket is full.) + def reserve_sql(purpose, drain_interval, max_wait) + case connection.adapter_name + when 'PostgreSQL', 'PostGIS' + # Postgres has native `interval` arithmetic and a statement-stable + # clock (`statement_timestamp()`), so we bind the intervals as ISO8601 + # strings and let the database do the math. + binds = { purpose: purpose, drain_interval: drain_interval.iso8601, max_wait: max_wait.iso8601 } + [<<~SQL.squish, binds] + INSERT INTO delayed_limits (purpose, drained_at) + VALUES (:purpose, statement_timestamp() + :drain_interval) + ON CONFLICT (purpose) DO UPDATE SET + drained_at = GREATEST(statement_timestamp(), delayed_limits.drained_at) + :drain_interval + WHERE delayed_limits.drained_at - statement_timestamp() <= :max_wait + RETURNING EXTRACT(EPOCH FROM (drained_at - statement_timestamp() - :drain_interval)) AS wait + SQL + when 'SQLite' + raise NotImplementedError, "Delayed::Limit requires SQLite #{MINIMUM_SQLITE_VERSION} or newer" unless supported? + + # SQLite has no interval type. Instead, `julianday` yields a + # sub-second-precise number of days, and we bind the intervals as a + # fraction of a day. (The `wait` is then scaled back to seconds.) + binds = { purpose: purpose, drain_interval: drain_interval.to_f / SECONDS_PER_DAY, max_wait: max_wait.to_f / SECONDS_PER_DAY } + [<<~SQL.squish, binds] + INSERT INTO delayed_limits (purpose, drained_at) + VALUES (:purpose, strftime('%Y-%m-%d %H:%M:%f', julianday('now') + :drain_interval)) + ON CONFLICT (purpose) DO UPDATE SET + drained_at = strftime('%Y-%m-%d %H:%M:%f', + MAX(julianday('now'), julianday(delayed_limits.drained_at)) + :drain_interval) + WHERE julianday(delayed_limits.drained_at) - julianday('now') <= :max_wait + RETURNING (julianday(drained_at) - julianday('now') - :drain_interval) * #{SECONDS_PER_DAY} AS wait + SQL + else + raise NotImplementedError, "Delayed::Limit is not supported on #{connection.adapter_name.inspect}" + end + end + end + end +end diff --git a/lib/delayed.rb b/lib/delayed.rb index fb4f8d7..7e3eb69 100644 --- a/lib/delayed.rb +++ b/lib/delayed.rb @@ -22,6 +22,7 @@ else require 'active_record' require_relative '../app/models/delayed/job' + require_relative '../app/models/delayed/limit' end ActiveSupport.on_load(:active_job) do diff --git a/spec/delayed/limit_spec.rb b/spec/delayed/limit_spec.rb new file mode 100644 index 0000000..2294e72 --- /dev/null +++ b/spec/delayed/limit_spec.rb @@ -0,0 +1,174 @@ +# frozen_string_literal: true + +require 'helper' + +RSpec::Matchers.define_negated_matcher :not_change, :change + +RSpec.describe Delayed::Limit do + before do + skip "Delayed::Limit is not supported on this version of #{current_adapter}" unless described_class.supported? + end + + after { described_class.delete_all } + + let(:purpose) { :test_bucket } + let(:limit) { { max: 1, per: 1.second } } + let(:expected_drain_rate) { 1.second } + + describe '.within_limit' do + it 'creates a limit row with drained_at set to now + drain_rate' do + now = Time.now.utc + expect { |b| described_class.within_limit(purpose, **limit, &b) } + .to change { described_class.count }.by(1) + .and yield_control.once + + expect(described_class.first.drained_at).to be_within(0.5.seconds).of(now + expected_drain_rate) + end + + it 'emits a notification when within the limit' do + expect { described_class.within_limit(purpose, **limit) { nil } } + .to emit_notification('delayed.limit.within_limit').with_payload(purpose: purpose) + end + + context 'with an existing limit row whose drained_at is in the past' do + before { described_class.create!(purpose: purpose, drained_at: 1.hour.ago) } + + it 'resets drained_at to now + drain_rate (does not accumulate against past time)' do + now = Time.now.utc + expect { |b| described_class.within_limit(purpose, **limit, &b) } + .to yield_control.once + .and not_change { described_class.count } + + expect(described_class.first.drained_at).to be_within(0.5.seconds).of(now + expected_drain_rate) + end + end + + context 'with an existing limit row whose drained_at is in the future' do + before { described_class.create!(purpose: purpose, drained_at: 10.seconds.from_now) } + + context 'when wait_timeout is within the returned wait time' do + it 'sleeps for the returned wait time and advances drained_at by drain_rate' do + expect(described_class).to receive(:sleep).with(a_value_within(1).of(10)).once + + expect { described_class.within_limit(purpose, **limit, wait_timeout: 15.seconds) { raise 'ohno' } } + .to raise_error(RuntimeError, 'ohno') + .and change { described_class.first.drained_at } + .by(a_value_within(0.5).of(expected_drain_rate).and(be_positive)) + end + end + + context 'when the returned wait time exceeds the wait_timeout' do + it 'raises LimitExceededError without advancing drained_at' do + expect { described_class.within_limit(purpose, **limit, wait_timeout: 1.second) { nil } } + .to raise_error(described_class::LimitExceededError) + .and not_change { described_class.first.drained_at } + end + + it 'emits a notification when the limit is exceeded' do + events = [] + callback = ->(name, *) { events << name } + ActiveSupport::Notifications.subscribed(callback, 'delayed.limit.exceeded') do + expect { described_class.within_limit(purpose, **limit, wait_timeout: 1.second) { nil } } + .to raise_error(described_class::LimitExceededError) + end + expect(events).to eq(['delayed.limit.exceeded']) + end + end + + context 'when the bucket drains before a subsequent call' do + it 'succeeds on a later attempt' do + expect { described_class.within_limit(purpose, **limit, wait_timeout: 1.second) { nil } } + .to raise_error(described_class::LimitExceededError) + + described_class.where(purpose: purpose).update_all(drained_at: 1.minute.ago) # rubocop:disable Rails/SkipsModelValidations + + now = Time.now.utc + expect { |b| described_class.within_limit(purpose, **limit, &b) } + .to yield_control.once + + expect(described_class.first.drained_at).to be_within(0.5.seconds).of(now + expected_drain_rate) + end + end + end + + context 'with a sub-second drain rate (max: 100, per: 1.minute => 0.6s)' do + let(:limit) { { max: 100, per: 1.minute } } + let(:expected_drain_rate) { 0.6.seconds } + + it 'creates a limit row with drained_at set 0.6 seconds in the future' do + now = Time.now.utc + expect { |b| described_class.within_limit(purpose, **limit, &b) } + .to change { described_class.count }.by(1) + .and yield_control.once + + expect(described_class.first.drained_at).to be_within(0.5.seconds).of(now + expected_drain_rate) + end + + context 'with an existing limit row whose drained_at is in the future' do + before { described_class.create!(purpose: purpose, drained_at: 1.minute.from_now) } + + context 'when wait_timeout is within the returned wait time' do + it 'sleeps for the returned wait time and advances drained_at by drain_rate' do + expect(described_class).to receive(:sleep).with(a_value_within(1).of(60)).once + + expect { described_class.within_limit(purpose, **limit, wait_timeout: 2.minutes) { raise 'ohno' } } + .to raise_error(RuntimeError, 'ohno') + .and change { described_class.first.drained_at } + .by(a_value_within(0.5).of(expected_drain_rate).and(be_positive)) + end + end + + context 'when the returned wait time exceeds the wait_timeout' do + it 'raises LimitExceededError without advancing drained_at' do + expect { described_class.within_limit(purpose, **limit, wait_timeout: 1.second) { nil } } + .to raise_error(described_class::LimitExceededError) + .and not_change { described_class.first.drained_at } + end + end + end + end + + context 'with a limit registered in advance via .register!' do + before { allow(described_class).to receive(:limits).and_return(purpose => limit) } + + it 'looks up the limit by purpose without requiring max/per' do + now = Time.now.utc + expect { |b| described_class.within_limit(purpose, &b) } + .to change { described_class.count }.by(1) + .and yield_control.once + + expect(described_class.first.drained_at).to be_within(0.5.seconds).of(now + expected_drain_rate) + end + + context 'when max and per are also passed directly' do + it 'raises ArgumentError without reserving capacity' do + expect { described_class.within_limit(purpose, **limit) { nil } } + .to raise_error(ArgumentError, /already registered/) + .and not_change { described_class.count } + end + end + end + end + + describe '.register!' do + around do |example| + original_limits = described_class.limits.dup + example.run + ensure + described_class.instance_variable_set(:@limits, original_limits) + end + + it 'registers a policy that can be looked up' do + described_class.register!(:some_purpose, max: 10, per: 1.minute) + + expect(described_class.limits.fetch(:some_purpose)).to eq(max: 10, per: 1.minute) + end + + it 'raises when a policy is already registered for the purpose' do + described_class.register!(:some_purpose, max: 10, per: 1.minute) + + expect { described_class.register!(:some_purpose, max: 5, per: 1.second) } + .to raise_error(ArgumentError, /already registered/) + end + end +end From e72ed3186f39528a2208b21d7d872004739ee207 Mon Sep 17 00:00:00 2001 From: smudge Date: Mon, 13 Jul 2026 17:37:25 -0400 Subject: [PATCH 3/5] docs: document the Delayed::Limit feature --- README.md | 89 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) diff --git a/README.md b/README.md index 1162541..7555307 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,7 @@ migration paths where possible. * [Priority-based Alerting Threshholds](#priority-based-alerting-threshholds) * [Continuous Monitoring](#continuous-monitoring) * [Configuration](#configuration) +* [Rate Limiting Jobs](#rate-limiting-jobs) * [Migrating from other ActiveJob backends](#migrating-from-other-activejob-backends) * [Migrating from DelayedJob](#migrating-from-delayedjob) * [How to Contribute](#how-to-contribute) @@ -531,6 +532,94 @@ Delayed.logger = Rails.logger Delayed.default_log_level = 'info' ``` +## Rate Limiting Jobs + +The `Delayed::Limit` class provides a database-backed **concurrency limiter/optimizer** for jobs +(via a [Generic Cell Rate Algorithm](https://en.wikipedia.org/wiki/Generic_cell_rate_algorithm) +implemented in SQL+Ruby). Use it to (e.g.) stay under a third-party API's published rate limit, or +to keep from overwhelming a downstream datastore: + +```ruby +Delayed::Limit.within_limit(:widgets_api, max: 100, per: 1.minute) do + WidgetsApi.create_widget!(...) +end +``` + +It will then attempt to maximize throughput without exceeding the limit. (Because its state lives in +the database, the limit applies across every worker and process at once.) + +If the limit would be exceeded within a configurable timeout, **the call will immediately raise a +`Delayed::Limit::LimitExceededError`**. (Background jobs will, in turn, fail fast and retry with the +usual back-off behavior.) + + +#### Setup + +To use this feature, make sure you have a `delayed_limits` table, or run `rake +delayed:install:migrations` to add it (see [Database Setup](#database-setup)). + +As of now, **only PostgreSQL and SQLite (3.35+) are supported.** The primary SQL query relies on an +upserting `RETURNING` clause and database-native timestamp arithmetic. You can check the current +connection at runtime with `Delayed::Limit.supported?`. + +#### Traffic Shaping vs Traffic Enforcement + +By default, `within_limit` will `sleep` up to 5 seconds (or a specified `wait_timeout`) before +yielding to the block. (This behavior is subject to the usual GIL and OS scheduling, so treat the +configured rate as a best-effort target rather than a hard guarantee.) + +Use a longer `wait_timeout` for even better throughput smoothing, at the cost of blocking threads +for longer: + +```ruby +Delayed::Limit.within_limit(:outbound_traffic, max: 100, per: 1.minute, wait_timeout: 30.seconds) do + # A longer wait timeout is best for longer-running or lower-priority background jobs. +end +``` + +Or set it to `0` to fail fast: + +```ruby +Delayed::Limit.within_limit(:inbound_traffic, max: 5, per: 1.second, wait_timeout: 0) do + # A shorter wait timeout is best for high-throughput or synchronous contexts (like web requests). +end +``` + +#### Shared Limits + +To avoid repeating the same purpose's `max` and `per` across multiple call sites, register a limit +in advance (e.g. in an initializer): + +```ruby +Delayed::Limit.register!(:widgets_api, max: 100, per: 1.minute) +``` + +Then, reference it by just its name at each call site: + +```ruby +Delayed::Limit.within_limit(:widgets_api) do + WidgetsApi.create_widget!(...) +end +``` + +Registered limits are cached indefinitely in memory and are not thread-safe on write, so avoid +registering them dynamically or at runtime! + +#### Handling "Burst" Throughput + +As of now, **there is no "burst" capacity.** The limiter allows one call per "drain interval" (`per +/ max`), so a limit of 60-per-minute behaves identically to 1-per-second (and will not allow +more than 1 call in the first second). + +This is generally acceptable for background job processing (and for traffic shaping in general), but +may be revisited in the future in order to support use cases like like API traffic enforcement. + +#### Monitoring Limit Usage + +Each call emits an `ActiveSupport::Notification` (`delayed.limit.within_limit` or +`delayed.limit.exceeded`), so you can monitor limiting activity the same way you would any other +event (see [Monitoring Jobs & Workers](#monitoring-jobs--workers)). + ## Migrating from other ActiveJob backends For the most part, standard ActiveJob APIs should be fully compatible. However, when migrating from From b82d3d4b900d91a598d1e5ea2ebbbe8437338b42 Mon Sep 17 00:00:00 2001 From: smudge Date: Wed, 15 Jul 2026 18:59:18 -0400 Subject: [PATCH 4/5] feat: add Delayed::RateLimitable job mixin --- README.md | 97 ++++++++--- app/models/delayed/limit.rb | 19 ++- lib/delayed.rb | 2 + lib/delayed/limitable.rb | 83 +++++++++ spec/delayed/limit_spec.rb | 12 +- spec/delayed/limitable_spec.rb | 303 +++++++++++++++++++++++++++++++++ 6 files changed, 480 insertions(+), 36 deletions(-) create mode 100644 lib/delayed/limitable.rb create mode 100644 spec/delayed/limitable_spec.rb diff --git a/README.md b/README.md index 7555307..de02b40 100644 --- a/README.md +++ b/README.md @@ -537,21 +537,28 @@ Delayed.default_log_level = 'info' The `Delayed::Limit` class provides a database-backed **concurrency limiter/optimizer** for jobs (via a [Generic Cell Rate Algorithm](https://en.wikipedia.org/wiki/Generic_cell_rate_algorithm) implemented in SQL+Ruby). Use it to (e.g.) stay under a third-party API's published rate limit, or -to keep from overwhelming a downstream datastore: +to keep from overwhelming a downstream datastore. + +The recommended interface is `with_limit`, provided via the `Delayed::Limitable` module (which is +included in all `ActiveJob` classes by default): ```ruby -Delayed::Limit.within_limit(:widgets_api, max: 100, per: 1.minute) do - WidgetsApi.create_widget!(...) +class TouchesThirdPartyApiJob < ApplicationJob + with_limit :third_party_api, max: 100, per: 1.minute + + def perform + # ... + end end ``` -It will then attempt to maximize throughput without exceeding the limit. (Because its state lives in -the database, the limit applies across every worker and process at once.) - -If the limit would be exceeded within a configurable timeout, **the call will immediately raise a -`Delayed::Limit::LimitExceededError`**. (Background jobs will, in turn, fail fast and retry with the -usual back-off behavior.) +The limiter will then attempt to maximize throughput without exceeding the limit. If the limit would +be exceeded within a configurable timeout, the job will immediately end and enqueue a retry attempt +with polynomial backoff. Because its state lives in the database, the limit applies across every +worker and process at once. +Plain (non-ActiveJob) classes may `include Delayed::Limitable` to use `with_limit`, but they must +define their own rescheduling/lifecycle behavior for `Delayed::Limit::LimitExceededError` errors. #### Setup @@ -564,42 +571,78 @@ connection at runtime with `Delayed::Limit.supported?`. #### Traffic Shaping vs Traffic Enforcement -By default, `within_limit` will `sleep` up to 5 seconds (or a specified `wait_timeout`) before -yielding to the block. (This behavior is subject to the usual GIL and OS scheduling, so treat the -configured rate as a best-effort target rather than a hard guarantee.) +By default, the limiter will `sleep` up to 5 seconds (or a specified `wait_timeout`) before +yielding to the limited work. (This behavior is subject to the usual GIL and OS scheduling, so +treat the configured rate as a best-effort target rather than a hard guarantee.) -Use a longer `wait_timeout` for even better throughput smoothing, at the cost of blocking threads -for longer: +Use a longer `wait_timeout` for even better throughput smoothing (at the cost of blocking threads): ```ruby -Delayed::Limit.within_limit(:outbound_traffic, max: 100, per: 1.minute, wait_timeout: 30.seconds) do - # A longer wait timeout is best for longer-running or lower-priority background jobs. -end +# A longer wait timeout is best for shaping outbound traffic in asynchronous contexts. +with_limit :outbound_traffic, max: 100, per: 1.minute, wait_timeout: 30.seconds +``` + +Or set it to `0` to fail fast, so that the job never blocks a worker thread: + +```ruby +# A zero wait timeout is best for enforcing inbound limits and shedding excess traffic. +with_limit :inbound_traffic, max: 5, per: 1.second, wait_timeout: 0 +``` + +#### Customizing `with_limit` + +The purpose defaults to the job's underscored class name, so it may be omitted entirely if the +limit is not shared with any other class: + +```ruby +with_limit max: 100, per: 1.minute ``` -Or set it to `0` to fail fast: +Use `on:` to wrap one or more other instance methods instead of `perform` (e.g. if only a portion +of the job's work is subject to the limit): + +```ruby +with_limit :third_party_api, on: :deliver! +``` + +**For ActiveJob classes only**, use `retry_attempts:`, `retry_wait:`, and `retry_jitter:` to +customize the retry behavior. (By default, jobs retry indefinitely with a polynomial backoff, with +the `wait_timeout` acting as a floor on the computed wait.) If `with_limit` is declared multiple +times on the same class (e.g. to apply different limits to different methods), only the first +declaration defines the job's retry behavior. + +#### Manually Limiting a Block of Code + +To rate limit code that doesn't belong to a job class, call `Delayed::Limit.within_limit` +directly. It accepts the same `purpose`, `max:`, `per:`, and `wait_timeout:` arguments as +`with_limit`, and wraps the limited work in a block: ```ruby -Delayed::Limit.within_limit(:inbound_traffic, max: 5, per: 1.second, wait_timeout: 0) do - # A shorter wait timeout is best for high-throughput or synchronous contexts (like web requests). +Delayed::Limit.within_limit(:widgets_api, max: 100, per: 1.minute) do + WidgetsApi.create_widget!(...) end ``` +The key difference is that there is no built-in retry behavior: if the limit would be exceeded +within the `wait_timeout`, the call raises `Delayed::Limit::LimitExceededError` immediately, and +it is up to the caller to rescue and/or retry. + #### Shared Limits -To avoid repeating the same purpose's `max` and `per` across multiple call sites, register a limit -in advance (e.g. in an initializer): +To avoid repeating the same purpose's `max` and `per` across multiple call sites, register a +limit in advance (e.g. in an initializer): ```ruby Delayed::Limit.register!(:widgets_api, max: 100, per: 1.minute) ``` -Then, reference it by just its name at each call site: +Then, reference it by just its name at each declaration: ```ruby -Delayed::Limit.within_limit(:widgets_api) do - WidgetsApi.create_widget!(...) -end +with_limit :widgets_api + +# or: +Delayed::Limit.within_limit(:widgets_api) { ... } ``` Registered limits are cached indefinitely in memory and are not thread-safe on write, so avoid @@ -612,7 +655,7 @@ As of now, **there is no "burst" capacity.** The limiter allows one call per "dr more than 1 call in the first second). This is generally acceptable for background job processing (and for traffic shaping in general), but -may be revisited in the future in order to support use cases like like API traffic enforcement. +may be revisited in the future in order to support use cases like API traffic enforcement. #### Monitoring Limit Usage diff --git a/app/models/delayed/limit.rb b/app/models/delayed/limit.rb index 539b128..737096d 100644 --- a/app/models/delayed/limit.rb +++ b/app/models/delayed/limit.rb @@ -70,10 +70,17 @@ def supported? # 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. def register!(purpose, max:, per:) - raise ArgumentError, "Limit policy '#{purpose}' already registered" if limits.key?(purpose.to_sym) + 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 => { max: max, per: per }.freeze).freeze + @limits = limits.merge(purpose.to_sym => config).freeze end # This method implements a leaky bucket algorithm (or, more specifically, @@ -89,11 +96,6 @@ def register!(purpose, max:, per:) # - T (emission interval) -> drain_interval # - t0 (time of request) -> the database's current time # - τ (bucket capacity) -> 1 call (implicitly) - # - # IMPORTANT: For best results, the calling code MUST make its best attempt - # to sleep for the returned 'wait' duration before proceeding. (Sleeping is - # what shapes traffic to a smooth rate; it's best-effort, subject to - # GIL / OS scheduling variability.) def within_limit(purpose, max: nil, per: nil, wait_timeout: 5.seconds) config = limits[purpose.to_sym] if config && (max || per) @@ -126,6 +128,9 @@ def within_limit(purpose, max: nil, per: nil, wait_timeout: 5.seconds) # 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? diff --git a/lib/delayed.rb b/lib/delayed.rb index 7e3eb69..7afc667 100644 --- a/lib/delayed.rb +++ b/lib/delayed.rb @@ -16,6 +16,7 @@ require 'delayed/helpers/migration' require 'delayed/worker' require 'delayed/job_wrapper' +require 'delayed/limitable' if defined?(Rails::Engine) require 'delayed/engine' @@ -30,6 +31,7 @@ ActiveJob::QueueAdapters::DelayedAdapter = Class.new(Delayed::ActiveJobAdapter) include Delayed::ActiveJobAdapter::EnqueuingPatch + include Delayed::Limitable end ActiveSupport.on_load(:action_mailer) do diff --git a/lib/delayed/limitable.rb b/lib/delayed/limitable.rb new file mode 100644 index 0000000..732aee9 --- /dev/null +++ b/lib/delayed/limitable.rb @@ -0,0 +1,83 @@ +# frozen_string_literal: true + +module Delayed + # A mixin that wraps a class's `perform` method (or other methods named via + # `on:`) in `Delayed::Limit.within_limit`. It is automatically included in + # ActiveJob classes, and configures the job to retry with a polynomial + # backoff when the limit's `wait_timeout` would be exceeded: + # + # class TouchesThirdPartyApiJob < ApplicationJob + # with_limit :third_party_api, max: 100, per: 1.minute + # + # def perform + # # ... + # end + # end + # + # The 'purpose' defaults to the job's underscored class name, and the limit + # is registered via `Delayed::Limit.register!` (unless the purpose was + # already registered, e.g. in an initializer, in which case the `max:`/`per:` + # config may be omitted entirely). Multiple job classes may share a purpose + # (and its limit) as long as their configs match exactly. + # + # Use `on:` to wrap one or more other instance methods instead of `perform`, + # e.g. if only a portion of the job's work is subject to the limit: + # + # with_limit :third_party_api, max: 100, per: 1.minute, on: :deliver! + # + # A class may declare `with_limit` more than once (e.g. to apply different + # limits to different methods), but only the first declaration defines the + # job's retry behavior (`attempts`, `wait`, and `jitter`, with `wait_timeout` + # acting as a floor on the computed wait). If two declarations' wait timeouts + # differ meaningfully, declare the one with the longer `wait_timeout` first. + # + # Retries rely on ActiveJob's `retry_on`, so when this mixin is included in + # a plain (non-ActiveJob) class, the named methods are still wrapped in + # `within_limit`, but the class must define its own rescue/retry behavior + # for `Delayed::Limit::LimitExceededError`. + module Limitable + extend ActiveSupport::Concern + + DEFAULT_RETRY_ATTEMPTS = if defined?(ActiveJob) && ActiveJob.gem_version >= Gem::Version.new('7.0') + :unlimited + else + Float::INFINITY + end + + class_methods do + def with_limit( + purpose = name.underscore.to_sym, + on: :perform, + max: nil, + per: nil, + wait_timeout: 5.seconds, + retry_jitter: 0.1, + retry_attempts: DEFAULT_RETRY_ATTEMPTS, + retry_wait: ->(attempt) { polynomial_backoff(wait_timeout, attempt, retry_jitter) } + ) + Delayed::Limit.register!(purpose, max: max, per: per) if max || per + + if defined?(ActiveJob::Base) && self < ActiveJob::Base && + rescue_handlers.none? { |klass, _| klass == Delayed::Limit::LimitExceededError.name } + retry_on(Delayed::Limit::LimitExceededError, attempts: retry_attempts, wait: retry_wait) + end + + prepend(Module.new do + Array(on).each do |method_name| + define_method(method_name) do |*args, **kwargs, &block| + Delayed::Limit.within_limit(purpose, wait_timeout: wait_timeout) do + super(*args, **kwargs, &block) + end + end + end + end) + end + + private + + def polynomial_backoff(min_wait, attempt, jitter) + [min_wait, (attempt**4)].max * (1 + rand(-jitter..jitter)) + end + end + end +end diff --git a/spec/delayed/limit_spec.rb b/spec/delayed/limit_spec.rb index 2294e72..4135f52 100644 --- a/spec/delayed/limit_spec.rb +++ b/spec/delayed/limit_spec.rb @@ -164,11 +164,19 @@ expect(described_class.limits.fetch(:some_purpose)).to eq(max: 10, per: 1.minute) end - it 'raises when a policy is already registered for the purpose' do + it 'is a no-op when re-registering a purpose with an identical config' do + described_class.register!(:some_purpose, max: 10, per: 1.minute) + + expect { described_class.register!(:some_purpose, max: 10, per: 1.minute) } + .not_to change { described_class.limits.fetch(:some_purpose) } + .from(max: 10, per: 1.minute) + end + + it 'raises when a policy is already registered with a conflicting config' do described_class.register!(:some_purpose, max: 10, per: 1.minute) expect { described_class.register!(:some_purpose, max: 5, per: 1.second) } - .to raise_error(ArgumentError, /already registered/) + .to raise_error(ArgumentError, /already registered and does not match/) end end end diff --git a/spec/delayed/limitable_spec.rb b/spec/delayed/limitable_spec.rb new file mode 100644 index 0000000..a0a5683 --- /dev/null +++ b/spec/delayed/limitable_spec.rb @@ -0,0 +1,303 @@ +# frozen_string_literal: true + +require 'helper' + +RSpec.describe Delayed::Limitable do + let(:purpose) { :limitable_spec } + + around do |example| + original_limits = Delayed::Limit.limits + example.run + ensure + Delayed::Limit.instance_variable_set(:@limits, original_limits) + end + + before do + allow(Delayed::Limit).to receive(:within_limit).and_yield + end + + it 'is automatically included in ActiveJob classes' do + expect(ActiveJob::Base).to respond_to(:with_limit) + end + + describe '.with_limit' do + context 'when on: is not specified' do + let(:job_class) do + p = purpose + Class.new(ActiveJob::Base) do + with_limit p, max: 4, per: 1.second + + define_method(:perform) { :result } + end + end + + it 'wraps perform with within_limit' do + job_class.new.perform + expect(Delayed::Limit).to have_received(:within_limit) + .with(purpose, wait_timeout: 5.seconds).once + end + + it 'registers the limit policy with Delayed::Limit' do + job_class + expect(Delayed::Limit.limits.fetch(purpose)).to eq(max: 4, per: 1.second) + end + end + + context 'when purpose is not specified' do + let(:job_class) do + Class.new(ActiveJob::Base) do + define_method(:perform) { :result } + end + end + + before do + stub_const('LimitableDefaultPurposeJob', job_class) + job_class.with_limit(max: 4, per: 1.second) + end + + it 'defaults the purpose to the underscored class name' do + job_class.new.perform + expect(Delayed::Limit).to have_received(:within_limit) + .with(:limitable_default_purpose_job, wait_timeout: 5.seconds).once + end + end + + context 'when wait_timeout: is specified' do + let(:job_class) do + p = purpose + Class.new(ActiveJob::Base) do + with_limit p, max: 4, per: 1.second, wait_timeout: 30.seconds + + define_method(:perform) { :result } + end + end + + it 'passes the wait_timeout through to within_limit' do + job_class.new.perform + expect(Delayed::Limit).to have_received(:within_limit) + .with(purpose, wait_timeout: 30.seconds).once + end + end + + context 'when on: is a single method name' do + let(:job_class) do + p = purpose + Class.new(ActiveJob::Base) do + with_limit p, max: 4, per: 1.second, on: :do_work + + define_method(:perform) { do_work && dont_do_work } + define_method(:do_work) { :result_b } + define_method(:dont_do_work) { :result_a } + end + end + + it 'wraps the specified method with within_limit' do + job_class.new.do_work + expect(Delayed::Limit).to have_received(:within_limit) + .with(purpose, wait_timeout: 5.seconds).once + end + + it 'does not add an extra within_limit call when perform delegates to the wrapped method' do + job_class.new.perform + expect(Delayed::Limit).to have_received(:within_limit).once + end + end + + context 'when on: is an array of method names' do + let(:job_class) do + p = purpose + Class.new(ActiveJob::Base) do + with_limit p, max: 4, per: 1.second, on: %i(do_work_a do_work_b) + + def perform; end + define_method(:do_work_a) { :result_a } + define_method(:do_work_b) { :result_b } + end + end + + it 'wraps each named method with within_limit' do + job_class.new.tap do |job| + job.perform + job.do_work_a + job.do_work_b + end + expect(Delayed::Limit).to have_received(:within_limit) + .with(purpose, wait_timeout: 5.seconds).twice + end + end + + context 'when the purpose was registered in advance and no config is given' do + before { Delayed::Limit.register!(purpose, max: 4, per: 1.second) } + + let(:job_class) do + p = purpose + Class.new(ActiveJob::Base) do + with_limit p + + define_method(:perform) { :result } + end + end + + it 'references the registered limit by purpose' do + job_class.new.perform + expect(Delayed::Limit).to have_received(:within_limit) + .with(purpose, wait_timeout: 5.seconds).once + end + end + + context 'when the config contains an unknown key' do + it 'raises ArgumentError' do + p = purpose + expect { + Class.new(ActiveJob::Base) { with_limit p, max: 4, per: 1.second, wait_timout: 1.second } + }.to raise_error(ArgumentError, /wait_timout/) + end + end + + context 'when with_limit is declared multiple times' do + let(:other_purpose) { :"#{purpose}_b" } + + let(:job_class) do + p = purpose + other = other_purpose + Class.new(ActiveJob::Base) do + with_limit p, max: 4, per: 1.second, wait_timeout: 100.seconds, on: :do_work_a, retry_jitter: 0 + with_limit other, max: 2, per: 1.minute, wait_timeout: 30.seconds, on: :do_work_b, retry_jitter: 0 + + define_method(:perform) { raise Delayed::Limit::LimitExceededError } + define_method(:do_work_a) { :result_a } + define_method(:do_work_b) { :result_b } + end + end + + before { stub_const('LimitableMultipleLimitsJob', job_class) } + + it 'registers each limit policy' do + job_class + expect(Delayed::Limit.limits.fetch(purpose)).to eq(max: 4, per: 1.second) + expect(Delayed::Limit.limits.fetch(other_purpose)).to eq(max: 2, per: 1.minute) + end + + it 'wraps each method with its own limit' do + job_class.new.tap do |job| + job.do_work_a + job.do_work_b + end + expect(Delayed::Limit).to have_received(:within_limit) + .with(purpose, wait_timeout: 100.seconds).once + expect(Delayed::Limit).to have_received(:within_limit) + .with(other_purpose, wait_timeout: 30.seconds).once + end + + it 'defines only a single retry handler' do + handlers = job_class.rescue_handlers.select do |class_name, _| + class_name == Delayed::Limit::LimitExceededError.name + end + expect(handlers.size).to eq(1) + end + + it 'retries according to the first declaration (its wait_timeout floors the retry wait)' do + job = job_class.new + allow(job).to receive(:retry_job) + job.perform_now + expect(job).to have_received(:retry_job) + .with(a_hash_including(wait: 100.seconds)) + end + end + + context 'when included in a plain (non-ActiveJob) class' do + let(:klass) do + p = purpose + Class.new do + include Delayed::Limitable + + with_limit p, max: 4, per: 1.second, on: :do_work + + define_method(:do_work) { :result } + end + end + + it 'wraps the specified method with within_limit' do + expect(klass.new.do_work).to eq(:result) + expect(Delayed::Limit).to have_received(:within_limit) + .with(purpose, wait_timeout: 5.seconds).once + end + + it 'registers the limit policy with Delayed::Limit' do + klass + expect(Delayed::Limit.limits.fetch(purpose)).to eq(max: 4, per: 1.second) + end + + it 'leaves LimitExceededError handling to the caller' do + allow(Delayed::Limit).to receive(:within_limit).and_raise(Delayed::Limit::LimitExceededError) + expect { klass.new.do_work }.to raise_error(Delayed::Limit::LimitExceededError) + end + end + end + + describe 'retry behavior' do + let(:job_class) do + p = purpose + Class.new(ActiveJob::Base) do + with_limit p, max: 4, per: 1.second, retry_jitter: 0 + + define_method(:perform) { raise Delayed::Limit::LimitExceededError } + end + end + + before { stub_const('LimitableRetryJob', job_class) } + + it 'retries the job when the limit is exceeded' do + job = job_class.new + allow(job).to receive(:retry_job) + job.perform_now + expect(job).to have_received(:retry_job) + .with(a_hash_including(wait: a_value > 0, error: an_instance_of(Delayed::Limit::LimitExceededError))) + end + + it 'floors the retry wait at the wait_timeout' do + job = job_class.new + allow(job).to receive(:retry_job) + job.perform_now + expect(job).to have_received(:retry_job) + .with(a_hash_including(wait: 5.seconds)) + end + + context 'when a custom wait: is specified' do + let(:job_class) do + p = purpose + Class.new(ActiveJob::Base) do + with_limit p, max: 4, per: 1.second, retry_wait: ->(_attempts) { 123 } + + define_method(:perform) { raise Delayed::Limit::LimitExceededError } + end + end + + it 'uses the custom wait' do + job = job_class.new + allow(job).to receive(:retry_job) + job.perform_now + expect(job).to have_received(:retry_job) + .with(a_hash_including(wait: 123)) + end + end + + context 'when retry_attempts: is specified' do + let(:job_class) do + p = purpose + Class.new(ActiveJob::Base) do + with_limit p, max: 4, per: 1.second, retry_attempts: 1 + + define_method(:perform) { raise Delayed::Limit::LimitExceededError } + end + end + + it 'stops retrying once the attempts are exhausted' do + job = job_class.new + allow(job).to receive(:retry_job) + expect { job.perform_now }.to raise_error(Delayed::Limit::LimitExceededError) + expect(job).not_to have_received(:retry_job) + end + end + end +end From ed77720e9651f4b1eeec4d7b2624d24422e5e19b Mon Sep 17 00:00:00 2001 From: smudge Date: Fri, 17 Jul 2026 17:14:16 -0400 Subject: [PATCH 5/5] v3.2.0 --- Gemfile.lock | 2 +- lib/delayed/version.rb | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Gemfile.lock b/Gemfile.lock index bf469e6..e7c2ef1 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -1,7 +1,7 @@ PATH remote: . specs: - delayed (3.1.0) + delayed (3.2.0) activerecord (>= 6.0) concurrent-ruby diff --git a/lib/delayed/version.rb b/lib/delayed/version.rb index d5ac1e9..ff207e3 100644 --- a/lib/delayed/version.rb +++ b/lib/delayed/version.rb @@ -1,5 +1,5 @@ # frozen_string_literal: true module Delayed - VERSION = '3.1.0' + VERSION = '3.2.0' end