liminal_cli/network/
live.rs

1//! The live rail: a liminal server through the Rust SDK, one connection per
2//! read or request, closed before the answer is returned.
3
4use std::future::Future;
5use std::pin::pin;
6use std::sync::atomic::{AtomicU64, Ordering};
7use std::task::{Context, Poll, Waker};
8
9use liminal_sdk::{
10    ConnectionPoolConfig, ConversationHandle, RemoteConfig, RemoteConversationHandle,
11    lys_pass_credential,
12};
13
14use crate::rail_link::{self, Subscription};
15use serde_json::Value;
16
17use super::{Counts, Rail, RailError};
18use crate::config::{Config, Credential, Limits};
19use crate::lys::LysAgent;
20
21/// The console's topology subject, the directory's surface (SRV-027 R3).
22pub const TOPOLOGY_SUBJECT: &str = "console.topology";
23
24/// The live rail.
25pub struct Live {
26    server: String,
27    channel: String,
28    source: Source,
29    limits: Limits,
30    counts: Counts,
31    conversations: AtomicU64,
32}
33
34impl std::fmt::Debug for Live {
35    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
36        f.debug_struct("Live")
37            .field("server", &self.server)
38            .field("channel", &self.channel)
39            .field("source", &self.source)
40            .field("limits", &self.limits)
41            .finish_non_exhaustive()
42    }
43}
44
45impl Live {
46    /// A rail on `config`'s server and pass.
47    ///
48    /// # Errors
49    /// Names a `[lys]` table that cannot ask Lys (see [`LysAgent::new`]).
50    pub fn new(config: &Config) -> Result<Self, &'static str> {
51        let source = match &config.credential {
52            Credential::Token(token) => Source::Fixed(lys_pass_credential(token.as_str())),
53            Credential::Lys(table) => {
54                Source::Lys(LysAgent::new(table, config.limits.reply_timeout())?)
55            }
56            Credential::Open { .. } => Source::Open(config.credential.open_record()),
57        };
58        Ok(Self {
59            server: config.server.clone(),
60            channel: config.channel.clone(),
61            source,
62            limits: config.limits,
63            counts: Counts::default(),
64            conversations: AtomicU64::new(0),
65        })
66    }
67
68    /// An open seat's `Connect` record (`open:<name>`, or empty when
69    /// anonymous); `None` for a seat holding a pass.
70    #[must_use]
71    pub fn open_record(&self) -> Option<Vec<u8>> {
72        match &self.source {
73            Source::Open(record) => Some(record.clone()),
74            Source::Fixed(_) | Source::Lys(_) => None,
75        }
76    }
77
78    /// The credential the next connection presents: the fixed pass, the
79    /// agent's pass from Lys, renewed when it is near expiry, or nothing at
80    /// all from an open seat (ACCESS-010 R3).
81    fn credential(&self) -> Result<Vec<u8>, RailError> {
82        match &self.source {
83            Source::Fixed(bytes) => Ok(bytes.clone()),
84            Source::Open(record) => Ok(record.clone()),
85            Source::Lys(agent) => agent
86                .pass()
87                .map(|pass| lys_pass_credential(&pass))
88                .map_err(|error| RailError::Refused(error.0)),
89        }
90    }
91}
92
93/// Where a connection's pass comes from.
94enum Source {
95    /// One pass, read at start.
96    Fixed(Vec<u8>),
97    /// The agent's own pass, asked of Lys and renewed.
98    Lys(LysAgent),
99    /// No credential: an open seat (ACCESS-010 R3) and the name it declares
100    /// as `open:<name>`, or nothing (ACCESS-012 R2).
101    Open(Vec<u8>),
102}
103
104impl Rail for Live {
105    fn topology(&self) -> Result<Value, RailError> {
106        let stream = self.admit(|credential| {
107            Counts::add(&self.counts.connections_opened);
108            Subscription::open(&self.server, TOPOLOGY_SUBJECT, credential).map_err(|error| {
109                Counts::add(&self.counts.connections_closed);
110                RailError::Refused(error.to_string())
111            })
112        })?;
113        Counts::add(&self.counts.subscribes);
114        let record = stream.recv_timeout(self.limits.reply_timeout());
115        drop(stream);
116        Counts::add(&self.counts.connections_closed);
117        let record = record.map_err(|_| RailError::NoRecord {
118            millis: self.limits.reply_timeout_ms,
119        })?;
120        Counts::add(&self.counts.records);
121        if record.payload().len() > self.limits.record_bytes {
122            return Err(RailError::RecordTooLarge {
123                limit: self.limits.record_bytes,
124            });
125        }
126        serde_json::from_slice(record.payload()).map_err(|error| RailError::Unreadable {
127            message: error.to_string(),
128        })
129    }
130
131    fn request(
132        &self,
133        subject: &str,
134        input: &Value,
135        reply: bool,
136    ) -> Result<Option<Value>, RailError> {
137        self.conversations.fetch_add(1, Ordering::SeqCst);
138        Counts::add(&self.counts.connections_opened);
139        let outcome = self.request_once(subject, input, reply);
140        Counts::add(&self.counts.connections_closed);
141        outcome
142    }
143
144    fn pass_for(&self, audience: &str) -> Result<String, RailError> {
145        match &self.source {
146            Source::Fixed(_) => Err(RailError::Refused(
147                "pass_for needs a [lys] table: a ticket on a fixed token cannot ask Lys for a pass"
148                    .to_owned(),
149            )),
150            Source::Open(_) => Err(RailError::Refused(
151                "pass_for needs a [lys] table: an open seat cannot ask Lys for a pass".to_owned(),
152            )),
153            Source::Lys(agent) => agent
154                .pass_for(audience)
155                .map_err(|error| RailError::Refused(error.0)),
156        }
157    }
158
159    fn tap(&self, channel: &str) -> Result<super::TapSubscription, RailError> {
160        self.admit(|credential| {
161            super::live_tap::open(&self.server, channel, credential, self.limits.record_bytes)
162        })
163    }
164
165    fn server_address(&self) -> &str {
166        &self.server
167    }
168
169    fn counts(&self) -> &Counts {
170        &self.counts
171    }
172
173    fn request_bytes(&self) -> usize {
174        self.limits.request_bytes
175    }
176
177    fn flow_steps(&self) -> usize {
178        self.limits.flow_steps
179    }
180}
181
182impl Live {
183    fn request_once(
184        &self,
185        subject: &str,
186        input: &Value,
187        reply: bool,
188    ) -> Result<Option<Value>, RailError> {
189        let mut attempted = false;
190        let config = self.admit(|credential| {
191            if attempted {
192                Counts::add(&self.counts.connections_closed);
193                Counts::add(&self.counts.connections_opened);
194            }
195            attempted = true;
196            RemoteConfig::new(
197                self.server.clone(),
198                self.channel.clone(),
199                subject.to_owned(),
200                ConnectionPoolConfig::new(1, 10, 16),
201            )
202            .and_then(|config| rail_link::connect(&self.server, config, credential))
203            .map_err(|error| RailError::Refused(error.to_string()))
204        })?;
205        let conversation = RemoteConversationHandle::new(&config);
206        Counts::add(&self.counts.requests);
207        if !reply {
208            conversation
209                .send(input)
210                .map_err(|error| RailError::Refused(error.to_string()))?;
211            return Ok(None);
212        }
213        conversation
214            .request(input)
215            .map_err(|error| RailError::Refused(error.to_string()))?;
216        let answer: Value = block_on(conversation.receive())
217            .and_then(|received| received.map_err(|error| error.to_string()))
218            .map_err(RailError::Refused)?;
219        Counts::add(&self.counts.replies);
220        Ok(Some(answer))
221    }
222}
223
224/// Drives the SDK's ready-or-never future once: the synchronous transport
225/// has the reply in hand by the time `receive` is polled.
226fn block_on<F: Future>(future: F) -> Result<F::Output, String> {
227    let mut future = pin!(future);
228    let waker = Waker::noop();
229    let mut context = Context::from_waker(waker);
230    match future.as_mut().poll(&mut context) {
231        Poll::Ready(value) => Ok(value),
232        Poll::Pending => Err("the synchronous transport parked unexpectedly".to_owned()),
233    }
234}
235
236#[path = "pass_renewal.rs"]
237mod pass_renewal;
238
239#[cfg(test)]
240#[path = "live_tests.rs"]
241mod tests;
242
243impl std::fmt::Debug for Source {
244    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
245        f.write_str(match self {
246            Self::Fixed(_) => "FixedCredential",
247            Self::Lys(_) => "LysCredential",
248            Self::Open(_) => "OpenCredential",
249        })
250    }
251}
liminal_cli/network/live.rs

liminal_cli/network/
live.rs
