vector_core/
blossom_stats.rs

1//! Per-server latency, speed and reachability, and the one-word status the
2//! list shows.
3//!
4//! One row per server, no history: latency and speed are exponentially
5//! weighted averages, the last few upload speeds ride along as a short array
6//! for a sparkline. The footprint does not grow with use.
7//!
8//! Latency is the BUD-06 preflight round-trip — a small request, so it is
9//! dominated by distance and server load and comparable between servers.
10//! Speed is this device's throughput to the server, which the uplink caps:
11//! shown as a number, never graded.
12
13use serde::{Deserialize, Serialize};
14
15/// Weight of a new sample against the running average.
16const ALPHA: f64 = 0.3;
17/// Below this an upload is all round-trip and says nothing about throughput.
18pub const MIN_SPEED_SAMPLE_BYTES: u64 = 256 * 1024;
19/// Sparkline length.
20const RECENT_MAX: usize = 12;
21/// A failure this old with nothing since no longer means the server is down.
22const OFFLINE_WINDOW_SECS: i64 = 30 * 60;
23/// A quota refusal is good for the rest of the day at most.
24const MAXED_WINDOW_SECS: i64 = 24 * 3600;
25
26pub const FAIL_OFFLINE: &str = "offline";
27pub const FAIL_MAXED: &str = "maxed";
28
29/// One recent upload: its speed and its size, stored as `[mbps, bytes]`.
30/// Rows written before sizes were kept hold a bare number and read back with
31/// `bytes: 0`.
32#[derive(Clone, Copy, Debug, PartialEq)]
33pub struct Sample {
34    pub mbps: f32,
35    pub bytes: u64,
36}
37
38impl Serialize for Sample {
39    fn serialize<S: serde::Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
40        (self.mbps, self.bytes).serialize(s)
41    }
42}
43
44impl<'de> Deserialize<'de> for Sample {
45    fn deserialize<D: serde::Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
46        #[derive(Deserialize)]
47        #[serde(untagged)]
48        enum Shape {
49            Pair(f32, u64),
50            Bare(f32),
51        }
52        Ok(match Shape::deserialize(d)? {
53            Shape::Pair(mbps, bytes) => Sample { mbps, bytes },
54            Shape::Bare(mbps) => Sample { mbps, bytes: 0 },
55        })
56    }
57}
58
59#[derive(Clone, Debug, Serialize, PartialEq)]
60pub struct ServerStats {
61    pub latency_ms: Option<f64>,
62    pub mbps: Option<f64>,
63    pub best_mbps: Option<f64>,
64    pub uploads: u64,
65    pub bytes_total: u64,
66    pub last_ok_at: Option<i64>,
67    pub last_fail_at: Option<i64>,
68    pub last_fail_kind: Option<String>,
69    /// Recent uploads, oldest first.
70    pub recent: Vec<Sample>,
71}
72
73/// The list's one-word answer to "will my next upload here work?".
74#[derive(Clone, Debug, Serialize, PartialEq, Eq)]
75pub struct ServerStatus {
76    /// `online` · `offline` · `maxed` · `no_access` · `disabled` · `untested`
77    pub state: &'static str,
78    pub label: &'static str,
79    /// A `.relay-status` modifier: `connected` · `connecting` · `disconnected` · `disabled`
80    pub tone: &'static str,
81    pub latency_ms: Option<u32>,
82}
83
84fn now_secs() -> i64 {
85    web_time::SystemTime::now()
86        .duration_since(web_time::UNIX_EPOCH)
87        .map(|d| d.as_secs() as i64)
88        .unwrap_or(0)
89}
90
91fn norm_url(url: &str) -> String {
92    url.trim().trim_end_matches('/').to_lowercase()
93}
94
95fn ewma(prev: Option<f64>, sample: f64) -> f64 {
96    match prev {
97        Some(p) => p + ALPHA * (sample - p),
98        None => sample,
99    }
100}
101
102// ============================================================================
103// Storage
104// ============================================================================
105
106pub fn migrate(tx: &rusqlite::Transaction<'_>) -> Result<(), String> {
107    tx.execute_batch(
108        "CREATE TABLE IF NOT EXISTS blossom_server_stats (
109            server_url     TEXT    PRIMARY KEY,
110            latency_ms     REAL,
111            mbps           REAL,
112            best_mbps      REAL,
113            uploads        INTEGER NOT NULL DEFAULT 0,
114            bytes_total    INTEGER NOT NULL DEFAULT 0,
115            last_ok_at     INTEGER,
116            last_fail_at   INTEGER,
117            last_fail_kind TEXT,
118            recent         TEXT    NOT NULL DEFAULT '[]',
119            updated_at     INTEGER NOT NULL
120        );",
121    )
122    .map_err(|e| format!("Failed to create blossom_server_stats table: {}", e))
123}
124
125fn read_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<ServerStats> {
126    let recent_json: String = row.get(8)?;
127    Ok(ServerStats {
128        latency_ms: row.get(0)?,
129        mbps: row.get(1)?,
130        best_mbps: row.get(2)?,
131        uploads: row.get::<_, i64>(3)? as u64,
132        bytes_total: row.get::<_, i64>(4)? as u64,
133        last_ok_at: row.get(5)?,
134        last_fail_at: row.get(6)?,
135        last_fail_kind: row.get(7)?,
136        recent: serde_json::from_str(&recent_json).unwrap_or_default(),
137    })
138}
139
140const SELECT: &str = "SELECT latency_ms, mbps, best_mbps, uploads, bytes_total, last_ok_at, last_fail_at, last_fail_kind, recent
141                        FROM blossom_server_stats WHERE server_url = ?1";
142
143pub fn stats_for(server_url: &str) -> Option<ServerStats> {
144    let conn = crate::db::get_db_connection_guard_static().ok()?;
145    conn.query_row(SELECT, rusqlite::params![norm_url(server_url)], read_row).ok()
146}
147
148fn load_or_default(conn: &rusqlite::Connection, server: &str) -> ServerStats {
149    conn.query_row(SELECT, rusqlite::params![server], read_row)
150        .unwrap_or(ServerStats {
151            latency_ms: None,
152            mbps: None,
153            best_mbps: None,
154            uploads: 0,
155            bytes_total: 0,
156            last_ok_at: None,
157            last_fail_at: None,
158            last_fail_kind: None,
159            recent: Vec::new(),
160        })
161}
162
163fn save(conn: &rusqlite::Connection, server: &str, s: &ServerStats) -> Result<(), String> {
164    conn.execute(
165        "INSERT INTO blossom_server_stats
166            (server_url, latency_ms, mbps, best_mbps, uploads, bytes_total, last_ok_at, last_fail_at, last_fail_kind, recent, updated_at)
167         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
168         ON CONFLICT(server_url) DO UPDATE SET
169            latency_ms = ?2, mbps = ?3, best_mbps = ?4, uploads = ?5, bytes_total = ?6,
170            last_ok_at = ?7, last_fail_at = ?8, last_fail_kind = ?9, recent = ?10, updated_at = ?11",
171        rusqlite::params![
172            server, s.latency_ms, s.mbps, s.best_mbps, s.uploads as i64, s.bytes_total as i64,
173            s.last_ok_at, s.last_fail_at, s.last_fail_kind,
174            serde_json::to_string(&s.recent).unwrap_or_else(|_| "[]".to_string()),
175            now_secs(),
176        ],
177    )
178    .map_err(|e| format!("Failed to save server stats: {}", e))?;
179    Ok(())
180}
181
182fn update(server_url: &str, f: impl FnOnce(&mut ServerStats)) {
183    let Ok(conn) = crate::db::get_write_connection_guard_static() else { return };
184    let server = norm_url(server_url);
185    let mut s = load_or_default(&conn, &server);
186    f(&mut s);
187    if let Err(e) = save(&conn, &server, &s) {
188        crate::log_warn!("[Blossom Stats] {}: {}", server, e);
189    }
190    drop(conn);
191    // The list's pill is derived from this row.
192    crate::traits::emit_event("blossom_status_updated", &());
193}
194
195/// A small request answered in `ms`.
196pub fn record_latency(server_url: &str, ms: f64) {
197    if ms <= 0.0 { return; }
198    update(server_url, |s| {
199        s.latency_ms = Some(ewma(s.latency_ms, ms));
200        s.last_ok_at = Some(now_secs());
201    });
202}
203
204/// `bytes` accepted in `ms`. Small uploads are ignored: they measure the
205/// round-trip, not the pipe.
206pub fn record_upload(server_url: &str, bytes: u64, ms: f64) {
207    if bytes < MIN_SPEED_SAMPLE_BYTES || ms <= 0.0 { return; }
208    let mbps = (bytes as f64 * 8.0) / (ms / 1000.0) / 1_000_000.0;
209    update(server_url, |s| {
210        s.mbps = Some(ewma(s.mbps, mbps));
211        s.best_mbps = Some(s.best_mbps.map_or(mbps, |b| b.max(mbps)));
212        s.uploads += 1;
213        s.bytes_total += bytes;
214        s.last_ok_at = Some(now_secs());
215        s.recent.push(Sample { mbps: mbps as f32, bytes });
216        if s.recent.len() > RECENT_MAX {
217            let drop = s.recent.len() - RECENT_MAX;
218            s.recent.drain(..drop);
219        }
220    });
221}
222
223/// The server answered, whatever it said: it is reachable.
224pub fn record_ok(server_url: &str) {
225    update(server_url, |s| s.last_ok_at = Some(now_secs()));
226}
227
228/// `kind` is [`FAIL_OFFLINE`] or [`FAIL_MAXED`].
229pub fn record_failure(server_url: &str, kind: &str) {
230    update(server_url, |s| {
231        s.last_fail_at = Some(now_secs());
232        s.last_fail_kind = Some(kind.to_string());
233    });
234}
235
236// ============================================================================
237// Status
238// ============================================================================
239
240/// Pure: what the list should say, from what is known. `info` is the
241/// server's document when it has one; `stats` this device's history.
242pub fn derive_status(
243    enabled: bool,
244    info: Option<&crate::blossom_info::ServerInfo>,
245    stats: Option<&ServerStats>,
246    now: i64,
247) -> ServerStatus {
248    let latency_ms = stats.and_then(|s| s.latency_ms).map(|l| l.round() as u32);
249    let status = |state: &'static str, label: &'static str, tone: &'static str| ServerStatus {
250        state, label, tone, latency_ms,
251    };
252    if !enabled {
253        return status("disabled", "Disabled", "disabled");
254    }
255    if let Some(c) = info.and_then(|i| i.caller.as_ref()) {
256        if !c.allowed {
257            return status("no_access", "No access", "disconnected");
258        }
259        let full = |used: u64, limit: u64| limit > 0 && used >= limit;
260        if full(c.storage_used, c.storage_limit) || full(c.daily_used, c.daily_limit) {
261            return status("maxed", "Maxed", "connecting");
262        }
263    }
264    if let Some(s) = stats {
265        let failed_since_ok = match (s.last_fail_at, s.last_ok_at) {
266            (Some(f), Some(o)) => f > o,
267            (Some(_), None) => true,
268            _ => false,
269        };
270        if failed_since_ok {
271            let age = now - s.last_fail_at.unwrap_or(now);
272            match s.last_fail_kind.as_deref() {
273                Some(FAIL_MAXED) if age <= MAXED_WINDOW_SECS => {
274                    return status("maxed", "Maxed", "connecting");
275                }
276                Some(FAIL_OFFLINE) if age <= OFFLINE_WINDOW_SECS => {
277                    return status("offline", "Offline", "disconnected");
278                }
279                _ => {}
280            }
281        }
282        if s.last_ok_at.is_some() {
283            return status("online", "Online", "connected");
284        }
285    }
286    if info.is_some() {
287        return status("online", "Online", "connected");
288    }
289    status("untested", "Untested", "disabled")
290}
291
292pub fn status_for(server_url: &str, enabled: bool) -> ServerStatus {
293    let info = crate::blossom_info::cached(server_url);
294    let stats = stats_for(server_url);
295    derive_status(enabled, info.as_ref(), stats.as_ref(), now_secs())
296}
297
298#[cfg(test)]
299mod tests {
300    use super::*;
301    use crate::blossom_info::{CallerInfo, ServerInfo};
302
303    fn stats() -> ServerStats {
304        ServerStats {
305            latency_ms: Some(48.4), mbps: Some(12.0), best_mbps: Some(20.0), uploads: 3,
306            bytes_total: 3_000_000, last_ok_at: Some(1000), last_fail_at: None,
307            last_fail_kind: None,
308            recent: vec![Sample { mbps: 10.0, bytes: 0 }, Sample { mbps: 12.0, bytes: 2_000_000 }],
309        }
310    }
311
312    fn doc(caller: Option<CallerInfo>) -> ServerInfo {
313        ServerInfo {
314            name: None, software: None, version: None, description: None, buds: vec![],
315            extensions: vec![], max_blob: None, mime: vec![], capacity_used: None,
316            capacity_limit: None, caller, fetched_at: 0, raw: serde_json::Value::Null,
317        }
318    }
319
320    fn caller(allowed: bool, storage: (u64, u64), daily: (u64, u64)) -> CallerInfo {
321        CallerInfo {
322            tier: "recognised".into(), max_blob: 1, storage_used: storage.0, storage_limit: storage.1,
323            blobs: 0, daily_used: daily.0, daily_limit: daily.1, allowed, reasons: vec![],
324            plan_title: None, plan_description: None,
325        }
326    }
327
328    #[test]
329    fn disabled_wins_over_everything() {
330        let s = derive_status(false, Some(&doc(Some(caller(false, (0, 1), (0, 1))))), Some(&stats()), 2000);
331        assert_eq!(s.state, "disabled");
332        assert_eq!(s.latency_ms, Some(48), "the number still rides along");
333    }
334
335    #[test]
336    fn the_document_decides_access_and_quota() {
337        assert_eq!(derive_status(true, Some(&doc(Some(caller(false, (0, 1), (0, 1))))), None, 0).state, "no_access");
338        assert_eq!(derive_status(true, Some(&doc(Some(caller(true, (5, 5), (0, 1))))), None, 0).state, "maxed");
339        assert_eq!(derive_status(true, Some(&doc(Some(caller(true, (0, 5), (1, 1))))), None, 0).state, "maxed");
340        assert_eq!(derive_status(true, Some(&doc(Some(caller(true, (0, 5), (0, 1))))), None, 0).state, "online");
341        // A zero limit is "no limit", not "full".
342        assert_eq!(derive_status(true, Some(&doc(Some(caller(true, (5, 0), (0, 0))))), None, 0).state, "online");
343    }
344
345    #[test]
346    fn a_recent_failure_with_nothing_since_is_offline_then_forgotten() {
347        let mut s = stats();
348        s.last_fail_at = Some(1500);
349        s.last_fail_kind = Some(FAIL_OFFLINE.into());
350        assert_eq!(derive_status(true, None, Some(&s), 1600).state, "offline");
351        assert_eq!(derive_status(true, None, Some(&s), 1500 + OFFLINE_WINDOW_SECS + 1).state, "online");
352        // A success after the failure clears it at once.
353        s.last_ok_at = Some(1501);
354        assert_eq!(derive_status(true, None, Some(&s), 1600).state, "online");
355    }
356
357    #[test]
358    fn a_quota_refusal_reads_maxed_for_the_day() {
359        let mut s = stats();
360        s.last_fail_at = Some(1500);
361        s.last_fail_kind = Some(FAIL_MAXED.into());
362        assert_eq!(derive_status(true, None, Some(&s), 1600).state, "maxed");
363        assert_eq!(derive_status(true, None, Some(&s), 1500 + MAXED_WINDOW_SECS + 1).state, "online");
364    }
365
366    #[test]
367    fn nothing_known_is_untested_and_a_document_alone_is_online() {
368        assert_eq!(derive_status(true, None, None, 0).state, "untested");
369        assert_eq!(derive_status(true, Some(&doc(None)), None, 0).state, "online");
370    }
371
372    #[test]
373    fn samples_round_trip_as_pairs_and_read_old_bare_numbers() {
374        let ring = vec![Sample { mbps: 4.5, bytes: 300 }];
375        let json = serde_json::to_string(&ring).unwrap();
376        assert_eq!(json, "[[4.5,300]]");
377        let back: Vec<Sample> = serde_json::from_str("[[4.5,300],7.25]").unwrap();
378        assert_eq!(back, vec![Sample { mbps: 4.5, bytes: 300 }, Sample { mbps: 7.25, bytes: 0 }]);
379    }
380
381    #[test]
382    fn ewma_starts_at_the_first_sample_and_then_leans_on_history() {
383        assert_eq!(ewma(None, 50.0), 50.0);
384        assert_eq!(ewma(Some(50.0), 100.0), 65.0);
385    }
386}
vector_core/blossom_stats.rs

vector_core/
blossom_stats.rs
