diff --git a/lib/vmpooler/pool_manager.rb b/lib/vmpooler/pool_manager.rb
index 0580d332..ed2fe704 100644
--- a/lib/vmpooler/pool_manager.rb
+++ b/lib/vmpooler/pool_manager.rb
@@ -12,6 +12,11 @@ class PoolManager
     CHECK_LOOP_DELAY_MAX_DEFAULT = 60
     CHECK_LOOP_DELAY_DECAY_DEFAULT = 2.0
 
+    # Number of pre-allocated mutexes that ondemand request_ids are hashed
+    # across. See #request_mutex for why this is striped rather than a
+    # per-request_id map.
+    REQUEST_MUTEX_STRIPE_COUNT = 256
+
     def initialize(config, logger, redis_connection_pool, metrics)
       $config = config
 
@@ -34,9 +39,10 @@ def initialize(config, logger, redis_connection_pool, metrics)
       $threads = Concurrent::Hash.new
 
       # Pool mutex
-      @reconfigure_pool = Concurrent::Hash.new
+      @reconfigure_pool = Concurrent::Map.new
 
-      @vm_mutex = Concurrent::Hash.new
+      @vm_mutex = Concurrent::Map.new
+      @request_mutex_stripes = Array.new(REQUEST_MUTEX_STRIPE_COUNT) { Mutex.new }
 
       # Name generator for generating host names
       @name_generator = Spicy::Proton.new
@@ -242,29 +248,34 @@ def move_pending_vm_to_ready(vm, pool, redis, request_id = nil)
       finish = format('%<time>.2f', time: Time.now - Time.parse(clone_time))
 
       if request_id
-        ondemandrequest_hash = redis.hgetall("vmpooler__odrequest__#{request_id}")
-        case ondemandrequest_hash['status']
-        when 'failed'
-          move_vm_queue(pool, vm, 'pending', 'completed', redis, "moved to completed queue. '#{request_id}' could not be filled in time")
-          return nil
-        when 'deleted'
-          move_vm_queue(pool, vm, 'pending', 'completed', redis, "moved to completed queue. '#{request_id}' has been deleted")
-          return nil
-        end
-        pool_alias = redis.hget("vmpooler__vm__#{vm}", 'pool_alias')
+        request_mutex(request_id).synchronize do
+          ondemandrequest_hash = redis.hgetall("vmpooler__odrequest__#{request_id}")
+          case ondemandrequest_hash['status']
+          when 'failed'
+            move_vm_queue(pool, vm, 'pending', 'completed', redis, "moved to completed queue. '#{request_id}' could not be filled in time")
+            return nil
+          when 'deleted'
+            move_vm_queue(pool, vm, 'pending', 'completed', redis, "moved to completed queue. '#{request_id}' has been deleted")
+            return nil
+          end
+          pool_alias = redis.hget("vmpooler__vm__#{vm}", 'pool_alias')
 
-        redis.pipelined do |pipeline|
-          pipeline.hset("vmpooler__active__#{pool}", vm, Time.now.to_s)
-          pipeline.hset("vmpooler__vm__#{vm}", 'checkout', Time.now.to_s)
-          if ondemandrequest_hash['token:token']
-            pipeline.hset("vmpooler__vm__#{vm}", 'token:token', ondemandrequest_hash['token:token'])
-            pipeline.hset("vmpooler__vm__#{vm}", 'token:user', ondemandrequest_hash['token:user'])
-            pipeline.hset("vmpooler__vm__#{vm}", 'lifetime', $config[:config]['vm_lifetime_auth'].to_i)
+          redis.pipelined do |pipeline|
+            pipeline.hset("vmpooler__active__#{pool}", vm, Time.now.to_s)
+            pipeline.hset("vmpooler__vm__#{vm}", 'checkout', Time.now.to_s)
+            if ondemandrequest_hash['token:token']
+              pipeline.hset("vmpooler__vm__#{vm}", 'token:token', ondemandrequest_hash['token:token'])
+              pipeline.hset("vmpooler__vm__#{vm}", 'token:user', ondemandrequest_hash['token:user'])
+              pipeline.hset("vmpooler__vm__#{vm}", 'lifetime', $config[:config]['vm_lifetime_auth'].to_i)
+            end
+            pipeline.sadd("vmpooler__#{request_id}__#{pool_alias}__#{pool}", vm)
           end
