Priority band for the pending set, enqueued_at instrumentation, and a retryable-starvation fix

.github/workflows/test.yml

@@ -22,7 +22,7 @@ env:
 
 jobs:
   test:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
 
     services:
       postgres:

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
 

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

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
 

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

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

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'))

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)