fix(cannon): Check sync status before deriving events (#182)

ethpandaops · Sep 6, 2023 · 2ceac41 · 2ceac41
1 parent fd241f0
commit 2ceac41
Show file tree

Hide file tree

Showing 8 changed files with 27 additions and 5 deletions.
diff --git a/pkg/cannon/cannon.go b/pkg/cannon/cannon.go
@@ -247,10 +247,6 @@ func (c *Cannon) syncClockDrift(ctx context.Context) error {
 }
 
 func (c *Cannon) handleNewDecoratedEvent(ctx context.Context, event *xatu.DecoratedEvent) error {
- if err := c.beacon.Synced(ctx); err != nil {
- return err
- }
-
  for _, sink := range c.sinks {
  if err := sink.HandleNewDecoratedEvent(ctx, event); err != nil {
  c.log.

diff --git a/pkg/cannon/deriver/beacon/eth/v2/attester_slashing.go b/pkg/cannon/deriver/beacon/eth/v2/attester_slashing.go
@@ -95,6 +95,10 @@ func (a *AttesterSlashingDeriver) run(ctx context.Context) {
  operation := func() error {
  time.Sleep(100 * time.Millisecond)
 
+ if err := a.beacon.Synced(ctx); err != nil {
+ return err
+ }
+
  // Get the next slot
  location, err := a.iterator.Next(ctx)
  if err != nil {

diff --git a/pkg/cannon/deriver/beacon/eth/v2/bls_to_execution_change.go b/pkg/cannon/deriver/beacon/eth/v2/bls_to_execution_change.go
@@ -98,6 +98,10 @@ func (b *BLSToExecutionChangeDeriver) run(ctx context.Context) {
  operation := func() error {
  time.Sleep(100 * time.Millisecond)
 
+ if err := b.beacon.Synced(ctx); err != nil {
+ return err
+ }
+
  // Get the next slot
  location, err := b.iterator.Next(ctx)
  if err != nil {

diff --git a/pkg/cannon/deriver/beacon/eth/v2/deposit.go b/pkg/cannon/deriver/beacon/eth/v2/deposit.go
@@ -96,6 +96,10 @@ func (b *DepositDeriver) run(ctx context.Context) {
  operation := func() error {
  time.Sleep(100 * time.Millisecond)
 
+ if err := b.beacon.Synced(ctx); err != nil {
+ return err
+ }
+
  // Get the next slot
  location, err := b.iterator.Next(ctx)
  if err != nil {

diff --git a/pkg/cannon/deriver/beacon/eth/v2/execution_transaction.go b/pkg/cannon/deriver/beacon/eth/v2/execution_transaction.go
@@ -99,6 +99,10 @@ func (b *ExecutionTransactionDeriver) run(ctx context.Context) {
  operation := func() error {
  time.Sleep(100 * time.Millisecond)
 
+ if err := b.beacon.Synced(ctx); err != nil {
+ return err
+ }
+
  // Get the next slot
  location, err := b.iterator.Next(ctx)
  if err != nil {

diff --git a/pkg/cannon/deriver/beacon/eth/v2/proposer_slashing.go b/pkg/cannon/deriver/beacon/eth/v2/proposer_slashing.go
@@ -95,6 +95,10 @@ func (b *ProposerSlashingDeriver) run(ctx context.Context) {
  operation := func() error {
  time.Sleep(100 * time.Millisecond)
 
+ if err := b.beacon.Synced(ctx); err != nil {
+ return err
+ }
+
  // Get the next slot
  location, err := b.iterator.Next(ctx)
  if err != nil {

diff --git a/pkg/cannon/deriver/beacon/eth/v2/voluntary_exit.go b/pkg/cannon/deriver/beacon/eth/v2/voluntary_exit.go
@@ -96,6 +96,10 @@ func (b *VoluntaryExitDeriver) run(ctx context.Context) {
  operation := func() error {
  time.Sleep(100 * time.Millisecond)
 
+ if err := b.beacon.Synced(ctx); err != nil {
+ return err
+ }
+
  // Get the next slot
  location, err := b.iterator.Next(ctx)
  if err != nil {

diff --git a/pkg/cannon/ethereum/beacon.go b/pkg/cannon/ethereum/beacon.go
@@ -33,12 +33,14 @@ type BeaconNode struct {
 func NewBeaconNode(ctx context.Context, name string, config *Config, log logrus.FieldLogger) (*BeaconNode, error) {
  opts := *beacon.
  DefaultOptions().
- EnableDefaultBeaconSubscription().
+ DisableEmptySlotDetection().
  DisablePrometheusMetrics()
 
  opts.HealthCheck.Interval.Duration = time.Second * 3
  opts.HealthCheck.SuccessfulResponses = 1
 
+ opts.BeaconSubscription.Enabled = false
+
  node := beacon.NewNode(log, &beacon.Config{
  Name: name,
  Addr: config.BeaconNodeAddress,