Nonblocking PublishRevisions (#1149)

* Add block parameter to PublishRevisionsRequest

* Nonblocking path

Author: gdbelvin

core/integration/client_tests.go

@@ -373,6 +373,7 @@ func (env *Env) setupHistory(ctx context.Context, directory *pb.Directory, userI
 				DirectoryId: directory.DirectoryId,
 				MinBatch:    1,
 				MaxBatch:    1,
+				Block:       true,
 			}); err != nil {
 				return fmt.Errorf("sequencer.RunBatch(%v): %v", i, err)
 			}

core/sequencer/sequencer_api.proto

@@ -56,6 +56,8 @@ message RunBatchRequest {
   int32 min_batch = 2;
   // max_batch is the maximum number of items in a batch.
   int32 max_batch = 3;
+  // block until a Signed Log Root has been published which encompases all map roots.
+  bool block = 4;
 }
 
 // DefineRevisionRequest contains information needed to define a new revision.
@@ -99,6 +101,8 @@ message ApplyRevisionResponse {
 // PublishRevisionsRequest copies all available SignedMapRoots into the Log of SignedMapRoots.
 message PublishRevisionsRequest {
   string directory_id = 1;
+  // block until a Signed Log Root has been published which encompases all map roots.
+  bool block = 2;
 }
 
 // PublishRevisionsResponse contains metrics about the publishing operation.

core/sequencer/sequencer_go_proto/sequencer_api.pb.go

@@ -166,7 +166,7 @@ func (s *Server) RunBatch(ctx context.Context, in *spb.RunBatchRequest) (*empty.
 		}
 	}
 
-	publishReq := &spb.PublishRevisionsRequest{DirectoryId: in.DirectoryId}
+	publishReq := &spb.PublishRevisionsRequest{DirectoryId: in.DirectoryId, Block: in.Block}
 	_, err = s.loopback.PublishRevisions(ctx, publishReq)
 	if err != nil {
 		return nil, err
@@ -377,9 +377,10 @@ func (s *Server) PublishRevisions(ctx context.Context,
 		return nil, err
 	}
 
-	// TODO(gbelvin): Remove wait when batching boundaries are deterministic.
-	if err := logClient.WaitForInclusion(ctx, latestRawMapRoot.GetMapRoot()); err != nil {
-		return nil, status.Errorf(codes.Internal, "WaitForInclusion(): %v", err)
+	if in.Block {
+		if err := logClient.WaitForInclusion(ctx, latestRawMapRoot.GetMapRoot()); err != nil {
+			return nil, status.Errorf(codes.Internal, "WaitForInclusion(): %v", err)
+		}
 	}
 	return &spb.PublishRevisionsResponse{Revisions: revs}, nil
 }