Merge branch 'main' into unlink

digidem · Jan 19, 2024 · 07ef013 · 07ef013
2 parents 4d8c94a + b85c3b6
commit 07ef013
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 8 deletions.
diff --git a/.github/workflows/node.js.yml b/.github/workflows/node.js.yml
@@ -14,8 +14,7 @@ jobs:
 
     strategy:
       matrix:
-        node-version: [16.17.x, 18.x, 20.x]
-        # See supported Node.js release schedule at https://nodejs.org/en/about/releases/
+        node-version: [16.17.1, 18.17.1, 20.x]
 
     steps:
       - uses: actions/checkout@v2

diff --git a/lib/multi-core-index-stream.js b/lib/multi-core-index-stream.js
@@ -188,7 +188,10 @@ class MultiCoreIndexStream extends Readable {
   #handleDrained() {
     let drained = true
     for (const stream of this.#streams.keys()) {
-      if (!stream.drained) drained = false
+      if (!stream.drained) {
+        drained = false
+        break
+      }
     }
     if (drained === this.#drained && !drained) return
     this.#drained = drained

diff --git a/test/helpers/index.js b/test/helpers/index.js
@@ -6,6 +6,7 @@ const ram = require('random-access-memory')
 const BLOCK_LENGTH = Buffer.from('block000000').byteLength
 
 /** @typedef {import('../../lib/types').Entry<'binary'>} Entry */
+/** @typedef {import('node:events').EventEmitter} EventEmitter */
 
 module.exports = {
   create,
@@ -14,6 +15,7 @@ module.exports = {
   generateFixtures,
   createMultiple,
   throttledDrain,
+  throttledIdle,
   sortEntries,
   logEntries,
   blocksToExpected,
@@ -87,24 +89,37 @@ async function generateFixtures(cores, count) {
  * The index stream can become momentarily drained between reads and
  * appends/downloads of new data. This throttle drained will resolve only when
  * the stream has remained drained for > 10ms
- * @param {import('events').EventEmitter} emitter
+ * @param {EventEmitter} emitter
  * @returns {Promise<void>}
  */
-async function throttledDrain(emitter) {
+function throttledDrain(emitter) {
+  return throttledStreamEvent(emitter, 'drained')
+}
+
+function throttledIdle(emitter) {
+  return throttledStreamEvent(emitter, 'idle')
+}
+
+/**
+ * @param {EventEmitter} emitter
+ * @param {string} eventName
+ * @returns {Promise<void>}
+ */
+function throttledStreamEvent(emitter, eventName) {
   return new Promise((resolve) => {
     /** @type {ReturnType<setTimeout>} */
     let timeoutId
 
-    function onDrained() {
+    function onEvent() {
       clearTimeout(timeoutId)
       timeoutId = setTimeout(() => {
-        emitter.off('drained', onDrained)
+        emitter.off(eventName, onEvent)
         emitter.off('indexing', onIndexing)
         resolve()
       }, 10)
     }
 
-    emitter.on('drained', onDrained)
+    emitter.on(eventName, onEvent)
     emitter.on('indexing', onIndexing)
     function onIndexing() {
       clearTimeout(timeoutId)