Refactor Storage Migration

Specs were reviewed and improved to better cover the current behavior.
There was some standardization done as well to facilitate the
implementation of the rollback functionality.

StorageMigratorWorker was extracted to HashedStorage namespace were
RollbackerWorker will live one as well.

6 files changed, 53 insertions, 37 deletions

diff --git a/app/services/projects/hashed_storage/migrate_attachments_service.rb b/app/services/projects/hashed_storage/migrate_attachments_service.rb
index a1f0302aeb7..7476980a117 100644
--- a/app/services/projects/hashed_storage/migrate_attachments_service.rb
+++ b/app/services/projects/hashed_storage/migrate_attachments_service.rb
@@ -34,22 +34,22 @@ module Projects
 
       private
 
-      def move_folder!(old_disk_path, new_disk_path)
-        unless File.directory?(old_disk_path)
-          logger.info("Skipped attachments migration from '#{old_disk_path}' to '#{new_disk_path}', source path doesn't exist or is not a directory (PROJECT_ID=#{project.id})")
-          return
+      def move_folder!(old_path, new_path)
+        unless File.directory?(old_path)
+          logger.info("Skipped attachments migration from '#{old_path}' to '#{new_path}', source path doesn't exist or is not a directory (PROJECT_ID=#{project.id})")
+          return true
         end
 
-        if File.exist?(new_disk_path)
-          logger.error("Cannot migrate attachments from '#{old_disk_path}' to '#{new_disk_path}', target path already exist (PROJECT_ID=#{project.id})")
-          raise AttachmentMigrationError, "Target path '#{new_disk_path}' already exist"
+        if File.exist?(new_path)
+          logger.error("Cannot migrate attachments from '#{old_path}' to '#{new_path}', target path already exist (PROJECT_ID=#{project.id})")
+          raise AttachmentMigrationError, "Target path '#{new_path}' already exist"
         end
 
         # Create hashed storage base path folder
-        FileUtils.mkdir_p(File.dirname(new_disk_path))
+        FileUtils.mkdir_p(File.dirname(new_path))
 
-        FileUtils.mv(old_disk_path, new_disk_path)
-        logger.info("Migrated project attachments from '#{old_disk_path}' to '#{new_disk_path}' (PROJECT_ID=#{project.id})")
+        FileUtils.mv(old_path, new_path)
+        logger.info("Migrated project attachments from '#{old_path}' to '#{new_path}' (PROJECT_ID=#{project.id})")
 
         true
       end
diff --git a/app/services/projects/hashed_storage/migration_service.rb b/app/services/projects/hashed_storage/migration_service.rb
index 140c35bdd52..7320f41ebf0 100644
--- a/app/services/projects/hashed_storage/migration_service.rb
+++ b/app/services/projects/hashed_storage/migration_service.rb
@@ -14,16 +14,26 @@ module Projects
       def execute
         # Migrate repository from Legacy to Hashed Storage
         unless project.hashed_storage?(:repository)
-          return unless HashedStorage::MigrateRepositoryService.new(project, old_disk_path, logger: logger).execute
+          return false unless migrate_repository
         end
 
         # Migrate attachments from Legacy to Hashed Storage
         unless project.hashed_storage?(:attachments)
-          HashedStorage::MigrateAttachmentsService.new(project, old_disk_path, logger: logger).execute
+          return false unless migrate_attachments
         end
 
         true
       end
+
+      private
+
+      def migrate_repository
+        HashedStorage::MigrateRepositoryService.new(project, old_disk_path, logger: logger).execute
+      end
+
+      def migrate_attachments
+        HashedStorage::MigrateAttachmentsService.new(project, old_disk_path, logger: logger).execute
+      end
     end
   end
 end
diff --git a/app/workers/all_queues.yml b/app/workers/all_queues.yml
index 4a306fc7a26..85c123c2704 100644
--- a/app/workers/all_queues.yml
+++ b/app/workers/all_queues.yml
@@ -45,6 +45,8 @@
 - github_importer:github_import_stage_import_pull_requests
 - github_importer:github_import_stage_import_repository
 
