// Copyright 2025 Google LLC

//

// Licensed under the Apache License, Version 2.0 (the "License");

// you may not use this file except in compliance with the License.

// You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software

// distributed under the License is distributed on an "AS IS" BASIS,

// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

// See the License for the specific language governing permissions and

// limitations under the License.

package storage

import (

"context"

"errors"

"fmt"

"hash/crc32"

"io"

"net/http"

"net/url"

"strings"

"sync"

"time"

gapic "cloud.google.com/go/storage/internal/apiv2"

"cloud.google.com/go/storage/internal/apiv2/storagepb"

gax "github.com/googleapis/gax-go/v2"

"google.golang.org/api/googleapi"

"google.golang.org/grpc/codes"

"google.golang.org/grpc/status"

"google.golang.org/protobuf/proto"

)

const (

// defaultWriteChunkRetryDeadline is the default deadline for the upload

// of a single chunk. It can be overwritten by Writer.ChunkRetryDeadline.

defaultWriteChunkRetryDeadline = 32 * time.Second

// maxPerMessageWriteSize is the maximum amount of content that can be sent

// per WriteObjectRequest message. A buffer reaching this amount will

// precipitate a flush of the buffer. It is only used by the gRPC Writer

// implementation.

maxPerMessageWriteSize int = int(storagepb.ServiceConstants_MAX_WRITE_CHUNK_BYTES)

)

func (w *gRPCWriter) Write(p []byte) (n int, err error) {

done := make(chan struct{})

cmd := &gRPCWriterCommandWrite{p: p, done: done}

select {

case <-w.donec:

return 0, w.streamResult

case w.writesChan <- cmd:

md5Provided := w.attrs != nil && w.attrs.MD5 != nil

// Update fullObjectChecksum on every write and send it on finalWrite if not disabled.

// Skip checksum calculation if user configures MD5 or CRC32C themselves.

if !w.disableAutoChecksum &&

!w.sendCRC32C &&

!md5Provided &&

!w.append {

w.fullObjectChecksum = crc32.Update(w.fullObjectChecksum, crc32cTable, p)

}

// write command successfully delivered to sender. We no longer own cmd.

break

}

select {

case <-w.donec:

return 0, w.streamResult

case <-done:

return len(p), nil

}

}

func (w *gRPCWriter) Flush() (int64, error) {

done := make(chan int64)

cmd := &gRPCWriterCommandFlush{done: done}

select {

case <-w.donec:

return 0, w.streamResult

case w.writesChan <- cmd:

// flush command successfully delivered to sender. We no longer own cmd.

break

}

select {

case <-w.donec:

return 0, w.streamResult

case f := <-done:

return f, nil

}

}

func (w *gRPCWriter) Close() error {

w.CloseWithError(nil)

return w.streamResult

}

func (w *gRPCWriter) CloseWithError(err error) error {

// N.B. CloseWithError always returns nil!

select {

case <-w.donec:

return nil

case w.writesChan <- &gRPCWriterCommandClose{err: err}:

break

}

<-w.donec

return nil

}

func (w *gRPCWriter) setAppendFinalCRC32C(sendAppendFinalCRC32C bool, c uint32) {

w.sendAppendFinalCRC32C = sendAppendFinalCRC32C

w.appendFinalCRC32C = c

}

