block: qcow: Preserve WRITE_ZEROES with backing files

QCOW2 empty L2 entries in an overlay mean that reads fall
through to the backing file. Reusing the punch_hole path for
WRITE_ZEROES therefore turns a full-cluster zero operation on an
unallocated overlay cluster into backing data exposure.

Keep discard/punch_hole behavior unchanged, but let WRITE_ZEROES
request a logical-zero marker when the image has a backing file.
ZERO_FLAG entries now read as zeros in both the legacy QcowFile
path and the shared runtime metadata path. Partial writes after
such entries seed new clusters from zeros instead of backing data.

Treat ZERO_FLAG entries as logical holes for SEEK_HOLE/SEEK_DATA.
Empty overlay entries with a backing file still report data because
the data exists in the backing file.

Avoid cluster-sized userspace zero buffers when materializing
zero-flagged clusters by zeroing the allocated host range directly.
This keeps recycled clusters safe without making partial writes
allocate large zero-filled Vecs.

Add regression coverage for legacy QcowFile, QcowSync, direct I/O,
QcowAsync/io_uring overlay paths, and a large-cluster partial-write
case.

Assisted-by: Codex:GPT-5

Signed-off-by: Ian Klemm <hi@ianklemm.de>

block/src/qcow/metadata.rs

@@ -22,6 +22,7 @@ use std::mem;
 use std::sync::{Arc, RwLock};
 
 use libc::{EINVAL, EIO};
+use vmm_sys_util::write_zeroes::WriteZeroesAt;
 
 use super::decoder::Decoder;
 use super::qcow_raw_file::QcowRawFile;
@@ -29,7 +30,7 @@ use super::refcount::RefCount;
 use super::util::{
     div_round_up_u64, l1_entry_make, l2_entry_compressed_cluster_layout, l2_entry_is_compressed,
     l2_entry_is_empty, l2_entry_is_zero, l2_entry_make_std, l2_entry_make_zero,
-    l2_entry_std_cluster_addr,
+    l2_entry_make_zero_plain, l2_entry_std_cluster_addr,
 };
 use super::vec_cache::{CacheMap, Cacheable, VecCache};
 use super::{QcowHeader, refcount};
@@ -270,15 +271,16 @@ impl QcowMetadata {
     }
 
     /// Deallocates a range of bytes. Full clusters are deallocated via metadata.
