fix(search): journal content deletes so lost index removals are retried (#37276)

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

@@ -3131,7 +3131,13 @@ private void removeContentAndProcessDependencies(final Contentlet contentlet,
 
         final DualIndexBulkRequest dualReq = bulkRequest instanceof DualIndexBulkRequest ? (DualIndexBulkRequest) bulkRequest : null;
 
-        for (final ContentletIndexOperations ops : router.writeProviders()) {
+        // writeProviders() is ordered primary-first in every phase (0 → [ES], 1/2 → [ES, OS],
+        // 3 → [OS]), so element 0 is the provider whose outcome the caller is entitled to.
+        final List<ContentletIndexOperations> deleteProviders = router.writeProviders();
+        final ContentletIndexOperations primary = deleteProviders.get(0);
+        int primaryDeleteOps = 0;
+
+        for (final ContentletIndexOperations ops : deleteProviders) {
             final ProviderIndices indices = loadProviderIndicesQuietly(ops);
             if (indices == null) {
                 continue;
@@ -3142,20 +3148,46 @@ private void removeContentAndProcessDependencies(final Contentlet contentlet,
             } else {
                 providerReq = bulkRequest;
             }
+            int opsAdded = 0;
             if (indices.live != null) {
                 ops.addDeleteOp(providerReq, indices.live, id);
+                opsAdded++;
             }
             if (indices.reindexLive != null) {
                 ops.addDeleteOp(providerReq, indices.reindexLive, id);
+                opsAdded++;
             }
             if (!onlyLive) {
                 if (indices.working != null) {
                     ops.addDeleteOp(providerReq, indices.working, id);
+                    opsAdded++;
                 }
                 if (indices.reindexWorking != null) {
                     ops.addDeleteOp(providerReq, indices.reindexWorking, id);
+                    opsAdded++;
                 }
             }
+            if (ops == primary) {
+                primaryDeleteOps = opsAdded;
+            }
+        }
+
+        // The primary contributing no delete operations means the removal did not happen, and
+        // putToIndex early-returns on an empty batch — so without this check it is
+        // indistinguishable from a completed removal (#37276, loss point L2).
+        //
+        // Counting the operations rather than testing for a null ProviderIndices covers both
+        // ways the primary can come up empty: its pointers failed to load (loadProviderIndices
+        // threw), or they loaded but hold no active index at all. The second reads as success
+        // just as silently as the first, and only the operation count sees both.
+        //
+        // A shadow provider keeps warn-and-continue, matching how putToIndex already isolates
+        // the OS leg under ADR-0009.
+        if (primaryDeleteOps == 0) {
+            throw new DotRuntimeException(
+                    "Cannot remove content from the index: the primary provider ("
+                            + primary.getClass().getSimpleName() + ") resolved no active index "
+                            + "for document " + id + ". The removal was NOT performed.");
         }
 
         if (!onlyLive && UtilMethods.isSet(relationships)) {

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

@@ -5,6 +5,7 @@
 import static com.dotmarketing.common.reindex.ReindexThread.ELASTICSEARCH_CONCURRENT_REQUESTS;
 
 import com.dotcms.content.index.ContentletIndexOperations;
+import com.google.common.annotations.VisibleForTesting;
 import com.dotcms.content.index.IndexAPI;
 import com.dotcms.content.index.opensearch.ContentletIndexOperationsOS;
 import com.dotcms.content.index.domain.CreateIndexStatus;
@@ -203,11 +204,7 @@ public void putToIndex(final IndexBulkRequest req) {
         try {
             final BulkResponse response = RestHighLevelClientProvider.getInstance()
                     .getClient().bulk(bulkRequest, RequestOptions.DEFAULT);
-            if (response != null && response.hasFailures()) {
-                Logger.error(this,
-                        "Error reindexing (" + response.getItems().length + ") content(s): "
-                                + response.buildFailureMessage());
-            }
+            handleBulkResponse(response);
         } catch (final Exception e) {
             if (ExceptionUtil.causedBy(e, IllegalStateException.class)) {
                 ContentletFactory.rebuildRestHighLevelClientIfNeeded(e);
@@ -217,6 +214,33 @@ public void putToIndex(final IndexBulkRequest req) {
         }
     }
 
+    /**
+     * Decides what a bulk response means to the caller.
+     *
+     * <p>Extracted from {@link #putToIndex(IndexBulkRequest)} so the policy can be exercised
+     * without a cluster — the HTTP call is not what is interesting here, the verdict is.</p>
+     *
+     * @param response the response from the bulk call; {@code null} is tolerated
+     */
+    @VisibleForTesting
+    void handleBulkResponse(final BulkResponse response) {
+        if (response != null && response.hasFailures()) {
+            // A bulk can return normally while rejecting individual items — a saturated write
+            // queue, an unavailable shard, a version conflict. Logging and returning made those
+            // indistinguishable from success, so a caller could commit a content deletion whose
+            // index removal never landed (#37276, loss point L3). Raising lets the journal entry
+            // be marked failed and retried instead.
+            //
+            // The message says "index operations" rather than "reindexing": the batch may well
+            // have carried removals, and the old wording is why searching production logs for
+            // failed deletes came back empty.
+            final String message = "Error applying (" + response.getItems().length
+                    + ") index operation(s): " + response.buildFailureMessage();
+            Logger.error(this, message);
+            throw new DotRuntimeException(message);
+        }
+    }
+
     // =========================================================================
     // Async bulk-processor write path
     // =========================================================================

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

@@ -3053,6 +3053,45 @@ private void deleteMultitrees(final Contentlet contentlet, final User user)
         }
     }
 
+    /**
+     * Records a durable index-removal intent in {@code dist_reindex_journal} for every identifier
+     * being destroyed, so that a failed or lost index write is retried instead of silently
+     * orphaning the document.
+     *
+     * <p>Must be called inside the transaction that deletes the contentlet rows: a rolled-back
+     * destroy must not leave the index owing a removal for content that still exists.</p>
+     *
+     * <p>A journal entry is <b>identifier-wide</b> — the consumer fans out across every language
+     * and variant of the identifier. That is correct here, because destruction removes every
+     * version and language by design. It is <b>not</b> transportable to the unpublish/archive
+     * path, where a removal is per language and an identifier-wide entry would drop languages
+     * that are still live.</p>
+     *
+     * <p>Takes identifiers rather than contentlets on purpose: they are collected <b>before</b>
+     * the rows are deleted, so this method cannot depend on the state of {@link Contentlet}
+     * objects that {@code contentFactory.delete} has already processed. The signature is what
+     * enforces that ordering — a comment would not.</p>
+     *
+     * @param identifiers identifiers of the contentlets being destroyed, collected before deletion
+     */
+    private void journalContentDeletes(final Set<String> identifiers) {
+
+        if (identifiers == null || identifiers.isEmpty()) {
+            return;
+        }
+
+        try {
+            APILocator.getReindexQueueAPI().addIdentifierDelete(identifiers);
+        } catch (final DotDataException e) {
+            // The journal row is the durability guarantee; without it the removal is only as good
+            // as the in-memory listener. Fail the destroy rather than commit a deletion whose
+            // index removal nothing is tracking.
+            throw new DotRuntimeException(
+                    "Unable to journal the index removal for destroyed content: " + e.getMessage(),
+                    e);
+        }
+    }
+
     /**
      * Completely destroys the given list of {@link Contentlet} objects (versions, relationships,
      * associated contents, binary files) in all of their languages.
@@ -3117,8 +3156,25 @@ private boolean destroyContentlets(final List<Contentlet> contentlets, final Use
 
         this.backupDestroyedContentlets(contentlets, user);
 
+        // Collected before the delete so the journal never depends on post-deletion object
+        // state — see journalContentDeletes.
+        final Set<String> destroyedIdentifiers = contentletsVersion.stream()
+                .map(Contentlet::getIdentifier)
+                .filter(UtilMethods::isSet)
+                .collect(Collectors.toSet());
+
         // Delete all the versions of the contentlets to delete
         this.contentFactory.delete(contentletsVersion);
+
+        // Record the index removal durably, in the SAME transaction that just deleted the rows.
+        // The commit listener below stays as the low-latency path, but it lives only in memory:
+        // if it is lost — the JVM stops between commit and execution, the shared pool rejects the
+        // task, or the bulk comes back with per-item failures — nothing would remember that the
+        // index still owes a removal, and the document is orphaned permanently. The journal row
+        // is what ReindexThread retries. This mirrors what the add path already does (see
+        // ContentletIndexAPIImpl#addContentToIndex, IndexPolicy.DEFER branch).
+        this.journalContentDeletes(destroyedIdentifiers);
+
         // Remove the contentlets from the search index and cache
         final Set<String> removedFromIndex = new HashSet<>();
         for (final Contentlet contentlet : contentletsVersion) {
@@ -3546,8 +3602,21 @@ public void deleteAllVersionsandBackup(List<Contentlet> contentlets, User user,
             contentletInodes.add(element.getInode());
         }
 
+        // Collected before the delete — see journalContentDeletes.
+        final Set<String> destroyedIdentifiers = contentletsVersion.stream()
+                .map(Contentlet::getIdentifier)
+                .filter(UtilMethods::isSet)
+                .collect(Collectors.toSet());
+
         contentFactory.delete(contentletsVersion);
 
+        // Same durability requirement as destroyContentlets: the index removal below is deferred
+        // and in-memory, so record the intent in the journal inside this transaction. Safe here
+        // because contentletsVersion was built from findAllVersions(identifier) above — every
+        // version and language of the identifier is going away, which is what an identifier-wide
+        // journal entry means.
+        this.journalContentDeletes(destroyedIdentifiers);
+
         for (Contentlet contentlet : perCons) {
             indexAPI.removeContentFromIndex(contentlet);
             CacheLocator.getIdentifierCache().removeFromCacheByVersionable(contentlet);
@@ -3599,6 +3668,11 @@ public void delete(List<Contentlet> contentlets, User user, boolean respectFront
 
         contentFactory.delete(contentletsVersion);
 
+        // Deliberately NOT journalled (#37276). Unlike destroyContentlets and
+        // deleteAllVersionsandBackup, contentletsVersion here is just the caller's list — this
+        // method can delete a subset of an identifier's versions or languages. A journal entry is
+        // identifier-wide, so recording one would remove index documents for languages that still
+        // exist. Same reason the unpublish/archive path is excluded.
         for (Contentlet contentlet : perCons) {
             indexAPI.removeContentFromIndex(contentlet);
             CacheLocator.getIdentifierCache().removeFromCacheByVersionable(contentlet);

dotCMS/src/main/java/com/dotcms/content/index/opensearch/ContentletIndexOperationsOS.java

@@ -21,6 +21,7 @@
 import com.dotmarketing.business.APILocator;
 import com.dotmarketing.common.reindex.ReindexThread;
 import com.dotmarketing.exception.DotDataException;
+import com.google.common.annotations.VisibleForTesting;
 import com.dotmarketing.exception.DotRuntimeException;
 import com.dotmarketing.util.Logger;
 import com.dotcms.rest.api.v1.DotObjectMapperProvider;
@@ -293,23 +294,52 @@ public void putToIndex(final IndexBulkRequest req) {
                         }
                         return b;
                     }));
-            if (response.errors()) {
-                for (final BulkResponseItem item : response.items()) {
-                    if (item.error() != null) {
-                        Logger.error(this,
-                                "OS bulk putToIndex error — id=" + item.id()
-                                        + " op=" + item.operationType()
-                                        + " type=" + item.error().type()
-                                        + " reason=" + item.error().reason());
-                    }
-                }
-            }
+            handleBulkResponse(response);
         } catch (final Exception e) {
             Logger.warnAndDebug(ContentletIndexOperationsOS.class, e);
             throw new DotRuntimeException(e.getMessage(), e);
         }
     }
 
+    /**
+     * Decides what a bulk response means to the caller.
+     *
+     * <p>Extracted from {@link #putToIndex(IndexBulkRequest)} so the policy can be exercised
+     * without a cluster — the HTTP call is not what is interesting here, the verdict is.</p>
+     *
+     * @param response the response from the bulk call; {@code null} is tolerated
+     */
+    @VisibleForTesting
+    void handleBulkResponse(final BulkResponse response) {
+        if (response == null || !response.errors()) {
+            return;
+        }
+
+        // Same contract as the Elasticsearch provider (#37276, loss point L3): a bulk that
+        // returns normally while rejecting items must not read as success to the caller.
+        //
+        // This matters most in phase 3, where OpenSearch is the sole provider and there is no
+        // shadow leg to absorb the loss. In dual-write phases the router already isolates the
+        // shadow (ContentletIndexAPIImpl#putToIndex), so raising here keeps ADR-0009 intact:
+        // an OS failure is still swallowed while OS is the shadow, and propagates once primary.
+        final StringBuilder detail = new StringBuilder();
+        for (final BulkResponseItem item : response.items()) {
+            if (item.error() != null) {
+                final String itemMessage = "OS bulk index operation error — id=" + item.id()
+                        + " op=" + item.operationType()
+                        + " type=" + item.error().type()
+                        + " reason=" + item.error().reason();
+                Logger.error(this, itemMessage);
+                if (detail.length() > 0) {
+                    detail.append("; ");
+                }
+                detail.append(itemMessage);
+            }
+        }
+
+        throw new DotRuntimeException(detail.toString());
+    }
+
     // =========================================================================
     // Async bulk-processor write path
     // =========================================================================

dotCMS/src/main/java/com/dotmarketing/common/reindex/ReindexQueueFactory.java

@@ -294,7 +294,15 @@ protected Map<String, ReindexEntry> findContentToReindex(final int recordsToRetu
         }
 
         for (ReindexEntry entry; (entry = queue.poll()) != null; ) {
-            contentToIndex.put(entry.getIdentToIndex(), entry);
+            // One outcome per identifier per batch, resolved by row id rather than by the order
+            // the entries happened to be polled in. Two entries for the same identifier are
+            // successive statements about what the index should hold, and only the newest is
+            // true: a DELETE written after a REINDEX means the content is gone, so applying the
+            // REINDEX afterwards would re-add a document for content that no longer exists.
+            // The loser stays in dist_reindex_journal and is collected on a later pass.
+            contentToIndex.merge(entry.getIdentToIndex(), entry,
+                    (existing, candidate) -> candidate.getId() > existing.getId()
+                            ? candidate : existing);
             if (contentToIndex.size() >= recordsToReturn) {
                 while (entry.equals(queue.peek())) {
                     // drain duplicate items

dotCMS/src/main/java/com/dotmarketing/portlets/contentlet/model/IndexPolicyProvider.java

@@ -25,7 +25,13 @@ public static IndexPolicyProvider getInstance() {
     } // getInstance.
 
     /**
-     * Give the index policy for single content, by default it is {@link IndexPolicy}.WAIT_FOR
+     * Give the index policy for single content. Defaults to {@link IndexPolicy#DEFER},
+     * overridable with {@code INDEX_POLICY_SINGLE_CONTENT}.
+     *
+     * <p>DEFER means the index write is handed to a post-commit listener rather than applied
+     * inline, which is why content deletion needs a durable journal record to survive a failed
+     * or lost listener (see #37276).</p>
+     *
      * @return  IndexPolicy
      */
     public IndexPolicy forSingleContent () {
@@ -38,7 +44,9 @@ public IndexPolicy forSingleContent () {
         return this.singleContentIndexPolicy;
     }
     /**
-     * Give the index policy for single content, by default it is {@link IndexPolicy}.WAIT_FOR
+     * Give the index policy for a content's dependencies. Defaults to
+     * {@link IndexPolicy#DEFER}, overridable with {@code INDEX_POLICY_DEPENDENCIES}.
+     *
      * @return  IndexPolicy
      */
     public IndexPolicy forContentDependencies () {