1 * workSource.finishedExecuting(node, null)

        then:
        1 * cancellationHandler.isCancellationRequested() >> false
        1 * workSource.executionState() >> WorkSource.State.NoMoreWorkToStart

        then:
        1 * workerLease.tryLock() >> true
        3 * workSource.allExecutionComplete() >> true
        1 * workSource.collectFailures([])
        0 * workSource._
    }

         * Note that this does not mean that all work has completed.
         */
        NoMoreWorkToStart
    }

    abstract class Selection<T> {
        private static final Selection<Object> NO_WORK_READY_TO_START = new Selection<Object>() {
            @Override
            public boolean isNoWorkReadyToStart() {

            while (iterator.hasNext()) {
                PlanDetails details = iterator.next();
                WorkSource.State state = details.source.executionState();
                if (state == WorkSource.State.NoMoreWorkToStart) {
                    if (details.source.allExecutionComplete()) {
                        iterator.remove();
                    }

        }
    }

    @Override
    public State executionState() {
        lockCoordinator.assertHasStateLock();
        if (waitingToStartNodes.isEmpty()) {
            return State.NoMoreWorkToStart;
        } else if (!readyNodes.isEmpty() && maybeNodesSelectable) {
            return State.MaybeWorkReadyToStart;
        } else {
            return State.NoWorkReadyToStart;
        }
    }

            } else {
                assert finalizedPlan.executionState() == WorkSource.State.NoMoreWorkToStart
            }
            assert finalizedPlan.selectNext().noMoreWorkToStart
            assert finalizedPlan.executionState() == WorkSource.State.NoMoreWorkToStart
            assert !finalizedPlan.allExecutionComplete()
        }
    }

            while (finalizedPlan.executionState() != WorkSource.State.NoMoreWorkToStart) {
                assert finalizedPlan.executionState() == WorkSource.State.MaybeWorkReadyToStart // There should always be a node ready to start when executing sequentially
                def selection = finalizedPlan.selectNext()
                if (selection.noMoreWorkToStart) {
                    break
                }