wowlab_sentinel/
presence.rs

1use async_trait::async_trait;
2use wowlab_centrifuge::error_codes;
3use wowlab_common::{NodePublicKey, NodeRealtimeMessage};
4use wowlab_types::sim::FastSet;
5
6use crate::{cron::CronJob, state::ServerState};
7
8crate::db_client!(PresenceDb, "presence", max = 2);
9
10/// Syncs node online/offline status between Centrifugo presence and the database.
11#[derive(Debug)]
12pub(crate) struct PresenceJob {
13    schedule: String,
14}
15
16impl PresenceJob {
17    pub(crate) fn new(schedule: &str) -> Self {
18        Self {
19            schedule: schedule.to_string(),
20        }
21    }
22}
23
24#[async_trait]
25impl CronJob for PresenceJob {
26    fn name(&self) -> &'static str {
27        "presence_sync"
28    }
29
30    fn schedule(&self) -> &str {
31        &self.schedule
32    }
33
34    async fn run(&self, state: &ServerState) {
35        let centrifugo_online: FastSet<NodePublicKey> =
36            match state.presence.get_online("nodes:online").await {
37                Ok(keys) => keys.into_iter().collect(),
38                Err(error) if error.server_code() == Some(error_codes::UNKNOWN_CHANNEL) => {
39                    FastSet::default()
40                }
41                Err(e) => {
42                    tracing::warn!(error = %e, "Failed to poll presence");
43
44                    return;
45                }
46            };
47
48        let db_online: FastSet<NodePublicKey> =
49            match fetch_online_node_keys(state.dbs.get::<PresenceDb>()).await {
50                Ok(keys) => keys.into_iter().collect(),
51                Err(e) => {
52                    tracing::warn!(error = %e, "Failed to fetch online nodes from DB");
53
54                    return;
55                }
56            };
57
58        tracing::info!(
59            centrifugo = centrifugo_online.len(),
60            db = db_online.len(),
61            "Presence sync"
62        );
63        tracing::debug!(nodes = ?centrifugo_online, "Centrifugo presence snapshot");
64
65        let to_online: Vec<NodePublicKey> =
66            centrifugo_online.difference(&db_online).cloned().collect();
67        let to_offline: Vec<NodePublicKey> =
68            db_online.difference(&centrifugo_online).cloned().collect();
69
70        if !to_online.is_empty() {
71            set_online(state, &to_online).await;
72        }
73
74        if !to_offline.is_empty() {
75            set_offline(state, &to_offline).await;
76        }
77
78        #[expect(
79            clippy::cast_precision_loss,
80            reason = "metrics gauges accept f64 and online node counts remain exactly representable"
81        )]
82        let online_count = centrifugo_online.len() as f64;
83
84        metrics::gauge!(crate::telemetry::NODES_ONLINE).set(online_count);
85        metrics::counter!(crate::telemetry::PRESENCE_POLLS).increment(1);
86        state.touch_presence();
87    }
88}
89
90fn keys_to_strings(keys: &[NodePublicKey]) -> Vec<String> {
91    keys.iter().map(ToString::to_string).collect()
92}
93
94async fn fetch_online_node_keys(db: &sqlx::PgPool) -> Result<Vec<NodePublicKey>, sqlx::Error> {
95    struct Row {
96        public_key: String,
97    }
98    let rows = sqlx::query_file_as!(Row, "queries/presence_fetch_online_keys.sql")
99        .fetch_all(db)
100        .await?;
101
102    Ok(rows
103        .into_iter()
104        .filter_map(|r| r.public_key.parse().ok())
105        .collect())
106}
107
108async fn set_online(state: &ServerState, keys: &[NodePublicKey]) {
109    let keys_str = keys_to_strings(keys);
110    let result = sqlx::query_file!("queries/presence_set_online.sql", &keys_str)
111        .execute(state.dbs.get::<PresenceDb>())
112        .await;
113
114    let Ok(r) = result else {
115        tracing::warn!(error = ?result.unwrap_err(), nodes = ?keys, "Failed to mark nodes online");
116
117        return;
118    };
119
120    if r.rows_affected() == 0 {
121        return;
122    }
123
124    tracing::info!(count = r.rows_affected(), "Nodes online");
125    publish_node_updates(state, keys).await;
126}
127
128async fn set_offline(state: &ServerState, keys: &[NodePublicKey]) {
129    let keys_str = keys_to_strings(keys);
130
131    let result = sqlx::query_file!("queries/presence_set_offline.sql", &keys_str)
132        .execute(state.dbs.get::<PresenceDb>())
133        .await;
134
135    let Ok(r) = result else {
136        tracing::warn!(error = ?result.unwrap_err(), nodes = ?keys, "Failed to mark nodes offline");
137
138        return;
139    };
140
141    if r.rows_affected() == 0 {
142        return;
143    }
144
145    tracing::info!(count = r.rows_affected(), "Nodes offline");
146    metrics::counter!(crate::telemetry::NODES_MARKED_OFFLINE).increment(r.rows_affected());
147    publish_node_updates(state, keys).await;
148}
149
150async fn publish_node_updates(state: &ServerState, keys: &[NodePublicKey]) {
151    let payload = refinement_update(keys);
152
153    state.publish("nodes:all", &payload).await;
154
155    for pk in keys {
156        // #t(rust_alloc_in_loop) per-node channel name
157        state.publish(&format!("nodes:{pk}"), &payload).await;
158    }
159}
160
161fn refinement_update(keys: &[NodePublicKey]) -> NodeRealtimeMessage {
162    NodeRealtimeMessage::RefinementUpdated {
163        ids: keys
164            .iter()
165            .map(|key| key.to_string().into_boxed_str())
166            .collect(),
167    }
168}
169
170#[cfg(test)]
171mod tests {
172    use googletest::prelude::*;
173
174    use super::*;
175
176    #[gtest]
177    fn publishes_the_shared_refinement_updated_shape() -> Result<()> {
178        let key = NodePublicKey::from_bytes([7; 32]);
179
180        let value =
181            serde_json::to_value(refinement_update(std::slice::from_ref(&key))).or_fail()?;
182
183        verify_eq!(
184            value,
185            serde_json::json!({
186                "type": "refinement_updated",
187                "payload": { "ids": [key.to_string()] }
188            })
189        )?;
190
191        Ok(())
192    }
193}
wowlab_sentinel/presence.rs

wowlab_sentinel/
presence.rs
