diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionService.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionService.java
index 1588e8dfafa97d..ac5711f7e72319 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionService.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionService.java
@@ -153,8 +153,10 @@
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Phaser;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -1809,36 +1811,111 @@ public void uploadOutputs(
 
     if (remoteOptions.getRemoteCacheAsync()
         && !action.getSpawn().getResourceOwner().mayModifySpawnOutputsAfterExecution()) {
-      var uploadDone = new CountDownLatch(1);
-      var future =
-          backgroundTaskExecutor.submit(
-              () -> {
-                try {
-                  doUploadOutputs(action, spawnResult, onUploadComplete);
-                } catch (ExecException e) {
-                  reportUploadError(e);
-                } catch (InterruptedException ignored) {
-                  // ThreadPerTaskExecutor does not care about interrupt status.
-                } finally {
-                  uploadDone.countDown();
-                }
-              });
-
-      if (outputService instanceof RemoteOutputService remoteOutputService
-          && remoteOutputService.getRewoundActionSynchronizer()
-              instanceof RemoteRewoundActionSynchronizer remoteRewoundActionSynchronizer) {
-        remoteRewoundActionSynchronizer.registerOutputUploadTask(
-            action.getRemoteActionExecutionContext().getSpawnOwner(),
-            () -> {
-              future.cancel(true);
-              uploadDone.await();
-            });
-      }
+      new OutputUploadTask(action, spawnResult, onUploadComplete).start(backgroundTaskExecutor);
     } else {
       doUploadOutputs(action, spawnResult, onUploadComplete);
     }
   }
 
