kv indexer: add separator to start key when matching ranges (#2925)

* kv indexer: add separator to start key when matching ranges

to avoid including false positives

Refs #2908

* refactor code

* add a test case

state/txindex/kv/kv.go

@@ -172,10 +172,10 @@ func (txi *TxIndex) Search(q *query.Query) ([]*types.TxResult, error) {
 
 		for _, r := range ranges {
 			if !hashesInitialized {
-				hashes = txi.matchRange(r, []byte(r.key))
+				hashes = txi.matchRange(r, startKey(r.key))
 				hashesInitialized = true
 			} else {
-				hashes = intersect(hashes, txi.matchRange(r, []byte(r.key)))
+				hashes = intersect(hashes, txi.matchRange(r, startKey(r.key)))
 			}
 		}
 	}
@@ -190,10 +190,10 @@ func (txi *TxIndex) Search(q *query.Query) ([]*types.TxResult, error) {
 		}
 
 		if !hashesInitialized {
-			hashes = txi.match(c, startKey(c, height))
+			hashes = txi.match(c, startKeyForCondition(c, height))
 			hashesInitialized = true
 		} else {
-			hashes = intersect(hashes, txi.match(c, startKey(c, height)))
+			hashes = intersect(hashes, txi.match(c, startKeyForCondition(c, height)))
 		}
 	}
 
@@ -359,14 +359,14 @@ func (txi *TxIndex) match(c query.Condition, startKey []byte) (hashes [][]byte)
 	return
 }
 
-func (txi *TxIndex) matchRange(r queryRange, prefix []byte) (hashes [][]byte) {
+func (txi *TxIndex) matchRange(r queryRange, startKey []byte) (hashes [][]byte) {
 	// create a map to prevent duplicates
 	hashesMap := make(map[string][]byte)
 
 	lowerBound := r.lowerBoundValue()
 	upperBound := r.upperBoundValue()
 
-	it := dbm.IteratePrefix(txi.store, prefix)
+	it := dbm.IteratePrefix(txi.store, startKey)
 	defer it.Close()
 LOOP:
 	for ; it.Valid(); it.Next() {
@@ -409,16 +409,6 @@ LOOP:
 ///////////////////////////////////////////////////////////////////////////////
 // Keys
 
-func startKey(c query.Condition, height int64) []byte {
-	var key string
-	if height > 0 {
-		key = fmt.Sprintf("%s/%v/%d/", c.Tag, c.Operand, height)
-	} else {
-		key = fmt.Sprintf("%s/%v/", c.Tag, c.Operand)
-	}
-	return []byte(key)
-}
-
 func isTagKey(key []byte) bool {
 	return strings.Count(string(key), tagKeySeparator) == 3
 }
@@ -429,11 +419,36 @@ func extractValueFromKey(key []byte) string {
 }
 
 func keyForTag(tag cmn.KVPair, result *types.TxResult) []byte {
-	return []byte(fmt.Sprintf("%s/%s/%d/%d", tag.Key, tag.Value, result.Height, result.Index))
+	return []byte(fmt.Sprintf("%s/%s/%d/%d",
+		tag.Key,
+		tag.Value,
+		result.Height,
+		result.Index,
+	))
 }
 
 func keyForHeight(result *types.TxResult) []byte {
-	return []byte(fmt.Sprintf("%s/%d/%d/%d", types.TxHeightKey, result.Height, result.Height, result.Index))
+	return []byte(fmt.Sprintf("%s/%d/%d/%d",
+		types.TxHeightKey,
+		result.Height,
+		result.Height,
+		result.Index,
+	))
+}
+
+func startKeyForCondition(c query.Condition, height int64) []byte {
+	if height > 0 {
+		return startKey(c.Tag, c.Operand, height)
+	}
+	return startKey(c.Tag, c.Operand)
+}
+
+func startKey(fields ...interface{}) []byte {
+	var b bytes.Buffer
+	for _, f := range fields {
+		b.Write([]byte(fmt.Sprintf("%v", f) + tagKeySeparator))
+	}
+	return b.Bytes()
 }
 
 ///////////////////////////////////////////////////////////////////////////////

state/txindex/kv/kv_test.go

@@ -126,7 +126,7 @@ func TestTxSearchOneTxWithMultipleSameTagsButDifferentValues(t *testing.T) {
 }
 
 func TestTxSearchMultipleTxs(t *testing.T) {
-	allowedTags := []string{"account.number"}
+	allowedTags := []string{"account.number", "account.number.id"}
 	indexer := NewTxIndex(db.NewMemDB(), IndexTags(allowedTags))
 
 	// indexed first, but bigger height (to test the order of transactions)
@@ -160,6 +160,17 @@ func TestTxSearchMultipleTxs(t *testing.T) {
 	err = indexer.Index(txResult3)
 	require.NoError(t, err)
 
+	// indexed fourth (to test we don't include txs with similar tags)
+	// https://github.com/tendermint/tendermint/issues/2908
+	txResult4 := txResultWithTags([]cmn.KVPair{
+		{Key: []byte("account.number.id"), Value: []byte("1")},
+	})
+	txResult4.Tx = types.Tx("Mike's account")
+	txResult4.Height = 2
+	txResult4.Index = 2
+	err = indexer.Index(txResult4)
+	require.NoError(t, err)
+
 	results, err := indexer.Search(query.MustParse("account.number >= 1"))
 	assert.NoError(t, err)