diff --git a/alioth/src/pci/cap.rs b/alioth/src/pci/cap.rs
index 833d7573..5365cfab 100644
--- a/alioth/src/pci/cap.rs
+++ b/alioth/src/pci/cap.rs
@@ -575,6 +575,25 @@ where
             *entry = MsixTableMmioEntry::default();
         }
     }
+
+    /// Replaces irqfds with plain entries that keep the address, data and
+    /// mask programmed by the guest.
+    pub fn release_irqfds(&self) {
+        let mut entries = self.entries.write();
+        for entry in entries.iter_mut() {
+            let MsixTableMmioEntry::IrqFd(fd) = entry else {
+                continue;
+            };
+            let mut control = MsixVectorCtrl(0);
+            control.set_masked(fd.get_masked());
+            *entry = MsixTableMmioEntry::Entry(MsixTableEntry {
+                addr_lo: fd.get_addr_lo(),
+                addr_hi: fd.get_addr_hi(),
+                data: fd.get_data(),
+                control,
+            });
+        }
+    }
 }
 
 impl<F> Mmio for MsixTableMmio<F>
diff --git a/alioth/src/pci/cap_test.rs b/alioth/src/pci/cap_test.rs
index 7fc8adc3..6df28c7f 100644
--- a/alioth/src/pci/cap_test.rs
+++ b/alioth/src/pci/cap_test.rs
@@ -311,6 +311,47 @@ fn test_msix_table_mmio() {
     );
 }
 
+#[test]
+fn test_msix_table_release_irqfds() {
+    let irqfd = TestIrqFd::default();
+    irqfd.set_addr_lo(0xfee0_0000).unwrap();
+    irqfd.set_addr_hi(0x1).unwrap();
+    irqfd.set_data(0x41).unwrap();
+    irqfd.set_masked(false).unwrap();
+    let entries = Box::new([
+        MsixTableMmioEntry::Entry(MsixTableEntry {
+            addr_lo: 0xfee0_1000,
+            addr_hi: 0,
+            data: 0x42,
+            control: MsixVectorCtrl(1),
+        }),
+        MsixTableMmioEntry::IrqFd(irqfd),
+    ]);
+    let table = MsixTableMmio {
+        entries: RwLock::new(entries),
+    };
+
+    // Irqfds are gone, but the guest programming stays.
+    table.release_irqfds();
+    assert_matches!(
+        **table.entries.read(),
+        [
+            MsixTableMmioEntry::Entry(MsixTableEntry {
+                addr_lo: 0xfee0_1000,
+                addr_hi: 0,
+                data: 0x42,
+                control: MsixVectorCtrl(1),
+            }),
+            MsixTableMmioEntry::Entry(MsixTableEntry {
+                addr_lo: 0xfee0_0000,
+                addr_hi: 0x1,
+                data: 0x41,
+                control: MsixVectorCtrl(0),
+            })
+        ]
+    );
+}
+
 #[test]
 fn test_pci_cap_list() {
     let caps: Vec<Box<dyn PciCap>> = vec![
diff --git a/alioth/src/virtio/dev/dev.rs b/alioth/src/virtio/dev/dev.rs
index 98178333..df8e0d9b 100644
--- a/alioth/src/virtio/dev/dev.rs
+++ b/alioth/src/virtio/dev/dev.rs
@@ -26,9 +26,11 @@ use std::fmt::Debug;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU8, AtomicU16, AtomicU32, Ordering};
 use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
 
 use bitflags::Flags;
-use flume::{Receiver, Sender};
+use flume::{Receiver, RecvTimeoutError, Sender, TryRecvError};
+use parking_lot::Mutex;
 use snafu::ResultExt;
 
 use crate::mem::emulated::Mmio;
@@ -93,6 +95,13 @@ impl Register {
 
 const TOKEN_WARKER: u64 = 1 << 63;
 
+/// How long a transport waits for the worker to reset a device when it
+/// cannot leave the waiting to the driver.
+///
+/// It is longer than the time the io_uring worker waits for in-flight
+/// requests when it leaves its event loop.
+pub(crate) const RESET_TIMEOUT: Duration = Duration::from_secs(10);
+
 #[derive(Debug, Clone)]
 pub struct StartParam<S>
 where
@@ -114,9 +123,15 @@ pub struct WorkerParam<S>
 where
     S: IrqSender,
 {
-    /// Receiving end of the wake event channel. The sending end is
-    /// [`VirtioDevice::event_tx`].
+    /// Receiving end of the wake event channel. The sending end is in
+    /// [`WorkerCtl`].
     pub(crate) event_rx: Receiver<WakeEvent<S>>,
+    /// Sending end of the report channel. The receiving end is in
+    /// [`WorkerCtl`].
+    ///
+    /// Only the worker holds it, so the channel disconnects once the worker
+    /// exits, which completes all pending resets.
+    pub(crate) report_tx: Sender<WorkerReport>,
     /// Guest memory.
     pub(crate) memory: Arc<RamBus>,
     /// Queue registers, shared with the transport.
@@ -139,7 +154,21 @@ where
     Start {
         param: StartParam<S>,
     },
-    Reset,
+    /// Resets the device. Once the worker no longer uses the queues, it
+    /// reports [`WorkerReport::ResetDone`] with `seq`.
+    Reset {
+        seq: u64,
+    },
+}
+
+/// What a worker reports to the transport.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum WorkerReport {
+    /// The worker has completed reset `seq`.
+    ResetDone { seq: u64 },
+    /// The worker has failed, and does not serve the queues until the
+    /// device is reset.
+    Failed,
 }
 
 #[derive(Debug, PartialEq, Eq)]
@@ -158,6 +187,161 @@ where
     backend: B,
 }
 
+/// The handle a transport uses to drive the worker thread of a virtio
+/// device.
+#[derive(Debug)]
+pub struct WorkerCtl<S>
+where
+    S: IrqSender,
+{
+    name: Arc<str>,
+    event_tx: Sender<WakeEvent<S>>,
+    notifier: Arc<Notifier>,
+    /// Sequence number of the last requested reset.
+    ///
+    /// The lock is held while sending the reset event, so that the worker
+    /// sees resets in the order of their sequence numbers.
+    reset_seq: Mutex<u64>,
+    reports: Mutex<Reports>,
+}
+
+/// Returns `true` if reset `seq` is not later than reset `done`.
+///
+/// Sequence numbers are compared in serial number arithmetic, which stays
+/// correct after the counter wraps around, as long as the two are less than
+/// 2^63 resets apart.
+fn reset_completed(done: u64, seq: u64) -> bool {
+    (done.wrapping_sub(seq) as i64) >= 0
+}
+
+/// What the worker has reported so far.
+#[derive(Debug)]
+struct Reports {
+    rx: Receiver<WorkerReport>,
+    /// Sequence number of the last completed reset.
+    reset_done: u64,
+    /// Whether the worker has failed since the last completed reset.
+    failed: bool,
+}
+
+impl Reports {
+    fn handle(&mut self, report: WorkerReport) {
+        match report {
+            WorkerReport::ResetDone { seq } => {
+                if !reset_completed(self.reset_done, seq) {
+                    self.reset_done = seq;
+                }
+                // A reset brings the device back into service.
+                self.failed = false;
+            }
+            WorkerReport::Failed => self.failed = true,
+        }
+    }
+
+    /// Handles all pending reports.
+    ///
+    /// Returns `false` if the worker has exited.
+    fn drain(&mut self) -> bool {
+        loop {
+            match self.rx.try_recv() {
+                Ok(report) => self.handle(report),
+                Err(TryRecvError::Empty) => break true,
+                Err(TryRecvError::Disconnected) => break false,
+            }
+        }
+    }
+}
+
+impl<S> WorkerCtl<S>
+where
+    S: IrqSender,
+{
+    pub(crate) fn new(
+        name: Arc<str>,
+        event_tx: Sender<WakeEvent<S>>,
+        notifier: Arc<Notifier>,
+        report_rx: Receiver<WorkerReport>,
+    ) -> Self {
+        WorkerCtl {
+            name,
+            event_tx,
+            notifier,
+            reset_seq: Mutex::new(0),
+            reports: Mutex::new(Reports {
+                rx: report_rx,
+                reset_done: 0,
+                failed: false,
+            }),
+        }
+    }
+
+    /// Sends `event` to the worker and wakes it up.
+    ///
+    /// Returns `false` if the worker has exited.
+    pub fn wake(&self, event: WakeEvent<S>) -> bool {
+        // A worker accepts `Start` only while it is pending, during which
+        // it waits on the channel instead of the notifier.
+        let is_start = matches!(event, WakeEvent::Start { .. });
+        if let Err(e) = self.event_tx.send(event) {
+            log::error!("{}: failed to send event: {e}", self.name);
+            return false;
+        }
+        if !is_start && let Err(e) = self.notifier.notify() {
+            log::error!("{}: failed to wake up device: {e}", self.name);
+        }
+        true
+    }
+
+    /// Asks the worker to reset the device.
+    ///
+    /// Returns the sequence number of this reset, which
+    /// [`poll_reset`](Self::poll_reset) and [`wait_reset`](Self::wait_reset)
+    /// take.
+    pub fn request_reset(&self) -> u64 {
+        let mut reset_seq = self.reset_seq.lock();
+        let seq = reset_seq.wrapping_add(1);
+        *reset_seq = seq;
+        self.wake(WakeEvent::Reset { seq });
+        seq
+    }
+
+    /// Returns `true` if the worker has completed reset `seq`.
+    ///
+    /// The worker handles resets in order, so this also means that all
+    /// earlier resets have completed.
+    pub fn poll_reset(&self, seq: u64) -> bool {
+        let mut reports = self.reports.lock();
+        let alive = reports.drain();
+        // An exited worker no longer uses the queues.
+        !alive || reset_completed(reports.reset_done, seq)
+    }
+
+    /// Waits until the worker has completed reset `seq`.
+    pub fn wait_reset(&self, seq: u64, timeout: Duration) -> Result<()> {
+        let deadline = Instant::now() + timeout;
+        let mut reports = self.reports.lock();
+        while !reset_completed(reports.reset_done, seq) {
+            match reports.rx.recv_deadline(deadline) {
+                Ok(report) => reports.handle(report),
+                Err(RecvTimeoutError::Disconnected) => break,
+                Err(RecvTimeoutError::Timeout) => {
+                    let name = self.name.clone();
+                    return error::ResetTimeout { name, timeout }.fail();
+                }
+            }
+        }
+        Ok(())
+    }
+
+    /// Returns `true` if the worker has failed, so that the device needs a
+    /// reset.
+    pub fn failed(&self) -> bool {
+        let mut reports = self.reports.lock();
+        reports.drain();
+        reports.failed
+    }
+}
+
 #[derive(Debug)]
 pub struct VirtioDevice<S>
 where
@@ -169,8 +353,7 @@ where
     pub device_feature: u128,
     pub queue_regs: Arc<[QueueReg]>,
     pub shared_mem_regions: Option<Arc<MemRegion>>,
-    pub notifier: Arc<Notifier>,
-    pub event_tx: Sender<WakeEvent<S>>,
+    pub ctl: Arc<WorkerCtl<S>>,
     pub(crate) worker_handle: Option<JoinHandle<()>>,
 }
 
