Increment failuresAfterMeaningfulProgress before retry (#104438)

The count for failuresAfterMeaningfulProgress should be incremented
before testing whether to retry s3 blob read. This is because we do not
want to count failuresAfterMeaningfulProgress towards actual number of
failures, i.e. we want to keep retrying if the last attempt has made
meaningful progress.  In addition, failuresAfterMeaningfulProgress
should _not_ be incremented on stream openning failures.

Resolves: #104436

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

@@ -171,6 +171,10 @@ class S3RetryingInputStream extends InputStream {
     }
 
     private void reopenStreamOrFail(IOException e) throws IOException {
+        final long meaningfulProgressSize = Math.max(1L, blobStore.bufferSizeInBytes() / 100L);
+        if (currentStreamProgress() >= meaningfulProgressSize) {
+            failuresAfterMeaningfulProgress += 1;
+        }
         final long delayInMillis = maybeLogAndComputeRetryDelay("reading", e);
         maybeAbort(currentStream);
         IOUtils.closeWhileHandlingException(currentStream);
@@ -213,11 +217,6 @@ class S3RetryingInputStream extends InputStream {
     }
 
     private void logForRetry(Level level, String action, Exception e) {
-        final long meaningfulProgressSize = Math.max(1L, blobStore.bufferSizeInBytes() / 100L);
-        final long currentStreamProgress = Math.subtractExact(Math.addExact(start, currentOffset), currentStreamFirstOffset);
-        if (currentStreamProgress >= meaningfulProgressSize) {
-            failuresAfterMeaningfulProgress += 1;
-        }
         logger.log(
             level,
             () -> format(
@@ -232,7 +231,7 @@ class S3RetryingInputStream extends InputStream {
                 start + currentOffset,
                 purpose.getKey(),
                 attempt,
-                currentStreamProgress,
+                currentStreamProgress(),
                 failuresAfterMeaningfulProgress,
                 maxRetriesForNoMeaningfulProgress()
             ),
@@ -240,6 +239,10 @@ class S3RetryingInputStream extends InputStream {
         );
     }
 
+    private long currentStreamProgress() {
+        return Math.subtractExact(Math.addExact(start, currentOffset), currentStreamFirstOffset);
+    }
+
     private boolean shouldRetry(int attempt) {
         if (purpose == OperationPurpose.REPOSITORY_ANALYSIS) {
             return false;

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3BlobContainerRetriesTests.java

@@ -510,7 +510,6 @@ public class S3BlobContainerRetriesTests extends AbstractBlobContainerRetriesTes
         assertEquals(blobSize, bytesReceived.get());
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/104436")
     public void testReadRetriesAfterMeaningfulProgress() throws Exception {
         final int maxRetries = between(0, 5);
         final int bufferSizeBytes = scaledRandomIntBetween(