Function fetch_with_retry

gateway/src/proxy/port_policy.rs:167–209

(state: &Proxy, instance_id: &str)

}

async fn fetch_with_retry(state: &Proxy, instance_id: &str) {
    let cfg = &state.config.proxy.port_policy_fetch;
    let mut attempt = 0u32;
    let mut backoff = cfg.backoff_initial;
    loop {
        let result =
            match tokio::time::timeout(cfg.timeout, fetch_and_store(state, instance_id)).await {
                Ok(r) => r,
                // The Info() RPC took too long. Treat as transient — the CVM
                // may just be slow to come up.
                Err(_) => Err(FetchError::Transient(anyhow::anyhow!(
                    "Info() rpc timed out after {:?}",
                    cfg.timeout
                ))),
            };
        match result {
            Ok(()) => {
                debug!("port_policy cached for instance {instance_id} (attempt {attempt})");
                return;
            }
            Err(FetchError::Permanent(err)) => {
                // Either the instance was recycled while queued, or the
                // agent responded with data we can't parse. Retrying won't
                // change either; bail.
                debug!("port_policy fetch for {instance_id}: permanent failure: {err:#}");
                return;
            }
            Err(FetchError::Transient(err)) => {
                warn!("port_policy fetch for {instance_id} failed (attempt {attempt}): {err:#}");
            }
        }
        if attempt >= cfg.max_retries {
            warn!(
                "port_policy 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);
    }
}

async fn fetch_and_store(state: &Proxy, instance_id: &str) -> Result<(), FetchError> {
    let (ip, agent_port) = {