Skip to content

feat: add default checksums for appendable writer (excludes appendable takeover writer) #13379

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
krishnamd-jkp merged 8 commits into from
Dec 15, 2025
Merged

feat: add default checksums for appendable writer (excludes appendable takeover writer) #13379

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
8e53769
feat: add default checksums for appendable writer
krishnamd-jkp Nov 19, 2025
30426a7
refactor checksum attrs and add integration tests
krishnamd-jkp Nov 24, 2025
8677c5e
resolve comments
krishnamd-jkp Nov 25, 2025
6f26708
resolve comments
krishnamd-jkp Dec 1, 2025
3a677d3
resolve comments
krishnamd-jkp Dec 8, 2025
befb811
move CPU ovehrhead docs to DisableAutoChecksum
krishnamd-jkp Dec 8, 2025
2a88960
replace string comp with error code
krishnamd-jkp Dec 8, 2025
510fb38
remove checksum attrs
krishnamd-jkp Dec 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions storage/doc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -407,6 +407,10 @@ roles which must be enabled in order to do the export successfully. To
disable this export, you can use the [WithDisabledClientMetrics] client
option.

The gRPC client automatically computes and sends CRC32C checksums for uploads using [Writer],
which provides an additional layer of data integrity validation when compared to the HTTP client.
This behavior can optionally be disabled by using [Writer.DisableAutoChecksum].

# Storage Control API

