Adjust review comments

Author: kathap

app/jobs/pollable_job_wrapper.rb

@@ -70,11 +70,12 @@ def error(job, exception)
       end
 
       def failure(job)
-        begin
-          unwrapped_job = Enqueuer.unwrap_job(@handler)
-          unwrapped_job.recover_from_failure
-        rescue StandardError => e
-          logger.error("failure recovery failed: #{e.class}: #{e.message}")
+        if @handler.respond_to?(:recover_from_failure)
+          begin
+            @handler.recover_from_failure
+          rescue StandardError => e
+            logger.error("failure recovery failed: #{e.class}: #{e.message}")
+          end
         end
 
         change_state(job, PollableJobModel::FAILED_STATE)

spec/unit/jobs/pollable_job_wrapper_spec.rb

@@ -202,52 +202,54 @@ class BigException < StandardError
           end
         end
 
-        context 'when the job implements recover_from_failure' do
-          let(:broker) do
-            VCAP::CloudController::ServiceBroker.create(
-              name: 'recovery-test-broker',
-              broker_url: 'http://example.org/broker-url',
-              auth_username: 'username',
-              auth_password: 'password',
-              state: VCAP::CloudController::ServiceBrokerStateEnum::SYNCHRONIZING
-            )
+        describe '#failure' do
+          let(:delayed_job) { instance_double(Delayed::Backend::Sequel::Job, guid: 'job-guid') }
+          let!(:pollable_job) do
+            VCAP::CloudController::PollableJobModel.make(delayed_job_guid: 'job-guid', state: VCAP::CloudController::PollableJobModel::PROCESSING_STATE)
           end
 
-          let(:update_request) do
-            VCAP::CloudController::ServiceBrokerUpdateRequest.create(
-              name: 'new-name',
-              service_broker_id: broker.id
-            )
-          end
+          context 'when handler implements recover_from_failure' do
+            let(:handler) do
+              instance_double(VCAP::CloudController::V3::UpdateBrokerJob, recover_from_failure: nil, warnings: nil)
+            end
+            let(:wrapper) { PollableJobWrapper.new(handler) }
 
-          let(:user_audit_info) { instance_double(VCAP::CloudController::UserAuditInfo, { user_guid: Sham.guid }) }
-          let(:update_job) { VCAP::CloudController::V3::UpdateBrokerJob.new(update_request.guid, broker.guid, 'AVAILABLE', user_audit_info:) }
-          let(:pollable_job) { PollableJobWrapper.new(update_job) }
+            it 'calls recover_from_failure and marks the pollable job failed' do
+              wrapper.failure(delayed_job)
 
-          before do
-            allow_any_instance_of(VCAP::CloudController::V3::UpdateBrokerJob::Perform).to receive(:perform).and_raise(StandardError.new('job failed'))
-          end
+              expect(handler).to have_received(:recover_from_failure)
+              expect(pollable_job.reload.state).to eq(VCAP::CloudController::PollableJobModel::FAILED_STATE)
+            end
 
-          it 'calls recover_from_failure and reverts broker state to AVAILABLE' do
-            enqueued_job = VCAP::CloudController::Jobs::Enqueuer.new.enqueue(pollable_job)
-            VCAP::CloudController::PollableJobModel.make(delayed_job_guid: enqueued_job.guid, state: 'PROCESSING')
+            context 'when recover_from_failure raises an error' do
+              let(:logger) { instance_double(Steno::Logger, error: nil) }
 
-            execute_all_jobs(expected_successes: 0, expected_failures: 1)
+              before do
+                allow(handler).to receive(:recover_from_failure).and_raise(StandardError.new('recovery failed'))
+                allow(Steno).to receive(:logger).with('cc.pollable.job.wrapper').and_return(logger)
+              end
 
-            broker.reload
-            expect(broker.state).to eq('SYNCHRONIZATION_FAILED')
-          end
-        end
+              it 'logs the error without re-raising' do
+                expect { wrapper.failure(delayed_job) }.not_to raise_error
+                expect(logger).to have_received(:error).with(/failure recovery failed/)
+              end
 
