feat: parallel queries

.vscode/launch.json

@@ -0,0 +1,24 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "args": [
+                "--timeout",
+                "999999",
+                "--colors",
+                "${relativeFile}"
+            ],
+            "internalConsoleOptions": "openOnSessionStart",
+            "name": "Mocha Tests",
+            "program": "${workspaceFolder}/node_modules/mocha/bin/_mocha",
+            "request": "launch",
+            "skipFiles": [
+                "<node_internals>/**"
+            ],
+            "type": "node"
+        }
+    ]
+}

eslint.config.js

@@ -2,7 +2,7 @@ import js from '@eslint/js';
 import globals from 'globals';
 import jsdoc from 'eslint-plugin-jsdoc';
 
-/** @type {import('@types/eslint').Linter.FlatConfig[]} */
+/** @type {import('@types/eslint').Linter.Config[]} */
 export default [
   js.configs.recommended,
   {

packages/core/src/Coordinator.js

@@ -44,6 +44,7 @@ export class Coordinator {
     consolidate = true,
     indexes = {}
   } = {}) {
+    /** @type QueryManager */
     this.manager = manager;
     this.manager.cache(cache);
     this.manager.consolidate(consolidate);

packages/core/src/QueryManager.js

@@ -1,29 +1,59 @@
 import { consolidator } from './QueryConsolidator.js';
 import { lruCache, voidCache } from './util/cache.js';
 import { priorityQueue } from './util/priority-queue.js';
-import { QueryResult } from './util/query-result.js';
+import { QueryResult, QueryState } from './util/query-result.js';
 
-export const Priority = { High: 0, Normal: 1, Low: 2 };
+export const Priority = Object.freeze({ High: 0, Normal: 1, Low: 2 });
 
 export class QueryManager {
-  constructor() {
+  constructor(
+    maxConcurrentRequests = 32
+  ) {
     this.queue = priorityQueue(3);
     this.db = null;
     this.clientCache = null;
     this._logger = null;
     this._logQueries = false;
     this.recorders = [];
-    this.pending = null;
     this._consolidate = null;
+    /**
+     * Requests pending with the query manager.
+     * 
+     * @type QueryResult[]
+     */
+    this.pendingResults = [];
+    this.maxConcurrentRequests = maxConcurrentRequests;
   }
 
   next() {
-    if (this.pending || this.queue.isEmpty()) return;
+    if (this.queue.isEmpty() || this.pendingResults.length > this.maxConcurrentRequests) {
+      return;
+    }
+
     const { request, result } = this.queue.next();
-    this.pending = this.submit(request, result);
-    this.pending.finally(() => { this.pending = null; this.next(); });
+
+    this.pendingResults.push(result);
+    this.submit(request, result).finally(() => {
+      // return from the queue all the prepared requests
+      while (this.pendingResults.length && this.pendingResults[0].state !== QueryState.pending) {
+        const result = this.pendingResults.shift();
+        if (result.state === QueryState.prepared) {
+          result.fulfill();
+        } else if (result.state === QueryState.done) {
+          console.warn('Found resolved query in pending results.');
+        }
+      }
+      this.next();
+    });
   }
 
+  /**
+   * Add an entry to the query queue with a priority.
+   * @param {object} entry The entry to add.
+   * @param {*} [entry.request] The query request.
+   * @param {QueryResult} [entry.result] The query result.
+   * @param {number} priority The query priority, defaults to `Priority.Normal`.
+   */
   enqueue(entry, priority = Priority.Normal) {
     this.queue.insert(entry, priority);
     this.next();
@@ -35,6 +65,11 @@ export class QueryManager {
     }
   }
 
+  /**
+   * Submit the query to the connector.
+   * @param {*} request The request.
+   * @param {QueryResult} result The query result.
+   */
   async submit(request, result) {
     try {
       const { query, type, cache = false, record = true, options } = request;
@@ -49,8 +84,9 @@ export class QueryManager {
       if (cache) {
         const cached = this.clientCache.get(sql);
         if (cached) {
+          const data = await cached;
           this._logger.debug('Cache');
-          result.fulfill(cached);
+          result.prepare(type === 'exec' ? null : data);
           return;
         }
       }
@@ -60,10 +96,16 @@ export class QueryManager {
       if (this._logQueries) {
         this._logger.debug('Query', { type, sql, ...options });
       }
-      const data = await this.db.query({ type, sql, ...options });
+
+      const promise = this.db.query({ type, sql, ...options });
+      if (cache) this.clientCache.set(sql, promise);
+
+      const data = await promise;
+
       if (cache) this.clientCache.set(sql, data);
+
       this._logger.debug(`Request: ${(performance.now() - t0).toFixed(1)}`);
-      result.fulfill(data);
+      result.prepare(type === 'exec' ? null : data);
     } catch (err) {
       result.reject(err);
     }
@@ -95,6 +137,12 @@ export class QueryManager {
     }
   }
 
+  /**
+   * Request a query result.
+   * @param {*} request The request.
+   * @param {number} priority The query priority, defaults to `Priority.Normal`.
+   * @returns {QueryResult} A query result promise.
+   */
   request(request, priority = Priority.Normal) {
     const result = new QueryResult();
     const entry = { request, result };
@@ -116,6 +164,12 @@ export class QueryManager {
         }
         return false;
       });
+
+      for (const result of this.pendingResults) {
+        if (set.has(result)) {
+          result.reject('Canceled');
+        }
+      }
     }
   }
 
@@ -124,6 +178,11 @@ export class QueryManager {
       result.reject('Cleared');
       return true;
     });
+
+    for (const result of this.pendingResults) {
+      result.reject('Cleared');
+    }
+    this.pendingResults = [];
   }
 
   record() {

packages/core/src/util/query-result.js

@@ -1,3 +1,10 @@
+export const QueryState = Object.freeze({
+  pending: Symbol('pending'),
+  prepared: Symbol('prepared'),
+  error: Symbol('error'),
+  done: Symbol('done')
+});
+
 /**
  * A query result Promise that can allows external callers
  * to resolve or reject the Promise.
@@ -15,15 +22,41 @@ export class QueryResult extends Promise {
     });
     this._resolve = resolve;
     this._reject = reject;
+    this._state = QueryState.pending;
+    this._value = undefined;
   }
 
   /**
-   * Resolve the result Promise with the provided value.
+   * Resolve the result Promise with a prepared value or the provided value.
+   * This method will only succeed if either a value is provided or prepared.
    * @param {*} value The result value.
    * @returns {this}
    */
   fulfill(value) {
-    this._resolve(value);
+    if (this._value !== undefined) {
+      if (value !== undefined) {
+        throw Error('Promise has prepared and provided value');
+      }
+      this._resolve(this._value);
+    } else if (value === undefined) {
+      throw Error('Promise has has neither prepared nor provided value');
+    } else {
+      this._resolve(value);
+    }
+
+    this._state = QueryState.done;
+
+    return this;
+  }
+
+  /**
+   * Prepare to resolve the result Promise with the provided value.
+   * @param {*} value The result value.
+   * @returns {this}
+   */
+  prepare(value) {
+    this._state = QueryState.prepared;
+    this._value = value;
     return this;
   }
 
@@ -33,9 +66,18 @@ export class QueryResult extends Promise {
    * @returns {this}
    */
   reject(error) {
+    this._state = QueryState.error;
     this._reject(error);
     return this;
   }
+
+  /**
+   * Returns the state of this query result.
+   * @returns {symbol}
+   */
+  get state() {
+    return this._state;
+  }
 }
 
 // necessary to make Promise subclass act like a Promise

packages/core/test/coordinator-test.js

@@ -1,8 +1,13 @@
-import assert from 'node:assert';
-import { Coordinator, coordinator } from '../src/index.js';
+import assert from "node:assert";
+import { coordinator, Coordinator } from "../src/index.js";
+import { QueryResult, QueryState } from "../src/util/query-result.js";
 
-describe('coordinator', () => {
-  it('has accessible singleton', () => {
+async function wait() {
+  return new Promise(setTimeout);
+}
+
+describe("coordinator", () => {
+  it("has accessible singleton", () => {
     const mc = coordinator();
     assert.ok(mc instanceof Coordinator);
 
@@ -11,4 +16,70 @@ describe('coordinator', () => {
 
     assert.strictEqual(mc2, coordinator());
   });
+
+  it("query results returned in correct order", async () => {
+    const promises = [];
+
+    const connector = {
+      async query() {
+        const promise = new QueryResult();
+        promises.push(promise);
+        return promise;
+      },
+    };
+
+    const coord = new Coordinator(connector);
+
+    const r0 = coord.query("SELECT 0");
+    const r1 = coord.query("SELECT 1");
+    const r2 = coord.query("SELECT 2");
+    const r3 = coord.query("SELECT 3");
+
+    // queries have not been sent yet
+    assert.equal(promises.length, 0);
+
+    await wait();
+
+    // all queries should have been sent to the connector
+    assert.equal(promises.length, 4);
+    assert.equal(coord.manager.pendingResults.length, 4);
+
+    // resolve promises in reverse order
+
+    promises.at(3).fulfill(0);
+    await wait();
+
+    assert.equal(r0.state, QueryState.pending);
+    assert.equal(r1.state, QueryState.pending);
+    assert.equal(r2.state, QueryState.pending);
+    assert.equal(r3.state, QueryState.prepared);
+
+    promises.at(1).fulfill(0);
+    await wait();
+
+    assert.equal(r0.state, QueryState.pending);
+    assert.equal(r1.state, QueryState.prepared);
+    assert.equal(r2.state, QueryState.pending);
+    assert.equal(r3.state, QueryState.prepared);
+
+    promises.at(0).fulfill(0);
+    await wait();
+
+    assert.equal(coord.manager.pendingResults.length, 2);
+
+    assert.equal(r0.state, QueryState.done);
+    assert.equal(r1.state, QueryState.done);
+    assert.equal(r2.state, QueryState.pending);
+    assert.equal(r3.state, QueryState.prepared);
+
+    promises.at(2).fulfill(0);
+    await wait();
+
+    assert.equal(coord.manager.pendingResults.length, 0);
+
+    assert.equal(r0.state, QueryState.done);
+    assert.equal(r1.state, QueryState.done);
+    assert.equal(r2.state, QueryState.done);
+    assert.equal(r3.state, QueryState.done);
+  });
 });