Merge pull request #170 from vulcanize/v1.10.13-statediff-0.0.29

log processing bug fix

Author: i-norden

statediff/indexer/indexer.go

@@ -299,7 +299,7 @@ type processArgs struct {
 	rctTrieNodes    []*ipld.EthRctTrie
 	txNodes         []*ipld.EthTx
 	txTrieNodes     []*ipld.EthTxTrie
-	logTrieNodes    [][]*ipld.EthLogTrie
+	logTrieNodes    [][]node.Node
 	logLeafNodeCIDs [][]cid.Cid
 	rctLeafNodeCIDs []cid.Cid
 }

statediff/indexer/indexer_test.go

@@ -49,6 +49,7 @@ var (
 	mockBlock                                              *types.Block
 	headerCID, trx1CID, trx2CID, trx3CID, trx4CID, trx5CID cid.Cid
 	rct1CID, rct2CID, rct3CID, rct4CID, rct5CID            cid.Cid
+	rctLeaf1, rctLeaf2, rctLeaf3, rctLeaf4, rctLeaf5       []byte
 	state1CID, state2CID, storageCID                       cid.Cid
 )
 
@@ -124,14 +125,49 @@ func init() {
 	trx3CID, _ = ipld.RawdataToCid(ipld.MEthTx, tx3, multihash.KECCAK_256)
 	trx4CID, _ = ipld.RawdataToCid(ipld.MEthTx, tx4, multihash.KECCAK_256)
 	trx5CID, _ = ipld.RawdataToCid(ipld.MEthTx, tx5, multihash.KECCAK_256)
-	rct1CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, rct1, multihash.KECCAK_256)
-	rct2CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, rct2, multihash.KECCAK_256)
-	rct3CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, rct3, multihash.KECCAK_256)
-	rct4CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, rct4, multihash.KECCAK_256)
-	rct5CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, rct5, multihash.KECCAK_256)
+	/*
+		rct1Node, _ := ipld.NewReceipt(rcts[0])
+		rct2Node, _ := ipld.NewReceipt(rcts[1])
+		rct3Node, _ := ipld.NewReceipt(rcts[2])
+		rct4Node, _ := ipld.NewReceipt(rcts[3])
+		rct5Node, _ := ipld.NewReceipt(rcts[4])
+	*/
 	state1CID, _ = ipld.RawdataToCid(ipld.MEthStateTrie, mocks.ContractLeafNode, multihash.KECCAK_256)
 	state2CID, _ = ipld.RawdataToCid(ipld.MEthStateTrie, mocks.AccountLeafNode, multihash.KECCAK_256)
 	storageCID, _ = ipld.RawdataToCid(ipld.MEthStorageTrie, mocks.StorageLeafNode, multihash.KECCAK_256)
