From 3a6cf52df44fb22210ad1979c91e72e9afd07b74 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Tue, 14 Jul 2026 20:21:11 +0800 Subject: [PATCH] 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 --- lib/diagnostics_channel.js | 20 ++++++--- lib/internal/process/pre_execution.js | 9 +++- src/node_diagnostics_channel.cc | 43 +++++++++---------- src/node_diagnostics_channel.h | 5 +-- test/cctest/test_diagnostics_channel.cc | 42 ++++++++++++++++++ .../test-diagnostics-channel-many-channels.js | 19 ++++++++ .../test-permission-diagnostics-channel.js | 6 +++ 7 files changed, 110 insertions(+), 34 deletions(-) create mode 100644 test/parallel/test-diagnostics-channel-many-channels.js diff --git a/lib/diagnostics_channel.js b/lib/diagnostics_channel.js index 7b78851208df66..c33908e59406e1 100644 --- a/lib/diagnostics_channel.js +++ b/lib/diagnostics_channel.js @@ -31,7 +31,6 @@ const { const { triggerUncaughtException } = internalBinding('errors'); 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 +131,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 +144,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 +154,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 +167,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; @@ -209,7 +208,7 @@ class Channel { 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,14 @@ function tracingChannel(nameOrChannels) { return new TracingChannel(nameOrChannels); } -dc_binding.linkNativeChannel((name) => channel(name)); +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, diff --git a/lib/internal/process/pre_execution.js b/lib/internal/process/pre_execution.js index 38ea6675928ad8..90474aba88052f 100644 --- a/lib/internal/process/pre_execution.js +++ b/lib/internal/process/pre_execution.js @@ -669,7 +669,14 @@ function setupDiagnosticsChannel() { // JS references are cleared during serialization. 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() { diff --git a/src/node_diagnostics_channel.cc b/src/node_diagnostics_channel.cc index 450a124c86959a..cfae019da62f02 100644 --- a/src/node_diagnostics_channel.cc +++ b/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 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& args) { - Realm* realm = Realm::GetCurrent(args); - BindingData* binding = realm->GetBindingData(); - 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& args) { Realm* realm = Realm::GetCurrent(args); BindingData* binding = realm->GetBindingData(); @@ -85,10 +83,11 @@ void BindingData::LinkNativeChannel(const FunctionCallbackInfo& args) { Local name = String::NewFromUtf8(isolate, channel_ptr->name_.c_str()) .ToLocalChecked(); - Local argv[] = {name}; + Local argv[] = { + name, Integer::NewFromUnsigned(isolate, channel_ptr->index_)}; Local 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()); @@ -102,6 +101,7 @@ bool BindingData::PrepareForSerialization(Local context, DCHECK_NULL(internal_field_info_); internal_field_info_ = InternalFieldInfoBase::New(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, void BindingData::CreatePerIsolateProperties(IsolateData* isolate_data, Local target) { Isolate* isolate = isolate_data->isolate(); - SetMethod( - isolate, target, "getOrCreateChannelIndex", GetOrCreateChannelIndex); SetMethod(isolate, target, "linkNativeChannel", LinkNativeChannel); } @@ -146,7 +144,6 @@ void BindingData::CreatePerContextProperties(Local 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 = env->context(); Local js_name = String::NewFromUtf8(isolate, name).ToLocalChecked(); - Local argv[] = {js_name}; + Local argv[] = {js_name, Integer::NewFromUnsigned(isolate, index)}; Local 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()); diff --git a/src/node_diagnostics_channel.h b/src/node_diagnostics_channel.h index 1c1831a0f9e45f..073e4e4f273bdf 100644 --- a/src/node_diagnostics_channel.h +++ b/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 channel_wrap_template_; std::vector> channels_; - static void GetOrCreateChannelIndex( - const v8::FunctionCallbackInfo& args); static void LinkNativeChannel( const v8::FunctionCallbackInfo& args); diff --git a/test/cctest/test_diagnostics_channel.cc b/test/cctest/test_diagnostics_channel.cc index b456f6aaf9eaab..f4295d3ba8b130 100644 --- a/test/cctest/test_diagnostics_channel.cc +++ b/test/cctest/test_diagnostics_channel.cc @@ -257,3 +257,45 @@ 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. +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()); +} diff --git a/test/parallel/test-diagnostics-channel-many-channels.js b/test/parallel/test-diagnostics-channel-many-channels.js new file mode 100644 index 00000000000000..3183bc10f485c1 --- /dev/null +++ b/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); diff --git a/test/parallel/test-permission-diagnostics-channel.js b/test/parallel/test-permission-diagnostics-channel.js index add99d5f8f0970..3593305e552c9d 100644 --- a/test/parallel/test-permission-diagnostics-channel.js +++ b/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);