v3: deprovision SI when LO returs state failed

[#172330022](https://www.pivotaltracker.com/story/show/172330022)

Co-authored-by: George Blue <gblue@pivotal.io>
Co-authored-by: Derik Evangelista <devangelista@pivotal.io>
Co-authored-by: George Blue <gblue@pivotal.io>

Author: Derik Evangelista

docs/v3/source/includes/experimental_resources/service_instances/_create.md.erb

@@ -99,7 +99,11 @@ Content-Type: application/json
 ```
 
 This endpoint creates a new service instance. Service instances can be of type `managed` or `user-provided`, and
-the required parameters are different for each type.
+the required parameters are different for each type. User provided service instances do not require interactions with
+service brokers.
+
+If failures occur when creating managed service instances, the API might execute orphan mitigation steps
+accordingly to cases outlined in the [OSBAPI specification](https://github.com/openservicebrokerapi/servicebroker/blob/master/spec.md#orphan-mitigation)
 
 #### Definition
 `POST /v3/service_instances`

lib/services/service_brokers/v2/client.rb

@@ -263,6 +263,10 @@ def fetch_service_instance_last_operation(instance)
           }
       }
 
+      if result[:last_operation][:state] == 'failed'
+        @orphan_mitigator.cleanup_failed_provision(@attrs, instance)
+      end
+
       result[:last_operation][:description] = last_operation_hash['description'] if last_operation_hash['description']
       result[:retry_after] = response[HttpResponse::HEADER_RETRY_AFTER] if response[HttpResponse::HEADER_RETRY_AFTER]
       result.merge(parsed_response.symbolize_keys)

spec/request/service_instances_spec.rb

@@ -1136,7 +1136,7 @@ def check_filtered_instances(*instances)
               expect(job.state).to eq(VCAP::CloudController::PollableJobModel::POLLING_STATE)
 
               Timecop.freeze(Time.now + 1.hour) do
-                execute_all_jobs(expected_successes: 0, expected_failures: 1)
+                execute_all_jobs(expected_successes: 0, expected_failures: 1, jobs_to_execute: 1)
               end
             end
 
@@ -1149,6 +1149,12 @@ def check_filtered_instances(*instances)
               expect(instance.last_operation.type).to eq('create')
               expect(instance.last_operation.state).to eq('failed')
             end
+
+            it 'fires an orphan mitigation job' do
+              jobs = Delayed::Job.where(failed_at: nil).all
+              expect(jobs).to have(1).jobs
+              expect(jobs.first).to be_a_fully_wrapped_job_of(VCAP::CloudController::Jobs::Services::DeleteOrphanedInstance)
+            end
           end
         end
       end

spec/support/background_job_helpers.rb

@@ -1,8 +1,8 @@
 module BackgroundJobHelpers
   include VCAP::CloudController
 
-  def execute_all_jobs(expected_successes:, expected_failures:)
-    successes, failures = Delayed::Worker.new.work_off
+  def execute_all_jobs(expected_successes:, expected_failures:, jobs_to_execute: 100)
+    successes, failures = Delayed::Worker.new.work_off(jobs_to_execute)
     failure_message = "Expected #{expected_successes} successful and #{expected_failures} failed jobs, got #{successes} successful and #{failures} failed jobs."
     fail_summaries = Delayed::Job.exclude(failed_at: nil).map { |j| "Handler: #{j.handler}, LastError: #{j.last_error}" }
     if fail_summaries.count > 0

spec/unit/lib/services/service_brokers/v2/client_spec.rb

@@ -436,6 +436,37 @@ module VCAP::Services::ServiceBrokers::V2
         end
       end
 
+      context 'when the broker returns 200 (ok)' do
+        let(:code) { 200 }
+        let(:message) { 'OK' }
+
+        context 'when the state is `failed`' do
+          let(:response_data) do
+            {
+              state: 'failed'
+            }
+          end
+
+          it 'performs orphan mitigation' do
+            client.fetch_service_instance_last_operation(instance)
+            expect(orphan_mitigator).to have_received(:cleanup_failed_provision).with(client_attrs, instance)
+          end
+        end
+
+        context 'when the state is something else' do
+          let(:response_data) do
+            {
+              state: 'succeeded'
+            }
+          end
+
+          it 'does not perform orphan mitigation' do
+            client.fetch_service_instance_last_operation(instance)
+            expect(orphan_mitigator).not_to have_received(:cleanup_failed_provision)
+          end
+        end
+      end
+
       context 'when the broker returns 410 (gone)' do
         let(:code) { 410 }
         let(:message) { 'GONE' }