feat: batch PrintBuffer fd flushes with a background sweeper

[ZaneHyattAB]

Follow-up to #1167 (merged). This PR only adds the sweeper on top of it.

Summary

#1167 made PrintBuffer.flush() also call sys.__stdout__.flush(), so output reaches a real pipe promptly. But logging.StreamHandler.emit calls flush() after every record, so every log line became one write(2): 23x the syscalls and about 10% lower throughput at 1 log line per record. This PR keeps #1167's delivery guarantee and batches the syscalls again. The fd flush moves off the hot path and into a lazy background sweeper.

flush():                       # hot path, called per log line
    drain StringIO -> sys.__stdout__   # userspace only, no syscall
    if not _pending_marked: _mark_pending()   # plain bool; first call lazily starts the sweeper

_sweep():                      # daemon thread "airbyte-print-buffer-sweeper"
    loop until _stop:
        _pending.wait()                # idle: blocked, zero CPU
        _stop.wait(flush_interval)     # coalesce for one interval (default 100 ms)
        with self.lock:                # same RLock as write/flush -> no torn messages
            _pending.clear(); _flush_to_fd()   # drain + sys.__stdout__.flush()

Lifecycle

• The sweeper starts on the first write/flush. __enter__ alone does not start it. Idle buffers never start a thread.
• __exit__: stop and join the sweeper (never while holding self.lock), then flush everything pending to the fd synchronously, then restore sys.stdout/sys.stderr. A later write starts a fresh sweeper, so re-entering the context works.
• Exit paths that skip __exit__: the first sweeper start also registers a module-level atexit hook (_shutdown_all). It switches each registered buffer to _sync_flush, stops and joins its sweeper, then does a final fd flush. atexit runs before the interpreter tears down daemon threads, so pending output is never left to a killed daemon thread. Buffers are tracked in a WeakSet.
• Fallback: if Thread.start() raises RuntimeError (for example at interpreter shutdown or a thread limit), the buffer switches to _sync_flush, which gives fix: flush sys.__stdout__ in PrintBuffer.flush() #1167's behaviour. If the sweeper's own fd flush raises anything (a broken pipe or closed stdout, or an unexpected error), the sweeper sets _sync_flush and exits, so output never waits on a dead sweeper and the next caller-thread flush() raises the error as it did before.

Fork safety: os.register_at_fork (skipped where unavailable, e.g. Windows):

• before: acquire every registered buffer's lock and flush pending bytes to the fd, so the child cannot re-emit them.
• after_in_parent: release the locks. The parent's sweeper keeps running.
• after_in_child: replace lock/_stop/_pending, and set _sweeper = None and _pending_marked = False. The child starts its own sweeper on its next write instead of believing a dead thread is running.

Unchanged for non-PrintBuffer users: nothing happens at import time. No hooks are registered and no thread is started until a PrintBuffer is written to.

Measurements

Method from the #1167 review thread: real child-process stdout pipe, real airbyte_cdk logger via PRINT_BUFFER and entrypoint-style print, strace -f -c -e trace=write, 3 runs each, Python 3.12.

Branch	1:1 (200k rec) rec/s	1:1 write()	1:1000 (1M rec) rec/s	1:1000 write()	Quiet-line latency
before #1167	79,201 / 78,869 / 71,220	9,092 / 9,092 / 9,092	558,912 / 527,816 / 533,837	1,020 / 1,021 / 1,021	3.001 / 3.001 / 3.001 s (only at exit)
#1167	70,063 / 71,179 / 71,143	200,059 / 200,046 / 200,061	546,177 / 526,588 / 529,110	1,018 / 1,020 / 1,020	0.000 / 0.000 / 0.000 s
this PR	79,778 / 79,690 / 78,530	9,107 / 9,103 / 9,105	521,099 / 525,892 / 547,932	1,025 / 1,030 / 1,033	0.101 / 0.100 / 0.100 s

Quiet-line latency is the time from logging one line and then going idle (the child sleeps 3 s before exiting) until the parent reads it from the pipe. On main, the line only arrives when the process exits.

strace adds a lot of noise at 1:1000. An interleaved run without strace (5 runs each, 1M records, 1:1000, sorted) shows no measurable difference: before #1167 611,586 to 633,211; #1167 507,014 to 621,803; this PR 598,663 to 631,848 rec/s.

Tests (unit_tests/utils/test_print_buffer.py)

