update

.gitignore

@@ -7,3 +7,4 @@ gen
 .DS_Store
 *.exe
 build
+data
\ No newline at end of file

Makefile

 .PHONY: default all clean dev
 
-GOBIN = $(shell pwd)/build
+GOBIN = $(shell pwd)/build/bin
 
 GOVERSION=$(shell go version | awk '{print $$3}')
 GITHASH=$(shell git show -s --format=%H)
@@ -19,13 +19,15 @@ BUILD_FLAGS=-ldflags "\
 
 
 witness:
-	# go build  -o MetaNet
-	go build $(BUILD_FLAGS) -o=${GOBIN}/$@ ./cmd
+	go build $(BUILD_FLAGS) -v -o=${GOBIN}/$@ ./cmd/witness
 
 
 dev:
-	go build $(BUILD_FLAGS) -o=${GOBIN}/$@ -gcflags "all=-N -l" ./cmd
+	go build $(BUILD_FLAGS) -v -o=${GOBIN}/$@ -gcflags "all=-N -l" ./cmd/witness
 
 
 clean:
-	rm -fr build
+	rm -rf build
+
+cleandb:
+	rm -rf data/*
\ No newline at end of file

api/grpc.go

 package api
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"math/big"
+	"net"
+	"witness/core"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
+	witnessv1 "github.com/odysseus/odysseus-protocol/gen/proto/go/witness/v1"
+	log "github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+)
+
+type Server struct {
+	witnessv1.UnimplementedWitnessServiceServer
+	w *core.Witness
+}
+
+func (s *Server) WitnessStatus(ctx context.Context, req *witnessv1.WitnessStatusRequest) (resp *witnessv1.WitnessStatusResponse, err error) {
+	return nil, nil
+}
+
+func (s *Server) PushProof(ctx context.Context, req *witnessv1.PushProofRequest) (resp *witnessv1.PushProofResponse, err error) {
+	resp = &witnessv1.PushProofResponse{}
+	proofs := req.GetProofs()
+	if len(proofs) == 0 {
+		return nil, errors.New("no proofs provided")
+	}
+	miner := common.HexToAddress(req.MinerAddress)
+	reward := common.HexToAddress(req.RewardAddress)
+	validatedProofs := make([]*witnessv1.ValidatedProof, 0, len(proofs))
+
+	log.WithFields(log.Fields{"miner": miner, "reward": reward, "proofs": len(proofs)}).Debug("push proof")
+
+	for _, proof := range proofs {
+		hashPayload := append(append([]byte(proof.TaskId), proof.ReqHash...), proof.RespHash...)
+		containerPubKey, err := crypto.SigToPub(crypto.Keccak256(hashPayload), proof.ContainerSignature)
+		if err != nil {
+			log.WithError(err).Error("failed to verify container signature")
+			return nil, err
+		}
+		containerSigner := crypto.PubkeyToAddress(*containerPubKey)
+		verified := s.w.IsContainerAddress(containerSigner)
+		if !verified {
+			log.WithFields(log.Fields{"signer": containerSigner}).Error("invalid container signature")
+			return nil, errors.New("invalid container signature")
+		}
+
+		minerHashPayload := append(append([]byte(proof.TaskId), proof.ReqHash...), proof.RespHash...)
+		minerPubKey, err := crypto.SigToPub(crypto.Keccak256(minerHashPayload), proof.MinerSignature)
+		if err != nil {
+			log.WithError(err).Error("failed to verify miner signature")
+			return nil, err
+		}
+
+		minerAddress := crypto.PubkeyToAddress(*minerPubKey)
+		if minerAddress.Hex() != miner.Hex() {
+			log.WithFields(log.Fields{"miner": miner, "signer": minerAddress}).Error("invalid miner address")
+			return nil, errors.New("invalid miner address")
+		}
+
+		managerHashPayload := bytes.NewBuffer(hashPayload)
+		managerHashPayload.Write(reward.Bytes())
+		managerHashPayload.Write(miner.Bytes())
+		managerHashPayload.Write(proof.ContainerSignature)
+		managerHashPayload.Write(proof.MinerSignature)
+		managerHashPayload.Write(big.NewInt(int64(proof.Workload)).Bytes())
+		managerPubKey, err := crypto.SigToPub(crypto.Keccak256(managerHashPayload.Bytes()), proof.ManagerSignature)
+		if err != nil {
+			log.WithError(err).Error("failed to verify manager signature")
+			return nil, err
+		}
+
+		managerSigner := crypto.PubkeyToAddress(*managerPubKey)
+		verified = s.w.IsNMAddress(managerSigner)
+		if !verified {
+			log.WithFields(log.Fields{"signer": managerSigner}).Error("invalid manager signature")
+			return nil, errors.New("invalid manager signature")
+		}
+		resp.Workload += proof.Workload
+		validatedProofs = append(validatedProofs, &witnessv1.ValidatedProof{
+			Workload: proof.Workload,
+		})
+	}
+	s.w.AddPendingProof(miner, validatedProofs)
+	return resp, nil
+}
+
+func (s *Server) GetMinerProof(ctx context.Context, req *witnessv1.GetMinerProofRequest) (*witnessv1.GetMinerProofResponse, error) {
+	return nil, nil
+}
+
+func (s *Server) GetProof(ctx context.Context, req *witnessv1.GetProofRequest) (*witnessv1.GetProofResponse, error) {
+	return nil, nil
+}
+
+func (s *Server) GetWithdrawProof(ctx context.Context, req *witnessv1.GetWithdrawProofRequest) (*witnessv1.GetWithdrawProofResponse, error) {
+	return nil, nil
+}
+
+func StartGRPC(listenAddress string, w *core.Witness) {
+	ln, err := net.Listen("tcp", listenAddress)
+	if err != nil {
+		log.WithError(err).Errorf("failed to listen on %s", listenAddress)
+		return
+	}
+
+	log.WithField("listen", listenAddress).Info("start gRPC server")
+	server := grpc.NewServer(grpc.MaxRecvMsgSize(1024*1024*1024), grpc.MaxSendMsgSize(1024*1024*1024))
+	witnessv1.RegisterWitnessServiceServer(server, &Server{w: w})
+	grpc_prometheus.Register(server)
+	err = server.Serve(ln)
+	if err != nil {
+		log.WithError(err).Error("failed to serve")
+		return
+	}
+	server.Stop()
+}

api/grpc_test.go

+package api
+
+import (
+	"context"
+	"testing"
+
+	witnessv1 "github.com/odysseus/odysseus-protocol/gen/proto/go/witness/v1"
+	log "github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+func TestClient(t *testing.T) {
+	client, err := grpc.Dial("127.0.0.1:9431", grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithDefaultCallOptions(
+			grpc.MaxCallRecvMsgSize(1024*1024*1024),
+			grpc.MaxCallSendMsgSize(1024*1024*1024)),
+	)
+	if err != nil {
+		panic(err)
+	}
+
+	gc := witnessv1.NewWitnessServiceClient(client)
+
+	req := &witnessv1.PushProofRequest{
+		Proofs: []*witnessv1.Proof{
+			{
+				Workload:           10,
+				TaskId:             "1",
+				ReqHash:            []byte("req"),
+				RespHash:           []byte("resp"),
+				ManagerSignature:   nil,
+				ContainerSignature: nil,
+				MinerSignature:     nil,
+			},
+		},
+		RewardAddress: "",
+		MinerAddress:  "",
+	}
+
+	resp, err := gc.PushProof(context.Background(), req)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	log.Info(resp.Workload)
+}

api/json.go

+package api
+
+import (
+	"encoding/json"
+)
+
+type jsonrpcMessage struct {
+	Version string          `json:"jsonrpc"`
+	ID      json.RawMessage `json:"id"`
+	Method  string          `json:"method,omitempty"`
+	Params  json.RawMessage `json:"params,omitempty"`
+	Error   *jsonError      `json:"error,omitempty"`
+	Result  json.RawMessage `json:"result,omitempty"`
+}
+
+type jsonError struct {
+	Code    int         `json:"code"`
+	Message string      `json:"message"`
+	Data    interface{} `json:"data,omitempty"`
+}

api/jsonrpc.go

-package api

api/metrics.go

 package api
-
-import (
-	"github.com/prometheus/client_golang/prometheus"
-)
-
-var (
-	_ = prometheus.AlreadyRegisteredError{}
-)

api/rpc.go

+package api
+
+import (
+	"encoding/json"
+	"net/http"
+	"witness/core"
+
+	"github.com/ethereum/go-ethereum/common"
+	geth "github.com/ethereum/go-ethereum/mobile"
+	log "github.com/sirupsen/logrus"
+)
+
+var witness *core.Witness
+
+func rpcHandle(w http.ResponseWriter, r *http.Request) {
+	req := &jsonrpcMessage{}
+	resp := &jsonrpcMessage{}
+
+	if r.Method != "POST" {
+		w.Write([]byte("method not allowed"))
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	contentType := r.Header.Get("Content-Type")
+	if contentType != "application/json" {
+		resp.Error = &jsonError{
+			Code:    -32600,
+			Message: "invalid Content-Type header",
+		}
+		resp.Version = "2.0"
+		_ = json.NewEncoder(w).Encode(resp)
+		return
+	}
+
+	err := json.NewDecoder(r.Body).Decode(req)
+	if err != nil {
+		resp.Error = &jsonError{
+			Code:    -32603,
+			Message: err.Error(),
+		}
+		resp.Version = "2.0"
+		_ = json.NewEncoder(w).Encode(resp)
+		return
+	}
+
+	if req.Version != "2.0" {
+		resp.Error = &jsonError{
+			Code:    -32600,
+			Message: "invalid jsonrpc version",
+		}
+		resp.Version = "2.0"
+		_ = json.NewEncoder(w).Encode(resp)
+		return
+	}
+	resp.Version = req.Version
+	resp.ID = req.ID
+
+	switch req.Method {
+	case "getWithdrawProofs":
+		getWithdrawProofs(req.Params, resp)
+		_ = json.NewEncoder(w).Encode(resp)
+		return
+	case "getPendingWorkload":
+		getPendingWorkload(req.Params, resp)
+		_ = json.NewEncoder(w).Encode(resp)
+		return
+
+	default:
+		resp.Error = &jsonError{
+			Code:    -32601,
+			Message: "method not found",
+		}
+		_ = json.NewEncoder(w).Encode(resp)
+		return
+	}
+
+}
+
+func getWithdrawProofs(params []byte, resp *jsonrpcMessage) {
+	addressList := make([]string, 0)
+	err := json.Unmarshal(params, &addressList)
+	if err != nil || len(addressList) != 1 {
+		resp.Error = &jsonError{
+			Code:    -32602,
+			Message: "invalid params",
+		}
+		return
+	}
+
+	_, err = geth.NewAddressFromHex(addressList[0])
+	if err != nil {
+		resp.Error = &jsonError{
+			Code:    -32602,
+			Message: "invalid params",
+		}
+		return
+	}
+
+	workload, proofs := witness.GetMerkleProof(common.HexToAddress(addressList[0]))
+
+	temp := map[string]interface{}{
+		"workload": workload,
+		"proofs":   proofs,
+	}
+
+	resp.Result, _ = json.Marshal(temp)
+	return
+}
+
+func getPendingWorkload(params []byte, resp *jsonrpcMessage) {
+	addressList := make([]string, 0)
+	err := json.Unmarshal(params, &addressList)
+	if err != nil || len(addressList) != 1 {
+		resp.Error = &jsonError{
+			Code:    -32602,
+			Message: "invalid params",
+		}
+		return
+	}
+
+	_, err = geth.NewAddressFromHex(addressList[0])
+	if err != nil {
+		resp.Error = &jsonError{
+			Code:    -32602,
+			Message: "invalid params",
+		}
+		return
+	}
+
+	workload := witness.GetPendingWorkload(common.HexToAddress(addressList[0]))
+
+	temp := map[string]interface{}{
+		"workload": workload,
+	}
+
+	resp.Result, _ = json.Marshal(temp)
+	return
+}
+
+func StartJSONRPC(listenAddress string, w *core.Witness) {
+	witness = w
+	http.HandleFunc("/", rpcHandle)
+	log.WithField("listen", listenAddress).Info("start JSON-RPC server")
+	err := http.ListenAndServe(listenAddress, nil)
+	if err != nil {
+		log.WithError(err).Fatal("failed to start JSON-RPC server")
+	}
+}

api/service.go

-package api

cmd/main.go

-package main
-
-import (
-	"fmt"
-	"os"
-	"witness/version"
-
-	log "github.com/sirupsen/logrus"
-	"github.com/urfave/cli/v2"
-)
-
-var (
-	appFlags = []cli.Flag{
-		configFileFlag,
-		grpcListenAddrFlag,
-		rpcListenAddrFlag,
-		rpcHostFlag,
-		storeContractFlag,
-		rewardContractFlag,
-	}
-)
-
-func main() {
-	// run()
-	fmt.Println(version.GOVersion)
-}
-
-func run() {
-	app := cli.App{}
-	app.Name = "witness"
-	app.Usage = "this is witness"
-	app.Version = version.Version
-	app.Before = func(c *cli.Context) error {
-		return loadFlagsFromConfig(c, appFlags)
-	}
-
-	if err := app.Run(os.Args); err != nil {
-		log.Error(err)
-	}
-}

cmd/flags.go → cmd/witness/flags.go

@@ -12,6 +12,26 @@ var (
 		Name:    "config",
 		Aliases: []string{"c"},
 		Usage:   "The path to the config file",
+		Value:   "config.toml",
+	}
+
+	dataDirFlag = &cli.StringFlag{
+		Name:    "data-dir",
+		Aliases: []string{"d"},
+		Usage:   "The path to the data directory",
+		Value:   "data",
+	}
+
+	metricsListenAddrFlag = &cli.StringFlag{
+		Name:  "metrics-listen",
+		Usage: "The listen address of the metrics server",
+		Value: "0.0.0.0:9429",
+	}
+
+	logLevelFlag = &cli.StringFlag{
+		Name:  "log-level",
+		Usage: "The log level",
+		Value: "info",
 	}
 
 	grpcListenAddrFlag = &cli.StringFlag{
@@ -26,7 +46,13 @@ var (
 		Value: "0.0.0.0:9431",
 	}
 
-	rpcHostFlag = &cli.StringFlag{
+	privateKeyFlag = &cli.StringFlag{
+		Name:    "private-key",
+		Usage:   "The private key of the account",
+		EnvVars: []string{"PRIVATE_KEY"},
+	}
+
+	chainRPCFlag = &cli.StringFlag{
 		Name:  "chain-rpc",
 		Usage: "The host of the chain json rpc server",
 	}
@@ -36,8 +62,8 @@ var (
 		Usage: "The address of the store contract",
 	}
 
-	rewardContractFlag = &cli.StringFlag{
-		Name:  "reward-contract",
+	witnessContractFlag = &cli.StringFlag{
+		Name:  "witness-contract",
 		Usage: "The address of the reward contract",
 	}
 )

cmd/witness/main.go

+package main
+
+import (
+	"net/http"
+	"os"
+	"witness/api"
+	"witness/conf"
+	"witness/core"
+	"witness/version"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+	log "github.com/sirupsen/logrus"
+	"github.com/urfave/cli/v2"
+)
+
+var (
+	appFlags = []cli.Flag{
+		configFileFlag,
+		dataDirFlag,
+		metricsListenAddrFlag,
+		logLevelFlag,
+		grpcListenAddrFlag,
+		rpcListenAddrFlag,
+		privateKeyFlag,
+		chainRPCFlag,
+		storeContractFlag,
+		witnessContractFlag,
+	}
+)
+
+func main() {
+	app := cli.App{}
+	app.Flags = wrapFlags(appFlags)
+	app.Name = "witness"
+	app.Usage = "this is witness"
+	app.Version = version.Version
+	app.Before = func(c *cli.Context) error {
+		return loadFlagsFromConfig(c, app.Flags)
+	}
+	app.Action = func(context *cli.Context) error {
+		run(context)
+		return nil
+	}
+	if err := app.Run(os.Args); err != nil {
+		log.Error(err)
+	}
+}
+
+func run(ctx *cli.Context) {
+	cfg := &conf.Config{
+		MetricsListenAddr: ctx.String(metricsListenAddrFlag.Name),
+		LogLevel:          ctx.String(logLevelFlag.Name),
+		GRPCListenAddr:    ctx.String(grpcListenAddrFlag.Name),
+		RPCListenAddr:     ctx.String(rpcListenAddrFlag.Name),
+		PrivateKey:        ctx.String(privateKeyFlag.Name),
+		ChainRPC:          ctx.String(chainRPCFlag.Name),
+		StoreContract:     ctx.String(storeContractFlag.Name),
+		WitnessContract:   ctx.String(witnessContractFlag.Name),
+		DataDir:           ctx.String(dataDirFlag.Name),
+	}
+	setLogLevel(cfg.LogLevel)
+	runMetrics(cfg.MetricsListenAddr)
+	w := core.NewWitness(cfg)
+	runGrpcServer(cfg.GRPCListenAddr, w)
+	runJSONRPCServer(cfg.RPCListenAddr, w)
+	select {}
+}
+
+func setLogLevel(level string) {
+	_level, err := log.ParseLevel(level)
+	if err != nil {
+		log.WithError(err).Error("failed to parse log level")
+		return
+	}
+	log.WithField("log level", _level).Info("set log level")
+	log.SetLevel(_level)
+}
+
+func runMetrics(listen string) {
+	http.Handle("/metrics", promhttp.Handler())
+	log.WithField("listen", listen).Info("start prometheus metrics server")
+	go func() {
+		if err := http.ListenAndServe(listen, nil); err != nil {
+			log.WithError(err).Error("failed to start prometheus metrics server")
+		}
+	}()
+}
+
+func runGrpcServer(listen string, w *core.Witness) {
+	go api.StartGRPC(listen, w)
+}
+
+func runJSONRPCServer(listen string, w *core.Witness) {
+	go api.StartJSONRPC(listen, w)
+}

conf/config.go

+package conf
+
+type Config struct {
+	MetricsListenAddr string
+	LogLevel          string
+	GRPCListenAddr    string
+	RPCListenAddr     string
+	ChainRPC          string
+	PrivateKey        string
+	StoreContract     string
+	WitnessContract   string
+	RewardContract    string
+	DataDir           string
+}

config.toml

 
-grpc-listen = "0.0.0.0:9430"
+log-level = "debug"
 
-rpc-listen = "0.0.0.0:9431"
+metrics-listen = "0.0.0.0:8425"
 
-chain-rpc = "https://1rpc.io/eth"
+grpc-listen = "0.0.0.0:9431"
 
-store-contract = "0x0000000000000000000000000000000000000000"
+rpc-listen = "0.0.0.0:9442"
 
-reward-contract = "0x0000000000000000000000000000000000000000"
+chain-rpc = "https://1rpc.io/holesky"
\ No newline at end of file
+
+private-key = "529f4efb80ac534f17d873104c71881c0970dbd5a886f183f63c5c6bb7a1fcd9"
+
+store-contract = "0x7Cd36Bc2a477f60A14f08442179B2f626bE026Ea"
+
+witness-contract = "0xf49133dD7B7ed75fA0f877413D293c05Bff0D8F0"
\ No newline at end of file

core/chain.go

+package core
+
+import (
+	"context"
+	"crypto/ecdsa"
+	"fmt"
+	"math/big"
+	"time"
+	"witness/contract"
+
+	"github.com/btcsuite/btcd/btcec/v2"
+	"github.com/ethereum/go-ethereum"
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethclient"
+	log "github.com/sirupsen/logrus"
+)
+
+type ChainRPC struct {
+	rpc             *ethclient.Client
+	chainID         *big.Int
+	privateKey      *ecdsa.PrivateKey
+	storageContract *contract.AddressStorage
+	witnessContract *contract.Witness
+}
+
+func newChain(rpc, privateKey, storageCa, witnessCa string) *ChainRPC {
+	ethRpc, err := ethclient.Dial(rpc)
+	if err != nil {
+		panic(err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	chainID, err := ethRpc.ChainID(ctx)
+	if err != nil {
+		panic(err)
+	}
+
+	log.WithField("chain_id", chainID.String()).Info("rpc get chain id")
+
+	storage, err := contract.NewAddressStorage(common.HexToAddress(storageCa), ethRpc)
+	if err != nil {
+		panic(err)
+	}
+
+	witness, err := contract.NewWitness(common.HexToAddress(witnessCa), ethRpc)
+	if err != nil {
+		panic(err)
+	}
+
+	binPrivateKey, err := hexutil.Decode(privateKey)
+	if err == hexutil.ErrMissingPrefix {
+		binPrivateKey, err = hexutil.Decode("0x" + privateKey)
+		if err != nil {
+			panic(err)
+		}
+	}
+	if err != nil && err != hexutil.ErrMissingPrefix {
+		panic(err)
+	}
+
+	k, _ := btcec.PrivKeyFromBytes(binPrivateKey)
+	return &ChainRPC{rpc: ethRpc, storageContract: storage, witnessContract: witness, chainID: chainID, privateKey: k.ToECDSA()}
+}
+
+func (r *ChainRPC) GetContainerAddresses() (addrs []common.Address, err error) {
+	return r.storageContract.GetContainerAddresses(nil)
+}
+
+func (r *ChainRPC) GetNMAddresses() (addrs []common.Address, err error) {
+	return r.storageContract.GetNmAddresses(nil)
+}
+
+func (r *ChainRPC) SubmitProofs(date string, merkleSumTreeRoot, merkleTreeRoot common.Hash) (txHash common.Hash, err error) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+	opts, err := bind.NewKeyedTransactorWithChainID(r.privateKey, r.chainID)
+	if err != nil {
+		return
+	}
+	opts.Context = ctx
+	signedTx, err := r.witnessContract.SubmitMerkleRoot(opts, date, merkleSumTreeRoot, merkleTreeRoot)
+	if err != nil {
+		return
+	}
+
+	receipt, err := r.WaitForReceipt(ctx, signedTx.Hash())
+	if err != nil {
+		return
+	}
+
+	if receipt.Status != types.ReceiptStatusSuccessful {
+		return common.Hash{}, fmt.Errorf("tx %s failed", signedTx.Hash().Hex())
+	}
+
+	return signedTx.Hash(), nil
+}
+
+func (r *ChainRPC) WaitForReceipt(ctx context.Context, txHash common.Hash) (receipt *types.Receipt, err error) {
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		default:
+			receipt, err = r.rpc.TransactionReceipt(ctx, txHash)
+			if err == nil {
+				return receipt, nil
+			}
+			if err == ethereum.NotFound {
+				time.Sleep(time.Second * 2)
+				continue
+			}
+			return nil, err
+		}
+	}
+
+}

core/statedb.go

+package core
+
+import (
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/trie"
+	"github.com/gogo/protobuf/proto"
+	witnessv1 "github.com/odysseus/odysseus-protocol/gen/proto/go/witness/v1"
+	log "github.com/sirupsen/logrus"
+)
+
+type StateDB struct {
+	db   *trie.Database
+	trie *trie.Trie
+}
+
+func NewStateDB(db *trie.Database, root common.Hash) (statedb *StateDB, err error) {
+	t, err := trie.New(common.Hash{}, root, db)
+	if err != nil {
+		return nil, err
+	}
+	return &StateDB{
+		db:   db,
+		trie: t,
+	}, nil
+}
+
+func (s *StateDB) GetMinerObject(miner common.Address) (object *witnessv1.MinerObject) {
+	object = &witnessv1.MinerObject{}
+	k := crypto.Keccak256(miner.Bytes())
+	v, err := s.trie.TryGet(k)
+	if err != nil {
+		return nil
+	}
+	err = proto.Unmarshal(v, object)
+	if err != nil {
+		return nil
+	}
+	return
+}
+
+func (s *StateDB) UpdateMinerObject(miner common.Address, object *witnessv1.MinerObject) (err error) {
+	k := crypto.Keccak256(miner.Bytes())
+	v, err := proto.Marshal(object)
+	if err != nil {
+		return
+	}
+
+	err = s.trie.TryUpdate(k, v)
+	return
+}
+
+func (s *StateDB) Commit() (root common.Hash, err error) {
+	root, nodes, err := s.trie.Commit(false)
+	if err != nil {
+		return
+	}
+	nodesSet := trie.NewMergedNodeSet()
+	if nodes != nil {
+		err = nodesSet.Merge(nodes)
+		if err != nil {
+			return
+		}
+	}
+	err = s.db.Update(nodesSet)
+	if err != nil {
+		return
+	}
+
+	err = s.db.Commit(root, true, nil)
+	return root, err
+}
+
+func (s *StateDB) IterAllObject() (objects []*witnessv1.MinerObject) {
+	iter := s.trie.NodeIterator(nil)
+	for iter.Next(true) {
+		if !iter.Leaf() {
+			continue
+		}
+		v := iter.LeafBlob()
+		object := &witnessv1.MinerObject{}
+		err := proto.Unmarshal(v, object)
+		if err != nil {
+			continue
+		}
+		log.WithField("workload", object.Workload).Debug("miner object")
+		objects = append(objects, object)
+	}
+	return
+}

core/trie.go

-package core

go.mod

@@ -3,24 +3,157 @@ module witness
 go 1.19
 
 require (
+	github.com/btcsuite/btcd/btcec/v2 v2.3.2
+	github.com/ethereum/go-ethereum v1.10.26
+	github.com/gogo/protobuf v1.3.2
+	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
+	github.com/libp2p/go-libp2p v0.26.0
+	github.com/libp2p/go-libp2p-pubsub v0.9.1
+	github.com/multiformats/go-multiaddr v0.12.1
+	github.com/odysseus/odysseus-protocol v0.0.0-00010101000000-000000000000
+	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.18.0
+	github.com/prysmaticlabs/prysm/v3 v3.2.2-rc.0
 	github.com/sirupsen/logrus v1.9.3
 	github.com/urfave/cli/v2 v2.27.1
+	google.golang.org/grpc v1.60.1
 )
 
 require (
 	github.com/BurntSushi/toml v1.3.2 // indirect
+	github.com/VictoriaMetrics/fastcache v1.10.0 // indirect
+	github.com/aristanetworks/goarista v0.0.0-20200805130819-fd197cf57d96 // indirect
+	github.com/benbjohnson/clock v1.3.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/containerd/cgroups v1.1.0 // indirect
+	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.3 // indirect
-	github.com/kr/text v0.2.0 // indirect
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect
+	github.com/deckarep/golang-set v1.8.0 // indirect
+	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 // indirect
+	github.com/docker/go-units v0.5.0 // indirect
+	github.com/edsrzf/mmap-go v1.1.0 // indirect
+	github.com/elastic/gosigar v0.14.2 // indirect
+	github.com/fjl/memsize v0.0.0-20190710130421-bcb5799ab5e5 // indirect
+	github.com/flynn/noise v1.0.0 // indirect
+	github.com/francoispqt/gojay v1.2.13 // indirect
+	github.com/gballet/go-libpcsclite v0.0.0-20191108122812-4678299bea08 // indirect
+	github.com/go-ole/go-ole v1.2.6 // indirect
+	github.com/go-stack/stack v1.8.1 // indirect
+	github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 // indirect
+	github.com/godbus/dbus/v5 v5.1.0 // indirect
+	github.com/golang-jwt/jwt/v4 v4.5.0 // indirect
+	github.com/golang/mock v1.6.0 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect
+	github.com/google/gopacket v1.1.19 // indirect
+	github.com/google/pprof v0.0.0-20231023181126-ff6d637d2a7b // indirect
+	github.com/google/uuid v1.3.1 // indirect
+	github.com/gorilla/websocket v1.5.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.0.1 // indirect
+	github.com/hashicorp/go-bexpr v0.1.10 // indirect
+	github.com/hashicorp/golang-lru v0.5.5-0.20210104140557-80c98217689d // indirect
+	github.com/hashicorp/golang-lru/v2 v2.0.5 // indirect
+	github.com/holiman/bloomfilter/v2 v2.0.3 // indirect
+	github.com/holiman/uint256 v1.2.4 // indirect
+	github.com/huin/goupnp v1.3.0 // indirect
+	github.com/ipfs/go-cid v0.4.1 // indirect
+	github.com/ipfs/go-log/v2 v2.5.1 // indirect
+	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
+	github.com/jbenet/go-temp-err-catcher v0.1.0 // indirect
+	github.com/klauspost/compress v1.17.2 // indirect
+	github.com/klauspost/cpuid/v2 v2.2.5 // indirect
+	github.com/koron/go-ssdp v0.0.4 // indirect
+	github.com/libp2p/go-buffer-pool v0.1.0 // indirect
+	github.com/libp2p/go-cidranger v1.1.0 // indirect
+	github.com/libp2p/go-flow-metrics v0.1.0 // indirect
+	github.com/libp2p/go-libp2p-asn-util v0.3.0 // indirect
+	github.com/libp2p/go-msgio v0.3.0 // indirect
+	github.com/libp2p/go-nat v0.2.0 // indirect
+	github.com/libp2p/go-netroute v0.2.1 // indirect
+	github.com/libp2p/go-reuseport v0.4.0 // indirect
+	github.com/libp2p/go-yamux/v4 v4.0.1 // indirect
+	github.com/marten-seemann/tcp v0.0.0-20210406111302-dfbc87cc63fd // indirect
+	github.com/mattn/go-colorable v0.1.8 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/mattn/go-runewidth v0.0.13 // indirect
 	github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
+	github.com/miekg/dns v1.1.56 // indirect
+	github.com/mikioh/tcpinfo v0.0.0-20190314235526-30a79bb1804b // indirect
+	github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc // indirect
+	github.com/minio/highwayhash v1.0.1 // indirect
+	github.com/minio/sha256-simd v1.0.1 // indirect
+	github.com/mitchellh/mapstructure v1.4.1 // indirect
+	github.com/mitchellh/pointerstructure v1.2.0 // indirect
+	github.com/mr-tron/base58 v1.2.0 // indirect
+	github.com/multiformats/go-base32 v0.1.0 // indirect
+	github.com/multiformats/go-base36 v0.2.0 // indirect
+	github.com/multiformats/go-multiaddr-dns v0.3.1 // indirect
+	github.com/multiformats/go-multiaddr-fmt v0.1.0 // indirect
+	github.com/multiformats/go-multibase v0.2.0 // indirect
+	github.com/multiformats/go-multicodec v0.9.0 // indirect
+	github.com/multiformats/go-multihash v0.2.3 // indirect
+	github.com/multiformats/go-multistream v0.5.0 // indirect
+	github.com/multiformats/go-varint v0.0.7 // indirect
+	github.com/olekukonko/tablewriter v0.0.5 // indirect
+	github.com/onsi/ginkgo/v2 v2.13.0 // indirect
+	github.com/opencontainers/runtime-spec v1.1.0 // indirect
+	github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 // indirect
 	github.com/prometheus/client_model v0.5.0 // indirect
 	github.com/prometheus/common v0.45.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
+	github.com/prometheus/tsdb v0.10.0 // indirect
+	github.com/prysmaticlabs/fastssz v0.0.0-20220628121656-93dfe28febab // indirect
+	github.com/prysmaticlabs/go-bitfield v0.0.0-20210809151128-385d8c5e3fb7 // indirect
+	github.com/prysmaticlabs/gohashtree v0.0.2-alpha // indirect
+	github.com/quic-go/qpack v0.4.0 // indirect
+	github.com/quic-go/qtls-go1-20 v0.3.4 // indirect
+	github.com/quic-go/quic-go v0.39.4 // indirect
+	github.com/quic-go/webtransport-go v0.6.0 // indirect
+	github.com/raulk/go-watchdog v1.3.0 // indirect
+	github.com/rivo/uniseg v0.3.4 // indirect
+	github.com/rjeczalik/notify v0.9.1 // indirect
+	github.com/rs/cors v1.7.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
+	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
+	github.com/spaolacci/murmur3 v1.1.0 // indirect
+	github.com/status-im/keycard-go v0.2.0 // indirect
+	github.com/stretchr/testify v1.8.2 // indirect
+	github.com/supranational/blst v0.3.11 // indirect
+	github.com/syndtr/goleveldb v1.0.1-0.20220721030215-126854af5e6d // indirect
+	github.com/thomaso-mirodin/intmath v0.0.0-20160323211736-5dc6d854e46e // indirect
+	github.com/tklauser/go-sysconf v0.3.12 // indirect
+	github.com/tklauser/numcpus v0.6.1 // indirect
+	github.com/tyler-smith/go-bip39 v1.1.0 // indirect
 	github.com/xrash/smetrics v0.0.0-20231213231151-1d8dd44e695e // indirect
+	github.com/yusufpapurcu/wmi v1.2.2 // indirect
+	go.uber.org/dig v1.17.1 // indirect
+	go.uber.org/fx v1.20.1 // indirect
+	go.uber.org/mock v0.3.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.26.0 // indirect
+	golang.org/x/crypto v0.18.0 // indirect
+	golang.org/x/exp v0.0.0-20240110193028-0dcbfd608b1e // indirect
+	golang.org/x/mod v0.14.0 // indirect
+	golang.org/x/net v0.20.0 // indirect
+	golang.org/x/sync v0.6.0 // indirect
 	golang.org/x/sys v0.16.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba // indirect
+	golang.org/x/tools v0.16.0 // indirect
+	google.golang.org/genproto v0.0.0-20240102182953-50ed04b92917 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20231212172506-995d672761c0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240108191215-35c7eff3a6b1 // indirect
 	google.golang.org/protobuf v1.32.0 // indirect
+	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
+	lukechampine.com/blake3 v1.2.1 // indirect
+	nhooyr.io/websocket v1.8.7 // indirect
 )
+
+replace github.com/odysseus/odysseus-protocol => ../odysseus-protocol
+
+replace github.com/MariusVanDerWijden/tx-fuzz => github.com/MariusVanDerWijden/tx-fuzz v1.3.2

p2p/bootnode_cache.go

+package p2p
+
+import (
+	"encoding/json"
+	"io"
+	"os"
+)
+
+const bootnodefile = ".bootnodes.json"
+
+var bootnodes []string
+
+func loadBootnodesFile() {
+	file, err := os.Open(bootnodefile)
+	if err != nil {
+		return
+	}
+	defer file.Close()
+
+	b, err := io.ReadAll(file)
+	if err != nil {
+		return
+	}
+	_ = json.Unmarshal(b, &bootnodes)
+	return
+}
+
+func attachBootnodeToFile(bootnode string) error {
+	for _, node := range bootnodes {
+		if node == bootnode {
+			return nil
+		}
+	}
+	bootnodes = append(bootnodes, bootnode)
+	b, err := json.Marshal(bootnodes)
+	if err != nil {
+		return err
+	}
+	file, err := os.OpenFile(bootnodefile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+	_, err = file.Write(b)
+	return err
+
+}

p2p/broadcaster.go

+package p2p
+
+import (
+	"context"
+
+	"github.com/gogo/protobuf/proto"
+	"github.com/pkg/errors"
+)
+
+// Broadcast a message to the p2p network, the message is assumed to be
+// broadcasted to the current fork.
+func (s *Service) Broadcast(ctx context.Context, msg proto.Message, topic string) error {
+	msgBytes, err := proto.Marshal(msg)
+	if err != nil {
+		return err
+	}
+	return s.broadcastObject(ctx, msgBytes, topic)
+}
+
+// method to broadcast messages to other peers in our gossip mesh.
+func (s *Service) broadcastObject(ctx context.Context, obj []byte, topic string) error {
+	if err := s.PublishToTopic(ctx, topic, obj); err != nil {
+		err := errors.Wrap(err, "could not publish message")
+		return err
+	}
+	return nil
+}

p2p/config.go

+package p2p
+
+// Config for the p2p service. These parameters are set from application level flags
+// to initialize the p2p service.
+type Config struct {
+	UseDiscovery        bool
+	StaticPeers         []string
+	BootstrapNodeAddr   []string
+	Discv5BootStrapAddr []string
+	LocalIP             string
+	HostAddress         string
+	PrivateKeyPath      string
+	TCPPort             uint
+	UDPPort             uint
+	MaxPeers            uint
+	MaxInboundPeers     uint
+	MaxOutboundPeers    uint
+	IgnoreLocalIP       bool
+	SingleNode          bool
+	EnableProm          bool
+	NebulaGrpcAddress   string
+}

p2p/connection_gater.go

+package p2p
+
+import (
+	"runtime"
+
+	"github.com/libp2p/go-libp2p/core/control"
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/multiformats/go-multiaddr"
+	manet "github.com/multiformats/go-multiaddr/net"
+	"github.com/sirupsen/logrus"
+)
+
+// InterceptPeerDial tests whether we're permitted to Dial the specified peer.
+func (_ *Service) InterceptPeerDial(_ peer.ID) (allow bool) {
+	return true
+}
+
+// InterceptAddrDial tests whether we're permitted to dial the specified
+// multiaddr for the given peer.
+func (s *Service) InterceptAddrDial(pid peer.ID, m multiaddr.Multiaddr) (allow bool) {
+	// Disallow bad peers from dialing in.
+	return !s.peers.IsBad(pid)
+}
+
+// InterceptAccept checks whether the incidental inbound connection is allowed.
+func (s *Service) InterceptAccept(n network.ConnMultiaddrs) (allow bool) {
+	// Deny all incoming connections before we are ready
+	if !s.started {
+		return false
+	}
+	if !s.validateDial(n.RemoteMultiaddr()) {
+		// Allow other go-routines to run in the event
+		// we receive a large amount of junk connections.
+		runtime.Gosched()
+		log.WithFields(logrus.Fields{"peer": n.RemoteMultiaddr(),
+			"reason": "exceeded dial limit"}).Trace("Not accepting inbound dial from ip address")
+		return false
+	}
+	if s.isPeerAtLimit(true /* inbound */) {
+		log.WithFields(logrus.Fields{"peer": n.RemoteMultiaddr(),
+			"reason": "at peer limit"}).Trace("Not accepting inbound dial")
+		return false
+	}
+	return true
+}
+
+// InterceptSecured tests whether a given connection, now authenticated,
+// is allowed.
+func (_ *Service) InterceptSecured(_ network.Direction, _ peer.ID, _ network.ConnMultiaddrs) (allow bool) {
+	return true
+}
+
+// InterceptUpgraded tests whether a fully capable connection is allowed.
+func (_ *Service) InterceptUpgraded(_ network.Conn) (allow bool, reason control.DisconnectReason) {
+	return true, 0
+}
+
+func (s *Service) validateDial(addr multiaddr.Multiaddr) bool {
+	_, err := manet.ToIP(addr)
+	return err == nil
+}

