Support Vent consuming events from Ethereum web3 JSONRPC

Provided an interface for Vent chain connection o abstract over Burrow and web3-supporting Ethereum chains and implemented for Burrow and web3

Also fix:

- Non-minimal size encoding in RLP (leading zeros)
- Eth-compliant hex strings
- EthRawTx hash and payload where incorrect (ChainID not in payload only
in digest)
- ChainID handling for mainline Ethereum

Added:

- NotEqual and Not operators to query grammar

Signed-off-by: Silas Davis <[email protected]>

.github/workflows/master.yaml

@@ -3,6 +3,7 @@ on:
   push:
     branches:
     - master
+    - release
 
 jobs:
   cover:

.github/workflows/test.yaml

@@ -36,7 +36,8 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v2
-    - run: make test_integration_vent_postgres
+    - run: make test_integration_ethereum
+    - run: make test_integration_vent_complete
 
   docker:
     runs-on: ubuntu-latest

.gitignore

@@ -26,3 +26,5 @@ node_modules
 js/dist
 
 helm/package
+build/
+.pid

Makefile

@@ -24,6 +24,7 @@ BIN_PATH?=$(GOPATH)/bin
 HELM_PATH?=helm/package
 HELM_PACKAGE=$(HELM_PATH)/burrow-$(VERSION).tgz
 ARCH?=linux-amd64
+PID_DIR=.pid
 
 export GO111MODULE=on
 
@@ -226,9 +227,38 @@ test_integration_vent:
 	# Include sqlite adapter with tests - will build with CGO but that's probably fine
 	go test -count=1 -v -tags 'integration sqlite' ./vent/...
 
-.PHONY:	test_integration_vent_postgres
-test_integration_vent_postgres:
-	docker-compose run burrow make test_integration_vent
+.PHONY:	test_integration_vent_complete
+test_integration_vent_complete:
+	docker-compose run burrow make test_integration_vent test_integration_vent_ethereum
+
+.PHONY:	test_integration_vent_ethereum
+test_integration_vent_ethereum: start_ganache
+	go test -count=1 -v -tags 'integration !sqlite ethereum' ./vent/...
+	$(MAKE) stop_ganache
+
+.PHONY:	test_integration_ethereum
+test_integration_ethereum: start_ganache
+	go test -v -tags 'integration ethereum' ./rpc/...
+	$(MAKE) stop_ganache
+
+$(PID_DIR)/ganache.pid:
+	mkdir -p $(PID_DIR)
+	yarn --cwd vent/test/eth install
+	@echo "Starting ganache in background..."
+	{ yarn --cwd vent/test/eth ganache & echo $$! > $@; }
+	@sleep 3
+	@echo "Ganache process started (pid at $@)"
+
+.PHONY: start_ganache
+start_ganache: $(PID_DIR)/ganache.pid
+
+.PHONY: stop_ganache
+stop_ganache: $(PID_DIR)/ganache.pid
+	@kill $(shell cat $<) && echo "Ganache process stopped." && rm $< || rm $<
+
+.PHONY: postgres
+postgres:
+	docker-compose up
 
 .PHONY: test_restore
 test_restore:
@@ -241,7 +271,7 @@ test_integration:
 	@go test -count=1 -v -tags integration ./integration/...
 
 .PHONY: test_integration_all
-test_integration_all: test_keys test_deploy test_integration_vent_postgres test_restore test_truffle test_integration
+test_integration_all: test_keys test_deploy test_integration_vent_complete test_restore test_truffle test_integration
 
 .PHONY: test_integration_all_no_postgres
 test_integration_all_no_postgres: test_keys test_deploy test_integration_vent test_restore test_truffle test_integration

acm/balance/balance.go

@@ -7,7 +7,7 @@ import (
 )
 
 var (
-	eth = new(big.Int).Exp(big.NewInt(10), big.NewInt(18), nil)
+	ethInWei = new(big.Int).Exp(big.NewInt(10), big.NewInt(18), nil)
 )
 
 type Balances []Balance
