Prevent duplicate updates

src/graph.luau

@@ -25,9 +25,7 @@ local function efn(err: string)
     local trace = debug.traceback(err, 2)
 
     if string.find(err, "^effect error stacktrace") then -- if effect error is nested
-        trace = string.gsub("    " .. trace, "\n", function() -- indent entire error
+        trace = string.gsub("    " .. trace, "\n", "\n    ") -- indent entire error
-            return "\n    "
-        end)
     end
 
     trace ..= "\nsource update stacktrace:"
@@ -144,9 +142,19 @@ local function destroy_owned<T>(node: Node<T>)
     end
 end
 
-local update_queue = { n = 0 } :: { n: number, [number]: Node<any> }
+local update_queue_n = 0
+local in_queue = {}
+local update_queue = {} :: { [number]: Node<any> }
+local node_deps_left = {} -- any node that goes into the update queue needs to store how many dependencies it's waiting on
 
-local function evaluate_node<T>(node: Node<T>)
+local function reset_queue()
+    update_queue_n = 0
+    table.clear(in_queue)
+    table.clear(update_queue)
+    table.clear(node_deps_left)
+end
+
+local function evaluate_node<T>(node: Node<T>, continue_on_error: boolean) -- if continue_on_error, the calling function wishes to continue even if the update fails (without using pcall)
     if flags.strict then
         if table.find(scopes, node) then
             error("a scope, that should rerun due to the update of a source, is already active", 0)
@@ -165,9 +173,13 @@ local function evaluate_node<T>(node: Node<T>)
             pop_scope()
 
             if not ok then
-                table.clear(update_queue)
+                local msg = debug.traceback(`effect error: {new_value}`, 2)
-                update_queue.n = 0
+                if continue_on_error then
-                error(`effect error stacktrace\n{new_value :: string}`, 0)
+                    task.spawn(error, msg, 0)
+                    return false
+                else
+                    error(msg, 2)
+                end
             end
 
             node.cache = new_value :: T
@@ -181,13 +193,17 @@ local function evaluate_node<T>(node: Node<T>)
         destroy_owned(node)
 
         push_scope(node)
-        local ok, new_value = pcall(node.effect :: (T) -> T, node.cache)
+        local ok, new_value = pcall(node.effect :: (T) -> T, cur_value)
         pop_scope()
 
         if not ok then
-            table.clear(update_queue)
+            local msg = debug.traceback(`effect error: {new_value}`, 2)
-            update_queue.n = 0
+            if continue_on_error then
-            error(`effect error:\n{new_value}\n`, 0)
+                task.spawn(error, msg, 0)
+                return false
+            else
+                error(msg, 2)
+            end
         end
 
         node.cache = new_value
@@ -195,58 +211,105 @@ local function evaluate_node<T>(node: Node<T>)
     end
 end
 
-local function queue_children_for_update<T>(node: SourceNode<T>)
+local function add_dependency_counts(node)
-    local i = update_queue.n
+    for i = 1, #node do
-    while node[1] do
+        local child = node[i]
-        i += 1
+        if not child.owner then continue end -- we won't be updating this anyway
-        update_queue[i] = node[1]
+        local deps_left = node_deps_left[child]
-        unparent(node[1])
+        if deps_left then
+            node_deps_left[child] = deps_left + 1
+        else
+            node_deps_left[child] = 1
+            add_dependency_counts(child) -- only recurse if we haven't seen this node before
+        end
+    end
+end
+local function remove_dependency_counts(node) -- to be called when updating a node if its value remains unchanged
+    for i = 1, #node do
+        local child = node[i]
+        if not node.owner then continue end -- we won't be updating this anyway
+        node_deps_left[child] -= 1
+        remove_dependency_counts(child)
+    end
+end
+
+local function queue_children_for_update<T>(node: SourceNode<T>)
+    for i = 1, #node do
+        local child = node[i]
+        node_deps_left[child] -= 1
+        if not in_queue[child] then
+            in_queue[child] = true
+            update_queue_n += 1
+            update_queue[update_queue_n] = child
+        end
     end
-    update_queue.n = i
 end
 
 local function get_update_queue_length()
-    return update_queue.n
+    return update_queue_n
 end
 
 local function flush_update_queue(from: number)
-    local i = from + 1
+    while true do
-    while i <= update_queue.n do
+        local i = from
+        if i >= update_queue_n then return end -- nothing more to do
+        local changed = false
+        local moveTo1 = from -- 1 less than where to store an update if we can't process it yet
+        while i < update_queue_n do
+            i += 1
             local node = update_queue[i]
             -- assert(node.effect)
-
+            if node.owner then
-        if node.owner and evaluate_node(node) then
+                local deps_left = node_deps_left[node]
+                if deps_left > 0 then -- waiting on more dependencies
+                    moveTo1 += 1
+                    if i ~= moveTo1 then
+                        update_queue[moveTo1] = node
+                        update_queue[i] = false :: any
+                    end
+                    continue
+                end
+                changed = true
+                -- Reset in_queue and node_deps_left before evaluation in case the node's depedencies are changed during evaluation (which usually implies an infinite loop, but we have explicit errors for this case)
+                in_queue[node] = nil
+                node_deps_left[node] = nil
+                unparent(node)
+                if evaluate_node(node, true) then
                     queue_children_for_update(node)
+                else
+                    remove_dependency_counts(node)
+                end
+            else -- otherwise node was cleaned up
+                in_queue[node] = nil
+                node_deps_left[node] = nil
             end
 
             update_queue[i] = false :: any
-        i += 1
         end
-    
+        if moveTo1 == 0 then -- no pending updates
-    update_queue.n = from
+            reset_queue()
+            return
+        else
+            update_queue_n = moveTo1
+            if not changed then -- Can occur in recursive updates (where 'from' is > 0)
+                if from == 0 then
+                    warn("Some nodes failed to update", update_queue, node_deps_left, debug.traceback())
+                    reset_queue()
+                end
+                return
+            end
+        end
+    end
 end
 
 local function update_descendants<T>(root: SourceNode<T>)
-    local n0 = update_queue.n
+    add_dependency_counts(root)
+    local n0 = update_queue_n
     queue_children_for_update(root)
 
     if flags.batch then return end
 
-    local i = n0 + 1
+    flush_update_queue(n0)
-    while i <= update_queue.n do
-        local node = update_queue[i]
-        --assert(node.effect)
-
-        -- check if node is still owned in case destroyed after queued
-        if node.owner and evaluate_node(node) then
-            queue_children_for_update(node)
-        end
-
-        update_queue[i] = false :: any -- false instead of nil to avoid sparse
-        i += 1
-    end
-
-    update_queue.n = n0
 end
 
 local function push_scope_as_child_of<T>(node: SourceNode<T>)

test/tests.luau

@@ -2847,14 +2847,14 @@ TEST("graph edge cases", wrap_root(function()
         CHECK(count.b == 2)
         CHECK(count.c == 2)
         CHECK(count.d == 2)
-        CHECK(count.e == 3) -- todo: redundant re-eval
+        CHECK(count.e == 2)
         CHECK(e() == 4)
 
         a(3)
         CHECK(count.b == 2)
         CHECK(count.c == 3)
         CHECK(count.d == 3)
-        CHECK(count.e == 4)
+        CHECK(count.e == 3)
         CHECK(e() == 12)
     end
 

todo.md

@@ -1,4 +1,3 @@
 # todo
 
 - improve error traces
-- prevent redundant re-eval of nodes in a complex diamond graph