Seen in soak testing:

```
org.elasticsearch.snapshots.SnapshotStressTestsIT > testRandomActivities {seed=[AB70F7282B153829:AB3327F808885EA1]} FAILED
    java.lang.AssertionError: add restore for [[index-0][2]] that already has an existing restore
        at __randomizedtesting.SeedInfo.seed([AB70F7282B153829]:0)
        at org.elasticsearch.repositories.blobstore.BlobStoreRepository.restoreShard(BlobStoreRepository.java:3399)
        at org.elasticsearch.index.shard.StoreRecovery.lambda$restore$14(StoreRecovery.java:569)
        at org.elasticsearch.action.ActionListenerImplementations$ResponseWrappingActionListener.onResponse(ActionListenerImplementations.java:236)
        at org.elasticsearch.action.support.SubscribableListener$SuccessResult.complete(SubscribableListener.java:310)
        at org.elasticsearch.action.support.SubscribableListener.tryComplete(SubscribableListener.java:230)
        at org.elasticsearch.action.support.SubscribableListener.addListener(SubscribableListener.java:133)
        at org.elasticsearch.action.support.SubscribableListener.addListener(SubscribableListener.java:108)
        at org.elasticsearch.index.shard.StoreRecovery.lambda$restore$15(StoreRecovery.java:567)
        at org.elasticsearch.action.ActionListenerImplementations$DelegatingFailureActionListener.onResponse(ActionListenerImplementations.java:212)
        at org.elasticsearch.action.ActionListenerImplementations$DelegatingResponseActionListener.onResponse(ActionListenerImplementations.java:182)
        at org.elasticsearch.index.CompositeIndexEventListener.callListeners(CompositeIndexEventListener.java:278)
        at org.elasticsearch.index.CompositeIndexEventListener.iterateBeforeIndexShardRecovery(CompositeIndexEventListener.java:287)
        at org.elasticsearch.index.CompositeIndexEventListener.beforeIndexShardRecovery(CompositeIndexEventListener.java:314)
        at org.elasticsearch.index.shard.IndexShard.preRecovery(IndexShard.java:1683)
        at org.elasticsearch.index.shard.StoreRecovery.restore(StoreRecovery.java:519)
        at org.elasticsearch.index.shard.StoreRecovery.recoverFromRepository(StoreRecovery.java:311)
        at org.elasticsearch.index.shard.IndexShard.restoreFromRepository(IndexShard.java:2349)
        at org.elasticsearch.index.shard.IndexShard.lambda$startRecovery$21(IndexShard.java:3132)
        at org.elasticsearch.action.ActionRunnable$3.doRun(ActionRunnable.java:73)
        at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:983)
        at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:26)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642)
        at java.base/java.lang.Thread.run(Thread.java:1583)
```

~This looks to happen if we attempt to do two concurrent restores of an index, one where the index is closed and the other where it is deleted. This test shouldn't do that since it is focussing on the happy path, and one of those restores should fail, but nor should it be possible to trip an assertion if we try it.~

Edit: hmm no that doesn't make sense, there's some confusing logging in this area. Looking again I see that we restore the same index twice in quick succession, and there are some recovery cancellations in the logs. Cancelling a snapshot-based recovery won't stop the restore (and clean up the `ongoingRestores` marker) straight away. But I'm not sure how that could happen in `SnapshotStressTestsIT` yet, it looks like we always hold all the permits for an index until the restore succeeds. 

---

I think the production consequences of this are mild, we mostly track the ongoing restores so that tests can wait for the repository to become idle.
server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java
%s""", ReferenceDocs.CONCURRENT_REPOSITORY_WRITERS, previousWriterMessage(previousWriterInfo)), cause);
    }
    private static String previousWriterMessage(@Nullable Tuple<Long, String> previousWriterInfo) {
        return previousWriterInfo == null
            ? ""
            : " The last cluster to write to this repository was ["
                + previousWriterInfo.v2()
                + "] at generation ["
                + previousWriterInfo.v1()
                + "].";
    }
    /**
     * Marks the repository as corrupted. This puts the repository in a state where its tracked value for
     * {@link RepositoryMetadata#pendingGeneration()} is unchanged while its value for {@link RepositoryMetadata#generation()} is set to
     * {@link RepositoryData#CORRUPTED_REPO_GEN}. In this state, the repository can not be used any longer and must be removed and
     * recreated after the problem that lead to it being marked as corrupted has been fixed.
     *
     * @param corruptedGeneration generation that failed to load because the index file was not found but that should have loaded
     * @param originalException   exception that lead to the failing to load the {@code index-N} blob
     * @param listener            listener to invoke once done
     */
    private void markRepoCorrupted(long corruptedGeneration, Exception originalException, ActionListener<Void> listener) {
        assert corruptedGeneration != RepositoryData.UNKNOWN_REPO_GEN;
        assert bestEffortConsistency == false;
        logger.warn(() -> "Marking repository [" + metadata.name() + "] as corrupted", originalException);
        submitUnbatchedTask(
            "mark repository corrupted [" + metadata.name() + "][" + corruptedGeneration + "]",
            new ClusterStateUpdateTask() {
                @Override
                public ClusterState execute(ClusterState currentState) {