[fxfs] Fix bug where pages may not be supplied.

It's possible that VMO size and 'byte_size' are different.
This fixes a bug where, if byte_size is smaller, the page handler
may fail to supply required pages.

It also ensures that read ahead will not supply zero pages beyond the end
of the VMO (uses max of byte_size and requested range) as this
will cause supply_pages to fail a range check and lead to a similar
result.

Bug: b/318773519
Change-Id: I87dd37a7065a6e454668b7e72ad1f7a98992b1e4
Reviewed-on: https://fuchsia-review.googlesource.com/c/fuchsia/+/971633
Commit-Queue: Aaron Drew <ripper@google.com>
Reviewed-by: Chris Suter <csuter@google.com>

Author: Aaron Drew

src/storage/fxfs/platform/src/fuchsia/pager.rs

@@ -402,7 +402,7 @@ pub trait PagerBacked: Sync + Send + Sized + 'static {
 }
 
 /// A generic page_in implementation that supplies pages using block-aligned reads.
-pub fn default_page_in<P: PagerBacked>(this: Arc<P>, mut range: Range<u64>) {
+pub fn default_page_in<P: PagerBacked>(this: Arc<P>, range: Range<u64>) {
     fxfs_trace::duration!(
         "start-page-in",
         "offset" => range.start,
@@ -425,27 +425,43 @@ pub fn default_page_in<P: PagerBacked>(this: Arc<P>, mut range: Range<u64>) {
     } else {
         round_down(READ_AHEAD_SIZE, read_alignment)
     };
+
+    // Two important subtleties to consider in this space:
+    //
+    // `byte_size` is the official size of the object. VMOs are page-aligned so `aligned_size` is
+    // the "offical" page length of the object. This may be smaller than Vmo::get_size because
+    // these two things are not updated atomically. The reverse is not true -- We do not currently
+    // ever shrink a VMO's size. We also do not update byte_size (self.handle.get_size()) if
+    // an independent handle is used to grow a file. This means the VMO's size should always be
+    // strictly equal or bigger than `byte_size`.
+    //
+    // It is valid to supply more pages than asked, but supplying pages outside of the VMO range
+    // will trigger OUT_OF_RANGE errors and the call will fail without supplying anything.
+    // We must supply the range requested under all circumstances to unblock any page misses but
+    // we should take care to never supply additional pages beyond the object length (`byte_size`)
+    // as there is a chance that we might serve a range outside of the VMO and fail to supply
+    // anything at all.
+
     let aligned_size = round_up(this.byte_size(), read_alignment).unwrap();
-    range = round_down(range.start, readahead_alignment)
-        ..round_up(range.end, readahead_alignment).unwrap();
-    if range.end > aligned_size {
-        range.end = aligned_size;
-    }
 
     // Zero-pad the tail if requested range exceeds the size of the thing we're reading.
+    // This can happen when we truncate and there are outstanding pager requests that the kernel
+    // was not able to cancel in time.
     let mut offset = std::cmp::max(range.start, aligned_size);
     while offset < range.end {
         let end = std::cmp::min(range.end, offset + ZERO_VMO_SIZE);
         pager.supply_pages(this.vmo(), offset..end, &ZERO_VMO, 0);
         offset = end;
     }
 
+    let mut readahead_range = round_down(range.start, readahead_alignment)
+        ..std::cmp::min(round_up(range.end, readahead_alignment).unwrap(), aligned_size);
     // Read in chunks of 128 KiB.
     let read_size = round_up(128 * 1024, read_alignment).unwrap();
-
-    while range.start < range.end {
-        let read_range = range.start..std::cmp::min(range.end, range.start + read_size);
-        range.start += read_size;
+    while readahead_range.start < readahead_range.end {
+        let read_range = readahead_range.start
+            ..std::cmp::min(readahead_range.end, readahead_range.start + read_size);
+        readahead_range.start += read_size;
 
         let this = this.clone();
         this.clone().pager().spawn(page_in_chunk(this, read_range, ref_guard.clone()));