diff --git a/app/graphql/graphql_triggers.rb b/app/graphql/graphql_triggers.rb
index 6d8b15ea48b2c995a73dbd4097f7cc2296440cf0..848e034e99c1b059df4a8bd4a3e62f677e41e78d 100644
--- a/app/graphql/graphql_triggers.rb
+++ b/app/graphql/graphql_triggers.rb
@@ -1,10 +1,10 @@
 # frozen_string_literal: true
 
 module GraphqlTriggers
-  def self.ci_job_created(job)
+  def self.ci_job_processed(job)
     return unless Feature.enabled?(:ci_job_created_subscription, job.project)
 
-    GitlabSchema.subscriptions.trigger(:ci_job_created, { project_id: job.project.to_gid }, job)
+    GitlabSchema.subscriptions.trigger(:ci_job_processed, { project_id: job.project.to_gid }, job)
   end
 
   def self.ci_job_status_updated(job)
diff --git a/app/graphql/subscriptions/ci/jobs/job_created.rb b/app/graphql/subscriptions/ci/jobs/job_created.rb
deleted file mode 100644
index 445a58dc9120cdef77e435315b7a96b6fad65145..0000000000000000000000000000000000000000
--- a/app/graphql/subscriptions/ci/jobs/job_created.rb
+++ /dev/null
@@ -1,40 +0,0 @@
-# frozen_string_literal: true
-
-module Subscriptions
-  module Ci
-    module Jobs
-      class JobCreated < Subscriptions::BaseSubscription
-        include Gitlab::Graphql::Laziness
-
-        argument :project_id,
-          ::Types::GlobalIDType[::Project],
-          required: true,
-          description: 'Global ID of the project.'
-
-        payload_type Types::Ci::JobType
-
-        def authorized?(project_id:)
-          project = force(GitlabSchema.find_by_gid(project_id))
-
-          unauthorized! unless project
-          unauthorized! unless Ability.allowed?(current_user, :read_build, project)
-
-          true
-        end
-
-        def update(project_id:)
-          created_job = object
-
-          return NO_UPDATE unless created_job
-
-          project = force(GitlabSchema.find_by_gid(project_id))
-
-          return NO_UPDATE unless project && created_job.project_id == project.id
-          return NO_UPDATE unless Ability.allowed?(current_user, :read_build, created_job)
-
-          created_job
-        end
-      end
-    end
-  end
-end
diff --git a/app/graphql/subscriptions/ci/jobs/job_processed.rb b/app/graphql/subscriptions/ci/jobs/job_processed.rb
new file mode 100644
index 0000000000000000000000000000000000000000..afc4b088ad983b35522be07855e83d3ec8caa2bb
--- /dev/null
+++ b/app/graphql/subscriptions/ci/jobs/job_processed.rb
@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+module Subscriptions
+  module Ci
+    module Jobs
+      class JobProcessed < Subscriptions::BaseSubscription
+        include Gitlab::Graphql::Laziness
+
+        argument :project_id,
+          ::Types::GlobalIDType[::Project],
+          required: true,
+          description: 'Global ID of the project.'
+
+        argument :with_artifacts,
+          GraphQL::Types::Boolean,
+          required: false,
+          description: 'Indicates if the job contains artifacts.'
+
+        payload_type Types::Ci::JobType
+
+        def authorized?(project_id:, **_kwargs)
+          project = force(GitlabSchema.find_by_gid(project_id))
+
+          unauthorized! unless project
+          unauthorized! unless Ability.allowed?(current_user, :read_build, project)
+
+          true
+        end
+
+        def update(project_id:, with_artifacts: false)
+          processed_job = object
+
+          return NO_UPDATE unless processed_job
+
+          project = force(GitlabSchema.find_by_gid(project_id))
+
+          return NO_UPDATE unless project && processed_job.project_id == project.id
+          return NO_UPDATE unless Ability.allowed?(current_user, :read_build, processed_job)
+
+          if processed_job.has_job_artifacts? && with_artifacts
+            return processed_job
+          elsif !processed_job.has_job_artifacts? && with_artifacts
+            return NO_UPDATE
+          end
+
+          processed_job
+        end
+      end
+    end
+  end
+end
diff --git a/app/graphql/types/subscription_type.rb b/app/graphql/types/subscription_type.rb
index bac1ab5ba98ae2d8f6853cfe7183c7f25c061782..9fafa7a9649d62dbeb380c574f4bafca501b9b77 100644
--- a/app/graphql/types/subscription_type.rb
+++ b/app/graphql/types/subscription_type.rb
@@ -4,9 +4,9 @@ module Types
   class SubscriptionType < ::Types::BaseObject
     graphql_name 'Subscription'
 
