Fix race condition when sending notifications (#159)

The race condition led to two (or more) notifications being sent 

Co-authored-by: Tema Bolshakov <either.free@gmail.com>

.rubocop.yml

@@ -12,3 +12,7 @@ Lint/MissingSuper:
 Lint/AmbiguousBlockAssociation:
   Exclude:
     - 'spec/**/*'
+
+Metrics/ClassLength:
+  Enabled: true
+  Max: 110

Gemfile.lock

@@ -2,6 +2,7 @@ PATH
   remote: .
   specs:
     stoplight (3.0.0)
+      redlock (~> 1.0)
 
 GEM
   remote: https://rubygems.org/
@@ -17,6 +18,8 @@ GEM
       multipart-post (>= 1.2, < 3)
     honeybadger (2.7.2)
     json (2.3.1)
+    mock_redis (0.34.0)
+      ruby2_keywords
     multipart-post (2.1.1)
     pagerduty (2.1.2)
       json (>= 1.7.7)
@@ -26,6 +29,8 @@ GEM
     rainbow (3.0.0)
     rake (13.0.1)
     redis (4.8.0)
+    redlock (1.3.2)
+      redis (>= 3.0.0, < 6.0)
     regexp_parser (1.8.2)
     rexml (3.2.5)
     rspec (3.11.0)
@@ -53,6 +58,7 @@ GEM
     rubocop-ast (1.1.1)
       parser (>= 2.7.1.5)
     ruby-progressbar (1.10.1)
+    ruby2_keywords (0.0.5)
     sentry-raven (1.2.3)
       faraday (>= 0.7.6, < 0.10.x)
     simplecov (0.21.2)
@@ -74,6 +80,7 @@ DEPENDENCIES
   bugsnag (~> 4.0)
   fakeredis (~> 0.8)
   honeybadger (~> 2.5)
+  mock_redis (~> 0.3)
   pagerduty (~> 2.1.1)
   rake (~> 13.0)
   redis (~> 4.1)

lib/stoplight/data_store/base.rb

@@ -52,6 +52,15 @@ def set_state(_light, _state)
       def clear_state(_light)
         raise NotImplementedError
       end
+
+      # @param _light [Light]
+      # @param _from_color [String]
+      # @param _to_color [String]
+      # @yield _block
+      # @return [Void]
+      def with_notification_lock(_light, _from_color, _to_color, &_block)
+        raise NotImplementedError
+      end
     end
   end
 end

lib/stoplight/data_store/memory.rb

@@ -7,10 +7,12 @@ module DataStore
     # @see Base
     class Memory < Base
       include MonitorMixin
+      KEY_SEPARATOR = ':'
 
       def initialize
         @failures = Hash.new { |h, k| h[k] = [] }
         @states = Hash.new { |h, k| h[k] = State::UNLOCKED }
+        @last_notifications = {}
         super() # MonitorMixin
       end
 
@@ -49,6 +51,30 @@ def set_state(light, state)
       def clear_state(light)
         synchronize { @states.delete(light.name) }
       end
+
+      def with_notification_lock(light, from_color, to_color)
+        synchronize do
+          if last_notification(light) != [from_color, to_color]
+            set_last_notification(light, from_color, to_color)
+
+            yield
+          end
+        end
+      end
+
+      # @param light [Stoplight::Light]
+      # @return [Array, nil]
+      def last_notification(light)
+        @last_notifications[light.name]
+      end
+
+      # @param light [Stoplight::Light]
+      # @param from_color [String]
+      # @param to_color [String]
+      # @return [void]
+      def set_last_notification(light, from_color, to_color)
+        @last_notifications[light.name] = [from_color, to_color]
+      end
     end
   end
 end

lib/stoplight/data_store/redis.rb

@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require 'redlock'
+
 module Stoplight
   module DataStore
     # @see Base
@@ -8,8 +10,9 @@ class Redis < Base
       KEY_SEPARATOR = ':'
 
       # @param redis [::Redis]
-      def initialize(redis)
+      def initialize(redis, redlock: Redlock::Client.new([redis]))
         @redis = redis
+        @redlock = redlock
       end
 
       def names
@@ -76,8 +79,34 @@ def clear_state(light)
         normalize_state(state)
       end
 
+      LOCK_TTL = 2_000 # milliseconds
+
+      def with_notification_lock(light, from_color, to_color)
+        @redlock.lock(notification_lock_key(light), LOCK_TTL) do
+          if last_notification(light) != [from_color, to_color]
+            set_last_notification(light, from_color, to_color)
+
+            yield
+          end
+        end
+      end
+
       private
 
+      # @param light [Stoplight::Light]
+      # @return [Array, nil]
+      def last_notification(light)
+        @redis.get(last_notification_key(light))&.split('->')
+      end
+
+      # @param light [Stoplight::Light]
+      # @param from_color [String]
+      # @param to_color [String]
+      # @return [void]
+      def set_last_notification(light, from_color, to_color)
+        @redis.set(last_notification_key(light), [from_color, to_color].join('->'))
+      end
+
       def query_failures(light, transaction: @redis)
         transaction.lrange(failures_key(light), 0, -1)
       end
@@ -103,6 +132,14 @@ def failures_key(light)
         key('failures', light.name)
       end
 
+      def notification_lock_key(light)
+        key('notification_lock', light.name)
+      end
+
+      def last_notification_key(light)
+        key('last_notification', light.name)
+      end
+
       def states_key
         key('states')
       end

lib/stoplight/light/runnable.rb

@@ -30,11 +30,15 @@ def run
 
       def run_green
         on_failure = lambda do |size, error|
-          notify(Color::GREEN, Color::RED, error) if size == threshold
+          notify(Color::GREEN, Color::RED, error) if failures_threshold_breached?(size, threshold)
         end
         run_code(nil, on_failure)
       end
 
+      def failures_threshold_breached?(current_failures_count, max_errors_threshold)
+        current_failures_count == max_errors_threshold
+      end
+
       def run_yellow
         on_success = lambda do |failures|
           notify(Color::RED, Color::GREEN) unless failures.empty?
@@ -75,8 +79,10 @@ def failures_and_state
       end
 
       def notify(from_color, to_color, error = nil)
-        notifiers.each do |notifier|
-          safely { notifier.notify(self, from_color, to_color, error) }
+        data_store.with_notification_lock(self, from_color, to_color) do
+          notifiers.each do |notifier|
+            safely { notifier.notify(self, from_color, to_color, error) }
+          end
         end
       end
 

spec/stoplight/data_store/base_spec.rb

@@ -61,4 +61,11 @@
         .to raise_error(NotImplementedError)
     end
   end
+
+  describe '#with_notification_lock' do
+    it 'is not implemented' do
+      expect { data_store.with_notification_lock(nil, nil, nil) }
+        .to raise_error(NotImplementedError)
+    end
+  end
 end

spec/stoplight/data_store/memory_spec.rb

@@ -130,4 +130,30 @@
       expect(data_store.get_state(light)).to eql(Stoplight::State::UNLOCKED)
     end
   end
+
+  describe '#with_notification_lock' do
+    context 'when notification is already sent' do
+      before do
+        data_store.with_notification_lock(light, Stoplight::Color::GREEN, Stoplight::Color::RED) {}
+      end
+
+      it 'does not yield passed block' do
+        expect do |b|
+          data_store.with_notification_lock(light, Stoplight::Color::GREEN, Stoplight::Color::RED, &b)
+        end.not_to yield_control
+      end
+    end
+
+    context 'when notification is not already sent' do
+      before do
+        data_store.with_notification_lock(light, Stoplight::Color::GREEN, Stoplight::Color::RED) {}
+      end
+
+      it 'yields passed block' do
+        expect do |b|
+          data_store.with_notification_lock(light, Stoplight::Color::RED, Stoplight::Color::GREEN, &b)
+        end.to yield_control
+      end
+    end
+  end
 end

spec/stoplight/data_store/redis_spec.rb

@@ -1,17 +1,16 @@
 # frozen_string_literal: true
 
 require 'spec_helper'
-require 'fakeredis'
+require 'mock_redis'
 
 RSpec.describe Stoplight::DataStore::Redis do
-  let(:data_store) { described_class.new(redis) }
-  let(:redis) { Redis.new }
+  let(:data_store) { described_class.new(redis, redlock: redlock) }
+  let(:redis) { MockRedis.new }
+  let(:redlock) { instance_double(Redlock::Client) }
   let(:light) { Stoplight::Light.new(name) {} }
   let(:name) { ('a'..'z').to_a.shuffle.join }
   let(:failure) { Stoplight::Failure.new('class', 'message', Time.new) }
 
-  before { Redis::Connection::Memory.reset_all_databases }
-
   it 'is a class' do
     expect(described_class).to be_a(Class)
   end
@@ -143,4 +142,36 @@
       expect(data_store.get_state(light)).to eql(Stoplight::State::UNLOCKED)
     end
   end
+
+  describe '#with_notification_lock' do
+    let(:lock_key) { "stoplight:notification_lock:#{name}" }
+
+    before do
+      allow(redlock).to receive(:lock).with(lock_key, 2_000).and_yield
+    end
+
+    context 'when notification is already sent' do
+      before do
+        data_store.with_notification_lock(light, Stoplight::Color::GREEN, Stoplight::Color::RED) {}
+      end
+
+      it 'does not yield passed block' do
+        expect do |b|
+          data_store.with_notification_lock(light, Stoplight::Color::GREEN, Stoplight::Color::RED, &b)
+        end.not_to yield_control
+      end
+    end
+
+    context 'when notification is not already sent' do
+      before do
+        data_store.with_notification_lock(light, Stoplight::Color::GREEN, Stoplight::Color::RED) {}
+      end
+
+      it 'yields passed block' do
+        expect do |b|
+          data_store.with_notification_lock(light, Stoplight::Color::RED, Stoplight::Color::GREEN, &b)
+        end.to yield_control
+      end
+    end
+  end
 end

spec/stoplight/light/runnable_spec.rb

@@ -109,6 +109,38 @@ def random_string
           expect(subject.data_store.get_failures(subject).size).to eql(1)
         end
 
+        context 'when we did not send notifications yet' do
+          it 'notifies when transitioning to red' do
+            subject.threshold.times do
+              expect(io.string).to eql('')
+              begin
+                subject.run
+              rescue error.class
+                nil
+              end
+            end
+            expect(io.string).to_not eql('')
+          end
+        end
+
+        context 'when we already sent notifications' do
+          before do
+            subject.data_store.with_notification_lock(subject, Stoplight::Color::GREEN, Stoplight::Color::RED) {}
+          end
+
+          it 'does not send new notifications' do
+            subject.threshold.times do
+              expect(io.string).to eql('')
+              begin
+                subject.run
+              rescue error.class
+                nil
+              end
+            end
+            expect(io.string).to eql('')
+          end
+        end
+
         it 'notifies when transitioning to red' do
           subject.threshold.times do
             expect(io.string).to eql('')

stoplight.gemspec

@@ -31,10 +31,13 @@ Gem::Specification.new do |gem|
 
   gem.required_ruby_version = '>= 2.6'
 
+  gem.add_dependency 'redlock', "~> 1.0"
+
   gem.add_development_dependency('benchmark-ips', '~> 2.3')
   gem.add_development_dependency('bugsnag', '~> 4.0')
   gem.add_development_dependency('fakeredis', '~> 0.8')
   gem.add_development_dependency('honeybadger', '~> 2.5')
+  gem.add_development_dependency('mock_redis', '~> 0.3')
   gem.add_development_dependency('pagerduty', '~> 2.1.1')
   gem.add_development_dependency('rake', '~> 13.0')
   gem.add_development_dependency('redis', '~> 4.1')