diff --git a/.github/workflows/publish-java.yml b/.github/workflows/publish-java.yml index 72717b32..735b3df6 100644 --- a/.github/workflows/publish-java.yml +++ b/.github/workflows/publish-java.yml @@ -95,7 +95,7 @@ jobs: uses: ./.github/actions/setup-rust - name: Build cdylib - run: cargo build -p taskito-java --release --features postgres,redis + run: cargo build -p taskito-java --release --features postgres,redis,workflows - name: Stage native binary under its classifier shell: bash diff --git a/Cargo.lock b/Cargo.lock index 563a3c3b..38dee34e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1516,6 +1516,7 @@ dependencies = [ "serde", "serde_json", "taskito-core", + "taskito-workflows", "tokio", "uuid", ] diff --git a/crates/taskito-java/Cargo.toml b/crates/taskito-java/Cargo.toml index 099f76b0..4c1c9883 100644 --- a/crates/taskito-java/Cargo.toml +++ b/crates/taskito-java/Cargo.toml @@ -11,11 +11,13 @@ crate-type = ["cdylib"] [features] default = [] -postgres = ["taskito-core/postgres"] -redis = ["taskito-core/redis"] +postgres = ["taskito-core/postgres", "taskito-workflows?/postgres"] +redis = ["taskito-core/redis", "taskito-workflows?/redis"] +workflows = ["dep:taskito-workflows"] [dependencies] taskito-core = { path = "../taskito-core" } +taskito-workflows = { path = "../taskito-workflows", optional = true } jni = "0.21" serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/taskito-java/src/backend.rs b/crates/taskito-java/src/backend.rs index 69688b0a..94e77a99 100644 --- a/crates/taskito-java/src/backend.rs +++ b/crates/taskito-java/src/backend.rs @@ -16,6 +16,48 @@ const DEFAULT_POSTGRES_SCHEMA: &str = "taskito"; pub struct QueueHandle { pub storage: StorageBackend, pub namespace: Option, + /// Workflow storage, built lazily on first workflow call (its constructor + /// runs the workflow-table migrations). Shares this queue's connection pool. + #[cfg(feature = "workflows")] + pub workflow_storage: std::sync::OnceLock, +} + +#[cfg(feature = "workflows")] +impl QueueHandle { + /// Return the workflow storage, initializing it (and its migrations) once. + pub fn workflow_store( + &self, + ) -> Result { + if let Some(wf) = self.workflow_storage.get() { + return Ok(wf.clone()); + } + let wf = build_workflow_storage(&self.storage)?; + // A racing thread's handle wraps the same pool, so either is fine. + let _ = self.workflow_storage.set(wf.clone()); + Ok(wf) + } +} + +/// Construct workflow storage matching this queue's core backend. +#[cfg(feature = "workflows")] +fn build_workflow_storage( + storage: &StorageBackend, +) -> Result { + use taskito_workflows::{WorkflowSqliteStorage, WorkflowStorageBackend}; + let wf = match storage { + StorageBackend::Sqlite(s) => { + WorkflowStorageBackend::Sqlite(WorkflowSqliteStorage::new(s.clone())?) + } + #[cfg(feature = "postgres")] + StorageBackend::Postgres(s) => WorkflowStorageBackend::Postgres( + taskito_workflows::WorkflowPostgresStorage::new(s.clone())?, + ), + #[cfg(feature = "redis")] + StorageBackend::Redis(s) => { + WorkflowStorageBackend::Redis(taskito_workflows::WorkflowRedisStorage::new(s.clone())?) + } + }; + Ok(wf) } /// Error for a backend that is unknown or whose cargo feature is not compiled in. @@ -66,5 +108,7 @@ pub fn open(options: OpenOptions) -> Result { Ok(QueueHandle { storage, namespace: options.namespace, + #[cfg(feature = "workflows")] + workflow_storage: std::sync::OnceLock::new(), }) } diff --git a/crates/taskito-java/src/error.rs b/crates/taskito-java/src/error.rs index 930f203a..715b950f 100644 --- a/crates/taskito-java/src/error.rs +++ b/crates/taskito-java/src/error.rs @@ -34,3 +34,11 @@ impl From for BindingError { Self::new(err.to_string()) } } + +/// Workflow errors (e.g. DAG validation) reach the boundary as `TaskitoException`. +#[cfg(feature = "workflows")] +impl From for BindingError { + fn from(err: taskito_workflows::WorkflowError) -> Self { + Self::new(err.to_string()) + } +} diff --git a/crates/taskito-java/src/ffi.rs b/crates/taskito-java/src/ffi.rs index c366defd..47a82310 100644 --- a/crates/taskito-java/src/ffi.rs +++ b/crates/taskito-java/src/ffi.rs @@ -78,6 +78,28 @@ pub fn read_bytes_array( Ok(out) } +/// Read a Java `String[]` argument into owned strings; a null array is empty. +#[cfg_attr(not(feature = "workflows"), allow(dead_code))] +pub fn read_string_array( + env: &mut JNIEnv, + value: &JObjectArray, +) -> Result, BindingError> { + if value.is_null() { + return Ok(Vec::new()); + } + let len = env + .get_array_length(value) + .map_err(|e| BindingError::new(format!("invalid String[] argument: {e}")))?; + let mut out = Vec::with_capacity(len as usize); + for index in 0..len { + let element = env + .get_object_array_element(value, index) + .map_err(|e| BindingError::new(format!("invalid String[] element: {e}")))?; + out.push(read_string(env, &JString::from(element))?); + } + Ok(out) +} + /// Build a Java `String` return value from an owned Rust `String`. pub fn new_string(env: &mut JNIEnv, value: String) -> Result { env.new_string(value) diff --git a/crates/taskito-java/src/lib.rs b/crates/taskito-java/src/lib.rs index daec4563..4d0fda19 100644 --- a/crates/taskito-java/src/lib.rs +++ b/crates/taskito-java/src/lib.rs @@ -17,6 +17,8 @@ mod handle; mod jvm; mod queue; mod worker; +#[cfg(feature = "workflows")] +mod workflows; use std::ffi::c_void; diff --git a/crates/taskito-java/src/workflows/mod.rs b/crates/taskito-java/src/workflows/mod.rs new file mode 100644 index 00000000..a2a453da --- /dev/null +++ b/crates/taskito-java/src/workflows/mod.rs @@ -0,0 +1,540 @@ +//! JNI entry points backing `org.byteveda.taskito.internal.NativeWorkflows`. +//! +//! Compiled only with the `workflows` feature. Static DAG steps are pre-enqueued +//! with a `depends_on` chain so the core scheduler runs them in topological +//! order; the worker-side `WorkflowTracker` records each node's terminal outcome +//! (and cascades a fail-fast skip) via [`mark_node_result`]. Deferred steps +//! (fan-out, gates, sub-workflows) are not produced by the Java builder yet, but +//! the submit path tracks them so the surface stays forward-compatible. + +use std::collections::{HashMap, HashSet}; + +use jni::objects::{JClass, JObjectArray, JString}; +use jni::sys::{jboolean, jint, jlong, jstring, JNI_FALSE}; +use jni::JNIEnv; +use serde::{Deserialize, Serialize}; +use taskito_core::job::{now_millis, NewJob}; +use taskito_core::Storage; +use taskito_workflows::dagron_core::DAG; +use taskito_workflows::{ + topological_order, StepMetadata, WorkflowDefinition, WorkflowNode, WorkflowNodeStatus, + WorkflowRun, WorkflowState, WorkflowStorage, WorkflowStorageBackend, +}; +use uuid::Uuid; + +use crate::backend::QueueHandle; +use crate::convert::{parse_json, to_json}; +use crate::error::BindingError; +use crate::ffi::{ + guard, new_string, read_bytes_array, read_optional_string, read_string, read_string_array, +}; + +const DEFAULT_QUEUE: &str = "default"; +const DEFAULT_MAX_RETRIES: i32 = 3; +const DEFAULT_TIMEOUT_MS: i64 = 300_000; + +/// One step as described by the Java builder. Edges come from `after`. +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct StepSpec { + name: String, + task_name: String, + #[serde(default)] + after: Vec, + queue: Option, + max_retries: Option, + timeout_ms: Option, + priority: Option, +} + +/// Combined run + node snapshot returned by `getWorkflowStatus`. Timestamps are +/// Unix milliseconds; `state`/`status` are the lowercase wire strings. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct WorkflowStatusView<'a> { + run_id: &'a str, + state: &'static str, + started_at: Option, + completed_at: Option, + error: Option<&'a str>, + nodes: Vec>, +} + +/// One node within a run. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct NodeView<'a> { + node_name: &'a str, + status: &'static str, + job_id: Option<&'a str>, + result_hash: Option<&'a str>, + fan_out_count: Option, + started_at: Option, + completed_at: Option, + error: Option<&'a str>, +} + +impl<'a> From<&'a WorkflowNode> for NodeView<'a> { + fn from(n: &'a WorkflowNode) -> Self { + Self { + node_name: &n.node_name, + status: n.status.as_str(), + job_id: n.job_id.as_deref(), + result_hash: n.result_hash.as_deref(), + fan_out_count: n.fan_out_count, + started_at: n.started_at, + completed_at: n.completed_at, + error: n.error.as_deref(), + } + } +} + +/// Borrow the queue behind a handle. +/// +/// # Safety +/// `handle` must be a live `QueueHandle` pointer (see [`crate::handle`]). +unsafe fn borrow_queue<'a>(handle: jlong) -> &'a QueueHandle { + crate::handle::borrow::(handle) +} + +/// `String submitWorkflow(long, String name, int version, String stepsJson, +/// String[] payloadNames, byte[][] payloads, String queueDefault, +/// String paramsJson, String[] deferredNames)` — record a run and pre-enqueue a +/// job per static step. Returns the run id. +#[no_mangle] +#[allow(clippy::too_many_arguments)] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_submitWorkflow<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + name: JString<'local>, + version: jint, + steps_json: JString<'local>, + payload_names: JObjectArray<'local>, + payloads: JObjectArray<'local>, + queue_default: JString<'local>, + params_json: JString<'local>, + deferred_names: JObjectArray<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let name = read_string(env, &name)?; + let steps_raw = read_string(env, &steps_json)?; + let names = read_string_array(env, &payload_names)?; + let payload_bytes = read_bytes_array(env, &payloads)?; + let queue_default = read_optional_string(env, &queue_default)?; + let params = read_optional_string(env, ¶ms_json)?; + let deferred = read_string_array(env, &deferred_names)?; + let run_id = submit( + queue, + name, + version, + &steps_raw, + names, + payload_bytes, + queue_default, + params, + deferred, + )?; + new_string(env, run_id) + }) +} + +#[allow(clippy::too_many_arguments)] +fn submit( + queue: &QueueHandle, + name: String, + version: i32, + steps_json: &str, + payload_names: Vec, + payloads: Vec>, + queue_default: Option, + params_json: Option, + deferred_names: Vec, +) -> Result { + if payload_names.len() != payloads.len() { + return Err(BindingError::new( + "payloadNames and payloads must have equal length", + )); + } + let wf = queue.workflow_store()?; + let specs: Vec = parse_json(steps_json, "workflow steps")?; + + // Add every node first, then the edges, so steps may be declared in any + // order (a step's `after` need not have been declared before it). + let mut dag: DAG<()> = DAG::new(); + for spec in &specs { + dag.add_node(spec.name.clone(), ()).map_err(|e| { + BindingError::new(format!("workflow add_node('{}') failed: {e}", spec.name)) + })?; + } + for spec in &specs { + for pred in &spec.after { + dag.add_edge(pred, &spec.name, None, None).map_err(|e| { + BindingError::new(format!( + "workflow add_edge('{pred}' -> '{}') failed: {e}", + spec.name + )) + })?; + } + } + let dag_bytes = dag + .to_json(|_| None) + .map_err(|e| BindingError::new(format!("workflow DAG serialize failed: {e}")))? + .into_bytes(); + + let step_meta: HashMap = specs + .iter() + .map(|s| { + ( + s.name.clone(), + StepMetadata { + task_name: s.task_name.clone(), + queue: s.queue.clone(), + max_retries: s.max_retries, + timeout_ms: s.timeout_ms, + priority: s.priority, + ..Default::default() + }, + ) + }) + .collect(); + + let node_payloads: HashMap> = payload_names.into_iter().zip(payloads).collect(); + let deferred: HashSet = deferred_names.into_iter().collect(); + let queue_default = queue_default.unwrap_or_else(|| DEFAULT_QUEUE.to_string()); + let ordered = topological_order(&dag_bytes)?; + + // Reuse a stored definition only when its DAG matches; otherwise the run + // would execute one topology while `definition_id` points at another. + let definition_id = match wf.get_workflow_definition(&name, Some(version))? { + Some(existing) => { + if existing.dag_data != dag_bytes { + return Err(BindingError::new(format!( + "workflow '{name}' v{version} already exists with a different DAG; bump the version" + ))); + } + existing.id + } + None => { + let def = WorkflowDefinition { + id: Uuid::now_v7().to_string(), + name: name.clone(), + version, + dag_data: dag_bytes.clone(), + step_metadata: step_meta.clone(), + created_at: now_millis(), + }; + let id = def.id.clone(); + wf.create_workflow_definition(&def)?; + id + } + }; + + let run_id = Uuid::now_v7().to_string(); + let now = now_millis(); + wf.create_workflow_run(&WorkflowRun { + id: run_id.clone(), + definition_id, + params: params_json, + state: WorkflowState::Pending, + started_at: Some(now), + completed_at: None, + error: None, + parent_run_id: None, + parent_node_name: None, + created_at: now, + })?; + + let mut job_ids: HashMap = HashMap::new(); + for topo in &ordered { + let meta = step_meta + .get(&topo.name) + .ok_or_else(|| BindingError::new(format!("step '{}' missing metadata", topo.name)))?; + + // Deferred node: tracked but not enqueued; the tracker creates its job + // once predecessors settle. (Not produced by the Java builder yet.) + if deferred.contains(&topo.name) { + wf.create_workflow_node(&new_node(&run_id, &topo.name, None))?; + continue; + } + + let payload = node_payloads + .get(&topo.name) + .cloned() + .ok_or_else(|| BindingError::new(format!("step '{}' missing payload", topo.name)))?; + // Only static predecessors gate this job; a deferred one has no job to + // wait on, so the tracker enqueues nodes downstream of it. + let depends_on = topo + .predecessors + .iter() + .filter(|p| !deferred.contains(*p)) + .map(|p| { + job_ids.get(p).cloned().ok_or_else(|| { + BindingError::new(format!("predecessor '{p}' of '{}' has no job", topo.name)) + }) + }) + .collect::, BindingError>>()?; + + let job = queue.storage.enqueue(NewJob { + queue: meta.queue.clone().unwrap_or_else(|| queue_default.clone()), + task_name: meta.task_name.clone(), + payload, + priority: meta.priority.unwrap_or(0), + scheduled_at: now, + max_retries: meta.max_retries.unwrap_or(DEFAULT_MAX_RETRIES), + timeout_ms: meta.timeout_ms.unwrap_or(DEFAULT_TIMEOUT_MS), + unique_key: None, + metadata: Some(workflow_metadata_json(&run_id, &topo.name)), + notes: None, + depends_on, + expires_at: None, + result_ttl_ms: None, + namespace: queue.namespace.clone(), + })?; + job_ids.insert(topo.name.clone(), job.id.clone()); + wf.create_workflow_node(&new_node(&run_id, &topo.name, Some(job.id)))?; + } + + wf.update_workflow_run_state(&run_id, WorkflowState::Running, None)?; + Ok(run_id) +} + +/// `String markWorkflowNodeResult(long, String jobId, boolean succeeded, +/// String error, boolean skipCascade)` — record a node's terminal outcome. +/// Returns the run's final state when it reached terminal, else `null` (also for +/// non-workflow jobs). +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_markWorkflowNodeResult< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + job_id: JString<'local>, + succeeded: jboolean, + error: JString<'local>, + skip_cascade: jboolean, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let job_id = read_string(env, &job_id)?; + let error = read_optional_string(env, &error)?; + match mark_node_result( + queue, + &job_id, + succeeded != JNI_FALSE, + error, + skip_cascade != JNI_FALSE, + )? { + Some(state) => new_string(env, state), + None => Ok(std::ptr::null_mut()), + } + }) +} + +fn mark_node_result( + queue: &QueueHandle, + job_id: &str, + succeeded: bool, + error: Option, + skip_cascade: bool, +) -> Result, BindingError> { + let wf = queue.workflow_store()?; + let job = match queue.storage.get_job(job_id)? { + Some(j) => j, + None => return Ok(None), + }; + let (run_id, node_name) = match parse_workflow_metadata(job.metadata.as_deref()) { + Some(pair) => pair, + None => return Ok(None), + }; + + let now = now_millis(); + if succeeded { + wf.set_workflow_node_completed(&run_id, &node_name, now, None)?; + } else { + let msg = error.clone().unwrap_or_else(|| "failed".to_string()); + wf.set_workflow_node_error(&run_id, &node_name, &msg)?; + if !skip_cascade { + let nodes = wf.get_workflow_nodes(&run_id)?; + cascade_skip_pending(queue, &wf, &run_id, &nodes); + } + } + + // Tracker-managed run: it owns cascade + finalization. Not used by the + // current Java tracker (static DAGs), kept for forward compatibility. + if skip_cascade { + return Ok(None); + } + + let nodes = wf.get_workflow_nodes(&run_id)?; + if !nodes.iter().all(|n| n.status.is_terminal()) { + return Ok(None); + } + let any_failed = nodes.iter().any(|n| n.status == WorkflowNodeStatus::Failed); + let final_state = if any_failed || !succeeded { + WorkflowState::Failed + } else { + WorkflowState::Completed + }; + wf.update_workflow_run_state( + &run_id, + final_state, + if final_state == WorkflowState::Failed { + error.as_deref() + } else { + None + }, + )?; + wf.set_workflow_run_completed(&run_id, now)?; + Ok(Some(final_state.as_str().to_string())) +} + +/// `String getWorkflowStatus(long, String runId)` — a JSON run + node snapshot, +/// or `null` if no such run exists. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWorkflowStatus< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let wf = queue.workflow_store()?; + let run = match wf.get_workflow_run(&run_id)? { + Some(r) => r, + None => return Ok(std::ptr::null_mut()), + }; + let nodes = wf.get_workflow_nodes(&run_id)?; + let view = WorkflowStatusView { + run_id: &run.id, + state: run.state.as_str(), + started_at: run.started_at, + completed_at: run.completed_at, + error: run.error.as_deref(), + nodes: nodes.iter().map(NodeView::from).collect(), + }; + new_string(env, to_json(&view)?) + }) +} + +/// `void cancelWorkflowRun(long, String runId)` — skip every pending node +/// (cancelling its job), request cancel on running nodes, and mark the run +/// `cancelled`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_cancelWorkflowRun< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let wf = queue.workflow_store()?; + for node in &wf.get_workflow_nodes(&run_id)? { + match node.status { + WorkflowNodeStatus::Pending | WorkflowNodeStatus::Ready => { + if let Some(job_id) = &node.job_id { + let _ = queue.storage.cancel_job(job_id); + } + wf.update_workflow_node_status( + &run_id, + &node.node_name, + WorkflowNodeStatus::Skipped, + )?; + } + WorkflowNodeStatus::Running => { + if let Some(job_id) = &node.job_id { + let _ = queue.storage.request_cancel(job_id); + } + } + _ => {} + } + } + wf.update_workflow_run_state(&run_id, WorkflowState::Cancelled, None)?; + wf.set_workflow_run_completed(&run_id, now_millis())?; + Ok(()) + }) +} + +/// A fresh `Pending` node. `job_id` is `None` for deferred nodes. +fn new_node(run_id: &str, node_name: &str, job_id: Option) -> WorkflowNode { + WorkflowNode { + id: Uuid::now_v7().to_string(), + run_id: run_id.to_string(), + node_name: node_name.to_string(), + job_id, + status: WorkflowNodeStatus::Pending, + result_hash: None, + fan_out_count: None, + fan_in_data: None, + started_at: None, + completed_at: None, + error: None, + compensation_job_id: None, + compensation_started_at: None, + compensation_completed_at: None, + compensation_error: None, + } +} + +/// Job-metadata blob linking a job back to its workflow node. +fn workflow_metadata_json(run_id: &str, node_name: &str) -> String { + serde_json::json!({ + "workflow_run_id": run_id, + "workflow_node_name": node_name, + }) + .to_string() +} + +/// Parse `{workflow_run_id, workflow_node_name}` from a job's metadata. Returns +/// `None` for compensation jobs so they never advance the forward run. +fn parse_workflow_metadata(metadata: Option<&str>) -> Option<(String, String)> { + let parsed: serde_json::Value = serde_json::from_str(metadata?).ok()?; + if parsed + .get("compensation") + .and_then(serde_json::Value::as_bool) + == Some(true) + { + return None; + } + let run_id = parsed.get("workflow_run_id")?.as_str()?.to_string(); + let node_name = parsed.get("workflow_node_name")?.as_str()?.to_string(); + Some((run_id, node_name)) +} + +/// Fail-fast: mark every still-pending node skipped and cancel its job. +/// Best-effort — per-node failures are logged, not propagated. +fn cascade_skip_pending( + queue: &QueueHandle, + wf: &WorkflowStorageBackend, + run_id: &str, + nodes: &[WorkflowNode], +) { + for node in nodes { + if !matches!( + node.status, + WorkflowNodeStatus::Pending | WorkflowNodeStatus::Ready + ) { + continue; + } + if let Some(job_id) = &node.job_id { + if let Err(e) = queue.storage.cancel_job(job_id) { + log::warn!("[taskito-java] cancel_job({job_id}) during workflow cascade: {e}"); + } + } + if let Err(e) = + wf.update_workflow_node_status(run_id, &node.node_name, WorkflowNodeStatus::Skipped) + { + log::warn!("[taskito-java] skip node '{}' failed: {e}", node.node_name); + } + } +} diff --git a/sdks/java/README.md b/sdks/java/README.md index 2bb8c725..0c8544a3 100644 --- a/sdks/java/README.md +++ b/sdks/java/README.md @@ -3,17 +3,23 @@ A typed Java 11+ client over the Taskito Rust core, via a hand-written JNI shell (`crates/taskito-java`). -> Status: **build-out** (Phases 0–3 core). Producer + inspection + admin + logs, -> worker task execution, middleware, signed/encrypted serializers, and the -> dashboard server are implemented and verified end-to-end. Webhooks and the CLI -> follow next. +> Status: **build-out**. Producer + inspection + admin + logs, worker task +> execution, middleware, signed/encrypted serializers, dashboard, webhooks, CLI, +> distributed locks, periodic/cron, and static-DAG workflows are implemented and +> verified end-to-end. ## Usage ### Enqueue ```java -import org.byteveda.taskito.*; +import org.byteveda.taskito.Queue; +import org.byteveda.taskito.Taskito; +import org.byteveda.taskito.model.Job; +import org.byteveda.taskito.model.JobStatus; +import org.byteveda.taskito.model.QueueStats; +import org.byteveda.taskito.task.EnqueueOptions; +import org.byteveda.taskito.task.Task; import java.util.Map; Task sendEmail = Task.of("send_email", Map.class) @@ -31,6 +37,8 @@ try (Queue queue = Taskito.builder().backend("sqlite").url("taskito.db").open()) ```java import org.byteveda.taskito.events.EventName; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; Task add = Task.of("add", Map.class); @@ -51,6 +59,45 @@ Runtime.getRuntime().addShutdownHook(new Thread(() -> { worker.awaitShutdown(); ``` +### Workflows + +Define a DAG of tasks; steps run in topological order once every predecessor +finishes. Attach `trackWorkflows()` to the worker so node and run state advance +as jobs complete. + +```java +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.byteveda.taskito.workflows.Workflow; +import org.byteveda.taskito.workflows.WorkflowRun; +import org.byteveda.taskito.workflows.WorkflowStatus; + +Task extract = Task.of("extract", Integer.class); +Task transform = Task.of("transform", Integer.class); +Task load = Task.of("load", Integer.class); + +Workflow wf = Workflow.named("etl") + .step("extract", extract, 1) + .step("transform", transform, 2, "extract") + .step("load", load, 3, "transform"); + +WorkflowRun run = queue.submitWorkflow(wf); + +try (Worker worker = queue.worker() + .handle(extract, p -> p * 10) + .handle(transform, p -> p + 1) + .handle(load, p -> p) + .trackWorkflows() + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(30)); + // status.state == WorkflowState.COMPLETED; status.node("load").get().status +} +``` + +A failed step (after its retries) fails the run and skips downstream nodes; +`run.cancel()` skips pending nodes. Fan-out, conditions, gates, and sagas are +not yet exposed. + ### Middleware ```java @@ -82,16 +129,27 @@ Queue secure = Taskito.builder() ## Structure +Packages are organized by feature; the root holds only the front door. + ```text org.byteveda.taskito ├── Taskito entry point — Taskito.builder()...open() ├── Queue public interface ├── DefaultQueue package-private impl (not exported) -├── Task / Job / JobStatus / QueueStats / EnqueueOptions typed model ├── TaskitoException unchecked error type +├── task/ Task, TaskFunction, EnqueueOptions +├── model/ Job, JobStatus, QueueStats, DeadJob, JobError, +│ TaskMetric, WorkerInfo, TaskLog, JobFilter (read-only views) +├── worker/ Worker runtime +├── locks/ Lock, LockInfo +├── scheduling/ PeriodicTask +├── workflows/ Workflow DAG builder, run, status, tracker ├── serialization/ Serializer SPI + JsonSerializer (Jackson) default +├── middleware/ Middleware hooks +├── events/ worker outcome events +├── dashboard/ webhooks/ cli/ ├── spi/ QueueBackend — seam between API and the native layer -└── internal/ JNI bindings (NativeQueue, NativeLoader, JniQueueBackend) +└── internal/ JNI bindings (NativeQueue, NativeWorkflows, NativeLoader, ...) ``` The `spi.QueueBackend` seam keeps the public API independent of JNI: it can be diff --git a/sdks/java/build.gradle.kts b/sdks/java/build.gradle.kts index f5d3a3d6..45c10315 100644 --- a/sdks/java/build.gradle.kts +++ b/sdks/java/build.gradle.kts @@ -96,7 +96,7 @@ val nativeStaging = layout.buildDirectory.dir("native") // Build the native library for the local platform. val cargoBuild by tasks.registering(Exec::class) { workingDir = crateDir.asFile - commandLine("cargo", "build", "--release", "--features", "postgres,redis") + commandLine("cargo", "build", "--release", "--features", "postgres,redis,workflows") } // Stage the built library under its platform-classifier resource path. diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java index f4871047..9a527a41 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java @@ -4,14 +4,33 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.util.ArrayList; import java.util.Arrays; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.concurrent.CopyOnWriteArrayList; +import org.byteveda.taskito.locks.Lock; +import org.byteveda.taskito.locks.LockInfo; import org.byteveda.taskito.middleware.EnqueueContext; import org.byteveda.taskito.middleware.Middleware; +import org.byteveda.taskito.model.DeadJob; +import org.byteveda.taskito.model.Job; +import org.byteveda.taskito.model.JobError; +import org.byteveda.taskito.model.JobFilter; +import org.byteveda.taskito.model.QueueStats; +import org.byteveda.taskito.model.TaskLog; +import org.byteveda.taskito.model.TaskMetric; +import org.byteveda.taskito.model.WorkerInfo; +import org.byteveda.taskito.scheduling.PeriodicTask; import org.byteveda.taskito.serialization.Serializer; import org.byteveda.taskito.spi.QueueBackend; +import org.byteveda.taskito.task.EnqueueOptions; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.byteveda.taskito.workflows.Step; +import org.byteveda.taskito.workflows.Workflow; +import org.byteveda.taskito.workflows.WorkflowRun; +import org.byteveda.taskito.workflows.WorkflowStatus; /** * Default {@link Queue}: maps the typed public API onto a {@link QueueBackend}, @@ -257,6 +276,56 @@ public long registerPeriodic(PeriodicTask task) { task.name, task.taskName, task.cron, payload, task.queue, task.timezone, task.enabled); } + // ── Workflows ─────────────────────────────────────────────────── + + @Override + public WorkflowRun submitWorkflow(Workflow workflow) { + List steps = workflow.steps(); + List> specs = new ArrayList<>(steps.size()); + String[] names = new String[steps.size()]; + byte[][] payloads = new byte[steps.size()][]; + for (int i = 0; i < steps.size(); i++) { + Step step = steps.get(i); + specs.add(stepSpec(step)); + names[i] = step.name; + payloads[i] = serializer.serialize(step.payload); + } + String runId = backend.submitWorkflow( + workflow.name(), workflow.version(), encode(specs), names, payloads, null, null, new String[0]); + return new WorkflowRun(backend, VIEWS, runId, workflow.name()); + } + + @Override + public Optional workflowStatus(String runId) { + return backend.getWorkflowStatusJson(runId).map(json -> decode(json, WorkflowStatus.class)); + } + + @Override + public void cancelWorkflow(String runId) { + backend.cancelWorkflowRun(runId); + } + + /** Encode a step's structure + per-step overrides for the native submit call. */ + private static Map stepSpec(Step step) { + Map spec = new LinkedHashMap<>(); + spec.put("name", step.name); + spec.put("taskName", step.taskName); + spec.put("after", step.after); + if (step.queue != null) { + spec.put("queue", step.queue); + } + if (step.maxRetries != null) { + spec.put("maxRetries", step.maxRetries); + } + if (step.timeoutMs != null) { + spec.put("timeoutMs", step.timeoutMs); + } + if (step.priority != null) { + spec.put("priority", step.priority); + } + return spec; + } + @Override public Worker.Builder worker() { return Worker.builder(backend, serializer, middleware); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java index 650cbe49..846cd66a 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Queue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Queue.java @@ -3,7 +3,24 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import org.byteveda.taskito.locks.Lock; +import org.byteveda.taskito.locks.LockInfo; import org.byteveda.taskito.middleware.Middleware; +import org.byteveda.taskito.model.DeadJob; +import org.byteveda.taskito.model.Job; +import org.byteveda.taskito.model.JobError; +import org.byteveda.taskito.model.JobFilter; +import org.byteveda.taskito.model.QueueStats; +import org.byteveda.taskito.model.TaskLog; +import org.byteveda.taskito.model.TaskMetric; +import org.byteveda.taskito.model.WorkerInfo; +import org.byteveda.taskito.scheduling.PeriodicTask; +import org.byteveda.taskito.task.EnqueueOptions; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.byteveda.taskito.workflows.Workflow; +import org.byteveda.taskito.workflows.WorkflowRun; +import org.byteveda.taskito.workflows.WorkflowStatus; /** A Taskito queue. Obtain one from {@link Taskito#builder()}. */ public interface Queue extends AutoCloseable { @@ -108,6 +125,17 @@ public interface Queue extends AutoCloseable { /** Register (or replace) a cron task; returns the next fire time (Unix ms). */ long registerPeriodic(PeriodicTask task); + // ── Workflows ─────────────────────────────────────────────────── + + /** Submit a workflow DAG; returns a handle to the run. */ + WorkflowRun submitWorkflow(Workflow workflow); + + /** Current status of a workflow run, or empty if it no longer exists. */ + Optional workflowStatus(String runId); + + /** Cancel a workflow run: skip its pending nodes and mark it cancelled. */ + void cancelWorkflow(String runId); + // ── Worker ────────────────────────────────────────────────────── /** Begin building a worker over this queue. */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/cli/Cli.java b/sdks/java/src/main/java/org/byteveda/taskito/cli/Cli.java index 48fe8b52..e47960de 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/cli/Cli.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/cli/Cli.java @@ -4,13 +4,13 @@ import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; -import org.byteveda.taskito.DeadJob; -import org.byteveda.taskito.Job; -import org.byteveda.taskito.JobFilter; -import org.byteveda.taskito.JobStatus; import org.byteveda.taskito.Queue; import org.byteveda.taskito.Taskito; import org.byteveda.taskito.dashboard.DashboardServer; +import org.byteveda.taskito.model.DeadJob; +import org.byteveda.taskito.model.Job; +import org.byteveda.taskito.model.JobFilter; +import org.byteveda.taskito.model.JobStatus; import picocli.CommandLine; import picocli.CommandLine.Command; import picocli.CommandLine.Option; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/Contract.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/Contract.java index dee31308..fbfb9f80 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/Contract.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/Contract.java @@ -2,11 +2,11 @@ import java.util.LinkedHashMap; import java.util.Map; -import org.byteveda.taskito.DeadJob; -import org.byteveda.taskito.Job; -import org.byteveda.taskito.QueueStats; -import org.byteveda.taskito.TaskMetric; -import org.byteveda.taskito.WorkerInfo; +import org.byteveda.taskito.model.DeadJob; +import org.byteveda.taskito.model.Job; +import org.byteveda.taskito.model.QueueStats; +import org.byteveda.taskito.model.TaskMetric; +import org.byteveda.taskito.model.WorkerInfo; /** * Maps the SDK's camelCase types to the snake_case JSON contract the React SPA diff --git a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java index eae3c931..1984431b 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/dashboard/DashboardServer.java @@ -21,11 +21,11 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; import java.util.stream.Collectors; -import org.byteveda.taskito.Job; -import org.byteveda.taskito.JobFilter; -import org.byteveda.taskito.JobStatus; import org.byteveda.taskito.Queue; import org.byteveda.taskito.TaskitoException; +import org.byteveda.taskito.model.Job; +import org.byteveda.taskito.model.JobFilter; +import org.byteveda.taskito.model.JobStatus; /** * A read/action dashboard API + static-SPA server backed by a {@link Queue}, diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java index 85601327..ba1be7d1 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java @@ -190,6 +190,35 @@ public long registerPeriodic( return NativeQueue.registerPeriodic(handle, name, taskName, cron, args, queue, timezone, enabled); } + @Override + public String submitWorkflow( + String name, + int version, + String stepsJson, + String[] payloadNames, + byte[][] payloads, + String queueDefault, + String paramsJson, + String[] deferredNames) { + return NativeWorkflows.submitWorkflow( + handle, name, version, stepsJson, payloadNames, payloads, queueDefault, paramsJson, deferredNames); + } + + @Override + public String markWorkflowNodeResult(String jobId, boolean succeeded, String error, boolean skipCascade) { + return NativeWorkflows.markWorkflowNodeResult(handle, jobId, succeeded, error, skipCascade); + } + + @Override + public Optional getWorkflowStatusJson(String runId) { + return Optional.ofNullable(NativeWorkflows.getWorkflowStatus(handle, runId)); + } + + @Override + public void cancelWorkflowRun(String runId) { + NativeWorkflows.cancelWorkflowRun(handle, runId); + } + @Override public WorkerControl startWorker(WorkerBridge bridge, String optionsJson) { return new JniWorkerControl(NativeQueue.runWorker(handle, bridge, optionsJson)); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.java b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.java new file mode 100644 index 00000000..ed5fa499 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.java @@ -0,0 +1,38 @@ +package org.byteveda.taskito.internal; + +/** + * JNI surface for workflow operations (the native {@code workflows} feature). + * + *

Opaque payloads cross as {@code byte[]}; step descriptions and views cross + * as JSON strings. Methods throw {@link org.byteveda.taskito.TaskitoException} + * on native failure. The {@code handle} is the queue handle from + * {@link NativeQueue#open}. + */ +public final class NativeWorkflows { + static { + NativeLoader.load(); + } + + private NativeWorkflows() {} + + /** Record a run and pre-enqueue a job per static step; returns the run id. */ + public static native String submitWorkflow( + long handle, + String name, + int version, + String stepsJson, + String[] payloadNames, + byte[][] payloads, + String queueDefault, + String paramsJson, + String[] deferredNames); + + /** Record a node's terminal outcome; returns the run's final state, or {@code null}. */ + public static native String markWorkflowNodeResult( + long handle, String jobId, boolean succeeded, String error, boolean skipCascade); + + /** Returns a JSON run + node snapshot, or {@code null} if the run is absent. */ + public static native String getWorkflowStatus(long handle, String runId); + + public static native void cancelWorkflowRun(long handle, String runId); +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Lock.java b/sdks/java/src/main/java/org/byteveda/taskito/locks/Lock.java similarity index 93% rename from sdks/java/src/main/java/org/byteveda/taskito/Lock.java rename to sdks/java/src/main/java/org/byteveda/taskito/locks/Lock.java index ef891c16..d5c130ff 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Lock.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/locks/Lock.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.locks; import java.util.UUID; import org.byteveda.taskito.spi.QueueBackend; @@ -15,7 +15,7 @@ public final class Lock implements AutoCloseable { private final long ttlMs; private boolean held; - Lock(QueueBackend backend, String name, long ttlMs) { + public Lock(QueueBackend backend, String name, long ttlMs) { this.backend = backend; this.name = name; this.ttlMs = ttlMs; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/LockInfo.java b/sdks/java/src/main/java/org/byteveda/taskito/locks/LockInfo.java similarity index 96% rename from sdks/java/src/main/java/org/byteveda/taskito/LockInfo.java rename to sdks/java/src/main/java/org/byteveda/taskito/locks/LockInfo.java index f450b57b..21abdad2 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/LockInfo.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/locks/LockInfo.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.locks; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/locks/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/locks/package-info.java new file mode 100644 index 00000000..f9e8bbed --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/locks/package-info.java @@ -0,0 +1,2 @@ +/** Distributed, TTL-bounded advisory locks. */ +package org.byteveda.taskito.locks; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java b/sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java index 16193ec1..0f877ac4 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/middleware/EnqueueContext.java @@ -1,6 +1,6 @@ package org.byteveda.taskito.middleware; -import org.byteveda.taskito.EnqueueOptions; +import org.byteveda.taskito.task.EnqueueOptions; /** * A job being enqueued, passed to {@link Middleware#onEnqueue} before diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DeadJob.java b/sdks/java/src/main/java/org/byteveda/taskito/model/DeadJob.java similarity index 97% rename from sdks/java/src/main/java/org/byteveda/taskito/DeadJob.java rename to sdks/java/src/main/java/org/byteveda/taskito/model/DeadJob.java index ffe43fa1..de6f61a1 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DeadJob.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/DeadJob.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.model; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Job.java b/sdks/java/src/main/java/org/byteveda/taskito/model/Job.java similarity index 98% rename from sdks/java/src/main/java/org/byteveda/taskito/Job.java rename to sdks/java/src/main/java/org/byteveda/taskito/model/Job.java index 1d331556..dbdd1b00 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Job.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/Job.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.model; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/JobError.java b/sdks/java/src/main/java/org/byteveda/taskito/model/JobError.java similarity index 96% rename from sdks/java/src/main/java/org/byteveda/taskito/JobError.java rename to sdks/java/src/main/java/org/byteveda/taskito/model/JobError.java index 5f81dad5..0344868a 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/JobError.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/JobError.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.model; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/JobFilter.java b/sdks/java/src/main/java/org/byteveda/taskito/model/JobFilter.java similarity index 98% rename from sdks/java/src/main/java/org/byteveda/taskito/JobFilter.java rename to sdks/java/src/main/java/org/byteveda/taskito/model/JobFilter.java index 4f03c252..7dcb1827 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/JobFilter.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/JobFilter.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.model; import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonProperty; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/JobStatus.java b/sdks/java/src/main/java/org/byteveda/taskito/model/JobStatus.java similarity index 90% rename from sdks/java/src/main/java/org/byteveda/taskito/JobStatus.java rename to sdks/java/src/main/java/org/byteveda/taskito/model/JobStatus.java index 85195385..be09d6d4 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/JobStatus.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/JobStatus.java @@ -1,8 +1,9 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.model; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonValue; import java.util.Locale; +import org.byteveda.taskito.TaskitoException; /** Lifecycle state of a job. Wire form is the lowercase name, shared across SDKs. */ public enum JobStatus { diff --git a/sdks/java/src/main/java/org/byteveda/taskito/QueueStats.java b/sdks/java/src/main/java/org/byteveda/taskito/model/QueueStats.java similarity index 96% rename from sdks/java/src/main/java/org/byteveda/taskito/QueueStats.java rename to sdks/java/src/main/java/org/byteveda/taskito/model/QueueStats.java index d177a626..36519a6b 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/QueueStats.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/QueueStats.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.model; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/TaskLog.java b/sdks/java/src/main/java/org/byteveda/taskito/model/TaskLog.java similarity index 96% rename from sdks/java/src/main/java/org/byteveda/taskito/TaskLog.java rename to sdks/java/src/main/java/org/byteveda/taskito/model/TaskLog.java index 09f7c4f6..ba913a6e 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/TaskLog.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/TaskLog.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.model; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/TaskMetric.java b/sdks/java/src/main/java/org/byteveda/taskito/model/TaskMetric.java similarity index 96% rename from sdks/java/src/main/java/org/byteveda/taskito/TaskMetric.java rename to sdks/java/src/main/java/org/byteveda/taskito/model/TaskMetric.java index 037639d9..2f640d01 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/TaskMetric.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/TaskMetric.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.model; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/WorkerInfo.java b/sdks/java/src/main/java/org/byteveda/taskito/model/WorkerInfo.java similarity index 97% rename from sdks/java/src/main/java/org/byteveda/taskito/WorkerInfo.java rename to sdks/java/src/main/java/org/byteveda/taskito/model/WorkerInfo.java index c9795acf..2088291a 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/WorkerInfo.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/WorkerInfo.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.model; import com.fasterxml.jackson.annotation.JsonCreator; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/model/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/model/package-info.java new file mode 100644 index 00000000..88b5042e --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/model/package-info.java @@ -0,0 +1,2 @@ +/** Immutable, read-only data the API returns: jobs, stats, dead letters, metrics, logs. */ +package org.byteveda.taskito.model; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/package-info.java index 67e0d8b5..d4bf14dc 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/package-info.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/package-info.java @@ -1,11 +1,20 @@ /** * Taskito Java SDK — a typed client over the Rust task-queue core. * - *

Open a {@link org.byteveda.taskito.Queue} via - * {@link org.byteveda.taskito.Taskito#builder()}. Enqueue typed payloads with a - * {@link org.byteveda.taskito.Task}; payloads serialize through a - * {@link org.byteveda.taskito.serialization.Serializer}. The implementation - * talks to the core through the {@link org.byteveda.taskito.spi.QueueBackend} - * seam, whose default implementation lives in {@code internal}. + *

This root package is the front door: open a + * {@link org.byteveda.taskito.Queue} via + * {@link org.byteveda.taskito.Taskito#builder()}. Everything else is grouped by + * feature: + * + *

    + *
  • {@code task} — {@link org.byteveda.taskito.task.Task} descriptors, + * handler functions, enqueue options + *
  • {@code model} — immutable views the API returns (jobs, stats, metrics) + *
  • {@code worker} — the worker runtime + *
  • {@code locks} / {@code scheduling} / {@code workflows} — feature surfaces + *
  • {@code serialization} / {@code middleware} / {@code events} — cross-cutting + *
  • {@code spi} — the {@link org.byteveda.taskito.spi.QueueBackend} seam, + * whose default implementation lives in {@code internal} + *
*/ package org.byteveda.taskito; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/PeriodicTask.java b/sdks/java/src/main/java/org/byteveda/taskito/scheduling/PeriodicTask.java similarity index 85% rename from sdks/java/src/main/java/org/byteveda/taskito/PeriodicTask.java rename to sdks/java/src/main/java/org/byteveda/taskito/scheduling/PeriodicTask.java index 3deaca03..76264c6a 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/PeriodicTask.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/scheduling/PeriodicTask.java @@ -1,14 +1,14 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.scheduling; /** A cron-scheduled task registration. The worker enqueues it when due. */ public final class PeriodicTask { - final String name; - final String taskName; - final String cron; - final Object payload; - final String queue; - final String timezone; - final boolean enabled; + public final String name; + public final String taskName; + public final String cron; + public final Object payload; + public final String queue; + public final String timezone; + public final boolean enabled; private PeriodicTask(Builder b) { this.name = b.name; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/scheduling/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/scheduling/package-info.java new file mode 100644 index 00000000..1f705abd --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/scheduling/package-info.java @@ -0,0 +1,2 @@ +/** Periodic (cron-scheduled) task registration. */ +package org.byteveda.taskito.scheduling; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java index ba1f3c9b..d8fc29ae 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java @@ -107,6 +107,24 @@ default long registerPeriodic( throw new UnsupportedOperationException("periodic tasks not supported by this backend"); } + // ── Workflows ─────────────────────────────────────────────────── + String submitWorkflow( + String name, + int version, + String stepsJson, + String[] payloadNames, + byte[][] payloads, + String queueDefault, + String paramsJson, + String[] deferredNames); + + /** Record a node's terminal outcome; returns the run's final state, or {@code null}. */ + String markWorkflowNodeResult(String jobId, boolean succeeded, String error, boolean skipCascade); + + Optional getWorkflowStatusJson(String runId); + + void cancelWorkflowRun(String runId); + // ── Worker ────────────────────────────────────────────────────── /** Start a worker that dispatches jobs to {@code bridge}; returns its control. */ WorkerControl startWorker(WorkerBridge bridge, String optionsJson); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/EnqueueOptions.java b/sdks/java/src/main/java/org/byteveda/taskito/task/EnqueueOptions.java similarity index 98% rename from sdks/java/src/main/java/org/byteveda/taskito/EnqueueOptions.java rename to sdks/java/src/main/java/org/byteveda/taskito/task/EnqueueOptions.java index 67fde0d0..e8b384a5 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/EnqueueOptions.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/task/EnqueueOptions.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.task; import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonProperty; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Task.java b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java similarity index 97% rename from sdks/java/src/main/java/org/byteveda/taskito/Task.java rename to sdks/java/src/main/java/org/byteveda/taskito/task/Task.java index 7aec0ef5..d0318126 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Task.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/task/Task.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.task; import java.util.Objects; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/TaskFunction.java b/sdks/java/src/main/java/org/byteveda/taskito/task/TaskFunction.java similarity index 84% rename from sdks/java/src/main/java/org/byteveda/taskito/TaskFunction.java rename to sdks/java/src/main/java/org/byteveda/taskito/task/TaskFunction.java index 704bf3b4..383acf1b 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/TaskFunction.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/task/TaskFunction.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.task; /** A task handler: receives a deserialized payload, returns a result (or null). */ @FunctionalInterface diff --git a/sdks/java/src/main/java/org/byteveda/taskito/task/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/task/package-info.java new file mode 100644 index 00000000..02b99ded --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/task/package-info.java @@ -0,0 +1,2 @@ +/** Defining and shaping work: task descriptors, handler functions, enqueue options. */ +package org.byteveda.taskito.task; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/RegisteredTask.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/RegisteredTask.java similarity index 80% rename from sdks/java/src/main/java/org/byteveda/taskito/RegisteredTask.java rename to sdks/java/src/main/java/org/byteveda/taskito/worker/RegisteredTask.java index 721514c5..b1e880f5 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/RegisteredTask.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/RegisteredTask.java @@ -1,4 +1,6 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.worker; + +import org.byteveda.taskito.task.TaskFunction; /** A worker-registered handler: payload type plus the function to run. */ final class RegisteredTask { diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java similarity index 92% rename from sdks/java/src/main/java/org/byteveda/taskito/Worker.java rename to sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java index 5eee23dd..37df4f71 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Worker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.worker; import com.fasterxml.jackson.databind.ObjectMapper; import java.util.ArrayList; @@ -13,6 +13,7 @@ import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import org.byteveda.taskito.TaskitoException; import org.byteveda.taskito.events.Emitter; import org.byteveda.taskito.events.EventName; import org.byteveda.taskito.events.OutcomeEvent; @@ -20,6 +21,9 @@ import org.byteveda.taskito.serialization.Serializer; import org.byteveda.taskito.spi.QueueBackend; import org.byteveda.taskito.spi.WorkerControl; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.task.TaskFunction; +import org.byteveda.taskito.workflows.WorkflowTracker; /** * A running worker. Build one with {@link Queue#worker()}, register handlers, @@ -135,6 +139,14 @@ public Builder on(EventName name, Consumer listener) { return this; } + /** Drive workflow node and run state from this worker's job outcomes. */ + public Builder trackWorkflows() { + WorkflowTracker tracker = new WorkflowTracker(backend); + on(EventName.SUCCESS, tracker::onSuccess); + on(EventName.DEAD, tracker::onDead); + return this; + } + public Worker start() { ExecutorService executor = concurrency > 0 ? Executors.newFixedThreadPool(concurrency) : Executors.newCachedThreadPool(); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java similarity index 99% rename from sdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.java rename to sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java index 0f5c95ba..1f09a109 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/WorkerDispatchBridge.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.java @@ -1,4 +1,4 @@ -package org.byteveda.taskito; +package org.byteveda.taskito.worker; import java.util.List; import java.util.Map; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/package-info.java new file mode 100644 index 00000000..a2d80e33 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/package-info.java @@ -0,0 +1,2 @@ +/** The worker runtime: handler registration, dispatch bridge, lifecycle. */ +package org.byteveda.taskito.worker; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/NodeSnapshot.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/NodeSnapshot.java new file mode 100644 index 00000000..d5f5415c --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/NodeSnapshot.java @@ -0,0 +1,38 @@ +package org.byteveda.taskito.workflows; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; + +/** A node's state within a workflow run. Timestamps are Unix milliseconds; nullable when unset. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class NodeSnapshot { + public final String nodeName; + public final NodeStatus status; + public final String jobId; + public final String resultHash; + public final Integer fanOutCount; + public final Long startedAt; + public final Long completedAt; + public final String error; + + @JsonCreator + public NodeSnapshot( + @JsonProperty("nodeName") String nodeName, + @JsonProperty("status") NodeStatus status, + @JsonProperty("jobId") String jobId, + @JsonProperty("resultHash") String resultHash, + @JsonProperty("fanOutCount") Integer fanOutCount, + @JsonProperty("startedAt") Long startedAt, + @JsonProperty("completedAt") Long completedAt, + @JsonProperty("error") String error) { + this.nodeName = nodeName; + this.status = status; + this.jobId = jobId; + this.resultHash = resultHash; + this.fanOutCount = fanOutCount; + this.startedAt = startedAt; + this.completedAt = completedAt; + this.error = error; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/NodeStatus.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/NodeStatus.java new file mode 100644 index 00000000..786b07e4 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/NodeStatus.java @@ -0,0 +1,30 @@ +package org.byteveda.taskito.workflows; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; +import java.util.Locale; + +/** Status of a single node within a workflow run. Wire form is the lowercase name. */ +public enum NodeStatus { + PENDING, + READY, + RUNNING, + COMPLETED, + FAILED, + SKIPPED, + WAITING_APPROVAL, + CACHE_HIT, + COMPENSATING, + COMPENSATED, + COMPENSATION_FAILED; + + @JsonValue + public String wire() { + return name().toLowerCase(Locale.ROOT); + } + + @JsonCreator + public static NodeStatus fromWire(String wire) { + return valueOf(wire.toUpperCase(Locale.ROOT)); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java new file mode 100644 index 00000000..ea1e7575 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java @@ -0,0 +1,88 @@ +package org.byteveda.taskito.workflows; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.byteveda.taskito.task.Task; + +/** One step in a {@link Workflow}: a task plus its payload, predecessors, and per-step overrides. */ +public final class Step { + public final String name; + public final String taskName; + public final Object payload; + public final List after; + public final String queue; + public final Integer maxRetries; + public final Long timeoutMs; + public final Integer priority; + + private Step(Builder builder) { + this.name = builder.name; + this.taskName = builder.taskName; + this.payload = builder.payload; + this.after = Collections.unmodifiableList(new ArrayList<>(builder.after)); + this.queue = builder.queue; + this.maxRetries = builder.maxRetries; + this.timeoutMs = builder.timeoutMs; + this.priority = builder.priority; + } + + /** Begin a step bound to a typed task. */ + public static Builder of(String name, Task task, T payload) { + return new Builder(name, task.name(), payload); + } + + /** Begin a step bound to a task name (untyped payload). */ + public static Builder of(String name, String taskName, Object payload) { + return new Builder(name, taskName, payload); + } + + /** Fluent builder for a {@link Step}. */ + public static final class Builder { + private final String name; + private final String taskName; + private final Object payload; + private final List after = new ArrayList<>(); + private String queue; + private Integer maxRetries; + private Long timeoutMs; + private Integer priority; + + private Builder(String name, String taskName, Object payload) { + this.name = name; + this.taskName = taskName; + this.payload = payload; + } + + /** Predecessor step names that must finish before this step runs. */ + public Builder after(String... predecessors) { + this.after.addAll(Arrays.asList(predecessors)); + return this; + } + + public Builder queue(String queue) { + this.queue = queue; + return this; + } + + public Builder maxRetries(int maxRetries) { + this.maxRetries = maxRetries; + return this; + } + + public Builder timeoutMs(long timeoutMs) { + this.timeoutMs = timeoutMs; + return this; + } + + public Builder priority(int priority) { + this.priority = priority; + return this; + } + + public Step build() { + return new Step(this); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java new file mode 100644 index 00000000..14e22dd1 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java @@ -0,0 +1,55 @@ +package org.byteveda.taskito.workflows; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import org.byteveda.taskito.task.Task; + +/** + * A workflow definition: a named, versioned DAG of {@link Step}s. Build one, then + * submit it with {@code Queue.submitWorkflow}. Steps run in topological order; + * a step waits for every predecessor named in its {@code after} list. + */ +public final class Workflow { + private final String name; + private int version = 1; + private final List steps = new ArrayList<>(); + + private Workflow(String name) { + this.name = name; + } + + /** Start a workflow named {@code name} (version defaults to 1). */ + public static Workflow named(String name) { + return new Workflow(name); + } + + /** Set the definition version. Bumping it lets the DAG change without conflict. */ + public Workflow version(int version) { + this.version = version; + return this; + } + + /** Add a step that runs after the named predecessors, using the task's defaults. */ + public Workflow step(String name, Task task, T payload, String... after) { + return step(Step.of(name, task, payload).after(after).build()); + } + + /** Add a fully-configured step. */ + public Workflow step(Step step) { + steps.add(step); + return this; + } + + public String name() { + return name; + } + + public int version() { + return version; + } + + public List steps() { + return Collections.unmodifiableList(steps); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowRun.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowRun.java new file mode 100644 index 00000000..f8dea1d7 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowRun.java @@ -0,0 +1,73 @@ +package org.byteveda.taskito.workflows; + +import com.fasterxml.jackson.databind.ObjectMapper; +import java.time.Duration; +import java.util.Optional; +import org.byteveda.taskito.TaskitoException; +import org.byteveda.taskito.spi.QueueBackend; + +/** A submitted workflow run. Query {@link #status()}, block on {@link #await}, or {@link #cancel()}. */ +public final class WorkflowRun { + private final QueueBackend backend; + private final ObjectMapper json; + private final String id; + private final String name; + + public WorkflowRun(QueueBackend backend, ObjectMapper json, String id, String name) { + this.backend = backend; + this.json = json; + this.id = id; + this.name = name; + } + + public String id() { + return id; + } + + public String name() { + return name; + } + + /** Current run + node snapshot, or empty if the run no longer exists. */ + public Optional status() { + return backend.getWorkflowStatusJson(id).map(this::decode); + } + + /** Block until the run reaches a terminal state, polling every 100ms. */ + public WorkflowStatus await(Duration timeout) { + return await(timeout, Duration.ofMillis(100)); + } + + /** Block until terminal, polling at {@code pollInterval}; throws on timeout. */ + public WorkflowStatus await(Duration timeout, Duration pollInterval) { + long deadline = System.nanoTime() + timeout.toNanos(); + while (true) { + WorkflowStatus status = status().orElseThrow(() -> new TaskitoException("workflow run not found: " + id)); + if (status.isTerminal()) { + return status; + } + if (System.nanoTime() >= deadline) { + throw new TaskitoException("workflow '" + id + "' did not finish within " + timeout); + } + try { + Thread.sleep(pollInterval.toMillis()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new TaskitoException("interrupted while awaiting workflow " + id, e); + } + } + } + + /** Cancel the run: skip its pending nodes and mark it cancelled. */ + public void cancel() { + backend.cancelWorkflowRun(id); + } + + private WorkflowStatus decode(String raw) { + try { + return json.readValue(raw, WorkflowStatus.class); + } catch (Exception e) { + throw new TaskitoException("failed to decode workflow status", e); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowState.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowState.java new file mode 100644 index 00000000..b72d77a9 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowState.java @@ -0,0 +1,44 @@ +package org.byteveda.taskito.workflows; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; +import java.util.Locale; + +/** State of a workflow run. Wire form is the lowercase name, shared across SDKs. */ +public enum WorkflowState { + PENDING, + RUNNING, + PAUSED, + COMPLETED, + COMPLETED_WITH_FAILURES, + FAILED, + CANCELLED, + COMPENSATING, + COMPENSATED, + COMPENSATION_FAILED; + + @JsonValue + public String wire() { + return name().toLowerCase(Locale.ROOT); + } + + @JsonCreator + public static WorkflowState fromWire(String wire) { + return valueOf(wire.toUpperCase(Locale.ROOT)); + } + + /** Whether the run has reached a final state (no further transitions). */ + public boolean isTerminal() { + switch (this) { + case COMPLETED: + case COMPLETED_WITH_FAILURES: + case FAILED: + case CANCELLED: + case COMPENSATED: + case COMPENSATION_FAILED: + return true; + default: + return false; + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowStatus.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowStatus.java new file mode 100644 index 00000000..b889b0c4 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowStatus.java @@ -0,0 +1,45 @@ +package org.byteveda.taskito.workflows; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +/** A snapshot of a workflow run's state and its nodes. Timestamps are Unix milliseconds. */ +@JsonIgnoreProperties(ignoreUnknown = true) +public final class WorkflowStatus { + public final String runId; + public final WorkflowState state; + public final Long startedAt; + public final Long completedAt; + public final String error; + public final List nodes; + + @JsonCreator + public WorkflowStatus( + @JsonProperty("runId") String runId, + @JsonProperty("state") WorkflowState state, + @JsonProperty("startedAt") Long startedAt, + @JsonProperty("completedAt") Long completedAt, + @JsonProperty("error") String error, + @JsonProperty("nodes") List nodes) { + this.runId = runId; + this.state = state; + this.startedAt = startedAt; + this.completedAt = completedAt; + this.error = error; + this.nodes = nodes == null ? Collections.emptyList() : Collections.unmodifiableList(nodes); + } + + /** Whether the run has reached a final state. */ + public boolean isTerminal() { + return state.isTerminal(); + } + + /** The named node, if present. */ + public Optional node(String nodeName) { + return nodes.stream().filter(n -> n.nodeName.equals(nodeName)).findFirst(); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java new file mode 100644 index 00000000..5efc12e6 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java @@ -0,0 +1,28 @@ +package org.byteveda.taskito.workflows; + +import org.byteveda.taskito.events.OutcomeEvent; +import org.byteveda.taskito.spi.QueueBackend; + +/** + * Advances workflow node and run state from worker outcomes. Attach via + * {@code Worker.Builder.trackWorkflows()}: on each terminal job outcome it + * records the node result (a no-op for non-workflow jobs), cascading a + * fail-fast skip and finalizing the run once every node settles. + */ +public final class WorkflowTracker { + private final QueueBackend backend; + + public WorkflowTracker(QueueBackend backend) { + this.backend = backend; + } + + /** Record a successful job as its workflow node completing. */ + public void onSuccess(OutcomeEvent event) { + backend.markWorkflowNodeResult(event.jobId, true, null, false); + } + + /** Record a dead-lettered (terminally failed) job as its workflow node failing. */ + public void onDead(OutcomeEvent event) { + backend.markWorkflowNodeResult(event.jobId, false, event.error, false); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/package-info.java new file mode 100644 index 00000000..dca33445 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/package-info.java @@ -0,0 +1,2 @@ +/** Workflow DAG definition, submission, status queries, and worker-side tracking. */ +package org.byteveda.taskito.workflows; diff --git a/sdks/java/src/test/java/org/byteveda/taskito/DashboardTest.java b/sdks/java/src/test/java/org/byteveda/taskito/DashboardTest.java index 2cfd5b94..a7293e50 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/DashboardTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/DashboardTest.java @@ -11,6 +11,8 @@ import java.util.Collections; import java.util.Map; import org.byteveda.taskito.dashboard.DashboardServer; +import org.byteveda.taskito.task.EnqueueOptions; +import org.byteveda.taskito.task.Task; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; import org.junit.jupiter.api.io.TempDir; diff --git a/sdks/java/src/test/java/org/byteveda/taskito/LockTest.java b/sdks/java/src/test/java/org/byteveda/taskito/LockTest.java index 943da487..296ce535 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/LockTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/LockTest.java @@ -5,6 +5,8 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.nio.file.Path; +import org.byteveda.taskito.locks.Lock; +import org.byteveda.taskito.scheduling.PeriodicTask; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; diff --git a/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java b/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java index 0dba5a79..c6c7b05b 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/QueueTest.java @@ -10,6 +10,12 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import org.byteveda.taskito.model.Job; +import org.byteveda.taskito.model.JobFilter; +import org.byteveda.taskito.model.JobStatus; +import org.byteveda.taskito.model.TaskLog; +import org.byteveda.taskito.task.EnqueueOptions; +import org.byteveda.taskito.task.Task; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java index 4c45fd5b..103728ac 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkerTest.java @@ -11,6 +11,8 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.byteveda.taskito.events.EventName; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; import org.junit.jupiter.api.io.TempDir; diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowTest.java new file mode 100644 index 00000000..84f1c066 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowTest.java @@ -0,0 +1,101 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicInteger; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.byteveda.taskito.workflows.NodeStatus; +import org.byteveda.taskito.workflows.Step; +import org.byteveda.taskito.workflows.Workflow; +import org.byteveda.taskito.workflows.WorkflowRun; +import org.byteveda.taskito.workflows.WorkflowState; +import org.byteveda.taskito.workflows.WorkflowStatus; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +class WorkflowTest { + + @Test + @Timeout(30) + void runsLinearDagToCompletion(@TempDir Path dir) throws Exception { + Task a = Task.of("wf.a", Integer.class); + Task b = Task.of("wf.b", Integer.class); + Task c = Task.of("wf.c", Integer.class); + try (Queue queue = + Taskito.builder().url(dir.resolve("wf.db").toString()).open()) { + Workflow wf = Workflow.named("pipeline") + .step("a", a, 1) + .step("b", b, 2, "a") + .step("c", c, 3, "b"); + + WorkflowRun run = queue.submitWorkflow(wf); + AtomicInteger ran = new AtomicInteger(); + try (Worker worker = queue.worker() + .handle(a, p -> ran.incrementAndGet()) + .handle(b, p -> ran.incrementAndGet()) + .handle(c, p -> ran.incrementAndGet()) + .trackWorkflows() + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.COMPLETED, status.state); + assertEquals(3, ran.get()); + assertEquals(NodeStatus.COMPLETED, status.node("c").orElseThrow().status); + } + } + } + + @Test + @Timeout(30) + void failingNodeFailsRunAndSkipsDownstream(@TempDir Path dir) throws Exception { + Task a = Task.of("fc.a", Integer.class); + Task b = Task.of("fc.b", Integer.class); + Task c = Task.of("fc.c", Integer.class); + try (Queue queue = + Taskito.builder().url(dir.resolve("fc.db").toString()).open()) { + Workflow wf = Workflow.named("failpipe") + .step("a", a, 1) + .step(Step.of("b", b, 2).after("a").maxRetries(0).build()) + .step("c", c, 3, "b"); + + WorkflowRun run = queue.submitWorkflow(wf); + AtomicInteger cRan = new AtomicInteger(); + try (Worker worker = queue.worker() + .handle(a, p -> p) + .handle(b, p -> { + throw new IllegalStateException("boom"); + }) + .handle(c, p -> cRan.incrementAndGet()) + .trackWorkflows() + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.FAILED, status.state); + assertEquals(NodeStatus.FAILED, status.node("b").orElseThrow().status); + assertEquals(NodeStatus.SKIPPED, status.node("c").orElseThrow().status); + assertEquals(0, cRan.get()); + } + } + } + + @Test + @Timeout(30) + void cancelSkipsPendingNodes(@TempDir Path dir) { + Task a = Task.of("cx.a", Integer.class); + Task b = Task.of("cx.b", Integer.class); + try (Queue queue = + Taskito.builder().url(dir.resolve("cx.db").toString()).open()) { + Workflow wf = Workflow.named("cancelpipe").step("a", a, 1).step("b", b, 2, "a"); + WorkflowRun run = queue.submitWorkflow(wf); + run.cancel(); + + WorkflowStatus status = run.status().orElseThrow(); + assertEquals(WorkflowState.CANCELLED, status.state); + assertTrue(status.node("a").orElseThrow().status == NodeStatus.SKIPPED); + assertTrue(status.node("b").orElseThrow().status == NodeStatus.SKIPPED); + } + } +}