Studio: only advertise a Cloudflare tunnel once it actually serves (#6264)

studio/backend/cloudflare_tunnel.py

@@ -24,12 +24,20 @@ from pathlib import Path
 from typing import Optional, Tuple
 
 # cloudflared logs the quick-tunnel URL; match only the URL so we do not depend
-# on the surrounding wording, which Cloudflare may change.
-_URL_RE = re.compile(r"https://[A-Za-z0-9-]+\.trycloudflare\.com")
+# on the surrounding wording, which Cloudflare may change. The negative lookahead
+# drops cloudflared's own API host, which appears in failure lines such as
+#   failed to request quick Tunnel: Post "https://api.trycloudflare.com/tunnel"
+# and must never be mistaken for a usable tunnel URL.
+_URL_RE = re.compile(r"https://(?!api\.)[A-Za-z0-9-]+\.trycloudflare\.com")
+
+# cloudflared logs this once per edge connection it establishes. Until at least
+# one appears the quick-tunnel URL returns Cloudflare error 1033 (HTTP 530), so
+# we wait for it before advertising the URL.
+_REGISTERED_MARKER = "Registered tunnel connection"
 
 _RELEASE_BASE = "https://github.com/cloudflare/cloudflared/releases/latest/download"
 
-_URL_TIMEOUT = 15.0  # seconds to wait for the public URL before giving up
+_READY_TIMEOUT = 15.0  # seconds to wait for the URL + a registered edge connection
 _DOWNLOAD_TIMEOUT = 60  # urlopen timeout for the one-time binary download
 
 
@@ -180,13 +188,24 @@ class CloudflareTunnel:
     upstream stays local-only.
     """
 
-    def __init__(self, port: int, binary: str):
+    def __init__(
+        self,
+        port: int,
+        binary: str,
+        protocol: Optional[str] = None,
+    ):
         self.port = port
         self.binary = binary
+        # None lets cloudflared pick its default (quic, with its own http2
+        # fallback); set to "http2" to force it when quic is blocked.
+        self.protocol = protocol
         self._proc: Optional[subprocess.Popen] = None
         self._lock = threading.Lock()
+        self._stopped = False
         self._url_event = threading.Event()
+        self._ready_event = threading.Event()
         self.url: Optional[str] = None
+        self.ready = False
         self.error: Optional[str] = None
 
     def start(self) -> None:
@@ -197,25 +216,33 @@ class CloudflareTunnel:
             f"http://localhost:{self.port}",
             "--no-autoupdate",
         ]
-        proc = subprocess.Popen(
-            cmd,
-            stdout = subprocess.PIPE,
-            stderr = subprocess.STDOUT,
-            stdin = subprocess.DEVNULL,
-            text = True,
-            errors = "replace",
-            bufsize = 1,
-            **_windows_hidden_kwargs(),
-        )
+        if self.protocol:
+            cmd += ["--protocol", self.protocol]
         with self._lock:
+            # A stop() that landed before us (e.g. a shutdown in the caller's
+            # register->start window) marks the tunnel stopped; spawning now would
+            # orphan a process nobody owns, so refuse.
+            if self._stopped:
+                return
+            proc = subprocess.Popen(
+                cmd,
+                stdout = subprocess.PIPE,
+                stderr = subprocess.STDOUT,
+                stdin = subprocess.DEVNULL,
+                text = True,
+                errors = "replace",
+                bufsize = 1,
+                **_windows_hidden_kwargs(),
+            )
             self._proc = proc
         threading.Thread(
             target = self._reader, args = (proc,), name = "cloudflared-reader", daemon = True
         ).start()
 
     def _reader(self, proc: subprocess.Popen) -> None:
-        # Drain cloudflared's output, capture the first trycloudflare URL, and
-        # keep draining so it never blocks on a full pipe.
+        # Drain cloudflared's output: capture the first trycloudflare URL and the
+        # first edge-connection registration, and keep draining so it never
+        # blocks on a full pipe.
         try:
             if proc.stdout is not None:
                 for line in proc.stdout:
@@ -224,20 +251,35 @@ class CloudflareTunnel:
                         if match:
                             self.url = match.group(0)
                             self._url_event.set()
+                    if not self.ready and _REGISTERED_MARKER in line:
+                        self.ready = True
+                        self._ready_event.set()
         except Exception:
             pass
         finally:
+            # stdout closed -> cloudflared has exited. Record why, and unblock any
+            # waiters at once instead of letting them wait out the full timeout.
             if self.url is None:
                 self.error = "cloudflared exited before emitting a tunnel URL"
-                self._url_event.set()
+            elif not self.ready:
+                self.error = "cloudflared exited before the tunnel connection registered"
+            self._url_event.set()
+            self._ready_event.set()
 
-    def wait_for_url(self, timeout: float = _URL_TIMEOUT) -> Optional[str]:
-        self._url_event.wait(timeout)
-        return self.url
+    def wait_for_ready(self, timeout: float = _READY_TIMEOUT) -> Optional[str]:
+        """Block until the tunnel is actually serving -- the URL has been minted
+        *and* at least one edge connection has registered -- or until timeout.
+
+        Returns the URL only when ready, so callers never advertise a URL that
+        would return Cloudflare error 1033 (HTTP 530)."""
+        self._ready_event.wait(timeout)
+        return self.url if self.ready else None
 
     def stop(self) -> None:
         """Terminate the tunnel. Idempotent and safe to call from a signal handler."""
         with self._lock:
+            # Mark stopped so a start() racing behind us refuses to spawn.
+            self._stopped = True
             proc, self._proc = self._proc, None
         if proc is None:
             return
@@ -260,43 +302,74 @@ class CloudflareTunnel:
 # enough; the lock guards the start/stop/shutdown races.
 _active_tunnel: Optional[CloudflareTunnel] = None
 _active_lock = threading.Lock()
+# Latched by stop_studio_tunnel so a shutdown landing *between* a start's retry
+# attempts aborts the loop instead of starting a tunnel nobody will ever stop.
+_shutdown_requested = False
 
 
-def start_studio_tunnel(port: int, timeout: float = _URL_TIMEOUT) -> Optional[str]:
-    """Start a quick tunnel and return its public URL, or None (best-effort).
+def start_studio_tunnel(port: int, timeout: float = _READY_TIMEOUT) -> Optional[str]:
+    """Start a quick tunnel and return its public URL once it is actually
+    serving, or None (best-effort).
 
-    On any failure (no binary, no URL within timeout, early crash) the tunnel is
-    stopped and None is returned, so the caller prints a hint and continues.
+    Waits for cloudflared to both mint the URL and register an edge connection
+    before returning, so the caller never advertises a URL that yields Cloudflare
+    error 1033 (HTTP 530). If a URL is minted but no connection registers within
+    the window (e.g. quic is blocked on this network), retries once forcing the
+    http2 protocol. On any failure the tunnel is stopped and None is returned.
     """
-    global _active_tunnel
+    global _active_tunnel, _shutdown_requested
     binary = ensure_cloudflared()
     if not binary:
         return None
-    tunnel = CloudflareTunnel(port, binary)
-    # Register before start/wait so a shutdown during the URL wait can stop it.
     with _active_lock:
-        prior, _active_tunnel = _active_tunnel, tunnel
-    if prior is not None:
-        prior.stop()
-    try:
-        tunnel.start()
-        url = tunnel.wait_for_url(timeout)
-    except Exception:
-        url = None
-    if url:
-        return url
-    # No URL (or crash): drop it unless a concurrent shutdown already replaced it.
-    with _active_lock:
-        if _active_tunnel is tunnel:
-            _active_tunnel = None
-    tunnel.stop()
+        _shutdown_requested = False  # fresh session
+    # Default protocol first (quic, with cloudflared's own http2 fallback); if a
+    # URL appears but no connection registers, quic is likely blocked -> retry
+    # once forcing http2.
+    for protocol in (None, "http2"):
+        # Create + register under the lock, and bail if a stop already landed
+        # (e.g. between this and the previous attempt) so we never start a tunnel
+        # after shutdown has run.
+        with _active_lock:
+            if _shutdown_requested:
+                _active_tunnel = None
+                return None
+            tunnel = CloudflareTunnel(port, binary, protocol = protocol)
+            prior, _active_tunnel = _active_tunnel, tunnel
+        if prior is not None:
+            prior.stop()
+        try:
+            tunnel.start()
+            url = tunnel.wait_for_ready(timeout)
+        except Exception:
+            url = None
+        if url:
+            return url
+        saw_url = tunnel.url is not None
+        # Not ready: drop it, but only if we are still the active tunnel.
+        with _active_lock:
+            was_active = _active_tunnel is tunnel
+            if was_active:
+                _active_tunnel = None
+        tunnel.stop()
+        # A concurrent shutdown or start took over while we waited; retrying would
+        # spawn a tunnel nobody owns (orphaned after shutdown), so bail instead.
+        if not was_active:
+            return None
+        # No URL at all is an API/network failure, not a protocol one; forcing
+        # http2 will not help, so do not burn another window on it.
+        if not saw_url:
+            return None
     return None
 
 
 def stop_studio_tunnel() -> None:
     """Terminate the active tunnel, if any. Idempotent."""
-    global _active_tunnel
+    global _active_tunnel, _shutdown_requested
     with _active_lock:
+        # Latch so an in-flight start_studio_tunnel won't start a fresh tunnel
+        # (e.g. its http2 retry) after we have already torn down.
+        _shutdown_requested = True
         tunnel, _active_tunnel = _active_tunnel, None
     if tunnel is not None:
         tunnel.stop()

studio/backend/run.py

@@ -1032,9 +1032,13 @@ def run_server(
     _cloudflare_enabled = cloudflare and host == "0.0.0.0" and not api_only and not _IS_COLAB
     if _cloudflare_enabled:
         try:  # best-effort: any failure must not block startup
-            from cloudflare_tunnel import start_studio_tunnel
+            from cloudflare_tunnel import start_studio_tunnel, stop_studio_tunnel
+
             _cloudflare_url = start_studio_tunnel(port)
             app.state.cloudflare_url = _cloudflare_url
+            # Backstop: tear the tunnel down even on an abnormal exit that bypasses
+            # _graceful_shutdown (e.g. an exception after startup -> sys.exit). Idempotent.
+            atexit.register(stop_studio_tunnel)
         except Exception as e:
             logger.debug("Cloudflare tunnel skipped: %s", e)
 

studio/backend/tests/test_cloudflare_tunnel.py

@@ -52,6 +52,26 @@ def test_url_regex_no_match_on_unrelated():
     assert ct._URL_RE.search("INF connecting to https://api.cloudflare.com/v4") is None
 
 
+def test_url_regex_ignores_api_endpoint():
+    # cloudflared's failure line names its own API host; it must never be taken
+    # as the tunnel URL (it returns a 404 and is not a quick tunnel).
+    line = (
+        'failed to request quick Tunnel: Post "https://api.trycloudflare.com/tunnel": '
+        "context deadline exceeded"
+    )
+    assert ct._URL_RE.search(line) is None
+
+
+def test_url_regex_skips_api_host_but_matches_real_url():
+    blob = (
+        'ERR failed to request quick Tunnel: Post "https://api.trycloudflare.com/tunnel"\n'
+        "INF |  https://brave-mountain-river-clouds.trycloudflare.com  |\n"
+    )
+    m = ct._URL_RE.search(blob)
+    assert m is not None
+    assert m.group(0) == "https://brave-mountain-river-clouds.trycloudflare.com"
+
+
 # ── asset mapping ────────────────────────────────────────────────────
 
 
@@ -295,9 +315,88 @@ def test_stop_terminates_process():
     t.stop()
 
 
-def test_wait_for_url_times_out_without_blocking():
+def test_start_after_stop_does_not_spawn(monkeypatch):
+    # If stop() lands before start() (a concurrent shutdown in the caller's
+    # register->start window), start() must NOT spawn a cloudflared process --
+    # nobody would own it and it would be orphaned.
     t = ct.CloudflareTunnel(8080, "/bin/cloudflared")
-    assert t.wait_for_url(timeout = 0.05) is None
+    spawned = []
+
+    class _FakeProc:
+        stdout = None
+
+        def poll(self):
+            return 0
+
+    monkeypatch.setattr(ct.subprocess, "Popen", lambda *a, **k: (spawned.append(a), _FakeProc())[1])
+    t.stop()  # proc is None -> no-op terminate, but marks the tunnel stopped
+    t.start()  # must short-circuit before Popen
+    assert spawned == []
+    assert t._proc is None
+
+
+def test_wait_for_ready_times_out_without_blocking():
+    t = ct.CloudflareTunnel(8080, "/bin/cloudflared")
+    assert t.wait_for_ready(timeout = 0.05) is None
+
+
+def _fake_proc(text):
+    return types.SimpleNamespace(stdout = io.StringIO(text))
+
+
+def test_reader_captures_url_and_registration():
+    t = ct.CloudflareTunnel(8080, "/bin/cloudflared")
+    t._reader(
+        _fake_proc(
+            "INF Requesting new quick Tunnel on trycloudflare.com...\n"
+            "INF |  https://words-here-abc.trycloudflare.com  |\n"
+            "INF Registered tunnel connection connIndex=0 protocol=http2\n"
+        )
+    )
+    assert t.url == "https://words-here-abc.trycloudflare.com"
+    assert t.ready is True
+    assert t.wait_for_ready(0) == t.url
+    assert t.error is None  # a fully-registered tunnel records no error
+
+
+def test_reader_url_without_registration_is_not_ready():
+    # A URL but no "Registered tunnel connection" (e.g. quic control stream
+    # fails) must not be advertised -- it returns Cloudflare error 1033.
+    t = ct.CloudflareTunnel(8080, "/bin/cloudflared")
+    t._reader(
+        _fake_proc(
+            "INF |  https://words-here-abc.trycloudflare.com  |\n"
+            'ERR failed to serve tunnel connection error="control stream failure"\n'
+        )
+    )
+    assert t.url == "https://words-here-abc.trycloudflare.com"
+    assert t.ready is False
+    assert t.wait_for_ready(0) is None
+    assert t.error == "cloudflared exited before the tunnel connection registered"
+
+
+def test_reader_handles_none_stdout():
+    # Popen.stdout can be None; _reader must not crash and must leave the tunnel
+    # un-ready so wait_for_ready returns None.
+    t = ct.CloudflareTunnel(8080, "/bin/cloudflared")
+    t._reader(types.SimpleNamespace(stdout = None))
+    assert t.url is None
+    assert t.ready is False
+    assert t.wait_for_ready(0) is None
+    assert t.error == "cloudflared exited before emitting a tunnel URL"
+
+
+def test_reader_ignores_api_endpoint_failure_line():
+    t = ct.CloudflareTunnel(8080, "/bin/cloudflared")
+    t._reader(
+        _fake_proc(
+            "ERR failed to request quick Tunnel: Post "
+            '"https://api.trycloudflare.com/tunnel": context deadline exceeded\n'
+        )
+    )
+    assert t.url is None
+    assert t.wait_for_ready(0) is None
+    assert t.error == "cloudflared exited before emitting a tunnel URL"
 
 
 def test_start_studio_tunnel_no_binary(monkeypatch):
@@ -306,18 +405,23 @@ def test_start_studio_tunnel_no_binary(monkeypatch):
 
 
 def test_start_studio_tunnel_registers_before_wait(monkeypatch):
-    # The tunnel must be visible to stop_studio_tunnel() during the URL wait,
-    # else a shutdown in that window orphans cloudflared.
+    # The tunnel must be visible to stop_studio_tunnel() during the readiness
+    # wait, else a shutdown in that window orphans cloudflared.
     seen = {}
 
     class _Stub:
-        def __init__(self, port, binary):
+        def __init__(
+            self,
+            port,
+            binary,
+            protocol = None,
+        ):
             self.url = None
 
         def start(self):
             pass
 
-        def wait_for_url(self, timeout):
+        def wait_for_ready(self, timeout):
             seen["active_during_wait"] = ct._active_tunnel is self
             self.url = "https://x.trycloudflare.com"
             return self.url
@@ -338,13 +442,18 @@ def test_start_studio_tunnel_clears_and_stops_on_no_url(monkeypatch):
     seen = {}
 
     class _Stub:
-        def __init__(self, port, binary):
+        def __init__(
+            self,
+            port,
+            binary,
+            protocol = None,
+        ):
             self.url = None
 
         def start(self):
             pass
 
-        def wait_for_url(self, timeout):
+        def wait_for_ready(self, timeout):
             return None
 
         def stop(self):
@@ -359,13 +468,18 @@ def test_start_studio_tunnel_clears_and_stops_on_no_url(monkeypatch):
 
 def test_start_studio_tunnel_returns_url(monkeypatch):
     class _StubTunnel:
-        def __init__(self, port, binary):
+        def __init__(
+            self,
+            port,
+            binary,
+            protocol = None,
+        ):
             self.url = None
 
         def start(self):
             self.url = "https://stub-xyz.trycloudflare.com"
 
-        def wait_for_url(self, timeout):
+        def wait_for_ready(self, timeout):
             return self.url
 
         def stop(self):
@@ -379,6 +493,169 @@ def test_start_studio_tunnel_returns_url(monkeypatch):
         ct.stop_studio_tunnel()
 
 
+def test_start_studio_tunnel_falls_back_to_http2(monkeypatch):
+    # First attempt mints a URL but never registers (quic blocked); the http2
+    # retry registers and wins.
+    attempts = []
+
+    class _Stub:
+        def __init__(
+            self,
+            port,
+            binary,
+            protocol = None,
+        ):
+            self.protocol = protocol
+            self.url = None
+            attempts.append(protocol)
+
+        def start(self):
+            self.url = "https://words.trycloudflare.com"  # URL always minted
+
+        def wait_for_ready(self, timeout):
+            return self.url if self.protocol == "http2" else None
+
+        def stop(self):
+            pass
+
+    monkeypatch.setattr(ct, "ensure_cloudflared", lambda: "/bin/cloudflared")
+    monkeypatch.setattr(ct, "CloudflareTunnel", _Stub)
+    try:
+        assert ct.start_studio_tunnel(8080) == "https://words.trycloudflare.com"
+        assert attempts == [None, "http2"]  # default first, then forced http2
+    finally:
+        ct.stop_studio_tunnel()
+
+
+def test_start_studio_tunnel_no_retry_when_shutdown_between_attempts(monkeypatch):
+    # A stop() landing in the gap AFTER the failed first attempt is cleaned up but
+    # BEFORE the http2 retry registers must abort the loop -- not start a second
+    # tunnel that nobody will ever stop (Codex review). Simulated by having the
+    # first attempt's stop() (called during cleanup) trigger the shutdown.
+    attempts = []
+
+    class _Stub:
+        def __init__(
+            self,
+            port,
+            binary,
+            protocol = None,
+        ):
+            self.url = None
+            attempts.append(protocol)
+
+        def start(self):
+            self.url = "https://words.trycloudflare.com"  # URL minted, never ready
+
+        def wait_for_ready(self, timeout):
+            return None
+
+        def stop(self):
+            ct.stop_studio_tunnel()  # a concurrent shutdown lands in the gap
+
+    monkeypatch.setattr(ct, "ensure_cloudflared", lambda: "/bin/cloudflared")
+    monkeypatch.setattr(ct, "CloudflareTunnel", _Stub)
+    assert ct.start_studio_tunnel(8080) is None
+    assert attempts == [None]  # http2 retry aborted after shutdown
+    assert ct._active_tunnel is None
+
+
+def test_start_studio_tunnel_no_http2_retry_when_no_url(monkeypatch):
+    # No URL at all is an API/network failure; the http2 fallback would not help,
+    # so it must be skipped (don't burn a second timeout window).
+    attempts = []
+
+    class _Stub:
+        def __init__(
+            self,
+            port,
+            binary,
+            protocol = None,
+        ):
+            self.url = None
+            attempts.append(protocol)
+
+        def start(self):
+            pass  # never mints a URL
+
+        def wait_for_ready(self, timeout):
+            return None
+
+        def stop(self):
+            pass
+
+    monkeypatch.setattr(ct, "ensure_cloudflared", lambda: "/bin/cloudflared")
+    monkeypatch.setattr(ct, "CloudflareTunnel", _Stub)
+    assert ct.start_studio_tunnel(8080) is None
+    assert attempts == [None]
+
+
+def test_start_studio_tunnel_both_protocols_fail_registration(monkeypatch):
+    # Both quic and http2 mint a URL but neither registers -> both attempts are
+    # exhausted and None is returned (no dead URL advertised).
+    attempts = []
+
+    class _Stub:
+        def __init__(
+            self,
+            port,
+            binary,
+            protocol = None,
+        ):
+            self.url = None
+            attempts.append(protocol)
+
+        def start(self):
+            self.url = "https://words.trycloudflare.com"  # URL minted, never ready
+
+        def wait_for_ready(self, timeout):
+            return None
+
+        def stop(self):
+            pass
+
+    monkeypatch.setattr(ct, "ensure_cloudflared", lambda: "/bin/cloudflared")
+    monkeypatch.setattr(ct, "CloudflareTunnel", _Stub)
+    assert ct.start_studio_tunnel(8080) is None
+    assert attempts == [None, "http2"]
+    assert ct._active_tunnel is None
+
+
+def test_start_studio_tunnel_aborts_retry_on_concurrent_shutdown(monkeypatch):
+    # If a concurrent stop_studio_tunnel() clears _active_tunnel while we wait,
+    # the retry loop must NOT start a second (http2) tunnel: shutdown is already
+    # done, so nothing would ever stop it and it would be orphaned.
+    attempts = []
+
+    class _Stub:
+        def __init__(
+            self,
+            port,
+            binary,
+            protocol = None,
+        ):
+            self.url = None
+            attempts.append(protocol)
+
+        def start(self):
+            self.url = "https://words.trycloudflare.com"  # URL minted (saw_url True)
+
+        def wait_for_ready(self, timeout):
+            # Simulate stop_studio_tunnel() landing during the wait.
+            with ct._active_lock:
+                ct._active_tunnel = None
+            return None  # never registered
+
+        def stop(self):
+            pass
+
+    monkeypatch.setattr(ct, "ensure_cloudflared", lambda: "/bin/cloudflared")
+    monkeypatch.setattr(ct, "CloudflareTunnel", _Stub)
+    assert ct.start_studio_tunnel(8080) is None
+    assert attempts == [None]  # no http2 retry -> no orphaned second tunnel
+    assert ct._active_tunnel is None
+
+
 # ── run.py source-level pins (AST / source, no heavy import) ─────────
 
 
@@ -419,6 +696,13 @@ def test_argparse_cloudflare_default_true():
     assert _argparse_default(_RUN_PY.read_text(), "--cloudflare") is True
 
 
+def test_run_server_registers_tunnel_atexit_backstop():
+    # An abnormal exit (exception after startup -> sys.exit) bypasses
+    # _graceful_shutdown; an atexit backstop must still stop the tunnel.
+    src = _RUN_PY.read_text()
+    assert "atexit.register(stop_studio_tunnel)" in src
+
+
 def test_run_server_gates_tunnel_on_wildcard():
     # Guard against accidentally widening the trigger beyond 0.0.0.0.
     source = _RUN_PY.read_text()