Skip to content

Commit a04d980

Browse files
krishnamd-jkpkrishnamd-jkp
authored
feat(go): Add full object checksum for negative offsets > size (#20026)
1 parent a25e93d commit a04d980

4 files changed

Lines changed: 113 additions & 80 deletions

File tree

‎storage/client_test.go‎

Lines changed: 79 additions & 68 deletions
Original file line numberDiff line numberDiff line change
@@ -2072,81 +2072,91 @@ func TestReadObjectWrongChecksumWholeObjectSizeEmulated(t *testing.T) {
20722072

20732073
for _, bidiReads := range []bool{false, true} {
20742074
for _, disableChecksum := range []bool{false, true} {
2075-
t.Run(fmt.Sprintf("bidiReads=%v/disableChecksum=%v", bidiReads, disableChecksum), func(t *testing.T) {
2076-
ctx := context.Background()
2077-
2078-
streamInterceptor := grpc.WithStreamInterceptor(
2079-
func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
2080-
clientStream, err := streamer(ctx, desc, cc, method, opts...)
2081-
2082-
switch method {
2083-
case "/google.storage.v2.Storage/ReadObject":
2084-
clientStream = &customObjectCRCReadStream{ClientStream: clientStream, isBidi: false}
2085-
case "/google.storage.v2.Storage/BidiReadObject":
2086-
clientStream = &customObjectCRCReadStream{ClientStream: clientStream, isBidi: true}
2087-
}
2088-
return clientStream, err
2089-
})
2090-
2091-
var clientOpts []option.ClientOption
2092-
clientOpts = append(clientOpts, option.WithGRPCDialOption(streamInterceptor))
2093-
if bidiReads {
2094-
clientOpts = append(clientOpts, experimental.WithGRPCBidiReads())
2095-
}
2096-
2097-
client, err := NewGRPCClient(ctx, clientOpts...)
2098-
if err != nil {
2099-
t.Fatalf("NewGRPCClient: %v", err)
2100-
}
2101-
2102-
var (
2103-
contents = randomBytes9MiB
2104-
prefix = time.Now().Nanosecond()
2105-
bucket = fmt.Sprintf("bucket-%d", prefix)
2106-
objName = fmt.Sprintf("%d-object", prefix)
2107-
o = client.Bucket(bucket).Object(objName)
2108-
)
2109-
2110-
if err := client.Bucket(bucket).Create(ctx, "project", nil); err != nil {
2111-
t.Fatalf("creating test bucket: %v", err)
2112-
}
2113-
w := o.NewWriter(ctx)
2114-
if _, err = w.Write(contents); err != nil {
2115-
t.Fatalf("writing test data: got %v; want ok", err)
2116-
}
2117-
if err := w.Close(); err != nil {
2118-
t.Fatalf("closing test data writer: got %v; want ok", err)
2119-
}
2120-
2121-
var readerOpts []ReaderOption
2122-
if disableChecksum {
2123-
readerOpts = append(readerOpts, WithDisableReaderChecksum())
2124-
}
2075+
for _, negativeOffset := range []bool{false, true} {
2076+
t.Run(fmt.Sprintf("bidiReads=%v/disableChecksum=%v/negativeOffset=%v", bidiReads, disableChecksum, negativeOffset), func(t *testing.T) {
2077+
ctx := context.Background()
2078+
2079+
streamInterceptor := grpc.WithStreamInterceptor(
2080+
func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
2081+
clientStream, err := streamer(ctx, desc, cc, method, opts...)
2082+
2083+
switch method {
2084+
case "/google.storage.v2.Storage/ReadObject":
2085+
clientStream = &customObjectCRCReadStream{ClientStream: clientStream, isBidi: false}
2086+
case "/google.storage.v2.Storage/BidiReadObject":
2087+
clientStream = &customObjectCRCReadStream{ClientStream: clientStream, isBidi: true}
2088+
}
2089+
return clientStream, err
2090+
})
21252091

2126-
r, err := o.NewRangeReader(ctx, 0, int64(len(contents)), readerOpts...)
2127-
if err != nil {
2128-
t.Fatalf("NewRangeReader: %v", err)
2129-
}
2092+
var clientOpts []option.ClientOption
2093+
clientOpts = append(clientOpts, option.WithGRPCDialOption(streamInterceptor))
2094+
if bidiReads {
2095+
clientOpts = append(clientOpts, experimental.WithGRPCBidiReads())
2096+
}
21302097

2131-
if disableChecksum {
2132-
buf := new(bytes.Buffer)
2133-
_, err = io.Copy(buf, r)
2098+
client, err := NewGRPCClient(ctx, clientOpts...)
21342099
if err != nil {
2135-
t.Fatalf("expected nil error with checksum disabled, got %v", err)
2100+
t.Fatalf("NewGRPCClient: %v", err)
21362101
}
2137-
if got, want := buf.Bytes(), contents; !bytes.Equal(got, want) {
2138-
t.Errorf("content mismatch: got %v bytes, want %v bytes", len(got), len(want))
2102+
defer client.Close()
2103+
var (
2104+
contents = randomBytes9MiB
2105+
prefix = time.Now().Nanosecond()
2106+
bucket = fmt.Sprintf("bucket-%d", prefix)
2107+
objName = fmt.Sprintf("%d-object", prefix)
2108+
o = client.Bucket(bucket).Object(objName)
2109+
)
2110+
2111+
if err := client.Bucket(bucket).Create(ctx, "project", nil); err != nil {
2112+
t.Fatalf("creating test bucket: %v", err)
21392113
}
2140-
} else {
2141-
_, err = io.Copy(io.Discard, r)
2142-
if err == nil {
2143-
t.Fatalf("expected error due to bad object CRC, got nil")
2114+
w := o.NewWriter(ctx)
2115+
if _, err = w.Write(contents); err != nil {
2116+
t.Fatalf("writing test data: got %v; want ok", err)
21442117
}
2145-
if got, want := err.Error(), "bad CRC on read"; !strings.Contains(got, want) {
2146-
t.Errorf("error mismatch: got %q, want to contain %q", got, want)
2118+
if err := w.Close(); err != nil {
2119+
t.Fatalf("closing test data writer: got %v; want ok", err)
21472120
}
2148-
}
2149-
})
2121+
2122+
var readerOpts []ReaderOption
2123+
if disableChecksum {
2124+
readerOpts = append(readerOpts, WithDisableReaderChecksum())
2125+
}
2126+
2127+
var offset, length int64
2128+
if negativeOffset {
2129+
offset = -int64(len(contents) + 1000)
2130+
length = -1
2131+
} else {
2132+
offset = 0
2133+
length = int64(len(contents))
2134+
}
2135+
r, err := o.NewRangeReader(ctx, offset, length, readerOpts...)
2136+
if err != nil {
2137+
t.Fatalf("NewRangeReader: %v", err)
2138+
}
2139+
defer r.Close()
2140+
if disableChecksum {
2141+
buf := new(bytes.Buffer)
2142+
_, err = io.Copy(buf, r)
2143+
if err != nil {
2144+
t.Fatalf("expected nil error with checksum disabled, got %v", err)
2145+
}
2146+
if got, want := buf.Bytes(), contents; !bytes.Equal(got, want) {
2147+
t.Errorf("content mismatch: got %v bytes, want %v bytes", len(got), len(want))
2148+
}
2149+
} else {
2150+
_, err = io.Copy(io.Discard, r)
2151+
if err == nil {
2152+
t.Fatalf("expected error due to bad object CRC, got nil")
2153+
}
2154+
if got, want := err.Error(), "bad CRC on read"; !strings.Contains(got, want) {
2155+
t.Errorf("error mismatch: got %q, want to contain %q", got, want)
2156+
}
2157+
}
2158+
})
2159+
}
21502160
}
21512161
}
21522162
}
@@ -2170,6 +2180,7 @@ func TestReadObjectWrongChecksumUnfinalizedWholeObjectSizeEmulated(t *testing.T)
21702180
if err != nil {
21712181
t.Fatalf("NewGRPCClient: %v", err)
21722182
}
2183+
defer client.Close()
21732184

21742185
var (
21752186
contents = randomBytes9MiB

‎storage/doc.go‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -419,6 +419,19 @@ apply to single-shot uploads when user-provided checksum is provided.
419419
420420
Automatic checksumming can be disabled using [Writer.DisableAutoChecksum].
421421
422+
# Read checksumming
423+
424+
By default, the client automatically computes and validates CRC32C checksums for reads
425+
when downloading an entire object, providing an additional layer of data integrity
426+
validation with a slight CPU overhead.
427+
428+
For gRPC clients, read checksumming is also performed for partial reads (range requests)
429+
by validating the checksum of each individual data chunk returned by the server.
430+
431+
Automatic read checksumming can be disabled using the [WithDisableReaderChecksum] option
432+
for a normal range reader, or the [WithDisableMRDReadChecksum] option for the multi-range
433+
downloader.
434+
422435
# Parallel Uploads
423436
424437
The parallel upload feature splits a large object into multiple parts and uploads them

‎storage/grpc_client.go‎

Lines changed: 10 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1372,31 +1372,30 @@ func (c *grpcStorageClient) NewRangeReader(ctx context.Context, params *newRange
13721372
params.length = obj.Size - params.offset
13731373
}
13741374

1375+
startOffset := params.offset
1376+
if params.offset < 0 {
1377+
startOffset = size + params.offset
1378+
}
1379+
// If caller has specified a negative start offset that's larger than the
1380+
// reported size, start at the beginning of the object.
1381+
if startOffset < 0 {
1382+
startOffset = 0
1383+
}
13751384
// Only support checksums when reading an entire object, not a range.
13761385
var (
13771386
wantCRC uint32
13781387
checkCRC bool
13791388
)
13801389
if checksums := obj.GetChecksums(); checksums != nil && checksums.Crc32C != nil {
13811390
if !params.disableCRCCheck &&
1382-
params.offset == 0 &&
1391+
startOffset == 0 &&
13831392
(params.length < 0 ||
13841393
finalized && params.length >= size) {
13851394
checkCRC = true
13861395
}
13871396
wantCRC = checksums.GetCrc32C()
13881397
}
13891398

1390-
startOffset := params.offset
1391-
if params.offset < 0 {
1392-
startOffset = size + params.offset
1393-
}
1394-
// If caller has specified a negative start offset that's larger than the
1395-
// reported size, start at the beginning of the object.
1396-
if startOffset < 0 {
1397-
startOffset = 0
1398-
}
1399-
14001399
// The remaining bytes are the lesser of the requested range and all bytes
14011400
// after params.offset.
14021401
length := params.length

‎storage/grpc_reader.go‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -191,13 +191,23 @@ func (c *grpcStorageClient) NewRangeReaderReadObject(ctx context.Context, params
191191
chunkCRCPresent = true
192192
chunkCRC = *msg.GetChecksummedData().Crc32C
193193
}
194+
startOffset := params.offset
195+
if params.offset < 0 {
196+
startOffset = size + params.offset
197+
}
198+
// If caller has specified a negative start offset that's larger than the
199+
// reported size, start at the beginning of the object.
200+
if startOffset < 0 {
201+
startOffset = 0
202+
}
203+
194204
var (
195205
wantCRC uint32
196206
checkCRC bool
197207
)
198208
if checksums := msg.GetObjectChecksums(); checksums != nil && checksums.Crc32C != nil {
199209
if !params.disableCRCCheck &&
200-
params.offset == 0 &&
210+
startOffset == 0 &&
201211
(params.length < 0 || (obj != nil && params.length >= size)) {
202212
checkCRC = true
203213
}

0 commit comments

Comments
 (0)