feat(java): workflow fan-out / fan-in

sdks/java/README.md

@@ -95,8 +95,22 @@ try (Worker worker = queue.worker()
 ```
 
 A failed step (after its retries) fails the run and skips downstream nodes;
-`run.cancel()` skips pending nodes. Fan-out, conditions, gates, and sagas are
-not yet exposed.
+`run.cancel()` skips pending nodes.
+
+**Fan-out / fan-in** map a step over a producer's result list and gather the
+results:
+
+```java
+Workflow wf = Workflow.named("pipeline")
+        .step("seed", seed, 4)                          // returns List.of(1,2,3,4)
+        .fanOut("square", square, "each", "seed")       // runs square(x) per item
+        .fanIn("sum", sum, "all", "square");            // sum receives [1,4,9,16]
+```
+
+`trackWorkflows()` advances run state from worker outcomes, so **every worker
+that processes workflow jobs must enable it** — in a multi-worker deployment, a
+run stalls on any node finished by a worker that did not opt in. Conditions,
+gates, and sagas are not yet exposed.
 
 ### Middleware
 

sdks/java/src/main/java/org/byteveda/taskito/DefaultQueue.java

@@ -4,10 +4,12 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 import org.byteveda.taskito.locks.Lock;
 import org.byteveda.taskito.locks.LockInfo;
@@ -281,20 +283,52 @@ public long registerPeriodic(PeriodicTask task) {
     @Override
     public WorkflowRun submitWorkflow(Workflow workflow) {
         List<Step> steps = workflow.steps();
+        Set<String> deferred = deferredNodes(steps);
         List<Map<String, Object>> specs = new ArrayList<>(steps.size());
-        String[] names = new String[steps.size()];
-        byte[][] payloads = new byte[steps.size()][];
-        for (int i = 0; i < steps.size(); i++) {
-            Step step = steps.get(i);
+        // Deferred nodes (fan-out/fan-in + their downstream) have no job — and so
+        // no payload — at submit; the tracker enqueues them at runtime.
+        List<String> payloadNames = new ArrayList<>();
+        List<byte[]> payloads = new ArrayList<>();
+        for (Step step : steps) {
             specs.add(stepSpec(step));
-            names[i] = step.name;
-            payloads[i] = serializer.serialize(step.payload);
+            if (!deferred.contains(step.name)) {
+                payloadNames.add(step.name);
+                payloads.add(serializer.serialize(step.payload));
+            }
         }
         String runId = backend.submitWorkflow(
-                workflow.name(), workflow.version(), encode(specs), names, payloads, null, null, new String[0]);
+                workflow.name(),
+                workflow.version(),
+                encode(specs),
+                payloadNames.toArray(new String[0]),
+                payloads.toArray(new byte[0][]),
+                null,
+                null,
+                deferred.toArray(new String[0]));
         return new WorkflowRun(backend, VIEWS, runId, workflow.name());
     }
 
+    /** Fan-out/fan-in nodes, plus everything transitively downstream of them, are deferred. */
+    private static Set<String> deferredNodes(List<Step> steps) {
+        Set<String> deferred = new HashSet<>();
+        for (Step step : steps) {
+            if (step.fanOut != null || step.fanIn != null) {
+                deferred.add(step.name);
+            }
+        }
+        boolean changed = true;
+        while (changed) {
+            changed = false;
+            for (Step step : steps) {
+                if (!deferred.contains(step.name) && step.after.stream().anyMatch(deferred::contains)) {
+                    deferred.add(step.name);
+                    changed = true;
+                }
+            }
+        }
+        return deferred;
+    }
+
     @Override
     public Optional<WorkflowStatus> workflowStatus(String runId) {
         return backend.getWorkflowStatusJson(runId).map(json -> decode(json, WorkflowStatus.class));
@@ -323,6 +357,12 @@ private static Map<String, Object> stepSpec(Step step) {
         if (step.priority != null) {
             spec.put("priority", step.priority);
         }
+        if (step.fanOut != null) {
+            spec.put("fanOut", step.fanOut);
+        }
+        if (step.fanIn != null) {
+            spec.put("fanIn", step.fanIn);
+        }
         return spec;
     }
 

sdks/java/src/main/java/org/byteveda/taskito/internal/JniQueueBackend.java

@@ -219,6 +219,60 @@ public void cancelWorkflowRun(String runId) {
         NativeWorkflows.cancelWorkflowRun(handle, runId);
     }
 
+    @Override
+    public Optional<String> getWorkflowPlanJson(String runId) {
+        return Optional.ofNullable(NativeWorkflows.getWorkflowPlan(handle, runId));
+    }
+
+    @Override
+    public Optional<String> workflowNodeForJobJson(String jobId) {
+        return Optional.ofNullable(NativeWorkflows.workflowNodeForJob(handle, jobId));
+    }
+
+    @Override
+    public String[] expandFanOut(
+            String runId,
+            String parentNode,
+            String[] childNames,
+            byte[][] childPayloads,
+            String taskName,
+            String queue,
+            int maxRetries,
+            long timeoutMs,
+            int priority) {
+        return NativeWorkflows.expandFanOut(
+                handle, runId, parentNode, childNames, childPayloads, taskName, queue, maxRetries, timeoutMs, priority);
+    }
+
+    @Override
+    public Optional<String> checkFanOutCompletionJson(String runId, String parentNode) {
+        return Optional.ofNullable(NativeWorkflows.checkFanOutCompletion(handle, runId, parentNode));
+    }
+
+    @Override
+    public String createDeferredJob(
+            String runId,
+            String nodeName,
+            byte[] payload,
+            String taskName,
+            String queue,
+            int maxRetries,
+            long timeoutMs,
+            int priority) {
+        return NativeWorkflows.createDeferredJob(
+                handle, runId, nodeName, payload, taskName, queue, maxRetries, timeoutMs, priority);
+    }
+
+    @Override
+    public void cascadeSkipPending(String runId) {
+        NativeWorkflows.cascadeSkipPending(handle, runId);
+    }
+
+    @Override
+    public Optional<String> finalizeRunIfTerminal(String runId) {
+        return Optional.ofNullable(NativeWorkflows.finalizeRunIfTerminal(handle, runId));
+    }
+
     @Override
     public WorkerControl startWorker(WorkerBridge bridge, String optionsJson) {
         return new JniWorkerControl(NativeQueue.runWorker(handle, bridge, optionsJson));

sdks/java/src/main/java/org/byteveda/taskito/internal/NativeWorkflows.java

@@ -35,4 +35,45 @@ public static native String markWorkflowNodeResult(
     public static native String getWorkflowStatus(long handle, String runId);
 
     public static native void cancelWorkflowRun(long handle, String runId);
+
+    // ── Fan-out / fan-in orchestration (driven by the worker-side tracker) ──
+
+    /** Returns the run's nodes with predecessors + step metadata (JSON), or {@code null}. */
+    public static native String getWorkflowPlan(long handle, String runId);
+
+    /** Returns {@code {runId, nodeName}} for a job (JSON), or {@code null} if non-workflow. */
+    public static native String workflowNodeForJob(long handle, String jobId);
+
+    /** Expand a fan-out parent into one child job per payload; returns the child job ids. */
+    public static native String[] expandFanOut(
+            long handle,
+            String runId,
+            String parentNode,
+            String[] childNames,
+            byte[][] childPayloads,
+            String taskName,
+            String queue,
+            int maxRetries,
+            long timeoutMs,
+            int priority);
+
+    /** Returns {@code {succeeded, childJobIds}} once all children settle (JSON), else {@code null}. */
+    public static native String checkFanOutCompletion(long handle, String runId, String parentNode);
+
+    /** Enqueue a job for a deferred node (e.g. the fan-in collector); returns the job id. */
+    public static native String createDeferredJob(
+            long handle,
+            String runId,
+            String nodeName,
+            byte[] payload,
+            String taskName,
+            String queue,
+            int maxRetries,
+            long timeoutMs,
+            int priority);
+
+    public static native void cascadeSkipPending(long handle, String runId);
+
+    /** Finalize the run if every node is terminal; returns the final state, or {@code null}. */
+    public static native String finalizeRunIfTerminal(long handle, String runId);
 }

sdks/java/src/main/java/org/byteveda/taskito/spi/QueueBackend.java

@@ -108,22 +108,79 @@ default long registerPeriodic(
     }
 
     // ── Workflows ───────────────────────────────────────────────────
-    String submitWorkflow(
+    // Optional capability: default to throwing so existing custom backends keep
+    // compiling and fail explicitly only when workflows are actually used.
+    String WORKFLOWS_UNSUPPORTED = "workflows not supported by this backend";
+
+    default String submitWorkflow(
             String name,
             int version,
             String stepsJson,
             String[] payloadNames,
             byte[][] payloads,
             String queueDefault,
             String paramsJson,
-            String[] deferredNames);
+            String[] deferredNames) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
 
     /** Record a node's terminal outcome; returns the run's final state, or {@code null}. */
-    String markWorkflowNodeResult(String jobId, boolean succeeded, String error, boolean skipCascade);
+    default String markWorkflowNodeResult(String jobId, boolean succeeded, String error, boolean skipCascade) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
+
+    default Optional<String> getWorkflowStatusJson(String runId) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
+
+    default void cancelWorkflowRun(String runId) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
+
+    default Optional<String> getWorkflowPlanJson(String runId) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
 
-    Optional<String> getWorkflowStatusJson(String runId);
+    default Optional<String> workflowNodeForJobJson(String jobId) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
+
+    default String[] expandFanOut(
+            String runId,
+            String parentNode,
+            String[] childNames,
+            byte[][] childPayloads,
+            String taskName,
+            String queue,
+            int maxRetries,
+            long timeoutMs,
+            int priority) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
+
+    default Optional<String> checkFanOutCompletionJson(String runId, String parentNode) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
 
-    void cancelWorkflowRun(String runId);
+    default String createDeferredJob(
+            String runId,
+            String nodeName,
+            byte[] payload,
+            String taskName,
+            String queue,
+            int maxRetries,
+            long timeoutMs,
+            int priority) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
+
+    default void cascadeSkipPending(String runId) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
+
+    default Optional<String> finalizeRunIfTerminal(String runId) {
+        throw new UnsupportedOperationException(WORKFLOWS_UNSUPPORTED);
+    }
 
     // ── Worker ──────────────────────────────────────────────────────
     /** Start a worker that dispatches jobs to {@code bridge}; returns its control. */

sdks/java/src/main/java/org/byteveda/taskito/worker/Worker.java

@@ -141,7 +141,7 @@ public Builder on(EventName name, Consumer<OutcomeEvent> listener) {
 
         /** Drive workflow node and run state from this worker's job outcomes. */
         public Builder trackWorkflows() {
-            WorkflowTracker tracker = new WorkflowTracker(backend);
+            WorkflowTracker tracker = new WorkflowTracker(backend, serializer);
             on(EventName.SUCCESS, tracker::onSuccess);
             on(EventName.DEAD, tracker::onDead);
             return this;

sdks/java/src/main/java/org/byteveda/taskito/workflows/PlanNode.java

@@ -0,0 +1,59 @@
+package org.byteveda.taskito.workflows;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Collections;
+import java.util.List;
+
+/** One node of a run's plan: predecessors plus the metadata needed to enqueue deferred work. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+final class PlanNode {
+    final String name;
+    final List<String> predecessors;
+    final String taskName;
+    final String queue;
+    final Integer maxRetries;
+    final Long timeoutMs;
+    final Integer priority;
+    final String fanOut;
+    final String fanIn;
+
+    @JsonCreator
+    PlanNode(
+            @JsonProperty("name") String name,
+            @JsonProperty("predecessors") List<String> predecessors,
+            @JsonProperty("taskName") String taskName,
+            @JsonProperty("queue") String queue,
+            @JsonProperty("maxRetries") Integer maxRetries,
+            @JsonProperty("timeoutMs") Long timeoutMs,
+            @JsonProperty("priority") Integer priority,
+            @JsonProperty("fanOut") String fanOut,
+            @JsonProperty("fanIn") String fanIn) {
+        this.name = name;
+        this.predecessors = predecessors == null ? Collections.emptyList() : predecessors;
+        this.taskName = taskName;
+        this.queue = queue;
+        this.maxRetries = maxRetries;
+        this.timeoutMs = timeoutMs;
+        this.priority = priority;
+        this.fanOut = fanOut;
+        this.fanIn = fanIn;
+    }
+
+    int retries() {
+        return maxRetries == null ? 3 : maxRetries;
+    }
+
+    long timeout() {
+        return timeoutMs == null ? 300_000L : timeoutMs;
+    }
+
+    int priorityOrDefault() {
+        return priority == null ? 0 : priority;
+    }
+
+    String queueOrDefault() {
+        return queue == null ? "default" : queue;
+    }
+}