[FLOC-4452] Delay invariant checks until all diff operations have been applied

flocker/control/_diffing.py

@@ -7,6 +7,8 @@
 flocker configuration or the flocker state.
 """
 
+from eliot import MessageType, Field
+
 from pyrsistent import (
     PClass,
     PMap,
@@ -15,6 +17,7 @@
     pvector,
     pvector_field,
 )
+from pyrsistent._transformations import _get
 
 from zope.interface import Interface, implementer
 
@@ -76,7 +79,9 @@ class _Set(PClass):
     value = field()
 
     def apply(self, obj):
-        return obj.transform(self.path, self.value)
+        return obj.transform(
+            self.path[:-1], lambda o: o.set(self.path[-1], self.value)
+        )
 
 
 @implementer(_IDiffChange)
@@ -95,6 +100,152 @@ def apply(self, obj):
         return obj.transform(self.path, lambda x: x.add(self.item))
 
 
+_sentinel = object()
+
+
+class _EvolverProxy(object):
+    """
+    This attempts to bunch all the diff operations for a particular object into
+    a single transaction so that related attributes can be ``set`` without
+    triggering an in invariant error.
+    Additionally, the leaf nodes are persisted first and in isolation, so as
+    not to trigger invariant errors in ancestor nodes.
+    """
+    def __init__(self, original):
+        """
+        :param PClass original: The root object to which transformations will
+            be applied.
+        """
+        self._original = original
+        self._evolver = original.evolver()
+        self._children = {}
+        self._operations = []
+
+    def _child(self, segment):
+        child = self._children.get(segment)
+        if child is not None:
+            return child
+        child = _get(self._original, segment, _sentinel)
+        if child is _sentinel:
+            raise KeyError(
+                'Segment not found in path. '
+                'Parent: {}, '
+                'Segment: {}'.format(self, segment)
+            )
+        proxy_for_child = _EvolverProxy(child)
+        self._children[segment] = proxy_for_child
+        return proxy_for_child
+
+    def transform(self, path, operation):
+        """
+        Traverse each segment of ``path`` to create a hierarchy of
+        ``_EvolverProxy`` objects and perform the ``operation`` on the
+        resulting leaf proxy object. This will infact perform the operation on
+        an evolver of the original Pyrsistent object.
+
+        :param PVector path: The path relative to ``original`` which will be
+            operated on.
+        :param callable operation: A function to be applied to an evolver of
+             the object at ``path``
+        :returns: ``self``
+        """
+        target = self
+        for segment in path:
+            target = target._child(segment)
+        operation(target)
+        return self
+
+    def add(self, item):
+        """
+        Add ``item`` to the ``original`` ``Pset`` or if the item is itself a
+        Pyrsistent object, add a new proxy for that item so that further
+        operations can be performed on it without triggering invariant checks
+        until the tree is finally committed.
+
+        :param item: An object to be added to the ``PSet`` wrapped by this
+            proxy.
+        :returns: ``self``
+        """
+        if hasattr(item, 'evolver'):
+            self._children[item] = _EvolverProxy(item)
+        else:
+            self._evolver.add(item)
+        return self
+
+    def set(self, key, item):
+        """
+        Set the ``item`` in an evolver of the ``original`` ``PMap`` or
+        ``PClass`` or if the item is itself a Pyrsistent object, add a new
+        proxy for that item so that further operations can be performed on it
+        without triggering invariant checks until the tree is finally
+        committed.
+
+        :param item: An object to be added or set on the ``PMap`` wrapped by
+            this proxy.
+        :returns: ``self``
+        """
+        if hasattr(item, 'evolver'):
+            # This will replace any existing proxy.
+            self._children[key] = _EvolverProxy(item)
+        else:
+            self._evolver.set(key, item)
+        return self
+
+    def remove(self, item):
+        """
+        Remove the ``item`` in an evolver of the ``original`` ``PMap``,
+        ``PClass``, or ``PSet`` and if the item is an uncommitted
+        ``_EvolverProxy`` remove it from the list of children so that the item
+        is not persisted when the structure is finally committed.
+
+        :param item: The object to be removed from the wrapped ``PSet`` or the
+            key to be removed from the wrapped ``PMap``
+        :returns: ``self``
+        """
+        self._children.pop(item, None)
+        # Attempt to remove the item from the evolver too.  It may be something
+        # that was replaced rather than added by a previous ``set`` operation.
+        try:
+            self._evolver.remove(item)
+        except KeyError:
+            pass
+        return self
+
+    def commit(self):
+        """
+        Persist all the changes made to the descendants of this structure, then
+        persist the resulting sub-objects and local changes to this root object
+        and finally return the resulting immutable structure.
+
+        :returns: The updated and persisted version of ``original``.
+        """
+        for segment, child_evolver_proxy in self._children.items():
+            child = child_evolver_proxy.commit()
+            # XXX this is ugly. Perhaps have a separate proxy for PClass, PMap
+            # and PSet collections
+            if hasattr(self._evolver, 'set'):
+                self._evolver.set(segment, child)
+            else:
+                self._evolver.add(child)
+        return self._evolver.persistent()
+
+
+TARGET_OBJECT = Field(
+    u"target_object", repr,
+    u"The object to which the diff was applied."
+)
+CHANGES = Field(
+    u"changes", repr,
+    u"The changes being applied."
+)
+
+DIFF_COMMIT_ERROR = MessageType(
+    u"flocker:control:Diff:commit_error",
+    [TARGET_OBJECT, CHANGES],
+    u"The target and changes that failed to apply."
+)
+
+
 @implementer(_IDiffChange)
 class Diff(PClass):
     """
@@ -111,9 +262,23 @@ class Diff(PClass):
     changes = pvector_field(object)
 
     def apply(self, obj):