-    field :ci_job_created,
-      subscription: Subscriptions::Ci::Jobs::JobCreated, null: true,
-      description: 'Triggered when a job is created.'
+    field :ci_job_processed,
+      subscription: Subscriptions::Ci::Jobs::JobProcessed, null: true,
+      description: 'Triggered when a job changes state.'
 
     field :ci_job_status_updated,
       subscription: Subscriptions::Ci::Jobs::StatusUpdated, null: true,
diff --git a/app/models/ci/build.rb b/app/models/ci/build.rb
index 952eca00bcce588bc8571329655d4a5bf37714f6..d34e0f7f06350bc25c771e9e02e2cd4923f6d386 100644
--- a/app/models/ci/build.rb
+++ b/app/models/ci/build.rb
@@ -255,7 +255,6 @@ class Build < Ci::Processable
       run_after_commit { build.execute_hooks }
     end
 
-    after_commit :trigger_job_create_subscription, on: :create
     after_commit :track_ci_secrets_management_id_tokens_usage, on: :create, if: :id_tokens?
     after_commit :track_ci_build_created_event, on: :create
 
@@ -417,6 +416,7 @@ def supported_keyset_orderings
       after_transition any => any do |build|
         build.run_after_commit do
           trigger_job_status_change_subscription
+          trigger_job_processed_subscription
         end
       end
     end
@@ -465,15 +465,15 @@ def trigger_job_status_change_subscription
       GraphqlTriggers.ci_job_status_updated(self)
     end
 
-    def trigger_job_create_subscription
+    def trigger_job_processed_subscription
       return unless Feature.enabled?(:ci_job_created_subscription, project)
 
       return if Gitlab::ApplicationRateLimiter.throttled?(
-        :ci_job_created_subscription,
+        :ci_job_processed_subscription,
         scope: project
       )
 
-      GraphqlTriggers.ci_job_created(self)
+      GraphqlTriggers.ci_job_processed(self)
     end
 
     # A Ci::Bridge may transition to `canceling` as a result of strategy: :depend
diff --git a/lib/gitlab/application_rate_limiter.rb b/lib/gitlab/application_rate_limiter.rb
index 2ee7b821c1b1c5a651108d0c5b093d7240e31aa4..11538ea17d9fc4f89bbc32606d5f53ef4cba9d02 100644
--- a/lib/gitlab/application_rate_limiter.rb
+++ b/lib/gitlab/application_rate_limiter.rb
@@ -25,7 +25,7 @@ def rate_limits # rubocop:disable Metrics/AbcSize
           autocomplete_users_unauthenticated: { threshold: -> { application_settings.autocomplete_users_unauthenticated_limit }, interval: 1.minute },
           bulk_delete_todos: { threshold: 6, interval: 1.minute },
           bulk_import: { threshold: 6, interval: 1.minute },
-          ci_job_created_subscription: { threshold: 50, interval: 1.minute },
+          ci_job_processed_subscription: { threshold: 50, interval: 1.minute },
           code_suggestions_api_endpoint: { threshold: -> { application_settings.code_suggestions_api_rate_limit }, interval: 1.minute },
           create_organization_api: { threshold: -> { application_settings.create_organization_api_limit }, interval: 1.minute },
           delete_all_todos: { threshold: 1, interval: 5.minutes },
diff --git a/spec/graphql/graphql_triggers_spec.rb b/spec/graphql/graphql_triggers_spec.rb
index 5875516c883a14a34911ad324a3f149c68f47230..070bd9d5318c8f01b7e985e9b83dda35d584c22f 100644
--- a/spec/graphql/graphql_triggers_spec.rb
+++ b/spec/graphql/graphql_triggers_spec.rb
@@ -245,17 +245,17 @@
     end
   end
 
-  describe '.ci_job_created' do
+  describe '.ci_job_processed' do
     let_it_be(:job) { create(:ci_build) }
 