p2p/discovery_test.go

+package p2p
+
+import (
+	"testing"
+
+	"github.com/ethereum/go-ethereum/p2p/enode"
+)
+
+func TestDecodeV5(t *testing.T) {
+	nodes := []string{
+		"enr:-JG4QL9QdmQjI-AOv-zFL1oaUAcWP6dPXBBOUAEXE_z4KZNfNTYDfmtAvxq-E0Xl1cqWGO_WmPldHoZIaVKzA0moWQ-GAYupV_MKgmlkgnY0gmlwhMCoAdOJc2VjcDI1NmsxoQLvx1jTRx75jpj8jD3KJ3FCZ_6UiICRIQJ_1MWmVx0IbYN0Y3CCdn6DdWRwgnZ-",
+		"enr:-JG4QPGTzcn6tp8UqoSmEBBj5f0_42wonWmoedFgvncdZL3Qf0DCEQVzDSHGi9jM35qdD76rvpWFuGpQ_rmoX14ZnACGAYupV_plgmlkgnY0gmlwhMCoAdOJc2VjcDI1NmsxoQLvx1jTRx75jpj8jD3KJ3FCZ_6UiICRIQJ_1MWmVx0IbYN0Y3CCdn6DdWRwgnZ-",
+		"enr:-JG4QP1aYRCaCknvIa3hSqE1zAUrm203DNFcvS8E1lQkqHq-PR0sQhIkzK7L4ZLLLPOijD9BTeCLWi8-0LVORt63dimGAYutv-WkgmlkgnY0gmlwhMCoAdyJc2VjcDI1NmsxoQJ81pyuyqa2QoPztOPsCqpkcW1gJwUxOcSJuWr9VUNavIN0Y3CCdn2DdWRwgnZ-",
+	}
+	r1, r2, r3 := parseGenericAddrs(nodes)
+	t.Log(r1)
+	t.Log(r2)
+	t.Log(r3)
+
+	for _, addr := range r1 {
+		bootnode, err := enode.Parse(enode.ValidSchemes, addr)
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Log(bootnode.String())
+		maddr, _ := convertToSingleMultiAddr(bootnode)
+		t.Log(maddr.String())
+	}
+
+}

