rust: move owned Value directly through callbacks and streams

rust/README.md

@@ -38,14 +38,18 @@ 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.
+runner and crypto handles remain thread-bound. `Value` is `Send`, not `Sync`:
+it can move between threads, and mutation requires exclusive `&mut` access.
+Get/listen callbacks receive ordinary owned `Box<Value>` objects and can move
+them directly through channels. 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.
+The same `Value` type is used for construction, callback delivery, async streams,
+and publication. `as_slice()` borrows the payload; `user_type()` and `push_type()`
+borrow strings. `owner()` returns a retained public-key handle without exporting
+it. `as_bytes()` explicitly copies the payload into a Rust `Vec<u8>`.
+Received values can be edited directly and passed to `put` without converting
+or cloning them. `boxed()` explicitly creates an additional independent copy.
 
 ```rust
 use opendht::{DhtRunner, InfoHash, Value};
@@ -56,11 +60,12 @@ 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()); }
+		if !expired { let _ = sent.send(value); }
 		true
 })?;
 runner.put(&hash, Value::new("hello"), |_| {}, false);
-let bytes = received.recv_timeout(std::time::Duration::from_secs(5))?;
+let value = received.recv_timeout(std::time::Duration::from_secs(5))?;
+assert_eq!(value.as_slice(), b"hello");
 runner.cancel_listen(&hash, token);
 runner.join()?;
 # Ok::<(), Box<dyn std::error::Error>>(())
@@ -120,13 +125,13 @@ runner.join()?;
 	concurrently and succeeds when any peer responds. Resolve hostnames through
 	the application's runtime before calling it.
 - `get`, `get_filtered`, and `query` return fallible streams. Get/listen values
-	are independent `ValueSnapshot`s containing payload, ID, type, sequence,
-	priority, user/push types, signer public-key bytes, and recipient. Listener
-	`ValueEvent`s also preserve the expiration flag.
+	are the same owned `Box<Value>` objects delivered by the callback API,
+	with borrowed access to payload and metadata. Listener `ValueEvent`s also
+	preserve the expiration flag.
 - Streams buffer at most 64 items by default; `with_capacity` changes this
 	limit. A full buffer stops delivery and yields `WouldBlock` after buffered
 	items, then ends. Restart the operation to obtain a fresh snapshot. Native
-	failures, invalid metadata, and runner termination also yield errors rather
+	failures and runner termination also yield errors rather
 	than successful EOF. Duplicate native deliveries are not suppressed.
 - Dropping a listener immediately queues native cancellation and releases its
 	token. Listeners retain the runner until dropped. Dropping get/query streams
@@ -134,11 +139,25 @@ runner.join()?;
 	Dropping a completion future abandons its result but does not cancel the
 	native operation or undo a permanent put.
 - Runner adapters and listeners remain thread-bound. Use `block_on`, a local
-	executor, or Tokio `LocalSet`, not `tokio::spawn`. Snapshots and query rows can
+	executor, or Tokio `LocalSet`, not `tokio::spawn`. Received values and query rows can
 	be sent to other threads. Non-threaded runners still require independent
 	`loop_once()` calls while waiting. `join()` remains blocking; await graceful
 	`shutdown()` first when persistence is required.
 
+Received values are copied once by the native callback bridge to isolate them
+from mutable DHT storage. Async streams move those exact boxes through their
+channels without copying payloads or rebuilding metadata. Borrowed data remains
+valid for the value's lifetime, even after the callback returns or the runner
+stops. Put operations consume their box, so Rust cannot continue mutating a value
+after native code retains it. Malformed UTF-8 metadata is reported when its
+string accessor is called, rather than eagerly failing the entire stream.
+
+`value.owner()` returns the existing `PublicKey` type, not a second key class or
+DER bytes. Call `pack()` only when exported bytes are actually needed. Shared
+const owner-key fingerprint caches are initialized on the callback thread before
+transfer. Replacing a returned key through `unpack()` changes that handle, not
+the value's owner key.
+
 ## Typed Records
 
 Enable `serde` for typed MessagePack operations on `AsyncDhtRunner`. This feature
@@ -175,7 +194,7 @@ let record = Service {
 };
 runner.put_typed(&hash, &record, false).await?;
 if let Some(event) = updates.try_next().await? {
-	println!("{}: {:?}", event.value.raw.id, event.value.data);
+	println!("{}: {:?}", event.value.raw.id(), event.value.data);
 }
 drop(updates);
 let records: Vec<_> = runner.get_typed::<Service>(&hash).try_collect().await?;
@@ -184,8 +203,10 @@ let records: Vec<_> = runner.get_typed::<Service>(&hash).try_collect().await?;
 Typed puts accept `Serialize`, including unsized values. Typed reads require
 `DeserializeOwned`, without imposing `Clone`, `Send`, or `Unpin` on the record.
 Decoding happens when the consumer polls, not in native callbacks. Each
-`TypedValue<T>` exposes decoded `data` and the original `raw` snapshot, retaining
-payload bytes and all metadata. `TypedEvent<T>` preserves the expiration flag.
+`TypedValue<T>` exposes decoded `data` and the original owned `raw: Box<Value>`, retaining
+payload bytes and all metadata. Decoding borrows the raw payload directly, with
+no intermediate byte-buffer copy; allocations required by the decoded record
+still apply. `TypedEvent<T>` preserves the expiration flag.
 
 `TypedError` distinguishes `Serialize`, `Deserialize { id, source }`, and native
 `Operation` errors, preserving the underlying error through `Error::source()`.
@@ -199,7 +220,7 @@ only when polled. Serialization failure does not start a native put. Dropping
 the future after publication starts does not undo the put. Typed puts use the
 default native metadata; for custom IDs/types or signed/encrypted publication,
 serialize into a `Value` and use the existing untyped methods. Raw get/listen
-snapshots can also be decoded with `TypedValue::<T>::try_from(snapshot)` or
+values can also be decoded with `TypedValue::<T>::try_from(value)` or
 `TypedEvent::<T>::try_from(event)`.
 
 The wire payload is compact MessagePack, following OpenDHT's serialization

rust/examples/dhtnode_async.rs

@@ -26,8 +26,8 @@ fn main() -> Result<(), Box<dyn Error>> {
             .ok_or_else(|| io::Error::other("listener ended before receiving a value"))??;
         println!(
             "LISTEN {}: {} (expired: {})",
-            event.value.id,
-            String::from_utf8_lossy(&event.value.data),
+            event.value.id(),
+            String::from_utf8_lossy(event.value.as_slice()),
             event.expired,
         );
         drop(updates);
@@ -60,7 +60,8 @@ fn main() -> Result<(), Box<dyn Error>> {
             })??;
             println!(
                 "TYPED LISTEN {}: {:?}",
-                event.value.raw.id, event.value.data
+                event.value.raw.id(),
+                event.value.data
             );
             drop(updates);
             let values: Vec<_> = destination

rust/src/aio.rs

@@ -1,7 +1,8 @@
 //! 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`; streamed snapshots contain only owned, thread-safe data.
