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
8 changes: 4 additions & 4 deletions lib/dispatcher/client-h2.js
Original file line number Diff line number Diff line change
Expand Up @@ -279,10 +279,10 @@ function connectH2 (client, socket) {
if (client[kRunning] > 0) {
// We are already processing requests

// Non-idempotent request cannot be retried.
// Ensure that no other requests are inflight and
// could cause failure.
if (request.idempotent === false) return true
// Unlike HTTP/1.1 pipelining, HTTP/2 multiplexes requests on
// independent streams, so non-idempotent requests can be dispatched
// concurrently. Retry eligibility is handled by stream/session error
// handling instead of by serializing all non-idempotent requests.
// Don't dispatch an upgrade until all preceding requests have completed.
// Possibly, we do not have remote settings confirmed yet.
if ((request.upgrade === 'websocket' || request.method === 'CONNECT') && session[kRemoteSettings] === false) return true
Expand Down
85 changes: 85 additions & 0 deletions test/issue-5390.js
Original file line number Diff line number Diff line change
@@ -0,0 +1,85 @@
'use strict'

const { test, after } = require('node:test')
const { createSecureServer } = require('node:http2')
const { once } = require('node:events')
const { setTimeout: sleep } = require('node:timers/promises')

const { tspl } = require('@matteo.collina/tspl')
const pem = require('@metcoder95/https-pem')

const { Agent, request } = require('..')

test('HTTP/2 POST requests multiplex on an established session', async t => {
const p = tspl(t, { plan: 1 })

const server = createSecureServer(await pem.generate({ opts: { keySize: 2048 } }))
const measuredStreams = []
const measuredRequests = 5
let releaseImmediately = false

function respond (stream) {
if (!stream.closed && !stream.destroyed) {
stream.respond({ ':status': 200 })
stream.end('ok')
}
}

function flushMeasuredStreams () {
for (const stream of measuredStreams) {
respond(stream)
}
}

server.on('stream', (stream, headers) => {
stream.on('error', () => {})

if (headers[':path'] === '/warmup') {
respond(stream)
return
}

measuredStreams.push(stream)

if (releaseImmediately || measuredStreams.length === measuredRequests) {
releaseImmediately = true
flushMeasuredStreams()
}
})

after(() => server.close())
await once(server.listen(0), 'listening')

const agent = new Agent({
connect: {
rejectUnauthorized: false
},
allowH2: true,
connections: 1
})
after(() => agent.close())

const origin = `https://localhost:${server.address().port}`
const warmup = await request(`${origin}/warmup`, { dispatcher: agent })
await warmup.body.text()

const requests = Promise.all(Array.from({ length: measuredRequests }, () => {
return request(origin, {
dispatcher: agent,
method: 'POST',
body: 'hello'
}).then(response => response.body.text())
}))

await sleep(500)

const concurrentStreams = measuredStreams.length

releaseImmediately = true
flushMeasuredStreams()
await requests

p.strictEqual(concurrentStreams, measuredRequests)

await p.completed
})
Loading