diff --git a/app/models/hooks/project_hook.rb b/app/models/hooks/project_hook.rb
index 16b95d2a2b999834a66099a3edc233e2f05ca245..9f45160d3a8d839258dbaaea193e78563b516016 100644
--- a/app/models/hooks/project_hook.rb
+++ b/app/models/hooks/project_hook.rb
@@ -41,6 +41,11 @@ def application_context
     super.merge(project: project)
   end
 
+  override :parent
+  def parent
+    project
+  end
+
   private
 
   override :web_hooks_disable_failed?
diff --git a/app/models/hooks/service_hook.rb b/app/models/hooks/service_hook.rb
index 1a466b333a55598bf62c4649af9148dbdd9caa46..80e167b350b5a6341b6adf035fb0186dd12fc869 100644
--- a/app/models/hooks/service_hook.rb
+++ b/app/models/hooks/service_hook.rb
@@ -2,6 +2,7 @@
 
 class ServiceHook < WebHook
   include Presentable
+  extend ::Gitlab::Utils::Override
 
   belongs_to :integration, foreign_key: :service_id
   validates :integration, presence: true
@@ -9,4 +10,7 @@ class ServiceHook < WebHook
   def execute(data, hook_name = 'service_hook')
     super(data, hook_name)
   end
+
+  override :parent
+  delegate :parent, to: :integration
 end
diff --git a/app/models/hooks/web_hook.rb b/app/models/hooks/web_hook.rb
index e8a55abfc8fdd404a5d080482ea9563d7da581dc..3320c13e87bc5c0bf50b8eaf681c95e44dc88f0f 100644
--- a/app/models/hooks/web_hook.rb
+++ b/app/models/hooks/web_hook.rb
@@ -122,6 +122,11 @@ def rate_limit
     nil
   end
 
+  # Returns the associated Project or Group for the WebHook if one exists.
+  # Overridden by inheriting classes.
+  def parent
+  end
+
   # Custom attributes to be included in the worker context.
   def application_context
     { related_class: type }
diff --git a/app/services/web_hook_service.rb b/app/services/web_hook_service.rb
index 79bdf34392f28150b24467457034ec666dd2d2b6..33e34ec41e2a4b4e689525afe04ab2cd2dd8a0e6 100644
--- a/app/services/web_hook_service.rb
+++ b/app/services/web_hook_service.rb
@@ -26,7 +26,6 @@ def initialize
   end
 
   REQUEST_BODY_SIZE_LIMIT = 25.megabytes
-  GITLAB_EVENT_HEADER = 'X-Gitlab-Event'
 
   attr_accessor :hook, :data, :hook_name, :request_options
   attr_reader :uniqueness_token
@@ -50,6 +49,10 @@ def initialize(hook, data, hook_name, uniqueness_token = nil)
   def execute
     return { status: :error, message: 'Hook disabled' } unless hook.executable?
 
+    log_recursion_limit if recursion_blocked?
+
+    Gitlab::WebHooks::RecursionDetection.register!(hook)
+
     start_time = Gitlab::Metrics::System.monotonic_time
 
     response = if parsed_url.userinfo.blank?
@@ -95,6 +98,10 @@ def async_execute
     Gitlab::ApplicationContext.with_context(hook.application_context) do
       break log_rate_limit if rate_limited?
 
+      log_recursion_limit if recursion_blocked?
+
+      data[:_gitlab_recursion_detection_request_uuid] = Gitlab::WebHooks::RecursionDetection::UUID.instance.request_uuid
+
       WebHookWorker.perform_async(hook.id, data, hook_name)
     end
   end
@@ -108,7 +115,7 @@ def parsed_url
   def make_request(url, basic_auth = false)
     Gitlab::HTTP.post(url,
       body: Gitlab::Json::LimitedEncoder.encode(data, limit: REQUEST_BODY_SIZE_LIMIT),
-      headers: build_headers(hook_name),
+      headers: build_headers,
       verify: hook.enable_ssl_verification,
       basic_auth: basic_auth,
       **request_options)
@@ -129,7 +136,7 @@ def log_execution(trigger:, url:, request_data:, response:, execution_duration:,
       trigger: trigger,
       url: url,
       execution_duration: execution_duration,
-      request_headers: build_headers(hook_name),
+      request_headers: build_headers,
       request_data: request_data,
       response_headers: format_response_headers(response),
       response_body: safe_response_body(response),
@@ -151,15 +158,16 @@ def response_category(response)
     end
   end
 
-  def build_headers(hook_name)
+  def build_headers
     @headers ||= begin
-      {
+      headers = {
         'Content-Type' => 'application/json',
         'User-Agent' => "GitLab/#{Gitlab::VERSION}",
-        GITLAB_EVENT_HEADER => self.class.hook_to_event(hook_name)
-      }.tap do |hash|
-        hash['X-Gitlab-Token'] = Gitlab::Utils.remove_line_breaks(hook.token) if hook.token.present?
-      end
+        Gitlab::WebHooks::GITLAB_EVENT_HEADER => self.class.hook_to_event(hook_name)
+      }
+
+      headers['X-Gitlab-Token'] = Gitlab::Utils.remove_line_breaks(hook.token) if hook.token.present?
+      headers.merge!(Gitlab::WebHooks::RecursionDetection.header(hook))
     end
   end
 
@@ -186,6 +194,10 @@ def rate_limited?
     )
   end
 
+  def recursion_blocked?
+    Gitlab::WebHooks::RecursionDetection.block?(hook)
+  end
+
   def rate_limit
     @rate_limit ||= hook.rate_limit
   end
@@ -199,4 +211,15 @@ def log_rate_limit
       **Gitlab::ApplicationContext.current
     )
   end
+
+  def log_recursion_limit
+    Gitlab::AuthLogger.error(
+      message: 'Webhook recursion detected and will be blocked in future',
+      hook_id: hook.id,
+      hook_type: hook.type,
+      hook_name: hook_name,
+      recursion_detection: ::Gitlab::WebHooks::RecursionDetection.to_log(hook),
+      **Gitlab::ApplicationContext.current
+    )
+  end
 end
