Continue compacting until content fits window during compaction

Also increase the virtual window size.

libraries/persistent_store/src/format.rs

@@ -266,24 +266,24 @@ impl Format {
 
     /// The total virtual capacity in words, denoted by V.
     ///
-    /// We have V = (N - 1) × (Q - 1) - M.
+    /// We have V = (N - 1) × Q - M - 1.
     ///
-    /// We can show V ≥ (N - 2) × (Q - 1) with the following steps:
+    /// We can show V ≥ (N - 2) × Q with the following steps:
-    /// - M ≤ Q - 1 from M < Q from [M](Format::max_prefix_len)'s definition
+    /// - M + 1 ≤ Q from M < Q from [M](Format::max_prefix_len)'s definition
-    /// - -M ≥ -(Q - 1) from above
+    /// - -M - 1 ≥ -Q from above
-    /// - V ≥ (N - 1) × (Q - 1) - (Q - 1) from V's definition
+    /// - V ≥ (N - 1) × Q - Q from V's definition
     pub fn virt_size(&self) -> Nat {
-        (self.num_pages() - 1) * (self.virt_page_size() - 1) - self.max_prefix_len()
+        (self.num_pages() - 1) * self.virt_page_size() - self.max_prefix_len() - 1
     }
 
     /// The total user capacity in words, denoted by C.
     ///
-    /// We have C = V - N = (N - 1) × (Q - 2) - M - 1.
+    /// We have C = V - N = (N - 1) × (Q - 1) - M - 2.
     ///
-    /// We can show C ≥ (N - 2) × (Q - 2) - 2 with the following steps:
+    /// We can show C ≥ (N - 2) × (Q - 1) - 2 with the following steps:
-    /// - V ≥ (N - 2) × (Q - 1) from [V](Format::virt_size)'s definition
+    /// - V ≥ (N - 2) × Q from [V](Format::virt_size)'s definition
-    /// - C ≥ (N - 2) × (Q - 1) - N from C's definition
+    /// - C ≥ (N - 2) × Q - N from C's definition
-    /// - (N - 2) × (Q - 1) - N = (N - 2) × (Q - 2) - 2 by calculus
+    /// - (N - 2) × Q - N = (N - 2) × (Q - 1) - 2 by calculus
     pub fn total_capacity(&self) -> Nat {
         // From the virtual capacity, we reserve N - 1 words for `Erase` entries and 1 word for a
         // `Clear` entry.

libraries/persistent_store/src/store.rs

@@ -354,7 +354,7 @@ impl<S: Storage> Store<S> {
             .build_internal(InternalEntry::Clear { min_key })?;
         // We always have one word available. We can't use `reserve` because this is internal
         // capacity, not user capacity.
-        while self.immediate_capacity()? < 1 {
+        while self.immediate_capacity()?.unwrap_or(0) < 1 {
             self.compact()?;
         }
         let tail = self.tail()?;
@@ -370,8 +370,9 @@ impl<S: Storage> Store<S> {
         if self.capacity()?.remaining() < length {
             return Err(StoreError::NoCapacity);
         }
-        if self.immediate_capacity()? < usize_to_nat(length) {
+        if self.immediate_capacity()?.unwrap_or(0) < usize_to_nat(length) {
             self.compact()?;
+            self.finish_compaction()?;
         }
         Ok(())
     }
@@ -521,10 +522,11 @@ impl<S: Storage> Store<S> {
         self.head = Some(head);
         let head_page = head.page(&self.format);
         match self.parse_compact(head_page)? {
-            WordState::Erased => Ok(()),
+            WordState::Erased => (),
-            WordState::Partial => self.compact(),
+            WordState::Partial => self.compact()?,
-            WordState::Valid(_) => self.compact_copy(),
+            WordState::Valid(_) => self.compact_copy()?,
         }
+        self.finish_compaction()
     }
 
     /// Recover a possible interrupted operation which is not a compaction.
@@ -685,7 +687,7 @@ impl<S: Storage> Store<S> {
         if self.capacity()?.remaining() < length as usize {
             return Err(StoreError::NoCapacity);
         }
-        while self.immediate_capacity()? < length {
+        while self.immediate_capacity()?.unwrap_or(0) < length {
             self.compact()?;
         }
         Ok(())
@@ -785,6 +787,23 @@ impl<S: Storage> Store<S> {
         Ok(())
     }
 
+    /// Finish a compaction sequence.
+    ///
+    /// Because compaction may temporarily move the tail outside the virtual window, we must keep
+    /// compacting until that invariant is restored. As such, this function must be called each time
+    /// the store is compacted without checking whether there is immediate capacity.
+    fn finish_compaction(&mut self) -> StoreResult<()> {
+        let mut count = self.format.num_pages() - 2;
+        while count > 0 && self.immediate_capacity()?.is_none() {
+            self.compact()?;
+            count -= 1;
+        }
+        if self.immediate_capacity()?.is_none() {
+            return Err(StoreError::InvalidStorage);
+        }
+        Ok(())
+    }
+
     /// Continues a transaction after it has been written.
     fn transaction_apply(&mut self, sorted_keys: &[Nat], marker: Position) -> StoreResult<()> {
         self.delete_keys(&sorted_keys, marker)?;
@@ -926,10 +945,10 @@ impl<S: Storage> Store<S> {
     }
 
     /// Returns the number of words that can be written without compaction.
-    fn immediate_capacity(&self) -> StoreResult<Nat> {
+    fn immediate_capacity(&self) -> StoreResult<Option<Nat>> {
         let tail = self.tail()?;
         let end = or_invalid(self.head)? + self.format.virt_size();
-        Ok(end.get().saturating_sub(tail.get()))
+        Ok(end.get().checked_sub(tail.get()))
     }
 
     /// Returns the position of the first word in the store.
@@ -1394,34 +1413,34 @@ mod tests {
         let mut driver = MINIMAL.new_driver().power_on().unwrap();
 
         // Don't compact if enough immediate capacity.
-        assert_eq!(driver.store().immediate_capacity().unwrap(), 39);
+        assert_eq!(driver.store().immediate_capacity().unwrap(), Some(42));
-        assert_eq!(driver.store().capacity().unwrap().remaining(), 34);
+        assert_eq!(driver.store().capacity().unwrap().remaining(), 37);
         assert_eq!(driver.store().head().unwrap().get(), 0);
-        driver.store_mut().prepare(34).unwrap();
+        driver.store_mut().prepare(37).unwrap();
         assert_eq!(driver.store().head().unwrap().get(), 0);
 
         // Fill the store.
         for key in 0..4 {
-            driver.insert(key, &[0x38; 28]).unwrap();
+            driver.insert(key, &[0x38; 32]).unwrap();
         }
         driver.check().unwrap();
-        assert_eq!(driver.store().immediate_capacity().unwrap(), 7);
+        assert_eq!(driver.store().immediate_capacity().unwrap(), Some(6));
-        assert_eq!(driver.store().capacity().unwrap().remaining(), 2);
+        assert_eq!(driver.store().capacity().unwrap().remaining(), 1);
         // Removing entries increases available capacity but not immediate capacity.
         driver.remove(0).unwrap();
         driver.remove(2).unwrap();
         driver.check().unwrap();
-        assert_eq!(driver.store().immediate_capacity().unwrap(), 7);
+        assert_eq!(driver.store().immediate_capacity().unwrap(), Some(6));
-        assert_eq!(driver.store().capacity().unwrap().remaining(), 18);
+        assert_eq!(driver.store().capacity().unwrap().remaining(), 19);
 
-        // Prepare for next write (7 words data + 1 word overhead).
+        // Prepare for next write (8 words data + 1 word overhead).
         assert_eq!(driver.store().head().unwrap().get(), 0);
-        driver.store_mut().prepare(8).unwrap();
+        driver.store_mut().prepare(9).unwrap();
         driver.check().unwrap();
-        assert_eq!(driver.store().head().unwrap().get(), 16);
+        assert_eq!(driver.store().head().unwrap().get(), 18);
         // The available capacity did not change, but the immediate capacity is above 8.
-        assert_eq!(driver.store().immediate_capacity().unwrap(), 14);
+        assert_eq!(driver.store().immediate_capacity().unwrap(), Some(14));
-        assert_eq!(driver.store().capacity().unwrap().remaining(), 18);
+        assert_eq!(driver.store().capacity().unwrap().remaining(), 19);
     }
 
     #[test]

libraries/persistent_store/src/test.rs

@@ -67,13 +67,13 @@ const TITAN: Config = Config {
 #[test]
 fn nordic_capacity() {
     let driver = NORDIC.new_driver().power_on().unwrap();
-    assert_eq!(driver.model().capacity().total, 19123);
+    assert_eq!(driver.model().capacity().total, 19141);
 }
 
 #[test]
 fn titan_capacity() {
     let driver = TITAN.new_driver().power_on().unwrap();
-    assert_eq!(driver.model().capacity().total, 4315);
+    assert_eq!(driver.model().capacity().total, 4323);
 }
 
 #[test]