Skip blanking a page that was never used

mvcc/src/database.rs

@@ -804,21 +804,17 @@ impl Inner {
                             sealed: true,
                         });
                     }
-                    let page_id = self
-                        .storage
-                        .reserve_page_ids(1)?
-                        .into_iter()
-                        .next()
-                        .ok_or(TransactionError::TimestampExhausted)?;
-                    // A page off the free pool still holds the records of its
-                    // previous life. Blank it here, while the open page lock
-                    // is held, so it is empty before any writer can append to
-                    // it. Clearing it as part of an append would depend on
-                    // which writer reached the log first.
-                    self.storage.commit_relaxed(vec![PageWrite {
-                        page_id,
-                        payload: Vec::new(),
-                    }])?;
+                    let (page_id, reused) = self.storage.reserve_one_page()?;
+                    // Only a recycled page carries records from a previous
+                    // life. A fresh one is past the end of the file and reads
+                    // as empty, so it needs no blanking and no round trip
+                    // while this lock is held.
+                    if reused {
+                        self.storage.commit_relaxed(vec![PageWrite {
+                            page_id,
+                            payload: Vec::new(),
+                        }])?;
+                    }
                     *open = Some(OpenPage { page_id, offset: 0 });
                 }
                 let page = open.as_mut().expect("a page is open");

storage/src/group_commit.rs

@@ -290,6 +290,11 @@ impl GroupCommitHandle {
         self.allocator.reserve(count)
     }
 
+    /// Reserve one page, reporting whether it came off the free pool.
+    pub fn reserve_one_page(&self) -> Result<(PageId, bool)> {
+        self.allocator.reserve_one()
+    }
+
     /// Return unreachable pages to the free pool.
     pub fn release_pages(&self, pages: impl IntoIterator<Item = PageId>) -> Result<()> {
         self.allocator.release(pages)

storage/src/store.rs

@@ -90,6 +90,20 @@ struct AllocatorState {
 }
 
 impl PageAllocator {
+    /// Reserve one page, reporting whether it came off the free pool.
+    pub fn reserve_one(&self) -> Result<(PageId, bool)> {
+        let mut state = self.inner.lock().map_err(|_| StorageError::Poisoned)?;
+        if let Some(reused) = state.free_pages.pop_first() {
+            return Ok((reused, true));
+        }
+        let page_id = PageId(state.next_page_id);
+        state.next_page_id = state
+            .next_page_id
+            .checked_add(1)
+            .ok_or_else(|| StorageError::Configuration("page ID space exhausted".to_owned()))?;
+        Ok((page_id, false))
+    }
+
     pub fn reserve(&self, count: usize) -> Result<Vec<PageId>> {
         let mut state = self.inner.lock().map_err(|_| StorageError::Poisoned)?;
         let mut page_ids = Vec::with_capacity(count);