From aff761ce02874b98c57e1b498f803a768e266dc1 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:27:22 +0530 Subject: [PATCH 01/11] feat(java): bind fan-out/fan-in workflow primitives over JNI --- crates/taskito-java/src/workflows/mod.rs | 413 ++++++++++++++++++++++- 1 file changed, 407 insertions(+), 6 deletions(-) diff --git a/crates/taskito-java/src/workflows/mod.rs b/crates/taskito-java/src/workflows/mod.rs index a2a453da..98b772d1 100644 --- a/crates/taskito-java/src/workflows/mod.rs +++ b/crates/taskito-java/src/workflows/mod.rs @@ -3,14 +3,16 @@ //! 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. +//! and drives the run forward. Deferred steps (fan-out / fan-in, and anything +//! downstream of them) carry a node row but no job at submit; the tracker +//! expands a fan-out into per-item child jobs and collects them into a fan-in +//! job at runtime via [`expand_fan_out`] / [`create_deferred_job`]. Gates and +//! sub-workflows are not produced by the Java builder yet. use std::collections::{HashMap, HashSet}; -use jni::objects::{JClass, JObjectArray, JString}; -use jni::sys::{jboolean, jint, jlong, jstring, JNI_FALSE}; +use jni::objects::{JByteArray, JClass, JObjectArray, JString}; +use jni::sys::{jboolean, jint, jlong, jobjectArray, jstring, JNI_FALSE}; use jni::JNIEnv; use serde::{Deserialize, Serialize}; use taskito_core::job::{now_millis, NewJob}; @@ -26,12 +28,16 @@ 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, + guard, new_string, new_string_array, read_bytes, 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; +/// Largest number of children one fan-out may expand into — guards against a +/// producer returning an enormous list and flooding storage in one batch. +const MAX_FAN_OUT: usize = 10_000; /// One step as described by the Java builder. Edges come from `after`. #[derive(Deserialize)] @@ -45,6 +51,10 @@ struct StepSpec { max_retries: Option, timeout_ms: Option, priority: Option, + /// Fan-out strategy (e.g. "each") — this node expands per predecessor result item. + fan_out: Option, + /// Fan-in strategy (e.g. "all") — this node collects its fan-out children's results. + fan_in: Option, } /// Combined run + node snapshot returned by `getWorkflowStatus`. Timestamps are @@ -194,6 +204,8 @@ fn submit( max_retries: s.max_retries, timeout_ms: s.timeout_ms, priority: s.priority, + fan_out: s.fan_out.clone(), + fan_in: s.fan_in.clone(), ..Default::default() }, ) @@ -465,6 +477,395 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_cancel }) } +/// One node of a run's plan: its predecessors plus the step metadata the tracker +/// needs to enqueue deferred (fan-out / fan-in) work. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct PlanNodeView<'a> { + name: &'a str, + predecessors: &'a [String], + task_name: Option<&'a str>, + queue: Option<&'a str>, + max_retries: Option, + timeout_ms: Option, + priority: Option, + fan_out: Option<&'a str>, + fan_in: Option<&'a str>, +} + +/// The run + node a job belongs to. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct NodeRefView<'a> { + run_id: &'a str, + node_name: &'a str, +} + +/// Outcome of a settled fan-out parent. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct FanOutCompletionView { + succeeded: bool, + child_job_ids: Vec, +} + +/// `String getWorkflowPlan(long, String runId)` — the run's nodes with their +/// predecessors and step metadata (JSON array), or `null` if the run is absent. +/// The tracker inverts predecessors into successors to route deferred work. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWorkflowPlan< + '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 def = match wf.get_workflow_definition_by_id(&run.definition_id)? { + Some(d) => d, + None => return Ok(std::ptr::null_mut()), + }; + let ordered = topological_order(&def.dag_data)?; + let views: Vec = ordered + .iter() + .map(|node| { + let meta = def.step_metadata.get(&node.name); + PlanNodeView { + name: &node.name, + predecessors: &node.predecessors, + task_name: meta.map(|m| m.task_name.as_str()), + queue: meta.and_then(|m| m.queue.as_deref()), + max_retries: meta.and_then(|m| m.max_retries), + timeout_ms: meta.and_then(|m| m.timeout_ms), + priority: meta.and_then(|m| m.priority), + fan_out: meta.and_then(|m| m.fan_out.as_deref()), + fan_in: meta.and_then(|m| m.fan_in.as_deref()), + } + }) + .collect(); + new_string(env, to_json(&views)?) + }) +} + +/// `String workflowNodeForJob(long, String jobId)` — the run + node a job +/// belongs to (JSON), or `null` for a non-workflow job. Lets the tracker classify +/// any worker outcome without an in-memory job→run map. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_workflowNodeForJob< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + job_id: JString<'local>, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let queue = unsafe { borrow_queue(handle) }; + let job_id = read_string(env, &job_id)?; + let job = match queue.storage.get_job(&job_id)? { + Some(j) => j, + None => return Ok(std::ptr::null_mut()), + }; + match parse_workflow_metadata(job.metadata.as_deref()) { + Some((run_id, node_name)) => new_string( + env, + to_json(&NodeRefView { + run_id: &run_id, + node_name: &node_name, + })?, + ), + None => Ok(std::ptr::null_mut()), + } + }) +} + +/// `String[] expandFanOut(long, String runId, String parentNode, String[] +/// childNames, byte[][] childPayloads, String taskName, String queue, +/// int maxRetries, long timeoutMs, int priority)` — expand a fan-out parent into +/// one child node + job per item. Returns the child job ids. +#[no_mangle] +#[allow(clippy::too_many_arguments)] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_expandFanOut<'local>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + parent_node: JString<'local>, + child_names: JObjectArray<'local>, + child_payloads: JObjectArray<'local>, + task_name: JString<'local>, + queue: JString<'local>, + max_retries: jint, + timeout_ms: jlong, + priority: jint, +) -> jobjectArray { + guard(&mut env, std::ptr::null_mut(), |env| { + let q = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let parent = read_string(env, &parent_node)?; + let names = read_string_array(env, &child_names)?; + let payloads = read_bytes_array(env, &child_payloads)?; + let task = read_string(env, &task_name)?; + let queue_name = read_string(env, &queue)?; + let ids = expand_fan_out( + q, + &run_id, + &parent, + names, + payloads, + &task, + &queue_name, + max_retries, + timeout_ms, + priority, + )?; + new_string_array(env, &ids) + }) +} + +#[allow(clippy::too_many_arguments)] +fn expand_fan_out( + queue: &QueueHandle, + run_id: &str, + parent: &str, + child_names: Vec, + child_payloads: Vec>, + task_name: &str, + queue_name: &str, + max_retries: i32, + timeout_ms: i64, + priority: i32, +) -> Result, BindingError> { + if child_names.len() != child_payloads.len() { + return Err(BindingError::new( + "childNames and childPayloads must have equal length", + )); + } + if child_names.len() > MAX_FAN_OUT { + return Err(BindingError::new(format!( + "fan-out of {} children exceeds the limit of {MAX_FAN_OUT}", + child_names.len() + ))); + } + let wf = queue.workflow_store()?; + let now = now_millis(); + let count = child_names.len() as i32; + if count == 0 { + wf.set_workflow_node_fan_out_count(run_id, parent, 0)?; + wf.set_workflow_node_completed(run_id, parent, now, None)?; + return Ok(Vec::new()); + } + + // Enqueue every child job first, then batch-insert their nodes in one + // transaction so a crash partway never leaves half-tracked children. + let mut child_job_ids = Vec::with_capacity(child_names.len()); + let mut nodes = Vec::with_capacity(child_names.len()); + for (child_name, payload) in child_names.iter().zip(child_payloads) { + let job = queue.storage.enqueue(NewJob { + queue: queue_name.to_string(), + task_name: task_name.to_string(), + payload, + priority, + scheduled_at: now, + max_retries, + timeout_ms, + unique_key: None, + metadata: Some(workflow_metadata_json(run_id, child_name)), + notes: None, + depends_on: vec![], + expires_at: None, + result_ttl_ms: None, + namespace: queue.namespace.clone(), + })?; + child_job_ids.push(job.id.clone()); + nodes.push(new_node(run_id, child_name, Some(job.id))); + } + // Children are enqueued; if node creation fails, cancel them so they don't + // run untracked outside the workflow. + if let Err(err) = wf.create_workflow_nodes_batch(&nodes) { + for id in &child_job_ids { + let _ = queue.storage.cancel_job(id); + } + return Err(err.into()); + } + wf.set_workflow_node_fan_out_count(run_id, parent, count)?; + Ok(child_job_ids) +} + +/// `String checkFanOutCompletion(long, String runId, String parentNode)` — when +/// every fan-out child is terminal, atomically finalize the parent (once) and +/// return `{succeeded, childJobIds}` (JSON); otherwise `null`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_checkFanOutCompletion< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + parent_node: 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 parent = read_string(env, &parent_node)?; + let wf = queue.workflow_store()?; + let prefix = format!("{parent}["); + let children = wf.get_workflow_nodes_by_prefix(&run_id, &prefix)?; + if children.is_empty() || !children.iter().all(|n| n.status.is_terminal()) { + return Ok(std::ptr::null_mut()); + } + let any_failed = children + .iter() + .any(|n| n.status == WorkflowNodeStatus::Failed); + let child_job_ids: Vec = children.iter().filter_map(|n| n.job_id.clone()).collect(); + let transitioned = wf.finalize_fan_out_parent( + &run_id, + &parent, + !any_failed, + if any_failed { + Some("fan-out child failed") + } else { + None + }, + now_millis(), + )?; + if !transitioned { + return Ok(std::ptr::null_mut()); + } + new_string( + env, + to_json(&FanOutCompletionView { + succeeded: !any_failed, + child_job_ids, + })?, + ) + }) +} + +/// `String createDeferredJob(long, String runId, String nodeName, byte[] payload, +/// String taskName, String queue, int maxRetries, long timeoutMs, int priority)` +/// — enqueue a job for a deferred node (e.g. the fan-in collector) and bind it to +/// that node. Returns the new job id. +#[no_mangle] +#[allow(clippy::too_many_arguments)] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_createDeferredJob< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + node_name: JString<'local>, + payload: JByteArray<'local>, + task_name: JString<'local>, + queue: JString<'local>, + max_retries: jint, + timeout_ms: jlong, + priority: jint, +) -> jstring { + guard(&mut env, std::ptr::null_mut(), |env| { + let q = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let node_name = read_string(env, &node_name)?; + let bytes = read_bytes(env, &payload)?; + let task = read_string(env, &task_name)?; + let queue_name = read_string(env, &queue)?; + let wf = q.workflow_store()?; + let job = q.storage.enqueue(NewJob { + queue: queue_name, + task_name: task, + payload: bytes, + priority, + scheduled_at: now_millis(), + max_retries, + timeout_ms, + unique_key: None, + metadata: Some(workflow_metadata_json(&run_id, &node_name)), + notes: None, + depends_on: vec![], + expires_at: None, + result_ttl_ms: None, + namespace: q.namespace.clone(), + })?; + // Cancel the job if binding fails, so it can't run untracked. + if let Err(err) = wf.set_workflow_node_job(&run_id, &node_name, &job.id) { + let _ = q.storage.cancel_job(&job.id); + return Err(err.into()); + } + new_string(env, job.id) + }) +} + +/// `void cascadeSkipPending(long, String runId)` — fail-fast: skip every +/// still-pending node of a run and cancel its job. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_cascadeSkipPending< + '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()?; + let nodes = wf.get_workflow_nodes(&run_id)?; + cascade_skip_pending(queue, &wf, &run_id, &nodes); + Ok(()) + }) +} + +/// `String finalizeRunIfTerminal(long, String runId)` — if every node is +/// terminal, set the run `completed` (or `failed`) and return that state; else +/// `null`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_finalizeRunIfTerminal< + '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 nodes = wf.get_workflow_nodes(&run_id)?; + if nodes.is_empty() || !nodes.iter().all(|n| n.status.is_terminal()) { + return Ok(std::ptr::null_mut()); + } + let any_failed = nodes.iter().any(|n| n.status == WorkflowNodeStatus::Failed); + let final_state = if any_failed { + WorkflowState::Failed + } else { + WorkflowState::Completed + }; + wf.update_workflow_run_state( + &run_id, + final_state, + if any_failed { + Some("a workflow node failed") + } else { + None + }, + )?; + wf.set_workflow_run_completed(&run_id, now_millis())?; + new_string(env, final_state.as_str().to_string()) + }) +} + /// 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 { From 1aa42ceb2ed7e30cf36709f0e72b5e2b3be430d0 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:27:23 +0530 Subject: [PATCH 02/11] feat(java): expose fan-out native methods on the backend SPI --- .../taskito/internal/JniQueueBackend.java | 54 +++++++++++++++++++ .../taskito/internal/NativeWorkflows.java | 41 ++++++++++++++ .../byteveda/taskito/spi/QueueBackend.java | 31 +++++++++++ 3 files changed, 126 insertions(+) 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 ba1be7d1..c59c64e6 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 @@ -219,6 +219,60 @@ public void cancelWorkflowRun(String runId) { NativeWorkflows.cancelWorkflowRun(handle, runId); } + @Override + public Optional getWorkflowPlanJson(String runId) { + return Optional.ofNullable(NativeWorkflows.getWorkflowPlan(handle, runId)); + } + + @Override + public Optional workflowNodeForJobJson(String jobId) { + return Optional.ofNullable(NativeWorkflows.workflowNodeForJob(handle, jobId)); + } + + @Override + public String[] expandFanOut( + String runId, + String parentNode, + String[] childNames, + byte[][] childPayloads, + String taskName, + String queue, + int maxRetries, + long timeoutMs, + int priority) { + return NativeWorkflows.expandFanOut( + handle, runId, parentNode, childNames, childPayloads, taskName, queue, maxRetries, timeoutMs, priority); + } + + @Override + public Optional checkFanOutCompletionJson(String runId, String parentNode) { + return Optional.ofNullable(NativeWorkflows.checkFanOutCompletion(handle, runId, parentNode)); + } + + @Override + public String createDeferredJob( + String runId, + String nodeName, + byte[] payload, + String taskName, + String queue, + int maxRetries, + long timeoutMs, + int priority) { + return NativeWorkflows.createDeferredJob( + handle, runId, nodeName, payload, taskName, queue, maxRetries, timeoutMs, priority); + } + + @Override + public void cascadeSkipPending(String runId) { + NativeWorkflows.cascadeSkipPending(handle, runId); + } + + @Override + public Optional finalizeRunIfTerminal(String runId) { + return Optional.ofNullable(NativeWorkflows.finalizeRunIfTerminal(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 index ed5fa499..e3a7c77d 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.java @@ -35,4 +35,45 @@ public static native String markWorkflowNodeResult( public static native String getWorkflowStatus(long handle, String runId); public static native void cancelWorkflowRun(long handle, String runId); + + // ── Fan-out / fan-in orchestration (driven by the worker-side tracker) ── + + /** Returns the run's nodes with predecessors + step metadata (JSON), or {@code null}. */ + public static native String getWorkflowPlan(long handle, String runId); + + /** Returns {@code {runId, nodeName}} for a job (JSON), or {@code null} if non-workflow. */ + public static native String workflowNodeForJob(long handle, String jobId); + + /** Expand a fan-out parent into one child job per payload; returns the child job ids. */ + public static native String[] expandFanOut( + long handle, + String runId, + String parentNode, + String[] childNames, + byte[][] childPayloads, + String taskName, + String queue, + int maxRetries, + long timeoutMs, + int priority); + + /** Returns {@code {succeeded, childJobIds}} once all children settle (JSON), else {@code null}. */ + public static native String checkFanOutCompletion(long handle, String runId, String parentNode); + + /** Enqueue a job for a deferred node (e.g. the fan-in collector); returns the job id. */ + public static native String createDeferredJob( + long handle, + String runId, + String nodeName, + byte[] payload, + String taskName, + String queue, + int maxRetries, + long timeoutMs, + int priority); + + public static native void cascadeSkipPending(long handle, String runId); + + /** Finalize the run if every node is terminal; returns the final state, or {@code null}. */ + public static native String finalizeRunIfTerminal(long handle, String runId); } 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 d8fc29ae..f26aaed8 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 @@ -125,6 +125,37 @@ String submitWorkflow( void cancelWorkflowRun(String runId); + Optional getWorkflowPlanJson(String runId); + + Optional workflowNodeForJobJson(String jobId); + + String[] expandFanOut( + String runId, + String parentNode, + String[] childNames, + byte[][] childPayloads, + String taskName, + String queue, + int maxRetries, + long timeoutMs, + int priority); + + Optional checkFanOutCompletionJson(String runId, String parentNode); + + String createDeferredJob( + String runId, + String nodeName, + byte[] payload, + String taskName, + String queue, + int maxRetries, + long timeoutMs, + int priority); + + void cascadeSkipPending(String runId); + + Optional finalizeRunIfTerminal(String runId); + // ── Worker ────────────────────────────────────────────────────── /** Start a worker that dispatches jobs to {@code bridge}; returns its control. */ WorkerControl startWorker(WorkerBridge bridge, String optionsJson); From 10d6cc0e1c59f609b5826d2a25c27d234f39ad53 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:27:23 +0530 Subject: [PATCH 03/11] feat(java): add fan-out/fan-in to the workflow builder --- .../org/byteveda/taskito/DefaultQueue.java | 54 ++++++++++++++++--- .../org/byteveda/taskito/workflows/Step.java | 23 ++++++++ .../byteveda/taskito/workflows/Workflow.java | 16 ++++++ 3 files changed, 86 insertions(+), 7 deletions(-) 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 9a527a41..a5cef92e 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java @@ -4,10 +4,12 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.util.ArrayList; import java.util.Arrays; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; import org.byteveda.taskito.locks.Lock; import org.byteveda.taskito.locks.LockInfo; @@ -281,20 +283,52 @@ public long registerPeriodic(PeriodicTask task) { @Override public WorkflowRun submitWorkflow(Workflow workflow) { List steps = workflow.steps(); + Set deferred = deferredNodes(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); + // Deferred nodes (fan-out/fan-in + their downstream) have no job — and so + // no payload — at submit; the tracker enqueues them at runtime. + List payloadNames = new ArrayList<>(); + List payloads = new ArrayList<>(); + for (Step step : steps) { specs.add(stepSpec(step)); - names[i] = step.name; - payloads[i] = serializer.serialize(step.payload); + if (!deferred.contains(step.name)) { + payloadNames.add(step.name); + payloads.add(serializer.serialize(step.payload)); + } } String runId = backend.submitWorkflow( - workflow.name(), workflow.version(), encode(specs), names, payloads, null, null, new String[0]); + workflow.name(), + workflow.version(), + encode(specs), + payloadNames.toArray(new String[0]), + payloads.toArray(new byte[0][]), + null, + null, + deferred.toArray(new String[0])); return new WorkflowRun(backend, VIEWS, runId, workflow.name()); } + /** Fan-out/fan-in nodes, plus everything transitively downstream of them, are deferred. */ + private static Set deferredNodes(List steps) { + Set deferred = new HashSet<>(); + for (Step step : steps) { + if (step.fanOut != null || step.fanIn != null) { + deferred.add(step.name); + } + } + boolean changed = true; + while (changed) { + changed = false; + for (Step step : steps) { + if (!deferred.contains(step.name) && step.after.stream().anyMatch(deferred::contains)) { + deferred.add(step.name); + changed = true; + } + } + } + return deferred; + } + @Override public Optional workflowStatus(String runId) { return backend.getWorkflowStatusJson(runId).map(json -> decode(json, WorkflowStatus.class)); @@ -323,6 +357,12 @@ private static Map stepSpec(Step step) { if (step.priority != null) { spec.put("priority", step.priority); } + if (step.fanOut != null) { + spec.put("fanOut", step.fanOut); + } + if (step.fanIn != null) { + spec.put("fanIn", step.fanIn); + } return spec; } 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 index ea1e7575..79aa5476 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java @@ -16,6 +16,8 @@ public final class Step { public final Integer maxRetries; public final Long timeoutMs; public final Integer priority; + public final String fanOut; + public final String fanIn; private Step(Builder builder) { this.name = builder.name; @@ -26,6 +28,8 @@ private Step(Builder builder) { this.maxRetries = builder.maxRetries; this.timeoutMs = builder.timeoutMs; this.priority = builder.priority; + this.fanOut = builder.fanOut; + this.fanIn = builder.fanIn; } /** Begin a step bound to a typed task. */ @@ -38,6 +42,11 @@ public static Builder of(String name, String taskName, Object payload) { return new Builder(name, taskName, payload); } + /** Begin a payload-less step (its payload is derived at runtime — fan-out/fan-in). */ + public static Builder of(String name, Task task) { + return new Builder(name, task.name(), null); + } + /** Fluent builder for a {@link Step}. */ public static final class Builder { private final String name; @@ -48,6 +57,8 @@ public static final class Builder { private Integer maxRetries; private Long timeoutMs; private Integer priority; + private String fanOut; + private String fanIn; private Builder(String name, String taskName, Object payload) { this.name = name; @@ -81,6 +92,18 @@ public Builder priority(int priority) { return this; } + /** Run this step once per item of its predecessor's result (strategy {@code "each"}). */ + public Builder fanOut(String strategy) { + this.fanOut = strategy; + return this; + } + + /** Collect a fan-out predecessor's child results into one list (strategy {@code "all"}). */ + public Builder fanIn(String strategy) { + this.fanIn = strategy; + 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 index 14e22dd1..97d75063 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java @@ -41,6 +41,22 @@ public Workflow step(Step step) { return this; } + /** + * Add a fan-out step: {@code task} runs once per item of its predecessor's + * result (a list). The predecessor named in {@code after} is the producer. + */ + public Workflow fanOut(String name, Task task, String strategy, String... after) { + return step(Step.of(name, task).fanOut(strategy).after(after).build()); + } + + /** + * Add a fan-in step that collects its fan-out predecessor's child results + * into one list and passes it to {@code task}. + */ + public Workflow fanIn(String name, Task task, String strategy, String... after) { + return step(Step.of(name, task).fanIn(strategy).after(after).build()); + } + public String name() { return name; } From 469ac003509918e1ca57e2c44da6898364b9c0c9 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:27:23 +0530 Subject: [PATCH 04/11] feat(java): drive fan-out/fan-in from a stateful tracker --- .../org/byteveda/taskito/worker/Worker.java | 2 +- .../byteveda/taskito/workflows/PlanNode.java | 59 ++++++ .../byteveda/taskito/workflows/RunPlan.java | 46 ++++ .../taskito/workflows/WorkflowTracker.java | 196 +++++++++++++++++- 4 files changed, 294 insertions(+), 9 deletions(-) create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java create mode 100644 sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java diff --git a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java index 37df4f71..39b6aae9 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java @@ -141,7 +141,7 @@ public Builder on(EventName name, Consumer listener) { /** Drive workflow node and run state from this worker's job outcomes. */ public Builder trackWorkflows() { - WorkflowTracker tracker = new WorkflowTracker(backend); + WorkflowTracker tracker = new WorkflowTracker(backend, serializer); on(EventName.SUCCESS, tracker::onSuccess); on(EventName.DEAD, tracker::onDead); return this; diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java new file mode 100644 index 00000000..cdd07861 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java @@ -0,0 +1,59 @@ +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; + +/** One node of a run's plan: predecessors plus the metadata needed to enqueue deferred work. */ +@JsonIgnoreProperties(ignoreUnknown = true) +final class PlanNode { + final String name; + final List predecessors; + final String taskName; + final String queue; + final Integer maxRetries; + final Long timeoutMs; + final Integer priority; + final String fanOut; + final String fanIn; + + @JsonCreator + PlanNode( + @JsonProperty("name") String name, + @JsonProperty("predecessors") List predecessors, + @JsonProperty("taskName") String taskName, + @JsonProperty("queue") String queue, + @JsonProperty("maxRetries") Integer maxRetries, + @JsonProperty("timeoutMs") Long timeoutMs, + @JsonProperty("priority") Integer priority, + @JsonProperty("fanOut") String fanOut, + @JsonProperty("fanIn") String fanIn) { + this.name = name; + this.predecessors = predecessors == null ? Collections.emptyList() : predecessors; + this.taskName = taskName; + this.queue = queue; + this.maxRetries = maxRetries; + this.timeoutMs = timeoutMs; + this.priority = priority; + this.fanOut = fanOut; + this.fanIn = fanIn; + } + + int retries() { + return maxRetries == null ? 3 : maxRetries; + } + + long timeout() { + return timeoutMs == null ? 300_000L : timeoutMs; + } + + int priorityOrDefault() { + return priority == null ? 0 : priority; + } + + String queueOrDefault() { + return queue == null ? "default" : queue; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java new file mode 100644 index 00000000..99237646 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java @@ -0,0 +1,46 @@ +package org.byteveda.taskito.workflows; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Predicate; + +/** A run's DAG: nodes by name plus successor adjacency, built from the native plan. */ +final class RunPlan { + private final Map byName; + private final Map> successors; + + private RunPlan(Map byName, Map> successors) { + this.byName = byName; + this.successors = successors; + } + + /** Build from the plan node list, inverting predecessors into successor adjacency. */ + static RunPlan from(List nodes) { + Map byName = new HashMap<>(); + Map> successors = new HashMap<>(); + for (PlanNode node : nodes) { + byName.put(node.name, node); + } + for (PlanNode node : nodes) { + for (String predecessor : node.predecessors) { + successors + .computeIfAbsent(predecessor, key -> new ArrayList<>()) + .add(node.name); + } + } + return new RunPlan(byName, successors); + } + + /** The first successor of {@code node} whose plan entry matches {@code match}, or null. */ + PlanNode successorMatching(String node, Predicate match) { + for (String successor : successors.getOrDefault(node, List.of())) { + PlanNode candidate = byName.get(successor); + if (candidate != null && match.test(candidate)) { + return candidate; + } + } + return null; + } +} 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 index 5efc12e6..d5a852a3 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java @@ -1,28 +1,208 @@ package org.byteveda.taskito.workflows; +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.databind.JavaType; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import org.byteveda.taskito.TaskitoException; import org.byteveda.taskito.events.OutcomeEvent; +import org.byteveda.taskito.serialization.Serializer; 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. + * {@code Worker.Builder.trackWorkflows()}. + * + *

