PLA-1859 Instrument feeder and add sandbox-only simulated delivery

README.md

@@ -8,6 +8,17 @@
 
 ### Rpush. The push notification service for Ruby.
 
+#### Thanx sandbox feeder probe
+
+On the Thanx fork, a sandbox notification whose data contains `pla1859_run` and
+whose token starts with the matching `pla1859-<run>-` prefix is always simulated
+before provider dispatch. Ordinary sandbox notifications continue to send.
+`RPUSH_SIMULATE_DELIVERY=true` remains an optional sandbox-only switch that
+simulates all notifications, and must not be left on after a test. Simulated
+notifications reach a terminal Redis state and emit `notifications_simulated`,
+not `notification_delivered`. This is a feeder-capacity test, not an end-to-end
+APNs/FCM throughput test.
+
 Rpush aims to be the *de facto* gem for sending push notifications in Ruby. Its core goals are ease of use, reliability and a rich feature set. Rpush provides numerous advanced features not found in others gems, giving you greater control & insight as your project grows. These are a few of the reasons why companies worldwide rely on Rpush to deliver their notifications.
 
 #### Supported Services

lib/rpush/configuration.rb

@@ -16,7 +16,7 @@ def configure
     end
   end
 
-  CURRENT_ATTRS = [:push_poll, :embedded, :pid_file, :batch_size, :push, :client, :logger, :log_file, :foreground, :foreground_logging, :log_level, :plugin, :apns]
+  CURRENT_ATTRS = [:push_poll, :embedded, :pid_file, :batch_size, :redis_fetch_batch_size, :push, :client, :logger, :log_file, :foreground, :foreground_logging, :log_level, :plugin, :apns]
   DEPRECATED_ATTRS = []
   CONFIG_ATTRS = CURRENT_ATTRS + DEPRECATED_ATTRS
 
@@ -47,6 +47,7 @@ def initialize
 
       self.push_poll = 2
       self.batch_size = 100
+      self.redis_fetch_batch_size = 1
       self.logger = nil
       self.log_file = 'log/rpush.log'
       self.pid_file = 'tmp/rpush.pid'

lib/rpush/daemon/batch.rb

@@ -4,12 +4,13 @@ class Batch
       include Reflectable
       include Loggable
 
-      attr_reader :num_processed, :notifications, :delivered, :failed, :retryable
+      attr_reader :num_processed, :notifications, :delivered, :failed, :retryable, :simulated
 
       def initialize(notifications)
         @notifications = notifications
         @num_processed = 0
         @delivered = []
+        @simulated = []
         @failed = {}
         @retryable = {}
         @mutex = Mutex.new
@@ -34,7 +35,7 @@ def each_delivered(&blk)
       # re-queue such notifications rather than lose them. Read-only; safe on any transport.
       def unresolved
         @mutex.synchronize do
-          resolved = @delivered + @failed.values.flatten(1) + @retryable.values.flatten(1)
+          resolved = @delivered + @simulated + @failed.values.flatten(1) + @retryable.values.flatten(1)
           @notifications - resolved
         end
       end
@@ -86,6 +87,21 @@ def mark_all_delivered
         end
       end
 
+      # A sandbox load-test outcome, not provider acceptance. Persist using the
+      # existing terminal Redis state so the feeder does not reclaim these rows,
+      # but never emit notification_delivered or its Nexus-side side effects.
+      def mark_simulated(notification)
+        @mutex.synchronize { @simulated << notification }
+        Rpush::Daemon.store.mark_delivered(notification, Time.now, persist: false)
+      end
+
+      def mark_all_simulated
+        @mutex.synchronize { @simulated = @notifications }
+        each_notification do |notification|
+          Rpush::Daemon.store.mark_delivered(notification, Time.now, persist: false)
+        end
+      end
+
       def mark_failed(notification, code, description)
         key = [code, description]
         @mutex.synchronize do
@@ -124,7 +140,7 @@ def all_processed
       def complete
         return if complete?
 
-        [:complete_delivered, :complete_failed, :complete_retried].each do |method|
+        [:complete_delivered, :complete_simulated, :complete_failed, :complete_retried].each do |method|
           begin
             send(method)
           rescue StandardError => e
@@ -143,6 +159,13 @@ def complete_delivered
         end
       end
 
+      def complete_simulated
+        return if @simulated.empty?
+
+        Rpush::Daemon.store.mark_batch_delivered(@simulated)
+        reflect(:notifications_simulated, @simulated.size)
+      end
+
       def complete_failed
         @failed.each do |(code, description), notifications|
           Rpush::Daemon.store.mark_batch_failed(notifications, code, description)

lib/rpush/daemon/dispatcher_loop.rb

@@ -63,11 +63,43 @@ def should_stop?(payload)
 
       def dispatch(payload)
         @dispatch_count += 1
