Fix empty slot detection never firing

pkg/beacon/beacon.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"strings"
 	"sync"
 	"time"
 
@@ -423,22 +422,7 @@ func (n *node) subscribeDownstream(ctx context.Context) error {
 	})
 
 	n.wallclock.OnSlotChanged(func(slot ethwallclock.Slot) {
-		if !n.options.DetectEmptySlots {
-			return
-		}
-
-		if n.stat.Syncing() {
-			return
-		}
-
-		_, err := n.FetchBlock(ctx, fmt.Sprintf("%v", slot.Number()-1))
-		if err != nil {
-			if strings.Contains(err.Error(), "404") {
-				n.publishEmptySlot(ctx, phase0.Slot(slot.Number()))
-			}
-
-			return
-		}
+		n.checkEmptySlot(ctx, slot)
 	})
 
 	n.OnFinalizedCheckpoint(ctx, func(ctx context.Context, ev *v1.FinalizedCheckpointEvent) error {
@@ -454,6 +438,29 @@ func (n *node) subscribeDownstream(ctx context.Context) error {
 	return nil
 }
 
+// checkEmptySlot fetches the block for the previous slot and publishes an
+// empty slot event if the beacon node has no block for it. FetchBlock
+// returns a nil block with a nil error on a 404, so a missing block is
+// detected by checking the returned block rather than the error.
+func (n *node) checkEmptySlot(ctx context.Context, slot ethwallclock.Slot) {
+	if !n.options.DetectEmptySlots {
+		return
+	}
+
+	if n.stat.Syncing() {
+		return
+	}
+
+	block, err := n.FetchBlock(ctx, fmt.Sprintf("%v", slot.Number()-1))
+	if err != nil {
+		return
+	}
+
+	if block == nil {
+		n.publishEmptySlot(ctx, phase0.Slot(slot.Number()))
+	}
+}
+
 func (n *node) fetchIsHealthy(ctx context.Context) error {
 	provider, isProvider := n.client.(eth2client.NodeSyncingProvider)
 	if !isProvider {

pkg/beacon/empty_slot_test.go

@@ -0,0 +1,119 @@
+package beacon
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/chuckpreslar/emission"
+	"github.com/ethpandaops/ethwallclock"
+	eapi "github.com/ethpandaops/go-eth2-client/api"
+	"github.com/ethpandaops/go-eth2-client/spec"
+	"github.com/ethpandaops/go-eth2-client/spec/all"
+	"github.com/sirupsen/logrus"
+)
+
+type emptySlotFakeClient struct {
+	blockErr error
+	block    *spec.VersionedSignedBeaconBlock
+}
+
+func (f *emptySlotFakeClient) Name() string    { return "fake" }
+func (f *emptySlotFakeClient) Address() string { return "fake://" }
+func (f *emptySlotFakeClient) IsActive() bool  { return true }
+func (f *emptySlotFakeClient) IsSynced() bool  { return true }
+
+func (f *emptySlotFakeClient) SignedBeaconBlock(
+	_ context.Context, _ *eapi.SignedBeaconBlockOpts,
+) (*eapi.Response[*spec.VersionedSignedBeaconBlock], error) {
+	if f.blockErr != nil {
+		return nil, f.blockErr
+	}
+
+	return &eapi.Response[*spec.VersionedSignedBeaconBlock]{Data: f.block}, nil
+}
+
+func (f *emptySlotFakeClient) AgnosticSignedBeaconBlock(
+	_ context.Context, _ *eapi.SignedBeaconBlockOpts,
+) (*eapi.Response[*all.SignedBeaconBlock], error) {
+	return nil, f.blockErr
+}
+
+func newEmptySlotTestNode(c *emptySlotFakeClient) *node {
+	log := logrus.New()
+	log.SetLevel(logrus.PanicLevel)
+
+	return &node{
+		log:     log,
+		options: &Options{DetectEmptySlots: true},
+		stat:    NewStatus(1, 1),
+		broker:  emission.NewEmitter(),
+		client:  c,
+	}
+}
+
+func TestCheckEmptySlot_PublishesOnMissingBlock(t *testing.T) {
+	n := newEmptySlotTestNode(&emptySlotFakeClient{
+		blockErr: &eapi.Error{Method: "GET", Endpoint: "/eth/v2/beacon/blocks/41", StatusCode: 404},
+	})
+
+	fired := make(chan *EmptySlotEvent, 1)
+	n.OnEmptySlot(context.Background(), func(_ context.Context, event *EmptySlotEvent) error {
+		fired <- event
+		return nil
+	})
+
+	n.checkEmptySlot(context.Background(), ethwallclock.NewSlot(42, time.Now(), time.Now()))
+
+	select {
+	case event := <-fired:
+		if event.Slot != 42 {
+			t.Fatalf("expected empty slot event for slot 42, got %d", event.Slot)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("expected an empty slot event to fire on a missing block, none did")
+	}
+}
+
+func TestCheckEmptySlot_DoesNotPublishWhenBlockExists(t *testing.T) {
+	n := newEmptySlotTestNode(&emptySlotFakeClient{
+		block: &spec.VersionedSignedBeaconBlock{},
+	})
+
+	fired := make(chan *EmptySlotEvent, 1)
+	n.OnEmptySlot(context.Background(), func(_ context.Context, event *EmptySlotEvent) error {
+		fired <- event
+		return nil
+	})
+
+	n.checkEmptySlot(context.Background(), ethwallclock.NewSlot(42, time.Now(), time.Now()))
+
+	select {
+	case event := <-fired:
+		t.Fatalf("expected no empty slot event when a block exists, got one for slot %d", event.Slot)
+	case <-time.After(200 * time.Millisecond):
+		// Good, nothing fired.
+	}
+}
+
+func TestCheckEmptySlot_DisabledByDefault(t *testing.T) {
+	n := newEmptySlotTestNode(&emptySlotFakeClient{
+		blockErr: &eapi.Error{Method: "GET", Endpoint: "/eth/v2/beacon/blocks/41", StatusCode: 404},
+	})
+	n.options.DetectEmptySlots = false
+
+	fired := make(chan *EmptySlotEvent, 1)
+	n.OnEmptySlot(context.Background(), func(_ context.Context, event *EmptySlotEvent) error {
+		fired <- event
+		return nil
+	})
+
+	n.checkEmptySlot(context.Background(), ethwallclock.NewSlot(42, time.Now(), time.Now()))
+
+	select {
+	case event := <-fired:
+		t.Fatalf("expected no empty slot event when detection is disabled, got one for slot %d", event.Slot)
+	case <-time.After(200 * time.Millisecond):
+		// Good, nothing fired.
+	}
+}