diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index a0399e026..738c21b1f 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -22,7 +22,7 @@ env:
 
 jobs:
   test:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
 
     services:
       postgres:
diff --git a/Gemfile.lock b/Gemfile.lock
index fd44332d4..70183734e 100644
--- a/Gemfile.lock
+++ b/Gemfile.lock
@@ -2,7 +2,7 @@ PATH
   remote: .
   specs:
     rpush (7.1.0)
-      activesupport (>= 5.2, < 7.1.0)
+      activesupport (>= 5.1)
       googleauth
       jwt (>= 1.5.6)
       multi_json (~> 1.0)
@@ -92,7 +92,7 @@ GEM
       crass (~> 1.0.2)
       nokogiri (>= 1.12.0)
     method_source (1.0.0)
-    mini_portile2 (2.8.5)
+    mini_portile2 (2.8.9)
     minitest (5.24.1)
     modis (4.3.0)
       activemodel (>= 5.2)
@@ -192,7 +192,7 @@ GEM
       simplecov_json_formatter (~> 0.1)
     simplecov-html (0.12.3)
     simplecov_json_formatter (0.1.4)
-    sqlite3 (2.0.2)
+    sqlite3 (1.6.9)
       mini_portile2 (~> 2.8.0)
     stackprof (0.2.17)
     stringio (3.1.1)
@@ -209,6 +209,7 @@ GEM
 
 PLATFORMS
   ruby
+  x86_64-linux
 
 DEPENDENCIES
   appraisal
@@ -226,7 +227,7 @@ DEPENDENCIES
   rubocop (~> 1.12.0)
   rubocop-performance
   simplecov
-  sqlite3
+  sqlite3 (~> 1.6.0)
   stackprof
   timecop
 
diff --git a/lib/rpush/client/redis/notification.rb b/lib/rpush/client/redis/notification.rb
index 856cc8bf7..9f0d784a7 100644
--- a/lib/rpush/client/redis/notification.rb
+++ b/lib/rpush/client/redis/notification.rb
@@ -8,10 +8,33 @@ class Notification
 
         enable_all_index false # prevent creation of massive rpush:notifications:all set
 
+        before_create :stamp_enqueued_at
         after_create :register_notification
 
         self.namespace = 'notifications'
 
+        # Delivery priority band for the pending sorted set. Lower band == served first.
+        #
+        # The pending set is scored so that ascending rank == ascending priority band,
+        # then ascending id within a band. The daemon claims by RANK
+        # (Store::Redis#pending_notification_ids uses ZRANGE 0..limit +
+        # ZREMRANGEBYRANK 0..limit), so ordering the score is sufficient to order
+        # delivery -- no change to the claim path is required.
+        #
+        # BAND is anchored on the DEFAULT class, not on zero, so that a notification
+        # at PRIORITY_CLASS_DEFAULT scores exactly its own id. That keeps the pending
+        # set byte-identical to pre-priority behaviour for every existing caller, which
+        # means (a) no transition window while a mixed-version fleet rolls, (b) no
+        # score inversion between old and new writers, and (c) a revert is score-safe.
+        # A class below the default scores negative and therefore sorts ahead of all
+        # default traffic.
+        #
+        # BAND must exceed the maximum notification id and stay inside the exact
+        # integer range of an IEEE-754 double (2**53), since Redis sorted-set scores
+        # are doubles. At 10**12 that holds for ids up to ~9.0e15.
+        PRIORITY_CLASS_DEFAULT = 1
+        PRIORITY_CLASS_BAND    = 10**12
+
         def self.absolute_pending_namespace
           "#{absolute_namespace}:pending"
         end
@@ -51,11 +74,38 @@ def self.absolute_retryable_namespace
         attribute :notification, :hash
         attribute :thread_id, :string
 
