Revert "Add a new slot ticker and use it on attestation aggregation (#…

…12377)"

This reverts commit f6764fe.

beacon-chain/node/node.go

@@ -581,8 +581,7 @@ func (b *BeaconNode) fetchBuilderService() *builder.Service {
 
 func (b *BeaconNode) registerAttestationPool() error {
 	s, err := attestations.NewService(b.ctx, &attestations.Config{
-		Pool:                b.attestationPool,
-		InitialSyncComplete: b.initialSyncComplete,
+		Pool: b.attestationPool,
 	})
 	if err != nil {
 		return errors.Wrap(err, "could not register atts pool service")

beacon-chain/operations/attestations/BUILD.bazel

@@ -18,7 +18,6 @@ go_library(
     deps = [
         "//beacon-chain/operations/attestations/kv:go_default_library",
         "//cache/lru:go_default_library",
-        "//config/features:go_default_library",
         "//config/params:go_default_library",
         "//consensus-types/primitives:go_default_library",
         "//crypto/hash:go_default_library",

beacon-chain/operations/attestations/prepare_forkchoice.go

@@ -7,39 +7,27 @@ import (
 	"time"
 
 	"github.com/prysmaticlabs/go-bitfield"
-	"github.com/prysmaticlabs/prysm/v4/config/features"
-	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/crypto/hash"
 	ethpb "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1"
 	attaggregation "github.com/prysmaticlabs/prysm/v4/proto/prysm/v1alpha1/attestation/aggregation/attestations"
 	"github.com/prysmaticlabs/prysm/v4/time/slots"
 	"go.opencensus.io/trace"
 )
 
+// Prepare attestations for fork choice three times per slot.
+var prepareForkChoiceAttsPeriod = slots.DivideSlotBy(3 /* times-per-slot */)
+
 // This prepares fork choice attestations by running batchForkChoiceAtts
 // every prepareForkChoiceAttsPeriod.
 func (s *Service) prepareForkChoiceAtts() {
-	intervals := features.Get().AggregateIntervals
-	slotDuration := time.Duration(params.BeaconConfig().SecondsPerSlot) * time.Second
-	// Adjust intervals for networks with a lower slot duration (Hive, e2e, etc)
-	for {
-		if intervals[len(intervals)-1] >= slotDuration {
-			for i, offset := range intervals {
-				intervals[i] = offset / 2
-			}
-		} else {
-			break
-		}
-	}
-	ticker := slots.NewSlotTickerWithIntervals(time.Unix(int64(s.genesisTime), 0), intervals)
+	ticker := time.NewTicker(prepareForkChoiceAttsPeriod)
+	defer ticker.Stop()
 	for {
 		select {
-		case <-ticker.C():
-			t := time.Now()
+		case <-ticker.C:
 			if err := s.batchForkChoiceAtts(s.ctx); err != nil {
 				log.WithError(err).Error("Could not prepare attestations for fork choice")
 			}
-			log.WithField("latency", time.Since(t).Milliseconds()).Debug("batched forkchoice attestations")
 		case <-s.ctx.Done():
 			log.Debug("Context closed, exiting routine")
 			return

beacon-chain/operations/attestations/service.go

@@ -5,7 +5,6 @@ package attestations
 
 import (
 	"context"
-	"errors"
 	"time"
 
 	lru "github.com/hashicorp/golang-lru"
@@ -27,9 +26,8 @@ type Service struct {
 
 // Config options for the service.
 type Config struct {
-	Pool                Pool
-	pruneInterval       time.Duration
-	InitialSyncComplete chan struct{}
+	Pool          Pool
+	pruneInterval time.Duration
 }
 
 // NewService instantiates a new attestation pool service instance that will
@@ -53,24 +51,10 @@ func NewService(ctx context.Context, cfg *Config) (*Service, error) {
 
 // Start an attestation pool service's main event loop.
 func (s *Service) Start() {
-	if err := s.waitForSync(s.cfg.InitialSyncComplete); err != nil {
-		log.WithError(err).Error("failed to wait for initial sync")
-		return
-	}
 	go s.prepareForkChoiceAtts()
 	go s.pruneAttsPool()
 }
 
-// waitForSync waits until the beacon node is synced to the latest head.
-func (s *Service) waitForSync(syncChan chan struct{}) error {
-	select {
-	case <-syncChan:
-		return nil
-	case <-s.ctx.Done():
-		return errors.New("context closed, exiting goroutine")
-	}
-}
-
 // Stop the beacon block attestation pool service's main event loop
 // and associated goroutines.
 func (s *Service) Stop() error {

config/features/config.go

@@ -72,9 +72,6 @@ type Flags struct {
 	// KeystoreImportDebounceInterval specifies the time duration the validator waits to reload new keys if they have
 	// changed on disk. This feature is for advanced use cases only.
 	KeystoreImportDebounceInterval time.Duration
-
-	// AggregateIntervals specifies the time durations at which we aggregate attestations preparing for forkchoice.
-	AggregateIntervals []time.Duration
 }
 
 var featureConfig *Flags
@@ -221,7 +218,6 @@ func ConfigureBeaconChain(ctx *cli.Context) error {
 		logEnabled(buildBlockParallel)
 		cfg.BuildBlockParallel = true
 	}
-	cfg.AggregateIntervals = []time.Duration{aggregateFirstInterval.Value, aggregateSecondInterval.Value, aggregateThirdInterval.Value}
 	Init(cfg)
 	return nil
 }

config/features/flags.go

@@ -50,24 +50,6 @@ var (
 		Usage: "(Danger): Writes the wallet password to the wallet directory on completing Prysm web onboarding. " +
 			"We recommend against this flag unless you are an advanced user.",
 	}
-	aggregateFirstInterval = &cli.DurationFlag{
-		Name:   "aggregate-first-interval",
-		Usage:  "(Advanced): Specifies the first interval in which attestations are aggregated in the slot (typically unnaggregated attestations are aggregated in this interval)",
-		Value:  7 * time.Second,
-		Hidden: true,
-	}
-	aggregateSecondInterval = &cli.DurationFlag{
-		Name:   "aggregate-second-interval",
-		Usage:  "(Advanced): Specifies the second interval in which attestations are aggregated in the slot",
-		Value:  9 * time.Second,
-		Hidden: true,
-	}
-	aggregateThirdInterval = &cli.DurationFlag{
-		Name:   "aggregate-third-interval",
-		Usage:  "(Advanced): Specifies the third interval in which attestations are aggregated in the slot",
-		Value:  11 * time.Second,
-		Hidden: true,
-	}
 	dynamicKeyReloadDebounceInterval = &cli.DurationFlag{
 		Name: "dynamic-key-reload-debounce-interval",
 		Usage: "(Advanced): Specifies the time duration the validator waits to reload new keys if they have " +
@@ -186,9 +168,6 @@ var BeaconChainFlags = append(deprecatedBeaconFlags, append(deprecatedFlags, []c
 	enableVerboseSigVerification,
 	enableOptionalEngineMethods,
 	prepareAllPayloads,
-	aggregateFirstInterval,
-	aggregateSecondInterval,
-	aggregateThirdInterval,
 	buildBlockParallel,
 }...)...)
 

time/slots/BUILD.bazel

@@ -37,6 +37,5 @@ go_test(
         "//time:go_default_library",
         "@com_github_sirupsen_logrus//:go_default_library",
         "@com_github_sirupsen_logrus//hooks/test:go_default_library",
-        "@com_github_stretchr_testify//require:go_default_library",
     ],
 )

time/slots/slotticker.go

@@ -4,7 +4,6 @@ package slots
 import (
 	"time"
 
-	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
 	prysmTime "github.com/prysmaticlabs/prysm/v4/time"
 )
@@ -105,65 +104,3 @@ func (s *SlotTicker) start(
 		}
 	}()
 }
-
-// startWithIntervals starts a ticker that emits a tick every slot at the
-// prescribed intervals. The caller is responsible to make these intervals increasing and
-// less than secondsPerSlot
-func (s *SlotTicker) startWithIntervals(
-	genesisTime time.Time,
-	until func(time.Time) time.Duration,
-	after func(time.Duration) <-chan time.Time,
-	intervals []time.Duration) {
-	go func() {
-		slot := Since(genesisTime)
-		slot++
-		interval := 0
-		nextTickTime := startFromTime(genesisTime, slot).Add(intervals[0])
-
-		for {
-			waitTime := until(nextTickTime)
-			select {
-			case <-after(waitTime):
-				s.c <- slot
-				interval++
-				if interval == len(intervals) {
-					interval = 0
-					slot++
-				}
-				nextTickTime = startFromTime(genesisTime, slot).Add(intervals[interval])
-			case <-s.done:
-				return
-			}
-		}
-	}()
-}
-
-// NewSlotTickerWithIntervals starts and returns a SlotTicker instance that allows
-// several offsets of time from genesis,
-// Caller is responsible to input the intervals in increasing order and none bigger or equal than
-// SecondsPerSlot
-func NewSlotTickerWithIntervals(genesisTime time.Time, intervals []time.Duration) *SlotTicker {
-	if genesisTime.Unix() == 0 {
-		panic("zero genesis time")
-	}
-	if len(intervals) == 0 {
-		panic("at least one interval has to be entered")
-	}
-	slotDuration := time.Duration(params.BeaconConfig().SecondsPerSlot) * time.Second
-	lastOffset := time.Duration(0)
-	for _, offset := range intervals {
-		if offset < lastOffset {
-			panic("invalid decreasing offsets")
-		}
-		if offset >= slotDuration {
-			panic("invalid ticker offset")
-		}
-		lastOffset = offset
-	}
-	ticker := &SlotTicker{
-		c:    make(chan primitives.Slot),
-		done: make(chan struct{}),
-	}
-	ticker.startWithIntervals(genesisTime, prysmTime.Until, time.After, intervals)
-	return ticker
-}

time/slots/slotticker_test.go

@@ -4,9 +4,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/prysmaticlabs/prysm/v4/config/params"
 	"github.com/prysmaticlabs/prysm/v4/consensus-types/primitives"
-	"github.com/stretchr/testify/require"
 )
 
 var _ Ticker = (*SlotTicker)(nil)
@@ -138,49 +136,3 @@ func TestGetSlotTickerWithOffset_OK(t *testing.T) {
 		}
 	}
 }
-
-func TestGetSlotTickerWitIntervals(t *testing.T) {
-	genesisTime := time.Now()
-	offset := time.Duration(params.BeaconConfig().SecondsPerSlot) * time.Second / 3
-	intervals := []time.Duration{offset, 2 * offset}
-
-	intervalTicker := NewSlotTickerWithIntervals(genesisTime, intervals)
-	normalTicker := NewSlotTicker(genesisTime, params.BeaconConfig().SecondsPerSlot)
-
-	firstTicked := 0
-	for {
-		select {
-		case <-intervalTicker.C():
-			// interval ticks starts in second slot
-			if firstTicked < 2 {
-				t.Fatal("Expected other ticker to tick first")
-			}
-			return
-		case <-normalTicker.C():
-			if firstTicked > 1 {
-				t.Fatal("Expected normal ticker to tick first")
-			}
-			firstTicked++
-		}
-	}
-}
-
-func TestSlotTickerWithIntervalsInputValidation(t *testing.T) {
-	var genesisTime time.Time
-	offset := time.Duration(params.BeaconConfig().SecondsPerSlot) * time.Second / 3
-	intervals := make([]time.Duration, 0)
-	panicCall := func() {
-		NewSlotTickerWithIntervals(genesisTime, intervals)
-	}
-	require.Panics(t, panicCall, "zero genesis time")
-	genesisTime = time.Now()
-	require.Panics(t, panicCall, "at least one interval has to be entered")
-	intervals = []time.Duration{2 * offset, offset}
-	require.Panics(t, panicCall, "invalid decreasing offsets")
-	intervals = []time.Duration{offset, 4 * offset}
-	require.Panics(t, panicCall, "invalid ticker offset")
-	intervals = []time.Duration{4 * offset, offset}
-	require.Panics(t, panicCall, "invalid ticker offset")
-	intervals = []time.Duration{offset, 2 * offset}
-	require.NotPanics(t, panicCall)
-}

time/slots/slottime.go

@@ -16,17 +16,12 @@ import (
 // incoming objects. (24 mins with mainnet spec)
 const MaxSlotBuffer = uint64(1 << 7)
 
-// startFromTime returns the slot start in terms of genesis time.Time
-func startFromTime(genesis time.Time, slot primitives.Slot) time.Time {
-	duration := time.Second * time.Duration(slot.Mul(params.BeaconConfig().SecondsPerSlot))
-	return genesis.Add(duration) // lint:ignore uintcast -- Genesis timestamp will not exceed int64 in your lifetime.
-}
-
 // StartTime returns the start time in terms of its unix epoch
 // value.
 func StartTime(genesis uint64, slot primitives.Slot) time.Time {
-	genesisTime := time.Unix(int64(genesis), 0) // lint:ignore uintcast -- Genesis timestamp will not exceed int64 in your lifetime.
-	return startFromTime(genesisTime, slot)
+	duration := time.Second * time.Duration(slot.Mul(params.BeaconConfig().SecondsPerSlot))
+	startTime := time.Unix(int64(genesis), 0).Add(duration) // lint:ignore uintcast -- Genesis timestamp will not exceed int64 in your lifetime.
+	return startTime
 }
 
 // SinceGenesis returns the number of slots since