Certain control plane and long-running operations for Cloud Storage (including Folder
Expand Down
105 changes: 68 additions & 37 deletions storage/grpc_writer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -819,11 +819,12 @@ func bidiWriteObjectRequest(r gRPCBidiWriteRequest, bufChecksum *uint32, objectC
}

type getObjectChecksumsParams struct {
fullObjectChecksum func() uint32
finishWrite bool
sendCRC32C bool
disableAutoChecksum bool
attrs *ObjectAttrs
objectAttrs *ObjectAttrs
fullObjectChecksum func() uint32
finishWrite bool
takeoverWriter bool
}

// getObjectChecksums determines what checksum information to include in the final
Expand All @@ -840,9 +841,10 @@ func getObjectChecksums(params *getObjectChecksumsParams) *storagepb.ObjectCheck

// send user's checksum on last write op if available
if params.sendCRC32C {
return toProtoChecksums(params.sendCRC32C, params.attrs)
return toProtoChecksums(params.sendCRC32C, params.objectAttrs)
}
if params.disableAutoChecksum {
// TODO(b/461982277): Enable checksum validation for appendable takeover writer gRPC
if params.disableAutoChecksum || params.takeoverWriter {
return nil
}
return &storagepb.ObjectChecksums{
Expand Down Expand Up @@ -879,8 +881,11 @@ type gRPCOneshotBidiWriteBufferSender struct {
firstMessage *storagepb.BidiWriteObjectRequest
streamErr error

checksumSettings func() (bool, bool, *ObjectAttrs)
fullObjectChecksum func() uint32
// Checksum related settings.
sendCRC32C bool
disableAutoChecksum bool
objectAttrs *ObjectAttrs
fullObjectChecksum func() uint32
}

func (w *gRPCWriter) newGRPCOneshotBidiWriteBufferSender() *gRPCOneshotBidiWriteBufferSender {
Expand All @@ -894,9 +899,9 @@ func (w *gRPCWriter) newGRPCOneshotBidiWriteBufferSender() *gRPCOneshotBidiWrite
CommonObjectRequestParams: toProtoCommonObjectRequestParams(w.encryptionKey),
ObjectChecksums: toProtoChecksums(w.sendCRC32C, w.attrs),
},
checksumSettings: func() (bool, bool, *ObjectAttrs) {
return w.sendCRC32C, w.disableAutoChecksum, w.attrs
},
sendCRC32C: w.sendCRC32C,
disableAutoChecksum: w.disableAutoChecksum,
objectAttrs: w.attrs,
fullObjectChecksum: func() uint32 {
return w.fullObjectChecksum
},
Expand Down Expand Up @@ -939,17 +944,16 @@ func (s *gRPCOneshotBidiWriteBufferSender) connect(ctx context.Context, cs gRPCB
continue
}

sendCrc32C, disableAutoChecksum, attrs := s.checksumSettings()
var bufChecksum *uint32
if !disableAutoChecksum {
if !s.disableAutoChecksum {
bufChecksum = proto.Uint32(crc32.Checksum(r.buf, crc32cTable))
}
objectChecksums := getObjectChecksums(&getObjectChecksumsParams{
sendCRC32C: s.sendCRC32C,
objectAttrs: s.objectAttrs,
fullObjectChecksum: s.fullObjectChecksum,
disableAutoChecksum: s.disableAutoChecksum,
finishWrite: r.finishWrite,
sendCRC32C: sendCrc32C,
disableAutoChecksum: disableAutoChecksum,
attrs: attrs,
})
req := bidiWriteObjectRequest(r, bufChecksum, objectChecksums)

Expand Down Expand Up @@ -996,8 +1000,11 @@ type gRPCResumableBidiWriteBufferSender struct {
startWriteRequest *storagepb.StartResumableWriteRequest
upid string

checksumSettings func() (bool, bool, *ObjectAttrs)
fullObjectChecksum func() uint32
// Checksum related settings.
sendCRC32C bool
disableAutoChecksum bool
objectAttrs *ObjectAttrs
fullObjectChecksum func() uint32

streamErr error
}
Expand All @@ -1011,9 +1018,9 @@ func (w *gRPCWriter) newGRPCResumableBidiWriteBufferSender() *gRPCResumableBidiW
CommonObjectRequestParams: toProtoCommonObjectRequestParams(w.encryptionKey),
ObjectChecksums: toProtoChecksums(w.sendCRC32C, w.attrs),
},
checksumSettings: func() (bool, bool, *ObjectAttrs) {
return w.sendCRC32C, w.disableAutoChecksum, w.attrs
},
sendCRC32C: w.sendCRC32C,
disableAutoChecksum: w.disableAutoChecksum,
objectAttrs: w.attrs,
fullObjectChecksum: func() uint32 {
return w.fullObjectChecksum
},
Expand Down Expand Up @@ -1076,17 +1083,16 @@ func (s *gRPCResumableBidiWriteBufferSender) connect(ctx context.Context, cs gRP
continue
}

sendCrc32C, disableAutoChecksum, attrs := s.checksumSettings()
var bufChecksum *uint32
if !disableAutoChecksum {
if !s.disableAutoChecksum {
bufChecksum = proto.Uint32(crc32.Checksum(r.buf, crc32cTable))
}
objectChecksums := getObjectChecksums(&getObjectChecksumsParams{
sendCRC32C: s.sendCRC32C,
objectAttrs: s.objectAttrs,
fullObjectChecksum: s.fullObjectChecksum,
disableAutoChecksum: s.disableAutoChecksum,
finishWrite: r.finishWrite,
sendCRC32C: sendCrc32C,
disableAutoChecksum: disableAutoChecksum,
attrs: attrs,
})
req := bidiWriteObjectRequest(r, bufChecksum, objectChecksums)

Expand Down Expand Up @@ -1142,12 +1148,18 @@ type gRPCAppendBidiWriteBufferSender struct {
bucket string
routingToken *string

firstMessage *storagepb.BidiWriteObjectRequest

objectChecksums *storagepb.ObjectChecksums
firstMessage *storagepb.BidiWriteObjectRequest
finalizeOnClose bool
objResource *storagepb.Object

// Checksum related settings.
sendCRC32C bool
disableAutoChecksum bool
objectAttrs *ObjectAttrs
fullObjectChecksum func() uint32

takeoverWriter bool

streamErr error
}

Expand All @@ -1164,8 +1176,13 @@ func (w *gRPCWriter) newGRPCAppendableObjectBufferSender() *gRPCAppendBidiWriteB
},
CommonObjectRequestParams: toProtoCommonObjectRequestParams(w.encryptionKey),
},
objectChecksums: toProtoChecksums(w.sendCRC32C, w.attrs),
finalizeOnClose: w.finalizeOnClose,
finalizeOnClose: w.finalizeOnClose,
sendCRC32C: w.sendCRC32C,
disableAutoChecksum: w.disableAutoChecksum,
objectAttrs: w.attrs,
fullObjectChecksum: func() uint32 {
return w.fullObjectChecksum
},
}
}

