dev: synchronize memory change with NotifyActor

compiler/src/lib.rs

@@ -49,7 +49,10 @@ pub mod service;
 /// Run the compiler in the system environment.
 #[cfg(feature = "system-compile")]
 pub(crate) mod system;
-use std::path::{Path, PathBuf};
+use std::{
+    path::{Path, PathBuf},
+    sync::Arc,
+};
 
 #[cfg(feature = "system-compile")]
 pub use system::TypstSystemWorld;
@@ -72,8 +75,15 @@ pub trait ShadowApi {
         unimplemented!()
     }
 
+    /// Get the shadow files.
+    fn shadow_paths(&self) -> Vec<Arc<Path>>;
+
     /// Reset the shadow files.
-    fn reset_shadow(&mut self);
+    fn reset_shadow(&mut self) {
+        for path in self.shadow_paths() {
+            self.unmap_shadow(&path).unwrap();
+        }
+    }
 
     /// Add a shadow file to the driver.
     fn map_shadow(&self, path: &Path, content: Bytes) -> FileResult<()>;

compiler/src/service/compile.rs

@@ -1,4 +1,11 @@
-use std::{num::NonZeroUsize, path::PathBuf, sync::Arc, thread::JoinHandle};
+use std::{
+    collections::HashSet,
+    num::NonZeroUsize,
+    ops::Deref,
+    path::{Path, PathBuf},
+    sync::Arc,
+    thread::JoinHandle,
+};
 
 use serde::Serialize;
 use tokio::sync::{mpsc, oneshot};
@@ -10,7 +17,7 @@ use typst::{
 };
 
 use crate::{
-    vfs::notify::{FileChangeSet, FilesystemEvent, NotifyMessage},
+    vfs::notify::{FilesystemEvent, MemoryEvent, NotifyMessage},
     world::{CompilerFeat, CompilerWorld},
     ShadowApi,
 };
@@ -33,11 +40,6 @@ fn ensure_single_thread<F: std::future::Future<Output = ()> + Send + 'static>(
     })
 }
 
