anng: adapt to NNG v2.0.0-alpha.7 API (#42)

* anng: adapt to NNG v2.0.0-alpha.7 API

Refactor address handling to use URL strings instead of a complex enum:

- Update nng-sys to 0.4.0-v2pre.2 (path dependency)
- Replace `Addr` enum (Inproc, Ipc, Inet, Inet6, Zt, Abstract) with `Url`
  wrapper around String, matching NNG's URL-based addressing model.
  Use `nng_url_sprintf` for URL formatting in `Listener::local_addr()`
- Use `nng_pipe_peer_addr` instead of deprecated `nng_pipe_get_addr`
- Retrieve URL scheme from pipe's listener/dialer endpoint
- Check pipe validity with `nng_pipe_id()` instead of internal field
- Enable full feature-powerset CI checks for anng

* anng: use unreachable! for validated pipe in remote_addr

The NNG_ENOENT branch in nng_pipe_peer_addr cannot be reached
because self.pipe() already validates the pipe before this point.

* anng: use unreachable! for invalid dialer/listener error paths

Replace verbose error-logging-and-debug-assert patterns with
unreachable! in remote_addr URL retrieval, consistent with the
approach taken for the validated pipe path.

* anng: make log message capitalization consistent

* anng: add TODO for upstream NNG issue #2228

* anng: accept &CStr in Url::from_nng to avoid caller conversion

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* anng: handle stale pipe in remote_addr gracefully

nng_pipe_peer_addr can return NNG_ENOENT when the pipe was closed
after self.pipe() returned a stale ID. Return None instead of
panicking.

* anng: replace unwrap with expect in Url::from_nng write! calls

* anng: use SocketAddrV6 Display for IPv6 URL formatting

* anng: restore lifetime nuance in CStr SAFETY comments

* anng: use split_once for URL scheme parsing in parse_tcp_url

* anng: use unreachable! for NNG_ENOENT in Listener::local_addr

&self guarantees the listener cannot be closed, so NNG_ENOENT
is unreachable.

* anng: take &nng_sockaddr by reference in Url::from_nng

* anng: validate UTF-8 on stack before allocating in local_addr

* anng: add TODO

* anng: preserve IPv6 scope_id in parse_tcp_url

SocketAddr::from_str() supports numeric IPv6 scope IDs, so there's no need
to strip them during parsing. This fixes data loss and maintains consistency
with Url::from_nng() which already preserves scope_id.

* anng: fix MSRV issue (core vs std)

---------

Co-authored-by: Felix Obenhuber <felix.obenhuber@helsing.ai>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: 3 people

.github/workflows/check.yml

@@ -127,7 +127,7 @@ jobs:
       # The effect of the failure (when running `hack` on the workspace) is here:
       # https://github.com/nanomsg/nng-rs/actions/runs/20988550343/job/60328143824
       #
-      # TODO: Remove this workaround once nng and anng are migrated to nng-sys v0.4/NNG v2
+      # TODO: Remove this workaround once nng is migrated to nng-sys v0.4/NNG v2
       # and revert to:
       # - name: cargo hack
       #   run: cargo hack --feature-powerset check --skip source-update-bindings
@@ -141,9 +141,7 @@ jobs:
       - name: build and install NNG v2 for nng-sys
         run: .github/scripts/install-nng.sh
       - name: cargo hack anng
-        # TODO(flxo): check anng just with feature vendored enabled because the needed nng version
-        # alpha.6 is not available
-        run: cargo hack -p anng --features vendored check
+        run: cargo hack -p anng --feature-powerset check
         env:
           # add /usr/local/include to the C compiler search path for NNG headers
           CFLAGS: -I/usr/local/include

Cargo.lock

@@ -27,9 +27,8 @@ tracing = "0.1.41"
 lazy_static = { version = "1.5.0", optional = true }
 
 [dependencies.nng-sys]
-version = "0.4.0-v2pre.1"
-# TODO(flxo): make anng use the nng-sys as path dependency from the workspace when updated
-# path = "../nng-sys"
+version = "0.4.0-v2pre.2"
+path = "../nng-sys"
 default-features = false
 features = ["std"]
 
@@ -40,6 +39,5 @@ tokio = ["tokio/rt-multi-thread"]
 
 [dev-dependencies]
 lexopt = "0.3"
-rstest = { version = "0.26.1", default-features = false }
 tokio = { version = "1.47.1", features = ["full"] }
 tracing-subscriber = "0.3.20"

anng/Cargo.toml

@@ -1,8 +1,8 @@
-use crate::{AioError, pipes::Addr};
+use crate::{AioError, pipes::Url};
 use bytes::{Buf, BufMut};
 use core::{
     cmp::max,
-    ffi::{c_char, c_void},
+    ffi::c_void,
     fmt,
     mem::MaybeUninit,
     ops::{Deref, DerefMut},
@@ -386,44 +386,124 @@ impl Message {
         }
     }
 
-    /// Returns the remote address of the message (ie, where it was received from), if available.
-    pub fn remote_addr(&self) -> Option<Addr> {
+    /// Returns the remote URL of the message (ie, where it was received from), if available.
+    pub fn remote_addr(&self) -> Option<Url> {
+        use core::ffi::CStr;
+
         let pipe = self.pipe()?;
 
-        // SAFETY: arg and addr are both valid, and on success, nng_pipe_get_addr initializes
-        let addr = unsafe {
-            let mut addr = MaybeUninit::<nng_sys::nng_sockaddr>::uninit();
-            let errno = nng_sys::nng_pipe_get_addr(
-                pipe,
-                nng_sys::NNG_OPT_REMADDR as *const _ as *const c_char,
-                addr.as_mut_ptr(),
-            );
-            match errno {
-                nng_err::NNG_OK => addr.assume_init(),
-                nng_err::NNG_ENOTSUP => {
-                    tracing::warn!("Message pipe does not support REMADDR");
-                    return None;
-                }
-                nng_err::NNG_ENOENT => {
-                    tracing::warn!("Message does not have a pipe");
-                    return None;
-                }
-                err if err.0 & nng_err::NNG_ESYSERR.0 != 0 => {
-                    tracing::warn!(
-                        "nng_pipe_get_addr returned a system error: {}",
-                        io::Error::from_raw_os_error((err.0 & !(nng_err::NNG_ESYSERR.0)) as i32)
-                    );
+        // Get the peer address from the pipe.
+        let mut addr = MaybeUninit::<nng_sys::nng_sockaddr>::uninit();
+        // SAFETY: pipe is valid (from self.pipe()), addr is valid for writing.
+        let errno = unsafe { nng_sys::nng_pipe_peer_addr(pipe, addr.as_mut_ptr()) };
+        let addr = match errno {
+            // SAFETY: nng_pipe_peer_addr initializes addr on success.
+            nng_err::NNG_OK => unsafe { addr.assume_init() },
+            nng_err::NNG_ENOTSUP => {
+                tracing::warn!("Message pipe does not support peer address");
+                return None;
+            }
+            nng_err::NNG_ENOENT => {
+                // The pipe ID stored in the message is stale - the pipe was closed
+                // between self.pipe() (which only checks the stored ID) and the
+                // nng_pipe_peer_addr call.
+                return None;
+            }
+            err if err.0 & nng_err::NNG_ESYSERR.0 != 0 => {
+                tracing::warn!(
+                    "nng_pipe_peer_addr returned a system error: {}",
+                    io::Error::from_raw_os_error((err.0 & !(nng_err::NNG_ESYSERR.0)) as i32)
+                );
+                return None;
+            }
+            _ => {
+                unreachable!(
+                    "nng_pipe_peer_addr documentation claims err \"{errno}\" is never returned"
+                );
+            }
+        };
+
+        // Get the URL scheme from the pipe's originating endpoint (listener or dialer).
+        //
+        // NNG does not provide a direct API to get a URL or transport type from a pipe.
+        // There's no `nng_pipe_get_url()` or transport type accessor. The peer address
+        // (`nng_sockaddr`) only contains address family, not the URL scheme. We must
+        // retrieve the scheme from the endpoint that created this pipe.
+        //
+        // TODO(flxo): once upstream fixes [2228](https://github.com/nanomsg/nng/issues/2228),
+        //             simplify this.
+        //
+        // In NNG, a pipe represents a single connection and is created by exactly one of:
+        // - A listener: when accepting an incoming connection
+        // - A dialer: when initiating an outgoing connection
+        //
+        // We use nng_listener_id/nng_dialer_id to determine which endpoint created this pipe.
+        // These functions return a positive ID for valid endpoints, or -1 for invalid ones.
+        // Only one will be valid for any given pipe.
+        //
+        // Note on URL ownership: nng_listener_get_url/nng_dialer_get_url return a borrowed
+        // pointer to NNG's internal URL structure (hence `const nng_url **`). This URL is
+        // owned by the listener/dialer and will be freed when that endpoint closes.
+        // We do NOT call nng_url_free here - that is only for URLs created by nng_url_parse().
+        let mut urlp: *const nng_sys::nng_url = core::ptr::null();
+
+        // Try to get URL from listener first, then dialer.
+        // Note: The endpoint could be closed between checking the ID and getting the URL.
+        // nng_listener_get_url and nng_dialer_get_url can only return:
+        // - 0 (NNG_OK): success
+        // - NNG_ENOENT: endpoint is invalid or was closed
+        // See nni_listener_find/nni_dialer_find in nng/src/core/listener.c and dialer.c.
+        const NNG_ENOENT: i32 = nng_sys::nng_err::NNG_ENOENT.0 as i32;
+
+        // SAFETY: pipe is valid (from self.pipe()).
+        let listener = unsafe { nng_sys::nng_pipe_listener(pipe) };
+        // SAFETY: listener is a stack value returned from nng_pipe_listener.
+        if unsafe { nng_sys::nng_listener_id(listener) } > 0 {
+            // SAFETY: listener is valid (positive ID), urlp is valid for writing.
+            let result = unsafe { nng_sys::nng_listener_get_url(listener, &mut urlp) };
+            match result {
+                0 => {} // success, urlp is now set
+                NNG_ENOENT => {
+                    tracing::debug!("listener was closed while retrieving URL");
                     return None;
                 }
-                _ => {
-                    unreachable!(
-                        "nng_pipe_get_addr documentation claims err \"{errno}\" is never returned"
-                    );
+                err => unreachable!("unexpected error from nng_listener_get_url: {err}"),
+            }
+        } else {
+            // SAFETY: pipe is valid (from self.pipe()).
+            let dialer = unsafe { nng_sys::nng_pipe_dialer(pipe) };
+            // SAFETY: dialer is a stack value returned from nng_pipe_dialer.
+            if unsafe { nng_sys::nng_dialer_id(dialer) } > 0 {
+                // SAFETY: dialer is valid (positive ID), urlp is valid for writing.
+                let result = unsafe { nng_sys::nng_dialer_get_url(dialer, &mut urlp) };
+                match result {
+                    0 => {} // success, urlp is now set
+                    NNG_ENOENT => {
+                        tracing::debug!("dialer was closed while retrieving URL");
+                        return None;
+                    }
+                    err => unreachable!("unexpected error from nng_dialer_get_url: {err}"),
                 }
+            } else {
+                // Neither listener nor dialer is valid - pipe may have been closed
+                return None;
+            }
+        }
+
+        let scheme = if !urlp.is_null() {
+            // SAFETY: urlp is non-null and was set by nng_listener_get_url or nng_dialer_get_url.
+            let scheme_ptr = unsafe { nng_sys::nng_url_scheme(urlp) };
+            if !scheme_ptr.is_null() {
+                // SAFETY: scheme_ptr is non-null and points to a valid C string owned by NNG.
+                unsafe { CStr::from_ptr(scheme_ptr) }
+            } else {
+                return None;
             }
+        } else {
+            return None;
         };
 
-        Addr::from_nng(addr)
+        Url::from_nng(&addr, scheme)
     }
 
     /// Reserves the minimum capacity for at least additional more bytes to be