Retry APNs (apnsp8) frames dropped by a mid-flight connection reset, and add structured push logging

lib/rpush/daemon/apnsp8/delivery.rb

@@ -9,6 +9,10 @@ class Delivery < Rpush::Daemon::Delivery
         RETRYABLE_CODES = [ 429, 500, 503 ]
         CLIENT_JOIN_TIMEOUT = 60
         DEFAULT_MAX_CONCURRENT_STREAMS = 100
+        # 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.
+        RECONNECT_RETRY_DELAY = 10.seconds
 
         def initialize(app, http2_client, token_provider, batch)
           @app = app
@@ -25,13 +29,24 @@ def perform
 
           # 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.
+          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, HTTP2::Error::StreamLimitExceeded => error
+        rescue Errno::ECONNREFUSED, SocketError, Errno::ECONNRESET, HTTP2::Error::StreamLimitExceeded => error
           # TODO restart connection when StreamLimitExceeded
-          mark_batch_retryable(Time.now + 10.seconds, error)
+          # ECONNRESET (an established connection reset by the peer) is retryable like a
+          # refused/failed connection: should it ever surface synchronously here rather than
+          # via the async on(:error) path above, it must not fall through to mark_batch_failed.
+          mark_batch_retryable(Time.now + RECONNECT_RETRY_DELAY, error)
           raise
         rescue StandardError => error
           mark_batch_failed(error)
@@ -103,6 +118,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,
@@ -114,16 +134,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) 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
 
         def build_request(notification)
@@ -158,15 +194,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

lib/rpush/daemon/app_runner.rb

@@ -33,7 +33,7 @@ def self.start_app(app)
         runner.start_loops
       rescue StandardError => e
         @runners.delete(app.id)
-        Rpush.logger.error("[#{app.name}] Exception raised during startup. Notifications will not be delivered for this app.")
+        log_push_event(:startup_failed, app: app, level: :error, reason: e.message)
         Rpush.logger.error(e)
         reflect(:error, e)
       end
@@ -123,11 +123,13 @@ def enqueue(notifications)
             batch = Batch.new(batch_notifications)
             queue.push(QueuePayload.new(batch))
           end
+          notifications.each { |notification| log_push_event(:enqueued, notification: notification) }
         else
           batch = Batch.new(notifications)
           notifications.each do |notification|
             queue.push(QueuePayload.new(batch, notification))
             reflect(:notification_enqueued, notification)
+            log_push_event(:enqueued, notification: notification)
           end
         end
       end

lib/rpush/daemon/batch.rb

@@ -27,6 +27,18 @@ def each_delivered(&blk)
         @delivered.each(&blk)
       end
 
+      # Notifications that reached no terminal outcome — neither delivered, failed, nor
+      # retryable. A dropped connection abandons in-flight HTTP/2 streams: their per-request
+      # on(:close) never fires, so nothing is recorded for them and they would be silently
+      # discarded when the batch completes. A transport calls this after a send attempt to
+      # 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)
+          @notifications - resolved
+        end
+      end
+
       def mark_retryable(notification, deliver_after)
         @mutex.synchronize do
           @retryable[deliver_after] ||= []

lib/rpush/daemon/dispatcher/apnsp8_http2.rb

@@ -35,7 +35,7 @@ def create_http2_client(app)
           url = URLS[app.environment.to_sym]
           client = NetHttp2::Client.new(url, 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

lib/rpush/daemon/loggable.rb

@@ -21,8 +21,34 @@ def log_error(e)
         end
       end
 
+      # Emit a single structured push-pipeline log line in logfmt-style
+      # `key=value` pairs, so Datadog indexes each field and the lines join to
+      # the nexus push logs on `rpush_notification_id`. Field order is stable:
+      # event, notification id, app, then the caller's fields in the order given.
+      # nil-valued fields are dropped; values with whitespace/`=`/`"` are quoted.
+      def log_push_event(event, app: nil, notification: nil, level: :info, **fields)
+        parts = ["event=#{event}"]
+        parts << "rpush_notification_id=#{notification.id}" if notification
+        name = (app || instance_variable_get('@app'))&.name
+        parts << "app=#{push_event_value(name)}" unless name.nil?
+        fields.each do |key, value|
+          next if value.nil?
+          parts << "#{key}=#{push_event_value(value)}"
+        end
+        Rpush.logger.public_send(level, parts.join(' '))
+      end
+
       private
 
+      def push_event_value(value)
+        # A structured record is one line: fold any newline in the value to a space so a
+        # multi-line message (e.g. an exception) cannot split the record.
+        str = value.to_s.gsub(/[\r\n]+/, ' ')
+        return str unless str.match?(/[\s"=]/)
+
+        %("#{str.gsub('"', '\"')}")
+      end
+
       def app_prefix(msg)
         app = instance_variable_get('@app')
         msg = "[#{app.name}] #{msg}" if app

spec/unit/daemon/apnsp8/delivery_spec.rb

@@ -3,7 +3,7 @@
 describe Rpush::Daemon::Apnsp8::Delivery do
   subject(:delivery) { described_class.new(app, http2_client, token_provider, batch) }
 
-  let(:app) { double(bundle_id: 'MY BUNDLE ID') }
+  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 }
 
@@ -21,14 +21,51 @@
     )
   end
 
-  let(:batch) { double(mark_delivered: nil, all_processed: nil) }
-  let(:logger) { double(info: nil) }
+  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)
+      allow(delivery).to receive(:reflect)
+      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
@@ -49,5 +86,60 @@
         end
       end
     end
+
+    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
+  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/app_runner_spec.rb

@@ -75,6 +75,14 @@ class Delivery
     expect(Rpush.logger).to receive(:error)
     Rpush::Daemon::AppRunner.start_app(app)
   end
+
+  it 'logs a structured startup_failed event when startup raises' do
+    allow(Rpush::Daemon::AppRunner).to receive(:new).with(app).and_return(runner)
+    allow(runner).to receive(:start_dispatchers).and_raise(StandardError, 'bad cert')
+    Rpush::Daemon::AppRunner.start_app(app)
+    expect(Rpush.logger).to have_received(:error)
+      .with('event=startup_failed app=test reason="bad cert"')
+  end
 end
 
 describe Rpush::Daemon::AppRunner, 'debug' do
@@ -139,7 +147,7 @@ class Delivery
   end
 
   describe 'enqueue' do
-    let(:notification) { double }
+    let(:notification) { double(id: 1) }
 
     it 'enqueues the batch' do
       expect(queue).to receive(:push) do |queue_payload|
@@ -154,11 +162,22 @@ class Delivery
       runner.enqueue([notification])
     end
 
+    it 'logs a structured enqueued event for each notification' do
+      expect(logger).to receive(:info).with('event=enqueued rpush_notification_id=1 app=test')
+      runner.enqueue([notification])
+    end
+
     describe 'a service that batches deliveries' do
       before do
         allow(runner.send(:service)).to receive_messages(batch_deliveries?: true)
       end
 
+      it 'logs a structured enqueued event for each notification' do
+        allow(runner).to receive(:num_dispatcher_loops).and_return(1)
+        expect(logger).to receive(:info).with('event=enqueued rpush_notification_id=1 app=test')
+        runner.enqueue([notification])
+      end
+
       describe '1 notification with more than one dispatcher loop' do
         it 'does not raise ArgumentError: invalid slice size' do
           # https://github.com/rpush/rpush/issues/57