[dbnode] Add IndexEntryChecksum field to entries in index file

docs/m3db/architecture/storage.md

@@ -19,24 +19,24 @@ A fileset has the following files:
 * **Checkpoint file:** Stores a digest of the digests file and written at the succesful completion of a fileset volume being persisted, allows for quickly checking if a volume was completed.
 
 ```
-                                                     ┌─────────────────────┐
-┌─────────────────────┐  ┌─────────────────────┐     │     Index File      │
-│      Info File      │  │   Summaries File    │     │   (sorted by ID)    │
-├─────────────────────┤  │   (sorted by ID)    │     ├─────────────────────┤
-│- Block Start        │  ├─────────────────────┤  ┌─>│- Idx                │
-│- Block Size         │  │- Idx                │  │  │- ID                 │
-│- Entries (Num)      │  │- ID                 │  │  │- Size               │
-│- Major Version      │  │- Index Entry Offset ├──┘  │- Checksum           │
-│- Summaries (Num)    │  └─────────────────────┘     │- Data Entry Offset  ├──┐
-│- BloomFilter (K/M)  │                              │- Encoded Tags       |  |
-│- Snapshot Time      │                              └─────────────────────┘  │
-│- Type (Flush/Snap)  |                                                       |
-│- Snapshot ID        |                                                       │
-│- Volume Index       |                                                       │
-│- Minor Version      |                                                       │
-└─────────────────────┘                                                       │
-                                                                              │
-                         ┌─────────────────────┐  ┌───────────────────────────┘
+                                                     ┌───────────────────────┐
+┌─────────────────────┐  ┌─────────────────────┐     │     Index File        │
+│      Info File      │  │   Summaries File    │     │   (sorted by ID)      │
+├─────────────────────┤  │   (sorted by ID)    │     ├───────────────────────┤
+│- Block Start        │  ├─────────────────────┤  ┌─>│- Idx                  │
+│- Block Size         │  │- Idx                │  │  │- ID                   │
+│- Entries (Num)      │  │- ID                 │  │  │- Size                 │
+│- Major Version      │  │- Index Entry Offset ├──┘  │- Checksum             │
+│- Summaries (Num)    │  └─────────────────────┘     │- Data Entry Offset    ├──┐
+│- BloomFilter (K/M)  │                              │- Encoded Tags         │  │
+│- Snapshot Time      │                              │- Index Entry Checksum │  │
+│- Type (Flush/Snap)  │                              └───────────────────────┘  │
+│- Snapshot ID        │                                                         │
+│- Volume Index       │                                                         │
+│- Minor Version      │                                                         │
+└─────────────────────┘                                                         │
+                                                                                │
+                         ┌─────────────────────┐  ┌─────────────────────────────┘
 ┌─────────────────────┐  │  Bloom Filter File  │  │
 │    Digests File     │  ├─────────────────────┤  │  ┌─────────────────────┐
 ├─────────────────────┤  │- Bitset             │  │  │      Data File      │

src/dbnode/persist/fs/msgpack/decoder.go

@@ -350,12 +350,25 @@ func (dec *Decoder) decodeIndexBloomFilterInfo() schema.IndexBloomFilterInfo {
 
 func (dec *Decoder) decodeIndexEntry(bytesPool pool.BytesPool) schema.IndexEntry {
 	var opts checkNumFieldsOptions
-	if dec.legacy.decodeLegacyV1IndexEntry {
+	switch dec.legacy.decodeLegacyIndexEntryVersion {
+	case legacyEncodingIndexEntryVersionV1:
 		// V1 had 5 fields.
 		opts.override = true
 		opts.numExpectedMinFields = 5
 		opts.numExpectedCurrFields = 5
+	case legacyEncodingIndexEntryVersionV2:
+		// V2 had 6 fields.
+		opts.override = true
+		opts.numExpectedMinFields = 5
+		opts.numExpectedCurrFields = 6
+	case legacyEncodingIndexEntryVersionCurrent:
+		// V3 is current version, no overrides needed
+		break
+	default:
+		dec.err = fmt.Errorf("invalid legacyEncodingIndexEntryVersion provided: %v", dec.legacy.decodeLegacyIndexEntryVersion)
+		return emptyIndexEntry
 	}
+
 	numFieldsToSkip, actual, ok := dec.checkNumFieldsFor(indexEntryType, opts)
 	if !ok {
 		return emptyIndexEntry
@@ -372,20 +385,35 @@ func (dec *Decoder) decodeIndexEntry(bytesPool pool.BytesPool) schema.IndexEntry
 
 	indexEntry.Size = dec.decodeVarint()
 	indexEntry.Offset = dec.decodeVarint()
-	indexEntry.Checksum = dec.decodeVarint()
+	indexEntry.DataChecksum = dec.decodeVarint()
 
-	if dec.legacy.decodeLegacyV1IndexEntry || actual < 6 {
+	// At this point, if its a V1 file, we've decoded all the available fields.
+	if dec.legacy.decodeLegacyIndexEntryVersion == legacyEncodingIndexEntryVersionV1 || actual < 6 {
 		dec.skip(numFieldsToSkip)
 		return indexEntry
 	}
 
+	// Decode fields added in V2
 	if bytesPool == nil {
 		indexEntry.EncodedTags, _, _ = dec.decodeBytes()
 	} else {
 		indexEntry.EncodedTags = dec.decodeBytesWithPool(bytesPool)
 	}
 
+	// At this point, if its a V2 file, we've decoded all the available fields.
+	if dec.legacy.decodeLegacyIndexEntryVersion == legacyEncodingIndexEntryVersionV2 || actual < 7 {
+		dec.skip(numFieldsToSkip)
+		return indexEntry
+	}
+
+	// Intentionally skip any extra fields here as we've stipulated that from V3 onward, IndexEntryChecksum will be the
+	// final field on index entries
 	dec.skip(numFieldsToSkip)
+
+	// Decode checksum field originally added in V3
+	// TODO(nate): actually use the checksum value for index entry validation - #2629
+	_ = dec.decodeVarint()
+
 	return indexEntry
 }
 

src/dbnode/persist/fs/msgpack/decoder_test.go

@@ -23,6 +23,7 @@ package msgpack
 import (
 	"testing"
 
+	"github.com/m3db/m3/src/dbnode/digest"
 	"github.com/stretchr/testify/require"
 )
 
@@ -128,12 +129,27 @@ func TestDecodeIndexEntryMoreFieldsThanExpected(t *testing.T) {
 	// Intentionally bump number of fields for the index entry object
 	enc.encodeNumObjectFieldsForFn = testGenEncodeNumObjectFieldsForFn(enc, indexEntryType, 1)
 	require.NoError(t, enc.EncodeIndexEntry(testIndexEntry))
+
+	// This hokey bit of logic is done so we can add extra fields in the correct location (since new IndexEntry fields
+	// will be added *before* the checksum). Confirm current checksum is correct, strip it, add unexpected field,
+	// and re-add updated checksum value
+
+	// Validate existing checksum
+	checksumPos := len(enc.Bytes()) - 5 // 5 bytes = 1 byte for integer code + 4 bytes for checksum
+	dec.Reset(NewByteDecoderStream(enc.Bytes()[checksumPos:]))
+	currChecksum := dec.decodeVarint()
+	require.Equal(t, currChecksum, int64(digest.Checksum(enc.Bytes()[:checksumPos])))
+
+	// Strip checksum, add new field, add updated checksum
+	enc.buf.Truncate(len(enc.Bytes()) - 5)
 	require.NoError(t, enc.enc.EncodeInt64(1234))
+	require.NoError(t, enc.enc.EncodeInt64(int64(digest.Checksum(enc.Bytes()))))
 
 	// Verify we can successfully skip unnecessary fields
 	dec.Reset(NewByteDecoderStream(enc.Bytes()))
 	res, err := dec.DecodeIndexEntry(nil)
 	require.NoError(t, err)
+
 	require.Equal(t, testIndexEntry, res)
 }
 

src/dbnode/persist/fs/msgpack/encoder.go

@@ -23,6 +23,7 @@ package msgpack
 import (
 	"bytes"
 
+	"github.com/m3db/m3/src/dbnode/digest"
 	"github.com/m3db/m3/src/dbnode/persist/schema"
 
 	"gopkg.in/vmihailenco/msgpack.v2"
@@ -64,20 +65,29 @@ const (
 	legacyEncodingIndexVersionV5
 )
 
+type legacyEncodingIndexEntryVersion int
+
+const (
+	legacyEncodingIndexEntryVersionCurrent                                 = legacyEncodingIndexEntryVersionV3
+	legacyEncodingIndexEntryVersionV1      legacyEncodingIndexEntryVersion = iota
+	legacyEncodingIndexEntryVersionV2
+	legacyEncodingIndexEntryVersionV3
+)
+
 type legacyEncodingOptions struct {
 	encodeLegacyIndexInfoVersion legacyEncodingIndexInfoVersion
 	decodeLegacyIndexInfoVersion legacyEncodingIndexInfoVersion
 
-	encodeLegacyV1IndexEntry bool
-	decodeLegacyV1IndexEntry bool
+	encodeLegacyIndexEntryVersion legacyEncodingIndexEntryVersion
+	decodeLegacyIndexEntryVersion legacyEncodingIndexEntryVersion
 }
 
 var defaultlegacyEncodingOptions = legacyEncodingOptions{
 	encodeLegacyIndexInfoVersion: legacyEncodingIndexVersionCurrent,
 	decodeLegacyIndexInfoVersion: legacyEncodingIndexVersionCurrent,
 
-	encodeLegacyV1IndexEntry: false,
-	decodeLegacyV1IndexEntry: false,
+	encodeLegacyIndexEntryVersion: legacyEncodingIndexEntryVersionCurrent,
+	decodeLegacyIndexEntryVersion: legacyEncodingIndexEntryVersionCurrent,
 }
 
 // NewEncoder creates a new encoder.
@@ -100,7 +110,8 @@ func newEncoder(legacy legacyEncodingOptions) *Encoder {
 	enc.encodeBytesFn = enc.encodeBytes
 	enc.encodeArrayLenFn = enc.encodeArrayLen
 
-	// Used primarily for testing.
+	// Used primarily for testing however legitimate production uses exist (e.g. addition of IndexEntryChecksum in
+	// IndexEntryV3)
 	enc.legacy = legacy
 
 	return enc
@@ -141,11 +152,19 @@ func (enc *Encoder) EncodeIndexEntry(entry schema.IndexEntry) error {
 	if enc.err != nil {
 		return enc.err
 	}
+
+	// There's no guarantee EncodeIndexEntry is called with an empty buffer so ensure
+	// only checksumming the bits we care about.
+	checksumStart := enc.buf.Len()
+
 	enc.encodeRootObject(indexEntryVersion, indexEntryType)
-	if enc.legacy.encodeLegacyV1IndexEntry {
+	switch enc.legacy.encodeLegacyIndexEntryVersion {
+	case legacyEncodingIndexEntryVersionV1:
 		enc.encodeIndexEntryV1(entry)
-	} else {
+	case legacyEncodingIndexEntryVersionV2:
 		enc.encodeIndexEntryV2(entry)
+	default:
+		enc.encodeIndexEntryV3(entry, checksumStart)
 	}
 	return enc.err
 }
@@ -283,17 +302,30 @@ func (enc *Encoder) encodeIndexEntryV1(entry schema.IndexEntry) {
 	enc.encodeBytesFn(entry.ID)
 	enc.encodeVarintFn(entry.Size)
 	enc.encodeVarintFn(entry.Offset)
-	enc.encodeVarintFn(entry.Checksum)
+	enc.encodeVarintFn(entry.DataChecksum)
 }
 
 func (enc *Encoder) encodeIndexEntryV2(entry schema.IndexEntry) {
+	enc.encodeArrayLenFn(6) // V2 had 6 fields.
+	enc.encodeVarintFn(entry.Index)
+	enc.encodeBytesFn(entry.ID)
+	enc.encodeVarintFn(entry.Size)
+	enc.encodeVarintFn(entry.Offset)
+	enc.encodeVarintFn(entry.DataChecksum)
+	enc.encodeBytesFn(entry.EncodedTags)
+}
+
+func (enc *Encoder) encodeIndexEntryV3(entry schema.IndexEntry, checksumStart int) {
 	enc.encodeNumObjectFieldsForFn(indexEntryType)
 	enc.encodeVarintFn(entry.Index)
 	enc.encodeBytesFn(entry.ID)
 	enc.encodeVarintFn(entry.Size)
 	enc.encodeVarintFn(entry.Offset)
-	enc.encodeVarintFn(entry.Checksum)
+	enc.encodeVarintFn(entry.DataChecksum)
 	enc.encodeBytesFn(entry.EncodedTags)
+
+	checksum := digest.Checksum(enc.Bytes()[checksumStart:])
+	enc.encodeVarintFn(int64(checksum))
 }
 
 func (enc *Encoder) encodeIndexSummary(summary schema.IndexSummary) {

src/dbnode/persist/fs/msgpack/encoder_test.go

@@ -40,19 +40,33 @@ func testCapturingEncoder(t *testing.T) (*Encoder, *[]interface{}) {
 	encoder := NewEncoder()
 
 	var result []interface{}
+	actualEncodeVarintFn := encoder.encodeVarintFn
 	encoder.encodeVarintFn = func(value int64) {
+		actualEncodeVarintFn(value)
 		result = append(result, value)
 	}
+
+	actualEncodeVarUintFn := encoder.encodeVarUintFn
 	encoder.encodeVarUintFn = func(value uint64) {
+		actualEncodeVarUintFn(value)
 		result = append(result, value)
 	}
+
+	actualEncodeFloat64Fn := encoder.encodeFloat64Fn
 	encoder.encodeFloat64Fn = func(value float64) {
+		actualEncodeFloat64Fn(value)
 		result = append(result, value)
 	}
+
+	actualEncodeBytesFn := encoder.encodeBytesFn
 	encoder.encodeBytesFn = func(value []byte) {
+		actualEncodeBytesFn(value)
 		result = append(result, value)
 	}
+
+	actualEncodeArrayLenFn := encoder.encodeArrayLenFn
 	encoder.encodeArrayLenFn = func(value int) {
+		actualEncodeArrayLenFn(value)
 		result = append(result, value)
 	}
 
@@ -98,8 +112,9 @@ func testExpectedResultForIndexEntry(t *testing.T, indexEntry schema.IndexEntry)
 		indexEntry.ID,
 		indexEntry.Size,
 		indexEntry.Offset,
-		indexEntry.Checksum,
+		indexEntry.DataChecksum,
 		indexEntry.EncodedTags,
+		testIndexEntryChecksum, // Checksum auto-added to the end of the index entry
 	}
 }