add few tests

core/storage/page_cache.rs

@@ -346,16 +346,15 @@ impl DumbLruPageCache {
         Ok(())
     }
 
+    /// Removes all pages from the cache with pgno greater than len
     pub fn truncate(&mut self, len: usize) -> Result<(), CacheError> {
         let head_ptr = *self.head.borrow();
         let mut current = head_ptr;
-        let mut has_non_removed = false;
         while let Some(node) = current {
             let node_ref = unsafe { node.as_ref() };
 
             current = node_ref.next;
             if node_ref.key.pgno <= len {
-                has_non_removed = true;
                 continue;
             }
 
@@ -369,10 +368,6 @@ impl DumbLruPageCache {
                 let _ = Box::from_raw(node.as_ptr());
             }
         }
-        if !has_non_removed {
-            let _ = self.head.take();
-            let _ = self.tail.take();
-        }
         Ok(())
     }
 
@@ -1269,6 +1264,38 @@ mod tests {
         assert!(cache.insert(create_key(4), page_with_content(4)).is_ok());
     }
 
+    #[test]
+    fn test_truncate_page_cache() {
+        let mut cache = DumbLruPageCache::new(10);
+        let _ = insert_page(&mut cache, 1);
+        let _ = insert_page(&mut cache, 4);
+        let _ = insert_page(&mut cache, 8);
+        let _ = insert_page(&mut cache, 10);
+        cache.truncate(4).unwrap();
+        assert!(cache.contains_key(&PageCacheKey { pgno: 1 }));
+        assert!(cache.contains_key(&PageCacheKey { pgno: 4 }));
+        assert!(!cache.contains_key(&PageCacheKey { pgno: 8 }));
+        assert!(!cache.contains_key(&PageCacheKey { pgno: 10 }));
+        assert_eq!(cache.len(), 2);
+        assert_eq!(cache.capacity, 10);
+        cache.verify_list_integrity();
+        assert!(cache.insert(create_key(8), page_with_content(8)).is_ok());
+    }
+
+    #[test]
+    fn test_truncate_page_cache_remove_all() {
+        let mut cache = DumbLruPageCache::new(10);
+        let _ = insert_page(&mut cache, 8);
+        let _ = insert_page(&mut cache, 10);
+        cache.truncate(4).unwrap();
+        assert!(!cache.contains_key(&PageCacheKey { pgno: 8 }));
+        assert!(!cache.contains_key(&PageCacheKey { pgno: 10 }));
+        assert_eq!(cache.len(), 0);
+        assert_eq!(cache.capacity, 10);
+        cache.verify_list_integrity();
+        assert!(cache.insert(create_key(8), page_with_content(8)).is_ok());
+    }
+
     #[test]
     #[ignore = "long running test, remove to verify"]
     fn test_clear_memory_stability() {

tests/integration/functions/test_wal_api.rs

@@ -627,3 +627,50 @@ fn test_wal_checkpoint_no_work() {
     );
     reader.execute("SELECT * FROM test").unwrap();
 }
+
+#[test]
+fn test_wal_revert_change_db_size() {
+    let db = TempDatabase::new_empty(false);
+    let writer = db.connect_limbo();
+
+    writer.execute("create table t(x, y)").unwrap();
+    let watermark = writer.wal_state().unwrap().max_frame;
+    writer
+        .execute("insert into t values (1, randomblob(10 * 4096))")
+        .unwrap();
+    writer
+        .execute("insert into t values (2, randomblob(20 * 4096))")
+        .unwrap();
+    let mut changed = writer.wal_changed_pages_after(watermark).unwrap();
+    changed.sort();
+
+    let mut frame = [0u8; 4096 + 24];
+
+    writer.wal_insert_begin().unwrap();
+    let mut frames_count = writer.wal_state().unwrap().max_frame;
+    for page_no in changed {
+        let page = &mut frame[24..];
+        if !writer
+            .try_wal_watermark_read_page(page_no, page, Some(watermark))
+            .unwrap()
+        {
+            continue;
+        }
+        let info = WalFrameInfo {
+            page_no: page_no,
+            db_size: if page_no == 2 { 2 } else { 0 },
+        };
+        info.put_to_frame_header(&mut frame);
+        frames_count += 1;
+        writer.wal_insert_frame(frames_count, &frame).unwrap();
+    }
+    writer.wal_insert_end().unwrap();
+
+    writer
+        .execute("insert into t values (3, randomblob(30 * 4096))")
+        .unwrap();
+    assert_eq!(
+        limbo_exec_rows(&db, &writer, "SELECT x, length(y) FROM t"),
+        vec![vec![Value::Integer(3), Value::Integer(30 * 4096)]]
+    );
+}