+
+	receiptTrie := ipld.NewRctTrie()
+
+	receiptTrie.Add(0, rct1)
+	receiptTrie.Add(1, rct2)
+	receiptTrie.Add(2, rct3)
+	receiptTrie.Add(3, rct4)
+	receiptTrie.Add(4, rct5)
+
+	rctLeafNodes, keys, _ := receiptTrie.GetLeafNodes()
+
+	rctleafNodeCids := make([]cid.Cid, len(rctLeafNodes))
+	orderedRctLeafNodes := make([][]byte, len(rctLeafNodes))
+	for i, rln := range rctLeafNodes {
+		var idx uint
+
+		r := bytes.NewReader(keys[i].TrieKey)
+		rlp.Decode(r, &idx)
+		rctleafNodeCids[idx] = rln.Cid()
+		orderedRctLeafNodes[idx] = rln.RawData()
+	}
+
+	rct1CID = rctleafNodeCids[0]
+	rct2CID = rctleafNodeCids[1]
+	rct3CID = rctleafNodeCids[2]
+	rct4CID = rctleafNodeCids[3]
+	rct5CID = rctleafNodeCids[4]
+
+	rctLeaf1 = orderedRctLeafNodes[0]
+	rctLeaf2 = orderedRctLeafNodes[1]
+	rctLeaf3 = orderedRctLeafNodes[2]
+	rctLeaf4 = orderedRctLeafNodes[3]
+	rctLeaf5 = orderedRctLeafNodes[4]
 }
 
 func setup(t *testing.T) {
@@ -416,7 +452,7 @@ func TestPublishAndIndexer(t *testing.T) {
 
 			switch c {
 			case rct1CID.String():
-				shared.ExpectEqual(t, data, rct1)
+				shared.ExpectEqual(t, data, rctLeaf1)
 				var postStatus uint64
 				pgStr = `SELECT post_status FROM eth.receipt_cids WHERE leaf_cid = $1`
 				err = db.Get(&postStatus, pgStr, c)
@@ -425,7 +461,7 @@ func TestPublishAndIndexer(t *testing.T) {
 				}
 				shared.ExpectEqual(t, postStatus, mocks.ExpectedPostStatus)
 			case rct2CID.String():
-				shared.ExpectEqual(t, data, rct2)
+				shared.ExpectEqual(t, data, rctLeaf2)
 				var postState string
 				pgStr = `SELECT post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
 				err = db.Get(&postState, pgStr, c)
@@ -434,7 +470,7 @@ func TestPublishAndIndexer(t *testing.T) {
 				}
 				shared.ExpectEqual(t, postState, mocks.ExpectedPostState1)
 			case rct3CID.String():
-				shared.ExpectEqual(t, data, rct3)
+				shared.ExpectEqual(t, data, rctLeaf3)
 				var postState string
 				pgStr = `SELECT post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
 				err = db.Get(&postState, pgStr, c)
@@ -443,7 +479,7 @@ func TestPublishAndIndexer(t *testing.T) {
 				}
 				shared.ExpectEqual(t, postState, mocks.ExpectedPostState2)
 			case rct4CID.String():
-				shared.ExpectEqual(t, data, rct4)
+				shared.ExpectEqual(t, data, rctLeaf4)
 				var postState string
 				pgStr = `SELECT post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
 				err = db.Get(&postState, pgStr, c)
@@ -452,7 +488,7 @@ func TestPublishAndIndexer(t *testing.T) {
 				}
 				shared.ExpectEqual(t, postState, mocks.ExpectedPostState3)
 			case rct5CID.String():
-				shared.ExpectEqual(t, data, rct5)
+				shared.ExpectEqual(t, data, rctLeaf5)
 				var postState string
 				pgStr = `SELECT post_state FROM eth.receipt_cids WHERE leaf_cid = $1`
 				err = db.Get(&postState, pgStr, c)

statediff/indexer/ipfs/ipld/eth_log_trie.go

@@ -3,6 +3,8 @@ package ipld
 import (
 	"fmt"
 
+	node "github.com/ipfs/go-ipld-format"
+
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ipfs/go-cid"
@@ -91,13 +93,13 @@ func newLogTrie() *logTrie {
 // getNodes invokes the localTrie, which computes the root hash of the
 // log trie and returns its database keys, to return a slice
 // of EthLogTrie nodes.
-func (rt *logTrie) getNodes() ([]*EthLogTrie, error) {
+func (rt *logTrie) getNodes() ([]node.Node, error) {
 	keys, err := rt.getKeys()
 	if err != nil {
 		return nil, err
 	}
 
-	out := make([]*EthLogTrie, 0, len(keys))
+	out := make([]node.Node, 0, len(keys))
 	for _, k := range keys {
 		n, err := rt.getNodeFromDB(k)
 		if err != nil {
@@ -114,14 +116,8 @@ func (rt *logTrie) getNodeFromDB(key []byte) (*EthLogTrie, error) {
 	if err != nil {
 		return nil, err
 	}
-
-	c, err := RawdataToCid(MEthLogTrie, rawdata, multihash.KECCAK_256)
-	if err != nil {
-		return nil, err
-	}
-
 	tn := &TrieNode{
-		cid:     c,
+		cid:     keccak256ToCid(MEthLogTrie, key),
 		rawdata: rawdata,
 	}
 	return &EthLogTrie{TrieNode: tn}, nil

statediff/indexer/ipfs/ipld/eth_parser.go

@@ -23,6 +23,8 @@ import (
 	"io"
 	"io/ioutil"
 
+	node "github.com/ipfs/go-ipld-format"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/rlp"
@@ -123,7 +125,7 @@ func FromBlockJSON(r io.Reader) (*EthHeader, []*EthTx, []*EthTxTrie, error) {
 
 // FromBlockAndReceipts takes a block and processes it
 // to return it a set of IPLD nodes for further processing.
-func FromBlockAndReceipts(block *types.Block, receipts []*types.Receipt) (*EthHeader, []*EthHeader, []*EthTx, []*EthTxTrie, []*EthReceipt, []*EthRctTrie, [][]*EthLogTrie, [][]cid.Cid, []cid.Cid, error) {
+func FromBlockAndReceipts(block *types.Block, receipts []*types.Receipt) (*EthHeader, []*EthHeader, []*EthTx, []*EthTxTrie, []*EthReceipt, []*EthRctTrie, [][]node.Node, [][]cid.Cid, []cid.Cid, error) {
 	// Process the header
 	headerNode, err := NewEthHeader(block.Header())
 	if err != nil {
@@ -148,10 +150,10 @@ func FromBlockAndReceipts(block *types.Block, receipts []*types.Receipt) (*EthHe
 	}
 
 	// Process the receipts and logs
-	rctNodes, tctTrieNodes, logTrieNodes, logLeafNodeCIDs, rctLeafNodeCIDs, err := processReceiptsAndLogs(receipts,
+	rctNodes, tctTrieNodes, logTrieAndLogNodes, logLeafNodeCIDs, rctLeafNodeCIDs, err := processReceiptsAndLogs(receipts,
 		block.Header().ReceiptHash[:])
 
-	return headerNode, uncleNodes, txNodes, txTrieNodes, rctNodes, tctTrieNodes, logTrieNodes, logLeafNodeCIDs, rctLeafNodeCIDs, err
+	return headerNode, uncleNodes, txNodes, txTrieNodes, rctNodes, tctTrieNodes, logTrieAndLogNodes, logLeafNodeCIDs, rctLeafNodeCIDs, err
 }
 
 // processTransactions will take the found transactions in a parsed block body
@@ -180,11 +182,11 @@ func processTransactions(txs []*types.Transaction, expectedTxRoot []byte) ([]*Et
 
 // processReceiptsAndLogs will take in receipts
 // to return IPLD node slices for eth-rct, eth-rct-trie, eth-log, eth-log-trie, eth-log-trie-CID, eth-rct-trie-CID
-func processReceiptsAndLogs(rcts []*types.Receipt, expectedRctRoot []byte) ([]*EthReceipt, []*EthRctTrie, [][]*EthLogTrie, [][]cid.Cid, []cid.Cid, error) {
+func processReceiptsAndLogs(rcts []*types.Receipt, expectedRctRoot []byte) ([]*EthReceipt, []*EthRctTrie, [][]node.Node, [][]cid.Cid, []cid.Cid, error) {
 	// Pre allocating memory.
 	ethRctNodes := make([]*EthReceipt, 0, len(rcts))
 	ethLogleafNodeCids := make([][]cid.Cid, 0, len(rcts))
-	ethLogTrieNodes := make([][]*EthLogTrie, 0, len(rcts))
+	ethLogTrieAndLogNodes := make([][]node.Node, 0, len(rcts))
 
 	receiptTrie := NewRctTrie()
 
@@ -195,7 +197,7 @@ func processReceiptsAndLogs(rcts []*types.Receipt, expectedRctRoot []byte) ([]*E
 			return nil, nil, nil, nil, nil, err
 		}
 		rct.LogRoot = logTrieHash
-		ethLogTrieNodes = append(ethLogTrieNodes, logTrieNodes)
+		ethLogTrieAndLogNodes = append(ethLogTrieAndLogNodes, logTrieNodes)
 		ethLogleafNodeCids = append(ethLogleafNodeCids, leafNodeCids)
 
 		ethRct, err := NewReceipt(rct)
@@ -235,22 +237,38 @@ func processReceiptsAndLogs(rcts []*types.Receipt, expectedRctRoot []byte) ([]*E
 		ethRctleafNodeCids[idx] = rln.Cid()
 	}
 
-	return ethRctNodes, rctTrieNodes, ethLogTrieNodes, ethLogleafNodeCids, ethRctleafNodeCids, err
+	return ethRctNodes, rctTrieNodes, ethLogTrieAndLogNodes, ethLogleafNodeCids, ethRctleafNodeCids, err
 }
 
-func processLogs(logs []*types.Log) ([]*EthLogTrie, []cid.Cid, common.Hash, error) {
+const keccak256Length = 32
+
+func processLogs(logs []*types.Log) ([]node.Node, []cid.Cid, common.Hash, error) {
 	logTr := newLogTrie()
+	shortLog := make(map[uint64]*EthLog, len(logs))
 	for idx, log := range logs {
-		ethLog, err := NewLog(log)
+		logRaw, err := rlp.EncodeToBytes(log)
 		if err != nil {
 			return nil, nil, common.Hash{}, err
 		}
-		if err = logTr.Add(idx, ethLog.RawData()); err != nil {
+		// if len(logRaw) <= keccak256Length it is possible this value's "leaf node"
+		// will be stored in its parent branch but only if len(partialPathOfTheNode) + len(logRaw) <= keccak256Length
+		// But we can't tell what the partial path will be until the trie is Commit()-ed
+		// So wait until we collect all the leaf nodes, and if we are missing any at the indexes we note in shortLogCIDs
+		// we know that these "leaf nodes" were internalized into their parent branch node and we move forward with
+		// using the cid.Cid we cached in shortLogCIDs
+		if len(logRaw) <= keccak256Length {
+			logNode, err := NewLog(log)
+			if err != nil {
+				return nil, nil, common.Hash{}, err
+			}
+			shortLog[uint64(idx)] = logNode
+		}
+		if err = logTr.Add(idx, logRaw); err != nil {
 			return nil, nil, common.Hash{}, err
 		}
 	}
 
-	logTrieNodes, err := logTr.getNodes()
+	logTrieAndLogNodes, err := logTr.getNodes()
 	if err != nil {
 		return nil, nil, common.Hash{}, err
 	}
@@ -259,8 +277,7 @@ func processLogs(logs []*types.Log) ([]*EthLogTrie, []cid.Cid, common.Hash, erro
 	if err != nil {
 		return nil, nil, common.Hash{}, err
 	}
-
-	leafNodeCids := make([]cid.Cid, len(leafNodes))
+	leafNodeCids := make([]cid.Cid, len(logs))
 	for i, ln := range leafNodes {
 		var idx uint
 
@@ -271,6 +288,15 @@ func processLogs(logs []*types.Log) ([]*EthLogTrie, []cid.Cid, common.Hash, erro
 		}
 		leafNodeCids[idx] = ln.Cid()
 	}
+	// this is where we check which logs <= keccak256Length were actually internalized into parent branch node
+	// and replace those that were with the cid.Cid for the raw log IPLD
+	for i, l := range shortLog {
+		if !leafNodeCids[i].Defined() {
+			leafNodeCids[i] = l.Cid()
+			// if the leaf node was internalized, we append an IPLD for log itself to the list of IPLDs we need to publish
+			logTrieAndLogNodes = append(logTrieAndLogNodes, l)
+		}
+	}
 
-	return logTrieNodes, leafNodeCids, common.BytesToHash(logTr.rootHash()), err
+	return logTrieAndLogNodes, leafNodeCids, common.BytesToHash(logTr.rootHash()), err
 }

statediff/indexer/ipfs/ipld/eth_receipt_trie.go

@@ -166,14 +166,8 @@ func (rt *rctTrie) getNodeFromDB(key []byte) (*EthRctTrie, error) {
 	if err != nil {
 		return nil, err
 	}
-
-	cid, err := RawdataToCid(MEthTxReceiptTrie, rawdata, multihash.KECCAK_256)
-	if err != nil {
-		return nil, err
-	}
-
 	tn := &TrieNode{
-		cid:     cid,
+		cid:     keccak256ToCid(MEthTxReceiptTrie, key),
 		rawdata: rawdata,
 	}
 

statediff/indexer/ipfs/ipld/eth_tx_trie.go

@@ -135,12 +135,8 @@ func (tt *txTrie) getNodes() ([]*EthTxTrie, error) {
 		if err != nil {
 			return nil, err
 		}
-		c, err := RawdataToCid(MEthTxTrie, rawdata, multihash.KECCAK_256)
-		if err != nil {
-			return nil, err
-		}
 		tn := &TrieNode{
-			cid:     c,
+			cid:     keccak256ToCid(MEthTxTrie, k),
 			rawdata: rawdata,
 		}
 		out = append(out, &EthTxTrie{TrieNode: tn})