Fix SSH tunnels not forwarding when the task holds over 1024 file descriptors

[firasbouzazi]

Follow-up to #74211. SSHTunnel._serve_forever waited with select.select(), which cannot watch a descriptor numbered 1024 or higher. The ValueError was caught and ended the forwarding thread, so with that many descriptors open SSHHook.get_tunnel() left a local port that accepted connections but never forwarded them.

The loop now uses selectors.DefaultSelector. Since connections come and go, the listening and shutdown sockets are registered once, and each forwarded connection's socket and channel are registered when accepted and unregistered before they are closed. Unregistering uses the descriptor recorded at registration: a closed socket has no descriptor, and paramiko.Channel.fileno() on a closed channel would create a new pipe instead of returning the old one. Connections whose channel the remote end closed are now closed and unregistered rather than only dropped from the list.

Tests: the in-process paramiko server fixtures from #74211 move to a shared conftest.py and now echo forwarded (direct-tcpip) channels. New tests forward data through SSHTunnel sequentially and concurrently while the process holds over 1024 descriptors (no data gets through before the fix), and check that connections closed by either end are unregistered. Against a local sshd, SSHHook.get_tunnel() with 1,100 extra descriptors open went from 0/10 to 10/10 round trips. The SSH provider unit tests (209) pass.

related: #74205

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code

Generated-by: Claude Code following the guidelines

[firasbouzazi]

@potiuk

[firasbouzazi]

The failing jobs look unrelated to this change: DB-core:Postgres:14:3.10:Always runs core tests, and the SSH provider tests from Compat 3.1.8 pass locally against Airflow 3.1.8 (209/209). Several jobs in this run also took 40+ minutes (e.g. the shared secrets_masker/plugins_manager tests, which take ~20s on main), so it looks like a runner issue. Could someone re-run the failed jobs? Thanks!