p2p/handshake.go

+package p2p
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"witness/p2p/peers"
+
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/sirupsen/logrus"
+)
+
+const (
+	// The time to wait for a status request.
+	timeForStatus = 10 * time.Second
+)
+
+func peerMultiaddrString(conn network.Conn) string {
+	return fmt.Sprintf("%s/p2p/%s", conn.RemoteMultiaddr().String(), conn.RemotePeer().String())
+}
+
+// AddConnectionHandler adds a callback function which handles the connection with a
+// newly added peer. It performs a handshake with that peer by sending a hello request
+// and validating the response from the peer.
+func (s *Service) AddConnectionHandler(reqFunc, goodByeFunc func(ctx context.Context, id peer.ID) error) {
+	// Peer map and lock to keep track of current connection attempts.
+	peerMap := make(map[peer.ID]bool)
+	peerLock := new(sync.Mutex)
+
+	// This is run at the start of each connection attempt, to ensure
+	// that there aren't multiple inflight connection requests for the
+	// same peer at once.
+	peerHandshaking := func(id peer.ID) bool {
+		peerLock.Lock()
+		defer peerLock.Unlock()
+
+		if peerMap[id] {
+			return true
+		}
+
+		peerMap[id] = true
+		return false
+	}
+
+	peerFinished := func(id peer.ID) {
+		peerLock.Lock()
+		defer peerLock.Unlock()
+
+		delete(peerMap, id)
+	}
+
+	s.host.Network().Notify(&network.NotifyBundle{
+		ConnectedF: func(net network.Network, conn network.Conn) {
+			remotePeer := conn.RemotePeer()
+			disconnectFromPeer := func() {
+				s.peers.SetConnectionState(remotePeer, peers.PeerDisconnecting)
+				// Only attempt a goodbye if we are still connected to the peer.
+				if s.host.Network().Connectedness(remotePeer) == network.Connected {
+					if err := goodByeFunc(context.TODO(), remotePeer); err != nil {
+						log.WithError(err).Error("Unable to disconnect from peer")
+					}
+				}
+				s.peers.SetConnectionState(remotePeer, peers.PeerDisconnected)
+			}
+			// Connection handler must be non-blocking as part of libp2p design.
+			go func() {
+				if peerHandshaking(remotePeer) {
+					// Exit this if there is already another connection
+					// request in flight.
+					return
+				}
+				defer peerFinished(remotePeer)
+				// Handle the various pre-existing conditions that will result in us not handshaking.
+				peerConnectionState, err := s.peers.ConnectionState(remotePeer)
+				if err == nil && (peerConnectionState == peers.PeerConnected || peerConnectionState == peers.PeerConnecting) {
+					log.WithField("currentState", peerConnectionState).WithField("reason", "already active").Trace("Ignoring connection request")
+					return
+				}
+				s.peers.Add(nil /* ENR */, remotePeer, conn.RemoteMultiaddr(), conn.Stat().Direction)
+				// Defensive check in the event we still get a bad peer.
+				if s.peers.IsBad(remotePeer) {
+					log.WithField("reason", "bad peer").Trace("Ignoring connection request")
+					disconnectFromPeer()
+					return
+				}
+				validPeerConnection := func() {
+					s.peers.SetConnectionState(conn.RemotePeer(), peers.PeerConnected)
+					// Go through the handshake process.
+					log.WithFields(logrus.Fields{
+						"direction":   conn.Stat().Direction,
+						"multiAddr":   peerMultiaddrString(conn),
+						"activePeers": len(s.peers.Active()),
+					}).Info("Peer connected")
+				}
+
+				// Do not perform handshake on inbound dials.
+				if conn.Stat().Direction == network.DirInbound {
+					// Wait for peer to initiate handshake
+					time.Sleep(timeForStatus)
+
+					// Exit if we are disconnected with the peer.
+					if s.host.Network().Connectedness(remotePeer) != network.Connected {
+						return
+					}
+
+					validPeerConnection()
+					return
+				}
+
+				s.peers.SetConnectionState(conn.RemotePeer(), peers.PeerConnecting)
+				if err := reqFunc(context.TODO(), conn.RemotePeer()); err != nil && err != io.EOF {
+					log.WithError(err).Trace("Handshake failed")
+					disconnectFromPeer()
+					return
+				}
+				validPeerConnection()
+			}()
+		},
+	})
+}
+
+// AddDisconnectionHandler disconnects from peers.  It handles updating the peer status.
+// This also calls the handler responsible for maintaining other parts of the sync or p2p system.
+func (s *Service) AddDisconnectionHandler(handler func(ctx context.Context, id peer.ID) error) {
+	s.host.Network().Notify(&network.NotifyBundle{
+		DisconnectedF: func(net network.Network, conn network.Conn) {
+			log := log.WithField("multiAddr", peerMultiaddrString(conn))
+			// Must be handled in a goroutine as this callback cannot be blocking.
+			go func() {
+				// Exit early if we are still connected to the peer.
+				if net.Connectedness(conn.RemotePeer()) == network.Connected {
+					return
+				}
+				priorState, err := s.peers.ConnectionState(conn.RemotePeer())
+				if err != nil {
+					// Can happen if the peer has already disconnected, so...
+					priorState = peers.PeerDisconnected
+				}
+				s.peers.SetConnectionState(conn.RemotePeer(), peers.PeerDisconnecting)
+				if err := handler(context.TODO(), conn.RemotePeer()); err != nil {
+					log.WithError(err).Error("Disconnect handler failed")
+				}
+				s.peers.SetConnectionState(conn.RemotePeer(), peers.PeerDisconnected)
+				// Only log disconnections if we were fully connected.
+				if priorState == peers.PeerConnected {
+					log.WithField("activePeers", len(s.peers.Active())).Info("Peer disconnected")
+				}
+			}()
+		},
+	})
+}

