Make timestamps of InternalOrderedBuildEvent immutable.

Similar to the sequence number, the timestamp should be immutable and computed when the event is enqueued (not at serialization time).

PiperOrigin-RevId: 206186036

2 files changed, 68 insertions, 35 deletions

diff --git a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceProtoUtil.java b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceProtoUtil.java
index dc6172cd08..4c0973f5b3 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceProtoUtil.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceProtoUtil.java
@@ -34,7 +34,7 @@ import com.google.devtools.build.v1.PublishLifecycleEventRequest;
 import com.google.devtools.build.v1.StreamId;
 import com.google.devtools.build.v1.StreamId.BuildComponent;
 import com.google.protobuf.Any;
-import com.google.protobuf.util.Timestamps;
+import com.google.protobuf.Timestamp;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import javax.annotation.Nullable;
@@ -47,7 +47,6 @@ public final class BuildEventServiceProtoUtil {
   private final String projectId;
   private final AtomicInteger streamSequenceNumber;
   private final String commandName;
-  private final Clock clock;
   private final Set<String> additionalKeywords;
 
   public BuildEventServiceProtoUtil(
@@ -61,42 +60,49 @@ public final class BuildEventServiceProtoUtil {
     this.buildInvocationId = buildInvocationId;
     this.projectId = projectId;
     this.commandName = commandName;
-    this.clock = clock;
     this.additionalKeywords = additionalKeywords;
     this.streamSequenceNumber = new AtomicInteger(1);
   }
 
-  public PublishLifecycleEventRequest buildEnqueued() {
-    return lifecycleEvent(projectId, 1,
+  public PublishLifecycleEventRequest buildEnqueued(Timestamp timestamp) {
+    return lifecycleEvent(
+            projectId,
+            1,
             com.google.devtools.build.v1.BuildEvent.newBuilder()
-                .setEventTime(Timestamps.fromMillis(clock.currentTimeMillis()))
+                .setEventTime(timestamp)
                 .setBuildEnqueued(BuildEnqueued.newBuilder()))
         .build();
   }
 
-  public PublishLifecycleEventRequest buildFinished(Result result) {
-    return lifecycleEvent(projectId, 2,
+  public PublishLifecycleEventRequest buildFinished(Timestamp timestamp, Result result) {
+    return lifecycleEvent(
+            projectId,
+            2,
             com.google.devtools.build.v1.BuildEvent.newBuilder()
-                .setEventTime(Timestamps.fromMillis(clock.currentTimeMillis()))
+                .setEventTime(timestamp)
                 .setBuildFinished(
                     BuildFinished.newBuilder()
                         .setStatus(BuildStatus.newBuilder().setResult(result))))
         .build();
   }
 
-  public PublishLifecycleEventRequest invocationStarted() {
-    return lifecycleEvent(projectId, 1,
+  public PublishLifecycleEventRequest invocationStarted(Timestamp timestamp) {
+    return lifecycleEvent(
+            projectId,
+            1,
             com.google.devtools.build.v1.BuildEvent.newBuilder()
-                .setEventTime(Timestamps.fromMillis(clock.currentTimeMillis()))
+                .setEventTime(timestamp)
                 .setInvocationAttemptStarted(
                     InvocationAttemptStarted.newBuilder().setAttemptNumber(1)))
         .build();
   }
 
-  public PublishLifecycleEventRequest invocationFinished(Result result) {
-    return lifecycleEvent(projectId, 2,
+  public PublishLifecycleEventRequest invocationFinished(Timestamp timestamp, Result result) {
+    return lifecycleEvent(
+            projectId,
+            2,
             com.google.devtools.build.v1.BuildEvent.newBuilder()
-                .setEventTime(Timestamps.fromMillis(clock.currentTimeMillis()))
+                .setEventTime(timestamp)
                 .setInvocationAttemptFinished(
                     InvocationAttemptFinished.newBuilder()
                         .setInvocationStatus(BuildStatus.newBuilder().setResult(result))))
@@ -108,20 +114,19 @@ public final class BuildEventServiceProtoUtil {
   }
 
   /** Creates a PublishBuildToolEventStreamRequest from a packed bazel event. */
-  public PublishBuildToolEventStreamRequest bazelEvent(Any packedEvent, int sequenceNumber) {
-    return bazelEvent(sequenceNumber, packedEvent);
-  }
-
-  @VisibleForTesting
-  public PublishBuildToolEventStreamRequest bazelEvent(int sequenceNumber, Any packedEvent) {
+  public PublishBuildToolEventStreamRequest bazelEvent(
+      int sequenceNumber, Timestamp timestamp, Any packedEvent) {
     return publishBuildToolEventStreamRequest(
         sequenceNumber,
+        timestamp,
         com.google.devtools.build.v1.BuildEvent.newBuilder().setBazelEvent(packedEvent));
   }
 
-  public PublishBuildToolEventStreamRequest streamFinished(int sequenceNumber) {
+  public PublishBuildToolEventStreamRequest streamFinished(
+      int sequenceNumber, Timestamp timestamp) {
     return publishBuildToolEventStreamRequest(
         sequenceNumber,
+        timestamp,
         BuildEvent.newBuilder()
             .setComponentStreamFinished(
                 BuildComponentStreamFinished.newBuilder().setType(FINISHED)));
@@ -129,14 +134,13 @@ public final class BuildEventServiceProtoUtil {
 
   @VisibleForTesting
   public PublishBuildToolEventStreamRequest publishBuildToolEventStreamRequest(
-      int sequenceNumber, BuildEvent.Builder besEvent) {
+      int sequenceNumber, Timestamp timestamp, BuildEvent.Builder besEvent) {
     PublishBuildToolEventStreamRequest.Builder builder =
         PublishBuildToolEventStreamRequest.newBuilder()
             .setOrderedBuildEvent(
                 OrderedBuildEvent.newBuilder()
                     .setSequenceNumber(sequenceNumber)
-                    .setEvent(
-                        besEvent.setEventTime(Timestamps.fromMillis(clock.currentTimeMillis())))
+                    .setEvent(besEvent.setEventTime(timestamp))
                     .setStreamId(streamId(besEvent.getEventCase())));
     if (sequenceNumber == 1) {
       builder.addAllNotificationKeywords(getKeywords());
diff --git a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java
index e62f2df64b..160c2ae4fc 100644
--- a/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java
+++ b/src/main/java/com/google/devtools/build/lib/buildeventservice/BuildEventServiceTransport.java
@@ -57,6 +57,8 @@ import com.google.devtools.build.v1.PublishBuildToolEventStreamRequest;
 import com.google.devtools.build.v1.PublishBuildToolEventStreamResponse;
 import com.google.devtools.build.v1.PublishLifecycleEventRequest;
 import com.google.protobuf.Any;
+import com.google.protobuf.Timestamp;
+import com.google.protobuf.util.Timestamps;
 import io.grpc.Status;
 import io.grpc.Status.Code;
 import io.grpc.StatusException;
@@ -109,6 +111,7 @@ public class BuildEventServiceTransport implements BuildEventTransport {
   private final BuildEventArtifactUploader artifactUploader;
   private final Sleeper sleeper;
   private final boolean errorsShouldFailTheBuild;
+  private final Clock clock;
   /** Contains all pendingAck events that might be retried in case of failures. */
   private ConcurrentLinkedDeque<InternalOrderedBuildEvent> pendingAck;
   /** Contains all events should be sent ordered by sequence number. */
@@ -220,6 +223,7 @@ public class BuildEventServiceTransport implements BuildEventTransport {
     this.sleeper = sleeper;
     this.besResultsUrl = besResultsUrl;
     this.errorsShouldFailTheBuild = errorsShouldFailTheBuild;
+    this.clock = clock;
   }
 
   public boolean isStreaming() {
@@ -259,11 +263,13 @@ public class BuildEventServiceTransport implements BuildEventTransport {
       return shutdownFuture;
     }
 
+    Timestamp lastEventTimestamp = timestamp();
     uploaderExecutorService.execute(
         () -> {
           try {
             sendOrderedBuildEvent(
-                new LastInternalOrderedBuildEvent(besProtoUtil.nextSequenceNumber()));
+                new LastInternalOrderedBuildEvent(
+                    besProtoUtil.nextSequenceNumber(), lastEventTimestamp));
 
             if (errorsReported) {
               // If we encountered errors before and have already reported them, then we should
@@ -345,10 +351,14 @@ public class BuildEventServiceTransport implements BuildEventTransport {
     ListenableFuture<PathConverter> upload = artifactUploader.upload(localFileMap.build());
     InternalOrderedBuildEvent buildEvent =
         new DefaultInternalOrderedBuildEvent(
-            event, namer, upload, besProtoUtil.nextSequenceNumber());
+            event, namer, upload, besProtoUtil.nextSequenceNumber(), timestamp());
     sendOrderedBuildEvent(buildEvent);
   }
 
+  private Timestamp timestamp() {
+    return Timestamps.fromMillis(clock.currentTimeMillis());
+  }
+
   private String errorMessageFromException(Throwable t) {
     String message;
     if (t instanceof TimeoutException) {
@@ -420,14 +430,14 @@ public class BuildEventServiceTransport implements BuildEventTransport {
     @Override
     public Void call() throws Exception {
       try {
-        publishLifecycleEvent(besProtoUtil.buildEnqueued());
-        publishLifecycleEvent(besProtoUtil.invocationStarted());
+        publishLifecycleEvent(besProtoUtil.buildEnqueued(timestamp()));
+        publishLifecycleEvent(besProtoUtil.invocationStarted(timestamp()));
         try {
           retryOnException(BuildEventServiceTransport.this::publishEventStream);
         } finally {
           Result result = getInvocationResult();
-          publishLifecycleEvent(besProtoUtil.invocationFinished(result));
-          publishLifecycleEvent(besProtoUtil.buildFinished(result));
+          publishLifecycleEvent(besProtoUtil.invocationFinished(timestamp(), result));
+          publishLifecycleEvent(besProtoUtil.buildFinished(timestamp(), result));
         }
       } finally {
         try {
@@ -658,10 +668,16 @@ public class BuildEventServiceTransport implements BuildEventTransport {
    * (serialized).
    */
   private interface InternalOrderedBuildEvent {
+
+    /** Returns whether this is the last event. */
     boolean isLastEvent();
 
+    /** Returns the immutable sequence number for this event. */
     int getSequenceNumber();
 
+    /** Returns the immutable Timestamp for this event. */
+    Timestamp getTimestamp();
+
     ListenableFuture<PathConverter> localFileUploadProgress();
 
     PublishBuildToolEventStreamRequest serialize(PathConverter pathConverter);
@@ -672,16 +688,19 @@ public class BuildEventServiceTransport implements BuildEventTransport {
     private final ArtifactGroupNamer artifactGroupNamer;
     private final ListenableFuture<PathConverter> artifactUpload;
     private final int sequenceNumber;
+    private final Timestamp timestamp;
 
     DefaultInternalOrderedBuildEvent(
         BuildEvent event,
         ArtifactGroupNamer artifactGroupNamer,
         ListenableFuture<PathConverter> artifactUpload,
-        int sequenceNumber) {
+        int sequenceNumber,
+        Timestamp timestamp) {
       this.event = Preconditions.checkNotNull(event);
       this.artifactGroupNamer = Preconditions.checkNotNull(artifactGroupNamer);
       this.artifactUpload = artifactUpload;
       this.sequenceNumber = sequenceNumber;
+      this.timestamp = timestamp;
     }
 
     @Override
@@ -694,6 +713,10 @@ public class BuildEventServiceTransport implements BuildEventTransport {
       return sequenceNumber;
     }
 
+    public Timestamp getTimestamp() {
+      return timestamp;
+    }
+
     @Override
     public ListenableFuture<PathConverter> localFileUploadProgress() {
       return artifactUpload;
@@ -719,15 +742,17 @@ public class BuildEventServiceTransport implements BuildEventTransport {
                   return protocolOptions;
                 }
               });
-      return besProtoUtil.bazelEvent(Any.pack(eventProto), sequenceNumber);
+      return besProtoUtil.bazelEvent(getSequenceNumber(), getTimestamp(), Any.pack(eventProto));
     }
   }
 
   private class LastInternalOrderedBuildEvent implements InternalOrderedBuildEvent {
     private final int sequenceNumber;
+    private final Timestamp timestamp;
 
-    LastInternalOrderedBuildEvent(int sequenceNumber) {
+    LastInternalOrderedBuildEvent(int sequenceNumber, Timestamp timestamp) {
       this.sequenceNumber = sequenceNumber;
+      this.timestamp = timestamp;
     }
 
     @Override
@@ -740,6 +765,10 @@ public class BuildEventServiceTransport implements BuildEventTransport {
       return sequenceNumber;
     }
 
+    public Timestamp getTimestamp() {
+      return timestamp;
+    }
+
     @Override
     public ListenableFuture<PathConverter> localFileUploadProgress() {
       return Futures.immediateFuture(PathConverter.NO_CONVERSION);
@@ -747,7 +776,7 @@ public class BuildEventServiceTransport implements BuildEventTransport {
 
     @Override
     public PublishBuildToolEventStreamRequest serialize(PathConverter pathConverter) {
-      return besProtoUtil.streamFinished(sequenceNumber);
+      return besProtoUtil.streamFinished(getSequenceNumber(), getTimestamp());
     }
   }