diff --git a/lib/rpush/daemon/apnsp8/delivery.rb b/lib/rpush/daemon/apnsp8/delivery.rb index 276525b7d..1c764d091 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, @@ -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 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/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/lib/rpush/daemon/dispatcher/apnsp8_http2.rb b/lib/rpush/daemon/dispatcher/apnsp8_http2.rb index 522d97bbb..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_error(error) + 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 015a3e22f..4c23f0a18 100644 --- a/lib/rpush/daemon/loggable.rb +++ b/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 diff --git a/spec/unit/daemon/apnsp8/delivery_spec.rb b/spec/unit/daemon/apnsp8/delivery_spec.rb index b1810cf5a..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') } + 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 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/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) 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..267c8c984 --- /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 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="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 + 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..f289ce1f9 --- /dev/null +++ b/spec/unit/daemon/loggable_spec.rb @@ -0,0 +1,75 @@ +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 '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 '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') + obj.log_push_event(:delivered, notification: notification, device_token: nil) + end + end +end