diff --git a/.github/actions/setup-java/action.yml b/.github/actions/setup-java/action.yml index 2e5c8a93..763878bc 100644 --- a/.github/actions/setup-java/action.yml +++ b/.github/actions/setup-java/action.yml @@ -2,7 +2,7 @@ name: Set up Java description: >- Installs a Temurin JDK and enables Gradle build caching. Used by every Java-touching job so toolchain setup lives in one place. The SDK targets Java - 11 bytecode via `--release 11`, so any JDK >= 11 works; we use 21 to match the + 17 bytecode via `--release 17`, so any JDK >= 17 works; we use 21 to match the publish workflow. inputs: diff --git a/.github/workflows/ci-java.yml b/.github/workflows/ci-java.yml index 6de3a097..372ad4b7 100644 --- a/.github/workflows/ci-java.yml +++ b/.github/workflows/ci-java.yml @@ -36,7 +36,7 @@ jobs: if-no-files-found: error retention-days: 1 - # Build + test on every supported JDK; the SDK targets 11 but runs on 11/17/21 + # Build + test on every supported JDK; the SDK targets 17 but runs on 17/21/25 # over a native .so, so JVM behaviour is runtime-sensitive. test: name: Java SDK Tests (JDK ${{ matrix.java }}) @@ -45,7 +45,7 @@ jobs: strategy: fail-fast: false matrix: - java: ["11", "17", "21"] + java: ["17", "21", "25"] steps: - name: Check out repository uses: actions/checkout@v7 @@ -65,7 +65,11 @@ jobs: # -x cargoBuild skips the Rust compile; copyNative still stages the .so. - name: Build and test the Java SDK working-directory: sdks/java - run: ./gradlew build -x cargoBuild --no-daemon + # Spotless/Checkstyle are JDK-independent (and palantir-java-format reaches + # into javac internals that newer JDKs change), so run the full `build` + # (which includes `check`) only on the baseline JDK; the other legs run the + # test suite to verify runtime behaviour on each JDK. + run: ./gradlew ${{ matrix.java == '17' && 'build' || 'test' }} -x cargoBuild --no-daemon # Smoke the other published platforms on one JDK: their native binary differs, # so they rebuild it (cargoBuild) instead of reusing the Linux artifact. diff --git a/crates/taskito-java/src/workflows/mod.rs b/crates/taskito-java/src/workflows/mod.rs index c087d605..6a111a1c 100644 --- a/crates/taskito-java/src/workflows/mod.rs +++ b/crates/taskito-java/src/workflows/mod.rs @@ -6,8 +6,12 @@ //! 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. +//! job at runtime via [`expand_fan_out`] / [`create_deferred_job`]. Approval +//! gates, conditional nodes, and sub-workflows are also deferred: the tracker +//! parks a gate node ([`set_workflow_node_waiting_approval`]) until resolved, +//! skips a node whose condition is false ([`skip_workflow_node`]), and submits a +//! sub-workflow as a child run (`submitWorkflow` with a parent link), resolving +//! the parent node when the child run finalizes. use std::collections::{HashMap, HashSet}; @@ -55,6 +59,13 @@ struct StepSpec { fan_out: Option, /// Fan-in strategy (e.g. "all") — this node collects its fan-out children's results. fan_in: Option, + /// Condition controlling whether this node runs: "on_success" / "on_failure" + /// / "always", or "callable" when the tracker holds a Java predicate. + condition: Option, + /// JSON `{timeoutMs, onTimeout, message}` marking an approval gate node. + gate: Option, + /// Serialized child-workflow spec marking a sub-workflow node. + sub_workflow: Option, } /// Combined run + node snapshot returned by `getWorkflowStatus`. Timestamps are @@ -109,8 +120,10 @@ unsafe fn borrow_queue<'a>(handle: jlong) -> &'a QueueHandle { /// `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. +/// String paramsJson, String[] deferredNames, String parentRunId, +/// String parentNodeName)` — record a run and pre-enqueue a job per static step. +/// `parentRunId`/`parentNodeName` link a sub-workflow child to its parent node +/// (both null for a top-level run). Returns the run id. #[no_mangle] #[allow(clippy::too_many_arguments)] pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_submitWorkflow<'local>( @@ -125,6 +138,8 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_submit queue_default: JString<'local>, params_json: JString<'local>, deferred_names: JObjectArray<'local>, + parent_run_id: JString<'local>, + parent_node_name: JString<'local>, ) -> jstring { guard(&mut env, std::ptr::null_mut(), |env| { let queue = unsafe { borrow_queue(handle) }; @@ -135,6 +150,8 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_submit 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 parent_run = read_optional_string(env, &parent_run_id)?; + let parent_node = read_optional_string(env, &parent_node_name)?; let run_id = submit( queue, name, @@ -145,6 +162,8 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_submit queue_default, params, deferred, + parent_run, + parent_node, )?; new_string(env, run_id) }) @@ -161,6 +180,8 @@ fn submit( queue_default: Option, params_json: Option, deferred_names: Vec, + parent_run_id: Option, + parent_node_name: Option, ) -> Result { if payload_names.len() != payloads.len() { return Err(BindingError::new( @@ -206,6 +227,9 @@ fn submit( priority: s.priority, fan_out: s.fan_out.clone(), fan_in: s.fan_in.clone(), + condition: s.condition.clone(), + gate: s.gate.clone(), + sub_workflow: s.sub_workflow.clone(), ..Default::default() }, ) @@ -253,8 +277,8 @@ fn submit( started_at: Some(now), completed_at: None, error: None, - parent_run_id: None, - parent_node_name: None, + parent_run_id, + parent_node_name, created_at: now, })?; @@ -491,6 +515,9 @@ struct PlanNodeView<'a> { priority: Option, fan_out: Option<&'a str>, fan_in: Option<&'a str>, + condition: Option<&'a str>, + gate: Option<&'a str>, + sub_workflow: Option<&'a str>, } /// The run + node a job belongs to. @@ -548,6 +575,9 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWor 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()), + condition: meta.and_then(|m| m.condition.as_deref()), + gate: meta.and_then(|m| m.gate.as_deref()), + sub_workflow: meta.and_then(|m| m.sub_workflow.as_deref()), } }) .collect(); @@ -555,6 +585,33 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_getWor }) } +/// `String workflowNameForRun(long, String runId)` — the run's definition name, +/// or `null` if the run (or its definition) is absent. Lets the tracker resolve a +/// run back to a registered `Workflow` so it can supply a deferred node's payload. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_workflowNameForRun< + '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()), + }; + match wf.get_workflow_definition_by_id(&run.definition_id)? { + Some(def) => new_string(env, def.name), + None => Ok(std::ptr::null_mut()), + } + }) +} + /// `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. @@ -869,6 +926,138 @@ pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_finali }) } +/// `void setWorkflowNodeWaitingApproval(long, String runId, String nodeName)` — +/// park an approval-gate node until it is resolved. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_setWorkflowNodeWaitingApproval< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + node_name: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let node_name = read_string(env, &node_name)?; + queue.workflow_store()?.update_workflow_node_status( + &run_id, + &node_name, + WorkflowNodeStatus::WaitingApproval, + )?; + Ok(()) + }) +} + +/// `void resolveWorkflowGate(long, String runId, String nodeName, +/// boolean approved, String error)` — settle a parked gate (or a sub-workflow +/// parent) as completed when approved, else failed with `error`. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_resolveWorkflowGate< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + node_name: JString<'local>, + approved: jboolean, + error: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let node_name = read_string(env, &node_name)?; + let error = read_optional_string(env, &error)?; + let wf = queue.workflow_store()?; + if approved != JNI_FALSE { + wf.set_workflow_node_completed(&run_id, &node_name, now_millis(), None)?; + } else { + let msg = error.unwrap_or_else(|| "gate rejected".to_string()); + wf.set_workflow_node_error(&run_id, &node_name, &msg)?; + } + Ok(()) + }) +} + +/// `void setWorkflowNodeRunning(long, String runId, String nodeName)` — promote a +/// gate / sub-workflow node to running without fan-out bookkeeping. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_setWorkflowNodeRunning< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + node_name: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let node_name = read_string(env, &node_name)?; + queue + .workflow_store()? + .set_workflow_node_running(&run_id, &node_name, now_millis())?; + Ok(()) + }) +} + +/// `void failWorkflowNode(long, String runId, String nodeName, String error)` — +/// mark a node failed (e.g. a sub-workflow whose child could not be submitted). +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_failWorkflowNode< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + node_name: JString<'local>, + error: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let node_name = read_string(env, &node_name)?; + let error = read_optional_string(env, &error)?.unwrap_or_else(|| "failed".to_string()); + queue + .workflow_store()? + .set_workflow_node_error(&run_id, &node_name, &error)?; + Ok(()) + }) +} + +/// `void skipWorkflowNode(long, String runId, String nodeName)` — mark a node +/// skipped (its condition evaluated false) and cancel any bound job. +#[no_mangle] +pub extern "system" fn Java_org_byteveda_taskito_internal_NativeWorkflows_skipWorkflowNode< + 'local, +>( + mut env: JNIEnv<'local>, + _class: JClass<'local>, + handle: jlong, + run_id: JString<'local>, + node_name: JString<'local>, +) { + guard(&mut env, (), |env| { + let queue = unsafe { borrow_queue(handle) }; + let run_id = read_string(env, &run_id)?; + let node_name = read_string(env, &node_name)?; + let wf = queue.workflow_store()?; + if let Some(node) = wf.get_workflow_node(&run_id, &node_name)? { + if let Some(job_id) = &node.job_id { + // Surface a storage failure rather than skipping while the bound job runs. + queue.storage.cancel_job(job_id)?; + } + } + wf.update_workflow_node_status(&run_id, &node_name, WorkflowNodeStatus::Skipped)?; + 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 { diff --git a/sdks/java/build.gradle.kts b/sdks/java/build.gradle.kts index bb0a72ea..986e4190 100644 --- a/sdks/java/build.gradle.kts +++ b/sdks/java/build.gradle.kts @@ -1,10 +1,9 @@ -import com.vanniktech.maven.publish.SonatypeHost plugins { `java-library` checkstyle - id("com.diffplug.spotless") version "6.25.0" - id("com.vanniktech.maven.publish") version "0.30.0" + id("com.diffplug.spotless") version "7.2.1" + id("com.vanniktech.maven.publish") version "0.37.0" } group = "org.byteveda" @@ -12,14 +11,15 @@ version = "0.18.0" java { // Sources + javadoc jars are added by the maven-publish plugin below. - // Compile to Java 11 bytecode with whatever JDK (>= 11) runs Gradle, rather - // than pinning a toolchain — `--release 11` also rejects post-11 stdlib APIs. - sourceCompatibility = JavaVersion.VERSION_11 - targetCompatibility = JavaVersion.VERSION_11 + // Compile to Java 17 bytecode with whatever JDK (>= 17) runs Gradle, rather + // than pinning a toolchain — `--release 17` also rejects post-17 stdlib APIs. + // Floor is 17 so every Spring Boot 3 app can adopt the SDK. + sourceCompatibility = JavaVersion.VERSION_17 + targetCompatibility = JavaVersion.VERSION_17 } tasks.withType().configureEach { - options.release.set(11) + options.release.set(17) } repositories { @@ -29,7 +29,7 @@ repositories { // --- Publishing: Maven Central via the Central Publisher Portal ------------ mavenPublishing { - publishToMavenCentral(SonatypeHost.CENTRAL_PORTAL) + publishToMavenCentral() signAllPublications() coordinates(group.toString(), "taskito", version.toString()) pom { @@ -102,13 +102,13 @@ val cargoTargetDir = layout.projectDirectory.dir("../../target") val nativeStaging = layout.buildDirectory.dir("native") // Build the native library for the local platform. -val cargoBuild by tasks.registering(Exec::class) { +val cargoBuild = tasks.register("cargoBuild") { workingDir = crateDir.asFile commandLine("cargo", "build", "--release", "--features", "postgres,redis,workflows") } // Stage the built library under its platform-classifier resource path. -val copyNative by tasks.registering(Copy::class) { +val copyNative = tasks.register("copyNative") { dependsOn(cargoBuild) from(cargoTargetDir.dir("release")) { include("libtaskito_java.so", "libtaskito_java.dylib", "taskito_java.dll") diff --git a/sdks/java/graalvm-smoke/build.gradle.kts b/sdks/java/graalvm-smoke/build.gradle.kts index e886383f..b9187843 100644 --- a/sdks/java/graalvm-smoke/build.gradle.kts +++ b/sdks/java/graalvm-smoke/build.gradle.kts @@ -4,7 +4,7 @@ plugins { application checkstyle - id("com.diffplug.spotless") version "6.25.0" + id("com.diffplug.spotless") version "7.2.1" id("org.graalvm.buildtools.native") version "0.10.6" } diff --git a/sdks/java/gradle/wrapper/gradle-wrapper.properties b/sdks/java/gradle/wrapper/gradle-wrapper.properties index 00cde9d2..2c68b418 100644 --- a/sdks/java/gradle/wrapper/gradle-wrapper.properties +++ b/sdks/java/gradle/wrapper/gradle-wrapper.properties @@ -1,6 +1,6 @@ distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists -distributionUrl=https\://services.gradle.org/distributions/gradle-8.12-bin.zip -distributionSha256Sum=7a00d51fb93147819aab76024feece20b6b84e420694101f276be952e08bef03 +distributionUrl=https\://services.gradle.org/distributions/gradle-9.6.1-bin.zip +distributionSha256Sum=9c0f7faeeb306cb14e4279a3e084ca6b596894089a0638e68a07c945a32c9e14 zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists diff --git a/sdks/java/processor/build.gradle.kts b/sdks/java/processor/build.gradle.kts index c3f7e5ac..4e53dc47 100644 --- a/sdks/java/processor/build.gradle.kts +++ b/sdks/java/processor/build.gradle.kts @@ -4,19 +4,19 @@ plugins { `java-library` checkstyle - id("com.diffplug.spotless") version "6.25.0" + id("com.diffplug.spotless") version "7.2.1" } group = "org.byteveda" version = "0.18.0" java { - sourceCompatibility = JavaVersion.VERSION_11 - targetCompatibility = JavaVersion.VERSION_11 + sourceCompatibility = JavaVersion.VERSION_17 + targetCompatibility = JavaVersion.VERSION_17 } tasks.withType().configureEach { - options.release.set(11) + options.release.set(17) } repositories { diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java index 81710064..33594c21 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -32,6 +32,7 @@ import org.byteveda.taskito.task.EnqueueOptions; import org.byteveda.taskito.task.Task; import org.byteveda.taskito.worker.Worker; +import org.byteveda.taskito.workflows.GateConfig; import org.byteveda.taskito.workflows.Step; import org.byteveda.taskito.workflows.Workflow; import org.byteveda.taskito.workflows.WorkflowRun; @@ -390,15 +391,21 @@ public WorkflowRun submitWorkflow(Workflow workflow, Map supplie payloads.toArray(new byte[0][]), null, null, - deferred.toArray(new String[0])); + deferred.toArray(new String[0]), + null, + null); return new WorkflowRun(backend, VIEWS, runId, workflow.name()); } - /** Fan-out/fan-in nodes, plus everything transitively downstream of them, are deferred. */ + /** + * Fan-out/fan-in and gate nodes — plus everything transitively downstream of + * them — are deferred: they carry a node row but no job at submit, and the + * worker tracker creates/parks them at runtime. + */ private static Set deferredNodes(List steps) { Set deferred = new HashSet<>(); for (Step step : steps) { - if (step.fanOut != null || step.fanIn != null) { + if (step.fanOut != null || step.fanIn != null || step.gate != null) { deferred.add(step.name); } } @@ -449,9 +456,29 @@ private static Map stepSpec(Step step) { if (step.fanIn != null) { spec.put("fanIn", step.fanIn); } + if (step.gate != null) { + spec.put("gate", encodeGate(step.gate)); + } return spec; } + /** + * Encode a gate as a JSON string {@code {timeoutMs, onTimeout, message}}. The + * core stores it opaquely on the step ({@code StepMetadata.gate}) and the + * worker tracker parses it back when the gate is reached. + */ + private static String encodeGate(GateConfig gate) { + Map blob = new LinkedHashMap<>(); + if (gate.timeout() != null) { + blob.put("timeoutMs", gate.timeout().toMillis()); + } + blob.put("onTimeout", gate.onTimeout().wire()); + if (gate.message() != null) { + blob.put("message", gate.message()); + } + return encode(blob); + } + @Override public Worker.Builder worker() { return Worker.builder(backend, serializer, middleware); 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 768e3bb4..795379e1 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 @@ -224,9 +224,21 @@ public String submitWorkflow( byte[][] payloads, String queueDefault, String paramsJson, - String[] deferredNames) { + String[] deferredNames, + String parentRunId, + String parentNodeName) { return NativeWorkflows.submitWorkflow( - handle, name, version, stepsJson, payloadNames, payloads, queueDefault, paramsJson, deferredNames); + handle, + name, + version, + stepsJson, + payloadNames, + payloads, + queueDefault, + paramsJson, + deferredNames, + parentRunId, + parentNodeName); } @Override @@ -254,6 +266,11 @@ public Optional workflowNodeForJobJson(String jobId) { return Optional.ofNullable(NativeWorkflows.workflowNodeForJob(handle, jobId)); } + @Override + public Optional workflowNameForRun(String runId) { + return Optional.ofNullable(NativeWorkflows.workflowNameForRun(handle, runId)); + } + @Override public String[] expandFanOut( String runId, @@ -298,6 +315,31 @@ public Optional finalizeRunIfTerminal(String runId) { return Optional.ofNullable(NativeWorkflows.finalizeRunIfTerminal(handle, runId)); } + @Override + public void setWorkflowNodeWaitingApproval(String runId, String nodeName) { + NativeWorkflows.setWorkflowNodeWaitingApproval(handle, runId, nodeName); + } + + @Override + public void resolveWorkflowGate(String runId, String nodeName, boolean approved, String error) { + NativeWorkflows.resolveWorkflowGate(handle, runId, nodeName, approved, error); + } + + @Override + public void setWorkflowNodeRunning(String runId, String nodeName) { + NativeWorkflows.setWorkflowNodeRunning(handle, runId, nodeName); + } + + @Override + public void failWorkflowNode(String runId, String nodeName, String error) { + NativeWorkflows.failWorkflowNode(handle, runId, nodeName, error); + } + + @Override + public void skipWorkflowNode(String runId, String nodeName) { + NativeWorkflows.skipWorkflowNode(handle, runId, nodeName); + } + @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 e3a7c77d..05272562 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 @@ -15,7 +15,11 @@ public final class NativeWorkflows { private NativeWorkflows() {} - /** Record a run and pre-enqueue a job per static step; returns the run id. */ + /** + * Record a run and pre-enqueue a job per static step; returns the run id. + * {@code parentRunId}/{@code parentNodeName} link a sub-workflow child to its + * parent node (both {@code null} for a top-level run). + */ public static native String submitWorkflow( long handle, String name, @@ -25,7 +29,9 @@ public static native String submitWorkflow( byte[][] payloads, String queueDefault, String paramsJson, - String[] deferredNames); + String[] deferredNames, + String parentRunId, + String parentNodeName); /** Record a node's terminal outcome; returns the run's final state, or {@code null}. */ public static native String markWorkflowNodeResult( @@ -44,6 +50,9 @@ public static native String markWorkflowNodeResult( /** Returns {@code {runId, nodeName}} for a job (JSON), or {@code null} if non-workflow. */ public static native String workflowNodeForJob(long handle, String jobId); + /** Returns the run's definition name, or {@code null} if the run is absent. */ + public static native String workflowNameForRun(long handle, String runId); + /** Expand a fan-out parent into one child job per payload; returns the child job ids. */ public static native String[] expandFanOut( long handle, @@ -76,4 +85,22 @@ public static native String createDeferredJob( /** Finalize the run if every node is terminal; returns the final state, or {@code null}. */ public static native String finalizeRunIfTerminal(long handle, String runId); + + // ── Gates / conditional nodes (driven by the worker-side tracker) ── + + /** Park an approval-gate node until it is resolved. */ + public static native void setWorkflowNodeWaitingApproval(long handle, String runId, String nodeName); + + /** Settle a parked gate (or sub-workflow parent): completed if approved, else failed with {@code error}. */ + public static native void resolveWorkflowGate( + long handle, String runId, String nodeName, boolean approved, String error); + + /** Promote a gate / sub-workflow node to running. */ + public static native void setWorkflowNodeRunning(long handle, String runId, String nodeName); + + /** Mark a node failed (e.g. a sub-workflow whose child could not be submitted). */ + public static native void failWorkflowNode(long handle, String runId, String nodeName, String error); + + /** Mark a node skipped (its condition evaluated false) and cancel any bound job. */ + public static native void skipWorkflowNode(long handle, String runId, String nodeName); } 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 c9913e27..18f9e577 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 @@ -145,7 +145,9 @@ default String submitWorkflow( byte[][] payloads, String queueDefault, String paramsJson, - String[] deferredNames) { + String[] deferredNames, + String parentRunId, + String parentNodeName) { throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); } @@ -170,6 +172,11 @@ default Optional workflowNodeForJobJson(String jobId) { throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); } + /** Returns the run's definition name, or empty if the run is absent. */ + default Optional workflowNameForRun(String runId) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + default String[] expandFanOut( String runId, String parentNode, @@ -207,6 +214,31 @@ default Optional finalizeRunIfTerminal(String runId) { throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); } + /** Park an approval-gate node until it is resolved. */ + default void setWorkflowNodeWaitingApproval(String runId, String nodeName) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + /** Settle a parked gate: completed if approved, else failed with {@code error}. */ + default void resolveWorkflowGate(String runId, String nodeName, boolean approved, String error) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + /** Promote a gate / sub-workflow node to running. */ + default void setWorkflowNodeRunning(String runId, String nodeName) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + /** Mark a node failed. */ + default void failWorkflowNode(String runId, String nodeName, String error) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + + /** Mark a node skipped (its condition evaluated false) and cancel any bound job. */ + default void skipWorkflowNode(String runId, String nodeName) { + throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED); + } + // ── 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/worker/Worker.java b/sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java index a86b6de1..008be28d 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 @@ -24,6 +24,7 @@ import org.byteveda.taskito.task.RetryPolicy; import org.byteveda.taskito.task.Task; import org.byteveda.taskito.task.TaskFunction; +import org.byteveda.taskito.workflows.Workflow; import org.byteveda.taskito.workflows.WorkflowTracker; /** @@ -36,12 +37,14 @@ public final class Worker implements AutoCloseable { private final WorkerControl control; private final ExecutorService executor; + private final WorkflowTracker tracker; private final CountDownLatch shutdown = new CountDownLatch(1); private boolean closed; - private Worker(WorkerControl control, ExecutorService executor) { + private Worker(WorkerControl control, ExecutorService executor, WorkflowTracker tracker) { this.control = control; this.executor = executor; + this.tracker = tracker; } public static Builder builder(QueueBackend backend, Serializer serializer, List middleware) { @@ -53,6 +56,26 @@ public void stop() { control.stop(); } + /** + * Approve a parked workflow gate so its successors run. Requires this worker + * to be tracking workflows ({@code trackWorkflows()} on the builder). + */ + public void approveGate(String runId, String nodeName) { + requireTracker().resolveGate(runId, nodeName, true, null); + } + + /** Reject a parked workflow gate; the gate fails and its successors are skipped. */ + public void rejectGate(String runId, String nodeName, String reason) { + requireTracker().resolveGate(runId, nodeName, false, reason == null ? "gate rejected" : reason); + } + + private WorkflowTracker requireTracker() { + if (tracker == null) { + throw new TaskitoException("worker is not tracking workflows; call trackWorkflows() on the builder"); + } + return tracker; + } + /** Block until {@link #close()} is called. */ public void awaitShutdown() throws InterruptedException { shutdown.await(); @@ -82,6 +105,9 @@ public synchronized void close() { executor.shutdownNow(); } control.close(); // now safe to free the native handle — no handler can touch it + if (tracker != null) { + tracker.close(); // stop the gate-timeout scheduler + } shutdown.countDown(); } @@ -99,6 +125,7 @@ public static final class Builder { private int concurrency; private Integer channelCapacity; private Integer batchSize; + private WorkflowTracker tracker; Builder(QueueBackend backend, Serializer serializer, List middleware) { this.backend = backend; @@ -173,12 +200,32 @@ 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, serializer); - on(EventName.SUCCESS, tracker::onSuccess); - on(EventName.DEAD, tracker::onDead); + ensureTracker(); + return this; + } + + /** + * Track workflows and register {@code workflows} so the tracker can supply + * the payloads of their deferred nodes (gates' downstream steps, etc.). + */ + public Builder trackWorkflows(Workflow... workflows) { + WorkflowTracker active = ensureTracker(); + for (Workflow workflow : workflows) { + active.register(workflow); + } return this; } + /** Create the tracker and wire its outcome listeners once. */ + private WorkflowTracker ensureTracker() { + if (tracker == null) { + tracker = new WorkflowTracker(backend, serializer); + on(EventName.SUCCESS, tracker::onSuccess); + on(EventName.DEAD, tracker::onDead); + } + return tracker; + } + public Worker start() { ExecutorService executor = concurrency > 0 ? Executors.newFixedThreadPool(concurrency) : Executors.newCachedThreadPool(); @@ -188,7 +235,7 @@ public Worker start() { new WorkerDispatchBridge(backend, handlers, serializer, executor, emitter, middleware); WorkerControl control = backend.startWorker(bridge, encodeOptions()); bridge.bind(control); - return new Worker(control, executor); + return new Worker(control, executor, tracker); } private String encodeOptions() { diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/GateAction.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/GateAction.java new file mode 100644 index 00000000..ec716bb9 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/GateAction.java @@ -0,0 +1,21 @@ +package org.byteveda.taskito.workflows; + +import java.util.Locale; + +/** What a {@link GateConfig} does when its approval timeout elapses. */ +public enum GateAction { + /** Treat the gate as approved — the node completes and successors run. */ + APPROVE, + /** Treat the gate as rejected — the node fails. */ + REJECT; + + /** Lowercase wire form used in the persisted gate metadata. */ + public String wire() { + return name().toLowerCase(Locale.ROOT); + } + + /** Parse a wire form ({@code "approve"}/{@code "reject"}); defaults to {@link #REJECT}. */ + public static GateAction fromWire(String wire) { + return "approve".equalsIgnoreCase(wire) ? APPROVE : REJECT; + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/GateConfig.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/GateConfig.java new file mode 100644 index 00000000..445c4390 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/GateConfig.java @@ -0,0 +1,40 @@ +package org.byteveda.taskito.workflows; + +import java.time.Duration; +import java.util.Objects; + +/** + * An approval gate on a workflow step. The node parks ({@code WAITING_APPROVAL}) + * until {@code Worker.approveGate}/{@code rejectGate} resolves it, or — if + * {@code timeout} is set — until the timeout elapses, when {@code onTimeout} + * decides the outcome. + * + * @param timeout how long to wait before auto-resolving; {@code null} waits forever + * @param onTimeout the action taken when {@code timeout} elapses (defaults to {@link GateAction#REJECT}) + * @param message an optional human-facing reason shown to the approver + */ +public record GateConfig(Duration timeout, GateAction onTimeout, String message) { + public GateConfig { + if (onTimeout == null) { + onTimeout = GateAction.REJECT; + } + if (timeout != null && (timeout.isNegative() || timeout.isZero())) { + throw new IllegalArgumentException("gate timeout must be positive"); + } + } + + /** A gate that waits indefinitely for a manual decision. */ + public static GateConfig manual() { + return new GateConfig(null, GateAction.REJECT, null); + } + + /** A gate that auto-resolves to {@code onTimeout} after {@code timeout}. */ + public static GateConfig timeout(Duration timeout, GateAction onTimeout) { + return new GateConfig(Objects.requireNonNull(timeout, "timeout"), onTimeout, null); + } + + /** A gate with a timeout and an approver-facing message. */ + public static GateConfig timeout(Duration timeout, GateAction onTimeout, String message) { + return new GateConfig(Objects.requireNonNull(timeout, "timeout"), onTimeout, message); + } +} 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 index cdd07861..53686ceb 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java @@ -18,6 +18,7 @@ final class PlanNode { final Integer priority; final String fanOut; final String fanIn; + final String gate; @JsonCreator PlanNode( @@ -29,7 +30,8 @@ final class PlanNode { @JsonProperty("timeoutMs") Long timeoutMs, @JsonProperty("priority") Integer priority, @JsonProperty("fanOut") String fanOut, - @JsonProperty("fanIn") String fanIn) { + @JsonProperty("fanIn") String fanIn, + @JsonProperty("gate") String gate) { this.name = name; this.predecessors = predecessors == null ? Collections.emptyList() : predecessors; this.taskName = taskName; @@ -39,6 +41,12 @@ final class PlanNode { this.priority = priority; this.fanOut = fanOut; this.fanIn = fanIn; + this.gate = gate; + } + + /** Whether this node is a fan-out/fan-in node (its job is created by the fan-out machinery). */ + boolean isFanNode() { + return fanOut != null || fanIn != null; } int retries() { 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 ff15ba62..ce4f5e7a 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 @@ -18,6 +18,7 @@ public final class Step { public final Integer priority; public final String fanOut; public final String fanIn; + public final GateConfig gate; private Step(Builder builder) { this.name = builder.name; @@ -30,6 +31,7 @@ private Step(Builder builder) { this.priority = builder.priority; this.fanOut = builder.fanOut; this.fanIn = builder.fanIn; + this.gate = builder.gate; } /** Begin a step bound to a typed task. */ @@ -59,6 +61,7 @@ public static final class Builder { private Integer priority; private String fanOut; private String fanIn; + private GateConfig gate; private Builder(String name, String taskName, Object payload) { this.name = name; @@ -114,10 +117,30 @@ public Builder fanIn(FanMode mode) { return fanIn(mode.wire()); } + /** + * Park this step for approval before it runs. The node waits until + * {@code Worker.approveGate}/{@code rejectGate}, or until the gate's + * timeout elapses. + */ + public Builder gate(GateConfig gate) { + this.gate = gate; + return this; + } + public Step build() { if (fanOut != null && fanIn != null) { throw new IllegalArgumentException("step '" + name + "' cannot be both fan-out and fan-in"); } + if (gate != null && (fanOut != null || fanIn != null)) { + throw new IllegalArgumentException("step '" + name + "' cannot be both a gate and a fan-out/fan-in"); + } + // A gate is a deferred control node (its task never enqueues), valid only + // on the Workflow.gate(...) sentinel — not on a normal task step. + if (gate != null && !Workflow.GATE_TASK.equals(taskName)) { + throw new IllegalArgumentException("step '" + name + + "': a gate may only be created via Workflow.gate(...); a gate on a normal task step " + + "would defer the node and never enqueue its task"); + } 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 bae45f67..627855b2 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 @@ -79,6 +79,20 @@ public Workflow fanIn(String name, Task task, FanMode mode, String... aft return fanIn(name, task, mode.wire(), after); } + /** + * Add an approval gate that runs after {@code after}. The gate is a control + * node — it runs no task; it parks until {@code Worker.approveGate}/ + * {@code rejectGate} (or its timeout) resolves it, then its successors run. + * The running worker must {@code trackWorkflows(this)} so its tracker holds + * the downstream steps' payloads. + */ + public Workflow gate(String name, GateConfig gate, String... after) { + return step(Step.of(name, GATE_TASK, null).gate(gate).after(after).build()); + } + + /** Sentinel task name for gate control nodes (never enqueued). */ + static final String GATE_TASK = "__gate__"; + // 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) { 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 9f01b71c..7a210443 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 @@ -6,10 +6,17 @@ import com.fasterxml.jackson.databind.JavaType; import com.fasterxml.jackson.databind.ObjectMapper; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import org.byteveda.taskito.TaskitoException; import org.byteveda.taskito.events.OutcomeEvent; import org.byteveda.taskito.serialization.Serializer; @@ -20,22 +27,57 @@ * {@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. + * run forward. A static node 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. An approval gate parks + * ({@code WAITING_APPROVAL}) until {@code Worker.approveGate}/{@code rejectGate} + * (or its timeout) resolves it; its successors then run. Deferred non-fan nodes + * (gates and steps downstream of a gate or fan-in) carry a node row but no job + * at submit — the tracker creates their job once predecessors settle, taking + * their payload from a {@link Workflow} registered with + * {@code trackWorkflows(workflow)}. + * + *

