Stop RateLimiter from registering a listener per queued task

Summary:
## Why

Every task that `RateLimiter.enqueueRun` had to queue parked itself on a fresh `'run'` listener
registered on the limiter's private `TypedEventEmitter`, and nothing ever removed it -- there was no
`.off()` anywhere in `RateLimiter.ts`. The listener set therefore grew with every task the limiter
had *ever* queued and only died with the limiter itself.

Three instances leak:

| Site | Concurrency | Scope |
| --- | --- | --- |
| `isl-server/src/Repository.ts` `configRateLimiter` | 1 | per `Repository` |
| `isl-server/src/Repository.ts` `catLimiter` | 4 | per `Repository` |
| `isl-server/src/facebook/phabricator/queryGraphql.ts` | 7 | module-level, process-wide singleton |

The visible symptom on every isl-server spawn, on both macOS and Linux, in production:

```
MaxListenersExceededWarning: Possible EventTarget memory leak detected.
11 run listeners added to EventTarget. MaxListeners is 10.
```

It says `EventTarget` rather than `EventEmitter` because `TypedEventEmitter` wraps a plain
`EventTarget`.

The warning badly under-reports the problem. Node latches a warned-once flag per
`(target, eventType)` pair, so it fires exactly once per process no matter how far past 11 the count
climbs -- the log can never show the growth. And memory is not the only cost: every stale listener
is still invoked on each subsequent `emit('run')`, so dequeuing one task is O(tasks ever queued) and
a long session is O(n^2).

Calling `off` in a `finally` would fix the leak but leave the warning reachable, because the
listener set would still be one live `EventTarget` listener per *currently waiting* task. That is
enough to re-emit the byte-identical warning on ordinary backpressure:
`vscode/extension/facebook/comments/LineRealigner.ts` issues one `repo.cat` per unique commented
`(hash, path)` inside a single `Promise.all`, into the per-`Repository` `catLimiter` whose
concurrency is `MAX_SIMULTANEOUS_CAT_CALLS = 4`, so a diff with more than 14 distinct commented
pairs trips it; the process-wide GraphQL limiter (concurrency 7) does the same at 18 concurrent
queries. Such a warning would be a false positive on transient backpressure, indistinguishable in a
log from the leak this diff exists to eliminate, and would cost the next engineer the same
investigation over again.

## What

`RateLimiter` no longer uses `TypedEventEmitter` at all; the import and the `runs` field are gone.
`queued` changes from `Array<Id>` to `Array<{id, allowedToRun}>`, where `allowedToRun` is that
waiter's own `Deferred`, and `run` resolves it directly instead of broadcasting `emit('run', id)`
for every waiter to filter on by id.

This makes the `EventTarget` flavor of `MaxListenersExceededWarning` structurally impossible rather
than merely less likely: it can only be raised from `addEventListener`, and the limiter now never
calls it, at any concurrency, for any number of waiters. (Node raises the same warning text from
`EventEmitter.addListener` as well, so parking waiters on a `node:events` emitter would reintroduce
it -- this removes the reachable path, not the warning from the runtime.) Handing out a turn also stops walking the
whole waiter set, and the `TypedEventEmitter` listener-to-wrapper `Map` that was leaking alongside
the `EventTarget` is gone with it.

The deferred is created before the task is pushed onto `queued`, so `run` can hand out a turn
whether or not `enqueueRun` has reached its `await` yet. Correctness therefore no longer rests on
the "`queued` non-empty implies `running` is at max" ordering argument that the emitter version
needed to guarantee a waiter had subscribed before its own wake-up fired.

The public API, the concurrency limits, the FIFO ordering and the `log` callback strings are all
unchanged. The one timing difference is that `now allowing ID:N to run` is logged by the resumed
waiter rather than from inside the dequeue, one microtask later. `setMaxListeners` was deliberately
*not* used, since that only silences the warning.

Reviewed By: evangrayk

Differential Revision: D115241638

fbshipit-source-id: 098ba01734776ce092c98ed17eebae602508500b

Author: mshamis-meta

addons/shared/RateLimiter.ts

@@ -5,10 +5,18 @@
  * LICENSE file in the root directory of this source tree.
  */
 
-import {TypedEventEmitter} from './TypedEventEmitter';
+import type {Deferred} from './utils';
+
+import {defer} from './utils';
 
 type Id = number;
 
