diff --git a/formal/tla/README.md b/formal/tla/README.md index 5720140..5a06bcc 100644 --- a/formal/tla/README.md +++ b/formal/tla/README.md @@ -20,10 +20,13 @@ the store, checked with TLC. The property it pins down: (airbnb/skipper#12) broke. - **Starting**: `startWorkflow` may fail before the workflow runs, and the caller is expected to retry. A retry may be a no-op only once the workflow task exists. +- **One run at a time**: no two workers run the workflow task at once, from loading the instance + to the final write (`NoOverlappingRuns`). Two runs at once would execute the same actions + concurrently. - **Liveness**: `EventuallyDone == <>[]Done`: the workflow is eventually created and stays terminal, `RETRIES_EXHAUSTED`, or parked in the DLQ. -The two invariants are the safety half of the argument, and they are much cheaper to check than the +`RunningIsDriven` and `WaitingIsDriven` are the safety half of the argument, and they are much cheaper to check than the liveness property. Each says that something durable, or an obligation someone will honour, will run the workflow again. @@ -33,8 +36,11 @@ the workflow again. - Scheduler workers. Each worker can fetch a task, lease it, renew the lease, run the task handler, and make the final versioned remove or reschedule (`SkipperSchedulerManager.finishTask`). +- One or more `startWorkflow` callers, each in its own JVM with its own in-memory queue. - The in-process `startWorkflow` path, which writes a backstop row and keeps an in-memory copy of the task in the caller's JVM. +- The start that finds a workflow stranded (`RUNNING` with no task row) and schedules it only in + the persistent scheduler, due after a lease (`StartRecover`). - The workflow body, abstracted to `NWaits` calls to `waitUntil(cond, timeout)` followed by a return. Wait *k*'s condition holds once *k* signals have been applied. Optionally the body can also fail as transient, non-retryable, or retries-exhausted. @@ -46,12 +52,16 @@ them. The fault kinds are switched on per model: worker/JVM crash, handler timeo expiry, a store error after the workflow row committed, `startWorkflow` failing after creating the row, and a signal call failing before it schedules. +`NoLongPauses` is an assumption some models make: no start stalls between its two writes, and no +in-memory copy waits in its queue, for as long as a lease. Without it, an in-memory copy can outlive +its lease and run next to the backstop row's delivery, which `inmem_start` already records. + Not modelled yet: compensation, execution timeouts, cancellation, child workflows, persisted-signal replay, task partitioning, several workflows sharing workers. | Action | Code | |---|---| -| `StartCreate`, `StartSched` | `SkipperEngine.startWorkflow`, `scheduleExecution`, `SchedulerExecutionQueue.schedule` | +| `StartCreate`, `StartSched`, `StartRecover` | `SkipperEngine.startWorkflow` (its `isStranded` check), `scheduleExecution`, `SchedulerExecutionQueue.schedule` / `scheduleAfterLease` | | `Fetch`, `MarkFailed`, `Finish` | `SqliteScheduler.fetch`, `SkipperSchedulerManager.handleTask` / `finishTask` / `scheduleRerunNow` | | `Take` | `SkipperSchedulerManager.startTakingTasks` on the in-memory queue | | `WfRead`, `WfRun`, `WfPersist`, `WfTimers` | `WorkflowExecutionTaskHandler.handle` (its instance read, then its timer read and the run) / `persistExecutionResult`, `Workflow.waitUntil` | @@ -87,31 +97,54 @@ CI runs the same script in `.github/workflows/formal.yml` (job `tla-model-check` `formal/` changes. It keeps TLC's output as the `tlc-output` run artifact. The job isn't a required check. -## Results (main at 619a628, plus the timer re-arm from airbnb/skipper#40) +## Results (main at 619a628, plus airbnb/skipper#40's timer re-arm and #46's stranded-start recovery) | Model | Scenario | Result | |---|---|---| | `baseline` | default gates, 2 workers, one signal, no faults | ok | | `baseline_user_failures` | as above, the workflow body may fail | ok | | `pr12_regression` | `BUMP_TASK_VERSION_ON_HONORED_LEASE` off | `RunningIsDriven` violated: the #12 race, found again (calibration) | -| `start_retry` | `startWorkflow` fails after creating the row; caller retries | **stuck forever** | +| `start_retry` | `startWorkflow` fails after creating the row; caller retries | ok (was stuck forever; fixed in #46) | | `start_retry_noop_off` | as above, `CREATE_EXISTING_WORKFLOW_IS_NOOP` off | ok | | `timer_arming_crash` | a worker crashes while running the workflow | ok (was stuck forever; fixed in #40) | | `timer_arming_store_error` | one store error after the workflow row committed | ok (was stuck forever; fixed in #40) | | `inmem_start` | in-process start, no faults | **stuck forever** | | `lease_faults` | a lease lapses under a live holder, 2 workers | **stuck forever** | | `signal_schedule` | a signal call fails before scheduling | `RunningIsDriven` violated; liveness holds only through the wait timer | +| `start_race` | two in-process starts from different JVMs, either may fail after creating the row; `NoLongPauses` | ok | +| `start_race_unguarded` | as above, with the stranded start scheduled like any other start | `NoOverlappingRuns` violated (calibration) | | `rearm_overlap` | a timer task goes missing and a lease lapses, so two executions schedule the same timer's task; checks `TimerSchedulesKeepLeases` only | ok | +`NoOverlappingRuns` is checked in every model except `inmem_start`, `lease_faults` and +`rearm_overlap`. Their leases can lapse under a queued copy or a live holder, the stale holders of +finding 3, so two runs can overlap there. + The findings behind these results: -1. **A retried `startWorkflow` can leave a new workflow `RUNNING` with no task.** (`createWorkflow` inserts - the row as `RUNNING`, not `CREATED`; trace validation caught the model's earlier `CREATED`.) - - `createWorkflow` (`SkipperEngine.kt:102`) and `scheduleExecution` (`:143`) are separate - transactions. +1. **A retried `startWorkflow` can leave a new workflow `RUNNING` with no task.** Fixed in + airbnb/skipper#46. (`createWorkflow` inserts the row as `RUNNING`, not `CREATED`; trace + validation caught the model's earlier `CREATED`.) + - `createWorkflow` and `scheduleExecution` are separate transactions. - If the call fails between them, the caller's retry hits `EntityAlreadyExists`. - - With `CREATE_EXISTING_WORKFLOW_IS_NOOP`, which is on by default (`:128`), that retry returns - without scheduling. + - With `CREATE_EXISTING_WORKFLOW_IS_NOOP`, which is on by default, that retry returned without + scheduling. + - The repair: a start that finds the workflow `RUNNING` with no task row schedules it anyway + (`SkipperEngine.isStranded`). + - That opens a race. The "stranded" workflow may belong to a start that is only slow and is about + to schedule it in memory. If the second start also schedules it like a normal start, the workflow + runs twice at once: + 1. One start writes the row. + 2. Another start finds the workflow `RUNNING` with no task and schedules it. + 3. A worker leases that task. + 4. The first start's in-memory schedule honours the lease but still queues its own copy. + 5. A second worker takes the copy. + + `start_race_unguarded` finds this. + - So the recovering start writes only the backstop row, due after a lease, with no in-memory + copy (`SchedulerExecutionQueue.scheduleAfterLease`). Whichever row write comes first, the + other honours its lease, and only one copy runs. `start_race` checks it, and trace + validation checks the engine against it on a recorded stranded child + (`ChildWorkflowStartFailureTest`). 2. **A `WAITING` workflow can end up with a timer that never fires.** Fixed in airbnb/skipper#40. - The timer row is written in the same transaction as `WAITING` diff --git a/formal/tla/SkipperWorkflowLiveness.tla b/formal/tla/SkipperWorkflowLiveness.tla index 7456519..86cb8a1 100644 --- a/formal/tla/SkipperWorkflowLiveness.tla +++ b/formal/tla/SkipperWorkflowLiveness.tla @@ -2,7 +2,7 @@ (***************************************************************************************) (* How the Skipper engine drives one workflow instance to completion, as a transition *) (* system over the durable store (workflow row, timer rows, scheduler task rows) and *) -(* the processes that touch it: scheduler workers, the caller of startWorkflow, a *) +(* the processes that touch it: scheduler workers, the callers of startWorkflow, a *) (* signal sender and an operator working the DLQ. *) (* *) (* The property under test: once a workflow is created it eventually settles in a *) @@ -31,7 +31,12 @@ CONSTANTS BumpOnHonoredLease, \* FeatureGate BUMP_TASK_VERSION_ON_HONORED_LEASE (default on) InMemoryStart, \* startWorkflow may take the in-process path (synchronous start) SignalInMemory, \* signals take the in-process path: FORCE_SIGNAL_WORKFLOW_EXEC_IN_SCHEDULER off - MaxStartCalls, \* startWorkflow calls the caller makes, beyond retries after a failure + MaxStartCalls, \* startWorkflow calls each caller makes, beyond retries after a failure + Callers, \* startWorkflow callers, each in its own JVM with its own in-memory queue + RecoverAfterLease, \* a start that finds the workflow stranded schedules it after a lease (FALSE: + \* like any other start, which only calibrates start_race_unguarded) + NoLongPauses, \* no start stalls between its two writes, and no in-memory copy waits in its + \* queue, for as long as a lease UserFailures, \* the workflow body may fail: transient, non-retryable, retries exhausted FreeBody \* the body's outcome is unconstrained (trace validation runs real workflows) @@ -51,6 +56,7 @@ ASSUME /\ Workers # {} /\ NoopOnExisting \in BOOLEAN /\ BumpOnHonoredLease \in BOOLEAN /\ UserFailures \in BOOLEAN /\ InMemoryStart \in BOOLEAN /\ SignalInMemory \in BOOLEAN /\ FreeBody \in BOOLEAN /\ MaxStartCalls \in Nat \ {0} + /\ Callers # {} /\ RecoverAfterLease \in BOOLEAN /\ NoLongPauses \in BOOLEAN ----------------------------------------------------------------------------------------- (* Task ids. The WORKFLOW task's id is the workflow id; TIMER task k belongs to wait k. *) @@ -78,8 +84,8 @@ VARIABLES tm, \* timer rows: k -> NONE | ACTIVE | EXPIRED | CANCELLED row, \* scheduler task rows: id -> [st: NONE|PENDING|RUNNING|FAILED, ra, ver, rc] wk, \* per-worker local state (see IdleW) - cpc, \* startWorkflow caller: ready (a call is due) | sched | done - lq, \* the caller JVM's in-memory queue: {} or {held snapshot} + cpc, \* startWorkflow caller c: ready (a call is due) | sched | recover | done + lq, \* caller c's JVM's in-memory queue: {} or {held snapshot} spc, \* signal sender: idle | exec | sched sRead, \* workflow version the in-flight signal read sigSent, faults, renewals, requeues, starts @@ -94,7 +100,8 @@ TypeOK == /\ row \in [TaskIds -> [st: {"NONE", "PENDING", "RUNNING", "FAILED"}, ra: Nat, ver: Nat, rc: Nat]] /\ \A p \in Workers : wk[p].pc \in {"idle", "dlq", "wf_exec", "wf_run", "wf_persist", "wf_timers", "wf_rearm", "tm_exec", "tm_sched", "finish"} - /\ cpc \in {"ready", "sched", "done"} + /\ cpc \in [Callers -> {"ready", "sched", "recover", "done"}] + /\ starts \in [Callers -> Nat] /\ spc \in {"idle", "exec", "sched"} ----------------------------------------------------------------------------------------- @@ -103,6 +110,12 @@ TypeOK == Live(t) == row[t].st \in {"PENDING", "RUNNING"} Queued(t) == row[t].st \in {"PENDING", "RUNNING", "FAILED"} \* FAILED = in the DLQ +(* Every in-memory copy of the workflow task, in any caller's JVM. *) +InMemoryCopies == UNION {lq[c] : c \in Callers} + +(* The JVM signals run in. Its in-memory queue is one of the callers'. *) +SignalNode == CHOOSE c \in Callers : TRUE + (* Task.hasActiveLease: PENDING or RUNNING with run_after <= now + lease. *) ActiveLease(r) == r.st \in {"PENDING", "RUNNING"} /\ r.ra # LATER @@ -124,7 +137,7 @@ Sched(t, ra, honor) == SchedReq(t, ra, honor, BumpOnHonoredLease) (* as a counter, and states that differ only in how many leases came before merge. *) NewLease == LET used == {FirstLease - 1} \cup {row[t].ra : t \in TaskIds} - \cup {wk[p].held.ra : p \in Workers} \cup {h.ra : h \in lq} + \cup {wk[p].held.ra : p \in Workers} \cup {h.ra : h \in InMemoryCopies} IN 1 + (CHOOSE m \in used : \A u \in used : u <= m) (* The holder of task t's lease is still working and will renew or finish it. *) @@ -211,11 +224,12 @@ Fetch(p) == !.att = TRUE]) /\ UNCHANGED <> -(* A consumer takes the in-memory copy startWorkflow queued in this JVM. *) +(* A consumer takes the in-memory copy startWorkflow queued in its JVM. *) Take(p) == - /\ wk[p].pc = "idle" /\ lq # {} - /\ \E h \in lq : SetW(p, [IdleW EXCEPT !.pc = "wf_exec", !.held = h, !.att = TRUE]) - /\ lq' = {} + /\ wk[p].pc = "idle" + /\ \E c \in Callers : \E h \in lq[c] : + /\ SetW(p, [IdleW EXCEPT !.pc = "wf_exec", !.held = h, !.att = TRUE]) + /\ lq' = [lq EXCEPT ![c] = {}] /\ UNCHANGED <> (* Scheduler.markAsFailed, versioned; a lost race is swallowed and the row left alone. *) @@ -383,8 +397,17 @@ Renew(p) == ----------------------------------------------------------------------------------------- (* Time. A lease whose holder is gone lapses; a backoff or timer comes due. *) +(* With NoLongPauses, a lease does not lapse while something that began within it is *) +(* still to happen: a start between its two writes, or an in-memory copy of that lease *) +(* waiting to be taken. The models that use it start from a fresh workflow, so every *) +(* workflow-task lease was granted after a pending start began. *) +LeaseStillClaimed(t) == + /\ NoLongPauses /\ t = WF + /\ \/ \E c \in Callers : cpc[c] = "sched" + \/ \E h \in InMemoryCopies : h.ra = row[WF].ra + ExpireFree(t) == - /\ Live(t) /\ row[t].ra >= FirstLease /\ ~LiveHolder(t) + /\ Live(t) /\ row[t].ra >= FirstLease /\ ~LiveHolder(t) /\ ~LeaseStillClaimed(t) /\ row' = [row EXCEPT ![t].ra = DUE] /\ UNCHANGED <> @@ -394,48 +417,67 @@ Tick(t) == /\ UNCHANGED <> ----------------------------------------------------------------------------------------- -(* Environment: the caller of startWorkflow, a signal sender, an operator. *) +(* Environment: the callers of startWorkflow, a signal sender, an operator. *) + +(* A RUNNING workflow with no task row: the start that wrote it never scheduled it, or *) +(* has not yet (SkipperEngine.isStranded). The store writes a new workflow as RUNNING. *) +Stranded == wf.st = "RUNNING" /\ row[WF].st = "NONE" (* SkipperEngine.startWorkflow, part 1: createWorkflow, or on EntityAlreadyExists *) (* return the existing instance without scheduling (terminal, or the no-op gate is on, *) -(* except RETRIES_EXHAUSTED which is always re-run). *) -StartCreate == - /\ cpc = "ready" +(* except RETRIES_EXHAUSTED which is always re-run). A stranded workflow is scheduled *) +(* even with the no-op gate on; the instance read and its task lookup are one step here. *) +StartCreate(c) == + /\ cpc[c] = "ready" /\ IF ~wf.ex THEN /\ wf' = [ex |-> TRUE, st |-> "RUNNING", ver |-> 1, sig |-> 0] - /\ cpc' = "sched" + /\ cpc' = [cpc EXCEPT ![c] = "sched"] ELSE /\ UNCHANGED wf - /\ cpc' = IF (wf.st \in Terminal \/ NoopOnExisting) /\ wf.st # "RETRIES_EXHAUSTED" - THEN "done" ELSE "sched" + /\ cpc' = [cpc EXCEPT ![c] = + IF wf.st = "RETRIES_EXHAUSTED" \/ ~(wf.st \in Terminal \/ NoopOnExisting) THEN "sched" + ELSE IF Stranded THEN IF RecoverAfterLease THEN "recover" ELSE "sched" + ELSE "done"] /\ UNCHANGED <> (* SchedulerExecutionQueue.schedule for a due workflow task: persist a leased backstop *) -(* row and queue the task in this JVM, unless it is already queued here. The queued *) -(* copy carries the workflow instance as it is now (wv): the handler runs on that *) -(* payload instead of reading the store again. *) -InMemorySchedule == - \/ /\ lq = {} +(* row and queue the task in JVM n, unless it is already queued there. It queues the *) +(* copy even when the row write honoured someone else's lease. The queued copy carries *) +(* the workflow instance as it is now (wv): the handler runs on that payload instead of *) +(* reading the store again. *) +InMemorySchedule(n) == + \/ /\ lq[n] = {} /\ LET r == Sched(WF, NewLease, TRUE) IN /\ row' = [row EXCEPT ![WF] = r] - /\ lq' = {[id |-> WF, ver |-> r.ver, ra |-> r.ra, rc |-> 0, rerun |-> FALSE, wv |-> wf.ver]} - \/ /\ lq # {} + /\ lq' = [lq EXCEPT ![n] = + {[id |-> WF, ver |-> r.ver, ra |-> r.ra, rc |-> 0, rerun |-> FALSE, wv |-> wf.ver]}] + \/ /\ lq[n] # {} /\ UNCHANGED <> (* Part 2: scheduleExecution, a separate transaction. Async goes to the persistent *) (* scheduler; in-process execution persists a leased backstop row and queues the task *) (* in this JVM (SchedulerExecutionQueue.schedule). *) -StartSched == - /\ cpc = "sched" - /\ cpc' = "done" +StartSched(c) == + /\ cpc[c] = "sched" + /\ cpc' = [cpc EXCEPT ![c] = "done"] /\ \/ /\ row' = [row EXCEPT ![WF] = Sched(WF, DUE, TRUE)] /\ UNCHANGED <> - \/ InMemoryStart /\ InMemorySchedule + \/ InMemoryStart /\ InMemorySchedule(c) /\ UNCHANGED <> +(* Part 2 for a stranded workflow (SchedulerExecutionQueue.scheduleAfterLease): only the *) +(* backstop row, due after a lease, and no in-memory copy. Its start may only be slow *) +(* and about to schedule in memory: whichever row write comes first, the other honours *) +(* its lease, and the only copy that runs is the one queued against it. *) +StartRecover(c) == + /\ cpc[c] = "recover" + /\ cpc' = [cpc EXCEPT ![c] = "done"] + /\ row' = [row EXCEPT ![WF] = Sched(WF, NewLease, TRUE)] + /\ UNCHANGED <> + (* The caller invokes the workflow method again (a sync caller reading its result). *) -StartAgain == - /\ cpc = "done" /\ starts < MaxStartCalls - /\ cpc' = "ready" /\ starts' = starts + 1 +StartAgain(c) == + /\ cpc[c] = "done" /\ starts[c] < MaxStartCalls + /\ cpc' = [cpc EXCEPT ![c] = "ready"] /\ starts' = [starts EXCEPT ![c] = @ + 1] /\ UNCHANGED <> (* SkipperEngine.runSignal: read, run the signal method, persist state and RUNNING *) @@ -461,7 +503,7 @@ SignalSched == /\ IF ~SignalInMemory THEN /\ row' = [row EXCEPT ![WF] = Sched(WF, DUE, TRUE)] /\ UNCHANGED <> - ELSE InMemorySchedule + ELSE InMemorySchedule(SignalNode) /\ UNCHANGED <> (* Scheduler.requeueFailedTask from the admin DLQ view. *) @@ -479,9 +521,9 @@ Crash(p) == /\ SetW(p, IdleW) /\ UNCHANGED <> -LoseLocalQueue == - /\ lq # {} /\ Fault("crash") - /\ lq' = {} +LoseLocalQueue(c) == + /\ lq[c] # {} /\ Fault("crash") + /\ lq' = [lq EXCEPT ![c] = {}] /\ UNCHANGED <> HandlerTimeout(p) == @@ -508,9 +550,9 @@ StoreError(p) == /\ UNCHANGED <> (* startWorkflow throws after the row exists; the caller is expected to retry. *) -StartFail == - /\ cpc \in {"sched", "done"} /\ wf.ex /\ Fault("caller_fail") - /\ cpc' = "ready" +StartFail(c) == + /\ cpc[c] \in {"sched", "recover", "done"} /\ wf.ex /\ Fault("caller_fail") + /\ cpc' = [cpc EXCEPT ![c] = "ready"] /\ UNCHANGED <> SignalFail == @@ -524,8 +566,8 @@ Init == /\ tm = [k \in Timers |-> "NONE"] /\ row = [t \in TaskIds |-> NoRow] /\ wk = [p \in Workers |-> IdleW] - /\ cpc = "ready" /\ lq = {} /\ spc = "idle" /\ sRead = 0 - /\ sigSent = 0 /\ faults = 0 /\ renewals = 0 /\ requeues = 0 /\ starts = 1 + /\ cpc = [c \in Callers |-> "ready"] /\ lq = [c \in Callers |-> {}] /\ spc = "idle" /\ sRead = 0 + /\ sigSent = 0 /\ faults = 0 /\ renewals = 0 /\ requeues = 0 /\ starts = [c \in Callers |-> 1] WorkerStep(p) == \/ Fetch(p) \/ Take(p) \/ MarkFailed(p) \/ WfRead(p) \/ WfRun(p) \/ WfPersist(p) \/ WfTimers(p) @@ -534,9 +576,9 @@ WorkerStep(p) == Next == \/ \E p \in Workers : WorkerStep(p) \/ Renew(p) \/ Crash(p) \/ HandlerTimeout(p) \/ StoreError(p) \/ \E t \in TaskIds : ExpireFree(t) \/ Tick(t) \/ ExpireEarly(t) \/ Requeue(t) - \/ StartCreate \/ StartSched \/ StartFail \/ StartAgain + \/ \E c \in Callers : StartCreate(c) \/ StartSched(c) \/ StartRecover(c) \/ StartFail(c) \/ StartAgain(c) + \/ LoseLocalQueue(c) \/ SignalStart \/ SignalPersist \/ SignalSched \/ SignalFail - \/ LoseLocalQueue (* Workers keep working, time passes, the caller retries until a call succeeds, and a *) (* signal call that started runs to its end. Faults, renewals, new signals and *) @@ -544,7 +586,7 @@ Next == Fairness == /\ \A p \in Workers : WF_vars(WorkerStep(p)) /\ \A t \in TaskIds : WF_vars(ExpireFree(t)) /\ WF_vars(Tick(t)) - /\ WF_vars(StartCreate) /\ WF_vars(StartSched) + /\ \A c \in Callers : WF_vars(StartCreate(c)) /\ WF_vars(StartSched(c)) /\ WF_vars(StartRecover(c)) /\ WF_vars(SignalPersist) /\ WF_vars(SignalSched) Spec == Init /\ [][Next]_vars /\ Fairness @@ -557,8 +599,8 @@ Spec == Init /\ [][Next]_vars /\ Fairness (* or a timer task whose timer already fired and which will schedule the workflow. *) WfDriven == \/ Queued(WF) - \/ lq # {} - \/ cpc \in {"ready", "sched"} + \/ InMemoryCopies # {} + \/ \E c \in Callers : cpc[c] \in {"ready", "sched", "recover"} \/ spc = "sched" \/ \E k \in Timers : tm[k] = "EXPIRED" /\ Queued(k) @@ -590,6 +632,14 @@ TimerSchedulesKeepLeases == [][\A p \in Workers, k \in Timers : SchedulesTimer(p, k) /\ ActiveLease(row[k]) => row'[k] = row[k]]_vars +(* Worker p is running the workflow task: from loading the instance to its final write. *) +RunsWorkflow(p) == + wk[p].held.id = WF /\ wk[p].pc \in {"wf_exec", "wf_run", "wf_persist", "wf_timers", "wf_rearm", "finish"} + +(* At most one execution of the workflow at a time. Two at once would run the same *) +(* actions concurrently; Skipper's retries assume a workflow's runs follow one another. *) +NoOverlappingRuns == \A p, q \in Workers : RunsWorkflow(p) /\ RunsWorkflow(q) => p = q + (* Terminal states are absorbing. *) TerminalIsStable == [][wf.st \in Terminal => wf'.st = wf.st]_vars diff --git a/formal/tla/models/baseline.cfg b/formal/tla/models/baseline.cfg index d0dca08..6b6e6d3 100644 --- a/formal/tla/models/baseline.cfg +++ b/formal/tla/models/baseline.cfg @@ -17,6 +17,9 @@ CONSTANTS SignalInMemory = FALSE MaxStartCalls = 1 FreeBody = FALSE -INVARIANTS TypeOK RunningIsDriven WaitingIsDriven + Callers = {c1} + RecoverAfterLease = TRUE + NoLongPauses = FALSE +INVARIANTS TypeOK RunningIsDriven WaitingIsDriven NoOverlappingRuns PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/baseline_user_failures.cfg b/formal/tla/models/baseline_user_failures.cfg index 2881973..80f9f4e 100644 --- a/formal/tla/models/baseline_user_failures.cfg +++ b/formal/tla/models/baseline_user_failures.cfg @@ -17,6 +17,9 @@ CONSTANTS SignalInMemory = FALSE MaxStartCalls = 1 FreeBody = FALSE -INVARIANTS TypeOK RunningIsDriven WaitingIsDriven + Callers = {c1} + RecoverAfterLease = TRUE + NoLongPauses = FALSE +INVARIANTS TypeOK RunningIsDriven WaitingIsDriven NoOverlappingRuns PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/inmem_start.cfg b/formal/tla/models/inmem_start.cfg index 06f4e54..5687121 100644 --- a/formal/tla/models/inmem_start.cfg +++ b/formal/tla/models/inmem_start.cfg @@ -17,6 +17,9 @@ CONSTANTS SignalInMemory = FALSE MaxStartCalls = 1 FreeBody = FALSE + Callers = {c1} + RecoverAfterLease = TRUE + NoLongPauses = FALSE INVARIANTS TypeOK RunningIsDriven WaitingIsDriven PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/lease_faults.cfg b/formal/tla/models/lease_faults.cfg index 90269f6..d795799 100644 --- a/formal/tla/models/lease_faults.cfg +++ b/formal/tla/models/lease_faults.cfg @@ -17,6 +17,9 @@ CONSTANTS SignalInMemory = FALSE MaxStartCalls = 1 FreeBody = FALSE + Callers = {c1} + RecoverAfterLease = TRUE + NoLongPauses = FALSE INVARIANTS TypeOK RunningIsDriven WaitingIsDriven PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/pr12_regression.cfg b/formal/tla/models/pr12_regression.cfg index da63fc5..bdf5255 100644 --- a/formal/tla/models/pr12_regression.cfg +++ b/formal/tla/models/pr12_regression.cfg @@ -17,6 +17,9 @@ CONSTANTS SignalInMemory = FALSE MaxStartCalls = 1 FreeBody = FALSE -INVARIANTS TypeOK RunningIsDriven WaitingIsDriven + Callers = {c1} + RecoverAfterLease = TRUE + NoLongPauses = FALSE +INVARIANTS TypeOK RunningIsDriven WaitingIsDriven NoOverlappingRuns PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/rearm_overlap.cfg b/formal/tla/models/rearm_overlap.cfg index b1b879d..4f29d06 100644 --- a/formal/tla/models/rearm_overlap.cfg +++ b/formal/tla/models/rearm_overlap.cfg @@ -20,6 +20,9 @@ CONSTANTS SignalInMemory = FALSE MaxStartCalls = 1 FreeBody = FALSE + Callers = {c1} + RecoverAfterLease = TRUE + NoLongPauses = FALSE INVARIANTS TypeOK PROPERTIES TerminalIsStable TimerSchedulesKeepLeases CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/signal_schedule.cfg b/formal/tla/models/signal_schedule.cfg index 8760ee6..6efe96e 100644 --- a/formal/tla/models/signal_schedule.cfg +++ b/formal/tla/models/signal_schedule.cfg @@ -17,6 +17,9 @@ CONSTANTS SignalInMemory = FALSE MaxStartCalls = 1 FreeBody = FALSE -INVARIANTS TypeOK RunningIsDriven WaitingIsDriven + Callers = {c1} + RecoverAfterLease = TRUE + NoLongPauses = FALSE +INVARIANTS TypeOK RunningIsDriven WaitingIsDriven NoOverlappingRuns PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/start_race.cfg b/formal/tla/models/start_race.cfg new file mode 100644 index 0000000..eac183b --- /dev/null +++ b/formal/tla/models/start_race.cfg @@ -0,0 +1,27 @@ +\* Two in-process starts of the same workflow from different JVMs, either of which may fail after +\* creating the row: the second can find the workflow RUNNING with no task while the first is +\* still about to schedule it. Assumes no start or queued copy stalls for a lease. +\* expect: ok +SPECIFICATION Spec +CONSTANTS + Workers = {p1, p2} + NWaits = 1 + MaxRetries = 4 + MaxSignals = 0 + MaxFaults = 1 + MaxRenewals = 1 + MaxRequeues = 0 + Faults = {"caller_fail"} + NoopOnExisting = TRUE + BumpOnHonoredLease = TRUE + UserFailures = FALSE + InMemoryStart = TRUE + SignalInMemory = FALSE + MaxStartCalls = 1 + FreeBody = FALSE + Callers = {c1, c2} + RecoverAfterLease = TRUE + NoLongPauses = TRUE +INVARIANTS TypeOK RunningIsDriven WaitingIsDriven NoOverlappingRuns +PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone +CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/start_race_unguarded.cfg b/formal/tla/models/start_race_unguarded.cfg new file mode 100644 index 0000000..67a3c38 --- /dev/null +++ b/formal/tla/models/start_race_unguarded.cfg @@ -0,0 +1,25 @@ +\* Calibration for start_race: the stranded start is scheduled like any other start. +\* expect: NoOverlappingRuns +SPECIFICATION Spec +CONSTANTS + Workers = {p1, p2} + NWaits = 1 + MaxRetries = 4 + MaxSignals = 0 + MaxFaults = 1 + MaxRenewals = 1 + MaxRequeues = 0 + Faults = {"caller_fail"} + NoopOnExisting = TRUE + BumpOnHonoredLease = TRUE + UserFailures = FALSE + InMemoryStart = TRUE + SignalInMemory = FALSE + MaxStartCalls = 1 + FreeBody = FALSE + Callers = {c1, c2} + RecoverAfterLease = FALSE + NoLongPauses = TRUE +INVARIANTS TypeOK RunningIsDriven WaitingIsDriven NoOverlappingRuns +PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone +CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/start_retry.cfg b/formal/tla/models/start_retry.cfg index 5271862..db695b5 100644 --- a/formal/tla/models/start_retry.cfg +++ b/formal/tla/models/start_retry.cfg @@ -1,5 +1,5 @@ \* startWorkflow fails after creating the row and the caller retries, CREATE_EXISTING_WORKFLOW_IS_NOOP on (default). -\* expect: RunningIsDriven +\* expect: ok SPECIFICATION Spec CONSTANTS Workers = {p1} @@ -17,6 +17,9 @@ CONSTANTS SignalInMemory = FALSE MaxStartCalls = 1 FreeBody = FALSE -INVARIANTS TypeOK RunningIsDriven WaitingIsDriven + Callers = {c1} + RecoverAfterLease = TRUE + NoLongPauses = FALSE +INVARIANTS TypeOK RunningIsDriven WaitingIsDriven NoOverlappingRuns PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/start_retry_noop_off.cfg b/formal/tla/models/start_retry_noop_off.cfg index f6482d0..6574162 100644 --- a/formal/tla/models/start_retry_noop_off.cfg +++ b/formal/tla/models/start_retry_noop_off.cfg @@ -17,6 +17,9 @@ CONSTANTS SignalInMemory = FALSE MaxStartCalls = 1 FreeBody = FALSE -INVARIANTS TypeOK RunningIsDriven WaitingIsDriven + Callers = {c1} + RecoverAfterLease = TRUE + NoLongPauses = FALSE +INVARIANTS TypeOK RunningIsDriven WaitingIsDriven NoOverlappingRuns PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/timer_arming_crash.cfg b/formal/tla/models/timer_arming_crash.cfg index e29eae8..6493c07 100644 --- a/formal/tla/models/timer_arming_crash.cfg +++ b/formal/tla/models/timer_arming_crash.cfg @@ -17,6 +17,9 @@ CONSTANTS SignalInMemory = FALSE MaxStartCalls = 1 FreeBody = FALSE -INVARIANTS TypeOK RunningIsDriven WaitingIsDriven + Callers = {c1} + RecoverAfterLease = TRUE + NoLongPauses = FALSE +INVARIANTS TypeOK RunningIsDriven WaitingIsDriven NoOverlappingRuns PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone CHECK_DEADLOCK FALSE diff --git a/formal/tla/models/timer_arming_store_error.cfg b/formal/tla/models/timer_arming_store_error.cfg index 06b4622..a6d1c34 100644 --- a/formal/tla/models/timer_arming_store_error.cfg +++ b/formal/tla/models/timer_arming_store_error.cfg @@ -17,6 +17,9 @@ CONSTANTS SignalInMemory = FALSE MaxStartCalls = 1 FreeBody = FALSE -INVARIANTS TypeOK RunningIsDriven WaitingIsDriven + Callers = {c1} + RecoverAfterLease = TRUE + NoLongPauses = FALSE +INVARIANTS TypeOK RunningIsDriven WaitingIsDriven NoOverlappingRuns PROPERTIES TerminalIsStable TimerSchedulesKeepLeases EventuallyDone CHECK_DEADLOCK FALSE diff --git a/formal/tla/trace/README.md b/formal/tla/trace/README.md index 60dfec7..5295282 100644 --- a/formal/tla/trace/README.md +++ b/formal/tla/trace/README.md @@ -25,6 +25,9 @@ the spec has drifted and needs updating, or the code has a bug. - Only the stock SQLite and MySQL backends are wrapped. Tests that plug in their own store or scheduler steer interleavings by blocking inside calls, and the recorder's lock would change what they exercise. + - A test that only wraps the stock backends, to make a call fail, can install tracing beneath its + wrapper (`ChildWorkflowStartFailureTest` does). A call the wrapper fails never reaches the trace, + and the recorder skips the wrapper's frames when it works out which engine method made a call. 2. **Conversion.** `trace_to_tla.py` splits each file by workflow and turns each write into the model action it corresponds to (the `ACTIONS` table). - State is reduced to model values. Task ids become 0 for the workflow task and 1..n for timers. @@ -37,7 +40,8 @@ the spec has drifted and needs updating, or the code has a bug. 3. **Checking.** [`SkipperTrace.tla`](SkipperTrace.tla) replays one workflow's events against the model. - A step either consumes the next event, taking its action and landing exactly on its snapshot, - or is silent. A silent step is a model step that changes no durable state: a lease lapsing, + or is silent. A start that found the workflow already there is a step too, although it + writes nothing: what that start does next depends on what it found. A silent step is a model step that changes no durable state: a lease lapsing, time passing, a worker's local bookkeeping. The logged reads can't happen silently. - The workflow body is left free (`FreeBody`), except in how a real body can run. It replays its waits in order and may stop anywhere, and it suspends only inside `waitUntil`, on a timer it @@ -113,10 +117,13 @@ is scheduled while its timer is still `ACTIVE`, which the model says can't happe that is exactly what should prompt either a spec update, if the new order is intended, or a rethink. -Current results on the traced suites: 90 of 90 in-scope workflow runs accepted, 113 skipped, and -442 of 442 mutants rejected, in about 7 minutes with 8 parallel checks. Of the skips, 95 are tests -that write to the store directly; the rest use compensation (14), execution timeouts (2), -cancellation (1) or rewind (1). +Current results on the traced suites: 110 of 110 in-scope workflow runs accepted, 121 skipped, and +540 of 540 mutants rejected. Of the skips, 96 are tests that write to the store directly; the rest +use compensation (17), cancellation (5), execution timeouts (2) or rewind (1). + +The stranded-start recovery of airbnb/skipper#46 is covered the same way. With the engine changed +back to scheduling a stranded child like any other start, the recorded run of +`ChildWorkflowStartFailureTest`'s stranded child is rejected at that schedule. ## What it has found so far diff --git a/formal/tla/trace/SkipperTrace.tla b/formal/tla/trace/SkipperTrace.tla index 942f7b8..988a5b8 100644 --- a/formal/tla/trace/SkipperTrace.tla +++ b/formal/tla/trace/SkipperTrace.tla @@ -50,8 +50,9 @@ Act(e) == [] e.act = "Finish" -> \E p \in Workers : Finish(p) [] e.act = "Renew" -> \E p \in Workers : Renew(p) [] e.act = "Requeue" -> \E t \in TaskIds : Requeue(t) - [] e.act = "StartCreate" -> StartCreate - [] e.act = "StartSched" -> StartSched + [] e.act = "StartCreate" -> \E c \in Callers : StartCreate(c) + [] e.act = "StartSched" -> \E c \in Callers : StartSched(c) + [] e.act = "StartRecover" -> \E c \in Callers : StartRecover(c) [] e.act = "SignalPersist" -> SignalPersist [] e.act = "SignalSched" -> SignalSched [] e.act = "SignalStart" -> SignalStart @@ -86,15 +87,17 @@ TimeNeeded == Silent == Next /\ DurableSame /\ ~LoggedRead /\ TimeNeeded /\ UNCHANGED l (* A logged read is its model step. A write that changed nothing the model tracks (a *) -(* lost optimistic lock, an entity that already existed, a no-op honour of a lease) is *) -(* consumed without a model step; the model action it came from, if any, runs as a *) -(* silent step. A refused signal changes nothing at all. *) +(* lost optimistic lock, a no-op honour of a lease) is consumed without a model step; *) +(* the model action it came from, if any, runs as a silent step. A refused signal *) +(* changes nothing at all. A start that found the workflow already there is its model *) +(* step, like a read: what it does next depends on what it found, so a retried start *) +(* cannot pass for the first start's late schedule. *) Consume == /\ l <= Len(TraceLog) /\ LET e == TraceLog[l] IN /\ CASE e.act = "SignalRejected" -> UNCHANGED vars /\ Matches(e) [] e.read -> Act(e) /\ Matches(e) - [] e.noop -> UNCHANGED vars /\ Matches(e) + [] e.noop /\ e.act # "StartCreate" -> UNCHANGED vars /\ Matches(e) [] OTHER -> Act(e) /\ Matches(e) /\ ReportProgress => PrintT(<<"matched", l, e.seq>>) /\ l' = l + 1 diff --git a/formal/tla/trace/trace_to_tla.py b/formal/tla/trace/trace_to_tla.py index e6880d1..4b780ba 100755 --- a/formal/tla/trace/trace_to_tla.py +++ b/formal/tla/trace/trace_to_tla.py @@ -29,6 +29,7 @@ ("createWorkflow", "SkipperEngine.startWorkflow", "StartCreate"), ("schedule", "SkipperEngine.runSignal", "SignalSched"), ("schedule", "SkipperEngine.scheduleExecution", "StartSched"), + ("schedule", "SchedulerExecutionQueue.scheduleAfterLease", "StartRecover"), ("schedule", "TimerTaskHandler.handle", "TmSched"), ("schedule", "WorkflowExecutionTaskHandler.rearmOrphanedTimers", "WfRearm"), ("schedule", "WorkflowExecutionTaskHandler.persistExecutionResult", "WfTimers"), @@ -137,8 +138,10 @@ def normalise(workflow, event, header, action): task = event.get("task") row = state["tasks"].get(task) if task else None if row and event["ok"]: - in_memory = event["op"] == "schedule" and any("SchedulerExecutionQueue.schedule" in f for f in event["site"]) - if event["op"] in ("fetch", "renewLease") or in_memory: + # An in-memory schedule leases its backstop row, and scheduleAfterLease writes the same row. + leasing_schedule = event["op"] == "schedule" and any( + f in ("SchedulerExecutionQueue.schedule", "SchedulerExecutionQueue.scheduleAfterLease") for f in event["site"]) + if event["op"] in ("fetch", "renewLease") or leasing_schedule: workflow.lease_values.setdefault(task, set()).add(row["ra"]) def cls(task_id, r): @@ -203,7 +206,8 @@ def model_files(workflow, header): "MaxFaults": EARLY_EXPIRIES, "MaxRenewals": count("Renew"), "MaxRequeues": count("Requeue"), - "Faults": '{"early_expiry"}', + # A start that ran again found the workflow already there: an earlier call may have failed. + "Faults": '{"early_expiry", "caller_fail"}' if count("StartCreate") > 1 else '{"early_expiry"}', "NoopOnExisting": gates["CREATE_EXISTING_WORKFLOW_IS_NOOP"], "BumpOnHonoredLease": gates["BUMP_TASK_VERSION_ON_HONORED_LEASE"], "InMemoryStart": True, @@ -211,6 +215,9 @@ def model_files(workflow, header): "MaxStartCalls": max(1, count("StartCreate")), "UserFailures": True, "FreeBody": True, + "Callers": "{c1}", + "RecoverAfterLease": True, + "NoLongPauses": False, } fmt = lambda v: ("TRUE" if v else "FALSE") if isinstance(v, bool) else str(v) cfg = "SPECIFICATION TraceSpec\nCONSTANTS\n" diff --git a/src/main/java/com/airbnb/skipper/IWorkflowFactory.kt b/src/main/java/com/airbnb/skipper/IWorkflowFactory.kt index 165f537..0d3ad8c 100644 --- a/src/main/java/com/airbnb/skipper/IWorkflowFactory.kt +++ b/src/main/java/com/airbnb/skipper/IWorkflowFactory.kt @@ -283,6 +283,16 @@ class InvocationBuilder( 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 = 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. diff --git a/src/main/java/com/airbnb/skipper/WorkflowFactory.kt b/src/main/java/com/airbnb/skipper/WorkflowFactory.kt index 0189849..67310f2 100644 --- a/src/main/java/com/airbnb/skipper/WorkflowFactory.kt +++ b/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,52 @@ 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 an [InternalError], the same + * error Skipper's own stores and schedulers throw when a statement fails. The parent runs + * again after the configured unexpected-error delay and starts the child again under the + * same id. If the failed attempt already wrote the child but not its task, that start + * schedules it. This includes bugs such as a NullPointerException from a store. Those + * retries stop at the scheduler's max retries, leaving the parent in TRANSIENT_ERROR to be + * retried once the bug is fixed, rather than in ERROR. + * + * 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 InternalError( + "unable to start child workflow ${request.workflowId} of ${request.parentWorkflowId}: ${e.message}", + e, + ) + } + private fun executeSignalWithRetries(request: RunRequest): SkipperEngine.SendSignalResult { val retries = 3 var lastSignalError: Throwable? = null diff --git a/src/main/java/com/airbnb/skipper/internal/SkipperEngine.kt b/src/main/java/com/airbnb/skipper/internal/SkipperEngine.kt index d09e665..1f58102 100644 --- a/src/main/java/com/airbnb/skipper/internal/SkipperEngine.kt +++ b/src/main/java/com/airbnb/skipper/internal/SkipperEngine.kt @@ -99,6 +99,7 @@ open class SkipperEngine ) .build() var workflowInstance: WorkflowInstance + var stranded = false try { workflowInstance = workflowStore.createWorkflow(workflowCreationRequest) eventPublisher.publishEvent( @@ -136,13 +137,27 @@ open class SkipperEngine // If the workflow is waiting for a manual retry, then force the execution regardless. shouldByPassExecution = false } + if (shouldByPassExecution && isStranded(workflowInstance)) { + // An earlier start wrote the instance and then failed to schedule it, so nothing + // will ever run it. Schedule it now even if starting an existing workflow is a + // no-op, or retrying that start (e.g. a parent starting its child again) would + // return an instance that never runs. + shouldByPassExecution = false + stranded = true + } if (shouldByPassExecution) { tags[METRIC_OUTCOME] = "existingNoOp" return workflowInstance } } - val inProcessExecution = !request.isRunAsync - scheduleExecution(workflowInstance, inProcessExecution) + if (stranded) { + // The start that wrote the instance may only be slow rather than failed, and about to + // schedule it in memory. Scheduling after a lease lets either one run it, never both. + scheduler.scheduleAfterLease(executionRequest(workflowInstance).build()) + } else { + val inProcessExecution = !request.isRunAsync + scheduleExecution(workflowInstance, inProcessExecution) + } if (request.executionTimeout != null) { val timeoutTaskId = String.format("%s:timeout", workflowInstance.workflowId) val executionTimeout = now.instant().plus(request.executionTimeout) @@ -181,25 +196,35 @@ open class SkipperEngine } } + /** + * Whether [workflowInstance] is RUNNING with no task to execute it. The store writes a new + * instance as RUNNING, and a RUNNING instance keeps its task until an execution persists a + * new status, so a missing task means the start that wrote it failed to schedule it. + */ + private fun isStranded(workflowInstance: WorkflowInstance): Boolean = + workflowInstance.status == WorkflowInstance.Status.RUNNING && + scheduler.getTask(workflowInstance.workflowId).isEmpty + open fun scheduleExecution( workflowInstance: WorkflowInstance, inMemoryExecution: Boolean ) { scheduler.schedule( - ScheduleRequest.builder() - .id(workflowInstance.workflowId) - .dedupToken(workflowInstance.workflowId) - .payload(workflowInstance) - .type(Task.Type.WORKFLOW) - .honorActiveLeaseWhenOverwriting(true) - .bumpVersionWhenHonoringLease( - featureGate.isEnabled(FeatureGate.Keys.BUMP_TASK_VERSION_ON_HONORED_LEASE) - ) - .inMemoryExecutionEnabled(inMemoryExecution) - .build() + executionRequest(workflowInstance).inMemoryExecutionEnabled(inMemoryExecution).build() ) } + private fun executionRequest(workflowInstance: WorkflowInstance): ScheduleRequest.ScheduleRequestBuilder = + ScheduleRequest.builder() + .id(workflowInstance.workflowId) + .dedupToken(workflowInstance.workflowId) + .payload(workflowInstance) + .type(Task.Type.WORKFLOW) + .honorActiveLeaseWhenOverwriting(true) + .bumpVersionWhenHonoringLease( + featureGate.isEnabled(FeatureGate.Keys.BUMP_TASK_VERSION_ON_HONORED_LEASE) + ) + /** * Clones a workflow instance. * diff --git a/src/main/java/com/airbnb/skipper/internal/scheduler/SchedulerExecutionQueue.kt b/src/main/java/com/airbnb/skipper/internal/scheduler/SchedulerExecutionQueue.kt index e427e88..0718a18 100644 --- a/src/main/java/com/airbnb/skipper/internal/scheduler/SchedulerExecutionQueue.kt +++ b/src/main/java/com/airbnb/skipper/internal/scheduler/SchedulerExecutionQueue.kt @@ -4,6 +4,7 @@ import com.airbnb.skipper.SkipperAnnotationNames.SCHEDULER_LEASE_DURATION import com.airbnb.skipper.SkipperAnnotationNames.UTC_CLOCK import com.airbnb.skipper.internal.common.UniqueBlockingQueue import com.google.common.annotations.VisibleForTesting +import io.vavr.control.Option import java.time.Clock import java.time.Duration import java.time.Instant @@ -123,6 +124,28 @@ open class SchedulerExecutionQueue return newTask } + /** + * Schedule a task in the persistent scheduler only, due once a lease from now has passed. + * + * This is the same row the in-memory path writes as its fallback, so it reads as an active lease + * to any other schedule of the same task. Paired with a concurrent in-memory schedule of that + * task, whichever writes the row first, the task runs once: + * 1. If the in-memory schedule writes first, this call honours its lease and enqueues nothing. + * 2. If this call writes first, the in-memory schedule honours this row's lease and runs the task + * in memory, as it would have with its own row. + * 3. If nothing else schedules the task, the persistent scheduler delivers it after the lease. + */ + open fun scheduleAfterLease(request: ScheduleRequest): Task = + persistentScheduler.schedule( + request.toBuilder() + .runAfter(now.instant().plus(leaseDuration)) + .payload(null) + .build(), + ) + + /** The persisted task with [taskId], if any. Every scheduled task is persisted, in-memory ones too. */ + open fun getTask(taskId: String): Option> = persistentScheduler.getTask(taskId) + open fun add(task: Task<*>): Boolean = getQueue(task.type).add(task, task.id) open fun take(type: Task.Type): Task { diff --git a/src/test/java/com/airbnb/skipper/WorkflowFactoryTest.kt b/src/test/java/com/airbnb/skipper/WorkflowFactoryTest.kt index 51e9e51..f487377 100644 --- a/src/test/java/com/airbnb/skipper/WorkflowFactoryTest.kt +++ b/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 { child.asyncWorkflowMethod("input") } + } + + @Test + fun testChildStartInfrastructureFailureRetriesTheParent() { + val thrown = startChildThrowing(java.util.concurrent.TimeoutException("store timed out")) + + // The same error the stores and schedulers throw, which the task handler already retries. + assertThat(thrown).isInstanceOf(com.airbnb.skipper.internal.InternalError::class.java) + assertThat(thrown.message).contains(workflowId).contains("parent-id").contains("store timed out") + assertThat(thrown.cause).isInstanceOf(java.util.concurrent.TimeoutException::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 { 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 diff --git a/src/test/java/com/airbnb/skipper/integtest/ChildWorkflowStartFailureTest.kt b/src/test/java/com/airbnb/skipper/integtest/ChildWorkflowStartFailureTest.kt new file mode 100644 index 0000000..80b1fae --- /dev/null +++ b/src/test/java/com/airbnb/skipper/integtest/ChildWorkflowStartFailureTest.kt @@ -0,0 +1,255 @@ +package com.airbnb.skipper.integtest + +import com.airbnb.skipper.ComponentFactory +import com.airbnb.skipper.FeatureGate +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.testutils.trace.Tracing +import com.airbnb.skipper.util.ExtraRequestData +import java.time.Duration +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 +import org.mockito.Mockito.`when` + +/** + * 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 schedule(request: ScheduleRequest): Task { + 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, + createExistingWorkflowIsNoop: Boolean = false, + ) { + val deps = TestRuntime() + `when`(deps.featureGate.isEnabled(FeatureGate.Keys.CREATE_EXISTING_WORKFLOW_IS_NOOP)).thenReturn(createExistingWorkflowIsNoop) + deps.config.taskUnexpectedErrorRetryDelay = Duration.ofMillis(200) + deps.config.schedulerTaskLeaseDuration = Duration.ofSeconds(1) + // Trace the stock backends beneath the faults (when tracing is on), so trace validation checks + // what reached the store, the recovery of a stranded child included. + deps.config.workflowStore = SqliteWorkflowStore.Factory() + deps.config.scheduler = SqliteScheduler.Factory() + Tracing.installIfRequested(deps.config) + val store = deps.config.workflowStore + val stockScheduler = deps.config.scheduler + deps.config.workflowStore = ComponentFactory { FaultyStore(store.create(it), storeFault) } + deps.config.scheduler = ComponentFactory { FaultyScheduler(stockScheduler.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 = 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 schedulerTimeoutWhileSchedulingAChildRunsTheChildWhenCreatingAnExistingWorkflowIsNoop() { + val parentId = "fan-out-${System.nanoTime()}" + // The gate's default. The retried start finds the child RUNNING with no task, so it schedules the + // child, due after the lease, instead of returning it as it is. + val fault = Fault(childId(parentId, 2)) { TimeoutException("injected timeout inserting the child's task") } + start(schedulerFault = fault, createExistingWorkflowIsNoop = true) + 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" + } +} diff --git a/src/test/java/com/airbnb/skipper/internal/SkipperEngineTest.kt b/src/test/java/com/airbnb/skipper/internal/SkipperEngineTest.kt index 297fefd..2bf67aa 100644 --- a/src/test/java/com/airbnb/skipper/internal/SkipperEngineTest.kt +++ b/src/test/java/com/airbnb/skipper/internal/SkipperEngineTest.kt @@ -248,6 +248,92 @@ class SkipperEngineTest { verify(mockScheduler, times(0)).schedule(any()) } + /** + * An earlier start wrote the instance, then failed to schedule it. Starting it again schedules it + * even though starting an existing workflow is a no-op, or it would never run. It schedules it after + * a lease, so an earlier start that is only slow can still run it in memory without a second copy. + */ + @Test + fun testStartWorkflowWhenWorkflowExistsWithoutATaskSchedulesItDespiteCreationNoop() { + whenever(mockFeatureGate.isEnabled(FeatureGate.Keys.CREATE_EXISTING_WORKFLOW_IS_NOOP)) + .thenReturn(true) + whenever(mockWorkflowStore.createWorkflow(any())).thenThrow(EntityAlreadyExists()) + whenever(mockWorkflowStore.getWorkflow(any())).thenReturn(Option.of(runningInstance())) + whenever(mockScheduler.getTask("test-workflow-id")).thenReturn(Option.none()) + whenever(mockScheduler.scheduleAfterLease(any>())) + .thenReturn(getTestTask(getWorkflowInstance())) + + skipperEngine.startWorkflow( + RunRequest.builder() + .workflowId("test-workflow-id") + .workflowClass(Workflow::class.java) + .input("test") + .workflowMethod("test-method") + .requestContext(REQUEST_CONTEXT) + .extraRequestData(ExtraRequestData()) + .runAsync(true) + .build(), + ) + + verify(mockScheduler).scheduleAfterLease( + argThat> { + id == "test-workflow-id" && type == Task.Type.WORKFLOW && isHonorActiveLeaseWhenOverwriting + }, + ) + verify(mockScheduler, times(0)).schedule(any()) + } + + @Test + fun testStartWorkflowWhenRunningWorkflowExistsWithATaskAndCreationIsNoopDoesNotScheduleIt() { + whenever(mockFeatureGate.isEnabled(FeatureGate.Keys.CREATE_EXISTING_WORKFLOW_IS_NOOP)) + .thenReturn(true) + whenever(mockWorkflowStore.createWorkflow(any())).thenThrow(EntityAlreadyExists()) + whenever(mockWorkflowStore.getWorkflow(any())).thenReturn(Option.of(runningInstance())) + whenever(mockScheduler.getTask("test-workflow-id")) + .thenReturn(Option.of(getTestTask(runningInstance()))) + + skipperEngine.startWorkflow( + RunRequest.builder() + .workflowId("test-workflow-id") + .workflowClass(Workflow::class.java) + .input("test") + .workflowMethod("test-method") + .requestContext(REQUEST_CONTEXT) + .extraRequestData(ExtraRequestData()) + .build(), + ) + + verify(mockScheduler, times(0)).schedule(any()) + } + + /** A WAITING instance has no task until a signal or timer wakes it, so its task is not looked up. */ + @Test + fun testStartWorkflowWhenWaitingWorkflowExistsAndCreationIsNoopDoesNotLookUpItsTask() { + whenever(mockFeatureGate.isEnabled(FeatureGate.Keys.CREATE_EXISTING_WORKFLOW_IS_NOOP)) + .thenReturn(true) + whenever(mockWorkflowStore.createWorkflow(any())).thenThrow(EntityAlreadyExists()) + whenever(mockWorkflowStore.getWorkflow(any())) + .thenReturn( + Option.of(TestUtils.getWorkflowInstance().toBuilder().status(WorkflowInstance.Status.WAITING).build()), + ) + + skipperEngine.startWorkflow( + RunRequest.builder() + .workflowId("test-workflow-id") + .workflowClass(Workflow::class.java) + .input("test") + .workflowMethod("test-method") + .requestContext(REQUEST_CONTEXT) + .extraRequestData(ExtraRequestData()) + .build(), + ) + + verify(mockScheduler, times(0)).getTask(any()) + verify(mockScheduler, times(0)).schedule(any()) + } + + private fun runningInstance(): WorkflowInstance = TestUtils.getWorkflowInstance().toBuilder().status(WorkflowInstance.Status.RUNNING).build() + @Test fun testStartWorkflowWhenRetriesAreExhaustedIgnoresRequestCreationNoop() { val instance = diff --git a/src/test/java/com/airbnb/skipper/internal/scheduler/QueueBasedSchedulerTest.kt b/src/test/java/com/airbnb/skipper/internal/scheduler/QueueBasedSchedulerTest.kt index c38d1cc..6d85270 100644 --- a/src/test/java/com/airbnb/skipper/internal/scheduler/QueueBasedSchedulerTest.kt +++ b/src/test/java/com/airbnb/skipper/internal/scheduler/QueueBasedSchedulerTest.kt @@ -76,4 +76,54 @@ class QueueBasedSchedulerTest { // the execution queue. assertEquals(1, persistentScheduler.realSize()) } + + @Test + fun testScheduleAfterLeaseOnlyWritesTheRowDueAfterTheLease() { + whenever(mockClock.instant()).thenReturn(Instant.EPOCH) + val task = scheduler.scheduleAfterLease(workflowTaskRequest()) + assertEquals(0, scheduler.getQueue(Task.Type.WORKFLOW).size()) + assertEquals(Instant.EPOCH.plus(leaseDuration), task.runAfter) + assertEquals(0, persistentScheduler.fetch(10).size(), "the row is not due before the lease passes") + whenever(mockClock.instant()).thenReturn(Instant.EPOCH.plus(leaseDuration).plus(Duration.ofSeconds(1))) + assertEquals(listOf(task.id), persistentScheduler.fetch(10).map { it.id }.toJavaList()) + } + + /** A start schedules the task in memory on one node while a retried start schedules it after the lease on another. */ + @Test + fun testInMemoryScheduleFollowedByScheduleAfterLeaseRunsTheTaskOnce() { + whenever(mockClock.instant()).thenReturn(Instant.EPOCH) + val otherNode = SchedulerExecutionQueue(persistentScheduler, mockClock, leaseDuration) + scheduler.schedule(workflowTaskRequest()) + otherNode.scheduleAfterLease(workflowTaskRequest()) + assertRunsOnlyInMemoryOn(scheduler, otherNode) + } + + @Test + fun testScheduleAfterLeaseFollowedByInMemoryScheduleRunsTheTaskOnce() { + whenever(mockClock.instant()).thenReturn(Instant.EPOCH) + val otherNode = SchedulerExecutionQueue(persistentScheduler, mockClock, leaseDuration) + otherNode.scheduleAfterLease(workflowTaskRequest()) + scheduler.schedule(workflowTaskRequest()) + assertRunsOnlyInMemoryOn(scheduler, otherNode) + } + + private fun assertRunsOnlyInMemoryOn( + node: SchedulerExecutionQueue, + otherNode: SchedulerExecutionQueue, + ) { + assertEquals(1, node.getQueue(Task.Type.WORKFLOW).size()) + assertEquals(0, otherNode.getQueue(Task.Type.WORKFLOW).size()) + assertEquals(0, persistentScheduler.fetch(10).size(), "the persistent row must not be due while the task runs in memory") + } + + private fun workflowTaskRequest(): ScheduleRequest { + val workflowInstance = TestUtils.getWorkflowInstance() + return ScheduleRequest.builder() + .id(workflowInstance.workflowId) + .dedupToken(workflowInstance.workflowId) + .type(Task.Type.WORKFLOW) + .payload(workflowInstance) + .honorActiveLeaseWhenOverwriting(true) + .build() + } } diff --git a/testutils/src/main/java/com/airbnb/skipper/testutils/trace/TraceRecorder.kt b/testutils/src/main/java/com/airbnb/skipper/testutils/trace/TraceRecorder.kt index b3da7af..a5106df 100644 --- a/testutils/src/main/java/com/airbnb/skipper/testutils/trace/TraceRecorder.kt +++ b/testutils/src/main/java/com/airbnb/skipper/testutils/trace/TraceRecorder.kt @@ -7,6 +7,7 @@ import com.airbnb.skipper.internal.storage.WorkflowStore import java.io.BufferedWriter import java.nio.file.Files import java.nio.file.Path +import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.locks.ReentrantLock import kotlin.concurrent.withLock @@ -155,16 +156,30 @@ class TraceRecorder internal constructor( */ internal fun calledFrom(methods: List): Boolean = callSite().firstOrNull() in methods - /** The engine methods that made the call, innermost first: what the checker maps to a model action. */ + /** + * The engine methods that made the call, innermost first: what the checker maps to a model action. + * A store or scheduler wrapped around the traced one, such as a test's fault injector, passes + * calls through to it, so its frames are skipped and the engine method behind it is the caller. + */ private fun callSite(): List = Thread.currentThread().stackTrace .filter { f -> f.className.startsWith("com.airbnb.skipper.") && EXCLUDED_PREFIXES.none { f.className.startsWith(it) } } + .dropWhile { f -> isBackendWrapper(f.className) } .map { f -> f.className.substringAfterLast('.') + "." + f.methodName } .take(8) + private val backendWrappers = ConcurrentHashMap() + + private fun isBackendWrapper(className: String): Boolean = + backendWrappers.computeIfAbsent(className) { + val cls = runCatching { Class.forName(it, false, TraceRecorder::class.java.classLoader) }.getOrNull() + cls != null && (WorkflowStore::class.java.isAssignableFrom(cls) || Scheduler::class.java.isAssignableFrom(cls)) + } + /** The frames of the test that started the runtime, for finding the test a trace came from. */ private fun origin(): List = Thread.currentThread().stackTrace + .filterNot { f -> isBackendWrapper(f.className) } .map { f -> f.className.substringAfterLast('.') + "." + f.methodName } .filter { it.contains("Test") || it.contains("Example") } .take(4) diff --git a/testutils/src/main/java/com/airbnb/skipper/testutils/trace/Tracing.kt b/testutils/src/main/java/com/airbnb/skipper/testutils/trace/Tracing.kt index af63f8e..79269e6 100644 --- a/testutils/src/main/java/com/airbnb/skipper/testutils/trace/Tracing.kt +++ b/testutils/src/main/java/com/airbnb/skipper/testutils/trace/Tracing.kt @@ -19,7 +19,9 @@ import java.util.UUID * * Only the stock JDBC backends are traced. A test that plugs in its own store or scheduler is * usually steering an interleaving by blocking inside a call, and the recorder's lock, held across - * each call, would stall the rest of the engine behind it and change what the test exercises. + * each call, would stall the rest of the engine behind it and change what the test exercises. A + * test that only wraps them, for example to make a call fail, can call this on the stock factories + * itself and wrap the traced ones: a call its wrapper fails never reaches the trace. */ object Tracing { const val DIR_PROPERTY = "skipper.trace.dir" diff --git a/web/src/content/docs/invoking-workflows.md b/web/src/content/docs/invoking-workflows.md index d7d1067..29488d0 100644 --- a/web/src/content/docs/invoking-workflows.md +++ b/web/src/content/docs/invoking-workflows.md @@ -256,3 +256,49 @@ lets Skipper hibernate the instance and notify you on completion, rather than ho that will time out anyway. For the Kotlin-specific picture — why `launch { }` is a poor substitute, and how threads are held on each side of a call — see [Kotlin Coroutines](/docs/kotlin-coroutines/). + +## Starting child workflows + +A workflow that starts other workflows from its workflow method starts them on its own execution +thread, so whatever a start throws escapes the parent's workflow method. There, an exception Skipper +doesn't recognize is taken for a bug in your workflow code and ends the parent in `ERROR` for good. +But a start writes the child to the workflow store and the scheduler, and a timeout on one of those +writes is no bug. In a fan-out that starts thousands of children, one such blip would end the whole +parent. + +Mark the invocation with `.asChildOf(this)`: + +```kotlin +@WorkflowMethod +open fun issueAll(requests: List) { + for (request in requests) { + workflowFactory.builder(IssueWorkflow.idFor(request)) + .callbackHandler(IssueCallbackHandler::class.java) + .asChildOf(this) + .build() + .issue(request) + } +} +``` + +`asChildOf` records the parent's id on the child (the same as `.parentWorkflowId(id)`), and an +invocation with a parent id sorts out what its start throws: + +1. **An error Skipper has already classified** — `NonRetryableError`, `RetryableError` and the + like — is rethrown unchanged. +2. **A request Skipper rejected as invalid** (`IllegalArgumentException`, `ValidationError`) is + rethrown unchanged too, and still fails the parent: running it again would only repeat it. +3. **Anything else**, such as a store or scheduler timeout, is rethrown as Skipper's internal error, + the same one its own stores and schedulers throw when a statement fails. The parent runs again + after `taskUnexpectedErrorRetryDelay` (30 seconds by default) and starts the child again under + the same id. Children it already started aren't duplicated, because the id is an idempotency + key. A child whose instance was written but whose task wasn't gets scheduled on that retry, even + with `createExistingWorkflowIsNoop` on. It runs once `schedulerTaskLeaseDuration` has passed + (8 minutes by default), so a first start that was only slow can't end up running the child + alongside it. This applies to every exception Skipper doesn't recognize, programming errors + such as a `NullPointerException` from a store included. The retries stop after + `schedulerTaskMaxRetries` (100 by default). The parent then stays in `TRANSIENT_ERROR`, which + the admin UI's Retry accepts once the cause is fixed. + +Only the start is covered: waiting for a child's result, and the child's own failures, reach the +parent as before. diff --git a/web/src/content/docs/reference/core-types.md b/web/src/content/docs/reference/core-types.md index c339612..9eb8a0a 100644 --- a/web/src/content/docs/reference/core-types.md +++ b/web/src/content/docs/reference/core-types.md @@ -52,6 +52,7 @@ Starting and addressing workflows. See [Invoking Workflows](/docs/invoking-workf | `.requestContext(payload)` | Opaque, persisted per-request payload surfaced to middleware. | | `.parentWorkflowId(id)` | Link this instance to a parent workflow. | | `.build(): T` | Build the workflow handle. | +| `.asChildOf(parent)` | Start this workflow as a child of `parent`, from `parent`'s workflow method: records the parent's id, and a store or scheduler failure while starting it retries the parent instead of failing it. See [Starting child workflows](/docs/invoking-workflows/#starting-child-workflows). | ## Lifecycle callbacks — `WorkflowCallbackHandler`