Skip to content

Commit 6c2f40f

Browse files
krishnamd-jkpkrishnamd-jkp
authored
chore(storage): Move checksumming to streams in MRD (#20084)
1 parent 39ab8d3 commit 6c2f40f

3 files changed

Lines changed: 359 additions & 35 deletions

File tree

‎storage/grpc_client.go‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1900,6 +1900,7 @@ type readResponseDecoder struct {
19001900
msg *storagepb.BidiReadObjectResponse // processed response message with all fields other than object data populated
19011901
dataOffsets map[int64]bufferSliceOffsets // Map ReadId to the offsets of the object data for that ID in the message.
19021902
done bool // true if the data has been completely read.
1903+
crcErrs map[int64]error // Map ReadId to the CRC validation error if it failed.
19031904
}
19041905

19051906
type bufferSliceOffsets struct {
@@ -2027,6 +2028,51 @@ func (d *readResponseDecoder) readAndUpdateCRC(p []byte, readID int64, updateCRC
20272028
return n, true
20282029
}
20292030

2031+
func (d *readResponseDecoder) verifyChecksums() {
2032+
if d.msg == nil {
2033+
return
2034+
}
2035+
for _, dataRange := range d.msg.GetObjectDataRanges() {
2036+
checksummedData := dataRange.GetChecksummedData()
2037+
if checksummedData == nil || checksummedData.Crc32C == nil {
2038+
continue
2039+
}
2040+
readID := dataRange.GetReadRange().GetReadId()
2041+
offsets, ok := d.dataOffsets[readID]
2042+
wantCRC := *checksummedData.Crc32C
2043+
var gotCRC uint32
2044+
2045+
if ok {
2046+
for i := offsets.startBuf; i <= offsets.endBuf; i++ {
2047+
if i < 0 || i >= len(d.databufs) {
2048+
continue
2049+
}
2050+
databuf := d.databufs[i]
2051+
var start uint64
2052+
if i == offsets.startBuf {
2053+
start = min(offsets.startOff, uint64(databuf.Len()))
2054+
}
2055+
end := uint64(databuf.Len())
2056+
if i == offsets.endBuf {
2057+
end = min(offsets.endOff, end)
2058+
}
2059+
if start >= end {
2060+
continue
2061+
}
2062+
dataSlice := databuf.ReadOnlyData()[start:end]
2063+
gotCRC = crc32.Update(gotCRC, crc32cTable, dataSlice)
2064+
}
2065+
}
2066+
2067+
if gotCRC != wantCRC {
2068+
if d.crcErrs == nil {
2069+
d.crcErrs = make(map[int64]error)
2070+
}
2071+
d.crcErrs[readID] = fmt.Errorf("storage: bad CRC on chunk read: got %d, want %d", gotCRC, wantCRC)
2072+
}
2073+
}
2074+
}
2075+
20302076
func (d *readResponseDecoder) writeToAndUpdateCRC(w io.Writer, readID int64, updateCRC func([]byte)) (totalWritten int64, found bool, err error) {
20312077
// For a completely empty message, just return 0
20322078
if len(d.databufs) == 0 {
@@ -2042,6 +2088,9 @@ func (d *readResponseDecoder) writeToAndUpdateCRC(w io.Writer, readID int64, upd
20422088

20432089
// Loop from the current buffer to the ending buffer for this specific data range.
20442090
for i := offsets.currBuf; i <= offsets.endBuf; i++ {
2091+
if i < 0 || i >= len(d.databufs) {
2092+
continue
2093+
}
20452094
databuf := d.databufs[i]
20462095

20472096
// Determine the start and end of the data slice for the current buffer.

‎storage/grpc_client_test.go‎

Lines changed: 297 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -647,3 +647,300 @@ func TestNewGRPCStorageClient_NoGlobalTimeout(t *testing.T) {
647647
verifyTimeoutIsZero("BidiWriteObject", opts.BidiWriteObject)
648648
verifyTimeoutIsZero("CancelResumableWrite", opts.CancelResumableWrite)
649649
}
650+
651+
func TestVerifyChecksums(t *testing.T) {
652+
helloWorldCRC := crc32.Checksum([]byte("hello world"), crc32cTable)
653+
654+
tests := []struct {
655+
desc string
656+
msg *storagepb.BidiReadObjectResponse
657+
databufs mem.BufferSlice
658+
dataOffsets map[int64]bufferSliceOffsets
659+
initialErrs map[int64]error
660+
wantErrs map[int64]string // readID -> substring of error message, or empty if no error expected
661+
}{
662+
{
663+
desc: "nil msg returns early and does not initialize crcErrs",
664+
msg: nil,
665+
databufs: mem.BufferSlice{
666+
mem.SliceBuffer([]byte("hello world")),
667+
},
668+
wantErrs: nil, // stays nil
669+
},
670+
{
671+
desc: "empty databufs with checksum return error",
672+
msg: &storagepb.BidiReadObjectResponse{
673+
ObjectDataRanges: []*storagepb.ObjectRangeData{
674+
{
675+
ChecksummedData: &storagepb.ChecksummedData{
676+
Crc32C: &helloWorldCRC,
677+
},
678+
ReadRange: &storagepb.ReadRange{ReadId: 1},
679+
},
680+
},
681+
},
682+
databufs: mem.BufferSlice{},
683+
wantErrs: map[int64]string{
684+
1: "storage: bad CRC on chunk read: got",
685+
},
686+
},
687+
{
688+
desc: "missing checksummed data is skipped",
689+
msg: &storagepb.BidiReadObjectResponse{
690+
ObjectDataRanges: []*storagepb.ObjectRangeData{
691+
{
692+
ReadRange: &storagepb.ReadRange{ReadId: 1},
693+
},
694+
},
695+
},
696+
databufs: mem.BufferSlice{
697+
mem.SliceBuffer([]byte("hello world")),
698+
},
699+
dataOffsets: map[int64]bufferSliceOffsets{
700+
1: {startBuf: 0, endBuf: 0, startOff: 0, endOff: 11},
701+
},
702+
wantErrs: map[int64]string{}, // initialized but empty
703+
},
704+
{
705+
desc: "missing Crc32C pointer is skipped",
706+
msg: &storagepb.BidiReadObjectResponse{
707+
ObjectDataRanges: []*storagepb.ObjectRangeData{
708+
{
709+
ChecksummedData: &storagepb.ChecksummedData{},
710+
ReadRange: &storagepb.ReadRange{ReadId: 1},
711+
},
712+
},
713+
},
714+
databufs: mem.BufferSlice{
715+
mem.SliceBuffer([]byte("hello world")),
716+
},
717+
dataOffsets: map[int64]bufferSliceOffsets{
718+
1: {startBuf: 0, endBuf: 0, startOff: 0, endOff: 11},
719+
},
720+
wantErrs: map[int64]string{},
721+
},
722+
{
723+
desc: "single buffer correct checksum",
724+
msg: &storagepb.BidiReadObjectResponse{
725+
ObjectDataRanges: []*storagepb.ObjectRangeData{
726+
{
727+
ChecksummedData: &storagepb.ChecksummedData{
728+
Crc32C: &helloWorldCRC,
729+
},
730+
ReadRange: &storagepb.ReadRange{ReadId: 1},
731+
},
732+
},
733+
},
734+
databufs: mem.BufferSlice{
735+
mem.SliceBuffer([]byte("hello world")),
736+
},
737+
dataOffsets: map[int64]bufferSliceOffsets{
738+
1: {startBuf: 0, endBuf: 0, startOff: 0, endOff: 11},
739+
},
740+
wantErrs: map[int64]string{},
741+
},
742+
{
743+
desc: "single buffer incorrect checksum",
744+
msg: &storagepb.BidiReadObjectResponse{
745+
ObjectDataRanges: []*storagepb.ObjectRangeData{
746+
{
747+
ChecksummedData: &storagepb.ChecksummedData{
748+
Crc32C: proto.Uint32(12345),
749+
},
750+
ReadRange: &storagepb.ReadRange{ReadId: 1},
751+
},
752+
},
753+
},
754+
databufs: mem.BufferSlice{
755+
mem.SliceBuffer([]byte("hello world")),
756+
},
757+
dataOffsets: map[int64]bufferSliceOffsets{
758+
1: {startBuf: 0, endBuf: 0, startOff: 0, endOff: 11},
759+
},
760+
wantErrs: map[int64]string{
761+
1: "storage: bad CRC on chunk read: got",
762+
},
763+
},
764+
{
765+
desc: "multiple buffers correct checksum",
766+
msg: &storagepb.BidiReadObjectResponse{
767+
ObjectDataRanges: []*storagepb.ObjectRangeData{
768+
{
769+
ChecksummedData: &storagepb.ChecksummedData{
770+
Crc32C: &helloWorldCRC,
771+
},
772+
ReadRange: &storagepb.ReadRange{ReadId: 1},
773+
},
774+
},
775+
},
776+
databufs: mem.BufferSlice{
777+
mem.SliceBuffer([]byte("prefix hello")),
778+
mem.SliceBuffer([]byte(" world diff message")),
779+
mem.SliceBuffer([]byte("suffix")),
780+
},
781+
dataOffsets: map[int64]bufferSliceOffsets{
782+
1: {startBuf: 0, endBuf: 1, startOff: 7, endOff: 6},
783+
},
784+
wantErrs: map[int64]string{},
785+
},
786+
{
787+
desc: "multiple buffers incorrect checksum",
788+
msg: &storagepb.BidiReadObjectResponse{
789+
ObjectDataRanges: []*storagepb.ObjectRangeData{
790+
{
791+
ChecksummedData: &storagepb.ChecksummedData{
792+
Crc32C: proto.Uint32(12345),
793+
},
794+
ReadRange: &storagepb.ReadRange{ReadId: 1},
795+
},
796+
},
797+
},
798+
databufs: mem.BufferSlice{
799+
mem.SliceBuffer([]byte("prefix_hello")),
800+
mem.SliceBuffer([]byte("_world_")),
801+
mem.SliceBuffer([]byte("suffix")),
802+
},
803+
dataOffsets: map[int64]bufferSliceOffsets{
804+
1: {startBuf: 0, endBuf: 1, startOff: 7, endOff: 6},
805+
},
806+
wantErrs: map[int64]string{
807+
1: "storage: bad CRC on chunk read: got",
808+
},
809+
},
810+
{
811+
desc: "startOff >= endOff empty segment",
812+
msg: &storagepb.BidiReadObjectResponse{
813+
ObjectDataRanges: []*storagepb.ObjectRangeData{
814+
{
815+
ChecksummedData: &storagepb.ChecksummedData{
816+
Crc32C: proto.Uint32(0),
817+
},
818+
ReadRange: &storagepb.ReadRange{ReadId: 1},
819+
},
820+
},
821+
},
822+
databufs: mem.BufferSlice{
823+
mem.SliceBuffer([]byte("hello world")),
824+
},
825+
dataOffsets: map[int64]bufferSliceOffsets{
826+
1: {startBuf: 0, endBuf: 0, startOff: 5, endOff: 5},
827+
},
828+
wantErrs: map[int64]string{},
829+
},
830+
{
831+
desc: "multiple ranges mix of results",
832+
msg: &storagepb.BidiReadObjectResponse{
833+
ObjectDataRanges: []*storagepb.ObjectRangeData{
834+
{
835+
ChecksummedData: &storagepb.ChecksummedData{
836+
Crc32C: &helloWorldCRC,
837+
},
838+
ReadRange: &storagepb.ReadRange{ReadId: 10},
839+
},
840+
{
841+
ChecksummedData: &storagepb.ChecksummedData{
842+
Crc32C: proto.Uint32(9999),
843+
},
844+
ReadRange: &storagepb.ReadRange{ReadId: 20},
845+
},
846+
{
847+
ReadRange: &storagepb.ReadRange{ReadId: 30},
848+
},
849+
},
850+
},
851+
databufs: mem.BufferSlice{
852+
mem.SliceBuffer([]byte("hello world")),
853+
},
854+
dataOffsets: map[int64]bufferSliceOffsets{
855+
10: {startBuf: 0, endBuf: 0, startOff: 0, endOff: 11},
856+
20: {startBuf: 0, endBuf: 0, startOff: 0, endOff: 11},
857+
30: {startBuf: 0, endBuf: 0, startOff: 0, endOff: 11},
858+
},
859+
wantErrs: map[int64]string{
860+
20: "storage: bad CRC on chunk read: got",
861+
},
862+
},
863+
{
864+
desc: "pre-existing errors are preserved",
865+
msg: &storagepb.BidiReadObjectResponse{
866+
ObjectDataRanges: []*storagepb.ObjectRangeData{
867+
{
868+
ChecksummedData: &storagepb.ChecksummedData{
869+
Crc32C: &helloWorldCRC,
870+
},
871+
ReadRange: &storagepb.ReadRange{ReadId: 10},
872+
},
873+
},
874+
},
875+
databufs: mem.BufferSlice{
876+
mem.SliceBuffer([]byte("hello world")),
877+
},
878+
dataOffsets: map[int64]bufferSliceOffsets{
879+
10: {startBuf: 0, endBuf: 0, startOff: 0, endOff: 11},
880+
},
881+
initialErrs: map[int64]error{
882+
99: fmt.Errorf("some other error"),
883+
},
884+
wantErrs: map[int64]string{
885+
99: "some other error",
886+
10: "",
887+
},
888+
},
889+
}
890+
891+
for _, tc := range tests {
892+
t.Run(tc.desc, func(t *testing.T) {
893+
decoder := &readResponseDecoder{
894+
msg: tc.msg,
895+
databufs: tc.databufs,
896+
dataOffsets: tc.dataOffsets,
897+
crcErrs: tc.initialErrs,
898+
}
899+
900+
decoder.verifyChecksums()
901+
902+
if tc.wantErrs == nil {
903+
if decoder.crcErrs != nil {
904+
t.Errorf("expected crcErrs to be nil, got: %v", decoder.crcErrs)
905+
}
906+
return
907+
}
908+
909+
// Count expected errors (we don't expect errors for empty strings in wantErrs, i.e. 10 is correct)
910+
var expectedCount int
911+
for _, wantSubstr := range tc.wantErrs {
912+
if wantSubstr != "" {
913+
expectedCount++
914+
}
915+
}
916+
917+
var gotCount int
918+
for _, err := range decoder.crcErrs {
919+
if err != nil {
920+
gotCount++
921+
}
922+
}
923+
924+
if gotCount != expectedCount {
925+
t.Errorf("mismatched error count: got %v, want %v", decoder.crcErrs, tc.wantErrs)
926+
}
927+
928+
for readID, wantSubstr := range tc.wantErrs {
929+
err, ok := decoder.crcErrs[readID]
930+
if wantSubstr == "" {
931+
if ok && err != nil {
932+
t.Errorf("unexpected error for read ID %d: %v", readID, err)
933+
}
934+
continue
935+
}
936+
if !ok {
937+
t.Errorf("expected error for read ID %d, but none was found", readID)
938+
continue
939+
}
940+
if !strings.Contains(err.Error(), wantSubstr) {
941+
t.Errorf("read ID %d error message: got %q, want it to contain %q", readID, err.Error(), wantSubstr)
942+
}
943+
}
944+
})
945+
}
946+
}

0 commit comments

Comments
 (0)