#1703 fixed wrongly emptied out search index on policy retrieval erro…

…r due to internal timeouts

Signed-off-by: Thomas Jäckle <thomas.jaeckle@beyonnex.io>

internal/utils/cache/src/main/java/org/eclipse/ditto/internal/utils/cache/entry/Entry.java

@@ -21,10 +21,6 @@
  */
 public interface Entry<T> {
 
-    static <T> Entry<T> permanent(final T value) {
-        return new ExistentEntry<>(Long.MAX_VALUE, value);
-    }
-
     static <T> Entry<T> of(final long revision, final T value) {
         return new ExistentEntry<>(revision, value);
     }
@@ -33,6 +29,10 @@ static <T> Entry<T> nonexistent() {
         return NonexistentEntry.getInstance();
     }
 
+    static <T> Entry<T> fetchError(final Throwable throwable) {
+        return FailedToFetchEntry.of(throwable);
+    }
+
     /**
      * Returns the revision of the cache entry.
      * An entry may only override those with smaller revisions.
@@ -43,6 +43,18 @@ static <T> Entry<T> nonexistent() {
 
     boolean exists();
 
+    /**
+     * @return whether the entry could not be fetched due to e.g. an internal timeout.
+     */
+    boolean isFetchError();
+
+    /**
+     * @return returns the cause of the internal fetch error if {@link #isFetchError()} was {@code true}.
+     */
+    default Optional<Throwable> getFetchErrorCause() {
+        return Optional.empty();
+    }
+
     /**
      * Retrieve the value if present.
      *

internal/utils/cache/src/main/java/org/eclipse/ditto/internal/utils/cache/entry/ExistentEntry.java

@@ -45,6 +45,11 @@ public boolean exists() {
         return true;
     }
 
+    @Override
+    public boolean isFetchError() {
+        return false;
+    }
+
     @Override
     public boolean equals(final Object o) {
         if (this == o) {

internal/utils/cache/src/main/java/org/eclipse/ditto/internal/utils/cache/entry/FailedToFetchEntry.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2023 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.internal.utils.cache.entry;
+
+import java.util.NoSuchElementException;
+import java.util.Optional;
+
+import javax.annotation.concurrent.Immutable;
+
+@Immutable
+final class FailedToFetchEntry<T> implements Entry<T> {
+
+    private final Throwable throwable;
+
+    /* this object is not supposed to be constructed anywhere else. */
+    private FailedToFetchEntry(final Throwable throwable) {
+        this.throwable = throwable;
+    }
+
+    @Override
+    public long getRevision() {
+        return 0;
+    }
+
+    @Override
+    public T getValueOrThrow() {
+        throw new NoSuchElementException();
+    }
+
+    @Override
+    public boolean exists() {
+        return false;
+    }
+
+    @Override
+    public boolean isFetchError() {
+        return true;
+    }
+
+    @Override
+    public Optional<Throwable> getFetchErrorCause() {
+        return Optional.of(throwable);
+    }
+
+    static <T> FailedToFetchEntry<T> of(final Throwable throwable) {
+        return new FailedToFetchEntry<>(throwable);
+    }
+
+}

internal/utils/cache/src/main/java/org/eclipse/ditto/internal/utils/cache/entry/NonexistentEntry.java

@@ -39,6 +39,11 @@ public boolean exists() {
         return false;
     }
 
