Fail stuck service instance 'update' operations

When CCDB is briefly unavailable during a broker polling cycle, the CC
polling job can fail permanently (max_attempts=1) while the broker is
still processing. This leaves an update stuck: last_operation.state
stays 'in progress' with no delayed job working on it, requiring
operator intervention.

Add ServiceOperationsUpdateInProgressCleanup, a periodic job that
detects stuck 'update' operations whose polling job has permanently
failed and marks the operation and its pollable job as 'failed', giving
clients a definitive final state. A FOR UPDATE SKIP LOCKED guard
prevents double processing across concurrent CC instances.

Unlike the create cleanup, no orphan mitigation is triggered: an update
targets a resource that already exists, so it must not be deprovisioned.

Author: kathap

app/jobs/runtime/service_operations_update_in_progress_cleanup.rb

@@ -0,0 +1,108 @@
+module VCAP::CloudController
+  module Jobs
+    module Runtime
+      class ServiceOperationsUpdateInProgressCleanup < VCAP::CloudController::Jobs::CCJob
+        BATCH_SIZE = 10
+
+        def perform
+          logger.info("Cleaning up service 'update' operations stuck in 'in progress'")
+          cleanup_operations(ServiceInstanceOperation, ServiceInstance, :service_instance_id, 'service_instance.update')
+        end
+
+        def max_attempts
+          1
+        end
+
+        private
+
+        def cleanup_operations(operation_model, instance_model, foreign_key, jobs_operation)
+          # Find stuck service instance 'in progress' update operations where the broker is still working
+          # but CC's polling job has permanently failed due to a transient error (e.g. brief db connection flip).
+          # Join path: service_instance_operations → service_instances → jobs → delayed_jobs.
+          #
+          # Filters:
+          #   - service_instance_operations.state='in progress': the broker has not yet reported a final state
+          #     (succeeded or failed) that CC could successfully persist; if CC had received and saved a final
+          #     state from the broker, this column would already be 'succeeded' or 'failed' — not 'in progress'
+          #   - service_instance_operations.type='update': scope to update operations only
+          #   - service_instance_operations.created_at > CURRENT_TIMESTAMP - max_duration: operations beyond the max async polling window
+          #     are intentionally excluded — the broker has given up on them too, so they are out of scope for this cleanup
+          #   - jobs.state IN (POLLING, FAILED): the pollable job has not reached COMPLETE (a successful job
+          #     would already be done and is out of scope); POLLING covers the case where the failure hook
+          #     itself couldn't write FAILED due to the DB flip
+          #   - jobs.operation='service_instance.update': prevents matching create/delete jobs for the same
+          #     service instance that happen to share the same resource_guid
+          #   - delayed_jobs.failed_at IS NOT NULL: the delayed job permanently failed (exhausted max_attempts);
+          #     jobs still alive or locked have failed_at=NULL and must not be touched
+          #
+          # Unlike the create cleanup, no orphan mitigation is triggered: an update targets a resource that
+          # already exists, so we must not deprovision it. We only give the client a definitive final state
+          # by marking the operation and its pollable job as failed.
+          operation_table = operation_model.table_name
+          instance_table = instance_model.table_name
+
+          stuck = operation_model.
+                  join(instance_table, id: Sequel[operation_table][foreign_key]).
+                  join(:jobs, resource_guid: Sequel[instance_table][:guid]).
+                  join(:delayed_jobs, guid: Sequel[:jobs][:delayed_job_guid]).
+                  where(Sequel[operation_table][:state] => 'in progress').
+                  where(Sequel[operation_table][:type] => 'update').
+                  where(Sequel.lit("#{operation_table}.created_at > CURRENT_TIMESTAMP - INTERVAL '?' SECOND", default_maximum_duration_seconds.to_i)).
+                  where(Sequel[:jobs][:state] => [PollableJobModel::POLLING_STATE, PollableJobModel::FAILED_STATE]).
+                  where(Sequel[:jobs][:operation] => jobs_operation).
+                  exclude(Sequel[:delayed_jobs][:failed_at] => nil).
+                  select(
+                    Sequel[:jobs][:guid].as(:pollable_guid),
+                    Sequel[operation_table][:id].as(:op_id),
+                    Sequel[operation_table][foreign_key].as(:resource_id)
+                  ).
+                  order(Sequel[operation_table][:created_at]).
+                  limit(BATCH_SIZE)
+
+          stuck.each do |row|
+            resolve_stuck(operation_model, instance_model, row[:op_id], row[:resource_id], row[:pollable_guid])
+          end
+        end
+
+        def resolve_stuck(operation_model, instance_model, op_id, resource_id, pollable_guid)
+          # Mark the stuck update operation as failed and mark its pollable job as failed,
+          # giving the client a definitive final state. No broker-side changes are reverted.
+          operation_model.db.transaction do
+            operation = operation_model.where(id: op_id, state: 'in progress').for_update.skip_locked.first
+            return unless operation
+
+            instance = instance_model.first(id: resource_id)
+            return unless instance
+
+            instance_type = instance_model.to_s.split('::').last
+
+            logger.info(
+              "#{instance_type} #{instance.guid} update operation is stuck in 'in progress'. " \
+              "Setting operation's state to 'failed' and pollable job's state to 'FAILED'.",
+              instance_type: instance_type,
+              instance_guid: instance.guid,
+              operation_id: op_id,
+              pollable_job_guid: pollable_guid
+            )
+
+            operation.update(state: 'failed',
+                             description: "Operation was stuck in 'in progress' state. Set to 'failed' by cleanup job.")
+            PollableJobModel.where(guid: pollable_guid).update(state: PollableJobModel::FAILED_STATE)
+          end
+        end
+
+        def default_maximum_duration_seconds
+          Config.config.get(:broker_client_max_async_poll_duration_minutes).minutes
+        end
+
+        def logger
+          @logger ||= Steno.logger('cc.background.service-operations-update-in-progress-cleanup')
+        end
+
+        def job_name_in_configuration
+          :service_operations_update_in_progress_cleanup
+        end
+      end
+    end
+  end
+end

