Reload process to use DB-assigned updated_at timestamps

When a process is saved, Sequel's timestamps plugin sets updated_at
in-memory before the DB transaction commits. If the runner is called
after the transaction commits (via after_commit callback), it receives
a stale in-memory timestamp that may differ from the DB-assigned value.

ProcessesSync compares process.updated_at against the LRP annotation
to detect drift. A stale timestamp causes unnecessary re-sync requests.

This commit adds process.reload before passing the process to runners
in ProcessObserver, ProcessRouteHandler, and AppUpdate. This ensures
the runner receives the DB-assigned timestamp.

Also removes an incorrect comment in ProcessRestart about timestamp
precision differences - testing confirmed the timestamps match when
the runner is called inside the transaction.

Adds skip_process_observer_on_update in AppUpdate to prevent redundant
ProcessObserver.updated calls when metric tags are being updated.

Author: philippthun

app/actions/app_update.rb

@@ -44,12 +44,20 @@ def update(app, message, lifecycle)
         )
 
         # update process timestamp to trigger convergence if sending fails
-        app.processes.each(&:save) if updating_metric_tags?(message)
+        if updating_metric_tags?(message)
+          app.processes.each do |process|
+            process.skip_process_observer_on_update = true
+            process.save
+          end
+        end
       end
 
       if updating_metric_tags?(message)
         app.processes.each do |process|
-          @runners.runner_for_process(process).update_metric_tags if process.state == ProcessModel::STARTED
+          # Reload to get the DB-assigned updated_at, which ProcessesSync compares against
+          # the LRP annotation to detect drift. Without this, the stale in-memory value
+          # causes an unnecessary re-sync.
+          @runners.runner_for_process(process.reload).update_metric_tags if process.state == ProcessModel::STARTED
         rescue Diego::Runner::CannotCommunicateWithDiegoError => e
           @logger.error("failed communicating with diego backend: #{e.message}")
         end

app/actions/process_restart.rb

@@ -9,11 +9,6 @@ def restart(process:, config:, stop_in_runtime:, revision: nil)
           process.lock!
           process.skip_process_observer_on_update = true
 
-          # A side-effect of submitting LRP requests before the transaction commits is that
-          # the annotation timestamp stored on the LRP is slightly different than the process.updated_at
-          # timestamp that is stored in the DB due to timestamp precision differences.
-          # This difference causes the sync job to submit an extra update LRP request which updates
-          # the LRP annotation at a later time. This should have no impact on the running LRP.
           if need_to_stop_in_runtime
             process.update(state: ProcessModel::STOPPED)
             runners(config).runner_for_process(process).stop

lib/cloud_controller/process_observer.rb

@@ -37,11 +37,15 @@ def react_to_state_change(process)
 
         process.update(revision: process.app.latest_revision) if process.revisions_enabled?
 
-        @runners.runner_for_process(process).start unless process.needs_staging?
+        # Reload to get the DB-assigned updated_at, which ProcessesSync compares against
+        # the LRP annotation to detect drift. Without this, the stale in-memory value
+        # causes an unnecessary re-sync.
+        @runners.runner_for_process(process.reload).start unless process.needs_staging?
       end
 
       def react_to_instances_change(process)
-        @runners.runner_for_process(process).scale if process.started? && process.active?
+        # Same as above: reload to get the DB-assigned updated_at before building the LRP annotation.
+        @runners.runner_for_process(process.reload).scale if process.started? && process.active?
       end
 
       def with_diego_communication_handling

lib/cloud_controller/process_route_handler.rb

@@ -28,7 +28,10 @@ def update_route_information(perform_validation: true, updated_ports: false)
     end
 
     def notify_backend_of_route_update
-      @runners.runner_for_process(@process).update_routes if @process && @process.staged? && @process.started?
+      # Reload to get the DB-assigned updated_at, which ProcessesSync compares against
+      # the LRP annotation to detect drift. Without this, the stale in-memory value
+      # causes an unnecessary re-sync.
+      @runners.runner_for_process(@process.reload).update_routes if @process && @process.staged? && @process.started?
     rescue Diego::Runner::CannotCommunicateWithDiegoError => e
       logger.error("failed communicating with diego backend: #{e.message}")
     end

spec/unit/actions/app_update_spec.rb

@@ -103,6 +103,25 @@ module VCAP::CloudController
             expect(runner).to have_received(:update_metric_tags).twice
           end
 
