From bae495679e7e4acf30c94eafc5864bf0e4364bbb Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Thu, 17 Sep 2026 16:29:40 +0000 Subject: [PATCH 1/5] fix: flush the underlying stdout in PrintBuffer.flush so messages reach the platform promptly Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- airbyte_cdk/utils/print_buffer.py | 6 ++ unit_tests/utils/test_print_buffer.py | 86 +++++++++++++++++++++++++++ 2 files changed, 92 insertions(+) create mode 100644 unit_tests/utils/test_print_buffer.py diff --git a/airbyte_cdk/utils/print_buffer.py b/airbyte_cdk/utils/print_buffer.py index ae5a2020c5..553d412ea6 100644 --- a/airbyte_cdk/utils/print_buffer.py +++ b/airbyte_cdk/utils/print_buffer.py @@ -52,9 +52,15 @@ def write(self, message: str) -> None: self.last_flush_time = current_time def flush(self) -> None: + """Writes the buffered content to the real stdout and pushes it through to the file descriptor. + + `sys.__stdout__` is block-buffered when stdout is a pipe, so without the explicit + `flush()` the messages would stay in the interpreter's ~8 KiB text buffer until it fills up. + """ with self.lock: combined_message = self.buffer.getvalue() sys.__stdout__.write(combined_message) # type: ignore[union-attr] + sys.__stdout__.flush() # type: ignore[union-attr] self.buffer = StringIO() def __enter__(self) -> "PrintBuffer": diff --git a/unit_tests/utils/test_print_buffer.py b/unit_tests/utils/test_print_buffer.py new file mode 100644 index 0000000000..b1bf88d3d4 --- /dev/null +++ b/unit_tests/utils/test_print_buffer.py @@ -0,0 +1,86 @@ +# +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +# + +import os +import select +import subprocess +import sys +import time +from unittest.mock import MagicMock + +from airbyte_cdk.utils.print_buffer import PrintBuffer + + +def test_flush_pushes_through_underlying_stdout(monkeypatch): + mock_stdout = MagicMock() + monkeypatch.setattr(sys, "__stdout__", mock_stdout) + + print_buffer = PrintBuffer() + print_buffer.write("hello") + print_buffer.flush() + + mock_stdout.write.assert_called_once_with("hello") + mock_stdout.flush.assert_called_once() + + +def test_output_reaches_pipe_before_process_exit(): + child = r""" +import time, sys +from airbyte_cdk.logger import PRINT_BUFFER, init_logger + +logger = init_logger("airbyte") +with PRINT_BUFFER: + print('{"type":"RECORD","record":{"stream":"s","data":{"id":1},"emitted_at":0}}\n', end="") + logger.info("Read 1 records from s stream") + time.sleep(10) +""" + env = os.environ.copy() + env.pop("PYTHONUNBUFFERED", None) + process = subprocess.Popen( + [sys.executable, "-c", child], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + ) + assert process.stdout is not None + assert process.stderr is not None + + output = bytearray() + deadline = time.monotonic() + 5.0 + record_marker = b'"type":"RECORD"' + log_marker = b"Read 1 records from s stream" + + try: + stdout_fd = process.stdout.fileno() + os.set_blocking(stdout_fd, False) + while time.monotonic() < deadline and not ( + record_marker in output and log_marker in output + ): + timeout = max(0.0, deadline - time.monotonic()) + readable, _, _ = select.select([process.stdout], [], [], timeout) + if not readable: + break + try: + chunk = os.read(stdout_fd, 65536) + except BlockingIOError: + continue + if not chunk: + break + output.extend(chunk) + + process_alive = process.poll() is None + finally: + if process.poll() is None: + process.kill() + process.wait() + stderr = process.stderr.read() + + assert record_marker in output and log_marker in output, ( + "Expected both records in child output before process exit. " + f"Received stdout: {bytes(output)!r}; stderr: {stderr!r}" + ) + assert process_alive, ( + "Child exited before output was observed. " + f"Received stdout: {bytes(output)!r}; stderr: {stderr!r}" + ) From b98cbe9ebe6d3608cf3e5b27dfe29e85ca4b264f Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Tue, 29 Sep 2026 20:25:58 +0000 Subject: [PATCH 2/5] feat: flush PrintBuffer output from a background sweeper instead of on every log line Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- airbyte_cdk/utils/print_buffer.py | 165 +++++++++++++++++- unit_tests/utils/test_print_buffer.py | 240 +++++++++++++++++++++++++- 2 files changed, 390 insertions(+), 15 deletions(-) diff --git a/airbyte_cdk/utils/print_buffer.py b/airbyte_cdk/utils/print_buffer.py index 553d412ea6..c0b578cb66 100644 --- a/airbyte_cdk/utils/print_buffer.py +++ b/airbyte_cdk/utils/print_buffer.py @@ -1,12 +1,67 @@ # Copyright (c) 2024 Airbyte, Inc., all rights reserved. +import atexit +import os import sys +import threading import time +import weakref from io import StringIO from threading import RLock from types import TracebackType from typing import Optional +_REGISTERED: "weakref.WeakSet[PrintBuffer]" = weakref.WeakSet() +_FORK_HELD: list["PrintBuffer"] = [] +_process_hooks_registered = False +_PROCESS_HOOKS_LOCK = threading.Lock() + + +def _register_process_hooks() -> None: + global _process_hooks_registered + if _process_hooks_registered: + return + with _PROCESS_HOOKS_LOCK: + if _process_hooks_registered: + return + atexit.register(_shutdown_all) + if hasattr(os, "register_at_fork"): # not available on Windows + os.register_at_fork( + before=_before_fork, + after_in_parent=_after_fork_in_parent, + after_in_child=_after_fork_in_child, + ) + _process_hooks_registered = True + + +def _shutdown_all() -> None: + for b in list(_REGISTERED): + b._shutdown() + + +def _before_fork() -> None: + buffers = list(_REGISTERED) + for b in buffers: + b.lock.acquire() + _FORK_HELD[:] = buffers + for b in buffers: + try: + b._flush_to_fd() # the child must not inherit (and later re-emit) pending bytes + except (OSError, ValueError): + pass + + +def _after_fork_in_parent() -> None: + for b in _FORK_HELD: + b.lock.release() + _FORK_HELD.clear() + + +def _after_fork_in_child() -> None: + for b in _FORK_HELD: + b._reset_after_fork() + _FORK_HELD.clear() + class PrintBuffer: """ @@ -17,18 +72,27 @@ class PrintBuffer: scenarios where you want to minimize the number of I/O operations by grouping multiple print statements together and flushing them as a single operation. + Buffer content is pushed through to the stdout file descriptor by a daemon + sweeper thread once per flush interval, so callers of `flush()` (such as + logging handlers, which flush after every record) do not each pay for a + file-descriptor flush. + Attributes: buffer (StringIO): A buffer to store the messages before flushing. flush_interval (float): The time interval (in seconds) after which the buffer is flushed. last_flush_time (float): The last time the buffer was flushed. lock (RLock): A reentrant lock to ensure thread-safe operations. + _sweeper (Optional[threading.Thread]): The daemon thread pushing pending output to the file descriptor, or None. + _stop (threading.Event): Signals the sweeper thread to exit. + _pending (threading.Event): Signals the sweeper thread that output is waiting to be pushed. + _sync_flush (bool): When True, `flush()` pushes to the file descriptor synchronously (no sweeper). Methods: write(message: str) -> None: Writes a message to the buffer and flushes if the interval has passed. flush() -> None: - Flushes the buffer content to the standard output. + Drains the buffer to the standard output and marks it pending for the sweeper thread. __enter__() -> "PrintBuffer": Enters the runtime context related to this object, redirecting stdout and stderr. @@ -42,26 +106,107 @@ def __init__(self, flush_interval: float = 0.1): self.flush_interval = flush_interval self.last_flush_time = time.monotonic() self.lock = RLock() + self._sweeper: Optional[threading.Thread] = None + self._stop = threading.Event() + self._pending = threading.Event() + self._sync_flush = False def write(self, message: str) -> None: with self.lock: self.buffer.write(message) + self._mark_pending() current_time = time.monotonic() if (current_time - self.last_flush_time) >= self.flush_interval: self.flush() self.last_flush_time = current_time def flush(self) -> None: - """Writes the buffered content to the real stdout and pushes it through to the file descriptor. + """Drains the buffered content to the real stdout and marks it pending for the sweeper. - `sys.__stdout__` is block-buffered when stdout is a pipe, so without the explicit - `flush()` the messages would stay in the interpreter's ~8 KiB text buffer until it fills up. + `sys.__stdout__` is block-buffered when stdout is a pipe; the sweeper thread + pushes pending output through to the file descriptor within one flush interval. + When `_sync_flush` is set, the push to the file descriptor happens here instead. """ with self.lock: - combined_message = self.buffer.getvalue() - sys.__stdout__.write(combined_message) # type: ignore[union-attr] - sys.__stdout__.flush() # type: ignore[union-attr] - self.buffer = StringIO() + if self._sync_flush: + self._flush_to_fd() + else: + self._drain() + self._mark_pending() + + def _drain(self) -> None: + combined_message = self.buffer.getvalue() + sys.__stdout__.write(combined_message) # type: ignore[union-attr] + self.buffer = StringIO() + + def _flush_to_fd(self) -> None: + self._drain() + sys.__stdout__.flush() # type: ignore[union-attr] + + def _mark_pending(self) -> None: + if self._sync_flush: + return + if self._sweeper is None: + self._start_sweeper() + if not self._pending.is_set(): + self._pending.set() + + def _start_sweeper(self) -> None: + stop, pending = threading.Event(), threading.Event() + thread = threading.Thread( + target=self._sweep, + args=(stop, pending), + name="airbyte-print-buffer-sweeper", + daemon=True, + ) + _register_process_hooks() + _REGISTERED.add(self) + try: + thread.start() + except RuntimeError: # e.g. interpreter shutdown / thread limit + self._sync_flush = True + return + self._sweeper, self._stop, self._pending = thread, stop, pending + + def _sweep(self, stop: threading.Event, pending: threading.Event) -> None: + while not stop.is_set(): + pending.wait() # idle buffers block here without consuming CPU + stop.wait(self.flush_interval) # coalesce for one interval + with self.lock: + pending.clear() + try: + self._flush_to_fd() + except (OSError, ValueError): # broken pipe / closed stdout + self._sync_flush = True + return + + def _stop_sweeper(self) -> None: + with self.lock: + thread, stop, pending = self._sweeper, self._stop, self._pending + self._sweeper = None + if thread is None: + return + stop.set() + pending.set() + if thread is not threading.current_thread(): + thread.join() + + def _shutdown(self) -> None: + with self.lock: + self._sync_flush = True + self._stop_sweeper() + with self.lock: + try: + self._flush_to_fd() + except (OSError, ValueError): + pass + + def _reset_after_fork(self) -> None: + # the sweeper thread does not survive fork; locks and events may be inconsistent + self.lock = RLock() + self._sweeper = None + self._stop = threading.Event() + self._pending = threading.Event() def __enter__(self) -> "PrintBuffer": self.old_stdout, self.old_stderr = sys.stdout, sys.stderr @@ -77,5 +222,7 @@ def __exit__( exc_val: Optional[BaseException], exc_tb: Optional[TracebackType], ) -> None: - self.flush() + self._stop_sweeper() + with self.lock: + self._flush_to_fd() sys.stdout, sys.stderr = self.old_stdout, self.old_stderr diff --git a/unit_tests/utils/test_print_buffer.py b/unit_tests/utils/test_print_buffer.py index b1bf88d3d4..1e1f79d6b9 100644 --- a/unit_tests/utils/test_print_buffer.py +++ b/unit_tests/utils/test_print_buffer.py @@ -2,26 +2,157 @@ # Copyright (c) 2025 Airbyte, Inc., all rights reserved. # +import io +import logging import os +import re import select +import signal import subprocess import sys +import threading import time from unittest.mock import MagicMock +import pytest + from airbyte_cdk.utils.print_buffer import PrintBuffer -def test_flush_pushes_through_underlying_stdout(monkeypatch): +def test_flush_drains_without_fd_flush_and_sweeper_flushes(monkeypatch): + flushed = threading.Event() mock_stdout = MagicMock() + mock_stdout.flush.side_effect = lambda *args, **kwargs: flushed.set() monkeypatch.setattr(sys, "__stdout__", mock_stdout) - print_buffer = PrintBuffer() - print_buffer.write("hello") - print_buffer.flush() + with PrintBuffer(flush_interval=1.0) as print_buffer: + print_buffer.write("hello") + print_buffer.flush() + + mock_stdout.write.assert_called_once_with("hello") + # ~1 s of margin before the sweeper would fire + assert not mock_stdout.flush.called + assert flushed.wait(timeout=10) + + +class _CountingRaw(io.RawIOBase): + def __init__(self): + self.write_calls = 0 + self.data = bytearray() + + def writable(self): + return True + + def write(self, b): + self.write_calls += 1 + self.data.extend(b) + return len(b) + + +def _buffered_counting_stdout(monkeypatch): + raw = _CountingRaw() + monkeypatch.setattr( + sys, "__stdout__", io.TextIOWrapper(io.BufferedWriter(raw, buffer_size=8192)) + ) + return raw + + +def test_exit_flushes_pending_and_stops_sweeper(monkeypatch): + raw = _buffered_counting_stdout(monkeypatch) + + with PrintBuffer(flush_interval=60) as print_buffer: + print_buffer.write("a\n") + print_buffer.flush() + print_buffer.write("b\n") + sweeper = print_buffer._sweeper + + assert bytes(raw.data) == b"a\nb\n" + assert sweeper is not None + assert not sweeper.is_alive() + - mock_stdout.write.assert_called_once_with("hello") - mock_stdout.flush.assert_called_once() +def test_syscall_count_stays_batched_at_one_log_line_per_record(monkeypatch): + raw = _buffered_counting_stdout(monkeypatch) + print_buffer = PrintBuffer(flush_interval=0.1) + logger = logging.Logger("print_buffer_syscalls") + logger.setLevel(logging.INFO) + logger.propagate = False + logger.addHandler(logging.StreamHandler(print_buffer)) + + n = 2000 + record_line = ( + '{"type":"RECORD","record":{"stream":"s","data":{"id":0,' + '"payload":"' + "x" * 200 + '"},"emitted_at":0}}\n' + ) + expected = "" + started = time.monotonic() + with print_buffer: + for i in range(n): + print_buffer.write(record_line) + expected += record_line + logger.info("Read %s records", i) + expected += f"Read {i} records\n" + elapsed = time.monotonic() - started + + total_bytes = len(expected.encode()) + assert bytes(raw.data) == expected.encode() + max_write_calls = total_bytes // 8192 + 1 + int(elapsed / 0.1) + 3 + assert raw.write_calls <= max_write_calls, ( + f"expected batched fd writes, got {raw.write_calls} raw writes " + f"for {n} log lines in {elapsed:.2f}s (bound {max_write_calls})" + ) + + +def _spawn_child(code): + env = os.environ.copy() + env.pop("PYTHONUNBUFFERED", None) + return subprocess.Popen( + [sys.executable, "-W", "ignore::DeprecationWarning", "-c", code], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=env, + start_new_session=True, + ) + + +def _read_until(process, markers, timeout): + """Read child stdout until every marker is seen or the deadline passes. + + Returns (output bytes, {marker: time.monotonic() at first sighting}). + """ + assert process.stdout is not None + stdout_fd = process.stdout.fileno() + os.set_blocking(stdout_fd, False) + output = bytearray() + seen = {} + deadline = time.monotonic() + timeout + while time.monotonic() < deadline and len(seen) < len(markers): + readable, _, _ = select.select( + [process.stdout], [], [], max(0.0, deadline - time.monotonic()) + ) + if not readable: + break + try: + chunk = os.read(stdout_fd, 65536) + except BlockingIOError: + continue + if not chunk: + break + output.extend(chunk) + now = time.monotonic() + for marker in markers: + if marker not in seen and marker in output: + seen[marker] = now + return bytes(output), seen + + +def _kill_process_group(process): + if process.poll() is None: + try: + os.killpg(process.pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError): + process.kill() + process.wait() def test_output_reaches_pipe_before_process_exit(): @@ -84,3 +215,100 @@ def test_output_reaches_pipe_before_process_exit(): "Child exited before output was observed. " f"Received stdout: {bytes(output)!r}; stderr: {stderr!r}" ) + + +@pytest.mark.skipif( + sys.platform != "linux", reason="relies on CLOCK_MONOTONIC being shared across processes" +) +def test_quiet_line_reaches_pipe_within_two_flush_intervals(): + child = r""" +import time +from airbyte_cdk.logger import PRINT_BUFFER, init_logger + +PRINT_BUFFER.flush_interval = 0.5 +logger = init_logger("airbyte") +with PRINT_BUFFER: + print('{"type":"RECORD","record":{"stream":"s","data":{"id":1},"emitted_at":0}}\n', end="") + logger.info(f"quiet-line sent_at={time.monotonic()!r}") + time.sleep(30) +""" + process = _spawn_child(child) + try: + output, seen = _read_until(process, [b"sent_at="], timeout=10) + stderr = b"" + match = re.search(rb"sent_at=([0-9.]+)", output) + assert match is not None, f"quiet-line not observed. stdout: {output!r}" + sent_at = float(match.group(1)) + latency = seen[b"sent_at="] - sent_at + print(f"\nquiet-line latency: {latency:.3f}s") + assert latency < 2 * 0.5, f"quiet-line took {latency:.3f}s to reach the pipe" + assert process.poll() is None, "child exited before output was observed" + finally: + _kill_process_group(process) + stderr = process.stderr.read() + if stderr: + print(f"\nchild stderr: {stderr!r}") + + +def test_pending_output_flushed_on_exit(): + child = r""" +import logging +from airbyte_cdk.logger import PRINT_BUFFER, init_logger + +logger = init_logger("airbyte") +with PRINT_BUFFER: + for i in range(1000): + print('{"type":"RECORD","record":{"stream":"s","data":{"id":%d,"e":0},"emitted_at":0}}\n' % i, end="") + logger.info("log-line-%d-done", i) +logger.info("after-with") +PRINT_BUFFER.write("raw-write-no-flush\n") +""" + process = _spawn_child(child) + try: + stdout, stderr = process.communicate(timeout=60) + finally: + _kill_process_group(process) + + for i in range(1000): + assert stdout.count(b'"id":%d,' % i) == 1, f"RECORD {i} missing or duplicated" + assert stdout.count(b"log-line-%d-done" % i) == 1, f"log line {i} missing or duplicated" + assert b"after-with" in stdout + assert b"raw-write-no-flush" in stdout + + +@pytest.mark.skipif(not hasattr(os, "fork"), reason="requires os.fork") +def test_forked_child_gets_a_working_sweeper(): + child = r""" +import os, time +from airbyte_cdk.logger import PRINT_BUFFER, init_logger + +logger = init_logger("airbyte") +with PRINT_BUFFER: + logger.info("before-fork") + pid = os.fork() + if pid == 0: + logger.info("in-fork-child") + time.sleep(30) + os._exit(0) + else: + logger.info("in-fork-parent") + time.sleep(30) +""" + process = _spawn_child(child) + try: + output, seen = _read_until(process, [b"in-fork-child", b"in-fork-parent"], timeout=10) + assert b"in-fork-child" in seen, ( + f"forked child's log line never arrived. stdout: {output!r}" + ) + assert b"in-fork-parent" in seen, ( + f"fork parent's log line never arrived. stdout: {output!r}" + ) + assert process.poll() is None, "child exited before output was observed" + assert output.count(b"before-fork") == 1, ( + f"pre-fork output emitted != 1 time. stdout: {output!r}" + ) + finally: + _kill_process_group(process) + stderr = process.stderr.read() + if stderr: + print(f"\nchild stderr: {stderr!r}") From 4d03a380a4ef3b0da382f0cee3dbd2eca7ac56d6 Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Tue, 29 Sep 2026 20:33:07 +0000 Subject: [PATCH 3/5] perf: guard sweeper notification with a plain bool on the write hot path Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- airbyte_cdk/utils/print_buffer.py | 17 +++++++++++++---- unit_tests/utils/test_print_buffer.py | 2 -- 2 files changed, 13 insertions(+), 6 deletions(-) diff --git a/airbyte_cdk/utils/print_buffer.py b/airbyte_cdk/utils/print_buffer.py index c0b578cb66..7fa6e90271 100644 --- a/airbyte_cdk/utils/print_buffer.py +++ b/airbyte_cdk/utils/print_buffer.py @@ -110,11 +110,13 @@ def __init__(self, flush_interval: float = 0.1): self._stop = threading.Event() self._pending = threading.Event() self._sync_flush = False + self._pending_marked = False def write(self, message: str) -> None: with self.lock: self.buffer.write(message) - self._mark_pending() + if not self._pending_marked: + self._mark_pending() current_time = time.monotonic() if (current_time - self.last_flush_time) >= self.flush_interval: self.flush() @@ -132,7 +134,8 @@ def flush(self) -> None: self._flush_to_fd() else: self._drain() - self._mark_pending() + if not self._pending_marked: + self._mark_pending() def _drain(self) -> None: combined_message = self.buffer.getvalue() @@ -148,8 +151,10 @@ def _mark_pending(self) -> None: return if self._sweeper is None: self._start_sweeper() - if not self._pending.is_set(): - self._pending.set() + if self._sweeper is None: + return # _start_sweeper fell back to synchronous flush + self._pending.set() + self._pending_marked = True def _start_sweeper(self) -> None: stop, pending = threading.Event(), threading.Event() @@ -174,6 +179,8 @@ def _sweep(self, stop: threading.Event, pending: threading.Event) -> None: stop.wait(self.flush_interval) # coalesce for one interval with self.lock: pending.clear() + if pending is self._pending: + self._pending_marked = False try: self._flush_to_fd() except (OSError, ValueError): # broken pipe / closed stdout @@ -184,6 +191,7 @@ def _stop_sweeper(self) -> None: with self.lock: thread, stop, pending = self._sweeper, self._stop, self._pending self._sweeper = None + self._pending_marked = False if thread is None: return stop.set() @@ -207,6 +215,7 @@ def _reset_after_fork(self) -> None: self._sweeper = None self._stop = threading.Event() self._pending = threading.Event() + self._pending_marked = False def __enter__(self) -> "PrintBuffer": self.old_stdout, self.old_stderr = sys.stdout, sys.stderr diff --git a/unit_tests/utils/test_print_buffer.py b/unit_tests/utils/test_print_buffer.py index 1e1f79d6b9..92d2d19eb3 100644 --- a/unit_tests/utils/test_print_buffer.py +++ b/unit_tests/utils/test_print_buffer.py @@ -235,7 +235,6 @@ def test_quiet_line_reaches_pipe_within_two_flush_intervals(): process = _spawn_child(child) try: output, seen = _read_until(process, [b"sent_at="], timeout=10) - stderr = b"" match = re.search(rb"sent_at=([0-9.]+)", output) assert match is not None, f"quiet-line not observed. stdout: {output!r}" sent_at = float(match.group(1)) @@ -252,7 +251,6 @@ def test_quiet_line_reaches_pipe_within_two_flush_intervals(): def test_pending_output_flushed_on_exit(): child = r""" -import logging from airbyte_cdk.logger import PRINT_BUFFER, init_logger logger = init_logger("airbyte") From ca58efe19e234d219a10b269ebc53a644861a0a1 Mon Sep 17 00:00:00 2001 From: "zane.hyatt" Date: Tue, 29 Sep 2026 20:44:10 +0000 Subject: [PATCH 4/5] chore: document intentionally ignored flush errors in PrintBuffer Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- airbyte_cdk/utils/print_buffer.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/airbyte_cdk/utils/print_buffer.py b/airbyte_cdk/utils/print_buffer.py index 7fa6e90271..604ab112be 100644 --- a/airbyte_cdk/utils/print_buffer.py +++ b/airbyte_cdk/utils/print_buffer.py @@ -48,6 +48,7 @@ def _before_fork() -> None: try: b._flush_to_fd() # the child must not inherit (and later re-emit) pending bytes except (OSError, ValueError): + # best effort: a broken/closed stdout must not block the fork pass @@ -207,6 +208,7 @@ def _shutdown(self) -> None: try: self._flush_to_fd() except (OSError, ValueError): + # stdout already broken/closed at shutdown; nothing more can be delivered pass def _reset_after_fork(self) -> None: From ad5e586cbfcb85db4f00cc65220c8a07f1d18e98 Mon Sep 17 00:00:00 2001 From: ZaneHyattAB Date: Mon, 5 Oct 2026 15:36:38 -0700 Subject: [PATCH 5/5] fix: fall back to synchronous flush on any sweeper error The sweeper only caught OSError/ValueError. Any other exception killed the thread while _sweeper still pointed at it, so later writes marked output pending for a dead thread and it sat in the stdout buffer again, silently. Treat any exception like a broken pipe: switch to synchronous fd flushes and exit. Co-Authored-By: Claude Opus 5.5 --- airbyte_cdk/utils/print_buffer.py | 4 +++- unit_tests/utils/test_print_buffer.py | 25 +++++++++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/airbyte_cdk/utils/print_buffer.py b/airbyte_cdk/utils/print_buffer.py index 604ab112be..29bd0e2d11 100644 --- a/airbyte_cdk/utils/print_buffer.py +++ b/airbyte_cdk/utils/print_buffer.py @@ -184,7 +184,9 @@ def _sweep(self, stop: threading.Event, pending: threading.Event) -> None: self._pending_marked = False try: self._flush_to_fd() - except (OSError, ValueError): # broken pipe / closed stdout + except Exception: + # broken pipe / closed stdout, or anything unexpected: hand fd flushes back to + # callers so output never waits on a dead sweeper self._sync_flush = True return diff --git a/unit_tests/utils/test_print_buffer.py b/unit_tests/utils/test_print_buffer.py index 92d2d19eb3..ce47cbcfbe 100644 --- a/unit_tests/utils/test_print_buffer.py +++ b/unit_tests/utils/test_print_buffer.py @@ -35,6 +35,31 @@ def test_flush_drains_without_fd_flush_and_sweeper_flushes(monkeypatch): assert flushed.wait(timeout=10) +def test_unexpected_sweeper_error_falls_back_to_synchronous_flush(monkeypatch): + sweeper_failed = threading.Event() + mock_stdout = MagicMock() + + def flush_fails_once(): + if not sweeper_failed.is_set(): + sweeper_failed.set() + raise RuntimeError("unexpected") + + mock_stdout.flush.side_effect = flush_fails_once + monkeypatch.setattr(sys, "__stdout__", mock_stdout) + + with PrintBuffer(flush_interval=0.01) as print_buffer: + print_buffer.write("a") + assert sweeper_failed.wait(timeout=10) + print_buffer._sweeper.join(timeout=10) + assert not print_buffer._sweeper.is_alive() + assert print_buffer._sync_flush + + mock_stdout.flush.reset_mock() + print_buffer.flush() + + mock_stdout.flush.assert_called_once() + + class _CountingRaw(io.RawIOBase): def __init__(self): self.write_calls = 0