chore(cli-integ): ResourcePool will not reacquire dead processes (#24496)

The ResourcePool acquires any available region from a collection of regions, each protected by a mutex.

The underlying mutex supports stealing locks from processes that have died, but the ResourcePool built on top never exercises that capability, because it will only try to acquire locks from mutexes it "knows" are free (i.e., for which it has seen an "unlock" event).

Try every mutex on every acquisition. This is less efficient, but it has the advantage of recovering from processes that died.

There is no test for this as the test is very awkward to write (depends on subprocesses being dead).

Also fix a bug in XpMutex itself where `acquire()` was supposed to call `tryAcquire()`, but that was fat-fingered into a recursive call.

----

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

Author: rix0rrr

packages/@aws-cdk-testing/cli-integ/lib/resource-pool.ts

@@ -44,9 +44,12 @@ export class ResourcePool<A extends string=string> {
     while (true) {
       // Start a wait on the unlock now -- if the unlock signal comes after
       // we try to acquire but before we start the wait, we might miss it.
-      const wait = this.pool.awaitUnlock(5000);
+      //
+      // (The timeout is in case the unlock signal doesn't come for whatever reason).
+      const wait = this.pool.awaitUnlock(10_000);
 
-      for (const res of this.unlockedResources()) {
+      // Try all mutexes, we might need to reacquire an expired lock
+      for (const res of this.resources) {
         const lease = await this.tryObtainLease(res);
         if (lease) {
           // Ignore the wait (count as handled)
@@ -107,13 +110,6 @@ export class ResourcePool<A extends string=string> {
     delete this.locks[value];
     await lock?.release();
   }
-
-  /**
-   * Return all resources that we definitely don't own the locks for
-   */
-  private unlockedResources(): A[] {
-    return this.resources.filter(res => !this.locks[res]);
-  }
 }
 
 /**

packages/@aws-cdk-testing/cli-integ/lib/xpmutex.ts

@@ -141,7 +141,7 @@ export class XpMutex {
       // signal due to unfortunate timing.
       const wait = this.pool.awaitUnlock(5000);
 
-      const lock = await this.acquire();
+      const lock = await this.tryAcquire();
       if (lock) {
         // Ignore the wait (count as handled)
         wait.then(() => {}, () => {});

packages/@aws-cdk-testing/cli-integ/test/xpmutex.test.ts

@@ -0,0 +1,46 @@
+import { XpMutexPool } from '../lib/xpmutex';
+
+const POOL = XpMutexPool.fromName('test-pool');
+
+test('acquire waits', async () => {
+  const mux = POOL.mutex('testA');
+  let secondLockAcquired = false;
+
+  // Current "process" acquires lock
+  const lock = await mux.acquire();
+
+  // Start a second "process" that tries to acquire the lock
+  const secondProcess = (async () => {
+    const secondLock = await mux.acquire();
+    try {
+      secondLockAcquired = true;
+    } finally {
+      await secondLock.release();
+    }
+  })();
+
+  // Once we release the lock the second process is free to take it
+  expect(secondLockAcquired).toBe(false);
+  await lock.release();
+
+  // We expect the variable to become true
+  await waitFor(() => secondLockAcquired);
+  expect(secondLockAcquired).toBe(true);
+
+  await secondProcess;
+});
+
+
+/**
+ * Poll for some condition every 10ms
+ */
+function waitFor(pred: () => boolean): Promise<void> {
+  return new Promise((ok) => {
+    const timerHandle = setInterval(() => {
+      if (pred()) {
+        clearInterval(timerHandle);
+        ok();
+      }
+    }, 5);
+  });
+}