+- hashed_storage:hashed_storage_migrator
+
 - mail_scheduler:mail_scheduler_issue_due
 - mail_scheduler:mail_scheduler_notification_service
 
@@ -131,7 +133,6 @@
 - repository_fork
 - repository_import
 - repository_remove_remote
-- storage_migrator
 - system_hook_push
 - update_merge_requests
 - upload_checksum
diff --git a/app/workers/hashed_storage/migrator_worker.rb b/app/workers/hashed_storage/migrator_worker.rb
new file mode 100644
index 00000000000..49e347d4060
--- /dev/null
+++ b/app/workers/hashed_storage/migrator_worker.rb
@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+
+module HashedStorage
+  class MigratorWorker
+    include ApplicationWorker
+
+    queue_namespace :hashed_storage
+
+    # @param [Integer] start initial ID of the batch
+    # @param [Integer] finish last ID of the batch
+    def perform(start, finish)
+      migrator = Gitlab::HashedStorage::Migrator.new
+      migrator.bulk_migrate(start: start, finish: finish)
+    end
+  end
+end
diff --git a/app/workers/project_migrate_hashed_storage_worker.rb b/app/workers/project_migrate_hashed_storage_worker.rb
index c3cdf102382..1c8f313e6e9 100644
--- a/app/workers/project_migrate_hashed_storage_worker.rb
+++ b/app/workers/project_migrate_hashed_storage_worker.rb
@@ -4,21 +4,25 @@ class ProjectMigrateHashedStorageWorker
   include ApplicationWorker
 
   LEASE_TIMEOUT = 30.seconds.to_i
+  LEASE_KEY_SEGMENT = 'project_migrate_hashed_storage_worker'.freeze
 
   # rubocop: disable CodeReuse/ActiveRecord
   def perform(project_id, old_disk_path = nil)
-    project = Project.find_by(id: project_id)
-    return if project.nil? || project.pending_delete?
-
     uuid = lease_for(project_id).try_obtain
+
     if uuid
-      ::Projects::HashedStorage::MigrationService.new(project, old_disk_path || project.full_path, logger: logger).execute
+      project = Project.find_by(id: project_id)
+      return if project.nil? || project.pending_delete?
+
+      old_disk_path ||= project.disk_path
+
+      ::Projects::HashedStorage::MigrationService.new(project, old_disk_path, logger: logger).execute
     else
-      false
+      return false
     end
-  rescue => ex
+
+  ensure
     cancel_lease_for(project_id, uuid) if uuid
-    raise ex
   end
   # rubocop: enable CodeReuse/ActiveRecord
 
@@ -29,7 +33,8 @@ class ProjectMigrateHashedStorageWorker
   private
 
   def lease_key(project_id)
-    "project_migrate_hashed_storage_worker:#{project_id}"
+    # we share the same lease key for both migration and rollback so they don't run simultaneously
+    "#{LEASE_KEY_SEGMENT}:#{project_id}"
   end
 
   def cancel_lease_for(project_id, uuid)
diff --git a/app/workers/storage_migrator_worker.rb b/app/workers/storage_migrator_worker.rb
deleted file mode 100644
index 714c4eed5b2..00000000000
--- a/app/workers/storage_migrator_worker.rb
+++ /dev/null
@@ -1,16 +0,0 @@
-# frozen_string_literal: true
-
-class StorageMigratorWorker
-  include ApplicationWorker
-
-  # @param [Integer] start initial ID of the batch
-  # @param [Integer] finish last ID of the batch
-  # @param [String] operation the operation to be performed: ['migrate', 'rollback']
-  def perform(start, finish, operation = :migrate)
-    # when scheduling a job symbols are converted to string, we need to convert back
-    operation = operation.to_sym if operation
-
-    migrator = Gitlab::HashedStorage::Migrator.new
-    migrator.bulk_migrate(start: start, finish: finish, operation: operation)
-  end
-end