feat: retry the parent when a child workflow started with asChildOf fails to start

src/main/java/com/airbnb/skipper/IWorkflowFactory.kt

@@ -283,6 +283,16 @@ class InvocationBuilder<T : Workflow>(
         return this
     }
 
+    /**
+     * Start this workflow as a child of [parent], from inside [parent]'s workflow method. Records
+     * [parent] as the child's parent workflow, the same as [parentWorkflowId] with [parent]'s id.
+     *
+     * A child's invocation treats a failure to start it as transient: a store or scheduler timeout
+     * retries [parent] instead of failing it for good. See [WorkflowFactory] for which failures that
+     * covers.
+     */
+    fun asChildOf(parent: Workflow): InvocationBuilder<T> = parentWorkflowId(parent.id)
+
     /**
      * Detach the invocation from the workflow result: the call returns as soon as the instance is
      * persisted and scheduled, rather than when execution finishes.

src/main/java/com/airbnb/skipper/WorkflowFactory.kt

@@ -4,6 +4,7 @@ import com.airbnb.skipper.SkipperAnnotationNames.RESULT_POLLING_SLEEP_DURATION
 import com.airbnb.skipper.SkipperAnnotationNames.RESULT_POLLING_TIME_LIMIT
 import com.airbnb.skipper.SkipperAnnotationNames.SKIPPER_MAIN_THREAD_POOL
 import com.airbnb.skipper.SkipperAnnotationNames.UTC_CLOCK
+import com.airbnb.skipper.internal.InternalError
 import com.airbnb.skipper.internal.SkipperEngine
 import com.airbnb.skipper.internal.api.ActionCheckpoint
 import com.airbnb.skipper.internal.api.RunRequest
@@ -181,7 +182,11 @@ open class WorkflowFactory
                         return@MethodHandler queryResult
                     }
                     val workflowInstance =
-                        skipperEngine.startWorkflow(request)
+                        if (parentWorkflowId != null) {
+                            startChildWorkflow(request)
+                        } else {
+                            skipperEngine.startWorkflow(request)
+                        }
                     result = workflowInstance.result
 
                     // Fire-and-forget: the caller opted out of the result, so return now that
@@ -312,6 +317,51 @@ open class WorkflowFactory
             }
         }
 
+        /**
+         * Starts a child workflow from its parent's workflow method, so that a failure to start it
+         * retries the parent instead of failing it.
+         *
+         * The parent runs this on its own execution thread, so whatever it throws escapes the parent's
+         * workflow method, and there an exception Skipper does not recognize is taken for a bug in the
+         * workflow code: the parent ends in ERROR for good. But starting a child writes to the
+         * workflow store and the scheduler, and a timeout or a dropped connection on one of those
+         * writes is no bug. What the start throws reaches the parent as follows:
+         *
+         * 1. An error Skipper has already classified (a [SkipperError], or one of Skipper's internal
+         *    errors) is rethrown unchanged, so it means what it meant before: a [RetryableError], such
+         *    as an [OptimisticLockingError], still retries the parent on its own delay, and a
+         *    [NonRetryableError] still fails it.
+         * 2. A request Skipper rejected as invalid ([IllegalArgumentException] or [ValidationError]),
+         *    such as a child id that is already taken while duplicates are not allowed, is rethrown
+         *    unchanged too. Running the parent again would only repeat it, so it still fails the parent.
+         * 3. Anything else, such as a store or scheduler timeout, becomes a [RetryableError]. The
+         *    parent runs again after [CHILD_START_RETRY_DELAY] and starts the child again under the
+         *    same id, which schedules an instance the failed attempt already wrote, unless
+         *    createExistingWorkflowIsNoop is on: then it returns that instance as it is, and a child
+         *    whose task was never scheduled does not run.
+         *
+         * Only the start is covered. Waiting for a child's result happens afterwards, and the child's
+         * own failures reach the parent as before.
+         */
+        @Suppress("TooGenericExceptionCaught")
+        private fun startChildWorkflow(request: RunRequest): WorkflowInstance =
+            try {
+                skipperEngine.startWorkflow(request)
+            } catch (e: Exception) {
+                if (e is SkipperError ||
+                    e is InternalError ||
+                    e is IllegalArgumentException ||
+                    e is ValidationError
+                ) {
+                    throw e
+                }
+                throw RetryableError(
+                    "unable to start child workflow ${request.workflowId} of ${request.parentWorkflowId}: ${e.message}",
+                    ApplicationError.fromException(e),
+                    CHILD_START_RETRY_DELAY,
+                )
+            }
+
         private fun executeSignalWithRetries(request: RunRequest): SkipperEngine.SendSignalResult {
             val retries = 3
             var lastSignalError: Throwable? = null
@@ -428,4 +478,14 @@ open class WorkflowFactory
                 skipperEngine.getActionCheckpoints(workflowInstance.workflowId).map(ActionCheckpoint::toView)
             }.getLastTransientError().getOrNull()
         }
