diagnostics_channel: grow native channel storage

lib/diagnostics_channel.js

@@ -30,8 +30,9 @@ const {
 
 const { triggerUncaughtException } = internalBinding('errors');
 
+// The subscriber buffer is replaced when native channel storage grows, so it
+// must always be accessed through the binding instead of cached.
 const dc_binding = internalBinding('diagnostics_channel');
-const { subscribers: subscriberCounts } = dc_binding;
 
 const { WeakReference, kEmptyObject } = require('internal/util');
 const { isPromise } = require('internal/util/types');
@@ -132,7 +133,7 @@ class ActiveChannel {
     this._subscribers = ArrayPrototypeSlice(this._subscribers);
     ArrayPrototypePush(this._subscribers, subscription);
     channels.incRef(this.name);
-    if (this._index !== undefined) subscriberCounts[this._index]++;
+    if (this._index !== undefined) dc_binding.subscribers[this._index]++;
   }
 
   unsubscribe(subscription) {
@@ -145,7 +146,7 @@ class ActiveChannel {
     ArrayPrototypePushApply(this._subscribers, after);
 
     channels.decRef(this.name);
-    if (this._index !== undefined) subscriberCounts[this._index]--;
+    if (this._index !== undefined) dc_binding.subscribers[this._index]--;
     maybeMarkInactive(this);
 
     return true;
@@ -155,7 +156,7 @@ class ActiveChannel {
     const replacing = this._stores.has(store);
     if (!replacing) {
       channels.incRef(this.name);
-      if (this._index !== undefined) subscriberCounts[this._index]++;
+      if (this._index !== undefined) dc_binding.subscribers[this._index]++;
     }
     this._stores.set(store, transform);
   }
@@ -168,7 +169,7 @@ class ActiveChannel {
     this._stores.delete(store);
 
     channels.decRef(this.name);
-    if (this._index !== undefined) subscriberCounts[this._index]--;
+    if (this._index !== undefined) dc_binding.subscribers[this._index]--;
     maybeMarkInactive(this);
 
     return true;
@@ -208,9 +209,7 @@ class Channel {
     this._subscribers = undefined;
     this._stores = undefined;
     this.name = name;
-    if (typeof name === 'string') {
-      this._index = dc_binding.getOrCreateChannelIndex(name);
-    }
+    this._index = undefined;
 
     channels.set(name, this);
   }
@@ -640,7 +639,15 @@ function tracingChannel(nameOrChannels) {
   return new TracingChannel(nameOrChannels);
 }
 
-dc_binding.linkNativeChannel((name) => channel(name));
+// Keep in sync with setupDiagnosticsChannel() in pre_execution.js.
+dc_binding.linkNativeChannel((name, index) => {
+  const linkedChannel = channel(name);
+  linkedChannel._index = index;
+  dc_binding.subscribers[index] =
+    (linkedChannel._subscribers?.length || 0) +
+    (linkedChannel._stores?.size || 0);
+  return linkedChannel;
+});
 
 module.exports = {
   channel,

lib/internal/process/pre_execution.js

@@ -667,9 +667,18 @@ function initializeClusterIPC() {
 function setupDiagnosticsChannel() {
   // Re-link native channels after snapshot deserialization since
   // JS references are cleared during serialization.
+  // Keep this callback in sync with the initial registration in
+  // lib/diagnostics_channel.js.
   const dc = require('diagnostics_channel');
   const dc_binding = internalBinding('diagnostics_channel');
-  dc_binding.linkNativeChannel((name) => dc.channel(name));
+  dc_binding.linkNativeChannel((name, index) => {
+    const channel = dc.channel(name);
+    channel._index = index;
+    dc_binding.subscribers[index] =
+      (channel._subscribers?.length || 0) +
+      (channel._stores?.size || 0);
+    return channel;
+  });
 }
 
 function initializePermission() {

src/node_diagnostics_channel.cc

@@ -16,6 +16,7 @@ using v8::Function;
 using v8::FunctionCallbackInfo;
 using v8::FunctionTemplate;
 using v8::HandleScope;
+using v8::Integer;
 using v8::Isolate;
 using v8::Local;
 using v8::Object;
@@ -28,8 +29,10 @@ BindingData::BindingData(Realm* realm,
                          Local<Object> wrap,
                          InternalFieldInfo* info)
     : SnapshotableObject(realm, wrap, type_int),
-      subscribers_(
-          realm->isolate(), kMaxChannels, MAYBE_FIELD_PTR(info, subscribers)) {
+      subscribers_(realm->isolate(),
+                   info == nullptr ? kInitialChannelCapacity
+                                   : info->subscribers_capacity,
+                   MAYBE_FIELD_PTR(info, subscribers)) {
   if (info == nullptr) {
     wrap->Set(realm->context(),
               FIXED_ONE_BYTE_STRING(realm->isolate(), "subscribers"),
@@ -50,25 +53,20 @@ uint32_t BindingData::GetOrCreateChannelIndex(const std::string& name) {
   if (it != channel_indices_.end()) {
     return it->second;
   }
-  CHECK_LT(next_channel_index_, kMaxChannels);
+  if (next_channel_index_ == subscribers_.Length()) {
+    subscribers_.reserve(subscribers_.Length() * 2);
+    object()
+        ->Set(realm()->context(),
+              FIXED_ONE_BYTE_STRING(realm()->isolate(), "subscribers"),
+              subscribers_.GetJSArray())
+        .Check();
+    subscribers_.MakeWeak();
+  }
   uint32_t index = next_channel_index_++;
   channel_indices_.emplace(name, index);
   return index;
 }
 
-void BindingData::GetOrCreateChannelIndex(
-    const FunctionCallbackInfo<Value>& args) {
-  Realm* realm = Realm::GetCurrent(args);
-  BindingData* binding = realm->GetBindingData<BindingData>();
-  CHECK_NOT_NULL(binding);
-
-  CHECK(args[0]->IsString());
-  Utf8Value name(realm->isolate(), args[0]);
-
-  uint32_t index = binding->GetOrCreateChannelIndex(*name);
-  args.GetReturnValue().Set(index);
-}
-
 void BindingData::LinkNativeChannel(const FunctionCallbackInfo<Value>& args) {
   Realm* realm = Realm::GetCurrent(args);
   BindingData* binding = realm->GetBindingData<BindingData>();
@@ -85,10 +83,11 @@ void BindingData::LinkNativeChannel(const FunctionCallbackInfo<Value>& args) {
       Local<String> name =
           String::NewFromUtf8(isolate, channel_ptr->name_.c_str())
               .ToLocalChecked();
-      Local<Value> argv[] = {name};
+      Local<Value> argv[] = {
+          name, Integer::NewFromUnsigned(isolate, channel_ptr->index_)};
       Local<Value> result;
       if (binding->link_callback_.Get(isolate)
-              ->Call(context, v8::Undefined(isolate), 1, argv)
+              ->Call(context, v8::Undefined(isolate), arraysize(argv), argv)
               .ToLocal(&result) &&
           result->IsObject()) {
         channel_ptr->Link(isolate, result.As<Object>());
@@ -102,6 +101,7 @@ bool BindingData::PrepareForSerialization(Local<Context> context,
   DCHECK_NULL(internal_field_info_);
   internal_field_info_ = InternalFieldInfoBase::New<InternalFieldInfo>(type());
   internal_field_info_->subscribers = subscribers_.Serialize(context, creator);
+  internal_field_info_->subscribers_capacity = subscribers_.Length();
   link_callback_.Reset();
   channel_wrap_template_.Reset();
   channels_.clear();
@@ -130,8 +130,6 @@ void BindingData::Deserialize(Local<Context> context,
 void BindingData::CreatePerIsolateProperties(IsolateData* isolate_data,
                                              Local<ObjectTemplate> target) {
   Isolate* isolate = isolate_data->isolate();
-  SetMethod(
-      isolate, target, "getOrCreateChannelIndex", GetOrCreateChannelIndex);
   SetMethod(isolate, target, "linkNativeChannel", LinkNativeChannel);
 }
 
@@ -146,7 +144,6 @@ void BindingData::CreatePerContextProperties(Local<Object> target,
 
 void BindingData::RegisterExternalReferences(
     ExternalReferenceRegistry* registry) {
-  registry->Register(GetOrCreateChannelIndex);
   registry->Register(LinkNativeChannel);
 }
 
@@ -226,10 +223,10 @@ Channel* Channel::Get(Environment* env, const char* name) {
     HandleScope handle_scope(isolate);
     Local<Context> context = env->context();
     Local<String> js_name = String::NewFromUtf8(isolate, name).ToLocalChecked();
-    Local<Value> argv[] = {js_name};
+    Local<Value> argv[] = {js_name, Integer::NewFromUnsigned(isolate, index)};
     Local<Value> result;
     if (binding->link_callback_.Get(isolate)
-            ->Call(context, v8::Undefined(isolate), 1, argv)
+            ->Call(context, v8::Undefined(isolate), arraysize(argv), argv)
             .ToLocal(&result) &&
         result->IsObject()) {
       channel->Link(isolate, result.As<Object>());

src/node_diagnostics_channel.h

@@ -20,10 +20,11 @@ class Channel;
 
 class BindingData : public SnapshotableObject {
  public:
-  static constexpr size_t kMaxChannels = 1024;
+  static constexpr size_t kInitialChannelCapacity = 1024;
 
   struct InternalFieldInfo : public node::InternalFieldInfoBase {
     AliasedBufferIndex subscribers;
+    size_t subscribers_capacity;
   };
 
   BindingData(Realm* realm,
@@ -48,8 +49,6 @@ class BindingData : public SnapshotableObject {
   v8::Global<v8::FunctionTemplate> channel_wrap_template_;
   std::vector<BaseObjectPtr<Channel>> channels_;
 
-  static void GetOrCreateChannelIndex(
-      const v8::FunctionCallbackInfo<v8::Value>& args);
   static void LinkNativeChannel(
       const v8::FunctionCallbackInfo<v8::Value>& args);
 

test/cctest/test_diagnostics_channel.cc

@@ -257,3 +257,47 @@ TEST_F(DiagnosticsChannelTest, JSChannelVisibleFromCpp) {
   EXPECT_TRUE(js_has_subs->IsTrue());
   EXPECT_TRUE(ch->HasSubscribers());
 }
+
+// Native channels grow the shared subscriber storage past its initial
+// capacity without losing the state of channels that were already linked.
+// Updating the first and last channels after growth also verifies that JS uses
+// the replacement buffer instead of a stale cached reference.
+TEST_F(DiagnosticsChannelTest, NativeChannelsGrowSubscriberStorage) {
+  const v8::HandleScope handle_scope(isolate_);
+  Argv argv;
+  Env env{handle_scope, argv};
+
+  SetProcessExitHandler(*env, [&](node::Environment* env_, int exit_code) {
+    EXPECT_EQ(exit_code, 0);
+    node::Stop(*env);
+  });
+
+  node::LoadEnvironment(
+      *env,
+      "globalThis.__dc = require('diagnostics_channel');"
+      "globalThis.__firstSubscriber = () => {};"
+      "globalThis.__dc.subscribe('test:cctest:grow:0', "
+      "                          globalThis.__firstSubscriber);");
+
+  Channel* first = Channel::Get(*env, "test:cctest:grow:0");
+  ASSERT_NE(first, nullptr);
+  ASSERT_TRUE(first->HasSubscribers());
+
+  Channel* last = nullptr;
+  for (size_t i = 1; i <= 1024; i++) {
+    std::string name = "test:cctest:grow:" + std::to_string(i);
+    last = Channel::Get(*env, name.c_str());
+    ASSERT_NE(last, nullptr);
+  }
+
+  RunJS(isolate_,
+        "globalThis.__dc.unsubscribe('test:cctest:grow:0', "
+        "                            globalThis.__firstSubscriber);");
+  EXPECT_FALSE(first->HasSubscribers());
+
+  RunJS(isolate_,
+        "globalThis.__lastSubscriber = () => {};"
+        "globalThis.__dc.subscribe('test:cctest:grow:1024', "
+        "                          globalThis.__lastSubscriber);");
+  EXPECT_TRUE(last->HasSubscribers());
+}

test/parallel/test-diagnostics-channel-many-channels.js

@@ -0,0 +1,19 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('node:assert');
+const dc = require('node:diagnostics_channel');
+
+let last;
+for (let i = 0; i < 1024 * 10 + 1; i++) {
+  last = dc.channel(`test:many-channels:${i}`);
+}
+
+const onMessage = common.mustCall((message, name) => {
+  assert.strictEqual(message, 'message');
+  assert.strictEqual(name, last.name);
+});
+
+last.subscribe(onMessage);
+last.publish('message');
+assert.strictEqual(last.unsubscribe(onMessage), true);

test/parallel/test-diagnostics-channel-symbol-named.js

@@ -12,6 +12,7 @@ const symbol = Symbol('test');
 
 // Individual channel objects can be created to avoid future lookups
 const channel = dc.channel(symbol);
+assert.strictEqual(Object.hasOwn(channel, '_index'), true);
 
 // Expect two successful publishes later
 channel.subscribe(common.mustCall((message, name) => {

test/parallel/test-permission-diagnostics-channel.js

@@ -12,6 +12,12 @@ const assert = require('node:assert');
 const dc = require('node:diagnostics_channel');
 const fs = require('node:fs');
 
+// JS-only channels must not consume the native subscriber storage used by the
+// permission audit publisher.
+for (let i = 0; i < 1024 * 10 + 1; i++) {
+  dc.channel(`test:permission:unrelated:${i}`);
+}
+
 const messages = [];
 dc.subscribe('node:permission-model:fs', (msg) => {
   messages.push(msg);