#218 - callbacks manager concurrency fix (#219)

CHANGELOG.md

@@ -1,5 +1,9 @@
 # WaterDrop changelog
 
+## 2.0.6 (2021-12-01)
+- #218 - Fixes a case, where dispatch of callbacks the same moment a new producer was created could cause a concurrency issue in the manager.
+- Fix some unstable specs.
+
 ## 2.0.5 (2021-11-28)
 
 ### Bug fixes

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    waterdrop (2.0.5)
+    waterdrop (2.0.6)
       concurrent-ruby (>= 1.1)
       dry-configurable (~> 0.13)
       dry-monitor (~> 0.5)

lib/water_drop/instrumentation/callbacks_manager.rb

@@ -13,8 +13,12 @@ def initialize
       # Invokes all the callbacks registered one after another
       #
       # @param args [Object] any args that should go to the callbacks
+      # @note We do not use `#each_value` here on purpose. With it being used, we cannot dispatch
+      #   callbacks and add new at the same time. Since we don't know when and in what thread
+      #   things are going to be added to the manager, we need to extract values into an array and
+      #   run it. That way we can add new things the same time.
       def call(*args)
-        @callbacks.each_value { |a| a.call(*args) }
+        @callbacks.values.each { |callback| callback.call(*args) }
       end
 
       # Adds a callback to the manager

lib/water_drop/version.rb

@@ -3,5 +3,5 @@
 # WaterDrop library
 module WaterDrop
   # Current WaterDrop version
-  VERSION = '2.0.5'
+  VERSION = '2.0.6'
 end

spec/lib/water_drop/instrumentation/callbacks_manager_spec.rb

@@ -4,6 +4,7 @@
   subject(:manager) { described_class.new }
 
   let(:id) { SecureRandom.uuid }
+  let(:changed) { [] }
 
   describe '#call' do
     context 'when there are no callbacks added' do
@@ -28,18 +29,44 @@
   end
 
   describe '#add' do
-    let(:changed) { [] }
-
     it 'expect after adding to be used' do
       manager.add(id, -> { changed << true })
       manager.call
       expect(changed).to eq([true])
     end
+
+    context 'when we are adding a callback but at the same time, we call callbacks' do
+      let(:added_id) { SecureRandom.uuid }
+      let(:callable) do
+        lambda do
+          changed << true
+          sleep(10)
+        end
+      end
+
+      before do
+        # This will simulate a long running callback when manager is called, so when we add new one
+        # The previous one is still running in a thread
+        manager.add(id, callable)
+        Thread.new { manager.call }
+        # This makes sure, that we wait until the thread kicks in
+        sleep(0.001) while changed.empty?
+      end
+
+      it { expect { manager.add(added_id, callable) }.not_to raise_error }
+
+      it 'expect to register the new callback' do
+        manager.delete(id)
+        manager.add(added_id, -> { changed << true })
+
+        manager.call
+
+        expect(changed).to eq([true, true])
+      end
+    end
   end
 
   describe '#delete' do
-    let(:changed) { [] }
-
     before { manager.add(id, -> { changed << true }) }
 
     it 'expect after removal not to be used' do

spec/lib/water_drop/producer_spec.rb

@@ -188,13 +188,15 @@
 
         producer.produce_sync(message)
 
-        sleep(0.001) while events.size < 2
+        sleep(0.001) while events.size < 3
       end
 
       it { expect(events.last.id).to eq('statistics.emitted') }
       it { expect(events.last[:producer_id]).to eq(producer.id) }
-      it { expect(events.last[:statistics]['msg_cnt']).to eq(1) }
-      it { expect(events.last[:statistics]['msg_cnt_d']).to eq(0) }
+      it { expect(events.last[:statistics]['ts']).to be > 0 }
+      # This is in microseconds. We needed a stable value for comparison, and the distance in
+      # between statistics events should always be within 1ms
+      it { expect(events.last[:statistics]['ts_d']).to be_between(90_000, 200_000) }
     end
 
     context 'when we have more producers' do