diagnostics_channel: grow native channel storage

Every string-named JavaScript channel consumed an entry in the fixed
native subscriber array. Creating more than 1,024 channels triggered a
CHECK and terminated the process.

Allocate slots only for native publishers and grow the aliased buffer
when it fills. Refresh the JavaScript view after resizing and preserve
the capacity in snapshots.

Signed-off-by: Stephen Belanger <admin@stephenbelanger.com>
PR-URL: #64497
Reviewed-By: Rafael Gonzaga <rafael.nunu@hotmail.com>
Reviewed-By: Gerhard Stöbich <deb2001-github@yahoo.de>
Reviewed-By: James M Snell <jasnell@gmail.com>

Author: Qard

lib/diagnostics_channel.js

@@ -31,8 +31,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 } = require('internal/util');
 
@@ -111,7 +112,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) {
@@ -124,7 +125,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;
@@ -134,7 +135,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);
   }
@@ -147,7 +148,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;
@@ -192,9 +193,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);
   }
@@ -446,7 +445,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

@@ -619,9 +619,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) => {