diff --git a/config/cloud_controller.yml b/config/cloud_controller.yml
index 45468fd1e81..1c07f6f74f3 100644
--- a/config/cloud_controller.yml
+++ b/config/cloud_controller.yml
@@ -275,6 +275,11 @@ rate_limiter_v2_api:
   global_admin_limit: 20000
   reset_interval_in_minutes: 60
 
+concurrency_rate_limiter:
+  enabled: false
+  blocking_limit: 10
+  logging_limit: 10
+
 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.rb b/lib/cloud_controller/config.rb
index 78f0d6679cb..3620dd814ab 100644
--- a/lib/cloud_controller/config.rb
+++ b/lib/cloud_controller/config.rb
@@ -64,9 +64,18 @@ def merge_defaults(orig_config)
         config = orig_config.dup
         config[:db] ||= {}
         ensure_config_has_database_parts(config)
+        apply_redis_counter_defaults(config)
         sanitize(config)
       end
 
+      def apply_redis_counter_defaults(config)
+        return unless config.dig(:concurrency_rate_limiter, :enabled) ||
+                      (config[:max_concurrent_service_broker_requests] || 0) > 0
+
+        config[:redis_connection_pool_size] = config.dig(:puma, :max_threads) || 1
+        config[:redis_counter_ttl_seconds] = config[:request_timeout_in_seconds] + 1
+      end
+
       def ensure_config_has_database_parts(config)
         abort_no_db_connection! if ENV['DB_CONNECTION_STRING'].nil? && config[:db][:database].nil?
         config[:db][:db_connection_string] ||= ENV.fetch('DB_CONNECTION_STRING', nil)
diff --git a/lib/cloud_controller/config_schemas/api_schema.rb b/lib/cloud_controller/config_schemas/api_schema.rb
index a4b041e3684..aa149d8ed96 100644
--- a/lib/cloud_controller/config_schemas/api_schema.rb
+++ b/lib/cloud_controller/config_schemas/api_schema.rb
@@ -384,6 +384,8 @@ class ApiSchema < VCAP::Config
             reset_interval_in_minutes: Integer
           },
           max_concurrent_service_broker_requests: Integer,
+          optional(:redis_connection_pool_size) => Integer,
+          optional(:redis_counter_ttl_seconds) => Integer,
           shared_isolation_segment_name: String,
 
           optional(:rate_limiter_v2_api) => {
@@ -395,6 +397,12 @@ class ApiSchema < VCAP::Config
             reset_interval_in_minutes: Integer
           },
 
