Skip stuck-operation recovery when a live pollable still drives it

The stuck-operation clock jobs (create cleanup, update fail, delete
retry, binding-delete retry) matched a resource's pollable by
resource_guid alone, so a permanently-failed pollable left by a previous
operation wrongly caused a subsequent healthy operation to be failed or
re-enqueued.

Add a correlated NOT EXISTS guard that skips a resource when it still has
a pollable actively driving the same operation (POLLING/PROCESSING backed
by a non-failed delayed_job).

Author: kathap

app/jobs/runtime/service_operations_binding_delete_stuck_in_progress_retry.rb

@@ -37,6 +37,7 @@ def retry_stuck(operation_model, instance_model, foreign_key, jobs_operation)
                   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),
@@ -96,6 +97,25 @@ def default_maximum_duration_seconds
           Config.config.get(:broker_client_max_async_poll_duration_minutes).minutes
         end
 
+        # NOT EXISTS guard: skip a binding 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
+        # binding must NOT trigger a spurious re-enqueue. 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 = binding.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-binding-delete-stuck-in-progress-retry')
         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

@@ -36,6 +36,7 @@ def retry_stuck(operation_model, instance_model, foreign_key, jobs_operation)
                   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),
@@ -95,6 +96,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 trigger a spurious re-enqueue. 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-delete-stuck-in-progress-retry')
         end

app/jobs/runtime/service_operations_update_stuck_in_progress_failed.rb

@@ -29,6 +29,7 @@ def mark_stuck_in_progress_failed(operation_model, instance_model, foreign_key,
                   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),
@@ -71,6 +72,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-update-stuck-in-progress-failed')
         end

spec/unit/jobs/runtime/service_operations_binding_delete_stuck_in_progress_retry_spec.rb