config/cloud_controller.yml

@@ -58,6 +58,9 @@ service_operations_initial_cleanup:
 service_operations_create_in_progress_cleanup:
   frequency_in_seconds: 3600 #1h
 
+service_operations_update_in_progress_cleanup:
+  frequency_in_seconds: 3600 #1h
+
 # One-off backfill - to be removed in a future version.
 lifecycle_type_backfill:
   frequency_in_seconds: 3600 #1h

lib/cloud_controller/clock/scheduler.rb

@@ -26,6 +26,7 @@ class Scheduler
       { name: 'failed_jobs', class: Jobs::Runtime::FailedJobsCleanup },
       { name: 'service_operations_initial_cleanup', class: Jobs::Runtime::ServiceOperationsInitialCleanup },
       { name: 'service_operations_create_in_progress_cleanup', class: Jobs::Runtime::ServiceOperationsCreateInProgressCleanup },
+      { name: 'service_operations_update_in_progress_cleanup', class: Jobs::Runtime::ServiceOperationsUpdateInProgressCleanup },
       # One-off backfill - to be removed in a future version.
       { name: 'lifecycle_type_backfill', class: Jobs::Runtime::LifecycleTypeBackfill }
     ].freeze

lib/cloud_controller/config_schemas/clock_schema.rb

@@ -37,6 +37,9 @@ class ClockSchema < VCAP::Config
           service_operations_create_in_progress_cleanup: {
             frequency_in_seconds: Integer
           },
