microsandbox_control_client/
connection.rs

1//! Automatic format discovery outside the generic framed router.
2
3use std::path::Path;
4use std::sync::Arc;
5
6use microsandbox_protocol::control::DEFAULT_REQUEST_TIMEOUT;
7use microsandbox_protocol_client::{
8    ClientError, ConnectOptions, Connector, Delivery, ErrorKind, LocalConnector, Message, Protocol,
9    RequestOptions,
10};
11use tokio::time::timeout_at;
12use tokio_util::sync::CancellationToken;
13
14use crate::{
15    CheckedControlRequest, ControlClient, ControlClientError, ControlClientResult, ControlProtocol,
16    IntoControlMessage, JsonControlClient, JsonReply, VerifiedControlConnector,
17    dialer::Dialer,
18    json_client::{check_deadline, deadline},
19};
20
21//--------------------------------------------------------------------------------------------------
22// Types
23//--------------------------------------------------------------------------------------------------
24
25/// Operation format selected during this connection's setup.
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub enum ControlMode {
28    /// Persistent, negotiated CBOR frames.
29    Framed,
30    /// Legacy JSON unary exchanges.
31    Json,
32}
33
34/// The actual reply format. JSON never receives synthetic IDs or CBOR bytes.
35#[derive(Debug)]
36pub enum ControlReply {
37    /// Original framed response, including unknown fields.
38    Framed(Message),
39    /// Original JSON line and its lossless inspection view.
40    Json(JsonReply),
41}
42
43/// Shared discovery result and operation adapter. Standalone JSON connections
44/// rediscover before fresh exchanges unless a verified runtime owner is supplied.
45#[derive(Clone)]
46pub struct ControlConnection {
47    inner: Arc<Inner>,
48}
49
50struct Inner {
51    selected: Selected,
52    dialer: Dialer,
53    options: ConnectOptions,
54    closed: CancellationToken,
55    capabilities: crate::Capabilities,
56}
57
58enum Selected {
59    Framed(ControlClient),
60    Json(JsonControlClient),
61}
62
63//--------------------------------------------------------------------------------------------------
64// Methods
65//--------------------------------------------------------------------------------------------------
66
67impl ControlConnection {
68    /// Discover support at an existing native endpoint. Path identity alone is
69    /// not sufficient to reuse JSON format evidence across process replacement.
70    pub async fn connect(path: impl AsRef<Path>) -> ControlClientResult<Self> {
71        Self::connect_with(path, |options| options).await
72    }
73
74    /// Configure one total deadline covering discovery, redial, and handshake.
75    pub async fn connect_with(
76        path: impl AsRef<Path>,
77        configure: impl FnOnce(ConnectOptions) -> ConnectOptions,
78    ) -> ControlClientResult<Self> {
79        Self::connect_connector_with(Arc::new(LocalConnector::new(path)), configure).await
80    }
81
82    /// Discover using a repeatable, owned connector without claimed OS identity.
83    pub async fn connect_connector(connector: Arc<dyn Connector>) -> ControlClientResult<Self> {
84        Self::connect_connector_with(connector, |options| options).await
85    }
86
87    /// Configure discovery over caller-provided independent transports.
88    pub async fn connect_connector_with(
89        connector: Arc<dyn Connector>,
90        configure: impl FnOnce(ConnectOptions) -> ConnectOptions,
91    ) -> ControlClientResult<Self> {
92        Self::establish(
93            Dialer::Unverified(connector),
94            configure(ConnectOptions::default()),
95        )
96        .await
97    }
98
99    /// Reuse JSON discovery while an external runtime owner verifies every peer
100    /// and rechecks its active run and process birth before each operation.
101    pub async fn connect_verified_connector(
102        connector: Arc<dyn VerifiedControlConnector>,
103    ) -> ControlClientResult<Self> {
104        Self::connect_verified_connector_with(connector, |options| options).await
105    }
106
107    /// Configure setup for an identity-verifying backend connector.
108    pub async fn connect_verified_connector_with(
109        connector: Arc<dyn VerifiedControlConnector>,
110        configure: impl FnOnce(ConnectOptions) -> ConnectOptions,
111    ) -> ControlClientResult<Self> {
112        Self::establish(
113            Dialer::Verified(connector),
114            configure(ConnectOptions::default()),
115        )
116        .await
117    }
118
119    async fn establish(dialer: Dialer, options: ConnectOptions) -> ControlClientResult<Self> {
120        options.limits.validate()?;
121        let until = deadline(options.setup_timeout)?;
122        timeout_at(until, async {
123            let json = JsonControlClient::configured(dialer.clone(), options.clone(), true);
124            let (mode, capabilities) = json.discover(until).await?;
125            let selected = match mode {
126                ControlMode::Json => Selected::Json(json),
127                ControlMode::Framed => {
128                    // JSON completes on its own stream. A positively advertised
129                    // framed path gets a fresh connection, never an in-place
130                    // upgrade or a fallback after a failed welcome.
131                    let transport = dialer.connect(until).await?;
132                    dialer.verify(until).await?;
133                    check_deadline(until)?;
134                    let established =
135                        ControlProtocol::establish(transport, options.clone()).await?;
136                    Selected::Framed(ControlClient::from_established(established).await?)
137                }
138            };
139            Ok(Self {
140                inner: Arc::new(Inner {
141                    selected,
142                    dialer,
143                    options,
144                    closed: CancellationToken::new(),
145                    capabilities,
146                }),
147            })
148        })
149        .await
150        .unwrap_or_else(|_| Err(ClientError::new(ErrorKind::Timeout).into()))
151        .map_err(not_sent)
152    }
153
154    /// Format selected by this setup. A later detected format change invalidates
155    /// the JSON handle before sending; callers explicitly establish a new one.
156    pub fn mode(&self) -> ControlMode {
157        match self.inner.selected {
158            Selected::Framed(_) => ControlMode::Framed,
159            Selected::Json(_) => ControlMode::Json,
160        }
161    }
162
163    /// Capability snapshot validated during discovery. Reading it performs no
164    /// I/O; use GetCapabilities to explicitly request a fresh observation.
165    pub fn capabilities(&self) -> &crate::Capabilities {
166        &self.inner.capabilities
167    }
168
169    /// Inspect shared closure without dialing or rediscovery.
170    pub fn is_closed(&self) -> bool {
171        match &self.inner.selected {
172            Selected::Framed(client) => client.is_closed(),
173            Selected::Json(client) => client.is_closed(),
174        }
175    }
176
177    /// Wait for the selected connection to close without dialing or polling.
178    pub async fn closed(&self) {
179        match &self.inner.selected {
180            Selected::Framed(client) => client.closed().await,
181            Selected::Json(client) => client.closed().await,
182        }
183    }
184
185    /// Obtain the complete generic framed surface. JSON fails locally, without
186    /// opening an endpoint or manufacturing an equivalent frame.
187    pub fn framed(&self) -> ControlClientResult<&ControlClient> {
188        match &self.inner.selected {
189            Selected::Framed(client) => Ok(client),
190            Selected::Json(_) => Err(ControlClientError::UnsupportedMode),
191        }
192    }
193
194    /// Close all shared handles. Subsequent requests cannot silently reconnect.
195    pub async fn close(&self) {
196        // Verification can be awaiting a backend/database operation before the
197        // generic client has a request to wake. It shares explicit close too.
198        self.inner.closed.cancel();
199        match &self.inner.selected {
200            Selected::Framed(client) => client.close().await,
201            Selected::Json(client) => client.close().await,
202        }
203    }
204
205    /// Send a named request and retain its actual reply representation.
206    pub async fn request(
207        &self,
208        message: impl IntoControlMessage,
209    ) -> ControlClientResult<ControlReply> {
210        self.request_with(message, |options| options).await
211    }
212
213    /// Configure one attempt, including any identity recheck or JSON rediscovery.
214    pub async fn request_with(
215        &self,
216        message: impl IntoControlMessage,
217        configure: impl FnOnce(RequestOptions) -> RequestOptions,
218    ) -> ControlClientResult<ControlReply> {
219        let options = configure(RequestOptions::default());
220        match &self.inner.selected {
221            Selected::Json(client) => Ok(ControlReply::Json(
222                client.operation(message.into_json()?, options).await?,
223            )),
224            Selected::Framed(client) => {
225                let options = self.verify_before_request(options).await?;
226                Ok(ControlReply::Framed(
227                    client.request_with(message, |_| options).await?,
228                ))
229            }
230        }
231    }
232
233    /// Normalize a checked request using the selected format's real decoder.
234    pub async fn request_typed<R: CheckedControlRequest>(
235        &self,
236        request: &R,
237    ) -> ControlClientResult<R::Response> {
238        self.request_typed_with(request, |options| options).await
239    }
240
241    /// Configure one checked operation's local wait.
242    pub async fn request_typed_with<R: CheckedControlRequest>(
243        &self,
244        request: &R,
245        configure: impl FnOnce(RequestOptions) -> RequestOptions,
246    ) -> ControlClientResult<R::Response> {
247        let options = configure(RequestOptions::default());
248        match &self.inner.selected {
249            Selected::Json(client) => {
250                let reply = client.operation(request.json_request()?, options).await?;
251                request.decode_json(reply)
252            }
253            Selected::Framed(client) => {
254                let options = self.verify_before_request(options).await?;
255                client.request_typed_with(request, |_| options).await
256            }
257        }
258    }
259
260    async fn verify_before_request(
261        &self,
262        options: RequestOptions,
263    ) -> ControlClientResult<RequestOptions> {
264        if self.is_closed() {
265            return Err(ClientError::new(ErrorKind::Closed).into());
266        }
267        let until = deadline(
268            options
269                .request_timeout
270                .or(self.inner.options.limits.request_timeout)
271                .unwrap_or(DEFAULT_REQUEST_TIMEOUT),
272        )?;
273        check_deadline(until)?;
274        let result = tokio::select! {
275            biased;
276            _ = self.inner.closed.cancelled() => Err(ClientError::new(ErrorKind::Closed).into()),
277            result = timeout_at(until, self.inner.dialer.verify(until)) => result
278                .unwrap_or_else(|_| Err(ClientError::new(ErrorKind::Timeout).into())),
279        };
280        if let Err(error) = result {
281            self.close().await;
282            return Err(not_sent(error));
283        }
284        check_deadline(until)?;
285        Ok(RequestOptions::default()
286            .request_timeout(until.saturating_duration_since(tokio::time::Instant::now())))
287    }
288}
289
290//--------------------------------------------------------------------------------------------------
291// Functions
292//--------------------------------------------------------------------------------------------------
293
294pub(crate) fn not_sent(error: ControlClientError) -> ControlClientError {
295    match error {
296        ControlClientError::Client(error) => error.with_delivery(Delivery::NotSent).into(),
297        error => error,
298    }
299}
microsandbox_control_client/connection.rs

microsandbox_control_client/
connection.rs
