fix(npmjs): follower hangs when doing a backfill (#1723)

(Re-roll of #1721 which was accidentally merged; in the previous version
I still did a `throw` from a non-`new Promise()` callback, which made
the exception uncaught. Solve in this one by more strictly splitting out
the Promise conversion to helper functions)

For reasons of the NPMJS replica behaving strangely[1], our follower has
changed to starting a full backfill of the entire catalog.

However, when doing the full backfill we're seeing strange behavior:
when doing fetches for a lot of packages in parallel, we notice that the
follower stops printing anything and the Lambda just times out after 5
minutes of inactivity. Looking at the logs, 47 requests are sent out in
parallel but only 15 are answered, and then nothing more happens.

Looking at the code it is a mess of NodeJS event handling, and from what
I can tell there is no handling of socket error events that occur in the
request (just socket error events that occur in the response). Best
theory: since these events are dropped on the floor, the corresponding
Promise is never resolved or rejected and the follower will just hang.
There also was no request timeout, I added that as well.

I refactored the code to be more straightforward:

- No recursion to do retries but a `while` loop
- Use a deadline-driven timeout (previous code would retry indefinitely)
- Put the classification of what makes an error retryable in once place
- Introduce helper functions to handle the conversion between callback
style and `async/await`.
- Importantly: add `req.on('error')` and `req.on('timeout')` to handle
problems with the request.
- Replace `x = new Error(); Error.captureStackTrace(x); ko(x)` with
`throw new Error()`, since we can now do that.

[1]:

https://github.com/orgs/community/discussions/152515#discussioncomment-13053455
----

*By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache-2.0 license*

---------

Signed-off-by: github-actions <github-actions@github.com>
Co-authored-by: github-actions <github-actions@github.com>

Author: rix0rrr

src/__tests__/__snapshots__/construct-hub.test.ts.snap

@@ -1,5 +1,5 @@
 import { EventEmitter } from 'events';
-import { OutgoingHttpHeaders } from 'http';
+import { IncomingMessage, OutgoingHttpHeaders } from 'http';
 import { Agent, request, RequestOptions } from 'https';
 import { Readable } from 'stream';
 import { URL } from 'url';
@@ -155,53 +155,30 @@ export class CouchChanges extends EventEmitter {
     let maxDelay = 100;
     while (true) {
       try {
-        return await new Promise((ok, ko) => {
-          const req = request(url, requestOptions, (res) => {
-            if (res.statusCode == null) {
-              throw new RetryableError('No status code available');
-            }
-
-            // Server errors. We can't know whether these are really retryable but we usually pretend that they are.
-            if (res.statusCode >= 500 && res.statusCode < 600) {
-              throw new RetryableError(
-                `HTTP ${res.statusCode} ${res.statusMessage}`
-              );
-            }
-
-            // Permanent (client) errors:
-            if (res.statusCode >= 400 && res.statusCode < 500) {
-              throw new Error(`HTTP ${res.statusCode} ${res.statusMessage}`);
-            }
-
-            console.log(
-              `Response: ${method.toUpperCase()} ${url} => HTTP ${
-                res.statusCode
-              } (${res.statusMessage})`
-            );
-
-            res.once('error', ko);
-
-            const json = JSONStream.parse(true);
-            json.once('data', ok);
-            json.once('error', ko);
-
-            const plainPayload =
-              res.headers['content-encoding'] === 'gzip' ? gunzip(res) : res;
-            plainPayload.pipe(json, { end: true });
-            plainPayload.once('error', ko);
-          });
-
-          req.on('error', ko);
-          req.on('timeout', () => {
-            req.destroy(
-              new RetryableError(
-                `Timeout after ${REQUEST_ATTEMPT_TIMEOUT_MS}ms, aborting request`
-              )
-            );
-          });
-
-          req.end(body && JSON.stringify(body, null, 2));
-        });
+        const res = await requestPromise(url, requestOptions, body);
+        if (res.statusCode == null) {
+          throw new RetryableError('No status code available');
+        }
+
+        // Server errors. We can't know whether these are really retryable but we usually pretend that they are.
+        if (res.statusCode >= 500 && res.statusCode < 600) {
+          throw new RetryableError(
+            `HTTP ${res.statusCode} ${res.statusMessage}`
+          );
+        }
+
+        // Permanent (client) errors:
+        if (res.statusCode >= 400 && res.statusCode < 500) {
+          throw new Error(`HTTP ${res.statusCode} ${res.statusMessage}`);
+        }
+
+        console.log(
+          `Response: ${method.toUpperCase()} ${url} => HTTP ${
+            res.statusCode
+          } (${res.statusMessage})`
+        );
+
+        return await readResponseJson(res);
       } catch (e: any) {
         if (Date.now() > deadline || !isRetryableError(e)) {
           throw e;
@@ -216,6 +193,45 @@ export class CouchChanges extends EventEmitter {
   }
 }
 
+/**
+ * A Promisified version of `https.request()` that also handles timeout events
+ */
+function requestPromise(
+  url: URL,
+  options: RequestOptions,
+  body?: Record<string, unknown>
+) {
+  return new Promise<IncomingMessage>((ok, ko) => {
+    const req = request(url, options ?? {}, ok);
+    req.on('error', ko);
+    req.on('timeout', () => {
+      req.destroy(
+        new RetryableError(
+          `Timeout after ${REQUEST_ATTEMPT_TIMEOUT_MS}ms, aborting request`
+        )
+      );
+    });
+    req.end(body && JSON.stringify(body, null, 2));
+  });
+}
+
+function readResponseJson(
+  res: IncomingMessage
+): Promise<Record<string, unknown>> {
+  return new Promise((ok, ko) => {
+    res.once('error', ko);
+
+    const json = JSONStream.parse(true);
+    json.once('data', ok);
+    json.once('error', ko);
+
+    const plainPayload =
+      res.headers['content-encoding'] === 'gzip' ? gunzip(res) : res;
+    plainPayload.pipe(json, { end: true });
+    plainPayload.once('error', ko);
+  });
+}
+
 class RetryableError extends Error {}
 
 function isRetryableError(e: Error): boolean {