ColtenOuO opened a new pull request, #72585:
URL: https://github.com/apache/airflow/pull/72585
### Summary
`TestWatchedSubprocess::test_reading_from_pipes` exists to prove that the
supervisor reassembles a log line that arrives across two separate writes. Its
own comment admitted the mechanism was shaky:
```python
# We need a short sleep for the main process to process things. I worry this
timing will be
# fragile, but I can't think of a better way. This lets the stdout be read
(partial line) and the
# stderr full line be read
sleep(0.1)
```
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.stdout` is an
`io.TextIOWrapper(binary, line_buffering=True)` (see `_reopen_std_io_handles`
in `task-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 the `sleep(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.stdout`
record reading `Message 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 of `make_buffered_socket_reader` could have regressed at
any point and this test would have kept passing.
### Test
By wrapping the parent's `task.stdout` handler and using `recv(...,
MSG_PEEK)` to record the exact bytes the parent saw on each read, with and
without an explicit flush in the child:
| version | bytes the parent actually read |
| --- | --- |
| current `main` | `b"I'm a short message\n"` then `b'Message split across
two writes\n'` |
| with `flush()` added | `b"I'm a short message\nMessage "` then `b'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
```mermaid
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 taken
```
### What happens now
```mermaid
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 record
```
The 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 == 0`
fails 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.py`
- The child flushes after writing the partial line, so it actually reaches
the parent.
- `sleep(0.1)` is replaced by a two-step handshake: the test wraps
`ActivitySubprocess._create_log_forwarder` for the `task.stdout` channel only
and signals a `multiprocessing.Event` after 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?
- [X] Yes — Claude Code (Opus 5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
