herocode/horus
14
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

move repos into monorepo

Browse Source
This commit is contained in:
Timur Gordon
2025-11-13 20:44:00 +01:00
commit 4b23e5eb7f
204 changed files with 33737 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
lib/runner/Cargo.toml Normal file
View File

@@ -0,0 +1,42 @@
[package]
name = "hero-runner"
version.workspace = true
edition.workspace = true
description = "Hero Runner library for executing jobs"
license = "MIT OR Apache-2.0"
[lib]
name = "hero_runner"
path = "lib.rs"
[dependencies]
# Core dependencies
anyhow.workspace = true
redis.workspace = true
serde.workspace = true
serde_json.workspace = true
tokio.workspace = true
log.workspace = true
env_logger.workspace = true
uuid.workspace = true
chrono.workspace = true
toml.workspace = true
thiserror.workspace = true
async-trait.workspace = true
# Crypto dependencies
secp256k1.workspace = true
sha2.workspace = true
hex.workspace = true
# Rhai scripting
rhai = { version = "1.21.0", features = ["std", "sync", "decimal", "internals", "serde"] }
# Hero dependencies
hero-job = { path = "../models/job" }
hero-job-client = { path = "../clients/job" }
hero_logger = { git = "https://git.ourworld.tf/herocode/baobab.git", branch = "logger" }
# Tracing
tracing = "0.1.41"
rand = "0.8"

270
lib/runner/async_runner.rs Normal file
View File

