Address comments.

pkg/snapshot/config.go

@@ -22,12 +22,12 @@ import (
 )
 
 const (
-	ancientDBPath   = "ANCIENT_DB_PATH"
+	ANCIENT_DB_PATH   = "ANCIENT_DB_PATH"
-	ethClientName   = "ETH_CLIENT_NAME"
+	ETH_CLIENT_NAME   = "ETH_CLIENT_NAME"
-	ethGenesisBlock = "ETH_GENESIS_BLOCK"
+	ETH_GENESIS_BLOCK = "ETH_GENESIS_BLOCK"
-	ethNetworkID    = "ETH_NETWORK_ID"
+	ETH_NETWORK_ID    = "ETH_NETWORK_ID"
-	ethNodeID       = "ETH_NODE_ID"
+	ETH_NODE_ID       = "ETH_NODE_ID"
-	lvlDBPath       = "LVL_DB_PATH"
+	LVL_DB_PATH       = "LVL_DB_PATH"
 )
 
 // Config is config parameters for DB.
@@ -42,12 +42,12 @@ type Config struct {
 // Init Initialises config
 func (c *Config) Init() {
 	c.dbInit()
-	viper.BindEnv("leveldb.path", lvlDBPath)
+	viper.BindEnv("leveldb.path", LVL_DB_PATH)
-	viper.BindEnv("ethereum.nodeID", ethNodeID)
+	viper.BindEnv("ethereum.nodeID", ETH_NODE_ID)
-	viper.BindEnv("ethereum.clientName", ethClientName)
+	viper.BindEnv("ethereum.clientName", ETH_CLIENT_NAME)
-	viper.BindEnv("ethereum.genesisBlock", ethGenesisBlock)
+	viper.BindEnv("ethereum.genesisBlock", ETH_GENESIS_BLOCK)
-	viper.BindEnv("ethereum.networkID", ethNetworkID)
+	viper.BindEnv("ethereum.networkID", ETH_NETWORK_ID)
-	viper.BindEnv("leveldb.ancient", ancientDBPath)
+	viper.BindEnv("leveldb.ancient", ANCIENT_DB_PATH)
 
 	c.Node = ethNode.Info{
 		ID:           viper.GetString("ethereum.nodeID"),

pkg/snapshot/publisher.go

@@ -17,10 +17,10 @@ package snapshot
 
 import (
 	"bytes"
+	"fmt"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
-	blockstore "github.com/ipfs/go-ipfs-blockstore"
-	dshelp "github.com/ipfs/go-ipfs-ds-help"
 	"github.com/jmoiron/sqlx"
 	"github.com/multiformats/go-multihash"
 
@@ -90,13 +90,6 @@ func (p *Publisher) PublishStateNode(node *node, headerID int64, tx *sqlx.Tx) (i
 		stateKey = node.key.Hex()
 	}
 
-	defer func() {
-		if rec := recover(); rec != nil {
-			shared.Rollback(tx)
-			panic(rec)
-		}
-	}()
-
 	stateCIDStr, mhKey, err := shared.PublishRaw(tx, ipld.MEthStateTrie, multihash.KECCAK_256, node.value)
 	if err != nil {
 		return 0, err
@@ -118,13 +111,6 @@ func (p *Publisher) PublishStorageNode(node *node, stateID int64, tx *sqlx.Tx) e
 		storageKey = node.key.Hex()
 	}
 
-	defer func() {
-		if rec := recover(); rec != nil {
-			shared.Rollback(tx)
-			panic(rec)
-		}
-	}()
-
 	storageCIDStr, mhKey, err := shared.PublishRaw(tx, ipld.MEthStorageTrie, multihash.KECCAK_256, node.value)
 	if err != nil {
 		return err
@@ -142,22 +128,15 @@ func (p *Publisher) PublishStorageNode(node *node, stateID int64, tx *sqlx.Tx) e
 }
 
 // PublishCode writes code to the ipfs backing pg datastore
-func (p *Publisher) PublishCode(code []byte, tx *sqlx.Tx) error {
+func (p *Publisher) PublishCode(codeHash common.Hash, codeBytes []byte, tx *sqlx.Tx) error {
 	// no codec for code, doesn't matter though since blockstore key is multihash-derived
-	return p.publishRaw(ipld.MEthStorageTrie, multihash.KECCAK_256, code, tx)
+	mhKey, err := shared.MultihashKeyFromKeccak256(codeHash)
+	if err != nil {
+		return fmt.Errorf("error deriving multihash key from codehash: %v", err)
 	}
 
-func (p *Publisher) publishRaw(codec, mh uint64, raw []byte, tx *sqlx.Tx) error {
+	if err = shared.PublishDirect(tx, mhKey, codeBytes); err != nil {
-	c, err := ipld.RawdataToCid(codec, raw, mh)
+		return fmt.Errorf("error publishing code IPLD: %v", err)
-	if err != nil {
-		return err
-	}
-
-	dbKey := dshelp.MultihashToDsKey(c.Hash())
-	prefixedKey := blockstore.BlockPrefix.String() + dbKey.String()
-	_, err = tx.Exec(`INSERT INTO public.blocks (key, data) VALUES ($1, $2) ON CONFLICT (key) DO NOTHING`, prefixedKey, raw)
-	if err != nil {
-		return err
 	}
 
 	p.currBatchSize++

pkg/snapshot/service.go

@@ -28,6 +28,7 @@ import (
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/statediff/indexer/postgres"
+	"github.com/ethereum/go-ethereum/statediff/indexer/shared"
 	"github.com/ethereum/go-ethereum/trie"
 	"github.com/jmoiron/sqlx"
 	"github.com/sirupsen/logrus"
@@ -136,7 +137,14 @@ func (s *Service) createSnapshot(it trie.NodeIterator, trieDB *trie.Database, he
 	}
 
 	defer func() {
+		if rec := recover(); rec != nil {
+			shared.Rollback(tx)
+			panic(rec)
+		} else if err != nil {
+			shared.Rollback(tx)
+		} else {
 			err = tx.Commit()
+		}
 	}()
 
 	for it.Next(true) {
@@ -203,13 +211,14 @@ func (s *Service) createSnapshot(it trie.NodeIterator, trieDB *trie.Database, he
 
 			// publish any non-nil code referenced by codehash
 			if !bytes.Equal(account.CodeHash, emptyCodeHash) {
-				codeBytes := rawdb.ReadCode(s.ethDB, common.BytesToHash(account.CodeHash))
+				codeHash := common.BytesToHash(account.CodeHash)
+				codeBytes := rawdb.ReadCode(s.ethDB, codeHash)
 				if len(codeBytes) == 0 {
 					logrus.Error("Code is missing", "account", common.BytesToHash(it.LeafKey()))
 					return errors.New("missing code")
 				}
 
-				if err = s.ipfsPublisher.PublishCode(codeBytes, tx); err != nil {
+				if err = s.ipfsPublisher.PublishCode(codeHash, codeBytes, tx); err != nil {
 					return err
 				}
 			}