Skip to main content

max / makenotwork

9.5 KB · 283 lines History Blame Raw
1 //! `pom serve`: starts the API/dashboard server and every background check task.
2
3 use tokio::task::JoinHandle;
4 use tracing::info;
5
6 use pom::alerts::Alerter;
7 use pom::config::Config;
8 use pom::db;
9 use pom::error::Result;
10 use pom::peer;
11
12 use super::tasks;
13 use super::tasks::CheckInterval;
14
15 pub(crate) async fn cmd_serve(pool: &sqlx::SqlitePool, config: &Config) -> Result<()> {
16 let default_interval = config.serve.interval_secs;
17 let prune_days = config.serve.prune_days;
18 let listen_addr = config.serve.listen.clone();
19
20 // Cancellation token for graceful shutdown
21 let token = tokio_util::sync::CancellationToken::new();
22
23 let instance_id =
24 peer::load_or_create_instance_id(config.instance.id.as_deref(), config.db_path()?.dir())?;
25 let instance_name = config.instance_name();
26 let instance_info = peer::InstanceInfo {
27 id: instance_id.clone(),
28 name: instance_name.clone(),
29 version: env!("CARGO_PKG_VERSION").to_string(),
30 targets: config.target_names(),
31 started_at: chrono::Utc::now().to_rfc3339(),
32 };
33
34 let alerter = match config.alerts.as_ref() {
35 Some(alert_config) => {
36 info!("Alerts enabled (to: {})", alert_config.to);
37 match Alerter::new(alert_config.clone(), pool.clone(), instance_name.clone()) {
38 Ok(a) => Some(a),
39 Err(e) => {
40 tracing::error!("alerts: client build failed, alerting disabled: {e}");
41 None
42 }
43 }
44 }
45 None => None,
46 };
47
48 info!("Instance: {instance_name} (id={instance_id})");
49 info!("Starting serve mode (default interval: {default_interval}s, prune: {prune_days}d)");
50
51 let mesh = peer::new_mesh_state(instance_info, &config.peers);
52
53 // Load known peer identities from DB
54 {
55 let mut state = mesh.write().await;
56 for (name, peer) in &mut state.peers {
57 if let Ok(Some(known_id)) = db::get_peer_identity(pool, name).await {
58 peer.known_id = Some(known_id);
59 }
60 }
61 }
62
63 let mut handles: Vec<JoinHandle<()>> = Vec::new();
64
65 handles.extend(tasks::spawn_health_tasks(
66 config,
67 pool,
68 &token,
69 alerter.as_ref(),
70 ));
71 handles.extend(tasks::spawn_tls_tasks(
72 config,
73 pool,
74 &token,
75 alerter.as_ref(),
76 ));
77 handles.extend(tasks::spawn_route_tasks(
78 config,
79 pool,
80 &token,
81 alerter.as_ref(),
82 ));
83 handles.extend(tasks::spawn_dns_tasks(
84 config,
85 pool,
86 &token,
87 alerter.as_ref(),
88 ));
89 handles.extend(tasks::spawn_whois_tasks(
90 config,
91 pool,
92 &token,
93 alerter.as_ref(),
94 ));
95 handles.extend(tasks::spawn_cors_tasks(
96 config,
97 pool,
98 &token,
99 alerter.as_ref(),
100 ));
101 handles.extend(tasks::spawn_backup_tasks(
102 config,
103 pool,
104 &token,
105 alerter.as_ref(),
106 ));
107 handles.extend(tasks::spawn_scan_pipeline_tasks(
108 config,
109 pool,
110 &token,
111 alerter.as_ref(),
112 ));
113 handles.extend(tasks::spawn_systemd_tasks(
114 config,
115 pool,
116 &token,
117 alerter.as_ref(),
118 ));
119 handles.extend(tasks::spawn_ca_bundle_tasks(
120 config,
121 pool,
122 &token,
123 alerter.as_ref(),
124 ));
125 handles.extend(tasks::spawn_test_tasks(
126 config,
127 pool,
128 &token,
129 alerter.as_ref(),
130 ));
131 // No alerter: the fleet readout has nothing to alert on. See the task module.
132 handles.extend(tasks::spawn_synckit_fleet_tasks(config, pool, &token));
133 handles.push(tasks::spawn_prune_task(pool, prune_days, &token));
134
135 // Spawn peer heartbeat tasks
136 if !config.peers.is_empty() {
137 let heartbeat_secs = config.serve.peer_heartbeat_secs;
138 info!(
139 "Peer mesh: {} peers, heartbeat every {heartbeat_secs}s",
140 config.peers.len()
141 );
142 let hb_handles = peer::spawn_heartbeat_tasks(
143 mesh.clone(),
144 pool.clone(),
145 heartbeat_secs,
146 alerter.clone(),
147 token.clone(),
148 )
149 .await;
150 handles.extend(hb_handles);
151 }
152
153 if let Some(handle) =
154 tasks::spawn_meta_alert_task(config, pool, default_interval, &token, alerter.as_ref())
155 {
156 handles.push(handle);
157 }
158
159 // Spawn the pending-alert retry drainer. When an alert's original send failed
160 // it was queued (the status transition fires only once, so without this the
161 // outage would go silent); re-deliver it here until it lands. This is the
162 // durable half of the fix for the transition-once alert-loss CRITICAL.
163 if let Some(retry_alerter) = alerter.clone() {
164 let pool = pool.clone();
165 let cancel = token.clone();
166 handles.push(tokio::spawn(async move {
167 const MAX_ATTEMPTS: i64 = 8; // ~ several hours of backoff before dead-letter
168 let mut ticks = CheckInterval::new(30, cancel);
169 while ticks.next().await {
170 let due = db::due_pending_alerts(&pool, 50).await.unwrap_or_default();
171 for p in due {
172 if retry_alerter.retry_pending(&p).await {
173 if let Err(e) = db::delete_pending_alert(&pool, p.id).await {
174 tracing::warn!(
175 alert_id = p.id,
176 alert_key = %p.alert_key,
177 error = %e,
178 "pending alert delete failed, alert will re-send"
179 );
180 }
181 info!("delivered previously-queued alert to {}", p.alert_key);
182 } else if p.attempts + 1 >= MAX_ATTEMPTS {
183 // Give up so the queue can't grow without bound; losing the
184 // row is logged loudly rather than silently.
185 tracing::error!(
186 "alert to {} undeliverable after {} attempts; dropping",
187 p.alert_key,
188 p.attempts + 1
189 );
190 if let Err(e) = db::delete_pending_alert(&pool, p.id).await {
191 tracing::warn!(
192 alert_id = p.id,
193 alert_key = %p.alert_key,
194 error = %e,
195 "pending alert delete failed, the drop retries next tick"
196 );
197 }
198 } else {
199 // Exponential backoff, capped at 1h.
200 let backoff =
201 std::cmp::min(30 * 2i64.pow((p.attempts.min(7)) as u32), 3600);
202 let next =
203 (chrono::Utc::now() + chrono::Duration::seconds(backoff)).to_rfc3339();
204 if let Err(e) = db::bump_pending_alert(&pool, p.id, &next).await {
205 tracing::warn!(
206 alert_id = p.id,
207 alert_key = %p.alert_key,
208 error = %e,
209 "pending alert backoff bump failed, row retries next tick"
210 );
211 }
212 }
213 }
214 }
215 }));
216 }
217
218 // Start HTTP API server
219 let api_app = pom::api::router(pool.clone(), config.clone(), Some(mesh.clone()));
220 let api_listener = tokio::net::TcpListener::bind(&listen_addr).await?;
221 info!("API server listening on {listen_addr}");
222 let api_cancel = token.clone();
223 handles.push(tokio::spawn(async move {
224 if let Err(e) = axum::serve(
225 api_listener,
226 api_app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
227 )
228 .with_graceful_shutdown(async move { api_cancel.cancelled().await })
229 .await
230 {
231 tracing::error!("API server error: {e}");
232 }
233 }));
234
235 // Wait for shutdown signal or unexpected task exit
236 let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
237
238 let mut watchdog_interval = tokio::time::interval(std::time::Duration::from_mins(1));
239 watchdog_interval.tick().await; // consume immediate first tick
240
241 loop {
242 tokio::select! {
243 _ = tokio::signal::ctrl_c() => {
244 info!("Received SIGINT, shutting down");
245 break;
246 }
247 _ = sigterm.recv() => {
248 info!("Received SIGTERM, shutting down");
249 break;
250 }
251 _ = watchdog_interval.tick() => {
252 for (i, handle) in handles.iter().enumerate() {
253 if handle.is_finished() {
254 tracing::error!("Background task {i} exited unexpectedly (possible panic)");
255 }
256 }
257 }
258 }
259 }
260
261 // Graceful shutdown: cancel all tasks, then wait with grace period
262 token.cancel();
263 info!("Waiting for tasks to finish (5s grace period)...");
264
265 let shutdown = async {
266 for handle in handles {
267 if let Err(e) = handle.await {
268 tracing::error!("Task shutdown error: {e}");
269 }
270 }
271 };
272
273 if tokio::time::timeout(std::time::Duration::from_secs(5), shutdown)
274 .await
275 .is_err()
276 {
277 tracing::warn!("Grace period elapsed, some tasks may not have finished cleanly");
278 }
279
280 info!("Shutdown complete");
281 Ok(())
282 }
283