-    it 'triggers the ci_job_created subscription' do
+    it 'triggers the ci_job_processed subscription' do
       expect(GitlabSchema.subscriptions).to receive(:trigger).with(
-        :ci_job_created,
+        :ci_job_processed,
         { project_id: job.project.to_gid },
         job
       )
 
-      described_class.ci_job_created(job)
+      described_class.ci_job_processed(job)
     end
 
     describe 'when ci_job_created_subscription is disabled' do
@@ -263,14 +263,14 @@
         stub_feature_flags(ci_job_created_subscription: false)
       end
 
-      it 'does not trigger the ci_job_created subscription' do
+      it 'does not trigger the ci_job_processed subscription' do
         expect(GitlabSchema.subscriptions).not_to receive(:trigger).with(
-          :ci_job_created,
+          :ci_job_processed,
           { job_id: job.project.to_gid },
           job
         )
 
-        described_class.ci_job_created(job)
+        described_class.ci_job_processed(job)
       end
     end
   end
diff --git a/spec/graphql/subscriptions/ci/jobs/job_created_spec.rb b/spec/graphql/subscriptions/ci/jobs/job_processed_spec.rb
similarity index 55%
rename from spec/graphql/subscriptions/ci/jobs/job_created_spec.rb
rename to spec/graphql/subscriptions/ci/jobs/job_processed_spec.rb
index bfe4f03cf2642bd5f75f2747cf56f1c03fd06185..35abe41587b69be7cfbc2012c5192d5ab23419ef 100644
--- a/spec/graphql/subscriptions/ci/jobs/job_created_spec.rb
+++ b/spec/graphql/subscriptions/ci/jobs/job_processed_spec.rb
@@ -2,21 +2,23 @@
 
 require 'spec_helper'
 
-RSpec.describe Subscriptions::Ci::Jobs::JobCreated, feature_category: :continuous_integration do
+RSpec.describe Subscriptions::Ci::Jobs::JobProcessed, feature_category: :continuous_integration do
   include GraphqlHelpers
 
-  it { expect(described_class).to have_graphql_arguments(:project_id) }
+  it { expect(described_class).to have_graphql_arguments(:project_id, :with_artifacts) }
   it { expect(described_class.payload_type).to eq(Types::Ci::JobType) }
 
   describe '#resolve' do
     let_it_be(:unauthorized_user) { create(:user) }
     let_it_be(:project) { create(:project) }
-    let_it_be(:job) { create(:ci_build, project: project) }
+    let_it_be(:pipeline) { create(:ci_pipeline, project: project) }
+
+    let_it_be(:job) { create(:ci_build, name: 'job_without_artifacts', pipeline: pipeline) }
 
     let(:current_user) { job.project.owners.first }
     let(:project_id) { project.to_gid }
 
-    subject(:subscription) { resolver.resolve_with_support(project_id: project_id) }
+    let(:subscription) { resolver.resolve_with_support(project_id: project_id) }
 
     context 'when initially subscribing to a projects jobs' do
       let(:resolver) { resolver_instance(described_class, ctx: query_context, subscription_update: false) }
@@ -47,6 +49,10 @@
         resolver_instance(described_class, obj: job, ctx: query_context, subscription_update: true)
       end
 
+      let(:resolver_with_artifacts) do
+        resolver_instance(described_class, obj: job_with_artifact, ctx: query_context, subscription_update: true)
+      end
+
       it 'returns the resolved object' do
         expect(subscription).to eq(job)
       end
@@ -62,6 +68,30 @@
           expect(subscription).to be_an(GraphQL::Execution::Skip)
         end
       end
+
+      context 'with job artifacts' do
+        let_it_be_with_reload(:job_with_artifact) do
+          create(:ci_build, :success, name: 'job_with_artifacts', pipeline: pipeline)
+        end
+
+        let_it_be_with_reload(:artifact) { create(:ci_job_artifact, job: job_with_artifact) }
+
+        let(:subscription_with_artifacts) do
+          resolver_with_artifacts.resolve_with_support(project_id: project_id, with_artifacts: true)
+        end
+
+        let(:subscription_no_job_artifacts) do
+          resolver.resolve_with_support(project_id: project_id, with_artifacts: true)
+        end
+
+        it 'returns the resolved object with artifacts' do
+          expect(subscription_with_artifacts).to eq(job_with_artifact)
+        end
+
+        it 'skips the update if the job has no artifacts but the argument `with_artifacts` was supplied' do
+          expect(subscription_no_job_artifacts).to be_an(GraphQL::Execution::Skip)
+        end
+      end
     end
   end
 end
