apeirogond 0.28.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

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

use a7n_proto::{ProtoError, v1::Envelope};

use super::*;

pub(crate) struct StalledStream {
    pub(crate) prefix: Option<Vec<u8>>,
    pub(crate) reads: usize,
}

impl FrameStream for StalledStream {
    async fn read_chunk(&mut self, max: usize) -> Result<Option<Chunk>, ReadError> {
        self.reads += 1;
        match self.prefix.take() {
            Some(prefix) => {
                assert!(prefix.len() <= max);
                Ok(Some(Chunk {
                    offset: 0,
                    bytes: prefix.into(),
                }))
            }
            None => std::future::pending().await,
        }
    }
}

#[tokio::test(start_paused = true)]
async fn read_oversized_or_stalled_frame_rejects_without_buffering_body() {
    // Arrange
    let duration = Duration::from_millis(20);
    let mut oversized = Vec::new();
    prost::encoding::encode_varint((FrameKind::Control.max_frame() + 1) as u64, &mut oversized);
    let cases = [(oversized, true), (vec![1], false)];
    for (prefix, too_large) in cases {
        let prefix_len = prefix.len();
        let mut stream = StalledStream {
            prefix: Some(prefix),
            reads: 0,
        };
        let mut buf = Vec::new();

        // Act
        let result = read::<Envelope>(
            &mut stream,
            &mut buf,
            FrameKind::Control,
            ChunkBound::Fixed(FrameKind::Control.max_frame()),
            Some(duration),
        )
        .await;

        // Assert
        if too_large {
            assert!(matches!(
                result,
                Err(LinkError::Proto(ProtoError::FrameTooLarge { .. }))
            ));
            assert_eq!(stream.reads, 1);
        } else {
            assert!(matches!(result, Err(LinkError::Timeout)));
        }
        assert_eq!(buf.len(), prefix_len);
        assert!(buf.capacity() < FrameKind::Control.max_frame());
    }
}

#[tokio::test(start_paused = true)]
async fn read_waiting_for_first_byte_keeps_caller_deadline() {
    // Arrange
    let duration = Duration::from_millis(20);
    let caller_deadline = duration * 2;
    let mut stream = StalledStream {
        prefix: None,
        reads: 0,
    };
    let mut buf = Vec::new();

    // Act
    let result = timeout(
        caller_deadline,
        read::<Envelope>(
            &mut stream,
            &mut buf,
            FrameKind::Control,
            ChunkBound::Fixed(usize::MAX),
            Some(duration),
        ),
    )
    .await;

    // Assert
    assert!(result.is_err());
    assert_eq!(buf, [] as [u8; 0]);
}