diff --git a/config/cloud_controller.yml b/config/cloud_controller.yml
index 45468fd1e81..60561d41a22 100644
--- a/config/cloud_controller.yml
+++ b/config/cloud_controller.yml
@@ -275,6 +275,14 @@ rate_limiter_v2_api:
   global_admin_limit: 20000
   reset_interval_in_minutes: 60
 
+concurrency_rate_limiter:
+  enabled: false
+  blocking_limit: 10
+  logging_limit: 10
+  redis_connection_pool_size: 40
+  redis_counter_ttl_seconds: 60
+
+
 temporary_enable_v2: true
 
 max_concurrent_service_broker_requests: 0
diff --git a/errors/v2.yml b/errors/v2.yml
index 710809685fd..259992e0f47 100644
--- a/errors/v2.yml
+++ b/errors/v2.yml
@@ -128,6 +128,17 @@
   http_code: 422
   message: "The %s is being deleted"
 
+10021:
+  name: ConcurrentRequestLimitExceeded
+  http_code: 429
+  message: "Too many concurrent requests. Please retry."
+
+10022:
+  name: IPBasedConcurrentRequestLimitExceeded
+  http_code: 429
+  message: "Too many concurrent requests from this IP. Please retry."
+
+
 20001:
   name: UserInvalid
   http_code: 400
diff --git a/lib/cloud_controller/config_schemas/api_schema.rb b/lib/cloud_controller/config_schemas/api_schema.rb
index a4b041e3684..d210704c7e3 100644
--- a/lib/cloud_controller/config_schemas/api_schema.rb
+++ b/lib/cloud_controller/config_schemas/api_schema.rb
@@ -395,6 +395,14 @@ class ApiSchema < VCAP::Config
             reset_interval_in_minutes: Integer
           },
 
+          optional(:concurrency_rate_limiter) => {
+            enabled: bool,
+            optional(:blocking_limit) => Integer,
+            optional(:logging_limit) => Integer,
+            optional(:redis_connection_pool_size) => Integer,
+            optional(:redis_counter_ttl_seconds) => Integer
+          },
+
           optional(:temporary_enable_v2) => bool,
 
           allow_app_ssh_access: bool,
diff --git a/lib/cloud_controller/rack_app_builder.rb b/lib/cloud_controller/rack_app_builder.rb
index 7ceae938cda..84dacbec720 100644
--- a/lib/cloud_controller/rack_app_builder.rb
+++ b/lib/cloud_controller/rack_app_builder.rb
@@ -7,13 +7,16 @@
 require 'rate_limiter'
 require 'service_broker_rate_limiter'
 require 'rate_limiter_v2_api'
+require 'concurrency_rate_limiter'
 require 'new_relic_custom_attributes'
 require 'zipkin'
 require 'block_v3_only_roles'
 require 'below_min_cli_warning'
+require 'user_context_setter'
 
 module VCAP::CloudController
   class RackAppBuilder
+    # rubocop:disable Metrics/MethodLength, Metrics/BlockLength
     def build(config, request_metrics, request_logs)
       token_decoder = VCAP::CloudController::UaaTokenDecoder.new(config.get(:uaa))
       configurer = VCAP::CloudController::Security::SecurityContextConfigurer.new(token_decoder)
@@ -21,14 +24,23 @@ def build(config, request_metrics, request_logs)
       logger = access_log(config)
 
       Rack::Builder.new do
-        use CloudFoundry::Middleware::RequestMetrics, request_metrics
         use CloudFoundry::Middleware::Cors, config.get(:allowed_cors_domains)
         use CloudFoundry::Middleware::VcapRequestContextSetter
         use CloudFoundry::Middleware::BelowMinCliWarning if config.get(:warn_if_below_min_cli_version)
         use CloudFoundry::Middleware::NewRelicCustomAttributes if config.get(:newrelic_enabled)
         use CloudFoundry::Middleware::SecurityContextSetter, configurer
         use CloudFoundry::Middleware::Zipkin
-        use CloudFoundry::Middleware::RequestLogs, request_logs
+
+        if config.get(:concurrency_rate_limiter, :enabled)
+          use CloudFoundry::Middleware::ConcurrencyRateLimiter, {
+            logger: Steno.logger('cc.concurrency_rate_limiter'),
+            blocking_limit: config.get(:concurrency_rate_limiter, :blocking_limit),
+            logging_limit: config.get(:concurrency_rate_limiter, :logging_limit),
+            redis_connection_pool_size: config.get(:concurrency_rate_limiter, :redis_connection_pool_size),
+            redis_counter_ttl_seconds: config.get(:concurrency_rate_limiter, :redis_counter_ttl_seconds)
+          }
+        end
+
         if config.get(:rate_limiter, :enabled)
           use CloudFoundry::Middleware::RateLimiter, {
             logger: Steno.logger('cc.rate_limiter'),
@@ -59,6 +71,10 @@ def build(config, request_metrics, request_logs)
           }
         end
 
