From 961ac5d4194b2f825c2bba986e99017c29b4dae1 Mon Sep 17 00:00:00 2001
From: Robert Stojanovski <robert@thanx.com>
Date: Thu, 20 Aug 2026 13:25:29 -0400
Subject: [PATCH 1/4] Retry APNs (apnsp8) frames dropped by a mid-flight
 connection reset

When the APNs HTTP/2 connection is reset while a batch is in flight,
net-http2 tears down the stream set and delivers the error to the
client's on(:error) callback; #join then returns normally. The in-flight
notifications never receive an on(:close), so they are marked neither
delivered, failed, nor retryable, and the batch discards them silently on
completion -- no retry, no failure record. An idle apnsp8 socket (e.g. a
low-volume Live Activity app) that APNs resets loses the next frame this
way.

Fix, localized to the apnsp8 transport plus a read-only Batch helper:

- Batch#unresolved returns notifications with no terminal outcome.
- Apnsp8::Delivery#perform reconciles after #join: any unresolved
  notification is re-queued (retryable) instead of dropped. No-op on the
  normal path where every stream reported a result.
- handle_response treats an absent status code (stream closed before APNs
  answered) as a transport failure -> retry, not a permanent failure.
- Errno::ECONNRESET added to the retryable rescue (matching apns2), in
  case a reset ever surfaces synchronously rather than via on(:error).

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}.

Specs: Batch#unresolved, plus apnsp8 reconciliation and no-status handling.
---
 lib/rpush/daemon/apnsp8/delivery.rb      | 43 ++++++++++++++--
 lib/rpush/daemon/batch.rb                | 12 +++++
 spec/unit/daemon/apnsp8/delivery_spec.rb | 65 ++++++++++++++++++++++--
 spec/unit/daemon/batch_spec.rb           | 27 ++++++++++
 4 files changed, 140 insertions(+), 7 deletions(-)

diff --git a/lib/rpush/daemon/apnsp8/delivery.rb b/lib/rpush/daemon/apnsp8/delivery.rb
index 276525b7d..a7aed6b49 100644
--- a/lib/rpush/daemon/apnsp8/delivery.rb
+++ b/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,
@@ -119,13 +139,28 @@ def ok(notification)
         end
 
         def service_unavailable(notification, response)
-          @batch.mark_retryable(notification, Time.now + 10.seconds)
+          @batch.mark_retryable(notification, Time.now + RECONNECT_RETRY_DELAY)
           # 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)
         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)
+        end
+
         def build_request(notification)
           {
             path:    "/3/device/#{notification.device_token}",
diff --git a/lib/rpush/daemon/batch.rb b/lib/rpush/daemon/batch.rb
index 92d585515..c7cfa22b5 100644
--- a/lib/rpush/daemon/batch.rb
+++ b/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] ||= []
diff --git a/spec/unit/daemon/apnsp8/delivery_spec.rb b/spec/unit/daemon/apnsp8/delivery_spec.rb
index b1810cf5a..ab1dba171 100644
--- a/spec/unit/daemon/apnsp8/delivery_spec.rb
+++ b/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,18 @@
     )
   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 '#perform' do
@@ -49,5 +53,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
diff --git a/spec/unit/daemon/batch_spec.rb b/spec/unit/daemon/batch_spec.rb
index aeb2f16f8..3df197521 100644
--- a/spec/unit/daemon/batch_spec.rb
+++ b/spec/unit/daemon/batch_spec.rb
@@ -135,6 +135,33 @@
     end
   end
 
