pulpod 0.1.0

Pulpo daemon — manages agent sessions via tmux/Docker
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

#[cfg(not(coverage))]
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    use clap::Parser;
    use tracing::info;

    let cli = pulpod::Cli::parse();

    let config = pulpod::config::load(&cli.config)?;
    let data_dir = std::path::PathBuf::from(config.data_dir());
    let _log_guard = pulpod::init_tracing(Some(&data_dir), config.node.log_retain_days)?;
    let (app, addr, shutdown_handle) = pulpod::build_app(&cli).await?;
    let listener = tokio::net::TcpListener::bind(&addr).await?;

    // After the shutdown signal fires, give in-flight streaming
    // connections (SSE, WebSocket) 3 seconds to close before forcing exit.
    let (shutdown_done_tx, shutdown_done_rx) = tokio::sync::oneshot::channel::<()>();
    let server = axum::serve(listener, app).with_graceful_shutdown(async move {
        shutdown_signal(shutdown_handle).await;
        let _ = shutdown_done_tx.send(());
    });

    tokio::select! {
        result = server => result?,
        () = async {
            let _ = shutdown_done_rx.await;
            tokio::time::sleep(std::time::Duration::from_secs(3)).await;
        } => {
            info!("Streaming connections still open — forcing shutdown");
        }
    }
    info!("pulpod shut down cleanly");

    Ok(())
}

/// Wait for SIGTERM or Ctrl+C, then signal all background loops to stop.
#[cfg(not(coverage))]
async fn shutdown_signal(handle: pulpod::ShutdownHandle) {
    use tracing::info;

    let ctrl_c = async {
        tokio::signal::ctrl_c()
            .await
            .expect("failed to install Ctrl+C handler");
    };

    #[cfg(unix)]
    let terminate = async {
        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
            .expect("failed to install SIGTERM handler")
            .recv()
            .await;
    };

    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();

    tokio::select! {
        () = ctrl_c => {},
        () = terminate => {},
    }

    info!("Shutdown signal received, stopping background tasks...");
    handle.shutdown();
}

// Dummy main for coverage builds — the real logic is tested via lib.rs
#[cfg(coverage)]
fn main() {}