wpt: make runner more resilient (nodejs#1884)

* wpt: make runner more resilient

* wpt: cleaner exit, hopefully

package.json

@@ -55,7 +55,7 @@
     "test:tdd": "tap test/*.js test/diagnostics-channel/*.js -w",
     "test:typescript": "tsd && tsc test/imports/undici-import.ts",
     "test:websocket": "node scripts/verifyVersion.js 18 || tap test/websocket/*.js",
-    "test:wpt": "node scripts/verifyVersion 18 || (node test/wpt/start-fetch.mjs && node test/wpt/start-FileAPI.mjs && node test/wpt/start-mimesniff.mjs && node test/wpt/start-xhr.mjs && node test/wpt/start-websockets.mjs)",
+    "test:wpt": "node scripts/verifyVersion 18 || (node test/wpt/start-fetch.mjs && node test/wpt/start-FileAPI.mjs && node test/wpt/start-mimesniff.mjs && node test/wpt/start-xhr.mjs && node --no-warnings test/wpt/start-websockets.mjs)",
     "coverage": "nyc --reporter=text --reporter=html npm run test",
     "coverage:ci": "nyc --reporter=lcov npm run test",
     "bench": "PORT=3042 concurrently -k -s first npm:bench:server npm:bench:run",

test/wpt/runner/runner/runner.mjs

@@ -1,6 +1,7 @@
 import { deepStrictEqual } from 'node:assert'
-import { EventEmitter } from 'node:events'
+import { EventEmitter, once } from 'node:events'
 import { readdirSync, readFileSync, statSync } from 'node:fs'
+import { cpus } from 'node:os'
 import { basename, isAbsolute, join, resolve } from 'node:path'
 import { fileURLToPath } from 'node:url'
 import { Worker } from 'node:worker_threads'
@@ -82,10 +83,11 @@ export class WPTRunner extends EventEmitter {
     return [...files]
   }
 
-  run () {
+  async run () {
     const workerPath = fileURLToPath(join(import.meta.url, '../worker.mjs'))
     /** @type {Set<Worker>} */
     const activeWorkers = new Set()
+    let finishedFiles = 0
 
     for (const test of this.#files) {
       const code = test.includes('.sub.')
@@ -95,6 +97,7 @@ export class WPTRunner extends EventEmitter {
 
       if (this.#status[basename(test)]?.skip) {
         this.#stats.skipped += 1
+        finishedFiles++
         continue
       }
 
@@ -131,10 +134,14 @@ export class WPTRunner extends EventEmitter {
         activeWorkers.delete(worker)
         clearTimeout(timeout)
 
-        if (activeWorkers.size === 0) {
+        if (++finishedFiles === this.#files.length) {
           this.handleRunnerCompletion()
         }
       })
+
+      if (activeWorkers.size === cpus().length) {
+        await once(worker, 'exit')
+      }
     }
   }
 

test/wpt/server/server.mjs

@@ -354,7 +354,7 @@ send({ server: `http://localhost:${server.address().port}` })
 
 process.on('message', (message) => {
   if (message === 'shutdown') {
-    server.close((err) => err ? send(err) : send({ message: 'shutdown' }))
+    server.close((err) => process.exit(err ? 1 : 0))
   }
 })
 

test/wpt/start-FileAPI.mjs

@@ -10,15 +10,17 @@ const child = fork(serverPath, [], {
   stdio: ['pipe', 'pipe', 'pipe', 'ipc']
 })
 
+child.on('exit', (code) => process.exit(code))
+
 for await (const [message] of on(child, 'message')) {
   if (message.server) {
     const runner = new WPTRunner('FileAPI', message.server)
     runner.run()
 
     runner.once('completion', () => {
-      child.send('shutdown')
+      if (child.connected) {
+        child.send('shutdown')
+      }
     })
-  } else if (message.message === 'shutdown') {
-    process.exit()
   }
 }

test/wpt/start-fetch.mjs

@@ -10,15 +10,17 @@ const child = fork(serverPath, [], {
   stdio: ['pipe', 'pipe', 'pipe', 'ipc']
 })
 
+child.on('exit', (code) => process.exit(code))
+
 for await (const [message] of on(child, 'message')) {
   if (message.server) {
     const runner = new WPTRunner('fetch', message.server)
     runner.run()
 
     runner.once('completion', () => {
-      child.send('shutdown')
+      if (child.connected) {
+        child.send('shutdown')
+      }
     })
-  } else if (message.message === 'shutdown') {
-    process.exit()
   }
 }

test/wpt/start-mimesniff.mjs

@@ -10,15 +10,17 @@ const child = fork(serverPath, [], {
   stdio: ['pipe', 'pipe', 'pipe', 'ipc']
 })
 
+child.on('exit', (code) => process.exit(code))
+
 for await (const [message] of on(child, 'message')) {
   if (message.server) {
     const runner = new WPTRunner('mimesniff', message.server)
     runner.run()
 
     runner.once('completion', () => {
-      child.send('shutdown')
+      if (child.connected) {
+        child.send('shutdown')
+      }
     })
-  } else if (message.message === 'shutdown') {
-    process.exit()
   }
 }

test/wpt/start-websockets.mjs

@@ -10,6 +10,8 @@ const child = fork(serverPath, [], {
   stdio: ['pipe', 'pipe', 'pipe', 'ipc']
 })
 
+child.on('exit', (code) => process.exit(code))
+
 for await (const [message] of on(child, 'message')) {
   if (message.server) {
     const runner = new WPTRunner('websockets', message.server)
@@ -20,7 +22,5 @@ for await (const [message] of on(child, 'message')) {
         child.send('shutdown')
       }
     })
-  } else if (message.message === 'shutdown') {
-    process.exit()
   }
 }