CalcJob: Assign outputs from node in case of cache hit (#5995)

If a `CalcJob` is launched and a cache hit occurs, the outputs of the
cache source are attached automatically to the process node of the
`CalcJob` instance in `Node.store`. The process execution itself,
shortcuts in `CalcJob.run` to directly return the exit status instead of
going to the upload stage.

This shortcut, however, also means that the `Parser` won't be called
which normally is responsible for adding the outputs; to the process
*node* but also to the *process* itself. By circumventing this, the
outputs mapping on the process instance will be empty. This is a problem
if the process was run as opposed to submitted. In this case, the
`Runner.run` method will return the results by taking it from the
`Process.outputs` method, but this will be empty.

Ultimately, this means that when a user *runs* a `CalcJob` that hits the
cache, the returned `results` dictionary will always be empty. To fix
it, the `CalcJob.run` method is updated to populate the `_outputs`
attribute of the process instance with the outputs that are stored on
the associated process node. Now the run method will return the correct
results dictionary.

aiida/engine/processes/calcjobs/calcjob.py

@@ -552,6 +552,11 @@ def run(self) -> Union[plumpy.process_states.Stop, int, plumpy.process_states.Wa
         # been overridden by the engine to `Running` so we cannot check that, but if the `exit_status` is anything other
         # than `None`, it should mean this node was taken from the cache, so the process should not be rerun.
         if self.node.exit_status is not None:
+            # Normally the outputs will be attached to the process by a ``Parser``, if defined in the inputs. But in
+            # this case, the parser will not be called. The outputs will already have been added to the process node
+            # though, so all that needs to be done here is just also assign them to the process instance. This such that
+            # when the process returns its results, it returns the actual outputs and not an empty dictionary.
+            self._outputs = self.node.get_outgoing(link_type=LinkType.CREATE).nested()  # pylint: disable=attribute-defined-outside-init
             return self.node.exit_status
 
         # Launch the upload operation

tests/engine/test_runners.py

@@ -15,9 +15,11 @@
 import plumpy
 import pytest
 
-from aiida.engine import Process
+from aiida.calculations.arithmetic.add import ArithmeticAddCalculation
+from aiida.engine import Process, launch
 from aiida.manage import get_manager
-from aiida.orm import Str, WorkflowNode
+from aiida.manage.caching import enable_caching
+from aiida.orm import Int, Str, WorkflowNode
 
 
 @pytest.fixture
@@ -80,3 +82,24 @@ def test_submit_args(runner):
     """
     with pytest.raises(TypeError, match=r'takes 2 positional arguments but 3 were given'):
         runner.submit(Proc, {'a': Str('input')})
+
+
+def test_run_return_value_cached(aiida_local_code_factory):
+    """Test that :meth:`aiida.engine.runners.Runner._run` return process results even when cached.
+
+    Regression test for https://github.com/aiidateam/aiida-core/issues/5994.
+    """
+    inputs = {
+        'code': aiida_local_code_factory('core.arithmetic.add', '/bin/bash'),
+        'x': Int(1),
+        'y': Int(-2),
+    }
+    results_source, node_source = launch.run_get_node(ArithmeticAddCalculation, **inputs)
+    assert node_source.is_valid_cache
+
+    with enable_caching():
+        results_cached, node_cached = launch.run_get_node(ArithmeticAddCalculation, **inputs)
+
+    assert node_cached.base.caching.get_cache_source() == node_source.uuid
+    assert sorted(results_cached.keys()) == sorted(results_source.keys())
+    assert sorted(results_cached.keys()) == ['remote_folder', 'retrieved', 'sum']