| 1351 | } |
| 1352 | |
| 1353 | async fn monitor_upstream( |
| 1354 | client: reqwest::Client, |
| 1355 | state: Arc<AppState>, |
| 1356 | mut rx: broadcast::Receiver<bool>, |
| 1357 | ) { |
| 1358 | info!(immediate = true, "Spawning upstream monitor task ..."); |
| 1359 | |
| 1360 | loop { |
| 1361 | tokio::select! { |
| 1362 | _ = rx.recv() => { |
| 1363 | break; |
| 1364 | } |
| 1365 | _ = sleep(Duration::from_secs(120)) => { |
| 1366 | for upstream_backend in state.cache.monitor_backends() { |
| 1367 | let url_str = upstream_backend.provider.as_str(); |
| 1368 | async { |
| 1369 | debug!("upstream checking -> {}", upstream_backend.provider); |
| 1370 | let r = client |
| 1371 | .head(url_str) |
| 1372 | .timeout(std::time::Duration::from_secs(8)) |
| 1373 | .send() |
| 1374 | .await |
| 1375 | .map(|resp| { |
| 1376 | debug!("upstream check -> {:?}", resp.status()); |
| 1377 | resp.status() == StatusCode::OK |
| 1378 | || resp.status() == StatusCode::FORBIDDEN |
| 1379 | }) |
| 1380 | .unwrap_or_else(|resp| { |
| 1381 | debug!(?resp); |
| 1382 | debug!( |
| 1383 | "upstream err check -> {:?}", |
| 1384 | resp.status() |
| 1385 | ); |
| 1386 | resp.status() == Some(StatusCode::OK) |
| 1387 | || resp.status() == Some(StatusCode::FORBIDDEN) |
| 1388 | }); |
| 1389 | |
| 1390 | upstream_backend.online.store(r, Ordering::Relaxed); |
| 1391 | info!("upstream online -> {}", r); |
| 1392 | } |
| 1393 | .instrument(tracing::info_span!("monitor_upstream", upstream = url_str)) |
| 1394 | .await; |
| 1395 | } // end for |
| 1396 | } |
| 1397 | } |
| 1398 | } |
| 1399 | |
| 1400 | info!(immediate = true, "Stopping upstream monitor task."); |
| 1401 | } |
| 1402 | |
| 1403 | struct PrefetchReq { |
| 1404 | fetch_url: Url, |