Ensure NamedSetOfFiles BEP event appears before event referencing the file_set.

Previously, when a `NestedSet<Artifact>` is referenced by multiple BuildEvents
(perhaps transitively) there is a race in `BuildEventStreamer`, which receives all
BuildEvents in an `@AllowConcurrentEvents` EventBus handler. If the BuildEvents are
posted by different threads, the first will acquire the name for the
`NamedSetOfFiles` and prepare the protobuf message from the NestedSet. In the
meantime, the second thread will see the NestedSet's name was already acquired and
may proceed to write its BuildEvent referencing the `named_set`. This leads to a
violation of the expected ordering.

Note that the ordering expectation here cannot be guaranteed by parent-child event
relationships, as a given `NamedSetOfFiles` cannot 'know' all of the events
that will reference it in advance. This is why `NamedSetOfFiles` events are children
of `Progress` events.

RELNOTES: Users consuming BEP may assume that a `named_set_of_files` event will
appear before any event referencing that `named_set` by ID. This allows consumers
to process the files for such events (eg. `TargetCompleted`) immediately.
PiperOrigin-RevId: 361822335
diff --git a/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java b/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java
index f645356..4c6cf98 100644
--- a/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java
+++ b/src/test/java/com/google/devtools/build/lib/runtime/BuildEventStreamerTest.java
@@ -69,6 +69,7 @@
 import com.google.devtools.build.lib.buildtool.buildevent.NoAnalyzeEvent;
 import com.google.devtools.build.lib.collect.nestedset.NestedSet;
 import com.google.devtools.build.lib.collect.nestedset.NestedSetBuilder;
+import com.google.devtools.build.lib.collect.nestedset.Order;
 import com.google.devtools.build.lib.server.FailureDetails.FailureDetail;
 import com.google.devtools.build.lib.server.FailureDetails.Spawn;
 import com.google.devtools.build.lib.server.FailureDetails.Spawn.Code;
@@ -86,6 +87,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -712,6 +714,110 @@
   }
 
   @Test
+  public void testArtifactSetsPrecedeReportingEvent() throws InterruptedException {
+    // Verify that reported artifacts appear as named_set_of_files before their ID is referenced by
+    // a reporting event.
+    BuildEvent startEvent =
+        new GenericBuildEvent(
+            testId("Initial"), ImmutableSet.of(ProgressEvent.INITIAL_PROGRESS_UPDATE));
+
+    // Prepare a dense NestedSet DAG with lots of shared references.
+    List<NestedSet<Artifact>> baseSets = new ArrayList<>();
+    baseSets.add(NestedSetBuilder.create(Order.STABLE_ORDER, makeArtifact("path/a")));
+    baseSets.add(NestedSetBuilder.create(Order.STABLE_ORDER, makeArtifact("path/b")));
+    baseSets.add(NestedSetBuilder.create(Order.STABLE_ORDER, makeArtifact("path/c")));
+    baseSets.add(NestedSetBuilder.create(Order.STABLE_ORDER, makeArtifact("path/d")));
+    List<NestedSet<Artifact>> depth2Sets = new ArrayList<>();
+    for (int i = 0; i < baseSets.size(); i++) {
+      depth2Sets.add(
+          NestedSetBuilder.<Artifact>stableOrder()
+              .addTransitive(baseSets.get(i))
+              .addTransitive(baseSets.get((i + 1) % baseSets.size()))
+              .build());
+    }
+    List<NestedSet<Artifact>> depth3Sets = new ArrayList<>();
+    for (int i = 0; i < depth2Sets.size(); i++) {
+      depth3Sets.add(
+          NestedSetBuilder.<Artifact>stableOrder()
+              .addTransitive(depth2Sets.get(i))
+              .addTransitive(depth2Sets.get((i + 1) % depth2Sets.size()))
+              .build());
+    }
+    List<NestedSet<Artifact>> depth4Sets = new ArrayList<>();
+    for (int i = 0; i < depth3Sets.size(); i++) {
+      depth4Sets.add(
+          NestedSetBuilder.<Artifact>stableOrder()
+              .addTransitive(depth3Sets.get(i))
+              .addTransitive(depth3Sets.get((i + 1) % depth3Sets.size()))
+              .build());
+    }
+    int numEvents = 20;
+    List<BuildEvent> eventsToPost = new ArrayList<>();
+    for (int i = 0; i < numEvents; i++) {
+      eventsToPost.add(
+          new GenericArtifactReportingEvent(
+              testId("reporting" + i), ImmutableSet.of(depth4Sets.get(i % depth4Sets.size()))));
+    }
+
+    streamer.buildEvent(startEvent);
+    // Publish `numEvents` different events that all report the same NamedSet of artifacts on
+    // `numEvents` different threads. Use latches to ensure:
+    //
+    // 1. all threads have started, before:
+    // 2. all threads send their event, before:
+    // 3. verifying the recorded events.
+    CountDownLatch readyToPublishLatch = new CountDownLatch(numEvents);
+    CountDownLatch startPublishingLatch = new CountDownLatch(1);
+    CountDownLatch donePublishingLatch = new CountDownLatch(numEvents);
+    for (int i = 0; i < numEvents; i++) {
+      int num = i;
+      new Thread(
+              () -> {
+                try {
+                  BuildEvent reportingArtifacts = eventsToPost.get(num);
+                  readyToPublishLatch.countDown();
+                  startPublishingLatch.await();
+                  streamer.buildEvent(reportingArtifacts);
+                } catch (InterruptedException e) {
+                  throw new RuntimeException(e);
+                }
+                donePublishingLatch.countDown();
+              })
+          .start();
+    }
+    readyToPublishLatch.await();
+    startPublishingLatch.countDown();
+    donePublishingLatch.await();
+
+    assertThat(streamer.isClosed()).isFalse();
+    List<BuildEvent> allEventsSeen = transport.getEvents();
+    List<BuildEventStreamProtos.BuildEvent> eventProtos = transport.getEventProtos();
+    // Each GenericArtifactReportingEvent and NamedArtifactGroup event has a corresponding Progress
+    // event posted immediately before.
+    assertThat(allEventsSeen)
+        .hasSize(1 + ((numEvents + baseSets.size() + depth2Sets.size() + depth3Sets.size()) * 2));
+    assertThat(allEventsSeen.get(0).getEventId()).isEqualTo(startEvent.getEventId());
+    // Verify that each named_set_of_files event is sent before all of the events that report that
+    // named_set.
+    Set<String> seenFileSets = new HashSet<>();
+    for (int i = 1; i < eventProtos.size(); i++) {
+      BuildEventStreamProtos.BuildEvent buildEvent = eventProtos.get(i);
+      if (buildEvent.getId().hasNamedSet()) {
+        // These are the separately-posted contents of reported artifacts.
+        seenFileSets.add(buildEvent.getId().getNamedSet().getId());
+        for (NamedSetOfFilesId nestedSetId : buildEvent.getNamedSetOfFiles().getFileSetsList()) {
+          assertThat(seenFileSets).contains(nestedSetId.getId());
+        }
+      } else if (buildEvent.getId().hasUnknown()) {
+        // These are the GenericArtifactReportingEvent that report artifacts.
+        for (NamedSetOfFilesId nestedSetId : buildEvent.getNamedSetOfFiles().getFileSetsList()) {
+          assertThat(seenFileSets).contains(nestedSetId.getId());
+        }
+      }
+    }
+  }
+
+  @Test
   public void testStdoutReported() {
     // Verify that stdout and stderr are reported in the build-event stream on progress
     // events.