MB-25244: memdb: Check shard order in LoadFromDisk

While linking built segments in Assemble(), verify that the tail of each
segment is lesser than the head of the next segment being linked.

Change-Id: I6c43eb20fa93bbc55dd298bb59c8d0f4a700d630

secondary/memdb/memdb.go

@@ -377,6 +377,33 @@ func (m *MemDB) newStoreConfig() skiplist.Config {
 		slCfg.BarrierDestructor = m.newBSDestructor()
 
 	}
+
+	slCfg.VerifyOrder = func(curr, next *skiplist.Node) bool {
+		if curr == nil || next == nil {
+			return true
+		}
+
+		cItem := curr.Item()
+		nItem := next.Item()
+		if cItem == nil {
+			if nItem != nil {
+				// next cannot be nil if curr is nil
+				return false
+			}
+			return true
+		} else if nItem == nil {
+			return true
+		}
+
+		// curr can be == next, curr can be < next
+		// return false if curr > next
+		if bytes.Compare((*Item)(cItem).Bytes(), (*Item)(nItem).Bytes()) > 0 {
+			return false
+		}
+
+		return true
+	}
+
 	return slCfg
 }
 
@@ -1094,7 +1121,11 @@ func (m *MemDB) LoadFromDisk(dir string, concurr int, callb ItemCallback) (*Snap
 		}
 	}
 
-	m.store = b.Assemble(segments...)
+	assembledStore := b.Assemble(segments...)
+	if assembledStore == nil {
+		return nil, ErrCorruptSnapshot
+	}
+	m.store = assembledStore
 
 	// Delta processing
 	if m.useDeltaFiles {

secondary/memdb/skiplist/builder.go

@@ -67,6 +67,10 @@ func (b *Builder) Assemble(segments ...*Segment) *Skiplist {
 	for _, seg := range segments {
 		for l := 0; l <= MaxLevel; l++ {
 			if tail[l] != nil && seg.head[l] != nil {
+				if !b.store.VerifyOrder(tail[l], seg.head[l]) {
+					return nil
+				}
+
 				tail[l].setNext(l, seg.head[l], false)
 			} else if head[l] == nil && seg.head[l] != nil {
 				head[l] = seg.head[l]

secondary/memdb/skiplist/skiplist.go

@@ -24,6 +24,7 @@ type FreeFn func(unsafe.Pointer)
 
 type Config struct {
 	ItemSize ItemSizeFn
+	VerifyOrder func(curr, next *Node) bool
 
 	UseMemoryMgmt     bool
 	Malloc            MallocFn
@@ -39,6 +40,9 @@ func DefaultConfig() Config {
 	return Config{
 		ItemSize:      defaultItemSize,
 		UseMemoryMgmt: false,
+		VerifyOrder:   func(curr, next *Node) bool {
+			return true
+		},
 	}
 }