build: prepare 4.5.0 release candidate 1

Co-authored-by: Copilot <223556219+Copilot@users.nply.github.com>
ci: install MessagePack C++ headers on Ubuntu
2026-11-09 21:07:17 +00:00 · 2026-11-03 22:48:09 -05:00 · 2026-11-03 22:44:50 -05:00 · 2026-11-03 22:41:04 -05:00 · 2026-11-03 22:40:02 -05:00 · 2026-10-29 22:46:57 -04:00
39 changed files with 2262 additions and 255 deletions
@@ -4,9 +4,9 @@ env:
  LLHTTP_VERSION: "v9.2.1"
  RESTINIO_VERSION: "0.7.10"
  EXPECTED_LITE_URL: "https://raw.githubusercontent.com/martinmoene/expected-lite/master/include/nonstd/expected.hpp"
-  UBUNTU_FULL_DEPS: "libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-dev"
-  UBUNTU_MESON_DEPS: "ninja-build libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-dev python3-setuptools python3-pip libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-dev libargon2-dev"
-  UBUNTU_MINIMAL_DEPS: "libncurses5-dev libreadline-dev nettle-dev libfmt-dev libgnutls28-dev libcppunit-dev libmsgpack-dev libargon2-dev"
+  UBUNTU_FULL_DEPS: "libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-cxx-dev libargon2-dev"
+  UBUNTU_MESON_DEPS: "ninja-build libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-dev python3-setuptools python3-pip libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-cxx-dev libargon2-dev"
+  UBUNTU_MINIMAL_DEPS: "libncurses5-dev libreadline-dev nettle-dev libfmt-dev libgnutls28-dev libcppunit-dev libmsgpack-cxx-dev libargon2-dev"
  DEBIAN13_DEPS: "git libtool pkg-config wget cmake libncurses5-dev libreadline-dev nettle-dev libasio-dev libgnutls28-dev libuv1-dev python3-pip python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev libfmt-dev libmsgpack-cxx-dev libargon2-dev"

 on: [push, pull_request]
@@ -33,6 +33,18 @@ jobs:
      - name: Build with Ninja
        run: cd build && ninja

+      - uses: dtolnay/rust-toolchain@stable
+
+      - name: Test Rust bindings
+        env:
+          LIBRARY_PATH: ${{ github.workspace }}/build
+          LD_LIBRARY_PATH: ${{ github.workspace }}/build
+          RUSTFLAGS: -L native=${{ github.workspace }}/build
+        run: |
+          cargo test --manifest-path rust/Cargo.toml --all-targets
+          cargo test --manifest-path rust/Cargo.toml --all-features --all-targets
+          cargo test --manifest-path rust/Cargo.toml --all-features --doc
+
  build-ubuntu-minimal:
    name: Ubuntu/GCC minimal build
    runs-on: ubuntu-latest
@@ -12,7 +12,7 @@ jobs:
          sudo apt update && \
          sudo apt install libncurses5-dev libreadline-dev nettle-dev \
          libgnutls28-dev libuv1-dev python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-          libfmt-dev libmsgpack-dev libargon2-dev libasio-dev \
+          libfmt-dev libmsgpack-cxx-dev libargon2-dev libasio-dev \
          llvm llvm-dev clang clang-tools && \
          sudo apt remove gcc g++

@@ -64,14 +64,26 @@ jobs:
      VCPKG_BINARY_SOURCES: clear;files,${{ github.workspace }}/vcpkg-binary-cache,readwrite
    steps:
      - uses: actions/checkout@v7
-      - uses: actions/setup-node@v6
+      - uses: actions/setup-node@v7
        with:
          node-version: '22'
          cache: npm
          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 }
@@ -136,7 +151,7 @@ jobs:
    runs-on: ubuntu-26.04
    steps:
      - uses: actions/checkout@v7
-      - uses: actions/setup-node@v6
+      - uses: actions/setup-node@v7
        with:
          node-version: '22'
      - uses: actions/download-artifact@v8
@@ -179,7 +194,7 @@ jobs:
        node: ['22', '24', '26']
    steps:
      - uses: actions/checkout@v7
-      - uses: actions/setup-node@v6
+      - uses: actions/setup-node@v7
        with:
          node-version: ${{ matrix.node }}
      - uses: actions/download-artifact@v8
@@ -4,10 +4,10 @@ permissions:
  contents: read

 on:
-  release:
-    types: [published]
  push:
+    branches: ['**']
  pull_request:
+  workflow_call:
  workflow_dispatch:

 jobs:
@@ -23,7 +23,7 @@ jobs:

      # Run locally with: pipx run cibuildwheel --platform linux python
      - name: Build wheels
-        uses: pypa/cibuildwheel@v4.3.4
+        uses: pypa/cibuildwheel@v4.3.5
        with:
          package-dir: python

@@ -82,19 +82,3 @@ jobs:
        with:
          name: opendht-wheels-win_amd64
          path: ./build/wheelhouse/*.whl
-
-  publish:
-    runs-on: ubuntu-latest
-    needs: [build_wheels, build_wheels_windows]
-    steps:
-    - uses: actions/download-artifact@v8
-      with:
-        pattern: opendht-wheels-*
-        path: dist
-        merge-multiple: true
-
-    - name: Publish package
-      if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags')
-      uses: pypa/gh-action-pypi-publish@release/v1
-      with:
-        password: ${{ secrets.PYPI_API_TOKEN }}
@@ -25,6 +25,12 @@ jobs:
      contents: read
      packages: read

+  python-wheels:
+    name: Build Python wheels
+    uses: ./.github/workflows/python-wheels.yml
+    permissions:
+      contents: read
+
  build-and-push-deps-image:
    name: Dependency Docker image
    runs-on: ubuntu-latest
@@ -593,6 +599,7 @@ jobs:
        uses: softprops/action-gh-release@v3
        with:
          draft: true
+          prerelease: ${{ contains(github.ref_name, '-') }}
          files: |
            downloaded_artifacts/windows_msi/*.msi
            downloaded_artifacts/static_dhtnode/**/*dhtnode.exe
@@ -602,12 +609,15 @@ jobs:
    name: Publish opendht to npm
    needs: create-github-release
    runs-on: ubuntu-26.04
+    environment:
+      name: npm
+      url: https://www.npmjs.com/package/opendht
    permissions:
      contents: read
      id-token: write
    steps:
      - uses: actions/checkout@v7
-      - uses: actions/setup-node@v6
+      - uses: actions/setup-node@v7
        with:
          node-version: '24'
          registry-url: https://registry.npmjs.org/
@@ -619,7 +629,6 @@ jobs:
      - name: Publish verified archive
        working-directory: nodejs
        env:
-          NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
          RELEASE_TAG: ${{ github.ref_name }}
        run: |
          dist_tag=latest
@@ -627,3 +636,51 @@ jobs:
            dist_tag=next
          fi
          npm publish ./dist/*.tgz --access public --provenance --tag "$dist_tag"
+
+  publish-pypi:
+    name: Publish opendht to PyPI
+    needs: [create-github-release, python-wheels]
+    runs-on: ubuntu-latest
+    environment:
+      name: pypi
+      url: https://pypi.org/p/opendht
+    permissions:
+      contents: read
+      id-token: write
+    steps:
+      - uses: actions/download-artifact@v8
+        with:
+          pattern: opendht-wheels-*
+          path: dist
+          merge-multiple: true
+      - name: Publish wheels
+        uses: pypa/gh-action-pypi-publish@release/v1
+
+  publish-crate:
+    name: Publish opendht to crates.io
+    needs: create-github-release
+    runs-on: ubuntu-latest
+    environment:
+      name: crates-io
+      url: https://crates.io/crates/opendht
+    permissions:
+      contents: read
+      id-token: write
+    steps:
+      - uses: actions/checkout@v7
+      - uses: taiki-e/install-action@v2
+        with:
+          tool: cargo-edit
+      - name: Set crate version from release tag
+        working-directory: rust
+        env:
+          RELEASE_TAG: ${{ github.ref_name }}
+        run: cargo set-version "${RELEASE_TAG#v}"
+      - name: Authenticate with crates.io
+        id: auth
+        uses: rust-lang/crates-io-auth-action@v1
+      - name: Publish crate
+        working-directory: rust
+        env:
+          CARGO_REGISTRY_TOKEN: ${{ steps.auth.outputs.token }}
+        run: cargo publish --allow-dirty
@@ -133,7 +133,7 @@ the assembled npm archive on Node 22, 24, and 26 before release attachment.
 Enable `OPENDHT_C=ON` to build the C ABI. With `BUILD_TESTING=ON`, CTest runs
 `test_c`; Meson registers the same test as `CBindings` when `c` and `tests` are
 enabled. Rust regression and callback-lifetime tests run with Cargo; see
-[the Rust binding guide](rust/README.md) for build and library-path settings.
+[the Rust build guide](rust/BUILD.md) for build and library-path settings.

 `dht_infohash` and `dht_pkid` have `uint32_t` alignment, matching the native hash
 types, while retaining their 20-byte and 32-byte sizes. Rebuild consumers against
@@ -10,10 +10,10 @@ if(POLICY CMP0167)
 endif()

 set (opendht_VERSION_MAJOR 4)
-set (opendht_VERSION_MINOR 4.99)
+set (opendht_VERSION_MINOR 5.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
-set (PACKAGE_VERSION ${opendht_VERSION})
-set (VERSION "${opendht_VERSION}")
+set (PACKAGE_VERSION "${opendht_VERSION}-rc.1")
+set (VERSION "${PACKAGE_VERSION}")
 project (opendht
    VERSION ${opendht_VERSION}
    DESCRIPTION "OpenDHT: A lightweight C++17 Distributed Hash Table library."
@@ -421,7 +421,7 @@ else ()
 endif ()
 set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_NO_BOOST -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")

-add_definitions(-DPACKAGE_VERSION="${opendht_VERSION}")
+add_definitions(-DPACKAGE_VERSION="${PACKAGE_VERSION}")

 if (ASIO_INCLUDE_DIR)
    include_directories (SYSTEM "${ASIO_INCLUDE_DIR}")
@@ -4,7 +4,10 @@
    <a id="user-content-opendht-" class="anchor" href="/savoirfairelinux/opendht/blob/master/README.md#opendht-" aria-hidden="true"></a>OpenDHT
 </h1>

-![PyPI - Version](https://img.shields.io/pypi/v/opendht?style=flat)
+[![PyPI - Version](https://img.shields.io/pypi/v/opendht?style=flat)](https://pypi.org/project/opendht/)
+[![crates.io - Version](https://img.shields.io/crates/v/opendht?style=flat)](https://crates.io/crates/opendht)
+[![npm - Version](https://img.shields.io/npm/v/opendht?style=flat)](https://www.npmjs.com/package/opendht)
+[![Docker Image](https://img.shields.io/badge/Docker-ghcr.io-2496ED?style=flat&logo=docker&logoColor=white)](https://github.com/savoirfairelinux/opendht/pkgs/container/opendht%2Fopendht)

 A lightweight C++17 Distributed Hash Table implementation.

@@ -17,7 +20,7 @@ Values are distributed over the network, with redundancy.
 * Public key cryptography layer providing optional data signature and encryption (using GnuTLS)
 * IPv4 and IPv6 support
 * Clean and powerful **C++17** map API
- * Bindings for **C, Rust & Python 3**
+ * Bindings for **C, Rust, Node.js & Python 3**
 * REST API with optional HTTP client+server with push notification support

 ## Documentation
@@ -8,6 +8,7 @@ RUN cd opendht && mkdir build && cd build \
 	&& cmake -GNinja .. -DCMAKE_INSTALL_PREFIX=/usr \
 	-DBUILD_TESTING=Off \
 	-DCMAKE_INTERPROCEDURAL_OPTIMIZATION=On \
+	-DCMAKE_BUILD_TYPE=RelWithDebInfo \
 	-DOPENDHT_C=On \
 	-DOPENDHT_PEER_DISCOVERY=On \
 	-DOPENDHT_PYTHON=On \
@@ -10,6 +10,7 @@ ENV DESTDIR=/install
 RUN cd opendht && mkdir build && cd build \
 	&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr \
                -DCMAKE_INTERPROCEDURAL_OPTIMIZATION=On \
+                -DCMAKE_BUILD_TYPE=RelWithDebInfo \
                -DBUILD_TESTING=Off \
                -DOPENDHT_C=On \
                -DOPENDHT_PEER_DISCOVERY=On \
@@ -8,6 +8,7 @@ RUN cd opendht && mkdir build && cd build \
 	&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr \
 				-DBUILD_TESTING=Off \
 				-DCMAKE_INTERPROCEDURAL_OPTIMIZATION=On \
+				-DCMAKE_BUILD_TYPE=RelWithDebInfo \
 				-DOPENDHT_C=On \
 				-DOPENDHT_PEER_DISCOVERY=On \
 				-DOPENDHT_PYTHON=On \
@@ -7,6 +7,7 @@ COPY . opendht
 RUN cd opendht && mkdir build && cd build \
 	&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr \
 				-DCMAKE_INTERPROCEDURAL_OPTIMIZATION=On \
+				-DCMAKE_BUILD_TYPE=RelWithDebInfo \
 				-DOPENDHT_C=On \
 				-DOPENDHT_PEER_DISCOVERY=On \
 				-DOPENDHT_PYTHON=On \
@@ -8,6 +8,7 @@ RUN cd opendht && mkdir build && cd build \
 	&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr \
 	-DBUILD_TESTING=Off \
 	-DCMAKE_INTERPROCEDURAL_OPTIMIZATION=On \
+	-DCMAKE_BUILD_TYPE=RelWithDebInfo \
 	-DOPENDHT_C=On \
 	-DOPENDHT_PEER_DISCOVERY=On \
 	-DOPENDHT_PYTHON=On \
@@ -19,7 +19,9 @@
 #include <mutex>

 namespace dht {
-enum class PushType { None = 0, Android, iOS, UnifiedPush };
+// Appended only: the values are serialized as integers by MSGPACK_ADD_ENUM,
+// so reordering would break compatibility with deployed proxies.
+enum class PushType { None = 0, Android, iOS, UnifiedPush, Huawei };
 }
 MSGPACK_ADD_ENUM(dht::PushType)

@@ -132,6 +134,7 @@ public:
        PushStats androidPush;
        PushStats iosPush;
        PushStats unifiedPush;
+        PushStats huaweiPush;

        /** Average requests per second */
        double requestRate {0};
@@ -397,6 +400,7 @@ private:
    PushStats androidPush_;
    PushStats iosPush_;
    PushStats unifiedPush_;