func (c *grpcStorageClient) OpenWriter(params *openWriterParams, opts ...storageOption) (internalWriter, error) {

if params.attrs.Retention != nil {

// TO-DO: remove once ObjectRetention is available - see b/308194853

return nil, status.Errorf(codes.Unimplemented, "storage: object retention is not supported in gRPC")

}

spec := &storagepb.WriteObjectSpec{

Resource: params.attrs.toProtoObject(params.bucket),

Appendable: proto.Bool(params.append),

}

// WriteObject doesn't support the generation condition, so use default.

if err := applyCondsProto("WriteObject", defaultGen, params.conds, spec); err != nil {

return nil, err

}

s := callSettings(c.settings, opts...)

if s.retry == nil {

s.retry = defaultRetry.clone()

}

if params.append {

s.retry = withBidiWriteObjectRedirectionErrorRetries(s)

}

chunkRetryDeadline := defaultWriteChunkRetryDeadline

if params.chunkRetryDeadline != 0 {

chunkRetryDeadline = params.chunkRetryDeadline

}

ctx := params.ctx

if s.userProject != "" {

ctx = setUserProjectMetadata(ctx, s.userProject)

}

chunkSize := gRPCChunkSize(params.chunkSize)

writeQuantum := maxPerMessageWriteSize

if writeQuantum > chunkSize {

writeQuantum = chunkSize

}

sendableUnits := chunkSize / writeQuantum

// There's no strict requirement that the chunk size be an exact multiple of

// the writeQuantum. In that case, there will be a tail segment of less than

// writeQuantum.

lastSegmentStart := sendableUnits * writeQuantum

if lastSegmentStart < chunkSize {

sendableUnits++

}

if params.append && params.appendGen >= 0 && params.setTakeoverOffset == nil {

return nil, errors.New("storage: no way to report offset for appendable takeover")

}

w := &gRPCWriter{

preRunCtx: ctx,

c: c,

settings: s,

bucket: params.bucket,

attrs: params.attrs,

conds: params.conds,

spec: spec,

encryptionKey: params.encryptionKey,

setError: params.setError,

progress: params.progress,

setObj: params.setObj,

setSize: params.setSize,

setTakeoverOffset: params.setTakeoverOffset,

flushSupported: params.append,

sendCRC32C: params.sendCRC32C,

disableAutoChecksum: params.disableAutoChecksum,

forceOneShot: params.chunkSize <= 0,

forceEmptyContentType: params.forceEmptyContentType,

append: params.append,

appendGen: params.appendGen,

finalizeOnClose: params.finalizeOnClose,

buf: nil, // Allocated lazily on first buffered write.

chunkSize: chunkSize,

writeQuantum: writeQuantum,

lastSegmentStart: lastSegmentStart,

sendableUnits: sendableUnits,

bufUnsentIdx: 0,

bufFlushedIdx: -1, // Handle flushes to length 0

bufBaseOffset: 0,

chunkRetryDeadline: chunkRetryDeadline,

abandonRetriesTime: time.Time{},

attempts: 0,

lastErr: nil,

streamSender: nil,

writesChan: make(chan gRPCWriterCommand, 1),

currentCommand: nil,

streamResult: nil,

donec: params.donec,

}

go func() {

if err := w.gatherFirstBuffer(); err != nil {

w.streamResult = err

w.setError(err)

close(w.donec)

return

}

if w.attrs.ContentType == "" && !w.forceEmptyContentType {

w.spec.Resource.ContentType = w.detectContentType()

}

w.streamSender = w.pickBufferSender()

// Writer does not use maxRetryDuration from retryConfig to maintain

// consistency with HTTP client behavior. Writers should use

// ChunkRetryDeadline for per-chunk timeouts and context for overall timeouts.

writerRetry := w.settings.retry

if writerRetry != nil {

writerRetry = writerRetry.clone()

writerRetry.maxRetryDuration = 0

}

w.streamResult = checkCanceled(run(w.preRunCtx, func(ctx context.Context) error {

w.lastErr = w.writeLoop(ctx)

return w.lastErr

}, writerRetry, w.settings.idempotent, withOperation("WriteObject"), withBucket(w.bucket), withObject(w.attrs.Name)))

w.setError(w.streamResult)

close(w.donec)

}()

return w, nil

}

// gRPCWriter is a wrapper around the gRPC client-stream API that manages

// sending chunks of data provided by the user over the stream.

