Background worker coordination by the queue processor.

apps/workers/operation_runner.rb

@@ -43,9 +43,7 @@ def initialize(host)
     # Extra arguments depend on the specific implementation of the runner.
     # Check particular classes for more information.
     def perform(operation, *args)
-      runner_for(operation).perform(*args).tap do |result|
-        update_next_run if result.success?
-      end
+      runner_for(operation).perform(*args)
     end
 
     private
@@ -68,10 +66,8 @@ def roomorama_client
     end
 
     # if the result of performing the operation was successful, we update
-    # the timestamp when the next synchronisation for the given host
-    # should happen.
+    # the timestamp when the next execution of the worker should happen.
     def update_next_run
-      return true
       one_day = 24 * 60 * 60 # TODO make this a dynamic value
       host.next_run_at = Time.now + one_day
 

apps/workers/processor.rb

@@ -69,13 +69,15 @@ def process!
     def run_worker(args)
       worker = BackgroundWorkerRepository.find(args[:background_worker_id])
 
-      timing_out(worker.type, args) do
-        host      = HostRepository.find(worker.host_id)
-        supplier  = SupplierRepository.find(host.supplier_id)
-        broadcast = [worker.type, ".", supplier.name].join
-
-        Concierge::Announcer.trigger(broadcast, host)
-        Result.new(true)
+      running(worker) do
+        timing_out(worker.type, args) do
+          host      = HostRepository.find(worker.host_id)
+          supplier  = SupplierRepository.find(host.supplier_id)
+          broadcast = [worker.type, ".", supplier.name].join
+
+          Concierge::Announcer.trigger(broadcast, host)
+          Result.new(true)
+        end
       end
     end
 
@@ -91,10 +93,37 @@ def timing_out(operation, params)
       Result.error(:timeout)
     end
 
+    # coordinates the +BackgroundWorker+ instance status and timestamps by changing
+    # the worker status to +running+, yielding the block (which is supposed to do
+    # the worker's job), and ensuring that the worker's status is set back to +idle+
+    # at the end of the process, as well as properly updating the +next_run_at+ column
+    # according to the specified worker +interval+.
+    def running(worker)
+      worker_started(worker)
+      yield
+
+    ensure
+      # reload the worker instance to make sure to account for any possible
+      # changes in the process
+      worker_completed(BackgroundWorkerRepository.find(worker.id))
+    end
+
     def message
       @message = json_decode(payload)
     end
 
+    def worker_started(worker)
+      worker.status = "running"
+      BackgroundWorkerRepository.update(worker)
+    end
+
+    def worker_completed(worker)
+      worker.status      = "idle"
+      worker.next_run_at = Time.now + worker.interval
+
+      BackgroundWorkerRepository.update(worker)
+    end
+
     # NOTE this time out should be shorter than the +VisibilityTimeout+ configured
     # on the SQS queue to be used by Concierge.
     def processing_timeout

spec/workers/operation_runner_spec.rb

@@ -32,18 +32,6 @@
       subject.perform(operation)
     end
 
-    xit "updates host next synchrnonisation time when successful" do
-      allow(subject).to receive(:runner_for) { double(perform: Result.new(true)) }
-      expect(host.next_run_at).to be_nil
-
-      identifiers = ["prop1"]
-      operation   = Roomorama::Client::Operations.disable(identifiers)
-      subject.perform(operation)
-
-      updated = HostRepository.find(host.id)
-      expect(updated.next_run_at > Time.now).to eq true
-    end
-
     it "raises an error if the operation is not recognised" do
       operation = nil
       expect {

spec/workers/processor_spec.rb

@@ -29,21 +29,27 @@
     it "triggers the associated supplier synchronisation mechanism on background worker operations" do
       invoked = false
 
+      supplier = create_supplier(name: "AcmeTest")
+      host     = create_host(username: "acme-host", supplier_id: supplier.id)
+      worker   = create_background_worker(host_id: host.id, type: "metadata", interval: 10)
+      payload[:data][:background_worker_id] = worker.id
+
       Concierge::Announcer.on("metadata.AcmeTest") do |host|
         expect(host.username).to eq "acme-host"
         expect(SupplierRepository.find(host.supplier_id).name).to eq "AcmeTest"
         invoked = true
-      end
 
-      supplier = create_supplier(name: "AcmeTest")
-      host     = create_host(username: "acme-host", supplier_id: supplier.id)
-      worker   = create_background_worker(host_id: host.id, type: "metadata")
-      payload[:data][:background_worker_id] = worker.id
+        expect(BackgroundWorkerRepository.find(worker.id).status).to eq "running"
+      end
 
       result = subject.process!
       expect(result).to be_a Result
       expect(result).to be_success
       expect(invoked).to eq true
+
+      reloaded_worker = BackgroundWorkerRepository.find(worker.id)
+      expect(reloaded_worker.next_run_at - Time.now).to be_within(1).of(worker.interval)
+      expect(reloaded_worker.status).to eq "idle"
     end
 
     it "times out and fails if the operation takes too long" do