Stuck service operations recovery

app/jobs/runtime/service_operations_binding_delete_stuck_in_progress_retry.rb

@@ -0,0 +1,109 @@
+module VCAP::CloudController
+  module Jobs
+    module Runtime
+      class ServiceOperationsBindingDeleteStuckInProgressRetry < VCAP::CloudController::Jobs::CCJob
+        BATCH_SIZE = 10
+
+        def perform
+          logger.info("Retrying stuck binding 'delete' operations")
+          retry_stuck(ServiceBindingOperation, ServiceBinding, :service_binding_id, 'service_bindings.delete')
+          retry_stuck(ServiceKeyOperation,     ServiceKey,     :service_key_id,     'service_keys.delete')
+        end
+
+        def max_attempts
+          1
+        end
+
+        private
+
+        def retry_stuck(operation_model, instance_model, foreign_key, jobs_operation)
+          # Find stuck binding 'delete' operations where the broker may still be working
+          # but CC's polling job has permanently failed due to a transient error (e.g. brief db connection flip).
+          #
+          # Unlike create we do not mark the operation failed and do not mitigate orphans: for a delete we
+          # re-enqueue the original polling job so the unbind is driven to completion. The original delayed_job's
+          # serialized handler is reused, preserving @start_time so the ReoccurringJob max-duration expiry
+          # (which marks the operation failed via handle_timeout) still fires against the original polling window.
+          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] => 'delete').
+                  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)
+          operation_model.db.transaction do
+            operation = operation_model.where(id: op_id, state: 'in progress').for_update.skip_locked.first
+            return unless operation
+
+            binding = instance_model.first(id: resource_id)
+            return unless binding
+
+            pollable = PollableJobModel.first(guid: pollable_guid)
+            return unless pollable
+
+            handler = deserialize_handler(pollable)
+            return unless handler
+
+            binding_type = instance_model.to_s.split('::').last
+
+            logger.info(
+              "#{binding_type} #{binding.guid} delete operation is stuck in 'in progress'. Re-enqueuing the polling job.",
+              binding_type: binding_type,
+              binding_guid: binding.guid,
+              operation_id: op_id,
+              pollable_job_guid: pollable_guid
+            )
+
+            pollable.update(state: PollableJobModel::POLLING_STATE, cf_api_error: nil)
+            Jobs::GenericEnqueuer.shared.enqueue_pollable(handler, existing_guid: pollable.guid, preserve_priority: true)
+          end
+        end
+
+        # Reuse the original delete polling job by deserializing the failed delayed_job's handler and unwrapping
+        # the wrapper chain (LoggingContextJob → TimeoutJob → PollableJobWrapper → DeleteBindingJob).
+        # This preserves the original @user_audit_info, @start_time and the binding @type.
+        def deserialize_handler(pollable)
+          delayed_job = Delayed::Job[guid: pollable.delayed_job_guid]
+          return unless delayed_job
+
+          Jobs::Enqueuer.unwrap_job(delayed_job.payload_object)
+        rescue StandardError => e
+          logger.error("Could not deserialize delayed job '#{pollable.delayed_job_guid}' for pollable '#{pollable.guid}': #{e.class}: #{e.message}")
+          nil
+        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-binding-delete-stuck-in-progress-retry')
+        end
+
+        def job_name_in_configuration
+          :service_operations_binding_delete_stuck_in_progress_retry
+        end
+      end
+    end
+  end
+end

app/jobs/runtime/service_operations_create_in_progress_cleanup.rb