type gRPCWriter struct {

preRunCtx context.Context

c *grpcStorageClient

settings *settings

bucket string

attrs *ObjectAttrs

conds *Conditions

spec *storagepb.WriteObjectSpec

encryptionKey []byte

setError func(error)

progress func(int64)

setObj func(*ObjectAttrs)

setSize func(int64)

setTakeoverOffset func(int64)

fullObjectChecksum uint32

appendFinalCRC32C uint32

sendAppendFinalCRC32C bool

flushSupported bool

sendCRC32C bool

disableAutoChecksum bool

forceOneShot bool

forceEmptyContentType bool

append bool

appendGen int64

finalizeOnClose bool

buf []byte

chunkSize int

// A writeQuantum is the largest quantity of data which can be sent to the

// service in a single message.

writeQuantum int

lastSegmentStart int

sendableUnits int

bufUnsentIdx int

bufFlushedIdx int

bufBaseOffset int64

chunkRetryDeadline time.Duration

abandonRetriesTime time.Time

attempts int

lastErr error

streamSender gRPCBidiWriteBufferSender

// Communication from the user goroutine to the stream management goroutines

writesChan chan gRPCWriterCommand

currentCommand gRPCWriterCommand

forcedStreamResult error

streamResult error

donec chan struct{}

}

func (w *gRPCWriter) pickBufferSender() gRPCBidiWriteBufferSender {

if w.append {

// Appendable object semantics

if w.appendGen >= 0 {

return w.newGRPCAppendTakeoverWriteBufferSender()

}

return w.newGRPCAppendableObjectBufferSender()

}

if w.forceOneShot {

// One shot semantics - no progress reports

w.progress = func(int64) {}

return w.newGRPCOneshotBidiWriteBufferSender()

}

// Resumable write semantics

return w.newGRPCResumableBidiWriteBufferSender()

}

// sendBufferToTarget uses cs to send slices of buf, which starts at baseOffset

// bytes into the object. Slices are sent until flushAt bytes have sent, in

// which case the final request is a flush, or until len(buf) < w.writeQuantum.

//

// handleCompletion is called for any completions that arrive during sends.

//

// Returns the last byte offset sent. Returns true if all desired requests were

// delivered, and false if cs.completions was closed before all requests could

// be delivered.

func (w *gRPCWriter) sendBufferToTarget(cs gRPCWriterCommandHandleChans, buf []byte, baseOffset int64, flushAt int, handleCompletion func(gRPCBidiWriteCompletion)) (int64, bool) {

sent := 0

if len(buf) > flushAt {

buf = buf[:flushAt]

}

for len(buf) > 0 && (len(buf) >= w.writeQuantum || len(buf) >= flushAt-sent) {

q := w.writeQuantum

if flushAt-sent < w.writeQuantum {

q = flushAt - sent

}

req := gRPCBidiWriteRequest{

buf: buf[:q],

offset: baseOffset + int64(sent),

flush: q == flushAt-sent,

}

if !cs.deliverRequestUnlessCompleted(req, handleCompletion) {

return baseOffset + int64(sent), false

}

buf = buf[q:]

sent += q

}

return baseOffset + int64(sent), true

}

func (w *gRPCWriter) handleCompletion(c gRPCBidiWriteCompletion) {

if c.resource != nil {

w.setObj(newObjectFromProto(c.resource))

}

// Already handled this completion

if c.flushOffset <= w.bufBaseOffset+int64(w.bufFlushedIdx) {

return

}

w.bufFlushedIdx = int(c.flushOffset - w.bufBaseOffset)

if w.bufFlushedIdx >= len(w.buf) {

// We can clear w.buf

w.bufBaseOffset = c.flushOffset

w.bufUnsentIdx = 0

w.bufFlushedIdx = 0

w.buf = w.buf[:0]

}

w.setSize(c.flushOffset)

w.progress(c.flushOffset)

}

func (w *gRPCWriter) withCommandRetryDeadline(f func() error) error {

w.abandonRetriesTime = time.Now().Add(w.chunkRetryDeadline)

err := f()

if err == nil {

w.abandonRetriesTime = time.Time{}

}

return err

}

// Gather write commands before starting the actual write. Returns nil if the

// stream should be started, and an error otherwise.

