Add latest changes from gitlab-org/gitlab@master

11 files changed, 175 insertions, 11 deletions

diff --git a/lib/gitlab/background_migration/extract_project_topics_into_separate_table.rb b/lib/gitlab/background_migration/extract_project_topics_into_separate_table.rb
index c5beb5a6865..31b5b5cdb73 100644
--- a/lib/gitlab/background_migration/extract_project_topics_into_separate_table.rb
+++ b/lib/gitlab/background_migration/extract_project_topics_into_separate_table.rb
@@ -33,11 +33,15 @@ module Gitlab
 
       def perform(start_id, stop_id)
         Tagging.includes(:tag).where(taggable_type: 'Project', id: start_id..stop_id).each do |tagging|
-          if Project.exists?(id: tagging.taggable_id)
-            topic = Topic.find_or_create_by(name: tagging.tag.name)
-            project_topic = ProjectTopic.find_or_create_by(project_id: tagging.taggable_id, topic: topic)
-
-            tagging.delete if project_topic.persisted?
+          if Project.exists?(id: tagging.taggable_id) && tagging.tag
+            begin
+              topic = Topic.find_or_create_by(name: tagging.tag.name)
+              project_topic = ProjectTopic.find_or_create_by(project_id: tagging.taggable_id, topic: topic)
+
+              tagging.delete if project_topic.persisted?
+            rescue StandardError => e
+              Gitlab::ErrorTracking.log_exception(e, tagging_id: tagging.id)
+            end
           else
             tagging.delete
           end
diff --git a/lib/gitlab/background_migration/mailers/unconfirm_mailer.rb b/lib/gitlab/background_migration/mailers/unconfirm_mailer.rb
index c096dae0631..3605b157f4f 100644
--- a/lib/gitlab/background_migration/mailers/unconfirm_mailer.rb
+++ b/lib/gitlab/background_migration/mailers/unconfirm_mailer.rb
@@ -14,7 +14,7 @@ module Gitlab
           mail(
             template_path: 'unconfirm_mailer',
             template_name: 'unconfirm_notification_email',
-            to: @user.notification_email,
+            to: @user.notification_email_or_default,
             subject: subject('GitLab email verification request')
           )
         end
diff --git a/lib/gitlab/database/connection.rb b/lib/gitlab/database/connection.rb
index 6eac115be13..cda6220ee6c 100644
--- a/lib/gitlab/database/connection.rb
+++ b/lib/gitlab/database/connection.rb
@@ -187,6 +187,19 @@ module Gitlab
         row['system_identifier']
       end
 