diff --git a/app/workers/web_hook_worker.rb b/app/workers/web_hook_worker.rb
index 5b4567dde29dadfcb1134fbe72d87222e009323f..952ac94d5e694664dbab95e2f395c9103e2719e1 100644
--- a/app/workers/web_hook_worker.rb
+++ b/app/workers/web_hook_worker.rb
@@ -13,11 +13,21 @@ class WebHookWorker
 
   worker_has_external_dependencies!
 
-  def perform(hook_id, data, hook_name)
+  # Webhook recursion detection properties are passed through the `data` arg.
+  # This will be migrated to the `params` arg over the next few releases.
+  # See https://gitlab.com/gitlab-org/gitlab/-/issues/347389.
+  def perform(hook_id, data, hook_name, params = {})
     hook = WebHook.find_by_id(hook_id)
     return unless hook
 
     data = data.with_indifferent_access
+
+    # Before executing the hook, reapply any recursion detection UUID that was
+    # initially present in the request header so the hook can pass this same header
+    # value in its request.
+    recursion_detection_uuid = data.delete(:_gitlab_recursion_detection_request_uuid)
+    Gitlab::WebHooks::RecursionDetection.set_request_uuid(recursion_detection_uuid)
+
     WebHookService.new(hook, data, hook_name, jid).execute
   end
 end
diff --git a/config/feature_flags/development/webhook_recursion_detection.yml b/config/feature_flags/development/webhook_recursion_detection.yml
new file mode 100644
index 0000000000000000000000000000000000000000..8ef0bf81491d415baa128b1755f5bbe1af425308
--- /dev/null
+++ b/config/feature_flags/development/webhook_recursion_detection.yml
@@ -0,0 +1,8 @@
+---
+name: webhook_recursion_detection
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/75821
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/349845
+milestone: '14.7'
+type: development
+group: group::integrations
+default_enabled: false
diff --git a/config/initializers/webhook_recursion_detection.rb b/config/initializers/webhook_recursion_detection.rb
new file mode 100644
index 0000000000000000000000000000000000000000..b345c005bace74804b153ffed6e350da5cdce5a6
--- /dev/null
+++ b/config/initializers/webhook_recursion_detection.rb
@@ -0,0 +1,5 @@
+# frozen_string_literal: true
+
+Rails.application.configure do |config|
+  config.middleware.insert_after RequestStore::Middleware, Gitlab::Middleware::WebhookRecursionDetection
+end
diff --git a/ee/app/models/hooks/group_hook.rb b/ee/app/models/hooks/group_hook.rb
index 3834c213d7ba584b237f25c2a2d5a561a57172ad..b681551144bdcb5609be687dc68503c9eeeba364 100644
--- a/ee/app/models/hooks/group_hook.rb
+++ b/ee/app/models/hooks/group_hook.rb
@@ -44,6 +44,11 @@ def rate_limit
     group.actual_limits.limit_for(:web_hook_calls)
   end
 
+  override :parent
+  def parent
+    group
+  end
+
   private
 
   override :web_hooks_disable_failed?
diff --git a/ee/spec/models/hooks/group_hook_spec.rb b/ee/spec/models/hooks/group_hook_spec.rb
index ef137b964e599c9bd8aae1db4b8667d82ad30cac..fdf036e1b0b0c05b1638169bede0b64a95071d5f 100644
--- a/ee/spec/models/hooks/group_hook_spec.rb
+++ b/ee/spec/models/hooks/group_hook_spec.rb
@@ -30,6 +30,15 @@
     end
   end
 
+  describe '#parent' do
+    it 'returns the associated group' do
+      group = build(:group)
+      hook = build(:group_hook, group: group)
+
+      expect(hook.parent).to eq(group)
+    end
+  end
+
   describe '#application_context' do
     let_it_be(:hook) { build(:group_hook) }
 
diff --git a/lib/api/ci/triggers.rb b/lib/api/ci/triggers.rb
index 6a2b16e1568f5fad096ce3a15064649a1284f6cf..ae89b475ef804615af2ad0614c2e9e06dabd8fff 100644
--- a/lib/api/ci/triggers.rb
+++ b/lib/api/ci/triggers.rb
@@ -5,7 +5,7 @@ module Ci
     class Triggers < ::API::Base
       include PaginationParams
 
-      HTTP_GITLAB_EVENT_HEADER = "HTTP_#{WebHookService::GITLAB_EVENT_HEADER}".underscore.upcase
+      HTTP_GITLAB_EVENT_HEADER = "HTTP_#{::Gitlab::WebHooks::GITLAB_EVENT_HEADER}".underscore.upcase
 
       feature_category :continuous_integration
 
