forked from cerc-io/ipld-eth-server
Sync only missing blocks on cold import
= Add eth_node_fingerprint to block that can be imitated by both hot and cold imports - Only sync missing blocks (blocks that are missing or don't share a fingerprint) on cold import - Set block is_final status after import
This commit is contained in:
Rob Mulholand
@@ -0,0 +1,13 @@
|
||||
package cold_import_test
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
. "github.com/onsi/ginkgo"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
func TestColdImport(t *testing.T) {
|
||||
RegisterFailHandler(Fail)
|
||||
RunSpecs(t, "ColdImport Suite")
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package cold_import
|
||||
|
||||
import (
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/ethereum"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/geth/converters/common"
|
||||
)
|
||||
|
||||
type ColdImporter struct {
|
||||
blockRepository datastore.BlockRepository
|
||||
converter common.BlockConverter
|
||||
ethDB ethereum.Database
|
||||
receiptRepository datastore.ReceiptRepository
|
||||
}
|
||||
|
||||
func NewColdImporter(ethDB ethereum.Database, blockRepository datastore.BlockRepository, receiptRepository datastore.ReceiptRepository, converter common.BlockConverter) *ColdImporter {
|
||||
return &ColdImporter{
|
||||
blockRepository: blockRepository,
|
||||
converter: converter,
|
||||
ethDB: ethDB,
|
||||
receiptRepository: receiptRepository,
|
||||
}
|
||||
}
|
||||
|
||||
func (ci *ColdImporter) Execute(startingBlockNumber int64, endingBlockNumber int64, nodeId string) error {
|
||||
missingBlocks := ci.blockRepository.MissingBlockNumbers(startingBlockNumber, endingBlockNumber, nodeId)
|
||||
for _, n := range missingBlocks {
|
||||
hash := ci.ethDB.GetBlockHash(n)
|
||||
|
||||
blockId, err := ci.createBlocksAndTransactions(hash, n)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = ci.createReceiptsAndLogs(hash, n, blockId)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
ci.blockRepository.SetBlocksStatus(endingBlockNumber)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ci *ColdImporter) createBlocksAndTransactions(hash []byte, i int64) (int64, error) {
|
||||
block := ci.ethDB.GetBlock(hash, i)
|
||||
coreBlock, err := ci.converter.ToCoreBlock(block)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return ci.blockRepository.CreateOrUpdateBlock(coreBlock)
|
||||
}
|
||||
|
||||
func (ci *ColdImporter) createReceiptsAndLogs(hash []byte, number int64, blockId int64) error {
|
||||
receipts := ci.ethDB.GetBlockReceipts(hash, number)
|
||||
coreReceipts := common.ToCoreReceipts(receipts)
|
||||
return ci.receiptRepository.CreateReceiptsAndLogs(blockId, coreReceipts)
|
||||
}
|
||||
@@ -0,0 +1,133 @@
|
||||
package cold_import_test
|
||||
|
||||
import (
|
||||
"github.com/ethereum/go-ethereum/core/types"
|
||||
. "github.com/onsi/ginkgo"
|
||||
. "github.com/onsi/gomega"
|
||||
|
||||
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres/repositories"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/fakes"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/geth/cold_import"
|
||||
vulcCommon "github.com/vulcanize/vulcanizedb/pkg/geth/converters/common"
|
||||
)
|
||||
|
||||
var _ = Describe("Geth cold importer", func() {
|
||||
var fakeGethBlock *types.Block
|
||||
|
||||
BeforeEach(func() {
|
||||
header := &types.Header{}
|
||||
transactions := []*types.Transaction{}
|
||||
uncles := []*types.Header{}
|
||||
receipts := []*types.Receipt{}
|
||||
fakeGethBlock = types.NewBlock(header, transactions, uncles, receipts)
|
||||
})
|
||||
|
||||
It("only populates missing blocks", func() {
|
||||
mockEthereumDatabase := fakes.NewMockEthereumDatabase()
|
||||
mockBlockRepository := fakes.NewMockBlockRepository()
|
||||
mockReceiptRepository := fakes.NewMockReceiptRepository()
|
||||
mockTransactionConverter := fakes.NewMockTransactionConverter()
|
||||
blockConverter := vulcCommon.NewBlockConverter(mockTransactionConverter)
|
||||
|
||||
nodeId := "node_id"
|
||||
startingBlockNumber := int64(120)
|
||||
missingBlockNumber := int64(123)
|
||||
endingBlockNumber := int64(125)
|
||||
fakeHash := []byte{1, 2, 3, 4, 5}
|
||||
mockBlockRepository.SetMissingBlockNumbersReturnArray([]int64{missingBlockNumber})
|
||||
mockEthereumDatabase.SetReturnHash(fakeHash)
|
||||
mockEthereumDatabase.SetReturnBlock(fakeGethBlock)
|
||||
importer := cold_import.NewColdImporter(mockEthereumDatabase, mockBlockRepository, mockReceiptRepository, blockConverter)
|
||||
|
||||
importer.Execute(startingBlockNumber, endingBlockNumber, nodeId)
|
||||
|
||||
mockBlockRepository.AssertMissingBlockNumbersCalledWith(startingBlockNumber, endingBlockNumber, nodeId)
|
||||
mockEthereumDatabase.AssertGetBlockHashCalledWith(missingBlockNumber)
|
||||
mockEthereumDatabase.AssertGetBlockCalledWith(fakeHash, missingBlockNumber)
|
||||
})
|
||||
|
||||
It("fetches missing blocks from level db and persists them to pg", func() {
|
||||
mockEthereumDatabase := fakes.NewMockEthereumDatabase()
|
||||
mockBlockRepository := fakes.NewMockBlockRepository()
|
||||
mockReceiptRepository := fakes.NewMockReceiptRepository()
|
||||
mockTransactionConverter := fakes.NewMockTransactionConverter()
|
||||
blockConverter := vulcCommon.NewBlockConverter(mockTransactionConverter)
|
||||
|
||||
blockNumber := int64(123)
|
||||
fakeHash := []byte{1, 2, 3, 4, 5}
|
||||
mockBlockRepository.SetMissingBlockNumbersReturnArray([]int64{blockNumber})
|
||||
mockEthereumDatabase.SetReturnHash(fakeHash)
|
||||
mockEthereumDatabase.SetReturnBlock(fakeGethBlock)
|
||||
importer := cold_import.NewColdImporter(mockEthereumDatabase, mockBlockRepository, mockReceiptRepository, blockConverter)
|
||||
|
||||
importer.Execute(blockNumber, blockNumber, "node_id")
|
||||
|
||||
mockEthereumDatabase.AssertGetBlockHashCalledWith(blockNumber)
|
||||
mockEthereumDatabase.AssertGetBlockCalledWith(fakeHash, blockNumber)
|
||||
mockTransactionConverter.AssertConvertTransactionsToCoreCalledWith(fakeGethBlock)
|
||||
convertedBlock, err := blockConverter.ToCoreBlock(fakeGethBlock)
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
mockBlockRepository.AssertCreateOrUpdateBlockCalledWith(convertedBlock)
|
||||
})
|
||||
|
||||
It("sets is_final status on populated blocks", func() {
|
||||
mockEthereumDatabase := fakes.NewMockEthereumDatabase()
|
||||
mockBlockRepository := fakes.NewMockBlockRepository()
|
||||
mockReceiptRepository := fakes.NewMockReceiptRepository()
|
||||
mockTransactionConverter := fakes.NewMockTransactionConverter()
|
||||
blockConverter := vulcCommon.NewBlockConverter(mockTransactionConverter)
|
||||
|
||||
startingBlockNumber := int64(120)
|
||||
endingBlockNumber := int64(125)
|
||||
fakeHash := []byte{1, 2, 3, 4, 5}
|
||||
mockBlockRepository.SetMissingBlockNumbersReturnArray([]int64{startingBlockNumber})
|
||||
mockEthereumDatabase.SetReturnHash(fakeHash)
|
||||
mockEthereumDatabase.SetReturnBlock(fakeGethBlock)
|
||||
importer := cold_import.NewColdImporter(mockEthereumDatabase, mockBlockRepository, mockReceiptRepository, blockConverter)
|
||||
|
||||
importer.Execute(startingBlockNumber, endingBlockNumber, "node_id")
|
||||
|
||||
mockBlockRepository.AssertSetBlockStatusCalledWith(endingBlockNumber)
|
||||
})
|
||||
|
||||
It("fetches receipts from level db and persists them to pg", func() {
|
||||
mockEthereumDatabase := fakes.NewMockEthereumDatabase()
|
||||
mockBlockRepository := fakes.NewMockBlockRepository()
|
||||
mockReceiptRepository := fakes.NewMockReceiptRepository()
|
||||
mockTransactionConverter := fakes.NewMockTransactionConverter()
|
||||
blockConverter := vulcCommon.NewBlockConverter(mockTransactionConverter)
|
||||
|
||||
blockNumber := int64(123)
|
||||
blockId := int64(999)
|
||||
mockBlockRepository.SetCreateOrUpdateBlockReturnVals(blockId, nil)
|
||||
fakeReceipts := types.Receipts{{}}
|
||||
mockBlockRepository.SetMissingBlockNumbersReturnArray([]int64{blockNumber})
|
||||
mockEthereumDatabase.SetReturnBlock(fakeGethBlock)
|
||||
mockEthereumDatabase.SetReturnReceipts(fakeReceipts)
|
||||
importer := cold_import.NewColdImporter(mockEthereumDatabase, mockBlockRepository, mockReceiptRepository, blockConverter)
|
||||
|
||||
importer.Execute(blockNumber, blockNumber, "node_id")
|
||||
|
||||
expectedReceipts := vulcCommon.ToCoreReceipts(fakeReceipts)
|
||||
mockReceiptRepository.AssertCreateReceiptsAndLogsCalledWith(blockId, expectedReceipts)
|
||||
})
|
||||
|
||||
It("does not fetch receipts if block already exists", func() {
|
||||
mockEthereumDatabase := fakes.NewMockEthereumDatabase()
|
||||
mockBlockRepository := fakes.NewMockBlockRepository()
|
||||
mockReceiptRepository := fakes.NewMockReceiptRepository()
|
||||
mockTransactionConverter := fakes.NewMockTransactionConverter()
|
||||
blockConverter := vulcCommon.NewBlockConverter(mockTransactionConverter)
|
||||
|
||||
blockNumber := int64(123)
|
||||
mockBlockRepository.SetMissingBlockNumbersReturnArray([]int64{})
|
||||
mockEthereumDatabase.SetReturnBlock(fakeGethBlock)
|
||||
mockBlockRepository.SetCreateOrUpdateBlockReturnVals(0, repositories.ErrBlockExists)
|
||||
importer := cold_import.NewColdImporter(mockEthereumDatabase, mockBlockRepository, mockReceiptRepository, blockConverter)
|
||||
|
||||
err := importer.Execute(blockNumber, blockNumber, "node_id")
|
||||
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
mockReceiptRepository.AssertCreateReceiptsAndLogsNotCalled()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,68 @@
|
||||
package cold_import
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/core"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/crypto"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/fs"
|
||||
"strings"
|
||||
)
|
||||
|
||||
const (
|
||||
ColdImportClientName = "LevelDbColdImport"
|
||||
ColdImportNetworkId float64 = 1
|
||||
)
|
||||
|
||||
var (
|
||||
NoChainDataErr = errors.New("Level DB path does not include chaindata extension.")
|
||||
NoGethRootErr = errors.New("Level DB path does not include root path to geth.")
|
||||
)
|
||||
|
||||
type ColdImportNodeBuilder struct {
|
||||
reader fs.Reader
|
||||
parser crypto.PublicKeyParser
|
||||
}
|
||||
|
||||
func NewColdImportNodeBuilder(reader fs.Reader, parser crypto.PublicKeyParser) ColdImportNodeBuilder {
|
||||
return ColdImportNodeBuilder{reader: reader, parser: parser}
|
||||
}
|
||||
|
||||
func (cinb ColdImportNodeBuilder) GetNode(genesisBlock []byte, levelPath string) (core.Node, error) {
|
||||
var coldNode core.Node
|
||||
nodeKeyPath, err := getNodeKeyPath(levelPath)
|
||||
if err != nil {
|
||||
return coldNode, err
|
||||
}
|
||||
nodeKey, err := cinb.reader.Read(nodeKeyPath)
|
||||
if err != nil {
|
||||
return coldNode, err
|
||||
}
|
||||
nodeId, err := cinb.parser.ParsePublicKey(string(nodeKey))
|
||||
if err != nil {
|
||||
return coldNode, err
|
||||
}
|
||||
genesisBlockHash := common.BytesToHash(genesisBlock).String()
|
||||
coldNode = core.Node{
|
||||
GenesisBlock: genesisBlockHash,
|
||||
NetworkID: ColdImportNetworkId,
|
||||
ID: nodeId,
|
||||
ClientName: ColdImportClientName,
|
||||
}
|
||||
return coldNode, nil
|
||||
}
|
||||
|
||||
func getNodeKeyPath(levelPath string) (string, error) {
|
||||
chaindataExtension := "chaindata"
|
||||
if !strings.Contains(levelPath, chaindataExtension) {
|
||||
return "", NoChainDataErr
|
||||
}
|
||||
chaindataExtensionLength := len(chaindataExtension)
|
||||
gethRootPathLength := len(levelPath) - chaindataExtensionLength
|
||||
if gethRootPathLength <= chaindataExtensionLength {
|
||||
return "", NoGethRootErr
|
||||
}
|
||||
gethRootPath := levelPath[:gethRootPathLength]
|
||||
nodeKeyPath := gethRootPath + "nodekey"
|
||||
return nodeKeyPath, nil
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
package cold_import_test
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
. "github.com/onsi/ginkgo"
|
||||
. "github.com/onsi/gomega"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/fakes"
|
||||
"github.com/vulcanize/vulcanizedb/pkg/geth/cold_import"
|
||||
)
|
||||
|
||||
var _ = Describe("Cold importer node builder", func() {
|
||||
Describe("when level path is not valid", func() {
|
||||
It("returns error if no chaindata extension", func() {
|
||||
gethPath := "path/to/geth"
|
||||
mockReader := fakes.NewMockFsReader()
|
||||
mockParser := fakes.NewMockCryptoParser()
|
||||
nodeBuilder := cold_import.NewColdImportNodeBuilder(mockReader, mockParser)
|
||||
|
||||
_, err := nodeBuilder.GetNode([]byte{1, 2, 3, 4, 5}, gethPath)
|
||||
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(err).To(MatchError(cold_import.NoChainDataErr))
|
||||
})
|
||||
|
||||
It("returns error if no root geth path", func() {
|
||||
chaindataPath := "chaindata"
|
||||
mockReader := fakes.NewMockFsReader()
|
||||
mockParser := fakes.NewMockCryptoParser()
|
||||
nodeBuilder := cold_import.NewColdImportNodeBuilder(mockReader, mockParser)
|
||||
|
||||
_, err := nodeBuilder.GetNode([]byte{1, 2, 3, 4, 5}, chaindataPath)
|
||||
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(err).To(MatchError(cold_import.NoGethRootErr))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("when reader fails", func() {
|
||||
It("returns err", func() {
|
||||
mockReader := fakes.NewMockFsReader()
|
||||
fakeError := errors.New("Failed")
|
||||
mockReader.SetReturnErr(fakeError)
|
||||
mockParser := fakes.NewMockCryptoParser()
|
||||
nodeBuilder := cold_import.NewColdImportNodeBuilder(mockReader, mockParser)
|
||||
|
||||
_, err := nodeBuilder.GetNode([]byte{1, 2, 3, 4, 5}, "path/to/geth/chaindata")
|
||||
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(err).To(MatchError(fakeError))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("when parser fails", func() {
|
||||
It("returns err", func() {
|
||||
mockReader := fakes.NewMockFsReader()
|
||||
mockParser := fakes.NewMockCryptoParser()
|
||||
fakeErr := errors.New("Failed")
|
||||
mockParser.SetReturnErr(fakeErr)
|
||||
nodeBuilder := cold_import.NewColdImportNodeBuilder(mockReader, mockParser)
|
||||
|
||||
_, err := nodeBuilder.GetNode([]byte{1, 2, 3, 4, 5}, "path/to/geth/chaindata")
|
||||
|
||||
Expect(err).To(HaveOccurred())
|
||||
Expect(err).To(MatchError(fakeErr))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("when path is valid and reader and parser succeed", func() {
|
||||
It("builds a node", func() {
|
||||
fakeGenesisBlock := []byte{1, 2, 3, 4, 5}
|
||||
fakeRootGethPath := "root/path/to/geth/"
|
||||
fakeLevelPath := fakeRootGethPath + "chaindata"
|
||||
fakeNodeKeyPath := fakeRootGethPath + "nodekey"
|
||||
fakePublicKeyBytes := []byte{5, 4, 3, 2, 1}
|
||||
fakePublicKeyString := "public_key"
|
||||
mockReader := fakes.NewMockFsReader()
|
||||
mockReader.SetReturnBytes(fakePublicKeyBytes)
|
||||
mockParser := fakes.NewMockCryptoParser()
|
||||
mockParser.SetReturnVal(fakePublicKeyString)
|
||||
nodeBuilder := cold_import.NewColdImportNodeBuilder(mockReader, mockParser)
|
||||
|
||||
result, err := nodeBuilder.GetNode(fakeGenesisBlock, fakeLevelPath)
|
||||
|
||||
Expect(err).NotTo(HaveOccurred())
|
||||
mockReader.AssertReadCalledWith(fakeNodeKeyPath)
|
||||
mockParser.AssertParsePublicKeyCalledWith(string(fakePublicKeyBytes))
|
||||
Expect(result).NotTo(BeNil())
|
||||
Expect(result.ClientName).To(Equal(cold_import.ColdImportClientName))
|
||||
expectedGenesisBlock := common.BytesToHash(fakeGenesisBlock).String()
|
||||
Expect(result.GenesisBlock).To(Equal(expectedGenesisBlock))
|
||||
Expect(result.ID).To(Equal(fakePublicKeyString))
|
||||
Expect(result.NetworkID).To(Equal(cold_import.ColdImportNetworkId))
|
||||
})
|
||||
})
|
||||
|
||||
})
|
||||
Reference in New Issue
Block a user