+//! 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};
@@ -16,43 +17,9 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 use std::task::{Context, Poll};
 
-/// An independent value snapshot containing no native handles.
-pub struct ValueSnapshot {
-    pub data: Vec<u8>,
-    pub id: u64,
-    pub value_type: u16,
-    pub seq: u16,
-    pub priority: u32,
-    pub user_type: String,
-    pub push_type: String,
-    /// DER-encoded public key of the signer, if present.
-    pub owner: Option<Vec<u8>>,
-    pub recipient: InfoHash,
-}
-
-impl ValueSnapshot {
-    fn copy(value: &Value) -> io::Result<Self> {
-        Ok(Self {
-            data: value.as_bytes(),
-            id: value.id(),
-            value_type: value.value_type(),
-            seq: value.seq(),
-            priority: value.priority(),
-            user_type: value
-                .user_type()
-                .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?,
-            push_type: value
-                .push_type()
-                .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?,
-            owner: value.owner().map(|key| key.pack()).transpose()?,
-            recipient: value.recipient(),
-        })
-    }
-}
-
 /// A listener update; expiration events retain the expired value's metadata.
 pub struct ValueEvent {
-    pub value: ValueSnapshot,
+    pub value: Box<Value>,
     pub expired: bool,
 }
 
@@ -142,7 +109,7 @@ impl<T> StreamSender<T> {
 /// A bounded get or query stream.
 ///
 /// Buffered items precede a terminal error. Native failure, runner shutdown,
-/// invalid metadata, and buffer overflow are never reported as successful EOF.
+/// 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"]
@@ -374,12 +341,12 @@ impl AsyncDhtRunner {
         pending
     }
 
-    pub fn get(&self, hash: &InfoHash) -> OperationStream<ValueSnapshot> {
+    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(ValueSnapshot::copy(&value)),
+            move |value| sender.send(Ok(value)),
             move |success| completion.finish(outcome(success)),
         );
         stream
@@ -389,13 +356,13 @@ impl AsyncDhtRunner {
         &self,
         hash: &InfoHash,
         filter: &str,
-    ) -> io::Result<OperationStream<ValueSnapshot>> {
+    ) -> 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(ValueSnapshot::copy(&value)),
+            move |value| sender.send(Ok(value)),
             move |success| completion.finish(outcome(success)),
         )?;
         Ok(stream)
