microsandbox 0.7.0

`microsandbox` is the core library for the microsandbox project.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

//! Best-effort creation telemetry. The creation result, not stream EOF, determines success.

use tokio::sync::mpsc;

pub use microsandbox_runtime::startup_progress::{StartupPhase, StartupProgress};

//--------------------------------------------------------------------------------------------------
// Types
//--------------------------------------------------------------------------------------------------

/// Progress from image preparation through VM activation.
#[derive(Clone, Debug, serde::Serialize)]
#[serde(tag = "kind", content = "progress", rename_all = "snake_case")]
pub enum CreationProgress {
    /// Existing per-layer image preparation progress.
    Pull(microsandbox_image::PullProgress),
    /// Runtime preparation or activation progress.
    Startup(StartupProgress),
}

/// Bounded progress stream. Dropping this handle does not cancel creation.
pub struct CreationProgressHandle {
    receiver: mpsc::Receiver<CreationProgress>,
}

//--------------------------------------------------------------------------------------------------
// Methods
//--------------------------------------------------------------------------------------------------

impl CreationProgressHandle {
    /// Receive the next event. Events may coalesce or be omitted for slow consumers.
    pub async fn recv(&mut self) -> Option<CreationProgress> {
        self.receiver.recv().await
    }
}

//--------------------------------------------------------------------------------------------------
// Functions
//--------------------------------------------------------------------------------------------------

pub(crate) fn channel() -> (CreationProgressHandle, mpsc::Sender<CreationProgress>) {
    let (sender, receiver) = mpsc::channel(256);
    (CreationProgressHandle { receiver }, sender)
}

pub(crate) fn report(
    observer: &Option<mpsc::WeakSender<CreationProgress>>,
    event: CreationProgress,
) {
    if let Some(sender) = observer.as_ref().and_then(mpsc::WeakSender::upgrade) {
        let _ = sender.try_send(event);
    }
}

//--------------------------------------------------------------------------------------------------
// Tests
//--------------------------------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn ignored_full_or_dropped_progress_never_blocks_the_producer() {
        let (mut events, sender) = channel();
        let weak = Some(sender.downgrade());
        for _ in 0..10000 {
            report(
                &weak,
                CreationProgress::Startup(StartupProgress::phase(StartupPhase::Activating)),
            );
        }
        assert!(events.recv().await.is_some());
        drop(events);
        report(
            &weak,
            CreationProgress::Startup(StartupProgress::phase(StartupPhase::Activating)),
        );
        drop(sender);
        assert!(weak.unwrap().upgrade().is_none());
    }

    #[test]
    fn startup_serialization_preserves_unknown_totals() {
        let event =
            CreationProgress::Startup(StartupProgress::phase(StartupPhase::SyncingMemoryBacking));
        let json = serde_json::to_value(event).unwrap();
        assert_eq!(json["kind"], "startup");
        assert_eq!(json["progress"]["phase"], "syncing_memory_backing");
        assert!(json["progress"]["total_bytes"].is_null());
    }
}