+          it 'does not trigger ProcessObserver when saving processes', isolation: :truncation do
+            expect(ProcessObserver).not_to receive(:updated)
+            app_update.update(app_model, message, lifecycle)
+          end
+
+          it 'passes a process with the correct updated_at timestamp to the runner', isolation: :truncation do
+            received = {}
+            allow(runners).to receive(:runner_for_process) do |p|
+              received[p.guid] = p.updated_at
+              runner
+            end
+
+            app_update.update(app_model, message, lifecycle)
+
+            expect(runners).to have_received(:runner_for_process).twice
+            expect(received[web_process.guid]).to eq(web_process.reload.updated_at)
+            expect(received[worker_process.guid]).to eq(worker_process.reload.updated_at)
+          end
+
           context 'when there is a CannotCommunicateWithDiegoError' do
             before do
               allow(runner).to receive(:update_metric_tags).and_invoke(
@@ -130,10 +149,12 @@ module VCAP::CloudController
               let!(:worker_process) { instance_double(VCAP::CloudController::ProcessModel) }
 
               before do
+                allow(web_process).to receive(:skip_process_observer_on_update=)
                 allow(web_process).to receive(:save)
-                allow(web_process).to receive(:state).and_return(ProcessModel::STARTED)
+                allow(web_process).to receive_messages(reload: web_process, state: ProcessModel::STARTED)
+                allow(worker_process).to receive(:skip_process_observer_on_update=)
                 allow(worker_process).to receive(:save)
-                allow(worker_process).to receive(:state).and_return(ProcessModel::STARTED)
+                allow(worker_process).to receive_messages(reload: worker_process, state: ProcessModel::STARTED)
                 allow(app_model).to receive(:processes).and_return([web_process, worker_process])
               end
 
@@ -180,10 +201,12 @@ module VCAP::CloudController
               let!(:worker_process) { instance_double(VCAP::CloudController::ProcessModel) }
 
               before do
+                allow(web_process).to receive(:skip_process_observer_on_update=)
                 allow(web_process).to receive(:save)
-                allow(web_process).to receive(:state).and_return(ProcessModel::STARTED)
+                allow(web_process).to receive_messages(reload: web_process, state: ProcessModel::STARTED)
+                allow(worker_process).to receive(:skip_process_observer_on_update=)
                 allow(worker_process).to receive(:save)
-                allow(worker_process).to receive(:state).and_return(ProcessModel::STARTED)
+                allow(worker_process).to receive_messages(reload: worker_process, state: ProcessModel::STARTED)
                 allow(app_model).to receive(:processes).and_return([web_process, worker_process])
               end
 

spec/unit/actions/process_restart_spec.rb

@@ -36,6 +36,19 @@ module VCAP::CloudController
         ProcessRestart.restart(process: process, config: config, stop_in_runtime: true)
       end
 
+      it 'passes a process with the correct updated_at timestamp to the runner', isolation: :truncation do
+        received = nil
+        allow(VCAP::CloudController::Diego::Runner).to receive(:new) do |p, _config|
+          received = p.updated_at
+          runner
+        end
+
+        ProcessRestart.restart(process: process, config: config, stop_in_runtime: false)
+
+        expect(VCAP::CloudController::Diego::Runner).to have_received(:new).once
+        expect(received).to eq(process.reload.updated_at)
+      end
+
       context 'when the process is STARTED' do
         it 'keeps process state as STARTED' do
           ProcessRestart.restart(process: process, config: config, stop_in_runtime: true)

spec/unit/lib/cloud_controller/process_observer_spec.rb

@@ -5,7 +5,7 @@ module VCAP::CloudController
     let(:stagers) { double(:stagers, stager_for_build: stager) }
     let(:runners) { instance_double(Runners, runner_for_process: runner) }
     let(:stager) { double(:stager) }
-    let(:runner) { instance_double(Diego::Runner, stop: nil, start: nil) }
+    let(:runner) { instance_double(Diego::Runner, stop: nil, start: nil, scale: nil) }
     let(:process_active) { true }
     let(:diego) { false }
     let(:process) do
@@ -33,6 +33,7 @@ module VCAP::CloudController
 
     before do
       ProcessObserver.configure(stagers, runners)
+      allow(process).to receive(:reload).and_return(process)
     end
 
     describe '.deleted' do
@@ -312,6 +313,37 @@ module VCAP::CloudController
           end
         end
       end
+
+      context 'updated_at annotation accuracy' do
+        let(:received) { [] }
+
+        before do
+          allow(runners).to receive(:runner_for_process) do |p|
+            received << p.updated_at
+            runner
+          end
+        end
+
+        context 'when the process instances have changed' do
+          it 'passes a process with the correct updated_at timestamp to the runner', isolation: :truncation do
+            process_model = ProcessModelFactory.make(state: 'STARTED')
+            process_model.update(instances: process_model.instances + 1)
+
+            expect(runners).to have_received(:runner_for_process).once
+            expect(received.last).to eq(process_model.reload.updated_at)
+          end
+        end
+
+        context 'when the process state has changed' do
+          it 'passes a process with the correct updated_at timestamp to the runner', isolation: :truncation do
+            process_model = ProcessModelFactory.make(state: 'STOPPED')
+            process_model.update(state: ProcessModel::STARTED)
+
+            expect(runners).to have_received(:runner_for_process).once
+            expect(received.last).to eq(process_model.reload.updated_at)
+          end
+        end
+      end
     end
   end
 end

spec/unit/lib/cloud_controller/process_route_handler_spec.rb

@@ -196,5 +196,22 @@ module VCAP::CloudController
         end
       end
     end
+
+    describe 'updated_at annotation accuracy' do
+      let(:process) { ProcessModelFactory.make(state: 'STARTED') }
+
+      it 'passes a process with the correct updated_at timestamp to the runner', isolation: :truncation do
+        received = nil
+        allow(runners).to receive(:runner_for_process) do |p|
+          received = p.updated_at
+          runner
+        end
+
+        handler.update_route_information
+
+        expect(runners).to have_received(:runner_for_process).once
+        expect(received).to eq(process.reload.updated_at)
+      end
+    end
   end
 end

spec/unit/models/runtime/process_model_spec.rb

@@ -1495,7 +1495,7 @@ def act_as_cf_admin
     end
 
     describe 'saving' do
-      it 'calls AppObserver.updated', isolation: :truncation do
+      it 'calls ProcessObserver.updated', isolation: :truncation do
         process = ProcessModelFactory.make
         expect(ProcessObserver).to receive(:updated).with(process)
         process.update(instances: process.instances + 1)