-    /// Partial clusters need the caller to write zeros. This method returns a
-    /// list of actions the caller should take.
+    /// If `zero_marker` is true, full-cluster deallocation records a logical
+    /// zero instead of an empty entry where backing data could otherwise be
+    /// exposed. Partial clusters need the caller to write zeros. This method
+    /// returns a list of actions the caller should take.
     pub(crate) fn deallocate_bytes(
         &self,
         address: u64,
         length: usize,
         sparse: bool,
-        virtual_size: u64,
-        cluster_size: u64,
+        zero_marker: bool,
         backing_file: Option<&dyn BackingRead>,
     ) -> io::Result<Vec<DeallocAction>> {
         if address.checked_add(length as u64).is_none() {
@@ -287,7 +289,8 @@ impl QcowMetadata {
         let mut inner = self.inner.write().unwrap();
         let mut actions = Vec::new();
 
-        let file_end = virtual_size;
+        let file_end = inner.header.size;
+        let cluster_size = inner.raw_file.cluster_size();
         let remaining_in_file = file_end.saturating_sub(address);
         let write_count = min(length as u64, remaining_in_file) as usize;
 
@@ -301,7 +304,11 @@ impl QcowMetadata {
             );
 
             if count == cluster_size as usize {
-                let punch_offset = inner.deallocate_cluster(curr_addr, sparse)?;
+                let punch_offset = inner.deallocate_cluster(
+                    curr_addr,
+                    sparse,
+                    zero_marker && backing_file.is_some(),
+                )?;
                 if let Some(host_offset) = punch_offset {
                     actions.push(DeallocAction::PunchHole {
                         host_offset,
@@ -406,14 +413,9 @@ impl QcowState {
                 has_backing_file,
             )))
         } else if l2_entry_is_zero(l2_entry) {
-            // Match original QcowFile::file_read semantics where zero flagged
-            // entries fall through to backing file when one exists or return
-            // zeros otherwise.
-            Ok(Some(self.unallocated_read_mapping(
-                address,
-                count,
-                has_backing_file,
-            )))
+            Ok(Some(ClusterReadMapping::Zero {
+                length: count as u64,
+            }))
         } else {
             let cluster_addr = l2_entry_std_cluster_addr(l2_entry);
             let cluster_size = self.raw_file.cluster_size();
@@ -469,10 +471,9 @@ impl QcowState {
                 length: count,
             })
         } else if l2_entry_is_zero(l2_entry) {
-            // Match original QcowFile::file_read semantics where zero flagged
-            // entries fall through to backing file when one exists or return
-            // zeros otherwise.
-            Ok(self.unallocated_read_mapping(address, count, has_backing_file))
+            Ok(ClusterReadMapping::Zero {
+                length: count as u64,
+            })
         } else {
             let cluster_addr = l2_entry_std_cluster_addr(l2_entry);
             let cluster_size = self.raw_file.cluster_size();
@@ -558,7 +559,11 @@ impl QcowState {
             self.deallocate_compressed_cluster(l2_entry)?;
             cluster_addr
         } else if l2_entry_is_empty(l2_entry) || l2_entry_is_zero(l2_entry) {
-            let cluster_addr = self.append_data_cluster(backing_data)?;
+            let cluster_addr = if l2_entry_is_zero(l2_entry) {
+                self.append_zeroed_data_cluster()?
+            } else {
+                self.append_data_cluster(backing_data)?
+            };
             self.update_cluster_addr(l1_index, l2_index, cluster_addr, &mut set_refcounts)?;
             cluster_addr
         } else {
@@ -683,6 +688,17 @@ impl QcowState {
         Ok(new_addr)
     }
 
+    /// Allocates a data cluster and zeroes it without building a cluster-sized buffer.
+    fn append_zeroed_data_cluster(&mut self) -> io::Result<u64> {
+        let new_addr = self.get_new_cluster(None)?;
+        let cluster_size = self.raw_file.cluster_size() as usize;
+        self.raw_file
+            .file_mut()
+            .write_zeroes_at(new_addr, cluster_size)?;
+        self.set_cluster_refcount_track_freed(new_addr, 1)?;
+        Ok(new_addr)
+    }
+
     /// Updates the L1 and L2 tables to point to a new cluster address.
     fn update_cluster_addr(
         &mut self,
@@ -821,13 +837,15 @@ impl QcowState {
     /// If sparse is true, fully deallocates and returns the host offset if
     /// the underlying storage should be punched after the refcount dropped
     /// to zero. If sparse is false, uses the zero flag optimization when
-    /// possible.
+    /// possible. If `zero_marker` is true, empty entries are replaced with
+    /// logical-zero entries so reads do not fall through to backing data.
     ///
     /// Returns None if no host punch_hole is needed.
     pub(super) fn deallocate_cluster(
         &mut self,
         address: u64,
         sparse: bool,
+        zero_marker: bool,
     ) -> io::Result<Option<u64>> {
         if address >= self.header.size {
             return Err(io::Error::from_raw_os_error(EINVAL));
@@ -839,21 +857,38 @@ impl QcowState {
             None => return Err(io::Error::from_raw_os_error(EINVAL)),
         };
         let l2_index = self.l2_table_index(address) as usize;
+        let dealloc_entry = if zero_marker {
+            l2_entry_make_zero_plain()
+        } else {
+            0
+        };
 
         if l2_addr_disk == 0 {
+            if zero_marker {
+                if let Some(new_addr) = self.cache_l2_cluster_alloc(l1_index, l2_addr_disk)? {
+                    self.set_cluster_refcount_track_freed(new_addr, 1)?;
+                }
+                self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = dealloc_entry;
+            }
             return Ok(None);
         }
 
         self.cache_l2_cluster(l1_index, l2_addr_disk)?;
 
         let l2_entry = self.l2_cache.get(l1_index).unwrap()[l2_index];
-        if l2_entry_is_empty(l2_entry) || l2_entry_is_zero(l2_entry) {
+        if l2_entry_is_empty(l2_entry) {
+            if zero_marker {
+                self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = dealloc_entry;
+            }
+            return Ok(None);
+        }
+        if l2_entry_is_zero(l2_entry) {
             return Ok(None);
         }
 
         if l2_entry_is_compressed(l2_entry) {
             self.deallocate_compressed_cluster(l2_entry)?;
-            self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = 0;
+            self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = dealloc_entry;
             return Ok(None);
         }
 
@@ -877,7 +912,7 @@ impl QcowState {
         if sparse {
             let new_refcount = refcount - 1;
             self.set_cluster_refcount_track_freed(cluster_addr, new_refcount)?;
-            self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = 0;
+            self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = dealloc_entry;
             if new_refcount == 0 {
                 self.unref_clusters.push(cluster_addr);
                 return Ok(Some(cluster_addr));
@@ -886,7 +921,7 @@ impl QcowState {
             self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = l2_entry_make_zero(cluster_addr);
         } else {
             self.set_cluster_refcount_track_freed(cluster_addr, refcount - 1)?;
-            self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = 0;
+            self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = dealloc_entry;
         }
         Ok(None)
     }

block/src/qcow/mod.rs

@@ -46,7 +46,8 @@ pub(crate) use util::MAX_NESTING_DEPTH;
 use util::{
     L1_TABLE_OFFSET_MASK, L2_TABLE_OFFSET_MASK, div_round_up_u32, div_round_up_u64, l1_entry_make,
     l2_entry_compressed_cluster_layout, l2_entry_is_compressed, l2_entry_is_empty,
-    l2_entry_is_zero, l2_entry_make_std, l2_entry_make_zero, l2_entry_std_cluster_addr,
+    l2_entry_is_zero, l2_entry_make_std, l2_entry_make_zero, l2_entry_make_zero_plain,
+    l2_entry_std_cluster_addr,
 };
 use vmm_sys_util::file_traits::{FileSetLen, FileSync};
 use vmm_sys_util::seek_hole::SeekHole;
@@ -1479,7 +1480,8 @@ impl QcowFile {
             buf[..count].copy_from_slice(&decompressed_cluster[start..end.unwrap()]);
         } else if l2_entry_is_zero(l2_entry) {
             // Cluster with zero flag reads as zeros without accessing disk.
-            return Ok(None);
+            buf[..count].fill(0);
+            return Ok(Some(()));
         } else {
             let cluster_addr = l2_entry_std_cluster_addr(l2_entry);
             if cluster_addr & (self.raw_file.cluster_size() - 1) != 0 {
@@ -1538,17 +1540,21 @@ impl QcowFile {
 
             cluster_addr
         } else if l2_entry_is_empty(l2_entry) || l2_entry_is_zero(l2_entry) {
-            let initial_data = if let Some(backing) = self.backing_file.as_mut() {
-                let cluster_size = self.raw_file.cluster_size();
-                let cluster_begin = address - (address % cluster_size);
-                let mut cluster_data = vec![0u8; cluster_size as usize];
-                backing.read_at(cluster_begin, &mut cluster_data)?;
-                Some(cluster_data)
+            let cluster_addr = if l2_entry_is_zero(l2_entry) {
+                self.append_zeroed_data_cluster()?
             } else {
-                None
+                let initial_data = if let Some(backing) = self.backing_file.as_mut() {
+                    let cluster_size = self.raw_file.cluster_size();
+                    let cluster_begin = address - (address % cluster_size);
+                    let mut cluster_data = vec![0u8; cluster_size as usize];
+                    backing.read_at(cluster_begin, &mut cluster_data)?;
+                    Some(cluster_data)
+                } else {
+                    None
+                };
+                self.append_data_cluster(initial_data)?
             };
             // Need to allocate a data cluster
-            let cluster_addr = self.append_data_cluster(initial_data)?;
             self.update_cluster_addr(l1_index, l2_index, cluster_addr, &mut set_refcounts)?;
             cluster_addr
         } else {
@@ -1641,6 +1647,18 @@ impl QcowFile {
         Ok(new_addr)
     }
 
+    // Allocate and initialize a zeroed data cluster without building a cluster-sized buffer.
+    fn append_zeroed_data_cluster(&mut self) -> std::io::Result<u64> {
+        let new_addr: u64 = self.get_new_cluster(None)?;
+        let cluster_size = self.raw_file.cluster_size() as usize;
+        self.raw_file
+            .file_mut()
+            .write_zeroes_at(new_addr, cluster_size)?;
+        // The cluster refcount starts at one indicating it is used but doesn't need COW.
+        self.set_cluster_refcount_track_freed(new_addr, 1)?;
+        Ok(new_addr)
+    }
+
     // Returns true if the cluster containing `address` is already allocated.
     fn cluster_allocated(&mut self, address: u64) -> std::io::Result<bool> {
         if address >= self.virtual_size() {
@@ -1655,16 +1673,25 @@ impl QcowFile {
         let l2_index = self.l2_table_index(address) as usize;
 
         if l2_addr_disk == 0 {
-            // The whole L2 table for this address is not allocated yet,
-            // so the cluster must also be unallocated.
-            return Ok(false);
+            // Empty overlay metadata means "consult backing" when a backing
+            // file exists; otherwise it is a hole in this image.
+            return Ok(self.backing_file.is_some());
         }
 
         self.cache_l2_cluster(l1_index, l2_addr_disk, false)?;
 
-        let cluster_addr = self.l2_cache.get(l1_index).unwrap()[l2_index];
-        // If cluster_addr != 0, the cluster is allocated.
-        Ok(cluster_addr != 0)
+        let l2_entry = self.l2_cache.get(l1_index).unwrap()[l2_index];
+        if l2_entry_is_empty(l2_entry) {
+            // Empty cluster with backing has existing data to seek in the backing file.
+            Ok(self.backing_file.is_some())
+        } else if l2_entry_is_compressed(l2_entry) {
+            Ok(true)
+        } else if l2_entry_is_zero(l2_entry) {
+            // Zero flagged cluster is a logical hole. It reads as zeros with no data to seek.
+            Ok(false)
+        } else {
+            Ok(true)
+        }
     }
 
     // Find the first guest address greater than or equal to `address` whose allocation state
@@ -1737,8 +1764,9 @@ impl QcowFile {
     }
 
     // Deallocate the storage for the cluster starting at `address`.
-    // Any future reads of this cluster will return all zeroes.
-    fn deallocate_cluster(&mut self, address: u64) -> std::io::Result<()> {
+    // If `zero_marker` is true, preserve WRITE_ZEROES semantics with a logical-zero
+    // entry instead of allowing backing data to reappear through an empty entry.
+    fn deallocate_cluster(&mut self, address: u64, zero_marker: bool) -> std::io::Result<()> {
         if address >= self.virtual_size() {
             return Err(std::io::Error::from_raw_os_error(EINVAL));
         }
@@ -1749,25 +1777,44 @@ impl QcowFile {
             .get(l1_index)
             .ok_or_else(|| std::io::Error::from_raw_os_error(EINVAL))?;
         let l2_index = self.l2_table_index(address) as usize;
+        let write_zero_marker = zero_marker && self.backing_file.is_some();
+        let dealloc_entry = if write_zero_marker {
+            l2_entry_make_zero_plain()
+        } else {
+            0
+        };
 
         if l2_addr_disk == 0 {
-            // The whole L2 table for this address is not allocated yet,
-            // so the cluster must also be unallocated.
+            // With a backing file, an empty L2 entry means "consult backing".
+            // WRITE_ZEROES needs a logical-zero marker instead.
+            if write_zero_marker {
+                if let Some(new_addr) = self.cache_l2_cluster(l1_index, l2_addr_disk, true)? {
+                    self.set_cluster_refcount_track_freed(new_addr, 1)?;
+                }
+                self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = dealloc_entry;
+            }
             return Ok(());
         }
 
         self.cache_l2_cluster(l1_index, l2_addr_disk, false)?;
 
         let l2_entry = self.l2_cache.get(l1_index).unwrap()[l2_index];
-        if l2_entry_is_empty(l2_entry) || l2_entry_is_zero(l2_entry) {
-            // Already unallocated or zero.
+        if l2_entry_is_empty(l2_entry) {
+            // With a backing file, empty means "consult backing"; preserve
+            // WRITE_ZEROES semantics with an explicit zero marker.
+            if write_zero_marker {
+                self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = dealloc_entry;
+            }
+            return Ok(());
+        }
+        if l2_entry_is_zero(l2_entry) {
             return Ok(());
         }
 
         // Compressed clusters cannot use the zero flag optimization, thus fully deallocate instead.
         if l2_entry_is_compressed(l2_entry) {
             self.deallocate_compressed_cluster(l2_entry)?;
-            self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = 0;
+            self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = dealloc_entry;
             return Ok(());
         }
 
@@ -1796,7 +1843,7 @@ impl QcowFile {
             self.set_cluster_refcount_track_freed(cluster_addr, new_refcount)?;
 
             // Rewrite the L2 entry to remove the cluster mapping (full deallocation).
-            self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = 0;
+            self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = dealloc_entry;
 
             if new_refcount == 0 {
                 let cluster_size = self.raw_file.cluster_size();
@@ -1820,15 +1867,27 @@ impl QcowFile {
                 // Multiple references - must decrement refcount and unmap this entry.
                 // Cannot use zero flag because other L2 entries still need the real data.
                 self.set_cluster_refcount_track_freed(cluster_addr, refcount - 1)?;
-                self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = 0;
+                self.l2_cache.get_mut(l1_index).unwrap()[l2_index] = dealloc_entry;
             }
         }
         Ok(())
     }
 
-    // Deallocate the storage for `length` bytes starting at `address`.
-    // Any future reads of this range will return all zeroes.
     fn deallocate_bytes(&mut self, address: u64, length: usize) -> std::io::Result<()> {
+        self.deallocate_bytes_impl(address, length, false)
+    }
+
+    // Apply WRITE_ZEROES semantics for `length` bytes starting at `address`.
+    fn write_zeroes_bytes(&mut self, address: u64, length: usize) -> std::io::Result<()> {
+        self.deallocate_bytes_impl(address, length, true)
+    }
+
+    fn deallocate_bytes_impl(
+        &mut self,
+        address: u64,
+        length: usize,
+        zero_marker: bool,
+    ) -> std::io::Result<()> {
         let write_count: usize = self.limit_range_file(address, length);
 
         let mut nwritten: usize = 0;
@@ -1838,7 +1897,7 @@ impl QcowFile {
 
             if count == self.raw_file.cluster_size() as usize {
                 // Full cluster - deallocate the storage.
-                self.deallocate_cluster(curr_addr)?;
+                self.deallocate_cluster(curr_addr, zero_marker)?;
             } else {
                 // Partial cluster - zero out the relevant bytes if it was allocated.
                 // Any space in unallocated clusters can be left alone, since
@@ -2159,7 +2218,7 @@ impl PunchHole for QcowFile {
 
 impl WriteZeroesAt for QcowFile {
     fn write_zeroes_at(&mut self, offset: u64, length: usize) -> io::Result<usize> {
-        self.punch_hole(offset, length as u64)?;
+        self.write_zeroes_bytes(offset, length)?;
         Ok(length)
     }
 }
@@ -2459,6 +2518,61 @@ mod unit_tests {
         testfn(qcow_file); // File closed when the function exits.
     }
 
+    fn header_with_cluster_bits(cluster_bits: u32, size: u64) -> QcowHeader {
+        let mut header = QcowHeader::create_for_size_and_path(3, size, None).unwrap();
+        let cluster_size = 1u32 << cluster_bits;
+        let entries_per_cluster = cluster_size / std::mem::size_of::<u64>() as u32;
+        let num_clusters = div_round_up_u64(size, u64::from(cluster_size)) as u32;
+        let num_l2_clusters = div_round_up_u32(num_clusters, entries_per_cluster);
+        let l1_clusters = div_round_up_u32(num_l2_clusters, entries_per_cluster);
+        let header_clusters =
+            div_round_up_u32(std::mem::size_of::<QcowHeader>() as u32, cluster_size);
+        let max_refcount_clusters = max_refcount_clusters(
+            DEFAULT_REFCOUNT_ORDER,
+            cluster_size,
+            num_clusters + l1_clusters + num_l2_clusters + header_clusters,
+        ) as u32;
+
+        header.cluster_bits = cluster_bits;
+        header.l1_size = num_l2_clusters;
+        header.l1_table_offset = u64::from(cluster_size);
+        header.refcount_table_offset = u64::from(cluster_size * (l1_clusters + 1));
+        header.refcount_table_clusters = div_round_up_u32(
+            max_refcount_clusters * std::mem::size_of::<u64>() as u32,
+            cluster_size,
+        );
+        header
+    }
+
+    fn qcow_file_with_cluster_bits(cluster_bits: u32, size: u64) -> QcowFile {
+        let disk_file = RawFile::new(TempFile::new().unwrap().into_file(), false);
+        let header = header_with_cluster_bits(cluster_bits, size);
+        QcowFile::new_from_header(disk_file, &header, true).unwrap()
+    }
+
+    fn qcow_overlay_with_backing_pattern(offset: u64, len: usize, value: u8) -> QcowFile {
+        qcow_overlay_with_backing_pattern_and_cluster_bits(offset, len, value, 16)
+    }
+
+    fn qcow_overlay_with_backing_pattern_and_cluster_bits(
+        offset: u64,
+        len: usize,
+        value: u8,
+        cluster_bits: u32,
+    ) -> QcowFile {
+        let cluster_size = 1u64 << cluster_bits;
+        let size = (offset + len as u64).max(cluster_size * 4);
+        let mut backing = qcow_file_with_cluster_bits(cluster_bits, size);
+        let backing_data = vec![value; len];
+        backing.seek(SeekFrom::Start(offset)).unwrap();
+        backing.write_all(&backing_data).unwrap();
+        backing.flush().unwrap();
+
+        let mut overlay = qcow_file_with_cluster_bits(cluster_bits, size);
+        overlay.set_backing_file(Some(Box::new(backing)));
+        overlay
+    }
+
     #[test]
     fn write_read_start_backing_v2() {
         let disk_file = basic_file(&valid_header_v2());
@@ -4606,6 +4720,90 @@ mod unit_tests {
         });
     }
 
+    #[test]
+    fn write_zeroes_unallocated_overlay_with_backing_must_read_zero() {
+        const CLUSTER_SIZE: usize = 0x10000;
+        const OFFSET: u64 = 0x10000;
+
+        let mut overlay = qcow_overlay_with_backing_pattern(OFFSET, CLUSTER_SIZE, 0xAB);
+        let nwritten = overlay
+            .write_zeroes_at(OFFSET, CLUSTER_SIZE)
+            .expect("failed to zero unallocated overlay cluster");
+        assert_eq!(nwritten, CLUSTER_SIZE);
+
+        let mut buf = [0xFFu8; CLUSTER_SIZE];
+        overlay.seek(SeekFrom::Start(OFFSET)).unwrap();
+        overlay.read_exact(&mut buf).unwrap();
+        assert!(
+            buf.iter().all(|&b| b == 0),
+            "zeroed unallocated overlay cluster exposed backing data"
+        );
+    }
+
+    #[test]
+    fn partial_write_after_write_zeroes_must_not_reintroduce_backing_data() {
+        const CLUSTER_SIZE: usize = 0x10000;
+        const OFFSET: u64 = 0x10000;
+        const PATCH_OFFSET: usize = 0x4000;
+        const PATCH_LEN: usize = 0x1000;
+
+        let mut overlay = qcow_overlay_with_backing_pattern(OFFSET, CLUSTER_SIZE, 0xAB);
+        overlay.write_zeroes_at(OFFSET, CLUSTER_SIZE).unwrap();
+
+        let patch = [0x99u8; PATCH_LEN];
+        overlay
+            .seek(SeekFrom::Start(OFFSET + PATCH_OFFSET as u64))
+            .unwrap();
+        overlay.write_all(&patch).unwrap();
+
+        let mut buf = [0xFFu8; CLUSTER_SIZE];
+        overlay.seek(SeekFrom::Start(OFFSET)).unwrap();
+        overlay.read_exact(&mut buf).unwrap();
+
+        assert!(buf[..PATCH_OFFSET].iter().all(|&b| b == 0));
+        assert!(
+            buf[PATCH_OFFSET..PATCH_OFFSET + PATCH_LEN]
+                .iter()
+                .all(|&b| b == 0x99)
+        );
+        assert!(buf[PATCH_OFFSET + PATCH_LEN..].iter().all(|&b| b == 0));
+    }
+
+    #[test]
+    fn partial_write_after_write_zeroes_large_cluster_must_not_reintroduce_backing_data() {
+        const CLUSTER_BITS: u32 = 20;
+        const CLUSTER_SIZE: usize = 1 << CLUSTER_BITS;
+        const OFFSET: u64 = CLUSTER_SIZE as u64;
+        const PATCH_OFFSET: usize = 0x4000;
+        const PATCH_LEN: usize = 0x1000;
+
+        let mut overlay = qcow_overlay_with_backing_pattern_and_cluster_bits(
+            OFFSET,
+            CLUSTER_SIZE,
+            0xAB,
+            CLUSTER_BITS,
+        );
+        overlay.write_zeroes_at(OFFSET, CLUSTER_SIZE).unwrap();
+
+        let patch = [0x99u8; PATCH_LEN];
+        overlay
+            .seek(SeekFrom::Start(OFFSET + PATCH_OFFSET as u64))
+            .unwrap();
+        overlay.write_all(&patch).unwrap();
+
+        let mut buf = vec![0xFFu8; CLUSTER_SIZE];
+        overlay.seek(SeekFrom::Start(OFFSET)).unwrap();
+        overlay.read_exact(&mut buf).unwrap();
+
+        assert!(buf[..PATCH_OFFSET].iter().all(|&b| b == 0));
+        assert!(
+            buf[PATCH_OFFSET..PATCH_OFFSET + PATCH_LEN]
+                .iter()
+                .all(|&b| b == 0x99)
+        );
+        assert!(buf[PATCH_OFFSET + PATCH_LEN..].iter().all(|&b| b == 0));
+    }
+
     #[test]
     fn test_compressed_read() {
         let cluster_size = 65536usize;

block/src/qcow/util.rs

@@ -63,6 +63,11 @@ pub(super) fn l2_entry_make_zero(cluster_addr: u64) -> u64 {
     (cluster_addr & L2_TABLE_OFFSET_MASK) | CLUSTER_USED_FLAG | ZERO_FLAG
 }
 
+/// Make L2 entry for an unallocated cluster that reads as logical zeros.
+pub(super) fn l2_entry_make_zero_plain() -> u64 {
+    ZERO_FLAG
+}
+
 /// Make L1 entry with optional flags.
 pub(super) fn l1_entry_make(cluster_addr: u64, refcount_is_one: bool) -> u64 {
     (cluster_addr & L1_TABLE_OFFSET_MASK) | (refcount_is_one as u64 * CLUSTER_USED_FLAG)

block/src/qcow_async.rs

@@ -200,17 +200,13 @@ impl AsyncIo for QcowAsync {
     }
 
     fn punch_hole(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()> {
-        let virtual_size = self.metadata.virtual_size();
-        let cluster_size = self.cluster_size;
-
         let result = self
             .metadata
             .deallocate_bytes(
                 offset,
                 length as usize,
                 self.sparse,
-                virtual_size,
-                cluster_size,
+                false,
                 self.backing_file.as_deref(),
             )
             .map_err(AsyncIoError::PunchHole);
@@ -238,9 +234,37 @@ impl AsyncIo for QcowAsync {
     }
 
     fn write_zeroes(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()> {
-        // For QCOW2, zeroing and hole punching are the same operation.
-        // Both discard guest data so the range reads back as zero.
-        self.punch_hole(offset, length, user_data)
+        let result = self
+            .metadata
+            .deallocate_bytes(
+                offset,
+                length as usize,
+                self.sparse,
+                true,
+                self.backing_file.as_deref(),
+            )
+            .map_err(AsyncIoError::WriteZeroes);
+
+        match result {
+            Ok(actions) => {
+                for action in &actions {
+                    self.apply_dealloc_action(action);
+                }
+                self.completion_list.push_back((user_data, 0));
+                self.eventfd.write(1).unwrap();
+                Ok(())
+            }
+            Err(e) => {
+                let errno = if let AsyncIoError::WriteZeroes(ref io_err) = e {
+                    -io_err.raw_os_error().unwrap_or(libc::EIO)
+                } else {
+                    -libc::EIO
+                };
+                self.completion_list.push_back((user_data, errno));
+                self.eventfd.write(1).unwrap();
+                Ok(())
+            }
+        }
     }
 
     fn batch_requests_enabled(&self) -> bool {
@@ -557,7 +581,7 @@ mod unit_tests {
 
     use super::*;
     use crate::disk_file::AsyncDiskFile;
-    use crate::qcow::{QcowFile, RawFile};
+    use crate::qcow::{BackingFileConfig, ImageType, QcowFile, RawFile};
     use crate::qcow_common::unit_tests::compress_allocated_clusters;
     use crate::qcow_disk::QcowDisk;
     use crate::{BatchRequest, RequestType, SECTOR_SIZE};
@@ -587,6 +611,38 @@ mod unit_tests {
         (temp_file, disk)
     }
 
+    fn create_overlay_disk_with_raw_backing_pattern(
+        file_size: u64,
+        value: u8,
+    ) -> (TempFile, TempFile, QcowDisk) {
+        let backing_temp = TempFile::new().unwrap();
+        let backing_data = vec![value; file_size as usize];
+        backing_temp.as_file().write_all(&backing_data).unwrap();
+        backing_temp.as_file().sync_all().unwrap();
+        let backing_path = backing_temp.as_path().to_str().unwrap().to_string();
+
+        let overlay_temp = TempFile::new().unwrap();
+        {
+            let raw = RawFile::new(overlay_temp.as_file().try_clone().unwrap(), false);
+            let backing_config = BackingFileConfig {
+                path: backing_path,
+                format: Some(ImageType::Raw),
+            };
+            QcowFile::new_from_backing(raw, 3, file_size, &backing_config, true).unwrap();
+        }
+
+        let disk = QcowDisk::new(
+            overlay_temp.as_file().try_clone().unwrap(),
+            false,
+            true,
+            true,
+            true,
+        )
+        .unwrap();
+
+        (backing_temp, overlay_temp, disk)
+    }
+
     fn wait_for_completion(async_io: &mut dyn AsyncIo) -> (u64, i32) {
         loop {
             if let Some(c) = async_io.next_completed_request() {
@@ -678,6 +734,28 @@ mod unit_tests {
         );
     }
 
+    #[test]
+    fn test_qcow_async_write_zeroes_unallocated_overlay_with_backing_must_read_zero() {
+        let cluster_size = 1u64 << 16;
+        let file_size = cluster_size * 4;
+        let offset = cluster_size;
+        let (_backing_temp, _overlay_temp, disk) =
+            create_overlay_disk_with_raw_backing_pattern(file_size, 0xAB);
+
+        let mut async_io = disk.create_async_io(1).unwrap();
+        async_io.write_zeroes(offset, cluster_size, 201).unwrap();
+        let (user_data, result) = wait_for_completion(async_io.as_mut());
+        assert_eq!(user_data, 201);
+        assert_eq!(result, 0, "write_zeroes should succeed");
+        drop(async_io);
+
+        let read_buf = async_read(&disk, offset, cluster_size as usize);
+        assert!(
+            read_buf.iter().all(|&b| b == 0),
+            "zeroed unallocated overlay cluster exposed backing data"
+        );
+    }
+
     #[test]
     fn test_qcow_async_write_read_roundtrip() {
         let file_size = 100 * 1024 * 1024;

block/src/qcow_sync.rs

@@ -57,6 +57,26 @@ impl QcowSync {
             completion_list: VecDeque::new(),
         }
     }
+
+    fn apply_dealloc_action(&mut self, action: &DeallocAction) {
+        match action {
+            DeallocAction::PunchHole {
+                host_offset,
+                length,
+            } => {
+                let _ = self.data_file.file_mut().punch_hole(*host_offset, *length);
+            }
+            DeallocAction::WriteZeroes {
+                host_offset,
+                length,
+            } => {
+                let _ = self
+                    .data_file
+                    .file_mut()
+                    .write_zeroes_at(*host_offset, *length);
+            }
+        }
+    }
 }
 
 impl AsyncIo for QcowSync {
@@ -249,41 +269,21 @@ impl AsyncIo for QcowSync {
     }
 
     fn punch_hole(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()> {
-        let virtual_size = self.metadata.virtual_size();
-        let cluster_size = self.cluster_size;
-
         let result = self
             .metadata
             .deallocate_bytes(
                 offset,
                 length as usize,
                 self.sparse,
-                virtual_size,
-                cluster_size,
+                false,
                 self.backing_file.as_deref(),
             )
             .map_err(AsyncIoError::PunchHole);
 
         match result {
             Ok(actions) => {
-                for action in actions {
-                    match action {
-                        DeallocAction::PunchHole {
-                            host_offset,
-                            length,
-                        } => {
-                            let _ = self.data_file.file_mut().punch_hole(host_offset, length);
-                        }
-                        DeallocAction::WriteZeroes {
-                            host_offset,
-                            length,
-                        } => {
-                            let _ = self
-                                .data_file
-                                .file_mut()
-                                .write_zeroes_at(host_offset, length);
-                        }
-                    }
+                for action in &actions {
+                    self.apply_dealloc_action(action);
                 }
                 self.completion_list.push_back((user_data, 0));
                 self.eventfd.write(1).unwrap();
@@ -303,9 +303,37 @@ impl AsyncIo for QcowSync {
     }
 
     fn write_zeroes(&mut self, offset: u64, length: u64, user_data: u64) -> AsyncIoResult<()> {
-        // For QCOW2 write_zeroes uses cluster deallocation, same as punch_hole.
-        // Unallocated clusters inherently read as zero in the QCOW2 format.
-        self.punch_hole(offset, length, user_data)
+        let result = self
+            .metadata
+            .deallocate_bytes(
+                offset,
+                length as usize,
+                self.sparse,
+                true,
+                self.backing_file.as_deref(),
+            )
+            .map_err(AsyncIoError::WriteZeroes);
+
+        match result {
+            Ok(actions) => {
+                for action in &actions {
+                    self.apply_dealloc_action(action);
+                }
+                self.completion_list.push_back((user_data, 0));
+                self.eventfd.write(1).unwrap();
+                Ok(())
+            }
+            Err(e) => {
+                let errno = if let AsyncIoError::WriteZeroes(ref io_err) = e {
+                    -io_err.raw_os_error().unwrap_or(libc::EIO)
+                } else {
+                    -libc::EIO
+                };
+                self.completion_list.push_back((user_data, errno));
+                self.eventfd.write(1).unwrap();
+                Ok(())
+            }
+        }
     }
 }
 
@@ -355,6 +383,39 @@ mod unit_tests {
         (temp_file, disk)
     }
 
+    fn create_overlay_disk_with_raw_backing_pattern(
+        file_size: u64,
+        value: u8,
+        direct_io: bool,
+    ) -> (TempFile, TempFile, QcowDisk) {
+        let backing_temp = TempFile::new().unwrap();
+        let backing_data = vec![value; file_size as usize];
+        backing_temp.as_file().write_all(&backing_data).unwrap();
+        backing_temp.as_file().sync_all().unwrap();
+        let backing_path = backing_temp.as_path().to_str().unwrap().to_string();
+
+        let overlay_temp = TempFile::new().unwrap();
+        {
+            let raw = RawFile::new(overlay_temp.as_file().try_clone().unwrap(), false);
+            let backing_config = BackingFileConfig {
+                path: backing_path,
+                format: Some(ImageType::Raw),
+            };
+            QcowFile::new_from_backing(raw, 3, file_size, &backing_config, true).unwrap();
+        }
+
+        let disk = QcowDisk::new(
+            overlay_temp.as_file().try_clone().unwrap(),
+            direct_io,
+            true,
+            true,
+            false,
+        )
+        .unwrap();
+
+        (backing_temp, overlay_temp, disk)
+    }
+
     fn async_read(disk: &QcowDisk, offset: u64, len: usize) -> Vec<u8> {
         let mut async_io = disk.create_async_io(1).unwrap();
         let mut buf = vec![0xFFu8; len];
@@ -829,6 +890,77 @@ mod unit_tests {
         }
     }
 
+    fn test_write_zeroes_unallocated_overlay_with_backing_must_read_zero_impl(direct_io: bool) {
+        let cluster_size = 1u64 << 16;
+        let file_size = cluster_size * 4;
+        let offset = cluster_size;
+        let (_backing_temp, _overlay_temp, disk) =
+            create_overlay_disk_with_raw_backing_pattern(file_size, 0xAB, direct_io);
+
+        let mut async_io = disk.create_async_io(1).unwrap();
+        async_io.write_zeroes(offset, cluster_size, 42).unwrap();
+        let (user_data, result) = async_io.next_completed_request().unwrap();
+        assert_eq!(user_data, 42);
+        assert_eq!(result, 0);
+        drop(async_io);
+
+        let buf = async_read(&disk, offset, cluster_size as usize);
+        assert!(
+            buf.iter().all(|&b| b == 0),
+            "zeroed unallocated overlay cluster exposed backing data"
+        );
+    }
+
+    #[test]
+    fn test_write_zeroes_unallocated_overlay_with_backing_must_read_zero() {
+        test_write_zeroes_unallocated_overlay_with_backing_must_read_zero_impl(false);
+    }
+
+    #[test]
+    fn test_write_zeroes_unallocated_overlay_with_backing_must_read_zero_direct_io() {
+        test_write_zeroes_unallocated_overlay_with_backing_must_read_zero_impl(true);
+    }
+
+    fn test_partial_write_after_write_zeroes_must_not_reintroduce_backing_data_impl(
+        direct_io: bool,
+    ) {
+        let cluster_size = 1u64 << 16;
+        let file_size = cluster_size * 4;
+        let offset = cluster_size;
+        let patch_offset = 0x4000usize;
+        let patch_len = 0x1000usize;
+        let (_backing_temp, _overlay_temp, disk) =
+            create_overlay_disk_with_raw_backing_pattern(file_size, 0xAB, direct_io);
+
+        let mut async_io = disk.create_async_io(1).unwrap();
+        async_io.write_zeroes(offset, cluster_size, 42).unwrap();
+        let (_user_data, result) = async_io.next_completed_request().unwrap();
+        assert_eq!(result, 0);
+        drop(async_io);
+
+        let patch = [0x99u8; 0x1000];
+        async_write(&disk, offset + patch_offset as u64, &patch[..patch_len]);
+
+        let buf = async_read(&disk, offset, cluster_size as usize);
+        assert!(buf[..patch_offset].iter().all(|&b| b == 0));
+        assert!(
+            buf[patch_offset..patch_offset + patch_len]
+                .iter()
+                .all(|&b| b == 0x99)
+        );
+        assert!(buf[patch_offset + patch_len..].iter().all(|&b| b == 0));
+    }
+
+    #[test]
+    fn test_partial_write_after_write_zeroes_must_not_reintroduce_backing_data() {
+        test_partial_write_after_write_zeroes_must_not_reintroduce_backing_data_impl(false);
+    }
+
+    #[test]
+    fn test_partial_write_after_write_zeroes_must_not_reintroduce_backing_data_direct_io() {
+        test_partial_write_after_write_zeroes_must_not_reintroduce_backing_data_impl(true);
+    }
+
     fn test_backing_file_read_qcow2_backing_impl(direct_io: bool) {
         let backing_temp = TempFile::new().unwrap();
         let cluster_size = 1u64 << 16;