+    PushStats huaweiPush_;

    // Thread-safe access to listeners map.
    std::mutex lockListener_;
@@ -5,7 +5,7 @@ project(
    'cython',
    license: 'MIT',
    license_files: 'LICENSE',
-    version: '4.4.99',
+    version: '4.5.0-rc.1',
    meson_version: '>=1.1.0',
    default_options: ['cpp_std=c++20', 'warning_level=3'],
 )
@@ -55,7 +55,12 @@ if host_machine.system() == 'linux' and io_uring.found()
 endif
 conf_data = configuration_data()

-add_project_arguments('-DMSGPACK_NO_BOOST', '-DASIO_STANDALONE', language: 'cpp')
+add_project_arguments(
+    '-DMSGPACK_NO_BOOST',
+    '-DASIO_STANDALONE',
+    '-DPACKAGE_VERSION="' + meson.project_version() + '"',
+    language: 'cpp',
+)
 if io_uring.found()
    add_project_arguments('-DASIO_HAS_IO_URING', '-DASIO_DISABLE_EPOLL', language: 'cpp')
    conf_data.set('iouring_lib', ', liburing')
@@ -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")
@@ -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,43 +32,16 @@ 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
 installation does not require a compiler or a system OpenDHT installation.

-## Example Tool
-
-After building, start the TypeScript example from the repository root:
-
-```sh
-npm --prefix nodejs run dhtnode -- --port 4222 --bootstrap bootstrap.jami.net
-```
-
-The source is [examples/dhtnode.ts](examples/dhtnode.ts). `npm run build` compiles
-it to JavaScript; run `npm run build:examples` after editing just the example.
-Bootstrap is optional;
-`--bootstrap-port` selects its UDP port. Use `--port 0` for a free local port.
-
-```text
-listen greeting
-put greeting hello world
-get greeting
-sign greeting signed hello
-cancel greeting
-quit
-```
-
-Keys are text labels, hashed automatically; values are UTF-8 text. The tool
-generates a fresh EC identity on each run and prints IPv4/IPv6 connectivity
-changes. Operations run concurrently, leaving the prompt responsive; dependent
-commands should be entered after their preceding operation completes. EOF waits
-for pending operations, then cancels listeners and closes the runner. Ctrl-C,
-SIGTERM, and `quit` close immediately, rejecting pending operations without
-printing cancellation errors. `--help` lists all options. Archives include both
-the TypeScript source and compiled `examples/dhtnode.js`, runnable without
-TypeScript or development dependencies, including from inside `node_modules`.
-
 ## Usage

 ```js
@@ -84,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 {
@@ -103,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'`.
@@ -113,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.
@@ -133,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

@@ -213,13 +231,6 @@ The [Node.js package workflow](../.github/workflows/nodejs-package.yml) builds:
 | macOS ARM64 and x64 | macOS 15 |
 | Windows x64 | Windows Server 2022 / MSVC |

-The Linux build uses `ghcr.io/savoirfairelinux/opendht/opendht-deps-noble`,
-built from [the Noble Dockerfile](../docker/DockerfileDepsNoble), restored from
-the Ubuntu 24.04 recipe in git history. The release workflow publishes this
-multi-architecture image alongside the other Docker images. Node builds use its
-published `latest` tag. simdutf is linked statically because Ubuntu 24.04 does not
-provide its development package.
-
 CI combines platform prebuilds into one npm archive and verifies that archive on
 Node 22, 24, and 26 on each platform, including Ubuntu 24.04 and 26.04. Branch
 pushes, pull requests, and manual runs build and test without publishing.
@@ -228,28 +239,5 @@ does not remove OS/libc or native dependency compatibility requirements. Review
 the included dependency licenses and source-distribution obligations before
 redistributing binaries.

