diff --git a/nodejs/README.md b/nodejs/README.md index a5e87815..2ddb20db 100644 --- a/nodejs/README.md +++ b/nodejs/README.md @@ -124,16 +124,12 @@ main().catch(console.error); Callbacks run only on their owning JS thread. There is no polling timer or worker thread waiting on each get/put. Explicitly close runners; GC cleanup and Node environment cleanup hooks are fallbacks, including forced worker termination. -Pending operations, active listeners, and registered error/status handlers keep their +Pending operations, active listeners, and registered status handlers keep their runner alive until completion, cancellation, handler removal, or explicit close. Graceful persistence cannot be guaranteed if the process is killed. -The cross-thread queue is limited to 1,024 events and 16 MiB, and each accumulated -get result is limited to 16 MiB and 65,536 values. Exceeding a limit closes the -runner, emits an `error` event with code `ERR_DHT_OVERFLOW`, and rejects pending -operations and `close()`. Install an `error` handler on long-lived runners. As -with other Node EventEmitters, an unhandled `error` event is fatal. There is no -silent dropping of events while continuing an apparently healthy listener. +The binding does not cap the callback queue or accumulated get results. Queued +callbacks retain their native values until delivery or cleanup. ## Crypto @@ -173,7 +169,7 @@ async function createAuthenticatedNode() { ``` Crypto methods take Uint8Array/Buffer inputs, not implicit text encodings. Imports -must be nonempty and crypto inputs are limited to 16 MiB. Malformed handles/data, +must be nonempty. Malformed handles/data, mismatched identities, and unsupported encryption keys raise JS exceptions. Only generation is asynchronous; import/export/sign/encrypt/decrypt are synchronous and should be used from worker threads for large workloads. diff --git a/nodejs/addon.cpp b/nodejs/addon.cpp index 2e50576e..c93ee74d 100644 --- a/nodejs/addon.cpp +++ b/nodejs/addon.cpp @@ -7,15 +7,13 @@ #include #include #include +#include #include #include #include namespace { -constexpr size_t maxQueuedEvents = 1024; -constexpr size_t maxResultBytes = 16 * 1024 * 1024; - struct GnuTlsContext { const int result = gnutls_global_init(); @@ -82,64 +80,28 @@ optionsObject(Napi::Value value) return value.As(); } -struct ValueData +Napi::Object +valueToJS(Napi::Env env, const dht::Value& value) { - explicit ValueData(const dht::Value& value) - : id(value.id) - , type(value.type) - , seq(value.seq) - , data(value.data) - , userType(value.user_type) - , owner(value.owner ? value.owner->getLongId().toString() : "") - , isSigned(value.isSigned()) - , isEncrypted(value.isEncrypted()) - {} - - Napi::Object toJS(Napi::Env env) const - { - auto result = Napi::Object::New(env); - result.Set("id", Napi::BigInt::New(env, id)); - result.Set("type", Napi::Number::New(env, type)); - result.Set("seq", Napi::Number::New(env, seq)); - result.Set("data", Napi::Buffer::Copy(env, data.data(), data.size())); - result.Set("userType", Napi::String::New(env, userType)); - result.Set("owner", owner.empty() ? Napi::Value(env.Null()) : Napi::Value(Napi::String::New(env, owner))); - result.Set("isSigned", Napi::Boolean::New(env, isSigned)); - result.Set("isEncrypted", Napi::Boolean::New(env, isEncrypted)); - return result; - } - - size_t size() const { return sizeof(ValueData) + data.size() + userType.size() + owner.size(); } - - uint64_t id; - uint16_t type; - uint16_t seq; - dht::Blob data; - std::string userType; - std::string owner; - bool isSigned; - bool isEncrypted; -}; - -struct Event -{ - enum class Kind { Value, Done, Overflow, Status }; - Kind kind; - uint64_t operation; - bool flag; - std::optional value; - dht::NodeStatus ipv4 = dht::NodeStatus::Disconnected; - dht::NodeStatus ipv6 = dht::NodeStatus::Disconnected; - - size_t size() const { return sizeof(Event) + (value ? value->size() : 0); } -}; + auto result = Napi::Object::New(env); + result.Set("id", Napi::BigInt::New(env, value.id)); + result.Set("type", Napi::Number::New(env, value.type)); + result.Set("seq", Napi::Number::New(env, value.seq)); + result.Set("data", Napi::Buffer::Copy(env, value.data.data(), value.data.size())); + result.Set("userType", Napi::String::New(env, value.user_type)); + result.Set("owner", + value.owner ? Napi::Value(Napi::String::New(env, value.owner->getLongId().toString())) + : Napi::Value(env.Null())); + result.Set("isSigned", Napi::Boolean::New(env, value.isSigned())); + result.Set("isEncrypted", Napi::Boolean::New(env, value.isEncrypted())); + return result; +} struct PendingOperation { Napi::Promise::Deferred deferred; bool get; - std::vector values; - size_t bytes = 0; + std::vector> values; }; struct Subscription @@ -151,6 +113,8 @@ struct Subscription struct RunnerState { + using Callback = std::function; + explicit RunnerState(Napi::Env environment) : env(environment) {} @@ -215,99 +179,44 @@ struct RunnerState } } - void send(std::unique_ptr event) - { - if (closing || overflowed) - return; - const auto eventSize = event->size(); - const auto previousCount = queuedEvents.fetch_add(1); - const auto previousBytes = queuedBytes.fetch_add(eventSize); - if (previousCount >= maxQueuedEvents || eventSize > maxResultBytes - || previousBytes > maxResultBytes - eventSize) { - queuedEvents--; - queuedBytes -= eventSize; - signalOverflow(); - return; - } - if (dispatcher.NonBlockingCall(static_cast(event.get())) == napi_ok) { - event.release(); - } else { - queuedEvents--; - queuedBytes -= eventSize; - } - } - - void sendValue(uint64_t operation, const dht::Value& value, bool expired = false) - { - if (closing || overflowed) - return; - if (value.data.size() > maxResultBytes || value.user_type.size() > maxResultBytes) { - signalOverflow(); - return; - } - send(std::make_unique(Event {Event::Kind::Value, operation, expired, ValueData(value)})); - } - - void signalOverflow() - { - if (overflowed.exchange(true)) - return; - auto event = std::make_unique(Event {Event::Kind::Overflow, 0, false, {}}); - if (dispatcher.NonBlockingCall(static_cast(event.get())) == napi_ok) - event.release(); - } - - void handle(Event& event) + void send(Callback callback) { if (closing) return; - if (event.kind == Event::Kind::Status) { - status4 = event.ipv4; - status6 = event.ipv6; - if (!onStatus.IsEmpty()) - onStatus.Call({status()}); - return; - } - if (event.kind == Event::Kind::Overflow) { - overflowed = true; - startClose(); - if (!onError.IsEmpty()) - onError.Call({makeCloseError().Value()}); - return; - } - if (event.kind == Event::Kind::Value) { - const auto listener = listeners.find(event.operation); - if (listener != listeners.end()) { - listener->second.callback.Call({event.value->toJS(env), Napi::Boolean::New(env, event.flag)}); - } else if (const auto pending = operations.find(event.operation); pending != operations.end()) { - auto& operation = pending->second; - operation.bytes += event.value->size(); - if (operation.bytes > maxResultBytes || operation.values.size() >= 65536) { - overflowed = true; - startClose(); - if (!onError.IsEmpty()) - onError.Call({makeCloseError().Value()}); - return; - } - operation.values.emplace_back(std::move(*event.value)); + auto task = std::make_unique(std::move(callback)); + if (dispatcher.NonBlockingCall(static_cast(task.get())) == napi_ok) + task.release(); + } + + void sendValue(uint64_t operation, std::shared_ptr value, bool expired = false) + { + send([operation, value = std::move(value), expired](RunnerState& state) mutable { + const auto listener = state.listeners.find(operation); + if (listener != state.listeners.end()) { + listener->second.callback.Call({valueToJS(state.env, *value), Napi::Boolean::New(state.env, expired)}); + } else if (const auto pending = state.operations.find(operation); pending != state.operations.end()) { + pending->second.values.emplace_back(std::move(value)); } - return; - } - const auto pending = operations.find(event.operation); + }); + } + + void completeOperation(uint64_t identifier, bool ok) + { + const auto pending = operations.find(identifier); if (pending == operations.end()) return; auto operation = std::move(pending->second); operations.erase(pending); if (!operation.get) { - operation.deferred.Resolve(Napi::Boolean::New(env, event.flag)); - } else if (!event.flag) { + operation.deferred.Resolve(Napi::Boolean::New(env, ok)); + } else if (!ok) { auto error = Napi::Error::New(env, "DHT get failed"); error.Set("code", "ERR_DHT_GET_FAILED"); operation.deferred.Reject(error.Value()); } else { auto values = Napi::Array::New(env, operation.values.size()); for (size_t index = 0; index < operation.values.size(); ++index) - values.Set(static_cast(index), operation.values[index].toJS(env)); + values.Set(static_cast(index), valueToJS(env, *operation.values[index])); operation.deferred.Resolve(values); } } @@ -327,15 +236,11 @@ struct RunnerState static void dispatch(napi_env environment, napi_value, void* context, void* data) { auto state = static_cast(context); - std::unique_ptr event(static_cast(data)); - if (event->kind != Event::Kind::Overflow) { - state->queuedEvents--; - state->queuedBytes -= event->size(); - } - if (!environment || state->envClosing) + std::unique_ptr callback(static_cast(data)); + if (!environment || state->envClosing || state->closing) return; try { - state->handle(*event); + (*callback)(*state); } catch (const Napi::Error& error) { napi_fatal_exception(environment, error.Value()); } catch (const std::exception& error) { @@ -383,11 +288,8 @@ struct RunnerState Napi::Error makeCloseError() const { - auto error = Napi::Error::New(env, - overflowed ? "OpenDHT result queue exceeded its limit" - : closeError.empty() ? "DHT runner is closed" - : closeError); - error.Set("code", overflowed ? "ERR_DHT_OVERFLOW" : "ERR_DHT_CLOSED"); + auto error = Napi::Error::New(env, closeError.empty() ? "DHT runner is closed" : closeError); + error.Set("code", "ERR_DHT_CLOSED"); return error; } @@ -402,7 +304,7 @@ struct RunnerState for (auto& pending : state->operations) pending.second.deferred.Reject(error.Value()); if (state->closeDeferred) { - if (!state->overflowed && state->closeError.empty()) + if (state->closeError.empty()) state->closeDeferred->Resolve(Napi::Env(environment).Undefined()); else state->closeDeferred->Reject(error.Value()); @@ -412,7 +314,6 @@ struct RunnerState state->listeners.clear(); state->closeDeferred.reset(); state->closePromise.Reset(); - state->onError.Reset(); state->onStatus.Reset(); state->owner.Reset(); napi_delete_async_work(environment, state->closeWork); @@ -427,9 +328,6 @@ struct RunnerState napi_async_cleanup_hook_handle cleanupHook {}; std::shared_ptr* closeOwner {}; std::atomic_bool closing {false}; - std::atomic_bool overflowed {false}; - std::atomic_size_t queuedEvents {0}; - std::atomic_size_t queuedBytes {0}; bool envClosing = false; bool started = false; bool hasIdentity = false; @@ -440,7 +338,6 @@ struct RunnerState std::unordered_map listeners; std::optional closeDeferred; Napi::Reference closePromise; - Napi::FunctionReference onError; Napi::FunctionReference onStatus; Napi::ObjectReference owner; dht::NodeStatus status4 = dht::NodeStatus::Disconnected; @@ -455,9 +352,7 @@ public: , state_(RunnerState::create(info.Env())) { if (info.Length() && info[0].IsFunction()) - state_->onError = Napi::Persistent(info[0].As()); - if (info.Length() > 1 && info[1].IsFunction()) - state_->onStatus = Napi::Persistent(info[1].As()); + state_->onStatus = Napi::Persistent(info[0].As()); } ~Runner() override { state_->startClose(); } @@ -510,7 +405,12 @@ private: const std::weak_ptr state = state_; context.statusChangedCallback = [state](dht::NodeStatus ipv4, dht::NodeStatus ipv6) { if (auto owner = state.lock()) - owner->send(std::make_unique(Event {Event::Kind::Status, 0, false, {}, ipv4, ipv6})); + owner->send([ipv4, ipv6](RunnerState& state) { + state.status4 = ipv4; + state.status6 = ipv6; + if (!state.onStatus.IsEmpty()) + state.onStatus.Call({state.status()}); + }); }; const auto service = std::to_string(port); state_->runner.run(ipv4.c_str(), ipv6.c_str(), service.c_str(), config, std::move(context)); @@ -528,7 +428,7 @@ private: return state_->closePromise.Value(); auto deferred = Napi::Promise::Deferred::New(info.Env()); if (state_->finished) { - if (state_->overflowed || !state_->closeError.empty()) + if (!state_->closeError.empty()) deferred.Reject(state_->makeCloseError().Value()); else deferred.Resolve(info.Env().Undefined()); @@ -543,7 +443,7 @@ private: Napi::Value isRunning(const Napi::CallbackInfo& info) { - return Napi::Boolean::New(info.Env(), !state_->closing && !state_->overflowed && state_->runner.isRunning()); + return Napi::Boolean::New(info.Env(), !state_->closing && state_->runner.isRunning()); } Napi::Value getStatus(const Napi::CallbackInfo& info) @@ -597,16 +497,17 @@ private: dht::GetCallback( [state, operation](const std::vector>& values) { const auto owner = state.lock(); - if (!owner || owner->closing || owner->overflowed) + if (!owner || owner->closing) return false; for (const auto& value : values) - owner->sendValue(operation, *value); - return !owner->closing && !owner->overflowed; + owner->sendValue(operation, value); + return !owner->closing; }), dht::DoneCallbackSimple([state, operation](bool ok) { if (const auto owner = state.lock()) - owner->send( - std::make_unique(Event {Event::Kind::Done, operation, ok, {}})); + owner->send([operation, ok](RunnerState& state) { + state.completeOperation(operation, ok); + }); })); } catch (...) { state_->operations.erase(operation); @@ -666,7 +567,7 @@ private: try { dht::DoneCallbackSimple done = [state, operation](bool ok) { if (const auto owner = state.lock()) - owner->send(std::make_unique(Event {Event::Kind::Done, operation, ok, {}})); + owner->send([operation, ok](RunnerState& state) { state.completeOperation(operation, ok); }); }; if (mode == PutMode::Signed) state_->runner.putSigned(key, value, std::move(done), permanent); @@ -697,11 +598,11 @@ private: [state, operation](const std::vector>& values, bool expired) { const auto owner = state.lock(); - if (!owner || owner->closing || owner->overflowed) + if (!owner || owner->closing) return false; for (const auto& value : values) - owner->sendValue(operation, *value, expired); - return !owner->closing && !owner->overflowed; + owner->sendValue(operation, value, expired); + return !owner->closing; })) .share(); state_->listeners.emplace(operation, diff --git a/nodejs/crypto.cpp b/nodejs/crypto.cpp index 4734b694..775a6e78 100644 --- a/nodejs/crypto.cpp +++ b/nodejs/crypto.cpp @@ -51,8 +51,6 @@ bytes(Napi::Value value) if (!value.IsTypedArray() || value.As().TypedArrayType() != napi_uint8_array) throw Napi::TypeError::New(value.Env(), "Expected a Uint8Array"); auto array = value.As(); - if (array.ElementLength() > 16 * 1024 * 1024) - throw Napi::RangeError::New(value.Env(), "Crypto input exceeds 16 MiB"); dht::Blob result(array.ElementLength()); if (!result.empty()) std::copy_n(array.Data(), result.size(), result.data()); diff --git a/nodejs/index.js b/nodejs/index.js index e811fd1f..40d64e27 100644 --- a/nodejs/index.js +++ b/nodejs/index.js @@ -21,12 +21,11 @@ class DhtRunner extends EventEmitter { constructor() { super(); const owner = new WeakRef(this); - this.#native = new native.NativeRunner(eventReporter(owner, 'error'), eventReporter(owner, 'statusChanged')); + this.#native = new native.NativeRunner(eventReporter(owner, 'statusChanged')); } #updateOwner() { - this.#native.setOwner(this.#listeners || this.#pending || this.listenerCount('error') - || this.listenerCount('statusChanged') ? this : null); + this.#native.setOwner(this.#listeners || this.#pending || this.listenerCount('statusChanged') ? this : null); } #track(operation) { diff --git a/nodejs/test/basic.test.js b/nodejs/test/basic.test.js index e0232db6..27788722 100644 --- a/nodejs/test/basic.test.js +++ b/nodejs/test/basic.test.js @@ -111,6 +111,7 @@ test('two loopback nodes exchange binary values and cancel listeners', { timeout const key = hash('node-addon-api integration'); const payload = Buffer.from([0, 1, 255, 0, 42]); let subscription; + let retainedValue; try { writer.bootstrap('127.0.0.1', reader.getBoundPort()); reader.bootstrap('127.0.0.1', writer.getBoundPort()); @@ -126,17 +127,26 @@ test('two loopback nodes exchange binary values and cancel listeners', { timeout }), true); const [value] = await received; assert.equal(value.id, 18446744073709551614n); + assert.equal(value.type, 0); + assert.ok(Buffer.isBuffer(value.data)); assert.deepEqual(value.data, payload); assert.equal(value.userType, 'nodejs-test'); assert.equal(typeof value.seq, 'number'); + assert.equal(value.owner, null); + assert.equal(value.isSigned, false); + assert.equal(value.isEncrypted, false); + value.data.fill(7); const values = await reader.get(key); - assert.ok(values.some(candidate => candidate.id === value.id && candidate.data.equals(payload))); + retainedValue = values.find(candidate => candidate.id === value.id); + assert.ok(retainedValue); + assert.deepEqual(retainedValue.data, payload); subscription.cancel(); subscription.cancel(); } finally { subscription?.cancel(); await Promise.all([writer.close(), reader.close()]); } + assert.deepEqual(retainedValue.data, payload); }); test('close settles pending operations and suppresses queued listener deliveries', { timeout: 10000 }, async () => { @@ -179,12 +189,15 @@ test('listener callbacks may cancel themselves and close the runner', { timeout: subscription.cancel(); finish(runner.close()); }); - const put = runner.put(key, Buffer.from('value')).catch(error => { - assert.equal(error.code, 'ERR_DHT_CLOSED'); - }); + const puts = []; + for (let index = 0; index < 3000; index++) { + puts.push(runner.put(key, Buffer.from('value'), { id: BigInt(index + 1) }).catch(error => { + assert.equal(error.code, 'ERR_DHT_CLOSED'); + })); + } try { await received; - await put; + await Promise.all(puts); assert.equal(deliveries, 1); } finally { await runner.close(); @@ -202,28 +215,23 @@ test('unref and garbage collection do not strand native threads', { timeout: 200 } }); -test('queue overflow reports an error and rejects close', { timeout: 15000 }, async () => { +test('callback queue completes bursts beyond the former event limit', { timeout: 15000 }, async () => { const runner = startNode(); - const key = hash('bounded listener queue'); - const reported = once(runner, 'error'); - const subscription = runner.listen(key, () => { }); + const key = hash('unbounded callback queue'); + const errors = []; + const count = 3000; + runner.on('error', error => errors.push(error)); const puts = []; try { - for (let index = 0; index < 3000; index++) { - puts.push(runner.put(key, Buffer.from('value'), { id: BigInt(index + 1) }).catch(error => { - assert.equal(error.code, 'ERR_DHT_OVERFLOW'); - })); - } - const deadline = performance.now() + 5000; - while (runner.isRunning() && performance.now() < deadline) { } - assert.equal(runner.isRunning(), false); - const [error] = await reported; - assert.equal(error.code, 'ERR_DHT_OVERFLOW'); - await Promise.all(puts); - await assert.rejects(runner.close(), { code: 'ERR_DHT_OVERFLOW' }); + for (let index = 0; index < count; index++) + puts.push(runner.put(key, Buffer.from('value'), { id: BigInt(index + 1) })); + const results = await Promise.all(puts); + assert.equal(results.length, count); + assert.ok(results.every(result => typeof result === 'boolean')); + assert.equal(runner.isRunning(), true); + assert.deepEqual(errors, []); } finally { - subscription.cancel(); - await runner.close().catch(() => { }); + await runner.close(); } }); diff --git a/nodejs/test/crypto.test.js b/nodejs/test/crypto.test.js index af20e0eb..91db4d1e 100644 --- a/nodejs/test/crypto.test.js +++ b/nodejs/test/crypto.test.js @@ -27,6 +27,8 @@ test('EC keys sign and invalid crypto inputs raise JS errors', async () => { const key = await PrivateKey.generate(); const restored = new PrivateKey(key.serialize()); assert.equal(restored.getPublicKey().checkSignature(payload, key.sign(payload)), true); + const largePayload = Buffer.alloc(16 * 1024 * 1024 + 1, 42); + assert.equal(restored.getPublicKey().checkSignature(largePayload, key.sign(largePayload)), true); assert.throws(() => key.getPublicKey().encrypt(payload)); for (const Type of [PrivateKey, PublicKey, Certificate]) { assert.throws(() => new Type(Buffer.alloc(0)));