microsandbox_control_client/
json_client.rs

1//! Legacy unary exchanges: one real JSON operation per independently owned stream.
2
3use std::path::Path;
4use std::sync::Arc;
5use std::time::Duration;
6
7use microsandbox_protocol::control::{ControlRequest, DEFAULT_REQUEST_TIMEOUT};
8use microsandbox_protocol_client::{
9    ClientError, ConnectOptions, Connector, Delivery, ErrorKind, LocalConnector, RequestOptions,
10};
11use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
12use tokio::time::{Instant, timeout_at};
13use tokio_util::sync::CancellationToken;
14use zeroize::Zeroizing;
15
16use crate::{
17    CheckedControlRequest, ControlClientError, ControlClientResult, ControlMode,
18    IntoControlMessage, JsonReply, dialer::Dialer,
19};
20
21//--------------------------------------------------------------------------------------------------
22// Constants
23//--------------------------------------------------------------------------------------------------
24
25/// Bound applies to new-client discovery replies, not legacy request lines.
26pub const MAX_DISCOVERY_RESPONSE_SIZE: usize = 64 * 1024;
27
28//--------------------------------------------------------------------------------------------------
29// Types
30//--------------------------------------------------------------------------------------------------
31
32/// Explicit JSON adapter. Construction is inert; every call opens one stream.
33#[derive(Clone)]
34pub struct JsonControlClient {
35    pub(crate) dialer: Dialer,
36    pub(crate) options: ConnectOptions,
37    closed: CancellationToken,
38    rediscover: bool,
39}
40
41//--------------------------------------------------------------------------------------------------
42// Methods
43//--------------------------------------------------------------------------------------------------
44
45impl JsonControlClient {
46    /// Select legacy JSON explicitly, without probing or opening a connection.
47    pub fn new(path: impl AsRef<Path>) -> Self {
48        Self::from_connector(Arc::new(LocalConnector::new(path)))
49    }
50
51    /// Store a repeatable dialer without I/O or hidden negotiation.
52    pub fn from_connector(connector: Arc<dyn Connector>) -> Self {
53        Self::configured(
54            Dialer::Unverified(connector),
55            ConnectOptions::default(),
56            false,
57        )
58    }
59
60    /// Configure local deadlines without performing I/O.
61    pub fn from_connector_with(
62        connector: Arc<dyn Connector>,
63        configure: impl FnOnce(ConnectOptions) -> ConnectOptions,
64    ) -> ControlClientResult<Self> {
65        let options = configure(ConnectOptions::default());
66        options.limits.validate()?;
67        Ok(Self::configured(
68            Dialer::Unverified(connector),
69            options,
70            false,
71        ))
72    }
73
74    pub(crate) fn configured(dialer: Dialer, options: ConnectOptions, rediscover: bool) -> Self {
75        Self {
76            dialer,
77            options,
78            closed: CancellationToken::new(),
79            rediscover,
80        }
81    }
82
83    /// Inspect shared closure, including unexpected exchange failure.
84    pub fn is_closed(&self) -> bool {
85        self.closed.is_cancelled()
86    }
87
88    /// Wait for shared closure. Successful per-operation JSON EOF is not a
89    /// session closure; explicit close or an unexpected exchange failure is.
90    pub async fn closed(&self) {
91        self.closed.cancelled().await;
92    }
93
94    /// Close every clone and wake each outstanding exchange with its own
95    /// admission certainty. No implicit reconnect follows an explicit close.
96    pub async fn close(&self) {
97        self.closed.cancel();
98    }
99
100    /// Translate a known native request, returning its actual JSON reply.
101    pub async fn request(
102        &self,
103        message: impl IntoControlMessage,
104    ) -> ControlClientResult<JsonReply> {
105        self.request_with(message, |options| options).await
106    }
107
108    /// Configure one total attempt deadline, including dial and any discovery.
109    pub async fn request_with(
110        &self,
111        message: impl IntoControlMessage,
112        configure: impl FnOnce(RequestOptions) -> RequestOptions,
113    ) -> ControlClientResult<JsonReply> {
114        let request = message.into_json()?;
115        self.operation(request, configure(RequestOptions::default()))
116            .await
117    }
118
119    /// Normalize a checked operation without manufacturing a CBOR response.
120    pub async fn request_typed<R: CheckedControlRequest>(
121        &self,
122        request: &R,
123    ) -> ControlClientResult<R::Response> {
124        self.request_typed_with(request, |options| options).await
125    }
126
127    /// Configure one checked unary attempt.
128    pub async fn request_typed_with<R: CheckedControlRequest>(
129        &self,
130        request: &R,
131        configure: impl FnOnce(RequestOptions) -> RequestOptions,
132    ) -> ControlClientResult<R::Response> {
133        let response = self
134            .operation(
135                request.json_request()?,
136                configure(RequestOptions::default()),
137            )
138            .await?;
139        request.decode_json(response)
140    }
141
142    pub(crate) async fn operation(
143        &self,
144        request: ControlRequest,
145        options: RequestOptions,
146    ) -> ControlClientResult<JsonReply> {
147        let until = deadline(
148            options
149                .request_timeout
150                .or(self.options.limits.request_timeout)
151                .unwrap_or(DEFAULT_REQUEST_TIMEOUT),
152        )?;
153        let setup_until = deadline(self.options.setup_timeout)?.min(until);
154        if self.rediscover && !self.dialer.verified() {
155            // A path-only caller cannot reuse evidence about an old process.
156            // Discover again before this new exchange; a format change closes
157            // this handle instead of silently retargeting its prepared request.
158            let mode = self
159                .discover(setup_until)
160                .await
161                .map_err(crate::connection::not_sent);
162            let mode = match mode {
163                Ok((mode, _)) => mode,
164                Err(error) => {
165                    self.close().await;
166                    return Err(error);
167                }
168            };
169            if mode != ControlMode::Json {
170                self.close().await;
171                return Err(ControlClientError::RuntimeChanged);
172            }
173        }
174        self.exchange(&request, until, setup_until, None).await
175    }
176
177    pub(crate) async fn discover(
178        &self,
179        until: Instant,
180    ) -> ControlClientResult<(ControlMode, crate::Capabilities)> {
181        let reply = self
182            .exchange(
183                &ControlRequest::Capabilities,
184                until,
185                until,
186                Some(MAX_DISCOVERY_RESPONSE_SIZE),
187            )
188            .await?;
189        let mode = reply.discovery_mode()?;
190        let capabilities = crate::json_reply::capabilities(
191            reply
192                .value()
193                .get("capabilities")
194                .ok_or_else(|| ClientError::new(ErrorKind::InvalidData))?,
195        )
196        .ok_or_else(|| ClientError::new(ErrorKind::InvalidData))?;
197        Ok((mode, capabilities))
198    }
199
200    async fn exchange(
201        &self,
202        request: &ControlRequest,
203        until: Instant,
204        setup_until: Instant,
205        max_reply: Option<usize>,
206    ) -> ControlClientResult<JsonReply> {
207        let mut line = Zeroizing::new(
208            serde_json::to_vec(request).map_err(|_| ClientError::new(ErrorKind::Encode))?,
209        );
210        line.push(b'\n');
211        let mut admitted = false;
212        let result = timeout_at(until, async {
213            tokio::select! {
214                biased;
215                _ = self.closed.cancelled() => Err(ClientError::new(ErrorKind::Closed).into()),
216                result = async {
217                    check_deadline(setup_until)?;
218                    let mut transport = timeout_at(setup_until, self.dialer.connect(setup_until)).await
219                        .map_err(|_| ClientError::new(ErrorKind::Timeout))??;
220                    check_deadline(setup_until)?;
221                    timeout_at(setup_until, self.dialer.verify(setup_until)).await
222                        .map_err(|_| ClientError::new(ErrorKind::Timeout))??;
223                    check_deadline(until)?;
224                    if self.closed.is_cancelled() { return Err(ClientError::new(ErrorKind::Closed).into()); }
225                    // Mark uncertainty before the first write poll: even a
226                    // partial failed write may have reached the peer.
227                    admitted = true;
228                    transport.write_all(&line).await.map_err(ClientError::from)?;
229                    transport.flush().await.map_err(ClientError::from)?;
230                    let mut reader = BufReader::new(transport);
231                    let mut reply = Vec::new();
232                    loop {
233                        let buffer = reader.fill_buf().await.map_err(ClientError::from)?;
234                        if buffer.is_empty() {
235                            if reply.is_empty() { return Err(ClientError::new(ErrorKind::PeerClosed).into()); }
236                            break;
237                        }
238                        let end = buffer.iter().position(|byte| *byte == b'\n').map(|index| index + 1);
239                        let count = end.unwrap_or(buffer.len());
240                        if max_reply.is_some_and(|limit| reply.len().saturating_add(count) > limit) {
241                            return Err(ClientError::new(ErrorKind::InvalidData).into());
242                        }
243                        reply.extend_from_slice(&buffer[..count]);
244                        reader.consume(count);
245                        if end.is_some() { break; }
246                    }
247                    JsonReply::parse(reply)
248                } => result,
249            }
250        }).await.unwrap_or_else(|_| Err(ClientError::new(ErrorKind::Timeout).into()));
251        match result {
252            Ok(reply) => Ok(reply),
253            Err(error) => {
254                self.closed.cancel();
255                Err(match error {
256                    ControlClientError::Client(error) => error
257                        .with_delivery(if admitted {
258                            Delivery::Unknown
259                        } else {
260                            Delivery::NotSent
261                        })
262                        .into(),
263                    error => error,
264                })
265            }
266        }
267    }
268}
269
270//--------------------------------------------------------------------------------------------------
271// Functions
272//--------------------------------------------------------------------------------------------------
273
274pub(crate) fn deadline(duration: Duration) -> ControlClientResult<Instant> {
275    Instant::now()
276        .checked_add(duration)
277        .ok_or_else(|| ClientError::new(ErrorKind::InvalidOptions).into())
278}
279
280pub(crate) fn check_deadline(until: Instant) -> ControlClientResult<()> {
281    if Instant::now() >= until {
282        Err(ClientError::new(ErrorKind::Timeout).into())
283    } else {
284        Ok(())
285    }
286}
microsandbox_control_client/json_client.rs

microsandbox_control_client/
json_client.rs
