Merge pull request #59 from vulcanize/develop

Include all changes needed for historical and knownGaps processing

.github/workflows/generic-testing.yml

@@ -3,22 +3,22 @@ on:
   workflow_call:
     inputs:
       stack-orchestrator-ref:
-        required: true
+        required: false
         type: string
       ipld-eth-beacon-db-ref:
-        required: true
+        required: false
         type: string
       ssz-data-ref:
-        required: true
+        required: false
         type: string
     secrets:
       GHA_KEY:
         required: true
 
 env:
-  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 }}
+  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' }}
   GOPATH: /tmp/go
 jobs:
   build:
@@ -49,7 +49,6 @@ jobs:
           echo vulcanize_ipld_eth_beacon_db=$GITHUB_WORKSPACE/ipld-eth-beacon-db/ > ./config.sh
           echo vulcanize_ipld_eth_beacon_indexer=$GITHUB_WORKSPACE/ipld-eth-beacon-indexer >> ./config.sh
           echo eth_beacon_config_file=$GITHUB_WORKSPACE/ipld-eth-beacon-indexer/config/cicd/boot.ipld-eth-beacon-indexer.json >> ./config.sh
-          echo eth_beacon_capture_mode=boot >> ./config.sh
           cat ./config.sh
 
       - name: Run docker compose
@@ -160,6 +159,7 @@ jobs:
         run: |
           echo vulcanize_ipld_eth_beacon_db=$GITHUB_WORKSPACE/ipld-eth-beacon-db/ > ./config.sh
           echo vulcanize_ipld_eth_beacon_indexer=$GITHUB_WORKSPACE/ipld-eth-beacon-indexer >> ./config.sh
+          echo eth_beacon_capture_mode=boot >> ./config.sh
           cat ./config.sh
 
       - name: Run docker compose

.github/workflows/on-pr.yml

@@ -0,0 +1,47 @@
+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}}

.github/workflows/on-publish.yml

@@ -2,50 +2,9 @@ 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 }}
@@ -53,8 +12,6 @@ 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 }}
@@ -68,11 +25,6 @@ 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
@@ -90,10 +42,6 @@ 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

.gitignore

