Handle bytestream edgecase of chunk containing only half of the multibyte character

Author: Pijukatel

src/apify_client/clients/resource_clients/log.py

@@ -231,12 +231,12 @@ def __init__(self, to_logger: logging.Logger, *, from_start: bool = True) -> Non
         self._to_logger = to_logger
         if self._force_propagate:
             to_logger.propagate = True
-        self._stream_buffer = list[str]()
-        self._split_marker = re.compile(r'(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z)')
+        self._stream_buffer = list[bytes]()
+        self._split_marker = re.compile(rb'(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z)')
         self._relevancy_time_limit: datetime | None = None if from_start else datetime.now(tz=timezone.utc)
 
     def _process_new_data(self, data: bytes) -> None:
-        new_chunk = data.decode('utf-8')
+        new_chunk = data
         self._stream_buffer.append(new_chunk)
         if re.findall(self._split_marker, new_chunk):
             # If complete split marker was found in new chunk, then log the buffer.
@@ -248,7 +248,7 @@ def _log_buffer_content(self, *, include_last_part: bool = False) -> None:
         Log the messages created from the split parts and remove them from buffer.
         The last part could be incomplete, and so it can be left unprocessed in the buffer until later.
         """
-        all_parts = re.split(self._split_marker, ''.join(self._stream_buffer))[1:]  # First split is empty string
+        all_parts = re.split(self._split_marker, b''.join(self._stream_buffer))[1:]  # The First split is empty
         if include_last_part:
             message_markers = all_parts[0::2]
             message_contents = all_parts[1::2]
@@ -260,12 +260,14 @@ def _log_buffer_content(self, *, include_last_part: bool = False) -> None:
             self._stream_buffer = all_parts[-2:]
 
         for marker, content in zip(message_markers, message_contents):
+            decoded_marker = marker.decode('utf-8')
+            decoded_content = content.decode('utf-8')
             if self._relevancy_time_limit:
-                log_time = datetime.fromisoformat(marker.replace('Z', '+00:00'))
+                log_time = datetime.fromisoformat(decoded_marker.replace('Z', '+00:00'))
                 if log_time < self._relevancy_time_limit:
                     # Skip irrelevant logs
                     continue
-            message = marker + content
+            message = decoded_marker + decoded_content
             self._to_logger.log(level=self._guess_log_level_from_message(message), msg=message.strip())
 
     @staticmethod

tests/unit/test_logging.py

@@ -23,20 +23,23 @@
 _MOCKED_ACTOR_LOGS = (
     b'2025-05-13T07:24:12.588Z ACTOR: Pulling Docker image of build.\n'
     b'2025-05-13T07:24:12.686Z ACTOR: Creating Docker container.\n'
-    b'2025-05-13T07:24:12.745Z ACTOR: Starting Docker container.',  # Several logs merged into one message
+    b'2025-05-13T07:24:12.745Z ACTOR: Starting Docker container.',  # Several logs merged into one chunk
+    b'2025-05-13T07:26:14.132Z [apify] DEBUG \xc3',  # Chunked log split in the middle of the multibyte character
+    b'\xa1',  # part 2
     b'2025-05-13T07:24:14.132Z [apify] INFO multiline \n log',
     b'2025-05-13T07:25:14.132Z [apify] WARNING some warning',
     b'2025-05-13T07:26:14.132Z [apify] DEBUG c',
-    b'2025-05-13T0',  # Chunked log that got split in the marker, part 1
-    b'7:26:14.132Z [apify] DEBUG d'  # Chunked log that got split in the marker, part 2
-    b'2025-05-13T07:26:14.132Z [apify] DEB',  # Chunked log that got split outside of marker, part 1
-    b'UG e',  # Chunked log that got split outside of marker, part 1
+    b'2025-05-13T0',  # Chunked log that got split in the marker
+    b'7:26:14.132Z [apify] DEBUG d'  # part 2
+    b'2025-05-13T07:26:14.132Z [apify] DEB',  # Chunked log that got split outside of marker
+    b'UG e',  # part 2
 )
 
 _EXPECTED_MESSAGES_AND_LEVELS = (
     ('2025-05-13T07:24:12.588Z ACTOR: Pulling Docker image of build.', logging.INFO),
     ('2025-05-13T07:24:12.686Z ACTOR: Creating Docker container.', logging.INFO),
     ('2025-05-13T07:24:12.745Z ACTOR: Starting Docker container.', logging.INFO),
+    ('2025-05-13T07:26:14.132Z [apify] DEBUG á', logging.DEBUG),
     ('2025-05-13T07:24:14.132Z [apify] INFO multiline \n log', logging.INFO),
     ('2025-05-13T07:25:14.132Z [apify] WARNING some warning', logging.WARNING),
     ('2025-05-13T07:26:14.132Z [apify] DEBUG c', logging.DEBUG),
@@ -115,7 +118,9 @@ def propagate_stream_logs() -> None:
     logging.getLogger(f'apify.{_MOCKED_ACTOR_NAME}-{_MOCKED_RUN_ID}').setLevel(logging.DEBUG)
 
 
-@pytest.mark.parametrize(('log_from_start', 'expected_log_count'), [(True, 8), (False, 5)])
+@pytest.mark.parametrize(
+    ('log_from_start', 'expected_log_count'), [(True, len(_EXPECTED_MESSAGES_AND_LEVELS)), (False, 6)]
+)
 @respx.mock
 async def test_redirected_logs_async(
     *,
@@ -148,7 +153,9 @@ async def test_redirected_logs_async(
         assert expected_message_and_level[1] == record.levelno
 
 
-@pytest.mark.parametrize(('log_from_start', 'expected_log_count'), [(True, 8), (False, 5)])
+@pytest.mark.parametrize(
+    ('log_from_start', 'expected_log_count'), [(True, len(_EXPECTED_MESSAGES_AND_LEVELS)), (False, 6)]
+)
 @respx.mock
 def test_redirected_logs_sync(
     *,
@@ -201,7 +208,7 @@ async def test_actor_call_redirect_logs_to_default_logger_async(
     assert isinstance(logger.handlers[0], logging.StreamHandler)
 
     # Ensure logs are propagated
-    assert len(caplog.records) == 8
+    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS)
     for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
         assert expected_message_and_level[0] == record.message
         assert expected_message_and_level[1] == record.levelno
@@ -228,7 +235,7 @@ def test_actor_call_redirect_logs_to_default_logger_sync(
     assert isinstance(logger.handlers[0], logging.StreamHandler)
 
     # Ensure logs are propagated
-    assert len(caplog.records) == 8
+    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS)
     for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
         assert expected_message_and_level[0] == record.message
         assert expected_message_and_level[1] == record.levelno
@@ -278,7 +285,7 @@ async def test_actor_call_redirect_logs_to_custom_logger_async(
     with caplog.at_level(logging.DEBUG, logger=logger_name):
         await run_client.call(logger=logger)
 
-    assert len(caplog.records) == 8
+    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS)
     for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
         assert expected_message_and_level[0] == record.message
         assert expected_message_and_level[1] == record.levelno
@@ -298,7 +305,7 @@ def test_actor_call_redirect_logs_to_custom_logger_sync(
     with caplog.at_level(logging.DEBUG, logger=logger_name):
         run_client.call(logger=logger)
 
-    assert len(caplog.records) == 8
+    assert len(caplog.records) == len(_EXPECTED_MESSAGES_AND_LEVELS)
     for expected_message_and_level, record in zip(_EXPECTED_MESSAGES_AND_LEVELS, caplog.records):
         assert expected_message_and_level[0] == record.message
         assert expected_message_and_level[1] == record.levelno