`connect_read_pipe` double-closes the underlying fd, possibly closing an unrelated fd

まだ誰も着手していません。

In our production use of uvloop, we found sporadic cases of file descriptors being randomly closed. (For more details of our specific case, see skypilot-org/skypilot#10681).
We traced this back to `loop.connect_read_pipe` double-closing its file descriptor. When the fd was re-allocated between the first and second close, this caused an unrelated thread to have its fd closed out from under.

I used this script to check the double-close behavior:
```python
import asyncio, io, os, sys

class BufferedReaderWithProbe(io.BufferedReader):
def close(self):
# Probe before status
fd = self.fileno()
print(f"closing {self} (fd {fd})")
try:
os.fstat(fd); fd_state = "still OPEN"
except OSError as e:
fd_state = f"already closed ({e.strerror})"
# Open a new pipe to see what fds we get
# Kernel should reuse lowest free number
newfd_reader, newfd_writer = os.pipe()
print(f" is fd {fd} already closed? {fd_state}")
print(f" fresh os.pipe() got fds {newfd_reader},{newfd_writer}")

# Do normal BufferedReader close
try:
print("calling BufferedReader.close")
super().close()
finally:

# Probe after status
print("Status after close:")
for fd in (newfd_reader, newfd_writer):
try:
os.fstat(fd); print(f" fresh fd {fd}: alive")
except OSError as e:
print(f" fresh fd {fd}: DEAD ({e.strerror}) <-- stolen")

async def main():
loop = asyncio.get_running_loop()
pipe_read_fd, pipe_write_fd = os.pipe()
probe_reader = BufferedReaderWithProbe(io.FileIO(pipe_read_fd, "rb"))
transport, _ = await loop.connect_read_pipe(asyncio.Protocol, probe_reader)
print(f"transport {type(transport).__name__} on fd {pipe_read_fd}")
transport.close()

if __name__ == "__main__":
if sys.argv[1:] == ["uvloop"]:
import uvloop; uvloop.install(); print(f"== uvloop {uvloop.__version__}")
else:
print(f"== stdlib asyncio {sys.version.split()[0]}")
asyncio.run(main())
```

```
$ python repro.py
== stdlib asyncio 3.10.16
transport _UnixReadPipeTransport on fd 6
closing (fd 6)
is fd 6 already closed? still OPEN
fresh os.pipe() got fds 8,9
calling BufferedReader.close
Status after close:
fresh fd 8: alive
fresh fd 9: alive
$ python repro.py uvloop
== uvloop 0.22.1
transport ReadUnixTransport on fd 13
closing (fd 13)
is fd 13 already closed? already closed (Bad file descriptor)
fresh os.pipe() got fds 13,16
calling BufferedReader.close
Status after close:
fresh fd 13: DEAD (Bad file descriptor) <-- stolen
fresh fd 16: alive
```

Here you can see that a asyncio event loop correctly leaves fd 6 open, allowing the underlying file object (in this case, our custom BufferedReader) to close the fd.
However, uvloop has already closed the fd, so when the file object is closed, it re-closes fd 13 which is now an unrelated `os.pipe()` fd.

This specific case was already fixed for sockets: https://github.com/MagicStack/uvloop/commit/d5195d7c10fbae81ef1dcb4609cee50f8aa746fa. It was assumed that the EBADF for non-sockets was benign, but it can actually be a problem if the underlying fd is reassigned before the `_fileobj.close()` can run.

You can observe the same behavior if the transport is not explicitly closed but is instead GC'd - although in this case it seems that the file object is closed before the `uv_close` call double-closes the fd. (The order is inverted.) This is actually the case we saw in prod.

It also looks like `connect_write_pipe` is similarly affected.

AI disclosure: This investigation was heavily assisted by AI tools, but this report was entirely human written without assistance except for the original repro script. The original repro script was then modified by me for clarity and manually validated.

トランスポートのエントリポイントである connect_read_pipe と connect_write_pipe から始め、それらのファイルディスクリプタのクリーンアップを、コミット d5195d7c10fbae81ef1dcb4609cee50f8aa746fa のソケット修正と比較します。提供された再現コードを uvloop と標準の asyncio ループで実行します。いずれかのパイプトランスポートを閉じるかガベージコレクションしても、無関係なファイルディスクリプタが閉じられなければ完了です。

索引モデルが issue の本文から書いたものです。