Feature/known gaps #217

Merged
abdulrabbani00 merged 15 commits from feature/known-gaps into v1.10.16-statediff-v3 2022-03-31 18:38:59 +00:00
34 changed files with 686 additions and 480 deletions

View File

@ -1,6 +1,7 @@
name: Docker Build and publish to Github
on:
workflow_dispatch:
push:
branches:
- statediff

8
.gitignore vendored
View File

@ -49,3 +49,11 @@ profile.cov
**/yarn-error.log
foundry/deployments/local-private-network/geth-linux-amd64
foundry/projects/local-private-network/geth-linux-amd64
# Helpful repos
related-repositories/foundry-test/**
related-repositories/hive/**
related-repositories/ipld-eth-db/**
statediff/indexer/database/sql/statediffing_test_file.sql
statediff/statediffing_test_file.sql
statediff/known_gaps.sql

5
.gitmodules vendored
View File

@ -5,7 +5,4 @@
[submodule "evm-benchmarks"]
path = tests/evm-benchmarks
url = https://github.com/ipsilon/evm-benchmarks
shallow = true
[submodule "foundry/projects/local-private-network/Stateful/lib/ds-test"]
path = foundry/projects/local-private-network/Stateful/lib/ds-test
url = https://github.com/dapphub/ds-test
shallow = true

View File

@ -252,13 +252,14 @@ func makeFullNode(ctx *cli.Context) (*node.Node, ethapi.Backend) {
}
}
p := statediff.Config{
IndexerConfig: indexerConfig,
ID: nodeID,
ClientName: clientName,
Context: context.Background(),
EnableWriteLoop: ctx.GlobalBool(utils.StateDiffWritingFlag.Name),
NumWorkers: ctx.GlobalUint(utils.StateDiffWorkersFlag.Name),
WaitForSync: ctx.GlobalBool(utils.StateDiffWaitForSync.Name),
IndexerConfig: indexerConfig,
KnownGapsFilePath: ctx.GlobalString(utils.StateDiffKnownGapsFilePath.Name),
ID: nodeID,
ClientName: clientName,
Context: context.Background(),
EnableWriteLoop: ctx.GlobalBool(utils.StateDiffWritingFlag.Name),
NumWorkers: ctx.GlobalUint(utils.StateDiffWorkersFlag.Name),
WaitForSync: ctx.GlobalBool(utils.StateDiffWaitForSync.Name),
}
utils.RegisterStateDiffService(stack, eth, &cfg.Eth, p, backend)
}

View File

@ -176,6 +176,7 @@ var (
utils.StateDiffWritingFlag,
utils.StateDiffWorkersFlag,
utils.StateDiffFilePath,
utils.StateDiffKnownGapsFilePath,
utils.StateDiffWaitForSync,
configFileFlag,
}

View File

@ -246,6 +246,7 @@ var AppHelpFlagGroups = []flags.FlagGroup{
utils.StateDiffWritingFlag,
utils.StateDiffWorkersFlag,
utils.StateDiffFilePath,
utils.StateDiffKnownGapsFilePath,
utils.StateDiffWaitForSync,
},
},

View File

@ -867,6 +867,11 @@ var (
Name: "statediff.file.path",
Usage: "Full path (including filename) to write statediff data out to when operating in file mode",
}
StateDiffKnownGapsFilePath = cli.StringFlag{
Name: "statediff.knowngapsfile.path",
Usage: "Full path (including filename) to write knownGaps statements when the DB is unavailable.",
Value: "./known_gaps.sql",
}
StateDiffDBClientNameFlag = cli.StringFlag{
Name: "statediff.db.clientname",
Usage: "Client name to use when writing state diffs to database",

View File

@ -1,9 +1,9 @@
version: '3.2'
version: "3.2"
services:
ipld-eth-db:
restart: always
image: vulcanize/ipld-eth-db:v0.3.1
image: vulcanize/ipld-eth-db:v3.2.0
environment:
POSTGRES_USER: "vdbm"
POSTGRES_DB: "vulcanize_testing_v3"

View File

@ -1,59 +0,0 @@
# Foundry README
# Overview
This document will go through the steps needed to test using Foundry. Currently, we use Foundry in the following capacity.
1. Create a private network with our internal version of Geth.
2. Deploy a smart contract to the private network.
3. Test the smart contract on the private network.
4. Create a transaction on the private network.
# Steps
The steps to create a new project are as follows.
## 1. Creating New Project
1. `cd foundry/projects`.
2. Create a directory that captures your project: `mkdir local-private-network; cd local-private-network`.
3. Create a [new foundry project](https://onbjerg.github.io/foundry-book/forge/creating-a-new-project.html): `forge init stateful`.
4. Follow the foundry [documentation](https://onbjerg.github.io/foundry-book/forge/tests.html) for writing smart contract tests.
## 2. Deployments
You can choose to have custom deployments for your workflow. However, it is recommended to utilize Docker.
# Existing Projects
Below, you can find existing projects and their descriptions.
## `local-private-network`
The purpose of this project is as follows:
1. Compile the geth from the local source.
2. Build a docker container with `ipld-eth-db` and another container for the `local-private-network`.
3. Run the compiled version of geth.
4. Deploy a smart contract to the private blockchain.
5. Trigger a transaction on the newly deployed smart contract.
## Using This Project
If you want to test your local geth code, do the following:
1. cd `foundry/projects/local-private-network`.
2. `./wrapper.sh` - This script will do all the heavy lifting for you.
3. Keep an eye out for the outputs from the docker container.
4. Enter the docker container and do as you please.
5. If you want to change your geth code, you will have to run `./wrapper.sh` for subsequent runs.
6. If you do not change your geth code, you have to run: `docker-compose up --build`.
### Key Notes:
- The command to [deploy](https://onbjerg.github.io/foundry-book/forge/deploying.html) the smart contract is: `forge create --keystore $ETH_KEYSTORE_FILE --rpc-url [http://127.0.0.1:8545](http://127.0.0.1:8545/) --constructor-args 1 --password "" --legacy /root/stateful/src/Stateful.sol:Stateful`
- The command to interact create a [transaction](https://onbjerg.github.io/foundry-book/reference/cast.html) is: `cast send --keystore $ETH_KEYSTORE_FILE --rpc-url [http://127.0.0.1:8545](http://127.0.0.1:8545/) --password "" --legacy $DEPLOYED_ADDRESS "off()"`
- The `Dockerfile` compiles `cast` and `forge`.
- The `foundry/projects/local-private-network/deploy-local-network.sh` file does most heavy lifting. It spins up geth and triggers various events.
- The `foundry/projects/local-private-network/start-private-network.sh` file triggers `deploy-local-network.sh`. This file runs all the tests.
- The `geth` node will stay running even after the tests are terminated.

View File

@ -1,37 +0,0 @@
FROM frolvlad/alpine-bash
# copy all files
RUN apk update ; apk add --no-cache --allow-untrusted ca-certificates curl bash git jq
ENV GLIBC_REPO=https://github.com/sgerrand/alpine-pkg-glibc
ENV GLIBC_VERSION=2.35-r0
RUN set -ex && \
apk --update add libstdc++ curl ca-certificates && \
for pkg in glibc-${GLIBC_VERSION} glibc-bin-${GLIBC_VERSION}; \
do curl -sSL ${GLIBC_REPO}/releases/download/${GLIBC_VERSION}/${pkg}.apk -o /tmp/${pkg}.apk; done && \
apk add --allow-untrusted /tmp/*.apk ; \
rm -v /tmp/*.apk ;/usr/glibc-compat/sbin/ldconfig /lib /usr/glibc-compat/lib
RUN apk add gcompat; echo "Sorry"
WORKDIR /root
COPY stateful ./stateful
ADD ./start-private-network.sh .
ADD ./deploy-local-network.sh .
ADD ../../geth-linux-amd64 /bin/geth
RUN curl -L https://foundry.paradigm.xyz | bash; \
/bin/bash -c 'source $HOME/.bashrc'; \
/root/.foundry/bin/foundryup
ENV PATH "$PATH:/root/.foundry/bin/"
RUN echo "export PATH=${PATH}" >> $HOME/.bashrc;
RUN chmod +x /bin/geth
EXPOSE 8545
EXPOSE 8546
ENTRYPOINT ["./start-private-network.sh"]

@ -1 +0,0 @@
Subproject commit 0a5da56b0d65960e6a994d2ec8245e6edd38c248

View File

@ -1,16 +0,0 @@
#!/bin/bash
set -e
GREEN='\033[0;32m'
RED='\033[0;31m'
NC='\033[0m'
start_path=$(pwd)
cd ../../../
echo -e "${GREEN}Building geth!${NC}"
docker build -t vulcanize/go-ethereum -f Dockerfile .
docker run --rm --entrypoint cat vulcanize/go-ethereum /usr/local/bin/geth > foundry/projects/local-private-network/geth-linux-amd64
chmod +x foundry/projects/local-private-network/geth-linux-amd64
echo -e "${GREEN}geth build complete!${NC}"
cd $start_path

View File

@ -1,201 +0,0 @@
#!/bin/bash
set -e
OPTS="./deploy-local-network.sh [<options>] <args>...
./deploy-local-network.sh --help
--
db-user=name database user
db-password=password database password
db-name=name database name
db-host=address database host
db-port=port database port
db-write=bool turn on database write mode
db-type=name the type of database
db-driver=name the driver used for the database
db-waitforsync=bool Should the statediff service start once geth has synced to head (default: false)
rpc-port=port change RPC port (default: 8545)
rpc-addr=address change RPC address (default: 127.0.0.1)
chain-id=number change chain ID (default: 99)
period=seconds use a block time instead of instamine
accounts=number create multiple accounts (default: 1)
address=address eth address to add to genesis
save=name after finishing, save snapshot
load=name start from a previously saved snapshot
dir=directory testnet directory
"
eval "$(
git rev-parse --parseopt -- "$@" <<<"$OPTS" || echo exit $?
)"
DB_USER=vdbm
DB_PASSWORD=password
DB_NAME=vulcanize_public
DB_HOST=127.0.0.1
DB_PORT=5432
DB_TYPE=postgres
DB_DRIVER=sqlx
DB_WAIT_FOR_SYNC=false
RPC_PORT=8545
RPC_ADDRESS=127.0.0.1
PERIOD=0
CHAINID=99
ACCOUNTS=0
ADDRESS=
gethdir=$HOME/testnet
while [[ $1 ]]; do
case $1 in
--) shift; break;;
--db-user) shift; DB_USER=$1;;
--db-password) shift; DB_PASSWORD=$1;;
--db-name) shift; DB_NAME=$1;;
--db-host) shift; DB_HOST=$1;;
--db-port) shift; DB_PORT=$1;;
--db-write) shift; DB_WRITE=$1;;
--db-type) shift; DB_TYPE=$1;;
--db-driver) shift; DB_DRIVER=$1;;
--db-waitforsync) shift; DB_WAIT_FOR_SYNC=$1;;
--rpc-port) shift; RPC_PORT=$1;;
--rpc-addr) shift; RPC_ADDRESS=$1;;
--chain-id) shift; CHAINID=$1;;
--period) shift; PERIOD=$1;;
--accounts) shift; ACCOUNTS=$1;;
--save) shift; SAVE=$1;;
--address) shift; ADDRESS=$1;;
--load) shift; LOAD=$1;;
--dir) shift; gethdir=$1;;
*) printf "${0##*/}: internal error: %q\\n" "$1"; exit 1
esac; shift
done
chaindir=$gethdir/$RPC_PORT
#while true; do
# if [[ ! -d "$gethdir/$CHAINID" ]]; then break; fi
# CHAINID=$((CHAINID + 1))
#done
mkdir -p "$chaindir/config"
#if [ -n "$ADDRESS" ]; then
# balance+=(-n {} -s "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" -i balance \
# -i "$ADDRESS")
#fi
for i in $(seq 0 "$ACCOUNTS"); do
address+=( "$(
geth 2>/dev/null account new --datadir "$chaindir" --password=<(exit) 2>/dev/null \
| grep -o -E "0x[A-Fa-f0-9]*" )" )
# balance+=(-n {} -s "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff" -i balance \
# -i "${address[i]}")
balance+=(' "'"${address[i]}"'": { "balance": "0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"}')
done
if [[ ! -f "./genesis.json" ]]
then
EXTRA_DATA="0x3132333400000000000000000000000000000000000000000000000000000000${address[0]#0x}0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
JSON_VAL='{
"config": {
"chainId": '"$CHAINID"',
"homesteadBlock": 0,
"eip150Block": 0,
"eip155Block": 0,
"eip158Block": 0,
"byzantiumBlock": 0,
"constantinopleBlock": 0,
"petersburgBlock": 0,
"istanbulBlock": 0,
"clique": {
"period": '"$PERIOD"',
"epoch": 3000
}
},
"difficulty": "0x1",
"gaslimit": "0xffffffffffff",
"extraData": "'"$EXTRA_DATA"'",
"alloc": {'"$balance"'}
}'
echo $JSON_VAL | jq . > $chaindir/config/genesis.json
geth 2>/dev/null --datadir "$chaindir" init "$chaindir/config/genesis.json"
else
echo "Using local genesis file"
geth 2>/dev/null --datadir "$chaindir" init "./genesis.json"
fi
export ETH_RPC_URL=http://$RPC_ADDRESS:$RPC_PORT
port=$((RPC_PORT + 30000))
geth version
echo >&2 "testnet: RPC URL: $ETH_RPC_URL"
echo >&2 "testnet: DB ADDRESS: $DB_HOST"
echo >&2 "testnet: TCP port: $port"
echo >&2 "testnet: Chain ID: $CHAINID"
echo >&2 "testnet: Database: $chaindir"
echo >&2 "testnet: Geth log: $chaindir/geth.log"
printf "%s\n" "${address[@]}" > "$chaindir/config/account"
echo "$ETH_RPC_URL" > "$chaindir/config/rpc-url"
echo "$port" > "$chaindir/config/node-port"
set +m
# Uncomment below once waitforsync has been merged
# geth \
# 2> >(tee "$chaindir/geth.log" | grep --line-buffered Success | sed 's/^/geth: /' >&2) \
# --datadir "$chaindir" --networkid "$CHAINID" --port="$port" \
# --mine --miner.threads=1 --allow-insecure-unlock \
# --http --http.api "web3,eth,net,debug,personal,statediff" --http.corsdomain '*' --http.vhosts '*' --nodiscover \
# --http.addr="$RPC_ADDRESS" --http.port="$RPC_PORT" --syncmode=full --gcmode=archive \
# --statediff --statediff.db.host="$DB_HOST" --statediff.db.port="$DB_PORT" --statediff.db.user="$DB_USER" \
# --statediff.db.password="$DB_PASSWORD" --statediff.db.name="$DB_NAME" \
# --statediff.db.nodeid 1 --statediff.db.clientname test1 --statediff.writing="$DB_WRITE" \
# --statediff.db.type="$DB_TYPE" --statediff.db.driver="$DB_DRIVER" --statediff.waitforsync="$DB_WAIT_FOR_SYNC" \
# --ws --ws.addr="0.0.0.0" --unlock="$(IFS=,; echo "${address[*]}")" --password=<(exit) &
echo "Starting Geth with following flags"
geth \
2> >(tee "$chaindir/geth.log" | grep --line-buffered Success | sed 's/^/geth: /' >&2) \
--datadir "$chaindir" --networkid "$CHAINID" --port="$port" \
--mine --miner.threads=1 --allow-insecure-unlock \
--http --http.api "admin,debug,eth,miner,net,personal,txpool,web3,statediff" --http.corsdomain '*' --http.vhosts '*' --nodiscover \
--http.addr="$RPC_ADDRESS" --http.port="$RPC_PORT" --syncmode=full --gcmode=archive \
--statediff --statediff.db.host="$DB_HOST" --statediff.db.port="$DB_PORT" --statediff.db.user="$DB_USER" \
--statediff.db.password="$DB_PASSWORD" --statediff.db.name="$DB_NAME" \
--statediff.db.nodeid 1 --statediff.db.clientname test1 --statediff.writing="$DB_WRITE" \
--statediff.db.type="$DB_TYPE" --statediff.db.driver="$DB_DRIVER" \
--ws --ws.addr="0.0.0.0" --ws.origins '*' --ws.api=admin,debug,eth,miner,net,personal,txpool,web3 \
--nat=none --miner.gasprice 16000000000 --nat=none \
--unlock="$(IFS=,; echo "${address[*]}")" --password=<(exit) &
gethpid=$!
echo "Geth started"
echo "Geth PID: $gethpid"
clean() {
( set -x; kill -INT $gethpid; wait )
if [[ $SAVE ]]; then
echo >&2 "testnet: saving $gethdir/snapshots/$SAVE"
mkdir -p "$gethdir/snapshots/$SAVE"
cp -r "$chaindir/keystore" "$gethdir/snapshots/$SAVE"
cp -r "$chaindir/config" "$gethdir/snapshots/$SAVE"
geth >/dev/null 2>&1 --datadir "$chaindir" \
export "$gethdir/snapshots/$SAVE/backup"
fi
( set -x; rm -rf "$chaindir" )
}
trap clean EXIT
echo "Curling: $ETH_RPC_URL"
until curl -s "$ETH_RPC_URL"; do sleep 1; done
echo "Curling: $ETH_RPC_URL complete"
# UPDATE
#ETH_FROM=$(seth --rpc-url="$ETH_RPC_URL" rpc eth_coinbase)
#export ETH_FROM
export ETH_KEYSTORE=$chaindir/keystore
export ETH_PASSWORD=/dev/null
printf 'testnet: Account: %s (default)\n' "${address[0]}" >&2
[[ "${#address[@]}" -gt 1 ]] && printf 'testnet: Account: %s\n' "${address[@]:1}" >&2
echo "Geth Start up completed!"
while true; do sleep 3600; done