diff --git a/lib/gitlab/middleware/webhook_recursion_detection.rb b/lib/gitlab/middleware/webhook_recursion_detection.rb
new file mode 100644
index 0000000000000000000000000000000000000000..2677445852ceb416bf3d7ca3f86d9334d95d1a53
--- /dev/null
+++ b/lib/gitlab/middleware/webhook_recursion_detection.rb
@@ -0,0 +1,19 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module Middleware
+    class WebhookRecursionDetection
+      def initialize(app)
+        @app = app
+      end
+
+      def call(env)
+        headers = ActionDispatch::Request.new(env).headers
+
+        ::Gitlab::WebHooks::RecursionDetection.set_from_headers(headers)
+
+        @app.call(env)
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/web_hooks.rb b/lib/gitlab/web_hooks.rb
new file mode 100644
index 0000000000000000000000000000000000000000..349c7a020cce030b72cb0ea13b8bea46d9fda42a
--- /dev/null
+++ b/lib/gitlab/web_hooks.rb
@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module WebHooks
+    GITLAB_EVENT_HEADER = 'X-Gitlab-Event'
+  end
+end
diff --git a/lib/gitlab/web_hooks/recursion_detection.rb b/lib/gitlab/web_hooks/recursion_detection.rb
new file mode 100644
index 0000000000000000000000000000000000000000..34f70093dcd965f4adf47c5db0557b45515fd60f
--- /dev/null
+++ b/lib/gitlab/web_hooks/recursion_detection.rb
@@ -0,0 +1,102 @@
+# frozen_string_literal: true
+
+# This module detects and blocks recursive webhook requests.
+#
+# Recursion can happen when a webhook has been configured to make a call
+# to its own GitLab instance (i.e., its API), and during the execution of
+# the call the webhook is triggered again to create an infinite loop of
+# being triggered.
+#
+# Additionally the module blocks a webhook once the number of requests to
+# the instance made by a series of webhooks triggering other webhooks reaches
+# a limit.
+#
+# Blocking recursive webhooks allows GitLab to continue to support workflows
+# that use webhooks to call the API non-recursively, or do not go on to
+# trigger an unreasonable number of other webhooks.
+module Gitlab
+  module WebHooks
+    module RecursionDetection
+      COUNT_LIMIT = 100
+      TOUCH_CACHE_TTL = 30.minutes
+
+      class << self
+        def set_from_headers(headers)
+          uuid = headers[UUID::HEADER]
+
+          return unless uuid
+
+          set_request_uuid(uuid)
+        end
+
+        def set_request_uuid(uuid)
+          UUID.instance.request_uuid = uuid
+        end
+
+        # Before a webhook is executed, `.register!` should be called.
+        # Adds the webhook ID to a cache (see `#cache_key_for_hook` for
+        # details of the cache).
+        def register!(hook)
+          return if disabled?(hook)
+
+          cache_key = cache_key_for_hook(hook)
+
+          ::Gitlab::Redis::SharedState.with do |redis|
+            redis.multi do
+              redis.sadd(cache_key, hook.id)
+              redis.expire(cache_key, TOUCH_CACHE_TTL)
+            end
+          end
+        end
+
+        # Returns true if the webhook ID is present in the cache, or if the
+        # number of IDs in the cache exceeds the limit (see
+        # `#cache_key_for_hook` for details of the cache).
+        def block?(hook)
+          return false if disabled?(hook)
+
+          # If a request UUID has not been set then we know the request was not
+          # made by a webhook, and no recursion is possible.
+          return false unless UUID.instance.request_uuid
+
+          cache_key = cache_key_for_hook(hook)
+
+          ::Gitlab::Redis::SharedState.with do |redis|
+            redis.sismember(cache_key, hook.id) ||
+              redis.scard(cache_key) >= COUNT_LIMIT
+          end
+        end
+
+        def header(hook)
+          UUID.instance.header(hook)
+        end
+
+        def to_log(hook)
+          {
+            uuid: UUID.instance.uuid_for_hook(hook),
+            ids: ::Gitlab::Redis::SharedState.with { |redis| redis.smembers(cache_key_for_hook(hook)).map(&:to_i) }
+          }
+        end
+
+        private
+
+        def disabled?(hook)
+          Feature.disabled?(:webhook_recursion_detection, hook.parent)
+        end
+
+        # Returns a cache key scoped to a UUID.
+        #
+        # The particular UUID will be either:
+        #
+        #   - A UUID that was recycled from the request headers if the request was made by a webhook.
+        #   - a new UUID initialized for the webhook.
+        #
+        # This means that cycles of webhooks that are triggered from other webhooks
+        # will share the same cache, and other webhooks will use a new cache.
+        def cache_key_for_hook(hook)
+          [:webhook_recursion_detection, UUID.instance.uuid_for_hook(hook)].join(':')
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/web_hooks/recursion_detection/uuid.rb b/lib/gitlab/web_hooks/recursion_detection/uuid.rb
new file mode 100644
index 0000000000000000000000000000000000000000..9c52399818de47e81f12bfed6b19416497acfecb
--- /dev/null
+++ b/lib/gitlab/web_hooks/recursion_detection/uuid.rb
@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module WebHooks
+    module RecursionDetection
+      class UUID
+        HEADER = "#{::Gitlab::WebHooks::GITLAB_EVENT_HEADER}-UUID"
+
+        include Singleton
+
+        attr_accessor :request_uuid
+
+        def initialize
+          self.new_uuids_for_hooks = {}
+        end
+
+        class << self
+          # Back the Singleton with RequestStore so it is isolated to this request.
+          def instance
+            Gitlab::SafeRequestStore[:web_hook_recursion_detection_uuid] ||= new
+          end
+        end
+
+        # Returns a UUID, which will be either:
+        #
+        #   - The UUID that was recycled from the request headers if the request was made by a webhook.
+        #   - A new UUID initialized for the webhook.
+        def uuid_for_hook(hook)
+          request_uuid || new_uuid_for_hook(hook)
+        end
+
+        def header(hook)
+          { HEADER => uuid_for_hook(hook) }
+        end
+
+        private
+
+        attr_accessor :new_uuids_for_hooks
+
+        def new_uuid_for_hook(hook)
+          new_uuids_for_hooks[hook.id] ||= SecureRandom.uuid
+        end
+      end
+    end
+  end
+end
diff --git a/spec/lib/gitlab/middleware/webhook_recursion_detection_spec.rb b/spec/lib/gitlab/middleware/webhook_recursion_detection_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..c8dbc990f8ca8b03d4e1f9f3b85c3d86b1a8ce5a
--- /dev/null
+++ b/spec/lib/gitlab/middleware/webhook_recursion_detection_spec.rb
@@ -0,0 +1,42 @@
+# frozen_string_literal: true
+
+require 'fast_spec_helper'
+require 'action_dispatch'
+require 'rack'
+require 'request_store'
+
+RSpec.describe Gitlab::Middleware::WebhookRecursionDetection do
+  let(:app) { double(:app) }
+  let(:middleware) { described_class.new(app) }
+  let(:env) { Rack::MockRequest.env_for("/").merge(headers) }
+
+  around do |example|
+    Gitlab::WithRequestStore.with_request_store { example.run }
+  end
+
+  describe '#call' do
+    subject(:call) { described_class.new(app).call(env) }
+
+    context 'when the recursion detection header is present' do
+      let(:new_uuid) { SecureRandom.uuid }
+      let(:headers) { { 'HTTP_X_GITLAB_EVENT_UUID' => new_uuid } }
+
+      it 'sets the request UUID from the header' do
+        expect(app).to receive(:call)
+        expect { call }.to change { Gitlab::WebHooks::RecursionDetection::UUID.instance.request_uuid }.to(new_uuid)
+      end
+    end
+
+    context 'when recursion headers are not present' do
+      let(:headers) { {} }
+
+      it 'works without errors' do
+        expect(app).to receive(:call)
+
+        call
+
+        expect(Gitlab::WebHooks::RecursionDetection::UUID.instance.request_uuid).to be_nil
+      end
+    end
+  end
+end
diff --git a/spec/lib/gitlab/web_hooks/recursion_detection_spec.rb b/spec/lib/gitlab/web_hooks/recursion_detection_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..4f92aa8b3d120e468a1e5960fd10c6e7e9356dd6
--- /dev/null
+++ b/spec/lib/gitlab/web_hooks/recursion_detection_spec.rb
@@ -0,0 +1,243 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::WebHooks::RecursionDetection, :clean_gitlab_redis_shared_state, :request_store do
+  let_it_be(:web_hook) { create(:project_hook) }
+
+  let!(:uuid_class) { described_class::UUID }
+
+  describe '.set_from_headers' do
+    let(:old_uuid) { SecureRandom.uuid }
+    let(:rack_headers) { Rack::MockRequest.env_for("/").merge(headers) }
+
+    subject(:set_from_headers) { described_class.set_from_headers(rack_headers) }
+
+    # Note, having a previous `request_uuid` value set before `.set_from_headers` is
+    # called is contrived and should not normally happen. However, testing with this scenario
+    # allows us to assert the ideal outcome if it ever were to happen.
+    before do
+      uuid_class.instance.request_uuid = old_uuid
+    end
+
+    context 'when the detection header is present' do
+      let(:new_uuid) { SecureRandom.uuid }
+
+      let(:headers) do
+        { uuid_class::HEADER => new_uuid }
+      end
+
+      it 'sets the request UUID value from the headers' do
+        set_from_headers
+
+        expect(uuid_class.instance.request_uuid).to eq(new_uuid)
+      end
+    end
+
+    context 'when detection header is not present' do
+      let(:headers) { {} }
+
+      it 'does not set the request UUID' do
+        set_from_headers
+
+        expect(uuid_class.instance.request_uuid).to eq(old_uuid)
+      end
+    end
+  end
+
+  describe '.set_request_uuid' do
+    it 'sets the request UUID value' do
+      new_uuid = SecureRandom.uuid
+
+      described_class.set_request_uuid(new_uuid)
+
+      expect(uuid_class.instance.request_uuid).to eq(new_uuid)
+    end
+  end
+
+  describe '.register!' do
+    let_it_be(:second_web_hook) { create(:project_hook) }
+    let_it_be(:third_web_hook) { create(:project_hook) }
+
+    def cache_key(hook)
+      described_class.send(:cache_key_for_hook, hook)
+    end
+
+    it 'stores IDs in the same cache when a request UUID is set, until the request UUID changes', :aggregate_failures do
+      # Register web_hook and second_web_hook against the same request UUID.
+      uuid_class.instance.request_uuid = SecureRandom.uuid
+      described_class.register!(web_hook)
+      described_class.register!(second_web_hook)
+      first_cache_key = cache_key(web_hook)
+      second_cache_key = cache_key(second_web_hook)
+
+      # Register third_web_hook against a new request UUID.
+      uuid_class.instance.request_uuid = SecureRandom.uuid
+      described_class.register!(third_web_hook)
+      third_cache_key = cache_key(third_web_hook)
+
+      expect(first_cache_key).to eq(second_cache_key)
+      expect(second_cache_key).not_to eq(third_cache_key)
+
+      ::Gitlab::Redis::SharedState.with do |redis|
+        members = redis.smembers(first_cache_key).map(&:to_i)
+        expect(members).to contain_exactly(web_hook.id, second_web_hook.id)
+
+        members = redis.smembers(third_cache_key).map(&:to_i)
+        expect(members).to contain_exactly(third_web_hook.id)
+      end
+    end
+
+    it 'stores IDs in unique caches when no request UUID is present', :aggregate_failures do
+      described_class.register!(web_hook)
+      described_class.register!(second_web_hook)
+      described_class.register!(third_web_hook)
+
+      first_cache_key = cache_key(web_hook)
+      second_cache_key = cache_key(second_web_hook)
+      third_cache_key = cache_key(third_web_hook)
+
+      expect([first_cache_key, second_cache_key, third_cache_key].compact.length).to eq(3)
+
+      ::Gitlab::Redis::SharedState.with do |redis|
+        members = redis.smembers(first_cache_key).map(&:to_i)
+        expect(members).to contain_exactly(web_hook.id)
+
+        members = redis.smembers(second_cache_key).map(&:to_i)
+        expect(members).to contain_exactly(second_web_hook.id)
+
+        members = redis.smembers(third_cache_key).map(&:to_i)
+        expect(members).to contain_exactly(third_web_hook.id)
+      end
+    end
+
+    it 'touches the storage ttl each time it is called', :aggregate_failures do
+      freeze_time do
+        described_class.register!(web_hook)
+
+        ::Gitlab::Redis::SharedState.with do |redis|
+          expect(redis.ttl(cache_key(web_hook))).to eq(described_class::TOUCH_CACHE_TTL.to_i)
+        end
+      end
+
+      travel_to(1.minute.from_now) do
+        described_class.register!(second_web_hook)
+
+        ::Gitlab::Redis::SharedState.with do |redis|
+          expect(redis.ttl(cache_key(web_hook))).to eq(described_class::TOUCH_CACHE_TTL.to_i)
+        end
+      end
+    end
+
+    it 'does not store anything if the feature flag is disabled' do
+      stub_feature_flags(webhook_recursion_detection: false)
+
+      described_class.register!(web_hook)
+
+      ::Gitlab::Redis::SharedState.with do |redis|
+        expect(redis.exists(cache_key(web_hook))).to eq(false)
+      end
+    end
+  end
+
+  describe 'block?' do
+    let_it_be(:registered_web_hooks) { create_list(:project_hook, 2) }
+
+    subject(:block?) { described_class.block?(web_hook) }
+
+    before do
+      # Register some previous webhooks.
+      uuid_class.instance.request_uuid = SecureRandom.uuid
+
+      registered_web_hooks.each do |web_hook|
+        described_class.register!(web_hook)
+      end
+    end
+
+    it 'returns false if webhook should not be blocked' do
+      is_expected.to eq(false)
+    end
+
+    context 'when the webhook has previously fired' do
+      before do
+        described_class.register!(web_hook)
+      end
+
+      it 'returns true' do
+        is_expected.to eq(true)
+      end
+
+      it 'returns false if the feature flag is disabled' do
+        stub_feature_flags(webhook_recursion_detection: false)
+
+        is_expected.to eq(false)
+      end
+
+      context 'when the request UUID changes again' do
+        before do
+          uuid_class.instance.request_uuid = SecureRandom.uuid
+        end
+
+        it 'returns false' do
+          is_expected.to eq(false)
+        end
+      end
+    end
+
+    context 'when the count limit has been reached' do
+      let_it_be(:registered_web_hooks) { create_list(:project_hook, 2) }
+
+      before do
+        registered_web_hooks.each do |web_hook|
+          described_class.register!(web_hook)
+        end
+
+        stub_const("#{described_class.name}::COUNT_LIMIT", registered_web_hooks.size)
+      end
+
+      it 'returns true' do
+        is_expected.to eq(true)
+      end
+
+      it 'returns false if the feature flag is disabled' do
+        stub_feature_flags(webhook_recursion_detection: false)
+
+        is_expected.to eq(false)
+      end
+
+      context 'when the request UUID changes again' do
+        before do
+          uuid_class.instance.request_uuid = SecureRandom.uuid
+        end
+
+        it 'returns false' do
+          is_expected.to eq(false)
+        end
+      end
+    end
+  end
+
+  describe '.header' do
+    subject(:header) { described_class.header(web_hook) }
+
+    it 'returns a header with the UUID value' do
+      uuid = SecureRandom.uuid
+      allow(uuid_class.instance).to receive(:uuid_for_hook).and_return(uuid)
+
+      is_expected.to eq({ uuid_class::HEADER => uuid })
+    end
+  end
+
+  describe '.to_log' do
+    subject(:to_log) { described_class.to_log(web_hook) }
+
+    it 'returns the UUID value and all registered webhook IDs in a Hash' do
+      uuid = SecureRandom.uuid
+      allow(uuid_class.instance).to receive(:uuid_for_hook).and_return(uuid)
+      registered_web_hooks = create_list(:project_hook, 2)
+      registered_web_hooks.each { described_class.register!(_1) }
+
+      is_expected.to eq({ uuid: uuid, ids: registered_web_hooks.map(&:id) })
+    end
+  end
+end
diff --git a/spec/models/hooks/project_hook_spec.rb b/spec/models/hooks/project_hook_spec.rb
index f0ee9a613d88e9222a7a53d5a9dc2f30d83ed6ba..ec2eca9675597e779b0953d1379d382cbc2a38be 100644
--- a/spec/models/hooks/project_hook_spec.rb
+++ b/spec/models/hooks/project_hook_spec.rb
@@ -40,6 +40,15 @@
     end
   end
 
