apeirogond 0.11.0

Daemon for Apeirogon, a cross-platform software KVM
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
93
94
95
96
97
98
99
100
101
102
103

// SPDX-License-Identifier: GPL-3.0-or-later

use std::path::PathBuf;
use std::sync::{Arc, atomic::Ordering};

use a7n_core::files::ValidManifest;
use a7n_proto::limits;
use quinn::{Connection, RecvStream};
use tokio::sync::{mpsc, oneshot, watch};
use tokio::time::timeout;

use super::{Event, Failure, Progress, Result, staging};
use crate::daemon::clipboard::fetch::read;

mod verify;
pub(super) use verify::apply;

pub(super) struct Receive {
    pub id: u64,
    pub manifest: ValidManifest,
    pub stage: PathBuf,
    pub destination: PathBuf,
    pub receiver: oneshot::Receiver<(RecvStream, Vec<u8>)>,
    pub cancelled: watch::Receiver<Failure>,
    pub progress: Arc<Progress>,
    pub events: mpsc::Sender<Event>,
    pub connection: Connection,
}

pub(super) async fn run(mut job: Receive) -> Result<Vec<PathBuf>> {
    let result = tokio::select! {
        biased;
        _ = job.cancelled.changed() => Err(*job.cancelled.borrow()),
        result = download(&job.manifest, &job.stage, &job.destination, &mut job.receiver, &job.progress, &job.events, job.id) => result,
    };
    if result.is_err() && result != Err(Failure::LinkLost) {
        let _ = std::fs::remove_dir_all(&job.stage);
    }
    if result == Err(Failure::Invalid) {
        // Malformed transfer framing closes the link, as for clipboard fetches.
        job.connection.close(0u32.into(), b"");
    }
    result
}

async fn download(
    manifest: &ValidManifest,
    stage: &std::path::Path,
    receive_dir: &std::path::Path,
    receiver: &mut oneshot::Receiver<(RecvStream, Vec<u8>)>,
    progress: &Progress,
    events: &mpsc::Sender<Event>,
    id: u64,
) -> Result<Vec<PathBuf>> {
    let mut partial = staging::prepare(stage, manifest)?;
    progress.bytes.store(
        partial
            .files
            .iter()
            .flatten()
            .map(|file| file.written)
            .sum(),
        Ordering::Relaxed,
    );
    events
        .send(Event::FilesReady {
            id,
            resume: partial.resume.clone(),
        })
        .await
        .map_err(|_| Failure::LinkLost)?;
    let (mut stream, mut buf) = timeout(limits::FRAME_READ_TIMEOUT, receiver)
        .await
        .map_err(|_| Failure::LinkLost)?
        .map_err(|_| Failure::LinkLost)?;
    loop {
        let frame = timeout(limits::FRAME_READ_TIMEOUT, read(&mut stream, &mut buf))
            .await
            .map_err(|_| Failure::LinkLost)?
            .map_err(|reason| {
                if reason == "frame" {
                    Failure::Invalid
                } else {
                    Failure::LinkLost
                }
            })?
            .ok_or(Failure::LinkLost)?;
        if apply(&mut partial, manifest, frame, progress)? {
            break;
        }
    }
    if timeout(limits::FRAME_READ_TIMEOUT, read(&mut stream, &mut buf))
        .await
        .map_err(|_| Failure::LinkLost)?
        .map_err(|_| Failure::LinkLost)?
        .is_some()
    {
        return Err(Failure::Invalid);
    }
    // Close all file handles before publishing, including on Windows.
    drop(partial);
    staging::publish(stage, receive_dir, manifest)
}