remove handle app update

wipe_task now happens in complete task, the moral successor of handle_app_update

but it now happens *before* the user observes a task complete.

this makes the task be able to be removed from the tasks table before the
user observes the future is complete.

you might say thats a new race condition: the user cannot see the task table entry
at the moment of completion. but that was never strongly guaranteed: the callback
to remove the entry could happen before or after the user observed the completion.

now theres a stronger assertion: it will definitely happen before the user observes
task completion via an AppFuture.

TODO: are there already tests about this?

When i removed wipe_task calls entirely, no per-config test failed...

parsl/tests/test_python_apps/test_garbage_collect.py
  - this test was very slightly racey before and this was noticeable
    in some race-condition fuzzing work I've done

Author: benclifford

parsl/dataflow/dflow.py

@@ -502,29 +502,6 @@ def handle_join_update(self, task_record: TaskRecord, inner_app_future: Optional
 
             self._log_std_streams(task_record)
 
-    def handle_app_update(self, task_record: TaskRecord, future: AppFuture) -> None:
-        """This function is called as a callback when an AppFuture
-        is in its final state.
-
-        It will trigger post-app processing such as checkpointing.
-
-        Args:
-             task_record : Task record
-             future (Future) : The relevant app future (which should be
-                 consistent with the task structure 'app_fu' entry
-
-        """
-
-        task_id = task_record['id']
-
-        if not task_record['app_fu'].done():
-            logger.error("Internal consistency error: app_fu is not done for task {}".format(task_id))
-        if not task_record['app_fu'] == future:
-            logger.error("Internal consistency error: callback future is not the app_fu in task structure, for task {}".format(task_id))
-
-        self.wipe_task(task_id)
-        return
-
     def _complete_task_result(self, task_record: TaskRecord, new_state: States, result: Any) -> None:
         """Set a task into a completed state
         """
@@ -541,6 +518,8 @@ def _complete_task_result(self, task_record: TaskRecord, new_state: States, resu
 
         self._send_task_log_info(task_record)
 
+        self.wipe_task(task_record['id'])
+
         with task_record['app_fu']._update_lock:
             task_record['app_fu'].set_result(result)
 
@@ -560,6 +539,8 @@ def _complete_task_exception(self, task_record: TaskRecord, new_state: States, e
 
         self._send_task_log_info(task_record)
 
+        self.wipe_task(task_record['id'])
+
         with task_record['app_fu']._update_lock:
             task_record['app_fu'].set_exception(exception)
 
@@ -1051,7 +1032,6 @@ def submit(self,
                                                               task_record['func_name'],
                                                               waiting_message))
 
-        app_fu.add_done_callback(partial(self.handle_app_update, task_record))
         self.update_task_state(task_record, States.pending)
         logger.debug("Task {} set to pending state with AppFuture: {}".format(task_id, task_record['app_fu']))