missed one

reactive_graph/src/computed/async_derived/arc_async_derived.rs

@@ -324,7 +324,7 @@ macro_rules! spawn_derived {
                     }
 
                     while rx.next().await.is_some() {
-                        let update_if_necessary = if $should_track {
+                        let update_if_necessary = !owner.paused() && if $should_track {
                             any_subscriber
                                 .with_observer(|| any_subscriber.update_if_necessary())
                         } else {

reactive_graph/src/effect/effect.rs

@@ -170,9 +170,10 @@ impl Effect<LocalStorage> {
 
                 async move {
                     while rx.next().await.is_some() {
-                        if subscriber
-                            .with_observer(|| subscriber.update_if_necessary())
-                            || first_run
+                        if !owner.paused()
+                            && (subscriber.with_observer(|| {
+                                subscriber.update_if_necessary()
+                            }) || first_run)
                         {
                             first_run = false;
                             subscriber.clear_sources(&subscriber);
@@ -321,9 +322,10 @@ impl Effect<LocalStorage> {
 
                 async move {
                     while rx.next().await.is_some() {
-                        if subscriber
-                            .with_observer(|| subscriber.update_if_necessary())
-                            || first_run
+                        if !owner.paused()
+                            && (subscriber.with_observer(|| {
+                                subscriber.update_if_necessary()
+                            }) || first_run)
                         {
                             subscriber.clear_sources(&subscriber);
 
@@ -388,9 +390,10 @@ impl Effect<SyncStorage> {
 
                 async move {
                     while rx.next().await.is_some() {
-                        if subscriber
-                            .with_observer(|| subscriber.update_if_necessary())
-                            || first_run
+                        if !owner.paused()
+                            && (subscriber.with_observer(|| {
+                                subscriber.update_if_necessary()
+                            }) || first_run)
                         {
                             first_run = false;
                             subscriber.clear_sources(&subscriber);
@@ -434,9 +437,10 @@ impl Effect<SyncStorage> {
 
             async move {
                 while rx.next().await.is_some() {
-                    if subscriber
-                        .with_observer(|| subscriber.update_if_necessary())
-                        || first_run
+                    if !owner.paused()
+                        && (subscriber
+                            .with_observer(|| subscriber.update_if_necessary())
+                            || first_run)
                     {
                         first_run = false;
                         subscriber.clear_sources(&subscriber);
@@ -487,9 +491,10 @@ impl Effect<SyncStorage> {
 
                 async move {
                     while rx.next().await.is_some() {
-                        if subscriber
-                            .with_observer(|| subscriber.update_if_necessary())
-                            || first_run
+                        if !owner.paused()
+                            && (subscriber.with_observer(|| {
+                                subscriber.update_if_necessary()
+                            }) || first_run)
                         {
                             subscriber.clear_sources(&subscriber);
 

reactive_graph/src/effect/render_effect.rs

@@ -91,9 +91,11 @@ where
 
                     async move {
                         while rx.next().await.is_some() {
-                            if subscriber.with_observer(|| {
-                                subscriber.update_if_necessary()
-                            }) {
+                            if !owner.paused()
+                                && subscriber.with_observer(|| {
+                                    subscriber.update_if_necessary()
+                                })
+                            {
                                 subscriber.clear_sources(&subscriber);
 
                                 let old_value = mem::take(
@@ -159,8 +161,10 @@ where
 
                 async move {
                     while rx.next().await.is_some() {
-                        if subscriber
-                            .with_observer(|| subscriber.update_if_necessary())
+                        if !owner.paused()
+                            && subscriber.with_observer(|| {
+                                subscriber.update_if_necessary()
+                            })
                         {
                             subscriber.clear_sources(&subscriber);
 

reactive_graph/src/owner.rs

@@ -130,6 +130,7 @@ impl Owner {
                     .and_then(|parent| parent.upgrade())
                     .map(|parent| parent.read().or_poisoned().arena.clone())
                     .unwrap_or_default(),
+                paused: false,
             })),
             #[cfg(feature = "hydration")]
             shared_context,
@@ -163,6 +164,7 @@ impl Owner {
                 children: Default::default(),
                 #[cfg(feature = "sandboxed-arenas")]
                 arena: Default::default(),
+                paused: false,
             })),
             #[cfg(feature = "hydration")]
             shared_context,
@@ -174,8 +176,10 @@ impl Owner {
     /// Creates a new `Owner` that is the child of the current `Owner`, if any.
     pub fn child(&self) -> Self {
         let parent = Some(Arc::downgrade(&self.inner));
+        let mut inner = self.inner.write().or_poisoned();
         #[cfg(feature = "sandboxed-arenas")]
-        let arena = self.inner.read().or_poisoned().arena.clone();
+        let arena = inner.arena.clone();
+        let paused = inner.paused;
         let child = Self {
             inner: Arc::new(RwLock::new(OwnerInner {
                 parent,
@@ -185,15 +189,12 @@ impl Owner {
                 children: Default::default(),
                 #[cfg(feature = "sandboxed-arenas")]
                 arena,
+                paused,
             })),
             #[cfg(feature = "hydration")]
             shared_context: self.shared_context.clone(),
         };
-        self.inner
-            .write()
-            .or_poisoned()
-            .children
-            .push(Arc::downgrade(&child.inner));
+        inner.children.push(Arc::downgrade(&child.inner));
         child
     }
 
@@ -337,6 +338,47 @@ impl Owner {
 
         inner(Box::new(fun))
     }
+
+    /// Pauses the execution of side effects for this owner, and any of its descendants.
+    ///
+    /// If this owner is the owner for an [`Effect`](crate::effect::Effect) or [`RenderEffect`](crate::effect::RenderEffect), this effect will not run until [`Owner::resume`] is called. All children of this effects are also paused.
+    ///
+    /// Any notifications will be ignored; effects that are notified will paused will not run when
+    /// resumed, until they are notified again by a source after being resumed.
+    pub fn pause(&self) {
+        let mut stack = Vec::with_capacity(16);
+        stack.push(Arc::downgrade(&self.inner));
+        while let Some(curr) = stack.pop() {
+            if let Some(curr) = curr.upgrade() {
+                let mut curr = curr.write().or_poisoned();
+                curr.paused = true;
+                stack.extend(curr.children.iter().map(Weak::clone));
+            }
+        }
+    }
+
+    /// Whether this owner has been paused by [`Owner::pause`].
+    pub fn paused(&self) -> bool {
+        self.inner.read().or_poisoned().paused
+    }
+
+    /// Resumes side effects that have been paused by [`Owner::pause`].
+    ///
+    /// All children will also be resumed.
+    ///
+    /// This will *not* cause side effects that were notified while paused to run, until they are
+    /// notified again by a source after being resumed.
+    pub fn resume(&self) {
+        let mut stack = Vec::with_capacity(16);
+        stack.push(Arc::downgrade(&self.inner));
+        while let Some(curr) = stack.pop() {
+            if let Some(curr) = curr.upgrade() {
+                let mut curr = curr.write().or_poisoned();
+                curr.paused = false;
+                stack.extend(curr.children.iter().map(Weak::clone));
+            }
+        }
+    }
 }
 
 #[doc(hidden)]
@@ -363,6 +405,7 @@ pub(crate) struct OwnerInner {
     pub children: Vec<Weak<RwLock<OwnerInner>>>,
     #[cfg(feature = "sandboxed-arenas")]
     arena: Arc<RwLock<ArenaMap>>,
+    paused: bool,
 }
 
 impl Debug for OwnerInner {

reactive_graph/src/signal/subscriber_traits.rs

@@ -55,7 +55,7 @@ impl<T: AsSubscriberSet + DefinedAt> ReactiveNode for T {
 
     fn mark_subscribers_check(&self) {
         if let Some(inner) = self.as_subscriber_set() {
-            let subs = inner.borrow().write().unwrap().take();
+            let subs = inner.borrow().read().unwrap().clone();
             for sub in subs {
                 sub.mark_dirty();
             }

reactive_graph/tests/effect.rs

@@ -196,3 +196,62 @@ async fn recursive_effect_runs_recursively() {
         })
         .await;
 }
+
+#[cfg(feature = "effects")]
+#[tokio::test]
+async fn paused_effect_pauses() {
+    use imports::*;
+    use reactive_graph::owner::StoredValue;
+
+    _ = Executor::init_tokio();
+    let owner = Owner::new();
+    owner.set();
+
+    task::LocalSet::new()
+        .run_until(async {
+            let a = RwSignal::new(-1);
+
+            // simulate an arbitrary side effect
+            let runs = StoredValue::new(0);
+
+            let owner = StoredValue::new(None);
+
+            Effect::new({
+                move || {
+                    *owner.write_value() = Owner::current();
+
+                    let _ = a.get();
+                    *runs.write_value() += 1;
+                }
+            });
+
+            Executor::tick().await;
+            assert_eq!(runs.get_value(), 1);
+
+            println!("setting to 1");
+            a.set(1);
+
+            Executor::tick().await;
+            assert_eq!(runs.get_value(), 2);
+
+            println!("pausing");
+            owner.get_value().unwrap().pause();
+
+            println!("setting to 2");
+            a.set(2);
+
+            Executor::tick().await;
+            assert_eq!(runs.get_value(), 2);
+
+            println!("resuming");
+            owner.get_value().unwrap().resume();
+
+            println!("setting to 3");
+            a.set(3);
+
+            Executor::tick().await;
+            println!("checking value");
+            assert_eq!(runs.get_value(), 3);
+        })
+        .await
+}