Compare commits

..
81 Commits
Author SHA1 Message Date
Ian Norden 0c2a9bc171 Merge pull request #67 from vulcanize/rct_size_bug
bump go-ethereum version
2021-12-01 22:47:37 -06:00
i-norden f35c5cf71e bump go-ethereum version 2021-12-01 22:46:45 -06:00
Ian Norden fe94c5a33c Merge pull request #66 from vulcanize/log_processing_fix
bump go-ethereum version
2021-12-01 16:01:29 -06:00
i-norden 55a20ac15f bump go-ethereum version 2021-12-01 16:00:53 -06:00
Ian Norden eee4330bf4 Merge pull request #65 from vulcanize/log_processing_fix
bump go-etheruem version
2021-11-30 18:21:12 -06:00
i-norden 5543a52472 bump go-etheruem version 2021-11-30 17:56:04 -06:00
Ian Norden db786ae3b0 Merge pull request #62 from vulcanize/schema_updates
Schema updates
2021-11-26 09:33:27 -06:00
i-norden d8a3177a20 bump statediffing geth version 2021-11-26 09:29:22 -06:00
i-norden fb54640823 add some missing ENVs 2021-11-26 09:25:01 -06:00
Ian Norden 731c986d75 Merge pull request #61 from vulcanize/schema_updates
fix cli
2021-11-22 13:56:01 -06:00
i-norden 30b481db24 fix cli 2021-11-22 13:55:29 -06:00
Ian Norden dcdfb9a05a Merge pull request #60 from vulcanize/schema_updates
update to new version of go-ethereum; minor README update
2021-11-20 14:50:19 -06:00
i-norden d78a503ba4 update to new version of go-ethereum; minor README update 2021-11-20 14:49:36 -06:00
Ian Norden ddbb690ee6 Merge pull request #59 from vulcanize/schema_updates
reflect changes in integrated service
2021-11-18 19:25:36 -06:00
i-norden bb5d5322d2 update README 2021-11-18 19:18:12 -06:00
i-norden 108a6f89f1 update to use new version of indexer; support file writing mode 2021-11-18 18:47:28 -06:00
i-norden d9eb68b2d2 update to use refactored statediffing geth 2021-11-12 18:49:24 -06:00
i-norden c5c230a2ea update to use new version of statediffing geth pkgs with public.blocks insert batching 2021-11-08 19:08:36 -06:00
i-norden dcf8afb96c bump Go version to 1.16 2021-11-08 19:05:35 -06:00
Ian Norden 212bf4c4b2 Merge pull request #52 from vulcanize/add-grafana-dashboard
Add grafana dashboard.
2021-11-04 10:09:21 -05:00
Arijit Das a4a3ec1d7a Add grafana dashboard. 2021-11-01 19:39:00 +05:30
Ian Norden efb334aef3 Merge pull request #51 from vulcanize/prerun
generate rand node id if none is provided
2021-10-28 10:52:35 -05:00
i-norden c83e8835ab generate rand node id if none is provided 2021-10-28 10:21:10 -05:00
S. R. Wadleigh 728642afec Merge pull request #48 from vulcanize/make_startup_script_executable
make startup_script.sh executable
2021-10-28 03:22:51 +00:00
Ian Norden 5875d1ec82 Update Dockerfile 2021-10-27 19:57:19 -05:00
erikdies 8d6919ab2b make startup_script.sh executable 2021-10-27 20:46:25 -04:00
Ian Norden abf72725ee Merge pull request #47 from vulcanize/prerun
cleanup
2021-10-27 17:46:13 -05:00
i-norden 8a55469403 cleanup 2021-10-27 17:45:47 -05:00
Ian Norden a3857bffe4 Merge pull request #46 from vulcanize/prerun
enable configuration of a prerun range by env variable; prerun only mode
2021-10-27 14:54:40 -05:00
i-norden 3956081874 sanitized prod config 2021-10-27 14:44:37 -05:00
i-norden e1abb399cd env vars for log stuff 2021-10-27 14:34:33 -05:00
i-norden 3d00e3ed05 enable configuration of a prerun range by env variable; prerun only mode 2021-10-27 13:36:19 -05:00
Ian Norden 84fff0aa03 Merge pull request #44 from vulcanize/stats
make DB stats collector registration optional
2021-10-25 18:57:40 -05:00
i-norden 79cb508d83 make DB stats collector registration optional 2021-10-25 18:54:54 -05:00
Ian Norden f51d89e873 Merge pull request #40 from vulcanize/metrics
finish prom metrics
2021-10-25 15:40:53 -05:00
i-norden a7ae868dd8 finish prom metrics 2021-10-25 15:40:11 -05:00
Ian Norden ee59ff3650 Merge pull request #39 from vulcanize/metrics
Metrics
2021-10-25 15:19:13 -05:00
i-norden 8571f35031 fixes 2021-10-25 14:54:55 -05:00
i-norden f679082c3b queue size metric; touchup 2021-10-25 14:27:11 -05:00
i-norden aace7793ba unify service RPC api and commands 2021-10-25 14:27:11 -05:00
Ian Norden b49e30d8e9 Merge pull request #38 from vulcanize/metrics
fix namespace
2021-10-24 18:39:07 -05:00
i-norden f207284e3a fix namespace 2021-10-24 18:37:01 -05:00
S. R. Wadleigh 3d3e68c0ae Merge pull request #37 from vulcanize/feature-docker-overlay
Update docker environment to support fs overlay
2021-10-21 20:19:53 +00:00
srwadleigh 66835c4c3e add trailing space to start script 2021-10-21 19:07:19 +00:00
srwadleigh 0d092cc535 add sudo and user permissions to Dockerfile, update startup_script.sh with overlay commands 2021-10-21 19:02:22 +00:00
Ian Norden a057940716 Merge pull request #35 from vulcanize/metrics
Metrics
2021-10-21 13:36:49 -05:00
i-norden 7dc6c52dec fix type 2021-10-21 13:29:30 -05:00
i-norden ca9d2a7bec update config 2021-10-21 13:10:16 -05:00
i-norden 072f035a02 wire up; customize metrics 2021-10-21 13:00:30 -05:00
i-norden 5a35f84de3 cli/cmd updates 2021-10-21 12:06:06 -05:00
i-norden 0f81dc32d2 prom pkg + monitoring 2021-10-21 11:57:44 -05:00
Arijit Das 6a912eef41 Merge pull request #34 from vulcanize/v1.10.11-statediff-0.0.27
Upgrade geth-statediff to version v1.10.11-statediff-0.0.27
2021-10-21 15:48:48 +05:30
Arijit Das 88f2eaed7f Upgrade geth-statediff to version v1.10.11-statediff-0.0.27 2021-10-21 15:44:56 +05:30
Ian Norden 26a88f06e7 Merge pull request #33 from vulcanize/update
fix CLI bug
2021-10-20 16:28:39 -05:00
i-norden 0dcec4ec8e fix CLI bug 2021-10-20 16:25:52 -05:00
Ian Norden 4cbaa630c7 Merge pull request #30 from vulcanize/updates
Updates
2021-10-19 17:25:50 -05:00
i-norden fb37ee9881 cli and toml bindings; fix unit tests; remove unused code 2021-10-19 16:49:04 -05:00
i-norden 01d0c21fd5 make cache configurable 2021-10-19 16:19:44 -05:00
Arijit Das 64e80eeb6a Merge pull request #28 from vulcanize/range-endpoint
Add RPC endpoint to write statediff.
2021-10-19 20:03:59 +05:30
Arijit Das 05b90107ca Add RPC endpoint to write statediff. 2021-10-19 17:27:51 +05:30
Arijit Das c4b95cc1b7 Merge pull request #27 from vulcanize/upgrade-geth
Upgrade geth to v1.10.10.
2021-10-19 11:44:40 +05:30
Arijit Das 7b9ca9b519 Update go.mod 2021-10-19 11:41:23 +05:30
Arijit Das b09e58c22a Upgrade geth to v1.10.9. 2021-10-14 12:17:38 +05:30
Ian Norden 450ff197a2 Merge pull request #20 from vulcanize/update
update to use v1.10.1-statediff-0.0.16
2021-03-24 18:54:46 -05:00
Ian Norden 41e2ae176f update to use v1.10.1-statediff-0.0.16; including fix for missing leaf keys in 'removed node' entries 2021-03-24 17:46:20 -05:00
Ian Norden fa0be51041 Merge pull request #19 from vulcanize/updates
Replicate latest statediffing geth updates here
2021-03-01 09:47:18 -06:00
Ian Norden 0315a31dd2 add endpoints for fetching/writing statediff by blockhash 2021-02-27 12:39:16 -06:00
Ian Norden 017daffb5a update to v1.9.25-statediff-v0.0.15 2021-02-27 12:39:03 -06:00
Roy Crihfield d3bed86fff Write directly to postgres (#16)
* Write directly to PostgreSQL

(port from geth statediff)

* Refactors for Node API change
* add namespacing to avoid name conflicts
2020-12-03 11:04:14 +08:00
roysc d61113ad12 doc headers 2020-12-03 10:08:20 +08:00
roysc 7a6b866163 docs, tweaks, fixes 2020-11-26 18:57:54 +08:00
roysc bd5d2992fa fix import
add namespacing to avoid name conflicts
2020-11-17 09:37:36 +08:00
roysc d886579f3d go.sum 2020-11-16 18:16:01 +08:00
roysc a93eb3ce35 fix http server 2020-11-16 18:15:53 +08:00
roysc abff9bf045 refactors
geth Node api change

sd ctor refactor
2020-11-16 17:49:59 +08:00
roysc ca8b3066d1 Write directly to PostgreSQL (port from geth statediff) 2020-11-15 15:18:34 +08:00
Ian Norden 294568da4e Merge pull request #14 from vulcanize/github_actions
GitHub actions
2020-11-09 12:53:37 -06:00
Ian Norden a5eea3f428 fix Dockerfile 2020-11-09 12:49:33 -06:00
Ian Norden 05328a1763 Makefile 2020-11-09 12:19:31 -06:00
Ian Norden 4901b3d35b github actions integration 2020-11-09 12:15:50 -06:00
Ian Norden f0a631109e env variables for server paths 2020-11-09 12:15:50 -06:00
36 changed files with 3448 additions and 963 deletions
+25
View File
@@ -0,0 +1,25 @@
name: Docker Compose Build
on:
push:
branches:
- master
jobs:
build:
name: Run docker build
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Get the version
id: vars
run: echo ::set-output name=sha::$(echo ${GITHUB_SHA:0:7})
- name: Run docker build
run: make docker-build
- name: Tag docker image
run: docker tag vulcanize/eth-statediff-service docker.pkg.github.com/vulcanize/eth-statediff-service/eth-statediff-service:${{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/eth-statediff-service/eth-statediff-service:${{steps.vars.outputs.sha}}
+12
View File
@@ -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: make docker-build
+25
View File
@@ -0,0 +1,25 @@
name: Publish Docker image
on:
release:
types: [published]
jobs:
push_to_registries:
name: Push Docker image to Docker Hub
runs-on: ubuntu-latest
steps:
- name: Get the version
id: vars
run: |
echo ::set-output name=sha::$(echo ${GITHUB_SHA:0:7})
echo ::set-output name=tag::$(echo ${GITHUB_REF#refs/tags/})
- 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/eth-statediff-service/eth-statediff-service:${{steps.vars.outputs.sha}}
- name: Docker Login to Docker Registry
run: echo ${{ secrets.VULCANIZEJENKINS_PAT }} | docker login -u vulcanizejenkins --password-stdin
- name: Tag docker image
run: docker tag docker.pkg.github.com/vulcanize/eth-statediff-service/eth-statediff-service:${{steps.vars.outputs.sha}} vulcanize/eth-statediff-service:${{steps.vars.outputs.tag}}
- name: Docker Push to Docker Hub
run: docker push vulcanize/eth-statediff-service:${{steps.vars.outputs.tag}}
+10 -5
View File
@@ -1,4 +1,4 @@
FROM golang:1.13-alpine as builder
FROM golang:1.16-alpine as builder
RUN apk --update --no-cache add make git g++ linux-headers
# DEBUG
@@ -6,16 +6,21 @@ RUN apk add busybox-extras
# Get and build ipfs-blockchain-watcher
ADD . /go/src/github.com/vulcanize/eth-statediff-service
#RUN git clone https://github.com/vulcanize/eth-statediff-service.git /go/src/github.com/vulcanize/eth-statediff-service
WORKDIR /go/src/github.com/vulcanize/eth-statediff-service
RUN GO111MODULE=on GCO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -ldflags '-extldflags "-static"' -o eth-statediff-service .
# app container
FROM alpine
ARG USER
ARG CONFIG_FILE
ARG EXPOSE_PORT
RUN adduser -Du 5000 $USER
ARG USER="vdm"
ARG CONFIG_FILE="./environments/config.toml"
ARG EXPOSE_PORT=8545
RUN adduser -Du 5000 $USER adm
RUN adduser $USER adm; apk --no-cache add sudo; echo '%adm ALL=(ALL) NOPASSWD:ALL' >> /etc/sudoers
WORKDIR /app
RUN chown $USER /app
USER $USER
+4
View File
@@ -0,0 +1,4 @@
## Build docker image
.PHONY: docker-build
docker-build:
docker build -t vulcanize/eth-statediff-service .
+74 -5
View File
@@ -1,4 +1,4 @@
## eth-statediff-service
# eth-statediff-service
[![Go Report Card](https://goreportcard.com/badge/github.com/vulcanize/eth-statediff-service)](https://goreportcard.com/report/github.com/vulcanize/eth-statediff-service)
@@ -10,11 +10,30 @@ Stand up a statediffing service directly on top of a go-ethereum leveldb instanc
This service can serve historical state data over the same rpc interface as
[statediffing geth](https://github.com/vulcanize/go-ethereum/releases/tag/v1.9.11-statediff-0.0.5) without needing to run a full node
Usage:
## Usage
`./eth-statediff-service serve --config={path to toml config file}`
### `serve`
Config:
To serve state diffs over RPC:
`eth-statediff-service serve --config=<config path>`
Available RPC methods are:
* `statediff_stateTrieAt()`
* `statediff_streamCodeAndCodeHash()`
* `statediff_stateDiffAt()`
* `statediff_writeStateDiffAt()`
* `statediff_writeStateDiffsInRange()`
e.g. `curl -X POST -H 'Content-Type: application/json' --data '{"jsonrpc":"2.0","method":"statediff_writeStateDiffsInRange","params":['"$BEGIN"', '"$END"', {"intermediateStateNodes":true,"intermediateStorageNodes":true,"includeBlock":true,"includeReceipts":true,"includeTD":true,"includeCode":true}],"id":1}' "$HOST":"$PORT"`
The process can be configured locally with sets of ranges to process as a "prerun" to processing directed by the server endpoints.
This is done by turning "prerun" on in the config (`statediff.prerun = true`) and defining ranged and params in the
`prerun` section of the config as shown below.
## Configuration
An example config file:
```toml
[leveldb]
@@ -22,13 +41,63 @@ Config:
ancient = "/Users/user/Library/Ethereum/geth/chaindata/ancient"
[server]
ipcPath = "~/.vulcanize/vulcanize.ipc"
ipcPath = ".ipc"
httpPath = "127.0.0.1:8545"
[statediff]
prerun = true
serviceWorkers = 1
workerQueueSize = 1024
trieWorkers = 4
[prerun]
only = false
ranges = [
[0, 1000]
]
[prerun.params]
intermediateStateNodes = true
intermediateStorageNodes = true
includeBlock = true
includeReceipts = true
includeTD = true
includeCode = true
watchedAddresses = []
watchedStorageKeys = []
[log]
file = ""
level = "info"
[eth]
chainID = 1
[database]
name = "vulcanize_test"
hostname = "localhost"
port = 5432
user = "vulcanize"
password = "..."
type = "postgres"
driver = "sqlx"
dumpDestination = ""
filePath = ""
[cache]
database = 1024
trie = 1024
[prom]
dbStats = false
metrics = true
http = true
httpAddr = "localhost"
httpPort = "8889"
[ethereum]
nodeID = ""
clientName = "eth-statediff-service"
genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
networkID = 1
chainID = 1
```
+139
View File
@@ -0,0 +1,139 @@
// VulcanizeDB
// Copyright © 2019 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"github.com/spf13/viper"
)
const (
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_CHAIN_ID = "ETH_CHAIN_ID"
DB_CACHE_SIZE_MB = "DB_CACHE_SIZE_MB"
TRIE_CACHE_SIZE_MB = "TRIE_CACHE_SIZE_MB"
LVLDB_PATH = "LVLDB_PATH"
LVLDB_ANCIENT = "LVLDB_ANCIENT"
STATEDIFF_PRERUN = "STATEDIFF_PRERUN"
STATEDIFF_TRIE_WORKERS = "STATEDIFF_TRIE_WORKERS"
STATEDIFF_SERVICE_WORKERS = "STATEDIFF_SERVICE_WORKERS"
STATEDIFF_WORKER_QUEUE_SIZE = "STATEDIFF_WORKER_QUEUE_SIZE"
SERVICE_IPC_PATH = "SERVICE_IPC_PATH"
SERVICE_HTTP_PATH = "SERVICE_HTTP_PATH"
PROM_METRICS = "PROM_METRICS"
PROM_HTTP = "PROM_HTTP"
PROM_HTTP_ADDR = "PROM_HTTP_ADDR"
PROM_HTTP_PORT = "PROM_HTTP_PORT"
PROM_DB_STATS = "PROM_DB_STATS"
PRERUN_ONLY = "PRERUN_ONLY"
PRERUN_RANGE_START = "PRERUN_RANGE_START"
PRERUN_RANGE_STOP = "PRERUN_RANGE_STOP"
PRERUN_INTERMEDIATE_STATE_NODES = "PRERUN_INTERMEDIATE_STATE_NODES"
PRERUN_INTERMEDIATE_STORAGE_NODES = "PRERUN_INTERMEDIATE_STORAGE_NODES"
PRERUN_INCLUDE_BLOCK = "PRERUN_INCLUDE_BLOCK"
PRERUN_INCLUDE_RECEIPTS = "PRERUN_INCLUDE_RECEIPTS"
PRERUN_INCLUDE_TD = "PRERUN_INCLUDE_TD"
PRERUN_INCLUDE_CODE = "PRERUN_INCLUDE_CODE"
LOG_LEVEL = "LOG_LEVEL"
LOG_FILE_PATH = "LOG_FILE_PATH"
DATABASE_NAME = "DATABASE_NAME"
DATABASE_HOSTNAME = "DATABASE_HOSTNAME"
DATABASE_PORT = "DATABASE_PORT"
DATABASE_USER = "DATABASE_USER"
DATABASE_PASSWORD = "DATABASE_PASSWORD"
DATABASE_TYPE = "DATABASE_TYPE"
DATABASE_DRIVER_TYPE = "DATABASE_DRIVER_TYPE"
DATABASE_DUMP_DST = "DATABASE_DUMP_DST"
DATABASE_FILE_PATH = "DATABASE_FILE_PATH"
DATABASE_MAX_IDLE_CONNECTIONS = "DATABASE_MAX_IDLE_CONNECTIONS"
DATABASE_MAX_OPEN_CONNECTIONS = "DATABASE_MAX_OPEN_CONNECTIONS"
DATABASE_MIN_OPEN_CONNS = "DATABASE_MIN_OPEN_CONNS"
DATABASE_MAX_CONN_LIFETIME = "DATABASE_MAX_CONN_LIFETIME"
DATABASE_CONN_TIMEOUT = "DATABSE_CONN_TIMEOUT"
DATABASE_MAX_CONN_IDLE_TIME = "DATABASE_MAX_CONN_IDLE_TIME"
)
// Bind env vars for eth node and DB configuration
func init() {
viper.BindEnv("server.ipcPath", SERVICE_IPC_PATH)
viper.BindEnv("server.httpPath", SERVICE_HTTP_PATH)
viper.BindEnv("ethereum.nodeID", ETH_NODE_ID)
viper.BindEnv("ethereum.clientName", ETH_CLIENT_NAME)
viper.BindEnv("ethereum.genesisBlock", ETH_GENESIS_BLOCK)
viper.BindEnv("ethereum.networkID", ETH_NETWORK_ID)
viper.BindEnv("ethereum.chainID", ETH_CHAIN_ID)
viper.BindEnv("database.name", DATABASE_NAME)
viper.BindEnv("database.hostname", DATABASE_HOSTNAME)
viper.BindEnv("database.port", DATABASE_PORT)
viper.BindEnv("database.user", DATABASE_USER)
viper.BindEnv("database.password", DATABASE_PASSWORD)
viper.BindEnv("database.maxIdle", DATABASE_MAX_IDLE_CONNECTIONS)
viper.BindEnv("database.maxOpen", DATABASE_MAX_OPEN_CONNECTIONS)
viper.BindEnv("database.minOpen", DATABASE_MIN_OPEN_CONNS)
viper.BindEnv("database.maxConnLifetime", DATABASE_MAX_CONN_LIFETIME)
viper.BindEnv("database.connTimeout", DATABASE_CONN_TIMEOUT)
viper.BindEnv("database.maxIdleTime", DATABASE_MAX_CONN_IDLE_TIME)
viper.BindEnv("database.type", DATABASE_TYPE)
viper.BindEnv("database.driver", DATABASE_DRIVER_TYPE)
viper.BindEnv("database.dumpDestination", DATABASE_DUMP_DST)
viper.BindEnv("database.filePath", DATABASE_FILE_PATH)
viper.BindEnv("cache.database", DB_CACHE_SIZE_MB)
viper.BindEnv("cache.trie", TRIE_CACHE_SIZE_MB)
viper.BindEnv("leveldb.path", LVLDB_PATH)
viper.BindEnv("leveldb.ancient", LVLDB_ANCIENT)
viper.BindEnv("prom.metrics", PROM_METRICS)
viper.BindEnv("prom.http", PROM_HTTP)
viper.BindEnv("prom.httpAddr", PROM_HTTP_ADDR)
viper.BindEnv("prom.httpPort", PROM_HTTP_PORT)
viper.BindEnv("prom.dbStats", PROM_DB_STATS)
viper.BindEnv("statediff.serviceWorkers", STATEDIFF_SERVICE_WORKERS)
viper.BindEnv("statediff.trieWorkers", STATEDIFF_TRIE_WORKERS)
viper.BindEnv("statediff.workerQueueSize", STATEDIFF_WORKER_QUEUE_SIZE)
viper.BindEnv("statediff.prerun", STATEDIFF_PRERUN)
viper.BindEnv("prerun.only", PRERUN_ONLY)
viper.BindEnv("prerun.start", PRERUN_RANGE_START)
viper.BindEnv("prerun.stop", PRERUN_RANGE_STOP)
viper.BindEnv("prerun.params.intermediateStateNodes", PRERUN_INTERMEDIATE_STATE_NODES)
viper.BindEnv("prerun.params.intermediateStorageNodes", PRERUN_INTERMEDIATE_STORAGE_NODES)
viper.BindEnv("prerun.params.includeBlock", PRERUN_INCLUDE_BLOCK)
viper.BindEnv("prerun.params.includeReceipts", PRERUN_INCLUDE_RECEIPTS)
viper.BindEnv("prerun.params.includeTD", PRERUN_INCLUDE_TD)
viper.BindEnv("prerun.params.includeCode", PRERUN_INCLUDE_CODE)
viper.BindEnv("log.level", LOG_LEVEL)
viper.BindEnv("log.file", LOG_FILE_PATH)
}
+261 -5
View File
@@ -18,12 +18,23 @@ package cmd
import (
"fmt"
"math/rand"
"os"
"strings"
"time"
"github.com/ethereum/go-ethereum/cmd/utils"
"github.com/ethereum/go-ethereum/statediff/indexer/database/dump"
"github.com/ethereum/go-ethereum/statediff/indexer/database/file"
"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
"github.com/ethereum/go-ethereum/statediff/indexer/interfaces"
"github.com/ethereum/go-ethereum/statediff/indexer/node"
"github.com/ethereum/go-ethereum/statediff/indexer/shared"
log "github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"github.com/spf13/viper"
"github.com/vulcanize/eth-statediff-service/pkg/prom"
)
var (
@@ -45,7 +56,7 @@ func Execute() {
}
func initFuncs(cmd *cobra.Command, args []string) {
logfile := viper.GetString("logfile")
logfile := viper.GetString("log.file")
if logfile != "" {
file, err := os.OpenFile(logfile,
os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
@@ -62,10 +73,24 @@ func initFuncs(cmd *cobra.Command, args []string) {
if err := logLevel(); err != nil {
log.Fatal("Could not set log level: ", err)
}
if viper.GetBool("prom.metrics") {
log.Info("initializing prometheus metrics")
prom.Init()
}
if viper.GetBool("prom.http") {
addr := fmt.Sprintf(
"%s:%s",
viper.GetString("prom.httpAddr"),
viper.GetString("prom.httpPort"),
)
log.Info("starting prometheus server")
prom.Listen(addr)
}
}
func logLevel() error {
viper.BindEnv("log.level", "LOGRUS_LEVEL")
lvl, err := log.ParseLevel(viper.GetString("log.level"))
if err != nil {
return err
@@ -83,14 +108,121 @@ func init() {
viper.SetEnvKeyReplacer(strings.NewReplacer(".", "_"))
viper.AutomaticEnv()
rootCmd.PersistentFlags().String("http-path", "", "vdb server http path")
rootCmd.PersistentFlags().String("ipc-path", "", "vdb server ipc path")
rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", "config file location")
rootCmd.PersistentFlags().String("log-file", "", "file path for logging")
rootCmd.PersistentFlags().String("log-level", log.InfoLevel.String(), "log level (trace, debug, info, warn, error, fatal, panic")
rootCmd.PersistentFlags().Int("workers", 0, "number of concurrent workers to use")
rootCmd.PersistentFlags().String("log-level", log.InfoLevel.String(),
"log level (trace, debug, info, warn, error, fatal, panic")
rootCmd.PersistentFlags().String("leveldb-path", "", "path to primary datastore")
rootCmd.PersistentFlags().String("ancient-path", "", "path to ancient datastore")
rootCmd.PersistentFlags().Bool("prerun", false, "turn on prerun of toml configured ranges")
rootCmd.PersistentFlags().Int("service-workers", 0, "number of range requests to process concurrently")
rootCmd.PersistentFlags().Int("trie-workers", 0, "number of workers to use for trie traversal and processing")
rootCmd.PersistentFlags().Int("worker-queue-size", 0, "size of the range request queue for service workers")
rootCmd.PersistentFlags().String("database-name", "vulcanize_public", "database name")
rootCmd.PersistentFlags().Int("database-port", 5432, "database port")
rootCmd.PersistentFlags().String("database-hostname", "localhost", "database hostname")
rootCmd.PersistentFlags().String("database-user", "", "database user")
rootCmd.PersistentFlags().String("database-password", "", "database password")
rootCmd.PersistentFlags().Int("database-max-idle", 0, "database max number of idle connections")
rootCmd.PersistentFlags().Int("database-max-open", 0, "database max number of open connections")
rootCmd.PersistentFlags().Int("database-min-open", 0, "database min number of open connections")
rootCmd.PersistentFlags().Duration("database-max-conn-lifetime", 0, "database max connection lifetime")
rootCmd.PersistentFlags().Duration("database-conn-timeout", 0, "database connection timeout")
rootCmd.PersistentFlags().Duration("database-max-idle-time", 0, "database max connection idle time")
rootCmd.PersistentFlags().String("database-type", "postgres", "database type (currently supported: postgres, dump)")
rootCmd.PersistentFlags().String("database-driver", "sqlx", "database driver type (currently supported: sqlx, pgx)")
rootCmd.PersistentFlags().String("database-dump-dst", "stdout", "dump destination (for database-type=dump; options: stdout, stderr, discard)")
rootCmd.PersistentFlags().String("database-file-path", "", "full file path (for database-type=file)")
rootCmd.PersistentFlags().String("eth-node-id", "", "eth node id")
rootCmd.PersistentFlags().String("eth-client-name", "eth-statediff-service", "eth client name")
rootCmd.PersistentFlags().String("eth-genesis-block",
"0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3", "eth genesis block hash")
rootCmd.PersistentFlags().String("eth-network-id", "1", "eth network id")
rootCmd.PersistentFlags().String("eth-chain-id", "1", "eth chain id")
rootCmd.PersistentFlags().Int("cache-db", 1024, "megabytes of memory allocated to database cache")
rootCmd.PersistentFlags().Int("cache-trie", 1024, "Megabytes of memory allocated to trie cache")
rootCmd.PersistentFlags().Bool("prom-http", false, "enable prometheus http service")
rootCmd.PersistentFlags().String("prom-http-addr", "127.0.0.1", "prometheus http host")
rootCmd.PersistentFlags().String("prom-http-port", "8080", "prometheus http port")
rootCmd.PersistentFlags().Bool("prom-db-stats", false, "enables prometheus db stats")
rootCmd.PersistentFlags().Bool("prom-metrics", false, "enable prometheus metrics")
rootCmd.PersistentFlags().Bool("prerun-only", false, "only process pre-configured ranges; exit afterwards")
rootCmd.PersistentFlags().Int("prerun-start", 0, "start height for a prerun range")
rootCmd.PersistentFlags().Int("prerun-stop", 0, "stop height for a prerun range")
rootCmd.PersistentFlags().Bool("prerun-intermediate-state-nodes", true, "include intermediate state nodes in state diff")
rootCmd.PersistentFlags().Bool("prerun-intermediate-storage-nodes", true, "include intermediate storage nodes in state diff")
rootCmd.PersistentFlags().Bool("prerun-include-block", true, "include block data in the statediff payload")
rootCmd.PersistentFlags().Bool("prerun-include-receipts", true, "include receipts in the statediff payload")
rootCmd.PersistentFlags().Bool("prerun-include-td", true, "include td in the statediff payload")
rootCmd.PersistentFlags().Bool("prerun-include-code", true, "include code and codehash mappings in statediff payload")
viper.BindPFlag("server.httpPath", rootCmd.PersistentFlags().Lookup("http-path"))
viper.BindPFlag("server.ipcPath", rootCmd.PersistentFlags().Lookup("ipc-path"))
viper.BindPFlag("log.file", rootCmd.PersistentFlags().Lookup("log-file"))
viper.BindPFlag("log.level", rootCmd.PersistentFlags().Lookup("log-level"))
viper.BindPFlag("statediff.workers", rootCmd.PersistentFlags().Lookup("workers"))
viper.BindPFlag("statediff.prerun", rootCmd.PersistentFlags().Lookup("prerun"))
viper.BindPFlag("statediff.serviceWorkers", rootCmd.PersistentFlags().Lookup("service-workers"))
viper.BindPFlag("statediff.trieWorkers", rootCmd.PersistentFlags().Lookup("trie-workers"))
viper.BindPFlag("statediff.workerQueueSize", rootCmd.PersistentFlags().Lookup("worker-queue-size"))
viper.BindPFlag("leveldb.path", rootCmd.PersistentFlags().Lookup("leveldb-path"))
viper.BindPFlag("leveldb.ancient", rootCmd.PersistentFlags().Lookup("ancient-path"))
viper.BindPFlag("database.name", rootCmd.PersistentFlags().Lookup("database-name"))
viper.BindPFlag("database.port", rootCmd.PersistentFlags().Lookup("database-port"))
viper.BindPFlag("database.hostname", rootCmd.PersistentFlags().Lookup("database-hostname"))
viper.BindPFlag("database.user", rootCmd.PersistentFlags().Lookup("database-user"))
viper.BindPFlag("database.password", rootCmd.PersistentFlags().Lookup("database-password"))
viper.BindPFlag("database.maxIdle", rootCmd.PersistentFlags().Lookup("database-max-idle"))
viper.BindPFlag("database.maxOpen", rootCmd.PersistentFlags().Lookup("database-max-open"))
viper.BindPFlag("database.minOpen", rootCmd.PersistentFlags().Lookup("database-min-open"))
viper.BindPFlag("database.maxConnLifetime", rootCmd.PersistentFlags().Lookup("database-max-conn-lifetime"))
viper.BindPFlag("database.connTimeout", rootCmd.PersistentFlags().Lookup("database-conn-timeout"))
viper.BindPFlag("database.maxIdleTime", rootCmd.PersistentFlags().Lookup("database-max-idle-time"))
viper.BindPFlag("database.type", rootCmd.PersistentFlags().Lookup("database-type"))
viper.BindPFlag("database.driver", rootCmd.PersistentFlags().Lookup("database-driver"))
viper.BindPFlag("database.dumpDestination", rootCmd.PersistentFlags().Lookup("database-dump-dst"))
viper.BindPFlag("database.filePath", rootCmd.PersistentFlags().Lookup("database-file-path"))
viper.BindPFlag("ethereum.nodeID", rootCmd.PersistentFlags().Lookup("eth-node-id"))
viper.BindPFlag("ethereum.clientName", rootCmd.PersistentFlags().Lookup("eth-client-name"))
viper.BindPFlag("ethereum.genesisBlock", rootCmd.PersistentFlags().Lookup("eth-genesis-block"))
viper.BindPFlag("ethereum.networkID", rootCmd.PersistentFlags().Lookup("eth-network-id"))
viper.BindPFlag("ethereum.chainID", rootCmd.PersistentFlags().Lookup("eth-chain-id"))
viper.BindPFlag("cache.database", rootCmd.PersistentFlags().Lookup("cache-db"))
viper.BindPFlag("cache.trie", rootCmd.PersistentFlags().Lookup("cache-trie"))
viper.BindPFlag("prom.http", rootCmd.PersistentFlags().Lookup("prom-http"))
viper.BindPFlag("prom.httpAddr", rootCmd.PersistentFlags().Lookup("prom-http-addr"))
viper.BindPFlag("prom.httpPort", rootCmd.PersistentFlags().Lookup("prom-http-port"))
viper.BindPFlag("prom.dbStats", rootCmd.PersistentFlags().Lookup("prom-db-stats"))
viper.BindPFlag("prom.metrics", rootCmd.PersistentFlags().Lookup("prom-metrics"))
viper.BindPFlag("prerun.only", rootCmd.PersistentFlags().Lookup("prerun-only"))
viper.BindPFlag("prerun.start", rootCmd.PersistentFlags().Lookup("prerun-start"))
viper.BindPFlag("prerun.stop", rootCmd.PersistentFlags().Lookup("prerun-stop"))
viper.BindPFlag("prerun.params.intermediateStateNodes", rootCmd.PersistentFlags().Lookup("prerun-intermediate-state-nodes"))
viper.BindPFlag("prerun.params.intermediateStorageNodes", rootCmd.PersistentFlags().Lookup("prerun-intermediate-storage-nodes"))
viper.BindPFlag("prerun.params.includeBlock", rootCmd.PersistentFlags().Lookup("prerun-include-block"))
viper.BindPFlag("prerun.params.includeReceipts", rootCmd.PersistentFlags().Lookup("prerun-include-receipts"))
viper.BindPFlag("prerun.params.includeTD", rootCmd.PersistentFlags().Lookup("prerun-include-td"))
viper.BindPFlag("prerun.params.includeCode", rootCmd.PersistentFlags().Lookup("prerun-include-code"))
rand.Seed(time.Now().UnixNano())
}
func initConfig() {
@@ -105,3 +237,127 @@ func initConfig() {
log.Warn("No config file passed with --config flag")
}
}
func getEthNodeInfo() node.Info {
var nodeID, genesisBlock, networkID, clientName string
var chainID uint64
if !viper.IsSet("ethereum.nodeID") {
nodeID = randSeq(12)
} else {
nodeID = viper.GetString("ethereum.nodeID")
}
genesisBlock = viper.GetString("ethereum.genesisBlock")
if !viper.IsSet("ethereum.genesisBlock") {
genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
} else {
genesisBlock = viper.GetString("ethereum.genesisBlock")
}
if !viper.IsSet("ethereum.chainID") {
chainID = 1
} else {
chainID = viper.GetUint64("ethereum.chainID")
}
if !viper.IsSet("ethereum.networkID") {
networkID = "1"
} else {
networkID = viper.GetString("ethereum.networkID")
}
if !viper.IsSet("ethereum.clientName") {
clientName = "eth-statediff-service"
} else {
clientName = viper.GetString("ethereum.clientName")
}
return node.Info{
ID: nodeID,
ClientName: clientName,
GenesisBlock: genesisBlock,
NetworkID: networkID,
ChainID: chainID,
}
}
var characters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
func randSeq(n int) string {
b := make([]rune, n)
for i := range b {
b[i] = characters[rand.Intn(len(characters))]
}
return string(b)
}
// getConfig constructs and returns the appropriate config from viper params
func getConfig(nodeInfo node.Info) (interfaces.Config, error) {
dbTypeStr := viper.GetString("database.type")
dbType, err := shared.ResolveDBType(dbTypeStr)
if err != nil {
return nil, err
}
logWithCommand.Infof("configuring service for database type: %s", dbType)
var indexerConfig interfaces.Config
switch dbType {
case shared.FILE:
logWithCommand.Info("starting in sql file writing mode")
filePathStr := viper.GetString("database.filePath")
if filePathStr == "" {
logWithCommand.Fatal("when operating in sql file writing mode a file path must be provided")
}
indexerConfig = file.Config{FilePath: filePathStr}
case shared.DUMP:
logWithCommand.Info("starting in data dump mode")
dumpDstStr := viper.GetString("database.dumpDestination")
dumpDst, err := dump.ResolveDumpType(dumpDstStr)
if err != nil {
return nil, err
}
switch dumpDst {
case dump.STDERR:
indexerConfig = dump.Config{Dump: os.Stdout}
case dump.STDOUT:
indexerConfig = dump.Config{Dump: os.Stderr}
case dump.DISCARD:
indexerConfig = dump.Config{Dump: dump.NewDiscardWriterCloser()}
default:
return nil, fmt.Errorf("unrecognized dump destination: %s", dumpDst)
}
case shared.POSTGRES:
logWithCommand.Info("starting in postgres mode")
driverTypeStr := viper.GetString("database.driver")
driverType, err := postgres.ResolveDriverType(driverTypeStr)
if err != nil {
utils.Fatalf("%v", err)
}
pgConfig := postgres.Config{
Hostname: viper.GetString("database.hostname"),
Port: viper.GetInt("database.port"),
DatabaseName: viper.GetString("database.name"),
Username: viper.GetString("database.user"),
Password: viper.GetString("database.password"),
ID: nodeInfo.ID,
ClientName: nodeInfo.ClientName,
Driver: driverType,
}
if viper.IsSet("database.maxIdle") {
pgConfig.MaxIdle = viper.GetInt("database.maxIdle")
}
if viper.IsSet("database.maxOpen") {
pgConfig.MaxConns = viper.GetInt("database.maxOpen")
}
if viper.IsSet("database.minOpen") {
pgConfig.MinConns = viper.GetInt("database.minOpen")
}
if viper.IsSet("database.maxConnLifetime") {
pgConfig.MaxConnLifetime = viper.GetDuration("database.maxConnLifetime")
}
if viper.IsSet("database.connTimeout") {
pgConfig.ConnTimeout = viper.GetDuration("database.connTimeout")
}
if viper.IsSet("database.maxIdleTime") {
pgConfig.MaxConnIdleTime = viper.GetDuration("database.maxIdleTime")
}
indexerConfig = pgConfig
default:
return nil, fmt.Errorf("unrecognized db type: %s", dbType)
}
return indexerConfig, nil
}
+28 -64
View File
@@ -16,24 +16,23 @@
package cmd
import (
"fmt"
"os"
"os/signal"
"sync"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rpc"
"github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"github.com/spf13/viper"
sd "github.com/vulcanize/eth-statediff-service/pkg"
srpc "github.com/vulcanize/eth-statediff-service/pkg/rpc"
)
// serveCmd represents the serve command
var serveCmd = &cobra.Command{
Use: "serve",
Short: "Standup a standalone statediffing RPC service on top of leveldb",
Short: "Stand up a standalone statediffing RPC service on top of leveldb",
Long: `Usage
./eth-statediff-service serve --config={path to toml config file}`,
@@ -44,72 +43,48 @@ var serveCmd = &cobra.Command{
},
}
func init() {
rootCmd.AddCommand(serveCmd)
}
func serve() {
logWithCommand.Info("starting statediff RPC service")
logWithCommand.Info("Running eth-statediff-service serve command")
// load params
path := viper.GetString("leveldb.path")
ancientPath := viper.GetString("leveldb.ancient")
if path == "" || ancientPath == "" {
logWithCommand.Fatal("require a valid eth leveldb primary datastore path and ancient datastore path")
}
chainID := viper.GetUint64("eth.chainID")
config, err := chainConfig(chainID)
statediffService, err := createStateDiffService()
if err != nil {
logWithCommand.Fatal(err)
}
// create leveldb reader
logWithCommand.Info("creating leveldb reader")
lvlDBReader, err := sd.NewLvlDBReader(path, ancientPath, config)
if err != nil {
logWithCommand.Fatal(err)
}
// create statediff service
logWithCommand.Info("creating statediff service")
statediffService, err := sd.NewStateDiffService(
lvlDBReader, sd.Config{Workers: viper.GetUint("statediff.workers")})
if err != nil {
logWithCommand.Fatal(err)
// short circuit if we only want to perform prerun
if viper.GetBool("prerun.only") {
if err := statediffService.Run(nil); err != nil {
logWithCommand.Fatal("unable to perform prerun: %v", err)
}
return
}
// start service and servers
logWithCommand.Info("starting statediff service")
logWithCommand.Info("Starting statediff service")
wg := new(sync.WaitGroup)
go statediffService.Loop(wg)
logWithCommand.Info("starting rpc servers")
if err := statediffService.Loop(wg); err != nil {
logWithCommand.Fatalf("unable to start statediff service: %v", err)
}
logWithCommand.Info("Starting RPC servers")
if err := startServers(statediffService); err != nil {
logWithCommand.Fatal(err)
}
logWithCommand.Info("rpc servers successfully spun up; awaiting requests")
logWithCommand.Info("RPC servers successfully spun up; awaiting requests")
// clean shutdown
shutdown := make(chan os.Signal)
signal.Notify(shutdown, os.Interrupt)
<-shutdown
logWithCommand.Info("received interrupt signal, shutting down")
logWithCommand.Info("Received interrupt signal, shutting down")
statediffService.Stop()
wg.Wait()
}
func init() {
rootCmd.AddCommand(serveCmd)
serveCmd.PersistentFlags().String("leveldb-path", "", "path to primary datastore")
serveCmd.PersistentFlags().String("ancient-path", "", "path to ancient datastore")
serveCmd.PersistentFlags().Uint("chain-id", 1, "ethereum chain id (mainnet = 1)")
serveCmd.PersistentFlags().String("http-path", "", "vdb server http path")
serveCmd.PersistentFlags().String("ipc-path", "", "vdb server ipc path")
viper.BindPFlag("leveldb.path", serveCmd.PersistentFlags().Lookup("leveldb-path"))
viper.BindPFlag("leveldb.ancient", serveCmd.PersistentFlags().Lookup("ancient-path"))
viper.BindPFlag("eth.chainID", serveCmd.PersistentFlags().Lookup("chain-id"))
viper.BindPFlag("server.httpPath", serveCmd.PersistentFlags().Lookup("http-path"))
viper.BindPFlag("server.ipcPath", serveCmd.PersistentFlags().Lookup("ipc-path"))
}
func startServers(serv sd.IService) error {
func startServers(serv sd.StateDiffService) error {
ipcPath := viper.GetString("server.ipcPath")
httpPath := viper.GetString("server.httpPath")
if ipcPath == "" && httpPath == "" {
@@ -117,31 +92,20 @@ func startServers(serv sd.IService) error {
}
if ipcPath != "" {
logWithCommand.Info("starting up IPC server")
_, _, err := rpc.StartIPCEndpoint(ipcPath, serv.APIs())
_, _, err := srpc.StartIPCEndpoint(ipcPath, serv.APIs())
if err != nil {
return err
}
}
if httpPath != "" {
logWithCommand.Info("starting up HTTP server")
if _, _, err := rpc.StartHTTPEndpoint(httpPath, serv.APIs(), []string{sd.APIName}, nil, nil, rpc.HTTPTimeouts{}); err != nil {
_, err := srpc.StartHTTPEndpoint(httpPath, serv.APIs(), []string{"statediff"}, nil, []string{"*"}, rpc.HTTPTimeouts{})
if err != nil {
return err
}
} else {
logWithCommand.Info("HTTP server is disabled")
}
return nil
}
func chainConfig(chainID uint64) (*params.ChainConfig, error) {
switch chainID {
case 1:
return params.MainnetChainConfig, nil
case 3:
return params.TestnetChainConfig, nil // Ropsten
case 4:
return params.RinkebyChainConfig, nil
case 5:
return params.GoerliChainConfig, nil
default:
return nil, fmt.Errorf("chain config for chainid %d not available", chainID)
}
}
+135
View File
@@ -0,0 +1,135 @@
package cmd
import (
"context"
"fmt"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/params"
gethsd "github.com/ethereum/go-ethereum/statediff"
ind "github.com/ethereum/go-ethereum/statediff/indexer"
"github.com/ethereum/go-ethereum/trie"
"github.com/spf13/viper"
sd "github.com/vulcanize/eth-statediff-service/pkg"
)
type blockRange [2]uint64
func createStateDiffService() (sd.StateDiffService, error) {
// load some necessary params
logWithCommand.Info("Loading statediff service parameters")
path := viper.GetString("leveldb.path")
ancientPath := viper.GetString("leveldb.ancient")
if path == "" || ancientPath == "" {
logWithCommand.Fatal("require a valid eth leveldb primary datastore path and ancient datastore path")
}
nodeInfo := getEthNodeInfo()
chainConf, err := chainConfig(nodeInfo.ChainID)
if err != nil {
logWithCommand.Fatal(err)
}
// create leveldb reader
logWithCommand.Info("Creating leveldb reader")
readerConf := sd.LvLDBReaderConfig{
TrieConfig: &trie.Config{
Cache: viper.GetInt("cache.trie"),
Journal: "",
Preimages: false,
},
ChainConfig: chainConf,
Path: path,
AncientPath: ancientPath,
DBCacheSize: viper.GetInt("cache.database"),
}
lvlDBReader, err := sd.NewLvlDBReader(readerConf)
if err != nil {
logWithCommand.Fatal(err)
}
// create statediff service
logWithCommand.Info("Setting up database")
conf, err := getConfig(nodeInfo)
if err != nil {
logWithCommand.Fatal(err)
}
logWithCommand.Info("Creating statediff indexer")
indexer, err := ind.NewStateDiffIndexer(context.Background(), chainConf, nodeInfo, conf)
if err != nil {
logWithCommand.Fatal(err)
}
logWithCommand.Info("Creating statediff service")
sdConf := sd.Config{
ServiceWorkers: viper.GetUint("statediff.serviceWorkers"),
TrieWorkers: viper.GetUint("statediff.trieWorkers"),
WorkerQueueSize: viper.GetUint("statediff.workerQueueSize"),
PreRuns: setupPreRunRanges(),
}
return sd.NewStateDiffService(lvlDBReader, indexer, sdConf)
}
func setupPreRunRanges() []sd.RangeRequest {
if !viper.GetBool("statediff.prerun") {
return nil
}
preRunParams := gethsd.Params{
IntermediateStateNodes: viper.GetBool("prerun.params.intermediateStateNodes"),
IntermediateStorageNodes: viper.GetBool("prerun.params.intermediateStorageNodes"),
IncludeBlock: viper.GetBool("prerun.params.includeBlock"),
IncludeReceipts: viper.GetBool("prerun.params.includeReceipts"),
IncludeTD: viper.GetBool("prerun.params.includeTD"),
IncludeCode: viper.GetBool("prerun.params.includeCode"),
}
var addrStrs []string
viper.UnmarshalKey("prerun.params.watchedAddresses", &addrStrs)
addrs := make([]common.Address, len(addrStrs))
for i, addrStr := range addrStrs {
addrs[i] = common.HexToAddress(addrStr)
}
preRunParams.WatchedAddresses = addrs
var storageKeyStrs []string
viper.UnmarshalKey("prerun.params.watchedStorageKeys", &storageKeyStrs)
keys := make([]common.Hash, len(storageKeyStrs))
for i, keyStr := range storageKeyStrs {
keys[i] = common.HexToHash(keyStr)
}
preRunParams.WatchedStorageSlots = keys
var rawRanges []blockRange
viper.UnmarshalKey("prerun.ranges", &rawRanges)
blockRanges := make([]sd.RangeRequest, len(rawRanges))
for i, rawRange := range rawRanges {
blockRanges[i] = sd.RangeRequest{
Start: rawRange[0],
Stop: rawRange[1],
Params: preRunParams,
}
}
if viper.IsSet("prerun.start") && viper.IsSet("prerun.stop") {
hardStart := viper.GetInt("prerun.start")
hardStop := viper.GetInt("prerun.stop")
blockRanges = append(blockRanges, sd.RangeRequest{
Start: uint64(hardStart),
Stop: uint64(hardStop),
Params: preRunParams,
})
}
return blockRanges
}
func chainConfig(chainID uint64) (*params.ChainConfig, error) {
switch chainID {
case 1:
return params.MainnetChainConfig, nil
case 3:
return params.RopstenChainConfig, nil // Ropsten
case 4:
return params.RinkebyChainConfig, nil
case 5:
return params.GoerliChainConfig, nil
default:
return nil, fmt.Errorf("chain config for chainid %d not available", chainID)
}
}
+1 -1
View File
@@ -6,7 +6,7 @@ services:
context: ./
cache_from:
- alpine:latest
- golang:1.13
- golang:1.16
dockerfile: ./Dockerfile
args:
USER: "vdbm"
+59
View File
@@ -0,0 +1,59 @@
[leveldb]
path = "/app/geth-rw/chaindata"
ancient = "/app/geth-rw/chaindata/ancient"
[server]
ipcPath = ""
httpPath = "0.0.0.0:8545"
[statediff]
prerun = true
serviceWorkers = 1
workerQueueSize = 1024
trieWorkers = 16
[prerun]
only = true
ranges = []
[prerun.params]
intermediateStateNodes = true
intermediateStorageNodes = true
includeBlock = true
includeReceipts = true
includeTD = true
includeCode = true
watchedAddresses = []
watchedStorageKeys = []
[log]
file = ""
level = "info"
[database]
name = ""
hostname = ""
port = 5432
user = ""
password = ""
type = "postgres"
driver = "sqlx"
dumpDestination = ""
filePath = ""
[cache]
database = 1024
trie = 4096
[prom]
dbStats = false
metrics = true
http = true
httpAddr = "0.0.0.0"
httpPort = 9100
[ethereum]
nodeID = ""
clientName = "eth-statediff-service"
genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
networkID = 1
chainID = 1
+47 -3
View File
@@ -7,11 +7,55 @@
httpPath = "127.0.0.1:8545"
[statediff]
workers = 4
prerun = true
serviceWorkers = 1
workerQueueSize = 1024
trieWorkers = 4
[prerun]
only = false
ranges = [
[0, 1000]
]
[prerun.params]
intermediateStateNodes = true
intermediateStorageNodes = true
includeBlock = true
includeReceipts = true
includeTD = true
includeCode = true
watchedAddresses = []
watchedStorageKeys = []
[log]
file = ""
level = "info"
[eth]
chainID = 1
[database]
name = "vulcanize_test"
hostname = "localhost"
port = 5432
user = "vulcanize"
password = "..."
type = "postgres"
driver = "sqlx"
dumpDestination = ""
filePath = ""
[cache]
database = 1024
trie = 1024
[prom]
dbStats = false
metrics = true
http = true
httpAddr = "localhost"
httpPort = "8889"
[ethereum]
nodeID = ""
clientName = "eth-statediff-service"
genesisBlock = "0xd4e56740f876aef8c010b86a40d5f56745a118d0906a34e69aec8c0db1cb8fa3"
networkID = 1
chainID = 1
+8 -6
View File
@@ -1,13 +1,15 @@
module github.com/vulcanize/eth-statediff-service
go 1.13
go 1.16
require (
github.com/ethereum/go-ethereum v1.9.11
github.com/sirupsen/logrus v1.6.0
github.com/spf13/cobra v1.0.0
github.com/ethereum/go-ethereum v1.10.11
github.com/jmoiron/sqlx v1.2.0
github.com/prometheus/client_golang v1.0.0
github.com/sirupsen/logrus v1.7.0
github.com/spf13/cobra v1.1.1
github.com/spf13/viper v1.7.1
github.com/vulcanize/go-eth-state-node-iterator v0.0.1-alpha
github.com/vulcanize/go-eth-state-node-iterator v0.0.1-alpha.0.20211014064906-d23d01ed8191
)
replace github.com/ethereum/go-ethereum v1.9.11 => github.com/vulcanize/go-ethereum v1.9.11-statediff-0.0.8
replace github.com/ethereum/go-ethereum v1.10.11 => github.com/vulcanize/go-ethereum v1.10.11-statediff-0.0.27h
+620 -91
View File
File diff suppressed because it is too large Load Diff
+449
View File
@@ -0,0 +1,449 @@
{
"annotations": {
"list": [
{
"builtIn": 1,
"datasource": "-- Grafana --",
"enable": true,
"hide": true,
"iconColor": "rgba(0, 211, 255, 1)",
"name": "Annotations & Alerts",
"type": "dashboard"
}
]
},
"editable": true,
"gnetId": null,
"graphTooltip": 0,
"id": 4,
"iteration": 1635774999288,
"links": [
{
"icon": "info",
"tags": [],
"targetBlank": true,
"title": "Grafana Docs",
"tooltip": "",
"type": "link",
"url": "https://grafana.com/docs/grafana/latest/"
},
{
"icon": "info",
"tags": [],
"targetBlank": true,
"title": "Prometheus Docs",
"type": "link",
"url": "http://prometheus.io/docs/introduction/overview/"
}
],
"panels": [
{
"aliasColors": {},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "eth-statediff-service",
"fieldConfig": {
"defaults": {
"custom": {},
"links": [],
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": null
},
{
"color": "red",
"value": 80
}
]
}
},
"overrides": []
},
"fill": 1,
"fillGradient": 0,
"gridPos": {
"h": 6,
"w": 18,
"x": 0,
"y": 0
},
"hiddenSeries": false,
"id": 3,
"legend": {
"avg": false,
"current": false,
"max": false,
"min": false,
"show": true,
"total": false,
"values": false
},
"lines": true,
"linewidth": 1,
"links": [],
"nullPointMode": "null",
"percentage": false,
"pluginVersion": "7.1.5",
"pointradius": 2,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "rate(eth_statediff_service_processed_height{instance=\"$instance\"}[1m])",
"interval": "",
"legendFormat": "",
"refId": "A"
}
],
"thresholds": [],
"timeFrom": null,
"timeRegions": [],
"timeShift": null,
"title": "Instance block processed (rate-1m)",
"tooltip": {
"shared": true,
"sort": 0,
"value_type": "individual"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
},
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
},
{
"content": "<span style=\"font-family: 'Open Sans', 'Helvetica Neue', Helvetica; font-size: 25px;vertical-align: text-top;color: #bbbfc2;margin-left: 10px;\">Prometheus</span>\n\n<p style=\"margin-top: 10px;\">You're using Prometheus, an open-source systems monitoring and alerting toolkit originally built at SoundCloud. For more information, check out the <a href=\"https://grafana.com/\">Grafana</a> and <a href=\"http://prometheus.io/\">Prometheus</a> projects.</p>",
"datasource": null,
"editable": true,
"error": false,
"fieldConfig": {
"defaults": {
"custom": {}
},
"overrides": []
},
"gridPos": {
"h": 5,
"w": 6,
"x": 18,
"y": 0
},
"id": 9,
"links": [],
"mode": "html",
"options": {
"content": "<span style=\"font-family: 'Open Sans', 'Helvetica Neue', Helvetica; font-size: 25px;vertical-align: text-top;color: #bbbfc2;margin-left: 10px;\">Prometheus</span>\n\n<p style=\"margin-top: 10px;\">You're using Prometheus, an open-source systems monitoring and alerting toolkit originally built at SoundCloud. For more information, check out the <a href=\"https://grafana.com/\">Grafana</a> and <a href=\"http://prometheus.io/\">Prometheus</a> projects.</p>",
"mode": "html"
},
"pluginVersion": "7.1.0",
"style": {},
"title": "",
"transparent": true,
"type": "text"
},
{
"content": "#### Samples Ingested\nThis graph displays the count of samples ingested by the Prometheus server, as measured over the last 5 minutes, per time series in the range vector. When troubleshooting an issue on IRC or Github, this is often the first stat requested by the Prometheus team. ",
"datasource": null,
"editable": true,
"error": false,
"fieldConfig": {
"defaults": {
"custom": {}
},
"overrides": []
},
"gridPos": {
"h": 6,
"w": 4,
"x": 18,
"y": 5
},
"id": 8,
"links": [],
"mode": "markdown",
"options": {
"content": "#### Samples Ingested\nThis graph displays the count of samples ingested by the Prometheus server, as measured over the last 5 minutes, per time series in the range vector. When troubleshooting an issue on IRC or Github, this is often the first stat requested by the Prometheus team. ",
"mode": "markdown"
},
"pluginVersion": "7.1.0",
"style": {},
"title": "",
"transparent": true,
"type": "text"
},
{
"aliasColors": {
"Value": "red",
"aggr": "red"
},
"bars": false,
"dashLength": 10,
"dashes": false,
"datasource": "eth-statediff-service",
"fieldConfig": {
"defaults": {
"custom": {},
"links": [],
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": null
},
{
"color": "red",
"value": 80
}
]
}
},
"overrides": []
},
"fill": 1,
"fillGradient": 0,
"gridPos": {
"h": 6,
"w": 18,
"x": 0,
"y": 6
},
"hiddenSeries": false,
"id": 17,
"legend": {
"avg": false,
"current": false,
"max": false,
"min": false,
"show": true,
"total": false,
"values": false
},
"lines": true,
"linewidth": 1,
"links": [],
"nullPointMode": "null",
"percentage": false,
"pluginVersion": "7.1.5",
"pointradius": 2,
"points": false,
"renderer": "flot",
"seriesOverrides": [],
"spaceLength": 10,
"stack": false,
"steppedLine": false,
"targets": [
{
"expr": "sum(increase(eth_statediff_service_processed_height{instance=~\".+\"}[1m]))",
"interval": "",
"legendFormat": "aggr",
"refId": "C"
}
],
"thresholds": [],
"timeFrom": null,
"timeRegions": [],
"timeShift": null,
"title": "Aggregate block processed (rate-1m)",
"tooltip": {
"shared": true,
"sort": 0,
"value_type": "individual"
},
"type": "graph",
"xaxis": {
"buckets": null,
"mode": "time",
"name": null,
"show": true,
"values": []
},
"yaxes": [
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
},
{
"format": "short",
"label": null,
"logBase": 1,
"max": null,
"min": null,
"show": true
}
],
"yaxis": {
"align": false,
"alignLevel": null
}
},
{
"content": "#### Scrapes\nPrometheus scrapes metrics from instrumented jobs, either directly or via an intermediary push gateway for short-lived jobs. Target scrapes will show how frequently targets are scraped, as measured over the last 5 minutes, per time series in the range vector. Scrape Duration will show how long the scrapes are taking, with percentiles available as series. ",
"datasource": null,
"editable": true,
"error": false,
"fieldConfig": {
"defaults": {
"custom": {}
},
"overrides": []
},
"gridPos": {
"h": 7,
"w": 6,
"x": 18,
"y": 11
},
"id": 11,
"links": [],
"mode": "markdown",
"options": {
"content": "#### Scrapes\nPrometheus scrapes metrics from instrumented jobs, either directly or via an intermediary push gateway for short-lived jobs. Target scrapes will show how frequently targets are scraped, as measured over the last 5 minutes, per time series in the range vector. Scrape Duration will show how long the scrapes are taking, with percentiles available as series. ",
"mode": "markdown"
},
"pluginVersion": "7.1.0",
"style": {},
"title": "",
"transparent": true,
"type": "text"
},
{
"content": "#### Rule Evaluation Duration\nThis graph panel plots the duration for all evaluations to execute. The 50th percentile, 90th percentile and 99th percentile are shown as three separate series to help identify outliers that may be skewing the data.",
"datasource": null,
"editable": true,
"error": false,
"fieldConfig": {
"defaults": {
"custom": {}
},
"overrides": []
},
"gridPos": {
"h": 7,
"w": 6,
"x": 18,
"y": 18
},
"id": 15,
"links": [],
"mode": "markdown",
"options": {
"content": "#### Rule Evaluation Duration\nThis graph panel plots the duration for all evaluations to execute. The 50th percentile, 90th percentile and 99th percentile are shown as three separate series to help identify outliers that may be skewing the data.",
"mode": "markdown"
},
"pluginVersion": "7.1.0",
"style": {},
"title": "",
"transparent": true,
"type": "text"
}
],
"refresh": false,
"revision": "1.0",
"schemaVersion": 26,
"style": "dark",
"tags": [],
"templating": {
"list": [
{
"allValue": null,
"current": {
"selected": true,
"text": "localhost:8888",
"value": "localhost:8888"
},
"datasource": "eth-statediff-service",
"definition": "label_values(instance)",
"hide": 0,
"includeAll": false,
"label": "SelectInstance",
"multi": false,
"name": "instance",
"options": [],
"query": "label_values(instance)",
"refresh": 2,
"regex": "",
"skipUrlSync": false,
"sort": 1,
"tagValuesQuery": "",
"tags": [],
"tagsQuery": "",
"type": "query",
"useTags": false
}
]
},
"time": {
"from": "now-30m",
"to": "now"
},
"timepicker": {
"now": true,
"refresh_intervals": [
"5s",
"10s",
"30s",
"1m",
"5m",
"15m",
"30m",
"1h",
"2h",
"1d"
],
"time_options": [
"5m",
"15m",
"1h",
"6h",
"12h",
"24h",
"2d",
"7d",
"30d"
]
},
"timezone": "browser",
"title": "Prometheus Block Process Stats",
"uid": "q1FUJHKnz",
"version": 14
}
+7
View File
@@ -0,0 +1,7 @@
global:
scrape_interval: 10s
scrape_configs:
- job_name: 'eth-statediff-service'
static_configs:
- targets: ['localhost:8100']
+12 -2
View File
@@ -30,11 +30,11 @@ const APIVersion = "0.0.1"
// PublicStateDiffAPI provides an RPC interface
// that can be used to fetch historical diffs from leveldb directly
type PublicStateDiffAPI struct {
sds IService
sds StateDiffService
}
// NewPublicStateDiffAPI creates an rpc interface for the underlying statediff service
func NewPublicStateDiffAPI(sds IService) *PublicStateDiffAPI {
func NewPublicStateDiffAPI(sds StateDiffService) *PublicStateDiffAPI {
return &PublicStateDiffAPI{
sds: sds,
}
@@ -49,3 +49,13 @@ func (api *PublicStateDiffAPI) StateDiffAt(ctx context.Context, blockNumber uint
func (api *PublicStateDiffAPI) StateTrieAt(ctx context.Context, blockNumber uint64, params sd.Params) (*sd.Payload, error) {
return api.sds.StateTrieAt(blockNumber, params)
}
// WriteStateDiffAt writes a state diff object directly to DB at the specific blockheight
func (api *PublicStateDiffAPI) WriteStateDiffAt(ctx context.Context, blockNumber uint64, params sd.Params) error {
return api.sds.WriteStateDiffAt(blockNumber, params)
}
// WriteStateDiffsInRange writes the state diff objects for the provided block range, with the provided params
func (api *PublicStateDiffAPI) WriteStateDiffsInRange(ctx context.Context, start, stop uint64, params sd.Params) error {
return api.sds.WriteStateDiffsInRange(start, stop, params)
}
+376 -294
View File
@@ -31,9 +31,12 @@ import (
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/trie"
sd "github.com/ethereum/go-ethereum/statediff"
sdtrie "github.com/ethereum/go-ethereum/statediff/trie_helpers"
sdtypes "github.com/ethereum/go-ethereum/statediff/types"
"github.com/ethereum/go-ethereum/trie"
"github.com/sirupsen/logrus"
iter "github.com/vulcanize/go-eth-state-node-iterator"
)
@@ -46,8 +49,9 @@ var (
// Builder interface exposes the method for building a state diff between two blocks
type Builder interface {
BuildStateDiffObject(args sd.Args, params sd.Params) (sd.StateObject, error)
BuildStateTrieObject(current *types.Block) (sd.StateObject, error)
BuildStateDiffObject(args sd.Args, params sd.Params) (sdtypes.StateObject, error)
BuildStateTrieObject(current *types.Block) (sdtypes.StateObject, error)
WriteStateDiffObject(args sdtypes.StateRoots, params sd.Params, output sdtypes.StateNodeSink, codeOutput sdtypes.CodeSink) error
}
type builder struct {
@@ -59,6 +63,48 @@ type iterPair struct {
older, newer trie.NodeIterator
}
func resolveNode(it trie.NodeIterator, trieDB *trie.Database) (sdtypes.StateNode, []interface{}, error) {
nodePath := make([]byte, len(it.Path()))
copy(nodePath, it.Path())
node, err := trieDB.Node(it.Hash())
if err != nil {
return sdtypes.StateNode{}, nil, err
}
var nodeElements []interface{}
if err := rlp.DecodeBytes(node, &nodeElements); err != nil {
return sdtypes.StateNode{}, nil, err
}
ty, err := sdtrie.CheckKeyType(nodeElements)
if err != nil {
return sdtypes.StateNode{}, nil, err
}
return sdtypes.StateNode{
NodeType: ty,
Path: nodePath,
NodeValue: node,
}, nodeElements, nil
}
// convenience
func stateNodeAppender(nodes *[]sdtypes.StateNode) sdtypes.StateNodeSink {
return func(node sdtypes.StateNode) error {
*nodes = append(*nodes, node)
return nil
}
}
func storageNodeAppender(nodes *[]sdtypes.StorageNode) sdtypes.StorageNodeSink {
return func(node sdtypes.StorageNode) error {
*nodes = append(*nodes, node)
return nil
}
}
func codeMappingAppender(data *[]sdtypes.CodeAndCodeHash) sdtypes.CodeSink {
return func(c sdtypes.CodeAndCodeHash) error {
*data = append(*data, c)
return nil
}
}
// NewBuilder is used to create a statediff builder
func NewBuilder(stateCache state.Database, workers uint) (Builder, error) {
if workers == 0 {
@@ -74,17 +120,17 @@ func NewBuilder(stateCache state.Database, workers uint) (Builder, error) {
}
// BuildStateTrieObject builds a state trie object from the provided block
func (sdb *builder) BuildStateTrieObject(current *types.Block) (sd.StateObject, error) {
func (sdb *builder) BuildStateTrieObject(current *types.Block) (sdtypes.StateObject, error) {
currentTrie, err := sdb.stateCache.OpenTrie(current.Root())
if err != nil {
return sd.StateObject{}, fmt.Errorf("error creating trie for block %d: %v", current.Number(), err)
return sdtypes.StateObject{}, fmt.Errorf("error creating trie for block %d: %v", current.Number(), err)
}
it := currentTrie.NodeIterator([]byte{})
stateNodes, codeAndCodeHashes, err := sdb.buildStateTrie(it)
if err != nil {
return sd.StateObject{}, fmt.Errorf("error collecting state nodes for block %d: %v", current.Number(), err)
return sdtypes.StateObject{}, fmt.Errorf("error collecting state nodes for block %d: %v", current.Number(), err)
}
return sd.StateObject{
return sdtypes.StateObject{
BlockNumber: current.Number(),
BlockHash: current.Hash(),
Nodes: stateNodes,
@@ -92,68 +138,32 @@ func (sdb *builder) BuildStateTrieObject(current *types.Block) (sd.StateObject,
}, nil
}
func resolveNode(it trie.NodeIterator, trieDB *trie.Database) (sd.StateNode, []interface{}, error) {
nodePath := make([]byte, len(it.Path()))
copy(nodePath, it.Path())
node, err := trieDB.Node(it.Hash())
if err != nil {
return sd.StateNode{}, nil, err
}
var nodeElements []interface{}
if err := rlp.DecodeBytes(node, &nodeElements); err != nil {
return sd.StateNode{}, nil, err
}
ty, err := sd.CheckKeyType(nodeElements)
if err != nil {
return sd.StateNode{}, nil, err
}
return sd.StateNode{
NodeType: ty,
Path: nodePath,
NodeValue: node,
}, nodeElements, nil
}
func (sdb *builder) buildStateTrie(it trie.NodeIterator) ([]sd.StateNode, []sd.CodeAndCodeHash, error) {
stateNodes := make([]sd.StateNode, 0)
codeAndCodeHashes := make([]sd.CodeAndCodeHash, 0)
func (sdb *builder) buildStateTrie(it trie.NodeIterator) ([]sdtypes.StateNode, []sdtypes.CodeAndCodeHash, error) {
stateNodes := make([]sdtypes.StateNode, 0)
codeAndCodeHashes := make([]sdtypes.CodeAndCodeHash, 0)
for it.Next(true) {
// skip value nodes and null nodes
// skip value nodes
if it.Leaf() || bytes.Equal(nullHashBytes, it.Hash().Bytes()) {
continue
}
nodePath := make([]byte, len(it.Path()))
copy(nodePath, it.Path())
node, err := sdb.stateCache.TrieDB().Node(it.Hash())
node, nodeElements, err := resolveNode(it, sdb.stateCache.TrieDB())
if err != nil {
return nil, nil, err
}
var nodeElements []interface{}
if err := rlp.DecodeBytes(node, &nodeElements); err != nil {
return nil, nil, err
}
ty, err := sd.CheckKeyType(nodeElements)
if err != nil {
return nil, nil, err
}
switch ty {
case sd.Leaf:
var account state.Account
switch node.NodeType {
case sdtypes.Leaf:
var account types.StateAccount
if err := rlp.DecodeBytes(nodeElements[1].([]byte), &account); err != nil {
return nil, nil, fmt.Errorf("error decoding account for leaf node at path %x nerror: %v", nodePath, err)
return nil, nil, fmt.Errorf("error decoding account for leaf node at path %x nerror: %v", node.Path, err)
}
partialPath := trie.CompactToHex(nodeElements[0].([]byte))
valueNodePath := append(nodePath, partialPath...)
valueNodePath := append(node.Path, partialPath...)
encodedPath := trie.HexToCompact(valueNodePath)
leafKey := encodedPath[1:]
node := sd.StateNode{
NodeType: ty,
Path: nodePath,
LeafKey: leafKey,
NodeValue: node,
}
node.LeafKey = leafKey
if !bytes.Equal(account.CodeHash, nullCodeHash) {
storageNodes, err := sdb.buildStorageNodesEventual(account.Root, nil, true)
var storageNodes []sdtypes.StorageNode
err := sdb.buildStorageNodesEventual(account.Root, nil, true, storageNodeAppender(&storageNodes))
if err != nil {
return nil, nil, fmt.Errorf("failed building eventual storage diffs for account %+v\r\nerror: %v", account, err)
}
@@ -164,27 +174,41 @@ func (sdb *builder) buildStateTrie(it trie.NodeIterator) ([]sd.StateNode, []sd.C
if err != nil {
return nil, nil, fmt.Errorf("failed to retrieve code for codehash %s\r\n error: %v", codeHash.String(), err)
}
codeAndCodeHashes = append(codeAndCodeHashes, sd.CodeAndCodeHash{
codeAndCodeHashes = append(codeAndCodeHashes, sdtypes.CodeAndCodeHash{
Hash: codeHash,
Code: code,
})
}
stateNodes = append(stateNodes, node)
case sd.Extension, sd.Branch:
stateNodes = append(stateNodes, sd.StateNode{
NodeType: ty,
Path: nodePath,
NodeValue: node,
})
case sdtypes.Extension, sdtypes.Branch:
stateNodes = append(stateNodes, node)
default:
return nil, nil, fmt.Errorf("unexpected node type %s", ty)
return nil, nil, fmt.Errorf("unexpected node type %s", node.NodeType)
}
}
return stateNodes, codeAndCodeHashes, it.Error()
}
// BuildStateDiff builds a statediff object from two blocks and the provided parameters
func (sdb *builder) BuildStateDiffObject(args sd.Args, params sd.Params) (sd.StateObject, error) {
// BuildStateDiffObject builds a statediff object from two blocks and the provided parameters
func (sdb *builder) BuildStateDiffObject(args sd.Args, params sd.Params) (sdtypes.StateObject, error) {
var stateNodes []sdtypes.StateNode
var codeAndCodeHashes []sdtypes.CodeAndCodeHash
err := sdb.WriteStateDiffObject(
sdtypes.StateRoots{OldStateRoot: args.OldStateRoot, NewStateRoot: args.NewStateRoot},
params, stateNodeAppender(&stateNodes), codeMappingAppender(&codeAndCodeHashes))
if err != nil {
return sdtypes.StateObject{}, err
}
return sdtypes.StateObject{
BlockHash: args.BlockHash,
BlockNumber: args.BlockNumber,
Nodes: stateNodes,
CodeAndCodeHashes: codeAndCodeHashes,
}, nil
}
// WriteStateDiffObject writes a statediff object to output callback
func (sdb *builder) WriteStateDiffObject(args sdtypes.StateRoots, params sd.Params, output sdtypes.StateNodeSink, codeOutput sdtypes.CodeSink) error {
if len(params.WatchedAddresses) > 0 {
// if we are watching only specific accounts then we are only diffing leaf nodes
log.Info("Ignoring intermediate state nodes because WatchedAddresses was passed")
@@ -194,95 +218,97 @@ func (sdb *builder) BuildStateDiffObject(args sd.Args, params sd.Params) (sd.Sta
// Load tries for old and new states
oldTrie, err := sdb.stateCache.OpenTrie(args.OldStateRoot)
if err != nil {
return sd.StateObject{}, fmt.Errorf("error creating trie for old state root: %v", err)
return fmt.Errorf("error creating trie for old state root: %v", err)
}
newTrie, err := sdb.stateCache.OpenTrie(args.NewStateRoot)
if err != nil {
return sd.StateObject{}, fmt.Errorf("error creating trie for new state root: %v", err)
return fmt.Errorf("error creating trie for new state root: %v", err)
}
// Split old and new tries into corresponding subtrie iterators
oldIterFac := iter.NewSubtrieIteratorFactory(oldTrie, sdb.numWorkers)
newIterFac := iter.NewSubtrieIteratorFactory(newTrie, sdb.numWorkers)
iterChan := make(chan []iterPair, sdb.numWorkers)
// Create iterators ahead of time to avoid race condition in state.Trie access
// We do two state iterations per subtrie: one for new/updated nodes,
// one for deleted/updated nodes; prepare 2 iterator instances for each task
var iterPairs [][]iterPair
for i := uint(0); i < sdb.numWorkers; i++ {
// two state iterations per diff build
iterChan <- []iterPair{
iterPair{
older: oldIterFac.IteratorAt(i),
newer: newIterFac.IteratorAt(i),
},
iterPair{
older: oldIterFac.IteratorAt(i),
newer: newIterFac.IteratorAt(i),
},
iterPairs = append(iterPairs, []iterPair{
{older: oldIterFac.IteratorAt(i), newer: newIterFac.IteratorAt(i)},
{older: oldIterFac.IteratorAt(i), newer: newIterFac.IteratorAt(i)},
})
}
// Dispatch workers to process trie data; sync and collect results here via channels
nodeChan := make(chan sdtypes.StateNode)
codeChan := make(chan sdtypes.CodeAndCodeHash)
go func() {
nodeSender := func(node sdtypes.StateNode) error { nodeChan <- node; return nil }
codeSender := func(code sdtypes.CodeAndCodeHash) error { codeChan <- code; return nil }
var wg sync.WaitGroup
for w := uint(0); w < sdb.numWorkers; w++ {
wg.Add(1)
go func(worker uint) {
defer wg.Done()
if err := sdb.buildStateDiff(iterPairs[worker], params, nodeSender, codeSender); err != nil {
logrus.Errorf("buildStateDiff error for worker %d, pparams %+v", worker, params)
}
}(w)
}
wg.Wait()
close(nodeChan)
close(codeChan)
}()
for nodeChan != nil || codeChan != nil {
select {
case node, more := <-nodeChan:
if more {
if err := output(node); err != nil {
return err
}
} else {
nodeChan = nil
}
case codeAndCodeHash, more := <-codeChan:
if more {
if err := codeOutput(codeAndCodeHash); err != nil {
return err
}
} else {
codeChan = nil
}
}
}
type packet struct {
nodes []sd.StateNode
codes []sd.CodeAndCodeHash
}
packetChan := make(chan packet)
var wg sync.WaitGroup
for w := uint(0); w < sdb.numWorkers; w++ {
wg.Add(1)
go func(iterChan <-chan []iterPair) error {
defer wg.Done()
if iters, more := <-iterChan; more {
subtrieNodes, subtrieCodes, err := sdb.buildStateDiff(iters, params)
if err != nil {
return err
}
packetChan <- packet{
nodes: subtrieNodes,
codes: subtrieCodes,
}
}
return nil
}(iterChan)
}
go func() {
defer close(packetChan)
defer close(iterChan)
wg.Wait()
}()
stateNodes := make([]sd.StateNode, 0)
codeAndCodeHashes := make([]sd.CodeAndCodeHash, 0)
for packet := range packetChan {
stateNodes = append(stateNodes, packet.nodes...)
codeAndCodeHashes = append(codeAndCodeHashes, packet.codes...)
}
return sd.StateObject{
BlockHash: args.BlockHash,
BlockNumber: args.BlockNumber,
Nodes: stateNodes,
CodeAndCodeHashes: codeAndCodeHashes,
}, nil
return nil
}
func (sdb *builder) buildStateDiff(args []iterPair, params sd.Params) ([]sd.StateNode, []sd.CodeAndCodeHash, error) {
func (sdb *builder) buildStateDiff(args []iterPair, params sd.Params, output sdtypes.StateNodeSink, codeOutput sdtypes.CodeSink) error {
// collect a slice of all the intermediate nodes that were touched and exist at B
// a map of their leafkey to all the accounts that were touched and exist at B
// and a slice of all the paths for the nodes in both of the above sets
createdOrUpdatedIntermediateNodes, diffAccountsAtB, diffPathsAtB, err := sdb.createdAndUpdatedState(
args[0], params.WatchedAddresses, params.IntermediateStateNodes)
var diffAccountsAtB AccountMap
var diffPathsAtB map[string]bool
var err error
if params.IntermediateStateNodes {
diffAccountsAtB, diffPathsAtB, err = sdb.createdAndUpdatedStateWithIntermediateNodes(args[0], output)
} else {
diffAccountsAtB, diffPathsAtB, err = sdb.createdAndUpdatedState(args[0], params.WatchedAddresses)
}
if err != nil {
return nil, nil, fmt.Errorf("error collecting createdAndUpdatedNodes: %v", err)
return fmt.Errorf("error collecting createdAndUpdatedNodes: %v", err)
}
// collect a slice of all the nodes that existed at a path in A that doesn't exist in B
// a map of their leafkey to all the accounts that were touched and exist at A
emptiedPaths, diffAccountsAtA, err := sdb.deletedOrUpdatedState(args[1], diffPathsAtB)
diffAccountsAtA, err := sdb.deletedOrUpdatedState(args[1], diffPathsAtB, output)
if err != nil {
return nil, nil, fmt.Errorf("error collecting deletedOrUpdatedNodes: %v", err)
return fmt.Errorf("error collecting deletedOrUpdatedNodes: %v", err)
}
// collect and sort the leafkey keys for both account mappings into a slice
@@ -295,57 +321,43 @@ func (sdb *builder) buildStateDiff(args []iterPair, params sd.Params) ([]sd.Stat
// and leaving the truly created or deleted keys in place
updatedKeys := findIntersection(createKeys, deleteKeys)
// build the diff nodes for the updated accounts using the mappings at both A and B
// as directed by the keys found as the intersection of the two
updatedAccounts, err := sdb.buildAccountUpdates(
// build the diff nodes for the updated accounts using the mappings at both A and B as directed by the keys found as the intersection of the two
err = sdb.buildAccountUpdates(
diffAccountsAtB, diffAccountsAtA, updatedKeys,
params.WatchedStorageSlots, params.IntermediateStorageNodes)
params.WatchedStorageSlots, params.IntermediateStorageNodes, output)
if err != nil {
return nil, nil, fmt.Errorf("error building diff for updated accounts: %v", err)
return fmt.Errorf("error building diff for updated accounts: %v", err)
}
// build the diff nodes for created accounts
createdAccounts, codeAndCodeHashes, err := sdb.buildAccountCreations(
diffAccountsAtB, params.WatchedStorageSlots, params.IntermediateStorageNodes)
err = sdb.buildAccountCreations(diffAccountsAtB, params.WatchedStorageSlots, params.IntermediateStorageNodes, output, codeOutput)
if err != nil {
return nil, nil, fmt.Errorf("error building diff for created accounts: %v", err)
return fmt.Errorf("error building diff for created accounts: %v", err)
}
// assemble all of the nodes into the statediff object, including the intermediate nodes
nodes := append(
append(
append(updatedAccounts, createdAccounts...),
createdOrUpdatedIntermediateNodes...,
),
emptiedPaths...)
return nodes, codeAndCodeHashes, nil
return nil
}
// createdAndUpdatedState returns
// a slice of all the intermediate nodes that exist in a different state at B than A
// a mapping of their leafkeys to all the accounts that exist in a different state at B than A
// and a slice of the paths for all of the nodes included in both
func (sdb *builder) createdAndUpdatedState(iters iterPair, watchedAddresses []common.Address, intermediates bool) ([]sd.StateNode, AccountMap, map[string]bool, error) {
createdOrUpdatedIntermediateNodes := make([]sd.StateNode, 0)
func (sdb *builder) createdAndUpdatedState(iters iterPair, watchedAddresses []common.Address) (AccountMap, map[string]bool, error) {
diffPathsAtB := make(map[string]bool)
diffAcountsAtB := make(AccountMap)
it, _ := trie.NewDifferenceIterator(iters.older, iters.newer)
for it.Next(true) {
// skip value nodes and null nodes
if it.Leaf() || bytes.Equal(nullHashBytes, it.Hash().Bytes()) {
continue
}
node, nodeElements, err := resolveNode(it, sdb.stateCache.TrieDB())
if err != nil {
return nil, nil, nil, err
return nil, nil, err
}
switch node.NodeType {
case sd.Leaf:
case sdtypes.Leaf:
// created vs updated is important for leaf nodes since we need to diff their storage
// so we need to map all changed accounts at B to their leafkey, since account can change pathes but not leafkey
var account state.Account
var account types.StateAccount
if err := rlp.DecodeBytes(nodeElements[1].([]byte), &account); err != nil {
return nil, nil, nil, fmt.Errorf("error decoding account for leaf node at path %x nerror: %v", node.Path, err)
return nil, nil, fmt.Errorf("error decoding account for leaf node at path %x nerror: %v", node.Path, err)
}
partialPath := trie.CompactToHex(nodeElements[0].([]byte))
valueNodePath := append(node.Path, partialPath...)
@@ -360,33 +372,22 @@ func (sdb *builder) createdAndUpdatedState(iters iterPair, watchedAddresses []co
Account: &account,
}
}
case sd.Extension, sd.Branch:
// create a diff for any intermediate node that has changed at b
// created vs updated makes no difference for intermediate nodes since we do not need to diff storage
if intermediates {
createdOrUpdatedIntermediateNodes = append(createdOrUpdatedIntermediateNodes, sd.StateNode{
NodeType: node.NodeType,
Path: node.Path,
NodeValue: node.NodeValue,
})
}
default:
return nil, nil, nil, fmt.Errorf("unexpected node type %s", node.NodeType)
}
// add both intermediate and leaf node paths to the list of diffPathsAtB
diffPathsAtB[common.Bytes2Hex(node.Path)] = true
}
return createdOrUpdatedIntermediateNodes, diffAcountsAtB, diffPathsAtB, it.Error()
return diffAcountsAtB, diffPathsAtB, it.Error()
}
// deletedOrUpdatedState returns a slice of all the paths that are emptied at B
// and a mapping of their leafkeys to all the accounts that exist in a different state at A than B
func (sdb *builder) deletedOrUpdatedState(iters iterPair, diffPathsAtB map[string]bool) ([]sd.StateNode, AccountMap, error) {
emptiedPaths := make([]sd.StateNode, 0)
diffAccountAtA := make(AccountMap)
it, _ := trie.NewDifferenceIterator(iters.newer, iters.older)
// createdAndUpdatedStateWithIntermediateNodes returns
// a slice of all the intermediate nodes that exist in a different state at B than A
// a mapping of their leafkeys to all the accounts that exist in a different state at B than A
// and a slice of the paths for all of the nodes included in both
func (sdb *builder) createdAndUpdatedStateWithIntermediateNodes(iters iterPair, output sdtypes.StateNodeSink) (AccountMap, map[string]bool, error) {
diffPathsAtB := make(map[string]bool)
diffAcountsAtB := make(AccountMap)
it, _ := trie.NewDifferenceIterator(iters.older, iters.newer)
for it.Next(true) {
// skip value nodes and null nodes
if it.Leaf() || bytes.Equal(nullHashBytes, it.Hash().Bytes()) {
continue
}
@@ -394,20 +395,11 @@ func (sdb *builder) deletedOrUpdatedState(iters iterPair, diffPathsAtB map[strin
if err != nil {
return nil, nil, err
}
// if this nodePath did not show up in diffPathsAtB
// that means the node at this path was deleted (or moved) in B
// emit an empty "removed" diff to signify as such
if _, ok := diffPathsAtB[common.Bytes2Hex(node.Path)]; !ok {
emptiedPaths = append(emptiedPaths, sd.StateNode{
Path: node.Path,
NodeValue: []byte{},
NodeType: sd.Removed,
})
}
switch node.NodeType {
case sd.Leaf:
// map all different accounts at A to their leafkey
var account state.Account
case sdtypes.Leaf:
// created vs updated is important for leaf nodes since we need to diff their storage
// so we need to map all changed accounts at B to their leafkey, since account can change paths but not leafkey
var account types.StateAccount
if err := rlp.DecodeBytes(nodeElements[1].([]byte), &account); err != nil {
return nil, nil, fmt.Errorf("error decoding account for leaf node at path %x nerror: %v", node.Path, err)
}
@@ -415,6 +407,56 @@ func (sdb *builder) deletedOrUpdatedState(iters iterPair, diffPathsAtB map[strin
valueNodePath := append(node.Path, partialPath...)
encodedPath := trie.HexToCompact(valueNodePath)
leafKey := encodedPath[1:]
diffAcountsAtB[common.Bytes2Hex(leafKey)] = accountWrapper{
NodeType: node.NodeType,
Path: node.Path,
NodeValue: node.NodeValue,
LeafKey: leafKey,
Account: &account,
}
case sdtypes.Extension, sdtypes.Branch:
// create a diff for any intermediate node that has changed at b
// created vs updated makes no difference for intermediate nodes since we do not need to diff storage
if err := output(sdtypes.StateNode{
NodeType: node.NodeType,
Path: node.Path,
NodeValue: node.NodeValue,
}); err != nil {
return nil, nil, err
}
default:
return nil, nil, fmt.Errorf("unexpected node type %s", node.NodeType)
}
// add both intermediate and leaf node paths to the list of diffPathsAtB
diffPathsAtB[common.Bytes2Hex(node.Path)] = true
}
return diffAcountsAtB, diffPathsAtB, it.Error()
}
// deletedOrUpdatedState returns a slice of all the paths that are emptied at B
// and a mapping of their leafkeys to all the accounts that exist in a different state at A than B
func (sdb *builder) deletedOrUpdatedState(iters iterPair, diffPathsAtB map[string]bool, output sdtypes.StateNodeSink) (AccountMap, error) {
diffAccountAtA := make(AccountMap)
it, _ := trie.NewDifferenceIterator(iters.newer, iters.older)
for it.Next(true) {
if it.Leaf() || bytes.Equal(nullHashBytes, it.Hash().Bytes()) {
continue
}
node, nodeElements, err := resolveNode(it, sdb.stateCache.TrieDB())
if err != nil {
return nil, err
}
switch node.NodeType {
case sdtypes.Leaf:
// map all different accounts at A to their leafkey
var account types.StateAccount
if err := rlp.DecodeBytes(nodeElements[1].([]byte), &account); err != nil {
return nil, fmt.Errorf("error decoding account for leaf node at path %x nerror: %v", node.Path, err)
}
partialPath := trie.CompactToHex(nodeElements[0].([]byte))
valueNodePath := append(node.Path, partialPath...)
encodedPath := trie.HexToCompact(valueNodePath)
leafKey := encodedPath[1:]
diffAccountAtA[common.Bytes2Hex(leafKey)] = accountWrapper{
NodeType: node.NodeType,
Path: node.Path,
@@ -422,55 +464,79 @@ func (sdb *builder) deletedOrUpdatedState(iters iterPair, diffPathsAtB map[strin
LeafKey: leafKey,
Account: &account,
}
case sd.Extension, sd.Branch:
// if this node's path did not show up in diffPathsAtB
// that means the node at this path was deleted (or moved) in B
// emit an empty "removed" diff to signify as such
if _, ok := diffPathsAtB[common.Bytes2Hex(node.Path)]; !ok {
if err := output(sdtypes.StateNode{
Path: node.Path,
NodeValue: []byte{},
NodeType: sdtypes.Removed,
LeafKey: leafKey,
}); err != nil {
return nil, err
}
}
case sdtypes.Extension, sdtypes.Branch:
// if this node's path did not show up in diffPathsAtB
// that means the node at this path was deleted (or moved) in B
// emit an empty "removed" diff to signify as such
if _, ok := diffPathsAtB[common.Bytes2Hex(node.Path)]; !ok {
if err := output(sdtypes.StateNode{
Path: node.Path,
NodeValue: []byte{},
NodeType: sdtypes.Removed,
}); err != nil {
return nil, err
}
}
// fall through, we did everything we need to do with these node types
default:
return nil, nil, fmt.Errorf("unexpected node type %s", node.NodeType)
return nil, fmt.Errorf("unexpected node type %s", node.NodeType)
}
}
return emptiedPaths, diffAccountAtA, it.Error()
return diffAccountAtA, it.Error()
}
// buildAccountUpdates uses the account diffs maps for A => B and B => A and the known intersection of their leafkeys
// to generate the statediff node objects for all of the accounts that existed at both A and B but in different states
// needs to be called before building account creations and deletions as this mutates
// those account maps to remove the accounts which were updated
func (sdb *builder) buildAccountUpdates(creations, deletions AccountMap, updatedKeys []string, watchedStorageKeys []common.Hash, intermediateStorageNodes bool) ([]sd.StateNode, error) {
updatedAccounts := make([]sd.StateNode, 0, len(updatedKeys))
func (sdb *builder) buildAccountUpdates(creations, deletions AccountMap, updatedKeys []string, watchedStorageKeys []common.Hash, intermediateStorageNodes bool, output sdtypes.StateNodeSink) error {
var err error
for _, key := range updatedKeys {
createdAcc := creations[key]
deletedAcc := deletions[key]
var storageDiffs []sd.StorageNode
var storageDiffs []sdtypes.StorageNode
if deletedAcc.Account != nil && createdAcc.Account != nil {
oldSR := deletedAcc.Account.Root
newSR := createdAcc.Account.Root
storageDiffs, err = sdb.buildStorageNodesIncremental(oldSR, newSR, watchedStorageKeys, intermediateStorageNodes)
err = sdb.buildStorageNodesIncremental(oldSR, newSR, watchedStorageKeys, intermediateStorageNodes, storageNodeAppender(&storageDiffs))
if err != nil {
return nil, fmt.Errorf("failed building incremental storage diffs for account with leafkey %s\r\nerror: %v", key, err)
return fmt.Errorf("failed building incremental storage diffs for account with leafkey %s\r\nerror: %v", key, err)
}
}
updatedAccounts = append(updatedAccounts, sd.StateNode{
if err = output(sdtypes.StateNode{
NodeType: createdAcc.NodeType,
Path: createdAcc.Path,
NodeValue: createdAcc.NodeValue,
LeafKey: createdAcc.LeafKey,
StorageNodes: storageDiffs,
})
}); err != nil {
return err
}
delete(creations, key)
delete(deletions, key)
}
return updatedAccounts, nil
return nil
}
// buildAccountCreations returns the statediff node objects for all the accounts that exist at B but not at A
// it also returns the code and codehash for created contract accounts
func (sdb *builder) buildAccountCreations(accounts AccountMap, watchedStorageKeys []common.Hash, intermediateStorageNodes bool) ([]sd.StateNode, []sd.CodeAndCodeHash, error) {
accountDiffs := make([]sd.StateNode, 0, len(accounts))
codeAndCodeHashes := make([]sd.CodeAndCodeHash, 0)
func (sdb *builder) buildAccountCreations(accounts AccountMap, watchedStorageKeys []common.Hash, intermediateStorageNodes bool, output sdtypes.StateNodeSink, codeOutput sdtypes.CodeSink) error {
for _, val := range accounts {
diff := sd.StateNode{
diff := sdtypes.StateNode{
NodeType: val.NodeType,
Path: val.Path,
LeafKey: val.LeafKey,
@@ -478,161 +544,128 @@ func (sdb *builder) buildAccountCreations(accounts AccountMap, watchedStorageKey
}
if !bytes.Equal(val.Account.CodeHash, nullCodeHash) {
// For contract creations, any storage node contained is a diff
storageDiffs, err := sdb.buildStorageNodesEventual(val.Account.Root, watchedStorageKeys, intermediateStorageNodes)
var storageDiffs []sdtypes.StorageNode
err := sdb.buildStorageNodesEventual(val.Account.Root, watchedStorageKeys, intermediateStorageNodes, storageNodeAppender(&storageDiffs))
if err != nil {
return nil, nil, fmt.Errorf("failed building eventual storage diffs for node %x\r\nerror: %v", val.Path, err)
return fmt.Errorf("failed building eventual storage diffs for node %x\r\nerror: %v", val.Path, err)
}
diff.StorageNodes = storageDiffs
// emit codehash => code mappings for new contracts
// emit codehash => code mappings for code
codeHash := common.BytesToHash(val.Account.CodeHash)
code, err := sdb.stateCache.ContractCode(common.Hash{}, codeHash)
if err != nil {
return nil, nil, fmt.Errorf("failed to retrieve code for codehash %s\r\n error: %v", codeHash.String(), err)
return fmt.Errorf("failed to retrieve code for codehash %s\r\n error: %v", codeHash.String(), err)
}
codeAndCodeHashes = append(codeAndCodeHashes, sd.CodeAndCodeHash{
if err := codeOutput(sdtypes.CodeAndCodeHash{
Hash: codeHash,
Code: code,
})
}); err != nil {
return err
}
}
if err := output(diff); err != nil {
return err
}
accountDiffs = append(accountDiffs, diff)
}
return accountDiffs, codeAndCodeHashes, nil
return nil
}
// buildStorageNodesEventual builds the storage diff node objects for a created account
// i.e. it returns all the storage nodes at this state, since there is no previous state
func (sdb *builder) buildStorageNodesEventual(sr common.Hash, watchedStorageKeys []common.Hash, intermediateNodes bool) ([]sd.StorageNode, error) {
func (sdb *builder) buildStorageNodesEventual(sr common.Hash, watchedStorageKeys []common.Hash, intermediateNodes bool, output sdtypes.StorageNodeSink) error {
if bytes.Equal(sr.Bytes(), emptyContractRoot.Bytes()) {
return nil, nil
return nil
}
log.Debug("Storage Root For Eventual Diff", "root", sr, sr.Hex())
log.Debug("Storage Root For Eventual Diff", "root", sr.Hex())
sTrie, err := sdb.stateCache.OpenTrie(sr)
if err != nil {
log.Info("error in build storage diff eventual", "error", err)
return nil, err
return err
}
it := sTrie.NodeIterator(make([]byte, 0))
return sdb.buildStorageNodesFromTrie(it, watchedStorageKeys, intermediateNodes)
err = sdb.buildStorageNodesFromTrie(it, watchedStorageKeys, intermediateNodes, output)
if err != nil {
return err
}
return nil
}
// buildStorageNodesFromTrie returns all the storage diff node objects in the provided node iterator
// if any storage keys are provided it will only return those leaf nodes
// including intermediate nodes can be turned on or off
func (sdb *builder) buildStorageNodesFromTrie(it trie.NodeIterator, watchedStorageKeys []common.Hash, intermediateNodes bool) ([]sd.StorageNode, error) {
storageDiffs := make([]sd.StorageNode, 0)
func (sdb *builder) buildStorageNodesFromTrie(it trie.NodeIterator, watchedStorageKeys []common.Hash, intermediateNodes bool, output sdtypes.StorageNodeSink) error {
for it.Next(true) {
// skip value nodes and null nodes
if it.Leaf() || bytes.Equal(nullHashBytes, it.Hash().Bytes()) {
continue
}
node, nodeElements, err := resolveNode(it, sdb.stateCache.TrieDB())
if err != nil {
return nil, err
return err
}
switch node.NodeType {
case sd.Leaf:
case sdtypes.Leaf:
partialPath := trie.CompactToHex(nodeElements[0].([]byte))
valueNodePath := append(node.Path, partialPath...)
encodedPath := trie.HexToCompact(valueNodePath)
leafKey := encodedPath[1:]
if isWatchedStorageKey(watchedStorageKeys, leafKey) {
storageDiffs = append(storageDiffs, sd.StorageNode{
if err := output(sdtypes.StorageNode{
NodeType: node.NodeType,
Path: node.Path,
NodeValue: node.NodeValue,
LeafKey: leafKey,
})
}); err != nil {
return err
}
}
case sd.Extension, sd.Branch:
case sdtypes.Extension, sdtypes.Branch:
if intermediateNodes {
storageDiffs = append(storageDiffs, sd.StorageNode{
if err := output(sdtypes.StorageNode{
NodeType: node.NodeType,
Path: node.Path,
NodeValue: node.NodeValue,
})
}); err != nil {
return err
}
}
default:
return nil, fmt.Errorf("unexpected node type %s", node.NodeType)
return fmt.Errorf("unexpected node type %s", node.NodeType)
}
}
return storageDiffs, it.Error()
return it.Error()
}
// buildStorageNodesIncremental builds the storage diff node objects for all nodes that exist in a different state at B than A
func (sdb *builder) buildStorageNodesIncremental(oldSR common.Hash, newSR common.Hash, watchedStorageKeys []common.Hash, intermediateNodes bool) ([]sd.StorageNode, error) {
func (sdb *builder) buildStorageNodesIncremental(oldSR common.Hash, newSR common.Hash, watchedStorageKeys []common.Hash, intermediateNodes bool, output sdtypes.StorageNodeSink) error {
if bytes.Equal(newSR.Bytes(), oldSR.Bytes()) {
return nil, nil
return nil
}
log.Debug("Storage Roots for Incremental Diff", "old", oldSR.Hex(), "new", newSR.Hex())
oldTrie, err := sdb.stateCache.OpenTrie(oldSR)
if err != nil {
return nil, err
return err
}
newTrie, err := sdb.stateCache.OpenTrie(newSR)
if err != nil {
return nil, err
return err
}
createdOrUpdatedStorage, diffPathsAtB, err := sdb.createdAndUpdatedStorage(oldTrie.NodeIterator([]byte{}), newTrie.NodeIterator([]byte{}), watchedStorageKeys, intermediateNodes)
diffPathsAtB, err := sdb.createdAndUpdatedStorage(oldTrie.NodeIterator([]byte{}), newTrie.NodeIterator([]byte{}), watchedStorageKeys, intermediateNodes, output)
if err != nil {
return nil, err
return err
}
deletedStorage, err := sdb.deletedOrUpdatedStorage(oldTrie.NodeIterator([]byte{}), newTrie.NodeIterator([]byte{}), diffPathsAtB, watchedStorageKeys, intermediateNodes)
err = sdb.deletedOrUpdatedStorage(oldTrie.NodeIterator([]byte{}), newTrie.NodeIterator([]byte{}), diffPathsAtB, watchedStorageKeys, intermediateNodes, output)
if err != nil {
return nil, err
return err
}
return append(createdOrUpdatedStorage, deletedStorage...), nil
return nil
}
func (sdb *builder) createdAndUpdatedStorage(a, b trie.NodeIterator, watchedKeys []common.Hash, intermediateNodes bool) ([]sd.StorageNode, map[string]bool, error) {
createdOrUpdatedStorage := make([]sd.StorageNode, 0)
func (sdb *builder) createdAndUpdatedStorage(a, b trie.NodeIterator, watchedKeys []common.Hash, intermediateNodes bool, output sdtypes.StorageNodeSink) (map[string]bool, error) {
diffPathsAtB := make(map[string]bool)
it, _ := trie.NewDifferenceIterator(a, b)
for it.Next(true) {
// skip value nodes and null nodes
if it.Leaf() || bytes.Equal(nullHashBytes, it.Hash().Bytes()) {
continue
}
node, nodeElements, err := resolveNode(it, sdb.stateCache.TrieDB())
if err != nil {
return nil, nil, err
}
switch node.NodeType {
case sd.Leaf:
partialPath := trie.CompactToHex(nodeElements[0].([]byte))
valueNodePath := append(node.Path, partialPath...)
encodedPath := trie.HexToCompact(valueNodePath)
leafKey := encodedPath[1:]
if isWatchedStorageKey(watchedKeys, leafKey) {
createdOrUpdatedStorage = append(createdOrUpdatedStorage, sd.StorageNode{
NodeType: node.NodeType,
Path: node.Path,
NodeValue: node.NodeValue,
LeafKey: leafKey,
})
}
case sd.Extension, sd.Branch:
if intermediateNodes {
createdOrUpdatedStorage = append(createdOrUpdatedStorage, sd.StorageNode{
NodeType: node.NodeType,
Path: node.Path,
NodeValue: node.NodeValue,
})
}
default:
return nil, nil, fmt.Errorf("unexpected node type %s", node.NodeType)
}
diffPathsAtB[common.Bytes2Hex(node.Path)] = true
}
return createdOrUpdatedStorage, diffPathsAtB, it.Error()
}
func (sdb *builder) deletedOrUpdatedStorage(a, b trie.NodeIterator, diffPathsAtB map[string]bool, watchedKeys []common.Hash, intermediateNodes bool) ([]sd.StorageNode, error) {
deletedStorage := make([]sd.StorageNode, 0)
it, _ := trie.NewDifferenceIterator(b, a)
for it.Next(true) {
// skip value nodes and null nodes
if it.Leaf() || bytes.Equal(nullHashBytes, it.Hash().Bytes()) {
continue
}
@@ -640,6 +673,50 @@ func (sdb *builder) deletedOrUpdatedStorage(a, b trie.NodeIterator, diffPathsAtB
if err != nil {
return nil, err
}
switch node.NodeType {
case sdtypes.Leaf:
partialPath := trie.CompactToHex(nodeElements[0].([]byte))
valueNodePath := append(node.Path, partialPath...)
encodedPath := trie.HexToCompact(valueNodePath)
leafKey := encodedPath[1:]
if isWatchedStorageKey(watchedKeys, leafKey) {
if err := output(sdtypes.StorageNode{
NodeType: node.NodeType,
Path: node.Path,
NodeValue: node.NodeValue,
LeafKey: leafKey,
}); err != nil {
return nil, err
}
}
case sdtypes.Extension, sdtypes.Branch:
if intermediateNodes {
if err := output(sdtypes.StorageNode{
NodeType: node.NodeType,
Path: node.Path,
NodeValue: node.NodeValue,
}); err != nil {
return nil, err
}
}
default:
return nil, fmt.Errorf("unexpected node type %s", node.NodeType)
}
diffPathsAtB[common.Bytes2Hex(node.Path)] = true
}
return diffPathsAtB, it.Error()
}
func (sdb *builder) deletedOrUpdatedStorage(a, b trie.NodeIterator, diffPathsAtB map[string]bool, watchedKeys []common.Hash, intermediateNodes bool, output sdtypes.StorageNodeSink) error {
it, _ := trie.NewDifferenceIterator(b, a)
for it.Next(true) {
if it.Leaf() || bytes.Equal(nullHashBytes, it.Hash().Bytes()) {
continue
}
node, nodeElements, err := resolveNode(it, sdb.stateCache.TrieDB())
if err != nil {
return err
}
// if this node path showed up in diffPathsAtB
// that means this node was updated at B and we already have the updated diff for it
// otherwise that means this node was deleted in B and we need to add a "removed" diff to represent that event
@@ -647,31 +724,36 @@ func (sdb *builder) deletedOrUpdatedStorage(a, b trie.NodeIterator, diffPathsAtB
continue
}
switch node.NodeType {
case sd.Leaf:
case sdtypes.Leaf:
partialPath := trie.CompactToHex(nodeElements[0].([]byte))
valueNodePath := append(node.Path, partialPath...)
encodedPath := trie.HexToCompact(valueNodePath)
leafKey := encodedPath[1:]
if isWatchedStorageKey(watchedKeys, leafKey) {
deletedStorage = append(deletedStorage, sd.StorageNode{
NodeType: sd.Removed,
if err := output(sdtypes.StorageNode{
NodeType: sdtypes.Removed,
Path: node.Path,
NodeValue: []byte{},
})
LeafKey: leafKey,
}); err != nil {
return err
}
}
case sd.Extension, sd.Branch:
case sdtypes.Extension, sdtypes.Branch:
if intermediateNodes {
deletedStorage = append(deletedStorage, sd.StorageNode{
NodeType: sd.Removed,
if err := output(sdtypes.StorageNode{
NodeType: sdtypes.Removed,
Path: node.Path,
NodeValue: []byte{},
})
}); err != nil {
return err
}
}
default:
return nil, fmt.Errorf("unexpected node type %s", node.NodeType)
return fmt.Errorf("unexpected node type %s", node.NodeType)
}
}
return deletedStorage, it.Error()
return it.Error()
}
// isWatchedAddress is used to check if a state account corresponds to one of the addresses the builder is configured to watch
+253 -239
View File
File diff suppressed because it is too large Load Diff
+9
View File
@@ -0,0 +1,9 @@
package statediff
// Config holds config params for the statediffing service
type Config struct {
ServiceWorkers uint
TrieWorkers uint
WorkerQueueSize uint
PreRuns []RangeRequest
}
+162
View File
@@ -0,0 +1,162 @@
// VulcanizeDB
// Copyright © 2021 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package prom
import (
"database/sql"
"github.com/prometheus/client_golang/prometheus"
)
const (
namespace = "eth_statediff_service"
subsystem = "connections"
)
// DBStatsGetter is an interface that gets sql.DBStats.
type DBStatsGetter interface {
Stats() sql.DBStats
}
// DBStatsCollector implements the prometheus.Collector interface.
type DBStatsCollector struct {
sg DBStatsGetter
// descriptions of exported metrics
maxOpenDesc *prometheus.Desc
openDesc *prometheus.Desc
inUseDesc *prometheus.Desc
idleDesc *prometheus.Desc
waitedForDesc *prometheus.Desc
blockedSecondsDesc *prometheus.Desc
closedMaxIdleDesc *prometheus.Desc
closedMaxLifetimeDesc *prometheus.Desc
}
// NewDBStatsCollector creates a new DBStatsCollector.
func NewDBStatsCollector(dbName string, sg DBStatsGetter) *DBStatsCollector {
labels := prometheus.Labels{"db_name": dbName}
return &DBStatsCollector{
sg: sg,
maxOpenDesc: prometheus.NewDesc(
prometheus.BuildFQName(namespace, subsystem, "max_open"),
"Maximum number of open connections to the database.",
nil,
labels,
),
openDesc: prometheus.NewDesc(
prometheus.BuildFQName(namespace, subsystem, "open"),
"The number of established connections both in use and idle.",
nil,
labels,
),
inUseDesc: prometheus.NewDesc(
prometheus.BuildFQName(namespace, subsystem, "in_use"),
"The number of connections currently in use.",
nil,
labels,
),
idleDesc: prometheus.NewDesc(
prometheus.BuildFQName(namespace, subsystem, "idle"),
"The number of idle connections.",
nil,
labels,
),
waitedForDesc: prometheus.NewDesc(
prometheus.BuildFQName(namespace, subsystem, "waited_for"),
"The total number of connections waited for.",
nil,
labels,
),
blockedSecondsDesc: prometheus.NewDesc(
prometheus.BuildFQName(namespace, subsystem, "blocked_seconds"),
"The total time blocked waiting for a new connection.",
nil,
labels,
),
closedMaxIdleDesc: prometheus.NewDesc(
prometheus.BuildFQName(namespace, subsystem, "closed_max_idle"),
"The total number of connections closed due to SetMaxIdleConns.",
nil,
labels,
),
closedMaxLifetimeDesc: prometheus.NewDesc(
prometheus.BuildFQName(namespace, subsystem, "closed_max_lifetime"),
"The total number of connections closed due to SetConnMaxLifetime.",
nil,
labels,
),
}
}
// Describe implements the prometheus.Collector interface.
func (c DBStatsCollector) Describe(ch chan<- *prometheus.Desc) {
ch <- c.maxOpenDesc
ch <- c.openDesc
ch <- c.inUseDesc
ch <- c.idleDesc
ch <- c.waitedForDesc
ch <- c.blockedSecondsDesc
ch <- c.closedMaxIdleDesc
ch <- c.closedMaxLifetimeDesc
}
// Collect implements the prometheus.Collector interface.
func (c DBStatsCollector) Collect(ch chan<- prometheus.Metric) {
stats := c.sg.Stats()
ch <- prometheus.MustNewConstMetric(
c.maxOpenDesc,
prometheus.GaugeValue,
float64(stats.MaxOpenConnections),
)
ch <- prometheus.MustNewConstMetric(
c.openDesc,
prometheus.GaugeValue,
float64(stats.OpenConnections),
)
ch <- prometheus.MustNewConstMetric(
c.inUseDesc,
prometheus.GaugeValue,
float64(stats.InUse),
)
ch <- prometheus.MustNewConstMetric(
c.idleDesc,
prometheus.GaugeValue,
float64(stats.Idle),
)
ch <- prometheus.MustNewConstMetric(
c.waitedForDesc,
prometheus.CounterValue,
float64(stats.WaitCount),
)
ch <- prometheus.MustNewConstMetric(
c.blockedSecondsDesc,
prometheus.CounterValue,
stats.WaitDuration.Seconds(),
)
ch <- prometheus.MustNewConstMetric(
c.closedMaxIdleDesc,
prometheus.CounterValue,
float64(stats.MaxIdleClosed),
)
ch <- prometheus.MustNewConstMetric(
c.closedMaxLifetimeDesc,
prometheus.CounterValue,
float64(stats.MaxLifetimeClosed),
)
}
+51
View File
@@ -0,0 +1,51 @@
// VulcanizeDB
// Copyright © 2020 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package prom
import (
"net/http"
"time"
"github.com/ethereum/go-ethereum/rpc"
)
// HTTPMiddleware http connection metric reader
func HTTPMiddleware(next http.Handler) http.Handler {
if !metrics {
return next
}
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
httpCount.Inc()
start := time.Now()
next.ServeHTTP(w, r)
duration := time.Now().Sub(start)
httpDuration.Observe(float64(duration.Seconds()))
})
}
// IPCMiddleware unix-socket connection counter
func IPCMiddleware(server *rpc.Server, client rpc.Conn) {
if metrics {
ipcCount.Inc()
}
server.ServeCodec(rpc.NewCodec(client), 0)
if metrics {
ipcCount.Dec()
}
}
+176
View File
@@ -0,0 +1,176 @@
// VulcanizeDB
// Copyright © 2021 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package prom
import (
"time"
"github.com/jmoiron/sqlx"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
)
const (
statsSubsystem = "stats"
subsystemHTTP = "http"
subsystemIPC = "ipc"
)
var (
metrics bool
queuedRanges prometheus.Gauge
lastLoadedHeight prometheus.Gauge
lastProcessedHeight prometheus.Gauge
tBlockLoad prometheus.Histogram
tBlockProcessing prometheus.Histogram
tStateProcessing prometheus.Histogram
tTxCommit prometheus.Histogram
httpCount prometheus.Counter
httpDuration prometheus.Histogram
ipcCount prometheus.Gauge
)
const (
RANGES_QUEUED = "ranges_queued"
LOADED_HEIGHT = "loaded_height"
PROCESSED_HEIGHT = "processed_height"
T_BLOCK_LOAD = "t_block_load"
T_BLOCK_PROCESSING = "t_block_processing"
T_STATE_PROCESSING = "t_state_processing"
T_POSTGRES_TX_COMMIT = "t_postgres_tx_commit"
)
// Init module initialization
func Init() {
metrics = true
queuedRanges = promauto.NewGauge(prometheus.GaugeOpts{
Namespace: namespace,
Name: RANGES_QUEUED,
Help: "Number of range requests currently queued",
})
lastLoadedHeight = promauto.NewGauge(prometheus.GaugeOpts{
Namespace: namespace,
Name: LOADED_HEIGHT,
Help: "The last block that was loaded for processing",
})
lastProcessedHeight = promauto.NewGauge(prometheus.GaugeOpts{
Namespace: namespace,
Name: PROCESSED_HEIGHT,
Help: "The last block that was processed",
})
tBlockLoad = promauto.NewHistogram(prometheus.HistogramOpts{
Namespace: namespace,
Subsystem: statsSubsystem,
Name: T_BLOCK_LOAD,
Help: "Block loading time",
})
tBlockProcessing = promauto.NewHistogram(prometheus.HistogramOpts{
Namespace: namespace,
Subsystem: statsSubsystem,
Name: T_BLOCK_PROCESSING,
Help: "Block (header, uncles, txs, rcts, tx trie, rct trie) processing time",
})
tStateProcessing = promauto.NewHistogram(prometheus.HistogramOpts{
Namespace: namespace,
Subsystem: statsSubsystem,
Name: T_STATE_PROCESSING,
Help: "State (state trie, storage tries, and code) processing time",
})
tTxCommit = promauto.NewHistogram(prometheus.HistogramOpts{
Namespace: namespace,
Subsystem: statsSubsystem,
Name: T_POSTGRES_TX_COMMIT,
Help: "Postgres tx commit time",
})
httpCount = promauto.NewCounter(prometheus.CounterOpts{
Namespace: namespace,
Subsystem: subsystemHTTP,
Name: "count",
Help: "http request count",
})
httpDuration = promauto.NewHistogram(prometheus.HistogramOpts{
Namespace: namespace,
Subsystem: subsystemHTTP,
Name: "duration",
Help: "http request duration",
})
ipcCount = promauto.NewGauge(prometheus.GaugeOpts{
Namespace: namespace,
Subsystem: subsystemIPC,
Name: "count",
Help: "unix socket connection count",
})
}
// RegisterDBCollector create metric collector for given connection
func RegisterDBCollector(name string, db *sqlx.DB) {
if metrics {
prometheus.Register(NewDBStatsCollector(name, db))
}
}
// IncQueuedRanges increments the number of queued range requests
func IncQueuedRanges() {
if metrics {
queuedRanges.Inc()
}
}
// DecQueuedRanges decrements the number of queued range requests
func DecQueuedRanges() {
if metrics {
queuedRanges.Dec()
}
}
// SetLastLoadedHeight sets last loaded height
func SetLastLoadedHeight(height int64) {
if metrics {
lastLoadedHeight.Set(float64(height))
}
}
// SetLastProcessedHeight sets last processed height
func SetLastProcessedHeight(height int64) {
if metrics {
lastProcessedHeight.Set(float64(height))
}
}
// SetTimeMetric time metric observation
func SetTimeMetric(name string, t time.Duration) {
if !metrics {
return
}
tAsF64 := t.Seconds()
switch name {
case T_BLOCK_LOAD:
tBlockLoad.Observe(tAsF64)
case T_BLOCK_PROCESSING:
tBlockProcessing.Observe(tAsF64)
case T_STATE_PROCESSING:
tStateProcessing.Observe(tAsF64)
case T_POSTGRES_TX_COMMIT:
tTxCommit.Observe(tAsF64)
}
}
+40
View File
@@ -0,0 +1,40 @@
// VulcanizeDB
// Copyright © 2021 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package prom
import (
"net/http"
"github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/sirupsen/logrus"
)
// Listen start listening http
func Listen(addr string) *http.Server {
mux := http.NewServeMux()
mux.Handle("/metrics", promhttp.Handler())
srv := http.Server{
Addr: addr,
Handler: mux,
}
go func() {
if err := srv.ListenAndServe(); err != nil {
logrus.Fatal(err)
}
}()
return &srv
}
+24 -6
View File
@@ -25,25 +25,43 @@ import (
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/trie"
)
// LvlDBReader exposes the necessary read functions on lvldb
// Reader interface required by the statediffing service
type Reader interface {
GetBlockByHash(hash common.Hash) (*types.Block, error)
GetBlockByNumber(number uint64) (*types.Block, error)
GetReceiptsByHash(hash common.Hash) (types.Receipts, error)
GetTdByHash(hash common.Hash) (*big.Int, error)
StateDB() state.Database
}
// LvlDBReader exposes the necessary Reader methods on lvldb
type LvlDBReader struct {
ethDB ethdb.Database
stateDB state.Database
chainConfig *params.ChainConfig
}
// NewLvlDBReader creates a new LvlDBReader
func NewLvlDBReader(path, ancient string, chainConfig *params.ChainConfig) (*LvlDBReader, error) {
edb, err := rawdb.NewLevelDBDatabaseWithFreezer(path, 1024, 256, ancient, "eth-statediff-service")
// LvLDBReaderConfig struct for initializing a LvlDBReader
type LvLDBReaderConfig struct {
TrieConfig *trie.Config
ChainConfig *params.ChainConfig
Path, AncientPath string
DBCacheSize int
}
// NewLvlDBReader creates a new Read using LevelDB
func NewLvlDBReader(conf LvLDBReaderConfig) (*LvlDBReader, error) {
edb, err := rawdb.NewLevelDBDatabaseWithFreezer(conf.Path, conf.DBCacheSize, 256, conf.AncientPath, "eth-statediff-service", true)
if err != nil {
return nil, err
}
return &LvlDBReader{
ethDB: edb,
stateDB: state.NewDatabase(edb),
chainConfig: chainConfig,
stateDB: state.NewDatabaseWithConfig(edb, conf.TrieConfig),
chainConfig: conf.ChainConfig,
}, nil
}
+37
View File
@@ -0,0 +1,37 @@
// VulcanizeDB
// Copyright © 2020 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package rpc
import "github.com/ethereum/go-ethereum/rpc"
// checkModuleAvailability check that all names given in modules are actually
// available API services.
func checkModuleAvailability(modules []string, apis []rpc.API) (bad, available []string) {
availableSet := make(map[string]struct{})
for _, api := range apis {
if _, ok := availableSet[api.Namespace]; !ok {
availableSet[api.Namespace] = struct{}{}
available = append(available, api.Namespace)
}
}
for _, name := range modules {
if _, ok := availableSet[name]; !ok {
bad = append(bad, name)
}
}
return bad, available
}
+49
View File
@@ -0,0 +1,49 @@
// VulcanizeDB
// Copyright © 2020 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package rpc
import (
"fmt"
"github.com/ethereum/go-ethereum/cmd/utils"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/rpc"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/eth-statediff-service/pkg/prom"
)
// StartHTTPEndpoint starts the HTTP RPC endpoint, configured with cors/vhosts/modules.
func StartHTTPEndpoint(endpoint string, apis []rpc.API, modules []string, cors []string, vhosts []string, timeouts rpc.HTTPTimeouts) (*rpc.Server, error) {
srv := rpc.NewServer()
err := node.RegisterApis(apis, modules, srv, false)
if err != nil {
utils.Fatalf("Could not register HTTP API: %w", err)
}
handler := node.NewHTTPHandlerStack(srv, cors, vhosts)
// start http server
_, addr, err := node.StartHTTPEndpoint(endpoint, rpc.DefaultHTTPTimeouts, prom.HTTPMiddleware(handler))
if err != nil {
utils.Fatalf("Could not start RPC api: %v", err)
}
extapiURL := fmt.Sprintf("http://%v/", addr)
log.Infof("HTTP endpoint opened %s", extapiURL)
return srv, err
}
+92
View File
@@ -0,0 +1,92 @@
// VulcanizeDB
// Copyright © 2020 Vulcanize
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package rpc
import (
"fmt"
"net"
"os"
"path/filepath"
"github.com/ethereum/go-ethereum/p2p/netutil"
"github.com/ethereum/go-ethereum/rpc"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/eth-statediff-service/pkg/prom"
)
var (
// On Linux, sun_path is 108 bytes in size
// see http://man7.org/linux/man-pages/man7/unix.7.html
maxPathSize = 108
)
// ipcListen will create a Unix socket on the given endpoint.
func ipcListen(endpoint string) (net.Listener, error) {
if len(endpoint) > int(maxPathSize) {
log.Warn(fmt.Sprintf("The ipc endpoint is longer than %d characters. ", maxPathSize),
"endpoint", endpoint)
}
// Ensure the IPC path exists and remove any previous leftover
if err := os.MkdirAll(filepath.Dir(endpoint), 0751); err != nil {
return nil, err
}
os.Remove(endpoint)
l, err := net.Listen("unix", endpoint)
if err != nil {
return nil, err
}
os.Chmod(endpoint, 0600)
return l, nil
}
func ipcServe(srv *rpc.Server, listener net.Listener) {
for {
conn, err := listener.Accept()
if netutil.IsTemporaryError(err) {
log.WithError(err).Warn("rpc accept error")
continue
}
if err != nil {
log.WithError(err).Warn("unknown error")
continue
}
log.WithField("addr", conn.RemoteAddr()).Trace("accepted ipc connection")
go prom.IPCMiddleware(srv, conn)
}
}
// StartIPCEndpoint starts an IPC endpoint.
func StartIPCEndpoint(ipcEndpoint string, apis []rpc.API) (net.Listener, *rpc.Server, error) {
// Register all the APIs exposed by the services.
handler := rpc.NewServer()
for _, api := range apis {
if err := handler.RegisterName(api.Namespace, api.Service); err != nil {
return nil, nil, err
}
log.Debug("IPC registered", "namespace", api.Namespace)
}
// All APIs registered, start the IPC listener.
listener, err := ipcListen(ipcEndpoint)
if err != nil {
return nil, nil, err
}
go ipcServe(handler, listener)
return listener, handler, nil
}
+239 -44
View File
@@ -20,43 +20,47 @@ import (
"fmt"
"math/big"
"sync"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
sd "github.com/ethereum/go-ethereum/statediff"
"github.com/ethereum/go-ethereum/statediff/indexer/interfaces"
sdtypes "github.com/ethereum/go-ethereum/statediff/types"
"github.com/sirupsen/logrus"
"github.com/vulcanize/eth-statediff-service/pkg/prom"
)
// lvlDBReader are the db interfaces required by the statediffing service
type lvlDBReader interface {
GetBlockByHash(hash common.Hash) (*types.Block, error)
GetBlockByNumber(number uint64) (*types.Block, error)
GetReceiptsByHash(hash common.Hash) (types.Receipts, error)
GetTdByHash(hash common.Hash) (*big.Int, error)
StateDB() state.Database
}
const defaultQueueSize = 1024
// IService is the state-diffing service interface
type IService interface {
// APIs(), Protocols(), Start() and Stop()
node.Service
// Main event loop for processing state diffs
Loop(wg *sync.WaitGroup)
// Method to get state diff object at specific block
// StateDiffService is the state-diffing service interface
type StateDiffService interface {
// Lifecycle Start() and Stop()
node.Lifecycle
// APIs and Protocols() interface for node service registration
APIs() []rpc.API
Protocols() []p2p.Protocol
// Loop is the main event loop for processing state diffs
Loop(wg *sync.WaitGroup) error
// Run is a one-off command to run on a predefined set of ranges
Run(ranges []RangeRequest) error
// StateDiffAt method to get state diff object at specific block
StateDiffAt(blockNumber uint64, params sd.Params) (*sd.Payload, error)
// Method to get state trie object at specific block
// StateDiffFor method to get state diff object at specific block
StateDiffFor(blockHash common.Hash, params sd.Params) (*sd.Payload, error)
// StateTrieAt method to get state trie object at specific block
StateTrieAt(blockNumber uint64, params sd.Params) (*sd.Payload, error)
}
// Server configuration
type Config struct {
// Number of goroutines to use
Workers uint
// WriteStateDiffAt method to write state diff object directly to DB
WriteStateDiffAt(blockNumber uint64, params sd.Params) error
// WriteStateDiffFor method to get state trie object at specific block
WriteStateDiffFor(blockHash common.Hash, params sd.Params) error
// WriteStateDiffsInRange method to wrtie state diff objects within the range directly to the DB
WriteStateDiffsInRange(start, stop uint64, params sd.Params) error
}
// Service is the underlying struct for the state diffing service
@@ -64,21 +68,35 @@ type Service struct {
// Used to build the state diff objects
Builder Builder
// Used to read data from leveldb
lvlDBReader lvlDBReader
lvlDBReader Reader
// Used to signal shutdown of the service
QuitChan chan bool
quitChan chan struct{}
// Interface for publishing statediffs as PG-IPLD objects
indexer interfaces.StateDiffIndexer
// range queue
queue chan RangeRequest
// number of ranges we can work over concurrently
workers uint
// ranges configured locally
preruns []RangeRequest
}
// NewStateDiffService creates a new Service
func NewStateDiffService(lvlDBReader lvlDBReader, cfg Config) (*Service, error) {
builder, err := NewBuilder(lvlDBReader.StateDB(), cfg.Workers)
func NewStateDiffService(lvlDBReader Reader, indexer interfaces.StateDiffIndexer, conf Config) (*Service, error) {
b, err := NewBuilder(lvlDBReader.StateDB(), conf.TrieWorkers)
if err != nil {
return nil, err
}
if conf.WorkerQueueSize == 0 {
conf.WorkerQueueSize = defaultQueueSize
}
return &Service{
lvlDBReader: lvlDBReader,
Builder: builder,
QuitChan: make(chan bool),
Builder: b,
indexer: indexer,
workers: conf.ServiceWorkers,
queue: make(chan RangeRequest, conf.WorkerQueueSize),
preruns: conf.PreRuns,
}, nil
}
@@ -99,17 +117,73 @@ func (sds *Service) APIs() []rpc.API {
}
}
// Loop is an empty service loop for awaiting rpc requests
func (sds *Service) Loop(wg *sync.WaitGroup) {
wg.Add(1)
for {
select {
case <-sds.QuitChan:
logrus.Info("closing the statediff service loop")
wg.Done()
return
// Run does a one-off processing run on the provided RangeRequests + any pre-runs, exiting afterwards
func (sds *Service) Run(rngs []RangeRequest) error {
for _, preRun := range sds.preruns {
logrus.Infof("processing prerun range (%d, %d)", preRun.Start, preRun.Stop)
for i := preRun.Start; i <= preRun.Stop; i++ {
if err := sds.WriteStateDiffAt(i, preRun.Params); err != nil {
return fmt.Errorf("error writing statediff at height %d in range (%d, %d) : %v", i, preRun.Start, preRun.Stop, err)
}
}
}
sds.preruns = nil
for _, rng := range rngs {
logrus.Infof("processing prerun range (%d, %d)", rng.Start, rng.Stop)
for i := rng.Start; i <= rng.Stop; i++ {
if err := sds.WriteStateDiffAt(i, rng.Params); err != nil {
return fmt.Errorf("error writing statediff at height %d in range (%d, %d) : %v", i, rng.Start, rng.Stop, err)
}
}
}
return nil
}
// Loop is an empty service loop for awaiting rpc requests
func (sds *Service) Loop(wg *sync.WaitGroup) error {
if sds.quitChan != nil {
return fmt.Errorf("service loop is already running")
}
sds.quitChan = make(chan struct{})
for i := 0; i < int(sds.workers); i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
for {
select {
case blockRange := <-sds.queue:
logrus.Infof("service worker %d received range (%d, %d) off of work queue, beginning processing", id, blockRange.Start, blockRange.Stop)
prom.DecQueuedRanges()
for j := blockRange.Start; j <= blockRange.Stop; j++ {
if err := sds.WriteStateDiffAt(j, blockRange.Params); err != nil {
logrus.Errorf("service worker %d error writing statediff at height %d in range (%d, %d) : %v", id, j, blockRange.Start, blockRange.Stop, err)
}
select {
case <-sds.quitChan:
logrus.Infof("closing service worker %d\n"+
"working in range (%d, %d)\n"+
"last processed height: %d", id, blockRange.Start, blockRange.Stop, j)
return
default:
logrus.Infof("service worker %d finished processing statediff height %d in range (%d, %d)", id, j, blockRange.Start, blockRange.Stop)
}
}
logrus.Infof("service worker %d finished processing range (%d, %d)", id, blockRange.Start, blockRange.Stop)
case <-sds.quitChan:
logrus.Infof("closing the statediff service loop worker %d", id)
return
}
}
}(i)
}
for _, preRun := range sds.preruns {
if err := sds.WriteStateDiffsInRange(preRun.Start, preRun.Stop, preRun.Params); err != nil {
close(sds.quitChan)
return err
}
}
return nil
}
// StateDiffAt returns a state diff object payload at the specific blockheight
@@ -119,7 +193,7 @@ func (sds *Service) StateDiffAt(blockNumber uint64, params sd.Params) (*sd.Paylo
if err != nil {
return nil, err
}
logrus.Info(fmt.Sprintf("sending state diff at block %d", blockNumber))
logrus.Infof("sending state diff at block %d", blockNumber)
if blockNumber == 0 {
return sds.processStateDiff(currentBlock, common.Hash{}, params)
}
@@ -130,6 +204,24 @@ func (sds *Service) StateDiffAt(blockNumber uint64, params sd.Params) (*sd.Paylo
return sds.processStateDiff(currentBlock, parentBlock.Root(), params)
}
// StateDiffFor returns a state diff object payload for the specific blockhash
// This operation cannot be performed back past the point of db pruning; it requires an archival node for historical data
func (sds *Service) StateDiffFor(blockHash common.Hash, params sd.Params) (*sd.Payload, error) {
currentBlock, err := sds.lvlDBReader.GetBlockByHash(blockHash)
if err != nil {
return nil, err
}
logrus.Infof("sending state diff at block %s", blockHash.Hex())
if currentBlock.NumberU64() == 0 {
return sds.processStateDiff(currentBlock, common.Hash{}, params)
}
parentBlock, err := sds.lvlDBReader.GetBlockByHash(currentBlock.ParentHash())
if err != nil {
return nil, err
}
return sds.processStateDiff(currentBlock, parentBlock.Root(), params)
}
// processStateDiff method builds the state diff payload from the current block, parent state root, and provided params
func (sds *Service) processStateDiff(currentBlock *types.Block, parentRoot common.Hash, params sd.Params) (*sd.Payload, error) {
stateDiff, err := sds.Builder.BuildStateDiffObject(sd.Args{
@@ -188,7 +280,7 @@ func (sds *Service) StateTrieAt(blockNumber uint64, params sd.Params) (*sd.Paylo
if err != nil {
return nil, err
}
logrus.Info(fmt.Sprintf("sending state trie at block %d", blockNumber))
logrus.Infof("sending state trie at block %d", blockNumber)
return sds.processStateTrie(currentBlock, params)
}
@@ -206,15 +298,118 @@ func (sds *Service) processStateTrie(block *types.Block, params sd.Params) (*sd.
}
// Start is used to begin the service
func (sds *Service) Start(*p2p.Server) error {
func (sds *Service) Start() error {
logrus.Info("starting statediff service")
go sds.Loop(new(sync.WaitGroup))
return nil
return sds.Loop(new(sync.WaitGroup))
}
// Stop is used to close down the service
func (sds *Service) Stop() error {
logrus.Info("stopping statediff service")
close(sds.QuitChan)
close(sds.quitChan)
return nil
}
// WriteStateDiffAt writes a state diff at the specific blockheight directly to the database
// This operation cannot be performed back past the point of db pruning; it requires an archival node
// for historical data
func (sds *Service) WriteStateDiffAt(blockNumber uint64, params sd.Params) error {
logrus.Infof("Writing state diff at block %d", blockNumber)
t := time.Now()
currentBlock, err := sds.lvlDBReader.GetBlockByNumber(blockNumber)
if err != nil {
return err
}
parentRoot := common.Hash{}
if blockNumber != 0 {
parentBlock, err := sds.lvlDBReader.GetBlockByHash(currentBlock.ParentHash())
if err != nil {
return err
}
parentRoot = parentBlock.Root()
}
return sds.writeStateDiff(currentBlock, parentRoot, params, t)
}
// WriteStateDiffFor writes a state diff for the specific blockHash directly to the database
// This operation cannot be performed back past the point of db pruning; it requires an archival node
// for historical data
func (sds *Service) WriteStateDiffFor(blockHash common.Hash, params sd.Params) error {
logrus.Infof("Writing state diff for block %s", blockHash.Hex())
t := time.Now()
currentBlock, err := sds.lvlDBReader.GetBlockByHash(blockHash)
if err != nil {
return err
}
parentRoot := common.Hash{}
if currentBlock.NumberU64() != 0 {
parentBlock, err := sds.lvlDBReader.GetBlockByHash(currentBlock.ParentHash())
if err != nil {
return err
}
parentRoot = parentBlock.Root()
}
return sds.writeStateDiff(currentBlock, parentRoot, params, t)
}
// Writes a state diff from the current block, parent state root, and provided params
func (sds *Service) writeStateDiff(block *types.Block, parentRoot common.Hash, params sd.Params, t time.Time) error {
var totalDifficulty *big.Int
var receipts types.Receipts
var err error
if params.IncludeTD {
totalDifficulty, err = sds.lvlDBReader.GetTdByHash(block.Hash())
}
if err != nil {
return err
}
if params.IncludeReceipts {
receipts, err = sds.lvlDBReader.GetReceiptsByHash(block.Hash())
}
if err != nil {
return err
}
height := block.Number().Int64()
prom.SetLastLoadedHeight(height)
prom.SetTimeMetric(prom.T_BLOCK_LOAD, time.Now().Sub(t))
t = time.Now()
tx, err := sds.indexer.PushBlock(block, receipts, totalDifficulty)
if err != nil {
return err
}
// defer handling of commit/rollback for any return case
output := func(node sdtypes.StateNode) error {
return sds.indexer.PushStateNode(tx, node, block.Hash().String())
}
codeOutput := func(c sdtypes.CodeAndCodeHash) error {
return sds.indexer.PushCodeAndCodeHash(tx, c)
}
prom.SetTimeMetric(prom.T_BLOCK_PROCESSING, time.Now().Sub(t))
t = time.Now()
err = sds.Builder.WriteStateDiffObject(sdtypes.StateRoots{
NewStateRoot: block.Root(),
OldStateRoot: parentRoot,
}, params, output, codeOutput)
prom.SetTimeMetric(prom.T_STATE_PROCESSING, time.Now().Sub(t))
t = time.Now()
err = tx.Submit(err)
prom.SetLastProcessedHeight(height)
prom.SetTimeMetric(prom.T_POSTGRES_TX_COMMIT, time.Now().Sub(t))
return err
}
// WriteStateDiffsInRange adds a RangeRequest to the work queue
func (sds *Service) WriteStateDiffsInRange(start, stop uint64, params sd.Params) error {
if stop < start {
return fmt.Errorf("invalid block range (%d, %d): stop height must be greater or equal to start height", start, stop)
}
blocked := time.NewTimer(30 * time.Second)
select {
case sds.queue <- RangeRequest{Start: start, Stop: stop, Params: params}:
prom.IncQueuedRanges()
logrus.Infof("added range (%d, %d) to the worker queue", start, stop)
return nil
case <-blocked.C:
return fmt.Errorf("unable to add range (%d, %d) to the worker queue", start, stop)
}
}
+2 -1
View File
@@ -32,8 +32,9 @@ import (
// the returned hash chain is ordered head->parent.
func MakeChain(n int, parent *types.Block, chainGen func(int, *core.BlockGen)) ([]*types.Block, *core.BlockChain) {
config := params.TestChainConfig
config.LondonBlock.Set(big.NewInt(10))
blocks, _ := core.GenerateChain(config, parent, ethash.NewFaker(), Testdb, n, chainGen)
chain, _ := core.NewBlockChain(Testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil)
chain, _ := core.NewBlockChain(Testdb, nil, params.TestChainConfig, ethash.NewFaker(), vm.Config{}, nil, nil)
return blocks, chain
}
-128
View File
@@ -1,128 +0,0 @@
// Copyright 2019 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package mocks
import (
"errors"
"math/big"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/event"
)
// BlockChain is a mock blockchain for testing
type BlockChain struct {
HashesLookedUp []common.Hash
blocksToReturnByHash map[common.Hash]*types.Block
blocksToReturnByNumber map[uint64]*types.Block
callCount int
ChainEvents []core.ChainEvent
Receipts map[common.Hash]types.Receipts
TDByHash map[common.Hash]*big.Int
}
// SetBlocksForHashes mock method
func (blockChain *BlockChain) SetBlocksForHashes(blocks map[common.Hash]*types.Block) {
if blockChain.blocksToReturnByHash == nil {
blockChain.blocksToReturnByHash = make(map[common.Hash]*types.Block)
}
blockChain.blocksToReturnByHash = blocks
}
// GetBlockByHash mock method
func (blockChain *BlockChain) GetBlockByHash(hash common.Hash) *types.Block {
blockChain.HashesLookedUp = append(blockChain.HashesLookedUp, hash)
var block *types.Block
if len(blockChain.blocksToReturnByHash) > 0 {
block = blockChain.blocksToReturnByHash[hash]
}
return block
}
// SetChainEvents mock method
func (blockChain *BlockChain) SetChainEvents(chainEvents []core.ChainEvent) {
blockChain.ChainEvents = chainEvents
}
// SubscribeChainEvent mock method
func (blockChain *BlockChain) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {
subErr := errors.New("subscription error")
var eventCounter int
subscription := event.NewSubscription(func(quit <-chan struct{}) error {
for _, chainEvent := range blockChain.ChainEvents {
if eventCounter > 1 {
time.Sleep(250 * time.Millisecond)
return subErr
}
select {
case ch <- chainEvent:
case <-quit:
return nil
}
eventCounter++
}
return nil
})
return subscription
}
// SetReceiptsForHash test method
func (blockChain *BlockChain) SetReceiptsForHash(hash common.Hash, receipts types.Receipts) {
if blockChain.Receipts == nil {
blockChain.Receipts = make(map[common.Hash]types.Receipts)
}
blockChain.Receipts[hash] = receipts
}
// GetReceiptsByHash mock method
func (blockChain *BlockChain) GetReceiptsByHash(hash common.Hash) types.Receipts {
return blockChain.Receipts[hash]
}
// SetBlockForNumber test method
func (blockChain *BlockChain) SetBlockForNumber(block *types.Block, number uint64) {
if blockChain.blocksToReturnByNumber == nil {
blockChain.blocksToReturnByNumber = make(map[uint64]*types.Block)
}
blockChain.blocksToReturnByNumber[number] = block
}
// GetBlockByNumber mock method
func (blockChain *BlockChain) GetBlockByNumber(number uint64) *types.Block {
return blockChain.blocksToReturnByNumber[number]
}
// GetTdByHash mock method
func (blockChain *BlockChain) GetTdByHash(hash common.Hash) *big.Int {
return blockChain.TDByHash[hash]
}
func (blockChain *BlockChain) SetTdByHash(hash common.Hash, td *big.Int) {
if blockChain.TDByHash == nil {
blockChain.TDByHash = make(map[common.Hash]*big.Int)
}
blockChain.TDByHash[hash] = td
}
func (blockChain *BlockChain) UnlockTrie(root common.Hash) {}
-57
View File
@@ -1,57 +0,0 @@
// Copyright 2019 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package mocks
import (
"github.com/ethereum/go-ethereum/core/types"
statediff "github.com/vulcanize/eth-statediff-service/pkg"
)
// Builder is a mock state diff builder
type Builder struct {
Args statediff.Args
Params statediff.Params
stateDiff statediff.StateObject
block *types.Block
stateTrie statediff.StateObject
builderError error
}
// BuildStateDiffObject mock method
func (builder *Builder) BuildStateDiffObject(args statediff.Args, params statediff.Params) (statediff.StateObject, error) {
builder.Args = args
builder.Params = params
return builder.stateDiff, builder.builderError
}
// BuildStateTrieObject mock method
func (builder *Builder) BuildStateTrieObject(block *types.Block) (statediff.StateObject, error) {
builder.block = block
return builder.stateTrie, builder.builderError
}
// SetStateDiffToBuild mock method
func (builder *Builder) SetStateDiffToBuild(stateDiff statediff.StateObject) {
builder.stateDiff = stateDiff
}
// SetBuilderError mock method
func (builder *Builder) SetBuilderError(err error) {
builder.builderError = err
}
-6
View File
@@ -18,7 +18,6 @@ package testhelpers
import (
"math/big"
"math/rand"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core"
@@ -42,12 +41,7 @@ func AddressToEncodedPath(address common.Address) []byte {
// Test variables
var (
EvenLeafFlag = []byte{byte(2) << 4}
BlockNumber = big.NewInt(rand.Int63())
BlockHash = "0xfa40fbe2d98d98b3363a778d52f2bcd29d6790b9b3f3cab2b167fd12d3550f73"
NullCodeHash = crypto.Keccak256Hash([]byte{})
StoragePath = common.HexToHash("0xc5d2460186f7233c927e7db2dcc703c0e500b653ca82273b7bfad8045d85a470").Bytes()
StorageKey = common.HexToHash("0000000000000000000000000000000000000000000000000000000000000001").Bytes()
StorageValue = common.Hex2Bytes("0x03")
NullHash = common.HexToHash("0x0000000000000000000000000000000000000000000000000000000000000000")
Testdb = rawdb.NewMemoryDatabase()
+10 -3
View File
@@ -20,8 +20,9 @@
package statediff
import (
"github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/types"
sd "github.com/ethereum/go-ethereum/statediff"
sdTypes "github.com/ethereum/go-ethereum/statediff/types"
)
// AccountMap is a mapping of hex encoded path => account wrapper
@@ -29,9 +30,15 @@ type AccountMap map[string]accountWrapper
// accountWrapper is used to temporary associate the unpacked node with its raw values
type accountWrapper struct {
Account *state.Account
NodeType sd.NodeType
Account *types.StateAccount
NodeType sdTypes.NodeType
Path []byte
NodeValue []byte
LeafKey []byte
}
// RangeRequest holds range quest work params
type RangeRequest struct {
Start, Stop uint64
Params sd.Params
}
Regular → Executable
+12 -3
View File
@@ -4,8 +4,17 @@ set -e
set +x
# Check the database variables are set
test $VDB_COMMAND
test "$VDB_COMMAND"
set +e
echo "Running the statediff service"
./eth-statediff-service ${VDB_COMMAND} --config=config.toml
# docker must be run in privilaged mode for mounts to work
echo "Setting up /app/geth-rw overlayed /app/geth-ro"
mkdir -p /tmp/overlay && \
sudo mount -t tmpfs tmpfs /tmp/overlay && \
mkdir -p /tmp/overlay/upper && \
mkdir -p /tmp/overlay/work && \
mkdir -p /app/geth-rw && \
sudo mount -t overlay overlay -o lowerdir=/app/geth-ro,upperdir=/tmp/overlay/upper,workdir=/tmp/overlay/work /app/geth-rw && \
echo "Running the statediff service" && \
sudo ./eth-statediff-service "$VDB_COMMAND" --config=config.toml