simpler serialization change detection

Author: essweine

SpiffWorkflow/bpmn/serializer/default/workflow.py

@@ -20,6 +20,7 @@
 from uuid import UUID
 
 from SpiffWorkflow.bpmn.specs.mixins.subworkflow_task import SubWorkflowTask
+from SpiffWorkflow.util.deep_merge import DeepMerge
 from SpiffWorkflow.util.copyonwrite import CopyOnWriteDict
 
 from ..helpers.bpmn_converter import BpmnConverter
@@ -29,16 +30,18 @@ class TaskConverter(BpmnConverter):
     def to_dict(self, task):
         # Optimize serialization by storing only local changes (delta)
         # instead of the full materialized data when task has a parent
-        if isinstance(task.data, CopyOnWriteDict) and task.parent is not None:
-            # Store only local modifications (delta from parent)
-            task_data = task.data.get_local_data()
-            delta_serialization = True
+
+        if task.parent is None:
+            data = task.data
+            delta = {}
         else:
-            # Root task or regular dict: serialize full data
-            task_data = task.data.materialize() if isinstance(task.data, CopyOnWriteDict) else task.data
-            delta_serialization = False
+            data = {}
+            delta = {
+                'updates': DeepMerge.get_updated_keys(task.parent.data, task.data),
+                'deletions': DeepMerge.get_deleted_keys(task.parent.data, task.data),
+            }
 
-        result = {
+        return {
             'id': str(task.id),
             'parent': str(task._parent) if task.parent is not None else None,
             'children': [ str(child) for child in task._children ],
@@ -47,16 +50,12 @@ def to_dict(self, task):
             'task_spec': task.task_spec.name,
             'triggered': task.triggered,
             'internal_data': self.registry.convert(task.internal_data),
-            'data': self.registry.convert(self.registry.clean(task_data)),
+            'data': data,
+            'delta': delta
         }
 
-        # Mark delta serialization for deserialization
-        if delta_serialization:
-            result['data_is_delta'] = True
-
-        return result
-
     def from_dict(self, dct, workflow):
+
         task_spec = workflow.spec.task_specs.get(dct['task_spec'])
         task = self.target_class(workflow, task_spec, state=dct['state'], id=UUID(dct['id']))
         task._parent = UUID(dct['parent']) if dct['parent'] is not None else None
@@ -65,14 +64,17 @@ def from_dict(self, dct, workflow):
         task.triggered = dct['triggered']
         task.internal_data = self.registry.restore(dct['internal_data'])
 
-        # Handle delta vs full data serialization
-        restored_data = self.registry.restore(dct['data'])
-        if dct.get('data_is_delta', False) and task.parent is not None:
-            # Reconstruct full data from parent + local delta
-            task.data = CopyOnWriteDict(parent=task.parent.data, **restored_data)
+        delta = dct.get('delta')
+        if delta:
+            data = DeepMerge.merge({}, task.parent.data)
+            data.update(self.registry.restore(delta.get('updates', {})))
+            for key in delta.get('deletions', {}):
+                if key in data:
+                    del data[key]
         else:
-            # Full data (backward compatible with old serializations)
-            task.data = restored_data
+            data = self.registry.restore(dct['data'])
+
+        task.data = data
 
         return task
 

SpiffWorkflow/task.py

@@ -25,7 +25,7 @@
 
 from .util.task import TaskState, TaskFilter, TaskIterator
 from .util.deep_merge import DeepMerge
-from .util.copyonwrite import CopyOnWriteDict
+#from .util.copyonwrite import CopyOnWriteDict
 from .exceptions import WorkflowException
 
 logger = logging.getLogger('spiff.task')
@@ -315,16 +315,16 @@ def _assign_new_thread_id(self, recursive=True):
         return self.thread_id
 
     def _inherit_data(self):
-        """Inherits data from the parent using copy-on-write semantics."""
+        """Inherits data from the parent."""
+        self.data = DeepMerge.merge(self.data, self.parent.data)
         # Preserve any data that was already set on this task before inheriting
         # (e.g., multi-instance input items set before _update is called)
         # But parent data takes precedence for conflicting keys (matches old behavior)
-        existing_only = {k: v for k, v in (self.data or {}).items()
-                        if k not in self.parent.data}
+        #existing_only = {k: v for k, v in (self.data or {}).items() if k not in self.parent.data}
 
         # Use CopyOnWriteDict to share parent data until modifications are made
         # This avoids expensive deepcopy operations for every task
-        self.data = CopyOnWriteDict(parent=self.parent.data, **existing_only)
+        #self.data = CopyOnWriteDict(parent=self.parent.data, **existing_only)
 
     def _set_internal_data(self, **kwargs):
         """Defines the given attribute/value pairs in this task's internal data."""

SpiffWorkflow/util/deep_merge.py

@@ -74,3 +74,13 @@ def merge_array(a, b, path=None):
 
         # Trim a back to the length of b.  In the end, the two arrays should match
         del a[len(b):]
+
+    @staticmethod
+    def get_updated_keys(a, b):
+        """get a list of keys from b that are different from a"""
+        return dict((key, b[key]) for key in b if key not in a or b[key] != a[key])
+
+    @staticmethod
+    def get_deleted_keys(a, b):
+        """get a list of keys from a that do not exist in b"""
+        return [key for key in a if key not in b]