-## Releases
-
-Pushing a `v*` tag starts [the release workflow](../.github/workflows/release-package.yml),
-which calls the Node.js package workflow. The npm archive version is taken from
-the tag without its leading `v`, for example `v4.4.99` becomes `opendht@4.4.99`.
-After all Node install tests and Windows builds pass, the release workflow attaches
-the verified npm archive alongside the Windows assets to the draft GitHub Release,
-then publishes that same archive to npm with provenance. Stable versions use the
-`latest` dist-tag; prerelease versions such as `v4.5.0-rc.1` use `next`.
-Publishing to npm happens on the tag push, not when the GitHub draft is published.
-Published npm versions cannot be overwritten; use a new version tag for changes.
-
-Configure an npm trusted publisher for the `opendht` package with GitHub owner
-`savoirfairelinux`, repository `opendht`, and workflow filename `release-package.yml`
-(not `nodejs-package.yml`). Allow direct `npm publish`. Publication runs on a
-GitHub-hosted runner with OIDC, so no npm token is needed once this is configured.
-
-For the first publication, before the npm package exists, set the repository's
-`NPM_TOKEN` secret to a granular npm token with publishing permission and 2FA
-bypass. After the first successful tag release, configure the trusted publisher
-in the npm package settings, then remove `NPM_TOKEN`. The publishing job supports
-the token for this bootstrap and prefers OIDC when available.
-
-Query filters, async iterators, AbortSignal support, certificate trust-store
-validation, CA issuance controls, and proxy controls remain future work.
+Query filters, AbortSignal support, certificate trust-store validation, CA
+issuance controls, and runtime proxy switching remain future work.
@@ -90,7 +90,7 @@ valueToJS(Napi::Env env, const dht::Value& value)
    result.Set("data", Napi::Buffer<uint8_t>::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<std::shared_ptr<const dht::Value>> values;
+    Napi::FunctionReference onValue;
+    std::shared_ptr<std::atomic_bool> 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<RunnerState> 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<std::atomic_bool>(false);
+            PendingOperation pending {deferred, true, {}, {}, cancelled};
+            if (stream)
+                pending.onValue = Napi::Persistent(info[1].As<Napi::Function>());
+            state_->operations.emplace(operation, std::move(pending));
            const std::weak_ptr<RunnerState> state = state_;
            try {
                state_->runner.get(key,
-                                   dht::GetCallback(
-                                       [state, operation](const std::vector<std::shared_ptr<dht::Value>>& 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<std::shared_ptr<dht::Value>>& 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<RunnerState> 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
@@ -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}`);
@@ -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<T> extends AsyncIterableIterator<T, undefined>, Listener, AsyncDisposable {
+    return(): Promise<IteratorResult<T, undefined>>;
+    throw(error?: unknown): Promise<IteratorResult<T, undefined>>;
+}
+
+export interface ListenerIterator extends ValueIterator<ListenEvent> { }
+
 export type NodeStatus = 'disconnected' | 'connecting' | 'connected';

 export interface ConnectivityStatus {
@@ -92,10 +106,12 @@ export class DhtRunner extends EventEmitter<RunnerEvents> implements AsyncDispos
    getNodeId(): string;
    getStatus(): ConnectivityStatus;
    bootstrap(host: string, port: number): void;
-    get(key: string): Promise<Value[]>;
+    get(key: string): ValueIterator<Value>;
+    getAll(key: string): Promise<Value[]>;
    put(key: string, data: Uint8Array, options?: PutOptions): Promise<boolean>;
    putSigned(key: string, data: Uint8Array, options?: PutOptions): Promise<boolean>;
    putEncrypted(key: string, data: Uint8Array, recipient: PublicKey, options?: PutOptions): Promise<boolean>;
+    listen(key: string): ListenerIterator;
    listen(key: string, callback: (value: Value, expired: boolean) => void): Listener;
    close(): Promise<void>;
    ref(): this;
@@ -1,10 +1,15 @@
 import { EventEmitter } from 'node:events';
+import { constants } from 'node:os';
 import { dlopen } from 'node:process';
 import load from 'node-gyp-build';
 import createCrypto from './crypto.js';

 const binding = { exports: {} };
-dlopen(binding, process.env.OPENDHT_NODE_BINDING || load.path(import.meta.dirname));
+const flags = process.platform === 'linux'
+    ? constants.dlopen.RTLD_LAZY | constants.dlopen.RTLD_DEEPBIND
+    : undefined;
+dlopen(binding, process.env.OPENDHT_NODE_BINDING || load.path(import.meta.dirname),
+    flags);
 const native = binding.exports;
 const { PublicKey, PrivateKey, Certificate, Identity, handle } = createCrypto(native);

@@ -17,6 +22,7 @@ class DhtRunner extends EventEmitter {
    #closing;
    #listeners = 0;
    #pending = 0;
+    #iterators = new Set();

    constructor() {
        super();
@@ -45,6 +51,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 +64,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 +79,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 +100,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(); }
@@ -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"
    }
 }
@@ -27,6 +27,9 @@ function copyNotices(dependencies) {
            name = match[2];
            notices = fs.readdirSync(match[1]).filter(file => /^(COPYING|LICENSE|COPYRIGHT)([._-]|$)/i.test(file))
                .map(file => path.join(match[1], file));
+        } else if (process.platform === 'linux' && /^libllhttp\.so(?:\.|$)/.test(path.basename(real))) {
+            name = 'llhttp';
+            notices = [path.join(repository, 'resources', 'llhttp-LICENSE-MIT')];
        } else if (process.platform === 'linux') {
            let owner;
            try { owner = run('dpkg-query', ['-S', real]); }
@@ -1,6 +1,7 @@
 import assert from 'node:assert/strict';
 import { execFile } from 'node:child_process';
 import { EventEmitter, once } from 'node:events';
+import { createServer } from 'node:http';
 import { test } from 'node:test';
 import { fileURLToPath } from 'node:url';
 import { promisify } from 'node:util';
@@ -29,6 +30,7 @@ test('runner lifecycle and argument validation', async () => {
    const runner = new DhtRunner();
    assert.equal(runner.isRunning(), false);
    assert.throws(() => runner.get(hash('idle')), /running/i);
+    assert.throws(() => runner.getAll(hash('idle')), /running/i);
    assert.throws(() => runner.run({ port: -1 }), RangeError);
    assert.throws(() => runner.run({ port: 65536 }), RangeError);
    assert.throws(() => runner.run({ port: 1.5 }), RangeError);
@@ -136,7 +138,7 @@ test('two loopback nodes exchange binary values and cancel listeners', { timeout
        assert.equal(value.isSigned, false);
        assert.equal(value.isEncrypted, false);
        value.data.fill(7);
-        const values = await reader.get(key);
+        const values = await reader.getAll(key);
        retainedValue = values.find(candidate => candidate.id === value.id);
        assert.ok(retainedValue);
        assert.deepEqual(retainedValue.data, payload);
@@ -153,7 +155,7 @@ test('close settles pending operations and suppresses queued listener deliveries
    const runner = startNode();
    let deliveries = 0;
    const subscription = runner.listen(hash('closing'), () => deliveries++);
-    const pending = runner.get(hash('unreachable'));
+    const pending = runner.getAll(hash('unreachable'));
    const settled = assert.rejects(pending, { code: 'ERR_DHT_CLOSED' });
    await runner.close();
    await settled;
@@ -161,6 +163,236 @@ test('close settles pending operations and suppresses queued listener deliveries
    assert.equal(deliveries, 0);
 });

+test('get iterators settle pending reads on cancellation and reject on runner close', { timeout: 10000 }, async () => {
+    const runner = startNode();
+    try {
+        const cancelled = runner.get(hash('cancel get'));
+        assert.equal(cancelled[Symbol.asyncIterator](), cancelled);
+        const pending = [cancelled.next(), cancelled.next()];
+        await cancelled.return();
+        cancelled.cancel();
+        assert.deepEqual(await Promise.all(pending), [
+            { value: undefined, done: true },
+            { value: undefined, done: true },
+        ]);
+        assert.deepEqual(await cancelled.next(), { value: undefined, done: true });
+
+        const closing = runner.get(hash('close get'));
+        const rejected = assert.rejects(closing.next(), { code: 'ERR_DHT_CLOSED' });
+        await runner.close();
+        await rejected;
+        await assert.rejects(closing.next(), { code: 'ERR_DHT_CLOSED' });
+    } finally {
+        await runner.close();
+    }
+});
+
+test('async listeners deliver queued values and cancel on loop exit', { timeout: 15000 }, async () => {
+    const writer = startNode();
+    const reader = startNode();
+    const key = hash('async listener');
+    let iterator;
+    try {
+        writer.bootstrap('127.0.0.1', reader.getBoundPort());
+        reader.bootstrap('127.0.0.1', writer.getBoundPort());
+        iterator = reader.listen(key);
+        assert.equal(iterator[Symbol.asyncIterator](), iterator);
+        const pending = iterator.next();
+        assert.equal(await writer.put(key, Buffer.from('first'), { id: 1n }), true);
+        const first = await pending;
+        assert.equal(first.done, false);
+        assert.equal(first.value.expired, false);
+        assert.equal(first.value.value.id, 1n);
+        assert.equal(first.value.value.data.toString(), 'first');
+
+        assert.equal(await writer.put(key, Buffer.from('second'), { id: 2n }), true);
+        for await (const { value, expired } of iterator) {
+            assert.equal(expired, false);
+            assert.equal(value.id, 2n);
+            assert.equal(value.data.toString(), 'second');
+            break;
+        }
+        assert.deepEqual(await iterator.next(), { value: undefined, done: true });
+        iterator.cancel();
+    } finally {
+        iterator?.cancel();
+        await Promise.all([writer.close(), reader.close()]);
+    }
+});
+
+test('async listener cancellation settles concurrent reads and drops buffered values', { timeout: 10000 }, async () => {
+    const runner = startNode();
+    const peer = startNode();
+    const key = hash('async cancellation');
+    let iterator;
+    try {
+        runner.bootstrap('127.0.0.1', peer.getBoundPort());
+        peer.bootstrap('127.0.0.1', runner.getBoundPort());
+        iterator = runner.listen(key);
+        const pending = [iterator.next(), iterator.next()];
+        iterator.cancel();
+        iterator.cancel();
+        assert.deepEqual(await Promise.all(pending), [
+            { value: undefined, done: true },
+            { value: undefined, done: true },
+        ]);
+        assert.equal(await runner.put(key, Buffer.from('cancelled'), { id: 3n }), true);
+        assert.deepEqual(await iterator.next(), { value: undefined, done: true });
+
+        const buffered = runner.listen(key);
+        await runner.getAll(key);
+        await buffered.return();
+        assert.deepEqual(await buffered.next(), { value: undefined, done: true });
+    } finally {
+        iterator?.cancel();
+        await Promise.all([runner.close(), peer.close()]);
+    }
+});
+
+test('async listener throw, disposal, and runner close settle pending reads', { timeout: 10000 }, async () => {
+    const runner = startNode();
+    try {
+        const thrown = runner.listen(hash('async throw'));
+        const pendingThrow = thrown.next();
+        const failure = new Error('consumer failed');
+        await assert.rejects(thrown.throw(failure), error => error === failure);
+        assert.deepEqual(await pendingThrow, { value: undefined, done: true });
+        assert.deepEqual(await thrown.next(), { value: undefined, done: true });
+
+        const disposed = runner.listen(hash('async dispose'));
+        const pendingDispose = disposed.next();
+        await disposed[Symbol.asyncDispose]();
+        assert.deepEqual(await pendingDispose, { value: undefined, done: true });
+
+        const closing = runner.listen(hash('async close'));
+        const pendingClose = [closing.next(), closing.next()];
+        const close = runner.close();
+        assert.equal(runner.close(), close);
+        assert.deepEqual(await Promise.all(pendingClose), [
+            { value: undefined, done: true },
+            { value: undefined, done: true },
+        ]);
+        await close;
+        assert.deepEqual(await closing.next(), { value: undefined, done: true });
+        assert.throws(() => runner.listen(hash('closed iterator')), /closed/i);
+    } finally {
+        await runner.close();
+    }
+});
+
+test('proxy configuration validates option types before starting', async () => {
+    const runner = new DhtRunner();
+    try {
+        assert.throws(() => runner.run({ proxyServer: 42 }), TypeError);
+        assert.throws(() => runner.run({ proxyUserAgent: false }), TypeError);
+        assert.throws(() => runner.run({ proxyUserAgent: 'client\r\nInjected: header' }), TypeError);
+        assert.throws(() => runner.run({ proxyServer: 'http://127.0.0.1\0' }), TypeError);
+        assert.throws(() => runner.run({ proxyServer: 'ftp://127.0.0.1' }), /http/i);
+        assert.throws(() => runner.run({ proxyServer: 'not a URL' }), /url/i);
+        assert.equal(runner.isRunning(), false);
+        runner.run({ port: 0, ipv4: '127.0.0.1', ipv6: '::1', proxyServer: '' });
+        assert.ok(runner.getBoundPort() > 0);
+    } finally {
+        await runner.close();
+    }
+});
+
+test('proxy configuration routes operations through HTTP with a custom user agent', { timeout: 15000 }, async () => {
+    const runner = new DhtRunner();
+    const requests = [];
+    const values = new Map();
+    let streamNextGet = false;
+    let streamResponse;
+    const server = createServer(async (request, response) => {
+        response.setHeader('content-type', 'application/json');
+        if (request.url === '/') {
+            response.end(JSON.stringify({
+                node_id: hash('proxy fixture'),
+                ipv4: { good: 1 },
+                ipv6: {},
+                public_ip: '127.0.0.1',
+            }));
+            return;
+        }
+        requests.push({ method: request.method, url: request.url, userAgent: request.headers['user-agent'] });
+        if (request.url === `/key/${hash('proxy failure')}`) {
+            response.statusCode = 500;
+            response.end();
+            return;
+        }
+        if (request.method === 'POST') {
+            const chunks = [];
+            for await (const chunk of request) chunks.push(chunk);
+            const value = JSON.parse(Buffer.concat(chunks).toString());
+            values.set(request.url, value);
+            response.end(JSON.stringify(value));
+        } else {
+            const value = values.get(request.url);
+            const body = value ? `${JSON.stringify(value)}\n` : '';
+            if (streamNextGet) {
+                streamNextGet = false;
+                streamResponse = response;
+                response.write(body);
+            } else response.end(body);
+        }
+    });
+    server.listen(0, '127.0.0.1');
+    try {
+        await once(server, 'listening');
+        runner.run({
+            proxyServer: `http://127.0.0.1:${server.address().port}`,
+            proxyUserAgent: 'opendht-node-test/1.0',
+            persistPath: '',
+        });
+        const key = hash('proxy configuration');
+        const payload = Buffer.from([0, 255, 42]);
+        assert.equal(await runner.put(key, payload, { id: 42n }), true);
+        const received = await runner.getAll(key);
+        assert.equal(received.length, 1);
+        assert.equal(received[0].id, 42n);
+        assert.deepEqual(received[0].data, payload);
+
+        streamNextGet = true;
+        const iterator = runner.get(key);
+        const first = await iterator.next();
+        assert.equal(first.done, false);
+        assert.equal(first.value.id, 42n);
+        assert.deepEqual(first.value.data, payload);
+        assert.equal(streamResponse.writableEnded, false);
+        streamResponse.end();
+        assert.deepEqual(await iterator.next(), { value: undefined, done: true });
+        assert.deepEqual(await iterator.next(), { value: undefined, done: true });
+        assert.deepEqual(requests.map(request => request.method), ['POST', 'GET', 'GET']);
+        assert.ok(requests.every(request => request.url === `/key/${key}`));
+        assert.ok(requests.every(request => request.userAgent === 'opendht-node-test/1.0'));
+
+        const empty = runner.get(hash('proxy empty'));
+        assert.deepEqual(await empty.next(), { value: undefined, done: true });
+        const failed = runner.get(hash('proxy failure'));
+        await assert.rejects(failed.next(), { code: 'ERR_DHT_GET_FAILED' });
+        await assert.rejects(runner.getAll(hash('proxy failure')), { code: 'ERR_DHT_GET_FAILED' });
+
+        const all = [];
+        for await (const value of runner.get(key)) all.push(value);
+        assert.equal(all.length, 1);
+        assert.deepEqual(all[0].data, payload);
+
+        streamNextGet = true;
+        const cancelled = runner.get(key);
+        for await (const value of cancelled) {
+            assert.equal(value.id, 42n);
+            break;
+        }
+        assert.equal(streamResponse.writableEnded, false);
+        assert.deepEqual(await cancelled.next(), { value: undefined, done: true });
+        streamResponse.end();
+    } finally {
+        streamResponse?.end();
+        await runner.close();
+        await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve()));
+    }
+});
+
 test('worker termination cleans up native runners', { timeout: 10000 }, async () => {
    const source = `
        import { parentPort } from 'node:worker_threads';
@@ -280,7 +512,7 @@ test('dhtnode example handles commands and quit while a get is pending', { timeo
        import { DhtRunner } from ${JSON.stringify(moduleURL)};
        const close = DhtRunner.prototype.close;
        let rejectGet;
-        DhtRunner.prototype.get = function () {
+        DhtRunner.prototype.getAll = function () {
            console.log('Get pending');
            return new Promise((resolve, reject) => { rejectGet = reject; });
        };
@@ -66,12 +66,12 @@ test('signed and encrypted values round-trip between authenticated nodes', { tim
        reader.bootstrap('127.0.0.1', writer.getBoundPort());
        const signedKey = hash('nodejs signed value');
        assert.equal(await writer.putSigned(signedKey, payload, { id: 41n }), true);
-        const signed = await reader.get(signedKey);
+        const signed = await reader.getAll(signedKey);
        assert.ok(signed.some(value => value.data.equals(payload) && value.isSigned));
        assert.equal(signed[0].owner, writerIdentity.certificate.getLongId());
        const encryptedKey = hash('nodejs encrypted value');
        assert.equal(await writer.putEncrypted(encryptedKey, payload, readerIdentity.certificate.getPublicKey(), { id: 42n }), true);
-        const decrypted = await reader.get(encryptedKey);
+        const decrypted = await reader.getAll(encryptedKey);
        assert.ok(decrypted.some(value => value.data.equals(payload)));
        assert.throws(() => writer.putEncrypted(encryptedKey, payload, {}), TypeError);
        assert.throws(() => writer.putEncrypted(encryptedKey, payload, writerIdentity.certificate.getPublicKey()), /RSA/i);
@@ -1,5 +1,5 @@
 import { Certificate, DhtRunner, Identity, PrivateKey, PublicKey, hash, version } from '../index.js';
-import type { ConnectivityStatus, NodeStatus, Value } from '../index.js';
+import type { ConnectivityStatus, ListenEvent, ListenerIterator, NodeStatus, Value, ValueIterator } from '../index.js';

 async function exercise(): Promise<void> {
    const privateKey: PrivateKey = await PrivateKey.generate({ algorithm: 'rsa', bits: 2048 });
@@ -22,7 +22,12 @@ async function exercise(): Promise<void> {
    const accepted: boolean = await runner.put(key, Buffer.from(version), { id: 42n, permanent: true });
    const signed: boolean = await runner.putSigned(key, plaintext);
    const encrypted: boolean = await runner.putEncrypted(key, plaintext, publicKey);
-    const values: Value[] = await runner.get(key);
+    const values: Value[] = await runner.getAll(key);
+    const get: ValueIterator<Value> = runner.get(key);
+    for await (const value of get) {
+        const payload: Buffer = value.data;
+        void payload;
+    }
    const listener = runner.listen(key, (value, expired) => {
        const payload: Buffer = value.data;
        const id: bigint = value.id;
@@ -30,7 +35,21 @@ async function exercise(): Promise<void> {
    });
    runner.ref().unref();
    listener.cancel();
+    const iterator: ListenerIterator = runner.listen(key);
+    for await (const event of iterator) {
+        const typed: ListenEvent = event;
+        const value: Value = typed.value;
+        const expired: boolean = typed.expired;
+        void [value, expired];
+        break;
+    }
+    iterator.cancel();
+    await iterator.return();
+    await iterator[Symbol.asyncDispose]();
    await runner[Symbol.asyncDispose]();
+    const proxy = new DhtRunner();
+    proxy.run({ proxyServer: 'https://proxy.example.com', proxyUserAgent: 'typed-client/1.0' });
+    await proxy.close();
    void [accepted, values, valid, signed, encrypted, status];
 }

@@ -0,0 +1,22 @@
+This software is licensed under the MIT License.
+
+Copyright Fedor Indutny, 2018.
+
+Permission is hereby granted, free of charge, to any person obtaining a
+copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to permit
+persons to whom the Software is furnished to do so, subject to the
+following conditions:
+
+The above copyright notice and this permission notice shall be included
+in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+USE OR OTHER DEALINGS IN THE SOFTWARE.
@@ -0,0 +1,70 @@
+# Building and Testing the Rust Crate
+
+The `opendht` crate links to the native OpenDHT C library. Cargo does not build
+or bundle that library. Install Rust and the native dependencies described in
+the [OpenDHT build guide](../BUILD.md).
+
+## Install the Native Library
+
+From the root of an OpenDHT source checkout, enable the C bindings:
+
+```sh
+cmake -S . -B build-bindings -DOPENDHT_C=ON -DBUILD_TESTING=ON
+cmake --build build-bindings
+ctest --test-dir build-bindings -R test_c --output-on-failure
+cmake --install build-bindings
+```
+
+Installation may require administrator privileges, depending on the install
+prefix. Set `CMAKE_INSTALL_PREFIX` when configuring to choose another location.
+The linker and runtime loader must be able to find the installed libraries.
+Applications can then depend on `opendht` from crates.io as shown in the
+[README](README.md).
+
+## Use an Uninstalled Native Build
+
+You can use the native build directly by setting its library search paths.
+For example, from the repository root on macOS:
+
+```sh
+export LIBRARY_PATH="$PWD/build-bindings${LIBRARY_PATH:+:$LIBRARY_PATH}"
+export DYLD_LIBRARY_PATH="$PWD/build-bindings${DYLD_LIBRARY_PATH:+:$DYLD_LIBRARY_PATH}"
+export RUSTFLAGS="-L native=$PWD/build-bindings"
+cargo test --manifest-path rust/Cargo.toml --all-targets
+```
+
+On Linux use `LD_LIBRARY_PATH` instead of `DYLD_LIBRARY_PATH`; on Windows add
+the native DLL directory to `PATH`. For an application outside this checkout,
+use absolute paths to the native build and run Cargo in the application directory.
+
+When linking an AddressSanitizer native build, use the same compiler runtime.
+For example, with Clang:
+
+```sh
+export RUSTFLAGS="-L native=$PWD/build-bindings -C linker=clang -C link-arg=-fsanitize=address"
+```
+
+## Build and Test the Crate
+
+With the native library available, run from the `rust` directory:
+
+```sh
+cargo build --all-targets
+cargo test --all-targets
+cargo test --doc
+cargo test --all-features --all-targets
+cargo test --all-features --doc
+```
+
+## Run the Examples
+
+From the `rust` directory:
+
+```sh
+cargo run --example dhtnode
+cargo run --features async --example dhtnode_async
+cargo run --features serde --example dhtnode_async
+```
+
+The callback example contacts a public bootstrap server. The async example uses
+two loopback nodes; enabling `serde` also demonstrates typed MessagePack records.
@@ -1,11 +1,30 @@
 [package]
 name = "opendht"
-version = "0.2.0"
+version = "0.6.0"
 authors = ["Sébastien Blin <sebastien.blin@savoirfairelinux.com>", "Adrien Beraud <adrien.beraud@savoirfairelinux.com>"]
 edition = "2024"
+description = "Rust bindings for OpenDHT"
+license = "MIT"
+license-file = "../LICENSE"
+repository = "https://github.com/savoirfairelinux/opendht"
+
+[features]
+async = ["dep:futures-channel", "dep:futures-core"]
+serde = ["async", "dep:serde", "dep:rmp-serde"]
+
+[[example]]
+name = "dhtnode_async"
+required-features = ["async"]

 [dependencies]
 libc="0.2"
 os_socketaddr="0.2.3"
+futures-channel = { version = "0.3", optional = true }
+futures-core = { version = "0.3", optional = true }
+serde = { version = "1", optional = true }
+rmp-serde = { version = "1", optional = true }

-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[dev-dependencies]
+futures = "0.3"
+serde = { version = "1", features = ["derive"] }
+serde_bytes = "0.11"
@@ -1,97 +1,152 @@
-# Rust Bindings for OpenDHT
+# OpenDHT: a Distributed Hash Table

-## Build and Test
+OpenDHT is a Kademlia-based distributed hash table for storing and retrieving
+small values over a peer-to-peer network without a central server. Common uses
+include peer discovery, service announcements, and decentralized applications.

-Install Rust and the native OpenDHT C library. From this directory:
+The `opendht` crate provides Rust bindings to [OpenDHT](https://opendht.net), with
+callback-based and runtime-independent async APIs. Capabilities include signed
+and encrypted values, filtered reads, metadata queries, and typed MessagePack
+records through Serde.

-```sh
-cargo build --all-targets
-cargo test --all-targets
-cargo test --doc
-cargo run --example dhtnode
+## Get Started
+
+Add the crate to your dependencies:
+
+```toml
+[dependencies]
+opendht = "0.6"
 ```

-To use an uninstalled native build, set its library search paths. For example,
-from the repository root on macOS:
+The native OpenDHT library with C bindings must be installed separately; it is
+not bundled with the crate. See the
+[build guide](https://github.com/savoirfairelinux/opendht/blob/master/rust/BUILD.md)
+for native installation, library paths, and source builds.

-```sh
-cmake -S . -B build-bindings -DOPENDHT_C=ON -DBUILD_TESTING=ON
-cmake --build build-bindings
-ctest --test-dir build-bindings -R test_c --output-on-failure
-LIBRARY_PATH="$PWD/build-bindings" \
-DYLD_LIBRARY_PATH="$PWD/build-bindings" \
-RUSTFLAGS="-L native=$PWD/build-bindings" \
-cargo test --manifest-path rust/Cargo.toml --all-targets
-```
+`DhtRunner` manages a node, `InfoHash` identifies a DHT key, and `Value` contains
+the payload and metadata stored under that key. `put` publishes values, `get`
+retrieves them, and `listen` subscribes to value updates and expiration events.

-On Linux use `LD_LIBRARY_PATH` instead of `DYLD_LIBRARY_PATH`; on Windows add
-the native DLL directory to `PATH`. When linking an AddressSanitizer native
-build, use the same compiler runtime, for example
-`RUSTFLAGS="-L native=$PWD/build-bindings -C linker=clang -C link-arg=-fsanitize=address"`.
-
-## Ownership and Errors
-
-Startup, crypto import/generation, serialization, and fallible settings return
-`std::io::Result`. Public-key `pack()` exports bytes; `unpack(bytes)` replaces a
-key only after successful import. Crypto methods accept binary buffers, including
-embedded NULs. Passwords and other C strings reject embedded NULs.
-
-Callbacks are owned `Send + 'static` closures, not borrowed `&mut` closures.
-Use `move` captures and channels or `Arc` for shared application state. Native
-handles are thread-bound; send copied data through channels rather than moving
-a runner or value handle between threads. Handles own stable native pointers
-and can be moved out of or back into their Rust boxes on the same thread.
-Callback panics are contained at the
-FFI boundary; a panicking value callback stops its operation.
-
-`Value::boxed()` creates an independent copy, including payload and metadata,
-so a retained value does not alias mutable native storage or in-flight puts.
+This example starts a node on an OS-assigned port, bootstraps into the public
+network, and listens for a value it publishes.

 ```rust
 use opendht::{DhtRunner, InfoHash, Value};
-use std::sync::mpsc;
+use std::{sync::mpsc, time::Duration};

-let mut runner = DhtRunner::new();
-runner.run(0)?;
-let hash = InfoHash::get("example");
-let (sent, received) = mpsc::channel();
-let token = runner.listen(&hash, move |value, expired| {
-		if !expired { let _ = sent.send(value.as_bytes()); }
-		true
-})?;
-runner.put(&hash, Value::new("hello"), |_| {}, false);
-let bytes = received.recv_timeout(std::time::Duration::from_secs(5))?;
-runner.cancel_listen(&hash, token);
-runner.join()?;
-# Ok::<(), Box<dyn std::error::Error>>(())
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let mut runner = DhtRunner::new();
+    runner.run(0)?;
+    runner.bootstrap("bootstrap.jami.net", 4222)?;
+
+    let key = InfoHash::get("my-app/hello");
+    let (sender, receiver) = mpsc::channel();
+    let listener = runner.listen(&key, move |value, expired| {
+        if !expired {
+            let _ = sender.send(value);
+        }
+        true
+    })?;
+
+    runner.put(&key, Value::new("Hello from Rust!"), |_| {}, false);
+    let value = receiver.recv_timeout(Duration::from_secs(5))?;
+    println!("{}", String::from_utf8_lossy(value.as_slice()));
+
+    runner.cancel_listen(&key, listener);
+    runner.join()?;
+    Ok(())
+}
 ```

-`join()` waits for native callbacks, releases registrations, and permits restart.
-Do not invoke it from a DHT callback. `shutdown(callback)` requests graceful
-shutdown; wait for its callback before `join()` when persistence is required.
-Dropping a listener token frees only the token; explicitly call `cancel_listen`
-to remove the registration, or call `join()` to stop all operations. Pending
-get/put/query completions are reported as failure when the runner stops.
+A bootstrap server supplies initial peers. Applications can use public
+bootstrap servers or known peers on a private network. Values expire, and their
+availability depends on participating nodes; the DHT does not provide durable
+storage guarantees.

-For non-threaded operation set `config.threaded = false` and repeatedly call
-`loop_once()`. Priority operations can take an iteration before normal work.
-`run_bind` selects IPv4/IPv6 bind addresses; `bound_port` reports OS-assigned ports.
+## Optional Features

-## Extended APIs
+| Feature | Capabilities |
+|---------|--------------|
+| Default | Callback-based `DhtRunner`, binary values, keys, certificates, and signed/encrypted operations. |
+| `async` | `aio::AsyncDhtRunner` with publication futures, bounded read/update streams, and automatic listener cancellation on drop. |
+| `serde` | Serde-backed MessagePack records with `put_typed`, `get_typed`, and `listen_typed`. Enables `async`. |

- Values expose explicit IDs, user types, value types, sequence numbers,
-	push priorities, and push types.
- Keys and certificates support persistence; identities support RSA and EC
-	generation. Private keys support binary signing and RSA decryption.
- `get_filtered`, `listen_filtered`, and `query` accept native SQL-like filters,
-	such as `WHERE value_type=17`, `WHERE seq=7`, and
-	`SELECT id,user_type WHERE value_type=17`.
-	Projected fields are optional owned values in `QueryResult`.
- `register_type` installs native type policy and expiration settings.
- Configuration accepts owned status and certificate-store callbacks.
-	Certificate lookup/registration and connectivity changes are exposed.
- Push token/topic/platform setters forward native settings.
-	`push_notification_received` returns a `PushNotification`; call `poll()` for
-	the native result. In non-threaded mode continue driving `loop_once()`.
-	Disabled builds return `IgnoredDisabled`; stopped runners return the native
-	ignored outcome. Push tokens remain valid after the runner stops.
+Enable `async` in the dependency configuration:
+
+```toml
+[dependencies]
+opendht = { version = "0.6", features = ["async"] }
+futures = "0.3"
+```
+
+Create the adapter with `AsyncDhtRunner::new(runner)` after starting a
+`DhtRunner`. The adapter is runtime-independent and thread-bound; use a local
+executor or Tokio's `LocalSet`. Dropping a listener stream cancels its
+subscription. Streams use bounded buffers and report `WouldBlock` on overflow.
+
+This example waits for a bootstrap peer, publishes a value with `.await`, and
+reads matching values as a stream:
+
+```rust,no_run
+use futures::{executor::block_on, TryStreamExt};
+use opendht::{aio::AsyncDhtRunner, DhtRunner, InfoHash, Value};
+use std::{io, net::ToSocketAddrs};
+
+fn main() -> io::Result<()> {
+    let mut runner = DhtRunner::new();
+    runner.run(0)?;
+    let runner = AsyncDhtRunner::new(runner);
+    let peers: Vec<_> = ("bootstrap.jami.net", 4222).to_socket_addrs()?.collect();
+
+    block_on(async {
+        runner.bootstrap(&peers).await?;
+        let key = InfoHash::get("my-app/hello-async");
+        runner.put(&key, Value::new("Hello from async Rust!"), false).await?;
+
+        let mut values = runner.get(&key);
+        while let Some(value) = values.try_next().await? {
+            println!("{}", String::from_utf8_lossy(value.as_slice()));
+        }
+
+        runner.shutdown().await?;
+        runner.join()
+    })
+}
+```
+
+### Typed Records
+
+The `serde` feature adds typed async operations and also enables `async`.
+`put_typed` serializes records implementing `Serialize`; `get_typed` and
+`listen_typed` decode records implementing `DeserializeOwned`. Typed reads retain
+the decoded record alongside its original payload and native metadata.
+
+Structs are serialized as compact MessagePack arrays. Field order is part of
+the wire format and must remain compatible across schema changes. `serde_bytes`
+encodes binary fields using MessagePack's binary representation.
+
+## Working With Values
+
+GET and LISTEN callbacks and async streams deliver owned `Box<Value>` instances.
+Received values implement `Send` but not `Sync`; they can be retained after the
+runner stops, moved to another thread, modified, and republished. `as_slice()`
+borrows the payload without copying; `as_bytes()` allocates a separate `Vec<u8>`.
+
+Values expose metadata including IDs, types, sequence numbers, and owner keys.
+Filters select values by metadata, and queries return selected metadata fields.
+
+`DhtRunner::listen` returns a token that must be passed to `cancel_listen` to
+unsubscribe. Dropping the token does not cancel the registration; dropping an
+async listener stream does.
+
+`join()` stops the runner and waits for its threads to exit. It must not be
+called from a callback. For graceful shutdown and configured state persistence,
+wait for `shutdown` to complete before calling `join()`.
+
+## Learn More
+
+- [API reference](https://docs.rs/opendht)
+- [Callback example](https://github.com/savoirfairelinux/opendht/blob/master/rust/examples/dhtnode.rs)
+- [Async and typed two-node example](https://github.com/savoirfairelinux/opendht/blob/master/rust/examples/dhtnode_async.rs)
+- [Build and test guide](https://github.com/savoirfairelinux/opendht/blob/master/rust/BUILD.md)
+- [OpenDHT project and issue tracker](https://github.com/savoirfairelinux/opendht)
@@ -0,0 +1,80 @@
+use futures::{executor::block_on, StreamExt, TryStreamExt};
+use opendht::{aio::AsyncDhtRunner, DhtRunner, DhtRunnerConfig, InfoHash, Value};
+use std::error::Error;
+use std::io;
+use std::net::SocketAddr;
+
+fn main() -> Result<(), Box<dyn Error>> {
+    let mut source = DhtRunner::new();
+    let mut destination = DhtRunner::new();
+    source.run_bind("127.0.0.1", "::1", 0, DhtRunnerConfig::new())?;
+    destination.run_bind("127.0.0.1", "::1", 0, DhtRunnerConfig::new())?;
+    let address = SocketAddr::from(([127, 0, 0, 1], destination.bound_port(libc::AF_INET as _)));
+    let source = AsyncDhtRunner::new(source);
+    let destination = AsyncDhtRunner::new(destination);
+
+    block_on(async {
+        source.bootstrap(&[address]).await?;
+        let hash = InfoHash::get("async example");
+        let mut updates = destination.listen(&hash)?;
+        source
+            .put(&hash, Value::with_id(b"hello from Rust async", 42), false)
+            .await?;
+        let event = updates
+            .next()
+            .await
+            .ok_or_else(|| io::Error::other("listener ended before receiving a value"))??;
+        println!(
+            "LISTEN {}: {} (expired: {})",
+            event.value.id(),
+            String::from_utf8_lossy(event.value.as_slice()),
+            event.expired,
+        );
+        drop(updates);
+
+        let values: Vec<_> = destination.get(&hash).try_collect().await?;
+        println!("GET: {} values", values.len());
+
+        #[cfg(feature = "serde")]
+        {
+            #[derive(Debug, serde::Serialize, serde::Deserialize)]
+            struct Service {
+                version: u32,
+                name: String,
+                port: u16,
+                #[serde(with = "serde_bytes")]
+                payload: Vec<u8>,
+            }
+
+            let hash = InfoHash::get("typed async example");
+            let mut updates = destination.listen_typed::<Service>(&hash)?;
+            let record = Service {
+                version: 1,
+                name: "chat".into(),
+                port: 4222,
+                payload: vec![0, 0x80, 0xff],
+            };
+            source.put_typed(&hash, &record, false).await?;
+            let event = updates.next().await.ok_or_else(|| {
+                io::Error::other("typed listener ended before receiving a value")
+            })??;
+            println!(
+                "TYPED LISTEN {}: {:?}",
+                event.value.raw.id(),
+                event.value.data
+            );
+            drop(updates);
+            let values: Vec<_> = destination
+                .get_typed::<Service>(&hash)
+                .try_collect()
+                .await?;
+            println!("TYPED GET: {} records", values.len());
+        }
+
+        source.shutdown().await?;
+        destination.shutdown().await?;
+        source.join()?;
+        destination.join()?;
+        Ok(())
+    })
+}
@@ -0,0 +1,429 @@
+//! Runtime-independent futures and bounded streams, enabled by the `async` feature.
+//!
+//! Runners and listeners stay on their creating thread. Use a local executor or
+//! Tokio `LocalSet`; received values own their native storage and can be
+//! transferred between threads without another payload copy.
+
+use crate::{DhtRunner, InfoHash, OpToken, PkId, QueryResult, Value};
+use futures_channel::{mpsc, oneshot};
+use futures_core::Stream;
+use std::cell::RefCell;
+use std::future::Future;
+use std::io;
+use std::net::SocketAddr;
+use std::pin::Pin;
+use std::rc::Rc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll};
+
+/// A listener update; expiration events retain the expired value's metadata.
+pub struct ValueEvent {
+    pub value: Box<Value>,
+    pub expired: bool,
+}
+
+/// A native completion mapped to an I/O result.
+///
+/// Operations start immediately, before this future is polled. Dropping it
+/// abandons the result but does not undo a put, ping, bootstrap, or shutdown.
+#[must_use = "operations run immediately; await their result or explicitly drop it"]
+pub struct Operation {
+    receiver: oneshot::Receiver<io::Result<()>>,
+}
+
+impl Future for Operation {
+    type Output = io::Result<()>;
+
+    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
+        Pin::new(&mut self.receiver).poll(context).map(|outcome| {
+            outcome.unwrap_or_else(|_| Err(stopped("DHT operation ended without completion")))
+        })
+    }
+}
+
+fn operation() -> (oneshot::Sender<io::Result<()>>, Operation) {
+    let (sender, receiver) = oneshot::channel();
+    (sender, Operation { receiver })
+}
+
+fn outcome(success: bool) -> io::Result<()> {
+    if success {
+        Ok(())
+    } else {
+        Err(io::Error::other(
+            "DHT operation failed or the runner stopped",
+        ))
+    }
+}
+
+fn stopped(message: &'static str) -> io::Error {
+    io::Error::new(io::ErrorKind::Interrupted, message)
+}
+
+#[derive(Clone)]
+struct StreamCompletion(Arc<Mutex<Option<oneshot::Sender<io::Result<()>>>>>);
+
+impl StreamCompletion {
+    fn finish(&self, result: io::Result<()>) {
+        let sender = self
+            .0
+            .lock()
+            .unwrap_or_else(|error| error.into_inner())
+            .take();
+        if let Some(sender) = sender {
+            let _ = sender.send(result);
+        }
+    }
+}
+
+struct StreamSender<T> {
+    values: mpsc::Sender<T>,
+    completion: StreamCompletion,
+}
+
+impl<T> StreamSender<T> {
+    fn send(&mut self, value: io::Result<T>) -> bool {
+        let value = match value {
+            Ok(value) => value,
+            Err(error) => {
+                self.completion.finish(Err(error));
+                return false;
+            }
+        };
+        match self.values.try_send(value) {
+            Ok(()) => true,
+            Err(error) => {
+                if error.is_full() {
+                    self.completion.finish(Err(io::Error::new(
+                        io::ErrorKind::WouldBlock,
+                        "DHT stream capacity exceeded; restart the operation to resynchronize",
+                    )));
+                }
+                false
+            }
+        }
+    }
+}
+
+/// A bounded get or query stream.
+///
+/// Buffered items precede a terminal error. Native failure, runner shutdown,
+/// and buffer overflow are never reported as successful EOF.
+/// Dropping the stream stops delivery; native work stops at its next callback
+/// or when the runner is joined.
+#[must_use = "streams must be polled to receive values and completion errors"]
+pub struct OperationStream<T> {
+    values: mpsc::Receiver<T>,
+    completion: oneshot::Receiver<io::Result<()>>,
+    completed: bool,
+    outcome: Option<io::Result<()>>,
+}
+
+impl<T: Unpin> Stream for OperationStream<T> {
+    type Item = io::Result<T>;
+
+    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let stream = self.get_mut();
+        if !stream.completed
+            && let Poll::Ready(result) = Pin::new(&mut stream.completion).poll(context)
+        {
+            stream.completed = true;
+            stream.outcome = Some(
+                result.unwrap_or_else(|_| Err(stopped("DHT stream ended without completion"))),
+            );
+            stream.values.close();
+        }
+        match Pin::new(&mut stream.values).poll_next(context) {
+            Poll::Ready(Some(value)) => Poll::Ready(Some(Ok(value))),
+            Poll::Ready(None) if stream.completed => {
+                Poll::Ready(stream.outcome.take().and_then(Result::err).map(Err))
+            }
+            _ => Poll::Pending,
+        }
+    }
+}
+
+fn stream_channel<T>(capacity: usize) -> (StreamSender<T>, OperationStream<T>) {
+    let (values, receiver) = mpsc::channel(capacity - 1);
+    let (completed, completion) = oneshot::channel();
+    (
+        StreamSender {
+            values,
+            completion: StreamCompletion(Arc::new(Mutex::new(Some(completed)))),
+        },
+        OperationStream {
+            values: receiver,
+            completion,
+            completed: false,
+            outcome: None,
+        },
+    )
+}
+
+struct RunnerState {
+    runner: RefCell<Box<DhtRunner>>,
+    stopping: Arc<AtomicBool>,
+}
+
+impl Drop for RunnerState {
+    fn drop(&mut self) {
+        self.stopping.store(true, Ordering::SeqCst);
+    }
+}
+
+/// A listener registration that cancels itself when dropped or terminated.
+///
+/// The runner is retained until this registration is dropped. Calling `join()`
+/// stops all registrations and makes pending listeners report an error.
+#[must_use = "dropping a listener cancels its native registration"]
+pub struct Listener {
+    state: Rc<RunnerState>,
+    hash: InfoHash,
+    token: Option<Box<OpToken>>,
+    stream: OperationStream<ValueEvent>,
+}
+
+impl Listener {
+    fn cancel(&mut self) {
+        if let Some(token) = self.token.take() {
+            self.state
+                .runner
+                .borrow_mut()
+                .cancel_listen(&self.hash, token);
+        }
+    }
+}
+
+impl Stream for Listener {
+    type Item = io::Result<ValueEvent>;
+
+    fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let result = Pin::new(&mut self.stream).poll_next(context);
+        if matches!(result, Poll::Ready(None) | Poll::Ready(Some(Err(_)))) {
+            self.cancel();
+        }
+        result
+    }
+}
+
+impl Drop for Listener {
+    fn drop(&mut self) {
+        self.cancel();
+    }
+}
+
+/// A shared, thread-bound async adapter for an already configured runner.
+///
+/// Clones share a runner without making native handles `Send` or `Sync`.
+/// Native calls are short; futures and streams never hold a mutable runner
+/// borrow while waiting. The default stream capacity is 64 items.
+///
+/// ```compile_fail
+/// use opendht::{aio::AsyncDhtRunner, DhtRunner};
+/// let runner = AsyncDhtRunner::new(DhtRunner::new());
+/// std::thread::spawn(move || runner.join().unwrap());
+/// ```
+#[derive(Clone)]
+pub struct AsyncDhtRunner {
+    state: Rc<RunnerState>,
+    capacity: usize,
+}
+
+impl AsyncDhtRunner {
+    pub fn new(runner: Box<DhtRunner>) -> Self {
+        Self::with_capacity(runner, 64).expect("default stream capacity is valid")
+    }
+
+    /// Set the maximum number of queued values per get, query, or listener.
+    /// A slow consumer receives `WouldBlock` and the stream terminates.
+    pub fn with_capacity(runner: Box<DhtRunner>, capacity: usize) -> io::Result<Self> {
+        if capacity == 0 {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "stream capacity must be positive",
+            ));
+        }
+        Ok(Self {
+            state: Rc::new(RunnerState {
+                runner: RefCell::new(runner),
+                stopping: Arc::new(AtomicBool::new(false)),
+            }),
+            capacity,
+        })
+    }
+
+    pub fn put(&self, hash: &InfoHash, value: Box<Value>, permanent: bool) -> Operation {
+        let (sender, pending) = operation();
+        self.state.runner.borrow_mut().put(
+            hash,
+            value,
+            move |success| {
+                let _ = sender.send(outcome(success));
+            },
+            permanent,
+        );
+        pending
+    }
+
+    pub fn put_signed(&self, hash: &InfoHash, value: Box<Value>, permanent: bool) -> Operation {
+        let (sender, pending) = operation();
+        self.state.runner.borrow_mut().put_signed(
+            hash,
+            value,
+            move |success| {
+                let _ = sender.send(outcome(success));
+            },
+            permanent,
+        );
+        pending
+    }
+
+    pub fn put_encrypted(
+        &self,
+        hash: &InfoHash,
+        recipient: &PkId,
+        value: Box<Value>,
+        permanent: bool,
+    ) -> Operation {
+        let (sender, pending) = operation();
+        self.state.runner.borrow_mut().put_encrypted(
+            hash,
+            recipient,
+            value,
+            move |success| {
+                let _ = sender.send(outcome(success));
+            },
+            permanent,
+        );
+        pending
+    }
+
+    pub fn ping(&self, address: SocketAddr) -> Operation {
+        let (sender, pending) = operation();
+        self.state
+            .runner
+            .borrow_mut()
+            .ping(address, move |success| {
+                let _ = sender.send(outcome(success));
+            });
+        pending
+    }
+
+    /// Bootstrap pre-resolved addresses concurrently; one successful peer suffices.
+    /// DNS resolution is left to the application's executor.
+    pub fn bootstrap(&self, addresses: &[SocketAddr]) -> Operation {
+        let (sender, pending) = operation();
+        if addresses.is_empty() {
+            let _ = sender.send(Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "at least one bootstrap address is required",
+            )));
+            return pending;
+        }
+        let completion = StreamCompletion(Arc::new(Mutex::new(Some(sender))));
+        let remaining = Arc::new(std::sync::atomic::AtomicUsize::new(addresses.len()));
+        for &address in addresses {
+            let completion = completion.clone();
+            let remaining = remaining.clone();
+            self.state
+                .runner
+                .borrow_mut()
+                .ping(address, move |success| {
+                    if success {
+                        completion.finish(Ok(()));
+                    }
+                    if remaining.fetch_sub(1, Ordering::SeqCst) == 1 {
+                        completion.finish(outcome(false));
+                    }
+                });
+        }
+        pending
+    }
+
+    pub fn get(&self, hash: &InfoHash) -> OperationStream<Box<Value>> {
+        let (mut sender, stream) = stream_channel(self.capacity);
+        let completion = sender.completion.clone();
+        self.state.runner.borrow_mut().get(
+            hash,
+            move |value| sender.send(Ok(value)),
+            move |success| completion.finish(outcome(success)),
+        );
+        stream
+    }
+
+    pub fn get_filtered(
+        &self,
+        hash: &InfoHash,
+        filter: &str,
+    ) -> io::Result<OperationStream<Box<Value>>> {
+        let (mut sender, stream) = stream_channel(self.capacity);
+        let completion = sender.completion.clone();
+        self.state.runner.borrow_mut().get_filtered(
+            hash,
+            filter,
+            move |value| sender.send(Ok(value)),
+            move |success| completion.finish(outcome(success)),
+        )?;
+        Ok(stream)
+    }
+
+    pub fn query(&self, hash: &InfoHash, query: &str) -> io::Result<OperationStream<QueryResult>> {
+        let (mut sender, stream) = stream_channel(self.capacity);
+        let completion = sender.completion.clone();
+        self.state.runner.borrow_mut().query(
+            hash,
+            query,
+            move |row| sender.send(Ok(row)),
+            move |success| completion.finish(outcome(success)),
+        )?;
+        Ok(stream)
+    }
+
+    pub fn listen(&self, hash: &InfoHash) -> io::Result<Listener> {
+        self.listen_filtered(hash, "")
+    }
+
+    pub fn listen_filtered(&self, hash: &InfoHash, filter: &str) -> io::Result<Listener> {
+        let (mut sender, stream) = stream_channel(self.capacity);
+        let token = self.state.runner.borrow_mut().listen_filtered(
+            hash,
+            filter,
+            move |value, expired| sender.send(Ok(ValueEvent { value, expired })),
+        )?;
+        Ok(Listener {
+            state: self.state.clone(),
+            hash: InfoHash { d: hash.d },
+            token: Some(token),
+            stream,
+        })
+    }
+
+    /// Request graceful shutdown. Call `join()` after this future succeeds when
+    /// persistence is required. Joining first reports an interrupted shutdown.
+    pub fn shutdown(&self) -> Operation {
+        let (sender, pending) = operation();
+        let stopping = self.state.stopping.clone();
+        self.state.runner.borrow_mut().shutdown(move || {
+            let result = if stopping.load(Ordering::SeqCst) {
+                Err(stopped("runner joined before graceful shutdown completed"))
+            } else {
+                Ok(())
+            };
+            let _ = sender.send(result);
+        });
+        pending
+    }
+
+    /// Drive a non-threaded runner without waiting for any async operation.
+    pub fn loop_once(&self) -> io::Result<()> {
+        self.state.runner.borrow_mut().loop_once()
+    }
+
+    /// Stop the runner and wait for native callbacks. This is a blocking call,
+    /// not an async wait, and must not be invoked from a native DHT callback.
+    pub fn join(&self) -> io::Result<()> {
+        self.state.stopping.store(true, Ordering::SeqCst);
+        self.state.runner.borrow_mut().join()
+    }
+}
@@ -272,7 +272,7 @@ extern "C" fn get_handler_cb(value: *const ffi::Value, ptr: *mut c_void) -> bool
    }
    callback(false, || unsafe {
        let handler = ptr as *mut GetHandler;
-        (*handler).get_cb(Value::from_raw(ffi::dht_value_copy(value)).expect("value copy failed"))
+        (*handler).get_cb(Value::copy_from_received(value).expect("value copy failed"))
    })
 }

