serve.rs 79.5%

70 of 88 lines run, 10 of 28 functions · run #15 on main · run missed

//! `spectra serve`: the daemon (spec §7.1, §21.1). It holds the ledger,
//! serves Tuner (and, as they land, the API and Radar) on the Unix socket
//! and on a loopback port, and writes a checkpoint when it stops.

use super::{ui, Ctx};
use anyhow::{bail, Context, Result};
use spectra_core::daemon::Daemon;
use spectra_mcp::client::Client;
use spectra_mcp::http::{AppState, PeerInfo};
use std::net::SocketAddr;
use std::sync::Arc;

pub async fn serve(ctx: &Ctx, bind: &str, insecure: bool, open: bool) -> Result<i32> {
    let addr: SocketAddr = bind.parse().with_context(|| format!("--bind {bind} is not an address and port"))?;
    if !addr.ip().is_loopback() && !insecure {
        bail!("Spectra serves loopback only unless it has TLS (spec §21.1); pass --insecure to bind {addr} in plain HTTP anyway");
    }
    let socket = ctx.socket();
    if socket.exists() {
        if Client::alive(&socket).await {
            bail!("Spectra is already serving {} (its socket answers)", ctx.data.display());
        }
        std::fs::remove_file(&socket).ok();
    }
    let daemon = Daemon::start(&ctx.data).await?;
    let unix = tokio::net::UnixListener::bind(&socket).with_context(|| format!("binding {}", socket.display()))?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o660))?;
    }
    let tcp = tokio::net::TcpListener::bind(addr).await.with_context(|| format!("binding {addr}"))?;
    let port = tcp.local_addr()?.port();
    let origins = vec![format!("http://127.0.0.1:{port}"), format!("http://localhost:{port}"), format!("http://{addr}")];
    let state = AppState { daemon: daemon.clone(), origins: Arc::new(origins), sessions: Default::default() };
    let app = spectra_mcp::http::router(state.clone()).merge(spectra_mcp::api::router(state.clone())).merge(radar::router(state));

    let (head, _) = daemon.ledger.head().await;
    let agents = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM agents WHERE disabled_ms IS NULL")
        .fetch_one(daemon.ledger.pool())
        .await
        .unwrap_or(0);
    println!("{} {}", ui::done("Serving"), ui::subject("Spectra"));
    println!("{}", ui::kv("socket", ui::path(socket.display())));
    println!("{}", ui::kv("http", format!("http://127.0.0.1:{port}")));
    println!("{}", ui::kv("ledger", format!("{head} events{}{agents} agent(s)", ui::sep())));
    println!("  {} {}", ui::note(), ui::dim("agents connect with: spectra mcp --agent NAME"));
    println!("  {} {}", ui::note(), ui::dim("open Radar with: spectra open"));
    if open {
        let ctx2 = Ctx { data: ctx.data.clone() };
        tokio::spawn(async move {
            // Once the socket answers, sign the link as the owner and open it.
            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
            if let Err(e) = super::open::open(&ctx2, None, true).await {
                eprintln!("{} {e:#}", ui::caution());
            }
        });
    }

    let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
    let stop = |mut rx: tokio::sync::watch::Receiver<bool>| async move {
        let _ = rx.wait_for(|s| *s).await;
    };
    let unix_srv =
        axum::serve(unix, app.clone().into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx.clone()));
    let tcp_srv = axum::serve(tcp, app.into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx));
    let signal = async move {
        shutdown_signal().await;
        let _ = stop_tx.send(true);
    };
    // After the signal, open connections get five seconds to finish: a
    // `subscriptions/listen` stream never does by itself.
    let (stopping_tx, mut stopping) = tokio::sync::watch::channel(false);
    tokio::spawn(async move {
        signal.await;
        let _ = stopping_tx.send(true);
    });
    let servers = async { tokio::join!(unix_srv, tcp_srv) };
    tokio::select! {
        (u, t) = servers => {
            u?;
            t?;
        }
        _ = async {
            let _ = stopping.wait_for(|s| *s).await;
            tokio::time::sleep(std::time::Duration::from_secs(5)).await;
        } => {
            tracing::info!("connections still open five seconds after the signal; stopping anyway");
        }
    }
    daemon.shutdown().await;
    std::fs::remove_file(&socket).ok();
    println!("{} {}", ui::done("Stopped"), ui::dim(format!("ledger checkpointed at {} events", daemon.ledger.head().await.0)));
    Ok(0)
}

async fn shutdown_signal() {
    #[cfg(unix)]
    {
        let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).expect("a SIGTERM handler");
        tokio::select! {
            _ = tokio::signal::ctrl_c() => {}
            _ = term.recv() => {}
        }
    }
    #[cfg(not(unix))]
    let _ = tokio::signal::ctrl_c().await;
}

