feat: Parse Beacon State (#171)

* feat: Parse Beacon State * style: Remove unnecessary timeout in action.yaml * fix: Update default networks in manual-integration.yaml * fix: Update network configuration to mainnet * feat: Add dynamic address for state-provider beacon node * feat: Add endpoint checks with retries and error handling * chore: Update shell script for endpoint checking * fix: Update endpoint URLs for debug mode * style: improve error message handling in endpoint check * refactor: Improve response handling and readability * style: improve code readability and consistency * chore: update integration workflow timeout to 10 minutes
ethpandaops · Jun 20, 2024 · 49ddba9 · 49ddba9
1 parent a2951b0
commit 49ddba9
Show file tree

Hide file tree

Showing 10 changed files with 152 additions and 47 deletions.
diff --git a/.github/actions/checkpoint-sync/action.yaml b/.github/actions/checkpoint-sync/action.yaml
@@ -28,6 +28,16 @@ runs:
     - name: Configure checkpointz
       shell: bash
       run: |
+        beacon_node=""
+        if [[ ${{ inputs.network }} == "mainnet" ]]; then
+          beacon_node="http://testing.mainnet.beacon-api.nimbus.team/"
+        elif [[ ${{ inputs.network }} == "holesky" ]]; then
+          beacon_node="http://testing.holesky.beacon-api.nimbus.team/"
+        else
+          echo "Unsupported network: ${{ inputs.network }}"
+          exit 1
+        fi
+
         cat <<EOF > checkpointz.yaml
         global:
           listenAddr: ":5555"
@@ -36,7 +46,7 @@ runs:
         beacon:
           upstreams:
           - name: state-provider
-            address: https://checkpoint-sync.${{ inputs.network }}.ethpandaops.io
+            address: $beacon_node
             timeoutSeconds: 30
             dataProvider: true
         checkpointz:
@@ -76,6 +86,101 @@ runs:
         echo "Waiting for checkpointz to have the genesis block...";
         bash -c 'while [[ "$(curl -s -o /dev/null -w ''%{http_code}'' localhost:5555/eth/v2/beacon/blocks/0)" != "200" ]]; do sleep 1; done';
         echo "Checkpointz has the genesis block.";
+    - name: Manually check endpoints
+      shell: bash
+      run: |
+        set +e
+        check_endpoint() {
+          local path=$1
+          local accept_header=$2
+          local response=$(curl -s -w "\n%{http_code}" -H "accept: ${accept_header}" "localhost:5555${path}" | tr -d '\0')
+          local http_code=$(echo "$response" | tail -n1)
+          local json_response=$(echo "$response" | sed '$d')
+          if [[ $http_code -ge 200 && $http_code -lt 300 ]]; then
+            echo $json_response
+            return 0
+          else
+            echo "Endpoint ${path} with accept header ${accept_header} is not available."
+            return 1
+          fi
+        }
+        fails=0
+
+        while true; do
+          if [[ $fails -gt 0 ]]; then
+            echo "Failed $fails times, retrying..."
+            sleep 1;
+          fi
+          fails=$((fails+1))
+          
+          echo "Checking endpoint /eth/v2/beacon/blocks/genesis with accept header application/json..."
+          genesis_block=$(check_endpoint "/eth/v2/beacon/blocks/genesis" "application/json")
+          [[ $? -ne 0 ]] && continue
+          echo "Genesis block is available."
+
+          echo "Checking endpoint /eth/v2/debug/beacon/states/genesis with accept header application/octet-stream..."
+          genesis_state=$(check_endpoint "/eth/v2/debug/beacon/states/genesis" "application/octet-stream")
+          [[ $? -ne 0 ]] && continue
+          echo "Genesis state is available."
+
+          echo "Checking endpoint /eth/v2/debug/beacon/states/finalized with accept header application/octet-stream..."
+          finalized_state=$(check_endpoint "/eth/v2/debug/beacon/states/finalized" "application/octet-stream")
+          [[ $? -ne 0 ]] && continue
+          echo "Finalized state is available."
+
+          echo "Checking endpoint /eth/v1/beacon/states/finalized/finality_checkpoints with accept header application/json..."
+          finality_checkpoints=$(check_endpoint "/eth/v1/beacon/states/finalized/finality_checkpoints" "application/json")
+          if [[ $? -ne 0 ]]; then
+            echo "Finality checkpoints endpoint is not available."
+            continue
+          fi
+          echo "Finality checkpoints endpoint is available."
+
+          finalized_root=$(echo $finality_checkpoints | jq -r '.data.finalized.root')
+          if [[ -z "$finalized_root" ]]; then
+            echo "Failed to extract finalized root from the finality checkpoints."
+            continue
+          fi
+          echo "Extracted finalized root: $finalized_root"
+
+          echo "Checking endpoint /eth/v2/beacon/blocks/$finalized_root with accept header application/json..."
+          finalized_block=$(check_endpoint "/eth/v2/beacon/blocks/$finalized_root" "application/json")
+          [[ $? -ne 0 ]] && continue
+          echo "Finalized block is available."
+
+          finalized_state_root=$(echo $finalized_block | jq -r '.data.message.state_root')
+          if [[ -z "$finalized_state_root" ]]; then
+            echo "Failed to extract finalized state root from the finalized block."
+            continue
+          fi
+          echo "Extracted finalized state root: $finalized_state_root"
+
+          echo "Checking endpoint /eth/v2/debug/beacon/states/$finalized_state_root with accept header application/octet-stream..."
+          finalized_state=$(check_endpoint "/eth/v2/debug/beacon/states/$finalized_state_root" "application/octet-stream")
+          [[ $? -ne 0 ]] && continue
+          echo "Finalized state is available."
+
+          echo "Fetching slot from /eth/v2/beacon/blocks/$finalized_root..."
+          finalized_slot=$(echo $finalized_block | jq -r '.data.message.slot')
+          if [[ -z "$finalized_slot" ]]; then
+            echo "Failed to extract slot from the finalized block."
+            continue
+          fi
+          echo "Extracted slot: $finalized_slot"
+
+          echo "Checking endpoint /eth/v2/beacon/blocks/$finalized_slot with accept header application/json..."
+          block=$(check_endpoint "/eth/v2/beacon/blocks/$finalized_slot" "application/json")
+          [[ $? -ne 0 ]] && continue
+          echo "Block for finalized slot $finalized_slot is available."
+
+          echo "Checking endpoint /eth/v2/beacon/blocks/finalized with accept header application/json..."
+          finalized_block_via_finalized=$(check_endpoint "/eth/v2/beacon/blocks/finalized" "application/json")
+          [[ $? -ne 0 ]] && continue
+          echo "Finalized block via 'finalized' endpoint is available."
+
+          echo "All endpoints are available."
+          break;
+        done;
     - name: Run teku client
       shell: bash
       if:  ${{ inputs.consensus == 'teku' }}

diff --git a/.github/workflows/goreleaser.yaml b/.github/workflows/goreleaser.yaml
@@ -10,10 +10,8 @@ jobs:
     permissions:
       contents: write
     runs-on:
-      - environment=production
-      - size=xlarge
-      - provider=ethpandaops
-      - realm=platform
+      - self-hosted-ghr
+      - size-l-x64
     steps:
       -
         name: Checkout

diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
@@ -13,9 +13,9 @@ jobs:
       fail-fast: false
       matrix:
         consensus: [lighthouse, teku, prysm, nimbus, lodestar]
-        network: [sepolia]
+        network: [mainnet]
     runs-on: ubuntu-latest
-    timeout-minutes: 5
+    timeout-minutes: 10
     steps:
       - uses: actions/checkout@v3
       - name: Print details

diff --git a/.github/workflows/manual-integration.yaml b/.github/workflows/manual-integration.yaml
@@ -11,7 +11,7 @@ on:
         description: 'Networks to use'    
         required: true
         type: string
-        default: ropsten, sepolia, goerli
+        default: mainnet, sepolia, holesky
 
 jobs:
   init:

diff --git a/pkg/api/handler.go b/pkg/api/handler.go
@@ -224,27 +224,12 @@ func (h *Handler) handleEthV2DebugBeaconStates(ctx context.Context, r *http.Requ
 		return NewUnsupportedMediaTypeResponse(nil), err
 	}
 
-	blockID, err := eth.NewBlockIdentifier(p.ByName("state_id"))
+	id, err := eth.NewStateIdentifier(p.ByName("state_id"))
 	if err != nil {
 		return NewBadRequestResponse(nil), err
 	}
 
-	block, err := h.eth.BeaconBlock(ctx, blockID)
-	if err != nil {
-		return NewInternalServerErrorResponse(nil), err
-	}
-
-	slot, err := block.Slot()
-	if err != nil {
-		return NewInternalServerErrorResponse(nil), err
-	}
-
-	stateID, err := eth.NewStateIdentifier(fmt.Sprintf("%d", slot))
-	if err != nil {
-		return NewInternalServerErrorResponse(nil), err
-	}
-
-	state, err := h.eth.BeaconState(ctx, stateID)
+	state, err := h.eth.BeaconState(ctx, id)
 	if err != nil {
 		return NewInternalServerErrorResponse(nil), err
 	}
@@ -255,22 +240,35 @@ func (h *Handler) handleEthV2DebugBeaconStates(ctx context.Context, r *http.Requ
 
 	rsp := NewSuccessResponse(ContentTypeResolvers{
 		ContentTypeSSZ: func() ([]byte, error) {
-			return *state, nil
+			switch state.Version {
+			case spec.DataVersionPhase0:
+				return state.Phase0.MarshalSSZ()
+			case spec.DataVersionAltair:
+				return state.Altair.MarshalSSZ()
+			case spec.DataVersionBellatrix:
+				return state.Bellatrix.MarshalSSZ()
+			case spec.DataVersionCapella:
+				return state.Capella.MarshalSSZ()
+			case spec.DataVersionDeneb:
+				return state.Deneb.MarshalSSZ()
+			default:
+				return nil, fmt.Errorf("unknown state version: %s", state.Version.String())
+			}
 		},
 	})
 
-	switch blockID.Type() {
-	case eth.BlockIDRoot, eth.BlockIDGenesis, eth.BlockIDSlot:
-		// TODO(sam.calder-mason): This should be calculated using the Weak-Subjectivity period.
+	switch id.Type() {
+	case eth.StateIDSlot:
 		rsp.SetCacheControl("public, s-max-age=6000")
-	case eth.BlockIDFinalized:
-		// TODO(sam.calder-mason): This should be calculated using the Weak-Subjectivity period.
+	case eth.StateIDFinalized:
 		rsp.SetCacheControl("public, s-max-age=180")
-	case eth.BlockIDHead:
+	case eth.StateIDRoot:
+		rsp.SetCacheControl("public, s-max-age=6000")
+	case eth.StateIDHead:
 		rsp.SetCacheControl("public, s-max-age=30")
 	}
 
-	rsp.SetEthConsensusVersion(block.Version.String())
+	rsp.SetEthConsensusVersion(state.Version.String())
 
 	return rsp, nil
 }

diff --git a/pkg/beacon/default.go b/pkg/beacon/default.go
@@ -613,7 +613,7 @@ func (d *Default) GetBlobSidecarsBySlot(ctx context.Context, slot phase0.Slot) (
 	return d.blobSidecars.GetBySlot(slot)
 }
 
-func (d *Default) GetBeaconStateBySlot(ctx context.Context, slot phase0.Slot) (*[]byte, error) {
+func (d *Default) GetBeaconStateBySlot(ctx context.Context, slot phase0.Slot) (*spec.VersionedBeaconState, error) {
 	block, err := d.GetBlockBySlot(ctx, slot)
 	if err != nil {
 		return nil, err
@@ -627,11 +627,11 @@ func (d *Default) GetBeaconStateBySlot(ctx context.Context, slot phase0.Slot) (*
 	return d.states.GetByStateRoot(stateRoot)
 }
 
-func (d *Default) GetBeaconStateByStateRoot(ctx context.Context, stateRoot phase0.Root) (*[]byte, error) {
+func (d *Default) GetBeaconStateByStateRoot(ctx context.Context, stateRoot phase0.Root) (*spec.VersionedBeaconState, error) {
 	return d.states.GetByStateRoot(stateRoot)
 }
 
-func (d *Default) GetBeaconStateByRoot(ctx context.Context, root phase0.Root) (*[]byte, error) {
+func (d *Default) GetBeaconStateByRoot(ctx context.Context, root phase0.Root) (*spec.VersionedBeaconState, error) {
 	block, err := d.GetBlockByRoot(ctx, root)
 	if err != nil {
 		return nil, err

diff --git a/pkg/beacon/download.go b/pkg/beacon/download.go
@@ -357,7 +357,10 @@ func (d *Default) fetchBundle(ctx context.Context, root phase0.Root, upstream *N
 		}
 	}
 
-	d.log.WithField("root", eth.RootAsString(root)).Infof("Successfully fetched bundle from %s", upstream.Config.Name)
+	d.log.WithFields(logrus.Fields{
+		"block_root": eth.RootAsString(root),
+		"state_root": eth.RootAsString(stateRoot),
+	}).Infof("Successfully fetched bundle from %s", upstream.Config.Name)
 
 	return block, nil
 }
@@ -369,7 +372,7 @@ func (d *Default) downloadAndStoreBeaconState(ctx context.Context, stateRoot pha
 		return nil
 	}
 
-	beaconState, err := node.Beacon.FetchRawBeaconState(ctx, eth.SlotAsString(slot), "application/octet-stream")
+	beaconState, err := node.Beacon.FetchBeaconState(ctx, eth.SlotAsString(slot))
 	if err != nil {
 		return fmt.Errorf("failed to fetch beacon state: %w", err)
 	}
@@ -383,7 +386,7 @@ func (d *Default) downloadAndStoreBeaconState(ctx context.Context, stateRoot pha
 		expiresAt = time.Now().Add(999999 * time.Hour)
 	}
 
-	if err := d.states.Add(stateRoot, &beaconState, expiresAt, slot); err != nil {
+	if err := d.states.Add(stateRoot, beaconState, expiresAt, slot); err != nil {
 		return fmt.Errorf("failed to store beacon state: %w", err)
 	}
 

diff --git a/pkg/beacon/finality_provider.go b/pkg/beacon/finality_provider.go
@@ -43,11 +43,11 @@ type FinalityProvider interface {
 	// GetBlockByStateRoot returns the block with the given root.
 	GetBlockByStateRoot(ctx context.Context, root phase0.Root) (*spec.VersionedSignedBeaconBlock, error)
 	// GetBeaconStateBySlot returns the beacon sate with the given slot.
-	GetBeaconStateBySlot(ctx context.Context, slot phase0.Slot) (*[]byte, error)
+	GetBeaconStateBySlot(ctx context.Context, slot phase0.Slot) (*spec.VersionedBeaconState, error)
 	// GetBeaconStateByStateRoot returns the beacon sate with the given state root.
-	GetBeaconStateByStateRoot(ctx context.Context, root phase0.Root) (*[]byte, error)
+	GetBeaconStateByStateRoot(ctx context.Context, root phase0.Root) (*spec.VersionedBeaconState, error)
 	// GetBeaconStateByRoot returns the beacon sate with the given root.
-	GetBeaconStateByRoot(ctx context.Context, root phase0.Root) (*[]byte, error)
+	GetBeaconStateByRoot(ctx context.Context, root phase0.Root) (*spec.VersionedBeaconState, error)
 	// GetBlobSidecarsBySlot returns the blob sidecars for the given slot.
 	GetBlobSidecarsBySlot(ctx context.Context, slot phase0.Slot) ([]*deneb.BlobSidecar, error)
 	// ListFinalizedSlots returns a slice of finalized slots.

diff --git a/pkg/beacon/store/state.go b/pkg/beacon/store/state.go
@@ -4,6 +4,7 @@ import (
 	"errors"
 	"time"
 
+	"github.com/attestantio/go-eth2-client/spec"
 	"github.com/attestantio/go-eth2-client/spec/phase0"
 	"github.com/ethpandaops/checkpointz/pkg/cache"
 	"github.com/ethpandaops/checkpointz/pkg/eth"
@@ -30,7 +31,7 @@ func NewBeaconState(log logrus.FieldLogger, config Config, namespace string) *Be
 	return c
 }
 
-func (c *BeaconState) Add(stateRoot phase0.Root, state *[]byte, expiresAt time.Time, slot phase0.Slot) error {
+func (c *BeaconState) Add(stateRoot phase0.Root, state *spec.VersionedBeaconState, expiresAt time.Time, slot phase0.Slot) error {
 	invincible := false
 	if slot == 0 {
 		invincible = true
@@ -48,7 +49,7 @@ func (c *BeaconState) Add(stateRoot phase0.Root, state *[]byte, expiresAt time.T
 	return nil
 }
 
-func (c *BeaconState) GetByStateRoot(stateRoot phase0.Root) (*[]byte, error) {
+func (c *BeaconState) GetByStateRoot(stateRoot phase0.Root) (*spec.VersionedBeaconState, error) {
 	data, _, err := c.store.Get(eth.RootAsString(stateRoot))
 	if err != nil {
 		return nil, err
@@ -57,8 +58,8 @@ func (c *BeaconState) GetByStateRoot(stateRoot phase0.Root) (*[]byte, error) {
 	return c.parseState(data)
 }
 
-func (c *BeaconState) parseState(data interface{}) (*[]byte, error) {
-	state, ok := data.(*[]byte)
+func (c *BeaconState) parseState(data interface{}) (*spec.VersionedBeaconState, error) {
+	state, ok := data.(*spec.VersionedBeaconState)
 	if !ok {
 		return nil, errors.New("invalid state")
 	}

diff --git a/pkg/service/eth/eth.go b/pkg/service/eth/eth.go
@@ -267,7 +267,7 @@ func (h *Handler) PeerCount(ctx context.Context) (uint64, error) {
 }
 
 // BeaconState returns the beacon state for the given state id.
-func (h *Handler) BeaconState(ctx context.Context, stateID StateIdentifier) (*[]byte, error) {
+func (h *Handler) BeaconState(ctx context.Context, stateID StateIdentifier) (*spec.VersionedBeaconState, error) {
 	var err error
 
 	const call = "beacon_state"