core: refetch payload attestation data at due time

core/consensus/timer/roundtimer.go

@@ -43,9 +43,9 @@ 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.
+	case duty.Type == core.DutyAttester && gloas, duty.Type == core.DutyPayloadAttestation:
+		// From gloas, attester and payload attestation 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()

core/consensus/timer/roundtimer_internal_test.go

@@ -310,18 +310,21 @@ func TestTimerSelection(t *testing.T) {
 	var (
 		genesisTime = time.Unix(1_000_000, 0)
 		slotStart   = genesisTime.Add(slot * slotDuration)
-		// timingFor returns the slot timing with attestations due at the offset into the slot, and
-		// proposals at its start.
+		// timingFor returns the slot timing with attestations due at the offset into the slot, payload
+		// attestations 3/4 into the slot, and proposals at its start.
 		timingFor = func(attestationDue time.Duration) slotTiming {
 			return slotTiming{
 				genesisTime:  genesisTime,
 				slotDuration: slotDuration,
 				slotOffset: func(duty core.Duty) time.Duration {
-					if duty.Type == core.DutyAttester {
+					switch duty.Type {
+					case core.DutyAttester:
 						return attestationDue
+					case core.DutyPayloadAttestation:
+						return 9 * time.Second
+					default:
+						return 0
 					}
-
-					return 0
 				},
 			}
 		}
@@ -381,6 +384,24 @@ func TestTimerSelection(t *testing.T) {
 			wantExtensions: []string{"double_round_on_rearm_min_1s"},
 			want:           []roundArming{{ms(4000), ms(5000)}, {ms(5000), ms(6000)}, {ms(6000), ms(7000)}, {ms(8000), ms(10000)}, {ms(11000), ms(14000)}},
 		},
+		{
+			name:           "payload attestation, defaults",
+			duty:           core.NewPayloadAttestationDuty(slot),
+			gloas:          true,
+			wantType:       TimerEagerAheadSplit,
+			wantExtensions: []string{"double_round_on_rearm_min_1s"},
+			want:           []roundArming{{ms(10000), ms(11000)}, {ms(11000), ms(12000)}, {ms(12000), ms(13000)}, {ms(14000), ms(16000)}, {ms(17000), ms(20000)}},
+		},
+		{
+			name:           "payload attestation, eager double linear disabled and linear enabled",
+			enable:         []featureset.Feature{featureset.Linear},
+			disable:        []featureset.Feature{featureset.EagerDoubleLinear},
+			duty:           core.NewPayloadAttestationDuty(slot),
+			gloas:          true,
+			wantType:       TimerEagerAheadSplit,
+			wantExtensions: []string{"double_round_on_rearm_min_1s"},
+			want:           []roundArming{{ms(10000), ms(11000)}, {ms(11000), ms(12000)}, {ms(12000), ms(13000)}, {ms(14000), ms(16000)}, {ms(17000), ms(20000)}},
+		},
 		{
 			name:           "gloas proposer, defaults",
 			duty:           core.NewProposerDuty(slot),

core/fetcher/fetcher.go

@@ -217,8 +217,7 @@ func (f *Fetcher) fetchGloasAttesterData(ctx context.Context, slot uint64, defSe
 
 	// Fall back to fetching from all beacon nodes at the attestation due time, if no block arrived in
 	// the block proposal interval.
-	slotStart := f.genesisTime.Add(f.slotDuration * time.Duration(slot))
-	dueTime := slotStart.Add(f.slotOffsetFunc(core.NewAttesterDuty(slot)))
+	dueTime := f.dueTime(core.NewAttesterDuty(slot))
 
 	timer := f.clock.NewTimer(dueTime.Sub(f.clock.Now()))
 	defer timer.Stop()
@@ -832,45 +831,83 @@ func (f *Fetcher) forkActive(fork eth2wrap.Fork, slot uint64) bool {
 	return f.forkSchedule().Active(fork, eth2p0.Epoch(slot/f.slotsPerEpoch))
 }
 
-// fetchPayloadAttestationData returns the fetched payload attestation data for the slot.
-// The data is per-slot, not per-validator, so all payload timeliness committee members
-// in the arg set attest to the same data.
+// dueTime returns the time the duty's data is due, its slot offset into its slot.
+func (f *Fetcher) dueTime(duty core.Duty) time.Time {
+	slotStart := f.genesisTime.Add(f.slotDuration * time.Duration(duty.Slot))
+
+	return slotStart.Add(f.slotOffsetFunc(duty))
+}
+
+// fetchPayloadAttestationData returns the payload attestation data of the slot. The data is per-slot,
+// not per-validator, so all payload timeliness committee members in the arg set attest to the same data.
+//
+// The spec has members attest as soon as the payload and its blob data are seen, but only report
+// either as missing once the payload attestation due time elapsed, since they may still arrive. So
+// data that doesn't report both, or a failed fetch, is fetched again at the due time.
 func (f *Fetcher) fetchPayloadAttestationData(ctx context.Context, slot uint64, defSet core.DutyDefinitionSet) (core.UnsignedDataSet, error) {
+	dueTime := f.dueTime(core.NewPayloadAttestationDuty(slot))
+
+	if f.clock.Now().Before(dueTime) {
+		data, err := f.fetchPayloadAttestationDataOnce(ctx, slot)
+		if err != nil {
+			log.Debug(ctx, "Payload attestation data fetch failed, fetching again at due time",
+				z.U64("slot", slot), z.Any("due_time", dueTime), z.Err(err))
+		} else if data.Gloas != nil && data.Gloas.PayloadPresent && data.Gloas.BlobDataAvailable {
+			return payloadAttestationSet(data, defSet), nil
+		} else {
+			log.Debug(ctx, "Payload or blob data not seen yet, fetching payload attestation data again at due time",
+				z.U64("slot", slot), z.Bool("no_block", data.NoPayload), z.Any("due_time", dueTime))
+		}
+
+		timer := f.clock.NewTimer(dueTime.Sub(f.clock.Now()))
+		defer timer.Stop()
+
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-timer.Chan():
+		}
+	}
+
+	data, err := f.fetchPayloadAttestationDataOnce(ctx, slot)
+	if err != nil {
+		return nil, err
+	}
+
+	return payloadAttestationSet(data, defSet), nil
+}
+
+// fetchPayloadAttestationDataOnce fetches the payload attestation data of the slot from the beacon node,
+// returning the no payload data if it has not seen a block for the slot.
+func (f *Fetcher) fetchPayloadAttestationDataOnce(ctx context.Context, slot uint64) (core.VersionedPayloadAttestationData, error) {
 	opts := &eth2api.PayloadAttestationDataOpts{
 		Slot: eth2p0.Slot(slot),
 	}
 
-	var data core.VersionedPayloadAttestationData
-
 	eth2Resp, err := f.eth2Cl.PayloadAttestationData(ctx, opts)
 	if errors.Is(err, eth2client.ErrNoPayloadAttestationData) {
 		// The beacon node has not seen a block for the slot. Propose the no payload data for it,
 		// so the cluster agrees on the no-block outcome like on any other value.
-		log.Debug(ctx, "No payload seen for slot's block", z.U64("slot", slot))
-
 		version := f.forkSchedule().DataVersion(eth2p0.Epoch(slot / f.slotsPerEpoch))
 
-		data, err = core.NewNoPayloadAttestationData(version, eth2p0.Slot(slot))
-		if err != nil {
-			return nil, err
-		}
+		return core.NewNoPayloadAttestationData(version, eth2p0.Slot(slot))
 	} else if err != nil {
-		return nil, err
+		return core.VersionedPayloadAttestationData{}, err
 	} else if eth2Resp.Data == nil {
-		return nil, errors.New("payload attestation data is nil")
-	} else {
-		data, err = core.NewVersionedPayloadAttestationData(eth2Resp.Data)
-		if err != nil {
-			return nil, err
-		}
+		return core.VersionedPayloadAttestationData{}, errors.New("payload attestation data is nil")
 	}
 
+	return core.NewVersionedPayloadAttestationData(eth2Resp.Data)
+}
+
+// payloadAttestationSet returns the unsigned data set attesting the data for each validator of the set.
+func payloadAttestationSet(data core.VersionedPayloadAttestationData, defSet core.DutyDefinitionSet) core.UnsignedDataSet {
 	resp := make(core.UnsignedDataSet)
 	for pubkey := range defSet {
 		resp[pubkey] = data
 	}
 
-	return resp, nil
+	return resp
 }
 
 // fetchContributionData fetches the sync committee contribution data.

core/fetcher/fetcher_internal_test.go

@@ -7,9 +7,11 @@ import (
 	"fmt"
 	"math/big"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
+	eth2client "github.com/attestantio/go-eth2-client"
 	eth2api "github.com/attestantio/go-eth2-client/api"
 	eth2v1 "github.com/attestantio/go-eth2-client/api/v1"
 	eth2spec "github.com/attestantio/go-eth2-client/spec"
@@ -334,6 +336,183 @@ func (c metadataClient) EPBSProposal(ctx context.Context, opts *eth2api.EPBSProp
 	return resp, nil
 }
 
+func TestFetchPayloadAttestationDue(t *testing.T) {
+	const slot = 1
+
+	var (
+		t0          = time.Unix(1_000_000, 0)
+		slotStart   = t0.Add(12 * time.Second)
+		payloadDue  = slotStart.Add(6 * time.Second) // The duty starts once the payload is due, half way into the slot.
+		dueTime     = slotStart.Add(9 * time.Second) // Payload attestations are due 3/4 into the slot.
+		errFetch    = errors.New("payload attestation data not yet final")
+		pubkey      = testutil.RandomCorePubKey(t)
+		duty        = core.NewPayloadAttestationDuty(slot)
+		defSet      = core.DutyDefinitionSet{pubkey: core.NewPTCDefinition(&eth2v1.PTCDuty{Slot: slot, ValidatorIndex: 2})}
+		root        = eth2p0.Root{0x01}
+		seen        = &gloas.PayloadAttestationData{BeaconBlockRoot: root, Slot: slot, PayloadPresent: true, BlobDataAvailable: true}
+		blobMissing = &gloas.PayloadAttestationData{BeaconBlockRoot: root, Slot: slot, PayloadPresent: true}
+	)
+
+	// response is a beacon node response to a payload attestation data request.
+	type response struct {
+		data *gloas.PayloadAttestationData
+		err  error
+	}
+
+	// setup returns a gloas fetcher at the time into the slot whose beacon node returns the responses
+	// in order, along with a channel of fetched data sets and the number of beacon node calls.
+	setup := func(t *testing.T, now time.Time, responses ...response) (*Fetcher, *clockwork.FakeClock, <-chan core.UnsignedDataSet, *atomic.Int64) {
+		t.Helper()
+
+		bmock, err := beaconmock.New(t.Context(), beaconmock.WithGenesisTime(t0), beaconmock.WithSlotsPerEpoch(1))
+		require.NoError(t, err)
+
+		var calls atomic.Int64
+
+		bmock.PayloadAttestationDataFunc = func(context.Context, eth2p0.Slot) (*eth2spec.VersionedPayloadAttestationData, error) {
+			resp := responses[calls.Add(1)-1]
+			if resp.err != nil {
+				return nil, resp.err
+			}
+
+			return &eth2spec.VersionedPayloadAttestationData{Version: eth2spec.DataVersionGloas, Gloas: resp.data}, nil
+		}
+
+		fetch, err := New(t.Context(), bmock, nil, false, &GraffitiBuilder{},
+			func() eth2wrap.ForkForkSchedule { return eth2wrap.ForkForkSchedule{eth2wrap.Gloas: {Epoch: 0}} }, 1, &gloas.BuilderConfig{}, false)
+		require.NoError(t, err)
+
+		clock := clockwork.NewFakeClockAt(now)
+		fetch.clock = clock
+
+		fetched := make(chan core.UnsignedDataSet, 1)
+
+		fetch.Subscribe(func(_ context.Context, _ core.Duty, set core.UnsignedDataSet) error {
+			fetched <- set
+			return nil
+		})
+
+		return fetch, clock, fetched, &calls
+	}
+
+	// fetchAsync runs Fetch in the background, returning its error channel.
+	fetchAsync := func(ctx context.Context, fetch *Fetcher) <-chan error {
+		errCh := make(chan error, 1)
+
+		go func() {
+			errCh <- fetch.Fetch(ctx, duty, defSet)
+		}()
+
+		return errCh
+	}
+
+	requireFetched := func(t *testing.T, fetched <-chan core.UnsignedDataSet, errCh <-chan error, expected *gloas.PayloadAttestationData) {
+		t.Helper()
+
+		select {
+		case err := <-errCh:
+			require.NoError(t, err)
+		case <-time.After(time.Second):
+			require.Fail(t, "payload attestation data not fetched")
+		}
+
+		set := <-fetched
+		require.Equal(t, expected, set[pubkey].(core.VersionedPayloadAttestationData).Gloas)
+	}
+
+	// requireRefetchedAtDue asserts that the fetcher waits for the due time before fetching again.
+	requireRefetchedAtDue := func(t *testing.T, clock *clockwork.FakeClock, fetched <-chan core.UnsignedDataSet) {
+		t.Helper()
+
+		require.NoError(t, clock.BlockUntilContext(t.Context(), 1))
+		clock.Advance(dueTime.Sub(payloadDue) - time.Millisecond)
+
+		select {
+		case <-fetched:
+			require.Fail(t, "fetched again before the due time")
+		case <-time.After(10 * time.Millisecond):
+		}
+
+		clock.Advance(time.Millisecond)
+	}
+
+	t.Run("payload and blob data seen", func(t *testing.T) {
+		fetch, _, fetched, calls := setup(t, payloadDue, response{data: seen})
+
+		requireFetched(t, fetched, fetchAsync(t.Context(), fetch), seen)
+		require.EqualValues(t, 1, calls.Load())
+	})
+
+	t.Run("blob data seen by the due time", func(t *testing.T) {
+		fetch, clock, fetched, calls := setup(t, payloadDue, response{data: blobMissing}, response{data: seen})
+
+		errCh := fetchAsync(t.Context(), fetch)
+
+		requireRefetchedAtDue(t, clock, fetched)
+
+		requireFetched(t, fetched, errCh, seen)
+		require.EqualValues(t, 2, calls.Load())
+	})
+
+	t.Run("blob data missing at the due time", func(t *testing.T) {
+		fetch, clock, fetched, calls := setup(t, payloadDue, response{data: blobMissing}, response{data: blobMissing})
+
+		errCh := fetchAsync(t.Context(), fetch)
+
+		requireRefetchedAtDue(t, clock, fetched)
+
+		requireFetched(t, fetched, errCh, blobMissing)
+		require.EqualValues(t, 2, calls.Load())
+	})
+
+	t.Run("block seen by the due time", func(t *testing.T) {
+		late := &gloas.PayloadAttestationData{BeaconBlockRoot: root, Slot: slot}
+		fetch, clock, fetched, calls := setup(t, payloadDue, response{err: eth2client.ErrNoPayloadAttestationData}, response{data: late})
+
+		errCh := fetchAsync(t.Context(), fetch)
+
+		requireRefetchedAtDue(t, clock, fetched)
+
+		requireFetched(t, fetched, errCh, late)
+		require.EqualValues(t, 2, calls.Load())
+	})
+
+	t.Run("fetch error before the due time", func(t *testing.T) {
+		fetch, clock, fetched, calls := setup(t, payloadDue, response{err: errFetch}, response{data: blobMissing})
+
+		errCh := fetchAsync(t.Context(), fetch)
+
+		requireRefetchedAtDue(t, clock, fetched)
+
+		requireFetched(t, fetched, errCh, blobMissing)
+		require.EqualValues(t, 2, calls.Load())
+	})
+
+	t.Run("fetch error at the due time", func(t *testing.T) {
+		fetch, clock, _, _ := setup(t, payloadDue, response{err: errFetch}, response{err: errFetch})
+
+		errCh := fetchAsync(t.Context(), fetch)
+
+		require.NoError(t, clock.BlockUntilContext(t.Context(), 1))
+		clock.Advance(dueTime.Sub(payloadDue))
+
+		select {
+		case err := <-errCh:
+			require.ErrorContains(t, err, "not yet final")
+		case <-time.After(time.Second):
+			require.Fail(t, "fetch did not return")
+		}
+	})
+
+	t.Run("due time already passed", func(t *testing.T) {
+		// Fetch starts after the due time, e.g. upon a retry, and fetches once.
+		fetch, _, fetched, calls := setup(t, dueTime.Add(time.Second), response{data: blobMissing})
+
+		requireFetched(t, fetched, fetchAsync(t.Context(), fetch), blobMissing)
+		require.EqualValues(t, 1, calls.Load())
+	})
+}
+
 func TestFetchEPBSBuilderSource(t *testing.T) {
 	const (
 		slot = 1