gw: retry+exponential backoff for port_attrs lazy fetch

Right after registration, the WireGuard handshake hasn't completed yet
and the agent's TCP port isn't reachable. The previous one-shot fetch
would fail and leave the cache empty, falling back to pp=false until
the next connection (which would itself eat one more failed fetch).

Move the timeout/retry policy into a dedicated config block so it can
be tuned per deployment:

  [core.proxy.port_attrs_fetch]
  timeout = "10s"          # per-attempt Info() RPC timeout
  max_retries = 5          # extra attempts after the initial try
  backoff_initial = "1s"   # doubles each retry up to backoff_max
  backoff_max = "30s"

Worst-case 1+2+4+8+16+30 ≈ 1 min covers a reasonable WG warmup window.

Bail out early when the instance is no longer in state (recycled while
queued) — the unknown-instance error chain is the signal.

Author: kvinwang

gateway/gateway.toml

@@ -61,6 +61,16 @@ max_connections_per_app = 2000
 # Whether to read PROXY protocol from inbound connections (e.g. from Cloudflare).
 inbound_pp_enabled = false
 
+[core.proxy.port_attrs_fetch]
+# Background lazy-fetch of port_attrs from legacy CVM agents.
+# Single Info() RPC timeout.
+timeout = "10s"
+# Retries cover the WireGuard / agent warmup window after registration.
+max_retries = 5
+# Exponential backoff between retries; doubles each attempt up to backoff_max.
+backoff_initial = "1s"
+backoff_max = "30s"
+
 [core.proxy.timeouts]
 # Timeout for establishing a connection to the target app.
 connect = "5s"
@@ -85,8 +95,6 @@ shutdown = "5s"
 total = "5h"
 # Timeout for proxy protocol header.
 pp_header = "5s"
-# Timeout for the background lazy fetch of port_attrs from a legacy CVM.
-port_attrs_fetch = "10s"
 
 [core.recycle]
 enabled = true

gateway/src/config.rs

@@ -124,6 +124,25 @@ pub struct ProxyConfig {
     /// (e.g. when behind a PP-aware load balancer like Cloudflare).
     #[serde(default)]
     pub inbound_pp_enabled: bool,
+    /// Background lazy-fetch behaviour for `port_attrs` (legacy CVMs).
+    pub port_attrs_fetch: PortAttrsFetchConfig,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct PortAttrsFetchConfig {
+    /// Timeout for a single `Info()` RPC attempt.
+    #[serde(with = "serde_duration")]
+    pub timeout: Duration,
+    /// Maximum number of attempts after the initial try (0 = no retry).
+    /// Retries cover the window where a freshly-registered CVM hasn't
+    /// finished its WireGuard handshake yet.
+    pub max_retries: u32,
+    /// Delay before the first retry; doubles on each subsequent retry,
+    /// capped at `backoff_max`.
+    #[serde(with = "serde_duration")]
+    pub backoff_initial: Duration,
+    #[serde(with = "serde_duration")]
+    pub backoff_max: Duration,
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -152,10 +171,6 @@ pub struct Timeouts {
     /// Timeout for reading the proxy protocol header from inbound connections.
     #[serde(with = "serde_duration")]
     pub pp_header: Duration,
-    /// Timeout for the background lazy fetch of `port_attrs` from a legacy CVM
-    /// agent's `Info()` RPC.
-    #[serde(with = "serde_duration")]
-    pub port_attrs_fetch: Duration,
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize)]

gateway/src/proxy/port_attrs.rs

@@ -45,13 +45,12 @@ pub(crate) fn should_send_pp(state: &Proxy, instance_id: &str, port: u16) -> boo
 /// Spawn the background lazy-fetch worker. Should be called once at startup.
 pub(crate) fn spawn_fetcher(state: Proxy, mut rx: UnboundedReceiver<String>) {
     let in_flight: Arc<Mutex<HashSet<String>>> = Default::default();
-    let timeout = state.config.proxy.timeouts.port_attrs_fetch;
     tokio::spawn(async move {
         while let Some(instance_id) = rx.recv().await {
-            // Dedupe: only spawn one fetch per instance at a time. After the
-            // fetch completes, the entry is removed so a later registration
-            // (with new compose_hash) can trigger a fresh fetch via the same
-            // path.
+            // Dedupe: only one fetch per instance at a time. The entry is
+            // removed once the retry loop terminates (success, exhausted,
+            // or unknown-instance), so a later registration with a new
+            // compose_hash can re-trigger via the same path.
             {
                 let mut in_flight = in_flight.lock().expect("port_attrs in_flight poisoned");
                 if !in_flight.insert(instance_id.clone()) {
@@ -62,15 +61,7 @@ pub(crate) fn spawn_fetcher(state: Proxy, mut rx: UnboundedReceiver<String>) {
             let in_flight = in_flight.clone();
             let id = instance_id.clone();
             tokio::spawn(async move {
-                match tokio::time::timeout(timeout, fetch_and_store(&state, &id)).await {
-                    Ok(Ok(())) => debug!("port_attrs cached for instance {id}"),
-                    Ok(Err(err)) => {
-                        warn!("port_attrs fetch for instance {id} failed: {err:#}")
-                    }
-                    Err(_) => {
-                        warn!("port_attrs fetch for instance {id} timed out after {timeout:?}")
-                    }
-                }
+                fetch_with_retry(&state, &id).await;
                 in_flight
                     .lock()
                     .expect("port_attrs in_flight poisoned")
@@ -80,6 +71,50 @@ pub(crate) fn spawn_fetcher(state: Proxy, mut rx: UnboundedReceiver<String>) {
     });
 }
 
+async fn fetch_with_retry(state: &Proxy, instance_id: &str) {
+    let cfg = &state.config.proxy.port_attrs_fetch;
+    let mut attempt = 0u32;
+    let mut backoff = cfg.backoff_initial;
+    loop {
+        match tokio::time::timeout(cfg.timeout, fetch_and_store(state, instance_id)).await {
+            Ok(Ok(())) => {
+                debug!("port_attrs cached for instance {instance_id} (attempt {attempt})");
+                return;
+            }
+            Ok(Err(err)) if is_unknown_instance(&err) => {
+                // Instance was recycled while the fetch was queued. No
+                // point retrying — the instance is gone.
+                debug!("port_attrs fetch for {instance_id}: instance no longer exists, giving up");
+                return;
+            }
+            Ok(Err(err)) => {
+                warn!("port_attrs fetch for {instance_id} failed (attempt {attempt}): {err:#}");
+            }
+            Err(_) => {
+                warn!(
+                    "port_attrs fetch for {instance_id} timed out after {:?} (attempt {attempt})",
+                    cfg.timeout
+                );
+            }
+        }
+        if attempt >= cfg.max_retries {
+            warn!(
+                "port_attrs fetch for {instance_id} giving up after {} attempts",
+                attempt + 1
+            );
+            return;
+        }
+        tokio::time::sleep(backoff).await;
+        attempt += 1;
+        backoff = (backoff * 2).min(cfg.backoff_max);
+    }
+}
+
+fn is_unknown_instance(err: &anyhow::Error) -> bool {
+    err.chain()
+        .any(|e| e.to_string().contains("unknown instance"))
+}
+
 async fn fetch_and_store(state: &Proxy, instance_id: &str) -> Result<()> {
     let (ip, agent_port) = {
         let guard = state.lock();