View File

@ -1,37 +0,0 @@
version: "3.2"
services:
foundry:
restart: unless-stopped
depends_on:
- ipld-eth-db
build: ./
environment:
DB_USER: vdbm
DB_NAME: vulcanize_testing
DB_HOST: ipld-eth-db
DB_PORT: 5432
DB_PASSWORD: password
DB_WRITE: "true"
DB_TYPE: postgres
DB_DRIVER: sqlx
DB_WAIT_FOR_SYNC: "true"
ports:
- "127.0.0.1:8545:8545"
- "127.0.0.1:8546:8546"
ipld-eth-db:
restart: always
image: vulcanize/ipld-eth-db:v3.0.6
environment:
POSTGRES_USER: "vdbm"
POSTGRES_DB: "vulcanize_testing"
POSTGRES_PASSWORD: "password"
volumes:
- vdb_db_eth_server:/var/lib/postgresql/data
ports:
- "127.0.0.1:8077:5432"
command: ["postgres", "-c", "log_statement=all"]
volumes:
vdb_db_eth_server:

View File

@ -1,50 +0,0 @@
#!/bin/bash
set -ex
# clean up
trap 'killall geth && rm -rf "$TMPDIR"' EXIT
trap "exit 1" SIGINT SIGTERM
TMPDIR=$(mktemp -d)
/bin/bash deploy-local-network.sh --rpc-addr 0.0.0.0 --chain-id 4 --db-user $DB_USER --db-password $DB_PASSWORD --db-name $DB_NAME \
--db-host $DB_HOST --db-port $DB_PORT --db-write $DB_WRITE --dir "$TMPDIR" --address $ADDRESS \
--db-type $DB_TYPE --db-driver $DB_DRIVER --db-waitforsync $DB_WAIT_FOR_SYNC --chain-id $CHAIN_ID &
echo "sleeping 90 sec"
# give it a few secs to start up
sleep 90
# Run tests
cd stateful
forge build
forge test --fork-url http://localhost:8545
# Deploy contracts
ETH_KEYSTORE_FILES=()
echo "ETH KEYSTORE: $TMPDIR/8545/keystore"
for entry in `ls $TMPDIR/8545/keystore`; do
ETH_KEYSTORE_FILES+=("${TMPDIR}/8545/keystore/${entry}")
done
echo "ETH_KEYSTORE_FILES: $ETH_KEYSTORE_FILES"
ETH_KEYSTORE_FILE=${ETH_KEYSTORE_FILES[0]}
if [ "${#ETH_KEYSTORE_FILES[@]}" -eq 1 ]; then
echo "Only one KEYSTORE"
else
echo "WARNING: More than one file in keystore: ${ETH_KEYSTORE_FILES}"
fi
DEPLOYED_ADDRESS=$(forge create --keystore $ETH_KEYSTORE_FILE --rpc-url http://127.0.0.1:8545 --constructor-args 1 --password "" --legacy /root/stateful/src/Stateful.sol:Stateful | grep "Deployed to:" | cut -d " " -f 3)
echo "Contract has been deployed to: $DEPLOYED_ADDRESS"
# Call a transaction
TX_OUT=$(cast send --keystore $ETH_KEYSTORE_FILE --rpc-url http://127.0.0.1:8545 --password "" --legacy $DEPLOYED_ADDRESS "off()")
echo "TX OUTPUT: $TX_OUT"
# Run forever
tail -f /dev/null

View File

@ -1,7 +0,0 @@
[default]
src = 'src'
out = 'out'
libs = ['lib']
remappings = ['ds-test/=lib/ds-test/src/']
# See more config options https://github.com/gakonst/foundry/tree/master/config

View File

@ -1,20 +0,0 @@
// SPDX-License-Identifier: UNLICENSED
pragma solidity 0.8.10;
contract Stateful {
uint x;
constructor(uint y) public {
x = y;
}
function off() public {
require(x == 1);
x = 0;
}
function on() public {
require(x == 0);
x = 1;
}
}

View File

@ -1,19 +0,0 @@
// SPDX-License-Identifier: UNLICENSED
pragma solidity 0.8.10;
import "ds-test/test.sol";
import {Stateful} from "../Stateful.sol";
contract StatefulTest is DSTest {
Stateful contractA;
//contractA A;
uint x;
function setUp() public {
x = 1;
contractA = new Stateful(x);
}
function testExample() public {
contractA.off();
}
}

View File

@ -1,7 +0,0 @@
#!/bin/bash
# This script will run everthing for you. Sit back and enjoy they show.
set -e
./compile-geth.sh
docker-compose up --build

View File

@ -0,0 +1,20 @@
# Overview
The folder will allow developers to clone/move related repositories to this directory. This will allow developers to locally work with these related repositories and `vulcanize/go-ethereum` in one place. This can make testing easier.
# Recommended Setup
## Moving Repositories
It is recommended that you move the following repositories under this folder. Keep the repository names!
- `vulcanize/foundry-tests`
- `vulcanize/ipld-eth-db`
## Symlinks
You can also create symlinks in this folder with the location of your repositories.
# Dependencies
If you want to use `foundry-test`, you will need to utilize the recommended setup.

View File

@ -26,7 +26,10 @@ import (
// Config contains instantiation parameters for the state diffing service
type Config struct {
// The configuration used for the stateDiff Indexer
IndexerConfig interfaces.Config
// The filepath to write knownGaps insert statements if we can't connect to the DB.
KnownGapsFilePath string
// A unique ID used for this service
ID string
// Name for the client this service is running

View File

@ -0,0 +1,17 @@
# Overview
This document will provide some insight into the `known_gaps` table, their use cases, and implementation. Please refer to the [following PR](https://github.com/vulcanize/go-ethereum/pull/217) and the [following epic](https://github.com/vulcanize/ops/issues/143) to grasp their inception.
![known gaps](diagrams/KnownGapsProcess.png)
# Use Cases
The known gaps table is updated when the following events occur:
1. At start up we check the latest block from the `eth.headers_cid` table. We compare the first block that we are processing with the latest block from the DB. If they are not one unit of expectedDifference away from each other, add the gap between the two blocks.
2. If there is any error in processing a block (db connection, deadlock, etc), add that block to the knownErrorBlocks slice, when the next block is successfully written, write this slice into the DB.
# Glossary
1. `expectedDifference (number)` - This number indicates what the difference between two blocks should be. If we are capturing all events on a geth node then this number would be `1`. But once we scale nodes, the `expectedDifference` might be `2` or greater.
2. `processingKey (number)` - This number can be used to keep track of different geth nodes and their specific `expectedDifference`.

3
statediff/docs/README.md Normal file
View File

@ -0,0 +1,3 @@
# Overview
This folder keeps tracks of random documents as they relate to the `statediff` service.

View File

@ -0,0 +1,21 @@
# Overview
This document will go through some notes on the database component of the statediff service.
# Components
- Indexer: The indexer creates IPLD and DB models to insert to the Postgres DB. It performs the insert utilizing and atomic function.
- Builder: The builder constructs the statediff object that needs to be inserted.
- Known Gaps: Captures any gaps that might have occured and either writes them to the DB, local sql file, to prometeus, or a local error.
# Making Code Changes
## Adding a New Function to the Indexer
If you want to implement a new feature for adding data to the database. Keep the following in mind:
1. You need to handle `sql`, `file`, and `dump`.
1. `sql` - Contains the code needed to write directly to the `sql` db.
2. `file` - Contains all the code required to write the SQL statements to a file.
3. `dump` - Contains all the code for outputting events to the console.
2. You will have to add it to the `interfaces.StateDiffIndexer` interface.

Binary file not shown.

After

Width:  |  Height:  |  Size: 33 KiB

View File

@ -31,22 +31,23 @@ import (
"github.com/ethereum/go-ethereum/statediff/indexer/shared"
)
// NewStateDiffIndexer creates and returns an implementation of the StateDiffIndexer interface
func NewStateDiffIndexer(ctx context.Context, chainConfig *params.ChainConfig, nodeInfo node.Info, config interfaces.Config) (interfaces.StateDiffIndexer, error) {
// NewStateDiffIndexer creates and returns an implementation of the StateDiffIndexer interface.
func NewStateDiffIndexer(ctx context.Context, chainConfig *params.ChainConfig, nodeInfo node.Info, config interfaces.Config) (sql.Database, interfaces.StateDiffIndexer, error) {
switch config.Type() {
case shared.FILE:
log.Info("Starting statediff service in SQL file writing mode")
fc, ok := config.(file.Config)
if !ok {
return nil, fmt.Errorf("file config is not the correct type: got %T, expected %T", config, file.Config{})
return nil, nil, fmt.Errorf("file config is not the correct type: got %T, expected %T", config, file.Config{})
}
fc.NodeInfo = nodeInfo
return file.NewStateDiffIndexer(ctx, chainConfig, fc)
ind, err := file.NewStateDiffIndexer(ctx, chainConfig, fc)
return nil, ind, err
case shared.POSTGRES:
log.Info("Starting statediff service in Postgres writing mode")
pgc, ok := config.(postgres.Config)
if !ok {
return nil, fmt.Errorf("postgres config is not the correct type: got %T, expected %T", config, postgres.Config{})
return nil, nil, fmt.Errorf("postgres config is not the correct type: got %T, expected %T", config, postgres.Config{})
}
var err error
var driver sql.Driver
@ -54,25 +55,27 @@ func NewStateDiffIndexer(ctx context.Context, chainConfig *params.ChainConfig, n
case postgres.PGX:
driver, err = postgres.NewPGXDriver(ctx, pgc, nodeInfo)
if err != nil {
return nil, err
return nil, nil, err
}
case postgres.SQLX:
driver, err = postgres.NewSQLXDriver(ctx, pgc, nodeInfo)
if err != nil {
return nil, err
return nil, nil, err
}
default:
return nil, fmt.Errorf("unrecongized Postgres driver type: %s", pgc.Driver)
return nil, nil, fmt.Errorf("unrecognized Postgres driver type: %s", pgc.Driver)
}
return sql.NewStateDiffIndexer(ctx, chainConfig, postgres.NewPostgresDB(driver))
db := postgres.NewPostgresDB(driver)
ind, err := sql.NewStateDiffIndexer(ctx, chainConfig, db)
return db, ind, err
case shared.DUMP:
log.Info("Starting statediff service in data dump mode")
dumpc, ok := config.(dump.Config)
if !ok {
return nil, fmt.Errorf("dump config is not the correct type: got %T, expected %T", config, dump.Config{})
return nil, nil, fmt.Errorf("dump config is not the correct type: got %T, expected %T", config, dump.Config{})
}
return dump.NewStateDiffIndexer(chainConfig, dumpc), nil
return nil, dump.NewStateDiffIndexer(chainConfig, dumpc), nil
default:
return nil, fmt.Errorf("unrecognized database type: %s", config.Type())
return nil, nil, fmt.Errorf("unrecognized database type: %s", config.Type())
}
}

View File

@ -553,3 +553,5 @@ func (sdi *StateDiffIndexer) PushCodeAndCodeHash(batch interfaces.Batch, codeAnd
func (sdi *StateDiffIndexer) Close() error {
return sdi.dbWriter.Close()
}
// Update the known gaps table with the gap information.

View File

@ -54,6 +54,7 @@ type Statements interface {
InsertStorageStm() string
InsertIPLDStm() string
InsertIPLDsStm() string
InsertKnownGapsStm() string
}
// Tx interface to accommodate different concrete SQL transaction types

View File

@ -36,6 +36,7 @@ type DB struct {
}
// InsertHeaderStm satisfies the sql.Statements interface
// Stm == Statement
func (db *DB) InsertHeaderStm() string {
return `INSERT INTO eth.header_cids (block_number, block_hash, parent_hash, cid, td, node_id, reward, state_root, tx_root, receipt_root, uncle_root, bloom, timestamp, mh_key, times_validated, coinbase)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
@ -99,3 +100,10 @@ func (db *DB) InsertIPLDStm() string {
func (db *DB) InsertIPLDsStm() string {
return `INSERT INTO public.blocks (key, data) VALUES (unnest($1::TEXT[]), unnest($2::BYTEA[])) ON CONFLICT (key) DO NOTHING`
}
// InsertKnownGapsStm satisfies the sql.Statements interface
func (db *DB) InsertKnownGapsStm() string {
return `INSERT INTO eth_meta.known_gaps (starting_block_number, ending_block_number, checked_out, processing_key) VALUES ($1, $2, $3, $4)
ON CONFLICT (starting_block_number) DO UPDATE SET (ending_block_number, processing_key) = ($2, $4)
WHERE eth_meta.known_gaps.ending_block_number <= $2`
}

View File

@ -147,3 +147,11 @@ type LogsModel struct {
Topic2 string `db:"topic2"`
Topic3 string `db:"topic3"`
}
// KnownGaps is the data structure for eth_meta.known_gaps
type KnownGapsModel struct {
StartingBlockNumber string `db:"starting_block_number"`
EndingBlockNumber string `db:"ending_block_number"`
CheckedOut bool `db:"checked_out"`
ProcessingKey int64 `db:"processing_key"`
}

278
statediff/known_gaps.go Normal file
View File

@ -0,0 +1,278 @@
// 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 statediff
import (
"context"
"fmt"
"io/ioutil"
"math/big"
"os"
"strings"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
"github.com/ethereum/go-ethereum/statediff/indexer/models"
)
var (
knownGapsInsert = "INSERT INTO eth_meta.known_gaps (starting_block_number, ending_block_number, checked_out, processing_key) " +
"VALUES ('%s', '%s', %t, %d) " +
"ON CONFLICT (starting_block_number) DO UPDATE SET (ending_block_number, processing_key) = ('%s', %d) " +
"WHERE eth_meta.known_gaps.ending_block_number <= '%s';\n"
dbQueryString = "SELECT MAX(block_number) FROM eth.header_cids"
defaultWriteFilePath = "./known_gaps.sql"
)
type KnownGapsState struct {
// Should we check for gaps by looking at the DB and comparing the latest block with head
checkForGaps bool
// Arbitrary processingKey that can be used down the line to differentiate different geth nodes.
processingKey int64
// This number indicates the expected difference between blocks.
// Currently, this is 1 since the geth node processes each block. But down the road this can be used in
// Tandom with the processingKey to differentiate block processing logic.
expectedDifference *big.Int
// Indicates if Geth is in an error state
// This is used to indicate the right time to upserts
errorState bool
// This array keeps track of errorBlocks as they occur.
// When the errorState is false again, we can process these blocks.
// Do we need a list, can we have /KnownStartErrorBlock and knownEndErrorBlock ints instead?
knownErrorBlocks []*big.Int
// The filepath to write SQL statements if we can't connect to the DB.
writeFilePath string
// DB object to use for reading and writing to the DB
db sql.Database
//Do we have entries in the local sql file that need to be written to the DB
sqlFileWaitingForWrite bool
// Metrics object used to track metrics.
statediffMetrics statediffMetricsHandles
}
// Create a new KnownGapsState struct, currently unused.
func NewKnownGapsState(checkForGaps bool, processingKey int64, expectedDifference *big.Int,
errorState bool, writeFilePath string, db sql.Database, statediffMetrics statediffMetricsHandles) *KnownGapsState {
return &KnownGapsState{
checkForGaps: checkForGaps,
processingKey: processingKey,
expectedDifference: expectedDifference,
errorState: errorState,
writeFilePath: writeFilePath,
db: db,
statediffMetrics: statediffMetrics,
}
}
func minMax(array []*big.Int) (*big.Int, *big.Int) {
var max *big.Int = array[0]
var min *big.Int = array[0]
for _, value := range array {
if max.Cmp(value) == -1 {
max = value
}
if min.Cmp(value) == 1 {
min = value
}
}
return min, max
}
// This function actually performs the write of the known gaps. It will try to do the following, it only goes to the next step if a failure occurs.
// 1. Write to the DB directly.
// 2. Write to sql file locally.
// 3. Write to prometheus directly.
// 4. Logs and error.
func (kg *KnownGapsState) pushKnownGaps(startingBlockNumber *big.Int, endingBlockNumber *big.Int, checkedOut bool, processingKey int64) error {
if startingBlockNumber.Cmp(endingBlockNumber) != -1 {
return fmt.Errorf("Starting Block %d, is greater than ending block %d", startingBlockNumber, endingBlockNumber)
}
knownGap := models.KnownGapsModel{
StartingBlockNumber: startingBlockNumber.String(),
EndingBlockNumber: endingBlockNumber.String(),
CheckedOut: checkedOut,
ProcessingKey: processingKey,
}
log.Info("Updating Metrics for the start and end block")
kg.statediffMetrics.knownGapStart.Update(startingBlockNumber.Int64())
kg.statediffMetrics.knownGapEnd.Update(endingBlockNumber.Int64())
var writeErr error
log.Info("Writing known gaps to the DB")
if kg.db != nil {
dbErr := kg.upsertKnownGaps(knownGap)
if dbErr != nil {
log.Warn("Error writing knownGaps to DB, writing them to file instead")
writeErr = kg.upsertKnownGapsFile(knownGap)
}
} else {
writeErr = kg.upsertKnownGapsFile(knownGap)
}
if writeErr != nil {
log.Error("Unsuccessful when writing to a file", "Error", writeErr)
log.Error("Updating Metrics for the start and end error block")
log.Error("Unable to write the following Gaps to DB or File", "startBlock", startingBlockNumber, "endBlock", endingBlockNumber)
kg.statediffMetrics.knownGapErrorStart.Update(startingBlockNumber.Int64())
kg.statediffMetrics.knownGapErrorEnd.Update(endingBlockNumber.Int64())
}
return nil
}
// This is a simple wrapper function to write gaps from a knownErrorBlocks array.
func (kg *KnownGapsState) captureErrorBlocks(knownErrorBlocks []*big.Int) {
startErrorBlock, endErrorBlock := minMax(knownErrorBlocks)
log.Warn("The following Gaps were found", "knownErrorBlocks", knownErrorBlocks)
log.Warn("Updating known Gaps table", "startErrorBlock", startErrorBlock, "endErrorBlock", endErrorBlock, "processingKey", kg.processingKey)
kg.pushKnownGaps(startErrorBlock, endErrorBlock, false, kg.processingKey)
}
// Users provide the latestBlockInDb and the latestBlockOnChain
// as well as the expected difference. This function does some simple math.
// The expected difference for the time being is going to be 1, but as we run
// More geth nodes, the expected difference might fluctuate.
func isGap(latestBlockInDb *big.Int, latestBlockOnChain *big.Int, expectedDifference *big.Int) bool {
latestBlock := big.NewInt(0)
if latestBlock.Sub(latestBlockOnChain, expectedDifference).Cmp(latestBlockInDb) != 0 {
log.Warn("We found a gap", "latestBlockInDb", latestBlockInDb, "latestBlockOnChain", latestBlockOnChain, "expectedDifference", expectedDifference)
return true
}
return false
}
// This function will check for Gaps and update the DB if gaps are found.
// The processingKey will currently be set to 0, but as we start to leverage horizontal scaling
// It might be a useful parameter to update depending on the geth node.
// TODO:
// REmove the return value
// Write to file if err in writing to DB
func (kg *KnownGapsState) findAndUpdateGaps(latestBlockOnChain *big.Int, expectedDifference *big.Int, processingKey int64) error {
// Make this global
latestBlockInDb, err := kg.queryDbToBigInt(dbQueryString)
if err != nil {
return err
}
gapExists := isGap(latestBlockInDb, latestBlockOnChain, expectedDifference)
if gapExists {
startBlock := big.NewInt(0)
endBlock := big.NewInt(0)
startBlock.Add(latestBlockInDb, expectedDifference)
endBlock.Sub(latestBlockOnChain, expectedDifference)
log.Warn("Found Gaps starting at", "startBlock", startBlock, "endingBlock", endBlock)
err := kg.pushKnownGaps(startBlock, endBlock, false, processingKey)
if err != nil {
log.Error("We were unable to write the following gap to the DB", "start Block", startBlock, "endBlock", endBlock, "error", err)
return err
}
}
return nil
}
// Upserts known gaps to the DB.
func (kg *KnownGapsState) upsertKnownGaps(knownGaps models.KnownGapsModel) error {
_, err := kg.db.Exec(context.Background(), kg.db.InsertKnownGapsStm(),
knownGaps.StartingBlockNumber, knownGaps.EndingBlockNumber, knownGaps.CheckedOut, knownGaps.ProcessingKey)
if err != nil {
return fmt.Errorf("error upserting known_gaps entry: %v", err)
}
log.Info("Successfully Wrote gaps to the DB", "startBlock", knownGaps.StartingBlockNumber, "endBlock", knownGaps.EndingBlockNumber)
return nil
}
// Write upsert statement into a local file.
func (kg *KnownGapsState) upsertKnownGapsFile(knownGaps models.KnownGapsModel) error {
insertStmt := []byte(fmt.Sprintf(knownGapsInsert, knownGaps.StartingBlockNumber, knownGaps.EndingBlockNumber, knownGaps.CheckedOut, knownGaps.ProcessingKey,
knownGaps.EndingBlockNumber, knownGaps.ProcessingKey, knownGaps.EndingBlockNumber))
log.Info("Trying to write file")
if kg.writeFilePath == "" {
kg.writeFilePath = defaultWriteFilePath
}
f, err := os.OpenFile(kg.writeFilePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
log.Info("Unable to open a file for writing")
return err
}
defer f.Close()
if _, err = f.Write(insertStmt); err != nil {
log.Info("Unable to open write insert statement to file")
return err
}
log.Info("Wrote the gaps to a local SQL file")
kg.sqlFileWaitingForWrite = true
return nil
}
func (kg *KnownGapsState) writeSqlFileStmtToDb() error {
log.Info("Writing the local SQL file for KnownGaps to the DB")
file, err := ioutil.ReadFile(kg.writeFilePath)
if err != nil {
log.Error("Unable to open local SQL File for writing")
return err
}
requests := strings.Split(string(file), ";")
for _, request := range requests {
_, err := kg.db.Exec(context.Background(), request)
if err != nil {
log.Error("Unable to run insert statement from file to the DB")
return err
}
}
if err := os.Truncate(kg.writeFilePath, 0); err != nil {
log.Info("Failed to empty knownGaps file after inserting statements to the DB", "error", err)
}
kg.sqlFileWaitingForWrite = false
return nil
}
// This is a simple wrapper function which will run QueryRow on the DB
func (kg *KnownGapsState) queryDb(queryString string) (string, error) {
var ret string
err := kg.db.QueryRow(context.Background(), queryString).Scan(&ret)
if err != nil {
log.Error(fmt.Sprint("Can't properly query the DB for query: ", queryString))
return "", err
}
return ret, nil
}
// This function is a simple wrapper which will call QueryDb but the return value will be
// a big int instead of a string
func (kg *KnownGapsState) queryDbToBigInt(queryString string) (*big.Int, error) {
ret := new(big.Int)
res, err := kg.queryDb(queryString)
if err != nil {
return ret, err
}
ret, ok := ret.SetString(res, 10)
if !ok {
log.Error(fmt.Sprint("Can't turn the res ", res, "into a bigInt"))
return ret, fmt.Errorf("Can't turn %s into a bigInt", res)
}
return ret, nil
}

View File

@ -0,0 +1,212 @@
package statediff
import (
"context"
"fmt"
"math/big"
"os"
"testing"
"github.com/ethereum/go-ethereum/metrics"
"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
"github.com/ethereum/go-ethereum/statediff/indexer/database/sql/postgres"
"github.com/stretchr/testify/require"
)
var (
knownGapsFilePath = "./known_gaps.sql"
)
type gapValues struct {
lastProcessedBlock int64
currentBlock int64
knownErrorBlocksStart int64
knownErrorBlocksEnd int64
expectedDif int64
processingKey int64
}
// Add clean db
// Test for failures when they are expected, when we go from smaller block to larger block
// We should no longer see the smaller block in DB
func TestKnownGaps(t *testing.T) {
tests := []gapValues{
// Known Gaps
{knownErrorBlocksStart: 115, knownErrorBlocksEnd: 120, expectedDif: 1, processingKey: 1},
/// Same tests as above with a new expected DIF
{knownErrorBlocksStart: 1150, knownErrorBlocksEnd: 1200, expectedDif: 2, processingKey: 2},
// Test update when block number is larger!!
{knownErrorBlocksStart: 1150, knownErrorBlocksEnd: 1204, expectedDif: 2, processingKey: 2},
// Update when processing key is different!
{knownErrorBlocksStart: 1150, knownErrorBlocksEnd: 1204, expectedDif: 2, processingKey: 10},
}
testWriteToDb(t, tests, true)
testWriteToFile(t, tests, true)
testFindAndUpdateGaps(t, true)
}
// test writing blocks to the DB
func testWriteToDb(t *testing.T, tests []gapValues, wipeDbBeforeStart bool) {
t.Log("Starting Write to DB test")
db := setupDb(t)
// Clear Table first, this is needed because we updated an entry to have a larger endblock number
// so we can't find the original start and endblock pair.
if wipeDbBeforeStart {
t.Log("Cleaning up eth_meta.known_gaps table")
db.Exec(context.Background(), "DELETE FROM eth_meta.known_gaps")
}
for _, tc := range tests {
// Create an array with knownGaps based on user inputs
knownGaps := KnownGapsState{
processingKey: tc.processingKey,
expectedDifference: big.NewInt(tc.expectedDif),
db: db,
statediffMetrics: RegisterStatediffMetrics(metrics.DefaultRegistry),
}
service := &Service{
KnownGaps: knownGaps,
}
knownErrorBlocks := (make([]*big.Int, 0))
knownErrorBlocks = createKnownErrorBlocks(knownErrorBlocks, tc.knownErrorBlocksStart, tc.knownErrorBlocksEnd)
service.KnownGaps.knownErrorBlocks = knownErrorBlocks
// Upsert
testCaptureErrorBlocks(t, service)
// Validate that the upsert was done correctly.
validateUpsert(t, service, tc.knownErrorBlocksStart, tc.knownErrorBlocksEnd)
}
tearDown(t, db)
}
// test writing blocks to file and then inserting them to DB
func testWriteToFile(t *testing.T, tests []gapValues, wipeDbBeforeStart bool) {
t.Log("Starting write to file test")
db := setupDb(t)
// Clear Table first, this is needed because we updated an entry to have a larger endblock number
// so we can't find the original start and endblock pair.
if wipeDbBeforeStart {
t.Log("Cleaning up eth_meta.known_gaps table")
db.Exec(context.Background(), "DELETE FROM eth_meta.known_gaps")
}
if _, err := os.Stat(knownGapsFilePath); err == nil {
err := os.Remove(knownGapsFilePath)
if err != nil {
t.Fatal("Can't delete local file")
}
}
tearDown(t, db)
for _, tc := range tests {
knownGaps := KnownGapsState{
processingKey: tc.processingKey,
expectedDifference: big.NewInt(tc.expectedDif),
writeFilePath: knownGapsFilePath,
statediffMetrics: RegisterStatediffMetrics(metrics.DefaultRegistry),
db: nil, // Only set to nil to be verbose that we can't use it
}
service := &Service{
KnownGaps: knownGaps,
}
knownErrorBlocks := (make([]*big.Int, 0))
knownErrorBlocks = createKnownErrorBlocks(knownErrorBlocks, tc.knownErrorBlocksStart, tc.knownErrorBlocksEnd)
service.KnownGaps.knownErrorBlocks = knownErrorBlocks
testCaptureErrorBlocks(t, service)
newDb := setupDb(t)
service.KnownGaps.db = newDb
if service.KnownGaps.sqlFileWaitingForWrite {
writeErr := service.KnownGaps.writeSqlFileStmtToDb()
require.NoError(t, writeErr)
}
// Validate that the upsert was done correctly.
validateUpsert(t, service, tc.knownErrorBlocksStart, tc.knownErrorBlocksEnd)
tearDown(t, newDb)
}
}
// Find a gap, if no gaps exist, it will create an arbitrary one
func testFindAndUpdateGaps(t *testing.T, wipeDbBeforeStart bool) {
db := setupDb(t)
if wipeDbBeforeStart {
db.Exec(context.Background(), "DELETE FROM eth_meta.known_gaps")
}
knownGaps := KnownGapsState{
processingKey: 1,
expectedDifference: big.NewInt(1),
db: db,
statediffMetrics: RegisterStatediffMetrics(metrics.DefaultRegistry),
}
service := &Service{
KnownGaps: knownGaps,
}
latestBlockInDb, err := service.KnownGaps.queryDbToBigInt("SELECT MAX(block_number) FROM eth.header_cids")
if err != nil {
t.Skip("Can't find a block in the eth.header_cids table.. Please put one there")
}
// Add the gapDifference for testing purposes
gapDifference := big.NewInt(10) // Set a difference between latestBlock in DB and on Chain
expectedDifference := big.NewInt(1) // Set what the expected difference between latestBlock in DB and on Chain should be
latestBlockOnChain := big.NewInt(0)
latestBlockOnChain.Add(latestBlockInDb, gapDifference)
t.Log("The latest block on the chain is: ", latestBlockOnChain)
t.Log("The latest block on the DB is: ", latestBlockInDb)
gapUpsertErr := service.KnownGaps.findAndUpdateGaps(latestBlockOnChain, expectedDifference, 0)
require.NoError(t, gapUpsertErr)
startBlock := big.NewInt(0)
endBlock := big.NewInt(0)
startBlock.Add(latestBlockInDb, gapDifference)
endBlock.Sub(latestBlockOnChain, gapDifference)
validateUpsert(t, service, startBlock.Int64(), endBlock.Int64())
}
// test capturing missed blocks
func testCaptureErrorBlocks(t *testing.T, service *Service) {
service.KnownGaps.captureErrorBlocks(service.KnownGaps.knownErrorBlocks)
}
// Helper function to create an array of gaps given a start and end block
func createKnownErrorBlocks(knownErrorBlocks []*big.Int, knownErrorBlocksStart int64, knownErrorBlocksEnd int64) []*big.Int {
for i := knownErrorBlocksStart; i <= knownErrorBlocksEnd; i++ {
knownErrorBlocks = append(knownErrorBlocks, big.NewInt(i))
}
return knownErrorBlocks
}
// Make sure the upsert was performed correctly
func validateUpsert(t *testing.T, service *Service, startingBlock int64, endingBlock int64) {
t.Logf("Starting to query blocks: %d - %d", startingBlock, endingBlock)
queryString := fmt.Sprintf("SELECT starting_block_number from eth_meta.known_gaps WHERE starting_block_number = %d AND ending_block_number = %d", startingBlock, endingBlock)
_, queryErr := service.KnownGaps.queryDb(queryString) // Figure out the string.
t.Logf("Updated Known Gaps table starting from, %d, and ending at, %d", startingBlock, endingBlock)
require.NoError(t, queryErr)
}
// Create a DB object to use
func setupDb(t *testing.T) sql.Database {
db, err := postgres.SetupSQLXDB()
if err != nil {
t.Error("Can't create a DB connection....")
t.Fatal(err)
}
return db
}
// Teardown the DB
func tearDown(t *testing.T, db sql.Database) {
t.Log("Starting tearDown")
db.Close()
}

View File

@ -50,6 +50,14 @@ type statediffMetricsHandles struct {
// Current length of chainEvent channels
serviceLoopChannelLen metrics.Gauge
writeLoopChannelLen metrics.Gauge
// The start block of the known gap
knownGapStart metrics.Gauge
// The end block of the known gap
knownGapEnd metrics.Gauge
// A known gaps start block which had an error being written to the DB
knownGapErrorStart metrics.Gauge
// A known gaps end block which had an error being written to the DB
knownGapErrorEnd metrics.Gauge
}
func RegisterStatediffMetrics(reg metrics.Registry) statediffMetricsHandles {
@ -59,6 +67,10 @@ func RegisterStatediffMetrics(reg metrics.Registry) statediffMetricsHandles {
lastStatediffHeight: metrics.NewGauge(),
serviceLoopChannelLen: metrics.NewGauge(),
writeLoopChannelLen: metrics.NewGauge(),
knownGapStart: metrics.NewGauge(),
knownGapEnd: metrics.NewGauge(),
knownGapErrorStart: metrics.NewGauge(),
knownGapErrorEnd: metrics.NewGauge(),
}
subsys := "service"
reg.Register(metricName(subsys, "last_sync_height"), ctx.lastSyncHeight)
@ -66,5 +78,9 @@ func RegisterStatediffMetrics(reg metrics.Registry) statediffMetricsHandles {
reg.Register(metricName(subsys, "last_statediff_height"), ctx.lastStatediffHeight)
reg.Register(metricName(subsys, "service_loop_channel_len"), ctx.serviceLoopChannelLen)
reg.Register(metricName(subsys, "write_loop_channel_len"), ctx.writeLoopChannelLen)
reg.Register(metricName(subsys, "known_gaps_start"), ctx.knownGapStart)
reg.Register(metricName(subsys, "known_gaps_end"), ctx.knownGapEnd)
reg.Register(metricName(subsys, "known_gaps_error_start"), ctx.knownGapErrorStart)
reg.Register(metricName(subsys, "known_gaps_error_end"), ctx.knownGapErrorEnd)
return ctx
}

View File

@ -41,8 +41,10 @@ import (
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
ind "github.com/ethereum/go-ethereum/statediff/indexer"
"github.com/ethereum/go-ethereum/statediff/indexer/database/sql"
"github.com/ethereum/go-ethereum/statediff/indexer/interfaces"
nodeinfo "github.com/ethereum/go-ethereum/statediff/indexer/node"
"github.com/ethereum/go-ethereum/statediff/indexer/shared"
types2 "github.com/ethereum/go-ethereum/statediff/types"
"github.com/ethereum/go-ethereum/trie"
)
@ -123,6 +125,8 @@ type Service struct {
BackendAPI ethapi.Backend
// Should the statediff service wait for geth to sync to head?
WaitForSync bool
// Used to signal if we should check for KnownGaps
KnownGaps KnownGapsState
// Whether or not we have any subscribers; only if we do, do we processes state diffs
subscribers int32
// Interface for publishing statediffs as PG-IPLD objects
@ -154,6 +158,7 @@ func NewBlockCache(max uint) BlockCache {
func New(stack *node.Node, ethServ *eth.Ethereum, cfg *ethconfig.Config, params Config, backend ethapi.Backend) error {
blockChain := ethServ.BlockChain()
var indexer interfaces.StateDiffIndexer
var db sql.Database
quitCh := make(chan bool)
if params.IndexerConfig != nil {
info := nodeinfo.Info{
@ -164,16 +169,34 @@ func New(stack *node.Node, ethServ *eth.Ethereum, cfg *ethconfig.Config, params
ClientName: params.ClientName,
}
var err error
indexer, err = ind.NewStateDiffIndexer(params.Context, blockChain.Config(), info, params.IndexerConfig)
db, indexer, err = ind.NewStateDiffIndexer(params.Context, blockChain.Config(), info, params.IndexerConfig)
if err != nil {
return err
}
indexer.ReportDBMetrics(10*time.Second, quitCh)
}
workers := params.NumWorkers
if workers == 0 {
workers = 1
}
// If we ever have multiple processingKeys we can update them here
// along with the expectedDifference
knownGaps := &KnownGapsState{
processingKey: 0,
expectedDifference: big.NewInt(1),
errorState: false,
writeFilePath: params.KnownGapsFilePath,
db: db,
statediffMetrics: statediffMetrics,
sqlFileWaitingForWrite: false,
}
if params.IndexerConfig.Type() == shared.POSTGRES {
knownGaps.checkForGaps = true
} else {
log.Info("We are not going to check for gaps on start up since we are not connected to Postgres!")
knownGaps.checkForGaps = false
}
sds := &Service{
Mutex: sync.Mutex{},
BlockChain: blockChain,
@ -184,6 +207,7 @@ func New(stack *node.Node, ethServ *eth.Ethereum, cfg *ethconfig.Config, params
BlockCache: NewBlockCache(workers),
BackendAPI: backend,
WaitForSync: params.WaitForSync,
KnownGaps: *knownGaps,
indexer: indexer,
enableWriteLoop: params.EnableWriteLoop,
numWorkers: workers,
@ -308,12 +332,42 @@ func (sds *Service) writeLoopWorker(params workerParams) {
sds.writeGenesisStateDiff(parentBlock, params.id)
}
// If for any reason we need to check for gaps,
// Check and update the gaps table.
if sds.KnownGaps.checkForGaps && !sds.KnownGaps.errorState {
log.Info("Checking for Gaps at", "current block", currentBlock.Number())
go sds.KnownGaps.findAndUpdateGaps(currentBlock.Number(), sds.KnownGaps.expectedDifference, sds.KnownGaps.processingKey)
sds.KnownGaps.checkForGaps = false
}
log.Info("Writing state diff", "block height", currentBlock.Number().Uint64(), "worker", params.id)
err := sds.writeStateDiffWithRetry(currentBlock, parentBlock.Root(), writeLoopParams)
if err != nil {
log.Error("statediff.Service.WriteLoop: processing error", "block height", currentBlock.Number().Uint64(), "error", err.Error(), "worker", params.id)
sds.KnownGaps.errorState = true
log.Warn("Updating the following block to knownErrorBlocks to be inserted into knownGaps table", "blockNumber", currentBlock.Number())
sds.KnownGaps.knownErrorBlocks = append(sds.KnownGaps.knownErrorBlocks, currentBlock.Number())
// Write object to startdiff
continue
}
sds.KnownGaps.errorState = false
if sds.KnownGaps.knownErrorBlocks != nil {
// We must pass in parameters by VALUE not reference.
// If we pass them in my reference, the references can change before the computation is complete!
staticKnownErrorBlocks := make([]*big.Int, len(sds.KnownGaps.knownErrorBlocks))
copy(staticKnownErrorBlocks, sds.KnownGaps.knownErrorBlocks)
sds.KnownGaps.knownErrorBlocks = nil
go sds.KnownGaps.captureErrorBlocks(staticKnownErrorBlocks)
}
if sds.KnownGaps.sqlFileWaitingForWrite {
log.Info("There are entries in the SQL file for knownGaps that should be written")
err := sds.KnownGaps.writeSqlFileStmtToDb()
if err != nil {
log.Error("Unable to write KnownGap sql file to DB")
}
}
// TODO: how to handle with concurrent workers
statediffMetrics.lastStatediffHeight.Update(int64(currentBlock.Number().Uint64()))
case <-sds.QuitChan: