fetch: fix wpt test request-upload.any.js (#3234)

* fix

* fix

Author: Uzlopak

test/wpt/server/lockedresource.mjs

@@ -0,0 +1,31 @@
+export class LockedResource {
+  constructor (resource) {
+    this.locked = false
+    this.waitingQueue = []
+    this.resource = resource
+  }
+
+  async acquire () {
+    return new Promise(resolve => {
+      if (!this.locked) {
+        // If the lock is not already acquired, acquire it immediately
+        this.locked = true
+        resolve(this.resource)
+      } else {
+        // If the lock is already acquired, queue the resolve function
+        this.waitingQueue.push(resolve)
+      }
+    })
+  }
+
+  release () {
+    if (this.waitingQueue.length > 0) {
+      // If there are functions waiting to acquire the lock, execute the next one in the queue
+      const nextResolve = this.waitingQueue.shift()
+      nextResolve(this.resource)
+    } else {
+      // If there are no functions waiting, release the lock
+      this.locked = false
+    }
+  }
+}

test/wpt/server/routes/network-partition-key.mjs

@@ -1,111 +1,132 @@
+import { LockedResource } from '../lockedresource.mjs'
+
 const stash = new Map()
+const lockedStash = new LockedResource(stash)
 
 /**
  * @see https://github.com/web-platform-tests/wpt/blob/master/fetch/connection-pool/resources/network-partition-key.py
- * @param {Parameters<import('http').RequestListener>[0]} req
- * @param {Parameters<import('http').RequestListener>[1]} res
+ * @param {Parameters<import('http').RequestListener>[0]} request
+ * @param {Parameters<import('http').RequestListener>[1]} response
  * @param {URL} url
  */
-export function route (req, res, { searchParams, port }) {
-  res.setHeader('Cache-Control', 'no-store')
+export async function route (request, response, { searchParams, port }) {
+  response.setHeader('Cache-Control', 'no-store')
 
   const dispatch = searchParams.get('dispatch')
   const uuid = searchParams.get('uuid')
   const partitionId = searchParams.get('partition_id')
 
   if (!uuid || !dispatch || !partitionId) {
-    res.statusCode = 404
-    res.end('Invalid query parameters')
-    return
+    return simpleResponse(request, response, 404, 'Not found', 'Invalid query parameters')
   }
 
   let testFailed = false
   let requestCount = 0
   let connectionCount = 0
 
-  if (searchParams.get('nocheck_partition') !== 'True') {
-    const addressKey = `${req.socket.localAddress}|${port}`
-    const serverState = stash.get(uuid) ?? {
-      testFailed: false,
-      requestCount: 0,
-      connectionCount: 0
-    }
+  if (searchParams.get('nocheck_partition') !== 'true') {
+    const stash = await lockedStash.acquire()
+    try {
+      const addressKey = `${request.socket.localAddress}|${port}`
+      const serverState = stash.get(uuid) ?? {
+        testFailed: false,
+        requestCount: 0,
+        connectionCount: 0,
+        sockets: new Set()
+      }
 
-    stash.delete(uuid)
-    requestCount = serverState.requestCount + 1
-    serverState.requestCount = requestCount
+      stash.delete(uuid)
+      requestCount = serverState.requestCount
+      requestCount += 1
+      serverState.requestCount = requestCount
 
-    if (Object.hasOwn(serverState, addressKey)) {
-      if (serverState[addressKey] !== partitionId) {
-        serverState.testFailed = true
+      if (addressKey in serverState) {
+        if (serverState[addressKey] !== partitionId) {
+          serverState.testFailed = true
+        }
+      }
+
+      // We can detect if a new connection is created by checking if the socket
+      // was already used in the test.
+      if (serverState.sockets.has(request.socket) === false) {
+        connectionCount = serverState.connectionCount
+        connectionCount += 1
+        serverState.connectionCount = connectionCount
+        serverState.sockets.add(request.socket)
       }
-    } else {
-      connectionCount = serverState.connectionCount + 1
-      serverState.connectionCount = connectionCount
-    }
 
-    serverState[addressKey] = partitionId
-    testFailed = serverState.testFailed
-    stash.set(uuid, serverState)
+      serverState[addressKey] = partitionId
+      testFailed = serverState.testFailed
+      stash.set(uuid, serverState)
+    } finally {
+      lockedStash.release()
+    }
   }
 
-  const origin = req.headers.origin
+  const origin = request.headers.origin
   if (origin) {
-    res.setHeader('Access-Control-Allow-Origin', origin)
-    res.setHeader('Access-Control-Allow-Credentials', 'true')
+    response.setHeader('Access-Control-Allow-Origin', origin)
+    response.setHeader('Access-Control-Allow-Credentials', 'true')
   }
 
-  if (req.method === 'OPTIONS') {
-    return handlePreflight(req, res)
+  if (request.method === 'OPTIONS') {
+    return handlePreflight(request, response)
   }
 
   if (dispatch === 'fetch_file') {
-    res.end()
-    return
+    // There is currently no relevant wpt test that uses this dispatch
+    return response.end()
   }
 
   if (dispatch === 'check_partition') {
     const status = searchParams.get('status') ?? 200
 
     if (testFailed) {
-      res.statusCode = status
-      res.end('Multiple partition IDs used on a socket')
-      return
+      return simpleResponse(request, response, status, 'OK', 'Multiple partition IDs used on a socket')
     }
 
     let body = 'ok'
-    if (searchParams.get('addcounter')) {
+    if (searchParams.get('addcounter') === 'true') {
       body += `. Request was sent ${requestCount} times. ${connectionCount} connections were created.`
-      res.statusCode = status
-      res.end(body)
-      return
+      return simpleResponse(request, response, status, 'OK', body)
     }
   }
 
   if (dispatch === 'clean_up') {
     stash.delete(uuid)
-    res.statusCode = 200
     if (testFailed) {
-      res.end('Test failed, but cleanup completed.')
-    } else {
-      res.end('cleanup complete')
+      return simpleResponse(request, response, 200, 'OK', 'Test failed, but cleanup completed.')
     }
-
-    return
+    return simpleResponse(request, response, 200, 'OK', 'cleanup complete')
   }
 
-  res.statusCode = 404
-  res.end('Unrecognized dispatch parameter: ' + dispatch)
+  return simpleResponse(request, response, 404, 'Not found', 'Unrecognized dispatch parameter: ' + dispatch)
+}
+
+/**
+ * @param {Parameters<import('http').RequestListener>[0]} request
+ * @param {Parameters<import('http').RequestListener>[1]} response
+ */
+function handlePreflight (request, response) {
+  response.statusCode = 200
+  response.statusMessage = 'OK'
+  response.setHeader('Access-Control-Allow-Methods', 'GET')
+  response.setHeader('Access-Control-Allow-Headers', 'header-to-force-cors')
+  response.setHeader('Access-Control-Max-Age', '86400')
+  response.end('Preflight request')
 }
 
 /**
- * @param {Parameters<import('http').RequestListener>[0]} req
- * @param {Parameters<import('http').RequestListener>[1]} res
+ * @param {Parameters<import('http').RequestListener>[0]} request
+ * @param {Parameters<import('http').RequestListener>[1]} response
+ * @param {number} statusCode
+ * @param {string} statusMessage
+ * @param {string} body
+ * @param {string} [contentType='text/plain']
  */
-function handlePreflight (req, res) {
-  res.statusCode = 200
-  res.setHeader('Access-Control-Allow-Methods', 'GET')
-  res.setHeader('Access-Control-Allow-Headers', 'header-to-force-cors')
-  res.setHeader('Access-Control-Max-Age', '86400')
-  res.end('Preflight request')
+function simpleResponse (request, response, statusCode, statusMessage, body, contentType = 'text/plain') {
+  response.statusCode = statusCode
+  response.statusMessage = statusMessage
+  response.setHeader('Content-Type', contentType)
+  response.end(body)
 }

test/wpt/status/fetch.status.json

@@ -68,11 +68,6 @@
         "note": "undici doesn't filter headers",
         "skip": true
       },
-      "request-upload.any.js": {
-        "fail": [
-          "Fetch with POST with text body on 421 response should be retried once on new connection."	
-        ]
-      },
       "request-upload.h2.any.js": {
         "note": "undici doesn't support http/2",
         "skip": true