mail4agent_messenger_shell/cli/
web_client.rs

1//! The web machine client as a process: [`MachineClient::from_env`], then
2//! [`MachineClient::tick`] in a loop, and
3//! [`MachineClient::poll_agent_directory`] on `M4A_AGENT_RESCAN_SECS`.
4//!
5//! Settings come from the environment, then from the env file
6//! (`M4A_ENV_FILE`, default `~/.config/mail4agent/web-client.env`) for
7//! anything unset. The client listens on the local send socket
8//! (`M4A_SEND_SOCK`, default `web-client.sock` under the store root) for
9//! `m4a-send`.
10//!
11//! Environment: everything `from_env` reads, plus `M4A_DRIVE_SECS` (full
12//! drive period, default 15) and `M4A_RUN_SECS` (exit after that many
13//! seconds; unset or 0 runs until killed). Prints nicks, event ids, room
14//! ids, and HTTP statuses only. Never prints URLs, keys, or bearers.
15
16use crate::MachineClient;
17use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
18
19/// Streak at which the log says once more that the daemon is still retrying.
20const HEAL_LOG_STREAK: u32 = 10;
21
22fn env_secs(key: &str, default: u64) -> u64 {
23    std::env::var(key)
24        .ok()
25        .and_then(|value| value.parse().ok())
26        .unwrap_or(default)
27}
28
29fn now_ms() -> i64 {
30    SystemTime::now()
31        .duration_since(UNIX_EPOCH)
32        .map(|elapsed| elapsed.as_millis() as i64)
33        .unwrap_or(0)
34}
35
36pub fn run(_args: Vec<String>) {
37    crate::load_env_file();
38    let mut client = match MachineClient::from_env() {
39        Ok(client) => client,
40        Err(err) => {
41            eprintln!("open failed: {err}");
42            std::process::exit(1);
43        }
44    };
45    match client.listen_for_sends_from_env() {
46        Ok(path) => println!("send socket {}", path.display()),
47        Err(err) => eprintln!("send socket not opened: {err}"),
48    }
49    println!("open ok; waiting on the push socket");
50    let drive_secs = env_secs("M4A_DRIVE_SECS", 15);
51    let run_secs = env_secs("M4A_RUN_SECS", 0);
52    let started = Instant::now();
53    let mut seen: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
54    let stale_secs = env_secs("M4A_STALE_SECS", 180);
55    let mut failing_since: Option<Instant> = None;
56    let mut last_probe = Instant::now();
57    loop {
58        let report = client.tick(now_ms(), drive_secs);
59        for (nick, event_id) in &report.pushed {
60            println!("push {nick} event={event_id}");
61        }
62        for (nick, text) in &report.alerts {
63            println!("alert {nick}: {text}");
64        }
65        for (nick, room) in &report.joined {
66            println!("joined {nick} room={room}");
67        }
68        for (from, to, reply) in &report.sent {
69            match (&reply.event_id, &reply.error) {
70                (_, Some(err)) => println!("send {from} -> {to} failed: {err}"),
71                (Some(event_id), None) => println!("send {from} -> {to} event={event_id}"),
72                (None, None) => println!("send {from} -> {to} ok"),
73            }
74        }
75        for (nick, err) in &report.errors {
76            eprintln!("drive {nick}: {err}");
77        }
78        // The log is per session, flattened: count what was printed for each nick, not overall.
79        let log = client.wake_log();
80        let mut now_counts: std::collections::HashMap<String, usize> = std::collections::HashMap::new();
81        for (nick, attempt) in &log {
82            let n = now_counts.entry(nick.clone()).or_insert(0);
83            *n += 1;
84            if *n > seen.get(nick).copied().unwrap_or(0) {
85                let status = attempt.status.map(|code| code.to_string()).unwrap_or_else(|| "queued (live deliver retried with backoff)".to_string());
86                println!("wake {nick} event={} status={status}", attempt.event_id);
87            }
88        }
89        seen = now_counts;
90        if let Err(err) = client.poll_agent_directory() {
91            eprintln!("agent poll: {err}");
92        }
93        if run_secs > 0 && started.elapsed().as_secs() >= run_secs {
94            println!("run time reached; exiting");
95            break;
96        }
97        // Sessions keep failing (homeserver down or restarting): back off instead of hammering it,
98        // up to 10 s. The client rebuilds its push socket itself and the next clean tick resets this.
99        let streak = client.error_streak();
100        if streak == 1 || streak == HEAL_LOG_STREAK {
101            eprintln!("homeserver unreachable or failing; backing off and retrying on its own");
102        }
103        // Health check: the client failing for minutes while a brand-new connection reaches the
104        // homeserver means state inside this process went stale (seen 13.10 and 16.10: hours of
105        // "error sending request" that a restart cured at once). Exit so the supervisor restarts it.
106        if streak == 0 { failing_since = None; } else if failing_since.is_none() { failing_since = Some(Instant::now()); }
107        if let Some(since) = failing_since {
108            if stale_secs > 0 && since.elapsed().as_secs() >= stale_secs && last_probe.elapsed().as_secs() >= 60 {
109                last_probe = Instant::now();
110                if let Some(code) = fresh_probe() {
111                    eprintln!("stale client: failing {}s while a fresh connection gets HTTP {code}; exiting for a restart", since.elapsed().as_secs());
112                    std::process::exit(75);
113                }
114            }
115        }
116        let pause_ms = if streak == 0 { 250 } else { (250u64 << streak.min(6)).min(10_000) };
117        std::thread::sleep(Duration::from_millis(pause_ms));
118    }
119}
120
121/// One request on a brand-new HTTP client (new pool, new DNS lookup): `Some(status)` if the
122/// homeserver answered at all.
123fn fresh_probe() -> Option<u16> {
124    let base = std::env::var("M4A_HOMESERVER_URL").or_else(|_| std::env::var("M4A_PRODUCT_URL")).ok()?;
125    let url = format!("{}/_matrix/client/versions", base.trim_end_matches('/'));
126    let client = reqwest::blocking::Client::builder().timeout(Duration::from_secs(10)).build().ok()?;
127    client.get(url).send().ok().map(|r| r.status().as_u16())
128}
mail4agent_messenger_shell/cli/web_client.rs

mail4agent_messenger_shell/cli/
web_client.rs
