Merge branch 'main' into pipeline-specific-schemas

Author: rreddy2

nodestream/metrics.py

@@ -214,7 +214,8 @@ def increment(self, metric: Metric, value: Number):
         self.metrics[metric] = self.metrics.get(metric, 0) + value
 
     def decrement(self, metric: Metric, value: Number):
-        self.metrics[metric] = self.metrics.get(metric, 0) - value
+        if not metric.accumulate:
+            self.metrics[metric] = self.metrics.get(metric, 0) - value
 
     def discharge(self) -> dict[Metric, Number]:
         metrics = {}
@@ -248,7 +249,8 @@ def increment(self, metric: Metric, value: Number):
         self.metrics[metric] = self.metrics.get(metric, 0) + value
 
     def decrement(self, metric: Metric, value: Number):
-        self.metrics[metric] = self.metrics.get(metric, 0) - value
+        if not metric.accumulate:
+            self.metrics[metric] = self.metrics.get(metric, 0) - value
 
     def discharge(self) -> dict[Metric, Number]:
         metrics = {}

nodestream/pipeline/channel.py

@@ -51,6 +51,7 @@ def create_with_naming(
         metric = Metric(
             f"buffered_{input_name}_to_{output_name}",
             f"Records buffered: {input_name} → {output_name}",
+            accumulate=True,
         )
         return cls(size, metric)
 

nodestream/pipeline/progress_reporter.py

@@ -64,10 +64,11 @@ def for_testing(cls, results_list: list) -> "PipelineProgressReporter":
         return cls(
             reporting_frequency=1,
             logger=getLogger("test"),
-            callback=lambda _, record: results_list.append(record),
+            callback=no_op,
             on_start_callback=no_op,
             on_finish_callback=no_op,
             on_fatal_error_callback=raise_exception,
+            observability_callback=lambda record: results_list.append(record),
         )
 
     def report(self, index, metrics: Metrics):

tests/unit/test_metrics.py

@@ -202,3 +202,40 @@ def test_json_log_metric_handler_discharge_with_accumulate(mocker):
     # Accumulating metric should be reset to 0, non-accumulating should remain
     assert handler.metrics[accumulating_metric] == 0
     assert handler.metrics[non_accumulating_metric] == 7
+
+
+def test_json_log_metric_handler_discharge_with_accumulate_and_decrement(mocker):
+    """Test that JsonLogMetricHandler does not decrement accumulating metrics"""
+    handler = JsonLogMetricHandler()
+
+    accumulating_metric = Metric("test_accumulate", accumulate=True)
+    non_accumulating_metric = Metric("test_no_accumulate", accumulate=False)
+
+    handler.increment(accumulating_metric, 10)
+    handler.decrement(accumulating_metric, 7)
+
+    handler.increment(non_accumulating_metric, 10)
+    handler.decrement(non_accumulating_metric, 7)
+
+    # Accumulating metric should be reset to 0, non-accumulating should remain
+    assert handler.metrics[accumulating_metric] == 10
+    assert handler.metrics[non_accumulating_metric] == 3
+
+
+def test_console_metric_handler_discharge_with_accumulate_and_decrement(mocker):
+    """Test that ConsoleMetricHandler does not decrement accumulating metrics"""
+    mock_command = mocker.Mock()
+    handler = ConsoleMetricHandler(mock_command)
+
+    accumulating_metric = Metric("test_accumulate", accumulate=True)
+    non_accumulating_metric = Metric("test_no_accumulate", accumulate=False)
+
+    handler.increment(accumulating_metric, 10)
+    handler.decrement(accumulating_metric, 7)
+
+    handler.increment(non_accumulating_metric, 10)
+    handler.decrement(non_accumulating_metric, 7)
+
+    # Accumulating metric should be reset to 0, non-accumulating should remain
+    assert handler.metrics[accumulating_metric] == 10
+    assert handler.metrics[non_accumulating_metric] == 3