+  describe '#parent' do
+    it 'returns the associated project' do
+      project = build(:project)
+      hook = build(:project_hook, project: project)
+
+      expect(hook.parent).to eq(project)
+    end
+  end
+
   describe '#application_context' do
     let_it_be(:hook) { build(:project_hook) }
 
diff --git a/spec/models/hooks/service_hook_spec.rb b/spec/models/hooks/service_hook_spec.rb
index 4ce2e729d892b2bdf0e939974dd6542ba5c052ed..85f433f5f814b72169832ca78be0842e13b7fddb 100644
--- a/spec/models/hooks/service_hook_spec.rb
+++ b/spec/models/hooks/service_hook_spec.rb
@@ -31,6 +31,36 @@
     end
   end
 
+  describe '#parent' do
+    let(:hook) { build(:service_hook, integration: integration) }
+
+    context 'with a project-level integration' do
+      let(:project) { build(:project) }
+      let(:integration) { build(:integration, project: project) }
+
+      it 'returns the associated project' do
+        expect(hook.parent).to eq(project)
+      end
+    end
+
+    context 'with a group-level integration' do
+      let(:group) { build(:group) }
+      let(:integration) { build(:integration, :group, group: group) }
+
+      it 'returns the associated group' do
+        expect(hook.parent).to eq(group)
+      end
+    end
+
+    context 'with an instance-level integration' do
+      let(:integration) { build(:integration, :instance) }
+
+      it 'returns nil' do
+        expect(hook.parent).to be_nil
+      end
+    end
+  end
+
   describe '#application_context' do
     let(:hook) { build(:service_hook) }
 
