diff --git a/lib/index.js b/lib/index.js index fa4d0f2..79a711e 100644 --- a/lib/index.js +++ b/lib/index.js @@ -4530,7 +4530,14 @@ class RTCPeerConnection extends SimpleEventTarget { } _handleNativeEvent(event) { - if (this._closed && event.type !== "datachannel") return; + if (this._closed) { + try { + event.channel?.close?.(); + } catch { + // Closed peers must ignore late native callbacks; channel close is best-effort. + } + return; + } if (this._shouldDeferIceEvent(event)) { this._deferredIceEvents.push(event); return; diff --git a/src/native/addon.cc b/src/native/addon.cc index a065d8f..af49dcb 100644 --- a/src/native/addon.cc +++ b/src/native/addon.cc @@ -54,6 +54,16 @@ struct PeerTeardownWork { std::vector> dataChannels; }; +void CloseDataChannelForTeardown(const std::shared_ptr &dataChannel) { + if (!dataChannel) + return; + try { + dataChannel->resetCallbacks(); + dataChannel->close(); + } catch (...) { + } +} + void RunPeerTeardown(PeerTeardownWork work) { auto closeSignal = std::make_shared(); work.peerConnection->onStateChange([closeSignal](rtc::PeerConnection::State state) { @@ -67,10 +77,7 @@ void RunPeerTeardown(PeerTeardownWork work) { }); for (auto &dataChannel : work.dataChannels) { - try { - dataChannel->close(); - } catch (...) { - } + CloseDataChannelForTeardown(dataChannel); } work.dataChannels.clear(); @@ -264,7 +271,9 @@ struct EventDispatcher : public std::enable_shared_from_this { private: EventDispatcher(Napi::Env env, Napi::Function callback) - : tsfn(Napi::ThreadSafeFunction::New(env, callback, "webrtc-node events", 0, 1)) {} + : tsfn(Napi::ThreadSafeFunction::New(env, callback, "webrtc-node events", 0, 1)) { + tsfn.Unref(env); + } void Drain(Napi::Env env, Napi::Function callback) { std::vector events; @@ -1043,6 +1052,10 @@ struct PeerBinding : public std::enable_shared_from_this { std::shared_ptr AddChannel(std::shared_ptr dataChannel, ChannelOptions options) { + if (shutdown.load()) { + CloseDataChannelForTeardown(dataChannel); + return nullptr; + } std::weak_ptr weak = shared_from_this(); auto channel = ChannelBinding::Create( std::move(dataChannel), dispatcher, std::move(options), @@ -1050,9 +1063,18 @@ struct PeerBinding : public std::enable_shared_from_this { if (auto self = weak.lock()) self->RemoveChannel(id, expected); }); + bool closeChannel = false; { std::lock_guard lock(channelsMutex); - channels[channel->id] = channel; + if (shutdown.load()) + closeChannel = true; + else + channels[channel->id] = channel; + } + if (closeChannel) { + channel->Destroy(); + CloseDataChannelForTeardown(channel->dataChannel); + return nullptr; } if (channel->dataChannel->isClosed()) RemoveChannel(channel->id, channel.get()); @@ -1063,6 +1085,12 @@ struct PeerBinding : public std::enable_shared_from_this { void Destroy() { ScheduleShutdown(); } + void DestroySync() { + auto work = PrepareShutdown(); + if (work) + RunPeerTeardown(std::move(*work)); + } + std::shared_ptr peerConnection; std::shared_ptr dispatcher; @@ -1201,6 +1229,8 @@ struct PeerBinding : public std::enable_shared_from_this { if (auto self = weak.lock()) { auto channel = self->AddChannel(dataChannel, ChannelBinding::IncomingOptions(dataChannel)); + if (!channel) + return; NativeEvent event; event.target = "peerconnection"; event.type = "datachannel"; @@ -1218,6 +1248,39 @@ struct PeerBinding : public std::enable_shared_from_this { std::unordered_map> channels; }; +std::mutex &PeerRegistryMutex() { + static std::mutex mutex; + return mutex; +} + +std::vector> &PeerRegistry() { + static std::vector> registry; + return registry; +} + +void RegisterPeerBinding(const std::shared_ptr &binding) { + std::lock_guard lock(PeerRegistryMutex()); + auto ®istry = PeerRegistry(); + registry.erase(std::remove_if(registry.begin(), registry.end(), + [](const auto &entry) { return entry.expired(); }), + registry.end()); + registry.push_back(binding); +} + +void CloseAllPeerBindings() { + std::vector> bindings; + { + std::lock_guard lock(PeerRegistryMutex()); + for (auto &entry : PeerRegistry()) { + if (auto binding = entry.lock()) + bindings.push_back(std::move(binding)); + } + PeerRegistry().clear(); + } + for (auto &binding : bindings) + binding->DestroySync(); +} + class NativePeerConnection : public Napi::ObjectWrap { public: static Napi::FunctionReference constructor; @@ -1261,6 +1324,7 @@ class NativePeerConnection : public Napi::ObjectWrap { throw Napi::TypeError::New(env, "NativePeerConnection requires an event callback"); auto dispatcher = EventDispatcher::Create(env, info[1].As()); binding_ = PeerBinding::Create(ParseConfiguration(info), dispatcher); + RegisterPeerBinding(binding_); } ~NativePeerConnection() override { @@ -1542,6 +1606,12 @@ void ConfigureLibDataChannelLogging() { rtc::InitLogger(rtc::LogLevel::Error); } +void CleanupNative() { + CloseAllPeerBindings(); + CloseAllIceUdpMuxBindings(); + rtc::Cleanup().wait(); +} + Napi::Object InitAll(Napi::Env env, Napi::Object exports) { ConfigureLibDataChannelLogging(); NativeDataChannel::Init(env, exports); @@ -1549,10 +1619,9 @@ Napi::Object InitAll(Napi::Env env, Napi::Object exports) { NativePeerConnection::Init(env, exports); exports.Set("generateCertificate", Napi::Function::New(env, GenerateCertificate)); exports.Set("importCertificate", Napi::Function::New(env, ImportCertificate)); - env.AddCleanupHook([]() { CloseAllIceUdpMuxBindings(); }); + env.AddCleanupHook([]() { CleanupNative(); }); exports.Set("cleanup", Napi::Function::New(env, [](const Napi::CallbackInfo &info) { - CloseAllIceUdpMuxBindings(); - rtc::Cleanup().wait(); + CleanupNative(); return info.Env().Undefined(); })); return exports; diff --git a/test/basic.test.js b/test/basic.test.js index d8a0f11..bf1a988 100644 --- a/test/basic.test.js +++ b/test/basic.test.js @@ -1,4 +1,6 @@ const assert = require("node:assert/strict"); +const { spawn } = require("node:child_process"); +const path = require("node:path"); const test = require("node:test"); const { RTCPeerConnection, @@ -81,6 +83,34 @@ function descriptionWithMaxMessageSize(description, value) { }; } +function fakeNativeDataChannel() { + return { + bindingId: 98765, + id: 0, + label: "late", + ordered: true, + maxPacketLifeTime: null, + maxRetransmits: null, + protocol: "", + negotiated: false, + bufferedAmount: 0, + isOpen: true, + isClosed: false, + maxMessageSize: 262144, + close() { + this.isOpen = false; + this.isClosed = true; + }, + setBufferedAmountLowThreshold() {}, + sendString() { + return true; + }, + sendBinary() { + return true; + }, + }; +} + function collectMessages(channel, count, timeout = 10000) { return new Promise((resolve, reject) => { const messages = []; @@ -185,6 +215,36 @@ async function closeAllAndWait(...peers) { await delay(1500); } +function runNodeScript(script, timeout = 5000) { + return new Promise((resolve, reject) => { + const child = spawn(process.execPath, ["-e", script], { + stdio: ["ignore", "pipe", "pipe"], + windowsHide: true, + }); + let output = ""; + const timer = setTimeout(() => { + child.kill(); + reject(new Error(`Child process did not exit within ${timeout}ms\n${output}`)); + }, timeout); + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (chunk) => { + output += chunk; + }); + child.stderr.on("data", (chunk) => { + output += chunk; + }); + child.once("error", (error) => { + clearTimeout(timer); + reject(error); + }); + child.once("exit", (code, signal) => { + clearTimeout(timer); + resolve({ code, signal, output }); + }); + }); +} + test("RTCSessionDescription and RTCIceCandidate expose WebRTC-shaped JSON", () => { const description = new RTCSessionDescription({ type: "offer", sdp: "v=0\r\n" }); assert.deepEqual(description.toJSON(), { type: "offer", sdp: "v=0\r\n" }); @@ -292,6 +352,42 @@ test("createDataChannel exposes core W3C attributes before negotiation", () => { pc.close(); }); +test("closed peer ignores late native datachannel events", async () => { + const pc = new RTCPeerConnection(); + let dataChannelEvents = 0; + pc.ondatachannel = () => { + dataChannelEvents += 1; + }; + + pc.close(); + const nativeChannel = fakeNativeDataChannel(); + pc._handleNativeEvent({ + target: "peerconnection", + type: "datachannel", + channelId: nativeChannel.bindingId, + channel: nativeChannel, + channelReadyState: "open", + }); + await delay(25); + + assert.equal(nativeChannel.isClosed, true); + assert.equal(pc._channels.size, 0); + assert.equal(pc._pendingDataChannelEvents.length, 0); + assert.equal(dataChannelEvents, 0); +}); + +test("unclosed native peer with a data channel does not keep Node alive", async () => { + const root = path.resolve(__dirname, ".."); + const script = ` + const { RTCPeerConnection } = require(${JSON.stringify(root)}); + const peerConnection = new RTCPeerConnection({ iceServers: [] }); + peerConnection.createDataChannel("unclosed-exit"); + `; + const result = await runNodeScript(script); + + assert.deepEqual(result, { code: 0, signal: null, output: "" }); +}); + test("native close suppression is restart-scoped and one-shot", () => { const pc = new RTCPeerConnection(); const dc = pc.createDataChannel("close-suppression");