diff --git a/.github/actions/tests/action.yml b/.github/actions/tests/action.yml index 6e1d14e..5fd6f03 100644 --- a/.github/actions/tests/action.yml +++ b/.github/actions/tests/action.yml @@ -13,12 +13,9 @@ 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: uv sync --extra tests + run: pip install -e .[tests] - name: Set up Rust uses: dtolnay/rust-toolchain@stable @@ -29,5 +26,5 @@ runs: - name: Run Tests shell: bash -euxo pipefail {0} run: | - uv run pytest tests/ + pytest tests/ cargo test --workspace --locked 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/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..0f4da39 --- /dev/null +++ b/crates/fluxqueue-worker/scripts/get_registry.py @@ -0,0 +1,59 @@ +import importlib +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): + 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) + + 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): + 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") + + 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..0907f42 100644 --- a/crates/fluxqueue-worker/src/task.rs +++ b/crates/fluxqueue-worker/src/task.rs @@ -1,62 +1,140 @@ use anyhow::{Context, Result, anyhow}; +use fluxqueue_common::Task; 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::{Arc, RwLock}; +use std::time::Instant; 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>>>>, + tasks: Registry>, + contexts: Registry>>, + context_objects: Registry>>, } 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 (tasks, contexts) = get_registry(module_path, queue_name)?; + + Ok(Self { + tasks: Arc::new(RwLock::new(tasks)), + contexts: Arc::new(RwLock::new(contexts)), + context_objects: Arc::new(RwLock::new(HashMap::new())), + }) } - 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>> { + 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 { + Ok(None) + } } } struct TaskRequest { - func: Arc>, + task_registry: Arc, + executor_id: Arc, + task_data: Arc, task_name: Arc, - raw_args: Arc>, - raw_kwargs: Arc>, + task: Arc, resp_tx: oneshot::Sender>, } 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.func, req.task_name, req.raw_args, req.raw_kwargs) - .await - .map_err(|e| PyRuntimeError::new_err(e.to_string()))?; + run_task( + req.task_registry, + req.executor_id, + req.task_data, + req.task_name, + req.task, + ) + .await + .map_err(|e| PyRuntimeError::new_err(e.to_string()))?; let _ = req.resp_tx.send(Ok(())); } @@ -69,24 +147,25 @@ impl PythonDispatcher { }); }); - Ok(Self { tx }) + Ok(Self { tx, task_registry }) } pub async fn execute( &self, - func: Arc>, + 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(); self.tx .send(TaskRequest { - func, + task_registry: self.task_registry.clone(), + executor_id, + task_data, task_name, - raw_args, - raw_kwargs, + task, resp_tx, }) .await @@ -97,26 +176,39 @@ impl PythonDispatcher { } } +struct MaybeCoro { + func: Arc>, + args: Py, + kwargs: Py, + context: Option>>, +} + async fn run_task( - task_function: Arc>, + task_registry: Arc, + executor_id: Arc, + task_data: Arc, task_name: Arc, - raw_args: Arc>, - raw_kwargs: Arc>, + task: Arc, ) -> Result<()> { - let task_args: Value = from_slice(&raw_args).context(format!( + let logger = Logger::new(format!("EXECUTOR {}", &executor_id)); + let duration_start = Instant::now(); + + 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 maybe_coro = Python::attach(|py| -> Result>> { + 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() @@ -124,30 +216,322 @@ async fn run_task( anyhow::bail!("Args must be an array/tuple, found {}", py_args.get_type()); }; + 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])?; + + 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 - .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)) + Ok(Some(MaybeCoro { + func: task_data.func.clone(), + args: args_tuple.unbind(), + kwargs: kwargs_dict.unbind(), + 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", + &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)))?; + if let Some(maybe_coro) = maybe_coro { + let fut = Python::attach(|py| { + 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 = context.call_method1( + py, + "_run_async_task", + ( + maybe_coro.func.as_any(), + task_metadata, + maybe_coro.args, + Some(maybe_coro.kwargs), + ), + )?; + into_future(result.into_bound(py)) + } else { + 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?; + + let duration_end = duration_start.elapsed(); + logger.info(format_args!( + "Task '{}' successfully finished in {}ms", + &task_name, + duration_end.as_millis() + )); } Ok(()) } + +type TasksAndContexts = ( + HashMap, Arc>, + HashMap, Arc>>, +); + +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, Arc> = registry + .get_item("tasks")? + .expect("tasks missing") + .cast::() + .map_err(|e| anyhow!("tasks is not a dict: {}", e))? + .iter() + .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::() + .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((Arc::new(name), Arc::new(func))) + }) + .collect(); + + Ok((tasks, contexts)) + })?; + + 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() { + 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(".")) +} + +#[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(&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(&Arc::new("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(&Arc::new("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()); + } +} diff --git a/crates/fluxqueue-worker/src/worker.rs b/crates/fluxqueue-worker/src/worker.rs index 8e2ba8c..b18c0aa 100644 --- a/crates/fluxqueue-worker/src/worker.rs +++ b/crates/fluxqueue-worker/src/worker.rs @@ -1,17 +1,13 @@ -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 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::{PythonDispatcher, TaskData, TaskRegistry}; use fluxqueue_common::{Task, deserialize_raw_task_data}; pub async fn run_worker( @@ -28,26 +24,25 @@ 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).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()?; 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 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)); @@ -64,7 +59,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) @@ -90,9 +85,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,8 +131,8 @@ 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, @@ -171,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", @@ -178,7 +174,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) @@ -188,8 +184,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.python_dispatcher.clone(), Arc::new(task), task_data.clone()).await; match task_result { Ok(_) => { @@ -198,25 +193,12 @@ 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 { - 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)); } } } @@ -235,8 +217,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 +230,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 +240,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 +295,25 @@ async fn janitor_loop( } async fn run_task( + executor_id: Arc, python_dispatcher: Arc, - task: &Task, - task_function: Arc>, + task: Arc, + task_data: Arc, ) -> Result<()> { - let task_name = Arc::new(task.name.clone()); - let raw_args = Arc::new(task.args.clone()); - let raw_kwargs = Arc::new(task.kwargs.clone()); + let task_name = Arc::new(format!("{}:{}", &task.name, &task.id)); python_dispatcher - .execute(task_function, task_name, raw_args, raw_kwargs) + .execute(executor_id, task_data, task_name, task) .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,93 +337,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"); - - 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_task_functions(&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_task_functions(&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_task_functions(&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_task_functions(&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_task_functions("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 module_path_str = get_test_module_path("test_tasks_module.py"); - let task_functions = get_task_functions(&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())?); - for (name, task_obj) in task_functions { - task_registry.insert(name, task_obj)?; - } - - 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 { @@ -540,7 +363,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(), + Arc::new(task), + task_func, + ) + .await; assert!(!result.is_err()); } @@ -549,16 +378,11 @@ 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 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 = 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 { @@ -572,7 +396,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(), + Arc::new(task), + task_func, + ) + .await; assert!(!result.is_err()); } 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/python/fluxqueue/__init__.py b/python/fluxqueue/__init__.py index 60393a9..b0256af 100644 --- a/python/fluxqueue/__init__.py +++ b/python/fluxqueue/__init__.py @@ -1,3 +1,3 @@ -__all__ = ["FluxQueue"] - -from .client import FluxQueue +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 new file mode 100644 index 0000000..7f9d9b0 --- /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}") + + 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 inspect.iscoroutinefunction(func): + + @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..23d6f16 --- /dev/null +++ b/python/fluxqueue/context.py @@ -0,0 +1,125 @@ +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") + + +class Context: + __fluxqueue_context__: str | None = None + + def __init__(self) -> None: + self._thread_local = threading.local() + self._metadata_var: ContextVar[TaskMetadata] = ContextVar("task_metadata") + + @property + def thread_storage(self) -> dict[str, Any]: + """ + Retrieves the thread-persistent storage dictionary. + + 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 + + @property + def metadata(self) -> TaskMetadata: + """ + Returns metadata isolated to the current task. + + 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() + + 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__: + cls.__fluxqueue_context__ = cls.__name__ + + +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]) + + wrapper = _task_decorator( + cast(Any, func), name=name, queue=queue, max_retries=max_retries, core=core + ) + + cast(Any, wrapper).__signature__ = new_sig + return wrapper 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.""" 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" "$@"