diff --git a/src/apify_client/_streamed_log.py b/src/apify_client/_streamed_log.py index c175ff17..40195297 100644 --- a/src/apify_client/_streamed_log.py +++ b/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: diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 48ada14e..1258b820 100644 --- a/tests/unit/test_logging.py +++ b/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,