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

remove btc stuff

Browse Source
This commit is contained in:
Ian Norden
2020-08-31 10:42:01 -05:00
parent 3fd2af61e8
commit 558599dd32
87 changed files with 0 additions and 11269 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/eth/cleaner.go
-356
View File
@@ -1,356 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth
import (
"fmt"
"github.com/jmoiron/sqlx"
"github.com/sirupsen/logrus"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/postgres"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
// Cleaner satisfies the shared.Cleaner interface fo ethereum
type Cleaner struct {
db *postgres.DB
}
// NewCleaner returns a new Cleaner struct that satisfies the shared.Cleaner interface
func NewCleaner(db *postgres.DB) *Cleaner {
return &Cleaner{
db: db,
}
}
// ResetValidation resets the validation level to 0 to enable revalidation
func (c *Cleaner) ResetValidation(rngs [][2]uint64) error {
tx, err := c.db.Beginx()
if err != nil {
return err
}
for _, rng := range rngs {
logrus.Infof("eth db cleaner resetting validation level to 0 for block range %d to %d", rng[0], rng[1])
pgStr := `UPDATE eth.header_cids
SET times_validated = 0
WHERE block_number BETWEEN $1 AND $2`
if _, err := tx.Exec(pgStr, rng[0], rng[1]); err != nil {
shared.Rollback(tx)
return err
}
}
return tx.Commit()
}
// Clean removes the specified data from the db within the provided block range
func (c *Cleaner) Clean(rngs [][2]uint64, t shared.DataType) error {
tx, err := c.db.Beginx()
if err != nil {
return err
}
for _, rng := range rngs {
logrus.Infof("eth db cleaner cleaning up block range %d to %d", rng[0], rng[1])
if err := c.clean(tx, rng, t); err != nil {
shared.Rollback(tx)
return err
}
}
if err := tx.Commit(); err != nil {
return err
}
logrus.Infof("eth db cleaner vacuum analyzing cleaned tables to free up space from deleted rows")
return c.vacuumAnalyze(t)
}
func (c *Cleaner) clean(tx *sqlx.Tx, rng [2]uint64, t shared.DataType) error {
switch t {
case shared.Full, shared.Headers:
return c.cleanFull(tx, rng)
case shared.Uncles:
if err := c.cleanUncleIPLDs(tx, rng); err != nil {
return err
}
return c.cleanUncleMetaData(tx, rng)
case shared.Transactions:
if err := c.cleanReceiptIPLDs(tx, rng); err != nil {
return err
}
if err := c.cleanTransactionIPLDs(tx, rng); err != nil {
return err
}
return c.cleanTransactionMetaData(tx, rng)
case shared.Receipts:
if err := c.cleanReceiptIPLDs(tx, rng); err != nil {
return err
}
return c.cleanReceiptMetaData(tx, rng)
case shared.State:
if err := c.cleanStorageIPLDs(tx, rng); err != nil {
return err
}
if err := c.cleanStateIPLDs(tx, rng); err != nil {
return err
}
return c.cleanStateMetaData(tx, rng)
case shared.Storage:
if err := c.cleanStorageIPLDs(tx, rng); err != nil {
return err
}
return c.cleanStorageMetaData(tx, rng)
default:
return fmt.Errorf("eth cleaner unrecognized type: %s", t.String())
}
}
func (c *Cleaner) vacuumAnalyze(t shared.DataType) error {
switch t {
case shared.Full, shared.Headers:
return c.vacuumFull()
case shared.Uncles:
if err := c.vacuumUncles(); err != nil {
return err
}
case shared.Transactions:
if err := c.vacuumTxs(); err != nil {
return err
}
if err := c.vacuumRcts(); err != nil {
return err
}
case shared.Receipts:
if err := c.vacuumRcts(); err != nil {
return err
}
case shared.State:
if err := c.vacuumState(); err != nil {
return err
}
if err := c.vacuumAccounts(); err != nil {
return err
}
if err := c.vacuumStorage(); err != nil {
return err
}
case shared.Storage:
if err := c.vacuumStorage(); err != nil {
return err
}
default:
return fmt.Errorf("eth cleaner unrecognized type: %s", t.String())
}
return c.vacuumIPLDs()
}
func (c *Cleaner) vacuumFull() error {
if err := c.vacuumHeaders(); err != nil {
return err
}
if err := c.vacuumUncles(); err != nil {
return err
}
if err := c.vacuumTxs(); err != nil {
return err
}
if err := c.vacuumRcts(); err != nil {
return err
}
if err := c.vacuumState(); err != nil {
return err
}
if err := c.vacuumAccounts(); err != nil {
return err
}
return c.vacuumStorage()
}
func (c *Cleaner) vacuumHeaders() error {
_, err := c.db.Exec(`VACUUM ANALYZE eth.header_cids`)
return err
}
func (c *Cleaner) vacuumUncles() error {
_, err := c.db.Exec(`VACUUM ANALYZE eth.uncle_cids`)
return err
}
func (c *Cleaner) vacuumTxs() error {
_, err := c.db.Exec(`VACUUM ANALYZE eth.transaction_cids`)
return err
}
func (c *Cleaner) vacuumRcts() error {
_, err := c.db.Exec(`VACUUM ANALYZE eth.receipt_cids`)
return err
}
func (c *Cleaner) vacuumState() error {
_, err := c.db.Exec(`VACUUM ANALYZE eth.state_cids`)
return err
}
func (c *Cleaner) vacuumAccounts() error {
_, err := c.db.Exec(`VACUUM ANALYZE eth.state_accounts`)
return err
}
func (c *Cleaner) vacuumStorage() error {
_, err := c.db.Exec(`VACUUM ANALYZE eth.storage_cids`)
return err
}
func (c *Cleaner) vacuumIPLDs() error {
_, err := c.db.Exec(`VACUUM ANALYZE public.blocks`)
return err
}
func (c *Cleaner) cleanFull(tx *sqlx.Tx, rng [2]uint64) error {
if err := c.cleanStorageIPLDs(tx, rng); err != nil {
return err
}
if err := c.cleanStateIPLDs(tx, rng); err != nil {
return err
}
if err := c.cleanReceiptIPLDs(tx, rng); err != nil {
return err
}
if err := c.cleanTransactionIPLDs(tx, rng); err != nil {
return err
}
if err := c.cleanUncleIPLDs(tx, rng); err != nil {
return err
}
if err := c.cleanHeaderIPLDs(tx, rng); err != nil {
return err
}
return c.cleanHeaderMetaData(tx, rng)
}
func (c *Cleaner) cleanStorageIPLDs(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM public.blocks A
USING eth.storage_cids B, eth.state_cids C, eth.header_cids D
WHERE A.key = B.mh_key
AND B.state_id = C.id
AND C.header_id = D.id
AND D.block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
func (c *Cleaner) cleanStorageMetaData(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM eth.storage_cids A
USING eth.state_cids B, eth.header_cids C
WHERE A.state_id = B.id
AND B.header_id = C.id
AND C.block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
func (c *Cleaner) cleanStateIPLDs(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM public.blocks A
USING eth.state_cids B, eth.header_cids C
WHERE A.key = B.mh_key
AND B.header_id = C.id
AND C.block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
func (c *Cleaner) cleanStateMetaData(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM eth.state_cids A
USING eth.header_cids B
WHERE A.header_id = B.id
AND B.block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
func (c *Cleaner) cleanReceiptIPLDs(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM public.blocks A
USING eth.receipt_cids B, eth.transaction_cids C, eth.header_cids D
WHERE A.key = B.mh_key
AND B.tx_id = C.id
AND C.header_id = D.id
AND D.block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
func (c *Cleaner) cleanReceiptMetaData(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM eth.receipt_cids A
USING eth.transaction_cids B, eth.header_cids C
WHERE A.tx_id = B.id
AND B.header_id = C.id
AND C.block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
func (c *Cleaner) cleanTransactionIPLDs(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM public.blocks A
USING eth.transaction_cids B, eth.header_cids C
WHERE A.key = B.mh_key
AND B.header_id = C.id
AND C.block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
func (c *Cleaner) cleanTransactionMetaData(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM eth.transaction_cids A
USING eth.header_cids B
WHERE A.header_id = B.id
AND B.block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
func (c *Cleaner) cleanUncleIPLDs(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM public.blocks A
USING eth.uncle_cids B, eth.header_cids C
WHERE A.key = B.mh_key
AND B.header_id = C.id
AND C.block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
func (c *Cleaner) cleanUncleMetaData(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM eth.uncle_cids A
USING eth.header_cids B
WHERE A.header_id = B.id
AND B.block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
func (c *Cleaner) cleanHeaderIPLDs(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM public.blocks A
USING eth.header_cids B
WHERE A.key = B.mh_key
AND B.block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
func (c *Cleaner) cleanHeaderMetaData(tx *sqlx.Tx, rng [2]uint64) error {
pgStr := `DELETE FROM eth.header_cids
WHERE block_number BETWEEN $1 AND $2`
_, err := tx.Exec(pgStr, rng[0], rng[1])
return err
}
pkg/eth/cleaner_test.go
-698
View File
@@ -1,698 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth_test
import (
"math/big"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/postgres"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
var (
// Block 0
// header variables
blockHash1 = crypto.Keccak256Hash([]byte{00, 02})
blocKNumber1 = big.NewInt(0)
headerCID1 = shared.TestCID([]byte("mockHeader1CID"))
headerMhKey1 = shared.MultihashKeyFromCID(headerCID1)
parentHash = crypto.Keccak256Hash([]byte{00, 01})
totalDifficulty = "50000000000000000000"
reward = "5000000000000000000"
headerModel = eth.HeaderModel{
BlockHash: blockHash1.String(),
BlockNumber: blocKNumber1.String(),
CID: headerCID1.String(),
MhKey: headerMhKey1,
ParentHash: parentHash.String(),
TotalDifficulty: totalDifficulty,
Reward: reward,
}
// tx variables
tx1CID = shared.TestCID([]byte("mockTx1CID"))
tx1MhKey = shared.MultihashKeyFromCID(tx1CID)
tx2CID = shared.TestCID([]byte("mockTx2CID"))
tx2MhKey = shared.MultihashKeyFromCID(tx2CID)
tx1Hash = crypto.Keccak256Hash([]byte{01, 01})
tx2Hash = crypto.Keccak256Hash([]byte{01, 02})
txSrc = common.HexToAddress("0x010a")
txDst = common.HexToAddress("0x020a")
txModels1 = []eth.TxModel{
{
CID: tx1CID.String(),
MhKey: tx1MhKey,
TxHash: tx1Hash.String(),
Index: 0,
},
{
CID: tx2CID.String(),
MhKey: tx2MhKey,
TxHash: tx2Hash.String(),
Index: 1,
},
}
// uncle variables
uncleCID = shared.TestCID([]byte("mockUncle1CID"))
uncleMhKey = shared.MultihashKeyFromCID(uncleCID)
uncleHash = crypto.Keccak256Hash([]byte{02, 02})
uncleParentHash = crypto.Keccak256Hash([]byte{02, 01})
uncleReward = "1000000000000000000"
uncleModels1 = []eth.UncleModel{
{
CID: uncleCID.String(),
MhKey: uncleMhKey,
Reward: uncleReward,
BlockHash: uncleHash.String(),
ParentHash: uncleParentHash.String(),
},
}
// receipt variables
rct1CID = shared.TestCID([]byte("mockRct1CID"))
rct1MhKey = shared.MultihashKeyFromCID(rct1CID)
rct2CID = shared.TestCID([]byte("mockRct2CID"))
rct2MhKey = shared.MultihashKeyFromCID(rct2CID)
rct1Contract = common.Address{}
rct2Contract = common.HexToAddress("0x010c")
receiptModels1 = map[common.Hash]eth.ReceiptModel{
tx1Hash: {
CID: rct1CID.String(),
MhKey: rct1MhKey,
ContractHash: crypto.Keccak256Hash(rct1Contract.Bytes()).String(),
},
tx2Hash: {
CID: rct2CID.String(),
MhKey: rct2MhKey,
ContractHash: crypto.Keccak256Hash(rct2Contract.Bytes()).String(),
},
}
// state variables
state1CID1 = shared.TestCID([]byte("mockState1CID1"))
state1MhKey1 = shared.MultihashKeyFromCID(state1CID1)
state1Path = []byte{'\x01'}
state1Key = crypto.Keccak256Hash(txSrc.Bytes())
state2CID1 = shared.TestCID([]byte("mockState2CID1"))
state2MhKey1 = shared.MultihashKeyFromCID(state2CID1)
state2Path = []byte{'\x02'}
state2Key = crypto.Keccak256Hash(txDst.Bytes())
stateModels1 = []eth.StateNodeModel{
{
CID: state1CID1.String(),
MhKey: state1MhKey1,
Path: state1Path,
NodeType: 2,
StateKey: state1Key.String(),
},
{
CID: state2CID1.String(),
MhKey: state2MhKey1,
Path: state2Path,
NodeType: 2,
StateKey: state2Key.String(),
},
}
// storage variables
storageCID = shared.TestCID([]byte("mockStorageCID1"))
storageMhKey = shared.MultihashKeyFromCID(storageCID)
storagePath = []byte{'\x01'}
storageKey = crypto.Keccak256Hash(common.Hex2Bytes("0x0000000000000000000000000000000000000000000000000000000000000000"))
storageModels1 = map[string][]eth.StorageNodeModel{
common.Bytes2Hex(state1Path): {
{
CID: storageCID.String(),
MhKey: storageMhKey,
StorageKey: storageKey.String(),
Path: storagePath,
NodeType: 2,
},
},
}
mockCIDPayload1 = &eth.CIDPayload{
HeaderCID: headerModel,
UncleCIDs: uncleModels1,
TransactionCIDs: txModels1,
ReceiptCIDs: receiptModels1,
StateNodeCIDs: stateModels1,
StorageNodeCIDs: storageModels1,
}
// Block 1
// header variables
blockHash2 = crypto.Keccak256Hash([]byte{00, 03})
blocKNumber2 = big.NewInt(1)
headerCID2 = shared.TestCID([]byte("mockHeaderCID2"))
headerMhKey2 = shared.MultihashKeyFromCID(headerCID2)
headerModel2 = eth.HeaderModel{
BlockHash: blockHash2.String(),
BlockNumber: blocKNumber2.String(),
CID: headerCID2.String(),
MhKey: headerMhKey2,
ParentHash: blockHash1.String(),
TotalDifficulty: totalDifficulty,
Reward: reward,
}
// tx variables
tx3CID = shared.TestCID([]byte("mockTx3CID"))
tx3MhKey = shared.MultihashKeyFromCID(tx3CID)
tx3Hash = crypto.Keccak256Hash([]byte{01, 03})
txModels2 = []eth.TxModel{
{
CID: tx3CID.String(),
MhKey: tx3MhKey,
TxHash: tx3Hash.String(),
Index: 0,
},
}
// receipt variables
rct3CID = shared.TestCID([]byte("mockRct3CID"))
rct3MhKey = shared.MultihashKeyFromCID(rct3CID)
receiptModels2 = map[common.Hash]eth.ReceiptModel{
tx3Hash: {
CID: rct3CID.String(),
MhKey: rct3MhKey,
ContractHash: crypto.Keccak256Hash(rct1Contract.Bytes()).String(),
},
}
// state variables
state1CID2 = shared.TestCID([]byte("mockState1CID2"))
state1MhKey2 = shared.MultihashKeyFromCID(state1CID2)
stateModels2 = []eth.StateNodeModel{
{
CID: state1CID2.String(),
MhKey: state1MhKey2,
Path: state1Path,
NodeType: 2,
StateKey: state1Key.String(),
},
}
mockCIDPayload2 = &eth.CIDPayload{
HeaderCID: headerModel2,
TransactionCIDs: txModels2,
ReceiptCIDs: receiptModels2,
StateNodeCIDs: stateModels2,
}
rngs = [][2]uint64{{0, 1}}
mhKeys = []string{
headerMhKey1,
headerMhKey2,
uncleMhKey,
tx1MhKey,
tx2MhKey,
tx3MhKey,
rct1MhKey,
rct2MhKey,
rct3MhKey,
state1MhKey1,
state2MhKey1,
state1MhKey2,
storageMhKey,
}
mockData = []byte{'\x01'}
)
var _ = Describe("Cleaner", func() {
var (
db *postgres.DB
repo *eth.CIDIndexer
cleaner *eth.Cleaner
)
BeforeEach(func() {
var err error
db, err = shared.SetupDB()
Expect(err).ToNot(HaveOccurred())
repo = eth.NewCIDIndexer(db)
cleaner = eth.NewCleaner(db)
})
Describe("Clean", func() {
BeforeEach(func() {
for _, key := range mhKeys {
_, err := db.Exec(`INSERT INTO public.blocks (key, data) VALUES ($1, $2)`, key, mockData)
Expect(err).ToNot(HaveOccurred())
}
err := repo.Index(mockCIDPayload1)
Expect(err).ToNot(HaveOccurred())
err = repo.Index(mockCIDPayload2)
Expect(err).ToNot(HaveOccurred())
tx, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
var startingIPFSBlocksCount int
pgStr := `SELECT COUNT(*) FROM public.blocks`
err = tx.Get(&startingIPFSBlocksCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var startingStorageCount int
pgStr = `SELECT COUNT(*) FROM eth.storage_cids`
err = tx.Get(&startingStorageCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var startingStateCount int
pgStr = `SELECT COUNT(*) FROM eth.state_cids`
err = tx.Get(&startingStateCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var startingReceiptCount int
pgStr = `SELECT COUNT(*) FROM eth.receipt_cids`
err = tx.Get(&startingReceiptCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var startingTxCount int
pgStr = `SELECT COUNT(*) FROM eth.transaction_cids`
err = tx.Get(&startingTxCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var startingUncleCount int
pgStr = `SELECT COUNT(*) FROM eth.uncle_cids`
err = tx.Get(&startingUncleCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var startingHeaderCount int
pgStr = `SELECT COUNT(*) FROM eth.header_cids`
err = tx.Get(&startingHeaderCount, pgStr)
Expect(err).ToNot(HaveOccurred())
err = tx.Commit()
Expect(err).ToNot(HaveOccurred())
Expect(startingIPFSBlocksCount).To(Equal(13))
Expect(startingStorageCount).To(Equal(1))
Expect(startingStateCount).To(Equal(3))
Expect(startingReceiptCount).To(Equal(3))
Expect(startingTxCount).To(Equal(3))
Expect(startingUncleCount).To(Equal(1))
Expect(startingHeaderCount).To(Equal(2))
})
AfterEach(func() {
eth.TearDownDB(db)
})
It("Cleans everything", func() {
err := cleaner.Clean(rngs, shared.Full)
Expect(err).ToNot(HaveOccurred())
tx, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
pgStr := `SELECT COUNT(*) FROM eth.header_cids`
var headerCount int
err = tx.Get(&headerCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var uncleCount int
pgStr = `SELECT COUNT(*) FROM eth.uncle_cids`
err = tx.Get(&uncleCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var txCount int
pgStr = `SELECT COUNT(*) FROM eth.transaction_cids`
err = tx.Get(&txCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var rctCount int
pgStr = `SELECT COUNT(*) FROM eth.receipt_cids`
err = tx.Get(&rctCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var stateCount int
pgStr = `SELECT COUNT(*) FROM eth.state_cids`
err = tx.Get(&stateCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var storageCount int
pgStr = `SELECT COUNT(*) FROM eth.storage_cids`
err = tx.Get(&storageCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var blocksCount int
pgStr = `SELECT COUNT(*) FROM public.blocks`
err = tx.Get(&blocksCount, pgStr)
Expect(err).ToNot(HaveOccurred())
err = tx.Commit()
Expect(err).ToNot(HaveOccurred())
Expect(headerCount).To(Equal(0))
Expect(uncleCount).To(Equal(0))
Expect(txCount).To(Equal(0))
Expect(rctCount).To(Equal(0))
Expect(stateCount).To(Equal(0))
Expect(storageCount).To(Equal(0))
Expect(blocksCount).To(Equal(0))
})
It("Cleans headers and all linked data (same as full)", func() {
err := cleaner.Clean(rngs, shared.Headers)
Expect(err).ToNot(HaveOccurred())
tx, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
var headerCount int
pgStr := `SELECT COUNT(*) FROM eth.header_cids`
err = tx.Get(&headerCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var uncleCount int
pgStr = `SELECT COUNT(*) FROM eth.uncle_cids`
err = tx.Get(&uncleCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var txCount int
pgStr = `SELECT COUNT(*) FROM eth.transaction_cids`
err = tx.Get(&txCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var rctCount int
pgStr = `SELECT COUNT(*) FROM eth.receipt_cids`
err = tx.Get(&rctCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var stateCount int
pgStr = `SELECT COUNT(*) FROM eth.state_cids`
err = tx.Get(&stateCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var storageCount int
pgStr = `SELECT COUNT(*) FROM eth.storage_cids`
err = tx.Get(&storageCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var blocksCount int
pgStr = `SELECT COUNT(*) FROM public.blocks`
err = tx.Get(&blocksCount, pgStr)
Expect(err).ToNot(HaveOccurred())
err = tx.Commit()
Expect(err).ToNot(HaveOccurred())
Expect(headerCount).To(Equal(0))
Expect(uncleCount).To(Equal(0))
Expect(txCount).To(Equal(0))
Expect(rctCount).To(Equal(0))
Expect(stateCount).To(Equal(0))
Expect(storageCount).To(Equal(0))
Expect(blocksCount).To(Equal(0))
})
It("Cleans uncles", func() {
err := cleaner.Clean(rngs, shared.Uncles)
Expect(err).ToNot(HaveOccurred())
tx, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
var headerCount int
pgStr := `SELECT COUNT(*) FROM eth.header_cids`
err = tx.Get(&headerCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var uncleCount int
pgStr = `SELECT COUNT(*) FROM eth.uncle_cids`
err = tx.Get(&uncleCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var txCount int
pgStr = `SELECT COUNT(*) FROM eth.transaction_cids`
err = tx.Get(&txCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var rctCount int
pgStr = `SELECT COUNT(*) FROM eth.receipt_cids`
err = tx.Get(&rctCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var stateCount int
pgStr = `SELECT COUNT(*) FROM eth.state_cids`
err = tx.Get(&stateCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var storageCount int
pgStr = `SELECT COUNT(*) FROM eth.storage_cids`
err = tx.Get(&storageCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var blocksCount int
pgStr = `SELECT COUNT(*) FROM public.blocks`
err = tx.Get(&blocksCount, pgStr)
Expect(err).ToNot(HaveOccurred())
err = tx.Commit()
Expect(err).ToNot(HaveOccurred())
Expect(headerCount).To(Equal(2))
Expect(uncleCount).To(Equal(0))
Expect(txCount).To(Equal(3))
Expect(rctCount).To(Equal(3))
Expect(stateCount).To(Equal(3))
Expect(storageCount).To(Equal(1))
Expect(blocksCount).To(Equal(12))
})
It("Cleans transactions and linked receipts", func() {
err := cleaner.Clean(rngs, shared.Transactions)
Expect(err).ToNot(HaveOccurred())
tx, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
var headerCount int
pgStr := `SELECT COUNT(*) FROM eth.header_cids`
err = tx.Get(&headerCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var uncleCount int
pgStr = `SELECT COUNT(*) FROM eth.uncle_cids`
err = tx.Get(&uncleCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var txCount int
pgStr = `SELECT COUNT(*) FROM eth.transaction_cids`
err = tx.Get(&txCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var rctCount int
pgStr = `SELECT COUNT(*) FROM eth.receipt_cids`
err = tx.Get(&rctCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var stateCount int
pgStr = `SELECT COUNT(*) FROM eth.state_cids`
err = tx.Get(&stateCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var storageCount int
pgStr = `SELECT COUNT(*) FROM eth.storage_cids`
err = tx.Get(&storageCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var blocksCount int
pgStr = `SELECT COUNT(*) FROM public.blocks`
err = tx.Get(&blocksCount, pgStr)
Expect(err).ToNot(HaveOccurred())
err = tx.Commit()
Expect(err).ToNot(HaveOccurred())
Expect(headerCount).To(Equal(2))
Expect(uncleCount).To(Equal(1))
Expect(txCount).To(Equal(0))
Expect(rctCount).To(Equal(0))
Expect(stateCount).To(Equal(3))
Expect(storageCount).To(Equal(1))
Expect(blocksCount).To(Equal(7))
})
It("Cleans receipts", func() {
err := cleaner.Clean(rngs, shared.Receipts)
Expect(err).ToNot(HaveOccurred())
tx, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
var headerCount int
pgStr := `SELECT COUNT(*) FROM eth.header_cids`
err = tx.Get(&headerCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var uncleCount int
pgStr = `SELECT COUNT(*) FROM eth.uncle_cids`
err = tx.Get(&uncleCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var txCount int
pgStr = `SELECT COUNT(*) FROM eth.transaction_cids`
err = tx.Get(&txCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var rctCount int
pgStr = `SELECT COUNT(*) FROM eth.receipt_cids`
err = tx.Get(&rctCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var stateCount int
pgStr = `SELECT COUNT(*) FROM eth.state_cids`
err = tx.Get(&stateCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var storageCount int
pgStr = `SELECT COUNT(*) FROM eth.storage_cids`
err = tx.Get(&storageCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var blocksCount int
pgStr = `SELECT COUNT(*) FROM public.blocks`
err = tx.Get(&blocksCount, pgStr)
Expect(err).ToNot(HaveOccurred())
err = tx.Commit()
Expect(err).ToNot(HaveOccurred())
Expect(headerCount).To(Equal(2))
Expect(uncleCount).To(Equal(1))
Expect(txCount).To(Equal(3))
Expect(rctCount).To(Equal(0))
Expect(stateCount).To(Equal(3))
Expect(storageCount).To(Equal(1))
Expect(blocksCount).To(Equal(10))
})
It("Cleans state and linked storage", func() {
err := cleaner.Clean(rngs, shared.State)
Expect(err).ToNot(HaveOccurred())
tx, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
var headerCount int
pgStr := `SELECT COUNT(*) FROM eth.header_cids`
err = tx.Get(&headerCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var uncleCount int
pgStr = `SELECT COUNT(*) FROM eth.uncle_cids`
err = tx.Get(&uncleCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var txCount int
pgStr = `SELECT COUNT(*) FROM eth.transaction_cids`
err = tx.Get(&txCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var rctCount int
pgStr = `SELECT COUNT(*) FROM eth.receipt_cids`
err = tx.Get(&rctCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var stateCount int
pgStr = `SELECT COUNT(*) FROM eth.state_cids`
err = tx.Get(&stateCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var storageCount int
pgStr = `SELECT COUNT(*) FROM eth.storage_cids`
err = tx.Get(&storageCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var blocksCount int
pgStr = `SELECT COUNT(*) FROM public.blocks`
err = tx.Get(&blocksCount, pgStr)
Expect(err).ToNot(HaveOccurred())
err = tx.Commit()
Expect(err).ToNot(HaveOccurred())
Expect(headerCount).To(Equal(2))
Expect(uncleCount).To(Equal(1))
Expect(txCount).To(Equal(3))
Expect(rctCount).To(Equal(3))
Expect(stateCount).To(Equal(0))
Expect(storageCount).To(Equal(0))
Expect(blocksCount).To(Equal(9))
})
It("Cleans storage", func() {
err := cleaner.Clean(rngs, shared.Storage)
Expect(err).ToNot(HaveOccurred())
tx, err := db.Beginx()
Expect(err).ToNot(HaveOccurred())
var headerCount int
pgStr := `SELECT COUNT(*) FROM eth.header_cids`
err = tx.Get(&headerCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var uncleCount int
pgStr = `SELECT COUNT(*) FROM eth.uncle_cids`
err = tx.Get(&uncleCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var txCount int
pgStr = `SELECT COUNT(*) FROM eth.transaction_cids`
err = tx.Get(&txCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var rctCount int
pgStr = `SELECT COUNT(*) FROM eth.receipt_cids`
err = tx.Get(&rctCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var stateCount int
pgStr = `SELECT COUNT(*) FROM eth.state_cids`
err = tx.Get(&stateCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var storageCount int
pgStr = `SELECT COUNT(*) FROM eth.storage_cids`
err = tx.Get(&storageCount, pgStr)
Expect(err).ToNot(HaveOccurred())
var blocksCount int
pgStr = `SELECT COUNT(*) FROM public.blocks`
err = tx.Get(&blocksCount, pgStr)
Expect(err).ToNot(HaveOccurred())
err = tx.Commit()
Expect(err).ToNot(HaveOccurred())
Expect(headerCount).To(Equal(2))
Expect(uncleCount).To(Equal(1))
Expect(txCount).To(Equal(3))
Expect(rctCount).To(Equal(3))
Expect(stateCount).To(Equal(3))
Expect(storageCount).To(Equal(0))
Expect(blocksCount).To(Equal(12))
})
})
Describe("ResetValidation", func() {
BeforeEach(func() {
for _, key := range mhKeys {
_, err := db.Exec(`INSERT INTO public.blocks (key, data) VALUES ($1, $2)`, key, mockData)
Expect(err).ToNot(HaveOccurred())
}
err := repo.Index(mockCIDPayload1)
Expect(err).ToNot(HaveOccurred())
err = repo.Index(mockCIDPayload2)
Expect(err).ToNot(HaveOccurred())
var validationTimes []int
pgStr := `SELECT times_validated FROM eth.header_cids`
err = db.Select(&validationTimes, pgStr)
Expect(err).ToNot(HaveOccurred())
Expect(len(validationTimes)).To(Equal(2))
Expect(validationTimes[0]).To(Equal(1))
Expect(validationTimes[1]).To(Equal(1))
err = repo.Index(mockCIDPayload1)
Expect(err).ToNot(HaveOccurred())
validationTimes = []int{}
pgStr = `SELECT times_validated FROM eth.header_cids ORDER BY block_number`
err = db.Select(&validationTimes, pgStr)
Expect(err).ToNot(HaveOccurred())
Expect(len(validationTimes)).To(Equal(2))
Expect(validationTimes[0]).To(Equal(2))
Expect(validationTimes[1]).To(Equal(1))
})
AfterEach(func() {
eth.TearDownDB(db)
})
It("Resets the validation level", func() {
err := cleaner.ResetValidation(rngs)
Expect(err).ToNot(HaveOccurred())
var validationTimes []int
pgStr := `SELECT times_validated FROM eth.header_cids`
err = db.Select(&validationTimes, pgStr)
Expect(err).ToNot(HaveOccurred())
Expect(len(validationTimes)).To(Equal(2))
Expect(validationTimes[0]).To(Equal(0))
Expect(validationTimes[1]).To(Equal(0))
err = repo.Index(mockCIDPayload2)
Expect(err).ToNot(HaveOccurred())
validationTimes = []int{}
pgStr = `SELECT times_validated FROM eth.header_cids ORDER BY block_number`
err = db.Select(&validationTimes, pgStr)
Expect(err).ToNot(HaveOccurred())
Expect(len(validationTimes)).To(Equal(2))
Expect(validationTimes[0]).To(Equal(0))
Expect(validationTimes[1]).To(Equal(1))
})
})
})
pkg/eth/converter.go
-155
View File
@@ -1,155 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth
import (
"fmt"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/statediff"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
// PayloadConverter satisfies the PayloadConverter interface for ethereum
type PayloadConverter struct {
chainConfig *params.ChainConfig
}
// NewPayloadConverter creates a pointer to a new PayloadConverter which satisfies the PayloadConverter interface
func NewPayloadConverter(chainConfig *params.ChainConfig) *PayloadConverter {
return &PayloadConverter{
chainConfig: chainConfig,
}
}
// Convert method is used to convert a eth statediff.Payload to an IPLDPayload
// Satisfies the shared.PayloadConverter interface
func (pc *PayloadConverter) Convert(payload shared.RawChainData) (shared.ConvertedData, error) {
stateDiffPayload, ok := payload.(statediff.Payload)
if !ok {
return nil, fmt.Errorf("eth converter: expected payload type %T got %T", statediff.Payload{}, payload)
}
// Unpack block rlp to access fields
block := new(types.Block)
if err := rlp.DecodeBytes(stateDiffPayload.BlockRlp, block); err != nil {
return nil, err
}
trxLen := len(block.Transactions())
convertedPayload := ConvertedPayload{
TotalDifficulty: stateDiffPayload.TotalDifficulty,
Block: block,
TxMetaData: make([]TxModel, 0, trxLen),
Receipts: make(types.Receipts, 0, trxLen),
ReceiptMetaData: make([]ReceiptModel, 0, trxLen),
StateNodes: make([]TrieNode, 0),
StorageNodes: make(map[string][]TrieNode),
}
signer := types.MakeSigner(pc.chainConfig, block.Number())
transactions := block.Transactions()
for i, trx := range transactions {
// Extract to and from data from the the transactions for indexing
from, err := types.Sender(signer, trx)
if err != nil {
return nil, err
}
txMeta := TxModel{
Dst: shared.HandleZeroAddrPointer(trx.To()),
Src: shared.HandleZeroAddr(from),
TxHash: trx.Hash().String(),
Index: int64(i),
Data: trx.Data(),
}
// txMeta will have same index as its corresponding trx in the convertedPayload.BlockBody
convertedPayload.TxMetaData = append(convertedPayload.TxMetaData, txMeta)
}
// Decode receipts for this block
receipts := make(types.Receipts, 0)
if err := rlp.DecodeBytes(stateDiffPayload.ReceiptsRlp, &receipts); err != nil {
return nil, err
}
// Derive any missing fields
if err := receipts.DeriveFields(pc.chainConfig, block.Hash(), block.NumberU64(), block.Transactions()); err != nil {
return nil, err
}
for i, receipt := range receipts {
// Extract topic and contract data from the receipt for indexing
topicSets := make([][]string, 4)
mappedContracts := make(map[string]bool) // use map to avoid duplicate addresses
for _, log := range receipt.Logs {
for i, topic := range log.Topics {
topicSets[i] = append(topicSets[i], topic.Hex())
}
mappedContracts[log.Address.String()] = true
}
// These are the contracts seen in the logs
logContracts := make([]string, 0, len(mappedContracts))
for addr := range mappedContracts {
logContracts = append(logContracts, addr)
}
// This is the contract address if this receipt is for a contract creation tx
contract := shared.HandleZeroAddr(receipt.ContractAddress)
var contractHash string
if contract != "" {
convertedPayload.TxMetaData[i].Deployment = true
contractHash = crypto.Keccak256Hash(common.HexToAddress(contract).Bytes()).String()
}
rctMeta := ReceiptModel{
Topic0s: topicSets[0],
Topic1s: topicSets[1],
Topic2s: topicSets[2],
Topic3s: topicSets[3],
Contract: contract,
ContractHash: contractHash,
LogContracts: logContracts,
}
// receipt and rctMeta will have same indexes
convertedPayload.Receipts = append(convertedPayload.Receipts, receipt)
convertedPayload.ReceiptMetaData = append(convertedPayload.ReceiptMetaData, rctMeta)
}
// Unpack state diff rlp to access fields
stateDiff := new(statediff.StateObject)
if err := rlp.DecodeBytes(stateDiffPayload.StateObjectRlp, stateDiff); err != nil {
return nil, err
}
for _, stateNode := range stateDiff.Nodes {
statePath := common.Bytes2Hex(stateNode.Path)
convertedPayload.StateNodes = append(convertedPayload.StateNodes, TrieNode{
Path: stateNode.Path,
Value: stateNode.NodeValue,
Type: stateNode.NodeType,
LeafKey: common.BytesToHash(stateNode.LeafKey),
})
for _, storageNode := range stateNode.StorageNodes {
convertedPayload.StorageNodes[statePath] = append(convertedPayload.StorageNodes[statePath], TrieNode{
Path: storageNode.Path,
Value: storageNode.NodeValue,
Type: storageNode.NodeType,
LeafKey: common.BytesToHash(storageNode.LeafKey),
})
}
}
return convertedPayload, nil
}
pkg/eth/converter_test.go
-54
View File
@@ -1,54 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth_test
import (
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rlp"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth/mocks"
)
var _ = Describe("Converter", func() {
Describe("Convert", func() {
It("Converts mock statediff.Payloads into the expected IPLDPayloads", func() {
converter := eth.NewPayloadConverter(params.MainnetChainConfig)
payload, err := converter.Convert(mocks.MockStateDiffPayload)
Expect(err).ToNot(HaveOccurred())
convertedPayload, ok := payload.(eth.ConvertedPayload)
Expect(ok).To(BeTrue())
Expect(convertedPayload.Block.Number().String()).To(Equal(mocks.BlockNumber.String()))
Expect(convertedPayload.Block.Hash().String()).To(Equal(mocks.MockBlock.Hash().String()))
Expect(convertedPayload.StateNodes).To(Equal(mocks.MockStateNodes))
Expect(convertedPayload.StorageNodes).To(Equal(mocks.MockStorageNodes))
Expect(convertedPayload.TotalDifficulty.Int64()).To(Equal(mocks.MockStateDiffPayload.TotalDifficulty.Int64()))
gotBody, err := rlp.EncodeToBytes(convertedPayload.Block.Body())
Expect(err).ToNot(HaveOccurred())
expectedBody, err := rlp.EncodeToBytes(mocks.MockBlock.Body())
Expect(err).ToNot(HaveOccurred())
Expect(gotBody).To(Equal(expectedBody))
gotHeader, err := rlp.EncodeToBytes(convertedPayload.Block.Header())
Expect(err).ToNot(HaveOccurred())
Expect(gotHeader).To(Equal(mocks.MockHeaderRlp))
Expect(convertedPayload.TxMetaData).To(Equal(mocks.MockTrxMeta))
Expect(convertedPayload.ReceiptMetaData).To(Equal(mocks.MockRctMeta))
})
})
})
pkg/eth/indexer.go
-206
View File
@@ -1,206 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth
import (
"fmt"
"github.com/ethereum/go-ethereum/common"
"github.com/jmoiron/sqlx"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/postgres"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
var (
nullHash = common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000000")
)
// Indexer satisfies the Indexer interface for ethereum
type CIDIndexer struct {
db *postgres.DB
}
// NewCIDIndexer creates a new pointer to a Indexer which satisfies the CIDIndexer interface
func NewCIDIndexer(db *postgres.DB) *CIDIndexer {
return &CIDIndexer{
db: db,
}
}
// Index indexes a cidPayload in Postgres
func (in *CIDIndexer) Index(cids shared.CIDsForIndexing) error {
cidPayload, ok := cids.(*CIDPayload)
if !ok {
return fmt.Errorf("eth indexer expected cids type %T got %T", &CIDPayload{}, cids)
}
// Begin new db tx
tx, err := in.db.Beginx()
if err != nil {
return err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
headerID, err := in.indexHeaderCID(tx, cidPayload.HeaderCID)
if err != nil {
log.Error("eth indexer error when indexing header")
return err
}
for _, uncle := range cidPayload.UncleCIDs {
if err := in.indexUncleCID(tx, uncle, headerID); err != nil {
log.Error("eth indexer error when indexing uncle")
return err
}
}
if err := in.indexTransactionAndReceiptCIDs(tx, cidPayload, headerID); err != nil {
log.Error("eth indexer error when indexing transactions and receipts")
return err
}
err = in.indexStateAndStorageCIDs(tx, cidPayload, headerID)
if err != nil {
log.Error("eth indexer error when indexing state and storage nodes")
}
return err
}
func (in *CIDIndexer) indexHeaderCID(tx *sqlx.Tx, header HeaderModel) (int64, error) {
var headerID int64
err := tx.QueryRowx(`INSERT INTO eth.header_cids (block_number, block_hash, parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
ON CONFLICT (block_number, block_hash) DO UPDATE SET (parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated) = ($3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, eth.header_cids.times_validated + 1)
RETURNING id`,
header.BlockNumber, header.BlockHash, header.ParentHash, header.CID, header.TotalDifficulty, in.db.NodeID, header.Reward, header.StateRoot, header.TxRoot,
header.RctRoot, header.UncleRoot, header.Bloom, header.Timestamp, header.MhKey, 1).Scan(&headerID)
return headerID, err
}
func (in *CIDIndexer) indexUncleCID(tx *sqlx.Tx, uncle UncleModel, headerID int64) error {
_, err := tx.Exec(`INSERT INTO eth.uncle_cids (block_hash, header_id, parent_hash, cid, reward, mh_key) VALUES ($1, $2, $3, $4, $5, $6)
ON CONFLICT (header_id, block_hash) DO UPDATE SET (parent_hash, cid, reward, mh_key) = ($3, $4, $5, $6)`,
uncle.BlockHash, headerID, uncle.ParentHash, uncle.CID, uncle.Reward, uncle.MhKey)
return err
}
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, 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 {
return err
}
receiptCidMeta, ok := payload.ReceiptCIDs[common.HexToHash(trxCidMeta.TxHash)]
if ok {
if err := in.indexReceiptCID(tx, receiptCidMeta, txID); err != nil {
return err
}
}
}
return nil
}
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, 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
}
func (in *CIDIndexer) indexReceiptCID(tx *sqlx.Tx, cidMeta ReceiptModel, txID int64) error {
_, err := tx.Exec(`INSERT INTO eth.receipt_cids (tx_id, cid, contract, contract_hash, topic0s, topic1s, topic2s, topic3s, log_contracts, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
ON CONFLICT (tx_id) DO UPDATE SET (cid, contract, contract_hash, topic0s, topic1s, topic2s, topic3s, log_contracts, mh_key) = ($2, $3, $4, $5, $6, $7, $8, $9, $10)`,
txID, cidMeta.CID, cidMeta.Contract, cidMeta.ContractHash, cidMeta.Topic0s, cidMeta.Topic1s, cidMeta.Topic2s, cidMeta.Topic3s, cidMeta.LogContracts, cidMeta.MhKey)
return err
}
func (in *CIDIndexer) indexStateAndStorageCIDs(tx *sqlx.Tx, payload *CIDPayload, headerID int64) error {
for _, stateCID := range payload.StateNodeCIDs {
var stateID int64
var stateKey string
if stateCID.StateKey != nullHash.String() {
stateKey = stateCID.StateKey
}
err := tx.QueryRowx(`INSERT INTO eth.state_cids (header_id, state_leaf_key, cid, state_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (header_id, state_path) DO UPDATE SET (state_leaf_key, cid, node_type, diff, mh_key) = ($2, $3, $5, $6, $7)
RETURNING id`,
headerID, stateKey, stateCID.CID, stateCID.Path, stateCID.NodeType, true, stateCID.MhKey).Scan(&stateID)
if err != nil {
return err
}
// If we have a state leaf node, index the associated account and storage nodes
if stateCID.NodeType == 2 {
statePath := common.Bytes2Hex(stateCID.Path)
for _, storageCID := range payload.StorageNodeCIDs[statePath] {
if err := in.indexStorageCID(tx, storageCID, stateID); err != nil {
return err
}
}
if stateAccount, ok := payload.StateAccounts[statePath]; ok {
if err := in.indexStateAccount(tx, stateAccount, stateID); err != nil {
return err
}
}
}
}
return nil
}
func (in *CIDIndexer) indexStateCID(tx *sqlx.Tx, stateNode StateNodeModel, headerID int64) (int64, error) {
var stateID int64
var stateKey string
if stateNode.StateKey != nullHash.String() {
stateKey = stateNode.StateKey
}
err := tx.QueryRowx(`INSERT INTO eth.state_cids (header_id, state_leaf_key, cid, state_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (header_id, state_path) DO UPDATE SET (state_leaf_key, cid, node_type, diff, mh_key) = ($2, $3, $5, $6, $7)
RETURNING id`,
headerID, stateKey, stateNode.CID, stateNode.Path, stateNode.NodeType, true, stateNode.MhKey).Scan(&stateID)
return stateID, err
}
func (in *CIDIndexer) indexStateAccount(tx *sqlx.Tx, stateAccount StateAccountModel, stateID int64) error {
_, err := tx.Exec(`INSERT INTO eth.state_accounts (state_id, balance, nonce, code_hash, storage_root) VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (state_id) DO UPDATE SET (balance, nonce, code_hash, storage_root) = ($2, $3, $4, $5)`,
stateID, stateAccount.Balance, stateAccount.Nonce, stateAccount.CodeHash, stateAccount.StorageRoot)
return err
}
func (in *CIDIndexer) indexStorageCID(tx *sqlx.Tx, storageCID StorageNodeModel, stateID int64) error {
var storageKey string
if storageCID.StorageKey != nullHash.String() {
storageKey = storageCID.StorageKey
}
_, err := tx.Exec(`INSERT INTO eth.storage_cids (state_id, storage_leaf_key, cid, storage_path, node_type, diff, mh_key) VALUES ($1, $2, $3, $4, $5, $6, $7)
ON CONFLICT (state_id, storage_path) DO UPDATE SET (storage_leaf_key, cid, node_type, diff, mh_key) = ($2, $3, $5, $6, $7)`,
stateID, storageKey, storageCID.CID, storageCID.Path, storageCID.NodeType, true, storageCID.MhKey)
return err
}
pkg/eth/indexer_test.go
-137
View File
@@ -1,137 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth_test
import (
"github.com/ethereum/go-ethereum/common"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth/mocks"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/postgres"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
var _ = Describe("Indexer", func() {
var (
db *postgres.DB
err error
repo *eth.CIDIndexer
)
BeforeEach(func() {
db, err = shared.SetupDB()
Expect(err).ToNot(HaveOccurred())
repo = eth.NewCIDIndexer(db)
// need entries in the public.blocks with the mhkeys or the FK constraint will fail
shared.PublishMockIPLD(db, mocks.HeaderMhKey, mockData)
shared.PublishMockIPLD(db, mocks.Trx1MhKey, mockData)
shared.PublishMockIPLD(db, mocks.Trx2MhKey, mockData)
shared.PublishMockIPLD(db, mocks.Trx3MhKey, mockData)
shared.PublishMockIPLD(db, mocks.Rct1MhKey, mockData)
shared.PublishMockIPLD(db, mocks.Rct2MhKey, mockData)
shared.PublishMockIPLD(db, mocks.Rct3MhKey, mockData)
shared.PublishMockIPLD(db, mocks.State1MhKey, mockData)
shared.PublishMockIPLD(db, mocks.State2MhKey, mockData)
shared.PublishMockIPLD(db, mocks.StorageMhKey, mockData)
})
AfterEach(func() {
eth.TearDownDB(db)
})
Describe("Index", func() {
It("Indexes CIDs and related metadata into vulcanizedb", func() {
err = repo.Index(mocks.MockCIDPayload)
Expect(err).ToNot(HaveOccurred())
pgStr := `SELECT cid, td, reward, id
FROM eth.header_cids
WHERE block_number = $1`
// check header was properly indexed
type res struct {
CID string
TD string
Reward string
ID int
}
headers := new(res)
err = db.QueryRowx(pgStr, 1).StructScan(headers)
Expect(err).ToNot(HaveOccurred())
Expect(headers.CID).To(Equal(mocks.HeaderCID.String()))
Expect(headers.TD).To(Equal(mocks.MockBlock.Difficulty().String()))
Expect(headers.Reward).To(Equal("5000000000000000000"))
// check trxs were properly indexed
trxs := make([]string, 0)
pgStr = `SELECT transaction_cids.cid FROM eth.transaction_cids INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.id)
WHERE header_cids.block_number = $1`
err = db.Select(&trxs, pgStr, 1)
Expect(err).ToNot(HaveOccurred())
Expect(len(trxs)).To(Equal(3))
Expect(shared.ListContainsString(trxs, mocks.Trx1CID.String())).To(BeTrue())
Expect(shared.ListContainsString(trxs, mocks.Trx2CID.String())).To(BeTrue())
Expect(shared.ListContainsString(trxs, mocks.Trx3CID.String())).To(BeTrue())
// check receipts were properly indexed
rcts := make([]string, 0)
pgStr = `SELECT receipt_cids.cid FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids
WHERE receipt_cids.tx_id = transaction_cids.id
AND transaction_cids.header_id = header_cids.id
AND header_cids.block_number = $1`
err = db.Select(&rcts, pgStr, 1)
Expect(err).ToNot(HaveOccurred())
Expect(len(rcts)).To(Equal(3))
Expect(shared.ListContainsString(rcts, mocks.Rct1CID.String())).To(BeTrue())
Expect(shared.ListContainsString(rcts, mocks.Rct2CID.String())).To(BeTrue())
Expect(shared.ListContainsString(rcts, mocks.Rct3CID.String())).To(BeTrue())
// check that state nodes were properly indexed
stateNodes := make([]eth.StateNodeModel, 0)
pgStr = `SELECT state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.id)
WHERE header_cids.block_number = $1`
err = db.Select(&stateNodes, pgStr, 1)
Expect(err).ToNot(HaveOccurred())
Expect(len(stateNodes)).To(Equal(2))
for _, stateNode := range stateNodes {
if stateNode.CID == mocks.State1CID.String() {
Expect(stateNode.NodeType).To(Equal(2))
Expect(stateNode.StateKey).To(Equal(common.BytesToHash(mocks.ContractLeafKey).Hex()))
Expect(stateNode.Path).To(Equal([]byte{'\x06'}))
}
if stateNode.CID == mocks.State2CID.String() {
Expect(stateNode.NodeType).To(Equal(2))
Expect(stateNode.StateKey).To(Equal(common.BytesToHash(mocks.AccountLeafKey).Hex()))
Expect(stateNode.Path).To(Equal([]byte{'\x0c'}))
}
}
// check that storage nodes were properly indexed
storageNodes := make([]eth.StorageNodeWithStateKeyModel, 0)
pgStr = `SELECT storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
FROM eth.storage_cids, eth.state_cids, eth.header_cids
WHERE storage_cids.state_id = state_cids.id
AND state_cids.header_id = header_cids.id
AND header_cids.block_number = $1`
err = db.Select(&storageNodes, pgStr, 1)
Expect(err).ToNot(HaveOccurred())
Expect(len(storageNodes)).To(Equal(1))
Expect(storageNodes[0]).To(Equal(eth.StorageNodeWithStateKeyModel{
CID: mocks.StorageCID.String(),
NodeType: 2,
StorageKey: common.BytesToHash(mocks.StorageLeafKey).Hex(),
StateKey: common.BytesToHash(mocks.ContractLeafKey).Hex(),
Path: []byte{},
}))
})
})
})
pkg/eth/mocks/test_data.go
-582
View File
@@ -1,582 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package mocks
import (
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"math/big"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/statediff"
"github.com/ethereum/go-ethereum/statediff/testhelpers"
"github.com/ipfs/go-block-format"
"github.com/multiformats/go-multihash"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/ipfs"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/ipfs/ipld"
)
// Test variables
var (
// block data
BlockNumber = big.NewInt(1)
MockHeader = types.Header{
Time: 0,
Number: new(big.Int).Set(BlockNumber),
Root: common.HexToHash("0x0"),
TxHash: common.HexToHash("0x0"),
ReceiptHash: common.HexToHash("0x0"),
Difficulty: big.NewInt(5000000),
Extra: []byte{},
}
MockTransactions, MockReceipts, SenderAddr = createTransactionsAndReceipts()
ReceiptsRlp, _ = rlp.EncodeToBytes(MockReceipts)
MockBlock = types.NewBlock(&MockHeader, MockTransactions, nil, MockReceipts)
MockBlockRlp, _ = rlp.EncodeToBytes(MockBlock)
MockHeaderRlp, _ = rlp.EncodeToBytes(MockBlock.Header())
Address = common.HexToAddress("0xaE9BEa628c4Ce503DcFD7E305CaB4e29E7476592")
AnotherAddress = common.HexToAddress("0xaE9BEa628c4Ce503DcFD7E305CaB4e29E7476593")
ContractAddress = crypto.CreateAddress(SenderAddr, MockTransactions[2].Nonce())
ContractHash = crypto.Keccak256Hash(ContractAddress.Bytes()).String()
MockContractByteCode = []byte{0, 1, 2, 3, 4, 5}
mockTopic11 = common.HexToHash("0x04")
mockTopic12 = common.HexToHash("0x06")
mockTopic21 = common.HexToHash("0x05")
mockTopic22 = common.HexToHash("0x07")
MockLog1 = &types.Log{
Address: Address,
Topics: []common.Hash{mockTopic11, mockTopic12},
Data: []byte{},
}
MockLog2 = &types.Log{
Address: AnotherAddress,
Topics: []common.Hash{mockTopic21, mockTopic22},
Data: []byte{},
}
HeaderCID, _ = ipld.RawdataToCid(ipld.MEthHeader, MockHeaderRlp, multihash.KECCAK_256)
HeaderMhKey = shared.MultihashKeyFromCID(HeaderCID)
Trx1CID, _ = ipld.RawdataToCid(ipld.MEthTx, MockTransactions.GetRlp(0), multihash.KECCAK_256)
Trx1MhKey = shared.MultihashKeyFromCID(Trx1CID)
Trx2CID, _ = ipld.RawdataToCid(ipld.MEthTx, MockTransactions.GetRlp(1), multihash.KECCAK_256)
Trx2MhKey = shared.MultihashKeyFromCID(Trx2CID)
Trx3CID, _ = ipld.RawdataToCid(ipld.MEthTx, MockTransactions.GetRlp(2), multihash.KECCAK_256)
Trx3MhKey = shared.MultihashKeyFromCID(Trx3CID)
Rct1CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, MockReceipts.GetRlp(0), multihash.KECCAK_256)
Rct1MhKey = shared.MultihashKeyFromCID(Rct1CID)
Rct2CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, MockReceipts.GetRlp(1), multihash.KECCAK_256)
Rct2MhKey = shared.MultihashKeyFromCID(Rct2CID)
Rct3CID, _ = ipld.RawdataToCid(ipld.MEthTxReceipt, MockReceipts.GetRlp(2), multihash.KECCAK_256)
Rct3MhKey = shared.MultihashKeyFromCID(Rct3CID)
State1CID, _ = ipld.RawdataToCid(ipld.MEthStateTrie, ContractLeafNode, multihash.KECCAK_256)
State1MhKey = shared.MultihashKeyFromCID(State1CID)
State2CID, _ = ipld.RawdataToCid(ipld.MEthStateTrie, AccountLeafNode, multihash.KECCAK_256)
State2MhKey = shared.MultihashKeyFromCID(State2CID)
StorageCID, _ = ipld.RawdataToCid(ipld.MEthStorageTrie, StorageLeafNode, multihash.KECCAK_256)
StorageMhKey = shared.MultihashKeyFromCID(StorageCID)
MockTrxMeta = []eth.TxModel{
{
CID: "", // This is empty until we go to publish to ipfs
MhKey: "",
Src: SenderAddr.Hex(),
Dst: Address.String(),
Index: 0,
TxHash: MockTransactions[0].Hash().String(),
Data: []byte{},
Deployment: false,
},
{
CID: "",
MhKey: "",
Src: SenderAddr.Hex(),
Dst: AnotherAddress.String(),
Index: 1,
TxHash: MockTransactions[1].Hash().String(),
Data: []byte{},
Deployment: false,
},
{
CID: "",
MhKey: "",
Src: SenderAddr.Hex(),
Dst: "",
Index: 2,
TxHash: MockTransactions[2].Hash().String(),
Data: MockContractByteCode,
Deployment: true,
},
}
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(),
Data: []byte{},
Deployment: false,
},
{
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(),
Data: MockContractByteCode,
Deployment: true,
},
}
MockRctMeta = []eth.ReceiptModel{
{
CID: "",
MhKey: "",
Topic0s: []string{
mockTopic11.String(),
},
Topic1s: []string{
mockTopic12.String(),
},
Contract: "",
ContractHash: "",
LogContracts: []string{
Address.String(),
},
},
{
CID: "",
MhKey: "",
Topic0s: []string{
mockTopic21.String(),
},
Topic1s: []string{
mockTopic22.String(),
},
Contract: "",
ContractHash: "",
LogContracts: []string{
AnotherAddress.String(),
},
},
{
CID: "",
MhKey: "",
Contract: ContractAddress.String(),
ContractHash: ContractHash,
LogContracts: []string{},
},
}
MockRctMetaPostPublish = []eth.ReceiptModel{
{
CID: Rct1CID.String(),
MhKey: Rct1MhKey,
Topic0s: []string{
mockTopic11.String(),
},
Topic1s: []string{
mockTopic12.String(),
},
Contract: "",
ContractHash: "",
LogContracts: []string{
Address.String(),
},
},
{
CID: Rct2CID.String(),
MhKey: Rct2MhKey,
Topic0s: []string{
mockTopic21.String(),
},
Topic1s: []string{
mockTopic22.String(),
},
Contract: "",
ContractHash: "",
LogContracts: []string{
AnotherAddress.String(),
},
},
{
CID: Rct3CID.String(),
MhKey: Rct3MhKey,
Contract: ContractAddress.String(),
ContractHash: ContractHash,
LogContracts: []string{},
},
}
// statediff data
storageLocation = common.HexToHash("0")
StorageLeafKey = crypto.Keccak256Hash(storageLocation[:]).Bytes()
StorageValue = common.Hex2Bytes("01")
StoragePartialPath = common.Hex2Bytes("20290decd9548b62a8d60345a988386fc84ba6bc95484008f6362f93160ef3e563")
StorageLeafNode, _ = rlp.EncodeToBytes([]interface{}{
StoragePartialPath,
StorageValue,
})
nonce1 = uint64(1)
ContractRoot = "0x821e2556a290c86405f8160a2d662042a431ba456b9db265c79bb837c04be5f0"
ContractCodeHash = common.HexToHash("0x753f98a8d4328b15636e46f66f2cb4bc860100aa17967cc145fcd17d1d4710ea")
contractPath = common.Bytes2Hex([]byte{'\x06'})
ContractLeafKey = testhelpers.AddressToLeafKey(ContractAddress)
ContractAccount, _ = rlp.EncodeToBytes(state.Account{
Nonce: nonce1,
Balance: big.NewInt(0),
CodeHash: ContractCodeHash.Bytes(),
Root: common.HexToHash(ContractRoot),
})
ContractPartialPath = common.Hex2Bytes("3114658a74d9cc9f7acf2c5cd696c3494d7c344d78bfec3add0d91ec4e8d1c45")
ContractLeafNode, _ = rlp.EncodeToBytes([]interface{}{
ContractPartialPath,
ContractAccount,
})
nonce0 = uint64(0)
AccountRoot = "0x56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"
AccountCodeHash = common.HexToHash("0xc5d2460186f7233c927e7db2dcc703c0e500b653ca82273b7bfad8045d85a470")
accountPath = common.Bytes2Hex([]byte{'\x0c'})
AccountAddresss = common.HexToAddress("0x0D3ab14BBaD3D99F4203bd7a11aCB94882050E7e")
AccountLeafKey = testhelpers.Account2LeafKey
Account, _ = rlp.EncodeToBytes(state.Account{
Nonce: nonce0,
Balance: big.NewInt(1000),
CodeHash: AccountCodeHash.Bytes(),
Root: common.HexToHash(AccountRoot),
})
AccountPartialPath = common.Hex2Bytes("3957f3e2f04a0764c3a0491b175f69926da61efbcc8f61fa1455fd2d2b4cdd45")
AccountLeafNode, _ = rlp.EncodeToBytes([]interface{}{
AccountPartialPath,
Account,
})
StateDiffs = []statediff.StateNode{
{
Path: []byte{'\x06'},
NodeType: statediff.Leaf,
LeafKey: ContractLeafKey,
NodeValue: ContractLeafNode,
StorageNodes: []statediff.StorageNode{
{
Path: []byte{},
NodeType: statediff.Leaf,
LeafKey: StorageLeafKey,
NodeValue: StorageLeafNode,
},
},
},
{
Path: []byte{'\x0c'},
NodeType: statediff.Leaf,
LeafKey: AccountLeafKey,
NodeValue: AccountLeafNode,
StorageNodes: []statediff.StorageNode{},
},
}
MockStateDiff = statediff.StateObject{
BlockNumber: new(big.Int).Set(BlockNumber),
BlockHash: MockBlock.Hash(),
Nodes: StateDiffs,
}
MockStateDiffBytes, _ = rlp.EncodeToBytes(MockStateDiff)
MockStateNodes = []eth.TrieNode{
{
LeafKey: common.BytesToHash(ContractLeafKey),
Path: []byte{'\x06'},
Value: ContractLeafNode,
Type: statediff.Leaf,
},
{
LeafKey: common.BytesToHash(AccountLeafKey),
Path: []byte{'\x0c'},
Value: AccountLeafNode,
Type: statediff.Leaf,
},
}
MockStateMetaPostPublish = []eth.StateNodeModel{
{
CID: State1CID.String(),
MhKey: State1MhKey,
Path: []byte{'\x06'},
NodeType: 2,
StateKey: common.BytesToHash(ContractLeafKey).Hex(),
},
{
CID: State2CID.String(),
MhKey: State2MhKey,
Path: []byte{'\x0c'},
NodeType: 2,
StateKey: common.BytesToHash(AccountLeafKey).Hex(),
},
}
MockStorageNodes = map[string][]eth.TrieNode{
contractPath: {
{
LeafKey: common.BytesToHash(StorageLeafKey),
Value: StorageLeafNode,
Type: statediff.Leaf,
Path: []byte{},
},
},
}
// aggregate payloads
MockStateDiffPayload = statediff.Payload{
BlockRlp: MockBlockRlp,
StateObjectRlp: MockStateDiffBytes,
ReceiptsRlp: ReceiptsRlp,
TotalDifficulty: MockBlock.Difficulty(),
}
MockConvertedPayload = eth.ConvertedPayload{
TotalDifficulty: MockBlock.Difficulty(),
Block: MockBlock,
Receipts: MockReceipts,
TxMetaData: MockTrxMeta,
ReceiptMetaData: MockRctMeta,
StorageNodes: MockStorageNodes,
StateNodes: MockStateNodes,
}
MockCIDPayload = &eth.CIDPayload{
HeaderCID: eth.HeaderModel{
BlockHash: MockBlock.Hash().String(),
BlockNumber: MockBlock.Number().String(),
CID: HeaderCID.String(),
MhKey: HeaderMhKey,
ParentHash: MockBlock.ParentHash().String(),
TotalDifficulty: MockBlock.Difficulty().String(),
Reward: "5000000000000000000",
StateRoot: MockBlock.Root().String(),
RctRoot: MockBlock.ReceiptHash().String(),
TxRoot: MockBlock.TxHash().String(),
UncleRoot: MockBlock.UncleHash().String(),
Bloom: MockBlock.Bloom().Bytes(),
Timestamp: MockBlock.Time(),
},
UncleCIDs: []eth.UncleModel{},
TransactionCIDs: MockTrxMetaPostPublsh,
ReceiptCIDs: map[common.Hash]eth.ReceiptModel{
MockTransactions[0].Hash(): MockRctMetaPostPublish[0],
MockTransactions[1].Hash(): MockRctMetaPostPublish[1],
MockTransactions[2].Hash(): MockRctMetaPostPublish[2],
},
StateNodeCIDs: MockStateMetaPostPublish,
StorageNodeCIDs: map[string][]eth.StorageNodeModel{
contractPath: {
{
CID: StorageCID.String(),
MhKey: StorageMhKey,
Path: []byte{},
StorageKey: common.BytesToHash(StorageLeafKey).Hex(),
NodeType: 2,
},
},
},
StateAccounts: map[string]eth.StateAccountModel{
contractPath: {
Balance: big.NewInt(0).String(),
Nonce: nonce1,
CodeHash: ContractCodeHash.Bytes(),
StorageRoot: common.HexToHash(ContractRoot).String(),
},
accountPath: {
Balance: big.NewInt(1000).String(),
Nonce: nonce0,
CodeHash: AccountCodeHash.Bytes(),
StorageRoot: common.HexToHash(AccountRoot).String(),
},
},
}
MockCIDWrapper = &eth.CIDWrapper{
BlockNumber: new(big.Int).Set(BlockNumber),
Header: eth.HeaderModel{
BlockNumber: "1",
BlockHash: MockBlock.Hash().String(),
ParentHash: "0x0000000000000000000000000000000000000000000000000000000000000000",
CID: HeaderCID.String(),
MhKey: HeaderMhKey,
TotalDifficulty: MockBlock.Difficulty().String(),
Reward: "5000000000000000000",
StateRoot: MockBlock.Root().String(),
RctRoot: MockBlock.ReceiptHash().String(),
TxRoot: MockBlock.TxHash().String(),
UncleRoot: MockBlock.UncleHash().String(),
Bloom: MockBlock.Bloom().Bytes(),
Timestamp: MockBlock.Time(),
TimesValidated: 1,
},
Transactions: MockTrxMetaPostPublsh,
Receipts: MockRctMetaPostPublish,
Uncles: []eth.UncleModel{},
StateNodes: MockStateMetaPostPublish,
StorageNodes: []eth.StorageNodeWithStateKeyModel{
{
Path: []byte{},
CID: StorageCID.String(),
MhKey: StorageMhKey,
NodeType: 2,
StateKey: common.BytesToHash(ContractLeafKey).Hex(),
StorageKey: common.BytesToHash(StorageLeafKey).Hex(),
},
},
}
HeaderIPLD, _ = blocks.NewBlockWithCid(MockHeaderRlp, HeaderCID)
Trx1IPLD, _ = blocks.NewBlockWithCid(MockTransactions.GetRlp(0), Trx1CID)
Trx2IPLD, _ = blocks.NewBlockWithCid(MockTransactions.GetRlp(1), Trx2CID)
Trx3IPLD, _ = blocks.NewBlockWithCid(MockTransactions.GetRlp(2), Trx3CID)
Rct1IPLD, _ = blocks.NewBlockWithCid(MockReceipts.GetRlp(0), Rct1CID)
Rct2IPLD, _ = blocks.NewBlockWithCid(MockReceipts.GetRlp(1), Rct2CID)
Rct3IPLD, _ = blocks.NewBlockWithCid(MockReceipts.GetRlp(2), Rct3CID)
State1IPLD, _ = blocks.NewBlockWithCid(ContractLeafNode, State1CID)
State2IPLD, _ = blocks.NewBlockWithCid(AccountLeafNode, State2CID)
StorageIPLD, _ = blocks.NewBlockWithCid(StorageLeafNode, StorageCID)
MockIPLDs = eth.IPLDs{
BlockNumber: new(big.Int).Set(BlockNumber),
Header: ipfs.BlockModel{
Data: HeaderIPLD.RawData(),
CID: HeaderIPLD.Cid().String(),
},
Transactions: []ipfs.BlockModel{
{
Data: Trx1IPLD.RawData(),
CID: Trx1IPLD.Cid().String(),
},
{
Data: Trx2IPLD.RawData(),
CID: Trx2IPLD.Cid().String(),
},
{
Data: Trx3IPLD.RawData(),
CID: Trx3IPLD.Cid().String(),
},
},
Receipts: []ipfs.BlockModel{
{
Data: Rct1IPLD.RawData(),
CID: Rct1IPLD.Cid().String(),
},
{
Data: Rct2IPLD.RawData(),
CID: Rct2IPLD.Cid().String(),
},
{
Data: Rct3IPLD.RawData(),
CID: Rct3IPLD.Cid().String(),
},
},
StateNodes: []eth.StateNode{
{
StateLeafKey: common.BytesToHash(ContractLeafKey),
Type: statediff.Leaf,
IPLD: ipfs.BlockModel{
Data: State1IPLD.RawData(),
CID: State1IPLD.Cid().String(),
},
Path: []byte{'\x06'},
},
{
StateLeafKey: common.BytesToHash(AccountLeafKey),
Type: statediff.Leaf,
IPLD: ipfs.BlockModel{
Data: State2IPLD.RawData(),
CID: State2IPLD.Cid().String(),
},
Path: []byte{'\x0c'},
},
},
StorageNodes: []eth.StorageNode{
{
StateLeafKey: common.BytesToHash(ContractLeafKey),
StorageLeafKey: common.BytesToHash(StorageLeafKey),
Type: statediff.Leaf,
IPLD: ipfs.BlockModel{
Data: StorageIPLD.RawData(),
CID: StorageIPLD.Cid().String(),
},
Path: []byte{},
},
},
}
)
// createTransactionsAndReceipts is a helper function to generate signed mock transactions and mock receipts with mock logs
func createTransactionsAndReceipts() (types.Transactions, types.Receipts, common.Address) {
// make transactions
trx1 := types.NewTransaction(0, Address, big.NewInt(1000), 50, big.NewInt(100), []byte{})
trx2 := types.NewTransaction(1, AnotherAddress, big.NewInt(2000), 100, big.NewInt(200), []byte{})
trx3 := types.NewContractCreation(2, big.NewInt(1500), 75, big.NewInt(150), MockContractByteCode)
transactionSigner := types.MakeSigner(params.MainnetChainConfig, new(big.Int).Set(BlockNumber))
mockCurve := elliptic.P256()
mockPrvKey, err := ecdsa.GenerateKey(mockCurve, rand.Reader)
if err != nil {
log.Fatal(err)
}
signedTrx1, err := types.SignTx(trx1, transactionSigner, mockPrvKey)
if err != nil {
log.Fatal(err)
}
signedTrx2, err := types.SignTx(trx2, transactionSigner, mockPrvKey)
if err != nil {
log.Fatal(err)
}
signedTrx3, err := types.SignTx(trx3, transactionSigner, mockPrvKey)
if err != nil {
log.Fatal(err)
}
SenderAddr, err := types.Sender(transactionSigner, signedTrx1) // same for both trx
if err != nil {
log.Fatal(err)
}
// make receipts
mockReceipt1 := types.NewReceipt(common.HexToHash("0x0").Bytes(), false, 50)
mockReceipt1.Logs = []*types.Log{MockLog1}
mockReceipt1.TxHash = signedTrx1.Hash()
mockReceipt2 := types.NewReceipt(common.HexToHash("0x1").Bytes(), false, 100)
mockReceipt2.Logs = []*types.Log{MockLog2}
mockReceipt2.TxHash = signedTrx2.Hash()
mockReceipt3 := types.NewReceipt(common.HexToHash("0x2").Bytes(), false, 75)
mockReceipt3.Logs = []*types.Log{}
mockReceipt3.TxHash = signedTrx3.Hash()
return types.Transactions{signedTrx1, signedTrx2, signedTrx3}, types.Receipts{mockReceipt1, mockReceipt2, mockReceipt3}, SenderAddr
}
pkg/eth/models.go
-126
View File
@@ -1,126 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth
import "github.com/lib/pq"
// HeaderModel is the db model for eth.header_cids
type HeaderModel struct {
ID int64 `db:"id"`
BlockNumber string `db:"block_number"`
BlockHash string `db:"block_hash"`
ParentHash string `db:"parent_hash"`
CID string `db:"cid"`
MhKey string `db:"mh_key"`
TotalDifficulty string `db:"td"`
NodeID int64 `db:"node_id"`
Reward string `db:"reward"`
StateRoot string `db:"state_root"`
UncleRoot string `db:"uncle_root"`
TxRoot string `db:"tx_root"`
RctRoot string `db:"receipt_root"`
Bloom []byte `db:"bloom"`
Timestamp uint64 `db:"timestamp"`
TimesValidated int64 `db:"times_validated"`
}
// UncleModel is the db model for eth.uncle_cids
type UncleModel struct {
ID int64 `db:"id"`
HeaderID int64 `db:"header_id"`
BlockHash string `db:"block_hash"`
ParentHash string `db:"parent_hash"`
CID string `db:"cid"`
MhKey string `db:"mh_key"`
Reward string `db:"reward"`
}
// TxModel is the db model for eth.transaction_cids
type TxModel struct {
ID int64 `db:"id"`
HeaderID int64 `db:"header_id"`
Index int64 `db:"index"`
TxHash string `db:"tx_hash"`
CID string `db:"cid"`
MhKey string `db:"mh_key"`
Dst string `db:"dst"`
Src string `db:"src"`
Data []byte `db:"tx_data"`
Deployment bool `db:"deployment"`
}
// ReceiptModel is the db model for eth.receipt_cids
type ReceiptModel struct {
ID int64 `db:"id"`
TxID int64 `db:"tx_id"`
CID string `db:"cid"`
MhKey string `db:"mh_key"`
Contract string `db:"contract"`
ContractHash string `db:"contract_hash"`
LogContracts pq.StringArray `db:"log_contracts"`
Topic0s pq.StringArray `db:"topic0s"`
Topic1s pq.StringArray `db:"topic1s"`
Topic2s pq.StringArray `db:"topic2s"`
Topic3s pq.StringArray `db:"topic3s"`
}
// StateNodeModel is the db model for eth.state_cids
type StateNodeModel struct {
ID int64 `db:"id"`
HeaderID int64 `db:"header_id"`
Path []byte `db:"state_path"`
StateKey string `db:"state_leaf_key"`
NodeType int `db:"node_type"`
CID string `db:"cid"`
MhKey string `db:"mh_key"`
Diff bool `db:"diff"`
}
// StorageNodeModel is the db model for eth.storage_cids
type StorageNodeModel struct {
ID int64 `db:"id"`
StateID int64 `db:"state_id"`
Path []byte `db:"storage_path"`
StorageKey string `db:"storage_leaf_key"`
NodeType int `db:"node_type"`
CID string `db:"cid"`
MhKey string `db:"mh_key"`
Diff bool `db:"diff"`
}
// StorageNodeWithStateKeyModel is a db model for eth.storage_cids + eth.state_cids.state_key
type StorageNodeWithStateKeyModel struct {
ID int64 `db:"id"`
StateID int64 `db:"state_id"`
Path []byte `db:"storage_path"`
StateKey string `db:"state_leaf_key"`
StorageKey string `db:"storage_leaf_key"`
NodeType int `db:"node_type"`
CID string `db:"cid"`
MhKey string `db:"mh_key"`
Diff bool `db:"diff"`
}
// StateAccountModel is a db model for an eth state account (decoded value of state leaf node)
type StateAccountModel struct {
ID int64 `db:"id"`
StateID int64 `db:"state_id"`
Balance string `db:"balance"`
Nonce uint64 `db:"nonce"`
CodeHash []byte `db:"code_hash"`
StorageRoot string `db:"storage_root"`
}
pkg/eth/payload_fetcher.go
-88
View File
@@ -1,88 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth
import (
"context"
"fmt"
"time"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/statediff"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
// BatchClient is an interface to a batch-fetching geth rpc client; created to allow mock insertion
type BatchClient interface {
BatchCallContext(ctx context.Context, batch []rpc.BatchElem) error
}
// PayloadFetcher satisfies the PayloadFetcher interface for ethereum
type PayloadFetcher struct {
// PayloadFetcher is thread-safe as long as the underlying client is thread-safe, since it has/modifies no other state
// http.Client is thread-safe
client BatchClient
timeout time.Duration
params statediff.Params
}
const method = "statediff_stateDiffAt"
// NewPayloadFetcher returns a PayloadFetcher
func NewPayloadFetcher(bc BatchClient, timeout time.Duration) *PayloadFetcher {
return &PayloadFetcher{
client: bc,
timeout: timeout,
params: statediff.Params{
IncludeReceipts: true,
IncludeTD: true,
IncludeBlock: true,
IntermediateStateNodes: true,
IntermediateStorageNodes: true,
},
}
}
// FetchAt fetches the statediff payloads at the given block heights
// Calls StateDiffAt(ctx context.Context, blockNumber uint64, params Params) (*Payload, error)
func (fetcher *PayloadFetcher) FetchAt(blockHeights []uint64) ([]shared.RawChainData, error) {
batch := make([]rpc.BatchElem, 0)
for _, height := range blockHeights {
batch = append(batch, rpc.BatchElem{
Method: method,
Args: []interface{}{height, fetcher.params},
Result: new(statediff.Payload),
})
}
ctx, cancel := context.WithTimeout(context.Background(), fetcher.timeout)
defer cancel()
if err := fetcher.client.BatchCallContext(ctx, batch); err != nil {
return nil, fmt.Errorf("ethereum PayloadFetcher batch err for block range %d-%d: %s", blockHeights[0], blockHeights[len(blockHeights)-1], err.Error())
}
results := make([]shared.RawChainData, 0, len(blockHeights))
for _, batchElem := range batch {
if batchElem.Error != nil {
return nil, fmt.Errorf("ethereum PayloadFetcher err at blockheight %d: %s", batchElem.Args[0].(uint64), batchElem.Error.Error())
}
payload, ok := batchElem.Result.(*statediff.Payload)
if ok {
results = append(results, *payload)
}
}
return results, nil
}
pkg/eth/payload_fetcher_test.go
-65
View File
@@ -1,65 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth_test
import (
"time"
"github.com/ethereum/go-ethereum/statediff"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth/mocks"
)
var _ = Describe("StateDiffFetcher", func() {
Describe("FetchStateDiffsAt", func() {
var (
mc *mocks.BackFillerClient
stateDiffFetcher *eth.PayloadFetcher
payload2 statediff.Payload
blockNumber2 uint64
)
BeforeEach(func() {
mc = new(mocks.BackFillerClient)
err := mc.SetReturnDiffAt(mocks.BlockNumber.Uint64(), mocks.MockStateDiffPayload)
Expect(err).ToNot(HaveOccurred())
payload2 = mocks.MockStateDiffPayload
payload2.BlockRlp = []byte{}
blockNumber2 = mocks.BlockNumber.Uint64() + 1
err = mc.SetReturnDiffAt(blockNumber2, payload2)
Expect(err).ToNot(HaveOccurred())
stateDiffFetcher = eth.NewPayloadFetcher(mc, time.Second*60)
})
It("Batch calls statediff_stateDiffAt", func() {
blockHeights := []uint64{
mocks.BlockNumber.Uint64(),
blockNumber2,
}
stateDiffPayloads, err := stateDiffFetcher.FetchAt(blockHeights)
Expect(err).ToNot(HaveOccurred())
Expect(len(stateDiffPayloads)).To(Equal(2))
payload1, ok := stateDiffPayloads[0].(statediff.Payload)
Expect(ok).To(BeTrue())
payload2, ok := stateDiffPayloads[1].(statediff.Payload)
Expect(ok).To(BeTrue())
Expect(payload1).To(Equal(mocks.MockStateDiffPayload))
Expect(payload2).To(Equal(payload2))
})
})
})
pkg/eth/publisher.go
-228
View File
@@ -1,228 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth
import (
"fmt"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/statediff"
"github.com/jmoiron/sqlx"
"github.com/multiformats/go-multihash"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/ipfs/ipld"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/postgres"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
// IPLDPublisher satisfies the IPLDPublisher interface for ethereum
// It interfaces directly with the public.blocks table of PG-IPFS rather than going through an ipfs intermediary
// It publishes and indexes IPLDs together in a single sqlx.Tx
type IPLDPublisher struct {
indexer *CIDIndexer
}
// NewIPLDPublisher creates a pointer to a new IPLDPublisher which satisfies the IPLDPublisher interface
func NewIPLDPublisher(db *postgres.DB) *IPLDPublisher {
return &IPLDPublisher{
indexer: NewCIDIndexer(db),
}
}
// Publish publishes an IPLDPayload to IPFS and returns the corresponding CIDPayload
func (pub *IPLDPublisher) Publish(payload shared.ConvertedData) error {
ipldPayload, ok := payload.(ConvertedPayload)
if !ok {
return fmt.Errorf("eth IPLDPublisher expected payload type %T got %T", ConvertedPayload{}, payload)
}
// Generate the iplds
headerNode, uncleNodes, txNodes, txTrieNodes, rctNodes, rctTrieNodes, err := ipld.FromBlockAndReceipts(ipldPayload.Block, ipldPayload.Receipts)
if err != nil {
return err
}
// Begin new db tx
tx, err := pub.indexer.db.Beginx()
if err != nil {
return err
}
defer func() {
if p := recover(); p != nil {
shared.Rollback(tx)
panic(p)
} else if err != nil {
shared.Rollback(tx)
} else {
err = tx.Commit()
}
}()
// Publish trie nodes
for _, node := range txTrieNodes {
if err := shared.PublishIPLD(tx, node); err != nil {
return err
}
}
for _, node := range rctTrieNodes {
if err := shared.PublishIPLD(tx, node); err != nil {
return err
}
}
// Publish and index header
if err := shared.PublishIPLD(tx, headerNode); err != nil {
return err
}
reward := CalcEthBlockReward(ipldPayload.Block.Header(), ipldPayload.Block.Uncles(), ipldPayload.Block.Transactions(), ipldPayload.Receipts)
header := HeaderModel{
CID: headerNode.Cid().String(),
MhKey: shared.MultihashKeyFromCID(headerNode.Cid()),
ParentHash: ipldPayload.Block.ParentHash().String(),
BlockNumber: ipldPayload.Block.Number().String(),
BlockHash: ipldPayload.Block.Hash().String(),
TotalDifficulty: ipldPayload.TotalDifficulty.String(),
Reward: reward.String(),
Bloom: ipldPayload.Block.Bloom().Bytes(),
StateRoot: ipldPayload.Block.Root().String(),
RctRoot: ipldPayload.Block.ReceiptHash().String(),
TxRoot: ipldPayload.Block.TxHash().String(),
UncleRoot: ipldPayload.Block.UncleHash().String(),
Timestamp: ipldPayload.Block.Time(),
}
headerID, err := pub.indexer.indexHeaderCID(tx, header)
if err != nil {
return err
}
// Publish and index uncles
for _, uncleNode := range uncleNodes {
if err := shared.PublishIPLD(tx, uncleNode); err != nil {
return err
}
uncleReward := CalcUncleMinerReward(ipldPayload.Block.Number().Int64(), uncleNode.Number.Int64())
uncle := UncleModel{
CID: uncleNode.Cid().String(),
MhKey: shared.MultihashKeyFromCID(uncleNode.Cid()),
ParentHash: uncleNode.ParentHash.String(),
BlockHash: uncleNode.Hash().String(),
Reward: uncleReward.String(),
}
if err := pub.indexer.indexUncleCID(tx, uncle, headerID); err != nil {
return err
}
}
// Publish and index txs and receipts
for i, txNode := range txNodes {
if err := shared.PublishIPLD(tx, txNode); err != nil {
return err
}
rctNode := rctNodes[i]
if err := shared.PublishIPLD(tx, rctNode); err != nil {
return err
}
txModel := ipldPayload.TxMetaData[i]
txModel.CID = txNode.Cid().String()
txModel.MhKey = shared.MultihashKeyFromCID(txNode.Cid())
txID, err := pub.indexer.indexTransactionCID(tx, txModel, headerID)
if err != nil {
return err
}
// If tx is a contract deployment, publish the data (code)
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
}
}
rctModel := ipldPayload.ReceiptMetaData[i]
rctModel.CID = rctNode.Cid().String()
rctModel.MhKey = shared.MultihashKeyFromCID(rctNode.Cid())
if err := pub.indexer.indexReceiptCID(tx, rctModel, txID); err != nil {
return err
}
}
// Publish and index state and storage
err = pub.publishAndIndexStateAndStorage(tx, ipldPayload, headerID)
return err // return err variable explicitly so that we return the err = tx.Commit() assignment in the defer
}
func (pub *IPLDPublisher) publishAndIndexStateAndStorage(tx *sqlx.Tx, ipldPayload ConvertedPayload, headerID int64) error {
// Publish and index state and storage
for _, stateNode := range ipldPayload.StateNodes {
stateCIDStr, err := shared.PublishRaw(tx, ipld.MEthStateTrie, multihash.KECCAK_256, stateNode.Value)
if err != nil {
return err
}
mhKey, _ := shared.MultihashKeyFromCIDString(stateCIDStr)
stateModel := StateNodeModel{
Path: stateNode.Path,
StateKey: stateNode.LeafKey.String(),
CID: stateCIDStr,
MhKey: mhKey,
NodeType: ResolveFromNodeType(stateNode.Type),
}
stateID, err := pub.indexer.indexStateCID(tx, stateModel, headerID)
if err != nil {
return err
}
// If we have a leaf, decode and index the account data and any associated storage diffs
if stateNode.Type == statediff.Leaf {
var i []interface{}
if err := rlp.DecodeBytes(stateNode.Value, &i); err != nil {
return err
}
if len(i) != 2 {
return fmt.Errorf("eth IPLDPublisher expected state leaf node rlp to decode into two elements")
}
var account state.Account
if err := rlp.DecodeBytes(i[1].([]byte), &account); err != nil {
return err
}
accountModel := StateAccountModel{
Balance: account.Balance.String(),
Nonce: account.Nonce,
CodeHash: account.CodeHash,
StorageRoot: account.Root.String(),
}
if err := pub.indexer.indexStateAccount(tx, accountModel, stateID); err != nil {
return err
}
for _, storageNode := range ipldPayload.StorageNodes[common.Bytes2Hex(stateNode.Path)] {
storageCIDStr, err := shared.PublishRaw(tx, ipld.MEthStorageTrie, multihash.KECCAK_256, storageNode.Value)
if err != nil {
return err
}
mhKey, _ := shared.MultihashKeyFromCIDString(storageCIDStr)
storageModel := StorageNodeModel{
Path: storageNode.Path,
StorageKey: storageNode.LeafKey.Hex(),
CID: storageCIDStr,
MhKey: mhKey,
NodeType: ResolveFromNodeType(storageNode.Type),
}
if err := pub.indexer.indexStorageCID(tx, storageModel, stateID); err != nil {
return err
}
}
}
}
return nil
}
pkg/eth/publisher_test.go
-233
View File
@@ -1,233 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth_test
import (
"github.com/ethereum/go-ethereum/common"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-ipfs-blockstore"
"github.com/ipfs/go-ipfs-ds-help"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth/mocks"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/postgres"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
var _ = Describe("PublishAndIndexer", func() {
var (
db *postgres.DB
err error
repo *eth.IPLDPublisher
ipfsPgGet = `SELECT data FROM public.blocks
WHERE key = $1`
)
BeforeEach(func() {
db, err = shared.SetupDB()
Expect(err).ToNot(HaveOccurred())
repo = eth.NewIPLDPublisher(db)
})
AfterEach(func() {
eth.TearDownDB(db)
})
Describe("Publish", func() {
It("Published and indexes header IPLDs in a single tx", func() {
err = repo.Publish(mocks.MockConvertedPayload)
Expect(err).ToNot(HaveOccurred())
pgStr := `SELECT cid, td, reward, id
FROM eth.header_cids
WHERE block_number = $1`
// check header was properly indexed
type res struct {
CID string
TD string
Reward string
ID int
}
header := new(res)
err = db.QueryRowx(pgStr, 1).StructScan(header)
Expect(err).ToNot(HaveOccurred())
Expect(header.CID).To(Equal(mocks.HeaderCID.String()))
Expect(header.TD).To(Equal(mocks.MockBlock.Difficulty().String()))
Expect(header.Reward).To(Equal("5000000000000000000"))
dc, err := cid.Decode(header.CID)
Expect(err).ToNot(HaveOccurred())
mhKey := dshelp.MultihashToDsKey(dc.Hash())
prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
var data []byte
err = db.Get(&data, ipfsPgGet, prefixedKey)
Expect(err).ToNot(HaveOccurred())
Expect(data).To(Equal(mocks.MockHeaderRlp))
})
It("Publishes and indexes transaction IPLDs in a single tx", func() {
err = repo.Publish(mocks.MockConvertedPayload)
Expect(err).ToNot(HaveOccurred())
// check that txs were properly indexed
trxs := make([]string, 0)
pgStr := `SELECT transaction_cids.cid FROM eth.transaction_cids INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.id)
WHERE header_cids.block_number = $1`
err = db.Select(&trxs, pgStr, 1)
Expect(err).ToNot(HaveOccurred())
Expect(len(trxs)).To(Equal(3))
Expect(shared.ListContainsString(trxs, mocks.Trx1CID.String())).To(BeTrue())
Expect(shared.ListContainsString(trxs, mocks.Trx2CID.String())).To(BeTrue())
Expect(shared.ListContainsString(trxs, mocks.Trx3CID.String())).To(BeTrue())
// and published
for _, c := range trxs {
dc, err := cid.Decode(c)
Expect(err).ToNot(HaveOccurred())
mhKey := dshelp.MultihashToDsKey(dc.Hash())
prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
var data []byte
err = db.Get(&data, ipfsPgGet, prefixedKey)
Expect(err).ToNot(HaveOccurred())
switch c {
case mocks.Trx1CID.String():
Expect(data).To(Equal(mocks.MockTransactions.GetRlp(0)))
case mocks.Trx2CID.String():
Expect(data).To(Equal(mocks.MockTransactions.GetRlp(1)))
case mocks.Trx3CID.String():
Expect(data).To(Equal(mocks.MockTransactions.GetRlp(2)))
}
}
})
It("Publishes and indexes receipt IPLDs in a single tx", func() {
err = repo.Publish(mocks.MockConvertedPayload)
Expect(err).ToNot(HaveOccurred())
// check receipts were properly indexed
rcts := make([]string, 0)
pgStr := `SELECT receipt_cids.cid FROM eth.receipt_cids, eth.transaction_cids, eth.header_cids
WHERE receipt_cids.tx_id = transaction_cids.id
AND transaction_cids.header_id = header_cids.id
AND header_cids.block_number = $1`
err = db.Select(&rcts, pgStr, 1)
Expect(err).ToNot(HaveOccurred())
Expect(len(rcts)).To(Equal(3))
Expect(shared.ListContainsString(rcts, mocks.Rct1CID.String())).To(BeTrue())
Expect(shared.ListContainsString(rcts, mocks.Rct2CID.String())).To(BeTrue())
Expect(shared.ListContainsString(rcts, mocks.Rct3CID.String())).To(BeTrue())
// and published
for _, c := range rcts {
dc, err := cid.Decode(c)
Expect(err).ToNot(HaveOccurred())
mhKey := dshelp.MultihashToDsKey(dc.Hash())
prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
var data []byte
err = db.Get(&data, ipfsPgGet, prefixedKey)
Expect(err).ToNot(HaveOccurred())
switch c {
case mocks.Rct1CID.String():
Expect(data).To(Equal(mocks.MockReceipts.GetRlp(0)))
case mocks.Rct2CID.String():
Expect(data).To(Equal(mocks.MockReceipts.GetRlp(1)))
case mocks.Rct3CID.String():
Expect(data).To(Equal(mocks.MockReceipts.GetRlp(2)))
}
}
})
It("Publishes and indexes state IPLDs in a single tx", func() {
err = repo.Publish(mocks.MockConvertedPayload)
Expect(err).ToNot(HaveOccurred())
// check that state nodes were properly indexed and published
stateNodes := make([]eth.StateNodeModel, 0)
pgStr := `SELECT state_cids.id, state_cids.cid, state_cids.state_leaf_key, state_cids.node_type, state_cids.state_path, state_cids.header_id
FROM eth.state_cids INNER JOIN eth.header_cids ON (state_cids.header_id = header_cids.id)
WHERE header_cids.block_number = $1`
err = db.Select(&stateNodes, pgStr, 1)
Expect(err).ToNot(HaveOccurred())
Expect(len(stateNodes)).To(Equal(2))
for _, stateNode := range stateNodes {
var data []byte
dc, err := cid.Decode(stateNode.CID)
Expect(err).ToNot(HaveOccurred())
mhKey := dshelp.MultihashToDsKey(dc.Hash())
prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
err = db.Get(&data, ipfsPgGet, prefixedKey)
Expect(err).ToNot(HaveOccurred())
pgStr = `SELECT * from eth.state_accounts WHERE state_id = $1`
var account eth.StateAccountModel
err = db.Get(&account, pgStr, stateNode.ID)
Expect(err).ToNot(HaveOccurred())
if stateNode.CID == mocks.State1CID.String() {
Expect(stateNode.NodeType).To(Equal(2))
Expect(stateNode.StateKey).To(Equal(common.BytesToHash(mocks.ContractLeafKey).Hex()))
Expect(stateNode.Path).To(Equal([]byte{'\x06'}))
Expect(data).To(Equal(mocks.ContractLeafNode))
Expect(account).To(Equal(eth.StateAccountModel{
ID: account.ID,
StateID: stateNode.ID,
Balance: "0",
CodeHash: mocks.ContractCodeHash.Bytes(),
StorageRoot: mocks.ContractRoot,
Nonce: 1,
}))
}
if stateNode.CID == mocks.State2CID.String() {
Expect(stateNode.NodeType).To(Equal(2))
Expect(stateNode.StateKey).To(Equal(common.BytesToHash(mocks.AccountLeafKey).Hex()))
Expect(stateNode.Path).To(Equal([]byte{'\x0c'}))
Expect(data).To(Equal(mocks.AccountLeafNode))
Expect(account).To(Equal(eth.StateAccountModel{
ID: account.ID,
StateID: stateNode.ID,
Balance: "1000",
CodeHash: mocks.AccountCodeHash.Bytes(),
StorageRoot: mocks.AccountRoot,
Nonce: 0,
}))
}
}
pgStr = `SELECT * from eth.state_accounts WHERE state_id = $1`
})
It("Publishes and indexes storage IPLDs in a single tx", func() {
err = repo.Publish(mocks.MockConvertedPayload)
Expect(err).ToNot(HaveOccurred())
// check that storage nodes were properly indexed
storageNodes := make([]eth.StorageNodeWithStateKeyModel, 0)
pgStr := `SELECT storage_cids.cid, state_cids.state_leaf_key, storage_cids.storage_leaf_key, storage_cids.node_type, storage_cids.storage_path
FROM eth.storage_cids, eth.state_cids, eth.header_cids
WHERE storage_cids.state_id = state_cids.id
AND state_cids.header_id = header_cids.id
AND header_cids.block_number = $1`
err = db.Select(&storageNodes, pgStr, 1)
Expect(err).ToNot(HaveOccurred())
Expect(len(storageNodes)).To(Equal(1))
Expect(storageNodes[0]).To(Equal(eth.StorageNodeWithStateKeyModel{
CID: mocks.StorageCID.String(),
NodeType: 2,
StorageKey: common.BytesToHash(mocks.StorageLeafKey).Hex(),
StateKey: common.BytesToHash(mocks.ContractLeafKey).Hex(),
Path: []byte{},
}))
var data []byte
dc, err := cid.Decode(storageNodes[0].CID)
Expect(err).ToNot(HaveOccurred())
mhKey := dshelp.MultihashToDsKey(dc.Hash())
prefixedKey := blockstore.BlockPrefix.String() + mhKey.String()
err = db.Get(&data, ipfsPgGet, prefixedKey)
Expect(err).ToNot(HaveOccurred())
Expect(data).To(Equal(mocks.StorageLeafNode))
})
})
})
pkg/eth/reward.go
-76
View File
@@ -1,76 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth
import (
"math/big"
"github.com/ethereum/go-ethereum/core/types"
)
func CalcEthBlockReward(header *types.Header, uncles []*types.Header, txs types.Transactions, receipts types.Receipts) *big.Int {
staticBlockReward := staticRewardByBlockNumber(header.Number.Int64())
transactionFees := calcEthTransactionFees(txs, receipts)
uncleInclusionRewards := calcEthUncleInclusionRewards(header, uncles)
tmp := transactionFees.Add(transactionFees, uncleInclusionRewards)
return tmp.Add(tmp, staticBlockReward)
}
func CalcUncleMinerReward(blockNumber, uncleBlockNumber int64) *big.Int {
staticBlockReward := staticRewardByBlockNumber(blockNumber)
rewardDiv8 := staticBlockReward.Div(staticBlockReward, big.NewInt(8))
mainBlock := big.NewInt(blockNumber)
uncleBlock := big.NewInt(uncleBlockNumber)
uncleBlockPlus8 := uncleBlock.Add(uncleBlock, big.NewInt(8))
uncleBlockPlus8MinusMainBlock := uncleBlockPlus8.Sub(uncleBlockPlus8, mainBlock)
return rewardDiv8.Mul(rewardDiv8, uncleBlockPlus8MinusMainBlock)
}
func staticRewardByBlockNumber(blockNumber int64) *big.Int {
staticBlockReward := new(big.Int)
//https://blog.ethereum.org/2017/10/12/byzantium-hf-announcement/
if blockNumber >= 7280000 {
staticBlockReward.SetString("2000000000000000000", 10)
} else if blockNumber >= 4370000 {
staticBlockReward.SetString("3000000000000000000", 10)
} else {
staticBlockReward.SetString("5000000000000000000", 10)
}
return staticBlockReward
}
func calcEthTransactionFees(txs types.Transactions, receipts types.Receipts) *big.Int {
transactionFees := new(big.Int)
for i, transaction := range txs {
receipt := receipts[i]
gasPrice := big.NewInt(transaction.GasPrice().Int64())
gasUsed := big.NewInt(int64(receipt.GasUsed))
transactionFee := gasPrice.Mul(gasPrice, gasUsed)
transactionFees = transactionFees.Add(transactionFees, transactionFee)
}
return transactionFees
}
func calcEthUncleInclusionRewards(header *types.Header, uncles []*types.Header) *big.Int {
uncleInclusionRewards := new(big.Int)
for range uncles {
staticBlockReward := staticRewardByBlockNumber(header.Number.Int64())
staticBlockReward.Div(staticBlockReward, big.NewInt(32))
uncleInclusionRewards.Add(uncleInclusionRewards, staticBlockReward)
}
return uncleInclusionRewards
}
pkg/eth/streamer.go
-72
View File
@@ -1,72 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth
import (
"context"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/statediff"
"github.com/sirupsen/logrus"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
const (
PayloadChanBufferSize = 20000 // the max eth sub buffer size
)
// StreamClient is an interface for subscribing and streaming from geth
type StreamClient interface {
Subscribe(ctx context.Context, namespace string, payloadChan interface{}, args ...interface{}) (*rpc.ClientSubscription, error)
}
// PayloadStreamer satisfies the PayloadStreamer interface for ethereum
type PayloadStreamer struct {
Client StreamClient
params statediff.Params
}
// NewPayloadStreamer creates a pointer to a new PayloadStreamer which satisfies the PayloadStreamer interface for ethereum
func NewPayloadStreamer(client StreamClient) *PayloadStreamer {
return &PayloadStreamer{
Client: client,
params: statediff.Params{
IncludeBlock: true,
IncludeTD: true,
IncludeReceipts: true,
IntermediateStorageNodes: true,
IntermediateStateNodes: true,
},
}
}
// Stream is the main loop for subscribing to data from the Geth state diff process
// Satisfies the shared.PayloadStreamer interface
func (ps *PayloadStreamer) Stream(payloadChan chan shared.RawChainData) (shared.ClientSubscription, error) {
stateDiffChan := make(chan statediff.Payload, PayloadChanBufferSize)
logrus.Debug("streaming diffs from geth")
go func() {
for {
select {
case payload := <-stateDiffChan:
payloadChan <- payload
}
}
}()
return ps.Client.Subscribe(context.Background(), "statediff", stateDiffChan, "stream", ps.params)
}
pkg/eth/streamer_test.go
-34
View File
@@ -1,34 +0,0 @@
// Copyright 2019 Vulcanize
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package eth_test
import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/eth/mocks"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/shared"
)
var _ = Describe("StateDiff Streamer", func() {
It("subscribes to the geth statediff service", func() {
client := &mocks.StreamClient{}
streamer := eth.NewPayloadStreamer(client)
payloadChan := make(chan shared.RawChainData)
_, err := streamer.Stream(payloadChan)
Expect(err).NotTo(HaveOccurred())
})
})
pkg/eth/types.go
-112
View File
@@ -1,112 +0,0 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package eth
import (
"math/big"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/statediff"
"github.com/vulcanize/ipfs-blockchain-watcher/pkg/ipfs"
)
// ConvertedPayload is a custom type which packages raw ETH data for publishing to IPFS and filtering to subscribers
// Returned by PayloadConverter
// Passed to IPLDPublisher and ResponseFilterer
type ConvertedPayload struct {
TotalDifficulty *big.Int
Block *types.Block
TxMetaData []TxModel
Receipts types.Receipts
ReceiptMetaData []ReceiptModel
StateNodes []TrieNode
StorageNodes map[string][]TrieNode
}
// Height satisfies the StreamedIPLDs interface
func (i ConvertedPayload) Height() int64 {
return i.Block.Number().Int64()
}
// Trie struct used to flag node as leaf or not
type TrieNode struct {
Path []byte
LeafKey common.Hash
Value []byte
Type statediff.NodeType
}
// CIDPayload is a struct to hold all the CIDs and their associated meta data for indexing in Postgres
// Returned by IPLDPublisher
// Passed to CIDIndexer
type CIDPayload struct {
HeaderCID HeaderModel
UncleCIDs []UncleModel
TransactionCIDs []TxModel
ReceiptCIDs map[common.Hash]ReceiptModel
StateNodeCIDs []StateNodeModel
StateAccounts map[string]StateAccountModel
StorageNodeCIDs map[string][]StorageNodeModel
}
// CIDWrapper is used to direct fetching of IPLDs from IPFS
// Returned by CIDRetriever
// Passed to IPLDFetcher
type CIDWrapper struct {
BlockNumber *big.Int
Header HeaderModel
Uncles []UncleModel
Transactions []TxModel
Receipts []ReceiptModel
StateNodes []StateNodeModel
StorageNodes []StorageNodeWithStateKeyModel
}
// IPLDs is used to package raw IPLD block data fetched from IPFS and returned by the server
// Returned by IPLDFetcher and ResponseFilterer
type IPLDs struct {
BlockNumber *big.Int
TotalDifficulty *big.Int
Header ipfs.BlockModel
Uncles []ipfs.BlockModel
Transactions []ipfs.BlockModel
Receipts []ipfs.BlockModel
StateNodes []StateNode
StorageNodes []StorageNode
}
// Height satisfies the StreamedIPLDs interface
func (i IPLDs) Height() int64 {
return i.BlockNumber.Int64()
}
type StateNode struct {
Type statediff.NodeType
StateLeafKey common.Hash
Path []byte
IPLD ipfs.BlockModel
}
type StorageNode struct {
Type statediff.NodeType
StateLeafKey common.Hash
StorageLeafKey common.Hash
Path []byte
IPLD ipfs.BlockModel
}