Merge 'core/wal: cache file size' from Pere Diaz Bou

Closes #2829

core/storage/sqlite3_ondisk.rs

@@ -1632,9 +1632,9 @@ pub fn read_entire_wal_dumb(file: &Arc<dyn File>) -> Result<Arc<UnsafeCell<WalFi
         write_lock: TursoRwLock::new(),
         loaded: AtomicBool::new(false),
         checkpoint_lock: TursoRwLock::new(),
+        initialized: AtomicBool::new(false),
     }));
     let wal_file_shared_for_completion = wal_file_shared_ret.clone();
-
     let complete: Box<ReadComplete> = Box::new(move |res: Result<(Arc<Buffer>, i32), _>| {
         let Ok((buf, bytes_read)) = res else {
             return;
@@ -1825,6 +1825,9 @@ pub fn read_entire_wal_dumb(file: &Arc<dyn File>) -> Result<Arc<UnsafeCell<WalFi
 
         wfs_data.nbackfills.store(0, Ordering::SeqCst);
         wfs_data.loaded.store(true, Ordering::SeqCst);
+        if size >= WAL_HEADER_SIZE as u64 {
+            wfs_data.initialized.store(true, Ordering::SeqCst);
+        }
     });
     let c = Completion::new_read(buf_for_pread, complete);
     let _c = file.pread(0, c)?;

core/storage/wal.rs

@@ -682,6 +682,7 @@ pub struct WalFileShared {
     /// Serialises checkpointer threads, only one checkpoint can be in flight at any time. Blocking and exclusive only
     pub checkpoint_lock: TursoRwLock,
     pub loaded: AtomicBool,
+    pub initialized: AtomicBool,
 }
 
 impl fmt::Debug for WalFileShared {
@@ -1629,6 +1630,7 @@ impl WalFile {
             )?)?;
         self.io
             .wait_for_completion(shared.file.sync(Completion::new_sync(|_| {}))?)?;
+        shared.initialized.store(true, Ordering::Release);
         Ok(())
     }
 
@@ -2035,6 +2037,7 @@ impl WalFile {
                 .file
                 .truncate(0, c)
                 .inspect_err(|e| unlock(Some(e)))?;
+            shared.initialized.store(false, Ordering::Release);
             self.io
                 .wait_for_completion(c)
                 .inspect_err(|e| unlock(Some(e)))?;
@@ -2141,7 +2144,7 @@ impl WalFileShared {
     }
 
     pub fn is_initialized(&self) -> Result<bool> {
-        Ok(self.file.size()? >= WAL_HEADER_SIZE as u64)
+        Ok(self.initialized.load(Ordering::Acquire))
     }
 
     pub fn new_shared(file: Arc<dyn File>) -> Result<Arc<UnsafeCell<WalFileShared>>> {
@@ -2181,6 +2184,7 @@ impl WalFileShared {
             write_lock: TursoRwLock::new(),
             checkpoint_lock: TursoRwLock::new(),
             loaded: AtomicBool::new(true),
+            initialized: AtomicBool::new(false),
         };
         Ok(Arc::new(UnsafeCell::new(shared)))
     }