@@ -8,6 +8,3 @@ temp/*
 pkg/beaconclient/ssz-data/
 *.test
 ipld-eth-beacon-indexer.log
-ipld-eth-beacon-indexer
-config/local.ipld-eth-beacon-indexer-config.json
-config/docker.ipld-eth-beacon-indexer-config.json

Dockerfile

@@ -1,7 +1,7 @@
 FROM golang:1.18-alpine as builder
 
 WORKDIR /go/src/github.com/vulcanize/ipld-eth-beacon-indexer
-RUN apk --no-cache add ca-certificates make git g++ linux-headers libstdc++
+RUN apk --no-cache add ca-certificates make git g++ linux-headers
 
 ENV GO111MODULE=on
 COPY go.mod .
@@ -9,11 +9,11 @@ COPY go.sum .
 RUN go mod tidy; go mod download
 COPY . .
 
-RUN GCO_ENABLED=0 GOOS=linux go build -race -ldflags="-s -w" -o ipld-eth-beacon-indexer .
+RUN GCO_ENABLED=0 GOOS=linux go build -race -a -installsuffix cgo -ldflags '-extldflags "-static"' -o ipld-eth-beacon-indexer .
 RUN chmod +x ipld-eth-beacon-indexer
 
 FROM frolvlad/alpine-bash:latest
-RUN apk --no-cache add ca-certificates libstdc++
+RUN apk --no-cache add ca-certificates
 WORKDIR /root/
 COPY --from=builder /go/src/github.com/vulcanize/ipld-eth-beacon-indexer/ipld-eth-beacon-indexer /root/ipld-eth-beacon-indexer
 ADD entrypoint.sh .

Makefile

@@ -67,7 +67,6 @@ unit-test-local:
 	go fmt ./...
 	$(GINKGO) -r --label-filter unit \
 	--randomize-all --randomize-suites \
-	--flake-attempts=3 \
 	--fail-on-pending --keep-going \
 	--trace
 
@@ -76,8 +75,7 @@ unit-test-ci:
 	go vet ./...
 	go fmt ./...
 	$(GINKGO) -r --label-filter unit \
-	--randomize-all --randomize-suites
-	--flake-attempts=3 \
+	--randomize-all --randomize-suites \
 	--fail-on-pending --keep-going \
 	--cover --coverprofile=cover.profile \
 	--trace --json-report=report.json

cmd/boot.go

@@ -18,10 +18,7 @@ package cmd
 
 import (
 	"context"
-	"fmt"
-	"net/http"
 	"os"
-	"strconv"
 	"syscall"
 
 	log "github.com/sirupsen/logrus"
@@ -63,12 +60,6 @@ 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,8 +50,6 @@ 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
@@ -116,8 +114,6 @@ 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
@@ -137,10 +133,6 @@ 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,7 +19,6 @@ package cmd
 import (
 	"context"
 	"fmt"
-	"net/http"
 	"strconv"
 
 	log "github.com/sirupsen/logrus"
@@ -108,12 +107,6 @@ 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,12 +81,6 @@ 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,9 +22,6 @@ import (
 	"os"
 	"strconv"
 
-	"net/http"
-	_ "net/http/pprof"
-
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -95,12 +92,6 @@ 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,9 +19,7 @@
     "checkDb": true
   },
   "t": {
-    "skipSync": true,
-    "pprof": true,
-    "pprofPort": 6060
+    "skipSync": true
   },
   "log": {
     "level": "debug",

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

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

entrypoint.sh

@@ -3,22 +3,17 @@
 sleep 10
 echo "Starting ipld-eth-beacon-indexer"
 
-echo /root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json > /root/ipld-eth-beacon-indexer.output
+echo /root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json
 
-if [ ${CAPTURE_MODE} == "boot" ]; then
-    /root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json > /root/ipld-eth-beacon-indexer.output
-    rv=$?
+/root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json
+rv=$?
 
-    if [ $rv != 0 ]; then
-      echo "ipld-eth-beacon-indexer boot failed"
-    else
-      echo "ipld-eth-beacon-indexer boot succeeded"
-    fi
-      echo $rv > /root/HEALTH
-      echo $rv
-      cat /root/ipld-eth-beacon-indexer.output
-
-    tail -f /dev/null
+if [ $rv != 0 ]; then
+  echo "ipld-eth-beacon-indexer startup failed"
+  echo 1 > /root/HEALTH
 else
-    exec /root/ipld-eth-beacon-indexer capture ${CAPTURE_MODE} --config /root/ipld-eth-beacon-config.json > /root/ipld-eth-beacon-indexer.output
-fi
+  echo "ipld-eth-beacon-indexer startup succeeded"
+  echo 0 > /root/HEALTH
+fi
+
+tail -f /dev/null

pkg/beaconclient/capturehistoric.go

@@ -31,7 +31,7 @@ import (
 func (bc *BeaconClient) CaptureHistoric(ctx context.Context, maxWorkers int) []error {
 	log.Info("We are starting the historical processing service.")
 	bc.HistoricalProcess = HistoricProcessing{db: bc.Db, metrics: bc.Metrics, uniqueNodeIdentifier: bc.UniqueNodeIdentifier}
-	errs := handleBatchProcess(ctx, maxWorkers, bc.HistoricalProcess, bc.HistoricalProcess.db, bc.ServerEndpoint, bc.Metrics, bc.CheckDb, bc.Metrics.IncrementHistoricSlotProcessed)
+	errs := handleBatchProcess(ctx, maxWorkers, bc.HistoricalProcess, bc.HistoricalProcess.db, bc.ServerEndpoint, bc.Metrics, bc.CheckDb)
 	log.Debug("Exiting Historical")
 	return errs
 }
@@ -39,8 +39,7 @@ func (bc *BeaconClient) CaptureHistoric(ctx context.Context, maxWorkers int) []e
 // This function will perform all the necessary clean up tasks for stopping historical processing.
 func (bc *BeaconClient) StopHistoric(cancel context.CancelFunc) error {
 	log.Info("We are stopping the historical processing service.")
-	cancel()
-	err := bc.HistoricalProcess.releaseDbLocks()
+	err := bc.HistoricalProcess.releaseDbLocks(cancel)
 	if err != nil {
 		loghelper.LogError(err).WithField("uniqueIdentifier", bc.UniqueNodeIdentifier).Error("We were unable to remove the locks from the eth_beacon.historic_processing table. Manual Intervention is needed!")
 	}
@@ -56,7 +55,7 @@ type BatchProcessing interface {
 	getSlotRange(context.Context, chan<- slotsToProcess) []error       // Write the slots to process in a channel, return an error if you cant get the next slots to write.
 	handleProcessingErrors(context.Context, <-chan batchHistoricError) // Custom logic to handle errors.
 	removeTableEntry(context.Context, <-chan slotsToProcess) error     // With the provided start and end slot, remove the entry from the database.
-	releaseDbLocks() error                                             // Update the checked_out column to false for whatever table is being updated.
+	releaseDbLocks(context.CancelFunc) error                           // Update the checked_out column to false for whatever table is being updated.
 }
 
 /// ^^^
@@ -91,24 +90,17 @@ type batchHistoricError struct {
 // 4. Remove the slot entry from the DB.
 //
 // 5. Handle any errors.
-func handleBatchProcess(ctx context.Context, maxWorkers int, bp BatchProcessing, db sql.Database, serverEndpoint string, metrics *BeaconClientMetrics, checkDb bool, incrementTracker func(uint64)) []error {
+func handleBatchProcess(ctx context.Context, maxWorkers int, bp BatchProcessing, db sql.Database, serverEndpoint string, metrics *BeaconClientMetrics, checkDb bool) []error {
 	slotsCh := make(chan slotsToProcess)
 	workCh := make(chan int)
 	processedCh := make(chan slotsToProcess)
 	errCh := make(chan batchHistoricError)
 	finalErrCh := make(chan []error, 1)
 
-	// Checkout Rows with same node Identifier.
-	err := bp.releaseDbLocks()
-	if err != nil {
-		loghelper.LogError(err).Error(("We are unable to un-checkout entries at the start!"))
-	}
-
 	// Start workers
 	for w := 1; w <= maxWorkers; w++ {
-		log.WithFields(log.Fields{"maxWorkers": maxWorkers}).Debug("Starting batch processing workers")
-
-		go processSlotRangeWorker(ctx, workCh, errCh, db, serverEndpoint, metrics, checkDb, incrementTracker)
+		log.WithFields(log.Fields{"maxWorkers": maxWorkers}).Debug("Starting batch  processing workers")
+		go processSlotRangeWorker(ctx, workCh, errCh, db, serverEndpoint, metrics, checkDb)
 	}
 
 	// Process all ranges and send each individual slot to the worker.
@@ -133,7 +125,6 @@ func handleBatchProcess(ctx context.Context, maxWorkers int, bp BatchProcessing,
 				} else if slots.startSlot == slots.endSlot {
 					log.WithField("slot", slots.startSlot).Debug("Added new slot to workCh")
 					workCh <- slots.startSlot
-					processedCh <- slots
 				} else {
 					for i := slots.startSlot; i <= slots.endSlot; i++ {
 						workCh <- i

pkg/beaconclient/databasewrite.go

@@ -90,12 +90,12 @@ type DatabaseWriter struct {
 	DbSlots              *DbSlots
 	DbSignedBeaconBlock  *DbSignedBeaconBlock
 	DbBeaconState        *DbBeaconState
-	rawBeaconState       *[]byte
-	rawSignedBeaconBlock *[]byte
+	rawBeaconState       []byte
+	rawSignedBeaconBlock []byte
 }
 
 func CreateDatabaseWrite(db sql.Database, slot int, stateRoot string, blockRoot string, parentBlockRoot string,
-	eth1BlockHash string, status string, rawSignedBeaconBlock *[]byte, rawBeaconState *[]byte, metrics *BeaconClientMetrics) (*DatabaseWriter, error) {
+	eth1BlockHash string, status string, rawSignedBeaconBlock []byte, rawBeaconState []byte, metrics *BeaconClientMetrics) (*DatabaseWriter, error) {
 	ctx := context.Background()
 	tx, err := db.Begin(ctx)
 	if err != nil {
@@ -194,7 +194,6 @@ func (dw *DatabaseWriter) transactFullSlot() error {
 		//	return err
 		//}
 		// Might want to seperate writing to public.blocks so we can do this concurrently...
-		// Cant concurrently write because we are using a transaction.
 		err := dw.transactSignedBeaconBlocks()
 		if err != nil {
 			loghelper.LogSlotError(dw.DbSlots.Slot, err).Error("We couldn't write to the eth_beacon block table...")
@@ -241,8 +240,8 @@ func (dw *DatabaseWriter) transactSignedBeaconBlocks() error {
 }
 
 // Upsert to public.blocks.
-func (dw *DatabaseWriter) upsertPublicBlocks(key string, data *[]byte) error {
-	_, err := dw.Tx.Exec(dw.Ctx, UpsertBlocksStmt, key, *data)
+func (dw *DatabaseWriter) upsertPublicBlocks(key string, data []byte) error {
+	_, err := dw.Tx.Exec(dw.Ctx, UpsertBlocksStmt, key, data)
 	if err != nil {
 		loghelper.LogSlotError(dw.DbSlots.Slot, err).Error("Unable to write to the slot to the public.blocks table")
 		return err
@@ -546,6 +545,7 @@ func IsSlotInDb(ctx context.Context, db sql.Database, slot string, blockRoot str
 	var (
 		isInBeaconState       bool
 		isInSignedBeaconBlock bool
+		err                   error
 	)
 	errG, _ := errgroup.WithContext(context.Background())
 	errG.Go(func() error {
@@ -553,7 +553,6 @@ func IsSlotInDb(ctx context.Context, db sql.Database, slot string, blockRoot str
 		case <-ctx.Done():
 			return nil
 		default:
-			var err error
 			isInBeaconState, err = checkSlotAndRoot(db, CheckBeaconStateStmt, slot, stateRoot)
 			if err != nil {
 				loghelper.LogError(err).Error("Unable to check if the slot and stateroot exist in eth_beacon.state")
@@ -566,7 +565,6 @@ func IsSlotInDb(ctx context.Context, db sql.Database, slot string, blockRoot str
 		case <-ctx.Done():
 			return nil
 		default:
-			var err error
 			isInSignedBeaconBlock, err = checkSlotAndRoot(db, CheckSignedBeaconBlockStmt, slot, blockRoot)
 			if err != nil {
 				loghelper.LogError(err).Error("Unable to check if the slot and block_root exist in eth_beacon.signed_block")

pkg/beaconclient/metrics.go

@@ -54,10 +54,6 @@ func CreateBeaconClientMetrics() (*BeaconClientMetrics, error) {
 	if err != nil {
 		return nil, err
 	}
-	err = prometheusRegisterHelper("historic_slots_processed", "Keeps track of the number of historic slots we successfully processed.", &metrics.HistoricSlotProcessed)
-	if err != nil {
-		return nil, err
-	}
 	err = prometheusRegisterHelper("head_error", "Keeps track of the number of errors we had processing head messages.", &metrics.HeadError)
 	if err != nil {
 		return nil, err
@@ -95,7 +91,6 @@ type BeaconClientMetrics struct {
 	KnownGapsInserts        uint64 // Number of known_gaps we successfully wrote to the DB.
 	KnownGapsProcessed      uint64 // Number of knownGaps processed.
 	KnownGapsReprocessError uint64 // Number of knownGaps that were updated with an error.
-	HistoricSlotProcessed   uint64 // Number of historic slots successfully processed.
 	HeadError               uint64 // Number of errors that occurred when decoding the head message.
 	HeadReorgError          uint64 // Number of errors that occurred when decoding the reorg message.
 }
@@ -140,11 +135,6 @@ func (m *BeaconClientMetrics) IncrementReorgError(inc uint64) {
 // Wrapper function to increment the number of knownGaps that were updated with reprocessing errors.
 //If we want to use mutexes later we can easily update all occurrences here.
 func (m *BeaconClientMetrics) IncrementKnownGapsReprocessError(inc uint64) {
+	log.Debug("Incrementing Known Gap Reprocessing: ", &m.KnownGapsReprocessError)
 	atomic.AddUint64(&m.KnownGapsReprocessError, inc)
 }
-
-// Wrapper function to increment the number of historicSlots that were processed successfully.
-// If we want to use mutexes later we can easily update all occurrences here.
-func (m *BeaconClientMetrics) IncrementHistoricSlotProcessed(inc uint64) {
-	atomic.AddUint64(&m.HistoricSlotProcessed, inc)
-}

pkg/beaconclient/processhistoric.go

@@ -81,8 +81,11 @@ func (hp HistoricProcessing) handleProcessingErrors(ctx context.Context, errMess
 }
 
 // "un"-checkout the rows held by this DB in the eth_beacon.historical_process table.
-func (hp HistoricProcessing) releaseDbLocks() error {
+func (hp HistoricProcessing) releaseDbLocks(cancel context.CancelFunc) error {
+	cancel()
 	log.Debug("Updating all the entries to eth_beacon.historical processing")
+	log.Debug("Db: ", hp.db)
+	log.Debug("hp.uniqueNodeIdentifier ", hp.uniqueNodeIdentifier)
 	res, err := hp.db.Exec(context.Background(), releaseHpLockStmt, hp.uniqueNodeIdentifier)
 	if err != nil {
 		return fmt.Errorf("Unable to remove lock from eth_beacon.historical_processing table for node %d, error is %e", hp.uniqueNodeIdentifier, err)
@@ -97,7 +100,7 @@ func (hp HistoricProcessing) releaseDbLocks() error {
 }
 
 // Process the slot range.
-func processSlotRangeWorker(ctx context.Context, workCh <-chan int, errCh chan<- batchHistoricError, db sql.Database, serverAddress string, metrics *BeaconClientMetrics, checkDb bool, incrementTracker func(uint64)) {
+func processSlotRangeWorker(ctx context.Context, workCh <-chan int, errCh chan<- batchHistoricError, db sql.Database, serverAddress string, metrics *BeaconClientMetrics, checkDb bool) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -112,8 +115,6 @@ func processSlotRangeWorker(ctx context.Context, workCh <-chan int, errCh chan<-
 					slot:       slot,
 				}
 				errCh <- errMs
-			} else {
-				incrementTracker(1)
 			}
 		}
 	}
@@ -148,7 +149,7 @@ func getBatchProcessRow(ctx context.Context, db sql.Database, getStartEndSlotStm
 				errCount = append(errCount, err)
 			}
 			if row < 1 {
-				time.Sleep(3 * time.Second)
+				time.Sleep(1000 * time.Millisecond)
 				log.Debug("We are checking rows, be patient")
 				break
 			}
@@ -175,7 +176,7 @@ func getBatchProcessRow(ctx context.Context, db sql.Database, getStartEndSlotStm
 			err = tx.QueryRow(dbCtx, getStartEndSlotStmt).Scan(&sp.startSlot, &sp.endSlot)
 			if err != nil {
 				if err == pgx.ErrNoRows {
-					time.Sleep(1 * time.Second)
+					time.Sleep(100 * time.Millisecond)
 					break
 				}
 				loghelper.LogSlotRangeStatementError(strconv.Itoa(sp.startSlot), strconv.Itoa(sp.endSlot), getStartEndSlotStmt, err).Error("Unable to get a row")
@@ -236,11 +237,8 @@ func removeRowPostProcess(ctx context.Context, db sql.Database, processCh <-chan
 		case slots := <-processCh:
 			// Make sure the start and end slot exist in the slots table.
 			go func() {
-				log.WithFields(log.Fields{
-					"startSlot": slots.startSlot,
-					"endSlot":   slots.endSlot,
-				}).Debug("Starting to check to see if the following slots have been processed")
-				for {
+				finishedProcess := false
+				for !finishedProcess {
 					isStartProcess, err := isSlotProcessed(db, checkProcessedStmt, strconv.Itoa(slots.startSlot))
 					if err != nil {
 						errCh <- err
@@ -250,9 +248,8 @@ func removeRowPostProcess(ctx context.Context, db sql.Database, processCh <-chan
 						errCh <- err
 					}
 					if isStartProcess && isEndProcess {
-						break
+						finishedProcess = true
 					}
-					time.Sleep(3 * time.Second)
 				}
 
 				_, err := db.Exec(context.Background(), removeStmt, strconv.Itoa(slots.startSlot), strconv.Itoa(slots.endSlot))

pkg/beaconclient/processknowngaps.go

@@ -61,7 +61,7 @@ type KnownGapsProcessing struct {
 func (bc *BeaconClient) ProcessKnownGaps(ctx context.Context, maxWorkers int) []error {
 	log.Info("We are starting the known gaps processing service.")
 	bc.KnownGapsProcess = KnownGapsProcessing{db: bc.Db, uniqueNodeIdentifier: bc.UniqueNodeIdentifier, metrics: bc.Metrics}
-	errs := handleBatchProcess(ctx, maxWorkers, bc.KnownGapsProcess, bc.KnownGapsProcess.db, bc.ServerEndpoint, bc.Metrics, bc.CheckDb, bc.Metrics.IncrementKnownGapsProcessed)
+	errs := handleBatchProcess(ctx, maxWorkers, bc.KnownGapsProcess, bc.KnownGapsProcess.db, bc.ServerEndpoint, bc.Metrics, bc.CheckDb)
 	log.Debug("Exiting known gaps processing service")
 	return errs
 }
@@ -69,8 +69,7 @@ func (bc *BeaconClient) ProcessKnownGaps(ctx context.Context, maxWorkers int) []
 // This function will perform all the necessary clean up tasks for stopping historical processing.
 func (bc *BeaconClient) StopKnownGapsProcessing(cancel context.CancelFunc) error {
 	log.Info("We are stopping the known gaps processing service.")
-	cancel()
-	err := bc.KnownGapsProcess.releaseDbLocks()
+	err := bc.KnownGapsProcess.releaseDbLocks(cancel)
 	if err != nil {
 		loghelper.LogError(err).WithField("uniqueIdentifier", bc.UniqueNodeIdentifier).Error("We were unable to remove the locks from the eth_beacon.known_gaps table. Manual Intervention is needed!")
 	}
@@ -122,8 +121,11 @@ func (kgp KnownGapsProcessing) handleProcessingErrors(ctx context.Context, errMe
 }
 
 // Updated checked_out column for the uniqueNodeIdentifier.
-func (kgp KnownGapsProcessing) releaseDbLocks() error {
+func (kgp KnownGapsProcessing) releaseDbLocks(cancel context.CancelFunc) error {
+	cancel()
 	log.Debug("Updating all the entries to eth_beacon.known_gaps")
+	log.Debug("Db: ", kgp.db)
+	log.Debug("kgp.uniqueNodeIdentifier ", kgp.uniqueNodeIdentifier)
 	res, err := kgp.db.Exec(context.Background(), releaseKgLockStmt, kgp.uniqueNodeIdentifier)
 	if err != nil {
 		return err

pkg/beaconclient/processslot.go

@@ -25,7 +25,6 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/jackc/pgx/v4"
 	si "github.com/prysmaticlabs/prysm/consensus-types/interfaces"
@@ -51,24 +50,23 @@ var (
 type ProcessSlot struct {
 	// Generic
 
-	Slot               int                  // The slot number.
-	Epoch              int                  // The epoch number.
-	BlockRoot          string               // The hex encoded string of the BlockRoot.
-	StateRoot          string               // The hex encoded string of the StateRoot.
-	ParentBlockRoot    string               // The hex encoded string of the parent block.
-	Status             string               // The status of the block
-	HeadOrHistoric     string               // Is this the head or a historic slot. This is critical when trying to analyze errors and skipped slots.
-	Db                 sql.Database         // The DB object used to write to the DB.
-	Metrics            *BeaconClientMetrics // An object to keep track of the beaconclient metrics
-	PerformanceMetrics PerformanceMetrics   // An object to keep track of performance metrics.
+	Slot            int                  // The slot number.
+	Epoch           int                  // The epoch number.
+	BlockRoot       string               // The hex encoded string of the BlockRoot.
+	StateRoot       string               // The hex encoded string of the StateRoot.
+	ParentBlockRoot string               // The hex encoded string of the parent block.
+	Status          string               // The status of the block
+	HeadOrHistoric  string               // Is this the head or a historic slot. This is critical when trying to analyze errors and skipped slots.
+	Db              sql.Database         // The DB object used to write to the DB.
+	Metrics         *BeaconClientMetrics // An object to keep track of the beaconclient 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.
@@ -76,19 +74,6 @@ type ProcessSlot struct {
 	DbBeaconState            *DbBeaconState       // The model being written to the state table.
 }
 
-type PerformanceMetrics struct {
-	BeaconNodeBlockRetrievalTime time.Duration // How long it took to get the BeaconBlock from the Beacon Node.
-	BeaconNodeStateRetrievalTime time.Duration // How long it took to get the BeaconState from the Beacon Node.
-	ParseBeaconObjectForHash     time.Duration // How long it took to get some information from the beacon objects.
-	CheckDbPreProcessing         time.Duration // How long it takes to check the DB before processing a block.
-	CreateDbWriteObject          time.Duration // How long it takes to create a DB write object.
-	TransactSlotOnly             time.Duration // How long it takes to transact the slot information only.
-	CheckReorg                   time.Duration // How long it takes to check for Reorgs
-	CommitTransaction            time.Duration // How long it takes to commit the final transaction.
-	TotalDbTransaction           time.Duration // How long it takes from start to committing the entire DB transaction.
-	TotalProcessing              time.Duration // How long it took to process the entire slot.
-}
-
 // This function will do all the work to process the slot and write it to the DB.
 // It will return the error and error process. The error process is used for providing reach detail to the
 // known_gaps table.
@@ -97,7 +82,6 @@ func processFullSlot(ctx context.Context, db sql.Database, serverAddress string,
 	case <-ctx.Done():
 		return nil, ""
 	default:
-		totalStart := time.Now()
 		ps := &ProcessSlot{
 			Slot:           slot,
 			BlockRoot:      blockRoot,
@@ -105,18 +89,6 @@ func processFullSlot(ctx context.Context, db sql.Database, serverAddress string,
 			HeadOrHistoric: headOrHistoric,
 			Db:             db,
 			Metrics:        metrics,
-			PerformanceMetrics: PerformanceMetrics{
-				BeaconNodeBlockRetrievalTime: 0,
-				BeaconNodeStateRetrievalTime: 0,
-				ParseBeaconObjectForHash:     0,
-				CheckDbPreProcessing:         0,
-				CreateDbWriteObject:          0,
-				TransactSlotOnly:             0,
-				CheckReorg:                   0,
-				CommitTransaction:            0,
-				TotalDbTransaction:           0,
-				TotalProcessing:              0,
-			},
 		}
 
 		g, _ := errgroup.WithContext(context.Background())
@@ -128,12 +100,10 @@ func processFullSlot(ctx context.Context, db sql.Database, serverAddress string,
 			case <-ctx.Done():
 				return nil
 			default:
-				start := time.Now()
 				err := ps.getBeaconState(serverAddress, vUnmarshalerCh)
 				if err != nil {
 					return err
 				}
-				ps.PerformanceMetrics.BeaconNodeStateRetrievalTime = time.Since(start)
 				return nil
 			}
 		})
@@ -144,34 +114,23 @@ func processFullSlot(ctx context.Context, db sql.Database, serverAddress string,
 			case <-ctx.Done():
 				return nil
 			default:
-				start := time.Now()
 				err := ps.getSignedBeaconBlock(serverAddress, vUnmarshalerCh)
 				if err != nil {
 					return err
 				}
-				ps.PerformanceMetrics.BeaconNodeBlockRetrievalTime = time.Since(start)
 				return nil
 			}
 		})
 
 		if err := g.Wait(); err != nil {
-			// Make sure channel is empty.
-			select {
-			case <-vUnmarshalerCh:
-			default:
-			}
 			return err, "processSlot"
 		}
 
-		parseBeaconTime := time.Now()
 		finalBlockRoot, finalStateRoot, finalEth1BlockHash, err := ps.provideFinalHash()
 		if err != nil {
 			return err, "CalculateBlockRoot"
 		}
-		ps.PerformanceMetrics.ParseBeaconObjectForHash = time.Since(parseBeaconTime)
-
 		if checkDb {
-			checkDbTime := time.Now()
 			inDb, err := IsSlotInDb(ctx, ps.Db, strconv.Itoa(ps.Slot), finalBlockRoot, finalStateRoot)
 			if err != nil {
 				return err, "checkDb"
@@ -180,35 +139,26 @@ func processFullSlot(ctx context.Context, db sql.Database, serverAddress string,
 				log.WithField("slot", slot).Info("Slot already in the DB.")
 				return nil, ""
 			}
-			ps.PerformanceMetrics.CheckDbPreProcessing = time.Since(checkDbTime)
 		}
 
 		// Get this object ready to write
-		createDbWriteTime := time.Now()
 		dw, err := ps.createWriteObjects(finalBlockRoot, finalStateRoot, finalEth1BlockHash)
 		if err != nil {
 			return err, "blockRoot"
 		}
-		ps.PerformanceMetrics.CreateDbWriteObject = time.Since(createDbWriteTime)
-
 		// Write the object to the DB.
-		dbFullTransactionTime := time.Now()
 		defer func() {
 			err := dw.Tx.Rollback(dw.Ctx)
 			if err != nil && err != pgx.ErrTxClosed {
 				loghelper.LogError(err).Error("We were unable to Rollback a transaction")
 			}
 		}()
-
-		transactionTime := time.Now()
 		err = dw.transactFullSlot()
 		if err != nil {
 			return err, "processSlot"
 		}
-		ps.PerformanceMetrics.TransactSlotOnly = time.Since(transactionTime)
 
 		// Handle any reorgs or skipped slots.
-		reorgTime := time.Now()
 		headOrHistoric = strings.ToLower(headOrHistoric)
 		if headOrHistoric != "head" && headOrHistoric != "historic" {
 			return fmt.Errorf("headOrHistoric must be either historic or head!"), ""
@@ -216,23 +166,11 @@ func processFullSlot(ctx context.Context, db sql.Database, serverAddress string,
 		if ps.HeadOrHistoric == "head" && previousSlot != 0 && previousBlockRoot != "" && ps.Status != "skipped" {
 			ps.checkPreviousSlot(dw.Tx, dw.Ctx, previousSlot, previousBlockRoot, knownGapsTableIncrement)
 		}
-		ps.PerformanceMetrics.CheckReorg = time.Since(reorgTime)
 
 		// Commit the transaction
-		commitTime := time.Now()
 		if err = dw.Tx.Commit(dw.Ctx); err != nil {
 			return err, "transactionCommit"
 		}
-		ps.PerformanceMetrics.CommitTransaction = time.Since(commitTime)
-
-		// Total metric capture time.
-		ps.PerformanceMetrics.TotalDbTransaction = time.Since(dbFullTransactionTime)
-		ps.PerformanceMetrics.TotalProcessing = time.Since(totalStart)
-
-		log.WithFields(log.Fields{
-			"slot":               slot,
-			"performanceMetrics": fmt.Sprintf("%+v\n", ps.PerformanceMetrics),
-		}).Debug("Performance Metric output!")
 
 		return nil, ""
 	}
@@ -275,7 +213,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
@@ -285,7 +223,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
@@ -305,14 +243,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
@@ -323,24 +261,23 @@ func (ps *ProcessSlot) getBeaconState(serverEndpoint string, vmCh chan<- *dt.Ver
 // Check to make sure that the previous block we processed is the parent of the current block.
 func (ps *ProcessSlot) checkPreviousSlot(tx sql.Tx, ctx context.Context, previousSlot int, previousBlockRoot string, knownGapsTableIncrement int) {
 	parentRoot := "0x" + hex.EncodeToString(ps.FullSignedBeaconBlock.Block().ParentRoot())
-	slot := int(ps.FullBeaconState.Slot())
-	if previousSlot == slot {
+	if previousSlot == int(ps.FullBeaconState.Slot()) {
 		log.WithFields(log.Fields{
-			"slot": slot,
+			"slot": ps.FullBeaconState.Slot(),
 			"fork": true,
 		}).Warn("A fork occurred! The previous slot and current slot match.")
 		transactReorgs(tx, ctx, strconv.Itoa(ps.Slot), ps.BlockRoot, ps.Metrics)
-	} else if previousSlot > slot {
+	} else if previousSlot > int(ps.FullBeaconState.Slot()) {
 		log.WithFields(log.Fields{
 			"previousSlot": previousSlot,
-			"curSlot":      slot,
+			"curSlot":      int(ps.FullBeaconState.Slot()),
 		}).Warn("We noticed the previous slot is greater than the current slot.")
-	} else if previousSlot+1 != slot {
+	} else if previousSlot+1 != int(ps.FullBeaconState.Slot()) {
 		log.WithFields(log.Fields{
 			"previousSlot": previousSlot,
-			"currentSlot":  slot,
+			"currentSlot":  ps.FullBeaconState.Slot(),
 		}).Error("We skipped a few slots.")
-		transactKnownGaps(tx, ctx, knownGapsTableIncrement, previousSlot+1, slot-1, fmt.Errorf("Gaps during head processing"), "headGaps", ps.Metrics)
+		transactKnownGaps(tx, ctx, knownGapsTableIncrement, previousSlot+1, int(ps.FullBeaconState.Slot())-1, fmt.Errorf("Gaps during head processing"), "headGaps", ps.Metrics)
 	} else if previousBlockRoot != parentRoot {
 		log.WithFields(log.Fields{
 			"previousBlockRoot":  previousBlockRoot,

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