zc2 0.0.27

P2P compute broker with credit-based billing, WAL, and broker mesh support
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

//! Local broker data source.
//!
//! Phase 1 of the terminal refactor (see `docs/TERMINAL_REFACTOR.md`).
//!
//! Mirrors the old `TuiApp::get_metrics` / `render_*` data gathering, producing
//! a [`Snapshot`] from a [`SharedBrokerState`] + [`StatsCollector`].

#![allow(dead_code)]

use std::sync::Arc;

use crate::broker::stats::StatsCollector;
use crate::broker::{SharedBrokerState, WorkerStatus};
use crate::tui::views::snapshot::{Connection, Snapshot, SourceMode, WorkerKind, WorkerRow};

use super::DataSource;

/// Reads the in-process broker state directly.
pub struct LocalSource {
    pub state: SharedBrokerState,
    pub stats: Arc<StatsCollector>,
}

impl LocalSource {
    pub fn new(state: SharedBrokerState, stats: Arc<StatsCollector>) -> Self {
        Self { state, stats }
    }
}

impl DataSource for LocalSource {
    fn snapshot(&self) -> Snapshot {
        let workers = self.state.workers.list();
        let active = workers
            .iter()
            .filter(|w| w.status == WorkerStatus::Healthy)
            .count();

        let metrics = self.stats.metrics(
            active,
            workers.len(),
            self.state.is_local_mode(),
            false,
            self.state.own_wireguard_ip.clone(),
        );

        let worker_rows: Vec<WorkerRow> = workers
            .iter()
            .map(|w| WorkerRow {
                status: match w.status {
                    WorkerStatus::Healthy => WorkerKind::Healthy,
                    WorkerStatus::Busy => WorkerKind::Busy,
                    WorkerStatus::Unhealthy => WorkerKind::Unhealthy,
                    WorkerStatus::Draining => WorkerKind::Draining,
                },
                name: w.name.clone(),
                cpus_available: w.resources.cpus_available,
                memory_gib: w.resources.memory_available as f64 / (1024.0 * 1024.0 * 1024.0),
                avg_latency_ms: w.avg_latency_ms,
            })
            .collect();

        let mode = if self.state.is_local_mode() {
            SourceMode::Local
        } else {
            SourceMode::P2p
        };

        Snapshot {
            title: "Zakuro Compute Broker".to_string(),
            mode,
            host_port: format!("{}:{}", self.state.config.host, self.state.config.port),
            wireguard_ip: metrics.wireguard_ip.clone(),
            ledger_connected: metrics.ledger_connected,
            connection: Connection::Connected,
            metrics,
            workers: worker_rows,
            transactions: self.stats.recent_transactions(50),
            task_offers: self.stats.recent_task_offers(30),
            rps_history: self.stats.rps_history(),
            events: Vec::new(),
        }
    }
}