-pub enum MemoryEvent {
-    Sync(FileChangeSet),
-    Update(FileChangeSet),
-}
-
 enum CompilerInterrupt<Ctx> {
     /// Interrupted by file system event.
     Fs(Option<FilesystemEvent>),
@@ -58,6 +60,7 @@ pub struct CompileActor<C: Compiler> {
     pub root: PathBuf,
     pub enable_watch: bool,
 
+    estimated_shadow_files: HashSet<Arc<Path>>,
     latest_doc: Option<Arc<TypstDocument>>,
 
     steal_send: mpsc::UnboundedSender<BorrowTask<Self>>,
@@ -84,6 +87,7 @@ where
             root,
             enable_watch: false,
 
+            estimated_shadow_files: Default::default(),
             latest_doc: None,
 
             steal_send,
@@ -98,7 +102,8 @@ where
         use CompilerResponse::*;
 
         // compile
-        self.compiler
+        self.latest_doc = self
+            .compiler
             .with_stage_diag::<true, _>("compiling", |driver| driver.compile());
 
         comemo::evict(30);
@@ -110,12 +115,34 @@ where
         // tx
     }
 
-    fn process(&mut self, event: CompilerInterrupt<Self>) -> bool {
+    fn process(&mut self, event: CompilerInterrupt<Self>, send: impl Fn(CompilerResponse)) -> bool {
+        use CompilerResponse::*;
+
         match event {
             CompilerInterrupt::Fs(event) => {
                 log::info!("CompileActor: fs event incoming {:?}", event);
 
-                if let Some(event) = event {
+                if let Some(mut event) = event {
+                    if let FilesystemEvent::UpstreamUpdate { upstream_event, .. } = &mut event {
+                        let event = upstream_event.take().unwrap().opaque;
+                        let event = *event.downcast::<MemoryEvent>().unwrap();
+
+                        if matches!(event, MemoryEvent::Sync(..)) {
+                            self.compiler.reset_shadow();
+                        }
+                        match event {
+                            MemoryEvent::Update(event) | MemoryEvent::Sync(event) => {
+                                for removes in event.removes {
+                                    let _ = self.compiler.unmap_shadow(&removes);
+                                }
+                                for (p, t) in event.inserts {
+                                    let _ =
+                                        self.compiler.map_shadow(&p, t.content().cloned().unwrap());
+                                }
+                            }
+                        }
+                    }
+
                     self.compiler.notify_fs_event(event);
                 }
 
@@ -124,21 +151,32 @@ where
             CompilerInterrupt::Memory(event) => {
                 log::info!("CompileActor: memory event incoming");
 
+                let mut files = HashSet::new();
                 if matches!(event, MemoryEvent::Sync(..)) {
-                    self.compiler.reset_shadow();
+                    files = self.estimated_shadow_files.clone();
+                    self.estimated_shadow_files.clear();
                 }
-                match event {
-                    MemoryEvent::Update(event) | MemoryEvent::Sync(event) => {
-                        for removes in event.removes {
-                            let _ = self.compiler.unmap_shadow(&removes);
+                match &event {
+                    MemoryEvent::Sync(event) | MemoryEvent::Update(event) => {
+                        for path in event.removes.iter().map(Deref::deref) {
+                            self.estimated_shadow_files.remove(path);
+                            files.insert(path.into());
                         }
-                        for (p, t) in event.inserts {
-                            let _ = self.compiler.map_shadow(&p, t.content().cloned().unwrap());
+                        for path in event.inserts.iter().map(|e| e.0.deref()) {
+                            self.estimated_shadow_files.insert(path.into());
+                            files.remove(path);
                         }
                     }
                 }
 
-                true
+                send(Notify(NotifyMessage::UpstreamUpdate(
+                    crate::vfs::notify::UpstreamUpdateEvent {
+                        invalidates: files.iter().map(|e| e.as_ref().to_owned()).collect(),
+                        opaque: Box::new(event),
+                    },
+                )));
+
+                false
             }
             CompilerInterrupt::Task(task) => {
                 task(self);
@@ -173,7 +211,7 @@ where
                 Some(it) = self.steal_recv.recv() => Some(CompilerInterrupt::Task(it)),
             } {
                 let mut need_recompile = false;
-                need_recompile = self.process(event) || need_recompile;
+                need_recompile = self.process(event, &compiler_ack) || need_recompile;
                 while let Some(event) = fs_rx
                     .try_recv()
                     .ok()
@@ -186,7 +224,7 @@ where
                     })
                     .or_else(|| self.steal_recv.try_recv().ok().map(CompilerInterrupt::Task))
                 {
-                    need_recompile = self.process(event) || need_recompile;
+                    need_recompile = self.process(event, &compiler_ack) || need_recompile;
                 }
 
                 if need_recompile {

compiler/src/service/driver.rs

@@ -1,4 +1,7 @@
-use std::path::{Path, PathBuf};
+use std::{
+    path::{Path, PathBuf},
+    sync::Arc,
+};
 
 use crate::{NotifyApi, ShadowApi};
 use typst::{diag::SourceResult, syntax::VirtualPath, World};
@@ -104,18 +107,27 @@ impl<W: World + WorkspaceProvider + NotifyApi> Compiler for CompileDriverImpl<W>
 }
 
 impl<W: World + ShadowApi> ShadowApi for CompileDriverImpl<W> {
+    #[inline]
     fn _shadow_map_id(&self, file_id: TypstFileId) -> typst::diag::FileResult<PathBuf> {
         self.world._shadow_map_id(file_id)
     }
 
+    #[inline]
+    fn shadow_paths(&self) -> Vec<Arc<Path>> {
+        self.world.shadow_paths()
+    }
+
+    #[inline]
     fn reset_shadow(&mut self) {
         self.world.reset_shadow()
     }
 
+    #[inline]
     fn map_shadow(&self, path: &Path, content: Bytes) -> typst::diag::FileResult<()> {
         self.world.map_shadow(path, content)
     }
 
+    #[inline]
     fn unmap_shadow(&self, path: &Path) -> typst::diag::FileResult<()> {
         self.world.unmap_shadow(path)
     }

compiler/src/service/mod.rs

@@ -254,6 +254,11 @@ where
         self.inner()._shadow_map_id(_file_id)
     }
 
+    #[inline]
+    fn shadow_paths(&self) -> Vec<Arc<Path>> {
+        self.inner().shadow_paths()
+    }
+
     #[inline]
     fn reset_shadow(&mut self) {
         self.inner_mut().reset_shadow()

compiler/src/service/watch.rs

@@ -18,7 +18,7 @@ use typst::diag::{FileError, FileResult};
 use typst_ts_core::Bytes;
 
 use crate::vfs::{
-    notify::{FileChangeSet, FilesystemEvent, NotifyFile, NotifyMessage},
+    notify::{FileChangeSet, FilesystemEvent, NotifyFile, NotifyMessage, UpstreamUpdateEvent},
     system::SystemAccessModel,
     AccessModel,
 };
@@ -139,7 +139,7 @@ impl NotifyActor {
         }
     }
 
-    /// Send a message to the actor.
+    /// Send a filesystem event to remove.
     fn send(&mut self, msg: FilesystemEvent) {
         self.sender.send(msg).unwrap();
     }
@@ -185,13 +185,18 @@ impl NotifyActor {
             // log::info!("vfs-notify event {event:?}");
             // function entries to handle some event
             match event {
+                ActorEvent::Message(NotifyMessage::UpstreamUpdate(event)) => {
+                    self.invalidate_upstream(event);
+                }
                 ActorEvent::Message(NotifyMessage::SyncDependency(paths)) => {
-                    self.update_watches(paths).await;
+                    if let Some(changeset) = self.update_watches(&paths) {
+                        self.send(FilesystemEvent::Update(changeset));
+                    }
                 }
                 ActorEvent::NotifyEvent(event) => {
                     // log::info!("notify event {event:?}");
                     if let Some(event) = log_notify_error(event, "failed to notify") {
-                        self.notify_event(event).await;
+                        self.notify_event(event);
                     }
                 }
                 ActorEvent::ReCheck(event) => {
@@ -201,8 +206,17 @@ impl NotifyActor {
         }
     }
 
+    /// Update the watches of corresponding invalidation
+    fn invalidate_upstream(&mut self, event: UpstreamUpdateEvent) {
+        let changeset = self.update_watches(&event.invalidates).unwrap_or_default();
+        self.send(FilesystemEvent::UpstreamUpdate {
+            changeset,
+            upstream_event: Some(event),
+        });
+    }
+
     /// Update the watches of corresponding files.
-    async fn update_watches(&mut self, paths: Vec<PathBuf>) {
+    fn update_watches(&mut self, paths: &[PathBuf]) -> Option<FileChangeSet> {
         // Increase the lifetime per external message.
         self.lifetime += 1;
 
@@ -247,7 +261,7 @@ impl NotifyActor {
         //
         // Also check whether the file is updated since there is a window
         // between unwatch the file and watch the file again.
-        for path in paths.into_iter() {
+        for path in paths.iter() {
             // Update or insert the entry with the new lifetime.
             let entry = self
                 .watched_entries
@@ -288,9 +302,9 @@ impl NotifyActor {
             } else {
                 let watched = self
                     .inner
-                    .content(&path)
+                    .content(path)
                     .map(|e| (meta.modified().unwrap(), e));
-                changeset.inserts.push((path, watched.into()));
+                changeset.inserts.push((path.clone(), watched.into()));
             }
         }
 
@@ -306,13 +320,11 @@ impl NotifyActor {
             }
         });
 
-        if !changeset.is_empty() {
-            self.send(FilesystemEvent::Update(changeset));
-        }
+        (!changeset.is_empty()).then_some(changeset)
     }
 
     /// Notify the event from the builtin watcher.
-    async fn notify_event(&mut self, event: notify::Event) {
+    fn notify_event(&mut self, event: notify::Event) {
         // Account file updates.
         let mut changeset = FileChangeSet::default();
         for path in event.paths.into_iter() {

compiler/src/vfs/mod.rs

@@ -153,6 +153,10 @@ impl<M: AccessModel + Sized> Vfs<M> {
         self.access_model.inner().clear_shadow();
     }
 
+    pub fn shadow_paths(&self) -> Vec<Arc<Path>> {
+        self.access_model.inner().file_paths()
+    }
+
     /// Set the `do_reparse` flag.
     pub fn set_do_reparse(&mut self, do_reparse: bool) {
         self.do_reparse = do_reparse;