Fix completion of pending direct upload

This fixes the bug of pending direct uploads not being
properly cleared off the Redis hash even on successful upload.

The cause of the problem was that the redis key pattern being used
to delete the redis field still includes the bucket prefix.

Changelog: fixed

app/uploaders/object_storage.rb

@@ -31,7 +31,8 @@ def store!(file)
       # The direct_upload_final_path is defined which means
       # file was uploaded to its final location so no need to move it.
       # Now we delete the pending upload entry as the upload is considered complete.
-      ObjectStorage::PendingDirectUpload.complete(@uploader.class.storage_location_identifier, file.path)
+      pending_upload_path = @uploader.class.without_bucket_prefix(file.path)
+      ObjectStorage::PendingDirectUpload.complete(@uploader.class.storage_location_identifier, pending_upload_path)
 
       file
     end
@@ -196,6 +197,10 @@ def with_bucket_prefix(path)
         File.join([object_store_options.bucket_prefix, path].compact)
       end
 
+      def without_bucket_prefix(path)
+        Pathname.new(path).relative_path_from(object_store_options.bucket_prefix.to_s).to_s
+      end
+
       def object_store_config
         ObjectStorage::Config.new(object_store_options)
       end

spec/uploaders/object_storage_spec.rb

@@ -1097,19 +1097,31 @@ def escape_path(path)
               let(:fog_config) do
                 Gitlab.config.uploads.object_store.tap do |config|
                   config[:remote_directory] = 'main-bucket'
-                  config[:bucket_prefix] = 'uploads'
+                  config[:bucket_prefix] = 'my/uploads'
                 end
               end
 
               let(:bucket) { 'main-bucket' }
-              let(:fog_file_path) { "uploads/#{final_path}" }
+              let(:fog_file_path) { "my/uploads/#{final_path}" }
 
               it 'stores the file final path in the db without the prefix' do
                 expect { subject }.not_to raise_error
 
-                expect(uploader.store_path).to eq("uploads/#{final_path}")
+                expect(uploader.store_path).to eq("my/uploads/#{final_path}")
                 expect(object.file_final_path).to eq(final_path)
               end
+
+              context 'and file is stored' do
+                subject do
+                  uploader.store!(uploaded_file)
+                end
+
+                it 'completes the matching pending upload entry' do
+                  expect { subject }
+                    .to change { ObjectStorage::PendingDirectUpload.exists?(uploader_class.storage_location_identifier, final_path) }
+                    .to(false)
+                end
+              end
             end
 
             context 'when file is stored' do