+        # Delivery band; see PRIORITY_CLASS_BAND. Defaults to PRIORITY_CLASS_DEFAULT so
+        # that callers which never set it are indistinguishable from today.
+        attribute :priority_class, :integer, default: PRIORITY_CLASS_DEFAULT
+
+        # When the notification entered the pending set. Stamped by a before_create
+        # callback rather than an attribute default, because Modis evaluates `default:`
+        # once at class-definition time -- a `default: Time.now` would brand every
+        # notification with the timestamp of process boot.
+        #
+        # Exists so the daemon can report how long a notification waited between being
+        # enqueued and being handed to APNs/FCM. Nothing else in the model records
+        # creation time: there is no created_at attribute.
+        attribute :enqueued_at, :timestamp
+
         def app
           return nil unless app_id
           @app ||= Rpush::Client::Redis::App.find(app_id)
         end
 
+        # Milliseconds between entering the pending set and now, or nil when the
+        # notification predates the enqueued_at attribute (every notification already
+        # resident in Redis at deploy time). Callers MUST handle nil.
+        #
+        # enqueued_at has whole-second resolution and is floored, so this is biased
+        # 0..999 ms high. That is immaterial against the multi-second budgets it exists
+        # to measure, but do not build sub-second assertions on it.
+        def age_ms(now = Time.now)
+          return nil unless enqueued_at
+
+          ((now - enqueued_at) * 1000).round
+        end
+
         def app=(app)
           @app = app
           if app
@@ -67,11 +117,25 @@ def app=(app)
 
         private
 
+        def stamp_enqueued_at
+          self.enqueued_at ||= Time.now
+        end
+
         def register_notification
           Modis.with_connection do |redis|
-            redis.zadd(self.class.absolute_pending_namespace, id, id)
+            redis.zadd(self.class.absolute_pending_namespace, pending_score, id)
           end
         end
+
+        # Anchored on the default band so default traffic scores exactly `id`.
+        #
+        # `|| PRIORITY_CLASS_DEFAULT` rather than `.to_i`: nil.to_i is 0, which would
+        # silently PROMOTE a notification whose priority_class failed to load into the
+        # highest band -- the unsafe direction. Falling back to the default is inert.
+        def pending_score
+          band = (priority_class || PRIORITY_CLASS_DEFAULT) - PRIORITY_CLASS_DEFAULT
+          (band * PRIORITY_CLASS_BAND) + id
+        end
       end
     end
   end
diff --git a/lib/rpush/daemon/store/redis.rb b/lib/rpush/daemon/store/redis.rb
index 54808e8cd..d88693e8b 100644
--- a/lib/rpush/daemon/store/redis.rb
+++ b/lib/rpush/daemon/store/redis.rb
@@ -4,6 +4,33 @@ module Store
       class Redis
         DEFAULT_MARK_OPTIONS = { persist: true }
 
+        # Ceiling on the share of a poll the retryable set may take in its first pass, so
+        # a deep retryable set cannot consume the budget and starve pending. It is a CAP,
+        # not a reservation: pending takes everything retryable leaves, and a second
+        # retryable pass mops up anything pending did not use -- so neither side starves
+        # and no slot goes idle while either side has work.
+        RETRYABLE_CLAIM_SHARE = 0.2
+
+        # Claim at most ARGV[2] members scoring at or below ARGV[1] (i.e. DUE), lowest
+        # score first, and remove exactly those members.
+        #
+        # A read-then-remove pair cannot be made safe here even inside MULTI, because
+        # MULTI cannot branch on the result of its own ZRANGE: a rank-bounded removal
+        # races another daemon's claim and can then remove -- and deliver -- a retry whose
+        # backoff has not elapsed. ZREMRANGEBYSCORE is no escape either; it would remove
+        # every due member while returning only `limit` of them, losing the rest. EVAL is
+        # what makes selection and removal one operation over one member set.
+        #
+        # `unpack` (Lua 5.1, which is what Redis embeds) is safe against a stack blowout
+        # here because the list is bounded by Rpush.config.batch_size.
+        RETRYABLE_CLAIM_SCRIPT = <<-LUA
+          local ids = redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', ARGV[1], 'LIMIT', 0, tonumber(ARGV[2]))
+          if #ids > 0 then
+            redis.call('ZREM', KEYS[1], unpack(ids))
+          end
+          return ids
+        LUA
+
         def app(app_id)
           Rpush::Client::ActiveRecord::App.find(app_id)
         end
@@ -12,10 +39,40 @@ def all_apps
           Rpush::Client::ActiveRecord::App.all
         end
 
