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-10 13:27:12 +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
27 changed files with 1720 additions and 149 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,7 +64,7 @@ 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
@@ -151,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
@@ -194,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/
@@ -626,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')
@@ -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