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    SubscriptionStream, lys_pass_credential,
12};
13use serde_json::Value;
14
15use super::{Counts, Rail, RailError};
16use crate::config::{Config, Credential, Limits};
17use crate::lys::LysAgent;
18
19/// The console's topology subject, the directory's surface (SRV-027 R3).
20pub const TOPOLOGY_SUBJECT: &str = "console.topology";
21
22/// The live rail.
23pub struct Live {
24    server: String,
25    channel: String,
26    source: Source,
27    limits: Limits,
28    counts: Counts,
29    conversations: AtomicU64,
30}
31
32impl std::fmt::Debug for Live {
33    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34        f.debug_struct("Live")
35            .field("server", &self.server)
36            .field("channel", &self.channel)
37            .field("source", &self.source)
38            .field("limits", &self.limits)
39            .finish_non_exhaustive()
40    }
41}
42
43impl Live {
44    /// A rail on `config`'s server and pass.
45    ///
46    /// # Errors
47    /// Names a `[lys]` table that cannot ask Lys (see [`LysAgent::new`]).
48    pub fn new(config: &Config) -> Result<Self, &'static str> {
49        let source = match &config.credential {
50            Credential::Token(token) => Source::Fixed(lys_pass_credential(token.as_str())),
51            Credential::Lys(table) => {
52                Source::Lys(LysAgent::new(table, config.limits.reply_timeout())?)
53            }
54        };
55        Ok(Self {
56            server: config.server.clone(),
57            channel: config.channel.clone(),
58            source,
59            limits: config.limits,
60            counts: Counts::default(),
61            conversations: AtomicU64::new(0),
62        })
63    }
64
65    /// The credential the next connection presents: the fixed pass, or the
66    /// agent's pass from Lys, renewed when it is near expiry.
67    fn credential(&self) -> Result<Vec<u8>, RailError> {
68        match &self.source {
69            Source::Fixed(bytes) => Ok(bytes.clone()),
70            Source::Lys(agent) => agent
71                .pass()
72                .map(|pass| lys_pass_credential(&pass))
73                .map_err(|error| RailError::Refused(error.0)),
74        }
75    }
76}
77
78/// Where a connection's pass comes from.
79#[derive(Debug)]
80enum Source {
81    /// One pass, read at start.
82    Fixed(Vec<u8>),
83    /// The agent's own pass, asked of Lys and renewed.
84    Lys(LysAgent),
85}
86
87impl Rail for Live {
88    fn topology(&self) -> Result<Value, RailError> {
89        let stream = self.admit(|credential| {
90            Counts::add(&self.counts.connections_opened);
91            SubscriptionStream::open_with_auth(
92                &self.server,
93                TOPOLOGY_SUBJECT,
94                Vec::new(),
95                credential,
96            )
97            .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::Lys(agent) => agent
140                .pass_for(audience)
141                .map_err(|error| RailError::Refused(error.0)),
142        }
143    }
144
145    fn tap(&self, channel: &str) -> Result<super::TapSubscription, RailError> {
146        self.admit(|credential| {
147            super::live_tap::open(&self.server, channel, credential, self.limits.record_bytes)
148        })
149    }
150
151    fn server_address(&self) -> &str {
152        &self.server
153    }
154
155    fn counts(&self) -> &Counts {
156        &self.counts
157    }
158
159    fn request_bytes(&self) -> usize {
160        self.limits.request_bytes
161    }
162
163    fn flow_steps(&self) -> usize {
164        self.limits.flow_steps
165    }
166}
167
168impl Live {
169    fn request_once(
170        &self,
171        subject: &str,
172        input: &Value,
173        reply: bool,
174    ) -> Result<Option<Value>, RailError> {
175        let mut attempted = false;
176        let config = self.admit(|credential| {
177            if attempted {
178                Counts::add(&self.counts.connections_closed);
179                Counts::add(&self.counts.connections_opened);
180            }
181            attempted = true;
182            RemoteConfig::new(
183                self.server.clone(),
184                self.channel.clone(),
185                subject.to_owned(),
186                ConnectionPoolConfig::new(1, 10, 16),
187            )
188            .and_then(|config| config.connect_tcp_with_auth(credential))
189            .map_err(|error| RailError::Refused(error.to_string()))
190        })?;
191        let conversation = RemoteConversationHandle::new(&config);
192        Counts::add(&self.counts.requests);
193        if !reply {
194            conversation
195                .send(input)
196                .map_err(|error| RailError::Refused(error.to_string()))?;
197            return Ok(None);
198        }
199        conversation
200            .request(input)
201            .map_err(|error| RailError::Refused(error.to_string()))?;
202        let answer: Value = block_on(conversation.receive())
203            .and_then(|received| received.map_err(|error| error.to_string()))
204            .map_err(RailError::Refused)?;
205        Counts::add(&self.counts.replies);
206        Ok(Some(answer))
207    }
208}
209
210/// Drives the SDK's ready-or-never future once: the synchronous transport
211/// has the reply in hand by the time `receive` is polled.
212fn block_on<F: Future>(future: F) -> Result<F::Output, String> {
213    let mut future = pin!(future);
214    let waker = Waker::noop();
215    let mut context = Context::from_waker(waker);
216    match future.as_mut().poll(&mut context) {
217        Poll::Ready(value) => Ok(value),
218        Poll::Pending => Err("the synchronous transport parked unexpectedly".to_owned()),
219    }
220}
221
222#[path = "pass_renewal.rs"]
223mod pass_renewal;
liminal_cli/network/live.rs

liminal_cli/network/
live.rs
