Function VerifySnapshot

x/debug.go:48–103

VerifySnapshot iterates over all the keys in badger. For all data keys it checks if key is a split key and it verifies if all part are present in badger as well.

(pstore *badger.DB, readTs uint64)

// VerifySnapshot iterates over all the keys in badger. For all data keys it checks
// if key is a split key and it verifies if all part are present in badger as well.
func VerifySnapshot(pstore *badger.DB, readTs uint64) {
	stream := pstore.NewStreamAt(readTs)
	stream.KeyToList = func(key []byte, itr *badger.Iterator) (*bpb.KVList, error) {
		for ; itr.Valid(); itr.Next() {
			item := itr.Item()
			if item.IsDeletedOrExpired() {
				break
			}
			if !bytes.Equal(key, item.Key()) {
				// Break out on the first encounter with another key.
				break
			}

			k := item.Key()
			parsedKey, kErr := Parse(k)
			Checkf(kErr, "Error parsing key: %v, version: %d", k, item.Version())
			if !parsedKey.IsData() {
				continue
			}

			err := item.Value(func(v []byte) error {
				plist := &pb.PostingList{}
				Check(proto.Unmarshal(v, plist))
				VerifyPack(plist)
				if len(plist.Splits) == 0 {
					return nil
				}
				if plist.Splits[0] != uint64(1) {
					log.Panic("First split UID is not 1 baseKey: ", k,
						" version ", item.Version())
				}
				for _, uid := range plist.Splits {
					sKey, kErr := SplitKey(k, uid)
					Checkf(kErr,
						"Error creating split key from base key: %v, version: %d", k,
						item.Version())
					newTxn := pstore.NewTransactionAt(readTs, false)
					_, dbErr := newTxn.Get(sKey)
					if dbErr != nil {
						log.Panic("Snapshot verification failed: Unable to find splitKey: ",
							sKey, "\nbaseKey: ", " version: ", item.Version(),
							parsedKey, "\nSplits: ", plist.Splits,
						)
					}
				}
				return nil
			})
			Checkf(err, "Error getting value of key: %v version: %v", k, item.Version())

			if item.DiscardEarlierVersions() {
				break
			}
		}
		return nil, nil
	}
}

// VerifyPostingSplits checks if all the keys from parts are