@@ -0,0 +1,270 @@
use crate::Job;
use log::{debug, error, info};
use rhai::{Engine, packages::Package};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{mpsc, Mutex};
use tokio::task::JoinHandle;
use crate::runner_trait::Runner;
/// Represents a running job with its handle and metadata
struct RunningJob {
job_id: String,
handle: JoinHandle<Result<String, Box<dyn std::error::Error + Send + Sync>>>,
started_at: std::time::Instant,
}
/// Builder for AsyncRunner
#[derive(Default)]
pub struct AsyncRunnerBuilder {
runner_id: Option<String>,
db_path: Option<String>,
redis_url: Option<String>,
default_timeout: Option<Duration>,
engine: Option<Arc<dyn Fn() -> Engine + Send + Sync>>,
}
impl AsyncRunnerBuilder {
pub fn new() -> Self {
Self::default()
}
pub fn runner_id<S: Into<String>>(mut self, runner_id: S) -> Self {
self.runner_id = Some(runner_id.into());
self
}
pub fn db_path<S: Into<String>>(mut self, db_path: S) -> Self {
self.db_path = Some(db_path.into());
self
}
pub fn redis_url<S: Into<String>>(mut self, redis_url: S) -> Self {
self.redis_url = Some(redis_url.into());
self
}
pub fn default_timeout(mut self, timeout: Duration) -> Self {
self.default_timeout = Some(timeout);
self
}
pub fn engine_factory<F>(mut self, factory: F) -> Self
where
F: Fn() -> Engine + Send + Sync + 'static,
{
self.engine = Some(Arc::new(factory));
self
}
pub fn build(self) -> Result<AsyncRunner, String> {
Ok(AsyncRunner {
runner_id: self.runner_id.ok_or("runner_id is required")?,
db_path: self.db_path.ok_or("db_path is required")?,
redis_url: self.redis_url.ok_or("redis_url is required")?,
default_timeout: self.default_timeout.unwrap_or(Duration::from_secs(300)),
engine_factory: self.engine.ok_or("engine factory is required")?,
running_jobs: Arc::new(Mutex::new(HashMap::new())),
})
}
}
/// Asynchronous runner that processes jobs concurrently
pub struct AsyncRunner {
pub runner_id: String,
pub db_path: String,
pub redis_url: String,
pub default_timeout: Duration,
pub engine_factory: Arc<dyn Fn() -> Engine + Send + Sync>,
running_jobs: Arc<Mutex<HashMap<String, RunningJob>>>,
}
impl AsyncRunner {
/// Create a new AsyncRunnerBuilder
pub fn builder() -> AsyncRunnerBuilder {
AsyncRunnerBuilder::new()
}
/// Add a running job to the tracking map
async fn add_running_job(&self, job_id: String, handle: JoinHandle<Result<String, Box<dyn std::error::Error + Send + Sync>>>) {
let running_job = RunningJob {
job_id: job_id.clone(),
handle,
started_at: std::time::Instant::now(),
};
let mut jobs = self.running_jobs.lock().await;
jobs.insert(job_id.clone(), running_job);
debug!("Async Runner: Added running job '{}'. Total running: {}",
job_id, jobs.len());
}
/// Remove a completed job from the tracking map
async fn remove_running_job(&self, job_id: &str) {
let mut jobs = self.running_jobs.lock().await;
if let Some(job) = jobs.remove(job_id) {
let duration = job.started_at.elapsed();
debug!("Async Runner: Removed completed job '{}' after {:?}. Remaining: {}",
job_id, duration, jobs.len());
}
}
/// Get the count of currently running jobs
pub async fn running_job_count(&self) -> usize {
let jobs = self.running_jobs.lock().await;
jobs.len()
}
/// Cleanup any finished jobs from the running jobs map
async fn cleanup_finished_jobs(&self) {
let mut jobs = self.running_jobs.lock().await;
let mut to_remove = Vec::new();
for (job_id, running_job) in jobs.iter() {
if running_job.handle.is_finished() {
to_remove.push(job_id.clone());
}
}
for job_id in to_remove {
if let Some(job) = jobs.remove(&job_id) {
let duration = job.started_at.elapsed();
debug!("Async Runner: Cleaned up finished job '{}' after {:?}",
job_id, duration);
}
}
}
}
impl Runner for AsyncRunner {
fn process_job(&self, job: Job) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
let job_id = job.id.clone();
let runner_id = &self.runner_id;
// Determine timeout (use job-specific timeout if available, otherwise default)
let job_timeout = if job.timeout > 0 {
Duration::from_secs(job.timeout)
} else {
self.default_timeout
};
info!("Async Runner '{}', Job {}: Spawning job execution task with timeout {:?}",
runner_id, job_id, job_timeout);
// Clone necessary data for the spawned task
let job_id_clone = job_id.clone();
let runner_id_clone = runner_id.clone();
let runner_id_debug = runner_id.clone();
let job_id_debug = job_id.clone();
let _redis_url_clone = self.redis_url.clone();
let running_jobs_clone = Arc::clone(&self.running_jobs);
let engine_factory = Arc::clone(&self.engine_factory);
let db_path_clone = self.db_path.clone();
// Spawn the job execution task
let job_handle = tokio::spawn(async move {
// Create a new engine instance (cheap with factory pattern)
let mut engine = engine_factory();
let mut db_config = rhai::Map::new();
db_config.insert("DB_PATH".into(), db_path_clone.into());
db_config.insert("CALLER_ID".into(), job.caller_id.clone().into());
db_config.insert("CONTEXT_ID".into(), job.context_id.clone().into());
engine.set_default_tag(rhai::Dynamic::from(db_config));
// Execute the Rhai script
let result = match engine.eval::<rhai::Dynamic>(&job.payload) {
Ok(result) => {
let result_str = if result.is::<String>() {
result.into_string().unwrap()
} else {
result.to_string()
};
info!("Async Runner '{}', Job {}: Script executed successfully. Result: {}",
runner_id_clone, job_id_clone, result_str);
Ok(result_str)
}
Err(e) => {
let error_msg = format!("Script execution error: {}", e);
error!("Async Runner '{}', Job {}: {}", runner_id_clone, job_id_clone, error_msg);
Err(Box::new(e) as Box<dyn std::error::Error + Send + Sync>)
}
};
// Remove this job from the running jobs map when it completes
let mut jobs = running_jobs_clone.lock().await;
if let Some(running_job) = jobs.remove(&job_id_clone) {
let duration = running_job.started_at.elapsed();
debug!("Async Runner '{}': Removed completed job '{}' after {:?}",
runner_id_debug, job_id_debug, duration);
}
result
});
// Add the job to the running jobs map
let running_job = RunningJob {
job_id: job_id.clone(),
handle: job_handle,
started_at: std::time::Instant::now(),
};
let running_jobs_clone = Arc::clone(&self.running_jobs);
let job_id_for_map = job_id.clone();
tokio::spawn(async move {
let mut jobs = running_jobs_clone.lock().await;
jobs.insert(job_id_for_map, running_job);
debug!("Async Runner: Added running job '{}'. Total running: {}",
job_id, jobs.len());
});
// For async runners, we return immediately with a placeholder
// The actual result will be handled by the spawned task
Ok("Job spawned for async processing".to_string())
}
fn runner_type(&self) -> &'static str {
"Async"
}
fn runner_id(&self) -> &str {
&self.runner_id
}
fn redis_url(&self) -> &str {
&self.redis_url
}
}
/// Convenience function to spawn an asynchronous runner using the trait interface
///
/// This function provides a clean interface for the new async runner implementation
/// with timeout support.
pub fn spawn_async_runner<F>(
runner_id: String,
db_path: String,
redis_url: String,
shutdown_rx: mpsc::Receiver<()>,
default_timeout: std::time::Duration,
engine_factory: F,
) -> JoinHandle<Result<(), Box<dyn std::error::Error + Send + Sync>>>
where
F: Fn() -> Engine + Send + Sync + 'static,
{
use std::sync::Arc;
let runner = Arc::new(
AsyncRunner::builder()
.runner_id(runner_id)
.db_path(db_path)
.redis_url(redis_url)
.default_timeout(default_timeout)
.engine_factory(engine_factory)
.build()
.expect("Failed to build AsyncRunner")
);
crate::runner_trait::spawn_runner(runner, shutdown_rx)
}

