Skip listing MPUs if TTL set to -1 (#127166) (#127205)

Recent versions of MinIO will sometimes leak multi-part uploads under
concurrent load, leaving them in the `ListMultipartUploads` output even
though they cannot be aborted. Today this causes repository analysis to
fail since compare-and-exchange operations will not even start if there
are any pre-existing uploads. This commit makes it possible to skip this
pre-flight check (and accept the performance consequences) by adjusting
the relevant settings.

Workaround for minio/minio#21189
Closes #122670

Author: DaveCTurner

modules/repository-s3/src/javaRestTest/java/org/elasticsearch/repositories/s3/RepositoryS3MinioBasicCredentialsRestIT.java

@@ -21,6 +21,9 @@
 
 import java.util.Locale;
 
+import static org.elasticsearch.repositories.s3.S3Service.REPOSITORY_S3_CAS_ANTI_CONTENTION_DELAY_SETTING;
+import static org.elasticsearch.repositories.s3.S3Service.REPOSITORY_S3_CAS_TTL_SETTING;
+
 @ThreadLeakFilters(filters = { TestContainersThreadFilter.class })
 @ThreadLeakScope(ThreadLeakScope.Scope.NONE) // https://github.com/elastic/elasticsearch/issues/102482
 public class RepositoryS3MinioBasicCredentialsRestIT extends AbstractRepositoryS3RestTestCase {
@@ -39,6 +42,10 @@ public class RepositoryS3MinioBasicCredentialsRestIT extends AbstractRepositoryS
         .keystore("s3.client." + CLIENT + ".access_key", ACCESS_KEY)
         .keystore("s3.client." + CLIENT + ".secret_key", SECRET_KEY)
         .setting("s3.client." + CLIENT + ".endpoint", minioFixture::getAddress)
+        // Skip listing of pre-existing uploads during a CAS because MinIO sometimes leaks them; also reduce the delay before proceeding
+        // TODO do not set these if running a MinIO version in which https://github.com/minio/minio/issues/21189 is fixed
+        .setting(REPOSITORY_S3_CAS_TTL_SETTING.getKey(), "-1")
+        .setting(REPOSITORY_S3_CAS_ANTI_CONTENTION_DELAY_SETTING.getKey(), "100ms")
         .build();
 
     @ClassRule

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3BlobContainer.java

@@ -686,6 +686,11 @@ void run(BytesReference expected, BytesReference updated, ActionListener<Optiona
          * @return {@code true} if there are already ongoing uploads, so we should not proceed with the operation
          */
         private boolean hasPreexistingUploads() {
+            final var timeToLiveMillis = blobStore.getCompareAndExchangeTimeToLive().millis();
+            if (timeToLiveMillis < 0) {
+                return false; // proceed always
+            }
+
             final var uploads = listMultipartUploads();
             logUploads("preexisting uploads", uploads);
 
@@ -694,11 +699,7 @@ private boolean hasPreexistingUploads() {
                 return false;
             }
 
-            final var expiryDate = Date.from(
-                Instant.ofEpochMilli(
-                    blobStore.getThreadPool().absoluteTimeInMillis() - blobStore.getCompareAndExchangeTimeToLive().millis()
-                )
-            );
+            final var expiryDate = Date.from(Instant.ofEpochMilli(blobStore.getThreadPool().absoluteTimeInMillis() - timeToLiveMillis));
             if (uploads.stream().anyMatch(upload -> upload.getInitiated().after(expiryDate))) {
                 logger.trace("[{}] fresh preexisting uploads vs {}", blobKey, expiryDate);
                 return true;
@@ -773,7 +774,7 @@ private int getUploadIndex(String targetUploadId, List<MultipartUpload> multipar
                     final var currentTimeMillis = blobStore.getThreadPool().absoluteTimeInMillis();
                     final var ageMillis = currentTimeMillis - multipartUpload.getInitiated().toInstant().toEpochMilli();
                     final var expectedAgeRangeMillis = blobStore.getCompareAndExchangeTimeToLive().millis();
-                    if (ageMillis < -expectedAgeRangeMillis || ageMillis > expectedAgeRangeMillis) {
+                    if (0 <= expectedAgeRangeMillis && (ageMillis < -expectedAgeRangeMillis || ageMillis > expectedAgeRangeMillis)) {
                         logger.warn(
                             """
                                 compare-and-exchange of blob [{}:{}] was initiated at [{}={}] \