Prevent DropletUpload from downgrading a staged droplet to failed (#5244)

If a worker is killed mid-job and restarts, it re-locks and re-runs its
own job. When another worker already completed the job in the meantime,
the droplet is already STAGED. The rescue block would unconditionally
overwrite STAGED with FAILED in that case.

Skip the state transition and log when the droplet is already staged.

Author: svkrieger

app/jobs/v3/droplet_upload.rb

@@ -35,9 +35,13 @@ def perform
           if droplet
             droplet.db.transaction do
               droplet.lock!
-              droplet.error_description = e.message
-              droplet.state = DropletModel::FAILED_STATE
-              droplet.save
+              if droplet.staged?
+                logger.info('droplet-upload.skipping-failed-state', droplet_guid: @droplet_guid, error: e.message)
+              else
+                droplet.error_description = e.message
+                droplet.state = DropletModel::FAILED_STATE
+                droplet.save
+              end
             end
           end
           raise
@@ -66,6 +70,10 @@ def resource_type
         def blobstore
           @blobstore ||= CloudController::DependencyLocator.instance.droplet_blobstore
         end
+
+        def logger
+          @logger ||= Steno.logger('cc.jobs.v3.droplet_upload')
+        end
       end
     end
   end

spec/unit/jobs/v3/droplet_upload_spec.rb

@@ -146,6 +146,24 @@ def reschedule_at(_, _=nil)
                 }.from(true).to(false)
               end
             end
+
+            context 'when the droplet is already STAGED' do
+              let(:staged_droplet) { create(:droplet_model, state: DropletModel::STAGED_STATE, droplet_hash: nil, sha256_checksum: nil, set_as_current_droplet: false) }
+
+              before do
+                staged_job = DropletUpload.new(local_file.path, staged_droplet.guid, skip_state_transition:)
+                Delayed::Job.enqueue(staged_job, queue: worker.name)
+                worker.work_off 1
+              end
+
+              it 'does not overwrite the droplet state' do
+                expect(staged_droplet.refresh.state).to eq(DropletModel::STAGED_STATE)
+              end
+
+              it 'does not set an error description' do
+                expect(staged_droplet.refresh.error_description).to be_nil
+              end
+            end
           end
 
           context 'if the file is missing' do