@@ -40,6 +40,10 @@ def cleanup_operations(operation_model, instance_model, foreign_key, jobs_operat
           #     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
+          #   - service_instances.guid NOT IN (live pollables for this operation): skip resources that still
+          #     have a POLLING/PROCESSING pollable driving this operation. A prior operation on the same
+          #     resource can leave a stale, permanently-failed pollable behind; without this guard that dead
+          #     row would match by resource_guid and cause the current healthy operation to be marked failed
           operation_table = operation_model.table_name
           instance_table = instance_model.table_name
 
@@ -53,6 +57,7 @@ def cleanup_operations(operation_model, instance_model, foreign_key, jobs_operat
                   where(Sequel[:jobs][:state] => [PollableJobModel::POLLING_STATE, PollableJobModel::FAILED_STATE]).
                   where(Sequel[:jobs][:operation] => jobs_operation).
                   exclude(Sequel[:delayed_jobs][:failed_at] => nil).
+                  exclude(live_pollable_exists(operation_model, instance_table, jobs_operation)).
                   select(
                     Sequel[:jobs][:guid].as(:pollable_guid),
                     Sequel[operation_table][:id].as(:op_id),
@@ -104,6 +109,25 @@ def default_maximum_duration_seconds
           Config.config.get(:broker_client_max_async_poll_duration_minutes).minutes
         end
 
+        # NOT EXISTS guard: skip a resource if it still has a pollable job actively driving
+        # THIS operation — state POLLING or PROCESSING AND backed by a delayed_job that has
+        # NOT permanently failed (failed_at IS NULL, or no delayed_job row yet). A stale,
+        # permanently-failed pollable left behind by a previous operation on the same
+        # resource must NOT cause the current healthy operation to be marked failed. A
+        # POLLING pollable whose delayed_job IS failed is itself stuck (the DB flip happened
+        # before the failure hook could write FAILED) and must NOT count as live. Correlated
+        # (resource_guid = instance.guid) so a NULL jobs.resource_guid elsewhere cannot
+        # poison the result the way a NOT IN subquery would.
+        def live_pollable_exists(operation_model, instance_table, jobs_operation)
+          operation_model.db[:jobs].
+            left_join(:delayed_jobs, guid: Sequel[:jobs][:delayed_job_guid]).
+            where(Sequel[:jobs][:operation] => jobs_operation).
+            where(Sequel[:jobs][:state] => [PollableJobModel::POLLING_STATE, PollableJobModel::PROCESSING_STATE]).
+            where(Sequel[:delayed_jobs][:failed_at] => nil).
+            where(Sequel[:jobs][:resource_guid] => Sequel[instance_table][:guid]).
+            exists
+        end
+
         def logger
           @logger ||= Steno.logger('cc.background.service-operations-create-in-progress-cleanup')
         end

app/jobs/runtime/service_operations_delete_stuck_in_progress_retry.rb

@@ -0,0 +1,108 @@
+module VCAP::CloudController
+  module Jobs
+    module Runtime
+      class ServiceOperationsDeleteStuckInProgressRetry < VCAP::CloudController::Jobs::CCJob
+        BATCH_SIZE = 10
+
+        def perform
+          logger.info("Retrying stuck service 'delete' operations")
+          retry_stuck(ServiceInstanceOperation, ServiceInstance, :service_instance_id, 'service_instance.delete')
+        end
+
+        def max_attempts
+          1
+        end
+
+        private
+
+        def retry_stuck(operation_model, instance_model, foreign_key, jobs_operation)
+          # Find stuck service instance 'delete' operations where the broker may still be working
+          # but CC's polling job has permanently failed due to a transient error (e.g. brief db connection flip).
+          #
+          # Unlike create/update we do not mark the operation failed: for a delete we re-enqueue the original
+          # polling job so the deprovision is driven to completion. The original delayed_job's serialized handler
+          # is reused, preserving @start_time so the ReoccurringJob max-duration expiry (which marks the operation
+          # failed via handle_timeout) still fires against the original polling window.
+          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] => 'delete').
+                  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)
+          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
+
+            pollable = PollableJobModel.first(guid: pollable_guid)
+            return unless pollable
+
+            handler = deserialize_handler(pollable)
+            return unless handler
+
+            instance_type = instance_model.to_s.split('::').last
+
+            logger.info(
+              "#{instance_type} #{instance.guid} delete operation is stuck in 'in progress'. Re-enqueuing the polling job.",
+              instance_type: instance_type,
+              instance_guid: instance.guid,
+              operation_id: op_id,
+              pollable_job_guid: pollable_guid
+            )
+
+            pollable.update(state: PollableJobModel::POLLING_STATE, cf_api_error: nil)
+            Jobs::GenericEnqueuer.shared.enqueue_pollable(handler, existing_guid: pollable.guid, preserve_priority: true)
+          end
+        end
+
+        # Reuse the original delete polling job by deserializing the failed delayed_job's handler and unwrapping
+        # the wrapper chain (LoggingContextJob → TimeoutJob → PollableJobWrapper → DeleteServiceInstanceJob).
+        # This preserves the original @user_audit_info, @start_time and the recursive-vs-plain delete variant.
+        def deserialize_handler(pollable)
+          delayed_job = Delayed::Job[guid: pollable.delayed_job_guid]
+          return unless delayed_job
+
+          Jobs::Enqueuer.unwrap_job(delayed_job.payload_object)
+        rescue StandardError => e
+          logger.error("Could not deserialize delayed job '#{pollable.delayed_job_guid}' for pollable '#{pollable.guid}': #{e.class}: #{e.message}")
+          nil
+        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-delete-stuck-in-progress-retry')
+        end
+
+        def job_name_in_configuration
+          :service_operations_delete_stuck_in_progress_retry
+        end
+      end
+    end
+  end
+end