-        @dispatcher.dispatch(payload)
+        if simulate_delivery?(payload)
+          # Keep the feeder, app queue, and Redis persistence in the benchmark, but
+          # never enter a provider dispatcher. A batch payload has no notification;
+          # single-delivery services share a batch across their queue payloads.
+          if payload.notification
+            payload.batch.mark_simulated(payload.notification)
+            payload.batch.notification_processed
+          else
+            payload.batch.mark_all_simulated
+            payload.batch.all_processed
+          end
+        else
+          @dispatcher.dispatch(payload)
+        end
       rescue StandardError => e
         log_error(e)
         reflect(:error, e)
       end
+
+      def simulate_delivery?(payload)
+        return false unless ENV['RAILS_ENV'] == 'sandbox'
+        return true if ENV['RPUSH_SIMULATE_DELIVERY'] == 'true'
+
+        # Probe records must never reach a provider even when the global test
+        # flag is off or a normal sandbox daemon claims one. Ordinary sandbox
+        # notifications keep their normal delivery path.
+        notifications = payload.notification ? [payload.notification] : payload.batch.notifications
+        marked = notifications.map { |notification| synthetic_probe_notification?(notification) }
+        raise 'mixed synthetic and real notifications in one provider batch' if marked.any? && !marked.all?
+
+        marked.any? && marked.all?
+      end
+
+      def synthetic_probe_notification?(notification)
+        marker = notification.data.is_a?(Hash) && notification.data['pla1859_run']
+        marker && notification.device_token.to_s.start_with?("pla1859-#{marker}-")
+      end
     end
   end
 end

lib/rpush/daemon/feeder.rb

@@ -52,10 +52,25 @@ def self.feed_forever
       end
 
       def self.enqueue_notifications
-        batch_size = Rpush.config.batch_size - Rpush::Daemon::AppRunner.total_queued
-        return if batch_size <= 0
+        poll_started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        queued_before = Rpush::Daemon::AppRunner.total_queued
+        queue_ms = elapsed_ms(poll_started)
+        batch_size = Rpush.config.batch_size - queued_before
+        if batch_size <= 0
+          reflect(:feeder_poll, requested: 0, claimed: 0, queued_before: queued_before,
+            queue_ms: queue_ms, claim_ms: 0, enqueue_ms: 0, work_ms: elapsed_ms(poll_started))
+          return
+        end
+
+        claim_started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
         notifications = Rpush::Daemon.store.deliverable_notifications(batch_size)
+        claim_ms = elapsed_ms(claim_started)
+
+        enqueue_started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
         Rpush::Daemon::AppRunner.enqueue(notifications)
+        reflect(:feeder_poll, requested: batch_size, claimed: notifications.size, queued_before: queued_before,
+          queue_ms: queue_ms, claim_ms: claim_ms, enqueue_ms: elapsed_ms(enqueue_started),
+          work_ms: elapsed_ms(poll_started))
       rescue StandardError => e
         Rpush.logger.error(e)
         reflect(:error, e)
@@ -64,6 +79,10 @@ def self.enqueue_notifications
       def self.interruptible_sleeper
         @interruptible_sleeper ||= InterruptibleSleep.new
       end
+
+      def self.elapsed_ms(started_at)
+        ((Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at) * 1000).round(2)
+      end
     end
   end
 end

lib/rpush/daemon/store/redis.rb

@@ -2,6 +2,8 @@ module Rpush
   module Daemon
     module Store
       class Redis
+        include Reflectable
+
         DEFAULT_MARK_OPTIONS = { persist: true }
 
         # Ceiling on the share of a poll the retryable set may take in its first pass, so
@@ -67,14 +69,22 @@ def all_apps
         def deliverable_notifications(limit)
           return [] unless limit > 0
 
+          claim_started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
           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
+          claim_ms = elapsed_ms(claim_started)
+          fetch_started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+          notifications = find_notifications_by_ids(ids)
+          reflect(:redis_claim, requested: limit, pending_claimed: pending_ids.size,
+            retryable_claimed: retryable_ids.size, missing: ids.size - notifications.size,
+            claim_ms: claim_ms, fetch_ms: elapsed_ms(fetch_started),
+            fetch_batch_size: Rpush.config.redis_fetch_batch_size)
+          notifications
         end
 
         def mark_delivered(notification, time, opts = {})
@@ -106,10 +116,7 @@ def mark_batch_failed(notifications, code, description)
         end
 
         def mark_ids_failed(ids, code, description, time)
-          ids.each do |id|
-            notification = find_notification_by_id(id)
-            next unless notification
-
+          find_notifications_by_ids(ids).each do |notification|
             mark_failed(notification, code, description, time)
           end
         end
@@ -137,10 +144,7 @@ def mark_batch_retryable(notifications, deliver_after)
         end
 
         def mark_ids_retryable(ids, deliver_after)
-          ids.each do |id|
-            notification = find_notification_by_id(id)
-            next unless notification
-
+          find_notifications_by_ids(ids).each do |notification|
             mark_retryable(notification, deliver_after)
           end
         end
@@ -193,13 +197,35 @@ def translate_integer_notification_id(id)
 
         private
 