• fix: flush sys.__stdout__ in PrintBuffer.flush() #1167's test_output_reaches_pipe_before_process_exit is unchanged and passes.
• The immediate-fd-flush unit test is replaced by a deferred one: flush() writes to stdout without calling stdout.flush(), and the sweeper calls it later. This is event-based, not sleep-based.
• __exit__ flushes everything pending and the sweeper thread is dead afterwards.
• Syscall count: 2,000 log lines through the real airbyte logger and a counting raw stream. Raw writes must stay at or below bytes // 8192 + 1 + elapsed / interval + 3. fix: flush sys.__stdout__ in PrintBuffer.flush() #1167's implementation fails this, which was checked by restoring it.
• Real pipe, subprocess: a line logged just before a quiet period arrives within 2 * flush_interval, and the child is still alive when it arrives.
• Real pipe, subprocess: all 1,000 records and log lines written inside with PRINT_BUFFER: reach the pipe exactly once on __exit__. A log line and a raw write() made after the context has exited (with no later flush) are delivered by the atexit hook.
• An unexpected exception in the sweeper's fd flush switches the buffer to synchronous flushing and the thread exits (fails without the fallback).
• os.fork() in a subprocess: parent and child each deliver post-fork lines through their own sweeper, and pre-fork output is not duplicated.
• Subprocess reads use select with generous deadlines and process-group cleanup. Nothing uses sleep for synchronization.

poetry run pytest unit_tests/utils/test_print_buffer.py unit_tests/test_logger.py unit_tests/test_entrypoint.py passed (66 passed, 1 skipped on macOS: the quiet-line latency test is Linux-only). The print-buffer file passed on repeated runs. ruff check, ruff format and mypy are clean.

Where output can still be delayed or lost

• Delay: bytes can sit in the sys.__stdout__ userspace buffer for about one flush_interval (100 ms) plus scheduling jitter before reaching the fd. A long-held self.lock or a starved thread can stretch that.
• Loss on hard termination: os._exit(), SIGKILL, a fatal signal, or a default-disposition SIGTERM skips atexit, so up to about one interval of output is lost. Before fix: flush sys.__stdout__ in PrintBuffer.flush() #1167, up to 8 KiB plus the StringIO contents was lost in the same cases. fix: flush sys.__stdout__ in PrintBuffer.flush() #1167 loses only the StringIO contents.
• Broken or closed stdout: the sweeper stops and the buffer switches to synchronous flushing. The error then shows up on the next caller-thread flush(), but the bytes are gone either way.
• _sync_flush fallback cases (thread start failure, after atexit, after a sweeper error) behave like fix: flush sys.__stdout__ in PrintBuffer.flush() #1167: one fd flush per flush().

Link to Devin session: https://app.devin.ai/sessions/e843f60236164bd4bd8b56d6ee7c009b
Open in Devin Desktop: https://app.devin.ai/desktop/session/e843f60236164bd4bd8b56d6ee7c009b?variant=devin
Requested by: ZaneHyattAB

[devin-ai-integration]

I'll fix CI failures and address comments from users with write access. I'll skip comments containing "(aside)".

• Disable automatic comment, CI, and merge conflict monitoring

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@devin/1790713331-print-buffer-sweeper#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch devin/1790713331-print-buffer-sweeper

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Fast)

4 988 tests  +6   4 976 ✅ +6   11m 57s ⏱️ + 2m 24s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit ad5e586. ± Comparison against base commit d214b0c.

This pull request removes 1 and adds 7 tests. Note that renamed tests count towards both.

unit_tests.utils.test_print_buffer ‑ test_flush_pushes_through_underlying_stdout

unit_tests.utils.test_print_buffer ‑ test_exit_flushes_pending_and_stops_sweeper
unit_tests.utils.test_print_buffer ‑ test_flush_drains_without_fd_flush_and_sweeper_flushes
unit_tests.utils.test_print_buffer ‑ test_forked_child_gets_a_working_sweeper
unit_tests.utils.test_print_buffer ‑ test_pending_output_flushed_on_exit
unit_tests.utils.test_print_buffer ‑ test_quiet_line_reaches_pipe_within_two_flush_intervals
unit_tests.utils.test_print_buffer ‑ test_syscall_count_stays_batched_at_one_log_line_per_record
unit_tests.utils.test_print_buffer ‑ test_unexpected_sweeper_error_falls_back_to_synchronous_flush

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

4 991 tests  +6   4 979 ✅ +6   9m 48s ⏱️ - 6m 16s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit ad5e586. ± Comparison against base commit d214b0c.

This pull request removes 1 and adds 7 tests. Note that renamed tests count towards both.

unit_tests.utils.test_print_buffer ‑ test_flush_pushes_through_underlying_stdout

unit_tests.utils.test_print_buffer ‑ test_exit_flushes_pending_and_stops_sweeper
unit_tests.utils.test_print_buffer ‑ test_flush_drains_without_fd_flush_and_sweeper_flushes
unit_tests.utils.test_print_buffer ‑ test_forked_child_gets_a_working_sweeper
unit_tests.utils.test_print_buffer ‑ test_pending_output_flushed_on_exit
unit_tests.utils.test_print_buffer ‑ test_quiet_line_reaches_pipe_within_two_flush_intervals
unit_tests.utils.test_print_buffer ‑ test_syscall_count_stays_batched_at_one_log_line_per_record
unit_tests.utils.test_print_buffer ‑ test_unexpected_sweeper_error_falls_back_to_synchronous_flush

♻️ This comment has been updated with latest results.