Fix reused/recovered bytes for files that are only partially recovered from cache (elastic#95987) (elastic#97102)

Before elastic#95891 a file was considered as reused in recovery if
it was fully cached in the cold persistent cache. Otherwise
the full file length was reported as recovered from the
blob store during prewarming.

While working on elastic#95891 I changed the
CachedBlobContainerIndexInput.prefetchPart()
method to be more precise on the number of bytes that were
effectively read from the blob store during prewarming.

But this obviously broke some tests (elastic#95970) because a file
cannot be partially recovered from disk and from remote. This
change restores the previous behavior with one adjustment:
the file is considered as reused if the prewarm method
effectively prefetched 0 bytes.

Closes elastic#95970
Closes elastic#95994

Author: tlrx

docs/changelog/95987.yaml

@@ -0,0 +1,8 @@
+pr: 95987
+summary: Fix reused/recovered bytes for files that are only partially recovered from
+  cache
+area: Snapshot/Restore
+type: bug
+issues:
+ - 95970
+ - 95994

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/SearchableSnapshotsIntegTests.java

@@ -94,7 +94,6 @@
 
 public class SearchableSnapshotsIntegTests extends BaseSearchableSnapshotsIntegTestCase {
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/pull/95987")
     public void testCreateAndRestoreSearchableSnapshot() throws Exception {
         final String fsRepoName = randomAlphaOfLength(10);
         final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);

x-pack/plugin/searchable-snapshots/src/internalClusterTest/java/org/elasticsearch/xpack/searchablesnapshots/recovery/SearchableSnapshotRecoveryStateIntegrationTests.java

@@ -61,7 +61,6 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
         return CollectionUtils.appendToCopy(super.nodePlugins(), TestRepositoryPlugin.class);
     }
 
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/95994")
     public void testRecoveryStateRecoveredBytesMatchPhysicalCacheState() throws Exception {
         final String fsRepoName = randomAlphaOfLength(10);
         final String indexName = randomAlphaOfLength(10).toLowerCase(Locale.ROOT);

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/store/SearchableSnapshotDirectory.java

@@ -506,13 +506,32 @@ private void prewarmCache(ActionListener<Void> listener, Supplier<Boolean> cance
 
                     final AtomicLong prefetchedBytes = new AtomicLong(0L);
                     try (var fileListener = new RefCountingListener(ActionListener.runBefore(completionListener.acquire().map(v -> {
+                        // we don't support files to be reported as partially recovered from disk and partially from the blob store, but
+                        // this is something that can happen for fully mounted searchable snapshots. It is possible that prewarming
+                        // prefetched nothing if a concurrent search was executing (and cached the data) or if the data were fetched from
+                        // the blob cache system index.
                         if (prefetchedBytes.get() == 0L) {
                             recoveryState.markIndexFileAsReused(file.physicalName());
                         } else {
-                            recoveryState.getIndex().addRecoveredFromSnapshotBytesToFile(file.physicalName(), prefetchedBytes.get());
+                            recoveryState.getIndex().addRecoveredFromSnapshotBytesToFile(file.physicalName(), file.length());
                         }
                         return v;
                     }), () -> IOUtils.closeWhileHandlingException(input)))) {
+
+                        if (input instanceof CachedBlobContainerIndexInput cachedIndexInput) {
+                            if (cachedIndexInput.getPersistentCacheInitialLength() == file.length()) {
+                                logger.trace(
+                                    () -> format(
+                                        "%s file [%s] is already available in cache (%d bytes)",
+                                        shardId,
+                                        file.physicalName(),
+                                        file.length()
+                                    )
+                                );
+                                continue;
+                            }
+                        }
+
                         for (int p = 0; p < file.numberOfParts(); p++) {
                             final int part = p;
                             prewarmTaskRunner.enqueueTask(fileListener.acquire().map(releasable -> {

x-pack/plugin/searchable-snapshots/src/main/java/org/elasticsearch/xpack/searchablesnapshots/store/input/CachedBlobContainerIndexInput.java

@@ -122,6 +122,13 @@ protected void readWithoutBlobCache(ByteBuffer b) throws Exception {
         assert bytesRead == length : bytesRead + " vs " + length;
     }
 
+    /**
+     * @return Returns the number of bytes already cached for the file in the cold persistent cache
+     */
+    public long getPersistentCacheInitialLength() throws Exception {
+        return cacheFileReference.get().getInitialLength();
+    }
+
     /**
      * Prefetches a complete part and writes it in cache. This method is used to prewarm the cache.
      *