Repository: discordjs/voice
Pull request #89 (2021-04-02 15:04:23)

feat: keep alive mechanism to detect connectivity to Discord
**Please describe the changes this PR makes and why it should be merged:**

Adds a keep alive mechanism to the UDP socket, that allows 60 seconds for Discord to echo keep alive messages back to the sender. If 60 seconds pass, the UDP socket emits an error warning that a connection between the client and Discord is likely unavailable, causing the voice connection to attempt to reconnect.

A breaking change because `connection.ping` is now of type ` { ws: ?number, udp: ?number }` instead of just `?number`.

**Status and versioning classification:**

- Code changes have been tested against the Discord API, or there are no code changes
- This PR changes the library's interface (methods or parameters added)
- This PR includes breaking changes (methods removed or renamed, parameters moved or removed)

codecov[bot]: # [Codecov](https://codecov.io/gh/discordjs/voice/pull/89?src=pr&el=h1) Report
> Merging [#89](https://codecov.io/gh/discordjs/voice/pull/89?src=pr&el=desc) (0b672e1) into [main](https://codecov.io/gh/discordjs/voice/commit/0ce79d62def0eea9d875eba57ec271a034ce5977?el=desc) (0ce79d6) will **increase** coverage by `1.53%`.
> The diff coverage is `89.65%`.

[![Impacted file tree graph](https://codecov.io/gh/discordjs/voice/pull/89/graphs/tree.svg?width=650&height=150&src=pr&token=u7oQ23UoxX)](https://codecov.io/gh/discordjs/voice/pull/89?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##             main      #89      +/-   ##
==========================================
+ Coverage   55.41%   56.94%   +1.53%     
==========================================
  Files          15       15              
  Lines         637      662      +25     
  Branches      143      147       +4     
==========================================
+ Hits          353      377      +24     
- Misses        282      283       +1     
  Partials        2        2              
```


| [Impacted Files](https://codecov.io/gh/discordjs/voice/pull/89?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [src/VoiceConnection.ts](https://codecov.io/gh/discordjs/voice/pull/89/diff?src=pr&el=tree#diff-c3JjL1ZvaWNlQ29ubmVjdGlvbi50cw==) | `78.57% <0.00%> (-0.71%)` | :arrow_down: |
| [src/networking/VoiceUDPSocket.ts](https://codecov.io/gh/discordjs/voice/pull/89/diff?src=pr&el=tree#diff-c3JjL25ldHdvcmtpbmcvVm9pY2VVRFBTb2NrZXQudHM=) | `97.91% <96.29%> (+2.08%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/discordjs/voice/pull/89?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/discordjs/voice/pull/89?src=pr&el=footer). Last update [0ce79d6...0b672e1](https://codecov.io/gh/discordjs/voice/pull/89?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/src/VoiceConnection.ts b/src/VoiceConnection.ts
index c3a898b..e00b4d1 100644
--- a/src/VoiceConnection.ts
+++ b/src/VoiceConnection.ts
@@ -485,21 +485,27 @@ export class VoiceConnection extends (EventEmitter as new () => TypedEmitter<Voi
 	}
 
 	/**
-	 * The latest ping (in milliseconds) for the WebSocket connection for this voice connection, if this
-	 * data is available.
+	 * The latest ping (in milliseconds) for the WebSocket connection and audio playback for this voice
+	 * connection, if this data is available.
 	 *
 	 * @remarks
 	 * For this data to be available, the VoiceConnection must be in the Ready state, and its underlying
-	 * WebSocket connection must have had at least one ping-pong exchange.
+	 * WebSocket connection and UDP socket must have had at least one ping-pong exchange.
 	 */
 	public get ping() {
 		if (
 			this.state.status === VoiceConnectionStatus.Ready &&
-			this.state.networking.state.code === NetworkingStatusCode.Ready &&
-			typeof this.state.networking.state.ws.ping !== 'undefined'
+			this.state.networking.state.code === NetworkingStatusCode.Ready
 		) {
-			return this.state.networking.state.ws.ping;
+			return {
+				ws: this.state.networking.state.ws.ping,
+				udp: this.state.networking.state.udp.ping,
+			};
 		}
+		return {
+			ws: undefined,
+			udp: undefined,
+		};
 	}
 
 	/**
diff --git a/src/networking/Networking.ts b/src/networking/Networking.ts
index c3182e0..105bcec 100644
--- a/src/networking/Networking.ts
+++ b/src/networking/Networking.ts
@@ -171,6 +171,8 @@ export class Networking extends EventEmitter {
 		this.onWsPacket = this.onWsPacket.bind(this);
 		this.onWsClose = this.onWsClose.bind(this);
 		this.onWsDebug = this.onWsDebug.bind(this);
+		this.onUdpDebug = this.onUdpDebug.bind(this);
+		this.onUdpClose = this.onUdpClose.bind(this);
 
 		this.debug = debug ? this.emit.bind(this, 'debug') : null;
 
@@ -220,6 +222,8 @@ export class Networking extends EventEmitter {
 		if (oldUdp && oldUdp !== newUdp) {
 			oldUdp.on('error', noop);
 			oldUdp.off('error', this.onChildError);
+			oldUdp.off('close', this.onUdpClose);
+			oldUdp.off('debug', this.onUdpDebug);
 			oldUdp.destroy();
 		}
 
@@ -317,6 +321,19 @@ export class Networking extends EventEmitter {
 		}
 	}
 
+	/**
+	 * Called when the UDP socket has closed itself if it has stopped receiving replies from Discord
+	 */
+	private onUdpClose() {
+		if (this.state.code === NetworkingStatusCode.Ready) {
+			this.state = {
+				...this.state,
+				code: NetworkingStatusCode.Resuming,
+				ws: this.createWebSocket(this.state.connectionOptions.endpoint),
+			};
+		}
+	}
+
 	/**
 	 * Called when a packet is received on the connection's WebSocket
 	 * @param packet - The received packet
@@ -329,6 +346,8 @@ export class Networking extends EventEmitter {
 
 			const udp = new VoiceUDPSocket({ ip, port });
 			udp.on('error', this.onChildError);
+			udp.on('debug', this.onUdpDebug);
+			udp.once('close', this.onUdpClose);
 			udp
 				.performIPDiscovery(ssrc)
 				.then((localConfig) => {
@@ -396,6 +415,15 @@ export class Networking extends EventEmitter {
 		this.debug?.(`[WS] ${message}`);
 	}
 
+	/**
+	 * Propagates debug messages from the child UDPSocket.
+	 *
+	 * @param message - The emitted debug message
+	 */
+	private onUdpDebug(message: string) {
+		this.debug?.(`[UDP] ${message}`);
+	}
+
 	/**
 	 * Prepares an Opus packet for playback. This includes attaching metadata to it and encrypting it.
 	 * It will be stored within the instance, and can be played by dispatchAudio().
diff --git a/src/networking/VoiceUDPSocket.ts b/src/networking/VoiceUDPSocket.ts
index cfc6854..7ad3677 100644
--- a/src/networking/VoiceUDPSocket.ts
+++ b/src/networking/VoiceUDPSocket.ts
@@ -11,6 +11,26 @@ export interface SocketConfig {
 	port: number;
 }
 
+interface KeepAlive {
+	value: number;
+	timestamp: number;
+}
+
+/**
+ * The interval in milliseconds at which keep alive datagrams are sent
+ */
+const KEEP_ALIVE_INTERVAL = 5e3;
+
+/**
+ * The maximum number of keep alive packets which can be missed
+ */
+const KEEP_ALIVE_LIMIT = 12;
+
+/**
+ * The maximum value of the keep alive counter
+ */
+const MAX_COUNTER_VALUE = 2 ** 32 - 1;
+
 /**
  * Manages the UDP networking for a voice connection.
  */
@@ -25,16 +45,92 @@ export class VoiceUDPSocket extends EventEmitter {
 	 */
 	private readonly remote: SocketConfig;
 
+	/**
+	 * A list of keep alives that are waiting to be acknowledged.
+	 */
+	private readonly keepAlives: KeepAlive[];
+
+	/**
+	 * The counter used in the keep alive mechanism
+	 */
+	private keepAliveCounter = 0;
+
+	/**
+	 * The buffer used to write the keep alive counter into
+	 */
+	private readonly keepAliveBuffer: Buffer;
+
+	/**
+	 * The Node.js interval for the keep-alive mechanism
+	 */
+	private readonly keepAliveInterval: NodeJS.Timeout;
+
+	/**
+	 * The time taken to receive a response to keep alive messages
+	 */
+	public ping?: number;
+
+	/**
+	 * The debug logger function, if debugging is enabled.
+	 */
+	private readonly debug: null | ((message: string) => void);
+
 	/**
 	 * Creates a new VoiceUDPSocket.
 	 *
 	 * @param remote - Details of the remote socket
 	 */
-	public constructor(remote: SocketConfig) {
+	public constructor(remote: SocketConfig, debug = false) {
 		super();
 		this.socket = createSocket('udp4');
 		this.socket.on('error', (error: Error) => this.emit('error', error));
+		this.socket.on('message', (buffer: Buffer) => this.onMessage(buffer));
+		this.socket.on('close', () => this.emit('close'));
 		this.remote = remote;
+		this.keepAlives = [];
+		this.keepAliveBuffer = Buffer.alloc(8);
+		this.keepAliveInterval = setInterval(() => this.keepAlive(), KEEP_ALIVE_INTERVAL);
+		setImmediate(() => this.keepAlive());
+
+		this.debug = debug ? this.emit.bind(this, 'debug') : null;
+	}
+
+	/**
+	 * Called when a message is received on the UDP socket
+	 * @param buffer The received buffer
+	 */
+	private onMessage(buffer: Buffer): void {
+		// Handle keep alive message
+		if (buffer.length === 8) {
+			const counter = buffer.readUInt32LE(0);
+			const index = this.keepAlives.findIndex(({ value }) => value === counter);
+			if (index === -1) return;
+			this.ping = Date.now() - this.keepAlives[index].timestamp;
+			// Delete all keep alives up to and including the received one
+			this.keepAlives.splice(0, index);
+		}
+	}
+
+	/**
+	 * Called at a regular interval to check whether we are still able to send datagrams to Discord
+	 */
+	private keepAlive() {
+		if (this.keepAlives.length >= KEEP_ALIVE_LIMIT) {
+			this.debug?.('UDP socket has not received enough responses from Discord - closing socket');
+			this.destroy();
+			return;
+		}
+
+		this.keepAliveBuffer.writeUInt32LE(this.keepAliveCounter, 0);
+		this.send(this.keepAliveBuffer);
+		this.keepAlives.push({
+			value: this.keepAliveCounter,
+			timestamp: Date.now(),
+		});
+		this.keepAliveCounter++;
+		if (this.keepAliveCounter > MAX_COUNTER_VALUE) {
+			this.keepAliveCounter = 0;
+		}
 	}
 
 	/**
@@ -51,6 +147,7 @@ export class VoiceUDPSocket extends EventEmitter {
 	 */
 	public destroy() {
 		this.socket.close();
+		clearInterval(this.keepAliveInterval);
 	}
 
 	/**
@@ -85,14 +182,14 @@ export class VoiceUDPSocket extends EventEmitter {
  */
 export function parseLocalPacket(message: Buffer): SocketConfig {
 	const packet = Buffer.from(message);
-	let ip = '';
-	for (let i = 4; i < packet.indexOf(0, i); i++) ip += String.fromCharCode(packet[i]);
+
+	const ip = packet.slice(4, packet.indexOf(0, 4)).toString('utf-8');
 
 	if (!isIPv4(ip)) {
 		throw new Error('Malformed IP address');
 	}
 
-	const port = parseInt(packet.readUIntLE(packet.length - 2, 2).toString(10), 10);
+	const port = packet.readUInt16LE(packet.length - 2);
 
 	return { ip, port };
 }


Test patch:
diff --git a/src/networking/__tests__/VoiceUDPSocket.test.ts b/src/networking/__tests__/VoiceUDPSocket.test.ts
index 8f988fb..543be38 100644
--- a/src/networking/__tests__/VoiceUDPSocket.test.ts
+++ b/src/networking/__tests__/VoiceUDPSocket.test.ts
@@ -1,9 +1,11 @@
 /* eslint-disable @typescript-eslint/no-empty-function */
 /* eslint-disable @typescript-eslint/no-unused-vars */
 import { createSocket as _createSocket } from 'dgram';
-import EventEmitter from 'events';
-import { parseLocalPacket, VoiceUDPSocket } from '../VoiceUDPSocket';
+import EventEmitter, { once } from 'events';
+import { VoiceUDPSocket } from '../VoiceUDPSocket';
+
 jest.mock('dgram');
+jest.useFakeTimers();
 
 const createSocket = (_createSocket as unknown) as jest.Mock<typeof _createSocket>;
 
@@ -25,7 +27,17 @@ const VALID_RESPONSE = Buffer.concat([
 	Buffer.from([0x8d, 0xa5]),
 ]);
 
+function wait() {
+	return new Promise((resolve) => process.nextTick(resolve));
+}
+
 describe('VoiceUDPSocket#performIPDiscovery', () => {
+	let socket: VoiceUDPSocket;
+
+	afterEach(() => {
+		socket.destroy();
+	});
+
 	/*
 		Ensures that the UDP socket sends data and parses the response correctly
 	*/
@@ -35,15 +47,16 @@ describe('VoiceUDPSocket#performIPDiscovery', () => {
 			fake.emit('message', VALID_RESPONSE);
 		});
 		createSocket.mockImplementation((type) => fake as any);
-		const socket = new VoiceUDPSocket({ ip: '1.2.3.4', port: 25565 });
+		socket = new VoiceUDPSocket({ ip: '1.2.3.4', port: 25565 });
 
 		expect(createSocket).toHaveBeenCalledWith('udp4');
+		expect(fake.listenerCount('message')).toBe(1);
 		await expect(socket.performIPDiscovery(1234)).resolves.toEqual({
 			ip: '94.195.157.2',
 			port: 42381,
 		});
 		// Ensure clean up occurs
-		expect(fake.listenerCount('message')).toBe(0);
+		expect(fake.listenerCount('message')).toBe(1);
 	});
 
 	/*
@@ -57,16 +70,17 @@ describe('VoiceUDPSocket#performIPDiscovery', () => {
 			fake.emit('message', fakeResponse);
 			setImmediate(() => fake.emit('message', VALID_RESPONSE));
 		});
-		createSocket.mockImplementation((type) => fake as any);
-		const socket = new VoiceUDPSocket({ ip: '1.2.3.4', port: 25565 });
+		createSocket.mockImplementation(() => fake as any);
+		socket = new VoiceUDPSocket({ ip: '1.2.3.4', port: 25565 });
 
 		expect(createSocket).toHaveBeenCalledWith('udp4');
+		expect(fake.listenerCount('message')).toBe(1);
 		await expect(socket.performIPDiscovery(1234)).resolves.toEqual({
 			ip: '94.195.157.2',
 			port: 42381,
 		});
 		// Ensure clean up occurs
-		expect(fake.listenerCount('message')).toBe(0);
+		expect(fake.listenerCount('message')).toBe(1);
 	});
 
 	test('Rejects if socket closes before IP discovery can be completed', async () => {
@@ -74,10 +88,67 @@ describe('VoiceUDPSocket#performIPDiscovery', () => {
 		fake.send = jest.fn().mockImplementation((buffer: Buffer, port: number, address: string) => {
 			setImmediate(() => fake.close());
 		});
-		createSocket.mockImplementation((type) => fake as any);
-		const socket = new VoiceUDPSocket({ ip: '1.2.3.4', port: 25565 });
+		createSocket.mockImplementation(() => fake as any);
+		socket = new VoiceUDPSocket({ ip: '1.2.3.4', port: 25565 });
 
 		expect(createSocket).toHaveBeenCalledWith('udp4');
 		await expect(socket.performIPDiscovery(1234)).rejects.toThrowError();
 	});
+
+	test('Stays alive when messages are echoed back', async () => {
+		const fake = new FakeSocket();
+		fake.send = jest.fn().mockImplementation((buffer: Buffer) => process.nextTick(() => fake.emit('message', buffer)));
+		createSocket.mockImplementation(() => fake as any);
+		socket = new VoiceUDPSocket({ ip: '1.2.3.4', port: 25565 });
+
+		let closed = false;
+		socket.on('close', () => (closed = true));
+
+		for (let i = 0; i < 30; i++) {
+			jest.advanceTimersToNextTimer();
+			await wait();
+		}
+
+		expect(closed).toBe(false);
+	});
+
+	test('Emits an error when no response received to keep alive messages', async () => {
+		const fake = new FakeSocket();
+		fake.send = jest.fn();
+		createSocket.mockImplementation(() => fake as any);
+		socket = new VoiceUDPSocket({ ip: '1.2.3.4', port: 25565 });
+
+		let closed = false;
+		socket.on('close', () => (closed = true));
+
+		for (let i = 0; i < 15; i++) {
+			jest.advanceTimersToNextTimer();
+			await wait();
+		}
+
+		expect(closed).toBe(true);
+	});
+
+	test('Recovers from intermittent responses', async () => {
+		const fake = new FakeSocket();
+		const fakeSend = jest.fn();
+		fake.send = fakeSend;
+		createSocket.mockImplementation(() => fake as any);
+		socket = new VoiceUDPSocket({ ip: '1.2.3.4', port: 25565 });
+
+		let closed = false;
+		socket.on('close', () => (closed = true));
+
+		for (let i = 0; i < 10; i++) {
+			jest.advanceTimersToNextTimer();
+			await wait();
+		}
+		fakeSend.mockImplementation((buffer: Buffer) => process.nextTick(() => fake.emit('message', buffer)));
+		expect(closed).toBe(false);
+		for (let i = 0; i < 30; i++) {
+			jest.advanceTimersToNextTimer();
+			await wait();
+		}
+		expect(closed).toBe(false);
+	});
 });