fix(reindex): bound per-entry mapping so hung storage I/O can't wedge indexing

dotCMS/src/main/java/com/dotcms/content/business/json/ContentletJsonAPIImpl.java

@@ -171,6 +171,7 @@ public ImmutableContentlet toImmutable(
         builder.identifier(UtilMethods.isNotSet(contentlet.getIdentifier()) ? StringPool.BLANK : contentlet.getIdentifier() );
         builder.inode( UtilMethods.isNotSet(contentlet.getInode()) ? StringPool.BLANK : contentlet.getInode() );
 
+
         final List<Field> fields = contentlet.getContentType().fields();
         for (final Field field : fields) {
             if (isNotMappable(field)) {
@@ -305,7 +306,8 @@ private Map<String, Object> getContentletMapFromImmutable(final Contentlet immut
                 value = identifierAPI.find(identifier).getAssetName();
             } else {
                 if (field instanceof BinaryField) {
-                    value = getBinary(field, inode).orElse(null);
+                    value = getBinary(field, inode, contentletFields.get(field.variable()))
+                            .orElse(null);
                 } else {
                     value = getValue(contentletFields, field);
                 }
@@ -393,9 +395,12 @@ private boolean isNotMappable(final Field field) {
      * Once a BinaryField is found this will rebuild it.
      * @param field
      * @param inode
+     * @param storedValue the field's {@link FieldValue} from the persisted json, which carries
+     *                    the binary's file name; may be null on legacy json
      * @return
      */
-    private Optional<File> getBinary(final Field field, final String inode) {
+    private Optional<File> getBinary(final Field field, final String inode,
+            final FieldValue<?> storedValue) {
         // This validation is here to prevent an exception.
         // Cause the json gets saved twice by internalCheckin and the first time it does it no inode is set yet
 
@@ -409,11 +414,22 @@ private Optional<File> getBinary(final Field field, final String inode) {
                         + inode
                         + java.io.File.separator
                         + field.variable());
-        if (binaryFileFolder.exists()) {
-            final java.io.File[] files = binaryFileFolder.listFiles(binaryFileFilter);
-            if (files != null && files.length > 0) {
-                return Optional.of(files[0]);
-            }
+
+        // The json stores the binary's file name (see BinaryField#fieldValue), so the path can
+        // be rebuilt with zero filesystem access. Listing the folder here — once per binary
+        // field per contentlet load — wedged the reindex pipeline on a hung S3-FUSE mount and
+        // is a per-load tax on any network-backed storage (issue #36498).
+        final Object storedName = null != storedValue ? storedValue.value() : null;
+        if (storedName instanceof String && isSet((String) storedName)
+                && !((String) storedName).contains("/") && !((String) storedName).contains("\\")) {
+            return Optional.of(new java.io.File(binaryFileFolder, (String) storedName));
+        }
+
+        // Legacy json without a stored file name: fall back to listing the folder. No exists()
+        // pre-check — listFiles() returns null for a missing folder.
+        final java.io.File[] files = binaryFileFolder.listFiles(binaryFileFilter);
+        if (files != null && files.length > 0) {
+            return Optional.of(files[0]);
         }
 
         return Optional.empty();

dotCMS/src/main/java/com/dotcms/content/elasticsearch/business/ContentletIndexAPIImpl.java

@@ -78,6 +78,7 @@
 import com.liferay.portal.util.PortalUtil;
 import com.liferay.util.StringPool;
 import com.rainerhahnekamp.sneakythrow.Sneaky;
+import io.vavr.Lazy;
 import io.vavr.control.Try;
 import java.io.IOException;
 import java.sql.Connection;
@@ -175,6 +176,20 @@ public class ContentletIndexAPIImpl implements ContentletIndexAPI {
 
     private static final ObjectMapper objectMapper = DotObjectMapperProvider.createDefaultMapper();
 
+    /**
+     * Max seconds a single reindex-journal entry may spend loading/mapping its contentlets
+     * before it is marked failed and the queue moves on — guards against hung filesystem I/O
+     * on network-backed storage (issue #36498). {@code 0} disables the guard.
+     */
+    public static final String REINDEX_CONTENTLET_MAPPING_TIMEOUT_SECONDS =
+            "REINDEX_CONTENTLET_MAPPING_TIMEOUT_SECONDS";
+
+    private static final Lazy<ReindexMappingRunner> mappingRunner = Lazy.of(() ->
+            new ReindexMappingRunner(
+                    () -> Config.getIntProperty(REINDEX_CONTENTLET_MAPPING_TIMEOUT_SECONDS, 120),
+                    Config.getIntProperty("REINDEX_CONTENTLET_MAPPING_MAX_THREADS", 8),
+                    DbConnectionFactory::closeSilently));
+
     public ContentletIndexAPIImpl() {
         this(new ContentletIndexOperationsES(),
              CDIUtils.getBeanThrows(ContentletIndexOperationsOS.class));
@@ -2318,17 +2333,43 @@ private void appendBulkRequestInternal(final IndexBulkRequest req, final Reindex
     private void appendBulkRequestToProcessor(final IndexBulkProcessor proc,
             final ReindexEntry idx) throws DotDataException {
         try {
-            for (final Contentlet contentlet : loadVersionInodes(idx).values()) {
-                Logger.debug(this, String.format("Indexing id: '%s', priority: '%s'",
-                        contentlet.getInode(), idx.getPriority()));
-                contentlet.setIndexPolicy(IndexPolicy.DEFER);
-                addBulkRequestToProcessor(proc, List.of(contentlet), idx.isReindex());
-            }
+            // Bounded timeout: loading and mapping touch binary files, and a hung stat on
+            // network-backed storage must fail this entry instead of wedging the reindex
+            // thread forever (issue #36498).
+            mappingRunner().run(() -> {
+                mapEntryForProcessor(proc, idx);
+                return null;
+            }, "reindex entry with identifier '" + idx.getIdentToIndex() + "'");
         } catch (final Exception e) {
             APILocator.getReindexQueueAPI().markAsFailed(idx, e.getMessage());
         }
     }
 
+    /**
+     * The shared timeout guard for per-entry mapping work. Seam for tests, which override it
+     * to supply a runner with a test-controlled timeout and no DB cleanup.
+     */
+    @VisibleForTesting
+    ReindexMappingRunner mappingRunner() {
+        return mappingRunner.get();
+    }
+
+    /**
+     * Loads all versions of the entry's contentlet and appends their index operations to the
+     * processor. Runs on a {@link ReindexMappingRunner} worker thread when the mapping timeout
+     * guard is enabled, so it must not rely on caller-thread state.
+     */
+    @VisibleForTesting
+    void mapEntryForProcessor(final IndexBulkProcessor proc, final ReindexEntry idx)
+            throws Exception {
+        for (final Contentlet contentlet : loadVersionInodes(idx).values()) {
+            Logger.debug(this, String.format("Indexing id: '%s', priority: '%s'",
+                    contentlet.getInode(), idx.getPriority()));
+            contentlet.setIndexPolicy(IndexPolicy.DEFER);
+            addBulkRequestToProcessor(proc, List.of(contentlet), idx.isReindex());
+        }
+    }
+
     private void appendBulkRequestFromContentlets(final IndexBulkRequest req,
             final List<Contentlet> contentToIndex) {
         this.addBulkRequest(req, contentToIndex, false);

dotCMS/src/main/java/com/dotcms/content/elasticsearch/business/ReindexMappingRunner.java

@@ -0,0 +1,101 @@
+package com.dotcms.content.elasticsearch.business;
+
+import com.dotmarketing.exception.DotRuntimeException;
+import com.dotmarketing.util.Logger;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.function.IntSupplier;
+
+/**
+ * Runs per-entry reindex mapping work under a bounded timeout so that a hung storage operation
+ * (e.g. a {@code File.exists()} that never returns on network-backed storage such as NFS, EFS or
+ * an S3 FUSE mount) cannot wedge the single reindex thread forever. See issue #36498: one
+ * unanswered {@code stat(2)} silently froze all content indexing for an instance.
+ *
+ * <p>Each task runs on its own virtual thread. A task that exceeds the timeout is abandoned, not
+ * interrupted-and-reused: a thread stuck in an uninterruptible native stat does not respond to
+ * interrupt. A wedged virtual thread pins a carrier thread (file I/O does not unmount), so
+ * in-flight tasks are capped by a semaphore whose permit is only released when the task actually
+ * finishes — wedged tasks hold their permit, and when every permit is held new work is rejected
+ * and loudly logged, since that means the storage itself is down.</p>
+ */
+class ReindexMappingRunner {
+
+    private final ExecutorService executor = Executors.newThreadPerTaskExecutor(
+            Thread.ofVirtual().name("dot-reindex-mapping-", 0).factory());
+    private final IntSupplier timeoutSeconds;
+    private final Runnable perTaskCleanup;
+    private final Semaphore inFlight;
+    private final int maxThreads;
+
+    /**
+     * @param timeoutSeconds resolved per task; {@code <= 0} disables the guard entirely and runs
+     *                       tasks inline on the calling thread (legacy behavior)
+     * @param maxThreads     hard cap on concurrent (including wedged) in-flight tasks
+     * @param perTaskCleanup runs on the worker thread after each task completes, wedged or not —
+     *                       used to release thread-local resources such as DB connections
+     */
+    ReindexMappingRunner(final IntSupplier timeoutSeconds, final int maxThreads,
+            final Runnable perTaskCleanup) {
+        this.timeoutSeconds = timeoutSeconds;
+        this.perTaskCleanup = perTaskCleanup;
+        this.maxThreads = maxThreads;
+        this.inFlight = new Semaphore(maxThreads);
+    }
+
+    /**
+     * Runs the task, failing with a {@link DotRuntimeException} if it does not complete within
+     * the configured timeout so the caller can mark the journal entry as failed and keep
+     * draining the queue.
+     *
+     * @throws Exception the task's own exception, or a {@link DotRuntimeException} on timeout or
+     *                   pool exhaustion
+     */
+    <T> T run(final Callable<T> task, final String description) throws Exception {
+        final int timeout = timeoutSeconds.getAsInt();
+        if (timeout <= 0) {
+            return task.call();
+        }
+        if (!inFlight.tryAcquire()) {
+            final String message = "Reindex mapping pool exhausted: all " + maxThreads
+                    + " in-flight tasks are busy or wedged in storage I/O — cannot map "
+                    + description + ". The underlying storage may be down.";
+            Logger.error(this, message);
+            throw new DotRuntimeException(message);
+        }
+        final Future<T> future = executor.submit(() -> {
+            try {
+                return task.call();
+            } finally {
+                try {
+                    perTaskCleanup.run();
+                } finally {
+                    // A wedged task holds its permit until the native call returns (if ever),
+                    // so wedged threads count against the cap instead of piling up unbounded —
+                    // and a throwing cleanup must never leak the permit.
+                    inFlight.release();
+                }
+            }
+        });
+        try {
+            return future.get(timeout, TimeUnit.SECONDS);
+        } catch (final TimeoutException timedOut) {
+            // Frees threads in interruptible waits; a thread wedged in native I/O ignores this
+            // and is simply abandoned.
+            future.cancel(true);
+            final String message = "Timed out after " + timeout + "s mapping " + description
+                    + " for reindex — likely hung storage I/O on a binary field. Marking the "
+                    + "journal entry as failed and continuing with the queue.";
+            Logger.error(this, message);
+            throw new DotRuntimeException(message, timedOut);
+        } catch (final ExecutionException failed) {
+            throw failed.getCause() instanceof Exception ? (Exception) failed.getCause() : failed;
+        }
+    }
+}