Fix #9: Fix VideoFrame attributes addition, keep the same message

don't create a new message when adding measurements attributes

Author: abukharov268

savant_cloudpin/services/_base.py

@@ -112,7 +112,7 @@ async def _inbound_ws_loop(self) -> None:
 
                 self._measurements.measure_sink_message_data(frame)
                 topic, msg, extra = protocol.unpack_stream_frame(frame)
-                msg = self._measurements.measure_sink_message(msg)
+                self._measurements.add_sink_message_measure(msg)
                 self._sink.send_message(topic, msg, extra)
                 self._sink_queue.task_done()
                 await asyncio.sleep(0)
@@ -141,8 +141,9 @@ async def _outbound_ws_loop(self) -> None:
                 await asyncio.sleep(self._io_timeout)
                 continue
 
-            message = self._measurements.measure_source_message(msg.message)
-            packed = protocol.pack_stream_frame(msg.topic, message, msg.data(0))
+            topic, message, extra = msg.topic, msg.message, msg.data(0)
+            self._measurements.add_source_message_measure(message)
+            packed = protocol.pack_stream_frame(topic, message, extra)
             transport.send(WSMsgType.BINARY, packed)
             self._measurements.measure_source_message_data(packed)
             await asyncio.sleep(0)

savant_cloudpin/services/_measuring.py

@@ -208,16 +208,16 @@ def _attrs(
                 pass
         return cast(Attributes, attrs)
 
-    def measure_sink_message(self, message: Message) -> Message:
-        return self._measure_message(message, socket="Sink")
+    def add_sink_message_measure(self, message: Message) -> None:
+        self._add_message_measure(message, socket="Sink")
 
-    def measure_source_message(self, message: Message) -> Message:
-        return self._measure_message(message, socket="Source")
+    def add_source_message_measure(self, message: Message) -> None:
+        self._add_message_measure(message, socket="Source")
 
-    def _measure_message(self, message: Message, socket: ZMQSocket) -> Message:
+    def _add_message_measure(self, message: Message, socket: ZMQSocket) -> None:
         self.metrics.messages.add(1, self._attrs(socket=socket))
         self._count_trace(message, socket)
-        return self._measure_video_frame(message, socket)
+        self._measure_video_frame(message, socket)
 
     def _count_trace(self, message: Message, socket: ZMQSocket) -> None:
         span = getattr(message, "span_context", None)
@@ -232,7 +232,7 @@ def _count_trace(self, message: Message, socket: ZMQSocket) -> None:
         )
         self.metrics.traces.add(1, attributes=attrs)
 
-    def _measure_video_frame(self, message: Message, socket: ZMQSocket) -> Message:
+    def _measure_video_frame(self, message: Message, socket: ZMQSocket) -> None:
         timings = VideoFrameTimings(message)
         match self._service, socket:
             case "Client", "Source":
@@ -245,7 +245,6 @@ def _measure_video_frame(self, message: Message, socket: ZMQSocket) -> Message:
                 timings.append_timing(LABEL_CLIENT_SINK)
 
         self._detect_video_frame_delay(timings)
-        return timings.message
 
     def _detect_video_frame_delay(self, timings: VideoFrameTimings) -> None:
         delay = timings.get_delay(LABEL_CLIENT_SOURCE, LABEL_SERVER_SINK)

savant_cloudpin/services/_video_frame.py

@@ -64,7 +64,6 @@ def append_timing(self, label: ValueLabel, truncate: bool = False) -> None:
 
         timings.values = values
         video_frame.set_attribute(timings)
-        self.message = Message.video_frame(video_frame)
         self.reset_cache()
 
     def get_delay(self, start_label: ValueLabel, end_label: ValueLabel) -> float | None:

tests/savant_cloudpin/test_observability.py

@@ -106,13 +106,13 @@ def test_measurements_for_video_frame(delay_mock: Mock) -> None:
     msg = MessageData.fake_video_frame().to_message()
 
     with freeze_time("2025-11-11", tz_offset=0) as frozen_time:
-        msg = client_measurements.measure_source_message(msg)
+        client_measurements.add_source_message_measure(msg)
         frozen_time.tick(delta=timedelta(seconds=2))
-        msg = server_measurements.measure_sink_message(msg)
+        server_measurements.add_sink_message_measure(msg)
         frozen_time.tick(delta=timedelta(seconds=10))
-        msg = server_measurements.measure_source_message(msg)
+        server_measurements.add_source_message_measure(msg)
         frozen_time.tick(delta=timedelta(seconds=3))
-        msg = client_measurements.measure_sink_message(msg)
+        client_measurements.add_sink_message_measure(msg)
 
     assert delay_mock.record.called
     assert delay_mock.record.call_args_list == [