+        use CloudFoundry::Middleware::RequestMetrics, request_metrics
+        use CloudFoundry::Middleware::UserContextSetter, configurer
+        use CloudFoundry::Middleware::RequestLogs, request_logs
+
         use CloudFoundry::Middleware::CefLogs, Logger.new(config.get(:security_event_logging, :file)), config.get(:local_route) if config.get(:security_event_logging, :enabled)
         use Rack::CommonLogger, logger if logger
 
@@ -76,6 +92,7 @@ def build(config, request_metrics, request_logs)
         end
       end
     end
+    # rubocop:enable Metrics/MethodLength, Metrics/BlockLength
 
     private
 
diff --git a/lib/cloud_controller/security/security_context_configurer.rb b/lib/cloud_controller/security/security_context_configurer.rb
index cef91925e84..0ccdcff5430 100644
--- a/lib/cloud_controller/security/security_context_configurer.rb
+++ b/lib/cloud_controller/security/security_context_configurer.rb
@@ -6,14 +6,31 @@ def initialize(token_decoder)
       end
 
       def configure(header_token)
+        configure_token_only(header_token)
+        return unless VCAP::CloudController::SecurityContext.valid_token?
+
+        configure_user
+      rescue VCAP::CloudController::UaaTokenDecoder::BadToken
+        VCAP::CloudController::SecurityContext.set(nil, :invalid_token, header_token)
+      end
+
+      def configure_token_only(header_token)
         VCAP::CloudController::SecurityContext.clear
         decoded_token = decode_token(header_token)
+        VCAP::CloudController::SecurityContext.set_token_only(decoded_token, header_token)
+      rescue VCAP::CloudController::UaaTokenDecoder::BadToken
+        VCAP::CloudController::SecurityContext.set(nil, :invalid_token, header_token)
+      end
+
+      def configure_user
+        decoded_token = VCAP::CloudController::SecurityContext.token
+        return unless decoded_token && decoded_token != :invalid_token
 
         user = user_from_token(decoded_token)
         set_is_oauth_client(user, decoded_token)
-        VCAP::CloudController::SecurityContext.set(user, decoded_token, header_token)
+        VCAP::CloudController::SecurityContext.set(user, decoded_token, VCAP::CloudController::SecurityContext.auth_token)
       rescue VCAP::CloudController::UaaTokenDecoder::BadToken
-        VCAP::CloudController::SecurityContext.set(nil, :invalid_token, header_token)
+        VCAP::CloudController::SecurityContext.set(nil, :invalid_token, VCAP::CloudController::SecurityContext.auth_token)
       end
 
       private
diff --git a/lib/cloud_controller/security_context.rb b/lib/cloud_controller/security_context.rb
index d0e55b87325..82015c5abb8 100644
--- a/lib/cloud_controller/security_context.rb
+++ b/lib/cloud_controller/security_context.rb
@@ -12,6 +12,12 @@ def self.set(user, token=nil, auth_token=nil)
       Thread.current[:vcap_auth_token] = auth_token
     end
 
+    def self.set_token_only(token, auth_token=nil)
+      Thread.current[:vcap_user] = nil
+      Thread.current[:vcap_token] = token
+      Thread.current[:vcap_auth_token] = auth_token
+    end
+
     def self.current_user
       Thread.current[:vcap_user]
     end