diff --git a/spec/models/integrations/datadog_spec.rb b/spec/models/integrations/datadog_spec.rb
index 9c3ff7aa35ba4ab09b9bc169dcf168c2cccf26e3..0607505e7d54bb08a41c0a78e4a2170bb6a7ee81 100644
--- a/spec/models/integrations/datadog_spec.rb
+++ b/spec/models/integrations/datadog_spec.rb
@@ -167,7 +167,7 @@
 
     context 'with pipeline data' do
       let(:data) { pipeline_data }
-      let(:expected_headers) { { WebHookService::GITLAB_EVENT_HEADER => 'Pipeline Hook' } }
+      let(:expected_headers) { { ::Gitlab::WebHooks::GITLAB_EVENT_HEADER => 'Pipeline Hook' } }
       let(:expected_body) { data.with_retried_builds.to_json }
 
       it { expect(a_request(:post, expected_hook_url).with(headers: expected_headers, body: expected_body)).to have_been_made }
@@ -175,7 +175,7 @@
 
     context 'with job data' do
       let(:data) { build_data }
-      let(:expected_headers) { { WebHookService::GITLAB_EVENT_HEADER => 'Job Hook' } }
+      let(:expected_headers) { { ::Gitlab::WebHooks::GITLAB_EVENT_HEADER => 'Job Hook' } }
       let(:expected_body) { data.to_json }
 
       it { expect(a_request(:post, expected_hook_url).with(headers: expected_headers, body: expected_body)).to have_been_made }
