fix cicd

.github/workflows/generic-testing.yml

@@ -3,22 +3,22 @@ on:
   workflow_call:
     inputs:
       stack-orchestrator-ref:
-        required: false
+        required: true
         type: string
       ipld-eth-beacon-db-ref:
-        required: false
+        required: true
         type: string
       ssz-data-ref:
-        required: false
+        required: true
         type: string
     secrets:
       GHA_KEY:
         required: true
 
 env:
-  stack-orchestrator-ref: ${{ inputs.stack-orchestrator-ref || '7fb664270a0ba09e2caa3095e8c91f3fdb5b38af' }}
-  ipld-eth-beacon-db-ref: ${{ inputs.ipld-eth-beacon-db-ref  || '3dfe416302d553f8240f6051c08a7899b0e39e12' }}
-  ssz-data-ref: ${{ inputs.ssz-data-ref || 'main' }}
+  stack-orchestrator-ref: ${{ inputs.stack-orchestrator-ref }}
+  ipld-eth-beacon-db-ref: ${{ inputs.ipld-eth-beacon-db-ref }}
+  ssz-data-ref: ${{ inputs.ssz-data-ref }}
   GOPATH: /tmp/go
 jobs:
   build:

.github/workflows/on-pr-publish.yml

@@ -2,9 +2,50 @@ name: Publish Docker image
 on:
   release:
     types: [published, edited]
+  workflow_dispatch:
+    inputs:
+      stack-orchestrator-ref:
+        description: "The branch, commit or sha from stack-orchestrator to checkout"
+        required: false
+        default: "7fb664270a0ba09e2caa3095e8c91f3fdb5b38af"
+      ipld-eth-beacon-db-ref:
+        description: "The branch, commit or sha from ipld-eth-beacon-db to checkout"
+        required: false
+        default: "3dfe416302d553f8240f6051c08a7899b0e39e12"
+      ssz-data-ref:
+        description: "The branch, commit or sha from ssz-data to checkout"
+        required: false
+        default: "main"
+  pull_request:
+    paths:
+      - "!**.md"
+      - "!.gitignore"
+      - "!LICENSE"
+      - "!.github/workflows/**"
+      - ".github/workflows/on-pr.yml"
+      - ".github/workflows/tests.yml"
+      - "**"
+  schedule:
+    - cron: '0 13 * * *' # Must be single quotes!!
 jobs:
+  pre_job:
+    # continue-on-error: true # Uncomment once integration is finished
+    runs-on: ubuntu-latest
+    # Map a step output to a job output
+    outputs:
+      should_skip: ${{ steps.skip_check.outputs.should_skip }}
+    steps:
+      - id: skip_check
+        uses: fkirc/skip-duplicate-actions@v4
+        with:
+          # All of these options are optional, so you can remove them if you are happy with the defaults
+          concurrent_skipping: "never"
+          skip_after_successful_duplicate: "true"
+          do_not_skip: '["workflow_dispatch", "schedule"]'
   trigger-tests:
     uses: ./.github/workflows/generic-testing.yml
+    if: ${{ needs.pre_job.outputs.should_skip != 'true' }}
+    needs: pre_job
     with:
       stack-orchestrator-ref: ${{ github.event.inputs.stack-orchestrator-ref }}
       ipld-eth-beacon-db-ref: ${{ github.event.inputs.ipld-eth-beacon-db-ref }}
@@ -12,6 +53,8 @@ jobs:
     secrets:
       GHA_KEY: ${{secrets.GHA_KEY}}
   system-testing:
+    if: ${{ needs.pre_job.outputs.should_skip != 'true' }}
+    needs: pre_job
     uses: ./.github/workflows/system-tests.yml
     with:
       stack-orchestrator-ref: ${{ github.event.inputs.stack-orchestrator-ref }}
@@ -25,6 +68,11 @@ jobs:
     needs:
       - trigger-tests
       - system-testing
+    if: |
+      always() &&
+      (needs.trigger-tests.result == 'success' || needs.trigger-tests.result == 'skipped') &&
+      (needs.system-testing.result == 'success' || needs.system-testing.result == 'skipped') &&
+      github.event_name == 'release'
     steps:
       - uses: actions/checkout@v2
       - name: Get the version
