feat(storage): change gRPC writes to use bi-directional streams

[BrennaEpp]

Emulated test is failing as bidi writes are not implemented in the emulator yet, so I added a quick skip for that test.

[noahdietz]

Are we 100% sure we can drop the stream reference here? If w.stream.Send gets an io.EOF, it means the backend wants to send an error. The error status won't come from CloseSend(), I think the client side still needs to Recv to get the actual Status (will send a link offline). I think we need to call Recv or CloseAndRecv if the bidi stream still has it, before Closing.

[tritone]

+1, we'd have to get clarification here and test it out in practice...

[BrennaEpp]

Thanks Noah, you are 100% correct. In this case, it would be a Recv() (the bidi stream does not have CloseAndRecv). Updated the code.

As for testing it out in practice, I agree. I think the best is to do this via the conformance tests, pending testbench implementation.

[tritone]

Overall looks pretty good, but I had some questions

[tritone]

+1, we'd have to get clarification here and test it out in practice...

[tritone]

Is it okay to do this without backoff? Wondering if this is a typical pattern.

[BrennaEpp]

I was wondering that too, but I can't find any resources on this because it seems that for the typical use case bidirectional clients receive and send in separate goroutines.
Here is what the typical pattern looks like:

for {
		resp, err := stream.Recv()
		if err == io.EOF {
			break
		} else if err != nil {
			return err
		}
		got = append(got, resp)
	}

All the resources I am looking at do say to receive from the stream until we get an error and I believe it is the same for response. For certain I know that the code as-is written may receive several responses (~1-4) before it will receive the object. Once we call CloseSend(), Recv() won't block and the server will always send, so this loop is just waiting for the server side to finish writing the object.
Either way, I've also added another loop following to make sure the stream is properly garbage collected, as seems to be required. I've also seen the following for example:

	// Keep reading to drain the stream.
	for {
		if _, err := stream.Recv(); err != nil {
			break
		}
	}

[tritone]

Shouldn't this happen outside of the if ShouldRetry() conditional?

[BrennaEpp]

I think it doesn't matter too much either way here. If we don't go into the conditional (ie. shouldRetry == false) it'll return from uploadBuffer with the err and we won't reuse the stream. If we do go into the conditional, I don't think it matters if it happened before or during. Unless I'm missing something?

[tritone]

You're right but it seems more readable to drop the stream reference immediately after the error if that's the reason we are doing it.

[BrennaEpp]

I moved this line

[tritone]

Are there docs that we can link to that officially document the "stream protocol" rather than just a commit message?

If not I think it'd be good to get a review from someone on the gRPC Go team if possible, given that these contracts are a little convoluted seemingly.

[BrennaEpp]

This is the actual docs: https://pkg.go.dev/google.golang.org/grpc#ClientConn.NewStream

I'll update it in the comment as well.

[tritone]

After looking at these docs I'm wondering if we can use the CloseAndRecv helper instead to avoid writing these loops? We already do this here:

google-cloud-go/storage/grpc_client.go

Line 1748 in cc83515

resp, err := w.stream.CloseAndRecv()

[BrennaEpp]

We don't have that helper function for the bidi stream

[tritone]

As I mentioned in another comment, I don't see why we would make this change; stream.CloseAndRecv should do what we want?

[tritone]

You're right but it seems more readable to drop the stream reference immediately after the error if that's the reason we are doing it.