From d80d33d8f192691905c3e14435a124751edecae0 Mon Sep 17 00:00:00 2001 From: Cosmin Atanasiu Date: Tue, 20 Oct 2026 12:06:47 -0700 Subject: [PATCH 01/10] PLA-1859 Instrument Rpush feeder claims for sandbox POC --- lib/rpush/daemon/feeder.rb | 23 +++++++++++++++++++++-- lib/rpush/daemon/store/redis.rb | 15 ++++++++++++++- lib/rpush/reflection_collection.rb | 3 ++- spec/unit/daemon/feeder_spec.rb | 13 +++++++++++++ spec/unit/daemon/store/redis_spec.rb | 8 ++++++++ 5 files changed, 58 insertions(+), 4 deletions(-) diff --git a/lib/rpush/daemon/feeder.rb b/lib/rpush/daemon/feeder.rb index 9bf447ad5..793f7a1ae 100644 --- a/lib/rpush/daemon/feeder.rb +++ b/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 diff --git a/lib/rpush/daemon/store/redis.rb b/lib/rpush/daemon/store/redis.rb index d88693e8b..35c708779 100644 --- a/lib/rpush/daemon/store/redis.rb +++ b/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,6 +69,7 @@ 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) @@ -74,7 +77,13 @@ def deliverable_notifications(limit) 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 = ids.map { |id| find_notification_by_id(id) }.compact + 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)) + notifications end def mark_delivered(notification, time, opts = {}) @@ -193,6 +202,10 @@ 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 diff --git a/lib/rpush/reflection_collection.rb b/lib/rpush/reflection_collection.rb index f2574870c..c72e74466 100644 --- a/lib/rpush/reflection_collection.rb +++ b/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 ] DEPRECATIONS = {} diff --git a/spec/unit/daemon/feeder_spec.rb b/spec/unit/daemon/feeder_spec.rb index d1ee32618..47f49a7ef 100644 --- a/spec/unit/daemon/feeder_spec.rb +++ b/spec/unit/daemon/feeder_spec.rb @@ -54,6 +54,19 @@ def start_and_stop start_and_stop end + it 'reports the requested and loaded notification counts per poll' do + expect(Rpush::Daemon::Feeder).to receive(:reflect).with(:feeder_poll, + hash_including(requested: Rpush.config.batch_size, claimed: 1, queued_before: 0)) + start_and_stop + end + + it 'reports when the internal queue consumes the whole claim budget' do + allow(Rpush::Daemon::AppRunner).to receive_messages(total_queued: Rpush.config.batch_size) + expect(Rpush::Daemon::Feeder).to receive(:reflect).with(:feeder_poll, + hash_including(requested: 0, claimed: 0, queued_before: Rpush.config.batch_size)) + start_and_stop + end + it "logs errors" do e = StandardError.new("bork") allow(Rpush::Daemon.store).to receive(:deliverable_notifications).and_raise(e) diff --git a/spec/unit/daemon/store/redis_spec.rb b/spec/unit/daemon/store/redis_spec.rb index ed96397cf..2ff8b1118 100644 --- a/spec/unit/daemon/store/redis_spec.rb +++ b/spec/unit/daemon/store/redis_spec.rb @@ -15,6 +15,14 @@ end describe 'deliverable_notifications' do + it 'reports the pending and retry claim counts without notification IDs' do + notification + expect(store).to receive(:reflect).with(:redis_claim, + hash_including(requested: Rpush.config.batch_size, pending_claimed: 1, + retryable_claimed: 0, missing: 0)) + store.deliverable_notifications(Rpush.config.batch_size) + end + it 'loads notifications in batches' do Rpush.config.batch_size = 100 allow(store).to receive_messages(pending_notification_ids: [1, 2, 3, 4]) From c4cecae4523e1e82c8202b018d3ce5ca05824943 Mon Sep 17 00:00:00 2001 From: Cosmin Atanasiu Date: Tue, 20 Oct 2026 14:59:37 -0700 Subject: [PATCH 02/10] PLA-1859 Add sandbox-only simulated Rpush delivery --- README.md | 9 +++++++ lib/rpush/daemon/batch.rb | 29 ++++++++++++++++++--- lib/rpush/daemon/dispatcher_loop.rb | 19 +++++++++++++- lib/rpush/reflection_collection.rb | 2 +- spec/unit/daemon/batch_spec.rb | 23 +++++++++++++++++ spec/unit/daemon/dispatcher_loop_spec.rb | 32 ++++++++++++++++++++++++ 6 files changed, 109 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 2b1fde050..1749cf433 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,15 @@ ### Rpush. The push notification service for Ruby. +#### Thanx sandbox feeder probe + +On the Thanx fork, `RPUSH_SIMULATE_DELIVERY=true` suppresses all provider dispatch +only when `RAILS_ENV=sandbox`. The daemon still claims, queues, and persists +each notification in a terminal Redis state; it emits +`notifications_simulated` rather than `notification_delivered`. This is for a +short feeder-capacity test, not an end-to-end APNs/FCM throughput test. Keep the +flag on until synthetic backlog is drained before restoring normal delivery. + 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 diff --git a/lib/rpush/daemon/batch.rb b/lib/rpush/daemon/batch.rb index 7037b4a62..107cd6b0c 100644 --- a/lib/rpush/daemon/batch.rb +++ b/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) diff --git a/lib/rpush/daemon/dispatcher_loop.rb b/lib/rpush/daemon/dispatcher_loop.rb index 7db6fd0f6..c12cbcb68 100644 --- a/lib/rpush/daemon/dispatcher_loop.rb +++ b/lib/rpush/daemon/dispatcher_loop.rb @@ -63,11 +63,28 @@ def should_stop?(payload) def dispatch(payload) @dispatch_count += 1 - @dispatcher.dispatch(payload) + if simulate_delivery? + # 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? + ENV['RAILS_ENV'] == 'sandbox' && ENV['RPUSH_SIMULATE_DELIVERY'] == 'true' + end end end end diff --git a/lib/rpush/reflection_collection.rb b/lib/rpush/reflection_collection.rb index c72e74466..b19d23a80 100644 --- a/lib/rpush/reflection_collection.rb +++ b/lib/rpush/reflection_collection.rb @@ -10,7 +10,7 @@ class NoSuchReflectionError < StandardError; end :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, - :feeder_poll, :redis_claim + :feeder_poll, :redis_claim, :notifications_simulated ] DEPRECATIONS = {} diff --git a/spec/unit/daemon/batch_spec.rb b/spec/unit/daemon/batch_spec.rb index 4a8ffdc5c..ac06eec18 100644 --- a/spec/unit/daemon/batch_spec.rb +++ b/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') } diff --git a/spec/unit/daemon/dispatcher_loop_spec.rb b/spec/unit/daemon/dispatcher_loop_spec.rb index ac6456455..a6f9e28c0 100644 --- a/spec/unit/daemon/dispatcher_loop_spec.rb +++ b/spec/unit/daemon/dispatcher_loop_spec.rb @@ -35,6 +35,38 @@ def run_dispatcher_loop run_dispatcher_loop end + describe 'sandbox simulated delivery' do + before do + queue.clear + allow(ENV).to receive(:[]).and_call_original + allow(ENV).to receive(:[]).with('RAILS_ENV').and_return('sandbox') + allow(ENV).to receive(:[]).with('RPUSH_SIMULATE_DELIVERY').and_return('true') + end + + it 'resolves a single notification without calling the provider dispatcher' do + queue.push(Rpush::Daemon::QueuePayload.new(batch, notification)) + expect(batch).to receive(:mark_simulated).with(notification) + expect(batch).to receive(:notification_processed) + expect(dispatcher).not_to receive(:dispatch) + run_dispatcher_loop + end + + it 'resolves an APNs batch without calling the provider dispatcher' do + queue.push(Rpush::Daemon::QueuePayload.new(batch)) + expect(batch).to receive(:mark_all_simulated) + expect(batch).to receive(:all_processed) + expect(dispatcher).not_to receive(:dispatch) + run_dispatcher_loop + end + + it 'does not simulate outside sandbox' do + allow(ENV).to receive(:[]).with('RAILS_ENV').and_return('production') + queue.push(Rpush::Daemon::QueuePayload.new(batch, notification)) + expect(dispatcher).to receive(:dispatch) + run_dispatcher_loop + end + end + describe 'stop' do before do queue.clear From cd53458d76bbaeecbaa0f7652da93ab5b392031c Mon Sep 17 00:00:00 2001 From: Cosmin Atanasiu Date: Tue, 20 Oct 2026 15:04:33 -0700 Subject: [PATCH 03/10] PLA-1859 Keep Rpush specs compatible with JSON 3 --- spec/spec_helper.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb index e9ff5119a..974a948de 100644 --- a/spec/spec_helper.rb +++ b/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 From eafa49cfdb41a50184f673c090170cebc3d26b15 Mon Sep 17 00:00:00 2001 From: Cosmin Atanasiu Date: Tue, 20 Oct 2026 15:07:18 -0700 Subject: [PATCH 04/10] PLA-1859 Pin test JSON below incompatible major version --- rpush.gemspec | 2 ++ 1 file changed, 2 insertions(+) diff --git a/rpush.gemspec b/rpush.gemspec index 03529128f..2c0396e88 100644 --- a/rpush.gemspec +++ b/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' From 376e1043d43719c3da14edff2562faca8066310a Mon Sep 17 00:00:00 2001 From: Cosmin Atanasiu Date: Tue, 20 Oct 2026 15:36:43 -0700 Subject: [PATCH 05/10] PLA-1859 Make marked sandbox probe records intrinsically no-send --- README.md | 14 +++++---- lib/rpush/daemon/dispatcher_loop.rb | 21 ++++++++++++-- spec/unit/daemon/dispatcher_loop_spec.rb | 37 ++++++++++++++++++++++++ 3 files changed, 63 insertions(+), 9 deletions(-) diff --git a/README.md b/README.md index 1749cf433..840fbe776 100644 --- a/README.md +++ b/README.md @@ -10,12 +10,14 @@ #### Thanx sandbox feeder probe -On the Thanx fork, `RPUSH_SIMULATE_DELIVERY=true` suppresses all provider dispatch -only when `RAILS_ENV=sandbox`. The daemon still claims, queues, and persists -each notification in a terminal Redis state; it emits -`notifications_simulated` rather than `notification_delivered`. This is for a -short feeder-capacity test, not an end-to-end APNs/FCM throughput test. Keep the -flag on until synthetic backlog is drained before restoring normal delivery. +On the Thanx fork, a sandbox notification whose data contains `pla1859_run` and +whose token starts with the matching `pla1859--` 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. diff --git a/lib/rpush/daemon/dispatcher_loop.rb b/lib/rpush/daemon/dispatcher_loop.rb index c12cbcb68..78258585f 100644 --- a/lib/rpush/daemon/dispatcher_loop.rb +++ b/lib/rpush/daemon/dispatcher_loop.rb @@ -63,7 +63,7 @@ def should_stop?(payload) def dispatch(payload) @dispatch_count += 1 - if simulate_delivery? + 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. @@ -82,8 +82,23 @@ def dispatch(payload) reflect(:error, e) end - def simulate_delivery? - ENV['RAILS_ENV'] == 'sandbox' && ENV['RPUSH_SIMULATE_DELIVERY'] == 'true' + 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 diff --git a/spec/unit/daemon/dispatcher_loop_spec.rb b/spec/unit/daemon/dispatcher_loop_spec.rb index a6f9e28c0..f267f57f2 100644 --- a/spec/unit/daemon/dispatcher_loop_spec.rb +++ b/spec/unit/daemon/dispatcher_loop_spec.rb @@ -67,6 +67,43 @@ def run_dispatcher_loop end end + describe 'sandbox synthetic probe records' do + before do + queue.clear + allow(ENV).to receive(:[]).and_call_original + allow(ENV).to receive(:[]).with('RAILS_ENV').and_return('sandbox') + allow(ENV).to receive(:[]).with('RPUSH_SIMULATE_DELIVERY').and_return(nil) + allow(notification).to receive_messages(data: {}, device_token: 'real-device-token') + end + + it 'never dispatches a marked synthetic notification to FCM' do + allow(notification).to receive_messages( + data: { 'pla1859_run' => 'probe1' }, device_token: 'pla1859-probe1-0' + ) + queue.push(Rpush::Daemon::QueuePayload.new(batch, notification)) + expect(batch).to receive(:mark_simulated).with(notification) + expect(batch).to receive(:notification_processed) + expect(dispatcher).not_to receive(:dispatch) + run_dispatcher_loop + end + + it 'continues to dispatch ordinary sandbox notifications' do + queue.push(Rpush::Daemon::QueuePayload.new(batch, notification)) + expect(dispatcher).to receive(:dispatch) + run_dispatcher_loop + end + + it 'does not simulate marked notifications outside sandbox' do + allow(ENV).to receive(:[]).with('RAILS_ENV').and_return('production') + allow(notification).to receive_messages( + data: { 'pla1859_run' => 'probe1' }, device_token: 'pla1859-probe1-0' + ) + queue.push(Rpush::Daemon::QueuePayload.new(batch, notification)) + expect(dispatcher).to receive(:dispatch) + run_dispatcher_loop + end + end + describe 'stop' do before do queue.clear From 67c5f4a66796f17061e0556acd694a554f00a054 Mon Sep 17 00:00:00 2001 From: Cosmin Atanasiu Date: Wed, 21 Oct 2026 10:12:40 -0700 Subject: [PATCH 06/10] PLA-1859 Pipeline bounded Redis notification fetches --- lib/rpush/configuration.rb | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/lib/rpush/configuration.rb b/lib/rpush/configuration.rb index 77ddcad1e..5c59692d5 100644 --- a/lib/rpush/configuration.rb +++ b/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' From 98740ecfdddbb11514c02270e1e3c4e9d757d0c7 Mon Sep 17 00:00:00 2001 From: Cosmin Atanasiu Date: Wed, 21 Oct 2026 10:12:43 -0700 Subject: [PATCH 07/10] PLA-1859 Pipeline bounded Redis notification fetches --- lib/rpush/daemon/store/redis.rb | 33 +++++++++++++++++++++++---------- 1 file changed, 23 insertions(+), 10 deletions(-) diff --git a/lib/rpush/daemon/store/redis.rb b/lib/rpush/daemon/store/redis.rb index 35c708779..a01b26d23 100644 --- a/lib/rpush/daemon/store/redis.rb +++ b/lib/rpush/daemon/store/redis.rb @@ -79,10 +79,11 @@ def deliverable_notifications(limit) ids = retryable_ids + pending_ids claim_ms = elapsed_ms(claim_started) fetch_started = Process.clock_gettime(Process::CLOCK_MONOTONIC) - notifications = ids.map { |id| find_notification_by_id(id) }.compact + 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)) + claim_ms: claim_ms, fetch_ms: elapsed_ms(fetch_started), + fetch_batch_size: Rpush.config.redis_fetch_batch_size) notifications end @@ -115,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 @@ -146,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 @@ -213,6 +208,24 @@ def find_notification_by_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 From ba81bb01ede675953dd966bd43025226a4be159e Mon Sep 17 00:00:00 2001 From: Cosmin Atanasiu Date: Wed, 21 Oct 2026 10:12:46 -0700 Subject: [PATCH 08/10] PLA-1859 Pipeline bounded Redis notification fetches --- spec/unit/configuration_spec.rb | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/spec/unit/configuration_spec.rb b/spec/unit/configuration_spec.rb index ff1032858..1a97988f6 100644 --- a/spec/unit/configuration_spec.rb +++ b/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' From 51656bed32596b549a53f22479c59f667746410f Mon Sep 17 00:00:00 2001 From: Cosmin Atanasiu Date: Wed, 21 Oct 2026 10:12:49 -0700 Subject: [PATCH 09/10] PLA-1859 Pipeline bounded Redis notification fetches --- spec/unit/daemon/store/redis_spec.rb | 33 ++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/spec/unit/daemon/store/redis_spec.rb b/spec/unit/daemon/store/redis_spec.rb index 2ff8b1118..0c1e4eacb 100644 --- a/spec/unit/daemon/store/redis_spec.rb +++ b/spec/unit/daemon/store/redis_spec.rb @@ -30,6 +30,30 @@ store.deliverable_notifications(Rpush.config.batch_size) end + it 'pipelines bounded groups while preserving notification order and STI types' do + notifications = 5.times.map { |i| Rpush::Client::Redis::Apns::Notification.create!(device_token: "#{i}" * 64, app: app) } + Rpush.config.redis_fetch_batch_size = 2 + + expect(Rpush::Client::Redis::Notification).to receive(:find).with(notifications[0].id.to_s, notifications[1].id.to_s).ordered.and_call_original + expect(Rpush::Client::Redis::Notification).to receive(:find).with(notifications[2].id.to_s, notifications[3].id.to_s).ordered.and_call_original + expect(Rpush::Client::Redis::Notification).to receive(:find).with(notifications[4].id.to_s).ordered.and_call_original + + loaded = store.deliverable_notifications(5) + expect(loaded.map(&:id)).to eq(notifications.map(&:id)) + expect(loaded).to all(be_a(Rpush::Client::Redis::Apns::Notification)) + end + + it 'falls back to individual reads only for a group containing a missing hash' do + notifications = 3.times.map { Rpush::Client::Redis::Apns::Notification.create!(device_token: 'a' * 64, app: app) } + Rpush.config.redis_fetch_batch_size = 3 + Modis.with_connection { |redis| redis.del(Rpush::Client::Redis::Notification.key_for(notifications[1].id)) } + + expect(logger).to receive(:warn).with("Couldn't find Rpush::Client::Redis::Notification with id=#{notifications[1].id}") + expect(store).to receive(:reflect).with(:redis_claim, hash_including(missing: 1, fetch_batch_size: 3)) + + expect(store.deliverable_notifications(3).map(&:id)).to eq([notifications[0].id, notifications[2].id]) + end + it 'loads an undelivered notification without deliver_after set' do notification.update!(delivered: false, deliver_after: nil) expect(store.deliverable_notifications(Rpush.config.batch_size)).to eq [notification] @@ -79,6 +103,15 @@ store.mark_ids_retryable([notification.id], deliver_after) end.not_to raise_exception end + + it 'bulk-fetches several notifications before marking them retryable' do + other = Rpush::Client::Redis::Apns::Notification.create!(device_token: 'b' * 64, app: app) + Rpush.config.redis_fetch_batch_size = 2 + expect(Rpush::Client::Redis::Notification).to receive(:find).with(notification.id, other.id).and_call_original + + store.mark_ids_retryable([notification.id, other.id], deliver_after) + expect([notification.reload.deliver_after, other.reload.deliver_after].map(&:to_s)).to all(eq(deliver_after.utc.to_s)) + end end describe 'mark_ids_failed' do From e2a1bae469947769b6bae8def7fe191c79a6b4c6 Mon Sep 17 00:00:00 2001 From: Cosmin Atanasiu Date: Wed, 21 Oct 2026 10:15:20 -0700 Subject: [PATCH 10/10] PLA-1859 Compare retry times independent of timezone label --- spec/unit/daemon/store/redis_spec.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spec/unit/daemon/store/redis_spec.rb b/spec/unit/daemon/store/redis_spec.rb index 0c1e4eacb..b969b59cf 100644 --- a/spec/unit/daemon/store/redis_spec.rb +++ b/spec/unit/daemon/store/redis_spec.rb @@ -110,7 +110,7 @@ expect(Rpush::Client::Redis::Notification).to receive(:find).with(notification.id, other.id).and_call_original store.mark_ids_retryable([notification.id, other.id], deliver_after) - expect([notification.reload.deliver_after, other.reload.deliver_after].map(&:to_s)).to all(eq(deliver_after.utc.to_s)) + expect([notification.reload.deliver_after, other.reload.deliver_after].map(&:to_i)).to all(eq(deliver_after.to_i)) end end