@@ -178,16 +361,14 @@ impl<S> VirtioDevice<S>
 where
     S: IrqSender,
 {
-    fn shutdown(&mut self) -> Result<(), Box<dyn std::error::Error>> {
+    fn shutdown(&mut self) {
         let Some(handle) = self.worker_handle.take() else {
-            return Ok(());
+            return;
         };
-        self.event_tx.send(WakeEvent::Shutdown)?;
-        self.notifier.notify()?;
+        self.ctl.wake(WakeEvent::Shutdown);
         if let Err(e) = handle.join() {
             log::error!("{}: failed to join worker thread: {e:?}", self.name)
         }
-        Ok(())
     }
 
     pub fn new<D>(
@@ -217,8 +398,10 @@ where
 
         let shared_mem_regions = dev.shared_mem_regions();
         let (event_tx, event_rx) = flume::unbounded();
+        let (report_tx, report_rx) = flume::unbounded();
         let param = WorkerParam {
             event_rx,
+            report_tx,
             memory,
             queue_regs: queue_regs.clone(),
         };
@@ -228,14 +411,14 @@ where
             VirtioFeature::from_bits_retain(device_feature & !D::Feature::all().bits()),
             D::Feature::from_bits_truncate(device_feature)
         );
+        let ctl = WorkerCtl::new(name.clone(), event_tx, notifier, report_rx);
         let virtio_dev = VirtioDevice {
             name,
             id,
             device_feature,
             queue_regs,
             worker_handle: Some(handle),
-            event_tx,
-            notifier,
+            ctl: Arc::new(ctl),
             device_config,
             shared_mem_regions,
         };
@@ -248,9 +431,7 @@ where
     S: IrqSender,
 {
     fn drop(&mut self) {
-        if let Err(e) = self.shutdown() {
-            log::error!("{}: failed to shutdown: {e}", self.name);
-        }
+        self.shutdown()
     }
 }
 
@@ -287,6 +468,9 @@ where
     pub dev: D,
     param: WorkerParam<S>,
     pub state: WorkerState,
+    /// The reset being handled, acknowledged once the worker no longer uses
+    /// the queues.
+    reset_seq: Option<u64>,
 }
 
 impl<D, S> Context<D, S>
@@ -294,6 +478,16 @@ where
     D: Virtio,
     S: IrqSender,
 {
+    fn report(&self, report: WorkerReport) {
+        // The transport might be gone, e.g. at VMM shutdown.
+        let _ = self.param.report_tx.send(report);
+    }
+
+    fn ack_reset(&self, seq: u64) {
+        log::debug!("{}: reset {seq} completed", self.dev.name());
+        self.report(WorkerReport::ResetDone { seq });
+    }
+
     fn handle_wake_events<B>(&mut self, backend: &mut B) -> Result<()>
     where
         B: ActiveBackend<D>,
@@ -310,9 +504,10 @@ where
                 }
                 #[cfg(target_os = "linux")]
                 WakeEvent::VuChannel { channel } => self.dev.set_vu_channel(channel),
-                WakeEvent::Reset => {
-                    log::info!("{}: guest requested reset", self.dev.name());
+                WakeEvent::Reset { seq } => {
+                    log::info!("{}: reset requested", self.dev.name());
                     self.state = WorkerState::Pending;
+                    self.reset_seq = Some(seq);
                     break;
                 }
             }
@@ -323,7 +518,8 @@ where
     fn wait_start(&mut self) -> Option<StartParam<S>> {
         for wake_event in self.param.event_rx.iter() {
             match wake_event {
-                WakeEvent::Reset => {}
+                // The device has not started, so there is nothing to reset.
+                WakeEvent::Reset { seq } => self.ack_reset(seq),
                 WakeEvent::Start { param } => {
                     self.state = WorkerState::Running;
                     return Some(param);
@@ -372,6 +568,7 @@ where
                 dev,
                 param,
                 state: WorkerState::Pending,
+                reset_seq: None,
             },
             backend,
         };
@@ -398,14 +595,16 @@ where
             VirtioFeature::from_bits_retain(param.feature & !D::Feature::all().bits()),
             D::Feature::from_bits_truncate(param.feature)
         );
-        self.backend
-            .event_loop(ram, &mut self.context, queues, param)
+        let ret = self
+            .backend
+            .event_loop(ram, &mut self.context, queues, param);
+        // Clean up even if the event loop failed, since the device might
+        // have been partially activated.
+        let reset = self.backend.reset(&mut self.context.dev);
+        ret.and(reset)
     }
 
-    fn loop_until_reset(&mut self) -> Result<()> {
-        let Some(param) = self.context.wait_start() else {
-            return Ok(());
-        };
+    fn run(&mut self, param: &StartParam<S>) -> Result<()> {
         let memory = self.context.param.memory.clone();
         let ram = memory.lock_layout();
         let feature = param.feature & !VirtioFeature::ACCESS_PLATFORM.bits();
@@ -420,7 +619,7 @@ where
                 Ok(Some(Queue::new(split_queue, reg, &ram)))
             };
             let queues: Result<Box<_>> = queue_regs.iter().map(new_queue).collect();
-            self.event_loop(&mut (queues?), &ram, &param)?;
+            self.event_loop(&mut (queues?), &ram, param)
         } else {
             let new_queue = |reg| {
                 let Some(split_queue) = SplitQueue::new(reg, &ram, event_idx)? else {
@@ -429,17 +628,37 @@ where
                 Ok(Some(Queue::new(split_queue, reg, &ram)))
             };
             let queues: Result<Box<_>> = queue_regs.iter().map(new_queue).collect();
-            self.event_loop(&mut (queues?), &ram, &param)?;
+            self.event_loop(&mut (queues?), &ram, param)
+        }
+    }
+
+    fn loop_until_reset(&mut self) -> Result<()> {
+        let Some(param) = self.context.wait_start() else {
+            return Ok(());
         };
-        self.backend.reset(&mut self.context.dev)?;
-        Ok(())
+        let ret = self.run(&param);
+        // Unless the worker failed in handling a reset or a shutdown, the
+        // device is out of service until the driver resets it. The worker
+        // stays alive for that reset.
+        if ret.is_err() && self.context.state == WorkerState::Running {
+            self.context.state = WorkerState::Pending;
+            self.context.report(WorkerReport::Failed);
+            // The transport sets DEVICE_NEEDS_RESET, after which a running
+            // device must notify the driver of a configuration change
+            // (VirtIO spec 1.4, Sec 2.1.2).
+            param.irq_sender.config_irq();
+        }
+        ret
     }
 
     fn do_work(mut self) {
         while self.context.state != WorkerState::Shutdown {
             if let Err(e) = self.loop_until_reset() {
-                log::error!("worker {}: {e:?}", self.context.dev.name(),);
-                return;
+                log::error!("worker {}: {e:?}", self.context.dev.name());
+            }
+            // The memory layout lock is released by now.
+            if let Some(seq) = self.context.reset_seq.take() {
+                self.context.ack_reset(seq);
             }
         }
         log::debug!("worker {}: done", self.context.dev.name())
diff --git a/alioth/src/virtio/dev/dev_test.rs b/alioth/src/virtio/dev/dev_test.rs
index 677f5082..fa2a3fff 100644
--- a/alioth/src/virtio/dev/dev_test.rs
+++ b/alioth/src/virtio/dev/dev_test.rs
@@ -12,9 +12,27 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::atomic::Ordering;
+use std::os::fd::BorrowedFd;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
+use std::thread::{self, JoinHandle};
+use std::time::{Duration, Instant};
 
-use super::Register;
+use assert_matches::assert_matches;
+use mio::Registry;
+use mio::event::Event;
+
+use crate::sync::notifier::Notifier;
+use crate::virtio::dev::entropy::{EntropyConfig, EntropyFeature};
+use crate::virtio::dev::{
+    Register, StartParam, Virtio, WakeEvent, WorkerCtl, WorkerParam, WorkerReport,
+};
+use crate::virtio::queue::VirtQueue;
+use crate::virtio::tests::{fixture_queues, fixture_ram_bus};
+use crate::virtio::worker::mio::{ActiveMio, Mio, VirtioMio};
+use crate::virtio::{DeviceId, Error, IrqSender, Result, VirtioFeature, error};
+
+const TIMEOUT: Duration = Duration::from_secs(5);
 
 #[test]
 fn test_virtio_register_get_driver_feature() {
@@ -32,3 +50,367 @@ fn test_virtio_register_get_driver_feature() {
         | ((0x0123_4567u128) << 96);
     assert_eq!(reg.get_driver_feature(), expected);
 }
+
+/// A device that only records what the worker does to it.
+///
+/// It borrows the config and feature types of the entropy device, which
+/// has neither.
+#[derive(Debug, Default)]
+struct FakeDev {
+    /// Makes activation fail, like a backend rejecting the driver setup.
+    fail_activate: Arc<AtomicBool>,
+    /// Counts the calls to [`VirtioMio::reset`].
+    resets: Arc<AtomicU32>,
+}
+
+impl Virtio for FakeDev {
+    type Config = EntropyConfig;
+    type Feature = EntropyFeature;
+
+    fn name(&self) -> &str {
+        "fake"
+    }
+
+    fn id(&self) -> DeviceId {
+        DeviceId::ENTROPY
+    }
+
+    fn num_queues(&self) -> u16 {
+        1
+    }
+
+    fn config(&self) -> Arc<EntropyConfig> {
+        Arc::new(EntropyConfig)
+    }
+
+    fn feature(&self) -> u128 {
+        0
+    }
+
+    fn spawn_worker<S>(self, param: WorkerParam<S>) -> Result<(JoinHandle<()>, Arc<Notifier>)>
+    where
+        S: IrqSender,
+    {
+        Mio::spawn_worker(self, param)
+    }
+}
+
+impl VirtioMio for FakeDev {
+    fn activate<'m, Q, S>(
+        &mut self,
+        _feature: u128,
+        _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>,
+    ) -> Result<()>
+    where
+        Q: VirtQueue<'m>,
+        S: IrqSender,
+    {
+        if self.fail_activate.load(Ordering::Acquire) {
+            return error::InvalidBuffer.fail();
+        }
+        Ok(())
+    }
+
+    fn handle_queue<'m, Q, S>(
+        &mut self,
+        _index: u16,
+        _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>,
+    ) -> Result<()>
+    where
+        Q: VirtQueue<'m>,
+        S: IrqSender,
+    {
+        Ok(())
+    }
+
+    fn handle_event<'m, Q, S>(
+        &mut self,
+        _event: &Event,
+        _active_mio: &mut ActiveMio<'_, '_, 'm, Q, S>,
+    ) -> Result<()>
+    where
+        Q: VirtQueue<'m>,
+        S: IrqSender,
+    {
+        Ok(())
+    }
+
+    fn reset(&mut self, _registry: &Registry) {
+        self.resets.fetch_add(1, Ordering::AcqRel);
+    }
+}
+
+/// Counts configuration interrupts.
+#[derive(Debug, Default)]
+struct TestIrqSender {
+    config_irqs: AtomicU32,
+}
+
+impl IrqSender for TestIrqSender {
+    fn queue_irq(&self, _idx: u16) {}
+
+    fn config_irq(&self) {
+        self.config_irqs.fetch_add(1, Ordering::AcqRel);
+    }
+
+    fn queue_irqfd<F, T>(&self, _idx: u16, _f: F) -> Result<T>
+    where
+        F: FnOnce(BorrowedFd) -> Result<T>,
+    {
+        unimplemented!()
+    }
+
+    fn config_irqfd<F, T>(&self, _f: F) -> Result<T>
+    where
+        F: FnOnce(BorrowedFd) -> Result<T>,
+    {
+        unimplemented!()
+    }
+}
+
+struct TestWorker {
+    ctl: WorkerCtl<TestIrqSender>,
+    handle: JoinHandle<()>,
+    irq_sender: Arc<TestIrqSender>,
+    fail_activate: Arc<AtomicBool>,
+    resets: Arc<AtomicU32>,
+}
+
+impl TestWorker {
+    fn spawn() -> Self {
+        let dev = FakeDev::default();
+        let fail_activate = dev.fail_activate.clone();
+        let resets = dev.resets.clone();
+        let (event_tx, event_rx) = flume::unbounded();
+        let (report_tx, report_rx) = flume::unbounded();
+        let param = WorkerParam {
+            event_rx,
+            report_tx,
+            memory: Arc::new(fixture_ram_bus()),
+            queue_regs: Arc::from(fixture_queues(1)),
+        };
+        let (handle, notifier) = dev.spawn_worker(param).unwrap();
+        TestWorker {
+            ctl: WorkerCtl::new("fake".into(), event_tx, notifier, report_rx),
+            handle,
+            irq_sender: Arc::default(),
+            fail_activate,
+            resets,
+        }
+    }
+
+    fn start(&self) {
+        let param = StartParam {
+            feature: VirtioFeature::VERSION_1.bits(),
+            irq_sender: self.irq_sender.clone(),
+            notifiers: None,
+        };
+        assert!(self.ctl.wake(WakeEvent::Start { param }));
+    }
+
+    fn reset(&self) {
+        let seq = self.ctl.request_reset();
+        assert_matches!(self.ctl.wait_reset(seq, TIMEOUT), Ok(()));
+        assert!(self.ctl.poll_reset(seq));
+    }
+
+    fn resets(&self) -> u32 {
+        self.resets.load(Ordering::Acquire)
+    }
+
+    fn config_irqs(&self) -> u32 {
+        self.irq_sender.config_irqs.load(Ordering::Acquire)
+    }
+
+    /// Waits until the worker reports a failure.
+    fn wait_failed(&self) {
+        let deadline = Instant::now() + TIMEOUT;
+        while !self.ctl.failed() {
+            assert!(Instant::now() < deadline, "no failure reported");
+            thread::sleep(Duration::from_millis(1));
+        }
+    }
+
+    fn shutdown(self) {
+        assert!(self.ctl.wake(WakeEvent::Shutdown));
+        self.handle.join().unwrap();
+    }
+}
+
+#[test]
+fn test_worker_reset_before_start() {
+    let worker = TestWorker::spawn();
+    worker.reset();
+    // The device has not started, so there was nothing to clean up.
+    assert_eq!(worker.resets(), 0);
+    worker.shutdown();
+}
+
+#[test]
+fn test_worker_reset_after_start() {
+    let worker = TestWorker::spawn();
+    worker.start();
+    worker.reset();
+    // The worker acknowledges a reset only after cleaning up.
+    assert_eq!(worker.resets(), 1);
+
+    worker.start();
+    worker.reset();
+    assert_eq!(worker.resets(), 2);
+    assert!(!worker.ctl.failed());
+    assert_eq!(worker.config_irqs(), 0);
+    worker.shutdown();
+}
+
+#[test]
+fn test_worker_survives_activation_failure() {
+    let worker = TestWorker::spawn();
+    worker.fail_activate.store(true, Ordering::Release);
+    worker.start();
+    // The worker cleans up the partially activated device before reporting
+    // the failure.
+    worker.wait_failed();
+    assert_eq!(worker.resets(), 1);
+
+    // The worker stays alive to complete the reset, which brings the device
+    // back into service.
+    worker.reset();
+    assert!(!worker.ctl.failed());
+    // The worker has notified the driver of the failure.
+    assert_eq!(worker.config_irqs(), 1);
+
+    worker.fail_activate.store(false, Ordering::Release);
+    worker.start();
+    worker.reset();
+    assert_eq!(worker.resets(), 2);
+    assert!(!worker.ctl.failed());
+    worker.shutdown();
+}
+
+#[test]
+fn test_worker_exit_completes_reset() {
+    let worker = TestWorker::spawn();
+    worker.start();
+    let TestWorker { ctl, handle, .. } = worker;
+    assert!(ctl.wake(WakeEvent::Shutdown));
+    handle.join().unwrap();
+
+    // An exited worker no longer uses the queues.
+    let seq = ctl.request_reset();
+    assert!(ctl.poll_reset(seq));
+    assert_matches!(ctl.wait_reset(seq, TIMEOUT), Ok(()));
+}
+
+#[test]
+fn test_wait_reset_timeout() {
+    let (event_tx, _event_rx) = flume::unbounded();
+    let (report_tx, report_rx) = flume::unbounded();
+    let notifier = Arc::new(Notifier::new().unwrap());
+    let ctl = WorkerCtl::<TestIrqSender>::new("fake".into(), event_tx, notifier, report_rx);
+
+    let seq1 = ctl.request_reset();
+    let seq2 = ctl.request_reset();
+    assert!(!ctl.poll_reset(seq1));
+    assert_matches!(
+        ctl.wait_reset(seq1, Duration::from_millis(10)),
+        Err(Error::ResetTimeout { timeout, .. }) if timeout == Duration::from_millis(10)
+    );
+
+    // Completing a reset also completes the earlier ones.
+    report_tx
+        .send(WorkerReport::ResetDone { seq: seq2 })
+        .unwrap();
+    assert!(ctl.poll_reset(seq1));
+    assert_matches!(ctl.wait_reset(seq2, Duration::ZERO), Ok(()));
+}
+
+#[test]
+fn test_reset_seq_wraps_around() {
+    let (event_tx, event_rx) = flume::unbounded();
+    let (report_tx, report_rx) = flume::unbounded();
+    let notifier = Arc::new(Notifier::new().unwrap());
+    let ctl = WorkerCtl::<TestIrqSender>::new("fake".into(), event_tx, notifier, report_rx);
+    // Pretend that almost all sequence numbers have been used up.
+    *ctl.reset_seq.lock() = u64::MAX - 1;
+    ctl.reports.lock().reset_done = u64::MAX - 1;
+
+    let seq1 = ctl.request_reset();
+    let seq2 = ctl.request_reset();
+    assert_eq!((seq1, seq2), (u64::MAX, 0));
+    for seq in [seq1, seq2] {
+        assert_matches!(event_rx.try_recv(), Ok(WakeEvent::Reset { seq: s }) if s == seq);
+    }
+    assert!(!ctl.poll_reset(seq1));
+
+    report_tx
+        .send(WorkerReport::ResetDone { seq: seq1 })
+        .unwrap();
+    assert!(ctl.poll_reset(seq1));
+    assert!(!ctl.poll_reset(seq2));
+
+    report_tx
+        .send(WorkerReport::ResetDone { seq: seq2 })
+        .unwrap();
+    assert!(ctl.poll_reset(seq1));
+    assert_matches!(ctl.wait_reset(seq2, Duration::ZERO), Ok(()));
+}
+
+#[test]
+fn test_request_reset_in_order() {
+    let (event_tx, event_rx) = flume::unbounded();
+    let notifier = Arc::new(Notifier::new().unwrap());
+    let report_rx = flume::unbounded().1;
+    let ctl = WorkerCtl::<TestIrqSender>::new("fake".into(), event_tx, notifier, report_rx);
+    let ctl = Arc::new(ctl);
+
+    let threads: Vec<_> = (0..4)
+        .map(|_| {
+            let ctl = ctl.clone();
+            thread::spawn(move || {
+                for _ in 0..100 {
+                    ctl.request_reset();
+                }
+            })
+        })
+        .collect();
+    for t in threads {
+        t.join().unwrap();
+    }
+
+    // The worker sees the resets in the order of their sequence numbers,
+    // even when they are requested concurrently.
+    let seqs: Vec<u64> = event_rx
+        .try_iter()
+        .map(|event| match event {
+            WakeEvent::Reset { seq } => seq,
+            event => panic!("unexpected event: {event:?}"),
+        })
+        .collect();
+    assert_eq!(seqs, (1..=400).collect::<Vec<_>>());
+}
+
+#[test]
+fn test_worker_failure_until_reset() {
+    let (event_tx, _event_rx) = flume::unbounded();
+    let (report_tx, report_rx) = flume::unbounded();
+    let notifier = Arc::new(Notifier::new().unwrap());
+    let ctl = WorkerCtl::<TestIrqSender>::new("fake".into(), event_tx, notifier, report_rx);
+    assert!(!ctl.failed());
+
+    // Reports arrive in order. A failure before a reset is gone with the
+    // reset, while a failure after it is not.
+    let seq = ctl.request_reset();
+    for report in [
+        WorkerReport::Failed,
+        WorkerReport::ResetDone { seq },
+        WorkerReport::Failed,
+    ] {
+        report_tx.send(report).unwrap();
+    }
+    assert!(ctl.failed());
+    assert!(ctl.poll_reset(seq));
+
+    let seq = ctl.request_reset();
+    report_tx.send(WorkerReport::ResetDone { seq }).unwrap();
+    assert!(!ctl.failed());
+}
diff --git a/alioth/src/virtio/dev/entropy_test.rs b/alioth/src/virtio/dev/entropy_test.rs
index ba613baa..5f809c8d 100644
--- a/alioth/src/virtio/dev/entropy_test.rs
+++ b/alioth/src/virtio/dev/entropy_test.rs
@@ -79,6 +79,7 @@ fn entropy_test() {
     let (handle, notifier) = dev
         .spawn_worker(WorkerParam {
             event_rx: rx,
+            report_tx: flume::unbounded().0,
             memory: ram_bus.clone(),
             queue_regs: regs,
         })
diff --git a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs
index 70874ada..d6a904f9 100644
--- a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs
+++ b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs
@@ -131,6 +131,7 @@ impl<'m> VsockTest<'m> {
         let (handle, notifier) = dev
             .spawn_worker(WorkerParam {
                 event_rx: rx,
+                report_tx: flume::unbounded().0,
                 memory: ram_bus.clone(),
                 queue_regs: regs,
             })
diff --git a/alioth/src/virtio/dev/vsock/vhost_vsock.rs b/alioth/src/virtio/dev/vsock/vhost_vsock.rs
index 9a18e7b3..49c60397 100644
--- a/alioth/src/virtio/dev/vsock/vhost_vsock.rs
+++ b/alioth/src/virtio/dev/vsock/vhost_vsock.rs
@@ -209,20 +209,24 @@ impl VirtioMio for VhostVsock {
     }
 
     fn reset(&mut self, registry: &Registry) {
-        self.vhost_dev.vsock_set_running(false).unwrap();
+        let name = &*self.name;
+        if let Err(e) = self.vhost_dev.vsock_set_running(false) {
+            log::error!("{name}: failed to stop vhost-vsock: {e:?}");
+        }
         for (index, error_fd) in self.error_fds.iter_mut().enumerate() {
             let Some(err_fd) = error_fd else {
                 continue;
             };
-            self.vhost_dev
-                .set_virtq_err(&VirtqFile {
-                    index: index as _,
-                    fd: VHOST_FILE_UNBIND,
-                })
-                .unwrap();
-            registry
-                .deregister(&mut SourceFd(&err_fd.as_raw_fd()))
-                .unwrap();
+            let unbind = VirtqFile {
+                index: index as _,
+                fd: VHOST_FILE_UNBIND,
+            };
+            if let Err(e) = self.vhost_dev.set_virtq_err(&unbind) {
+                log::error!("{name}: queue-{index}: failed to unbind error fd: {e:?}");
+            }
+            if let Err(e) = registry.deregister(&mut SourceFd(&err_fd.as_raw_fd())) {
+                log::error!("{name}: queue-{index}: failed to deregister error fd: {e:?}");
+            }
             *error_fd = None;
         }
     }
diff --git a/alioth/src/virtio/pci.rs b/alioth/src/virtio/pci.rs
index af8e2bcd..bb9d3228 100644
--- a/alioth/src/virtio/pci.rs
+++ b/alioth/src/virtio/pci.rs
@@ -18,13 +18,14 @@ use std::mem::size_of;
 use std::os::fd::{AsFd, AsRawFd, BorrowedFd};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU16, Ordering};
+use std::time::Duration;
 
 use alioth_macros::Layout;
-use flume::Sender;
 use parking_lot::{Mutex, RwLock};
 use zerocopy::{FromZeros, Immutable, IntoBytes};
 
 use crate::device::Pause;
+use crate::errors::BoxTrace;
 use crate::hv::{IrqFd, MsiSender, NotifierRegistry};
 use crate::mem::emulated::{Action, Mmio};
 use crate::mem::{MemRange, MemRegion, MemRegionCallback, MemRegionEntry};
@@ -39,7 +40,7 @@ use crate::pci::config::{
 use crate::pci::{self, Pci, PciBar};
 use crate::sync::notifier::Notifier;
 use crate::utils::{get_atomic_high32, get_atomic_low32, set_atomic_high32, set_atomic_low32};
-use crate::virtio::dev::{Register, StartParam, VirtioDevice, WakeEvent};
+use crate::virtio::dev::{RESET_TIMEOUT, Register, StartParam, VirtioDevice, WakeEvent, WorkerCtl};
 use crate::virtio::queue::{QUEUE_SIZE_MAX, QueueReg};
 use crate::virtio::{DevStatus, DeviceId, IrqSender, Result, VirtioFeature, error};
 use crate::{consts, impl_mmio_for_zerocopy, mem};
@@ -193,35 +194,32 @@ where
     queues: Arc<[QueueReg]>,
     irq_sender: Arc<PciIrqSender<M>>,
     notifiers: Option<Arc<[Notifier]>>,
-    event_tx: Sender<WakeEvent<PciIrqSender<M>>>,
-    notifier: Arc<Notifier>,
+    ctl: Arc<WorkerCtl<PciIrqSender<M>>>,
+    /// Sequence number of the device reset in progress, if any.
+    ///
+    /// Writes to the common configuration hold the lock, so that they never
+    /// interleave with a reset.
+    reset_seq: Mutex<Option<u64>>,
 }
 
 impl<M> VirtioPciRegisterMmio<M>
 where
     M: MsiSender,
 {
-    fn wake_up_dev(&self, event: WakeEvent<PciIrqSender<M>>) {
-        let is_start = matches!(event, WakeEvent::Start { .. });
-        if let Err(e) = self.event_tx.send(event) {
-            log::error!("{}: failed to send event: {e}", self.name);
-            return;
-        }
-        if is_start {
-            return;
-        }
-        if let Err(e) = self.notifier.notify() {
-            log::error!("{}: failed to wake up device: {e}", self.name);
-        }
-    }
-
-    fn reset(&self) {
+    /// Returns the registers to their initial values.
+    ///
+    /// The worker must not be using the queues, i.e. it either has not
+    /// started or has completed the reset.
+    ///
+    /// The MSI-X table belongs to the PCI function, so a device reset keeps
+    /// what the guest has programmed into it.
+    fn complete_reset(&self) {
         let config_msix = &self.irq_sender.msix_vector.config;
         config_msix.store(VIRTIO_MSI_NO_VECTOR, Ordering::Release);
         for q_vector in self.irq_sender.msix_vector.queues.iter() {
             q_vector.store(VIRTIO_MSI_NO_VECTOR, Ordering::Release);
         }
-        self.irq_sender.msix_table.reset();
+        self.irq_sender.msix_table.release_irqfds();
         for q in self.queues.iter() {
             // Disable the queue before clearing its ring addresses, so that
             // a reader that checks `enabled` after loading the addresses
@@ -232,6 +230,65 @@ where
             q.driver.store(0, Ordering::Release);
             q.device.store(0, Ordering::Release);
         }
+        // Clear the status last, since it tells the driver that the reset
+        // has completed.
+        self.reg.status.store(0, Ordering::Release);
+    }
+
+    /// Polls a pending reset and completes it once the worker stops.
+    ///
+    /// Returns `true` if no reset is in progress.
+    fn poll_reset(&self, reset_seq: &mut Option<u64>) -> bool {
+        let Some(seq) = *reset_seq else {
+            return true;
+        };
+        if !self.ctl.poll_reset(seq) {
+            return false;
+        }
+        self.complete_reset();
+        *reset_seq = None;
+        true
+    }
+
+    /// Tells the driver that the device needs a reset, after the worker of
+    /// the running device has failed (VirtIO spec 1.4, Sec 2.1.2).
+    fn set_needs_reset(&self) {
+        let status = DevStatus::from_bits_retain(self.reg.status.load(Ordering::Acquire));
+        if status.contains(DevStatus::DRIVER_OK) {
+            let status = status | DevStatus::NEEDS_RESET;
+            self.reg.status.store(status.bits(), Ordering::Release);
+        }
+    }
+
+    /// Resets the device and waits for the worker to complete the reset.
+    fn reset_sync(&self, timeout: Duration) -> Result<()> {
+        let mut reset_seq = self.reset_seq.lock();
+        let seq = self.ctl.request_reset();
+        self.ctl.wait_reset(seq, timeout)?;
+        self.complete_reset();
+        *reset_seq = None;
+        Ok(())
+    }
+
+    fn notify_queue(&self, offset: usize, size: u8, val: u64) {
+        let q_index = (offset - VirtioPciRegister::OFFSET_QUEUE_NOTIFY) / size_of::<u32>();
+        if q_index >= self.queues.len() {
+            self.log_invalid_write(offset, size, val);
+            return;
+        }
+        if self.notifiers.is_some() {
+            log::warn!("{}: notifying queue-{q_index} by vm exit!", self.name);
+        }
+        let q_index = q_index as u16;
+        self.ctl.wake(WakeEvent::Notify { q_index });
+    }
+
+    fn log_invalid_write(&self, offset: usize, size: u8, val: u64) {
+        log::error!(
+            "{}: write 0x{val:0width$x} to invalid register offset = {offset:#x}",
+            self.name,
+            width = 2 * size as usize
+        );
     }
 
     /// Returns the queue `q_sel` if the driver may still change its ring
@@ -310,7 +367,14 @@ where
                 self.irq_sender.msix_vector.config.load(Ordering::Acquire) as u64
             }
             VirtioCommonCfg::LAYOUT_NUM_QUEUES => self.queues.len() as u64,
-            VirtioCommonCfg::LAYOUT_DEVICE_STATUS => reg.status.load(Ordering::Acquire) as u64,
+            VirtioCommonCfg::LAYOUT_DEVICE_STATUS => {
+                // The driver polls device_status until a reset completes.
+                let mut reset_seq = self.reset_seq.lock();
+                if self.poll_reset(&mut reset_seq) && self.ctl.failed() {
+                    self.set_needs_reset();
+                }
+                reg.status.load(Ordering::Acquire) as u64
+            }
             VirtioCommonCfg::LAYOUT_CONFIG_GENERATION => {
                 0 // TODO: support device config change at runtime
             }
@@ -407,8 +471,27 @@ where
     }
 
     fn write(&self, offset: u64, size: u8, val: u64) -> mem::Result<Action> {
+        let offset = offset as usize;
+        // Queue notifications change no registers, so they skip the reset
+        // lock and stay fast.
+        if offset >= VirtioPciRegister::OFFSET_QUEUE_NOTIFY {
+            self.notify_queue(offset, size, val);
+            return Ok(Action::None);
+        }
+        let mut reset_seq = self.reset_seq.lock();
+        if !self.poll_reset(&mut reset_seq) {
+            // The driver must wait for device_status to read 0 before it
+            // reinitializes the device (VirtIO spec 1.4, Sec 4.1.4.3.2).
+            if offset != VirtioCommonCfg::OFFSET_DEVICE_STATUS || val != 0 {
+                log::warn!(
+                    "{}: ignored write {val:#x} to {offset:#x} during reset",
+                    self.name
+                );
+            }
+            return Ok(Action::None);
+        }
         let reg = &self.reg;
-        match (offset as usize, size as usize) {
+        match (offset, size as usize) {
             VirtioCommonCfg::LAYOUT_DEVICE_FEATURE_SELECT => {
                 reg.device_feature_sel.store(val as u32, Ordering::Release);
             }
@@ -456,21 +539,26 @@ where
                     log::warn!("{}: invalid status change: {old:#x} -> {val:#x}", self.name);
                     return Ok(Action::None);
                 }
-                reg.status.store(status.bits(), Ordering::Release);
                 if status.is_empty() {
-                    self.reset();
                     if old.contains(DevStatus::DRIVER_OK) {
-                        self.wake_up_dev(WakeEvent::Reset);
+                        // The worker might still be using the queues. Keep
+                        // all registers, including the non-zero status, until
+                        // it completes the reset.
+                        *reset_seq = Some(self.ctl.request_reset());
+                    } else {
+                        self.complete_reset();
+                    }
+                } else {
+                    reg.status.store(status.bits(), Ordering::Release);
+                    if !old.contains(DevStatus::DRIVER_OK) && status.contains(DevStatus::DRIVER_OK)
+                    {
+                        let param = StartParam {
+                            feature: reg.get_driver_feature(),
+                            irq_sender: self.irq_sender.clone(),
+                            notifiers: self.notifiers.clone(),
+                        };
+                        self.ctl.wake(WakeEvent::Start { param });
                     }
-                } else if !old.contains(DevStatus::DRIVER_OK)
-                    && status.contains(DevStatus::DRIVER_OK)
-                {
-                    let param = StartParam {
-                        feature: reg.get_driver_feature(),
-                        irq_sender: self.irq_sender.clone(),
-                        notifiers: self.notifiers.clone(),
-                    };
-                    self.wake_up_dev(WakeEvent::Start { param });
                 }
             }
             VirtioCommonCfg::LAYOUT_QUEUE_SELECT => {
@@ -591,26 +679,7 @@ where
                     log::warn!("{}: queue reset requested but not supported", self.name);
                 }
             }
-            (offset, _)
-                if offset >= VirtioPciRegister::OFFSET_QUEUE_NOTIFY
-                    && offset
-                        < VirtioPciRegister::OFFSET_QUEUE_NOTIFY
-                            + size_of::<u32>() * self.queues.len() =>
-            {
-                let q_index = (offset - VirtioPciRegister::OFFSET_QUEUE_NOTIFY) as u16 / 4;
-                if self.notifiers.is_some() {
-                    log::warn!("{}: notifying queue-{q_index} by vm exit!", self.name);
-                }
-                let event = WakeEvent::Notify { q_index };
-                self.wake_up_dev(event)
-            }
-            _ => {
-                log::error!(
-                    "{}: write 0x{val:0width$x} to invalid register offset = {offset:#x}",
-                    self.name,
-                    width = 2 * size as usize
-                );
-            }
+            _ => self.log_invalid_write(offset, size, val),
         }
         Ok(Action::None)
     }
