Skip to content

Commit 39f93e1

Browse files
UlisesGasconmcollina
authored andcommitted
fix(websocket): bound automatic PONG writes under backpressure
Signed-off-by: Matteo Collina <hello@matteocollina.com> (cherry picked from commit b217373) (cherry picked from commit 5394462)
1 parent c43b23f commit 39f93e1

2 files changed

Lines changed: 115 additions & 0 deletions

File tree

‎lib/web/websocket/receiver.js‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -249,6 +249,20 @@ class ByteParser extends Writable {
249249
if (isControlFrame(this.#info.opcode)) {
250250
this.#loop = this.parseControlFrame(body)
251251
this.#state = parserStates.INFO
252+
253+
// If the automatic PONG/CLOSE reply filled the socket's write buffer,
254+
// stop parsing and resume on drain so a PING flood cannot grow the
255+
// outbound queue without bound.
256+
const socket = this.ws[kResponse].socket
257+
if (this.#loop && socket.writableNeedDrain) {
258+
this.#loop = false
259+
socket.once('drain', () => {
260+
if (this.destroyed) return
261+
this.#loop = true
262+
this.run(callback)
263+
})
264+
return
265+
}
252266
} else {
253267
if (!this.#info.compressed) {
254268
if (!this.writeFragments(body)) {
Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
'use strict'
2+
3+
const assert = require('node:assert')
4+
const { test } = require('node:test')
5+
const { EventEmitter } = require('node:events')
6+
const { ByteParser } = require('../../lib/web/websocket/receiver')
7+
const { kReceivedClose, kResponse } = require('../../lib/web/websocket/symbols')
8+
9+
// An unmasked, empty PING frame as sent by a server (FIN + opcode 0x9).
10+
const serverPingFrame = Buffer.from([0x89, 0x00])
11+
12+
// A socket stub whose writes report backpressure (write() returns false and
13+
// writableNeedDrain stays set) until drain() is called.
14+
function createBackpressuredSocket () {
15+
const socket = new EventEmitter()
16+
socket.pongWrites = 0
17+
socket.blocked = true
18+
socket.writableNeedDrain = false
19+
socket.write = () => {
20+
socket.pongWrites += 1
21+
if (socket.blocked) {
22+
socket.writableNeedDrain = true
23+
return false
24+
}
25+
return true
26+
}
27+
socket.drain = () => {
28+
socket.writableNeedDrain = false
29+
socket.emit('drain')
30+
}
31+
return socket
32+
}
33+
34+
function createPingHandler (socket) {
35+
return {
36+
[kReceivedClose]: false,
37+
[kResponse]: { socket }
38+
}
39+
}
40+
41+
test('ByteParser stops auto-writing PONGs once the socket applies backpressure', () => {
42+
const socket = createBackpressuredSocket()
43+
const parser = new ByteParser(createPingHandler(socket))
44+
const pingFlood = Buffer.concat(Array.from({ length: 50 }, () => serverPingFrame))
45+
46+
parser.write(pingFlood)
47+
48+
return new Promise((resolve) => {
49+
setImmediate(() => {
50+
assert.strictEqual(socket.pongWrites, 1)
51+
parser.destroy()
52+
resolve()
53+
})
54+
})
55+
})
56+
57+
test('ByteParser resumes auto-writing PONGs after the socket drains', () => {
58+
const socket = createBackpressuredSocket()
59+
const parser = new ByteParser(createPingHandler(socket))
60+
const pingFlood = Buffer.concat(Array.from({ length: 50 }, () => serverPingFrame))
61+
62+
parser.write(pingFlood)
63+
64+
return new Promise((resolve) => {
65+
setImmediate(() => {
66+
assert.strictEqual(socket.pongWrites, 1)
67+
68+
socket.blocked = false
69+
socket.drain()
70+
71+
setImmediate(() => {
72+
assert.strictEqual(socket.pongWrites, 50)
73+
parser.destroy()
74+
resolve()
75+
})
76+
})
77+
})
78+
})
79+
80+
test('ByteParser does not resume PONG writes if destroyed while paused', () => {
81+
const socket = createBackpressuredSocket()
82+
const parser = new ByteParser(createPingHandler(socket))
83+
const pingFlood = Buffer.concat(Array.from({ length: 50 }, () => serverPingFrame))
84+
85+
parser.write(pingFlood)
86+
87+
return new Promise((resolve) => {
88+
setImmediate(() => {
89+
assert.strictEqual(socket.pongWrites, 1)
90+
91+
parser.destroy()
92+
socket.blocked = false
93+
socket.drain()
94+
95+
setImmediate(() => {
96+
assert.strictEqual(socket.pongWrites, 1)
97+
resolve()
98+
})
99+
})
100+
})
101+
})

0 commit comments

Comments
 (0)