+type QueuedTask = {
+  id: Id;
+  /** Resolved by the limiter to hand this task its turn. */
+  allowedToRun: Deferred<void>;
+};
+
 /**
  * Rate limits requests to run an arbitrary task.
  * Up to `maxSimultaneousRunning` tasks can run at once,
@@ -23,9 +31,8 @@ type Id = number;
  * ```
  */
 export class RateLimiter {
-  private queued: Array<Id> = [];
+  private queued: Array<QueuedTask> = [];
   private running: Array<Id> = [];
-  private runs = new TypedEventEmitter<'run', Id>();
 
   constructor(
     private maxSimultaneousRunning: number,
@@ -39,20 +46,17 @@ export class RateLimiter {
 
   async enqueueRun<T>(runner: () => Promise<T>): Promise<T> {
     const id = this.generateId();
+    // Created before the task is queued so that `run` can always hand out the turn, whether or not
+    // this function has reached the `await` below by the time the turn is granted.
+    const task: QueuedTask = {id, allowedToRun: defer<void>()};
 
-    this.queued.push(id);
+    this.queued.push(task);
     this.tryDequeueNext();
 
     if (!this.running.includes(id)) {
       this.log?.(`${this.running.length} tasks are already running, enqueuing ID:${id}`);
-      await new Promise(res => {
-        this.runs.on('run', ran => {
-          if (ran === id) {
-            this.log?.(`now allowing ID:${id} to run`);
-            res(undefined);
-          }
-        });
-      });
+      await task.allowedToRun.promise;
+      this.log?.(`now allowing ID:${id} to run`);
     }
 
     try {
@@ -76,8 +80,8 @@ export class RateLimiter {
     }
   }
 
-  private run(id: Id) {
-    this.running.push(id);
-    this.runs.emit('run', id);
+  private run(task: QueuedTask) {
+    this.running.push(task.id);
+    task.allowedToRun.resolve(undefined);
   }
 }

addons/shared/__tests__/RateLimiter.test.ts

@@ -9,6 +9,17 @@ import {RateLimiter} from '../RateLimiter';
 import {nextTick} from '../testUtils';
 import {defer} from '../utils';
 
+/**
+ * Number of tasks parked in `RateLimiter`'s queue waiting for a turn. Each one owns a `Deferred`
+ * the limiter still has to resolve, so this array is the only place a waiter can accumulate. It is
+ * reachable only through private fields, but asserting on it is the only way to prove queued tasks
+ * don't leak.
+ */
+function waitingTaskCount(rateLimiter: RateLimiter): number {
+  const {queued} = rateLimiter as unknown as {queued: Array<unknown>};
+  return queued.length;
+}
+
 describe('RateLimiter', () => {
   it('immediately invokes if less than max simultaneous requests are running', () => {
     const d1 = defer();
@@ -130,4 +141,145 @@ describe('RateLimiter', () => {
     expect(hasId3Resolved).toBe(true);
     expect(sawError).toBe(true);
   });
+
+  it('parks one waiter per queued task and releases it once that task may run', async () => {
+    const rateLimiter = new RateLimiter(1);
+    expect(waitingTaskCount(rateLimiter)).toBe(0);
+
+    const total = 5;
+    const deferreds = Array.from({length: total}, () => defer<number>());
+    const results = deferreds.map((deferred, i) =>
+      rateLimiter.enqueueRun(() => deferred.promise.then(() => i)),
+    );
+
+    await nextTick();
+    // every task but the running one is waiting for its turn
+    expect(waitingTaskCount(rateLimiter)).toBe(total - 1);
+
+    // letting one task through only releases that task
+    deferreds[0].resolve(0);
+    await nextTick();
+    expect(waitingTaskCount(rateLimiter)).toBe(total - 2);
+
+    deferreds.forEach((deferred, i) => deferred.resolve(i));
+    expect(await Promise.all(results)).toEqual([0, 1, 2, 3, 4]);
+    expect(waitingTaskCount(rateLimiter)).toBe(0);
+  });
+
+  it('does not accumulate waiters over the lifetime of the limiter', async () => {
+    const rateLimiter = new RateLimiter(1);
+    // Guards this design's own accumulator, NOT the per-waiter subscription leak that motivated it
+    // — `queued` drained correctly even in the leaking version, so this passes against it. The
+    // regression guard for that leak is the `addEventListener` spy in the next test.
+    // Runs one task while a second waits, so the number of waiters alive at any moment is 1
+    // no matter how many pairs run: the count can only grow if finished tasks stay in the queue.
+    const runPair = async (remaining: number): Promise<void> => {
+      if (remaining === 0) {
+        return;
+      }
+      const first = defer<void>();
+      const second = defer<void>();
+      const runs = [
+        rateLimiter.enqueueRun(() => first.promise),
+        rateLimiter.enqueueRun(() => second.promise),
+      ];
+
+      await nextTick();
+      expect(waitingTaskCount(rateLimiter)).toBe(1);
+
+      first.resolve(undefined);
+      second.resolve(undefined);
+      await Promise.all(runs);
+      expect(waitingTaskCount(rateLimiter)).toBe(0);
+
+      return runPair(remaining - 1);
+    };
+
+    await runPair(20);
+  });
+
+  it('adds no EventTarget listener for a burst of waiters past the warning threshold', async () => {
+    // This is the regression guard for the per-waiter subscription leak. The EventTarget flavor of
+    // `MaxListenersExceededWarning` can only come from `addEventListener`, and `TypedEventEmitter`
+    // wraps an `EventTarget`, so never calling it is what makes that warning unreachable rather
+    // than merely rare. (Node raises the same warning from `EventEmitter.addListener` too, which
+    // this does not cover — parking waiters on a `node:events` emitter would reintroduce it.)
+    // Capturing `process.on('warning')` cannot prove this here: jest-environment-node gives each
+    // test a copy of `process`, and warnings raised by the real one never reach that copy.
+    const addEventListener = jest.spyOn(EventTarget.prototype, 'addEventListener');
+    try {
+      const maxSimultaneous = 4;
+      const total = 20; // leaves 16 tasks waiting at once, well past the threshold of 10
+      const rateLimiter = new RateLimiter(maxSimultaneous);
+      const deferreds = Array.from({length: total}, () => defer<void>());
+      const started: Array<number> = [];
+      const results = deferreds.map((deferred, i) =>
+        rateLimiter.enqueueRun(async () => {
+          started.push(i);
+          await deferred.promise;
+          return i;
+        }),
+      );
+
+      await nextTick();
+      expect(waitingTaskCount(rateLimiter)).toBe(total - maxSimultaneous);
+      expect(addEventListener).not.toHaveBeenCalled();
+
+      deferreds.forEach(deferred => deferred.resolve(undefined));
+      const inOrder = Array.from({length: total}, (_, i) => i);
+      expect(await Promise.all(results)).toEqual(inOrder);
+      expect(started).toEqual(inOrder);
+      expect(waitingTaskCount(rateLimiter)).toBe(0);
+      expect(addEventListener).not.toHaveBeenCalled();
+    } finally {
+      addEventListener.mockRestore();
+    }
+  });
+
+  it('still runs every task in order without exceeding the concurrency limit', async () => {
+    const rateLimiter = new RateLimiter(2);
+    const deferreds = Array.from({length: 6}, () => defer<number>());
+    const started: Array<number> = [];
+    let running = 0;
+    let maxRunning = 0;
+
+    const results = deferreds.map((deferred, i) =>
+      rateLimiter.enqueueRun(async () => {
+        started.push(i);
+        running++;
+        maxRunning = Math.max(maxRunning, running);
+        const result = await deferred.promise;
+        running--;
+        return result;
+      }),
+    );
+
+    await nextTick();
+    expect(started).toEqual([0, 1]);
+
+    deferreds.forEach((deferred, i) => deferred.resolve(i * 10));
+    expect(await Promise.all(results)).toEqual([0, 10, 20, 30, 40, 50]);
+    expect(started).toEqual([0, 1, 2, 3, 4, 5]);
+    expect(maxRunning).toBe(2);
+  });
+
+  it('still logs when a task is enqueued and when it is allowed to run', async () => {
+    const log = jest.fn();
+    const rateLimiter = new RateLimiter(1, log);
+    const d1 = defer<void>();
+    const d2 = defer<void>();
+
+    const first = rateLimiter.enqueueRun(() => d1.promise);
+    const second = rateLimiter.enqueueRun(() => d2.promise);
+    expect(log).toHaveBeenCalledWith('1 tasks are already running, enqueuing ID:2');
+    expect(log).not.toHaveBeenCalledWith('now allowing ID:2 to run');
+
+    d1.resolve(undefined);
+    await nextTick();
+    expect(log).toHaveBeenCalledWith('now allowing ID:2 to run');
+
+    d2.resolve(undefined);
+    await Promise.all([first, second]);
+    expect(log).toHaveBeenCalledTimes(2);
+  });
 });