cufflink_native/
service.rs

1use crate::dispatch::Dispatcher;
2use crate::nats::{NatsProvider, NatsServer};
3use crate::socket::SocketServer;
4use crate::Extension;
5use cufflink_types::ext::{callers, SOCKET_VAR, TOKEN_VAR};
6use cufflink_types::manifest::validate::validate_provides;
7use cufflink_types::slug::{is_valid_service_name, SERVICE_NAME_MAX};
8use cufflink_types::{ProvidedExtension, ServiceManifest, ServiceMode};
9use std::future::Future;
10use std::path::Path;
11use std::process::ExitCode;
12use std::sync::Arc;
13
14const DEFAULT_CONCURRENCY: usize = 64;
15
16/// A native service named `name`: add its extensions, then
17/// [`run`](Service::run) it.
18pub fn service(name: impl Into<String>) -> Service {
19    Service {
20        name: name.into(),
21        extensions: Vec::new(),
22        concurrency: DEFAULT_CONCURRENCY,
23    }
24}
25
26/// A native service and the extensions it provides.
27#[must_use]
28pub struct Service {
29    name: String,
30    extensions: Vec<(ProvidedExtension, Arc<dyn Extension>)>,
31    concurrency: usize,
32}
33
34impl Service {
35    /// Provide `extension` as `name` to `callers`: service names, or `*` for
36    /// every service of the tenant.
37    pub fn extension<C: Into<String>>(
38        mut self,
39        name: impl Into<String>,
40        extension: impl Extension,
41        callers: impl IntoIterator<Item = C>,
42    ) -> Self {
43        let provided = ProvidedExtension {
44            name: name.into(),
45            callers: callers.into_iter().map(Into::into).collect(),
46        };
47        self.extensions.push((provided, Arc::new(extension)));
48        self
49    }
50
51    /// Run at most `calls` calls at once (64 unless set); the others wait
52    /// for a slot until their deadline.
53    pub fn concurrency(mut self, calls: usize) -> Self {
54        self.concurrency = calls;
55        self
56    }
57
58    /// The `mode: native` manifest of the service, which the CLI deploys:
59    /// its name and what it provides, each extension's callers in order,
60    /// once each.
61    pub fn manifest(&self) -> eyre::Result<ServiceManifest> {
62        eyre::ensure!(
63            is_valid_service_name(&self.name),
64            "'{}' is not a service name (1 to {SERVICE_NAME_MAX} of A-Z, a-z, 0-9, '-' and '_')",
65            self.name
66        );
67        let declared: Vec<ProvidedExtension> =
68            self.extensions.iter().map(|(p, _)| p.clone()).collect();
69        validate_provides("provides", &declared).map_err(eyre::Report::msg)?;
70        let mut provides = declared
71            .into_iter()
72            .map(|p| {
73                callers(p.callers).map(|callers| ProvidedExtension {
74                    name: p.name,
75                    callers,
76                })
77            })
78            .collect::<Result<Vec<_>, _>>()
79            .map_err(eyre::Report::msg)?;
80        provides.sort_by(|a, b| a.name.cmp(&b.name));
81        let manifest = ServiceManifest {
82            name: self.name.clone(),
83            mode: ServiceMode::Native,
84            provides,
85            ..Default::default()
86        };
87        Ok(ServiceManifest {
88            required_platform_features: manifest
89                .derived_features()
90                .into_iter()
91                .map(str::to_string)
92                .collect(),
93            ..manifest
94        })
95    }
96
97    /// The manifest of a service that can serve.
98    fn checked(&self) -> eyre::Result<ServiceManifest> {
99        eyre::ensure!(
100            self.concurrency > 0,
101            "concurrency: at least one call must be able to run"
102        );
103        self.manifest()
104    }
105
106    fn validated(self) -> eyre::Result<(String, Dispatcher)> {
107        self.checked()?;
108        let extensions = self
109            .extensions
110            .into_iter()
111            .map(|(provided, extension)| (provided.name, extension))
112            .collect();
113        Ok((self.name, Dispatcher::new(extensions, self.concurrency)))
114    }
115
116    /// Run the service as its binary's `main`: print its manifest under
117    /// `--emit-manifest`, else serve the Unix socket `CUFFLINK_EXT_SOCKET`
118    /// names, to requests bearing the token `CUFFLINK_EXT_TOKEN` holds, until
119    /// SIGTERM or SIGINT. Fails, saying why, when the service or its
120    /// environment is invalid.
121    ///
122    /// On Linux the process first makes itself undumpable, so other
123    /// processes of its user (the runner's other children) cannot read its
124    /// environment or memory.
125    pub fn run(self) -> ExitCode {
126        let emit = std::env::args_os()
127            .skip(1)
128            .any(|arg| arg == "--emit-manifest");
129        let served = self.checked().and_then(|manifest| {
130            if emit {
131                println!("{}", serde_json::to_string_pretty(&manifest)?);
132                return Ok(());
133            }
134            forbid_inspection()?;
135            init_tracing();
136            let path = std::env::var_os(SOCKET_VAR)
137                .ok_or_else(|| eyre::eyre!("{SOCKET_VAR} must name the socket to serve on"))?;
138            let token = std::env::var(TOKEN_VAR)
139                .ok()
140                .filter(|token| !token.is_empty())
141                .ok_or_else(|| {
142                    eyre::eyre!(
143                        "{TOKEN_VAR} must hold the token the runner sends with each request"
144                    )
145                })?;
146            block_on(async move {
147                let shutdown = shutdown_signal()?;
148                self.bind_socket(path)
149                    .await?
150                    .require_token(token)
151                    .serve(shutdown)
152                    .await
153            })
154        });
155        served.map_or_else(
156            |e| {
157                eprintln!("cufflink-native: {e:#}");
158                ExitCode::FAILURE
159            },
160            |()| ExitCode::SUCCESS,
161        )
162    }
163
164    /// Serve the extensions from NATS as `provider` until SIGTERM or
165    /// SIGINT, undumpable on Linux as [`run`](Self::run) is.
166    pub fn run_nats(self, provider: NatsProvider) -> eyre::Result<()> {
167        forbid_inspection()?;
168        init_tracing();
169        block_on(async move { self.serve_nats(provider, shutdown_signal()?).await })
170    }
171
172    /// Listen on the Unix socket `path`, replacing a stale one.
173    pub async fn bind_socket(self, path: impl AsRef<Path>) -> eyre::Result<SocketServer> {
174        let (_, dispatcher) = self.validated()?;
175        SocketServer::bind(path.as_ref(), dispatcher)
176    }
177
178    /// Connect to `provider`'s NATS and subscribe to the calls of each
179    /// extension in the queue group every provider shares.
180    pub async fn bind_nats(self, provider: NatsProvider) -> eyre::Result<NatsServer> {
181        let (name, dispatcher) = self.validated()?;
182        NatsServer::bind(&name, provider, dispatcher).await
183    }
184
185    /// [`bind_nats`](Self::bind_nats), then [`serve`](NatsServer::serve)
186    /// until `shutdown`, which also ends a bind still waiting for NATS.
187    pub async fn serve_nats(
188        self,
189        provider: NatsProvider,
190        shutdown: impl Future<Output = ()>,
191    ) -> eyre::Result<()> {
192        let mut shutdown = std::pin::pin!(shutdown);
193        let bound = tokio::select! {
194            bound = self.bind_nats(provider) => bound?,
195            () = &mut shutdown => return Ok(()),
196        };
197        bound.serve(shutdown).await
198    }
199}
200
201/// Keep other processes of this user from reading this one's environment
202/// (`/proc/<pid>/environ`) or memory: execve makes every process dumpable
203/// again, so each must do it itself.
204#[cfg(target_os = "linux")]
205fn forbid_inspection() -> eyre::Result<()> {
206    // SAFETY: PR_SET_DUMPABLE takes plain integers and touches no memory.
207    match unsafe { libc::prctl(libc::PR_SET_DUMPABLE, 0, 0, 0, 0) } {
208        0 => Ok(()),
209        _ => Err(eyre::Report::new(std::io::Error::last_os_error())
210            .wrap_err("cannot make the process undumpable")),
211    }
212}
213
214#[cfg(not(target_os = "linux"))]
215fn forbid_inspection() -> eyre::Result<()> {
216    Ok(())
217}
218
219fn init_tracing() {
220    let filter = tracing_subscriber::EnvFilter::try_from_default_env()
221        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));
222    let _ = tracing_subscriber::fmt()
223        .with_env_filter(filter)
224        .with_writer(std::io::stderr)
225        .try_init();
226}
227
228fn block_on(served: impl Future<Output = eyre::Result<()>>) -> eyre::Result<()> {
229    let runtime = tokio::runtime::Builder::new_multi_thread()
230        .enable_all()
231        .build()?;
232    let result = runtime.block_on(served);
233    runtime.shutdown_background();
234    result
235}
236
237fn shutdown_signal() -> eyre::Result<impl Future<Output = ()>> {
238    use tokio::signal::unix::{signal, SignalKind};
239    let (mut terminate, mut interrupt) = (
240        signal(SignalKind::terminate())?,
241        signal(SignalKind::interrupt())?,
242    );
243    Ok(async move {
244        tokio::select! {
245            _ = terminate.recv() => {}
246            _ = interrupt.recv() => {}
247        }
248        tracing::info!("shutting down");
249    })
250}
251
252#[cfg(test)]
253mod tests {
254    use super::*;
255    use crate::{CallContext, Result};
256    use cufflink_types::features::NATIVE_SERVICES;
257    use serde_json::Value;
258
259    fn token(_: &CallContext, payload: Value) -> Result<Value> {
260        Ok(payload)
261    }
262
263    fn refused(service: Service) -> String {
264        match service.validated() {
265            Ok(_) => panic!("accepted"),
266            Err(e) => e.to_string(),
267        }
268    }
269
270    #[test]
271    fn the_manifest_is_native_and_lists_what_it_provides() {
272        let manifest = service("ext-google")
273            .extension("pdf.render", token, ["*"])
274            .extension(
275                "google.token",
276                token,
277                ["reports-service", "billing", "reports-service"],
278            )
279            .manifest()
280            .expect("manifest");
281        let emitted: Value =
282            serde_json::from_str(&serde_json::to_string_pretty(&manifest).expect("json"))
283                .expect("parse");
284        assert_eq!(emitted["name"], "ext-google");
285        assert_eq!(emitted["mode"], "native");
286        assert_eq!(emitted["tables"], serde_json::json!([]));
287        assert_eq!(
288            emitted["provides"],
289            serde_json::json!([
290                {"name": "google.token", "callers": ["billing", "reports-service"]},
291                {"name": "pdf.render", "callers": ["*"]},
292            ])
293        );
294        assert_eq!(
295            emitted["required_platform_features"],
296            serde_json::json!([NATIVE_SERVICES])
297        );
298        let parsed: ServiceManifest = serde_json::from_value(emitted).expect("a manifest");
299        assert!(!parsed.requires_wasm());
300    }
301
302    #[test]
303    fn an_invalid_service_says_why() {
304        for (service, reason) in [
305            (
306                service("ext google").extension("google.token", token, ["*"]),
307                "'ext google' is not a service name (1 to 64 of A-Z, a-z, 0-9, '-' and '_')",
308            ),
309            (
310                service("ext-google"),
311                "provides: a native service provides at least one extension",
312            ),
313            (
314                service("ext-google").extension("Google.Token", token, ["*"]),
315                "provides[0]: 'Google.Token' is not an extension name (segments of a-z, 0-9 and \
316                 '-' joined by '.', at most 128 bytes)",
317            ),
318            (
319                service("ext-google")
320                    .extension("google.token", token, ["*"])
321                    .extension("google.token", token, ["billing"]),
322                "provides: extension 'google.token' is listed twice",
323            ),
324            (
325                service("ext-google").extension("google.token", token, Vec::<String>::new()),
326                "provides[0].callers: at least one caller is required",
327            ),
328            (
329                service("ext-google").extension("google.token", token, ["reports.service"]),
330                "provides[0].callers: 'reports.service' is neither a service name nor '*'",
331            ),
332            (
333                service("ext-google")
334                    .extension("google.token", token, ["*"])
335                    .concurrency(0),
336                "concurrency: at least one call must be able to run",
337            ),
338        ] {
339            assert_eq!(refused(service), reason);
340        }
341    }
342
343    #[cfg(target_os = "linux")]
344    #[test]
345    fn a_process_can_make_itself_undumpable() {
346        forbid_inspection().expect("undumpable");
347        // SAFETY: PR_GET_DUMPABLE reads a flag of this process.
348        assert_eq!(unsafe { libc::prctl(libc::PR_GET_DUMPABLE, 0, 0, 0, 0) }, 0);
349    }
350
351    #[test]
352    fn a_valid_service_dispatches_its_extensions() {
353        let (name, dispatcher) = service("ext-google")
354            .extension("google.token", token, ["*"])
355            .validated()
356            .expect("valid");
357        assert_eq!(name, "ext-google");
358        assert_eq!(dispatcher.names().collect::<Vec<_>>(), ["google.token"]);
359    }
360}
cufflink_native/service.rs

cufflink_native/
service.rs
