feat(worker): Add client library version check before worker initialization

Author: CCXLV

crates/fluxqueue-core/src/lib.rs

@@ -76,6 +76,7 @@ impl FluxQueueCore {
 
 #[pymodule]
 fn _core(m: &Bound<'_, PyModule>) -> PyResult<()> {
+    m.add("__version__", env!("CARGO_PKG_VERSION"))?;
     m.add_class::<FluxQueueCore>()?;
     Ok(())
 }

crates/fluxqueue-worker/src/worker.rs

@@ -1,4 +1,6 @@
-use anyhow::Result;
+use anyhow::{Result, anyhow};
+use pyo3::Python;
+use pyo3::types::PyAnyMethods;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::time::Duration;
@@ -10,6 +12,8 @@ use crate::redis_client::RedisClient;
 use crate::task::{PythonDispatcher, TaskData, TaskRegistry};
 use fluxqueue_common::{Task, deserialize_raw_task_data};
 
+static MIN_CLIENT_LIB_VERSION: &str = "0.3.0";
+
 pub async fn run_worker(
     mut shutdown: watch::Receiver<bool>,
     concurrency: usize,
@@ -18,6 +22,11 @@ pub async fn run_worker(
     queue_name: String,
     save_dead_tasks: bool,
 ) -> Result<()> {
+    check_client_library_version().map_err(|e| {
+        tracing::error!("{}", e);
+        std::process::exit(1);
+    })?;
+
     let redis_client = RedisClient::new(&redis_url).await.map_err(|e| {
         tracing::error!("{}", e);
         std::process::exit(1);
@@ -331,6 +340,52 @@ fn check_worker_is_ready(ready_check: ReadyCheck) {
     }
 }
 
+fn check_client_library_version() -> Result<()> {
+    let library_version = Python::attach(|py| -> Result<String> {
+        let module = py.import("fluxqueue")?;
+        let version = module.getattr("__version__")?;
+        Ok(version.to_string())
+    })?;
+
+    let comparison = compare_versions(MIN_CLIENT_LIB_VERSION, &library_version);
+    if comparison == -1 {
+        tracing::warn!(
+            "Worker version '{}' is older than client library '{}'. For full functionality, update the worker to match the client version.",
+            MIN_CLIENT_LIB_VERSION,
+            &library_version
+        );
+    }
+
+    if comparison == 1 {
+        return Err(anyhow!(
+            "Minimum required client library version is: {}, found: {}",
+            MIN_CLIENT_LIB_VERSION,
+            &library_version
+        ));
+    }
+
+    Ok(())
+}
+
+fn compare_versions(v1: &str, v2: &str) -> i8 {
+    let mut parts1: Vec<u32> = v1.split('.').map(|p| p.parse().unwrap_or(0)).collect();
+    let mut parts2: Vec<u32> = v2.split('.').map(|p| p.parse().unwrap_or(0)).collect();
+
+    let len = parts1.len().max(parts2.len());
+    parts1.resize(len, 0);
+    parts2.resize(len, 0);
+
+    for (a, b) in parts1.iter().zip(parts2.iter()) {
+        if a > b {
+            return 1;
+        }
+        if a < b {
+            return -1;
+        }
+    }
+    0
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

python/fluxqueue/__init__.py

@@ -1,3 +1,4 @@
+from ._core import __version__ as __version__
 from .client import FluxQueue as FluxQueue
 from .context import Context as Context
 from .models import TaskMetadata as TaskMetadata