func (w *gRPCWriter) gatherFirstBuffer() error {

if w.append && w.appendGen >= 0 {

// For takeovers, kick off the stream immediately since we need to know the

// takeover offset to issue writes.

return nil

}

for cmd := range w.writesChan {

switch v := cmd.(type) {

case *gRPCWriterCommandWrite:

// If zero-copy one-shot is requested, OR the payload is larger than the buffer,

// bypass buffering entirely and hand off to the writeLoop immediately.

if w.forceOneShot || len(w.buf)+len(v.p) > w.chunkSize {

w.currentCommand = cmd

return nil

}

// Otherwise, lazily allocate and stage the small write (normal buffered path)

if w.buf == nil {

w.buf = make([]byte, 0, w.chunkSize)

}

// We have not started sending yet, and we can stage all data without

// starting a send. Compare against w.chunkSize instead of

// w.writeQuantum: that way we can perform a oneshot upload for objects

// which fit in one chunk, even though we will cut the request into

// w.writeQuantum units when we do start sending.

origLen := len(w.buf)

w.buf = w.buf[:origLen+len(v.p)]

copy(w.buf[origLen:], v.p)

close(v.done)

break

case *gRPCWriterCommandClose:

// If we get here, data (if any) fits in w.buf, so we can force oneshot.

w.forceOneShot = true

w.currentCommand = cmd

// No need to start sending if v.err is not nil.

return v.err

default:

// Have to start sending!

w.currentCommand = cmd

return nil

}

}

// Nothing should ever close w.writesChan, so we should never get here

return errors.New("storage.Writer: unexpectedly closed w.writesChan")

}

func (w *gRPCWriter) writeLoop(ctx context.Context) error {

w.attempts++

// Return an error if we've been waiting for a single operation for too long.

if !w.abandonRetriesTime.IsZero() && time.Now().After(w.abandonRetriesTime) {

return fmt.Errorf("storage: retry deadline of %s reached after %v attempts; last error: %w", w.chunkRetryDeadline, w.attempts, w.lastErr)

}

// Allow each request in w.buf to be sent and result in a completion without

// blocking.

requests := make(chan gRPCBidiWriteRequest, w.sendableUnits)

completions := make(chan gRPCBidiWriteCompletion, w.sendableUnits)

// Only one request ack will be outstanding at a time.

requestAcks := make(chan struct{}, 1)

chcs := gRPCWriterCommandHandleChans{requests, requestAcks, completions}

bscs := gRPCBufSenderChans{requests, requestAcks, completions}

ctx, cancel := context.WithCancel(ctx)

defer cancel()

w.streamSender.connect(ctx, bscs, w.settings.gax...)

// Drain any initial completions (like QueryWriteStatus results).

Loop:

for {

select {

case c, ok := <-completions:

if !ok {

return w.streamSender.err()

}

w.handleCompletion(c)

default:

break Loop

}

}

if w.bufFlushedIdx > 0 {

copy(w.buf, w.buf[w.bufFlushedIdx:])

w.buf = w.buf[:len(w.buf)-w.bufFlushedIdx]

w.bufBaseOffset += int64(w.bufFlushedIdx)

w.bufUnsentIdx -= w.bufFlushedIdx

if w.bufUnsentIdx < 0 {

w.bufUnsentIdx = 0

}

w.bufFlushedIdx = -1

}

// Send any full quantum in w.buf, possibly including a flush

if err := w.withCommandRetryDeadline(func() error {

sentOffset, ok := w.sendBufferToTarget(chcs, w.buf, w.bufBaseOffset, cap(w.buf),

w.handleCompletion)

if !ok {

return w.streamSender.err()

}

w.bufUnsentIdx = int(sentOffset - w.bufBaseOffset)

// We may have observed a completion that is after all of w.buf if we also

// have a write command in w.currentCommand which sent a flush, but failed

// before the completion could be delivered.

if w.bufUnsentIdx < 0 {

w.bufUnsentIdx = 0

}

return nil

}); err != nil {

return err

}

err := func() error {

for {

if w.currentCommand != nil {

if err := w.withCommandRetryDeadline(func() error {

return w.currentCommand.handle(w, chcs)

}); err != nil {

return err

}

w.currentCommand = nil

}

select {

case c, ok := <-completions:

if !ok {

return w.streamSender.err()

}

w.handleCompletion(c)

case cmd, ok := <-w.writesChan:

if !ok {

// Nothing should ever close w.writesChan, so we should never get here

return errors.New("storage.Writer: unexpectedly closed w.writesChan")

}

w.currentCommand = cmd

}

}

}()

if err == nil {

err = errors.New("storage.Writer: unexpected nil error from write loop")

}

var closeErr *gRPCWriterCommandClose

if !errors.As(err, &closeErr) {

// Not a shutdown.

return err

}

if closeErr.err == nil {

// Clean shutdown. Send any remaining tail.

req := gRPCBidiWriteRequest{

buf: w.buf[w.bufUnsentIdx:],

offset: w.bufBaseOffset + int64(w.bufUnsentIdx),

flush: true,

finishWrite: true,

}

if err := w.withCommandRetryDeadline(func() error {

if !chcs.deliverRequestUnlessCompleted(req, w.handleCompletion) {

return w.streamSender.err()

}

return nil

}); err != nil {

return err

}

} else {

// Unclean shutdown. Cancel the context so we clean up expeditiously.

cancel()

}

close(requests)

for c := range completions {

w.handleCompletion(c)

}

if closeErr.err == nil {

return w.streamSender.err()

}

return closeErr.err

}