-          pipeline.sadd("vmpooler__#{request_id}__#{pool_alias}__#{pool}", vm)
+          move_vm_queue(pool, vm, 'pending', 'running', redis)
+          # already holding request_mutex(request_id) from the top of this
+          # block, so call the lock-free variant directly to avoid deadlocking
+          # on Ruby's non-reentrant Mutex
+          check_ondemand_request_ready_locked(request_id, redis)
         end
-        move_vm_queue(pool, vm, 'pending', 'running', redis)
-        check_ondemand_request_ready(request_id, redis)
       else
         redis.smove("vmpooler__pending__#{pool}", "vmpooler__ready__#{pool}", vm)
       end
@@ -1775,11 +1786,41 @@ def check_pool(pool,
     end
 
     def pool_mutex(poolname)
-      @reconfigure_pool[poolname] || @reconfigure_pool[poolname] = Mutex.new
+      @reconfigure_pool.compute_if_absent(poolname) { Mutex.new }
     end
 
     def vm_mutex(vmname)
-      @vm_mutex[vmname] || @vm_mutex[vmname] = Mutex.new
+      @vm_mutex.compute_if_absent(vmname) { Mutex.new }
+    end
+
+    # Striped rather than per-request_id: request_ids are one-shot and never
+    # revisited once terminal (ready/failed/deleted), so a growable
+    # compute_if_absent+delete map (like pool_mutex/vm_mutex) would leak
+    # forever if never deleted, or - if deleted - risks a caller that already
+    # fetched the old Mutex still being blocked/about to synchronize on it
+    # while a later caller, finding the entry gone, gets handed a brand new
+    # Mutex for the same request_id. That would let both callers enter the
+    # request's critical section concurrently, defeating the whole point of
+    # this lock (see puppetlabs/vmpooler#702 review discussion). A fixed-size
+    # array of pre-allocated mutexes needs no creation or deletion at all, so
+    # this can't happen; unrelated request_ids occasionally sharing a stripe
+    # just means incidental extra contention, not a correctness problem.
+    #
+    # LOCK ORDERING: _check_pending_vm holds vm_mutex(vm) while calling
+    # move_pending_vm_to_ready, which acquires request_mutex(request_id) - so
+    # the established order is vm_mutex -> request_mutex. Anything acquiring
+    # request_mutex first (e.g. the ondemand_provisioner sweep) must not then
+    # acquire vm_mutex, or the two will deadlock. Note in particular that
+    # remove_vms_for_failed_request currently destroys VMs without taking
+    # their vm_mutex; adding that locking would need this order respected.
+    #
+    # SCOPE: this is an in-process Mutex, so it only serializes threads within
+    # a single manager process. It does not coordinate with the api process
+    # (which mutates request status and VM queues in the ondemand delete
+    # endpoint), and it provides no protection at all if the manager is run
+    # with more than one replica.
+    def request_mutex(request_id)
+      @request_mutex_stripes[request_id.hash.abs % REQUEST_MUTEX_STRIPE_COUNT]
     end
 
     def dereference_mutex(vmname)
@@ -2385,22 +2426,55 @@ def check_ondemand_requests_ready(redis)
       in_progress_requests.length
     end
 
+    # Acquires the per-request lock before delegating. Callers that already
+    # hold request_mutex(request_id) (e.g. move_pending_vm_to_ready) must call
+    # check_ondemand_request_ready_locked directly instead - Ruby's Mutex is
+    # not reentrant, so re-synchronizing here would deadlock.
     def check_ondemand_request_ready(request_id, redis, score = nil)
+      request_mutex(request_id).synchronize do
+        check_ondemand_request_ready_locked(request_id, redis, score)
+      end
+    end
+
+    # MUST be called with request_mutex(request_id) already held - it does no
+    # locking of its own. Use check_ondemand_request_ready if you do not
+    # already hold the lock.
+    def check_ondemand_request_ready_locked(request_id, redis, score = nil)
       # default expiration is one month to ensure the data does not stay in redis forever
       default_expiration = 259_200_0
       processing_key = 'vmpooler__provisioning__processing'
       ondemand_hash_key = "vmpooler__odrequest__#{request_id}"
-      score ||= redis.zscore(processing_key, request_id)
-      return if request_expired?(request_id, score, redis)
 
-      return unless vms_ready?(request_id, redis)
+      # Never resurrect a request that already reached a terminal state. The
+      # ondemand delete endpoint sets 'deleted' (from the api process, which
+      # this lock cannot coordinate with) without removing the request from
+      # the processing set, so the sweep keeps visiting it; drop it here so we
+      # neither overwrite its status nor revisit it forever.
+      if %w[failed deleted].include?(redis.hget(ondemand_hash_key, 'status'))
+        redis.zrem(processing_key, request_id)
+        return
+      end
 
-      redis.hset(ondemand_hash_key, 'status', 'ready')
-      redis.expire(ondemand_hash_key, default_expiration)
-      redis.zrem(processing_key, request_id)
+      # Check readiness before expiry: a request that just became fully
+      # provisioned must not be killed by a concurrent expiry sweep landing
+      # on the same tick.
+      if vms_ready?(request_id, redis)
+        redis.hset(ondemand_hash_key, 'status', 'ready')
+        redis.expire(ondemand_hash_key, default_expiration)
+        redis.zrem(processing_key, request_id)
+        return
+      end
+
+      score ||= redis.zscore(processing_key, request_id)
+      request_expired?(request_id, score, redis)
     end
 
     def request_expired?(request_id, score, redis)
+      # A missing score means there is nothing in the processing set to
+      # expire (e.g. the request already completed) - treat as not expired
+      # rather than letting `nil.to_i` (0) produce a runaway delta.
+      return false if score.nil?
+
       delta = Time.now.to_i - score.to_i
       ondemand_request_ttl = $config[:config]['ondemand_request_ttl']
       return false unless delta > ondemand_request_ttl * 60
diff --git a/spec/unit/pool_manager_spec.rb b/spec/unit/pool_manager_spec.rb
index 1b2ccef2..eb897a65 100644
--- a/spec/unit/pool_manager_spec.rb
+++ b/spec/unit/pool_manager_spec.rb
@@ -1,5 +1,6 @@
 require 'spec_helper'
 require 'time'
+require 'timeout'
 require 'mock_redis'
 
 # Custom RSpec :Matchers
@@ -541,7 +542,7 @@
     context 'with request_id' do
       context 'with a pending request' do
         before(:each) do
-          allow(subject).to receive(:check_ondemand_request_ready)
+          allow(subject).to receive(:check_ondemand_request_ready_locked)
           config[:config]['ondemand_request_ttl'] = 20
         end
 
@@ -566,6 +567,142 @@
         end
       end
 
+      context 'when this is the last vm needed to fulfill the request' do
+        let(:platform_alias) { pool }
+        let(:platforms_string) { "#{platform_alias}:#{pool}:1" }
+
+        before(:each) do
+          config[:config]['ondemand_request_ttl'] = 20
+          redis_connection_pool.with do |redis|
+            create_ondemand_request_for_test(request_id, current_time.to_i, platforms_string, redis)
+            create_ondemand_processing(request_id, current_time.to_i, redis)
+            redis.hset("vmpooler__vm__#{vm}", 'pool_alias', pool)
+          end
+        end
+
+        it 'does not deadlock marking the request ready (regression: move_pending_vm_to_ready already holds request_mutex)' do
+          redis_connection_pool.with do |redis|
+            expect { subject.move_pending_vm_to_ready(vm, pool, redis, request_id) }.to_not raise_error
+          end
+        end
+
+        it 'marks the request as ready' do
+          redis_connection_pool.with do |redis|
+            subject.move_pending_vm_to_ready(vm, pool, redis, request_id)
+            expect(redis.hget("vmpooler__odrequest__#{request_id}", 'status')).to eq('ready')
+          end
+        end
+      end
+
+      context 'when the expiry sweep runs concurrently with the last vm completing the request' do
+        # Deterministic two-thread test: proves request_mutex actually
+        # provides mutual exclusion between move_pending_vm_to_ready and the
+        # expiry sweep (check_ondemand_request_ready), rather than merely
+        # proving the recursive-lock deadlock is gone. Uses a single shared
+        # MockRedis instance directly (not redis_connection_pool, which is
+        # sized to 1 connection here and would otherwise block on pool
+        # checkout rather than on the mutex we're testing).
+        let(:platform_alias) { pool }
+        let(:platforms_string) { "#{platform_alias}:#{pool}:1" }
+        # older than the ttl configured below, so the sweep considers the
+        # request expired the instant it can inspect it
+        let(:expired_score) { (Time.now - 960).to_i }
+        let(:shared_redis) { MockRedis.new }
+
+        before(:each) do
+          config[:config]['ondemand_request_ttl'] = 5
+          create_ondemand_request_for_test(request_id, expired_score, platforms_string, shared_redis)
+          create_ondemand_processing(request_id, expired_score, shared_redis)
+          shared_redis.hset("vmpooler__vm__#{vm}", 'pool_alias', pool)
+          shared_redis.hset("vmpooler__vm__#{vm}", 'clone', Time.now.to_s)
+          shared_redis.sadd("vmpooler__pending__#{pool}", vm)
+        end
+
+        it 'blocks the sweep from entering its critical section until move_pending_vm_to_ready releases it, and never runs both concurrently' do
+          reached_move_pause = Queue.new
+          release_move_pending = Queue.new
+          entered_locked_check = Queue.new
+          bookkeeping_mutex = Mutex.new
+          active_count = 0
+          max_active = 0
+
+          # Pause move_pending_vm_to_ready only at its 'pending' -> 'running'
+          # transition (unique to this call site) while it still holds
+          # request_mutex, so we can deterministically run the sweep
+          # concurrently and prove it can't make progress until this releases.
+          allow(subject).to receive(:move_vm_queue).and_wrap_original do |original, *args|
+            if args[2] == 'pending' && args[3] == 'running'
+              reached_move_pause << true
+              release_move_pending.pop
+            end
+            original.call(*args)
+          end
+
+          # Instrument the actual critical section both callers eventually
+          # reach, so we can directly observe: (a) whether the sweep manages
+          # to enter it while move_pending_vm_to_ready is paused holding the
+          # lock, and (b) whether the two ever overlap.
+          original_locked_check = subject.method(:check_ondemand_request_ready_locked)
+          allow(subject).to receive(:check_ondemand_request_ready_locked) do |*args|
+            entered_locked_check << true
+            bookkeeping_mutex.synchronize do
+              active_count += 1
+              max_active = active_count if active_count > max_active
+            end
+            begin
+              original_locked_check.call(*args)
+            ensure
+              bookkeeping_mutex.synchronize { active_count -= 1 }
+            end
+          end
+
+          Timeout.timeout(5) do
+            move_thread = Thread.new do
+              subject.move_pending_vm_to_ready(vm, pool, shared_redis, request_id)
+            end
+
+            reached_move_pause.pop
+
+            sweep_thread = Thread.new do
+              subject.check_ondemand_request_ready(request_id, shared_redis, expired_score)
+            end
+
+            # While move_pending_vm_to_ready still holds request_mutex, the
+            # sweep must not be able to enter its own critical section yet -
+            # give it a generous 0.3s window (far longer than an in-memory
+            # mock-redis call sequence needs) to prove it's genuinely blocked,
+            # not just slow.
+            #
+            # Note the flake direction is deliberately one-sided: a working
+            # lock can never spuriously fail this (the sweep is blocked
+            # indefinitely, not merely for longer than 0.3s), whereas a broken
+            # lock on a pathologically slow runner could occasionally escape
+            # detection. It can under-report a regression, never invent one.
+            sweep_progress = Thread.new do
+              Timeout.timeout(0.3) { entered_locked_check.pop }
+              :entered
+            rescue Timeout::Error
+              :blocked
+            end
+            expect(sweep_progress.value).to eq(:blocked)
+
+            release_move_pending << true
+
+            move_thread.join
+            sweep_thread.join
+          end
+
+          # Covers only overlap of the check_ondemand_request_ready_locked
+          # call itself; the :blocked assertion above is what proves the sweep
+          # can't overlap the earlier part of move_pending_vm_to_ready's
+          # critical section (the status read and fulfillment-set write).
+          expect(max_active).to eq(1)
+          expect(shared_redis.hget("vmpooler__odrequest__#{request_id}", 'status')).to eq('ready')
+          expect(shared_redis.sismember("vmpooler__completed__#{pool}", vm)).to be false
+          expect(shared_redis.sismember("vmpooler__running__#{pool}", vm)).to be true
+        end
+      end
+
       context 'when the request has been marked as failed' do
         before(:each) do
           redis_connection_pool.with do |redis|
@@ -617,7 +754,7 @@
         let(:score) { current_time.to_i }
         before(:each) do
           config[:config]['ondemand_request_ttl'] = 20
-          allow(subject).to receive(:check_ondemand_request_ready)
+          allow(subject).to receive(:check_ondemand_request_ready_locked)
           redis_connection_pool.with do |redis|
             create_ondemand_request_for_test(request_id, score, platforms_string, redis, user, token)
           end
@@ -5428,7 +5565,9 @@
       before(:each) do
         expect(subject).to receive(:vms_ready?).and_return(true)
         redis_connection_pool.with do |redis|
-          expect(redis).to receive(:zscore).and_return(score)
+          # readiness is checked before expiry, so a ready request must not
+          # need to look up its score at all
+          allow(redis).to receive(:zscore).and_return(score)
         end
       end
 
@@ -5468,12 +5607,58 @@
     context 'when a request has taken too long to be filled' do
       it 'should return true for request_expired?' do
         redis_connection_pool.with do |redis|
+          expect(subject).to receive(:vms_ready?).and_return(false)
           expect(redis).to receive(:zscore).and_return(score)
           expect(subject).to receive(:request_expired?).with(request_id, Float, redis).and_return(true)
           subject.check_ondemand_request_ready(request_id, redis)
         end
       end
     end
+
+    %w[failed deleted].each do |terminal_status|
+      context "when the request has already been marked as '#{terminal_status}'" do
+        before(:each) do
+          redis_connection_pool.with do |redis|
+            create_ondemand_processing(request_id, current_time.to_i, redis)
+            set_ondemand_request_status(request_id, terminal_status, redis)
+          end
+        end
+
+        it 'does not overwrite the status' do
+          redis_connection_pool.with do |redis|
+            subject.check_ondemand_request_ready(request_id, redis)
+            expect(redis.hget("vmpooler__odrequest__#{request_id}", 'status')).to eq(terminal_status)
+          end
+        end
+
+        it 'stops revisiting it by removing it from the processing set' do
+          redis_connection_pool.with do |redis|
+            subject.check_ondemand_request_ready(request_id, redis)
+            expect(redis.zscore('vmpooler__provisioning__processing', request_id)).to be nil
+          end
+        end
+
+        it 'does not evaluate readiness or expiry' do
+          redis_connection_pool.with do |redis|
+            expect(subject).to_not receive(:vms_ready?)
+            expect(subject).to_not receive(:request_expired?)
+            subject.check_ondemand_request_ready(request_id, redis)
+          end
+        end
+      end
+    end
+
+    context 'when the request is not ready and the score has already been removed from processing' do
+      it 'does not treat a missing score as an infinitely old request' do
+        redis_connection_pool.with do |redis|
+          expect(subject).to receive(:vms_ready?).and_return(false)
+          expect(redis).to receive(:zscore).and_return(nil)
+          expect(subject).to receive(:request_expired?).with(request_id, nil, redis).and_call_original
+          expect(subject).to_not receive(:remove_vms_for_failed_request)
+          subject.check_ondemand_request_ready(request_id, redis)
+        end
+      end
+    end
   end
 
   describe '#request_expired?' do
@@ -5549,6 +5734,16 @@
         end
       end
     end
+
+    context 'with a nil score' do
+      it 'returns false without treating the missing score as an infinitely old request' do
+        redis_connection_pool.with do |redis|
+          expect(subject).to_not receive(:remove_vms_for_failed_request)
+          result = subject.request_expired?(request_id, nil, redis)
+          expect(result).to be false
+        end
+      end
+    end
   end
 
   describe '#remove_vms_for_failed_request)' do