feat(publish): enable throttling when publishing modules (lerna#4013)

Author: Ayowel

libs/commands/publish/README.md

@@ -70,6 +70,7 @@ This is useful when a previous `lerna publish` failed to publish all packages to
 - [`--registry <url>`](#--registry-url)
 - [`--tag-version-prefix`](#--tag-version-prefix)
 - [`--temp-tag`](#--temp-tag)
+- [`--throttle`](#--throttle)
 - [`--yes`](#--yes)
 - [`--summary-file <dir>`](#--summary-file)
 
@@ -302,6 +303,20 @@ new version(s) to the dist-tag configured by [`--dist-tag`](#--dist-tag-tag) (de
 This is not generally necessary, as Lerna will publish packages in topological
 order (all dependencies before dependents) by default.
 
+### `--throttle`
+
+This option class allows to throttle the timing at which modules are published to the configured registry.
+
+- `--throttle`: Enable throttling when publishing modules
+- `--throttle-size`: The amount of modules that may be published at once (defaults to `25`)
+- `--throttle-delay`: How long to wait after a module was successfully published (defaults to 30 seconds)
+
+This is usefull to avoid errors/retries when publishing to rate-limited repositories on huge monorepos:
+
+```bash
+lerna publish from-git --throttle --throttle-delay=$((3600*24))
+```
+
 ### `--yes`
 
 ```sh

libs/commands/publish/src/command.ts

@@ -107,6 +107,18 @@ const command: CommandModule = {
         describe: "Create a temporary tag while publishing.",
         type: "boolean",
       },
+      throttle: {
+        describe: "Throttle module publication. This is implicit if a throttle size or delay is provided",
+        type: "boolean",
+      },
+      "throttle-size": {
+        describe: "Bucket size used to throttle module publication.",
+        type: "number",
+      },
+      "throttle-delay": {
+        describe: "Delay between throttle bucket items publications (in seconds).",
+        type: "number",
+      },
       "no-verify-access": {
         // proxy for --verify-access
         describe: "Do not verify package read-write access for current npm user.",

libs/commands/publish/src/index.ts

@@ -51,6 +51,7 @@ import { getTwoFactorAuthRequired } from "./lib/get-two-factor-auth-required";
 import { gitCheckout } from "./lib/git-checkout";
 import { interpolate } from "./lib/interpolate";
 import { removeTempLicenses } from "./lib/remove-temp-licenses";
+import { Queue, TailHeadQueue } from "./lib/throttle-queue";
 import { verifyNpmPackageAccess } from "./lib/verify-npm-package-access";
 
 module.exports = function factory(argv: Arguments<PublishCommandConfigOptions>) {
@@ -87,6 +88,9 @@ interface PublishCommandConfigOptions extends CommandConfigOptions {
   rejectCycles?: boolean;
   distTag?: string;
   preDistTag?: string;
+  throttle?: boolean;
+  throttleSize?: number;
+  throttleDelay?: number;
 }
 
 class PublishCommand extends Command {
@@ -973,14 +977,29 @@ class PublishCommand extends Command {
     };
     process.on("log", logListener);
 
+    let queue: Queue | undefined = undefined;
+    if (this.options.throttle) {
+      const DEFAULT_QUEUE_THROTTLE_SIZE = 25;
+      const DEFAULT_QUEUE_THROTTLE_DELAY = 30;
+      queue = new TailHeadQueue(
+        this.options.throttleSize !== undefined ? this.options.throttleSize : DEFAULT_QUEUE_THROTTLE_SIZE,
+        (this.options.throttleDelay !== undefined
+          ? this.options.throttleDelay
+          : DEFAULT_QUEUE_THROTTLE_DELAY) * 1000
+      );
+    }
     const mapper = pPipe(
       ...[
         (pkg: Package) => {
           const preDistTag = this.getPreDistTag(pkg);
           const tag = !this.options.tempTag && preDistTag ? preDistTag : opts.tag;
           const pkgOpts = Object.assign({}, opts, { tag });
 
-          return pulseTillDone(npmPublish(pkg, pkg.packed.tarFilePath, pkgOpts, this.otpCache))
+          return pulseTillDone(
+            queue
+              ? queue.queue(() => npmPublish(pkg, pkg.packed.tarFilePath, pkgOpts, this.otpCache))
+              : npmPublish(pkg, pkg.packed.tarFilePath, pkgOpts, this.otpCache)
+          )
             .then(() => {
               this.publishedPackages.push(pkg);
 

libs/commands/publish/src/lib/throttle-queue.spec.ts

@@ -0,0 +1,46 @@
+import * as throttling from "./throttle-queue";
+
+describe("verifyTailHeadQueueBehavior", () => {
+  test("immediately runs all provided resolving promises within queue size", async () => {
+    const count = 100;
+    const queue = new throttling.TailHeadQueue(count, 1000);
+    const queue_promises = Array(count)
+      .fill(undefined)
+      .map(() => queue.queue(async () => Date.now()));
+    const acc = await Promise.all(queue_promises);
+    expect(acc.length).toBe(count);
+    acc.sort();
+    expect(acc[count - 1] - acc[0]).toBeLessThan(100);
+  });
+  test("immediately runs all provided rejecting promises within queue size", async () => {
+    const count = 100;
+    const queue = new throttling.TailHeadQueue(count, 1000);
+    const queue_promises = Array(count)
+      .fill(undefined)
+      .map(() => queue.queue(async () => new Promise((_, r) => r(Date.now()))));
+    const acc = await Promise.allSettled(queue_promises);
+    expect(acc.length).toBe(count);
+    const resolved_sorted_acc = acc
+      .map((r) => {
+        expect(r.status).toBe("rejected");
+        if (r.status === "rejected") {
+          return r.reason;
+        }
+      })
+      .sort();
+    expect(resolved_sorted_acc[count - 1] - resolved_sorted_acc[0]).toBeLessThan(100);
+  });
+  test("runs all provided resolving promises with a delay if they exceed queue size", async () => {
+    const count = 100;
+    const queue = new throttling.TailHeadQueue(count / 3 + 1, 2 * 1000);
+    const queue_promises = Array(count)
+      .fill(undefined)
+      .map(() => queue.queue(async () => Date.now()));
+    const acc = await Promise.all(queue_promises);
+    expect(acc.length).toBe(count);
+    acc.sort();
+    const total_time = acc[count - 1] - acc[0];
+    expect(total_time).toBeGreaterThan(3.8 * 1000);
+    expect(total_time).toBeLessThan(4.2 * 1000);
+  });
+});

libs/commands/publish/src/lib/throttle-queue.ts

@@ -0,0 +1,81 @@
+/**
+ * A queue abstraction for tasks that may require a delayed execution
+ */
+export interface Queue {
+  /**
+   * Run a task once the queue is cleared
+   * @param f A function that will be executed once the queue is useable
+   * @return A promise that wraps the value returned by f
+   */
+  queue<T>(f: () => Promise<T>): Promise<T>;
+}
+
+/**
+ * A sized queue that adds a delay between the end of an item's execution.
+ *
+ * Use only with async code, multi-threading is not supported.
+ */
+export class TailHeadQueue implements Queue {
+  private queue_list: ((v: any) => any)[];
+  private queue_size: number;
+  private queue_period: number;
+  private allowance: number;
+  private last_end: number[];
+
+  /**
+   * @param size The number of items that may run concurrently
+   * @param period The time between the end of the execution of an item and the start of the execution of the next one (ms)
+   */
+  constructor(size, period) {
+    this.queue_list = [];
+    this.queue_size = Math.floor(size);
+    this.queue_period = period;
+    this.allowance = this.queue_size;
+    this.last_end = [];
+  }
+
+  /**
+   * Validate the execution of a queue item and schedule the execution of the next one
+   */
+  _on_settled() {
+    const next = this.queue_list.shift();
+    if (next !== undefined) {
+      setTimeout(next, this.queue_period);
+    } else {
+      // If we ever reach this point and THEN a new item is queued, we need to
+      // explicitly keep track of execution times to be able to add a delay.
+      //
+      // We can't simply wrap this in a setTimeout as it would add a flat
+      // this.queue_period delay to lerna's execution end after the last item
+      // in the queue is processed.
+      this.last_end.push(Date.now());
+      this.allowance += 1;
+    }
+  }
+
+  async queue<T>(f: () => Promise<T>): Promise<T> {
+    let p: Promise<T>;
+    if (this.allowance > 0) {
+      this.allowance -= 1;
+      // Check if the queue should delay the promise's execution
+      if (this.allowance + 1 <= this.last_end.length) {
+        const time_offset = Date.now() - (this.last_end.shift() || 0);
+        if (time_offset < this.queue_period) {
+          p = new Promise((r) => setTimeout(r, this.queue_period - time_offset)).then(f);
+        }
+      }
+      if (p === undefined) {
+        p = f();
+      }
+    } else {
+      p = new Promise((r) => {
+        this.queue_list.push(r);
+      }).then(f);
+    }
+    return p.finally(() => {
+      // Don't wait on _on_settled as it is for the queue's continuation
+      // and should not delay processing of the queued item's result.
+      this._on_settled();
+    });
+  }
+}