Fix in_bulk leaking thread local on exception

lib/gouda/bulk.rb

@@ -18,7 +18,8 @@ module Gouda
   #    end
   # @return [Object] the return value of the block
   def self.in_bulk(&blk)
-    if Thread.current[:gouda_bulk_buffer].nil?
+    outermost = Thread.current[:gouda_bulk_buffer].nil?
+    if outermost
       Thread.current[:gouda_bulk_buffer] = []
       retval = yield
       buf, Thread.current[:gouda_bulk_buffer] = Thread.current[:gouda_bulk_buffer], nil
@@ -27,6 +28,8 @@ def self.in_bulk(&blk)
     else # There already is an open bulk
       yield
     end
+  ensure
+    Thread.current[:gouda_bulk_buffer] = nil if outermost
   end
 
   # This method exists in edge Rails so probably can be replaced later:

test/gouda/concurrency_extension_test.rb

@@ -4,13 +4,15 @@
 
 class GoudaConcurrencyExtensionTest < ActiveSupport::TestCase
   include AssertHelper
+
   class TestJobWithoutConcurrency < ActiveJob::Base
     self.queue_adapter = Gouda::Adapter.new
   end
 
   class TestJobWithPerformConcurrency < ActiveJob::Base
     self.queue_adapter = Gouda::Adapter.new
     include Gouda::ActiveJobExtensions::Concurrency
+
     gouda_control_concurrency_with(perform_limit: 1)
 
     def perform(*args)
@@ -42,6 +44,7 @@ def perform(*args)
   class TestJobWithCommonConcurrency < ActiveJob::Base
     self.queue_adapter = Gouda::Adapter.new
     include Gouda::ActiveJobExtensions::Concurrency
+
     gouda_control_concurrency_with(total_limit: 1)
 
     def perform(*args)
@@ -79,6 +82,7 @@ def perform(*args)
   class TestJobWithEnqueueConcurrency < ActiveJob::Base
     self.queue_adapter = Gouda::Adapter.new
     include Gouda::ActiveJobExtensions::Concurrency
+
     gouda_control_concurrency_with(enqueue_limit: 1)
 
     def perform(*args)
@@ -105,6 +109,7 @@ def perform(*args)
   class TestJobWithCustomKey < ActiveJob::Base
     self.queue_adapter = Gouda::Adapter.new
     include Gouda::ActiveJobExtensions::Concurrency
+
     gouda_control_concurrency_with total_limit: 1, key: "42"
   end
 
@@ -117,6 +122,7 @@ class TestJobWithCustomKey < ActiveJob::Base
   class TestJobWithCustomKeyProc < ActiveJob::Base
     self.queue_adapter = Gouda::Adapter.new
     include Gouda::ActiveJobExtensions::Concurrency
+
     gouda_control_concurrency_with total_limit: 1, key: -> { @ivar }
 
     def initialize(...)

test/gouda/gouda_test.rb

@@ -58,6 +58,7 @@ def enqueue_concurrency_key
 
   class JobWithEnqueueConcurrencyViaGoudaAndEnqueueLimit < GoudaTestJob
     include Gouda::ActiveJobExtensions::Concurrency
+
     gouda_control_concurrency_with(enqueue_limit: 1, key: -> { self.class.to_s })
     def perform
       "perform result of #{self.class}"
@@ -66,6 +67,7 @@ def perform
 
   class JobWithEnqueueConcurrencyViaGoudaAndTotalLimit < GoudaTestJob
     include Gouda::ActiveJobExtensions::Concurrency
+
     gouda_control_concurrency_with(total_limit: 1, key: -> { self.class.to_s })
     def perform
       "perform result of #{self.class}"
@@ -84,6 +86,7 @@ def execution_concurrency_key
 
   class JobWithExecutionConcurrencyViaGoudaAndTotalLimit < GoudaTestJob
     include Gouda::ActiveJobExtensions::Concurrency
+
     gouda_control_concurrency_with(total_limit: 1, key: -> { self.class.to_s })
     def perform
       "perform result of #{self.class}"
@@ -92,6 +95,7 @@ def perform
 
   class JobWithExecutionConcurrencyViaGoudaAndPerformLimit < GoudaTestJob
     include Gouda::ActiveJobExtensions::Concurrency
+
     gouda_control_concurrency_with(perform_limit: 1, key: -> { self.class.to_s })
     def perform
       "perform result of #{self.class}"
@@ -384,6 +388,29 @@ def perform
     end
   end
 
+  test "resets the bulk buffer thread local when the block raises" do
+    assert_raises(RuntimeError) do
+      Gouda.in_bulk do
+        raise "boom"
+      end
+    end
+    assert_nil Thread.current[:gouda_bulk_buffer], "The bulk buffer thread local must be reset after an exception"
+  end
+
+  test "enqueues jobs normally after a failed in_bulk block" do
+    assert_raises(RuntimeError) do
+      Gouda.in_bulk do
+        StandardJob.perform_later
+        raise "boom"
+      end
+    end
+
+    # Jobs enqueued after the failed bulk block should still be enqueued normally
+    assert_changes_by(-> { Gouda::Workload.count }, exactly: 1) do
+      StandardJob.perform_later
+    end
+  end
+
   test "sets the correct queue with perform_later" do
     assert_changes_by(-> { Gouda::Workload.count }, exactly: 1) do
       HeavyJob.perform_later

test/gouda/scheduler_test.rb

@@ -19,6 +19,7 @@ def perform(regular = "ok", mandatory:, optional: "hidden")
 
   class FailingJob < ActiveJob::Base
     include Gouda::ActiveJobExtensions::Concurrency
+
     self.queue_adapter = Gouda::Adapter.new
 
     class MegaError < StandardError