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 960e392a..16734ba3 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -5,6 +5,7 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; +import java.util.Base64; import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; @@ -408,7 +409,11 @@ private static Set deferredNodes(List steps) { // "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) { + if (step.fanOut != null + || step.fanIn != null + || step.gate != null + || step.subWorkflow != null + || runtimeCondition) { deferred.add(step.name); } } @@ -436,7 +441,7 @@ public void cancelWorkflow(String runId) { } /** Encode a step's structure + per-step overrides for the native submit call. */ - private static Map stepSpec(Step step) { + private Map stepSpec(Step step) { Map spec = new LinkedHashMap<>(); spec.put("name", step.name); spec.put("taskName", step.taskName); @@ -465,9 +470,50 @@ private static Map stepSpec(Step step) { if (step.condition != null) { spec.put("condition", step.condition); } + if (step.subWorkflow != null) { + spec.put("subWorkflow", encodeChild(step.subWorkflow)); + } return spec; } + /** + * Encode a child workflow as a JSON string the worker tracker submits when the + * sub-workflow node is reached: its name/version, the child's step specs, its + * deferred set, and every child step's payload (Base64). Recurses through + * {@link #stepSpec}, so a child may itself contain gates/conditions/sub-workflows. + */ + private String encodeChild(Workflow child) { + List steps = child.steps(); + Set deferred = deferredNodes(steps); + List> specs = new ArrayList<>(steps.size()); + Map payloads = new LinkedHashMap<>(); + for (Step step : steps) { + // A child run has no submit-time payload map and no callable registry, so + // any step that depends on either would lose state — reject it at build time. + if ("callable".equals(step.condition)) { + throw new TaskitoException("child workflow step '" + step.name + + "' uses condition(Condition); a sub-workflow cannot carry callable predicates"); + } + boolean controlNode = + step.fanOut != null || step.fanIn != null || step.gate != null || step.subWorkflow != null; + if (!controlNode && step.payload == null) { + throw new TaskitoException("child workflow step '" + step.name + + "' has no payload; a sub-workflow cannot supply child step payloads at submit"); + } + specs.add(stepSpec(step)); + if (step.payload != null) { + payloads.put(step.name, Base64.getEncoder().encodeToString(serializer.serialize(step.payload))); + } + } + Map blob = new LinkedHashMap<>(); + blob.put("name", child.name()); + blob.put("version", child.version()); + blob.put("stepsJson", encode(specs)); + blob.put("deferred", new ArrayList<>(deferred)); + blob.put("payloads", payloads); + return encode(blob); + } + /** * Encode a gate as a JSON string {@code {timeoutMs, onTimeout, message}}. The * core stores it opaquely on the step ({@code StepMetadata.gate}) and the 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 c2ac415c..a0e57a5d 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 @@ -20,6 +20,7 @@ final class PlanNode { final String fanIn; final String gate; final String condition; + final String subWorkflow; @JsonCreator PlanNode( @@ -33,7 +34,8 @@ final class PlanNode { @JsonProperty("fanOut") String fanOut, @JsonProperty("fanIn") String fanIn, @JsonProperty("gate") String gate, - @JsonProperty("condition") String condition) { + @JsonProperty("condition") String condition, + @JsonProperty("subWorkflow") String subWorkflow) { this.name = name; this.predecessors = predecessors == null ? Collections.emptyList() : predecessors; this.taskName = taskName; @@ -45,6 +47,7 @@ final class PlanNode { this.fanIn = fanIn; this.gate = gate; this.condition = condition; + this.subWorkflow = subWorkflow; } /** 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 e4d7e860..d86e09b4 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 @@ -21,6 +21,7 @@ public final class Step { public final GateConfig gate; public final String condition; public final Condition callableCondition; + public final Workflow subWorkflow; private Step(Builder builder) { this.name = builder.name; @@ -36,6 +37,7 @@ private Step(Builder builder) { this.gate = builder.gate; this.condition = builder.condition; this.callableCondition = builder.callableCondition; + this.subWorkflow = builder.subWorkflow; } /** Begin a step bound to a typed task. */ @@ -68,6 +70,7 @@ public static final class Builder { private GateConfig gate; private String condition; private Condition callableCondition; + private Workflow subWorkflow; private Builder(String name, String taskName, Object payload) { this.name = name; @@ -177,6 +180,17 @@ public Builder condition(Condition predicate) { return this; } + /** + * Make this step a sub-workflow: instead of running a task it submits + * {@code child} as a child run and completes when the child finalizes + * (failing if the child fails). The running worker must + * {@code trackWorkflows(parent)}. + */ + public Builder subWorkflow(Workflow child) { + this.subWorkflow = child; + return this; + } + public Step build() { if (fanOut != null && fanIn != null) { throw new IllegalArgumentException("step '" + name + "' cannot be both fan-out and fan-in"); @@ -191,6 +205,10 @@ public Step build() { + "': 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"); } + if (subWorkflow != null && (fanOut != null || fanIn != null || gate != null)) { + throw new IllegalArgumentException( + "step '" + name + "' cannot be both a sub-workflow and a gate/fan-out/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 627855b2..ffbffdb2 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 @@ -93,6 +93,21 @@ public Workflow gate(String name, GateConfig gate, String... after) { /** Sentinel task name for gate control nodes (never enqueued). */ static final String GATE_TASK = "__gate__"; + /** + * Add a sub-workflow step: when reached it submits {@code child} as a child + * run and completes when the child finalizes (failing if the child fails). + * The running worker must {@code trackWorkflows(this)}. + */ + public Workflow subWorkflow(String name, Workflow child, String... after) { + return step(Step.of(name, SUB_WORKFLOW_TASK, null) + .subWorkflow(child) + .after(after) + .build()); + } + + /** Sentinel task name for sub-workflow control nodes (never enqueued). */ + static final String SUB_WORKFLOW_TASK = "__subworkflow__"; + // 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 311a44a7..97b28903 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 @@ -7,6 +7,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import java.util.ArrayList; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Optional; @@ -51,6 +52,8 @@ public final class WorkflowTracker { /** Deferred-node payloads from registered workflows: {@code wfName -> node -> serialized payload}. */ private final ConcurrentMap> deferredPayloads = new ConcurrentHashMap<>(); + /** Sub-workflow child runs' deferred-node payloads, keyed by run id so same-named runs don't collide. */ + private final ConcurrentMap> childRunPayloads = 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. */ @@ -59,6 +62,8 @@ public final class WorkflowTracker { 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(); + /** Sub-workflow child run → its parent {@code [runId, nodeName]}, so a finalized child resolves its parent node. */ + private final ConcurrentMap childToParent = new ConcurrentHashMap<>(); public WorkflowTracker(QueueBackend backend, Serializer serializer) { this.backend = backend; @@ -254,6 +259,10 @@ private void promoteOrSkip(String runId, PlanNode node, Map deferred = new HashSet<>(child.deferred); + List payloadNames = new ArrayList<>(); + List payloadBytes = new ArrayList<>(); + Map allPayloads = new HashMap<>(); + child.payloads.forEach((nodeName, base64) -> { + byte[] bytes = java.util.Base64.getDecoder().decode(base64); + allPayloads.put(nodeName, bytes); + if (!deferred.contains(nodeName)) { + payloadNames.add(nodeName); + payloadBytes.add(bytes); + } + }); + String childRun = backend.submitWorkflow( + child.name, + child.version, + child.stepsJson, + payloadNames.toArray(new String[0]), + payloadBytes.toArray(new byte[0][]), + null, + null, + deferred.toArray(new String[0]), + runId, + node.name); + // Register the child's payloads under its run id so the tracker can promote the + // child's own deferred nodes — run-scoped, so two same-named child runs can't stomp. + childRunPayloads.put(childRun, allPayloads); + childToParent.put(childRun, new String[] {runId, node.name}); + backend.setWorkflowNodeRunning(runId, node.name); + } + /** Whether {@code node}'s condition holds given its predecessors' outcomes. */ private boolean shouldRun(String runId, PlanNode node, Map statuses) { if ("callable".equals(node.condition)) { @@ -374,6 +416,10 @@ private void createDeferredJobFor(String runId, PlanNode node) { } private byte[] deferredPayload(String runId, String nodeName) { + Map childPayloads = childRunPayloads.get(runId); + if (childPayloads != null) { + return childPayloads.get(nodeName); // a child run resolves only its run-scoped payloads + } String wfName = workflowName(runId); if (wfName == null) { return null; @@ -395,11 +441,32 @@ private String workflowName(String runId) { } private void finalizeIfTerminal(String runId) { - if (backend.finalizeRunIfTerminal(runId).isPresent()) { - forget(runId); // run reached a terminal state — drop its cached state + Optional finalState = backend.finalizeRunIfTerminal(runId); + if (finalState.isEmpty()) { + return; + } + String[] parent = childToParent.remove(runId); + forget(runId); // run reached a terminal state — drop its cached state + if (parent != null) { + // This run is a sub-workflow child — resolve its parent node, then advance the parent. + boolean succeeded = "completed".equals(finalState.get()); + resolveSubWorkflowParent( + parent[0], + parent[1], + succeeded, + succeeded ? null : "sub-workflow " + runId + " " + finalState.get()); } } + private void resolveSubWorkflowParent(String parentRun, String parentNode, boolean succeeded, String error) { + backend.resolveWorkflowGate(parentRun, parentNode, succeeded, error); + RunPlan plan = plans.computeIfAbsent(parentRun, this::loadPlan); + if (plan != null) { + promoteSuccessors(parentRun, parentNode, plan); + } + finalizeIfTerminal(parentRun); + } + /** Drop all per-run state once a run is terminal, cancelling any lingering timers. */ private void forget(String runId) { plans.remove(runId); @@ -416,6 +483,9 @@ private void forget(String runId) { }); resolvedGates.removeIf(key -> key.runId.equals(runId)); promotedNodes.removeIf(key -> key.runId.equals(runId)); + childToParent.remove(runId); + childToParent.values().removeIf(parent -> parent[0].equals(runId)); + childRunPayloads.remove(runId); } /** Stop the gate-timeout scheduler. Called when the owning worker closes. */ @@ -577,6 +647,30 @@ private static final class GateMeta { } } + /** A serialized child workflow carried in a sub-workflow node's metadata. */ + @JsonIgnoreProperties(ignoreUnknown = true) + private static final class ChildSpec { + final String name; + final int version; + final String stepsJson; + final List deferred; + final Map payloads; + + @JsonCreator + ChildSpec( + @JsonProperty("name") String name, + @JsonProperty("version") int version, + @JsonProperty("stepsJson") String stepsJson, + @JsonProperty("deferred") List deferred, + @JsonProperty("payloads") Map payloads) { + this.name = name; + this.version = version; + this.stepsJson = stepsJson; + this.deferred = deferred == null ? List.of() : deferred; + this.payloads = payloads == null ? Map.of() : payloads; + } + } + /** {@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/WorkflowSubWorkflowTest.java b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSubWorkflowTest.java new file mode 100644 index 00000000..b54261f6 --- /dev/null +++ b/sdks/java/src/test/java/org/byteveda/taskito/WorkflowSubWorkflowTest.java @@ -0,0 +1,112 @@ +package org.byteveda.taskito; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.nio.file.Path; +import java.time.Duration; +import java.util.concurrent.atomic.AtomicInteger; +import org.byteveda.taskito.task.Task; +import org.byteveda.taskito.worker.Worker; +import org.byteveda.taskito.workflows.NodeStatus; +import org.byteveda.taskito.workflows.Step; +import org.byteveda.taskito.workflows.Workflow; +import org.byteveda.taskito.workflows.WorkflowRun; +import org.byteveda.taskito.workflows.WorkflowState; +import org.byteveda.taskito.workflows.WorkflowStatus; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.io.TempDir; + +class WorkflowSubWorkflowTest { + + private static final Task PREP = Task.of("sw.prep", Integer.class); + private static final Task CHILD_A = Task.of("sw.childA", Integer.class); + private static final Task CHILD_B = Task.of("sw.childB", Integer.class); + private static final Task FINISH = Task.of("sw.finish", Integer.class); + + @Test + @Timeout(30) + void childCompletesThenParentContinues(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("sw.db").toString()).open()) { + Workflow child = Workflow.named("child").step("a", CHILD_A, 10).step("b", CHILD_B, 20, "a"); + Workflow parent = Workflow.named("parent") + .step("prep", PREP, 1) + .subWorkflow("sub", child, "prep") + .step("finish", FINISH, 2, "sub"); + + WorkflowRun run = queue.submitWorkflow(parent); + AtomicInteger childRan = new AtomicInteger(); + AtomicInteger finished = new AtomicInteger(); + try (Worker worker = queue.worker() + .handle(PREP, p -> p) + .handle(CHILD_A, p -> childRan.incrementAndGet()) + .handle(CHILD_B, p -> childRan.incrementAndGet()) + .handle(FINISH, p -> finished.incrementAndGet()) + .trackWorkflows(parent) + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.COMPLETED, status.state); + assertEquals(NodeStatus.COMPLETED, status.node("sub").orElseThrow().status); + assertEquals(NodeStatus.COMPLETED, status.node("finish").orElseThrow().status); + assertEquals(2, childRan.get()); + assertEquals(1, finished.get()); + } + } + } + + @Test + @Timeout(30) + void childFailureFailsParentAndSkipsDownstream(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("sw.db").toString()).open()) { + Workflow child = Workflow.named("child") + .step(Step.of("a", CHILD_A, 10).maxRetries(0).build()); + Workflow parent = Workflow.named("parent") + .step("prep", PREP, 1) + .subWorkflow("sub", child, "prep") + .step("finish", FINISH, 2, "sub"); + + WorkflowRun run = queue.submitWorkflow(parent); + AtomicInteger finished = new AtomicInteger(); + try (Worker worker = queue.worker() + .handle(PREP, p -> p) + .handle(CHILD_A, p -> { + throw new IllegalStateException("boom"); + }) + .handle(FINISH, p -> finished.incrementAndGet()) + .trackWorkflows(parent) + .start()) { + WorkflowStatus status = run.await(Duration.ofSeconds(20)); + assertEquals(WorkflowState.FAILED, status.state); + assertEquals(NodeStatus.FAILED, status.node("sub").orElseThrow().status); + assertEquals(NodeStatus.SKIPPED, status.node("finish").orElseThrow().status); + assertEquals(0, finished.get()); + } + } + } + + @Test + @Timeout(30) + void rejectsChildStepsThatLoseStateAtSubmit(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("sw.db").toString()).open()) { + // A child has no submit-time payload map: a payloadless plain task step can't run. + Workflow noPayload = + Workflow.named("child").step(Step.of("a", CHILD_A).build()); + Workflow parentNoPayload = Workflow.named("p1").subWorkflow("sub", noPayload); + assertThrows(TaskitoException.class, () -> queue.submitWorkflow(parentNoPayload)); + + // A child has no callable registry across the submit boundary. + Workflow callableChild = Workflow.named("child") + .step("a", CHILD_A, 1) + .step(Step.of("b", CHILD_B, 2) + .after("a") + .condition(ctx -> true) + .build()); + Workflow parentCallable = Workflow.named("p2").subWorkflow("sub", callableChild); + assertThrows(TaskitoException.class, () -> queue.submitWorkflow(parentCallable)); + } + } +}