80
lib/runner/lib.rs Normal file
View File

@@ -0,0 +1,80 @@
// Core modules
pub mod async_runner;
pub mod sync_runner;
pub mod runner_trait;
pub mod script_mode;
// Public exports for convenience
pub use runner_trait::{Runner, RunnerConfig, spawn_runner};
pub use async_runner::{AsyncRunner, spawn_async_runner};
pub use sync_runner::{SyncRunner, SyncRunnerConfig, spawn_sync_runner};
// Re-export job types from hero-job crate
pub use hero_job::{Job, JobStatus, JobError, JobBuilder, JobSignature};
// Re-export job client
pub use hero_job_client::{Client, ClientBuilder};
pub use redis::AsyncCommands;
use log::{error, info};
const BLPOP_TIMEOUT_SECONDS: usize = 5;
/// Initialize Redis connection for the runner
pub async fn initialize_redis_connection(
runner_id: &str,
redis_url: &str,
) -> Result<redis::aio::MultiplexedConnection, Box<dyn std::error::Error + Send + Sync>> {
let redis_client = redis::Client::open(redis_url)
.map_err(|e| {
error!("Runner for Runner ID '{}': Failed to open Redis client: {}", runner_id, e);
e
})?;
let redis_conn = redis_client.get_multiplexed_async_connection().await
.map_err(|e| {
error!("Runner for Runner ID '{}': Failed to get Redis connection: {}", runner_id, e);
e
})?;
info!("Runner for Runner ID '{}' successfully connected to Redis.", runner_id);
Ok(redis_conn)
}
// /// Load job from Redis using the supervisor's Job API
// pub async fn load_job_from_redis(
// redis_conn: &mut redis::aio::MultiplexedConnection,
// job_id: &str,
// runner_id: &str,
// ) -> Result<Job, JobError> {
// debug!("Runner '{}', Job {}: Loading job from Redis", runner_id, job_id);
// // Load job data from Redis hash
// let job_data: std::collections::HashMap<String, String> = redis_conn.hgetall(&client.job_key(job_id)).await
// .map_err(JobError::Redis)?;
// if job_data.is_empty() {
// return Err(JobError::NotFound(job_id.to_string()));
// }
// // Parse job from hash data using the supervisor's Job struct
// let job = Job {
// id: job_id.to_string(),
// caller_id: job_data.get("caller_id").unwrap_or(&"".to_string()).clone(),
// context_id: job_data.get("context_id").unwrap_or(&"".to_string()).clone(),
// payload: job_data.get("payload").unwrap_or(&"".to_string()).clone(),
// runner: job_data.get("runner").unwrap_or(&"default".to_string()).clone(),
// executor: job_data.get("executor").unwrap_or(&"rhai".to_string()).clone(),
// timeout: job_data.get("timeout").and_then(|s| s.parse().ok()).unwrap_or(300),
// env_vars: serde_json::from_str(job_data.get("env_vars").unwrap_or(&"{}".to_string()))
// .map_err(JobError::Serialization)?,
// created_at: job_data.get("created_at")
// .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
// .map(|dt| dt.with_timezone(&chrono::Utc))
// .unwrap_or_else(chrono::Utc::now),
// updated_at: job_data.get("updated_at")
// .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
// .map(|dt| dt.with_timezone(&chrono::Utc))
// .unwrap_or_else(chrono::Utc::now),
// };
// Ok(job)
// }

