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    /// The credential the next connection presents: the fixed pass, the
69    /// agent's pass from Lys, renewed when it is near expiry, or nothing at
70    /// all from an open seat (ACCESS-010 R3).
71    fn credential(&self) -> Result<Vec<u8>, RailError> {
72        match &self.source {
73            Source::Fixed(bytes) => Ok(bytes.clone()),
74            Source::Open(record) => Ok(record.clone()),
75            Source::Lys(agent) => agent
76                .pass()
77                .map(|pass| lys_pass_credential(&pass))
78                .map_err(|error| RailError::Refused(error.0)),
79        }
80    }
81}
82
83/// Where a connection's pass comes from.
84enum Source {
85    /// One pass, read at start.
86    Fixed(Vec<u8>),
87    /// The agent's own pass, asked of Lys and renewed.
88    Lys(LysAgent),
89    /// No credential: an open seat (ACCESS-010 R3) and the name it declares
90    /// as `open:<name>`, or nothing (ACCESS-012 R2).
91    Open(Vec<u8>),
92}
93
94impl Rail for Live {
95    fn topology(&self) -> Result<Value, RailError> {
96        let stream = self.admit(|credential| {
97            Counts::add(&self.counts.connections_opened);
98            Subscription::open(&self.server, TOPOLOGY_SUBJECT, credential).map_err(|error| {
99                Counts::add(&self.counts.connections_closed);
100                RailError::Refused(error.to_string())
101            })
102        })?;
103        Counts::add(&self.counts.subscribes);
104        let record = stream.recv_timeout(self.limits.reply_timeout());
105        drop(stream);
106        Counts::add(&self.counts.connections_closed);
107        let record = record.map_err(|_| RailError::NoRecord {
108            millis: self.limits.reply_timeout_ms,
109        })?;
110        Counts::add(&self.counts.records);
111        if record.payload().len() > self.limits.record_bytes {
112            return Err(RailError::RecordTooLarge {
113                limit: self.limits.record_bytes,
114            });
115        }
116        serde_json::from_slice(record.payload()).map_err(|error| RailError::Unreadable {
117            message: error.to_string(),
118        })
119    }
120
121    fn request(
122        &self,
123        subject: &str,
124        input: &Value,
125        reply: bool,
126    ) -> Result<Option<Value>, RailError> {
127        self.conversations.fetch_add(1, Ordering::SeqCst);
128        Counts::add(&self.counts.connections_opened);
129        let outcome = self.request_once(subject, input, reply);
130        Counts::add(&self.counts.connections_closed);
131        outcome
132    }
133
134    fn pass_for(&self, audience: &str) -> Result<String, RailError> {
135        match &self.source {
136            Source::Fixed(_) => Err(RailError::Refused(
137                "pass_for needs a [lys] table: a ticket on a fixed token cannot ask Lys for a pass"
138                    .to_owned(),
139            )),
140            Source::Open(_) => Err(RailError::Refused(
141                "pass_for needs a [lys] table: an open seat cannot ask Lys for a pass".to_owned(),
142            )),
143            Source::Lys(agent) => agent
144                .pass_for(audience)
145                .map_err(|error| RailError::Refused(error.0)),
146        }
147    }
148
149    fn tap(&self, channel: &str) -> Result<super::TapSubscription, RailError> {
150        self.admit(|credential| {
151            super::live_tap::open(&self.server, channel, credential, self.limits.record_bytes)
152        })
153    }
154
155    fn server_address(&self) -> &str {
156        &self.server
157    }
158
159    fn counts(&self) -> &Counts {
160        &self.counts
161    }
162
163    fn request_bytes(&self) -> usize {
164        self.limits.request_bytes
165    }
166
167    fn flow_steps(&self) -> usize {
168        self.limits.flow_steps
169    }
170}
171
172impl Live {
173    fn request_once(
174        &self,
175        subject: &str,
176        input: &Value,
177        reply: bool,
178    ) -> Result<Option<Value>, RailError> {
179        let mut attempted = false;
180        let config = self.admit(|credential| {
181            if attempted {
182                Counts::add(&self.counts.connections_closed);
183                Counts::add(&self.counts.connections_opened);
184            }
185            attempted = true;
186            RemoteConfig::new(
187                self.server.clone(),
188                self.channel.clone(),
189                subject.to_owned(),
190                ConnectionPoolConfig::new(1, 10, 16),
191            )
192            .and_then(|config| rail_link::connect(&self.server, config, credential))
193            .map_err(|error| RailError::Refused(error.to_string()))
194        })?;
195        let conversation = RemoteConversationHandle::new(&config);
196        Counts::add(&self.counts.requests);
197        if !reply {
198            conversation
199                .send(input)
200                .map_err(|error| RailError::Refused(error.to_string()))?;
201            return Ok(None);
202        }
203        conversation
204            .request(input)
205            .map_err(|error| RailError::Refused(error.to_string()))?;
206        let answer: Value = block_on(conversation.receive())
207            .and_then(|received| received.map_err(|error| error.to_string()))
208            .map_err(RailError::Refused)?;
209        Counts::add(&self.counts.replies);
210        Ok(Some(answer))
211    }
212}
213
214/// Drives the SDK's ready-or-never future once: the synchronous transport
215/// has the reply in hand by the time `receive` is polled.
216fn block_on<F: Future>(future: F) -> Result<F::Output, String> {
217    let mut future = pin!(future);
218    let waker = Waker::noop();
219    let mut context = Context::from_waker(waker);
220    match future.as_mut().poll(&mut context) {
221        Poll::Ready(value) => Ok(value),
222        Poll::Pending => Err("the synchronous transport parked unexpectedly".to_owned()),
223    }
224}
225
226#[path = "pass_renewal.rs"]
227mod pass_renewal;
228
229#[cfg(test)]
230#[path = "live_tests.rs"]
231mod tests;
232
233impl std::fmt::Debug for Source {
234    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
235        f.write_str(match self {
236            Self::Fixed(_) => "FixedCredential",
237            Self::Lys(_) => "LysCredential",
238            Self::Open(_) => "OpenCredential",
239        })
240    }
241}
liminal_cli/network/live.rs

liminal_cli/network/
live.rs