+
+        companion object {
+            /**
+             * How long a parent waits before running again after one of its children failed to start:
+             * long enough for a brief store or scheduler outage to pass, short enough not to stall a
+             * fan-out.
+             */
+            @JvmField
+            val CHILD_START_RETRY_DELAY: Duration = Duration.ofSeconds(5)
+        }
     }

src/test/java/com/airbnb/skipper/WorkflowFactoryTest.kt

@@ -155,6 +155,49 @@ class WorkflowFactoryTest {
         )
     }
 
+    private val parent = SampleWorkflow().apply { id = "parent-id" }
+
+    private fun startChildThrowing(error: Exception): Throwable {
+        whenever(skipperEngine.startWorkflow(any())).thenAnswer { throw error }
+        val child = workflowFactory.builder(SampleWorkflow::class.java, workflowId).asChildOf(parent).build()
+        return assertThrows<Throwable> { child.asyncWorkflowMethod("input") }
+    }
+
+    @Test
+    fun testChildStartInfrastructureFailureRetriesTheParent() {
+        val thrown = startChildThrowing(java.util.concurrent.TimeoutException("store timed out"))
+
+        assertThat(thrown).isInstanceOf(RetryableError::class.java)
+        assertThat((thrown as RetryableError).getNextRetryDelay()).isEqualTo(WorkflowFactory.CHILD_START_RETRY_DELAY)
+        assertThat(thrown.message).contains(workflowId).contains("parent-id").contains("store timed out")
+        assertThat(thrown.cause).isInstanceOf(ApplicationError::class.java)
+        verify(skipperEngine).startWorkflow(org.mockito.kotlin.argThat { parentWorkflowId == "parent-id" })
+    }
+
+    @Test
+    fun testChildStartClassifiedOrInvalidFailureIsRethrownUnchanged() {
+        val unchanged =
+            listOf(
+                NonRetryableError("child failed"),
+                RetryableError("child asked to retry", null, java.time.Duration.ofSeconds(1)),
+                com.airbnb.skipper.internal.InternalError("engine failure"),
+                IllegalArgumentException("invalid child request"),
+                ValidationError("invalid child input"),
+            )
+        for (error in unchanged) {
+            assertThat(startChildThrowing(error)).isSameAs(error)
+        }
+    }
+
+    @Test
+    fun testStartFailureWithoutAParentIsNotReclassified() {
+        val error = IllegalStateException("store timed out")
+        whenever(skipperEngine.startWorkflow(any())).thenAnswer { throw error }
+        val workflow = workflowFactory.builder(SampleWorkflow::class.java, workflowId).build()
+
+        assertThat(assertThrows<Throwable> { workflow.asyncWorkflowMethod("input") }).isSameAs(error)
+    }
+
     /**
      * A detached invocation returns before the workflow executes, so its future must not report
      * success — a caller that awaits it by mistake would otherwise take an immediate completion as

src/test/java/com/airbnb/skipper/integtest/ChildWorkflowStartFailureTest.kt

@@ -0,0 +1,229 @@
+package com.airbnb.skipper.integtest
+
+import com.airbnb.skipper.ComponentFactory
+import com.airbnb.skipper.IWorkflowFactory
+import com.airbnb.skipper.OptimisticLockingError
+import com.airbnb.skipper.Workflow
+import com.airbnb.skipper.WorkflowInstance
+import com.airbnb.skipper.WorkflowMethod
+import com.airbnb.skipper.internal.SkipperEngine
+import com.airbnb.skipper.internal.SkipperSchedulerManager
+import com.airbnb.skipper.internal.api.RunRequest
+import com.airbnb.skipper.internal.scheduler.ScheduleRequest
+import com.airbnb.skipper.internal.scheduler.Scheduler
+import com.airbnb.skipper.internal.scheduler.Task
+import com.airbnb.skipper.internal.scheduler.sqlite.SqliteScheduler
+import com.airbnb.skipper.internal.storage.WorkflowCreationRequest
+import com.airbnb.skipper.internal.storage.WorkflowStore
+import com.airbnb.skipper.internal.storage.sqlite.SqliteWorkflowStore
+import com.airbnb.skipper.testutils.TestHelper
+import com.airbnb.skipper.testutils.TestRequestContext
+import com.airbnb.skipper.testutils.TestRuntime
+import com.airbnb.skipper.util.ExtraRequestData
+import java.util.concurrent.TimeoutException
+import java.util.concurrent.atomic.AtomicInteger
+import javax.inject.Inject
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.parallel.Execution
+import org.junit.jupiter.api.parallel.ExecutionMode
+
+/**
+ * A parent workflow starts its children on its own execution thread, so a store or scheduler failure
+ * while starting one of them escapes the parent's workflow method. Started with
+ * [com.airbnb.skipper.InvocationBuilder.asChildOf], a transient one (a timeout) retries the parent,
+ * which then starts the remaining children and completes. Started without it, the same failure is
+ * taken for a bug in the workflow code and fails the parent terminally. A request the
+ * store rejects as invalid fails the parent either way, since retrying it cannot succeed.
+ *
+ * Each test fails the first store or scheduler call for the second of three children.
+ */
+@Execution(ExecutionMode.SAME_THREAD)
+class ChildWorkflowStartFailureTest {
+    /** Throws [error] from the first call that targets [workflowId]; every other call passes through. */
+    class Fault(
+        private val workflowId: String,
+        private val error: () -> Exception,
+    ) {
+        val hits = AtomicInteger()
+
+        fun maybeThrow(id: String) {
+            if (id == workflowId && hits.incrementAndGet() == 1) {
+                throw error()
+            }
+        }
+    }
+
+    class FaultyStore(
+        private val delegate: WorkflowStore,
+        private val fault: Fault?,
+    ) : WorkflowStore by delegate {
+        override fun createWorkflow(request: WorkflowCreationRequest): WorkflowInstance {
+            fault?.maybeThrow(request.workflowId)
+            return delegate.createWorkflow(request)
+        }
+    }
+
+    class FaultyScheduler(
+        private val delegate: Scheduler,
+        private val fault: Fault?,
+    ) : Scheduler by delegate {
+        override fun <T> schedule(request: ScheduleRequest<T>): Task<T> {
+            if (request.type == Task.Type.WORKFLOW) {
+                fault?.maybeThrow(request.id)
+            }
+            return delegate.schedule(request)
+        }
+    }
+
+    open class ChildWorkflow : Workflow() {
+        @WorkflowMethod
+        open fun run() {}
+    }
+
+    open class FanOutWorkflow : Workflow() {
+        @Inject lateinit var workflowFactory: IWorkflowFactory
+
+        @WorkflowMethod
+        open fun run(): String {
+            for (i in 1..CHILDREN) {
+                workflowFactory.builder(ChildWorkflow::class.java, childId(id, i)).runAsync().detached().asChildOf(this).build().run()
+            }
+            return "started $CHILDREN children"
+        }
+    }
+
+    open class PlainFanOutWorkflow : Workflow() {
+        @Inject lateinit var workflowFactory: IWorkflowFactory
+
+        @WorkflowMethod
+        open fun run(): String {
+            for (i in 1..CHILDREN) {
+                workflowFactory.builder(ChildWorkflow::class.java, childId(id, i)).runAsync().detached().build().run()
+            }
+            return "started $CHILDREN children"
+        }
+    }
+
+    private lateinit var engine: SkipperEngine
+    private lateinit var scheduler: Scheduler
+    private lateinit var manager: SkipperSchedulerManager
+
+    @AfterEach
+    fun tearDown() {
+        if (::manager.isInitialized) manager.forceStop()
+    }
+
+    private fun start(
+        storeFault: Fault? = null,
+        schedulerFault: Fault? = null,
+    ) {
+        val deps = TestRuntime()
+        deps.config.workflowStore = ComponentFactory { FaultyStore(SqliteWorkflowStore.Factory().create(it), storeFault) }
+        deps.config.scheduler = ComponentFactory { FaultyScheduler(SqliteScheduler.Factory().create(it), schedulerFault) }
+        val runtime = deps.runtime
+        engine = runtime.skipperEngine.get()
+        scheduler = runtime.scheduler.get()
+        manager = runtime.skipperSchedulerManager.get()
+        manager.start()
+    }
+
+    private fun startParent(
+        parentId: String,
+        parentClass: Class<out Workflow> = FanOutWorkflow::class.java,
+    ) {
+        engine.startWorkflow(
+            RunRequest.builder()
+                .workflowId(parentId)
+                .workflowClass(parentClass)
+                .workflowMethod("run")
+                .requestContext(TestRequestContext.builder().build())
+                .extraRequestData(ExtraRequestData())
+                .runAsync(true) // hand it to the scheduler so the task handler retries it like any other task
+                .build(),
+        )
+    }
+
+    private fun awaitTerminal(workflowId: String): WorkflowInstance =
+        TestHelper(engine, scheduler, workflowId)
+            .waitForWorkflowToReachStatus(setOf(WorkflowInstance.Status.COMPLETED, WorkflowInstance.Status.ERROR))
+
+    private fun assertParentRetriedAndCompleted(
+        parentId: String,
+        fault: Fault,
+    ) {
+        val parent = awaitTerminal(parentId)
+        assertEquals(WorkflowInstance.Status.COMPLETED, parent.status, "a transient child-start failure must not fail the parent")
+        assertEquals("started $CHILDREN children", parent.result.get())
+        assertEquals(2, fault.hits.get(), "the parent is retried and starts the failed child again")
+        for (i in 1..CHILDREN) {
+            val child = awaitTerminal(childId(parentId, i))
+            assertEquals(WorkflowInstance.Status.COMPLETED, child.status)
+            assertEquals(parentId, child.parentWorkflowId)
+        }
+    }
+
+    @Test
+    fun schedulerTimeoutWhileSchedulingAChildRetriesTheParent() {
+        val parentId = "fan-out-${System.nanoTime()}"
+        // The child's instance commits, then inserting its task times out. The retried parent finds the
+        // instance already there and schedules it again.
+        val fault = Fault(childId(parentId, 2)) { TimeoutException("injected timeout inserting the child's task") }
+        start(schedulerFault = fault)
+        startParent(parentId)
+        assertParentRetriedAndCompleted(parentId, fault)
+    }
+
+    @Test
+    fun storeTimeoutWhileCreatingAChildRetriesTheParent() {
+        val parentId = "fan-out-${System.nanoTime()}"
+        val fault = Fault(childId(parentId, 2)) { TimeoutException("injected timeout creating the child's instance") }
+        start(storeFault = fault)
+        startParent(parentId)
+        assertParentRetriedAndCompleted(parentId, fault)
+    }
+
+    @Test
+    fun retryableErrorWhileCreatingAChildRetriesTheParent() {
+        val parentId = "fan-out-${System.nanoTime()}"
+        // Already a RetryableError, so it is rethrown unchanged and the executor retries the parent on it.
+        val fault = Fault(childId(parentId, 2)) { OptimisticLockingError("injected conflict creating the child's instance") }
+        start(storeFault = fault)
+        startParent(parentId)
+        assertParentRetriedAndCompleted(parentId, fault)
+    }
+
+    @Test
+    fun invalidChildRequestStillFailsTheParent() {
+        val parentId = "fan-out-${System.nanoTime()}"
+        val fault = Fault(childId(parentId, 2)) { IllegalArgumentException("injected invalid child request") }
+        start(storeFault = fault)
+        startParent(parentId)
+
+        val parent = awaitTerminal(parentId)
+        assertEquals(WorkflowInstance.Status.ERROR, parent.status)
+        assertEquals(1, fault.hits.get(), "an invalid request is not retried")
+    }
+
+    @Test
+    fun invocationWithoutAParentStillFailsTheParentOnATransientStartFailure() {
+        val parentId = "fan-out-${System.nanoTime()}"
+        val fault = Fault(childId(parentId, 2)) { TimeoutException("injected timeout creating the child's instance") }
+        start(storeFault = fault)
+        startParent(parentId, PlainFanOutWorkflow::class.java)
+
+        val parent = awaitTerminal(parentId)
+        assertEquals(WorkflowInstance.Status.ERROR, parent.status, "only a child invocation retries the parent")
+        assertEquals(1, fault.hits.get())
+    }
+
+    private companion object {
+        const val CHILDREN = 3
+
+        fun childId(
+            parentId: String,
+            i: Int
+        ) = "$parentId-child-$i"
+    }
+}