@@ -42,6 +90,10 @@ jobs:
     name: Push Docker image to Docker Hub
     runs-on: ubuntu-latest
     needs: build
+    if: |
+      always() &&
+      (needs.build.result == 'success') &&
+      github.event_name == 'release'
     steps:
       - name: Get the version
         id: vars

.github/workflows/on-pr.yml

@@ -1,47 +0,0 @@
-name: Test Application On PR
-
-on:
-  workflow_dispatch:
-    inputs:
-      stack-orchestrator-ref:
-        description: "The branch, commit or sha from stack-orchestrator to checkout"
-        required: false
-        default: "main"
-      ipld-eth-beacon-db-ref:
-        description: "The branch, commit or sha from ipld-eth-beacon-db to checkout"
-        required: false
-        default: "main"
-      ssz-data-ref:
-        description: "The branch, commit or sha from ssz-data to checkout"
-        required: false
-        default: "main"
-  pull_request:
-    paths:
-      - "!**.md"
-      - "!.gitignore"
-      - "!LICENSE"
-      - "!.github/workflows/**"
-      - ".github/workflows/on-pr.yml"
-      - ".github/workflows/tests.yml"
-      - "**"
-  schedule:
-    - cron: '0 13 * * *' # Must be single quotes!!
-
-jobs:
-  trigger-tests:
-    if: github.event_name != 'schedule'
-    uses: ./.github/workflows/generic-testing.yml
-    with:
-      stack-orchestrator-ref: ${{ github.event.inputs.stack-orchestrator-ref }}
-      ipld-eth-beacon-db-ref: ${{ github.event.inputs.ipld-eth-beacon-db-ref }}
-      ssz-data-ref: ${{ github.event.inputs.ssz-data-ref }}
-    secrets:
-      GHA_KEY: ${{secrets.GHA_KEY}}
-  system-testing:
-    uses: ./.github/workflows/system-tests.yml
-    with:
-      stack-orchestrator-ref: ${{ github.event.inputs.stack-orchestrator-ref }}
-      ipld-eth-beacon-db-ref: ${{ github.event.inputs.ipld-eth-beacon-db-ref }}
-    secrets:
-      GHA_KEY: ${{secrets.GHA_KEY}}
-      BC_ADDRESS: ${{secrets.BC_ADDRESS}}

cmd/boot.go