Each run's DAG (the plan) is loaded lazily from storage 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<>(); + private final ConcurrentMap runNames = new ConcurrentHashMap<>(); + private final ScheduledExecutorService gateScheduler = + Executors.newSingleThreadScheduledExecutor(WorkflowTracker::daemon); + + /** Deferred-node payloads from registered workflows: {@code wfName -> node -> serialized payload}. */ + private final ConcurrentMap> deferredPayloads = new ConcurrentHashMap<>(); + /** Live gate timeout timers, so a manual resolution can cancel a pending auto-resolution. */ + private final ConcurrentMap> gateTimers = new ConcurrentHashMap<>(); + /** Gate keys already resolved, so a timer and a manual call never both fire. */ + private final Set resolvedGates = ConcurrentHashMap.newKeySet(); + /** Deferred nodes already promoted (job created or gate parked), to dedupe concurrent outcomes. */ + private final Set promotedNodes = ConcurrentHashMap.newKeySet(); public WorkflowTracker(QueueBackend backend, Serializer serializer) { this.backend = backend; this.serializer = serializer; } + /** + * Register a workflow so the tracker can supply its deferred nodes' payloads + * at runtime. Required for any workflow with a gate (or other deferred node) + * whose downstream steps run a task. + */ + public void register(Workflow workflow) { + Map byNode = new HashMap<>(); + for (Step step : workflow.steps()) { + if (step.payload != null) { + byNode.put(step.name, serializer.serialize(step.payload)); + } + } + deferredPayloads.put(workflow.name(), byNode); + } + /** Route a successful job outcome. */ public void onSuccess(OutcomeEvent event) { onOutcome(event.jobId, true, null); @@ -63,6 +105,7 @@ private void onOutcome(String jobId, boolean succeeded, String error) { } if (succeeded && plan != null) { expandFanOutIfAny(ref.runId, ref.nodeName, jobId, plan); + promoteSuccessors(ref.runId, ref.nodeName, plan); } else if (!succeeded) { backend.cascadeSkipPending(ref.runId); } @@ -122,11 +165,14 @@ private void handleFanOutChild(String runId, String childName, RunPlan plan) { collectFanIns(runId, parent, results, plan); } - /** Create a fan-in job for every fan-in successor of {@code parent}; finalize if none. */ + /** Create a fan-in job for every fan-in successor of {@code parent}; promote others; 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()) { + if (plan != null) { + promoteSuccessors(runId, parent, plan); + } finalizeIfTerminal(runId); return; } @@ -144,12 +190,150 @@ private void collectFanIns(String runId, String parent, List results, Ru } } + /** + * Promote every deferred non-fan successor of {@code settledNode} whose + * predecessors have all settled: park a gate, or create the node's job when + * all predecessors completed, or skip-and-cascade when one did not. + */ + private void promoteSuccessors(String runId, String settledNode, RunPlan plan) { + Map statuses = statusMap(runId); + for (PlanNode succ : plan.successorsMatching(settledNode, candidate -> !candidate.isFanNode())) { + promoteDeferred(runId, succ, statuses, plan); + } + } + + private void promoteDeferred(String runId, PlanNode node, Map statuses, RunPlan plan) { + NodeSnapshot snap = statuses.get(node.name); + if (snap == null || snap.jobId != null || snap.status != NodeStatus.PENDING) { + return; // static (scheduler-run), already promoted, or already parked + } + if (!allTerminal(node.predecessors, statuses)) { + return; // a predecessor is still in flight + } + GateKey promotionKey = new GateKey(runId, node.name); + if (!promotedNodes.add(promotionKey)) { + return; // a concurrent outcome already promoted it + } + // Roll back the dedupe marker if promotion fails, else the node wedges PENDING. + try { + if (!allCompleted(node.predecessors, statuses)) { + backend.skipWorkflowNode(runId, node.name); + promoteSuccessors(runId, node.name, plan); // cascade the skip downstream + return; + } + if (node.gate != null) { + enterGate(runId, node); + return; + } + createDeferredJobFor(runId, node); + } catch (RuntimeException e) { + promotedNodes.remove(promotionKey); + throw e; + } + } + + /** Park a gate node for approval, scheduling its timeout auto-resolution if any. */ + private void enterGate(String runId, PlanNode node) { + backend.setWorkflowNodeWaitingApproval(runId, node.name); + GateMeta meta = node.gate == null ? null : decode(node.gate, GateMeta.class); + if (meta != null && meta.timeoutMs != null) { + GateKey key = new GateKey(runId, node.name); + gateTimers.put( + key, + gateScheduler.schedule( + () -> onGateTimeout(runId, node.name, meta.onTimeout), + meta.timeoutMs, + TimeUnit.MILLISECONDS)); + } + } + + private void onGateTimeout(String runId, String nodeName, String onTimeout) { + boolean approve = GateAction.fromWire(onTimeout) == GateAction.APPROVE; + resolveGate(runId, nodeName, approve, approve ? null : "gate timeout"); + } + + /** + * Resolve a parked gate: complete it (and run its successors) when + * {@code approved}, else fail it (and skip its successors). Idempotent — the + * first of a manual call and a timeout wins. + */ + public void resolveGate(String runId, String nodeName, boolean approved, String error) { + NodeSnapshot snap = statusMap(runId).get(nodeName); + if (snap == null || snap.status != NodeStatus.WAITING_APPROVAL) { + return; // only a parked gate can be resolved (wrong node, not yet parked, or already settled) + } + GateKey key = new GateKey(runId, nodeName); + if (!resolvedGates.add(key)) { + return; // already resolved by a timer or another caller + } + ScheduledFuture timer = gateTimers.remove(key); + if (timer != null) { + timer.cancel(false); + } + backend.resolveWorkflowGate(runId, nodeName, approved, error); + RunPlan plan = plans.computeIfAbsent(runId, this::loadPlan); + if (plan != null) { + // On reject the gate is Failed, so promoteSuccessors skips its successors. + promoteSuccessors(runId, nodeName, plan); + } + finalizeIfTerminal(runId); + } + + private void createDeferredJobFor(String runId, PlanNode node) { + byte[] payload = deferredPayload(runId, node.name); + if (payload == null) { + throw new TaskitoException("no payload for deferred workflow node '" + node.name + + "'; register the workflow on the worker via trackWorkflows(workflow)"); + } + backend.createDeferredJob( + runId, + node.name, + payload, + node.taskName, + node.queueOrDefault(), + node.retries(), + node.timeout(), + node.priorityOrDefault()); + } + + private byte[] deferredPayload(String runId, String nodeName) { + String wfName = runNames.computeIfAbsent( + runId, id -> backend.workflowNameForRun(id).orElse(null)); + if (wfName == null) { + return null; + } + return deferredPayloads.getOrDefault(wfName, Map.of()).get(nodeName); + } + private void finalizeIfTerminal(String runId) { if (backend.finalizeRunIfTerminal(runId).isPresent()) { - plans.remove(runId); // run reached a terminal state — drop its cached plan + forget(runId); // run reached a terminal state — drop its cached state } } + /** Drop all per-run state once a run is terminal, cancelling any lingering timers. */ + private void forget(String runId) { + plans.remove(runId); + runNames.remove(runId); + gateTimers.keySet().removeIf(key -> { + if (key.runId.equals(runId)) { + ScheduledFuture timer = gateTimers.get(key); + if (timer != null) { + timer.cancel(false); + } + return true; + } + return false; + }); + resolvedGates.removeIf(key -> key.runId.equals(runId)); + promotedNodes.removeIf(key -> key.runId.equals(runId)); + } + + /** Stop the gate-timeout scheduler. Called when the owning worker closes. */ + public void close() { + gateScheduler.shutdownNow(); + } + /** * 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 @@ -172,6 +356,58 @@ private byte[] fetchResult(String jobId) { throw new TaskitoException("no result for workflow job " + jobId + " after 5s"); } + private Map statusMap(String runId) { + WorkflowStatus status = backend.getWorkflowStatusJson(runId) + .map(raw -> decode(raw, WorkflowStatus.class)) + .orElse(null); + if (status == null) { + return Map.of(); + } + Map byName = new HashMap<>(); + for (NodeSnapshot node : status.nodes) { + byName.put(node.nodeName, node); + } + return byName; + } + + private static boolean allTerminal(List predecessors, Map statuses) { + for (String pred : predecessors) { + NodeSnapshot snap = statuses.get(pred); + if (snap == null || !isTerminal(snap.status)) { + return false; + } + } + return true; + } + + private static boolean allCompleted(List predecessors, Map statuses) { + for (String pred : predecessors) { + NodeSnapshot snap = statuses.get(pred); + if (snap == null || !isCompleted(snap.status)) { + return false; + } + } + return true; + } + + private static boolean isTerminal(NodeStatus status) { + switch (status) { + case COMPLETED: + case FAILED: + case SKIPPED: + case CACHE_HIT: + case COMPENSATED: + case COMPENSATION_FAILED: + return true; + default: + return false; + } + } + + private static boolean isCompleted(NodeStatus status) { + return status == NodeStatus.COMPLETED || status == NodeStatus.CACHE_HIT; + } + private RunPlan loadPlan(String runId) { return backend.getWorkflowPlanJson(runId) .map(raw -> RunPlan.from(decodeList(raw))) @@ -195,6 +431,53 @@ private List decodeList(String raw) { } } + private static Thread daemon(Runnable runnable) { + Thread thread = new Thread(runnable, "taskito-workflow-gates"); + thread.setDaemon(true); + return thread; + } + + /** A run + node identity used to key gate timers and dedupe promotion. */ + private static final class GateKey { + final String runId; + final String nodeName; + + GateKey(String runId, String nodeName) { + this.runId = runId; + this.nodeName = nodeName; + } + + @Override + public boolean equals(Object other) { + if (!(other instanceof GateKey)) { + return false; + } + GateKey that = (GateKey) other; + return runId.equals(that.runId) && nodeName.equals(that.nodeName); + } + + @Override + public int hashCode() { + return 31 * runId.hashCode() + nodeName.hashCode(); + } + } + + /** {@code {timeoutMs, onTimeout, message}} parsed from a node's gate metadata. */ + @JsonIgnoreProperties(ignoreUnknown = true) + private static final class GateMeta { + final Long timeoutMs; + final String onTimeout; + + @JsonCreator + GateMeta( + @JsonProperty("timeoutMs") Long timeoutMs, + @JsonProperty("onTimeout") String onTimeout, + @JsonProperty("message") String message) { + this.timeoutMs = timeoutMs; + this.onTimeout = onTimeout; + } + } + /** {@code {runId, nodeName}} from the native node-ref view. */ @JsonIgnoreProperties(ignoreUnknown = true) private static final class NodeRef { diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowGateTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowGateTest.java new file mode 100644 index 00000000..686074a3 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowGateTest.java @@ -0,0 +1,140 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.fail; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicInteger; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.byteveda.taskito.workflows.GateAction; +import org.byteveda.taskito.workflows.GateConfig; +import org.byteveda.taskito.workflows.NodeSnapshot; +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 WorkflowGateTest { + + private static final Task PROCESS = Task.of("g.process", Integer.class); + private static final Task DEPLOY = Task.of("g.deploy", Integer.class); + + private static Workflow gatedWorkflow(GateConfig gate) { + return Workflow.named("gated") + .step("process", PROCESS, 1) + .gate("gate", gate, "process") + .step("deploy", DEPLOY, 2, "gate"); + } + + @Test + @Timeout(30) + void approvedGateRunsDownstream(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("g.db").toString()).open()) { + Workflow wf = gatedWorkflow(GateConfig.manual()); + WorkflowRun run = queue.submitWorkflow(wf); + AtomicInteger deployed = new AtomicInteger(); + try (Worker worker = queue.worker() + .handle(PROCESS, p -> p) + .handle(DEPLOY, p -> deployed.incrementAndGet()) + .trackWorkflows(wf) + .start()) { + awaitGate(run, NodeStatus.WAITING_APPROVAL); + worker.approveGate(run.runId(), "gate"); + + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.COMPLETED, status.state); + assertEquals(NodeStatus.COMPLETED, status.node("gate").orElseThrow().status); + assertEquals(NodeStatus.COMPLETED, status.node("deploy").orElseThrow().status); + assertEquals(1, deployed.get()); + } + } + } + + @Test + @Timeout(30) + void rejectedGateSkipsDownstream(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("g.db").toString()).open()) { + Workflow wf = gatedWorkflow(GateConfig.manual()); + WorkflowRun run = queue.submitWorkflow(wf); + AtomicInteger deployed = new AtomicInteger(); + try (Worker worker = queue.worker() + .handle(PROCESS, p -> p) + .handle(DEPLOY, p -> deployed.incrementAndGet()) + .trackWorkflows(wf) + .start()) { + awaitGate(run, NodeStatus.WAITING_APPROVAL); + worker.rejectGate(run.runId(), "gate", "no"); + + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.FAILED, status.state); + assertEquals(NodeStatus.FAILED, status.node("gate").orElseThrow().status); + assertEquals(NodeStatus.SKIPPED, status.node("deploy").orElseThrow().status); + assertEquals(0, deployed.get()); + } + } + } + + @Test + @Timeout(30) + void gateTimeoutAutoApproves(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("g.db").toString()).open()) { + Workflow wf = gatedWorkflow(GateConfig.timeout(Duration.ofMillis(300), GateAction.APPROVE)); + WorkflowRun run = queue.submitWorkflow(wf); + AtomicInteger deployed = new AtomicInteger(); + try (Worker worker = queue.worker() + .handle(PROCESS, p -> p) + .handle(DEPLOY, p -> deployed.incrementAndGet()) + .trackWorkflows(wf) + .start()) { + // No manual resolution — the timeout drives it to approval. + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.COMPLETED, status.state); + assertEquals(1, deployed.get()); + } + } + } + + @Test + @Timeout(30) + void gateTimeoutAutoRejects(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("g.db").toString()).open()) { + Workflow wf = gatedWorkflow(GateConfig.timeout(Duration.ofMillis(300), GateAction.REJECT)); + WorkflowRun run = queue.submitWorkflow(wf); + AtomicInteger deployed = new AtomicInteger(); + try (Worker worker = queue.worker() + .handle(PROCESS, p -> p) + .handle(DEPLOY, p -> deployed.incrementAndGet()) + .trackWorkflows(wf) + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.FAILED, status.state); + assertEquals(NodeStatus.SKIPPED, status.node("deploy").orElseThrow().status); + assertEquals(0, deployed.get()); + } + } + } + + /** Poll until the gate node reaches {@code expected}, or fail after 10s. */ + private static void awaitGate(WorkflowRun run, NodeStatus expected) throws InterruptedException { + long deadline = System.nanoTime() + Duration.ofSeconds(10).toNanos(); + while (System.nanoTime() < deadline) { + Optional gate = run.status().flatMap(s -> s.node("gate")); + if (gate.isPresent() && gate.get().status == expected) { + return; + } + Thread.sleep(50); + } + fail("gate did not reach " + expected); + } +} diff --git a/sdks/java/test-support/build.gradle.kts b/sdks/java/test-support/build.gradle.kts index b17505d4..033124d5 100644 --- a/sdks/java/test-support/build.gradle.kts +++ b/sdks/java/test-support/build.gradle.kts @@ -4,19 +4,19 @@ plugins { `java-library` checkstyle - id("com.diffplug.spotless") version "6.25.0" + id("com.diffplug.spotless") version "7.2.1" } group = "org.byteveda" version = "0.18.0" java { - sourceCompatibility = JavaVersion.VERSION_11 - targetCompatibility = JavaVersion.VERSION_11 + sourceCompatibility = JavaVersion.VERSION_17 + targetCompatibility = JavaVersion.VERSION_17 } tasks.withType().configureEach { - options.release.set(11) + options.release.set(17) } repositories { diff --git a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java index 9a8e1296..dae38bee 100644 --- a/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java +++ b/sdks/java/test-support/src/main/java/org/byteveda/taskito/test/InMemoryQueueBackend.java @@ -476,7 +476,9 @@ public String submitWorkflow( byte[][] payloads, String queueDefault, String paramsJson, - String[] deferredNames) { + String[] deferredNames, + String parentRunId, + String parentNodeName) { return unsupported(); }