.github/workflows/nodejs.yml

@@ -20,6 +20,9 @@ jobs:
       include: |
         - runs-on: ubuntu-latest
           node-version: 16.8
+      exclude: |
+        - runs-on: windows-latest
+          node-version: 16
   automerge:
     if: >
         github.event_name == 'pull_request' &&

docs/best-practices/proxy.md

@@ -20,10 +20,10 @@ import { createServer } from 'http'
 import proxy from 'proxy'
 
 const server = await buildServer()
-const proxy = await buildProxy()
+const proxyServer = await buildProxy()
 
 const serverUrl = `http://localhost:${server.address().port}`
-const proxyUrl = `http://localhost:${proxy.address().port}`
+const proxyUrl = `http://localhost:${proxyServer.address().port}`
 
 server.on('request', (req, res) => {
   console.log(req.url) // '/hello?foo=bar'
@@ -47,7 +47,7 @@ console.log(response.statusCode) // 200
 console.log(JSON.parse(data)) // { hello: 'world' }
 
 server.close()
-proxy.close()
+proxyServer.close()
 client.close()
 
 function buildServer () {
@@ -73,12 +73,12 @@ import { createServer } from 'http'
 import proxy from 'proxy'
 
 const server = await buildServer()
-const proxy = await buildProxy()
+const proxyServer = await buildProxy()
 
 const serverUrl = `http://localhost:${server.address().port}`
-const proxyUrl = `http://localhost:${proxy.address().port}`
+const proxyUrl = `http://localhost:${proxyServer.address().port}`
 
-proxy.authenticate = function (req, fn) {
+proxyServer.authenticate = function (req, fn) {
   fn(null, req.headers['proxy-authorization'] === `Basic ${Buffer.from('user:pass').toString('base64')}`)
 }
 
@@ -107,7 +107,7 @@ console.log(response.statusCode) // 200
 console.log(JSON.parse(data)) // { hello: 'world' }
 
 server.close()
-proxy.close()
+proxyServer.close()
 client.close()
 
 function buildServer () {
@@ -124,3 +124,4 @@ function buildProxy () {
   })
 }
 ```
+

lib/balanced-pool.js

@@ -18,6 +18,17 @@ const { parseOrigin } = require('./core/util')
 const kFactory = Symbol('factory')
 
 const kOptions = Symbol('options')
+const kGreatestCommonDivisor = Symbol('kGreatestCommonDivisor')
+const kCurrentWeight = Symbol('kCurrentWeight')
+const kIndex = Symbol('kIndex')
+const kWeight = Symbol('kWeight')
+const kMaxWeightPerServer = Symbol('kMaxWeightPerServer')
+const kErrorPenalty = Symbol('kErrorPenalty')
+
+function getGreatestCommonDivisor (a, b) {
+  if (b === 0) return a
+  return getGreatestCommonDivisor(b, a % b)
+}
 
 function defaultFactory (origin, opts) {
   return new Pool(origin, opts)
@@ -28,6 +39,11 @@ class BalancedPool extends PoolBase {
     super()
 
     this[kOptions] = opts
+    this[kIndex] = -1
+    this[kCurrentWeight] = 0
+
+    this[kMaxWeightPerServer] = this[kOptions].maxWeightPerServer || 100
+    this[kErrorPenalty] = this[kOptions].errorPenalty || 15
 
     if (!Array.isArray(upstreams)) {
       upstreams = [upstreams]
@@ -42,6 +58,7 @@ class BalancedPool extends PoolBase {
     for (const upstream of upstreams) {
       this.addUpstream(upstream)
     }
+    this._updateBalancedPoolStats()
   }
 
   addUpstream (upstream) {
@@ -54,12 +71,40 @@ class BalancedPool extends PoolBase {
     ))) {
       return this
     }
+    const pool = this[kFactory](upstreamOrigin, Object.assign({}, this[kOptions]))
+
+    this[kAddClient](pool)
+    pool.on('connect', () => {
+      pool[kWeight] = Math.min(this[kMaxWeightPerServer], pool[kWeight] + this[kErrorPenalty])
+    })
+
+    pool.on('connectionError', () => {
+      pool[kWeight] = Math.max(1, pool[kWeight] - this[kErrorPenalty])
+      this._updateBalancedPoolStats()
+    })
+
+    pool.on('disconnect', (...args) => {
+      const err = args[2]
+      if (err && err.code === 'UND_ERR_SOCKET') {
+        // decrease the weight of the pool.
+        pool[kWeight] = Math.max(1, pool[kWeight] - this[kErrorPenalty])
+        this._updateBalancedPoolStats()
+      }
+    })
+
+    for (const client of this[kClients]) {
+      client[kWeight] = this[kMaxWeightPerServer]
+    }
 
-    this[kAddClient](this[kFactory](upstreamOrigin, Object.assign({}, this[kOptions])))
+    this._updateBalancedPoolStats()
 
     return this
   }
 
+  _updateBalancedPoolStats () {
+    this[kGreatestCommonDivisor] = this[kClients].map(p => p[kWeight]).reduce(getGreatestCommonDivisor, 0)
+  }
+
   removeUpstream (upstream) {
     const upstreamOrigin = parseOrigin(upstream).origin
 
@@ -100,10 +145,42 @@ class BalancedPool extends PoolBase {
       return
     }
 
-    this[kClients].splice(this[kClients].indexOf(dispatcher), 1)
-    this[kClients].push(dispatcher)
+    const allClientsBusy = this[kClients].map(pool => pool[kNeedDrain]).reduce((a, b) => a && b, true)
+
+    if (allClientsBusy) {
+      return
+    }
+
+    let counter = 0
+
+    let maxWeightIndex = this[kClients].findIndex(pool => !pool[kNeedDrain])
+
+    while (counter++ < this[kClients].length) {
+      this[kIndex] = (this[kIndex] + 1) % this[kClients].length
+      const pool = this[kClients][this[kIndex]]
+
+      // find pool index with the largest weight
+      if (pool[kWeight] > this[kClients][maxWeightIndex][kWeight] && !pool[kNeedDrain]) {
+        maxWeightIndex = this[kIndex]
+      }
+
+      // decrease the current weight every `this[kClients].length`.
+      if (this[kIndex] === 0) {
+        // Set the current weight to the next lower weight.
+        this[kCurrentWeight] = this[kCurrentWeight] - this[kGreatestCommonDivisor]
+
+        if (this[kCurrentWeight] <= 0) {
+          this[kCurrentWeight] = this[kMaxWeightPerServer]
+        }
+      }
+      if (pool[kWeight] >= this[kCurrentWeight] && (!pool[kNeedDrain])) {
+        return pool
+      }
+    }
 
-    return dispatcher
+    this[kCurrentWeight] = this[kClients][maxWeightIndex][kWeight]
+    this[kIndex] = maxWeightIndex
+    return this[kClients][maxWeightIndex]
   }
 }
 

lib/core/request.js

@@ -7,6 +7,27 @@ const {
 const assert = require('assert')
 const util = require('./util')
 
+// tokenRegExp and headerCharRegex have been lifted from
+// https://github.com/nodejs/node/blob/main/lib/_http_common.js
+
+/**
+ * Verifies that the given val is a valid HTTP token
+ * per the rules defined in RFC 7230
+ * See https://tools.ietf.org/html/rfc7230#section-3.2.6
+ */
+const tokenRegExp = /^[\^_`a-zA-Z\-0-9!#$%&'*+.|~]+$/
+
+/**
+ * Matches if val contains an invalid field-vchar
+ *  field-value    = *( field-content / obs-fold )
+ *  field-content  = field-vchar [ 1*( SP / HTAB ) field-vchar ]
+ *  field-vchar    = VCHAR / obs-text
+ */
+const headerCharRegex = /[^\t\x20-\x7e\x80-\xff]/
+
+// Verifies that a given path is valid does not contain control chars \x00 to \x20
+const invalidPathRegex = /[^\u0021-\u00ff]/
+
 const kHandler = Symbol('handler')
 
 const channels = {}
@@ -54,10 +75,14 @@ class Request {
       method !== 'CONNECT'
     ) {
       throw new InvalidArgumentError('path must be an absolute URL or start with a slash')
+    } else if (invalidPathRegex.exec(path) !== null) {
+      throw new InvalidArgumentError('invalid request path')
     }
 
     if (typeof method !== 'string') {
       throw new InvalidArgumentError('method must be a string')
+    } else if (tokenRegExp.exec(method) === null) {
+      throw new InvalidArgumentError('invalid request method')
     }
 
     if (upgrade && typeof upgrade !== 'string') {
@@ -301,6 +326,10 @@ function processHeader (request, key, val) {
     key.toLowerCase() === 'expect'
   ) {
     throw new NotSupportedError('expect header not supported')
+  } else if (tokenRegExp.exec(key) === null) {
+    throw new InvalidArgumentError('invalid header key')
+  } else if (headerCharRegex.exec(val) !== null) {
+    throw new InvalidArgumentError(`invalid ${key} header`)
   } else {
     request.headers += `${key}: ${val}\r\n`
   }

lib/fetch/body.js

@@ -291,6 +291,10 @@ function bodyMixinMethods (instance) {
       const chunks = []
 
       for await (const chunk of consumeBody(this[kState].body)) {
+        if (!isUint8Array(chunk)) {
+          throw new TypeError('Expected Uint8Array chunk')
+        }
+
         // Assemble one final large blob with Uint8Array's can exhaust memory.
         // That's why we create create multiple blob's and using references
         chunks.push(new Blob([chunk]))
@@ -314,6 +318,10 @@ function bodyMixinMethods (instance) {
         let offset = 0
 
         for await (const chunk of consumeBody(this[kState].body)) {
+          if (!isUint8Array(chunk)) {
+            throw new TypeError('Expected Uint8Array chunk')
+          }
+
           buffer.set(chunk, offset)
           offset += chunk.length
         }
@@ -331,6 +339,10 @@ function bodyMixinMethods (instance) {
       let size = 0
 
       for await (const chunk of consumeBody(this[kState].body)) {
+        if (!isUint8Array(chunk)) {
+          throw new TypeError('Expected Uint8Array chunk')
+        }
+
         chunks.push(chunk)
         size += chunk.byteLength
       }
@@ -355,6 +367,10 @@ function bodyMixinMethods (instance) {
       const textDecoder = new TextDecoder()
 
       for await (const chunk of consumeBody(this[kState].body)) {
+        if (!isUint8Array(chunk)) {
+          throw new TypeError('Expected Uint8Array chunk')
+        }
+
         result += textDecoder.decode(chunk, { stream: true })
       }
 

lib/handler/redirect.js

@@ -186,7 +186,8 @@ function shouldRemoveHeader (header, removeContent, unknownOrigin) {
   return (
     (header.length === 4 && header.toString().toLowerCase() === 'host') ||
     (removeContent && header.toString().toLowerCase().indexOf('content-') === 0) ||
-    (unknownOrigin && header.length === 13 && header.toString().toLowerCase() === 'authorization')
+    (unknownOrigin && header.length === 13 && header.toString().toLowerCase() === 'authorization') ||
+    (unknownOrigin && header.length === 6 && header.toString().toLowerCase() === 'cookie')
   )
 }
 

lib/mock/mock-utils.js

@@ -8,7 +8,7 @@ const {
   kOrigin,
   kGetNetConnect
 } = require('./mock-symbols')
-const { buildURL } = require('../core/util')
+const { buildURL, nop } = require('../core/util')
 
 function matchValue (match, value) {
   if (typeof match === 'string') {
@@ -288,6 +288,7 @@ function mockDispatch (opts, handler) {
     const responseHeaders = generateKeyValues(headers)
     const responseTrailers = generateKeyValues(trailers)
 
+    handler.abort = nop
     handler.onHeaders(statusCode, responseHeaders, resume, getStatusText(statusCode))
     handler.onData(Buffer.from(responseData))
     handler.onComplete(responseTrailers)

package.json

@@ -1,6 +1,6 @@
 {
   "name": "undici",
-  "version": "5.6.1",
+  "version": "5.8.0",
   "description": "An HTTP/1.1 client, written from scratch for Node.js",
   "homepage": "https://undici.nodejs.org",
   "bugs": {