private List<Node> prepareWithLocks(ApplicationId application, ClusterSpec cluster, NodeSpec requested, NodeIndices indices) {
        Runnable waiter = null;
        List<Node> acceptedNodes;
        try (Mutex lock = nodeRepository.applications().lock(application);
             ApplicationMutex parentLockOrNull = parentLockOrNull(requested.type());
             Mutex allocationLock = nodeRepository.nodes().lockUnallocated()) {
            LockedNodeList allNodes = nodeRepository.nodes().list(allocationLock);
            NodeAllocation allocation = prepareAllocation(application, cluster, requested, indices::next, allNodes);
            NodeType hostType = allocation.nodeType().hostType();
            if (canProvisionDynamically(hostType) && allocation.hostDeficit().isPresent()) {
                HostSharing sharing = hostSharing(cluster, hostType);
                Version osVersion = nodeRepository.osVersions().targetFor(hostType).orElse(Version.emptyVersion);
                NodeAllocation.HostDeficit deficit = allocation.hostDeficit().get();
                Set<Node> hosts = new LinkedHashSet<>();
                Consumer<List<ProvisionedHost>> whenProvisioned = provisionedHosts -> {
                    List<Node> newHosts = provisionedHosts.stream()
                                                          .map(host -> host.generateHost(requested.hostTTL()))
                                                          .filter(hosts::add)
                                                          .toList();
                    nodeRepository.nodes().addNodes(newHosts, Agent.application);

                    // Offer the nodes on the newly provisioned hosts, this should be enough to cover the deficit
                    List<NodeCandidate> candidates = provisionedHosts.stream()
                            .map(host -> NodeCandidate.createNewExclusiveChild(host.generateNode(),
                                                                               host.generateHost(requested.hostTTL())))
                            .toList();
                    allocation.offer(candidates);
                };
                try {
                    if (throttler.throttle(allNodes, Agent.system)) {
                        throw new NodeAllocationException("Host provisioning is being throttled", true);
                    }

                    HostProvisionRequest request = new HostProvisionRequest(allocation.provisionIndices(deficit.count()),
                                                                            hostType,
                                                                            deficit.resources(),
                                                                            application,
                                                                            osVersion,
                                                                            Vtag.currentVersion,
                                                                            sharing,
                                                                            Optional.of(cluster.type()),
                                                                            Optional.of(cluster.id()),
                                                                            requested.cloudAccount(),
                                                                            deficit.dueToFlavorUpgrade(),
                                                                            false);
                    Predicate<NodeResources> realHostResourcesWithinLimits =
                            resources -> nodeRepository.nodeResourceLimits().isWithinRealLimits(resources, cluster);
                    waiter = hostProvisioner.get().provisionHosts(request, realHostResourcesWithinLimits, whenProvisioned);
                } catch (NodeAllocationException e) {
                    // Mark the nodes that were written to ZK in the consumer for deprovisioning. While these hosts do
                    // not exist, we cannot remove them from ZK here because other nodes may already have been
                    // allocated on them, so let HostDeprovisioner deal with it
                    hosts.forEach(host -> nodeRepository.nodes().parkRecursively(host.hostname(), Agent.system, true,
                            "Failed to provision: " + Exceptions.toMessageString(e)));
                    throw e;
                }
            } else if (allocation.hostDeficit().isPresent() && requested.canFail() &&
                       allocation.hasRetiredJustNow() && requested instanceof NodeSpec.CountNodeSpec cns) {
                // Non-dynamically provisioned zone with a deficit because we just now retired some nodes.
                // Try again, but without retiring
                indices.resetProbe();
                List<Node> accepted = prepareWithLocks(application, cluster, cns.withoutRetiring(), indices);
                log.warning("Prepared " + application + " " + cluster.id() + " without retirement due to lack of capacity");
                return accepted;
            }

            if (! allocation.fulfilled() && requested.canFail())
                throw new NodeAllocationException("Could not satisfy " + requested + " in " + application + " " + cluster +
                                                  allocation.allocationFailureDetails(), true);

            // Carry out and return allocation
            if (parentLockOrNull != null) {
                List<Node> exclusiveParents = allocation.parentsRequiredToBeExclusive();
                nodeRepository.nodes().setExclusiveToApplicationId(exclusiveParents, parentLockOrNull, application);
            }
            acceptedNodes = allocation.finalNodes();
            nodeRepository.nodes().reserve(allocation.reservableNodes());
            nodeRepository.nodes().addReservedNodes(new LockedNodeList(allocation.newNodes(), allocationLock));

            if (requested.rejectNonActiveParent()) { // TODO: Move into offer() - currently this must be done *after* reserving
                NodeList activeHosts = allNodes.state(Node.State.active).parents().nodeType(requested.type().hostType());
                acceptedNodes = acceptedNodes.stream()
                                             .filter(node -> node.parentHostname().isEmpty() || activeHosts.parentOf(node).isPresent())
                                             .toList();
            }
        }

        if (waiter != null) waiter.run();
        return acceptedNodes;
    }
Ensure sufficient nodes are reserved or active for the given application, group and cluster

@param application        the application we are allocating to
@param cluster            the cluster and group we are allocating to
@param requested          a specification of the requested nodes
@return the list of nodes this cluster group will have allocated if activated
node-repository/src/main/java/com/yahoo/vespa/hosted/provision/provisioning/Preparer.java