+  @Nullable
+  private RemoteRewoundActionSynchronizer getRewoundActionSynchronizer() {
+    if (outputService instanceof RemoteOutputService remoteOutputService
+        && remoteOutputService.getRewoundActionSynchronizer()
+            instanceof RemoteRewoundActionSynchronizer rewoundActionSynchronizer) {
+      return rewoundActionSynchronizer;
+    }
+    return null;
+  }
+
+  /**
+   * A cancellable background upload of an action's outputs.
+   *
+   * <p>Registers itself with the {@link RemoteRewoundActionSynchronizer}, if there is one, before
+   * the upload starts and unregisters itself when it is done, so that a rewinding of the action
+   * waits for uploads that are still in flight.
+   *
+   * <p>Ensures that the completion callback runs exactly once and that {@link #cancel} only returns
+   * once the upload no longer accesses the action's outputs.
+   */
+  private final class OutputUploadTask implements RemoteRewoundActionSynchronizer.Cancellable {
+    private final RemoteAction action;
+    private final SpawnResult spawnResult;
+    private final Runnable onUploadComplete;
+    private final ActionExecutionMetadata spawnOwner;
+    @Nullable private final RemoteRewoundActionSynchronizer rewoundActionSynchronizer;
+    private final CountDownLatch done = new CountDownLatch(1);
+
+    // The thread running the upload, published by the upload itself before it reads cancelled.
+    @Nullable private volatile Thread thread;
+    // Set by cancel before it reads thread.
+    private volatile boolean cancelled;
+
+    OutputUploadTask(RemoteAction action, SpawnResult spawnResult, Runnable onUploadComplete) {
+      this.action = action;
+      this.spawnResult = spawnResult;
+      this.onUploadComplete = onUploadComplete;
+      this.spawnOwner = action.getRemoteActionExecutionContext().getSpawnOwner();
+      this.rewoundActionSynchronizer = getRewoundActionSynchronizer();
+    }
+
+    /** Registers the task for cancellation and starts the upload on the given executor. */
+    void start(Executor executor) {
+      // Register before starting the upload so that it can't unregister itself before it has been
+      // registered.
+      if (rewoundActionSynchronizer != null) {
+        rewoundActionSynchronizer.registerOutputUploadTask(spawnOwner, this);
+      }
+      try {
+        // Runs the upload rather than submitting it: the body of a task submitted to an
+        // ExecutorService is skipped entirely if its future is cancelled before it starts, which
+        // would leave the completion callback unrun and cancel waiting forever.
+        executor.execute(this::upload);
+      } catch (RejectedExecutionException e) {
+        // The upload will never run, so complete the task on its behalf.
+        finish();
+        throw e;
+      }
+    }
+
+    private void upload() {
+      // Publish the thread before reading cancelled, which cancel writes before it reads the
+      // thread. At least one of the two thus observes the other and the upload is either skipped
+      // or interrupted.
+      thread = Thread.currentThread();
+      try {
+        if (cancelled) {
+          onUploadComplete.run();
+        } else {
+          doUploadOutputs(action, spawnResult, onUploadComplete);
+        }
+      } catch (ExecException e) {
+        reportUploadError(e);
+      } catch (InterruptedException ignored) {
+        // ThreadPerTaskExecutor does not care about interrupt status.
+      } finally {
+        finish();
+      }
+    }
+
+    @Override
+    public void cancel() throws InterruptedException {
+      cancelled = true;
+      var localThread = thread;
+      if (localThread != null) {
+        localThread.interrupt();
+      }
+      done.await();
+    }
+
+    /** Signals that the upload no longer accesses the action's outputs. */
+    private void finish() {
+      done.countDown();
+      if (rewoundActionSynchronizer != null) {
+        rewoundActionSynchronizer.unregisterOutputUploadTask(spawnOwner, this);
+      }
+    }
+  }
+
   private void doUploadOutputs(
       RemoteAction action, SpawnResult spawnResult, Runnable onUploadComplete)
       throws ExecException, InterruptedException {
diff --git a/src/main/java/com/google/devtools/build/lib/remote/RemoteRewoundActionSynchronizer.java b/src/main/java/com/google/devtools/build/lib/remote/RemoteRewoundActionSynchronizer.java
index b14263d6ddb3c7..c59a4c3a8e046c 100644
--- a/src/main/java/com/google/devtools/build/lib/remote/RemoteRewoundActionSynchronizer.java
+++ b/src/main/java/com/google/devtools/build/lib/remote/RemoteRewoundActionSynchronizer.java
@@ -14,6 +14,7 @@
 
 package com.google.devtools.build.lib.remote;
 
+
 import com.github.benmanes.caffeine.cache.Caffeine;
 import com.github.benmanes.caffeine.cache.LoadingCache;
 import com.google.common.collect.ImmutableList;
@@ -41,13 +42,22 @@
  * while they are being read.
  */
 final class RemoteRewoundActionSynchronizer implements RewoundActionSynchronizer {
-  /** A task with a cancellation callback. */
+  /** A cancellable task that operates on the outputs of an action. */
   public interface Cancellable {
+    /**
+     * Cancels the task and returns only once it no longer accesses the outputs of the action it has
+     * been registered for, which are about to be regenerated by a rewound execution of that action.
+     *
+     * <p>May be called after the task has completed, in which case it returns immediately.
+     */
     void cancel() throws InterruptedException;
   }
 
   private final RemoteActionInputFetcher actionInputFetcher;
-  private final ConcurrentHashMap<ActionExecutionMetadata, Cancellable> outputUploadTasks =
+
+  // An action generally has at most one such task in flight, but nothing prevents an action from
+  // executing multiple spawns whose outputs are uploaded concurrently.
+  private final ConcurrentHashMap<ActionLookupData, ImmutableList<Cancellable>> outputUploadTasks =
       new ConcurrentHashMap<>();
 
   // A single coarse lock is used to synchronize rewound actions (writers) and both rewound and
@@ -92,33 +102,40 @@ on any (rewound or non-rewound) action executions while it holds read locks and
   * Nodes are given by the currently active Skyframe action execution threads, each of which is
     identified with the action it is (or will be) executing. Actions are in one-to-one
     correspondence with the ActionLookupData that is used as the key in the fine locks map.
-  * For each pair of actions A_1 and A_2, there is an edge from A_1 to A_2 labeled with XY(A_3)
-    if A_1 is waiting for the X lock of A_3 and A_2 currently holds the Y lock of A_3, where X and
-    Y are either R (for read) or W (for write). The resulting graph may have parallel edges with
-    distinct labels.
+  * For each pair of actions A_1 and A_2, there is an edge from A_1 to A_2 labeled with XY(K)
+    if A_1 is waiting for the X lock of the key K and A_2 currently holds the Y lock of K, where X
+    and Y are either R (for read) or W (for write). The resulting graph may have parallel edges
+    with distinct labels.
 
-  Let C be any directed cycle in the graph representing a deadlock, let A_1 -[XY(A_3)]-> A_2 be an
+  Say that an action A "covers" a key K if A is the action identified by K, or if K identifies an
+  ActionTemplate and A is one of its expanded actions. By construction of outputKeyFor, the write
+  lock of K is only ever acquired by actions covering K, and every action covers exactly one key.
+
+  Let C be any directed cycle in the graph representing a deadlock, let A_1 -[XY(K)]-> A_2 be an
   edge in C and consider the following cases for the pair XY:
 
   * RR: Since a read-write lock whose read lock is held by at least one thread doesn't
         block any other thread from acquiring its read lock, this case doesn't occur.
-  * WW: The write lock of A_3 is only ever (attempted to be) acquired by A_3 itself when it is
-        rewound, which means that the edge would necessarily be of the shape A_3 -[WW(A_3)]-> A_3.
-        But this isn't possible since the write lock for an action is only acquired in one place (
-        enterActionPreparationForRewinding) and not recursively.
-  * WR: In this case, A_1 attempts to acquire a write lock, which only happens when A_1 is a
-        rewound action about to prepare for its (re-)execution. This means that the edge is
-        necessarily of the shape A_1 -[WR(A_1)]-> A_2. While a rewound action is waiting for its
-        own write lock in enterActionPreparation, it doesn't hold any locks since
-        enterActionExecution hasn't been called yet in SkyframeActionExecutor and all past
-        executions of the action have released all their locks due to use of try-with-resources.
-        This means that A_1 can't have any incoming edges in the wait-for graph, which is a
-        contradiction to the assumption that it is contained in the directed cycle C.
-
-   We conclude that XY = RW. Since the write lock of A_3 is only ever acquired by A_3 itself, all
-   edges in C are of the form A_1 -[RW(A_2)]-> A_2. But by construction of inputKeysFor, the
-   action A_1 is attempting to acquire the read locks of all its inputs' generating actions, and
-   thus the action A_1 depends on one of the outputs of A_2 (*).
+  * WW and WR: In both cases, A_1 attempts to acquire a write lock, which only happens when A_1 is
+        a rewound action about to prepare for its (re-)execution. While a rewound action is waiting
+        for a write lock in enterActionPreparation, it doesn't hold any locks: enterActionExecution
+        hasn't been called yet in SkyframeActionExecutor, it only ever acquires the single write
+        lock it is waiting for, and all past executions of the action have released all their locks
+        due to use of try-with-resources. This means that A_1 can't have any incoming edges in the
+        wait-for graph, which is a contradiction to the assumption that it is contained in the
+        directed cycle C.
+
+   We conclude that XY = RW, so all edges in C are of the form A_1 -[RW(K)]-> A_2 with A_2 covering
+   K. Since every node of C also has an incoming edge, every node of C holds a write lock and thus
+   covers the key of that lock.
+
+   By construction of inputKeysFor, A_1 is waiting for R(K) because it has an input guarded by K,
+   which is either an output of the action identified by K, or a file in a tree artifact declared
+   by the ActionTemplate identified by K. In the latter case, if the input is an individual file
+   rather than the tree artifact itself, then A_1 is an expanded action of that template and thus
+   covers K - but A_1 covers exactly one key, namely the one of the write lock it holds, which A_2
+   holds instead. So A_1 depends on the tree artifact in its entirety and thus on all actions
+   covering K, in particular on A_2 (*).
 
    Applied to all edges of C, we conclude that there is a corresponding directed cycle in the
    action graph, which is a contradiction since Bazel disallows dependency cycles.
@@ -127,6 +144,12 @@ thus the action A_1 depends on one of the outputs of A_2 (*).
    * The proof would not go through at (*) if fineLocks were replaced by a Striped lock structure
      with a fixed number of locks. In fact, this gives rise to a deadlock if the number of stripes
      is at least 2, but low enough that distinct generating actions hash to the same stripe.
+   * It is crucial that an action only ever acquires a single write lock: a rewound action holding
+     one write lock while waiting for another could deadlock with a reader acquiring the same two
+     locks in the opposite order, and readers acquire their locks in an arbitrary order.
+   * A rewound action must skip the read lock of the key guarding its own outputs, which it already
+     holds the write lock of: the locks aren't reentrant, so an expanded action consuming the
+     outputs of another action from the same expansion would otherwise deadlock with itself.
    */
 
   @Override
@@ -149,7 +172,10 @@ private SilentCloseable enterActionPreparationForRewinding(Action action)
     if (localCoarseLock != null) {
       // This is the first time a rewound action has attempted to prepare for its execution.
       // Switch to using the fine locks under the protection of the coarse write lock.
-      localCoarseLock.writeLock().lockInterruptibly();
+      try (SilentCloseable c =
+          Profiler.instance().profile(ProfilerTask.ACTION_LOCK, "action.prepareFirstRewinding")) {
+        localCoarseLock.writeLock().lockInterruptibly();
+      }
       try {
         // Check again under the lock to avoid a race between multiple rewound actions attempting
         // to prepare for execution at the same time.
@@ -166,6 +192,8 @@ private SilentCloseable enterActionPreparationForRewinding(Action action)
                   // (https://github.com/openjdk/jdk/blob/b349f661ea5f14b258191134714a7e712c90ef3e/src/java.base/share/classes/java/util/concurrent/locks/StampedLock.java#L1039),
                   // TODO: Investigate the effect of fair locks on build wall time.
                   .build((ActionLookupData unused) -> new StampedLock().asReadWriteLock());
+          // Must be assigned after fineLocks as lockArtifactsForConsumption relies on a null
+          // coarseLock implying a non-null fineLocks.
           coarseLock = null;
         }
       } finally {
@@ -174,8 +202,18 @@ private SilentCloseable enterActionPreparationForRewinding(Action action)
     }
 
     var writeLock = fineLocks.get(outputKeyFor(action)).writeLock();
-    writeLock.lockInterruptibly();
-    prepareOutputsForRewinding(action);
+    try (SilentCloseable c =
+        Profiler.instance()
+            .profile(ProfilerTask.ACTION_LOCK, "action.awaitRewoundActionConsumers")) {
+      writeLock.lockInterruptibly();
+    }
+    try (SilentCloseable c =
+        Profiler.instance().profile(ProfilerTask.INFO, "action.prepareOutputsForRewinding")) {
+      prepareOutputsForRewinding(action);
+    } catch (Throwable t) {
+      writeLock.unlock();
+      throw t;
+    }
     return writeLock::unlock;
   }
 
