Fix race condition (#30)

Co-authored-by: Natalie Weizenbaum <nweiz@google.com>

Author: ntkme

.eslintignore

@@ -47,7 +47,6 @@ jobs:
 
   deploy:
     name: Deploy
-    runs-on: ubuntu-latest
     if: "startsWith(github.ref, 'refs/tags/') && github.repository == 'sass/sync-message-port'"
     needs: [static_analysis, tests]
     permissions:

.eslintrc

@@ -1,3 +1,15 @@
+## 1.2.0
+
+* Calling `SyncMessagePort.receiveMessage()` or
+  `SyncMessagePort.receiveMessageIfAvailable()` while the port has listeners is
+  no longer an error. If these calls consume a message, it won't be propagated
+  to the event listener, and vice versa. If a `receiveMessage()` call is active
+  at the same time as a listener, an incoming message will be handled
+  preferentially by `receiveMessage()`.
+
+* Fix a race condition where sending many messages in a row could (rarely) cause
+  a crash.
+
 ## 1.1.3
 
 * No user-visible changes.

.github/workflows/ci.yml

@@ -0,0 +1,26 @@
+import { defineConfig } from 'eslint/config';
+import gts from 'gts';
+
+export default defineConfig([
+  gts,
+  {
+    rules: {
+      '@typescript-eslint/explicit-function-return-type': [
+        'error',
+        {'allowExpressions': true}
+      ],
+      'func-style': ['error', 'declaration'],
+      'prefer-const': ['error', {'destructuring': 'all'}],
+      // It would be nice to sort import declaration order as well, but that's not
+      // autofixable and it's not worth the effort of handling manually.
+      'sort-imports': ['error', {'ignoreDeclarationSort': true}],
+    },
+  },
+  {
+    ignores: [
+      'build/',
+      'dist/',
+      '**/*.js',
+    ],
+  },
+]);

CHANGELOG.md

@@ -0,0 +1,72 @@
+/**
+ * A counter that can be atomically incremented and decremented, and which
+ * callers can synchronously listen for it becoming non-zero.
+ *
+ * This can be "open" or "closed"; once it's closed, {@link wait} will no longer
+ * emit useful events.
+ */
+export class AtomicCounter {
+  /**
+   * The underlying BigInt64Array.
+   *
+   * The first BigInt64 represents the current value of the counter. The second
+   * BigInt64 is used to track the closed state.
+   */
+  private readonly buffer: BigInt64Array;
+
+  constructor(buffer: SharedArrayBuffer) {
+    if (buffer.byteLength !== 16) {
+      throw new Error('SharedArrayBuffer must have a byteLength of 16.');
+    }
+    this.buffer = new BigInt64Array(buffer);
+  }
+
+  /** Atomically decrement the current value by one. */
+  decrement(): void {
+    Atomics.sub(this.buffer, 0, 1n);
+  }
+
+  /** Atomically increment the current value by one. */
+  increment(): void {
+    if (Atomics.add(this.buffer, 0, 1n) === 0n) {
+      Atomics.notify(this.buffer, 0, 1);
+    }
+  }
+
+  /**
+   * Closes the counter.
+   *
+   * This will cause any outstanding calls to {@link wait} on any thread to
+   * return `true` immediately.
+   */
+  close(): void {
+    // The current value is no longer relevant once closed, therefore set it to
+    // non-zero to prevent a potential deadlock when `Atomics.notify` is called
+    // immediately before `Atomics.wait`.
+    if (
+      Atomics.compareExchange(this.buffer, 1, 0n, 1n) === 0n &&
+      Atomics.compareExchange(this.buffer, 0, 0n, 1n) === 0n
+    ) {
+      Atomics.notify(this.buffer, 0);
+    }
+  }
+
+  /**
+   * Waits until the current value is not zero or the counter is closed.
+   *
+   * Returns `true` when the counter is non-zero *or* when it's closed. Returns
+   * `false` if the counter remains zero for `timeout` milliseconds.
+   */
+  wait(timeout?: number): boolean {
+    while (
+      Atomics.load(this.buffer, 0) === 0n &&
+      Atomics.load(this.buffer, 1) === 0n
+    ) {
+      const result = Atomics.wait(this.buffer, 0, 0n, timeout);
+      if (result !== 'ok') {
+        return result !== 'timed-out';
+      }
+    }
+    return true;
+  }
+}