Async sync committee aggregator selection

This PR refactors `ProcessExecutionPayload` with `ApplyExecutionPayload`
so caller can use Apply method to calculate post state root. Note that
validations are not required for Apply function. We really need the
state mutation lines that's:
```
1. Ensure latest_block_header.state_root is set (if zero, set it to the pre‑payload HashTreeRoot)...
2. processExecutionRequests()
3. QueueBuilderPayment()
4. SetExecutionPayloadAvailability(state.Slot(), true)
5. SetLatestBlockHash(payload.BlockHash())
```
I decided to keep them there because a. it's cheap b. it makes refactor
cleaner to reason c. API/caller may want to validate envelope and bid
consistency (ex: beacon api has option to validate consensus)

beacon-chain/core/gloas/payload.go

@@ -112,6 +112,34 @@ func ProcessExecutionPayload(
 		return errors.Wrap(err, "signature verification failed")
 	}
 
+	envelope, err := signedEnvelope.Envelope()
+	if err != nil {
+		return errors.Wrap(err, "could not get envelope from signed envelope")
+	}
+
+	if err := ApplyExecutionPayload(ctx, st, envelope); err != nil {
+		return err
+	}
+
+	r, err := st.HashTreeRoot(ctx)
+	if err != nil {
+		return errors.Wrap(err, "could not get hash tree root")
+	}
+	if r != envelope.StateRoot() {
+		return fmt.Errorf("state root mismatch: expected %#x, got %#x", envelope.StateRoot(), r)
+	}
+
+	return nil
+}
+
+// ApplyExecutionPayload applies the execution payload envelope to the state and performs the same
+// consistency checks as the full processing path. This keeps the post-payload state root computation
+// on a shared code path, even though some bid/payload checks are not strictly required for the root itself.
+func ApplyExecutionPayload(
+	ctx context.Context,
+	st state.BeaconState,
+	envelope interfaces.ROExecutionPayloadEnvelope,
+) error {
 	latestHeader := st.LatestBlockHeader()
 	if len(latestHeader.StateRoot) == 0 || bytes.Equal(latestHeader.StateRoot, make([]byte, 32)) {
 		previousStateRoot, err := st.HashTreeRoot(ctx)
@@ -128,10 +156,6 @@ func ProcessExecutionPayload(
 	if err != nil {
 		return errors.Wrap(err, "could not compute block header root")
 	}
-	envelope, err := signedEnvelope.Envelope()
-	if err != nil {
-		return errors.Wrap(err, "could not get envelope from signed envelope")
-	}
 
 	beaconBlockRoot := envelope.BeaconBlockRoot()
 	if !bytes.Equal(beaconBlockRoot[:], blockHeaderRoot[:]) {
@@ -217,14 +241,6 @@ func ProcessExecutionPayload(
 		return errors.Wrap(err, "could not set latest block hash")
 	}
 
-	r, err := st.HashTreeRoot(ctx)
-	if err != nil {
-		return errors.Wrap(err, "could not get hash tree root")
-	}
-	if r != envelope.StateRoot() {
-		return fmt.Errorf("state root mismatch: expected %#x, got %#x", envelope.StateRoot(), r)
-	}
-
 	return nil
 }
 

changelog/tt_apply_execution_payload.md

@@ -0,0 +1,2 @@
+### Ignored
+- Refactor ProcessExecutionPayload to ApplyExecutionPayload

validator/client/validator.go

@@ -60,6 +60,8 @@ var (
 	ErrValidatorsAllExited          = errors.New("All validators are exited, no more work to perform...")
 )
 
+const syncCommitteeSelectionTimeout = 1 * time.Second
+
 var (
 	msgCouldNotFetchKeys = "could not fetch validating keys"
 	msgNoKeysFetched     = "No validating keys fetched. Waiting for keys..."
@@ -896,8 +898,16 @@ func (v *validator) isSyncCommitteeAggregator(ctx context.Context, slot primitiv
 
 	// Override selections with aggregated ones if the node is part of a Distributed Validator.
 	if v.distributed && len(selections) > 0 {
-		var err error
-		selections, err = v.validatorClient.AggregatedSyncSelections(ctx, selections)
+		var (
+			aggCtx = ctx
+			cancel context.CancelFunc
+			err    error
+		)
+		if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) > syncCommitteeSelectionTimeout {
+			aggCtx, cancel = context.WithTimeout(ctx, syncCommitteeSelectionTimeout)
+			defer cancel()
+		}
+		selections, err = v.validatorClient.AggregatedSyncSelections(aggCtx, selections)
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to get aggregated sync selections")
 		}