diff --git a/middleware/concurrency_rate_limiter.rb b/middleware/concurrency_rate_limiter.rb
new file mode 100644
index 00000000000..398f3ff325b
--- /dev/null
+++ b/middleware/concurrency_rate_limiter.rb
@@ -0,0 +1,228 @@
+require 'mixins/client_ip'
+
+module CloudFoundry
+  module Middleware
+    class StoreError < StandardError; end
+
+    class ConcurrentRedisStore
+      def initialize(redis, counter_ttl_seconds: nil)
+        @redis = redis
+        @counter_ttl_seconds = counter_ttl_seconds
+      end
+
+      def self.new_socket(socket, connection_pool_size: nil, counter_ttl_seconds: nil)
+        connection_pool_size ||= VCAP::CloudController::Config.config.get(:puma, :max_threads) || 1
+        redis = ConnectionPool::Wrapper.new(size: connection_pool_size) do
+          Redis.new(timeout: 1, path: socket)
+        end
+        new(redis, counter_ttl_seconds: counter_ttl_seconds)
+      end
+
+      def increment(key, logger)
+        count = @redis.incr(key).to_i
+        @redis.expire(key, @counter_ttl_seconds) if @counter_ttl_seconds
+        count
+      rescue Redis::BaseError => e
+        logger.error("Redis error: #{e.class} - #{e.message}")
+        raise StoreError.new("increment failed: #{e.message}")
+      end
+
+      def decrement(key, logger)
+        count = @redis.decr(key).to_i
+        @redis.incr(key) if count < 0
+        [count, 0].max
+      rescue Redis::BaseError => e
+        logger.error("Redis error: #{e.class} - #{e.message}")
+        raise StoreError.new("decrement failed: #{e.message}")
+      end
+    end
+
+    class ConcurrentInMemoryStore
+      def initialize
+        @mutex = Mutex.new
+        @data = {}
+      end
+
+      def increment(key, _logger)
+        @mutex.synchronize do
+          @data[key] = (@data[key] || 0) + 1
+        end
+      end
+
+      def decrement(key, _logger)
+        @mutex.synchronize do
+          return 0 unless @data.key?(key)
+
+          @data[key] -= 1
+          @data.delete(key) if @data[key] <= 0
+          @data[key] || 0
+        end
+      end
+    end
+
+    class ConcurrencyLimiter
+      @instance_mutex = Mutex.new
+
+      def self.instance(logger, blocking_limit: nil, logging_limit: nil, redis_connection_pool_size: nil, redis_counter_ttl_seconds: nil)
+        return @instance if @instance
+
+        @instance_mutex.synchronize do
+          @instance ||= new(logger,
+                            blocking_limit: blocking_limit,
+                            logging_limit: logging_limit,
+                            redis_connection_pool_size: redis_connection_pool_size,
+                            redis_counter_ttl_seconds: redis_counter_ttl_seconds)
+        end
+        @instance
+      end
+
+      def initialize(logger, blocking_limit: nil, logging_limit: nil, redis_connection_pool_size: nil, redis_counter_ttl_seconds: nil)
+        @blocking_limit = blocking_limit
+        @logging_limit = logging_limit
+        @redis_connection_pool_size = redis_connection_pool_size
+        @redis_counter_ttl_seconds = redis_counter_ttl_seconds
+        @logger = logger
+      end
+
+      def try_increment?(user_guid)
+        return true unless @blocking_limit&.>=(0) || @logging_limit&.>=(0)
+
+        key = "#{key_prefix}:#{user_guid}"
+        count = store.increment(key, @logger)
+
+        if @logging_limit&.>=(0) && count > @logging_limit
+          @logger.info("Concurrency limit warning for user '#{user_guid}', count=#{count} exceeded logging_limit=#{@logging_limit}")
+        end
+
+        if @blocking_limit&.>=(0) && count > @blocking_limit
+          store.decrement(key, @logger)
+          @logger.info("Concurrent rate limit exceeded for user '#{user_guid}', limit=#{@blocking_limit} remaining=0")
+          return false
+        end
+
+        true
+      rescue StoreError
+        # fail open
+        true
+      end
+
+      def decrement(user_guid)
+        return unless @blocking_limit&.>=(0) || @logging_limit&.>=(0)
+
+        key = "#{key_prefix}:#{user_guid}"
+        store.decrement(key, @logger)
+      rescue StoreError
+        # fail open
+      end
+
+      def suggested_retry_after
+        rand(1..5).to_i
+      end
+
+      def error_name
+        'ConcurrentRequestLimitExceeded'
+      end
+
+      def error_name_ip_based
+        'IPBasedConcurrentRequestLimitExceeded'
+      end
+
+      private
+
+      def key_prefix
+        'concurrent-rate-limit'
+      end
+
+      def store
+        return @store if defined?(@store)
+
+        redis_socket = VCAP::CloudController::Config.config.get(:redis, :socket)
+        @store = if redis_socket.nil?
+                   ConcurrentInMemoryStore.new
+                 else
+                   ConcurrentRedisStore.new_socket(redis_socket, connection_pool_size: @redis_connection_pool_size, counter_ttl_seconds: @redis_counter_ttl_seconds)
+                 end
+      end
+    end
+
+    class ConcurrencyRateLimiter
+      include CloudFoundry::Middleware::ClientIp
+
+      def initialize(app, opts)
+        @app = app
+        @logger = opts[:logger]
+        @concurrency_limiter = ConcurrencyLimiter.instance(
+          opts[:logger],
+          blocking_limit: opts[:blocking_limit],
+          logging_limit: opts[:logging_limit],
+          redis_connection_pool_size: opts[:redis_connection_pool_size],
+          redis_counter_ttl_seconds: opts[:redis_counter_ttl_seconds]
+        )
+      end
+
+      def call(env)
+        user_guid = nil
+        incremented = false
+
+        if apply_rate_limiting?(env)
+          user_guid = get_user_id(env)
+          incremented = @concurrency_limiter.try_increment?(user_guid)
+          return too_many_requests!(env) unless incremented
+        end
+
+        status, headers, body = @app.call(env)
+        [status, headers, body]
+      ensure
+        @concurrency_limiter.decrement(user_guid) if incremented
+      end
+
+      private
+
+      def get_user_id(env)
+        user_token?(env) ? env['cf.user_guid'] : client_ip(ActionDispatch::Request.new(env))
+      end
+
+      def user_token?(env)
+        !!env['cf.user_guid']
+      end
+
+      def too_many_requests!(env)
+        headers = {}
+        headers['Retry-After'] = @concurrency_limiter.suggested_retry_after.to_s
+        headers['Content-Type'] = 'text/plain; charset=utf-8'
+        message = rate_limit_error(env).to_json
+        headers['Content-Length'] = message.length.to_s
+        [429, headers, [message]]
+      end
+
+      def apply_rate_limiting?(env)
+        request = ActionDispatch::Request.new(env)
+        !basic_auth?(env) && !internal_api?(request) && !root_api?(request)
+      end
+
+      def root_api?(request)
+        request.fullpath.match(%r{\A(?:/v2/info|/v3|/|/healthz)\z})
+      end
+
+      def internal_api?(request)
+        request.fullpath.match(%r{\A/internal})
+      end
+
+      def basic_auth?(env)
+        auth = Rack::Auth::Basic::Request.new(env)
+        auth.provided? && auth.basic?
+      end
+
+      def rate_limit_error(env)
+        error_name = user_token?(env) ? @concurrency_limiter.error_name : @concurrency_limiter.error_name_ip_based
+        api_error = CloudController::Errors::ApiError.new_from_details(error_name)
+        version   = env['PATH_INFO'][0..2]
+        if version == '/v2'
+          ErrorPresenter.new(api_error, Rails.env.test?, V2ErrorHasher.new(api_error)).to_hash
+        elsif version == '/v3'
+          ErrorPresenter.new(api_error, Rails.env.test?, V3ErrorHasher.new(api_error)).to_hash
+        end
+      end
+    end
+  end
+end
diff --git a/middleware/security_context_setter.rb b/middleware/security_context_setter.rb
index bdfe6fd0acc..06a24deec0c 100644
--- a/middleware/security_context_setter.rb
+++ b/middleware/security_context_setter.rb
@@ -29,7 +29,7 @@ def call(env)
           end
         end
 
