From 39da52132753a7bed99b220e0f67d75458e2061c Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sat, 28 Feb 2026 17:53:41 +0400 Subject: [PATCH 1/3] feat(worker): Add client library version check before worker initialization --- crates/fluxqueue-core/src/lib.rs | 1 + crates/fluxqueue-worker/src/worker.rs | 57 ++++++++++++++++++++++++++- python/fluxqueue/__init__.py | 1 + 3 files changed, 58 insertions(+), 1 deletion(-) 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/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index b18c0aa..ee8a7a1 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; @@ -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, 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 { + 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 = 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 From 9d17b892a750a9a2a7b8bf9bd335a08569815ca5 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sat, 28 Feb 2026 18:07:57 +0400 Subject: [PATCH 2/3] Update versions --- Cargo.lock | 4 ++-- crates/fluxqueue-core/Cargo.toml | 2 +- crates/fluxqueue-worker/Cargo.toml | 2 +- crates/fluxqueue-worker/src/worker.rs | 10 +++++----- 4 files changed, 9 insertions(+), 9 deletions(-) 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-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 ee8a7a1..a2ac40d 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/crates/fluxqueue-worker/src/worker.rs @@ -12,8 +12,6 @@ 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, concurrency: usize, @@ -347,11 +345,13 @@ fn check_client_library_version() -> Result<()> { Ok(version.to_string()) })?; - let comparison = compare_versions(MIN_CLIENT_LIB_VERSION, &library_version); + let worker_version = env!("CARGO_PKG_VERSION"); + + 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.", - MIN_CLIENT_LIB_VERSION, + worker_version, &library_version ); } @@ -359,7 +359,7 @@ fn check_client_library_version() -> Result<()> { if comparison == 1 { return Err(anyhow!( "Minimum required client library version is: {}, found: {}", - MIN_CLIENT_LIB_VERSION, + worker_version, &library_version )); } From d62fc8976587be12a605686f5fa184a5630d06ea Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sat, 28 Feb 2026 18:09:26 +0400 Subject: [PATCH 3/3] Update --- crates/fluxqueue-worker/src/worker.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/crates/fluxqueue-worker/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index a2ac40d..acf824b 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/crates/fluxqueue-worker/src/worker.rs @@ -339,14 +339,13 @@ 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 worker_version = env!("CARGO_PKG_VERSION"); - let comparison = compare_versions(worker_version, &library_version); if comparison == -1 { tracing::warn!(