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

@@ -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.
 constdc_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);
 
 returntrue;
@@ -155,7 +156,7 @@ class ActiveChannel {
 constreplacing=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);
 
 returntrue;
@@ -208,9 +209,7 @@ class Channel {
 this._subscribers=undefined;
 this._stores=undefined;
 this.name=name;
-if(typeofname==='string'){
-this._index=dc_binding.getOrCreateChannelIndex(name);
-}
+this._index=undefined;
 
 channels.set(name,this);
 }
@@ -640,7 +639,15 @@ function tracingChannel(nameOrChannels) {
 returnnewTracingChannel(nameOrChannels);
 }
 
-dc_binding.linkNativeChannel((name)=>channel(name));
+// Keep in sync with setupDiagnosticsChannel() in pre_execution.js.
+dc_binding.linkNativeChannel((name,index)=>{
+constlinkedChannel=channel(name);
+linkedChannel._index=index;
+dc_binding.subscribers[index]=
+(linkedChannel._subscribers?.length||0)+
+(linkedChannel._stores?.size||0);
+returnlinkedChannel;
+});
 
 module.exports={
  channel,

lib/internal/process/pre_execution.js

@@ -657,9 +657,18 @@ function initializeClusterIPC() {
 functionsetupDiagnosticsChannel(){
 // 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.
 constdc=require('diagnostics_channel');
 constdc_binding=internalBinding('diagnostics_channel');
-dc_binding.linkNativeChannel((name)=>dc.channel(name));
+dc_binding.linkNativeChannel((name,index)=>{
+constchannel=dc.channel(name);
+channel._index=index;
+dc_binding.subscribers[index]=
+(channel._subscribers?.length||0)+
+(channel._stores?.size||0);
+returnchannel;
+});
 }
 
 functioninitializePermission(){

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;
 }
 
-voidBindingData::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);
-}
-
 voidBindingData::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,
 voidBindingData::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,
 
 voidBindingData::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;
 
 classBindingData : publicSnapshotableObject {
 public:
-staticconstexprsize_tkMaxChannels = 1024;
+staticconstexprsize_tkInitialChannelCapacity = 1024;
 
 structInternalFieldInfo : publicnode::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_;
 
-staticvoidGetOrCreateChannelIndex(
-const v8::FunctionCallbackInfo<v8::Value>& args);
 staticvoidLinkNativeChannel(
 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';
+
+constcommon=require('../common');
+constassert=require('node:assert');
+constdc=require('node:diagnostics_channel');
+
+letlast;
+for(leti=0;i<1024*10+1;i++){
+last=dc.channel(`test:many-channels:${i}`);
+}
+
+constonMessage=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
 constchannel=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');
 constdc=require('node:diagnostics_channel');
 constfs=require('node:fs');
 
+// JS-only channels must not consume the native subscriber storage used by the
+// permission audit publisher.
+for(leti=0;i<1024*10+1;i++){
+dc.channel(`test:permission:unrelated:${i}`);
+}
+
 constmessages=[];
 dc.subscribe('node:permission-model:fs',(msg)=>{
 messages.push(msg);