272
lib/runner/runner_trait.rs Normal file
View File

@@ -0,0 +1,272 @@
//! # Runner Trait Abstraction
//!
//! This module provides a trait-based abstraction for Rhai runners that eliminates
//! code duplication between synchronous and asynchronous runner implementations.
//!
//! The `Runner` trait defines the common interface and behavior, while specific
//! implementations handle job processing differently (sync vs async).
//!
//! ## Architecture
//!
//! ```text
//! ┌─────────────────┐ ┌─────────────────┐
//! │ SyncRunner │ │ AsyncRunner │
//! │ │ │ │
//! │ process_job() │ │ process_job() │
//! │ (sequential) │ │ (concurrent) │
//! └─────────────────┘ └─────────────────┘
//! │ │
//! └───────┬───────────────┘
//! │
//! ┌───────▼───────┐
//! │ Runner Trait │
//! │ │
//! │ spawn() │
//! │ config │
//! │ common loop │
//! └───────────────┘
//! ```
use crate::{Job, JobStatus, Client};
use log::{debug, error, info};
use redis::AsyncCommands;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use crate::{initialize_redis_connection, BLPOP_TIMEOUT_SECONDS};
/// Configuration for runner instances
#[derive(Debug, Clone)]
pub struct RunnerConfig {
pub runner_id: String,
pub db_path: String,
pub redis_url: String,
pub default_timeout: Option<Duration>, // Only used by async runners
}
impl RunnerConfig {
/// Create a new runner configuration
pub fn new(
runner_id: String,
db_path: String,
redis_url: String,
) -> Self {
Self {
runner_id,
db_path,
redis_url,
default_timeout: None,
}
}
/// Set default timeout for async runners
pub fn with_default_timeout(mut self, timeout: Duration) -> Self {
self.default_timeout = Some(timeout);
self
}
}
/// Trait defining the common interface for Rhai runners
///
/// This trait abstracts the common functionality between synchronous and
/// asynchronous runners, allowing them to share the same spawn logic and
/// Redis polling loop while implementing different job processing strategies.
pub trait Runner: Send + Sync + 'static {
/// Process a single job
///
/// This is the core method that differentiates runner implementations:
/// - Sync runners process jobs sequentially, one at a time
/// - Async runners spawn concurrent tasks for each job
///
/// # Arguments
///
/// * `job` - The job to process
///
/// Note: The engine is now owned by the runner implementation as a field
/// For sync runners, this should be a blocking operation
/// For async runners, this can spawn tasks and return immediately
fn process_job(&self, job: Job) -> Result<String, Box<dyn std::error::Error + Send + Sync>>;
/// Get the runner type name for logging
fn runner_type(&self) -> &'static str;
/// Get runner ID for this runner instance
fn runner_id(&self) -> &str;
/// Get Redis URL for this runner instance
fn redis_url(&self) -> &str;
/// Spawn the runner
///
/// This method provides the common runner loop implementation that both
/// sync and async runners can use. It handles:
/// - Redis connection setup
/// - Job polling from Redis queue
/// - Shutdown signal handling
/// - Delegating job processing to the implementation
///
/// Note: The engine is now owned by the runner implementation as a field
fn spawn(
self: Arc<Self>,
mut shutdown_rx: mpsc::Receiver<()>,
) -> JoinHandle<Result<(), Box<dyn std::error::Error + Send + Sync>>> {
tokio::spawn(async move {
let runner_id = self.runner_id();
let redis_url = self.redis_url();
// Create client to get the proper queue key
let client = Client::builder()
.redis_url(redis_url)
.build()
.await
.map_err(|e| format!("Failed to create client: {}", e))?;
let queue_key = client.runner_key(runner_id);
info!(
"{} Runner '{}' starting. Connecting to Redis at {}. Listening on queue: {}",
self.runner_type(),
runner_id,
redis_url,
queue_key
);
let mut redis_conn = initialize_redis_connection(runner_id, redis_url).await?;
loop {
let blpop_keys = vec![queue_key.clone()];
tokio::select! {
// Listen for shutdown signal
_ = shutdown_rx.recv() => {
info!("{} Runner '{}': Shutdown signal received. Terminating loop.",
self.runner_type(), runner_id);
break;
}
// Listen for tasks from Redis
blpop_result = redis_conn.blpop(&blpop_keys, BLPOP_TIMEOUT_SECONDS as f64) => {
debug!("{} Runner '{}': Attempting BLPOP on queue: {}",
self.runner_type(), runner_id, queue_key);
let response: Option<(String, String)> = match blpop_result {
Ok(resp) => resp,
Err(e) => {
error!("{} Runner '{}': Redis BLPOP error on queue {}: {}. Runner for this circle might stop.",
self.runner_type(), runner_id, queue_key, e);
return Err(Box::new(e) as Box<dyn std::error::Error + Send + Sync>);
}
};
if let Some((_queue_name_recv, job_id)) = response {
info!("{} Runner '{}' received job_id: {} from queue: {}",
self.runner_type(), runner_id, job_id, _queue_name_recv);
// Load the job from Redis
match client.load_job_from_redis(&job_id).await {
Ok(job) => {
// Check for ping job and handle it directly
if job.payload.trim() == "ping" {
info!("{} Runner '{}': Received ping job '{}', responding with pong",
self.runner_type(), runner_id, job_id);
// Update job status to started
if let Err(e) = client.set_job_status(&job_id, JobStatus::Started).await {
error!("{} Runner '{}': Failed to update ping job '{}' status to Started: {}",
self.runner_type(), runner_id, job_id, e);
}
// Set result to "pong" and mark as finished
if let Err(e) = client.set_result(&job_id, "pong").await {
error!("{} Runner '{}': Failed to set ping job '{}' result: {}",
self.runner_type(), runner_id, job_id, e);
}
if let Err(e) = client.set_job_status(&job_id, JobStatus::Finished).await {
error!("{} Runner '{}': Failed to update ping job '{}' status to Finished: {}",
self.runner_type(), runner_id, job_id, e);
}
info!("{} Runner '{}': Successfully responded to ping job '{}' with pong",
self.runner_type(), runner_id, job_id);
} else {
// Update job status to started
if let Err(e) = client.set_job_status(&job_id, JobStatus::Started).await {
error!("{} Runner '{}': Failed to update job '{}' status to Started: {}",
self.runner_type(), runner_id, job_id, e);
}
// Delegate job processing to the implementation
match self.process_job(job) {
Ok(result) => {
// Set result and mark as finished
if let Err(e) = client.set_result(&job_id, &result).await {
error!("{} Runner '{}': Failed to set job '{}' result: {}",
self.runner_type(), runner_id, job_id, e);
}
if let Err(e) = client.set_job_status(&job_id, JobStatus::Finished).await {
error!("{} Runner '{}': Failed to update job '{}' status to Finished: {}",
self.runner_type(), runner_id, job_id, e);
}
}
Err(e) => {
let error_str = format!("{:?}", e);
error!("{} Runner '{}': Job '{}' processing failed: {}",
self.runner_type(), runner_id, job_id, error_str);
// Set error and mark as error
if let Err(e) = client.set_error(&job_id, &error_str).await {
error!("{} Runner '{}': Failed to set job '{}' error: {}",
self.runner_type(), runner_id, job_id, e);
}
if let Err(e) = client.set_job_status(&job_id, JobStatus::Error).await {
error!("{} Runner '{}': Failed to update job '{}' status to Error: {}",
self.runner_type(), runner_id, job_id, e);
}
}
}
}
}
Err(e) => {
error!("{} Runner '{}': Failed to load job '{}': {}",
self.runner_type(), runner_id, job_id, e);
}
}
} else {
debug!("{} Runner '{}': BLPOP timed out on queue {}. No new tasks.",
self.runner_type(), runner_id, queue_key);
}
}
}
}
info!("{} Runner '{}' has shut down.", self.runner_type(), runner_id);
Ok(())
})
}
}
/// Convenience function to spawn a runner with the trait-based interface
///
/// This function provides a unified interface for spawning any runner implementation
/// that implements the Runner trait.
///
/// # Arguments
///
/// * `runner` - The runner implementation to spawn
/// * `shutdown_rx` - Channel receiver for shutdown signals
///
/// # Returns
///
/// Returns a `JoinHandle` that can be awaited to wait for runner shutdown.
pub fn spawn_runner<W: Runner>(
runner: Arc<W>,
shutdown_rx: mpsc::Receiver<()>,
) -> JoinHandle<Result<(), Box<dyn std::error::Error + Send + Sync>>> {
runner.spawn(shutdown_rx)
}

