Add a new slot ticker and use it on attestation aggregation (#12377)

* Add slot ticker with intervals

* add flags for aggregation duration

* misspelling

* hide flags

* fix flags and default durations

* lint

* wait for initial sync

* deep source

* add log

* Preston's review

* fix error message

---------

Co-authored-by: prylabs-bulldozer[bot] <58059840+prylabs-bulldozer[bot]@users.noreply.github.com>

time/slots/BUILD.bazel

@@ -37,5 +37,6 @@ 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,6 +4,7 @@ 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"
 )
@@ -104,3 +105,65 @@ 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,7 +4,9 @@ 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)
@@ -136,3 +138,49 @@ 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,12 +16,17 @@ 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 {
-	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
+	genesisTime := time.Unix(int64(genesis), 0) // lint:ignore uintcast -- Genesis timestamp will not exceed int64 in your lifetime.
+	return startFromTime(genesisTime, slot)
 }
 
 // SinceGenesis returns the number of slots since