@@ -947,8 +1016,8 @@ where
                 device_feature,
                 ..Default::default()
             },
-            event_tx: dev.event_tx.clone(),
-            notifier: dev.notifier.clone(),
+            ctl: dev.ctl.clone(),
+            reset_seq: Mutex::new(None),
             queues: dev.queue_regs.clone(),
             irq_sender: Arc::new(PciIrqSender {
                 msix_vector,
@@ -1015,9 +1084,10 @@ where
     }
 
     fn reset(&self) -> pci::Result<()> {
-        self.registers.wake_up_dev(WakeEvent::Reset);
-        self.registers.reset();
-        self.registers.reg.status.store(0, Ordering::Release);
+        let ret = self.registers.reset_sync(RESET_TIMEOUT);
+        ret.box_trace(pci::error::Reset)?;
+        // Unlike a device reset, a PCI reset also clears the MSI-X table.
+        self.registers.irq_sender.msix_table.reset();
         Ok(())
     }
 }
diff --git a/alioth/src/virtio/pci_test.rs b/alioth/src/virtio/pci_test.rs
index b1aad476..b618c8ee 100644
--- a/alioth/src/virtio/pci_test.rs
+++ b/alioth/src/virtio/pci_test.rs
@@ -16,12 +16,15 @@ use std::mem::size_of;
 use std::os::fd::AsRawFd;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU64, Ordering};
