singbox_driver/
readiness.rs

1//! Loopback-only readiness and live version observation for sing-box.
2
3use crate::api::clash_api::{SingBoxClashApiClient, SingBoxCoreVersionProvider};
4use crate::api::controller_configuration::SingBoxControllerConfiguration;
5use crate::api::http::{ReqwestSingBoxClashHttpPort, SingBoxClashHttpPort};
6use async_trait::async_trait;
7use lurker_core_contract::identity::CoreVersion;
8use lurker_core_ports::core_process::CoreReadiness;
9use lurker_core_ports::error::PortError;
10use std::sync::Arc;
11use std::time::Duration;
12use tokio::time::{Instant, sleep};
13
14const RETRY_INTERVAL: Duration = Duration::from_millis(100);
15
16enum ReadinessWaitState {
17    Probe,
18    Retry(PortError),
19    Ready,
20}
21
22/// The loopback readiness probe of the sing-box Clash API.
23pub struct SingBoxControllerReadiness {
24    api: SingBoxClashApiClient,
25    version_provider: SingBoxCoreVersionProvider,
26}
27
28impl SingBoxControllerReadiness {
29    /// Creates the readiness probe with the production HTTP boundary.
30    pub fn with_reqwest(controller: SingBoxControllerConfiguration) -> Result<Self, PortError> {
31        Self::new(controller, Arc::new(ReqwestSingBoxClashHttpPort))
32    }
33
34    /// Creates the readiness probe with an injected HTTP boundary.
35    pub fn new(
36        controller: SingBoxControllerConfiguration,
37        http: Arc<dyn SingBoxClashHttpPort>,
38    ) -> Result<Self, PortError> {
39        let api = SingBoxClashApiClient::new(controller, http)?;
40        let version_provider = SingBoxCoreVersionProvider::new(api.clone());
41        Ok(Self {
42            api,
43            version_provider,
44        })
45    }
46
47    /// Waits until the controller answers, or the timeout elapses.
48    ///
49    /// The last failure is returned when the bound elapses; the probe never
50    /// reports readiness it did not observe.
51    pub async fn wait_until_ready(&self, timeout: Duration) -> Result<(), PortError> {
52        let deadline = Instant::now() + timeout;
53        let mut state = ReadinessWaitState::Probe;
54        loop {
55            state = match state {
56                ReadinessWaitState::Probe => match self.probe().await {
57                    Ok(_) => ReadinessWaitState::Ready,
58                    Err(error) => ReadinessWaitState::Retry(error),
59                },
60                ReadinessWaitState::Retry(error) => {
61                    let remaining = deadline.saturating_duration_since(Instant::now());
62                    if remaining.is_zero() {
63                        return Err(error);
64                    }
65                    sleep(remaining.min(RETRY_INTERVAL)).await;
66                    ReadinessWaitState::Probe
67                }
68                ReadinessWaitState::Ready => return Ok(()),
69            };
70        }
71    }
72}
73
74#[async_trait]
75impl CoreReadiness for SingBoxControllerReadiness {
76    async fn probe(&self) -> Result<String, PortError> {
77        let _ = self.version_provider.observe().await?;
78        Ok(self
79            .api
80            .endpoint()
81            .as_str()
82            .trim_end_matches('/')
83            .to_owned())
84    }
85
86    async fn active_core_version(&self) -> Result<Option<CoreVersion>, PortError> {
87        self.version_provider.observe().await.map(Some)
88    }
89}
singbox_driver/readiness.rs

singbox_driver/
readiness.rs
