Statediffing geth

* Write state diff to CSV (#2)

* port statediff from https://github.com/jpmorganchase/quorum/blob/9b7fd9af8082795eeeb6863d9746f12b82dd5078/statediff/statediff.go; minor fixes

* integrating state diff extracting, building, and persisting into geth processes

* work towards persisting created statediffs in ipfs; based off github.com/vulcanize/eth-block-extractor

* Add a state diff service

* Remove diff extractor from blockchain

* Update imports

* Move statediff on/off check to geth cmd config

* Update starting state diff service

* Add debugging logs for creating diff

* Add statediff extractor and builder tests and small refactoring

* Start to write statediff to a CSV

* Restructure statediff directory

* Pull CSV publishing methods into their own file

* Reformatting due to go fmt

* Add gomega to vendor dir

* Remove testing focuses

* Update statediff tests to use golang test pkg

instead of ginkgo

- builder_test
- extractor_test
- publisher_test

* Use hexutil.Encode instead of deprecated common.ToHex

* Remove OldValue from DiffBigInt and DiffUint64 fields

* Update builder test

* Remove old storage value from updated accounts

* Remove old values from created/deleted accounts

* Update publisher to account for only storing current account values

* Update service loop and fetching previous block

* Update testing

- remove statediff ginkgo test suite file
- move mocks to their own dir

* Updates per go fmt

* Updates to tests

* Pass statediff mode and path in through cli

* Return filename from publisher

* Remove some duplication in builder

* Remove code field from state diff output

this is the contract byte code, and it can still be obtained by querying
the db by the codeHash

* Consolidate acct diff structs for updated & updated/deleted accts

* Include block number in csv filename

* Clean up error logging

* Cleanup formatting, spelling, etc

* Address PR comments

* Add contract address and storage value to csv

* Refactor accumulating account row in csv publisher

* Add DiffStorage struct

* Add storage key to csv

* Address PR comments

* Fix publisher to include rows for accounts that don't have store updates

* Update builder test after merging in release/1.8

* Update test contract to include storage on contract intialization

- so that we're able to test that storage diffing works for created and
deleted accounts (not just updated accounts).

* Factor out a common trie iterator method in builder

* Apply goimports to statediff

* Apply gosimple changes to statediff

* Gracefully exit geth command(#4)

* Statediff for full node (#6)

* Open a trie from the in-memory database

* Use a node's LeafKey as an identifier instead of the address

It was proving difficult to find look the address up from a given path
with a full node (sometimes the value wouldn't exist in the disk db).
So, instead, for now we are using the node's LeafKey with is a Keccak256
hash of the address, so if we know the address we can figure out which
LeafKey it matches up to.

* Make sure that statediff has been processed before pruning

* Use blockchain stateCache.OpenTrie for storage diffs

* Clean up log lines and remove unnecessary fields from builder

* Apply go fmt changes

* Add a sleep to the blockchain test

* Address PR comments

* Address PR comments

* refactoring/reorganizing packages

* refactoring statediff builder and types and adjusted to relay proofs and paths (still need to make this optional)

* refactoring state diff service and adding api which allows for streaming state diff payloads over an rpc websocket subscription

* make proofs and paths optional + compress service loop into single for loop (may be missing something here)

* option to process intermediate nodes

* make state diff rlp serializable

* cli parameter to limit statediffing to select account addresses + test

* review fixes and fixes for issues ran into in integration

* review fixes; proper method signature for api; adjust service so that statediff processing is halted/paused until there is at least one subscriber listening for the results

* adjust buffering to improve stability; doc.go; fix notifier
err handling

* relay receipts with the rest of the data + review fixes/changes

* rpc method to get statediff at specific block; requires archival node or the block be within the pruning range

* review fixes

* fixes after rebase

* statediff verison meta

* fix linter issues

* include total difficulty to the payload

* fix state diff builder: emit actual leaf nodes instead of value nodes; diff on the leaf not on the value; emit correct path for intermediate nodes

* adjust statediff builder tests to changes and extend to test intermediate nodes; golint

* add genesis block to test; handle block 0 in StateDiffAt

* rlp files for mainnet blocks 0-3, for tests

* builder test on mainnet blocks

* common.BytesToHash(path) => crypto.Keaccak256(hash) in builder; BytesToHash produces same hash for e.g. []byte{} and []byte{\x00} - prefix \x00 steps are inconsequential to the hash result

* complete tests for early mainnet blocks

* diff type for representing deleted accounts

* fix builder so that we handle account deletions properly and properly diff storage when an account is moved to a new path; update params

* remove cli params; moving them to subscriber defined

* remove unneeded bc methods

* update service and api; statediffing params are now defined by user through api rather than by service provider by cli

* update top level tests

* add ability to watch specific storage slots (leaf keys) only

* comments; explain logic

* update mainnet blocks test

* update api_test.go

* storage leafkey filter test

* cleanup chain maker

* adjust chain maker for tests to add an empty account in block1 and switch to EIP-158 afterwards (now we just need to generate enough accounts until one causes the empty account to be touched and removed post-EIP-158 so we can simulate and test that process...); also added 2 new blocks where more contract storage is set and old slots are set to zero so they are removed so we can test that

* found an account whose creation causes the empty account to be moved to a new path; this should count as 'touching; the empty account and cause it to be removed according to eip-158... but it doesn't

* use new contract in unit tests that has self-destruct ability, so we can test eip-158 since simply moving an account to new path doesn't count as 'touchin' it

* handle storage deletions

* tests for eip-158 account removal and storage value deletions; there is one edge case left to test where we remove 1 account when only two exist such that the remaining account is moved up and replaces the root branch node

* finish testing known edge cases

* add endpoint to fetch all state and storage nodes at a given blockheight; useful for generating a recent atate cache/snapshot that we can diff forward from rather than needing to collect all diffs from genesis

* test for state trie builder

* minor changes/fixes

* update version meta

* if statediffing is on, lock tries in triedb until the statediffing service signals they are done using them

* update version meta

* fix mock blockchain; golint; bump patch

* increase maxRequestContentLength; bump patch

* log the sizes of the state objects we are sending

* CI build (#20)

* CI: run build on PR and on push to master

* CI: debug building geth

* CI: fix coping file

* CI: fix coping file v2

* CI: temporary upload file to release asset

* CI: get release upload_url by tag, upload asset to current relase

* CI: fix tag name

* fix ci build on statediff_at_anyblock-1.9.11 branch

* fix publishing assets in release

* bump version meta

* use context deadline for timeout in eth_call

* collect and emit codehash=>code mappings for state objects

* subscription endpoint for retrieving all the codehash=>code mappings that exist at provided height

* bump version meta

* Implement WriteStateDiffAt

* Writes state diffs directly to postgres

* Adds CLI flags to configure PG

* Refactors builder output with callbacks

* Copies refactored postgres handling code from ipld-eth-indexer

* rename PostgresCIDWriter.{index->upsert}*

* less ambiguous

* go.mod update

* rm unused

* cleanup

* output code & codehash iteratively

* had to rf some types for this

* prometheus metrics output

* duplicate recent eth-indexer changes

* migrations and metrics...

* [wip] prom.Init() here? another CLI flag?

* cleanup

* tidy & DRY

* statediff WriteLoop service + CLI flag

* [wip] update test mocks

* todo - do something meaningful to test write loop

* logging

* use geth log

* port tests to go testing

* drop ginkgo/gomega

* fix and cleanup tests

* fail before defer statement

* delete vendor/ dir

* unused

* bump version meta

Author: elizabethengelman

.github/workflows/build.yml

@@ -0,0 +1,12 @@
+name: Docker Build
+
+on: [pull_request]
+
+jobs:
+  build:
+    name: Run docker build
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Run docker build
+        run: docker build -t vulcanize/go-ethereum -f Dockerfile.amd64 .

.github/workflows/on-master.yaml

@@ -0,0 +1,25 @@
+name: Docker Build and publish to Github
+
+on:
+  push:
+    branches:
+      - statediff_at_anyblock-1.9.11
+
+jobs:
+  build:
+    name: Run docker build and publish
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Run docker build
+        run: docker build -t vulcanize/go-ethereum -f Dockerfile.amd64 .
+      - name: Get the version
+        id: vars
+        run: echo ::set-output name=sha::$(echo ${GITHUB_SHA:0:7})
+      - name: Tag docker image
+        run: docker tag vulcanize/go-ethereum docker.pkg.github.com/vulcanize/go-ethereum/go-ethereum:${{steps.vars.outputs.sha}}
+      - name: Docker Login
+        run: echo ${{ secrets.GITHUB_TOKEN }} | docker login https://docker.pkg.github.com -u vulcanize --password-stdin
+      - name: Docker Push
+        run: docker push docker.pkg.github.com/vulcanize/go-ethereum/go-ethereum:${{steps.vars.outputs.sha}}
+

.github/workflows/publish.yaml

@@ -0,0 +1,34 @@
+name: Publish geth to release
+on:
+  release:
+    types: [published]
+jobs:
+  push_to_registries:
+    name: Publish assets to Release
+    runs-on: ubuntu-latest
+    steps:
+      - name: Get the version
+        id: vars
+        run: |
+          echo ::set-output name=sha::$(echo ${GITHUB_SHA:0:7})
+      - name: Docker Login to Github Registry
+        run: echo ${{ secrets.GITHUB_TOKEN }} | docker login https://docker.pkg.github.com -u vulcanize --password-stdin
+      - name: Docker Pull
+        run: docker pull docker.pkg.github.com/vulcanize/go-ethereum/go-ethereum:${{steps.vars.outputs.sha}}
+      - name: Copy ethereum binary file
+        run: docker run --rm --entrypoint cat docker.pkg.github.com/vulcanize/go-ethereum/go-ethereum:${{steps.vars.outputs.sha}} /go-ethereum/build/bin/geth > geth-linux-amd64
+      - name: Get release
+        id: get_release
+        uses: bruceadams/[email protected]
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      - name: Upload Release Asset
+        id: upload-release-asset
+        uses: actions/upload-release-asset@v1
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          upload_url: ${{ steps.get_release.outputs.upload_url }}
+          asset_path: geth-linux-amd64
+          asset_name: geth-linux-amd64
+          asset_content_type: application/octet-stream

Dockerfile.amd64

@@ -0,0 +1,7 @@
+# Build Geth in a stock Go builder container
+FROM golang:1.13 as builder
+
+#RUN apk add --no-cache make gcc musl-dev linux-headers git
+
+ADD . /go-ethereum
+RUN cd /go-ethereum && make geth

cmd/geth/config.go

@@ -145,6 +145,9 @@ func makeConfigNode(ctx *cli.Context) (*node.Node, gethConfig) {
 		cfg.Ethstats.URL = ctx.GlobalString(utils.EthStatsURLFlag.Name)
 	}
 	utils.SetShhConfig(ctx, stack)
+	if ctx.GlobalBool(utils.StateDiffFlag.Name) {
+		cfg.Eth.Diffing = true
+	}
 
 	return stack, cfg
 }
@@ -165,6 +168,26 @@ func makeFullNode(ctx *cli.Context) (*node.Node, ethapi.Backend) {
 	backend := utils.RegisterEthService(stack, &cfg.Eth)
 
 	checkWhisper(ctx)
+
+	if ctx.GlobalBool(utils.StateDiffFlag.Name) {
+		var dbParams *[3]string
+		if ctx.GlobalIsSet(utils.StateDiffDBFlag.Name) {
+			dbParams = new([3]string)
+			dbParams[0] = ctx.GlobalString(utils.StateDiffDBFlag.Name)
+			if ctx.GlobalIsSet(utils.StateDiffDBNodeIDFlag.Name) {
+				dbParams[1] = ctx.GlobalString(utils.StateDiffDBNodeIDFlag.Name)
+			} else {
+				utils.Fatalf("Must specify node ID for statediff DB output")
+			}
+			if ctx.GlobalIsSet(utils.StateDiffDBClientNameFlag.Name) {
+				dbParams[2] = ctx.GlobalString(utils.StateDiffDBClientNameFlag.Name)
+			} else {
+				utils.Fatalf("Must specify client name for statediff DB output")
+			}
+		}
+		utils.RegisterStateDiffService(stack, dbParams, ctx.GlobalBool(utils.StateDiffWritingFlag.Name))
+	}
+
 	// Configure GraphQL if requested
 	if ctx.GlobalIsSet(utils.GraphQLEnabledFlag.Name) {
 		utils.RegisterGraphQLService(stack, backend, cfg.Node)

cmd/geth/main.go

@@ -157,6 +157,11 @@ var (
 		utils.GpoMaxGasPriceFlag,
 		utils.EWASMInterpreterFlag,
 		utils.EVMInterpreterFlag,
+		utils.StateDiffFlag,
+		utils.StateDiffDBFlag,
+		utils.StateDiffDBNodeIDFlag,
+		utils.StateDiffDBClientNameFlag,
+		utils.StateDiffWritingFlag,
 		configFileFlag,
 	}
 

cmd/geth/usage.go

@@ -234,6 +234,16 @@ var AppHelpFlagGroups = []flags.FlagGroup{
 			utils.LegacyGraphQLPortFlag,
 		}, debug.DeprecatedFlags...),
 	},
+	{
+		Name: "STATE DIFF",
+		Flags: []cli.Flag{
+			utils.StateDiffFlag,
+			utils.StateDiffDBFlag,
+			utils.StateDiffDBNodeIDFlag,
+			utils.StateDiffDBClientNameFlag,
+			utils.StateDiffWritingFlag,
+		},
+	},
 	{
 		Name: "MISC",
 		Flags: []cli.Flag{

cmd/utils/flags.go

@@ -31,6 +31,8 @@ import (
 	"text/template"
 	"time"
 
+	cli "gopkg.in/urfave/cli.v1"
+
 	"github.com/ethereum/go-ethereum/accounts"
 	"github.com/ethereum/go-ethereum/accounts/keystore"
 	"github.com/ethereum/go-ethereum/common"
@@ -63,8 +65,9 @@ import (
 	"github.com/ethereum/go-ethereum/p2p/nat"
 	"github.com/ethereum/go-ethereum/p2p/netutil"
 	"github.com/ethereum/go-ethereum/params"
+	"github.com/ethereum/go-ethereum/statediff"
+
 	pcsclite "github.com/gballet/go-libpcsclite"
-	cli "gopkg.in/urfave/cli.v1"
 )
 
 func init() {
@@ -722,6 +725,27 @@ var (
 		Usage: "External EVM configuration (default = built-in interpreter)",
 		Value: "",
 	}
+
+	StateDiffFlag = cli.BoolFlag{
+		Name:  "statediff",
+		Usage: "Enables the processing of state diffs between each block",
+	}
+	StateDiffDBFlag = cli.StringFlag{
+		Name:  "statediff.db",
+		Usage: "PostgreSQL database connection string for writing state diffs",
+	}
+	StateDiffDBNodeIDFlag = cli.StringFlag{
+		Name:  "statediff.dbnodeid",
+		Usage: "Node ID to use when writing state diffs to database",
+	}
+	StateDiffDBClientNameFlag = cli.StringFlag{
+		Name:  "statediff.dbclientname",
+		Usage: "Client name to use when writing state diffs to database",
+	}
+	StateDiffWritingFlag = cli.BoolFlag{
+		Name:  "statediff.writing",
+		Usage: "Activates progressive writing of state diffs to database as new block are synced",
+	}
 )
 
 // MakeDataDir retrieves the currently requested data directory, terminating
@@ -987,6 +1011,9 @@ func setWS(ctx *cli.Context, cfg *node.Config) {
 	if ctx.GlobalIsSet(WSApiFlag.Name) {
 		cfg.WSModules = SplitAndTrim(ctx.GlobalString(WSApiFlag.Name))
 	}
+	if ctx.GlobalBool(StateDiffFlag.Name) {
+		cfg.WSModules = append(cfg.WSModules, "statediff")
+	}
 }
 
 // setIPC creates an IPC path configuration from the set command line flags,
@@ -1715,6 +1742,21 @@ func RegisterGraphQLService(stack *node.Node, backend ethapi.Backend, cfg node.C
 	}
 }
 
+// RegisterStateDiffService configures and registers a service to stream state diff data over RPC
+// dbParams are: Postgres connection URI, Node ID, client name
+func RegisterStateDiffService(stack *node.Node, dbParams *[3]string, startWriteLoop bool) {
+	if err := stack.Register(func(ctx *node.ServiceContext) (node.Service, error) {
+		var ethServ *eth.Ethereum
+		err := ctx.Service(&ethServ)
+		if err != nil {
+			return nil, err
+		}
+		return statediff.NewStateDiffService(ethServ, dbParams, startWriteLoop)
+	}); err != nil {
+		Fatalf("Failed to register State Diff Service", err)
+	}
+}
+
 func SetupMetrics(ctx *cli.Context) {
 	if metrics.Enabled {
 		log.Info("Enabling metrics collection")

core/blockchain.go

@@ -131,6 +131,7 @@ type CacheConfig struct {
 	SnapshotLimit       int           // Memory allowance (MB) to use for caching snapshot entries in memory
 
 	SnapshotWait bool // Wait for snapshot construction on startup. TODO(karalabe): This is a dirty hack for testing, nuke it
+	StateDiffing        bool          // Whether or not the statediffing service is running
 }
 
 // defaultCacheConfig are the default caching values if none are specified by the
@@ -210,6 +211,10 @@ type BlockChain struct {
 	badBlocks       *lru.Cache                     // Bad block cache
 	shouldPreserve  func(*types.Block) bool        // Function used to determine whether should preserve the given block.
 	terminateInsert func(common.Hash, uint64) bool // Testing hook used to terminate ancient receipt chain insertion.
+
+	// Locked roots and their mutex
+	trieLock    sync.Mutex
+	lockedRoots map[common.Hash]bool
 }
 
 // NewBlockChain returns a fully initialised block chain using information
@@ -226,7 +231,6 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
 	txLookupCache, _ := lru.New(txLookupCacheLimit)
 	futureBlocks, _ := lru.New(maxFutureBlocks)
 	badBlocks, _ := lru.New(badBlockLimit)
-
 	bc := &BlockChain{
 		chainConfig:    chainConfig,
 		cacheConfig:    cacheConfig,
@@ -244,6 +248,7 @@ func NewBlockChain(db ethdb.Database, cacheConfig *CacheConfig, chainConfig *par
 		engine:         engine,
 		vmConfig:       vmConfig,
 		badBlocks:      badBlocks,
+		lockedRoots:    make(map[common.Hash]bool),
 	}
 	bc.validator = NewBlockValidator(chainConfig, bc, engine)
 	bc.prefetcher = newStatePrefetcher(chainConfig, bc, engine)
@@ -967,7 +972,10 @@ func (bc *BlockChain) Stop() {
 			}
 		}
 		for !bc.triegc.Empty() {
-			triedb.Dereference(bc.triegc.PopItem().(common.Hash))
+			pruneRoot := bc.triegc.PopItem().(common.Hash)
+			if !bc.TrieLocked(pruneRoot) {
+				triedb.Dereference(pruneRoot)
+			}
 		}
 		if size, _ := triedb.Size(); size != 0 {
 			log.Error("Dangling trie nodes after full cleanup")
@@ -1473,6 +1481,11 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
 		triedb.Reference(root, common.Hash{}) // metadata reference to keep trie alive
 		bc.triegc.Push(root, -int64(block.NumberU64()))
 
+		// If we are statediffing, lock the trie until the statediffing service is done using it
+		if bc.cacheConfig.StateDiffing {
+			bc.LockTrie(root)
+		}
+
 		if current := block.NumberU64(); current > TriesInMemory {
 			// If we exceeded our memory allowance, flush matured singleton nodes to disk
 			var (
@@ -1511,7 +1524,11 @@ func (bc *BlockChain) writeBlockWithState(block *types.Block, receipts []*types.
 					bc.triegc.Push(root, number)
 					break
 				}
-				triedb.Dereference(root.(common.Hash))
+				pruneRoot := root.(common.Hash)
+				if !bc.TrieLocked(pruneRoot) {
+					log.Debug("Dereferencing", "root", root.(common.Hash).Hex())
+					triedb.Dereference(pruneRoot)
+				}
 			}
 		}
 	}
@@ -2481,3 +2498,28 @@ func (bc *BlockChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscript
 func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription {
 	return bc.scope.Track(bc.blockProcFeed.Subscribe(ch))
 }
+
+// TrieLocked returns whether the trie associated with the provided root is locked for use
+func (bc *BlockChain) TrieLocked(root common.Hash) bool {
+	bc.trieLock.Lock()
+	locked, ok := bc.lockedRoots[root]
+	bc.trieLock.Unlock()
+	if !ok {
+		return false
+	}
+	return locked
+}
+
+// LockTrie prevents dereferencing of the provided root
+func (bc *BlockChain) LockTrie(root common.Hash) {
+	bc.trieLock.Lock()
+	bc.lockedRoots[root] = true
+	bc.trieLock.Unlock()
+}
+
+// UnlockTrie allows dereferencing of the provided root- provided it was previously locked
+func (bc *BlockChain) UnlockTrie(root common.Hash) {
+	bc.trieLock.Lock()
+	bc.lockedRoots[root] = false
+	bc.trieLock.Unlock()
+}

eth/backend.go

@@ -169,6 +169,7 @@ func New(stack *node.Node, config *Config) (*Ethereum, error) {
 			TrieDirtyDisabled:   config.NoPruning,
 			TrieTimeLimit:       config.TrieTimeout,
 			SnapshotLimit:       config.SnapshotCache,
+			StateDiffing:        config.Diffing,
 		}
 	)
 	eth.blockchain, err = core.NewBlockChain(chainDb, cacheConfig, chainConfig, eth.engine, vmConfig, eth.shouldPreserve, &config.TxLookupLimit)