Add Bazel support for --rewind_lost_inputs (bazelbuild#25477)

As of bazelbuild#25396, action rewinding (controlled by `--rewind_lost_inputs`) and build rewinding (controlled by `--experimental_remote_cache_eviction_retries`) are equally effective at recovering lost inputs.
However, action rewinding in Bazel is prone to races, which renders it unusable in practice - in fact, there are races even if `--jobs=1`, as discovered in bazelbuild#25412. It does have a number of benefits compared to build rewinding, which makes it worth fixing these issues:
* When a lost input is detected, the progress of actions running concurrently isn't lost.
* Build rewinding can start a large number of invocations with their own build lifecycle, which greatly complicates build observability.
* Finding a good value for the allowed number of build retries is difficult since a single input may be lost multiple times and rewinding can discover additional lost inputs, but the at the same time builds that ultimately fail shouldn't be retried indefinitely.
* Build rewinding drops all action cache entries that mention remote files when it encounters a lost input, which can compound remote cache issues.

This PR adds Bazel support for `--rewind_lost_inputs` with arbitrary `--jobs` values by synchronizing action preparation, execution and post-processing in the presence of rewound actions. This is necessary with Bazel's remote filesystem since it is backed by the local filesystem and needs to support local execution of actions, whereas Blaze uses a content-addressed filesystem that can be updated atomically.

Synchronization is achieved by adding try-with-resources scopes backed by a new `RewoundActionSynchronizer` interface to `SkyframeActionExecutor` that wrap action preparation (which primarily deletes action outputs) and action execution, thus preventing a rewound action from deleting its outputs while downstream actions read them concurrently. Additional synchronization is required to handle async remote cache uploads (`--remote_cache_async`).

The synchronization scheme relies on a single `ReadWriteLock` that is only ever locked for reading until the first time an action is rewound, which ensures that performance doesn't regress for the common case of builds without lost inputs. Upon the first time an action is rewound, the single lock is inflated to a concurrent map of locks that permits concurrency between actions as long as dependency relations between rewound and non-rewound actions are honored (i.e., an action consuming a non-lost input of a rewound action can't execute concurrently with that action's preparation and execution). See the comment in `RemoteRewoundActionSynchronizer` for details as well as a proof that this scheme is free of deadlocks.
________

Subsumes the previously reviewed bazelbuild#25412, which couldn't be merged due to the lack of synchronization.

Tested for races manually by running the following command (also with `ActionRewindStrategy.MAX_ACTION_REWIND_EVENTS = 10`):
```
bazel test //src/test/java/com/google/devtools/build/lib/skyframe/rewinding:RewindingTest --test_filter=com.google.devtools.build.lib.skyframe.rewinding.RewindingTest#multipleLostInputsForRewindPlan --runs_per_test=1000 --runs_per_test_detects_flakes --test_sharding_strategy=disabled
```

Fixes bazelbuild#26657

RELNOTES: Bazel now has experimental support for --rewind_lost_inputs, which can rerun actions within a single build to recover from (remote or disk) cache evictions.

Closes bazelbuild#25477.

PiperOrigin-RevId: 882050264
Change-Id: I79b7d22bdb83224088a34be62c492a966e9be132
(cherry picked from commit 464eacb)

Author: fmeum

src/main/java/com/google/devtools/build/lib/remote/AbstractActionInputPrefetcher.java

@@ -78,7 +78,7 @@ public abstract class AbstractActionInputPrefetcher implements ActionInputPrefet
   protected final Path execRoot;
   protected final RemoteOutputChecker remoteOutputChecker;
 
-  @Nullable private final ActionOutputDirectoryHelper outputDirectoryHelper;
+  @Nullable protected final ActionOutputDirectoryHelper outputDirectoryHelper;
 
   /** The state of a directory tracked by {@link DirectoryTracker}, as explained below. */
   enum DirectoryState {
@@ -140,8 +140,9 @@ private void setWritable(Path dir, DirectoryState newState) throws IOException {
       directoryStateMap.compute(
           dir,
           (unusedKey, oldState) -> {
-            if (oldState == DirectoryState.TEMPORARILY_WRITABLE
-                || oldState == DirectoryState.PERMANENTLY_WRITABLE) {
+            if (!forceRefetch(dir)
+                && (oldState == DirectoryState.TEMPORARILY_WRITABLE
+                    || oldState == DirectoryState.PERMANENTLY_WRITABLE)) {
               // Already writable, but must potentially upgrade from temporary to permanent.
               return newState == DirectoryState.PERMANENTLY_WRITABLE ? newState : oldState;
             }
@@ -177,8 +178,9 @@ void setOutputPermissions(Path dir) throws IOException {
       directoryStateMap.compute(
           dir,
           (unusedKey, oldState) -> {
-            if (oldState == DirectoryState.OUTPUT_PERMISSIONS
-                || oldState == DirectoryState.PERMANENTLY_WRITABLE) {
+            if (!forceRefetch(dir)
+                && (oldState == DirectoryState.OUTPUT_PERMISSIONS
+                    || oldState == DirectoryState.PERMANENTLY_WRITABLE)) {
               // Either the output permissions have already been set, or we're not changing the
               // permissions ever again.
               return oldState;
@@ -256,6 +258,12 @@ private static boolean shouldDownloadFile(Path path, FileArtifactValue metadata)
 
   protected abstract boolean canDownloadFile(Path path, FileArtifactValue metadata);
 
+  /**
+   * If true, then all previously acquired knowledge of the file system state of this path (e.g. the
+   * existence of tree artifact directories or previously downloaded files) must be discarded.
+   */
+  protected abstract boolean forceRefetch(Path path);
+
   /**
    * Downloads file to the given path via its metadata.
    *
@@ -605,15 +613,16 @@ private Completable downloadFileNoCheckRx(
                           alreadyDeleted.set(true);
                         }));
 
-    return downloadCache.executeIfNot(
+    return downloadCache.execute(
         finalPath,
         Completable.defer(
             () -> {
               if (shouldDownloadFile(finalPath, metadata)) {
                 return download;
               }
               return Completable.complete();
-            }));
+            }),
+        forceRefetch(finalPath));
   }
 
   private void finalizeDownload(
@@ -690,7 +699,7 @@ private static void deletePartialDownload(Path path) {
   }
 
   private Completable plantSymlink(Symlink symlink) {
-    return downloadCache.executeIfNot(
+    return downloadCache.execute(
         symlink.linkPath(),
         Completable.defer(
             () -> {
@@ -699,7 +708,8 @@ private Completable plantSymlink(Symlink symlink) {
               symlink.linkPath().delete();
               symlink.linkPath().createSymbolicLink(symlink.targetPath());
               return Completable.complete();
-            }));
+            }),
+        forceRefetch(symlink.linkPath));
   }
 
   public ImmutableSet<Path> downloadedFiles() {
@@ -736,7 +746,7 @@ public void finalizeAction(Action action, OutputMetadataStore outputMetadataStor
       }
 
       var metadata = outputMetadataStore.getOutputMetadata(output);
-      if (!metadata.isRemote()) {
+      if (!canDownloadFile(output.getPath(), metadata)) {
         continue;
       }
 

src/main/java/com/google/devtools/build/lib/remote/BUILD

@@ -212,6 +212,7 @@ java_library(
     name = "remote_output_checker",
     srcs = ["RemoteOutputChecker.java"],
     deps = [
+        ":concurrent_artifact_path_trie",
         "//src/main/java/com/google/devtools/build/lib/actions:artifacts",
         "//src/main/java/com/google/devtools/build/lib/actions:file_metadata",
         "//src/main/java/com/google/devtools/build/lib/analysis:analysis_cluster",
@@ -283,6 +284,32 @@ java_library(
     ],
 )
 
+java_library(
+    name = "concurrent_artifact_path_trie",
+    srcs = ["ConcurrentArtifactPathTrie.java"],
+    deps = [
+        "//src/main/java/com/google/devtools/build/lib/actions:artifacts",
+        "//src/main/java/com/google/devtools/build/lib/vfs:pathfragment",
+        "//third_party:guava",
+    ],
+)
+
+java_library(
+    name = "remote_rewound_action_synchronizer",
+    srcs = ["RemoteRewoundActionSynchronizer.java"],
+    deps = [
+        ":remote_action_input_fetcher",
+        "//src/main/java/com/google/devtools/build/lib/actions",
+        "//src/main/java/com/google/devtools/build/lib/actions:action_lookup_data",
+        "//src/main/java/com/google/devtools/build/lib/actions:artifacts",
+        "//src/main/java/com/google/devtools/build/lib/profiler",
+        "//src/main/java/com/google/devtools/build/lib/vfs:output_service",
+        "//third_party:caffeine",
+        "//third_party:guava",
+        "//third_party:jsr305",
+    ],
+)
+
 java_library(
     name = "store",
     srcs = ["Store.java"],
@@ -293,13 +320,16 @@ java_library(
     srcs = ["RemoteImportantOutputHandler.java"],
     deps = [
         ":remote_output_checker",
+        ":remote_rewound_action_synchronizer",
         "//src/main/java/com/google/devtools/build/lib/actions",
         "//src/main/java/com/google/devtools/build/lib/actions:artifacts",
         "//src/main/java/com/google/devtools/build/lib/actions:file_metadata",
         "//src/main/java/com/google/devtools/build/lib/actions:important_output_handler",
+        "//src/main/java/com/google/devtools/build/lib/profiler",
         "//src/main/java/com/google/devtools/build/lib/remote/common:bulk_transfer_exception",
         "//src/main/java/com/google/devtools/build/lib/remote/util",
         "//src/main/java/com/google/devtools/build/lib/vfs",
+        "//src/main/java/com/google/devtools/build/lib/vfs:output_service",
         "//src/main/java/com/google/devtools/build/lib/vfs:pathfragment",
         "//src/main/java/com/google/devtools/build/skyframe",
         "//src/main/protobuf:failure_details_java_proto",

src/main/java/com/google/devtools/build/lib/remote/ConcurrentArtifactPathTrie.java

@@ -0,0 +1,57 @@
+// Copyright 2026 The Bazel Authors. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package com.google.devtools.build.lib.remote;
+
+import com.google.common.base.Preconditions;
+import com.google.devtools.build.lib.actions.ActionInput;
+import com.google.devtools.build.lib.actions.Artifact;
+import com.google.devtools.build.lib.vfs.PathFragment;
+import java.util.concurrent.ConcurrentSkipListSet;
+
+/**
+ * A specialized concurrent trie that stores paths of artifacts and allows checking whether a given
+ * path is contained in (in the case of a tree artifact) or exactly matches (in any other case) an
+ * artifact in the trie.
+ */
+final class ConcurrentArtifactPathTrie {
+  // Invariant: no path in this set is a prefix of another path.
+  private final ConcurrentSkipListSet<PathFragment> paths =
+      new ConcurrentSkipListSet<>(PathFragment.HIERARCHICAL_COMPARATOR);
+
+  /**
+   * Adds the given {@link ActionInput} to the trie.
+   *
+   * <p>The caller must ensure that no object's path passed to this method is a prefix of any
+   * previously added object's path. Bazel enforces this for non-aggregate artifacts. Callers must
+   * not pass in {@link Artifact.TreeFileArtifact}s (which have exec paths that have their parent
+   * tree artifact's exec path as a prefix) or non-Artifact {@link ActionInput}s that violate this
+   * invariant.
+   */
+  void add(ActionInput input) {
+    Preconditions.checkArgument(
+        !(input instanceof Artifact.TreeFileArtifact),
+        "TreeFileArtifacts should not be added to the trie: %s",
+        input);
+    paths.add(input.getExecPath());
+  }
+
+  /** Checks whether the given {@link PathFragment} is contained in an artifact in the trie. */
+  boolean contains(PathFragment execPath) {
+    // By the invariant of this set, there is at most one prefix of execPath in the set. Since the
+    // comparator sorts all children of a path right after the path itself, if such a prefix
+    // exists, it must thus sort right before execPath (or be equal to it).
+    var floorPath = paths.floor(execPath);
+    return floorPath != null && execPath.startsWith(floorPath);
+  }
+}

src/main/java/com/google/devtools/build/lib/remote/RemoteActionInputFetcher.java

@@ -13,7 +13,6 @@
 // limitations under the License.
 package com.google.devtools.build.lib.remote;
 
-import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.util.concurrent.Futures.immediateFailedFuture;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 
@@ -25,6 +24,7 @@
 import com.google.devtools.build.lib.actions.ActionExecutionMetadata;
 import com.google.devtools.build.lib.actions.ActionInput;
 import com.google.devtools.build.lib.actions.ActionOutputDirectoryHelper;
+import com.google.devtools.build.lib.actions.Artifact;
 import com.google.devtools.build.lib.actions.FileArtifactValue;
 import com.google.devtools.build.lib.actions.cache.VirtualActionInput;
 import com.google.devtools.build.lib.events.Reporter;
@@ -35,7 +35,9 @@
 import com.google.devtools.build.lib.util.TempPathGenerator;
 import com.google.devtools.build.lib.vfs.OutputPermissions;
 import com.google.devtools.build.lib.vfs.Path;
+import com.google.devtools.build.lib.vfs.Symlinks;
 import java.io.IOException;
+import java.util.Collection;
 import javax.annotation.Nullable;
 
 /**
@@ -50,6 +52,7 @@ public class RemoteActionInputFetcher extends AbstractActionInputPrefetcher {
   private final String buildRequestId;
   private final String commandId;
   private final CombinedCache combinedCache;
+  private final ConcurrentArtifactPathTrie rewoundActionOutputs = new ConcurrentArtifactPathTrie();
 
   RemoteActionInputFetcher(
       Reporter reporter,
@@ -80,7 +83,18 @@ protected void prefetchVirtualActionInput(VirtualActionInput input) throws IOExc
 
   @Override
   protected boolean canDownloadFile(Path path, FileArtifactValue metadata) {
-    return metadata.isRemote();
+    // When action rewinding is enabled, an action that had remote metadata at some point during the
+    // build may have been re-executed locally to regenerate lost inputs, but may then be rewound
+    // again and thus have its (now local) outputs deleted. In this case, we need to download the
+    // outputs again, even if they are now considered local.
+    return metadata.isRemote() || (forceRefetch(path) && !path.exists(Symlinks.NOFOLLOW));
+  }
+
+  @Override
+  protected boolean forceRefetch(Path path) {
+    // Caches for download operations and output directory creation need to be disregarded for the
+    // outputs of rewound actions as they may have been deleted after they were first created.
+    return path.startsWith(execRoot) && rewoundActionOutputs.contains(path.relativeTo(execRoot));
   }
 
   @Override
@@ -93,7 +107,6 @@ protected ListenableFuture<Void> doDownloadFile(
       Priority priority,
       Reason reason)
       throws IOException {
-    checkArgument(metadata.isRemote(), "Cannot download file that is not a remote file.");
     RequestMetadata requestMetadata =
         TracingMetadataUtils.buildMetadata(
             buildRequestId,
@@ -140,4 +153,22 @@ protected ListenableFuture<Void> doDownloadFile(
                 }),
         directExecutor());
   }
+
+  public void handleRewoundActionOutputs(Collection<Artifact> outputs) {
+    // SkyframeActionExecutor#prepareForRewinding does *not* call this method because the
+    // RemoteActionFileSystem corresponds to an ActionFileSystemType with inMemoryFileSystem() ==
+    // true. While it is true that resetting outputDirectoryHelper isn't necessary to undo the
+    // caching of output directory creation during action preparation, we still need to reset here
+    // since outputDirectoryHelper is also used by AbstractActionInputPrefetcher.
+    outputDirectoryHelper.invalidateTreeArtifactDirectoryCreation(outputs);
+    for (Artifact output : outputs) {
+      // Action templates have TreeFileArtifacts as outputs, which isn't supported by the trie. We
+      // only need to track the tree artifacts themselves.
+      if (output instanceof Artifact.TreeFileArtifact) {
+        rewoundActionOutputs.add(output.getParent());
+      } else {
+        rewoundActionOutputs.add(output);
+      }
+    }
+  }
 }

src/main/java/com/google/devtools/build/lib/remote/RemoteExecutionService.java

@@ -64,6 +64,8 @@
 import com.google.common.eventbus.Subscribe;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.devtools.build.lib.actions.ActionInput;
 import com.google.devtools.build.lib.actions.Artifact;
@@ -151,8 +153,8 @@
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.concurrent.CancellationException;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Phaser;
 import java.util.concurrent.Semaphore;
@@ -192,9 +194,10 @@ public class RemoteExecutionService {
   private final Set<String> reportedErrors = new HashSet<>();
 
   @SuppressWarnings("AllowVirtualThreads")
-  private final ExecutorService backgroundTaskExecutor =
-      Executors.newThreadPerTaskExecutor(
-          Thread.ofVirtual().name("remote-execution-bg-", 0).factory());
+  private final ListeningExecutorService backgroundTaskExecutor =
+      MoreExecutors.listeningDecorator(
+          Executors.newThreadPerTaskExecutor(
+              Thread.ofVirtual().name("remote-execution-bg-", 0).factory()));
 
   private final AtomicBoolean shutdown = new AtomicBoolean(false);
   private final AtomicBoolean buildInterrupted = new AtomicBoolean(false);
@@ -1761,16 +1764,31 @@ public void uploadOutputs(
 
     if (remoteOptions.remoteCacheAsync
         && !action.getSpawn().getResourceOwner().mayModifySpawnOutputsAfterExecution()) {
-      backgroundTaskExecutor.execute(
-          () -> {
-            try {
-              doUploadOutputs(action, spawnResult, onUploadComplete);
-            } catch (ExecException e) {
-              reportUploadError(e);
-            } catch (InterruptedException ignored) {
-              // ThreadPerTaskExecutor does not care about interrupt status.
-            }
-          });
+      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();
+            });
+      }
     } else {
       doUploadOutputs(action, spawnResult, onUploadComplete);
     }