nodejs/node#55116

.github/workflows/pull-request.yml

@@ -97,8 +97,9 @@ jobs:
           - 20.13.0 # introduces TC early exit and TC#hasSubscribers()
           - 21.0.0 # nothing special, first 21
           - 21.x # nothing special, latest 21
-          - 22.0.0 # introduces TC early exit and TC#hasSubscribers()
+          - 2.20.0 # introduces TC early exit and TC#hasSubscribers()
           - 22.x # nothing special, full support DC
+          - add me
     steps:
     - uses: actions/checkout@v2
     - uses: actions/setup-node@v1

checks.js

@@ -62,5 +62,11 @@ module.exports.hasSyncUnsubscribeBug = hasSyncUnsubscribeBug;
 function hasTracingChannelHasSubscribers() {
   return MAJOR >= 22
     || (MAJOR == 20 && MINOR >= 13);
-};
+}
 module.exports.hasTracingChannelHasSubscribers = hasTracingChannelHasSubscribers;
+
+function hasSubscribersMutationBug() {
+  // TODO: version TBD
+  return MAJOR <= 22 && MINOR <= 9;
+}
+module.exports.hasSubscribersMutationBug = hasSubscribersMutationBug;

dc-polyfill.js

@@ -34,5 +34,9 @@ if (!checks.hasTracingChannelHasSubscribers()) {
   dc = require('./patch-tracing-channel-has-subscribers.js')(dc);
 }
 
+if (!checks.hasSubscribersMutationBug()) {
+  dc = require('./patch-subscribers-mutation-bug.js')(dc);
+}
+
 module.exports = dc;
 

patch-subscribers-mutation-bug.js

@@ -0,0 +1,48 @@
+const {
+    ArrayPrototypePushApply,
+    ArrayPrototypeSlice
+} = require('./primordials.js');
+
+// The ch.unsubscribe() method doesn't return a value
+// Recent versions return if an unsubscribe succeeded
+// @see https://github.com/nodejs/node/pull/40433
+module.exports = function (unpatched) {
+  const channels = new WeakSet();
+
+  const dc_channel = unpatched.channel;
+
+  const dc = { ...unpatched };
+
+  dc.channel = function () {
+    const ch = dc_channel.apply(this, arguments);
+
+    if (channels.has(ch)) return ch;
+
+    const { subscribe, unsubscribe, publish } = ch;
+
+    ch.subscribe = function () {
+      this._subscribers = ArrayPrototypeSlice(this._subscribers);
+
+      return subscribe.apply(this, arguments);
+    };
+
+    ch.unsubscribe = function () {
+      // TODO: update me
+      this._subscribers = ArrayPrototypeSlice(this._subscribers);
+
+      return subscribe.apply(this, arguments);
+    };
+
+    ch.publish = function () {
+      const self = Object.assign({}, this);
+
+      return publish.apply(self, arguments)
+    };
+
+    channels.add(ch);
+
+    return ch;
+  };
+
+  return dc;
+};

primordials.js

@@ -17,6 +17,9 @@ const ReflectApply = Reflect.apply;
 const PromiseReject = Promise.reject.bind(Promise);
 const PromiseResolve = Promise.resolve;
 const PromisePrototypeThen = makeCall(Promise.prototype.then);
+const ArrayPrototypePush = makeCall(Array.prototype.push);
+const ArrayPrototypePushApply = (...args) => Array.prototype.push.apply(...args);
+const ArrayPrototypeSlice = makeCall(Array.prototype.slice);
 const ArrayPrototypeSplice = makeCall(Array.prototype.splice);
 const ArrayPrototypeAt = makeCall(Array.prototype.at || arrayAtPolyfill);
 const ObjectDefineProperty = Object.defineProperty;
@@ -31,6 +34,9 @@ module.exports = {
   PromiseReject,
   PromiseResolve,
   PromisePrototypeThen,
+  ArrayPrototypePush,
+  ArrayPrototypePushApply,
+  ArrayPrototypeSlice,
   ArrayPrototypeSplice,
   ArrayPrototypeAt,
   ObjectDefineProperty,

test/test-diagnostics-channel-sync-unsubscribe.spec.js

@@ -14,6 +14,7 @@ test('test-diagnostics-channel-sync-unsubscribe', (t) => {
   const onMessageHandler = common.mustCall(() => dc.unsubscribe(channel_name, onMessageHandler));
 
   dc.subscribe(channel_name, onMessageHandler);
+  dc.subscribe(channel_name, common.mustCall());
 
   // This must not throw.
   dc.channel(channel_name).publish(published_data);