Merge pull request #4 from vulcanize/snapshot

Updates

cmd/stateSnapshot.go

@@ -60,9 +60,11 @@ func stateSnapshot() {
 func init() {
 	rootCmd.AddCommand(stateSnapshotCmd)
 
-	stateSnapshotCmd.PersistentFlags().String("leveldb-path", "", "path to leveldb")
+	stateSnapshotCmd.PersistentFlags().String("leveldb-path", "", "path to primary datastore")
+	stateSnapshotCmd.PersistentFlags().String("ancient-path", "", "path to ancient datastore")
 	stateSnapshotCmd.PersistentFlags().String("block-height", "", "blockheight to extract state at")
 
 	viper.BindPFlag("leveldb.path", stateSnapshotCmd.PersistentFlags().Lookup("leveldb-path"))
+	viper.BindPFlag("leveldb.ancient", stateSnapshotCmd.PersistentFlags().Lookup("ancient-path"))
 	viper.BindPFlag("snapshot.blockHeight", stateSnapshotCmd.PersistentFlags().Lookup("block-height"))
 }

go.mod

@@ -4,6 +4,8 @@ go 1.13
 
 require (
 	github.com/ethereum/go-ethereum v1.9.11
+	github.com/ipfs/go-ipfs-blockstore v1.0.0
+	github.com/ipfs/go-ipfs-ds-help v1.0.0
 	github.com/multiformats/go-multihash v0.0.13
 	github.com/sirupsen/logrus v1.6.0
 	github.com/spf13/cobra v1.0.0

pkg/snapshot/config.go

@@ -23,17 +23,19 @@ import (
 )
 
 const (
-	LVL_DB_PATH       = "LVL_DB_PATH"
+	ANCIENT_DB_PATH   = "ANCIENT_DB_PATH"
-	ETH_NODE_ID       = "ETH_NODE_ID"
 	ETH_CLIENT_NAME   = "ETH_CLIENT_NAME"
 	ETH_GENESIS_BLOCK = "ETH_GENESIS_BLOCK"
 	ETH_NETWORK_ID    = "ETH_NETWORK_ID"
+	ETH_NODE_ID       = "ETH_NODE_ID"
+	LVL_DB_PATH       = "LVL_DB_PATH"
 )
 
 type Config struct {
-	LevelDBPath string
+	LevelDBPath   string
-	Node        core.Node
+	AncientDBPath string
-	DBConfig    config.Database
+	Node          core.Node
+	DBConfig      config.Database
 }
 
 func (c *Config) Init() {
@@ -43,6 +45,7 @@ func (c *Config) Init() {
 	viper.BindEnv("ethereum.clientName", ETH_CLIENT_NAME)
 	viper.BindEnv("ethereum.genesisBlock", ETH_GENESIS_BLOCK)
 	viper.BindEnv("ethereum.networkID", ETH_NETWORK_ID)
+	viper.BindEnv("leveldb.ancient", ANCIENT_DB_PATH)
 
 	c.Node = core.Node{
 		ID:           viper.GetString("ethereum.nodeID"),
@@ -51,4 +54,5 @@ func (c *Config) Init() {
 		NetworkID:    viper.GetString("ethereum.networkID"),
 	}
 	c.LevelDBPath = viper.GetString("leveldb.path")
+	c.AncientDBPath = viper.GetString("leveldb.ancient")
 }

pkg/snapshot/publisher.go

@@ -19,6 +19,8 @@ import (
 	"bytes"
 
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ipfs/go-ipfs-blockstore"
+	"github.com/ipfs/go-ipfs-ds-help"
 	"github.com/multiformats/go-multihash"
 
 	"github.com/vulcanize/ipfs-blockchain-watcher/pkg/ipfs/ipld"
@@ -36,6 +38,7 @@ func NewPublisher(db *postgres.DB) *Publisher {
 	}
 }
 
+// PublishHeader writes the header to the ipfs backing pg datastore and adds secondary indexes in the header_cids table
 func (p *Publisher) PublishHeader(header *types.Header) (int64, error) {
 	headerNode, err := ipld.NewEthHeader(header)
 	if err != nil {
@@ -69,6 +72,7 @@ func (p *Publisher) PublishHeader(header *types.Header) (int64, error) {
 	return headerID, err
 }
 
+// PublishStateNode writes the state node to the ipfs backing datastore and adds secondary indexes in the state_cids table
 func (p *Publisher) PublishStateNode(node Node, headerID int64) (int64, error) {
 	var stateID int64
 	var stateKey string
@@ -101,6 +105,7 @@ func (p *Publisher) PublishStateNode(node Node, headerID int64) (int64, error) {
 	return stateID, err
 }
 
+// PublishStorageNode writes the storage node to the ipfs backing pg datastore and adds secondary indexes in the storage_cids table
 func (p *Publisher) PublishStorageNode(node Node, stateID int64) error {
 	var storageKey string
 	if !bytes.Equal(node.Key.Bytes(), nullHash.Bytes()) {
@@ -130,3 +135,20 @@ func (p *Publisher) PublishStorageNode(node Node, stateID int64) error {
 		stateID, storageKey, storageCIDStr, node.Path, node.NodeType, false, mhKey)
 	return err
 }
+
+// PublishCode writes code to the ipfs backing pg datastore
+func (p *Publisher) PublishCode(code []byte) error {
+	// no codec for code, doesn't matter though since blockstore key is multihash-derived
+	return p.publishRaw(ipld.MEthStorageTrie, multihash.KECCAK_256, code)
+}
+
+func (p *Publisher) publishRaw(codec, mh uint64, raw []byte) error {
+	c, err := ipld.RawdataToCid(codec, raw, mh)
+	if err != nil {
+		return err
+	}
+	dbKey := dshelp.MultihashToDsKey(c.Hash())
+	prefixedKey := blockstore.BlockPrefix.String() + dbKey.String()
+	_, err = p.db.Exec(`INSERT INTO public.blocks (key, data) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING`, prefixedKey, raw)
+	return err
+}

pkg/snapshot/service.go

@@ -35,6 +35,7 @@ import (
 var (
 	nullHash          = common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000000")
 	emptyNode, _      = rlp.EncodeToBytes([]byte{})
+	emptyCodeHash     = crypto.Keccak256([]byte{})
 	emptyContractRoot = crypto.Keccak256Hash(emptyNode)
 )
 
@@ -49,7 +50,7 @@ func NewSnapshotService(con Config) (*Service, error) {
 	if err != nil {
 		return nil, err
 	}
-	edb, err := rawdb.NewLevelDBDatabase(con.LevelDBPath, 256, 1024, "eth-pg-ipfs-state-snapshot")
+	edb, err := rawdb.NewLevelDBDatabaseWithFreezer(con.LevelDBPath, 1024, 256, con.AncientDBPath, "eth-pg-ipfs-state-snapshot")
 	if err != nil {
 		return nil, err
 	}
@@ -136,7 +137,7 @@ func (s *Service) createSnapshot(it trie.NodeIterator, trieDB *trie.Database, he
 		}
 		switch ty {
 		case Leaf:
-			// if the node is a leaf, decode the account and if publish the associated storage trie nodes if there are any
+			// if the node is a leaf, decode the account and publish the associated storage trie nodes if there are any
 			var account state.Account
 			if err := rlp.DecodeBytes(nodeElements[1].([]byte), &account); err != nil {
 				return fmt.Errorf("error decoding account for leaf node at path %x nerror: %v", nodePath, err)
@@ -150,6 +151,16 @@ func (s *Service) createSnapshot(it trie.NodeIterator, trieDB *trie.Database, he
 			if err != nil {
 				return err
 			}
+			// publish any non-nil code referenced by codehash
+			if !bytes.Equal(account.CodeHash, emptyCodeHash) {
+				codeBytes, err := s.ethDB.Get(account.CodeHash)
+				if err != nil {
+					return err
+				}
+				if err := s.ipfsPublisher.PublishCode(codeBytes); err != nil {
+					return err
+				}
+			}
 			if err := s.storageSnapshot(account.Root, stateID); err != nil {
 				return fmt.Errorf("failed building storage snapshot for account %+v\r\nerror: %v", account, err)
 			}
@@ -162,7 +173,7 @@ func (s *Service) createSnapshot(it trie.NodeIterator, trieDB *trie.Database, he
 			return errors.New("unexpected node type")
 		}
 	}
-	return nil
+	return it.Error()
 }
 
 func (s *Service) storageSnapshot(sr common.Hash, stateID int64) error {
@@ -217,5 +228,5 @@ func (s *Service) storageSnapshot(sr common.Hash, stateID int64) error {
 			return err
 		}
 	}
-	return nil
+	return it.Error()
 }