168
lib/runner/script_mode.rs Normal file
View File

@@ -0,0 +1,168 @@
use std::time::Duration;
use tokio::time::timeout;
use crate::{JobBuilder, JobStatus, Client};
use log::{info, error};
use tokio::sync::mpsc;
use std::sync::Arc;
use crate::async_runner::AsyncRunner;
use crate::runner_trait::{Runner, RunnerConfig};
/// Execute a script in single-job mode
/// Creates a job, submits it, waits for completion, and returns the result
pub async fn execute_script_mode<F>(
script_content: &str,
runner_id: &str,
redis_url: String,
job_timeout: Duration,
engine_factory: F,
) -> Result<String, Box<dyn std::error::Error + Send + Sync>>
where
F: Fn() -> rhai::Engine + Send + Sync + 'static,
{
info!("Executing script in single-job mode");
// Create job client
let job_client = Client::builder()
.redis_url(&redis_url)
.build()
.await?;
// Create the job using JobBuilder
let job = JobBuilder::new()
.caller_id("script_mode")
.payload(script_content)
.runner(runner_id)
.executor("rhai")
.timeout(job_timeout.as_secs())
.build()?;
let job_id = job.id.clone();
info!("Created job with ID: {}", job_id);
// Submit the job
job_client.store_job_in_redis(&job).await?;
info!("Job stored in Redis");
// Dispatch the job to the runner's queue
job_client.job_run(&job_id, runner_id).await?;
info!("Job dispatched to runner queue: {}", runner_id);
// Create and spawn a temporary runner to process the job
let (shutdown_tx, shutdown_rx) = mpsc::channel::<()>(1);
let config = RunnerConfig {
runner_id: runner_id.to_string(),
db_path: "/tmp".to_string(), // Temporary path for script mode
redis_url: redis_url.clone(),
default_timeout: Some(job_timeout),
};
let runner = Arc::new(
AsyncRunner::builder()
.runner_id(&config.runner_id)
.db_path(&config.db_path)
.redis_url(&config.redis_url)
.default_timeout(config.default_timeout.unwrap_or(job_timeout))
.engine_factory(engine_factory)
.build()
.map_err(|e| format!("Failed to build runner: {}", e))?
);
let runner_handle = runner.spawn(shutdown_rx);
info!("Temporary runner spawned for job processing");
// Wait for job completion with timeout
let result = timeout(job_timeout, wait_for_job_completion(&job_client, &job_id)).await;
// Shutdown the temporary runner
let _ = shutdown_tx.send(()).await;
let _ = runner_handle.await;
match result {
Ok(job_result) => {
match job_result {
Ok(job_status) => {
match job_status {
JobStatus::Finished => {
info!("Job completed successfully");
// Get the job result from Redis
match job_client.get_result(&job_id).await {
Ok(Some(result)) => Ok(result),
Ok(None) => Ok("Job completed with no result".to_string()),
Err(e) => {
error!("Failed to get job result: {}", e);
Ok("Job completed but result unavailable".to_string())
}
}
}
JobStatus::Error => {
// Get the job error from Redis - for now just return a generic error
error!("Job failed with status: Error");
return Err("Job execution failed".into());
/*match job_client.get_job_error(&job_id).await {
Ok(Some(error_msg)) => {
error!("Job failed: {}", error_msg);
Err(format!("Job failed: {}", error_msg).into())
}
Ok(None) => {
error!("Job failed with no error message");
Err("Job failed with no error message".into())
}
Err(e) => {
error!("Failed to get job error: {}", e);
Err("Job failed but error details unavailable".into())
}
}*/
}
_ => {
error!("Job ended in unexpected status: {:?}", job_status);
Err(format!("Job ended in unexpected status: {:?}", job_status).into())
}
}
}
Err(e) => {
error!("Error waiting for job completion: {}", e);
Err(e)
}
}
}
Err(_) => {
error!("Job execution timed out after {:?}", job_timeout);
// Try to cancel the job
let _ = job_client.set_job_status(&job_id, JobStatus::Error).await;
Err("Job execution timed out".into())
}
}
}
/// Wait for job completion by polling Redis
async fn wait_for_job_completion(
job_client: &Client,
job_id: &str,
) -> Result<JobStatus, Box<dyn std::error::Error + Send + Sync>> {
let poll_interval = Duration::from_millis(500);
loop {
match job_client.get_status(job_id).await {
Ok(status) => {
match status {
JobStatus::Finished | JobStatus::Error => {
return Ok(status);
}
JobStatus::Created | JobStatus::Dispatched | JobStatus::WaitingForPrerequisites | JobStatus::Started => {
// Continue polling
tokio::time::sleep(poll_interval).await;
}
JobStatus::Stopping => {
// Job is being stopped, wait a bit more
tokio::time::sleep(poll_interval).await;
}
}
}
Err(e) => {
error!("Error polling job status: {}", e);
tokio::time::sleep(poll_interval).await;
}
}
}
}

