diff --git a/lib/gouda/bulk.rb b/lib/gouda/bulk.rb index e5973d2..638f7c9 100644 --- a/lib/gouda/bulk.rb +++ b/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: diff --git a/test/gouda/concurrency_extension_test.rb b/test/gouda/concurrency_extension_test.rb index 528ec5d..91119a9 100644 --- a/test/gouda/concurrency_extension_test.rb +++ b/test/gouda/concurrency_extension_test.rb @@ -4,6 +4,7 @@ class GoudaConcurrencyExtensionTest < ActiveSupport::TestCase include AssertHelper + class TestJobWithoutConcurrency < ActiveJob::Base self.queue_adapter = Gouda::Adapter.new end @@ -11,6 +12,7 @@ class TestJobWithoutConcurrency < ActiveJob::Base 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(...) diff --git a/test/gouda/gouda_test.rb b/test/gouda/gouda_test.rb index 561773a..c5513af 100644 --- a/test/gouda/gouda_test.rb +++ b/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 diff --git a/test/gouda/scheduler_test.rb b/test/gouda/scheduler_test.rb index 937145b..cb9c8c1 100644 --- a/test/gouda/scheduler_test.rb +++ b/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