From 7a91d136cb3ab7551908e4048d17cd8f704bc84e Mon Sep 17 00:00:00 2001 From: Robert Stojanovski Date: Fri, 21 Aug 2026 08:54:16 -0400 Subject: [PATCH] Retry APNs2 frames dropped by a mid-flight connection reset Apns2::Delivery has the same latent silent-drop pattern PR #7 fixed for Apnsp8, called out there as a deliberate follow-up: a dropped HTTP/2 connection tears down its in-flight streams via net-http2's on(:error) callback rather than raising into #perform, so those notifications never receive an on(:close) and are marked neither delivered, failed, nor retryable -- silently discarded when the batch completes. This is the transport a cert-based app (e.g. currypizzahouse_ios) uses, observed in production as a connection that goes completely silent for hours -- no sends, no errors logged -- then resumes on its own with no restart. Reused a single push message's rpush_notifications rows confirm the same request succeeding on one attempt and silently vanishing (no delivered/failed/retryable outcome) on another, minutes apart, against the same two device tokens. Fix, mirroring Apnsp8::Delivery#perform and reusing Batch#unresolved (already added by #7, transport-agnostic): - Apns2::Delivery#perform reconciles after #join: any unresolved notification is re-queued (retryable) instead of dropped. - handle_response treats an absent status code (stream closed before APNs answered) as a transport failure -> retry, not a permanent failure (previously this fell through to the `else` branch and was marked permanently *failed* -- worse than Apnsp8's pre-#7 silent drop). - Also fixes the untested per-notification SSLError rescue named in #7's "Follow-ups" section: preparing a request could raise before the notification ever got a stream, and the old code just logged and moved on, leaving it with no outcome at all. Now retried via the same path. Every existing APNs status outcome (200/410/400/429/500/503) is unchanged; only the "no verdict from APNs" cases move from {silent drop, permanent fail} to {retry} -- strictly safer, matching #7's regression-safety argument for Apnsp8. Also brings structured push-event logging (#7) to this transport for parity: delivered/failed/retrying events on Delivery, and the dispatcher's connection_error now includes the error message (not just its class) -- addressing the one open review comment on #7 before it repeats here. Tests mirror #7's apnsp8 coverage: the reconnection sweep, no-status handling, the SSLError-at-prepare-time path, and the logging format cases. Stacked on rstojano/apnsp8-retry-dropped-frames (#7) to reuse Batch#unresolved and Loggable#log_push_event without redefining them; rebase onto master once #7 merges. --- lib/rpush/daemon/apns2/delivery.rb | 92 +++++++++-- lib/rpush/daemon/dispatcher/apns_http2.rb | 6 +- spec/functional/apns2_spec.rb | 10 +- spec/unit/daemon/apns2/delivery_spec.rb | 146 ++++++++++++++++++ .../unit/daemon/dispatcher/apns_http2_spec.rb | 35 +++++ 5 files changed, 269 insertions(+), 20 deletions(-) create mode 100644 spec/unit/daemon/apns2/delivery_spec.rb create mode 100644 spec/unit/daemon/dispatcher/apns_http2_spec.rb diff --git a/lib/rpush/daemon/apns2/delivery.rb b/lib/rpush/daemon/apns2/delivery.rb index f67e96d7e..88461a216 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 88a159bb4..7ab666a91 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 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/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