Today the `ReservedClusterStateService` decides whether or not to submit an update-error task depending on the content of the last-applied state as exposed by `ClusterService#state()`. This may happen concurrently with another state being published, so does not give a fully consistent view of the latest state. Instead it should check the state within the cluster state task executor, triggering a no-op update as appropriate.
server/src/main/java/org/elasticsearch/reservedstate/service/ReservedClusterStateService.java
for (var tuple : tuples) {
                stateMap.put(tuple.v1(), tuple.v2());
            }
            return new ReservedStateChunk(stateMap, (ReservedStateVersion) a[1]);
        }
    );
    /**
     * Controller class for saving and reserving {@link ClusterState}.
     * @param clusterService for fetching and saving the modified state
     * @param handlerList a list of reserved state handlers, which we use to transform the state
     */
    public ReservedClusterStateService(ClusterService clusterService, List<ReservedClusterStateHandler<?>> handlerList) {
        this.clusterService = clusterService;
        this.updateStateTaskExecutor = new ReservedStateUpdateTaskExecutor(clusterService.getRerouteService());
        this.errorStateTaskExecutor = new ReservedStateErrorTaskExecutor();
        this.handlers = handlerList.stream().collect(Collectors.toMap(ReservedClusterStateHandler::name, Function.identity()));
        stateChunkParser.declareNamedObjects(ConstructingObjectParser.constructorArg(), (p, c, name) -> {
            if (handlers.containsKey(name) == false) {
                throw new IllegalStateException("Missing handler definition for content key [" + name + "]");
            }
            p.nextToken();
            return new Tuple<>(name, handlers.get(name).fromXContent(p));
        }, STATE_FIELD);
        stateChunkParser.declareObject(ConstructingObjectParser.constructorArg(), (p, c) -> ReservedStateVersion.parse(p), METADATA_FIELD);
    }
    /**
     * Saves and reserves a chunk of the cluster state under a given 'namespace' from {@link XContentParser}
     *
     * @param namespace the namespace under which we'll store the reserved keys in the cluster state metadata
     * @param parser the XContentParser to process