+          optional(:concurrency_rate_limiter) => {
+            enabled: bool,
+            optional(:blocking_limit) => Integer,
+            optional(:logging_limit) => 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..67b32abe60b 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(:redis_connection_pool_size),
+            redis_counter_ttl_seconds: config.get(:redis_counter_ttl_seconds)
+          }
+        end
+
         if config.get(:rate_limiter, :enabled)
           use CloudFoundry::Middleware::RateLimiter, {
             logger: Steno.logger('cc.rate_limiter'),
@@ -45,7 +57,9 @@ def build(config, request_metrics, request_logs)
           use CloudFoundry::Middleware::ServiceBrokerRateLimiter, {
             logger: Steno.logger('cc.service_broker_rate_limiter'),
             max_concurrent_requests: config.get(:max_concurrent_service_broker_requests),
-            broker_timeout_seconds: config.get(:broker_client_timeout_seconds)
+            broker_timeout_seconds: config.get(:broker_client_timeout_seconds),
+            redis_connection_pool_size: config.get(:redis_connection_pool_size),
+            redis_counter_ttl_seconds: config.get(:redis_counter_ttl_seconds)
           }
         end
         if config.get(:rate_limiter_v2_api, :enabled)
@@ -59,6 +73,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 +94,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..0826fc194ad 100644
--- a/lib/cloud_controller/security/security_context_configurer.rb
+++ b/lib/cloud_controller/security/security_context_configurer.rb
@@ -6,14 +6,29 @@ def initialize(token_decoder)
       end
 
       def configure(header_token)
+        configure_token_only(header_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(nil, decoded_token, header_token)
+      rescue VCAP::CloudController::UaaTokenDecoder::BadToken
+        VCAP::CloudController::SecurityContext.set(nil, :invalid_token, header_token)
+      end
 
+      def configure_user
+        return unless VCAP::CloudController::SecurityContext.valid_token?
+
+        decoded_token = VCAP::CloudController::SecurityContext.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/middleware/base_rate_limiter.rb b/middleware/base_rate_limiter.rb
index ca00db858be..9c77e9bf47a 100644
--- a/middleware/base_rate_limiter.rb
+++ b/middleware/base_rate_limiter.rb
@@ -1,4 +1,6 @@
-require 'mixins/client_ip'
+require 'mixins/user_id'
+require 'mixins/too_many_requests'
+require 'mixins/basic_auth'
 require 'mixins/user_reset_interval'
 require 'redis'
 
@@ -89,7 +91,9 @@ def to_hash
     end
 
     class BaseRateLimiter
-      include CloudFoundry::Middleware::ClientIp
+      include CloudFoundry::Middleware::UserId
+      include CloudFoundry::Middleware::TooManyRequests
+      include CloudFoundry::Middleware::BasicAuth
 
       def initialize(app, logger, expiring_request_counter, reset_interval, header_suffix=nil)
         @app = app
@@ -111,17 +115,13 @@ def call(env)
           rate_limit_headers.reset = (Time.now.to_i + expires_in).to_s
           rate_limit_headers.remaining = estimate_remaining(env, count)
 
-          return too_many_requests!(expires_in, env, rate_limit_headers) if exceeded_rate_limit(count, env)
+          return too_many_requests!(env, rate_limit_error_name(env), retry_after: expires_in, extra_headers: rate_limit_headers.to_hash) if exceeded_rate_limit(count, env)
         end
 
         status, headers, body = @app.call(env)
         [status, headers.merge(rate_limit_headers.to_hash), body]
       end
 
-      def get_user_id(env)
-        user_token?(env) ? env['cf.user_guid'] : client_ip(ActionDispatch::Request.new(env))
-      end
-
       private
 
       def apply_rate_limiting?(_env)
@@ -132,7 +132,7 @@ def global_request_limit(_env)
         raise 'method should be implemented in concrete class'
       end
 
-      def rate_limit_error(_env)
+      def rate_limit_error_name(_env)
         raise 'method should be implemented in concrete class'
       end
 
@@ -153,27 +153,9 @@ def estimate_remaining(env, new_count)
         [0, estimate].max.to_i.to_s
       end
 
-      def user_token?(env)
-        !!env['cf.user_guid']
-      end
-
-      def basic_auth?(env)
-        auth = Rack::Auth::Basic::Request.new(env)
-        auth.provided? && auth.basic?
-      end
-
       def admin?
         VCAP::CloudController::SecurityContext.admin? || VCAP::CloudController::SecurityContext.admin_read_only?
       end
-
-      def too_many_requests!(expires_in, env, rate_limit_headers)
-        headers = {}
-        headers['Retry-After'] = expires_in.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, rate_limit_headers.to_hash.merge(headers), [message]]
-      end
     end
   end
 end
diff --git a/middleware/concurrency_rate_limiter.rb b/middleware/concurrency_rate_limiter.rb
new file mode 100644
index 00000000000..ae2a5fd2e8f
--- /dev/null
+++ b/middleware/concurrency_rate_limiter.rb
@@ -0,0 +1,58 @@
+require 'mixins/user_id'
+require 'mixins/internal_or_root_api'
+require 'mixins/too_many_requests'
+require 'mixins/basic_auth'
+require 'concurrent_request_counter'
+
+module CloudFoundry
+  module Middleware
+    class ConcurrencyRateLimiter
+      include CloudFoundry::Middleware::UserId
+      include CloudFoundry::Middleware::InternalOrRootApi
+      include CloudFoundry::Middleware::TooManyRequests
+      include CloudFoundry::Middleware::BasicAuth
+
+      def initialize(app, opts)
+        @app = app
+        @logger = opts[:logger]
+        @counter = ConcurrentRequestCounter.instance(
+          'concurrent-rate-limit',
+          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
+        decrement_after_call = false
+
+        if apply_rate_limiting?(env)
+          user_guid = get_user_id(env)
+          decrement_after_call = @counter.try_increment?(user_guid, @logger)
+          return too_many_requests!(env, rate_limit_error_name(env), retry_after: suggested_retry_after) unless decrement_after_call
+        end
+
+        @app.call(env)
+      ensure
+        @counter.decrement(user_guid, @logger) if decrement_after_call
+      end
+
+      private
+
+      def suggested_retry_after
+        rand(1..5).to_i
+      end
+
+      def apply_rate_limiting?(env)
+        request = ActionDispatch::Request.new(env)
+        !basic_auth?(env) && !internal_api?(request) && !root_api?(request)
+      end
+
+      def rate_limit_error_name(env)
+        user_token?(env) ? 'ConcurrentRequestLimitExceeded' : 'IPBasedConcurrentRequestLimitExceeded'
+      end
+    end
+  end
+end
diff --git a/middleware/concurrent_request_counter.rb b/middleware/concurrent_request_counter.rb
new file mode 100644
index 00000000000..3c198f6f8fe
--- /dev/null
+++ b/middleware/concurrent_request_counter.rb
@@ -0,0 +1,199 @@
+require 'digest'
+
+module CloudFoundry
+  module Middleware
+    class StoreError < StandardError; end
+
+    class ConcurrentRequestCounter
+      @instances = {}
+      @mutex = Mutex.new
+
+      def self.instance(key_prefix, **opts)
+        return @instances[key_prefix] if @instances[key_prefix]
+
+        @mutex.synchronize do
+          @instances[key_prefix] ||= new(key_prefix, **opts)
+        end
+        @instances[key_prefix]
+      end
+
+      def initialize(key_prefix, blocking_limit: nil, logging_limit: nil, redis_connection_pool_size: nil, redis_counter_ttl_seconds: nil)
+        @key_prefix = key_prefix
+        @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
+
+      def try_increment?(user_guid, logger)
+        return true unless blocking_active? || logging_active?
+
+        key = "#{@key_prefix}:#{user_guid}"
+
+        if blocking_active?
+          count = store.try_increment(key, @blocking_limit, logger)
+          return false if count.nil?
+        else
+          count = store.try_log_increment(key, logger)
+          logger.info("Concurrency limit warning for '#{user_guid}', count=#{count} exceeded logging_limit=#{@logging_limit}") if count > @logging_limit
+        end
+
+        true
+      rescue StoreError
+        # fail open
+        true
+      end
+
+      def decrement(user_guid, logger)
+        return unless blocking_active? || logging_active?
+
+        key = "#{@key_prefix}:#{user_guid}"
+        store.decrement(key, logger)
+      rescue StoreError
+        # fail open
+      end
+
+      private
+
+      def blocking_active?
+        @blocking_limit&.>=(0)
+      end
+
+      def logging_active?
+        @logging_limit&.>=(0)
+      end
+
+      def store
+        return @store if defined?(@store)
+
+        redis_socket = VCAP::CloudController::Config.config.get(:redis, :socket)
+        @store = if redis_socket.nil?
+                   InMemoryStore.new
+                 else
+                   RedisStore.new(redis_socket, @redis_connection_pool_size, @redis_counter_ttl_seconds)
+                 end
+      end
+
+      class RedisStore
+        # Atomic check-then-increment: admitted when under limit, -1 when rejected (not incremented).
+        # KEYS[1] = counter key
+        # ARGV[1] = limit (>= 0): 0 rejects all, N caps at N
+        # ARGV[2] = ttl in seconds (<= 0 = no expiry)
+        INCREMENT_SCRIPT = <<~LUA.freeze
+          local limit = tonumber(ARGV[1])
+          local ttl = tonumber(ARGV[2])
+          local current = tonumber(redis.call('get', KEYS[1]) or '0')
+          if current >= limit then
+            return -1
+          end
+          local count = redis.call('incr', KEYS[1])
+          if count == 1 and ttl > 0 then
+            redis.call('expire', KEYS[1], ttl)
+          end
+          return count
+        LUA
+        INCREMENT_SHA = Digest::SHA1.hexdigest(INCREMENT_SCRIPT).freeze
+
+        # Always increments with no cap. Used in logging-only mode to observe real concurrency.
+        # KEYS[1] = counter key
+        # ARGV[1] = ttl in seconds (<= 0 = no expiry)
+        LOG_INCREMENT_SCRIPT = <<~LUA.freeze
+          local ttl = tonumber(ARGV[1])
+          local count = redis.call('incr', KEYS[1])
+          if count == 1 and ttl > 0 then
+            redis.call('expire', KEYS[1], ttl)
+          end
+          return count
+        LUA
+        LOG_INCREMENT_SHA = Digest::SHA1.hexdigest(LOG_INCREMENT_SCRIPT).freeze
+
+        def initialize(socket, connection_pool_size, counter_ttl_seconds)
+          @redis = ConnectionPool::Wrapper.new(size: connection_pool_size) do
+            Redis.new(timeout: 1, path: socket)
+          end
+          @counter_ttl_seconds = counter_ttl_seconds
+        end
+
+        # Returns post-increment count when incremented, nil when limit is reached.
+        def try_increment(key, limit, logger)
+          count = eval_increment(key, limit).to_i
+          count.negative? ? nil : count
+        rescue Redis::BaseError => e
+          logger.error("Redis error: #{e.class} - #{e.message}")
+          raise StoreError.new("increment failed: #{e.message}")
+        end
+
+        # Always increments and returns count. Used in logging-only mode.
+        def try_log_increment(key, logger)
+          eval_log_increment(key).to_i
+        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
+
+        private
+
+        def eval_increment(key, limit)
+          argv = [limit, @counter_ttl_seconds || 0]
+          @redis.evalsha(INCREMENT_SHA, keys: [key], argv: argv)
+        rescue Redis::CommandError => e
+          raise unless e.message.include?('NOSCRIPT')
+
+          @redis.eval(INCREMENT_SCRIPT, keys: [key], argv: argv)
+        end
+
+        def eval_log_increment(key)
+          argv = [@counter_ttl_seconds || 0]
+          @redis.evalsha(LOG_INCREMENT_SHA, keys: [key], argv: argv)
+        rescue Redis::CommandError => e
+          raise unless e.message.include?('NOSCRIPT')
+
+          @redis.eval(LOG_INCREMENT_SCRIPT, keys: [key], argv: argv)
+        end
+      end
+
+      class InMemoryStore
+        def initialize
+          @mutex = Mutex.new
+          @data = {}
+        end
+
+        # Atomic check-and-increment. Returns post-increment count when admitted, nil when at limit.
+        def try_increment(key, limit, _logger)
+          @mutex.synchronize do
+            current = @data[key] || 0
+            return nil if current >= limit
+
+            @data[key] = current + 1
+          end
+        end
+
+        # Always increments and returns count. Never rejects. Used in logging-only mode.
+        def try_log_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
+    end
+  end
+end
diff --git a/middleware/mixins/basic_auth.rb b/middleware/mixins/basic_auth.rb
new file mode 100644
index 00000000000..aa18e210abb
--- /dev/null
+++ b/middleware/mixins/basic_auth.rb
@@ -0,0 +1,12 @@
+module CloudFoundry
+  module Middleware
+    module BasicAuth
+      private
+
+      def basic_auth?(env)
+        auth = Rack::Auth::Basic::Request.new(env)
+        auth.provided? && auth.basic?
+      end
+    end
+  end
+end
diff --git a/middleware/mixins/internal_or_root_api.rb b/middleware/mixins/internal_or_root_api.rb
new file mode 100644
index 00000000000..eeeecb7e41a
--- /dev/null
+++ b/middleware/mixins/internal_or_root_api.rb
@@ -0,0 +1,13 @@
+module CloudFoundry
+  module Middleware
+    module InternalOrRootApi
+      def internal_api?(request)
+        request.fullpath.match(%r{\A/internal})
+      end
+
+      def root_api?(request)
+        request.fullpath.match(%r{\A(?:/v2/info|/v3|/|/healthz)\z})
+      end
+    end
+  end
+end
diff --git a/middleware/mixins/too_many_requests.rb b/middleware/mixins/too_many_requests.rb
new file mode 100644
index 00000000000..2591b67f99d
--- /dev/null
+++ b/middleware/mixins/too_many_requests.rb
@@ -0,0 +1,26 @@
+module CloudFoundry
+  module Middleware
+    module TooManyRequests
+      def too_many_requests!(env, error_name, retry_after:, extra_headers: {})
+        headers = {}
+        headers['Retry-After'] = retry_after.to_s
+        headers['Content-Type'] = 'text/plain; charset=utf-8'
+        message = rate_limit_error(env, error_name).to_json
+        headers['Content-Length'] = message.bytesize.to_s
+        [429, extra_headers.to_hash.merge(headers), [message]]
+      end
+
+      private
+
+      def rate_limit_error(env, error_name)
+        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/mixins/user_id.rb b/middleware/mixins/user_id.rb
new file mode 100644
index 00000000000..6544758871f
--- /dev/null
+++ b/middleware/mixins/user_id.rb
@@ -0,0 +1,19 @@
+require 'mixins/client_ip'
+
+module CloudFoundry
+  module Middleware
+    module UserId
+      include CloudFoundry::Middleware::ClientIp
+
+      def get_user_id(env)
+        user_token?(env) ? env['cf.user_guid'] : client_ip(ActionDispatch::Request.new(env))
+      end
+
+      private
+
+      def user_token?(env)
+        !!env['cf.user_guid']
+      end
+    end
+  end
+end
diff --git a/middleware/rate_limiter.rb b/middleware/rate_limiter.rb
index 1aa088afbe3..6b57f5eaa57 100644
--- a/middleware/rate_limiter.rb
+++ b/middleware/rate_limiter.rb
@@ -1,8 +1,11 @@
 require 'base_rate_limiter'
+require 'mixins/internal_or_root_api'
 
 module CloudFoundry
   module Middleware
     class RateLimiter < BaseRateLimiter
+      include CloudFoundry::Middleware::InternalOrRootApi
+
       EXPIRING_REQUEST_COUNTER = ExpiringRequestCounter.new('rate-limit')
 
       def initialize(app, opts)
@@ -26,14 +29,6 @@ def apply_rate_limiting?(env)
         true
       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 global_request_limit(env)
         return @global_admin_limit if admin?
 
@@ -46,15 +41,8 @@ def per_process_request_limit(env)
         user_token?(env) ? @per_process_general_limit : @per_process_unauthenticated_limit
       end
 
-      def rate_limit_error(env)
-        error_name = user_token?(env) ? 'RateLimitExceeded' : 'IPBasedRateLimitExceeded'
-        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
+      def rate_limit_error_name(env)
+        user_token?(env) ? 'RateLimitExceeded' : 'IPBasedRateLimitExceeded'
       end
     end
   end
diff --git a/middleware/rate_limiter_v2_api.rb b/middleware/rate_limiter_v2_api.rb
index 285f1e39b72..231398ee6f6 100644
--- a/middleware/rate_limiter_v2_api.rb
+++ b/middleware/rate_limiter_v2_api.rb
@@ -36,9 +36,8 @@ def v2_rate_limit_exempted?
         VCAP::CloudController::SecurityContext.v2_rate_limit_exempted?
       end
 
-      def rate_limit_error(_env)
-        api_error = CloudController::Errors::ApiError.new_from_details('RateLimitV2APIExceeded')
-        ErrorPresenter.new(api_error, Rails.env.test?, V2ErrorHasher.new(api_error)).to_hash
+      def rate_limit_error_name(_env)
+        'RateLimitV2APIExceeded'
       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/service_broker_rate_limiter.rb b/middleware/service_broker_rate_limiter.rb
index 1a828692354..a30a64abfcd 100644
--- a/middleware/service_broker_rate_limiter.rb
+++ b/middleware/service_broker_rate_limiter.rb
@@ -1,5 +1,5 @@
-require 'concurrent-ruby'
-require 'redis'
+require 'mixins/too_many_requests'
+require 'concurrent_request_counter'
 
 module CloudFoundry
   module Middleware
@@ -10,103 +10,36 @@ module Middleware
       RateLimitEndpoint.new(%r{\A/v3/(service_instances|service_credential_bindings|service_route_bindings)/.+/parameters\z}, %w[GET])
     ].freeze
 
-    class ConcurrentRequestCounter
-      def initialize(key_prefix, redis_connection_pool_size: nil)
-        @key_prefix = key_prefix
-        @redis_connection_pool_size = redis_connection_pool_size
-      end
-
-      def try_increment?(user_guid, max_concurrent_requests, logger)
-        key = "#{@key_prefix}:#{user_guid}"
-        store.try_increment?(key, max_concurrent_requests, logger)
-      end
-
-      def decrement(user_guid, logger)
-        key = "#{@key_prefix}:#{user_guid}"
-        store.decrement(key, logger)
-      end
-
-      private
-
-      def store
-        return @store if defined?(@store)
-
-        redis_socket = VCAP::CloudController::Config.config.get(:redis, :socket)
-        @store = redis_socket.nil? ? InMemoryStore.new : RedisStore.new(redis_socket, @redis_connection_pool_size)
-      end
-
-      class InMemoryStore
-        def initialize
-          @mutex = Mutex.new
-          @data = {}
-        end
-
-        def try_increment?(key, max_concurrent_requests, _)
-          @mutex.synchronize do
-            @data[key] = Concurrent::Semaphore.new(max_concurrent_requests) unless @data.key?(key)
-            @data[key].try_acquire
-          end
-        end
-
-        def decrement(key, _)
-          @mutex.synchronize do
-            @data[key].release if @data.key?(key)
-          end
-        end
-      end
-
-      class RedisStore
-        def initialize(socket, connection_pool_size)
-          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
-        end
-
-        def try_increment?(key, max_concurrent_requests, logger)
-          count_str = @redis.incr(key)
-          return true if count_str.to_i <= max_concurrent_requests
-
-          @redis.decr(key)
-          false
-        rescue Redis::BaseError => e
-          logger.error("Redis error: #{e.class} - #{e.message}")
-          true
-        end
-
-        def decrement(key, logger)
-          count_str = @redis.decr(key)
-          @redis.incr(key) if count_str.to_i < 0
-        rescue Redis::BaseError => e
-          logger.error("Redis error: #{e.class} - #{e.message}")
-        end
-      end
-    end
-
     class ServiceBrokerRateLimiter
-      CONCURRENT_REQUEST_COUNTER = ConcurrentRequestCounter.new('service-broker-rate-limit')
+      include CloudFoundry::Middleware::TooManyRequests
 
       def initialize(app, opts)
         @app = app
         @logger = opts[:logger]
-        @max_concurrent_requests = opts[:max_concurrent_requests]
         @broker_timeout_seconds = opts[:broker_timeout_seconds]
-        @concurrent_request_counter = CONCURRENT_REQUEST_COUNTER
+        @counter = ConcurrentRequestCounter.instance(
+          'service-broker-rate-limit',
+          blocking_limit: opts[:max_concurrent_requests],
+          redis_connection_pool_size: opts[:redis_connection_pool_size],
+          redis_counter_ttl_seconds: opts[:redis_counter_ttl_seconds]
+        )
       end
 
       def call(env)
         decrement_after_call = false
         user_guid = env['cf.user_guid']
-        if apply_rate_limiting?(env)
-          return too_many_requests!(env, user_guid) unless @concurrent_request_counter.try_increment?(user_guid, @max_concurrent_requests, @logger)
-
-          decrement_after_call = true
 
+        if apply_rate_limiting?(env)
+          decrement_after_call = @counter.try_increment?(user_guid, @logger)
+          unless decrement_after_call
+            @logger.info("Service broker concurrent rate limit exceeded for user '#{user_guid}'")
+            return too_many_requests!(env, 'ServiceBrokerRateLimitExceeded', retry_after: suggested_retry_after)
+          end
         end
 
         @app.call(env)
       ensure
-        @concurrent_request_counter.decrement(user_guid, @logger) if decrement_after_call
+        @counter.decrement(user_guid, @logger) if decrement_after_call
       end
 
       private
@@ -130,26 +63,6 @@ def suggested_retry_after
         delay_range = (@broker_timeout_seconds * 0.5).floor..(@broker_timeout_seconds * 1.5).ceil
         rand(delay_range).to_i
       end
-
-      def rate_limit_error(env)
-        api_error = CloudController::Errors::ApiError.new_from_details('ServiceBrokerRateLimitExceeded')
-        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
-
-      def too_many_requests!(env, user_guid)
-        @logger.info("Service broker concurrent rate limit exceeded for user '#{user_guid}'")
-        headers = {}
-        headers['Retry-After'] = 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
     end
   end
 end
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/config_spec.rb b/spec/unit/lib/cloud_controller/config_spec.rb
index dd91faa8c64..3aafd8f7dbf 100644
--- a/spec/unit/lib/cloud_controller/config_spec.rb
+++ b/spec/unit/lib/cloud_controller/config_spec.rb
@@ -103,6 +103,54 @@ module VCAP::CloudController
           expect(config_hash[:local_route]).to eq('127.0.0.1')
         end
       end
+
+      describe 'redis counter defaults' do
+        let(:base_hash) { YAMLConfig.safe_load_file('config/cloud_controller.yml') }
+
+        context 'when concurrency_rate_limiter is enabled' do
+          let(:config_hash) do
+            base_hash['concurrency_rate_limiter'] = { 'enabled' => true, 'blocking_limit' => 10 }
+            base_hash['puma'] = { 'automatic_worker_count' => false, 'workers' => 2, 'max_threads' => 5 }
+            Config.load_from_hash(base_hash, context: :api).config_hash
+          end
+
+          it 'derives redis_connection_pool_size from puma max_threads' do
+            expect(config_hash[:redis_connection_pool_size]).to eq(5)
+          end
+
+          it 'derives redis_counter_ttl_seconds from request_timeout_in_seconds + 1' do
+            expect(config_hash[:redis_counter_ttl_seconds]).to eq(config_hash[:request_timeout_in_seconds] + 1)
+          end
+        end
+
+        context 'when max_concurrent_service_broker_requests is set' do
+          let(:config_hash) do
+            base_hash['max_concurrent_service_broker_requests'] = 3
+            base_hash['puma'] = { 'automatic_worker_count' => false, 'workers' => 2, 'max_threads' => 4 }
+            Config.load_from_hash(base_hash, context: :api).config_hash
+          end
+
+          it 'derives redis_connection_pool_size from puma max_threads' do
+            expect(config_hash[:redis_connection_pool_size]).to eq(4)
+          end
+        end
+
+        context 'when neither rate limiter is active' do
+          let(:config_hash) do
+            base_hash['concurrency_rate_limiter'] = { 'enabled' => false }
+            base_hash['max_concurrent_service_broker_requests'] = 0
+            Config.load_from_hash(base_hash, context: :api).config_hash
+          end
+
+          it 'does not set redis_connection_pool_size' do
+            expect(config_hash[:redis_connection_pool_size]).to be_nil
+          end
+
+          it 'does not set redis_counter_ttl_seconds' do
+            expect(config_hash[:redis_counter_ttl_seconds]).to be_nil
+          end
+        end
+      end
     end
 
     describe '.load_from_file' do
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..36c472e551e 100644
--- a/spec/unit/lib/cloud_controller/rack_app_builder_spec.rb
+++ b/spec/unit/lib/cloud_controller/rack_app_builder_spec.rb
@@ -110,7 +110,9 @@ module VCAP::CloudController
               anything,
               logger: instance_of(Steno::Logger),
               max_concurrent_requests: TestConfig.config_instance.get(:max_concurrent_service_broker_requests),
-              broker_timeout_seconds: TestConfig.config_instance.get(:broker_client_timeout_seconds)
+              broker_timeout_seconds: TestConfig.config_instance.get(:broker_client_timeout_seconds),
+              redis_connection_pool_size: TestConfig.config_instance.get(:redis_connection_pool_size),
+              redis_counter_ttl_seconds: TestConfig.config_instance.get(:redis_counter_ttl_seconds)
             )
           end
         end
