From bf1d30af347a6c87c79d50c8023ce8de74717fa0 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 10 Nov 2026 10:50:24 +0000 Subject: [PATCH 1/3] refactor: redirect each streamed log line as soon as its newline arrives `StreamedLog` now flushes every complete line instead of holding the last message until the next one starts, matching the JS client. Lines of a multi-line message that arrive in a later chunk are redirected as a separate record at the level of the message they continue. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017ocuAJXagvBd6TZVvVAYzM --- src/apify_client/_streamed_log.py | 45 ++++++++++------- tests/unit/test_logging.py | 80 +++++++++++++++++++++++++++++-- 2 files changed, 104 insertions(+), 21 deletions(-) diff --git a/src/apify_client/_streamed_log.py b/src/apify_client/_streamed_log.py index c175ff17..4f008573 100644 --- a/src/apify_client/_streamed_log.py +++ b/src/apify_client/_streamed_log.py @@ -51,43 +51,52 @@ def __init__(self, to_logger: logging.Logger, *, from_start: bool = True) -> Non self._split_marker = re.compile(rb'(?:\n|^)(\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=UTC) self._received_data = False + self._last_message_level: int | None = None + """Level of the last redirected message, or `None` if the last message was skipped or none came yet.""" 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:] - - for marker, content in zip(message_markers, message_contents, strict=False): + complete, rest = buffer[:last_newline], buffer[last_newline + 1 :] + self._stream_buffer = [rest] if rest else [] + + all_parts = re.split(self._split_marker, complete) + # Lines before the first marker continue a message whose first lines an earlier chunk already redirected. + continuation = all_parts[0].decode('utf-8').strip() + if continuation and self._last_message_level is not None: + self._to_logger.log(level=self._last_message_level, msg=continuation) + + message_markers = all_parts[1::2] + message_contents = all_parts[2::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: log_time = datetime.fromisoformat(decoded_marker) if log_time < self._relevancy_time_limit: # Skip irrelevant logs + self._last_message_level = None continue message = decoded_marker + decoded_content - self._to_logger.log(level=self._guess_log_level_from_message(message), msg=message.strip()) + self._last_message_level = self._guess_log_level_from_message(message) + self._to_logger.log(level=self._last_message_level, msg=message.strip()) def _process_whole_log(self, log: bytes | None) -> None: """Redirect a whole log read in one request, including its last part.""" diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 48ada14e..60b4817f 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,80 @@ 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 ERROR boom\n', b' traceback line\n', b'2025-05-13T07:24:13.588Z next\n'], + [ + (logging.ERROR, '2025-05-13T07:24:12.588Z ERROR boom'), + (logging.ERROR, 'traceback line'), + (logging.INFO, '2025-05-13T07:24:13.588Z next'), + ], + id='continuation-in-later-chunk-keeps-level', + ), + 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, From 6f306fee5be4eef486692af91881e3db6ea6ad74 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 10 Nov 2026 10:59:22 +0000 Subject: [PATCH 2/3] refactor: keep the indentation of continuation lines in streamed logs Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017ocuAJXagvBd6TZVvVAYzM --- src/apify_client/_streamed_log.py | 2 +- tests/unit/test_logging.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/apify_client/_streamed_log.py b/src/apify_client/_streamed_log.py index 4f008573..0422e0fc 100644 --- a/src/apify_client/_streamed_log.py +++ b/src/apify_client/_streamed_log.py @@ -79,7 +79,7 @@ def _log_buffer_content(self, *, include_last_part: bool = False) -> None: all_parts = re.split(self._split_marker, complete) # Lines before the first marker continue a message whose first lines an earlier chunk already redirected. - continuation = all_parts[0].decode('utf-8').strip() + continuation = all_parts[0].decode('utf-8').lstrip('\n').rstrip() if continuation and self._last_message_level is not None: self._to_logger.log(level=self._last_message_level, msg=continuation) diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 60b4817f..5c7b6e59 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -1385,7 +1385,7 @@ def generate_logs() -> Iterator[bytes]: [b'2025-05-13T07:24:12.588Z ERROR boom\n', b' traceback line\n', b'2025-05-13T07:24:13.588Z next\n'], [ (logging.ERROR, '2025-05-13T07:24:12.588Z ERROR boom'), - (logging.ERROR, 'traceback line'), + (logging.ERROR, ' traceback line'), (logging.INFO, '2025-05-13T07:24:13.588Z next'), ], id='continuation-in-later-chunk-keeps-level', From 70d3ae486a8e6caba5e76050708a4bfe032e9fc8 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 10 Nov 2026 15:18:36 +0000 Subject: [PATCH 3/3] refactor: drop handling of log lines without a timestamp marker Every line of a platform log starts with a timestamp, so a chunk never begins in the middle of a message. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_017ocuAJXagvBd6TZVvVAYzM --- src/apify_client/_streamed_log.py | 17 ++++------------- tests/unit/test_logging.py | 9 --------- 2 files changed, 4 insertions(+), 22 deletions(-) diff --git a/src/apify_client/_streamed_log.py b/src/apify_client/_streamed_log.py index 0422e0fc..40195297 100644 --- a/src/apify_client/_streamed_log.py +++ b/src/apify_client/_streamed_log.py @@ -51,8 +51,6 @@ def __init__(self, to_logger: logging.Logger, *, from_start: bool = True) -> Non self._split_marker = re.compile(rb'(?:\n|^)(\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=UTC) self._received_data = False - self._last_message_level: int | None = None - """Level of the last redirected message, or `None` if the last message was skipped or none came yet.""" def _process_new_data(self, data: bytes) -> None: if data: @@ -77,14 +75,9 @@ def _log_buffer_content(self, *, include_last_part: bool = False) -> None: complete, rest = buffer[:last_newline], buffer[last_newline + 1 :] self._stream_buffer = [rest] if rest else [] - all_parts = re.split(self._split_marker, complete) - # Lines before the first marker continue a message whose first lines an earlier chunk already redirected. - continuation = all_parts[0].decode('utf-8').lstrip('\n').rstrip() - if continuation and self._last_message_level is not None: - self._to_logger.log(level=self._last_message_level, msg=continuation) - - message_markers = all_parts[1::2] - message_contents = all_parts[2::2] + 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') @@ -92,11 +85,9 @@ def _log_buffer_content(self, *, include_last_part: bool = False) -> None: log_time = datetime.fromisoformat(decoded_marker) if log_time < self._relevancy_time_limit: # Skip irrelevant logs - self._last_message_level = None continue message = decoded_marker + decoded_content - self._last_message_level = self._guess_log_level_from_message(message) - self._to_logger.log(level=self._last_message_level, msg=message.strip()) + self._to_logger.log(level=self._guess_log_level_from_message(message), msg=message.strip()) def _process_whole_log(self, log: bytes | None) -> None: """Redirect a whole log read in one request, including its last part.""" diff --git a/tests/unit/test_logging.py b/tests/unit/test_logging.py index 5c7b6e59..1258b820 100644 --- a/tests/unit/test_logging.py +++ b/tests/unit/test_logging.py @@ -1381,15 +1381,6 @@ def generate_logs() -> Iterator[bytes]: @pytest.mark.parametrize( ('chunks', 'expected_calls'), [ - pytest.param( - [b'2025-05-13T07:24:12.588Z ERROR boom\n', b' traceback line\n', b'2025-05-13T07:24:13.588Z next\n'], - [ - (logging.ERROR, '2025-05-13T07:24:12.588Z ERROR boom'), - (logging.ERROR, ' traceback line'), - (logging.INFO, '2025-05-13T07:24:13.588Z next'), - ], - id='continuation-in-later-chunk-keeps-level', - ), 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')],