@@ -49,6 +49,18 @@ def prepare_stuck_binding(
         { binding: binding, pjob: pjob, delayed_job: dj }
       end
 
+      # Attach an additional live pollable job (POLLING/PROCESSING, delayed_job NOT failed)
+      # for the same binding + operation. Mirrors a second delete that is actively polling
+      # while a stale, permanently-failed pollable from a previous attempt lingers.
+      def add_live_pollable(binding_type, binding, state: PollableJobModel::POLLING_STATE)
+        operation = binding_type == :credential ? 'service_bindings.delete' : 'service_keys.delete'
+        resource_type = binding_type == :credential ? 'service_bindings' : 'service_keys'
+        delete_job = V3::DeleteBindingJob.new(binding_type, binding.guid, user_audit_info: user_audit_info)
+        pjob = Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(delete_job)
+        pjob.update(state: state, operation: operation, resource_type: resource_type)
+        pjob
+      end
+
       it { is_expected.to be_a_valid_job }
 
       %i[credential key].each do |binding_type|
@@ -122,6 +134,34 @@ def prepare_stuck_binding(
             it_behaves_like 'does not retry the operation'
           end
 
+          context 'when a live pollable job is still driving the same operation' do
+            # A previous delete attempt left a stale, permanently-failed pollable behind; a
+            # second delete on the same binding is now actively polling. The stale row must
+            # not trigger a spurious re-enqueue.
+            it 'does not retry and leaves both pollables untouched' do
+              scenario = prepare_stuck_binding(binding_type: binding_type)
+              live_pjob = add_live_pollable(binding_type, scenario[:binding])
+
+              job.perform
+
+              expect(scenario[:binding].last_operation.reload.state).to eq('in progress')
+              expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+              expect(live_pjob.reload.state).to eq(PollableJobModel::POLLING_STATE)
+              expect(enqueuer).not_to have_received(:enqueue_pollable)
+            end
+
+            it 'still retries once the live pollable is gone' do
+              scenario = prepare_stuck_binding(binding_type: binding_type)
+              live_pjob = add_live_pollable(binding_type, scenario[:binding])
+              live_pjob.destroy
+
+              job.perform
+
+              expect(scenario[:pjob].reload.state).to eq(PollableJobModel::POLLING_STATE)
+              expect(enqueuer).to have_received(:enqueue_pollable)
+            end
+          end
+
           context 'when a binding delete job is stuck with state FAILED' do
             it 'resets the pollable job to POLLING and re-enqueues the original delete job' do
               scenario = prepare_stuck_binding(binding_type: binding_type)

spec/unit/jobs/runtime/service_operations_create_in_progress_cleanup_spec.rb

@@ -55,6 +55,24 @@ def prepare_stuck_service_instance(
         { service_instance: service_instance, pjob: pjob, delayed_job: dj }
       end
 
+      # Attach an additional live pollable job (POLLING/PROCESSING, delayed_job NOT failed)
+      # for the same instance + operation. Mirrors a second create actively polling while a
+      # stale, permanently-failed pollable from a previous attempt lingers.
+      def add_live_pollable(service_instance, operation: 'service_instance.create', state: PollableJobModel::POLLING_STATE)
+        dj = Delayed::Job.create!(
+          guid: SecureRandom.uuid,
+          handler: 'fake',
+          run_at: Time.now,
+          queue: 'cc-generic'
+        )
+        create(:pollable_job_model,
+               state: state,
+               operation: operation,
+               resource_guid: service_instance.guid,
+               resource_type: 'service_instances',
+               delayed_job_guid: dj.guid)
+      end
+
       shared_examples 'does not trigger orphan mitigation' do
         before { job.perform }
 
@@ -119,6 +137,34 @@ def prepare_stuck_service_instance(
           it_behaves_like 'does not trigger orphan mitigation'
         end
 
+        context 'when a live pollable job is still driving the same operation' do
+          # A previous create attempt left a stale, permanently-failed pollable behind; a
+          # second create on the same instance is now actively polling. The stale row must
+          # not cause the healthy current operation to be marked failed / mitigated.
+          it 'does not mitigate and leaves both pollables untouched' do
+            scenario = prepare_stuck_service_instance
+            live_pjob = add_live_pollable(scenario[:service_instance])
+
+            job.perform
+
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('in progress')
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+            expect(live_pjob.reload.state).to eq(PollableJobModel::POLLING_STATE)
+            expect(fake_mitigator).not_to have_received(:cleanup_failed_provision)
+          end
+
+          it 'still mitigates once the live pollable is gone' do
+            scenario = prepare_stuck_service_instance
+            live_pjob = add_live_pollable(scenario[:service_instance])
+            live_pjob.destroy
+
+            job.perform
+
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('failed')
+            expect(fake_mitigator).to have_received(:cleanup_failed_provision).with(scenario[:service_instance])
+          end
+        end
+
         context 'when a service instance create job is stuck with state FAILED' do
           it 'sets operation to failed, pollable job to FAILED, and triggers orphan mitigation' do
             scenario = prepare_stuck_service_instance

spec/unit/jobs/runtime/service_operations_delete_stuck_in_progress_retry_spec.rb

@@ -44,6 +44,16 @@ def prepare_stuck_service_instance(
         { service_instance: service_instance, pjob: pjob, delayed_job: dj }
       end
 
+      # Attach an additional live pollable job (POLLING/PROCESSING, delayed_job NOT failed)
+      # for the same instance + operation. Mirrors a second delete that is actively polling
+      # while a stale, permanently-failed pollable from a previous attempt lingers.
+      def add_live_pollable(service_instance, operation: 'service_instance.delete', state: PollableJobModel::POLLING_STATE)
+        delete_job = V3::DeleteServiceInstanceJob.new(service_instance.guid, user_audit_info)
+        pjob = Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(delete_job)
+        pjob.update(state: state, operation: operation)
+        pjob
+      end
+
       shared_examples 'does not retry the operation' do
         it 'leaves the operation in progress, the pollable job untouched, and does not re-enqueue' do
           scenario = subject_scenario
@@ -110,6 +120,34 @@ def prepare_stuck_service_instance(
           it_behaves_like 'does not retry the operation'
         end
 
+        context 'when a live pollable job is still driving the same operation' do
+          # A previous delete attempt left a stale, permanently-failed pollable behind; a
+          # second delete on the same instance is now actively polling. The stale row must
+          # not trigger a spurious re-enqueue.
+          it 'does not retry and leaves both pollables untouched' do
+            scenario = prepare_stuck_service_instance
+            live_pjob = add_live_pollable(scenario[:service_instance])
+
+            job.perform
+
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('in progress')
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+            expect(live_pjob.reload.state).to eq(PollableJobModel::POLLING_STATE)
+            expect(enqueuer).not_to have_received(:enqueue_pollable)
+          end
+
+          it 'still retries once the live pollable is gone' do
+            scenario = prepare_stuck_service_instance
+            live_pjob = add_live_pollable(scenario[:service_instance])
+            live_pjob.destroy
+
+            job.perform
+
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::POLLING_STATE)
+            expect(enqueuer).to have_received(:enqueue_pollable)
+          end
+        end
+
         context 'when a service instance delete job is stuck with state FAILED' do
           it 'resets the pollable job to POLLING and re-enqueues the original delete job' do
             scenario = prepare_stuck_service_instance