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

liminal_cli/network/
live.rs