+          service_operations_update_in_progress_cleanup: {
+            frequency_in_seconds: Integer
+          },
           # One-off backfill - to be removed in a future version.
           lifecycle_type_backfill: {
             frequency_in_seconds: Integer

lib/cloud_controller/jobs.rb

@@ -26,6 +26,7 @@
 require 'jobs/runtime/expired_orphaned_blob_cleanup'
 require 'jobs/runtime/expired_resource_cleanup'
 require 'jobs/runtime/service_operations_create_in_progress_cleanup'
+require 'jobs/runtime/service_operations_update_in_progress_cleanup'
 require 'jobs/runtime/failed_jobs_cleanup'
 require 'jobs/runtime/service_operations_initial_cleanup'
 require 'jobs/runtime/legacy_jobs'

spec/unit/jobs/runtime/service_operations_update_in_progress_cleanup_spec.rb

@@ -0,0 +1,188 @@
+require 'spec_helper'
+
+module VCAP::CloudController
+  module Jobs::Runtime
+    RSpec.describe ServiceOperationsUpdateInProgressCleanup, job_context: :worker do
+      subject(:job) { ServiceOperationsUpdateInProgressCleanup.new }
+
+      let(:fake_logger) { instance_double(Steno::Logger, info: nil, warn: nil) }
+      let(:max_poll_duration_minutes) { 60 }
+
+      before do
+        allow(Steno).to receive(:logger).and_return(fake_logger)
+        TestConfig.override(broker_client_max_async_poll_duration_minutes: max_poll_duration_minutes)
+      end
+
+      # Builds a fully stuck scenario for ServiceInstance update that the job should pick up and resolve.
+      # All filter conditions are satisfied: sio is in progress/update/within cutoff,
+      # pjob is FAILED with operation=service_instance.update, delayed_job has failed_at set.
+      # Override individual parameters to break a single filter and test exclusion.
+      def prepare_stuck_service_instance(
+        service_instance_state: 'in progress',
+        service_instance_type: 'update',
+        service_instance_created_at: Time.now,
+        pollable_job_state: PollableJobModel::FAILED_STATE,
+        pollable_job_operation: 'service_instance.update',
+        delayed_job_failed_at: Time.now
+      )
+        service_instance = create(:managed_service_instance)
+
+        create(:service_instance_operation,
+               service_instance_id: service_instance.id,
+               type: service_instance_type,
+               state: service_instance_state,
+               created_at: service_instance_created_at)
+
+        dj = Delayed::Job.create!(
+          guid: SecureRandom.uuid,
+          handler: 'fake',
+          run_at: Time.now,
+          failed_at: delayed_job_failed_at,
+          queue: 'cc-generic'
+        )
+
+        pjob = create(:pollable_job_model,
+                      state: pollable_job_state,
+                      operation: pollable_job_operation,
+                      resource_guid: service_instance.guid,
+                      resource_type: 'service_instances',
+                      delayed_job_guid: dj.guid)
+
+        { service_instance: service_instance, pjob: pjob, delayed_job: dj }
+      end
+
+      shared_examples 'does not resolve the operation' do
+        it 'leaves the operation in progress and the pollable job untouched' do
+          scenario = subject_scenario
+          job.perform
+          expect(scenario[:service_instance].last_operation.reload.state).to eq('in progress')
+          expect(scenario[:pjob].reload.state).to eq(scenario[:pjob].state)
+        end
+      end
+
+      it { is_expected.to be_a_valid_job }
+
+      describe '#perform' do
+        context 'when sio state is not in progress' do
+          it 'does not resolve when state is succeeded' do
+            scenario = prepare_stuck_service_instance(service_instance_state: 'succeeded')
+            job.perform
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('succeeded')
+          end
+
+          it 'does not resolve when state is failed' do
+            scenario = prepare_stuck_service_instance(service_instance_state: 'failed')
+            job.perform
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('failed')
+          end
+        end
+
+        context 'when sio type is not update' do
+          let(:subject_scenario) { prepare_stuck_service_instance(service_instance_type: 'create') }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when sio created_at is beyond the max polling window' do
+          let(:subject_scenario) { prepare_stuck_service_instance(service_instance_created_at: Time.now - (max_poll_duration_minutes + 1).minutes) }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when delayed_job.failed_at is nil (job still running or locked)' do
+          let(:subject_scenario) { prepare_stuck_service_instance(delayed_job_failed_at: nil) }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when pollable job state is COMPLETE' do
+          let(:subject_scenario) { prepare_stuck_service_instance(pollable_job_state: PollableJobModel::COMPLETE_STATE) }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when pollable job state is PROCESSING' do
+          let(:subject_scenario) { prepare_stuck_service_instance(pollable_job_state: PollableJobModel::PROCESSING_STATE) }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when pollable job operation is not service_instance.update' do
+          let(:subject_scenario) { prepare_stuck_service_instance(pollable_job_operation: 'service_instance.create') }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when a service instance update job is stuck with state FAILED' do
+          it 'sets operation to failed and pollable job to FAILED' do
+            scenario = prepare_stuck_service_instance
+            job.perform
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('failed')
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+          end
+        end
+
+        context 'when a service instance update job is stuck with state POLLING (DB flip before failure hook)' do
+          it 'sets operation to failed and pollable job to FAILED' do
+            scenario = prepare_stuck_service_instance(pollable_job_state: PollableJobModel::POLLING_STATE)
+            job.perform
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('failed')
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+          end
+        end
+
+        context 'when there are multiple stuck jobs within the batch size' do
+          it 'resolves each one' do
+            3.times { prepare_stuck_service_instance }
+            job.perform
+            expect(ServiceInstanceOperation.where(state: 'failed').count).to eq(3)
+          end
+        end
+
+        context 'when there are more stuck jobs than the batch size' do
+          it 'processes only up to BATCH_SIZE jobs per run' do
+            (ServiceOperationsUpdateInProgressCleanup::BATCH_SIZE + 1).times { prepare_stuck_service_instance }
+            job.perform
+            expect(ServiceInstanceOperation.where(state: 'failed').count).to eq(ServiceOperationsUpdateInProgressCleanup::BATCH_SIZE)
+          end
+        end
+      end
+
+      describe '#resolve_stuck' do
+        context 'when another process already resolved it (skip_locked returns nil)' do
+          it 'does nothing' do
+            scenario = prepare_stuck_service_instance
+
+            expect do
+              job.send(:resolve_stuck, ServiceInstanceOperation, ServiceInstance,
+                       -1, scenario[:service_instance].id, scenario[:pjob].guid)
+            end.not_to raise_error
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('in progress')
+          end
+        end
+
+        context 'when the operation is stuck in progress' do
+          it 'sets the operation state from in progress to failed' do
+            scenario = prepare_stuck_service_instance
+            op = scenario[:service_instance].last_operation
+
+            expect do
+              job.send(:resolve_stuck, ServiceInstanceOperation, ServiceInstance,
+                       op.id, scenario[:service_instance].id, scenario[:pjob].guid)
+            end.to change { op.reload.state }.from('in progress').to('failed')
+          end
+
+          it 'sets the pollable job state to FAILED' do
+            scenario = prepare_stuck_service_instance(pollable_job_state: PollableJobModel::POLLING_STATE)
+            op = scenario[:service_instance].last_operation
+
+            expect do
+              job.send(:resolve_stuck, ServiceInstanceOperation, ServiceInstance,
+                       op.id, scenario[:service_instance].id, scenario[:pjob].guid)
+            end.to change { scenario[:pjob].reload.state }.from(PollableJobModel::POLLING_STATE).to(PollableJobModel::FAILED_STATE)
+          end
+        end
+      end
+    end
+  end
+end

spec/unit/lib/cloud_controller/clock/scheduler_spec.rb

@@ -22,6 +22,7 @@ module VCAP::CloudController
           pollable_jobs: { cutoff_age_in_days: 2 },
           service_operations_initial_cleanup: { frequency_in_seconds: 600 },
           service_operations_create_in_progress_cleanup: { frequency_in_seconds: 600 },
+          service_operations_update_in_progress_cleanup: { frequency_in_seconds: 600 },
           lifecycle_type_backfill: { frequency_in_seconds: 500 },
           service_usage_events: { cutoff_age_in_days: 5 },
           completed_tasks: { cutoff_age_in_days: 6 },
@@ -169,6 +170,12 @@ module VCAP::CloudController
           expect(block.call).to be_instance_of(Jobs::Runtime::ServiceOperationsCreateInProgressCleanup)
         end
 
+        expect(clock).to receive(:schedule_frequent_worker_job) do |args, &block|
+          expect(args).to eql(name: 'service_operations_update_in_progress_cleanup', interval: 600)
+          expect(Jobs::Runtime::ServiceOperationsUpdateInProgressCleanup).to receive(:new).and_call_original
+          expect(block.call).to be_instance_of(Jobs::Runtime::ServiceOperationsUpdateInProgressCleanup)
+        end
+
         expect(clock).to receive(:schedule_frequent_worker_job) do |args, &block|
           expect(args).to eql(name: 'lifecycle_type_backfill', interval: 500)
           expect(Jobs::Runtime::LifecycleTypeBackfill).to receive(:new).and_call_original