-        context 'when the job does not implement recover_from_failure' do
-          it 'still marks the job as FAILED without error' do
-            enqueued_job = VCAP::CloudController::Jobs::Enqueuer.new.enqueue(pollable_job)
-            job_model = VCAP::CloudController::PollableJobModel.make(delayed_job_guid: enqueued_job.guid, state: 'PROCESSING')
+              it 'still marks the pollable job as failed' do
+                wrapper.failure(delayed_job)
 
-            execute_all_jobs(expected_successes: 0, expected_failures: 1)
+                expect(pollable_job.reload.state).to eq(VCAP::CloudController::PollableJobModel::FAILED_STATE)
+              end
+            end
+          end
 
-            job_model.reload
-            expect(job_model.state).to eq('FAILED')
+          context 'when handler does not implement recover_from_failure' do
+            let(:handler) { double('HandlerWithoutRecovery', warnings: nil) }
+            let(:wrapper) { PollableJobWrapper.new(handler) }
+
+            it 'still marks the pollable job as failed without error' do
+              expect { wrapper.failure(delayed_job) }.not_to raise_error
+              expect(pollable_job.reload.state).to eq(VCAP::CloudController::PollableJobModel::FAILED_STATE)
+            end
           end
         end
       end

spec/unit/jobs/v3/services/synchronize_broker_catalog_job_spec.rb

@@ -201,6 +201,70 @@ def incompatible_catalog
             allow(catalog).to receive_messages(valid?: true, compatible?: false, incompatibility_errors: incompatibility_errors)
           end
         end
+
+        describe '#recover_from_failure' do
+          let(:broker) do
+            ServiceBroker.create(
+              name: 'test-broker',
+              broker_url: 'http://example.org/broker-url',
+              auth_username: 'username',
+              auth_password: 'password'
+            )
+          end
+          let(:user_audit_info) { instance_double(UserAuditInfo, { user_guid: Sham.guid }) }
+
+          subject(:job) do
+            SynchronizeBrokerCatalogJob.new(broker.guid, user_audit_info:)
+          end
+
+          context 'when broker is in SYNCHRONIZING state' do
+            before do
+              broker.update(state: ServiceBrokerStateEnum::SYNCHRONIZING)
+            end
+
+            it 'sets the broker to SYNCHRONIZATION_FAILED' do
+              expect do
+                job.recover_from_failure
+              end.to change { broker.reload.state }.
+                from(ServiceBrokerStateEnum::SYNCHRONIZING).
+                to(ServiceBrokerStateEnum::SYNCHRONIZATION_FAILED)
+            end
+          end
+
+          shared_examples 'does not change the broker state' do |expected_state|
+            it 'leaves the state unchanged' do
+              broker.update(state: expected_state)
+              job.recover_from_failure
+
+              expect(broker.reload.state).to eq(expected_state)
+            end
+          end
+
+          context 'when broker is in a different state' do
+            include_examples 'does not change the broker state', ServiceBrokerStateEnum::AVAILABLE
+            include_examples 'does not change the broker state', ServiceBrokerStateEnum::DELETE_IN_PROGRESS
+          end
+
+          context 'when database error occurs during recovery' do
+            let(:dataset) { instance_double(Sequel::Dataset) }
+            let(:logger) { instance_double(Steno::Logger, error: nil) }
+
+            before do
+              broker.update(state: ServiceBrokerStateEnum::SYNCHRONIZING)
+              allow(ServiceBroker).to receive(:where).
+                with(guid: broker.guid, state: ServiceBrokerStateEnum::SYNCHRONIZING).
+                and_return(dataset)
+              allow(dataset).to receive(:update).and_raise(Sequel::DatabaseError.new(RuntimeError.new('connection lost')))
+              allow(Steno).to receive(:logger).with('cc.background').and_return(logger)
+            end
+
+            it 'logs the error and does not raise' do
+              expect { job.recover_from_failure }.not_to raise_error
+              expect(logger).to have_received(:error).with(/Failed to recover broker state/)
+              expect(broker.reload.state).to eq(ServiceBrokerStateEnum::SYNCHRONIZING)
+            end
+          end
+        end
       end
     end
   end

spec/unit/jobs/v3/services/update_broker_job_spec.rb

@@ -448,25 +448,21 @@ module V3
 
           context 'when database disconnects during state rollback' do
             let(:catalog_error) { StandardError.new('Catalog fetch failed') }
