Repository navigation
Conversation
The test that is supposed to exercise the parent reading a partial line never did. The child's stdout is line buffered, so writing "Message " without a newline left the text in the child's own buffer instead of sending it down the socket; the parent only ever saw the complete line once the second write flushed it. The sleep the test relied on to make the parent read a partial line had nothing to read during it, and both outcomes produce the same log record, so the missing coverage could never show up as a failure. The child now flushes the partial line and waits for the parent to confirm it has consumed it before writing the rest, so the partial-line path is guaranteed to be taken and a regression there fails the test instead of passing quietly.
potiuk
added
the
closed because of open PR limit
|
Hello @ColtenOuO - thank you for your contributions to Apache Airflow! The Airflow community has introduced a limit of 5 open pull requests at a time for contributors without write access to the repository. You currently have 19 open pull requests, so - as a one-time step of introducing the limit - we closed the ones where maintainers have not engaged yet:
These pull requests stay open because maintainers are already engaged in them - they count towards your limit:
This is not a judgement of you or of your changes. We never told contributors before that opening many pull requests at once was a problem, so there is nothing to feel bad about - and nothing is lost: your branches, commits and the review history stay where they are. What we ask you to do is to make your first prioritization decision: choose which of the pull requests above matter most to you, and reopen them (up to 5 open at a time, including the ones still open) with the "Reopen pull request" button or While your pull requests are waiting for review, the most valuable thing you can do is help in other ways - reviewing other contributors' pull requests, helping with issues, and taking part in the discussions on the devlist and Slack. Why we introduced the limit, what it means for you and how to reopen or restore a pull request is explained in https://github.com/apache/airflow/blob/main/contributing-docs/32_open_pull_request_limit.rst. Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting |
2 participants
Summary
TestWatchedSubprocess::test_reading_from_pipesexists to prove that the supervisor reassembles a log line that arrives across two separate writes. Its own comment admitted the mechanism was shaky:Investigating that fragility turned up something worse than a timing race: the test never exercised the split at all, and could not have failed if the partial-line handling broke.
The child process writes the first half of the line with
sys.stdout.write("Message "). Inside the forked child,sys.stdoutis anio.TextIOWrapper(binary, line_buffering=True)(see_reopen_std_io_handlesintask-sdk/src/airflow/sdk/execution_time/supervisor.py). Line buffering only flushes on a newline, and"Message "has none — so that text sat in the child's own userspace buffer and never reached the socket. The parent had nothing partial to read during thesleep(0.1); the bytes were only handed over later, when the second write appended"split across two writes\n"and the newline triggered a single flush of the whole line.Both paths produce exactly the same assertion target — one
task.stdoutrecord readingMessage split across two writes— so the test passed either way. This was not a flaky failure but a silent loss of coverage: the partial-line branch ofmake_buffered_socket_readercould have regressed at any point and this test would have kept passing.Test
By wrapping the parent's
task.stdouthandler and usingrecv(..., MSG_PEEK)to record the exact bytes the parent saw on each read, with and without an explicit flush in the child:mainb"I'm a short message\n"thenb'Message split across two writes\n'flush()addedb"I'm a short message\nMessage "thenb'split across two writes\n'The first row is the bug: the parent's second read already contains the complete line, terminating newline included. No partial line ever crossed the socket.
What was happening before
sequenceDiagram autonumber participant C as child code participant B as child stdout buffer<br/>TextIOWrapper, line buffered participant P as parent supervisor C->>B: print "I'm a short message" B->>P: flush on newline: I'm a short message + LF Note over P: first read emits one log record C->>B: write "Message " Note over B: no newline, so no flush<br/>the bytes never reach the socket Note over C,P: sleep 0.1s — parent sits in select with nothing to read C->>B: write "split across two writes" + LF B->>P: flush on newline: Message split across two writes + LF Note over P: second read already holds the complete line<br/>the partial-line branch is never takenWhat happens now
sequenceDiagram autonumber participant C as child code participant B as child stdout buffer participant P as parent supervisor C->>B: print "I'm a short message" B->>P: flush on newline: I'm a short message + LF P-->>C: set stdout_consumed after first read Note over C: wait for stdout_consumed, then clear it C->>B: write "Message " C->>B: flush B->>P: Message Note over P: second read has no newline<br/>the partial line is held in the reader buffer P-->>C: set stdout_consumed after second read Note over C: wait for stdout_consumed C->>B: write "split across two writes" + LF B->>P: split across two writes + LF Note over P: third read completes the buffered line<br/>and emits one log recordThe guarantee is structural rather than probabilistic. When the second wait returns, the only thing the parent can have consumed is the partial line, because the first wait already confirmed the short message was drained and then cleared the event. If the parent were to somehow not read the partial line, the child raises after a 10 second timeout and exits non-zero, so
assert rc == 0fails the test — the old "quietly skipped the interesting case" outcome is no longer reachable.The reworked test was run 20 times in a row, passing 20/20, with no sleep left in it.
Change
task-sdk/tests/task_sdk/execution_time/test_supervisor.pysleep(0.1)is replaced by a two-step handshake: the test wrapsActivitySubprocess._create_log_forwarderfor thetask.stdoutchannel only and signals amultiprocessing.Eventafter each read; the child waits for the short message to be consumed, writes and flushes the partial line, waits for that to be consumed, and only then writes the remainder.Was generative AI tooling used to co-author this PR?