@@ -422,9 +389,7 @@ impl AsyncDhtRunner {
         let token = self.state.runner.borrow_mut().listen_filtered(
             hash,
             filter,
-            move |value, expired| {
-                sender.send(ValueSnapshot::copy(&value).map(|value| ValueEvent { value, expired }))
-            },
+            move |value, expired| sender.send(Ok(ValueEvent { value, expired })),
         )?;
         Ok(Listener {
             state: self.state.clone(),

rust/src/dhtrunner.rs

@@ -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

rust/src/typed.rs

@@ -30,7 +30,7 @@
 //!     };
 //!     runner.put_typed(key, &record, false).await?;
 //!     if let Some(event) = updates.try_next().await? {
-//!         println!("{}: {:?}", event.value.raw.id, event.value.data);
+//!         println!("{}: {:?}", event.value.raw.id(), event.value.data);
 //!     }
 //!     drop(updates);
 //!     let records: Vec<_> = runner.get_typed::<Service>(key).try_collect().await?;
@@ -38,7 +38,7 @@
 //! }
 //! ```
 
-use crate::aio::{AsyncDhtRunner, ValueEvent, ValueSnapshot};
+use crate::aio::{AsyncDhtRunner, ValueEvent};
 use crate::{InfoHash, Value};
 use futures_core::Stream;
 use serde::{de::DeserializeOwned, Serialize};
@@ -51,7 +51,7 @@ 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: ValueSnapshot,
+    pub raw: Box<Value>,
 }
 
 impl<T: fmt::Debug> fmt::Debug for TypedValue<T> {
@@ -59,8 +59,8 @@ impl<T: fmt::Debug> fmt::Debug for TypedValue<T> {
         formatter
             .debug_struct("TypedValue")
             .field("data", &self.data)
-            .field("id", &self.raw.id)
-            .field("seq", &self.raw.seq)
+            .field("id", &self.raw.id())
+            .field("seq", &self.raw.seq())
             .finish_non_exhaustive()
     }
 }
@@ -106,12 +106,15 @@ impl Error for TypedError {
     }
 }
 
-impl<T: DeserializeOwned> TryFrom<ValueSnapshot> for TypedValue<T> {
+impl<T: DeserializeOwned> TryFrom<Box<Value>> for TypedValue<T> {
     type Error = TypedError;
 
-    fn try_from(raw: ValueSnapshot) -> Result<Self, Self::Error> {
-        let data = rmp_serde::from_slice(&raw.data)
-            .map_err(|source| TypedError::Deserialize { id: raw.id, source })?;
+    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 })
     }
 }

rust/src/value.rs

@@ -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)?;