@@ -184,20 +222,28 @@ private SilentCloseable enterActionPreparationForRewinding(Action action)
    * their prefetching state.
    */
   private void prepareOutputsForRewinding(Action action) throws InterruptedException {
-    Cancellable task = outputUploadTasks.remove(action);
-    if (task != null) {
-      task.cancel();
+    ImmutableList<Cancellable> tasks = outputUploadTasks.remove(actionKeyFor(action));
+    if (tasks != null) {
+      for (Cancellable task : tasks) {
+        task.cancel();
+      }
     }
     actionInputFetcher.handleRewoundActionOutputs(action.getOutputs());
   }
 
   @Override
-  public SilentCloseable enterActionExecution(Action action, InputMetadataProvider metadataProvider)
+  public SilentCloseable enterActionExecution(
+      Action action, boolean wasRewound, InputMetadataProvider metadataProvider)
       throws InterruptedException {
     try (SilentCloseable c =
         Profiler.instance().profile(ProfilerTask.ACTION_LOCK, "action.enterActionExecution")) {
       return lockArtifactsForConsumption(
-          () -> action.getInputs().toList().iterator(), metadataProvider);
+          action.getInputs().toList(),
+          metadataProvider,
+          // A rewound action already holds the write lock on the key guarding its outputs and the
+          // locks aren't reentrant. Actions generated by an ActionTemplate can consume the outputs
+          // of other actions from the same expansion, which are guarded by the same key.
+          wasRewound ? outputKeyFor(action) : null);
     }
   }
 
