diff --git a/pkg/beacon/beacon.go b/pkg/beacon/beacon.go index 0b1eb83..e9ec6bd 100644 --- a/pkg/beacon/beacon.go +++ b/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 { diff --git a/pkg/beacon/empty_slot_test.go b/pkg/beacon/empty_slot_test.go new file mode 100644 index 0000000..29113b3 --- /dev/null +++ b/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. + } +}