fix: ensure metrics are emitted even if batch fails (#65)

* ensure that metrics are emitted even if process_batch errors continuously

* bump version, fix lockfiles

Author: effron

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    journaled (6.2.5)
+    journaled (6.2.6)
       activejob
       activerecord
       activesupport
@@ -258,4 +258,4 @@ DEPENDENCIES
   webmock
 
 BUNDLED WITH
-   2.6.3
+  4.0.7

gemfiles/rails_7_2.gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: ..
   specs:
-    journaled (6.2.5)
+    journaled (6.2.6)
       activejob
       activerecord
       activesupport
@@ -258,4 +258,4 @@ DEPENDENCIES
   webmock
 
 BUNDLED WITH
-   2.6.3
+  4.0.7

gemfiles/rails_8_0.gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: ..
   specs:
-    journaled (6.2.5)
+    journaled (6.2.6)
       activejob
       activerecord
       activesupport
@@ -259,4 +259,4 @@ DEPENDENCIES
   webmock
 
 BUNDLED WITH
-   2.6.3
+  4.0.7

lib/journaled/outbox/worker.rb

@@ -59,9 +59,10 @@ def run_loop
             break
           end
 
+          emit_metrics_if_needed
+
           begin
             process_batch
-            emit_metrics_if_needed
           rescue StandardError => e
             Rails.logger.error("Worker error: #{e.class} - #{e.message}")
             Rails.logger.error(e.backtrace.join("\n"))

lib/journaled/version.rb

@@ -1,5 +1,5 @@
 # frozen_string_literal: true
 
 module Journaled
-  VERSION = "6.2.5"
+  VERSION = "6.2.6"
 end

spec/lib/journaled/outbox/worker_spec.rb

@@ -180,6 +180,28 @@
         worker.start
         expect(processor).to have_received(:process_batch).at_least(:twice)
       end
+
+      it 'still emits queue metrics after errors' do
+        call_count = 0
+        allow(processor).to receive(:process_batch) do
+          call_count += 1
+          Timecop.travel(61.seconds) if call_count == 1
+          worker.shutdown if call_count >= 2
+          raise StandardError, 'Kinesis error'
+        end
+
+        emitted = {}
+        callback = ->(name, _started, _finished, _unique_id, payload) { emitted[name] = payload }
+
+        ActiveSupport::Notifications.subscribed(callback, /journaled\.worker\.queue_/) do
+          worker.start
+
+          timeout = 2.seconds.from_now
+          sleep 0.1 until emitted.key?('journaled.worker.queue_total_count') || Time.current > timeout
+        end
+
+        expect(emitted).to have_key('journaled.worker.queue_total_count')
+      end
     end
   end