spill 0.5.0

A single-user BitTorrent daemon for Linux and macOS
Documentation 
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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

//! BEP 11 per-connection state. Untrusted additions are bounded; dropped hints
//! never disconnect peers. Output reports connection deltas at most once/minute.
use crate::engine::core::extensions::{invalid, limits};
use crate::engine::network::SocketAddr;
use crate::protocol::{
    bencode::{self, Node, Value},
    compact,
    peer::pex,
};
use std::{
    collections::BTreeMap,
    io,
    time::{Duration, Instant},
};

pub(in crate::engine) const ID: u8 = 2;
const INTERVAL: Duration = Duration::from_secs(60);
const UPDATE: usize = 50;
const INITIAL: usize = 200;

#[derive(Default)]
pub(in crate::engine) struct Pex {
    pub(in crate::engine) enabled: bool,
    pub(in crate::engine) endpoint: Option<SocketAddr>,
    pub(in crate::engine) local_ip: Option<crate::engine::network::IpAddr>,
    pub(in crate::engine) inbound: bool,
    pub(in crate::engine) received: Vec<SocketAddr>,
    received_at: Option<Instant>,
    sent_at: Option<Instant>,
    advertised: Vec<SocketAddr>,
}
impl Pex {
    pub(in crate::engine) fn receive(
        &mut self,
        payload: &[u8],
        capacity: usize,
        now: Instant,
    ) -> io::Result<()> {
        let cap = if self.received_at.is_none() {
            INITIAL
        } else {
            UPDATE
        };
        // The bencode byte budget bounds parsing; peer counts only bound the
        // hints retained below, not whether a valid sender stays connected.
        let message =
            pex::Message::parse(payload, limits(), usize::MAX, usize::MAX).map_err(invalid)?;
        if self
            .received_at
            .is_some_and(|at| now.saturating_duration_since(at) < INTERVAL)
        {
            return Ok(());
        }
        self.received_at = Some(now);
        for address in message
            .added_v4()
            .iter()
            .map(SocketAddr::V4)
            .chain(message.added_v6().iter().map(SocketAddr::V6))
            .take(cap)
        {
            if self.received.len() >= capacity.min(INITIAL) {
                break;
            }
            if crate::engine::network::eligible(address) && !self.received.contains(&address) {
                self.received.push(address);
            }
        }
        Ok(())
    }
    pub(in crate::engine) fn due(&self, now: Instant) -> bool {
        self.sent_at
            .is_none_or(|at| now.saturating_duration_since(at) >= INTERVAL)
    }
    pub(in crate::engine) fn update(
        &self,
        peers: &[SocketAddr],
        now: Instant,
    ) -> io::Result<Option<Vec<u8>>> {
        if !self.due(now) {
            return Ok(None);
        }
        let added: Vec<_> = peers
            .iter()
            .copied()
            .filter(|p| !self.advertised.contains(p))
            .take(UPDATE)
            .collect();
        let dropped: Vec<_> = self
            .advertised
            .iter()
            .copied()
            .filter(|p| !peers.contains(p))
            .take(UPDATE)
            .collect();
        if added.is_empty() && dropped.is_empty() {
            return Ok(None);
        }
        let mut fields = BTreeMap::new();
        let mut buffers = [Vec::new(), Vec::new(), Vec::new(), Vec::new()];
        for (list, v4, v6) in [(&added, 0, 1), (&dropped, 2, 3)] {
            for address in list {
                match address {
                    SocketAddr::V4(a) => buffers[v4].extend_from_slice(&compact::encode_v4(*a)),
                    SocketAddr::V6(a) => {
                        buffers[v6].extend_from_slice(&compact::encode_v6(*a).map_err(invalid)?)
                    }
                }
            }
        }
        for (key, bytes) in [b"added".as_slice(), b"added6", b"dropped", b"dropped6"]
            .into_iter()
            .zip(&buffers)
        {
            fields.insert(key, Node::new(Value::Bytes(bytes)));
        }
        let bytes =
            bencode::encode(&Node::new(Value::Dictionary(fields)), limits()).map_err(invalid)?;
        Ok(Some(bytes))
    }
    // Commit only after output admission, so backpressure never loses changes.
    pub(in crate::engine) fn sent(&mut self, payload: &[u8], now: Instant) -> io::Result<()> {
        let message = pex::Message::parse(payload, limits(), UPDATE, UPDATE).map_err(invalid)?;
        let dropped: Vec<_> = message
            .dropped_v4()
            .iter()
            .map(SocketAddr::V4)
            .chain(message.dropped_v6().iter().map(SocketAddr::V6))
            .collect();
        self.advertised.retain(|p| !dropped.contains(p));
        self.advertised.extend(
            message
                .added_v4()
                .iter()
                .map(SocketAddr::V4)
                .chain(message.added_v6().iter().map(SocketAddr::V6)),
        );
        self.sent_at = Some(now);
        Ok(())
    }
    pub(in crate::engine) fn disabled(&mut self) {
        self.advertised.clear();
        self.received.clear();
    }
}