virtio: retire a queue after a fatal fetch error

vm/devices/virtio/virtio/src/common.rs

@@ -318,7 +318,14 @@ impl VirtioQueue {
     /// new data arrived during arming, it returns immediately without sleeping.
     /// On wakeup, kicks are suppressed to avoid unnecessary doorbells while
     /// the caller drains the queue.
+    ///
+    /// Returns `Poll::Pending` forever once the queue has failed, since no
+    /// further work can ever be fetched. This keeps callers that loop on
+    /// "kick, then retry" from spinning.
     pub fn poll_kick(&mut self, cx: &mut Context<'_>) -> Poll<()> {
+        if self.core.failed() {
+            return Poll::Pending;
+        }
         if self.core.arm_for_kick() {
             ready!(self.queue_event.wait().poll_unpin(cx)).expect("waits on Event cannot fail");
         }
@@ -424,6 +431,13 @@ impl VirtioQueue {
     }
 }
 
+/// Yields each descriptor chain the guest makes available.
+///
+/// After yielding a [`QueueError`] the queue is retired, and the stream parks
+/// forever rather than ending: returning `None` would be ready synchronously on
+/// every poll, so a caller looping over the stream would spin on it. Parking
+/// makes it impossible for any caller to busy-loop, and the worker stays
+/// cancellable.
 impl Stream for VirtioQueue {
     type Item = Result<VirtioQueueCallbackWork, Error>;
 

vm/devices/virtio/virtio/src/queue.rs

@@ -203,6 +203,19 @@ pub(crate) struct QueueCoreGetWork {
     inner: QueueGetWorkInner,
     /// Whether kick notification is currently armed.
     armed: bool,
+    /// Whether a fatal error has retired the fetch side of the queue.
+    ///
+    /// Every [`QueueError`] raised while fetching work means the guest violated
+    /// the queue protocol, and the offending chain is rejected *without* being
+    /// consumed — the available index does not move. Re-reading the ring would
+    /// therefore hit the same descriptors and raise the same error forever, so a
+    /// caller that logs the error and keeps polling would spin without making
+    /// progress. Latch the failure instead: report it once, then report the
+    /// queue as permanently empty.
+    ///
+    /// This only retires fetching. Descriptors already consumed can still be
+    /// completed.
+    failed: bool,
     /// Consumed-but-not-completed ring capacity, in the format's native unit:
     /// buffers (heads) for split, descriptors (slots) for packed — the two forms
     /// of the virtio "Queue Size" limit (spec §2.7.1 vs §2.8.1).
@@ -277,10 +290,17 @@ impl QueueCoreGetWork {
             mem,
             inner,
             armed: false,
+            failed: false,
             in_flight,
         })
     }
 
+    /// Whether a fatal error has retired the fetch side of this queue. Once
+    /// set, no further work will ever be returned. See [`Self::failed`].
+    pub fn failed(&self) -> bool {
+        self.failed
+    }
+
     pub fn try_next_work(&mut self) -> Result<Option<VirtioQueueCallbackWork>, QueueError> {
         match self.try_peek_work() {
             Ok(Some(work)) => {
@@ -296,7 +316,21 @@ impl QueueCoreGetWork {
     /// consume the peeked descriptor and move to the next one. Calling this
     /// again without advancing will return the same descriptor, but note that
     /// the guest may have modified the descriptor memory in the meantime.
+    ///
+    /// Returns `Ok(None)` forever once the queue has failed, so that the error
+    /// is reported exactly once.
     pub fn try_peek_work(&mut self) -> Result<Option<VirtioQueueCallbackWork>, QueueError> {
+        if self.failed {
+            return Ok(None);
+        }
+        let r = self.try_peek_work_inner();
+        if r.is_err() {
+            self.failed = true;
+        }
+        r
+    }
+
+    fn try_peek_work_inner(&mut self) -> Result<Option<VirtioQueueCallbackWork>, QueueError> {
         let index = match &mut self.inner {
             QueueGetWorkInner::Split(split) => split.is_available()?,
             QueueGetWorkInner::Packed(packed) => packed.is_available()?,

vm/devices/virtio/virtio/src/tests.rs

@@ -5578,12 +5578,15 @@ fn new_bound_test_queue_result(
     Ok((mem, queue))
 }
 
-fn assert_in_flight_error(error: io::Error, in_flight: u16, requested: u16) {
-    let error = error
+fn queue_error(error: &io::Error) -> Option<&QueueError> {
+    error
         .get_ref()
-        .and_then(|error| error.downcast_ref::<QueueError>());
+        .and_then(|error| error.downcast_ref::<QueueError>())
+}
+
+fn assert_in_flight_error(error: io::Error, in_flight: u16, requested: u16) {
     assert!(matches!(
-        error,
+        queue_error(&error),
         Some(QueueError::TooManyInFlightDescriptors {
             in_flight: actual_in_flight,
             requested: actual_requested,
@@ -5592,6 +5595,31 @@ fn assert_in_flight_error(error: io::Error, in_flight: u16, requested: u16) {
     ));
 }
 
+/// Points descriptor 0 at itself and makes it available. The chain never
+/// terminates, so the queue rejects it with [`QueueError::TooLong`].
+fn make_split_cyclic_chain(mem: &GuestMemory) {
+    use crate::test_helpers::make_available;
+    use crate::test_helpers::write_descriptor;
+
+    write_descriptor(
+        mem,
+        BOUND_TEST_DESC_ADDR,
+        0,
+        BOUND_TEST_DATA_ADDR,
+        0x100,
+        DescriptorFlags::new().with_write(true).with_next(true),
+        0,
+    );
+    let mut avail_idx = 0;
+    make_available(
+        mem,
+        BOUND_TEST_AVAIL_ADDR,
+        BOUND_TEST_QUEUE_SIZE,
+        0,
+        &mut avail_idx,
+    );
+}
+
 fn next_error(queue: &mut VirtioQueue) -> io::Error {
     match queue.try_next() {
         Err(error) => error,
@@ -5704,8 +5732,12 @@ async fn packed_queue_rejects_chain_exceeding_free_space(driver: DefaultDriver)
     assert_in_flight_error(next_error(&mut queue), 2, 3);
 }
 
+/// Oversubscribing the ring retires the queue like any other protocol
+/// violation. Completing outstanding work frees ring capacity but does not
+/// revive it: the rejected chain was never consumed, so the queue would hand
+/// out a descriptor the guest had already oversubscribed.
 #[async_test]
-async fn split_queue_capacity_recovers_after_completion(driver: DefaultDriver) {
+async fn split_queue_stays_retired_after_capacity_error(driver: DefaultDriver) {
     let (mem, mut queue) = new_bound_test_queue(&driver, false, None);
     make_split_bound_test_work(&mem, BOUND_TEST_QUEUE_SIZE + 1, 0);
 
@@ -5714,8 +5746,14 @@ async fn split_queue_capacity_recovers_after_completion(driver: DefaultDriver) {
         works.push(queue.try_next().unwrap().expect("within queue size"));
     }
     assert_in_flight_error(next_error(&mut queue), BOUND_TEST_QUEUE_SIZE, 1);
+
+    // Completions still publish to the used ring, so a device can drain its
+    // in-flight work after the failure.
     queue.complete(works.remove(0), 1);
-    let _ = queue.try_next().unwrap().unwrap();
+    assert!(
+        matches!(queue.try_next(), Ok(None)),
+        "freeing capacity must not revive a retired queue"
+    );
 }
 
 #[async_test]
@@ -5788,3 +5826,50 @@ async fn queue_new_rejects_corrupt_saved_state(driver: DefaultDriver) {
         ));
     }
 }
+
+/// A rejected descriptor chain is never consumed, so the available index does
+/// not move and the same chain would be re-read on the next call. Report the
+/// failure once and then report the queue as empty, so that a device worker
+/// which logs the error and keeps polling makes no further progress instead of
+/// spinning on the same bad chain forever.
+#[async_test]
+async fn queue_reports_failure_only_once(driver: DefaultDriver) {
+    let (mem, mut queue) = new_bound_test_queue(&driver, false, None);
+    make_split_cyclic_chain(&mem);
+
+    let Err(error) = queue.try_next() else {
+        panic!("a cyclic descriptor chain must be rejected");
+    };
+    assert!(matches!(queue_error(&error), Some(QueueError::TooLong)));
+
+    assert!(
+        matches!(queue.try_next(), Ok(None)),
+        "failed queue must report no work rather than repeat the error"
+    );
+    assert!(
+        matches!(queue.try_peek(), Ok(None)),
+        "failed queue must report no work rather than repeat the error"
+    );
+}
+
+/// Driven as a stream, a failed queue must park rather than yield the same
+/// error on every poll. It must not end the stream either: `None` would be
+/// ready synchronously on every poll, so a caller looping over the stream
+/// would spin on it exactly as it would on the repeated error.
+#[async_test]
+async fn queue_stream_parks_after_failure(driver: DefaultDriver) {
+    let (mem, mut queue) = new_bound_test_queue(&driver, false, None);
+    make_split_cyclic_chain(&mem);
+
+    let Some(Err(error)) = queue.next().await else {
+        panic!("the stream must yield the failure");
+    };
+    assert!(matches!(queue_error(&error), Some(QueueError::TooLong)));
+
+    assert!(
+        poll_fn(|cx| std::task::Poll::Ready(queue.poll_next_unpin(cx)))
+            .await
+            .is_pending(),
+        "failed queue must park rather than repeat the error or end the stream"
+    );
+}

vm/devices/virtio/virtio_blk/src/integration_tests.rs

@@ -11,15 +11,22 @@ use crate::VirtioBlkDevice;
 use disk_backend::Disk;
 use disk_backend::DiskError;
 use disk_backend::DiskIo;
+use futures::future::Either;
+use futures::future::select;
 use guestmem::GuestMemory;
 use guestmem::MemoryRead;
 use guestmem::MemoryWrite;
 use inspect::Inspect;
 use pal_async::DefaultDriver;
+use pal_async::DefaultPool;
 use pal_async::async_test;
+use pal_async::timer::PolledTimer;
 use pal_event::Event;
 use parking_lot::Mutex;
 use scsi_buffers::RequestBuffers;
+use std::future::Future;
+use std::pin::pin;
+use std::time::Duration;
 use test_with_tracing::test;
 use virtio::QueueResources;
 use virtio::VirtioDevice;
@@ -70,12 +77,26 @@ struct TestHarness {
 impl TestHarness {
     /// Create a harness with a RAM disk of the given size.
     fn new(driver: &DefaultDriver, disk: Disk, read_only: bool) -> Self {
+        Self::with_device_driver(driver, driver, disk, read_only)
+    }
+
+    /// Like [`TestHarness::new`], but runs the device's worker task on
+    /// `device_driver` rather than on the test's own executor. A test that
+    /// must stay responsive while the worker misbehaves puts the device on a
+    /// separate thread's executor.
+    fn with_device_driver(
+        driver: &DefaultDriver,
+        device_driver: &DefaultDriver,
+        disk: Disk,
+        read_only: bool,
+    ) -> Self {
         let mem = GuestMemory::allocate(TOTAL_MEM_SIZE);
 
         init_avail_ring(&mem, AVAIL_ADDR);
         init_used_ring(&mem, USED_ADDR);
 
-        let driver_source = VmTaskDriverSource::new(SingleDriverBackend::new(driver.clone()));
+        let driver_source =
+            VmTaskDriverSource::new(SingleDriverBackend::new(device_driver.clone()));
         let device = VirtioBlkDevice::new(&driver_source, disk, read_only);
 
         let queue_event = Event::new();
@@ -510,6 +531,20 @@ fn ram_disk(size: u64, read_only: bool) -> Disk {
     disklayer_ram::ram_disk(size, read_only).unwrap()
 }
 
+/// Awaits `fut`, panicking with `msg` if it does not complete within `timeout`.
+async fn with_timeout<F: Future>(
+    driver: &DefaultDriver,
+    timeout: Duration,
+    msg: &str,
+    fut: F,
+) -> F::Output {
+    let mut timer = PolledTimer::new(driver);
+    match select(pin!(fut), pin!(timer.sleep(timeout))).await {
+        Either::Left((output, _)) => output,
+        Either::Right(_) => panic!("{msg}"),
+    }
+}
+
 /// Write 1 sector then read it back. Verifies basic write and read roundtrip.
 #[async_test]
 async fn write_then_read_roundtrip(driver: DefaultDriver) {
@@ -1284,3 +1319,62 @@ async fn bounce_buffer_write_read_roundtrip(driver: DefaultDriver) {
         .unwrap();
     assert_eq!(read_status[0], VIRTIO_BLK_S_OK, "read should succeed");
 }
+
+/// A descriptor whose `next` link points back at itself forms a chain that
+/// never terminates, and the queue rejects it without consuming it — the
+/// available index does not move, so the same chain is still there on the next
+/// poll. A worker that logs the error and keeps polling therefore spins inside
+/// a single `poll` call, never returning `Pending`. That burns a CPU forever
+/// and, because the cancel future is only polled once the work future returns
+/// `Pending`, leaves the worker task impossible to stop — wedging device
+/// teardown, reset, save/restore, and inspect.
+///
+/// The device runs on its own executor thread so that a spinning worker wedges
+/// only that thread, letting this test fail on a timeout instead of hanging.
+#[async_test]
+async fn cyclic_descriptor_chain_does_not_wedge_worker(driver: DefaultDriver) {
+    let (_device_thread, device_driver) = DefaultPool::spawn_on_thread("virtio-blk-device");
+    let disk = ram_disk(64 * 1024, false);
+    let mut harness = TestHarness::with_device_driver(&driver, &device_driver, disk, false);
+    harness.enable().await;
+
+    // Run one valid request first, so the worker is known to be up and parked
+    // waiting for a kick by the time the bad chain arrives.
+    harness.post_flush_request(0);
+    let (used_id, _) = harness.wait_for_used().await;
+    assert_eq!(used_id, 0);
+
+    // Descriptor 4 chains to itself.
+    let gpa = harness.alloc_data(REQ_HEADER_SIZE);
+    write_descriptor(
+        &harness.mem,
+        DESC_ADDR,
+        4,
+        gpa,
+        REQ_HEADER_SIZE,
+        DescriptorFlags::new().with_next(true),
+        4,
+    );
+    make_available(
+        &harness.mem,
+        AVAIL_ADDR,
+        QUEUE_SIZE,
+        4,
+        &mut harness.avail_idx,
+    );
+    harness.queue_event.signal();
+
+    // Give the worker time to wake on the kick and reject the chain.
+    PolledTimer::new(&driver)
+        .sleep(Duration::from_millis(250))
+        .await;
+
+    // A worker spinning on the bad chain never observes the stop request.
+    with_timeout(
+        &driver,
+        Duration::from_secs(5),
+        "virtio-blk worker could not be stopped after an invalid descriptor chain",
+        harness.device.stop_queue(0),
+    )
+    .await;
+}

vm/devices/virtio/virtio_blk/src/lib.rs

@@ -169,7 +169,13 @@ impl AsyncRun<BlkQueueState> for BlkWorker {
         state: &mut BlkQueueState,
     ) -> Result<(), task_control::Cancelled> {
         stop.until_stopped(async {
-            loop {
+            // Set once the queue can no longer produce work, because the guest
+            // violated the queue protocol. Fetching stops, but in-flight IOs
+            // keep draining so their descriptors are still completed; the
+            // worker exits once they are done.
+            let mut queue_done = false;
+
+            while !(queue_done && self.ios.is_empty()) {
                 enum Event {
                     NewWork(Result<VirtioQueueCallbackWork, std::io::Error>),
                     Completed(IoCompletion),
@@ -181,7 +187,7 @@ impl AsyncRun<BlkQueueState> for BlkWorker {
                         return Poll::Ready(Event::Completed(completion));
                     }
                     // Accept new work if under the depth limit.
-                    if self.ios.len() < MAX_IO_DEPTH {
+                    if !queue_done && self.ios.len() < MAX_IO_DEPTH {
                         if let Poll::Ready(item) = state.queue.poll_next_unpin(cx) {
                             let item = item.expect("virtio queue stream never ends");
                             return Poll::Ready(Event::NewWork(item));
@@ -201,10 +207,13 @@ impl AsyncRun<BlkQueueState> for BlkWorker {
                         }));
                     }
                     Event::NewWork(Err(err)) => {
+                        // The queue is retired: the rejected chain is never
+                        // consumed, so retrying would fail identically forever.
                         tracelimit::error_ratelimited!(
                             error = &err as &dyn std::error::Error,
-                            "error reading from virtio queue"
+                            "error reading from virtio queue, stopping worker"
                         );
+                        queue_done = true;
                     }
                     Event::Completed(completion) => {
                         self.finish_io(&mut state.queue, completion);