vortix 0.3.1

Terminal UI for WireGuard and OpenVPN with real-time telemetry and leak guarding
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

//! Journal writer task — drains the mpsc and writes to disk + broadcast + tail.

use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use tokio::io::AsyncWriteExt;
use tokio::sync::{broadcast, mpsc};
use tracing::warn;

use crate::vortix_core::engine::event::EventEnvelope;

use super::TailBuffer;

/// Disk-backed writer. Appends one JSON line per event to `path`, fans out to
/// broadcast subscribers, and pushes into the tail buffer.
pub(crate) async fn run(
    path: PathBuf,
    mut mpsc_rx: mpsc::UnboundedReceiver<EventEnvelope>,
    bcast_tx: broadcast::Sender<EventEnvelope>,
    tail: Arc<Mutex<TailBuffer>>,
) {
    let mut file = match tokio::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&path)
        .await
    {
        Ok(f) => f,
        Err(e) => {
            warn!(
                target: "vortix::journal",
                path = %path.display(),
                error = %e,
                "failed to open journal file; events will be dropped"
            );
            // Keep draining so producers don't see WriterGone unexpectedly,
            // but discard records.
            while mpsc_rx.recv().await.is_some() {}
            return;
        }
    };

    while let Some(env) = mpsc_rx.recv().await {
        // 1. Persist.
        match serde_json::to_vec(&env) {
            Ok(mut bytes) => {
                bytes.push(b'\n');
                if let Err(e) = file.write_all(&bytes).await {
                    warn!(
                        target: "vortix::journal",
                        path = %path.display(),
                        error = %e,
                        "journal write failed"
                    );
                } else if let Err(e) = file.flush().await {
                    warn!(target: "vortix::journal", error = %e, "journal flush failed");
                }
            }
            Err(e) => {
                warn!(
                    target: "vortix::journal",
                    error = %e,
                    "failed to serialise journal record"
                );
            }
        }

        // 2. Broadcast (lossy — fine if no subscribers).
        let _ = bcast_tx.send(env.clone());

        // 3. Tail buffer.
        tail.lock().unwrap().push(env);
    }
}

/// Disk-disabled writer. Same fan-out minus the file.
pub(crate) async fn run_in_memory(
    mut mpsc_rx: mpsc::UnboundedReceiver<EventEnvelope>,
    bcast_tx: broadcast::Sender<EventEnvelope>,
    tail: Arc<Mutex<TailBuffer>>,
) {
    while let Some(env) = mpsc_rx.recv().await {
        let _ = bcast_tx.send(env.clone());
        tail.lock().unwrap().push(env);
    }
}