p2p/log.go

+package p2p
+
+import (
+	"strings"
+
+	"github.com/libp2p/go-libp2p/core/peer"
+	ma "github.com/multiformats/go-multiaddr"
+	"github.com/sirupsen/logrus"
+)
+
+var log = logrus.WithField("module", "p2p")
+
+func logIPAddr(id peer.ID, addrs ...ma.Multiaddr) {
+	var correctAddr ma.Multiaddr
+	for _, addr := range addrs {
+		if strings.Contains(addr.String(), "/ip4/") || strings.Contains(addr.String(), "/ip6/") {
+			correctAddr = addr
+			break
+		}
+	}
+	if correctAddr != nil {
+		log.WithField(
+			"multiAddr",
+			correctAddr.String()+"/p2p/"+id.String(),
+		).Info("Node started p2p server")
+	}
+}
+
+func logExternalIPAddr(id peer.ID, addr string, port uint) {
+	if addr != "" {
+		multiAddr, err := MultiAddressBuilder(addr, port)
+		if err != nil {
+			log.WithError(err).Error("Could not create multiaddress")
+			return
+		}
+		log.WithField(
+			"multiAddr",
+			multiAddr.String()+"/p2p/"+id.String(),
+		).Info("Node started external p2p server")
+	}
+}

p2p/message_id.go

