@@ -392,16 +392,25 @@ def _add_group(
392392 return
393393 visited .add (group )
394394 parent = group .parent
395- if parent is not None and parent in group_set :
396- self ._add_group (parent , group_set , new_root_groups , visited , parent_task )
395+ if parent is not None :
396+ if parent in group_set :
397+ self ._add_group (
398+ parent , group_set , new_root_groups , visited , parent_task
399+ )
400+
401+ # Discard newly discovered children of removed parents.
402+ if parent not in self ._group_nodes :
403+ return
397404
398405 self ._group_nodes [group ] = _GroupNode ()
399406
400407 if parent_task is None and not parent :
401408 new_root_groups .append (group )
402409 elif parent :
403410 parent_node = self ._group_nodes .get (parent )
404- if parent_node :
411+ # defensive guard mirroring the JS version; children of
412+ # removed parents have already been discarded above
413+ if parent_node : # pragma: no branch
405414 parent_node .child_groups .append (group )
406415
407416 def _add_task (self , task : WorkTask ) -> None :
@@ -421,7 +430,9 @@ def _add_streams(
421430 if not parent_task :
422431 return streams
423432 task_node = self ._task_nodes .get (parent_task )
424- if task_node :
433+ # defensive guard mirroring the JS version; work of removed
434+ # tasks is ignored before it is integrated
435+ if task_node : # pragma: no branch
425436 task_node .child_streams .extend (streams )
426437 return []
427438
@@ -485,14 +496,18 @@ def _start_group(self, group: Group) -> None:
485496 def _start_task (self , task : WorkTask ) -> None :
486497 """Start a task, pushing a graph event when its computation settles."""
487498 task_nodes = self ._task_nodes
488- if task in task_nodes :
489- return
490- task_nodes [task ] = _TaskNode ()
499+ already_started = task in task_nodes
500+ if not already_started :
501+ task_nodes [task ] = _TaskNode ()
491502 try :
492503 result = task .computation .result ()
493- except Exception as error :
504+ except ( Exception , CancelledError ) as error :
494505 self ._push (_TaskFailure (task , error ))
495506 return
507+ if already_started :
508+ # Pending tasks already have a result handler. Successful results are
509+ # handled once for all owning groups, including those not yet started.
510+ return
496511 if is_awaitable (result ):
497512 future : Future [WorkResult ] = ensure_future (result )
498513
@@ -560,8 +575,9 @@ def _task_success(
560575 task , result = graph_event
561576 value , work = result
562577 task_node = self ._task_nodes .get (task )
563- if task_node :
564- task_node .value = value
578+ if not task_node :
579+ return []
580+ task_node .value = value
565581 self ._maybe_integrate_work (work , task )
566582
567583 group_events : list [GroupValuesEvent | GroupSuccessEvent ] = []
@@ -604,11 +620,13 @@ def _maybe_finish_group(
604620 def _task_failure (self , graph_event : _TaskFailure ) -> Sequence [GroupFailureEvent ]:
605621 """Handle a task failure, removing all groups it belongs to."""
606622 task , error = graph_event
607- self ._task_nodes .pop (task , None )
623+ if task not in self ._task_nodes :
624+ return []
608625 group_failure_events : list [GroupFailureEvent ] = []
626+ root_groups = self ._root_groups
609627 for group in task .groups :
610628 group_node = self ._group_nodes .get (group )
611- if group_node :
629+ if group_node and group in root_groups :
612630 group_failure_events .append (
613631 self ._finish_group_failure (group , group_node , error )
614632 )
@@ -676,19 +694,21 @@ def _finish_group_failure(
676694 ) -> GroupFailureEvent :
677695 """Finish a failed group, removing its whole subtree."""
678696 self ._remove_group (group , group_node )
679- self ._root_groups .pop (group , None )
680697 return GroupFailureEvent (group , error )
681698
682699 def _remove_group (self , group : Group , group_node : _GroupNode ) -> None :
683700 """Remove a group with its tasks and child groups from the graph."""
701+ self ._root_groups .pop (group , None )
684702 group_nodes = self ._group_nodes
685703 del group_nodes [group ]
686704 for task in list (group_node .tasks ):
687705 if all (task_group not in group_nodes for task_group in task .groups ):
688706 self ._remove_task (task )
689707 for child_group in group_node .child_groups :
690708 child_group_node = group_nodes .get (child_group )
691- if child_group_node :
709+ # defensive guard mirroring the JS version; only root groups
710+ # fail, so the children of a removed group still exist
711+ if child_group_node : # pragma: no branch
692712 self ._remove_group (child_group , child_group_node )
693713
694714 def _remove_task (self , task : WorkTask ) -> None :
0 commit comments