A vsock reader wired through tokio `AsyncFd` is never woken after the first readable event

On Hermit after PR #2434 is merged (which makes accep return a real connection socket), a non-blocking `AF_VSOCK` socket driven through `tokio::io::unix::AsyncFd` stops being woken after the **first** readable event. The first read works; every subsequent `readable().await` on the same fd hangs forever, even though the peer has sent more data.

The guest listens on a vsock port, accepts one connection, and reads in "waves" (await readiness, drain to `WouldBlock`) and responds with "received" back. The host sends `test`, waits, then sends `test` again. Observed guest output:

```
REPRO: listening on vsock port 9999
REPRO: accepted connection
REPRO: before awaiting readable. will hang on next line after second call
REPRO: after awaiting readable, we only reach this first call
REPRO: wave 1 = "test\n"
REPRO: awaiting wave 2...
REPRO: before awaiting readable. will hang on next line after second call
```

Wave 1 is read and responded too. Wave 2 never wakes the reader.

A control using the **blocking** vsock path (no tokio) — the existing hermit-rs `vsock` example — works fine on the same VM, so this is specific to the `AsyncFd`/`SourceFd` readiness path, not the vsock transport or the kernel's accept/read implementation.

## Root cause

mio's `poll(2)` selector is edge-triggered by construction. After an fd's event fires, it clears that fd's interest from the pollfd [`mio-1.2.1/src/sys/unix/selector/poll.rs`](https://github.com/tokio-rs/mio/blob/release-v1.2.1/src/sys/unix/selector/poll.rs#L292):

```rust
// Remove the interest which just got triggered the IoSourceState's do_io
// wrapper used with this selector will add back the interest using
// reregister.
poll_fd.events &= !poll_fd.revents;
```

