yggdrasil/tasks/
cache_sync.rs

1//! PostgreSQL 通知驱动双机缓存失效，持久化版本与定时核对负责断线补偿。
2use std::time::Duration;
3
4use tokio_postgres::{AsyncMessage, Client, NoTls};
5
6const CHANNEL: &str = "ygg_content_changed";
7const RECHECK_INTERVAL: Duration = Duration::from_secs(5);
8
9struct ConnectionTask(tokio::task::JoinHandle<()>);
10impl Drop for ConnectionTask {
11    fn drop(&mut self) {
12        self.0.abort();
13    }
14}
15
16async fn refresh_revision(
17    client: &Client,
18    observed: &mut Option<i64>,
19) -> Result<(), tokio_postgres::Error> {
20    let revision: i64 = client
21        .query_one(
22            "SELECT revision FROM content_cache_revision WHERE singleton",
23            &[],
24        )
25        .await?
26        .get(0);
27    if *observed != Some(revision) {
28        crate::cache::invalidate_shared_content();
29        *observed = Some(revision);
30        tracing::debug!(revision, "共享内容缓存已同步失效");
31    }
32    Ok(())
33}
34
35async fn listen(
36    config: &tokio_postgres::Config,
37) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
38    let (client, mut connection) = config.connect(NoTls).await?;
39    let (tx, mut rx) = tokio::sync::mpsc::channel(1);
40    let _driver = ConnectionTask(tokio::spawn(async move {
41        while let Some(message) = futures::future::poll_fn(|cx| connection.poll_message(cx)).await {
42            match message {
43                Ok(AsyncMessage::Notification(notification))
44                    if notification.channel() == CHANNEL =>
45                {
46                    // 版本查询会合并突发变更；通道满时无需再排队。
47                    let _ = tx.try_send(());
48                }
49                Err(_) => break,
50                _ => {}
51            }
52        }
53    }));
54    // 先监听再读取版本，避免读取版本和 LISTEN 之间漏掉提交。
55    client.batch_execute("LISTEN ygg_content_changed").await?;
56    let mut observed = None;
57    refresh_revision(&client, &mut observed).await?;
58    tracing::info!("双机内容缓存通知已连接");
59    let mut interval = tokio::time::interval(RECHECK_INTERVAL);
60    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
61    loop {
62        tokio::select! {
63            signal = rx.recv() => {
64                if signal.is_none() { return Err(std::io::Error::other("cache notification connection closed").into()); }
65            }
66            _ = interval.tick() => {}
67        }
68        refresh_revision(&client, &mut observed).await?;
69    }
70}
71
72pub async fn run() {
73    let Ok(mut config) = crate::db::pool::build_pg_config() else {
74        tracing::error!("无法初始化内容缓存通知：数据库配置无效");
75        return;
76    };
77    config.connect_timeout(Duration::from_secs(5));
78    config.application_name("yggdrasil-cache-sync");
79    config.options("-c statement_timeout=5000");
80    loop {
81        if let Err(error) = listen(&config).await {
82            tracing::warn!(%error, "内容缓存通知断开，5 秒后重连；HTML TTL 继续兜底");
83            crate::cache::invalidate_shared_content();
84        }
85        tokio::time::sleep(RECHECK_INTERVAL).await;
86    }
87}
yggdrasil/tasks/cache_sync.rs

yggdrasil/tasks/
cache_sync.rs