On each terminal job outcome it records the node result, then drives the + * run forward: a static node simply finalizes the run when all nodes settle; a + * fan-out producer's result is split into per-item child jobs; a settled fan-out + * is collected into one fan-in job. Each run's DAG is loaded lazily from storage + * (the plan) and cached until the run reaches a terminal state. */ public final class WorkflowTracker { private final QueueBackend backend; + private final Serializer serializer; + private final ObjectMapper json = new ObjectMapper(); + private final ConcurrentMap plans = new ConcurrentHashMap<>(); - public WorkflowTracker(QueueBackend backend) { + public WorkflowTracker(QueueBackend backend, Serializer serializer) { this.backend = backend; + this.serializer = serializer; } - /** Record a successful job as its workflow node completing. */ + /** Route a successful job outcome. */ public void onSuccess(OutcomeEvent event) { - backend.markWorkflowNodeResult(event.jobId, true, null, false); + onOutcome(event.jobId, true, null); } - /** Record a dead-lettered (terminally failed) job as its workflow node failing. */ + /** Route a dead-lettered (terminally failed) job outcome. */ public void onDead(OutcomeEvent event) { - backend.markWorkflowNodeResult(event.jobId, false, event.error, false); + onOutcome(event.jobId, false, event.error); + } + + private void onOutcome(String jobId, boolean succeeded, String error) { + NodeRef ref = backend.workflowNodeForJobJson(jobId) + .map(raw -> decode(raw, NodeRef.class)) + .orElse(null); + if (ref == null) { + return; // not a workflow job + } + // Record the node outcome; the tracker (not the core) drives cascade + finalize. + backend.markWorkflowNodeResult(jobId, succeeded, error, true); + RunPlan plan = plans.computeIfAbsent(ref.runId, this::loadPlan); + + if (ref.nodeName.indexOf('[') >= 0) { + handleFanOutChild(ref.runId, ref.nodeName, plan); + return; + } + if (succeeded && plan != null) { + expandFanOutIfAny(ref.runId, ref.nodeName, jobId, plan); + } else if (!succeeded) { + backend.cascadeSkipPending(ref.runId); + } + finalizeIfTerminal(ref.runId); + } + + /** If the just-completed node feeds a fan-out, split its result list into child jobs. */ + private void expandFanOutIfAny(String runId, String node, String jobId, RunPlan plan) { + PlanNode fanOut = plan.successorMatching(node, candidate -> candidate.fanOut != null); + if (fanOut == null) { + return; + } + byte[] result = fetchResult(jobId); + if (result == null) { + return; + } + List items = serializer.deserialize(result, List.class); + String[] names = new String[items.size()]; + byte[][] payloads = new byte[items.size()][]; + for (int i = 0; i < items.size(); i++) { + names[i] = fanOut.name + "[" + i + "]"; + payloads[i] = serializer.serialize(items.get(i)); + } + backend.expandFanOut( + runId, + fanOut.name, + names, + payloads, + fanOut.taskName, + fanOut.queueOrDefault(), + fanOut.retries(), + fanOut.timeout(), + fanOut.priorityOrDefault()); + } + + /** When every fan-out child has settled, collect their results into one fan-in job. */ + private void handleFanOutChild(String runId, String childName, RunPlan plan) { + String parent = childName.substring(0, childName.indexOf('[')); + FanOutCompletion done = backend.checkFanOutCompletionJson(runId, parent) + .map(raw -> decode(raw, FanOutCompletion.class)) + .orElse(null); + if (done == null) { + return; // siblings still running, or already handled by a concurrent outcome + } + if (!done.succeeded) { + backend.cascadeSkipPending(runId); + finalizeIfTerminal(runId); + return; + } + PlanNode fanIn = plan == null ? null : plan.successorMatching(parent, candidate -> candidate.fanIn != null); + if (fanIn == null) { + finalizeIfTerminal(runId); + return; + } + List results = new ArrayList<>(done.childJobIds.size()); + for (String childJobId : done.childJobIds) { + byte[] bytes = fetchResult(childJobId); + results.add(bytes == null ? null : serializer.deserialize(bytes, Object.class)); + } + backend.createDeferredJob( + runId, + fanIn.name, + serializer.serialize(results), + fanIn.taskName, + fanIn.queueOrDefault(), + fanIn.retries(), + fanIn.timeout(), + fanIn.priorityOrDefault()); + } + + private void finalizeIfTerminal(String runId) { + if (backend.finalizeRunIfTerminal(runId).isPresent()) { + plans.remove(runId); // run reached a terminal state — drop its cached plan + } + } + + /** Read a job's serialized result, briefly polling in case the event outran the DB write. */ + private byte[] fetchResult(String jobId) { + for (int attempt = 0; attempt < 50; attempt++) { + Optional result = backend.getResult(jobId); + if (result.isPresent()) { + return result.get(); + } + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } + } + return null; + } + + private RunPlan loadPlan(String runId) { + return backend.getWorkflowPlanJson(runId) + .map(raw -> RunPlan.from(decodeList(raw))) + .orElse(null); + } + + private T decode(String raw, Class type) { + try { + return json.readValue(raw, type); + } catch (Exception e) { + throw new TaskitoException("failed to decode workflow data", e); + } + } + + private List decodeList(String raw) { + JavaType type = json.getTypeFactory().constructCollectionType(List.class, PlanNode.class); + try { + return json.readValue(raw, type); + } catch (Exception e) { + throw new TaskitoException("failed to decode workflow plan", e); + } + } + + /** {@code {runId, nodeName}} from the native node-ref view. */ + @JsonIgnoreProperties(ignoreUnknown = true) + private static final class NodeRef { + final String runId; + final String nodeName; + + @JsonCreator + NodeRef(@JsonProperty("runId") String runId, @JsonProperty("nodeName") String nodeName) { + this.runId = runId; + this.nodeName = nodeName; + } + } + + /** {@code {succeeded, childJobIds}} from the native fan-out completion view. */ + @JsonIgnoreProperties(ignoreUnknown = true) + private static final class FanOutCompletion { + final boolean succeeded; + final List childJobIds; + + @JsonCreator + FanOutCompletion( + @JsonProperty("succeeded") boolean succeeded, @JsonProperty("childJobIds") List childJobIds) { + this.succeeded = succeeded; + this.childJobIds = childJobIds == null ? List.of() : childJobIds; + } } } From d9a761858f7672dd5001ead5891ea93c38ac0e35 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:27:24 +0530 Subject: [PATCH 05/11] test(java): cover fan-out/fan-in execution --- .../byteveda/taskito/WorkflowFanOutTest.java | 70 +++++++++++++++++++ 1 file changed, 70 insertions(+) create mode 100644 sdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java new file mode 100644 index 00000000..b667eb27 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowFanOutTest.java @@ -0,0 +1,70 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.byteveda.taskito.workflows.NodeStatus; +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 WorkflowFanOutTest { + + @Test + @Timeout(30) + void fansOutPerItemAndFansInResults(@TempDir Path dir) throws Exception { + Task seed = Task.of("fo.seed", Integer.class); + Task square = Task.of("fo.square", Integer.class); + Task sum = Task.of("fo.sum", List.class); + try (Queue queue = + Taskito.builder().url(dir.resolve("fo.db").toString()).open()) { + // seed(4) -> [1,2,3,4]; square each -> [1,4,9,16]; sum all -> 30 + Workflow wf = Workflow.named("fanpipe") + .step("seed", seed, 4) + .fanOut("square", square, "each", "seed") + .fanIn("sum", sum, "all", "square"); + + WorkflowRun run = queue.submitWorkflow(wf); + try (Worker worker = queue.worker() + .handle(seed, n -> { + List out = new ArrayList<>(); + for (int i = 1; i <= n; i++) { + out.add(i); + } + return out; + }) + .handle(square, x -> x * x) + .handle(sum, (List xs) -> { + int total = 0; + for (Object x : xs) { + total += ((Number) x).intValue(); + } + return total; + }) + .trackWorkflows() + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.COMPLETED, status.state); + assertEquals(NodeStatus.COMPLETED, status.node("square").orElseThrow().status); + assertEquals(NodeStatus.COMPLETED, status.node("sum").orElseThrow().status); + + long children = status.nodes.stream() + .filter(node -> node.nodeName.startsWith("square[")) + .count(); + assertEquals(4, children); + + String sumJob = status.node("sum").orElseThrow().jobId; + assertEquals(30, queue.getResult(sumJob, Integer.class).orElseThrow()); + } + } + } +} From 09d1bb85227aa4dc349f52e067b1a490457dadfb Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:27:24 +0530 Subject: [PATCH 06/11] docs(java): document fan-out/fan-in --- sdks/java/README.md | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/sdks/java/README.md b/sdks/java/README.md index 0c8544a3..cb97dd79 100644 --- a/sdks/java/README.md +++ b/sdks/java/README.md @@ -95,8 +95,20 @@ try (Worker worker = queue.worker() ``` 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. +`run.cancel()` skips pending nodes. + +**Fan-out / fan-in** map a step over a producer's result list and gather the +results: + +```java +Workflow wf = Workflow.named("pipeline") + .step("seed", seed, 4) // returns List.of(1,2,3,4) + .fanOut("square", square, "each", "seed") // runs square(x) per item + .fanIn("sum", sum, "all", "square"); // sum receives [1,4,9,16] +``` + +`trackWorkflows()` must be attached for fan-out expansion and aggregation. +Conditions, gates, and sagas are not yet exposed. ### Middleware From 39f529efe636ad65b008a317308e7f1aefdf0f3f Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:50:24 +0530 Subject: [PATCH 07/11] refactor(java): make workflow SPI methods default-throwing --- .../byteveda/taskito/spi/QueueBackend.java | 54 ++++++++++++++----- 1 file changed, 40 insertions(+), 14 deletions(-) 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 f26aaed8..b0766eda 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 @@ -108,7 +108,11 @@ default long registerPeriodic( } // ── Workflows ─────────────────────────────────────────────────── - String submitWorkflow( + // Optional capability: default to throwing so existing custom backends keep + // compiling and fail explicitly only when workflows are actually used. + String WORKFLOWS_UNSUPPORTED = "workflows not supported by this backend"; + + default String submitWorkflow( String name, int version, String stepsJson, @@ -116,20 +120,32 @@ String submitWorkflow( byte[][] payloads, String queueDefault, String paramsJson, - String[] deferredNames); + String[] deferredNames) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } /** 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); + default String markWorkflowNodeResult(String jobId, boolean succeeded, String error, boolean skipCascade) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } - Optional getWorkflowStatusJson(String runId); + default Optional getWorkflowStatusJson(String runId) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } - void cancelWorkflowRun(String runId); + default void cancelWorkflowRun(String runId) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } - Optional getWorkflowPlanJson(String runId); + default Optional getWorkflowPlanJson(String runId) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } - Optional workflowNodeForJobJson(String jobId); + default Optional workflowNodeForJobJson(String jobId) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } - String[] expandFanOut( + default String[] expandFanOut( String runId, String parentNode, String[] childNames, @@ -138,11 +154,15 @@ String[] expandFanOut( String queue, int maxRetries, long timeoutMs, - int priority); + int priority) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } - Optional checkFanOutCompletionJson(String runId, String parentNode); + default Optional checkFanOutCompletionJson(String runId, String parentNode) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } - String createDeferredJob( + default String createDeferredJob( String runId, String nodeName, byte[] payload, @@ -150,11 +170,17 @@ String createDeferredJob( String queue, int maxRetries, long timeoutMs, - int priority); + int priority) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } - void cascadeSkipPending(String runId); + default void cascadeSkipPending(String runId) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } - Optional finalizeRunIfTerminal(String runId); + default Optional finalizeRunIfTerminal(String runId) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } // ── Worker ────────────────────────────────────────────────────── /** Start a worker that dispatches jobs to {@code bridge}; returns its control. */ From 239098387edd2140e6edc18a9f1ce289835789a5 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:50:24 +0530 Subject: [PATCH 08/11] fix(java): validate fan-out/fan-in step construction --- .../java/org/byteveda/taskito/workflows/Step.java | 3 +++ .../java/org/byteveda/taskito/workflows/Workflow.java | 11 +++++++++++ 2 files changed, 14 insertions(+) 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 index 79aa5476..24a7b9c9 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Step.java @@ -105,6 +105,9 @@ public Builder fanIn(String strategy) { } public Step build() { + if (fanOut != null && fanIn != null) { + throw new IllegalArgumentException("step '" + name + "' cannot be both fan-out and fan-in"); + } 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 index 97d75063..b4c22a95 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Workflow.java @@ -46,6 +46,7 @@ public Workflow step(Step step) { * result (a list). The predecessor named in {@code after} is the producer. */ public Workflow fanOut(String name, Task task, String strategy, String... after) { + requireSinglePredecessor("fan-out", name, after); return step(Step.of(name, task).fanOut(strategy).after(after).build()); } @@ -54,9 +55,19 @@ public Workflow fanOut(String name, Task task, String strategy, String... * into one list and passes it to {@code task}. */ public Workflow fanIn(String name, Task task, String strategy, String... after) { + requireSinglePredecessor("fan-in", name, after); return step(Step.of(name, task).fanIn(strategy).after(after).build()); } + // A fan-out/fan-in node has exactly one runtime trigger — its single producer. + // Zero predecessors would never enqueue; multiple could fire from the wrong one. + private static void requireSinglePredecessor(String kind, String name, String[] after) { + if (after.length != 1) { + throw new IllegalArgumentException( + kind + " step '" + name + "' needs exactly one predecessor, got " + after.length); + } + } + public String name() { return name; } From 5113eb3425742a0711deacd0f15db0ba21564626 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:50:24 +0530 Subject: [PATCH 09/11] fix(java): harden fan-out/fan-in routing (all successors, empty producers, missing result) --- .../byteveda/taskito/workflows/RunPlan.java | 9 +- .../taskito/workflows/WorkflowTracker.java | 102 ++++++++++-------- 2 files changed, 64 insertions(+), 47 deletions(-) diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java index 99237646..b1d7fbc6 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/RunPlan.java @@ -33,14 +33,15 @@ static RunPlan from(List nodes) { return new RunPlan(byName, successors); } - /** The first successor of {@code node} whose plan entry matches {@code match}, or null. */ - PlanNode successorMatching(String node, Predicate match) { + /** Every successor of {@code node} whose plan entry matches {@code match}. */ + List successorsMatching(String node, Predicate match) { + List matches = new ArrayList<>(); for (String successor : successors.getOrDefault(node, List.of())) { PlanNode candidate = byName.get(successor); if (candidate != null && match.test(candidate)) { - return candidate; + matches.add(candidate); } } - return null; + return matches; } } 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 index d5a852a3..9f01b71c 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowTracker.java @@ -69,36 +69,39 @@ private void onOutcome(String jobId, boolean succeeded, String error) { finalizeIfTerminal(ref.runId); } - /** If the just-completed node feeds a fan-out, split its result list into child jobs. */ + /** If the just-completed node feeds fan-outs, split its result list into child jobs. */ private void expandFanOutIfAny(String runId, String node, String jobId, RunPlan plan) { - PlanNode fanOut = plan.successorMatching(node, candidate -> candidate.fanOut != null); - if (fanOut == null) { + List fanOuts = plan.successorsMatching(node, candidate -> candidate.fanOut != null); + if (fanOuts.isEmpty()) { return; } - byte[] result = fetchResult(jobId); - if (result == null) { - return; + List items = serializer.deserialize(fetchResult(jobId), List.class); + for (PlanNode fanOut : fanOuts) { + String[] names = new String[items.size()]; + byte[][] payloads = new byte[items.size()][]; + for (int i = 0; i < items.size(); i++) { + names[i] = fanOut.name + "[" + i + "]"; + payloads[i] = serializer.serialize(items.get(i)); + } + backend.expandFanOut( + runId, + fanOut.name, + names, + payloads, + fanOut.taskName, + fanOut.queueOrDefault(), + fanOut.retries(), + fanOut.timeout(), + fanOut.priorityOrDefault()); + // An empty producer list yields no child jobs, so no child outcome will + // ever drive the fan-in — collect it now with an empty result list. + if (items.isEmpty()) { + collectFanIns(runId, fanOut.name, List.of(), plan); + } } - List items = serializer.deserialize(result, List.class); - String[] names = new String[items.size()]; - byte[][] payloads = new byte[items.size()][]; - for (int i = 0; i < items.size(); i++) { - names[i] = fanOut.name + "[" + i + "]"; - payloads[i] = serializer.serialize(items.get(i)); - } - backend.expandFanOut( - runId, - fanOut.name, - names, - payloads, - fanOut.taskName, - fanOut.queueOrDefault(), - fanOut.retries(), - fanOut.timeout(), - fanOut.priorityOrDefault()); } - /** When every fan-out child has settled, collect their results into one fan-in job. */ + /** When every fan-out child has settled, collect their results into the fan-in job(s). */ private void handleFanOutChild(String runId, String childName, RunPlan plan) { String parent = childName.substring(0, childName.indexOf('[')); FanOutCompletion done = backend.checkFanOutCompletionJson(runId, parent) @@ -112,25 +115,33 @@ private void handleFanOutChild(String runId, String childName, RunPlan plan) { finalizeIfTerminal(runId); return; } - PlanNode fanIn = plan == null ? null : plan.successorMatching(parent, candidate -> candidate.fanIn != null); - if (fanIn == null) { + List results = new ArrayList<>(done.childJobIds.size()); + for (String childJobId : done.childJobIds) { + results.add(serializer.deserialize(fetchResult(childJobId), Object.class)); + } + collectFanIns(runId, parent, results, plan); + } + + /** Create a fan-in job for every fan-in successor of {@code parent}; finalize if none. */ + private void collectFanIns(String runId, String parent, List results, RunPlan plan) { + List fanIns = + plan == null ? List.of() : plan.successorsMatching(parent, candidate -> candidate.fanIn != null); + if (fanIns.isEmpty()) { finalizeIfTerminal(runId); return; } - List results = new ArrayList<>(done.childJobIds.size()); - for (String childJobId : done.childJobIds) { - byte[] bytes = fetchResult(childJobId); - results.add(bytes == null ? null : serializer.deserialize(bytes, Object.class)); - } - backend.createDeferredJob( - runId, - fanIn.name, - serializer.serialize(results), - fanIn.taskName, - fanIn.queueOrDefault(), - fanIn.retries(), - fanIn.timeout(), - fanIn.priorityOrDefault()); + byte[] payload = serializer.serialize(results); + for (PlanNode fanIn : fanIns) { + backend.createDeferredJob( + runId, + fanIn.name, + payload, + fanIn.taskName, + fanIn.queueOrDefault(), + fanIn.retries(), + fanIn.timeout(), + fanIn.priorityOrDefault()); + } } private void finalizeIfTerminal(String runId) { @@ -139,7 +150,12 @@ private void finalizeIfTerminal(String runId) { } } - /** Read a job's serialized result, briefly polling in case the event outran the DB write. */ + /** + * Read a job's serialized result, briefly polling in case the outcome event + * outran the DB write. A missing result after the window is fatal to the + * step: returning null here would silently drop a fan-out expansion or fan in + * a {@code null} element, so we fail loudly instead. + */ private byte[] fetchResult(String jobId) { for (int attempt = 0; attempt < 50; attempt++) { Optional result = backend.getResult(jobId); @@ -150,10 +166,10 @@ private byte[] fetchResult(String jobId) { Thread.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); - return null; + throw new TaskitoException("interrupted awaiting result for workflow job " + jobId, e); } } - return null; + throw new TaskitoException("no result for workflow job " + jobId + " after 5s"); } private RunPlan loadPlan(String runId) { From 927eaadabf869287bf2af5610839990766bd4e68 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:50:24 +0530 Subject: [PATCH 10/11] fix(java): order fan-out children by item index --- crates/taskito-java/src/workflows/mod.rs | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/crates/taskito-java/src/workflows/mod.rs b/crates/taskito-java/src/workflows/mod.rs index 98b772d1..c087d605 100644 --- a/crates/taskito-java/src/workflows/mod.rs +++ b/crates/taskito-java/src/workflows/mod.rs @@ -719,10 +719,13 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_checkF let parent = read_string(env, &parent_node)?; let wf = queue.workflow_store()?; let prefix = format!("{parent}["); - let children = wf.get_workflow_nodes_by_prefix(&run_id, &prefix)?; + let mut children = wf.get_workflow_nodes_by_prefix(&run_id, &prefix)?; if children.is_empty() || !children.iter().all(|n| n.status.is_terminal()) { return Ok(std::ptr::null_mut()); } + // Order children by their item index (`parent[i]`) so the fan-in list + // matches the producer's list order, not storage-return order. + children.sort_by_key(|n| fan_out_child_index(&n.node_name)); let any_failed = children .iter() .any(|n| n.status == WorkflowNodeStatus::Failed); @@ -896,6 +899,15 @@ fn workflow_metadata_json(run_id: &str, node_name: &str) -> String { .to_string() } +/// Item index `i` parsed from a fan-out child name `parent[i]`. Unparseable names +/// sort last so they never silently reorder valid children. +fn fan_out_child_index(name: &str) -> u64 { + name.rsplit_once('[') + .and_then(|(_, rest)| rest.strip_suffix(']')) + .and_then(|idx| idx.parse().ok()) + .unwrap_or(u64::MAX) +} + /// 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)> { From 8c5a6e2e9cca46a6a3e8a6ad1ad0f0431df5b2fb Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Fri, 26 Jun 2026 09:50:52 +0530 Subject: [PATCH 11/11] docs(java): clarify trackWorkflows is required per worker --- sdks/java/README.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sdks/java/README.md b/sdks/java/README.md index cb97dd79..c77b5e19 100644 --- a/sdks/java/README.md +++ b/sdks/java/README.md @@ -107,8 +107,10 @@ Workflow wf = Workflow.named("pipeline") .fanIn("sum", sum, "all", "square"); // sum receives [1,4,9,16] ``` -`trackWorkflows()` must be attached for fan-out expansion and aggregation. -Conditions, gates, and sagas are not yet exposed. +`trackWorkflows()` advances run state from worker outcomes, so **every worker +that processes workflow jobs must enable it** — in a multi-worker deployment, a +run stalls on any node finished by a worker that did not opt in. Conditions, +gates, and sagas are not yet exposed. ### Middleware