Repository: coinbase/rosetta-cli
Pull request #76 (2020-07-24 22:03:36)

Implement FindTransaction Function
Related PR: #68 

### Changes
To support the `Construction API`, we need to search for processed transactions (to confirm them) after some confirmation depth has passed (to protect against reorgs).

### WARNING
This PR changes how transaction hashes are stored!



Patch:
diff --git a/internal/storage/block_storage.go b/internal/storage/block_storage.go
index 80fff8f..c1f5b2a 100644
--- a/internal/storage/block_storage.go
+++ b/internal/storage/block_storage.go
@@ -19,6 +19,7 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"math"
 
 	"github.com/coinbase/rosetta-sdk-go/syncer"
 	"github.com/coinbase/rosetta-sdk-go/types"
@@ -53,6 +54,10 @@ var (
 	// found in BlockStorage.
 	ErrBlockNotFound = errors.New("block not found")
 
+	// ErrTransactionNotFound is returned when a transaction
+	// is not found in BlockStorage.
+	ErrTransactionNotFound = errors.New("transaction not found")
+
 	// ErrDuplicateBlockHash is returned when a block hash
 	// cannot be stored because it is a duplicate.
 	ErrDuplicateBlockHash = errors.New("duplicate block hash")
@@ -60,10 +65,6 @@ var (
 	// ErrDuplicateTransactionHash is returned when a transaction
 	// hash cannot be stored because it is a duplicate.
 	ErrDuplicateTransactionHash = errors.New("duplicate transaction hash")
-
-	// ErrDuplicateKey is returned when trying to store a key that
-	// already exists (this is used by storeHash).
-	ErrDuplicateKey = errors.New("duplicate key")
 )
 
 func getHeadBlockKey() []byte {
@@ -76,12 +77,12 @@ func getBlockKey(blockIdentifier *types.BlockIdentifier) []byte {
 	)
 }
 
-func getBlockHashKey(hash string) []byte {
-	return []byte(fmt.Sprintf("%s/%s", blockHashNamespace, hash))
+func getBlockHashKey(blockIdentifier *types.BlockIdentifier) []byte {
+	return []byte(fmt.Sprintf("%s/%s", blockHashNamespace, blockIdentifier.Hash))
 }
 
-func getTransactionHashKey(blockHash string, transactionHash string) []byte {
-	return []byte(fmt.Sprintf("%s/%s/%s", transactionHashNamespace, blockHash, transactionHash))
+func getTransactionHashKey(transactionIdentifier *types.TransactionIdentifier) []byte {
+	return []byte(fmt.Sprintf("%s/%s", transactionHashNamespace, transactionIdentifier.Hash))
 }
 
 // BlockWorker is an interface that allows for work
@@ -214,33 +215,20 @@ func (b *BlockStorage) AddBlock(
 	}
 
 	// Store block hash
-	blockHashKey := getBlockHashKey(block.BlockIdentifier.Hash)
-	err = b.storeHash(ctx, transaction, blockHashKey)
-	if errors.Is(err, ErrDuplicateKey) {
-		return fmt.Errorf(
-			"%w %s",
-			ErrDuplicateBlockHash,
-			block.BlockIdentifier.Hash,
-		)
-	} else if err != nil {
+	err = b.storeBlockHash(ctx, transaction, block.BlockIdentifier)
+	if err != nil {
 		return fmt.Errorf("%w: unable to store block hash", err)
 	}
 
 	// Store all transaction hashes
 	for _, txn := range block.Transactions {
-		transactionHashKey := getTransactionHashKey(
-			block.BlockIdentifier.Hash,
-			txn.TransactionIdentifier.Hash,
+		err = b.storeTransactionHash(
+			ctx,
+			transaction,
+			block.BlockIdentifier,
+			txn.TransactionIdentifier,
 		)
-		err = b.storeHash(ctx, transaction, transactionHashKey)
-		if errors.Is(err, ErrDuplicateKey) {
-			return fmt.Errorf(
-				"%w transaction %s appears multiple times in block %s",
-				ErrDuplicateTransactionHash,
-				txn.TransactionIdentifier.Hash,
-				block.BlockIdentifier.Hash,
-			)
-		} else if err != nil {
+		if err != nil {
 			return fmt.Errorf("%w: unable to store transaction hash", err)
 		}
 	}
@@ -266,23 +254,20 @@ func (b *BlockStorage) RemoveBlock(
 
 	// Remove all transaction hashes
 	for _, txn := range block.Transactions {
-		err = transaction.Delete(
-			ctx,
-			getTransactionHashKey(block.BlockIdentifier.Hash, txn.TransactionIdentifier.Hash),
-		)
+		err = b.removeTransactionHash(ctx, transaction, blockIdentifier, txn.TransactionIdentifier)
 		if err != nil {
 			return err
 		}
 	}
 
 	// Remove block hash
-	err = transaction.Delete(ctx, getBlockHashKey(block.BlockIdentifier.Hash))
+	err = transaction.Delete(ctx, getBlockHashKey(blockIdentifier))
 	if err != nil {
 		return err
 	}
 
 	// Remove block
-	if err := transaction.Delete(ctx, getBlockKey(block.BlockIdentifier)); err != nil {
+	if err := transaction.Delete(ctx, getBlockKey(blockIdentifier)); err != nil {
 		return err
 	}
 
@@ -396,23 +381,161 @@ func (b *BlockStorage) CreateBlockCache(ctx context.Context) []*types.BlockIdent
 	return cache
 }
 
-// storeHash stores either a block or transaction hash.
-// TODO: store block hash in transaction hash value
-func (b *BlockStorage) storeHash(
+func (b *BlockStorage) storeBlockHash(
 	ctx context.Context,
 	transaction DatabaseTransaction,
-	hashKey []byte,
+	block *types.BlockIdentifier,
 ) error {
+	hashKey := getBlockHashKey(block)
 	exists, _, err := transaction.Get(ctx, hashKey)
 	if err != nil {
 		return err
 	}
 
 	if exists {
-		return ErrDuplicateKey
+		return fmt.Errorf("%w: duplicate block hash %s found", ErrDuplicateBlockHash, block.Hash)
 	}
 
 	return transaction.Set(ctx, hashKey, []byte(""))
 }
 
-// TODO: Wait until a transaction has a depth of X
+func (b *BlockStorage) storeTransactionHash(
+	ctx context.Context,
+	transaction DatabaseTransaction,
+	blockIdentifier *types.BlockIdentifier,
+	transactionIdentifier *types.TransactionIdentifier,
+) error {
+	hashKey := getTransactionHashKey(transactionIdentifier)
+	exists, val, err := transaction.Get(ctx, hashKey)
+	if err != nil {
+		return err
+	}
+
+	if !exists {
+		m := make(map[string]int64)
+		m[blockIdentifier.Hash] = blockIdentifier.Index
+		encodedResult, err := encode(m)
+		if err != nil {
+			return fmt.Errorf("%w: unable to encode transaction data", err)
+		}
+
+		return transaction.Set(ctx, hashKey, encodedResult)
+	}
+
+	var blocks map[string]int64
+	if err := decode(val, &blocks); err != nil {
+		return fmt.Errorf("%w: could not decode transaction hash contents", err)
+	}
+
+	if _, exists := blocks[blockIdentifier.Hash]; exists {
+		return fmt.Errorf(
+			"%w: duplicate transaction %s found in block %s:%d",
+			ErrDuplicateTransactionHash,
+			transactionIdentifier.Hash,
+			blockIdentifier.Hash,
+			blockIdentifier.Index,
+		)
+	}
+
+	blocks[blockIdentifier.Hash] = blockIdentifier.Index
+	encodedResult, err := encode(blocks)
+	if err != nil {
+		return fmt.Errorf("%w: unable to encode transaction data", err)
+	}
+
+	return transaction.Set(ctx, hashKey, encodedResult)
+}
+
+func (b *BlockStorage) removeTransactionHash(
+	ctx context.Context,
+	transaction DatabaseTransaction,
+	blockIdentifier *types.BlockIdentifier,
+	transactionIdentifier *types.TransactionIdentifier,
+) error {
+	hashKey := getTransactionHashKey(transactionIdentifier)
+	exists, val, err := transaction.Get(ctx, hashKey)
+	if err != nil {
+		return err
+	}
+
+	if !exists {
+		return fmt.Errorf("could not remove transaction %s", transactionIdentifier.Hash)
+	}
+
+	var blocks map[string]int64
+	if err := decode(val, &blocks); err != nil {
+		return fmt.Errorf("%w: could not decode transaction hash contents", err)
+	}
+
+	if _, exists := blocks[blockIdentifier.Hash]; !exists {
+		return fmt.Errorf("saved blocks at transaction does not contain %s", blockIdentifier.Hash)
+	}
+
+	delete(blocks, blockIdentifier.Hash)
+
+	if len(blocks) == 0 {
+		return transaction.Delete(ctx, hashKey)
+	}
+
+	encodedResult, err := encode(blocks)
+	if err != nil {
+		return fmt.Errorf("%w: unable to encode transaction data", err)
+	}
+
+	return transaction.Set(ctx, hashKey, encodedResult)
+}
+
+// FindTransaction returns the []*types.BlockIdentifier containing the
+// transaction and the depth from the current head of the first transaction
+// sigting (almost always this will just be a single block). If not found,
+// it returns a ErrTransactionNotFound error.
+func (b *BlockStorage) FindTransaction(
+	ctx context.Context,
+	transactionIdentifier *types.TransactionIdentifier,
+) ([]*types.BlockIdentifier, int64, error) {
+	txn := b.db.NewDatabaseTransaction(ctx, false)
+	defer txn.Discard(ctx)
+
+	txExists, tx, err := txn.Get(ctx, getTransactionHashKey(transactionIdentifier))
+	if err != nil {
+		return nil, -1, fmt.Errorf("%w: unable to query database for transaction", err)
+	}
+
+	if !txExists {
+		return nil, -1, nil
+	}
+
+	var blocks map[string]int64
+	if err := decode(tx, &blocks); err != nil {
+		return nil, -1, fmt.Errorf("%w: unable to decode block data for transaction", err)
+	}
+
+	blockExists, block, err := txn.Get(ctx, getHeadBlockKey())
+	if err != nil {
+		return nil, -1, fmt.Errorf("%w: unable to query database for head block", err)
+	}
+
+	if !blockExists {
+		// It should not be possible to enter this conditional
+		// as the existence of a tx hash implies at least
+		// one block is synced.
+		return nil, -1, ErrHeadBlockNotFound
+	}
+
+	var head types.BlockIdentifier
+	err = decode(block, &head)
+	if err != nil {
+		return nil, -1, fmt.Errorf("%w: could not decode head block", err)
+	}
+
+	ids := []*types.BlockIdentifier{}
+	oldestBlock := int64(math.MaxInt64)
+	for hash, index := range blocks {
+		ids = append(ids, &types.BlockIdentifier{Hash: hash, Index: index})
+		if index < oldestBlock {
+			oldestBlock = index
+		}
+	}
+
+	return ids, head.Index - oldestBlock, nil
+}


Test patch:
diff --git a/internal/storage/block_storage_test.go b/internal/storage/block_storage_test.go
index 49723ac..49abebc 100644
--- a/internal/storage/block_storage_test.go
+++ b/internal/storage/block_storage_test.go
@@ -262,6 +262,16 @@ func TestBlock(t *testing.T) {
 
 	storage := NewBlockStorage(database)
 
+	t.Run("Get non-existent tx", func(t *testing.T) {
+		txBlocks, headDistance, err := storage.FindTransaction(
+			ctx,
+			newBlock.Transactions[0].TransactionIdentifier,
+		)
+		assert.NoError(t, err)
+		assert.Nil(t, txBlocks)
+		assert.Equal(t, int64(-1), headDistance)
+	})
+
 	t.Run("Set and get block", func(t *testing.T) {
 		err := storage.AddBlock(ctx, newBlock)
 		assert.NoError(t, err)
@@ -273,6 +283,15 @@ func TestBlock(t *testing.T) {
 		head, err := storage.GetHeadBlockIdentifier(ctx)
 		assert.NoError(t, err)
 		assert.Equal(t, newBlock.BlockIdentifier, head)
+
+		txBlocks, headDistance, err := storage.FindTransaction(
+			ctx,
+			newBlock.Transactions[0].TransactionIdentifier,
+		)
+		assert.NoError(t, err)
+		assert.Len(t, txBlocks, 1)
+		assert.Equal(t, newBlock.BlockIdentifier, txBlocks[0])
+		assert.Equal(t, int64(0), headDistance)
 	})
 
 	t.Run("Get non-existent block", func(t *testing.T) {
@@ -287,11 +306,7 @@ func TestBlock(t *testing.T) {
 
 	t.Run("Set duplicate block hash", func(t *testing.T) {
 		err = storage.AddBlock(ctx, newBlock)
-		assert.EqualError(t, err, fmt.Errorf(
-			"%w %s",
-			ErrDuplicateBlockHash,
-			newBlock.BlockIdentifier.Hash,
-		).Error())
+		assert.Contains(t, err.Error(), ErrDuplicateBlockHash.Error())
 	})
 
 	t.Run("Set duplicate transaction hash (from prior block)", func(t *testing.T) {
@@ -305,6 +320,19 @@ func TestBlock(t *testing.T) {
 		head, err := storage.GetHeadBlockIdentifier(ctx)
 		assert.NoError(t, err)
 		assert.Equal(t, newBlock2.BlockIdentifier, head)
+
+		txBlocks, headDistance, err := storage.FindTransaction(
+			ctx,
+			newBlock.Transactions[0].TransactionIdentifier,
+		)
+		assert.NoError(t, err)
+		assert.Len(t, txBlocks, 2)
+		assert.ElementsMatch(
+			t,
+			[]*types.BlockIdentifier{newBlock.BlockIdentifier, newBlock2.BlockIdentifier},
+			txBlocks,
+		)
+		assert.Equal(t, int64(1), headDistance)
 	})
 
 	t.Run("Remove block and re-set block of same hash", func(t *testing.T) {
@@ -321,6 +349,19 @@ func TestBlock(t *testing.T) {
 		head, err = storage.GetHeadBlockIdentifier(ctx)
 		assert.NoError(t, err)
 		assert.Equal(t, newBlock2.BlockIdentifier, head)
+
+		txBlocks, headDistance, err := storage.FindTransaction(
+			ctx,
+			newBlock.Transactions[0].TransactionIdentifier,
+		)
+		assert.NoError(t, err)
+		assert.Len(t, txBlocks, 2)
+		assert.ElementsMatch(
+			t,
+			[]*types.BlockIdentifier{newBlock.BlockIdentifier, newBlock2.BlockIdentifier},
+			txBlocks,
+		)
+		assert.Equal(t, int64(1), headDistance)
 	})
 
 	t.Run("Add block with complex metadata", func(t *testing.T) {
@@ -338,12 +379,7 @@ func TestBlock(t *testing.T) {
 
 	t.Run("Set duplicate transaction hash (same block)", func(t *testing.T) {
 		err = storage.AddBlock(ctx, duplicateTxBlock)
-		assert.EqualError(t, err, fmt.Errorf(
-			"%w transaction %s appears multiple times in block %s",
-			ErrDuplicateTransactionHash,
-			"blahTx3",
-			"blah 4",
-		).Error())
+		assert.Contains(t, err.Error(), ErrDuplicateTransactionHash.Error())
 
 		head, err := storage.GetHeadBlockIdentifier(ctx)
 		assert.NoError(t, err)