@@ -308,7 +308,7 @@ extern "C" fn listen_handler(value: *const ffi::Value, expired: bool, ptr: *mut
    callback(false, || unsafe {
        let handler = ptr as *mut ListenHandler;
        (*handler).cb(
-            Value::from_raw(ffi::dht_value_copy(value)).expect("value copy failed"),
+            Value::copy_from_received(value).expect("value copy failed"),
            expired,
        )
    })
@@ -537,6 +537,7 @@ impl DhtRunner {
        result(unsafe { ffi::dht_runner_connectivity_changed(self.as_ptr()) })
    }

+    /// Deliver owned values that callbacks may mutate or move to another thread.
    pub fn get(
        &mut self,
        h: &InfoHash,
@@ -676,7 +677,8 @@ impl DhtRunner {
        unsafe { ffi::dht_runner_cancel_put(self.as_ptr(), h, vid) }
    }

-    /// Registers an owned callback. It is released after cancellation or `join()`.
+    /// Registers an owned callback receiving ordinary owned `Box<Value>` objects.
+    /// It is released after cancellation or `join()`.
    /// Borrowed callbacks cannot outlive their stack frame:
    ///
    /// ```compile_fail
@@ -40,6 +40,12 @@ mod infohash;
 mod pkid;
 mod value;

+#[cfg(feature = "async")]
+pub mod aio;
+
+#[cfg(feature = "serde")]
+pub mod typed;
+
 pub use blob::Blob;
 pub use dhtrunner::{
    DhtNodeConfig, DhtRunner, DhtRunnerConfig, DhtSecureConfig, NodeStatus, OpToken,
@@ -0,0 +1,201 @@
+//! MessagePack-encoded typed operations, enabled by the `serde` feature.
+//!
+//! This feature enables `async` and adds methods to [`AsyncDhtRunner`]. Values
+//! use compact MessagePack payloads, with no binding-specific envelope or
+//! implicit schema version. Structs are encoded as arrays in field order.
+//! Applications can include a version field in their records and use
+//! `serde_bytes` for binary fields rather than arrays of integers.
+//!
+//! ```no_run
+//! use futures::TryStreamExt;
+//! use opendht::{aio::AsyncDhtRunner, InfoHash};
+//! use serde::{Deserialize, Serialize};
+//!
+//! #[derive(Debug, Serialize, Deserialize)]
+//! struct Service {
+//!     version: u32,
+//!     name: String,
+//!     port: u16,
+//!     #[serde(with = "serde_bytes")]
+//!     payload: Vec<u8>,
+//! }
+//!
+//! async fn exchange(
+//!     runner: &AsyncDhtRunner,
+//!     key: &InfoHash,
+//! ) -> Result<(), Box<dyn std::error::Error>> {
+//!     let mut updates = runner.listen_typed::<Service>(key)?;
+//!     let record = Service {
+//!         version: 1, name: "chat".into(), port: 4222, payload: vec![0, 0x80, 0xff],
+//!     };
+//!     runner.put_typed(key, &record, false).await?;
+//!     if let Some(event) = updates.try_next().await? {
+//!         println!("{}: {:?}", event.value.raw.id(), event.value.data);
+//!     }
+//!     drop(updates);
+//!     let records: Vec<_> = runner.get_typed::<Service>(key).try_collect().await?;
+//!     Ok(())
+//! }
+//! ```
+
+use crate::aio::{AsyncDhtRunner, ValueEvent};
+use crate::{InfoHash, Value};
+use futures_core::Stream;
+use serde::{de::DeserializeOwned, Serialize};
+use std::error::Error;
+use std::fmt;
+use std::io;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+/// A decoded record together with its original payload and native metadata.
+pub struct TypedValue<T> {
+    pub data: T,
+    pub raw: Box<Value>,
+}
+
+impl<T: fmt::Debug> fmt::Debug for TypedValue<T> {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("TypedValue")
+            .field("data", &self.data)
+            .field("id", &self.raw.id())
+            .field("seq", &self.raw.seq())
+            .finish_non_exhaustive()
+    }
+}
+
+/// A decoded listener update, including whether the record expired.
+#[derive(Debug)]
+pub struct TypedEvent<T> {
+    pub value: TypedValue<T>,
+    pub expired: bool,
+}
+
+/// Typed operations distinguish native failures from payload format errors.
+#[derive(Debug)]
+#[non_exhaustive]
+pub enum TypedError {
+    Operation(io::Error),
+    Serialize(rmp_serde::encode::Error),
+    Deserialize {
+        id: u64,
+        source: rmp_serde::decode::Error,
+    },
+}
+
+impl fmt::Display for TypedError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Operation(error) => write!(formatter, "DHT operation failed: {error}"),
+            Self::Serialize(error) => write!(formatter, "could not serialize typed value: {error}"),
+            Self::Deserialize { id, source } => {
+                write!(formatter, "could not deserialize DHT value {id}: {source}")
+            }
+        }
+    }
+}
+
+impl Error for TypedError {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        match self {
+            Self::Operation(error) => Some(error),
+            Self::Serialize(error) => Some(error),
+            Self::Deserialize { source, .. } => Some(source),
+        }
+    }
+}
+
+impl<T: DeserializeOwned> TryFrom<Box<Value>> for TypedValue<T> {
+    type Error = TypedError;
+
+    fn try_from(raw: Box<Value>) -> Result<Self, Self::Error> {
+        let data =
+            rmp_serde::from_slice(raw.as_slice()).map_err(|source| TypedError::Deserialize {
+                id: raw.id(),
+                source,
+            })?;
+        Ok(Self { data, raw })
+    }
+}
+
+impl<T: DeserializeOwned> TryFrom<ValueEvent> for TypedEvent<T> {
+    type Error = TypedError;
+
+    fn try_from(event: ValueEvent) -> Result<Self, Self::Error> {
+        Ok(Self {
+            value: event.value.try_into()?,
+            expired: event.expired,
+        })
+    }
+}
+
+struct DecodeStream<Source: Stream, Output> {
+    source: Source,
+    decode: fn(Source::Item) -> Result<Output, TypedError>,
+}
+
+impl<Source: Stream + Unpin, Output> Stream for DecodeStream<Source, Output> {
+    type Item = Result<Output, TypedError>;
+
+    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let stream = self.get_mut();
+        Pin::new(&mut stream.source)
+            .poll_next(context)
+            .map(|item| item.map(stream.decode))
+    }
+}
+
+fn decode<Input, Output: TryFrom<Input, Error = TypedError>>(
+    input: io::Result<Input>,
+) -> Result<Output, TypedError> {
+    input.map_err(TypedError::Operation)?.try_into()
+}
+
+impl AsyncDhtRunner {
+    /// Serialize a record as MessagePack and publish it when this future is polled.
+    ///
+    /// Unlike untyped `put`, this operation is lazy. Serialization failure does
+    /// not start a native put. Dropping it after publication starts does not
+    /// undo the put. Use the untyped API to customize value IDs and metadata.
+    pub async fn put_typed<T: Serialize + ?Sized>(
+        &self,
+        hash: &InfoHash,
+        value: &T,
+        permanent: bool,
+    ) -> Result<(), TypedError> {
+        let bytes = rmp_serde::to_vec(value).map_err(TypedError::Serialize)?;
+        self.put(hash, Value::from_bytes(&bytes), permanent)
+            .await
+            .map_err(TypedError::Operation)
+    }
+
+    /// Start a get and decode each MessagePack payload on the consumer's thread.
+    ///
+    /// A malformed or incompatible record yields a per-value error; polling can
+    /// continue to subsequent records. Native errors remain terminal. The stream
+    /// does not borrow the runner or hash, and records need not be `Unpin`.
+    pub fn get_typed<T: DeserializeOwned>(
+        &self,
+        hash: &InfoHash,
+    ) -> impl Stream<Item = Result<TypedValue<T>, TypedError>> + Unpin + use<T> {
+        DecodeStream {
+            source: self.get(hash),
+            decode: decode::<_, TypedValue<T>>,
+        }
+    }
+
+    /// Listen for MessagePack records, preserving metadata and expiration events.
+    ///
+    /// Decode errors do not cancel the registration. Dropping this stream does
+    /// cancel it, including when a short-circuiting consumer stops on an error.
+    pub fn listen_typed<T: DeserializeOwned>(
+        &self,
+        hash: &InfoHash,
+    ) -> io::Result<impl Stream<Item = Result<TypedEvent<T>, TypedError>> + Unpin + use<T>> {
+        Ok(DecodeStream {
+            source: self.listen(hash)?,
+            decode: decode::<_, TypedEvent<T>>,
+        })
+    }
+}
@@ -14,7 +14,40 @@ pub use crate::ffi::DataView;

 native_handle!(Value, dht_value_unref);