+use std::thread;
+use std::time::Duration;
 
 use assert_matches::assert_matches;
-use flume::Receiver;
+use flume::{Receiver, Sender};
 use parking_lot::{Mutex, RwLock};
 use rstest::rstest;
 
+use crate::hv::IrqFd;
 use crate::hv::tests::{RegisteredAddr, TestIrqFd, TestMsiSender, TestNotifierRegistry};
 use crate::mem::emulated::{Action, Mmio};
 use crate::mem::{self, MemRange, MemRegion, MemRegionEntry, MemRegionType};
@@ -31,20 +34,23 @@ use crate::pci::cap::{
 use crate::pci::config::{BAR_MEM32, BAR_MEM64, BAR_PREFETCHABLE, PciConfigArea};
 use crate::pci::{Pci, PciBar};
 use crate::sync::notifier::Notifier;
-use crate::virtio::dev::{Register, VirtioDevice, WakeEvent};
+use crate::virtio::dev::{Register, VirtioDevice, WakeEvent, WorkerCtl, WorkerReport};
 use crate::virtio::pci::{
     PciIrqSender, VIRTIO_MSI_NO_VECTOR, VirtioCommonCfg, VirtioPciCap, VirtioPciCap64,
     VirtioPciDevice, VirtioPciMsixVector, VirtioPciNotifyCap, VirtioPciRegister,
     VirtioPciRegisterMmio,
 };
 use crate::virtio::queue::{QUEUE_SIZE_MAX, QueueReg};
-use crate::virtio::{DevStatus, DeviceId, IrqSender, VirtioFeature};
+use crate::virtio::{DevStatus, DeviceId, Error, IrqSender, VirtioFeature};
 
 type TestMmio = VirtioPciRegisterMmio<TestMsiSender>;
 type TestWakeReceiver = flume::Receiver<WakeEvent<PciIrqSender<TestMsiSender>>>;
 
-fn create_test_mmio(queues: Arc<[QueueReg]>) -> (TestMmio, TestWakeReceiver) {
+/// Creates the registers of a device, together with the worker ends of the
+/// wake event channel and the reset completion channel.
+fn create_test_mmio(queues: Arc<[QueueReg]>) -> (TestMmio, TestWakeReceiver, Sender<WorkerReport>) {
     let (event_tx, event_rx) = flume::unbounded();
+    let (report_tx, report_rx) = flume::unbounded();
     let notifier = Arc::new(Notifier::new().unwrap());
     let msi_sender = TestMsiSender::default();
     let msix_table = Arc::new(MsixTableMmio {
@@ -70,10 +76,15 @@ fn create_test_mmio(queues: Arc<[QueueReg]>) -> (TestMmio, TestWakeReceiver) {
         queues,
         irq_sender,
         notifiers: None,
-        event_tx,
-        notifier,
+        ctl: Arc::new(WorkerCtl::new(
+            "test-virtio-pci".into(),
+            event_tx,
+            notifier,
+            report_rx,
+        )),
+        reset_seq: Mutex::new(None),
     };
-    (mmio, event_rx)
+    (mmio, event_rx, report_tx)
 }
 
 #[rstest]
@@ -90,7 +101,7 @@ fn test_queue_address_reads(#[case] offset: usize, #[case] expected: u64) {
         device: AtomicU64::new(0x0123_4567_89ab_cdef),
         ..Default::default()
     }]);
-    let (mmio, _) = create_test_mmio(queues);
+    let (mmio, _, _report_tx) = create_test_mmio(queues);
     assert_matches!(mmio.read(offset as u64, 4), Ok(val) if val == expected);
 }
 
@@ -119,7 +130,7 @@ fn test_queue_alignment(
         device: AtomicU64::new(0x3000_0000),
         ..Default::default()
     }]);
