Single-binary self-hosted uptime monitoring and status pages on Rust axum: HTTP probes, Lighthouse audits, SEO crawler, and PDF reports.
axumdockerrustself-hostedsqlitestatus-pageuptime-monitoringvite
1use crate::checker;
2use crate::crawler;
3use crate::db::now_ms;
4use crate::lighthouse;
5use crate::models::PropertyRow;
6use crate::Config;
7use sqlx::SqlitePool;
8use std::sync::Arc;
9use std::time::Duration;
10use tokio::sync::Semaphore;
11use tokio::task::JoinHandle;
12
13const CYCLE_SECS: u64 = 30;
14// Two pools so slow lighthouse/crawler work can't starve quick HTTP pings.
15const FAST_PERMITS: usize = 2;
16const SLOW_PERMITS: usize = 2;
17const CLEANUP_INTERVAL_SECS: i64 = 86_400;
18const CHECK_RETENTION_DAYS: i64 = 3;
19const CRAWL_WEDGE_SECS: i64 = 900;
20const LH_WEDGE_SECS: i64 = 300;
21const CRAWL_INTERVAL_DAYS: i64 = 7;
22const LH_INTERVAL_DAYS: i64 = 1;
23
24/// Wipe queued/running rows on startup. Anything in those states is
25/// leftover from a prior crash; tasks didn't survive the restart so the
26/// rows must not block new work.
27pub async fn reset_states_on_boot(pool: &SqlitePool) -> anyhow::Result<()> {
28 sqlx::query("UPDATE properties SET crawl_state = 'idle' WHERE crawl_state IN ('queued', 'running')")
29 .execute(pool)
30 .await?;
31 sqlx::query(
32 "UPDATE properties SET lighthouse_state = 'idle' WHERE lighthouse_state IN ('queued', 'running')",
33 )
34 .execute(pool)
35 .await?;
36 Ok(())
37}
38
39pub fn spawn(pool: SqlitePool, config: Arc<Config>) -> JoinHandle<()> {
40 let fast = Arc::new(Semaphore::new(FAST_PERMITS));
41 let slow = Arc::new(Semaphore::new(SLOW_PERMITS));
42 tokio::spawn(async move {
43 let mut last_cleanup: Option<i64> = None;
44 loop {
45 if let Err(e) = enqueue_status(&pool, &config, &fast).await {
46 tracing::warn!("[scheduler] enqueue_status: {e}");
47 }
48 if let Err(e) = enqueue_lighthouse(&pool, &config, &slow).await {
49 tracing::warn!("[scheduler] enqueue_lighthouse: {e}");
50 }
51 if let Err(e) = enqueue_crawler(&pool, &config, &slow).await {
52 tracing::warn!("[scheduler] enqueue_crawler: {e}");
53 }
54 if let Err(e) = reset_wedged_states(&pool).await {
55 tracing::warn!("[scheduler] reset_wedged_states: {e}");
56 }
57 if let Err(e) = maybe_cleanup(&pool, &mut last_cleanup).await {
58 tracing::warn!("[scheduler] cleanup: {e}");
59 }
60 tokio::time::sleep(Duration::from_secs(CYCLE_SECS)).await;
61 }
62 })
63}
64
65async fn reset_wedged_states(pool: &SqlitePool) -> sqlx::Result<()> {
66 let now = now_ms();
67 let crawl_cutoff = now - CRAWL_WEDGE_SECS * 1000;
68 let lh_cutoff = now - LH_WEDGE_SECS * 1000;
69
70 sqlx::query(
71 "UPDATE properties SET crawl_state = 'idle', last_crawl_error = 'Crawl timed out or was interrupted' \
72 WHERE crawl_state = 'running' AND crawl_started_at IS NOT NULL AND crawl_started_at < ?",
73 )
74 .bind(crawl_cutoff)
75 .execute(pool)
76 .await?;
77
78 sqlx::query(
79 "UPDATE properties SET lighthouse_state = 'idle', last_lighthouse_error = 'Lighthouse run timed out or was interrupted' \
80 WHERE lighthouse_state = 'running' AND lighthouse_started_at IS NOT NULL AND lighthouse_started_at < ?",
81 )
82 .bind(lh_cutoff)
83 .execute(pool)
84 .await?;
85 Ok(())
86}
87
88async fn maybe_cleanup(pool: &SqlitePool, last: &mut Option<i64>) -> sqlx::Result<()> {
89 let now = now_ms();
90 if let Some(t) = *last {
91 if (now - t) / 1000 < CLEANUP_INTERVAL_SECS {
92 return Ok(());
93 }
94 }
95 let cutoff = now - CHECK_RETENTION_DAYS * 24 * 3600 * 1000;
96 let res = sqlx::query("DELETE FROM checks WHERE created_at < ?")
97 .bind(cutoff)
98 .execute(pool)
99 .await?;
100 tracing::info!("[scheduler] cleaned {} checks older than {}d", res.rows_affected(), CHECK_RETENTION_DAYS);
101 *last = Some(now);
102 Ok(())
103}
104
105async fn enqueue_status(
106 pool: &SqlitePool,
107 config: &Arc<Config>,
108 sem: &Arc<Semaphore>,
109) -> sqlx::Result<()> {
110 let now = now_ms();
111 let due: Vec<PropertyRow> = sqlx::query_as(
112 "SELECT * FROM properties \
113 WHERE last_run_at IS NULL OR next_run_at IS NULL OR next_run_at <= ?",
114 )
115 .bind(now)
116 .fetch_all(pool)
117 .await?;
118 for prop in due {
119 let next = checker::next_3min_boundary();
120 sqlx::query(
121 "UPDATE properties SET next_run_at = ?, last_run_at = ?, updated_at = ? WHERE id = ?",
122 )
123 .bind(next)
124 .bind(now)
125 .bind(now)
126 .bind(prop.id.clone())
127 .execute(pool)
128 .await?;
129
130 let pool = pool.clone();
131 let sem = sem.clone();
132 let config = config.clone();
133 tokio::spawn(async move {
134 let _permit = match sem.acquire_owned().await {
135 Ok(p) => p,
136 Err(_) => return,
137 };
138 tracing::info!("[scheduler] checking status {}", prop.url);
139 if let Err(e) = checker::process_check(&pool, &config, &prop).await {
140 tracing::warn!("[scheduler] status check failed for {}: {e:#}", prop.url);
141 }
142 });
143 }
144 Ok(())
145}
146
147async fn enqueue_lighthouse(
148 pool: &SqlitePool,
149 config: &Arc<Config>,
150 sem: &Arc<Semaphore>,
151) -> sqlx::Result<()> {
152 let now = now_ms();
153 let due: Vec<PropertyRow> = sqlx::query_as(
154 "SELECT * FROM properties \
155 WHERE (last_lighthouse_run_at IS NULL OR next_lighthouse_run_at IS NULL OR next_lighthouse_run_at <= ?) \
156 AND lighthouse_state NOT IN ('queued', 'running')",
157 )
158 .bind(now)
159 .fetch_all(pool)
160 .await?;
161
162 for prop in due {
163 let next = now + LH_INTERVAL_DAYS * 24 * 3600 * 1000;
164 sqlx::query(
165 "UPDATE properties SET next_lighthouse_run_at = ?, last_lighthouse_run_at = ?, lighthouse_state = 'queued', updated_at = ? \
166 WHERE id = ?",
167 )
168 .bind(next)
169 .bind(now)
170 .bind(now)
171 .bind(prop.id.clone())
172 .execute(pool)
173 .await?;
174
175 let pool = pool.clone();
176 let sem = sem.clone();
177 let config = config.clone();
178 tokio::spawn(async move {
179 let _permit = match sem.acquire_owned().await {
180 Ok(p) => p,
181 Err(_) => return,
182 };
183 tracing::info!("[scheduler] lighthouse {}", prop.url);
184 run_lighthouse_for(&pool, &config, &prop).await;
185 });
186 }
187 Ok(())
188}
189
190async fn run_lighthouse_for(pool: &SqlitePool, config: &Arc<Config>, prop: &PropertyRow) {
191 let now = now_ms();
192 let _ = sqlx::query(
193 "UPDATE properties SET lighthouse_state = 'running', lighthouse_started_at = ?, updated_at = ? WHERE id = ?",
194 )
195 .bind(now)
196 .bind(now)
197 .bind(prop.id.clone())
198 .execute(pool)
199 .await;
200
201 let started = std::time::Instant::now();
202 match lighthouse::fetch(&config.root, &prop.url).await {
203 Ok(results) => {
204 match lighthouse::parse_scores(&results) {
205 Ok(scores) => {
206 let details = lighthouse::parse_details(&results);
207 let scores_json = serde_json::to_string(&scores).unwrap_or_else(|_| "{}".into());
208 let details_json = details
209 .as_ref()
210 .and_then(|d| serde_json::to_string(d).ok())
211 .unwrap_or_else(|| "null".into());
212 let dur = started.elapsed().as_millis() as i64;
213 let _ = sqlx::query(
214 "UPDATE properties SET \
215 lighthouse_scores = ?, lighthouse_details = ?, \
216 last_lighthouse_success_at = ?, last_lighthouse_error = NULL, \
217 last_lighthouse_duration_ms = ?, lighthouse_state = 'idle', updated_at = ? \
218 WHERE id = ?",
219 )
220 .bind(scores_json)
221 .bind(details_json)
222 .bind(now_ms())
223 .bind(dur)
224 .bind(now_ms())
225 .bind(prop.id.clone())
226 .execute(pool)
227 .await;
228 }
229 Err(e) => store_lh_error(pool, prop, &format!("{e}"), started).await,
230 }
231 }
232 Err(e) => store_lh_error(pool, prop, &format!("{e}"), started).await,
233 }
234}
235
236async fn store_lh_error(
237 pool: &SqlitePool,
238 prop: &PropertyRow,
239 msg: &str,
240 started: std::time::Instant,
241) {
242 tracing::warn!("[scheduler] lighthouse failed for {}: {msg}", prop.url);
243 let dur = started.elapsed().as_millis() as i64;
244 let _ = sqlx::query(
245 "UPDATE properties SET lighthouse_state = 'idle', last_lighthouse_error = ?, last_lighthouse_duration_ms = ?, updated_at = ? \
246 WHERE id = ?",
247 )
248 .bind(msg)
249 .bind(dur)
250 .bind(now_ms())
251 .bind(prop.id.clone())
252 .execute(pool)
253 .await;
254}
255
256async fn enqueue_crawler(
257 pool: &SqlitePool,
258 _config: &Arc<Config>,
259 sem: &Arc<Semaphore>,
260) -> sqlx::Result<()> {
261 let now = now_ms();
262 let due: Vec<PropertyRow> = sqlx::query_as(
263 "SELECT * FROM properties \
264 WHERE (last_run_at_crawler IS NULL OR next_run_at_crawler IS NULL OR next_run_at_crawler <= ?) \
265 AND crawl_state NOT IN ('queued', 'running')",
266 )
267 .bind(now)
268 .fetch_all(pool)
269 .await?;
270
271 for prop in due {
272 let next = now + CRAWL_INTERVAL_DAYS * 24 * 3600 * 1000;
273 sqlx::query(
274 "UPDATE properties SET next_run_at_crawler = ?, last_run_at_crawler = ?, crawl_state = 'queued', updated_at = ? \
275 WHERE id = ?",
276 )
277 .bind(next)
278 .bind(now)
279 .bind(now)
280 .bind(prop.id.clone())
281 .execute(pool)
282 .await?;
283
284 let pool = pool.clone();
285 let sem = sem.clone();
286 tokio::spawn(async move {
287 let _permit = match sem.acquire_owned().await {
288 Ok(p) => p,
289 Err(_) => return,
290 };
291 tracing::info!("[scheduler] crawler {}", prop.url);
292 run_crawler_for(&pool, &prop).await;
293 });
294 }
295 Ok(())
296}
297
298async fn run_crawler_for(pool: &SqlitePool, prop: &PropertyRow) {
299 let now = now_ms();
300 let _ = sqlx::query(
301 "UPDATE properties SET crawl_state = 'running', crawl_started_at = ?, last_crawl_pages_count = 0, updated_at = ? WHERE id = ?",
302 )
303 .bind(now)
304 .bind(now)
305 .bind(prop.id.clone())
306 .execute(pool)
307 .await;
308
309 let pool_for_progress = pool.clone();
310 let id_for_progress = prop.id.clone();
311 let progress_cb = move |pages: usize| {
312 let pool = pool_for_progress.clone();
313 let id = id_for_progress.clone();
314 tokio::spawn(async move {
315 let _ = sqlx::query(
316 "UPDATE properties SET last_crawl_pages_count = ? WHERE id = ?",
317 )
318 .bind(pages as i64)
319 .bind(id)
320 .execute(&pool)
321 .await;
322 });
323 };
324
325 let started = std::time::Instant::now();
326 let outcome = crawler::run_seo_spider(&prop.url, progress_cb).await;
327 let duration_ms = started.elapsed().as_millis() as i64;
328 match outcome {
329 Ok(insights) => {
330 let json = serde_json::to_string(&insights).unwrap_or_else(|_| "[]".into());
331 let _ = sqlx::query(
332 "UPDATE properties SET \
333 crawler_insights = ?, crawl_state = 'idle', \
334 last_crawl_success_at = ?, last_crawl_error = NULL, \
335 last_crawl_duration_ms = ?, updated_at = ? \
336 WHERE id = ?",
337 )
338 .bind(json)
339 .bind(now_ms())
340 .bind(duration_ms)
341 .bind(now_ms())
342 .bind(prop.id.clone())
343 .execute(pool)
344 .await;
345 }
346 Err(e) => {
347 tracing::warn!("[scheduler] crawl failed for {}: {e:#}", prop.url);
348 let _ = sqlx::query(
349 "UPDATE properties SET crawl_state = 'idle', last_crawl_error = ?, last_crawl_duration_ms = ?, updated_at = ? WHERE id = ?",
350 )
351 .bind(format!("{e:#}"))
352 .bind(duration_ms)
353 .bind(now_ms())
354 .bind(prop.id.clone())
355 .execute(pool)
356 .await;
357 }
358 }
359}