Fix potential issues with atomic pipeline updates + tests

mpenn · mpenn · commit 1a7b01097303 · 2025-09-04T21:44:55.000-04:00
Signed-off-by: Matthew Penn &lt;mpenn@nvidia.com&gt;
diff --git a/src/nat/observability/exporter/processing_exporter.py b/src/nat/observability/exporter/processing_exporter.py
@@ -97,25 +97,25 @@ def add_processor(self,
         # Validate type compatibility at insertion point
         self._validate_insertion_compatibility(processor, insert_position)
 
-        # Insert the processor at the calculated position
-        if insert_position == len(self._processors):
-            self._processors.append(processor)
-        else:
-            self._processors.insert(insert_position, processor)
-
-        # Validate and register processor name if provided (before position updates)
+        # Pre-validate name (no side effects yet)
         if name is not None:
             if not isinstance(name, str):
                 raise TypeError(f"Processor name must be a string, got {type(name).__name__}")
             if name in self._processor_names:
                 raise ValueError(f"Processor name '{name}' already exists")
 
-        # Always update positions for processors that shifted
-        for proc_name, pos in self._processor_names.items():
+        # Shift existing name positions (do this before list mutation)
+        for proc_name, pos in list(self._processor_names.items()):
             if pos >= insert_position:
                 self._processor_names[proc_name] = pos + 1
 
-        # Register the processor name after position updates
+        # Insert the processor
+        if insert_position == len(self._processors):
+            self._processors.append(processor)
+        else:
+            self._processors.insert(insert_position, processor)
+
+        # Record the new processor name, if provided
         if name is not None:
             self._processor_names[name] = insert_position
 
diff --git a/tests/nat/observability/exporter/test_processing_exporter.py b/tests/nat/observability/exporter/test_processing_exporter.py
@@ -301,13 +301,89 @@ def test_add_processor_duplicate_name_raises_error(self, processing_exporter):
         with pytest.raises(ValueError, match="Processor name 'test_name' already exists"):
             processing_exporter.add_processor(processor2, name="test_name")
 
+    def test_add_processor_atomicity_on_name_validation_failure(self, processing_exporter):
+        """Test that failed name validation leaves processor pipeline unchanged (atomicity)."""
+        # Set up initial state with multiple processors
+        processor1 = MockProcessor("proc1")  # str -> int
+        processor2 = MockBatchProcessor("proc2")  # int -> list[int]
+
+        processing_exporter.add_processor(processor1, name="first")
+        processing_exporter.add_processor(processor2, name="second")
+
+        # Capture initial state
+        initial_processor_count = len(processing_exporter._processors)
+        initial_processor_objects = processing_exporter._processors.copy()
+        initial_name_mapping = processing_exporter._processor_names.copy()
+
+        # Attempt to add processor with duplicate name (should fail)
+        # Make processor3 compatible with processor2's output (list[int] -> ?)
+        class ListToIntProcessor(Processor[list[int], int]):
+
+            async def process(self, item: list[int]) -> int:
+                return sum(item)
+
+        processor3 = ListToIntProcessor()  # list[int] -> int (compatible)
+
+        with pytest.raises(ValueError, match="Processor name 'first' already exists"):
+            processing_exporter.add_processor(processor3, name="first")  # Duplicate name
+
+        # Verify complete atomicity - no partial state changes
+        assert len(processing_exporter._processors) == initial_processor_count, \
+            "Processor count changed after failed operation"
+        assert processing_exporter._processors == initial_processor_objects, \
+            "Processor list modified after failed operation"
+        assert processing_exporter._processor_names == initial_name_mapping, \
+            "Name mapping modified after failed operation"
+
+        # Verify the failed processor was not added anywhere
+        assert processor3 not in processing_exporter._processors, \
+            "Failed processor found in processor list"
+
     def test_add_processor_non_string_name_raises_error(self, processing_exporter):
         """Test that non-string processor names raise TypeError."""
         processor = MockProcessor()
 
         with pytest.raises(TypeError, match="Processor name must be a string"):
             processing_exporter.add_processor(processor, name=123)  # Invalid type
 
+    def test_add_processor_atomicity_on_type_validation_failure(self, processing_exporter):
+        """Test that failed type validation leaves processor pipeline unchanged (atomicity)."""
+        # Set up initial state with multiple processors
+        processor1 = MockProcessor("proc1")  # str -> int
+        processor2 = MockBatchProcessor("proc2")  # int -> list[int]
+
+        processing_exporter.add_processor(processor1, name="first")
+        processing_exporter.add_processor(processor2, name="second")
+
+        # Capture initial state
+        initial_processor_count = len(processing_exporter._processors)
+        initial_processor_objects = processing_exporter._processors.copy()
+        initial_name_mapping = processing_exporter._processor_names.copy()
+
+        # Attempt to add processor with invalid name type (should fail)
+        # Make processor3 compatible with processor2's output (list[int] -> ?)
+        class ListToStringProcessor(Processor[list[int], str]):
+
+            async def process(self, item: list[int]) -> str:
+                return str(sum(item))
+
+        processor3 = ListToStringProcessor()  # list[int] -> str (compatible)
+
+        with pytest.raises(TypeError, match="Processor name must be a string"):
+            processing_exporter.add_processor(processor3, name=123)  # Invalid type
+
+        # Verify complete atomicity - no partial state changes
+        assert len(processing_exporter._processors) == initial_processor_count, \
+            "Processor count changed after failed operation"
+        assert processing_exporter._processors == initial_processor_objects, \
+            "Processor list modified after failed operation"
+        assert processing_exporter._processor_names == initial_name_mapping, \
+            "Name mapping modified after failed operation"
+
+        # Verify the failed processor was not added anywhere
+        assert processor3 not in processing_exporter._processors, \
+            "Failed processor found in processor list"
+
     def test_get_processor_by_name_exists(self, processing_exporter):
         """Test getting processor by name when it exists."""
         processor = MockProcessor()