+  describe 'unresolved' do
+    it 'returns every notification with no outcome' do
+      expect(batch.unresolved).to eq [notification1, notification2]
+    end
+
+    it 'excludes delivered notifications' do
+      batch.mark_delivered(notification1)
+      expect(batch.unresolved).to eq [notification2]
+    end
+
+    it 'excludes failed notifications' do
+      batch.mark_failed(notification1, 400, 'BadDeviceToken')
+      expect(batch.unresolved).to eq [notification2]
+    end
+
+    it 'excludes retryable notifications' do
+      batch.mark_retryable(notification1, time)
+      expect(batch.unresolved).to eq [notification2]
+    end
+
+    it 'returns nothing once every notification resolved' do
+      batch.mark_delivered(notification1)
+      batch.mark_failed(notification2, 400, 'BadDeviceToken')
+      expect(batch.unresolved).to be_empty
+    end
+  end
+
   describe 'complete' do
     before do
       allow(Rpush).to receive_messages(logger: double.as_null_object)

From 0629bdc7c23deb5115d57cfe50681c8ac25cfca6 Mon Sep 17 00:00:00 2001
From: Robert Stojanovski <robert@thanx.com>
Date: Thu, 20 Aug 2026 16:29:28 -0400
Subject: [PATCH 2/4] Structured push-pipeline logging (apnsp8 transport +
 Loggable helper)

Add Loggable#log_push_event, which emits logfmt-style key=value lines so
Datadog indexes each field and the lines join to the nexus push logs on
rpush_notification_id. Give every apnsp8 notification a discoverable
outcome:

- delivery: delivered / failed / retrying (with the device token truncated),
  and a retryable APNs response is logged as a retry, not a failure
- app_runner: enqueued per notification (both the batch and non-batch
  paths), and startup_failed when an app cannot start (bad cert)
- dispatcher: connection_error on a socket-level client error, so a
  mid-flight reset is findable

Logging only; no delivery behavior changes.
---
 lib/rpush/daemon/apnsp8/delivery.rb           | 33 ++++++----
 lib/rpush/daemon/app_runner.rb                |  4 +-
 lib/rpush/daemon/dispatcher/apnsp8_http2.rb   |  2 +-
 lib/rpush/daemon/loggable.rb                  | 24 +++++++
 spec/unit/daemon/apnsp8/delivery_spec.rb      | 35 ++++++++++-
 spec/unit/daemon/app_runner_spec.rb           | 21 ++++++-
 .../daemon/dispatcher/apnsp8_http2_spec.rb    | 34 ++++++++++
 spec/unit/daemon/loggable_spec.rb             | 63 +++++++++++++++++++
 8 files changed, 200 insertions(+), 16 deletions(-)
 create mode 100644 spec/unit/daemon/dispatcher/apnsp8_http2_spec.rb
 create mode 100644 spec/unit/daemon/loggable_spec.rb

diff --git a/lib/rpush/daemon/apnsp8/delivery.rb b/lib/rpush/daemon/apnsp8/delivery.rb
index a7aed6b49..1c764d091 100644
--- a/lib/rpush/daemon/apnsp8/delivery.rb
+++ b/lib/rpush/daemon/apnsp8/delivery.rb
@@ -134,16 +134,17 @@ 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 + RECONNECT_RETRY_DELAY)
-          # 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)
+          # 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
@@ -158,7 +159,7 @@ def retry_unresolved
         # 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)
+          retry_message_to_log(notification, reason: 'connection_lost')
         end
 
         def build_request(notification)
@@ -193,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
diff --git a/lib/rpush/daemon/app_runner.rb b/lib/rpush/daemon/app_runner.rb
index 6e7311d24..6e1e83499 100644
--- a/lib/rpush/daemon/app_runner.rb
+++ b/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
diff --git a/lib/rpush/daemon/dispatcher/apnsp8_http2.rb b/lib/rpush/daemon/dispatcher/apnsp8_http2.rb
index 522d97bbb..6c98a0237 100644
--- a/lib/rpush/daemon/dispatcher/apnsp8_http2.rb
+++ b/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)
             reflect(:error, error)
           end
           client
