eggress_runtime/
supervisor.rs

1use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
2use std::sync::{Arc, Mutex};
3use std::time::{Duration, Instant};
4
5use arc_swap::ArcSwap;
6#[cfg(feature = "operations")]
7use eggress_admin::{AdminSnapshot, AdminSnapshotProvider, ListenerInfo};
8use eggress_core::listener::{TcpListener, TcpListenerConfig};
9use eggress_core::ProtocolId;
10use eggress_routing::health::HealthManager;
11use eggress_routing::upstream::UpstreamRuntime;
12use eggress_routing::{RouteService, SharedRoutingService};
13use tokio_util::sync::CancellationToken;
14use tokio_util::task::TaskTracker;
15use tracing::Instrument;
16
17use crate::error::RuntimeError;
18use crate::platform::{check_capability, PlatformCapability};
19use crate::snapshot::{compile_runtime_snapshot, CompiledRuntimeSnapshot};
20
21/// Per-listener connection slot for listener implementations that cannot use
22/// the core TCP listener wrapper (transparent and Unix sockets).
23struct ListenerConnectionSlot {
24    active: Arc<AtomicU64>,
25}
26
27impl ListenerConnectionSlot {
28    fn try_acquire(active: &Arc<AtomicU64>, limit: u64) -> Option<Self> {
29        active
30            .fetch_update(Ordering::AcqRel, Ordering::Relaxed, |current| {
31                (current < limit).then_some(current + 1)
32            })
33            .ok()
34            .map(|_| Self {
35                active: active.clone(),
36            })
37    }
38}
39
40impl Drop for ListenerConnectionSlot {
41    fn drop(&mut self) {
42        self.active.fetch_sub(1, Ordering::Release);
43    }
44}
45
46/// Result of a reload attempt.
47#[derive(Debug)]
48pub enum ReloadResult {
49    /// Reload was applied successfully.
50    Applied { generation: u64, upstreams: usize },
51    /// Reload was rejected due to unsupported changes.
52    Rejected { reason: String },
53    /// Reload failed due to a compile or build error.
54    Failed { error: String },
55}
56
57/// Adapter that exposes the runtime's compiled snapshot to the admin server.
58///
59/// Implements `AdminSnapshotProvider` so that admin handlers see live data:
60/// each request reads the current `ArcSwap<CompiledRuntimeSnapshot>` rather
61/// than a startup-captured copy. Reloads take effect on the next request.
62#[cfg(feature = "operations")]
63pub struct RuntimeAdminListenerInfos {
64    state: Arc<RuntimeState>,
65}
66
67#[cfg(feature = "operations")]
68impl AdminSnapshotProvider for RuntimeAdminListenerInfos {
69    fn snapshot(&self) -> AdminSnapshot {
70        let snap = self.state.snapshot.load();
71        let addrs = self
72            .state
73            .listener_addrs
74            .lock()
75            .unwrap_or_else(|e| e.into_inner());
76        let listeners: Vec<ListenerInfo> = snap
77            .listeners
78            .iter()
79            .enumerate()
80            .map(|(idx, lcfg)| {
81                let mode = if lcfg.transparent.as_ref().is_some_and(|t| t.enabled) {
82                    Some("transparent".to_string())
83                } else if lcfg.unix.is_some() {
84                    Some("unix".to_string())
85                } else {
86                    Some("standard".to_string())
87                };
88
89                let capability_status = if lcfg.transparent.as_ref().is_some_and(|t| t.enabled) {
90                    let cap = crate::platform::check_capability(
91                        crate::platform::PlatformCapability::LinuxOriginalDstIpv4,
92                    );
93                    Some(cap.to_string())
94                } else {
95                    None
96                };
97
98                let original_dst_support = if lcfg.transparent.as_ref().is_some_and(|t| t.enabled) {
99                    let cap = crate::platform::check_capability(
100                        crate::platform::PlatformCapability::LinuxOriginalDstIpv4,
101                    );
102                    Some(cap == crate::platform::CapabilityStatus::Available)
103                } else {
104                    None
105                };
106
107                let (unix_socket_path, unix_socket_unlink_existing) =
108                    if let Some(ref unix_cfg) = lcfg.unix {
109                        (
110                            Some(unix_cfg.path.display().to_string()),
111                            Some(unix_cfg.unlink_existing),
112                        )
113                    } else {
114                        (None, None)
115                    };
116
117                ListenerInfo {
118                    name: lcfg.name.clone(),
119                    bind: lcfg.bind.clone(),
120                    local_addr: addrs
121                        .get(idx)
122                        .and_then(|a| *a)
123                        .map(|a| a.to_string())
124                        .unwrap_or_default(),
125                    protocols: lcfg.protocols.iter().map(|p| p.to_string()).collect(),
126                    udp_enabled: lcfg.udp.as_ref().is_some_and(|u| u.enabled),
127                    mode,
128                    capability_status,
129                    original_dst_support,
130                    unix_socket_path,
131                    unix_socket_unlink_existing,
132                }
133            })
134            .collect();
135        AdminSnapshot {
136            generation: snap.generation,
137            router: snap.router.clone(),
138            pac: snap.admin.as_ref().and_then(|a| a.pac.clone()),
139            static_routes: snap
140                .admin
141                .as_ref()
142                .map(|a| a.static_content.clone())
143                .unwrap_or_default(),
144            listeners,
145        }
146    }
147}
148
149/// What is and isn't reloaded on SIGHUP:
150///
151/// **Reloaded (hot-swap, no downtime):**
152/// - Upstream chains and health config (with Arc reuse for unchanged upstreams)
153/// - Upstream groups, schedulers, and fallback policies
154/// - Routing rules and default action
155/// - Listener configuration metadata (name, bind, protocols, auth)
156/// - Admin PAC and static content configuration
157///
158/// **NOT reloaded (requires full restart):**
159/// - Listener socket bindings (bound before readiness)
160/// - Process-level settings (log format, log level, shutdown grace)
161/// - Timeout configuration
162/// - Admin bind address
163///
164/// **UDP-specific reload semantics:**
165/// - UDP limits apply to new associations only; existing associations keep their limits.
166/// - UDP bind changes are restart-required.
167/// - UDP advertise address changes are restart-required if socket bind changes.
168/// - Route changes apply immediately to future UDP packets.
169///
170/// Classify whether a reload is supported given old and new listener
171/// configs. Returns `Ok(())` if the reload is safe, or `Err(reason)`
172/// if it should be rejected.
173fn classify_listeners(
174    old_listeners: &[eggress_config::compile::ListenerConfig],
175    new_listeners: &[eggress_config::compile::ListenerConfig],
176) -> Result<(), String> {
177    if old_listeners.len() != new_listeners.len() {
178        return Err(format!(
179            "listener count changed ({} -> {}); restart required",
180            old_listeners.len(),
181            new_listeners.len()
182        ));
183    }
184
185    for (old, new) in old_listeners.iter().zip(new_listeners.iter()) {
186        if old.name != new.name {
187            return Err(format!(
188                "listener name changed ('{}' -> '{}'); restart required",
189                old.name, new.name
190            ));
191        }
192        if old.bind != new.bind {
193            return Err(format!(
194                "listener bind address changed for '{}': '{}' -> '{}'; restart required",
195                old.name, old.bind, new.bind
196            ));
197        }
198        match (&old.udp, &new.udp) {
199            (Some(old_udp), Some(new_udp)) => {
200                if old_udp.bind != new_udp.bind {
201                    return Err(format!(
202                        "UDP bind address changed for '{}': '{}' -> '{}'; restart required",
203                        old.name, old_udp.bind, new_udp.bind
204                    ));
205                }
206            }
207            (None, Some(new_udp)) => {
208                if !new_udp.bind.ip().is_unspecified() {
209                    return Err(format!(
210                        "UDP bind address added for '{}': '{}'; restart required",
211                        new.name, new_udp.bind
212                    ));
213                }
214            }
215            (Some(_old_udp), None) => {}
216            (None, None) => {}
217        }
218
219        match (&old.transparent, &new.transparent) {
220            (Some(old_t), Some(new_t)) => {
221                if old_t.enabled != new_t.enabled {
222                    return Err(format!(
223                        "transparent config changed for '{}': enabled {} -> {}; restart required",
224                        old.name, old_t.enabled, new_t.enabled
225                    ));
226                }
227            }
228            (None, Some(new_t)) => {
229                if new_t.enabled {
230                    return Err(format!(
231                        "transparent proxy enabled for '{}'; restart required",
232                        new.name
233                    ));
234                }
235            }
236            (Some(_old_t), None) => {}
237            (None, None) => {}
238        }
239
240        match (&old.unix, &new.unix) {
241            (Some(old_u), Some(new_u)) => {
242                if old_u.path != new_u.path {
243                    return Err(format!(
244                        "unix socket path changed for '{}': '{}' -> '{}'; restart required",
245                        old.name,
246                        old_u.path.display(),
247                        new_u.path.display()
248                    ));
249                }
250            }
251            (None, Some(_new_u)) => {
252                return Err(format!(
253                    "unix socket added for '{}'; restart required",
254                    new.name
255                ));
256            }
257            (Some(_old_u), None) => {
258                return Err(format!(
259                    "unix socket removed for '{}'; restart required",
260                    old.name
261                ));
262            }
263            (None, None) => {}
264        }
265    }
266
267    Ok(())
268}
269
270struct PreparedListener {
271    name: String,
272    #[allow(dead_code)] // used only with `operations` feature
273    bind: String,
274    protocols: Vec<ProtocolId>,
275    listener: TcpListener,
276    local_addr: std::net::SocketAddr,
277    auth: eggress_server::accept::InboundAuthentication,
278    handshake_timeout: Duration,
279    udp: Option<eggress_config::compile::CompiledListenerUdpConfig>,
280    tls: Option<eggress_config::compile::CompiledListenerTlsConfig>,
281    shadowsocks: Option<eggress_config::model::ShadowsocksListenerConfig>,
282    trojan: Option<eggress_config::model::ListenerTrojanConfig>,
283    fixed_target: Option<eggress_core::TargetAddr>,
284    local_bind: Option<String>,
285}
286
287#[cfg(feature = "quic")]
288struct PreparedQuicListener {
289    name: String,
290    protocols: Vec<ProtocolId>,
291    listener: Arc<eggress_transport_quic::QuicListener>,
292    local_addr: std::net::SocketAddr,
293    auth: eggress_server::accept::InboundAuthentication,
294    handshake_timeout: Duration,
295    connection_limit: u64,
296}
297
298#[cfg(feature = "extended")]
299type PreparedShadowsocksUdpRelay = (
300    Arc<tokio::net::UdpSocket>,
301    eggress_udp::standalone_shadowsocks::ShadowsocksStandaloneUdpConfig,
302);
303
304#[cfg(feature = "extended")]
305async fn prepare_shadowsocks_udp_relay(
306    prepared_listener: &PreparedListener,
307    udp_cfg: &eggress_config::compile::CompiledListenerUdpConfig,
308    routing: Arc<dyn RouteService>,
309    state: &RuntimeState,
310) -> Result<PreparedShadowsocksUdpRelay, RuntimeError> {
311    let ss = prepared_listener.shadowsocks.as_ref().ok_or_else(|| {
312        RuntimeError::Other(format!(
313            "listener '{}' shadowsocks_udp mode requires shadowsocks config",
314            prepared_listener.name
315        ))
316    })?;
317    let method =
318        eggress_protocol_shadowsocks::CipherMethod::parse_method(&ss.method).map_err(|e| {
319            RuntimeError::Other(format!(
320                "listener '{}' has invalid shadowsocks method '{}': {}",
321                prepared_listener.name, ss.method, e
322            ))
323        })?;
324    let socket = Arc::new(
325        tokio::net::UdpSocket::bind(udp_cfg.bind)
326            .await
327            .map_err(|e| RuntimeError::ListenerBind {
328                addr: udp_cfg.bind.to_string(),
329                source: e,
330            })?,
331    );
332    let local_addr = socket
333        .local_addr()
334        .map_err(|e| RuntimeError::ListenerBind {
335            addr: udp_cfg.bind.to_string(),
336            source: e,
337        })?;
338    tracing::info!(
339        "shadowsocks UDP relay listening on {local_addr} ({})",
340        prepared_listener.name
341    );
342
343    let relay_config = eggress_udp::standalone_shadowsocks::ShadowsocksStandaloneUdpConfig {
344        routing,
345        udp_metrics: state.udp_metrics.clone(),
346        shadowsocks_metrics: Some(state.shadowsocks_metrics.clone()),
347        limits: eggress_udp::limits::UdpLimits::from_listener_config(
348            udp_cfg.max_associations_global,
349            udp_cfg.max_associations,
350            udp_cfg.max_targets_per_association,
351            udp_cfg.max_datagram_size,
352            udp_cfg.idle_timeout,
353            udp_cfg.client_pin,
354            udp_cfg.target_idle_timeout,
355        ),
356        listener: prepared_listener.name.clone(),
357        generation: state.snapshot.load().generation,
358        method,
359        password: ss.password.clone(),
360        allow_private_egress: udp_cfg.allow_private_egress,
361    };
362
363    Ok((socket, relay_config))
364}
365
366/// Compute the advertised IP for the SOCKS5 UDP ASSOCIATE reply.
367///
368/// Derivation rules:
369/// 1. If `advertise` is configured, use it.
370/// 2. Else if UDP bind IP is not unspecified, use UDP bind IP.
371/// 3. Else if TCP peer is loopback, use loopback matching address family.
372/// 4. Else return a config error requiring explicit `advertise`.
373fn compute_advertise_ip(
374    configured_advertise: Option<std::net::IpAddr>,
375    udp_bind_ip: std::net::IpAddr,
376    tcp_peer: std::net::SocketAddr,
377) -> Result<std::net::IpAddr, eggress_udp::error::UdpError> {
378    if let Some(ip) = configured_advertise {
379        return Ok(ip);
380    }
381
382    if !udp_bind_ip.is_unspecified() {
383        return Ok(udp_bind_ip);
384    }
385
386    if tcp_peer.ip().is_loopback() {
387        match tcp_peer {
388            std::net::SocketAddr::V4(_) => {
389                return Ok(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST));
390            }
391            std::net::SocketAddr::V6(_) => {
392                return Ok(std::net::IpAddr::V6(std::net::Ipv6Addr::LOCALHOST));
393            }
394        }
395    }
396
397    Err(eggress_udp::error::UdpError::Other(
398        "UDP relay requires explicit advertise address when bind is unspecified and client is not loopback".to_string()
399    ))
400}
401
402struct RuntimeUdpService {
403    _listener_name: String,
404    udp_config: eggress_config::compile::CompiledListenerUdpConfig,
405    registry: Arc<eggress_udp::registry::UdpAssociationRegistry>,
406    metrics: Arc<dyn eggress_server::SessionMetrics>,
407    udp_metrics: Arc<eggress_udp::metrics::UdpMetrics>,
408    routing: Arc<SharedRoutingService>,
409    udp_tasks: TaskTracker,
410}
411
412impl eggress_server::UdpService for RuntimeUdpService {
413    fn create_association(
414        &self,
415        listener: &str,
416        client_tcp_peer: std::net::SocketAddr,
417        identity: eggress_core::ClientIdentity,
418        generation: u64,
419    ) -> std::pin::Pin<
420        Box<
421            dyn std::future::Future<
422                    Output = Result<
423                        eggress_server::UdpAssociationHandle,
424                        eggress_udp::error::UdpError,
425                    >,
426                > + Send
427                + 'static,
428        >,
429    > {
430        let registry = self.registry.clone();
431        let metrics = self.metrics.clone();
432        let udp_metrics = self.udp_metrics.clone();
433        let routing = self.routing.clone();
434        let udp_tasks = self.udp_tasks.clone();
435        let udp_config = self.udp_config.clone();
436        let listener = listener.to_string();
437        Box::pin(async move {
438            let assoc = registry
439                .create_association(&listener, client_tcp_peer, identity, generation)
440                .await?;
441            metrics.record_udp_association_created();
442
443            let relay_socket =
444                std::sync::Arc::new(tokio::net::UdpSocket::bind(udp_config.bind).await?);
445            let local_addr = relay_socket.local_addr()?;
446
447            let advertised_ip =
448                compute_advertise_ip(udp_config.advertise, local_addr.ip(), client_tcp_peer)?;
449            let relay_addr = std::net::SocketAddr::new(advertised_ip, local_addr.port());
450
451            let relay_config = eggress_udp::relay::RelayConfig {
452                routing: routing as Arc<dyn RouteService>,
453                udp_metrics: udp_metrics.clone(),
454                limits: eggress_udp::limits::UdpLimits::from_listener_config(
455                    udp_config.max_associations_global,
456                    udp_config.max_associations,
457                    udp_config.max_targets_per_association,
458                    udp_config.max_datagram_size,
459                    udp_config.idle_timeout,
460                    udp_config.client_pin,
461                    udp_config.target_idle_timeout,
462                ),
463                listener: listener.clone(),
464                generation,
465                identity: assoc.meta.identity.clone(),
466                client_tcp_peer,
467                registry: registry.clone(),
468            };
469
470            let relay_assoc = assoc.clone();
471            let relay_cancel = assoc.cancel.clone();
472            let assoc_id = assoc.id;
473            udp_tasks.spawn(async move {
474                let result = eggress_udp::relay::udp_relay_loop(
475                    relay_socket,
476                    relay_assoc,
477                    relay_config,
478                    relay_cancel,
479                )
480                .await;
481                if let Err(error) = result {
482                    tracing::debug!(
483                        %error,
484                        association_id = ?assoc_id,
485                        "UDP relay ended with error"
486                    );
487                }
488            });
489
490            Ok(eggress_server::UdpAssociationHandle {
491                id: assoc.id,
492                relay_addr,
493                cancel: assoc.cancel.clone(),
494            })
495        })
496    }
497
498    fn is_enabled(&self) -> bool {
499        self.udp_config.enabled
500    }
501
502    fn active_count(
503        &self,
504    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = usize> + Send + 'static>> {
505        let registry = self.registry.clone();
506        Box::pin(async move { registry.active_count().await })
507    }
508}
509
510pub struct RuntimeState {
511    pub snapshot: Arc<ArcSwap<CompiledRuntimeSnapshot>>,
512    pub routing: Arc<SharedRoutingService>,
513    pub metrics: Arc<dyn eggress_server::SessionMetrics>,
514    pub readiness: Arc<AtomicBool>,
515    pub start_time: Instant,
516    pub active_connections: Arc<AtomicU64>,
517    pub connection_counter: Arc<AtomicU64>,
518    pub admin_local_addr: Arc<Mutex<Option<std::net::SocketAddr>>>,
519    pub listener_addrs: Arc<Mutex<Vec<Option<std::net::SocketAddr>>>>,
520    pub udp_registry: Arc<eggress_udp::registry::UdpAssociationRegistry>,
521    pub udp_metrics: Arc<eggress_udp::metrics::UdpMetrics>,
522    #[cfg(feature = "extended")]
523    pub shadowsocks_metrics: Arc<eggress_protocol_shadowsocks::ShadowsocksMetrics>,
524    pub udp_tasks: TaskTracker,
525    pub transparent_accepted_total: Arc<AtomicU64>,
526    pub transparent_original_dst_failed_total: Arc<AtomicU64>,
527    #[cfg(feature = "reverse")]
528    pub reverse_registry: Arc<eggress_admin::ReverseRegistry>,
529    #[cfg(feature = "reverse")]
530    pub reverse_metrics: Arc<eggress_protocol_reverse::metrics::ReverseMetrics>,
531}
532
533impl RuntimeState {
534    pub fn generation(&self) -> u64 {
535        self.snapshot.load().generation
536    }
537}
538
539#[allow(dead_code)]
540pub struct ServiceSupervisor {
541    config_path: Option<String>,
542    state: Arc<RuntimeState>,
543    metrics_registry: Arc<eggress_metrics::MetricsRegistry>,
544    cancel: CancellationToken,
545    listener_cancel: CancellationToken,
546    connection_cancel: CancellationToken,
547    health_cancel: CancellationToken,
548    admin_cancel: CancellationToken,
549    health: Option<HealthManager>,
550    tasks: TaskTracker,
551    connection_tasks: TaskTracker,
552    admin_tasks: TaskTracker,
553    shutdown_grace: Duration,
554    rt_config: eggress_config::compile::RuntimeConfig,
555    tls_client_config: Option<std::sync::Arc<rustls::ClientConfig>>,
556    #[cfg(feature = "ssh")]
557    ssh_sessions: Arc<eggress_transport_ssh::SshSessionCache>,
558    compatibility_options: CompatibilityOptions,
559}
560
561/// Explicit options used only by the pproxy compatibility executable.
562/// Native configuration startup always uses the default value.
563#[derive(Debug, Clone, Default)]
564pub struct CompatibilityOptions {
565    pub auth_timeout: Option<Duration>,
566    pub system_proxy: bool,
567    /// Compatibility-only debug propagation. Native Eggress never enables it.
568    pub debug: bool,
569    /// Compatibility verbosity count, used for human-readable session and
570    /// traffic events backed by the normal session reports.
571    pub verbose_level: u8,
572}
573
574impl ServiceSupervisor {
575    pub fn start(config_path: &str) -> Result<Self, RuntimeError> {
576        let (rt_config, warnings) = eggress_config::load_and_validate_with_warnings(config_path)
577            .map_err(|e| RuntimeError::Config(e.to_string()))?;
578
579        for warning in &warnings {
580            tracing::warn!("config security warning: {warning}");
581        }
582
583        Self::init_with_config(
584            rt_config,
585            Some(config_path.to_string()),
586            CompatibilityOptions::default(),
587        )
588    }
589
590    /// Start from an already-validated [`RuntimeConfig`] without reading a file.
591    ///
592    /// When `config_path` is provided (file-backed startup), SIGHUP reload
593    /// is enabled. When `None` (in-memory/compatibility startup), SIGHUP
594    /// reload is disabled because there is no stable user-authored config
595    /// file to reload from.
596    pub fn start_from_config(
597        rt_config: eggress_config::compile::RuntimeConfig,
598        config_path: Option<String>,
599    ) -> Result<Self, RuntimeError> {
600        Self::init_with_config(rt_config, config_path, CompatibilityOptions::default())
601    }
602
603    pub fn start_from_config_with_options(
604        rt_config: eggress_config::compile::RuntimeConfig,
605        config_path: Option<String>,
606        compatibility_options: CompatibilityOptions,
607    ) -> Result<Self, RuntimeError> {
608        Self::init_with_config(rt_config, config_path, compatibility_options)
609    }
610
611    fn init_with_config(
612        rt_config: eggress_config::compile::RuntimeConfig,
613        config_path: Option<String>,
614        compatibility_options: CompatibilityOptions,
615    ) -> Result<Self, RuntimeError> {
616        #[cfg(not(feature = "reverse"))]
617        if !rt_config.reverse_servers.is_empty() || !rt_config.reverse_clients.is_empty() {
618            return Err(RuntimeError::Other(
619                "reverse proxy support not included in this build".to_string(),
620            ));
621        }
622
623        #[cfg(not(feature = "operations"))]
624        if rt_config.admin.as_ref().is_some_and(|a| a.enabled) {
625            return Err(RuntimeError::Other(
626                "admin server support not included in this build; \
627                 enable the 'operations' feature or remove [admin] from config"
628                    .to_string(),
629            ));
630        }
631
632        for lcfg in &rt_config.listeners {
633            if lcfg.unix.is_none() {
634                let _bind_addr: std::net::SocketAddr =
635                    lcfg.bind.parse().map_err(|e| RuntimeError::ListenerBind {
636                        addr: lcfg.bind.clone(),
637                        source: std::io::Error::new(std::io::ErrorKind::InvalidInput, e),
638                    })?;
639            }
640        }
641
642        let udp_metrics = Arc::new(eggress_udp::metrics::UdpMetrics::new());
643        #[cfg(feature = "extended")]
644        let shadowsocks_metrics = Arc::new(eggress_protocol_shadowsocks::ShadowsocksMetrics::new());
645
646        let metrics_registry = Arc::new(eggress_metrics::MetricsRegistry::new());
647        metrics_registry.set_udp_metrics(udp_metrics.clone());
648        #[cfg(feature = "extended")]
649        {
650            metrics_registry.set_shadowsocks_metrics(shadowsocks_metrics.clone());
651        }
652        let metrics: Arc<dyn eggress_server::SessionMetrics> = metrics_registry.clone();
653        let readiness = Arc::new(AtomicBool::new(false));
654
655        let snapshot = compile_runtime_snapshot(&rt_config, None)
656            .map_err(|e| RuntimeError::Config(e.to_string()))?;
657        let snapshot = Arc::new(ArcSwap::from_pointee(snapshot));
658
659        let routing = Arc::new(SharedRoutingService::new_arc(
660            snapshot.load().router.clone(),
661        ));
662
663        let active_connections = Arc::new(AtomicU64::new(0));
664        let connection_counter = Arc::new(AtomicU64::new(1));
665
666        let udp_global_limit = rt_config
667            .listeners
668            .iter()
669            .find_map(|l| l.udp.as_ref().map(|u| u.max_associations_global))
670            .unwrap_or(1024);
671
672        let udp_registry = Arc::new(eggress_udp::registry::UdpAssociationRegistry::new(
673            eggress_udp::limits::UdpLimits {
674                max_associations_global: udp_global_limit,
675                ..Default::default()
676            },
677        ));
678
679        #[cfg(feature = "reverse")]
680        let reverse_metrics = Arc::new(eggress_protocol_reverse::metrics::ReverseMetrics::new());
681        let udp_tasks = TaskTracker::new();
682
683        let state = Arc::new(RuntimeState {
684            snapshot: snapshot.clone(),
685            routing: routing.clone(),
686            metrics: metrics.clone(),
687            readiness,
688            start_time: Instant::now(),
689            active_connections,
690            connection_counter,
691            admin_local_addr: Arc::new(Mutex::new(None)),
692            listener_addrs: Arc::new(Mutex::new(Vec::new())),
693            udp_registry,
694            udp_metrics,
695            #[cfg(feature = "extended")]
696            shadowsocks_metrics,
697            udp_tasks: udp_tasks.clone(),
698            transparent_accepted_total: Arc::new(AtomicU64::new(0)),
699            transparent_original_dst_failed_total: Arc::new(AtomicU64::new(0)),
700            #[cfg(feature = "reverse")]
701            reverse_registry: Arc::new(eggress_admin::ReverseRegistry::new()),
702            #[cfg(feature = "reverse")]
703            reverse_metrics,
704        });
705
706        // Bridge transparent proxy atomics to MetricsRegistry for /metrics
707        metrics_registry.set_transparent_counters(
708            state.transparent_accepted_total.clone(),
709            state.transparent_original_dst_failed_total.clone(),
710        );
711
712        let cancel = CancellationToken::new();
713        let listener_cancel = CancellationToken::new();
714        let connection_cancel = CancellationToken::new();
715        let health_cancel = CancellationToken::new();
716        let admin_cancel = CancellationToken::new();
717        #[cfg(feature = "ssh")]
718        let ssh_sessions = Arc::new(eggress_transport_ssh::SshSessionCache::new());
719
720        let tasks = TaskTracker::new();
721        let connection_tasks = TaskTracker::new();
722
723        let mut health: Option<HealthManager> = None;
724
725        {
726            let upstream_runtimes: Vec<Arc<UpstreamRuntime>> =
727                snapshot.load().upstreams.values().cloned().collect();
728
729            if !upstream_runtimes.is_empty() {
730                let hm = HealthManager::new(health_cancel.clone());
731                health = Some(hm);
732            }
733        }
734
735        let shutdown_grace = rt_config.process.shutdown_grace;
736
737        Ok(ServiceSupervisor {
738            config_path,
739            state,
740            metrics_registry,
741            cancel,
742            listener_cancel,
743            connection_cancel,
744            health_cancel,
745            admin_cancel,
746            health,
747            tasks,
748            connection_tasks,
749            admin_tasks: TaskTracker::new(),
750            shutdown_grace,
751            rt_config,
752            tls_client_config: None,
753            #[cfg(feature = "ssh")]
754            ssh_sessions,
755            compatibility_options,
756        })
757    }
758
759    pub fn state(&self) -> &Arc<RuntimeState> {
760        &self.state
761    }
762
763    pub fn shutdown_token(&self) -> CancellationToken {
764        self.cancel.clone()
765    }
766
767    /// Override the TLS client config used for upstream connections (e.g., Trojan).
768    /// Intended for test-only use (e.g., insecure TLS for self-signed certs).
769    #[allow(dead_code)]
770    pub fn with_tls_client_config(mut self, config: std::sync::Arc<rustls::ClientConfig>) -> Self {
771        self.tls_client_config = Some(config);
772        self
773    }
774
775    /// Classify whether a reload is supported given old and new listener configs.
776    /// Returns `Ok(())` if the reload is safe, or `Err(reason)` if it should be rejected.
777    ///
778    /// UDP-specific reload semantics:
779    /// - UDP bind changes are restart-required.
780    /// - UDP advertise address changes are restart-required if socket bind changes.
781    /// - UDP limits apply to new associations only; existing ones keep their limits.
782    /// - Route changes apply immediately to future UDP packets.
783    fn classify_reload(
784        &self,
785        new_config: &eggress_config::compile::RuntimeConfig,
786    ) -> Result<(), String> {
787        classify_listeners(&self.rt_config.listeners, &new_config.listeners)
788    }
789
790    /// Attempt to reload configuration. Encapsulates the full reload transaction:
791    /// 1. Load and compile new config
792    /// 2. Classify unsupported changes (reject if listener topology changed)
793    /// 3. Build new snapshot with previous snapshot for Arc reuse
794    /// 4. Atomically swap routing and snapshot
795    /// 5. Update stored rt_config for subsequent reloads
796    pub fn reload_config(&mut self) -> ReloadResult {
797        let config_path = match self.config_path {
798            Some(ref p) => p.clone(),
799            None => {
800                return ReloadResult::Rejected {
801                    reason: "no config file path available for reload".to_string(),
802                };
803            }
804        };
805        let new_rt_config = match eggress_config::compile::load_and_compile(&config_path) {
806            Ok(c) => c,
807            Err(e) => {
808                return ReloadResult::Failed {
809                    error: format!("config load: {e}"),
810                };
811            }
812        };
813
814        if let Err(reason) = self.classify_reload(&new_rt_config) {
815            return ReloadResult::Rejected { reason };
816        }
817
818        let prev_snapshot = self.state.snapshot.load();
819        let prev_ref: Option<&CompiledRuntimeSnapshot> = Some(&prev_snapshot);
820        let new_snapshot = match compile_runtime_snapshot(&new_rt_config, prev_ref) {
821            Ok(s) => s,
822            Err(e) => {
823                return ReloadResult::Failed {
824                    error: format!("snapshot build: {e}"),
825                };
826            }
827        };
828
829        let upstream_count = new_snapshot.upstreams.len();
830        let gen = new_snapshot.generation;
831
832        // Snapshot must be published before the router swap. Readers that observe
833        // the new generation via `snapshot.load()` pull the router from that
834        // same snapshot Arc, so any reader seeing the new generation also sees
835        // the router that belongs to it.
836        let new_snapshot = Arc::new(new_snapshot);
837        self.state.snapshot.store(new_snapshot.clone());
838        self.state.routing.swap_arc(new_snapshot.router.clone());
839
840        self.rt_config = new_rt_config;
841
842        ReloadResult::Applied {
843            generation: gen,
844            upstreams: upstream_count,
845        }
846    }
847
848    pub fn run(&mut self) -> Result<(), RuntimeError> {
849        #[allow(unused_variables)]
850        let config_path = self.config_path.clone().unwrap_or_default();
851        let routing = self.state.routing.clone();
852        let listener_cancel = self.listener_cancel.clone();
853        let connection_cancel = self.connection_cancel.clone();
854        let health_cancel = self.health_cancel.clone();
855        let admin_cancel = self.admin_cancel.clone();
856        let cancel = self.cancel.clone();
857        #[allow(unused_variables)]
858        let metrics = self.state.metrics.clone();
859        let readiness = self.state.readiness.clone();
860        #[cfg(feature = "operations")]
861        let admin_state_ref = self.state.clone();
862        let active_connections = self.state.active_connections.clone();
863        let shutdown_grace = self.shutdown_grace;
864        let tasks = self.tasks.clone();
865        let connection_tasks = self.connection_tasks.clone();
866        let admin_tasks = self.admin_tasks.clone();
867        let health = std::sync::Arc::new(std::sync::Mutex::new(self.health.take()));
868        #[allow(unused_variables)]
869        let health_clone = health.clone();
870        let health_for_run = health.clone();
871        let snapshot = self.state.snapshot.clone();
872        let state_ref = self.state.clone();
873        let rt_config = self.rt_config.clone();
874        let tls_client_config = self.tls_client_config.clone();
875        let compatibility_options = self.compatibility_options.clone();
876        #[cfg(feature = "ssh")]
877        let ssh_sessions = self.ssh_sessions.clone();
878
879        let handshake_timeout = rt_config.timeouts.handshake;
880        let connect_timeout = rt_config.timeouts.connect;
881
882        #[cfg(feature = "operations")]
883        let listener_infos_provider: Arc<RuntimeAdminListenerInfos> =
884            Arc::new(RuntimeAdminListenerInfos {
885                state: admin_state_ref.clone(),
886            });
887
888        #[cfg(feature = "operations")]
889        let metrics_registry_for_admin = self.metrics_registry.clone();
890
891        let run_async = async move {
892            #[cfg(feature = "operations")]
893            let mut compatibility_system_proxy: Option<
894                eggress_system_proxy::AppliedProxy,
895            > = None;
896
897            #[cfg(feature = "operations")]
898            let metrics_registry = metrics_registry_for_admin;
899            // Start health probes inside the runtime context
900            {
901                let mut guard = health_for_run.lock().unwrap_or_else(|e| e.into_inner());
902                if let Some(ref mut hm) = *guard {
903                    let upstream_runtimes: Vec<Arc<UpstreamRuntime>> =
904                        snapshot.load().upstreams.values().cloned().collect();
905                    if !upstream_runtimes.is_empty() {
906                        hm.start_probes(&upstream_runtimes);
907                    }
908                }
909            }
910
911            let current_snapshot = snapshot.load();
912            let listener_configs = current_snapshot.listeners.clone();
913            #[cfg(feature = "operations")]
914            let admin_config = current_snapshot.admin.clone();
915            drop(current_snapshot);
916
917            if listener_configs.is_empty() {
918                tracing::warn!("no listeners configured; the proxy will not accept connections");
919            }
920
921            let mut prepared = Vec::new();
922            #[cfg(feature = "quic")]
923            let mut prepared_quic = Vec::<PreparedQuicListener>::new();
924            let compatibility_auth_reuse = compatibility_options
925                .auth_timeout
926                .map(eggress_server::accept::AuthReuseCache::new)
927                .map(Arc::new);
928            #[cfg(unix)]
929            let mut unix_listener_args = Vec::new();
930            let mut transparent_listener_args = Vec::new();
931
932            for lcfg in &listener_configs {
933                let protocols: Vec<ProtocolId> = lcfg.protocols.to_vec();
934
935                let auth = match &lcfg.auth {
936                    Some(auth_cfg) => {
937                        if auth_cfg.auth_type == "password" {
938                            let username = auth_cfg.username.clone().unwrap_or_default();
939                            let password = auth_cfg.password.clone().unwrap_or_default();
940                            if let Some(reuse) = compatibility_auth_reuse.clone() {
941                                eggress_server::accept::InboundAuthentication::UsernamePasswordWithReuse {
942                                    username,
943                                    password,
944                                    reuse,
945                                }
946                            } else {
947                                eggress_server::accept::InboundAuthentication::UsernamePassword {
948                                    username,
949                                    password,
950                                }
951                            }
952                        } else {
953                            eggress_server::accept::InboundAuthentication::None
954                        }
955                    }
956                    None => eggress_server::accept::InboundAuthentication::None,
957                };
958
959                let connection_limit = lcfg.connection_limit.unwrap_or(1024) as usize;
960
961                // Handle Unix domain socket listeners separately
962                #[allow(unused_variables)]
963                if let Some(ref unix_cfg) = lcfg.unix {
964                    #[cfg(unix)]
965                    {
966                        match eggress_server::listener::unix::create_unix_listener(
967                            &eggress_server::listener::unix::UnixListenerConfig::from_compiled(
968                                &unix_cfg.path,
969                                unix_cfg.unlink_existing,
970                                Some(unix_cfg.mode),
971                            ),
972                        ) {
973                            Ok(unix_listener) => {
974                                tracing::info!(
975                                    "unix socket listener created at {} ({})",
976                                    unix_cfg.path.display(),
977                                    lcfg.name
978                                );
979                                unix_listener_args.push((
980                                    lcfg.name.clone(),
981                                    unix_listener,
982                                    protocols,
983                                    auth,
984                                    handshake_timeout,
985                                    connection_limit as u64,
986                                    lcfg.tls.clone(),
987                                    lcfg.shadowsocks.clone(),
988                                    lcfg.trojan.clone(),
989                                    lcfg.udp.clone(),
990                                ));
991                                continue;
992                            }
993                            Err(e) => {
994                                tracing::error!(
995                                    "failed to bind unix socket at {} for listener '{}': {e}",
996                                    unix_cfg.path.display(),
997                                    lcfg.name
998                                );
999                                continue;
1000                            }
1001                        }
1002                    }
1003                    #[cfg(not(unix))]
1004                    {
1005                        tracing::error!(
1006                            "unix socket listener '{}' skipped: not supported on this platform",
1007                            lcfg.name
1008                        );
1009                        continue;
1010                    }
1011                }
1012
1013                // Handle transparent TCP listeners
1014                if let Some(ref transparent_cfg) = lcfg.transparent {
1015                    if transparent_cfg.enabled {
1016                        let capability = check_capability(PlatformCapability::LinuxOriginalDstIpv4);
1017                        if capability != crate::platform::CapabilityStatus::Available {
1018                            #[cfg(feature = "operations")]
1019                            state_ref.metrics.record_platform_capability_check_failure();
1020                            let _cap_span = tracing::info_span!(
1021                                "capability_check_failed",
1022                                capability = %PlatformCapability::LinuxOriginalDstIpv4,
1023                                status = %capability,
1024                                listener = %lcfg.name,
1025                            );
1026                            tracing::warn!(
1027                                "transparent proxy not available for listener '{}' ({}); \
1028                                 falling back to normal TCP listener",
1029                                lcfg.name,
1030                                capability
1031                            );
1032                        } else {
1033                            let bind_addr: std::net::SocketAddr =
1034                                lcfg.bind.parse().map_err(|e| RuntimeError::ListenerBind {
1035                                    addr: lcfg.bind.clone(),
1036                                    source: std::io::Error::new(
1037                                        std::io::ErrorKind::InvalidInput,
1038                                        e,
1039                                    ),
1040                                })?;
1041
1042                            let transparent_listener =
1043                                eggress_server::listener::transparent::TransparentListener::bind(
1044                                    &bind_addr.to_string(),
1045                                )
1046                                .await
1047                                .map_err(|e| {
1048                                    RuntimeError::ListenerBind {
1049                                        addr: lcfg.bind.clone(),
1050                                        source: e,
1051                                    }
1052                                })?;
1053
1054                            let local_addr = transparent_listener.local_addr().map_err(|e| {
1055                                RuntimeError::ListenerBind {
1056                                    addr: lcfg.bind.clone(),
1057                                    source: e,
1058                                }
1059                            })?;
1060
1061                            tracing::info!(
1062                                "transparent TCP listener listening on {local_addr} ({})",
1063                                lcfg.name
1064                            );
1065
1066                            transparent_listener_args.push((
1067                                lcfg.name.clone(),
1068                                transparent_listener,
1069                                protocols,
1070                                auth,
1071                                handshake_timeout,
1072                                connection_limit as u64,
1073                                lcfg.tls.clone(),
1074                                lcfg.shadowsocks.clone(),
1075                                lcfg.trojan.clone(),
1076                                lcfg.udp.clone(),
1077                            ));
1078                            continue;
1079                        }
1080                    }
1081                }
1082
1083                #[cfg(feature = "quic")]
1084                if protocols.contains(&ProtocolId::Quic) || protocols.contains(&ProtocolId::Http3) {
1085                    let tls = lcfg.tls.clone().ok_or_else(|| {
1086                        RuntimeError::Other(format!(
1087                            "QUIC/HTTP3 listener '{}' requires certificate and key material",
1088                            lcfg.name
1089                        ))
1090                    })?;
1091                    let bind_addr: std::net::SocketAddr =
1092                        lcfg.bind.parse().map_err(|e| RuntimeError::ListenerBind {
1093                            addr: lcfg.bind.clone(),
1094                            source: std::io::Error::new(std::io::ErrorKind::InvalidInput, e),
1095                        })?;
1096                    let listener = eggress_transport_quic::QuicListener::bind(
1097                        bind_addr,
1098                        eggress_transport_quic::QuicServerConfig {
1099                            certificate_pem: tls.cert_pem.clone(),
1100                            private_key_pem: tls.key_pem.clone(),
1101                            idle_timeout: Duration::from_secs(60),
1102                            max_concurrent_streams: lcfg.connection_limit.unwrap_or(1024).max(1),
1103                            alpn_protocols: if protocols.contains(&ProtocolId::Http3) {
1104                                vec![b"h3".to_vec()]
1105                            } else {
1106                                Vec::new()
1107                            },
1108                        },
1109                    )
1110                    .await
1111                    .map_err(|e| RuntimeError::ListenerBind {
1112                        addr: lcfg.bind.clone(),
1113                        source: std::io::Error::other(e.to_string()),
1114                    })?;
1115                    let local_addr =
1116                        listener
1117                            .local_addr()
1118                            .map_err(|e| RuntimeError::ListenerBind {
1119                                addr: lcfg.bind.clone(),
1120                                source: std::io::Error::other(e.to_string()),
1121                            })?;
1122                    tracing::info!("QUIC listening on {local_addr} ({})", lcfg.name);
1123                    prepared_quic.push(PreparedQuicListener {
1124                        name: lcfg.name.clone(),
1125                        protocols,
1126                        listener,
1127                        local_addr,
1128                        auth,
1129                        handshake_timeout,
1130                        connection_limit: lcfg.connection_limit.unwrap_or(1024) as u64,
1131                    });
1132                    continue;
1133                }
1134
1135                // Standard TCP listener path
1136                let bind_addr: std::net::SocketAddr =
1137                    lcfg.bind.parse().map_err(|e| RuntimeError::ListenerBind {
1138                        addr: lcfg.bind.clone(),
1139                        source: std::io::Error::new(std::io::ErrorKind::InvalidInput, e),
1140                    })?;
1141
1142                let config = TcpListenerConfig {
1143                    bind_addr,
1144                    protocols: protocols.clone(),
1145                    auth_required: false,
1146                    handshake_timeout,
1147                    connection_limit,
1148                };
1149
1150                let listener = TcpListener::new_with_reuse_port(
1151                    &config,
1152                    listener_cancel.clone(),
1153                    lcfg.reuse_port.unwrap_or(false),
1154                )
1155                .await
1156                .map_err(|e| RuntimeError::ListenerBind {
1157                    addr: lcfg.bind.clone(),
1158                    source: e,
1159                })?;
1160                let local_addr = listener
1161                    .local_addr()
1162                    .map_err(|e| RuntimeError::ListenerBind {
1163                        addr: lcfg.bind.clone(),
1164                        source: e,
1165                    })?;
1166                tracing::info!("listening on {local_addr} ({})", lcfg.name);
1167
1168                prepared.push(PreparedListener {
1169                    name: lcfg.name.clone(),
1170                    bind: lcfg.bind.clone(),
1171                    protocols,
1172                    listener,
1173                    local_addr,
1174                    auth,
1175                    handshake_timeout,
1176                    udp: lcfg.udp.clone(),
1177                    tls: lcfg.tls.clone(),
1178                    shadowsocks: lcfg.shadowsocks.clone(),
1179                    trojan: lcfg.trojan.clone(),
1180                    fixed_target: lcfg.fixed_target.clone(),
1181                    local_bind: lcfg.local_bind.clone(),
1182                });
1183            }
1184
1185            #[cfg(feature = "operations")]
1186            {
1187                let listener_infos: Vec<eggress_admin::ListenerInfo> = prepared
1188                    .iter()
1189                    .map(|p| eggress_admin::ListenerInfo {
1190                        name: p.name.clone(),
1191                        bind: p.bind.clone(),
1192                        local_addr: p.local_addr.to_string(),
1193                        protocols: p.protocols.iter().map(|p| p.to_string()).collect(),
1194                        udp_enabled: p.udp.as_ref().is_some_and(|u| u.enabled),
1195                        mode: Some("standard".to_string()),
1196                        capability_status: None,
1197                        original_dst_support: None,
1198                        unix_socket_path: None,
1199                        unix_socket_unlink_existing: None,
1200                    })
1201                    .collect();
1202                drop(listener_infos);
1203            }
1204
1205            // Store listener addresses for admin snapshot, indexed by config order.
1206            // Build a lookup map from all listener types (standard, transparent, unix).
1207            {
1208                let mut addr_map: std::collections::HashMap<String, Option<std::net::SocketAddr>> =
1209                    std::collections::HashMap::new();
1210                for p in &prepared {
1211                    addr_map.insert(p.name.clone(), Some(p.local_addr));
1212                }
1213                #[cfg(feature = "quic")]
1214                for p in &prepared_quic {
1215                    addr_map.insert(p.name.clone(), Some(p.local_addr));
1216                }
1217                for (name, transparent_listener, _, _, _, _, _, _, _, _) in
1218                    &transparent_listener_args
1219                {
1220                    let addr = transparent_listener.local_addr().ok();
1221                    addr_map.insert(name.clone(), addr);
1222                }
1223                #[cfg(unix)]
1224                for (name, _, _, _, _, _, _, _, _, _) in &unix_listener_args {
1225                    // Unix domain sockets don't have a meaningful TCP socket address
1226                    addr_map.insert(name.clone(), None);
1227                }
1228                let addrs: Vec<Option<std::net::SocketAddr>> = listener_configs
1229                    .iter()
1230                    .map(|lcfg| addr_map.get(&lcfg.name).copied().flatten())
1231                    .collect();
1232                *state_ref
1233                    .listener_addrs
1234                    .lock()
1235                    .unwrap_or_else(|e| e.into_inner()) = addrs;
1236            }
1237
1238            #[cfg(feature = "extended")]
1239            let mut shadowsocks_udp_relays = Vec::new();
1240            let mut echo_udp_relays = Vec::new();
1241
1242            for prepared_listener in &prepared {
1243                if let Some(ref udp_cfg) = prepared_listener.udp {
1244                    #[cfg(feature = "extended")]
1245                    if udp_cfg.mode == eggress_udp::UdpMode::ShadowsocksUdp {
1246                        shadowsocks_udp_relays.push(
1247                            prepare_shadowsocks_udp_relay(
1248                                prepared_listener,
1249                                udp_cfg,
1250                                routing.clone(),
1251                                &state_ref,
1252                            )
1253                            .await?,
1254                        );
1255                    }
1256                    if udp_cfg.mode == eggress_udp::UdpMode::Echo {
1257                        let socket =
1258                            Arc::new(tokio::net::UdpSocket::bind(udp_cfg.bind).await.map_err(
1259                                |e| RuntimeError::ListenerBind {
1260                                    addr: udp_cfg.bind.to_string(),
1261                                    source: e,
1262                                },
1263                            )?);
1264                        echo_udp_relays.push((prepared_listener.name.clone(), socket, None));
1265                    } else if udp_cfg.mode == eggress_udp::UdpMode::FixedTarget {
1266                        let socket =
1267                            Arc::new(tokio::net::UdpSocket::bind(udp_cfg.bind).await.map_err(
1268                                |e| RuntimeError::ListenerBind {
1269                                    addr: udp_cfg.bind.to_string(),
1270                                    source: e,
1271                                },
1272                            )?);
1273                        echo_udp_relays.push((
1274                            prepared_listener.name.clone(),
1275                            socket,
1276                            udp_cfg.fixed_target.clone(),
1277                        ));
1278                    }
1279                }
1280            }
1281
1282            #[cfg(feature = "extended")]
1283            for (socket, relay_config) in shadowsocks_udp_relays {
1284                let relay_cancel = cancel.clone();
1285                tasks.spawn(async move {
1286                    let result =
1287                        eggress_udp::standalone_shadowsocks::shadowsocks_standalone_udp_relay(
1288                            socket,
1289                            relay_config,
1290                            relay_cancel,
1291                        )
1292                        .await;
1293                    if let Err(error) = result {
1294                        tracing::debug!(
1295                            %error,
1296                            "Shadowsocks UDP relay ended with error"
1297                        );
1298                    }
1299                });
1300            }
1301
1302            for (listener_name, socket, fixed_target) in echo_udp_relays {
1303                let relay_cancel = cancel.clone();
1304                let max_size = listener_configs
1305                    .iter()
1306                    .find(|l| l.name == listener_name)
1307                    .and_then(|l| l.udp.as_ref())
1308                    .map(|u| u.max_datagram_size)
1309                    .unwrap_or(65535);
1310                tasks.spawn(async move {
1311                    let mut buf = vec![0u8; max_size];
1312                    let target_socket = if let Some(target) = fixed_target {
1313                        let addr = match target.host {
1314                            eggress_core::TargetHost::Ip(ip) => std::net::SocketAddr::new(ip, target.port),
1315                            eggress_core::TargetHost::Domain(domain) => match tokio::net::lookup_host((domain.as_str(), target.port)).await {
1316                                Ok(mut addrs) => match addrs.next() { Some(addr) => addr, None => return },
1317                                Err(_) => return,
1318                            },
1319                        };
1320                        let target_socket = match tokio::net::UdpSocket::bind("0.0.0.0:0").await {
1321                            Ok(s) => s,
1322                            Err(_) => return,
1323                        };
1324                        if target_socket.connect(addr).await.is_err() { return; }
1325                        Some(target_socket)
1326                    } else { None };
1327                    let mut response = vec![0u8; max_size];
1328                    loop {
1329                        tokio::select! {
1330                            result = socket.recv_from(&mut buf) => match result {
1331                                Ok((n, peer)) => {
1332                                    if let Some(ref target_socket) = target_socket {
1333                                        if target_socket.send(&buf[..n]).await.is_ok() {
1334                                            if let Ok(m) = target_socket.recv(&mut response).await { let _ = socket.send_to(&response[..m], peer).await; }
1335                                        }
1336                                    } else { let _ = socket.send_to(&buf[..n], peer).await; }
1337                                }
1338                                Err(_) => break,
1339                            },
1340                            _ = relay_cancel.cancelled() => break,
1341                        }
1342                    }
1343                });
1344            }
1345
1346            // Spawn transparent listener accept loops
1347            for (
1348                listener_name,
1349                transparent_listener,
1350                protocols,
1351                auth,
1352                hs_timeout,
1353                connection_limit,
1354                tls_cfg,
1355                ss_cfg,
1356                trojan_cfg,
1357                udp_cfg,
1358            ) in transparent_listener_args
1359            {
1360                let routing = routing.clone();
1361                let state = state_ref.clone();
1362                let conn_tasks = connection_tasks.clone();
1363                let conn_cancel = connection_cancel.clone();
1364                let tls_client_config = tls_client_config.clone();
1365                let listener_cancel = listener_cancel.clone();
1366
1367                #[cfg(feature = "ssh")]
1368                let listener_ssh_sessions = ssh_sessions.clone();
1369
1370                tasks.spawn(async move {
1371                    let proto_slice: Arc<[ProtocolId]> = protocols.clone().into();
1372                    let transparent_listener_inner = transparent_listener.inner();
1373                    let listener_active = Arc::new(AtomicU64::new(0));
1374
1375                    let transparent_accepted = state.transparent_accepted_total.clone();
1376                    let transparent_dst_failed =
1377                        state.transparent_original_dst_failed_total.clone();
1378
1379                    loop {
1380                        let accept_result = tokio::select! {
1381                            result = transparent_listener_inner.accept() => result,
1382                            _ = listener_cancel.cancelled() => {
1383                                break;
1384                            }
1385                        };
1386
1387                        let (stream, _peer) = match accept_result {
1388                            Ok(s) => s,
1389                            Err(e) => {
1390                                if e.to_string().contains("listener cancelled") {
1391                                    break;
1392                                }
1393                                tracing::error!(
1394                                    "transparent accept error on '{}': {e}",
1395                                    listener_name
1396                                );
1397                                continue;
1398                            }
1399                        };
1400
1401                        let connection_slot = match ListenerConnectionSlot::try_acquire(
1402                            &listener_active,
1403                            connection_limit,
1404                        ) {
1405                            Some(slot) => slot,
1406                            None => {
1407                                tracing::debug!(
1408                                    listener = %listener_name,
1409                                    limit = connection_limit,
1410                                    "dropping transparent connection: connection limit reached"
1411                                );
1412                                drop(stream);
1413                                continue;
1414                            }
1415                        };
1416
1417                        transparent_accepted.fetch_add(1, Ordering::Relaxed);
1418
1419                        let original_dst =
1420                            match eggress_server::listener::transparent::get_original_destination(&stream) {
1421                                Ok(addr) => addr,
1422                                Err(e) => {
1423                                    drop(connection_slot);
1424                                    transparent_dst_failed.fetch_add(1, Ordering::Relaxed);
1425                                    let _span = tracing::info_span!(
1426                                        "transparent_original_dst_failed",
1427                                        listener = %listener_name,
1428                                        error = %e,
1429                                    );
1430                                    tracing::warn!(
1431                                        "failed to get original destination for transparent connection on '{}': {e}",
1432                                        listener_name
1433                                    );
1434                                    continue;
1435                                }
1436                            };
1437
1438                        let peer = stream
1439                            .peer_addr()
1440                            .unwrap_or_else(|_| {
1441                                std::net::SocketAddr::new(
1442                                    std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED),
1443                                    0,
1444                                )
1445                            });
1446
1447                        let routing = routing.clone();
1448                        let tls_client_config = tls_client_config.clone();
1449                        let listener_str = listener_name.clone();
1450                        let conn_id = state
1451                            .connection_counter
1452                            .fetch_add(1, Ordering::Relaxed);
1453                        let conn_protocols = proto_slice.clone();
1454                        let conn_auth = auth.clone();
1455                        let conn_metrics = state.metrics.clone();
1456                        #[cfg(feature = "extended")]
1457                        let conn_ss_metrics = state.shadowsocks_metrics.clone();
1458                        let active = state.active_connections.clone();
1459                        let conn_cancel = conn_cancel.child_token();
1460                        let generation = state.snapshot.load().generation;
1461                        let tls_config = tls_cfg.clone();
1462                        let ss_config = ss_cfg.clone();
1463                        let trojan_config = trojan_cfg.clone();
1464
1465                        let udp_svc = udp_cfg.as_ref().map(|udp_config| {
1466                            Arc::new(RuntimeUdpService {
1467                                _listener_name: listener_name.clone(),
1468                                udp_config: udp_config.clone(),
1469                                registry: state.udp_registry.clone(),
1470                                metrics: state.metrics.clone(),
1471                                udp_metrics: state.udp_metrics.clone(),
1472                                routing: routing.clone(),
1473                                udp_tasks: state.udp_tasks.clone(),
1474                            }) as Arc<dyn eggress_server::UdpService>
1475                        });
1476
1477                        active.fetch_add(1, Ordering::Relaxed);
1478
1479                        #[cfg(feature = "ssh")]
1480                        let conn_ssh_sessions = listener_ssh_sessions.clone();
1481                        conn_tasks.spawn(async move {
1482                            let _connection_slot = connection_slot;
1483                            let started = std::time::Instant::now();
1484
1485                            let stream: eggress_core::BoxStream =
1486                                if let Some(ref tls_cfg) = tls_config {
1487                                    let server_config = match eggress_transport_tls::TlsServerConfigBuilder::new()
1488                                        .with_certificate_pem(&tls_cfg.cert_pem)
1489                                        .and_then(|b| b.with_key_pem(&tls_cfg.key_pem))
1490                                        .and_then(|b| {
1491                                            let b = if tls_cfg.alpn.is_empty() { b } else { b.with_alpn(tls_cfg.alpn.clone()) };
1492                                            b.build()
1493                                        }) {
1494                                            Ok(c) => c,
1495                                            Err(e) => {
1496                                                tracing::error!(%peer, "TLS config error: {e}");
1497                                                active.fetch_sub(1, Ordering::Relaxed);
1498                                                return;
1499                                            }
1500                                        };
1501                                    match eggress_transport_tls::tls_accept(Box::new(stream), server_config).await {
1502                                        Ok(s) => s,
1503                                        Err(e) => {
1504                                            tracing::debug!(%peer, "TLS accept failed: {e}");
1505                                            active.fetch_sub(1, Ordering::Relaxed);
1506                                            return;
1507                                        }
1508                                    }
1509                                } else {
1510                                    Box::new(stream)
1511                                };
1512
1513                            let config = eggress_server::ConnectionConfig {
1514                                routing: routing as Arc<dyn RouteService>,
1515                                context: eggress_server::ConnectionContext {
1516                                    source: Some(peer),
1517                                    listener: listener_str.clone(),
1518                                    generation,
1519                                },
1520                                handshake_timeout: hs_timeout,
1521                                connect_timeout,
1522                                protocols: conn_protocols,
1523                                authentication: conn_auth,
1524                                metrics: Some(conn_metrics),
1525                                udp: udp_svc,
1526                                tls_client_config,
1527                                shadowsocks: ss_config.map(
1528                                    |ss| eggress_server::accept::InboundShadowsocksConfig {
1529                                        method: ss.method,
1530                                        password: ss.password,
1531                                        #[cfg(feature = "pproxy-legacy")]
1532                                        auth_prefix: ss.auth_prefix.map(String::into_bytes),
1533                                        #[cfg(feature = "pproxy-legacy")]
1534                                        plugins: ss.plugins,
1535                                    },
1536                                ),
1537                                #[cfg(feature = "extended")]
1538                                shadowsocks_metrics: Some(conn_ss_metrics),
1539                                #[cfg(not(feature = "extended"))]
1540                                shadowsocks_metrics: None,
1541                                trojan: trojan_config.map(
1542                                    |t| eggress_server::accept::InboundTrojanConfig {
1543                                        password: t.password,
1544                                        fallback: t.fallback,
1545                                    },
1546                                ),
1547                                fixed_target: None,
1548                                local_bind: None,
1549                                #[cfg(feature = "ssh")]
1550                                ssh_sessions: Some(conn_ssh_sessions),
1551                            };
1552
1553                            let report = tokio::select! {
1554                                report = eggress_server::serve_connection(stream, config)
1555                                    .instrument(tracing::info_span!(
1556                                        "conn",
1557                                        id = conn_id,
1558                                        peer = %peer,
1559                                        original_dst = %original_dst,
1560                                        listener_type = "transparent",
1561                                        listener = %listener_str,
1562                                    )) => {
1563                                report
1564                            }
1565                                _ = conn_cancel.cancelled() => {
1566                                    eggress_server::SessionReport::cancelled(
1567                                        None,
1568                                        None,
1569                                        String::new(),
1570                                    )
1571                                }
1572                            };
1573
1574                            active.fetch_sub(1, Ordering::Relaxed);
1575
1576                            tracing::info!(
1577                                protocol = ?report.protocol,
1578                                target = ?report.target,
1579                                original_dst = %original_dst,
1580                                route = %report.route,
1581                                outcome = ?report.outcome,
1582                                bytes_upstream = report.bytes_upstream,
1583                                bytes_downstream = report.bytes_downstream,
1584                                duration_ms = started.elapsed().as_millis() as u64,
1585                                "transparent connection completed",
1586                            );
1587                        });
1588                    }
1589                });
1590            }
1591
1592            // Spawn Unix domain socket accept loops
1593            #[cfg(unix)]
1594            for (
1595                listener_name,
1596                unix_listener,
1597                protocols,
1598                auth,
1599                hs_timeout,
1600                connection_limit,
1601                tls_cfg,
1602                ss_cfg,
1603                trojan_cfg,
1604                udp_cfg,
1605            ) in unix_listener_args
1606            {
1607                let routing = routing.clone();
1608                let state = state_ref.clone();
1609                let conn_tasks = connection_tasks.clone();
1610                let conn_cancel = connection_cancel.clone();
1611                let tls_client_config = tls_client_config.clone();
1612                let listener_cancel = listener_cancel.clone();
1613
1614                let socket_path = unix_listener.path().display().to_string();
1615
1616                #[cfg(feature = "ssh")]
1617                let listener_ssh_sessions = ssh_sessions.clone();
1618                tasks.spawn(async move {
1619                    let proto_slice: Arc<[ProtocolId]> = protocols.clone().into();
1620                    let listener_active = Arc::new(AtomicU64::new(0));
1621
1622                    let _accept_loop_span = tracing::info_span!(
1623                        "unix_accept_loop",
1624                        listener = %listener_name,
1625                        socket_path = %socket_path,
1626                    );
1627
1628                    loop {
1629                        let (stream, _peer_addr) = tokio::select! {
1630                            result = unix_listener.accept() => match result {
1631                                Ok(r) => r,
1632                                Err(e) => {
1633                                    tracing::error!("unix accept error on '{}': {e}", listener_name);
1634                                    continue;
1635                                }
1636                            },
1637                            _ = listener_cancel.cancelled() => {
1638                                break;
1639                            }
1640                        };
1641
1642                        let connection_slot = match ListenerConnectionSlot::try_acquire(
1643                            &listener_active,
1644                            connection_limit,
1645                        ) {
1646                            Some(slot) => slot,
1647                            None => {
1648                                tracing::debug!(
1649                                    listener = %listener_name,
1650                                    limit = connection_limit,
1651                                    "dropping Unix connection: connection limit reached"
1652                                );
1653                                drop(stream);
1654                                continue;
1655                            }
1656                        };
1657
1658                        state.metrics.record_unix_listener_connection_accepted();
1659
1660                        let routing = routing.clone();
1661                        let tls_client_config = tls_client_config.clone();
1662                        let listener_str = listener_name.clone();
1663                        let conn_id =
1664                            state.connection_counter.fetch_add(1, Ordering::Relaxed);
1665                        let conn_protocols = proto_slice.clone();
1666                        let conn_auth = auth.clone();
1667                        let conn_metrics = state.metrics.clone();
1668                        #[cfg(feature = "extended")]
1669                        let conn_ss_metrics = state.shadowsocks_metrics.clone();
1670                        let active = state.active_connections.clone();
1671                        let conn_cancel = conn_cancel.child_token();
1672                        let generation = state.snapshot.load().generation;
1673
1674                        let tls_config = tls_cfg.clone();
1675                        let ss_config = ss_cfg.clone();
1676                        let trojan_config = trojan_cfg.clone();
1677                        let socket_path_clone = socket_path.clone();
1678                        let listener_str_for_span = listener_str.clone();
1679
1680                        let udp_svc = udp_cfg.as_ref().map(|udp_config| {
1681                            Arc::new(RuntimeUdpService {
1682                                _listener_name: listener_name.clone(),
1683                                udp_config: udp_config.clone(),
1684                                registry: state.udp_registry.clone(),
1685                                metrics: state.metrics.clone(),
1686                                udp_metrics: state.udp_metrics.clone(),
1687                                routing: routing.clone(),
1688                                udp_tasks: state.udp_tasks.clone(),
1689                            }) as Arc<dyn eggress_server::UdpService>
1690                        });
1691
1692                        active.fetch_add(1, Ordering::Relaxed);
1693
1694                        #[cfg(feature = "ssh")]
1695                        let conn_ssh_sessions = listener_ssh_sessions.clone();
1696                        conn_tasks.spawn(async move {
1697                            let _connection_slot = connection_slot;
1698                            let started = std::time::Instant::now();
1699
1700                            let stream: eggress_core::BoxStream =
1701                                if let Some(ref tls_cfg) = tls_config {
1702                                    let server_config = match eggress_transport_tls::TlsServerConfigBuilder::new()
1703                                        .with_certificate_pem(&tls_cfg.cert_pem)
1704                                        .and_then(|b| b.with_key_pem(&tls_cfg.key_pem))
1705                                        .and_then(|b| {
1706                                            let b = if tls_cfg.alpn.is_empty() { b } else { b.with_alpn(tls_cfg.alpn.clone()) };
1707                                            b.build()
1708                                        }) {
1709                                            Ok(c) => c,
1710                                            Err(e) => {
1711                                                tracing::error!("TLS config error for unix connection: {e}");
1712                                                active.fetch_sub(1, Ordering::Relaxed);
1713                                                return;
1714                                            }
1715                                        };
1716                                    match eggress_transport_tls::tls_accept(Box::new(stream), server_config).await {
1717                                        Ok(s) => s,
1718                                        Err(e) => {
1719                                            tracing::debug!("TLS accept failed for unix connection: {e}");
1720                                            active.fetch_sub(1, Ordering::Relaxed);
1721                                            return;
1722                                        }
1723                                    }
1724                                } else {
1725                                    Box::new(stream)
1726                                };
1727
1728                            let peer = std::net::SocketAddr::new(
1729                                std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
1730                                0,
1731                            );
1732
1733                            let config = eggress_server::ConnectionConfig {
1734                                routing: routing as Arc<dyn RouteService>,
1735                                context: eggress_server::ConnectionContext {
1736                                    source: Some(peer),
1737                                    listener: listener_str,
1738                                    generation,
1739                                },
1740                                handshake_timeout: hs_timeout,
1741                                connect_timeout,
1742                                protocols: conn_protocols,
1743                                authentication: conn_auth,
1744                                metrics: Some(conn_metrics),
1745                                udp: udp_svc,
1746                                tls_client_config,
1747                                shadowsocks: ss_config.map(
1748                                    |ss| eggress_server::accept::InboundShadowsocksConfig {
1749                                        method: ss.method,
1750                                        password: ss.password,
1751                                        #[cfg(feature = "pproxy-legacy")]
1752                                        auth_prefix: ss.auth_prefix.map(String::into_bytes),
1753                                        #[cfg(feature = "pproxy-legacy")]
1754                                        plugins: ss.plugins,
1755                                    },
1756                                ),
1757                                #[cfg(feature = "extended")]
1758                                shadowsocks_metrics: Some(conn_ss_metrics),
1759                                #[cfg(not(feature = "extended"))]
1760                                shadowsocks_metrics: None,
1761                                trojan: trojan_config.map(
1762                                    |t| eggress_server::accept::InboundTrojanConfig {
1763                                        password: t.password,
1764                                        fallback: t.fallback,
1765                                    },
1766                                ),
1767                                fixed_target: None,
1768                                local_bind: None,
1769                                #[cfg(feature = "ssh")]
1770                                ssh_sessions: Some(conn_ssh_sessions),
1771                            };
1772
1773                            let report = tokio::select! {
1774                                report = eggress_server::serve_connection(stream, config)
1775                                    .instrument(tracing::info_span!(
1776                                        "conn",
1777                                        id = conn_id,
1778                                        peer = %peer,
1779                                        listener_type = "unix",
1780                                        listener = %listener_str_for_span,
1781                                        socket_path = %socket_path_clone,
1782                                    )) => {
1783                                    report
1784                                }
1785                                _ = conn_cancel.cancelled() => {
1786                                    eggress_server::SessionReport::cancelled(
1787                                        None,
1788                                        None,
1789                                        String::new(),
1790                                    )
1791                                }
1792                            };
1793
1794                            active.fetch_sub(1, Ordering::Relaxed);
1795
1796                            tracing::info!(
1797                                protocol = ?report.protocol,
1798                                target = ?report.target,
1799                                route = %report.route,
1800                                outcome = ?report.outcome,
1801                                bytes_upstream = report.bytes_upstream,
1802                                bytes_downstream = report.bytes_downstream,
1803                                duration_ms = started.elapsed().as_millis() as u64,
1804                                "unix connection completed",
1805                            );
1806                        });
1807                    }
1808
1809                    // Cleanup socket file on shutdown
1810                    unix_listener.cleanup().unwrap_or_else(|e| {
1811                        tracing::warn!("failed to cleanup unix socket: {e}");
1812                    });
1813                });
1814            }
1815
1816            // Spawn standard TCP accept loops
1817            #[cfg(feature = "operations")]
1818            let compatibility_proxy_selection = prepared
1819                .iter()
1820                .find(|listener| {
1821                    listener.protocols.contains(&ProtocolId::Socks5)
1822                        && listener.local_addr.port() != 0
1823                })
1824                .map(|listener| {
1825                    (
1826                        eggress_system_proxy::CompatibilityProxyKind::Socks5,
1827                        listener.local_addr.port(),
1828                    )
1829                })
1830                .or_else(|| {
1831                    prepared
1832                        .iter()
1833                        .find(|listener| {
1834                            listener.protocols.contains(&ProtocolId::Http)
1835                                && listener.local_addr.port() != 0
1836                        })
1837                        .map(|listener| {
1838                            (
1839                                eggress_system_proxy::CompatibilityProxyKind::Http,
1840                                listener.local_addr.port(),
1841                            )
1842                        })
1843                });
1844
1845            // All compatibility TCP listeners are bound before this point,
1846            // so --sys can use the actual selected port. Apply before any
1847            // accept loop or admin task is started; an apply failure is
1848            // therefore still a pre-run startup error.
1849            if compatibility_options.system_proxy {
1850                #[cfg(feature = "operations")]
1851                {
1852                    let selected = compatibility_proxy_selection.ok_or_else(|| {
1853                        RuntimeError::Other(
1854                            "--sys requires a usable local HTTP or SOCKS5 listener".to_string(),
1855                        )
1856                    })?;
1857                    let address = std::net::SocketAddr::new(
1858                        std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
1859                        selected.1,
1860                    );
1861                    compatibility_system_proxy = Some(
1862                        eggress_system_proxy::apply_compatibility_proxy(selected.0, address)
1863                            .map_err(RuntimeError::Other)?,
1864                    );
1865                }
1866                #[cfg(not(feature = "operations"))]
1867                {
1868                    return Err(RuntimeError::Other(
1869                        "--sys requires the operations feature".to_string(),
1870                    ));
1871                }
1872            }
1873
1874            #[cfg(feature = "quic")]
1875            for prepared_listener in prepared_quic {
1876                let listener_name = prepared_listener.name.clone();
1877                let listener_protocols: Arc<[ProtocolId]> = prepared_listener
1878                    .protocols
1879                    .iter()
1880                    .copied()
1881                    .filter(|protocol| !matches!(protocol, ProtocolId::Quic | ProtocolId::Http3))
1882                    .collect::<Vec<_>>()
1883                    .into();
1884                let routing = routing.clone();
1885                let state = state_ref.clone();
1886                let conn_tasks = connection_tasks.clone();
1887                let conn_cancel = connection_cancel.clone();
1888                let listener_cancel = listener_cancel.clone();
1889                let is_h3 = prepared_listener.protocols.contains(&ProtocolId::Http3);
1890                let auth = prepared_listener.auth.clone();
1891                let handshake_timeout_for_listener = prepared_listener.handshake_timeout;
1892                let connection_limit = prepared_listener.connection_limit;
1893
1894                #[cfg(feature = "ssh")]
1895                let listener_ssh_sessions = ssh_sessions.clone();
1896
1897                if is_h3 {
1898                    let listener = prepared_listener.listener.clone();
1899                    let tls_client_config_for_listener = tls_client_config.clone();
1900                    tasks.spawn(async move {
1901                        let active_streams = Arc::new(AtomicU64::new(0));
1902                        loop {
1903                            let connection = match listener.accept_connection(&listener_cancel).await {
1904                                Ok(Some(connection)) => connection,
1905                                Ok(None) => break,
1906                                Err(error) => {
1907                                    tracing::debug!(%error, listener = %listener_name, "H3 connection failed");
1908                                    continue;
1909                                }
1910                            };
1911                            let routing = routing.clone();
1912                            let state = state.clone();
1913                            let conn_tasks = conn_tasks.clone();
1914                            let conn_cancel = conn_cancel.child_token();
1915                            let listener_name = listener_name.clone();
1916                            let auth = auth.clone();
1917                            let authorization = match &auth {
1918                                eggress_server::accept::InboundAuthentication::None => None,
1919                                eggress_server::accept::InboundAuthentication::UsernamePassword { username, password }
1920                                | eggress_server::accept::InboundAuthentication::UsernamePasswordWithReuse { username, password, .. } => {
1921                                    Some((username.clone(), password.clone()))
1922                                }
1923                            };
1924                            let active_streams = active_streams.clone();
1925                            let protocols = listener_protocols.clone();
1926                            let connection_limit = connection_limit;
1927                            let tls_client_config_for_connection = tls_client_config_for_listener.clone();
1928                            conn_tasks.spawn(async move {
1929                                let result = eggress_protocol_h3::serve_connection(
1930                                    connection,
1931                                    conn_cancel.clone(),
1932                                    authorization,
1933                                    move |request, stream, peer| {
1934                                        let routing = routing.clone();
1935                                        let state = state.clone();
1936                                        let listener_name = listener_name.clone();
1937                                        let auth = auth.clone();
1938                                        let protocols = protocols.clone();
1939                                        let active_streams = active_streams.clone();
1940                                        let tls_client_config = tls_client_config_for_connection.clone();
1941                                        async move {
1942                                            let slot = match ListenerConnectionSlot::try_acquire(&active_streams, connection_limit) {
1943                                                Some(slot) => slot,
1944                                                None => return,
1945                                            };
1946                                            let target = match request.target() {
1947                                                Ok(target) => target,
1948                                                Err(error) => {
1949                                                    tracing::debug!(%error, "invalid H3 CONNECT authority");
1950                                                    drop(slot);
1951                                                    return;
1952                                                }
1953                                            };
1954                                            let generation = state.snapshot.load().generation;
1955                                            let config = eggress_server::ConnectionConfig {
1956                                                routing: routing as Arc<dyn RouteService>,
1957                                                context: eggress_server::ConnectionContext {
1958                                                    source: Some(peer),
1959                                                    listener: listener_name,
1960                                                    generation,
1961                                                },
1962                                                handshake_timeout: handshake_timeout_for_listener,
1963                                                connect_timeout,
1964                                                protocols,
1965                                                authentication: auth,
1966                                                metrics: Some(state.metrics.clone()),
1967                                                udp: None,
1968                                                tls_client_config,
1969                                                shadowsocks: None,
1970                                                #[cfg(feature = "extended")]
1971                                                shadowsocks_metrics: Some(state.shadowsocks_metrics.clone()),
1972                                                #[cfg(not(feature = "extended"))]
1973                                                shadowsocks_metrics: None,
1974                                                trojan: None,
1975                                                fixed_target: None,
1976                                                local_bind: None,
1977                                                #[cfg(feature = "ssh")]
1978                                                ssh_sessions: Some(listener_ssh_sessions.clone()),
1979                                            };
1980                                            let pending = eggress_server::accept::PendingTunnel {
1981                                                target: target.clone(),
1982                                                client: stream,
1983                                                protocol: eggress_server::accept::TunnelProtocol::Http3,
1984                                                reply_context: eggress_server::accept::ReplyContext::Http3,
1985                                                identity: eggress_core::ClientIdentity::Anonymous,
1986                                            };
1987                                            state.metrics.record_session_start();
1988                                            let report = eggress_server::execute::execute(
1989                                                eggress_server::accept::AcceptedSession::Tunnel(pending),
1990                                                &config,
1991                                            ).await;
1992                                            state.metrics.record_session(&report);
1993                                            drop(slot);
1994                                        }
1995                                    },
1996                                ).await;
1997                                if let Err(error) = result {
1998                                    tracing::debug!(%error, "H3 connection ended");
1999                                }
2000                            });
2001                        }
2002                    });
2003                } else {
2004                    let listener = prepared_listener.listener.clone();
2005                    let auth = prepared_listener.auth.clone();
2006                    let tls_client_config_for_listener = tls_client_config.clone();
2007                    tasks.spawn(async move {
2008                        let active_streams = Arc::new(AtomicU64::new(0));
2009                        let listener_name_for_handler = listener_name.clone();
2010                        let protocols_for_handler = listener_protocols.clone();
2011                        let routing_for_handler = routing.clone();
2012                        let state_for_handler = state.clone();
2013                        let auth_for_handler = auth.clone();
2014                        #[cfg(feature = "ssh")]
2015                        let listener_ssh_sessions_for_handler = Some(listener_ssh_sessions.clone());
2016                        let result = listener
2017                            .run(listener_cancel, move |stream, peer| {
2018                                let routing = routing_for_handler.clone();
2019                                let state = state_for_handler.clone();
2020                                let listener_name = listener_name_for_handler.clone();
2021                                let protocols = protocols_for_handler.clone();
2022                                let auth = auth_for_handler.clone();
2023                                let active_streams = active_streams.clone();
2024                                #[cfg(feature = "ssh")]
2025                                let ssh_sessions = listener_ssh_sessions_for_handler.clone();
2026                                let tls_client_config = tls_client_config_for_listener.clone();
2027                                async move {
2028                                    let Some(slot) = ListenerConnectionSlot::try_acquire(
2029                                        &active_streams,
2030                                        connection_limit,
2031                                    ) else {
2032                                        return;
2033                                    };
2034                                    let generation = state.snapshot.load().generation;
2035                                    let config = eggress_server::ConnectionConfig {
2036                                        routing: routing as Arc<dyn RouteService>,
2037                                        context: eggress_server::ConnectionContext {
2038                                            source: Some(peer),
2039                                            listener: listener_name,
2040                                            generation,
2041                                        },
2042                                        handshake_timeout: handshake_timeout_for_listener,
2043                                        connect_timeout,
2044                                        protocols,
2045                                        authentication: auth,
2046                                        metrics: Some(state.metrics.clone()),
2047                                        udp: None,
2048                                        tls_client_config,
2049                                        shadowsocks: None,
2050                                        #[cfg(feature = "extended")]
2051                                        shadowsocks_metrics: Some(
2052                                            state.shadowsocks_metrics.clone(),
2053                                        ),
2054                                        #[cfg(not(feature = "extended"))]
2055                                        shadowsocks_metrics: None,
2056                                        trojan: None,
2057                                        fixed_target: None,
2058                                        local_bind: None,
2059                                        #[cfg(feature = "ssh")]
2060                                        ssh_sessions,
2061                                    };
2062                                    let _ = eggress_server::serve_connection(stream, config).await;
2063                                    drop(slot);
2064                                }
2065                            })
2066                            .await;
2067                        if let Err(error) = result {
2068                            tracing::debug!(%error, "QUIC listener ended");
2069                        }
2070                    });
2071                }
2072            }
2073
2074            for prepared_listener in prepared {
2075                let routing = routing.clone();
2076                let state = state_ref.clone();
2077                let conn_tasks = connection_tasks.clone();
2078                let conn_cancel = connection_cancel.clone();
2079                let tls_client_config = tls_client_config.clone();
2080
2081                #[cfg(feature = "ssh")]
2082                let listener_ssh_sessions = ssh_sessions.clone();
2083                tasks.spawn(async move {
2084                    let proto_slice: Arc<[ProtocolId]> = prepared_listener.protocols.clone().into();
2085
2086                    loop {
2087                        let conn = match prepared_listener.listener.accept().await {
2088                            Ok(c) => c,
2089                            Err(e) => {
2090                                if e.to_string().contains("listener cancelled") {
2091                                    break;
2092                                }
2093                                tracing::error!("accept error: {e}");
2094                                continue;
2095                            }
2096                        };
2097
2098                        let routing = routing.clone();
2099                        let tls_client_config = tls_client_config.clone();
2100                        let peer = conn.peer_addr;
2101                        let listener_str = prepared_listener.name.clone();
2102                        let conn_id = state.connection_counter.fetch_add(1, Ordering::Relaxed);
2103                        let conn_protocols = proto_slice.clone();
2104                        let conn_auth = prepared_listener.auth.clone();
2105                        let conn_metrics = state.metrics.clone();
2106                        #[cfg(feature = "extended")]
2107                        let conn_ss_metrics = state.shadowsocks_metrics.clone();
2108                        let active = state.active_connections.clone();
2109                        let conn_cancel = conn_cancel.child_token();
2110                        let generation = state.snapshot.load().generation;
2111
2112                        active.fetch_add(1, Ordering::Relaxed);
2113
2114                        let tls_config = prepared_listener.tls.clone();
2115                        let ss_config = prepared_listener.shadowsocks.clone();
2116                        let trojan_config = prepared_listener.trojan.clone();
2117                        let fixed_target = prepared_listener.fixed_target.clone();
2118                        let local_bind = prepared_listener.local_bind.clone();
2119
2120                        let udp_svc = if let Some(ref udp_config) = prepared_listener.udp {
2121                            Some(Arc::new(RuntimeUdpService {
2122                                _listener_name: prepared_listener.name.clone(),
2123                                udp_config: udp_config.clone(),
2124                                registry: state.udp_registry.clone(),
2125                                metrics: state.metrics.clone(),
2126                                udp_metrics: state.udp_metrics.clone(),
2127                                routing: routing.clone(),
2128                                udp_tasks: state.udp_tasks.clone(),
2129                            })
2130                                as Arc<dyn eggress_server::UdpService>)
2131                        } else {
2132                            None
2133                        };
2134                        #[cfg(feature = "ssh")]
2135                        let conn_ssh_sessions = listener_ssh_sessions.clone();
2136                        conn_tasks.spawn(async move {
2137                            let started = std::time::Instant::now();
2138
2139                            // Apply TLS if configured for this listener
2140                            let stream: eggress_core::BoxStream =
2141                                if let Some(ref tls_cfg) = tls_config {
2142                                    let server_config =
2143                                        match eggress_transport_tls::TlsServerConfigBuilder::new()
2144                                            .with_certificate_pem(&tls_cfg.cert_pem)
2145                                            .and_then(|b| b.with_key_pem(&tls_cfg.key_pem))
2146                                            .and_then(|b| {
2147                                                let b = if tls_cfg.alpn.is_empty() {
2148                                                    b
2149                                                } else {
2150                                                    b.with_alpn(tls_cfg.alpn.clone())
2151                                                };
2152                                                b.build()
2153                                            }) {
2154                                            Ok(c) => c,
2155                                            Err(e) => {
2156                                                tracing::error!(%peer, "TLS config error: {e}");
2157                                                active.fetch_sub(1, Ordering::Relaxed);
2158                                                return;
2159                                            }
2160                                        };
2161                                    match eggress_transport_tls::tls_accept(
2162                                        Box::new(conn.stream),
2163                                        server_config,
2164                                    )
2165                                    .await
2166                                    {
2167                                        Ok(s) => s,
2168                                        Err(e) => {
2169                                            tracing::debug!(%peer, "TLS accept failed: {e}");
2170                                            active.fetch_sub(1, Ordering::Relaxed);
2171                                            return;
2172                                        }
2173                                    }
2174                                } else {
2175                                    Box::new(conn.stream)
2176                                };
2177
2178                            #[cfg(feature = "extended")]
2179                            let advanced_protocol = conn_protocols.first().copied();
2180                            #[cfg(feature = "extended")]
2181                            let advanced_is_single = conn_protocols.len() == 1;
2182                            #[cfg(feature = "extended")]
2183                            let advanced_fixed_target = fixed_target.clone();
2184                            let config = eggress_server::ConnectionConfig {
2185                                routing: routing as Arc<dyn RouteService>,
2186                                context: eggress_server::ConnectionContext {
2187                                    source: Some(peer),
2188                                    listener: listener_str,
2189                                    generation,
2190                                },
2191                                handshake_timeout: prepared_listener.handshake_timeout,
2192                                connect_timeout,
2193                                protocols: conn_protocols,
2194                                authentication: conn_auth,
2195                                metrics: Some(conn_metrics),
2196                                udp: udp_svc,
2197                                tls_client_config: tls_client_config.clone(),
2198                                shadowsocks: ss_config.map(|ss| {
2199                                    eggress_server::accept::InboundShadowsocksConfig {
2200                                        method: ss.method,
2201                                        password: ss.password,
2202                                        #[cfg(feature = "pproxy-legacy")]
2203                                        auth_prefix: ss.auth_prefix.map(String::into_bytes),
2204                                        #[cfg(feature = "pproxy-legacy")]
2205                                        plugins: ss.plugins,
2206                                    }
2207                                }),
2208                                #[cfg(feature = "extended")]
2209                                shadowsocks_metrics: Some(conn_ss_metrics),
2210                                #[cfg(not(feature = "extended"))]
2211                                shadowsocks_metrics: None,
2212                                trojan: trojan_config.map(|t| {
2213                                    eggress_server::accept::InboundTrojanConfig {
2214                                        password: t.password,
2215                                        fallback: t.fallback,
2216                                    }
2217                                }),
2218                                fixed_target,
2219                                local_bind,
2220                                #[cfg(feature = "ssh")]
2221                                ssh_sessions: Some(conn_ssh_sessions),
2222                            };
2223
2224                            #[cfg(feature = "extended")]
2225                            if matches!(
2226                                advanced_protocol,
2227                                Some(ProtocolId::Http2 | ProtocolId::WebSocket)
2228                            ) && advanced_is_single
2229                            {
2230                                let advanced_result = match advanced_protocol {
2231                                    Some(ProtocolId::Http2) => {
2232                                        eggress_server::advanced::serve_h2_connection(
2233                                            stream, config,
2234                                        )
2235                                        .await
2236                                    }
2237                                    Some(ProtocolId::WebSocket) => match advanced_fixed_target {
2238                                        Some(target) => {
2239                                            eggress_server::advanced::serve_websocket_connection(
2240                                                stream, config, target,
2241                                            )
2242                                            .await
2243                                        }
2244                                        None => Err("WebSocket listener requires a fixed target"
2245                                            .to_string()),
2246                                    },
2247                                    _ => unreachable!(),
2248                                };
2249                                if let Err(error) = advanced_result {
2250                                    tracing::debug!(%peer, %error, "advanced listener ended");
2251                                }
2252                                active.fetch_sub(1, Ordering::Relaxed);
2253                                return;
2254                            }
2255
2256                            let report = tokio::select! {
2257                                report = eggress_server::serve_connection(stream, config)
2258                                    .instrument(tracing::info_span!(
2259                                        "conn",
2260                                        id = conn_id,
2261                                        peer = %peer,
2262                                    )) => {
2263                                    report
2264                                }
2265                                _ = conn_cancel.cancelled() => {
2266                                    eggress_server::SessionReport::cancelled(
2267                                        None,
2268                                        None,
2269                                        String::new(),
2270                                    )
2271                                }
2272                            };
2273
2274                            active.fetch_sub(1, Ordering::Relaxed);
2275
2276                            if compatibility_options.debug && report.failure.is_some() {
2277                                tracing::error!(
2278                                    protocol = ?report.protocol,
2279                                    target = ?report.target,
2280                                    route = %report.route,
2281                                    outcome = ?report.outcome,
2282                                    failure = ?report.failure,
2283                                    "pproxy debug connection failure",
2284                                );
2285                            } else {
2286                                tracing::info!(
2287                                    protocol = ?report.protocol,
2288                                    target = ?report.target,
2289                                    route = %report.route,
2290                                    outcome = ?report.outcome,
2291                                    bytes_upstream = report.bytes_upstream,
2292                                    bytes_downstream = report.bytes_downstream,
2293                                    duration_ms = started.elapsed().as_millis() as u64,
2294                                    "connection completed",
2295                                );
2296                            }
2297                            if compatibility_options.verbose_level >= 1 {
2298                                tracing::info!(
2299                                    protocol = ?report.protocol,
2300                                    target = ?report.target,
2301                                    route = %report.route,
2302                                    "pproxy connection event",
2303                                );
2304                            }
2305                            if compatibility_options.verbose_level >= 2 {
2306                                tracing::info!(
2307                                    bytes_upstream = report.bytes_upstream,
2308                                    bytes_downstream = report.bytes_downstream,
2309                                    "pproxy traffic stats",
2310                                );
2311                            }
2312                        });
2313                    }
2314                });
2315            }
2316
2317            // Spawn reverse servers and clients
2318            #[cfg(feature = "reverse")]
2319            {
2320                let current_snapshot = snapshot.load();
2321                let reverse_servers = current_snapshot.reverse_servers.clone();
2322                let reverse_clients = current_snapshot.reverse_clients.clone();
2323                drop(current_snapshot);
2324
2325                for rs_cfg in reverse_servers {
2326                    if rs_cfg.pproxy_compat {
2327                        let server_config =
2328                            eggress_protocol_reverse::compat_pproxy::PproxyBackwardServerConfig {
2329                                control_bind: rs_cfg.control_bind,
2330                                external_bind: rs_cfg.external_bind,
2331                                auth: eggress_protocol_reverse::compat_pproxy::raw_auth(
2332                                    rs_cfg.auth_username.as_deref(),
2333                                    rs_cfg.auth_password.as_deref(),
2334                                ),
2335                                max_control_connections: rs_cfg.max_control_connections as usize,
2336                                max_pending_external: rs_cfg.max_pending_external as usize,
2337                                read_timeout_ms: rs_cfg.read_timeout_ms,
2338                                socks5_target: None,
2339                                client_framing:
2340                                    eggress_protocol_reverse::compat_pproxy::PproxyBackwardFraming::Raw,
2341                            };
2342                        let server =
2343                            eggress_protocol_reverse::compat_pproxy::PproxyBackwardServer::new(
2344                                server_config,
2345                            );
2346                        let server_cancel = server.cancel_token();
2347                        let cancel_clone = cancel.clone();
2348                        tasks.spawn(async move {
2349                            let result = tokio::select! {
2350                                r = server.run() => r,
2351                                _ = cancel_clone.cancelled() => {
2352                                    server_cancel.cancel();
2353                                    Ok(())
2354                                }
2355                            };
2356                            if let Err(e) = result {
2357                                tracing::error!(error = %e, "pproxy backward server error");
2358                            }
2359                        });
2360                        continue;
2361                    }
2362                    let server_config = eggress_protocol_reverse::server::ReverseServerConfig {
2363                        control_bind: rs_cfg.control_bind,
2364                        external_bind: Some(rs_cfg.external_bind),
2365                        auth_username: rs_cfg.auth_username,
2366                        auth_password: rs_cfg.auth_password,
2367                        max_control_connections: rs_cfg.max_control_connections,
2368                        read_timeout_ms: rs_cfg.read_timeout_ms,
2369                        allow_bind: rs_cfg.allow_bind,
2370                        max_listeners_per_client: rs_cfg.max_listeners_per_client,
2371                        max_streams_per_listener: rs_cfg.max_streams_per_listener,
2372                        max_pending_external: rs_cfg.max_pending_external,
2373                    };
2374                    // Defense-in-depth: validate the configuration before
2375                    // spawning the task so unsafe configurations fail at
2376                    // startup rather than silently at bind time.
2377                    if let Err(e) = server_config.validate() {
2378                        tracing::error!(
2379                            server_id = %rs_cfg.id,
2380                            error = %e,
2381                            "reverse server configuration validation failed; skipping",
2382                        );
2383                        continue;
2384                    }
2385                    let mut server =
2386                        eggress_protocol_reverse::server::ReverseServer::new(server_config);
2387                    server.set_metrics(state_ref.reverse_metrics.clone());
2388                    let server_state = server.state_handle();
2389                    let server_cancel = server.cancel_token();
2390
2391                    state_ref
2392                        .reverse_registry
2393                        .register(eggress_admin::ReverseServerEntry {
2394                            id: eggress_admin::ReverseServerId::from(rs_cfg.id.as_str()),
2395                            control_bind: rs_cfg.control_bind.to_string(),
2396                            state: server_state,
2397                        });
2398
2399                    let cancel_clone = cancel.clone();
2400                    tasks.spawn(async move {
2401                        let result = tokio::select! {
2402                            r = server.run() => r,
2403                            _ = cancel_clone.cancelled() => {
2404                                server_cancel.cancel();
2405                                Ok(())
2406                            }
2407                        };
2408                        if let Err(e) = result {
2409                            tracing::error!(error = %e, "reverse server error");
2410                        }
2411                    });
2412                }
2413
2414                for rc_cfg in reverse_clients {
2415                    let host = rc_cfg
2416                        .default_target_host
2417                        .clone()
2418                        .unwrap_or_else(|| "127.0.0.1".to_string());
2419                    let port = rc_cfg.default_target_port.unwrap_or(0);
2420
2421                    let parallel = rc_cfg.parallel_connections.max(1);
2422                    for conn_idx in 0..parallel {
2423                        if rc_cfg.pproxy_compat {
2424                            let client_config = eggress_protocol_reverse::compat_pproxy::PproxyBackwardClientConfig {
2425                                server_addr: rc_cfg.server_addr,
2426                                server_chain: rc_cfg.server_chain.clone(),
2427                                auth: eggress_protocol_reverse::compat_pproxy::raw_auth(
2428                                    rc_cfg.auth_username.as_deref(),
2429                                    rc_cfg.auth_password.as_deref(),
2430                                ),
2431                                reconnect_initial_ms: rc_cfg.reconnect_initial_ms,
2432                                reconnect_max_ms: rc_cfg.reconnect_max_ms,
2433                                read_timeout_ms: rc_cfg.read_timeout_ms,
2434                                target_connect_timeout_ms: 10_000,
2435                                server_framing:
2436                                    eggress_protocol_reverse::compat_pproxy::PproxyBackwardFraming::Raw,
2437                            };
2438                            let client =
2439                                eggress_protocol_reverse::compat_pproxy::PproxyBackwardClient::new(
2440                                    client_config,
2441                                    std::sync::Arc::new(
2442                                        crate::reverse::RouteEngineTargetResolver::new(
2443                                            routing.clone(),
2444                                            host.clone(),
2445                                            port,
2446                                            std::sync::Arc::from(rc_cfg.id.as_str()),
2447                                            Some(rc_cfg.server_addr),
2448                                        ),
2449                                    ),
2450                                );
2451                            let cancel_clone = cancel.clone();
2452                            let client_cancel = client.cancel_token();
2453                            let client_id = rc_cfg.id.clone();
2454                            let server_addr = rc_cfg.server_addr;
2455                            tasks.spawn(async move {
2456                                let result = tokio::select! {
2457                                    r = client.run() => r,
2458                                    _ = cancel_clone.cancelled() => {
2459                                        client_cancel.cancel();
2460                                        Ok(())
2461                                    }
2462                                };
2463                                if let Err(e) = result {
2464                                    tracing::error!(error = %e, client_id = %client_id, server = %server_addr, conn = conn_idx, "pproxy backward client error");
2465                                }
2466                            });
2467                            continue;
2468                        }
2469                        let client_config = eggress_protocol_reverse::client::ReverseClientConfig {
2470                            server_addr: rc_cfg.server_addr,
2471                            auth_username: rc_cfg.auth_username.clone(),
2472                            auth_password: rc_cfg.auth_password.clone(),
2473                            reconnect_initial_ms: rc_cfg.reconnect_initial_ms,
2474                            reconnect_max_ms: rc_cfg.reconnect_max_ms,
2475                            default_target_host: rc_cfg.default_target_host.clone(),
2476                            default_target_port: rc_cfg.default_target_port,
2477                            read_timeout_ms: rc_cfg.read_timeout_ms,
2478                            drain_grace_ms: rc_cfg.drain_grace_ms,
2479                            ..Default::default()
2480                        };
2481                        let mut client =
2482                            eggress_protocol_reverse::client::ReverseClient::new(client_config);
2483                        client.set_metrics(state_ref.reverse_metrics.clone());
2484
2485                        let resolver = crate::reverse::RouteEngineTargetResolver::new(
2486                            routing.clone(),
2487                            host.clone(),
2488                            port,
2489                            std::sync::Arc::from(rc_cfg.id.as_str()),
2490                            Some(rc_cfg.server_addr),
2491                        );
2492                        client.set_resolver(std::sync::Arc::new(resolver));
2493
2494                        let cancel_clone = cancel.clone();
2495                        let client_cancel = client.cancel_token();
2496                        let client_id = rc_cfg.id.clone();
2497                        let server_addr = rc_cfg.server_addr;
2498
2499                        tasks.spawn(async move {
2500                            let result = tokio::select! {
2501                                r = client.run() => r,
2502                                _ = cancel_clone.cancelled() => {
2503                                    client_cancel.cancel();
2504                                    Ok(())
2505                                }
2506                            };
2507                            if let Err(e) = result {
2508                                tracing::error!(error = %e, client_id = %client_id, server = %server_addr, conn = conn_idx, "reverse client error");
2509                            }
2510                        });
2511                    }
2512                }
2513            }
2514
2515            // Pre-bind admin listener before marking readiness so bind failures
2516            // are surfaced as startup errors rather than silent background failures.
2517            #[cfg(feature = "operations")]
2518            let pre_bound_admin = if let Some(ref admin_cfg) = admin_config {
2519                if admin_cfg.enabled {
2520                    let bind = admin_cfg.bind.clone();
2521                    let admin_cancel_token = admin_cancel.clone();
2522                    let server =
2523                        match eggress_admin::AdminServer::new(&bind, admin_cancel_token).await {
2524                            Ok(s) => Some(s),
2525                            Err(e) => {
2526                                return Err(RuntimeError::ListenerBind {
2527                                    addr: bind,
2528                                    source: std::io::Error::new(
2529                                        std::io::ErrorKind::AddrInUse,
2530                                        e.to_string(),
2531                                    ),
2532                                });
2533                            }
2534                        };
2535                    server
2536                } else {
2537                    None
2538                }
2539            } else {
2540                None
2541            };
2542
2543            #[cfg(feature = "operations")]
2544            if let (Some(server), Some(admin_cfg)) = (pre_bound_admin, admin_config.as_ref()) {
2545                let metrics_enabled = admin_cfg.metrics;
2546                let state_ref = state_ref.clone();
2547                let provider: Arc<dyn AdminSnapshotProvider> = listener_infos_provider.clone();
2548                if let Ok(addr) = server.local_addr() {
2549                    *state_ref
2550                        .admin_local_addr
2551                        .lock()
2552                        .unwrap_or_else(|e| e.into_inner()) = Some(addr);
2553                }
2554                let admin_auth = admin_cfg.auth.clone();
2555                admin_tasks.spawn(async move {
2556                    let admin_state = eggress_admin::AdminState {
2557                        metrics: metrics_registry.clone(),
2558                        start_time: state_ref.start_time,
2559                        readiness: state_ref.readiness.clone(),
2560                        active_connections: Some(state_ref.active_connections.clone()),
2561                        provider,
2562                        udp_registry: state_ref.udp_registry.clone(),
2563                        #[cfg(feature = "reverse")]
2564                        reverse_registry: state_ref.reverse_registry.clone(),
2565                        #[cfg(not(feature = "reverse"))]
2566                        reverse_registry: std::sync::Arc::new(eggress_admin::ReverseRegistry::new()),
2567                        metrics_enabled,
2568                        auth: admin_auth,
2569                    };
2570                    if let Err(e) = server.run(admin_state).await {
2571                        tracing::error!("admin server error: {e}");
2572                    }
2573                });
2574            }
2575
2576            readiness.store(true, Ordering::Release);
2577
2578            #[cfg(unix)]
2579            {
2580                let mut sigterm =
2581                    tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate());
2582                let mut sighup =
2583                    tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup());
2584
2585                if let Err(ref e) = sigterm {
2586                    tracing::warn!("failed to register SIGTERM handler: {e}");
2587                }
2588                if let Err(ref e) = sighup {
2589                    tracing::warn!("failed to register SIGHUP handler: {e}");
2590                }
2591
2592                loop {
2593                    tokio::select! {
2594                        _ = cancel.cancelled() => {
2595                            tracing::info!("shutdown requested via cancel token");
2596                            break;
2597                        }
2598                        _ = tokio::signal::ctrl_c() => {
2599                            tracing::info!("shutdown signal received");
2600                            break;
2601                        }
2602                        _ = async { sigterm.as_mut().ok()?.recv().await }, if sigterm.is_ok() => {
2603                            tracing::info!("shutdown signal received");
2604                            break;
2605                        }
2606                        _ = async { sighup.as_mut().ok()?.recv().await }, if sighup.is_ok() && !config_path.is_empty() => {
2607                            tracing::info!("reload signal received, reloading config from {config_path}");
2608                            let prev_snapshot = snapshot.load();
2609                            let prev_ref: Option<&CompiledRuntimeSnapshot> = Some(&prev_snapshot);
2610                            let config_path_clone = config_path.clone();
2611                            let load_result = tokio::task::spawn_blocking(move || {
2612                                eggress_config::compile::load_and_compile(&config_path_clone)
2613                            }).await;
2614                            match load_result {
2615                                Ok(Ok(new_rt_config)) => {
2616                                    // Classify unsupported changes: reject if listener topology changed
2617                                    let old_listeners = &snapshot.load().listeners;
2618                                    if let Err(reason) = classify_listeners(old_listeners, &new_rt_config.listeners) {
2619                                        tracing::error!("reload rejected: {reason}");
2620                                        metrics.record_reload(false);
2621                                        continue;
2622                                    }
2623                                    match compile_runtime_snapshot(&new_rt_config, prev_ref) {
2624                                        Ok(new_snapshot) => {
2625                                            let upstream_count = new_snapshot.upstreams.len();
2626                                            let gen = new_snapshot.generation;
2627
2628                                            // Snapshot must be published before the router swap so any reader
2629                                            // observing the new generation via
2630                                            // `snapshot.load()` also sees the
2631                                            // matching router.
2632                                            let new_snapshot = Arc::new(new_snapshot);
2633                                            snapshot.store(new_snapshot.clone());
2634                                            routing.swap_arc(new_snapshot.router.clone());
2635
2636                                            metrics.set_config_generation(gen);
2637                                            metrics.record_reload(true);
2638
2639                                            if let Ok(mut guard) = health_clone.lock() {
2640                                                if let Some(ref mut hm) = *guard {
2641                                                    hm.stop_all();
2642                                                }
2643                                                let upstream_runtimes: Vec<Arc<UpstreamRuntime>> = snapshot
2644                                                    .load()
2645                                                    .upstreams
2646                                                    .values()
2647                                                    .cloned()
2648                                                    .collect();
2649                                                if !upstream_runtimes.is_empty() {
2650                                                    let mut hm = HealthManager::new(health_cancel.clone());
2651                                                    hm.start_probes(&upstream_runtimes);
2652                                                    *guard = Some(hm);
2653                                                } else {
2654                                                    *guard = None;
2655                                                }
2656                                            }
2657
2658                                            tracing::info!(
2659                                                generation = gen,
2660                                                upstreams = upstream_count,
2661                                                "config reloaded successfully"
2662                                            );
2663                                        }
2664                                        Err(e) => {
2665                                            metrics.record_reload(false);
2666                                            tracing::error!("reload failed (snapshot build): {e}");
2667                                        }
2668                                    }
2669                                }
2670                                Ok(Err(e)) => {
2671                                    metrics.record_reload(false);
2672                                    tracing::error!("reload failed (config load): {e}");
2673                                }
2674                                Err(join_err) => {
2675                                    metrics.record_reload(false);
2676                                    tracing::error!("reload task panicked: {join_err}");
2677                                }
2678                            }
2679                        }
2680                    }
2681                }
2682            }
2683
2684            #[cfg(not(unix))]
2685            {
2686                tokio::select! {
2687                    _ = cancel.cancelled() => {
2688                        tracing::info!("shutdown requested via cancel token");
2689                    }
2690                    _ = tokio::signal::ctrl_c() => {
2691                        tracing::info!("shutdown signal received");
2692                    }
2693                }
2694            }
2695
2696            // 1. Set readiness false (admin /-/ready will report 503 during drain)
2697            readiness.store(false, Ordering::Release);
2698
2699            // 2. Stop listeners (no new connections accepted)
2700            listener_cancel.cancel();
2701
2702            // 3. Stop health probes
2703            health_cancel.cancel();
2704
2705            // 4. Close all UDP associations
2706            state_ref.udp_registry.close_all().await;
2707
2708            // 5. Wait for UDP relay tasks to complete
2709            state_ref.udp_tasks.close();
2710            let _ = tokio::time::timeout(shutdown_grace, state_ref.udp_tasks.wait()).await;
2711
2712            // 6. Wait for listener accept loops to exit so they cannot hand
2713            //    new connections to the connection tracker.
2714            tasks.close();
2715            tasks.wait().await;
2716
2717            // 7. Drain active connections within the grace period; force-cancel
2718            //    afterwards. Admin stays up through this window so operators
2719            //    can observe drain progress via /-/ready, /-/status, /metrics.
2720            tracing::info!("draining active connections");
2721
2722            let deadline = tokio::time::Instant::now() + shutdown_grace;
2723            loop {
2724                let active = active_connections.load(Ordering::Relaxed);
2725                if active == 0 {
2726                    tracing::info!("all connections drained");
2727                    break;
2728                }
2729                if tokio::time::Instant::now() >= deadline {
2730                    tracing::warn!(active, "drain timeout reached, forcing shutdown");
2731                    connection_cancel.cancel();
2732                    break;
2733                }
2734                tokio::time::sleep(Duration::from_millis(100)).await;
2735            }
2736
2737            // 8. Wait for connection tasks (either drained naturally or force-cancelled)
2738            connection_tasks.close();
2739            connection_tasks.wait().await;
2740
2741            #[cfg(feature = "ssh")]
2742            ssh_sessions.shutdown().await;
2743
2744            // 9. Now that the proxy has fully stopped accepting and serving
2745            //    traffic, stop the admin server. /-/ready has been reporting
2746            //    503 since step 1.
2747            admin_cancel.cancel();
2748            admin_tasks.close();
2749            admin_tasks.wait().await;
2750
2751            #[cfg(feature = "operations")]
2752            if let Some(mut proxy) = compatibility_system_proxy {
2753                proxy.restore().map_err(RuntimeError::Other)?;
2754            }
2755
2756            Ok::<_, RuntimeError>(())
2757        };
2758
2759        let result = if tokio::runtime::Handle::try_current().is_err() {
2760            // Caller is not inside a tokio runtime; create one and block on it.
2761            let rt = tokio::runtime::Runtime::new()?;
2762            rt.block_on(run_async)
2763        } else {
2764            // Caller is already inside a tokio runtime. Driving a long-lived
2765            // supervisor body via Handle::current().block_on from a worker
2766            // thread would panic or deadlock. Run on a dedicated OS thread
2767            // with its own runtime.
2768            std::thread::Builder::new()
2769                .name("eggress-supervisor".to_string())
2770                .spawn(move || -> Result<(), RuntimeError> {
2771                    let rt = tokio::runtime::Builder::new_multi_thread()
2772                        .enable_all()
2773                        .build()
2774                        .map_err(RuntimeError::RuntimeInit)?;
2775                    rt.block_on(run_async)
2776                })
2777                .map_err(RuntimeError::RuntimeInit)?
2778                .join()
2779                .map_err(|payload| {
2780                    let message = payload
2781                        .downcast_ref::<String>()
2782                        .map(String::as_str)
2783                        .or_else(|| payload.downcast_ref::<&'static str>().copied())
2784                        .unwrap_or("unknown panic payload");
2785                    RuntimeError::Other(format!("supervisor thread panicked: {message}"))
2786                })?
2787        };
2788
2789        self.health = std::sync::Arc::try_unwrap(health)
2790            .ok()
2791            .and_then(|m| m.into_inner().ok())
2792            .flatten();
2793
2794        match &result {
2795            Ok(()) => tracing::info!("eggress stopped"),
2796            Err(e) => tracing::error!(error = %e, "eggress stopped with error"),
2797        }
2798
2799        result
2800    }
2801}
2802
2803#[cfg(test)]
2804mod tests {
2805    use super::*;
2806    use std::io::Write;
2807    use tempfile::NamedTempFile;
2808
2809    use crate::snapshot::compile_runtime_snapshot;
2810    use eggress_config::compile::{GroupFallback, ProcessConfig, RuntimeConfig, TimeoutConfig};
2811    use eggress_routing::scheduler::SchedulerKind;
2812    use eggress_routing::{MatchExpr, RouteActionSpec, RuleId, UpstreamGroupId};
2813
2814    fn write_config(content: &str) -> NamedTempFile {
2815        let mut f = NamedTempFile::new().unwrap();
2816        f.write_all(content.as_bytes()).unwrap();
2817        f.flush().unwrap();
2818        f
2819    }
2820
2821    #[test]
2822    fn build_router_direct_only() {
2823        let rt_config = RuntimeConfig {
2824            process: ProcessConfig::default(),
2825            timeouts: TimeoutConfig::default(),
2826            listeners: vec![],
2827            upstreams: vec![],
2828            groups: vec![],
2829            rules: vec![],
2830            default_action: RouteActionSpec::Direct,
2831            admin: None,
2832            reverse_servers: vec![],
2833            reverse_clients: vec![],
2834        };
2835        let snap = compile_runtime_snapshot(&rt_config, None).unwrap();
2836        assert!(snap.router.rules().is_empty());
2837    }
2838
2839    #[test]
2840    fn build_router_with_group_references_unknown_upstream() {
2841        let rt_config = RuntimeConfig {
2842            process: ProcessConfig::default(),
2843            timeouts: TimeoutConfig::default(),
2844            listeners: vec![],
2845            upstreams: vec![],
2846            groups: vec![eggress_config::compile::UpstreamGroupConfig {
2847                id: UpstreamGroupId(Arc::from("main")),
2848                scheduler: SchedulerKind::RoundRobin,
2849                members: vec!["nonexistent".to_string()],
2850                fallback: GroupFallback::Reject,
2851            }],
2852            rules: vec![],
2853            default_action: RouteActionSpec::Direct,
2854            admin: None,
2855            reverse_servers: vec![],
2856            reverse_clients: vec![],
2857        };
2858        let result = compile_runtime_snapshot(&rt_config, None);
2859        assert!(result.is_err(), "expected error, got Ok");
2860        let err_msg = result.err().unwrap().to_string();
2861        assert!(err_msg.contains("nonexistent"));
2862    }
2863
2864    #[test]
2865    fn build_router_with_valid_group() {
2866        let rt_config = RuntimeConfig {
2867            process: ProcessConfig::default(),
2868            timeouts: TimeoutConfig::default(),
2869            listeners: vec![],
2870            upstreams: vec![eggress_config::compile::UpstreamConfig {
2871                id: "proxy1".to_string(),
2872                chain: eggress_uri::ProxyChainSpec { hops: vec![] },
2873                health: eggress_routing::health::HealthConfig::default(),
2874                h2: None,
2875            }],
2876            groups: vec![eggress_config::compile::UpstreamGroupConfig {
2877                id: UpstreamGroupId(Arc::from("main")),
2878                scheduler: SchedulerKind::RoundRobin,
2879                members: vec!["proxy1".to_string()],
2880                fallback: GroupFallback::Reject,
2881            }],
2882            rules: vec![],
2883            default_action: RouteActionSpec::Direct,
2884            admin: None,
2885            reverse_servers: vec![],
2886            reverse_clients: vec![],
2887        };
2888        let snap = compile_runtime_snapshot(&rt_config, None).unwrap();
2889        assert!(snap.router.rules().is_empty());
2890    }
2891
2892    #[test]
2893    fn build_router_rule_references_unknown_group() {
2894        let rt_config = RuntimeConfig {
2895            process: ProcessConfig::default(),
2896            timeouts: TimeoutConfig::default(),
2897            listeners: vec![],
2898            upstreams: vec![],
2899            groups: vec![],
2900            rules: vec![eggress_routing::CompiledRule {
2901                id: RuleId(Arc::from("r1")),
2902                matcher: MatchExpr::Any,
2903                action: RouteActionSpec::UpstreamGroup(UpstreamGroupId(Arc::from("missing"))),
2904            }],
2905            default_action: RouteActionSpec::Direct,
2906            admin: None,
2907            reverse_servers: vec![],
2908            reverse_clients: vec![],
2909        };
2910        let result = compile_runtime_snapshot(&rt_config, None);
2911        assert!(result.is_err(), "expected error, got Ok");
2912        let err_msg = result.err().unwrap().to_string();
2913        assert!(err_msg.contains("missing"));
2914    }
2915
2916    #[tokio::test]
2917    async fn load_config_start_supervisor() {
2918        let config = r#"
2919version = 1
2920
2921[[listeners]]
2922name = "test"
2923bind = "127.0.0.1:0"
2924protocols = ["http"]
2925"#;
2926        let f = write_config(config);
2927        let path = f.path().to_str().unwrap();
2928        let result = ServiceSupervisor::start(path);
2929        assert!(result.is_ok(), "expected Ok, got {:?}", result.err());
2930    }
2931
2932    #[test]
2933    fn active_connections_counter_increments_and_decrements() {
2934        let active = Arc::new(AtomicU64::new(0));
2935        assert_eq!(active.load(Ordering::Relaxed), 0);
2936        active.fetch_add(1, Ordering::Relaxed);
2937        assert_eq!(active.load(Ordering::Relaxed), 1);
2938        active.fetch_add(1, Ordering::Relaxed);
2939        assert_eq!(active.load(Ordering::Relaxed), 2);
2940        active.fetch_sub(1, Ordering::Relaxed);
2941        assert_eq!(active.load(Ordering::Relaxed), 1);
2942        active.fetch_sub(1, Ordering::Relaxed);
2943        assert_eq!(active.load(Ordering::Relaxed), 0);
2944    }
2945
2946    #[test]
2947    fn readiness_flag_controls_ready_endpoint() {
2948        let readiness = Arc::new(AtomicBool::new(true));
2949        assert!(readiness.load(Ordering::Relaxed));
2950        readiness.store(false, Ordering::Relaxed);
2951        assert!(!readiness.load(Ordering::Relaxed));
2952        readiness.store(true, Ordering::Relaxed);
2953        assert!(readiness.load(Ordering::Relaxed));
2954    }
2955
2956    #[test]
2957    fn reload_rejects_listener_name_change() {
2958        let config1 = r#"
2959version = 1
2960
2961[[listeners]]
2962name = "http-in"
2963bind = "127.0.0.1:8080"
2964protocols = ["http"]
2965"#;
2966        let config2 = r#"
2967version = 1
2968
2969[[listeners]]
2970name = "http-changed"
2971bind = "127.0.0.1:8080"
2972protocols = ["http"]
2973"#;
2974        let f1 = write_config(config1);
2975        let f2 = write_config(config2);
2976        let path1 = f1.path().to_str().unwrap();
2977        let path2 = f2.path().to_str().unwrap();
2978
2979        let sup = ServiceSupervisor::start(path1).unwrap();
2980        let new_config = eggress_config::compile::load_and_compile(path2).unwrap();
2981        let result = sup.classify_reload(&new_config);
2982        assert!(result.is_err(), "listener name change should be rejected");
2983        assert!(result.unwrap_err().contains("name changed"));
2984    }
2985
2986    #[test]
2987    fn reload_rejects_listener_bind_change() {
2988        let config1 = r#"
2989version = 1
2990
2991[[listeners]]
2992name = "http-in"
2993bind = "127.0.0.1:8080"
2994protocols = ["http"]
2995"#;
2996        let config2 = r#"
2997version = 1
2998
2999[[listeners]]
3000name = "http-in"
3001bind = "127.0.0.1:9090"
3002protocols = ["http"]
3003"#;
3004        let f1 = write_config(config1);
3005        let f2 = write_config(config2);
3006        let path1 = f1.path().to_str().unwrap();
3007        let path2 = f2.path().to_str().unwrap();
3008
3009        let sup = ServiceSupervisor::start(path1).unwrap();
3010        let new_config = eggress_config::compile::load_and_compile(path2).unwrap();
3011        let result = sup.classify_reload(&new_config);
3012        assert!(result.is_err(), "listener bind change should be rejected");
3013        assert!(result.unwrap_err().contains("bind"));
3014    }
3015
3016    #[test]
3017    fn reload_accepts_unchanged_listeners() {
3018        let config = r#"
3019version = 1
3020
3021[[listeners]]
3022name = "http-in"
3023bind = "127.0.0.1:8080"
3024protocols = ["http"]
3025"#;
3026        let f = write_config(config);
3027        let path = f.path().to_str().unwrap();
3028
3029        let sup = ServiceSupervisor::start(path).unwrap();
3030        let new_config = eggress_config::compile::load_and_compile(path).unwrap();
3031        let result = sup.classify_reload(&new_config);
3032        assert!(result.is_ok(), "unchanged listeners should be accepted");
3033    }
3034
3035    #[test]
3036    fn reload_rejects_listener_count_change() {
3037        let config1 = r#"
3038version = 1
3039
3040[[listeners]]
3041name = "http-in"
3042bind = "127.0.0.1:8080"
3043protocols = ["http"]
3044"#;
3045        let config2 = r#"
3046version = 1
3047
3048[[listeners]]
3049name = "http-in"
3050bind = "127.0.0.1:8080"
3051protocols = ["http"]
3052
3053[[listeners]]
3054name = "socks-in"
3055bind = "127.0.0.1:1080"
3056protocols = ["socks5"]
3057"#;
3058        let f1 = write_config(config1);
3059        let f2 = write_config(config2);
3060        let path1 = f1.path().to_str().unwrap();
3061        let path2 = f2.path().to_str().unwrap();
3062
3063        let sup = ServiceSupervisor::start(path1).unwrap();
3064        let new_config = eggress_config::compile::load_and_compile(path2).unwrap();
3065        let result = sup.classify_reload(&new_config);
3066        assert!(result.is_err(), "listener count change should be rejected");
3067        assert!(result.unwrap_err().contains("listener count"));
3068    }
3069
3070    #[test]
3071    fn reload_rejects_transparent_enabled_change() {
3072        let config1 = r#"
3073version = 1
3074
3075[[listeners]]
3076name = "http-in"
3077bind = "127.0.0.1:8080"
3078protocols = ["http"]
3079"#;
3080        let config2 = r#"
3081version = 1
3082
3083[[listeners]]
3084name = "http-in"
3085bind = "127.0.0.1:8080"
3086protocols = ["http"]
3087
3088[listeners.transparent]
3089enabled = true
3090"#;
3091        let f1 = write_config(config1);
3092        let f2 = write_config(config2);
3093        let path1 = f1.path().to_str().unwrap();
3094        let path2 = f2.path().to_str().unwrap();
3095
3096        let sup = ServiceSupervisor::start(path1).unwrap();
3097        let new_config = eggress_config::compile::load_and_compile(path2).unwrap();
3098        let result = sup.classify_reload(&new_config);
3099        assert!(
3100            result.is_err(),
3101            "transparent enabled change should be rejected"
3102        );
3103        assert!(result.unwrap_err().contains("transparent"));
3104    }
3105
3106    #[test]
3107    fn reload_rejects_unix_path_change() {
3108        let config1 = r#"
3109version = 1
3110
3111[[listeners]]
3112name = "http-in"
3113bind = "127.0.0.1:8080"
3114protocols = ["http"]
3115
3116[listeners.unix]
3117path = "/tmp/eggress.sock"
3118"#;
3119        let config2 = r#"
3120version = 1
3121
3122[[listeners]]
3123name = "http-in"
3124bind = "127.0.0.1:8080"
3125protocols = ["http"]
3126
3127[listeners.unix]
3128path = "/tmp/eggress-new.sock"
3129"#;
3130        let f1 = write_config(config1);
3131        let f2 = write_config(config2);
3132        let path1 = f1.path().to_str().unwrap();
3133        let path2 = f2.path().to_str().unwrap();
3134
3135        let sup = ServiceSupervisor::start(path1).unwrap();
3136        let new_config = eggress_config::compile::load_and_compile(path2).unwrap();
3137        let result = sup.classify_reload(&new_config);
3138        assert!(result.is_err(), "unix path change should be rejected");
3139        assert!(result.unwrap_err().contains("unix socket path"));
3140    }
3141
3142    #[test]
3143    fn compute_advertise_explicit() {
3144        let result = compute_advertise_ip(
3145            Some("10.0.0.1".parse().unwrap()),
3146            "0.0.0.0".parse().unwrap(),
3147            "127.0.0.1:5000".parse().unwrap(),
3148        );
3149        assert_eq!(
3150            result.unwrap(),
3151            std::net::IpAddr::V4("10.0.0.1".parse().unwrap())
3152        );
3153    }
3154
3155    #[test]
3156    fn compute_advertise_bind_ip() {
3157        let result = compute_advertise_ip(
3158            None,
3159            "192.168.1.1".parse().unwrap(),
3160            "127.0.0.1:5000".parse().unwrap(),
3161        );
3162        assert_eq!(
3163            result.unwrap(),
3164            std::net::IpAddr::V4("192.168.1.1".parse().unwrap())
3165        );
3166    }
3167
3168    #[test]
3169    fn compute_advertise_loopback_fallback() {
3170        let result = compute_advertise_ip(
3171            None,
3172            "0.0.0.0".parse().unwrap(),
3173            "127.0.0.1:5000".parse().unwrap(),
3174        );
3175        assert_eq!(
3176            result.unwrap(),
3177            std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)
3178        );
3179    }
3180
3181    #[test]
3182    fn compute_advertise_unspecified_non_loopback_rejected() {
3183        let result = compute_advertise_ip(
3184            None,
3185            "0.0.0.0".parse().unwrap(),
3186            "192.168.1.10:5000".parse().unwrap(),
3187        );
3188        assert!(
3189            result.is_err(),
3190            "non-loopback with unspecified bind should fail"
3191        );
3192    }
3193
3194    #[test]
3195    fn compute_advertise_ipv6_loopback() {
3196        let result =
3197            compute_advertise_ip(None, "::".parse().unwrap(), "[::1]:5000".parse().unwrap());
3198        assert_eq!(
3199            result.unwrap(),
3200            std::net::IpAddr::V6(std::net::Ipv6Addr::LOCALHOST)
3201        );
3202    }
3203
3204    #[test]
3205    fn compute_advertise_explicit_overrides_bind() {
3206        let result = compute_advertise_ip(
3207            Some("10.0.0.1".parse().unwrap()),
3208            "192.168.1.1".parse().unwrap(),
3209            "127.0.0.1:5000".parse().unwrap(),
3210        );
3211        assert_eq!(
3212            result.unwrap(),
3213            std::net::IpAddr::V4("10.0.0.1".parse().unwrap())
3214        );
3215    }
3216}
eggress_runtime/supervisor.rs

eggress_runtime/
supervisor.rs