1		//! `spectra serve`: the daemon (spec §7.1, §21.1). It holds the ledger,
2		//! serves Tuner (and, as they land, the API and Radar) on the Unix socket
3		//! and on a loopback port, and writes a checkpoint when it stops.
4
5		use super::{ui, Ctx};
6		use anyhow::{bail, Context, Result};
7		use spectra_core::daemon::Daemon;
8		use spectra_mcp::client::Client;
9		use spectra_mcp::http::{AppState, PeerInfo};
10		use std::net::SocketAddr;
11		use std::sync::Arc;
12
13	10×	pub async fn serve(ctx: &Ctx, bind: &str, insecure: bool, open: bool) -> Result<i32> {
14	10×	let addr: SocketAddr = bind.parse().with_context(\|\| format!("--bind {bind} is not an address and port"))?;
15	10×	if !addr.ip().is_loopback() && !insecure {
16	0×	bail!("Spectra serves loopback only unless it has TLS (spec §21.1); pass --insecure to bind {addr} in plain HTTP anyway");
17	10×	}
18	10×	let socket = ctx.socket();
19	10×	if socket.exists() {
20	0×	if Client::alive(&socket).await {
21	0×	bail!("Spectra is already serving {} (its socket answers)", ctx.data.display());
22	0×	}
23	0×	std::fs::remove_file(&socket).ok();
24	10×	}
25	10×	let daemon = Daemon::start(&ctx.data).await?;
26	10×	let unix = tokio::net::UnixListener::bind(&socket).with_context(\|\| format!("binding {}", socket.display()))?;
27		#[cfg(unix)]
28		{
29		use std::os::unix::fs::PermissionsExt;
30	10×	std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o660))?;
31		}
32	10×	let tcp = tokio::net::TcpListener::bind(addr).await.with_context(\|\| format!("binding {addr}"))?;
33	10×	let port = tcp.local_addr()?.port();
34	10×	let origins = vec![format!("http://127.0.0.1:{port}"), format!("http://localhost:{port}"), format!("http://{addr}")];
35	10×	let state = AppState { daemon: daemon.clone(), origins: Arc::new(origins), sessions: Default::default() };
36	10×	let app = spectra_mcp::http::router(state.clone()).merge(spectra_mcp::api::router(state.clone())).merge(radar::router(state));
37
38	10×	let (head, _) = daemon.ledger.head().await;
39	10×	let agents = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM agents WHERE disabled_ms IS NULL")
40	10×	.fetch_one(daemon.ledger.pool())
41	10×	.await
42	10×	.unwrap_or(0);
43	10×	println!("{} {}", ui::done("Serving"), ui::subject("Spectra"));
44	10×	println!("{}", ui::kv("socket", ui::path(socket.display())));
45	10×	println!("{}", ui::kv("http", format!("http://127.0.0.1:{port}")));
46	10×	println!("{}", ui::kv("ledger", format!("{head} events{}{agents} agent(s)", ui::sep())));
47	10×	println!(" {} {}", ui::note(), ui::dim("agents connect with: spectra mcp --agent NAME"));
48	10×	println!(" {} {}", ui::note(), ui::dim("open Radar with: spectra open"));
49	10×	if open {
50	0×	let ctx2 = Ctx { data: ctx.data.clone() };
51	0×	tokio::spawn(async move {
52		// Once the socket answers, sign the link as the owner and open it.
53	0×	tokio::time::sleep(std::time::Duration::from_millis(200)).await;
54	0×	if let Err(e) = super::open::open(&ctx2, None, true).await {
55	0×	eprintln!("{} {e:#}", ui::caution());
56	0×	}
57	0×	});
58	10×	}
59
60	10×	let (stop_tx, stop_rx) = tokio::sync::watch::channel(false);
61	20×	let stop = \|mut rx: tokio::sync::watch::Receiver<bool>\| async move {
62	20×	let _ = rx.wait_for(\|s\| *s).await;
63	40×	};
64	10×	let unix_srv =
65	10×	axum::serve(unix, app.clone().into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx.clone()));
66	10×	let tcp_srv = axum::serve(tcp, app.into_make_service_with_connect_info::<PeerInfo>()).with_graceful_shutdown(stop(stop_rx));
67	10×	let signal = async move {
68	10×	shutdown_signal().await;
69	10×	let _ = stop_tx.send(true);
70	10×	};
71		// After the signal, open connections get five seconds to finish: a
72		// `subscriptions/listen` stream never does by itself.
73	10×	let (stopping_tx, mut stopping) = tokio::sync::watch::channel(false);
74	10×	tokio::spawn(async move {
75	10×	signal.await;
76	10×	let _ = stopping_tx.send(true);
77	10×	});
78	10×	let servers = async { tokio::join!(unix_srv, tcp_srv) };
79	10×	tokio::select! {
80	10×	(u, t) = servers => {
81	10×	u?;
82	10×	t?;
83		}
84	10×	_ = async {
85	10×	let _ = stopping.wait_for(\|s\| *s).await;
86	6×	tokio::time::sleep(std::time::Duration::from_secs(5)).await;
87	0×	} => {
88	0×	tracing::info!("connections still open five seconds after the signal; stopping anyway");
89		}
90		}
91	10×	daemon.shutdown().await;
92	10×	std::fs::remove_file(&socket).ok();
93	10×	println!("{} {}", ui::done("Stopped"), ui::dim(format!("ledger checkpointed at {} events", daemon.ledger.head().await.0)));
94	10×	Ok(0)
95	10×	}
96
97	10×	async fn shutdown_signal() {
98		#[cfg(unix)]
99		{
100	10×	let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).expect("a SIGTERM handler");
101	10×	tokio::select! {
102	10×	_ = tokio::signal::ctrl_c() => {}
103	10×	_ = term.recv() => {}
104		}
105		}
106		#[cfg(not(unix))]
107		let _ = tokio::signal::ctrl_c().await;
108	10×	}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / src/cli/serve.rs 79.5%

Keyboard