Merge pull request #3 from vulcanize/snapshot

Updates

cmd/stateSnapshot.go

@@ -43,9 +43,16 @@ func stateSnapshot() {
 	if err != nil {
 		logWithCommand.Fatal(err)
 	}
-	height := uint64(viper.GetInt64("snapshot.blockHeight"))
+	height := viper.GetInt64("snapshot.blockHeight")
-	if err := snapshotService.CreateSnapshot(height); err != nil {
+	if height < 0 {
-		logWithCommand.Fatal(err)
+		if err := snapshotService.CreateLatestSnapshot(); err != nil {
+			logWithCommand.Fatal(err)
+		}
+	} else {
+		uHeight := uint64(height)
+		if err := snapshotService.CreateSnapshot(uHeight); err != nil {
+			logWithCommand.Fatal(err)
+		}
 	}
 	logWithCommand.Infof("state snapshot at height %d is complete", height)
 }

db/migrations/00001_create_ipfs_blocks_table.sql

@@ -0,0 +1,8 @@
+-- +goose Up
+CREATE TABLE IF NOT EXISTS public.blocks (
+  key TEXT UNIQUE NOT NULL,
+  data BYTEA NOT NULL
+);
+
+-- +goose Down
+DROP TABLE public.blocks;

db/migrations/00002_create_nodes_table.sql

@@ -0,0 +1,12 @@
+-- +goose Up
+CREATE TABLE nodes (
+  id            SERIAL PRIMARY KEY,
+  client_name   VARCHAR,
+  genesis_block VARCHAR(66),
+  network_id    VARCHAR,
+  node_id       VARCHAR(128),
+  CONSTRAINT node_uc UNIQUE (genesis_block, network_id, node_id)
+);
+
+-- +goose Down
+DROP TABLE nodes;

db/migrations/00003_create_eth_schema.sql

@@ -0,0 +1,5 @@
+-- +goose Up
+CREATE SCHEMA eth;
+
+-- +goose Down
+DROP SCHEMA eth;

db/migrations/00004_create_eth_header_cids_table.sql

@@ -0,0 +1,23 @@
+-- +goose Up
+CREATE TABLE eth.header_cids (
+  id                    SERIAL PRIMARY KEY,
+  block_number          BIGINT NOT NULL,
+  block_hash            VARCHAR(66) NOT NULL,
+  parent_hash           VARCHAR(66) NOT NULL,
+  cid                   TEXT NOT NULL,
+  mh_key                TEXT NOT NULL REFERENCES public.blocks (key) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  td                    NUMERIC NOT NULL,
+  node_id               INTEGER NOT NULL REFERENCES nodes (id) ON DELETE CASCADE,
+  reward                NUMERIC NOT NULL,
+  state_root            VARCHAR(66) NOT NULL,
+  tx_root               VARCHAR(66) NOT NULL,
+  receipt_root          VARCHAR(66) NOT NULL,
+  uncle_root            VARCHAR(66) NOT NULL,
+  bloom                 BYTEA NOT NULL,
+  timestamp             NUMERIC NOT NULL,
+  times_validated       INTEGER NOT NULL DEFAULT 1,
+  UNIQUE (block_number, block_hash)
+);
+
+-- +goose Down
+DROP TABLE eth.header_cids;

db/migrations/00005_create_eth_uncle_cids_table.sql

@@ -0,0 +1,14 @@
+-- +goose Up
+CREATE TABLE eth.uncle_cids (
+  id                    SERIAL PRIMARY KEY,
+  header_id             INTEGER NOT NULL REFERENCES eth.header_cids (id) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  block_hash            VARCHAR(66) NOT NULL,
+  parent_hash           VARCHAR(66) NOT NULL,
+  cid                   TEXT NOT NULL,
+  mh_key                TEXT NOT NULL REFERENCES public.blocks (key) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  reward                NUMERIC NOT NULL,
+  UNIQUE (header_id, block_hash)
+);
+
+-- +goose Down
+DROP TABLE eth.uncle_cids;

db/migrations/00006_create_eth_transaction_cids_table.sql

@@ -0,0 +1,15 @@
+-- +goose Up
+CREATE TABLE eth.transaction_cids (
+  id                    SERIAL PRIMARY KEY,
+  header_id             INTEGER NOT NULL REFERENCES eth.header_cids (id) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  tx_hash               VARCHAR(66) NOT NULL,
+  index                 INTEGER NOT NULL,
+  cid                   TEXT NOT NULL,
+  mh_key                TEXT NOT NULL REFERENCES public.blocks (key) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  dst                   VARCHAR(66) NOT NULL,
+  src                   VARCHAR(66) NOT NULL,
+  UNIQUE (header_id, tx_hash)
+);
+
+-- +goose Down
+DROP TABLE eth.transaction_cids;

db/migrations/00007_create_eth_receipt_cids_table.sql

@@ -0,0 +1,18 @@
+-- +goose Up
+CREATE TABLE eth.receipt_cids (
+  id                    SERIAL PRIMARY KEY,
+  tx_id                 INTEGER NOT NULL REFERENCES eth.transaction_cids (id) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  cid                   TEXT NOT NULL,
+  mh_key                TEXT NOT NULL REFERENCES public.blocks (key) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  contract              VARCHAR(66),
+  contract_hash         VARCHAR(66),
+  topic0s               VARCHAR(66)[],
+  topic1s               VARCHAR(66)[],
+  topic2s               VARCHAR(66)[],
+  topic3s               VARCHAR(66)[],
+  log_contracts         VARCHAR(66)[],
+  UNIQUE (tx_id)
+);
+
+-- +goose Down
+DROP TABLE eth.receipt_cids;

db/migrations/00008_create_eth_state_cids_table.sql

@@ -0,0 +1,15 @@
+-- +goose Up
+CREATE TABLE eth.state_cids (
+  id                    SERIAL PRIMARY KEY,
+  header_id             INTEGER NOT NULL REFERENCES eth.header_cids (id) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  state_leaf_key        VARCHAR(66),
+  cid                   TEXT NOT NULL,
+  mh_key                TEXT NOT NULL REFERENCES public.blocks (key) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  state_path            BYTEA,
+  node_type             INTEGER,
+  diff                  BOOLEAN NOT NULL DEFAULT FALSE,
+  UNIQUE (header_id, state_path, diff)
+);
+
+-- +goose Down
+DROP TABLE eth.state_cids;

db/migrations/00009_create_eth_storage_cids_table.sql

@@ -0,0 +1,15 @@
+-- +goose Up
+CREATE TABLE eth.storage_cids (
+  id                    SERIAL PRIMARY KEY,
+  state_id              INTEGER NOT NULL REFERENCES eth.state_cids (id) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  storage_leaf_key      VARCHAR(66),
+  cid                   TEXT NOT NULL,
+  mh_key                TEXT NOT NULL REFERENCES public.blocks (key) ON DELETE CASCADE DEFERRABLE INITIALLY DEFERRED,
+  storage_path          BYTEA,
+  node_type             INTEGER NOT NULL,
+  diff                  BOOLEAN NOT NULL DEFAULT FALSE,
+  UNIQUE (state_id, storage_path, diff)
+);
+
+-- +goose Down
+DROP TABLE eth.storage_cids;

db/migrations/00010_create_eth_state_accouts_table.sql

@@ -0,0 +1,13 @@
+-- +goose Up
+CREATE TABLE eth.state_accounts (
+  id                    SERIAL PRIMARY KEY,
+  state_id              INTEGER NOT NULL REFERENCES eth.state_cids (id) ON DELETE CASCADE,
+  balance               NUMERIC NOT NULL,
+  nonce                 INTEGER NOT NULL,
+  code_hash             BYTEA NOT NULL,
+  storage_root          VARCHAR(66) NOT NULL,
+  UNIQUE (state_id)
+);
+
+-- +goose Down
+DROP TABLE eth.state_accounts;

pkg/snapshot/publisher.go

@@ -126,7 +126,7 @@ func (p *Publisher) PublishStorageNode(node Node, stateID int64) error {
 	}
 	mhKey, _ := shared.MultihashKeyFromCIDString(storageCIDStr)
 	_, 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)`,
+ 							  ON CONFLICT (state_id, storage_path, diff) DO UPDATE SET (storage_leaf_key, cid, node_type, mh_key) = ($2, $3, $5, $7)`,
 		stateID, storageKey, storageCIDStr, node.Path, node.NodeType, false, mhKey)
 	return err
 }

