Repository: buchgr/bazel-remote
Pull request #150 (2020-01-01 10:46:13)

gRPC: more inlining fixes


Patch:
diff --git a/server/grpc.go b/server/grpc.go
index d15a6d9..fafb29f 100644
--- a/server/grpc.go
+++ b/server/grpc.go
@@ -20,6 +20,7 @@ import (
 
 const (
 	hashKeyLength = 64
+	emptySha256   = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
 )
 
 var (
@@ -90,7 +91,17 @@ func (s *grpcServer) GetCapabilities(ctx context.Context,
 }
 
 // Return an error if `hash` is not a valid cache key.
-func (s *grpcServer) validateHash(hash string, logPrefix string) error {
+func (s *grpcServer) validateHash(hash string, size int64, logPrefix string) error {
+	if size == int64(0) {
+		if hash == emptySha256 {
+			return nil
+		}
+
+		msg := "Invalid zero-length SHA256 hash"
+		s.accessLogger.Printf("%s %s: %s", logPrefix, hash, msg)
+		return status.Error(codes.InvalidArgument, msg)
+	}
+
 	if len(hash) != hashKeyLength {
 		msg := fmt.Sprintf("Hash length must be length %d", hashKeyLength)
 		s.accessLogger.Printf("%s %s: %s", logPrefix, hash, msg)
diff --git a/server/grpc_ac.go b/server/grpc_ac.go
index d4da291..a857ba6 100644
--- a/server/grpc_ac.go
+++ b/server/grpc_ac.go
@@ -38,7 +38,7 @@ func (s *grpcServer) GetActionResult(ctx context.Context,
 	req *pb.GetActionResultRequest) (*pb.ActionResult, error) {
 
 	errorPrefix := "GRPC AC GET"
-	err := s.validateHash(req.ActionDigest.Hash, errorPrefix)
+	err := s.validateHash(req.ActionDigest.Hash, req.ActionDigest.SizeBytes, errorPrefix)
 	if err != nil {
 		return nil, err
 	}
@@ -152,7 +152,7 @@ func (s *grpcServer) UpdateActionResult(ctx context.Context,
 	req *pb.UpdateActionResultRequest) (*pb.ActionResult, error) {
 
 	errorPrefix := "GRPC AC PUT"
-	err := s.validateHash(req.ActionDigest.Hash, errorPrefix)
+	err := s.validateHash(req.ActionDigest.Hash, req.ActionDigest.SizeBytes, errorPrefix)
 	if err != nil {
 		return nil, err
 	}
diff --git a/server/grpc_cas.go b/server/grpc_cas.go
index d4c6914..62853d9 100644
--- a/server/grpc_cas.go
+++ b/server/grpc_cas.go
@@ -32,11 +32,17 @@ func (s *grpcServer) FindMissingBlobs(ctx context.Context,
 	errorPrefix := "GRPC CAS GET"
 	for _, digest := range req.BlobDigests {
 		hash := digest.GetHash()
-		err := s.validateHash(hash, errorPrefix)
+		err := s.validateHash(hash, digest.SizeBytes, errorPrefix)
 		if err != nil {
 			return nil, err
 		}
 
+		if digest.SizeBytes == 0 {
+			// The hash was validated, so we know it's OK.
+			s.accessLogger.Printf("GRPC CAS HEAD %s OK", hash)
+			continue
+		}
+
 		if !s.cache.Contains(cache.CAS, hash) {
 			s.accessLogger.Printf("GRPC CAS HEAD %s NOT FOUND", hash)
 			resp.MissingBlobDigests = append(resp.MissingBlobDigests, digest)
@@ -59,7 +65,7 @@ func (s *grpcServer) BatchUpdateBlobs(ctx context.Context,
 	errorPrefix := "GRPC CAS PUT"
 	for _, req := range in.Requests {
 		// TODO: consider fanning-out goroutines here.
-		err := s.validateHash(req.Digest.Hash, errorPrefix)
+		err := s.validateHash(req.Digest.Hash, req.Digest.SizeBytes, errorPrefix)
 		if err != nil {
 			return nil, err
 		}
@@ -91,6 +97,14 @@ func (s *grpcServer) BatchUpdateBlobs(ctx context.Context,
 // found, the returned error is errBlobNotFound. Only use this
 // function when it's OK to buffer the entire blob in memory.
 func (s *grpcServer) getBlobData(hash string, size int64) ([]byte, error) {
+	if size < 0 {
+		return []byte{}, errBadSize
+	}
+
+	if size == 0 {
+		return []byte{}, nil
+	}
+
 	rdr, sizeBytes, err := s.cache.Get(cache.CAS, hash)
 	if err != nil {
 		rdr.Close()
@@ -135,6 +149,7 @@ func (s *grpcServer) getBlobResponse(digest *pb.Digest) *pb.BatchReadBlobsRespon
 	r.Data = data
 
 	s.accessLogger.Printf("GRPC CAS GET %s OK", digest.Hash)
+	r.Status = &status.Status{Code: int32(codes.OK)}
 	return &r
 }
 
@@ -149,7 +164,7 @@ func (s *grpcServer) BatchReadBlobs(ctx context.Context,
 	errorPrefix := "GRPC CAS GET"
 	for _, digest := range in.Digests {
 		// TODO: consider fanning-out goroutines here.
-		err := s.validateHash(digest.Hash, errorPrefix)
+		err := s.validateHash(digest.Hash, digest.SizeBytes, errorPrefix)
 		if err != nil {
 			return nil, err
 		}
@@ -166,7 +181,7 @@ func (s *grpcServer) GetTree(in *pb.GetTreeRequest,
 		Directories: make([]*pb.Directory, 0),
 	}
 	errorPrefix := "GRPC CAS GETTREEREQUEST"
-	err := s.validateHash(in.RootDigest.Hash, errorPrefix)
+	err := s.validateHash(in.RootDigest.Hash, in.RootDigest.SizeBytes, errorPrefix)
 	if err != nil {
 		return err
 	}
@@ -212,7 +227,7 @@ func (s *grpcServer) fillDirectories(resp *pb.GetTreeResponse, dir *pb.Directory
 	// Recursively append all the child dirs.
 	for _, dirNode := range dir.Directories {
 
-		err := s.validateHash(dirNode.Digest.Hash, errorPrefix)
+		err := s.validateHash(dirNode.Digest.Hash, dirNode.Digest.SizeBytes, errorPrefix)
 		if err != nil {
 			return err
 		}


Test patch:
diff --git a/server/grpc_bytestream.go b/server/grpc_bytestream.go
index 30b037f..90bf90e 100644
--- a/server/grpc_bytestream.go
+++ b/server/grpc_bytestream.go
@@ -68,16 +68,23 @@ func (s *grpcServer) Read(req *bytestream.ReadRequest,
 		return status.Error(codes.InvalidArgument, msg)
 	}
 
-	hash := rem[0]
-	err := s.validateHash(hash, errorPrefix)
+	size, err := strconv.ParseInt(rem[1], 10, 64)
 	if err != nil {
-		return err
+		msg := fmt.Sprintf("Invalid size: %s", rem[1])
+		s.accessLogger.Printf("%s: %s", errorPrefix, msg)
+		return status.Error(codes.InvalidArgument, msg)
+	}
+	if size < 0 {
+		msg := fmt.Sprintf("Invalid size (must be non-negative): %s", rem[1])
+		s.accessLogger.Printf("%s: %s", errorPrefix, msg)
+		return status.Error(codes.InvalidArgument, msg)
 	}
 
-	size, err := strconv.ParseInt(rem[1], 10, 64)
+	hash := rem[0]
+
+	err = s.validateHash(hash, size, errorPrefix)
 	if err != nil {
-		s.accessLogger.Printf("GRPC BYTESTREAM READ FAILED: unable to parse size: %s", err)
-		return status.Error(codes.Unknown, err.Error())
+		return err
 	}
 
 	if req.ReadOffset > size {
@@ -204,14 +211,14 @@ func (s *grpcServer) parseWriteResource(r string) (string, int64, error) {
 	}
 
 	hash := rem[2]
-	err := s.validateHash(hash, "GRPC BYTESTREAM READ FAILED")
+	size, err := strconv.ParseInt(rem[3], 10, 64)
 	if err != nil {
-		return "", 0, err
+		return "", 0, status.Errorf(codes.InvalidArgument, "Unable to parse size: %s", rem[3])
 	}
 
-	size, err := strconv.ParseInt(rem[3], 10, 64)
+	err = s.validateHash(hash, size, "GRPC BYTESTREAM READ FAILED")
 	if err != nil {
-		return "", 0, status.Errorf(codes.InvalidArgument, "Unable to parse size: %s", rem[3])
+		return "", 0, err
 	}
 
 	return hash, size, nil
diff --git a/server/grpc_test.go b/server/grpc_test.go
index e04aa88..f162ed9 100644
--- a/server/grpc_test.go
+++ b/server/grpc_test.go
@@ -351,7 +351,30 @@ func TestGrpcAcRequestInlinedBlobs(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	// Check that we can download the blobs individually.
+	// Check that the blobs exist.
+
+	missingReq := pb.FindMissingBlobsRequest{
+		BlobDigests: []*pb.Digest{
+			&outputFileDigest,
+			&emptyFileDigest,
+			&stdoutDigest,
+		},
+	}
+
+	missingResp, err := casClient.FindMissingBlobs(ctx, &missingReq)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(missingResp.MissingBlobDigests) != 0 {
+		for _, d := range missingResp.MissingBlobDigests {
+			t.Log("Blob missing from the CAS:", d.Hash, d.SizeBytes)
+		}
+		t.Fatal("Expected", len(missingReq.BlobDigests), "blobs, missing",
+			len(missingResp.MissingBlobDigests))
+	}
+
+	// Double-check that we can actually download the blobs individually.
 
 	downReq := pb.BatchReadBlobsRequest{
 		Digests: []*pb.Digest{
@@ -371,6 +394,20 @@ func TestGrpcAcRequestInlinedBlobs(t *testing.T) {
 		t.Fatal("Expected", len(downReq.Digests), "responses, got",
 			len(downResp.GetResponses()))
 	}
+
+	for _, r := range downResp.GetResponses() {
+		if r == nil {
+			t.Fatal("nil response in BatchReadBlobsResponse")
+		}
+
+		if r.Status == nil {
+			t.Fatal("nil status in BatchReadBlobsResponse_Response", r.Digest)
+		}
+
+		if r.Status.GetCode() != int32(codes.OK) {
+			t.Fatal("missing blob:", r.Digest)
+		}
+	}
 }
 
 func TestGrpcByteStream(t *testing.T) {