+        # Splits a poll's budget between the two sources instead of letting retryable
+        # consume all of it.
+        #
+        # Previously `retryable_notification_ids` claimed EVERY due retry with no
+        # bound, then `limit` was decremented by that count. Whenever the due retryable
+        # set was at least `limit` deep, pending received exactly zero for the poll --
+        # and because the claimed batch also overshot `limit`, the Feeder's
+        # `batch_size - AppRunner.total_queued` short-circuit then suppressed subsequent
+        # polls until the oversized batch drained, extending the starvation well beyond
+        # a single poll. Measured live: a retryable set 35x batch_size deep.
+        #
+        # Retryable is claimed first but CAPPED, so it can no longer take the poll;
+        # pending then claims the ENTIRE remainder, so an empty retryable set costs
+        # nothing. Any budget neither side used goes back to retryable, so retryable is
+        # not starved when pending is shallow either.
+        #
+        # The cap sits on retryable rather than on pending on purpose. Reserving a fixed
+        # share for RETRYABLE instead -- and giving pending only the rest -- leaves that
+        # share unused on every poll where nothing is due, which is every poll of a large
+        # campaign drain. That is a flat ~20% off the top of the throughput the feeder is
+        # capable of, and the feeder's claim rate is the binding constraint on how long a
+        # campaign takes: measured at 5 tasks, deliveries plateaued at 2,672/min/task
+        # against the 3,000/min that batch_size 100 over a 2s push_poll allows, i.e. 89%
+        # of the claim ceiling, with dispatch, the database and Redis all idle-ish behind
+        # it. Losing a fifth of that is 10 minutes on a 2M-notification campaign.
         def deliverable_notifications(limit)
-          retryable_ids = retryable_notification_ids
-          limit -= retryable_ids.size
-          pending_ids = limit > 0 ? pending_notification_ids(limit) : []
+          return [] unless limit > 0
+
+          retryable_ids = retryable_notification_ids(retryable_budget(limit))
+          pending_ids   = pending_notification_ids(limit - retryable_ids.size)
+
+          unclaimed = limit - retryable_ids.size - pending_ids.size
+          retryable_ids += retryable_notification_ids(unclaimed) if unclaimed > 0
+
           ids = retryable_ids + pending_ids
           ids.map { |id| find_notification_by_id(id) }.compact
         end