@@ -221,6 +223,57 @@ 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
+                                              }), 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: TestConfig.config_instance.get(:redis_connection_pool_size),
+              redis_counter_ttl_seconds: TestConfig.config_instance.get(:redis_counter_ttl_seconds)
+            )
+          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..e472d7432f4 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,79 @@ 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
+          expect { configurer.configure_token_only(auth_token) }.to have_queried_db_times(/select .* from .users./i, 0)
+          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
+        before do
+          SecurityContext.set(nil, { 'user_id' => 'user-id-1' }, '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(nil, 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..99ff4168aa4
--- /dev/null
+++ b/spec/unit/middleware/concurrency_rate_limiter_spec.rb
@@ -0,0 +1,170 @@
+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(:counter) do
+        instance_double(ConcurrentRequestCounter, try_increment?: true, decrement: nil)
+      end
+
+      let(:middleware) do
+        ConcurrencyRateLimiter.new(app, logger: logger, blocking_limit: blocking_limit, logging_limit: logging_limit)
+      end
+
+      before do
+        allow(ConcurrentRequestCounter).to receive(:instance).and_return(counter)
+      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(counter).to have_received(:decrement).with(user_guid, logger)
+          end
+        end
+
+        context 'when over the limit' do
+          before do
+            allow(counter).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(counter).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(counter).to have_received(:decrement).with(user_guid, logger)
+          end
+        end
+
+        context 'when using an unauthenticated request (IP-based)' do
+          let(:ip_env) { { 'PATH_INFO' => '/v3/apps' } }
+          let(:fake_request) do
+            headers = ActionDispatch::Http::Headers.from_hash({ 'HTTP_X_FORWARDED_FOR' => '1.2.3.4' })
+            instance_double(ActionDispatch::Request, fullpath: '/v3/apps', ip: '10.0.0.1', headers: headers)
+          end
+
+          before do
+            allow(ActionDispatch::Request).to receive(:new).and_return(fake_request)
+          end
+
+          it 'uses the X-Forwarded-For IP as the user identifier' do
+            middleware.call(ip_env)
+            expect(counter).to have_received(:try_increment?).with('1.2.3.4', anything)
+          end
+
+          context 'when over the limit' do
+            before do
+              allow(counter).to receive(:try_increment?).and_return(false)
+            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(:request_double) { instance_double(ActionDispatch::Request, fullpath: '/internal/v4/asg_latest_update') }
+
+            before { allow(ActionDispatch::Request).to receive(:new).and_return(request_double) }
+
+            it 'does not rate limit' do
+              middleware.call(internal_env)
+              expect(counter).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(:request_double) { instance_double(ActionDispatch::Request, fullpath: path) }
+
+                before { allow(ActionDispatch::Request).to receive(:new).and_return(request_double) }
+
+                it 'does not rate limit' do
+                  middleware.call(root_env)
+                  expect(counter).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(counter).not_to have_received(:try_increment?)
+            end
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/spec/unit/middleware/concurrent_request_counter_spec.rb b/spec/unit/middleware/concurrent_request_counter_spec.rb
new file mode 100644
index 00000000000..3eb7ee4f07b
--- /dev/null
+++ b/spec/unit/middleware/concurrent_request_counter_spec.rb
@@ -0,0 +1,340 @@
+require 'spec_helper'
+
+module CloudFoundry
+  module Middleware
+    RSpec.describe ConcurrentRequestCounter do
+      let(:logger) { double('logger', info: nil, error: nil) }
+      let(:blocking_limit) { 3 }
+      let(:logging_limit) { 2 }
+      let(:user_guid) { 'user-id-1' }
+
+      subject(:counter) do
+        ConcurrentRequestCounter.new('test', blocking_limit: blocking_limit, logging_limit: logging_limit)
+      end
+
+      before do
+        counter.instance_variable_set(:@store, ConcurrentRequestCounter::InMemoryStore.new)
+      end
+
+      describe '#try_increment?' do
+        it 'returns true when under the blocking limit' do
+          expect(counter.try_increment?(user_guid, logger)).to be true
+        end
+
+        it 'returns false when over the blocking limit' do
+          blocking_limit.times { counter.try_increment?(user_guid, logger) }
+          expect(counter.try_increment?(user_guid, logger)).to be false
+        end
+
+        it 'does not consume a slot for a rejected request (no phantom count)' do
+          blocking_limit.times { counter.try_increment?(user_guid, logger) }
+          5.times { expect(counter.try_increment?(user_guid, logger)).to be false }
+          counter.decrement(user_guid, logger)
+          expect(counter.try_increment?(user_guid, logger)).to be true
+          expect(counter.try_increment?(user_guid, logger)).to be false
+        end
+
+        context 'with only logging_limit (no blocking_limit)' do
+          subject(:counter) { ConcurrentRequestCounter.new('test', logging_limit: logging_limit) }
+
+          before do
+            counter.instance_variable_set(:@store, ConcurrentRequestCounter::InMemoryStore.new)
+          end
+
+          it 'always returns true' do
+            10.times { counter.try_increment?(user_guid, logger) }
+            expect(counter.try_increment?(user_guid, logger)).to be true
+          end
+
+          it 'logs a warning when count exceeds logging_limit but does not block' do
+            logging_limit.times { counter.try_increment?(user_guid, logger) }
+            result = counter.try_increment?(user_guid, logger)
+            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(:counter) { ConcurrentRequestCounter.new('test', blocking_limit: blocking_limit) }
+
+          before do
+            counter.instance_variable_set(:@store, ConcurrentRequestCounter::InMemoryStore.new)
+          end
+
+          it 'returns false when over the blocking limit' do
+            blocking_limit.times { counter.try_increment?(user_guid, logger) }
+            expect(counter.try_increment?(user_guid, logger)).to be false
+          end
+
+          it 'does not log any warnings' do
+            blocking_limit.times { counter.try_increment?(user_guid, logger) }
+            counter.try_increment?(user_guid, logger)
+            expect(logger).not_to have_received(:info).with(/Concurrency limit warning/)
+          end
+        end
+
+        context 'with neither blocking_limit nor logging_limit' do
+          it 'always returns true without hitting the store' do
+            counter = ConcurrentRequestCounter.new('test')
+            store = instance_double(ConcurrentRequestCounter::InMemoryStore)
+            allow(store).to receive(:try_increment)
+            counter.instance_variable_set(:@store, store)
+            expect(counter.try_increment?(user_guid, logger)).to be true
+            expect(store).not_to have_received(:try_increment)
+          end
+        end
+
+        context 'with negative limits (disabled)' do
+          it 'always returns true without hitting the store' do
+            counter = ConcurrentRequestCounter.new('test', blocking_limit: -1, logging_limit: -1)
+            store = instance_double(ConcurrentRequestCounter::InMemoryStore)
+            allow(store).to receive(:try_increment)
+            counter.instance_variable_set(:@store, store)
+            expect(counter.try_increment?(user_guid, logger)).to be true
+            expect(store).not_to have_received(:try_increment)
+          end
+        end
+
+        context 'when store raises StoreError' do
+          before do
+            store = instance_double(ConcurrentRequestCounter::InMemoryStore)
+            allow(store).to receive(:try_increment).and_raise(StoreError)
+            counter.instance_variable_set(:@store, store)
+          end
+
+          it 'fails open and returns true' do
+            expect(counter.try_increment?(user_guid, logger)).to be true
+          end
+        end
+      end
+
+      describe '#decrement' do
+        it 'decrements the counter' do
+          blocking_limit.times { counter.try_increment?(user_guid, logger) }
+          counter.decrement(user_guid, logger)
+          expect(counter.try_increment?(user_guid, logger)).to be true
+        end
+
+        context 'with neither blocking_limit nor logging_limit' do
+          it 'does not hit the store' do
+            counter = ConcurrentRequestCounter.new('test')
+            store = instance_double(ConcurrentRequestCounter::InMemoryStore)
+            allow(store).to receive(:decrement)
+            counter.instance_variable_set(:@store, store)
+            counter.decrement(user_guid, logger)
+            expect(store).not_to have_received(:decrement)
+          end
+        end
+
+        context 'with negative limits (disabled)' do
+          it 'does not hit the store' do
+            counter = ConcurrentRequestCounter.new('test', blocking_limit: -1, logging_limit: -1)
+            store = instance_double(ConcurrentRequestCounter::InMemoryStore)
+            allow(store).to receive(:decrement)
+            counter.instance_variable_set(:@store, store)
+            counter.decrement(user_guid, logger)
+            expect(store).not_to have_received(:decrement)
+          end
+        end
+
+        context 'when store raises StoreError' do
+          before do
+            store = instance_double(ConcurrentRequestCounter::InMemoryStore)
+            allow(store).to receive(:decrement).and_raise(StoreError)
+            counter.instance_variable_set(:@store, store)
+          end
+
+          it 'does not raise' do
+            expect { counter.decrement(user_guid, logger) }.not_to raise_error
+          end
+        end
+      end
+    end
+
+    RSpec.describe ConcurrentRequestCounter::InMemoryStore do
+      let(:store) { ConcurrentRequestCounter::InMemoryStore.new }
+      let(:logger) { double('logger') }
+      let(:key) { 'test-key' }
+
+      describe '#try_increment' do
+        it 'returns 1 for a new key' do
+          expect(store.try_increment(key, 10, logger)).to eq(1)
+        end
+
+        it 'increments on each successful call' do
+          store.try_increment(key, 10, logger)
+          expect(store.try_increment(key, 10, logger)).to eq(2)
+        end
+
+        it 'returns nil without incrementing when at the limit' do
+          2.times { store.try_increment(key, 2, logger) }
+          expect(store.try_increment(key, 2, logger)).to be_nil
+          store.decrement(key, logger)
+          expect(store.try_increment(key, 2, logger)).to eq(2)
+        end
+      end
+
+      describe '#try_log_increment' do
+        it 'always increments with no cap' do
+          5.times { store.try_log_increment(key, logger) }
+          expect(store.try_log_increment(key, logger)).to eq(6)
+        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.try_increment(key, 10, logger)
+          store.try_increment(key, 10, logger)
+          expect(store.decrement(key, logger)).to eq(1)
+        end
+
+        it 'removes the key when count reaches 0' do
+          store.try_increment(key, 10, 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 ConcurrentRequestCounter::RedisStore do
+      let(:logger) { double('logger', error: nil) }
+      let(:key) { 'test-key' }
+      let(:redis) { MockRedis.new }
+      let(:store) do
+        s = ConcurrentRequestCounter::RedisStore.new('/tmp/test.sock', 1, nil)
+        s.instance_variable_set(:@redis, redis)
+        s
+      end
+
+      before do
+        allow(redis).to receive(:evalsha) do |sha, **opts|
+          k = opts[:keys].first
+          if sha == ConcurrentRequestCounter::RedisStore::INCREMENT_SHA
+            limit = opts[:argv][0].to_i
+            ttl   = opts[:argv][1].to_i
+            current = redis.get(k).to_i
+            if current >= limit
+              -1
+            else
+              count = redis.incr(k)
+              redis.expire(k, ttl) if count == 1 && ttl > 0
+              count
+            end
+          elsif sha == ConcurrentRequestCounter::RedisStore::LOG_INCREMENT_SHA
+            ttl   = opts[:argv][0].to_i
+            count = redis.incr(k)
+            redis.expire(k, ttl) if count == 1 && ttl > 0
+            count
+          end
+        end
+      end
+
+      describe '#try_increment' do
+        it 'returns 1 for a new key' do
+          expect(store.try_increment(key, 10, logger)).to eq(1)
+        end
+
+        it 'increments on each successful call' do
+          store.try_increment(key, 10, logger)
+          expect(store.try_increment(key, 10, logger)).to eq(2)
+        end
+
+        it 'returns nil without incrementing when at the limit' do
+          2.times { store.try_increment(key, 2, logger) }
+          expect(store.try_increment(key, 2, logger)).to be_nil
+          expect(redis.get(key).to_i).to eq(2)
+        end
+
+        context 'with TTL configured' do
+          let(:store) do
+            s = ConcurrentRequestCounter::RedisStore.new('/tmp/test.sock', 1, 60)
+            s.instance_variable_set(:@redis, redis)
+            s
+          end
+
+          it 'sets TTL only on the first increment' do
+            allow(redis).to receive(:expire).and_call_original
+            store.try_increment(key, 10, logger)
+            store.try_increment(key, 10, logger)
+            expect(redis).to have_received(:expire).with(key, 60).once
+          end
+        end
+
+        context 'when NOSCRIPT is raised' do
+          before do
+            allow(redis).to receive(:evalsha).and_raise(Redis::CommandError.new('NOSCRIPT No matching script'))
+            # allow(redis).to receive(:eval) is unreliable because Kernel#eval shadows the stub;
+            # define_singleton_method directly on the instance avoids that.
+            redis.define_singleton_method(:eval) { |*_args, **_kwargs| 1 }
+          end
+
+          it 'falls back to eval and returns count' do
+            expect(store.try_increment(key, 10, logger)).to eq(1)
+          end
+        end
+
+        context 'when Redis raises an error' do
+          before { allow(redis).to receive(:evalsha).and_raise(Redis::ConnectionError) }
+
+          it 'logs the error and raises StoreError' do
+            expect { store.try_increment(key, 10, logger) }.to raise_error(StoreError)
+            expect(logger).to have_received(:error).with(/Redis error/)
+          end
+        end
+      end
+
+      describe '#try_log_increment' do
+        it 'always increments with no cap' do
+          5.times { store.try_log_increment(key, logger) }
+          expect(store.try_log_increment(key, logger)).to eq(6)
+        end
+
+        context 'when Redis raises an error' do
+          before { allow(redis).to receive(:evalsha).and_raise(Redis::ConnectionError) }
+
+          it 'logs the error and raises StoreError' do
+            expect { store.try_log_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.try_increment(key, 10, logger)
+          store.try_increment(key, 10, 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.try_increment(key, 10, logger)).to eq(1)
+        end
+
+        it 'returns 0 when key expired mid-flight' do
+          store.try_increment(key, 10, logger)
+          store.instance_variable_get(:@redis).del(key)
+          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/mixins/basic_auth_spec.rb b/spec/unit/middleware/mixins/basic_auth_spec.rb
new file mode 100644
index 00000000000..7f94aabf4e5
--- /dev/null
+++ b/spec/unit/middleware/mixins/basic_auth_spec.rb
@@ -0,0 +1,28 @@
+require 'spec_helper'
+require 'mixins/basic_auth'
+
+module CloudFoundry
+  module Middleware
+    RSpec.describe 'BasicAuth mixin' do
+      let(:implementor) do
+        Class.new { include CloudFoundry::Middleware::BasicAuth }.new
+      end
+
+      describe '#basic_auth?' do
+        it 'returns true when a Basic auth header is present' do
+          env = { 'HTTP_AUTHORIZATION' => 'Basic ' + Base64.encode64('user:pass').strip }
+          expect(implementor.send(:basic_auth?, env)).to be true
+        end
+
+        it 'returns false when no authorization header is present' do
+          expect(implementor.send(:basic_auth?, {})).to be false
+        end
+
+        it 'returns false for a Bearer token' do
+          env = { 'HTTP_AUTHORIZATION' => 'Bearer some-token' }
+          expect(implementor.send(:basic_auth?, env)).to be false
+        end
+      end
+    end
+  end
+end
diff --git a/spec/unit/middleware/mixins/internal_or_root_api_spec.rb b/spec/unit/middleware/mixins/internal_or_root_api_spec.rb
new file mode 100644
index 00000000000..b96ccf19a14
--- /dev/null
+++ b/spec/unit/middleware/mixins/internal_or_root_api_spec.rb
@@ -0,0 +1,40 @@
+require 'spec_helper'
+require 'mixins/internal_or_root_api'
+
+module CloudFoundry
+  module Middleware
+    RSpec.describe 'InternalOrRootApi mixin' do
+      let(:implementor) do
+        Class.new { include CloudFoundry::Middleware::InternalOrRootApi }.new
+      end
+
+      def request_for(path)
+        instance_double(ActionDispatch::Request, fullpath: path)
+      end
+
+      describe '#internal_api?' do
+        it 'returns truthy for /internal paths' do
+          expect(implementor).to be_internal_api(request_for('/internal/v4/asg_latest_update'))
+        end
+
+        it 'returns falsy for non-internal paths' do
+          expect(implementor).not_to be_internal_api(request_for('/v3/apps'))
+        end
+      end
+
+      describe '#root_api?' do
+        ['/v2/info', '/v3', '/', '/healthz'].each do |path|
+          it "returns truthy for #{path}" do
+            expect(implementor).to be_root_api(request_for(path))
+          end
+        end
+
+        ['/v2/apps', '/v3/apps', '/v2/info/extra'].each do |path|
+          it "returns falsy for #{path}" do
+            expect(implementor).not_to be_root_api(request_for(path))
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/spec/unit/middleware/mixins/too_many_requests_spec.rb b/spec/unit/middleware/mixins/too_many_requests_spec.rb
new file mode 100644
index 00000000000..5b549ce0142
--- /dev/null
+++ b/spec/unit/middleware/mixins/too_many_requests_spec.rb
@@ -0,0 +1,60 @@
+require 'spec_helper'
+require 'mixins/too_many_requests'
+
+module CloudFoundry
+  module Middleware
+    RSpec.describe 'TooManyRequests mixin' do
+      let(:implementor) do
+        Class.new { include CloudFoundry::Middleware::TooManyRequests }.new
+      end
+
+      let(:v2_env) { { 'PATH_INFO' => '/v2/apps' } }
+      let(:v3_env) { { 'PATH_INFO' => '/v3/apps' } }
+
+      describe '#too_many_requests!' do
+        it 'returns a 429 status' do
+          status, = implementor.too_many_requests!(v3_env, 'RateLimitExceeded', retry_after: 1)
+          expect(status).to eq(429)
+        end
+
+        it 'sets the Retry-After header' do
+          _, headers, = implementor.too_many_requests!(v3_env, 'RateLimitExceeded', retry_after: 1)
+          expect(headers['Retry-After']).to eq('1')
+        end
+
+        it 'sets Content-Type to text/plain' do
+          _, headers, = implementor.too_many_requests!(v3_env, 'RateLimitExceeded', retry_after: 1)
+          expect(headers['Content-Type']).to eq('text/plain; charset=utf-8')
+        end
+
+        it 'sets Content-Length matching the body size' do
+          _, headers, body = implementor.too_many_requests!(v3_env, 'RateLimitExceeded', retry_after: 1)
+          expect(headers['Content-Length']).to eq(body.first.bytesize.to_s)
+        end
+
+        it 'merges extra_headers without overriding standard headers' do
+          _, headers, = implementor.too_many_requests!(v3_env, 'RateLimitExceeded', retry_after: 1,
+                                                                                    extra_headers: { 'X-RateLimit-Limit' => '100' })
+          expect(headers['X-RateLimit-Limit']).to eq('100')
+          expect(headers['Retry-After']).to eq('1')
+        end
+
+        context 'when the path is /v2' do
+          it 'formats the body as a v2 error' do
+            _, _, body = implementor.too_many_requests!(v2_env, 'RateLimitExceeded', retry_after: 1)
+            json = Oj.load(body.first)
+            expect(json).to include('error_code' => 'CF-RateLimitExceeded')
+          end
+        end
+
+        context 'when the path is /v3' do
+          it 'formats the body as a v3 error' do
+            _, _, body = implementor.too_many_requests!(v3_env, 'RateLimitExceeded', retry_after: 1)
+            json = Oj.load(body.first)
+            expect(json['errors'].first).to include('title' => 'CF-RateLimitExceeded')
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/spec/unit/middleware/mixins/user_id_spec.rb b/spec/unit/middleware/mixins/user_id_spec.rb
new file mode 100644
index 00000000000..a81e1515a24
--- /dev/null
+++ b/spec/unit/middleware/mixins/user_id_spec.rb
@@ -0,0 +1,66 @@
+require 'spec_helper'
+require 'mixins/user_id'
+
+module CloudFoundry
+  module Middleware
+    RSpec.describe 'UserId mixin' do
+      let(:implementor) do
+        Class.new { include CloudFoundry::Middleware::UserId }.new
+      end
+
+      describe '#get_user_id' do
+        context 'when cf.user_guid is present' do
+          it 'returns the user guid' do
+            env = { 'cf.user_guid' => 'some-user-guid', 'PATH_INFO' => '/v3/apps' }
+            expect(implementor.get_user_id(env)).to eq('some-user-guid')
+          end
+        end
+
+        context 'when cf.user_guid is absent' do
+          context 'when X-Forwarded-For is present' do
+            let(:env) { { 'PATH_INFO' => '/v3/apps', 'HTTP_X_FORWARDED_FOR' => '1.2.3.4' } }
+            let(:request_double) do
+              headers = ActionDispatch::Http::Headers.from_hash({ 'HTTP_X_FORWARDED_FOR' => '1.2.3.4' })
+              instance_double(ActionDispatch::Request, headers: headers, ip: '10.0.0.1')
+            end
+
+            before { allow(ActionDispatch::Request).to receive(:new).with(env).and_return(request_double) }
+
+            it 'returns the X-Forwarded-For IP' do
+              expect(implementor.get_user_id(env)).to eq('1.2.3.4')
+            end
+          end
+
+          context 'when X-Forwarded-For is absent' do
+            let(:env) { { 'PATH_INFO' => '/v3/apps', 'REMOTE_ADDR' => '10.0.0.1' } }
+            let(:request_double) do
+              headers = ActionDispatch::Http::Headers.from_hash({})
+              instance_double(ActionDispatch::Request, headers: headers, ip: '10.0.0.1')
+            end
+
+            before { allow(ActionDispatch::Request).to receive(:new).with(env).and_return(request_double) }
+
+            it 'falls back to request.ip' do
+              expect(implementor.get_user_id(env)).to eq('10.0.0.1')
+            end
+          end
+        end
+      end
+
+      describe '#user_token?' do
+        it 'returns true when cf.user_guid is set' do
+          env = { 'cf.user_guid' => 'some-user-guid' }
+          expect(implementor.send(:user_token?, env)).to be true
+        end
+
+        it 'returns false when cf.user_guid is absent' do
+          expect(implementor.send(:user_token?, {})).to be false
+        end
+
+        it 'returns false when cf.user_guid is nil' do
+          expect(implementor.send(:user_token?, { 'cf.user_guid' => nil })).to be false
+        end
+      end
+    end
+  end
+end
diff --git a/spec/unit/middleware/service_broker_rate_limiter_spec.rb b/spec/unit/middleware/service_broker_rate_limiter_spec.rb
index a192aed424e..30b7f62e2f4 100644
--- a/spec/unit/middleware/service_broker_rate_limiter_spec.rb
+++ b/spec/unit/middleware/service_broker_rate_limiter_spec.rb
@@ -4,7 +4,7 @@ module CloudFoundry
   module Middleware
     RSpec.describe ServiceBrokerRateLimiter, isolation: :truncation do
       let(:app) { double(:app) }
-      let(:logger) { double }
+      let(:logger) { double('logger', info: nil) }
       let(:instance) { create(:service, instances_retrievable: true) }
       let(:path_info) { '/v2/service_instances' }
       let(:user_guid) { SecureRandom.uuid }
@@ -13,11 +13,14 @@ module Middleware
       let(:fake_request) { instance_double(ActionDispatch::Request, fullpath: '/v2/service_instances', method: 'POST') }
       let(:max_concurrent_requests) { 1 }
       let(:broker_timeout) { 60 }
+      let(:counter) { instance_double(ConcurrentRequestCounter, try_increment?: true, decrement: nil) }
+
       let(:middleware) do
         ServiceBrokerRateLimiter.new(app, logger: logger, max_concurrent_requests: max_concurrent_requests, broker_timeout_seconds: broker_timeout)
       end
 
       before do
+        allow(ConcurrentRequestCounter).to receive(:instance).and_return(counter)
         allow(ActionDispatch::Request).to receive(:new).and_return(fake_request)
         allow(logger).to receive(:info)
         allow(app).to receive(:call).and_return([200, {}, 'a body'])
@@ -80,6 +83,12 @@ module Middleware
                 let(:request_method) { method }
                 let(:fake_request) { instance_double(ActionDispatch::Request, fullpath: endpoint, method: method) }
 
+                before do
+                  real_counter = ConcurrentRequestCounter.new('test-service-broker', blocking_limit: max_concurrent_requests)
+                  real_counter.instance_variable_set(:@store, ConcurrentRequestCounter::InMemoryStore.new)
+                  middleware.instance_variable_set(:@counter, real_counter)
+                end
+
                 it 'does not allow more than the max number of concurrent requests' do
                   gate = Queue.new
                   allow(app).to receive(:call) do
@@ -134,62 +143,62 @@ module Middleware
             end
           end
         end
+      end
 
-        describe 'errors' do
-          let(:max_concurrent_requests) { 0 }
-
-          context 'when the path is /v2/*' do
-            let(:path_info) { '/v2/service_instances' }
-
-            it 'formats the response error in v2 format' do
-              Timecop.freeze do
-                _, response_headers, body = middleware.call(user_env)
-                json_body = Oj.load(body.first)
-                expect(json_body).to include(
-                  'code' => 10_016,
-                  'description' => 'Service broker concurrent request limit exceeded',
-                  'error_code' => 'CF-ServiceBrokerRateLimitExceeded'
-                )
-                expect(response_headers['Retry-After'].to_i).to be_between((broker_timeout * 0.5).floor, (broker_timeout * 1.5).ceil)
-              end
+      describe 'errors' do
+        let(:max_concurrent_requests) { 0 }
+
+        before do
+          allow(counter).to receive(:try_increment?).and_return(false)
+        end
+
+        context 'when the path is /v2/*' do
+          let(:path_info) { '/v2/service_instances' }
+
+          it 'formats the response error in v2 format' do
+            Timecop.freeze do
+              _, response_headers, body = middleware.call(user_env)
+              json_body = Oj.load(body.first)
+              expect(json_body).to include(
+                'code' => 10_016,
+                'description' => 'Service broker concurrent request limit exceeded',
+                'error_code' => 'CF-ServiceBrokerRateLimitExceeded'
+              )
+              expect(response_headers['Retry-After'].to_i).to be_between((broker_timeout * 0.5).floor, (broker_timeout * 1.5).ceil)
             end
           end
+        end
 
-          context 'when the path is /v3/*' do
-            let(:path_info) { '/v3/service_instances' }
-
-            it 'formats the response error in v3 format' do
-              Timecop.freeze do
-                _, response_headers, body = middleware.call(user_env)
-                json_body = Oj.load(body.first)
-                expect(json_body['errors'].first).to include(
-                  'code' => 10_016,
-                  'detail' => 'Service broker concurrent request limit exceeded',
-                  'title' => 'CF-ServiceBrokerRateLimitExceeded'
-                )
-                expect(response_headers['Retry-After'].to_i).to be_between((broker_timeout * 0.5).floor, (broker_timeout * 1.5).ceil)
-              end
+        context 'when the path is /v3/*' do
+          let(:path_info) { '/v3/service_instances' }
+
+          it 'formats the response error in v3 format' do
+            Timecop.freeze do
+              _, response_headers, body = middleware.call(user_env)
+              json_body = Oj.load(body.first)
+              expect(json_body['errors'].first).to include(
+                'code' => 10_016,
+                'detail' => 'Service broker concurrent request limit exceeded',
+                'title' => 'CF-ServiceBrokerRateLimitExceeded'
+              )
+              expect(response_headers['Retry-After'].to_i).to be_between((broker_timeout * 0.5).floor, (broker_timeout * 1.5).ceil)
             end
           end
+        end
 
-          context 'when broker_client_timeout_seconds is reduced' do
-            let(:broker_timeout) { 3 }
+        context 'when broker_client_timeout_seconds is reduced' do
+          let(:broker_timeout) { 3 }
 
-            it 'reduces the suggested delay in the Retry-After header' do
-              Timecop.freeze do
-                _, response_headers, = middleware.call(user_env)
-                expect(response_headers['Retry-After'].to_i).to be_between((broker_timeout * 0.5).floor, (broker_timeout * 1.5).ceil)
-              end
+          it 'reduces the suggested delay in the Retry-After header' do
+            Timecop.freeze do
+              _, response_headers, = middleware.call(user_env)
+              expect(response_headers['Retry-After'].to_i).to be_between((broker_timeout * 0.5).floor, (broker_timeout * 1.5).ceil)
             end
           end
         end
       end
 
       describe 'skipped requests' do
-        let(:concurrent_request_counter) { double }
-
-        before { middleware.instance_variable_set('@concurrent_request_counter', concurrent_request_counter) }
-
         context 'user is an admin' do
           before do
             allow(VCAP::CloudController::SecurityContext).to receive(:admin?).and_return(true)
@@ -197,7 +206,7 @@ module Middleware
 
           it 'does not rate limit them' do
             middleware.call(user_env)
-            expect(concurrent_request_counter).not_to receive(:try_increment?)
+            expect(counter).not_to have_received(:try_increment?)
             expect(app).to have_received(:call)
           end
         end
@@ -207,7 +216,7 @@ module Middleware
 
           it 'does not rate limit them' do
             middleware.call(user_env)
-            expect(concurrent_request_counter).not_to receive(:try_increment?)
+            expect(counter).not_to have_received(:try_increment?)
             expect(app).to have_received(:call)
           end
         end
@@ -217,134 +226,11 @@ module Middleware
 
           it 'does not rate limit them' do
             middleware.call(user_env)
-            expect(concurrent_request_counter).not_to receive(:try_increment?)
+            expect(counter).not_to have_received(:try_increment?)
             expect(app).to have_received(:call)
           end
         end
       end
     end
-
-    RSpec.describe ConcurrentRequestCounter do
-      store_implementations = []
-      store_implementations << :in_memory   # Test the ConcurrentRequestCounter::InMemoryStore
-      store_implementations << :mock_redis  # Test the ConcurrentRequestCounter::RedisStore with MockRedis
-
-      store_implementations.each do |store_implementation|
-        describe store_implementation do
-          let(:concurrent_request_counter) { ConcurrentRequestCounter.new('test') }
-          let(:store) { concurrent_request_counter.instance_variable_get(:@store) }
-          let(:user_guid) { SecureRandom.uuid }
-          let(:max_concurrent_requests) { 5 }
-          let(:logger) { double('logger', info: nil) }
-
-          before do
-            TestConfig.override(redis: { socket: 'foo' }, puma: { max_threads: 123 }) unless store_implementation == :in_memory
-
-            allow(ConnectionPool::Wrapper).to receive(:new).and_call_original
-            concurrent_request_counter.send(:store) # instantiate counter and store implementation
-          end
-
-          describe '#initialize' do
-            it 'sets the @key_prefix' do
-              expect(concurrent_request_counter.instance_variable_get(:@key_prefix)).to eq('test')
-            end
-
-            it 'instantiates the appropriate store class' do
-              if store_implementation == :in_memory
-                expect(store).to be_a(ConcurrentRequestCounter::InMemoryStore)
-              else
-                expect(store).to be_a(ConcurrentRequestCounter::RedisStore)
-              end
-            end
-
-            it 'uses a connection pool size that equals the maximum puma threads' do
-              skip('Not relevant for InMemoryStore') if store_implementation == :in_memory
-
-              expect(ConnectionPool::Wrapper).to have_received(:new).with(size: 123)
-            end
-
-            context 'with custom connection pool size' do
-              let(:concurrent_request_counter) { ConcurrentRequestCounter.new('test', redis_connection_pool_size: 456) }
-
-              it 'uses the provided connection pool size' do
-                skip('Not relevant for InMemoryStore') if store_implementation == :in_memory
-
-                expect(ConnectionPool::Wrapper).to have_received(:new).with(size: 456)
-              end
-            end
-          end
-
-          describe '#try_increment?' do
-            it 'calls @store.try_increment? with the prefixed user guid and the given maximum concurrent requests and logger' do
-              allow(store).to receive(:try_increment?).and_call_original
-              concurrent_request_counter.try_increment?(user_guid, max_concurrent_requests, logger)
-              expect(store).to have_received(:try_increment?).with("test:#{user_guid}", max_concurrent_requests, logger)
-            end
-
-            it 'returns true for a new user' do
-              expect(concurrent_request_counter).to be_try_increment(user_guid, max_concurrent_requests, logger)
-            end
-
-            it 'returns true for a recurring user performing the maximum allowed concurrent requests' do
-              (max_concurrent_requests - 1).times { concurrent_request_counter.try_increment?(user_guid, max_concurrent_requests, logger) }
-              expect(concurrent_request_counter).to be_try_increment(user_guid, max_concurrent_requests, logger)
-            end
-
-            it 'returns false for a recurring user with too many concurrent requests' do
-              max_concurrent_requests.times { concurrent_request_counter.try_increment?(user_guid, max_concurrent_requests, logger) }
-              expect(concurrent_request_counter).not_to be_try_increment(user_guid, max_concurrent_requests, logger)
-            end
-
-            it 'returns true again for a recurring user after a single decrement' do
-              (max_concurrent_requests + 1).times { concurrent_request_counter.try_increment?(user_guid, max_concurrent_requests, logger) }
-              concurrent_request_counter.decrement(user_guid, logger)
-              expect(concurrent_request_counter).to be_try_increment(user_guid, max_concurrent_requests, logger)
-            end
-
-            it 'returns true in case of a Redis error' do
-              skip('Not relevant for InMemoryStore') if store_implementation == :in_memory
-
-              allow_any_instance_of(MockRedis::StringMethods).to receive(:incr).and_raise(Redis::ConnectionError)
-              allow_any_instance_of(Redis).to receive(:incr).and_raise(Redis::ConnectionError)
-              allow(logger).to receive(:error)
-
-              expect(concurrent_request_counter).to be_try_increment(user_guid, max_concurrent_requests, logger)
-              expect(logger).to have_received(:error).with(/Redis error/)
-            end
-          end
-
-          describe '#decrement' do
-            it 'calls @store.decrement with the prefixed user guid and the given logger' do
-              allow(store).to receive(:decrement).and_call_original
-              concurrent_request_counter.decrement(user_guid, logger)
-              expect(store).to have_received(:decrement).with("test:#{user_guid}", logger)
-            end
-
-            it 'decreases the number of concurrent requests, allowing for another concurrent request' do
-              max_concurrent_requests.times { concurrent_request_counter.try_increment?(user_guid, max_concurrent_requests, logger) }
-              concurrent_request_counter.decrement(user_guid, logger)
-              expect(concurrent_request_counter).to be_try_increment(user_guid, max_concurrent_requests, logger)
-            end
-
-            it 'does not decrease the number of concurrent requests below zero' do
-              concurrent_request_counter.decrement(user_guid, logger)
-              max_concurrent_requests.times { concurrent_request_counter.try_increment?(user_guid, max_concurrent_requests, logger) }
-              expect(concurrent_request_counter).not_to be_try_increment(user_guid, max_concurrent_requests, logger)
-            end
-
-            it 'writes an error log in case of a Redis error' do
-              skip('Not relevant for InMemoryStore') if store_implementation == :in_memory
-
-              allow_any_instance_of(MockRedis::StringMethods).to receive(:decr).and_raise(Redis::ConnectionError)
-              allow_any_instance_of(Redis).to receive(:decr).and_raise(Redis::ConnectionError)
-              allow(logger).to receive(:error)
-
-              concurrent_request_counter.decrement(user_guid, logger)
-              expect(logger).to have_received(:error).with(/Redis error/)
-            end
-          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..be200608cb8
--- /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
+          middleware.call(env)
+        end
+      end
+    end
+  end
+end