Isolate and report serialization errors

lib/segment/analytics/message_batch.rb

@@ -5,6 +5,8 @@ module Segment
   class Analytics
     # A batch of `Message`s to be sent to the API
     class MessageBatch
+      class JSONGenerationError < StandardError; end
+
       extend Forwardable
       include Segment::Analytics::Logging
       include Segment::Analytics::Defaults::MessageBatch
@@ -16,8 +18,13 @@ def initialize(max_message_count)
       end
 
       def <<(message)
-        message_json_size = message.to_json.bytesize
+        begin
+          message_json = message.to_json
+        rescue StandardError => e
+          raise JSONGenerationError, "Serialization error: #{e}"
+        end
 
+        message_json_size = message_json.bytesize
         if message_too_big?(message_json_size)
           logger.error('a message exceeded the maximum allowed size')
         else

lib/segment/analytics/worker.rb

@@ -38,11 +38,10 @@ def run
           return if @queue.empty?
 
           @lock.synchronize do
-            @batch << @queue.pop until @batch.full? || @queue.empty?
+            consume_message_from_queue! until @batch.full? || @queue.empty?
           end
 
           res = Request.new.post @write_key, @batch
-
           @on_error.call(res.status, res.error) unless res.status == 200
 
           @lock.synchronize { @batch.clear }
@@ -54,6 +53,14 @@ def run
       def is_requesting?
         @lock.synchronize { !@batch.empty? }
       end
+
+      private
+
+      def consume_message_from_queue!
+        @batch << @queue.pop
+      rescue MessageBatch::JSONGenerationError => e
+        @on_error.call(-1, e.to_s)
+      end
     end
   end
 end

spec/segment/analytics/worker_spec.rb

@@ -83,6 +83,29 @@ class Analytics
 
           expect(queue).to be_empty
         end
+
+        it 'calls on_error for bad json' do
+          bad_obj = Object.new
+          def bad_obj.to_json(*_args)
+            raise "can't serialize to json"
+          end
+
+          on_error = proc {}
+          expect(on_error).to receive(:call).once.with(-1, /serialize to json/)
+
+          good_message = Requested::TRACK
+          bad_message = Requested::TRACK.merge({ 'bad_obj' => bad_obj })
+
+          queue = Queue.new
+          queue << good_message
+          queue << bad_message
+
+          worker = described_class.new(queue,
+                                       'testsecret',
+                                       :on_error => on_error)
+          worker.run
+          expect(queue).to be_empty
+        end
       end
 
       describe '#is_requesting?' do