Cap concurrent requests to get Neptune schema (#58)

Previously, for a graph with `n` nodes or edges, there would be `n`
concurrent requests made to Neptune via `queryNeptune`. Depending on
your instance class or VPC settings, you could hit errors such as
`MemoryLimitExceededException` or connection limits/sockets abruptly
closing, with increasing likelihood as the size of the graph increased.
Initially observed abrupt socket closing just like
#20.

Now, the concurrent request Promises are resolved in batches using
`mapAll`. For now, the batch size is hardcoded to 20. For very small
instance classes (e.g. from the "Development and testing" template),
`MemoryLimitExceededException`s are still likely.

Further improvements could include:
- relating the batch size to the instance class
- allowing the user to specify the batch size in the process args
- exponential/dynamic backoff on transient errors as indicated in
  https://docs.aws.amazon.com/neptune/latest/userguide/errors-engine-codes.html#errors-query

Author: danielfinke

src/NeptuneSchema.js

@@ -15,6 +15,7 @@ import { aws4Interceptor } from "aws4-axios";
 import { fromNodeProviderChain  } from "@aws-sdk/credential-providers";
 import { NeptunedataClient, ExecuteOpenCypherQueryCommand } from "@aws-sdk/client-neptunedata";
 import { loggerDebug, loggerError, loggerInfo, yellow } from "./logger.js";
+import { mapAll } from "./util-promise.js";
 import { ExecuteQueryCommand, NeptuneGraphClient } from "@aws-sdk/client-neptune-graph";
 
 const NEPTUNE_DB = 'neptune-db';
@@ -191,7 +192,7 @@ async function findFromAndToLabels(edgeStructure) {
 
 
 async function getEdgesDirections() {
-    await Promise.all(schema.edgeStructures.map(findFromAndToLabels))
+    await mapAll(schema.edgeStructures, findFromAndToLabels);
 }
 
 
@@ -254,9 +255,7 @@ async function getEdgeProperties(edge) {
 
 async function getEdgesProperties() {
     loggerInfo('Retrieving edge properties');
-    await Promise.all(schema.edgeStructures.map(async (edge) => {
-        await getEdgeProperties(edge);
-      }));
+    await mapAll(schema.edgeStructures, getEdgeProperties);
 }
 
 
@@ -281,9 +280,7 @@ async function getNodeProperties(node) {
 
 async function getNodesProperties() {
     loggerInfo('Retrieving node properties');
-    await Promise.all(schema.nodeStructures.map(async (node) => {
-        await getNodeProperties(node);
-      }));
+    await mapAll(schema.nodeStructures, getNodeProperties);
 }
 
 
@@ -323,9 +320,7 @@ async function getEdgesDirectionsCardinality() {
         }
     }
 
-    await Promise.all(possibleDirections.map(async (d) => {
-        await checkEdgeDirectionCardinality(d);     
-    }));    
+    await mapAll(possibleDirections, checkEdgeDirectionCardinality);
 }
 
 

src/test/util-promise.test.js

@@ -0,0 +1,62 @@
+import { mapAll } from '../util-promise.js'
+
+describe('mapAll', () => {
+    test('should map each element, resolve each Promise, and resolve to the mapped values', async () => {
+        await expect(mapAll([1, 2, 3, 4, 5, 6, 7, 8], double))
+            .resolves
+            .toEqual([2, 4, 6, 8, 10, 12, 14, 16]);
+    });
+
+    test('should map each element in batches', async () => {
+        const count = 100;
+        const batchSize = 50;
+        const batchCounter = makeBatchCounter(count);
+
+        await expect(mapAll(Array(count + 1).fill(0), (i) => batchCounter.fn(i), batchSize))
+            .rejects
+            .toThrow(Error);
+        expect(batchCounter.count).toEqual(count);
+    });
+
+    test('should reject with the same error as a rejected mapped Promise', async () => {
+        await expect(mapAll([2, 4, 6, 3, 8, 10], assertEven))
+            .rejects
+            .toThrow(OddNumberError);
+    });
+
+    test('should resolve to an empty array for empty input arrays', async () => {
+        await expect(mapAll([], Promise.resolve)).resolves.toEqual([]);
+    });
+
+    test('should resolve to an empty array for non-positive batch sizes', async () => {
+        await expect(mapAll([1, 2, 3], Promise.resolve, -1)).resolves.toEqual([]);
+        await expect(mapAll([1, 2, 3], Promise.resolve, 0)).resolves.toEqual([]);
+    });
+
+    function double(i) {
+        return Promise.resolve(i * 2);
+    }
+
+    function makeBatchCounter(n) {
+        return {
+            count: 0,
+            fn: function (i) {
+                if (this.count >= n) {
+                    throw new Error();
+                }
+                this.count++;
+                return Promise.resolve(i);
+            }
+        };
+    }
+
+    function assertEven(i) {
+        if (i % 2 !== 0) {
+            throw new OddNumberError();
+        }
+
+        return Promise.resolve(i);
+    }
+
+    class OddNumberError extends Error {}
+});

src/util-promise.js

@@ -0,0 +1,44 @@
+const BATCH_SIZE = 20;
+
+/**
+ * @callback MapCallback
+ * @param {T} value
+ * @param {number} index
+ * @param {T[]} array
+ * @returns {U}
+ * @template T, U
+ */
+
+/**
+ * Calls a Promise-returning callback function on each element of an array and creates a Promise
+ * that is resolved with an array of results when all of the returned Promises resolve, or rejected
+ * when any Promise is rejected.
+ *
+ * The elements of the array are mapped and then resolved in batches of size `batchSize`, ensuring
+ * that no more than `batchSize` Promises are pending at one time. This is useful for e.g. avoiding
+ * too many simultaneous web requests.
+ *
+ * @param {T[]} array an array to map over
+ * @param {MapCallback<T, Promise<U>>} callbackfn a function that accepts up to three arguments. The
+ * mapAll function calls the callbackfn function one time for each element in the array
+ * @param {number} batchSize the number of Promises to concurrently resolve
+ * @returns {Promise<U[]>}
+ * @template T, U
+ */
+async function mapAll(array, callbackfn, batchSize = BATCH_SIZE) {
+    if (batchSize <= 0) {
+        return [];
+    }
+
+    const results = [];
+
+    for (let i = 0; i < array.length; i += batchSize) {
+        const promises = array.slice(i, i + batchSize).map(callbackfn);
+        const batchResults = await Promise.all(promises);
+        results.push(...batchResults);
+    }
+
+    return results;
+}
+
+export { mapAll };