diff --git a/spec/requests/api/ci/triggers_spec.rb b/spec/requests/api/ci/triggers_spec.rb
index d270a16d28d4cbc50ddee4ee3cf696d4f3bb0d2b..a036a55f5f316d8553764d4c74de774ea74c7860 100644
--- a/spec/requests/api/ci/triggers_spec.rb
+++ b/spec/requests/api/ci/triggers_spec.rb
@@ -162,7 +162,7 @@
         expect do
           post api("/projects/#{project.id}/ref/master/trigger/pipeline?token=#{trigger_token}"),
             params: { ref: 'refs/heads/other-branch' },
-            headers: { WebHookService::GITLAB_EVENT_HEADER => 'Pipeline Hook' }
+            headers: { ::Gitlab::WebHooks::GITLAB_EVENT_HEADER => 'Pipeline Hook' }
         end.not_to change(Ci::Pipeline, :count)
 
         expect(response).to have_gitlab_http_status(:forbidden)
diff --git a/spec/requests/recursive_webhook_detection_spec.rb b/spec/requests/recursive_webhook_detection_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..09a352da84ba0ec5d3a91b5d7363d6417326cc01
--- /dev/null
+++ b/spec/requests/recursive_webhook_detection_spec.rb
@@ -0,0 +1,200 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe 'Recursive webhook detection', :sidekiq_inline, :clean_gitlab_redis_shared_state, :request_store do
+  include StubRequests
+
+  let_it_be(:user) { create(:user) }
+  let_it_be(:project) { create(:project, :repository, namespace: user.namespace, creator: user) }
+  let_it_be(:merge_request) { create(:merge_request, source_project: project) }
+  let_it_be(:project_hook) { create(:project_hook, project: project, merge_requests_events: true) }
+  let_it_be(:system_hook) { create(:system_hook, merge_requests_events: true) }
+
+  # Trigger a change to the merge request to fire the webhooks.
+  def trigger_web_hooks
+    params = { merge_request: { description: FFaker::Lorem.sentence } }
+    put project_merge_request_path(project, merge_request), params: params, headers: headers
+  end
+
+  def stub_requests
+    stub_full_request(project_hook.url, method: :post, ip_address: '8.8.8.8')
+    stub_full_request(system_hook.url, method: :post, ip_address: '8.8.8.9')
+  end
+
+  before do
+    login_as(user)
+  end
+
+  context 'when the request headers include the recursive webhook detection header' do
+    let(:uuid) { SecureRandom.uuid }
+    let(:headers) { { Gitlab::WebHooks::RecursionDetection::UUID::HEADER => uuid } }
+
+    it 'executes all webhooks, logs no errors, and the webhook requests contain the same UUID header', :aggregate_failures do
+      stub_requests
+
+      expect(Gitlab::AuthLogger).not_to receive(:error)
+
+      trigger_web_hooks
+
+      expect(WebMock).to have_requested(:post, stubbed_hostname(project_hook.url))
+        .with { |req| req.headers['X-Gitlab-Event-Uuid'] == uuid }
+        .once
+      expect(WebMock).to have_requested(:post, stubbed_hostname(system_hook.url))
+        .with { |req| req.headers['X-Gitlab-Event-Uuid'] == uuid }
+        .once
+    end
+
+    shared_examples 'when the feature flag is disabled' do
+      it 'executes and logs no errors' do
+        stub_feature_flags(webhook_recursion_detection: false)
+        stub_requests
+
+        expect(Gitlab::AuthLogger).not_to receive(:error)
+
+        trigger_web_hooks
+
+        expect(WebMock).to have_requested(:post, stubbed_hostname(project_hook.url)).once
+        expect(WebMock).to have_requested(:post, stubbed_hostname(system_hook.url)).once
+      end
+    end
+
+    context 'when one of the webhooks is recursive' do
+      before do
+        # Recreate the necessary state for the previous request to be
+        # considered made from the webhook.
+        Gitlab::WebHooks::RecursionDetection.set_request_uuid(uuid)
+        Gitlab::WebHooks::RecursionDetection.register!(project_hook)
+        Gitlab::WebHooks::RecursionDetection.set_request_uuid(nil)
+      end
+
+      it 'executes all webhooks and logs an error for the recursive hook', :aggregate_failures do
+        stub_requests
+
+        expect(Gitlab::AuthLogger).to receive(:error).with(
+          include(
+            message: 'Webhook recursion detected and will be blocked in future',
+            hook_id: project_hook.id,
+            recursion_detection: {
+              uuid: uuid,
+              ids: [project_hook.id]
+            }
+          )
+        ).twice # Twice: once in `#async_execute`, and again in `#execute`.
+
+        trigger_web_hooks
+
+        expect(WebMock).to have_requested(:post, stubbed_hostname(project_hook.url)).once
+        expect(WebMock).to have_requested(:post, stubbed_hostname(system_hook.url)).once
+      end
+
+      include_examples 'when the feature flag is disabled'
+    end
+
+    context 'when the count limit has been reached' do
+      let_it_be(:previous_hooks) { create_list(:project_hook, 3) }
+
+      before do
+        stub_const('Gitlab::WebHooks::RecursionDetection::COUNT_LIMIT', 2)
+        # Recreate the necessary state for a number of previous webhooks to
+        # have been triggered previously.
+        Gitlab::WebHooks::RecursionDetection.set_request_uuid(uuid)
+        previous_hooks.each { Gitlab::WebHooks::RecursionDetection.register!(_1) }
+        Gitlab::WebHooks::RecursionDetection.set_request_uuid(nil)
+      end
+
+      it 'executes and logs errors for all hooks', :aggregate_failures do
+        stub_requests
+        previous_hook_ids = previous_hooks.map(&:id)
+
+        expect(Gitlab::AuthLogger).to receive(:error).with(
+          include(
+            message: 'Webhook recursion detected and will be blocked in future',
+            hook_id: project_hook.id,
+            recursion_detection: {
+              uuid: uuid,
+              ids: include(*previous_hook_ids)
+            }
+          )
+        ).twice
+        expect(Gitlab::AuthLogger).to receive(:error).with(
+          include(
+            message: 'Webhook recursion detected and will be blocked in future',
+            hook_id: system_hook.id,
+            recursion_detection: {
+              uuid: uuid,
+              ids: include(*previous_hook_ids)
+            }
+          )
+        ).twice
+
+        trigger_web_hooks
+
+        expect(WebMock).to have_requested(:post, stubbed_hostname(project_hook.url)).once
+        expect(WebMock).to have_requested(:post, stubbed_hostname(system_hook.url)).once
+      end
+    end
+
+    include_examples 'when the feature flag is disabled'
+  end
+
+  context 'when the recursive webhook detection header is absent' do
+    let(:headers) { {} }
+
+    let(:uuid_header_spy) do
+      Class.new do
+        attr_reader :values
+
+        def initialize
+          @values = []
+        end
+
+        def to_proc
+          proc do |method, *args|
+            method.call(*args).tap do |headers|
+              @values << headers[Gitlab::WebHooks::RecursionDetection::UUID::HEADER]
+            end
+          end
+        end
+      end.new
+    end
+
+    before do
+      allow(Gitlab::WebHooks::RecursionDetection).to receive(:header).at_least(:once).and_wrap_original(&uuid_header_spy)
+    end
+
+    it 'executes all webhooks, logs no errors, and the webhook requests contain different UUID headers', :aggregate_failures do
+      stub_requests
+
+      expect(Gitlab::AuthLogger).not_to receive(:error)
+
+      trigger_web_hooks
+
+      uuid_headers = uuid_header_spy.values
+
+      expect(uuid_headers).to all(be_present)
+      expect(uuid_headers.uniq.length).to eq(2)
+      expect(WebMock).to have_requested(:post, stubbed_hostname(project_hook.url))
+        .with { |req| uuid_headers.include?(req.headers['X-Gitlab-Event-Uuid']) }
+        .once
+      expect(WebMock).to have_requested(:post, stubbed_hostname(system_hook.url))
+        .with { |req| uuid_headers.include?(req.headers['X-Gitlab-Event-Uuid']) }
+        .once
+    end
+
+    it 'uses new UUID values between requests' do
+      stub_requests
+
+      trigger_web_hooks
+      trigger_web_hooks
+
+      uuid_headers = uuid_header_spy.values
+
+      expect(uuid_headers).to all(be_present)
+      expect(uuid_headers.length).to eq(4)
+      expect(uuid_headers.uniq.length).to eq(4)
+      expect(WebMock).to have_requested(:post, stubbed_hostname(project_hook.url)).twice
+      expect(WebMock).to have_requested(:post, stubbed_hostname(system_hook.url)).twice
+    end
+  end
+end
diff --git a/spec/services/web_hook_service_spec.rb b/spec/services/web_hook_service_spec.rb
index 2aebd2adab913703eb591219e4daa3ffe63ce4a2..7d933ea9c5cd0b1e252e0aae3b46c0de6090dc62 100644
--- a/spec/services/web_hook_service_spec.rb
+++ b/spec/services/web_hook_service_spec.rb
@@ -2,20 +2,12 @@
 
 require 'spec_helper'
 
