test: make unique address creation thread-safe

Author: aminya

package.json

@@ -29,6 +29,7 @@
     "@types/fs-extra": "^9.0.13",
     "@types/mocha": "^10.0.6",
     "@types/node": "^18.19.34",
+    "@types/proper-lockfile": "^4.1.4",
     "@types/semver": "^7.5.8",
     "@types/shelljs": "^0.8.15",
     "@types/which": "^2.0.2",
@@ -49,6 +50,8 @@
     "npm-run-all2": "^6.2.0",
     "prebuildify": "^6.0.1",
     "prettier": "^2.8.8",
+    "proper-lockfile": "^4.1.2",
+    "random-words": "^1",
     "rocha": "^2.5.10",
     "semver": "^7.6.2",
     "ts-node": "~10.9.2",

pnpm-lock.yaml

@@ -1,6 +1,7 @@
 import * as path from "path"
 import * as semver from "semver"
 import * as fs from "fs"
+import * as lockfile from "proper-lockfile"
 
 import {spawn} from "child_process"
 
@@ -12,18 +13,63 @@ if (semver.satisfies(zmq.version, ">= 4.2")) {
   zmq.context.blocky = false
 }
 
-/* Windows cannot bind on a ports just above 1014; start higher to be safe. */
-let seq = 5000
+/**
+ * Get a unique id to be used as a port number or IPC path.
+ * This function is thread-safe and will use a lock file to ensure that the id is unique.
+ */
+let idFallback = 5000
+async function getUniqueId() {
+  const idPath = path.resolve(__dirname, "../../tmp/id")
+  await fs.promises.mkdir(path.dirname(idPath), {recursive: true})
+
+  try {
+    // Create the file if it doesn't exist
+    if (!fs.existsSync(idPath)) {
+      await fs.promises.writeFile(idPath, "5000", "utf8")
+
+      /* Windows cannot bind on a ports just above 1014; start higher to be safe. */
+      return 5000
+    }
+
+    await lockfile.lock(idPath, {retries: 10})
+
+    // Read the current number from the file
+    const idString = await fs.promises.readFile(idPath, "utf8")
+    let id = parseInt(idString, 10)
+
+    // Increment the number
+    id++
+
+    // Ensure the number is within the valid port range
+    if (id > 65535) {
+      idFallback++
+      id = idFallback
+    }
+
+    // Write the new number back to the file
+    await fs.promises.writeFile(idPath, id.toString(), "utf8")
+
+    return id
+  } catch (err) {
+    console.error(`Error getting unique id via id file: ${err}`)
+    return idFallback++
+  } finally {
+    // Release the lock
+    try {
+      await lockfile.unlock(idPath)
+    } catch {
+      // ignore
+    }
+  }
+}
 
 type Proto = "ipc" | "tcp" | "udp" | "inproc"
 