-        security_context_configurer.configure(header_token)
+        security_context_configurer.configure_token_only(header_token)
 
         if VCAP::CloudController::SecurityContext.valid_token?
           env['cf.user_guid'] = id_from_token
diff --git a/middleware/user_context_setter.rb b/middleware/user_context_setter.rb
new file mode 100644
index 00000000000..63fed0e59a4
--- /dev/null
+++ b/middleware/user_context_setter.rb
@@ -0,0 +1,15 @@
+module CloudFoundry
+  module Middleware
+    class UserContextSetter
+      def initialize(app, security_context_configurer)
+        @app = app
+        @security_context_configurer = security_context_configurer
+      end
+
+      def call(env)
+        @security_context_configurer.configure_user
+        @app.call(env)
+      end
+    end
+  end
+end
diff --git a/spec/unit/lib/cloud_controller/rack_app_builder_spec.rb b/spec/unit/lib/cloud_controller/rack_app_builder_spec.rb
index b5702afd474..bb4047ac768 100644
--- a/spec/unit/lib/cloud_controller/rack_app_builder_spec.rb
+++ b/spec/unit/lib/cloud_controller/rack_app_builder_spec.rb
@@ -221,6 +221,59 @@ module VCAP::CloudController
         end
       end
 
+      describe 'ConcurrencyRateLimiter' do
+        before do
+          allow(CloudFoundry::Middleware::ConcurrencyRateLimiter).to receive(:new)
+        end
+
+        context 'when enabled' do
+          before do
+            builder.build(TestConfig.override(concurrency_rate_limiter: {
+                                                enabled: true,
+                                                blocking_limit: 10,
+                                                logging_limit: 5,
+                                                redis_connection_pool_size: 4,
+                                                redis_counter_ttl_seconds: 600
+                                              }), request_metrics, request_logs).to_app
+          end
+
+          it 'enables the ConcurrencyRateLimiter middleware' do
+            expect(CloudFoundry::Middleware::ConcurrencyRateLimiter).to have_received(:new).with(
+              anything,
+              logger: instance_of(Steno::Logger),
+              blocking_limit: 10,
+              logging_limit: 5,
+              redis_connection_pool_size: 4,
+              redis_counter_ttl_seconds: 600
+            )
+          end
+        end
+
+        context 'when disabled' do
+          before do
+            builder.build(TestConfig.override(concurrency_rate_limiter: { enabled: false }), request_metrics, request_logs).to_app
+          end
+
+          it 'does not enable the ConcurrencyRateLimiter middleware' do
+            expect(CloudFoundry::Middleware::ConcurrencyRateLimiter).not_to have_received(:new)
+          end
+        end
+      end
+
+      describe 'UserContextSetter' do
+        before do
+          allow(CloudFoundry::Middleware::UserContextSetter).to receive(:new)
+        end
+
+        it 'wires UserContextSetter with the security context configurer' do
+          builder.build(TestConfig.config_instance, request_metrics, request_logs).to_app
+          expect(CloudFoundry::Middleware::UserContextSetter).to have_received(:new).with(
+            anything,
+            instance_of(VCAP::CloudController::Security::SecurityContextConfigurer)
+          )
+        end
+      end
+
       describe 'Below Min Cli Warning' do
         before do
           allow(CloudFoundry::Middleware::BelowMinCliWarning).to receive(:new)
diff --git a/spec/unit/lib/cloud_controller/security/security_context_configurer_spec.rb b/spec/unit/lib/cloud_controller/security/security_context_configurer_spec.rb
index 7137438b677..b6b2ca06ea6 100644
--- a/spec/unit/lib/cloud_controller/security/security_context_configurer_spec.rb
+++ b/spec/unit/lib/cloud_controller/security/security_context_configurer_spec.rb
@@ -197,6 +197,81 @@ module Security
           end
         end
       end
