respect the state update counter if present

Author: svenfuchs

lib/travis/hub/service/state_update.rb

@@ -0,0 +1,84 @@
+module Travis
+  module Hub
+    module Service
+      class StateUpdate < Struct.new(:data)
+        class Counter < Struct.new(:job_id, :redis)
+          TTL = 3600 * 12
+
+          def count
+            @count ||= redis.get(key).to_i
+          end
+
+          def increment
+            count = redis.incr(key)
+            redis.expire(key, TTL)
+            count
+          end
+
+          private
+
+            def key
+              "job:state_update_count:#{job_id}"
+            end
+        end
+
+        include Helper::Context
+
+        MSGS = {
+          missing:   'Received state update with no count for job id=%p, last known count: %p.',
+          ordered:   'Received state update %p for job id=%p, last known count: %p',
+          unordered: 'Received state update %p for job id=%p, last known count: %p. %s',
+          skip:      'Skipping the message.'
+        }
+
+        def apply?
+          return missing unless given?
+          apply = ordered? ? ordered : unordered
+          return true unless ENV['UPDATE_COUNT']
+          apply
+        end
+
+        private
+
+          def given?
+            !count.nil?
+          end
+
+          def missing
+            warn :missing, job_id, counter.count
+            true
+          end
+
+          def ordered
+            info :ordered, count, job_id, counter.count
+            true
+          end
+
+          def unordered
+            warn :unordered, count, job_id, counter.count, ENV['UPDATE_COUNT'] ? MSGS[:skip] : ''
+            false
+          end
+
+          def ordered?
+            count >= counter.count
+          end
+
+          def counter
+            @counter ||= Counter.new(job_id, redis)
+          end
+
+          def job_id
+            data[:id]
+          end
+
+          def count
+            meta[:state_update_count]
+          end
+
+          def meta
+            data[:meta] || {}
+          end
+      end
+    end
+  end
+end

lib/travis/hub/service/update_job.rb

@@ -3,6 +3,7 @@
 require 'travis/hub/helper/locking'
 require 'travis/hub/model/job'
 require 'travis/hub/service/error_job'
+require 'travis/hub/service/state_update'
 require 'travis/hub/service/notify_workers'
 require 'travis/hub/helper/limit'
 
@@ -16,7 +17,7 @@ class UpdateJob < Struct.new(:event, :data)
         EVENTS = [:receive, :reset, :start, :finish, :cancel, :restart]
 
         MSGS = {
-          skipped: 'Skipped event job:%s for <Job id=%s> trying to update state from %s to %s data=%s',
+          skipped: 'Skipped event job:%s for <Job id=%s> trying to update state from %p to %p data=%s',
         }
 
         def run
@@ -37,6 +38,7 @@ def job
         private
 
           def update_job
+            return skipped unless apply_state_update?
             return error_job if event == :reset && resets.limited? && !job.finished?
             return recancel if recancel?
             return skipped if skip_canceled?
@@ -72,6 +74,10 @@ def skipped
             warn :skipped, event, job.id, job.state, data[:state], data
           end
 
+          def apply_state_update?
+            StateUpdate.new(context, data).apply?
+          end
+
           def resets
             @resets ||= Limit.new(redis, :resets, job.id, config.limit.resets)
           end

spec/spec_helper.rb

@@ -75,6 +75,7 @@
     Travis::Event.instance_variable_set(:@subscriptions, nil)
     # Travis::Addons.setup({ host: 'host.com', encryption: { key: 'secret' * 10 } }, logger)
     Time.stubs(:now).returns(NOW)
+    context.redis.flushall
   end
 end
 

spec/support/context.rb

@@ -6,6 +6,7 @@ module Context
 
     included do
       let(:stdout)  { StringIO.new }
+      let(:log)     { stdout.string }
       let(:logger)  { Travis::Logger.new(stdout) }
       let(:context) { Travis::Hub::Context.new(logger: logger) }
       before        { Travis::Hub.context = context }

spec/travis/hub/service/update_job_spec.rb

@@ -210,4 +210,68 @@ def recieve(msg)
       expect(job.reload.state).to eql :passed
     end
   end
+
+  describe 'state update count' do
+    let(:job)   { FactoryGirl.create(:job, state: :received) }
+    let(:event) { :start }
+    let(:data)  { { id: job.id, state: :started, meta: meta } }
+
+    before { ENV['UPDATE_COUNT'] = 'true' }
+    after  { ENV['UPDATE_COUNT'] = nil }
+
+    describe 'with no count stored' do
+      describe 'given no meta' do
+        let(:meta) { nil }
+        before { subject.run }
+
+        it { expect(job.reload.state).to eq :started }
+        it { expect(log).to include "W Received state update with no count for job id=#{job.id}, last known count: 0" }
+      end
+
+      describe 'given no count' do
+        let(:meta) { {} }
+        before { subject.run }
+
+        it { expect(job.reload.state).to eq :started }
+        it { expect(log).to include "W Received state update with no count for job id=#{job.id}, last known count: 0" }
+      end
+
+      describe 'given a count' do
+        let(:meta) { { state_update_count: 2 } }
+        before { subject.run }
+
+        it { expect(job.reload.state).to eq :started }
+        it { expect(log).to include "I Received state update 2 for job id=#{job.id}, last known count: 0" }
+      end
+    end
+
+    describe 'with a count stored' do
+      before { context.redis.set("job:state_update_count:#{job.id}", 3) }
+
+      describe 'given no meta it skips the message' do
+        let(:meta) { nil }
+        before { subject.run }
+
+        it { expect(job.reload.state).to eq :started }
+        it { expect(log).to include "W Received state update with no count for job id=#{job.id}, last known count: 3" }
+      end
+
+      describe 'given no count it skips the message' do
+        let(:meta) { {} }
+        before { subject.run }
+
+        it { expect(job.reload.state).to eq :started }
+        it { expect(log).to include "W Received state update with no count for job id=#{job.id}, last known count: 3" }
+      end
+
+      describe 'given a count it skips the message' do
+        let(:meta) { { state_update_count: 2 } }
+        before { subject.run }
+
+        it { expect(job.reload.state).to eq :received }
+        it { expect(log).to include "W Received state update 2 for job id=#{job.id}, last known count: 3. Skipping the message." }
+        it { expect(log).to include "W Skipped event job:start for <Job id=#{job.id}> trying to update state from :received to :started" }
+      end
+    end
+  end
 end