+        proxy = _EvolverProxy(original=obj)
         for c in self.changes:
-            obj = c.apply(obj)
-        return obj
+            if len(c.path) > 0:
+                proxy = c.apply(proxy)
+            else:
+                assert type(c) is _Set
+                proxy = _EvolverProxy(original=c.value)
+        try:
+            return proxy.commit()
+        except:
+            # Imported here to avoid circular dependencies.
+            from ._persistence import wire_encode
+            DIFF_COMMIT_ERROR(
+                target_object=wire_encode(obj),
+                changes=wire_encode(self.changes),
+            ).write()
+            raise
 
 
 def _create_diffs_for_sets(current_path, set_a, set_b):

flocker/control/test/test_diffing.py

@@ -7,16 +7,18 @@
 from json import dumps
 from uuid import uuid4
 
+from eliot.testing import capture_logging, assertHasMessage
 from hypothesis import given
 import hypothesis.strategies as st
-from pyrsistent import PClass, field, pmap, pset
+from pyrsistent import PClass, field, pmap, pset, InvariantException
 
-from .._diffing import create_diff, compose_diffs
+from .._diffing import create_diff, compose_diffs, DIFF_COMMIT_ERROR
 from .._persistence import wire_encode, wire_decode
 from .._model import Node, Port
 from ..testtools import (
     application_strategy,
     deployment_strategy,
+    node_strategy,
     related_deployments_strategy
 )
 
@@ -183,7 +185,6 @@ def test_different_objects(self):
         object_a = DiffTestObj(a=pset(xrange(1000)))
         object_b = pmap({'1': 34})
         diff = create_diff(object_a, object_b)
-
         self.assertThat(
             wire_decode(wire_encode(diff)).apply(object_a),
             Equals(object_b)
@@ -202,3 +203,140 @@ def test_different_uuids(self):
             wire_decode(wire_encode(diff)).apply(object_a),
             Equals(object_b)
         )
+
+
+class DiffTestObjInvariant(PClass):
+    """
+    Simple pyrsistent object with an invariant.
+    """
+    _perform_invariant_check = True
+    a = field()
+    b = field()
+
+    def __invariant__(self):
+        if self._perform_invariant_check and self.a == self.b:
+            return (False, "a must not equal b")
+        else:
+            return (True, "")
+
+
+class InvariantDiffTests(TestCase):
+    """
+    Tests for creating and applying diffs to objects with invariant checks.
+    """
+    def test_straight_swap(self):
+        """
+        A diff composed of two separate ``set`` operations can be applied to an
+        object without triggering an invariant exception.
+        """
+        o1 = DiffTestObjInvariant(
+            a=1,
+            b=2,
+        )
+        o2 = DiffTestObjInvariant(
+            a=2,
+            b=1,
+        )
+        diff = create_diff(o1, o2)
+        self.assertEqual(2, len(diff.changes))
+        self.assertEqual(
+            o2,
+            diff.apply(o1)
+        )
+
+    def test_deep_swap(self):
+        """
+        A diff composed of two separate ``set`` operations can be applied to a
+        nested object without triggering an invariant exception.
+        """
+        a = DiffTestObjInvariant(
+            a=1,
+            b=2,
+        )
+        b = DiffTestObjInvariant(
+            a=3,
+            b=4,
+        )
+        o1 = DiffTestObjInvariant(
+            a=a,
+            b=b,
+        )
+        o2 = o1.transform(
+            ['a'],
+            lambda o: o.evolver().set('a', 2).set('b', 1).persistent()
+        )
+        diff = create_diff(o1, o2)
+
+        self.assertEqual(
+            o2,
+            diff.apply(o1)
+        )
+
+    @capture_logging(assertHasMessage, DIFF_COMMIT_ERROR)
+    def test_error_logging(self, logger):
+        """
+        Failures while applying a diff emit a log message containing the full
+        diff.
+        """
+        o1 = DiffTestObjInvariant(
+            a=1,
+            b=2,
+        )
+        DiffTestObjInvariant._perform_invariant_check = False
+        o2 = o1.set('b', 1)
+        DiffTestObjInvariant._perform_invariant_check = True
+        diff = create_diff(o1, o2)
+        self.assertRaises(
+            InvariantException,
+            diff.apply,
+            o1,
+        )
+
+    def test_application_add(self):
+        """
+        A diff on a Node, which *adds* and application with a volume *and* the
+        manifestation for the volume, can be applied without triggering an
+        invariant error on the Node.
+        """
+        node2 = node_strategy(
+            min_number_of_applications=1,
+            stateful_applications=True,
+        ).example()
+        application = node2.applications.values()[0]
+        node1 = node2.transform(
+            ['applications'],
+            lambda o: o.remove(application.name)
+        ).transform(
+            ['manifestations'],
+            lambda o: o.remove(application.volume.manifestation.dataset_id)
+        )
+        diff = create_diff(node1, node2)
+        self.assertEqual(
+            node2,
+            diff.apply(node1),
+        )
+
+    def test_application_modify(self):
+        """
+        A diff on a Node, which adds a volume to an *existing* application
+        volume *and* the manifestation for the volume, can be applied without
+        triggering an invariant error on the Node.
+        """
+        node2 = node_strategy(
+            min_number_of_applications=1,
+            stateful_applications=True,
+        ).example()
+        application = node2.applications.values()[0]
+        volume = application.volume
+        node1 = node2.transform(
+            ['applications', application.name],
+            lambda o: o.set('volume', None)
+        ).transform(
+            ['manifestations'],
+            lambda o: o.remove(volume.manifestation.dataset_id)
+        )
+        diff = create_diff(node1, node2)
+        self.assertEqual(
+            node2,
+            diff.apply(node1),
+        )