Expand Down Expand Up @@ -1278,8 +1295,14 @@ func (w *gRPCWriter) newGRPCAppendTakeoverWriteBufferSender() *gRPCAppendTakeove
AppendObjectSpec: writeObjectSpecAsAppendObjectSpec(w.spec, w.appendGen),
},
},
objectChecksums: toProtoChecksums(w.sendCRC32C, w.attrs),
finalizeOnClose: w.finalizeOnClose,
finalizeOnClose: w.finalizeOnClose,
takeoverWriter: true,
sendCRC32C: w.sendCRC32C,
disableAutoChecksum: w.disableAutoChecksum,
objectAttrs: w.attrs,
fullObjectChecksum: func() uint32 {
return w.fullObjectChecksum
},
},
takeoverReported: false,
handleTakeoverCompletion: func(c gRPCBidiWriteCompletion) {
Expand Down Expand Up @@ -1409,12 +1432,20 @@ func (s *gRPCAppendBidiWriteBufferSender) send(stream storagepb.Storage_BidiWrit
flush: flush,
finishWrite: finalizeObject,
}
// TODO(b/453869602): implement default checksumming for appendable writes
req := bidiWriteObjectRequest(r, nil, nil)
if finalizeObject {
// appendable objects pass checksums on the finalize message only
req.ObjectChecksums = s.objectChecksums

var bufChecksum *uint32
if !s.disableAutoChecksum {
bufChecksum = proto.Uint32(crc32.Checksum(r.buf, crc32cTable))
}
objectChecksums := getObjectChecksums(&getObjectChecksumsParams{
sendCRC32C: s.sendCRC32C,
objectAttrs: s.objectAttrs,
fullObjectChecksum: s.fullObjectChecksum,
disableAutoChecksum: s.disableAutoChecksum,
finishWrite: finalizeObject,
takeoverWriter: s.takeoverWriter,
})
req := bidiWriteObjectRequest(r, bufChecksum, objectChecksums)
if sendFirstMessage {
proto.Merge(req, s.firstMessage)
}
Expand Down
15 changes: 12 additions & 3 deletions storage/grpc_writer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ func TestGetObjectChecksums(t *testing.T) {
fullObjectChecksum func() uint32
finishWrite bool
sendCRC32C bool
takeoverWriter bool
disableAutoChecksum bool
attrs *ObjectAttrs
want *storagepb.ObjectChecksums
Expand Down Expand Up @@ -73,16 +74,24 @@ func TestGetObjectChecksums(t *testing.T) {
Crc32C: proto.Uint32(456),
},
},
// TODO(b/461982277): remove this testcase once checksums for takeover writer is implemented
{
name: "takeover writer should return nil",
finishWrite: true,
takeoverWriter: true,
want: nil,
},
}

for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := getObjectChecksums(&getObjectChecksumsParams{
disableAutoChecksum: tt.disableAutoChecksum,
sendCRC32C: tt.sendCRC32C,
objectAttrs: tt.attrs,
fullObjectChecksum: tt.fullObjectChecksum,
finishWrite: tt.finishWrite,
sendCRC32C: tt.sendCRC32C,
disableAutoChecksum: tt.disableAutoChecksum,
attrs: tt.attrs,
takeoverWriter: tt.takeoverWriter,
})
if !proto.Equal(got, tt.want) {
t.Errorf("getObjectChecksums() = %v, want %v", got, tt.want)
Expand Down
Loading
Loading