From 8cf1190f7a80472a0c24ce8587c25f9ecb50c85f Mon Sep 17 00:00:00 2001
From: Julien Cristau <jcristau@mozilla.com>
Date: Wed, 15 Jan 2025 14:42:13 +0100
Subject: [PATCH] Bug 1941166 - handle long lines in pipe_to_log (#679)

Avoid deadlock with full pipe buffer when the output contains long lines
(>64kB).  If we go this long without a newline, log what we have and
keep going.
---
 src/scriptworker/log.py | 10 +++++++++-
 tests/test_log.py       | 14 ++++++++++++++
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/src/scriptworker/log.py b/src/scriptworker/log.py
index 8b8853a3..b3a90916 100644
--- a/src/scriptworker/log.py
+++ b/src/scriptworker/log.py
@@ -5,6 +5,7 @@
     log (logging.Logger): the log object for this module.
 
 """
+import asyncio
 import logging
 import logging.handlers
 import os
@@ -82,7 +83,14 @@ async def pipe_to_log(pipe: StreamReader, filehandles: Sequence[IO[str]] = (), l
 
     """
     while True:
-        line = await pipe.readline()  # type: Union[str, bytes]
+        try:
+            line = await pipe.readuntil()  # type: Union[str, bytes]
+        except asyncio.exceptions.IncompleteReadError as e:
+            line = e.partial
+        except asyncio.exceptions.LimitOverrunError:
+            # line too long
+            line = bytes(pipe._buffer)
+            pipe._buffer.clear()
         if line:
             line = to_unicode(line)
             log.log(level, line.rstrip())
diff --git a/tests/test_log.py b/tests/test_log.py
index 359776dc..83c2a196 100644
--- a/tests/test_log.py
+++ b/tests/test_log.py
@@ -62,6 +62,20 @@ async def test_pipe_to_log(rw_context):
     assert read(log_file) in ("foo\nbar\n", "bar\nfoo\n")
 
 
+@pytest.mark.asyncio
+async def test_pipe_to_log_limit(rw_context):
+    cmd = ["echo", "a" * 100_000]
+    proc = await asyncio.create_subprocess_exec(*cmd, stdout=PIPE, stderr=PIPE, stdin=None)
+    tasks = []
+    with swlog.get_log_filehandle(rw_context) as log_fh:
+        tasks.append(asyncio.create_task(swlog.pipe_to_log(proc.stderr, filehandles=[log_fh])))
+        tasks.append(asyncio.create_task(swlog.pipe_to_log(proc.stdout, filehandles=[log_fh])))
+        await asyncio.wait(tasks)
+        await proc.wait()
+    log_file = swlog.get_log_filename(rw_context)
+    assert len(read(log_file)) == 100_001
+
+
 def test_update_logging_config_verbose(rw_context):
     rw_context.config["verbose"] = True
     swlog.update_logging_config(rw_context, log_name=rw_context.config["log_dir"])