-RSpec.describe WebHookService do
+RSpec.describe WebHookService, :request_store, :clean_gitlab_redis_shared_state do
   include StubRequests
 
   let_it_be(:project) { create(:project) }
   let_it_be_with_reload(:project_hook) { create(:project_hook, project: project) }
 
-  let(:headers) do
-    {
-      'Content-Type' => 'application/json',
-      'User-Agent' => "GitLab/#{Gitlab::VERSION}",
-      'X-Gitlab-Event' => 'Push Hook'
-    }
-  end
-
   let(:data) do
     { before: 'oldrev', after: 'newrev', ref: 'ref' }
   end
@@ -61,6 +53,21 @@
   end
 
   describe '#execute' do
+    let!(:uuid) { SecureRandom.uuid }
+    let(:headers) do
+      {
+        'Content-Type' => 'application/json',
+        'User-Agent' => "GitLab/#{Gitlab::VERSION}",
+        'X-Gitlab-Event' => 'Push Hook',
+        'X-Gitlab-Event-UUID' => uuid
+      }
+    end
+
+    before do
+      # Set a stable value for the `X-Gitlab-Event-UUID` header.
+      Gitlab::WebHooks::RecursionDetection.set_request_uuid(uuid)
+    end
+
     context 'when token is defined' do
       let_it_be(:project_hook) { create(:project_hook, :token) }
 
@@ -127,11 +134,74 @@
       expect(service_instance.execute).to eq({ status: :error, message: 'Hook disabled' })
     end
 