// gRPCWriterCommandHandleChans contains the channels that a gRPCWriterCommand

// implementation must use to send requests and get notified of completions.

// Requests are delivered on a write-only channel, request acks and completions

// arrive on read-only channels.

type gRPCWriterCommandHandleChans struct {

requests chan<- gRPCBidiWriteRequest

requestAcks <-chan struct{}

completions <-chan gRPCBidiWriteCompletion

}

// gRPCBufSenderChans contains the channels that a gRPCBidiWriteBufferSender

// must use to get notified of requests and deliver completions. Requests arrive

// on a read-only channel, request acks and completions are delivered on

// write-only channels.

type gRPCBufSenderChans struct {

requests <-chan gRPCBidiWriteRequest

requestAcks chan<- struct{}

completions chan<- gRPCBidiWriteCompletion

}

// deliverRequestUnlessCompleted submits req to cs.requests, unless

// cs.completions is closed first. If a completion arrives before the request is

// enqueued, handleCompletion is called.

//

// Returns true if request was successfully enqueued, and false if completions

// was closed first.

func (cs gRPCWriterCommandHandleChans) deliverRequestUnlessCompleted(req gRPCBidiWriteRequest, handleCompletion func(gRPCBidiWriteCompletion)) bool {

for {

select {

case cs.requests <- req:

return true

case c, ok := <-cs.completions:

if !ok {

return false

}

handleCompletion(c)

}

}

}

// gRPCWriterCommand represents an operation on a gRPCWriter

type gRPCWriterCommand interface {

// handle applies the command to a gRPCWriter.

//

// Implementations may return an error. In that case, the command may be

// retried with a new gRPCWriterCommandHandleChans instance.

handle(*gRPCWriter, gRPCWriterCommandHandleChans) error

}

type gRPCWriterCommandWrite struct {

p []byte

done chan struct{}

initialOffset int64

hasStarted bool

closeOnce sync.Once

}