+
+      describe '#configure_token_only' do
+        let(:auth_token) { 'auth-token' }
+        let(:token_information) { { 'user_id' => 'user-id-1' } }
+
+        before do
+          allow(token_decoder).to receive(:decode_token).with(auth_token).and_return(token_information)
+        end
+
+        it 'sets the token without looking up the user in the DB' do
+          configurer.configure_token_only(auth_token)
+          expect(SecurityContext.token).to eq(token_information)
+          expect(SecurityContext.auth_token).to eq(auth_token)
+          expect(SecurityContext.current_user).to be_nil
+        end
+
+        it 'clears the security context first' do
+          SecurityContext.set('foo', 'bar', 'baz')
+          configurer.configure_token_only(auth_token)
+          expect(SecurityContext.current_user).to be_nil
+        end
+
+        context 'when the auth_token is invalid' do
+          before do
+            allow(token_decoder).to receive(:decode_token).with(auth_token).and_raise(VCAP::CloudController::UaaTokenDecoder::BadToken)
+          end
+
+          it 'sets invalid token without raising' do
+            expect { configurer.configure_token_only(auth_token) }.not_to raise_error
+            expect(SecurityContext.token).to eq(:invalid_token)
+          end
+        end
+      end
+
+      describe '#configure_user' do
+        let(:token_information) { { 'user_id' => 'user-id-1' } }
+
+        before do
+          SecurityContext.set_token_only(token_information, 'auth-token')
+        end
+
+        context 'when user does not exist' do
+          it 'creates and sets the user' do
+            expect { configurer.configure_user }.to change(User, :count).by(1)
+            expect(SecurityContext.current_user.guid).to eq('user-id-1')
+          end
+        end
+
+        context 'when user already exists' do
+          let!(:user) { create(:user, guid: 'user-id-1') }
+
+          it 'sets the existing user on the security context' do
+            configurer.configure_user
+            expect(SecurityContext.current_user.id).to eq(user.id)
+          end
+        end
+
+        context 'when token is nil' do
+          before { SecurityContext.set_token_only(nil, nil) }
+
+          it 'does not raise and leaves current_user nil' do
+            expect { configurer.configure_user }.not_to raise_error
+            expect(SecurityContext.current_user).to be_nil
+          end
+        end
+
+        context 'when token is invalid_token' do
+          before { SecurityContext.set(nil, :invalid_token, 'auth-token') }
+
+          it 'does not raise and leaves current_user nil' do
+            expect { configurer.configure_user }.not_to raise_error
+            expect(SecurityContext.current_user).to be_nil
+          end
+        end
+      end
     end
   end
 end