pkg/snapshot/service.go

@@ -20,8 +20,6 @@ import (
 	"errors"
 	"fmt"
 
-	"github.com/sirupsen/logrus"
-
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/state"
@@ -29,6 +27,7 @@ import (
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/ethereum/go-ethereum/trie"
+	"github.com/sirupsen/logrus"
 
 	"github.com/vulcanize/ipfs-blockchain-watcher/pkg/postgres"
 )
@@ -61,12 +60,41 @@ func NewSnapshotService(con Config) (*Service, error) {
 	}, nil
 }
 
+func (s *Service) CreateLatestSnapshot() error {
+	// extract header from lvldb and publish to PG-IPFS
+	// hold onto the headerID so that we can link the state nodes to this header
+	logrus.Info("Creating snapshot at head")
+	hash := rawdb.ReadHeadHeaderHash(s.ethDB)
+	height := rawdb.ReadHeaderNumber(s.ethDB, hash)
+	if height == nil {
+		return fmt.Errorf("unable to read header height for header hash %s", hash.String())
+	}
+	header := rawdb.ReadHeader(s.ethDB, hash, *height)
+	if header == nil {
+		return fmt.Errorf("unable to read canonical header at height %d", height)
+	}
+	logrus.Infof("head hash: %s head height: %d", hash.Hex(), *height)
+	headerID, err := s.ipfsPublisher.PublishHeader(header)
+	if err != nil {
+		return err
+	}
+	t, err := s.stateDB.OpenTrie(header.Root)
+	if err != nil {
+		return err
+	}
+	trieDB := s.stateDB.TrieDB()
+	return s.createSnapshot(t.NodeIterator([]byte{}), trieDB, headerID)
+}
+
 func (s *Service) CreateSnapshot(height uint64) error {
 	// extract header from lvldb and publish to PG-IPFS
 	// hold onto the headerID so that we can link the state nodes to this header
 	logrus.Infof("Creating snapshot at height %d", height)
 	hash := rawdb.ReadCanonicalHash(s.ethDB, height)
 	header := rawdb.ReadHeader(s.ethDB, hash, height)
+	if header == nil {
+		return fmt.Errorf("unable to read canonical header at height %d", height)
+	}
 	headerID, err := s.ipfsPublisher.PublishHeader(header)
 	if err != nil {
 		return err