diff --git a/CHANGELOG.md b/CHANGELOG.md index 5606977..88a4efb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,9 +4,10 @@ All notable changes to this project are documented here. The format follows Keep ## [Unreleased] -Requires QQQ 4.1.0-RC.8 or later (central health checks, help bundles and the spec-managed UI API extension contract). +Requires QQQ 4.1.0-RC.8 or later (central health checks, help bundles and the spec-managed UI API extension contract), and a QQQ 4.1 build with `ClusterCaches` for multi-node deployments (the pom builds against 4.1.0-SNAPSHOT until the next QQQ RC; a release must pin that RC). ### Fixed +- Multi-node deployments: a Quick Search config change (an index row, its fields or the settings row, from the admin UI or a config process) now clears the cached config on every node, not only the node that made the change. The runtime's config cache (the config store snapshot and settings, and the per-table enabled flags) is registered with QQQ `ClusterCaches` as `qbit-quick-search.config`, and `QuickSearchRuntime.invalidateConfig()` publishes a cluster invalidation. The 30 s and 60 s TTLs remain as the safety net. `QuickSearchFieldRules`' warn-once table set is registered as node-local. `QuickSearchClusterCachesTest` audits the package for unregistered static caches and covers a remote invalidation. - Indexing bookkeeping is no longer record-audited: the `quick_search_index` runtime fields written by basepull, full reindex and real-time indexing (status, last basepull time, document count, last run, real-time error count) and the settings row's priming outcome (`lastPrimedAt`, summary) are written with `omitDmlAudit`. On a production host these wrote about 470 "Record was Edited during process: Update search index" audit rows an hour. Admin edits to an index, its fields or the settings are still audited. - Real-time indexing failures now increase the table's error count by the number of failed events. Successful basepulls and full reindexes reset the count. Counter update failures are logged without failing the source write. diff --git a/pom.xml b/pom.xml index 6aaa9bf..be02741 100644 --- a/pom.xml +++ b/pom.xml @@ -283,7 +283,7 @@ true - 4.1.0-RC.8 + 4.1.0-RC.14 diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java index ad822ec..bc758fc 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java @@ -23,6 +23,8 @@ import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.WeakHashMap; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; @@ -34,6 +36,7 @@ import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient; import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher; import com.kingsrook.qbits.quicksearch.publisher.SynchronousIndexEventPublisher; import com.kingsrook.qqq.backend.core.actions.tables.QueryAction; +import com.kingsrook.qqq.backend.core.cluster.ClusterCaches; import com.kingsrook.qqq.backend.core.context.QContext; import com.kingsrook.qqq.backend.core.exceptions.QException; import com.kingsrook.qqq.backend.core.logging.QLogger; @@ -70,6 +73,17 @@ public class QuickSearchRuntime private static final long ENABLED_CACHE_TTL_MILLIS = 60_000L; + /******************************************************************************* + ** Cluster cache name for every runtime's config cache (the config store's + ** snapshot and settings, and the enabled flags). A config row edited on + ** one node clears it on every node; the TTLs remain the safety net. + *******************************************************************************/ + public static final String CONFIG_CACHE_NAME = "qbit-quick-search.config"; + + private static final Set RUNTIMES = ClusterCaches.shared(CONFIG_CACHE_NAME, + Collections.synchronizedSet(Collections.newSetFromMap(new WeakHashMap<>())), + QuickSearchRuntime::invalidateAllLocal, null); + public static final long RECORD_SEARCH_BACKOFF_MILLIS = 30_000L; private final StatusCache statusCache = new StatusCache(); @@ -128,6 +142,7 @@ public class QuickSearchRuntime { this.config = config; this.configStore = new QuickSearchConfigStore(this); + RUNTIMES.add(this); } @@ -261,9 +276,21 @@ public class QuickSearchRuntime /******************************************************************************* - ** Drop every cached config (a config row changed). + ** Drop every cached config (a config row changed): on this node now, and + ** on the other nodes through the cluster bus. *******************************************************************************/ public void invalidateConfig() + { + invalidateConfigLocal(); + ClusterCaches.invalidate(CONFIG_CACHE_NAME); + } + + + + /******************************************************************************* + ** Drop this runtime's cached config on this node only. + *******************************************************************************/ + void invalidateConfigLocal() { configStore.invalidate(); enabledCache.clear(); @@ -271,6 +298,25 @@ public class QuickSearchRuntime + /******************************************************************************* + ** Drop the cached config of every live runtime on this node (a cluster + ** invalidation arrived). + *******************************************************************************/ + private static void invalidateAllLocal() + { + List runtimes; + synchronized(RUNTIMES) + { + runtimes = new ArrayList<>(RUNTIMES); + } + for(QuickSearchRuntime runtime : runtimes) + { + runtime.invalidateConfigLocal(); + } + } + + + /******************************************************************************* ** Return the OpenSearch client, building it on first use. Building the ** client does not contact the cluster; {@link #ensureIndexReady()} does. @@ -422,6 +468,16 @@ public class QuickSearchRuntime + /******************************************************************************* + ** How many enabled flags are cached (for tests). + *******************************************************************************/ + int getCachedEnabledFlagCount() + { + return (enabledCache.size()); + } + + + /******************************************************************************* ** Forget cached enabled flags (after an operator toggles a table). *******************************************************************************/ @@ -741,6 +797,7 @@ public class QuickSearchRuntime return; } closed = true; + RUNTIMES.remove(this); statusCache.close(); if(publisher != null) diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/config/QuickSearchConfigStore.java b/src/main/java/com/kingsrook/qbits/quicksearch/config/QuickSearchConfigStore.java index 843dcef..2b8a91b 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/config/QuickSearchConfigStore.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/config/QuickSearchConfigStore.java @@ -60,8 +60,9 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair; ** before 1.1.0-RC.3. ** ** Reads are cached for CACHE_TTL_MILLIS and dropped at once by the config - ** tables' customizers when a row changes on this server; other servers pick - ** the change up when their cache expires. + ** tables' customizers when a row changes, on every server: the runtime + ** publishes a cluster cache invalidation (QuickSearchRuntime.CONFIG_CACHE_NAME). + ** The TTL remains the safety net when the cluster bus is down. *******************************************************************************/ public class QuickSearchConfigStore { diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/config/QuickSearchFieldRules.java b/src/main/java/com/kingsrook/qbits/quicksearch/config/QuickSearchFieldRules.java index a043e25..974a4d3 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/config/QuickSearchFieldRules.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/config/QuickSearchFieldRules.java @@ -29,6 +29,7 @@ import com.kingsrook.qqq.backend.core.actions.ai.AiDataClassifications; import com.kingsrook.qqq.backend.core.actions.ai.DataClassification; import com.kingsrook.qqq.backend.core.actions.permissions.PermissionsHelper; import com.kingsrook.qqq.backend.core.actions.permissions.TablePermissionSubType; +import com.kingsrook.qqq.backend.core.cluster.ClusterCaches; import com.kingsrook.qqq.backend.core.context.QContext; import com.kingsrook.qqq.backend.core.logging.QLogger; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput; @@ -59,7 +60,8 @@ import com.kingsrook.qbits.quicksearch.model.QuickSearchFieldRole; public final class QuickSearchFieldRules { private static final QLogger LOG = QLogger.getLogger(QuickSearchFieldRules.class); - private static final WeakHashMap WARNED_TABLES = new WeakHashMap<>(); + private static final WeakHashMap WARNED_TABLES = ClusterCaches.nodeLocal(new WeakHashMap<>(), + "warn-once log de-duplication for this node's table metadata objects"); public static final int MAX_DEFAULT_FIELDS = 20; public static final int TITLE_WEIGHT = 3; diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchClusterCachesTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchClusterCachesTest.java new file mode 100644 index 0000000..12f09a9 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchClusterCachesTest.java @@ -0,0 +1,193 @@ +/* + * Copyright 2026 Kingsrook, LLC + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.kingsrook.qbits.quicksearch; + + +import java.util.EnumSet; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.function.Consumer; +import com.kingsrook.qqq.backend.core.cluster.ClusterBus; +import com.kingsrook.qqq.backend.core.cluster.ClusterBusInterface; +import com.kingsrook.qqq.backend.core.cluster.ClusterBusMetaData; +import com.kingsrook.qqq.backend.core.cluster.ClusterCacheAudit; +import com.kingsrook.qqq.backend.core.cluster.ClusterEvent; +import com.kingsrook.qqq.backend.core.cluster.QNodeRoles; +import com.kingsrook.qqq.backend.core.context.QContext; +import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference; +import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex; +import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexField; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import static org.assertj.core.api.Assertions.assertThat; + + +/******************************************************************************* + ** The Quick Search caches are correct on several nodes: every static cache is + ** registered with ClusterCaches, and a config change on one node clears the + ** config cache on the others. + *******************************************************************************/ +class QuickSearchClusterCachesTest extends BaseQuickSearchTest +{ + /** the "wire": events every node's transport sees */ + static final List> WIRE = new CopyOnWriteArrayList<>(); + + + + /******************************************************************************* + ** A transport over the static WIRE, so a "remote" publish reaches us. + *******************************************************************************/ + public static class WireBus implements ClusterBusInterface + { + /******************************************************************************* + ** + *******************************************************************************/ + @Override + public void publish(ClusterEvent event) + { + WIRE.forEach(listener -> listener.accept(event)); + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + @Override + public AutoCloseable subscribe(Consumer listener) + { + WIRE.add(listener); + return (() -> WIRE.remove(listener)); + } + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + @AfterEach + void afterEach() + { + ClusterBus.close(QContext.getQInstance()); + WIRE.clear(); + QNodeRoles.set(null); + } + + + + /******************************************************************************* + ** Connect the current instance to the wire as node-a. + *******************************************************************************/ + private static void connectAsNodeA() + { + QInstance instance = QContext.getQInstance(); + instance.withSupplementalMetaData(new ClusterBusMetaData().withBus(new QCodeReference(WireBus.class))); + QNodeRoles.set(new QNodeRoles.Settings(EnumSet.allOf(QNodeRoles.Role.class), true, true, true, null, "node-a")); + ClusterBus.connect(instance); + } + + + + /******************************************************************************* + ** A runtime with one code-declared table. + *******************************************************************************/ + private static QuickSearchRuntime newRuntime() + { + /////////////////////////////////////////////////////////////////////// + // without the field table the config comes from code, so the enabled // + // flag is read from the index row and cached // + /////////////////////////////////////////////////////////////////////// + QContext.getQInstance().getTables().remove(QuickSearchIndexField.TABLE_NAME); + QContext.getQInstance().getTable(QuickSearchIndex.TABLE_NAME).getAssociations().clear(); + + QuickSearchRuntime runtime = new QuickSearchRuntime(new QuickSearchQBitConfig().withBackendName(TEST_BACKEND_NAME).withOpensearchHost("localhost").withOpensearchPort(9200).withOpensearchIndexName("t")); + runtime.setDiscoveredTables(List.of(new QuickSearchableTableConfig().withTableName(TEST_ENTITY_TABLE).withPrimaryKeyField("id").withSearchableFields(List.of("name")))); + return (runtime); + } + + + + /******************************************************************************* + ** Every static Memoization in the QBit is registered (shared or node-local). + *******************************************************************************/ + @Test + void testNoUnregisteredStaticCaches() throws Exception + { + assertThat(ClusterCacheAudit.unregisteredStaticCaches("com.kingsrook.qbits.quicksearch")).isEmpty(); + } + + + + /******************************************************************************* + ** Another node's config invalidation clears this node's cached config. + *******************************************************************************/ + @Test + void testRemoteInvalidationClearsConfigCache() + { + connectAsNodeA(); + QuickSearchRuntime runtime = newRuntime(); + try + { + runtime.isTableEnabled(TEST_ENTITY_TABLE); + assertThat(runtime.getCachedEnabledFlagCount()).isEqualTo(1); + + ClusterEvent remote = new ClusterEvent(ClusterEvent.TYPE_MEMO_INVALIDATE, QuickSearchRuntime.CONFIG_CACHE_NAME, null); + remote.setOriginNodeId("node-b"); + WIRE.forEach(listener -> listener.accept(remote)); + + assertThat(runtime.getCachedEnabledFlagCount()).isZero(); + } + finally + { + runtime.close(); + } + } + + + + /******************************************************************************* + ** A config change here clears the cache here and tells the other nodes. + *******************************************************************************/ + @Test + void testInvalidateConfigPublishesToOtherNodes() + { + connectAsNodeA(); + List sent = new CopyOnWriteArrayList<>(); + WIRE.add(sent::add); + + QuickSearchRuntime runtime = newRuntime(); + try + { + runtime.isTableEnabled(TEST_ENTITY_TABLE); + runtime.invalidateConfig(); + + assertThat(runtime.getCachedEnabledFlagCount()).isZero(); + assertThat(sent).anySatisfy(event -> + { + assertThat(event.getType()).isEqualTo(ClusterEvent.TYPE_MEMO_INVALIDATE); + assertThat(event.getName()).isEqualTo(QuickSearchRuntime.CONFIG_CACHE_NAME); + assertThat(event.getOriginNodeId()).isEqualTo("node-a"); + }); + } + finally + { + runtime.close(); + } + } +}