+      def pg_wal_lsn_diff(location1, location2)
+        lsn1 = connection.quote(location1)
+        lsn2 = connection.quote(location2)
+
+        query = <<-SQL.squish
+            SELECT pg_wal_lsn_diff(#{lsn1}, #{lsn2})
+              AS result
+        SQL
+
+        row = connection.select_all(query).first
+        row['result'] if row
+      end
+
       # @param [ActiveRecord::Connection] ar_connection
       # @return [String]
       def get_write_location(ar_connection)
diff --git a/lib/gitlab/database/load_balancing/sidekiq_server_middleware.rb b/lib/gitlab/database/load_balancing/sidekiq_server_middleware.rb
index f7fda14b215..15f8f0fb240 100644
--- a/lib/gitlab/database/load_balancing/sidekiq_server_middleware.rb
+++ b/lib/gitlab/database/load_balancing/sidekiq_server_middleware.rb
@@ -57,7 +57,7 @@ module Gitlab
         end
 
         def get_wal_locations(job)
-          job['wal_locations'] || legacy_wal_location(job)
+          job['dedup_wal_locations'] || job['wal_locations'] || legacy_wal_location(job)
         end
 
         # Already scheduled jobs could still contain legacy database write location.
diff --git a/lib/gitlab/database/migration_helpers/loose_foreign_key_helpers.rb b/lib/gitlab/database/migration_helpers/loose_foreign_key_helpers.rb
new file mode 100644
index 00000000000..30601bffd7a
--- /dev/null
+++ b/lib/gitlab/database/migration_helpers/loose_foreign_key_helpers.rb
@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Database
+    module MigrationHelpers
+      module LooseForeignKeyHelpers
+        include Gitlab::Database::SchemaHelpers
+
+        DELETED_RECORDS_INSERT_FUNCTION_NAME = 'insert_into_loose_foreign_keys_deleted_records'
+
+        def track_record_deletions(table)
+          execute(<<~SQL)
+          CREATE TRIGGER #{record_deletion_trigger_name(table)}
+          AFTER DELETE ON #{table} REFERENCING OLD TABLE AS old_table
+          FOR EACH STATEMENT
+          EXECUTE FUNCTION #{DELETED_RECORDS_INSERT_FUNCTION_NAME}();
+          SQL
+        end
+
+        def untrack_record_deletions(table)
+          drop_trigger(table, record_deletion_trigger_name(table))
+        end
+
+        private
+
+        def record_deletion_trigger_name(table)
+          "#{table}_loose_fk_trigger"
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/git/user.rb b/lib/gitlab/git/user.rb
index 05ae3391040..0798cc51055 100644
--- a/lib/gitlab/git/user.rb
+++ b/lib/gitlab/git/user.rb
@@ -6,7 +6,7 @@ module Gitlab
       attr_reader :username, :name, :email, :gl_id, :timezone
 
       def self.from_gitlab(gitlab_user)
-        new(gitlab_user.username, gitlab_user.name, gitlab_user.commit_email, Gitlab::GlId.gl_id(gitlab_user), gitlab_user.timezone)
+        new(gitlab_user.username, gitlab_user.name, gitlab_user.commit_email_or_default, Gitlab::GlId.gl_id(gitlab_user), gitlab_user.timezone)
       end
 
       def self.from_gitaly(gitaly_user)
diff --git a/lib/gitlab/sidekiq_logging/structured_logger.rb b/lib/gitlab/sidekiq_logging/structured_logger.rb
index 842e53b2ffb..1aebce987fe 100644
--- a/lib/gitlab/sidekiq_logging/structured_logger.rb
+++ b/lib/gitlab/sidekiq_logging/structured_logger.rb
@@ -69,6 +69,7 @@ module Gitlab
         message = base_message(payload)
 
         payload['load_balancing_strategy'] = job['load_balancing_strategy'] if job['load_balancing_strategy']
+        payload['dedup_wal_locations'] = job['dedup_wal_locations'] if job['dedup_wal_locations'].present?
 
         if job_exception
           payload['message'] = "#{message}: fail: #{payload['duration_s']} sec"
diff --git a/lib/gitlab/sidekiq_middleware/duplicate_jobs/duplicate_job.rb b/lib/gitlab/sidekiq_middleware/duplicate_jobs/duplicate_job.rb
index c1dc616cbb2..aeb58d7c153 100644
--- a/lib/gitlab/sidekiq_middleware/duplicate_jobs/duplicate_job.rb
+++ b/lib/gitlab/sidekiq_middleware/duplicate_jobs/duplicate_job.rb
@@ -17,10 +17,26 @@ module Gitlab
       #
       # When new jobs can be scheduled again, the strategy calls `#delete`.
       class DuplicateJob
+        include Gitlab::Utils::StrongMemoize
+
         DUPLICATE_KEY_TTL = 6.hours
+        WAL_LOCATION_TTL = 60.seconds
+        MAX_REDIS_RETRIES = 5
         DEFAULT_STRATEGY = :until_executing
         STRATEGY_NONE = :none
 
+        LUA_SET_WAL_SCRIPT = <<~EOS
+          local key, wal, offset, ttl = KEYS[1], ARGV[1], tonumber(ARGV[2]), ARGV[3]
+          local existing_offset = redis.call("LINDEX", key, -1)
+          if existing_offset == false then
+            redis.call("RPUSH", key, wal, offset)
+            redis.call("EXPIRE", key, ttl)
+          elseif offset > tonumber(existing_offset) then
+            redis.call("LSET", key, 0, wal)
+            redis.call("LSET", key, -1, offset)
+          end
+        EOS
+
         attr_reader :existing_jid
 
         def initialize(job, queue_name)
@@ -44,22 +60,59 @@ module Gitlab
         # This method will return the jid that was set in redis
         def check!(expiry = DUPLICATE_KEY_TTL)
           read_jid = nil
+          read_wal_locations = {}
 
           Sidekiq.redis do |redis|
             redis.multi do |multi|
               redis.set(idempotency_key, jid, ex: expiry, nx: true)
+              read_wal_locations = check_existing_wal_locations!(redis, expiry)
               read_jid = redis.get(idempotency_key)
             end
           end
 
           job['idempotency_key'] = idempotency_key
 
+          # We need to fetch values since the read_wal_locations and read_jid were obtained inside transaction, under redis.multi command.
+          self.existing_wal_locations = read_wal_locations.transform_values(&:value)
           self.existing_jid = read_jid.value
         end
 
+        def update_latest_wal_location!
+          return unless job_wal_locations.present?
+
+          Sidekiq.redis do |redis|
+            redis.multi do
+              job_wal_locations.each do |connection_name, location|
+                redis.eval(LUA_SET_WAL_SCRIPT, keys: [wal_location_key(connection_name)], argv: [location, pg_wal_lsn_diff(connection_name).to_i, WAL_LOCATION_TTL])
+              end
+            end
+          end
+        end
+
+        def latest_wal_locations
+          return {} unless job_wal_locations.present?
+
+          strong_memoize(:latest_wal_locations) do
+            read_wal_locations = {}
+
+            Sidekiq.redis do |redis|
+              redis.multi do
+                job_wal_locations.keys.each do |connection_name|
+                  read_wal_locations[connection_name] = redis.lindex(wal_location_key(connection_name), 0)
+                end
+              end
+            end
+
+            read_wal_locations.transform_values(&:value).compact
+          end
+        end
+
         def delete!
           Sidekiq.redis do |redis|
-            redis.del(idempotency_key)
+            redis.multi do |multi|
+              redis.del(idempotency_key)
+              delete_wal_locations!(redis)
+            end
           end
         end
 
@@ -93,6 +146,7 @@ module Gitlab
 
         private
 
+        attr_accessor :existing_wal_locations
         attr_reader :queue_name, :job
         attr_writer :existing_jid
 
@@ -100,6 +154,10 @@ module Gitlab
           @worker_klass ||= worker_class_name.to_s.safe_constantize
         end
 
+        def pg_wal_lsn_diff(connection_name)
+          Gitlab::Database::DATABASES[connection_name].pg_wal_lsn_diff(job_wal_locations[connection_name], existing_wal_locations[connection_name])
+        end
+
         def strategy
           return DEFAULT_STRATEGY unless worker_klass
           return DEFAULT_STRATEGY unless worker_klass.respond_to?(:idempotent?)
@@ -120,6 +178,20 @@ module Gitlab
           job['jid']
         end
 
+        def job_wal_locations
+          return {} unless preserve_wal_location?
+
+          job['wal_locations'] || {}
+        end
+
+        def existing_wal_location_key(connection_name)
+          "#{idempotency_key}:#{connection_name}:existing_wal_location"
+        end
+
+        def wal_location_key(connection_name)
+          "#{idempotency_key}:#{connection_name}:wal_location"
+        end
+
         def idempotency_key
           @idempotency_key ||= job['idempotency_key'] || "#{namespace}:#{idempotency_hash}"
         end
@@ -135,6 +207,29 @@ module Gitlab
         def idempotency_string
           "#{worker_class_name}:#{Sidekiq.dump_json(arguments)}"
         end
+
+        def delete_wal_locations!(redis)
+          job_wal_locations.keys.each do |connection_name|
+            redis.del(wal_location_key(connection_name))
+            redis.del(existing_wal_location_key(connection_name))
+          end
+        end
+
+        def check_existing_wal_locations!(redis, expiry)
+          read_wal_locations = {}
+
+          job_wal_locations.each do |connection_name, location|
+            key = existing_wal_location_key(connection_name)
+            redis.set(key, location, ex: expiry, nx: true)
+            read_wal_locations[connection_name] = redis.get(key)
+          end
+
+          read_wal_locations
+        end
+
+        def preserve_wal_location?
+          Feature.enabled?(:preserve_latest_wal_locations_for_idempotent_jobs, default_enabled: :yaml)
+        end
       end
     end
   end
diff --git a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/deduplicates_when_scheduling.rb b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/deduplicates_when_scheduling.rb
index 469033a5e52..fc58d4f5323 100644
--- a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/deduplicates_when_scheduling.rb
+++ b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/deduplicates_when_scheduling.rb
@@ -14,6 +14,8 @@ module Gitlab
               job['duplicate-of'] = duplicate_job.existing_jid
 
               if duplicate_job.idempotent?
+                duplicate_job.update_latest_wal_location!
+
                 Gitlab::SidekiqLogging::DeduplicationLogger.instance.log(
                   job, "dropped #{strategy_name}", duplicate_job.options)
                 return false
@@ -23,8 +25,16 @@ module Gitlab
             yield
           end
 
+          def perform(job)
+            update_job_wal_location!(job)
+          end
+
           private
 
+          def update_job_wal_location!(job)
+            job['dedup_wal_locations'] = duplicate_job.latest_wal_locations if duplicate_job.latest_wal_locations.present?
+          end
+
           def deduplicatable_job?
             !duplicate_job.scheduled? || duplicate_job.options[:including_scheduled]
           end
diff --git a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executed.rb b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executed.rb
index 738efa36fc8..5164b994267 100644
--- a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executed.rb
+++ b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executed.rb
@@ -8,9 +8,14 @@ module Gitlab
         # removes the lock after the job has executed preventing a new job to be queued
         # while a job is still executing.
         class UntilExecuted < Base
+          extend ::Gitlab::Utils::Override
+
           include DeduplicatesWhenScheduling
 
-          def perform(_job)
+          override :perform
+          def perform(job)
+            super
+
             yield
 
             duplicate_job.delete!
diff --git a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executing.rb b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executing.rb
index 68d66383b2b..1f7e3a4ea30 100644
--- a/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executing.rb
+++ b/lib/gitlab/sidekiq_middleware/duplicate_jobs/strategies/until_executing.rb
@@ -8,9 +8,13 @@ module Gitlab
         # removes the lock before the job starts allowing a new job to be queued
         # while a job is still executing.
         class UntilExecuting < Base
+          extend ::Gitlab::Utils::Override
+
           include DeduplicatesWhenScheduling
 
-          def perform(_job)
+          override :perform
+          def perform(job)
+            super
             duplicate_job.delete!
 
             yield