Merge pull request #33 from probe-lab/raulk/fixes

implement Status and Metadata handlers + close write side of streams on request + misc fixes

Author: raulk

api/client.go

@@ -17,6 +17,7 @@ type ClientConfig struct {
 	Endpoint     string
 	StateTimeout time.Duration
 	QueryTimeout time.Duration
+	Logger       log.FieldLogger
 }
 
 type Client struct {
@@ -85,21 +86,36 @@ func (c *Client) get(
 	// we will only handle JSONs
 	req.Header.Set("Accept", "application/json")
 
+	l := c.cfg.Logger.WithFields(log.Fields{
+		"url":    callURL,
+		"method": req.Method,
+	})
+	l.Info("requesting beacon API")
 	resp, err := c.client.Do(req)
+
 	if err != nil {
+		l.WithError(err).Warn("error requesting beacon API")
 		return respBody, errors.Wrap(err, fmt.Sprintf("unable to request URL %s", callURL.String()))
 	}
 	if resp == nil {
-		return respBody, errors.New("got empty response from the API")
+		err := errors.New("got empty response from the API")
+		l.WithError(err).Warn("error requesting beacon API")
+		return respBody, err
 	}
 	defer resp.Body.Close()
 
 	respBody, err = io.ReadAll(resp.Body)
 	if err != nil {
+		l.WithError(err).Warn("failed to read response body")
 		return respBody, errors.Wrap(err, "reading response body")
 	}
 
-	// return copy of the request
+	if len(respBody) > 1024 {
+		l.Infof("successful beacon API response: [omitted due to length %d]", len(respBody))
+	} else {
+		l.Infof("successful beacon API response: %s", respBody)
+	}
+
 	return respBody, nil
 }
 

api/client_test.go

@@ -36,7 +36,7 @@ func Test_ApiGetPeerDASstate(t *testing.T) {
 	httpCli, testMainCtx, cancel := genTestAPICli(t)
 	defer cancel()
 
-	_, err := httpCli.GetPeerDASstate(testMainCtx)
+	_, err := httpCli.GetBeaconStateHead(testMainCtx)
 	require.NoError(t, err)
 }
 

api/fork_choice.go

@@ -31,8 +31,8 @@ type ForkChoiceNode struct {
 	ExecutionBlockHash string `json:"execution_block_hash"`
 }
 
-func (c *Client) GetForkChoice(ctx context.Context) (ForkChoiceNode, error) {
-	var forkChoice ForkChoiceNode
+func (c *Client) GetForkChoice(ctx context.Context) (ForkChoice, error) {
+	var forkChoice ForkChoice
 
 	resp, err := c.get(ctx, c.cfg.QueryTimeout, ForkChoiceBase, "")
 	if err != nil {

api/node_identity.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"strconv"
 
+	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/pkg/errors"
 )
 
@@ -17,23 +18,33 @@ type NodeIdentity struct {
 		Maddrs     []string `json:"p2p_addresses"`
 		DiscvAddrs []string `json:"discovery_addresses"`
 		Metadata   struct {
-			SeqNum   string `json:"seq_number"`
-			Attnets  string `json:"attnets"`
-			Syncnets string `json:"syncnets"`
-			Cgc      string `json:"custody_group_count"`
+			SeqNum   string        `json:"seq_number"`
+			Attnets  hexutil.Bytes `json:"attnets"`
+			Syncnets hexutil.Bytes `json:"syncnets"`
+			Cgc      string        `json:"custody_group_count"`
 		} `json:"metadata"`
 	} `json:"data"`
 }
 
 func (i *NodeIdentity) CustodyInt() (int, error) {
+	// TODO remove patch for Prysm, adding dummy data in missing fields.
+	// https://github.com/OffchainLabs/prysm/pull/15506
+	if i.Data.Metadata.Cgc == "" {
+		return 128, nil
+	}
 	return strconv.Atoi(i.Data.Metadata.Cgc)
 }
 
-func (i *NodeIdentity) Attnets() string {
+func (i *NodeIdentity) Attnets() []byte {
 	return i.Data.Metadata.Attnets
 }
 
-func (i *NodeIdentity) Syncnets() string {
+func (i *NodeIdentity) Syncnets() []byte {
+	// TODO remove patch for Prysm, adding dummy data in missing fields.
+	// https://github.com/OffchainLabs/prysm/pull/15506
+	if i.Data.Metadata.Syncnets == nil {
+		return []byte{0x00}
+	}
 	return i.Data.Metadata.Syncnets
 }
 

api/state.go

@@ -10,15 +10,15 @@ import (
 
 var BeaconStateBase = "eth/v2/debug/beacon/states/head"
 
-type PeerDASstate struct {
+type BeaconState struct {
 	Version             string              `json:"version"`
 	ExecutionOptimistic bool                `json:"execution_optimistic"`
 	Finalized           bool                `json:"finalized"`
 	Data                electra.BeaconState `json:"data"`
 }
 
-func (c *Client) GetPeerDASstate(ctx context.Context) (PeerDASstate, error) {
-	var state PeerDASstate
+func (c *Client) GetBeaconStateHead(ctx context.Context) (BeaconState, error) {
+	var state BeaconState
 
 	resp, err := c.get(ctx, c.cfg.QueryTimeout, BeaconStateBase, "")
 	if err != nil {

beacon_api.go

@@ -41,11 +41,11 @@ type BeaconAPIConfig struct {
 }
 
 type BeaconAPIImpl struct {
-	cfg    BeaconAPIConfig
-	apiCli *api.Client
+	cfg       BeaconAPIConfig
+	apiClient *api.Client
 
 	specs         map[string]any
-	headState     *api.PeerDASstate
+	headState     *api.BeaconState
 	forkSchedules api.ForkSchedule
 	fuluForkEpoch uint64
 }
@@ -55,27 +55,28 @@ func NewBeaconAPI(cfg BeaconAPIConfig) (BeaconAPI, error) {
 		Endpoint:     cfg.Endpoint,
 		StateTimeout: ApiStateTimeout,
 		QueryTimeout: ApiQueryTimeout,
+		Logger:       cfg.Logger,
 	}
 	apiCli, err := api.NewClient(ethApiCfg)
 	if err != nil {
 		return nil, err
 	}
 
 	return &BeaconAPIImpl{
-		cfg:    cfg,
-		apiCli: apiCli,
+		cfg:       cfg,
+		apiClient: apiCli,
 	}, nil
 }
 
 func (b *BeaconAPIImpl) Init(ctx context.Context) error {
 	// check api connection
-	if err := b.apiCli.CheckConnection(ctx); err != nil {
+	if err := b.apiClient.CheckConnection(ctx); err != nil {
 		return fmt.Errorf("connection to %s was stablished, but not active - %s", b.cfg.Endpoint, err.Error())
 	}
 	b.cfg.Logger.Info("connected to the beacon API...")
 
 	// Get node identity and ENR
-	nodeIdentity, err := b.apiCli.GetNodeIdentity(ctx)
+	nodeIdentity, err := b.apiClient.GetNodeIdentity(ctx)
 	if err != nil {
 		b.cfg.Logger.WithError(err).Warn("failed to get node identity")
 	} else {
@@ -92,14 +93,14 @@ func (b *BeaconAPIImpl) Init(ctx context.Context) error {
 	}
 
 	// get the config specs from the apiCli
-	specs, err := b.apiCli.GetConfigSpecs(ctx)
+	specs, err := b.apiClient.GetConfigSpecs(ctx)
 	if err != nil {
 		return err
 	}
 	b.specs = specs
 
 	// Get genesis data from the proper endpoint
-	genesisData, err := b.apiCli.GetGenesis(ctx)
+	genesisData, err := b.apiClient.GetGenesis(ctx)
 	if err != nil {
 		return errors.Wrap(err, "failed to get genesis data")
 	}
@@ -133,7 +134,7 @@ func (b *BeaconAPIImpl) Init(ctx context.Context) error {
 	}
 
 	// get the network configuration from the apiCli
-	forkSchedules, err := b.apiCli.GetNetworkConfig(ctx)
+	forkSchedules, err := b.apiClient.GetNetworkConfig(ctx)
 	if err != nil {
 		return err
 	}
@@ -143,7 +144,7 @@ func (b *BeaconAPIImpl) Init(ctx context.Context) error {
 	}
 
 	// compose and get the local Metadata
-	currentState, err := b.apiCli.GetPeerDASstate(ctx)
+	currentState, err := b.apiClient.GetBeaconStateHead(ctx)
 	if err != nil {
 		return err
 	}
@@ -190,7 +191,7 @@ func (b *BeaconAPIImpl) Init(ctx context.Context) error {
 				return fmt.Errorf("tooled closed without reaching fulu upgrade")
 
 			case <-time.After(secondsToFulu):
-				currentState, err = b.apiCli.GetPeerDASstate(ctx)
+				currentState, err = b.apiClient.GetBeaconStateHead(ctx)
 				if err != nil {
 					return err
 				}
@@ -412,9 +413,9 @@ func (b *BeaconAPIImpl) GetFuluForkEpoch() uint64 {
 }
 
 func (b *BeaconAPIImpl) GetNodeIdentity(ctx context.Context) (*api.NodeIdentity, error) {
-	return b.apiCli.GetNodeIdentity(ctx)
+	return b.apiClient.GetNodeIdentity(ctx)
 }
 
 func (b *BeaconAPIImpl) GetBeaconBlock(ctx context.Context, slot uint64) (*spec.VersionedSignedBeaconBlock, error) {
-	return b.apiCli.GetBeaconBlock(ctx, slot)
+	return b.apiClient.GetBeaconBlock(ctx, slot)
 }

custody_utils.go

@@ -2,13 +2,13 @@ package dasguardian
 
 import (
 	"encoding/binary"
-	"encoding/hex"
 	"math"
 	"sort"
 
 	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/ethereum/go-ethereum/p2p/enr"
 	"github.com/holiman/uint256"
-	errors "github.com/pkg/errors"
+	"github.com/pkg/errors"
 )
 
 var (
@@ -39,31 +39,28 @@ type AttnetsEntry []byte
 
 func (c AttnetsEntry) ENRKey() string { return "attnets" }
 
-func GetAttnetsFromEnr(ethNode *enode.Node) string {
+func GetAttnetsFromEnr(ethNode *enode.Node) (AttnetsEntry, error) {
 	enr := ethNode.Record()
 
 	var attnetsEntry AttnetsEntry
 	err := enr.Load(&attnetsEntry)
-	if err != nil {
-		return "no-attnets"
-	}
-	return "0x" + hex.EncodeToString(attnetsEntry)
+	return attnetsEntry, err
 }
 
 // syncnets
 type SyncnetsEntry []byte
 
 func (c SyncnetsEntry) ENRKey() string { return "syncnets" }
 
-func GetSyncnetsFromEnr(ethNode *enode.Node) string {
-	enr := ethNode.Record()
+func GetSyncnetsFromEnr(ethNode *enode.Node) (SyncnetsEntry, error) {
+	record := ethNode.Record()
 
 	var syncnetsEntry SyncnetsEntry
-	err := enr.Load(&syncnetsEntry)
-	if err != nil {
-		return "no-syncnets"
+	err := record.Load(&syncnetsEntry)
+	if enr.IsNotFound(err) {
+		return nil, nil
 	}
-	return "0x" + hex.EncodeToString(syncnetsEntry)
+	return syncnetsEntry, err
 }
 
 // Mainly from: prysm/beacon-chain/core/peerdas/helpers.go

encoding.go

@@ -8,10 +8,11 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/golang/snappy"
 	"github.com/libp2p/go-libp2p/core/network"
 	dynssz "github.com/pk910/dynamic-ssz"
-	errors "github.com/pkg/errors"
+	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -127,12 +128,19 @@ func readVarint(r io.Reader) (uint64, error) {
 
 // writeRequest writes a request to the stream with SSZ+Snappy encoding
 func (r *ReqResp) writeRequest(stream network.Stream, req any) error {
+	defer stream.CloseWrite()
+
 	// Set write deadline
 	if err := stream.SetWriteDeadline(time.Now().Add(r.cfg.WriteTimeout)); err != nil {
 		return fmt.Errorf("failed to set write deadline: %w", err)
 	}
 
-	// Marshal to SSZ
+	if req == nil {
+		// we close the write side of the stream immediately, communicating we have no data to send
+		return nil
+	}
+
+	// Marshal to SSZ if the request is not nil
 	data, err := sszCodec.MarshalSSZ(req)
 	if err != nil {
 		return fmt.Errorf("failed to marshal SSZ: %w", err)
@@ -157,6 +165,14 @@ func (r *ReqResp) writeRequest(stream network.Stream, req any) error {
 		return fmt.Errorf("failed to compress data: %w", err)
 	}
 
+	log.WithFields(log.Fields{
+		"protocol":  stream.Protocol(),
+		"data":      hexutil.Encode(data),
+		"data_len":  len(data),
+		"wire_data": hexutil.Encode(buf.Bytes()),
+		"wire_len":  buf.Len(),
+	}).Debug("writing request")
+
 	// Write buffer to the stream
 	if _, err := io.Copy(stream, &buf); err != nil {
 		return fmt.Errorf("failed to write final payload to stream: %w", err)
@@ -271,7 +287,8 @@ func (r *ReqResp) readResponse(stream network.Stream, resp any) error {
 		}).Debug("Raw response code received")
 	}
 
-	if code[0] != ResponseCodeSuccess {
+	success := code[0] == ResponseCodeSuccess
+	if !success {
 		if log.GetLevel() >= log.DebugLevel {
 			errorType := getResponseCodeName(code[0])
 			log.WithFields(log.Fields{
@@ -280,7 +297,6 @@ func (r *ReqResp) readResponse(stream network.Stream, resp any) error {
 				"error_type":    errorType,
 			}).Debug("Non-success response code received")
 		}
-		return fmt.Errorf("RPC error code: %d", code[0])
 	}
 
 	// Read uncompressed length prefix
@@ -330,6 +346,21 @@ func (r *ReqResp) readResponse(stream network.Stream, resp any) error {
 		}).Debug("Raw response data received")
 	}
 
+	if !success {
+		var errorMessage ErrorMessage
+		l := log.WithFields(log.Fields{
+			"response_type": responseType,
+			"raw_data_hex":  fmt.Sprintf("0x%x", data),
+		})
+		if err := sszCodec.UnmarshalSSZ(&errorMessage, data); err != nil {
+			l.WithError(err).Error("failed to unmarshal SSZ error message")
+			return fmt.Errorf("failed to unmarshal SSZ error message: %w", err)
+		}
+		msg := string(errorMessage)
+		l.Warnf("RPC failed; error message: %s", msg)
+		return fmt.Errorf("RPC failed: %s", msg)
+	}
+
 	// Unmarshal from SSZ
 	if err := sszCodec.UnmarshalSSZ(resp, data); err != nil {
 		if log.GetLevel() >= log.DebugLevel {