wowlab_node/core/
mod.rs

1use std::{sync::Arc, time::Duration};
2
3use tokio::sync::mpsc;
4use tokio_util::sync::CancellationToken;
5use uuid::Uuid;
6use wowlab_engine_ports::{ContentCatalog, DataResolver, DynDataResolver};
7
8use crate::{
9    ConnectionStatus, NodeState, WorkBatchResult, WorkerPool, claim, config::NodeConfig,
10    realtime::RealtimeEvent, sentinel::SentinelClient, utils::backoff::ExponentialBackoff,
11    work_context::WorkContextCache,
12};
13
14mod connection;
15mod lifecycle;
16mod unlink;
17mod work;
18
19pub use unlink::{
20    LocalIdentityOutcome, RemoteUnlinkOutcome, UnlinkError, UnlinkErrorCategory, UnlinkOutcome,
21};
22
23const EVENT_CHANNEL_SIZE: usize = 32;
24const INITIAL_BACKOFF_SECS: u64 = 5;
25const MAX_BACKOFF_SECS: u64 = 300;
26
27fn validated_capacity(total_cores: i32, max_parallel: i32) -> Option<(u32, usize)> {
28    let total_cores = u32::try_from(total_cores).ok().filter(|cores| *cores > 0)?;
29    let max_parallel = u32::try_from(max_parallel)
30        .ok()
31        .filter(|workers| *workers > 0)?;
32    let max_parallel = usize::try_from(total_cores.min(max_parallel)).ok()?;
33
34    Some((total_cores, max_parallel))
35}
36
37/// Events emitted by `NodeCore`.
38#[derive(Debug)]
39// #t(rust_non_exhaustive_on_public) internal event enum matched exhaustively in node-headless and node-gui
40pub enum NodeCoreEvent {
41    StateChanged(NodeState),
42    ConnectionChanged(ConnectionStatus),
43    ChunkAssigned {
44        job_id: Uuid,
45        chunk_index: i32,
46        iterations: i32,
47    },
48    ChunkCompleted {
49        job_id: Uuid,
50        chunk_index: i32,
51        mean_dps: f32,
52    },
53    ChunkFailed {
54        job_id: Uuid,
55        chunk_index: i32,
56        error: String,
57    },
58    Error(String),
59    UnlinkCompleted(Result<UnlinkOutcome, UnlinkError>),
60}
61
62/// Owned event stream emitted by a [`NodeCore`].
63#[derive(Debug)]
64pub struct NodeEvents {
65    receiver: mpsc::Receiver<NodeCoreEvent>,
66    unlink_receiver: mpsc::UnboundedReceiver<NodeCoreEvent>,
67}
68
69impl NodeEvents {
70    fn new(
71        receiver: mpsc::Receiver<NodeCoreEvent>,
72        unlink_receiver: mpsc::UnboundedReceiver<NodeCoreEvent>,
73    ) -> Self {
74        Self {
75            receiver,
76            unlink_receiver,
77        }
78    }
79
80    pub fn try_recv(&mut self) -> Option<NodeCoreEvent> {
81        self.unlink_receiver
82            .try_recv()
83            .ok()
84            .or_else(|| self.receiver.try_recv().ok())
85    }
86}
87
88enum RegisterResult {
89    Success,
90    Failed(String),
91}
92
93type VerifyReceiver =
94    mpsc::Receiver<Result<wowlab_types::sensitive::Sensitive<String>, VerifyError>>;
95
96#[derive(Debug, thiserror::Error)]
97#[non_exhaustive]
98pub(crate) enum VerifyError {
99    #[error("node not found")]
100    NotFound,
101    #[error("server unavailable")]
102    Unavailable,
103}
104
105/// Host application metadata attached to node registration requests.
106#[derive(Clone, Debug)]
107// #t(rust_similar_structs) Node registration identity is distinct from doc-generation package metadata.
108pub struct NodeApplication {
109    name: String,
110    version: String,
111}
112
113impl NodeApplication {
114    #[must_use]
115    pub fn new(name: impl Into<String>, version: impl Into<String>) -> Self {
116        Self {
117            name: name.into(),
118            version: version.into(),
119        }
120    }
121}
122
123/// Event-driven node controller.
124pub struct NodeCore {
125    runtime: tokio::runtime::Runtime,
126    config: NodeConfig,
127    sentinel: SentinelClient,
128    worker_pool: WorkerPool,
129    state: NodeState,
130    connection_status: ConnectionStatus,
131    registered: bool,
132    node_name: String,
133    total_cores: u32,
134
135    app_name: String,
136    app_version: String,
137
138    work_context_cache: WorkContextCache,
139
140    verify_rx: Option<VerifyReceiver>,
141    register_rx: Option<mpsc::Receiver<RegisterResult>>,
142    realtime_rx: Option<mpsc::Receiver<RealtimeEvent>>,
143    realtime_shutdown: Option<CancellationToken>,
144    result_rx: Option<mpsc::Receiver<WorkBatchResult>>,
145    unlink_rx: Option<unlink::UnlinkReceiver>,
146    unlink_cancel: Option<CancellationToken>,
147    unlink_origin: Option<unlink::UnlinkOrigin>,
148
149    backoff: ExponentialBackoff,
150
151    event_tx: mpsc::Sender<NodeCoreEvent>,
152    unlink_event_tx: mpsc::UnboundedSender<NodeCoreEvent>,
153
154    started: bool,
155}
156
157impl NodeCore {
158    /// Construct a node using a caller-supplied data resolver.
159    ///
160    /// # Errors
161    ///
162    /// Returns an error when the Tokio runtime or signed Sentinel client cannot be initialized.
163    pub fn with_resolver<R>(
164        config: NodeConfig,
165        resolver: R,
166        application: NodeApplication,
167        catalog: &'static ContentCatalog,
168    ) -> Result<(Self, NodeEvents), crate::sentinel::SentinelError>
169    where
170        R: DataResolver + Send + Sync + 'static,
171    {
172        let runtime = tokio::runtime::Runtime::new()
173            .map_err(|error| crate::sentinel::SentinelError::client_build(error.to_string()))?;
174        let resolver = DynDataResolver::new_arc(resolver);
175
176        Self::new(runtime, resolver, config, application, catalog)
177    }
178
179    /// Construct a node wired to the Supabase data resolver.
180    ///
181    /// # Errors
182    ///
183    /// Returns an error when the Tokio runtime, Supabase client, metadata lookup, or data resolver cannot be initialized.
184    #[cfg(feature = "supabase")]
185    pub fn with_supabase(
186        config: NodeConfig,
187        application: NodeApplication,
188        catalog: &'static ContentCatalog,
189    ) -> Result<(Self, NodeEvents), crate::sentinel::SentinelError> {
190        use wowlab_engine_adapter_data::{GameDataCache, SupabaseResolver};
191        use wowlab_supabase::SupabaseClient;
192
193        #[derive(serde::Deserialize)]
194        struct MetaRow {
195            patch_version: String,
196        }
197
198        let runtime = tokio::runtime::Runtime::new()
199            .map_err(|error| crate::sentinel::SentinelError::client_build(error.to_string()))?;
200        let engine_client = SupabaseClient::new(&config.supabase_url, &config.supabase_anon_key)?;
201
202        let meta: Vec<MetaRow> = runtime
203            .block_on(engine_client.get_json("meta?id=eq.1&select=patch_version", "game"))?;
204        let patch_version = meta
205            .into_iter()
206            .next()
207            .map(|m| m.patch_version)
208            .ok_or_else(|| {
209                crate::sentinel::SentinelError::client_build(
210                    "game.meta not seeded — run `wowlab snapshot sync` first",
211                )
212            })?;
213
214        let cache_dir = config.game_data_cache_dir.clone().ok_or_else(|| {
215            crate::sentinel::SentinelError::client_build(
216                "game-data cache directory is unavailable; set NODE_CACHE_DIR",
217            )
218        })?;
219        let engine_cache = GameDataCache::new(engine_client, patch_version, cache_dir)?;
220        let resolver: Arc<DynDataResolver<'static>> =
221            DynDataResolver::new_arc(SupabaseResolver::new(engine_cache));
222
223        Self::new(runtime, resolver, config, application, catalog)
224    }
225
226    fn new(
227        runtime: tokio::runtime::Runtime,
228        resolver: Arc<DynDataResolver<'static>>,
229        config: NodeConfig,
230        application: NodeApplication,
231        catalog: &'static ContentCatalog,
232    ) -> Result<(Self, NodeEvents), crate::sentinel::SentinelError> {
233        let mut config = config;
234        let sentinel = SentinelClient::new(config.sentinel_url.clone(), config.clone())?;
235        let (total_cores, enabled_cores) =
236            validated_capacity(config.total_cores, config.enabled_cores).ok_or_else(|| {
237                crate::sentinel::SentinelError::client_build(
238                    "NODE_TOTAL_CORES and NODE_ENABLED_CORES must be positive",
239                )
240            })?;
241
242        if let Some(ref token) = config.token_claim {
243            if let Err(msg) = claim::validate_token(token.expose()) {
244                let detail = if msg.is_empty() {
245                    "empty token".to_string()
246                } else {
247                    msg.to_string()
248                };
249
250                tracing::error!(%detail, "Invalid NODE_CLAIM_TOKEN");
251                config.token_claim = None;
252            }
253        }
254
255        let state = if config.token_claim.is_some() {
256            NodeState::Registering
257        } else {
258            NodeState::Verifying
259        };
260
261        let mut worker_pool = WorkerPool::new(enabled_cores, catalog);
262
263        worker_pool.set_dyn_resolver(resolver);
264
265        let (event_tx, event_rx) = mpsc::channel(EVENT_CHANNEL_SIZE);
266        let (unlink_event_tx, unlink_event_rx) = mpsc::unbounded_channel();
267
268        let core = Self {
269            runtime,
270            state,
271            sentinel,
272            worker_pool,
273            registered: false,
274            node_name: config.name.clone(),
275            total_cores,
276            app_name: application.name,
277            app_version: application.version,
278            work_context_cache: WorkContextCache::new(),
279            connection_status: ConnectionStatus::Connecting,
280            verify_rx: None,
281            register_rx: None,
282            realtime_rx: None,
283            realtime_shutdown: None,
284            result_rx: None,
285            unlink_rx: None,
286            unlink_cancel: None,
287            unlink_origin: None,
288            backoff: ExponentialBackoff::new(
289                Duration::from_secs(INITIAL_BACKOFF_SECS),
290                Duration::from_secs(MAX_BACKOFF_SECS),
291            ),
292            event_tx,
293            unlink_event_tx,
294            config,
295            started: false,
296        };
297
298        Ok((core, NodeEvents::new(event_rx, unlink_event_rx)))
299    }
300}
301
302impl std::fmt::Debug for NodeCore {
303    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
304        f.debug_struct("NodeCore")
305            .field("state", &self.state)
306            .field("connection_status", &self.connection_status)
307            .field("registered", &self.registered)
308            .field("node_name", &self.node_name)
309            .field("max_parallel", &self.worker_pool.stats().max_workers)
310            .field("total_cores", &self.total_cores)
311            .field("app_name", &self.app_name)
312            .field("app_version", &self.app_version)
313            .field("started", &self.started)
314            .finish_non_exhaustive()
315    }
316}
317
318#[cfg(test)]
319mod tests {
320    use googletest::prelude::*;
321
322    use super::validated_capacity;
323
324    #[gtest]
325    fn capacity_rejects_non_positive_values_and_caps_workers_to_total_cores() -> Result<()> {
326        verify_eq!(validated_capacity(8, 4), Some((8, 4)))?;
327        verify_eq!(validated_capacity(4, 8), Some((4, 4)))?;
328        verify_that!(validated_capacity(0, 4), none())?;
329        verify_that!(validated_capacity(4, 0), none())?;
330        verify_that!(validated_capacity(-1, 4), none())?;
331        verify_that!(validated_capacity(4, -8), none())?;
332
333        verify_that!(validated_capacity(i32::MIN, i32::MIN), none())
334    }
335}
wowlab_node/core/mod.rs

wowlab_node/core/
mod.rs