-    let (mmio, _event_rx) = create_test_mmio(queues);
+    let (mmio, _event_rx, _report_tx) = create_test_mmio(queues);
     if packed {
         assert_matches!(
             mmio.write(VirtioCommonCfg::OFFSET_DRIVER_FEATURE_SELECT as u64, 4, 1),
@@ -146,7 +157,7 @@ fn test_queue_alignment(
 #[case(VirtioCommonCfg::OFFSET_QUEUE_ENABLE, 2, 1)]
 fn test_queue_registers_write_read(#[case] offset: usize, #[case] size: u8, #[case] val: u64) {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, _) = create_test_mmio(queues);
+    let (mmio, _, _report_tx) = create_test_mmio(queues);
 
     assert_matches!(
         mmio.write(VirtioCommonCfg::OFFSET_QUEUE_SELECT as u64, 2, 0),
@@ -166,7 +177,7 @@ fn test_queue_size_write_read(#[case] packed: bool, #[case] input: u64, #[case]
         size: AtomicU16::new(128),
         ..Default::default()
     }]);
-    let (mmio, _event_rx) = create_test_mmio(queues);
+    let (mmio, _event_rx, _report_tx) = create_test_mmio(queues);
     if packed {
         assert_matches!(
             mmio.write(VirtioCommonCfg::OFFSET_DRIVER_FEATURE_SELECT as u64, 4, 1),
@@ -191,7 +202,7 @@ fn test_queue_size_write_read(#[case] packed: bool, #[case] input: u64, #[case]
 #[test]
 fn test_queue_size_locked_when_enabled() {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, _) = create_test_mmio(queues);
+    let (mmio, _, _report_tx) = create_test_mmio(queues);
 
     assert_matches!(
         mmio.write(VirtioCommonCfg::OFFSET_QUEUE_SELECT as u64, 2, 0),
@@ -215,7 +226,7 @@ fn test_queue_size_locked_when_enabled() {
 #[test]
 fn test_queue_addrs_locked_when_enabled() {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, _) = create_test_mmio(queues.clone());
+    let (mmio, _, _report_tx) = create_test_mmio(queues.clone());
 
     let offsets = [
         VirtioCommonCfg::OFFSET_QUEUE_DESC_LO,
@@ -270,7 +281,7 @@ fn test_queue_addrs_locked_when_enabled() {
 #[case(VirtioCommonCfg::OFFSET_QUEUE_ENABLE, 2, 1)]
 fn test_out_of_bounds_queue_writes(#[case] offset: usize, #[case] size: u8, #[case] val: u64) {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, _) = create_test_mmio(queues);
+    let (mmio, _, _report_tx) = create_test_mmio(queues);
 
     assert_matches!(
         mmio.write(VirtioCommonCfg::OFFSET_QUEUE_SELECT as u64, 2, 99),
@@ -319,7 +330,7 @@ fn test_common_cfg_queue_reads(
             ..Default::default()
         },
     ]);
-    let (mmio, _) = create_test_mmio(queues);
+    let (mmio, _, _report_tx) = create_test_mmio(queues);
 
     assert_matches!(
         mmio.write(VirtioCommonCfg::OFFSET_QUEUE_SELECT as u64, 2, q_sel),
@@ -336,7 +347,7 @@ fn test_common_cfg_queue_reads(
 #[case(VirtioCommonCfg::OFFSET_QUEUE_RESET, 2, 0)]
 fn test_common_cfg_misc_reads(#[case] offset: usize, #[case] size: u8, #[case] expected: u64) {
     let queues = Arc::new([QueueReg::default(), QueueReg::default()]);
-    let (mmio, _) = create_test_mmio(queues);
+    let (mmio, _, _report_tx) = create_test_mmio(queues);
     assert_matches!(mmio.read(offset as u64, size), Ok(val) if val == expected);
 }
 
@@ -346,7 +357,7 @@ fn test_common_cfg_misc_reads(#[case] offset: usize, #[case] size: u8, #[case] e
 #[case(10, 0)]
 fn test_device_feature_reads(#[case] sel: u64, #[case] expected_feature: u64) {
     let queues = Arc::new([QueueReg::default()]);
-    let (mut mmio, _) = create_test_mmio(queues);
+    let (mut mmio, _, _report_tx) = create_test_mmio(queues);
     mmio.reg.device_feature = [0x1111_2222, 0x3333_4444, 0, 0];
 
     assert_matches!(
@@ -374,7 +385,7 @@ fn test_device_feature_reads(#[case] sel: u64, #[case] expected_feature: u64) {
 #[case(10, 0x1234, 0)]
 fn test_driver_features(#[case] bank: u64, #[case] write_val: u64, #[case] expected: u64) {
     let queues = Arc::new([QueueReg::default()]);
-    let (mut mmio, _event_rx) = create_test_mmio(queues);
+    let (mut mmio, _event_rx, _report_tx) = create_test_mmio(queues);
     mmio.reg.device_feature = [0x1234_5678, 0x0000_0005, 0, 0];
 
     assert_matches!(
@@ -402,7 +413,7 @@ fn test_driver_features(#[case] bank: u64, #[case] write_val: u64, #[case] expec
 #[test]
 fn test_driver_features_locked_after_features_ok() {
     let queues = Arc::new([QueueReg::default()]);
-    let (mut mmio, _event_rx) = create_test_mmio(queues);
+    let (mut mmio, _event_rx, _report_tx) = create_test_mmio(queues);
     mmio.reg.device_feature = [0x1234_5678, 0x0000_0005, 0, 0];
 
     assert_matches!(
@@ -447,7 +458,7 @@ fn test_driver_features_locked_after_features_ok() {
 #[test]
 fn test_device_status_valid_transitions() {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, event_rx) = create_test_mmio(queues.clone());
+    let (mmio, event_rx, report_tx) = create_test_mmio(queues.clone());
 
     // Initially status is 0 (empty)
     assert_matches!(
@@ -621,12 +632,28 @@ fn test_device_status_valid_transitions() {
         mmio.write(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1, 0),
         Ok(Action::None)
     );
+    let Ok(WakeEvent::Reset { seq }) = event_rx.try_recv() else {
+        panic!("expected a reset event");
+    };
+    assert!(event_rx.is_empty());
+
+    // The worker might still be using the queues, so nothing changes until
+    // it completes the reset.
+    assert_matches!(
+        mmio.read(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1),
+        Ok(status) if status == needs_reset.bits() as u64
+    );
+    assert!(queues[0].enabled.load(Ordering::Acquire));
+    assert_matches!(
+        mmio.read(VirtioCommonCfg::OFFSET_CONFIG_MSIX_VECTOR as u64, 2),
+        Ok(0)
+    );
+
+    report_tx.send(WorkerReport::ResetDone { seq }).unwrap();
     assert_matches!(
         mmio.read(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1),
         Ok(0)
     );
-    assert_matches!(event_rx.try_recv(), Ok(WakeEvent::Reset));
-    assert!(event_rx.is_empty());
     assert!(!queues[0].enabled.load(Ordering::Acquire));
     assert_matches!(
         mmio.read(VirtioCommonCfg::OFFSET_CONFIG_MSIX_VECTOR as u64, 2),
@@ -634,6 +661,150 @@ fn test_device_status_valid_transitions() {
     );
 }
 
+/// Drives the device to DRIVER_OK with queue 0 enabled, and then asks for a
+/// reset. Returns the sequence number of the reset.
+fn start_then_reset(mmio: &TestMmio, event_rx: &TestWakeReceiver) -> u64 {
+    let driver_ok =
+        DevStatus::ACK | DevStatus::DRIVER | DevStatus::FEATURES_OK | DevStatus::DRIVER_OK;
+    for (offset, size, val) in [
+        (VirtioCommonCfg::OFFSET_QUEUE_SELECT, 2, 0),
+        (VirtioCommonCfg::OFFSET_QUEUE_ENABLE, 2, 1),
+        (
+            VirtioCommonCfg::OFFSET_DEVICE_STATUS,
+            1,
+            driver_ok.bits() as u64,
+        ),
+        (VirtioCommonCfg::OFFSET_DEVICE_STATUS, 1, 0),
+    ] {
+        assert_matches!(mmio.write(offset as u64, size, val), Ok(Action::None));
+    }
+    assert_matches!(event_rx.try_recv(), Ok(WakeEvent::Start { .. }));
+    let Ok(WakeEvent::Reset { seq }) = event_rx.try_recv() else {
+        panic!("expected a reset event");
+    };
+    seq
+}
+
+#[test]
+fn test_device_reset_in_progress() {
+    let queues = Arc::new([QueueReg::default()]);
+    let (mmio, event_rx, report_tx) = create_test_mmio(queues.clone());
+    let seq = start_then_reset(&mmio, &event_rx);
+
+    // The driver must leave the device alone until the reset completes, so
+    // all writes are ignored, and writing 0 again does not reset it twice.
+    for (offset, size, val) in [
+        (VirtioCommonCfg::OFFSET_DEVICE_STATUS, 1, 0),
+        (
+            VirtioCommonCfg::OFFSET_DEVICE_STATUS,
+            1,
+            DevStatus::ACK.bits() as u64,
+        ),
+        (VirtioCommonCfg::OFFSET_QUEUE_SELECT, 2, 1),
+        (VirtioCommonCfg::OFFSET_CONFIG_MSIX_VECTOR, 2, 1),
+    ] {
+        assert_matches!(mmio.write(offset as u64, size, val), Ok(Action::None));
+    }
+    assert!(event_rx.is_empty());
+    let driver_ok =
+        DevStatus::ACK | DevStatus::DRIVER | DevStatus::FEATURES_OK | DevStatus::DRIVER_OK;
+    assert_matches!(
+        mmio.read(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1),
+        Ok(status) if status == driver_ok.bits() as u64
+    );
+    assert_matches!(
+        mmio.read(VirtioCommonCfg::OFFSET_QUEUE_SELECT as u64, 2),
+        Ok(0)
+    );
+    assert_matches!(
+        mmio.read(VirtioCommonCfg::OFFSET_CONFIG_MSIX_VECTOR as u64, 2),
+        Ok(vector) if vector == VIRTIO_MSI_NO_VECTOR as u64
+    );
+    assert!(queues[0].enabled.load(Ordering::Acquire));
+
+    // Once the worker has completed the reset, the driver can set up the
+    // device again.
+    report_tx.send(WorkerReport::ResetDone { seq }).unwrap();
+    assert_matches!(
+        mmio.write(
+            VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64,
+            1,
+            DevStatus::ACK.bits() as u64
+        ),
+        Ok(Action::None)
+    );
+    assert_matches!(
+        mmio.read(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1),
+        Ok(status) if status == DevStatus::ACK.bits() as u64
+    );
+    assert!(!queues[0].enabled.load(Ordering::Acquire));
+}
+
+#[test]
+fn test_device_needs_reset_after_worker_failure() {
+    let queues = Arc::new([QueueReg::default()]);
+    let (mmio, event_rx, report_tx) = create_test_mmio(queues);
+    let driver_ok =
+        DevStatus::ACK | DevStatus::DRIVER | DevStatus::FEATURES_OK | DevStatus::DRIVER_OK;
+    assert_matches!(
+        mmio.write(
+            VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64,
+            1,
+            driver_ok.bits() as u64
+        ),
+        Ok(Action::None)
+    );
+    assert_matches!(event_rx.try_recv(), Ok(WakeEvent::Start { .. }));
+
+    report_tx.send(WorkerReport::Failed).unwrap();
+    let needs_reset = driver_ok | DevStatus::NEEDS_RESET;
+    assert_matches!(
+        mmio.read(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1),
+        Ok(status) if status == needs_reset.bits() as u64
+    );
+
+    // Resetting the device brings it back into service.
+    assert_matches!(
+        mmio.write(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1, 0),
+        Ok(Action::None)
+    );
+    let Ok(WakeEvent::Reset { seq }) = event_rx.try_recv() else {
+        panic!("expected a reset event");
+    };
+    report_tx.send(WorkerReport::ResetDone { seq }).unwrap();
+    assert_matches!(
+        mmio.read(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1),
+        Ok(0)
+    );
+    assert_matches!(
+        mmio.write(
+            VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64,
+            1,
+            DevStatus::ACK.bits() as u64
+        ),
+        Ok(Action::None)
+    );
+    assert_matches!(
+        mmio.read(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1),
+        Ok(status) if status == DevStatus::ACK.bits() as u64
+    );
+}
+
+#[test]
+fn test_device_reset_worker_exited() {
+    let queues = Arc::new([QueueReg::default()]);
+    let (mmio, event_rx, report_tx) = create_test_mmio(queues.clone());
+    start_then_reset(&mmio, &event_rx);
+
+    // A worker that has exited no longer uses the queues.
+    drop(report_tx);
+    assert_matches!(
+        mmio.read(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1),
+        Ok(0)
+    );
+    assert!(!queues[0].enabled.load(Ordering::Acquire));
+}
+
 #[rstest]
 // Invalid transitions from empty state (skipping ACK or invalid combinations)
 #[case(DevStatus::empty(), DevStatus::DRIVER.bits() as u64)]
@@ -668,7 +839,7 @@ fn test_device_status_valid_transitions() {
 )]
 fn test_device_status_invalid_transitions(#[case] initial: DevStatus, #[case] invalid_write: u64) {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, event_rx) = create_test_mmio(queues);
+    let (mmio, event_rx, _report_tx) = create_test_mmio(queues);
 
     if !initial.is_empty() {
         assert_matches!(
@@ -722,7 +893,7 @@ fn test_device_status_multistep_transition(
     #[case] expect_start: bool,
 ) {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, event_rx) = create_test_mmio(queues);
+    let (mmio, event_rx, _report_tx) = create_test_mmio(queues);
 
     // Set driver features
     assert_matches!(
@@ -788,7 +959,7 @@ fn test_device_status_multistep_transition(
 #[test]
 fn test_device_status_reset_without_driver_ok() {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, event_rx) = create_test_mmio(queues.clone());
+    let (mmio, event_rx, _report_tx) = create_test_mmio(queues.clone());
 
     // Set status to ACK | DRIVER
     let ack_driver = DevStatus::ACK | DevStatus::DRIVER;
@@ -839,6 +1010,53 @@ fn test_device_status_reset_without_driver_ok() {
     );
 }
 
+#[test]
+fn test_device_reset_keeps_msix_table() {
+    let queues = Arc::new([QueueReg::default()]);
+    let (mmio, _event_rx, _report_tx) = create_test_mmio(queues);
+
+    let irqfd = TestIrqFd::default();
+    irqfd.set_addr_lo(0xfee0_0000).unwrap();
+    irqfd.set_data(0x41).unwrap();
+    irqfd.set_masked(false).unwrap();
+    *mmio.irq_sender.msix_table.entries.write() = vec![
+        MsixTableMmioEntry::Entry(MsixTableEntry {
+            addr_lo: 0xfee0_1000,
+            addr_hi: 0,
+            data: 0x42,
+            control: MsixVectorCtrl(0),
+        }),
+        MsixTableMmioEntry::IrqFd(irqfd),
+    ]
+    .into_boxed_slice();
+
+    assert_matches!(
+        mmio.write(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1, 0),
+        Ok(Action::None)
+    );
+
+    // The device reset releases the irqfds, but keeps what the guest has
+    // programmed into the MSI-X table.
+    let entries = mmio.irq_sender.msix_table.entries.read();
+    assert_matches!(
+        **entries,
+        [
+            MsixTableMmioEntry::Entry(MsixTableEntry {
+                addr_lo: 0xfee0_1000,
+                data: 0x42,
+                ..
+            }),
+            MsixTableMmioEntry::Entry(MsixTableEntry {
+                addr_lo: 0xfee0_0000,
+                data: 0x41,
+                ..
+            })
+        ]
+    );
+    assert!(!entries[0].get_masked());
+    assert!(!entries[1].get_masked());
+}
+
 #[test]
 fn test_device_reset_queue_registers() {
     let queues = Arc::new([
@@ -851,7 +1069,7 @@ fn test_device_reset_queue_registers() {
             ..Default::default()
         },
     ]);
-    let (mmio, event_rx) = create_test_mmio(queues.clone());
+    let (mmio, event_rx, report_tx) = create_test_mmio(queues.clone());
 
     // Initialize MSI-X table with 2 Entry slots
     *mmio.irq_sender.msix_table.entries.write() = vec![
@@ -904,8 +1122,15 @@ fn test_device_reset_queue_registers() {
         mmio.write(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1, 0),
         Ok(Action::None)
     );
-    assert_matches!(event_rx.try_recv(), Ok(WakeEvent::Reset));
+    let Ok(WakeEvent::Reset { seq }) = event_rx.try_recv() else {
+        panic!("expected a reset event");
+    };
     assert!(event_rx.is_empty());
+    report_tx.send(WorkerReport::ResetDone { seq }).unwrap();
+    assert_matches!(
+        mmio.read(VirtioCommonCfg::OFFSET_DEVICE_STATUS as u64, 1),
+        Ok(0)
+    );
 
     // Verify all queue registers are reset
     for (q_index, _) in queues.iter().enumerate() {
@@ -945,7 +1170,7 @@ fn test_device_reset_queue_registers() {
 #[case(VirtioCommonCfg::OFFSET_QUEUE_MSIX_VECTOR, Some(0))]
 fn test_msix_vector_configuration(#[case] offset: usize, #[case] queue_sel: Option<u16>) {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, _) = create_test_mmio(queues);
+    let (mmio, _, _report_tx) = create_test_mmio(queues);
 
     // Initialize MSI-X table with 2 Entry slots
     *mmio.irq_sender.msix_table.entries.write() = vec![
@@ -982,7 +1207,7 @@ fn test_msix_vector_configuration(#[case] offset: usize, #[case] queue_sel: Opti
 #[case(VirtioPciRegister::OFFSET_QUEUE_NOTIFY + size_of::<u32>(), false)]
 fn test_queue_notify(#[case] offset: usize, #[case] expect_wake: bool) {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, event_rx) = create_test_mmio(queues);
+    let (mmio, event_rx, _report_tx) = create_test_mmio(queues);
 
     assert_matches!(mmio.write(offset as u64, 2, 0), Ok(Action::None));
     if expect_wake {
@@ -1017,8 +1242,13 @@ fn test_notify_with_notifiers() {
         queues,
         irq_sender,
         notifiers: Some(Arc::new([Notifier::new().unwrap()])),
-        event_tx,
-        notifier,
+        ctl: Arc::new(WorkerCtl::new(
+            "test-virtio-pci-notifier".into(),
+            event_tx,
+            notifier,
+            flume::unbounded().1,
+        )),
+        reset_seq: Mutex::new(None),
     };
 
     assert_matches!(
@@ -1031,7 +1261,7 @@ fn test_notify_with_notifiers() {
 #[test]
 fn test_wake_up_dev_channel_error() {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, event_rx) = create_test_mmio(queues);
+    let (mmio, event_rx, _report_tx) = create_test_mmio(queues);
 
     // Drop receiver so send will fail
     drop(event_rx);
@@ -1050,7 +1280,7 @@ fn test_wake_up_dev_channel_error() {
 #[case(0xdead_beef, 4, 0x1234)]
 fn test_invalid_register_writes(#[case] offset: u64, #[case] size: u8, #[case] val: u64) {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, _) = create_test_mmio(queues);
+    let (mmio, _, _report_tx) = create_test_mmio(queues);
     assert_matches!(mmio.write(offset, size, val), Ok(Action::None));
 }
 
@@ -1059,7 +1289,7 @@ fn test_invalid_register_writes(#[case] offset: u64, #[case] size: u8, #[case] v
 #[case(0, 3)]
 fn test_invalid_register_reads(#[case] offset: u64, #[case] size: u8) {
     let queues = Arc::new([QueueReg::default()]);
-    let (mmio, _) = create_test_mmio(queues);
+    let (mmio, _, _report_tx) = create_test_mmio(queues);
     assert_matches!(mmio.read(offset, size), Ok(0));
 }
 
@@ -1068,7 +1298,7 @@ fn test_invalid_register_reads(#[case] offset: u64, #[case] size: u8) {
 #[case(3, (size_of::<VirtioPciRegister>() + size_of::<u32>() * 4) as u64)]
 fn test_mmio_size(#[case] num_queues: usize, #[case] expected_size: u64) {
     let queues = (0..num_queues).map(|_| QueueReg::default()).collect();
-    let (mmio, _) = create_test_mmio(queues);
+    let (mmio, _, _report_tx) = create_test_mmio(queues);
     assert_eq!(mmio.size(), expected_size);
 }
 
@@ -1096,11 +1326,16 @@ fn create_test_virtio_device<S>(
     config_size: u64,
     shared_mem: Option<Arc<MemRegion>>,
     num_queues: usize,
-) -> (VirtioDevice<S>, Receiver<WakeEvent<S>>)
+) -> (
+    VirtioDevice<S>,
+    Receiver<WakeEvent<S>>,
+    Sender<WorkerReport>,
+)
 where
     S: IrqSender,
 {
     let (event_tx, event_rx) = flume::unbounded();
+    let (report_tx, report_rx) = flume::unbounded();
     let notifier = Arc::new(Notifier::new().unwrap());
     let queue_regs = (0..num_queues)
         .map(|_| QueueReg::default())
@@ -1112,11 +1347,15 @@ where
         device_feature: (VirtioFeature::VERSION_1 | VirtioFeature::RING_PACKED).bits(),
         queue_regs,
         shared_mem_regions: shared_mem,
-        notifier,
-        event_tx,
+        ctl: Arc::new(WorkerCtl::new(
+            "test-dev".into(),
+            event_tx,
+            notifier,
+            report_rx,
+        )),
         worker_handle: None,
     };
-    (dev, event_rx)
+    (dev, event_rx, report_tx)
 }
 
 #[test]
@@ -1324,7 +1563,8 @@ fn test_virtio_pci_device_classes(
     #[case] expected_subclass: u8,
     #[case] expected_dev_id: u16,
 ) {
-    let (dev, _rx) = create_test_virtio_device::<PciIrqSender<TestMsiSender>>(id, 0, None, 2);
+    let (dev, _rx, _report_tx) =
+        create_test_virtio_device::<PciIrqSender<TestMsiSender>>(id, 0, None, 2);
     let pci_dev = VirtioPciDevice::new(
         dev,
         TestMsiSender::default(),
@@ -1377,7 +1617,7 @@ fn test_virtio_pci_device_with_config_and_shared_memory_prefetchable() {
         ],
         callbacks: Mutex::new(vec![]),
     });
-    let (dev, _rx) = create_test_virtio_device::<PciIrqSender<TestMsiSender>>(
+    let (dev, _rx, _report_tx) = create_test_virtio_device::<PciIrqSender<TestMsiSender>>(
         DeviceId::FILE_SYSTEM,
         32,
         Some(shared_mem),
@@ -1429,7 +1669,7 @@ fn test_virtio_pci_device_shared_memory_non_prefetchable() {
         }],
         callbacks: Mutex::new(vec![]),
     });
-    let (dev, _rx) = create_test_virtio_device::<PciIrqSender<TestMsiSender>>(
+    let (dev, _rx, _report_tx) = create_test_virtio_device::<PciIrqSender<TestMsiSender>>(
         DeviceId::FILE_SYSTEM,
         0,
         Some(shared_mem),
@@ -1448,7 +1688,7 @@ fn test_virtio_pci_device_shared_memory_non_prefetchable() {
 
 #[test]
 fn test_virtio_pci_device_notifier_callback() {
-    let (dev, _rx) =
+    let (dev, _rx, _report_tx) =
         create_test_virtio_device::<PciIrqSender<TestMsiSender>>(DeviceId::NET, 0, None, 2);
     let registry = TestNotifierRegistry::default();
     let registered = registry.registered.clone();
@@ -1493,7 +1733,7 @@ fn test_virtio_pci_device_notifier_callback() {
 
 #[test]
 fn test_virtio_pci_device_notifier_fallback() {
-    let (dev, _rx) =
+    let (dev, _rx, _report_tx) =
         create_test_virtio_device::<PciIrqSender<TestMsiSender>>(DeviceId::NET, 0, None, 1);
     let pci_dev = VirtioPciDevice::new(
         dev,
@@ -1513,7 +1753,7 @@ fn test_virtio_pci_device_notifier_fallback() {
 
 #[test]
 fn test_virtio_pci_device_pci_reset() {
-    let (dev, event_rx) =
+    let (dev, event_rx, report_tx) =
         create_test_virtio_device::<PciIrqSender<TestMsiSender>>(DeviceId::NET, 0, None, 1);
     let pci_dev = VirtioPciDevice::new(
         dev,
@@ -1523,24 +1763,74 @@ fn test_virtio_pci_device_pci_reset() {
     .unwrap();
 
     // Set queue enabled and device status
-    pci_dev.registers.queues[0]
-        .enabled
-        .store(true, Ordering::Release);
-    pci_dev
-        .registers
-        .reg
-        .status
-        .store(DevStatus::DRIVER_OK.bits(), Ordering::Release);
+    let registers = pci_dev.registers.clone();
+    registers.queues[0].enabled.store(true, Ordering::Release);
+    let driver_ok = DevStatus::DRIVER_OK.bits();
+    registers.reg.status.store(driver_ok, Ordering::Release);
+    registers.irq_sender.msix_table.entries.write()[0] =
+        MsixTableMmioEntry::Entry(MsixTableEntry {
+            addr_lo: 0xfee0_0000,
+            addr_hi: 0,
+            data: 0x41,
+            control: MsixVectorCtrl(0),
+        });
+
+    // A fake worker checks that nothing is cleared before it completes the
+    // reset.
+    let worker = thread::spawn(move || {
+        let Ok(WakeEvent::Reset { seq }) = event_rx.recv() else {
+            panic!("expected a reset event");
+        };
+        assert_eq!(registers.reg.status.load(Ordering::Acquire), driver_ok);
+        assert!(registers.queues[0].enabled.load(Ordering::Acquire));
+        report_tx.send(WorkerReport::ResetDone { seq }).unwrap();
+    });
 
-    // Call Pci::reset
+    // Pci::reset returns only after the worker has completed the reset.
     assert_matches!(pci_dev.reset(), Ok(()));
+    worker.join().unwrap();
 
-    // Device status should be cleared
     assert_eq!(pci_dev.registers.reg.status.load(Ordering::Acquire), 0);
-    // Queue should be disabled
     assert!(!pci_dev.registers.queues[0].enabled.load(Ordering::Acquire));
-    // Reset event sent
-    assert_matches!(event_rx.try_recv(), Ok(WakeEvent::Reset));
+    // Unlike a device reset, a PCI reset also clears the MSI-X table.
+    let entries = pci_dev.registers.irq_sender.msix_table.entries.read();
+    assert_matches!(
+        &entries[0],
+        MsixTableMmioEntry::Entry(MsixTableEntry {
+            addr_lo: 0,
+            data: 0,
+            ..
+        })
+    );
+    assert!(entries[0].get_masked());
+}
+
+#[test]
+fn test_virtio_pci_device_pci_reset_timeout() {
+    let (dev, event_rx, _report_tx) =
+        create_test_virtio_device::<PciIrqSender<TestMsiSender>>(DeviceId::NET, 0, None, 1);
+    let pci_dev = VirtioPciDevice::new(
+        dev,
+        TestMsiSender::default(),
+        Option::<TestNotifierRegistry>::None,
+    )
+    .unwrap();
+    let driver_ok = DevStatus::DRIVER_OK.bits();
+    pci_dev
+        .registers
+        .reg
+        .status
+        .store(driver_ok, Ordering::Release);
+
+    // The worker never completes the reset.
+    let timeout = Duration::from_millis(10);
+    assert_matches!(
+        pci_dev.registers.reset_sync(timeout),
+        Err(Error::ResetTimeout { timeout: t, .. }) if t == timeout
+    );
+    assert_matches!(event_rx.try_recv(), Ok(WakeEvent::Reset { .. }));
+    let status = pci_dev.registers.reg.status.load(Ordering::Acquire);
+    assert_eq!(status, driver_ok);
 }
 
 #[test]
diff --git a/alioth/src/virtio/virtio.rs b/alioth/src/virtio/virtio.rs
index b566d594..222e46d6 100644
--- a/alioth/src/virtio/virtio.rs
+++ b/alioth/src/virtio/virtio.rs
@@ -28,6 +28,8 @@ pub mod worker;
 use std::fmt::Debug;
 use std::os::fd::BorrowedFd;
 use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
 
 use bitflags::Flags;
 use snafu::Snafu;
@@ -79,6 +81,8 @@ pub enum Error {
     InvalidMsixVector { vector: u16 },
     #[snafu(display("Invalid virtq buffer"))]
     InvalidBuffer,
+    #[snafu(display("{name}: device did not complete reset within {timeout:?}"))]
+    ResetTimeout { name: Arc<str>, timeout: Duration },
     #[cfg(target_os = "linux")]
     #[snafu(display("vhost-user error"), context(false))]
     Vu { source: Box<vu::Error> },
diff --git a/alioth/src/virtio/vu/backend.rs b/alioth/src/virtio/vu/backend.rs
index 8a648319..76cdaba2 100644
--- a/alioth/src/virtio/vu/backend.rs
+++ b/alioth/src/virtio/vu/backend.rs
@@ -28,7 +28,7 @@ use zerocopy::IntoBytes;
 use crate::errors::DebugTrace;
 use crate::mem::mapped::{ArcMemPages, RamBus};
 use crate::sync::notifier::Notifier;
-use crate::virtio::dev::{StartParam, VirtioDevice, WakeEvent};
+use crate::virtio::dev::{RESET_TIMEOUT, StartParam, VirtioDevice, WakeEvent};
 use crate::virtio::vu::Error as VuError;
 use crate::virtio::vu::bindings::{
     MAX_CONFIG_SIZE, MemoryRegion, MemorySingleRegion, Message, VirtqAddr, VirtqState, VuFeature,
@@ -49,6 +49,8 @@ pub enum Error {
     Vu {
         source: Box<crate::virtio::vu::Error>,
     },
+    #[snafu(display("Virtio device error"), context(false))]
+    Virtio { source: Box<crate::virtio::Error> },
     #[snafu(display("failed to parse the payload of {req:?}"))]
     Parse { req: VuFrontMsg },
     #[snafu(display("frontend requested invalid queue index: {index}"))]
@@ -140,6 +142,11 @@ pub struct VuBackend {
     session: VuSession,
     channel: Option<Arc<VuChannel>>,
     status: DevStatus,
+    /// Whether the worker has started and not stopped since.
+    ///
+    /// The frontend can stop the worker with GET_VRING_BASE, which does not
+    /// change the device status.
+    running: bool,
     memory: Arc<RamBus>,
     dev: VirtioDevice<VuIrqSender>,
     init: VuInit,
@@ -159,6 +166,7 @@ impl VuBackend {
             dev,
             memory,
             status: DevStatus::empty(),
+            running: false,
             init: VuInit {
                 drv_feat: 0,
                 queues: (0..queue_num).map(|_| VuQueueInit::default()).collect(),
@@ -171,18 +179,19 @@ impl VuBackend {
         self.dev.name.as_ref()
     }
 
-    fn wake_up_dev(&self, event: WakeEvent<VuIrqSender>) {
-        let is_start = matches!(event, WakeEvent::Start { .. });
-        if let Err(e) = self.dev.event_tx.send(event) {
-            log::error!("{}: failed to send event: {e}", self.dev.name);
-            return;
-        }
-        if is_start {
-            return;
-        }
-        if let Err(e) = self.dev.notifier.notify() {
-            log::error!("{}: failed to wake up device: {e}", self.dev.name);
+    /// Stops the worker and waits until it no longer uses the queues.
+    ///
+    /// The frontend takes the reply to its request as the completion of the
+    /// stop, so the backend has to wait before replying.
+    fn stop_worker(&mut self) -> Result<()> {
+        if !self.running {
+            return Ok(());
         }
+        let ctl = &self.dev.ctl;
+        let seq = ctl.request_reset();
+        ctl.wait_reset(seq, RESET_TIMEOUT)?;
+        self.running = false;
+        Ok(())
     }
 
     fn convert_frontend_hva(&self, hva: u64) -> Result<u64> {
@@ -247,7 +256,7 @@ impl VuBackend {
     }
 
     fn handle_msg(&mut self, msg: &mut Message, fds: &mut [Option<OwnedFd>; 8]) -> Result<()> {
-        let name = &*self.dev.name;
+        let name = self.dev.name.clone();
         let (req, size) = (VuFrontMsg::from(msg.request), msg.size);
 
         match (req, size) {
@@ -296,10 +305,10 @@ impl VuBackend {
                 let channel = Arc::new(VuChannel {
                     conn: UnixStream::from(fd),
                 });
-                let r = self.dev.event_tx.send(WakeEvent::VuChannel {
+                let event = WakeEvent::VuChannel {
                     channel: channel.clone(),
-                });
-                if r.is_err() {
+                };
+                if !self.dev.ctl.wake(event) {
                     return error::SendChannel.fail();
                 }
                 self.channel = Some(channel);
@@ -360,6 +369,13 @@ impl VuBackend {
                 let Some(_q) = self.init.queues.get_mut(index as usize) else {
                     return error::InvalidQueue { index }.fail();
                 };
+                // GET_VRING_BASE stops the ring. One worker serves all the
+                // queues, so stop all of them. The frontend has no way to
+                // learn about a failure other than waiting forever for the
+                // reply, so reply anyway.
+                if let Err(e) = self.stop_worker() {
+                    log::error!("{name}: queue-{index}: failed to stop: {e:?}");
+                }
                 virtq_base.val = 0;
                 self.session.reply(req, &virtq_base, &[])?;
                 msg.flag.set_need_reply(false);
@@ -394,7 +410,7 @@ impl VuBackend {
                     return error::MissingFd { req: msg.request }.fail();
                 };
                 let region = &single.region;
-                if self.status.contains(DevStatus::DRIVER_OK) {
+                if self.running {
                     return error::ChangeMemoryLayout.fail();
                 }
                 log::debug!("{name}: add mem: {region:x?}, fd: {}", fd.as_raw_fd());
@@ -410,7 +426,7 @@ impl VuBackend {
             (VuFrontMsg::REM_MEM_REG, 40) => {
                 let single: MemorySingleRegion = self.session.recv_payload()?;
                 let region = &single.region;
-                if self.status.contains(DevStatus::DRIVER_OK) {
+                if self.running {
                     return error::ChangeMemoryLayout.fail();
                 }
                 for (index, r) in self.init.regions.iter().enumerate() {
@@ -423,6 +439,10 @@ impl VuBackend {
                 }
             }
             (VuFrontMsg::GET_STATUS, 0) => {
+                if self.running && self.dev.ctl.failed() {
+                    // VirtIO spec 1.4, Sec 2.1.2
+                    self.status |= DevStatus::NEEDS_RESET;
+                }
                 let status = self.status.bits() as u64;
                 self.session.reply(req, &status, &[])?;
                 msg.flag.set_need_reply(false);
@@ -431,18 +451,15 @@ impl VuBackend {
             (VuFrontMsg::SET_STATUS, 8) => {
                 let status: u64 = self.session.recv_payload()?;
                 let new = DevStatus::from_bits_retain(status as u8);
-                let old = self.status;
-                self.status = new;
-                log::debug!("{name}: set status: {old:x?} -> {new:x?}");
-                if (old ^ new).contains(DevStatus::DRIVER_OK) {
-                    let event = if new.contains(DevStatus::DRIVER_OK) {
-                        let param = self.parse_init()?;
-                        WakeEvent::Start { param }
-                    } else {
-                        WakeEvent::Reset
-                    };
-                    self.wake_up_dev(event);
+                log::debug!("{name}: set status: {:x?} -> {new:x?}", self.status);
+                if !new.contains(DevStatus::DRIVER_OK) {
+                    self.stop_worker()?;
+                } else if !self.running {
+                    let param = self.parse_init()?;
+                    self.dev.ctl.wake(WakeEvent::Start { param });
+                    self.running = true;
                 }
+                self.status = new;
             }
             (VuFrontMsg::GET_CONFIG, 12..) => {
                 let mut region = [0u8; MAX_CONFIG_SIZE];
@@ -508,3 +525,7 @@ impl VuBackend {
         Ok(())
     }
 }
+
+#[cfg(test)]
+#[path = "backend_test.rs"]
+mod tests;
diff --git a/alioth/src/virtio/vu/backend_test.rs b/alioth/src/virtio/vu/backend_test.rs
new file mode 100644
index 00000000..212c7165
--- /dev/null
+++ b/alioth/src/virtio/vu/backend_test.rs
@@ -0,0 +1,175 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::os::fd::AsFd;
+use std::os::unix::net::UnixStream;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
+use std::thread::{self, JoinHandle};
+use std::time::{Duration, Instant};
+
+use flume::{Receiver, Sender};
+
+use crate::mem;
+use crate::mem::emulated::{Action, Mmio};
+use crate::mem::mapped::RamBus;
+use crate::sync::notifier::Notifier;
+use crate::virtio::dev::{VirtioDevice, WakeEvent, WorkerCtl, WorkerReport};
+use crate::virtio::queue::QueueReg;
+use crate::virtio::vu::backend::{VuBackend, VuIrqSender};
+use crate::virtio::vu::bindings::VirtqState;
+use crate::virtio::vu::conn::VuSession;
+use crate::virtio::{DevStatus, DeviceId};
+
+#[derive(Debug)]
+struct TestDevConfig;
+
+impl Mmio for TestDevConfig {
+    fn size(&self) -> u64 {
+        0
+    }
+
+    fn read(&self, _offset: u64, _size: u8) -> mem::Result<u64> {
+        Ok(0)
+    }
+
+    fn write(&self, _offset: u64, _size: u8, _val: u64) -> mem::Result<Action> {
+        Ok(Action::None)
+    }
+}
+
+type TestWakeReceiver = Receiver<WakeEvent<VuIrqSender>>;
+
+/// Serves a device with one queue, whose worker is `worker`, on a backend
+/// thread. Returns the frontend end of the connection and the thread.
+///
+/// The thread joins the worker after the frontend disconnects.
+fn spawn_backend<W>(worker: W) -> (VuSession, JoinHandle<()>)
+where
+    W: FnOnce(TestWakeReceiver, Sender<WorkerReport>) + Send + 'static,
+{
+    let (event_tx, event_rx) = flume::unbounded();
+    let (report_tx, report_rx) = flume::unbounded();
+    let notifier = Arc::new(Notifier::new().unwrap());
+    let dev = VirtioDevice {
+        name: "vu-backend".into(),
+        id: DeviceId::NET,
+        device_config: Arc::new(TestDevConfig),
+        device_feature: 0,
+        queue_regs: Arc::new([QueueReg::default()]),
+        shared_mem_regions: None,
+        ctl: Arc::new(WorkerCtl::new(
+            "vu-backend".into(),
+            event_tx,
+            notifier,
+            report_rx,
+        )),
+        worker_handle: None,
+    };
+    let worker = thread::spawn(move || worker(event_rx, report_tx));
+
+    let (frontend_conn, backend_conn) = UnixStream::pair().unwrap();
+    let mut backend = VuBackend::new(backend_conn, dev, Arc::new(RamBus::new())).unwrap();
+    let backend_thread = thread::spawn(move || {
+        backend.run().unwrap();
+        // Dropping the backend disconnects the worker.
+        drop(backend);
+        worker.join().unwrap();
+    });
+    (VuSession::from(frontend_conn), backend_thread)
+}
+
+fn driver_ok() -> u64 {
+    let status = DevStatus::ACK | DevStatus::DRIVER | DevStatus::FEATURES_OK | DevStatus::DRIVER_OK;
+    status.bits() as u64
+}
+
+/// Starts the device. The backend takes the kick fd for each start.
+fn start(frontend: &VuSession) {
+    let kick = Notifier::new().unwrap();
+    frontend.set_virtq_kick(&0, kick.as_fd()).unwrap();
+    frontend.set_status(&driver_ok()).unwrap();
+}
+
+#[test]
+fn test_vu_backend_stops_worker_before_reply() {
+    // A fake worker that takes a while to stop.
+    let starts = Arc::new(AtomicU32::new(0));
+    let stopped = Arc::new(AtomicBool::new(false));
+    let (worker_starts, worker_stopped) = (starts.clone(), stopped.clone());
+    let (frontend, backend_thread) = spawn_backend(move |event_rx, report_tx| {
+        for event in event_rx.iter() {
+            match event {
+                WakeEvent::Start { .. } => {
+                    worker_starts.fetch_add(1, Ordering::AcqRel);
+                }
+                WakeEvent::Reset { seq } => {
+                    thread::sleep(Duration::from_millis(50));
+                    worker_stopped.store(true, Ordering::Release);
+                    report_tx.send(WorkerReport::ResetDone { seq }).unwrap();
+                }
+                event => panic!("unexpected event: {event:?}"),
+            }
+        }
+    });
+
+    // The reply to SET_STATUS(0) means that the worker has stopped.
+    start(&frontend);
+    frontend.set_status(&0).unwrap();
+    assert!(stopped.swap(false, Ordering::AcqRel));
+
+    // So does the reply to GET_VRING_BASE.
+    start(&frontend);
+    let virtq_base = VirtqState { index: 0, val: 0 };
+    assert_eq!(frontend.get_virtq_base(&virtq_base).unwrap().val, 0);
+    assert!(stopped.swap(false, Ordering::AcqRel));
+
+    // The worker has already stopped, so there is nothing left to stop.
+    frontend.set_status(&0).unwrap();
+    assert!(!stopped.load(Ordering::Acquire));
+
+    drop(frontend);
+    backend_thread.join().unwrap();
+    assert_eq!(starts.load(Ordering::Acquire), 2);
+}
+
+#[test]
+fn test_vu_backend_needs_reset_after_worker_failure() {
+    // A fake worker that fails as soon as it starts.
+    let (frontend, backend_thread) = spawn_backend(move |event_rx, report_tx| {
+        for event in event_rx.iter() {
+            let report = match event {
+                WakeEvent::Start { .. } => WorkerReport::Failed,
+                WakeEvent::Reset { seq } => WorkerReport::ResetDone { seq },
+                event => panic!("unexpected event: {event:?}"),
+            };
+            report_tx.send(report).unwrap();
+        }
+    });
+
+    start(&frontend);
+    let needs_reset = driver_ok() | DevStatus::NEEDS_RESET.bits() as u64;
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while frontend.get_status().unwrap() != needs_reset {
+        assert!(Instant::now() < deadline, "no DEVICE_NEEDS_RESET");
+        thread::sleep(Duration::from_millis(1));
+    }
+
+    // Resetting the device brings it back into service.
+    frontend.set_status(&0).unwrap();
+    assert_eq!(frontend.get_status().unwrap(), 0);
+
+    drop(frontend);
+    backend_thread.join().unwrap();
+}
diff --git a/alioth/src/virtio/vu/frontend.rs b/alioth/src/virtio/vu/frontend.rs
index a4f2de46..67018fed 100644
--- a/alioth/src/virtio/vu/frontend.rs
+++ b/alioth/src/virtio/vu/frontend.rs
@@ -410,6 +410,17 @@ impl VirtioMio for VuFrontend {
     fn reset(&mut self, registry: &Registry) {
         let name = &*self.name;
         for index in 0..self.num_queues {
+            // A disabled ring is still processed. Only GET_VRING_BASE stops
+            // it, and once all rings are stopped, the backend no longer
+            // touches guest memory or sends notifications.
+            let virtq_base = VirtqState {
+                index: index as _,
+                val: 0,
+            };
+            match self.session.get_virtq_base(&virtq_base) {
+                Ok(base) => log::trace!("{name}: queue-{index}: stopped at {}", base.val),
+                Err(e) => log::error!("{name}: failed to stop queue-{index}: {e:?}"),
+            }
             let disable = VirtqState {
                 index: index as _,
                 val: 0,
@@ -453,3 +464,7 @@ impl DevSpec for VuFrontendSpec {
         true
     }
 }
+
+#[cfg(test)]
+#[path = "frontend_test.rs"]
+mod tests;
diff --git a/alioth/src/virtio/vu/frontend_test.rs b/alioth/src/virtio/vu/frontend_test.rs
new file mode 100644
index 00000000..ccd96314
--- /dev/null
+++ b/alioth/src/virtio/vu/frontend_test.rs
@@ -0,0 +1,86 @@
+// Copyright 2026 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::os::unix::net::UnixStream;
+use std::sync::Arc;
+use std::thread;
+
+use mio::Poll;
+
+use crate::virtio::DeviceId;
+use crate::virtio::vu::bindings::{VirtqState, VuFeature, VuFrontMsg};
+use crate::virtio::vu::conn::VuSession;
+use crate::virtio::vu::frontend::VuFrontend;
+use crate::virtio::worker::mio::VirtioMio;
+
+#[test]
+fn test_vu_frontend_reset_stops_rings() {
+    let (frontend_conn, backend_conn) = UnixStream::pair().unwrap();
+    let backend = VuSession::from(backend_conn);
+    let backend_thread = thread::spawn(move || {
+        let mut fds = [const { None }; 8];
+        let mut requests = vec![];
+        while let Ok(msg) = backend.recv_msg(&mut fds) {
+            let req = VuFrontMsg::from(msg.request);
+            match req {
+                VuFrontMsg::GET_VIRTQ_BASE => {
+                    let mut virtq_base: VirtqState = backend.recv_payload().unwrap();
+                    requests.push((req, virtq_base.index as u64));
+                    virtq_base.val = 0;
+                    backend.reply(req, &virtq_base, &[]).unwrap();
+                }
+                VuFrontMsg::SET_VIRTQ_ENABLE => {
+                    let virtq_enable: VirtqState = backend.recv_payload().unwrap();
+                    assert_eq!(virtq_enable.val, 0);
+                    requests.push((req, virtq_enable.index as u64));
+                    backend.reply(req, &0u64, &[]).unwrap();
+                }
+                VuFrontMsg::SET_STATUS => {
+                    let status: u64 = backend.recv_payload().unwrap();
+                    requests.push((req, status));
+                    backend.reply(req, &0u64, &[]).unwrap();
+                }
+                req => panic!("unexpected request: {req:?}"),
+            }
+        }
+        requests
+    });
+
+    let mut frontend = VuFrontend {
+        name: "vu-frontend".into(),
+        session: Arc::new(VuSession::from(frontend_conn)),
+        channel: None,
+        id: DeviceId::NET,
+        vu_feature: VuFeature::STATUS,
+        device_feature: 0,
+        num_queues: 2,
+        err_fds: Box::default(),
+    };
+    let poll = Poll::new().unwrap();
+    frontend.reset(poll.registry());
+    drop(frontend);
+
+    // Every ring is stopped before the device is reset.
+    let requests = backend_thread.join().unwrap();
+    assert_eq!(
+        requests,
+        [
+            (VuFrontMsg::GET_VIRTQ_BASE, 0),
+            (VuFrontMsg::SET_VIRTQ_ENABLE, 0),
+            (VuFrontMsg::GET_VIRTQ_BASE, 1),
+            (VuFrontMsg::SET_VIRTQ_ENABLE, 1),
+            (VuFrontMsg::SET_STATUS, 0),
+        ]
+    );
+}
diff --git a/alioth/src/virtio/worker/io_uring_test.rs b/alioth/src/virtio/worker/io_uring_test.rs
index 168afec5..694626e5 100644
--- a/alioth/src/virtio/worker/io_uring_test.rs
+++ b/alioth/src/virtio/worker/io_uring_test.rs
@@ -138,6 +138,7 @@ fn io_uring_drain_test() {
     let (handle, notifier) = dev
         .spawn_worker(WorkerParam {
             event_rx: rx,
+            report_tx: flume::unbounded().0,
             memory: ram_bus.clone(),
             queue_regs: regs.clone(),
         })