+/// Rust owns an unaliased native value: constructors allocate it, received
+/// values are deep-copied once, and boxed() deep-copies explicitly. Puts consume
+/// the handle before native code retains it. Shared owner keys are const and
+/// their lazy fingerprints are initialized before received values cross threads.
+/// Mutation requires exclusive access; native storage may be freed on any thread.
+/// Value is Send, not Sync.
+unsafe impl Send for Value {}
+
+impl AsRef<[u8]> for Value {
+    fn as_ref(&self) -> &[u8] {
+        self.as_slice()
+    }
+}
+
+impl fmt::Debug for Value {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("Value")
+            .field("id", &self.id())
+            .field("seq", &self.seq())
+            .finish_non_exhaustive()
+    }
+}
+
 impl Value {
+    pub(crate) unsafe fn copy_from_received(value: *const ffi::Value) -> io::Result<Box<Self>> {
+        let value = unsafe { Self::from_raw(ffi::dht_value_copy(value))? };
+        if let Some(owner) = value.owner() {
+            owner.id();
+            owner.long_id();
+        }
+        Ok(value)
+    }
+
    pub fn with_id(data: &[u8], id: u64) -> Box<Value> {
        unsafe {
            Self::from_raw(ffi::dht_value_with_id_new(data.as_ptr(), data.len(), id))
@@ -41,13 +74,16 @@ impl Value {
    }

    pub fn as_bytes(&self) -> Vec<u8> {
-        unsafe {
-            let dv = ffi::dht_value_get_data(self.as_ptr());
-            if dv.size == 0 {
-                return Vec::new();
-            }
-            let slice = slice::from_raw_parts(dv.data, dv.size);
-            slice.to_vec()
+        self.as_slice().to_vec()
+    }
+
+    /// Borrow the payload without allocating or copying it.
+    pub fn as_slice(&self) -> &[u8] {
+        let view = self.dataview();
+        if view.size == 0 {
+            &[]
+        } else {
+            unsafe { slice::from_raw_parts(view.data, view.size) }
        }
    }

@@ -63,12 +99,8 @@ impl Value {
        unsafe { PublicKey::from_raw(ffi::dht_value_get_owner(self.as_ptr())).ok() }
    }

-    pub fn user_type(&self) -> Result<String, str::Utf8Error> {
-        unsafe {
-            Ok(CStr::from_ptr(ffi::dht_value_get_user_type(self.as_ptr()))
-                .to_str()?
-                .to_owned())
-        }
+    pub fn user_type(&self) -> Result<&str, str::Utf8Error> {
+        unsafe { CStr::from_ptr(ffi::dht_value_get_user_type(self.as_ptr())).to_str() }
    }

    pub fn set_user_type(&mut self, value: &str) -> io::Result<()> {
@@ -103,12 +135,8 @@ impl Value {
            ffi::dht_value_set_priority(self.as_ptr(), value);
        }
    }
-    pub fn push_type(&self) -> Result<String, str::Utf8Error> {
-        unsafe {
-            Ok(CStr::from_ptr(ffi::dht_value_get_push_type(self.as_ptr()))
-                .to_str()?
-                .to_owned())
-        }
+    pub fn push_type(&self) -> Result<&str, str::Utf8Error> {
+        unsafe { CStr::from_ptr(ffi::dht_value_get_push_type(self.as_ptr())).to_str() }
    }
    pub fn set_push_type(&mut self, value: &str) -> io::Result<()> {
        let value = CString::new(value)?;
@@ -131,3 +159,33 @@ impl fmt::Display for Value {
        }
    }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn value_moves_between_threads_without_copying_payload() {
+        let mut value = Value::from_bytes(b"native\0payload\xff");
+        value.set_user_type("borrowed metadata").unwrap();
+        let view = unsafe { ffi::dht_value_get_data(value.as_ptr()) };
+        let value = std::thread::spawn(move || {
+            value.set_seq(7);
+            value
+        })
+        .join()
+        .unwrap();
+        assert_eq!(value.as_slice().as_ptr(), view.data);
+        assert_eq!(value.as_slice(), b"native\0payload\xff");
+        assert_eq!(value.seq(), 7);
+        let user_type: &str = value.user_type().unwrap();
+        assert_eq!(user_type, "borrowed metadata");
+    }
+
+    #[test]
+    fn empty_value_payload_is_borrowed_safely() {
+        let value = Value::from_bytes(&[]);
+        assert!(value.as_slice().is_empty());
+        assert_eq!(value.as_slice(), &[]);
+    }
+}
@@ -55,7 +55,6 @@ fn native_handles_can_move_out_of_their_boxes() {
    let token = *runner
        .listen(&hash, move |value, expired| {
            if !expired {
-                let value = *value;
                sent.send(value.as_bytes()).unwrap();
            }
            true
@@ -246,7 +245,7 @@ fn shutdown_owns_its_callback_and_contains_panics() {
 fn startup_errors_reach_rust() {
    let socket4 = std::net::UdpSocket::bind("0.0.0.0:0").unwrap();
    let port = socket4.local_addr().unwrap().port();
-    let _socket6 = std::net::UdpSocket::bind((std::net::Ipv6Addr::UNSPECIFIED, port)).unwrap();
+    let _socket6 = std::net::UdpSocket::bind((std::net::Ipv6Addr::LOCALHOST, port)).unwrap();
    assert!(DhtRunner::new().run(port).is_err());
 }

@@ -481,3 +480,566 @@ fn threaded_signed_and_encrypted_roundtrips() {
        runner.join().unwrap();
    }
 }
+
+#[test]
+fn callbacks_move_mutable_values_without_copying() {
+    let mut runner = DhtRunner::new();
+    let mut config = DhtRunnerConfig::new();
+    config.threaded = false;
+    runner.run_config(0, config).unwrap();
+    let hash = InfoHash::get("single Value callback ownership");
+    let (listen_sent, listen_received) = mpsc::channel();
+    let _token = runner
+        .listen(&hash, move |mut value: Box<Value>, expired| {
+            if !expired {
+                value.set_seq(8);
+                let pointer = value.as_slice().as_ptr() as usize;
+                let _ = listen_sent.send((pointer, value));
+            }
+            true
+        })
+        .unwrap();
+    let mut value = Value::with_id(b"immutable\0payload\xff", 42);
+    value.set_seq(7);
+    value.set_user_type("application/immutable").unwrap();
+    value.set_push_type("call").unwrap();
+    runner.put(&hash, value, |_| {}, false);
+    runner.loop_once().unwrap();
+
+    let (get_sent, get_received) = mpsc::channel();
+    runner.get(
+        &hash,
+        move |value: Box<Value>| {
+            let pointer = value.as_slice().as_ptr() as usize;
+            let _ = get_sent.send((pointer, value));
+            true
+        },
+        |_| {},
+    );
+    runner.loop_once().unwrap();
+    runner.join().unwrap();
+
+    for (received, expected_seq) in [(listen_received, 8), (get_received, 7)] {
+        let (pointer, value) = received.recv_timeout(Duration::from_secs(5)).unwrap();
+        let value = std::thread::spawn(move || {
+            let mut value = value;
+            assert_eq!(value.as_slice().as_ptr() as usize, pointer);
+            assert_eq!(value.as_slice(), b"immutable\0payload\xff");
+            assert_eq!(value.id(), 42);
+            assert_eq!(value.seq(), expected_seq);
+            let user_type: &str = value.user_type().unwrap();
+            assert_eq!(user_type, "application/immutable");
+            let push_type: &str = value.push_type().unwrap();
+            assert_eq!(push_type, "call");
+            value.set_seq(9);
+            assert_eq!(value.as_slice().as_ptr() as usize, pointer);
+            value
+        })
+        .join()
+        .unwrap();
+        let mut editable = value.boxed();
+        editable.set_seq(10);
+        assert_eq!(value.seq(), 9);
+        assert_eq!(editable.seq(), 10);
+    }
+}
+
+#[cfg(feature = "async")]
+mod asynchronous {
+    use super::*;
+    use futures::{executor::block_on, future, Future, StreamExt, TryStreamExt};
+    use opendht::aio::AsyncDhtRunner;
+    use std::pin::pin;
+    use std::time::Instant;
+
+    fn nonthreaded(capacity: usize) -> AsyncDhtRunner {
+        let mut runner = DhtRunner::new();
+        let mut config = DhtRunnerConfig::new();
+        config.threaded = false;
+        runner.run_bind("127.0.0.1", "::1", 0, config).unwrap();
+        runner.register_type(17, "async test", 1, false).unwrap();
+        AsyncDhtRunner::with_capacity(runner, capacity).unwrap()
+    }
+
+    fn drive<Output>(runner: &AsyncDhtRunner, operation: impl Future<Output = Output>) -> Output {
+        let mut operation = pin!(operation);
+        let deadline = Instant::now() + Duration::from_secs(15);
+        block_on(future::poll_fn(|context| {
+            assert!(Instant::now() < deadline, "async operation timed out");
+            runner.loop_once().unwrap();
+            context.waker().wake_by_ref();
+            operation.as_mut().poll(context)
+        }))
+    }
+
+    fn wait<Output>(operation: impl Future<Output = Output>) -> Output {
+        let (finished, stopped) = mpsc::channel();
+        let (expired, deadline) = futures::channel::oneshot::channel();
+        let watchdog = std::thread::spawn(move || {
+            if stopped.recv_timeout(Duration::from_secs(20)).is_err() {
+                let _ = expired.send(());
+            }
+        });
+        let outcome = block_on(future::select(Box::pin(operation), deadline));
+        let _ = finished.send(());
+        watchdog.join().unwrap();
+        match outcome {
+            future::Either::Left((output, _)) => output,
+            future::Either::Right(_) => panic!("async operation timed out"),
+        }
+    }
+
+    #[test]
+    fn async_listener_preserves_owned_metadata() {
+        let runner = nonthreaded(8);
+        let hash = InfoHash::get("async metadata");
+        let mut listener = runner.listen_filtered(&hash, "WHERE seq=7").unwrap();
+        let mut value = Value::with_id(b"async\0binary\xff", 42);
+        value.set_value_type(17);
+        value.set_seq(7);
+        value.set_priority(2);
+        value.set_user_type("application/async").unwrap();
+        value.set_push_type("call").unwrap();
+        let _pending = runner.put(&hash, value, false);
+        let event = drive(&runner, listener.next()).unwrap().unwrap();
+        assert!(!event.expired);
+        assert_eq!(event.value.as_slice(), b"async\0binary\xff");
+        assert_eq!(event.value.id(), 42);
+        assert_eq!(event.value.value_type(), 17);
+        assert_eq!(event.value.seq(), 7);
+        assert_eq!(event.value.priority(), 2);
+        assert_eq!(event.value.user_type().unwrap(), "application/async");
+        assert_eq!(event.value.push_type().unwrap(), "call");
+        assert!(event.value.owner().is_none());
+        assert!(event.value.recipient().is_zero());
+        let event = std::thread::spawn(move || event).join().unwrap();
+        assert_eq!(event.value.as_slice(), b"async\0binary\xff");
+        drop(listener);
+        runner.loop_once().unwrap();
+        runner.join().unwrap();
+    }
+
+    #[test]
+    fn async_listener_delivers_expiration_events() {
+        let runner = nonthreaded(8);
+        let hash = InfoHash::get("async expiration");
+        let mut listener = runner.listen(&hash).unwrap();
+        let mut value = Value::with_id(b"expires", 42);
+        value.set_value_type(17);
+        let _pending = runner.put(&hash, value, false);
+        assert!(!drive(&runner, listener.next()).unwrap().unwrap().expired);
+        let event = drive(&runner, listener.next()).unwrap().unwrap();
+        assert!(event.expired);
+        assert_eq!(event.value.id(), 42);
+        assert_eq!(event.value.as_slice(), b"expires");
+        drop(listener);
+        runner.join().unwrap();
+    }
+
+    #[test]
+    fn async_overflow_is_reported_after_buffered_values() {
+        let runner = nonthreaded(1);
+        let hash = InfoHash::get("async overflow");
+        let mut listener = runner.listen(&hash).unwrap();
+        for id in 1..=3 {
+            let _pending = runner.put(&hash, Value::with_id(b"queued", id), false);
+            runner.loop_once().unwrap();
+        }
+        assert_eq!(
+            drive(&runner, listener.next()).unwrap().unwrap().value.id(),
+            1
+        );
+        let error = drive(&runner, listener.next()).unwrap().err().unwrap();
+        assert_eq!(error.kind(), std::io::ErrorKind::WouldBlock);
+        assert!(drive(&runner, listener.next()).is_none());
+        drop(listener);
+        runner.join().unwrap();
+    }
+
+    #[test]
+    fn async_listener_retains_runner_until_drop() {
+        let released = Arc::new(AtomicUsize::new(0));
+        let counter = DropCounter(released.clone());
+        let mut native = DhtRunner::new();
+        let mut config = DhtRunnerConfig::new();
+        config.threaded = false;
+        config.set_status_callback(move |_, _| {
+            let _ = &counter;
+        });
+        native.run_config(0, config).unwrap();
+        let runner = AsyncDhtRunner::new(native);
+        let listener = runner.listen(&InfoHash::get("retained runner")).unwrap();
+        drop(runner);
+        assert_eq!(released.load(Ordering::SeqCst), 0);
+        drop(listener);
+        assert_eq!(released.load(Ordering::SeqCst), 1);
+    }
+
+    #[test]
+    fn async_join_completes_pending_operations_and_streams() {
+        let runner = nonthreaded(8);
+        let hash = InfoHash::get("async shutdown failures");
+        let put = runner.put(&hash, Value::new("pending"), false);
+        let mut values = runner.get(&hash);
+        let mut query = runner.query(&hash, "SELECT id").unwrap();
+        let mut listener = runner.listen(&hash).unwrap();
+        let ping = runner.ping("127.0.0.1:9".parse().unwrap());
+        let bootstrap = runner.bootstrap(&["127.0.0.1:9".parse().unwrap()]);
+        let shutdown = runner.shutdown();
+        runner.join().unwrap();
+        assert!(wait(put).is_err());
+        assert!(wait(ping).is_err());
+        assert!(wait(bootstrap).is_err());
+        assert!(wait(shutdown).is_err());
+        assert!(wait(values.next()).unwrap().is_err());
+        assert!(wait(values.next()).is_none());
+        assert!(wait(query.next()).unwrap().is_err());
+        assert!(wait(query.next()).is_none());
+        assert!(wait(listener.next()).unwrap().is_err());
+        assert!(wait(listener.next()).is_none());
+        drop(listener);
+    }
+
+    #[test]
+    fn async_stream_reports_runner_destruction() {
+        let runner = nonthreaded(8);
+        let mut values = runner.get(&InfoHash::get("dropped async runner"));
+        drop(runner);
+        assert!(wait(values.next()).unwrap().is_err());
+        assert!(wait(values.next()).is_none());
+    }
+
+    #[test]
+    fn async_invalid_inputs_and_dropped_operations() {
+        assert!(AsyncDhtRunner::with_capacity(DhtRunner::new(), 0).is_err());
+        let runner = nonthreaded(8);
+        let hash = InfoHash::get("async cancellation");
+        let error = wait(runner.bootstrap(&[])).unwrap_err();
+        assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);
+        assert!(runner.get_filtered(&hash, "WHERE user_type=\0").is_err());
+        assert!(runner.listen_filtered(&hash, "WHERE user_type=\0").is_err());
+        assert!(runner.query(&hash, "SELECT\0 id").is_err());
+        let mut invalid = runner.get_filtered(&hash, "WHERE invalid=1").unwrap();
+        assert!(wait(invalid.next()).unwrap().is_err());
+        assert!(wait(invalid.next()).is_none());
+        drop(runner.get(&hash));
+        drop(runner.query(&hash, "SELECT id").unwrap());
+        drop(runner.put(&hash, Value::new("abandoned"), false));
+        drop(runner.ping("127.0.0.1:9".parse().unwrap()));
+        let listener = runner.listen(&hash).unwrap();
+        drop(listener);
+        runner.loop_once().unwrap();
+        runner.join().unwrap();
+    }
+
+    #[test]
+    fn async_threaded_network_roundtrips() {
+        let mut nodes = Vec::new();
+        for name in ["async source", "async destination"] {
+            let mut runner = DhtRunner::new();
+            let mut config = DhtRunnerConfig::new();
+            config.dht_config.id = Some(DhtIdentity::new(name).unwrap());
+            runner.run_bind("127.0.0.1", "::1", 0, config).unwrap();
+            nodes.push(runner);
+        }
+        let destination_native = nodes.pop().unwrap();
+        let address = (
+            [127, 0, 0, 1],
+            destination_native.bound_port(libc::AF_INET as _),
+        )
+            .into();
+        let recipient = destination_native.public_key().unwrap().long_id();
+        let signer = nodes.last().unwrap().public_key().unwrap().id();
+        let source = AsyncDhtRunner::new(nodes.pop().unwrap());
+        let destination = AsyncDhtRunner::new(destination_native);
+        wait(source.bootstrap(&["127.0.0.1:9".parse().unwrap(), address])).unwrap();
+
+        for mode in ["plain", "signed", "encrypted"] {
+            let hash = InfoHash::get(mode);
+            let mut listener = destination.listen(&hash).unwrap();
+            let mut value = Value::with_id(b"async network\0value", 42);
+            value.set_seq(7);
+            let put = match mode {
+                "signed" => source.put_signed(&hash, value, false),
+                "encrypted" => source.put_encrypted(&hash, &recipient, value, false),
+                _ => source.put(&hash, value, false),
+            };
+            wait(put).unwrap();
+            let event = wait(listener.next()).unwrap().unwrap();
+            assert!(!event.expired);
+            assert_eq!(event.value.as_slice(), b"async network\0value");
+            if mode != "plain" {
+                assert!(event.value.owner().is_some());
+                assert!(event.value.owner().unwrap().id() == signer);
+                let event = std::thread::spawn(move || event).join().unwrap();
+                assert!(event.value.owner().unwrap().id() == signer);
+            }
+            drop(listener);
+            if mode == "signed" {
+                let values: Vec<_> = wait(
+                    destination
+                        .get_filtered(&hash, "WHERE seq=7")
+                        .unwrap()
+                        .try_collect(),
+                )
+                .unwrap();
+                assert!(values
+                    .iter()
+                    .any(|value| value.id() == 42 && value.seq() == 7));
+            }
+            let values: Vec<_> = wait(destination.get(&hash).try_collect()).unwrap();
+            assert!(values
+                .iter()
+                .any(|value| value.as_slice() == b"async network\0value"));
+            let rows: Vec<_> =
+                wait(destination.query(&hash, "SELECT id").unwrap().try_collect()).unwrap();
+            assert!(rows.iter().any(|row| row.id == Some(42)));
+        }
+
+        wait(source.shutdown()).unwrap();
+        wait(destination.shutdown()).unwrap();
+        source.join().unwrap();
+        destination.join().unwrap();
+    }
+
+    #[cfg(feature = "serde")]
+    mod typed {
+        use super::*;
+        use opendht::typed::TypedError;
+        use serde::{Deserialize, Serialize};
+        use std::error::Error;
+        use std::marker::PhantomPinned;
+
+        #[derive(Debug, PartialEq, Serialize, Deserialize)]
+        struct Record {
+            version: u32,
+            name: String,
+            ports: Vec<u16>,
+            #[serde(with = "serde_bytes")]
+            payload: Vec<u8>,
+        }
+
+        fn record() -> Record {
+            Record {
+                version: 1,
+                name: "node-\u{03bb}\0".into(),
+                ports: vec![4222, 8443],
+                payload: vec![0, 0x80, 0xff, 1],
+            }
+        }
+
+        #[test]
+        fn typed_listener_decodes_compact_msgpack_binary() {
+            #[derive(Debug, PartialEq, Deserialize)]
+            struct Packet {
+                #[serde(with = "serde_bytes")]
+                payload: Vec<u8>,
+            }
+
+            let runner = nonthreaded(8);
+            let hash = InfoHash::get("typed MessagePack binary fixture");
+            let mut listener = runner.listen_typed::<Packet>(&hash).unwrap();
+            let fixture = [0x91, 0xc4, 0x04, 0x00, 0x80, 0xff, 0x01];
+            let _pending = runner.put(&hash, Value::with_id(&fixture, 42), false);
+            let event = drive(&runner, listener.next()).unwrap().unwrap();
+            assert_eq!(event.value.raw.as_slice(), fixture);
+            assert_eq!(event.value.data.payload, [0, 0x80, 0xff, 1]);
+            drop(listener);
+            runner.join().unwrap();
+        }
+
+        #[test]
+        fn typed_listener_preserves_metadata_and_expiration() {
+            let runner = nonthreaded(8);
+            let hash = InfoHash::get("typed metadata");
+            let mut listener = runner.listen_typed::<Record>(&hash).unwrap();
+            let mut value = Value::with_id(&rmp_serde::to_vec(&record()).unwrap(), 42);
+            value.set_value_type(17);
+            value.set_seq(7);
+            value.set_user_type("application/typed-test").unwrap();
+            let _pending = runner.put(&hash, value, false);
+            let event = drive(&runner, listener.next()).unwrap().unwrap();
+            assert!(!event.expired);
+            assert_eq!(event.value.data, record());
+            assert_eq!(event.value.raw.id(), 42);
+            assert_eq!(event.value.raw.value_type(), 17);
+            assert_eq!(event.value.raw.seq(), 7);
+            assert_eq!(
+                event.value.raw.user_type().unwrap(),
+                "application/typed-test"
+            );
+            assert_eq!(
+                rmp_serde::from_slice::<Record>(event.value.raw.as_slice()).unwrap(),
+                record()
+            );
+            let event = drive(&runner, listener.next()).unwrap().unwrap();
+            assert!(event.expired);
+            assert_eq!(event.value.data, record());
+            assert_eq!(event.value.raw.id(), 42);
+            drop(listener);
+            runner.join().unwrap();
+        }
+
+        #[test]
+        fn typed_decode_errors_do_not_end_listener() {
+            let runner = nonthreaded(8);
+            let hash = InfoHash::get("typed mixed payloads");
+            let mut listener = runner.listen_typed::<Record>(&hash).unwrap();
+            for (id, payload) in [(1, [0xc1].as_slice()), (2, [0x11].as_slice())] {
+                let _pending = runner.put(&hash, Value::with_id(payload, id), false);
+                let error = drive(&runner, listener.next()).unwrap().err().unwrap();
+                assert!(error.source().is_some());
+                assert!(
+                    matches!(error, TypedError::Deserialize { id: value_id, .. } if value_id == id)
+                );
+            }
+            let _pending = runner.put(
+                &hash,
+                Value::with_id(&rmp_serde::to_vec(&record()).unwrap(), 3),
+                false,
+            );
+            let event = drive(&runner, listener.next()).unwrap().unwrap();
+            assert_eq!(event.value.raw.id(), 3);
+            assert_eq!(event.value.data, record());
+            runner.join().unwrap();
+            let error = wait(listener.next()).unwrap().err().unwrap();
+            assert!(matches!(error, TypedError::Operation(_)));
+            assert!(wait(listener.next()).is_none());
+        }
+
+        #[test]
+        fn typed_put_distinguishes_serialization_and_native_errors() {
+            struct Unserializable;
+
+            impl Serialize for Unserializable {
+                fn serialize<Serializer>(
+                    &self,
+                    _: Serializer,
+                ) -> Result<Serializer::Ok, Serializer::Error>
+                where
+                    Serializer: serde::Serializer,
+                {
+                    Err(serde::ser::Error::custom(
+                        "intentional serialization failure",
+                    ))
+                }
+            }
+
+            let runner = AsyncDhtRunner::new(DhtRunner::new());
+            let hash = InfoHash::get("typed put errors");
+            let error = wait(runner.put_typed(&hash, &Unserializable, false)).unwrap_err();
+            assert!(error.source().is_some());
+            assert!(matches!(error, TypedError::Serialize(_)));
+            let error = wait(runner.put_typed(&hash, &record(), false)).unwrap_err();
+            assert!(error.source().is_some());
+            assert!(matches!(error, TypedError::Operation(_)));
+            runner.join().unwrap();
+        }
+
+        #[test]
+        fn typed_streams_support_non_unpin_records_and_detached_lifetimes() {
+            #[derive(Serialize, Deserialize)]
+            struct PinnedRecord {
+                message: String,
+                #[serde(skip)]
+                _pin: PhantomPinned,
+            }
+
+            let runner = nonthreaded(8);
+            let hash = InfoHash::get("typed pinned records");
+            let mut listener = runner.listen_typed::<PinnedRecord>(&hash).unwrap();
+            let value = PinnedRecord {
+                message: "owned".into(),
+                _pin: PhantomPinned,
+            };
+            let _pending = runner.put(
+                &hash,
+                Value::from_bytes(&rmp_serde::to_vec(&value).unwrap()),
+                false,
+            );
+            let event = drive(&runner, listener.next()).unwrap().unwrap();
+            assert_eq!(event.value.data.message, "owned");
+            drop(listener);
+            let mut values = runner.get_typed::<PinnedRecord>(&InfoHash::get("temporary hash"));
+            drop(runner);
+            assert!(matches!(
+                wait(values.next()).unwrap(),
+                Err(TypedError::Operation(_))
+            ));
+            assert!(wait(values.next()).is_none());
+        }
+
+        #[test]
+        fn typed_listener_inherits_bounded_stream_errors() {
+            let runner = nonthreaded(1);
+            let hash = InfoHash::get("typed overflow");
+            let mut listener = runner.listen_typed::<Record>(&hash).unwrap();
+            for id in 1..=3 {
+                let _pending = runner.put(
+                    &hash,
+                    Value::with_id(&rmp_serde::to_vec(&record()).unwrap(), id),
+                    false,
+                );
+                runner.loop_once().unwrap();
+            }
+            assert_eq!(
+                drive(&runner, listener.next())
+                    .unwrap()
+                    .unwrap()
+                    .value
+                    .raw
+                    .id(),
+                1
+            );
+            let error = drive(&runner, listener.next()).unwrap().err().unwrap();
+            assert!(
+                matches!(error, TypedError::Operation(error) if error.kind() == std::io::ErrorKind::WouldBlock)
+            );
+            assert!(drive(&runner, listener.next()).is_none());
+            drop(listener);
+            runner.join().unwrap();
+        }
+
+        #[test]
+        fn typed_threaded_put_get_listen_roundtrip() {
+            let mut source = DhtRunner::new();
+            let mut destination = DhtRunner::new();
+            source
+                .run_bind("127.0.0.1", "::1", 0, DhtRunnerConfig::new())
+                .unwrap();
+            destination
+                .run_bind("127.0.0.1", "::1", 0, DhtRunnerConfig::new())
+                .unwrap();
+            let address = ([127, 0, 0, 1], destination.bound_port(libc::AF_INET as _)).into();
+            let source = AsyncDhtRunner::new(source);
+            let destination = AsyncDhtRunner::new(destination);
+            wait(source.bootstrap(&[address])).unwrap();
+            let hash = InfoHash::get("typed network record");
+            let mut listener = destination.listen_typed::<Record>(&hash).unwrap();
+            wait(source.put_typed(&hash, &record(), false)).unwrap();
+            let event = wait(listener.next()).unwrap().unwrap();
+            assert_eq!(event.value.data, record());
+            assert_eq!(event.value.raw.as_slice().first(), Some(&0x94));
+            assert!(event
+                .value
+                .raw
+                .as_slice()
+                .ends_with(&[0xc4, 0x04, 0x00, 0x80, 0xff, 0x01]));
+            let id = event.value.raw.id();
+            drop(listener);
+            let values: Vec<_> =
+                wait(destination.get_typed::<Record>(&hash).try_collect()).unwrap();
+            assert!(values
+                .iter()
+                .any(|value| value.data == record() && value.raw.id() == id));
+            let values: Vec<_> = wait(destination.get(&hash).try_collect()).unwrap();
+            assert!(values.iter().any(|value| {
+                rmp_serde::from_slice::<Record>(value.as_slice()).unwrap() == record()
+            }));
+            wait(source.shutdown()).unwrap();
+            wait(destination.shutdown()).unwrap();
+            source.join().unwrap();
+            destination.join().unwrap();
+        }
+    }
+}
@@ -97,7 +97,7 @@ std::string
 DhtProxyServer::ServerStats::toString() const
 {
    auto ret = fmt::format("Listens: {}, Puts: {}, PushListeners: {}\n"
-                           "Push requests in the last {}: [Android: {}], [iOS: {}], [Unified: {}]\n"
+                           "Push requests in the last {}: [Android: {}], [iOS: {}], [Unified: {}], [Huawei: {}]\n"
                           "Requests: {} per second.",
                           listenCount,
                           putCount,
@@ -106,6 +106,7 @@ DhtProxyServer::ServerStats::toString() const
                           androidPush.toString(),
                           iosPush.toString(),
                           unifiedPush.toString(),
+                           huaweiPush.toString(),
                           requestRate);
    if (nodeInfo) {
        auto& ipv4 = nodeInfo->ipv4;
@@ -131,6 +132,7 @@ DhtProxyServer::ServerStats::toJson() const
    result["androidPush"] = androidPush.toJson();
    result["iosPush"] = iosPush.toJson();
    result["unifiedPush"] = unifiedPush.toJson();
+    result["huaweiPush"] = huaweiPush.toJson();
    result["requestRate"] = requestRate;
    if (nodeInfo)
        result["nodeInfo"] = nodeInfo->toJson();
@@ -639,6 +641,7 @@ DhtProxyServer::updateStats(std::shared_ptr<NodeInfo> info) const
        stats.androidPush = androidPush_;
        stats.iosPush = iosPush_;
        stats.unifiedPush = unifiedPush_;
+        stats.huaweiPush = huaweiPush_;
    }
 #endif
    stats.serverStartTime = serverStartTime_;
@@ -864,6 +867,8 @@ DhtProxyServer::getTypeFromString(const std::string& type)
        return PushType::iOS;
    else if (type == "unifiedpush")
        return PushType::UnifiedPush;
+    else if (type == "huawei")
+        return PushType::Huawei;
    return PushType::None;
 }

@@ -1275,15 +1280,25 @@ DhtProxyServer::sendPushNotification(
            Json::Value tokens(Json::arrayValue);
            tokens[0] = token;
            notification["tokens"] = std::move(tokens);
-            notification["platform"] = type == PushType::Android ? 2 : 1;
-            notification["data"] = std::move(json);
+            // gorush platform codes: 1 = iOS (APNs), 2 = Android (FCM), 3 = Huawei (HMS)
+            notification["platform"] = type == PushType::Android  ? 2
+                                       : type == PushType::Huawei ? 3
+                                                                  : 1;
            auto priority = highPriority ? "high" : "normal";
-            if (type == PushType::Android) {
+            if (type == PushType::Huawei) {
+                // HMS carries the payload as a serialized JSON string in its own field:
+                // the generic "data" object is ignored for this platform by gorush.
+                notification["huawei_data"] = Json::writeString(jsonBuilder_, json);
+                notification["huawei_ttl"] = "86400s"; // time to live = 24 hours
+                notification["priority"] = priority;
+            } else if (type == PushType::Android) {
+                notification["data"] = std::move(json);
                Json::Value androidConfig(Json::objectValue);
                androidConfig["priority"] = priority;
                androidConfig["ttl"] = "86400s"; // time to live = 24 hours
                notification["android"] = std::move(androidConfig);
            } else {
+                notification["data"] = std::move(json);
                notification["priority"] = priority;
                const auto expiration = std::chrono::system_clock::now() + std::chrono::hours(24);
                uint32_t exp = std::chrono::duration_cast<std::chrono::seconds>(expiration.time_since_epoch()).count();
@@ -1336,6 +1351,9 @@ DhtProxyServer::sendPushNotification(
        case PushType::UnifiedPush:
            unifiedPush_.increment(highPriority);
            break;
+        case PushType::Huawei:
+            huaweiPush_.increment(highPriority);
+            break;
        default:
            break;
        }
@@ -1,6 +1,6 @@
 {
  "name": "opendht",
-  "version-string": "4.4.99",
+  "version-string": "4.5.0-rc.1",
  "port-version": 1,
  "description": "OpenDHT: A lightweight C++17 Distributed Hash Table library.",
  "homepage": "https://opendht.net",
Author	SHA1	Message	Date
Adrien BeraudandCopilot	dff27203a4	build: prepare 4.5.0 release candidate 1 Co-authored-by: Copilot <223556219+Copilot@users.nply.github.com>	2026-11-03 22:48:09 -05:00
Adrien BeraudandCopilot	506d9f9cbb	ci: install MessagePack C++ headers on Ubuntu Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>	2026-11-03 22:44:50 -05:00
dependabot[bot]	349f4e5fc6	build(deps): bump pypa/cibuildwheel from 4.3.4 to 4.3.5 Bumps [pypa/cibuildwheel](https://github.com/pypa/cibuildwheel) from 4.3.4 to 4.3.5. - [Release notes](https://github.com/pypa/cibuildwheel/releases) - [Changelog](https://github.com/pypa/cibuildwheel/blob/main/docs/changelog.md) - [Commits](https://github.com/pypa/cibuildwheel/compare/v4.3.4...v4.3.5) --- updated-dependencies: - dependency-name: pypa/cibuildwheel dependency-version: 4.3.5 dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] <support@github.com>	2026-11-03 22:41:04 -05:00
dependabot[bot]	11d69d02e8	build(deps): bump actions/setup-node from 6 to 7 Bumps [actions/setup-node](https://github.com/actions/setup-node) from 6 to 7. - [Release notes](https://github.com/actions/setup-node/releases) - [Commits](https://github.com/actions/setup-node/compare/v6...v7) --- updated-dependencies: - dependency-name: actions/setup-node dependency-version: '7' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com>	2026-11-03 22:40:02 -05:00
Adrien Beraud	3c87991e92	docs: make Rust README precise and factual	2026-10-29 22:46:57 -04:00
Adrien Beraud	52287bf608	ci: use npm environment for publishing	2026-10-29 22:27:34 -04:00
Adrien Beraud	061ce2deb4	docker: build with RelWithDebInfo	2026-10-29 22:27:04 -04:00
Adrien Beraud	64b536fb53	ci: publish Python wheels from the tag release workflow	2026-10-29 21:48:33 -04:00
Adrien Beraud	67c76f17de	README: add Node.js bindings	2026-10-29 21:20:25 -04:00
Adrien Beraud	25fde756e8	docs: add linked package and Docker badges	2026-10-29 21:12:35 -04:00
Adrien Beraud	f6b3eda280	rust: rewrite README	2026-10-29 20:55:07 -04:00
Adrien Beraud	9aee407a42	rust: avoid dual-stack conflict in startup error test	2026-10-29 10:35:51 -04:00
Adrien Beraud	722ef44c7d	rust: move owned Value directly through callbacks and streams	2026-10-28 23:51:46 -04:00
Adrien Beraud	6ed2720bd7	rust: test single-copy async value ownership	2026-10-28 23:51:46 -04:00
Adrien Beraud	9005f73141	rust: add optional Serde typed operations	2026-10-28 23:51:46 -04:00
Adrien Beraud	9785c0a127	rust: add typed operation regression tests	2026-10-28 23:51:41 -04:00
Adrien Beraud	5662cc36d1	rust: publish crate on release with trusted publishing	2026-10-28 22:04:57 -04:00
Adrien Beraud	e116cfe548	rust: add optional runtime-independent async API	2026-10-28 21:11:43 -04:00
Adrien Beraud	1c5e8378b8	rust: add async adapter regression tests	2026-10-28 21:11:42 -04:00
Adrien Beraud	c630e80345	ci: use trusted publishing for Python wheels	2026-10-29 00:26:07 +00:00
cyrilleandCopilot	17f83edce4	proxy: add Huawei (HMS) push type Route push notifications to Huawei devices, which have no Google Play Services and therefore cannot receive FCM. gorush, already used as the push relay, supports HMS natively as platform 3. HMS takes its payload as a serialized JSON string in "huawei_data" and carries its own ttl/priority fields, so it needs its own branch rather than the generic "data" object used for FCM and APNs. The enum value is appended last: MSGPACK_ADD_ENUM serializes it as an integer, so reordering would break compatibility with deployed proxies. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>	2026-10-29 00:02:02 +00:00
Adrien Beraud	afab251376	nodejs: isolate native Linux dependency symbols Use RTLD_DEEPBIND when loading the Linux addon so its llhttp calls do not resolve to Node's incompatible bundled parser. The CI image uses llhttp 9.2.1 while Node 22 exports 9.4.3, corrupting the heap when the proxy client starts. Preserve default loading on other platforms.	2026-10-27 23:04:32 -04:00
Adrien Beraud	86d452fbb6	nodejs: bundle source-built llhttp license notices The Noble dependency image builds llhttp from source, so dpkg-query cannot identify its license package. Include the pinned upstream MIT notice when bundling Linux llhttp libraries while retaining package-owner checks for other dependencies.	2026-10-27 23:04:32 -04:00
Adrien Beraud	08ff775232	ci: rely on npm trusted publishing	2026-10-27 22:48:58 -04:00
Adrien Beraud	738e0e22bf	nodejs: add streaming iterators and startup proxy configuration	2026-10-27 22:46:34 -04:00
Adrien Beraud	8ec5b47c86	nodejs: test streaming iterators and proxy configuration	2026-10-27 22:09:21 -04:00
Adrien Beraud	e1c31568f9	readme: cleanup	2026-10-27 22:01:05 -04:00