From 738e0e22bfbce35f3118bcb9b2f8e3c27e485dcd Mon Sep 17 00:00:00 2001 From: Adrien Beraud Date: Tue, 27 Oct 2026 22:09:21 -0400 Subject: [PATCH] nodejs: add streaming iterators and startup proxy configuration --- .github/workflows/nodejs-package.yml | 17 ++++- nodejs/CMakeLists.txt | 1 + nodejs/README.md | 64 ++++++++++++++++-- nodejs/addon.cpp | 85 +++++++++++++++++------- nodejs/examples/dhtnode.ts | 2 +- nodejs/index.d.ts | 18 ++++- nodejs/index.js | 99 +++++++++++++++++++++++++++- nodejs/package.json | 2 +- 8 files changed, 253 insertions(+), 35 deletions(-) diff --git a/.github/workflows/nodejs-package.yml b/.github/workflows/nodejs-package.yml index b695d874..e72b370c 100644 --- a/.github/workflows/nodejs-package.yml +++ b/.github/workflows/nodejs-package.yml @@ -71,7 +71,19 @@ jobs: cache-dependency-path: nodejs/package-lock.json - name: Install macOS dependencies if: runner.os == 'macOS' - run: brew install cmake ninja pkg-config dylibbundler gnutls nettle argon2 msgpack-cxx fmt jsoncpp simdutf + run: brew install cmake ninja pkg-config dylibbundler gnutls nettle argon2 msgpack-cxx fmt jsoncpp simdutf asio llhttp openssl + - name: Install restinio + if: runner.os == 'macOS' + env: + RESTINIO_VERSION: '0.7.10' + run: | + include="$RUNNER_TEMP/opendht-proxy/include" + mkdir -p "$include/nonstd" + curl -fsSL --retry 3 https://raw.githubusercontent.com/martinmoene/expected-lite/v0.10.0/include/nonstd/expected.hpp -o "$include/nonstd/expected.hpp" + curl -fsSL --retry 3 "https://github.com/Stiffstream/restinio/releases/download/v${RESTINIO_VERSION}/restinio-${RESTINIO_VERSION}.tar.bz2" -o "$RUNNER_TEMP/restinio.tar.bz2" + tar -xjf "$RUNNER_TEMP/restinio.tar.bz2" -C "$RUNNER_TEMP" + cp -R "$RUNNER_TEMP/restinio-${RESTINIO_VERSION}/dev/restinio" "$include/" + echo "CMAKE_PREFIX_PATH=$RUNNER_TEMP/opendht-proxy" >> "$GITHUB_ENV" - name: Set up MSVC if: runner.os == 'Windows' uses: ilammy/msvc-dev-cmd@v1 @@ -98,6 +110,8 @@ jobs: run: | if [[ "$RUNNER_OS" == 'macOS' ]]; then export MACOSX_DEPLOYMENT_TARGET=15.0 + openssl_prefix=$(brew --prefix openssl) + export PKG_CONFIG_PATH="$openssl_prefix/lib/pkgconfig${PKG_CONFIG_PATH:+:$PKG_CONFIG_PATH}" fi npm run build -- --CDOPENDHT_DOWNLOAD_DEPS=OFF - name: Build Windows addon @@ -110,6 +124,7 @@ jobs: "--CDVCPKG_INSTALLED_DIR=$env:VCPKG_INSTALLED_DIR" ` "--CDVCPKG_TARGET_TRIPLET=$env:VCPKG_DEFAULT_TRIPLET" ` --CDVCPKG_MANIFEST_NO_DEFAULT_FEATURES=ON ` + --CDVCPKG_MANIFEST_FEATURES=proxy ` --CDCMAKE_MSVC_RUNTIME_LIBRARY=MultiThreadedDLL ` --CDOPENDHT_DOWNLOAD_DEPS=OFF if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } diff --git a/nodejs/CMakeLists.txt b/nodejs/CMakeLists.txt index 68c47e1c..8b0156d6 100644 --- a/nodejs/CMakeLists.txt +++ b/nodejs/CMakeLists.txt @@ -6,6 +6,7 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON) set(CMAKE_POSITION_INDEPENDENT_CODE ON) set(OPENDHT_TOOLS OFF CACHE BOOL "Build DHT tools") set(OPENDHT_PYTHON OFF CACHE BOOL "Build Python bindings") +set(OPENDHT_PROXY_CLIENT ON CACHE BOOL "Enable DHT proxy client") set(OPENDHT_PEER_DISCOVERY OFF CACHE BOOL "Enable multicast peer discovery") set(OPENDHT_DOCUMENTATION OFF CACHE BOOL "Build documentation") set(BUILD_TESTING OFF CACHE BOOL "Build native tests") diff --git a/nodejs/README.md b/nodejs/README.md index 5edfd269..37f11542 100644 --- a/nodejs/README.md +++ b/nodejs/README.md @@ -12,7 +12,8 @@ npm install opendht ## Build Install the native dependencies described in [BUILD.md](../BUILD.md), plus Node.js, -a C++17 compiler, and CMake 3.21 or later. Ninja is optional. +a C++17 compiler, and CMake 3.21 or later. Ninja is optional. Proxy-client support +defaults to enabled and requires ASIO, llhttp, RESTinio, and expected-lite headers. ```sh cd nodejs @@ -31,6 +32,11 @@ Native CMake options can be passed through, for example: npm run build -- --CDOPENDHT_PEER_DISCOVERY=ON ``` +For an existing build, change cached options directly with CMake, for example +`cmake -S . -B build -DOPENDHT_PROXY_CLIENT=ON`, then rebuild. A custom build can +disable proxy support with `-DOPENDHT_PROXY_CLIENT=OFF`; requesting a proxy then +throws an explicit unsupported-build error. + The loader checks local builds, then compatible Node-API binaries under `prebuilds/`. Set `OPENDHT_NODE_BINDING` to an absolute addon path to select an explicit build. Packaged binaries include OpenDHT and non-system native dependencies, so archive @@ -52,7 +58,7 @@ async function main() { const key = hash('example'); const stored = await writer.put(key, Buffer.from('hello'), { id: 42n }); if (!stored) throw new Error('DHT put failed'); - for (const value of await reader.get(key)) { + for await (const value of reader.get(key)) { console.log(value.id, value.data.toString()); } } finally { @@ -71,6 +77,7 @@ main().catch(console.error); - `run({port, ipv4, ipv6, network, persistPath, identity})` starts a threaded node. Port 0 selects a free port. The defaults bind both IP families on all interfaces. `persistPath` is optional; use an empty string to disable persistence. + `proxyServer` and `proxyUserAgent` configure proxy mode (see below). - `getBoundPort(4 | 6)` defaults to IPv4; `getNodeId()` returns the node's hex ID. - `getStatus()` returns the latest JS-dispatched `{ipv4, ipv6}` connectivity snapshot. Each family is `'disconnected'`, `'connecting'`, or `'connected'`. @@ -81,12 +88,16 @@ main().catch(console.error); peer is connected; subsequent DHT operations report their own completion. - `put(key, Uint8Array, {id, type, userType, permanent})` copies the input and resolves to the native success boolean. IDs are unsigned 64-bit `bigint` values. -- `get(key)` resolves to value snapshots, or rejects with `ERR_DHT_GET_FAILED`. +- `get(key)` returns an async iterator of values as they arrive. The iterator + ends when the lookup completes, or rejects with `ERR_DHT_GET_FAILED` on failure. + `getAll(key)` returns a single promise resolving to the complete value array. Values contain `id`, `type`, `seq`, `userType`, copied `Buffer` data, `isSigned`, `isEncrypted`, and an `owner` long fingerprint (or null for unsigned values). - `listen(key, (value, expired) => {})` returns a listener with idempotent `cancel()`. Queued deliveries are suppressed after cancellation. Callback return values are ignored; callbacks may cancel or close the runner. +- `listen(key)` without a callback returns an async iterator of `{value, expired}` + events. It continues until cancelled or the runner closes. - `close()` is idempotent and returns the same promise on repeated calls. It waits off the JS thread for graceful shutdown, then joins the native threads. After two seconds it forces the join if graceful shutdown has not completed. @@ -101,8 +112,47 @@ 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 binding does not cap the callback queue or accumulated get results. Queued -callbacks retain their native values until delivery or cleanup. +The binding does not cap the callback queue, accumulated `getAll()` results, or +iterator buffers. Slow iterator consumers can accumulate values without +backpressure. Queued callbacks retain their native values until delivery or cleanup. + +## Async Iteration + +`get()` and callback-free `listen()` start immediately and return iterators with +`next()`, `return()`, `throw()`, `cancel()`, and `Symbol.asyncDispose`. +Leaving a `for await` loop with `break` or an exception cancels the subscription. +Explicit cancellation discards buffered values and settles pending reads as done. +Closing the runner rejects active get iterators with `ERR_DHT_CLOSED`; listener +iterators end normally. Successful get completion preserves unread buffered values. + +```js +for await (const value of runner.get(hash('example'))) { + console.log(value.id, value.data); +} +const snapshots = await runner.getAll(hash('example')); + +for await (const { value, expired } of runner.listen(hash('updates'))) { + console.log(value.id, expired); + if (expired) break; +} +``` + +## Proxy Configuration + +Pass the URL of an OpenDHT HTTP(S) proxy at startup: + +```js +runner.run({ + proxyServer: 'https://proxy.example.com', + proxyUserAgent: 'my-app/1.0', +}); +``` + +This selects the native OpenDHT proxy client, not an HTTP CONNECT or SOCKS proxy. +The node does not bind UDP sockets in this mode; `port`, `ipv4`, `ipv6`, and +`bootstrap()` are for direct DHT nodes. Omit `proxyServer` or use an empty string +for direct networking. Proxy settings are startup-only; create a new runner to +change transports or servers. Packaged binaries include proxy-client support. ## Crypto @@ -189,5 +239,5 @@ does not remove OS/libc or native dependency compatibility requirements. Review the included dependency licenses and source-distribution obligations before redistributing binaries. -Query filters, async iterators, AbortSignal support, certificate trust-store -validation, CA issuance controls, and proxy controls remain future work. \ No newline at end of file +Query filters, AbortSignal support, certificate trust-store validation, CA +issuance controls, and runtime proxy switching remain future work. \ No newline at end of file diff --git a/nodejs/addon.cpp b/nodejs/addon.cpp index c93ee74d..989d9da5 100644 --- a/nodejs/addon.cpp +++ b/nodejs/addon.cpp @@ -90,7 +90,7 @@ valueToJS(Napi::Env env, const dht::Value& value) 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())) + value.owner ? Napi::Value(Napi::String::New(env, value.owner->getLongId().to_view())) : Napi::Value(env.Null())); result.Set("isSigned", Napi::Boolean::New(env, value.isSigned())); result.Set("isEncrypted", Napi::Boolean::New(env, value.isEncrypted())); @@ -102,6 +102,8 @@ struct PendingOperation Napi::Promise::Deferred deferred; bool get; std::vector> values; + Napi::FunctionReference onValue; + std::shared_ptr cancelled; }; struct Subscription @@ -195,7 +197,10 @@ struct RunnerState 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)); + if (!pending->second.onValue.IsEmpty()) + pending->second.onValue.Call({valueToJS(state.env, *value)}); + else + pending->second.values.emplace_back(std::move(value)); } }); } @@ -213,6 +218,8 @@ struct RunnerState auto error = Napi::Error::New(env, "DHT get failed"); error.Set("code", "ERR_DHT_GET_FAILED"); operation.deferred.Reject(error.Value()); + } else if (!operation.onValue.IsEmpty()) { + operation.deferred.Resolve(env.Undefined()); } else { auto values = Napi::Array::New(env, operation.values.size()); for (size_t index = 0; index < operation.values.size(); ++index) @@ -224,12 +231,8 @@ struct RunnerState Napi::Object status() const { auto result = Napi::Object::New(env); - result.Set("ipv4", - Napi::String::New(env, - std::string(dht::statusToStr(closing ? dht::NodeStatus::Disconnected : status4)))); - result.Set("ipv6", - Napi::String::New(env, - std::string(dht::statusToStr(closing ? dht::NodeStatus::Disconnected : status6)))); + result.Set("ipv4", Napi::String::New(env, dht::statusToStr(closing ? dht::NodeStatus::Disconnected : status4))); + result.Set("ipv6", Napi::String::New(env, dht::statusToStr(closing ? dht::NodeStatus::Disconnected : status6))); return result; } @@ -370,6 +373,8 @@ public: InstanceMethod("getStatus", &Runner::getStatus), InstanceMethod("bootstrap", &Runner::bootstrap), InstanceMethod("get", &Runner::get), + InstanceMethod("getAll", &Runner::getAll), + InstanceMethod("cancelGet", &Runner::cancelGet), InstanceMethod("put", &Runner::put), InstanceMethod("putSigned", &Runner::putSigned), InstanceMethod("putEncrypted", &Runner::putEncrypted), @@ -401,6 +406,13 @@ private: config.dht_config.node_config.persist_path = text(options.Get("persistPath"), "persistPath"); if (options.Has("identity")) config.dht_config.id = opendht_node::readIdentity(options.Get("identity")); + if (options.Has("proxyServer")) + config.proxy_server = text(options.Get("proxyServer"), "proxyServer"); + if (options.Has("proxyUserAgent")) { + config.proxy_user_agent = text(options.Get("proxyUserAgent"), "proxyUserAgent"); + if (config.proxy_user_agent.find_first_of("\r\n") != std::string::npos) + throw Napi::TypeError::New(info.Env(), "proxyUserAgent must not contain line breaks"); + } dht::DhtRunner::Context context; const std::weak_ptr state = state_; context.statusChangedCallback = [state](dht::NodeStatus ipv4, dht::NodeStatus ipv6) { @@ -466,7 +478,7 @@ private: { return guarded(info, [&]() -> Napi::Value { state_->requireRunning(); - return Napi::String::New(info.Env(), state_->runner.getNodeId().toString()); + return Napi::String::New(info.Env(), state_->runner.getNodeId().to_view()); }); } @@ -483,26 +495,36 @@ private: }); } - Napi::Value get(const Napi::CallbackInfo& info) + Napi::Value get(const Napi::CallbackInfo& info) { return getValues(info, true); } + + Napi::Value getAll(const Napi::CallbackInfo& info) { return getValues(info, false); } + + Napi::Value getValues(const Napi::CallbackInfo& info, bool stream) { return guarded(info, [&]() -> Napi::Value { state_->requireRunning(); const auto key = infoHash(info[0]); + if (stream && !info[1].IsFunction()) + throw Napi::TypeError::New(info.Env(), "callback must be a function"); const auto operation = state_->nextOperation++; auto deferred = Napi::Promise::Deferred::New(info.Env()); - state_->operations.emplace(operation, PendingOperation {deferred, true, {}}); + auto cancelled = std::make_shared(false); + PendingOperation pending {deferred, true, {}, {}, cancelled}; + if (stream) + pending.onValue = Napi::Persistent(info[1].As()); + state_->operations.emplace(operation, std::move(pending)); const std::weak_ptr state = state_; try { state_->runner.get(key, - dht::GetCallback( - [state, operation](const std::vector>& values) { - const auto owner = state.lock(); - if (!owner || owner->closing) - return false; - for (const auto& value : values) - owner->sendValue(operation, value); - return !owner->closing; - }), + dht::GetCallback([state, operation, cancelled]( + const std::vector>& values) { + const auto owner = state.lock(); + if (!owner || owner->closing || *cancelled) + return false; + for (const auto& value : values) + owner->sendValue(operation, value); + return !owner->closing && !*cancelled; + }), dht::DoneCallbackSimple([state, operation](bool ok) { if (const auto owner = state.lock()) owner->send([operation, ok](RunnerState& state) { @@ -513,10 +535,29 @@ private: state_->operations.erase(operation); throw; } + if (stream) { + auto result = Napi::Object::New(info.Env()); + result.Set("token", Napi::BigInt::New(info.Env(), operation)); + result.Set("done", deferred.Promise()); + return result; + } return deferred.Promise(); }); } + Napi::Value cancelGet(const Napi::CallbackInfo& info) + { + return guarded(info, [&]() -> Napi::Value { + const auto operation = state_->operations.find(unsignedBigInt(info[0])); + if (operation != state_->operations.end() && !operation->second.onValue.IsEmpty()) { + *operation->second.cancelled = true; + operation->second.deferred.Resolve(info.Env().Undefined()); + state_->operations.erase(operation); + } + return info.Env().Undefined(); + }); + } + Napi::Value put(const Napi::CallbackInfo& info) { return putValue(info, PutMode::Plain); } Napi::Value putSigned(const Napi::CallbackInfo& info) { return putValue(info, PutMode::Signed); } @@ -562,7 +603,7 @@ private: } const auto operation = state_->nextOperation++; auto deferred = Napi::Promise::Deferred::New(info.Env()); - state_->operations.emplace(operation, PendingOperation {deferred, false, {}}); + state_->operations.emplace(operation, PendingOperation {deferred, false, {}, {}, {}}); const std::weak_ptr state = state_; try { dht::DoneCallbackSimple done = [state, operation](bool ok) { @@ -672,7 +713,7 @@ hash(const Napi::CallbackInfo& info) } else { throw Napi::TypeError::New(env, "Expected a string or Uint8Array"); } - return Napi::String::New(env, result.toString()); + return Napi::String::New(env, result.to_view()); } Napi::Object diff --git a/nodejs/examples/dhtnode.ts b/nodejs/examples/dhtnode.ts index 54bed460..b4e8788e 100644 --- a/nodejs/examples/dhtnode.ts +++ b/nodejs/examples/dhtnode.ts @@ -96,7 +96,7 @@ Keys are text labels hashed with SHA-1. Values are UTF-8 text.`); const location = hash(key); switch (command) { case 'get': { - const found = await runner.get(location); + const found = await runner.getAll(location); if (closing) return; for (const value of found) printValue(key, value); if (!found.length) print(`No values for ${key}`); diff --git a/nodejs/index.d.ts b/nodejs/index.d.ts index 059512a4..b6506494 100644 --- a/nodejs/index.d.ts +++ b/nodejs/index.d.ts @@ -49,6 +49,8 @@ export interface RunOptions { network?: number; persistPath?: string; identity?: Identity; + proxyServer?: string; + proxyUserAgent?: string; } export interface PutOptions { @@ -73,6 +75,18 @@ export interface Listener { cancel(): void; } +export interface ListenEvent { + value: Value; + expired: boolean; +} + +export interface ValueIterator extends AsyncIterableIterator, Listener, AsyncDisposable { + return(): Promise>; + throw(error?: unknown): Promise>; +} + +export interface ListenerIterator extends ValueIterator { } + export type NodeStatus = 'disconnected' | 'connecting' | 'connected'; export interface ConnectivityStatus { @@ -92,10 +106,12 @@ export class DhtRunner extends EventEmitter implements AsyncDispos getNodeId(): string; getStatus(): ConnectivityStatus; bootstrap(host: string, port: number): void; - get(key: string): Promise; + get(key: string): ValueIterator; + getAll(key: string): Promise; put(key: string, data: Uint8Array, options?: PutOptions): Promise; putSigned(key: string, data: Uint8Array, options?: PutOptions): Promise; putEncrypted(key: string, data: Uint8Array, recipient: PublicKey, options?: PutOptions): Promise; + listen(key: string): ListenerIterator; listen(key: string, callback: (value: Value, expired: boolean) => void): Listener; close(): Promise; ref(): this; diff --git a/nodejs/index.js b/nodejs/index.js index 40d64e27..11ed0e6e 100644 --- a/nodejs/index.js +++ b/nodejs/index.js @@ -17,6 +17,7 @@ class DhtRunner extends EventEmitter { #closing; #listeners = 0; #pending = 0; + #iterators = new Set(); constructor() { super(); @@ -45,6 +46,11 @@ class DhtRunner extends EventEmitter { removeAllListeners(...args) { super.removeAllListeners(...args); this.#updateOwner(); return this; } run(options) { + if (typeof options?.proxyServer === 'string' && options.proxyServer) { + const proxy = new URL(options.proxyServer); + if (proxy.protocol !== 'http:' && proxy.protocol !== 'https:') + throw new TypeError('proxyServer must be an HTTP(S) URL'); + } if (options?.identity !== undefined) options = { ...options, identity: handle(options.identity, Identity) }; this.#native.run(options); } @@ -53,7 +59,14 @@ class DhtRunner extends EventEmitter { getNodeId() { return this.#native.getNodeId(); } getStatus() { return this.#native.getStatus(); } bootstrap(host, port) { this.#native.bootstrap(host, port); } - get(key) { return this.#track(this.#native.get(key)); } + get(key) { + return this.#iterate((push, finish) => { + const operation = this.#native.get(key, push); + this.#track(operation.done).then(() => finish(), finish); + return () => this.#native.cancelGet(operation.token); + }, true); + } + getAll(key) { return this.#track(this.#native.getAll(key)); } put(key, data, options) { return this.#track(this.#native.put(key, data, options)); } putSigned(key, data, options) { return this.#track(this.#native.putSigned(key, data, options)); } putEncrypted(key, data, recipient, options) { @@ -61,6 +74,12 @@ class DhtRunner extends EventEmitter { } listen(key, callback) { + if (callback === undefined) { + return this.#iterate(push => { + const subscription = this.listen(key, (value, expired) => push({ value, expired })); + return () => subscription.cancel(); + }); + } const token = this.#native.listen(key, callback); this.#listeners++; this.#updateOwner(); @@ -76,7 +95,83 @@ class DhtRunner extends EventEmitter { }; } - close() { return this.#closing ??= this.#native.close(); } + #iterate(subscribe, rejectOnClose = false) { + let values = []; + let offset = 0; + const pending = []; + let done = false; + let failure; + const finish = error => { + if (done) return; + done = true; + failure = error; + unsubscribe(); + this.#iterators.delete(onClose); + if (failure) { + values = []; + offset = 0; + } + for (const { resolve, reject } of pending.splice(0)) { + if (failure) reject(failure); + else resolve({ value: undefined, done: true }); + } + }; + const cancel = () => { + values = []; + offset = 0; + failure = undefined; + finish(); + }; + const onClose = () => { + if (rejectOnClose) { + const error = new Error('DHT runner is closed'); + error.code = 'ERR_DHT_CLOSED'; + finish(error); + } else cancel(); + }; + const unsubscribe = subscribe(value => { + if (done) return; + const waiter = pending.shift(); + if (waiter) waiter.resolve({ value, done: false }); + else values.push(value); + }, finish); + this.#iterators.add(onClose); + return { + next() { + if (failure) return Promise.reject(failure); + if (offset < values.length) { + const value = values[offset]; + values[offset++] = undefined; + if (offset === values.length) { + values = []; + offset = 0; + } + return Promise.resolve({ value, done: false }); + } + if (done) return Promise.resolve({ value: undefined, done: true }); + return new Promise((resolve, reject) => pending.push({ resolve, reject })); + }, + return() { + cancel(); + return Promise.resolve({ value: undefined, done: true }); + }, + throw(error) { + cancel(); + return Promise.reject(error); + }, + cancel, + [Symbol.asyncIterator]() { return this; }, + [Symbol.asyncDispose]() { cancel(); return Promise.resolve(); }, + }; + } + + close() { + if (!this.#closing) { + this.#closing = this.#native.close(); + for (const cancel of this.#iterators) cancel(); + } + return this.#closing; + } ref() { this.#native.ref(); return this; } unref() { this.#native.unref(); return this; } [Symbol.asyncDispose]() { return this.close(); } diff --git a/nodejs/package.json b/nodejs/package.json index 23d83b61..a8467c8f 100644 --- a/nodejs/package.json +++ b/nodejs/package.json @@ -52,7 +52,7 @@ "devDependencies": { "@types/node": "^22.20.5", "cmake-js": "^8.0.0", - "node-addon-api": "^8.5.0", + "node-addon-api": "^8.8.0", "typescript": "^7.0.2" } } \ No newline at end of file