Tell memoizer the task result earlier, to fix race condition #3762 (#3979)

Prior to this PR, the memoizer was told about the result in the
AppFuture result callback.

By the time this happens, the user workflow code can also observe that
the task is completed and also observe that the memoizer has not got a
copy of the result: for example, by invoking the same app again
immediately and seeing a duplicate execution.

The AppFuture callback shouldn't have anything in it that is related to
user-observable task completion, but this PR does not remove other stuff
in there.

This PR moves update_memo earlier, to before setting the AppFuture
result, so that the memoizer is strictly before future changes state.

# Changed Behaviour

Checkpointing behaviour will now be slightly less racy. In situations
where that race condition would have fired, the user will now see fewer
apps actually executed, and memoized results used instead. This is, I
think, a very rare situation.

# Fixes

#3762 

## Type of change

- Bug fix

Author: benclifford

parsl/dataflow/dflow.py

@@ -375,6 +375,7 @@ def handle_exec_update(self, task_record: TaskRecord, future: Future) -> None:
                 logger.info("Task {} failed due to dependency failure so skipping retries".format(task_id))
                 task_record['time_returned'] = datetime.datetime.now()
                 self._send_task_log_info(task_record)
+                self.memoizer.update_memo(task_record)
                 with task_record['app_fu']._update_lock:
                     task_record['app_fu'].set_exception(e)
 
@@ -400,6 +401,7 @@ def handle_exec_update(self, task_record: TaskRecord, future: Future) -> None:
                 self.update_task_state(task_record, States.failed)
                 task_record['time_returned'] = datetime.datetime.now()
                 self._send_task_log_info(task_record)
+                self.memoizer.update_memo(task_record)
                 with task_record['app_fu']._update_lock:
                     task_record['app_fu'].set_exception(e)
 
@@ -446,6 +448,7 @@ def handle_exec_update(self, task_record: TaskRecord, future: Future) -> None:
                         self.update_task_state(task_record, States.failed)
                         task_record['time_returned'] = datetime.datetime.now()
                         self._send_task_log_info(task_record)
+                        self.memoizer.update_memo(task_record)
                         with task_record['app_fu']._update_lock:
                             task_record['app_fu'].set_exception(
                                 TypeError(f"join_app body must return a Future or list of Futures, got {joinable} of type {type(joinable)}"))
@@ -521,6 +524,7 @@ def handle_join_update(self, task_record: TaskRecord, inner_app_future: Optional
 
                 self.update_task_state(task_record, States.failed)
                 task_record['time_returned'] = datetime.datetime.now()
+                self.memoizer.update_memo(task_record)
                 with task_record['app_fu']._update_lock:
                     task_record['app_fu'].set_exception(e)
 
@@ -561,8 +565,6 @@ def handle_app_update(self, task_record: TaskRecord, future: AppFuture) -> None:
         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.memoizer.update_memo(task_record)
-
         # Cover all checkpointing cases here:
         # Do we need to checkpoint now, or queue for later,
         # or do nothing?
@@ -591,6 +593,7 @@ def _complete_task(self, task_record: TaskRecord, new_state: States, result: Any
         logger.info(f"Task {task_record['id']} completed ({old_state.name} -> {new_state.name})")
         task_record['time_returned'] = datetime.datetime.now()
 
+        self.memoizer.update_memo(task_record)
         with task_record['app_fu']._update_lock:
             task_record['app_fu'].set_result(result)