175
lib/runner/sync_runner.rs Normal file
View File

@@ -0,0 +1,175 @@
use crate::Job;
use crate::runner_trait::Runner;
use log::{debug, error, info};
use rhai::{Engine, Dynamic};
use std::sync::Arc;
use tracing::subscriber::with_default;
/// Configuration for sync runner instances
#[derive(Debug, Clone)]
pub struct SyncRunnerConfig {
pub runner_id: String,
pub redis_url: String,
}
/// Synchronous runner that processes jobs sequentially
pub struct SyncRunner {
pub config: SyncRunnerConfig,
pub engine_factory: Arc<dyn Fn() -> Engine + Send + Sync>,
}
impl SyncRunner {
/// Create a new SyncRunner with the provided engine factory
pub fn new<F>(config: SyncRunnerConfig, engine_factory: F) -> Self
where
F: Fn() -> Engine + Send + Sync + 'static,
{
Self {
config,
engine_factory: Arc::new(engine_factory),
}
}
/// Execute a job with the given engine, setting proper job context
///
/// This function sets up the engine with job context (DB_PATH, CALLER_ID, CONTEXT_ID)
/// and evaluates the script. It returns the result or error.
fn execute_job_with_engine(
engine: &mut Engine,
job: &Job,
) -> Result<Dynamic, Box<rhai::EvalAltResult>> {
// Set up job context in the engine
let mut db_config = rhai::Map::new();
db_config.insert("CALLER_ID".into(), job.caller_id.clone().into());
db_config.insert("CONTEXT_ID".into(), job.context_id.clone().into());
// Extract signatories from job signatures, or fall back to env_vars
let signatories: Vec<Dynamic> = if !job.signatures.is_empty() {
// Use signatures from the job
job.signatures.iter()
.map(|sig| Dynamic::from(sig.public_key.clone()))
.collect()
} else {
Vec::new()
};
db_config.insert("SIGNATORIES".into(), Dynamic::from(signatories));
engine.set_default_tag(Dynamic::from(db_config));
debug!("Sync Runner for Context ID '{}': Evaluating script with Rhai engine (job context set).", job.context_id);
// Execute the script with the configured engine
engine.eval::<Dynamic>(&job.payload)
}
}
impl Runner for SyncRunner {
fn process_job(&self, job: Job) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
let job_id = &job.id;
let runner_id = &self.config.runner_id;
debug!("Sync Runner '{}', Job {}: Processing started.", runner_id, job_id);
info!("Sync Runner '{}' processing job_id: {}. Script: {:.50}...", job.context_id, job_id, job.payload);
// Determine logs directory (default to ~/hero/logs)
let logs_root = if let Some(home) = std::env::var_os("HOME") {
std::path::PathBuf::from(home).join("hero").join("logs")
} else {
std::path::PathBuf::from("logs")
};
// Create job-specific logger
let job_logger_result = hero_logger::create_job_logger_with_guard(
&logs_root,
runner_id, // Use runner_id as the actor_type
job_id,
);
// Verify signatures before executing (if any)
if let Err(e) = job.verify_signatures() {
error!("Job {} signature verification failed: {}", job_id, e);
return Err(Box::new(e));
}
// Execute job within logging context
let result = match job_logger_result {
Ok((job_logger, _guard)) => {
// Execute ALL job processing within logging context
with_default(job_logger, || {
tracing::info!("Job {} started", job_id);
// Create a new engine instance and configure Rhai logging
let mut engine = (self.engine_factory)();
// Reconfigure Rhai logging for this specific job context
// This ensures print() and debug() calls go to the job logger
hero_logger::rhai_integration::configure_rhai_logging(&mut engine, runner_id);
// Execute the script
let script_result = Self::execute_job_with_engine(&mut engine, &job);
tracing::info!("Job {} completed", job_id);
script_result
})
}
Err(e) => {
error!("Failed to create job logger for job {}: {}", job_id, e);
// Fallback: execute without job-specific logging
let mut engine = (self.engine_factory)();
Self::execute_job_with_engine(&mut engine, &job)
}
};
// Process result
match result {
Ok(result) => {
let output_str = if result.is::<String>() {
result.into_string().unwrap()
} else {
result.to_string()
};
info!("Sync Runner for Context ID '{}' job {} completed. Output: {}", job.context_id, job.id, output_str);
Ok(output_str)
}
Err(e) => {
let error_str = format!("{:?}", *e);
error!("Sync Runner for Context ID '{}' job {} script evaluation failed. Error: {}", job.context_id, job.id, error_str);
Err(Box::new(e) as Box<dyn std::error::Error + Send + Sync>)
}
}
}
fn runner_type(&self) -> &'static str {
"Sync"
}
fn runner_id(&self) -> &str {
&self.config.runner_id
}
fn redis_url(&self) -> &str {
&self.config.redis_url
}
}
/// Convenience function to spawn a synchronous runner using the trait interface
pub fn spawn_sync_runner<F>(
runner_id: String,
redis_url: String,
shutdown_rx: tokio::sync::mpsc::Receiver<()>,
engine_factory: F,
) -> tokio::task::JoinHandle<Result<(), Box<dyn std::error::Error + Send + Sync>>>
where
F: Fn() -> Engine + Send + Sync + 'static,
{
let config = SyncRunnerConfig {
runner_id,
redis_url,
};
let runner = Arc::new(SyncRunner::new(config, engine_factory));
crate::runner_trait::spawn_runner(runner, shutdown_rx)
}