+        def elapsed_ms(started_at)
+          ((Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at) * 1000).round(2)
+        end
+
         def find_notification_by_id(id)
           Rpush::Client::Redis::Notification.find(id)
         rescue Modis::RecordNotFound
           Rpush.logger.warn("Couldn't find Rpush::Client::Redis::Notification with id=#{id}")
           nil
         end
 
+        # Modis#find pipelines a multi-ID lookup, preserving the caller's order and
+        # STI model types. Keep the default single-record path until a deployment
+        # explicitly opts in. A missing hash makes Modis raise for the whole slice;
+        # retry that small slice individually to retain the old skip-and-warn behavior.
+        def find_notifications_by_ids(ids)
+          ids.each_slice(Rpush.config.redis_fetch_batch_size).flat_map do |slice|
+            if slice.size == 1
+              [find_notification_by_id(slice.first)].compact
+            else
+              begin
+                Rpush::Client::Redis::Notification.find(*slice)
+              rescue Modis::RecordNotFound
+                slice.map { |id| find_notification_by_id(id) }.compact
+              end
+            end
+          end
+        end
+
         def create_fcm_like_notification(notification, attrs, data, app) # rubocop:disable Metrics/ParameterLists
           notification.assign_attributes(attrs)
           notification.data = data

lib/rpush/reflection_collection.rb

@@ -9,7 +9,8 @@ class NoSuchReflectionError < StandardError; end
       :fcm_delivered_to_recipient, :fcm_failed_to_recipient, :fcm_canonical_id, :fcm_invalid_device_token,
       :error, :adm_canonical_id, :adm_failed_to_recipient, :wns_invalid_channel,
       :tcp_connection_lost, :ssl_certificate_will_expire, :ssl_certificate_revoked,
-      :notification_id_will_retry, :notification_id_failed
+      :notification_id_will_retry, :notification_id_failed,
+      :feeder_poll, :redis_claim, :notifications_simulated
     ]
 
     DEPRECATIONS = {}

rpush.gemspec

@@ -55,6 +55,8 @@ Gem::Specification.new do |s|
   s.add_development_dependency 'appraisal'
   s.add_development_dependency 'codeclimate-test-reporter', '1.0.7'
   s.add_development_dependency 'simplecov'
+  # Rails 6/7.0 JSON encoding passes quirks_mode, which json 3 no longer accepts.
+  s.add_development_dependency 'json', '< 3'
   s.add_development_dependency 'rubocop', '~> 1.12.0'
   s.add_development_dependency 'rubocop-performance'
   s.add_development_dependency 'byebug'

spec/spec_helper.rb

@@ -54,7 +54,7 @@ def redis?
 TEST_CERT = File.read(File.join(path, 'cert_without_password.pem'))
 TEST_CERT_WITH_PASSWORD = File.read(File.join(path, 'cert_with_password.pem'))
 
-VAPID_KEYPAIR = WebPush.generate_key.to_hash.merge(subject: 'rpush-test@example.org').to_json
+VAPID_KEYPAIR = JSON.generate(WebPush.generate_key.to_hash.merge(subject: 'rpush-test@example.org'))
 
 def after_example_cleanup
   Rpush.logger = nil

spec/unit/configuration_spec.rb

@@ -28,6 +28,17 @@
     end
   end
 
+  it 'keeps Redis notification reads serial by default' do
+    expect(config.redis_fetch_batch_size).to eq(1)
+  end
+
+  it 'copies the Redis fetch batch size on update' do
+    new_config = Rpush::Configuration.new
+    new_config.redis_fetch_batch_size = 25
+    config.update(new_config)
+    expect(config.redis_fetch_batch_size).to eq(25)
+  end
+
   it 'sets the pid_file relative if not absolute' do
     config.pid_file = 'tmp/rpush.pid'
     expect(config.pid_file).to eq '/tmp/rails_root/tmp/rpush.pid'

spec/unit/daemon/batch_spec.rb

@@ -50,6 +50,29 @@
     end
   end
 
+  describe 'simulated delivery' do
+    it 'persists a terminal state without classifying it as provider delivery' do
+      expect(store).to receive(:mark_delivered).with(notification1, time, persist: false)
+      batch.mark_simulated(notification1)
+      expect(batch.simulated).to eq [notification1]
+      expect(batch.delivered).to be_empty
+    end
+
+    it 'resolves every notification in a batch' do
+      batch.mark_all_simulated
+      expect(batch.unresolved).to be_empty
+    end
+
+    it 'emits only the simulated reflection on completion' do
+      allow(batch).to receive(:reflect)
+      expect(store).to receive(:mark_batch_delivered).with([notification1, notification2])
+      expect(batch).to receive(:reflect).with(:notifications_simulated, 2)
+      expect(batch).not_to receive(:reflect).with(:notification_delivered, anything)
+      batch.mark_all_simulated
+      batch.all_processed
+    end
+  end
+
   describe 'mark_all_retryable' do
     let(:error) { StandardError.new('Exception') }