+    it 'executes and registers the hook with the recursion detection', :aggregate_failures do
+      stub_full_request(project_hook.url, method: :post)
+      cache_key = Gitlab::WebHooks::RecursionDetection.send(:cache_key_for_hook, project_hook)
+
+      ::Gitlab::Redis::SharedState.with do |redis|
+        expect { service_instance.execute }.to change {
+          redis.sismember(cache_key, project_hook.id)
+        }.to(true)
+      end
+
+      expect(WebMock).to have_requested(:post, stubbed_hostname(project_hook.url))
+        .with(headers: headers)
+        .once
+    end
+
+    it 'executes and logs if a recursive web hook is detected', :aggregate_failures do
+      stub_full_request(project_hook.url, method: :post)
+      Gitlab::WebHooks::RecursionDetection.register!(project_hook)
+
+      expect(Gitlab::AuthLogger).to receive(:error).with(
+        include(
+          message: 'Webhook recursion detected and will be blocked in future',
+          hook_id: project_hook.id,
+          hook_type: 'ProjectHook',
+          hook_name: 'push_hooks',
+          recursion_detection: Gitlab::WebHooks::RecursionDetection.to_log(project_hook),
+          'correlation_id' => kind_of(String)
+        )
+      )
+
+      service_instance.execute
+
+      expect(WebMock).to have_requested(:post, stubbed_hostname(project_hook.url))
+        .with(headers: headers)
+        .once
+    end
+
+    it 'executes and logs if the recursion count limit would be exceeded', :aggregate_failures do
+      stub_full_request(project_hook.url, method: :post)
+      stub_const("#{Gitlab::WebHooks::RecursionDetection.name}::COUNT_LIMIT", 3)
+      previous_hooks = create_list(:project_hook, 3)
+      previous_hooks.each { Gitlab::WebHooks::RecursionDetection.register!(_1) }
+
+      expect(Gitlab::AuthLogger).to receive(:error).with(
+        include(
+          message: 'Webhook recursion detected and will be blocked in future',
+          hook_id: project_hook.id,
+          hook_type: 'ProjectHook',
+          hook_name: 'push_hooks',
+          recursion_detection: Gitlab::WebHooks::RecursionDetection.to_log(project_hook),
+          'correlation_id' => kind_of(String)
+        )
+      )
+
+      service_instance.execute
+
+      expect(WebMock).to have_requested(:post, stubbed_hostname(project_hook.url))
+        .with(headers: headers)
+        .once
+    end
+
     it 'handles exceptions' do
       exceptions = Gitlab::HTTP::HTTP_ERRORS + [
         Gitlab::Json::LimitedEncoder::LimitExceeded, URI::InvalidURIError
       ]
 
+      allow(Gitlab::WebHooks::RecursionDetection).to receive(:block?).and_return(false)
+
       exceptions.each do |exception_class|
         exception = exception_class.new('Exception message')
         project_hook.enable!
@@ -420,6 +490,57 @@ def expect_to_rate_limit(hook, threshold:, throttled: false)
       end
     end
 
+    context 'recursion detection' do
+      before do
+        # Set a request UUID so `RecursionDetection.block?` will query redis.
+        Gitlab::WebHooks::RecursionDetection.set_request_uuid(SecureRandom.uuid)
+      end
+
+      it 'queues a worker and logs an error if the call chain limit would be exceeded' do
+        stub_const("#{Gitlab::WebHooks::RecursionDetection.name}::COUNT_LIMIT", 3)
+        previous_hooks = create_list(:project_hook, 3)
+        previous_hooks.each { Gitlab::WebHooks::RecursionDetection.register!(_1) }
+
+        expect(WebHookWorker).to receive(:perform_async)
+        expect(Gitlab::AuthLogger).to receive(:error).with(
+          include(
+            message: 'Webhook recursion detected and will be blocked in future',
+            hook_id: project_hook.id,
+            hook_type: 'ProjectHook',
+            hook_name: 'push_hooks',
+            recursion_detection: Gitlab::WebHooks::RecursionDetection.to_log(project_hook),
+            'correlation_id' => kind_of(String),
+            'meta.project' => project.full_path,
+            'meta.related_class' => 'ProjectHook',
+            'meta.root_namespace' => project.root_namespace.full_path
+          )
+        )
+
+        service_instance.async_execute
+      end
+
+      it 'queues a worker and logs an error if a recursive call chain is detected' do
+        Gitlab::WebHooks::RecursionDetection.register!(project_hook)
+
+        expect(WebHookWorker).to receive(:perform_async)
+        expect(Gitlab::AuthLogger).to receive(:error).with(
+          include(
+            message: 'Webhook recursion detected and will be blocked in future',
+            hook_id: project_hook.id,
+            hook_type: 'ProjectHook',
+            hook_name: 'push_hooks',
+            recursion_detection: Gitlab::WebHooks::RecursionDetection.to_log(project_hook),
+            'correlation_id' => kind_of(String),
+            'meta.project' => project.full_path,
+            'meta.related_class' => 'ProjectHook',
+            'meta.root_namespace' => project.root_namespace.full_path
+          )
+        )
+
+        service_instance.async_execute
+      end
+    end
+
     context 'when hook has custom context attributes' do
       it 'includes the attributes in the worker context' do
         expect(WebHookWorker).to receive(:perform_async) do
diff --git a/spec/workers/web_hook_worker_spec.rb b/spec/workers/web_hook_worker_spec.rb
index 0f40177eb7d4173a12a1cfb27f3f1de3d8fdac2c..bbb8844a447ab8cfd0ae140aa6cfdace2d3425e2 100644
--- a/spec/workers/web_hook_worker_spec.rb
+++ b/spec/workers/web_hook_worker_spec.rb
@@ -19,6 +19,15 @@
       expect { subject.perform(non_existing_record_id, data, hook_name) }.not_to raise_error
     end
 
+    it 'retrieves recursion detection data, reinstates it, and cleans it from payload', :request_store, :aggregate_failures do
+      uuid = SecureRandom.uuid
+      full_data = data.merge({ _gitlab_recursion_detection_request_uuid: uuid })
+
+      expect_next(WebHookService, project_hook, data.with_indifferent_access, hook_name, anything).to receive(:execute)
+      expect { subject.perform(project_hook.id, full_data, hook_name) }
+        .to change { Gitlab::WebHooks::RecursionDetection::UUID.instance.request_uuid }.to(uuid)
+    end
+
     it_behaves_like 'worker with data consistency',
                   described_class,
                   data_consistency: :delayed