@@ -161,21 +218,40 @@ def create_gcm_like_notification(notification, attrs, data, registration_ids, de
           notification
         end
 
-        def retryable_notification_ids
-          retryable_ns = Rpush::Client::Redis::Notification.absolute_retryable_namespace
+        # Retryable's first-pass cap. `.min` with `limit` matters at limit == 1, which the
+        # Feeder reaches whenever AppRunner still holds batch_size - 1 queued: without it
+        # the ceil would ask for more than the poll has.
+        #
+        # Nothing is reserved for pending here, because nothing needs to be: pending
+        # claims `limit - retryable_ids.size`, so it always receives at least
+        # limit - ceil(limit * SHARE), and the whole budget whenever nothing is due.
+        def retryable_budget(limit)
+          [(limit * RETRYABLE_CLAIM_SHARE).ceil, limit].min
+        end
 
-          Modis.with_connection do |redis|
-            retryable_results = redis.multi do |transaction|
-              now = Time.now.to_i
-              transaction.zrangebyscore(retryable_ns, 0, now)
-              transaction.zremrangebyscore(retryable_ns, 0, now)
-            end
+        # Claims at most `limit` DUE retries, oldest-due first, removing exactly the
+        # members it returns. Selection and removal are one atomic operation -- see
+        # RETRYABLE_CLAIM_SCRIPT for why that cannot be done with MULTI.
+        def retryable_notification_ids(limit)
+          return [] unless limit > 0
 
-            retryable_results.first
+          Modis.with_connection do |redis|
+            redis.eval(
+              RETRYABLE_CLAIM_SCRIPT,
+              keys: [Rpush::Client::Redis::Notification.absolute_retryable_namespace],
+              argv: [Time.now.to_i, limit]
+            )
           end
         end
 
+        # NOTE the early return. ZRANGE bounds are INCLUSIVE, so the `limit - 1` below
+        # turns a request for 0 into `zrange 0 0`, which returns ONE id -- claiming a
+        # notification the caller had no budget for and, worse, removing it from the
+        # pending set to do so. The previous caller papered over that with its own
+        # `limit > 0 ?` ternary; guarding here fixes it for every caller instead.
         def pending_notification_ids(limit)
+          return [] unless limit > 0
+
           limit = [0, limit - 1].max # 'zrange key 0 1' will return 2 values, not 1.
           pending_ns = Rpush::Client::Redis::Notification.absolute_pending_namespace
 
diff --git a/rpush.gemspec b/rpush.gemspec
index cdd42448a..03529128f 100644
--- a/rpush.gemspec
+++ b/rpush.gemspec
@@ -61,5 +61,9 @@ Gem::Specification.new do |s|
 
   s.add_development_dependency 'pg'
   s.add_development_dependency 'mysql2'
-  s.add_development_dependency 'sqlite3'
+  # Pin to the 1.6.x line: it ships precompiled x86_64-linux binaries for Ruby 2.7
+  # (production's version), so CI installs a prebuilt gem instead of compiling from
+  # source. 1.7+ dropped the precompiled 2.7 build. Only used by the active_record
+  # test client; production rpush uses the redis store.
+  s.add_development_dependency 'sqlite3', '~> 1.6.0'
 end
diff --git a/spec/functional_spec_helper.rb b/spec/functional_spec_helper.rb
index 46da6f137..85138e055 100644
--- a/spec/functional_spec_helper.rb
+++ b/spec/functional_spec_helper.rb
@@ -27,6 +27,8 @@ def stub_tcp_connection(tcp_socket, ssl_socket, io_double)
   end
 
   config.after(:each) do
-    DatabaseCleaner.clean if active_record? && functional_example?(self.class.metadata)
+    # Apps live in ActiveRecord under both clients, so clean the RDBMS after every
+    # functional example regardless of the notification store.
+    DatabaseCleaner.clean if functional_example?(self.class.metadata)
   end
 end
diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb
index 8911d55bf..e9ff5119a 100644
--- a/spec/spec_helper.rb
+++ b/spec/spec_helper.rb
@@ -32,7 +32,10 @@ def redis?
   client == :redis
 end
 
-require 'support/active_record_setup' if active_record?
+# The ActiveRecord schema is needed under both clients: Store::Redis#all_apps reads apps from
+# ActiveRecord (this fork's hybrid — apps in the RDBMS, notifications in the per-client store),
+# so even the redis client needs the apps table.
+require 'support/active_record_setup'
 
 RPUSH_ROOT = '/tmp/rails_root'
 
@@ -43,6 +46,10 @@ def redis?
 
 RPUSH_CLIENT = Rpush.config.client
 
+# Under the redis client, keep the ActiveRecord app store in sync with the redis one, so the
+# daemon (which reads apps from ActiveRecord) sees the apps the specs create in Redis.
+require 'support/redis_app_mirror' if redis?
+
 path = File.join(File.dirname(__FILE__), 'support')
 TEST_CERT = File.read(File.join(path, 'cert_without_password.pem'))
 TEST_CERT_WITH_PASSWORD = File.read(File.join(path, 'cert_with_password.pem'))
diff --git a/spec/support/redis_app_mirror.rb b/spec/support/redis_app_mirror.rb
new file mode 100644
index 000000000..5eadd924d
--- /dev/null
+++ b/spec/support/redis_app_mirror.rb
@@ -0,0 +1,36 @@
+# Production runs a hybrid store: apps live in Redis — where Notification#app resolves them and
+# the notification store is keyed — AND in ActiveRecord, where the daemon reads them
+# (Store::Redis#all_apps / #app use Rpush::Client::ActiveRecord::App; see the "Use ActiveRecord
+# to fetch Apps instead of Redis" change). The upstream specs only create Redis apps, so under
+# the redis client the daemon can't find them. Mirror every Redis app into ActiveRecord under
+# the same id so both sides agree, exactly as production keeps them in sync. Test-only.
+module RedisAppMirror
+  module_function
+
+  def active_record_class_for(redis_app)
+    redis_app.class.name.sub('Rpush::Client::Redis', 'Rpush::Client::ActiveRecord').constantize
+  end
+
+  def mirror(redis_app)
+    ar_class = active_record_class_for(redis_app)
+    record = ar_class.find_or_initialize_by(id: redis_app.id)
+    copyable = redis_app.attributes.slice(*ar_class.column_names).except('id', 'type')
+    record.assign_attributes(copyable)
+    record.id = redis_app.id
+    record.save!(validate: false)
+  end
+
+  def remove(id)
+    Rpush::Client::ActiveRecord::App.where(id: id).delete_all
+  end
+end
+
+Rpush::Client::Redis::App.class_eval do
+  after_save { RedisAppMirror.mirror(self) }
+  after_destroy { RedisAppMirror.remove(id) }
+end
+
+# The redis Feedback model disables Modis' all-index in production to avoid a huge "all" set.
+# Re-enable it for the test suite so specs can list feedback via .all; the data volumes that
+# motivate disabling it in production do not exist in tests.
+Rpush::Client::Redis::Apns::Feedback.enable_all_index(true)
diff --git a/spec/unit/client/redis/notification_spec.rb b/spec/unit/client/redis/notification_spec.rb
index 5d02d828b..4287450ac 100644
--- a/spec/unit/client/redis/notification_spec.rb
+++ b/spec/unit/client/redis/notification_spec.rb
@@ -2,4 +2,94 @@
 
 describe Rpush::Client::Redis::Notification do
   it_behaves_like 'Rpush::Client::Notification'
+
+  let(:app) do
+    Rpush::Client::Redis::Apns::App.create!(name: 'my_app', environment: 'development', certificate: TEST_CERT)
+  end
+
+  def create_notification(attrs = {})
+    Rpush::Client::Redis::Apns::Notification.create!({ device_token: 'a' * 64, app: app }.merge(attrs))
+  end
+
+  def pending_score_for(notification)
+    Modis.with_connection do |redis|
+      redis.zscore(described_class.absolute_pending_namespace, notification.id)
+    end
+  end
+
+  def pending_ids_in_rank_order
+    Modis.with_connection do |redis|
+      redis.zrange(described_class.absolute_pending_namespace, 0, -1).map(&:to_i)
+    end
+  end
+
+  describe 'pending-set score' do
+    it 'scores a default notification as exactly its own id' do
+      notification = create_notification
+      expect(pending_score_for(notification)).to eq notification.id.to_f
+    end
+
+    it 'scores a priority-class-0 notification one band below its id' do
+      notification = create_notification(priority_class: 0)
+      expected = notification.id - described_class::PRIORITY_CLASS_BAND
+      expect(pending_score_for(notification)).to eq expected.to_f
+    end
+
+    it 'ranks a priority-class-0 notification ahead of a default one created earlier' do
+      earlier_default = create_notification
+      later_priority  = create_notification(priority_class: 0)
+
+      ranked = pending_ids_in_rank_order
+      expect(ranked.index(later_priority.id)).to be < ranked.index(earlier_default.id)
+    end
+
+    it 'treats an explicitly nil priority_class as the default rather than promoting it' do
+      notification = create_notification(priority_class: nil)
+      expect(pending_score_for(notification)).to eq notification.id.to_f
+    end
+
+    it 'keeps the band inside the exact-integer range of a double' do
+      expect(described_class::PRIORITY_CLASS_BAND).to be < 2**53
+    end
+  end
+
+  describe 'priority_class' do
+    it 'defaults to PRIORITY_CLASS_DEFAULT' do
+      expect(create_notification.priority_class).to eq described_class::PRIORITY_CLASS_DEFAULT
+    end
+  end
+
+  describe 'enqueued_at' do
+    it 'is stamped on create' do
+      expect(create_notification.enqueued_at).not_to be_nil
+    end
+
+    it 'differs between notifications created at different times' do
+      first = nil
+      second = nil
+      Timecop.freeze(Time.now) { first = create_notification }
+      Timecop.freeze(Time.now + 5.minutes) { second = create_notification }
+
+      expect(second.enqueued_at.to_i).to be > first.enqueued_at.to_i
+    end
+
+    it 'is not overwritten when the caller supplies one' do
+      supplied = Time.now - 1.hour
+      expect(create_notification(enqueued_at: supplied).enqueued_at.to_i).to eq supplied.to_i
+    end
+  end
+
+  describe '#age_ms' do
+    it 'is nil when enqueued_at is absent' do
+      notification = create_notification
+      notification.enqueued_at = nil
+      expect(notification.age_ms).to be_nil
+    end
+
+    it 'measures whole seconds from enqueued_at' do
+      notification = create_notification
+      notification.enqueued_at = Time.now - 3
+      expect(notification.age_ms(Time.now)).to be_within(999).of(3000)
+    end
+  end
 end if redis?
diff --git a/spec/unit/daemon/shared/store.rb b/spec/unit/daemon/shared/store.rb
index c3d67f97d..46f80e034 100644
--- a/spec/unit/daemon/shared/store.rb
+++ b/spec/unit/daemon/shared/store.rb
@@ -31,12 +31,14 @@
   end
 
   it 'finds an app by ID' do
-    expect(store.app(app.id)).to eq(app)
+    # Compare by id: the redis store resolves apps from ActiveRecord (this fork's hybrid), so it
+    # returns an ActiveRecord::App representation of the same app the spec created in Redis.
+    expect(store.app(app.id).id).to eq(app.id)
   end
 
   it 'finds all apps' do
     app
-    expect(store.all_apps).to eq([app])
+    expect(store.all_apps.map(&:id)).to eq([app.id])
   end
 
   it 'translates an Integer notification ID' do
diff --git a/spec/unit/daemon/store/redis_spec.rb b/spec/unit/daemon/store/redis_spec.rb
index e484a4fce..ed96397cf 100644
--- a/spec/unit/daemon/store/redis_spec.rb
+++ b/spec/unit/daemon/store/redis_spec.rb
@@ -89,4 +89,135 @@
       end.not_to raise_exception
     end
   end
+
+  describe 'retryable and pending budget' do
+    let(:pending_ns)   { Rpush::Client::Redis::Notification.absolute_pending_namespace }
+    let(:retryable_ns) { Rpush::Client::Redis::Notification.absolute_retryable_namespace }
+
+    def new_notification
+      Rpush::Client::Redis::Apns::Notification.create!(device_token: 'a' * 64, app: app)
+    end
+
+    # Creation registers into pending, so a retryable fixture has to be moved across.
+    def move_to_retryable(notification, due_at)
+      Modis.with_connection do |redis|
+        redis.zrem(pending_ns, notification.id)
+        redis.zadd(retryable_ns, due_at.to_i, notification.id)
+      end
+    end
+
+    # The discriminating example: 11 due retries against a budget of 10. Under the
+    # previous unbounded claim, retryable_notification_ids returned all 11, `limit`
+    # went negative and pending received ZERO -- so the pending notification was
+    # absent and the batch overshot the budget. Both assertions fail on that code.
+    it 'does not let a deep retryable set starve pending' do
+      11.times { move_to_retryable(new_notification, time - 1.hour) }
+      pending = new_notification
+
+      delivered = store.deliverable_notifications(10)
+
+      expect(delivered.size).to be <= 10
+      expect(delivered).to include(pending)
+    end
+
+    # Retryable's cap is a first-pass cap only: a second pass hands back whatever pending
+    # left, so a retry backlog still drains at the full batch_size when pending is empty.
+    it 'gives retryable the whole budget when pending is empty' do
+      11.times { move_to_retryable(new_notification, time - 1.hour) }
+      Modis.with_connection { |redis| redis.del(pending_ns) }
+
+      expect(store.deliverable_notifications(10).size).to eq 10
+    end
+
+    it 'does not claim a retry whose deliver_after is still in the future' do
+      move_to_retryable(new_notification, time + 1.hour)
+      Modis.with_connection { |redis| redis.del(pending_ns) }
+
+      expect(store.deliverable_notifications(10)).to be_empty
+    end
+
+    it 'leaves unclaimed due retries in the retryable set' do
+      11.times { move_to_retryable(new_notification, time - 1.hour) }
+      Modis.with_connection { |redis| redis.del(pending_ns) }
+
+      store.deliverable_notifications(10)
+
+      remaining = Modis.with_connection { |redis| redis.zcard(retryable_ns) }
+      expect(remaining).to eq 1
+    end
+
+    # THE example for the campaign target: a drain with nothing due must get the WHOLE
+    # batch, not batch minus retryable's share. Reserving that share instead of capping it
+    # idles a fifth of the feeder's claim rate on every poll of a campaign -- and the
+    # claim rate is what sets how long the campaign takes.
+    it 'gives pending the whole budget when nothing is due to retry' do
+      10.times { new_notification }
+
+      expect(store.deliverable_notifications(10).size).to eq 10
+    end
+
+    # limit == 1 is what the Feeder passes whenever AppRunner still holds batch_size - 1
+    # queued.
+    it 'delivers pending work when the whole budget is a single slot and nothing is due' do
+      pending = new_notification
+
+      expect(store.deliverable_notifications(1)).to eq [pending]
+    end
+
+    # ZRANGE bounds are INCLUSIVE, so a zero pending budget used to claim one id anyway
+    # (`zrange 0 0` returns one element). Both halves are asserted, because the harmful
+    # half is the second: the notification was REMOVED from the pending set to be
+    # returned over budget, so a caller that dropped the overflow would drop the
+    # notification with it.
+    it 'gives the single slot to the due retry when there is one' do
+      retryable = new_notification
+      move_to_retryable(retryable, time - 1.hour)
+      new_notification
+
+      expect(store.deliverable_notifications(1)).to eq [retryable]
+    end
+
+    it 'leaves the pending notification in the set when its budget was zero' do
+      retryable = new_notification
+      move_to_retryable(retryable, time - 1.hour)
+      new_notification
+
+      store.deliverable_notifications(1)
+
+      remaining = Modis.with_connection { |redis| redis.zcard(pending_ns) }
+      expect(remaining).to eq 1
+    end
+
+    it 'delivers nothing rather than raising when the budget is zero' do
+      new_notification
+
+      expect(store.deliverable_notifications(0)).to be_empty
+    end
+
+    # The claim selects and removes ONE member set in a single Redis operation, so a
+    # concurrent claimer cannot make it remove a member whose backoff has not elapsed.
+    # Under a rank-bounded claim the future retry would be taken here as soon as the
+    # due count read before the removal disagreed with the set.
+    it 'leaves a future retry queued when it shares the set with a due one' do
+      due = new_notification
+      future = new_notification
+      move_to_retryable(due, time - 1.hour)
+      move_to_retryable(future, time + 1.hour)
+      Modis.with_connection { |redis| redis.del(pending_ns) }
+
+      expect(store.deliverable_notifications(10)).to eq [due]
+    end
+
+    it 'keeps the future retry in the retryable set for its own backoff' do
+      due = new_notification
+      future = new_notification
+      move_to_retryable(due, time - 1.hour)
+      move_to_retryable(future, time + 1.hour)
+
+      store.deliverable_notifications(10)
+
+      remaining = Modis.with_connection { |redis| redis.zrange(retryable_ns, 0, -1) }
+      expect(remaining).to eq [future.id.to_s]
+    end
+  end
 end if redis?
diff --git a/spec/unit_spec_helper.rb b/spec/unit_spec_helper.rb
index 9a9c453dd..bea195536 100644
--- a/spec/unit_spec_helper.rb
+++ b/spec/unit_spec_helper.rb
@@ -14,14 +14,16 @@ def unit_example?(metadata)
       redis.keys('rpush:*').each { |key| redis.del(key) }
     end if redis? && unit_example?(self.class.metadata)
 
-    if active_record? && unit_example?(self.class.metadata)
+    # Apps live in ActiveRecord under both clients, so wrap every unit example in a
+    # transaction and roll it back, regardless of the notification store.
+    if unit_example?(self.class.metadata)
       connection = ActiveRecord::Base.connection
       connection.begin_transaction joinable: false
     end
   end
 
   config.after(:each) do
-    if active_record? && unit_example?(self.class.metadata)
+    if unit_example?(self.class.metadata)
       connection = ActiveRecord::Base.connection
       connection.rollback_transaction if connection.transaction_open?
     end