Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 19 additions & 6 deletions lib/dispatcher/client-h2.js
Original file line number Diff line number Diff line change
Expand Up @@ -152,6 +152,21 @@ function requeueUnsentRequest (client, request) {
client[kQueue].splice(client[kPendingIdx] + 1, 0, request)
}

function completeRequest (client, request, resetPendingIdx = false) {
const index = client[kQueue].indexOf(request, client[kRunningIdx])

if (index === -1 || index >= client[kPendingIdx]) {
return
}

client[kQueue].splice(index, 1)
client[kPendingIdx]--

if (resetPendingIdx && client[kPendingIdx] < client[kRunningIdx]) {
client[kPendingIdx] = client[kRunningIdx]
}
}

function canRetryRequestAfterGoAway (request) {
const { body } = request

Expand Down Expand Up @@ -479,7 +494,9 @@ function onHttp2SessionClose () {
const requests = client[kQueue].splice(client[kRunningIdx])
for (let i = 0; i < requests.length; i++) {
const request = requests[i]
util.errorRequest(client, request, err)
if (request != null) {
util.errorRequest(client, request, err)
}
}
}
}
Expand Down Expand Up @@ -742,11 +759,7 @@ function writeH2 (client, request) {
}

requestFinalized = true
client[kQueue][client[kRunningIdx]++] = null

if (resetPendingIdx && client[kPendingIdx] < client[kRunningIdx]) {
client[kPendingIdx] = client[kRunningIdx]
}
completeRequest(client, request, resetPendingIdx)

client[kResume]()
}
Expand Down
8 changes: 6 additions & 2 deletions lib/dispatcher/client.js
Original file line number Diff line number Diff line change
Expand Up @@ -395,7 +395,9 @@ class Client extends DispatcherBase {
const requests = this[kQueue].splice(this[kPendingIdx])
for (let i = 0; i < requests.length; i++) {
const request = requests[i]
util.errorRequest(this, request, err)
if (request != null) {
util.errorRequest(this, request, err)
}
}

const callback = () => {
Expand Down Expand Up @@ -434,7 +436,9 @@ function onError (client, err) {

for (let i = 0; i < requests.length; i++) {
const request = requests[i]
util.errorRequest(client, request, err)
if (request != null) {
util.errorRequest(client, request, err)
}
}
assert(client[kSize] === 0)
}
Expand Down
69 changes: 69 additions & 0 deletions test/issue-5404.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
'use strict'

const { test } = require('node:test')
const assert = require('node:assert')
const { once } = require('node:events')
const { createServer } = require('node:http2')

const { Client } = require('..')
const { kQueue } = require('../lib/core/symbols')

// Regression test for https://github.com/nodejs/undici/issues/5404.
// HTTP/2 streams can complete out of order. Completing the second stream first
// must not clear the first request's queue slot; otherwise destroying the
// client can lose or mis-error the still-running request.
test('h2: out-of-order completion preserves running requests during destroy', async () => {
const server = createServer()
let firstStream

server.on('sessionError', () => {})
server.on('stream', (stream, headers) => {
switch (headers[':path']) {
case '/first':
firstStream = stream
break
case '/second':
stream.respond({ ':status': 200 })
stream.end('second')
break
case '/third':
stream.respond({ ':status': 200 })
stream.end('third')
break
default:
stream.respond({ ':status': 404 })
stream.end()
}
})

await once(server.listen(0), 'listening')

const client = new Client(`http://localhost:${server.address().port}`, {
allowH2: true,
useH2c: true,
maxConcurrentStreams: 2
})

try {
const first = client.request({ path: '/first', method: 'GET' })
const firstError = first.then(
() => null,
err => err
)

const second = await client.request({ path: '/second', method: 'GET' })
assert.strictEqual(await second.body.text(), 'second')

const third = await client.request({ path: '/third', method: 'GET' })
assert.strictEqual(await third.body.text(), 'third')

assert.strictEqual(firstStream.destroyed, false)
assert.deepStrictEqual(client[kQueue].map(request => request?.path), ['/first'])

await client.destroy(new Error('boom'))
assert.strictEqual((await firstError).message, 'boom')
} finally {
await client.destroy().catch(() => {})
server.close()
}
})
Loading