class InternalStepRunner implements StepRunner {

    private Optional<RunStatus> deploy(ApplicationId id, JobType type, Supplier<ActivateResult> deployment,
                                       Instant startTime, DualLogger logger) {
        try {
            PrepareResponse prepareResponse = deployment.get().prepareResponse();
            if (prepareResponse.log != null)
                logger.logAll(prepareResponse.log.stream()
                                                 .map(entry -> new LogEntry(0, // Sequenced by BufferedLogStore.
                                                                            Instant.ofEpochMilli(entry.time),
                                                                            LogEntry.typeOf(LogLevel.parse(entry.level)),
                                                                            entry.message))
                                                 .collect(toList()));
            if ( ! prepareResponse.configChangeActions.refeedActions.stream().allMatch(action -> action.allowed)) {
                List<String> messages = new ArrayList<>();
                messages.add("Deploy failed due to non-compatible changes that require re-feed.");
                messages.add("Your options are:");
                messages.add("1. Revert the incompatible changes.");
                messages.add("2. If you think it is safe in your case, you can override this validation, see");
                messages.add("   http://docs.vespa.ai/documentation/reference/validation-overrides.html");
                messages.add("3. Deploy as a new application under a different name.");
                messages.add("Illegal actions:");
                prepareResponse.configChangeActions.refeedActions.stream()
                                                                 .filter(action -> ! action.allowed)
                                                                 .flatMap(action -> action.messages.stream())
                                                                 .forEach(messages::add);
                logger.log(messages);
                return Optional.of(deploymentFailed);
            }

            if (prepareResponse.configChangeActions.restartActions.isEmpty())
                logger.log("No services requiring restart.");
            else
                prepareResponse.configChangeActions.restartActions.stream()
                                                                  .flatMap(action -> action.services.stream())
                                                                  .map(service -> service.hostName)
                                                                  .sorted().distinct()
                                                                  .map(HostName::from)
                                                                  .forEach(hostname -> {
                                                                      controller.applications().restart(new DeploymentId(id, type.zone(controller.system())), new RestartFilter().withHostName(Optional.of(hostname)));
                                                                      logger.log("Schedule service restart on host " + hostname.value() + ".");
                                                                  });
            logger.log("Deployment successful.");
            if (prepareResponse.message != null)
                logger.log(prepareResponse.message);
            return Optional.of(running);
        }
        catch (ConfigServerException e) {
            // Retry certain failures for up to one hour.
            Optional<RunStatus> result = startTime.isBefore(controller.clock().instant().minus(Duration.ofHours(1)))
                                         ? Optional.of(deploymentFailed) : Optional.empty();
            switch (e.getErrorCode()) {
                case CERTIFICATE_NOT_READY:
                    if (startTime.plus(timeouts.endpointCertificate()).isBefore(controller.clock().instant())) {
                        logger.log("Deployment failed to find provisioned endpoint certificate after " + timeouts.endpointCertificate());
                        return Optional.of(RunStatus.endpointCertificateTimeout);
                    }
                    return result;
                case ACTIVATION_CONFLICT:
                case APPLICATION_LOCK_FAILURE:
                    logger.log("Deployment failed with possibly transient error " + e.getErrorCode() +
                            ", will retry: " + e.getMessage());
                    return result;
                case LOAD_BALANCER_NOT_READY:
                case PARENT_HOST_NOT_READY:
                    logger.log(e.getServerMessage());
                    return result;
                case OUT_OF_CAPACITY:
                    logger.log(e.getServerMessage());
                    return controller.system().isCd() && startTime.plus(timeouts.capacity()).isAfter(controller.clock().instant())
                           ? Optional.empty()
                           : Optional.of(outOfCapacity);
                case INVALID_APPLICATION_PACKAGE:
                case BAD_REQUEST:
                    logger.log(e.getMessage());
                    return Optional.of(deploymentFailed);
            }

            throw e;
        }
        catch (EndpointCertificateException e) {
            switch (e.type()) {
                case CERT_NOT_AVAILABLE:
                    // Same as CERTIFICATE_NOT_READY above, only from the controller
                    if (startTime.plus(timeouts.endpointCertificate()).isBefore(controller.clock().instant())) {
                        logger.log("Deployment failed to find provisioned endpoint certificate after " + timeouts.endpointCertificate());
                        return Optional.of(RunStatus.endpointCertificateTimeout);
                    }
                    return Optional.empty();
                default:
                    throw e; // Should be surfaced / fail deployment
            }
        }
    }

    public  InternalStepRunner(Controller controller);

}