diff --git a/spec/graphql/types/subscription_type_spec.rb b/spec/graphql/types/subscription_type_spec.rb
index 8f566c03742db22613bbf5008696ded0d93abe43..30a4d6baac7d1c8fce132e94dda9d21045ff68fe 100644
--- a/spec/graphql/types/subscription_type_spec.rb
+++ b/spec/graphql/types/subscription_type_spec.rb
@@ -5,7 +5,7 @@
 RSpec.describe GitlabSchema.types['Subscription'], feature_category: :subscription_management do
   it 'has the expected fields' do
     expected_fields = %i[
-      ci_job_created
+      ci_job_processed
       ci_pipeline_schedule_status_updated
       ci_job_status_updated
       ci_pipeline_status_updated
diff --git a/spec/models/ci/build_spec.rb b/spec/models/ci/build_spec.rb
index a379ebb91cd3f34d325f228723eebddd392c0093..093453ef51628575c3f1c0ce2171f815067a1cbb 100644
--- a/spec/models/ci/build_spec.rb
+++ b/spec/models/ci/build_spec.rb
@@ -257,46 +257,63 @@
       context 'with rate limiting enabled' do
         before do
           allow(Gitlab::ApplicationRateLimiter).to receive(:throttled?)
-                                                     .with(:ci_job_created_subscription, scope: build.project)
+                                                     .with(:ci_job_processed_subscription, scope: build.project)
                                                      .and_return(true)
         end
 
-        it 'does not trigger GraphQL subscription when rate limited' do
-          expect_next(described_class).to receive(:execute_hooks)
-          expect(GraphqlTriggers).not_to receive(:ci_job_created)
+        %w[cancel! drop! run! skip! success!].each do |action|
+          shared_examples "when build receives #{action} event" do
+            it 'does not trigger GraphQL subscription ciJobProcessed' do
+              expect(GraphqlTriggers).not_to receive(:ci_job_processed).with(build)
 
-          create(:ci_build, pipeline: pipeline)
+              build.public_send(action)
+            end
+          end
         end
       end
 
       context 'without rate limiting enabled' do
         before do
           allow(Gitlab::ApplicationRateLimiter).to receive(:throttled?)
-                                                     .with(:ci_job_created_subscription, scope: build.project)
+                                                     .with(:ci_job_processed_subscription, scope: build.project)
                                                      .and_return(false)
         end
 
-        context 'with feature flag enabled' do
-          it 'triggers GraphQL subscription ciJobCreated' do
-            expect_next(described_class).to receive(:execute_hooks)
+        context 'with ci_job_created_subscription turned on' do
+          %w[cancel! drop! run! skip! success!].each do |action|
+            shared_examples "when build receives #{action} event" do
+              it 'triggers GraphQL subscription ciJobProcessed' do
+                expect(GraphqlTriggers).to receive(:ci_job_processed).with(build)
+
+                build.public_send(action)
+              end
+            end
+
+            it_behaves_like "when build receives #{action} event"
 
-            expect(GraphqlTriggers).to receive(:ci_job_created).with(instance_of(described_class))
+            context 'when FF `stop_writing_builds_metadata` is disabled' do
+              before do
+                stub_feature_flags(stop_writing_builds_metadata: false)
+              end
 
-            create(:ci_build, pipeline: pipeline)
+              it_behaves_like "when build receives #{action} event"
+            end
           end
         end
 
-        context 'with feature flag turned off' do
+        context 'with ci_job_created_subscription feature flag turned off' do
           before do
             stub_feature_flags(ci_job_created_subscription: false)
           end
 
-          it 'does not trigger GraphQL subscription ciJobCreated' do
-            expect_next(described_class).to receive(:execute_hooks)
+          %w[cancel! drop! run! skip! success!].each do |action|
+            shared_examples "when build receives #{action} event" do
+              it 'does not trigger GraphQL subscription ciJobProcessed' do
+                expect(GraphqlTriggers).not_to receive(:ci_job_processed).with(build)
 
-            expect(GraphqlTriggers).not_to receive(:ci_job_created).with(instance_of(described_class))
-
-            create(:ci_build, pipeline: pipeline)
+                build.public_send(action)
+              end
+            end
           end
         end
       end