+    @Override
+    public boolean isFetchError() {
+        return false;
+    }
+
     @SuppressWarnings("unchecked")
     static <T> NonexistentEntry<T> getInstance() {
         return (NonexistentEntry<T>) INSTANCE;

thingsearch/service/src/main/java/org/eclipse/ditto/thingsearch/service/persistence/write/model/ThingWriteModel.java

@@ -115,6 +115,16 @@ public static ThingWriteModel ofEmptiedOut(final Metadata metadata) {
         return new ThingWriteModel(metadata, emptiedOutThingDocument, false, 0L);
     }
 
+    /**
+     * Create a Thing write model which does not update the search index, but does "no operation" on it instead.
+     *
+     * @param metadata the metadata.
+     * @return a Thing write model.
+     */
+    public static ThingWriteModel noopWriteModel(final Metadata metadata) {
+        return new ThingWriteModel(metadata, new BsonDocument(), false, 0L);
+    }
+
     @Override
     public Optional<MongoWriteModel> toIncrementalMongo(@Nullable final AbstractWriteModel previousWriteModel,
             final int maxWireVersion) {
@@ -211,8 +221,14 @@ private Optional<MongoWriteModel> computeDiff(final ThingWriteModel lastWriteMod
             PATCH_SKIP_COUNT.increment();
             return Optional.empty();
         }
-        final var diff = tryComputeDiff(getThingDocument(), lastWriteModel.getThingDocument(), maxWireVersion);
-        if (diff.isPresent() && diff.get().isDiffSmaller()) {
+        final BsonDocument currentWriteModel = getThingDocument();
+        final var diff = tryComputeDiff(currentWriteModel, lastWriteModel.getThingDocument(), maxWireVersion);
+        if (currentWriteModel.isEmpty()) {
+            LOGGER.debug("Skipping update due to empty currentWriteModel <{}>",
+                    ((AbstractWriteModel) this).getClass().getSimpleName());
+            PATCH_SKIP_COUNT.increment();
+            return Optional.empty();
+        } else if (diff.isPresent() && diff.get().isDiffSmaller()) {
             final var aggregationPipeline = diff.get().consumeAndExport();
             if (aggregationPipeline.isEmpty()) {
                 LOGGER.debug("Skipping update due to {} <{}>", "empty diff",

thingsearch/service/src/main/java/org/eclipse/ditto/thingsearch/service/persistence/write/streaming/EnforcementFlow.java

@@ -24,6 +24,7 @@
 
 import javax.annotation.Nullable;
 
+import org.eclipse.ditto.base.model.exceptions.AskException;
 import org.eclipse.ditto.internal.models.signalenrichment.CachingSignalEnrichmentFacade;
 import org.eclipse.ditto.internal.models.streaming.AbstractEntityIdWithRevision;
 import org.eclipse.ditto.internal.utils.cache.Cache;
@@ -179,7 +180,7 @@ public <T> Source<List<AbstractWriteModel>, T> create(
 
         return source.flatMapConcat(changes -> Source.fromIterator(changes::iterator)
                         .flatMapMerge(parallelismPerBulkShard, changedMetadata ->
-                                retrieveThingFromCachingFacade(changedMetadata.getThingId(), changedMetadata, 2)
+                                retrieveThingFromCachingFacade(changedMetadata.getThingId(), changedMetadata, 3)
                                         .flatMapConcat(pair -> {
                                             final JsonObject thing = pair.second();
                                             searchUpdateObserver.process(changedMetadata, thing);
@@ -198,7 +199,7 @@ public <T> Source<List<AbstractWriteModel>, T> create(
      */
     public Flow<ThingUpdater.Data, MongoWriteModel, NotUsed> create(final SearchUpdateMapper mapper) {
         return Flow.<ThingUpdater.Data>create()
-                .flatMapConcat(data -> retrieveThingFromCachingFacade(data.metadata().getThingId(), data.metadata(), 2)
+                .flatMapConcat(data -> retrieveThingFromCachingFacade(data.metadata().getThingId(), data.metadata(), 3)
                         .flatMapConcat(pair -> {
                             final JsonObject thing = pair.second();
                             searchUpdateObserver.process(data.metadata(), thing);
@@ -222,14 +223,20 @@ private Source<Pair<ThingId, JsonObject>, NotUsed> retrieveThingFromCachingFacad
                 .map(thing -> Pair.create(thingId, thing))
                 .recoverWithRetries(1, new PFBuilder<Throwable, Source<Pair<ThingId, JsonObject>, NotUsed>>()
                         .match(Throwable.class, error -> {
-                            if (leftRetryAttempts > 0 && error instanceof CompletionException completionException &&
+                            if (error instanceof CompletionException completionException &&
                                     completionException.getCause() instanceof AskTimeoutException) {
-                                // retry ask timeouts
-                                return retrieveThingFromCachingFacade(thingId, metadata, leftRetryAttempts - 1);
+                                if (leftRetryAttempts > 0) {
+                                    // retry ask timeouts
+                                    return retrieveThingFromCachingFacade(thingId, metadata, leftRetryAttempts - 1);
+                                } else {
+                                    log.warn("No retries left, try to SudoRetrieveThing via cache, therefore giving " +
+                                                    "up for thingId <{}>", thingId);
+                                    return Source.empty();
+                                }
                             }
 
-                            log.error("Unexpected exception during SudoRetrieveThing via cache: <{}> - {}", thingId,
-                                    error.getClass().getSimpleName(), error);
+                            log.error("Unexpected exception during SudoRetrieveThing via cache for thingId <{}> - {}",
+                                    thingId, error.getClass().getSimpleName(), error);
                             return Source.empty();
                         })
                         .build());
@@ -274,11 +281,22 @@ private Source<AbstractWriteModel, NotUsed> computeWriteModel(final Metadata met
                                 return ThingWriteModel.ofEmptiedOut(metadata);
                             }
                         } else {
-                            // no enforcer; "empty out" thing in search index
-                            log.warn(
-                                    "Computed - due to missing enforcer - 'emptied out' ThingWriteModel for metadata <{}> " +
-                                            "and thing <{}>", metadata, thing);
-                            return ThingWriteModel.ofEmptiedOut(metadata);
+                            if (entry.isFetchError()) {
+                                final Throwable fetchErrorCause = entry.getFetchErrorCause().orElse(
+                                        new IllegalStateException("No fetch error cause present when it should be")
+                                );
+                                log.warn("Computed - due to fetch error <{}: {}> on policy cache - 'no op' ThingWriteModel " +
+                                                "for metadata <{}> and thing <{}>",
+                                        fetchErrorCause.getClass().getSimpleName(), fetchErrorCause.getMessage(),
+                                        metadata, thing, fetchErrorCause
+                                );
+                                return ThingWriteModel.noopWriteModel(metadata);
+                            } else {
+                                // no enforcer; "empty out" thing in search index
+                                log.warn("Computed - due to missing enforcer - 'emptied out' ThingWriteModel for " +
+                                                "metadata <{}> and thing <{}>", metadata, thing);
+                                return ThingWriteModel.ofEmptiedOut(metadata);
+                            }
                         }
                     });
         }
@@ -320,7 +338,11 @@ private Source<Entry<Pair<Policy, Set<PolicyTag>>>, NotUsed> readCachedEnforcer(
                                 }
                             })
                             .exceptionally(error -> {
-                                log.error("Failed to read policyEnforcerCache", error);
+                                final Throwable cause = error instanceof CompletionException ? error.getCause() : error;
+                                log.warn("Failed to read policyEnforcerCache", cause);
+                                if (cause instanceof AskException askException) {
+                                    return Source.single(Entry.fetchError(askException));
+                                }
                                 return POLICY_NONEXISTENT;
                             });