Retry APNs2 frames dropped by a mid-flight connection reset, and add structured push logging

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

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

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
 

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

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