ORM: ProcessNode.is_valid_cache is False for unsealed nodes

When a `ProcessNode` is not yet sealed, it has not officially been
terminated. At this point it cannot yet be considered a valid cache
source. However, this condition was not considered by the property
`ProcessNode.is_valid_cache`.

This bug manifested itself in very rare situations where a race
condition could lead to a process being cached from an unsealed node.
When a node is stored from the cache, it copies all the attribute except
for the sealed key and adds the outputs. The sealing is then left to the
engine which will complete the cached process as if it had run normally.
The problem arises when the cache source was not yet sealed, and so the
outputs had not yet been added. The cached node will then miss the
output nodes.

aiida/orm/nodes/process/process.py

@@ -40,7 +40,7 @@ def is_valid_cache(self) -> bool:
 
         :returns: True if this process node is valid to be used for caching, False otherwise
         """
-        if not (super().is_valid_cache and self._node.is_finished):
+        if not (super().is_valid_cache and self._node.is_finished and self._node.is_sealed):
             return False
 
         try:

tests/orm/nodes/process/test_process.py

@@ -1,7 +1,12 @@
 # -*- coding: utf-8 -*-
+# pylint: disable=redefined-outer-name
 """Tests for :mod:`aiida.orm.nodes.process.process`."""
-from aiida.engine import ExitCode
+import pytest
+
+from aiida.engine import ExitCode, ProcessState
+from aiida.orm.nodes.caching import NodeCaching
 from aiida.orm.nodes.process.process import ProcessNode
+from aiida.orm.nodes.process.workflow import WorkflowNode
 
 
 def test_exit_code():
@@ -14,3 +19,56 @@ def test_exit_code():
 
     node.set_exit_message('I am a teapot')
     assert node.exit_code == ExitCode(418, 'I am a teapot')
+
+
+@pytest.fixture
+@pytest.mark.usefixtures('aiida_profile')
+def process_nodes():
+    """Return a list of tuples of a process node and whether they should be a valid cache source."""
+    entry_point = 'aiida.calculations:core.arithmetic.add'
+
+    node_invalid_cache_extra = WorkflowNode(label='node_invalid_cache_extra')
+    node_invalid_cache_extra.set_process_state(ProcessState.FINISHED)
+    node_invalid_cache_extra.base.extras.set(NodeCaching._VALID_CACHE_KEY, False)  # pylint: disable=protected-access
+
+    node_no_process_class = WorkflowNode(label='node_no_process_class')
+    node_no_process_class.set_process_state(ProcessState.FINISHED)
+
+    node_invalid_process_class = WorkflowNode(label='node_invalid_process_class', process_type='aiida.calculations:no')
+    node_invalid_process_class.set_process_state(ProcessState.FINISHED)
+
+    node_excepted = WorkflowNode(label='node_excepted', process_type=entry_point)
+    node_excepted.set_process_state(ProcessState.EXCEPTED)
+
+    node_excepted_stored = WorkflowNode(label='node_excepted_stored', process_type=entry_point)
+    node_excepted_stored.set_process_state(ProcessState.EXCEPTED)
+
+    node_excepted_sealed = WorkflowNode(label='node_excepted_sealed', process_type=entry_point)
+    node_excepted_sealed.set_process_state(ProcessState.EXCEPTED)
+
+    node_finished = WorkflowNode(label='node_finished', process_type=entry_point)
+    node_finished.set_process_state(ProcessState.FINISHED)
+
+    node_finished_stored = WorkflowNode(label='node_finished_stored', process_type=entry_point)
+    node_finished_stored.set_process_state(ProcessState.FINISHED)
+
+    node_finished_sealed = WorkflowNode(label='node_finished_sealed', process_type=entry_point)
+    node_finished_sealed.set_process_state(ProcessState.FINISHED)
+
+    return (
+        (node_invalid_cache_extra.store().seal(), False),
+        (node_no_process_class.store().seal(), False),
+        (node_invalid_process_class.store().seal(), False),
+        (node_excepted, False),
+        (node_excepted_stored.store(), False),
+        (node_excepted_sealed.store().seal(), False),
+        (node_finished, False),
+        (node_finished_stored.store(), False),
+        (node_finished_sealed.store().seal(), True),
+    )
+
+
+def test_is_valid_cache(process_nodes):
+    """Test the :meth:`aiida.orm.nodes.process.process.ProcessNode.is_valid_cache` property."""
+    for node, is_valid_cache in process_nodes:
+        assert node.is_valid_cache == is_valid_cache, node

tests/orm/nodes/test_repository.py

@@ -20,7 +20,7 @@ def cacheable_node():
     node = CalcJobNode(process_type='aiida.calculations:core.arithmetic.add')
     node.set_process_state(ProcessState.FINISHED)
     node.base.repository.put_object_from_bytes(b'content', 'relative/path')
-    node.store()
+    node.store().seal()
     assert node.base.caching.is_valid_cache
 
     return node