Skip to content

Commit 1b2f5af

Browse files
krishnamd-jkpkrishnamd-jkp
authored
feat(storage): Accept CRC32C for appendable objects (#20104)
1 parent 1c073b8 commit 1b2f5af

6 files changed

Lines changed: 343 additions & 33 deletions

File tree

‎storage/grpc_writer.go‎

Lines changed: 51 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,8 @@ func (w *gRPCWriter) Write(p []byte) (n int, err error) {
5858
// Skip checksum calculation if user configures MD5 or CRC32C themselves.
5959
if !w.disableAutoChecksum &&
6060
!w.sendCRC32C &&
61-
!md5Provided {
61+
!md5Provided &&
62+
!w.append {
6263
w.fullObjectChecksum = crc32.Update(w.fullObjectChecksum, crc32cTable, p)
6364
}
6465
// write command successfully delivered to sender. We no longer own cmd.
@@ -109,6 +110,11 @@ func (w *gRPCWriter) CloseWithError(err error) error {
109110
return nil
110111
}
111112

113+
func (w *gRPCWriter) setAppendFinalCRC32C(sendAppendFinalCRC32C bool, c uint32) {
114+
w.sendAppendFinalCRC32C = sendAppendFinalCRC32C
115+
w.appendFinalCRC32C = c
116+
}
117+
112118
func (c *grpcStorageClient) OpenWriter(params *openWriterParams, opts ...storageOption) (internalWriter, error) {
113119
if params.attrs.Retention != nil {
114120
// TO-DO: remove once ObjectRetention is available - see b/308194853
@@ -259,7 +265,9 @@ type gRPCWriter struct {
259265
setSize func(int64)
260266
setTakeoverOffset func(int64)
261267

262-
fullObjectChecksum uint32
268+
fullObjectChecksum uint32
269+
appendFinalCRC32C uint32
270+
sendAppendFinalCRC32C bool
263271

264272
flushSupported bool
265273
sendCRC32C bool
@@ -948,9 +956,9 @@ type getObjectChecksumsParams struct {
948956
sendCRC32C bool
949957
disableAutoChecksum bool
950958
objectAttrs *ObjectAttrs
951-
fullObjectChecksum func() uint32
959+
fullObjectChecksum func() *uint32
952960
finishWrite bool
953-
takeoverWriter bool
961+
append bool
954962
}
955963

956964
// getObjectChecksums determines what checksum information to include in the final
@@ -965,20 +973,30 @@ func getObjectChecksums(params *getObjectChecksumsParams) *storagepb.ObjectCheck
965973
return nil
966974
}
967975

976+
// For append operations, send user's final append checksum on last write op if available.
977+
// Auto checksum is not supported for appendable writes.
978+
var crc32c *uint32
979+
if params.fullObjectChecksum != nil {
980+
crc32c = params.fullObjectChecksum()
981+
}
982+
983+
if params.append && crc32c != nil {
984+
return &storagepb.ObjectChecksums{Crc32C: crc32c}
985+
}
986+
968987
// send user's checksum on last write op if available
969988
if params.sendCRC32C || (params.objectAttrs != nil && params.objectAttrs.MD5 != nil) {
970989
return toProtoChecksums(params.sendCRC32C, params.objectAttrs)
971990
}
972-
// TODO(b/461982277): Enable checksum validation for appendable takeover writer gRPC
973-
if params.disableAutoChecksum || params.takeoverWriter {
974-
return nil
975-
}
976-
if params.fullObjectChecksum == nil {
991+
992+
if params.append || params.disableAutoChecksum || params.fullObjectChecksum == nil {
977993
return nil
978994
}
979-
return &storagepb.ObjectChecksums{
980-
Crc32C: proto.Uint32(params.fullObjectChecksum()),
995+
996+
if crc32c != nil {
997+
return &storagepb.ObjectChecksums{Crc32C: crc32c}
981998
}
999+
return nil
9821000
}
9831001

9841002
type gRPCBidiWriteBufferSender interface {
@@ -1014,7 +1032,7 @@ type gRPCOneshotBidiWriteBufferSender struct {
10141032
sendCRC32C bool
10151033
disableAutoChecksum bool
10161034
objectAttrs *ObjectAttrs
1017-
fullObjectChecksum func() uint32
1035+
fullObjectChecksum func() *uint32
10181036
}
10191037

10201038
func (w *gRPCWriter) newGRPCOneshotBidiWriteBufferSender() *gRPCOneshotBidiWriteBufferSender {
@@ -1031,8 +1049,9 @@ func (w *gRPCWriter) newGRPCOneshotBidiWriteBufferSender() *gRPCOneshotBidiWrite
10311049
sendCRC32C: w.sendCRC32C,
10321050
disableAutoChecksum: w.disableAutoChecksum,
10331051
objectAttrs: w.attrs,
1034-
fullObjectChecksum: func() uint32 {
1035-
return w.fullObjectChecksum
1052+
fullObjectChecksum: func() *uint32 {
1053+
checksum := w.fullObjectChecksum
1054+
return &checksum
10361055
},
10371056
}
10381057
}
@@ -1151,7 +1170,7 @@ type gRPCResumableBidiWriteBufferSender struct {
11511170
sendCRC32C bool
11521171
disableAutoChecksum bool
11531172
objectAttrs *ObjectAttrs
1154-
fullObjectChecksum func() uint32
1173+
fullObjectChecksum func() *uint32
11551174

11561175
streamErr error
11571176
}
@@ -1168,8 +1187,9 @@ func (w *gRPCWriter) newGRPCResumableBidiWriteBufferSender() *gRPCResumableBidiW
11681187
sendCRC32C: w.sendCRC32C,
11691188
disableAutoChecksum: w.disableAutoChecksum,
11701189
objectAttrs: w.attrs,
1171-
fullObjectChecksum: func() uint32 {
1172-
return w.fullObjectChecksum
1190+
fullObjectChecksum: func() *uint32 {
1191+
checksum := w.fullObjectChecksum
1192+
return &checksum
11731193
},
11741194
}
11751195
}
@@ -1299,7 +1319,7 @@ type gRPCAppendBidiWriteBufferSender struct {
12991319
sendCRC32C bool
13001320
disableAutoChecksum bool
13011321
objectAttrs *ObjectAttrs
1302-
fullObjectChecksum func() uint32
1322+
fullObjectChecksum func() *uint32
13031323

13041324
takeoverWriter bool
13051325

@@ -1323,8 +1343,12 @@ func (w *gRPCWriter) newGRPCAppendableObjectBufferSender() *gRPCAppendBidiWriteB
13231343
sendCRC32C: w.sendCRC32C,
13241344
disableAutoChecksum: w.disableAutoChecksum,
13251345
objectAttrs: w.attrs,
1326-
fullObjectChecksum: func() uint32 {
1327-
return w.fullObjectChecksum
1346+
fullObjectChecksum: func() *uint32 {
1347+
if !w.sendAppendFinalCRC32C {
1348+
return nil
1349+
}
1350+
checksum := w.appendFinalCRC32C
1351+
return &checksum
13281352
},
13291353
}
13301354
}
@@ -1439,8 +1463,12 @@ func (w *gRPCWriter) newGRPCAppendTakeoverWriteBufferSender() *gRPCAppendTakeove
14391463
sendCRC32C: w.sendCRC32C,
14401464
disableAutoChecksum: w.disableAutoChecksum,
14411465
objectAttrs: w.attrs,
1442-
fullObjectChecksum: func() uint32 {
1443-
return w.fullObjectChecksum
1466+
fullObjectChecksum: func() *uint32 {
1467+
if !w.sendAppendFinalCRC32C {
1468+
return nil
1469+
}
1470+
checksum := w.appendFinalCRC32C
1471+
return &checksum
14441472
},
14451473
},
14461474
takeoverReported: false,
@@ -1582,7 +1610,7 @@ func (s *gRPCAppendBidiWriteBufferSender) send(stream storagepb.Storage_BidiWrit
15821610
fullObjectChecksum: s.fullObjectChecksum,
15831611
disableAutoChecksum: s.disableAutoChecksum,
15841612
finishWrite: finalizeObject,
1585-
takeoverWriter: s.takeoverWriter,
1613+
append: true,
15861614
})
15871615
req := bidiWriteObjectRequest(r, bufChecksum, objectChecksums)
15881616
if sendFirstMessage {

‎storage/grpc_writer_test.go‎

Lines changed: 33 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -30,12 +30,12 @@ import (
3030
func TestGetObjectChecksums(t *testing.T) {
3131
tests := []struct {
3232
name string
33-
fullObjectChecksum func() uint32
33+
fullObjectChecksum func() *uint32
3434
finishWrite bool
3535
sendCRC32C bool
36-
takeoverWriter bool
3736
disableAutoChecksum bool
3837
attrs *ObjectAttrs
38+
append bool
3939
want *storagepb.ObjectChecksums
4040
}{
4141
{
@@ -93,7 +93,7 @@ func TestGetObjectChecksums(t *testing.T) {
9393
},
9494
{
9595
name: "CRC32C enabled, no user-provided checksum",
96-
fullObjectChecksum: func() uint32 { return 456 },
96+
fullObjectChecksum: func() *uint32 { return proto.Uint32(456) },
9797
finishWrite: true,
9898
sendCRC32C: false,
9999
disableAutoChecksum: false,
@@ -102,12 +102,36 @@ func TestGetObjectChecksums(t *testing.T) {
102102
Crc32C: proto.Uint32(456),
103103
},
104104
},
105-
// TODO(b/461982277): remove this testcase once checksums for takeover writer is implemented
106105
{
107-
name: "takeover writer should return nil",
108-
finishWrite: true,
109-
takeoverWriter: true,
110-
want: nil,
106+
name: "CRC32C enabled, but callback returns nil (missing initial checksum)",
107+
fullObjectChecksum: func() *uint32 { return nil },
108+
finishWrite: true,
109+
sendCRC32C: false,
110+
disableAutoChecksum: false,
111+
attrs: &ObjectAttrs{},
112+
want: nil,
113+
},
114+
{
115+
name: "Append operation without final user-provided CRC32C (callback returns nil)",
116+
fullObjectChecksum: func() *uint32 { return nil },
117+
finishWrite: true,
118+
append: true,
119+
sendCRC32C: false,
120+
disableAutoChecksum: false,
121+
attrs: &ObjectAttrs{},
122+
want: nil,
123+
},
124+
{
125+
name: "Append operation with final CRC32C and initial CRC32C",
126+
fullObjectChecksum: func() *uint32 { return proto.Uint32(123) },
127+
finishWrite: true,
128+
append: true,
129+
sendCRC32C: false,
130+
disableAutoChecksum: false,
131+
attrs: &ObjectAttrs{CRC32C: 456},
132+
want: &storagepb.ObjectChecksums{
133+
Crc32C: proto.Uint32(123),
134+
},
111135
},
112136
}
113137

@@ -119,7 +143,7 @@ func TestGetObjectChecksums(t *testing.T) {
119143
objectAttrs: tt.attrs,
120144
fullObjectChecksum: tt.fullObjectChecksum,
121145
finishWrite: tt.finishWrite,
122-
takeoverWriter: tt.takeoverWriter,
146+
append: tt.append,
123147
})
124148
if !proto.Equal(got, tt.want) {
125149
t.Errorf("getObjectChecksums() = %v, want %v", got, tt.want)

‎storage/http_client.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1120,6 +1120,9 @@ func (hiw *httpInternalWriter) Flush() (int64, error) {
11201120
return 0, errors.New("Writer.Flush is only supported for gRPC-based clients")
11211121
}
11221122

1123+
// Not supported on HTTP Client as this is for setting CRC on appendable objects.
1124+
func (hiw *httpInternalWriter) setAppendFinalCRC32C(sendAppendFinalCRC32C bool, c uint32) {}
1125+
11231126
func (c *httpStorageClient) OpenWriter(params *openWriterParams, opts ...storageOption) (internalWriter, error) {
11241127
if params.append {
11251128
return nil, errors.New("storage: append not supported on HTTP Client; use gRPC")

0 commit comments

Comments
 (0)