@@ -134,9 +134,9 @@ func (bs Balances) HasPower() bool {
 func NativeToWei(n uint64) *big.Int {
 	// 1 native unit to 1 ether (wei)
 	x := new(big.Int).SetUint64(n)
-	return new(big.Int).Mul(x, eth)
+	return new(big.Int).Mul(x, ethInWei)
 }
 
 func WeiToNative(x *big.Int) *big.Int {
-	return new(big.Int).Div(x, eth)
+	return new(big.Int).Div(x, ethInWei)
 }

cmd/burrow/commands/vent.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/hyperledger/burrow/config/source"
+	"github.com/hyperledger/burrow/crypto"
 	"github.com/hyperledger/burrow/execution/evm/abi"
 	"github.com/hyperledger/burrow/logging/logconfig"
 	"github.com/hyperledger/burrow/vent/config"
@@ -29,9 +30,11 @@ func Vent(output Output) func(cmd *cli.Cmd) {
 				cfg := config.DefaultVentConfig()
 
 				dbOpts := sqlDBOpts(cmd, cfg)
-				grpcAddrOpt := cmd.StringOpt("grpc-addr", cfg.GRPCAddr, "Address to connect to the Hyperledger Burrow gRPC server")
-				httpAddrOpt := cmd.StringOpt("http-addr", cfg.HTTPAddr, "Address to bind the HTTP server")
+				grpcAddrOpt := cmd.StringOpt("chain-addr", cfg.ChainAddress, "Address to connect to the Hyperledger Burrow gRPC server")
+				httpAddrOpt := cmd.StringOpt("http-addr", cfg.HTTPListenAddress, "Address to bind the HTTP server")
 				logLevelOpt := cmd.StringOpt("log-level", cfg.LogLevel, "Logging level (error, warn, info, debug)")
+				watchAddressesOpt := cmd.StringsOpt("watch", nil, "Add contract address to global watch filter")
+				minimumHeightOpt := cmd.IntOpt("minimum-height", 0, "Add contract address to global watch filter")
 				abiFileOpt := cmd.StringsOpt("abi", cfg.AbiFileOrDirs, "EVM Contract ABI file or folder")
 				specFileOrDirOpt := cmd.StringsOpt("spec", cfg.SpecFileOrDirs, "SQLSol specification file or folder")
 				dbBlockOpt := cmd.BoolOpt("blocks", false, "Create block tables and persist related data")
@@ -44,9 +47,18 @@ func Vent(output Output) func(cmd *cli.Cmd) {
 					cfg.DBAdapter = *dbOpts.adapter
 					cfg.DBURL = *dbOpts.url
 					cfg.DBSchema = *dbOpts.schema
-					cfg.GRPCAddr = *grpcAddrOpt
-					cfg.HTTPAddr = *httpAddrOpt
+					cfg.ChainAddress = *grpcAddrOpt
+					cfg.HTTPListenAddress = *httpAddrOpt
 					cfg.LogLevel = *logLevelOpt
+					cfg.WatchAddresses = make([]crypto.Address, len(*watchAddressesOpt))
+					cfg.MinimumHeight = uint64(*minimumHeightOpt)
+					var err error
+					for i, wa := range *watchAddressesOpt {
+						cfg.WatchAddresses[i], err = crypto.AddressFromHexString(wa)
+						if err != nil {
+							output.Fatalf("could not parse watch address: %w", err)
+						}
+					}
 					cfg.AbiFileOrDirs = *abiFileOpt
 					cfg.SpecFileOrDirs = *specFileOrDirOpt
 					if *dbBlockOpt {
@@ -65,8 +77,10 @@ func Vent(output Output) func(cmd *cli.Cmd) {
 					}
 				}
 
-				cmd.Spec = "--spec=<spec file or dir> [--abi=<abi file or dir>] [--db-adapter] [--db-url] [--db-schema] " +
-					"[--blocks] [--txs] [--grpc-addr] [--http-addr] [--log-level] [--announce-every=<duration>]"
+				cmd.Spec = "--spec=<spec file or dir>... [--abi=<abi file or dir>...] " +
+					"[--watch=<contract address>...] [--minimum-height=<lowest height from which to read>] " +
+					"[--db-adapter] [--db-url] [--db-schema] [--blocks] [--txs] [--chain-addr] [--http-addr] " +
+					"[--log-level] [--announce-every=<duration>]"
 
 				cmd.Action = func() {
 					log, err := logconfig.New().NewLogger()

core/kernel.go

@@ -20,6 +20,7 @@ import (
 	"github.com/hyperledger/burrow/consensus/tendermint"
 	"github.com/hyperledger/burrow/crypto"
 	"github.com/hyperledger/burrow/dump"
+	"github.com/hyperledger/burrow/rpc/web3"
 
 	// GRPC Codec
 	_ "github.com/hyperledger/burrow/encoding"
@@ -52,7 +53,7 @@ type Kernel struct {
 	// Expose these public-facing interfaces to allow programmatic extension of the Kernel by other projects
 	Emitter        *event.Emitter
 	Service        *rpc.Service
-	EthService     *rpc.EthService
+	EthService     *web3.EthService
 	Launchers      []process.Launcher
 	State          *state.State
 	Blockchain     *bcm.Blockchain

core/processes.go

@@ -143,7 +143,7 @@ func TendermintLauncher(kern *Kernel) process.Launcher {
 			nodeRegState := kern.State
 			validatorState := kern.State
 			kern.Service = rpc.NewService(accountState, nameRegState, nodeRegState, kern.Blockchain, validatorState, nodeView, kern.Logger)
-			kern.EthService = rpc.NewEthService(accountState, eventsState, kern.Blockchain, validatorState, nodeView, kern.Transactor, kern.keyStore, kern.Logger)
+			kern.EthService = web3.NewEthService(accountState, eventsState, kern.Blockchain, validatorState, nodeView, kern.Transactor, kern.keyStore, kern.Logger)
 
 			if err := kern.Node.Start(); err != nil {
 				return nil, fmt.Errorf("%s error starting Tendermint node: %v", errHeader, err)

crypto/crypto.go

@@ -74,3 +74,7 @@ type Signer interface {
 type Signable interface {
 	SignBytes(chainID string) ([]byte, error)
 }
+
+func (pk *PrivateKey) GetAddress() Address {
+	return pk.GetPublicKey().GetAddress()
+}

crypto/signature.go

@@ -118,7 +118,12 @@ func (sig *Signature) String() string {
 }
 
 func GetEthChainID(chainID string) *big.Int {
-	return new(big.Int).SetBytes(Keccak256([]byte(chainID)))
+	b := new(big.Int)
+	id, ok := b.SetString(chainID, 10)
+	if ok {
+		return id
+	}
+	return b.SetBytes([]byte(chainID))
 }
 
 func GetEthSignatureRecoveryID(chainID string, parity *big.Int) *big.Int {

deploy/compile/compilers_test.go

@@ -40,7 +40,7 @@ func TestLocalMulti(t *testing.T) {
 	expectedSolcResponse := BlankSolcResponse()
 	actualOutput, err := exec.Command("solc", "--combined-json", "bin,abi", "contractImport1.sol").CombinedOutput()
 	if err != nil {
-		t.Fatal(err)
+		t.Fatalf("solc failed %v: %s", err, actualOutput)
 	}
 
 	warning, responseJSON := extractWarningJSON(strings.TrimSpace(string(actualOutput)))
@@ -85,7 +85,7 @@ func TestLocalSingle(t *testing.T) {
 	shellCmd := exec.Command("solc", "--combined-json", "bin,abi", "simpleContract.sol")
 	actualOutput, err := shellCmd.CombinedOutput()
 	if err != nil {
-		t.Fatal(err)
+		t.Fatalf("solc failed %v: %s", err, actualOutput)
 	}
 
 	warning, responseJSON := extractWarningJSON(strings.TrimSpace(string(actualOutput)))

encoding/rlp/rlp.go

@@ -90,15 +90,22 @@ func Decode(src []byte, dst interface{}) error {
 			}
 		}
 	case reflect.Struct:
-		if val.NumField() != len(fields) {
-			return fmt.Errorf("wrong number of fields; have %d, want %d", len(fields), val.NumField())
-		}
+		rt := val.Type()
+		numExportedFields := 0
 		for i := 0; i < val.NumField(); i++ {
-			err := decodeField(val.Field(i), fields[i])
-			if err != nil {
-				return err
+			// Skip unexported fields
+			if rt.Field(i).PkgPath == "" {
+				err := decodeField(val.Field(i), fields[numExportedFields])
+				if err != nil {
+					return err
+				}
+				numExportedFields++
 			}
 		}
+		if numExportedFields != len(fields) {
+			return fmt.Errorf("wrong number of fields; have %d, want %d", len(fields), numExportedFields)
+		}
+
 	default:
 		return fmt.Errorf("cannot decode into unsupported type %v", reflect.TypeOf(dst))
 	}
@@ -146,11 +153,11 @@ func encodeLength(n int, offset magicOffset) []byte {
 	if n <= ShortLength {
 		return []uint8{uint8(offset) + uint8(n)}
 	}
-
 	i := uint64(n)
 	b := make([]byte, 8)
 	binary.BigEndian.PutUint64(b, i)
-	byteLengthOfLength := bits.Len64(i)/8 + 1
+	// Byte-wise ceiling
+	byteLengthOfLength := (bits.Len64(i) + 7) / 8
 	// > If a string is more than 55 bytes long, the RLP encoding consists of a single byte with value 0xb7
 	// > plus the length in bytes of the length of the string in binary form, followed by the length of the string,
 	// > followed by the string
@@ -186,12 +193,18 @@ func encodeList(val reflect.Value) ([]byte, error) {
 func encodeStruct(val reflect.Value) ([]byte, error) {
 	out := make([][]byte, 0)
 
+	rt := val.Type()
+
 	for i := 0; i < val.NumField(); i++ {
-		data, err := encode(val.Field(i))
-		if err != nil {
-			return nil, err
+		field := val.Field(i)
+		// Skip unexported fields
+		if rt.Field(i).PkgPath == "" {
+			data, err := encode(field)
+			if err != nil {
+				return nil, err
+			}
+			out = append(out, data)
 		}
-		out = append(out, data)
 	}
 	sum := bytes.Join(out, []byte{})
 	length := encodeLength(len(sum), SliceOffset)