## EPF Week 14 Update

This week, I focused on building out payload timeliness committee related functionality into the validator client.

### PTC Duty Fetching
First, I extended the `duties_service` to maintain a dedicated `ptc_duties` map:

type PtcMap = HashMap<Epoch, (DependentRoot, HashMap<PublicKeyBytes, PtcDuty>)>

Within the same process responsible for polling the beacon node for attesters, proposers, and sync committee members every slot, we query a new `POST validator/duties/ptc/{epoch}` endpoint to fetch the ptc duties for a given epoch, which come back in shape:

pub struct PtcDuty {
    pub pubkey: PublicKeyBytes,
    #[serde(with = "serde_utils::quoted_u64")]
    pub validator_index: u64,
    pub slot: Slot,
}

I mentioned that we make this query every slot, but it's unlikely that the ptc committee will change mid-epoch. This would be a re-org scenario. In the happy case, the ptc committee members for an epoch can be known a full epoch in advance. We can tell if there has been a re-org by just checking each slot whether the `dependent_root` has changed by making a small query to the ptc duties endpoint to get back the `dependent_root`. If it is the same as the previous slot, we can stop there. If it has changed, we will hit the endpoint again but request duties for every validator being managed by the VC.

We use the results to update the `PTCMap` to track the slots in which validators managed by our VC are responsible for submitting a payload attestation 3/4 into the slot, per the [spec](https://ethereum.github.io/consensus-specs/specs/gloas/validator/#payload-timeliness-attestation).

The implementation can be seen [here](https://github.com/sigp/lighthouse/compare/gloas-containers...shane-moore:gloas-vc-ptc-duty).

### PayloadAttestation Submissions
If one of the validators in the validator client is part of the ptc committee for a slot, the validator client needs to request a payload attestation from the beacon node 3/4ths into the slot. Upon receiving, it should sign and submit back to the beacon node for gossip. We'll need to build a payload attestation service in the validator client that can accomplish this performantly.

In order to gain some inspiration, we should take a look at how the attestation committee service in lighthouse's validator client works and see if we can gain some inspiration for our ptc attestation service.

A key difference right off the bat is that:
- attesters in the validator client will be split into attestation sub-committees per slot, and their votes will be aggregated, whereas there are no sub-committees nor aggregation for ptc attestations.

However, they do share lots of similarities, namely that a validator in either the attestation committee or the ptc committee both are responsible for grabbing a vote from the beacon node to then sign and broadcast by a certain time in the slot.

#### `attestation_service.rs`
The entry point is `start_update_service`, which creates a persistent loop that uses tokio's async timer to wait for 1/3rd through each slot and then spawning attestation related tasks via `spawn_attestation_tasks` in the background as to avoid blocking the main service loop.

In `spawn_attestation_tasks`, we then group validators by committee:
`HashMap<CommitteeIndex, Vec<DutyAndProof>>`
and then span seperate async task for each committee as to parallelize the work per committee of publishing and aggregating attestations:

 self.inner.executor.spawn_ignoring_error(
                    self.clone().publish_attestations_and_aggregates(
                        slot,
                        committee_index,
                        validator_duties,
                        aggregate_production_instant,
                    ),
                    "attestation publish",
                );

`publish_attestations_and_aggregates` will start by downloading, signing, and publishing an `Attestation` for each validator in our VC via `produce_and_publish_attestations`:

 let attestation_opt = self.produce_and_publish_attestations(slot, committee_index, &validator_duties)

The following step in `publish_attestations_and_aggregates` is to create aggregates from the attestations, but we can skip that aspect since there will be no aggregation for payload attestations. Let's look deeper into `produce_and_publish_attestations` instead.

`produce_and_publish_attestations` will query the beacon node for attestation data based off slot and `committee_index`:

beacon_node.get_validator_attestation_data(slot, committee_index)

Then, we loop over each validator in our VC with duties for this slot and create futures for attestations to be signed by each validator with a duty. A `join_all` is then used to drive each of the underlying futures to completion, enabling parallel signing.

    // Execute all the futures in parallel, collecting any successful results.
let (ref attestations, ref validator_indices): (Vec<_>, Vec<_>) = join_all(signing_futures)
            .await
            .into_iter()
            .flatten()
            .unzip();

The resulting signed attestations are submitted to the BN in a single POST:

 beacon_node.post_beacon_pool_attestations_v2::<S::E>(single_attestations, fork_name)

For our `payload_attestation_service.rs`, I think we can uses the following patterns from `attestation_service.rs`:
- 1 fetch per slot of `AttestationData`
- parallel signing of `AttestationDataMessage` for validators in our VC who are on a PTC committee per slot
- all signed `AttestationDataMessage` can be sent to the BN in a single `POST`
- Use `beacon_nodes.first_success()` for automatic failover between multiple beacon nodes
- Not discussed above, but we can do similar metrics integrations for timing of events like signing
- similar validation that duty's slot matches the current slot before signing

## Vision for `payload_attestation_service.rs`

We'll start off by using a builder pattern as to respect dependency injection and create a `PayloadAttestationServiceBuilder`.

We'll have a top level data structure for the service:

/// Attempts to produce payload attestations for all known PTC validators 3/4th of the way through each slot.
pub struct PayloadAttestationService<S, T> {
    inner: Arc<Inner<S, T>>,
}

pub struct Inner<S, T> {
    duties_service: Arc<DutiesService<S, T>>,
    validator_store: Arc<S>,
    slot_clock: T,
    beacon_nodes: Arc<BeaconNodeFallback<T>>,
    executor: TaskExecutor,
    chain_spec: Arc<ChainSpec>,
    disable: bool,
}

Note that we use an `Inner` struct within an Arc as to minimize memory overhead. Essentially, if you clone the service during spawns, it will just increase the reference count of the inner `Arc`. Without it, we'd need to increase the reference count of all the inner `Arc` on clone.

Next, we'll create a `start_update_service` that will create a persistent loop in a tokio task responsible for spawning payload attestation tasks (`spawn_payload_attestation_tasks`) at 3/4 through a slot.

The `spawn_payload_attestation_tasks` will then get the list of ptc duties for that slot:

 let duties = self.duties_service.get_ptc_duties_for_slot(slot);

We still don't want to block the main thread from being able to precisely spawn payload attestation tasks at 3/4 for every slot, so we create a new tokio task to actually query, sign, and publish the payload attestations:

   self.executor.spawn_ignoring_error(
            self.clone().publish_payload_attestations_batch(slot, duties),
            "payload_attestation_batch_publish",
        );

`publish_payload_attestations`:
First, we query the beacon node for `PayloadAttestationData` for a given slot:

   beacon_node.get_validator_payload_attestation_data(slot)

Once returned, we can have each of our validators in the VC who are in a ptc committee for this slot sign in parallel:

let signing_futures = duties.iter().map(|duty| {
 match service.sign_payload_attestation_data(&pubkey, &attestation_data, slot).await
{
    Ok(signature) =>
    Some(PayloadAttestationMessage {
            validator_index,
            data: attestation_data,
            signature,
                                    }),
...
    
 let signed_attestations: Vec<PayloadAttestationMessage> = join_all(signing_futures)

Finally, we `POST` the signed attestations to the BN in one call:

beacon_node.post_beacon_pool_payload_attestations(signed_attestations, fork_name)

I believe this approach provides a variety of optimizations:
- never block the main thread from trying to produce payload attestations 3/4 into a block
- single `GET` and `POST` calls to the BN regardless of how many validators in your VC are in a PTC that slot
- parallelized payload attestation signing as to reduce the footprint of signing from impacting the ptc hot path.

The implementation for the `payload_attestation_service.rs` can be seen [here](https://github.com/shane-moore/lighthouse/compare/gloas-vc-ptc-duty...gloas-vc-payload-attestation-service).