Skip to content

feat: only subscribe to sampled subnets #7860

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
twoeths wants to merge 1 commit into from
Closed

feat: only subscribe to sampled subnets #7860

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions packages/beacon-node/src/network/core/networkCore.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,7 +178,7 @@ export class NetworkCore implements INetworkCore {
);

const gossip = new Eth2Gossipsub(opts, {
config,
networkConfig,
libp2p,
logger,
metricsRegister: metricsRegistry,
Expand Down Expand Up @@ -331,7 +331,7 @@ export class NetworkCore implements INetworkCore {
}

for (const fork of getActiveForks(this.config, this.clock.currentEpoch)) {
this.subscribeCoreTopicsAtFork(this.config, fork);
this.subscribeCoreTopicsAtFork(this.networkConfig, fork);
}
}

Expand All @@ -340,7 +340,7 @@ export class NetworkCore implements INetworkCore {
*/
async unsubscribeGossipCoreTopics(): Promise<void> {
for (const fork of this.subscribedForks.values()) {
this.unsubscribeCoreTopicsAtFork(this.config, fork);
this.unsubscribeCoreTopicsAtFork(this.networkConfig, fork);
}
}

Expand Down Expand Up @@ -496,7 +496,7 @@ export class NetworkCore implements INetworkCore {
if (epoch === forkEpoch - FORK_EPOCH_LOOKAHEAD) {
// Don't subscribe to new fork if the node is not subscribed to any topic
if (await this.isSubscribedToGossipCoreTopics()) {
this.subscribeCoreTopicsAtFork(this.config, nextFork);
this.subscribeCoreTopicsAtFork(this.networkConfig, nextFork);
this.logger.info("Subscribing gossip topics before fork", {nextFork});
} else {
this.logger.info("Skipping subscribing gossip topics before fork", {nextFork});
Expand All @@ -515,7 +515,7 @@ export class NetworkCore implements INetworkCore {
// After fork transition
if (epoch === forkEpoch + FORK_EPOCH_LOOKAHEAD) {
this.logger.info("Unsubscribing gossip topics from prev fork", {prevFork});
this.unsubscribeCoreTopicsAtFork(this.config, prevFork);
this.unsubscribeCoreTopicsAtFork(this.networkConfig, prevFork);
this.attnetsService.unsubscribeSubnetsFromPrevFork(prevFork);
this.syncnetsService.unsubscribeSubnetsFromPrevFork(prevFork);
}
Expand All @@ -526,25 +526,25 @@ export class NetworkCore implements INetworkCore {
}
};

private subscribeCoreTopicsAtFork(config: BeaconConfig, fork: ForkName): void {
private subscribeCoreTopicsAtFork(networkConfig: NetworkConfig, fork: ForkName): void {
if (this.subscribedForks.has(fork)) return;
this.subscribedForks.add(fork);
const {subscribeAllSubnets, disableLightClientServer} = this.opts;

for (const topic of getCoreTopicsAtFork(config, fork, {
for (const topic of getCoreTopicsAtFork(networkConfig, fork, {
subscribeAllSubnets,
disableLightClientServer,
})) {
this.gossip.subscribeTopic({...topic, fork});
}
}

private unsubscribeCoreTopicsAtFork(config: BeaconConfig, fork: ForkName): void {
private unsubscribeCoreTopicsAtFork(networkConfig: NetworkConfig, fork: ForkName): void {
if (!this.subscribedForks.has(fork)) return;
this.subscribedForks.delete(fork);
const {subscribeAllSubnets, disableLightClientServer} = this.opts;

for (const topic of getCoreTopicsAtFork(config, fork, {
for (const topic of getCoreTopicsAtFork(networkConfig, fork, {
subscribeAllSubnets,
disableLightClientServer,
})) {
Expand Down
20 changes: 13 additions & 7 deletions packages/beacon-node/src/network/gossip/gossipsub.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import {GossipTopic, GossipType} from "./interface.js";
import {Eth2GossipsubMetrics, createEth2GossipsubMetrics} from "./metrics.js";
import {GossipTopicCache, getCoreTopicsAtFork, stringifyGossipTopic} from "./topic.js";

import {NetworkConfig} from "../networkConfig.js";
import {
GOSSIP_D,
GOSSIP_D_HIGH,
Expand All @@ -39,7 +40,7 @@ export type Eth2Context = {
};

export type Eth2GossipsubModules = {
config: BeaconConfig;
networkConfig: NetworkConfig;
libp2p: Libp2p;
logger: Logger;
metricsRegister: RegistryMetricCreator | null;
Expand Down Expand Up @@ -84,10 +85,11 @@ export class Eth2Gossipsub extends GossipSub {

constructor(opts: Eth2GossipsubOpts, modules: Eth2GossipsubModules) {
const {allowPublishToZeroPeers, gossipsubD, gossipsubDLow, gossipsubDHigh} = opts;
const gossipTopicCache = new GossipTopicCache(modules.config);
const {networkConfig, logger, metricsRegister, peersData, events} = modules;
const config = networkConfig.getConfig();
const gossipTopicCache = new GossipTopicCache(config);

const scoreParams = computeGossipPeerScoreParams(modules);
const {config, logger, metricsRegister, peersData, events} = modules;
const scoreParams = computeGossipPeerScoreParams({config, eth2Context: modules.eth2Context});

// Gossipsub parameters defined here:
// https://github.com/ethereum/consensus-specs/blob/v1.1.10/specs/phase0/p2p-interface.md#the-gossip-domain-gossipsub
Expand Down Expand Up @@ -126,7 +128,7 @@ export class Eth2Gossipsub extends GossipSub {
),
metricsRegister: metricsRegister as MetricsRegister | null,
metricsTopicStrToLabel: metricsRegister
? getMetricsTopicStrToLabel(config, {disableLightClientServer: opts.disableLightClientServer ?? false})
? getMetricsTopicStrToLabel(networkConfig, {disableLightClientServer: opts.disableLightClientServer ?? false})
: undefined,
asyncValidation: true,

Expand Down Expand Up @@ -330,11 +332,15 @@ function attSubnetLabel(subnet: SubnetID): string {
return `0${subnet}`;
}

function getMetricsTopicStrToLabel(config: BeaconConfig, opts: {disableLightClientServer: boolean}): TopicStrToLabel {
function getMetricsTopicStrToLabel(
networkConfig: NetworkConfig,
opts: {disableLightClientServer: boolean}
): TopicStrToLabel {
const metricsTopicStrToLabel = new Map<TopicStr, TopicLabel>();
const config = networkConfig.getConfig();

for (const {name: fork} of config.forksAscendingEpochOrder) {
const topics = getCoreTopicsAtFork(config, fork, {
const topics = getCoreTopicsAtFork(networkConfig, fork, {
subscribeAllSubnets: true,
disableLightClientServer: opts.disableLightClientServer,
});
Expand Down
2 changes: 1 addition & 1 deletion packages/beacon-node/src/network/gossip/scoringParameters.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ type TopicScoreInput = {
export function computeGossipPeerScoreParams({
config,
eth2Context,
}: Pick<Eth2GossipsubModules, "config" | "eth2Context">): Partial<PeerScoreParams> {
}: {config: BeaconConfig; eth2Context: Eth2Context}): Partial<PeerScoreParams> {
const decayIntervalMs = config.SECONDS_PER_SLOT * 1000;
const decayToZero = 0.01;
const epochDurationMs = config.SECONDS_PER_SLOT * SLOTS_PER_EPOCH * 1000;
Expand Down
7 changes: 5 additions & 2 deletions packages/beacon-node/src/network/gossip/topic.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import {
import {Attestation, SingleAttestation, ssz, sszTypesFor} from "@lodestar/types";

import {GossipAction, GossipActionError, GossipErrorCode} from "../../chain/errors/gossipValidation.js";
import {NetworkConfig} from "../networkConfig.js";
import {DEFAULT_ENCODING} from "./constants.js";
import {GossipEncoding, GossipTopic, GossipTopicTypeMap, GossipType, SSZTypeOfGossipTopic} from "./interface.js";

Expand Down Expand Up @@ -227,7 +228,7 @@ export function parseGossipTopic(forkDigestContext: ForkDigestContext, topicStr:
* De-duplicate logic to pick fork topics between subscribeCoreTopicsAtFork and unsubscribeCoreTopicsAtFork
*/
export function getCoreTopicsAtFork(
config: ChainConfig,
networkConfig: NetworkConfig,
fork: ForkName,
opts: {subscribeAllSubnets?: boolean; disableLightClientServer?: boolean}
): GossipTopicTypeMap[keyof GossipTopicTypeMap][] {
Expand All @@ -243,13 +244,15 @@ export function getCoreTopicsAtFork(
// After fulu also track data_column_sidecar_{index}
if (ForkSeq[fork] >= ForkSeq.fulu) {
// TODO: @matthewkeil check if this needs to be updated for custody groups
for (let subnet = 0; subnet < DATA_COLUMN_SIDECAR_SUBNET_COUNT; subnet++) {
const subnets = networkConfig.getCustodyConfig().sampledSubnets;
for (const subnet of subnets) {
topics.push({type: GossipType.data_column_sidecar, subnet});
}
}

// After Deneb also track blob_sidecar_{subnet_id}
if (ForkSeq[fork] >= ForkSeq.deneb) {
const config = networkConfig.getConfig();
const subnetCount = isForkPostElectra(fork)
? config.BLOB_SIDECAR_SUBNET_COUNT_ELECTRA
: config.BLOB_SIDECAR_SUBNET_COUNT;
Expand Down
Loading