vcon_server/
serve.rs

1//! `vcon-server serve`: the server, run from its configuration.
2
3use std::future::Future;
4use std::sync::Arc;
5use std::time::Duration;
6use std::{fmt, io};
7
8use tokio::net::TcpListener;
9
10use crate::api::{self, App, Limits, TlsError};
11use crate::clock::{Clock, SystemClock};
12use crate::config::Config;
13use crate::datadir::{self, LockError, Work};
14use crate::envelope::{Key, KeyError};
15use crate::fs::OsFs;
16use crate::retention::Retention;
17use crate::tenants::{KeyOrder, RegistryError, RotationBacklog, TenantError, Tenants, with_causes};
18use time::OffsetDateTime;
19
20/// Why the server did not start, or stopped.
21#[derive(Debug)]
22pub enum ServeError {
23    /// The data directory could not be taken.
24    DataDir(LockError),
25    /// The master key file could not be read.
26    MasterKey(KeyError),
27    /// server.sqlite was made under another master key.
28    WrongMasterKey,
29    /// The registry could not be opened.
30    Open(RegistryError),
31    /// server.sqlite could not be read.
32    Registry(TenantError),
33    /// Some tenant's data is not under its current key.
34    KeysOutOfOrder(KeyOrder),
35    /// The certificate or key could not be used.
36    Tls(TlsError),
37    /// The listening address could not be bound.
38    Listen(io::Error),
39    /// Serving failed.
40    Serve(io::Error),
41}
42
43impl fmt::Display for ServeError {
44    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
45        match self {
46            ServeError::DataDir(e) => write!(f, "{e}"),
47            ServeError::MasterKey(e) => write!(f, "the master key cannot be read: {e:?}"),
48            ServeError::WrongMasterKey => {
49                f.write_str("the master key is not the one server.sqlite was made under")
50            }
51            ServeError::Open(e) => {
52                write!(f, "the registry cannot be opened: {}", with_causes(e))
53            }
54            ServeError::Registry(e) => write!(f, "server.sqlite cannot be opened: {e:?}"),
55            ServeError::KeysOutOfOrder(order) => {
56                let mut parts = Vec::new();
57                if order.wrapped_elsewhere > 0 {
58                    parts.push(format!(
59                        "{} tenant key(s) do not unwrap under the master key",
60                        order.wrapped_elsewhere
61                    ));
62                }
63                if order.rotations_stopped > 0 {
64                    parts.push(format!(
65                        "{} key rotation(s) stopped part way; run rekey tenant again",
66                        order.rotations_stopped
67                    ));
68                }
69                if order.data_under_other_keys > 0 {
70                    parts.push(format!(
71                        "{} tenant(s) with data under another key",
72                        order.data_under_other_keys
73                    ));
74                }
75                if order.catalogs_unopened > 0 {
76                    let why = order
77                        .unopened_error
78                        .as_ref()
79                        .map(|e| format!(": {e}"))
80                        .unwrap_or_default();
81                    parts.push(format!(
82                        "{} tenant catalog(s) cannot be opened{why}",
83                        order.catalogs_unopened
84                    ));
85                }
86                write!(
87                    f,
88                    "{}; vcon-server check names the tenants",
89                    parts.join("; ")
90                )
91            }
92            ServeError::Tls(e) => write!(f, "the TLS certificate or key cannot be used: {e:?}"),
93            ServeError::Listen(e) => write!(f, "the listening address cannot be bound: {e}"),
94            ServeError::Serve(e) => write!(f, "serving failed: {e}"),
95        }
96    }
97}
98
99/// Runs the server until `shutdown` completes: locks the data directory,
100/// opens the registry under the master key, and serves the API over TLS.
101pub async fn run(config: &Config, shutdown: impl Future<Output = ()>) -> Result<(), ServeError> {
102    let _lock =
103        datadir::lock(&config.data_dir, Work::Serve.sharing()).map_err(ServeError::DataDir)?;
104    let master = Key::load_master(&config.master_key).map_err(ServeError::MasterKey)?;
105    let tenants = Tenants::open(
106        &config.data_dir,
107        master,
108        Arc::new(OsFs),
109        Arc::new(SystemClock),
110    )
111    .map_err(|e| match e {
112        RegistryError::WrongKey => ServeError::WrongMasterKey,
113        other => ServeError::Open(other),
114    })?;
115    let order = tenants.key_order().map_err(ServeError::Registry)?;
116    if !order.is_sound() {
117        return Err(ServeError::KeysOutOfOrder(order));
118    }
119    let tls = api::tls_config(&config.tls_cert, &config.tls_key).map_err(ServeError::Tls)?;
120    let listener = TcpListener::bind(config.listen)
121        .await
122        .map_err(ServeError::Listen)?;
123    // The address bound, which names the port the system chose when the
124    // configuration gives port 0.
125    let bound = listener.local_addr().map_err(ServeError::Listen)?;
126    let tenants = Arc::new(tenants);
127    // The purge sweep: at start, then purge_interval after each sweep ends,
128    // until the server stops.
129    let (stop, stopped) = tokio::sync::oneshot::channel::<()>();
130    let (swept, default) = (tenants.clone(), config.retention_default);
131    let sweeper = tokio::spawn(every(
132        config.purge_interval,
133        async {
134            let _ = stopped.await;
135        },
136        move || {
137            let tenants = swept.clone();
138            async move {
139                let _ = tokio::task::spawn_blocking(move || sweep(&tenants, default)).await;
140            }
141        },
142    ));
143    // The rotation warning: at start, then each day.
144    let (stop_warning, warning_stopped) = tokio::sync::oneshot::channel::<()>();
145    let (watched, rotation_warn) = (tenants.clone(), config.rotation_warn);
146    let warner = tokio::spawn(every(
147        ROTATION_CHECK,
148        async {
149            let _ = warning_stopped.await;
150        },
151        move || {
152            let tenants = watched.clone();
153            async move {
154                let _ = tokio::task::spawn_blocking(move || {
155                    warn_if_rotation_overdue(&tenants, rotation_warn);
156                })
157                .await;
158            }
159        },
160    ));
161    let app = app_from(config, tenants);
162    tracing::info!(listen = %bound, "serving");
163    let served = api::serve(listener, tls, api::router(Arc::new(app)), shutdown).await;
164    let _ = stop.send(());
165    let _ = stop_warning.send(());
166    let _ = sweeper.await;
167    let _ = warner.await;
168    served.map_err(ServeError::Serve)?;
169    tracing::info!("stopped");
170    Ok(())
171}
172
173/// What one purge sweep did.
174#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
175pub struct SweepReport {
176    /// Versions purged, over every tenant.
177    pub purged: u64,
178    /// Tenants that could not be swept; they wait for the next sweep.
179    pub failed: u64,
180}
181
182/// One purge sweep: each tenant purges what is due under its own retention
183/// period, else `default`. A tenant that cannot be swept, such as one whose
184/// period does not read back or that is being removed, is counted and left
185/// for the next sweep. The sweep is logged with counts only, never a tenant.
186pub fn sweep(tenants: &Tenants, default: Retention) -> SweepReport {
187    let mut report = SweepReport::default();
188    match tenants.ids() {
189        Ok(ids) => {
190            for id in ids {
191                let purged = tenants.retention(id).and_then(|own| {
192                    let store = tenants.store(id)?;
193                    Ok(store.purge_due(crate::tenants::kept_for(own, default)))
194                });
195                match purged {
196                    Ok(Ok(done)) => report.purged += done.purged,
197                    _ => report.failed += 1,
198                }
199            }
200        }
201        Err(_) => report.failed += 1,
202    }
203    tracing::info!(
204        purged = report.purged,
205        failed = report.failed,
206        "purge sweep"
207    );
208    report
209}
210
211/// Runs `job` at once, then again `interval` after each run ends, until
212/// `shutdown` completes; a run still going when it does is not awaited.
213pub async fn every<F, Fut>(interval: Duration, shutdown: impl Future<Output = ()>, mut job: F)
214where
215    F: FnMut() -> Fut,
216    Fut: Future<Output = ()>,
217{
218    tokio::pin!(shutdown);
219    loop {
220        tokio::select! {
221            biased;
222            () = &mut shutdown => return,
223            () = job() => {}
224        }
225        tokio::select! {
226            biased;
227            () = &mut shutdown => return,
228            () = tokio::time::sleep(interval) => {}
229        }
230    }
231}
232
233/// What the API serves from, as the configuration sets it.
234fn app_from(config: &Config, tenants: Arc<Tenants>) -> App {
235    App {
236        tenants,
237        limits: Limits {
238            body: config.body_limit,
239            inflated: config.inflated_limit,
240            body_timeout: config.read_timeout,
241        },
242        supported: config.supported_critical.clone(),
243        retention_default: config.retention_default,
244        mcp: config.mcp,
245    }
246}
247
248/// Whether key rotation is overdue: the oldest purge or removal waiting
249/// for it is older than `rotation_warn`.
250pub fn rotation_overdue(
251    backlog: &RotationBacklog,
252    rotation_warn: Duration,
253    now: OffsetDateTime,
254) -> bool {
255    [backlog.oldest_purge, backlog.oldest_removal]
256        .into_iter()
257        .flatten()
258        .any(|at| now - at > rotation_warn)
259}
260
261/// How often `serve` looks at the rotation backlog.
262const ROTATION_CHECK: Duration = Duration::from_secs(86_400);
263
264/// Logs a warning, with counts only, when key rotation is overdue.
265fn warn_if_rotation_overdue(tenants: &Tenants, rotation_warn: Duration) {
266    match tenants.rotation_backlog() {
267        Ok(backlog) if rotation_overdue(&backlog, rotation_warn, SystemClock.now()) => {
268            tracing::warn!(
269                tenants_with_purges = backlog.tenants_with_purges,
270                purges = backlog.purges,
271                removals = backlog.removals,
272                "key rotation overdue: purged vCons and removed tenants stay readable on old copies until it"
273            );
274        }
275        Ok(_) => {}
276        Err(_) => tracing::warn!("the key rotation backlog could not be read"),
277    }
278}
279
280#[cfg(test)]
281mod tests {
282    use std::time::Duration;
283
284    use super::*;
285
286    mod startup {
287        use time::macros::datetime;
288
289        use super::*;
290
291        const NOW: OffsetDateTime = datetime!(2026-10-25 12:00 UTC);
292        const DAY: Duration = Duration::from_secs(86_400);
293
294        #[test]
295        fn rotation_is_overdue_once_the_oldest_waiting_work_passes_rotation_warn() {
296            let backlog =
297                |purge: Option<OffsetDateTime>, removal: Option<OffsetDateTime>| RotationBacklog {
298                    tenants_with_purges: u64::from(purge.is_some()),
299                    purges: u64::from(purge.is_some()),
300                    oldest_purge: purge,
301                    removals: u64::from(removal.is_some()),
302                    oldest_removal: removal,
303                };
304            let day_ago = Some(datetime!(2026-10-24 12:00 UTC));
305            let older = Some(datetime!(2026-10-24 11:59:59 UTC));
306            for (b, overdue) in [
307                (backlog(None, None), false),
308                (backlog(day_ago, None), false),
309                (backlog(older, None), true),
310                (backlog(None, older), true),
311                (backlog(day_ago, older), true),
312            ] {
313                assert_eq!(rotation_overdue(&b, DAY, NOW), overdue, "{b:?}");
314            }
315        }
316
317        #[test]
318        fn a_refusal_for_keys_out_of_order_counts_and_points_at_check() {
319            let order = KeyOrder {
320                wrapped_elsewhere: 1,
321                rotations_stopped: 2,
322                data_under_other_keys: 3,
323                ..KeyOrder::default()
324            };
325            assert_eq!(
326                ServeError::KeysOutOfOrder(order).to_string(),
327                "1 tenant key(s) do not unwrap under the master key; \
328                 2 key rotation(s) stopped part way; run rekey tenant again; \
329                 3 tenant(s) with data under another key; \
330                 vcon-server check names the tenants",
331                "each count in its fixed place, then where to look"
332            );
333        }
334
335        /// Catalogs the server may not open are counted apart from keys,
336        /// with the system's error, and never a tenant or a path.
337        #[test]
338        fn a_refusal_for_catalogs_it_may_not_open_gives_the_error_not_a_key() {
339            let order = KeyOrder {
340                catalogs_unopened: 2,
341                unopened_error: Some("Permission denied (os error 13)".to_owned()),
342                ..KeyOrder::default()
343            };
344            assert_eq!(
345                ServeError::KeysOutOfOrder(order).to_string(),
346                "2 tenant catalog(s) cannot be opened: Permission denied (os error 13); vcon-server check names the tenants"
347            );
348        }
349    }
350    use crate::retention::Retention;
351
352    #[test]
353    fn the_api_takes_its_limits_and_extensions_from_the_configuration() {
354        let dir = tempfile::tempdir().expect("temp dir");
355        let tenants = Arc::new(
356            Tenants::open(
357                dir.path(),
358                Key::generate().expect("key"),
359                Arc::new(OsFs),
360                Arc::new(SystemClock),
361            )
362            .expect("opens"),
363        );
364        let config = Config {
365            listen: "127.0.0.1:8443".parse().expect("address"),
366            tls_cert: "cert.pem".into(),
367            tls_key: "key.pem".into(),
368            data_dir: dir.path().to_owned(),
369            master_key: "master.key".into(),
370            retention_default: Retention::Months(18),
371            purge_interval: Duration::from_secs(3600),
372            rotation_warn: Duration::from_secs(86400),
373            body_limit: 12_345,
374            inflated_limit: 67_890,
375            read_timeout: Duration::from_secs(17),
376            supported_critical: ["known".to_owned()].into(),
377            mcp: crate::mcp::Limits {
378                calls_per_second: 3,
379                concurrent: 4,
380                response_bytes: 5,
381            },
382        };
383        let app = app_from(&config, tenants.clone());
384        assert!(Arc::ptr_eq(&app.tenants, &tenants));
385        assert_eq!(app.limits.body, 12_345);
386        assert_eq!(app.limits.inflated, 67_890);
387        assert_eq!(app.limits.body_timeout, Duration::from_secs(17));
388        assert_eq!(app.supported, config.supported_critical);
389        assert_eq!(app.retention_default, config.retention_default);
390        assert_eq!(app.mcp, config.mcp);
391    }
392
393    mod sweeps {
394        use std::sync::Arc;
395        use std::sync::atomic::{AtomicUsize, Ordering};
396        use std::time::Duration;
397
398        use time::macros::datetime;
399        use uuid::Uuid;
400
401        use crate::audit::Actor;
402        use crate::ids::{KeyId, TenantId};
403        use crate::retention::Retention;
404        use crate::serve::{SweepReport, every, sweep};
405        use crate::store::Caller;
406        use crate::store::testing::vcon;
407        use crate::tenants::{Tenants, fixture};
408
409        const CALLER: Caller = Caller { key: KeyId([7; 8]) };
410
411        /// A vCon stored and deleted in `id` while seven years apply, so
412        /// it waits hidden for a sweep.
413        fn deleted(tenants: &Tenants, id: TenantId, uuid: &str) -> Uuid {
414            let store = tenants.store(id).expect("its store");
415            let (found, body) = vcon(uuid, "kept");
416            store.create(&CALLER, &found, &body).expect("stored");
417            store
418                .delete(&CALLER, found.uuid, Retention::Years(7))
419                .expect("deleted");
420            found.uuid
421        }
422
423        /// How many versions `id` still keeps.
424        fn kept(tenants: &Tenants, id: TenantId) -> u64 {
425            tenants
426                .store(id)
427                .expect("its store")
428                .counts()
429                .expect("counts")
430                .versions
431        }
432
433        #[test]
434        fn a_sweep_purges_what_is_due_in_each_tenant_under_its_own_period() {
435            let f = fixture();
436            let own = f
437                .tenants
438                .create(Actor::Cli, "own", Some(Retention::Days(1)))
439                .expect("made");
440            let default = f.tenants.create(Actor::Cli, "default", None).expect("made");
441            let long = f
442                .tenants
443                .create(Actor::Cli, "long", Some(Retention::Years(7)))
444                .expect("made");
445            deleted(&f.tenants, own, "0199e5c6-0d7a-7c11-8000-000000000001");
446            deleted(&f.tenants, default, "0199e5c6-0d7a-7c11-8000-000000000002");
447            deleted(&f.tenants, long, "0199e5c6-0d7a-7c11-8000-000000000003");
448            f.clock.set(datetime!(2026-10-25 13:00 UTC));
449
450            let report = sweep(&f.tenants, Retention::Days(1));
451
452            assert_eq!(
453                report,
454                SweepReport {
455                    purged: 2,
456                    failed: 0
457                }
458            );
459            assert_eq!(kept(&f.tenants, own), 0);
460            assert_eq!(kept(&f.tenants, default), 0);
461            assert_eq!(kept(&f.tenants, long), 1, "seven years have not passed");
462        }
463
464        #[test]
465        fn a_sweep_goes_on_past_a_tenant_it_cannot_sweep() {
466            let f = fixture();
467            let damaged = f.tenants.create(Actor::Cli, "damaged", None).expect("made");
468            let fine = f.tenants.create(Actor::Cli, "fine", None).expect("made");
469            deleted(&f.tenants, damaged, "0199e5c6-0d7a-7c11-8000-000000000001");
470            deleted(&f.tenants, fine, "0199e5c6-0d7a-7c11-8000-000000000002");
471            f.tenants
472                .conn_for_test()
473                .execute(
474                    "UPDATE tenant SET retention = '7w' WHERE id = ?1",
475                    [damaged.0],
476                )
477                .expect("damaged");
478            f.clock.set(datetime!(2026-10-25 13:00 UTC));
479
480            let report = sweep(&f.tenants, Retention::Days(1));
481
482            assert_eq!(
483                report,
484                SweepReport {
485                    purged: 1,
486                    failed: 1
487                }
488            );
489            assert_eq!(kept(&f.tenants, fine), 0);
490            assert_eq!(kept(&f.tenants, damaged), 1, "left for a later sweep");
491        }
492
493        #[tokio::test(start_paused = true)]
494        async fn every_runs_at_start_then_after_each_interval_until_shutdown() {
495            let runs = Arc::new(AtomicUsize::new(0));
496            let (stop, stopped) = tokio::sync::oneshot::channel::<()>();
497            let counted = runs.clone();
498            let task = tokio::spawn(every(
499                Duration::from_secs(60),
500                async move {
501                    let _ = stopped.await;
502                },
503                move || {
504                    counted.fetch_add(1, Ordering::SeqCst);
505                    async {}
506                },
507            ));
508            let settle = || async {
509                for _ in 0..10 {
510                    tokio::task::yield_now().await;
511                }
512            };
513
514            settle().await;
515            assert_eq!(runs.load(Ordering::SeqCst), 1, "at start");
516            tokio::time::advance(Duration::from_secs(59)).await;
517            settle().await;
518            assert_eq!(runs.load(Ordering::SeqCst), 1, "not before the interval");
519            tokio::time::advance(Duration::from_secs(1)).await;
520            settle().await;
521            assert_eq!(runs.load(Ordering::SeqCst), 2, "after the interval");
522
523            stop.send(()).expect("still running");
524            task.await.expect("ends on shutdown");
525            tokio::time::advance(Duration::from_secs(600)).await;
526            assert_eq!(runs.load(Ordering::SeqCst), 2, "none after shutdown");
527        }
528    }
529}
vcon_server/serve.rs

vcon_server/
serve.rs