+            let(:mock_dataset) { instance_double(Sequel::Postgres::Dataset) }
 
             before do
               allow_any_instance_of(VCAP::CloudController::V3::ServiceBrokerCatalogUpdater).to receive(:refresh).and_raise(catalog_error)
-
-              mock_dataset = instance_double(Sequel::Postgres::Dataset)
               allow(mock_dataset).to receive(:update).and_raise(Sequel::DatabaseDisconnectError.new('connection lost'))
-
               allow(ServiceBroker).to receive(:where).and_call_original
-              allow(ServiceBroker).to receive(:where).with(id: broker.id).and_return(mock_dataset)
+              allow(ServiceBroker).to receive(:where).
+                with(id: update_broker_request.service_broker_id, state: ServiceBrokerStateEnum::SYNCHRONIZING).
+                and_return(mock_dataset)
             end
 
-            it 'swallows the database error and re-raises the original catalog error' do
+            it 're-raises the original error instead of the rollback database error' do
               expect { job.perform }.to raise_error(catalog_error)
             end
 
-            it 'does not raise a database connection error' do
-              expect { job.perform }.not_to raise_error(Sequel::DatabaseDisconnectError)
-            end
-
             it 'still cleans up the update request' do
               expect { job.perform }.to raise_error(catalog_error)
               expect(ServiceBrokerUpdateRequest.where(id: update_broker_request.id).all).to be_empty
@@ -497,48 +493,45 @@ module V3
               end
 
               it 'sets the broker to SYNCHRONIZATION_FAILED' do
-                job.recover_from_failure
-
-                broker.reload
-                expect(broker.state).to eq(ServiceBrokerStateEnum::SYNCHRONIZATION_FAILED)
+                expect do
+                  job.recover_from_failure
+                end.to change { broker.reload.state }.
+                  from(ServiceBrokerStateEnum::SYNCHRONIZING).
+                  to(ServiceBrokerStateEnum::SYNCHRONIZATION_FAILED)
               end
             end
 
-            context 'when broker is not in SYNCHRONIZING state' do
-              before do
-                broker.update(state: ServiceBrokerStateEnum::AVAILABLE)
-              end
-
-              it 'does not change the broker state' do
+            shared_examples 'does not change the broker state' do |expected_state|
+              it 'leaves the state unchanged' do
+                broker.update(state: expected_state)
                 job.recover_from_failure
 
-                broker.reload
-                expect(broker.state).to eq(ServiceBrokerStateEnum::AVAILABLE)
+                expect(broker.reload.state).to eq(expected_state)
               end
             end
 
-            context 'when broker state has changed to something else' do
-              before do
-                broker.update(state: ServiceBrokerStateEnum::DELETE_IN_PROGRESS)
-              end
-
-              it 'does not overwrite the newer state' do
-                job.recover_from_failure
-
-                broker.reload
-                expect(broker.state).to eq(ServiceBrokerStateEnum::DELETE_IN_PROGRESS)
-              end
+            context 'when broker is in a different state' do
+              include_examples 'does not change the broker state', ServiceBrokerStateEnum::AVAILABLE
+              include_examples 'does not change the broker state', ServiceBrokerStateEnum::DELETE_IN_PROGRESS
             end
 
             context 'when database error occurs during recovery' do
+              let(:dataset) { instance_double(Sequel::Dataset) }
+              let(:logger) { instance_double(Steno::Logger, error: nil) }
+
               before do
                 broker.update(state: ServiceBrokerStateEnum::SYNCHRONIZING)
-                allow(ServiceBroker).to receive(:where).and_raise(Sequel::DatabaseError.new('connection lost'))
-                allow(Steno).to receive(:logger).and_return(double(error: nil))
+                allow(ServiceBroker).to receive(:where).
+                  with(guid: broker.guid, state: ServiceBrokerStateEnum::SYNCHRONIZING).
+                  and_return(dataset)
+                allow(dataset).to receive(:update).and_raise(Sequel::DatabaseError.new(RuntimeError.new('connection lost')))
+                allow(Steno).to receive(:logger).with('cc.background').and_return(logger)
               end
 
               it 'logs the error and does not raise' do
                 expect { job.recover_from_failure }.not_to raise_error
+                expect(logger).to have_received(:error).with(/Failed to recover broker state/)
+                expect(broker.reload.state).to eq(ServiceBrokerStateEnum::SYNCHRONIZING)
               end
             end
           end