Online Indexer: replace the synchronized runner with a heartbeat

   Each indexing session, for each index, will create a key-value heartbeat of the format:
        [prefix, xid] -> [indexing-type, genesis time, heartbeat time]
   Indexing session that are expected to be exclusive will throw an exception if another, active, session exists.

   Motivation:

        1. Represent the heartbeat in every index during multi target indexing (currently - only the master index has a sync lock)
        2. Keep a heartbeat during mutual indexing, which can allow better automatic decision making
        3. Decide about exclusiveness according to the indexing method (currently - user input)

Resolve #3529

Author: jjezra

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java

@@ -45,13 +45,10 @@
 import com.apple.foundationdb.record.provider.common.StoreTimer;
 import com.apple.foundationdb.record.provider.common.StoreTimerSnapshot;
 import com.apple.foundationdb.record.provider.foundationdb.indexing.IndexingRangeSet;
-import com.apple.foundationdb.record.provider.foundationdb.synchronizedsession.SynchronizedSessionRunner;
 import com.apple.foundationdb.record.query.plan.RecordQueryPlanner;
 import com.apple.foundationdb.record.query.plan.synthetic.SyntheticRecordFromStoredRecordPlan;
 import com.apple.foundationdb.record.query.plan.synthetic.SyntheticRecordPlanner;
 import com.apple.foundationdb.subspace.Subspace;
-import com.apple.foundationdb.synchronizedsession.SynchronizedSession;
-import com.apple.foundationdb.synchronizedsession.SynchronizedSessionLockedException;
 import com.apple.foundationdb.tuple.ByteArrayUtil2;
 import com.apple.foundationdb.tuple.Tuple;
 import com.google.protobuf.Message;
@@ -81,7 +78,6 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
 import java.util.function.Function;
