diff --git a/Cargo.lock b/Cargo.lock index 845f236..cf90b63 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -272,7 +272,7 @@ dependencies = [ [[package]] name = "fluxqueue-worker" -version = "0.2.1" +version = "0.3.0-rc1" dependencies = [ "anyhow", "chrono", @@ -297,7 +297,7 @@ dependencies = [ [[package]] name = "fluxqueue_core" -version = "0.2.1" +version = "0.3.0-rc1" dependencies = [ "deadpool-redis", "fluxqueue-common", diff --git a/crates/fluxqueue-core/Cargo.toml b/crates/fluxqueue-core/Cargo.toml index 592c692..53ab80a 100644 --- a/crates/fluxqueue-core/Cargo.toml +++ b/crates/fluxqueue-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "fluxqueue_core" -version = "0.2.1" +version = "0.3.0-rc1" repository.workspace = true homepage.workspace = true authors.workspace = true diff --git a/crates/fluxqueue-core/src/lib.rs b/crates/fluxqueue-core/src/lib.rs index a460584..44cf2d6 100644 --- a/crates/fluxqueue-core/src/lib.rs +++ b/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::()?; Ok(()) } diff --git a/crates/fluxqueue-worker/Cargo.toml b/crates/fluxqueue-worker/Cargo.toml index aceb0ee..662448d 100644 --- a/crates/fluxqueue-worker/Cargo.toml +++ b/crates/fluxqueue-worker/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "fluxqueue-worker" -version = "0.2.1" +version = "0.3.0-rc1" edition = "2024" [lib] diff --git a/crates/fluxqueue-worker/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index b18c0aa..acf824b 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/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; @@ -18,6 +20,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 +338,53 @@ fn check_worker_is_ready(ready_check: ReadyCheck) { } } +fn check_client_library_version() -> Result<()> { + let worker_version = env!("CARGO_PKG_VERSION"); + let library_version = Python::attach(|py| -> Result { + let module = py.import("fluxqueue")?; + let version = module.getattr("__version__")?; + Ok(version.to_string()) + })?; + + let comparison = compare_versions(worker_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.", + worker_version, + &library_version + ); + } + + if comparison == 1 { + return Err(anyhow!( + "Minimum required client library version is: {}, found: {}", + worker_version, + &library_version + )); + } + + Ok(()) +} + +fn compare_versions(v1: &str, v2: &str) -> i8 { + let mut parts1: Vec = v1.split('.').map(|p| p.parse().unwrap_or(0)).collect(); + let mut parts2: Vec = 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::*; diff --git a/python/fluxqueue/__init__.py b/python/fluxqueue/__init__.py index b0256af..de2f3ed 100644 --- a/python/fluxqueue/__init__.py +++ b/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