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/daemon/apns2/delivery.rb b/lib/rpush/daemon/apns2/delivery.rb index f67e96d7e..9cda5fb29 100644 --- a/lib/rpush/daemon/apns2/delivery.rb +++ b/lib/rpush/daemon/apns2/delivery.rb @@ -8,6 +8,10 @@ module Apns2 class Delivery < Rpush::Daemon::Delivery RETRYABLE_CODES = [ 429, 500, 503 ] CLIENT_JOIN_TIMEOUT = 60 + # How long to defer a notification whose delivery could not be confirmed at the + # transport level: a dropped connection, an SSL failure, or a stream that closed + # with no APNs status. Matches the service-unavailable backoff. + RECONNECT_RETRY_DELAY = 10.seconds def initialize(app, http2_client, batch) @app = app @@ -19,19 +23,29 @@ def perform @batch.each_notification do |notification| begin prepare_async_post(notification) - rescue OpenSSL::SSL::SSLError => error - log_error("Notification #{notification.id} failed with SSL error") + rescue OpenSSL::SSL::SSLError + # The TLS handshake/write failed before this notification got a stream, so it + # holds no outcome. Re-queue it (and keep processing the rest of the batch) + # instead of dropping it with only a log line. + connection_lost(notification) end end # Send all preprocessed requests at once @client.join(timeout: CLIENT_JOIN_TIMEOUT) + + # A dropped connection tears down its in-flight streams WITHOUT raising here: + # net-http2 hands the socket error to the client's on(:error) callback and #join + # returns once the stream set is emptied. Those notifications never received an + # on(:close), so they hold no outcome and would be silently discarded when the + # batch completes. Re-queue them instead. No-op on the normal path. + retry_unresolved rescue NetHttp2::AsyncRequestTimeout => error - mark_batch_retryable(Time.now + 10.seconds, error) + mark_batch_retryable(Time.now + RECONNECT_RETRY_DELAY, error) @client.close raise rescue Errno::ECONNREFUSED, SocketError, Errno::ECONNRESET => error - mark_batch_retryable(Time.now + 10.seconds, error) + mark_batch_retryable(Time.now + RECONNECT_RETRY_DELAY, error) raise rescue StandardError => error mark_batch_failed(error) @@ -74,6 +88,11 @@ def handle_response(notification, response) ok(notification) when *RETRYABLE_CODES service_unavailable(notification, response) + when nil + # The stream closed before any :status header arrived — APNs returned no verdict + # (the connection dropped mid-flight). Transport failure, not an APNs rejection, + # so retry rather than mark it permanently failed. + connection_lost(notification) else reflect(:notification_id_failed, @app, @@ -85,16 +104,32 @@ def handle_response(notification, response) end def ok(notification) - log_info("#{notification.id} sent to #{notification.device_token}") + log_push_event(:delivered, notification: notification, + device_token: truncate_device_token(notification.device_token)) @batch.mark_delivered(notification) end def service_unavailable(notification, response) - @batch.mark_retryable(notification, Time.now + 10.seconds) - # Logs should go last as soon as we need to initialize - # retry time to display it in log - failed_message_to_log(notification, response) - retry_message_to_log(notification) + @batch.mark_retryable(notification, Time.now + RECONNECT_RETRY_DELAY) + # A retryable APNs response (429/500/503) is not a failure: log it only as a retry + # so the failure signal stays clean. Logs go last, after mark_retryable has set the + # deliver_after we want to display. + retry_message_to_log(notification, reason: response[:code]) + end + + # Re-queue every notification the batch never resolved — one whose HTTP/2 stream was + # abandoned when the connection dropped, so its on(:close) never fired and it holds no + # delivered/failed/retryable outcome. A no-op when every stream reported a result. + def retry_unresolved + @batch.unresolved.each { |notification| connection_lost(notification) } + end + + # A notification with no delivery outcome from APNs: retry it on a fresh connection + # rather than discard it. Shared by the mid-flight-drop sweep (#retry_unresolved), the + # no-status branch of #handle_response, and an SSL failure during #perform. + def connection_lost(notification) + @batch.mark_retryable(notification, Time.now + RECONNECT_RETRY_DELAY) + retry_message_to_log(notification, reason: 'connection_lost') end def build_request(notification) @@ -125,15 +160,23 @@ def notification_data(notification) notification.data || {} end - def retry_message_to_log(notification) - log_warn("Notification #{notification.id} will be retried after "\ - "#{notification.deliver_after.strftime('%Y-%m-%d %H:%M:%S')} "\ - "(retry #{notification.retries}).") + def retry_message_to_log(notification, reason:) + log_push_event(:retrying, notification: notification, level: :warn, + reason: reason, + retry: notification.retries, + deliver_after: notification.deliver_after&.strftime('%Y-%m-%d %H:%M:%S')) end def failed_message_to_log(notification, response) - log_error("Notification #{notification.id} failed, "\ - "#{response[:code]}/#{response[:failure_reason]}") + log_push_event(:failed, notification: notification, level: :error, + code: response[:code], reason: response[:failure_reason]) + end + + # Keep the raw device token out of logs; a short prefix is enough to correlate. + def truncate_device_token(token) + return token if token.nil? || token.length <= 8 + + "#{token[0, 8]}…" end end end diff --git a/lib/rpush/daemon/dispatcher/apns_http2.rb b/lib/rpush/daemon/dispatcher/apns_http2.rb index 88a159bb4..be364a7c3 100644 --- a/lib/rpush/daemon/dispatcher/apns_http2.rb +++ b/lib/rpush/daemon/dispatcher/apns_http2.rb @@ -34,7 +34,7 @@ def create_http2_client(app) url = URLS[app.environment.to_sym] client = NetHttp2::Client.new(url, ssl_context: prepare_ssl_context, connect_timeout: DEFAULT_TIMEOUT) client.on(:error) do |error| - log_error(error) + log_push_event(:connection_error, level: :error, error: "#{error.class}: #{error.message}") reflect(:error, error) end client 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/apns2_spec.rb b/spec/functional/apns2_spec.rb index 5824d087f..3415a685e 100644 --- a/spec/functional/apns2_spec.rb +++ b/spec/functional/apns2_spec.rb @@ -270,10 +270,12 @@ def create_notification expect(fake_client).to receive(:call_async) { raise(OpenSSL::SSL::SSLError) } end - it 'logs the error' do - expect(Rpush.logger).to receive(:error) - create_notification - Rpush.push + it 'fails but retries delivery several times' do + notification = create_notification + expect do + Rpush.push + notification.reload + end.to change(notification, :retries) end 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/daemon/apns2/delivery_spec.rb b/spec/unit/daemon/apns2/delivery_spec.rb new file mode 100644 index 000000000..c8f446314 --- /dev/null +++ b/spec/unit/daemon/apns2/delivery_spec.rb @@ -0,0 +1,146 @@ +require 'unit_spec_helper' + +describe Rpush::Daemon::Apns2::Delivery do + subject(:delivery) { described_class.new(app, http2_client, batch) } + + let(:app) { double(bundle_id: 'MY BUNDLE ID', name: 'my_app') } + let(:notification1) { double('Notification 1', data: {}, as_json: {}).as_null_object } + let(:notification2) { double('Notification 2', data: {}, as_json: {}).as_null_object } + + let(:http_request) { double(on: nil) } + let(:http2_client) do + double( + call_async: nil, + join: nil, + prepare_request: http_request + ) + end + + let(:batch) do + double(mark_delivered: nil, mark_retryable: nil, mark_failed: nil, + all_processed: nil, unresolved: []) + end + let(:logger) { double('logger').as_null_object } + + before do + allow(batch).to receive(:each_notification) do |&blk| + [notification1, notification2].each(&blk) + end + allow(Rpush).to receive_messages(logger: logger) + allow(delivery).to receive(:reflect) + end + + describe 'structured outcome logging' do + let(:notification) do + double('Notification', + id: 7, + device_token: 'abcdef0123456789', + retries: 1, + deliver_after: Time.parse('2026-08-20 16:21:36 UTC')) + end + + it 'logs a structured delivered event on success' do + allow(batch).to receive(:mark_delivered) + expect(logger).to receive(:info) + .with('event=delivered rpush_notification_id=7 app=my_app device_token=abcdef01…') + delivery.send(:ok, notification) + end + + it 'logs a structured failed event on an APNs rejection' do + allow(batch).to receive(:mark_failed) + expect(logger).to receive(:error) + .with('event=failed rpush_notification_id=7 app=my_app code=410 reason=Unregistered') + delivery.send(:handle_response, notification, code: 410, failure_reason: 'Unregistered') + end + + it 'logs a structured retrying event for a retryable code and does not log a failure' do + allow(batch).to receive(:mark_retryable) + expect(logger).not_to receive(:error) + expect(logger).to receive(:warn) + .with('event=retrying rpush_notification_id=7 app=my_app reason=503 retry=1 deliver_after="2026-08-20 16:21:36"') + delivery.send(:service_unavailable, notification, code: 503, failure_reason: 'ServiceUnavailable') + end + end + + describe '#perform' do + context 'when the connection drops and a stream is left unresolved' do + before { allow(batch).to receive(:unresolved).and_return([notification1]) } + + it 'marks the abandoned notification retryable rather than discarding it' do + expect(batch).to receive(:mark_retryable).with(notification1, anything) + delivery.perform + end + end + + context 'when every stream resolved' do + it 'marks nothing retryable' do + expect(batch).not_to receive(:mark_retryable) + delivery.perform + end + end + + context 'when preparing a request raises an SSL error before it gets a stream' do + before do + allow(delivery).to receive(:prepare_async_post) do |notification| + raise OpenSSL::SSL::SSLError, 'session ticket not found' if notification == notification1 + end + end + + it 'marks the notification retryable instead of silently skipping it' do + expect(batch).to receive(:mark_retryable).with(notification1, anything) + delivery.perform + end + + it 'does not abort the batch — the next notification is still attempted' do + attempted = [] + allow(delivery).to receive(:prepare_async_post) do |notification| + attempted << notification + raise OpenSSL::SSL::SSLError, 'session ticket not found' if notification == notification1 + end + + delivery.perform + + expect(attempted).to eq([notification1, notification2]) + end + end + end + + describe '#handle_response' do + def handle(response) + delivery.send(:handle_response, notification1, response) + end + + context 'with a 200 status' do + it 'marks the notification delivered' do + expect(batch).to receive(:mark_delivered).with(notification1) + handle(code: 200) + end + end + + context 'with a retryable status' do + it 'marks the notification retryable' do + expect(batch).to receive(:mark_retryable).with(notification1, anything) + handle(code: 503) + end + end + + context 'with no status (stream closed before APNs responded)' do + it 'marks the notification retryable, not failed' do + expect(batch).to receive(:mark_retryable).with(notification1, anything) + handle({}) + end + + it 'does not mark the notification failed' do + expect(batch).not_to receive(:mark_failed) + handle({}) + end + end + + context 'with a permanent error status' do + it 'marks the notification failed' do + expect(batch).to receive(:mark_failed).with(notification1, 400, 'BadDeviceToken') + handle(code: 400, failure_reason: 'BadDeviceToken') + end + end + end +end diff --git a/spec/unit/daemon/dispatcher/apns_http2_spec.rb b/spec/unit/daemon/dispatcher/apns_http2_spec.rb new file mode 100644 index 000000000..076a690cd --- /dev/null +++ b/spec/unit/daemon/dispatcher/apns_http2_spec.rb @@ -0,0 +1,35 @@ +require 'unit_spec_helper' + +describe Rpush::Daemon::Dispatcher::ApnsHttp2 do + let(:app) { double(name: 'my_app', environment: 'production', certificate: 'cert', password: 'pass') } + let(:delivery_class) { double('DeliveryClass') } + let(:logger) { double('logger').as_null_object } + let(:client) { double('NetHttp2::Client') } + let(:ssl_context) { double('OpenSSL::SSL::SSLContext') } + + subject(:dispatcher) { described_class.new(app, delivery_class) } + + before do + allow(Rpush).to receive_messages(logger: logger) + allow_any_instance_of(described_class).to receive(:prepare_ssl_context).and_return(ssl_context) + @callbacks = {} + allow(NetHttp2::Client).to receive(:new).and_return(client) + allow(client).to receive(:on) { |event, &blk| @callbacks[event] = blk } + end + + describe 'the client error callback' do + it 'logs a structured connection_error event including the error message, not just its class' do + dispatcher + allow(dispatcher).to receive(:reflect) + expect(logger).to receive(:error) + .with('event=connection_error app=my_app error="Errno::ECONNRESET: Connection reset by peer"') + @callbacks[:error].call(Errno::ECONNRESET.new) + end + + it 'still reflects the error so upstream handlers fire' do + dispatcher + expect(dispatcher).to receive(:reflect).with(:error, kind_of(Errno::ECONNRESET)) + @callbacks[:error].call(Errno::ECONNRESET.new('Connection reset by peer')) + end + end +end 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_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