diff --git a/lib/rpush/daemon/loggable.rb b/lib/rpush/daemon/loggable.rb
index 015a3e22f..544234ae6 100644
--- a/lib/rpush/daemon/loggable.rb
+++ b/lib/rpush/daemon/loggable.rb
@@ -21,8 +21,32 @@ 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=#{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)
+        str = value.to_s
+        return str unless str.match?(/[\s"=]/)
+
+        %("#{str.gsub('"', '\"')}")
+      end
+
       def app_prefix(msg)
         app = instance_variable_get('@app')
         msg = "[#{app.name}] #{msg}" if app
diff --git a/spec/unit/daemon/apnsp8/delivery_spec.rb b/spec/unit/daemon/apnsp8/delivery_spec.rb
index ab1dba171..9298dd52b 100644
--- a/spec/unit/daemon/apnsp8/delivery_spec.rb
+++ b/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', name: 'MY APP') }
+  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 }
 
@@ -35,6 +35,39 @@
     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
     context 'with an HTTP2 client where max concurrent streams is not set' do
       let(:max_concurrent_streams) { 0x7fffffff }
diff --git a/spec/unit/daemon/app_runner_spec.rb b/spec/unit/daemon/app_runner_spec.rb
index e979b2cfb..e2ba8f271 100644
--- a/spec/unit/daemon/app_runner_spec.rb
+++ b/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
diff --git a/spec/unit/daemon/dispatcher/apnsp8_http2_spec.rb b/spec/unit/daemon/dispatcher/apnsp8_http2_spec.rb
new file mode 100644
index 000000000..f3694d190
--- /dev/null
+++ b/spec/unit/daemon/dispatcher/apnsp8_http2_spec.rb
@@ -0,0 +1,34 @@
+require 'unit_spec_helper'
+
+describe Rpush::Daemon::Dispatcher::Apnsp8Http2 do
+  let(:app) { double(name: 'my_app', environment: 'production') }
+  let(:delivery_class) { double('DeliveryClass') }
+  let(:logger) { double('logger').as_null_object }
+  let(:client) { double('NetHttp2::Client') }
+
+  subject(:dispatcher) { described_class.new(app, delivery_class) }
+
+  before do
+    allow(Rpush).to receive_messages(logger: logger)
+    allow(Rpush::Daemon::Apnsp8::Token).to receive(:new).and_return(double('Token'))
+    @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 when the connection raises a socket error' do
+      dispatcher
+      allow(dispatcher).to receive(:reflect)
+      expect(logger).to receive(:error)
+        .with('event=connection_error app=my_app error=Errno::ECONNRESET')
+      @callbacks[:error].call(Errno::ECONNRESET.new('Connection reset by peer'))
+    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/loggable_spec.rb b/spec/unit/daemon/loggable_spec.rb
new file mode 100644
index 000000000..07e887a42
--- /dev/null
+++ b/spec/unit/daemon/loggable_spec.rb
@@ -0,0 +1,63 @@
+require 'unit_spec_helper'
+
+describe Rpush::Daemon::Loggable do
+  let(:logger) { double(info: nil, warn: nil, error: nil) }
+  before { allow(Rpush).to receive_messages(logger: logger) }
+
+  let(:klass) do
+    Class.new do
+      include Rpush::Daemon::Loggable
+      def initialize(app = nil)
+        @app = app
+      end
+    end
+  end
+
+  let(:app) { double(name: 'my_app') }
+  let(:notification) { double(id: 42) }
+
+  describe '#log_push_event' do
+    it 'writes a structured line with event, notification id, app, and fields at info level' do
+      obj = klass.new(app)
+      expect(logger).to receive(:info).with('event=delivered rpush_notification_id=42 app=my_app code=200')
+      obj.log_push_event(:delivered, notification: notification, code: 200)
+    end
+
+    it 'routes the warn level to the warn logger' do
+      obj = klass.new(app)
+      expect(logger).to receive(:warn).with('event=retrying rpush_notification_id=42 app=my_app reason=503')
+      obj.log_push_event(:retrying, notification: notification, level: :warn, reason: 503)
+    end
+
+    it 'routes the error level to the error logger' do
+      obj = klass.new(app)
+      expect(logger).to receive(:error).with('event=failed rpush_notification_id=42 app=my_app code=410 reason=Unregistered')
+      obj.log_push_event(:failed, notification: notification, level: :error, code: 410, reason: 'Unregistered')
+    end
+
+    it 'omits the notification id when no notification is given' do
+      obj = klass.new(app)
+      expect(logger).to receive(:error).with('event=connection_error app=my_app error=Errno::ECONNRESET')
+      obj.log_push_event(:connection_error, level: :error, error: 'Errno::ECONNRESET')
+    end
+
+    it 'prefers an explicitly passed app over the instance app' do
+      obj = klass.new(nil)
+      other = double(name: 'other_app')
+      expect(logger).to receive(:info).with('event=startup_failed app=other_app reason="bad cert"')
+      obj.log_push_event(:startup_failed, app: other, reason: 'bad cert')
+    end
+
+    it 'quotes field values that contain whitespace' do
+      obj = klass.new(app)
+      expect(logger).to receive(:warn).with('event=retrying rpush_notification_id=42 app=my_app deliver_after="2026-08-20 16:21:36"')
+      obj.log_push_event(:retrying, notification: notification, level: :warn, deliver_after: '2026-08-20 16:21:36')
+    end
+
+    it 'omits fields whose value is nil' do
+      obj = klass.new(app)
+      expect(logger).to receive(:info).with('event=delivered rpush_notification_id=42 app=my_app')
+      obj.log_push_event(:delivered, notification: notification, device_token: nil)
+    end
+  end
+end

From de49c70370e02bb3538069a55d3ca3c2e8b3078a Mon Sep 17 00:00:00 2001
From: Robert Stojanovski <robert@thanx.com>
Date: Thu, 20 Aug 2026 17:08:37 -0400
Subject: [PATCH 3/4] Quote the app name in structured push logs

The app name bypassed push_event_value, so a name containing whitespace,
=, or " produced an unparseable logfmt line and broke Datadog field
indexing. Route it through the same quoter as every other field.
---
 lib/rpush/daemon/loggable.rb      | 2 +-
 spec/unit/daemon/loggable_spec.rb | 6 ++++++
 2 files changed, 7 insertions(+), 1 deletion(-)

diff --git a/lib/rpush/daemon/loggable.rb b/lib/rpush/daemon/loggable.rb
index 544234ae6..f1052444e 100644
--- a/lib/rpush/daemon/loggable.rb
+++ b/lib/rpush/daemon/loggable.rb
@@ -30,7 +30,7 @@ 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=#{name}" unless name.nil?
+        parts << "app=#{push_event_value(name)}" unless name.nil?
         fields.each do |key, value|
           next if value.nil?
           parts << "#{key}=#{push_event_value(value)}"
diff --git a/spec/unit/daemon/loggable_spec.rb b/spec/unit/daemon/loggable_spec.rb
index 07e887a42..4402e754e 100644
--- a/spec/unit/daemon/loggable_spec.rb
+++ b/spec/unit/daemon/loggable_spec.rb
@@ -54,6 +54,12 @@ def initialize(app = nil)
       obj.log_push_event(:retrying, notification: notification, level: :warn, deliver_after: '2026-08-20 16:21:36')
     end
 
+    it 'quotes the app name when it contains whitespace' do
+      obj = klass.new(double(name: 'My App'))
+      expect(logger).to receive(:info).with('event=delivered rpush_notification_id=42 app="My App"')
+      obj.log_push_event(:delivered, notification: notification)
+    end
+
     it 'omits fields whose value is nil' do
       obj = klass.new(app)
       expect(logger).to receive(:info).with('event=delivered rpush_notification_id=42 app=my_app')

From cb86da04d07428f24d3ea613f3cc4012e9dd8a0e Mon Sep 17 00:00:00 2001
From: Robert Stojanovski <robert@thanx.com>
Date: Fri, 21 Aug 2026 06:14:42 -0400
Subject: [PATCH 4/4] Log the error message on connection_error, and keep push
 logs single-line

Review follow-up: the connection_error line logged only the exception class,
so two same-class socket errors with different messages were indistinguishable
in a Datadog query. Include the message. Also fold newlines in structured
field values so a multi-line message cannot split the one-line record.
---
 lib/rpush/daemon/dispatcher/apnsp8_http2.rb      | 2 +-
 lib/rpush/daemon/loggable.rb                     | 4 +++-
 spec/unit/daemon/dispatcher/apnsp8_http2_spec.rb | 6 +++---
 spec/unit/daemon/loggable_spec.rb                | 6 ++++++
 4 files changed, 13 insertions(+), 5 deletions(-)

diff --git a/lib/rpush/daemon/dispatcher/apnsp8_http2.rb b/lib/rpush/daemon/dispatcher/apnsp8_http2.rb
index 6c98a0237..5b8ade3ef 100644
--- a/lib/rpush/daemon/dispatcher/apnsp8_http2.rb
+++ b/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_push_event(:connection_error, level: :error, error: error.class)
+            log_push_event(:connection_error, level: :error, error: "#{error.class}: #{error.message}")
             reflect(:error, error)
           end
           client
diff --git a/lib/rpush/daemon/loggable.rb b/lib/rpush/daemon/loggable.rb
index f1052444e..4c23f0a18 100644
--- a/lib/rpush/daemon/loggable.rb
+++ b/lib/rpush/daemon/loggable.rb
@@ -41,7 +41,9 @@ def log_push_event(event, app: nil, notification: nil, level: :info, **fields)
       private
 
       def push_event_value(value)
-        str = value.to_s
+        # 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('"', '\"')}")
diff --git a/spec/unit/daemon/dispatcher/apnsp8_http2_spec.rb b/spec/unit/daemon/dispatcher/apnsp8_http2_spec.rb
index f3694d190..267c8c984 100644
--- a/spec/unit/daemon/dispatcher/apnsp8_http2_spec.rb
+++ b/spec/unit/daemon/dispatcher/apnsp8_http2_spec.rb
@@ -17,12 +17,12 @@
   end
 
   describe 'the client error callback' do
-    it 'logs a structured connection_error event when the connection raises a socket error' do
+    it 'logs a structured connection_error event with the error class and message' do
       dispatcher
       allow(dispatcher).to receive(:reflect)
       expect(logger).to receive(:error)
-        .with('event=connection_error app=my_app error=Errno::ECONNRESET')
-      @callbacks[:error].call(Errno::ECONNRESET.new('Connection reset by peer'))
+        .with('event=connection_error app=my_app error="SocketError: Socket was remotely closed"')
+      @callbacks[:error].call(SocketError.new('Socket was remotely closed'))
     end
 
     it 'still reflects the error so upstream handlers fire' do
diff --git a/spec/unit/daemon/loggable_spec.rb b/spec/unit/daemon/loggable_spec.rb
index 4402e754e..f289ce1f9 100644
--- a/spec/unit/daemon/loggable_spec.rb
+++ b/spec/unit/daemon/loggable_spec.rb
@@ -60,6 +60,12 @@ def initialize(app = nil)
       obj.log_push_event(:delivered, notification: notification)
     end
 
+    it 'collapses newlines in field values so the record stays one line' do
+      obj = klass.new(app)
+      expect(logger).to receive(:error).with('event=connection_error app=my_app error="a b"')
+      obj.log_push_event(:connection_error, level: :error, error: "a\nb")
+    end
+
     it 'omits fields whose value is nil' do
       obj = klass.new(app)
       expect(logger).to receive(:info).with('event=delivered rpush_notification_id=42 app=my_app')