From ce5a14870bddbb71c6158c9afe7707f035aafc77 Mon Sep 17 00:00:00 2001 From: James Maes Date: Tue, 6 Oct 2026 18:03:15 -0500 Subject: [PATCH 1/3] fix(#14): replay deletes captured during a full reindex While a full reindex rebuilds, the listener records each delete as an AWAITING_REINDEX failed event (the index row is REBUILDING), and the reindex applies them through the alias after the swap. Closes #14 --- README.md | 4 +- .../QuickSearchRecordChangeListener.java | 164 ++++++++++-- .../model/QuickSearchFailedEvent.java | 6 + .../processes/AbstractIndexingStep.java | 6 +- .../processes/FullReindexStep.java | 171 +++++++++++- .../processes/ReconcileIndexStep.java | 8 +- ...ullReindexDeleteReplayIntegrationTest.java | 244 ++++++++++++++++++ .../QuickSearchRecordChangeListenerTest.java | 95 +++++++ .../processes/BasepullIndexStepTest.java | 46 ++++ .../processes/FullReindexStepTest.java | 116 +++++++++ .../processes/ReconcileIndexStepTest.java | 24 ++ 11 files changed, 857 insertions(+), 27 deletions(-) create mode 100644 src/test/java/com/kingsrook/qbits/quicksearch/FullReindexDeleteReplayIntegrationTest.java diff --git a/README.md b/README.md index b8499f8..1558e1d 100644 --- a/README.md +++ b/README.md @@ -167,9 +167,9 @@ Without `schedulerName` the processes exist but nothing runs them; the QBit logs ## Operational tables -- `quickSearchIndex`: one row per table with `enabled` (honoured by indexing and search, cached for one minute), `basepullIntervalMinutes`, `lastBasepullTime`, `lastReconcileTime`, `lastFullReindexTime`, `lastRunStatus`, `lastErrorMessage`, `recordCount`, `documentCount`, `realTimeErrorCount`, `status` (`ACTIVE` or `NEEDS_REINDEX` when the configured fields changed since the row was created). Unique on `tableName`. +- `quickSearchIndex`: one row per table with `enabled` (honoured by indexing and search, cached for one minute), `basepullIntervalMinutes`, `lastBasepullTime`, `lastReconcileTime`, `lastFullReindexTime`, `lastRunStatus`, `lastErrorMessage`, `recordCount`, `documentCount`, `realTimeErrorCount`, `status` (`ACTIVE`, `NEEDS_REINDEX` when the configured fields changed since the row was created, or `REBUILDING` while a full reindex of all tables runs). Unique on `tableName`. - `quickSearchIndexRun`: `runType` (`BASEPULL`, `RECONCILE`, `FULL_REINDEX`), `status`, counts, `errorMessage`. -- `quickSearchFailedEvent`: real-time events that could not be applied; `status` `PENDING` or `EXHAUSTED` after 10 attempts. +- `quickSearchFailedEvent`: real-time events that could not be applied; `status` `PENDING` or `EXHAUSTED` after 10 attempts. Deletes made while a full reindex of all tables runs are also recorded here with status `AWAITING_REINDEX`: the reindex applies them to the new index after the alias swap (basepull does not replay them). If a full reindex is killed mid-run, its tables stay `REBUILDING` and deletes keep collecting as `AWAITING_REINDEX` until the next full reindex completes. ## Extending diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java b/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java index 127ec78..26f9724 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java @@ -22,6 +22,7 @@ import java.util.ArrayList; import java.util.List; import com.kingsrook.qqq.backend.core.actions.tables.InsertAction; import com.kingsrook.qqq.backend.core.actions.tables.QueryAction; +import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction; import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeEvent; import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeListenerInterface; import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeType; @@ -33,11 +34,13 @@ import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperat import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput; +import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput; import com.kingsrook.qqq.backend.core.model.data.QRecord; import com.kingsrook.qqq.backend.core.utils.CollectionUtils; import com.kingsrook.qbits.quicksearch.QuickSearchRuntime; import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig; import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent; +import com.kingsrook.qbits.quicksearch.processes.AbstractIndexingStep; import com.kingsrook.qbits.quicksearch.publisher.IndexEvent; import com.kingsrook.qbits.quicksearch.publisher.IndexEventAction; import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher; @@ -57,6 +60,11 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair; ** This listener never fails the host's write. When publishing fails, each ** affected (table, record, action) is written to quickSearchFailedEvent for ** the scheduled basepull to replay. + ** + ** While a full reindex rebuilds into a new physical index, the alias still + ** points at the old one, so a published delete never reaches the new index. + ** Such deletes are also written to quickSearchFailedEvent with status + ** AWAITING_REINDEX; the reindex applies them after the alias swap. *******************************************************************************/ public class QuickSearchRecordChangeListener implements RecordChangeListenerInterface { @@ -129,9 +137,21 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte /******************************************************************************* - ** Publish, recording any failure durably. Never throws. + ** Publish, recording any failure durably, then capture the deletes if a full + ** reindex is rebuilding the table. Never throws. *******************************************************************************/ static void publish(QuickSearchRuntime runtime, List indexEvents, List deleteEvents) + { + publishOrRecordFailures(runtime, indexEvents, deleteEvents); + captureDeletesDuringRebuild(runtime, deleteEvents); + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + private static void publishOrRecordFailures(QuickSearchRuntime runtime, List indexEvents, List deleteEvents) { IndexEventPublisher publisher; try @@ -188,29 +208,13 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte try { - String failedEventTable = runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME); - if(QContext.getQInstance() == null || QContext.getQInstance().getTable(failedEventTable) == null) + if(!hasTable(runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME))) { LOG.error("quickSearchFailedEvent table is not in the QInstance; failed events are lost", logPair("eventCount", events.size())); return; } - List rows = new ArrayList<>(); - for(IndexEvent event : events) - { - rows.add(new QRecord() - .withValue("tableName", event.getTableName()) - .withValue("recordId", event.getRecordId()) - .withValue("action", event.getAction() == null ? null : event.getAction().name()) - .withValue("errorMessage", truncate(message, 4000)) - .withValue("attempts", 0) - .withValue("status", QuickSearchFailedEvent.STATUS_PENDING)); - } - - InsertInput insertInput = new InsertInput(); - insertInput.setTableName(failedEventTable); - insertInput.setRecords(rows); - new InsertAction().execute(insertInput); + insertFailedEventRows(runtime, events, message, QuickSearchFailedEvent.STATUS_PENDING); } catch(Exception e) { @@ -220,6 +224,128 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte + /******************************************************************************* + ** Record deletes made while a full reindex is rebuilding the table, so the + ** reindex can apply them to the new physical index after the alias swap. + ** basepull does not replay AWAITING_REINDEX rows, so it cannot consume one + ** against the old index before the swap. + ** + ** The table's status is read from the database on every call (not cached) + ** so a reindex running on another node is seen. This runs after the delete + ** committed; the reindex marks the table REBUILDING before reading it, so a + ** delete of a record the reindex may have read always sees that status. + ** Never throws. + *******************************************************************************/ + static void captureDeletesDuringRebuild(QuickSearchRuntime runtime, List deleteEvents) + { + if(deleteEvents.isEmpty()) + { + return; + } + + String tableName = deleteEvents.get(0).getTableName(); + try + { + if(!hasTable(runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME)) || !isRebuilding(runtime, tableName)) + { + return; + } + + List inserted = insertFailedEventRows(runtime, deleteEvents, "deleted during a full reindex", QuickSearchFailedEvent.STATUS_AWAITING_REINDEX); + handOverIfRebuildFinished(runtime, tableName, inserted); + } + catch(Exception e) + { + LOG.error("Could not capture deletes made during a full reindex; the next full reindex or reconcile removes them", e, + logPair("tableName", tableName), logPair("eventCount", deleteEvents.size())); + } + } + + + + /******************************************************************************* + ** The reindex may have finished, and drained its captured deletes, between + ** the status read and the insert. Read the status again: if the table is no + ** longer REBUILDING, make the rows PENDING so basepull replays them against + ** the current index (deletes are idempotent). If it still is, the reindex + ** has not drained yet and will see these rows. + *******************************************************************************/ + static void handOverIfRebuildFinished(QuickSearchRuntime runtime, String tableName, List inserted) throws QException + { + if(inserted.isEmpty() || isRebuilding(runtime, tableName)) + { + return; + } + + List updates = new ArrayList<>(); + for(QRecord row : inserted) + { + updates.add(new QRecord().withValue("id", row.getValue("id")).withValue("status", QuickSearchFailedEvent.STATUS_PENDING)); + } + + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME)); + updateInput.setRecords(updates); + new UpdateAction().execute(updateInput); + } + + + + /******************************************************************************* + ** Whether the table's quickSearchIndex row is REBUILDING. + *******************************************************************************/ + private static boolean isRebuilding(QuickSearchRuntime runtime, String tableName) throws QException + { + String indexTable = runtime.getConfig().getQuickSearchIndexTableName(); + if(!hasTable(indexTable)) + { + return (false); + } + + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(indexTable); + queryInput.setFilter(new QQueryFilter(new QFilterCriteria("tableName", QCriteriaOperator.EQUALS, tableName)).withLimit(1)); + List rows = new QueryAction().execute(queryInput).getRecords(); + return (CollectionUtils.nullSafeHasContents(rows) && AbstractIndexingStep.STATUS_REBUILDING.equals(rows.get(0).getValueString("status"))); + } + + + + /******************************************************************************* + ** Insert one quickSearchFailedEvent row per event; returns the inserted rows. + *******************************************************************************/ + static List insertFailedEventRows(QuickSearchRuntime runtime, List events, String message, String status) throws QException + { + List rows = new ArrayList<>(); + for(IndexEvent event : events) + { + rows.add(new QRecord() + .withValue("tableName", event.getTableName()) + .withValue("recordId", event.getRecordId()) + .withValue("action", event.getAction() == null ? null : event.getAction().name()) + .withValue("errorMessage", truncate(message, 4000)) + .withValue("attempts", 0) + .withValue("status", status)); + } + + InsertInput insertInput = new InsertInput(); + insertInput.setTableName(runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME)); + insertInput.setRecords(rows); + return (CollectionUtils.nonNullList(new InsertAction().execute(insertInput).getRecords())); + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + private static boolean hasTable(String tableName) + { + return (QContext.getQInstance() != null && QContext.getQInstance().getTable(tableName) != null); + } + + + /******************************************************************************* ** *******************************************************************************/ diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/model/QuickSearchFailedEvent.java b/src/main/java/com/kingsrook/qbits/quicksearch/model/QuickSearchFailedEvent.java index ff84817..6c37272 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/model/QuickSearchFailedEvent.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/model/QuickSearchFailedEvent.java @@ -34,6 +34,12 @@ public class QuickSearchFailedEvent extends QRecordEntity public static final String STATUS_PENDING = "PENDING"; public static final String STATUS_EXHAUSTED = "EXHAUSTED"; + /////////////////////////////////////////////////////////////////////////// + // a delete made while a full reindex rebuilds into a new physical index; // + // the reindex applies it after the swap, and basepull does not replay it // + /////////////////////////////////////////////////////////////////////////// + public static final String STATUS_AWAITING_REINDEX = "AWAITING_REINDEX"; + @QField() private Integer id; diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java b/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java index 5909048..00fb387 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java @@ -65,6 +65,7 @@ public abstract class AbstractIndexingStep implements BackendStep public static final String STATUS_ACTIVE = "ACTIVE"; public static final String STATUS_NEEDS_REINDEX = "NEEDS_REINDEX"; + public static final String STATUS_REBUILDING = "REBUILDING"; public static final String RUN_COMPLETED = "COMPLETED"; public static final String RUN_FAILED = "FAILED"; public static final String RUN_RUNNING = "RUNNING"; @@ -219,6 +220,8 @@ public abstract class AbstractIndexingStep implements BackendStep /******************************************************************************* ** Compare the configured fields with what the row was created with; mark ** the row NEEDS_REINDEX when they differ. Returns true when drift was found. + ** A REBUILDING row is left alone: the full reindex in progress records the + ** current fields, and the status tells the listener to capture deletes. *******************************************************************************/ protected boolean detectDrift(QRecord row, QuickSearchableTableConfig tableConfig) throws QException { @@ -229,7 +232,8 @@ public abstract class AbstractIndexingStep implements BackendStep String current = buildSearchableFieldsJson(tableConfig); String recorded = row.getValueString("searchableFieldsJson"); - if(recorded == null || recorded.equals(current) || STATUS_NEEDS_REINDEX.equals(row.getValueString("status"))) + String status = row.getValueString("status"); + if(recorded == null || recorded.equals(current) || STATUS_NEEDS_REINDEX.equals(status) || STATUS_REBUILDING.equals(status)) { return (recorded != null && !recorded.equals(current)); } diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java b/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java index 48f0430..185e736 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java @@ -24,13 +24,28 @@ import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction; +import com.kingsrook.qqq.backend.core.actions.tables.QueryAction; +import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction; import com.kingsrook.qqq.backend.core.exceptions.QException; import com.kingsrook.qqq.backend.core.logging.QLogger; import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput; import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput; +import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput; +import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator; +import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria; +import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy; +import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter; +import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput; +import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput; import com.kingsrook.qqq.backend.core.model.data.QRecord; +import com.kingsrook.qqq.backend.core.utils.CollectionUtils; +import com.kingsrook.qbits.quicksearch.QuickSearchQBitConfig; import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig; +import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent; import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun; +import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult; +import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument; import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient; import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair; @@ -45,6 +60,14 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair; ** created by an older release (outdated mapping) is upgraded. On failure the ** new physical index is deleted and nothing changes. ** + ** While the new index is built, real-time writes still go through the alias + ** to the old one. Updates are caught by the next basepull (lastBasepullTime + ** is set to each table's start). Deletes are caught here: each table's + ** quickSearchIndex row is REBUILDING for the whole rebuild, which makes the + ** record change listener (on any node) also record each delete as an + ** AWAITING_REINDEX failed event. Once the rebuild's own writes are done and + ** the alias is swapped, those deletes are applied through the alias. + ** ** With a "tableName" input, that table is rebuilt in place with the reconcile ** algorithm (index everything, then delete what was not re-indexed). *******************************************************************************/ @@ -111,15 +134,18 @@ public class FullReindexStep extends AbstractIndexingStep client.createPhysicalIndex(physicalIndex); LOG.info("Rebuilding Quick Search index", logPair("physicalIndex", physicalIndex), logPair("tables", tables.size())); - Map runs = new LinkedHashMap<>(); - Map counts = new LinkedHashMap<>(); - Map startTimes = new LinkedHashMap<>(); - Integer indexed = 0; + Map runs = new LinkedHashMap<>(); + Map counts = new LinkedHashMap<>(); + Map startTimes = new LinkedHashMap<>(); + Map priorStatus = new LinkedHashMap<>(); + Integer indexed = 0; boolean swapped = false; try { + markRebuilding(tables, priorStatus); + for(QuickSearchableTableConfig tableConfig : tables) { String tableName = tableConfig.getTableName(); @@ -154,10 +180,13 @@ public class FullReindexStep extends AbstractIndexingStep values.put("searchableFieldsJson", buildSearchableFieldsJson(tableConfig)); values.put("status", STATUS_ACTIVE); updateIndexRow(runs.get(tableConfig).getQuickSearchIndexId(), values); + priorStatus.remove(tableName); completeRunRecord(runs.get(tableConfig), RUN_COMPLETED, tableCounts.processed(), tableCounts.indexed(), 0, null); } + applyCapturedDeletes(client, tables); + LOG.info("Full reindex complete", logPair("physicalIndex", physicalIndex), logPair("recordsIndexed", indexed)); return (indexed); } @@ -187,6 +216,9 @@ public class FullReindexStep extends AbstractIndexingStep } } + restoreStatus(priorStatus); + applyCapturedDeletes(client, tables); + if(e instanceof QException qException) { throw qException; @@ -197,6 +229,137 @@ public class FullReindexStep extends AbstractIndexingStep + /******************************************************************************* + ** Mark each table's index row REBUILDING before any of them is read, + ** remembering the status it had. A stale REBUILDING left by a crashed run + ** is remembered as ACTIVE. + *******************************************************************************/ + private void markRebuilding(List tables, Map priorStatus) throws QException + { + for(QuickSearchableTableConfig tableConfig : tables) + { + QRecord row = queryIndexRow(tableConfig.getTableName()); + String status = row.getValueString("status"); + priorStatus.put(tableConfig.getTableName(), status == null || STATUS_REBUILDING.equals(status) ? STATUS_ACTIVE : status); + updateIndexRow(row.getValueInteger("id"), Map.of("status", STATUS_REBUILDING)); + } + } + + + + /******************************************************************************* + ** After a failed rebuild, put back the status of each table still marked + ** REBUILDING. Never throws. + *******************************************************************************/ + private void restoreStatus(Map priorStatus) + { + for(Map.Entry entry : priorStatus.entrySet()) + { + try + { + updateIndexRow(queryIndexId(entry.getKey()), Map.of("status", entry.getValue())); + } + catch(Exception e) + { + LOG.warn("Could not restore the index row status after a failed full reindex", e, logPair("tableName", entry.getKey())); + } + } + } + + + + /******************************************************************************* + ** Apply the deletes the listener captured during the rebuild (failed-event + ** rows with status AWAITING_REINDEX) through the alias, after the rebuild's + ** own writes, so an in-flight page cannot bring a deleted record back. + ** Deleting a missing document is harmless. Applied rows are removed; rows + ** the cluster rejects become PENDING for basepull to retry. Called once the + ** tables are no longer REBUILDING, so the listener stops adding rows (one + ** that races the status change hands its own rows to basepull). Never throws. + *******************************************************************************/ + void applyCapturedDeletes(QuickSearchOpenSearchClient client, List tables) + { + List tableNames = new ArrayList<>(); + tables.forEach(tableConfig -> tableNames.add(tableConfig.getTableName())); + if(tableNames.isEmpty()) + { + return; + } + + Integer applied = 0; + try + { + QuickSearchQBitConfig config = getConfig(); + String failedEventTable = config.applyPrefix(QuickSearchFailedEvent.TABLE_NAME); + + while(true) + { + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(failedEventTable); + queryInput.setFilter(new QQueryFilter( + new QFilterCriteria("status", QCriteriaOperator.EQUALS, QuickSearchFailedEvent.STATUS_AWAITING_REINDEX), + new QFilterCriteria("tableName", QCriteriaOperator.IN, tableNames)) + .withOrderBy(new QFilterOrderBy("id", true)) + .withLimit(config.getSourceBatchSize())); + List rows = new QueryAction().execute(queryInput).getRecords(); + if(CollectionUtils.nullSafeIsEmpty(rows)) + { + break; + } + + List documentIds = new ArrayList<>(); + for(QRecord row : rows) + { + documentIds.add(OpenSearchDocument.buildDocumentId(row.getValueString("tableName"), row.getValueString("recordId"))); + } + + String error; + try + { + BulkIndexResult result = client.deleteDocuments(documentIds, config.getBulkBatchSize()); + error = result.getFailureCount() == 0 ? null : String.join("; ", result.getErrors()); + } + catch(Exception e) + { + error = e.getMessage() == null ? e.toString() : e.getMessage(); + } + + if(error == null) + { + DeleteInput deleteInput = new DeleteInput(); + deleteInput.setTableName(failedEventTable); + deleteInput.setPrimaryKeys(rows.stream().map(row -> row.getValue("id")).toList()); + new DeleteAction().execute(deleteInput); + applied = applied + rows.size(); + } + else + { + List updates = new ArrayList<>(); + for(QRecord row : rows) + { + updates.add(new QRecord() + .withValue("id", row.getValue("id")) + .withValue("errorMessage", truncate(error, 4000)) + .withValue("status", QuickSearchFailedEvent.STATUS_PENDING)); + } + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(failedEventTable); + updateInput.setRecords(updates); + new UpdateAction().execute(updateInput); + LOG.warn("Could not apply deletes captured during the full reindex; basepull will retry them", logPair("count", rows.size()), logPair("error", error)); + } + } + } + catch(Exception e) + { + LOG.warn("Could not apply deletes captured during the full reindex; they stay AWAITING_REINDEX for the next full reindex", e); + } + + LOG.info("Applied deletes captured during the full reindex", logPair("count", applied)); + } + + + /******************************************************************************* ** Rebuild one table in place (kept for callers that used this directly). *******************************************************************************/ diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStep.java b/src/main/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStep.java index 74450bd..7aa7ef0 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStep.java @@ -146,7 +146,13 @@ public class ReconcileIndexStep extends AbstractIndexingStep values.put("recordCount", counts.processed()); values.put("documentCount", safeCount(client, tableName)); values.put("searchableFieldsJson", buildSearchableFieldsJson(tableConfig)); - values.put("status", STATUS_ACTIVE); + //////////////////////////////////////////////////////////////////// + // a full reindex running alongside owns the status until it ends // + //////////////////////////////////////////////////////////////////// + if(!STATUS_REBUILDING.equals(queryIndexRow(tableName).getValueString("status"))) + { + values.put("status", STATUS_ACTIVE); + } if(ReconcileIndexStep.RUN_TYPE.equals(runType) == false) { values.put("lastFullReindexTime", Instant.now()); diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/FullReindexDeleteReplayIntegrationTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/FullReindexDeleteReplayIntegrationTest.java new file mode 100644 index 0000000..c4e526b --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/quicksearch/FullReindexDeleteReplayIntegrationTest.java @@ -0,0 +1,244 @@ +/* + * Copyright 2024 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.io.Serializable; +import java.util.List; +import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction; +import com.kingsrook.qqq.backend.core.actions.tables.InsertAction; +import com.kingsrook.qqq.backend.core.actions.tables.QueryAction; +import com.kingsrook.qqq.backend.core.context.QContext; +import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput; +import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput; +import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput; +import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput; +import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria; +import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput; +import com.kingsrook.qqq.backend.core.model.data.QRecord; +import com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType; +import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData; +import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType; +import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData; +import com.kingsrook.qqq.backend.core.model.session.QSession; +import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule; +import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore; +import com.kingsrook.qbits.quicksearch.actions.QuickSearchAction; +import com.kingsrook.qbits.quicksearch.actions.QuickSearchInput; +import com.kingsrook.qbits.quicksearch.actions.QuickSearchResult; +import com.kingsrook.qbits.quicksearch.annotations.QuickSearchField; +import com.kingsrook.qbits.quicksearch.annotations.QuickSearchable; +import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent; +import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient; +import com.kingsrook.qbits.quicksearch.processes.FullReindexStep; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import static org.assertj.core.api.Assertions.assertThat; + + +/******************************************************************************* + ** Integration test, against a real OpenSearch, that a record deleted while a + ** full reindex rebuilds the index is gone from the new physical index after + ** the alias swap, even though the delete itself only reached the old one. + *******************************************************************************/ +@ExtendWith(RequiresDockerCondition.class) +@Testcontainers +class FullReindexDeleteReplayIntegrationTest +{ + private static final String BACKEND_NAME = "reindexDeleteMemoryBackend"; + private static final String TABLE_NAME = "reindexCustomer"; + private static final String INDEX_NAME = "quick_search_reindex_delete_test"; + + @Container + static GenericContainer opensearch = new GenericContainer<>(System.getProperty("opensearch.test.image", "opensearchproject/opensearch:2.19.6")) + .withExposedPorts(9200) + .withEnv("discovery.type", "single-node") + .withEnv("DISABLE_SECURITY_PLUGIN", "true") + .withEnv("DISABLE_INSTALL_DEMO_CONFIG", "true") + .waitingFor(Wait.forHttp("/_cluster/health").forStatusCode(200)); + + + + /******************************************************************************* + ** Entity annotated for quick search indexing. + *******************************************************************************/ + @QuickSearchable(tableName = TABLE_NAME) + static class ReindexCustomer + { + @QuickSearchField(weight = 2) + private String name; + + @QuickSearchField + private String city; + } + + + + /******************************************************************************* + ** Build the instance, produce the QBit with real-time indexing on and the + ** record-security re-read off (search returns whatever the index holds), + ** and initialize QContext. + *******************************************************************************/ + @BeforeAll + static void setUpAll() throws Exception + { + MemoryRecordStore.getInstance().reset(); + + QInstance qInstance = new QInstance(); + qInstance.addBackend(new QBackendMetaData() + .withName(BACKEND_NAME) + .withBackendType(MemoryBackendModule.class)); + qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData() + .withName("anonymous") + .withType(QAuthenticationType.FULLY_ANONYMOUS)); + qInstance.addTable(new QTableMetaData() + .withName(TABLE_NAME) + .withBackendName(BACKEND_NAME) + .withPrimaryKeyField("id") + .withField(new QFieldMetaData("id", QFieldType.INTEGER).withIsEditable(false)) + .withField(new QFieldMetaData("name", QFieldType.STRING)) + .withField(new QFieldMetaData("city", QFieldType.STRING))); + + QuickSearchQBitConfig config = new QuickSearchQBitConfig() + .withBackendName(BACKEND_NAME) + .withOpensearchHost(opensearch.getHost()) + .withOpensearchPort(opensearch.getMappedPort(9200)) + .withOpensearchIndexName(INDEX_NAME) + .withSearchableEntityClasses(List.of(ReindexCustomer.class)) + .withEnableRealTimeIndexing(true) + .withApplyRecordSecurityLocks(false); + + new QuickSearchQBitProducer().withConfig(config).produce(qInstance); + + QContext.init(qInstance, new QSession()); + } + + + + /******************************************************************************* + ** Clean up QContext and QBit state after all tests complete. + *******************************************************************************/ + @AfterAll + static void tearDownAll() + { + QContext.clear(); + QuickSearchQBitContext.clear(); + MemoryRecordStore.getInstance().reset(); + } + + + + /******************************************************************************* + ** Test: a record deleted after the rebuild read it (its document is already + ** in the new physical index) is absent from the index and from search once + ** the rebuild has swapped the alias. + *******************************************************************************/ + @Test + void testDeleteDuringFullReindex_isAbsentAfterTheSwap() throws Exception + { + InsertInput insertInput = new InsertInput(); + insertInput.setTableName(TABLE_NAME); + insertInput.setRecords(List.of( + new QRecord().withValue("id", 1).withValue("name", "Gwendolyn").withValue("city", "Rebuildton"), + new QRecord().withValue("id", 2).withValue("name", "Stanford").withValue("city", "Rebuildton"))); + new InsertAction().execute(insertInput); + assertThat(searchRecordIds("rebuildton")).containsExactlyInAnyOrder("1", "2"); + + RunBackendStepOutput output = new RunBackendStepOutput(); + new FullReindexStep() + { + /////////////////////////////////////////////////////////////////////// + // delete a record once its document is in the new physical index, // + // while the alias still points at the old one // + /////////////////////////////////////////////////////////////////////// + @Override + protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List extraCriteria, String targetIndex) throws QException + { + IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex); + + DeleteInput deleteInput = new DeleteInput(); + deleteInput.setTableName(TABLE_NAME); + deleteInput.setPrimaryKeys(List.of(1)); + new DeleteAction().execute(deleteInput); + + return (counts); + } + }.run(new RunBackendStepInput(), output); + + assertThat(output.getValueBoolean("aliasSwapped")).isTrue(); + assertThat(documentRecordIds("rebuildton")) + .as("the deleted record's document must not survive the rebuild") + .containsExactly("2"); + assertThat(searchRecordIds("rebuildton")).containsExactly("2"); + + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(QuickSearchFailedEvent.TABLE_NAME); + assertThat(new QueryAction().execute(queryInput).getRecords()) + .as("the captured delete was applied and its row removed") + .isEmpty(); + } + + + + /******************************************************************************* + ** Refresh the index, then return the record IDs search returns for the term. + *******************************************************************************/ + private List searchRecordIds(String term) throws QException + { + client().refreshIndex(); + return (new QuickSearchAction().execute(new QuickSearchInput().withSearchTerm(term).withTableName(TABLE_NAME)) + .getResults().stream() + .map(QuickSearchResult::getRecordId) + .toList()); + } + + + + /******************************************************************************* + ** Refresh the index, then return the record IDs of the documents matching + ** the term, straight from OpenSearch. + *******************************************************************************/ + private List documentRecordIds(String term) throws QException + { + client().refreshIndex(); + return (client().search(term, List.of(TABLE_NAME), 10, 0, QuickSearchQBitContext.getDiscoveredTables()) + .hits().hits().stream() + .map(hit -> hit.source().getRecordId()) + .toList()); + } + + + + /******************************************************************************* + ** The live OpenSearch client from the QBit context. + *******************************************************************************/ + private QuickSearchOpenSearchClient client() + { + return (QuickSearchQBitContext.getClient()); + } + +} diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java index bbaac71..6a0a297 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java @@ -41,6 +41,7 @@ import com.kingsrook.qbits.quicksearch.QuickSearchRuntime; import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig; import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent; import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex; +import com.kingsrook.qbits.quicksearch.processes.AbstractIndexingStep; import com.kingsrook.qbits.quicksearch.publisher.IndexEvent; import com.kingsrook.qbits.quicksearch.publisher.IndexEventAction; import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher; @@ -230,4 +231,98 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest verify(publisher, never()).publishIndexEvents(anyList()); } + + + @Test + void testDelete_duringFullReindex_isCapturedForTheNewIndex() throws QException + { + insertIndexRowWithStatus(AbstractIndexingStep.STATUS_REBUILDING); + Integer id = insert("widget", null); + delete(id); + + verify(publisher, times(1)).publishDeleteEvents(anyList()); + + List captured = queryFailedEvents(); + assertThat(captured).hasSize(1); + assertThat(captured.get(0).getValueString("tableName")).isEqualTo(TEST_ENTITY_TABLE); + assertThat(captured.get(0).getValueString("recordId")).isEqualTo(String.valueOf(id)); + assertThat(captured.get(0).getValueString("action")).isEqualTo("DELETE"); + assertThat(captured.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX); + } + + + + @Test + void testDelete_publishFailureDuringFullReindex_recordsBothRows() throws QException + { + doThrow(new QException("cluster down")).when(publisher).publishDeleteEvents(anyList()); + insertIndexRowWithStatus(AbstractIndexingStep.STATUS_REBUILDING); + delete(insert("widget", null)); + + assertThat(queryFailedEvents()).extracting(r -> r.getValueString("status")) + .containsExactlyInAnyOrder(QuickSearchFailedEvent.STATUS_PENDING, QuickSearchFailedEvent.STATUS_AWAITING_REINDEX); + } + + + + @Test + void testDelete_withoutFullReindex_isNotCaptured() throws QException + { + insertIndexRowWithStatus(AbstractIndexingStep.STATUS_ACTIVE); + delete(insert("widget", null)); + + verify(publisher, times(1)).publishDeleteEvents(anyList()); + assertThat(queryFailedEvents()).isEmpty(); + } + + + + @Test + void testCapture_rebuildFinishedBeforeInsert_handsRowsToBasepull() throws QException + { + //////////////////////////////////////////////////////////////////////// + // the second read sees the rebuild finished, as if it ended between // + // the first read and the insert // + //////////////////////////////////////////////////////////////////////// + insertIndexRowWithStatus(AbstractIndexingStep.STATUS_ACTIVE); + List events = List.of(new IndexEvent().withAction(IndexEventAction.DELETE).withTableName(TEST_ENTITY_TABLE).withRecordId("7")); + + List inserted = QuickSearchRecordChangeListener.insertFailedEventRows(runtime, events, "captured", QuickSearchFailedEvent.STATUS_AWAITING_REINDEX); + QuickSearchRecordChangeListener.handOverIfRebuildFinished(runtime, TEST_ENTITY_TABLE, inserted); + + List rows = queryFailedEvents(); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING); + } + + + + private void delete(Integer id) throws QException + { + DeleteInput deleteInput = new DeleteInput(); + deleteInput.setTableName(TEST_ENTITY_TABLE); + deleteInput.setPrimaryKeys(List.of(id)); + new DeleteAction().execute(deleteInput); + } + + + + private void insertIndexRowWithStatus(String status) throws QException + { + InsertInput insertInput = new InsertInput(); + insertInput.setTableName(QuickSearchIndex.TABLE_NAME); + insertInput.setRecords(List.of(new QRecord().withValue("tableName", TEST_ENTITY_TABLE).withValue("enabled", true).withValue("status", status))); + new InsertAction().execute(insertInput); + runtime.invalidateEnabledCache(); + } + + + + private List queryFailedEvents() throws QException + { + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(QuickSearchFailedEvent.TABLE_NAME); + return (new QueryAction().execute(queryInput).getRecords()); + } + } diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java index b704bee..ca456a9 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java @@ -864,4 +864,50 @@ class BasepullIndexStepTest extends BaseQuickSearchTest assertThat(new BasepullIndexStep().isDueForBasepull(index)).isTrue(); } + + + /******************************************************************************* + ** Test: a delete captured during a full reindex is left for that reindex + ** to apply to the new index, not replayed against the old one. + *******************************************************************************/ + @Test + void testReplayFailedEvents_awaitingReindexRowsAreNotReplayed() throws QException + { + insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60); + QRecord captured = insertFailedEvent(TEST_ENTITY_TABLE, "7", "DELETE", 0); + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(QuickSearchFailedEvent.TABLE_NAME); + updateInput.setRecords(List.of(new QRecord().withValue("id", captured.getValue("id")).withValue("status", QuickSearchFailedEvent.STATUS_AWAITING_REINDEX))); + new UpdateAction().execute(updateInput); + + RunBackendStepOutput output = new RunBackendStepOutput(); + new BasepullIndexStep().run(new RunBackendStepInput(), output); + + verify(mockClient, never()).deleteDocuments(anyList(), anyInt()); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).hasSize(1); + assertThat(output.getValueInteger("failedEventsReplayed")).isEqualTo(0); + } + + + + /******************************************************************************* + ** Test: drift found while a full reindex is running does not replace the + ** REBUILDING status the listener relies on. + *******************************************************************************/ + @Test + void testDriftDetection_duringFullReindex_keepsRebuilding() throws QException + { + QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60); + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(QuickSearchIndex.TABLE_NAME); + updateInput.setRecords(List.of(new QRecord().withValue("id", row.getValue("id")) + .withValue("status", AbstractIndexingStep.STATUS_REBUILDING) + .withValue("searchableFieldsJson", "[{\"fieldName\":\"name\",\"weight\":1,\"includeLabel\":false}]"))); + new UpdateAction().execute(updateInput); + + new BasepullIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()); + + assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING); + } + } diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java index 2488c00..95f62ef 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java @@ -21,6 +21,7 @@ import java.util.ArrayList; import java.util.List; import com.kingsrook.qqq.backend.core.actions.tables.InsertAction; import com.kingsrook.qqq.backend.core.actions.tables.QueryAction; +import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction; import com.kingsrook.qqq.backend.core.exceptions.QException; import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput; import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput; @@ -29,9 +30,12 @@ import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperat import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput; +import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput; import com.kingsrook.qqq.backend.core.model.data.QRecord; import com.kingsrook.qbits.quicksearch.BaseQuickSearchTest; import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext; +import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig; +import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent; import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex; import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun; import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult; @@ -474,4 +478,116 @@ class FullReindexStepTest extends BaseQuickSearchTest verify(mockClient, times(3)).indexDocuments(eq(PHYSICAL_INDEX), anyList(), anyInt()); } + + + /******************************************************************************* + ** Test: the index row reads REBUILDING while tables are read, and a delete + ** captured during the rebuild is applied through the alias after the swap, + ** then its failed-event row is removed. + *******************************************************************************/ + @Test + void testAllTables_appliesCapturedDeletesAfterTheSwap() throws QException + { + insertTestEntities(3); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1)); + + List statusesDuringRebuild = new ArrayList<>(); + new FullReindexStep() + { + @Override + protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List extraCriteria, String targetIndex) throws QException + { + IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex); + statusesDuringRebuild.add(FullReindexStepTest.this.queryIndexRow().getValueString("status")); + insertCapturedDelete("2"); + return (counts); + } + }.run(new RunBackendStepInput(), new RunBackendStepOutput()); + + assertThat(statusesDuringRebuild).containsExactly(AbstractIndexingStep.STATUS_REBUILDING); + + InOrder inOrder = inOrder(mockClient); + inOrder.verify(mockClient).swapAliasTo(PHYSICAL_INDEX); + inOrder.verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":2")), anyInt()); + + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty(); + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + } + + + + /******************************************************************************* + ** Test: a captured delete the cluster rejects is handed to basepull. + *******************************************************************************/ + @Test + void testAllTables_capturedDeleteFailure_becomesPendingForBasepull() throws QException + { + insertTestEntities(1); + insertCapturedDelete("9"); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenThrow(new QException("cluster down")); + + new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()); + + List rows = queryAll(QuickSearchFailedEvent.TABLE_NAME); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING); + assertThat(rows.get(0).getValueString("errorMessage")).contains("cluster down"); + } + + + + /******************************************************************************* + ** Test: when the rebuild fails before the swap, the previous status comes + ** back and the captured deletes still leave the AWAITING_REINDEX state. + *******************************************************************************/ + @Test + void testAllTables_failureBeforeSwap_restoresStatusAndDrainsCapturedDeletes() throws QException + { + insertTestEntities(1); + insertIndexRow(true); + updateIndexRowStatus(AbstractIndexingStep.STATUS_NEEDS_REINDEX); + insertCapturedDelete("4"); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1)); + doThrow(new QException("swap refused")).when(mockClient).swapAliasTo(PHYSICAL_INDEX); + + assertThatThrownBy(() -> new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput())) + .isInstanceOf(QException.class); + + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_NEEDS_REINDEX); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty(); + verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":4")), anyInt()); + } + + + + /******************************************************************************* + ** Insert an AWAITING_REINDEX delete for testEntity, as the listener does + ** during a full reindex. + *******************************************************************************/ + private void insertCapturedDelete(String recordId) throws QException + { + InsertInput insertInput = new InsertInput(); + insertInput.setTableName(QuickSearchFailedEvent.TABLE_NAME); + insertInput.setRecords(List.of(new QRecord() + .withValue("tableName", TEST_ENTITY_TABLE) + .withValue("recordId", recordId) + .withValue("action", "DELETE") + .withValue("attempts", 0) + .withValue("status", QuickSearchFailedEvent.STATUS_AWAITING_REINDEX))); + new InsertAction().execute(insertInput); + } + + + + /******************************************************************************* + ** Set the status on the testEntity quickSearchIndex row. + *******************************************************************************/ + private void updateIndexRowStatus(String status) throws QException + { + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(QuickSearchIndex.TABLE_NAME); + updateInput.setRecords(List.of(new QRecord().withValue("id", queryIndexRow().getValue("id")).withValue("status", status))); + new UpdateAction().execute(updateInput); + } + } diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStepTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStepTest.java index b765a7c..cd08eae 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStepTest.java @@ -386,4 +386,28 @@ class ReconcileIndexStepTest extends BaseQuickSearchTest assertThat(row.getValueString("lastErrorMessage")).contains("connection refused"); } + + + /******************************************************************************* + ** Test: a reconcile finishing while a full reindex runs leaves the + ** REBUILDING status in place, so deletes are still captured. + *******************************************************************************/ + @Test + void testReconcile_duringFullReindex_keepsRebuilding() throws QException + { + insertTestEntities(1); + InsertInput insertInput = new InsertInput(); + insertInput.setTableName(QuickSearchIndex.TABLE_NAME); + insertInput.setRecords(List.of(new QRecord() + .withValue("tableName", TEST_ENTITY_TABLE) + .withValue("enabled", true) + .withValue("status", AbstractIndexingStep.STATUS_REBUILDING))); + new InsertAction().execute(insertInput); + + new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()); + + assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING); + assertThat(indexedRecordIds()).containsExactly("1"); + } + } -- 2.54.0 From 89c675bea55f9614d245f179c5b70b24418bca67 Mon Sep 17 00:00:00 2001 From: James Maes Date: Tue, 6 Oct 2026 18:28:47 -0500 Subject: [PATCH 2/3] fix(#14): read rebuild status before publish, harden drain The listener reads REBUILDING before publishing deletes, drift detection re-reads the row before writing, and the drain pages past rows it cannot remove and skips records re-created with the same key. Refs #14 --- README.md | 2 +- .../QuickSearchRecordChangeListener.java | 58 +++-- .../processes/AbstractIndexingStep.java | 10 + .../processes/FullReindexStep.java | 204 +++++++++++++----- .../QuickSearchRecordChangeListenerTest.java | 41 ++++ .../processes/BasepullIndexStepTest.java | 28 +++ .../processes/FullReindexStepTest.java | 73 ++++++- 7 files changed, 346 insertions(+), 70 deletions(-) diff --git a/README.md b/README.md index 1558e1d..0593730 100644 --- a/README.md +++ b/README.md @@ -169,7 +169,7 @@ Without `schedulerName` the processes exist but nothing runs them; the QBit logs - `quickSearchIndex`: one row per table with `enabled` (honoured by indexing and search, cached for one minute), `basepullIntervalMinutes`, `lastBasepullTime`, `lastReconcileTime`, `lastFullReindexTime`, `lastRunStatus`, `lastErrorMessage`, `recordCount`, `documentCount`, `realTimeErrorCount`, `status` (`ACTIVE`, `NEEDS_REINDEX` when the configured fields changed since the row was created, or `REBUILDING` while a full reindex of all tables runs). Unique on `tableName`. - `quickSearchIndexRun`: `runType` (`BASEPULL`, `RECONCILE`, `FULL_REINDEX`), `status`, counts, `errorMessage`. -- `quickSearchFailedEvent`: real-time events that could not be applied; `status` `PENDING` or `EXHAUSTED` after 10 attempts. Deletes made while a full reindex of all tables runs are also recorded here with status `AWAITING_REINDEX`: the reindex applies them to the new index after the alias swap (basepull does not replay them). If a full reindex is killed mid-run, its tables stay `REBUILDING` and deletes keep collecting as `AWAITING_REINDEX` until the next full reindex completes. +- `quickSearchFailedEvent`: real-time events that could not be applied; `status` `PENDING` or `EXHAUSTED` after 10 attempts. Deletes made while a full reindex of all tables runs are also recorded here with status `AWAITING_REINDEX`: the reindex applies them to the new index after the alias swap (basepull does not replay them). While a table is `REBUILDING`, drift detection is suspended for it. If a full reindex is killed mid-run, its tables stay `REBUILDING` and deletes keep collecting as `AWAITING_REINDEX`; no data is lost, and the recovery is to run a full reindex of all tables again, which applies the collected deletes and returns the tables to `ACTIVE`. ## Extending diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java b/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java index 26f9724..990f884 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java @@ -137,13 +137,25 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte /******************************************************************************* - ** Publish, recording any failure durably, then capture the deletes if a full + ** Publish, recording any failure durably, and capture the deletes if a full ** reindex is rebuilding the table. Never throws. + ** + ** The status is read before the deletes are published, never after: a + ** reindex can swap the alias and clear REBUILDING while the publish is in + ** flight, and a read after that would miss a delete that reached only the + ** old index. A read before the publish that sees no REBUILDING means either + ** the reindex has not started reading the table (the record, deleted and + ** committed, cannot be read into the new index) or the alias already points + ** at the new index (the publish reaches it). *******************************************************************************/ static void publish(QuickSearchRuntime runtime, List indexEvents, List deleteEvents) { + boolean rebuilding = isRebuildingBeforePublish(runtime, deleteEvents); publishOrRecordFailures(runtime, indexEvents, deleteEvents); - captureDeletesDuringRebuild(runtime, deleteEvents); + if(rebuilding) + { + captureDeletesDuringRebuild(runtime, deleteEvents); + } } @@ -225,32 +237,42 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte /******************************************************************************* - ** Record deletes made while a full reindex is rebuilding the table, so the - ** reindex can apply them to the new physical index after the alias swap. - ** basepull does not replay AWAITING_REINDEX rows, so it cannot consume one - ** against the old index before the swap. - ** - ** The table's status is read from the database on every call (not cached) - ** so a reindex running on another node is seen. This runs after the delete - ** committed; the reindex marks the table REBUILDING before reading it, so a - ** delete of a record the reindex may have read always sees that status. - ** Never throws. + ** Whether a full reindex is rebuilding the deletes' table. The status is + ** read from the database (not cached) so a reindex on another node is seen. + ** Runs after the delete committed. Never throws. *******************************************************************************/ - static void captureDeletesDuringRebuild(QuickSearchRuntime runtime, List deleteEvents) + static boolean isRebuildingBeforePublish(QuickSearchRuntime runtime, List deleteEvents) { if(deleteEvents.isEmpty()) { - return; + return (false); } String tableName = deleteEvents.get(0).getTableName(); try { - if(!hasTable(runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME)) || !isRebuilding(runtime, tableName)) - { - return; - } + return (hasTable(runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME)) && isRebuilding(runtime, tableName)); + } + catch(Exception e) + { + LOG.error("Could not read whether a full reindex is running; deletes made during one are not captured", e, logPair("tableName", tableName)); + return (false); + } + } + + + /******************************************************************************* + ** Record deletes made while a full reindex is rebuilding the table, so the + ** reindex can apply them to the new physical index after the alias swap. + ** basepull does not replay AWAITING_REINDEX rows, so it cannot consume one + ** against the old index before the swap. Never throws. + *******************************************************************************/ + static void captureDeletesDuringRebuild(QuickSearchRuntime runtime, List deleteEvents) + { + String tableName = deleteEvents.get(0).getTableName(); + try + { List inserted = insertFailedEventRows(runtime, deleteEvents, "deleted during a full reindex", QuickSearchFailedEvent.STATUS_AWAITING_REINDEX); handOverIfRebuildFinished(runtime, tableName, inserted); } diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java b/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java index 00fb387..0ada7e8 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java @@ -238,6 +238,16 @@ public abstract class AbstractIndexingStep implements BackendStep return (recorded != null && !recorded.equals(current)); } + /////////////////////////////////////////////////////////////////////////// + // the row may be a snapshot taken before a full reindex started; read // + // it again so the REBUILDING status the listener relies on is kept // + /////////////////////////////////////////////////////////////////////////// + QRecord latest = queryIndexRow(tableConfig.getTableName()); + if(latest != null && STATUS_REBUILDING.equals(latest.getValueString("status"))) + { + return (true); + } + LOG.warn("Searchable field configuration changed since the index row was created; marking NEEDS_REINDEX", logPair("tableName", tableConfig.getTableName())); updateIndexRow(row.getValueInteger("id"), Map.of("status", STATUS_NEEDS_REINDEX)); return (true); diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java b/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java index 185e736..9ced2dd 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java @@ -21,9 +21,11 @@ import java.io.Serializable; import java.time.Instant; import java.util.ArrayList; import java.util.HashMap; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction; import com.kingsrook.qqq.backend.core.actions.tables.QueryAction; import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction; @@ -32,6 +34,7 @@ import com.kingsrook.qqq.backend.core.logging.QLogger; import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput; import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput; import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput; +import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteOutput; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy; @@ -272,10 +275,14 @@ public class FullReindexStep extends AbstractIndexingStep ** Apply the deletes the listener captured during the rebuild (failed-event ** rows with status AWAITING_REINDEX) through the alias, after the rebuild's ** own writes, so an in-flight page cannot bring a deleted record back. - ** Deleting a missing document is harmless. Applied rows are removed; rows - ** the cluster rejects become PENDING for basepull to retry. Called once the - ** tables are no longer REBUILDING, so the listener stops adding rows (one - ** that races the status change hands its own rows to basepull). Never throws. + ** Called once the tables are no longer REBUILDING, so the listener stops + ** adding rows (one that races the status change hands its own rows to + ** basepull). + ** + ** Rows are read in id order, past the previous batch, so a batch whose rows + ** could not be removed or updated (QQQ reports those as per-row errors, not + ** exceptions) is not read again; such rows stay AWAITING_REINDEX for the next + ** full reindex. Never throws. *******************************************************************************/ void applyCapturedDeletes(QuickSearchOpenSearchClient client, List tables) { @@ -291,63 +298,31 @@ public class FullReindexStep extends AbstractIndexingStep { QuickSearchQBitConfig config = getConfig(); String failedEventTable = config.applyPrefix(QuickSearchFailedEvent.TABLE_NAME); + Serializable afterId = null; while(true) { - QueryInput queryInput = new QueryInput(); - queryInput.setTableName(failedEventTable); - queryInput.setFilter(new QQueryFilter( + QQueryFilter filter = new QQueryFilter( new QFilterCriteria("status", QCriteriaOperator.EQUALS, QuickSearchFailedEvent.STATUS_AWAITING_REINDEX), new QFilterCriteria("tableName", QCriteriaOperator.IN, tableNames)) .withOrderBy(new QFilterOrderBy("id", true)) - .withLimit(config.getSourceBatchSize())); + .withLimit(config.getSourceBatchSize()); + if(afterId != null) + { + filter.addCriteria(new QFilterCriteria("id", QCriteriaOperator.GREATER_THAN, afterId)); + } + + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(failedEventTable); + queryInput.setFilter(filter); List rows = new QueryAction().execute(queryInput).getRecords(); if(CollectionUtils.nullSafeIsEmpty(rows)) { break; } - List documentIds = new ArrayList<>(); - for(QRecord row : rows) - { - documentIds.add(OpenSearchDocument.buildDocumentId(row.getValueString("tableName"), row.getValueString("recordId"))); - } - - String error; - try - { - BulkIndexResult result = client.deleteDocuments(documentIds, config.getBulkBatchSize()); - error = result.getFailureCount() == 0 ? null : String.join("; ", result.getErrors()); - } - catch(Exception e) - { - error = e.getMessage() == null ? e.toString() : e.getMessage(); - } - - if(error == null) - { - DeleteInput deleteInput = new DeleteInput(); - deleteInput.setTableName(failedEventTable); - deleteInput.setPrimaryKeys(rows.stream().map(row -> row.getValue("id")).toList()); - new DeleteAction().execute(deleteInput); - applied = applied + rows.size(); - } - else - { - List updates = new ArrayList<>(); - for(QRecord row : rows) - { - updates.add(new QRecord() - .withValue("id", row.getValue("id")) - .withValue("errorMessage", truncate(error, 4000)) - .withValue("status", QuickSearchFailedEvent.STATUS_PENDING)); - } - UpdateInput updateInput = new UpdateInput(); - updateInput.setTableName(failedEventTable); - updateInput.setRecords(updates); - new UpdateAction().execute(updateInput); - LOG.warn("Could not apply deletes captured during the full reindex; basepull will retry them", logPair("count", rows.size()), logPair("error", error)); - } + afterId = rows.get(rows.size() - 1).getValue("id"); + applied = applied + applyCapturedDeleteBatch(client, config, failedEventTable, rows); } } catch(Exception e) @@ -360,6 +335,137 @@ public class FullReindexStep extends AbstractIndexingStep + /******************************************************************************* + ** Apply one batch of captured deletes; returns how many were applied. + ** A record that exists again (re-created with the same key since it was + ** deleted) keeps its document, and its row is just removed. Deleting a + ** missing document is harmless. Applied rows are removed; rows the cluster + ** rejects become PENDING for basepull to retry. + *******************************************************************************/ + private Integer applyCapturedDeleteBatch(QuickSearchOpenSearchClient client, QuickSearchQBitConfig config, String failedEventTable, List rows) throws QException + { + Map> rowsByTable = new LinkedHashMap<>(); + for(QRecord row : rows) + { + rowsByTable.computeIfAbsent(row.getValueString("tableName"), k -> new ArrayList<>()).add(row); + } + + List finished = new ArrayList<>(); + List toApply = new ArrayList<>(); + for(Map.Entry> entry : rowsByTable.entrySet()) + { + Set existingIds = queryExistingRecordIds(entry.getKey(), entry.getValue()); + for(QRecord row : entry.getValue()) + { + if(existingIds.contains(row.getValueString("recordId"))) + { + finished.add(row); + } + else + { + toApply.add(row); + } + } + } + + String error = null; + if(!toApply.isEmpty()) + { + List documentIds = new ArrayList<>(); + for(QRecord row : toApply) + { + documentIds.add(OpenSearchDocument.buildDocumentId(row.getValueString("tableName"), row.getValueString("recordId"))); + } + + try + { + BulkIndexResult result = client.deleteDocuments(documentIds, config.getBulkBatchSize()); + error = result.getFailureCount() == 0 ? null : String.join("; ", result.getErrors()); + } + catch(Exception e) + { + error = e.getMessage() == null ? e.toString() : e.getMessage(); + } + + if(error == null) + { + finished.addAll(toApply); + } + } + + if(!finished.isEmpty()) + { + DeleteInput deleteInput = new DeleteInput(); + deleteInput.setTableName(failedEventTable); + deleteInput.setPrimaryKeys(finished.stream().map(row -> row.getValue("id")).toList()); + DeleteOutput deleteOutput = new DeleteAction().execute(deleteInput); + if(CollectionUtils.nullSafeHasContents(deleteOutput.getRecordsWithErrors())) + { + LOG.warn("Could not remove applied captured-delete rows; they stay AWAITING_REINDEX and are applied again by the next full reindex", + logPair("count", deleteOutput.getRecordsWithErrors().size()), logPair("firstError", deleteOutput.getRecordsWithErrors().get(0).getErrors())); + } + } + + if(error != null) + { + List updates = new ArrayList<>(); + for(QRecord row : toApply) + { + updates.add(new QRecord() + .withValue("id", row.getValue("id")) + .withValue("errorMessage", truncate(error, 4000)) + .withValue("status", QuickSearchFailedEvent.STATUS_PENDING)); + } + + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(failedEventTable); + updateInput.setRecords(updates); + List notUpdated = CollectionUtils.nonNullList(new UpdateAction().execute(updateInput).getRecords()).stream() + .filter(record -> CollectionUtils.nullSafeHasContents(record.getErrors())) + .toList(); + LOG.warn("Could not apply deletes captured during the full reindex; basepull will retry them", logPair("count", toApply.size()), logPair("error", error)); + if(!notUpdated.isEmpty()) + { + LOG.warn("Could not hand captured deletes to basepull; they stay AWAITING_REINDEX for the next full reindex", + logPair("count", notUpdated.size()), logPair("firstError", notUpdated.get(0).getErrors())); + } + return (0); + } + + return (toApply.size()); + } + + + + /******************************************************************************* + ** The record IDs, among the rows' recordIds, that exist in the source table. + *******************************************************************************/ + private Set queryExistingRecordIds(String tableName, List rows) throws QException + { + QuickSearchableTableConfig tableConfig = getTableConfig(tableName); + if(tableConfig == null) + { + return (Set.of()); + } + + String primaryKeyField = tableConfig.getPrimaryKeyField() == null ? "id" : tableConfig.getPrimaryKeyField(); + List recordIds = new ArrayList<>(); + rows.forEach(row -> recordIds.add(row.getValueString("recordId"))); + + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(tableName); + queryInput.setFilter(new QQueryFilter(new QFilterCriteria(primaryKeyField, QCriteriaOperator.IN, recordIds))); + + Set existingIds = new HashSet<>(); + for(QRecord record : CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords())) + { + existingIds.add(record.getValueString(primaryKeyField)); + } + return (existingIds); + } + + + /******************************************************************************* ** Rebuild one table in place (kept for callers that used this directly). *******************************************************************************/ diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java index 6a0a297..713a289 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java @@ -50,6 +50,7 @@ import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -325,4 +326,44 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest return (new QueryAction().execute(queryInput).getRecords()); } + + + @Test + void testDelete_rebuildEndsDuringPublish_isStillCaptured() throws QException + { + //////////////////////////////////////////////////////////////////////////// + // the reindex swaps the alias and clears REBUILDING while the delete is // + // being published to the old index: the status read must come first // + //////////////////////////////////////////////////////////////////////////// + insertIndexRowWithStatus(AbstractIndexingStep.STATUS_REBUILDING); + Integer id = insert("widget", null); + doAnswer(invocation -> + { + setIndexRowStatus(AbstractIndexingStep.STATUS_ACTIVE); + return (null); + }).when(publisher).publishDeleteEvents(anyList()); + + delete(id); + + List rows = queryFailedEvents(); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getValueString("recordId")).isEqualTo(String.valueOf(id)); + assertThat(rows.get(0).getValueString("action")).isEqualTo("DELETE"); + assertThat(rows.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING); + } + + + + private void setIndexRowStatus(String status) throws QException + { + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(QuickSearchIndex.TABLE_NAME); + QRecord row = new QueryAction().execute(queryInput).getRecords().get(0); + + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(QuickSearchIndex.TABLE_NAME); + updateInput.setRecords(List.of(new QRecord().withValue("id", row.getValue("id")).withValue("status", status))); + new UpdateAction().execute(updateInput); + } + } diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java index ca456a9..4624729 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java @@ -910,4 +910,32 @@ class BasepullIndexStepTest extends BaseQuickSearchTest assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING); } + + + /******************************************************************************* + ** Test: drift found on a row snapshot taken before a full reindex started + ** does not overwrite the REBUILDING status now in the database. + *******************************************************************************/ + @Test + void testDriftDetection_staleSnapshot_keepsRebuilding() throws QException + { + QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60); + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(QuickSearchIndex.TABLE_NAME); + updateInput.setRecords(List.of(new QRecord().withValue("id", row.getValue("id")).withValue("status", AbstractIndexingStep.STATUS_REBUILDING))); + new UpdateAction().execute(updateInput); + + QRecord staleSnapshot = new QRecord() + .withValue("id", row.getValue("id")) + .withValue("tableName", TEST_ENTITY_TABLE) + .withValue("status", AbstractIndexingStep.STATUS_ACTIVE) + .withValue("searchableFieldsJson", "[{\"fieldName\":\"name\",\"weight\":1,\"includeLabel\":false}]"); + QuickSearchableTableConfig tableConfig = QuickSearchQBitContext.getDiscoveredTables().stream() + .filter(table -> TEST_ENTITY_TABLE.equals(table.getTableName())) + .findFirst().orElseThrow(); + + assertThat(new BasepullIndexStep().detectDrift(staleSnapshot, tableConfig)).isTrue(); + assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING); + } + } diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java index 95f62ef..c51bef0 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java @@ -19,12 +19,15 @@ package com.kingsrook.qbits.quicksearch.processes; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction; import com.kingsrook.qqq.backend.core.actions.tables.InsertAction; import com.kingsrook.qqq.backend.core.actions.tables.QueryAction; import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction; import com.kingsrook.qqq.backend.core.exceptions.QException; import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput; import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput; +import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteOutput; import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria; @@ -32,6 +35,7 @@ import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter; import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput; import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput; import com.kingsrook.qqq.backend.core.model.data.QRecord; +import com.kingsrook.qqq.backend.core.model.statusmessages.SystemErrorStatusMessage; import com.kingsrook.qbits.quicksearch.BaseQuickSearchTest; import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext; import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig; @@ -45,6 +49,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.mockito.InOrder; +import org.mockito.MockedConstruction; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; @@ -55,6 +60,7 @@ import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -499,7 +505,7 @@ class FullReindexStepTest extends BaseQuickSearchTest { IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex); statusesDuringRebuild.add(FullReindexStepTest.this.queryIndexRow().getValueString("status")); - insertCapturedDelete("2"); + insertCapturedDelete("7"); return (counts); } }.run(new RunBackendStepInput(), new RunBackendStepOutput()); @@ -508,7 +514,7 @@ class FullReindexStepTest extends BaseQuickSearchTest InOrder inOrder = inOrder(mockClient); inOrder.verify(mockClient).swapAliasTo(PHYSICAL_INDEX); - inOrder.verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":2")), anyInt()); + inOrder.verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":7")), anyInt()); assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty(); assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); @@ -590,4 +596,67 @@ class FullReindexStepTest extends BaseQuickSearchTest new UpdateAction().execute(updateInput); } + + + /******************************************************************************* + ** Test: a captured delete for a record re-created with the same key since + ** is not applied (its document stays), and its row is removed. + *******************************************************************************/ + @Test + void testAllTables_capturedDeleteOfRecreatedRecord_isNotApplied() throws QException + { + insertTestEntities(3); + insertCapturedDelete("2"); + insertCapturedDelete("9"); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1)); + + new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()); + + verify(mockClient, times(1)).deleteDocuments(anyList(), anyInt()); + verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":9")), anyInt()); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty(); + } + + + + /******************************************************************************* + ** Test: rows the delete action reports as not removed (per-row errors, not + ** an exception) are not read again; the drain moves on and ends. + *******************************************************************************/ + @Test + void testAllTables_capturedRowsNotRemoved_drainStillEnds() throws QException + { + QuickSearchQBitContext.getConfig().withSourceBatchSize(1); + insertTestEntities(1); + insertCapturedDelete("8"); + insertCapturedDelete("9"); + + AtomicInteger deleteCalls = new AtomicInteger(0); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenAnswer(invocation -> + { + /////////////////////////////////////////////////////////////////// + // fail loudly instead of hanging if the same batch is re-read // + /////////////////////////////////////////////////////////////////// + if(deleteCalls.incrementAndGet() > 10) + { + throw (new IllegalStateException("drain re-read the same rows")); + } + return (new BulkIndexResult().withSuccessCount(1)); + }); + + try(MockedConstruction deleteActions = mockConstruction(DeleteAction.class, (deleteAction, context) -> + { + DeleteOutput output = new DeleteOutput(); + output.setRecordsWithErrors(List.of(new QRecord().withValue("id", 1).withError(new SystemErrorStatusMessage("row is locked")))); + when(deleteAction.execute(any())).thenReturn(output); + })) + { + new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()); + } + + assertThat(deleteCalls.get()).isEqualTo(2); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).extracting(row -> row.getValueString("status")) + .containsOnly(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX); + } + } -- 2.54.0 From 02785dabcdfa9f1e045b7b4dc1679223cec53d2c Mon Sep 17 00:00:00 2001 From: James Maes Date: Tue, 6 Oct 2026 18:39:55 -0500 Subject: [PATCH 3/3] fix(#14): judge drift on the re-read row, isolate drain batches Drift detection compares the configured fields against the freshly read row, and a captured-delete batch that fails no longer stops the later batches. Refs #14 --- .../processes/AbstractIndexingStep.java | 42 ++++++++++++++----- .../processes/FullReindexStep.java | 12 +++++- .../processes/BasepullIndexStepTest.java | 32 +++++++++++++- .../processes/FullReindexStepTest.java | 38 +++++++++++++++++ 4 files changed, 111 insertions(+), 13 deletions(-) diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java b/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java index 0ada7e8..83ddf12 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java @@ -230,31 +230,53 @@ public abstract class AbstractIndexingStep implements BackendStep return (false); } - String current = buildSearchableFieldsJson(tableConfig); - String recorded = row.getValueString("searchableFieldsJson"); - String status = row.getValueString("status"); - if(recorded == null || recorded.equals(current) || STATUS_NEEDS_REINDEX.equals(status) || STATUS_REBUILDING.equals(status)) + String current = buildSearchableFieldsJson(tableConfig); + if(!isUnmarkedDrift(row, current)) { - return (recorded != null && !recorded.equals(current)); + return (hasDrift(row, current)); } /////////////////////////////////////////////////////////////////////////// - // the row may be a snapshot taken before a full reindex started; read // - // it again so the REBUILDING status the listener relies on is kept // + // the row may be a snapshot taken before a full reindex started or // + // finished; decide on a fresh read, so REBUILDING is kept and fields a // + // reindex has since recorded are not flagged // /////////////////////////////////////////////////////////////////////////// QRecord latest = queryIndexRow(tableConfig.getTableName()); - if(latest != null && STATUS_REBUILDING.equals(latest.getValueString("status"))) + if(latest == null || !isUnmarkedDrift(latest, current)) { - return (true); + return (latest != null && hasDrift(latest, current)); } + // a status change in the moment between this read and the update below is not guarded LOG.warn("Searchable field configuration changed since the index row was created; marking NEEDS_REINDEX", logPair("tableName", tableConfig.getTableName())); - updateIndexRow(row.getValueInteger("id"), Map.of("status", STATUS_NEEDS_REINDEX)); + updateIndexRow(latest.getValueInteger("id"), Map.of("status", STATUS_NEEDS_REINDEX)); return (true); } + /******************************************************************************* + ** Whether the row records fields that differ from the configured ones. + *******************************************************************************/ + private static boolean hasDrift(QRecord row, String current) + { + String recorded = row.getValueString("searchableFieldsJson"); + return (recorded != null && !recorded.equals(current)); + } + + + + /******************************************************************************* + ** Whether the row has drifted and is not yet NEEDS_REINDEX or REBUILDING. + *******************************************************************************/ + private static boolean isUnmarkedDrift(QRecord row, String current) + { + String status = row.getValueString("status"); + return (hasDrift(row, current) && !STATUS_NEEDS_REINDEX.equals(status) && !STATUS_REBUILDING.equals(status)); + } + + + /******************************************************************************* ** JSON array describing a table's searchable fields, weights and labels. *******************************************************************************/ diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java b/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java index 9ced2dd..392a2dd 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java @@ -322,7 +322,15 @@ public class FullReindexStep extends AbstractIndexingStep } afterId = rows.get(rows.size() - 1).getValue("id"); - applied = applied + applyCapturedDeleteBatch(client, config, failedEventTable, rows); + try + { + applied = applied + applyCapturedDeleteBatch(client, config, failedEventTable, rows); + } + catch(Exception e) + { + LOG.warn("Could not apply a batch of deletes captured during the full reindex; they stay AWAITING_REINDEX for the next full reindex", e, + logPair("tableNames", rows.stream().map(row -> row.getValueString("tableName")).distinct().toList()), logPair("count", rows.size())); + } } } catch(Exception e) @@ -440,7 +448,7 @@ public class FullReindexStep extends AbstractIndexingStep /******************************************************************************* ** The record IDs, among the rows' recordIds, that exist in the source table. *******************************************************************************/ - private Set queryExistingRecordIds(String tableName, List rows) throws QException + Set queryExistingRecordIds(String tableName, List rows) throws QException { QuickSearchableTableConfig tableConfig = getTableConfig(tableName); if(tableConfig == null) diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java index 4624729..cda9564 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java @@ -934,8 +934,38 @@ class BasepullIndexStepTest extends BaseQuickSearchTest .filter(table -> TEST_ENTITY_TABLE.equals(table.getTableName())) .findFirst().orElseThrow(); - assertThat(new BasepullIndexStep().detectDrift(staleSnapshot, tableConfig)).isTrue(); + new BasepullIndexStep().detectDrift(staleSnapshot, tableConfig); assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING); } + + + /******************************************************************************* + ** Test: a stale snapshot shows drift, but a full reindex has since recorded + ** the current fields; the row is not marked NEEDS_REINDEX. + *******************************************************************************/ + @Test + void testDriftDetection_staleSnapshot_reindexSinceRecordedCurrentFields() throws QException + { + QuickSearchableTableConfig tableConfig = QuickSearchQBitContext.getDiscoveredTables().stream() + .filter(table -> TEST_ENTITY_TABLE.equals(table.getTableName())) + .findFirst().orElseThrow(); + QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60); + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(QuickSearchIndex.TABLE_NAME); + updateInput.setRecords(List.of(new QRecord().withValue("id", row.getValue("id")) + .withValue("status", AbstractIndexingStep.STATUS_ACTIVE) + .withValue("searchableFieldsJson", AbstractIndexingStep.buildSearchableFieldsJson(tableConfig)))); + new UpdateAction().execute(updateInput); + + QRecord staleSnapshot = new QRecord() + .withValue("id", row.getValue("id")) + .withValue("tableName", TEST_ENTITY_TABLE) + .withValue("status", AbstractIndexingStep.STATUS_ACTIVE) + .withValue("searchableFieldsJson", "[{\"fieldName\":\"name\",\"weight\":1,\"includeLabel\":false}]"); + + assertThat(new BasepullIndexStep().detectDrift(staleSnapshot, tableConfig)).isFalse(); + assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + } + } diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java index c51bef0..01cd065 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java @@ -19,6 +19,7 @@ package com.kingsrook.qbits.quicksearch.processes; import java.util.ArrayList; import java.util.List; +import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction; import com.kingsrook.qqq.backend.core.actions.tables.InsertAction; @@ -659,4 +660,41 @@ class FullReindexStepTest extends BaseQuickSearchTest .containsOnly(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX); } + + + /******************************************************************************* + ** Test: a batch that fails before it is applied (here, the source lookup) + ** stays AWAITING_REINDEX, and the later batches are still applied. + *******************************************************************************/ + @Test + void testAllTables_failedCapturedBatch_laterBatchesStillApplied() throws QException + { + QuickSearchQBitContext.getConfig().withSourceBatchSize(1); + insertTestEntities(1); + insertCapturedDelete("8"); + insertCapturedDelete("9"); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1)); + + new FullReindexStep() + { + @Override + Set queryExistingRecordIds(String tableName, List rows) throws QException + { + if("8".equals(rows.get(0).getValueString("recordId"))) + { + throw (new QException("source table unavailable")); + } + return (super.queryExistingRecordIds(tableName, rows)); + } + }.run(new RunBackendStepInput(), new RunBackendStepOutput()); + + verify(mockClient, times(1)).deleteDocuments(anyList(), anyInt()); + verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":9")), anyInt()); + + List rows = queryAll(QuickSearchFailedEvent.TABLE_NAME); + assertThat(rows).hasSize(1); + assertThat(rows.get(0).getValueString("recordId")).isEqualTo("8"); + assertThat(rows.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX); + } + } -- 2.54.0