+package p2p
+
+import (
+	"fmt"
+
+	pubsubpb "github.com/libp2p/go-libp2p-pubsub/pb"
+	"github.com/prysmaticlabs/prysm/v3/crypto/hash"
+)
+
+func msgID(pmsg *pubsubpb.Message) string {
+	h := hash.Hash(pmsg.Data)
+	return fmt.Sprintf("%x", h[:20])
+}

p2p/message_id_test.go

+package p2p
+
+import (
+	"testing"
+
+	pubsubpb "github.com/libp2p/go-libp2p-pubsub/pb"
+)
+
+func TestMessageID(t *testing.T) {
+	tt := []pubsubpb.Message{
+		{
+			Data: []byte("hello"),
+		},
+		{
+			Data: []byte("world"),
+		},
+	}
+	wants := []string{
+		"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c",
+		"486ea46224d1bb4fb680f34f7c9ad96a8f24ec88",
+	}
+
+	for i := range tt {
+		t.Run(string(tt[i].Data), func(t *testing.T) {
+			if msgID(&tt[i]) != wants[i] {
+				t.Errorf("Wanted %s, got %s", wants[i], msgID(&tt[i]))
+			}
+		})
+	}
+}

p2p/options.go

+package p2p
+
+import (
+	"crypto/ecdsa"
+	"fmt"
+	"net"
+	"witness/version"
+
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/p2p/muxer/yamux"
+	"github.com/libp2p/go-libp2p/p2p/security/noise"
+	"github.com/libp2p/go-libp2p/p2p/transport/tcp"
+	ma "github.com/multiformats/go-multiaddr"
+	"github.com/pkg/errors"
+	ecdsaprysm "github.com/prysmaticlabs/prysm/v3/crypto/ecdsa"
+)
+
+// MultiAddressBuilder takes in an ip address string and port to produce a go multiaddr format.
+func MultiAddressBuilder(ipAddr string, port uint) (ma.Multiaddr, error) {
+	parsedIP := net.ParseIP(ipAddr)
+	if parsedIP.To4() == nil && parsedIP.To16() == nil {
+		return nil, errors.Errorf("invalid ip address provided: %s", ipAddr)
+	}
+	if parsedIP.To4() != nil {
+		return ma.NewMultiaddr(fmt.Sprintf("/ip4/%s/tcp/%d", ipAddr, port))
+	}
+	return ma.NewMultiaddr(fmt.Sprintf("/ip6/%s/tcp/%d", ipAddr, port))
+}
+
+// buildOptions for the libp2p host.
+func (s *Service) buildOptions(ip net.IP, priKey *ecdsa.PrivateKey) []libp2p.Option {
+	cfg := s.cfg
+	listen, err := MultiAddressBuilder(ip.String(), cfg.TCPPort)
+	if err != nil {
+		log.WithError(err).Fatal("Failed to p2p listen")
+	}
+	if cfg.LocalIP != "" {
+		if net.ParseIP(cfg.LocalIP) == nil {
+			log.Fatalf("Invalid local ip provided: %s", cfg.LocalIP)
+		}
+		listen, err = MultiAddressBuilder(cfg.LocalIP, cfg.TCPPort)
+		if err != nil {
+			log.WithError(err).Fatal("Failed to p2p listen")
+		}
+	}
+	ifaceKey, err := ecdsaprysm.ConvertToInterfacePrivkey(priKey)
+	if err != nil {
+		log.WithError(err).Fatal("Failed to retrieve private key")
+	}
+	id, err := peer.IDFromPublicKey(ifaceKey.GetPublic())
+	if err != nil {
+		log.WithError(err).Fatal("Failed to retrieve peer id")
+	}
+	log.Infof("Running node with peer id of %s ", id.String())
+	myPeerIDToFile(id.String())
+
+	log.Debugf("listen %s", listen.String())
+
+	options := []libp2p.Option{
+		privKeyOption(priKey),
+		libp2p.ListenAddrs(listen),
+		libp2p.UserAgent(version.Version),
+		libp2p.ConnectionGater(s),
+		libp2p.Transport(tcp.NewTCPTransport),
+		libp2p.Muxer("/yamux/6.7.0", yamux.DefaultTransport),
+		libp2p.DefaultMuxers,
+	}
+
+	options = append(options, libp2p.Security(noise.ID, noise.New))
+	options = append(options, libp2p.DisableRelay())
+
+	if cfg.HostAddress != "" {
+		options = append(options, libp2p.AddrsFactory(func(addrs []ma.Multiaddr) []ma.Multiaddr {
+			external, err := MultiAddressBuilder(cfg.HostAddress, cfg.TCPPort)
+			if err != nil {
+				log.WithError(err).Error("Unable to create external multiaddress")
+			} else {
+				addrs = append(addrs, external)
+			}
+			return addrs
+		}))
+	}
+
+	return options
+}
+
+func multiAddressBuilderWithID(ipAddr, protocol string, port uint, id peer.ID) (ma.Multiaddr, error) {
+	parsedIP := net.ParseIP(ipAddr)
+	if parsedIP.To4() == nil && parsedIP.To16() == nil {
+		return nil, errors.Errorf("invalid ip address provided: %s", ipAddr)
+	}
+	if id.String() == "" {
+		return nil, errors.New("empty peer id given")
+	}
+	if parsedIP.To4() != nil {
+		return ma.NewMultiaddr(fmt.Sprintf("/ip4/%s/%s/%d/p2p/%s", ipAddr, protocol, port, id.String()))
+	}
+	return ma.NewMultiaddr(fmt.Sprintf("/ip6/%s/%s/%d/p2p/%s", ipAddr, protocol, port, id.String()))
+}
+
+// Adds a private key to the libp2p option if the option was provided.
+// If the private key file is missing or cannot be read, or if the
+// private key contents cannot be marshaled, an exception is thrown.
+func privKeyOption(privkey *ecdsa.PrivateKey) libp2p.Option {
+	return func(cfg *libp2p.Config) error {
+		ifaceKey, err := ecdsaprysm.ConvertToInterfacePrivkey(privkey)
+		if err != nil {
+			return err
+		}
+		log.Debug("ECDSA private key generated")
+		return cfg.Apply(libp2p.Identity(ifaceKey))
+	}
+}