diff --git a/lib/rpush/daemon/apns2/delivery.rb b/lib/rpush/daemon/apns2/delivery.rb index f67e96d7..88461a21 100644 --- a/lib/rpush/daemon/apns2/delivery.rb +++ b/lib/rpush/daemon/apns2/delivery.rb @@ -8,6 +8,11 @@ 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, or a stream that closed with no APNs + # status. Matches the existing service-unavailable / connection-error backoff. + # Mirrors Apnsp8::Delivery. + RECONNECT_RETRY_DELAY = 10.seconds def initialize(app, http2_client, batch) @app = app @@ -20,18 +25,32 @@ def perform begin prepare_async_post(notification) rescue OpenSSL::SSL::SSLError => error - log_error("Notification #{notification.id} failed with SSL error") + # Building the request for THIS notification raised before it ever got a + # stream, so it will never receive an on(:close) — left alone it would hold no + # outcome at all (neither delivered, failed, nor retryable) and silently + # vanish from the batch when the next notification is processed. Retry it + # explicitly instead of just logging and moving on. + prepare_failed(notification, error) 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 so the frame lands on a fresh connection instead + # of vanishing. No-op on the normal path where every stream reported a result. + # (Mirrors Apnsp8::Delivery#perform — see PR #7.) + 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 +93,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). This is a 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 +109,45 @@ 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) and + # the no-status branch of #handle_response. + def connection_lost(notification) + @batch.mark_retryable(notification, Time.now + RECONNECT_RETRY_DELAY) + retry_message_to_log(notification, reason: 'connection_lost') + end + + # The per-notification request-preparation step raised before its stream existed + # (e.g. an SSL renegotiation error on this cert-based connection), so it will never + # receive an on(:close). Retry it like any other transport-level drop rather than + # silently skipping to the next notification in the batch. + def prepare_failed(notification, error) + @batch.mark_retryable(notification, Time.now + RECONNECT_RETRY_DELAY) + # Keep `reason` a stable, low-cardinality value for classification and put the + # exception in its own `error` field, consistent with the dispatcher's + # connection_error event. + retry_message_to_log(notification, reason: 'prepare_failed', + error: "#{error.class}: #{error.message}") end def build_request(notification) @@ -125,15 +178,24 @@ 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:, error: nil) + log_push_event(:retrying, notification: notification, level: :warn, + reason: reason, + error: error, + 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 88a159bb..7ab666a9 100644 --- a/lib/rpush/daemon/dispatcher/apns_http2.rb +++ b/lib/rpush/daemon/dispatcher/apns_http2.rb @@ -34,7 +34,11 @@ 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) + # Include the message, not just the class: this is the one line meant to make a + # mid-flight reset findable by app and time in Datadog, and two different socket + # errors of the same class (e.g. two distinct SSLError causes) are otherwise + # indistinguishable here. + log_push_event(:connection_error, level: :error, error: "#{error.class}: #{error.message}") reflect(:error, error) end client diff --git a/spec/functional/apns2_spec.rb b/spec/functional/apns2_spec.rb index 5824d087..3415a685 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/unit/daemon/apns2/delivery_spec.rb b/spec/unit/daemon/apns2/delivery_spec.rb new file mode 100644 index 00000000..c8f44631 --- /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 00000000..076a690c --- /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