Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 1 addition & 0 deletions packages/lodestar/src/db/api/beacon/repositories/index.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
export * from "./attestations";
export * from "./attesterSlashing";
export * from "./block";
export * from "./blockArchive";
export * from "./chain";
export * from "./deposit";
export * from "./merkleTree";
Expand Down
37 changes: 30 additions & 7 deletions packages/lodestar/src/sync/reqResp/reqResp.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@ import {INetwork} from "../../network";
import {ILogger} from "../../logger";
import {ISyncOptions, ISyncReqResp} from "./interface";
import {ReputationStore} from "../IReputation";
import {computeStartSlotOfEpoch} from "@chainsafe/eth2.0-state-transition";
import {signingRoot} from "@chainsafe/ssz";

export interface ISyncReqRespModules {
config: IBeaconConfig;
Expand All @@ -28,6 +30,12 @@ export interface ISyncReqRespModules {
logger: ILogger;
}

enum GoodByeReasonCode {
CLIENT_SHUTDOWN = 1,
IRRELEVANT_NETWORK = 2,
ERROR = 3,
}

/**
* The SyncReqResp module handles app-level requests / responses from other peers,
* fetching state from the chain and database as needed.
Expand Down Expand Up @@ -64,7 +72,7 @@ export class SyncReqResp implements ISyncReqResp {
this.network.reqResp.removeListener("request", this.onRequest);
await Promise.all(
this.network.getPeers().map((peerInfo) =>
this.network.reqResp.goodbye(peerInfo, 0n)));

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would extract reason codes to enum

this.network.reqResp.goodbye(peerInfo, BigInt(GoodByeReasonCode.CLIENT_SHUTDOWN))));
}

public onRequest = async (
Expand Down Expand Up @@ -97,7 +105,24 @@ export class SyncReqResp implements ISyncReqResp {
} catch (e) {
this.network.reqResp.sendResponse(id, e, null);
}
// TODO handle incorrect forkVersion or disjoint finalizedCheckpoint
if (await this.shouldDisconnectOnHello(request)) {
this.network.reqResp.goodbye(peerInfo, BigInt(GoodByeReasonCode.IRRELEVANT_NETWORK));
}
}

public async shouldDisconnectOnHello(request: Hello): Promise<boolean> {
const headBlock = await this.db.block.getChainHead();
const state = await this.db.state.get(headBlock.stateRoot);
if (!state.fork.currentVersion.equals(request.headForkVersion)) {
return true;
}
Comment thread
mpetrunic marked this conversation as resolved.
Resolved
const startSlot = computeStartSlotOfEpoch(this.config, request.finalizedEpoch);
const startBlock = await this.db.blockArchive.get(startSlot);
if (state.finalizedCheckpoint.epoch >= request.finalizedEpoch &&
!request.finalizedRoot.equals(signingRoot(this.config.types.BeaconBlock, startBlock))) {
return true;
}
return false;
}

// eslint-disable-next-line @typescript-eslint/no-unused-vars
Expand Down Expand Up @@ -153,11 +178,9 @@ export class SyncReqResp implements ISyncReqResp {
finalizedRoot = ZERO_HASH;
} else {
headSlot = await this.db.chain.getChainHeadSlot();
const [bRoot, state] = await Promise.all([
this.db.chain.getBlockRoot(headSlot),
this.db.state.getLatest(),
]);
headRoot = bRoot;
const headBlock = await this.db.block.getChainHead();
const state = await this.db.state.get(headBlock.stateRoot);
headRoot = signingRoot(this.config.types.BeaconBlock, headBlock);
finalizedEpoch = state.finalizedCheckpoint.epoch;
finalizedRoot = state.finalizedCheckpoint.root;
}
Expand Down
20 changes: 16 additions & 4 deletions packages/lodestar/test/e2e/sync/reqResp.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,9 @@ import {WinstonLogger} from "../../../src/logger";
import {INetworkOptions} from "../../../src/network/options";
import {BeaconMetrics} from "../../../src/metrics";
import {generateState} from "../../utils/state";
import {BlockRepository, ChainRepository, StateRepository} from "../../../src/db/api/beacon/repositories";
import {BlockRepository, ChainRepository, StateRepository, BlockArchiveRepository} from "../../../src/db/api/beacon/repositories";
import {generateEmptyBlock} from "../../utils/block";
import {signingRoot} from "@chainsafe/ssz";

const multiaddr = "/ip4/127.0.0.1/tcp/0";
const opts: INetworkOptions = {
Expand Down Expand Up @@ -49,18 +51,28 @@ describe("[sync] rpc", function () {
networkId: 0n,
});
const state = generateState();
const block = generateEmptyBlock();
state.finalizedCheckpoint = {
epoch: 0,
root: signingRoot(config.types.BeaconBlock, block),
};
// @ts-ignore
const db = {
state: sandbox.createStubInstance(StateRepository),
chain: sandbox.createStubInstance(ChainRepository),
block: sandbox.createStubInstance(BlockRepository),
blockArchive: sandbox.createStubInstance(BlockArchiveRepository),
} as BeaconDb;
// @ts-ignore
db.state.getLatest.resolves(state);
// @ts-ignore
db.chain.getBlockRoot.resolves(Buffer.alloc(32));
db.state.get.resolves(state);
// @ts-ignore
db.chain.getChainHeadSlot.resolves(0);
// @ts-ignore
db.block.getChainHead.resolves(block);
// @ts-ignore
db.block.get.resolves(block);
// @ts-ignore
db.blockArchive.get.resolves(block);
chain.latestState = state;
rpcA = new SyncReqResp({}, {
config,
Expand Down
74 changes: 66 additions & 8 deletions packages/lodestar/test/unit/sync/reqRes.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,21 +4,22 @@ import PeerInfo from "peer-info";
import PeerId from "peer-id";
import {Goodbye, Hello,} from "@chainsafe/eth2.0-types";
import {config} from "@chainsafe/eth2.0-config/lib/presets/mainnet";
import * as ssz from "@chainsafe/ssz/lib/core/signingRoot";

import {Method, ZERO_HASH} from "../../../src/constants";
import {BeaconChain} from "../../../src/chain";
import {Libp2pNetwork} from "../../../src/network";
import {WinstonLogger} from "../../../src/logger";
import {generateState} from "../../utils/state";
import {SyncReqResp} from "../../../src/sync/reqResp";
import {BlockRepository, ChainRepository, StateRepository} from "../../../src/db/api/beacon/repositories";
import {BlockRepository, ChainRepository, StateRepository, BlockArchiveRepository} from "../../../src/db/api/beacon/repositories";
import {ReqResp} from "../../../src/network/reqResp";
import {ReputationStore} from "../../../src/sync/IReputation";

describe("syncing", function () {
let sandbox = sinon.createSandbox();
let syncRpc: SyncReqResp;
let chainStub, networkStub, dbStub, repsStub, logger, reqRespStub;
let chainStub, networkStub, dbStub, repsStub, logger, reqRespStub, signingRootStub;

beforeEach(() => {
chainStub = sandbox.createStubInstance(BeaconChain);
Expand All @@ -31,8 +32,10 @@ describe("syncing", function () {
chain: sandbox.createStubInstance(ChainRepository),
state: sandbox.createStubInstance(StateRepository),
block: sandbox.createStubInstance(BlockRepository),
blockArchive: sandbox.createStubInstance(BlockArchiveRepository),
};
repsStub = sandbox.createStubInstance(ReputationStore);
signingRootStub = sandbox.stub(ssz, "signingRoot");
logger = new WinstonLogger();
logger.silent = true;

Expand All @@ -52,7 +55,7 @@ describe("syncing", function () {
});


it('should able to create Hello - genesis time', async function () {
it("should able to create Hello - genesis time", async function () {
chainStub.genesisTime = 0;
chainStub.networkId = 1n;
chainStub.chainId = 1;
Expand All @@ -73,7 +76,7 @@ describe("syncing", function () {
expect.fail(e.stack);
}
});
it('should start and stop sync rpc', async function () {
it("should start and stop sync rpc", async function () {
const peerInfo: PeerInfo = new PeerInfo(new PeerId(Buffer.from("lodestar")));
networkStub.hasPeer.returns(true);
networkStub.getPeers.returns([peerInfo, peerInfo]);
Expand All @@ -91,7 +94,7 @@ describe("syncing", function () {
}
});

it('should handle request - onHello(success)', async function () {
it("should handle request - onHello(success)", async function () {
const peerInfo: PeerInfo = new PeerInfo(new PeerId(Buffer.from("lodestar")));
const body: Hello = {
headForkVersion: Buffer.alloc(4),
Expand All @@ -104,15 +107,18 @@ describe("syncing", function () {
latestHello: null,
});
reqRespStub.sendResponse.resolves(0);
dbStub.block.getChainHead.resolves(Buffer.alloc(0));
dbStub.state.get.resolves(generateState());
try {
await syncRpc.onRequest(peerInfo, Method.Hello, "hello", body);
expect(reqRespStub.sendResponse.calledOnce).to.be.true;
expect(reqRespStub.goodbye.called).to.be.false;
}catch (e) {
expect.fail(e.stack);
}
});

it('should handle request - onHello(error)', async function () {
it("should handle request - onHello(error)", async function () {
const peerInfo: PeerInfo = new PeerInfo(new PeerId(Buffer.from("lodestar")));
const body: Hello = {
headForkVersion: Buffer.alloc(4),
Expand All @@ -132,8 +138,60 @@ describe("syncing", function () {
}
});

it("should disconnect on hello - incorrect headForkVersion", async function() {
const body: Hello = {
headForkVersion: Buffer.alloc(4),
finalizedRoot: Buffer.alloc(32),
finalizedEpoch: 1,
headRoot: Buffer.alloc(32),
headSlot: 1,
};

dbStub.block.getChainHead.resolves(Buffer.alloc(0));
const state = generateState();
state.fork.currentVersion = Buffer.from("efgh");
dbStub.state.get.resolves(state);
expect(await syncRpc.shouldDisconnectOnHello(body)).to.be.true;
});

it("should disconnect on hello - incorrect finalized checkpoint", async function() {
const body: Hello = {
headForkVersion: Buffer.alloc(4),
finalizedRoot: Buffer.from("xyz"),
finalizedEpoch: 1,
headRoot: Buffer.alloc(32),
headSlot: 1,
};

dbStub.block.getChainHead.resolves(Buffer.alloc(0));
const state = generateState();
state.fork.currentVersion = Buffer.alloc(4);
state.finalizedCheckpoint.epoch = 2;
dbStub.state.get.resolves(state);
signingRootStub.returns(Buffer.from("not xyz"));
expect(await syncRpc.shouldDisconnectOnHello(body)).to.be.true;
});

it("should not disconnect on hello", async function() {
const body: Hello = {
headForkVersion: Buffer.alloc(4),
finalizedRoot: Buffer.from("xyz"),
finalizedEpoch: 1,
headRoot: Buffer.alloc(32),
headSlot: 1,
};

dbStub.block.getChainHead.resolves(Buffer.alloc(0));
const state = generateState();
state.fork.currentVersion = Buffer.alloc(4);
state.finalizedCheckpoint.epoch = 1;
dbStub.state.get.resolves(state);
signingRootStub.returns(Buffer.from("xyz"));

expect(await syncRpc.shouldDisconnectOnHello(body)).to.be.false;
});

it('should handle request - onGoodbye', async function () {
it("should handle request - onGoodbye", async function () {
const peerInfo: PeerInfo = new PeerInfo(new PeerId(Buffer.from("lodestar")));
const goodbye: Goodbye = 1n;
networkStub.disconnect.resolves(0);
Expand All @@ -145,7 +203,7 @@ describe("syncing", function () {
}
});

it('should fail to handle request ', async function () {
it("should fail to handle request ", async function () {
const peerInfo: PeerInfo = new PeerInfo(new PeerId(Buffer.from("lodestar")));
try {
await syncRpc.onRequest(peerInfo, null, "null", null);
Expand Down