Add latest changes from gitlab-org/gitlab@master

31 files changed, 1101 insertions, 672 deletions

diff --git a/spec/finders/issues_finder_spec.rb b/spec/finders/issues_finder_spec.rb
index 28764f98fe1..dbf5abe64a5 100644
--- a/spec/finders/issues_finder_spec.rb
+++ b/spec/finders/issues_finder_spec.rb
@@ -3,6 +3,7 @@
 require 'spec_helper'
 
 RSpec.describe IssuesFinder do
+  using RSpec::Parameterized::TableSyntax
   include_context 'IssuesFinder context'
 
   describe '#execute' do
@@ -1022,4 +1023,33 @@ RSpec.describe IssuesFinder do
       end
     end
   end
+
+  describe '#parent_param=' do
+    let(:finder) { described_class.new(nil) }
+
+    subject { finder.parent_param = obj }
+
+    where(:klass, :param) do
+      :Project | :project_id
+      :Group   | :group_id
+    end
+
+    with_them do
+      let(:obj) { Object.const_get(klass, false).new }
+
+      it 'sets the params' do
+        subject
+
+        expect(finder.params[param]).to eq(obj)
+      end
+    end
+
+    context 'unexpected parent' do
+      let(:obj) { MergeRequest.new }
+
+      it 'raises an error' do
+        expect { subject }.to raise_error('Unexpected parent: MergeRequest')
+      end
+    end
+  end
 end
diff --git a/spec/fixtures/gitlab/database/structure_example_cleaned.sql b/spec/fixtures/gitlab/database/structure_example_cleaned.sql
index 42eed974e64..dc112da7037 100644
--- a/spec/fixtures/gitlab/database/structure_example_cleaned.sql
+++ b/spec/fixtures/gitlab/database/structure_example_cleaned.sql
@@ -1,8 +1,6 @@
-SET search_path=public;
+CREATE EXTENSION IF NOT EXISTS pg_trgm;
 
