Function wait_for_socket

scripts/repro_live_duplicate.py:113–126

(path: Path, timeout_s: float = 20.0)

def wait_for_socket(path: Path, timeout_s: float = 20.0) -> None:
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        if path.exists():
            try:
                s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
                s.settimeout(0.2)
                s.connect(str(path))
                s.close()
                return
            except OSError:
                pass
        time.sleep(0.02)
    raise RuntimeError(f"socket not ready: {path}")


# ──────────────────────────────────────────────────────────────────────────────