From 8ff2fe61751cd5c559f58cec3eed4115b590991d 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 | 25 +++++++---- lib/internal/process/pre_execution.js | 11 ++++- src/node_diagnostics_channel.cc | 43 +++++++++--------- src/node_diagnostics_channel.h | 5 +-- test/cctest/test_diagnostics_channel.cc | 44 +++++++++++++++++++ .../test-diagnostics-channel-many-channels.js | 19 ++++++++ .../test-diagnostics-channel-symbol-named.js | 1 + .../test-permission-diagnostics-channel.js | 6 +++ 8 files changed, 118 insertions(+), 36 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..93a2e85857ab6c 100644 --- a/lib/diagnostics_channel.js +++ b/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, diff --git a/lib/internal/process/pre_execution.js b/lib/internal/process/pre_execution.js index 38ea6675928ad8..804d588b62ba19 100644 --- a/lib/internal/process/pre_execution.js +++ b/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() { 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..30a004b59070b5 100644 --- a/test/cctest/test_diagnostics_channel.cc +++ b/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()); +} 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-diagnostics-channel-symbol-named.js b/test/parallel/test-diagnostics-channel-symbol-named.js index 96fe0fa53596e2..84213d4e22e51e 100644 --- a/test/parallel/test-diagnostics-channel-symbol-named.js +++ b/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) => { 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);