core: start gloas attester consensus at slot start

app/app.go

@@ -627,16 +627,18 @@ func wireCoreWorkflow(ctx context.Context, life *lifecycle.Manager, conf Config,
 			z.Int("builder_urls", len(conf.BuilderURLs)))
 	}
 
-	fetch, err := fetcher.New(eth2Cl, builderRegSvc.FeeRecipient, conf.BuilderAPI, graffitiBuilder,
+	fetch, err := fetcher.New(ctx, eth2Cl, builderRegSvc.FeeRecipient, conf.BuilderAPI, graffitiBuilder,
 		forkSchedule, slotsPerEpoch, builderConfig(conf), featureset.Enabled(featureset.FetchOnlyCommIdx0))
 	if err != nil {
 		return err
 	}
 
-	// Invalidate early-fetched (head-event-triggered) attestation data on reorgs, since the cached
-	// data was verified against a head that may no longer be canonical.
-	if featureset.Enabled(featureset.FetchAttOnBlock) || featureset.Enabled(featureset.FetchAttOnBlockWithDelay) {
+	if featureset.Enabled(featureset.FetchAttOnBlock) {
+		// Invalidate early-fetched (head-event-triggered) attestation data on reorgs, since the cached
+		// data was verified against a head that may no longer be canonical.
 		sseListener.SubscribeChainReorgEvent(fetch.HandleChainReorg)
+		// From gloas, head events trigger fetching attestation data for consensus directly.
+		sseListener.SubscribeHeadEvent(fetch.HandleHeadEvent)
 	}
 
 	dutyDB := dutydb.NewMemDB(deadlinerFunc("dutydb"))

app/featureset/featureset.go

@@ -76,33 +76,27 @@ const (
 	// head event is received in time.
 	FetchAttOnBlock = "fetch_att_on_block"
 
-	// FetchAttOnBlockWithDelay enables fetching attestation data with 300ms delay.
-	// When enabled with FetchAttOnBlock, uses the attestation deadline plus 300ms as fallback timeout.
-	// When enabled alone, uses the attestation deadline plus 300ms as timeout.
-	FetchAttOnBlockWithDelay = "fetch_att_on_block_with_delay"
-
 	// DisableDutiesCache is a safety measure to disable duties cache.
 	DisableDutiesCache = "disable_duties_cache"
 )
 
 var (
 	// state defines the current rollout status of each feature.
 	state = map[Feature]status{
-		EagerDoubleLinear:        statusStable,
-		ConsensusParticipate:     statusStable,
-		MockAlpha:                statusAlpha,
-		JSONRequests:             statusAlpha,
-		GnosisBlockHotfix:        statusAlpha,
-		Linear:                   statusAlpha,
-		SSEReorgDuties:           statusAlpha,
-		AttestationInclusion:     statusAlpha,
-		ProposalTimeout:          statusStable,
-		QUIC:                     statusAlpha,
-		FetchOnlyCommIdx0:        statusStable,
-		ChainSplitHalt:           statusAlpha,
-		FetchAttOnBlock:          statusAlpha,
-		FetchAttOnBlockWithDelay: statusAlpha,
-		DisableDutiesCache:       statusAlpha,
+		EagerDoubleLinear:    statusStable,
+		ConsensusParticipate: statusStable,
+		MockAlpha:            statusAlpha,
+		JSONRequests:         statusAlpha,
+		GnosisBlockHotfix:    statusAlpha,
+		Linear:               statusAlpha,
+		SSEReorgDuties:       statusAlpha,
+		AttestationInclusion: statusAlpha,
+		ProposalTimeout:      statusStable,
+		QUIC:                 statusAlpha,
+		FetchOnlyCommIdx0:    statusStable,
+		ChainSplitHalt:       statusAlpha,
+		FetchAttOnBlock:      statusStable,
+		DisableDutiesCache:   statusAlpha,
 		// Add all features and their status here.
 	}
 

core/consensus/qbft/qbft.go

@@ -277,9 +277,9 @@ func NewConsensus(ctx context.Context, eth2Cl eth2wrap.Client, forkSchedule func
 		return nil, errors.Wrap(err, "fetch genesis time")
 	}
 
-	slotDuration, _, err := eth2wrap.FetchSlotsConfig(ctx, eth2Cl)
+	slotDuration, slotsPerEpoch, err := eth2wrap.FetchSlotsConfig(ctx, eth2Cl)
 	if err != nil {
-		return nil, errors.Wrap(err, "fetch slot duration")
+		return nil, errors.Wrap(err, "fetch slots config")
 	}
 
 	slotOffsetFunc, err := core.NewSlotOffsetFunc(ctx, eth2Cl, forkSchedule)
@@ -298,7 +298,7 @@ func NewConsensus(ctx context.Context, eth2Cl eth2wrap.Client, forkSchedule func
 		snifferFunc:         snifferFunc,
 		gaterFunc:           gaterFunc,
 		dropFilter:          log.Filter(),
-		timerFunc:           timer.GetRoundTimerFunc(genesisTime, slotDuration, slotOffsetFunc),
+		timerFunc:           timer.GetRoundTimerFunc(genesisTime, slotDuration, slotsPerEpoch, slotOffsetFunc, forkSchedule),
 		metrics:             metrics.NewConsensusMetrics(protocols.QBFTv2ProtocolID),
 		compareAttestations: compareAttestations,
 	}

core/consensus/qbft/qbft_internal_test.go

@@ -888,7 +888,8 @@ func TestInstanceIO_MaybeStart(t *testing.T) {
 		c.gaterFunc = func(core.Duty) bool { return true }
 		c.mutable.instances = make(map[core.Duty]*instance.IO[Msg])
 		// Use zero values for tests to use default clock.Now() behavior
-		c.timerFunc = timer.GetRoundTimerFunc(time.Time{}, 0, func(core.Duty) time.Duration { return 0 })
+		c.timerFunc = timer.GetRoundTimerFunc(time.Time{}, 0, 32, func(core.Duty) time.Duration { return 0 },
+			func() eth2wrap.ForkForkSchedule { return eth2wrap.ForkForkSchedule{} })
 
 		// Generate a p2p private key pair.
 		p2pKey := testutil.GenerateInsecureK1Key(t, 0)

core/consensus/qbft/roundtimer_internal_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 
+	"github.com/obolnetwork/charon/app/eth2wrap"
 	"github.com/obolnetwork/charon/app/featureset"
 	"github.com/obolnetwork/charon/core"
 	"github.com/obolnetwork/charon/core/consensus/timer"
@@ -24,24 +25,25 @@ import (
 // all peers are blocked, so it neither waits in real time nor depends on goroutine scheduling.
 func TestRoundTimersFirstLeaderDown(t *testing.T) {
 	const (
-		nodes        = 4
-		slot         = 1
-		slotDuration = 12 * time.Second
-		latency      = 10 * time.Millisecond
+		nodes         = 4
+		slot          = 1
+		slotsPerEpoch = 32
+		slotDuration  = 12 * time.Second
+		latency       = 10 * time.Millisecond
 		// decideAfter is how long the second round takes to decide after the first round times out:
 		// the round changes, then the pre-prepare, prepares and commits, each delayed by the latency.
 		decideAfter = 4 * latency
 	)
 
-	// Duties start at the start of the slot.
-	zeroOffset := func(core.Duty) time.Duration { return 0 }
-
 	tests := []struct {
 		name     string
 		enable   []featureset.Feature
 		disable  []featureset.Feature
 		dutyType core.DutyType
-		wantType timer.Type
+		gloas    bool // Whether gloas is active at the duty's slot.
+		// dutyOffset is the duty's offset into the slot, when it is due.
+		dutyOffset time.Duration
+		wantType   timer.Type
 		// firstRoundEnd is when the first round times out, relative to the start of the slot.
 		firstRoundEnd time.Duration
 	}{
@@ -58,6 +60,14 @@ func TestRoundTimersFirstLeaderDown(t *testing.T) {
 			wantType:      timer.TimerIncreasing,
 			firstRoundEnd: time.Second,
 		},
+		{
+			name:          "gloas attester, defaults",
+			dutyType:      core.DutyAttester,
+			gloas:         true,
+			dutyOffset:    3 * time.Second,
+			wantType:      timer.TimerEagerAheadSplit,
+			firstRoundEnd: 4 * time.Second,
+		},
 		{
 			name:          "proposer, defaults",
 			dutyType:      core.DutyProposer,
@@ -94,7 +104,16 @@ func TestRoundTimersFirstLeaderDown(t *testing.T) {
 				// The bubble's fake time starts at the start of the slot.
 				slotStart := time.Now()
 				genesisTime := slotStart.Add(-slot * slotDuration)
-				timerFunc := timer.GetRoundTimerFunc(genesisTime, slotDuration, zeroOffset)
+				forkSchedule := func() eth2wrap.ForkForkSchedule {
+					if !test.gloas {
+						return eth2wrap.ForkForkSchedule{}
+					}
+
+					return eth2wrap.ForkForkSchedule{eth2wrap.Gloas: {Epoch: 0}}
+				}
+
+				slotOffset := func(core.Duty) time.Duration { return test.dutyOffset }
+				timerFunc := timer.GetRoundTimerFunc(genesisTime, slotDuration, slotsPerEpoch, slotOffset, forkSchedule)
 
 				duty := core.Duty{Slot: slot, Type: test.dutyType}
 				require.Equal(t, test.wantType, timerFunc(duty).Type())

core/consensus/timer/extensions.go

@@ -25,9 +25,9 @@ type extension interface {
 	// onArm returns the time the extension adds to the round's duration when the round is armed.
 	onArm(round int64) time.Duration
 	// onRearm returns the new deadline of the round armed again, given
-	// the round's duration and end including extras and its deadline when armed, or false if the
-	// extension doesn't act upon it.
-	onRearm(duration, end time.Duration, deadline, now time.Time) (time.Time, bool)
+	// the round's duration and end including extras, the anchor's lead and its deadline when armed,
+	// or false if the extension doesn't act upon it.
+	onRearm(duration, end, lead time.Duration, deadline, now time.Time) (time.Time, bool)
 }
 
 // noopExtension provides the default extension behaviour: applying to every duty without acting.
@@ -38,7 +38,7 @@ func (noopExtension) appliesForDuty(core.Duty) bool { return true }
 
 func (noopExtension) onArm(int64) time.Duration { return 0 }
 
-func (noopExtension) onRearm(time.Duration, time.Duration, time.Time, time.Time) (time.Time, bool) {
+func (noopExtension) onRearm(time.Duration, time.Duration, time.Duration, time.Time, time.Time) (time.Time, bool) {
 	return time.Time{}, false
 }
 
@@ -48,24 +48,39 @@ type resetOnRearm struct{ noopExtension }
 
 func (resetOnRearm) name() string { return "reset_on_rearm" }
 
-func (resetOnRearm) onRearm(duration, _ time.Duration, _, now time.Time) (time.Time, bool) {
+func (resetOnRearm) onRearm(duration, _, _ time.Duration, _, now time.Time) (time.Time, bool) {
 	return now.Add(duration), true
 }
 
-// doubleOnRearm extends the round when it is armed again, from its deadline by the round's end,
+// doubleTotalOnRearm extends the round when it is armed again, from its deadline by the round's end,
 // doubling the time since the start of the first round. Extending from the deadline, rather than
 // resetting the round's timer, keeps round end times aligned across peers: QBFT arms a round again
 // upon a justified pre-prepare, so resetting has no effect on the leader, who resets at the start of
 // the round, while it has a large effect on the other peers, who reset when they receive the
 // justified pre-prepare.
-type doubleOnRearm struct{ noopExtension }
+type doubleTotalOnRearm struct{ noopExtension }
 
-func (doubleOnRearm) name() string { return "double_on_rearm" }
+func (doubleTotalOnRearm) name() string { return "double_total_on_rearm" }
 
-func (doubleOnRearm) onRearm(_, end time.Duration, deadline, _ time.Time) (time.Time, bool) {
+func (doubleTotalOnRearm) onRearm(_, end, _ time.Duration, deadline, _ time.Time) (time.Time, bool) {
 	return deadline.Add(end), true
 }
 
+// doubleRoundOnRearm extends the round from its deadline by its duration after the duty start, at
+// least by minimum, when it is armed again, doubling the round. The lead a first round anchored ahead
+// of the duty start spans doesn't count.
+type doubleRoundOnRearm struct {
+	noopExtension
+
+	minimum time.Duration
+}
+
+func (e doubleRoundOnRearm) name() string { return "double_round_on_rearm_min_" + e.minimum.String() }
+
+func (e doubleRoundOnRearm) onRearm(duration, end, lead time.Duration, deadline, _ time.Time) (time.Time, bool) {
+	return deadline.Add(max(e.minimum, min(duration, end-lead))), true
+}
+
 // proposalTimeout adds 500ms to the first round of proposer duties.
 type proposalTimeout struct{ noopExtension }
 

core/consensus/timer/roundtimer.go

@@ -7,35 +7,48 @@
 	"testing"
 	"time"
 
+	eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0"
 	"github.com/jonboulle/clockwork"
 
+	"github.com/obolnetwork/charon/app/eth2wrap"
 	"github.com/obolnetwork/charon/app/featureset"
 	"github.com/obolnetwork/charon/core"
 )
 
 // RoundTimerFunc is a function that returns a round timer.
 type RoundTimerFunc func(core.Duty) RoundTimer
 
-// GetRoundTimerFunc returns a timer function based on the enabled features.
+// GetRoundTimerFunc returns a timer function based on the fork schedule and the enabled features.
 // Genesis time and slot duration are required to calculate deterministic slot start times, while
 // the slot offset function provides the duty's offset into the slot at which consensus starts.
-func GetRoundTimerFunc(genesisTime time.Time, slotDuration time.Duration, slotOffsetFunc core.SlotOffsetFunc) RoundTimerFunc {
+func GetRoundTimerFunc(genesisTime time.Time, slotDuration time.Duration, slotsPerEpoch uint64,
+	slotOffsetFunc core.SlotOffsetFunc, forkSchedule func() eth2wrap.ForkForkSchedule,
+) RoundTimerFunc {
 	timing := slotTiming{
 		genesisTime:  genesisTime,
 		slotDuration: slotDuration,
 		slotOffset:   slotOffsetFunc,
 	}
 
 	return func(duty core.Duty) RoundTimer {
-		return newRoundTimer(duty, selectTimer(duty), timing, clockwork.NewRealClock())
+		gloas := forkSchedule().Active(eth2wrap.Gloas, eth2p0.Epoch(duty.Slot/slotsPerEpoch))
+
+		return newRoundTimer(duty, selectTimer(duty, gloas), timing, clockwork.NewRealClock())
 	}
 }
 
-// selectTimer returns the timer for the duty, as selected by the feature set, with its extensions.
-func selectTimer(duty core.Duty) timerDef {
+// selectTimer returns the timer for the duty, as selected by the fork, whether gloas is active at the
+// duty's slot, and the feature set, with its extensions.
+func selectTimer(duty core.Duty, gloas bool) timerDef {
 	var def timerDef
 
 	switch {
+	case duty.Type == core.DutyAttester && gloas:
+		// From gloas, attester duties use the eager ahead split timer regardless of the feature set,
+		// since round end times must be identical across the cluster.
+		// TODO(post-gloas): make this the default for attester duties without the gloas check, removing
+		// the fork schedule from GetRoundTimerFunc.
+		def = eagerAheadSplitTimer()
 	case duty.Type == core.DutyProposer && featureset.Enabled(featureset.Linear):
 		// The linear timer has precedence over the eager double linear timer, but only for proposer duties.
 		def = linearTimer()
@@ -65,6 +78,7 @@
 	TimerIncreasing        Type = "inc"
 	TimerEagerDoubleLinear Type = "eager_dlinear"
 	TimerLinear            Type = "linear"
+	TimerEagerAheadSplit   Type = "eager_ahead_split"
 )
 
 // RoundTimer provides the duration for each consensus round.
@@ -127,6 +141,8 @@
 	// start returns the start of the duty's round armed at now, given the total duration of the rounds
 	// before it.
 	start(duty core.Duty, timing slotTiming, now time.Time, priorRoundsDuration time.Duration) time.Time
+	// lead returns how long before the duty start the first round starts.
+	lead() time.Duration
 }
 
 // anchorLocalDutyStart starts the first round when the duty starts on this node, and each later round
@@ -138,6 +154,8 @@
 	return now
 }
 
+func (anchorLocalDutyStart) lead() time.Duration { return 0 }
+
 // anchorDutyStart starts the first round at the duty's start, its offset into the slot, with later
 // rounds following on at absolute times, aligned across peers.
 type anchorDutyStart struct{}
@@ -152,22 +170,51 @@
 	return start.Add(priorRoundsDuration)
 }
 
+func (anchorDutyStart) lead() time.Duration { return 0 }
+
+// anchorPreviousInterval starts the first round at the start of the interval preceding the duty's
+// own, one interval before the duty start, with later rounds following on at absolute times, aligned
+// across peers.
+type anchorPreviousInterval struct {
+	leadTime time.Duration
+}
+
+func (a anchorPreviousInterval) start(duty core.Duty, timing slotTiming, now time.Time, priorRoundsDuration time.Duration) time.Time {
+	dutyStart := timing.dutyStart(duty)
+	if dutyStart.IsZero() {
+		// Without slot timing (only in tests), the first round starts when armed.
+		return now.Add(priorRoundsDuration)
+	}
+
+	return dutyStart.Add(priorRoundsDuration - a.lead())
+}
+
+func (a anchorPreviousInterval) lead() time.Duration { return a.leadTime }
+
 // slotTiming provides the slot start times anchored timers start at.
 type slotTiming struct {
 	genesisTime  time.Time
 	slotDuration time.Duration
 	slotOffset   core.SlotOffsetFunc
 }
 
+// slotStart returns the start of the duty's slot, or the zero time without slot timing (only in tests).
+func (s slotTiming) slotStart(duty core.Duty) time.Time {
+	if s.genesisTime.IsZero() || s.slotDuration <= 0 {
+		return time.Time{}
+	}
+
+	return s.genesisTime.Add(s.slotDuration * time.Duration(duty.Slot))
+}
+
 // dutyStart returns the start of the duty's slot plus the duty's offset into the slot, or the zero
 // time without slot timing (only in tests).
 func (s slotTiming) dutyStart(duty core.Duty) time.Time {
-	if s.genesisTime.IsZero() || s.slotDuration <= 0 {
+	slotStart := s.slotStart(duty)
+	if slotStart.IsZero() {
 		return time.Time{}
 	}
 
-	slotStart := s.genesisTime.Add(s.slotDuration * time.Duration(duty.Slot))
-
 	return slotStart.Add(s.slotOffset(duty))
 }
 
@@ -227,11 +274,12 @@
 
 // rearmDeadline returns the deadline of the round armed again, as defined by the first extension
 // acting upon it, or its deadline if none does. In practice, QBFT rearms the current round
-// upon a justified pre-prepare for it. Two extensions currently act upon this rearming, declared
-// by the timers: resetOnRearm by inc and linear, and doubleOnRearm by eager_dlinear.
+// upon a justified pre-prepare for it. Three extensions currently act upon this rearming, declared
+// by the timers: resetOnRearm by inc and linear, doubleTotalOnRearm by eager_dlinear, and
+// doubleRoundOnRearm by eager_ahead_split.
 func (t *roundTimer) rearmDeadline(round int64, currentDeadline, now time.Time) time.Time {
 	for _, ext := range t.def.extensions {
-		adjustedDeadline, ok := ext.onRearm(t.duration(round), t.end(round), currentDeadline, now)
+		adjustedDeadline, ok := ext.onRearm(t.duration(round), t.end(round), t.def.anchor.lead(), currentDeadline, now)
 		if ok {
 			return adjustedDeadline
 		}