refactor: redirect each streamed log line as soon as its newline arrives

src/apify_client/_streamed_log.py

@@ -55,30 +55,30 @@ def __init__(self, to_logger: logging.Logger, *, from_start: bool = True) -> Non
     def _process_new_data(self, data: bytes) -> None:
         if data:
             self._received_data = True
-        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.
+        self._stream_buffer.append(data)
+        if b'\n' in data:
             self._log_buffer_content(include_last_part=False)
 
     def _log_buffer_content(self, *, include_last_part: bool = False) -> None:
-        """Merge the whole buffer and split it into parts based on the marker.
+        """Redirect every complete line in the buffer and keep only the unterminated last line.
 
-        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.
+        A line is complete once its newline arrives, so a message is redirected as soon as it is written, not when the
+        next message starts. With `include_last_part`, the unterminated last line is redirected too.
         """
-        all_parts = re.split(self._split_marker, b''.join(self._stream_buffer))[1:]  # The First split is empty
+        buffer = b''.join(self._stream_buffer)
+        last_newline = buffer.rfind(b'\n')
         if include_last_part:
-            message_markers = all_parts[0::2]
-            message_contents = all_parts[1::2]
-            self._stream_buffer = []
+            complete, rest = buffer, b''
+        elif last_newline == -1:
+            complete, rest = b'', buffer
         else:
-            message_markers = all_parts[0:-2:2]
-            message_contents = all_parts[1:-2:2]
-            # The last two parts (marker and message) are possibly not complete and will be left in the buffer
-            self._stream_buffer = all_parts[-2:]
+            complete, rest = buffer[:last_newline], buffer[last_newline + 1 :]
+        self._stream_buffer = [rest] if rest else []
 
-        for marker, content in zip(message_markers, message_contents, strict=False):
+        all_parts = re.split(self._split_marker, complete)[1:]  # Every platform log line starts with a marker
+        message_markers = all_parts[0::2]
+        message_contents = all_parts[1::2]
+        for marker, content in zip(message_markers, message_contents, strict=True):
             decoded_marker = marker.decode('utf-8')
             decoded_content = content.decode('utf-8')
             if self._relevancy_time_limit:

tests/unit/test_logging.py

@@ -915,15 +915,15 @@ def _register_run_and_actor_endpoints(httpserver: HTTPServer) -> None:
 
 
 def _wait_for_blocking_read(streamed_log: StreamedLogBase, *, timeout: float = 5) -> None:
-    """Wait until a chunk has been buffered, which proves the reader reached the blocking read.
+    """Wait until a chunk has been received, which proves the reader reached the blocking read.
 
     A fixed sleep could end before the reader gets there. Async callers reach this through `asyncio.to_thread`, so
     the streaming task keeps the event loop while the wait runs.
     """
     deadline = time.monotonic() + timeout
-    while not streamed_log._stream_buffer and time.monotonic() < deadline:
+    while not streamed_log._received_data and time.monotonic() < deadline:
         time.sleep(0.01)
-    assert streamed_log._stream_buffer, 'the log stream never reached its blocking read'
+    assert streamed_log._received_data, 'the log stream never reached its blocking read'
 
 
 @pytest.mark.usefixtures('propagate_stream_logs')
@@ -1345,6 +1345,71 @@ def generate_logs() -> Iterator[bytes]:
     streamed_log.stop()
 
 
+def test_streamed_log_sync_redirects_line_before_next_message(httpserver: HTTPServer) -> None:
+    """A complete line is redirected while the stream stays open and silent, without waiting for the next message."""
+    release_server = threading.Event()
+    line = '2025-05-13T07:24:12.588Z ACTOR: going quiet'
+
+    def _silent_handler(_request: Request) -> Response:
+        def generate_logs() -> Iterator[bytes]:
+            yield f'{line}\n'.encode()
+            release_server.wait(timeout=30)
+
+        return Response(response=generate_logs(), status=200, mimetype='application/octet-stream')
+
+    httpserver.expect_request(
+        f'/v2/actor-runs/{_MOCKED_RUN_ID}/log', method='GET', query_string='stream=true&raw=true'
+    ).respond_with_handler(_silent_handler)
+
+    api_url = httpserver.url_for('/').removesuffix('/')
+    log_client = ApifyClient(token='mocked_token', api_url=api_url).run(run_id=_MOCKED_RUN_ID).log()
+    to_logger = Mock(spec=logging.Logger)
+    streamed_log = StreamedLog(log_client, to_logger=to_logger)
+
+    streaming_thread = streamed_log.start()
+    try:
+        deadline = time.monotonic() + 5
+        while not to_logger.log.called and time.monotonic() < deadline:
+            time.sleep(0.01)
+        to_logger.log.assert_called_once_with(level=logging.INFO, msg=line)
+    finally:
+        release_server.set()
+        streaming_thread.join(timeout=5)
+        streamed_log.stop()
+
+
+@pytest.mark.parametrize(
+    ('chunks', 'expected_calls'),
+    [
+        pytest.param(
+            [b'2025-05-13T07:24:12.588Z first\n2025-05-13T07:24:13', b'.588Z second\n'],
+            [(logging.INFO, '2025-05-13T07:24:12.588Z first'), (logging.INFO, '2025-05-13T07:24:13.588Z second')],
+            id='marker-split-across-chunks',
+        ),
+        pytest.param(
+            [b'2025-05-13T07:24:12.588Z unterminated'],
+            [],
+            id='unterminated-line-waits-for-newline',
+        ),
+        pytest.param(
+            [b'preamble without marker\n', b'2025-05-13T07:24:12.588Z first\n'],
+            [(logging.INFO, '2025-05-13T07:24:12.588Z first')],
+            id='text-before-first-message-is-skipped',
+        ),
+    ],
+)
+def test_streamed_log_redirects_each_complete_line(chunks: list[bytes], expected_calls: list[tuple[int, str]]) -> None:
+    """Each chunk redirects the messages its newlines complete and keeps only the unterminated rest."""
+    to_logger = Mock(spec=logging.Logger)
+    streamed_log = StreamedLogBase(to_logger)
+
+    for chunk in chunks:
+        streamed_log._process_new_data(chunk)
+
+    calls = [(call.kwargs['level'], call.kwargs['msg']) for call in to_logger.log.call_args_list]
+    assert calls == expected_calls
+
+
 @pytest.mark.usefixtures('propagate_stream_logs')
 def test_streamed_log_sync_stop_reports_failing_tail_flush(
     caplog: LogCaptureFixture,