cerc-io/ipld-eth-server
12
0
Fork 1
Code Issues 10 Pull Requests 2 Actions Packages Projects Releases 37 Wiki Activity

eth json rpc optimizations; begin ipfs-ethdb integration

Browse Source
This commit is contained in:
Ian Norden
2020-08-14 13:14:19 -05:00
parent d842a5e796
commit 611fc2c11e
11 changed files with 77 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/eth/backend.go
+38 -100
View File
@@ -22,15 +22,17 @@ import (
"fmt"
"math/big"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
ipfsethdb "github.com/vulcanize/pg-ipfs-ethdb"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/ipfs"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/postgres"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
var (
@@ -41,13 +43,16 @@ type Backend struct {
Retriever *CIDRetriever
Fetcher *IPLDFetcher
DB *postgres.DB
EthDB ethdb.Database
}
func NewEthBackend(db *postgres.DB) (*Backend, error) {
r := NewCIDRetriever(db)
return &Backend{
Retriever: r,
Fetcher: NewIPLDFetcher(db),
DB: db,
EthDB: ipfsethdb.NewDatabase(db.DB),
}, nil
}
@@ -64,42 +69,16 @@ func (b *Backend) HeaderByNumber(ctx context.Context, blockNumber rpc.BlockNumbe
return nil, errPendingBlockNumber
}
// Begin tx
tx, err := b.DB.Beginx()
if err != nil {
return nil, err
var headerBytes []byte
pgStr := `SELECT DISTINCT ON (block_number) data
FROM public.blocks INNER JOIN eth.header_cids ON (header_cids.mh_key = blocks.key)
WHERE block_number = $1
ORDER BY block_number, times_validated DESC`
if err := b.DB.Get(&headerBytes, pgStr, number); err != nil {
return nil, fmt.Errorf("header at block %d is not available; err %s", number, err.Error())
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
// Retrieve the CIDs for headers at this height
headerCids, err := b.Retriever.RetrieveHeaderCIDs(tx, number)
if err != nil {
return nil, err
}
// If there are none, throw an error
if len(headerCids) < 1 {
return nil, fmt.Errorf("header at block %d is not available", number)
}
// Fetch the header IPLDs for those CIDs
headerIPLD, err := b.Fetcher.FetchHeader(tx, headerCids[0])
if err != nil {
return nil, err
}
// Decode the first header at this block height and return it
// We throw an error in FetchHeaders() if the number of headers does not match the number of CIDs and we already
// confirmed the number of CIDs is greater than 0 so there is no need to bound check the slice before accessing
var header types.Header
err = rlp.DecodeBytes(headerIPLD.Data, &header)
return &header, err
header := new(types.Header)
return header, rlp.DecodeBytes(headerBytes, header)
}
// GetTd retrieves and returns the total difficulty at the given block hash
@@ -120,41 +99,24 @@ func (b *Backend) GetTd(blockHash common.Hash) (*big.Int, error) {
// GetLogs returns all the logs for the given block hash
func (b *Backend) GetLogs(ctx context.Context, hash common.Hash) ([][]*types.Log, error) {
// Begin tx
tx, err := b.DB.Beginx()
if err != nil {
pgStr := `SELECT blocks.data FROM public.blocks, eth.receipt_cids, eth.transaction_cids, eth.header_cids
WHERE header_cids.block_hash = $1
AND receipt_cids.tx_id = transaction_cids.id
AND transaction_cids.header_id = header_cids.id
ORDER BY transaction_cids.index ASC`
var receiptBytes [][]byte
if err := b.DB.Select(&receiptBytes, pgStr, hash.Hex()); err != nil {
return nil, err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
receiptCIDs, err := b.Retriever.RetrieveRctCIDs(tx, ReceiptFilter{}, 0, &hash, nil)
if err != nil {
return nil, err
}
if len(receiptCIDs) == 0 {
return nil, nil
}
receiptIPLDs, err := b.Fetcher.FetchRcts(tx, receiptCIDs)
if err != nil {
return nil, err
}
logs := make([][]*types.Log, len(receiptIPLDs))
for i, rctIPLD := range receiptIPLDs {
logs := make([][]*types.Log, len(receiptBytes))
for i, rctBytes := range receiptBytes {
var rct types.Receipt
if err := rlp.DecodeBytes(rctIPLD.Data, &rct); err != nil {
if err := rlp.DecodeBytes(rctBytes, &rct); err != nil {
return nil, err
}
logs[i] = rct.Logs
}
return logs, err
return logs, nil
}
// BlockByNumber returns the requested canonical block.
@@ -326,48 +288,24 @@ func (b *Backend) BlockByHash(ctx context.Context, hash common.Hash) (*types.Blo
// GetTransaction retrieves a tx by hash
// It also returns the blockhash, blocknumber, and tx index associated with the transaction
func (b *Backend) GetTransaction(ctx context.Context, txHash common.Hash) (*types.Transaction, common.Hash, uint64, uint64, error) {
pgStr := `SELECT transaction_cids.mh_key, transaction_cids.index, header_cids.block_hash, header_cids.block_number
FROM eth.transaction_cids, eth.header_cids
WHERE transaction_cids.header_id = header_cids.id
AND transaction_cids.tx_hash = $1`
var txCIDWithHeaderInfo struct {
MhKey string `db:"mh_key"`
Index int64 `db:"index"`
var tempTxStruct struct {
Data []byte `db:"data"`
BlockHash string `db:"block_hash"`
BlockNumber int64 `db:"block_number"`
BlockNumber uint64 `db:"block_number"`
Index uint64 `db:"index"`
}
if err := b.DB.Get(&txCIDWithHeaderInfo, pgStr, txHash.String()); err != nil {
return nil, common.Hash{}, 0, 0, err
}
// Begin tx
tx, err := b.DB.Beginx()
if err != nil {
return nil, common.Hash{}, 0, 0, err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
txIPLD, err := b.Fetcher.FetchTrxs(tx, []TxModel{{MhKey: txCIDWithHeaderInfo.MhKey}})
if err != nil {
return nil, common.Hash{}, 0, 0, err
}
if err := tx.Commit(); err != nil {
pgStr := `SELECT blocks.data, block_hash, block_number, index FROM public.blocks, eth.transaction_cids, eth.header_cids
WHERE blocks.key = transaction_cids.mh_key
AND transaction_cids.header_id = header_cids.id
AND transaction_cids.tx_hash = $1`
if err := b.DB.Get(&tempTxStruct, pgStr, txHash.String()); err != nil {
return nil, common.Hash{}, 0, 0, err
}
var transaction types.Transaction
if err := rlp.DecodeBytes(txIPLD[0].Data, &transaction); err != nil {
if err := rlp.DecodeBytes(tempTxStruct.Data, &transaction); err != nil {
return nil, common.Hash{}, 0, 0, err
}
return &transaction, common.HexToHash(txCIDWithHeaderInfo.BlockHash), uint64(txCIDWithHeaderInfo.BlockNumber), uint64(txCIDWithHeaderInfo.Index), err
return &transaction, common.HexToHash(tempTxStruct.BlockHash), tempTxStruct.BlockNumber, tempTxStruct.Index, nil
}
// extractLogsOfInterest returns logs from the receipt IPLD
pkg/eth/cid_retriever.go
+2 -1
View File
@@ -187,7 +187,8 @@ func (ecr *CIDRetriever) RetrieveTxCIDs(tx *sqlx.Tx, txFilter TxFilter, headerID
id := 1
pgStr := fmt.Sprintf(`SELECT transaction_cids.id, transaction_cids.header_id,
transaction_cids.tx_hash, transaction_cids.cid, transaction_cids.mh_key,
transaction_cids.dst, transaction_cids.src, transaction_cids.index
transaction_cids.dst, transaction_cids.src, transaction_cids.index,
transaction_cids.tx_data, transaction_cids.deployment
FROM eth.transaction_cids INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.id)
WHERE header_cids.id = $%d`, id)
args = append(args, headerID)
pkg/eth/eth_suite_test.go
+1 -1
View File
@@ -27,7 +27,7 @@ import (
func TestETHWatcher(t *testing.T) {
RegisterFailHandler(Fail)
RunSpecs(t, "ETH IPFS WatcherSuite Test")
RunSpecs(t, "ETH IPFS Watcher Suite Test")
}
var _ = BeforeSuite(func() {
pkg/eth/indexer.go
+4 -4
View File
@@ -109,8 +109,8 @@ func (in *CIDIndexer) indexUncleCID(tx *sqlx.Tx, uncle UncleModel, headerID int6
func (in *CIDIndexer) indexTransactionAndReceiptCIDs(tx *sqlx.Tx, payload *CIDPayload, headerID int64) error {
for _, trxCidMeta := range payload.TransactionCIDs {
var txID int64
err := tx.QueryRowx(`INSERT INTO eth.transaction_cids (header_id, tx_hash, cid, dst, src, index, mh_key, data, deployment) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (header_id, tx_hash) DO UPDATE SET (cid, dst, src, index, mh_key, data, deployment) = ($3, $4, $5, $6, $7, $8, $9)
err := tx.QueryRowx(`INSERT INTO eth.transaction_cids (header_id, tx_hash, cid, dst, src, index, mh_key, tx_data, deployment) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (header_id, tx_hash) DO UPDATE SET (cid, dst, src, index, mh_key, tx_data, deployment) = ($3, $4, $5, $6, $7, $8, $9)
RETURNING id`,
headerID, trxCidMeta.TxHash, trxCidMeta.CID, trxCidMeta.Dst, trxCidMeta.Src, trxCidMeta.Index, trxCidMeta.MhKey, trxCidMeta.Data, trxCidMeta.Deployment).Scan(&txID)
if err != nil {
@@ -128,8 +128,8 @@ func (in *CIDIndexer) indexTransactionAndReceiptCIDs(tx *sqlx.Tx, payload *CIDPa
func (in *CIDIndexer) indexTransactionCID(tx *sqlx.Tx, transaction TxModel, headerID int64) (int64, error) {
var txID int64
err := tx.QueryRowx(`INSERT INTO eth.transaction_cids (header_id, tx_hash, cid, dst, src, index, mh_key, data, deployment) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (header_id, tx_hash) DO UPDATE SET (cid, dst, src, index, mh_key, data, deployment) = ($3, $4, $5, $6, $7, $8, $9)
err := tx.QueryRowx(`INSERT INTO eth.transaction_cids (header_id, tx_hash, cid, dst, src, index, mh_key, tx_data, deployment) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (header_id, tx_hash) DO UPDATE SET (cid, dst, src, index, mh_key, tx_data, deployment) = ($3, $4, $5, $6, $7, $8, $9)
RETURNING id`,
headerID, transaction.TxHash, transaction.CID, transaction.Dst, transaction.Src, transaction.Index, transaction.MhKey, transaction.Data, transaction.Deployment).Scan(&txID)
return txID, err
pkg/eth/mocks/test_data.go
+24 -18
View File
@@ -132,28 +132,34 @@ var (
}
MockTrxMetaPostPublsh = []eth.TxModel{
{
CID: Trx1CID.String(), // This is empty until we go to publish to ipfs
MhKey: Trx1MhKey,
Src: SenderAddr.Hex(),
Dst: Address.String(),
Index: 0,
TxHash: MockTransactions[0].Hash().String(),
CID: Trx1CID.String(), // This is empty until we go to publish to ipfs
MhKey: Trx1MhKey,
Src: SenderAddr.Hex(),
Dst: Address.String(),
Index: 0,
TxHash: MockTransactions[0].Hash().String(),
Data: []byte{},
Deployment: false,
},
{
CID: Trx2CID.String(),
MhKey: Trx2MhKey,
Src: SenderAddr.Hex(),
Dst: AnotherAddress.String(),
Index: 1,
TxHash: MockTransactions[1].Hash().String(),
CID: Trx2CID.String(),
MhKey: Trx2MhKey,
Src: SenderAddr.Hex(),
Dst: AnotherAddress.String(),
Index: 1,
TxHash: MockTransactions[1].Hash().String(),
Data: []byte{},
Deployment: false,
},
{
CID: Trx3CID.String(),
MhKey: Trx3MhKey,
Src: SenderAddr.Hex(),
Dst: "",
Index: 2,
TxHash: MockTransactions[2].Hash().String(),
CID: Trx3CID.String(),
MhKey: Trx3MhKey,
Src: SenderAddr.Hex(),
Dst: "",
Index: 2,
TxHash: MockTransactions[2].Hash().String(),
Data: MockContractByteCode,
Deployment: true,
},
}
MockRctMeta = []eth.ReceiptModel{
pkg/eth/models.go
+1 -1
View File
@@ -59,7 +59,7 @@ type TxModel struct {
MhKey string `db:"mh_key"`
Dst string `db:"dst"`
Src string `db:"src"`
Data []byte `db:"data"`
Data []byte `db:"tx_data"`
Deployment bool `db:"deployment"`
}
pkg/eth/publisher.go
+1 -1
View File
@@ -145,7 +145,7 @@ func (pub *IPLDPublisher) Publish(payload shared.ConvertedData) error {
return err
}
// If tx is a contract deployment, publish the data (code)
if txModel.Deployment {
if txModel.Deployment { // codec doesn't matter in this case sine we are not interested in the cid and the db key is multihash-derived
if _, err = shared.PublishRaw(tx, ipld.MEthStorageTrie, multihash.KECCAK_256, txModel.Data); err != nil {
return err
}