@@ -211,29 +257,50 @@ public SilentCloseable enterProcessOutputsAndGetLostArtifacts(
     try (SilentCloseable c =
         Profiler.instance()
             .profile(ProfilerTask.ACTION_LOCK, "action.enterProcessOutputsAndGetLostArtifacts")) {
-      return lockArtifactsForConsumption(importantOutputs, fullMetadataProvider);
+      return lockArtifactsForConsumption(
+          importantOutputs, fullMetadataProvider, /* writeLockedKey= */ null);
     }
   }
 
   /**
    * Registers a cancellation callback for an upload of action outputs that may still be running
    * after the action has completed.
+   *
+   * <p>Must be paired with a call to {@link #unregisterOutputUploadTask} once the upload has
+   * completed so that the task doesn't remain registered (and thus retained) for the rest of the
+   * build.
    */
   public void registerOutputUploadTask(ActionExecutionMetadata action, Cancellable task) {
-    // We don't expect to have multiple output upload tasks for the same action registered at the
-    // same time.
-    outputUploadTasks.merge(
-        action,
-        task,
-        (oldTask, newTask) -> {
-          throw new IllegalStateException(
-              "Attempted to register multiple output upload tasks for %s: %s and %s"
-                  .formatted(action, oldTask, newTask));
+    // The task is added to the list inside the mapping function so that registration is atomic
+    // with respect to the removal of the entry in prepareOutputsForRewinding.
+    outputUploadTasks.compute(
+        actionKeyFor(action),
+        (unusedKey, tasks) ->
+            tasks == null
+                ? ImmutableList.of(task)
+                : ImmutableList.<Cancellable>builderWithExpectedSize(tasks.size() + 1)
+                    .addAll(tasks)
+                    .add(task)
+                    .build());
+  }
+
+  /** Unregisters a task previously registered via {@link #registerOutputUploadTask}. */
+  public void unregisterOutputUploadTask(ActionExecutionMetadata action, Cancellable task) {
+    outputUploadTasks.computeIfPresent(
+        actionKeyFor(action),
+        (unusedKey, tasks) -> {
+          // Identity comparison: a task is only ever registered once, and a task registered by a
+          // re-execution of the action must not be unregistered by its predecessor.
+          var remainingTasks =
+              tasks.stream().filter(t -> t != task).collect(ImmutableList.toImmutableList());
+          return remainingTasks.isEmpty() ? null : remainingTasks;
         });
   }
 
   private SilentCloseable lockArtifactsForConsumption(
-      Iterable<Artifact> artifacts, InputMetadataProvider metadataProvider)
+      Iterable<Artifact> artifacts,
+      InputMetadataProvider metadataProvider,
+      @Nullable ActionLookupData writeLockedKey)
       throws InterruptedException {
     var localCoarseLock = coarseLock;
     if (localCoarseLock != null) {
@@ -256,7 +323,7 @@ private SilentCloseable lockArtifactsForConsumption(
       localCoarseLock.readLock().unlock();
     }
     var allReadWriteLocks =
-        localFineLocks.getAll(inputKeysFor(artifacts, metadataProvider)).values();
+        localFineLocks.getAll(inputKeysFor(artifacts, metadataProvider, writeLockedKey)).values();
     var locksToUnlockBuilder =
         ImmutableList.<Lock>builderWithExpectedSize(allReadWriteLocks.size());
     try {
@@ -265,7 +332,7 @@ private SilentCloseable lockArtifactsForConsumption(
         readLock.lockInterruptibly();
         locksToUnlockBuilder.add(readLock);
       }
-    } catch (InterruptedException e) {
+    } catch (Throwable e) {
       for (var readLock : locksToUnlockBuilder.build()) {
         readLock.unlock();
       }
@@ -276,7 +343,9 @@ private SilentCloseable lockArtifactsForConsumption(
   }
 
   private static Iterable<ActionLookupData> inputKeysFor(
-      Iterable<Artifact> artifacts, InputMetadataProvider metadataProvider) {
+      Iterable<Artifact> artifacts,
+      InputMetadataProvider metadataProvider,
+      @Nullable ActionLookupData writeLockedKey) {
     var allArtifacts =
         Iterables.concat(
             artifacts,
@@ -284,12 +353,44 @@ private static Iterable<ActionLookupData> inputKeysFor(
                 Iterables.transform(
                     metadataProvider.getRunfilesTrees(),
                     runfilesTree -> runfilesTree.getArtifacts().toList())));
-    return Iterables.transform(
-        Iterables.filter(allArtifacts, artifact -> artifact instanceof DerivedArtifact),
-        artifact -> ((DerivedArtifact) artifact).getGeneratingActionKey());
+    var result =
+        Iterables.transform(
+            Iterables.filter(allArtifacts, artifact -> artifact instanceof DerivedArtifact),
+            artifact -> lockKeyFor((DerivedArtifact) artifact));
+    if (writeLockedKey == null) {
+      return result;
+    }
+    return Iterables.filter(result, key -> !key.equals(writeLockedKey));
   }
 
-  private static ActionLookupData outputKeyFor(Action action) {
+  /** Returns the key that uniquely identifies the given action. */
+  private static ActionLookupData actionKeyFor(ActionExecutionMetadata action) {
     return ((DerivedArtifact) action.getPrimaryOutput()).getGeneratingActionKey();
   }
+
+  /**
+   * Returns the key of the lock that guards the given artifact, which is the generating action key
+   * of the outermost tree artifact containing it, or its own if it isn't contained in one.
+   *
+   * <p>This is the artifact's own generating action key except for the outputs of an {@link
+   * com.google.devtools.build.lib.actions.ActionTemplate} expansion, which are guarded by the key
+   * of the template: they are only ever consumed as part of a tree artifact the template declares,
+   * either by actions outside the expansion, which depend on that tree artifact, or by other
+   * actions of the same expansion, which depend on individual files in it.
+   */
+  private static ActionLookupData lockKeyFor(DerivedArtifact artifact) {
+    var outermost = artifact;
+    for (var parent = artifact.getParent(); parent != null; parent = parent.getParent()) {
+      outermost = parent;
+    }
+    return outermost.getGeneratingActionKey();
+  }
+
+  /**
+   * Returns the key of the lock that guards the outputs of the given action, which is the key its
+   * consumers acquire the read lock of.
+   */
+  private static ActionLookupData outputKeyFor(Action action) {
+    return lockKeyFor((DerivedArtifact) action.getPrimaryOutput());
+  }
 }
diff --git a/src/main/java/com/google/devtools/build/lib/skyframe/SkyframeActionExecutor.java b/src/main/java/com/google/devtools/build/lib/skyframe/SkyframeActionExecutor.java
index 956f51e6b9c3cd..ddc2b97d3b42ac 100644
--- a/src/main/java/com/google/devtools/build/lib/skyframe/SkyframeActionExecutor.java
+++ b/src/main/java/com/google/devtools/build/lib/skyframe/SkyframeActionExecutor.java
@@ -1141,8 +1141,9 @@ public ActionStepOrResult run(Environment env)
           }
           env.getListener().post(event);
           var rewoundActionSynchronizer = outputService.getRewoundActionSynchronizer();
+          boolean wasRewound = wasRewound(action);
           try (SilentCloseable outerLock =
-              rewoundActionSynchronizer.enterActionPreparation(action, wasRewound(action))) {
+              rewoundActionSynchronizer.enterActionPreparation(action, wasRewound)) {
             if (actionFileSystemType().shouldDoEagerActionPrep()) {
               try (SilentCloseable d =
                   Profiler.instance().profile(ProfilerTask.INFO, "action.prepare")) {
@@ -1177,7 +1178,7 @@ public ActionStepOrResult run(Environment env)
 
             try (SilentCloseable innerLock =
                 rewoundActionSynchronizer.enterActionExecution(
-                    action, actionExecutionContext.getInputMetadataProvider())) {
+                    action, wasRewound, actionExecutionContext.getInputMetadataProvider())) {
               return executeAction(env.getListener(), action);
             }
           }
diff --git a/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java b/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java
index 6b034956d90076..4b93b730e75e09 100644
--- a/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java
+++ b/src/main/java/com/google/devtools/build/lib/vfs/OutputService.java
@@ -291,7 +291,8 @@ SilentCloseable enterActionPreparation(Action action, boolean wasRewound)
         throws InterruptedException;
 
     /** Guards an action from the beginning to the end of its {@link Action#execute execution}. */
-    SilentCloseable enterActionExecution(Action action, InputMetadataProvider metadataProvider)
+    SilentCloseable enterActionExecution(
+        Action action, boolean wasRewound, InputMetadataProvider metadataProvider)
         throws InterruptedException;
 
     /**
@@ -307,7 +308,7 @@ public SilentCloseable enterActionPreparation(Action action, boolean wasRewound)
 
           @Override
           public SilentCloseable enterActionExecution(
-              Action action, InputMetadataProvider metadataProvider) {
+              Action action, boolean wasRewound, InputMetadataProvider metadataProvider) {
             return () -> {};
           }
         };
diff --git a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/BUILD b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/BUILD
index a96f75249d4487..7296b03461aa2f 100644
--- a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/BUILD
+++ b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/BUILD
@@ -38,6 +38,7 @@ java_library(
         "//src/main/java/com/google/devtools/build/lib/skyframe:aspect_key_creator",
         "//src/main/java/com/google/devtools/build/lib/skyframe/rewinding",
         "//src/main/java/com/google/devtools/build/lib/skyframe/rewinding:lost_important_output_handler_module",
+        "//src/main/java/com/google/devtools/build/lib/util:os",
         "//src/main/java/com/google/devtools/build/lib/vfs",
         "//src/main/java/com/google/devtools/build/lib/vfs:pathfragment",
         "//src/main/java/com/google/devtools/build/skyframe",
diff --git a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTest.java b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTest.java
index dc12784dafd741..7467ca06ce1218 100644
--- a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTest.java
+++ b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTest.java
@@ -266,6 +266,11 @@ public void treeArtifactRewound_oneFileLost() throws Exception {
     helper.runTreeArtifactRewound_oneFileLost_spawnFailed();
   }
 
+  @Test
+  public void actionTemplateExpansionRewound_notConcurrentWithTreeConsumers() throws Exception {
+    helper.runActionTemplateExpansionRewound_notConcurrentWithTreeConsumers();
+  }
+
   @Test
   public void generatedRunfilesRewound_allFilesLost() throws Exception {
     helper.runGeneratedRunfilesRewound_allFilesLost_spawnFailed();
diff --git a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTestsHelper.java b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTestsHelper.java
index 142d1bcfd17834..fb7be69562a599 100644
--- a/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTestsHelper.java
+++ b/src/test/java/com/google/devtools/build/lib/skyframe/rewinding/RewindingTestsHelper.java
@@ -23,6 +23,7 @@
 import static com.google.devtools.build.lib.vfs.FileSystemUtils.readContentAsLatin1;
 import static com.google.devtools.build.lib.vfs.FileSystemUtils.writeContent;
 import static java.util.Arrays.stream;
+import static java.util.stream.Collectors.joining;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
@@ -33,6 +34,7 @@
 import com.google.common.collect.ImmutableMultiset;
 import com.google.common.collect.ImmutableSetMultimap;
 import com.google.common.collect.Iterables;
+import com.google.common.eventbus.AllowConcurrentEvents;
 import com.google.common.eventbus.Subscribe;
 import com.google.common.flogger.GoogleLogger;
 import com.google.common.util.concurrent.Uninterruptibles;
@@ -49,6 +51,7 @@
 import com.google.devtools.build.lib.actions.FilesetOutputSymlink;
 import com.google.devtools.build.lib.actions.LostInputsExecException;
 import com.google.devtools.build.lib.actions.Spawn;
+import com.google.devtools.build.lib.actions.SpawnExecutedEvent;
 import com.google.devtools.build.lib.actions.SpawnResult;
 import com.google.devtools.build.lib.analysis.AspectCompleteEvent;
 import com.google.devtools.build.lib.analysis.TargetCompleteEvent;
@@ -79,6 +82,7 @@
 import com.google.devtools.build.lib.testutil.SpawnController.SpawnShim;
 import com.google.devtools.build.lib.testutil.SpawnInputUtils;
 import com.google.devtools.build.lib.testutil.TestConstants;
+import com.google.devtools.build.lib.util.OS;
 import com.google.devtools.build.lib.vfs.Path;
 import com.google.devtools.build.lib.vfs.PathFragment;
 import com.google.devtools.build.skyframe.NodeEntry.DirtyType;
@@ -1664,6 +1668,259 @@ final void runTreeArtifactRewoundWhenTreeFilesLost(
     assertArtifactKey(rewoundKeys.get(2), "tree/_pic_objs/consumes_tree/make_cc_dir");
   }
 
+  /**
+   * The number of actions that consume the tree artifact populated by an action template expansion
+   * concurrently in {@link #runActionTemplateExpansionRewound_notConcurrentWithTreeConsumers}.
+   *
+   * <p>The more of them are reading it when the expansion is rewound, the more likely it is that a
+   * missing mutual exclusion is detected.
+   */
+  private static final int TREE_CONSUMER_COUNT = 8;
+
+  /**
+   * A tool that copies its second argument to its first.
+   *
+   * <p>On Windows, {@code cmd.exe} treats forward slashes as option prefixes, so the paths have to
+   * be translated before they are passed to {@code copy}.
+   */
+  private static final String COPY_TOOL_SCRIPT =
+      OS.getCurrent() == OS.WINDOWS
+          ? """
+            @echo off
+            set "OUT=%~1"
+            set "IN=%~2"
+            copy /Y "%IN:/=\\%" "%OUT:/=\\%" >NUL
+            """
+          : """
+            #!/bin/bash
+            cp "$2" "$1"
+            """;
+
+  /**
+   * Verifies that an action generated by an {@link
+   * com.google.devtools.build.lib.actions.ActionTemplate} does not prepare for its re-execution
+   * while a consumer of the tree artifact it populates is running.
+   *
+   * <p>The outputs of such an action are {@link
+   * com.google.devtools.build.lib.actions.Artifact.TreeFileArtifact}s owned by the expansion,
+   * whereas its consumers depend on the parent tree artifact, which is owned by the template. A
+   * rewound expansion action must therefore synchronize on the template rather than on itself.
+   *
+   * <p>The rewound action also consumes the output of another action of the same expansion, which
+   * is a special case that only happens with expansions and must not result in deadlocks.
+   */
+  public final void runActionTemplateExpansionRewound_notConcurrentWithTreeConsumers() throws Exception {
+    // All consumers and the action that reports the lost input have to run concurrently for the
+    // expansion to be rewound while the tree artifact is being read.
+    ensureMinimumJobs(TREE_CONSUMER_COUNT + 1);
+    testCase.addOptions("--experimental_allow_map_directory");
+    testCase.write(
+        "foo/defs.bzl",
+        """
+        def _copy_tool_impl(ctx):
+            tool = ctx.actions.declare_file(ctx.attr.name + ".bat")
+            ctx.actions.write(tool, r\"\"\"COPY_TOOL_SCRIPT\"\"\", is_executable = True)
+            return DefaultInfo(files = depset([tool]), executable = tool)
+
+        copy_tool = rule(implementation = _copy_tool_impl, executable = True)
+
+        def _map_impl(template_ctx, input_directories, output_directories, tools, **kwargs):
+            for child in input_directories["seed"].children:
+                # The two actions form a chain within the expansion, so the second one consumes a
+                # file of the very tree artifact it populates.
+                mid = template_ctx.declare_file(
+                    child.basename + ".mid",
+                    directory = output_directories["mapped"],
+                )
+                args = template_ctx.args()
+                args.add_all([mid, child])
+                template_ctx.run(
+                    inputs = [child],
+                    outputs = [mid],
+                    executable = tools["copy_tool"],
+                    arguments = [args],
+                    progress_message = "Mapping foo/mapped_dir (1)",
+                )
+                out = template_ctx.declare_file(
+                    child.basename + ".out",
+                    directory = output_directories["mapped"],
+                )
+                args = template_ctx.args()
+                args.add_all([out, mid])
+                template_ctx.run(
+                    inputs = [mid],
+                    outputs = [out],
+                    executable = tools["copy_tool"],
+                    arguments = [args],
+                    progress_message = "Mapping foo/mapped_dir (2)",
+                )
+
+        def _mapped_tree_impl(ctx):
+            seed = ctx.actions.declare_directory("seed_dir")
+            ctx.actions.run_shell(
+                outputs = [seed],
+                command = "echo seed > $1/f1",
+                arguments = [seed.path],
+                progress_message = "Seeding foo/seed_dir",
+            )
+            mapped = ctx.actions.declare_directory("mapped_dir")
+            ctx.actions.map_directory(
+                implementation = _map_impl,
+                input_directories = {"seed": seed},
+                output_directories = {"mapped": mapped},
+                tools = {"copy_tool": ctx.attr._copy_tool.files_to_run},
+                # Ensure that the rewound expansion action re-executes its spawn instead of
+                # picking up the result of its first execution from the cache.
+                execution_requirements = {"no-cache": "1"},
+            )
+            return DefaultInfo(files = depset([mapped]))
+
+        mapped_tree = rule(
+            implementation = _mapped_tree_impl,
+            attrs = {
+                "_copy_tool": attr.label(
+                    default = ":copy_tool",
+                    executable = True,
+                    cfg = "exec",
+                ),
+            },
+        )
+
+        def _consumer_impl(ctx):
+            out = ctx.actions.declare_file(ctx.attr.name + ".out")
+            ctx.actions.run_shell(
+                inputs = ctx.files.srcs,
+                outputs = [out],
+                command = "echo consumed > $1",
+                arguments = [out.path],
+                progress_message = "Consuming //foo:" + ctx.attr.name,
+            )
+            return DefaultInfo(files = depset([out]))
+
+        consumer = rule(
+            implementation = _consumer_impl,
+            attrs = {"srcs": attr.label_list(allow_files = True)},
+        )
+        """
+            .replace("COPY_TOOL_SCRIPT", COPY_TOOL_SCRIPT));
+    testCase.write(
+        "foo/BUILD",
+        """
+        load(":defs.bzl", "consumer", "copy_tool", "mapped_tree")
+
+        copy_tool(name = "copy_tool")
+
+        mapped_tree(name = "mapped_tree")
+
+        genrule(
+            name = "warmup_gen",
+            outs = ["warmup.out"],
+            cmd = "echo warmup > $@",
+            tags = ["no-cache"],
+        )
+
+        genrule(
+            name = "warmup_consumer",
+            srcs = ["warmup.out"],
+            outs = ["warmup_consumed.out"],
+            cmd = "cp $< $@",
+        )
+
+        consumer(
+            name = "losing_consumer",
+            srcs = [
+                "warmup_consumed.out",
+                ":mapped_tree",
+            ],
+        )
+        """
+            + IntStream.range(0, TREE_CONSUMER_COUNT)
+                .mapToObj(
+                    i ->
+                        """
+                        consumer(
+                            name = "consumer_%d",
+                            srcs = [
+                                "warmup_consumed.out",
+                                ":mapped_tree",
+                            ],
+                        )
+                        """
+                            .formatted(i))
+                .collect(joining("\n")));
+
+    // The first rewound action of a build waits for all in-flight actions to finish before it
+    // prepares for its re-execution, which would mask the behavior under test. Rewind an unrelated
+    // action first; all consumers depend on its output and thus can't start any earlier.
+    addSpawnShim(
+        "Executing genrule //foo:warmup_consumer",
+        (spawn, context) -> createLostInputsExecException(spawn, context, "warmup.out"));
+
+    // A consumer holds the read lock on the tree artifact from before its spawn shim runs until
+    // after its spawn has been executed, so the count is a lower bound on the number of consumers
+    // that are reading the tree artifact.
+    AtomicInteger consumersReadingTree = new AtomicInteger();
+    AtomicInteger maxConsumersReadingDuringExpansion = new AtomicInteger();
+    testCase.getRuntimeWrapper()
+        .registerSubscriber(
+            new Object() {
+              @Subscribe
+              @AllowConcurrentEvents
+              @SuppressWarnings("unused")
+              public void accept(SpawnExecutedEvent event) {
+                if (event.getActionMetadata().describe().startsWith("Consuming //foo:consumer_")) {
+                  consumersReadingTree.decrementAndGet();
+                }
+              }
+            });
+    for (int i = 0; i < TREE_CONSUMER_COUNT; i++) {
+      addSpawnShim(
+          "Consuming //foo:consumer_" + i,
+          (spawn, context) -> {
+            consumersReadingTree.incrementAndGet();
+            return ExecResult.delegate();
+          });
+    }
+    // The expansion actions are the only writers, so they must never observe a reader. Both of them
+    // run twice: once initially and once after the expansion has been rewound.
+    for (String expansionAction :
+        ImmutableList.of("Mapping foo/mapped_dir (1)", "Mapping foo/mapped_dir (2)")) {
+      for (int run = 0; run < 2; run++) {
+        addSpawnShim(
+            expansionAction,
+            (spawn, context) -> {
+              maxConsumersReadingDuringExpansion.accumulateAndGet(
+                  consumersReadingTree.get(), Math::max);
+              return ExecResult.delegate();
+            });
+      }
+    }
+    // Fails without executing a spawn, so that the expansion is rewound while the other consumers
+    // are as likely as possible to still be reading the tree artifact. The consumers deliberately
+    // don't wait for each other: making them do so would deadlock whenever Skyframe can't run all
+    // of them concurrently.
+    addSpawnShim(
+        "Consuming //foo:losing_consumer",
+        (spawn, context) -> {
+          SpecialArtifact mappedTree = SpawnInputUtils.getTreeArtifactWithName(spawn, "mapped_dir");
+          return createLostInputsExecException(
+              context, SpawnInputUtils.getExpandedToArtifact("f1.out", mappedTree, spawn, context));
+        });
+
+    testCase.buildTarget("//foo:all");
+
+    verifyAllSpawnShimsConsumed();
+    assertWithMessage(
+            "the rewound action template expansion prepared for its re-execution while consumers"
+                + " were still reading the tree artifact it populates")
+        .that(maxConsumersReadingDuringExpansion.get())
+        .isEqualTo(0);
+    // Rewinding an expanded action re-expands the template, so both actions of the chain re-run.
+    var executedSpawns = ImmutableMultiset.copyOf(getExecutedSpawnDescriptions());
+    assertThat(executedSpawns).hasCount("Mapping foo/mapped_dir (1)", 2);
+    assertThat(executedSpawns).hasCount("Mapping foo/mapped_dir (2)", 2);
+  }
+
   public final void runGeneratedRunfilesRewound_allFilesLost_spawnFailed() throws Exception {
     // This test demonstrates that rewinding works when an action fails due to lost inputs which are
     // generated files in the action's runfiles. Rewinding must propagate across the runfiles tree
@@ -3316,10 +3573,20 @@ static boolean isActionExecutionKey(Object key, Label label) {
    * CPU.
    */
   private void ensureMultipleJobs() throws Exception {
+    ensureMinimumJobs(2);
+  }
+
+  /**
+   * Ensures that the value of the {@code --jobs} flag is at least {@code minJobs}.
+   *
+   * <p>Note that the default value for {@code --jobs} is automatically calculated based on host
+   * CPU.
+   */
+  private void ensureMinimumJobs(int minJobs) throws Exception {
     int autoJobs = new JobsConverter().convert("auto");
-    if (autoJobs == 1) {
-      logger.atInfo().log("Setting --jobs=2 (was 1)");
-      testCase.addOptions("--jobs=2");
+    if (autoJobs < minJobs) {
+      logger.atInfo().log("Setting --jobs=%s (was %s)", minJobs, autoJobs);
+      testCase.addOptions("--jobs=" + minJobs);
     } else {
       logger.atInfo().log("Keeping default value of --jobs=%s", autoJobs);
     }