fix(storage): force first message on next sendBuffer when nothing sent on current (googleapis#11871)

* fix(storage): force first message on next sendBuffer when nothing is sent

If nothing is sent but the stream is reconnected, next send will not recognize the stream as reconnected without this change.

* remove reconnected var

---------

Co-authored-by: Chris Cotter <cjcotter@google.com>

Author: BrennaEpp

storage/grpc_client.go

@@ -2839,7 +2839,6 @@ func (s *gRPCResumableBidiWriteBufferSender) queryProgress(ctx context.Context)
 }
 
 func (s *gRPCResumableBidiWriteBufferSender) sendBuffer(ctx context.Context, buf []byte, offset int64, flush, finishWrite bool) (obj *storagepb.Object, err error) {
-	reconnected := false
 	if s.stream == nil {
 		// Determine offset and reconnect
 		s.flushOffset, err = s.queryProgress(ctx)
@@ -2850,7 +2849,7 @@ func (s *gRPCResumableBidiWriteBufferSender) sendBuffer(ctx context.Context, buf
 		if err != nil {
 			return
 		}
-		reconnected = true
+		s.forceFirstMessage = true
 	}
 
 	// clean up buf. We'll still write the message if a flush/finishWrite was
@@ -2868,7 +2867,7 @@ func (s *gRPCResumableBidiWriteBufferSender) sendBuffer(ctx context.Context, buf
 	}
 
 	req := bidiWriteObjectRequest(buf, offset, flush, finishWrite)
-	if s.forceFirstMessage || reconnected {
+	if s.forceFirstMessage {
 		req.FirstMessage = &storagepb.BidiWriteObjectRequest_UploadId{UploadId: s.upid}
 		s.forceFirstMessage = false
 	}

storage/retry_conformance_test.go

@@ -613,8 +613,9 @@ var methods = map[string][]retryFunc{
 				obj = obj.If(Conditions{DoesNotExist: true})
 			}
 			w := obj.NewWriter(ctx)
-			// Set Writer.ChunkSize to 2 MiB to perform resumable uploads.
-			w.ChunkSize = 2097152
+			// Set Writer.ChunkSize to 4MiB to perform resumable uploads on a smaller object size.
+			// Set it larger than 2MiB so it can test boundaries for max message size.
+			w.ChunkSize = 4 * MiB
 
 			if _, err := w.Write(randomBytes9MiB); err != nil {
 				return fmt.Errorf("writing object: %v", err)