-import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 /**
@@ -106,6 +102,7 @@ public abstract class IndexingBase {
     private final long startingTimeMillis;
     private long lastTypeStampCheckMillis;
     private Map<String, IndexingMerger> indexingMergerMap = null;
+    private IndexingHeartbeat heartbeat = null; // this will stay null for index scrubbing
 
     IndexingBase(@Nonnull IndexingCommon common,
                  @Nonnull OnlineIndexer.IndexingPolicy policy) {
@@ -157,28 +154,13 @@ public CompletableFuture<Void> buildIndexAsync(boolean markReadable, boolean use
         KeyValueLogMessage message = KeyValueLogMessage.build("build index online",
                 LogMessageKeys.SHOULD_MARK_READABLE, markReadable);
         long startNanos = System.nanoTime();
-        final CompletableFuture<Void> buildIndexAsyncFuture;
         FDBDatabaseRunner runner = getRunner();
-        Index index = common.getPrimaryIndex();
-        if (runner.getTimer() != null) {
-            lastProgressSnapshot = StoreTimerSnapshot.from(runner.getTimer());
+        final FDBStoreTimer timer = runner.getTimer();
+        if ( timer != null) {
+            lastProgressSnapshot = StoreTimerSnapshot.from(timer);
         }
-        if (useSyncLock) {
-            buildIndexAsyncFuture = runner
-                    .runAsync(context -> openRecordStore(context).thenApply(store -> IndexingSubspaces.indexBuildLockSubspace(store, index)),
-                            common.indexLogMessageKeyValues("IndexingBase::indexBuildLockSubspace"))
-                    .thenCompose(lockSubspace -> runner.startSynchronizedSessionAsync(lockSubspace, common.config.getLeaseLengthMillis()))
-                    .thenCompose(synchronizedRunner -> {
-                        message.addKeyAndValue(LogMessageKeys.SESSION_ID, synchronizedRunner.getSessionId());
-                        return runWithSynchronizedRunnerAndEndSession(synchronizedRunner,
-                                () -> handleStateAndDoBuildIndexAsync(markReadable, message));
-                    });
-        } else {
-            message.addKeyAndValue(LogMessageKeys.SESSION_ID, "none");
-            common.setSynchronizedSessionRunner(null);
-            buildIndexAsyncFuture = handleStateAndDoBuildIndexAsync(markReadable, message);
-        }
-        return buildIndexAsyncFuture.whenComplete((vignore, ex) -> {
+        message.addKeyAndValue(LogMessageKeys.SESSION_ID, common.getUuid());
+        return handleStateAndDoBuildIndexAsync(markReadable, message).whenComplete((vignore, ex) -> {
             message.addKeysAndValues(indexingLogMessageKeyValues()) // add these here to pick up state accumulated during build
                     .addKeysAndValues(common.indexLogMessageKeyValues())
                     .addKeyAndValue(LogMessageKeys.TOTAL_MICROS, TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - startNanos));
@@ -193,36 +175,6 @@ public CompletableFuture<Void> buildIndexAsync(boolean markReadable, boolean use
         });
     }
 
-    @SuppressWarnings("PMD.CloseResource")
-    private <T> CompletableFuture<T> runWithSynchronizedRunnerAndEndSession(
-            @Nonnull SynchronizedSessionRunner newSynchronizedRunner, @Nonnull Supplier<CompletableFuture<T>> runnable) {
-        final SynchronizedSessionRunner currentSynchronizedRunner1 = common.getSynchronizedSessionRunner();
-        if (currentSynchronizedRunner1 == null) {
-            common.setSynchronizedSessionRunner(newSynchronizedRunner);
-            return MoreAsyncUtil.composeWhenComplete(runnable.get(), (result, ex) -> {
-                final SynchronizedSessionRunner currentSynchronizedRunner2 = common.getSynchronizedSessionRunner();
-                if (newSynchronizedRunner.equals(currentSynchronizedRunner2)) {
-                    common.setSynchronizedSessionRunner(null);
-                } else {
-                    if (LOGGER.isWarnEnabled()) {
-                        LOGGER.warn(KeyValueLogMessage.build("synchronizedSessionRunner was modified during the run",
-                                LogMessageKeys.SESSION_ID, newSynchronizedRunner.getSessionId(),
-                                LogMessageKeys.INDEXER_SESSION_ID, currentSynchronizedRunner2 == null ? null : currentSynchronizedRunner2.getSessionId())
-                                .addKeysAndValues(common.indexLogMessageKeyValues())
-                                .toString());
-                    }
-                }
-                return newSynchronizedRunner.endSessionAsync();
-            }, getRunner().getDatabase()::mapAsyncToSyncException);
-        } else {
-            return newSynchronizedRunner.endSessionAsync().thenApply(vignore -> {
-                throw new RecordCoreException("another synchronized session is running on the indexer",
-                        LogMessageKeys.SESSION_ID, newSynchronizedRunner.getSessionId(),
-                        LogMessageKeys.INDEXER_SESSION_ID, currentSynchronizedRunner1.getSessionId());
-            });
-        }
-    }
-
     abstract List<Object> indexingLogMessageKeyValues();
 
     @Nonnull
@@ -314,7 +266,7 @@ private CompletableFuture<Void> markIndexesWriteOnly(boolean continueBuild, FDBR
     @Nonnull
     public CompletableFuture<Boolean> markReadableIfBuilt() {
         AtomicBoolean allReadable = new AtomicBoolean(true);
-        return common.getNonSynchronizedRunner().runAsync(context -> openRecordStore(context).thenCompose(store ->
+        return common.getRunner().runAsync(context -> openRecordStore(context).thenCompose(store ->
             forEachTargetIndex(index -> {
                 if (store.isIndexReadable(index)) {
                     return AsyncUtil.DONE;
@@ -335,6 +287,7 @@ public CompletableFuture<Boolean> markReadableIfBuilt() {
         ).thenApply(ignore -> allReadable.get()), common.indexLogMessageKeyValues("IndexingBase::markReadableIfBuilt"));
     }
 
+
     @Nonnull
     public CompletableFuture<Boolean> markIndexReadable(boolean markReadablePlease) {
         if (!markReadablePlease) {
@@ -360,12 +313,16 @@ public CompletableFuture<Boolean> markIndexReadable(boolean markReadablePlease)
     private CompletableFuture<Boolean> markIndexReadableSingleTarget(Index index, AtomicBoolean anythingChanged,
                                                                      AtomicReference<RuntimeException> runtimeExceptionAtomicReference) {
         // An extension function to reduce markIndexReadable's complexity
-        return common.getNonSynchronizedRunner().runAsync(context ->
+        return common.getRunner().runAsync(context ->
                 common.getRecordStoreBuilder().copyBuilder().setContext(context).openAsync()
-                        .thenCompose(store ->
-                                policy.shouldAllowUniquePendingState(store) ?
-                                store.markIndexReadableOrUniquePending(index) :
-                                store.markIndexReadable(index))
+                        .thenCompose(store -> {
+                            if (heartbeat != null) {
+                                heartbeat.clearHeartbeat(store, index);
+                            }
+                            return policy.shouldAllowUniquePendingState(store) ?
+                                   store.markIndexReadableOrUniquePending(index) :
+                                   store.markIndexReadable(index);
+                        })
         ).handle((changed, ex) -> {
             if (ex == null) {
                 if (Boolean.TRUE.equals(changed)) {
@@ -388,6 +345,7 @@ public void enforceStampOverwrite() {
     private CompletableFuture<Void> setIndexingTypeOrThrow(FDBRecordStore store, boolean continuedBuild) {
         // continuedBuild is set if this session isn't a continuation of a previous indexing
         IndexBuildProto.IndexBuildIndexingStamp indexingTypeStamp = getIndexingTypeStamp(store);
+        heartbeat = new IndexingHeartbeat(common.getUuid(), indexingTypeStamp.getMethod());
 
         return forEachTargetIndex(index -> setIndexingTypeOrThrow(store, continuedBuild, index, indexingTypeStamp));
     }
@@ -428,21 +386,6 @@ private CompletableFuture<Void> setIndexingTypeOrThrow(FDBRecordStore store, boo
                     }
                     // Here: check if type conversion is allowed
                     if (continuedBuild && shouldAllowTypeConversionContinue(newStamp, savedStamp)) {
-                        // Special case: partly built by another indexing method, but may be continued with the current one
-                        if (savedStamp.getMethod().equals(IndexBuildProto.IndexBuildIndexingStamp.Method.MULTI_TARGET_BY_RECORDS)) {
-                            // Here: throw an exception if there is an active multi-target session that includes this index
-                            final String otherPrimaryIndexName = savedStamp.getTargetIndex(0);
-                            if (!otherPrimaryIndexName.equals(common.getPrimaryIndex().getName())) {
-                                // Note: For protection, avoid breaking an active multi-target session. This leads to a certain
-                                // inconsistency for buildIndex that is called with a false `useSyncLock` - sync lock will be
-                                // checked during a method conversion, but not during a simple "same method" continue.
-                                return throwIfSyncedLock(otherPrimaryIndexName, store, newStamp, savedStamp)
-                                        .thenCompose(ignore -> {
-                                            store.saveIndexingTypeStamp(index, newStamp);
-                                            return AsyncUtil.DONE;
-                                        });
-                            }
-                        }
                         store.saveIndexingTypeStamp(index, newStamp);
                         return AsyncUtil.DONE;
                     }
@@ -476,23 +419,6 @@ private static IndexBuildProto.IndexBuildIndexingStamp blocklessStampOf(IndexBui
                 .build();
     }
 
-    CompletableFuture<Void> throwIfSyncedLock(String otherIndexName, FDBRecordStore store, IndexBuildProto.IndexBuildIndexingStamp newStamp, IndexBuildProto.IndexBuildIndexingStamp savedStamp) {
-        final Index otherIndex = store.getRecordMetaData().getIndex(otherIndexName);
-        final Subspace mainLockSubspace = IndexingSubspaces.indexBuildLockSubspace(store, otherIndex);
-        return SynchronizedSession.checkActiveSessionExists(store.ensureContextActive(), mainLockSubspace)
-                        .thenApply(hasActiveSession -> {
-                            if (Boolean.TRUE.equals(hasActiveSession)) {
-                                throw new SynchronizedSessionLockedException("Failed to takeover indexing while part of a multi-target with an existing session in progress")
-                                        .addLogInfo(LogMessageKeys.SUBSPACE, mainLockSubspace)
-                                        .addLogInfo(LogMessageKeys.PRIMARY_INDEX, otherIndexName)
-                                        .addLogInfo(LogMessageKeys.EXPECTED, PartlyBuiltException.stampToString(newStamp))
-                                        .addLogInfo(LogMessageKeys.ACTUAL, PartlyBuiltException.stampToString(savedStamp));
-                            }
-                            return null;
-                        });
-
-    }
-
     @Nonnull
     private CompletableFuture<Void> throwAsByRecordsUnlessNoRecordWasScanned(boolean noRecordScanned,
                                                                              FDBRecordStore store,
@@ -885,21 +811,43 @@ private CompletableFuture<Boolean> hadTransactionReachedLimits(FDBRecordStore st
     }
 
     private CompletableFuture<Void> validateTypeStamp(@Nonnull FDBRecordStore store) {
+        if (shouldValidate()) {
+            // check other heartbeats (if exclusive) & typestamp
+            final IndexBuildProto.IndexBuildIndexingStamp expectedTypeStamp = getIndexingTypeStamp(store);
+            return forEachTargetIndex(index -> CompletableFuture.allOf(
+                    updateHeartbeat(true, store, index),
+                    store.loadIndexingTypeStampAsync(index)
+                            .thenAccept(typeStamp -> validateTypeStamp(typeStamp, expectedTypeStamp, index))
+                    ));
+        } else {
+            // update only
+            return forEachTargetIndex(index -> updateHeartbeat(false, store, index));
+        }
+    }
+
+    private CompletableFuture<Void> updateHeartbeat(boolean validate, FDBRecordStore store, Index index) {
+        if (heartbeat != null) {
+            if (validate) {
+                return heartbeat.checkAndUpdateHeartbeat(store, index);
+            }
+            heartbeat.updateHeartbeat(store, index);
+        }
+        return AsyncUtil.DONE;
+    }
+
+    private boolean shouldValidate() {
         final long minimalInterval = policy.getCheckIndexingMethodFrequencyMilliseconds();
         if (minimalInterval < 0 || isScrubber) {
-            return AsyncUtil.DONE;
+            return false;
         }
         if (minimalInterval > 0) {
             final long now = System.currentTimeMillis();
             if (now < lastTypeStampCheckMillis + minimalInterval) {
-                return AsyncUtil.DONE;
+                return false;
             }
             lastTypeStampCheckMillis = now;
         }
-        final IndexBuildProto.IndexBuildIndexingStamp expectedTypeStamp = getIndexingTypeStamp(store);
-        return forEachTargetIndex(index ->
-                store.loadIndexingTypeStampAsync(index)
-                        .thenAccept(typeStamp -> validateTypeStamp(typeStamp, expectedTypeStamp, index)));
+        return true;
     }
 
     private void validateTypeStamp(final IndexBuildProto.IndexBuildIndexingStamp typeStamp,

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingCommon.java

@@ -27,7 +27,6 @@
 import com.apple.foundationdb.record.metadata.Index;
 import com.apple.foundationdb.record.metadata.MetaDataException;
 import com.apple.foundationdb.record.metadata.RecordType;
-import com.apple.foundationdb.record.provider.foundationdb.synchronizedsession.SynchronizedSessionRunner;
 import com.apple.foundationdb.record.query.plan.synthetic.SyntheticRecordPlanner;
 import com.apple.foundationdb.tuple.Tuple;
 
@@ -50,10 +49,10 @@
 
 @API(API.Status.INTERNAL)
 public class IndexingCommon {
+    // TODO? get uuid from caller to allow lock takeover
     private final UUID uuid = UUID.randomUUID();
 
     @Nonnull private final FDBDatabaseRunner runner;
-    @Nullable private SynchronizedSessionRunner synchronizedSessionRunner = null;
 
     @Nonnull private final FDBRecordStore.Builder recordStoreBuilder;
     @Nonnull private final AtomicLong totalRecordsScanned;
@@ -176,11 +175,6 @@ private void logIf(boolean condition, List<Object> list, @Nonnull Object... a) {
 
     @Nonnull
     public FDBDatabaseRunner getRunner() {
-        return synchronizedSessionRunner == null ? runner : synchronizedSessionRunner;
-    }
-
-    @Nonnull
-    public FDBDatabaseRunner getNonSynchronizedRunner() {
         return runner;
     }
 
@@ -258,15 +252,6 @@ public FDBRecordStore.Builder getRecordStoreBuilder() {
         return recordStoreBuilder;
     }
 
-    @Nullable
-    public SynchronizedSessionRunner getSynchronizedSessionRunner() {
-        return synchronizedSessionRunner;
-    }
-
-    public void setSynchronizedSessionRunner(@Nullable final SynchronizedSessionRunner synchronizedSessionRunner) {
-        this.synchronizedSessionRunner = synchronizedSessionRunner;
-    }
-
     @Nonnull
     public AtomicLong getTotalRecordsScanned() {
         return totalRecordsScanned;
@@ -287,8 +272,5 @@ public boolean loadConfig() {
 
     public void close() {
         runner.close();
-        if (synchronizedSessionRunner != null) {
-            synchronizedSessionRunner.close();
-        }
     }
 }