add truncate method to the page cache

core/storage/page_cache.rs

@@ -1,8 +1,11 @@
+use std::sync::atomic::Ordering;
 use std::{cell::RefCell, ptr::NonNull};
 
 use std::sync::Arc;
 use tracing::{debug, trace};
 
+use crate::turso_assert;
+
 use super::pager::PageRef;
 
 /// FIXME: https://github.com/tursodatabase/turso/issues/1661
@@ -343,13 +346,48 @@ impl DumbLruPageCache {
         Ok(())
     }
 
+    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;
+            }
+
+            self.map.borrow_mut().remove(&node_ref.key);
+            turso_assert!(!node_ref.page.is_dirty(), "page must be clean");
+            turso_assert!(!node_ref.page.is_locked(), "page must be unlocked");
+            turso_assert!(!node_ref.page.is_pinned(), "page must be unpinned");
+            self.detach(node, true)?;
+
+            unsafe {
+                let _ = Box::from_raw(node.as_ptr());
+            }
+        }
+        if !has_non_removed {
+            let _ = self.head.take();
+            let _ = self.tail.take();
+        }
+        Ok(())
+    }
+
     pub fn print(&self) {
         tracing::debug!("page_cache_len={}", self.map.borrow().len());
         let head_ptr = *self.head.borrow();
         let mut current = head_ptr;
         while let Some(node) = current {
             unsafe {
-                tracing::debug!("page={:?}", node.as_ref().key);
+                tracing::debug!(
+                    "page={:?}, flags={}, pin_count={}",
+                    node.as_ref().key,
+                    node.as_ref().page.get().flags.load(Ordering::SeqCst),
+                    node.as_ref().page.get().pin_count.load(Ordering::SeqCst),
+                );
                 let node_ref = node.as_ref();
                 current = node_ref.next;
             }

core/storage/pager.rs

@@ -1413,7 +1413,16 @@ impl Pager {
                 page.get().id == header.page_number as usize,
                 "page has unexpected id"
             );
-            self.add_dirty(&page);
+        }
+        if header.page_number == 1 {
+            let db_size = self
+                .io
+                .block(|| self.with_header(|header| header.database_size))?;
+            tracing::debug!("truncate page_cache as first page was written: {}", db_size);
+            let mut page_cache = self.page_cache.write();
+            page_cache.truncate(db_size.get() as usize).map_err(|e| {
+                LimboError::InternalError(format!("Failed to truncate page cache: {e:?}"))
+            })?;
         }
         if header.is_commit_frame() {
             for page_id in self.dirty_pages.borrow().iter() {