diff --git a/packages/api/src/beacon/routes/validator.ts b/packages/api/src/beacon/routes/validator.ts index 12f8fb97139f..458255b5d86f 100644 --- a/packages/api/src/beacon/routes/validator.ts +++ b/packages/api/src/beacon/routes/validator.ts @@ -16,6 +16,7 @@ import { ValidatorIndex, RootHex, StringType, + SubcommitteeIndex, Wei, } from "@lodestar/types"; import {ApiClientResponse} from "../../interfaces.js"; @@ -97,6 +98,32 @@ export type SyncDuty = { validatorSyncCommitteeIndices: number[]; }; +/** + * From https://github.com/ethereum/beacon-APIs/pull/224 + */ +export type BeaconCommitteeSelection = { + /** Index of the validator */ + validatorIndex: ValidatorIndex; + /** The slot at which a validator is assigned to attest */ + slot: Slot; + /** The `slot_signature` calculated by the validator for the upcoming attestation slot */ + selectionProof: BLSSignature; +}; + +/** + * From https://github.com/ethereum/beacon-APIs/pull/224 + */ +export type SyncCommitteeSelection = { + /** Index of the validator */ + validatorIndex: ValidatorIndex; + /** The slot at which validator is assigned to produce a sync committee contribution */ + slot: Slot; + /** SubcommitteeIndex to which the validator is assigned */ + subcommitteeIndex: SubcommitteeIndex; + /** The `slot_signature` calculated by the validator for the upcoming sync committee slot */ + selectionProof: BLSSignature; +}; + export type LivenessResponseData = { index: ValidatorIndex; epoch: Epoch; @@ -303,6 +330,52 @@ export type Api = { proposers: ProposerPreparationData[] ): Promise>; + /** + * Determine if a distributed validator has been selected to aggregate attestations + * + * This endpoint is implemented by a distributed validator middleware client to exchange + * partial beacon committee selection proofs for combined/aggregated selection proofs to allow + * a validator client to correctly determine if one of its validators has been selected to + * perform an aggregation duty in this slot. + * + * Note that this endpoint is not implemented by the beacon node and will return a 501 error + * + * @param requestBody An array of partial beacon committee selection proofs + * @returns An array of threshold aggregated beacon committee selection proofs + * @throws ApiError + */ + submitBeaconCommitteeSelections( + selections: BeaconCommitteeSelection[] + ): Promise< + ApiClientResponse< + {[HttpStatusCode.OK]: {data: BeaconCommitteeSelection[]}}, + HttpStatusCode.BAD_REQUEST | HttpStatusCode.NOT_IMPLEMENTED | HttpStatusCode.SERVICE_UNAVAILABLE + > + >; + + /** + * Determine if a distributed validator has been selected to make a sync committee contribution + * + * This endpoint is implemented by a distributed validator middleware client to exchange + * partial sync committee selection proofs for combined/aggregated selection proofs to allow + * a validator client to correctly determine if one of its validators has been selected to + * perform a sync committee contribution (sync aggregation) duty in this slot. + * + * Note that this endpoint is not implemented by the beacon node and will return a 501 error + * + * @param requestBody An array of partial sync committee selection proofs + * @returns An array of threshold aggregated sync committee selection proofs + * @throws ApiError + */ + submitSyncCommitteeSelections( + selections: SyncCommitteeSelection[] + ): Promise< + ApiClientResponse< + {[HttpStatusCode.OK]: {data: SyncCommitteeSelection[]}}, + HttpStatusCode.BAD_REQUEST | HttpStatusCode.NOT_IMPLEMENTED | HttpStatusCode.SERVICE_UNAVAILABLE + > + >; + /** Returns validator indices that have been observed to be active on the network */ getLiveness( indices: ValidatorIndex[], @@ -332,6 +405,8 @@ export const routesData: RoutesData = { prepareBeaconCommitteeSubnet: {url: "/eth/v1/validator/beacon_committee_subscriptions", method: "POST"}, prepareSyncCommitteeSubnets: {url: "/eth/v1/validator/sync_committee_subscriptions", method: "POST"}, prepareBeaconProposer: {url: "/eth/v1/validator/prepare_beacon_proposer", method: "POST"}, + submitBeaconCommitteeSelections: {url: "/eth/v1/validator/beacon_committee_selections", method: "POST"}, + submitSyncCommitteeSelections: {url: "/eth/v1/validator/sync_committee_selections", method: "POST"}, getLiveness: {url: "/eth/v1/validator/liveness", method: "GET"}, registerValidator: {url: "/eth/v1/validator/register_validator", method: "POST"}, }; @@ -352,10 +427,31 @@ export type ReqTypes = { prepareBeaconCommitteeSubnet: {body: unknown}; prepareSyncCommitteeSubnets: {body: unknown}; prepareBeaconProposer: {body: unknown}; + submitBeaconCommitteeSelections: {body: unknown}; + submitSyncCommitteeSelections: {body: unknown}; getLiveness: {query: {indices: ValidatorIndex[]; epoch: Epoch}}; registerValidator: {body: unknown}; }; +const BeaconCommitteeSelection = new ContainerType( + { + validatorIndex: ssz.ValidatorIndex, + slot: ssz.Slot, + selectionProof: ssz.BLSSignature, + }, + {jsonCase: "eth2"} +); + +const SyncCommitteeSelection = new ContainerType( + { + validatorIndex: ssz.ValidatorIndex, + slot: ssz.Slot, + subcommitteeIndex: ssz.SubcommitteeIndex, + selectionProof: ssz.BLSSignature, + }, + {jsonCase: "eth2"} +); + export function getReqSerializers(): ReqSerializers { const BeaconCommitteeSubscription = new ContainerType( { @@ -461,6 +557,14 @@ export function getReqSerializers(): ReqSerializers { ], schema: {body: Schema.ObjectArray}, }, + submitBeaconCommitteeSelections: { + writeReq: (items) => ({body: ArrayOf(BeaconCommitteeSelection).toJson(items)}), + parseReq: () => [[]], + }, + submitSyncCommitteeSelections: { + writeReq: (items) => ({body: ArrayOf(SyncCommitteeSelection).toJson(items)}), + parseReq: () => [[]], + }, getLiveness: { writeReq: (indices, epoch) => ({query: {indices, epoch}}), parseReq: ({query}) => [query.indices, query.epoch], @@ -532,6 +636,8 @@ export function getReturnTypes(): ReturnTypes { produceAttestationData: ContainerData(ssz.phase0.AttestationData), produceSyncCommitteeContribution: ContainerData(ssz.altair.SyncCommitteeContribution), getAggregatedAttestation: ContainerData(ssz.phase0.Attestation), + submitBeaconCommitteeSelections: ContainerData(ArrayOf(BeaconCommitteeSelection)), + submitSyncCommitteeSelections: ContainerData(ArrayOf(SyncCommitteeSelection)), getLiveness: jsonType("snake"), }; } diff --git a/packages/api/test/unit/beacon/testData/validator.ts b/packages/api/test/unit/beacon/testData/validator.ts index e6939614ec73..90792f467d7f 100644 --- a/packages/api/test/unit/beacon/testData/validator.ts +++ b/packages/api/test/unit/beacon/testData/validator.ts @@ -6,6 +6,7 @@ import {GenericServerTestCases} from "../../../utils/genericServerTest.js"; const ZERO_HASH = Buffer.alloc(32, 0); const ZERO_HASH_HEX = "0x" + ZERO_HASH.toString("hex"); const randaoReveal = Buffer.alloc(96, 1); +const selectionProof = Buffer.alloc(96, 1); const graffiti = "a".repeat(32); export const testData: GenericServerTestCases = { @@ -90,6 +91,14 @@ export const testData: GenericServerTestCases = { args: [[{validatorIndex: "1", feeRecipient: "0xa94f5374fce5edbc8e2a8697c15331677e6ebf0b"}]], res: undefined, }, + submitBeaconCommitteeSelections: { + args: [[]], + res: {data: [{validatorIndex: 1, slot: 2, selectionProof}]}, + }, + submitSyncCommitteeSelections: { + args: [[]], + res: {data: [{validatorIndex: 1, slot: 2, subcommitteeIndex: 3, selectionProof}]}, + }, getLiveness: { args: [[0], 0], res: {data: []}, diff --git a/packages/beacon-node/src/api/impl/errors.ts b/packages/beacon-node/src/api/impl/errors.ts index d520013a53de..2830a0438f3b 100644 --- a/packages/beacon-node/src/api/impl/errors.ts +++ b/packages/beacon-node/src/api/impl/errors.ts @@ -31,3 +31,11 @@ export class NodeIsSyncing extends ApiError { super(503, `Node is syncing - ${statusMsg}`); } } + +// Error thrown by beacon node APIs that are only supported by distributed validator middleware clients +// For example https://github.com/ethereum/beacon-APIs/blob/f087fbf2764e657578a6c29bdf0261b36ee8db1e/apis/validator/beacon_committee_selections.yaml +export class OnlySupportedByDVT extends ApiError { + constructor() { + super(501, "Only supported by distributed validator middleware clients"); + } +} diff --git a/packages/beacon-node/src/api/impl/validator/index.ts b/packages/beacon-node/src/api/impl/validator/index.ts index fa7c27864e49..f43b0d5c03fb 100644 --- a/packages/beacon-node/src/api/impl/validator/index.ts +++ b/packages/beacon-node/src/api/impl/validator/index.ts @@ -19,7 +19,7 @@ import {ZERO_HASH} from "../../../constants/index.js"; import {SyncState} from "../../../sync/index.js"; import {isOptimisticBlock} from "../../../util/forkChoice.js"; import {toGraffitiBuffer} from "../../../util/graffiti.js"; -import {ApiError, NodeIsSyncing} from "../errors.js"; +import {ApiError, NodeIsSyncing, OnlySupportedByDVT} from "../errors.js"; import {validateSyncCommitteeGossipContributionAndProof} from "../../../chain/validation/syncCommitteeContributionAndProof.js"; import {CommitteeSubscription} from "../../../network/subnets/index.js"; import {ApiModules} from "../types.js"; @@ -698,6 +698,14 @@ export function getValidatorApi({ await chain.updateBeaconProposerData(chain.clock.currentEpoch, proposers); }, + async submitBeaconCommitteeSelections() { + throw new OnlySupportedByDVT(); + }, + + async submitSyncCommitteeSelections() { + throw new OnlySupportedByDVT(); + }, + async getLiveness(indices: ValidatorIndex[], epoch: Epoch) { if (indices.length === 0) { return { diff --git a/packages/validator/src/services/attestation.ts b/packages/validator/src/services/attestation.ts index 31d21c1cba0b..7e9affb65a41 100644 --- a/packages/validator/src/services/attestation.ts +++ b/packages/validator/src/services/attestation.ts @@ -1,7 +1,7 @@ -import {phase0, Slot, ssz} from "@lodestar/types"; -import {computeEpochAtSlot} from "@lodestar/state-transition"; +import {BLSSignature, phase0, Slot, ssz} from "@lodestar/types"; +import {computeEpochAtSlot, isAggregatorFromCommitteeLength} from "@lodestar/state-transition"; import {sleep} from "@lodestar/utils"; -import {Api, ApiError} from "@lodestar/api"; +import {Api, ApiError, routes} from "@lodestar/api"; import {toHexString} from "@chainsafe/ssz"; import {IClock, LoggerVc} from "../util/index.js"; import {PubkeyHex} from "../types.js"; @@ -15,6 +15,7 @@ import {ValidatorEventEmitter} from "./emitter.js"; export type AttestationServiceOpts = { afterBlockDelaySlotFraction?: number; disableAttestationGrouping?: boolean; + distributedAggregationSelection?: boolean; }; /** @@ -42,7 +43,9 @@ export class AttestationService { private readonly metrics: Metrics | null, private readonly opts?: AttestationServiceOpts ) { - this.dutiesService = new AttestationDutiesService(logger, api, clock, validatorStore, chainHeadTracker, metrics); + this.dutiesService = new AttestationDutiesService(logger, api, clock, validatorStore, chainHeadTracker, metrics, { + distributedAggregationSelection: opts?.distributedAggregationSelection, + }); // At most every slot, check existing duties from AttestationDutiesService and run tasks clock.runEverySlot(this.runAttestationTasks); @@ -59,6 +62,18 @@ export class AttestationService { return; } + if (this.opts?.distributedAggregationSelection) { + // Validator in distributed cluster only has a key share, not the full private key. + // The partial selection proofs must be exchanged for combined selection proofs by + // calling submitBeaconCommitteeSelections on the distributed validator middleware client. + // This will run in parallel to other attestation tasks but must be finished before starting + // attestation aggregation as it is required to correctly determine if validator is aggregator + // and to produce a AggregateAndProof that can be threshold aggregated by the middleware client. + this.runDistributedAggregationSelectionTasks(duties, slot, signal).catch((e) => + this.logger.error("Error on attestation aggregation selection", {slot}, e) + ); + } + // A validator should create and broadcast the attestation to the associated attestation subnet when either // (a) the validator has received a valid block from the expected block proposer for the assigned slot or // (b) one-third of the slot has transpired (SECONDS_PER_SLOT / 3 seconds after the start of slot) -- whichever comes first. @@ -276,4 +291,94 @@ export class AttestationService { } } } + + /** + * Performs additional attestation aggregation tasks required if validator is part of distributed cluster + * + * 1. Exchange partial for combined selection proofs + * 2. Determine validators that should aggregate attestations + * 3. Mutate duty objects to set selection proofs for aggregators + * 4. Resubscribe validators as aggregators on beacon committee subnets + * + * See https://docs.google.com/document/d/1q9jOTPcYQa-3L8luRvQJ-M0eegtba4Nmon3dpO79TMk/mobilebasic + */ + private async runDistributedAggregationSelectionTasks( + duties: AttDutyAndProof[], + slot: number, + signal: AbortSignal + ): Promise { + const partialSelections: routes.validator.BeaconCommitteeSelection[] = duties.map( + ({duty, partialSelectionProof}) => ({ + validatorIndex: duty.validatorIndex, + slot, + selectionProof: partialSelectionProof as BLSSignature, + }) + ); + + this.logger.debug("Submitting partial beacon committee selection proofs", {slot, count: partialSelections.length}); + + const res = await Promise.race([ + this.api.validator + .submitBeaconCommitteeSelections(partialSelections) + .catch((e) => this.logger.error("Error on submitBeaconCommitteeSelections", {slot}, e)), + // Exit attestation aggregation flow if there is no response after 1/3 of slot as + // beacon node would likely not have enough time to prepare an aggregate attestation. + // Note that the aggregations flow is not explicitly exited but rather will be skipped + // due to the fact that calculation of `is_aggregator` in AttestationDutiesService is not done + // and selectionProof is set to null, meaning no validator will be considered an aggregator. + sleep(this.clock.msToSlot(slot + 1 / 3), signal), + ]); + + if (!res) { + throw new Error("submitBeaconCommitteeSelections did not resolve after 1/3 of slot"); + } + ApiError.assert(res, "Error receiving combined selection proofs"); + + const combinedSelections = res.response.data; + this.logger.debug("Received combined beacon committee selection proofs", {slot, count: combinedSelections.length}); + + const beaconCommitteeSubscriptions: routes.validator.BeaconCommitteeSubscription[] = []; + + for (const dutyAndProof of duties) { + const {validatorIndex, committeeIndex, committeeLength, committeesAtSlot} = dutyAndProof.duty; + const logCtxValidator = {slot, index: committeeIndex, validatorIndex}; + + const combinedSelection = combinedSelections.find((s) => s.validatorIndex === validatorIndex && s.slot === slot); + + if (!combinedSelection) { + this.logger.warn("Did not receive combined beacon committee selection proof", logCtxValidator); + continue; + } + + const isAggregator = isAggregatorFromCommitteeLength(committeeLength, combinedSelection.selectionProof); + + if (isAggregator) { + // Update selection proof by mutating duty object + dutyAndProof.selectionProof = combinedSelection.selectionProof; + + // Only push subnet subscriptions with `isAggregator=true` as all validators + // with duties for slot are already subscribed to subnets with `isAggregator=false`. + beaconCommitteeSubscriptions.push({ + validatorIndex, + committeesAtSlot, + committeeIndex, + slot, + isAggregator, + }); + this.logger.debug("Resubscribing validator as aggregator on beacon committee subnet", logCtxValidator); + } + } + + // If there are any subscriptions with aggregators, push them out to the beacon node. + if (beaconCommitteeSubscriptions.length > 0) { + ApiError.assert( + await this.api.validator.prepareBeaconCommitteeSubnet(beaconCommitteeSubscriptions), + "Failed to resubscribe to beacon committee subnets" + ); + this.logger.debug("Resubscribed validators as aggregators on beacon committee subnets", { + slot, + count: beaconCommitteeSubscriptions.length, + }); + } + } } diff --git a/packages/validator/src/services/attestationDuties.ts b/packages/validator/src/services/attestationDuties.ts index 8b419f8da792..916c02f91994 100644 --- a/packages/validator/src/services/attestationDuties.ts +++ b/packages/validator/src/services/attestationDuties.ts @@ -26,11 +26,17 @@ export type AttDutyAndProof = { duty: routes.validator.AttesterDuty; /** This value is only set to not null if the proof indicates that the validator is an aggregator. */ selectionProof: BLSSignature | null; + /** This value will only be set if validator is part of distributed cluster and only has a key share */ + partialSelectionProof?: BLSSignature; }; // To assist with readability type AttDutiesAtEpoch = {dependentRoot: RootHex; dutiesByIndex: Map}; +type AttestationDutiesServiceOpts = { + distributedAggregationSelection?: boolean; +}; + export class AttestationDutiesService { /** Maps a validator public key to their duties for each epoch */ private readonly dutiesByIndexByEpoch = new Map(); @@ -46,7 +52,8 @@ export class AttestationDutiesService { private clock: IClock, private readonly validatorStore: ValidatorStore, chainHeadTracker: ChainHeaderTracker, - private readonly metrics: Metrics | null + private readonly metrics: Metrics | null, + private readonly opts?: AttestationDutiesServiceOpts ) { // Running this task every epoch is safe since a re-org of two epochs is very unlikely // TODO: If the re-org event is reliable consider re-running then @@ -326,6 +333,15 @@ export class AttestationDutiesService { private async getDutyAndProof(duty: routes.validator.AttesterDuty): Promise { const selectionProof = await this.validatorStore.signAttestationSelectionProof(duty.pubkey, duty.slot); + + if (this.opts?.distributedAggregationSelection) { + // Validator in distributed cluster only has a key share, not the full private key. + // Passing a partial selection proof to `is_aggregator` would produce incorrect result. + // AttestationService will exchange partial for combined selection proofs retrieved from + // distributed validator middleware client and determine aggregators at beginning of every slot. + return {duty, selectionProof: null, partialSelectionProof: selectionProof}; + } + const isAggregator = isAggregatorFromCommitteeLength(duty.committeeLength, selectionProof); return { diff --git a/packages/validator/src/services/syncCommittee.ts b/packages/validator/src/services/syncCommittee.ts index 362fe78ac655..adbda3231697 100644 --- a/packages/validator/src/services/syncCommittee.ts +++ b/packages/validator/src/services/syncCommittee.ts @@ -1,8 +1,8 @@ import {ChainForkConfig} from "@lodestar/config"; -import {Slot, CommitteeIndex, altair, Root} from "@lodestar/types"; +import {Slot, CommitteeIndex, altair, Root, BLSSignature} from "@lodestar/types"; import {sleep} from "@lodestar/utils"; -import {computeEpochAtSlot} from "@lodestar/state-transition"; -import {Api, ApiError} from "@lodestar/api"; +import {computeEpochAtSlot, isSyncCommitteeAggregator} from "@lodestar/state-transition"; +import {Api, ApiError, routes} from "@lodestar/api"; import {IClock, LoggerVc} from "../util/index.js"; import {PubkeyHex} from "../types.js"; import {Metrics} from "../metrics.js"; @@ -12,8 +12,9 @@ import {groupSyncDutiesBySubcommitteeIndex, SubcommitteeDuty} from "./utils.js"; import {ChainHeaderTracker} from "./chainHeaderTracker.js"; import {ValidatorEventEmitter} from "./emitter.js"; -type SyncCommitteeServiceOpts = { +export type SyncCommitteeServiceOpts = { scAfterBlockDelaySlotFraction?: number; + distributedAggregationSelection?: boolean; }; /** @@ -33,7 +34,9 @@ export class SyncCommitteeService { private readonly metrics: Metrics | null, private readonly opts?: SyncCommitteeServiceOpts ) { - this.dutiesService = new SyncCommitteeDutiesService(config, logger, api, clock, validatorStore, metrics); + this.dutiesService = new SyncCommitteeDutiesService(config, logger, api, clock, validatorStore, metrics, { + distributedAggregationSelection: opts?.distributedAggregationSelection, + }); // At most every slot, check existing duties from SyncCommitteeDutiesService and run tasks clock.runEverySlot(this.runSyncCommitteeTasks); @@ -56,6 +59,18 @@ export class SyncCommitteeService { return; } + if (this.opts?.distributedAggregationSelection) { + // Validator in distributed cluster only has a key share, not the full private key. + // The partial selection proofs must be exchanged for combined selection proofs by + // calling submitSyncCommitteeSelections on the distributed validator middleware client. + // This will run in parallel to other sync committee tasks but must be finished before starting + // sync committee contributions as it is required to correctly determine if validator is aggregator + // and to produce a ContributionAndProof that can be threshold aggregated by the middleware client. + this.runDistributedAggregationSelectionTasks(dutiesAtSlot, slot, signal).catch((e) => + this.logger.error("Error on sync committee aggregation selection", {slot}, e) + ); + } + // unlike Attestation, SyncCommitteeSignature could be published asap // especially with lodestar, it's very busy at 1/3 of slot // see https://github.com/ChainSafe/lodestar/issues/4608 @@ -214,4 +229,84 @@ export class SyncCommitteeService { } } } + + /** + * Performs additional sync committee contribution tasks required if validator is part of distributed cluster + * + * 1. Exchange partial for combined selection proofs + * 2. Determine validators that should produce sync committee contribution + * 3. Mutate duty objects to set selection proofs for aggregators + * + * See https://docs.google.com/document/d/1q9jOTPcYQa-3L8luRvQJ-M0eegtba4Nmon3dpO79TMk/mobilebasic + */ + private async runDistributedAggregationSelectionTasks( + duties: SyncDutyAndProofs[], + slot: number, + signal: AbortSignal + ): Promise { + const partialSelections: routes.validator.SyncCommitteeSelection[] = []; + + for (const {duty, selectionProofs} of duties) { + const validatorSelections: routes.validator.SyncCommitteeSelection[] = selectionProofs.map( + ({subcommitteeIndex, partialSelectionProof}) => ({ + validatorIndex: duty.validatorIndex, + slot, + subcommitteeIndex, + selectionProof: partialSelectionProof as BLSSignature, + }) + ); + partialSelections.push(...validatorSelections); + } + + this.logger.debug("Submitting partial sync committee selection proofs", {slot, count: partialSelections.length}); + + const res = await Promise.race([ + this.api.validator + .submitSyncCommitteeSelections(partialSelections) + .catch((e) => this.logger.error("Error on submitSyncCommitteeSelections", {slot}, e)), + // Exit sync committee contributions flow if there is no response after 2/3 of slot. + // This is in contrast to attestations aggregations flow which is already exited at 1/3 of the slot + // because for sync committee is not required to resubscribe to subnets as beacon node will assume + // validator always aggregates. This allows us to wait until we have to produce sync committee contributions. + // Note that the sync committee contributions flow is not explicitly exited but rather will be skipped + // due to the fact that calculation of `is_sync_committee_aggregator` in SyncCommitteeDutiesService is not done + // and selectionProof is set to null, meaning no validator will be considered an aggregator. + sleep(this.clock.msToSlot(slot + 2 / 3), signal), + ]); + + if (!res) { + throw new Error("submitSyncCommitteeSelections did not resolve after 2/3 of slot"); + } + ApiError.assert(res, "Error receiving combined selection proofs"); + + const combinedSelections = res.response.data; + this.logger.debug("Received combined sync committee selection proofs", {slot, count: combinedSelections.length}); + + for (const dutyAndProofs of duties) { + const {validatorIndex, subnets} = dutyAndProofs.duty; + + for (const subnet of subnets) { + const logCtxValidator = {slot, index: subnet, validatorIndex}; + + const combinedSelection = combinedSelections.find( + (s) => s.validatorIndex === validatorIndex && s.slot === slot && s.subcommitteeIndex === subnet + ); + + if (!combinedSelection) { + this.logger.warn("Did not receive combined sync committee selection proof", logCtxValidator); + continue; + } + + const isAggregator = isSyncCommitteeAggregator(combinedSelection.selectionProof); + + if (isAggregator) { + const selectionProofObject = dutyAndProofs.selectionProofs.find((p) => p.subcommitteeIndex === subnet); + if (selectionProofObject) { + // Update selection proof by mutating proof objects in duty object + selectionProofObject.selectionProof = combinedSelection.selectionProof; + } + } + } + } + } } diff --git a/packages/validator/src/services/syncCommitteeDuties.ts b/packages/validator/src/services/syncCommitteeDuties.ts index 9e9df14d9a41..764cafa0e3e1 100644 --- a/packages/validator/src/services/syncCommitteeDuties.ts +++ b/packages/validator/src/services/syncCommitteeDuties.ts @@ -42,6 +42,8 @@ export type SyncDutySubnet = { export type SyncSelectionProof = { /** This value is only set to not null if the proof indicates that the validator is an aggregator. */ selectionProof: BLSSignature | null; + /** This value will only be set if validator is part of distributed cluster and only has a key share */ + partialSelectionProof?: BLSSignature; subcommitteeIndex: number; }; @@ -60,6 +62,10 @@ export type SyncDutyAndProofs = { // To assist with readability type DutyAtPeriod = {duty: SyncDutySubnet}; +type SyncCommitteeDutiesServiceOpts = { + distributedAggregationSelection?: boolean; +}; + /** * Validators are part of a static long (~27h) sync committee, and part of static subnets. * However, the isAggregator role changes per slot. @@ -74,7 +80,8 @@ export class SyncCommitteeDutiesService { private readonly api: Api, clock: IClock, private readonly validatorStore: ValidatorStore, - metrics: Metrics | null + metrics: Metrics | null, + private readonly opts?: SyncCommitteeDutiesServiceOpts ) { // Running this task every epoch is safe since a re-org of many epochs is very unlikely // TODO: If the re-org event is reliable consider re-running then @@ -285,11 +292,23 @@ export class SyncCommitteeDutiesService { const dutiesAndProofs: SyncSelectionProof[] = []; for (const subnet of duty.subnets) { const selectionProof = await this.validatorStore.signSyncCommitteeSelectionProof(duty.pubkey, slot, subnet); - dutiesAndProofs.push({ - // selectionProof === null is used to check if is aggregator - selectionProof: isSyncCommitteeAggregator(selectionProof) ? selectionProof : null, - subcommitteeIndex: subnet, - }); + if (this.opts?.distributedAggregationSelection) { + // Validator in distributed cluster only has a key share, not the full private key. + // Passing a partial selection proof to `is_sync_committee_aggregator` would produce incorrect result. + // SyncCommitteeService will exchange partial for combined selection proofs retrieved from + // distributed validator middleware client and determine aggregators at beginning of every slot. + dutiesAndProofs.push({ + selectionProof: null, + partialSelectionProof: selectionProof, + subcommitteeIndex: subnet, + }); + } else { + dutiesAndProofs.push({ + // selectionProof === null is used to check if is aggregator + selectionProof: isSyncCommitteeAggregator(selectionProof) ? selectionProof : null, + subcommitteeIndex: subnet, + }); + } } return dutiesAndProofs; } diff --git a/packages/validator/src/validator.ts b/packages/validator/src/validator.ts index d52e02a5bf82..4700ad2a9dda 100644 --- a/packages/validator/src/validator.ts +++ b/packages/validator/src/validator.ts @@ -129,6 +129,7 @@ export class Validator { { afterBlockDelaySlotFraction: opts.afterBlockDelaySlotFraction, disableAttestationGrouping: opts.disableAttestationGrouping || opts.distributed, + distributedAggregationSelection: opts.distributed, } ); @@ -141,7 +142,10 @@ export class Validator { emitter, chainHeaderTracker, metrics, - {scAfterBlockDelaySlotFraction: opts.scAfterBlockDelaySlotFraction} + { + scAfterBlockDelaySlotFraction: opts.scAfterBlockDelaySlotFraction, + distributedAggregationSelection: opts.distributed, + } ); this.config = config; diff --git a/packages/validator/test/unit/services/attestation.test.ts b/packages/validator/test/unit/services/attestation.test.ts index a53314dd359e..754b9a133ff7 100644 --- a/packages/validator/test/unit/services/attestation.test.ts +++ b/packages/validator/test/unit/services/attestation.test.ts @@ -3,7 +3,7 @@ import sinon from "sinon"; import bls from "@chainsafe/bls"; import {toHexString} from "@chainsafe/ssz"; import {ssz} from "@lodestar/types"; -import {HttpStatusCode} from "@lodestar/api"; +import {HttpStatusCode, routes} from "@lodestar/api"; import {AttestationService, AttestationServiceOpts} from "../../../src/services/attestation.js"; import {AttDutyAndProof} from "../../../src/services/attestationDuties.js"; import {ValidatorStore} from "../../../src/services/validatorStore.js"; @@ -42,104 +42,157 @@ describe("AttestationService", function () { sandbox.resetHistory(); }); - context("With attestation grouping enabled", () => { - const opts: AttestationServiceOpts = {disableAttestationGrouping: false}; - - it("Should produce, sign, and publish an attestation + aggregate", async () => { - await testAttestationTasks(opts); - }); - }); - - context("With attestation grouping disabled", () => { - const opts: AttestationServiceOpts = {disableAttestationGrouping: true}; - - it("Should produce, sign, and publish an attestation + aggregate", async () => { - await testAttestationTasks(opts); - }); - }); - - async function testAttestationTasks(opts?: AttestationServiceOpts): Promise { - const clock = new ClockMock(); - const attestationService = new AttestationService( - loggerVc, - api, - clock, - validatorStore, - emitter, - chainHeadTracker, - null, - opts - ); - - const attestation = ssz.phase0.Attestation.defaultValue(); - const aggregate = ssz.phase0.SignedAggregateAndProof.defaultValue(); - const duties: AttDutyAndProof[] = [ - { - duty: { - slot: 0, - committeeIndex: attestation.data.index, - committeeLength: 120, - committeesAtSlot: 120, - validatorCommitteeIndex: 1, - validatorIndex: 0, - pubkey: pubkeys[0], - }, - selectionProof: ZERO_HASH, - }, - ]; - - // Return empty replies to duties service - api.beacon.getStateValidators.resolves({ - response: {executionOptimistic: false, data: []}, - ok: true, - status: HttpStatusCode.OK, - }); - api.validator.getAttesterDuties.resolves({ - response: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false, data: []}, - ok: true, - status: HttpStatusCode.OK, - }); - - // Mock duties service to return some duties directly - attestationService["dutiesService"].getDutiesAtSlot = sinon.stub().returns(duties); - - // Mock beacon's attestation and aggregates endpoints - - api.validator.produceAttestationData.resolves({ - response: {data: attestation.data}, - ok: true, - status: HttpStatusCode.OK, - }); - api.validator.getAggregatedAttestation.resolves({ - response: {data: attestation}, - ok: true, - status: HttpStatusCode.OK, + const testContexts: [string, AttestationServiceOpts][] = [ + ["With default configuration", {}], + ["With attestation grouping disabled", {disableAttestationGrouping: true}], + ["With distributed aggregation selection enabled", {distributedAggregationSelection: true}], + ]; + + for (const [title, opts] of testContexts) { + context(title, () => { + it("Should produce, sign, and publish an attestation + aggregate", async () => { + const clock = new ClockMock(); + const attestationService = new AttestationService( + loggerVc, + api, + clock, + validatorStore, + emitter, + chainHeadTracker, + null, + opts + ); + + const attestation = ssz.phase0.Attestation.defaultValue(); + const aggregate = ssz.phase0.SignedAggregateAndProof.defaultValue(); + const duties: AttDutyAndProof[] = [ + { + duty: { + slot: 0, + committeeIndex: attestation.data.index, + committeeLength: 120, + committeesAtSlot: 120, + validatorCommitteeIndex: 1, + validatorIndex: 0, + pubkey: pubkeys[0], + }, + selectionProof: opts.distributedAggregationSelection ? null : ZERO_HASH, + partialSelectionProof: opts.distributedAggregationSelection ? ZERO_HASH : undefined, + }, + ]; + + // Return empty replies to duties service + api.beacon.getStateValidators.resolves({ + response: {executionOptimistic: false, data: []}, + ok: true, + status: HttpStatusCode.OK, + }); + api.validator.getAttesterDuties.resolves({ + response: {dependentRoot: ZERO_HASH_HEX, executionOptimistic: false, data: []}, + ok: true, + status: HttpStatusCode.OK, + }); + + // Mock duties service to return some duties directly + attestationService["dutiesService"].getDutiesAtSlot = sinon.stub().returns(duties); + + // Mock beacon's attestation and aggregates endpoints + + api.validator.produceAttestationData.resolves({ + response: {data: attestation.data}, + ok: true, + status: HttpStatusCode.OK, + }); + api.validator.getAggregatedAttestation.resolves({ + response: {data: attestation}, + ok: true, + status: HttpStatusCode.OK, + }); + api.beacon.submitPoolAttestations.resolves({ + response: undefined, + ok: true, + status: HttpStatusCode.OK, + }); + api.validator.publishAggregateAndProofs.resolves({ + response: undefined, + ok: true, + status: HttpStatusCode.OK, + }); + + if (opts.distributedAggregationSelection) { + // Mock distributed validator middleware client selections endpoint + // and return a selection proof that passes `is_aggregator` test + api.validator.submitBeaconCommitteeSelections.resolves({ + response: {data: [{validatorIndex: 0, slot: 0, selectionProof: Buffer.alloc(1, 0x10)}]}, + ok: true, + status: HttpStatusCode.OK, + }); + // Accept all subscriptions + api.validator.prepareBeaconCommitteeSubnet.resolves({ + response: undefined, + ok: true, + status: HttpStatusCode.OK, + }); + } + + // Mock signing service + validatorStore.signAttestation.resolves(attestation); + validatorStore.signAggregateAndProof.resolves(aggregate); + + // Trigger clock onSlot for slot 0 + await clock.tickSlotFns(0, controller.signal); + + if (opts.distributedAggregationSelection) { + // Must submit partial beacon committee selection proof based on duty + const selection: routes.validator.BeaconCommitteeSelection = { + validatorIndex: 0, + slot: 0, + selectionProof: ZERO_HASH, + }; + expect(api.validator.submitBeaconCommitteeSelections.callCount).to.equal( + 1, + "submitBeaconCommitteeSelections() must be called once" + ); + expect(api.validator.submitBeaconCommitteeSelections.getCall(0).args).to.deep.equal( + [[selection]], // 1 arg, = selection[] + "wrong submitBeaconCommitteeSelections() args" + ); + + // Must resubscribe validator as aggregator on beacon committee subnet + const subscription: routes.validator.BeaconCommitteeSubscription = { + validatorIndex: 0, + committeeIndex: 0, + committeesAtSlot: 120, + slot: 0, + isAggregator: true, + }; + expect(api.validator.prepareBeaconCommitteeSubnet.callCount).to.equal( + 1, + "prepareBeaconCommitteeSubnet() must be called once" + ); + expect(api.validator.prepareBeaconCommitteeSubnet.getCall(0).args).to.deep.equal( + [[subscription]], // 1 arg, = subscription[] + "wrong prepareBeaconCommitteeSubnet() args" + ); + } + + // Must submit the attestation received through produceAttestationData() + expect(api.beacon.submitPoolAttestations.callCount).to.equal(1, "submitAttestations() must be called once"); + expect(api.beacon.submitPoolAttestations.getCall(0).args).to.deep.equal( + [[attestation]], // 1 arg, = attestation[] + "wrong submitAttestations() args" + ); + + // Must submit the aggregate received through getAggregatedAttestation() then createAndSignAggregateAndProof() + expect(api.validator.publishAggregateAndProofs.callCount).to.equal( + 1, + "publishAggregateAndProofs() must be called once" + ); + expect(api.validator.publishAggregateAndProofs.getCall(0).args).to.deep.equal( + [[aggregate]], // 1 arg, = aggregate[] + "wrong publishAggregateAndProofs() args" + ); + }); }); - api.beacon.submitPoolAttestations.resolves(); - api.validator.publishAggregateAndProofs.resolves(); - - // Mock signing service - validatorStore.signAttestation.resolves(attestation); - validatorStore.signAggregateAndProof.resolves(aggregate); - - // Trigger clock onSlot for slot 0 - await clock.tickSlotFns(0, controller.signal); - - // Must submit the attestation received through produceAttestationData() - expect(api.beacon.submitPoolAttestations.callCount).to.equal(1, "submitAttestations() must be called once"); - expect(api.beacon.submitPoolAttestations.getCall(0).args).to.deep.equal( - [[attestation]], // 1 arg, = attestation[] - "wrong submitAttestations() args" - ); - - // Must submit the aggregate received through getAggregatedAttestation() then createAndSignAggregateAndProof() - expect(api.validator.publishAggregateAndProofs.callCount).to.equal( - 1, - "publishAggregateAndProofs() must be called once" - ); - expect(api.validator.publishAggregateAndProofs.getCall(0).args).to.deep.equal( - [[aggregate]], // 1 arg, = aggregate[] - "wrong publishAggregateAndProofs() args" - ); } }); diff --git a/packages/validator/test/unit/services/syncCommittee.test.ts b/packages/validator/test/unit/services/syncCommittee.test.ts index c7815570d393..9316f11eb483 100644 --- a/packages/validator/test/unit/services/syncCommittee.test.ts +++ b/packages/validator/test/unit/services/syncCommittee.test.ts @@ -5,8 +5,8 @@ import {toHexString} from "@chainsafe/ssz"; import {createChainForkConfig} from "@lodestar/config"; import {config as mainnetConfig} from "@lodestar/config/default"; import {ssz} from "@lodestar/types"; -import {HttpStatusCode} from "@lodestar/api"; -import {SyncCommitteeService} from "../../../src/services/syncCommittee.js"; +import {HttpStatusCode, routes} from "@lodestar/api"; +import {SyncCommitteeService, SyncCommitteeServiceOpts} from "../../../src/services/syncCommittee.js"; import {SyncDutyAndProofs} from "../../../src/services/syncCommitteeDuties.js"; import {ValidatorStore} from "../../../src/services/validatorStore.js"; import {getApiClientStub} from "../../utils/apiStub.js"; @@ -47,87 +47,144 @@ describe("SyncCommitteeService", function () { let controller: AbortController; // To stop clock beforeEach(() => (controller = new AbortController())); - afterEach(() => controller.abort()); - - it("Should produce, sign, and publish a sync committee + contribution", async () => { - const clock = new ClockMock(); - const syncCommitteeService = new SyncCommitteeService( - config, - loggerVc, - api, - clock, - validatorStore, - emitter, - chainHeaderTracker, - null - ); - - const beaconBlockRoot = Buffer.alloc(32, 0x4d); - const syncCommitteeSignature = ssz.altair.SyncCommitteeMessage.defaultValue(); - const contribution = ssz.altair.SyncCommitteeContribution.defaultValue(); - const contributionAndProof = ssz.altair.SignedContributionAndProof.defaultValue(); - const duties: SyncDutyAndProofs[] = [ - { - duty: { - pubkey: toHexString(pubkeys[0]), - validatorIndex: 0, - subnets: [0], - }, - selectionProofs: [{selectionProof: ZERO_HASH, subcommitteeIndex: 0}], - }, - ]; - - // Return empty replies to duties service - api.beacon.getStateValidators.resolves({ - response: {data: [], executionOptimistic: false}, - ok: true, - status: HttpStatusCode.OK, - }); - api.validator.getSyncCommitteeDuties.resolves({ - response: {data: [], executionOptimistic: false}, - ok: true, - status: HttpStatusCode.OK, - }); - - // Mock duties service to return some duties directly - syncCommitteeService["dutiesService"].getDutiesAtSlot = sinon.stub().returns(duties); - - // Mock beacon's sync committee and contribution routes + afterEach(() => { + controller.abort(); + sandbox.resetHistory(); + }); - chainHeaderTracker.getCurrentChainHead.returns(beaconBlockRoot); - api.beacon.submitPoolSyncCommitteeSignatures.resolves(); - api.validator.produceSyncCommitteeContribution.resolves({ - response: {data: contribution}, - ok: true, - status: HttpStatusCode.OK, + const testContexts: [string, SyncCommitteeServiceOpts][] = [ + ["With default configuration", {}], + ["With distributed aggregation selection enabled", {distributedAggregationSelection: true}], + ]; + + for (const [title, opts] of testContexts) { + context(title, () => { + it("Should produce, sign, and publish a sync committee + contribution", async () => { + const clock = new ClockMock(); + const syncCommitteeService = new SyncCommitteeService( + config, + loggerVc, + api, + clock, + validatorStore, + emitter, + chainHeaderTracker, + null, + opts + ); + + const beaconBlockRoot = Buffer.alloc(32, 0x4d); + const syncCommitteeSignature = ssz.altair.SyncCommitteeMessage.defaultValue(); + const contribution = ssz.altair.SyncCommitteeContribution.defaultValue(); + const contributionAndProof = ssz.altair.SignedContributionAndProof.defaultValue(); + const duties: SyncDutyAndProofs[] = [ + { + duty: { + pubkey: toHexString(pubkeys[0]), + validatorIndex: 0, + subnets: [0], + }, + selectionProofs: [ + { + selectionProof: opts.distributedAggregationSelection ? null : ZERO_HASH, + partialSelectionProof: opts.distributedAggregationSelection ? ZERO_HASH : undefined, + subcommitteeIndex: 0, + }, + ], + }, + ]; + + // Return empty replies to duties service + api.beacon.getStateValidators.resolves({ + response: {data: [], executionOptimistic: false}, + ok: true, + status: HttpStatusCode.OK, + }); + api.validator.getSyncCommitteeDuties.resolves({ + response: {data: [], executionOptimistic: false}, + ok: true, + status: HttpStatusCode.OK, + }); + + // Mock duties service to return some duties directly + syncCommitteeService["dutiesService"].getDutiesAtSlot = sinon.stub().returns(duties); + + // Mock beacon's sync committee and contribution routes + + chainHeaderTracker.getCurrentChainHead.returns(beaconBlockRoot); + api.beacon.submitPoolSyncCommitteeSignatures.resolves({ + response: undefined, + ok: true, + status: HttpStatusCode.OK, + }); + api.validator.produceSyncCommitteeContribution.resolves({ + response: {data: contribution}, + ok: true, + status: HttpStatusCode.OK, + }); + api.validator.publishContributionAndProofs.resolves({ + response: undefined, + ok: true, + status: HttpStatusCode.OK, + }); + + if (opts.distributedAggregationSelection) { + // Mock distributed validator middleware client selections endpoint + // and return a selection proof that passes `is_sync_committee_aggregator` test + api.validator.submitSyncCommitteeSelections.resolves({ + response: { + data: [{validatorIndex: 0, slot: 0, subcommitteeIndex: 0, selectionProof: Buffer.alloc(1, 0x19)}], + }, + ok: true, + status: HttpStatusCode.OK, + }); + } + + // Mock signing service + validatorStore.signSyncCommitteeSignature.resolves(syncCommitteeSignature); + validatorStore.signContributionAndProof.resolves(contributionAndProof); + + // Trigger clock onSlot for slot 0 + await clock.tickSlotFns(0, controller.signal); + + if (opts.distributedAggregationSelection) { + // Must submit partial sync committee selection proof based on duty + const selection: routes.validator.SyncCommitteeSelection = { + validatorIndex: 0, + slot: 0, + subcommitteeIndex: 0, + selectionProof: ZERO_HASH, + }; + expect(api.validator.submitSyncCommitteeSelections.callCount).to.equal( + 1, + "submitSyncCommitteeSelections() must be called once" + ); + expect(api.validator.submitSyncCommitteeSelections.getCall(0).args).to.deep.equal( + [[selection]], // 1 arg, = selection[] + "wrong submitSyncCommitteeSelections() args" + ); + } + + // Must submit the signature received through signSyncCommitteeSignature() + expect(api.beacon.submitPoolSyncCommitteeSignatures.callCount).to.equal( + 1, + "submitPoolSyncCommitteeSignatures() must be called once" + ); + expect(api.beacon.submitPoolSyncCommitteeSignatures.getCall(0).args).to.deep.equal( + [[syncCommitteeSignature]], // 1 arg, = syncCommitteeSignature[] + "wrong submitPoolSyncCommitteeSignatures() args" + ); + + // Must submit the aggregate received through produceSyncCommitteeContribution() then signContributionAndProof() + expect(api.validator.publishContributionAndProofs.callCount).to.equal( + 1, + "publishContributionAndProofs() must be called once" + ); + expect(api.validator.publishContributionAndProofs.getCall(0).args).to.deep.equal( + [[contributionAndProof]], // 1 arg, = contributionAndProof[] + "wrong publishContributionAndProofs() args" + ); + }); }); - api.validator.publishContributionAndProofs.resolves(); - - // Mock signing service - validatorStore.signSyncCommitteeSignature.resolves(syncCommitteeSignature); - validatorStore.signContributionAndProof.resolves(contributionAndProof); - - // Trigger clock onSlot for slot 0 - await clock.tickSlotFns(0, controller.signal); - - // Must submit the signature received through signSyncCommitteeSignature() - expect(api.beacon.submitPoolSyncCommitteeSignatures.callCount).to.equal( - 1, - "submitPoolSyncCommitteeSignatures() must be called once" - ); - expect(api.beacon.submitPoolSyncCommitteeSignatures.getCall(0).args).to.deep.equal( - [[syncCommitteeSignature]], // 1 arg, = syncCommitteeSignature[] - "wrong submitPoolSyncCommitteeSignatures() args" - ); - - // Must submit the aggregate received through produceSyncCommitteeContribution() then signContributionAndProof() - expect(api.validator.publishContributionAndProofs.callCount).to.equal( - 1, - "publishContributionAndProofs() must be called once" - ); - expect(api.validator.publishContributionAndProofs.getCall(0).args).to.deep.equal( - [[contributionAndProof]], // 1 arg, = contributionAndProof[] - "wrong publishContributionAndProofs() args" - ); - }); + } });