-export function uniqAddress(proto: Proto) {
-  const id = seq++
+export async function uniqAddress(proto: Proto) {
+  const id = await getUniqueId()
   switch (proto) {
     case "ipc": {
       const sock = path.resolve(__dirname, `../../tmp/${proto}-${id}`)
-      // create the directory
-      fs.mkdirSync(path.dirname(sock), {recursive: true})
 
       return `${proto}://${sock}`
     }

test/unit/helpers.ts

@@ -22,8 +22,8 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
     beforeEach(async function () {
       proxy = new zmq.Proxy(new zmq.Router(), new zmq.Dealer())
 
-      frontAddress = uniqAddress(proto)
-      backAddress = uniqAddress(proto)
+      frontAddress = await uniqAddress(proto)
+      backAddress = await uniqAddress(proto)
 
       req = new zmq.Request()
       rep = new zmq.Reply()

test/unit/proxy-router-dealer-test.ts

@@ -25,7 +25,7 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
     })
 
     it("should fail if front end is not bound or connected", async function () {
-      await proxy.backEnd.bind(uniqAddress(proto))
+      await proxy.backEnd.bind(await uniqAddress(proto))
 
       try {
         await proxy.run()
@@ -39,8 +39,8 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
     })
 
     it("should fail if front end is not open", async function () {
-      await proxy.frontEnd.bind(uniqAddress(proto))
-      await proxy.backEnd.bind(uniqAddress(proto))
+      await proxy.frontEnd.bind(await uniqAddress(proto))
+      await proxy.backEnd.bind(await uniqAddress(proto))
       proxy.frontEnd.close()
 
       try {
@@ -55,7 +55,7 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
     })
 
     it("should fail if back end is not bound or connected", async function () {
-      await proxy.frontEnd.bind(uniqAddress(proto))
+      await proxy.frontEnd.bind(await uniqAddress(proto))
 
       try {
         await proxy.run()
@@ -71,8 +71,8 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
     })
 
     it("should fail if back end is not open", async function () {
-      await proxy.frontEnd.bind(uniqAddress(proto))
-      await proxy.backEnd.bind(uniqAddress(proto))
+      await proxy.frontEnd.bind(await uniqAddress(proto))
+      await proxy.backEnd.bind(await uniqAddress(proto))
       proxy.backEnd.close()
 
       try {

test/unit/proxy-run-test.ts

@@ -25,8 +25,8 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
     })
 
     it("should throw if called after termination", async function () {
-      await proxy.frontEnd.bind(uniqAddress(proto))
-      await proxy.backEnd.bind(uniqAddress(proto))
+      await proxy.frontEnd.bind(await uniqAddress(proto))
+      await proxy.backEnd.bind(await uniqAddress(proto))
 
       const sleep_ms = 50
 

test/unit/proxy-terminate-test.ts

@@ -19,12 +19,12 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
 
     describe("bind", function () {
       it("should resolve", async function () {
-        await sock.bind(uniqAddress(proto))
+        await sock.bind(await uniqAddress(proto))
         assert.ok(true)
       })
 
       it("should throw error if not bound to endpoint", async function () {
-        const address = uniqAddress(proto)
+        const address = await uniqAddress(proto)
         try {
           await sock.unbind(address)
           assert.ok(false)
@@ -72,8 +72,8 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
       it("should fail during other bind", async function () {
         let promise
         try {
-          promise = sock.bind(uniqAddress(proto))
-          await sock.bind(uniqAddress(proto))
+          promise = sock.bind(await uniqAddress(proto))
+          await sock.bind(await uniqAddress(proto))
           assert.ok(false)
         } catch (err) {
           if (!isFullError(err)) {
@@ -92,7 +92,7 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
 
     describe("unbind", function () {
       it("should unbind", async function () {
-        const address = uniqAddress(proto)
+        const address = await uniqAddress(proto)
         await sock.bind(address)
         await sock.unbind(address)
         assert.ok(true)
@@ -130,7 +130,7 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
 
       it("should fail during other unbind", async function () {
         let promise
-        const address = uniqAddress(proto)
+        const address = await uniqAddress(proto)
         await sock.bind(address)
         try {
           promise = sock.unbind(address)

test/unit/socket-bind-unbind-test.ts

@@ -60,15 +60,15 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
       })
 
       it("should close after successful bind", async function () {
-        const promise = sock.bind(uniqAddress(proto))
+        const promise = sock.bind(await uniqAddress(proto))
         sock.close()
         assert.equal(sock.closed, false)
         await promise
         assert.equal(sock.closed, true)
       })
 
       it("should close after unsuccessful bind", async function () {
-        const address = uniqAddress(proto)
+        const address = await uniqAddress(proto)
         await sock.bind(address)
         const promise = sock.bind(address)
         sock.close()
@@ -83,7 +83,7 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
       })
 
       it("should close after successful unbind", async function () {
-        const address = uniqAddress(proto)
+        const address = await uniqAddress(proto)
         await sock.bind(address)
         const promise = sock.unbind(address)
         sock.close()
@@ -93,7 +93,7 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
       })
 
       it("should close after unsuccessful unbind", async function () {
-        const address = uniqAddress(proto)
+        const address = await uniqAddress(proto)
         const promise = sock.unbind(address)
         sock.close()
         assert.equal(sock.closed, false)
@@ -130,7 +130,7 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
           context = undefined
 
           global.gc!()
-          socket.connect(uniqAddress(proto))
+          socket.connect(await uniqAddress(proto))
           await socket.send(Buffer.from("foo"))
           socket.close()
         }

test/unit/socket-close-test.ts

@@ -52,15 +52,15 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
       if (semver.satisfies(zmq.version, ">= 4.1")) {
         it("should allow setting routing id on router", async function () {
           sock = new zmq.Router({mandatory: true, linger: 0})
-          await sock.connect(uniqAddress(proto), {routingId: "remoteId"})
+          await sock.connect(await uniqAddress(proto), {routingId: "remoteId"})
           await sock.send(["remoteId", "hi"])
         })
       }
     })
 
     describe("disconnect", function () {
       it("should throw error if not connected to endpoint", async function () {
-        const address = uniqAddress(proto)
+        const address = await uniqAddress(proto)
         try {
           await sock.disconnect(address)
           assert.ok(false)

test/unit/socket-connect-disconnect-test.ts

@@ -39,7 +39,7 @@ for (const proto of testProtos("tcp", "ipc", "inproc")) {
 
     describe("when connected", function () {
       beforeEach(async function () {
-        const address = uniqAddress(proto)
+        const address = await uniqAddress(proto)
         await sockB.bind(address)
         await sockA.connect(address)
       })