diff --git a/packages/contracts/source/contracts/p2p/endpoints.ts b/packages/contracts/source/contracts/p2p/endpoints.ts index 9401cc47d..7cf5c448a 100644 --- a/packages/contracts/source/contracts/p2p/endpoints.ts +++ b/packages/contracts/source/contracts/p2p/endpoints.ts @@ -33,9 +33,17 @@ export interface GetBlocksResponse extends Response { blocks: Buffer[]; } +export interface GetMessagesQuery { + blockNumber: number; + round: number; + validatorsSignedPrevote: readonly boolean[]; + validatorsSignedPrecommit: readonly boolean[]; +} + export interface GetMessagesRequest extends Request { payload: { headers: HeaderData; + query: GetMessagesQuery; }; } diff --git a/packages/contracts/source/contracts/p2p/header.ts b/packages/contracts/source/contracts/p2p/header.ts index e8c8c9545..a245957e0 100644 --- a/packages/contracts/source/contracts/p2p/header.ts +++ b/packages/contracts/source/contracts/p2p/header.ts @@ -18,8 +18,6 @@ export interface Header { validatorsSignedPrevote: readonly boolean[]; toData(): HeaderData; - getValidatorsSignedPrecommitCount(): number; - getValidatorsSignedPrevoteCount(): number; } export type HeaderFactory = () => Header; diff --git a/packages/contracts/source/contracts/p2p/peer-communicator.ts b/packages/contracts/source/contracts/p2p/peer-communicator.ts index a27325027..e37d581e6 100644 --- a/packages/contracts/source/contracts/p2p/peer-communicator.ts +++ b/packages/contracts/source/contracts/p2p/peer-communicator.ts @@ -1,6 +1,7 @@ import type { GetApiNodesResponse, GetBlocksResponse, + GetMessagesQuery, GetMessagesResponse, GetPeersResponse, GetProposalResponse, @@ -20,7 +21,7 @@ export interface PeerCommunicator { getPeers(peer: Peer): Promise; getApiNodes(peer: Peer): Promise; - getMessages(peer: Peer): Promise; + getMessages(peer: Peer, query: GetMessagesQuery): Promise; getProposal(peer: Peer): Promise; getBlocks( peer: Peer, diff --git a/packages/p2p/source/downloader/message-downloader.test.ts b/packages/p2p/source/downloader/message-downloader.test.ts index 002cb9bfc..7c0901763 100644 --- a/packages/p2p/source/downloader/message-downloader.test.ts +++ b/packages/p2p/source/downloader/message-downloader.test.ts @@ -1,5 +1,6 @@ import { Enums, Events, Identifiers } from "@mainsail/constants"; import { Application } from "@mainsail/kernel"; +import { sleep } from "@mainsail/utils"; import { describe } from "@mainsail/test-runner"; import { MessageDownloader } from "./message-downloader"; @@ -19,6 +20,7 @@ describe<{ const messageProcessor = { process: async () => Enums.Consensus.ProcessorResult.Accepted }; const cryptoConfiguration = { getMilestone: () => ({ roundValidators: 2 }) }; const state = { resetLastMessageTime: () => {} }; + const logger = { debug: () => {} }; // By the time the async listener runs, the store may already be past the applied block; // the purge must use the event payload, so this deliberately disagrees with it. const stateStore = { getBlockNumber: () => 3 }; @@ -28,8 +30,6 @@ describe<{ // prevotes and precommits (indexes 0 and 1) that we are missing. context.ourHeader = { blockNumber: 2, - getValidatorsSignedPrecommitCount: () => 0, - getValidatorsSignedPrevoteCount: () => 0, round: 0, validatorsSignedPrecommit: [false, false], validatorsSignedPrevote: [false, false], @@ -55,6 +55,7 @@ describe<{ context.app.bind(Identifiers.Consensus.Processor.Message).toConstantValue(messageProcessor); context.app.bind(Identifiers.Cryptography.Configuration).toConstantValue(cryptoConfiguration); context.app.bind(Identifiers.P2P.State).toConstantValue(state); + context.app.bind(Identifiers.Services.Log.Service).toConstantValue(logger); context.app.bind(Identifiers.State.Store).toConstantValue(stateStore); context.app.bind(Identifiers.Services.EventDispatcher.Service).toConstantValue({ listen: (event: string, listener: { handle: (payload: { data: { number: number } }) => Promise }) => { @@ -67,6 +68,398 @@ describe<{ context.downloader = context.app.resolve(MessageDownloader); }); + it("#download - should request exactly the job's block, round and our bitmaps for a partial download", ({ + downloader, + peer, + }) => { + const getMessages = stub(communicator, "getMessages").returnValue(new Promise(() => {})); + + downloader.download(peer); + + getMessages.calledWith(peer, { + blockNumber: 2, + round: 0, + validatorsSignedPrecommit: [false, false], + validatorsSignedPrevote: [false, false], + }); + }); + + it("#download - should keep the queried bitmaps as they were when the job was created", ({ + downloader, + peer, + ourHeader, + }) => { + const getMessages = stub(communicator, "getMessages").returnValue(new Promise(() => {})); + + downloader.download(peer); + + // A message arrives while the request is in flight; the live header flips, but the + // query was pinned at job creation and must not change under the request. + ourHeader.validatorsSignedPrevote[0] = true; + + getMessages.calledWith(peer, { + blockNumber: 2, + round: 0, + validatorsSignedPrecommit: [false, false], + validatorsSignedPrevote: [false, false], + }); + }); + + it("#download - should request the whole target round for a full download", ({ downloader, peer }) => { + stub(cryptoConfiguration, "getMilestone").returnValue({ roundValidators: 5 }); + peer.header.round = 1; + peer.header.validatorsSignedPrevote = [true, true, false, false, false]; + peer.header.validatorsSignedPrecommit = [false, false, false, false, false]; + + const getMessages = stub(communicator, "getMessages").returnValue(new Promise(() => {})); + + downloader.download(peer); + + // All-false bitmaps: "we have nothing of that round, send everything". + getMessages.calledWith(peer, { + blockNumber: 2, + round: 1, + validatorsSignedPrecommit: [false, false, false, false, false], + validatorsSignedPrevote: [false, false, false, false, false], + }); + }); + + it("#download - should accept a complete reply for the requested round", async ({ downloader, peer }) => { + peer.header.validatorsSignedPrevote = [true, false]; + peer.header.validatorsSignedPrecommit = [true, false]; + + stub(communicator, "getMessages").resolvedValue({ + precommits: [Buffer.alloc(1)], + prevotes: [Buffer.alloc(1)], + }); + stub(factory, "makeMessageFromBytes").resolvedValueSequence([ + { blockNumber: 2, round: 0, validatorIndex: 0 }, + { blockNumber: 2, round: 0, validatorIndex: 0 }, + ]); + const process = stub(messageProcessor, "process").resolvedValue(Enums.Consensus.ProcessorResult.Accepted); + const resetLastMessageTime = stub(state, "resetLastMessageTime"); + const banPeer = stub(peerDisposer, "banPeer"); + + downloader.download(peer); + await sleep(10); + + assert.false(downloader.isDownloading()); + process.calledTimes(2); + resetLastMessageTime.calledOnce(); + banPeer.neverCalled(); + }); + + it("#download - should not ban a peer that delivers the blocking minority it advertised", async ({ + downloader, + peer, + }) => { + // Early in a round there are legitimately no precommits, and the advertised blocking + // minority of prevotes is all the peer ever promised. + stub(cryptoConfiguration, "getMilestone").returnValue({ roundValidators: 5 }); + peer.header.round = 1; + peer.header.validatorsSignedPrevote = [true, true, false, false, false]; + peer.header.validatorsSignedPrecommit = [false, false, false, false, false]; + + stub(communicator, "getMessages").resolvedValue({ + precommits: [], + prevotes: [Buffer.alloc(1), Buffer.alloc(1)], + }); + stub(factory, "makeMessageFromBytes").resolvedValueSequence([ + { blockNumber: 2, round: 1, validatorIndex: 0 }, + { blockNumber: 2, round: 1, validatorIndex: 1 }, + ]); + const process = stub(messageProcessor, "process").resolvedValue(Enums.Consensus.ProcessorResult.Accepted); + const banPeer = stub(peerDisposer, "banPeer"); + + downloader.download(peer); + await sleep(10); + + banPeer.neverCalled(); + process.calledTimes(2); + assert.false(downloader.isDownloading()); + }); + + it("#download - should download a full round advertised by a blocking minority of precommits", ({ + downloader, + peer, + }) => { + // A blocking minority of precommits lets consensus jump rounds just like one of + // prevotes, so it justifies a full download all the same. + stub(cryptoConfiguration, "getMilestone").returnValue({ roundValidators: 5 }); + peer.header.round = 1; + peer.header.validatorsSignedPrevote = [false, false, false, false, false]; + peer.header.validatorsSignedPrecommit = [true, true, false, false, false]; + + const getMessages = stub(communicator, "getMessages").returnValue(new Promise(() => {})); + + downloader.download(peer); + + getMessages.calledWith(peer, { + blockNumber: 2, + round: 1, + validatorsSignedPrecommit: [false, false, false, false, false], + validatorsSignedPrevote: [false, false, false, false, false], + }); + }); + + it("#download - should not ban a peer that delivers the blocking minority of precommits it advertised", async ({ + downloader, + peer, + }) => { + stub(cryptoConfiguration, "getMilestone").returnValue({ roundValidators: 5 }); + peer.header.round = 1; + peer.header.validatorsSignedPrevote = [false, false, false, false, false]; + peer.header.validatorsSignedPrecommit = [true, true, false, false, false]; + + stub(communicator, "getMessages").resolvedValue({ + precommits: [Buffer.alloc(1), Buffer.alloc(1)], + prevotes: [], + }); + stub(factory, "makeMessageFromBytes").resolvedValueSequence([ + { blockNumber: 2, round: 1, validatorIndex: 0 }, + { blockNumber: 2, round: 1, validatorIndex: 1 }, + ]); + const process = stub(messageProcessor, "process").resolvedValue(Enums.Consensus.ProcessorResult.Accepted); + const banPeer = stub(peerDisposer, "banPeer"); + + downloader.download(peer); + await sleep(10); + + banPeer.neverCalled(); + process.calledTimes(2); + assert.false(downloader.isDownloading()); + }); + + it("#download - should not punish a reply below the advertised blocking minority of precommits", async ({ + downloader, + peer, + }) => { + stub(cryptoConfiguration, "getMilestone").returnValue({ roundValidators: 5 }); + peer.header.round = 1; + peer.header.validatorsSignedPrevote = [false, false, false, false, false]; + peer.header.validatorsSignedPrecommit = [true, true, false, false, false]; + + // It answers for the requested round, but below any blocking minority. A crash on the + // peer's side can explain the shortfall, so it is not provable misbehavior. + stub(communicator, "getMessages").resolvedValue({ precommits: [Buffer.alloc(1)], prevotes: [] }); + stub(factory, "makeMessageFromBytes").resolvedValue({ blockNumber: 2, round: 1, validatorIndex: 0 }); + const process = stub(messageProcessor, "process").resolvedValue(Enums.Consensus.ProcessorResult.Accepted); + const banPeer = stub(peerDisposer, "banPeer"); + const debug = stub(logger, "debug"); + + downloader.download(peer); + await sleep(10); + + banPeer.neverCalled(); + debug.calledOnce(); + process.calledOnce(); + assert.false(downloader.isDownloading()); + }); + + it("#download - should keep a reply that is missing a requested message and not punish it", async ({ + downloader, + peer, + }) => { + // The peer's header promised these messages, but a crash on its side can explain the + // shortfall; what did arrive is kept. + stub(communicator, "getMessages").resolvedValue({ precommits: [], prevotes: [Buffer.alloc(1)] }); + stub(factory, "makeMessageFromBytes").resolvedValue({ blockNumber: 2, round: 0, validatorIndex: 0 }); + const process = stub(messageProcessor, "process").resolvedValue(Enums.Consensus.ProcessorResult.Accepted); + const banPeer = stub(peerDisposer, "banPeer"); + const debug = stub(logger, "debug"); + + downloader.download(peer); + await sleep(10); + + banPeer.neverCalled(); + debug.calledOnce(); + process.calledOnce(); + assert.false(downloader.isDownloading()); + }); + + it("#download - should not punish a reply below the advertised blocking minority of prevotes", async ({ + downloader, + peer, + }) => { + stub(cryptoConfiguration, "getMilestone").returnValue({ roundValidators: 5 }); + peer.header.round = 1; + peer.header.validatorsSignedPrevote = [true, true, false, false, false]; + peer.header.validatorsSignedPrecommit = [false, false, false, false, false]; + + // It answers for the requested round, but without the promised prevotes. A crash on + // the peer's side can explain the shortfall, so it is not provable misbehavior. + stub(communicator, "getMessages").resolvedValue({ precommits: [Buffer.alloc(1)], prevotes: [] }); + stub(factory, "makeMessageFromBytes").resolvedValue({ blockNumber: 2, round: 1, validatorIndex: 0 }); + const process = stub(messageProcessor, "process").resolvedValue(Enums.Consensus.ProcessorResult.Accepted); + const banPeer = stub(peerDisposer, "banPeer"); + const debug = stub(logger, "debug"); + + downloader.download(peer); + await sleep(10); + + banPeer.neverCalled(); + debug.calledOnce(); + process.calledOnce(); + assert.false(downloader.isDownloading()); + }); + + it("#download - should not police a round the peer's header never described", async ({ downloader, peer }) => { + // The peer is a round ahead without a blocking minority, so the job settles for the + // round below — about which the peer promised nothing. A thin reply is fine. + stub(cryptoConfiguration, "getMilestone").returnValue({ roundValidators: 5 }); + peer.header.round = 1; + peer.header.validatorsSignedPrevote = [true, false, false, false, false]; + peer.header.validatorsSignedPrecommit = [true, false, false, false, false]; + + stub(communicator, "getMessages").resolvedValue({ precommits: [], prevotes: [Buffer.alloc(1)] }); + stub(factory, "makeMessageFromBytes").resolvedValue({ blockNumber: 2, round: 0, validatorIndex: 0 }); + const process = stub(messageProcessor, "process").resolvedValue(Enums.Consensus.ProcessorResult.Accepted); + const banPeer = stub(peerDisposer, "banPeer"); + + downloader.download(peer); + await sleep(10); + + banPeer.neverCalled(); + process.calledOnce(); + assert.false(downloader.isDownloading()); + }); + + it("#download - should ban a peer that answers for a round the job never requested", async ({ + downloader, + peer, + }) => { + // The responder serves exactly the queried round or nothing; anything else cannot + // come from an honest peer. + stub(communicator, "getMessages").resolvedValue({ precommits: [], prevotes: [Buffer.alloc(1)] }); + stub(factory, "makeMessageFromBytes").resolvedValue({ blockNumber: 2, round: 1, validatorIndex: 0 }); + const process = stub(messageProcessor, "process"); + const banPeer = stub(peerDisposer, "banPeer"); + + downloader.download(peer); + await sleep(10); + + banPeer.calledOnce(); + process.neverCalled(); + assert.false(downloader.isDownloading()); + }); + + it("#download - should ban a peer that answers for a block the job never requested", async ({ + downloader, + peer, + }) => { + stub(communicator, "getMessages").resolvedValue({ precommits: [], prevotes: [Buffer.alloc(1)] }); + stub(factory, "makeMessageFromBytes").resolvedValue({ blockNumber: 3, round: 0, validatorIndex: 0 }); + const process = stub(messageProcessor, "process"); + const banPeer = stub(peerDisposer, "banPeer"); + + downloader.download(peer); + await sleep(10); + + banPeer.calledOnce(); + process.neverCalled(); + assert.false(downloader.isDownloading()); + }); + + it("#download - should ban the peer when one list mixes rounds", async ({ downloader, peer }) => { + // Timing cannot explain a single list about two different rounds. + stub(communicator, "getMessages").resolvedValue({ + precommits: [], + prevotes: [Buffer.alloc(1), Buffer.alloc(1)], + }); + stub(factory, "makeMessageFromBytes").resolvedValueSequence([ + { blockNumber: 2, round: 0, validatorIndex: 0 }, + { blockNumber: 2, round: 1, validatorIndex: 1 }, + ]); + const banPeer = stub(peerDisposer, "banPeer"); + + downloader.download(peer); + await sleep(10); + + banPeer.calledOnce(); + assert.false(downloader.isDownloading()); + }); + + it("#download - should ban the peer and retry when a message is invalid", async ({ downloader, peer }) => { + stub(communicator, "getMessages").resolvedValue({ precommits: [], prevotes: [Buffer.alloc(1)] }); + stub(factory, "makeMessageFromBytes").resolvedValue({ blockNumber: 2, round: 0, validatorIndex: 0 }); + stub(messageProcessor, "process").resolvedValue(Enums.Consensus.ProcessorResult.Invalid); + const banPeer = stub(peerDisposer, "banPeer"); + + downloader.download(peer); + await sleep(10); + + banPeer.calledOnce(); + assert.false(downloader.isDownloading()); + }); + + it("#download - should release the downloads under the round the job actually used", async ({ + downloader, + peer, + }) => { + // Peer a round ahead without a blocking minority: the job runs for our round and its + // slot must be marked and cleared under that same round, or it can never be re-pulled. + stub(cryptoConfiguration, "getMilestone").returnValue({ roundValidators: 5 }); + peer.header.round = 1; + peer.header.validatorsSignedPrevote = [true, false, false, false, false]; + peer.header.validatorsSignedPrecommit = [false, false, false, false, false]; + + const getMessages = stub(communicator, "getMessages").resolvedValue({ precommits: [], prevotes: [] }); + + downloader.download(peer); + await sleep(10); + + assert.false(downloader.isDownloading()); + + downloader.download(peer); + await sleep(10); + + getMessages.calledTimes(2); + }); + + it("#download - should not report itself busy after probing a peer with nothing to serve", ({ + downloader, + peer, + }) => { + const getMessages = stub(communicator, "getMessages"); + + peer.header.validatorsSignedPrevote = [false, false]; + peer.header.validatorsSignedPrecommit = [false, false]; + downloader.download(peer); + + getMessages.neverCalled(); + assert.false(downloader.isDownloading()); + }); + + it("#download - should not download twice for the same messages while pending", ({ downloader, peer }) => { + const getMessages = stub(communicator, "getMessages").returnValue(new Promise(() => {})); + + downloader.download(peer); + downloader.download(peer); + + getMessages.calledOnce(); + }); + + it("#download - should release the downloads on an empty reply and allow them to be re-pulled", async ({ + downloader, + peer, + }) => { + const getMessages = stub(communicator, "getMessages").resolvedValue({ precommits: [], prevotes: [] }); + const banPeer = stub(peerDisposer, "banPeer"); + + downloader.download(peer); + assert.true(downloader.isDownloading()); + + await sleep(10); + + assert.false(downloader.isDownloading()); + banPeer.neverCalled(); + + downloader.download(peer); + await sleep(10); + + getMessages.calledTimes(2); + }); + it("#initialize - should purge pending downloads for the applied block number", async (context) => { // The common case: the handler runs before anything else is applied. stub(stateStore, "getBlockNumber").returnValue(2); diff --git a/packages/p2p/source/downloader/message-downloader.ts b/packages/p2p/source/downloader/message-downloader.ts index a88558963..2cb1fa9ba 100644 --- a/packages/p2p/source/downloader/message-downloader.ts +++ b/packages/p2p/source/downloader/message-downloader.ts @@ -1,6 +1,6 @@ import type { Contracts } from "@mainsail/contracts"; -import { isMajority, isMinority } from "@mainsail/blockchain-utils"; +import { isMinority } from "@mainsail/blockchain-utils"; import { Enums, Events, Identifiers } from "@mainsail/constants"; import { inject, injectable, postConstruct } from "@mainsail/container"; import { ensureError } from "@mainsail/utils"; @@ -12,6 +12,8 @@ type DownloadsByRound = { prevotes: boolean[]; }; +class IncompleteResponseError extends Error {} + type DownloadJob = { isFullDownload: boolean; peer: Contracts.P2P.Peer; @@ -60,6 +62,9 @@ export class MessageDownloader implements Contracts.P2P.Downloader { @inject(Identifiers.P2P.State) private readonly state!: Contracts.P2P.State; + @inject(Identifiers.Services.Log.Service) + private readonly logger!: Contracts.Kernel.Logger; + #fullDownloadsByBlockNumber: Map> = new Map(); #downloadsByBlockNumber: Map> = new Map(); @@ -83,7 +88,6 @@ export class MessageDownloader implements Contracts.P2P.Downloader { const header = this.headerFactory(); let peers = this.repository.getPeers(); - // Create download jobs as long as we can download while ((peers = peers.filter((peer) => this.#canDownload(header, peer.header))) && peers.length > 0) { this.download(getRandomPeer(peers)); } @@ -101,7 +105,7 @@ export class MessageDownloader implements Contracts.P2P.Downloader { const round = this.#getHighestRoundToDownload(ourHeader, peer.header); if (ourHeader.round === round) { - const downloads = this.#getDownloadsByRound(peer.header.blockNumber, peer.header.round); + const downloads = this.#getDownloadsByRound(peer.header.blockNumber, round); const job: DownloadJob = { blockNumber: ourHeader.blockNumber, @@ -145,7 +149,7 @@ export class MessageDownloader implements Contracts.P2P.Downloader { const round = this.#getHighestRoundToDownload(ourHeader, peerHeader); if (ourHeader.round === round) { - const downloads = this.#getDownloadsByRound(peerHeader.blockNumber, peerHeader.round); + const downloads = this.#peekDownloadsByRound(peerHeader.blockNumber, round); const prevoteIndexes = this.#getPrevoteIndexesToDownload(ourHeader, peerHeader, downloads.prevotes); const precommitIndexes = this.#getPrecommitIndexesToDownload(ourHeader, peerHeader, downloads.precommits); @@ -167,7 +171,10 @@ export class MessageDownloader implements Contracts.P2P.Downloader { const { roundValidators } = this.cryptoConfiguration.getMilestone(ourHeader.blockNumber); - if (isMinority(peerHeader.validatorsSignedPrevote.filter(Boolean).length, roundValidators)) { + if ( + isMinority(peerHeader.validatorsSignedPrevote.filter(Boolean).length, roundValidators) || + isMinority(peerHeader.validatorsSignedPrecommit.filter(Boolean).length, roundValidators) + ) { return peerHeader.round; } @@ -204,35 +211,31 @@ export class MessageDownloader implements Contracts.P2P.Downloader { const roundsByBlockNumber = this.#downloadsByBlockNumber.get(blockNumber)!; if (!roundsByBlockNumber.has(round)) { - roundsByBlockNumber.set(round, { - precommits: Array.from({ - length: this.cryptoConfiguration.getMilestone(blockNumber).roundValidators, - }).fill(false), - prevotes: Array.from({ - length: this.cryptoConfiguration.getMilestone(blockNumber).roundValidators, - }).fill(false), - }); + roundsByBlockNumber.set(round, this.#makeDownloadsByRound(blockNumber)); } return roundsByBlockNumber.get(round)!; } - #checkMessage(message: Contracts.Crypto.Message, firstMessage: Contracts.Crypto.Message, job: DownloadJob): void { - if (message.blockNumber !== firstMessage.blockNumber || message.round !== firstMessage.round) { - throw new Error( - `Received message blockNumber ${message.blockNumber} and round ${message.round} does not match expected blockNumber ${firstMessage.blockNumber} and round ${firstMessage.round}`, - ); - } + #peekDownloadsByRound(blockNumber: number, round: number): DownloadsByRound { + return this.#downloadsByBlockNumber.get(blockNumber)?.get(round) ?? this.#makeDownloadsByRound(blockNumber); + } + + #makeDownloadsByRound(blockNumber: number): DownloadsByRound { + const { roundValidators } = this.cryptoConfiguration.getMilestone(blockNumber); + + return { + precommits: Array.from({ length: roundValidators }).fill(false), + prevotes: Array.from({ length: roundValidators }).fill(false), + }; + } - if (message.blockNumber !== job.blockNumber) { + #checkMessage(message: Contracts.Crypto.Message, job: DownloadJob): void { + if (message.blockNumber !== job.blockNumber || message.round !== job.round) { throw new Error( - `Received message blockNumber ${message.blockNumber} does not match expected blockNumber ${job.blockNumber}`, + `Received message blockNumber ${message.blockNumber} and round ${message.round} does not match requested blockNumber ${job.blockNumber} and round ${job.round}`, ); } - - if (message.round < job.round) { - throw new Error(`Received message round ${message.round} is lower than requested round ${job.round}`); - } } #checkResponse( @@ -243,17 +246,17 @@ export class MessageDownloader implements Contracts.P2P.Downloader { const prevotes = [...prevotesMap.values()]; const precommits = [...precommitsMap.values()]; - // Allow response to be empty if (prevotes.length === 0 && precommits.length === 0) { return; } this.state.resetLastMessageTime(); - // Check actual received round, because we might have received a full response even if we marked request as a partial - const receivedRound = prevotes.length > 0 ? prevotes[0].round : precommits[0].round; + if (job.peerHeader.round !== job.round) { + return; + } - if (job.ourHeader.round < receivedRound) { + if (job.isFullDownload) { this.#checkFullRoundResponse(prevotesMap, precommitsMap, job); } else { this.#checkPartialRoundResponse(prevotesMap, precommitsMap, job); @@ -267,12 +270,8 @@ export class MessageDownloader implements Contracts.P2P.Downloader { ) { const { roundValidators } = this.cryptoConfiguration.getMilestone(job.blockNumber); - if (!isMajority(prevotes.size + job.ourHeader.getValidatorsSignedPrevoteCount(), roundValidators)) { - throw new Error(`Peer didn't return enough prevotes for +2/3 majority`); - } - - if (!isMajority(precommits.size + job.ourHeader.getValidatorsSignedPrecommitCount(), roundValidators)) { - throw new Error(`Peer didn't return enough precommits for +2/3 majority`); + if (!isMinority(prevotes.size, roundValidators) && !isMinority(precommits.size, roundValidators)) { + throw new IncompleteResponseError(`Peer didn't return a blocking minority of prevotes or precommits`); } } @@ -281,16 +280,15 @@ export class MessageDownloader implements Contracts.P2P.Downloader { precommits: Map, job: DownloadJob, ) { - // Check if received all the requested data for (const index of job.prevoteIndexes) { if (!prevotes.has(index)) { - throw new Error(`Missing prevote for validator ${index}`); + throw new IncompleteResponseError(`Missing prevote for validator ${index}`); } } for (const index of job.precommitIndexes) { if (!precommits.has(index)) { - throw new Error(`Missing precommit for validator ${index}`); + throw new IncompleteResponseError(`Missing precommit for validator ${index}`); } } } @@ -299,18 +297,26 @@ export class MessageDownloader implements Contracts.P2P.Downloader { let error: Error | undefined; try { - const result = await this.communicator.getMessages(job.peer); + const { roundValidators } = this.cryptoConfiguration.getMilestone(job.blockNumber); + const nothingSigned = Array.from({ length: roundValidators }).fill(false); + + const result = await this.communicator.getMessages(job.peer, { + blockNumber: job.blockNumber, + round: job.round, + validatorsSignedPrecommit: job.isFullDownload + ? nothingSigned + : [...job.ourHeader.validatorsSignedPrecommit], + validatorsSignedPrevote: job.isFullDownload + ? nothingSigned + : [...job.ourHeader.validatorsSignedPrevote], + }); - let firstPrevote: Contracts.Crypto.Message | undefined; const prevotes: Map = new Map(); for (const buffer of result.prevotes) { const prevote = await this.factory.makeMessageFromBytes(buffer); prevotes.set(prevote.validatorIndex, prevote); - if (firstPrevote === undefined) { - firstPrevote = prevote; - } - this.#checkMessage(prevote, firstPrevote, job); + this.#checkMessage(prevote, job); const response = await this.messageProcessor.process(prevote, false); @@ -319,16 +325,12 @@ export class MessageDownloader implements Contracts.P2P.Downloader { } } - let firstPrecommit: Contracts.Crypto.Message | undefined; const precommits: Map = new Map(); for (const buffer of result.precommits) { const precommit = await this.factory.makeMessageFromBytes(buffer); precommits.set(precommit.validatorIndex, precommit); - if (firstPrecommit === undefined) { - firstPrecommit = precommit; - } - this.#checkMessage(precommit, firstPrecommit, job); + this.#checkMessage(precommit, job); const response = await this.messageProcessor.process(precommit, false); @@ -345,7 +347,12 @@ export class MessageDownloader implements Contracts.P2P.Downloader { this.#removeDownloadJob(job); if (error) { - this.peerDisposer.banPeer(job.peer.ip, error); + if (error instanceof IncompleteResponseError) { + this.logger.debug(`Incomplete response from ${job.peer.ip}: ${error.message}`, "p2p"); + } else { + this.peerDisposer.banPeer(job.peer.ip, error); + } + this.tryToDownload(); } } @@ -371,14 +378,12 @@ export class MessageDownloader implements Contracts.P2P.Downloader { #removeFullDownloadJob(job: DownloadJob) { this.#fullDownloadsByBlockNumber.get(job.blockNumber)?.delete(job.round); - // Cleanup if (this.#fullDownloadsByBlockNumber.get(job.blockNumber)?.size === 0) { this.#fullDownloadsByBlockNumber.delete(job.blockNumber); } } #removePartialDownloadJob(job: DownloadJob) { - // Return if the blockNumber was already removed, because the block was applied. const roundsByBlockNumber = this.#downloadsByBlockNumber.get(job.blockNumber); if (!roundsByBlockNumber) { return; @@ -397,7 +402,6 @@ export class MessageDownloader implements Contracts.P2P.Downloader { downloadsByRound.precommits[index] = false; } - // Cleanup if ( downloadsByRound.prevotes.every((value) => !value) && downloadsByRound.precommits.every((value) => !value) @@ -441,7 +445,6 @@ export class MessageDownloader implements Contracts.P2P.Downloader { ): number[] { const indexes: number[] = []; - // Request missing messages for (const [index, precommit] of messages.entries()) { if (!precommit && peerValidatorsSignedMessage[index] && !ourValidatorsSignedMessage[index]) { indexes.push(index); diff --git a/packages/p2p/source/header.ts b/packages/p2p/source/header.ts index 26117778e..89ff94f6f 100644 --- a/packages/p2p/source/header.ts +++ b/packages/p2p/source/header.ts @@ -44,12 +44,4 @@ export class Header implements Contracts.P2P.Header { version: this.app.version(), }; } - - public getValidatorsSignedPrecommitCount(): number { - return this.validatorsSignedPrecommit.filter(Boolean).length; - } - - public getValidatorsSignedPrevoteCount(): number { - return this.validatorsSignedPrevote.filter(Boolean).length; - } } diff --git a/packages/p2p/source/peer-communicator.ts b/packages/p2p/source/peer-communicator.ts index 6e03163c4..e306e535b 100644 --- a/packages/p2p/source/peer-communicator.ts +++ b/packages/p2p/source/peer-communicator.ts @@ -78,11 +78,14 @@ export class PeerCommunicator implements Contracts.P2P.PeerCommunicator { ); } - public async getMessages(peer: Contracts.P2P.Peer): Promise { + public async getMessages( + peer: Contracts.P2P.Peer, + query: Contracts.P2P.GetMessagesQuery, + ): Promise { const response = await this.#emit( peer, Routes.GetMessages, - {}, + { query }, { timeout: 5000 }, ); return response.data; diff --git a/packages/p2p/source/socket-server/codecs/proto/get-messages.proto b/packages/p2p/source/socket-server/codecs/proto/get-messages.proto index b48e1da12..f295127bd 100644 --- a/packages/p2p/source/socket-server/codecs/proto/get-messages.proto +++ b/packages/p2p/source/socket-server/codecs/proto/get-messages.proto @@ -4,8 +4,16 @@ import "shared.proto"; package getMessages; +message GetMessagesQuery { + uint32 blockNumber = 1; + uint32 round = 2; + repeated bool validatorsSignedPrevote = 3; + repeated bool validatorsSignedPrecommit = 4; +} + message GetMessagesRequest { shared.Headers headers = 1; + GetMessagesQuery query = 2; } message GetMessagesResponse { diff --git a/packages/p2p/source/socket-server/codecs/proto/protos.d.ts b/packages/p2p/source/socket-server/codecs/proto/protos.d.ts index 8fcc059b2..52664f668 100644 --- a/packages/p2p/source/socket-server/codecs/proto/protos.d.ts +++ b/packages/p2p/source/socket-server/codecs/proto/protos.d.ts @@ -907,6 +907,141 @@ export namespace getBlocks { /** Namespace getMessages. */ export namespace getMessages { + /** + * Properties of a GetMessagesQuery. + * @deprecated Use getMessages.GetMessagesQuery.$Properties instead. + */ + interface IGetMessagesQuery extends getMessages.GetMessagesQuery.$Properties { + } + + /** Represents a GetMessagesQuery. */ + class GetMessagesQuery { + + /** + * Constructs a new GetMessagesQuery. + * @param [properties] Properties to set + */ + constructor(properties?: getMessages.GetMessagesQuery.$Properties); + + /** Unknown fields preserved while decoding when enabled */ + $unknowns?: Uint8Array[]; + + /** GetMessagesQuery blockNumber. */ + blockNumber: number; + + /** GetMessagesQuery round. */ + round: number; + + /** GetMessagesQuery validatorsSignedPrevote. */ + validatorsSignedPrevote: boolean[]; + + /** GetMessagesQuery validatorsSignedPrecommit. */ + validatorsSignedPrecommit: boolean[]; + + /** + * Creates a new GetMessagesQuery instance using the specified properties. + * @param [properties] Properties to set + * @returns GetMessagesQuery instance + */ + static create(properties: getMessages.GetMessagesQuery.$Shape): getMessages.GetMessagesQuery & getMessages.GetMessagesQuery.$Shape; + static create(properties?: getMessages.GetMessagesQuery.$Properties): getMessages.GetMessagesQuery; + + /** + * Encodes the specified GetMessagesQuery message. Does not implicitly {@link getMessages.GetMessagesQuery.verify|verify} messages. + * @param message GetMessagesQuery message or plain object to encode + * @param [writer] Writer to encode to + * @returns Writer + */ + static encode(message: getMessages.GetMessagesQuery.$Properties, writer?: $protobuf.Writer): $protobuf.Writer; + + /** + * Encodes the specified GetMessagesQuery message, length delimited. Does not implicitly {@link getMessages.GetMessagesQuery.verify|verify} messages. + * @param message GetMessagesQuery message or plain object to encode + * @param [writer] Writer to encode to + * @returns Writer + */ + static encodeDelimited(message: getMessages.GetMessagesQuery.$Properties, writer?: $protobuf.Writer): $protobuf.Writer; + + /** + * Decodes a GetMessagesQuery message from the specified reader or buffer. + * @param reader Reader or buffer to decode from + * @param [length] Message length if known beforehand + * @returns {getMessages.GetMessagesQuery & getMessages.GetMessagesQuery.$Shape} GetMessagesQuery + * @throws {Error} If the payload is not a reader or valid buffer + * @throws {$protobuf.util.ProtocolError} If required fields are missing + */ + static decode(reader: ($protobuf.Reader|Uint8Array), length?: number): getMessages.GetMessagesQuery & getMessages.GetMessagesQuery.$Shape; + + /** + * Decodes a GetMessagesQuery message from the specified reader or buffer, length delimited. + * @param reader Reader or buffer to decode from + * @returns {getMessages.GetMessagesQuery & getMessages.GetMessagesQuery.$Shape} GetMessagesQuery + * @throws {Error} If the payload is not a reader or valid buffer + * @throws {$protobuf.util.ProtocolError} If required fields are missing + */ + static decodeDelimited(reader: ($protobuf.Reader|Uint8Array)): getMessages.GetMessagesQuery & getMessages.GetMessagesQuery.$Shape; + + /** + * Verifies a GetMessagesQuery message. + * @param message Plain object to verify + * @returns `null` if valid, otherwise the reason why it is not + */ + static verify(message: { [k: string]: any }): (string|null); + + /** + * Creates a GetMessagesQuery message from a plain object. Also converts values to their respective internal types. + * @param object Plain object + * @returns GetMessagesQuery + */ + static fromObject(object: { [k: string]: any }): getMessages.GetMessagesQuery; + + /** + * Creates a plain object from a GetMessagesQuery message. Also converts values to other types if specified. + * @param message GetMessagesQuery + * @param [options] Conversion options + * @returns Plain object + */ + static toObject(message: getMessages.GetMessagesQuery, options?: $protobuf.IConversionOptions): { [k: string]: any }; + + /** + * Converts this GetMessagesQuery to JSON. + * @returns JSON object + */ + toJSON(): { [k: string]: any }; + + /** + * Gets the type url for GetMessagesQuery + * @param [prefix] Custom type url prefix, defaults to `"type.googleapis.com"` + * @returns The type url + */ + static getTypeUrl(prefix?: string): string; + } + + namespace GetMessagesQuery { + + /** Properties of a GetMessagesQuery. */ + interface $Properties { + + /** GetMessagesQuery blockNumber */ + blockNumber?: (number|null); + + /** GetMessagesQuery round */ + round?: (number|null); + + /** GetMessagesQuery validatorsSignedPrevote */ + validatorsSignedPrevote?: (boolean[]|null); + + /** GetMessagesQuery validatorsSignedPrecommit */ + validatorsSignedPrecommit?: (boolean[]|null); + + /** Unknown fields preserved while decoding when enabled */ + $unknowns?: Uint8Array[]; + } + + /** Shape of a GetMessagesQuery. */ + type $Shape = getMessages.GetMessagesQuery.$Properties; + } + /** * Properties of a GetMessagesRequest. * @deprecated Use getMessages.GetMessagesRequest.$Properties instead. @@ -929,6 +1064,9 @@ export namespace getMessages { /** GetMessagesRequest headers. */ headers?: (shared.Headers.$Properties|null); + /** GetMessagesRequest query. */ + query?: (getMessages.GetMessagesQuery.$Properties|null); + /** * Creates a new GetMessagesRequest instance using the specified properties. * @param [properties] Properties to set @@ -1016,6 +1154,9 @@ export namespace getMessages { /** GetMessagesRequest headers */ headers?: (shared.Headers.$Properties|null); + /** GetMessagesRequest query */ + query?: (getMessages.GetMessagesQuery.$Properties|null); + /** Unknown fields preserved while decoding when enabled */ $unknowns?: Uint8Array[]; } diff --git a/packages/p2p/source/socket-server/codecs/proto/protos.js b/packages/p2p/source/socket-server/codecs/proto/protos.js index cf53f9db5..af91c0047 100644 --- a/packages/p2p/source/socket-server/codecs/proto/protos.js +++ b/packages/p2p/source/socket-server/codecs/proto/protos.js @@ -2318,12 +2318,409 @@ export const getMessages = $root.getMessages = (() => { */ const getMessages = {}; + getMessages.GetMessagesQuery = (function() { + + /** + * Properties of a GetMessagesQuery. + * @typedef {Object} getMessages.GetMessagesQuery.$Properties + * @property {number|null} [blockNumber] GetMessagesQuery blockNumber + * @property {number|null} [round] GetMessagesQuery round + * @property {Array.|null} [validatorsSignedPrevote] GetMessagesQuery validatorsSignedPrevote + * @property {Array.|null} [validatorsSignedPrecommit] GetMessagesQuery validatorsSignedPrecommit + * @property {Array.} [$unknowns] Unknown fields preserved while decoding when enabled + */ + + /** + * Properties of a GetMessagesQuery. + * @memberof getMessages + * @interface IGetMessagesQuery + * @augments getMessages.GetMessagesQuery.$Properties + * @deprecated Use getMessages.GetMessagesQuery.$Properties instead. + */ + + /** + * Shape of a GetMessagesQuery. + * @typedef {getMessages.GetMessagesQuery.$Properties} getMessages.GetMessagesQuery.$Shape + */ + + /** + * Constructs a new GetMessagesQuery. + * @memberof getMessages + * @classdesc Represents a GetMessagesQuery. + * @constructor + * @param {getMessages.GetMessagesQuery.$Properties=} [properties] Properties to set + * @property {Array.} [$unknowns] Unknown fields preserved while decoding when enabled + */ + const GetMessagesQuery = function (properties) { + this.validatorsSignedPrevote = []; + this.validatorsSignedPrecommit = []; + if (properties) + for (let keys = $Object.keys(properties), i = 0; i < keys.length; ++i) + if (properties[keys[i]] != null && keys[i] !== "__proto__") + this[keys[i]] = properties[keys[i]]; + }; + + /** + * GetMessagesQuery blockNumber. + * @member {number} blockNumber + * @memberof getMessages.GetMessagesQuery + * @instance + */ + GetMessagesQuery.prototype.blockNumber = 0; + + /** + * GetMessagesQuery round. + * @member {number} round + * @memberof getMessages.GetMessagesQuery + * @instance + */ + GetMessagesQuery.prototype.round = 0; + + /** + * GetMessagesQuery validatorsSignedPrevote. + * @member {Array.} validatorsSignedPrevote + * @memberof getMessages.GetMessagesQuery + * @instance + */ + GetMessagesQuery.prototype.validatorsSignedPrevote = $util.emptyArray; + + /** + * GetMessagesQuery validatorsSignedPrecommit. + * @member {Array.} validatorsSignedPrecommit + * @memberof getMessages.GetMessagesQuery + * @instance + */ + GetMessagesQuery.prototype.validatorsSignedPrecommit = $util.emptyArray; + + /** + * Creates a new GetMessagesQuery instance using the specified properties. + * @function create + * @memberof getMessages.GetMessagesQuery + * @static + * @param {getMessages.GetMessagesQuery.$Properties=} [properties] Properties to set + * @returns {getMessages.GetMessagesQuery} GetMessagesQuery instance + * @type {{ + * (properties: getMessages.GetMessagesQuery.$Shape): getMessages.GetMessagesQuery & getMessages.GetMessagesQuery.$Shape; + * (properties?: getMessages.GetMessagesQuery.$Properties): getMessages.GetMessagesQuery; + * }} + */ + GetMessagesQuery.create = function(properties) { + return new GetMessagesQuery(properties); + }; + + /** + * Encodes the specified GetMessagesQuery message. Does not implicitly {@link getMessages.GetMessagesQuery.verify|verify} messages. + * @function encode + * @memberof getMessages.GetMessagesQuery + * @static + * @param {getMessages.GetMessagesQuery.$Properties} message GetMessagesQuery message or plain object to encode + * @param {$protobuf.Writer} [writer] Writer to encode to + * @returns {$protobuf.Writer} Writer + */ + GetMessagesQuery.encode = function (message, writer, _depth) { + if (!writer) + writer = $Writer.create(); + if (_depth === $undefined) + _depth = 0; + if (_depth > $util.recursionLimit) + throw $Error("max depth exceeded"); + if (message.blockNumber != null && $Object.hasOwnProperty.call(message, "blockNumber")) + writer.uint32(/* id 1, wireType 0 =*/8).uint32(message.blockNumber); + if (message.round != null && $Object.hasOwnProperty.call(message, "round")) + writer.uint32(/* id 2, wireType 0 =*/16).uint32(message.round); + if (message.validatorsSignedPrevote != null && message.validatorsSignedPrevote.length) { + writer.uint32(/* id 3, wireType 2 =*/26).fork(); + for (let i = 0; i < message.validatorsSignedPrevote.length; ++i) + writer.bool(message.validatorsSignedPrevote[i]); + writer.ldelim(); + } + if (message.validatorsSignedPrecommit != null && message.validatorsSignedPrecommit.length) { + writer.uint32(/* id 4, wireType 2 =*/34).fork(); + for (let i = 0; i < message.validatorsSignedPrecommit.length; ++i) + writer.bool(message.validatorsSignedPrecommit[i]); + writer.ldelim(); + } + if (message.$unknowns != null && $Object.hasOwnProperty.call(message, "$unknowns")) + for (let i = 0; i < message.$unknowns.length; ++i) + writer.raw(message.$unknowns[i]); + return writer; + }; + + /** + * Encodes the specified GetMessagesQuery message, length delimited. Does not implicitly {@link getMessages.GetMessagesQuery.verify|verify} messages. + * @function encodeDelimited + * @memberof getMessages.GetMessagesQuery + * @static + * @param {getMessages.GetMessagesQuery.$Properties} message GetMessagesQuery message or plain object to encode + * @param {$protobuf.Writer} [writer] Writer to encode to + * @returns {$protobuf.Writer} Writer + */ + GetMessagesQuery.encodeDelimited = function(message, writer) { + return this.encode(message, writer && writer.len ? writer.fork() : writer).ldelim(); + }; + + /** + * Decodes a GetMessagesQuery message from the specified reader or buffer. + * @function decode + * @memberof getMessages.GetMessagesQuery + * @static + * @param {$protobuf.Reader|Uint8Array} reader Reader or buffer to decode from + * @param {number} [length] Message length if known beforehand + * @returns {getMessages.GetMessagesQuery & getMessages.GetMessagesQuery.$Shape} GetMessagesQuery + * @throws {Error} If the payload is not a reader or valid buffer + * @throws {$protobuf.util.ProtocolError} If required fields are missing + */ + GetMessagesQuery.decode = function (reader, length, _end, _depth, _target) { + if (!(reader instanceof $Reader)) + reader = $Reader.create(reader); + if (_depth === $undefined) + _depth = 0; + if (_depth > $Reader.recursionLimit) + throw $Error("max depth exceeded"); + let end = length === $undefined ? reader.len : reader.pos + length, message = _target || new $root.getMessages.GetMessagesQuery(), value; + while (reader.pos < end) { + let start = reader.pos; + let tag = reader.tag(); + if (tag === _end) { + _end = $undefined; + break; + } + let wireType = tag & 7; + switch (tag >>>= 3) { + case 1: { + if (wireType !== 0) + break; + if (value = reader.uint32()) + message.blockNumber = value; + else + delete message.blockNumber; + continue; + } + case 2: { + if (wireType !== 0) + break; + if (value = reader.uint32()) + message.round = value; + else + delete message.round; + continue; + } + case 3: { + if (wireType === 2) { + if (!(message.validatorsSignedPrevote && message.validatorsSignedPrevote.length)) + message.validatorsSignedPrevote = []; + let end2 = reader.uint32() + reader.pos; + while (reader.pos < end2) + message.validatorsSignedPrevote.push(reader.bool()); + continue; + } + if (wireType !== 0) + break; + if (!(message.validatorsSignedPrevote && message.validatorsSignedPrevote.length)) + message.validatorsSignedPrevote = []; + message.validatorsSignedPrevote.push(reader.bool()); + continue; + } + case 4: { + if (wireType === 2) { + if (!(message.validatorsSignedPrecommit && message.validatorsSignedPrecommit.length)) + message.validatorsSignedPrecommit = []; + let end2 = reader.uint32() + reader.pos; + while (reader.pos < end2) + message.validatorsSignedPrecommit.push(reader.bool()); + continue; + } + if (wireType !== 0) + break; + if (!(message.validatorsSignedPrecommit && message.validatorsSignedPrecommit.length)) + message.validatorsSignedPrecommit = []; + message.validatorsSignedPrecommit.push(reader.bool()); + continue; + } + } + reader.skipType(wireType, _depth, tag); + if (!reader.discardUnknown) { + $util.makeProp(message, "$unknowns", false); + (message.$unknowns || (message.$unknowns = [])).push(reader.raw(start, reader.pos)); + } + } + if (_end !== $undefined) + throw $Error("missing end group"); + return message; + }; + + /** + * Decodes a GetMessagesQuery message from the specified reader or buffer, length delimited. + * @function decodeDelimited + * @memberof getMessages.GetMessagesQuery + * @static + * @param {$protobuf.Reader|Uint8Array} reader Reader or buffer to decode from + * @returns {getMessages.GetMessagesQuery & getMessages.GetMessagesQuery.$Shape} GetMessagesQuery + * @throws {Error} If the payload is not a reader or valid buffer + * @throws {$protobuf.util.ProtocolError} If required fields are missing + */ + GetMessagesQuery.decodeDelimited = function(reader) { + if (!(reader instanceof $Reader)) + reader = new $Reader(reader); + return this.decode(reader, reader.uint32()); + }; + + /** + * Verifies a GetMessagesQuery message. + * @function verify + * @memberof getMessages.GetMessagesQuery + * @static + * @param {Object.} message Plain object to verify + * @returns {string|null} `null` if valid, otherwise the reason why it is not + */ + GetMessagesQuery.verify = function (message, _depth) { + if (typeof message !== "object" || message === null) + return "object expected"; + if (_depth === $undefined) + _depth = 0; + if (_depth > $util.recursionLimit) + return "max depth exceeded"; + if (message.blockNumber != null && $Object.hasOwnProperty.call(message, "blockNumber")) + if (!$util.isInteger(message.blockNumber)) + return "blockNumber: integer expected"; + if (message.round != null && $Object.hasOwnProperty.call(message, "round")) + if (!$util.isInteger(message.round)) + return "round: integer expected"; + if (message.validatorsSignedPrevote != null && $Object.hasOwnProperty.call(message, "validatorsSignedPrevote")) { + if (!$Array.isArray(message.validatorsSignedPrevote)) + return "validatorsSignedPrevote: array expected"; + for (let i = 0; i < message.validatorsSignedPrevote.length; ++i) + if (typeof message.validatorsSignedPrevote[i] !== "boolean") + return "validatorsSignedPrevote: boolean[] expected"; + } + if (message.validatorsSignedPrecommit != null && $Object.hasOwnProperty.call(message, "validatorsSignedPrecommit")) { + if (!$Array.isArray(message.validatorsSignedPrecommit)) + return "validatorsSignedPrecommit: array expected"; + for (let i = 0; i < message.validatorsSignedPrecommit.length; ++i) + if (typeof message.validatorsSignedPrecommit[i] !== "boolean") + return "validatorsSignedPrecommit: boolean[] expected"; + } + return null; + }; + + /** + * Creates a GetMessagesQuery message from a plain object. Also converts values to their respective internal types. + * @function fromObject + * @memberof getMessages.GetMessagesQuery + * @static + * @param {Object.} object Plain object + * @returns {getMessages.GetMessagesQuery} GetMessagesQuery + */ + GetMessagesQuery.fromObject = function (object, _depth) { + if (object instanceof $root.getMessages.GetMessagesQuery) + return object; + if (!$util.isObject(object)) + throw $TypeError(".getMessages.GetMessagesQuery: object expected"); + if (_depth === $undefined) + _depth = 0; + if (_depth > $util.recursionLimit) + throw $Error("max depth exceeded"); + let message = new $root.getMessages.GetMessagesQuery(); + if (object.blockNumber != null) + if ($Number(object.blockNumber) !== 0) + message.blockNumber = object.blockNumber >>> 0; + if (object.round != null) + if ($Number(object.round) !== 0) + message.round = object.round >>> 0; + if (object.validatorsSignedPrevote) { + if (!$Array.isArray(object.validatorsSignedPrevote)) + throw $TypeError(".getMessages.GetMessagesQuery.validatorsSignedPrevote: array expected"); + message.validatorsSignedPrevote = $Array(object.validatorsSignedPrevote.length); + for (let i = 0; i < object.validatorsSignedPrevote.length; ++i) + message.validatorsSignedPrevote[i] = $Boolean(object.validatorsSignedPrevote[i]); + } + if (object.validatorsSignedPrecommit) { + if (!$Array.isArray(object.validatorsSignedPrecommit)) + throw $TypeError(".getMessages.GetMessagesQuery.validatorsSignedPrecommit: array expected"); + message.validatorsSignedPrecommit = $Array(object.validatorsSignedPrecommit.length); + for (let i = 0; i < object.validatorsSignedPrecommit.length; ++i) + message.validatorsSignedPrecommit[i] = $Boolean(object.validatorsSignedPrecommit[i]); + } + return message; + }; + + /** + * Creates a plain object from a GetMessagesQuery message. Also converts values to other types if specified. + * @function toObject + * @memberof getMessages.GetMessagesQuery + * @static + * @param {getMessages.GetMessagesQuery} message GetMessagesQuery + * @param {$protobuf.IConversionOptions} [options] Conversion options + * @returns {Object.} Plain object + */ + GetMessagesQuery.toObject = function (message, options, _depth) { + if (!options) + options = {}; + if (_depth === $undefined) + _depth = 0; + if (_depth > $util.recursionLimit) + throw $Error("max depth exceeded"); + let object = {}; + if (options.arrays || options.defaults) { + object.validatorsSignedPrevote = []; + object.validatorsSignedPrecommit = []; + } + if (options.defaults) { + object.blockNumber = 0; + object.round = 0; + } + if (message.blockNumber != null && $Object.hasOwnProperty.call(message, "blockNumber")) + object.blockNumber = message.blockNumber; + if (message.round != null && $Object.hasOwnProperty.call(message, "round")) + object.round = message.round; + if (message.validatorsSignedPrevote && message.validatorsSignedPrevote.length) { + object.validatorsSignedPrevote = $Array(message.validatorsSignedPrevote.length); + for (let j = 0; j < message.validatorsSignedPrevote.length; ++j) + object.validatorsSignedPrevote[j] = message.validatorsSignedPrevote[j]; + } + if (message.validatorsSignedPrecommit && message.validatorsSignedPrecommit.length) { + object.validatorsSignedPrecommit = $Array(message.validatorsSignedPrecommit.length); + for (let j = 0; j < message.validatorsSignedPrecommit.length; ++j) + object.validatorsSignedPrecommit[j] = message.validatorsSignedPrecommit[j]; + } + return object; + }; + + /** + * Converts this GetMessagesQuery to JSON. + * @function toJSON + * @memberof getMessages.GetMessagesQuery + * @instance + * @returns {Object.} JSON object + */ + GetMessagesQuery.prototype.toJSON = function() { + return GetMessagesQuery.toObject(this, $protobuf.util.toJSONOptions); + }; + + /** + * Gets the type url for GetMessagesQuery + * @function getTypeUrl + * @memberof getMessages.GetMessagesQuery + * @static + * @param {string} [prefix] Custom type url prefix, defaults to `"type.googleapis.com"` + * @returns {string} The type url + */ + GetMessagesQuery.getTypeUrl = function(prefix) { + if (prefix === $undefined) + prefix = "type.googleapis.com"; + return prefix + "/getMessages.GetMessagesQuery"; + }; + + return GetMessagesQuery; + })(); + getMessages.GetMessagesRequest = (function() { /** * Properties of a GetMessagesRequest. * @typedef {Object} getMessages.GetMessagesRequest.$Properties * @property {shared.Headers.$Properties|null} [headers] GetMessagesRequest headers + * @property {getMessages.GetMessagesQuery.$Properties|null} [query] GetMessagesRequest query * @property {Array.} [$unknowns] Unknown fields preserved while decoding when enabled */ @@ -2363,6 +2760,14 @@ export const getMessages = $root.getMessages = (() => { */ GetMessagesRequest.prototype.headers = null; + /** + * GetMessagesRequest query. + * @member {getMessages.GetMessagesQuery.$Properties|null|undefined} query + * @memberof getMessages.GetMessagesRequest + * @instance + */ + GetMessagesRequest.prototype.query = null; + /** * Creates a new GetMessagesRequest instance using the specified properties. * @function create @@ -2397,6 +2802,8 @@ export const getMessages = $root.getMessages = (() => { throw $Error("max depth exceeded"); if (message.headers != null && $Object.hasOwnProperty.call(message, "headers")) $root.shared.Headers.encode(message.headers, writer.uint32(/* id 1, wireType 2 =*/10).fork(), _depth + 1).ldelim(); + if (message.query != null && $Object.hasOwnProperty.call(message, "query")) + $root.getMessages.GetMessagesQuery.encode(message.query, writer.uint32(/* id 2, wireType 2 =*/18).fork(), _depth + 1).ldelim(); if (message.$unknowns != null && $Object.hasOwnProperty.call(message, "$unknowns")) for (let i = 0; i < message.$unknowns.length; ++i) writer.raw(message.$unknowns[i]); @@ -2450,6 +2857,12 @@ export const getMessages = $root.getMessages = (() => { message.headers = $root.shared.Headers.decode(reader, reader.uint32(), $undefined, _depth + 1, message.headers); continue; } + case 2: { + if (wireType !== 2) + break; + message.query = $root.getMessages.GetMessagesQuery.decode(reader, reader.uint32(), $undefined, _depth + 1, message.query); + continue; + } } reader.skipType(wireType, _depth, tag); if (!reader.discardUnknown) { @@ -2498,6 +2911,11 @@ export const getMessages = $root.getMessages = (() => { if (error) return "headers." + error; } + if (message.query != null && $Object.hasOwnProperty.call(message, "query")) { + let error = $root.getMessages.GetMessagesQuery.verify(message.query, _depth + 1); + if (error) + return "query." + error; + } return null; }; @@ -2524,6 +2942,11 @@ export const getMessages = $root.getMessages = (() => { throw $TypeError(".getMessages.GetMessagesRequest.headers: object expected"); message.headers = $root.shared.Headers.fromObject(object.headers, _depth + 1); } + if (object.query != null) { + if (!$util.isObject(object.query)) + throw $TypeError(".getMessages.GetMessagesRequest.query: object expected"); + message.query = $root.getMessages.GetMessagesQuery.fromObject(object.query, _depth + 1); + } return message; }; @@ -2544,10 +2967,14 @@ export const getMessages = $root.getMessages = (() => { if (_depth > $util.recursionLimit) throw $Error("max depth exceeded"); let object = {}; - if (options.defaults) + if (options.defaults) { object.headers = null; + object.query = null; + } if (message.headers != null && $Object.hasOwnProperty.call(message, "headers")) object.headers = $root.shared.Headers.toObject(message.headers, options, _depth + 1); + if (message.query != null && $Object.hasOwnProperty.call(message, "query")) + object.query = $root.getMessages.GetMessagesQuery.toObject(message.query, options, _depth + 1); return object; }; diff --git a/packages/p2p/source/socket-server/controllers/get-messages.test.ts b/packages/p2p/source/socket-server/controllers/get-messages.test.ts new file mode 100644 index 000000000..cb91ac4cc --- /dev/null +++ b/packages/p2p/source/socket-server/controllers/get-messages.test.ts @@ -0,0 +1,125 @@ +import { Identifiers } from "@mainsail/constants"; +import { Application } from "@mainsail/kernel"; + +import { describe } from "@mainsail/test-runner"; +import { GetMessagesController } from "./get-messages"; + +describe<{ + app: Application; + controller: GetMessagesController; + roundStates: Map; +}>("GetMessagesController", ({ it, assert, beforeEach }) => { + const consensus = { getBlockNumber: () => 2, getRound: () => 5 }; + + // A round state holding serialized prevotes/precommits for validators 0 and 1. + const makeRoundState = (round: number, indexes: number[]) => ({ + getPrecommit: (index: number) => + indexes.includes(index) ? { serialized: Buffer.from([20 + index]) } : undefined, + getPrecommits: () => indexes.map((index) => ({ serialized: Buffer.from([20 + index]) })), + getPrevote: (index: number) => + indexes.includes(index) ? { serialized: Buffer.from([10 + index]) } : undefined, + getPrevotes: () => indexes.map((index) => ({ serialized: Buffer.from([10 + index]) })), + hasMinorityPrevotesOrPrecommits: () => true, + round, + }); + + beforeEach((context) => { + context.roundStates = new Map(); + + context.app = new Application(); + context.app.bind(Identifiers.Consensus.Service).toConstantValue(consensus); + context.app.bind(Identifiers.Consensus.RoundStateRepository).toConstantValue({ + getRoundState: (blockNumber: number, round: number) => + context.roundStates.get(`${blockNumber}-${round}`) ?? makeRoundState(round, []), + }); + + context.controller = context.app.resolve(GetMessagesController); + }); + + it("should serve the queried round even while consensus is on a later one", async (context) => { + // Consensus sits on round 5; the requester asks for round 3, which is still retained. + context.roundStates.set("2-3", makeRoundState(3, [0, 1])); + + const response = await context.controller.handle( + { + payload: { + headers: { blockNumber: 2, round: 5, validatorsSignedPrecommit: [], validatorsSignedPrevote: [] }, + query: { + blockNumber: 2, + round: 3, + validatorsSignedPrecommit: [false, false], + validatorsSignedPrevote: [false, false], + }, + }, + } as any, + {} as any, + ); + + assert.equal(response.prevotes, [Buffer.from([10]), Buffer.from([11])]); + assert.equal(response.precommits, [Buffer.from([20]), Buffer.from([21])]); + }); + + it("should return only what the query is missing", async (context) => { + context.roundStates.set("2-3", makeRoundState(3, [0, 1])); + + const response = await context.controller.handle( + { + payload: { + headers: { blockNumber: 2, round: 5, validatorsSignedPrecommit: [], validatorsSignedPrevote: [] }, + query: { + blockNumber: 2, + round: 3, + validatorsSignedPrecommit: [true, false], + validatorsSignedPrevote: [false, true], + }, + }, + } as any, + {} as any, + ); + + assert.equal(response.prevotes, [Buffer.from([10])]); + assert.equal(response.precommits, [Buffer.from([21])]); + }); + + it("should answer empty for a block it is not deciding", async (context) => { + context.roundStates.set("3-0", makeRoundState(0, [0])); + + const response = await context.controller.handle( + { + payload: { + headers: { blockNumber: 2, round: 5, validatorsSignedPrecommit: [], validatorsSignedPrevote: [] }, + query: { + blockNumber: 3, + round: 0, + validatorsSignedPrecommit: [false, false], + validatorsSignedPrevote: [false, false], + }, + }, + } as any, + {} as any, + ); + + assert.equal(response.prevotes, []); + assert.equal(response.precommits, []); + }); + + it("should answer empty for a round it has not reached", async (context) => { + const response = await context.controller.handle( + { + payload: { + headers: { blockNumber: 2, round: 5, validatorsSignedPrecommit: [], validatorsSignedPrevote: [] }, + query: { + blockNumber: 2, + round: 7, + validatorsSignedPrecommit: [false, false], + validatorsSignedPrevote: [false, false], + }, + }, + } as any, + {} as any, + ); + + assert.equal(response.prevotes, []); + assert.equal(response.precommits, []); + }); +}); diff --git a/packages/p2p/source/socket-server/controllers/get-messages.ts b/packages/p2p/source/socket-server/controllers/get-messages.ts index 31ff912c1..23b1b5528 100644 --- a/packages/p2p/source/socket-server/controllers/get-messages.ts +++ b/packages/p2p/source/socket-server/controllers/get-messages.ts @@ -13,38 +13,25 @@ export class GetMessagesController implements Contracts.P2P.Controller { request: Contracts.P2P.GetMessagesRequest, h: Hapi.ResponseToolkit, ): Promise { - const { blockNumber, round, validatorsSignedPrecommit, validatorsSignedPrevote } = request.payload.headers; - const consensus = this.app.get(Identifiers.Consensus.Service); const roundStateRepo = this.app.get( Identifiers.Consensus.RoundStateRepository, ); - if (blockNumber !== consensus.getBlockNumber() || round > consensus.getRound()) { + const { query } = request.payload; + if (query.blockNumber !== consensus.getBlockNumber() || query.round > consensus.getRound()) { return { precommits: [], prevotes: [], }; } - // Use the highest round with minority prevotes - let roundState = roundStateRepo.getRoundState(blockNumber, consensus.getRound()); - if (roundState.round >= 1 && roundState.round > round && !roundState.hasMinorityPrevotesOrPrecommits()) { - roundState = roundStateRepo.getRoundState(blockNumber, consensus.getRound() - 1); - } + const roundState = roundStateRepo.getRoundState(query.blockNumber, query.round); - if (round === roundState.round) { - // Return only missing messages - return { - precommits: this.getPrecommits(validatorsSignedPrecommit, roundState), - prevotes: this.getPrevotes(validatorsSignedPrevote, roundState), - }; - } else { - return { - precommits: roundState.getPrecommits().map((precommit) => precommit.serialized), - prevotes: roundState.getPrevotes().map((prevote) => prevote.serialized), - }; - } + return { + precommits: this.getPrecommits(query.validatorsSignedPrecommit, roundState), + prevotes: this.getPrevotes(query.validatorsSignedPrevote, roundState), + }; } private getPrevotes( diff --git a/packages/p2p/source/socket-server/routes/get-messages.test.ts b/packages/p2p/source/socket-server/routes/get-messages.test.ts new file mode 100644 index 000000000..c46172694 --- /dev/null +++ b/packages/p2p/source/socket-server/routes/get-messages.test.ts @@ -0,0 +1,29 @@ +import { Identifiers } from "@mainsail/constants"; +import { Application } from "@mainsail/kernel"; +import { describe } from "@mainsail/test-runner"; + +import { GetMessagesRoute } from "./get-messages"; + +describe<{ + app: Application; + route: GetMessagesRoute; +}>("GetMessagesRoute", ({ it, assert, beforeEach }) => { + const configuration = { getMaxRoundValidators: () => 200 }; + + beforeEach((context) => { + context.app = new Application(); + + context.app.bind(Identifiers.ServiceProvider.Configuration).toConstantValue({}).whenTagged("plugin", "p2p"); + context.app.bind(Identifiers.Cryptography.Configuration).toConstantValue(configuration); + + context.route = context.app.resolve(GetMessagesRoute); + }); + + it("should scale the payload limit with the maximum validator count", ({ route }) => { + const { maxBytes } = route.getRoutesConfigByPath()["/getMessages"]; + + // The request carries four validator bitmaps (two in the headers, two in the query), + // each one byte per validator on the wire. + assert.equal(maxBytes, 1024 + 4 * 200); + }); +}); diff --git a/packages/p2p/source/socket-server/routes/get-messages.ts b/packages/p2p/source/socket-server/routes/get-messages.ts index c6c0228a5..e629761fc 100644 --- a/packages/p2p/source/socket-server/routes/get-messages.ts +++ b/packages/p2p/source/socket-server/routes/get-messages.ts @@ -15,7 +15,7 @@ export class GetMessagesRoute extends Route { "/getMessages": { codec: Codecs.getMessages, id: Routes.GetMessages, - maxBytes: 1024, + maxBytes: 1024 + 4 * this.cryptoConfiguration.getMaxRoundValidators(), validation: Schemas.getMessages(this.cryptoConfiguration), }, }; diff --git a/packages/p2p/source/socket-server/schemas/get-messages.test.ts b/packages/p2p/source/socket-server/schemas/get-messages.test.ts new file mode 100644 index 000000000..6b0ef3d14 --- /dev/null +++ b/packages/p2p/source/socket-server/schemas/get-messages.test.ts @@ -0,0 +1,44 @@ +import { describe } from "@mainsail/test-runner"; +import { getMessages } from "./get-messages"; + +describe("getMessages schema", ({ it, assert }) => { + const makeSchema = () => getMessages({ getMaxRoundValidators: () => 2 } as any); + + const headers = { + blockNumber: 2, + // eslint-disable-next-line unicorn/no-null + proposedBlockHash: null, + round: 0, + step: 1, + validatorsSignedPrecommit: [false, false], + validatorsSignedPrevote: [false, false], + version: "0.0.1", + }; + const query = { + blockNumber: 2, + round: 0, + validatorsSignedPrecommit: [false, false], + validatorsSignedPrevote: [false, false], + }; + + it("should accept a request with a query", () => { + assert.undefined(makeSchema().validate({ headers, query }).error); + }); + + it("should reject a request without a query", () => { + assert.defined(makeSchema().validate({ headers }).error); + }); + + it("should reject an incomplete query", () => { + assert.defined(makeSchema().validate({ headers, query: { blockNumber: 2, round: 0 } }).error); + }); + + it("should reject a query with oversized bitmaps", () => { + assert.defined( + makeSchema().validate({ + headers, + query: { ...query, validatorsSignedPrevote: [false, false, false] }, + }).error, + ); + }); +}); diff --git a/packages/p2p/source/socket-server/schemas/get-messages.ts b/packages/p2p/source/socket-server/schemas/get-messages.ts index 1007cd9c5..a97b5980a 100644 --- a/packages/p2p/source/socket-server/schemas/get-messages.ts +++ b/packages/p2p/source/socket-server/schemas/get-messages.ts @@ -4,7 +4,16 @@ import Joi from "joi"; import { makeHeaders } from "./shared.js"; -export const getMessages = (configuration: Contracts.Crypto.Configuration): Joi.ObjectSchema => - Joi.object({ +export const getMessages = (configuration: Contracts.Crypto.Configuration): Joi.ObjectSchema => { + const roundValidators = configuration.getMaxRoundValidators(); + + return Joi.object({ headers: makeHeaders(configuration), + query: Joi.object({ + blockNumber: Joi.number().integer().min(1).required(), + round: Joi.number().integer().min(0).required(), + validatorsSignedPrecommit: Joi.array().items(Joi.boolean()).max(roundValidators).required(), + validatorsSignedPrevote: Joi.array().items(Joi.boolean()).max(roundValidators).required(), + }).required(), }); +};