vcon_server/
serve.rs

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

vcon_server/
serve.rs