func (c *gRPCWriterCommandWrite) handle(w *gRPCWriter, cs gRPCWriterCommandHandleChans) error {

if len(c.p) == 0 {

// No data to write.

c.markDone()

return nil

}

if !c.hasStarted {

c.initialOffset = w.bufBaseOffset + int64(len(w.buf))

c.hasStarted = true

} else {

// Retrying this command; check if server has persisted some bytes of this command's payload.

bytesPersisted := w.bufBaseOffset - c.initialOffset

if bytesPersisted > 0 {

if int64(len(c.p)) < bytesPersisted {

bytesPersisted = int64(len(c.p))

}

c.p = c.p[bytesPersisted:]

c.initialOffset = w.bufBaseOffset

if len(c.p) == 0 {

c.markDone()

return nil

}

}

}

// Zero-Copy send.

if w.forceOneShot {

err := c.zeroCopyWrite(w, cs)

if err != nil {

return err

}

// If zeroCopyWrite returns without error, the write is done.

return nil

}

if w.buf == nil {

w.buf = make([]byte, 0, w.chunkSize)

}

wblen := len(w.buf)

allKnownBytes := wblen + len(c.p)

fullBufs := allKnownBytes / cap(w.buf)

partialBuf := allKnownBytes % cap(w.buf)

if partialBuf == 0 {

// If we would exactly fill some number of cap(w.buf) units, we don't need

// to block on the flush for the last one. We know that c.p is not empty, so

// allKnownBytes is not 0 and therefore if partialBuf is 0, fullBufs is not

// 0.

fullBufs--

partialBuf = cap(w.buf)

}

if fullBufs == 0 {

// Everything fits in w.buf. Copy in and send from there.

w.buf = w.buf[:allKnownBytes]

copied := copy(w.buf[wblen:], c.p)

// Now that it's in w.buf, clear it from the command in case we retry.

c.p = c.p[copied:]

sending := w.buf[w.bufUnsentIdx:]

sentOffset, ok := w.sendBufferToTarget(cs, sending, w.bufBaseOffset+int64(w.bufUnsentIdx), cap(sending),

w.handleCompletion)

if !ok {

return w.streamSender.err()

}

w.bufUnsentIdx = int(sentOffset - w.bufBaseOffset)

c.markDone()

return nil

}

// We have at least one full buffer, followed by a partial. The first full

// buffer is the interesting one. We don't actually have to copy all of c.p

// in: we can send from it in place, except for any partial quantum at the

// tail of w.buf. Send that quantum...

toNextWriteQuantum := func() int {

if wblen > w.lastSegmentStart {

return cap(w.buf) - wblen

}

if wblen%w.writeQuantum == 0 {

return 0

}

return w.writeQuantum - (wblen % w.writeQuantum)

}()

w.buf = w.buf[:wblen+toNextWriteQuantum]

copied := copy(w.buf[wblen:], c.p)

c.p = c.p[copied:]

c.initialOffset += int64(copied)

firstFullBufFromCmd := cap(w.buf) - len(w.buf)

sending := w.buf[w.bufUnsentIdx:]

sentOffset, ok := w.sendBufferToTarget(cs, sending, w.bufBaseOffset+int64(w.bufUnsentIdx), cap(sending),

w.handleCompletion)

if !ok {

return w.streamSender.err()

}

// ...then send the prefix of c.p which could fill w.buf

cmdBaseOffset := w.bufBaseOffset + int64(len(w.buf))

cmdBuf := c.p

trimCommandBuf := func(cmp gRPCBidiWriteCompletion) {

w.handleCompletion(cmp)

// After a completion, keep c.p up to date with w.buf's tail.

bufTail := w.bufBaseOffset + int64(len(w.buf))

if bufTail <= cmdBaseOffset {

return

}

trim := int(bufTail - cmdBaseOffset)

if len(c.p) < trim {

trim = len(c.p)

}

c.p = c.p[trim:]

c.initialOffset += int64(trim)

cmdBaseOffset = bufTail

}

offset := cmdBaseOffset

sentOffset, ok = w.sendBufferToTarget(cs, cmdBuf, offset, firstFullBufFromCmd,

trimCommandBuf)

if !ok {

return w.streamSender.err()

}

cmdBuf = cmdBuf[int(sentOffset-offset):]

offset = sentOffset

// Remaining full buffers can be satisfied entirely from cmdBuf with no copies.

for i := 0; i < fullBufs-1; i++ {

sentOffset, ok = w.sendBufferToTarget(cs, cmdBuf, offset, cap(w.buf),

trimCommandBuf)

if !ok {

return w.streamSender.err()

}

cmdBuf = cmdBuf[int(sentOffset-offset):]

offset = sentOffset

}

// Send the last partial buffer. We need to flush to offset before we can copy

// the rest of cmdBuf into w.buf and complete this command.

sentOffset, ok = w.sendBufferToTarget(cs, cmdBuf, offset, cap(w.buf),

trimCommandBuf)

if !ok {

return w.streamSender.err()

}

// Finally, we need the sender to ack to let us know c.p can be released.

if !cs.deliverRequestUnlessCompleted(gRPCBidiWriteRequest{requestAck: true}, trimCommandBuf) {

return w.streamSender.err()

}

ackOutstanding := true

for ackOutstanding || (w.bufBaseOffset+int64(w.bufFlushedIdx)) < offset {

select {

case cmp, ok := <-cs.completions:

if !ok {

return w.streamSender.err()

}

trimCommandBuf(cmp)

case <-cs.requestAcks:

ackOutstanding = false

}

}

toCopyIn := cmdBuf[int(w.bufBaseOffset-offset):]

w.buf = w.buf[:len(toCopyIn)]

copy(w.buf, toCopyIn)

w.bufUnsentIdx = int(sentOffset - w.bufBaseOffset)

c.markDone()

return nil

}