app/jobs/runtime/service_operations_update_stuck_in_progress_failed.rb

@@ -0,0 +1,104 @@
+module VCAP::CloudController
+  module Jobs
+    module Runtime
+      class ServiceOperationsUpdateStuckInProgressFailed < VCAP::CloudController::Jobs::CCJob
+        BATCH_SIZE = 10
+
+        def perform
+          logger.info("Marking stuck service 'update' operations as 'failed'")
+          mark_stuck_in_progress_failed(ServiceInstanceOperation, ServiceInstance, :service_instance_id, 'service_instance.update')
+        end
+
+        def max_attempts
+          1
+        end
+
+        private
+
+        def mark_stuck_in_progress_failed(operation_model, instance_model, foreign_key, jobs_operation)
+          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).
+                  exclude(live_pollable_exists(operation_model, instance_table, jobs_operation)).
+                  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)
+          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
+
+        # NOT EXISTS guard: skip a resource if it still has a pollable job actively driving
+        # THIS operation — state POLLING or PROCESSING AND backed by a delayed_job that has
+        # NOT permanently failed (failed_at IS NULL, or no delayed_job row yet). A stale,
+        # permanently-failed pollable left behind by a previous operation on the same
+        # resource must NOT cause the current healthy operation to be marked failed. A
+        # POLLING pollable whose delayed_job IS failed is itself stuck (the DB flip happened
+        # before the failure hook could write FAILED) and must NOT count as live.
+        # Correlated (resource_guid = instance.guid) so a NULL jobs.resource_guid elsewhere
+        # cannot poison the result the way a NOT IN subquery would.
+        def live_pollable_exists(operation_model, instance_table, jobs_operation)
+          operation_model.db[:jobs].
+            left_join(:delayed_jobs, guid: Sequel[:jobs][:delayed_job_guid]).
+            where(Sequel[:jobs][:operation] => jobs_operation).
+            where(Sequel[:jobs][:state] => [PollableJobModel::POLLING_STATE, PollableJobModel::PROCESSING_STATE]).
+            where(Sequel[:delayed_jobs][:failed_at] => nil).
+            where(Sequel[:jobs][:resource_guid] => Sequel[instance_table][:guid]).
+            exists
+        end
+
+        def logger
+          @logger ||= Steno.logger('cc.background.service-operations-update-stuck-in-progress-failed')
+        end
+
+        def job_name_in_configuration
+          :service_operations_update_stuck_in_progress_failed
+        end
+      end
+    end
+  end
+end

config/cloud_controller.yml

@@ -58,6 +58,15 @@ service_operations_initial_cleanup:
 service_operations_create_in_progress_cleanup:
   frequency_in_seconds: 3600 #1h
 
+service_operations_update_stuck_in_progress_failed:
+  frequency_in_seconds: 3600 #1h
+
+service_operations_delete_stuck_in_progress_retry:
+  frequency_in_seconds: 3600 #1h
+
+service_operations_binding_delete_stuck_in_progress_retry:
+  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,9 @@ 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_stuck_in_progress_failed', class: Jobs::Runtime::ServiceOperationsUpdateStuckInProgressFailed },
+      { name: 'service_operations_delete_stuck_in_progress_retry', class: Jobs::Runtime::ServiceOperationsDeleteStuckInProgressRetry },
+      { name: 'service_operations_binding_delete_stuck_in_progress_retry', class: Jobs::Runtime::ServiceOperationsBindingDeleteStuckInProgressRetry },
       # One-off backfill - to be removed in a future version.
       { name: 'lifecycle_type_backfill', class: Jobs::Runtime::LifecycleTypeBackfill }
     ].freeze