As the comment says, it relies on `IoSourceState::do_io` to re-arm the interest.
But `do_io` only reregisters when the I/O closure returns `WouldBlock`
[`mio-1.2.1/src/sys/unix/selector/poll.rs`](https://github.com/tokio-rs/mio/blob/release-v1.2.1/src/sys/unix/selector/poll.rs#L726):

```rust
if err.kind() == io::ErrorKind::WouldBlock {
self.inner.as_ref().map_or(Ok(()), |state| {
state
.selector
.reregister(state.fd, state.token, state.interests)
})?;
}
```

tokio's `AsyncFd` registers the fd via `mio::unix::SourceFd` (a raw fd) and never routes I/O through mio's `IoSource`, so `do_io` is never called. The interest is cleared once and never restored. The next `readable()` therefore waits on an interest mask that no longer contains the read bit, and never fires.

**NOTE:** This does not affect TcpStream/UdpSocket: tokio wraps them in mio's native mio::net::TcpStream/UdpSocket, which hold the fd as an IoSource and route all I/O through IoSource::do_io — the method that reregisters interest on WouldBlock (poll.rs:733). Only raw fds registered via SourceFd (as AsyncFd does, and as this vsock example does) miss the rearm and get stuck

## Reproducer

A self-contained `main.rs` example below. It needs the code in the second block `raw_vsock.rs` to compile and run.

```rust
// main.rs
use tokio::io::unix::AsyncFd;

async fn read_wave(afd: &AsyncFd) -> std::io::Result> {
loop {
println!("REPRO: before awaiting readable. will hang on next line after second call");
let mut guard = afd.readable().await?;
println!("REPRO: after awaiting readable, we only reach this first call");
let mut out = Vec::new();
let mut buf = [0u8; 256];
loop {
match guard.try_io(|inner| {
use std::io::Read as _;
(&*inner.get_ref()).read(&mut buf)
}) {
Ok(Ok(0)) => return Ok(out), // EOF
Ok(Ok(n)) => out.extend_from_slice(&buf[..n]),
Ok(Err(e)) => return Err(e),
Err(_would_block) => {
if !out.is_empty() {
return Ok(out);
}
break; // spurious wake with no data: re-await
}
}
}
}
}

async fn run() -> std::io::Result<()> {
let listener = listen_vsock(PORT)?;
println!("REPRO: listening on vsock port {PORT}");

let conn = accept_one(&listener).await?;
println!("REPRO: accepted connection");

// Wave 1 works: the fresh `AsyncFd` registration arms the interest.
let first = read_wave(&conn).await?;
println!("REPRO: wave 1 = {:?}", String::from_utf8_lossy(&first));
write_all(&conn, b"received").await?;

// Wave 2 hangs: mio cleared the interest after wave 1 and `AsyncFd`
// (SourceFd) is never re-armed, so this `readable()` never wakes.
println!("REPRO: awaiting wave 2...");
let second = read_wave(&conn).await?;
println!("REPRO: wave 2 = {:?}", String::from_utf8_lossy(&second));
write_all(&conn, b"received again").await?;

println!("REPRO: responded to wave 2 -- OK (bug NOT present)");
Ok(())
}
```

And for completeness the supporting code:
```rust
// raw_vsock.rs

use std::io::{self, Read, Write};
use std::mem::size_of;
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd};

#[cfg(target_os = "hermit")]
use hermit_abi::{
accept, bind, fcntl, listen, read, sockaddr, sockaddr_vm, socket, socklen_t, write, AF_VSOCK,
F_SETFL, O_NONBLOCK, SOCK_STREAM, VMADDR_CID_ANY,
};

use tokio::io::unix::AsyncFd;

fn last_err() -> io::Error {
io::Error::last_os_error()
}

fn make_sockaddr_vm(cid: u32, port: u32) -> sockaddr_vm {
sockaddr_vm {
#[cfg(target_os = "hermit")]
svm_len: size_of::() as u8,
svm_reserved1: 0,
svm_family: AF_VSOCK as _,
svm_cid: cid,
svm_port: port,
svm_zero: [0; 4],
}
}

fn set_nonblocking(fd: RawFd) -> io::Result<()> {
if unsafe { fcntl(fd, F_SETFL, O_NONBLOCK) } < 0 {
return Err(last_err());
}
Ok(())
}

pub struct RawVsock {
fd: OwnedFd,
}

impl RawVsock {
fn from_raw(fd: RawFd) -> io::Result {
set_nonblocking(fd)?;
Ok(Self {
fd: unsafe { OwnedFd::from_raw_fd(fd) },
})
}
}

impl AsRawFd for RawVsock {
fn as_raw_fd(&self) -> RawFd {
self.fd.as_raw_fd()
}
}

impl Read for &RawVsock {
fn read(&mut self, buf: &mut [u8]) -> io::Result {
let r = unsafe { read(self.fd.as_raw_fd(), buf.as_mut_ptr(), buf.len()) };
if r < 0 {
Err(last_err())
} else {
Ok(r as usize)
}
}
}

impl Write for &RawVsock {
fn write(&mut self, buf: &[u8]) -> io::Result {
let r = unsafe { write(self.fd.as_raw_fd(), buf.as_ptr(), buf.len()) };
if r < 0 {
Err(last_err())
} else {
Ok(r as usize)
}
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}

/// Bind + listen on `VMADDR_CID_ANY:port` and return the listener as an
/// `AsyncFd`.
pub fn listen_vsock(port: u32) -> io::Result> {
let lfd = unsafe { socket(AF_VSOCK, SOCK_STREAM, 0) };
if lfd < 0 {
return Err(last_err());
}
let addr = make_sockaddr_vm(VMADDR_CID_ANY, port);
if unsafe {
bind(
lfd,
&addr as *const _ as *const sockaddr,
size_of::() as socklen_t,
)
} < 0
{
return Err(last_err());
}
if unsafe { listen(lfd, 1) } < 0 {
return Err(last_err());
}
AsyncFd::new(RawVsock::from_raw(lfd)?)
}

pub async fn accept_one(listener: &AsyncFd) -> io::Result> {
loop {
let mut guard = listener.readable().await?;
let mut peer = make_sockaddr_vm(0, 0);
let mut len = size_of::() as socklen_t;
match guard.try_io(|inner| {
let fd = unsafe {
accept(
inner.get_ref().as_raw_fd(),
&mut peer as *mut _ as *mut sockaddr,
&mut len as *mut socklen_t,
)
};
if fd < 0 {
Err(last_err())
} else {
Ok(fd)
}
}) {
Ok(Ok(fd)) => return AsyncFd::new(RawVsock::from_raw(fd)?),
Ok(Err(e)) => return Err(e),
Err(_would_block) => continue,
}
}
}

/// Write all of `buf`, awaiting writability between partial writes.
pub async fn write_all(afd: &AsyncFd, mut buf: &[u8]) -> io::Result<()> {
while !buf.is_empty() {
let mut guard = afd.writable().await?;
match guard.try_io(|inner| (&*inner.get_ref()).write(buf)) {
Ok(Ok(n)) => buf = &buf[n..],
Ok(Err(e)) => return Err(e),
Err(_would_block) => continue,
}
}
Ok(())
}

```

Build the above with hermit and run it in qemu:
```
sudo qemu-system-x86_64 -enable-kvm -cpu host -smp 1 -m 512M -machine q35 \
-display none -serial stdio \
-kernel kernel/hermit-loader-x86_64 \
-initrd target/x86_64-unknown-hermit/debug/vsock-async-fd-issue \
-device vhost-vsock-pci,disable-legacy=on,guest-cid=3
```

Use e.g. `socat - VSOCK-CONNECT:3:9999` to connect and send a message that get's echoed back. Then try sending a second message. The AsyncFd will then cause a hang.

## Workaround
Consumers can force a re-arm by deregistering and reregistering the fd after each drained read — in practice, dropping and recreating the `AsyncFd` over the same raw fd (`into_inner()` deregisters without closing the fd; `AsyncFd::new` reregisters with fresh interest). `clear_ready()` alone is not enough: it resets tokio's internal readiness but not mio's pollfd interest mask.

## Possible directions

- Make the `poll(2)` selector re-arm interest for fds registered via `SourceFd` (not just those going through `IoSource::do_io`) — i.e. fix it in mio.
- Or handle the `SourceFd` + `poll(2)`-selector combination in the Hermit integration so `AsyncFd`-based code works without a per-read re-arm.
- Or switch to `epoll` for hermit.

Start with the Hermit vsock AsyncFd reproducer in main.rs and raw_vsock.rs, then inspect mio-1.2.1/src/sys/unix/selector/poll.rs around selector interest clearing and IoSourceState::do_io. Run the provided Hermit/QEMU command and connect with socat to confirm the second readable wave hangs. Done means repeated AsyncFd readable events work without dropping and recreating the AsyncFd, with the existing blocking vsock path unaffected.

Written by the indexing model from the issue text.