From 41642b8bcb0539cad509ca56aa83c37446774ae3 Mon Sep 17 00:00:00 2001 From: Chase Wright Date: Wed, 16 Sep 2026 13:18:33 -0500 Subject: [PATCH] feat(discv5)!: prefer IPv4 for dual-stack ENRs and fail requests on send error Dual-stack ENRs are contacted over IPv4 first (configurable via dualStackPreferredFamily). UDP send errors fail the request at once with SendFailed. The address a node last sent an authenticated message from is reused for later requests. The IPv6 vote-harvest ping is an isolated probe. verifyEnr accepts a dual-stack ENR observed over either family. --- packages/discv5/README.md | 4 + packages/discv5/src/config/index.ts | 1 + packages/discv5/src/metrics.ts | 7 + packages/discv5/src/service/service.ts | 62 ++++++-- packages/discv5/src/service/types.ts | 4 + packages/discv5/src/session/nodeInfo.ts | 22 ++- packages/discv5/src/session/service.ts | 71 ++++++--- packages/discv5/src/session/types.ts | 18 +++ packages/discv5/src/transport/udp.ts | 22 +-- packages/discv5/src/util/ip.ts | 38 +++-- packages/discv5/test/e2e/dualStack.test.ts | 82 +++++++++++ .../discv5/test/unit/session/nodeInfo.test.ts | 49 +++++++ .../discv5/test/unit/session/service.test.ts | 137 +++++++++++++++++- .../discv5/test/unit/transport/udp.test.ts | 43 ++++++ packages/discv5/test/unit/util/ip.test.ts | 59 +++++--- 15 files changed, 538 insertions(+), 81 deletions(-) create mode 100644 packages/discv5/test/e2e/dualStack.test.ts create mode 100644 packages/discv5/test/unit/session/nodeInfo.test.ts diff --git a/packages/discv5/README.md b/packages/discv5/README.md index 9ed0e85..59ce033 100644 --- a/packages/discv5/README.md +++ b/packages/discv5/README.md @@ -36,6 +36,10 @@ const libp2p = createLibp2p({ }); ``` +## Dual-stack + +With both `ip4` and `ip6` bind addresses, a peer that advertises both families is contacted over IPv4 first. Pass `config: {dualStackPreferredFamily: 6}` to try IPv6 first. Once a session exists with a peer, later requests reuse its address. A request whose packet cannot be sent (for example, no route to the peer's IPv6 address) fails at once with `SendFailed` instead of waiting for the request timeout. + ## License Apache-2.0 diff --git a/packages/discv5/src/config/index.ts b/packages/discv5/src/config/index.ts index 8ac1519..c377fb5 100644 --- a/packages/discv5/src/config/index.ts +++ b/packages/discv5/src/config/index.ts @@ -21,6 +21,7 @@ export type IDiscv5Config = ISessionConfig & export const defaultConfig: IDiscv5Config = { addrVotesToUpdateEnr: 10, allowUnverifiedSessions: false, + dualStackPreferredFamily: 4, enrUpdate: true, lookupNumResults: 16, lookupParallelism: 3, diff --git a/packages/discv5/src/metrics.ts b/packages/discv5/src/metrics.ts index 7261053..bda4bd5 100644 --- a/packages/discv5/src/metrics.ts +++ b/packages/discv5/src/metrics.ts @@ -24,6 +24,7 @@ export interface IDiscv5Metrics { sentMessageCount: Gauge<{type: string}>; rcvdMessageCount: Gauge<{type: string}>; + requestFailedCount: Gauge<{reason: string}>; rateLimitHitIP: Gauge; rateLimitHitTotal: Gauge; @@ -72,6 +73,12 @@ export function createDiscv5Metrics(register: MetricsRegister) { labelNames: ["type"], name: "discv5_rcvd_message_count", }), + /** Total number of failed requests by failure reason */ + requestFailedCount: register.gauge<{reason: string}>({ + help: "Count of the discv5 requests that failed by reason", + labelNames: ["reason"], + name: "discv5_request_failed_count", + }), /** Total number messages sent by message type */ sentMessageCount: register.gauge<{type: string}>({ help: "Count of the discv5 messages sent by message type", diff --git a/packages/discv5/src/service/service.ts b/packages/discv5/src/service/service.ts index 1cd899c..429ddb5 100644 --- a/packages/discv5/src/service/service.ts +++ b/packages/discv5/src/service/service.ts @@ -9,8 +9,9 @@ import { createPeerIdFromPublicKey, } from "@chainsafe/enr"; import type {PeerId, PrivateKey} from "@libp2p/interface"; -import type {Multiaddr} from "@multiformats/multiaddr"; +import {type Multiaddr, isMultiaddr} from "@multiformats/multiaddr"; import {randomBytes} from "@noble/hashes/utils.js"; +import {LRUCache} from "lru-cache"; import debug from "weald"; import {type IDiscv5Config, defaultConfig} from "../config/index.js"; import { @@ -48,6 +49,7 @@ import { type INodeAddress, type NodeContact, createNodeContact, + createNodeContactAt, getNodeAddress, getNodeId, } from "../session/nodeInfo.js"; @@ -168,6 +170,15 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { */ private connectedPeers: Map; + /** + * Socket address each node last sent us an authenticated message from. Later requests reuse it + * instead of re-selecting a family from the ENR, so a peer that reached us over one family stays + * on it. Written only on inbound traffic: an outgoing handshake proves nothing about the address. + * + * BOUNDED: same bounds as the session cache + */ + private confirmedAddrs: LRUCache; + /** * Id for the next lookup that we start */ @@ -197,6 +208,7 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { this.activeRequests = new Map(); this.activeNodesResponses = new Map(); this.connectedPeers = new Map(); + this.confirmedAddrs = new LRUCache({max: config.sessionCacheCapacity, ttl: config.sessionTimeout}); this.nextLookupId = 1; this.ipMode = this.sessionService.transport.ipMode; this.addrVotes = { @@ -282,6 +294,7 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { clearInterval(intervalId); } this.connectedPeers.clear(); + this.confirmedAddrs.clear(); this.sessionService.off("established", this.onEstablished); this.sessionService.off("request", this.handleRpcRequest); this.sessionService.off("response", this.handleRpcResponse); @@ -404,7 +417,7 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { * Send FINDNODE message to remote and returns response */ async sendFindNode(remote: ENR | Multiaddr, distances: number[]): Promise { - const contact = createNodeContact(remote, this.ipMode); + const contact = this.contactFor(remote); const request = createFindNodeMessage(distances); return await new Promise((resolve, reject) => { @@ -423,7 +436,7 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { * Send TALKREQ message to dstId and returns response */ async sendTalkReq(remote: ENR | Multiaddr, payload: Uint8Array, protocol: string | Uint8Array): Promise { - const contact = createNodeContact(remote, this.ipMode); + const contact = this.contactFor(remote); const request = createTalkRequestMessage(payload, protocol); return await new Promise((resolve, reject) => { @@ -456,8 +469,8 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { /** * Sends a PING request to a node and returns response */ - async sendPing(nodeAddr: ENR | Multiaddr): Promise { - const contact = createNodeContact(nodeAddr, this.ipMode); + async sendPing(nodeAddr: ENR | Multiaddr, opts?: {preferFamily?: 4 | 6; probe?: boolean}): Promise { + const contact = this.contactFor(nodeAddr, opts?.preferFamily); const request = createPingMessage(this.enr.seq); return await new Promise((resolve, reject) => { @@ -467,11 +480,30 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { resolve: resolve as (val: PongResponse) => void, }, contact, + probe: opts?.probe, request, }); }); } + /** + * Builds the contact for a remote. An explicit family wins, then the address the node last sent + * us an authenticated message from, then the configured dual-stack preference. + */ + private contactFor(remote: ENR | Multiaddr, preferFamily?: 4 | 6): NodeContact { + if (isMultiaddr(remote)) { + return createNodeContact(remote, this.ipMode); + } + if (preferFamily !== undefined) { + return createNodeContact(remote, this.ipMode, preferFamily); + } + const confirmedAddr = this.confirmedAddrs.get(remote.nodeId); + if (confirmedAddr) { + return createNodeContactAt(remote, confirmedAddr); + } + return createNodeContact(remote, this.ipMode, this.config.dualStackPreferredFamily ?? 4); + } + /** * Ping all peers connected in the routing table */ @@ -568,7 +600,7 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { } this.sendRpcRequest({ - contact: createNodeContact(enr, this.ipMode), + contact: this.contactFor(enr), lookupId, request, }); @@ -589,7 +621,7 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { const nodeAddr = getNodeAddress(activeRequest.contact); log("Sending %s to node: %o", MessageType[activeRequest.request.type], nodeAddr); try { - this.sessionService.sendRequest(activeRequest.contact, activeRequest.request); + this.sessionService.sendRequest(activeRequest.contact, activeRequest.request, {probe: activeRequest.probe}); this.metrics?.sentMessageCount.inc({type: MessageType[activeRequest.request.type]}); } catch (e) { this.activeRequests.delete(bytesToBigint(activeRequest.request.id)); @@ -692,7 +724,7 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { ) { const enr = newStatus.enr; setTimeout(() => { - this.sendPing(enr).catch((e) => + this.sendPing(enr, {preferFamily: 6, probe: true}).catch((e) => log("Error pinging peer for IP vote %o: %s", enr, (e as Error).message) ); }, 0); @@ -731,6 +763,7 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { } clearInterval(this.connectedPeers.get(nodeId) as NodeJS.Timeout); this.connectedPeers.delete(nodeId); + this.confirmedAddrs.delete(nodeId); break; } } @@ -840,6 +873,7 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { private handleRpcRequest = (nodeAddr: INodeAddress, request: RequestMessage): void => { const requestType = MessageType[request.type]; this.metrics?.rcvdMessageCount.inc({type: requestType}); + this.confirmedAddrs.set(nodeAddr.nodeId, nodeAddr.socketAddr); try { switch (request.type) { @@ -866,7 +900,7 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { // check if we need to update the known ENR const entry = this.kbuckets.getWithPending(nodeAddr.nodeId); if (entry && entry.value.seq < message.enrSeq) { - this.requestEnr(createNodeContact(entry.value, this.ipMode)); + this.requestEnr(this.contactFor(entry.value)); } const ipUDP = multiaddrToSocketAddress(nodeAddr.socketAddr); @@ -980,6 +1014,7 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { ); return; } + this.confirmedAddrs.set(nodeAddr.nodeId, nodeAddr.socketAddr); // Check that the response type matches the request if (!requestMatchesResponse(activeRequest.request, response)) { @@ -1104,11 +1139,12 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { */ private rpcFailure = (rpcId: RequestId, error: RequestErrorType): void => { log("RPC error, removing request. Reason: %s, id %s", RequestErrorType[error], rpcId); + this.metrics?.requestFailedCount.inc({reason: RequestErrorType[error]}); const req = this.activeRequests.get(bytesToBigint(rpcId)); if (!req) { return; } - const {request, contact, lookupId, callbackPromise} = req; + const {request, contact, lookupId, callbackPromise, probe} = req; this.activeRequests.delete(bytesToBigint(request.id)); const nodeId = getNodeId(contact); @@ -1143,8 +1179,10 @@ export class Discv5 extends (EventEmitter as {new (): Discv5EventEmitter}) { } } - // report the node as being disconnected - this.connectionUpdated(nodeId, {type: ConnectionStatusType.Disconnected}); + // report the node as being disconnected, unless the request was only a probe + if (!probe) { + this.connectionUpdated(nodeId, {type: ConnectionStatusType.Disconnected}); + } // If this is initiated by the user, return the error on the callback. callbackPromise?.reject(new CodeError("RPC failure", RequestErrorType[error])); diff --git a/packages/discv5/src/service/types.ts b/packages/discv5/src/service/types.ts index 43df114..d01aae4 100644 --- a/packages/discv5/src/service/types.ts +++ b/packages/discv5/src/service/types.ts @@ -77,6 +77,10 @@ export interface IActiveRequest; + private pendingRequests: Map; /** * Currently in-progress handshakes with peers. @@ -144,8 +145,6 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter */ private sessions: LRUCache; - private ipMode: IPMode; - constructor(config: ISessionConfig, enr: SignableENR, keypair: IKeypair, transport: ITransportService) { super(); @@ -165,7 +164,6 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter this.pendingRequests = new Map(); this.activeChallenges = new LRUCache({max: config.sessionCacheCapacity, ttl: config.requestTimeout * 2}); this.sessions = new LRUCache({max: config.sessionCacheCapacity, ttl: config.sessionTimeout}); - this.ipMode = this.transport.ipMode; } /** @@ -198,7 +196,7 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter /** * Sends an RequestMessage to a node. */ - sendRequest(contact: NodeContact, request: RequestMessage): void { + sendRequest(contact: NodeContact, request: RequestMessage, opts?: RequestOpts): void { const nodeAddr = getNodeAddress(contact); const nodeAddrStr = nodeAddressToString(nodeAddr); @@ -215,7 +213,7 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter pendingRequests = []; this.pendingRequests.set(nodeAddrStr, pendingRequests); } - pendingRequests.push([contact, request]); + pendingRequests.push([contact, request, opts]); return; } @@ -239,6 +237,7 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter handshakeSent: false, initiatingSession, packet, + probe: opts?.probe, request, retries: 1, }; @@ -246,10 +245,10 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter // let the filter know we are expecting a response this.addExpectedResponse(nodeAddr.socketAddr); - this.send(nodeAddr, packet); - this.activeRequestsNonceMapping.set(bytesToHex(packet.header.nonce), nodeAddr); this.activeRequests.set(nodeAddrStr, call); + + this.sendRequestPacket(call); } /** @@ -454,7 +453,7 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter this.insertActiveRequest(requestCall); // Send the actual packet - this.send(nodeAddr, authPacket); + this.sendRequestPacket(requestCall); // Notify the application that the session has been established this.emit("established", nodeAddr, requestCall.contact.enr, connectionDirection, true); @@ -473,7 +472,7 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter this.insertActiveRequest(requestCall); // Send the actual request - this.send(nodeAddr, authPacket); + this.sendRequestPacket(requestCall); // send FINDNODE 0 const request = createFindNodeMessage([0]); @@ -491,11 +490,18 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter * Returns true if they match */ private verifyEnr(enr: ENR, nodeAddr: INodeAddress): boolean { - const enrMultiaddrIP4 = getSocketAddressMultiaddrOnENR(enr, {...this.ipMode, ip6: false} as IPMode); - const enrMultiaddrIP6 = getSocketAddressMultiaddrOnENR(enr, {...this.ipMode, ip4: false} as IPMode); + if (enr.nodeId !== nodeAddr.nodeId) { + return false; + } + const enrAddr4 = getSocketAddressOnENRByFamily(enr, 4); + const enrAddr6 = getSocketAddressOnENRByFamily(enr, 6); + if (!enrAddr4 && !enrAddr6) { + return true; + } + const observed = multiaddrToSocketAddress(nodeAddr.socketAddr); return ( - enr.nodeId === nodeAddr.nodeId && - (enrMultiaddrIP4?.equals(nodeAddr.socketAddr) ?? enrMultiaddrIP6?.equals(nodeAddr.socketAddr) ?? true) + (enrAddr4 !== undefined && isEqualSocketAddress(enrAddr4, observed)) || + (enrAddr6 !== undefined && isEqualSocketAddress(enrAddr6, observed)) ); } @@ -783,9 +789,9 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter // increment the request retry count and restart the timeout log("Resending message to: %o", nodeAddr); - this.send(nodeAddr, requestCall.packet); requestCall.retries++; this.activeRequests.set(nodeAddrStr, requestCall); + this.sendRequestPacket(requestCall); } } @@ -803,7 +809,7 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter } log("Sending next awaiting message. Node: %o", request[0]); - this.sendRequest(request[0], request[1]); + this.sendRequest(request[0], request[1], request[2]); } } } @@ -820,6 +826,11 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter this.emit("requestFailed", requestCall.request.id, error); const nodeAddr = getNodeAddress(requestCall.contact); + if (requestCall.probe) { + // only handleResponse drains the queue otherwise, so a failed probe must hand over itself + this.sendNextRequest(nodeAddr); + return; + } this.failSession(nodeAddr, error, removeSession); } @@ -847,4 +858,28 @@ export class SessionService extends (EventEmitter as {new (): StrictEventEmitter .send(nodeAddr.socketAddr, nodeAddr.nodeId, packet) .catch((e) => log("Error sending packet to node %o: %s", nodeAddr, (e as Error).message)); } + + /** + * Sends the packet of an active request. A send error fails the request at once instead of + * leaving it to the request timeout. + */ + private sendRequestPacket(requestCall: IRequestCall): void { + const nodeAddr = getNodeAddress(requestCall.contact); + this.transport + .send(nodeAddr.socketAddr, nodeAddr.nodeId, requestCall.packet) + .catch((e) => this.handleSendFailure(requestCall, e as Error)); + } + + private handleSendFailure(requestCall: IRequestCall, e: Error): void { + const nodeAddr = getNodeAddress(requestCall.contact); + const nodeAddrStr = nodeAddressToString(nodeAddr); + // the callback is asynchronous: the request may have been answered or replaced already + if (this.activeRequests.get(nodeAddrStr) !== requestCall) { + return; + } + log("Send failed to node %o: %s", nodeAddr, e.message); + this.activeRequests.delete(nodeAddrStr); + this.removeExpectedResponse(nodeAddr.socketAddr); + this.failRequest(requestCall, RequestErrorType.SendFailed, false); + } } diff --git a/packages/discv5/src/session/types.ts b/packages/discv5/src/session/types.ts index 1529852..36758ac 100644 --- a/packages/discv5/src/session/types.ts +++ b/packages/discv5/src/session/types.ts @@ -35,6 +35,20 @@ export interface ISessionConfig { * UDP endpoint does not match observed socket address) */ allowUnverifiedSessions: boolean; + /** + * Address family to contact first when both sockets are bound and the remote ENR advertises both. + * Defaults to IPv4. Only affects dual-stack operation. + */ + dualStackPreferredFamily?: 4 | 6; +} + +/** Per-request options */ +export interface RequestOpts { + /** + * A probe is sent for a side effect only (e.g. to collect an address vote). Its failure must not + * fail other requests queued to the same node address or change the node's connection status. + */ + probe?: boolean; } export enum RequestErrorType { @@ -54,6 +68,8 @@ export enum RequestErrorType { EncryptionFailed = 6, /** The multiaddr provided is invalid */ InvalidMultiaddr = 7, + /** The transport could not send the request packet (e.g. no route to the destination) */ + SendFailed = 8, } export enum ResponseErrorType { @@ -128,6 +144,8 @@ export interface IRequestCall { * determine the connection direction of the session. */ initiatingSession: boolean; + /** See {@link RequestOpts.probe} */ + probe?: boolean; } export interface ISessionEvents { diff --git a/packages/discv5/src/transport/udp.ts b/packages/discv5/src/transport/udp.ts index 92d1f55..0c5fefa 100644 --- a/packages/discv5/src/transport/udp.ts +++ b/packages/discv5/src/transport/udp.ts @@ -103,17 +103,19 @@ export class UDPTransportService async send(to: Multiaddr, toId: string, packet: IPacket): Promise { const nodeAddr = multiaddrToObject(to); - if (nodeAddr.family === 4) { - if (!this.ip4) { - throw new Error("Cannot send to an IPv4 address without a bound IPv4 socket"); - } - this.ip4.socket?.send(encodePacket(toId, packet), nodeAddr.port, nodeAddr.host); - } else if (nodeAddr.family === 6) { - if (!this.ip6) { - throw new Error("Cannot send to an IPv6 address without a bound IPv6 socket"); - } - this.ip6.socket?.send(encodePacket(toId, packet), nodeAddr.port, nodeAddr.host); + const socketOpts = nodeAddr.family === 4 ? this.ip4 : this.ip6; + if (!socketOpts) { + throw new Error(`Cannot send to an IPv${nodeAddr.family} address without a bound IPv${nodeAddr.family} socket`); + } + const socket = socketOpts.socket; + if (!socket) { + throw new Error("Transport not started"); } + // dgram reports send errors (e.g. ENETUNREACH) only through the callback + const data = encodePacket(toId, packet); + await new Promise((resolve, reject) => { + socket.send(data, nodeAddr.port, nodeAddr.host, (err) => (err ? reject(err) : resolve())); + }); } addExpectedResponse(ipAddress: string): void { diff --git a/packages/discv5/src/util/ip.ts b/packages/discv5/src/util/ip.ts index 42b090f..d1ea095 100644 --- a/packages/discv5/src/util/ip.ts +++ b/packages/discv5/src/util/ip.ts @@ -67,15 +67,25 @@ export function isEqualSocketAddress(s1: SocketAddress, s2: SocketAddress): bool return s1.port === s2.port; } -export function getSocketAddressMultiaddrOnENR(enr: BaseENR, ipMode: IPMode): Multiaddr | undefined { - if (ipMode.ip6) { - const multiaddr = enr.getLocationMultiaddr("udp6"); - if (multiaddr) return multiaddr; - } - if (ipMode.ip4) { - const multiaddr = enr.getLocationMultiaddr("udp4"); +/** + * Families to try for a dual-stack ENR, in order. IPv4 first by default: it is reachable from nearly + * every host, while a bound IPv6 socket says nothing about IPv6 connectivity. + */ +function familyOrder(ipMode: IPMode, preferFamily: 4 | 6): (4 | 6)[] { + const order: (4 | 6)[] = preferFamily === 6 ? [6, 4] : [4, 6]; + return order.filter((family) => (family === 4 ? ipMode.ip4 : ipMode.ip6)); +} + +export function getSocketAddressMultiaddrOnENR( + enr: BaseENR, + ipMode: IPMode, + preferFamily: 4 | 6 = 4 +): Multiaddr | undefined { + for (const family of familyOrder(ipMode, preferFamily)) { + const multiaddr = enr.getLocationMultiaddr(family === 4 ? "udp4" : "udp6"); if (multiaddr) return multiaddr; } + return undefined; } export function getSocketAddressOnENRByFamily(enr: BaseENR, family: 4 | 6): SocketAddress | undefined { @@ -93,13 +103,13 @@ export function getSocketAddressOnENRByFamily(enr: BaseENR, family: 4 | 6): Sock return {ip, port}; } -export function getSocketAddressOnENR(enr: BaseENR, ipMode: IPMode): SocketAddress | undefined { - if (ipMode.ip6) { - const socketAddr = getSocketAddressOnENRByFamily(enr, 6); - if (socketAddr) return socketAddr; - } - if (ipMode.ip4) { - const socketAddr = getSocketAddressOnENRByFamily(enr, 4); +export function getSocketAddressOnENR( + enr: BaseENR, + ipMode: IPMode, + preferFamily: 4 | 6 = 4 +): SocketAddress | undefined { + for (const family of familyOrder(ipMode, preferFamily)) { + const socketAddr = getSocketAddressOnENRByFamily(enr, family); if (socketAddr) return socketAddr; } return undefined; diff --git a/packages/discv5/test/e2e/dualStack.test.ts b/packages/discv5/test/e2e/dualStack.test.ts new file mode 100644 index 0000000..999881b --- /dev/null +++ b/packages/discv5/test/e2e/dualStack.test.ts @@ -0,0 +1,82 @@ +import {SignableENR} from "@chainsafe/enr"; +import {generateKeyPair} from "@libp2p/crypto/keys"; +import {multiaddr} from "@multiformats/multiaddr"; +import {afterEach, describe, expect, it, vi} from "vitest"; +import {type IDiscv5Config, defaultConfig} from "../../src/config/index.js"; +import {Discv5} from "../../src/index.js"; +import {INodeContactType} from "../../src/session/nodeInfo.js"; +import type {BindAddrs} from "../../src/transport/index.js"; + +/** + * Loopback-only dual-stack scenarios. A peer that advertises an unreachable IPv6 address next to a + * reachable IPv4 one stands in for the common case of a bootnode reached from a host without IPv6. + * The IPv4-mapped address is refused by an IPv6-only socket on every platform; an unrouted global + * address like 2001:db8::1 is silently dropped by kernels that have an IPv6 default route. + */ +const unreachableIp6 = "::ffff:127.0.0.1"; +describe("dual-stack peers", () => { + const nodes: Discv5[] = []; + let nextPort = 10200; + + type NodeOpts = {ip6?: boolean; enrIp6?: string; config?: Partial}; + + async function startNode(opts: NodeOpts): Promise<{discv5: Discv5; enr: SignableENR; port: number}> { + const port = nextPort++; + const privateKey = await generateKeyPair("secp256k1"); + const enr = SignableENR.createFromPrivateKey(privateKey); + enr.ip = "127.0.0.1"; + enr.udp = port; + if (opts.enrIp6) { + enr.ip6 = opts.enrIp6; + enr.udp6 = port; + } + const bindAddrs: BindAddrs = opts.ip6 + ? {ip4: multiaddr(`/ip4/127.0.0.1/udp/${port}`), ip6: multiaddr(`/ip6/::1/udp/${port}`)} + : {ip4: multiaddr(`/ip4/127.0.0.1/udp/${port}`)}; + const discv5 = Discv5.create({bindAddrs, config: opts.config, enr, privateKey}); + nodes.push(discv5); + await discv5.start(); + return {discv5, enr, port}; + } + + afterEach(async () => { + for (const node of nodes.splice(0)) { + await node.stop(); + } + }); + + it("reaches a dual-stack peer over IPv4 when its IPv6 address is unreachable", async () => { + const a = await startNode({enrIp6: "::1", ip6: true}); + const b = await startNode({enrIp6: "2001:db8::1"}); + + const pong = await a.discv5.sendPing(b.enr.toENR()); + expect(pong.enrSeq).to.equal(b.enr.seq); + }); + + it("fails fast instead of timing out when the preferred family cannot be reached", async () => { + const a = await startNode({config: {dualStackPreferredFamily: 6}, enrIp6: "::1", ip6: true}); + const b = await startNode({enrIp6: unreachableIp6}); + + const start = Date.now(); + await expect(a.discv5.sendPing(b.enr.toENR())).rejects.toMatchObject({code: "SendFailed"}); + expect(Date.now() - start).to.be.lessThan(defaultConfig.requestTimeout); + }); + + it("keeps talking to a peer over the family it used to reach us", async () => { + const a = await startNode({enrIp6: "::1", ip6: true}); + const c = await startNode({config: {dualStackPreferredFamily: 6}, enrIp6: "::1", ip6: true}); + + await c.discv5.sendPing(a.enr.toENR()); + + const sendRequest = vi.spyOn(a.discv5.sessionService, "sendRequest"); + await a.discv5.sendPing(c.enr.toENR()); + const contacts = sendRequest.mock.calls + .map(([contact]) => contact) + .filter((contact) => contact.nodeAddress.nodeId === c.enr.nodeId); + expect(contacts.length).to.be.greaterThan(0); + for (const contact of contacts) { + expect(contact.type).to.equal(INodeContactType.Enr); + expect(contact.nodeAddress.socketAddr.toString()).to.equal(`/ip6/::1/udp/${c.port}`); + } + }); +}); diff --git a/packages/discv5/test/unit/session/nodeInfo.test.ts b/packages/discv5/test/unit/session/nodeInfo.test.ts new file mode 100644 index 0000000..06e299d --- /dev/null +++ b/packages/discv5/test/unit/session/nodeInfo.test.ts @@ -0,0 +1,49 @@ +import {SignableENR} from "@chainsafe/enr"; +import {multiaddr} from "@multiformats/multiaddr"; +import {describe, expect, it} from "vitest"; +import {generateKeypair} from "../../../src/index.js"; +import {INodeContactType, createNodeContact, createNodeContactAt} from "../../../src/session/nodeInfo.js"; + +describe("createNodeContact", () => { + const signableEnr = SignableENR.createV4(generateKeypair("secp256k1").privateKey); + signableEnr.ip = "127.0.0.1"; + signableEnr.udp = 9000; + signableEnr.ip6 = "::1"; + signableEnr.udp6 = 9001; + const enr = signableEnr.toENR(); + + it("contacts a dual-stack ENR over IPv4 by default", () => { + expect(createNodeContact(enr, {ip4: true, ip6: true}).nodeAddress.socketAddr.toString()).to.equal( + "/ip4/127.0.0.1/udp/9000" + ); + }); + + it("contacts a dual-stack ENR over IPv6 when requested", () => { + expect(createNodeContact(enr, {ip4: true, ip6: true}, 6).nodeAddress.socketAddr.toString()).to.equal( + "/ip6/::1/udp/9001" + ); + }); + + it("ignores the preference when only one family is bound", () => { + expect(createNodeContact(enr, {ip4: false, ip6: true}).nodeAddress.socketAddr.toString()).to.equal( + "/ip6/::1/udp/9001" + ); + expect(createNodeContact(enr, {ip4: true, ip6: false}, 6).nodeAddress.socketAddr.toString()).to.equal( + "/ip4/127.0.0.1/udp/9000" + ); + }); + + it("creates an ENR contact at a known socket address", () => { + const contact = createNodeContactAt(enr, multiaddr("/ip6/::1/udp/9001")); + expect(contact.type).to.equal(INodeContactType.Enr); + expect(contact.nodeAddress.nodeId).to.equal(enr.nodeId); + expect(contact.nodeAddress.socketAddr.toString()).to.equal("/ip6/::1/udp/9001"); + }); + + it("rejects a socket address that is not a bare ip/udp multiaddr", () => { + expect(() => + createNodeContactAt(enr, multiaddr(`/ip4/127.0.0.1/udp/9000/p2p/${enr.peerId.toString()}`)) + ).to.throw(); + expect(() => createNodeContactAt(enr, multiaddr("/ip4/127.0.0.1/tcp/9000"))).to.throw(); + }); +}); diff --git a/packages/discv5/test/unit/session/service.test.ts b/packages/discv5/test/unit/session/service.test.ts index 8092071..06ca4ad 100644 --- a/packages/discv5/test/unit/session/service.test.ts +++ b/packages/discv5/test/unit/session/service.test.ts @@ -1,14 +1,21 @@ +import {EventEmitter} from "node:events"; import {SignableENR} from "@chainsafe/enr"; import {type Multiaddr, multiaddr} from "@multiformats/multiaddr"; import {hexToBytes} from "ethereum-cryptography/utils.js"; import {afterEach, beforeEach, describe, expect, it} from "vitest"; import {defaultConfig} from "../../../src/config/index.js"; -import {createKeypair} from "../../../src/keypair/index.js"; -import {createFindNodeMessage} from "../../../src/message/index.js"; +import {createKeypair, generateKeypair} from "../../../src/keypair/index.js"; +import { + MessageType, + type RequestId, + type ResponseMessage, + createFindNodeMessage, + createPingMessage, +} from "../../../src/message/index.js"; import {type IPacket, PacketType, createWhoAreYouPacket} from "../../../src/packet/index.js"; -import {SessionService} from "../../../src/session/index.js"; -import {createNodeContact} from "../../../src/session/nodeInfo.js"; -import {UDPTransportService} from "../../../src/transport/index.js"; +import {RequestErrorType, SessionService} from "../../../src/session/index.js"; +import {type NodeContact, createNodeContact} from "../../../src/session/nodeInfo.js"; +import {type ITransportService, UDPTransportService} from "../../../src/transport/index.js"; describe("session service", () => { const kp0 = createKeypair({ @@ -94,3 +101,123 @@ describe("session service", () => { // TODO implement or delete this }); }); + +describe("session service send failures and probes", () => { + const kpA = generateKeypair("secp256k1"); + const kpB = generateKeypair("secp256k1"); + const enrA = SignableENR.createV4(kpA.privateKey); + const enrB = SignableENR.createV4(kpB.privateKey); + enrB.ip = "127.0.0.1"; + enrB.udp = 49040; + const contactB = (opts?: {preferFamily?: 4 | 6}): NodeContact => + createNodeContact(enrB.toENR(), {ip4: true, ip6: true}, opts?.preferFamily); + + /** Transport that records sends and resolves or rejects them in the given order */ + function stubTransport(outcomes: ("ok" | "fail")[]): {transport: ITransportService; sends: Multiaddr[]} { + const sends: Multiaddr[] = []; + const transport = Object.assign(new EventEmitter(), { + bindAddrs: [], + getContactableAddr: () => undefined, + ipMode: {ip4: true, ip6: true}, + send: async (to: Multiaddr): Promise => { + sends.push(to); + if ((outcomes.shift() ?? "ok") === "fail") { + throw Object.assign(new Error("send ENETUNREACH"), {code: "ENETUNREACH"}); + } + }, + start: () => Promise.resolve(), + stop: () => Promise.resolve(), + }) as unknown as ITransportService; + return {sends, transport}; + } + + async function withService( + outcomes: ("ok" | "fail")[], + fn: (service: SessionService, sends: Multiaddr[], failed: [RequestId, RequestErrorType][]) => Promise + ): Promise { + const {transport, sends} = stubTransport(outcomes); + const service = new SessionService(defaultConfig, enrA, kpA, transport); + const failed: [RequestId, RequestErrorType][] = []; + service.on("requestFailed", (id, err) => failed.push([id, err])); + await service.start(); + try { + await fn(service, sends, failed); + } finally { + await service.stop(); + } + } + + const tick = (): Promise => new Promise((resolve) => setTimeout(resolve, 0)); + + it("should fail the request at once with SendFailed when the transport cannot send", async () => { + await withService(["fail"], async (service, _sends, failed) => { + const request = createFindNodeMessage([0]); + const start = Date.now(); + service.sendRequest(contactB(), request); + await tick(); + expect(failed).to.deep.equal([[request.id, RequestErrorType.SendFailed]]); + expect(Date.now() - start).to.be.lessThan(defaultConfig.requestTimeout / 2); + }); + }); + + it("should fail requests queued behind a request whose send failed", async () => { + await withService(["fail"], async (service, sends, failed) => { + const request1 = createFindNodeMessage([0]); + const request2 = createFindNodeMessage([1]); + service.sendRequest(contactB(), request1); + service.sendRequest(contactB(), request2); + await tick(); + expect(failed.map(([id]) => id)).to.deep.equal([request1.id, request2.id]); + expect(sends.length).to.equal(1); + }); + }); + + it("should not fail queued requests when a probe fails, and should send the next one", async () => { + await withService(["fail", "ok"], async (service, sends, failed) => { + const probe = createPingMessage(1n); + const request = createFindNodeMessage([0]); + service.sendRequest(contactB(), probe, {probe: true}); + service.sendRequest(contactB(), request); + await tick(); + expect(failed).to.deep.equal([[probe.id, RequestErrorType.SendFailed]]); + expect(sends.length).to.equal(2); + }); + }); + + it("should keep the probe flag on a request dequeued after another completes", async () => { + await withService(["ok", "fail", "ok"], async (service, sends, failed) => { + const request1 = createFindNodeMessage([0]); + const probe = createPingMessage(1n); + const request3 = createFindNodeMessage([1]); + const contact = contactB(); + service.sendRequest(contact, request1); + service.sendRequest(contact, probe, {probe: true}); + service.sendRequest(contact, request3); + await tick(); + expect(sends.length).to.equal(1); + + service["handleResponse"](contact.nodeAddress, {id: request1.id, type: MessageType.PONG} as ResponseMessage); + await tick(); + expect(failed).to.deep.equal([[probe.id, RequestErrorType.SendFailed]]); + expect(sends.length).to.equal(3); + }); + }); + + it("should verify a dual-stack ENR observed over either family", () => { + const {transport} = stubTransport([]); + const service = new SessionService(defaultConfig, enrA, kpA, transport); + const signableEnr = SignableENR.createV4(kpB.privateKey); + signableEnr.ip = "127.0.0.1"; + signableEnr.udp = 9000; + signableEnr.ip6 = "::1"; + signableEnr.udp6 = 9001; + const enr = signableEnr.toENR(); + const verify = (addr: string): boolean => + service["verifyEnr"](enr, {nodeId: enr.nodeId, socketAddr: multiaddr(addr)}); + + expect(verify("/ip4/127.0.0.1/udp/9000")).to.be.true; + expect(verify("/ip6/::1/udp/9001")).to.be.true; + expect(verify("/ip4/127.0.0.1/udp/9001")).to.be.false; + expect(verify("/ip6/::1/udp/9002")).to.be.false; + }); +}); diff --git a/packages/discv5/test/unit/transport/udp.test.ts b/packages/discv5/test/unit/transport/udp.test.ts index 7ef4585..4d99989 100644 --- a/packages/discv5/test/unit/transport/udp.test.ts +++ b/packages/discv5/test/unit/transport/udp.test.ts @@ -278,3 +278,46 @@ describe("UDP4+6 transport", () => { }); }); }); + +describe("UDP transport send errors", () => { + const nodeId = bytesToHex(new Uint8Array(32).fill(3)); + const packet: IPacket = { + header: { + authdata: new Uint8Array(32).fill(2), + authdataSize: 32, + flag: PacketType.Message, + nonce: new Uint8Array(NONCE_SIZE), + protocolId: "discv5", + version: 1, + }, + maskingIv: new Uint8Array(MASKING_IV_SIZE), + message: new Uint8Array(44).fill(1), + }; + + it("should reject when the datagram cannot be delivered", async () => { + const t = new UDPTransportService({bindAddrs: {ip6: multiaddr("/ip6/::1/udp/49530")}, nodeId}); + await t.start(); + try { + // an IPv4-mapped destination is refused by an IPv6-only socket on every platform, while an + // unrouted global address is silently accepted by kernels that have an IPv6 default route + await expect(t.send(multiaddr("/ip6/::ffff:127.0.0.1/udp/9"), nodeId, packet)).rejects.toThrow(); + } finally { + await t.stop(); + } + }); + + it("should reject before start", async () => { + const t = new UDPTransportService({bindAddrs: {ip4: multiaddr("/ip4/127.0.0.1/udp/49531")}, nodeId}); + await expect(t.send(multiaddr("/ip4/127.0.0.1/udp/9"), nodeId, packet)).rejects.toThrow("Transport not started"); + }); + + it("should reject a family without a bound socket", async () => { + const t = new UDPTransportService({bindAddrs: {ip4: multiaddr("/ip4/127.0.0.1/udp/49532")}, nodeId}); + await t.start(); + try { + await expect(t.send(multiaddr("/ip6/::1/udp/9"), nodeId, packet)).rejects.toThrow("without a bound IPv6 socket"); + } finally { + await t.stop(); + } + }); +}); diff --git a/packages/discv5/test/unit/util/ip.test.ts b/packages/discv5/test/unit/util/ip.test.ts index 3c1b8cc..d4e9710 100644 --- a/packages/discv5/test/unit/util/ip.test.ts +++ b/packages/discv5/test/unit/util/ip.test.ts @@ -5,6 +5,7 @@ import {generateKeypair} from "../../../src/index.js"; import { type Ip, type SocketAddress, + getSocketAddressMultiaddrOnENR, getSocketAddressOnENR, getSocketAddressOnENRByFamily, isEqualSocketAddress, @@ -193,27 +194,51 @@ describe("get/set SocketAddress on ENR", () => { expect(getSocketAddressOnENRByFamily(enr, 6)).to.deep.equal(addr6); }); - it("keeps dual-stack ENR lookup preference for IPv6", () => { - const addr4: SocketAddress = { - ip: { - octets: Uint8Array.from([127, 0, 0, 1]), - type: 4, - }, - port: 53, - }; - const addr6: SocketAddress = { - ip: { - octets: Uint8Array.from([0, 1, 0, 2, 0, 3, 0, 4, 0, 5, 0, 6, 0, 7, 0, 8]), - type: 6, - }, - port: 54, - }; + const addr4: SocketAddress = { + ip: { + octets: Uint8Array.from([127, 0, 0, 1]), + type: 4, + }, + port: 53, + }; + const addr6: SocketAddress = { + ip: { + octets: Uint8Array.from([0, 1, 0, 2, 0, 3, 0, 4, 0, 5, 0, 6, 0, 7, 0, 8]), + type: 6, + }, + port: 54, + }; + const dualStackEnr = (): SignableENR => { const enr = SignableENR.createV4(generateKeypair("secp256k1").privateKey); - setSocketAddressOnENR(enr, addr4); setSocketAddressOnENR(enr, addr6); + return enr; + }; + + it("prefers IPv4 for dual-stack ENR by default", () => { + const enr = dualStackEnr(); + expect(getSocketAddressOnENR(enr, {ip4: true, ip6: true})).to.deep.equal(addr4); + expect(getSocketAddressMultiaddrOnENR(enr, {ip4: true, ip6: true})?.toString()).to.equal( + multiaddrFromSocketAddress(addr4).toString() + ); + }); + + it("prefers IPv6 for dual-stack ENR when requested", () => { + const enr = dualStackEnr(); + expect(getSocketAddressOnENR(enr, {ip4: true, ip6: true}, 6)).to.deep.equal(addr6); + expect(getSocketAddressMultiaddrOnENR(enr, {ip4: true, ip6: true}, 6)?.toString()).to.equal( + multiaddrFromSocketAddress(addr6).toString() + ); + }); + + it("falls back to the other family when the preferred one is unavailable", () => { + const enr6 = SignableENR.createV4(generateKeypair("secp256k1").privateKey); + setSocketAddressOnENR(enr6, addr6); + expect(getSocketAddressOnENR(enr6, {ip4: true, ip6: true})).to.deep.equal(addr6); - expect(getSocketAddressOnENR(enr, {ip4: true, ip6: true})).to.deep.equal(addr6); + const enr = dualStackEnr(); + expect(getSocketAddressOnENR(enr, {ip4: false, ip6: true})).to.deep.equal(addr6); + expect(getSocketAddressOnENR(enr, {ip4: true, ip6: false}, 6)).to.deep.equal(addr4); }); });