-CREATE EXTENSION IF NOT EXISTS pg_trgm WITH SCHEMA public;
-
-CREATE TABLE public.abuse_reports (
+CREATE TABLE abuse_reports (
     id integer NOT NULL,
     reporter_id integer,
     user_id integer,
@@ -13,20 +11,18 @@ CREATE TABLE public.abuse_reports (
     cached_markdown_version integer
 );
 
-CREATE SEQUENCE public.abuse_reports_id_seq
+CREATE SEQUENCE abuse_reports_id_seq
     START WITH 1
     INCREMENT BY 1
     NO MINVALUE
     NO MAXVALUE
     CACHE 1;
 
-ALTER TABLE ONLY public.abuse_reports ALTER COLUMN id SET DEFAULT nextval('public.abuse_reports_id_seq'::regclass);
+ALTER TABLE ONLY abuse_reports ALTER COLUMN id SET DEFAULT nextval('abuse_reports_id_seq'::regclass);
 
-ALTER TABLE ONLY public.abuse_reports
+ALTER TABLE ONLY abuse_reports
     ADD CONSTRAINT abuse_reports_pkey PRIMARY KEY (id);
 
-CREATE INDEX index_abuse_reports_on_user_id ON public.abuse_reports USING btree (user_id);
-
--- schema_migrations.version information is no longer stored in this file,
+CREATE INDEX index_abuse_reports_on_user_id ON abuse_reports USING btree (user_id);-- schema_migrations.version information is no longer stored in this file,
 -- but instead tracked in the db/schema_migrations directory
 -- see https://gitlab.com/gitlab-org/gitlab/-/issues/218590 for details
diff --git a/spec/frontend/design_management/components/design_notes/__snapshots__/design_note_spec.js.snap b/spec/frontend/design_management/components/design_notes/__snapshots__/design_note_spec.js.snap
index 8711083ad8c..084a7e5d712 100644
--- a/spec/frontend/design_management/components/design_notes/__snapshots__/design_note_spec.js.snap
+++ b/spec/frontend/design_management/components/design_notes/__snapshots__/design_note_spec.js.snap
@@ -60,7 +60,7 @@ exports[`Design note component should match the snapshot 1`] = `
     </div>
      
     <div
-      class="gl-display-flex"
+      class="gl-display-flex gl-align-items-baseline"
     >
        
       <!---->
diff --git a/spec/graphql/resolvers/issue_status_counts_resolver_spec.rb b/spec/graphql/resolvers/issue_status_counts_resolver_spec.rb
index 69e940ee6ca..16eb190efc6 100644
--- a/spec/graphql/resolvers/issue_status_counts_resolver_spec.rb
+++ b/spec/graphql/resolvers/issue_status_counts_resolver_spec.rb
@@ -41,6 +41,12 @@ RSpec.describe Resolvers::IssueStatusCountsResolver do
 
     it_behaves_like 'returns expected results'
 
+    context 'project used as parent' do
+      let(:parent) { project }
+
+      it_behaves_like 'returns expected results'
+    end
+
     context 'group used as parent' do
       let(:parent) { project.group }
 
diff --git a/spec/lib/gitlab/database/custom_structure_spec.rb b/spec/lib/gitlab/database/custom_structure_spec.rb
index b3bdca0acdd..04ce1e4ad9a 100644
--- a/spec/lib/gitlab/database/custom_structure_spec.rb
+++ b/spec/lib/gitlab/database/custom_structure_spec.rb
@@ -9,7 +9,6 @@ RSpec.describe Gitlab::Database::CustomStructure do
     <<~DATA
       -- this file tracks custom GitLab data, such as foreign keys referencing partitioned tables
       -- more details can be found in the issue: https://gitlab.com/gitlab-org/gitlab/-/issues/201872
-      SET search_path=public;
     DATA
   end
 
diff --git a/spec/lib/gitlab/database/schema_cleaner_spec.rb b/spec/lib/gitlab/database/schema_cleaner_spec.rb
index 1303ad7a311..950759c7f96 100644
--- a/spec/lib/gitlab/database/schema_cleaner_spec.rb
+++ b/spec/lib/gitlab/database/schema_cleaner_spec.rb
@@ -15,8 +15,8 @@ RSpec.describe Gitlab::Database::SchemaCleaner do
     expect(subject).not_to include('COMMENT ON EXTENSION')
   end
 
-  it 'sets the search_path' do
-    expect(subject.split("\n").first).to eq('SET search_path=public;')
+  it 'no assumption about public being the default schema' do
+    expect(subject).not_to match(/public\.\w+/)
   end
 
   it 'cleans up the full schema as expected (blackbox test with example)' do
diff --git a/spec/lib/gitlab/email/handler/create_note_handler_spec.rb b/spec/lib/gitlab/email/handler/create_note_handler_spec.rb
index 07b8070be30..ef448ee96a4 100644
--- a/spec/lib/gitlab/email/handler/create_note_handler_spec.rb
+++ b/spec/lib/gitlab/email/handler/create_note_handler_spec.rb
@@ -65,24 +65,15 @@ RSpec.describe Gitlab::Email::Handler::CreateNoteHandler do
         end
       end
 
-      [true, false].each do |state_tracking_enabled|
-        context "and current user can update noteable #{state_tracking_enabled ? 'enabled' : 'disabled'}" do
-          before do
-            stub_feature_flags(track_resource_state_change_events: state_tracking_enabled)
-
-            project.add_developer(user)
-          end
+      context "and current user can update noteable" do
+        before do
+          project.add_developer(user)
+        end
 
-          it 'does not raise an error' do
-            if state_tracking_enabled
-              expect { receiver.execute }.to change { noteable.resource_state_events.count }.by(1)
-            else
-              # One system note is created for the 'close' event
-              expect { receiver.execute }.to change { noteable.notes.count }.by(1)
-            end
+        it 'does not raise an error' do
+          expect { receiver.execute }.to change { noteable.resource_state_events.count }.by(1)
 
-            expect(noteable.reload).to be_closed
-          end
+          expect(noteable.reload).to be_closed
         end
       end
     end
diff --git a/spec/lib/gitlab/graphql/loaders/issuable_loader_spec.rb b/spec/lib/gitlab/graphql/loaders/issuable_loader_spec.rb
index 180966de895..33a9d40931e 100644
--- a/spec/lib/gitlab/graphql/loaders/issuable_loader_spec.rb
+++ b/spec/lib/gitlab/graphql/loaders/issuable_loader_spec.rb
@@ -6,9 +6,26 @@ RSpec.describe Gitlab::Graphql::Loaders::IssuableLoader do
   subject { described_class.new(parent, finder) }
 
   let(:params) { HashWithIndifferentAccess.new }
+  let(:finder_params) { finder.params.to_h.with_indifferent_access }
+
+  # Dumb finder class, that only implements what we need, and has
+  # predictable query counts.
+  let(:finder_class) do
+    Class.new(IssuesFinder) do
+      def execute
+        params[:project_id].issues.where(iid: params[:iids])
+      end
+
+      private
+
+      def params_class
+        IssuesFinder::Params
+      end
+    end
+  end
 
   describe '#find_all' do
-    let(:finder) { double(:finder, params: params, execute: %i[x y z]) }
+    let(:finder) { issuable_finder(params: params, result: [:x, :y, :z]) }
 
     where(:factory, :param_name) do
       %i[project group].map { |thing| [thing, :"#{thing}_id"] }
@@ -19,7 +36,7 @@ RSpec.describe Gitlab::Graphql::Loaders::IssuableLoader do
 
       it 'assignes the parent parameter, and batching_find_alls the finder' do
         expect(subject.find_all).to contain_exactly(:x, :y, :z)
-        expect(params).to include(param_name => parent)
+        expect(finder_params).to include(param_name => parent)
       end
     end
 
@@ -34,12 +51,12 @@ RSpec.describe Gitlab::Graphql::Loaders::IssuableLoader do
 
   describe '#batching_find_all' do
     context 'the finder params are anything other than [iids]' do
-      let(:finder) { double(:finder, params: params, execute: [:foo]) }
+      let(:finder) { issuable_finder(params: params, result: [:foo]) }
       let(:parent) { build_stubbed(:project) }
 
       it 'batching_find_alls the finder, setting the correct parent parameter' do
         expect(subject.batching_find_all).to eq([:foo])
-        expect(params[:project_id]).to eq(parent)
+        expect(finder_params[:project_id]).to eq(parent)
       end
 
       it 'allows a post-process block' do
@@ -48,23 +65,6 @@ RSpec.describe Gitlab::Graphql::Loaders::IssuableLoader do
     end
 
     context 'the finder params are exactly [iids]' do
-      # Dumb finder class, that only implements what we need, and has
-      # predictable query counts.
-      let(:finder_class) do
-        Class.new do
-          attr_reader :current_user, :params
-
-          def initialize(user, args)
-            @current_user = user
-            @params = HashWithIndifferentAccess.new(args.to_h)
-          end
-
-          def execute
-            params[:project_id].issues.where(iid: params[:iids])
-          end
-        end
-      end
-
       it 'batches requests' do
         issue_a = create(:issue)
         issue_b = create(:issue)
@@ -93,4 +93,13 @@ RSpec.describe Gitlab::Graphql::Loaders::IssuableLoader do
       end
     end
   end
+
+  private
+
+  def issuable_finder(user: double(:user), params: {}, result: nil)
+    new_finder = finder_class.new(user, params)
+    allow(new_finder).to receive(:execute).and_return(result) if result
+
+    new_finder
+  end
 end
diff --git a/spec/lib/gitlab/middleware/multipart/handler_for_jwt_params_spec.rb b/spec/lib/gitlab/middleware/multipart/handler_for_jwt_params_spec.rb
new file mode 100644
index 00000000000..59ec743f6ca
--- /dev/null
+++ b/spec/lib/gitlab/middleware/multipart/handler_for_jwt_params_spec.rb
@@ -0,0 +1,53 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Middleware::Multipart::HandlerForJWTParams do
+  using RSpec::Parameterized::TableSyntax
+
+  let_it_be(:env) { Rack::MockRequest.env_for('/', method: 'post', params: {}) }
+  let_it_be(:message) { { 'rewritten_fields' => {} } }
+
+  describe '#allowed_paths' do
+    let_it_be(:expected_allowed_paths) do
+      [
+        Dir.tmpdir,
+        ::FileUploader.root,
+        ::Gitlab.config.uploads.storage_path,
+        ::JobArtifactUploader.workhorse_upload_path,
+        ::LfsObjectUploader.workhorse_upload_path,
+        File.join(Rails.root, 'public/uploads/tmp')
+      ]
+    end
+
+    let_it_be(:expected_with_packages_path) { expected_allowed_paths + [::Packages::PackageFileUploader.workhorse_upload_path] }
+
+    subject { described_class.new(env, message).send(:allowed_paths) }
+
+    where(:package_features_enabled, :object_storage_enabled, :direct_upload_enabled, :expected_paths) do
+      false | false | true  | :expected_allowed_paths
+      false | false | false | :expected_allowed_paths
+      false | true  | true  | :expected_allowed_paths
+      false | true  | false | :expected_allowed_paths
+      true  | false | true  | :expected_with_packages_path
+      true  | false | false | :expected_with_packages_path
+      true  | true  | true  | :expected_allowed_paths
+      true  | true  | false | :expected_with_packages_path
+    end
+
+    with_them do
+      before do
+        stub_config(packages: {
+          enabled: package_features_enabled,
+          object_store: {
+            enabled: object_storage_enabled,
+            direct_upload: direct_upload_enabled
+          },
+          storage_path: '/any/dir'
+        })
+      end
+
+      it { is_expected.to eq(send(expected_paths)) }
+    end
+  end
+end
diff --git a/spec/lib/gitlab/middleware/multipart_with_handler_for_jwt_params_spec.rb b/spec/lib/gitlab/middleware/multipart_with_handler_for_jwt_params_spec.rb
new file mode 100644
index 00000000000..875e3820011
--- /dev/null
+++ b/spec/lib/gitlab/middleware/multipart_with_handler_for_jwt_params_spec.rb
@@ -0,0 +1,171 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Gitlab::Middleware::Multipart do
+  include MultipartHelpers
+
+  describe '#call' do
+    let(:app) { double(:app) }
+    let(:middleware) { described_class.new(app) }
+    let(:secret) { Gitlab::Workhorse.secret }
+    let(:issuer) { 'gitlab-workhorse' }
+
+    subject do
+      env = post_env(
+        rewritten_fields: rewritten_fields,
+        params: params,
+        secret: secret,
+        issuer: issuer
+      )
+      middleware.call(env)
+    end
+
+    before do
+      stub_feature_flags(upload_middleware_jwt_params_handler: true)
+    end
+
+    context 'remote file mode' do
+      let(:mode) { :remote }
+
+      it_behaves_like 'handling all upload parameters conditions'
+
+      context 'and a path set' do
+        include_context 'with one temporary file for multipart'
+
+        let(:rewritten_fields) { rewritten_fields_hash('file' => uploaded_filepath) }
+        let(:params) { upload_parameters_for(key: 'file', filename: filename, remote_id: remote_id).merge('file.path' => '/should/not/be/read') }
+
+        it 'builds an UploadedFile' do
+          expect_uploaded_files(original_filename: filename, remote_id: remote_id, size: uploaded_file.size, params_path: %w(file))
+
+          subject
+        end
+      end
+    end
+
+    context 'local file mode' do
+      let(:mode) { :local }
+
+      it_behaves_like 'handling all upload parameters conditions'
+
+      context 'when file is' do
+        include_context 'with one temporary file for multipart'
+
+        let(:allowed_paths) { [Dir.tmpdir] }
+
+        before do
+          expect_next_instance_of(::Gitlab::Middleware::Multipart::HandlerForJWTParams) do |handler|
+            expect(handler).to receive(:allowed_paths).and_return(allowed_paths)
+          end
+        end
+
+        context 'in allowed paths' do
+          let(:rewritten_fields) { rewritten_fields_hash('file' => uploaded_filepath) }
+          let(:params) { upload_parameters_for(filepath: uploaded_filepath, key: 'file', filename: filename) }
+
+          it 'builds an UploadedFile' do
+            expect_uploaded_files(filepath: uploaded_filepath, original_filename: filename, size: uploaded_file.size, params_path: %w(file))
+
+            subject
+          end
+        end
+
+        context 'not in allowed paths' do
+          let(:allowed_paths) { [] }
+
+          let(:rewritten_fields) { rewritten_fields_hash('file' => uploaded_filepath) }
+          let(:params) { upload_parameters_for(filepath: uploaded_filepath, key: 'file') }
+
+          it 'returns an error' do
+            result = subject
+
+            expect(result[0]).to eq(400)
+            expect(result[2]).to include('insecure path used')
+          end
+        end
+      end
+    end
+
+    context 'with dummy params in remote mode' do
+      let(:rewritten_fields) { { 'file' => 'should/not/be/read' } }
+      let(:params) { upload_parameters_for(key: 'file') }
+      let(:mode) { :remote }
+
+      context 'with an invalid secret' do
+        let(:secret) { 'INVALID_SECRET' }
+
+        it { expect { subject }.to raise_error(JWT::VerificationError) }
+      end
+
+      context 'with an invalid issuer' do
+        let(:issuer) { 'INVALID_ISSUER' }
+
+        it { expect { subject }.to raise_error(JWT::InvalidIssuerError) }
+      end
+
+      context 'with invalid rewritten field key' do
+        invalid_keys = [
+          '[file]',
+          ';file',
+          'file]',
+          ';file]',
+          'file]]',
+          'file;;'
+        ]
+
+        invalid_keys.each do |invalid_key|
+          context invalid_key do
+            let(:rewritten_fields) { { invalid_key => 'should/not/be/read' } }
+
+            it { expect { subject }.to raise_error(RuntimeError, "invalid field: \"#{invalid_key}\"") }
+          end
+        end
+      end
+
+      context 'with invalid key in parameters' do
+        include_context 'with one temporary file for multipart'
+
+        let(:rewritten_fields) { rewritten_fields_hash('file' => uploaded_filepath) }
+        let(:params) { upload_parameters_for(filepath: uploaded_filepath, key: 'wrong_key', filename: filename, remote_id: remote_id) }
+
+        it 'raises an error' do
+          expect { subject }.to raise_error(RuntimeError, 'Empty JWT param: file.gitlab-workhorse-upload')
+        end
+      end
+
+      context 'with a modified JWT payload' do
+        include_context 'with one temporary file for multipart'
+
+        let(:rewritten_fields) { rewritten_fields_hash('file' => uploaded_filepath) }
+        let(:crafted_payload) { Base64.urlsafe_encode64({ 'path' => 'test' }.to_json) }
+        let(:params) do
+          upload_parameters_for(filepath: uploaded_filepath, key: 'file', filename: filename, remote_id: remote_id).tap do |params|
+            header, _, sig = params['file.gitlab-workhorse-upload'].split('.')
+            params['file.gitlab-workhorse-upload'] = [header, crafted_payload, sig].join('.')
+          end
+        end
+
+        it 'raises an error' do
+          expect { subject }.to raise_error(JWT::VerificationError, 'Signature verification raised')
+        end
+      end
+
+      context 'with a modified JWT sig' do
+        include_context 'with one temporary file for multipart'
+
+        let(:rewritten_fields) { rewritten_fields_hash('file' => uploaded_filepath) }
+        let(:params) do
+          upload_parameters_for(filepath: uploaded_filepath, key: 'file', filename: filename, remote_id: remote_id).tap do |params|
+            header, payload, sig = params['file.gitlab-workhorse-upload'].split('.')
+            params['file.gitlab-workhorse-upload'] = [header, payload, "#{sig}modified"].join('.')
+          end
+        end
+
+        it 'raises an error' do
+          expect { subject }.to raise_error(JWT::VerificationError, 'Signature verification raised')
+        end
+      end
+    end
+  end
+end
diff --git a/spec/lib/gitlab/middleware/multipart_spec.rb b/spec/lib/gitlab/middleware/multipart_with_handler_spec.rb
index 781f3e0289b..742a5639ace 100644
--- a/spec/lib/gitlab/middleware/multipart_spec.rb
+++ b/spec/lib/gitlab/middleware/multipart_with_handler_spec.rb
@@ -21,7 +21,11 @@ RSpec.describe Gitlab::Middleware::Multipart do
       middleware.call(env)
     end
 
-    context 'with remote file mode params' do
+    before do
+      stub_feature_flags(upload_middleware_jwt_params_handler: false)
+    end
+
+    context 'remote file mode' do
       let(:mode) { :remote }
 
       it_behaves_like 'handling all upload parameters conditions'
@@ -58,10 +62,10 @@ RSpec.describe Gitlab::Middleware::Multipart do
 
         context 'in allowed paths' do
           let(:rewritten_fields) { rewritten_fields_hash('file' => uploaded_filepath) }
-          let(:params) { upload_parameters_for(filepath: uploaded_filepath, key: 'file', filename: filename, remote_id: remote_id) }
+          let(:params) { upload_parameters_for(filepath: uploaded_filepath, key: 'file', filename: filename) }
 
           it 'builds an UploadedFile' do
-            expect_uploaded_files(filepath: uploaded_filepath, original_filename: filename, remote_id: remote_id, size: uploaded_file.size, params_path: %w(file))
+            expect_uploaded_files(filepath: uploaded_filepath, original_filename: filename, size: uploaded_file.size, params_path: %w(file))
 
             subject
           end
diff --git a/spec/lib/gitlab/usage_data_spec.rb b/spec/lib/gitlab/usage_data_spec.rb
index 6631a0d3cc6..7c413f3bcca 100644
--- a/spec/lib/gitlab/usage_data_spec.rb
+++ b/spec/lib/gitlab/usage_data_spec.rb
@@ -628,6 +628,7 @@ RSpec.describe Gitlab::UsageData, :aggregate_failures do
         expect(subject[:gitlab_shared_runners_enabled]).to eq(Gitlab.config.gitlab_ci.shared_runners_enabled)
         expect(subject[:web_ide_clientside_preview_enabled]).to eq(Gitlab::CurrentSettings.web_ide_clientside_preview_enabled?)
         expect(subject[:grafana_link_enabled]).to eq(Gitlab::CurrentSettings.grafana_enabled?)
+        expect(subject[:gitpod_enabled]).to eq(Gitlab::CurrentSettings.gitpod_enabled?)
       end
 
       context 'with embedded Prometheus' do
@@ -657,6 +658,20 @@ RSpec.describe Gitlab::UsageData, :aggregate_failures do
           expect(subject[:grafana_link_enabled]).to eq(false)
         end
       end
+
+      context 'with Gitpod' do
+        it 'returns true when is enabled' do
+          stub_application_setting(gitpod_enabled: true)
+
+          expect(subject[:gitpod_enabled]).to eq(true)
+        end
+
+        it 'returns false when is disabled' do
+          stub_application_setting(gitpod_enabled: false)
+
+          expect(subject[:gitpod_enabled]).to eq(false)
+        end
+      end
     end
 
     describe '.components_usage_data' do
diff --git a/spec/lib/uploaded_file_spec.rb b/spec/lib/uploaded_file_spec.rb
index cf2ab04b457..8425e1dbd46 100644
--- a/spec/lib/uploaded_file_spec.rb
+++ b/spec/lib/uploaded_file_spec.rb
@@ -14,226 +14,343 @@ RSpec.describe UploadedFile do
     FileUtils.rm_f(temp_file)
   end
 
-  describe ".from_params" do
-    let(:upload_path) { nil }
-    let(:file_path_override) { nil }
+  context 'from_params functions' do
+    RSpec.shared_examples 'using the file path' do |filename:, content_type:, sha256:, path_suffix:|
+      it { is_expected.not_to be_nil }
+
+      it 'sets properly the attributes' do
+        expect(subject.original_filename).to eq(filename)
+        expect(subject.content_type).to eq(content_type)
+        expect(subject.sha256).to eq(sha256)
+        expect(subject.remote_id).to be_nil
+        expect(subject.path).to end_with(path_suffix)
+      end
+
+      it 'handles a blank path' do
+        params['file.path'] = ''
+
+        # Not a real file, so can't determine size itself
+        params['file.size'] = 1.byte
 
-    after do
-      FileUtils.rm_r(upload_path) if upload_path
+        expect { described_class.from_params(params, :file, upload_path) }
+          .not_to raise_error
+      end
     end
 
-    subject do
-      described_class.from_params(params, :file, [upload_path, Dir.tmpdir], file_path_override)
+    RSpec.shared_examples 'using the remote id' do |filename:, content_type:, sha256:, size:, remote_id:|
+      it { is_expected.not_to be_nil }
+
+      it 'sets properly the attributes' do
+        expect(subject.original_filename).to eq(filename)
+        expect(subject.content_type).to eq('application/octet-stream')
+        expect(subject.sha256).to eq('sha256')
+        expect(subject.path).to be_nil
+        expect(subject.size).to eq(123456)
+        expect(subject.remote_id).to eq('1234567890')
+      end
     end
 
-    context 'when valid file is specified' do
-      context 'only local path is specified' do
-        let(:params) do
-          { 'file.path' => temp_file.path }
-        end
+    describe '.from_params_without_field' do
+      let(:upload_path) { nil }
 
-        it { is_expected.not_to be_nil }
+      after do
+        FileUtils.rm_r(upload_path) if upload_path
+      end
 
-        it "generates filename from path" do
-          expect(subject.original_filename).to eq(::File.basename(temp_file.path))
-        end
+      subject do
+        described_class.from_params_without_field(params, [upload_path, Dir.tmpdir])
       end
 
-      context 'all parameters are specified' do
-        RSpec.shared_context 'filepath override' do
-          let(:temp_file_override) { Tempfile.new(%w[override override], temp_dir) }
-          let(:file_path_override) { temp_file_override.path }
+      context 'when valid file is specified' do
+        context 'only local path is specified' do
+          let(:params) { { 'path' => temp_file.path } }
 
-          before do
-            FileUtils.touch(temp_file_override)
-          end
+          it { is_expected.not_to be_nil }
 
-          after do
-            FileUtils.rm_f(temp_file_override)
+          it 'generates filename from path' do
+            expect(subject.original_filename).to eq(::File.basename(temp_file.path))
           end
         end
 
-        RSpec.shared_examples 'using the file path' do |filename:, content_type:, sha256:, path_suffix:|
-          it 'sets properly the attributes' do
-            expect(subject.original_filename).to eq(filename)
-            expect(subject.content_type).to eq(content_type)
-            expect(subject.sha256).to eq(sha256)
-            expect(subject.remote_id).to be_nil
-            expect(subject.path).to end_with(path_suffix)
+        context 'all parameters are specified' do
+          context 'with a filepath' do
+            let(:params) do
+              { 'path' => temp_file.path,
+                'name' => 'dir/my file&.txt',
+                'type' => 'my/type',
+                'sha256' => 'sha256' }
+            end
+
+            it_behaves_like 'using the file path',
+                            filename: 'my_file_.txt',
+                            content_type: 'my/type',
+                            sha256: 'sha256',
+                            path_suffix: 'test'
           end
 
-          it 'handles a blank path' do
-            params['file.path'] = ''
-
-            # Not a real file, so can't determine size itself
-            params['file.size'] = 1.byte
+          context 'with a remote id' do
+            let(:params) do
+              {
+                'name' => 'dir/my file&.txt',
+                'sha256' => 'sha256',
+                'remote_url' => 'http://localhost/file',
+                'remote_id' => '1234567890',
+                'etag' => 'etag1234567890',
+                'size' => '123456'
+              }
+            end
+
+            it_behaves_like 'using the remote id',
+                            filename: 'my_file_.txt',
+                            content_type: 'application/octet-stream',
+                            sha256: 'sha256',
+                            size: 123456,
+                            remote_id: '1234567890'
+          end
 
-            expect { described_class.from_params(params, :file, upload_path) }
-              .not_to raise_error
+          context 'with a path and a remote id' do
+            let(:params) do
+              {
+                'path' => temp_file.path,
+                'name' => 'dir/my file&.txt',
+                'sha256' => 'sha256',
+                'remote_url' => 'http://localhost/file',
+                'remote_id' => '1234567890',
+                'etag' => 'etag1234567890',
+                'size' => '123456'
+              }
+            end
+
+            it_behaves_like 'using the remote id',
+                            filename: 'my_file_.txt',
+                            content_type: 'application/octet-stream',
+                            sha256: 'sha256',
+                            size: 123456,
+                            remote_id: '1234567890'
           end
         end
+      end
 
-        RSpec.shared_examples 'using the remote id' do |filename:, content_type:, sha256:, size:, remote_id:|
-          it 'sets properly the attributes' do
-            expect(subject.original_filename).to eq(filename)
-            expect(subject.content_type).to eq('application/octet-stream')
-            expect(subject.sha256).to eq('sha256')
-            expect(subject.path).to be_nil
-            expect(subject.size).to eq(123456)
-            expect(subject.remote_id).to eq('1234567890')
-          end
+      context 'when no params are specified' do
+        let(:params) { {} }
+
+        it 'does not return an object' do
+          is_expected.to be_nil
         end
+      end
 
-        context 'with a filepath' do
-          let(:params) do
-            { 'file.path' => temp_file.path,
-              'file.name' => 'dir/my file&.txt',
-              'file.type' => 'my/type',
-              'file.sha256' => 'sha256' }
-          end
+      context 'when verifying allowed paths' do
+        let(:params) { { 'path' => temp_file.path } }
+
+        context 'when file is stored in system temporary folder' do
+          let(:temp_dir) { Dir.tmpdir }
 
           it { is_expected.not_to be_nil }
+        end
+
+        context 'when file is stored in user provided upload path' do
+          let(:upload_path) { Dir.mktmpdir }
+          let(:temp_dir) { upload_path }
 
-          it_behaves_like 'using the file path',
-                          filename: 'my_file_.txt',
-                          content_type: 'my/type',
-                          sha256: 'sha256',
-                          path_suffix: 'test'
+          it { is_expected.not_to be_nil }
         end
 
-        context 'with a filepath override' do
-          include_context 'filepath override'
+        context 'when file is stored outside of user provided upload path' do
+          let!(:generated_dir) { Dir.mktmpdir }
+          let!(:temp_dir) { Dir.mktmpdir }
 
-          let(:params) do
-            { 'file.path' => temp_file.path,
-              'file.name' => 'dir/my file&.txt',
-              'file.type' => 'my/type',
-              'file.sha256' => 'sha256' }
+          before do
+            # We overwrite default temporary path
+            allow(Dir).to receive(:tmpdir).and_return(generated_dir)
           end
 
-          it { is_expected.not_to be_nil }
-
-          it_behaves_like 'using the file path',
-                          filename: 'my_file_.txt',
-                          content_type: 'my/type',
-                          sha256: 'sha256',
-                          path_suffix: 'override'
+          it 'raises an error' do
+            expect { subject }.to raise_error(UploadedFile::InvalidPathError, /insecure path used/)
+          end
         end
+      end
+    end
 
-        context 'with a remote id' do
-          let(:params) do
-            {
-              'file.name' => 'dir/my file&.txt',
-              'file.sha256' => 'sha256',
-              'file.remote_url' => 'http://localhost/file',
-              'file.remote_id' => '1234567890',
-              'file.etag' => 'etag1234567890',
-              'file.size' => '123456'
-            }
-          end
+    describe '.from_params' do
+      let(:upload_path) { nil }
+      let(:file_path_override) { nil }
 
-          it { is_expected.not_to be_nil }
+      after do
+        FileUtils.rm_r(upload_path) if upload_path
+      end
+
+      subject do
+        described_class.from_params(params, :file, [upload_path, Dir.tmpdir], file_path_override)
+      end
 
-          it_behaves_like 'using the remote id',
-                          filename: 'my_file_.txt',
-                          content_type: 'application/octet-stream',
-                          sha256: 'sha256',
-                          size: 123456,
-                          remote_id: '1234567890'
+      RSpec.shared_context 'filepath override' do
+        let(:temp_file_override) { Tempfile.new(%w[override override], temp_dir) }
+        let(:file_path_override) { temp_file_override.path }
+
+        before do
+          FileUtils.touch(temp_file_override)
         end
 
-        context 'with a path and a remote id' do
+        after do
+          FileUtils.rm_f(temp_file_override)
+        end
+      end
+
+      context 'when valid file is specified' do
+        context 'only local path is specified' do
           let(:params) do
-            {
-              'file.path' => temp_file.path,
-              'file.name' => 'dir/my file&.txt',
-              'file.sha256' => 'sha256',
-              'file.remote_url' => 'http://localhost/file',
-              'file.remote_id' => '1234567890',
-              'file.etag' => 'etag1234567890',
-              'file.size' => '123456'
-            }
+            { 'file.path' => temp_file.path }
           end
 
           it { is_expected.not_to be_nil }
 
-          it_behaves_like 'using the remote id',
-                          filename: 'my_file_.txt',
-                          content_type: 'application/octet-stream',
-                          sha256: 'sha256',
-                          size: 123456,
-                          remote_id: '1234567890'
+          it "generates filename from path" do
+            expect(subject.original_filename).to eq(::File.basename(temp_file.path))
+          end
         end
 
-        context 'with a path override and a remote id' do
-          include_context 'filepath override'
+        context 'all parameters are specified' do
+          context 'with a filepath' do
+            let(:params) do
+              { 'file.path' => temp_file.path,
+                'file.name' => 'dir/my file&.txt',
+                'file.type' => 'my/type',
+                'file.sha256' => 'sha256' }
+            end
+
+            it_behaves_like 'using the file path',
+                            filename: 'my_file_.txt',
+                            content_type: 'my/type',
+                            sha256: 'sha256',
+                            path_suffix: 'test'
+          end
 
-          let(:params) do
-            {
-              'file.name' => 'dir/my file&.txt',
-              'file.sha256' => 'sha256',
-              'file.remote_url' => 'http://localhost/file',
-              'file.remote_id' => '1234567890',
-              'file.etag' => 'etag1234567890',
-              'file.size' => '123456'
-            }
+          context 'with a filepath override' do
+            include_context 'filepath override'
+
+            let(:params) do
+              { 'file.path' => temp_file.path,
+                'file.name' => 'dir/my file&.txt',
+                'file.type' => 'my/type',
+                'file.sha256' => 'sha256' }
+            end
+
+            it_behaves_like 'using the file path',
+                            filename: 'my_file_.txt',
+                            content_type: 'my/type',
+                            sha256: 'sha256',
+                            path_suffix: 'override'
           end
 
-          it { is_expected.not_to be_nil }
+          context 'with a remote id' do
+            let(:params) do
+              {
+                'file.name' => 'dir/my file&.txt',
+                'file.sha256' => 'sha256',
+                'file.remote_url' => 'http://localhost/file',
+                'file.remote_id' => '1234567890',
+                'file.etag' => 'etag1234567890',
+                'file.size' => '123456'
+              }
+            end
+
+            it_behaves_like 'using the remote id',
+                            filename: 'my_file_.txt',
+                            content_type: 'application/octet-stream',
+                            sha256: 'sha256',
+                            size: 123456,
+                            remote_id: '1234567890'
+          end
 
-          it_behaves_like 'using the remote id',
-                          filename: 'my_file_.txt',
-                          content_type: 'application/octet-stream',
-                          sha256: 'sha256',
-                          size: 123456,
-                          remote_id: '1234567890'
+          context 'with a path and a remote id' do
+            let(:params) do
+              {
+                'file.path' => temp_file.path,
+                'file.name' => 'dir/my file&.txt',
+                'file.sha256' => 'sha256',
+                'file.remote_url' => 'http://localhost/file',
+                'file.remote_id' => '1234567890',
+                'file.etag' => 'etag1234567890',
+                'file.size' => '123456'
+              }
+            end
+
+            it_behaves_like 'using the remote id',
+                            filename: 'my_file_.txt',
+                            content_type: 'application/octet-stream',
+                            sha256: 'sha256',
+                            size: 123456,
+                            remote_id: '1234567890'
+          end
+
+          context 'with a path override and a remote id' do
+            include_context 'filepath override'
+
+            let(:params) do
+              {
+                'file.name' => 'dir/my file&.txt',
+                'file.sha256' => 'sha256',
+                'file.remote_url' => 'http://localhost/file',
+                'file.remote_id' => '1234567890',
+                'file.etag' => 'etag1234567890',
+                'file.size' => '123456'
+              }
+            end
+
+            it_behaves_like 'using the remote id',
+                            filename: 'my_file_.txt',
+                            content_type: 'application/octet-stream',
+                            sha256: 'sha256',
+                            size: 123456,
+                            remote_id: '1234567890'
+          end
         end
       end
-    end
 
-    context 'when no params are specified' do
-      let(:params) do
-        {}
-      end
+      context 'when no params are specified' do
+        let(:params) do
+          {}
+        end
 
-      it "does not return an object" do
-        is_expected.to be_nil
+        it "does not return an object" do
+          is_expected.to be_nil
+        end
       end
-    end
 
-    context 'when verifying allowed paths' do
-      let(:params) do
-        { 'file.path' => temp_file.path }
-      end
+      context 'when verifying allowed paths' do
+        let(:params) do
+          { 'file.path' => temp_file.path }
+        end
 
-      context 'when file is stored in system temporary folder' do
-        let(:temp_dir) { Dir.tmpdir }
+        context 'when file is stored in system temporary folder' do
+          let(:temp_dir) { Dir.tmpdir }
 
-        it "succeeds" do
-          is_expected.not_to be_nil
+          it "succeeds" do
+            is_expected.not_to be_nil
+          end
         end
-      end
 
-      context 'when file is stored in user provided upload path' do
-        let(:upload_path) { Dir.mktmpdir }
-        let(:temp_dir) { upload_path }
+        context 'when file is stored in user provided upload path' do
+          let(:upload_path) { Dir.mktmpdir }
+          let(:temp_dir) { upload_path }
 
-        it "succeeds" do
-          is_expected.not_to be_nil
+          it "succeeds" do
+            is_expected.not_to be_nil
+          end
         end
-      end
 
-      context 'when file is stored outside of user provided upload path' do
-        let!(:generated_dir) { Dir.mktmpdir }
-        let!(:temp_dir) { Dir.mktmpdir }
+        context 'when file is stored outside of user provided upload path' do
+          let!(:generated_dir) { Dir.mktmpdir }
+          let!(:temp_dir) { Dir.mktmpdir }
 
-        before do
-          # We overwrite default temporary path
-          allow(Dir).to receive(:tmpdir).and_return(generated_dir)
-        end
+          before do
+            # We overwrite default temporary path
+            allow(Dir).to receive(:tmpdir).and_return(generated_dir)
+          end
 
-        it "raises an error" do
-          expect { subject }.to raise_error(UploadedFile::InvalidPathError, /insecure path used/)
+          it "raises an error" do
+            expect { subject }.to raise_error(UploadedFile::InvalidPathError, /insecure path used/)
+          end
         end
       end
     end
diff --git a/spec/models/ci/job_artifact_spec.rb b/spec/models/ci/job_artifact_spec.rb
index 779839df670..a66e96d8a19 100644
--- a/spec/models/ci/job_artifact_spec.rb
+++ b/spec/models/ci/job_artifact_spec.rb
@@ -19,7 +19,7 @@ RSpec.describe Ci::JobArtifact do
 
   it_behaves_like 'having unique enum values'
 
-  it_behaves_like 'UpdateProjectStatistics' do
+  it_behaves_like 'UpdateProjectStatistics', :with_counter_attribute do
     let_it_be(:job, reload: true) { create(:ci_build) }
 
     subject { build(:ci_job_artifact, :archive, job: job, size: 107464) }
diff --git a/spec/models/concerns/counter_attribute_spec.rb b/spec/models/concerns/counter_attribute_spec.rb
index f23865a5dbb..a19fbae3cfb 100644
--- a/spec/models/concerns/counter_attribute_spec.rb
+++ b/spec/models/concerns/counter_attribute_spec.rb
@@ -12,6 +12,36 @@ RSpec.describe CounterAttribute, :counter_attribute, :clean_gitlab_redis_shared_
     let(:model) { CounterAttributeModel.find(project_statistics.id) }
   end
 
+  describe 'after_flush callbacks' do
+    let(:attribute) { model.class.counter_attributes.first}
+
+    subject { model.flush_increments_to_database!(attribute) }
+
+    it 'has registered callbacks' do # defined in :counter_attribute RSpec tag
+      expect(model.class.after_flush_callbacks.size).to eq(1)
+    end
+
+    context 'when there are increments to flush' do
+      before do
+        model.delayed_increment_counter(attribute, 10)
+      end
+
+      it 'executes the callbacks' do
+        subject
+
+        expect(model.flushed).to be_truthy
+      end
+    end
+
+    context 'when there are no increments to flush' do
+      it 'does not execute the callbacks' do
+        subject
+
+        expect(model.flushed).to be_nil
+      end
+    end
+  end
+
   describe '.steal_increments' do
     let(:increment_key) { 'counters:Model:123:attribute' }
     let(:flushed_key) { 'counter:Model:123:attribute:flushed' }
diff --git a/spec/models/merge_request_spec.rb b/spec/models/merge_request_spec.rb
index 98f709a0610..a14549ebff3 100644
--- a/spec/models/merge_request_spec.rb
+++ b/spec/models/merge_request_spec.rb
@@ -2358,48 +2358,43 @@ RSpec.describe MergeRequest, factory_default: :keep do
       end
     end
 
-    context 'when state event tracking is disabled' do
+    context 'when no metrics or merge event exists' do
+      let(:user) { create(:user) }
+      let(:merge_request) { create(:merge_request, :merged) }
+
       before do
-        stub_feature_flags(track_resource_state_change_events: false)
+        merge_request.metrics.destroy!
       end
 
-      context 'when merging note is persisted, but no metrics or merge event exists' do
-        let(:user) { create(:user) }
-        let(:merge_request) { create(:merge_request, :merged) }
-
+      context 'when resource event for the merge exists' do
         before do
-          merge_request.metrics.destroy!
-
           SystemNoteService.change_status(merge_request,
                                           merge_request.target_project,
                                           user,
                                           merge_request.state, nil)
         end
 
-        it 'returns merging note creation date' do
+        it 'returns the resource event creation date' do
           expect(merge_request.reload.metrics).to be_nil
           expect(merge_request.merge_event).to be_nil
-          expect(merge_request.notes.count).to eq(1)
-          expect(merge_request.merged_at).to eq(merge_request.notes.first.created_at)
+          expect(merge_request.resource_state_events.count).to eq(1)
+          expect(merge_request.merged_at).to eq(merge_request.resource_state_events.first.created_at)
         end
       end
-    end
-
-    context 'when state event tracking is enabled' do
-      let(:user) { create(:user) }
-      let(:merge_request) { create(:merge_request, :merged) }
-
-      before do
-        merge_request.metrics.destroy!
 
-        SystemNoteService.change_status(merge_request,
-                                        merge_request.target_project,
-                                        user,
-                                        merge_request.state, nil)
-      end
+      context 'when system note for the merge exists' do
+        before do
+          # We do not create these system notes anymore but we need this to work for existing MRs
+          # that used system notes instead of resource state events
+          create(:note, :system, noteable: merge_request, note: 'merged')
+        end
 
-      it 'does not create a system note' do
-        expect(merge_request.notes).to be_empty
+        it 'returns the merging note creation date' do
+          expect(merge_request.reload.metrics).to be_nil
+          expect(merge_request.merge_event).to be_nil
+          expect(merge_request.notes.count).to eq(1)
+          expect(merge_request.merged_at).to eq(merge_request.notes.first.created_at)
+        end
       end
     end
   end
diff --git a/spec/models/project_statistics_spec.rb b/spec/models/project_statistics_spec.rb
index 383fabcfffb..47e57972cc1 100644
--- a/spec/models/project_statistics_spec.rb
+++ b/spec/models/project_statistics_spec.rb
@@ -324,22 +324,51 @@ RSpec.describe ProjectStatistics do
   describe '.increment_statistic' do
     shared_examples 'a statistic that increases storage_size' do
       it 'increases the statistic by that amount' do
-        expect { described_class.increment_statistic(project.id, stat, 13) }
+        expect { described_class.increment_statistic(project, stat, 13) }
           .to change { statistics.reload.send(stat) || 0 }
           .by(13)
       end
 
       it 'increases also storage size by that amount' do
-        expect { described_class.increment_statistic(project.id, stat, 20) }
+        expect { described_class.increment_statistic(project, stat, 20) }
           .to change { statistics.reload.storage_size }
           .by(20)
       end
     end
 
+    shared_examples 'a statistic that increases storage_size asynchronously' do
+      it 'stores the increment temporarily in Redis', :clean_gitlab_redis_shared_state do
+        described_class.increment_statistic(project, stat, 13)
+
+        Gitlab::Redis::SharedState.with do |redis|
+          increment = redis.get(statistics.counter_key(stat))
+          expect(increment.to_i).to eq(13)
+        end
+      end
+
+      it 'schedules a worker to update the statistic and storage_size async' do
+        expect(FlushCounterIncrementsWorker)
+          .to receive(:perform_in)
+          .with(CounterAttribute::WORKER_DELAY, described_class.name, statistics.id, stat)
+
+        expect(FlushCounterIncrementsWorker)
+          .to receive(:perform_in)
+          .with(CounterAttribute::WORKER_DELAY, described_class.name, statistics.id, :storage_size)
+
+        described_class.increment_statistic(project, stat, 20)
+      end
+    end
+
     context 'when adjusting :build_artifacts_size' do
       let(:stat) { :build_artifacts_size }
 
-      it_behaves_like 'a statistic that increases storage_size'
+      it_behaves_like 'a statistic that increases storage_size asynchronously'
+
+      it_behaves_like 'a statistic that increases storage_size' do
+        before do
+          stub_feature_flags(efficient_counter_attribute: false)
+        end
+      end
     end
 
     context 'when adjusting :pipeline_artifacts_size' do
diff --git a/spec/services/alert_management/process_prometheus_alert_service_spec.rb b/spec/services/alert_management/process_prometheus_alert_service_spec.rb
index b14cc65506a..ad4ab26c198 100644
--- a/spec/services/alert_management/process_prometheus_alert_service_spec.rb
+++ b/spec/services/alert_management/process_prometheus_alert_service_spec.rb
@@ -148,28 +148,20 @@ RSpec.describe AlertManagement::ProcessPrometheusAlertService do
               expect { execute }.to change { alert.reload.resolved? }.to(true)
             end
 
-            [true, false].each do |state_tracking_enabled|
-              context 'existing issue' do
-                before do
-                  stub_feature_flags(track_resource_state_change_events: state_tracking_enabled)
-                end
-
-                let!(:alert) { create(:alert_management_alert, :with_issue, project: project, fingerprint: fingerprint) }
-
-                it 'closes the issue' do
-                  issue = alert.issue
-
-                  expect { execute }
-                    .to change { issue.reload.state }
-                    .from('opened')
-                    .to('closed')
-                end
-
-                if state_tracking_enabled
-                  specify { expect { execute }.to change(ResourceStateEvent, :count).by(1) }
-                else
-                  specify { expect { execute }.to change(Note, :count).by(1) }
-                end
+            context 'existing issue' do
+              let!(:alert) { create(:alert_management_alert, :with_issue, project: project, fingerprint: fingerprint) }
+
+              it 'closes the issue' do
+                issue = alert.issue
+
+                expect { execute }
+                  .to change { issue.reload.state }
+                  .from('opened')
+                  .to('closed')
+              end
+
+              it 'creates a resource state event' do
+                expect { execute }.to change(ResourceStateEvent, :count).by(1)
               end
             end
           end
diff --git a/spec/services/issues/close_service_spec.rb b/spec/services/issues/close_service_spec.rb
index 4db6e5cac12..9076fb11c9b 100644
--- a/spec/services/issues/close_service_spec.rb
+++ b/spec/services/issues/close_service_spec.rb
@@ -233,26 +233,11 @@ RSpec.describe Issues::CloseService do
         expect(email.subject).to include(issue.title)
       end
 
-      context 'when resource state events are disabled' do
-        before do
-          stub_feature_flags(track_resource_state_change_events: false)
-        end
-
-        it 'creates system note about the issue being closed' do
-          close_issue
-
-          note = issue.notes.last
-          expect(note.note).to include "closed"
-        end
-      end
-
-      context 'when resource state events are enabled' do
-        it 'creates resource state event about the issue being closed' do
-          close_issue
+      it 'creates resource state event about the issue being closed' do
+        close_issue
 
-          event = issue.resource_state_events.last
-          expect(event.state).to eq('closed')
-        end
+        event = issue.resource_state_events.last
+        expect(event.state).to eq('closed')
       end
 
       it 'marks todos as done' do
diff --git a/spec/services/merge_requests/close_service_spec.rb b/spec/services/merge_requests/close_service_spec.rb
index e7ac286f48b..67fb4eaade5 100644
--- a/spec/services/merge_requests/close_service_spec.rb
+++ b/spec/services/merge_requests/close_service_spec.rb
@@ -19,54 +19,45 @@ RSpec.describe MergeRequests::CloseService do
   describe '#execute' do
     it_behaves_like 'cache counters invalidator'
 
-    [true, false].each do |state_tracking_enabled|
-      context "valid params with state_tracking #{state_tracking_enabled ? 'enabled' : 'disabled'}" do
-        let(:service) { described_class.new(project, user, {}) }
+    context 'valid params' do
+      let(:service) { described_class.new(project, user, {}) }
 
-        before do
-          stub_feature_flags(track_resource_state_change_events: state_tracking_enabled)
-
-          allow(service).to receive(:execute_hooks)
+      before do
+        allow(service).to receive(:execute_hooks)
 
-          perform_enqueued_jobs do
-            @merge_request = service.execute(merge_request)
-          end
+        perform_enqueued_jobs do
+          @merge_request = service.execute(merge_request)
         end
+      end
 
-        it { expect(@merge_request).to be_valid }
-        it { expect(@merge_request).to be_closed }
+      it { expect(@merge_request).to be_valid }
+      it { expect(@merge_request).to be_closed }
 
-        it 'executes hooks with close action' do
-          expect(service).to have_received(:execute_hooks)
-                               .with(@merge_request, 'close')
-        end
+      it 'executes hooks with close action' do
+        expect(service).to have_received(:execute_hooks)
+                             .with(@merge_request, 'close')
+      end
 
-        it 'sends email to user2 about assign of new merge_request', :sidekiq_might_not_need_inline do
-          email = ActionMailer::Base.deliveries.last
-          expect(email.to.first).to eq(user2.email)
-          expect(email.subject).to include(merge_request.title)
-        end
+      it 'sends email to user2 about assign of new merge_request', :sidekiq_might_not_need_inline do
+        email = ActionMailer::Base.deliveries.last
+        expect(email.to.first).to eq(user2.email)
+        expect(email.subject).to include(merge_request.title)
+      end
 
-        it 'creates system note about merge_request reassign' do
-          if state_tracking_enabled
-            event = @merge_request.resource_state_events.last
-            expect(event.state).to eq('closed')
-          else
-            note = @merge_request.notes.last
-            expect(note.note).to include 'closed'
-          end
-        end
+      it 'creates a resource event' do
+        event = @merge_request.resource_state_events.last
+        expect(event.state).to eq('closed')
+      end
 
-        it 'marks todos as done' do
-          expect(todo.reload).to be_done
-        end
+      it 'marks todos as done' do
+        expect(todo.reload).to be_done
+      end
 
-        context 'when auto merge is enabled' do
-          let(:merge_request) { create(:merge_request, :merge_when_pipeline_succeeds) }
+      context 'when auto merge is enabled' do
+        let(:merge_request) { create(:merge_request, :merge_when_pipeline_succeeds) }
 
-          it 'cancels the auto merge' do
-            expect(@merge_request).not_to be_auto_merge_enabled
-          end
+        it 'cancels the auto merge' do
+          expect(@merge_request).not_to be_auto_merge_enabled
         end
       end
     end
diff --git a/spec/services/merge_requests/ff_merge_service_spec.rb b/spec/services/merge_requests/ff_merge_service_spec.rb
index 55856deeaca..3038e48a184 100644
--- a/spec/services/merge_requests/ff_merge_service_spec.rb
+++ b/spec/services/merge_requests/ff_merge_service_spec.rb
@@ -22,74 +22,65 @@ RSpec.describe MergeRequests::FfMergeService do
   end
 
   describe '#execute' do
-    [true, false].each do |state_tracking_enabled|
-      context "valid params with state_tracking #{state_tracking_enabled ? 'enabled' : 'disabled'}" do
-        let(:service) { described_class.new(project, user, valid_merge_params) }
-
-        def execute_ff_merge
-          perform_enqueued_jobs do
-            service.execute(merge_request)
-          end
-        end
-
-        before do
-          stub_feature_flags(track_resource_state_change_events: state_tracking_enabled)
+    context 'valid params' do
+      let(:service) { described_class.new(project, user, valid_merge_params) }
 
-          allow(service).to receive(:execute_hooks)
+      def execute_ff_merge
+        perform_enqueued_jobs do
+          service.execute(merge_request)
         end
+      end
 
-        it "does not create merge commit" do
-          execute_ff_merge
+      before do
+        allow(service).to receive(:execute_hooks)
+      end
 
-          source_branch_sha = merge_request.source_project.repository.commit(merge_request.source_branch).sha
-          target_branch_sha = merge_request.target_project.repository.commit(merge_request.target_branch).sha
+      it "does not create merge commit" do
+        execute_ff_merge
 
-          expect(source_branch_sha).to eq(target_branch_sha)
-        end
+        source_branch_sha = merge_request.source_project.repository.commit(merge_request.source_branch).sha
+        target_branch_sha = merge_request.target_project.repository.commit(merge_request.target_branch).sha
 
-        it 'keeps the merge request valid' do
-          expect { execute_ff_merge }
-            .not_to change { merge_request.valid? }
-        end
+        expect(source_branch_sha).to eq(target_branch_sha)
+      end
 
-        it 'updates the merge request to merged' do
-          expect { execute_ff_merge }
-            .to change { merge_request.merged? }
-            .from(false)
-            .to(true)
-        end
+      it 'keeps the merge request valid' do
+        expect { execute_ff_merge }
+          .not_to change { merge_request.valid? }
+      end
 
-        it 'sends email to user2 about merge of new merge_request' do
-          execute_ff_merge
+      it 'updates the merge request to merged' do
+        expect { execute_ff_merge }
+          .to change { merge_request.merged? }
+          .from(false)
+          .to(true)
+      end
 
-          email = ActionMailer::Base.deliveries.last
-          expect(email.to.first).to eq(user2.email)
-          expect(email.subject).to include(merge_request.title)
-        end
+      it 'sends email to user2 about merge of new merge_request' do
+        execute_ff_merge
 
-        it 'creates system note about merge_request merge' do
-          execute_ff_merge
+        email = ActionMailer::Base.deliveries.last
+        expect(email.to.first).to eq(user2.email)
+        expect(email.subject).to include(merge_request.title)
+      end
 
-          if state_tracking_enabled
-            event = merge_request.resource_state_events.last
-            expect(event.state).to eq('merged')
-          else
-            note = merge_request.notes.last
-            expect(note.note).to include 'merged'
-          end
-        end
+      it 'creates resource event about merge_request merge' do
+        execute_ff_merge
 
-        it 'does not update squash_commit_sha if it is not a squash' do
-          expect { execute_ff_merge }.not_to change { merge_request.squash_commit_sha }
-        end
+        event = merge_request.resource_state_events.last
+        expect(event.state).to eq('merged')
+      end
 
-        it 'updates squash_commit_sha if it is a squash' do
-          merge_request.update!(squash: true)
+      it 'does not update squash_commit_sha if it is not a squash' do
+        expect { execute_ff_merge }.not_to change { merge_request.squash_commit_sha }
+      end
 
-          expect { execute_ff_merge }
-            .to change { merge_request.squash_commit_sha }
-            .from(nil)
-        end
+      it 'updates squash_commit_sha if it is a squash' do
+        merge_request.update!(squash: true)
+
+        expect { execute_ff_merge }
+          .to change { merge_request.squash_commit_sha }
+          .from(nil)
       end
     end
 
diff --git a/spec/services/merge_requests/merge_service_spec.rb b/spec/services/merge_requests/merge_service_spec.rb
index 8328f461029..184d54fe09b 100644
--- a/spec/services/merge_requests/merge_service_spec.rb
+++ b/spec/services/merge_requests/merge_service_spec.rb
@@ -20,11 +20,7 @@ RSpec.describe MergeRequests::MergeService do
     end
 
     context 'valid params' do
-      let(:state_tracking) { true }
-
       before do
-        stub_feature_flags(track_resource_state_change_events: state_tracking)
-
         allow(service).to receive(:execute_hooks)
 
         perform_enqueued_jobs do
@@ -47,20 +43,9 @@ RSpec.describe MergeRequests::MergeService do
       end
 
       context 'note creation' do
-        context 'when resource state event tracking is disabled' do
-          let(:state_tracking) { false }
-
-          it 'creates system note about merge_request merge' do
-            note = merge_request.notes.last
-            expect(note.note).to include 'merged'
-          end
-        end
-
-        context 'when resource state event tracking is enabled' do
-          it 'creates resource state event about merge_request merge' do
-            event = merge_request.resource_state_events.last
-            expect(event.state).to eq('merged')
-          end
+        it 'creates resource state event about merge_request merge' do
+          event = merge_request.resource_state_events.last
+          expect(event.state).to eq('merged')
         end
       end
 
diff --git a/spec/services/merge_requests/refresh_service_spec.rb b/spec/services/merge_requests/refresh_service_spec.rb
index cace1e0bf09..ca0c4b29ebe 100644
--- a/spec/services/merge_requests/refresh_service_spec.rb
+++ b/spec/services/merge_requests/refresh_service_spec.rb
@@ -367,76 +367,58 @@ RSpec.describe MergeRequests::RefreshService do
       end
     end
 
-    [true, false].each do |state_tracking_enabled|
-      context "push to origin repo target branch with state tracking #{state_tracking_enabled ? 'enabled' : 'disabled'}", :sidekiq_might_not_need_inline do
+    context 'push to origin repo target branch', :sidekiq_might_not_need_inline do
+      context 'when all MRs to the target branch had diffs' do
         before do
-          stub_feature_flags(track_resource_state_change_events: state_tracking_enabled)
+          service.new(@project, @user).execute(@oldrev, @newrev, 'refs/heads/feature')
+          reload_mrs
         end
 
-        context 'when all MRs to the target branch had diffs' do
-          before do
-            service.new(@project, @user).execute(@oldrev, @newrev, 'refs/heads/feature')
-            reload_mrs
-          end
+        it 'updates the merge state' do
+          expect(@merge_request).to be_merged
+          expect(@fork_merge_request).to be_merged
+          expect(@build_failed_todo).to be_done
+          expect(@fork_build_failed_todo).to be_done
 
-          it 'updates the merge state' do
-            expect(@merge_request).to be_merged
-            expect(@fork_merge_request).to be_merged
-            expect(@build_failed_todo).to be_done
-            expect(@fork_build_failed_todo).to be_done
-
-            if state_tracking_enabled
-              expect(@merge_request.resource_state_events.last.state).to eq('merged')
-              expect(@fork_merge_request.resource_state_events.last.state).to eq('merged')
-            else
-              expect(@merge_request.notes.last.note).to include('merged')
-              expect(@fork_merge_request.notes.last.note).to include('merged')
-            end
-          end
+          expect(@merge_request.resource_state_events.last.state).to eq('merged')
+          expect(@fork_merge_request.resource_state_events.last.state).to eq('merged')
         end
+      end
 
-        context 'when an MR to be closed was empty already' do
-          let!(:empty_fork_merge_request) do
-            create(:merge_request,
-                   source_project: @fork_project,
-                   source_branch: 'master',
-                   target_branch: 'master',
-                   target_project: @project)
-          end
+      context 'when an MR to be closed was empty already' do
+        let!(:empty_fork_merge_request) do
+          create(:merge_request,
+                  source_project: @fork_project,
+                  source_branch: 'master',
+                  target_branch: 'master',
+                  target_project: @project)
+        end
 
-          before do
-            # This spec already has a fake push, so pretend that we were targeting
-            # feature all along.
-            empty_fork_merge_request.update_columns(target_branch: 'feature')
+        before do
+          # This spec already has a fake push, so pretend that we were targeting
+          # feature all along.
+          empty_fork_merge_request.update_columns(target_branch: 'feature')
 
-            service.new(@project, @user).execute(@oldrev, @newrev, 'refs/heads/feature')
-            reload_mrs
-            empty_fork_merge_request.reload
-          end
+          service.new(@project, @user).execute(@oldrev, @newrev, 'refs/heads/feature')
+          reload_mrs
+          empty_fork_merge_request.reload
+        end
 
-          it 'only updates the non-empty MRs' do
-            expect(@merge_request).to be_merged
-            expect(@fork_merge_request).to be_merged
-
-            expect(empty_fork_merge_request).to be_open
-            expect(empty_fork_merge_request.merge_request_diff.state).to eq('empty')
-            expect(empty_fork_merge_request.notes).to be_empty
-
-            if state_tracking_enabled
-              expect(@merge_request.resource_state_events.last.state).to eq('merged')
-              expect(@fork_merge_request.resource_state_events.last.state).to eq('merged')
-            else
-              expect(@merge_request.notes.last.note).to include('merged')
-              expect(@fork_merge_request.notes.last.note).to include('merged')
-            end
-          end
+        it 'only updates the non-empty MRs' do
+          expect(@merge_request).to be_merged
+          expect(@fork_merge_request).to be_merged
+
+          expect(empty_fork_merge_request).to be_open
+          expect(empty_fork_merge_request.merge_request_diff.state).to eq('empty')
+          expect(empty_fork_merge_request.notes).to be_empty
+
+          expect(@merge_request.resource_state_events.last.state).to eq('merged')
+          expect(@fork_merge_request.resource_state_events.last.state).to eq('merged')
         end
       end
 
-      context "manual merge of source branch #{state_tracking_enabled ? 'enabled' : 'disabled'}", :sidekiq_might_not_need_inline do
+      context 'manual merge of source branch', :sidekiq_might_not_need_inline do
         before do
-          stub_feature_flags(track_resource_state_change_events: state_tracking_enabled)
-
           # Merge master -> feature branch
           @project.repository.merge(@user, @merge_request.diff_head_sha, @merge_request, 'Test message')
           commit = @project.repository.commit('feature')
@@ -445,13 +427,8 @@ RSpec.describe MergeRequests::RefreshService do
         end
 
         it 'updates the merge state' do
-          if state_tracking_enabled
-            expect(@merge_request.resource_state_events.last.state).to eq('merged')
-            expect(@fork_merge_request.resource_state_events.last.state).to eq('merged')
-          else
-            expect(@merge_request.notes.last.note).to include('merged')
-            expect(@fork_merge_request.notes.last.note).to include('merged')
-          end
+          expect(@merge_request.resource_state_events.last.state).to eq('merged')
+          expect(@fork_merge_request.resource_state_events.last.state).to eq('merged')
 
           expect(@merge_request).to be_merged
           expect(@merge_request.diffs.size).to be > 0
@@ -616,29 +593,21 @@ RSpec.describe MergeRequests::RefreshService do
       end
     end
 
-    [true, false].each do |state_tracking_enabled|
-      context "push to origin repo target branch after fork project was removed #{state_tracking_enabled ? 'enabled' : 'disabled'}" do
-        before do
-          stub_feature_flags(track_resource_state_change_events: state_tracking_enabled)
+    context 'push to origin repo target branch after fork project was removed' do
+      before do
+        @fork_project.destroy!
+        service.new(@project, @user).execute(@oldrev, @newrev, 'refs/heads/feature')
+        reload_mrs
+      end
 
-          @fork_project.destroy!
-          service.new(@project, @user).execute(@oldrev, @newrev, 'refs/heads/feature')
-          reload_mrs
-        end
+      it 'updates the merge request state' do
+        expect(@merge_request.resource_state_events.last.state).to eq('merged')
 
-        it 'updates the merge request state' do
-          if state_tracking_enabled
-            expect(@merge_request.resource_state_events.last.state).to eq('merged')
-          else
-            expect(@merge_request.notes.last.note).to include('merged')
-          end
-
-          expect(@merge_request).to be_merged
-          expect(@fork_merge_request).to be_open
-          expect(@fork_merge_request.notes).to be_empty
-          expect(@build_failed_todo).to be_done
-          expect(@fork_build_failed_todo).to be_done
-        end
+        expect(@merge_request).to be_merged
+        expect(@fork_merge_request).to be_open
+        expect(@fork_merge_request.notes).to be_empty
+        expect(@build_failed_todo).to be_done
+        expect(@fork_build_failed_todo).to be_done
       end
     end
 
diff --git a/spec/services/merge_requests/reopen_service_spec.rb b/spec/services/merge_requests/reopen_service_spec.rb
index 0066834180e..ffc2ebb344c 100644
--- a/spec/services/merge_requests/reopen_service_spec.rb
+++ b/spec/services/merge_requests/reopen_service_spec.rb
@@ -20,11 +20,8 @@ RSpec.describe MergeRequests::ReopenService do
 
     context 'valid params' do
       let(:service) { described_class.new(project, user, {}) }
-      let(:state_tracking) { true }
 
       before do
-        stub_feature_flags(track_resource_state_change_events: state_tracking)
-
         allow(service).to receive(:execute_hooks)
 
         perform_enqueued_jobs do
@@ -47,20 +44,9 @@ RSpec.describe MergeRequests::ReopenService do
       end
 
       context 'note creation' do
-        context 'when state event tracking is disabled' do
-          let(:state_tracking) { false }
-
-          it 'creates system note about merge_request reopen' do
-            note = merge_request.notes.last
-            expect(note.note).to include 'reopened'
-          end
-        end
-
-        context 'when state event tracking is enabled' do
-          it 'creates resource state event about merge_request reopen' do
-            event = merge_request.resource_state_events.last
-            expect(event.state).to eq('reopened')
-          end
+        it 'creates resource state event about merge_request reopen' do
+          event = merge_request.resource_state_events.last
+          expect(event.state).to eq('reopened')
         end
       end
     end
diff --git a/spec/services/projects/alerting/notify_service_spec.rb b/spec/services/projects/alerting/notify_service_spec.rb
index 77a0e330109..68764990886 100644
--- a/spec/services/projects/alerting/notify_service_spec.rb
+++ b/spec/services/projects/alerting/notify_service_spec.rb
@@ -127,23 +127,8 @@ RSpec.describe Projects::Alerting::NotifyService do
                   let(:alert) { create(:alert_management_alert, :with_issue, project: project, fingerprint: fingerprint_sha) }
                   let(:issue) { alert.issue }
 
-                  context 'state_tracking is enabled' do
-                    before do
-                      stub_feature_flags(track_resource_state_change_events: true)
-                    end
-
-                    it { expect { subject }.to change { issue.reload.state }.from('opened').to('closed') }
-                    it { expect { subject }.to change(ResourceStateEvent, :count).by(1) }
-                  end
-
-                  context 'state_tracking is disabled' do
-                    before do
-                      stub_feature_flags(track_resource_state_change_events: false)
-                    end
-
-                    it { expect { subject }.to change { issue.reload.state }.from('opened').to('closed') }
-                    it { expect { subject }.to change(Note, :count).by(1) }
-                  end
+                  it { expect { subject }.to change { issue.reload.state }.from('opened').to('closed') }
+                  it { expect { subject }.to change(ResourceStateEvent, :count).by(1) }
                 end
               end
             end
diff --git a/spec/services/system_notes/issuables_service_spec.rb b/spec/services/system_notes/issuables_service_spec.rb
index fec2a711dc2..f978b09faba 100644
--- a/spec/services/system_notes/issuables_service_spec.rb
+++ b/spec/services/system_notes/issuables_service_spec.rb
@@ -131,43 +131,11 @@ RSpec.describe ::SystemNotes::IssuablesService do
   describe '#change_status' do
     subject { service.change_status(status, source) }
 
-    context 'when resource state event tracking is enabled' do
-      let(:status) { 'reopened' }
-      let(:source) { nil }
+    let(:status) { 'reopened' }
+    let(:source) { nil }
 
-      it 'does not change note count' do
-        expect { subject }.not_to change { Note.count }
-      end
-    end
-
-    context 'with status reopened' do
-      before do
-        stub_feature_flags(track_resource_state_change_events: false)
-      end
-
-      let(:status) { 'reopened' }
-      let(:source) { nil }
-
-      it_behaves_like 'a note with overridable created_at'
-
-      it_behaves_like 'a system note' do
-        let(:action) { 'opened' }
-      end
-    end
-
-    context 'with a source' do
-      before do
-        stub_feature_flags(track_resource_state_change_events: false)
-      end
-
-      let(:status) { 'opened' }
-      let(:source) { double('commit', gfm_reference: 'commit 123456') }
-
-      it_behaves_like 'a note with overridable created_at'
-
-      it 'sets the note text' do
-        expect(subject.note).to eq "#{status} via commit 123456"
-      end
+    it 'creates a resource state event' do
+      expect { subject }.to change { ResourceStateEvent.count }.by(1)
     end
   end
 
@@ -636,67 +604,26 @@ RSpec.describe ::SystemNotes::IssuablesService do
   describe '#close_after_error_tracking_resolve' do
     subject { service.close_after_error_tracking_resolve }
 
-    context 'when state tracking is enabled' do
-      before do
-        stub_feature_flags(track_resource_state_change_events: true)
-      end
-
-      it 'creates the expected state event' do
-        subject
+    it 'creates the expected state event' do
+      subject
 
-        event = ResourceStateEvent.last
+      event = ResourceStateEvent.last
 
-        expect(event.close_after_error_tracking_resolve).to eq(true)
-        expect(event.state).to eq('closed')
-      end
-    end
-
-    context 'when state tracking is disabled' do
-      before do
-        stub_feature_flags(track_resource_state_change_events: false)
-      end
-
-      it_behaves_like 'a system note' do
-        let(:action) { 'closed' }
-      end
-
-      it 'creates the expected system note' do
-        expect(subject.note)
-          .to eq('resolved the corresponding error and closed the issue.')
-      end
+      expect(event.close_after_error_tracking_resolve).to eq(true)
+      expect(event.state).to eq('closed')
     end
   end
 
   describe '#auto_resolve_prometheus_alert' do
     subject { service.auto_resolve_prometheus_alert }
 
-    context 'when state tracking is enabled' do
-      before do
-        stub_feature_flags(track_resource_state_change_events: true)
-      end
-
-      it 'creates the expected state event' do
-        subject
-
-        event = ResourceStateEvent.last
-
-        expect(event.close_auto_resolve_prometheus_alert).to eq(true)
-        expect(event.state).to eq('closed')
-      end
-    end
-
-    context 'when state tracking is disabled' do
-      before do
-        stub_feature_flags(track_resource_state_change_events: false)
-      end
+    it 'creates the expected state event' do
+      subject
 
-      it_behaves_like 'a system note' do
-        let(:action) { 'closed' }
-      end
+      event = ResourceStateEvent.last
 
-      it 'creates the expected system note' do
-        expect(subject.note).to eq('automatically closed this issue because the alert resolved.')
-      end
+      expect(event.close_auto_resolve_prometheus_alert).to eq(true)
+      expect(event.state).to eq('closed')
     end
   end
 end
diff --git a/spec/support/counter_attribute.rb b/spec/support/counter_attribute.rb
index ea71b25b4c0..8bd40b72dcf 100644
--- a/spec/support/counter_attribute.rb
+++ b/spec/support/counter_attribute.rb
@@ -9,6 +9,12 @@ RSpec.configure do |config|
 
       counter_attribute :build_artifacts_size
       counter_attribute :commit_count
+
+      attr_accessor :flushed
+
+      counter_attribute_after_flush do |subject|
+        subject.flushed = true
+      end
     end
   end
 end
diff --git a/spec/support/helpers/multipart_helpers.rb b/spec/support/helpers/multipart_helpers.rb
index f068d5e102d..043cb6e1420 100644
--- a/spec/support/helpers/multipart_helpers.rb
+++ b/spec/support/helpers/multipart_helpers.rb
@@ -1,6 +1,8 @@
 # frozen_string_literal: true
 
 module MultipartHelpers
+  include WorkhorseHelpers
+
   def post_env(rewritten_fields:, params:, secret:, issuer:)
     token = JWT.encode({ 'iss' => issuer, 'rewritten_fields' => rewritten_fields }, secret, 'HS256')
     Rack::MockRequest.env_for(
@@ -29,7 +31,14 @@ module MultipartHelpers
       raise ArgumentError, "can't handle #{mode} mode"
     end
 
-    result
+    return result if ::Feature.disabled?(:upload_middleware_jwt_params_handler)
+
+    # the HandlerForJWTParams expects a jwt token with the upload parameters
+    # *without* the "#{key}." prefix
+    result.deep_transform_keys! { |k| k.remove("#{key}.") }
+    {
+      "#{key}.gitlab-workhorse-upload" => jwt_token('upload' => result)
+    }
   end
 
   # This function assumes a `mode` variable to be set
diff --git a/spec/support/helpers/workhorse_helpers.rb b/spec/support/helpers/workhorse_helpers.rb
index f16b6c1e910..7e95f49aea2 100644
--- a/spec/support/helpers/workhorse_helpers.rb
+++ b/spec/support/helpers/workhorse_helpers.rb
@@ -3,6 +3,8 @@
 module WorkhorseHelpers
   extend self
 
+  UPLOAD_PARAM_NAMES = %w[name size path remote_id sha256 type].freeze
+
   def workhorse_send_data
     @_workhorse_send_data ||= begin
       header = response.headers[Gitlab::Workhorse::SEND_DATA_HEADER]
@@ -59,6 +61,7 @@ module WorkhorseHelpers
       file = workhorse_params.delete(key)
       rewritten_fields[key] = file.path if file
       workhorse_params = workhorse_disk_accelerated_file_params(key, file).merge(workhorse_params)
+      workhorse_params = workhorse_params.merge(jwt_file_upload_param(key: key, params: workhorse_params))
     end
 
     headers = if send_rewritten_field
@@ -74,8 +77,19 @@ module WorkhorseHelpers
 
   private
 
-  def jwt_token(data = {})
-    JWT.encode({ 'iss' => 'gitlab-workhorse' }.merge(data), Gitlab::Workhorse.secret, 'HS256')
+  def jwt_file_upload_param(key:, params:)
+    upload_params = UPLOAD_PARAM_NAMES.map do |file_upload_param|
+      [file_upload_param, params["#{key}.#{file_upload_param}"]]
+    end
+    upload_params = upload_params.to_h.compact
+
+    return {} if upload_params.empty?
+
+    { "#{key}.gitlab-workhorse-upload" => jwt_token('upload' => upload_params) }
+  end
+
+  def jwt_token(data = {}, issuer: 'gitlab-workhorse', secret: Gitlab::Workhorse.secret, algorithm: 'HS256')
+    JWT.encode({ 'iss' => issuer }.merge(data), secret, algorithm)
   end
 
   def workhorse_rewritten_fields_header(fields)
diff --git a/spec/support/shared_examples/features/file_uploads_shared_examples.rb b/spec/support/shared_examples/features/file_uploads_shared_examples.rb
index d586bf03b59..ea8c8d44501 100644
--- a/spec/support/shared_examples/features/file_uploads_shared_examples.rb
+++ b/spec/support/shared_examples/features/file_uploads_shared_examples.rb
@@ -2,6 +2,28 @@
 
 RSpec.shared_examples 'handling file uploads' do |shared_examples_name|
   context 'with object storage disabled' do
-    it_behaves_like shared_examples_name
+    context 'with upload_middleware_jwt_params_handler disabled' do
+      before do
+        stub_feature_flags(upload_middleware_jwt_params_handler: false)
+
+        expect_next_instance_of(Gitlab::Middleware::Multipart::Handler) do |handler|
+          expect(handler).to receive(:with_open_files).and_call_original
+        end
+      end
+
+      it_behaves_like shared_examples_name
+    end
+
+    context 'with upload_middleware_jwt_params_handler enabled' do
+      before do
+        stub_feature_flags(upload_middleware_jwt_params_handler: true)
+
+        expect_next_instance_of(Gitlab::Middleware::Multipart::HandlerForJWTParams) do |handler|
+          expect(handler).to receive(:with_open_files).and_call_original
+        end
+      end
+
+      it_behaves_like shared_examples_name
+    end
   end
 end
diff --git a/spec/support/shared_examples/models/update_project_statistics_shared_examples.rb b/spec/support/shared_examples/models/update_project_statistics_shared_examples.rb
index 557025569b8..7b591ad84d1 100644
--- a/spec/support/shared_examples/models/update_project_statistics_shared_examples.rb
+++ b/spec/support/shared_examples/models/update_project_statistics_shared_examples.rb
@@ -1,6 +1,6 @@
 # frozen_string_literal: true
 
-RSpec.shared_examples 'UpdateProjectStatistics' do
+RSpec.shared_examples 'UpdateProjectStatistics' do |with_counter_attribute|
   let(:project) { subject.project }
   let(:project_statistics_name) { described_class.project_statistics_name }
   let(:statistic_attribute) { described_class.statistic_attribute }
@@ -13,108 +13,230 @@ RSpec.shared_examples 'UpdateProjectStatistics' do
     subject.read_attribute(statistic_attribute).to_i
   end
 
-  it { is_expected.to be_new_record }
+  def read_pending_increment
+    Gitlab::Redis::SharedState.with do |redis|
+      key = project.statistics.counter_key(project_statistics_name)
+      redis.get(key).to_i
+    end
+  end
 
-  context 'when creating' do
-    it 'updates the project statistics' do
-      delta0 = reload_stat
+  it { is_expected.to be_new_record }
 
-      subject.save!
+  context 'when feature flag efficient_counter_attribute is disabled' do
+    before do
+      stub_feature_flags(efficient_counter_attribute: false)
+    end
 
-      delta1 = reload_stat
+    context 'when creating' do
+      it 'updates the project statistics' do
+        delta0 = reload_stat
 
-      expect(delta1).to eq(delta0 + read_attribute)
-      expect(delta1).to be > delta0
-    end
+        subject.save!
 
-    it 'schedules a namespace statistics worker' do
-      expect(Namespaces::ScheduleAggregationWorker)
-        .to receive(:perform_async).once
+        delta1 = reload_stat
 
-      subject.save!
-    end
-  end
+        expect(delta1).to eq(delta0 + read_attribute)
+        expect(delta1).to be > delta0
+      end
 
-  context 'when updating' do
-    let(:delta) { 42 }
+      it 'schedules a namespace statistics worker' do
+        expect(Namespaces::ScheduleAggregationWorker)
+          .to receive(:perform_async).once
 
-    before do
-      subject.save!
+        subject.save!
+      end
     end
 
-    it 'updates project statistics' do
-      expect(ProjectStatistics)
-        .to receive(:increment_statistic)
-        .and_call_original
+    context 'when updating' do
+      let(:delta) { 42 }
 
-      subject.write_attribute(statistic_attribute, read_attribute + delta)
+      before do
+        subject.save!
+      end
 
-      expect { subject.save! }
-        .to change { reload_stat }
-        .by(delta)
-    end
+      it 'updates project statistics' do
+        expect(ProjectStatistics)
+          .to receive(:increment_statistic)
+          .and_call_original
 
-    it 'schedules a namespace statistics worker' do
-      expect(Namespaces::ScheduleAggregationWorker)
-        .to receive(:perform_async).once
+        subject.write_attribute(statistic_attribute, read_attribute + delta)
 
-      subject.write_attribute(statistic_attribute, read_attribute + delta)
-      subject.save!
-    end
+        expect { subject.save! }
+          .to change { reload_stat }
+          .by(delta)
+      end
 
-    it 'avoids N + 1 queries' do
-      subject.write_attribute(statistic_attribute, read_attribute + delta)
+      it 'schedules a namespace statistics worker' do
+        expect(Namespaces::ScheduleAggregationWorker)
+          .to receive(:perform_async).once
 
-      control_count = ActiveRecord::QueryRecorder.new do
+        subject.write_attribute(statistic_attribute, read_attribute + delta)
         subject.save!
       end
 
-      subject.write_attribute(statistic_attribute, read_attribute + delta)
+      it 'avoids N + 1 queries' do
+        subject.write_attribute(statistic_attribute, read_attribute + delta)
 
-      expect do
-        subject.save!
-      end.not_to exceed_query_limit(control_count)
-    end
-  end
+        control_count = ActiveRecord::QueryRecorder.new do
+          subject.save!
+        end
 
-  context 'when destroying' do
-    before do
-      subject.save!
+        subject.write_attribute(statistic_attribute, read_attribute + delta)
+
+        expect do
+          subject.save!
+        end.not_to exceed_query_limit(control_count)
+      end
     end
 
-    it 'updates the project statistics' do
-      delta0 = reload_stat
+    context 'when destroying' do
+      before do
+        subject.save!
+      end
 
-      subject.destroy!
+      it 'updates the project statistics' do
+        delta0 = reload_stat
 
-      delta1 = reload_stat
+        subject.destroy!
 
-      expect(delta1).to eq(delta0 - read_attribute)
-      expect(delta1).to be < delta0
-    end
+        delta1 = reload_stat
+
+        expect(delta1).to eq(delta0 - read_attribute)
+        expect(delta1).to be < delta0
+      end
+
+      it 'schedules a namespace statistics worker' do
+        expect(Namespaces::ScheduleAggregationWorker)
+          .to receive(:perform_async).once
 
-    it 'schedules a namespace statistics worker' do
-      expect(Namespaces::ScheduleAggregationWorker)
-        .to receive(:perform_async).once
+        subject.destroy!
+      end
+
+      context 'when it is destroyed from the project level' do
+        it 'does not update the project statistics' do
+          expect(ProjectStatistics)
+            .not_to receive(:increment_statistic)
+
+          project.update!(pending_delete: true)
+          project.destroy!
+        end
+
+        it 'does not schedule a namespace statistics worker' do
+          expect(Namespaces::ScheduleAggregationWorker)
+            .not_to receive(:perform_async)
 
-      subject.destroy!
+          project.update!(pending_delete: true)
+          project.destroy!
+        end
+      end
     end
+  end
 
-    context 'when it is destroyed from the project level' do
-      it 'does not update the project statistics' do
-        expect(ProjectStatistics)
-          .not_to receive(:increment_statistic)
+  def expect_flush_counter_increments_worker_performed
+    expect(FlushCounterIncrementsWorker)
+      .to receive(:perform_in)
+      .with(CounterAttribute::WORKER_DELAY, project.statistics.class.name, project.statistics.id, project_statistics_name)
+    expect(FlushCounterIncrementsWorker)
+      .to receive(:perform_in)
+      .with(CounterAttribute::WORKER_DELAY, project.statistics.class.name, project.statistics.id, :storage_size)
 
-        project.update!(pending_delete: true)
-        project.destroy!
+    yield
+
+    # simulate worker running now
+    expect(Namespaces::ScheduleAggregationWorker).to receive(:perform_async)
+    FlushCounterIncrementsWorker.new.perform(project.statistics.class.name, project.statistics.id, project_statistics_name)
+  end
+
+  if with_counter_attribute
+    context 'when statistic is a counter attribute', :clean_gitlab_redis_shared_state do
+      context 'when creating' do
+        it 'stores pending increments for async update' do
+          initial_stat = reload_stat
+          expected_increment = read_attribute
+
+          expect_flush_counter_increments_worker_performed do
+            subject.save!
+
+            expect(read_pending_increment).to eq(expected_increment)
+            expect(expected_increment).to be > initial_stat
+            expect(expected_increment).to be_positive
+          end
+        end
       end
 
-      it 'does not schedule a namespace statistics worker' do
-        expect(Namespaces::ScheduleAggregationWorker)
-          .not_to receive(:perform_async)
+      context 'when updating' do
+        let(:delta) { 42 }
+
+        before do
+          subject.save!
+          redis_shared_state_cleanup!
+        end
+
+        it 'stores pending increments for async update' do
+          expect(ProjectStatistics)
+            .to receive(:increment_statistic)
+            .and_call_original
+
+          subject.write_attribute(statistic_attribute, read_attribute + delta)
+
+          expect_flush_counter_increments_worker_performed do
+            subject.save!
+
+            expect(read_pending_increment).to eq(delta)
+          end
+        end
+
+        it 'avoids N + 1 queries' do
+          subject.write_attribute(statistic_attribute, read_attribute + delta)
+
+          control_count = ActiveRecord::QueryRecorder.new do
+            subject.save!
+          end
+
+          subject.write_attribute(statistic_attribute, read_attribute + delta)
+
+          expect do
+            subject.save!
+          end.not_to exceed_query_limit(control_count)
+        end
+      end
 
-        project.update!(pending_delete: true)
-        project.destroy!
+      context 'when destroying' do
+        before do
+          subject.save!
+          redis_shared_state_cleanup!
+        end
+
+        it 'stores pending increment for async update' do
+          initial_stat = reload_stat
+          expected_increment = -read_attribute
+
+          expect_flush_counter_increments_worker_performed do
+            subject.destroy!
+
+            expect(read_pending_increment).to eq(expected_increment)
+            expect(expected_increment).to be < initial_stat
+            expect(expected_increment).to be_negative
+          end
+        end
+
+        context 'when it is destroyed from the project level' do
+          it 'does not update the project statistics' do
+            expect(ProjectStatistics)
+              .not_to receive(:increment_statistic)
+
+            project.update!(pending_delete: true)
+            project.destroy!
+          end
+
+          it 'does not schedule a namespace statistics worker' do
+            expect(Namespaces::ScheduleAggregationWorker)
+              .not_to receive(:perform_async)
+
+            project.update!(pending_delete: true)
+            project.destroy!
+          end
+        end
       end
     end
   end