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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
//! Peer connection state and wire-level bookkeeping.
//!
//! A [`Peer`] owns everything about one TCP connection: phase, availability,
//! extension state, rates, reputation (anti-leech ledger), and the outgoing
//! byte buffer. The choke/unchoke **algorithm** itself lives in
//! [`crate::leech`] (reciprocity scoring, snubs, bans); this module only
//! supplies the per-connection state it operates on.
use crate::bitfield::Bitfield;
use crate::consts::REQUEST_PIPELINE;
use crate::leech::PeerReputation;
use crate::monitoring::DiscoverySource;
use crate::platform::{ConnId, NetAddr};
use crate::wire::{ExtHandshake, Message, MessageStream};
use alloc::vec::Vec;
/// Lifecycle phase of a peer connection.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PeerPhase {
/// Connecting (non-blocking connect in progress).
Connecting,
/// Our handshake sent; waiting for theirs.
Handshake,
/// Handshake complete; exchanging bitfield/extension messages.
Ready,
/// Dead.
Closed,
}
/// Per-connection peer state.
#[derive(Debug)]
pub struct Peer {
/// Host connection handle.
pub id: ConnId,
/// Remote endpoint.
pub addr: NetAddr,
/// Lifecycle phase.
pub phase: PeerPhase,
/// Their peer id (after handshake).
pub peer_id: Option<[u8; 20]>,
/// Their reserved bits.
pub reserved: [u8; 8],
/// We are choking them.
pub am_choking: bool,
/// We are interested in them.
pub am_interested: bool,
/// They are choking us.
pub peer_choking: bool,
/// They are interested in us.
pub peer_interested: bool,
/// Their piece availability (empty if have_all/have_none).
pub have: Bitfield,
/// `have_all` received (fast extension).
pub have_all: bool,
/// `have_none` received.
pub have_none: bool,
/// Their extended-handshake data.
pub ext: Option<ExtHandshake>,
/// `ut_metadata` extended id (if any).
pub ext_metadata: Option<u8>,
/// `ut_pex` extended id (if any).
pub ext_pex: Option<u8>,
/// They are a seed (have everything).
pub is_seed: bool,
/// Supports fast extension.
pub fast: bool,
/// Supports DHT.
pub supports_dht: bool,
/// Supports v2 (BEP-52).
pub supports_v2: bool,
/// Bytes received total.
pub down_total: u64,
/// Bytes sent total.
pub up_total: u64,
/// Download rate (bytes/s, smoothed).
pub down_rate: u32,
/// Upload rate (bytes/s, smoothed).
pub up_rate: u32,
/// Timestamp of last inbound activity.
pub last_active: u64,
/// Timestamp of last data received (for snub detection).
pub last_data_in: u64,
/// Timestamp of connection start.
pub connected_at: u64,
/// Snubbed (no data for a while despite requests).
pub snubbed: bool,
/// This peer holds an optimistic-unchoke slot.
pub optimistic: bool,
/// Outstanding request count on this connection.
pub requests_in_flight: u32,
/// Last time they requested a block from us (ms; 0 = never). Drives
/// the idle-slot (bandwidth squatting) detection.
pub last_request_at: u64,
/// Blocks we have served them on this connection.
pub served_requests: u32,
/// Cancels for blocks we never had outstanding to them (abuse signal).
pub spurious_cancels: u32,
/// Structurally invalid requests (abuse signal).
pub invalid_requests: u32,
/// Outgoing byte buffer (engine drains via host).
pub out: Vec<u8>,
/// Incoming message stream.
pub msgs: MessageStream,
/// Handshake prefix buffer.
pub handshake_buf: Vec<u8>,
/// How this peer was found.
pub source: DiscoverySource,
/// Anti-leech reputation ledger (client, corrupt blocks, violations).
pub rep: PeerReputation,
/// Last PEX message sent.
pub pex_sent_at: u64,
/// Rate-sampling window accumulators.
pub window_down: u64,
/// Bytes uploaded in the current window.
pub window_up: u64,
/// When the current rate window started.
pub window_started: u64,
}
impl Peer {
/// Create a new peer (outbound or inbound).
pub fn new(id: ConnId, addr: NetAddr, piece_count: u32, source: DiscoverySource) -> Self {
Peer {
id,
addr,
phase: PeerPhase::Connecting,
peer_id: None,
reserved: [0u8; 8],
am_choking: true,
am_interested: false,
peer_choking: true,
peer_interested: false,
have: Bitfield::new(piece_count),
have_all: false,
have_none: false,
ext: None,
ext_metadata: None,
ext_pex: None,
is_seed: false,
fast: false,
supports_dht: false,
supports_v2: false,
down_total: 0,
up_total: 0,
down_rate: 0,
up_rate: 0,
last_active: 0,
last_data_in: 0,
connected_at: 0,
snubbed: false,
optimistic: false,
requests_in_flight: 0,
last_request_at: 0,
served_requests: 0,
spurious_cancels: 0,
invalid_requests: 0,
out: Vec::with_capacity(16 * 1024),
msgs: MessageStream::new(),
handshake_buf: Vec::new(),
source,
rep: PeerReputation::default(),
pex_sent_at: 0,
window_down: 0,
window_up: 0,
window_started: 0,
}
}
/// Whether we know the peer has piece `p`.
pub fn has_piece(&self, p: u32) -> bool {
self.have_all || self.have.get(p)
}
/// Whether we should be interested (they have something we lack).
pub fn should_be_interested(&self, our_have: &Bitfield) -> bool {
if self.peer_choking || self.have_none {
return false;
}
if self.have_all {
return !our_have.all_set();
}
// any piece they have that we don't
let n = self.have.len().min(our_have.len());
let mut i = self.have.first_set();
while let Some(p) = i {
if p >= n {
break;
}
if !our_have.get(p) {
return true;
}
i = self.have.next_set_from(p + 1);
}
false
}
/// Max request pipeline depth for this peer.
pub fn max_pipeline(&self) -> u32 {
REQUEST_PIPELINE
}
/// Queue a message for sending.
pub fn send(&mut self, m: &Message) {
self.out.extend_from_slice(&m.encode());
}
/// Whether the out buffer is empty.
pub fn flushed(&self) -> bool {
self.out.is_empty()
}
/// Reset the rate window (called on a fixed cadence).
pub fn roll_window(&mut self, now: u64) {
let span_ms = now.saturating_sub(self.window_started).max(1);
self.down_rate = ((self.window_down * 1000) / span_ms) as u32;
self.up_rate = ((self.window_up * 1000) / span_ms) as u32;
self.window_down = 0;
self.window_up = 0;
self.window_started = now;
}
/// Record downloaded bytes.
pub fn on_data_in(&mut self, n: usize, now: u64) {
self.down_total += n as u64;
self.window_down += n as u64;
self.last_active = now;
self.last_data_in = now;
self.snubbed = false;
}
/// Record uploaded bytes.
pub fn on_data_out(&mut self, n: usize, now: u64) {
self.up_total += n as u64;
self.window_up += n as u64;
self.last_active = now;
}
/// Refresh the snub flag: we are snubbed when we expect data from them
/// (they unchoked us and we have outstanding requests) but they stay
/// quiet for `timeout_ms`. The flag clears as soon as the situation
/// improves (data arrives or expectations end).
pub fn refresh_snub(&mut self, now: u64, timeout_ms: u64) {
self.snubbed = self.phase == PeerPhase::Ready
&& !self.peer_choking
&& self.requests_in_flight > 0
&& now.saturating_sub(self.last_data_in) > timeout_ms;
}
}
#[cfg(test)]
mod tests {
use super::*;
fn peer(id: ConnId) -> Peer {
let mut p = Peer::new(
id,
NetAddr::V4([127, 0, 0, 1], 6881),
64,
DiscoverySource::Tracker,
);
p.phase = PeerPhase::Ready;
p
}
#[test]
fn snub_detection_and_clear() {
let mut p = peer(1);
p.peer_choking = false;
p.requests_in_flight = 2;
p.last_data_in = 0;
p.refresh_snub(1000, 500);
assert!(p.snubbed);
// data arrives → cleared
p.on_data_in(16, 1001);
assert!(!p.snubbed);
// no outstanding requests → never snubbed
p.requests_in_flight = 0;
p.last_data_in = 0;
p.refresh_snub(2000, 500);
assert!(!p.snubbed);
}
#[test]
fn interested_detection() {
let mut p = peer(1);
p.peer_choking = false;
p.have.set(5);
let mut ours = Bitfield::new(64);
ours.set(5);
assert!(!p.should_be_interested(&ours));
ours.clear(5);
assert!(p.should_be_interested(&ours));
}
}