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 33594c21..960e392a 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -398,14 +398,17 @@ public WorkflowRun submitWorkflow(Workflow workflow, Map supplie } /** - * 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. + * Fan-out/fan-in, gate, and conditional 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/evaluates them at runtime. */ private static Set deferredNodes(List steps) { Set deferred = new HashSet<>(); for (Step step : steps) { - if (step.fanOut != null || step.fanIn != null || step.gate != null) { + // "on_success" is the default/static path — only runtime-evaluated + // conditions (on_failure / always / callable) force a deferred node. + boolean runtimeCondition = step.condition != null && !"on_success".equals(step.condition); + if (step.fanOut != null || step.fanIn != null || step.gate != null || runtimeCondition) { deferred.add(step.name); } } @@ -459,6 +462,9 @@ private static Map stepSpec(Step step) { if (step.gate != null) { spec.put("gate", encodeGate(step.gate)); } + if (step.condition != null) { + spec.put("condition", step.condition); + } return spec; } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/Condition.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Condition.java new file mode 100644 index 00000000..739bfc88 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/Condition.java @@ -0,0 +1,12 @@ +package org.byteveda.taskito.workflows; + +/** + * A predicate deciding whether a workflow step runs, given the run's state when + * its predecessors have settled. Registered with the running worker via + * {@code trackWorkflows(workflow)} — it is code, so it is not persisted; a + * workflow using a callable condition must be tracked on the worker that runs it. + */ +@FunctionalInterface +public interface Condition { + boolean test(WorkflowContext context); +} 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 53686ceb..c2ac415c 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 @@ -19,6 +19,7 @@ final class PlanNode { final String fanOut; final String fanIn; final String gate; + final String condition; @JsonCreator PlanNode( @@ -31,7 +32,8 @@ final class PlanNode { @JsonProperty("priority") Integer priority, @JsonProperty("fanOut") String fanOut, @JsonProperty("fanIn") String fanIn, - @JsonProperty("gate") String gate) { + @JsonProperty("gate") String gate, + @JsonProperty("condition") String condition) { this.name = name; this.predecessors = predecessors == null ? Collections.emptyList() : predecessors; this.taskName = taskName; @@ -42,6 +44,7 @@ final class PlanNode { this.fanOut = fanOut; this.fanIn = fanIn; this.gate = gate; + this.condition = condition; } /** Whether this node is a fan-out/fan-in node (its job is created by the fan-out machinery). */ 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 ce4f5e7a..e4d7e860 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 @@ -19,6 +19,8 @@ public final class Step { public final String fanOut; public final String fanIn; public final GateConfig gate; + public final String condition; + public final Condition callableCondition; private Step(Builder builder) { this.name = builder.name; @@ -32,6 +34,8 @@ private Step(Builder builder) { this.fanOut = builder.fanOut; this.fanIn = builder.fanIn; this.gate = builder.gate; + this.condition = builder.condition; + this.callableCondition = builder.callableCondition; } /** Begin a step bound to a typed task. */ @@ -62,6 +66,8 @@ public static final class Builder { private String fanOut; private String fanIn; private GateConfig gate; + private String condition; + private Condition callableCondition; private Builder(String name, String taskName, Object payload) { this.name = name; @@ -127,6 +133,50 @@ public Builder gate(GateConfig gate) { return this; } + /** + * Run this step only when {@code condition} holds: {@code "on_success"} + * (every predecessor completed — the default), {@code "on_failure"} (any + * predecessor failed), or {@code "always"} (once predecessors settle). A + * conditional step is evaluated by the worker tracker, not pre-enqueued. + */ + public Builder condition(String condition) { + if (condition != null + && !"on_success".equals(condition) + && !"on_failure".equals(condition) + && !"always".equals(condition)) { + throw new IllegalArgumentException("unknown condition '" + condition + + "'; use on_success, on_failure, always, or condition(Condition)"); + } + this.condition = condition; + return this; + } + + /** Run this step only if every predecessor completed (the default). */ + public Builder onSuccess() { + return condition("on_success"); + } + + /** Run this step only if a predecessor failed (a recovery branch). */ + public Builder onFailure() { + return condition("on_failure"); + } + + /** Run this step once predecessors settle, regardless of their outcome. */ + public Builder always() { + return condition("always"); + } + + /** + * Run this step only when {@code predicate} holds. The predicate is code, + * so the workflow must be registered on the running worker via + * {@code trackWorkflows(workflow)}. + */ + public Builder condition(Condition predicate) { + this.callableCondition = predicate; + this.condition = "callable"; + return this; + } + public Step build() { if (fanOut != null && fanIn != null) { throw new IllegalArgumentException("step '" + name + "' cannot be both fan-out and fan-in"); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowContext.java b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowContext.java new file mode 100644 index 00000000..b937c3d2 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/workflows/WorkflowContext.java @@ -0,0 +1,32 @@ +package org.byteveda.taskito.workflows; + +import java.util.Map; +import java.util.Optional; + +/** + * The run's state passed to a callable {@link Condition} when deciding whether a + * step should run: the results and statuses of already-settled nodes. + * + * @param runId the workflow run id + * @param results deserialized results of completed nodes, by node name + * @param statuses every settled node's status, by node name + * @param successCount how many nodes completed + * @param failureCount how many nodes failed + */ +public record WorkflowContext( + String runId, + Map results, + Map statuses, + int successCount, + int failureCount) { + + /** The result of {@code nodeName}, if it completed. */ + public Optional result(String nodeName) { + return Optional.ofNullable(results.get(nodeName)); + } + + /** The status of {@code nodeName}, if it has settled. */ + public Optional status(String nodeName) { + return Optional.ofNullable(statuses.get(nodeName)); + } +} 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 7a210443..311a44a7 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 @@ -51,6 +51,8 @@ public final class WorkflowTracker { /** Deferred-node payloads from registered workflows: {@code wfName -> node -> serialized payload}. */ private final ConcurrentMap> deferredPayloads = new ConcurrentHashMap<>(); + /** Callable conditions from registered workflows: {@code wfName -> node -> predicate}. */ + private final ConcurrentMap> callableConditions = 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. */ @@ -70,12 +72,17 @@ public WorkflowTracker(QueueBackend backend, Serializer serializer) { */ public void register(Workflow workflow) { Map byNode = new HashMap<>(); + Map conditions = new HashMap<>(); for (Step step : workflow.steps()) { if (step.payload != null) { byNode.put(step.name, serializer.serialize(step.payload)); } + if (step.callableCondition != null) { + conditions.put(step.name, step.callableCondition); + } } deferredPayloads.put(workflow.name(), byNode); + callableConditions.put(workflow.name(), conditions); } /** Route a successful job outcome. */ @@ -103,12 +110,20 @@ private void onOutcome(String jobId, boolean succeeded, String error) { handleFanOutChild(ref.runId, ref.nodeName, plan); return; } - if (succeeded && plan != null) { + if (plan == null) { + // No plan to drive (e.g. an untracked run) — fall back to fail-fast. + if (!succeeded) { + backend.cascadeSkipPending(ref.runId); + } + finalizeIfTerminal(ref.runId); + return; + } + if (succeeded) { expandFanOutIfAny(ref.runId, ref.nodeName, jobId, plan); - promoteSuccessors(ref.runId, ref.nodeName, plan); - } else if (!succeeded) { - backend.cascadeSkipPending(ref.runId); } + // Evaluate every successor's condition — running on-failure recovery nodes + // and skip-cascading the rest. Replaces a blanket fail-fast cascade. + promoteSuccessors(ref.runId, ref.nodeName, plan); finalizeIfTerminal(ref.runId); } @@ -154,7 +169,13 @@ private void handleFanOutChild(String runId, String childName, RunPlan plan) { return; // siblings still running, or already handled by a concurrent outcome } if (!done.succeeded) { - backend.cascadeSkipPending(runId); + // The fan-out parent is now Failed; evaluate its successors' conditions + // (on_failure/always run, on_success skip-cascade) like any other node. + if (plan != null) { + promoteSuccessors(runId, parent, plan); + } else { + backend.cascadeSkipPending(runId); + } finalizeIfTerminal(runId); return; } @@ -191,36 +212,44 @@ 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. + * Evaluate every successor of {@code settledNode} whose predecessors have all + * settled: run it (its condition holds) or skip-and-cascade it. A satisfied + * gate parks; a satisfied deferred node's job is created; a satisfied static + * node is left for the scheduler; an unsatisfied node is skipped and the skip + * cascades to its own successors. */ 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); + for (PlanNode succ : plan.successorsMatching(settledNode, candidate -> true)) { + promoteOrSkip(runId, succ, statuses, plan); } } - private void promoteDeferred(String runId, PlanNode node, Map statuses, RunPlan plan) { + private void promoteOrSkip(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 (snap == null || snap.status != NodeStatus.PENDING) { + return; // already running, parked, promoted, or terminal } 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 + return; // a concurrent outcome already decided this node } // Roll back the dedupe marker if promotion fails, else the node wedges PENDING. try { - if (!allCompleted(node.predecessors, statuses)) { + if (!shouldRun(runId, node, statuses)) { backend.skipWorkflowNode(runId, node.name); promoteSuccessors(runId, node.name, plan); // cascade the skip downstream return; } + if (node.fanOut != null || node.fanIn != null) { + return; // fan nodes are driven by expandFanOut / fan-out completion + } + if (snap.jobId != null) { + return; // static node — the scheduler runs it once its deps complete + } if (node.gate != null) { enterGate(runId, node); return; @@ -232,6 +261,54 @@ private void promoteDeferred(String runId, PlanNode node, Map statuses) { + if ("callable".equals(node.condition)) { + Condition predicate = callableCondition(runId, node.name); + if (predicate == null) { + throw new TaskitoException("callable condition for workflow node '" + node.name + + "' is not registered; track the workflow on the worker via trackWorkflows(workflow)"); + } + return predicate.test(buildContext(runId, statuses)); + } + if ("on_failure".equals(node.condition)) { + return anyFailed(node.predecessors, statuses); + } + if ("always".equals(node.condition)) { + return true; + } + // null / "on_success": run only when every predecessor completed. + return allCompleted(node.predecessors, statuses); + } + + private Condition callableCondition(String runId, String nodeName) { + String wfName = workflowName(runId); + if (wfName == null) { + return null; + } + return callableConditions.getOrDefault(wfName, Map.of()).get(nodeName); + } + + /** Build the run snapshot a callable condition sees: completed nodes' results + statuses. */ + private WorkflowContext buildContext(String runId, Map statuses) { + Map results = new HashMap<>(); + Map statusByNode = new HashMap<>(); + int succeeded = 0; + int failed = 0; + for (NodeSnapshot node : statuses.values()) { + statusByNode.put(node.nodeName, node.status); + if (isCompleted(node.status)) { + succeeded++; + if (node.jobId != null) { + results.put(node.nodeName, serializer.deserialize(fetchResult(node.jobId), Object.class)); + } + } else if (node.status == NodeStatus.FAILED) { + failed++; + } + } + return new WorkflowContext(runId, results, statusByNode, succeeded, failed); + } + /** 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); @@ -297,14 +374,26 @@ private void createDeferredJobFor(String runId, PlanNode node) { } private byte[] deferredPayload(String runId, String nodeName) { - String wfName = runNames.computeIfAbsent( - runId, id -> backend.workflowNameForRun(id).orElse(null)); + String wfName = workflowName(runId); if (wfName == null) { return null; } return deferredPayloads.getOrDefault(wfName, Map.of()).get(nodeName); } + /** The run's workflow-definition name, cached (drives the deferred-payload + condition registries). */ + private String workflowName(String runId) { + String cached = runNames.get(runId); + if (cached != null) { + return cached; + } + String name = backend.workflowNameForRun(runId).orElse(null); + if (name != null) { + runNames.put(runId, name); + } + return name; + } + private void finalizeIfTerminal(String runId) { if (backend.finalizeRunIfTerminal(runId).isPresent()) { forget(runId); // run reached a terminal state — drop its cached state @@ -390,6 +479,16 @@ private static boolean allCompleted(List predecessors, Map predecessors, Map statuses) { + for (String pred : predecessors) { + NodeSnapshot snap = statuses.get(pred); + if (snap != null && snap.status == NodeStatus.FAILED) { + return true; + } + } + return false; + } + private static boolean isTerminal(NodeStatus status) { switch (status) { case COMPLETED: diff --git a/sdks/java/src/test/java/org/byteveda/taskito/WorkflowConditionTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowConditionTest.java new file mode 100644 index 00000000..c08ff11e --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowConditionTest.java @@ -0,0 +1,159 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicBoolean; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.byteveda.taskito.workflows.NodeStatus; +import org.byteveda.taskito.workflows.Step; +import org.byteveda.taskito.workflows.Workflow; +import org.byteveda.taskito.workflows.WorkflowRun; +import org.byteveda.taskito.workflows.WorkflowState; +import org.byteveda.taskito.workflows.WorkflowStatus; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +class WorkflowConditionTest { + + private static final Task RISKY = Task.of("c.risky", Integer.class); + private static final Task RECOVER = Task.of("c.recover", String.class); + private static final Task PRODUCER = Task.of("c.producer", Integer.class); + private static final Task CHECK = Task.of("c.check", Integer.class); + + @Test + @Timeout(30) + void onFailureRunsRecoveryNode(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("c.db").toString()).open()) { + Workflow wf = Workflow.named("cond") + .step(Step.of("risky", RISKY, 1).maxRetries(0).build()) + .step(Step.of("recover", RECOVER, "fix") + .onFailure() + .after("risky") + .build()); + WorkflowRun run = queue.submitWorkflow(wf); + AtomicBoolean recovered = new AtomicBoolean(); + try (Worker worker = queue.worker() + .handle(RISKY, p -> { + throw new IllegalStateException("boom"); + }) + .handle(RECOVER, p -> { + recovered.set(true); + return p; + }) + .trackWorkflows(wf) + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.FAILED, status.state); + assertEquals(NodeStatus.FAILED, status.node("risky").orElseThrow().status); + assertEquals(NodeStatus.COMPLETED, status.node("recover").orElseThrow().status); + assertEquals(true, recovered.get()); + } + } + } + + @Test + @Timeout(30) + void onSuccessSkipsRecoveryNode(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("c.db").toString()).open()) { + Workflow wf = Workflow.named("cond") + .step("risky", RISKY, 1) + .step(Step.of("recover", RECOVER, "fix") + .onFailure() + .after("risky") + .build()); + WorkflowRun run = queue.submitWorkflow(wf); + AtomicBoolean recovered = new AtomicBoolean(); + try (Worker worker = queue.worker() + .handle(RISKY, p -> p) + .handle(RECOVER, p -> { + recovered.set(true); + return p; + }) + .trackWorkflows(wf) + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.COMPLETED, status.state); + assertEquals(NodeStatus.SKIPPED, status.node("recover").orElseThrow().status); + assertEquals(false, recovered.get()); + } + } + } + + @Test + @Timeout(30) + void alwaysRunsAfterFailure(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("c.db").toString()).open()) { + Workflow wf = Workflow.named("cond") + .step(Step.of("risky", RISKY, 1).maxRetries(0).build()) + .step(Step.of("cleanup", RECOVER, "clean") + .always() + .after("risky") + .build()); + WorkflowRun run = queue.submitWorkflow(wf); + AtomicBoolean cleaned = new AtomicBoolean(); + try (Worker worker = queue.worker() + .handle(RISKY, p -> { + throw new IllegalStateException("boom"); + }) + .handle(RECOVER, p -> { + cleaned.set(true); + return p; + }) + .trackWorkflows(wf) + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.FAILED, status.state); + assertEquals(NodeStatus.COMPLETED, status.node("cleanup").orElseThrow().status); + assertEquals(true, cleaned.get()); + } + } + } + + @Test + @Timeout(30) + void callableConditionRunsWhenTrue(@TempDir Path dir) throws Exception { + assertCallable(dir, 10, true); + } + + @Test + @Timeout(30) + void callableConditionSkipsWhenFalse(@TempDir Path dir) throws Exception { + assertCallable(dir, 3, false); + } + + private static void assertCallable(Path dir, int produced, boolean expectChecked) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("c.db").toString()).open()) { + Workflow wf = Workflow.named("cond") + .step("producer", PRODUCER, produced) + .step(Step.of("check", CHECK, 0) + .condition(ctx -> ((Number) ctx.result("producer").orElse(0)).intValue() > 5) + .after("producer") + .build()); + WorkflowRun run = queue.submitWorkflow(wf); + AtomicBoolean checked = new AtomicBoolean(); + try (Worker worker = queue.worker() + .handle(PRODUCER, p -> p) + .handle(CHECK, p -> { + checked.set(true); + return p; + }) + .trackWorkflows(wf) + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.COMPLETED, status.state); + assertEquals( + expectChecked ? NodeStatus.COMPLETED : NodeStatus.SKIPPED, + status.node("check").orElseThrow().status); + assertEquals(expectChecked, checked.get()); + } + } + } +}