@@ -18,7 +18,10 @@ package cmd
 
 import (
 	"context"
+	"fmt"
+	"net/http"
 	"os"
+	"strconv"
 	"syscall"
 
 	log "github.com/sirupsen/logrus"
@@ -60,6 +63,12 @@ func bootApp() {
 		notifierCh <- syscall.SIGTERM
 	}()
 
+	if viper.GetBool("t.pprof") {
+		go func() {
+			log.Println(http.ListenAndServe(fmt.Sprint("localhost:"+strconv.Itoa(viper.GetInt("t.pprofPort"))), nil))
+		}()
+	}
+
 	err = shutdown.ShutdownBoot(ctx, notifierCh, maxWaitSecondsShutdown, Db, Bc)
 	if err != nil {
 		loghelper.LogError(err).Error("Ungracefully Shutdown ipld-eth-beacon-indexer!")

cmd/capture.go

@@ -50,6 +50,8 @@ var (
 	maxWaitSecondsShutdown     time.Duration  = time.Duration(20) * time.Second
 	notifierCh                 chan os.Signal = make(chan os.Signal, 1)
 	testDisregardSync          bool
+	isTestPprof                bool
+	testPprofPort              int
 )
 
 // captureCmd represents the capture command
@@ -114,6 +116,8 @@ func init() {
 
 	//// Testing Specific
 	captureCmd.PersistentFlags().BoolVar(&testDisregardSync, "t.skipSync", false, "Should we disregard the head sync?")
+	captureCmd.PersistentFlags().BoolVar(&isTestPprof, "t.pprof", false, "Should we start pprof?")
+	captureCmd.PersistentFlags().IntVar(&testPprofPort, "t.pprofPort", 6060, "What port should we export pprof at?")
 
 	// Bind Flags with Viper
 	//// DB Flags
@@ -133,6 +137,10 @@ func init() {
 	//// Testing Specific
 	err = viper.BindPFlag("t.skipSync", captureCmd.PersistentFlags().Lookup("t.skipSync"))
 	exitErr(err)
+	err = viper.BindPFlag("t.pprof", captureCmd.PersistentFlags().Lookup("t.pprof"))
+	exitErr(err)
+	err = viper.BindPFlag("t.pprofPort", captureCmd.PersistentFlags().Lookup("t.pprofPort"))
+	exitErr(err)
 
 	//// LH specific
 	err = viper.BindPFlag("bc.address", captureCmd.PersistentFlags().Lookup("bc.address"))

cmd/full.go

@@ -19,6 +19,7 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"net/http"
 	"strconv"
 
 	log "github.com/sirupsen/logrus"
@@ -107,6 +108,12 @@ func startFullProcessing() {
 		}()
 	}
 
+	if viper.GetBool("t.pprof") {
+		go func() {
+			log.Println(http.ListenAndServe(fmt.Sprint("localhost:"+strconv.Itoa(viper.GetInt("t.pprofPort"))), nil))
+		}()
+	}
+
 	// Shutdown when the time is right.
 	err = shutdown.ShutdownFull(ctx, KgCancel, hpCancel, notifierCh, maxWaitSecondsShutdown, Db, Bc)
 	if err != nil {

cmd/head.go

@@ -81,6 +81,12 @@ func startHeadTracking() {
 		}()
 	}
 
+	if viper.GetBool("t.pprof") {
+		go func() {
+			log.Println(http.ListenAndServe(fmt.Sprint("localhost:"+strconv.Itoa(viper.GetInt("t.pprofPort"))), nil))
+		}()
+	}
+
 	// Shutdown when the time is right.
 	err = shutdown.ShutdownHeadTracking(ctx, KgCancel, notifierCh, maxWaitSecondsShutdown, Db, Bc)
 	if err != nil {

cmd/historic.go

@@ -22,6 +22,9 @@ import (
 	"os"
 	"strconv"
 
+	"net/http"
+	_ "net/http/pprof"
+
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -92,6 +95,12 @@ func startHistoricProcessing() {
 		}()
 	}
 
+	if viper.GetBool("t.pprof") {
+		go func() {
+			log.Println(http.ListenAndServe(fmt.Sprint("localhost:"+strconv.Itoa(viper.GetInt("t.pprofPort"))), nil))
+		}()
+	}
+
 	// Shutdown when the time is right.
 	err = shutdown.ShutdownHistoricProcessing(ctx, kgCancel, hpCancel, notifierCh, maxWaitSecondsShutdown, Db, Bc)
 	if err != nil {

config/cicd/boot.ipld-eth-beacon-indexer.json

@@ -19,7 +19,9 @@
     "checkDb": true
   },
   "t": {
-    "skipSync": true
+    "skipSync": true,
+    "pprof": true,
+    "pprofPort": 6060
   },
   "log": {
     "level": "debug",

config/example.ipld-eth-beacon-indexer-config.json

@@ -19,7 +19,9 @@
     "checkDb": true
   },
   "t": {
-    "skipSync": true
+    "skipSync": true,
+    "pprof": true,
+    "pprofPort": 6060
   },
   "log": {
     "level": "debug",

pkg/beaconclient/processslot.go

@@ -63,12 +63,12 @@ type ProcessSlot struct {
 	PerformanceMetrics PerformanceMetrics   // An object to keep track of performance metrics.
 	// BeaconBlock
 
-	SszSignedBeaconBlock  []byte               // The entire SSZ encoded SignedBeaconBlock
+	SszSignedBeaconBlock  *[]byte              // The entire SSZ encoded SignedBeaconBlock
 	FullSignedBeaconBlock si.SignedBeaconBlock // The unmarshaled BeaconState object, the unmarshalling could have errors.
 
 	// BeaconState
 	FullBeaconState state.BeaconState // The unmarshaled BeaconState object, the unmarshalling could have errors.
-	SszBeaconState  []byte            // The entire SSZ encoded BeaconState
+	SszBeaconState  *[]byte           // The entire SSZ encoded BeaconState
 
 	// DB Write objects
 	DbSlotsModel             *DbSlots             // The model being written to the slots table.
@@ -155,6 +155,11 @@ func processFullSlot(ctx context.Context, db sql.Database, serverAddress string,
 		})
 
 		if err := g.Wait(); err != nil {
+			// Make sure channel is empty.
+			select {
+			case <-vUnmarshalerCh:
+			default:
+			}
 			return err, "processSlot"
 		}
 
@@ -270,7 +275,7 @@ func (ps *ProcessSlot) getSignedBeaconBlock(serverAddress string, vmCh <-chan *d
 	vm := <-vmCh
 	if rc != 200 {
 		ps.FullSignedBeaconBlock = &wrapper.Phase0SignedBeaconBlock{}
-		ps.SszSignedBeaconBlock = []byte{}
+		ps.SszSignedBeaconBlock = &[]byte{}
 		ps.ParentBlockRoot = ""
 		ps.Status = "skipped"
 		return nil
@@ -280,7 +285,7 @@ func (ps *ProcessSlot) getSignedBeaconBlock(serverAddress string, vmCh <-chan *d
 		return fmt.Errorf(VersionedUnmarshalerError)
 	}
 
-	ps.FullSignedBeaconBlock, err = vm.UnmarshalBeaconBlock(ps.SszSignedBeaconBlock)
+	ps.FullSignedBeaconBlock, err = vm.UnmarshalBeaconBlock(*ps.SszSignedBeaconBlock)
 	if err != nil {
 		loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Warn("Unable to process the slots SignedBeaconBlock")
 		return nil
@@ -300,14 +305,14 @@ func (ps *ProcessSlot) getBeaconState(serverEndpoint string, vmCh chan<- *dt.Ver
 	stateEndpoint := serverEndpoint + BcStateQueryEndpoint + stateIdentifier
 	ps.SszBeaconState, _, _ = querySsz(stateEndpoint, strconv.Itoa(ps.Slot))
 
-	versionedUnmarshaler, err := dt.FromState(ps.SszBeaconState)
+	versionedUnmarshaler, err := dt.FromState(*ps.SszBeaconState)
 	if err != nil {
 		loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error(VersionedUnmarshalerError)
 		vmCh <- nil
 		return fmt.Errorf(VersionedUnmarshalerError)
 	}
 	vmCh <- versionedUnmarshaler
-	ps.FullBeaconState, err = versionedUnmarshaler.UnmarshalBeaconState(ps.SszBeaconState)
+	ps.FullBeaconState, err = versionedUnmarshaler.UnmarshalBeaconState(*ps.SszBeaconState)
 	if err != nil {
 		loghelper.LogSlotError(strconv.Itoa(ps.Slot), err).Error("Unable to process the slots BeaconState")
 		return err
@@ -356,7 +361,7 @@ func (ps *ProcessSlot) createWriteObjects(blockRoot, stateRoot, eth1BlockHash st
 		status = "proposed"
 	}
 
-	dw, err := CreateDatabaseWrite(ps.Db, ps.Slot, stateRoot, blockRoot, ps.ParentBlockRoot, eth1BlockHash, status, &ps.SszSignedBeaconBlock, &ps.SszBeaconState, ps.Metrics)
+	dw, err := CreateDatabaseWrite(ps.Db, ps.Slot, stateRoot, blockRoot, ps.ParentBlockRoot, eth1BlockHash, status, ps.SszSignedBeaconBlock, ps.SszBeaconState, ps.Metrics)
 	if err != nil {
 		return dw, err
 	}

pkg/beaconclient/queryserver.go

@@ -28,7 +28,7 @@ import (
 )
 
 // A helper function to query endpoints that utilize slots.
-func querySsz(endpoint string, slot string) ([]byte, int, error) {
+func querySsz(endpoint string, slot string) (*[]byte, int, error) {
 	log.WithFields(log.Fields{"endpoint": endpoint}).Debug("Querying endpoint")
 	client := &http.Client{}
 	req, err := http.NewRequest("GET", endpoint, nil)
@@ -49,7 +49,7 @@ func querySsz(endpoint string, slot string) ([]byte, int, error) {
 		loghelper.LogSlotError(slot, err).Error("Unable to turn response into a []bytes array!")
 		return nil, rc, fmt.Errorf("Unable to turn response into a []bytes array!: %s", err.Error())
 	}
-	return body, rc, nil
+	return &body, rc, nil
 }
 
 // Object to unmarshal the BlockRootResponse