func (c *gRPCWriterCommandWrite) zeroCopyWrite(w *gRPCWriter, cs gRPCWriterCommandHandleChans) error {

// Pre-emptively get the context channel to avoid closure overhead in the loop.

ctxDone := w.preRunCtx.Done()

// sendBufferToTarget handles the quantum breakdown.

newOffset, ok := w.sendBufferToTarget(cs, c.p, w.bufBaseOffset, len(c.p), w.handleCompletion)

if !ok {

return w.streamSender.err()

}

// Request an ack from the sender goroutine to ensure the buffer has been

// dispatched to gRPC and is safe for the user to reuse.

if !cs.deliverRequestUnlessCompleted(gRPCBidiWriteRequest{requestAck: true}, w.handleCompletion) {

return w.streamSender.err()

}

ackOutstanding := true

// Wait for server acknowledgement and sender transmissions to enable incremental progress.

for ackOutstanding || w.bufBaseOffset < newOffset {

select {

case completion, ok := <-cs.completions:

if !ok {

return w.streamSender.err()

}

w.handleCompletion(completion)

case <-cs.requestAcks:

ackOutstanding = false

case <-ctxDone:

return w.preRunCtx.Err()

}

}

c.p = nil

c.markDone()

return nil

}

// Helper to ensure we don't close done twice and keep the main logic clean.

func (c *gRPCWriterCommandWrite) markDone() {

c.closeOnce.Do(func() { close(c.done) })

}

type gRPCWriterCommandFlush struct {

done chan int64

}

func (c *gRPCWriterCommandFlush) handle(w *gRPCWriter, cs gRPCWriterCommandHandleChans) error {

flushTarget := w.bufBaseOffset + int64(len(w.buf))

// We know that there are at most w.writeQuantum bytes in

// w.buf[w.bufUnsentIdx:], because we send anything more inline when handling

// a write.

req := gRPCBidiWriteRequest{

buf: w.buf[w.bufUnsentIdx:],

offset: w.bufBaseOffset + int64(w.bufUnsentIdx),

flush: true,

finishWrite: false,

}

if !cs.deliverRequestUnlessCompleted(req, w.handleCompletion) {

return w.streamSender.err()

}

// Successful flushes will clear w.buf.

for (w.bufBaseOffset + int64(w.bufFlushedIdx)) < flushTarget {

c, ok := <-cs.completions

if !ok {

// Stream failure

return w.streamSender.err()

}

w.handleCompletion(c)

}

// handleCompletion has cleared w.buf and updated w.bufUnsentIdx by now.

c.done <- flushTarget

return nil

}

type gRPCWriterCommandClose struct {

err error

}

func (e *gRPCWriterCommandClose) Error() string {

return e.err.Error()

}

func (c *gRPCWriterCommandClose) handle(w *gRPCWriter, cs gRPCWriterCommandHandleChans) error {

// N.B. c is not nil, even if c.err is nil!

return c

}

// Detect content type using bytes first from baseBuf, then from pendingBuf if

// there are not enough bytes in baseBuf.