diff --git a/spec/unit/middleware/concurrency_rate_limiter_spec.rb b/spec/unit/middleware/concurrency_rate_limiter_spec.rb
new file mode 100644
index 00000000000..998b4f1aeac
--- /dev/null
+++ b/spec/unit/middleware/concurrency_rate_limiter_spec.rb
@@ -0,0 +1,437 @@
+require 'spec_helper'
+
+module CloudFoundry
+  module Middleware
+    RSpec.describe ConcurrencyRateLimiter do
+      let(:app) { double(:app, call: [200, {}, 'a body']) }
+      let(:logger) { double('logger', info: nil, error: nil) }
+      let(:user_guid) { 'user-id-1' }
+      let(:user_env) { { 'cf.user_guid' => user_guid, 'PATH_INFO' => '/v3/apps' } }
+      let(:blocking_limit) { 2 }
+      let(:logging_limit) { nil }
+      let(:concurrency_limiter) do
+        instance_double(ConcurrencyLimiter, try_increment?: true, decrement: nil, suggested_retry_after: 1,
+                                            error_name: 'ConcurrentRequestLimitExceeded',
+                                            error_name_ip_based: 'IPBasedConcurrentRequestLimitExceeded')
+      end
+
+      let(:middleware) do
+        ConcurrencyRateLimiter.new(app, logger: logger, blocking_limit: blocking_limit, logging_limit: logging_limit)
+      end
+
+      before do
+        allow(ConcurrencyLimiter).to receive(:instance).and_return(concurrency_limiter)
+      end
+
+      describe '#call' do
+        context 'when under the limit' do
+          it 'passes the request through' do
+            status, = middleware.call(user_env)
+            expect(status).to eq(200)
+          end
+
+          it 'decrements after the request completes' do
+            middleware.call(user_env)
+            expect(concurrency_limiter).to have_received(:decrement).with(user_guid)
+          end
+        end
+
+        context 'when over the limit' do
+          before do
+            allow(concurrency_limiter).to receive(:try_increment?).and_return(false)
+          end
+
+          it 'returns 429' do
+            status, = middleware.call(user_env)
+            expect(status).to eq(429)
+          end
+
+          it 'does not call the app' do
+            middleware.call(user_env)
+            expect(app).not_to have_received(:call)
+          end
+
+          it 'does not decrement after the blocked request' do
+            middleware.call(user_env)
+            expect(concurrency_limiter).not_to have_received(:decrement)
+          end
+
+          it 'includes Retry-After header as seconds' do
+            _, response_headers, = middleware.call(user_env)
+            expect(response_headers['Retry-After'].to_i).to be > 0
+          end
+
+          context 'when the path is /v3' do
+            it 'formats the error in v3 format' do
+              _, _, body = middleware.call(user_env)
+              json_body = Oj.load(body.first)
+              expect(json_body['errors'].first).to include(
+                'title' => 'CF-ConcurrentRequestLimitExceeded',
+                'code' => 10_021
+              )
+            end
+          end
+
+          context 'when the path is /v2' do
+            let(:user_env) { { 'cf.user_guid' => user_guid, 'PATH_INFO' => '/v2/apps' } }
+
+            it 'formats the error in v2 format' do
+              _, _, body = middleware.call(user_env)
+              json_body = Oj.load(body.first)
+              expect(json_body).to include(
+                'error_code' => 'CF-ConcurrentRequestLimitExceeded',
+                'code' => 10_021
+              )
+            end
+          end
+        end
+
+        context 'when an error is raised in the app' do
+          before do
+            allow(app).to receive(:call).and_raise('an error')
+          end
+
+          it 'still decrements' do
+            expect { middleware.call(user_env) }.to raise_error('an error')
+            expect(concurrency_limiter).to have_received(:decrement).with(user_guid)
+          end
+        end
+
+        context 'when using an unauthenticated request (IP-based)' do
+          let(:ip_env) { { 'PATH_INFO' => '/v3/apps', 'REMOTE_ADDR' => '1.2.3.4', 'HTTP_X_FORWARDED_FOR' => '1.2.3.4' } }
+          let(:fake_request) { instance_double(ActionDispatch::Request, fullpath: '/v3/apps', ip: '1.2.3.4', headers: { 'HTTP_X_FORWARDED_FOR' => '1.2.3.4' }) }
+
+          before do
+            allow(ActionDispatch::Request).to receive(:new).and_return(fake_request)
+          end
+
+          it 'uses IP as the user identifier' do
+            middleware.call(ip_env)
+            expect(concurrency_limiter).to have_received(:try_increment?).with('1.2.3.4')
+          end
+
+          context 'when over the limit' do
+            before do
+              allow(concurrency_limiter).to receive_messages(try_increment?: false, error_name_ip_based: 'IPBasedConcurrentRequestLimitExceeded')
+            end
+
+            it 'uses the IP-based error name' do
+              _, _, body = middleware.call(ip_env)
+              json_body = Oj.load(body.first)
+              expect(json_body['errors'].first['title']).to eq('CF-IPBasedConcurrentRequestLimitExceeded')
+            end
+          end
+        end
+
+        describe 'bypassed requests' do
+          context 'internal API' do
+            let(:internal_env) { { 'cf.user_guid' => user_guid, 'PATH_INFO' => '/internal/v4/asg_latest_update' } }
+            let(:fake_request) { instance_double(ActionDispatch::Request, fullpath: '/internal/v4/asg_latest_update') }
+
+            before { allow(ActionDispatch::Request).to receive(:new).and_return(fake_request) }
+
+            it 'does not rate limit' do
+              middleware.call(internal_env)
+              expect(concurrency_limiter).not_to have_received(:try_increment?)
+            end
+          end
+
+          context 'root API paths' do
+            %w[/v2/info /v3 / /healthz].each do |path|
+              context path do
+                let(:root_env) { { 'cf.user_guid' => user_guid, 'PATH_INFO' => path } }
+                let(:fake_request) { instance_double(ActionDispatch::Request, fullpath: path) }
+
+                before { allow(ActionDispatch::Request).to receive(:new).and_return(fake_request) }
+
+                it 'does not rate limit' do
+                  middleware.call(root_env)
+                  expect(concurrency_limiter).not_to have_received(:try_increment?)
+                end
+              end
+            end
+          end
+
+          context 'basic auth request' do
+            let(:basic_auth_env) do
+              user_env.merge('HTTP_AUTHORIZATION' => 'Basic ' + Base64.encode64('user:pass').strip)
+            end
+
+            it 'does not rate limit' do
+              middleware.call(basic_auth_env)
+              expect(concurrency_limiter).not_to have_received(:try_increment?)
+            end
+          end
+        end
+      end
+    end
+
+    RSpec.describe ConcurrencyLimiter do
+      let(:logger) { double('logger', info: nil, error: nil) }
+      let(:blocking_limit) { 3 }
+      let(:logging_limit) { 2 }
+      let(:user_guid) { 'user-id-1' }
+
+      subject(:limiter) { ConcurrencyLimiter.new(logger, blocking_limit: blocking_limit, logging_limit: logging_limit) }
+
+      before do
+        limiter.instance_variable_set(:@store, ConcurrentInMemoryStore.new)
+      end
+
+      describe '#try_increment?' do
+        it 'returns true when under the blocking limit' do
+          expect(limiter.try_increment?(user_guid)).to be true
+        end
+
+        it 'returns false when over the blocking limit' do
+          blocking_limit.times { limiter.try_increment?(user_guid) }
+          expect(limiter.try_increment?(user_guid)).to be false
+        end
+
+        it 'logs a warning when count exceeds logging_limit' do
+          logging_limit.times { limiter.try_increment?(user_guid) }
+          limiter.try_increment?(user_guid)
+          expect(logger).to have_received(:info).with(/Concurrency limit warning/)
+        end
+
+        it 'does not log warning when under logging_limit' do
+          limiter.try_increment?(user_guid)
+          expect(logger).not_to have_received(:info)
+        end
+
+        it 'logs rate limit exceeded when blocked' do
+          blocking_limit.times { limiter.try_increment?(user_guid) }
+          limiter.try_increment?(user_guid)
+          expect(logger).to have_received(:info).with(/Concurrent rate limit exceeded/)
+        end
+
+        context 'with only logging_limit (no blocking_limit)' do
+          subject(:limiter) { ConcurrencyLimiter.new(logger, logging_limit: logging_limit) }
+
+          it 'always returns true' do
+            10.times { limiter.try_increment?(user_guid) }
+            expect(limiter.try_increment?(user_guid)).to be true
+          end
+
+          it 'logs a warning when count exceeds logging_limit but does not block' do
+            logging_limit.times { limiter.try_increment?(user_guid) }
+            result = limiter.try_increment?(user_guid)
+            expect(result).to be true
+            expect(logger).to have_received(:info).with(/Concurrency limit warning/)
+          end
+        end
+
+        context 'with only blocking_limit (no logging_limit)' do
+          subject(:limiter) { ConcurrencyLimiter.new(logger, blocking_limit: blocking_limit) }
+
+          it 'returns false when over the blocking limit' do
+            blocking_limit.times { limiter.try_increment?(user_guid) }
+            expect(limiter.try_increment?(user_guid)).to be false
+          end
+
+          it 'does not log any warnings' do
+            blocking_limit.times { limiter.try_increment?(user_guid) }
+            limiter.try_increment?(user_guid)
+            expect(logger).not_to have_received(:info).with(/Concurrency limit warning/)
+          end
+        end
+
+        context 'with neither blocking_limit nor logging_limit' do
+          subject(:limiter) { ConcurrencyLimiter.new(logger) }
+
+          it 'always returns true without hitting the store' do
+            store = instance_double(ConcurrentInMemoryStore)
+            allow(store).to receive(:increment)
+            limiter.instance_variable_set(:@store, store)
+            expect(limiter.try_increment?(user_guid)).to be true
+            expect(store).not_to have_received(:increment)
+          end
+        end
+
+        context 'with negative limits (disabled)' do
+          subject(:limiter) { ConcurrencyLimiter.new(logger, blocking_limit: -1, logging_limit: -1) }
+
+          it 'always returns true without hitting the store' do
+            store = instance_double(ConcurrentInMemoryStore)
+            allow(store).to receive(:increment)
+            limiter.instance_variable_set(:@store, store)
+            expect(limiter.try_increment?(user_guid)).to be true
+            expect(store).not_to have_received(:increment)
+          end
+        end
+
+        context 'when store raises StoreError' do
+          before do
+            store = instance_double(ConcurrentInMemoryStore)
+            allow(store).to receive(:increment).and_raise(StoreError)
+            limiter.instance_variable_set(:@store, store)
+          end
+
+          it 'fails open and returns true' do
+            expect(limiter.try_increment?(user_guid)).to be true
+          end
+        end
+      end
+
+      describe '#decrement' do
+        it 'decrements the counter' do
+          blocking_limit.times { limiter.try_increment?(user_guid) }
+          limiter.decrement(user_guid)
+          expect(limiter.try_increment?(user_guid)).to be true
+        end
+
+        context 'with neither blocking_limit nor logging_limit' do
+          subject(:limiter) { ConcurrencyLimiter.new(logger) }
+
+          it 'does not hit the store' do
+            store = instance_double(ConcurrentInMemoryStore)
+            allow(store).to receive(:decrement)
+            limiter.instance_variable_set(:@store, store)
+            limiter.decrement(user_guid)
+            expect(store).not_to have_received(:decrement)
+          end
+        end
+
+        context 'with negative limits (disabled)' do
+          subject(:limiter) { ConcurrencyLimiter.new(logger, blocking_limit: -1, logging_limit: -1) }
+
+          it 'does not hit the store' do
+            store = instance_double(ConcurrentInMemoryStore)
+            allow(store).to receive(:decrement)
+            limiter.instance_variable_set(:@store, store)
+            limiter.decrement(user_guid)
+            expect(store).not_to have_received(:decrement)
+          end
+        end
+
+        context 'when store raises StoreError' do
+          before do
+            store = instance_double(ConcurrentInMemoryStore)
+            allow(store).to receive(:decrement).and_raise(StoreError)
+            limiter.instance_variable_set(:@store, store)
+          end
+
+          it 'does not raise' do
+            expect { limiter.decrement(user_guid) }.not_to raise_error
+          end
+        end
+      end
+
+      describe '#suggested_retry_after' do
+        it 'returns a positive integer' do
+          expect(limiter.suggested_retry_after).to be >= 1
+        end
+
+        it 'returns a value between 1 and 5' do
+          100.times { expect(limiter.suggested_retry_after).to be_between(1, 5) }
+        end
+      end
+    end
+
+    RSpec.describe ConcurrentInMemoryStore do
+      let(:store) { ConcurrentInMemoryStore.new }
+      let(:logger) { double('logger') }
+      let(:key) { 'test-key' }
+
+      describe '#increment' do
+        it 'returns 1 for a new key' do
+          expect(store.increment(key, logger)).to eq(1)
+        end
+
+        it 'increments on each call' do
+          store.increment(key, logger)
+          expect(store.increment(key, logger)).to eq(2)
+        end
+      end
+
+      describe '#decrement' do
+        it 'returns 0 for a non-existent key' do
+          expect(store.decrement(key, logger)).to eq(0)
+        end
+
+        it 'decrements the counter' do
+          store.increment(key, logger)
+          store.increment(key, logger)
+          expect(store.decrement(key, logger)).to eq(1)
+        end
+
+        it 'removes the key when count reaches 0' do
+          store.increment(key, logger)
+          store.decrement(key, logger)
+          expect(store.instance_variable_get(:@data)).not_to have_key(key)
+        end
+
+        it 'does not go below 0' do
+          store.decrement(key, logger)
+          expect(store.decrement(key, logger)).to eq(0)
+        end
+      end
+    end
+
+    RSpec.describe ConcurrentRedisStore do
+      let(:logger) { double('logger', error: nil) }
+      let(:key) { 'test-key' }
+      let(:store) { ConcurrentRedisStore.new(MockRedis.new) }
+
+      describe '#increment' do
+        it 'returns 1 for a new key' do
+          expect(store.increment(key, logger)).to eq(1)
+        end
+
+        it 'increments on each call' do
+          store.increment(key, logger)
+          expect(store.increment(key, logger)).to eq(2)
+        end
+
+        context 'with TTL configured' do
+          let(:store) { ConcurrentRedisStore.new(MockRedis.new, counter_ttl_seconds: 60) }
+
+          it 'sets TTL on every increment' do
+            redis = store.instance_variable_get(:@redis)
+            allow(redis).to receive(:expire).and_call_original
+            store.increment(key, logger)
+            store.increment(key, logger)
+            expect(redis).to have_received(:expire).with(key, 60).twice
+          end
+        end
+
+        context 'when Redis raises an error' do
+          before { allow(store.instance_variable_get(:@redis)).to receive(:incr).and_raise(Redis::ConnectionError) }
+
+          it 'logs the error and raises StoreError' do
+            expect { store.increment(key, logger) }.to raise_error(StoreError)
+            expect(logger).to have_received(:error).with(/Redis error/)
+          end
+        end
+      end
+
+      describe '#decrement' do
+        it 'decrements the counter' do
+          store.increment(key, logger)
+          store.increment(key, logger)
+          expect(store.decrement(key, logger)).to eq(1)
+        end
+
+        it 'does not go below 0 when key does not exist' do
+          store.decrement(key, logger)
+          expect(store.increment(key, logger)).to eq(1)
+        end
+
+        it 'returns 0 when key expired mid-flight' do
+          store.increment(key, logger)
+          store.instance_variable_get(:@redis).del(key) # simulate TTL expiry
+          expect(store.decrement(key, logger)).to eq(0)
+        end
+
+        context 'when Redis raises an error' do
+          before { allow(store.instance_variable_get(:@redis)).to receive(:decr).and_raise(Redis::ConnectionError) }
+
+          it 'logs the error and raises StoreError' do
+            expect { store.decrement(key, logger) }.to raise_error(StoreError)
+            expect(logger).to have_received(:error).with(/Redis error/)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/spec/unit/middleware/user_context_setter_spec.rb b/spec/unit/middleware/user_context_setter_spec.rb
new file mode 100644
index 00000000000..42421fa89e0
--- /dev/null
+++ b/spec/unit/middleware/user_context_setter_spec.rb
@@ -0,0 +1,38 @@
+require 'spec_helper'
+require 'user_context_setter'
+
+module CloudFoundry
+  module Middleware
+    RSpec.describe UserContextSetter do
+      let(:app) { double(:app, call: [200, {}, 'a body']) }
+      let(:security_context_configurer) { instance_double(VCAP::CloudController::Security::SecurityContextConfigurer, configure_user: nil) }
+      let(:middleware) { UserContextSetter.new(app, security_context_configurer) }
+      let(:env) { { 'cf.user_guid' => 'user-id-1', 'PATH_INFO' => '/v3/apps' } }
+
+      describe '#call' do
+        it 'calls configure_user on the security context configurer' do
+          middleware.call(env)
+          expect(security_context_configurer).to have_received(:configure_user)
+        end
+
+        it 'passes the request to the app' do
+          middleware.call(env)
+          expect(app).to have_received(:call).with(env)
+        end
+
+        it 'returns the app response' do
+          status, headers, body = middleware.call(env)
+          expect(status).to eq(200)
+          expect(headers).to eq({})
+          expect(body).to eq('a body')
+        end
+
+        it 'calls configure_user before the app' do
+          expect(security_context_configurer).to receive(:configure_user).ordered
+          expect(app).to receive(:call).ordered.and_return([200, {}, 'a body'])
+          middleware.call(env)
+        end
+      end
+    end
+  end
+end