Prep 8.12.2 for release (#2142)

* Backport changelog for 8.12.1

* Add changelog for 8.12.2

docs/changelog.asciidoc

@@ -1,6 +1,28 @@
 [[changelog-client]]
 == Release notes
 
+[discrete]
+=== 8.12.2
+
+[discrete]
+==== Fixes
+
+[discrete]
+===== Upgrade transport to 8.4.1 https://github.com/elastic/elasticsearch-js/pull/2137[#2137]
+
+Upgrades `@elastic/transport` to 8.4.1 to resolve https://github.com/elastic/elastic-transport-js/pull/83[a bug] where arrays in error diagnostics were unintentionally transformed into objects.
+
+[discrete]
+=== 8.12.1
+
+[discrete]
+==== Fixes
+
+[discrete]
+===== Fix hang in bulk helper semaphore https://github.com/elastic/elasticsearch-js/pull/2027[#2027]
+
+The failing state could be reached when a server's response times are slower than flushInterval.
+
 [discrete]
 === 8.12.0
 

package.json

@@ -1,7 +1,7 @@
 {
   "name": "@elastic/elasticsearch",
-  "version": "8.12.0",
-  "versionCanary": "8.12.0-canary.0",
+  "version": "8.12.2",
+  "versionCanary": "8.12.2-canary.0",
   "description": "The official Elasticsearch client for Node.js",
   "main": "index.js",
   "types": "index.d.ts",
@@ -83,7 +83,7 @@
     "zx": "^7.2.2"
   },
   "dependencies": {
-    "@elastic/transport": "^8.4.0",
+    "@elastic/transport": "^8.4.1",
     "tslib": "^2.4.0"
   },
   "tap": {
@@ -93,4 +93,4 @@
     "coverage": false,
     "check-coverage": false
   }
-}
+}

src/helpers.ts

