wowlab_sentinel/
application.rs

1use std::sync::Arc;
2
3use metrics_exporter_prometheus::PrometheusBuilder;
4use tokio::sync::RwLock;
5use tokio_util::sync::CancellationToken;
6use wowlab_centrifuge::{Client, ClientConfig, ClientEvent, Presence, generate_token};
7use wowlab_types::sim::FastMap;
8
9use crate::{
10    config::{Config, ExposeSecret},
11    cron, db, http, scheduler,
12    state::{HealthState, McpState, RuntimeState, ServerState},
13};
14
15#[derive(Debug)]
16struct Startup {
17    state: Arc<ServerState>,
18}
19
20const CENTRIFUGE_CONNECT_TIMEOUT: std::time::Duration =
21    std::time::Duration::from_secs(wowlab_types::constants::HTTP_CONNECT_TIMEOUT_SECS);
22
23/// Starts all Sentinel services and runs until one exits or shutdown is requested.
24///
25/// # Panics
26///
27/// Panics when required configuration is missing or invalid, startup state cannot be initialized, or the metrics/tracing recorders cannot be installed.
28pub async fn run() {
29    load_env();
30    init_tracing();
31
32    let config = Config::from_env().expect("required SENTINEL_* env var missing");
33
34    config.log();
35
36    let startup = build_state(config).await;
37
38    initialize_runtime(&startup.state).await;
39
40    tracing::info!("Starting wowlab-sentinel");
41    serve(startup).await;
42}
43
44fn load_env() {
45    if dotenvy::dotenv().is_err() {
46        let crate_dir = env!("CARGO_MANIFEST_DIR");
47        let _ = dotenvy::from_path(format!("{crate_dir}/.env"));
48    }
49}
50
51fn init_tracing() {
52    tracing_subscriber::fmt()
53        .with_env_filter(
54            tracing_subscriber::EnvFilter::from_default_env().add_directive(
55                "wowlab_sentinel=info"
56                    .parse()
57                    .expect("static tracing directive must parse"),
58            ),
59        )
60        .init();
61}
62
63async fn build_state(config: Config) -> Startup {
64    let prometheus = PrometheusBuilder::new()
65        .install_recorder()
66        .expect("Failed to install prometheus recorder");
67    let dbs = db::DbRegistry::connect(&config)
68        .await
69        .expect("Failed to connect to database");
70    let filters = Arc::new(RwLock::new(FastMap::default()));
71    let centrifuge = connect_centrifuge(&config).await;
72    let presence = Presence::new(
73        &config.centrifugo_url,
74        config.centrifugo_key.expose_secret(),
75    );
76    let latitude = config.latitude_api_key.as_ref().map(|key| {
77        crate::latitude::LatitudeClient::new(key.clone(), config.latitude_base_url.clone())
78    });
79
80    let state = Arc::new(ServerState {
81        config,
82        dbs,
83        filters,
84        health: HealthState::new(prometheus),
85        runtime: RuntimeState::default(),
86        centrifuge,
87        presence,
88        mcp: Arc::new(
89            McpState::new().expect("static MCP table descriptors must match the Types registry"),
90        ),
91        latitude,
92    });
93
94    Startup { state }
95}
96
97async fn connect_centrifuge(config: &Config) -> Client {
98    let initial_token = generate_token("sentinel", config.centrifugo_token_secret.expose_secret())
99        .expect("Failed to generate beacon token");
100    let token_secret = config.centrifugo_token_secret.clone();
101    let centrifuge = Client::new(
102        ClientConfig::new(&config.centrifugo_url, initial_token)
103            .name(env!("CARGO_PKG_NAME"))
104            .version(env!("CARGO_PKG_VERSION"))
105            .get_token(move || {
106                let secret = token_secret.clone();
107
108                async move {
109                    generate_token("sentinel", secret.expose_secret())
110                        .map_err(|error| wowlab_centrifuge::Error::protocol(error.to_string()))
111                }
112            }),
113    );
114
115    let mut events = centrifuge.events().await;
116
117    centrifuge.connect();
118    let connected = tokio::time::timeout(CENTRIFUGE_CONNECT_TIMEOUT, async {
119        let mut errors = 0usize;
120
121        while let Some(event) = events.recv().await {
122            match event {
123                ClientEvent::Connected(_) => return (true, errors),
124                ClientEvent::Error(_) => errors += 1,
125                _ => {}
126            }
127        }
128
129        (false, errors)
130    })
131    .await;
132
133    let (connection_succeeded, error_count) = connected.unwrap_or((false, 0));
134
135    if error_count > 0 {
136        tracing::warn!(error_count, "Centrifuge reported startup connection errors");
137    }
138
139    if connection_succeeded {
140        tracing::info!("Connected to Beacon");
141    } else {
142        tracing::warn!("Failed to connect to Beacon on startup");
143    }
144
145    centrifuge
146}
147
148async fn initialize_runtime(state: &ServerState) {
149    crate::telemetry::init();
150    crate::telemetry::init_running_chunks_gauge(state.dbs.get::<scheduler::SchedulerDb>()).await;
151    fail_interrupted_jobs(state).await;
152    cron::docs::refresh(&state.config.docs_base_url, &state.mcp).await;
153}
154
155async fn fail_interrupted_jobs(state: &ServerState) {
156    // Runtime state is process-memory only, so jobs left `running` by a prior process are unrecoverable.
157    // docref:start database-restart-fail-running
158    match sqlx::query_file!("queries/scheduler_fail_running_on_restart.sql")
159        .fetch_all(state.dbs.get::<scheduler::SchedulerDb>())
160        .await
161    {
162        Ok(rows) => {
163            let mut log_failures = 0usize;
164            for row in &rows {
165                if sqlx::query_file!(
166                    "queries/chunk_event_log_insert.sql",
167                    row.id,
168                    "restart_dropped",
169                    Option::<uuid::Uuid>::None,
170                    Option::<String>::None,
171                    Some("sentinel restart dropped in-flight runtime state"),
172                )
173                .execute(state.dbs.get::<scheduler::SchedulerDb>())
174                .await
175                .is_err()
176                {
177                    log_failures += 1;
178                }
179            }
180            if log_failures > 0 {
181                tracing::error!(log_failures, "Failed to record restart_dropped events");
182            }
183            if !rows.is_empty() {
184                tracing::warn!(count = rows.len(), "Failed running jobs on restart");
185            }
186        }
187        Err(error) => tracing::error!(%error, "Failed to fail running jobs on restart"),
188    }
189    // docref:end database-restart-fail-running
190}
191
192async fn serve(startup: Startup) {
193    let Startup { state } = startup;
194    let shutdown = CancellationToken::new();
195
196    spawn_shutdown_signal(shutdown.clone());
197    let centrifuge = state.centrifuge.clone();
198
199    tokio::select! {
200        result = scheduler::run(Arc::clone(&state), shutdown.clone()) => {
201            if let Err(error) = result {
202                tracing::error!(%error, "Scheduler exited with error");
203            }
204        }
205        result = cron::run(Arc::clone(&state), shutdown.clone()) => {
206            if let Err(error) = result {
207                tracing::error!(%error, "Cron scheduler exited with error");
208            }
209        }
210        result = http::run(Arc::clone(&state), shutdown.clone()) => {
211            if let Err(error) = result {
212                tracing::error!(%error, "HTTP server exited with error");
213            }
214        }
215    }
216
217    shutdown.cancel();
218    tracing::info!("Server shutting down, cleaning up resources...");
219    centrifuge.disconnect();
220    tracing::debug!("Disconnected from Beacon");
221    state.dbs.close().await;
222    tracing::debug!("Closed database connections");
223    tracing::info!("Shutdown complete");
224}
225
226fn spawn_shutdown_signal(shutdown: CancellationToken) {
227    tokio::spawn(async move {
228        #[cfg(unix)]
229        {
230            use tokio::signal::unix::{SignalKind, signal};
231            let mut sigterm =
232                signal(SignalKind::terminate()).expect("Failed to register SIGTERM handler");
233            let mut sigint =
234                signal(SignalKind::interrupt()).expect("Failed to register SIGINT handler");
235
236            tokio::select! {
237                _ = sigterm.recv() => tracing::info!("Received SIGTERM, initiating graceful shutdown"),
238                _ = sigint.recv() => tracing::info!("Received SIGINT, initiating graceful shutdown"),
239            }
240        }
241
242        #[cfg(windows)]
243        {
244            if let Err(error) = tokio::signal::ctrl_c().await {
245                tracing::error!(%error, "Failed to listen for Ctrl-C");
246            }
247
248            tracing::info!("Received Ctrl-C, initiating graceful shutdown");
249        }
250
251        shutdown.cancel();
252    });
253}
wowlab_sentinel/application.rs

wowlab_sentinel/
application.rs
