async fn run(
    data_dir: PathBuf,
    startup_admin_addr: SocketAddr,
    options: ServeOptions,
    startup_locks: rust_proxy::settings::StartupLocks,
    update_check: rust_proxy::update_check::Config,
) -> Result<()> {
    ensure!(
        startup_admin_addr.port() > 0 && options.proxy_listen.port() > 0,
        "listener ports must be nonzero"
    );
    ensure!(
        (1..=4096).contains(&options.max_connections) && (1..=4096).contains(&options.max_inflight),
        "limits must be 1..4096"
    );
    ensure!(
        options.upstream_timeout_secs > 0
            && options.body_idle_timeout_secs > 0
            && options.shutdown_timeout_secs > 0
            && options.websocket_idle_timeout_secs > 0,
        "timeouts must be positive"
    );
    ensure!(
        startup_admin_addr.port() != options.proxy_listen.port(),
        "admin and proxy ports must differ"
    );
    if let Some(tls) = options.tls_listen {
        ensure!(
            tls.port() > 0
                && tls.port() != options.proxy_listen.port()
                && tls.port() != startup_admin_addr.port(),
            "TLS listener port must be nonzero and differ from HTTP/admin"
        );
    }
    if options.demo {
        ensure!(
            options.demo_listen.ip().is_loopback() && options.demo_listen.port() > 0,
            "demo must bind to a nonzero loopback port"
        );
        ensure!(
            options.demo_listen.port() != startup_admin_addr.port()
                && options.demo_listen.port() != options.proxy_listen.port()
                && options
                    .tls_listen
                    .is_none_or(|tls| tls.port() != options.demo_listen.port()),
            "demo port must differ from proxy/admin"
        );
    }
    auth::prepare_directory(&data_dir)?;
    let _process_lock = auth::lock_directory(&data_dir)?;
    auth::prepare_database(&data_dir.join("config.sqlite3"))?;
    let startup_listeners = (options.proxy_listen, options.tls_listen);
    let initial_listeners =
        rust_proxy::settings::Listeners::from_startup(options.proxy_listen, options.tls_listen);
    let initial = if options.demo {
        let mut config = config::demo_config(options.demo_listen);
        config.listeners = initial_listeners;
        config.vhosts[0].frontend.listener_ids = config
            .listeners
            .endpoints
            .iter()
            .map(|listener| listener.id.clone())
            .collect();
        config
    } else {
        let mut config = Config {
            listeners: initial_listeners,
            ..Config::default()
        };
        // Only a NEW database receives this neutral landing page.
        config.pages.direct_ip = Some(Default::default());
        config.pages.unknown_host = Some(Default::default());
        config
    };
    let store = Store::open(&data_dir.join("config.sqlite3"), &initial).await?;
    let mut saved = store.load().await?;
    startup_locks.ensure_config(&mut saved.config);
    startup_locks
        .validate(&saved.config, startup_admin_addr)
        .context("stored configuration conflicts with a listener fixed by systemd/startup")?;
    let admin_addr = saved
        .config
        .administration
        .resolved_listen(startup_admin_addr);
    let admin_tls_enabled = saved.config.administration.tls;
    let remote_admin =
        validate_admin_binding(admin_addr, admin_tls_enabled, options.allow_remote_admin)?;
    ensure!(
        admin_addr.port() != options.proxy_listen.port()
            && options
                .tls_listen
                .is_none_or(|tls| tls.port() != admin_addr.port())
            && (!options.demo || options.demo_listen.port() != admin_addr.port()),
        "active administration port conflicts with another startup listener"
    );
    let resolved_policy = ValidationPolicy {
        proxy_addr: options.proxy_listen,
        admin_addr,
        tls_addr: options.tls_listen,
        cert_dir: None,
        data_dir: Some(data_dir.clone()),
    };
    saved.config.listeners.validate(&resolved_policy)?;
    // The certificate library is available even on an HTTP-only installation.
    // Initializing private files does not bind or expose a TLS listener.
    let directory = data_dir.join("certs");
    let initialize = directory.clone();
    tokio::task::spawn_blocking(move || rust_proxy::tls::initialize_default(&initialize)).await??;
    rust_proxy::secret_store::initialize(&data_dir)?;
    let migrated =
        rust_proxy::secret_store::migrate_acme_credentials(&store, &data_dir, &directory).await?;
    if migrated > 0 {
        tracing::info!(
            migrated,
            "legacy ACME credentials migrated to encrypted Secret Store"
        );
    }
    let cert_dir = Some(directory.clone());
    let policy = ValidationPolicy {
        proxy_addr: options.proxy_listen,
        admin_addr,
        tls_addr: options.tls_listen,
        cert_dir,
        data_dir: Some(data_dir.clone()),
    };
    let validation_policy = policy.clone();
    let snapshot = tokio::task::spawn_blocking(move || {
        config::compile(saved.config, saved.revision, &validation_policy)
    })
    .await?
    .context("stored configuration is invalid; no automatic reset was performed")?;
    rust_proxy::logging::configure_error_log(snapshot.envelope.config.error_log.clone());
    let admin_tls_config = {
        let tls = snapshot.envelope.config.tls.clone();
        let administration = snapshot.envelope.config.administration.clone();
        let cert_dir = policy.cert_dir.clone();
        tokio::task::spawn_blocking(move || {
            rust_proxy::tls::compile_admin(&tls, &administration, cert_dir.as_deref())
        })
        .await??
    };
    let setup_required = store.admin_record().await?.is_none();
    let authentication = auth::Auth::new(store.clone(), data_dir.clone()).await?;
    if !setup_required {
        authentication.remove_bootstrap_files().await;
    }
    let renewal_observations = store.load_certificate_renewals().await?;
    let geoip = match rust_proxy::geoip::Database::load(&data_dir.join("geoip")) {
        Ok(database) => std::sync::Arc::new(database),
        Err(error) => {
            tracing::warn!(%error, "GeoIP country database unavailable; country allow lists fail closed");
            std::sync::Arc::new(rust_proxy::geoip::Database::empty())
        }
    };
    let state = App::new_with_startup_locks_geoip(
        store.clone(),
        snapshot,
        policy,
        authentication,
        options.max_inflight,
        Duration::from_secs(options.upstream_timeout_secs),
        Duration::from_secs(options.body_idle_timeout_secs),
        Duration::from_secs(options.websocket_idle_timeout_secs),
        startup_listeners,
        startup_locks,
        geoip,
    );
    state.acme.restore_renewals(renewal_observations);
    state.start_update_check(update_check);
    // Bind EVERY listener before spawning servers: a missing private IP or
    // occupied port fails startup atomically, never falling back to 0.0.0.0.
    let mut public_listeners = Vec::new();
    for binding in &state.active_listeners {
        if options.demo {
            ensure!(
                !rust_proxy::vhost::socket_overlap(binding.address, options.demo_listen),
                "vhost listener conflicts with demo backend"
            );
        }
        let listener = TcpListener::bind(binding.address).await.with_context(|| {
            format!(
                "cannot bind frontend {} (TLS={}); check local IP, port permissions and conflicts",
                binding.address, binding.tls
            )
        })?;
        public_listeners.push((*binding, listener));
    }
    let mut quic_listeners = Vec::new();
    for binding in &state.active_listeners {
        if binding.http3.is_some() {
            quic_listeners.push(rust_proxy::http3::bind(*binding, &state)?);
        }
    }
    let mut tcp_stream_listeners = Vec::new();
    let mut udp_stream_listeners = Vec::new();
    for stream in state.active_streams.iter().filter(|stream| stream.enabled) {
        let stream_metrics = state
            .stream_metrics
            .get(&stream.id)
            .cloned()
            .context("missing stream metrics registry entry")?;
        if options.demo && stream.protocol == rust_proxy::stream::Protocol::Tcp {
            ensure!(
                !rust_proxy::vhost::socket_overlap(stream.listen, options.demo_listen),
                "TCP stream listener conflicts with demo backend"
            );
        }
        match stream.protocol {
            rust_proxy::stream::Protocol::Tcp => {
                let listener = TcpListener::bind(stream.listen).await.with_context(|| {
                    format!("cannot bind TCP stream {} on {}", stream.id, stream.listen)
                })?;
                tcp_stream_listeners.push((stream.clone(), listener, stream_metrics));
            }
            rust_proxy::stream::Protocol::Udp => {
                let socket = UdpSocket::bind(stream.listen).await.with_context(|| {
                    format!("cannot bind UDP stream {} on {}", stream.id, stream.listen)
                })?;
                udp_stream_listeners.push((stream.clone(), socket, stream_metrics));
            }
        }
    }
    let metrics_token = if state.active_prometheus.enabled {
        state.active_prometheus.token()?
    } else {
        None
    };
    let metrics_listener = if state.active_prometheus.enabled {
        ensure!(
            !options.demo
                || !rust_proxy::vhost::socket_overlap(
                    state.active_prometheus.listen,
                    options.demo_listen
                ),
            "metrics exporter overlaps demo origin"
        );
        Some(
            TcpListener::bind(state.active_prometheus.listen)
                .await
                .context("cannot bind Prometheus exporter")?,
        )
    } else {
        None
    };
    let admin_listener = TcpListener::bind(admin_addr)
        .await
        .context("cannot bind admin listener")?;
    let demo_listener = if options.demo {
        Some(
            TcpListener::bind(options.demo_listen)
                .await
                .context("cannot bind demo backend")?,
        )
    } else {
        None
    };
    let shutdown = state.shutdown.clone();
    let grace = Duration::from_secs(options.shutdown_timeout_secs);
    let mut tasks = JoinSet::new();
    let connection_slots =
        std::sync::Arc::new(tokio::sync::Semaphore::new(options.max_connections));
    for (binding, listener) in public_listeners {
        tasks.spawn(rust_proxy::server::serve_frontend_with_http2(
            listener,
            rust_proxy::proxy::router(state.clone()),
            shutdown.clone(),
            connection_slots.clone(),
            grace,
            state.metrics.clone(),
            binding.tls.then(|| (state.clone(), binding.http2)),
            binding.h2c,
            Some(state.clone()),
            binding.http2_settings.unwrap_or_default(),
        ));
    }
    for listener in quic_listeners {
        tasks.spawn(rust_proxy::http3::serve(
            listener,
            rust_proxy::proxy::router(state.clone()),
            state.clone(),
            grace,
        ));
    }
    for (stream, listener, metrics) in tcp_stream_listeners {
        tasks.spawn(rust_proxy::stream::serve_tcp(
            stream,
            listener,
            shutdown.clone(),
            grace,
            metrics,
        ));
    }
    for (stream, socket, metrics) in udp_stream_listeners {
        tasks.spawn(rust_proxy::stream::serve_udp(
            stream,
            socket,
            shutdown.clone(),
            metrics,
        ));
    }
    if let Some(tls) = admin_tls_config {
        tasks.spawn(rust_proxy::server::serve_tls_with_plain_redirect(
            admin_listener,
            rust_proxy::admin::router(state.clone()),
            rust_proxy::admin::https_redirect_router(),
            shutdown.clone(),
            32,
            grace,
            tls,
        ));
    } else {
        tasks.spawn(rust_proxy::server::serve(
            admin_listener,
            rust_proxy::admin::router(state.clone()),
            shutdown.clone(),
            32,
            grace,
            None,
        ));
    }
    if let Some(listener) = metrics_listener {
        tasks.spawn(rust_proxy::server::serve(
            listener,
            rust_proxy::prometheus::router(state.clone(), metrics_token),
            shutdown.clone(),
            16,
            grace,
            None,
        ));
    }
    if let Some(listener) = demo_listener {
        tasks.spawn(rust_proxy::server::serve(
            listener,
            rust_proxy::demo::router(),
            shutdown.clone(),
            256,
            grace,
            None,
        ));
    }
    println!("\n  RustProxy {}\n", env!("CARGO_PKG_VERSION"));
    for binding in &state.active_listeners {
        println!(
            "  Frontend: {}://{}  ({})",
            if binding.tls { "https" } else { "http" },
            binding.address,
            if binding.http3.is_some() && binding.http2 {
                "HTTP/3 (UDP) + HTTP/2 + HTTP/1.1"
            } else if binding.http3.is_some() {
                "HTTP/3 (UDP) + HTTP/1.1"
            } else if binding.tls && binding.http2 {
                "HTTP/2 + HTTP/1.1"
            } else if binding.h2c {
                "h2c prior-knowledge + HTTP/1.1"
            } else {
                "HTTP/1.1"
            }
        );
    }
    for stream in state.active_streams.iter().filter(|stream| stream.enabled) {
        println!(
            "  Stream: {}://{} -> {}  ({})",
            match stream.protocol {
                rust_proxy::stream::Protocol::Tcp => "tcp",
                rust_proxy::stream::Protocol::Udp => "udp",
            },
            stream.listen,
            stream.upstream,
            stream.id
        );
    }
    println!(
        "  Panel:  {}://{admin_addr}",
        if admin_tls_enabled { "https" } else { "http" }
    );
    if remote_admin && !admin_tls_enabled {
        tracing::warn!(
            "REMOTE ADMIN ENABLED OVER PLAIN HTTP: credentials and session cookies can be exposed; restrict access with a firewall/private network and prefer a protected tunnel"
        );
    }
    if state.active_prometheus.enabled {
        println!(
            "  Metrics: http://{}{}",
            state.active_prometheus.listen, state.active_prometheus.path
        );
    }
    if setup_required {
        println!(
            "  Setup:  rust-proxy --data-dir {} admin setup-token",
            data_dir.display()
        );
        println!(
            "          Or: rust-proxy --data-dir {} admin init --username YOUR_NAME",
            data_dir.display()
        );
    } else {
        println!("  Sign in with your administrator account.");
    }
    println!("  Data:   {}\n  Stop:   Ctrl+C\n", data_dir.display());
    if state
        .active_listeners
        .iter()
        .any(|l| !l.address.ip().is_loopback())
    {
        tracing::warn!("public binding enabled: development release; configure host/route ACLs and trusted certificates before exposure");
    }
    let result = tokio::select! {
        result = shutdown_signal() => result,
        result = tasks.join_next() => match result {
            Some(Ok(Ok(()))) => Err(anyhow::anyhow!("listener stopped unexpectedly")),
            Some(Ok(Err(error))) => Err(error),
            Some(Err(error)) => Err(error.into()),
            None => Err(anyhow::anyhow!("all listeners stopped")),
        },
    };
    shutdown.cancel();
    while tasks.join_next().await.is_some() {}
    state.tunnels.close();
    let _ = tokio::time::timeout(grace, state.tunnels.wait()).await;
    state.drain_updates().await;
    let managed_update = data_dir.join(".managed-update").is_file();
    let cache_flush_budget = if managed_update {
        std::cmp::max(grace, Duration::from_secs(90))
    } else {
        grace
    };
    let flush_result = if managed_update {
        tokio::time::timeout(
            cache_flush_budget,
            state.cache.flush_hot_to_cold_for_update(),
        )
        .await
    } else {
        tokio::time::timeout(cache_flush_budget, state.cache.flush_hot_to_cold()).await
    };
    match flush_result {
        Ok(moved) if moved > 0 && managed_update => tracing::info!(
            moved,
            "managed update moved hot cache objects to cold storage"
        ),
        Ok(moved) if moved > 0 => tracing::info!(
            moved,
            "flushed hot cache objects to cold storage before shutdown"
        ),
        Err(_) if managed_update => {
            tracing::warn!("managed-update hot-to-cold cache handoff exceeded its shutdown budget")
        }
        Err(_) => tracing::warn!("cold-cache shutdown flush exceeded graceful shutdown timeout"),
        _ => {}
    }
    store.close().await;
    result
}