func (w *gRPCWriter) detectContentType() string {

wblen := len(w.buf)

// If the current command is a write, we want to be able to update it in

// place. If the

cmdbuf := &([]byte{})

if c, ok := w.currentCommand.(*gRPCWriterCommandWrite); ok {

cmdbuf = &c.p

}

if wblen == 0 {

// Use the command in place

return http.DetectContentType(*cmdbuf)

}

if wblen >= w.writeQuantum {

// Use w.buf in place

return http.DetectContentType(w.buf)

}

// We need to put bytes from the command onto w.buf. Try to fill a

// writeQuantum since we'll have to do that in order to send, anyway.

newSz := w.writeQuantum

if wblen+len(*cmdbuf) < newSz {

newSz = wblen + len(*cmdbuf)

}

w.buf = w.buf[:newSz]

copied := copy(w.buf[wblen:], *cmdbuf)

*cmdbuf = (*cmdbuf)[copied:]

return http.DetectContentType(w.buf)

}

type gRPCBidiWriteRequest struct {

buf []byte

offset int64

flush bool

finishWrite bool

// If requestAck is true, no other message fields may be set. Buffer senders

// must ack on the requestAcks channel if all prior messages on the requests

// channel have been delivered to gRPC.

requestAck bool

}

type gRPCBidiWriteCompletion struct {

flushOffset int64

resource *storagepb.Object

}

func completion(r *storagepb.BidiWriteObjectResponse) *gRPCBidiWriteCompletion {

switch c := r.WriteStatus.(type) {

case *storagepb.BidiWriteObjectResponse_PersistedSize:

return &gRPCBidiWriteCompletion{flushOffset: c.PersistedSize}

case *storagepb.BidiWriteObjectResponse_Resource:

return &gRPCBidiWriteCompletion{flushOffset: c.Resource.GetSize(), resource: c.Resource}

default:

return nil

}

}

// Server contract expects full object checksum to be sent only on first or last write.

// Checksums of full object are already being sent on first Write during initialization of sender.

// Send objectChecksums only on final request and nil in other cases.

func bidiWriteObjectRequest(r gRPCBidiWriteRequest, bufChecksum *uint32, objectChecksums *storagepb.ObjectChecksums) *storagepb.BidiWriteObjectRequest {

var data *storagepb.BidiWriteObjectRequest_ChecksummedData

if r.buf != nil {

data = &storagepb.BidiWriteObjectRequest_ChecksummedData{

ChecksummedData: &storagepb.ChecksummedData{

Content: r.buf,

Crc32C: bufChecksum,

},

}

}

req := &storagepb.BidiWriteObjectRequest{

Data: data,

WriteOffset: r.offset,

FinishWrite: r.finishWrite,

Flush: r.flush,

StateLookup: r.flush,

ObjectChecksums: objectChecksums,

}

return req

}

type getObjectChecksumsParams struct {

sendCRC32C bool

disableAutoChecksum bool

objectAttrs *ObjectAttrs

fullObjectChecksum func() *uint32

finishWrite bool

append bool

}

// getObjectChecksums determines what checksum information to include in the final

// gRPC request

//

// function returns a populated ObjectChecksums only when finishWrite is true

// If CRC32C is disabled, it returns the user-provided checksum if available.

// If CRC32C is enabled, it returns the user-provided checksum if available,

// or the computed checksum of the entire object.

func getObjectChecksums(params *getObjectChecksumsParams) *storagepb.ObjectChecksums {

if !params.finishWrite {

return nil

}

// For append operations, send user's final append checksum on last write op if available.

// Auto checksum is not supported for appendable writes.

var crc32c *uint32

if params.fullObjectChecksum != nil {

crc32c = params.fullObjectChecksum()

}

if params.append && crc32c != nil {

return &storagepb.ObjectChecksums{Crc32C: crc32c}

}

// send user's checksum on last write op if available

if params.sendCRC32C || (params.objectAttrs != nil && params.objectAttrs.MD5 != nil) {

return toProtoChecksums(params.sendCRC32C, params.objectAttrs)

}

if params.append || params.disableAutoChecksum || params.fullObjectChecksum == nil {

return nil

}

if crc32c != nil {

return &storagepb.ObjectChecksums{Crc32C: crc32c}

}

return nil

}