When a shard is to be snapshotted, we acquire a store ref (wrapped in an index commit) and then enqueue the shard snapshot task for execution by the repository:

https://github.com/elastic/elasticsearch/blob/b124763772748d8993911e4a7ba94b17472a9539/server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java#L369-L384

If the shard is closed while its snapshot is in the queue then it looks like we do not release the store ref promptly. Instead, we keep hold of it until it reaches the front of the queue, which could be many minutes later. During this time the shard cannot fully shut down and release the shard lock, leading to a `ShardLockObtainFailedException` being thrown on attempts to reinitialise the shard.
server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java
}
    private static InputStream maybeRateLimit(
        InputStream stream,
        Supplier<RateLimiter> rateLimiterSupplier,
        RateLimitingInputStream.Listener throttleListener
    ) {
        return new RateLimitingInputStream(stream, rateLimiterSupplier, throttleListener);
    }
    /**
     * Wrap the restore rate limiter (controlled by the repository setting `max_restore_bytes_per_sec` and the cluster setting
     * `indices.recovery.max_bytes_per_sec`) around the given stream. Any throttling is reported to the given listener and not otherwise
     * recorded in the value returned by {@link BlobStoreRepository#getRestoreThrottleTimeInNanos}.
     */
    public InputStream maybeRateLimitRestores(InputStream stream) {
        return maybeRateLimitRestores(stream, restoreRateLimitingTimeInNanos::inc);
    }
    /**
     * Wrap the restore rate limiter (controlled by the repository setting `max_restore_bytes_per_sec` and the cluster setting
     * `indices.recovery.max_bytes_per_sec`) around the given stream. Any throttling is recorded in the value returned by {@link
     * BlobStoreRepository#getRestoreThrottleTimeInNanos}.
     */
    public InputStream maybeRateLimitRestores(InputStream stream, RateLimitingInputStream.Listener throttleListener) {
        return maybeRateLimit(
            maybeRateLimit(stream, () -> restoreRateLimiter, throttleListener),
            recoverySettings::rateLimiter,
            throttleListener
        );
    }
    /**
     * Wrap the snapshot rate limiter around the given stream. Any throttling is recorded in the value returned by