@@ -624,7 +624,7 @@ export default class Helpers {
       let chunkBytes = 0
       timeoutRef = setTimeout(onFlushTimeout, flushInterval) // eslint-disable-line
 
-      // @ts-expect-error datasoruce is an iterable
+      // @ts-expect-error datasource is an iterable
       for await (const chunk of datasource) {
         if (shouldAbort) break
         timeoutRef.refresh()
@@ -656,15 +656,16 @@ export default class Helpers {
 
         if (chunkBytes >= flushBytes) {
           stats.bytes += chunkBytes
-          const send = await semaphore()
-          send(bulkBody.slice())
+          const bulkBodyCopy = bulkBody.slice()
           bulkBody.length = 0
           chunkBytes = 0
+          const send = await semaphore()
+          send(bulkBodyCopy)
         }
       }
 
       clearTimeout(timeoutRef)
-      // In some cases the previos http call does not have finished,
+      // In some cases the previous http call has not finished,
       // or we didn't reach the flush bytes threshold, so we force one last operation.
       if (!shouldAbort && chunkBytes > 0) {
         const send = await semaphore()
@@ -708,8 +709,8 @@ export default class Helpers {
     // to guarantee that no more than the number of operations
     // allowed to run at the same time are executed.
     // It returns a semaphore function which resolves in the next tick
-    // if we didn't reach the maximim concurrency yet, otherwise it returns
-    // a promise that resolves as soon as one of the running request has finshed.
+    // if we didn't reach the maximum concurrency yet, otherwise it returns
+    // a promise that resolves as soon as one of the running requests has finished.
     // The semaphore function resolves a send function, which will be used
     // to send the actual bulk request.
     // It also returns a finish function, which returns a promise that is resolved

test/unit/helpers/bulk.test.ts

@@ -23,9 +23,11 @@ import { createReadStream } from 'fs'
 import * as http from 'http'
 import { join } from 'path'
 import split from 'split2'
+import { Readable } from 'stream'
 import { test } from 'tap'
 import { Client, errors } from '../../../'
 import { buildServer, connection } from '../../utils'
+const { sleep } = require('../../integration/helper')
 
 let clientVersion: string = require('../../../package.json').version // eslint-disable-line
 if (clientVersion.includes('-')) {
@@ -1594,3 +1596,150 @@ test('Flush interval', t => {
 
   t.end()
 })
+
+test(`flush timeout does not lock process when flushInterval is less than server timeout`, async t => {
+  const flushInterval = 500
+
+  async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
+    setTimeout(() => {
+      res.writeHead(200, { 'content-type': 'application/json' })
+      res.end(JSON.stringify({ errors: false, items: [{}] }))
+    }, 1000)
+  }
+
+  const [{ port }, server] = await buildServer(handler)
+  const client = new Client({ node: `http://localhost:${port}` })
+
+  async function * generator () {
+    const data = dataset.slice()
+    for (const doc of data) {
+      await sleep(flushInterval)
+      yield doc
+    }
+  }
+
+  const result = await client.helpers.bulk({
+    datasource: Readable.from(generator()),
+    flushBytes: 1,
+    flushInterval: flushInterval,
+    concurrency: 1,
+    onDocument (_) {
+      return {
+        index: { _index: 'test' }
+      }
+    },
+    onDrop (_) {
+      t.fail('This should never be called')
+    }
+  })
+
+  t.type(result.time, 'number')
+  t.type(result.bytes, 'number')
+  t.match(result, {
+    total: 3,
+    successful: 3,
+    retry: 0,
+    failed: 0,
+    aborted: false
+  })
+
+  server.stop()
+})
+
+test(`flush timeout does not lock process when flushInterval is greater than server timeout`, async t => {
+  const flushInterval = 500
+
+  async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
+    setTimeout(() => {
+      res.writeHead(200, { 'content-type': 'application/json' })
+      res.end(JSON.stringify({ errors: false, items: [{}] }))
+    }, 250)
+  }
+
+  const [{ port }, server] = await buildServer(handler)
+  const client = new Client({ node: `http://localhost:${port}` })
+
+  async function * generator () {
+    const data = dataset.slice()
+    for (const doc of data) {
+      await sleep(flushInterval)
+      yield doc
+    }
+  }
+
+  const result = await client.helpers.bulk({
+    datasource: Readable.from(generator()),
+    flushBytes: 1,
+    flushInterval: flushInterval,
+    concurrency: 1,
+    onDocument (_) {
+      return {
+        index: { _index: 'test' }
+      }
+    },
+    onDrop (_) {
+      t.fail('This should never be called')
+    }
+  })
+
+  t.type(result.time, 'number')
+  t.type(result.bytes, 'number')
+  t.match(result, {
+    total: 3,
+    successful: 3,
+    retry: 0,
+    failed: 0,
+    aborted: false
+  })
+
+  server.stop()
+})
+
+test(`flush timeout does not lock process when flushInterval is equal to server timeout`, async t => {
+  const flushInterval = 500
+
+  async function handler (req: http.IncomingMessage, res: http.ServerResponse) {
+    setTimeout(() => {
+      res.writeHead(200, { 'content-type': 'application/json' })
+      res.end(JSON.stringify({ errors: false, items: [{}] }))
+    }, flushInterval)
+  }
+
+  const [{ port }, server] = await buildServer(handler)
+  const client = new Client({ node: `http://localhost:${port}` })
+
+  async function * generator () {
+    const data = dataset.slice()
+    for (const doc of data) {
+      await sleep(flushInterval)
+      yield doc
+    }
+  }
+
+  const result = await client.helpers.bulk({
+    datasource: Readable.from(generator()),
+    flushBytes: 1,
+    flushInterval: flushInterval,
+    concurrency: 1,
+    onDocument (_) {
+      return {
+        index: { _index: 'test' }
+      }
+    },
+    onDrop (_) {
+      t.fail('This should never be called')
+    }
+  })
+
+  t.type(result.time, 'number')
+  t.type(result.bytes, 'number')
+  t.match(result, {
+    total: 3,
+    successful: 3,
+    retry: 0,
+    failed: 0,
+    aborted: false
+  })
+
+  server.stop()
+})