From 96ba848ff6aca7fabad7bdb7fb357eeb1b98c731 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Fri, 20 Feb 2026 17:08:03 +0400 Subject: [PATCH 01/31] feat(context): Start working on contexts feature --- python/fluxqueue/__init__.py | 3 +- python/fluxqueue/_task.py | 69 ++++++++++++++++++++++++++++ python/fluxqueue/client.py | 64 ++++++++++++++------------ python/fluxqueue/context.py | 89 ++++++++++++++++++++++++++++++++++++ 4 files changed, 194 insertions(+), 31 deletions(-) create mode 100644 python/fluxqueue/_task.py create mode 100644 python/fluxqueue/context.py diff --git a/python/fluxqueue/__init__.py b/python/fluxqueue/__init__.py index 60393a9..f7d1134 100644 --- a/python/fluxqueue/__init__.py +++ b/python/fluxqueue/__init__.py @@ -1,3 +1,4 @@ -__all__ = ["FluxQueue"] +__all__ = ["Context", "FluxQueue"] from .client import FluxQueue +from .context import Context diff --git a/python/fluxqueue/_task.py b/python/fluxqueue/_task.py new file mode 100644 index 0000000..eea25fb --- /dev/null +++ b/python/fluxqueue/_task.py @@ -0,0 +1,69 @@ +import inspect +from collections.abc import Callable, Coroutine +from functools import wraps +from typing import Any, ParamSpec, cast, get_type_hints, overload + +from ._core import FluxQueueCore +from .utils import get_task_name + +P = ParamSpec("P") + + +@overload +def _task_decorator( + func: Callable[P, None], + *, + name: str | None, + queue: str, + max_retries: int, + core: FluxQueueCore, +) -> Callable[P, None]: ... + + +@overload +def _task_decorator( + func: Callable[P, Coroutine[Any, Any, None]], + *, + name: str | None, + queue: str, + max_retries: int, + core: FluxQueueCore, +) -> Callable[P, Coroutine[Any, Any, None]]: ... + + +def _task_decorator( + func: Callable[P, None | Coroutine[Any, Any, None]], + *, + name: str | None, + queue: str, + max_retries: int, + core: FluxQueueCore, +) -> Callable[P, None | Coroutine[Any, Any, None]]: + type_hints = get_type_hints(func) + return_type = type_hints.get("return") + + if return_type and return_type is not type(None): + raise TypeError(f"Task function must return None, got {return_type}") + + is_async = inspect.iscoroutinefunction(func) + task_name = get_task_name(func, name) + + cast(Any, func).task_name = task_name + cast(Any, func).queue = queue + + if is_async: + + @wraps(func) + async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> None: + await core._enqueue_async(task_name, queue, max_retries, args, kwargs) + return None + + return async_wrapper + else: + + @wraps(func) + def sync_wrapper(*args: P.args, **kwargs: P.kwargs) -> None: + core._enqueue(task_name, queue, max_retries, args, kwargs) + return None + + return sync_wrapper diff --git a/python/fluxqueue/client.py b/python/fluxqueue/client.py index f211e94..045a76f 100644 --- a/python/fluxqueue/client.py +++ b/python/fluxqueue/client.py @@ -1,10 +1,9 @@ -import inspect from collections.abc import Callable, Coroutine -from functools import wraps -from typing import Any, ParamSpec, cast, get_type_hints, overload +from typing import Any, Concatenate, ParamSpec, cast, overload from ._core import FluxQueueCore -from .utils import get_task_name +from ._task import _task_decorator +from .context import C, _with_context P = ParamSpec("P") @@ -62,35 +61,40 @@ def decorator( def decorator( func: Callable[P, None | Coroutine[Any, Any, None]], ) -> Callable[P, None | Coroutine[Any, Any, None]]: - type_hints = get_type_hints(func) - return_type = type_hints.get("return") + return _task_decorator( + cast(Any, func), + name=name, + queue=queue, + max_retries=max_retries, + core=self._core, + ) - if return_type and return_type is not type(None): - raise TypeError(f"Task function must return None, got {return_type}") - - is_async = inspect.iscoroutinefunction(func) - task_name = get_task_name(func, name) - - cast(Any, func).task_name = task_name - cast(Any, func).queue = queue - - if is_async: - - @wraps(func) - async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> None: - await self._core._enqueue_async( - task_name, queue, max_retries, args, kwargs - ) - return None + return decorator - return async_wrapper - else: + def task_with_context( + self, + *, + name: str | None = None, + queue: str = "default", + max_retries: int = 3, + ): + @overload + def decorator(func: Callable[Concatenate[C, P], None]) -> Callable[P, None]: ... - @wraps(func) - def sync_wrapper(*args: P.args, **kwargs: P.kwargs) -> None: - self._core._enqueue(task_name, queue, max_retries, args, kwargs) - return None + @overload + def decorator( + func: Callable[Concatenate[C, P], Coroutine[Any, Any, None]], + ) -> Callable[P, Coroutine[Any, Any, None]]: ... - return sync_wrapper + def decorator( + func: Callable[Concatenate[C, P], None | Coroutine[Any, Any, None]], + ) -> Callable[P, None | Coroutine[Any, Any, None]]: + return _with_context( + cast(Any, func), + name=name, + queue=queue, + max_retries=max_retries, + core=self._core, + ) return decorator diff --git a/python/fluxqueue/context.py b/python/fluxqueue/context.py new file mode 100644 index 0000000..8fee552 --- /dev/null +++ b/python/fluxqueue/context.py @@ -0,0 +1,89 @@ +import inspect +from collections.abc import Callable, Coroutine +from typing import Any, Concatenate, ParamSpec, TypeVar, cast, get_type_hints, overload + +from ._core import FluxQueueCore +from ._task import _task_decorator + +P = ParamSpec("P") + + +class Context: + __fluxqueue_context__: str | None = None + + def __init_subclass__(cls) -> None: + if not cls.__fluxqueue_context__: + cls.__fluxqueue_context__ = cls.__name__ + # raise NotImplementedError( + # f"{cls.__name__} is not implemented properly, make sure to add '__fluxqueue_context__' attribute." + # ) + + +C = TypeVar("C", bound=Context) + + +@overload +def _with_context( + func: Callable[Concatenate[C, P], None], + *, + name: str | None, + queue: str, + max_retries: int, + core: FluxQueueCore, +) -> Callable[P, None]: ... + + +@overload +def _with_context( + func: Callable[Concatenate[C, P], Coroutine[Any, Any, None]], + *, + name: str | None, + queue: str, + max_retries: int, + core: FluxQueueCore, +) -> Callable[P, Coroutine[Any, Any, None]]: ... + + +def _with_context( + func: Callable[Concatenate[C, P], None | Coroutine[Any, Any, None]], + *, + name: str | None, + queue: str, + max_retries: int, + core: FluxQueueCore, +) -> Callable[P, None | Coroutine[Any, Any, None]]: + sig = inspect.signature(func) + hints = get_type_hints(func) + + all_param_names = list(sig.parameters.keys()) + + context_params = { + name: hints[name] + for name in all_param_names + if name in hints + and isinstance(hints[name], type) + and issubclass(hints[name], Context) + } + + if len(context_params) != 1: + raise TypeError( + f"Expected exactly one context parameter, found {len(context_params)}: {list(context_params.keys())}" + ) + + non_context_params = [ + name for name in all_param_names if name not in context_params + ] + + new_sig = sig.replace(parameters=[sig.parameters[n] for n in non_context_params]) + + if inspect.iscoroutinefunction(func): + wrapper = _task_decorator( + cast(Any, func), name=name, queue=queue, max_retries=max_retries, core=core + ) + else: + wrapper = _task_decorator( + cast(Any, func), name=name, queue=queue, max_retries=max_retries, core=core + ) + + cast(Any, wrapper).__signature__ = new_sig + return wrapper From 856eb99b88dc250959641257f35be3fb0a64c534 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sat, 21 Feb 2026 15:59:43 +0400 Subject: [PATCH 02/31] Fix clippy error by moving args to a struct --- crates/fluxqueue-worker/src/worker.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/crates/fluxqueue-worker/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index 8e2ba8c..3351be1 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/crates/fluxqueue-worker/src/worker.rs @@ -14,6 +14,19 @@ use crate::redis_client::RedisClient; use crate::task::{PythonDispatcher, TaskRegistry}; use fluxqueue_common::{Task, deserialize_raw_task_data}; +struct ExecutorContext { + queue_name: Arc, + executor_id: Arc, + redis_client: Arc, + task_registry: Arc, + python_dispatcher: Arc, +} + +struct ReadyCheck { + concurrency: Arc, + counter: Arc, +} + pub async fn run_worker( mut shutdown: watch::Receiver, concurrency: usize, From fed2e2a94c9d6e2684d227bda6fb13fc9df3b026 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sat, 21 Feb 2026 16:01:01 +0400 Subject: [PATCH 03/31] Move structs below the run_worker func --- crates/fluxqueue-worker/src/worker.rs | 13 ------------- 1 file changed, 13 deletions(-) diff --git a/crates/fluxqueue-worker/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index 3351be1..8e2ba8c 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/crates/fluxqueue-worker/src/worker.rs @@ -14,19 +14,6 @@ use crate::redis_client::RedisClient; use crate::task::{PythonDispatcher, TaskRegistry}; use fluxqueue_common::{Task, deserialize_raw_task_data}; -struct ExecutorContext { - queue_name: Arc, - executor_id: Arc, - redis_client: Arc, - task_registry: Arc, - python_dispatcher: Arc, -} - -struct ReadyCheck { - concurrency: Arc, - counter: Arc, -} - pub async fn run_worker( mut shutdown: watch::Receiver, concurrency: usize, From eba4d34b5c8216962468384efa6590f964491b52 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sun, 22 Feb 2026 21:09:34 +0400 Subject: [PATCH 04/31] Start working on context in the worker --- .../fluxqueue-worker/scripts/get_functions.py | 27 -- .../fluxqueue-worker/scripts/get_registry.py | 37 +++ crates/fluxqueue-worker/src/logger.rs | 4 +- crates/fluxqueue-worker/src/redis_client.rs | 4 +- crates/fluxqueue-worker/src/task.rs | 212 +++++++++++- crates/fluxqueue-worker/src/worker.rs | 307 ++++++------------ python/fluxqueue/_task.py | 1 + 7 files changed, 344 insertions(+), 248 deletions(-) delete mode 100644 crates/fluxqueue-worker/scripts/get_functions.py create mode 100644 crates/fluxqueue-worker/scripts/get_registry.py diff --git a/crates/fluxqueue-worker/scripts/get_functions.py b/crates/fluxqueue-worker/scripts/get_functions.py deleted file mode 100644 index 05b5527..0000000 --- a/crates/fluxqueue-worker/scripts/get_functions.py +++ /dev/null @@ -1,27 +0,0 @@ -import importlib -import inspect -import sys -from pathlib import Path - - -def list_functions(module_path: str, queue: str, module_dir: str | None = None): - if module_dir: - module_dir_path = Path(module_dir).resolve() - if str(module_dir_path) not in sys.path: - sys.path.insert(0, str(module_dir_path)) - - module = importlib.import_module(module_path) - funcs = {} - for _name, obj in inspect.getmembers(module): - task_name = getattr(obj, "task_name", None) - task_queue = getattr(obj, "queue", None) - if not task_queue or task_queue != queue: - continue - - if inspect.isfunction(obj) or (inspect.isbuiltin(obj) and task_name): - if funcs.get(task_name): - raise ValueError(f"Task name '{task_name}' is duplicated") - - original_func = getattr(obj, "__wrapped__", obj) - funcs[task_name] = original_func - return funcs diff --git a/crates/fluxqueue-worker/scripts/get_registry.py b/crates/fluxqueue-worker/scripts/get_registry.py new file mode 100644 index 0000000..2857e36 --- /dev/null +++ b/crates/fluxqueue-worker/scripts/get_registry.py @@ -0,0 +1,37 @@ +import importlib +import inspect +import sys +from pathlib import Path + + +def get_registry(module_path: str, queue: str, module_dir: str | None = None): + if module_dir: + module_dir_path = Path(module_dir).resolve() + if str(module_dir_path) not in sys.path: + sys.path.insert(0, str(module_dir_path)) + + module = importlib.import_module(module_path) + registry = {"tasks": {}, "contexts": {}} + for _name, obj in inspect.getmembers(module): + if inspect.isfunction(obj): + task_name = getattr(obj, "task_name", None) + task_queue = getattr(obj, "queue", None) + if not task_queue or task_queue != queue: + continue + + if registry["tasks"].get(task_name): + raise ValueError(f"Task '{task_name}' is duplicated") + + original_func = getattr(obj, "__wrapped__", obj) + registry["tasks"][task_name] = original_func + elif inspect.isclass(obj): + context_name = getattr(obj, "__fluxqueue_context__", None) + if not context_name: + continue + + if registry["contexts"].get(context_name): + raise ValueError(f"Context '{context_name}' is duplicated") + + registry["contexts"][context_name] = obj + + return registry diff --git a/crates/fluxqueue-worker/src/logger.rs b/crates/fluxqueue-worker/src/logger.rs index fff49c4..df72f76 100644 --- a/crates/fluxqueue-worker/src/logger.rs +++ b/crates/fluxqueue-worker/src/logger.rs @@ -34,13 +34,15 @@ pub fn initial_logs( concurrency: usize, redis_url: &str, tasks_module_path: &str, - tasks: &Vec<&String>, + tasks: Vec, + contexts: Vec, ) { info!("Queue: {}", queue_name); info!("Concurrency: {}", concurrency); info!("Redis: {}", redis_url); info!("Tasks module: {}", tasks_module_path); info!("Tasks found: {:?}", tasks); + info!("Contexts found: {:?}", contexts); info!("Starting up the executors..."); } diff --git a/crates/fluxqueue-worker/src/redis_client.rs b/crates/fluxqueue-worker/src/redis_client.rs index 78e1466..27fde37 100644 --- a/crates/fluxqueue-worker/src/redis_client.rs +++ b/crates/fluxqueue-worker/src/redis_client.rs @@ -33,7 +33,7 @@ impl RedisClient { Ok(()) } - pub async fn set_executors_heartbeat(&self, executor_ids: Arc>>) -> Result<()> { + pub async fn set_executors_heartbeat(&self, executor_ids: Arc>>) -> Result<()> { for id in executor_ids.iter() { self.set_executor_heartbeat(id).await?; } @@ -60,7 +60,7 @@ impl RedisClient { pub async fn cleanup_executors_registry( &self, queue_name: &str, - ids: Arc>>, + ids: Arc>>, ) -> Result<()> { let mut conn = self.redis_pool.get().await?; let executors_key = keys::get_executors_key(queue_name); diff --git a/crates/fluxqueue-worker/src/task.rs b/crates/fluxqueue-worker/src/task.rs index bd2ae9c..8aeea5d 100644 --- a/crates/fluxqueue-worker/src/task.rs +++ b/crates/fluxqueue-worker/src/task.rs @@ -1,32 +1,120 @@ use anyhow::{Context, Result, anyhow}; use pyo3::exceptions::PyRuntimeError; use pyo3::prelude::*; -use pyo3::types::{PyDict, PyList, PyTuple}; +use pyo3::types::{PyAnyMethods, PyDict, PyDictMethods, PyList, PyModule, PyTuple}; use pyo3_async_runtimes::tokio::into_future; use pythonize::pythonize; use rmp_serde::from_slice; use rmpv::Value; use std::collections::HashMap; +use std::ffi::CString; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, RwLock}; +use std::time::Instant; use tokio::sync::{mpsc, oneshot}; +use crate::logger::Logger; + +#[derive(Debug)] pub struct TaskRegistry { tasks: Arc>>>>, + contexts: Arc>>>>, } impl TaskRegistry { - pub fn new() -> Self { - Self { - tasks: Arc::new(RwLock::new(HashMap::new())), + pub fn new(module_path: &str, queue_name: &str) -> Result { + let script = include_str!("../scripts/get_registry.py"); + let script_cstr = CString::new(script)?; + let filename = CString::new("get_registry.py")?; + let module_name = CString::new("get_registry")?; + + let full_current_dir = std::env::current_dir().unwrap(); + let full_module_path = full_current_dir.join(module_path); + let clean_module_path = normalize_path(&full_module_path); + let project_root = full_current_dir + .ancestors() + .find(|p| p.join("tests").exists()) + .unwrap_or(&full_current_dir); + let real_module_path = path_to_module_path(project_root, &clean_module_path); + + if !clean_module_path.exists() || real_module_path.is_none() { + return Err(anyhow!( + "Tasks module path {:?} doesn't exist.", + clean_module_path + )); } + + let real_module_path = real_module_path.unwrap(); + let module_dir = project_root.to_string_lossy().to_string(); + + let (tasks, contexts) = Python::attach( + |py| -> Result<( + HashMap>>, + HashMap>>, + )> { + let module = PyModule::from_code( + py, + script_cstr.as_c_str(), + filename.as_c_str(), + module_name.as_c_str(), + ) + .map_err(|e| anyhow!("Failed to import python module: {}", e))?; + + let registry: Bound<'_, PyDict> = module + .getattr("get_registry") + .map_err(|e| anyhow!("Failed to get 'get_registry' script: {}", e))? + .call1((real_module_path, queue_name, module_dir)) + .map_err(|e| anyhow!("Failed to get tasks: {}", e))? + .cast_into::() + .map_err(|_| anyhow!("Failed to cast result to a Python Dictionary"))?; + + let tasks: HashMap>> = registry + .get_item("tasks")? + .expect("tasks missing") + .cast::() + .map_err(|e| anyhow!("tasks is not a dict: {}", e))? + .iter() + .filter_map(|(key, value)| { + let name: String = key.extract().ok()?; + let func: Py = value.unbind(); + Some((name, Arc::new(func))) + }) + .collect(); + + let contexts: HashMap>> = registry + .get_item("contexts")? + .expect("contexts missing") + .cast::() + .map_err(|e| anyhow!("contexts is not a dict: {}", e))? + .iter() + .filter_map(|(key, value): (Bound, Bound)| { + let name: String = key.extract().ok()?; + let func: Py = value.unbind(); + Some((name, Arc::new(func))) + }) + .collect(); + + Ok((tasks, contexts)) + }, + )?; + + Ok(Self { + tasks: Arc::new(RwLock::new(tasks)), + contexts: Arc::new(RwLock::new(contexts)), + }) } - pub fn insert(&self, name: String, func: Py) -> Result<()> { - let mut tasks = self.tasks.write().map_err(|_| { - anyhow!("Internal Error: Task registry lock poisoned (a thread panicked)") - })?; - tasks.insert(name, Arc::new(func)); - Ok(()) + pub fn get_registered_tasks(&self) -> Result> { + let tasks = self.tasks.read().map_err(|e| anyhow!(e.to_string()))?; + let task_names: Vec<_> = tasks.iter().map(|t| t.0.to_string()).collect(); + Ok(task_names) + } + + pub fn get_registered_contexts(&self) -> Result> { + let contexts = self.contexts.read().map_err(|e| anyhow!(e.to_string()))?; + let context_names: Vec<_> = contexts.iter().map(|t| t.0.to_string()).collect(); + Ok(context_names) } pub fn get(&self, name: &str) -> Option>> { @@ -36,6 +124,7 @@ impl TaskRegistry { } struct TaskRequest { + executor_id: Arc, func: Arc>, task_name: Arc, raw_args: Arc>, @@ -54,9 +143,15 @@ impl PythonDispatcher { let dispatcher = async move { while let Some(req) = rx.recv().await { - run_task(req.func, req.task_name, req.raw_args, req.raw_kwargs) - .await - .map_err(|e| PyRuntimeError::new_err(e.to_string()))?; + run_task( + req.executor_id, + req.func, + req.task_name, + req.raw_args, + req.raw_kwargs, + ) + .await + .map_err(|e| PyRuntimeError::new_err(e.to_string()))?; let _ = req.resp_tx.send(Ok(())); } @@ -74,6 +169,7 @@ impl PythonDispatcher { pub async fn execute( &self, + executor_id: Arc, func: Arc>, task_name: Arc, raw_args: Arc>, @@ -83,6 +179,7 @@ impl PythonDispatcher { self.tx .send(TaskRequest { + executor_id, func, task_name, raw_args, @@ -98,11 +195,15 @@ impl PythonDispatcher { } async fn run_task( + executor_id: Arc, task_function: Arc>, task_name: Arc, raw_args: Arc>, raw_kwargs: Arc>, ) -> Result<()> { + let logger = Logger::new(format!("EXECUTOR {}", &executor_id)); + let duration_start = Instant::now(); + let task_args: Value = from_slice(&raw_args).context(format!( "Failed to deserialize task '{}' function args", &task_name @@ -140,14 +241,99 @@ async fn run_task( if is_coroutine { Ok(Some(result)) } else { + let duration_end = duration_start.elapsed(); + logger.info(format_args!( + "Task '{}' successfully finished in {}ms", + &task_name, + duration_end.as_millis() + )); + Ok(None) } + }) + .map_err(|e| { + let duration_end = duration_start.elapsed(); + logger.error(format_args!( + "Task '{}' failed in {}ms: {}", + &task_name, + duration_end.as_millis(), + e + )); + anyhow!(e.to_string()) })?; if let Some(coro) = maybe_coro { let fut = Python::attach(|py| into_future(coro.into_bound(py)))?; fut.await?; + + let duration_end = duration_start.elapsed(); + logger.info(format_args!( + "Task '{}' successfully finished in {}ms", + &task_name, + duration_end.as_millis() + )); } Ok(()) } + +pub struct DispatcherPool { + dispatchers: Vec>, + index: AtomicUsize, +} + +impl DispatcherPool { + pub fn new(concurrency: usize) -> Result { + // let logical_cores = num_cpus::get(); + // let pool_size = (concurrency / 4) + // .max(1) + // .min(logical_cores * 2) + // .max(logical_cores); + + let mut dispatchers = Vec::with_capacity(concurrency); + for _ in 0..concurrency { + dispatchers.push(Arc::new(PythonDispatcher::new()?)); + } + + Ok(Self { + dispatchers, + index: AtomicUsize::new(0), + }) + } + + pub fn get(&self) -> Arc { + let idx = self.index.fetch_add(1, Ordering::Relaxed) % self.dispatchers.len(); + self.dispatchers[idx].clone() + } +} + +fn normalize_path(path: &Path) -> PathBuf { + let mut components = Vec::new(); + for comp in path.components() { + match comp { + std::path::Component::ParentDir => { + components.pop(); + } + std::path::Component::CurDir => {} + other => components.push(other), + } + } + components.iter().collect() +} + +fn path_to_module_path(current_dir: &Path, target_path: &Path) -> Option { + let rel_path = target_path.strip_prefix(current_dir).ok()?; + + let mut components: Vec = rel_path + .components() + .map(|c| c.as_os_str().to_string_lossy().to_string()) + .collect(); + + if let Some(last) = components.last_mut() + && let Some(pos) = last.rfind('.') + { + last.truncate(pos); + } + + Some(components.join(".")) +} diff --git a/crates/fluxqueue-worker/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index 8e2ba8c..ae0a3a1 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/crates/fluxqueue-worker/src/worker.rs @@ -1,17 +1,14 @@ -use anyhow::{Result, anyhow}; -use pyo3::types::{PyAnyMethods, PyDict, PyDictMethods, PyModule}; -use pyo3::{Bound, Py, PyAny, Python}; -use std::ffi::CString; -use std::path::{Path, PathBuf}; +use anyhow::Result; +use pyo3::{Py, PyAny}; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::time::{Duration, Instant}; +use std::time::Duration; use tokio::sync::watch; use tokio::task::JoinSet; use crate::logger::{Logger, initial_logs}; use crate::redis_client::RedisClient; -use crate::task::{PythonDispatcher, TaskRegistry}; +use crate::task::{DispatcherPool, TaskRegistry}; use fluxqueue_common::{Task, deserialize_raw_task_data}; pub async fn run_worker( @@ -28,26 +25,21 @@ pub async fn run_worker( })?; let redis_client = Arc::new(redis_client); - let task_functions = get_task_functions(&tasks_module_path, &queue_name).map_err(|e| { - tracing::error!("{}", e); - std::process::exit(1); - })?; - let task_names: Vec<&String> = task_functions.iter().map(|(name, _obj)| name).collect(); + let task_registry = Arc::new(TaskRegistry::new(&tasks_module_path, &queue_name)?); + let registered_tasks = task_registry.get_registered_tasks()?; + let registered_contexts = task_registry.get_registered_contexts()?; initial_logs( &queue_name, concurrency, &redis_url, &tasks_module_path, - &task_names, + registered_tasks, + registered_contexts, ); - let task_registry = Arc::new(TaskRegistry::new()); - for (name, task_obj) in task_functions { - task_registry.insert(name, task_obj)?; - } - - let queue_name = Arc::from(queue_name.to_string()); + let dispatcher_pool = Arc::new(DispatcherPool::new(concurrency)?); + let queue_name = Arc::new(queue_name); let executor_ids = generate_executor_ids(concurrency); let atomic_concurrency = Arc::new(AtomicUsize::new(concurrency)); let started_executors_count = Arc::new(AtomicUsize::new(0)); @@ -62,10 +54,9 @@ pub async fn run_worker( let queue_name = Arc::clone(&queue_name); let executor_id = Arc::clone(&executor_ids[i]); let task_registry = Arc::clone(&task_registry); + let dispatcher_pool = Arc::clone(&dispatcher_pool); async move { - let python_dispatcher = Arc::new(PythonDispatcher::new()?); - redis_client .register_executor(&queue_name, &executor_id) .await?; @@ -82,7 +73,7 @@ pub async fn run_worker( executor_id, redis_client, task_registry, - python_dispatcher, + dispatcher_pool, }; Ok::<_, anyhow::Error>((shutdown, ready_check, executor_context)) @@ -90,9 +81,9 @@ pub async fn run_worker( }) .collect(); - let results = futures::future::join_all(executor_futures).await; + let results = futures::future::try_join_all(executor_futures).await?; for result in results { - let (shutdown, ready_check, executor_context) = result?; + let (shutdown, ready_check, executor_context) = result; executors.spawn(executor_loop(shutdown, ready_check, executor_context)); } @@ -136,11 +127,11 @@ pub async fn run_worker( } struct ExecutorContext { - queue_name: Arc, - executor_id: Arc, + queue_name: Arc, + executor_id: Arc, redis_client: Arc, task_registry: Arc, - python_dispatcher: Arc, + dispatcher_pool: Arc, } struct ReadyCheck { @@ -188,8 +179,7 @@ async fn executor_loop( return Ok(()); }; - let duration_start = Instant::now(); - let task_result = run_task(ctx.python_dispatcher.clone(), &task, task_function).await; + let task_result = run_task(ctx.executor_id.clone(), ctx.dispatcher_pool.clone(), &task, task_function).await; match task_result { Ok(_) => { @@ -198,21 +188,8 @@ async fn executor_loop( .await { logger.error(format_args!("Failed to remove the task after successful run: {}", e)); } - let duration_end = duration_start.elapsed(); - logger.info(format_args!( - "Task '{}' successfully finished in {}ms", - &task_name, - duration_end.as_millis() - )); } - Err(e) => { - let duration_end = duration_start.elapsed(); - logger.error(format_args!( - "Task '{}' failed in {}ms: {}", - &task_name, - duration_end.as_millis(), - e - )); + Err(_) => { if let Err(err) = ctx.redis_client .mark_as_failed(&ctx.queue_name, &ctx.executor_id, &raw_data) .await { @@ -235,8 +212,8 @@ async fn executor_loop( async fn janitor_loop( mut shutdown: watch::Receiver, ready_check: ReadyCheck, - queue_name: Arc, - executor_ids: Arc>>, + queue_name: Arc, + executor_ids: Arc>>, save_dead_tasks: Arc, redis_client: Arc, ) -> Result<()> { @@ -248,6 +225,7 @@ async fn janitor_loop( loop { tokio::select! { _ = shutdown.changed() => { + // TODO: Add a log on shutdown to notify how many tasks are left in the queue if any. return Ok(()) } @@ -257,10 +235,11 @@ async fn janitor_loop( match tasks_res { Ok(Some(raw_data)) => { let task = deserialize_raw_task_data(&raw_data)?; + let task_name = format!("{}:{}", &task.name, &task.id); logger.info(format_args!( "Received a failed task '{}': retries={}, max retries={}", - &task.name, + &task_name, &task.retries, &task.max_retries )); @@ -311,112 +290,28 @@ async fn janitor_loop( } async fn run_task( - python_dispatcher: Arc, + executor_id: Arc, + dispatcher_pool: Arc, task: &Task, task_function: Arc>, ) -> Result<()> { - let task_name = Arc::new(task.name.clone()); + let task_name = Arc::new(format!("{}:{}", &task.name, &task.id)); let raw_args = Arc::new(task.args.clone()); let raw_kwargs = Arc::new(task.kwargs.clone()); - python_dispatcher - .execute(task_function, task_name, raw_args, raw_kwargs) + let dispatcher = dispatcher_pool.get(); + dispatcher + .execute(executor_id, task_function, task_name, raw_args, raw_kwargs) .await?; Ok(()) } -fn get_task_functions(module_path: &str, queue_name: &str) -> Result)>> { - let script = include_str!("../scripts/get_functions.py"); - let script_cstr = CString::new(script)?; - let filename = CString::new("get_functions.py")?; - let module_name = CString::new("get_functions")?; - - let full_current_dir = std::env::current_dir().unwrap(); - let full_module_path = full_current_dir.join(module_path); - let clean_module_path = normalize_path(&full_module_path); - let project_root = full_current_dir - .ancestors() - .find(|p| p.join("tests").exists()) - .unwrap_or(&full_current_dir); - let real_module_path = path_to_module_path(project_root, &clean_module_path); - - if !clean_module_path.exists() || real_module_path.is_none() { - return Err(anyhow!( - "Tasks module path {:?} doesn't exist.", - clean_module_path - )); - } - - let real_module_path = real_module_path.unwrap(); - let module_dir = project_root.to_string_lossy().to_string(); - - Python::attach(|py| { - let module = PyModule::from_code( - py, - script_cstr.as_c_str(), - filename.as_c_str(), - module_name.as_c_str(), - ) - .map_err(|e| anyhow!("Failed to import python module: {}", e))?; - - let py_funcs: Bound<'_, PyDict> = module - .getattr("list_functions") - .map_err(|e| anyhow!("Failed to get 'list_functions' script: {}", e))? - .call1((real_module_path, queue_name, module_dir)) - .map_err(|e| anyhow!("Failed to get tasks: {}", e))? - .cast_into::() - .map_err(|_| anyhow!("Failed to cast result to a Python Dictionary"))?; - - let funcs: Vec<_> = py_funcs - .iter() - .filter_map(|(key, value): (Bound, Bound)| { - let name: String = key.extract().ok()?; - let func: Py = value.unbind(); - Some((name, func)) - }) - .collect(); - - Ok(funcs) - }) -} - -fn normalize_path(path: &Path) -> PathBuf { - let mut components = Vec::new(); - for comp in path.components() { - match comp { - std::path::Component::ParentDir => { - components.pop(); - } - std::path::Component::CurDir => {} - other => components.push(other), - } - } - components.iter().collect() -} - -fn path_to_module_path(current_dir: &Path, target_path: &Path) -> Option { - let rel_path = target_path.strip_prefix(current_dir).ok()?; - - let mut components: Vec = rel_path - .components() - .map(|c| c.as_os_str().to_string_lossy().to_string()) - .collect(); - - if let Some(last) = components.last_mut() - && let Some(pos) = last.rfind('.') - { - last.truncate(pos); - } - - Some(components.join(".")) -} - -fn generate_executor_ids(num_executors: usize) -> Arc>> { +fn generate_executor_ids(num_executors: usize) -> Arc>> { let mut ids = Vec::with_capacity(num_executors); for _ in 0..num_executors { - let id: Arc = Arc::from(uuid::Uuid::new_v4().to_string()); + let id: Arc = Arc::from(uuid::Uuid::new_v4().to_string()); ids.push(id); } @@ -440,19 +335,19 @@ mod tests { use crate::logger::TestWriter; use std::sync::{Arc, Mutex}; - #[test] - fn test_path_to_module_path() -> Result<()> { - let current_dir = Path::new("project"); - let tasks_path = Path::new("../project/tasks.py"); - let normalized_path = normalize_path(tasks_path); - let module_path = path_to_module_path(current_dir, &normalized_path); - let expected_path = Path::new("project/tasks.py"); + // #[test] + // fn test_path_to_module_path() -> Result<()> { + // let current_dir = Path::new("project"); + // let tasks_path = Path::new("../project/tasks.py"); + // let normalized_path = normalize_path(tasks_path); + // let module_path = path_to_module_path(current_dir, &normalized_path); + // let expected_path = Path::new("project/tasks.py"); - assert_eq!(normalized_path, expected_path); - assert_eq!(module_path, Some("tasks".to_string())); + // assert_eq!(normalized_path, expected_path); + // assert_eq!(module_path, Some("tasks".to_string())); - Ok(()) - } + // Ok(()) + // } fn get_test_module_path(filename: &str) -> String { let current_dir = std::env::current_dir().unwrap(); @@ -460,71 +355,66 @@ mod tests { test_module_path.to_str().unwrap().to_string() } - #[test] - fn test_get_task_functions_valid_module() -> Result<()> { - let module_path_str = get_test_module_path("test_tasks_module.py"); - let functions = get_task_functions(&module_path_str, "default")?; + // #[test] + // fn test_get_task_functions_valid_module() -> Result<()> { + // let module_path_str = get_test_module_path("test_tasks_module.py"); + // let functions = get_registry(&module_path_str, "default")?; - assert_eq!(functions.len(), 3); + // assert_eq!(functions.len(), 3); - let task_names: Vec = functions.iter().map(|(name, _)| name.clone()).collect(); - assert!(task_names.contains(&"task-1".to_string())); - assert!(task_names.contains(&"task-2".to_string())); - assert!(task_names.contains(&"async-task".to_string())); + // let task_names: Vec = functions.iter().map(|(name, _)| name.clone()).collect(); + // assert!(task_names.contains(&"task-1".to_string())); + // assert!(task_names.contains(&"task-2".to_string())); + // assert!(task_names.contains(&"async-task".to_string())); - assert!(!task_names.contains(&"high-priority-task".to_string())); + // assert!(!task_names.contains(&"high-priority-task".to_string())); - Ok(()) - } + // Ok(()) + // } - #[test] - fn test_get_task_functions_different_queue() -> Result<()> { - let module_path_str = get_test_module_path("test_tasks_module.py"); - let functions = get_task_functions(&module_path_str, "high-priority")?; + // #[test] + // fn test_get_task_functions_different_queue() -> Result<()> { + // let module_path_str = get_test_module_path("test_tasks_module.py"); + // let functions = get_registry(&module_path_str, "high-priority")?; - assert_eq!(functions.len(), 1); - assert_eq!(functions[0].0, "high-priority-task"); + // assert_eq!(functions.len(), 1); + // assert_eq!(functions[0].0, "high-priority-task"); - Ok(()) - } + // Ok(()) + // } - #[test] - fn test_get_task_functions_empty_module() -> Result<()> { - let module_path_str = get_test_module_path("test_tasks_empty.py"); - let functions = get_task_functions(&module_path_str, "default")?; + // #[test] + // fn test_get_task_functions_empty_module() -> Result<()> { + // let module_path_str = get_test_module_path("test_tasks_empty.py"); + // let functions = get_registry(&module_path_str, "default")?; - assert_eq!(functions.len(), 0); + // assert_eq!(functions.len(), 0); - Ok(()) - } + // Ok(()) + // } - #[test] - fn test_get_task_functions_duplicate_names() { - let module_path_str = get_test_module_path("test_tasks_duplicate.py"); + // #[test] + // fn test_get_task_functions_duplicate_names() { + // let module_path_str = get_test_module_path("test_tasks_duplicate.py"); - let result = get_task_functions(&module_path_str, "default"); - assert!(result.is_err()); + // let result = get_registry(&module_path_str, "default"); + // assert!(result.is_err()); - let error_msg = result.unwrap_err().to_string(); - assert!(error_msg.contains("duplicated") || error_msg.contains("duplicate")); - } + // let error_msg = result.unwrap_err().to_string(); + // assert!(error_msg.contains("duplicated") || error_msg.contains("duplicate")); + // } - #[test] - fn test_get_task_functions_invalid_path() { - let result = get_task_functions("nonexistent/path/to/module.py", "default"); - assert!(result.is_err()); - } + // #[test] + // fn test_get_task_functions_invalid_path() { + // let result = get_registry("nonexistent/path/to/module.py", "default"); + // assert!(result.is_err()); + // } #[tokio::test] async fn test_run_task_with_sync_function() -> Result<()> { - let task_registry = TaskRegistry::new(); - let python_dispatcher = Arc::new(PythonDispatcher::new()?); + let dispatcher_pool = Arc::new(DispatcherPool::new(1)?); let module_path_str = get_test_module_path("test_tasks_module.py"); - let task_functions = get_task_functions(&module_path_str, "default")?; - - for (name, task_obj) in task_functions { - task_registry.insert(name, task_obj)?; - } + let task_registry = TaskRegistry::new(&module_path_str, "default")?; let task = task_registry.get("task-1"); assert!(task.is_some()); @@ -540,7 +430,13 @@ mod tests { max_retries: 3, }; - let result = run_task(python_dispatcher.clone(), &task, task_func).await; + let result = run_task( + Arc::new("test".to_string()), + dispatcher_pool.clone(), + &task, + task_func, + ) + .await; assert!(!result.is_err()); } @@ -549,14 +445,9 @@ mod tests { #[tokio::test] async fn test_run_task_with_async_function() -> Result<()> { - let task_registry = TaskRegistry::new(); - let python_dispatcher = Arc::new(PythonDispatcher::new()?); + let dispatcher_pool = Arc::new(DispatcherPool::new(1)?); let module_path_str = get_test_module_path("test_tasks_module.py"); - let task_functions = get_task_functions(&module_path_str, "default")?; - - for (name, task_obj) in task_functions { - task_registry.insert(name, task_obj)?; - } + let task_registry = TaskRegistry::new(&module_path_str, "default")?; let task = task_registry.get("async-task"); assert!(task.is_some()); @@ -572,7 +463,13 @@ mod tests { max_retries: 3, }; - let result = run_task(python_dispatcher.clone(), &task, task_func).await; + let result = run_task( + Arc::new("test".to_string()), + dispatcher_pool.clone(), + &task, + task_func, + ) + .await; assert!(!result.is_err()); } diff --git a/python/fluxqueue/_task.py b/python/fluxqueue/_task.py index eea25fb..207f857 100644 --- a/python/fluxqueue/_task.py +++ b/python/fluxqueue/_task.py @@ -48,6 +48,7 @@ def _task_decorator( is_async = inspect.iscoroutinefunction(func) task_name = get_task_name(func, name) + # TODO: Add unique identifier 'fluxqueue' just to be 100% sure cast(Any, func).task_name = task_name cast(Any, func).queue = queue From e65bef3dce1eabeaa38b9974197eac90451be355 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sun, 22 Feb 2026 21:13:27 +0400 Subject: [PATCH 05/31] Fix clippy error --- crates/fluxqueue-worker/src/task.rs | 100 ++++++++++++++-------------- 1 file changed, 50 insertions(+), 50 deletions(-) diff --git a/crates/fluxqueue-worker/src/task.rs b/crates/fluxqueue-worker/src/task.rs index 8aeea5d..0262495 100644 --- a/crates/fluxqueue-worker/src/task.rs +++ b/crates/fluxqueue-worker/src/task.rs @@ -16,6 +16,11 @@ use tokio::sync::{mpsc, oneshot}; use crate::logger::Logger; +type TasksAndContexts = ( + HashMap>>, + HashMap>>, +); + #[derive(Debug)] pub struct TaskRegistry { tasks: Arc>>>>, @@ -48,56 +53,51 @@ impl TaskRegistry { let real_module_path = real_module_path.unwrap(); let module_dir = project_root.to_string_lossy().to_string(); - let (tasks, contexts) = Python::attach( - |py| -> Result<( - HashMap>>, - HashMap>>, - )> { - let module = PyModule::from_code( - py, - script_cstr.as_c_str(), - filename.as_c_str(), - module_name.as_c_str(), - ) - .map_err(|e| anyhow!("Failed to import python module: {}", e))?; - - let registry: Bound<'_, PyDict> = module - .getattr("get_registry") - .map_err(|e| anyhow!("Failed to get 'get_registry' script: {}", e))? - .call1((real_module_path, queue_name, module_dir)) - .map_err(|e| anyhow!("Failed to get tasks: {}", e))? - .cast_into::() - .map_err(|_| anyhow!("Failed to cast result to a Python Dictionary"))?; - - let tasks: HashMap>> = registry - .get_item("tasks")? - .expect("tasks missing") - .cast::() - .map_err(|e| anyhow!("tasks is not a dict: {}", e))? - .iter() - .filter_map(|(key, value)| { - let name: String = key.extract().ok()?; - let func: Py = value.unbind(); - Some((name, Arc::new(func))) - }) - .collect(); - - let contexts: HashMap>> = registry - .get_item("contexts")? - .expect("contexts missing") - .cast::() - .map_err(|e| anyhow!("contexts is not a dict: {}", e))? - .iter() - .filter_map(|(key, value): (Bound, Bound)| { - let name: String = key.extract().ok()?; - let func: Py = value.unbind(); - Some((name, Arc::new(func))) - }) - .collect(); - - Ok((tasks, contexts)) - }, - )?; + let (tasks, contexts) = Python::attach(|py| -> Result { + let module = PyModule::from_code( + py, + script_cstr.as_c_str(), + filename.as_c_str(), + module_name.as_c_str(), + ) + .map_err(|e| anyhow!("Failed to import python module: {}", e))?; + + let registry: Bound<'_, PyDict> = module + .getattr("get_registry") + .map_err(|e| anyhow!("Failed to get 'get_registry' script: {}", e))? + .call1((real_module_path, queue_name, module_dir)) + .map_err(|e| anyhow!("Failed to get tasks: {}", e))? + .cast_into::() + .map_err(|_| anyhow!("Failed to cast result to a Python Dictionary"))?; + + let tasks: HashMap>> = registry + .get_item("tasks")? + .expect("tasks missing") + .cast::() + .map_err(|e| anyhow!("tasks is not a dict: {}", e))? + .iter() + .filter_map(|(key, value)| { + let name: String = key.extract().ok()?; + let func: Py = value.unbind(); + Some((name, Arc::new(func))) + }) + .collect(); + + let contexts: HashMap>> = registry + .get_item("contexts")? + .expect("contexts missing") + .cast::() + .map_err(|e| anyhow!("contexts is not a dict: {}", e))? + .iter() + .filter_map(|(key, value): (Bound, Bound)| { + let name: String = key.extract().ok()?; + let func: Py = value.unbind(); + Some((name, Arc::new(func))) + }) + .collect(); + + Ok((tasks, contexts)) + })?; Ok(Self { tasks: Arc::new(RwLock::new(tasks)), From 154a1ef5c43e7529c07c3ec01210ccc0a4f517af Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sun, 22 Feb 2026 22:15:38 +0400 Subject: [PATCH 06/31] Remove dispatcher pool, was no different without it --- crates/fluxqueue-worker/src/task.rs | 31 --------------------------- crates/fluxqueue-worker/src/worker.rs | 21 +++++++++--------- 2 files changed, 10 insertions(+), 42 deletions(-) diff --git a/crates/fluxqueue-worker/src/task.rs b/crates/fluxqueue-worker/src/task.rs index 0262495..4f222b2 100644 --- a/crates/fluxqueue-worker/src/task.rs +++ b/crates/fluxqueue-worker/src/task.rs @@ -9,7 +9,6 @@ use rmpv::Value; use std::collections::HashMap; use std::ffi::CString; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, RwLock}; use std::time::Instant; use tokio::sync::{mpsc, oneshot}; @@ -277,36 +276,6 @@ async fn run_task( Ok(()) } -pub struct DispatcherPool { - dispatchers: Vec>, - index: AtomicUsize, -} - -impl DispatcherPool { - pub fn new(concurrency: usize) -> Result { - // let logical_cores = num_cpus::get(); - // let pool_size = (concurrency / 4) - // .max(1) - // .min(logical_cores * 2) - // .max(logical_cores); - - let mut dispatchers = Vec::with_capacity(concurrency); - for _ in 0..concurrency { - dispatchers.push(Arc::new(PythonDispatcher::new()?)); - } - - Ok(Self { - dispatchers, - index: AtomicUsize::new(0), - }) - } - - pub fn get(&self) -> Arc { - let idx = self.index.fetch_add(1, Ordering::Relaxed) % self.dispatchers.len(); - self.dispatchers[idx].clone() - } -} - fn normalize_path(path: &Path) -> PathBuf { let mut components = Vec::new(); for comp in path.components() { diff --git a/crates/fluxqueue-worker/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index ae0a3a1..328b0a4 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/crates/fluxqueue-worker/src/worker.rs @@ -8,7 +8,7 @@ use tokio::task::JoinSet; use crate::logger::{Logger, initial_logs}; use crate::redis_client::RedisClient; -use crate::task::{DispatcherPool, TaskRegistry}; +use crate::task::{PythonDispatcher, TaskRegistry}; use fluxqueue_common::{Task, deserialize_raw_task_data}; pub async fn run_worker( @@ -38,7 +38,6 @@ pub async fn run_worker( registered_contexts, ); - let dispatcher_pool = Arc::new(DispatcherPool::new(concurrency)?); let queue_name = Arc::new(queue_name); let executor_ids = generate_executor_ids(concurrency); let atomic_concurrency = Arc::new(AtomicUsize::new(concurrency)); @@ -54,9 +53,10 @@ pub async fn run_worker( let queue_name = Arc::clone(&queue_name); let executor_id = Arc::clone(&executor_ids[i]); let task_registry = Arc::clone(&task_registry); - let dispatcher_pool = Arc::clone(&dispatcher_pool); async move { + let python_dispatcher = Arc::new(PythonDispatcher::new()?); + redis_client .register_executor(&queue_name, &executor_id) .await?; @@ -73,7 +73,7 @@ pub async fn run_worker( executor_id, redis_client, task_registry, - dispatcher_pool, + python_dispatcher, }; Ok::<_, anyhow::Error>((shutdown, ready_check, executor_context)) @@ -131,7 +131,7 @@ struct ExecutorContext { executor_id: Arc, redis_client: Arc, task_registry: Arc, - dispatcher_pool: Arc, + python_dispatcher: Arc, } struct ReadyCheck { @@ -179,7 +179,7 @@ async fn executor_loop( return Ok(()); }; - let task_result = run_task(ctx.executor_id.clone(), ctx.dispatcher_pool.clone(), &task, task_function).await; + let task_result = run_task(ctx.executor_id.clone(), ctx.python_dispatcher.clone(), &task, task_function).await; match task_result { Ok(_) => { @@ -291,7 +291,7 @@ async fn janitor_loop( async fn run_task( executor_id: Arc, - dispatcher_pool: Arc, + python_dispatcher: Arc, task: &Task, task_function: Arc>, ) -> Result<()> { @@ -299,8 +299,7 @@ async fn run_task( let raw_args = Arc::new(task.args.clone()); let raw_kwargs = Arc::new(task.kwargs.clone()); - let dispatcher = dispatcher_pool.get(); - dispatcher + python_dispatcher .execute(executor_id, task_function, task_name, raw_args, raw_kwargs) .await?; @@ -412,7 +411,7 @@ mod tests { #[tokio::test] async fn test_run_task_with_sync_function() -> Result<()> { - let dispatcher_pool = Arc::new(DispatcherPool::new(1)?); + let dispatcher_pool = Arc::new(PythonDispatcher::new()?); let module_path_str = get_test_module_path("test_tasks_module.py"); let task_registry = TaskRegistry::new(&module_path_str, "default")?; @@ -445,7 +444,7 @@ mod tests { #[tokio::test] async fn test_run_task_with_async_function() -> Result<()> { - let dispatcher_pool = Arc::new(DispatcherPool::new(1)?); + let dispatcher_pool = Arc::new(PythonDispatcher::new()?); let module_path_str = get_test_module_path("test_tasks_module.py"); let task_registry = TaskRegistry::new(&module_path_str, "default")?; From 8f1b706926422d5a1d9d466dff3d8132b3798537 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sun, 22 Feb 2026 23:47:54 +0400 Subject: [PATCH 07/31] Add tests for tasks --- crates/fluxqueue-worker/src/task.rs | 235 +++++++++++++++++++--------- 1 file changed, 161 insertions(+), 74 deletions(-) diff --git a/crates/fluxqueue-worker/src/task.rs b/crates/fluxqueue-worker/src/task.rs index 4f222b2..5a8726e 100644 --- a/crates/fluxqueue-worker/src/task.rs +++ b/crates/fluxqueue-worker/src/task.rs @@ -15,11 +15,6 @@ use tokio::sync::{mpsc, oneshot}; use crate::logger::Logger; -type TasksAndContexts = ( - HashMap>>, - HashMap>>, -); - #[derive(Debug)] pub struct TaskRegistry { tasks: Arc>>>>, @@ -28,75 +23,7 @@ pub struct TaskRegistry { impl TaskRegistry { pub fn new(module_path: &str, queue_name: &str) -> Result { - let script = include_str!("../scripts/get_registry.py"); - let script_cstr = CString::new(script)?; - let filename = CString::new("get_registry.py")?; - let module_name = CString::new("get_registry")?; - - let full_current_dir = std::env::current_dir().unwrap(); - let full_module_path = full_current_dir.join(module_path); - let clean_module_path = normalize_path(&full_module_path); - let project_root = full_current_dir - .ancestors() - .find(|p| p.join("tests").exists()) - .unwrap_or(&full_current_dir); - let real_module_path = path_to_module_path(project_root, &clean_module_path); - - if !clean_module_path.exists() || real_module_path.is_none() { - return Err(anyhow!( - "Tasks module path {:?} doesn't exist.", - clean_module_path - )); - } - - let real_module_path = real_module_path.unwrap(); - let module_dir = project_root.to_string_lossy().to_string(); - - let (tasks, contexts) = Python::attach(|py| -> Result { - let module = PyModule::from_code( - py, - script_cstr.as_c_str(), - filename.as_c_str(), - module_name.as_c_str(), - ) - .map_err(|e| anyhow!("Failed to import python module: {}", e))?; - - let registry: Bound<'_, PyDict> = module - .getattr("get_registry") - .map_err(|e| anyhow!("Failed to get 'get_registry' script: {}", e))? - .call1((real_module_path, queue_name, module_dir)) - .map_err(|e| anyhow!("Failed to get tasks: {}", e))? - .cast_into::() - .map_err(|_| anyhow!("Failed to cast result to a Python Dictionary"))?; - - let tasks: HashMap>> = registry - .get_item("tasks")? - .expect("tasks missing") - .cast::() - .map_err(|e| anyhow!("tasks is not a dict: {}", e))? - .iter() - .filter_map(|(key, value)| { - let name: String = key.extract().ok()?; - let func: Py = value.unbind(); - Some((name, Arc::new(func))) - }) - .collect(); - - let contexts: HashMap>> = registry - .get_item("contexts")? - .expect("contexts missing") - .cast::() - .map_err(|e| anyhow!("contexts is not a dict: {}", e))? - .iter() - .filter_map(|(key, value): (Bound, Bound)| { - let name: String = key.extract().ok()?; - let func: Py = value.unbind(); - Some((name, Arc::new(func))) - }) - .collect(); - - Ok((tasks, contexts)) - })?; + let (tasks, contexts) = get_registry(module_path, queue_name)?; Ok(Self { tasks: Arc::new(RwLock::new(tasks)), @@ -276,6 +203,85 @@ async fn run_task( Ok(()) } +type TasksAndContexts = ( + HashMap>>, + HashMap>>, +); + +fn get_registry(module_path: &str, queue_name: &str) -> Result { + let script = include_str!("../scripts/get_registry.py"); + let script_cstr = CString::new(script)?; + let filename = CString::new("get_registry.py")?; + let module_name = CString::new("get_registry")?; + + let full_current_dir = std::env::current_dir().unwrap(); + let full_module_path = full_current_dir.join(module_path); + let clean_module_path = normalize_path(&full_module_path); + let project_root = full_current_dir + .ancestors() + .find(|p| p.join("tests").exists()) + .unwrap_or(&full_current_dir); + let real_module_path = path_to_module_path(project_root, &clean_module_path); + + if !clean_module_path.exists() || real_module_path.is_none() { + return Err(anyhow!( + "Tasks module path {:?} doesn't exist.", + clean_module_path + )); + } + + let real_module_path = real_module_path.unwrap(); + let module_dir = project_root.to_string_lossy().to_string(); + + let result = Python::attach(|py| -> Result { + let module = PyModule::from_code( + py, + script_cstr.as_c_str(), + filename.as_c_str(), + module_name.as_c_str(), + ) + .map_err(|e| anyhow!("Failed to import python module: {}", e))?; + + let registry: Bound<'_, PyDict> = module + .getattr("get_registry") + .map_err(|e| anyhow!("Failed to get 'get_registry' script: {}", e))? + .call1((real_module_path, queue_name, module_dir)) + .map_err(|e| anyhow!("Failed to get tasks: {}", e))? + .cast_into::() + .map_err(|_| anyhow!("Failed to cast result to a Python Dictionary"))?; + + let tasks: HashMap>> = registry + .get_item("tasks")? + .expect("tasks missing") + .cast::() + .map_err(|e| anyhow!("tasks is not a dict: {}", e))? + .iter() + .filter_map(|(key, value)| { + let name: String = key.extract().ok()?; + let func: Py = value.unbind(); + Some((name, Arc::new(func))) + }) + .collect(); + + let contexts: HashMap>> = registry + .get_item("contexts")? + .expect("contexts missing") + .cast::() + .map_err(|e| anyhow!("contexts is not a dict: {}", e))? + .iter() + .filter_map(|(key, value): (Bound, Bound)| { + let name: String = key.extract().ok()?; + let func: Py = value.unbind(); + Some((name, Arc::new(func))) + }) + .collect(); + + Ok((tasks, contexts)) + })?; + + Ok(result) +} + fn normalize_path(path: &Path) -> PathBuf { let mut components = Vec::new(); for comp in path.components() { @@ -306,3 +312,84 @@ fn path_to_module_path(current_dir: &Path, target_path: &Path) -> Option Some(components.join(".")) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_path_to_module_path() -> Result<()> { + let current_dir = Path::new("project"); + let tasks_path = Path::new("../project/tasks.py"); + let normalized_path = normalize_path(tasks_path); + let module_path = path_to_module_path(current_dir, &normalized_path); + let expected_path = Path::new("project/tasks.py"); + + assert_eq!(normalized_path, expected_path); + assert_eq!(module_path, Some("tasks".to_string())); + + Ok(()) + } + + fn get_test_module_path(filename: &str) -> String { + let current_dir = std::env::current_dir().unwrap(); + let test_module_path = current_dir.join("tests").join(filename); + test_module_path.to_str().unwrap().to_string() + } + + #[test] + fn test_get_task_functions_valid_module() -> Result<()> { + let module_path_str = get_test_module_path("test_tasks_module.py"); + let (tasks, _) = get_registry(&module_path_str, "default")?; + + assert_eq!(tasks.len(), 3); + + let task_names: Vec = tasks.iter().map(|(name, _)| name.clone()).collect(); + assert!(task_names.contains(&"task-1".to_string())); + assert!(task_names.contains(&"task-2".to_string())); + assert!(task_names.contains(&"async-task".to_string())); + + assert!(!task_names.contains(&"high-priority-task".to_string())); + + Ok(()) + } + + #[test] + fn test_get_task_functions_different_queue() -> Result<()> { + let module_path_str = get_test_module_path("test_tasks_module.py"); + let (tasks, _) = get_registry(&module_path_str, "high-priority")?; + + let task_names: Vec = tasks.iter().map(|(name, _)| name.clone()).collect(); + assert_eq!(tasks.len(), 1); + assert!(task_names.contains(&"high-priority-task".to_string())); + + Ok(()) + } + + #[test] + fn test_get_task_functions_empty_module() -> Result<()> { + let module_path_str = get_test_module_path("test_tasks_empty.py"); + let (tasks, _) = get_registry(&module_path_str, "default")?; + + assert_eq!(tasks.len(), 0); + + Ok(()) + } + + #[test] + fn test_get_task_functions_duplicate_names() { + let module_path_str = get_test_module_path("test_tasks_duplicate.py"); + + let result = get_registry(&module_path_str, "default"); + assert!(result.is_err()); + + let error_msg = result.unwrap_err().to_string(); + assert!(error_msg.contains("duplicated") || error_msg.contains("duplicate")); + } + + #[test] + fn test_get_task_functions_invalid_path() { + let result = get_registry("nonexistent/path/to/module.py", "default"); + assert!(result.is_err()); + } +} From 3a30b06b305984e6cf1fe6c8a41b32217d0485a1 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sun, 22 Feb 2026 23:48:34 +0400 Subject: [PATCH 08/31] Remove commented lines --- crates/fluxqueue-worker/src/worker.rs | 69 --------------------------- 1 file changed, 69 deletions(-) diff --git a/crates/fluxqueue-worker/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index 328b0a4..9e536f5 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/crates/fluxqueue-worker/src/worker.rs @@ -334,81 +334,12 @@ mod tests { use crate::logger::TestWriter; use std::sync::{Arc, Mutex}; - // #[test] - // fn test_path_to_module_path() -> Result<()> { - // let current_dir = Path::new("project"); - // let tasks_path = Path::new("../project/tasks.py"); - // let normalized_path = normalize_path(tasks_path); - // let module_path = path_to_module_path(current_dir, &normalized_path); - // let expected_path = Path::new("project/tasks.py"); - - // assert_eq!(normalized_path, expected_path); - // assert_eq!(module_path, Some("tasks".to_string())); - - // Ok(()) - // } - fn get_test_module_path(filename: &str) -> String { let current_dir = std::env::current_dir().unwrap(); let test_module_path = current_dir.join("tests").join(filename); test_module_path.to_str().unwrap().to_string() } - // #[test] - // fn test_get_task_functions_valid_module() -> Result<()> { - // let module_path_str = get_test_module_path("test_tasks_module.py"); - // let functions = get_registry(&module_path_str, "default")?; - - // assert_eq!(functions.len(), 3); - - // let task_names: Vec = functions.iter().map(|(name, _)| name.clone()).collect(); - // assert!(task_names.contains(&"task-1".to_string())); - // assert!(task_names.contains(&"task-2".to_string())); - // assert!(task_names.contains(&"async-task".to_string())); - - // assert!(!task_names.contains(&"high-priority-task".to_string())); - - // Ok(()) - // } - - // #[test] - // fn test_get_task_functions_different_queue() -> Result<()> { - // let module_path_str = get_test_module_path("test_tasks_module.py"); - // let functions = get_registry(&module_path_str, "high-priority")?; - - // assert_eq!(functions.len(), 1); - // assert_eq!(functions[0].0, "high-priority-task"); - - // Ok(()) - // } - - // #[test] - // fn test_get_task_functions_empty_module() -> Result<()> { - // let module_path_str = get_test_module_path("test_tasks_empty.py"); - // let functions = get_registry(&module_path_str, "default")?; - - // assert_eq!(functions.len(), 0); - - // Ok(()) - // } - - // #[test] - // fn test_get_task_functions_duplicate_names() { - // let module_path_str = get_test_module_path("test_tasks_duplicate.py"); - - // let result = get_registry(&module_path_str, "default"); - // assert!(result.is_err()); - - // let error_msg = result.unwrap_err().to_string(); - // assert!(error_msg.contains("duplicated") || error_msg.contains("duplicate")); - // } - - // #[test] - // fn test_get_task_functions_invalid_path() { - // let result = get_registry("nonexistent/path/to/module.py", "default"); - // assert!(result.is_err()); - // } - #[tokio::test] async fn test_run_task_with_sync_function() -> Result<()> { let dispatcher_pool = Arc::new(PythonDispatcher::new()?); From 315d0f495029f6b8ed07faeb2a5887245f8ad9e6 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Tue, 24 Feb 2026 00:50:29 +0400 Subject: [PATCH 09/31] Finish context subclasses feature --- .../fluxqueue-worker/scripts/get_registry.py | 28 +++- crates/fluxqueue-worker/src/task.rs | 152 ++++++++++++++---- crates/fluxqueue-worker/src/worker.rs | 25 ++- 3 files changed, 157 insertions(+), 48 deletions(-) diff --git a/crates/fluxqueue-worker/scripts/get_registry.py b/crates/fluxqueue-worker/scripts/get_registry.py index 2857e36..8834b9b 100644 --- a/crates/fluxqueue-worker/scripts/get_registry.py +++ b/crates/fluxqueue-worker/scripts/get_registry.py @@ -2,6 +2,9 @@ import inspect import sys from pathlib import Path +from typing import get_type_hints + +from fluxqueue import Context def get_registry(module_path: str, queue: str, module_dir: str | None = None): @@ -23,12 +26,31 @@ def get_registry(module_path: str, queue: str, module_dir: str | None = None): raise ValueError(f"Task '{task_name}' is duplicated") original_func = getattr(obj, "__wrapped__", obj) - registry["tasks"][task_name] = original_func + + hints = get_type_hints(original_func) + sig = inspect.signature(original_func) + context_params = { + name: hints[name] + for name in sig.parameters + if name in hints + and isinstance(hints[name], type) + and issubclass(hints[name], Context) + } + if not context_params: + context_name = None + else: + context = context_params[next(iter(context_params))] + context_name = getattr(context, "__fluxqueue_context__", None) + + registry["tasks"][task_name] = { + "func": original_func, + "context_name": context_name + } elif inspect.isclass(obj): - context_name = getattr(obj, "__fluxqueue_context__", None) - if not context_name: + if not issubclass(obj, Context): continue + context_name = getattr(obj, "__fluxqueue_context__", None) if registry["contexts"].get(context_name): raise ValueError(f"Context '{context_name}' is duplicated") diff --git a/crates/fluxqueue-worker/src/task.rs b/crates/fluxqueue-worker/src/task.rs index 5a8726e..57ada48 100644 --- a/crates/fluxqueue-worker/src/task.rs +++ b/crates/fluxqueue-worker/src/task.rs @@ -15,10 +15,19 @@ use tokio::sync::{mpsc, oneshot}; use crate::logger::Logger; +type Registry = Arc, T>>>; + +#[derive(Debug)] +pub struct TaskData { + func: Arc>, + context_name: Option>, +} + #[derive(Debug)] pub struct TaskRegistry { - tasks: Arc>>>>, - contexts: Arc>>>>, + tasks: Registry>, + contexts: Registry>>, + context_objects: Registry>>, } impl TaskRegistry { @@ -28,6 +37,7 @@ impl TaskRegistry { Ok(Self { tasks: Arc::new(RwLock::new(tasks)), contexts: Arc::new(RwLock::new(contexts)), + context_objects: Arc::new(RwLock::new(HashMap::new())), }) } @@ -43,15 +53,64 @@ impl TaskRegistry { Ok(context_names) } - pub fn get(&self, name: &str) -> Option>> { + pub fn get_task(&self, name: Arc) -> Option> { let tasks = self.tasks.read().ok()?; - tasks.get(name).cloned() + tasks.get(&name).cloned() + } + + pub fn get_context(&self, name: Arc) -> Option>> { + let contexts = self.contexts.read().ok()?; + contexts.get(&name).cloned() + } + + pub fn get_context_object(&self, name: Arc) -> Option>> { + let ctx_objects = self.context_objects.read().ok()?; + ctx_objects.get(&name).cloned() + } + + pub fn get_task_context(&self, task_data: Arc) -> Result>>> { + let Some(context_name) = task_data.context_name.clone() else { + return Ok(None); + }; + + let context = self.get_context_object(context_name.clone()); + if let Some(context) = context { + return Ok(Some(context.clone())); + } + + let context_class = self.get_context(context_name.clone()); + + if let Some(context_class) = context_class { + Python::attach(|py| -> Result>>> { + let context = context_class.call0(py)?; + let context = Arc::new(context); + + py.detach(|| -> Result<()> { + let mut map = self.context_objects.write().map_err(|_| { + anyhow!( + "Internal Error: 'context_objects' lock poisoned (a thread panicked)" + ) + })?; + + map.insert(context_name.clone(), context.clone()); + Ok(()) + })?; + + Ok(Some(context)) + }) + } else { + Err(anyhow!( + "Context '{}' wasn't found in the registry", + &context_name + )) + } } } struct TaskRequest { + task_registry: Arc, executor_id: Arc, - func: Arc>, + task_data: Arc, task_name: Arc, raw_args: Arc>, raw_kwargs: Arc>, @@ -60,18 +119,20 @@ struct TaskRequest { pub struct PythonDispatcher { tx: mpsc::Sender, + task_registry: Arc, } impl PythonDispatcher { - pub fn new() -> Result { + pub fn new(task_registry: Arc) -> Result { let logical_cores = num_cpus::get(); let (tx, mut rx) = mpsc::channel::(logical_cores * 2); let dispatcher = async move { while let Some(req) = rx.recv().await { run_task( + req.task_registry, req.executor_id, - req.func, + req.task_data, req.task_name, req.raw_args, req.raw_kwargs, @@ -90,13 +151,13 @@ impl PythonDispatcher { }); }); - Ok(Self { tx }) + Ok(Self { tx, task_registry }) } pub async fn execute( &self, executor_id: Arc, - func: Arc>, + task_data: Arc, task_name: Arc, raw_args: Arc>, raw_kwargs: Arc>, @@ -105,8 +166,9 @@ impl PythonDispatcher { self.tx .send(TaskRequest { + task_registry: self.task_registry.clone(), executor_id, - func, + task_data, task_name, raw_args, raw_kwargs, @@ -121,8 +183,9 @@ impl PythonDispatcher { } async fn run_task( + task_registry: Arc, executor_id: Arc, - task_function: Arc>, + task_data: Arc, task_name: Arc, raw_args: Arc>, raw_kwargs: Arc>, @@ -139,11 +202,13 @@ async fn run_task( &task_name ))?; + let context = task_registry.get_task_context(task_data.clone())?; + let maybe_coro = Python::attach(|py| -> Result>> { let py_args = pythonize(py, &task_args).context("Failed to pythonize args")?; let py_kwargs = pythonize(py, &task_kwargs).context("Failed to pythonize kwargs")?; - let args_tuple = if let Ok(list) = py_args.cast::() { + let mut args_tuple = if let Ok(list) = py_args.cast::() { list.to_tuple() } else if let Ok(tuple) = py_args.cast::() { tuple.clone() @@ -151,11 +216,22 @@ async fn run_task( anyhow::bail!("Args must be an array/tuple, found {}", py_args.get_type()); }; + if let Some(context) = context { + let context = context.as_any(); + let prefix = PyTuple::new(py, &[context])?; + + let new_tuple = prefix.add(args_tuple.clone())?; + if let Ok(tuple) = new_tuple.cast::() { + args_tuple = tuple.clone(); + } + } + let kwargs_dict = py_kwargs .cast_into::() .map_err(|_| anyhow!("Kwargs must be a map/dict"))?; - let result = task_function + let result = task_data + .func .call(py, args_tuple, Some(&kwargs_dict)) .map_err(|e| anyhow!("Failed to call Python function: {:?}", e))?; @@ -204,8 +280,8 @@ async fn run_task( } type TasksAndContexts = ( - HashMap>>, - HashMap>>, + HashMap, Arc>, + HashMap, Arc>>, ); fn get_registry(module_path: &str, queue_name: &str) -> Result { @@ -250,20 +326,32 @@ fn get_registry(module_path: &str, queue_name: &str) -> Result .cast_into::() .map_err(|_| anyhow!("Failed to cast result to a Python Dictionary"))?; - let tasks: HashMap>> = registry + let tasks: HashMap, Arc> = registry .get_item("tasks")? .expect("tasks missing") .cast::() .map_err(|e| anyhow!("tasks is not a dict: {}", e))? .iter() - .filter_map(|(key, value)| { - let name: String = key.extract().ok()?; - let func: Py = value.unbind(); - Some((name, Arc::new(func))) - }) - .collect(); - - let contexts: HashMap>> = registry + .map( + |(key, value)| -> Result<(Arc, Arc), anyhow::Error> { + let name: String = key.extract()?; + let data = value.cast::().map_err(|e| anyhow!(e.to_string()))?; + + let func = data + .get_item("func")? + .ok_or_else(|| anyhow!("Couldn't get the function"))?; + let func = Arc::new(func.unbind()); + + let context_name = data + .get_item("context_name")? + .map(|c| Arc::new(c.unbind().to_string())); + + Ok((Arc::new(name), Arc::new(TaskData { func, context_name }))) + }, + ) + .collect::, _>>()?; + + let contexts: HashMap, Arc>> = registry .get_item("contexts")? .expect("contexts missing") .cast::() @@ -272,7 +360,7 @@ fn get_registry(module_path: &str, queue_name: &str) -> Result .filter_map(|(key, value): (Bound, Bound)| { let name: String = key.extract().ok()?; let func: Py = value.unbind(); - Some((name, Arc::new(func))) + Some((Arc::new(name), Arc::new(func))) }) .collect(); @@ -344,12 +432,12 @@ mod tests { assert_eq!(tasks.len(), 3); - let task_names: Vec = tasks.iter().map(|(name, _)| name.clone()).collect(); - assert!(task_names.contains(&"task-1".to_string())); - assert!(task_names.contains(&"task-2".to_string())); - assert!(task_names.contains(&"async-task".to_string())); + let task_names: Vec> = tasks.iter().map(|(name, _)| name.clone()).collect(); + assert!(task_names.contains(&Arc::new("task-1".to_string()))); + assert!(task_names.contains(&Arc::new("task-2".to_string()))); + assert!(task_names.contains(&Arc::new("async-task".to_string()))); - assert!(!task_names.contains(&"high-priority-task".to_string())); + assert!(!task_names.contains(&Arc::new("high-priority-task".to_string()))); Ok(()) } @@ -359,9 +447,9 @@ mod tests { let module_path_str = get_test_module_path("test_tasks_module.py"); let (tasks, _) = get_registry(&module_path_str, "high-priority")?; - let task_names: Vec = tasks.iter().map(|(name, _)| name.clone()).collect(); + let task_names: Vec> = tasks.iter().map(|(name, _)| name.clone()).collect(); assert_eq!(tasks.len(), 1); - assert!(task_names.contains(&"high-priority-task".to_string())); + assert!(task_names.contains(&Arc::new("high-priority-task".to_string()))); Ok(()) } diff --git a/crates/fluxqueue-worker/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index 9e536f5..f4e0924 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/crates/fluxqueue-worker/src/worker.rs @@ -1,5 +1,4 @@ use anyhow::Result; -use pyo3::{Py, PyAny}; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; @@ -8,7 +7,7 @@ use tokio::task::JoinSet; use crate::logger::{Logger, initial_logs}; use crate::redis_client::RedisClient; -use crate::task::{PythonDispatcher, TaskRegistry}; +use crate::task::{PythonDispatcher, TaskData, TaskRegistry}; use fluxqueue_common::{Task, deserialize_raw_task_data}; pub async fn run_worker( @@ -55,7 +54,7 @@ pub async fn run_worker( let task_registry = Arc::clone(&task_registry); async move { - let python_dispatcher = Arc::new(PythonDispatcher::new()?); + let python_dispatcher = Arc::new(PythonDispatcher::new(task_registry.clone())?); redis_client .register_executor(&queue_name, &executor_id) @@ -169,7 +168,7 @@ async fn executor_loop( raw_data.len() )); - let Some(task_function) = ctx.task_registry.get(&task.name) else { + let Some(task_data) = ctx.task_registry.get_task(Arc::new(task.name.to_string())) else { logger.warn(format_args!("Task '{}' not found in registry. Skipping", &task.name)); if let Err(e) = ctx.redis_client .remove_from_processing(&ctx.queue_name, &ctx.executor_id, &raw_data) @@ -179,7 +178,7 @@ async fn executor_loop( return Ok(()); }; - let task_result = run_task(ctx.executor_id.clone(), ctx.python_dispatcher.clone(), &task, task_function).await; + let task_result = run_task(ctx.executor_id.clone(), ctx.python_dispatcher.clone(), &task, task_data.clone()).await; match task_result { Ok(_) => { @@ -293,14 +292,14 @@ async fn run_task( executor_id: Arc, python_dispatcher: Arc, task: &Task, - task_function: Arc>, + task_data: Arc, ) -> Result<()> { let task_name = Arc::new(format!("{}:{}", &task.name, &task.id)); let raw_args = Arc::new(task.args.clone()); let raw_kwargs = Arc::new(task.kwargs.clone()); python_dispatcher - .execute(executor_id, task_function, task_name, raw_args, raw_kwargs) + .execute(executor_id, task_data, task_name, raw_args, raw_kwargs) .await?; Ok(()) @@ -342,11 +341,11 @@ mod tests { #[tokio::test] async fn test_run_task_with_sync_function() -> Result<()> { - let dispatcher_pool = Arc::new(PythonDispatcher::new()?); let module_path_str = get_test_module_path("test_tasks_module.py"); - let task_registry = TaskRegistry::new(&module_path_str, "default")?; + let task_registry = Arc::new(TaskRegistry::new(&module_path_str, "default")?); + let dispatcher_pool = Arc::new(PythonDispatcher::new(task_registry.clone())?); - let task = task_registry.get("task-1"); + let task = task_registry.get_task(Arc::new("task-1".to_string())); assert!(task.is_some()); if let Some(task_func) = task { @@ -375,11 +374,11 @@ mod tests { #[tokio::test] async fn test_run_task_with_async_function() -> Result<()> { - let dispatcher_pool = Arc::new(PythonDispatcher::new()?); let module_path_str = get_test_module_path("test_tasks_module.py"); - let task_registry = TaskRegistry::new(&module_path_str, "default")?; + let task_registry = Arc::new(TaskRegistry::new(&module_path_str, "default")?); + let dispatcher_pool = Arc::new(PythonDispatcher::new(task_registry.clone())?); - let task = task_registry.get("async-task"); + let task = task_registry.get_task(Arc::new("async-task".to_string())); assert!(task.is_some()); if let Some(task_func) = task { From 302aaed86ac213d45f450be916762866443166c3 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Tue, 24 Feb 2026 00:56:38 +0400 Subject: [PATCH 10/31] Fix clippy errors --- crates/fluxqueue-worker/src/task.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/fluxqueue-worker/src/task.rs b/crates/fluxqueue-worker/src/task.rs index 57ada48..cd72c4e 100644 --- a/crates/fluxqueue-worker/src/task.rs +++ b/crates/fluxqueue-worker/src/task.rs @@ -218,7 +218,7 @@ async fn run_task( if let Some(context) = context { let context = context.as_any(); - let prefix = PyTuple::new(py, &[context])?; + let prefix = PyTuple::new(py, [context])?; let new_tuple = prefix.add(args_tuple.clone())?; if let Ok(tuple) = new_tuple.cast::() { From 0897f5d3e962b48c30266a47b404bd7c83bdbf3f Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Tue, 24 Feb 2026 20:49:08 +0400 Subject: [PATCH 11/31] Add process exit on task registry init failure --- crates/fluxqueue-worker/src/worker.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/fluxqueue-worker/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index f4e0924..5f786c9 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/crates/fluxqueue-worker/src/worker.rs @@ -24,7 +24,10 @@ pub async fn run_worker( })?; let redis_client = Arc::new(redis_client); - let task_registry = Arc::new(TaskRegistry::new(&tasks_module_path, &queue_name)?); + let task_registry = Arc::new(TaskRegistry::new(&tasks_module_path, &queue_name).map_err(|e| { + tracing::error!("{}", e); + std::process::exit(1); + })?); let registered_tasks = task_registry.get_registered_tasks()?; let registered_contexts = task_registry.get_registered_contexts()?; From 60baa8ac3c75383112977631c7093bd37963e508 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Tue, 24 Feb 2026 21:07:15 +0400 Subject: [PATCH 12/31] Add thread storage to the context --- crates/fluxqueue-worker/scripts/get_registry.py | 2 +- python/fluxqueue/context.py | 13 ++++++++++--- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/crates/fluxqueue-worker/scripts/get_registry.py b/crates/fluxqueue-worker/scripts/get_registry.py index 8834b9b..0f4da39 100644 --- a/crates/fluxqueue-worker/scripts/get_registry.py +++ b/crates/fluxqueue-worker/scripts/get_registry.py @@ -44,7 +44,7 @@ def get_registry(module_path: str, queue: str, module_dir: str | None = None): registry["tasks"][task_name] = { "func": original_func, - "context_name": context_name + "context_name": context_name, } elif inspect.isclass(obj): if not issubclass(obj, Context): diff --git a/python/fluxqueue/context.py b/python/fluxqueue/context.py index 8fee552..7db19d5 100644 --- a/python/fluxqueue/context.py +++ b/python/fluxqueue/context.py @@ -1,4 +1,5 @@ import inspect +import threading from collections.abc import Callable, Coroutine from typing import Any, Concatenate, ParamSpec, TypeVar, cast, get_type_hints, overload @@ -11,12 +12,18 @@ class Context: __fluxqueue_context__: str | None = None + def __init__(self) -> None: + self._thread_local = threading.local() + + @property + def thread_storage(self) -> dict[str, Any]: + if not hasattr(self._thread_local, "storage"): + self._thread_local.storage = {} + return self._thread_local.storage + def __init_subclass__(cls) -> None: if not cls.__fluxqueue_context__: cls.__fluxqueue_context__ = cls.__name__ - # raise NotImplementedError( - # f"{cls.__name__} is not implemented properly, make sure to add '__fluxqueue_context__' attribute." - # ) C = TypeVar("C", bound=Context) From 04df64d92a038531767c455bab544a1d7d4c0b5e Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Wed, 25 Feb 2026 01:18:08 +0400 Subject: [PATCH 13/31] Add run-worker wrapper script --- .python-version | 1 + crates/fluxqueue-worker/Cargo.toml | 8 +++-- pyproject.toml | 57 ++++++++++++++---------------- scripts/run-worker.sh | 9 +++++ 4 files changed, 43 insertions(+), 32 deletions(-) create mode 100644 .python-version create mode 100755 scripts/run-worker.sh diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..05879da --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.13.11 \ No newline at end of file diff --git a/crates/fluxqueue-worker/Cargo.toml b/crates/fluxqueue-worker/Cargo.toml index 8c9f164..18ed4ed 100644 --- a/crates/fluxqueue-worker/Cargo.toml +++ b/crates/fluxqueue-worker/Cargo.toml @@ -12,6 +12,7 @@ name = "fluxqueue-worker" path = "src/main.rs" [dependencies] +pyo3 = { version = "0.27.2", features = ["auto-initialize"] } pyo3-async-runtimes = { version = "0.27.0", features = ["tokio-runtime"] } uuid = { version = "1.20.0", features = ["v4"] } redis = "0.32.7" @@ -22,12 +23,15 @@ tokio = { version = "1.49.0", features = [ "signal", "time", ] } -pyo3 = { version = "0.27.2", features = ["auto-initialize"] } rmp-serde = "1.3.1" rmpv = { version = "1.3.1", features = ["with-serde"] } serde = { version = "1.0.228", features = ["derive"] } tracing = "0.1.44" -tracing-subscriber = { version = "0.3.22", features = ["env-filter", "fmt", "time"] } +tracing-subscriber = { version = "0.3.22", features = [ + "env-filter", + "fmt", + "time", +] } time = { version = "0.3.47", features = ["local-offset", "macros"] } clap = { version = "4.5.56", features = ["derive", "env"] } anyhow = "1.0.100" diff --git a/pyproject.toml b/pyproject.toml index bc72abe..802f7d5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,26 +8,33 @@ dynamic = ["version"] description = "A lightweight, resource-efficient, high-throughput task queue for Python, written in Rust." readme = "README.md" requires-python = ">=3.11,<3.15" -keywords = ["task-queue", "queue", "redis", "async", "background-tasks", "worker", "rust", "fluxqueue"] +keywords = [ + "task-queue", + "queue", + "redis", + "async", + "background-tasks", + "worker", + "rust", + "fluxqueue", +] license = "Apache-2.0" license-files = ["LICENSE"] -authors = [ - {name = "Giorgi Merebashvili", email = "mereba2627@gmail.com"} -] +authors = [{ name = "Giorgi Merebashvili", email = "mereba2627@gmail.com" }] classifiers = [ - "Development Status :: 5 - Production/Stable", - "Intended Audience :: Developers", - "Programming Language :: Python", - "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3 :: Only", - "Programming Language :: Python :: 3.11", - "Programming Language :: Python :: 3.12", - "Programming Language :: Python :: 3.13", - "Programming Language :: Python :: 3.14", - "Programming Language :: Rust", - "Operating System :: OS Independent", - "Topic :: Software Development :: Libraries", - "Topic :: System :: Distributed Computing", + "Development Status :: 5 - Production/Stable", + "Intended Audience :: Developers", + "Programming Language :: Python", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3 :: Only", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3.14", + "Programming Language :: Rust", + "Operating System :: OS Independent", + "Topic :: Software Development :: Libraries", + "Topic :: System :: Distributed Computing", ] [project.urls] @@ -39,19 +46,9 @@ Changelog = "https://github.com/CCXLV/fluxqueue/releases" [project.optional-dependencies] dev = ["ruff"] -tests = [ - "pytest-asyncio", - "python-dotenv", - "redis" -] -cli = [ - "fluxqueue-cli>=0.1.0b4" -] -docs = [ - "mkdocs-material", - "mkdocstrings", - "mkdocstrings-python", -] +tests = ["pytest-asyncio", "python-dotenv", "redis"] +cli = ["fluxqueue-cli>=0.1.0b4"] +docs = ["mkdocs-material", "mkdocstrings", "mkdocstrings-python"] build = ["maturin"] [tool.maturin] diff --git a/scripts/run-worker.sh b/scripts/run-worker.sh new file mode 100755 index 0000000..7778beb --- /dev/null +++ b/scripts/run-worker.sh @@ -0,0 +1,9 @@ +#!/bin/bash + +LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") + +if [ -n "$LIB_DIR" ]; then + export LD_LIBRARY_PATH="$LIB_DIR:$LD_LIBRARY_PATH" +fi + +exec "./fluxqueue-worker" "$@" From 5a01f6dc949b961e7276ed34a981fb6e5377429d Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Wed, 25 Feb 2026 02:18:50 +0400 Subject: [PATCH 14/31] Fix unnecessary check in context decorator --- python/fluxqueue/context.py | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/python/fluxqueue/context.py b/python/fluxqueue/context.py index 7db19d5..e77596a 100644 --- a/python/fluxqueue/context.py +++ b/python/fluxqueue/context.py @@ -83,14 +83,9 @@ def _with_context( new_sig = sig.replace(parameters=[sig.parameters[n] for n in non_context_params]) - if inspect.iscoroutinefunction(func): - wrapper = _task_decorator( - cast(Any, func), name=name, queue=queue, max_retries=max_retries, core=core - ) - else: - wrapper = _task_decorator( - cast(Any, func), name=name, queue=queue, max_retries=max_retries, core=core - ) + wrapper = _task_decorator( + cast(Any, func), name=name, queue=queue, max_retries=max_retries, core=core + ) cast(Any, wrapper).__signature__ = new_sig return wrapper From 180aa29747b79cce26bca8de3c5d11d2fcfa57f0 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Wed, 25 Feb 2026 13:58:24 +0400 Subject: [PATCH 15/31] Update thread storage to ContextVar instead of thread.local() --- python/fluxqueue/context.py | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/python/fluxqueue/context.py b/python/fluxqueue/context.py index e77596a..5863540 100644 --- a/python/fluxqueue/context.py +++ b/python/fluxqueue/context.py @@ -1,6 +1,6 @@ import inspect -import threading from collections.abc import Callable, Coroutine +from contextvars import ContextVar from typing import Any, Concatenate, ParamSpec, TypeVar, cast, get_type_hints, overload from ._core import FluxQueueCore @@ -13,13 +13,28 @@ class Context: __fluxqueue_context__: str | None = None def __init__(self) -> None: - self._thread_local = threading.local() + self._thread_storage: ContextVar[dict[str, Any]] = ContextVar( + "thread_storage", default=None + ) @property def thread_storage(self) -> dict[str, Any]: - if not hasattr(self._thread_local, "storage"): - self._thread_local.storage = {} - return self._thread_local.storage + """ + Context-scoped storage dictionary. + + Returns a mutable dictionary associated with the current + execution context. The storage is isolated per thread or + async task using ContextVar, ensuring safe concurrent usage. + + The dictionary is initialized lazily on first access. + """ + storage = self._thread_storage.get(None) + + if storage is None: + storage = {} + self._thread_storage.set(storage) + + return storage def __init_subclass__(cls) -> None: if not cls.__fluxqueue_context__: From acdb46813cfdae400c9689ed936b2c75ebdea7b8 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Wed, 25 Feb 2026 23:03:08 +0400 Subject: [PATCH 16/31] Add task metadata feat --- crates/fluxqueue-worker/src/task.rs | 112 ++++++++++++++++++++------ crates/fluxqueue-worker/src/worker.rs | 25 +++--- python/fluxqueue/__init__.py | 7 +- python/fluxqueue/_task.py | 3 +- python/fluxqueue/context.py | 45 ++++++++--- python/fluxqueue/models.py | 17 ++++ 6 files changed, 153 insertions(+), 56 deletions(-) create mode 100644 python/fluxqueue/models.py diff --git a/crates/fluxqueue-worker/src/task.rs b/crates/fluxqueue-worker/src/task.rs index cd72c4e..70fab87 100644 --- a/crates/fluxqueue-worker/src/task.rs +++ b/crates/fluxqueue-worker/src/task.rs @@ -1,4 +1,5 @@ use anyhow::{Context, Result, anyhow}; +use fluxqueue_common::Task; use pyo3::exceptions::PyRuntimeError; use pyo3::prelude::*; use pyo3::types::{PyAnyMethods, PyDict, PyDictMethods, PyList, PyModule, PyTuple}; @@ -112,8 +113,7 @@ struct TaskRequest { executor_id: Arc, task_data: Arc, task_name: Arc, - raw_args: Arc>, - raw_kwargs: Arc>, + task: Arc, resp_tx: oneshot::Sender>, } @@ -134,8 +134,7 @@ impl PythonDispatcher { req.executor_id, req.task_data, req.task_name, - req.raw_args, - req.raw_kwargs, + req.task, ) .await .map_err(|e| PyRuntimeError::new_err(e.to_string()))?; @@ -159,8 +158,7 @@ impl PythonDispatcher { executor_id: Arc, task_data: Arc, task_name: Arc, - raw_args: Arc>, - raw_kwargs: Arc>, + task: Arc, ) -> Result<()> { let (resp_tx, resp_rx) = oneshot::channel(); @@ -170,8 +168,7 @@ impl PythonDispatcher { executor_id, task_data, task_name, - raw_args, - raw_kwargs, + task, resp_tx, }) .await @@ -182,29 +179,39 @@ impl PythonDispatcher { } } +struct CoroWithContext { + args: Py, + kwargs: Py, + context: Arc>, +} + +struct MaybeCoro { + func: Arc>, + with_context: Option, +} + async fn run_task( task_registry: Arc, executor_id: Arc, task_data: Arc, task_name: Arc, - raw_args: Arc>, - raw_kwargs: Arc>, + task: Arc, ) -> Result<()> { let logger = Logger::new(format!("EXECUTOR {}", &executor_id)); let duration_start = Instant::now(); - let task_args: Value = from_slice(&raw_args).context(format!( + let task_args: Value = from_slice(&task.args).context(format!( "Failed to deserialize task '{}' function args", &task_name ))?; - let task_kwargs: Value = from_slice(&raw_kwargs).context(format!( + let task_kwargs: Value = from_slice(&task.kwargs).context(format!( "Failed to deserialize task '{}' function kwargs", &task_name ))?; let context = task_registry.get_task_context(task_data.clone())?; - let maybe_coro = Python::attach(|py| -> Result>> { + let maybe_coro = Python::attach(|py| -> Result> { let py_args = pythonize(py, &task_args).context("Failed to pythonize args")?; let py_kwargs = pythonize(py, &task_kwargs).context("Failed to pythonize kwargs")?; @@ -216,7 +223,11 @@ async fn run_task( anyhow::bail!("Args must be an array/tuple, found {}", py_args.get_type()); }; - if let Some(context) = context { + if let Some(context) = context.as_ref() { + let task_metadata = get_task_metadata(py, task.clone())?; + let metadata_var = context.getattr(py, "_metadata_var")?; + metadata_var.call_method1(py, "set", (task_metadata,))?; + let context = context.as_any(); let prefix = PyTuple::new(py, [context])?; @@ -230,19 +241,25 @@ async fn run_task( .cast_into::() .map_err(|_| anyhow!("Kwargs must be a map/dict"))?; - let result = task_data - .func - .call(py, args_tuple, Some(&kwargs_dict)) - .map_err(|e| anyhow!("Failed to call Python function: {:?}", e))?; - - let bound_result = result.bind(py); - let is_coroutine = bound_result - .hasattr("__await__") - .map_err(|_| anyhow!("Failed to check if result is awaitable"))?; + let is_coroutine = is_coroutine(py, task_data.func.clone())?; if is_coroutine { - Ok(Some(result)) + let with_context = context.map(|context| CoroWithContext { + args: args_tuple.unbind(), + kwargs: kwargs_dict.unbind(), + context, + }); + + Ok(Some(MaybeCoro { + func: task_data.func.clone(), + with_context, + })) } else { + task_data + .func + .call(py, args_tuple.clone(), Some(&kwargs_dict)) + .map_err(|e| anyhow!("Failed to call Python function: {:?}", e))?; + let duration_end = duration_start.elapsed(); logger.info(format_args!( "Task '{}' successfully finished in {}ms", @@ -264,8 +281,27 @@ async fn run_task( anyhow!(e.to_string()) })?; - if let Some(coro) = maybe_coro { - let fut = Python::attach(|py| into_future(coro.into_bound(py)))?; + if let Some(maybe_coro) = maybe_coro { + let fut = Python::attach(|py| { + if let Some(with_context) = maybe_coro.with_context { + let task_metadata = get_task_metadata(py, task.clone()) + .map_err(|e| PyRuntimeError::new_err(e.to_string()))?; + let result = with_context.context.call_method1( + py, + "_run_async_task", + ( + maybe_coro.func.as_any(), + task_metadata, + with_context.args, + Some(with_context.kwargs), + ), + )?; + into_future(result.into_bound(py)) + } else { + let func = maybe_coro.func.clone_ref(py); + into_future(func.into_bound(py)) + } + })?; fut.await?; let duration_end = duration_start.elapsed(); @@ -370,6 +406,30 @@ fn get_registry(module_path: &str, queue_name: &str) -> Result Ok(result) } +fn get_task_metadata(py: Python<'_>, task: Arc) -> Result> { + let module = py.import("fluxqueue.models")?.unbind(); + let task_metadata = module.call_method1( + py, + "TaskMetadata", + ( + task.id.clone(), + task.retries, + task.max_retries, + task.created_at, + ), + )?; + + Ok(task_metadata) +} + +fn is_coroutine(py: Python<'_>, func: Arc>) -> Result { + let inspect = py.import("inspect")?; + let is_coro: bool = inspect + .call_method1("iscoroutinefunction", (func.as_any(),))? + .extract()?; + Ok(is_coro) +} + fn normalize_path(path: &Path) -> PathBuf { let mut components = Vec::new(); for comp in path.components() { diff --git a/crates/fluxqueue-worker/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index 5f786c9..b18c0aa 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/crates/fluxqueue-worker/src/worker.rs @@ -24,10 +24,12 @@ pub async fn run_worker( })?; let redis_client = Arc::new(redis_client); - let task_registry = Arc::new(TaskRegistry::new(&tasks_module_path, &queue_name).map_err(|e| { - tracing::error!("{}", e); - std::process::exit(1); - })?); + let task_registry = Arc::new(TaskRegistry::new(&tasks_module_path, &queue_name).map_err( + |e| { + tracing::error!("{}", e); + std::process::exit(1); + }, + )?); let registered_tasks = task_registry.get_registered_tasks()?; let registered_contexts = task_registry.get_registered_contexts()?; @@ -164,6 +166,7 @@ async fn executor_loop( Ok(Some(raw_data)) => { let task = deserialize_raw_task_data(&raw_data)?; let task_name = format!("{}:{}", &task.name, &task.id); + let actual_task_name = task.name.clone(); logger.info(format_args!( "Received a task '{}' with a total of {} Bytes", @@ -181,7 +184,7 @@ async fn executor_loop( return Ok(()); }; - let task_result = run_task(ctx.executor_id.clone(), ctx.python_dispatcher.clone(), &task, task_data.clone()).await; + let task_result = run_task(ctx.executor_id.clone(), ctx.python_dispatcher.clone(), Arc::new(task), task_data.clone()).await; match task_result { Ok(_) => { @@ -195,7 +198,7 @@ async fn executor_loop( if let Err(err) = ctx.redis_client .mark_as_failed(&ctx.queue_name, &ctx.executor_id, &raw_data) .await { - logger.error(format_args!("Failed to mark the task '{}' as failed: {}", &task.name, err)); + logger.error(format_args!("Failed to mark the task '{}' as failed: {}", actual_task_name, err)); } } } @@ -294,15 +297,13 @@ async fn janitor_loop( async fn run_task( executor_id: Arc, python_dispatcher: Arc, - task: &Task, + task: Arc, task_data: Arc, ) -> Result<()> { let task_name = Arc::new(format!("{}:{}", &task.name, &task.id)); - let raw_args = Arc::new(task.args.clone()); - let raw_kwargs = Arc::new(task.kwargs.clone()); python_dispatcher - .execute(executor_id, task_data, task_name, raw_args, raw_kwargs) + .execute(executor_id, task_data, task_name, task) .await?; Ok(()) @@ -365,7 +366,7 @@ mod tests { let result = run_task( Arc::new("test".to_string()), dispatcher_pool.clone(), - &task, + Arc::new(task), task_func, ) .await; @@ -398,7 +399,7 @@ mod tests { let result = run_task( Arc::new("test".to_string()), dispatcher_pool.clone(), - &task, + Arc::new(task), task_func, ) .await; diff --git a/python/fluxqueue/__init__.py b/python/fluxqueue/__init__.py index f7d1134..b0256af 100644 --- a/python/fluxqueue/__init__.py +++ b/python/fluxqueue/__init__.py @@ -1,4 +1,3 @@ -__all__ = ["Context", "FluxQueue"] - -from .client import FluxQueue -from .context import Context +from .client import FluxQueue as FluxQueue +from .context import Context as Context +from .models import TaskMetadata as TaskMetadata diff --git a/python/fluxqueue/_task.py b/python/fluxqueue/_task.py index 207f857..7f9d9b0 100644 --- a/python/fluxqueue/_task.py +++ b/python/fluxqueue/_task.py @@ -45,14 +45,13 @@ def _task_decorator( if return_type and return_type is not type(None): raise TypeError(f"Task function must return None, got {return_type}") - is_async = inspect.iscoroutinefunction(func) task_name = get_task_name(func, name) # TODO: Add unique identifier 'fluxqueue' just to be 100% sure cast(Any, func).task_name = task_name cast(Any, func).queue = queue - if is_async: + if inspect.iscoroutinefunction(func): @wraps(func) async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> None: diff --git a/python/fluxqueue/context.py b/python/fluxqueue/context.py index 5863540..13727e5 100644 --- a/python/fluxqueue/context.py +++ b/python/fluxqueue/context.py @@ -1,10 +1,12 @@ import inspect +import threading from collections.abc import Callable, Coroutine from contextvars import ContextVar from typing import Any, Concatenate, ParamSpec, TypeVar, cast, get_type_hints, overload from ._core import FluxQueueCore from ._task import _task_decorator +from .models import TaskMetadata P = ParamSpec("P") @@ -13,28 +15,47 @@ class Context: __fluxqueue_context__: str | None = None def __init__(self) -> None: - self._thread_storage: ContextVar[dict[str, Any]] = ContextVar( - "thread_storage", default=None + self._thread_local = threading.local() + self._metadata_var: ContextVar[TaskMetadata] = ContextVar( + "task_metadata", default=None ) @property def thread_storage(self) -> dict[str, Any]: """ - Context-scoped storage dictionary. + Retrieves the thread-persistent storage dictionary. - Returns a mutable dictionary associated with the current - execution context. The storage is isolated per thread or - async task using ContextVar, ensuring safe concurrent usage. + Returns a dictionary that persists across all tasks executed by the current worker. + Used for storing long-lived resources like database engines and connection + pools to avoid re-initialization overhead. + """ + if not hasattr(self._thread_local, "storage"): + self._thread_local.storage = {} + + return self._thread_local.storage - The dictionary is initialized lazily on first access. + @property + def metadata(self) -> TaskMetadata: """ - storage = self._thread_storage.get(None) + Returns metadata isolated to the current task. - if storage is None: - storage = {} - self._thread_storage.set(storage) + Returns a TaskMetadata instance containing execution details like + retry counts and task IDs. This property uses ContextVars to ensure + data isolation during concurrent task execution on the same thread. + """ + return self._metadata_var.get() - return storage + async def _run_async_task( + self, func: Callable, metadata: TaskMetadata, args, kwargs + ): + """ + This function is for internal use only. + """ + token = self._metadata_var.set(metadata) + try: + await func(*args, **kwargs) + finally: + self._metadata_var.reset(token) def __init_subclass__(cls) -> None: if not cls.__fluxqueue_context__: diff --git a/python/fluxqueue/models.py b/python/fluxqueue/models.py new file mode 100644 index 0000000..a6812cc --- /dev/null +++ b/python/fluxqueue/models.py @@ -0,0 +1,17 @@ +from dataclasses import dataclass + + +@dataclass(slots=True) +class TaskMetadata: + """ + Read-only metadata for a FluxQueue task. + """ + + task_id: str + """Unique identifier for the current task execution.""" + retry_count: int + """Number of times this task has been retried.""" + max_retries: int + """Maximum number of retry attempts allowed before failure.""" + enqueued_at: int + """ISO 8601 timestamp of when the task was originally enqueued.""" From e24bd556a8d35d06e0792bffa9dffcd8121ad227 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 19:51:36 +0400 Subject: [PATCH 17/31] Finish --- crates/fluxqueue-worker/src/task.rs | 40 ++++++++++++----------------- python/fluxqueue/context.py | 4 +-- 2 files changed, 18 insertions(+), 26 deletions(-) diff --git a/crates/fluxqueue-worker/src/task.rs b/crates/fluxqueue-worker/src/task.rs index 70fab87..0907f42 100644 --- a/crates/fluxqueue-worker/src/task.rs +++ b/crates/fluxqueue-worker/src/task.rs @@ -100,10 +100,7 @@ impl TaskRegistry { Ok(Some(context)) }) } else { - Err(anyhow!( - "Context '{}' wasn't found in the registry", - &context_name - )) + Ok(None) } } } @@ -179,15 +176,11 @@ impl PythonDispatcher { } } -struct CoroWithContext { - args: Py, - kwargs: Py, - context: Arc>, -} - struct MaybeCoro { func: Arc>, - with_context: Option, + args: Py, + kwargs: Py, + context: Option>>, } async fn run_task( @@ -244,15 +237,11 @@ async fn run_task( let is_coroutine = is_coroutine(py, task_data.func.clone())?; if is_coroutine { - let with_context = context.map(|context| CoroWithContext { + Ok(Some(MaybeCoro { + func: task_data.func.clone(), args: args_tuple.unbind(), kwargs: kwargs_dict.unbind(), context, - }); - - Ok(Some(MaybeCoro { - func: task_data.func.clone(), - with_context, })) } else { task_data @@ -283,23 +272,28 @@ async fn run_task( if let Some(maybe_coro) = maybe_coro { let fut = Python::attach(|py| { - if let Some(with_context) = maybe_coro.with_context { + if let Some(context) = maybe_coro.context { let task_metadata = get_task_metadata(py, task.clone()) .map_err(|e| PyRuntimeError::new_err(e.to_string()))?; - let result = with_context.context.call_method1( + let result = context.call_method1( py, "_run_async_task", ( maybe_coro.func.as_any(), task_metadata, - with_context.args, - Some(with_context.kwargs), + maybe_coro.args, + Some(maybe_coro.kwargs), ), )?; into_future(result.into_bound(py)) } else { - let func = maybe_coro.func.clone_ref(py); - into_future(func.into_bound(py)) + let result = maybe_coro + .func + .call(py, maybe_coro.args, Some(maybe_coro.kwargs.bind(py))) + .map_err(|e| anyhow!("Failed to call Python function: {:?}", e)) + .map_err(|e| PyRuntimeError::new_err(e.to_string()))?; + + into_future(result.into_bound(py)) } })?; fut.await?; diff --git a/python/fluxqueue/context.py b/python/fluxqueue/context.py index 13727e5..23d6f16 100644 --- a/python/fluxqueue/context.py +++ b/python/fluxqueue/context.py @@ -16,9 +16,7 @@ class Context: def __init__(self) -> None: self._thread_local = threading.local() - self._metadata_var: ContextVar[TaskMetadata] = ContextVar( - "task_metadata", default=None - ) + self._metadata_var: ContextVar[TaskMetadata] = ContextVar("task_metadata") @property def thread_storage(self) -> dict[str, Any]: From 084478157f48c55358fcc892451dfb91f4a4a930 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 20:44:47 +0400 Subject: [PATCH 18/31] Update cargo tests --- .github/actions/tests/action.yml | 1 + scripts/set_ld_path.sh | 4 ++++ 2 files changed, 5 insertions(+) create mode 100755 scripts/set_ld_path.sh diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index 6e1d14e..b47dfdb 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -30,4 +30,5 @@ runs: shell: bash -euxo pipefail {0} run: | uv run pytest tests/ + ./scripts/set_ld_path.sh cargo test --workspace --locked diff --git a/scripts/set_ld_path.sh b/scripts/set_ld_path.sh new file mode 100755 index 0000000..96e3d46 --- /dev/null +++ b/scripts/set_ld_path.sh @@ -0,0 +1,4 @@ +#!/bin/bash + +LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") +export LD_LIBRARY_PATH="$LIB_DIR:$LD_LIBRARY_PATH" From d6902196854007abc8c79f48bfec30fe0309d0ac Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 20:53:26 +0400 Subject: [PATCH 19/31] Fix tests --- .github/actions/tests/action.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index b47dfdb..4dff2fc 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -30,5 +30,5 @@ runs: shell: bash -euxo pipefail {0} run: | uv run pytest tests/ - ./scripts/set_ld_path.sh + source ./scripts/set_ld_path.sh cargo test --workspace --locked From e6db82f98a52125a471764182f4662990c29e5e3 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 21:02:01 +0400 Subject: [PATCH 20/31] Fix tests --- .github/actions/tests/action.yml | 3 ++- scripts/set_ld_path.sh | 4 ---- 2 files changed, 2 insertions(+), 5 deletions(-) delete mode 100755 scripts/set_ld_path.sh diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index 4dff2fc..d74633e 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -30,5 +30,6 @@ runs: shell: bash -euxo pipefail {0} run: | uv run pytest tests/ - source ./scripts/set_ld_path.sh + LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") + export LD_LIBRARY_PATH="$LIB_DIR:$LD_LIBRARY_PATH" cargo test --workspace --locked diff --git a/scripts/set_ld_path.sh b/scripts/set_ld_path.sh deleted file mode 100755 index 96e3d46..0000000 --- a/scripts/set_ld_path.sh +++ /dev/null @@ -1,4 +0,0 @@ -#!/bin/bash - -LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") -export LD_LIBRARY_PATH="$LIB_DIR:$LD_LIBRARY_PATH" From ece79b7eefa8b68d9e7a5b128a2d83cd3112d669 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 21:06:46 +0400 Subject: [PATCH 21/31] Fix tests --- .github/actions/tests/action.yml | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index d74633e..6dcc66c 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -30,6 +30,16 @@ runs: shell: bash -euxo pipefail {0} run: | uv run pytest tests/ + LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") - export LD_LIBRARY_PATH="$LIB_DIR:$LD_LIBRARY_PATH" - cargo test --workspace --locked + PYTHON_PREFIX=$(python3 -c "import sys; print(sys.prefix)") + + if [[ "$RUNNER_OS" == "macOS" ]]; then + export DYLD_LIBRARY_PATH="$LIB_DIR:${DYLD_LIBRARY_PATH:-}" + else + export LD_LIBRARY_PATH="$LIB_DIR:${LD_LIBRARY_PATH:-}" + fi + + export PYTHONHOME="$PYTHON_PREFIX" + + cargo test --workspace --locked -- --nocapture \ No newline at end of file From d4c06ced890a84a2acfc28a108ed55c53580a9f5 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 21:15:08 +0400 Subject: [PATCH 22/31] Fix tests --- .github/actions/tests/action.yml | 3 --- 1 file changed, 3 deletions(-) diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index 6dcc66c..791e8ec 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -32,7 +32,6 @@ runs: uv run pytest tests/ LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") - PYTHON_PREFIX=$(python3 -c "import sys; print(sys.prefix)") if [[ "$RUNNER_OS" == "macOS" ]]; then export DYLD_LIBRARY_PATH="$LIB_DIR:${DYLD_LIBRARY_PATH:-}" @@ -40,6 +39,4 @@ runs: export LD_LIBRARY_PATH="$LIB_DIR:${LD_LIBRARY_PATH:-}" fi - export PYTHONHOME="$PYTHON_PREFIX" - cargo test --workspace --locked -- --nocapture \ No newline at end of file From 7f57bf40de5d2f9563e05cac88d7a5722da1d1a1 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 21:24:30 +0400 Subject: [PATCH 23/31] Fix tests once again --- .github/actions/tests/action.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index 791e8ec..ea1f9ba 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -29,8 +29,6 @@ runs: - name: Run Tests shell: bash -euxo pipefail {0} run: | - uv run pytest tests/ - LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") if [[ "$RUNNER_OS" == "macOS" ]]; then @@ -39,4 +37,6 @@ runs: export LD_LIBRARY_PATH="$LIB_DIR:${LD_LIBRARY_PATH:-}" fi + export PYTHONPATH="$(pwd)/python:$PYTHONPATH" + cargo test --workspace --locked -- --nocapture \ No newline at end of file From 371f6b16fd4f65f1536d00b91bfcf56f6b84a873 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 21:29:09 +0400 Subject: [PATCH 24/31] Fix tests --- .github/actions/tests/action.yml | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index ea1f9ba..58ca247 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -29,6 +29,8 @@ runs: - name: Run Tests shell: bash -euxo pipefail {0} run: | + pytest tests/ + LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") if [[ "$RUNNER_OS" == "macOS" ]]; then @@ -37,6 +39,6 @@ runs: export LD_LIBRARY_PATH="$LIB_DIR:${LD_LIBRARY_PATH:-}" fi - export PYTHONPATH="$(pwd)/python:$PYTHONPATH" + source ./venv/bin/activate cargo test --workspace --locked -- --nocapture \ No newline at end of file From a27a5f28ccee79b1f1d1c6cb0d5296dee749bb65 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 21:39:51 +0400 Subject: [PATCH 25/31] Move tests to bash script --- .github/actions/tests/action.yml | 15 +-------------- scripts/run-tests.sh | 13 +++++++++++++ 2 files changed, 14 insertions(+), 14 deletions(-) create mode 100755 scripts/run-tests.sh diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index 58ca247..91935bc 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -28,17 +28,4 @@ runs: - name: Run Tests shell: bash -euxo pipefail {0} - run: | - pytest tests/ - - LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") - - if [[ "$RUNNER_OS" == "macOS" ]]; then - export DYLD_LIBRARY_PATH="$LIB_DIR:${DYLD_LIBRARY_PATH:-}" - else - export LD_LIBRARY_PATH="$LIB_DIR:${LD_LIBRARY_PATH:-}" - fi - - source ./venv/bin/activate - - cargo test --workspace --locked -- --nocapture \ No newline at end of file + run: scripts/run-tests.sh \ No newline at end of file diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh new file mode 100755 index 0000000..eb2d8e9 --- /dev/null +++ b/scripts/run-tests.sh @@ -0,0 +1,13 @@ +#!/bin/bash + +set -euxo pipefail + +# Running Python tests +uv run pytest tests/ + +# Setting up LD_LIBRARY_PATH environment variable +# https://pyo3.rs/v0.28.2/building-and-distribution.html#dynamically-embedding-the-python-interpreter +LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") +export LD_LIBRARY_PATH="$LIB_DIR:${LD_LIBRARY_PATH:-}" + +uv run cargo test --workspace --locked From 8435fa6b6805c227cc9b15a09db7c9704a99cea1 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 21:44:53 +0400 Subject: [PATCH 26/31] Fix run-tests --- scripts/run-tests.sh | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index eb2d8e9..4d01f17 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -2,12 +2,12 @@ set -euxo pipefail -# Running Python tests -uv run pytest tests/ +source .venv/bin/activate # Setting up LD_LIBRARY_PATH environment variable # https://pyo3.rs/v0.28.2/building-and-distribution.html#dynamically-embedding-the-python-interpreter LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") export LD_LIBRARY_PATH="$LIB_DIR:${LD_LIBRARY_PATH:-}" -uv run cargo test --workspace --locked +pytest tests/ +cargo test --workspace --locked From fc4d10d7316b3f93f9cfb939b48195013ec63abd Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 21:50:43 +0400 Subject: [PATCH 27/31] Fix run-tests --- scripts/run-tests.sh | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh index 4d01f17..465c8bf 100755 --- a/scripts/run-tests.sh +++ b/scripts/run-tests.sh @@ -6,8 +6,15 @@ source .venv/bin/activate # Setting up LD_LIBRARY_PATH environment variable # https://pyo3.rs/v0.28.2/building-and-distribution.html#dynamically-embedding-the-python-interpreter +OS="$(uname)" LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") -export LD_LIBRARY_PATH="$LIB_DIR:${LD_LIBRARY_PATH:-}" + +if [[ "$OS" == "Darwin" ]]; then + export DYLD_LIBRARY_PATH="$LIB_DIR:${DYLD_LIBRARY_PATH:-}" +else + export LD_LIBRARY_PATH="$LIB_DIR:${LD_LIBRARY_PATH:-}" +fi pytest tests/ + cargo test --workspace --locked From d507fd9e334b59042c1f2a8730fa26b289f53814 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 22:26:29 +0400 Subject: [PATCH 28/31] Fix tests --- .github/actions/tests/action.yml | 6 ++++-- scripts/run-tests.sh | 20 -------------------- 2 files changed, 4 insertions(+), 22 deletions(-) delete mode 100755 scripts/run-tests.sh diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index 91935bc..fb9c917 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -18,7 +18,7 @@ runs: - name: Install dependencies shell: bash -euxo pipefail {0} - run: uv sync --extra tests + run: pip install -e .[tests] - name: Set up Rust uses: dtolnay/rust-toolchain@stable @@ -28,4 +28,6 @@ runs: - name: Run Tests shell: bash -euxo pipefail {0} - run: scripts/run-tests.sh \ No newline at end of file + run: | + pytest tests/ + cargo test --workspace --locked diff --git a/scripts/run-tests.sh b/scripts/run-tests.sh deleted file mode 100755 index 465c8bf..0000000 --- a/scripts/run-tests.sh +++ /dev/null @@ -1,20 +0,0 @@ -#!/bin/bash - -set -euxo pipefail - -source .venv/bin/activate - -# Setting up LD_LIBRARY_PATH environment variable -# https://pyo3.rs/v0.28.2/building-and-distribution.html#dynamically-embedding-the-python-interpreter -OS="$(uname)" -LIB_DIR=$(python3 -c "import sysconfig; print(sysconfig.get_config_var('LIBDIR'))") - -if [[ "$OS" == "Darwin" ]]; then - export DYLD_LIBRARY_PATH="$LIB_DIR:${DYLD_LIBRARY_PATH:-}" -else - export LD_LIBRARY_PATH="$LIB_DIR:${LD_LIBRARY_PATH:-}" -fi - -pytest tests/ - -cargo test --workspace --locked From 7f1bd549b613103b90ca72c8e0b168827531d09f Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 22:29:51 +0400 Subject: [PATCH 29/31] Update tests --- .github/actions/tests/action.yml | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index fb9c917..562cb6a 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -18,7 +18,9 @@ runs: - name: Install dependencies shell: bash -euxo pipefail {0} - run: pip install -e .[tests] + run: | + python3 -m venv .venv + uv sync --extra tests - name: Set up Rust uses: dtolnay/rust-toolchain@stable @@ -29,5 +31,5 @@ runs: - name: Run Tests shell: bash -euxo pipefail {0} run: | - pytest tests/ - cargo test --workspace --locked + uv run pytest tests/ + uv run cargo test --workspace --locked From fd2c9792cee89b4c06565981faa37d691a3a8d87 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 22:34:32 +0400 Subject: [PATCH 30/31] Revert "Update tests" This reverts commit 7f1bd549b613103b90ca72c8e0b168827531d09f. --- .github/actions/tests/action.yml | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index 562cb6a..fb9c917 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -18,9 +18,7 @@ runs: - name: Install dependencies shell: bash -euxo pipefail {0} - run: | - python3 -m venv .venv - uv sync --extra tests + run: pip install -e .[tests] - name: Set up Rust uses: dtolnay/rust-toolchain@stable @@ -31,5 +29,5 @@ runs: - name: Run Tests shell: bash -euxo pipefail {0} run: | - uv run pytest tests/ - uv run cargo test --workspace --locked + pytest tests/ + cargo test --workspace --locked From 5e593a333f677f35a7df619e3cd51d170c92bb49 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Thu, 26 Feb 2026 22:50:26 +0400 Subject: [PATCH 31/31] Remove uv installation --- .github/actions/tests/action.yml | 3 --- 1 file changed, 3 deletions(-) diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index fb9c917..5fd6f03 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -13,9 +13,6 @@ runs: with: python-version: ${{ inputs.python-version }} - - name: Install uv - uses: astral-sh/setup-uv@v7 - - name: Install dependencies shell: bash -euxo pipefail {0} run: pip install -e .[tests]