ChainSafe · wemeetagain · Apr 17, 2023 · Apr 3, 2023 · Apr 13, 2023 · Apr 13, 2023
@@ -53,7 +53,7 @@ export function getBeaconPoolApi({
         attestations.map(async (attestation, i) => {
           try {
             // eslint-disable-next-line @typescript-eslint/explicit-function-return-type
-            const validateFn = () => validateGossipAttestation(chain, attestation, null, null);
+            const validateFn = () => validateGossipAttestation(chain, {attestation, serializedData: null}, null);
             const {slot, beaconBlockRoot} = attestation.data;
             // when a validator is configured with multiple beacon node urls, this attestation data may come from another beacon node
             // and the block hasn't been in our forkchoice since we haven't seen / processing that block

@@ -493,6 +493,12 @@ export class BeaconChain implements IBeaconChain {
     }
   }
 
+  persistInvalidSszBytes(typeName: string, sszBytes: Uint8Array, suffix?: string): void {
+    if (this.opts.persistInvalidSszObjects) {
+      void this.persistInvalidSszObject(typeName, sszBytes, sszBytes, suffix);
+    }
+  }
+
   persistInvalidSszView(view: TreeView<CompositeTypeAny>, suffix?: string): void {
     if (this.opts.persistInvalidSszObjects) {
       void this.persistInvalidSszObject(view.type.typeName, view.serialize(), view.hashTreeRoot(), suffix);

@@ -126,6 +126,10 @@ export enum AttestationErrorCode {
    * Invalid attestation indexes: not sorted or unique
    */
   INVALID_INDEXED_ATTESTATION = "ATTESTATION_ERROR_INVALID_INDEXED_ATTESTATION",
+  /**
+   * Invalid ssz bytes.
+   */
+  INVALID_SERIALIZED_BYTES = "ATTESTATION_ERROR_INVALID_SERIALIZED_BYTES",
 }
 
 export type AttestationErrorType =
@@ -158,7 +162,8 @@ export type AttestationErrorType =
   | {code: AttestationErrorCode.COMMITTEE_INDEX_OUT_OF_RANGE; index: number}
   | {code: AttestationErrorCode.MISSING_ATTESTATION_HEAD_STATE; error: Error}
   | {code: AttestationErrorCode.INVALID_AGGREGATOR}
-  | {code: AttestationErrorCode.INVALID_INDEXED_ATTESTATION};
+  | {code: AttestationErrorCode.INVALID_INDEXED_ATTESTATION}
+  | {code: AttestationErrorCode.INVALID_SERIALIZED_BYTES};
 
 export class AttestationError extends GossipActionError<AttestationErrorType> {
   getMetadata(): Record<string, string | number | null> {

@@ -5,6 +5,8 @@ export enum GossipAction {
   REJECT = "REJECT",
 }
 
+export const INVALID_SERIALIZED_BYTES_ERROR_CODE = "GOSSIP_ERROR_INVALID_SERIALIZED_BYTES";
+
 export class GossipActionError<T extends {code: string}> extends LodestarError<T> {
   action: GossipAction;
 

@@ -134,6 +134,7 @@ export interface IBeaconChain {
   updateBeaconProposerData(epoch: Epoch, proposers: ProposerPreparationData[]): Promise<void>;
 
   persistInvalidSszValue<T>(type: Type<T>, sszObject: T | Uint8Array, suffix?: string): void;
+  persistInvalidSszBytes(type: string, sszBytes: Uint8Array, suffix?: string): void;
   /** Persist bad items to persistInvalidSszObjectsDir dir, for example invalid state, attestations etc. */
   persistInvalidSszView(view: TreeView<CompositeTypeAny>, suffix?: string): void;
   updateBuilderStatus(clockSlot: Slot): void;

@@ -1,4 +1,4 @@
-import {RootHex, Slot} from "@lodestar/types";
+import {phase0, RootHex, Slot} from "@lodestar/types";
 import {MapDef} from "@lodestar/utils";
 import {Metrics} from "../../metrics/metrics.js";
 import {AttDataBase64} from "../../util/sszBytes.js";
@@ -9,8 +9,11 @@ export type AttestationDataCacheEntry = {
   committeeIndices: number[];
   // IndexedAttestationData signing root, 32 bytes
   signingRoot: Uint8Array;
-  // to be consumed by forkchoice
+  // to be consumed by forkchoice and oppool
   attDataRootHex: RootHex;
+  // caching this for 3 slots take 600 instances max, this is nothing compared to attestations processed per slot
+  // for example in a mainnet node subscribing to all subnets, attestations are processed up to 20k per slot
+  attestationData: phase0.AttestationData;
   subnet: number;
 };
 

@@ -13,21 +13,42 @@ import {IBeaconChain} from "..";
 import {AttestationError, AttestationErrorCode, GossipAction} from "../errors/index.js";
 import {MAXIMUM_GOSSIP_CLOCK_DISPARITY_SEC} from "../../constants/index.js";
 import {RegenCaller} from "../regen/index.js";
-import {getAttDataBase64FromAttestationSerialized} from "../../util/sszBytes.js";
+import {
+  AttDataBase64,
+  getAggregationBitsFromAttestationSerialized,
+  getAttDataBase64FromAttestationSerialized,
+  getSignatureFromAttestationSerialized,
+} from "../../util/sszBytes.js";
+import {AttestationDataCacheEntry} from "../seenCache/seenAttestationData.js";
+import {sszDeserializeAttestation} from "../../network/gossip/topic.js";
 
 export type AttestationValidationResult = {
+  attestation: phase0.Attestation;
   indexedAttestation: phase0.IndexedAttestation;
   subnet: number;
   attDataRootHex: RootHex;
 };
 
+export type AttestationOrBytes =
+  // for api
+  | {attestation: phase0.Attestation; serializedData: null}
+  // for gossip
+  | {
+      attestation: null;
+      serializedData: Uint8Array;
+      // available in NetworkProcessor since we check for unknown block root attestations
+      attSlot: Slot;
+    };
+
+/**
+ * Only deserialize the attestation if needed, use the cached AttestationData instead
+ * This is to avoid deserializing similar attestation multiple times which could help the gc
+ */
 export async function validateGossipAttestation(
   chain: IBeaconChain,
-  attestation: phase0.Attestation,
+  attestationOrBytes: AttestationOrBytes,
   /** Optional, to allow verifying attestations through API with unknown subnet */
-  subnet: number | null,
-  // available for gossip attestations, null for api attestations
-  serializedData: Uint8Array | null = null
+  subnet: number | null
 ): Promise<AttestationValidationResult> {
   // Do checks in this order:
   // - do early checks (w/o indexed attestation)
@@ -38,17 +59,39 @@ export async function validateGossipAttestation(
 
   // verify_early_checks
   // Run the checks that happen before an indexed attestation is constructed.
-  const attData = attestation.data;
+
+  let attestationOrCache:
+    | {attestation: phase0.Attestation; cache: null}
+    | {attestation: null; cache: AttestationDataCacheEntry; serializedData: Uint8Array};
+  let attDataBase64: AttDataBase64 | null;
+  if (attestationOrBytes.serializedData) {
+    // gossip
+    attDataBase64 = getAttDataBase64FromAttestationSerialized(attestationOrBytes.serializedData);
+    const attSlot = attestationOrBytes.attSlot;
+    const cachedAttData = attDataBase64 !== null ? chain.seenAttestationDatas.get(attSlot, attDataBase64) : null;
+    if (cachedAttData === null) {
+      const attestation = sszDeserializeAttestation(attestationOrBytes.serializedData);
+      // only deserialize on the first AttestationData that's not cached
+      attestationOrCache = {attestation, cache: null};
+    } else {
+      attestationOrCache = {attestation: null, cache: cachedAttData, serializedData: attestationOrBytes.serializedData};
+    }
+  } else {
+    // api
+    attDataBase64 = null;
+    attestationOrCache = {attestation: attestationOrBytes.attestation, cache: null};
+  }
+
+  const attData: phase0.AttestationData = attestationOrCache.attestation
+    ? attestationOrCache.attestation.data
+    : attestationOrCache.cache.attestationData;
   const attSlot = attData.slot;
   const attIndex = attData.index;
   const attEpoch = computeEpochAtSlot(attSlot);
   const attTarget = attData.target;
   const targetEpoch = attTarget.epoch;
 
-  const attDataBase64 = serializedData ? getAttDataBase64FromAttestationSerialized(serializedData) : null;
-  const cachedAttData = attDataBase64 ? chain.seenAttestationDatas.get(attSlot, attDataBase64) : null;
-
-  if (!cachedAttData) {
+  if (!attestationOrCache.cache) {
     // [REJECT] The attestation's epoch matches its target -- i.e. attestation.data.target.epoch == compute_epoch_at_slot(attestation.data.slot)
     if (targetEpoch !== attEpoch) {
       throw new AttestationError(GossipAction.REJECT, {
@@ -59,13 +102,21 @@ export async function validateGossipAttestation(
     // [IGNORE] attestation.data.slot is within the last ATTESTATION_PROPAGATION_SLOT_RANGE slots (within a MAXIMUM_GOSSIP_CLOCK_DISPARITY allowance)
     //  -- i.e. attestation.data.slot + ATTESTATION_PROPAGATION_SLOT_RANGE >= current_slot >= attestation.data.slot
     // (a client MAY queue future attestations for processing at the appropriate slot).
-    verifyPropagationSlotRange(chain, attSlot);
+    verifyPropagationSlotRange(chain, attestationOrCache.attestation.data.slot);
   }
 
   // [REJECT] The attestation is unaggregated -- that is, it has exactly one participating validator
   // (len([bit for bit in attestation.aggregation_bits if bit]) == 1, i.e. exactly 1 bit is set).
   // > TODO: Do this check **before** getting the target state but don't recompute zipIndexes
-  const aggregationBits = attestation.aggregationBits;
+  const aggregationBits = attestationOrCache.attestation
+    ? attestationOrCache.attestation.aggregationBits
+    : getAggregationBitsFromAttestationSerialized(attestationOrCache.serializedData);
+  if (aggregationBits === null) {
+    throw new AttestationError(GossipAction.REJECT, {
+      code: AttestationErrorCode.INVALID_SERIALIZED_BYTES,
+    });
+  }
+
   const bitIndex = aggregationBits.getSingleTrueBit();
   if (bitIndex === null) {
     throw new AttestationError(GossipAction.REJECT, {
@@ -76,10 +127,11 @@ export async function validateGossipAttestation(
   let committeeIndices: number[];
   let getSigningRoot: () => Uint8Array;
   let expectedSubnet: number;
-  if (cachedAttData) {
-    committeeIndices = cachedAttData.committeeIndices;
-    getSigningRoot = () => cachedAttData.signingRoot;
-    expectedSubnet = cachedAttData.subnet;
+  if (attestationOrCache.cache) {
+    committeeIndices = attestationOrCache.cache.committeeIndices;
+    const signingRoot = attestationOrCache.cache.signingRoot;
+    getSigningRoot = () => signingRoot;
+    expectedSubnet = attestationOrCache.cache.subnet;
   } else {
     // Attestations must be for a known block. If the block is unknown, we simply drop the
     // attestation and do not delay consideration for later.
@@ -88,7 +140,12 @@ export async function validateGossipAttestation(
 
     // [IGNORE] The block being voted for (attestation.data.beacon_block_root) has been seen (via both gossip
     // and non-gossip sources) (a client MAY queue attestations for processing once block is retrieved).
-    const attHeadBlock = verifyHeadBlockAndTargetRoot(chain, attData.beaconBlockRoot, attTarget.root, attEpoch);
+    const attHeadBlock = verifyHeadBlockAndTargetRoot(
+      chain,
+      attestationOrCache.attestation.data.beaconBlockRoot,
+      attestationOrCache.attestation.data.target.root,
+      attEpoch
+    );
 
     // [REJECT] The block being voted for (attestation.data.beacon_block_root) passes validation.
     // > Altready check in `verifyHeadBlockAndTargetRoot()`
@@ -162,19 +219,28 @@ export async function validateGossipAttestation(
   const attestingIndices = [validatorIndex];
   let signatureSet: ISignatureSet;
   let attDataRootHex: RootHex;
-  if (cachedAttData) {
+  const signature = attestationOrCache.attestation
+    ? attestationOrCache.attestation.signature
+    : getSignatureFromAttestationSerialized(attestationOrCache.serializedData);
+  if (signature === null) {
+    throw new AttestationError(GossipAction.REJECT, {
+      code: AttestationErrorCode.INVALID_SERIALIZED_BYTES,
+    });
+  }
+
+  if (attestationOrCache.cache) {
     // there could be up to 6% of cpu time to compute signing root if we don't clone the signature set
     signatureSet = createAggregateSignatureSetFromComponents(
       attestingIndices.map((i) => chain.index2pubkey[i]),
-      cachedAttData.signingRoot,
-      attestation.signature
+      attestationOrCache.cache.signingRoot,
+      signature
     );
-    attDataRootHex = cachedAttData.attDataRootHex;
+    attDataRootHex = attestationOrCache.cache.attDataRootHex;
   } else {
     signatureSet = createAggregateSignatureSetFromComponents(
       attestingIndices.map((i) => chain.index2pubkey[i]),
       getSigningRoot(),
-      attestation.signature
+      signature
     );
 
     // add cached attestation data before verifying signature
@@ -187,6 +253,7 @@ export async function validateGossipAttestation(
         // precompute this to be used in forkchoice
         // root of AttestationData was already cached during getIndexedAttestationSignatureSet
         attDataRootHex,
+        attestationData: attData,
       });
     }
   }
@@ -213,9 +280,17 @@ export async function validateGossipAttestation(
   const indexedAttestation: phase0.IndexedAttestation = {
     attestingIndices,
     data: attData,
-    signature: attestation.signature,
+    signature,
   };
-  return {indexedAttestation, subnet: expectedSubnet, attDataRootHex};
+
+  const attestation: phase0.Attestation = attestationOrCache.attestation
+    ? attestationOrCache.attestation
+    : {
+        aggregationBits,
+        data: attData,
+        signature,
+      };
+  return {attestation, indexedAttestation, subnet: expectedSubnet, attDataRootHex};
 }
 
 /**

@@ -4,7 +4,7 @@ import {Message, TopicValidatorResult} from "@libp2p/interface-pubsub";
 import StrictEventEmitter from "strict-event-emitter-types";
 import {PeerIdStr} from "@chainsafe/libp2p-gossipsub/types";
 import {ForkName} from "@lodestar/params";
-import {allForks, altair, capella, deneb, phase0} from "@lodestar/types";
+import {allForks, altair, capella, deneb, phase0, Slot} from "@lodestar/types";
 import {BeaconConfig} from "@lodestar/config";
 import {Logger} from "@lodestar/utils";
 import {IBeaconChain} from "../../chain/index.js";
@@ -65,6 +65,10 @@ export type GossipTopicMap = {
  */
 export type GossipTopic = GossipTopicMap[keyof GossipTopicMap];
 
+export type SSZTypeOfGossipTopic<T extends GossipTopic> = T extends {type: infer K extends GossipType}
+  ? GossipTypeMap[K]
+  : never;
+
 export type GossipTypeMap = {
   [GossipType.beacon_block]: allForks.SignedBeaconBlock;
   [GossipType.beacon_block_and_blobs_sidecar]: deneb.SignedBeaconBlockAndBlobsSidecar;
@@ -152,7 +156,8 @@ export type GossipValidatorFn = (
   topic: GossipTopic,
   msg: Message,
   propagationSource: PeerIdStr,
-  seenTimestampSec: number
+  seenTimestampSec: number,
+  msgSlot?: Slot
 ) => Promise<TopicValidatorResult>;
 
 export type ValidatorFnsByType = {[K in GossipType]: GossipValidatorFn};
@@ -161,20 +166,24 @@ export type GossipJobQueues = {
   [K in GossipType]: JobItemQueue<Parameters<GossipValidatorFn>, ResolvedType<GossipValidatorFn>>;
 };
 
+export type GossipData = {
+  serializedData: Uint8Array;
+  msgSlot?: Slot;
+};
+
 export type GossipHandlerFn = (
-  object: GossipTypeMap[GossipType],
+  gossipData: GossipData,
   topic: GossipTopicMap[GossipType],
   peerIdStr: string,
-  seenTimestampSec: number,
-  gossipSerializedData: Uint8Array
+  seenTimestampSec: number
 ) => Promise<void>;
+
 export type GossipHandlers = {
   [K in GossipType]: (
-    object: GossipTypeMap[K],
+    gossipData: GossipData,
     topic: GossipTopicMap[K],
     peerIdStr: string,
-    seenTimestampSec: number,
-    gossipSerializedData: Uint8Array
+    seenTimestampSec: number
   ) => Promise<void>;
 };
 

@@ -1,4 +1,4 @@
-import {ssz} from "@lodestar/types";
+import {phase0, ssz} from "@lodestar/types";
 import {ForkDigestContext} from "@lodestar/config";
 import {
   ATTESTATION_SUBNET_COUNT,
@@ -8,7 +8,12 @@ import {
   isForkLightClient,
 } from "@lodestar/params";
 
-import {GossipEncoding, GossipTopic, GossipType, GossipTopicTypeMap} from "./interface.js";
+import {
+  GossipAction,
+  GossipActionError,
+  INVALID_SERIALIZED_BYTES_ERROR_CODE,
+} from "../../chain/errors/gossipValidation.js";
+import {GossipEncoding, GossipTopic, GossipType, GossipTopicTypeMap, SSZTypeOfGossipTopic} from "./interface.js";
 import {DEFAULT_ENCODING} from "./constants.js";
 
 export interface IGossipTopicCache {
@@ -110,6 +115,29 @@ export function getGossipSSZType(topic: GossipTopic) {
   }
 }
 
+/**
+ * Deserialize a gossip serialized data into an ssz object.
+ */
+export function sszDeserialize<T extends GossipTopic>(topic: T, serializedData: Uint8Array): SSZTypeOfGossipTopic<T> {
+  const sszType = getGossipSSZType(topic);
+  try {
+    return sszType.deserialize(serializedData) as SSZTypeOfGossipTopic<T>;
+  } catch (e) {
+    throw new GossipActionError(GossipAction.REJECT, {code: INVALID_SERIALIZED_BYTES_ERROR_CODE});
+  }
+}
+
+/**
+ * Deserialize a gossip serialized data into an Attestation object.
+ */
+export function sszDeserializeAttestation(serializedData: Uint8Array): phase0.Attestation {
+  try {
+    return ssz.phase0.Attestation.deserialize(serializedData);
+  } catch (e) {
+    throw new GossipActionError(GossipAction.REJECT, {code: INVALID_SERIALIZED_BYTES_ERROR_CODE});
+  }
+}
+
 // Parsing
 
 const gossipTopicRegex = new RegExp("^/eth2/(\\w+)/(\\w+)/(\\w+)");