From 59f44b43ef3b29053b03d2bc89f3af2ae8a177f0 Mon Sep 17 00:00:00 2001 From: James Maes Date: Thu, 8 Oct 2026 07:39:31 -0500 Subject: [PATCH 1/3] fix(#24): recover a stuck REBUILDING status after a killed reindex A full reindex now refuses to start while another is alive, judged by a per-page heartbeat on its FULL_REINDEX run records. A run stale for fullReindexStaleMinutes (default 30) is recovered by the next basepull or full reindex: its records become FAILED, its tables ACTIVE, and its captured deletes PENDING for basepull. Closes #24 --- README.md | 7 +- .../quicksearch/QuickSearchQBitConfig.java | 54 +++ .../processes/AbstractIndexingStep.java | 27 +- .../processes/BasepullIndexStep.java | 18 +- .../processes/FullReindexStep.java | 406 ++++++++++++++++- .../processes/ReconcileIndexStep.java | 2 +- ...ullReindexDeleteReplayIntegrationTest.java | 5 +- .../FullReindexRecoveryIntegrationTest.java | 321 ++++++++++++++ .../QuickSearchQBitConfigValidationTest.java | 14 + .../processes/BasepullIndexStepTest.java | 68 ++- .../processes/FullReindexStepTest.java | 412 +++++++++++++++++- 11 files changed, 1303 insertions(+), 31 deletions(-) create mode 100644 src/test/java/com/kingsrook/qbits/quicksearch/FullReindexRecoveryIntegrationTest.java diff --git a/README.md b/README.md index b6758ed..0339ebb 100644 --- a/README.md +++ b/README.md @@ -128,9 +128,9 @@ Search runs as the current `QContext` session: tables the session cannot read ar | Process | Trigger | What it does | |---|---|---| -| Quick Search Basepull Index | Scheduled every `basepullRepeatSeconds` when `schedulerName` is set; also on demand | Replays failed real-time events, re-indexes rows changed since the last run start minus `basepullOverlapSeconds`, purges old run history | +| Quick Search Basepull Index | Scheduled every `basepullRepeatSeconds` when `schedulerName` is set; also on demand | Recovers a full reindex that stopped without finishing, replays failed real-time events, re-indexes rows changed since the last run start minus `basepullOverlapSeconds`, purges old run history | | Quick Search Reconcile Index | Cron via `reconcileCronExpression` when `schedulerName` is set; also on demand (optional `tableName` input) | Re-indexes everything in place, removes documents without a source row and documents of unconfigured tables | -| Quick Search Full Reindex | On demand (optional `tableName` input) | All tables: build a fresh physical index and swap the alias. One table: reconcile algorithm in place | +| Quick Search Full Reindex | On demand (optional `tableName` input) | All tables: build a fresh physical index and swap the alias. One table: reconcile algorithm in place. Fails to start while another full reindex is running | Without `schedulerName` the processes exist but nothing runs them; the QBit logs a warning at startup. @@ -155,6 +155,7 @@ Without `schedulerName` the processes exist but nothing runs them; the QBit logs | `basepullOverlapSeconds` | 300 | Re-read window before the previous run start | | `defaultBasepullIntervalMinutes` | 5 | Per-table due interval for config-driven tables | | `runHistoryRetentionDays` | 30 | Purge run records older than this; null keeps them | +| `fullReindexStaleMinutes` | 30 | A running full reindex that shows no sign of life for this long is treated as stopped: it no longer blocks a new one, and basepull recovers its tables. Keep it well above the time one page of `sourceBatchSize` records takes; null means the default | | `bulkBatchSize`, `maxBulkRequestBytes`, `sourceBatchSize` | 500, 5 MiB, 1000 | Batch sizes | | `maxFieldLength` | 10000 | Truncate each indexed value | | `maxSearchLimit` | 100 | Largest page size | @@ -171,7 +172,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). 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`. +- `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. Only one full reindex runs at a time: starting one while another is running fails. A full reindex shows it is running through its `FULL_REINDEX` run records, which stay `RUNNING` with `modifyDate` refreshed after every page. If it is killed mid-run (pod restart, OOM, deploy), recovery is automatic once its run records have shown no sign of life for `fullReindexStaleMinutes` (30 by default): the next basepull, or the next full reindex, marks those run records `FAILED`, returns the tables to `ACTIVE` with `lastBasepullTime` moved back to the run's start, and turns the `AWAITING_REINDEX` rows `PENDING` so basepull replays the deletes. No data is lost. A run that was only slow and is judged stopped notices before its alias swap and fails without swapping. ## Extending diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfig.java b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfig.java index 01b3da3..c726f08 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfig.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfig.java @@ -61,6 +61,15 @@ public class QuickSearchQBitConfig implements QBitConfig public static final String DEFAULT_ADMIN_PERMISSION_BASE_NAME = "quickSearchAdmin"; public static final String DEFAULT_BASEPULL_TIMESTAMP_FIELD = "modifyDate"; + /////////////////////////////////////////////////////////////////////////////// + // a live full reindex refreshes its run record after every page; a page is // + // one source query plus sourceBatchSize / bulkBatchSize bulk requests, each // + // bounded by the transport timeouts (a few minutes at most by default), // + // so 30 minutes leaves a wide margin before a slow run is judged dead, and // + // basepull (every 5 minutes by default) recovers a dead one soon after // + /////////////////////////////////////////////////////////////////////////////// + public static final Integer DEFAULT_FULL_REINDEX_STALE_MINUTES = 30; + private static final Pattern INDEX_NAME_PATTERN = Pattern.compile("^[a-z0-9][a-z0-9_+.-]*$"); private static final Set AWS_SERVICE_NAMES = Set.of("es", "aoss"); @@ -95,6 +104,7 @@ public class QuickSearchQBitConfig implements QBitConfig private Integer basepullOverlapSeconds = 300; private Integer defaultBasepullIntervalMinutes = 5; private Integer runHistoryRetentionDays = 30; + private Integer fullReindexStaleMinutes = DEFAULT_FULL_REINDEX_STALE_MINUTES; private Integer bulkBatchSize = 500; private Integer sourceBatchSize = 1000; private Integer maxFieldLength = 10000; @@ -389,6 +399,7 @@ public class QuickSearchQBitConfig implements QBitConfig requirePositive(maxSearchLimit, "maxSearchLimit", errors); requirePositive(maxFieldLength, "maxFieldLength", errors); requirePositive(runHistoryRetentionDays, "runHistoryRetentionDays", errors); + requirePositive(fullReindexStaleMinutes, "fullReindexStaleMinutes", errors); if(basepullOverlapSeconds != null && basepullOverlapSeconds < 0) { @@ -558,6 +569,17 @@ public class QuickSearchQBitConfig implements QBitConfig + /******************************************************************************* + ** fullReindexStaleMinutes, defaulting to DEFAULT_FULL_REINDEX_STALE_MINUTES. + *******************************************************************************/ + @JsonIgnore + public Integer getEffectiveFullReindexStaleMinutes() + { + return (fullReindexStaleMinutes == null ? DEFAULT_FULL_REINDEX_STALE_MINUTES : fullReindexStaleMinutes); + } + + + /******************************************************************************* ** *******************************************************************************/ @@ -1738,6 +1760,38 @@ public class QuickSearchQBitConfig implements QBitConfig + /******************************************************************************* + ** Getter for fullReindexStaleMinutes: a RUNNING full reindex whose run record shows no sign of life (start or heartbeat) for this many minutes is treated as stopped; null means the default + *******************************************************************************/ + public Integer getFullReindexStaleMinutes() + { + return (this.fullReindexStaleMinutes); + } + + + + /******************************************************************************* + ** Setter for fullReindexStaleMinutes + *******************************************************************************/ + public void setFullReindexStaleMinutes(Integer fullReindexStaleMinutes) + { + this.fullReindexStaleMinutes = fullReindexStaleMinutes; + } + + + + /******************************************************************************* + ** Fluent setter for fullReindexStaleMinutes + *******************************************************************************/ + public QuickSearchQBitConfig withFullReindexStaleMinutes(Integer fullReindexStaleMinutes) + { + this.fullReindexStaleMinutes = fullReindexStaleMinutes; + return (this); + } + + + + /******************************************************************************* ** Getter for bulkBatchSize: documents per OpenSearch bulk request *******************************************************************************/ 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 83ddf12..4bb6ea2 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java @@ -376,6 +376,26 @@ public abstract class AbstractIndexingStep implements BackendStep + /******************************************************************************* + ** Refresh the run record's modifyDate: the heartbeat by which other nodes + ** tell a RUNNING full reindex that is still going from one that stopped + ** (see FullReindexStep.recoverDeadRun). Does nothing without a run. + *******************************************************************************/ + protected void heartbeat(QuickSearchIndexRun run) throws QException + { + if(run == null || run.getId() == null) + { + return; + } + + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(getConfig().getQuickSearchIndexRunTableName()); + updateInput.setRecords(List.of(new QRecord().withValue("id", run.getId()).withValue("modifyDate", Instant.now()))); + new UpdateAction().execute(updateInput); + } + + + /******************************************************************************* ** Update fields on a quickSearchIndex row. *******************************************************************************/ @@ -429,9 +449,10 @@ public abstract class AbstractIndexingStep implements BackendStep /******************************************************************************* ** Index every source record matching the criteria into targetIndex (null ** for the alias), paging by primary key so each record is read once even - ** while rows are inserted or deleted. + ** while rows are inserted or deleted. The run's heartbeat is refreshed + ** after every page. *******************************************************************************/ - protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List extraCriteria, String targetIndex) throws QException + protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException { QuickSearchQBitConfig config = getConfig(); String tableName = tableConfig.getTableName(); @@ -491,6 +512,8 @@ public abstract class AbstractIndexingStep implements BackendStep { throw new QException("Source record has no value in primary key field [" + primaryKeyField + "]"); } + + heartbeat(run); } return (new IndexCounts(processed, indexed, errors, skipped, firstError)); diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStep.java b/src/main/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStep.java index 9b98c93..0183818 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStep.java @@ -55,8 +55,10 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair; /******************************************************************************* ** Scheduled catch-up. For each enabled table that is due, re-indexes records ** whose basepull timestamp is at or after the previous run's start minus the - ** configured overlap, paging by primary key. First replays failed real-time - ** events, and afterwards purges old run history. + ** configured overlap, paging by primary key. First recovers what a full + ** reindex that stopped without finishing left behind (so its captured deletes + ** are replayed in the same run) and replays failed real-time events, and + ** afterwards purges old run history. ** ** The watermark stored on the quickSearchIndex row is the run's start time, ** captured before the first query, so rows modified during the run are read @@ -83,6 +85,16 @@ public class BasepullIndexStep extends AbstractIndexingStep { List failures = new ArrayList<>(); + try + { + new FullReindexStep().recoverDeadRun(true); + } + catch(Exception e) + { + LOG.warn("Recovering from a stopped full reindex did not complete", e); + failures.add("full reindex recovery: " + e.getMessage()); + } + try { replayFailedEvents(output); @@ -211,7 +223,7 @@ public class BasepullIndexStep extends AbstractIndexingStep criteria.add(new QFilterCriteria(timestampField, QCriteriaOperator.GREATER_THAN_OR_EQUALS, since)); } - IndexCounts counts = indexAllRecords(client, tableConfig, criteria, null); + IndexCounts counts = indexAllRecords(client, tableConfig, criteria, null, run); if(counts.errors() == 0) { 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 7c6efae..a22086c 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java @@ -19,12 +19,15 @@ package com.kingsrook.qbits.quicksearch.processes; import java.io.Serializable; import java.time.Instant; +import java.time.temporal.ChronoUnit; import java.util.ArrayList; +import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction; import com.kingsrook.qqq.backend.core.actions.tables.QueryAction; @@ -71,6 +74,14 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair; ** AWAITING_REINDEX failed event. Once the rebuild's own writes are done and ** the alias is swapped, those deletes are applied through the alias. ** + ** Only one full reindex runs at a time. Each table's run record is RUNNING + ** before any table is marked REBUILDING, and its modifyDate is refreshed + ** after every page; a run whose records show no sign of life for + ** fullReindexStaleMinutes has stopped (killed JVM, lost node). Every full + ** reindex, and every basepull, first recovers what such a run left behind + ** (see recoverDeadRun); a full reindex then refuses to start while another + ** one is alive. + ** ** With a "tableName" input, that table is rebuilt in place with the reconcile ** algorithm (index everything, then delete what was not re-indexed). *******************************************************************************/ @@ -88,12 +99,18 @@ public class FullReindexStep extends AbstractIndexingStep @Override public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException { - String tableNameFilter = input.getValueString("tableName"); + String tableNameFilter = input.getValueString("tableName"); + boolean allTables = tableNameFilter == null || tableNameFilter.isBlank(); + + //////////////////////////////////////////////////////////////////////////// + // an all-tables rebuild applies the captured deletes itself after its swap // + //////////////////////////////////////////////////////////////////////////// + refuseIfAlive(recoverDeadRun(!allTables)); List tables = new ArrayList<>(); for(QuickSearchableTableConfig tableConfig : getDiscoveredTables()) { - if(tableNameFilter != null && !tableNameFilter.isBlank() && !tableNameFilter.equals(tableConfig.getTableName())) + if(!allTables && !tableNameFilter.equals(tableConfig.getTableName())) { continue; } @@ -108,7 +125,7 @@ public class FullReindexStep extends AbstractIndexingStep } } - if(tableNameFilter != null && !tableNameFilter.isBlank()) + if(!allTables) { Integer recordsIndexed = 0; for(QuickSearchableTableConfig tableConfig : tables) @@ -134,29 +151,42 @@ public class FullReindexStep extends AbstractIndexingStep QuickSearchOpenSearchClient client = getClient(); String physicalIndex = client.newPhysicalIndexName(); - 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<>(); Map priorStatus = new LinkedHashMap<>(); Integer indexed = 0; + boolean created = false; + boolean marked = false; boolean swapped = false; try { + ///////////////////////////////////////////////////////////////////////// + // the run records exist before any table is REBUILDING, so whoever // + // sees a table REBUILDING also sees this run alive; then make sure no // + // other run started at the same moment // + ///////////////////////////////////////////////////////////////////////// + for(QuickSearchableTableConfig tableConfig : tables) + { + runs.put(tableConfig, createRunRecord(queryIndexId(tableConfig.getTableName()), RUN_TYPE)); + } + refuseIfAlive(queryLiveRunsOtherThan(runs.values())); + + client.createPhysicalIndex(physicalIndex); + created = true; + LOG.info("Rebuilding Quick Search index", logPair("physicalIndex", physicalIndex), logPair("tables", tables.size())); + + marked = true; markRebuilding(tables, priorStatus); for(QuickSearchableTableConfig tableConfig : tables) { - String tableName = tableConfig.getTableName(); - QuickSearchIndexRun run = createRunRecord(queryIndexId(tableName), RUN_TYPE); - runs.put(tableConfig, run); + String tableName = tableConfig.getTableName(); startTimes.put(tableName, Instant.now()); - IndexCounts tableCounts = indexAllRecords(client, tableConfig, List.of(), physicalIndex); + IndexCounts tableCounts = indexAllRecords(client, tableConfig, List.of(), physicalIndex, runs.get(tableConfig)); counts.put(tableName, tableCounts); indexed = indexed + tableCounts.indexed(); @@ -167,6 +197,7 @@ public class FullReindexStep extends AbstractIndexingStep } client.refreshIndex(physicalIndex); + verifyStillOwned(tables, runs.values()); client.swapAliasTo(physicalIndex); swapped = true; @@ -188,17 +219,26 @@ public class FullReindexStep extends AbstractIndexingStep 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); } + /////////////////////////////////////////////////////////////////////////// + // the run records stay RUNNING until the captured deletes are applied, // + // so a basepull meanwhile does not take them for ones left behind // + /////////////////////////////////////////////////////////////////////////// applyCapturedDeletes(client, tables); + for(QuickSearchableTableConfig tableConfig : tables) + { + IndexCounts tableCounts = counts.get(tableConfig.getTableName()); + completeRunRecord(runs.get(tableConfig), RUN_COMPLETED, tableCounts.processed(), tableCounts.indexed(), 0, null); + } + LOG.info("Full reindex complete", logPair("physicalIndex", physicalIndex), logPair("recordsIndexed", indexed)); return (indexed); } catch(Exception e) { + boolean takenOver = marked && isTakenOverOrUnknown(runs.values()); LOG.warn(swapped ? "Full reindex failed after the alias swap; the new index stays in service" : "Full reindex failed; the previous index stays in service", e, logPair("physicalIndex", physicalIndex)); for(Map.Entry entry : runs.entrySet()) @@ -211,7 +251,7 @@ public class FullReindexStep extends AbstractIndexingStep e.getMessage()); } - if(!swapped) + if(created && !swapped) { try { @@ -223,8 +263,16 @@ public class FullReindexStep extends AbstractIndexingStep } } - restoreStatus(priorStatus); - applyCapturedDeletes(client, tables); + ////////////////////////////////////////////////////////////////////////// + // a recovery that judged this run stopped has already put the tables // + // back and handed the captured deletes to basepull, and a newer run // + // may own both by now // + ////////////////////////////////////////////////////////////////////////// + if(marked && !takenOver) + { + restoreStatus(priorStatus); + applyCapturedDeletes(client, tables); + } if(e instanceof QException qException) { @@ -275,6 +323,329 @@ public class FullReindexStep extends AbstractIndexingStep + /******************************************************************************* + ** Recover what a full reindex that stopped without finishing (killed JVM, + ** lost node) left behind, unless a full reindex is alive. Returns the live + ** RUNNING FULL_REINDEX run records, and changes nothing when there are any. + ** + ** A run is alive while the startTime or modifyDate (its heartbeat) of one of + ** its RUNNING records is within fullReindexStaleMinutes. When none is + ** alive: each stale RUNNING record is marked FAILED; each REBUILDING table + ** goes back to ACTIVE (the status it had before is not stored; drift + ** detection marks NEEDS_REINDEX again where it applies), with its + ** lastBasepullTime moved back to the stopped run's start, so basepull + ** re-reads updates that a swap may have left out; and, with + ** handCapturedDeletesToBasepull, every AWAITING_REINDEX row becomes PENDING. + ** Replaying those deletes is harmless whether or not the stopped run + ** swapped: before a swap the alias points at the old index, which the + ** listener already deleted from (a missing document counts as success); + ** after one, it removes the copy the run read before the delete. + ** + ** The REBUILDING rows and captured rows are read before the run records. A + ** run creates its records before it marks a table, and the listener + ** captures a delete only once it sees REBUILDING, so anything read here + ** that belongs to a live run comes with a live record, and nothing of a + ** live run is touched. A run that starts after the records are read can + ** have a table it just marked put back to ACTIVE, and a run misjudged as + ** stopped finds its records FAILED; both check before their swap and fail + ** without swapping, so no delete is lost. + *******************************************************************************/ + List recoverDeadRun(boolean handCapturedDeletesToBasepull) throws QException + { + List rebuildingRows = queryRebuildingIndexRows(); + Serializable lastCapturedId = handCapturedDeletesToBasepull ? queryLastCapturedDeleteId() : null; + List runningRuns = queryRunningRuns(); + + Instant cutoff = staleCutoff(); + List alive = runningRuns.stream().filter(run -> isAlive(run, cutoff)).toList(); + if(!alive.isEmpty() || (rebuildingRows.isEmpty() && runningRuns.isEmpty() && lastCapturedId == null)) + { + return (alive); + } + + LOG.warn("Recovering what a full reindex that stopped without finishing left behind", + logPair("staleRunIds", runningRuns.stream().map(run -> run.getValue("id")).toList()), + logPair("tableNames", rebuildingRows.stream().map(row -> row.getValueString("tableName")).toList())); + + for(QRecord runRecord : runningRuns) + { + QuickSearchIndexRun run = new QuickSearchIndexRun() + .withId(runRecord.getValueInteger("id")) + .withQuickSearchIndexId(runRecord.getValueInteger("quickSearchIndexId")) + .withRunType(RUN_TYPE); + completeRunRecord(run, RUN_FAILED, 0, 0, 1, "Full reindex stopped without finishing (no heartbeat since " + lastSeen(runRecord) + "); recovered automatically"); + } + + for(QRecord row : rebuildingRows) + { + Map values = new HashMap<>(); + values.put("status", STATUS_ACTIVE); + + Instant runStart = earliestStart(runningRuns, row.getValueInteger("id")); + Instant lastBasepullTime = row.getValueInstant("lastBasepullTime"); + if(runStart != null && lastBasepullTime != null && runStart.isBefore(lastBasepullTime)) + { + values.put("lastBasepullTime", runStart); + } + updateIndexRow(row.getValueInteger("id"), values); + } + + if(lastCapturedId != null) + { + handCapturedDeletesToBasepull(lastCapturedId); + } + return (alive); + } + + + + /******************************************************************************* + ** Make the AWAITING_REINDEX rows up to lastCapturedId PENDING, for basepull + ** to replay. Rows are read in id order past the previous batch, so a row + ** that could not be updated is not read again. + *******************************************************************************/ + private void handCapturedDeletesToBasepull(Serializable lastCapturedId) throws QException + { + QuickSearchQBitConfig config = getConfig(); + String failedEventTable = config.applyPrefix(QuickSearchFailedEvent.TABLE_NAME); + Serializable afterId = null; + Integer handedOver = 0; + + while(true) + { + QQueryFilter filter = new QQueryFilter( + new QFilterCriteria("status", QCriteriaOperator.EQUALS, QuickSearchFailedEvent.STATUS_AWAITING_REINDEX), + new QFilterCriteria("id", QCriteriaOperator.LESS_THAN_OR_EQUALS, lastCapturedId)) + .withOrderBy(new QFilterOrderBy("id", true)) + .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; + } + + afterId = rows.get(rows.size() - 1).getValue("id"); + List updates = new ArrayList<>(); + rows.forEach(row -> updates.add(new QRecord().withValue("id", row.getValue("id")).withValue("status", QuickSearchFailedEvent.STATUS_PENDING))); + + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(failedEventTable); + updateInput.setRecords(updates); + new UpdateAction().execute(updateInput); + handedOver = handedOver + rows.size(); + } + + LOG.info("Handed deletes captured during a stopped full reindex to basepull", logPair("count", handedOver)); + } + + + + /******************************************************************************* + ** Fail when another full reindex is alive. + *******************************************************************************/ + private void refuseIfAlive(List alive) throws QException + { + if(alive.isEmpty()) + { + return; + } + + QRecord run = alive.get(0); + throw (new QException("A full reindex is already running (run " + run.getValue("id") + ", last heartbeat " + lastSeen(run) + "); try again once it finishes. " + + "A run with no heartbeat for " + getConfig().getEffectiveFullReindexStaleMinutes() + " minutes is treated as stopped and recovered automatically.")); + } + + + + /******************************************************************************* + ** Just before the swap, fail unless this run still owns its tables: its + ** run records are still RUNNING (a recovery that judged it stopped marks + ** them FAILED) and every table is still REBUILDING (otherwise a delete + ** since then may have reached only the old index, uncaptured). + *******************************************************************************/ + private void verifyStillOwned(List tables, Collection runs) throws QException + { + if(isTakenOver(runs)) + { + throw (new QException("Full reindex was recovered as stopped (no heartbeat for " + getConfig().getEffectiveFullReindexStaleMinutes() + " minutes) while still running; the alias was not swapped")); + } + + for(QuickSearchableTableConfig tableConfig : tables) + { + String status = queryIndexRow(tableConfig.getTableName()).getValueString("status"); + if(!STATUS_REBUILDING.equals(status)) + { + throw (new QException("Table [" + tableConfig.getTableName() + "] is no longer REBUILDING (now " + status + "), so a delete since then may not have been captured; the alias was not swapped")); + } + } + } + + + + /******************************************************************************* + ** Whether a recovery has marked any of the run records finished. + *******************************************************************************/ + private boolean isTakenOver(Collection runs) throws QException + { + if(runs.isEmpty()) + { + return (false); + } + + List ids = new ArrayList<>(); + runs.forEach(run -> ids.add(run.getId())); + + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(getConfig().getQuickSearchIndexRunTableName()); + queryInput.setFilter(new QQueryFilter(new QFilterCriteria("id", QCriteriaOperator.IN, ids))); + return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords()).stream() + .anyMatch(run -> !RUN_RUNNING.equals(run.getValueString("status")))); + } + + + + /******************************************************************************* + ** isTakenOver for the failure path: never throws, and answers true when it + ** cannot tell, which leaves the tables to the next recovery. + *******************************************************************************/ + private boolean isTakenOverOrUnknown(Collection runs) + { + try + { + return (isTakenOver(runs)); + } + catch(Exception e) + { + LOG.warn("Could not read whether a recovery took over the failed full reindex; leaving its tables to the next recovery", e); + return (true); + } + } + + + + /******************************************************************************* + ** The live RUNNING FULL_REINDEX run records other than the given ones. + *******************************************************************************/ + private List queryLiveRunsOtherThan(Collection own) throws QException + { + Set ownIds = new HashSet<>(); + own.forEach(run -> ownIds.add(run.getId())); + + Instant cutoff = staleCutoff(); + return (queryRunningRuns().stream() + .filter(run -> !ownIds.contains(run.getValueInteger("id")) && isAlive(run, cutoff)) + .toList()); + } + + + + /******************************************************************************* + ** Every RUNNING FULL_REINDEX run record, oldest first. + *******************************************************************************/ + private List queryRunningRuns() throws QException + { + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(getConfig().getQuickSearchIndexRunTableName()); + queryInput.setFilter(new QQueryFilter( + new QFilterCriteria("runType", QCriteriaOperator.EQUALS, RUN_TYPE), + new QFilterCriteria("status", QCriteriaOperator.EQUALS, RUN_RUNNING)) + .withOrderBy(new QFilterOrderBy("id", true))); + return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords())); + } + + + + /******************************************************************************* + ** Every quickSearchIndex row that is REBUILDING. + *******************************************************************************/ + private List queryRebuildingIndexRows() throws QException + { + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(getConfig().getQuickSearchIndexTableName()); + queryInput.setFilter(new QQueryFilter(new QFilterCriteria("status", QCriteriaOperator.EQUALS, STATUS_REBUILDING))); + return (CollectionUtils.nonNullList(new QueryAction().execute(queryInput).getRecords())); + } + + + + /******************************************************************************* + ** The id of the newest AWAITING_REINDEX row, or null. Ids grow in insert + ** order, so a row captured after this read is beyond it. + *******************************************************************************/ + private Serializable queryLastCapturedDeleteId() throws QException + { + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME)); + queryInput.setFilter(new QQueryFilter(new QFilterCriteria("status", QCriteriaOperator.EQUALS, QuickSearchFailedEvent.STATUS_AWAITING_REINDEX)) + .withOrderBy(new QFilterOrderBy("id", false)) + .withLimit(1)); + List rows = new QueryAction().execute(queryInput).getRecords(); + return (CollectionUtils.nullSafeIsEmpty(rows) ? null : rows.get(0).getValue("id")); + } + + + + /******************************************************************************* + ** Run records last seen before this are stale. + *******************************************************************************/ + private Instant staleCutoff() throws QException + { + return (Instant.now().minus(getConfig().getEffectiveFullReindexStaleMinutes(), ChronoUnit.MINUTES)); + } + + + + /******************************************************************************* + ** Whether a run record was last seen after the cutoff. + *******************************************************************************/ + private static boolean isAlive(QRecord run, Instant cutoff) + { + Instant lastSeen = lastSeen(run); + return (lastSeen != null && lastSeen.isAfter(cutoff)); + } + + + + /******************************************************************************* + ** A run record's latest sign of life: its startTime or its heartbeat + ** (modifyDate), whichever is later; null when it has neither. + *******************************************************************************/ + private static Instant lastSeen(QRecord run) + { + Instant startTime = run.getValueInstant("startTime"); + Instant modifyDate = run.getValueInstant("modifyDate"); + if(startTime == null || modifyDate == null) + { + return (startTime == null ? modifyDate : startTime); + } + return (startTime.isAfter(modifyDate) ? startTime : modifyDate); + } + + + + /******************************************************************************* + ** The earliest startTime among the run records of an index row, or null. + *******************************************************************************/ + private static Instant earliestStart(List runs, Integer indexId) + { + return (runs.stream() + .filter(run -> Objects.equals(indexId, run.getValueInteger("quickSearchIndexId"))) + .map(run -> run.getValueInstant("startTime")) + .filter(Objects::nonNull) + .min(Instant::compareTo) + .orElse(null)); + } + + + /******************************************************************************* ** Apply the deletes the listener captured during the rebuild (failed-event ** rows with status AWAITING_REINDEX) through the alias, after the rebuild's @@ -285,8 +656,9 @@ public class FullReindexStep extends AbstractIndexingStep ** ** 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. + ** exceptions) is not read again; such rows stay AWAITING_REINDEX until the + ** next full reindex, or a basepull that finds none running, picks them up. + ** Never throws. *******************************************************************************/ void applyCapturedDeletes(QuickSearchOpenSearchClient client, List tables) { 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 d01c61c..8955088 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStep.java @@ -132,7 +132,7 @@ public class ReconcileIndexStep extends AbstractIndexingStep try { Instant reconcileStart = Instant.now(); - IndexCounts counts = indexAllRecords(client, tableConfig, List.of(), null); + IndexCounts counts = indexAllRecords(client, tableConfig, List.of(), null, run); Long documentsRemoved = 0L; if(counts.errors() == 0) diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/FullReindexDeleteReplayIntegrationTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/FullReindexDeleteReplayIntegrationTest.java index c4e526b..6f318aa 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/FullReindexDeleteReplayIntegrationTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/FullReindexDeleteReplayIntegrationTest.java @@ -47,6 +47,7 @@ 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.model.QuickSearchIndexRun; import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient; import com.kingsrook.qbits.quicksearch.processes.FullReindexStep; import org.junit.jupiter.api.AfterAll; @@ -176,9 +177,9 @@ class FullReindexDeleteReplayIntegrationTest // while the alias still points at the old one // /////////////////////////////////////////////////////////////////////// @Override - protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List extraCriteria, String targetIndex) throws QException + protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException { - IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex); + IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex, run); DeleteInput deleteInput = new DeleteInput(); deleteInput.setTableName(TABLE_NAME); diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/FullReindexRecoveryIntegrationTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/FullReindexRecoveryIntegrationTest.java new file mode 100644 index 0000000..57b6d44 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/quicksearch/FullReindexRecoveryIntegrationTest.java @@ -0,0 +1,321 @@ +/* + * 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.time.Instant; +import java.time.temporal.ChronoUnit; +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.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.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.actions.tables.update.UpdateInput; +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.model.QuickSearchIndex; +import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun; +import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient; +import com.kingsrook.qbits.quicksearch.processes.AbstractIndexingStep; +import com.kingsrook.qbits.quicksearch.processes.BasepullIndexStep; +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; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + + +/******************************************************************************* + ** Integration test, against a real OpenSearch, that a full reindex killed + ** right after its alias swap (table left REBUILDING, RUNNING run record, a + ** captured delete never applied, so the deleted record is back in search) is + ** recovered by the scheduled basepull once its run record goes stale. + *******************************************************************************/ +@ExtendWith(RequiresDockerCondition.class) +@Testcontainers +class FullReindexRecoveryIntegrationTest +{ + private static final String BACKEND_NAME = "reindexRecoveryMemoryBackend"; + private static final String TABLE_NAME = "recoveryCustomer"; + private static final String INDEX_NAME = "quick_search_reindex_recovery_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 RecoveryCustomer + { + @QuickSearchField(weight = 2) + private String name; + + @QuickSearchField + private String city; + } + + + + /******************************************************************************* + ** Thrown to stop a full reindex the way a killed JVM does: it is not an + ** Exception, so none of the step's failure handling runs. + *******************************************************************************/ + static class SimulatedKill extends Error + { + } + + + + /******************************************************************************* + ** 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)) + .withField(new QFieldMetaData("modifyDate", QFieldType.DATE_TIME))); + + QuickSearchQBitConfig config = new QuickSearchQBitConfig() + .withBackendName(BACKEND_NAME) + .withOpensearchHost(opensearch.getHost()) + .withOpensearchPort(opensearch.getMappedPort(9200)) + .withOpensearchIndexName(INDEX_NAME) + .withSearchableEntityClasses(List.of(RecoveryCustomer.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 during a full reindex that is then killed right + ** after its swap is back in search (the new index holds the copy read + ** before the delete, and the captured delete was never applied). While the + ** dead run's record is fresh it blocks a new full reindex; once it is + ** stale, basepull recovers the table and replays the delete, and the + ** record is gone from search. + *******************************************************************************/ + @Test + void testKilledFullReindex_isRecoveredByBasepull() throws Exception + { + InsertInput insertInput = new InsertInput(); + insertInput.setTableName(TABLE_NAME); + insertInput.setRecords(List.of( + new QRecord().withValue("id", 1).withValue("name", "Gwendolyn").withValue("city", "Stalledton"), + new QRecord().withValue("id", 2).withValue("name", "Stanford").withValue("city", "Stalledton"))); + new InsertAction().execute(insertInput); + assertThat(searchRecordIds("stalledton")).containsExactlyInAnyOrder("1", "2"); + + assertThatThrownBy(() -> new FullReindexStep() + { + //////////////////////////////////////////////////////////////////// + // delete a record once its document is in the new physical index // + //////////////////////////////////////////////////////////////////// + @Override + protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException + { + IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex, run); + + DeleteInput deleteInput = new DeleteInput(); + deleteInput.setTableName(TABLE_NAME); + deleteInput.setPrimaryKeys(List.of(1)); + new DeleteAction().execute(deleteInput); + + return (counts); + } + + + + //////////////////////////////////////////////////////////////////// + // die right after the swap, before any table is finished // + //////////////////////////////////////////////////////////////////// + @Override + protected void updateIndexRow(Integer indexId, Map values) throws QException + { + if(values.containsKey("lastFullReindexTime")) + { + throw (new SimulatedKill()); + } + super.updateIndexRow(indexId, values); + } + }.run(new RunBackendStepInput(), new RunBackendStepOutput())) + .isInstanceOf(SimulatedKill.class); + + assertThat(indexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING); + assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).singleElement() + .satisfies(run -> assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING)); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).singleElement() + .satisfies(row -> assertThat(row.getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX)); + assertThat(searchRecordIds("stalledton")) + .as("the swapped-in index still holds the deleted record") + .containsExactlyInAnyOrder("1", "2"); + + assertThatThrownBy(() -> new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput())) + .as("the dead run's record is still fresh, so it counts as running") + .isInstanceOf(QException.class) + .hasMessageContaining("already running"); + + backdateRunRecords(); + new BasepullIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()); + + assertThat(searchRecordIds("stalledton")).containsExactly("2"); + assertThat(indexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty(); + assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)) + .filteredOn(run -> FullReindexStep.RUN_TYPE.equals(run.getValueString("runType"))) + .singleElement() + .satisfies(run -> + { + assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED); + assertThat(run.getValueString("errorMessage")).contains("stopped without finishing"); + }); + } + + + + /******************************************************************************* + ** Move the start and heartbeat of every RUNNING run record two hours into + ** the past, as if its run had stopped refreshing it. + *******************************************************************************/ + private void backdateRunRecords() throws QException + { + Instant twoHoursAgo = Instant.now().minus(2, ChronoUnit.HOURS); + for(QRecord run : queryAll(QuickSearchIndexRun.TABLE_NAME)) + { + if(AbstractIndexingStep.RUN_RUNNING.equals(run.getValueString("status"))) + { + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(QuickSearchIndexRun.TABLE_NAME); + updateInput.setOmitModifyDateUpdate(true); + updateInput.setRecords(List.of(new QRecord().withValue("id", run.getValue("id")).withValue("startTime", twoHoursAgo).withValue("modifyDate", twoHoursAgo))); + new UpdateAction().execute(updateInput); + } + } + } + + + + /******************************************************************************* + ** The table's quickSearchIndex row. + *******************************************************************************/ + private QRecord indexRow() throws QException + { + return (queryAll(QuickSearchIndex.TABLE_NAME).stream() + .filter(row -> TABLE_NAME.equals(row.getValueString("tableName"))) + .findFirst().orElseThrow()); + } + + + + /******************************************************************************* + ** All rows of a table. + *******************************************************************************/ + private List queryAll(String tableName) throws QException + { + QueryInput queryInput = new QueryInput(); + queryInput.setTableName(tableName); + return (new QueryAction().execute(queryInput).getRecords()); + } + + + + /******************************************************************************* + ** Refresh the index, then return the record IDs search returns for the term. + *******************************************************************************/ + private List searchRecordIds(String term) throws QException + { + QuickSearchQBitContext.getClient().refreshIndex(); + return (new QuickSearchAction().execute(new QuickSearchInput().withSearchTerm(term).withTableName(TABLE_NAME)) + .getResults().stream() + .map(QuickSearchResult::getRecordId) + .toList()); + } + +} diff --git a/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfigValidationTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfigValidationTest.java index de957c0..01cd253 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfigValidationTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfigValidationTest.java @@ -279,6 +279,7 @@ class QuickSearchQBitConfigValidationTest .withBasepullRepeatSeconds(0) .withBasepullOverlapSeconds(-5) .withRunHistoryRetentionDays(0) + .withFullReindexStaleMinutes(0) .withMaxFieldLength(0)); assertThat(errors).anyMatch(e -> e.startsWith("connectTimeoutMillis")); assertThat(errors).anyMatch(e -> e.startsWith("responseTimeoutMillis")); @@ -287,11 +288,23 @@ class QuickSearchQBitConfigValidationTest assertThat(errors).anyMatch(e -> e.startsWith("basepullRepeatSeconds")); assertThat(errors).anyMatch(e -> e.startsWith("basepullOverlapSeconds")); assertThat(errors).anyMatch(e -> e.startsWith("runHistoryRetentionDays")); + assertThat(errors).anyMatch(e -> e.startsWith("fullReindexStaleMinutes")); assertThat(errors).anyMatch(e -> e.startsWith("maxFieldLength")); } + @Test + void testFullReindexStaleMinutes_nullIsValid_andMeansTheDefault() + { + QuickSearchQBitConfig config = baseConfig().withFullReindexStaleMinutes(null); + assertThat(validate(config)).isEmpty(); + assertThat(config.getEffectiveFullReindexStaleMinutes()).isEqualTo(QuickSearchQBitConfig.DEFAULT_FULL_REINDEX_STALE_MINUTES); + assertThat(config.withFullReindexStaleMinutes(5).getEffectiveFullReindexStaleMinutes()).isEqualTo(5); + } + + + @Test void testValidate_nullInstance_isSafe() { @@ -337,6 +350,7 @@ class QuickSearchQBitConfigValidationTest assertThat(config.getResponseTimeoutMillis()).isEqualTo(60000); assertThat(config.getMaxSearchLimit()).isEqualTo(100); assertThat(config.getBasepullOverlapSeconds()).isEqualTo(300); + assertThat(config.getFullReindexStaleMinutes()).isEqualTo(30); assertThat(config.getAdminPermissionRules().getPermissionBaseName()).isEqualTo(QuickSearchQBitConfig.DEFAULT_ADMIN_PERMISSION_BASE_NAME); assertThat(config.getApplyRecordSecurityLocks()).isTrue(); assertThat(config.getDefaultBackendNameForTables()).isNull(); 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 2e41682..9bd9b66 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java @@ -224,6 +224,34 @@ class BasepullIndexStepTest extends BaseQuickSearchTest + /******************************************************************************* + ** Insert a RUNNING FULL_REINDEX run record for the index row whose start + ** and last heartbeat were the given number of minutes ago. + *******************************************************************************/ + private QRecord insertRunningFullReindex(QRecord indexRow, long minutesAgo) throws QException + { + InsertInput input = new InsertInput(); + input.setTableName(QuickSearchIndexRun.TABLE_NAME); + input.setRecords(List.of(new QRecord() + .withValue("quickSearchIndexId", indexRow.getValue("id")) + .withValue("runType", FullReindexStep.RUN_TYPE) + .withValue("status", AbstractIndexingStep.RUN_RUNNING))); + QRecord run = new InsertAction().execute(input).getRecords().get(0); + + //////////////////////////////////////////////////////////////////// + // set both times without QQQ stamping modifyDate with the present // + //////////////////////////////////////////////////////////////////// + Instant lastSeen = Instant.now().minus(minutesAgo, ChronoUnit.MINUTES); + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(QuickSearchIndexRun.TABLE_NAME); + updateInput.setOmitModifyDateUpdate(true); + updateInput.setRecords(List.of(new QRecord().withValue("id", run.getValue("id")).withValue("startTime", lastSeen).withValue("modifyDate", lastSeen))); + new UpdateAction().execute(updateInput); + return (run); + } + + + /******************************************************************************* ** Query all rows of a table. *******************************************************************************/ @@ -923,7 +951,8 @@ class BasepullIndexStepTest extends BaseQuickSearchTest @Test void testReplayFailedEvents_awaitingReindexRowsAreNotReplayed() throws QException { - insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60); + QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60); + insertRunningFullReindex(row, 1); QRecord captured = insertFailedEvent(TEST_ENTITY_TABLE, "7", "DELETE", 0); UpdateInput updateInput = new UpdateInput(); updateInput.setTableName(QuickSearchFailedEvent.TABLE_NAME); @@ -940,6 +969,42 @@ class BasepullIndexStepTest extends BaseQuickSearchTest + /******************************************************************************* + ** Test: a full reindex that stopped without finishing (table left + ** REBUILDING, run record without a heartbeat for two hours) is recovered + ** by the scheduled basepull, which replays its captured deletes in the same + ** run. + *******************************************************************************/ + @Test + void testBasepull_recoversAStoppedFullReindex_andReplaysItsCapturedDeletes() throws QException + { + QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60); + QRecord deadRun = insertRunningFullReindex(row, 120); + 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); + 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); + + RunBackendStepOutput output = new RunBackendStepOutput(); + new BasepullIndexStep().run(new RunBackendStepInput(), output); + + verify(mockClient).deleteDocuments(List.of(TEST_ENTITY_TABLE + ":7"), QuickSearchQBitContext.getConfig().getBulkBatchSize()); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty(); + assertThat(output.getValueInteger("failedEventsReplayed")).isEqualTo(1); + assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)) + .filteredOn(run -> deadRun.getValue("id").equals(run.getValue("id"))) + .singleElement() + .satisfies(run -> assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED)); + } + + + /******************************************************************************* ** Test: drift found while a full reindex is running does not replace the ** REBUILDING status the listener relies on. @@ -948,6 +1013,7 @@ class BasepullIndexStepTest extends BaseQuickSearchTest void testDriftDetection_duringFullReindex_keepsRebuilding() throws QException { QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60); + insertRunningFullReindex(row, 1); UpdateInput updateInput = new UpdateInput(); updateInput.setTableName(QuickSearchIndex.TABLE_NAME); updateInput.setRecords(List.of(new QRecord().withValue("id", row.getValue("id")) 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 902bc70..4c44651 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java @@ -17,9 +17,12 @@ package com.kingsrook.qbits.quicksearch.processes; +import java.time.Instant; +import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.List; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction; import com.kingsrook.qqq.backend.core.actions.tables.InsertAction; @@ -556,9 +559,9 @@ class FullReindexStepTest extends BaseQuickSearchTest new FullReindexStep() { @Override - protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List extraCriteria, String targetIndex) throws QException + protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException { - IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex); + IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex, run); statusesDuringRebuild.add(FullReindexStepTest.this.queryIndexRow().getValueString("status")); insertCapturedDelete("7"); return (counts); @@ -751,4 +754,409 @@ class FullReindexStepTest extends BaseQuickSearchTest assertThat(rows.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX); } + + + /******************************************************************************* + ** Insert a RUNNING FULL_REINDEX run record for the given index row, started + ** startedMinutesAgo and last refreshed (heartbeat) heartbeatMinutesAgo. + *******************************************************************************/ + private QRecord insertRunningRun(Integer quickSearchIndexId, long startedMinutesAgo, long heartbeatMinutesAgo) throws QException + { + InsertInput insertInput = new InsertInput(); + insertInput.setTableName(QuickSearchIndexRun.TABLE_NAME); + insertInput.setRecords(List.of(new QRecord() + .withValue("quickSearchIndexId", quickSearchIndexId) + .withValue("runType", FullReindexStep.RUN_TYPE) + .withValue("status", AbstractIndexingStep.RUN_RUNNING))); + QRecord run = new InsertAction().execute(insertInput).getRecords().get(0); + setRunTimes(run.getValueInteger("id"), Instant.now().minus(startedMinutesAgo, ChronoUnit.MINUTES), Instant.now().minus(heartbeatMinutesAgo, ChronoUnit.MINUTES)); + return (queryRun(run.getValueInteger("id"))); + } + + + + /******************************************************************************* + ** Move a run record's start and heartbeat two hours into the past, as if + ** its run had stopped refreshing it. + *******************************************************************************/ + private void backdateRun(Integer runId) throws QException + { + Instant twoHoursAgo = Instant.now().minus(2, ChronoUnit.HOURS); + setRunTimes(runId, twoHoursAgo, twoHoursAgo); + } + + + + /******************************************************************************* + ** Set a run record's startTime and modifyDate, without QQQ stamping the + ** modifyDate with the current time. + *******************************************************************************/ + private void setRunTimes(Integer runId, Instant startTime, Instant modifyDate) throws QException + { + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(QuickSearchIndexRun.TABLE_NAME); + updateInput.setOmitModifyDateUpdate(true); + updateInput.setRecords(List.of(new QRecord().withValue("id", runId).withValue("startTime", startTime).withValue("modifyDate", modifyDate))); + new UpdateAction().execute(updateInput); + } + + + + /******************************************************************************* + ** The run record with the given id. + *******************************************************************************/ + private QRecord queryRun(Integer runId) throws QException + { + return (queryAll(QuickSearchIndexRun.TABLE_NAME).stream() + .filter(run -> runId.equals(run.getValueInteger("id"))) + .findFirst().orElseThrow()); + } + + + + /******************************************************************************* + ** Set the status and lastBasepullTime on the testEntity quickSearchIndex row. + *******************************************************************************/ + private void updateIndexRow(String status, Instant lastBasepullTime) 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) + .withValue("lastBasepullTime", lastBasepullTime))); + new UpdateAction().execute(updateInput); + } + + + + /******************************************************************************* + ** Test: a full reindex whose run record is fresh blocks a second one, which + ** fails before it creates anything. + *******************************************************************************/ + @Test + void testLiveRun_blocksASecondRun() throws QException + { + insertTestEntities(1); + insertIndexRow(true); + insertRunningRun(queryIndexRow().getValueInteger("id"), 1, 1); + + assertThatThrownBy(() -> new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput())) + .isInstanceOf(QException.class) + .hasMessageContaining("already running"); + + verify(mockClient, never()).createPhysicalIndex(anyString()); + verify(mockClient, never()).indexDocuments(anyString(), anyList(), anyInt()); + assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).singleElement() + .satisfies(run -> assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING)); + } + + + + /******************************************************************************* + ** Test: a live full reindex also blocks a single-table full reindex. + *******************************************************************************/ + @Test + void testLiveRun_blocksASingleTableRun() throws QException + { + insertTestEntities(1); + insertIndexRow(true); + insertRunningRun(queryIndexRow().getValueInteger("id"), 1, 1); + + assertThatThrownBy(() -> new FullReindexStep().run(inputForTable(TEST_ENTITY_TABLE), new RunBackendStepOutput())) + .isInstanceOf(QException.class) + .hasMessageContaining("already running"); + + verify(mockClient, never()).indexDocuments(anyList(), anyInt()); + assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).hasSize(1); + } + + + + /******************************************************************************* + ** Test: a dead run (no heartbeat past the threshold) does not block a new + ** one; it is recovered first, and the deletes it captured wait for the new + ** run, which applies them after its swap. + *******************************************************************************/ + @Test + void testStaleRun_doesNotBlock_andIsRecovered() throws QException + { + insertTestEntities(2); + insertIndexRow(true); + updateIndexRowStatus(AbstractIndexingStep.STATUS_REBUILDING); + Integer staleRunId = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120).getValueInteger("id"); + insertCapturedDelete("9"); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1)); + + RunBackendStepOutput output = new RunBackendStepOutput(); + new FullReindexStep().run(new RunBackendStepInput(), output); + + assertThat(output.getValueBoolean("aliasSwapped")).isTrue(); + assertThat(queryRun(staleRunId).getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED); + assertThat(queryRun(staleRunId).getValueString("errorMessage")).contains("stopped without finishing"); + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + + InOrder inOrder = inOrder(mockClient); + inOrder.verify(mockClient).swapAliasTo(PHYSICAL_INDEX); + inOrder.verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":9")), anyInt()); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty(); + } + + + + /******************************************************************************* + ** Test: the run records stay RUNNING while the captured deletes are applied + ** after the swap, so a basepull meanwhile leaves those rows to this run. + *******************************************************************************/ + @Test + void testAllTables_runStaysRunningUntilCapturedDeletesAreApplied() throws QException + { + insertTestEntities(1); + insertCapturedDelete("9"); + + List runStatusesWhileApplying = new ArrayList<>(); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenAnswer(invocation -> + { + queryAll(QuickSearchIndexRun.TABLE_NAME).forEach(run -> runStatusesWhileApplying.add(run.getValueString("status"))); + return (new BulkIndexResult().withSuccessCount(1)); + }); + + new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()); + + assertThat(runStatusesWhileApplying).containsExactly(AbstractIndexingStep.RUN_RUNNING); + assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).singleElement() + .satisfies(run -> assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_COMPLETED)); + } + + + + /******************************************************************************* + ** Test: the run record's modifyDate is refreshed after every page, so a + ** long rebuild keeps showing it is alive. + *******************************************************************************/ + @Test + void testHeartbeat_refreshesTheRunRecordAfterEachPage() throws QException + { + QuickSearchQBitContext.getConfig().withSourceBatchSize(2); + insertTestEntities(5); + insertIndexRow(true); + Integer runId = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120).getValueInteger("id"); + + List heartbeatsSeenByEachBulk = new ArrayList<>(); + when(mockClient.indexDocuments(anyString(), anyList(), anyInt())).thenAnswer(invocation -> + { + heartbeatsSeenByEachBulk.add(queryRun(runId).getValueInstant("modifyDate")); + return (new BulkIndexResult().withSuccessCount(((List) invocation.getArgument(1)).size())); + }); + + Instant before = Instant.now(); + new FullReindexStep().indexAllRecords(mockClient, QuickSearchQBitContext.getTableConfig(TEST_ENTITY_TABLE), List.of(), PHYSICAL_INDEX, new QuickSearchIndexRun().withId(runId)); + + assertThat(heartbeatsSeenByEachBulk).hasSize(3); + assertThat(heartbeatsSeenByEachBulk.get(0)).isBefore(before); + assertThat(heartbeatsSeenByEachBulk.get(1)).isAfterOrEqualTo(before); + assertThat(queryRun(runId).getValueInstant("modifyDate")).isAfterOrEqualTo(heartbeatsSeenByEachBulk.get(2)); + } + + + + /******************************************************************************* + ** Test: a run that stalls past the threshold is recovered by another node + ** while still going. It notices before its swap and stops without swapping + ** and without touching the tables or the captured deletes, which the + ** recovery already handed to basepull. + *******************************************************************************/ + @Test + void testRunJudgedDead_recoveredWhileGoing_doesNotSwap() throws QException + { + insertTestEntities(1); + + assertThatThrownBy(() -> new FullReindexStep() + { + @Override + protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException + { + IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex, run); + insertCapturedDelete("7"); + backdateRun(run.getId()); + assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty(); + return (counts); + } + }.run(new RunBackendStepInput(), new RunBackendStepOutput())) + .isInstanceOf(QException.class) + .hasMessageContaining("recovered"); + + verify(mockClient, never()).swapAliasTo(anyString()); + verify(mockClient).deletePhysicalIndex(PHYSICAL_INDEX); + verify(mockClient, never()).deleteDocuments(anyList(), anyInt()); + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).singleElement() + .satisfies(row -> assertThat(row.getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING)); + } + + + + /******************************************************************************* + ** Test: a table that is no longer REBUILDING before the swap (a recovery + ** that read the run records just before this run created its own put it + ** back to ACTIVE) means a delete since may have reached only the old index, + ** so the run fails without swapping and applies its captured deletes to the + ** old index. + *******************************************************************************/ + @Test + void testTableNoLongerRebuildingBeforeTheSwap_doesNotSwap() throws QException + { + insertTestEntities(1); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1)); + + assertThatThrownBy(() -> new FullReindexStep() + { + @Override + protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List extraCriteria, String targetIndex, QuickSearchIndexRun run) throws QException + { + IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex, run); + insertCapturedDelete("7"); + updateIndexRowStatus(AbstractIndexingStep.STATUS_ACTIVE); + return (counts); + } + }.run(new RunBackendStepInput(), new RunBackendStepOutput())) + .isInstanceOf(QException.class) + .hasMessageContaining("no longer REBUILDING"); + + verify(mockClient, never()).swapAliasTo(anyString()); + verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":7")), anyInt()); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty(); + assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).singleElement() + .satisfies(run -> assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED)); + } + + + + /******************************************************************************* + ** Test: when another node starts a full reindex at the same moment (after + ** this one checked, before it created its run records), this one sees the + ** other's fresh run record and stops before it touches anything. + *******************************************************************************/ + @Test + void testSimultaneousStart_secondRunStopsBeforeMarking() throws QException + { + insertTestEntities(1); + AtomicBoolean otherStarted = new AtomicBoolean(false); + + assertThatThrownBy(() -> new FullReindexStep() + { + @Override + protected QuickSearchIndexRun createRunRecord(Integer indexId, String runType) throws QException + { + if(otherStarted.compareAndSet(false, true)) + { + insertRunningRun(indexId, 0, 0); + } + return (super.createRunRecord(indexId, runType)); + } + }.run(new RunBackendStepInput(), new RunBackendStepOutput())) + .isInstanceOf(QException.class) + .hasMessageContaining("already running"); + + verify(mockClient, never()).createPhysicalIndex(anyString()); + verify(mockClient, never()).swapAliasTo(anyString()); + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).extracting(run -> run.getValueString("status")) + .containsExactlyInAnyOrder(AbstractIndexingStep.RUN_RUNNING, AbstractIndexingStep.RUN_FAILED); + } + + + + /******************************************************************************* + ** Test: recovering a dead run marks its record FAILED, puts the table back + ** to ACTIVE with the basepull watermark rewound to the run's start, and + ** hands the captured deletes to basepull. + *******************************************************************************/ + @Test + void testRecovery_deadRun_restoresTableAndHandsCapturedDeletesToBasepull() throws QException + { + insertIndexRow(true); + updateIndexRow(AbstractIndexingStep.STATUS_REBUILDING, Instant.now().minus(5, ChronoUnit.MINUTES)); + QRecord deadRun = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 90); + insertCapturedDelete("4"); + insertCapturedDelete("5"); + + assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty(); + + QRecord row = queryIndexRow(); + assertThat(row.getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + assertThat(row.getValueInstant("lastBasepullTime")).isEqualTo(deadRun.getValueInstant("startTime")); + assertThat(row.getValueString("lastRunStatus")).isEqualTo("FULL_REINDEX FAILED"); + assertThat(row.getValueString("lastErrorMessage")).contains("stopped without finishing"); + + QRecord run = queryRun(deadRun.getValueInteger("id")); + assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED); + assertThat(run.getValue("endTime")).isNotNull(); + + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).extracting(event -> event.getValueString("status")) + .containsExactly(QuickSearchFailedEvent.STATUS_PENDING, QuickSearchFailedEvent.STATUS_PENDING); + } + + + + /******************************************************************************* + ** Test: a run started long ago but with a recent heartbeat is alive; + ** recovery returns it and leaves its table, run record and captured + ** deletes alone. + *******************************************************************************/ + @Test + void testRecovery_liveRun_leavesItsTableAndRowsAlone() throws QException + { + insertIndexRow(true); + updateIndexRowStatus(AbstractIndexingStep.STATUS_REBUILDING); + QRecord liveRun = insertRunningRun(queryIndexRow().getValueInteger("id"), 180, 1); + insertCapturedDelete("4"); + + assertThat(new FullReindexStep().recoverDeadRun(true)).extracting(run -> run.getValueInteger("id")) + .containsExactly(liveRun.getValueInteger("id")); + + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING); + assertThat(queryRun(liveRun.getValueInteger("id")).getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).singleElement() + .satisfies(event -> assertThat(event.getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX)); + } + + + + /******************************************************************************* + ** Test: a REBUILDING table whose run record is gone (purged) is recovered + ** too, without moving its basepull watermark. + *******************************************************************************/ + @Test + void testRecovery_withoutRunRecord_restoresTheTable() throws QException + { + Instant lastBasepullTime = Instant.now().minus(5, ChronoUnit.MINUTES); + insertIndexRow(true); + updateIndexRow(AbstractIndexingStep.STATUS_REBUILDING, lastBasepullTime); + insertCapturedDelete("4"); + + assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty(); + + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + assertThat(queryIndexRow().getValueInstant("lastBasepullTime")).isEqualTo(lastBasepullTime); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).singleElement() + .satisfies(event -> assertThat(event.getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING)); + } + + + + /******************************************************************************* + ** Test: with nothing left behind, recovery changes nothing. + *******************************************************************************/ + @Test + void testRecovery_nothingLeftBehind_changesNothing() throws QException + { + insertIndexRow(true); + + assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty(); + + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + assertThat(queryIndexRow().getValueString("lastRunStatus")).isNull(); + assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).isEmpty(); + } + } -- 2.54.0 From bdb50511697b9df973571c848331517c551cf3e0 Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 9 Oct 2026 18:38:33 -0500 Subject: [PATCH 2/3] fix(#24): keep a slow captured-delete drain from touching a newer run's rows The drain now heartbeats every batch, reads only rows up to the newest one at its start and stops once a recovery takes the run over; recovery marks dead runs FAILED last, basepull skips replayed deletes of re-created records, and the heartbeat is best effort and FULL_REINDEX only. Refs #24 --- README.md | 2 +- .../quicksearch/QuickSearchQBitConfig.java | 2 - .../processes/AbstractIndexingStep.java | 58 ++++- .../processes/BasepullIndexStep.java | 41 ++- .../processes/FullReindexStep.java | 165 ++++++------ .../processes/BasepullIndexStepTest.java | 23 ++ .../processes/FullReindexStepTest.java | 234 +++++++++++++++++- 7 files changed, 414 insertions(+), 111 deletions(-) diff --git a/README.md b/README.md index 0339ebb..4b901f3 100644 --- a/README.md +++ b/README.md @@ -172,7 +172,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). While a table is `REBUILDING`, drift detection is suspended for it. Only one full reindex runs at a time: starting one while another is running fails. A full reindex shows it is running through its `FULL_REINDEX` run records, which stay `RUNNING` with `modifyDate` refreshed after every page. If it is killed mid-run (pod restart, OOM, deploy), recovery is automatic once its run records have shown no sign of life for `fullReindexStaleMinutes` (30 by default): the next basepull, or the next full reindex, marks those run records `FAILED`, returns the tables to `ACTIVE` with `lastBasepullTime` moved back to the run's start, and turns the `AWAITING_REINDEX` rows `PENDING` so basepull replays the deletes. No data is lost. A run that was only slow and is judged stopped notices before its alias swap and fails without swapping. +- `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. Only one full reindex runs at a time: starting one while another is running fails. A full reindex shows it is running through its `FULL_REINDEX` run records, which stay `RUNNING` with `modifyDate` refreshed after every page. If it is killed mid-run (pod restart, OOM, deploy), recovery is automatic once its run records have shown no sign of life for `fullReindexStaleMinutes` (30 by default): the next basepull, or the next full reindex, marks those run records `FAILED`, returns the tables to `ACTIVE` with `lastBasepullTime` moved back to the run's start, and turns the `AWAITING_REINDEX` rows `PENDING` so basepull replays the deletes. No data is lost. A run that was only slow and is judged stopped notices before its alias swap and fails without swapping, or, after its swap, stops applying its captured deletes; basepull skips a replayed delete whose record exists again. ## Extending diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfig.java b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfig.java index c726f08..7b50e60 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfig.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchQBitConfig.java @@ -1759,7 +1759,6 @@ public class QuickSearchQBitConfig implements QBitConfig - /******************************************************************************* ** Getter for fullReindexStaleMinutes: a RUNNING full reindex whose run record shows no sign of life (start or heartbeat) for this many minutes is treated as stopped; null means the default *******************************************************************************/ @@ -1791,7 +1790,6 @@ public class QuickSearchQBitConfig implements QBitConfig - /******************************************************************************* ** Getter for bulkBatchSize: documents per OpenSearch bulk request *******************************************************************************/ 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 4bb6ea2..eb18608 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java @@ -23,8 +23,10 @@ import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import com.kingsrook.qqq.backend.core.actions.processes.BackendStep; import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction; import com.kingsrook.qqq.backend.core.actions.tables.InsertAction; @@ -377,21 +379,59 @@ public abstract class AbstractIndexingStep implements BackendStep /******************************************************************************* - ** Refresh the run record's modifyDate: the heartbeat by which other nodes - ** tell a RUNNING full reindex that is still going from one that stopped - ** (see FullReindexStep.recoverDeadRun). Does nothing without a run. + ** Refresh a FULL_REINDEX run record's modifyDate: the heartbeat by which + ** other nodes tell a full reindex that is still going from one that + ** stopped (see FullReindexStep.recoverDeadRun). Nothing reads it for other + ** run types. Best effort: a failure is logged and never stops the run (one + ** judged stopped as a result fails safely before its swap). *******************************************************************************/ - protected void heartbeat(QuickSearchIndexRun run) throws QException + protected void heartbeat(QuickSearchIndexRun run) { - if(run == null || run.getId() == null) + if(run == null || run.getId() == null || !FullReindexStep.RUN_TYPE.equals(run.getRunType())) { return; } - UpdateInput updateInput = new UpdateInput(); - updateInput.setTableName(getConfig().getQuickSearchIndexRunTableName()); - updateInput.setRecords(List.of(new QRecord().withValue("id", run.getId()).withValue("modifyDate", Instant.now()))); - new UpdateAction().execute(updateInput); + try + { + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(getConfig().getQuickSearchIndexRunTableName()); + updateInput.setRecords(List.of(new QRecord().withValue("id", run.getId()).withValue("modifyDate", Instant.now()))); + new UpdateAction().execute(updateInput); + } + catch(Exception e) + { + LOG.warn("Could not refresh the full reindex heartbeat", e, logPair("runId", run.getId())); + } + } + + + + /******************************************************************************* + ** The record IDs, among the rows' recordIds, that exist in the source table. + *******************************************************************************/ + 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); } diff --git a/src/main/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStep.java b/src/main/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStep.java index 0183818..91a03ec 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStep.java @@ -25,6 +25,7 @@ import java.util.HashMap; 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; @@ -268,8 +269,10 @@ public class BasepullIndexStep extends AbstractIndexingStep /******************************************************************************* ** Replay PENDING failed real-time events: re-read and index records for - ** INDEX events, delete documents for DELETE events. Successes are removed; - ** failures bump attempts and become EXHAUSTED at the limit. + ** INDEX events, delete documents for DELETE events. A DELETE whose record + ** exists again (re-created with the same key since) keeps its document and + ** counts as replayed. Successes are removed; failures bump attempts and + ** become EXHAUSTED at the limit. *******************************************************************************/ void replayFailedEvents(RunBackendStepOutput output) throws QException { @@ -304,19 +307,37 @@ public class BasepullIndexStep extends AbstractIndexingStep for(Map.Entry> entry : deleteByTable.entrySet()) { - List documentIds = new ArrayList<>(); - for(QRecord event : entry.getValue()) - { - documentIds.add(OpenSearchDocument.buildDocumentId(entry.getKey(), event.getValueString("recordId"))); - } + List toDelete = entry.getValue(); try { - BulkIndexResult result = client.deleteDocuments(documentIds, config.getBulkBatchSize()); - markReplayOutcome(entry.getValue(), result.getFailureCount() == 0, result.getErrors(), succeeded, failed); + Set existingIds = queryExistingRecordIds(entry.getKey(), entry.getValue()); + toDelete = new ArrayList<>(); + for(QRecord event : entry.getValue()) + { + if(existingIds.contains(event.getValueString("recordId"))) + { + succeeded.add(event.getValue("id")); + } + else + { + toDelete.add(event); + } + } + + if(!toDelete.isEmpty()) + { + List documentIds = new ArrayList<>(); + for(QRecord event : toDelete) + { + documentIds.add(OpenSearchDocument.buildDocumentId(entry.getKey(), event.getValueString("recordId"))); + } + BulkIndexResult result = client.deleteDocuments(documentIds, config.getBulkBatchSize()); + markReplayOutcome(toDelete, result.getFailureCount() == 0, result.getErrors(), succeeded, failed); + } } catch(Exception e) { - markReplayOutcome(entry.getValue(), false, List.of(e.getMessage() == null ? e.toString() : e.getMessage()), succeeded, failed); + markReplayOutcome(toDelete, false, List.of(e.getMessage() == null ? e.toString() : e.getMessage()), succeeded, failed); } } 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 a22086c..974b685 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java @@ -102,9 +102,9 @@ public class FullReindexStep extends AbstractIndexingStep String tableNameFilter = input.getValueString("tableName"); boolean allTables = tableNameFilter == null || tableNameFilter.isBlank(); - //////////////////////////////////////////////////////////////////////////// + ////////////////////////////////////////////////////////////////////////////// // an all-tables rebuild applies the captured deletes itself after its swap // - //////////////////////////////////////////////////////////////////////////// + ////////////////////////////////////////////////////////////////////////////// refuseIfAlive(recoverDeadRun(!allTables)); List tables = new ArrayList<>(); @@ -197,6 +197,7 @@ public class FullReindexStep extends AbstractIndexingStep } client.refreshIndex(physicalIndex); + keepAlive(runs.values()); verifyStillOwned(tables, runs.values()); client.swapAliasTo(physicalIndex); swapped = true; @@ -205,6 +206,7 @@ public class FullReindexStep extends AbstractIndexingStep { String tableName = tableConfig.getTableName(); IndexCounts tableCounts = counts.get(tableName); + keepAlive(runs.values()); Map values = new HashMap<>(); values.put("lastFullReindexTime", Instant.now()); @@ -221,11 +223,13 @@ public class FullReindexStep extends AbstractIndexingStep priorStatus.remove(tableName); } - /////////////////////////////////////////////////////////////////////////// - // the run records stay RUNNING until the captured deletes are applied, // - // so a basepull meanwhile does not take them for ones left behind // - /////////////////////////////////////////////////////////////////////////// - applyCapturedDeletes(client, tables); + ///////////////////////////////////////////////////////////////////////////// + // the run records stay RUNNING, refreshed per batch, until the captured // + // deletes are applied; if a recovery takes the run over anyway, the drain // + // stops, and it never reads rows captured after it started, which belong // + // to a newer run // + ///////////////////////////////////////////////////////////////////////////// + applyCapturedDeletes(client, tables, runs.values()); for(QuickSearchableTableConfig tableConfig : tables) { @@ -241,16 +245,6 @@ public class FullReindexStep extends AbstractIndexingStep boolean takenOver = marked && isTakenOverOrUnknown(runs.values()); LOG.warn(swapped ? "Full reindex failed after the alias swap; the new index stays in service" : "Full reindex failed; the previous index stays in service", e, logPair("physicalIndex", physicalIndex)); - for(Map.Entry entry : runs.entrySet()) - { - IndexCounts tableCounts = counts.get(entry.getKey().getTableName()); - completeRunRecord(entry.getValue(), RUN_FAILED, - tableCounts == null ? 0 : tableCounts.processed(), - tableCounts == null ? 0 : tableCounts.indexed(), - tableCounts == null ? 1 : Math.max(1, tableCounts.errors()), - e.getMessage()); - } - if(created && !swapped) { try @@ -263,15 +257,28 @@ public class FullReindexStep extends AbstractIndexingStep } } - ////////////////////////////////////////////////////////////////////////// - // a recovery that judged this run stopped has already put the tables // - // back and handed the captured deletes to basepull, and a newer run // - // may own both by now // - ////////////////////////////////////////////////////////////////////////// - if(marked && !takenOver) + /////////////////////////////////////////////////////////////////////////// + // a recovery that took this run over owns its tables, captured deletes // + // and run records (a newer run may own the first two by now); when that // + // cannot be read, the run records stay RUNNING for the next recovery // + /////////////////////////////////////////////////////////////////////////// + if(!takenOver) { - restoreStatus(priorStatus); - applyCapturedDeletes(client, tables); + if(marked) + { + restoreStatus(priorStatus); + applyCapturedDeletes(client, tables, runs.values()); + } + + for(Map.Entry entry : runs.entrySet()) + { + IndexCounts tableCounts = counts.get(entry.getKey().getTableName()); + completeRunRecord(entry.getValue(), RUN_FAILED, + tableCounts == null ? 0 : tableCounts.processed(), + tableCounts == null ? 0 : tableCounts.indexed(), + tableCounts == null ? 1 : Math.max(1, tableCounts.errors()), + e.getMessage()); + } } if(e instanceof QException qException) @@ -330,25 +337,29 @@ public class FullReindexStep extends AbstractIndexingStep ** ** A run is alive while the startTime or modifyDate (its heartbeat) of one of ** its RUNNING records is within fullReindexStaleMinutes. When none is - ** alive: each stale RUNNING record is marked FAILED; each REBUILDING table - ** goes back to ACTIVE (the status it had before is not stored; drift - ** detection marks NEEDS_REINDEX again where it applies), with its - ** lastBasepullTime moved back to the stopped run's start, so basepull - ** re-reads updates that a swap may have left out; and, with - ** handCapturedDeletesToBasepull, every AWAITING_REINDEX row becomes PENDING. - ** Replaying those deletes is harmless whether or not the stopped run - ** swapped: before a swap the alias points at the old index, which the - ** listener already deleted from (a missing document counts as success); - ** after one, it removes the copy the run read before the delete. + ** alive: each REBUILDING table goes back to ACTIVE (the status it had + ** before is not stored; drift detection marks NEEDS_REINDEX again where it + ** applies), with its lastBasepullTime moved back to the stopped run's + ** start, so basepull re-reads updates that a swap may have left out; with + ** handCapturedDeletesToBasepull, every AWAITING_REINDEX row up to the + ** newest one read here becomes PENDING; and only then is each stale RUNNING + ** record marked FAILED, so a recovery that fails partway leaves them for + ** the next one, run start included. Replaying those deletes is harmless + ** whether or not the stopped run swapped: before a swap the alias points at + ** the old index, which the listener already deleted from (a missing + ** document counts as success); after one, it removes the copy the run read + ** before the delete. ** ** The REBUILDING rows and captured rows are read before the run records. A ** run creates its records before it marks a table, and the listener ** captures a delete only once it sees REBUILDING, so anything read here ** that belongs to a live run comes with a live record, and nothing of a - ** live run is touched. A run that starts after the records are read can - ** have a table it just marked put back to ACTIVE, and a run misjudged as - ** stopped finds its records FAILED; both check before their swap and fail - ** without swapping, so no delete is lost. + ** live run is touched. A run that starts after the records are read, or a + ** live run misjudged as stopped, can have a table put back to ACTIVE or its + ** records marked FAILED; it checks both right after a heartbeat, before its + ** swap, and fails without swapping. A misjudged run that already swapped + ** stops applying its captured deletes once its records are FAILED, and + ** never reads rows captured after its drain started. *******************************************************************************/ List recoverDeadRun(boolean handCapturedDeletesToBasepull) throws QException { @@ -367,15 +378,6 @@ public class FullReindexStep extends AbstractIndexingStep logPair("staleRunIds", runningRuns.stream().map(run -> run.getValue("id")).toList()), logPair("tableNames", rebuildingRows.stream().map(row -> row.getValueString("tableName")).toList())); - for(QRecord runRecord : runningRuns) - { - QuickSearchIndexRun run = new QuickSearchIndexRun() - .withId(runRecord.getValueInteger("id")) - .withQuickSearchIndexId(runRecord.getValueInteger("quickSearchIndexId")) - .withRunType(RUN_TYPE); - completeRunRecord(run, RUN_FAILED, 0, 0, 1, "Full reindex stopped without finishing (no heartbeat since " + lastSeen(runRecord) + "); recovered automatically"); - } - for(QRecord row : rebuildingRows) { Map values = new HashMap<>(); @@ -394,6 +396,15 @@ public class FullReindexStep extends AbstractIndexingStep { handCapturedDeletesToBasepull(lastCapturedId); } + + for(QRecord runRecord : runningRuns) + { + QuickSearchIndexRun run = new QuickSearchIndexRun() + .withId(runRecord.getValueInteger("id")) + .withQuickSearchIndexId(runRecord.getValueInteger("quickSearchIndexId")) + .withRunType(RUN_TYPE); + completeRunRecord(run, RUN_FAILED, 0, 0, 1, "Full reindex stopped without finishing (no heartbeat since " + lastSeen(runRecord) + "); recovered automatically"); + } return (alive); } @@ -490,6 +501,16 @@ public class FullReindexStep extends AbstractIndexingStep + /******************************************************************************* + ** Refresh one of the run's records, which keeps the whole run alive. + *******************************************************************************/ + private void keepAlive(Collection runs) + { + runs.stream().findFirst().ifPresent(this::heartbeat); + } + + + /******************************************************************************* ** Whether a recovery has marked any of the run records finished. *******************************************************************************/ @@ -652,7 +673,9 @@ public class FullReindexStep extends AbstractIndexingStep ** own writes, so an in-flight page cannot bring a deleted record back. ** 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). + ** basepull). Only rows up to the newest one read at the start are applied, + ** since a later row belongs to a newer run. The run is kept alive before + ** every batch, and the drain stops once a recovery has taken it over. ** ** 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 @@ -660,7 +683,7 @@ public class FullReindexStep extends AbstractIndexingStep ** next full reindex, or a basepull that finds none running, picks them up. ** Never throws. *******************************************************************************/ - void applyCapturedDeletes(QuickSearchOpenSearchClient client, List tables) + void applyCapturedDeletes(QuickSearchOpenSearchClient client, List tables, Collection runs) { List tableNames = new ArrayList<>(); tables.forEach(tableConfig -> tableNames.add(tableConfig.getTableName())); @@ -674,13 +697,22 @@ public class FullReindexStep extends AbstractIndexingStep { QuickSearchQBitConfig config = getConfig(); String failedEventTable = config.applyPrefix(QuickSearchFailedEvent.TABLE_NAME); + Serializable lastCapturedId = queryLastCapturedDeleteId(); Serializable afterId = null; - while(true) + while(lastCapturedId != null) { + if(isTakenOver(runs)) + { + LOG.warn("A recovery took this full reindex over while it applied its captured deletes; leaving the rest to the recovery", logPair("applied", applied)); + break; + } + keepAlive(runs); + QQueryFilter filter = new QQueryFilter( new QFilterCriteria("status", QCriteriaOperator.EQUALS, QuickSearchFailedEvent.STATUS_AWAITING_REINDEX), - new QFilterCriteria("tableName", QCriteriaOperator.IN, tableNames)) + new QFilterCriteria("tableName", QCriteriaOperator.IN, tableNames), + new QFilterCriteria("id", QCriteriaOperator.LESS_THAN_OR_EQUALS, lastCapturedId)) .withOrderBy(new QFilterOrderBy("id", true)) .withLimit(config.getSourceBatchSize()); if(afterId != null) @@ -821,35 +853,6 @@ public class FullReindexStep extends AbstractIndexingStep - /******************************************************************************* - ** The record IDs, among the rows' recordIds, that exist in the source table. - *******************************************************************************/ - 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/processes/BasepullIndexStepTest.java b/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java index 9bd9b66..c0ea648 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java @@ -969,6 +969,29 @@ class BasepullIndexStepTest extends BaseQuickSearchTest + /******************************************************************************* + ** Test: a DELETE event whose record exists again (re-created with the same + ** key since) keeps the record's document; the event is resolved. + *******************************************************************************/ + @Test + void testReplayFailedEvents_deleteOfRecreatedRecord_keepsItsDocument() throws QException + { + insertTestEntities(1); + insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60); + insertFailedEvent(TEST_ENTITY_TABLE, "1", "DELETE", 0); + insertFailedEvent(TEST_ENTITY_TABLE, "9", "DELETE", 0); + + RunBackendStepOutput output = new RunBackendStepOutput(); + new BasepullIndexStep().run(new RunBackendStepInput(), output); + + verify(mockClient, times(1)).deleteDocuments(anyList(), anyInt()); + verify(mockClient).deleteDocuments(List.of(TEST_ENTITY_TABLE + ":9"), QuickSearchQBitContext.getConfig().getBulkBatchSize()); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty(); + assertThat(output.getValueInteger("failedEventsReplayed")).isEqualTo(2); + } + + + /******************************************************************************* ** Test: a full reindex that stopped without finishing (table left ** REBUILDING, run record without a heartbeat for two hours) is recovered 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 4c44651..34d76aa 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java @@ -17,10 +17,12 @@ package com.kingsrook.qbits.quicksearch.processes; +import java.io.Serializable; import java.time.Instant; import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -802,6 +804,24 @@ class FullReindexStepTest extends BaseQuickSearchTest + /******************************************************************************* + ** The run record with the given id, for lambdas and overrides that cannot + ** throw a checked exception. + *******************************************************************************/ + private QRecord queryRunQuietly(Integer runId) + { + try + { + return (queryRun(runId)); + } + catch(QException e) + { + throw (new IllegalStateException(e)); + } + } + + + /******************************************************************************* ** The run record with the given id. *******************************************************************************/ @@ -941,20 +961,77 @@ class FullReindexStepTest extends BaseQuickSearchTest insertIndexRow(true); Integer runId = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120).getValueInteger("id"); - List heartbeatsSeenByEachBulk = new ArrayList<>(); + AtomicInteger pagesIndexed = new AtomicInteger(0); when(mockClient.indexDocuments(anyString(), anyList(), anyInt())).thenAnswer(invocation -> { - heartbeatsSeenByEachBulk.add(queryRun(runId).getValueInstant("modifyDate")); + pagesIndexed.incrementAndGet(); return (new BulkIndexResult().withSuccessCount(((List) invocation.getArgument(1)).size())); }); - Instant before = Instant.now(); - new FullReindexStep().indexAllRecords(mockClient, QuickSearchQBitContext.getTableConfig(TEST_ENTITY_TABLE), List.of(), PHYSICAL_INDEX, new QuickSearchIndexRun().withId(runId)); + List pagesIndexedAtEachHeartbeat = new ArrayList<>(); + List heartbeatsWritten = new ArrayList<>(); + FullReindexStep step = new FullReindexStep() + { + @Override + protected void heartbeat(QuickSearchIndexRun run) + { + super.heartbeat(run); + pagesIndexedAtEachHeartbeat.add(pagesIndexed.get()); + heartbeatsWritten.add(queryRunQuietly(runId).getValueInstant("modifyDate")); + } + }; - assertThat(heartbeatsSeenByEachBulk).hasSize(3); - assertThat(heartbeatsSeenByEachBulk.get(0)).isBefore(before); - assertThat(heartbeatsSeenByEachBulk.get(1)).isAfterOrEqualTo(before); - assertThat(queryRun(runId).getValueInstant("modifyDate")).isAfterOrEqualTo(heartbeatsSeenByEachBulk.get(2)); + Instant before = Instant.now(); + step.indexAllRecords(mockClient, QuickSearchQBitContext.getTableConfig(TEST_ENTITY_TABLE), List.of(), PHYSICAL_INDEX, + new QuickSearchIndexRun().withId(runId).withRunType(FullReindexStep.RUN_TYPE)); + + assertThat(pagesIndexedAtEachHeartbeat).containsExactly(1, 2, 3); + assertThat(heartbeatsWritten).hasSize(3).allSatisfy(heartbeat -> assertThat(heartbeat).isAfterOrEqualTo(before)); + assertThat(heartbeatsWritten).isSorted(); + } + + + + /******************************************************************************* + ** Test: only FULL_REINDEX runs have a heartbeat (nothing reads one for the + ** other run types). + *******************************************************************************/ + @Test + void testHeartbeat_onlyForFullReindexRuns() throws QException + { + insertIndexRow(true); + QRecord run = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120); + + new FullReindexStep().heartbeat(new QuickSearchIndexRun().withId(run.getValueInteger("id")).withRunType(BasepullIndexStep.RUN_TYPE)); + assertThat(queryRun(run.getValueInteger("id")).getValueInstant("modifyDate")).isEqualTo(run.getValueInstant("modifyDate")); + + new FullReindexStep().heartbeat(new QuickSearchIndexRun().withId(run.getValueInteger("id")).withRunType(FullReindexStep.RUN_TYPE)); + assertThat(queryRun(run.getValueInteger("id")).getValueInstant("modifyDate")).isAfter(run.getValueInstant("modifyDate")); + } + + + + /******************************************************************************* + ** Test: a heartbeat that cannot be written is logged and the pages are + ** still indexed. + *******************************************************************************/ + @Test + void testHeartbeat_failureDoesNotStopIndexing() throws QException + { + QuickSearchQBitContext.getConfig().withSourceBatchSize(2); + insertTestEntities(3); + + AbstractIndexingStep.IndexCounts counts; + try(MockedConstruction updateActions = mockConstruction(UpdateAction.class, (updateAction, context) -> + when(updateAction.execute(any())).thenThrow(new QException("database unavailable")))) + { + counts = new FullReindexStep().indexAllRecords(mockClient, QuickSearchQBitContext.getTableConfig(TEST_ENTITY_TABLE), List.of(), PHYSICAL_INDEX, + new QuickSearchIndexRun().withId(1).withRunType(FullReindexStep.RUN_TYPE)); + assertThat(updateActions.constructed()).hasSize(2); + } + + assertThat(counts.processed()).isEqualTo(3); + assertThat(counts.indexed()).isEqualTo(3); } @@ -1144,6 +1221,147 @@ class FullReindexStepTest extends BaseQuickSearchTest + /******************************************************************************* + ** Test: recovery hands over only the captured rows that existed when it + ** read them; a row captured after (by a run that started meanwhile) stays + ** AWAITING_REINDEX for that run. + *******************************************************************************/ + @Test + void testRecovery_handsOverOnlyRowsCapturedBeforeItRead() throws QException + { + insertIndexRow(true); + updateIndexRowStatus(AbstractIndexingStep.STATUS_REBUILDING); + insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120); + insertCapturedDelete("4"); + + AtomicBoolean capturedMeanwhile = new AtomicBoolean(false); + new FullReindexStep() + { + @Override + protected void updateIndexRow(Integer indexId, Map values) throws QException + { + super.updateIndexRow(indexId, values); + if(capturedMeanwhile.compareAndSet(false, true)) + { + insertCapturedDelete("5"); + } + } + }.recoverDeadRun(true); + + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).extracting(event -> event.getValueString("recordId") + " " + event.getValueString("status")) + .containsExactlyInAnyOrder("4 " + QuickSearchFailedEvent.STATUS_PENDING, "5 " + QuickSearchFailedEvent.STATUS_AWAITING_REINDEX); + } + + + + /******************************************************************************* + ** Test: a recovery that fails while putting a table back leaves the dead + ** run's record RUNNING, so the next recovery still finds the run's start + ** and rewinds the basepull watermark. + *******************************************************************************/ + @Test + void testRecovery_failsPartway_nextRecoveryStillRewinds() throws QException + { + insertIndexRow(true); + updateIndexRow(AbstractIndexingStep.STATUS_REBUILDING, Instant.now().minus(5, ChronoUnit.MINUTES)); + QRecord deadRun = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120); + + assertThatThrownBy(() -> new FullReindexStep() + { + @Override + protected void updateIndexRow(Integer indexId, Map values) throws QException + { + throw (new QException("database unavailable")); + } + }.recoverDeadRun(true)) + .isInstanceOf(QException.class) + .hasMessageContaining("database unavailable"); + + assertThat(queryRun(deadRun.getValueInteger("id")).getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING); + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING); + + assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty(); + + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + assertThat(queryIndexRow().getValueInstant("lastBasepullTime")).isEqualTo(deadRun.getValueInstant("startTime")); + assertThat(queryRun(deadRun.getValueInteger("id")).getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED); + } + + + + /******************************************************************************* + ** Test: a run whose drain of captured deletes is so slow that it is judged + ** stopped, while a new full reindex starts, marks the table and captures a + ** delete of its own: the drain stops at its next batch and leaves both its + ** remaining row and the new run's row alone. + *******************************************************************************/ + @Test + void testDrainTakenOverMidway_leavesTheNewRunsRowsAlone() throws QException + { + QuickSearchQBitContext.getConfig().withSourceBatchSize(1); + insertTestEntities(1); + insertCapturedDelete("8"); + insertCapturedDelete("9"); + + AtomicBoolean newRunStarted = new AtomicBoolean(false); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenAnswer(invocation -> + { + if(newRunStarted.compareAndSet(false, true)) + { + for(QRecord run : queryAll(QuickSearchIndexRun.TABLE_NAME)) + { + backdateRun(run.getValueInteger("id")); + } + assertThat(new FullReindexStep().recoverDeadRun(false)).isEmpty(); + insertRunningRun(queryIndexRow().getValueInteger("id"), 0, 0); + updateIndexRowStatus(AbstractIndexingStep.STATUS_REBUILDING); + insertCapturedDelete("7"); + } + return (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 + ":8")), anyInt()); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).extracting(event -> event.getValueString("recordId") + " " + event.getValueString("status")) + .containsExactlyInAnyOrder("9 " + QuickSearchFailedEvent.STATUS_AWAITING_REINDEX, "7 " + QuickSearchFailedEvent.STATUS_AWAITING_REINDEX); + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING); + } + + + + /******************************************************************************* + ** Test: the drain applies only rows up to the newest one at its start; a + ** row captured while it runs is left alone. + *******************************************************************************/ + @Test + void testDrain_stopsAtTheNewestRowReadAtItsStart() throws QException + { + QuickSearchQBitContext.getConfig().withSourceBatchSize(1); + insertTestEntities(1); + insertCapturedDelete("8"); + insertCapturedDelete("9"); + + AtomicBoolean capturedMeanwhile = new AtomicBoolean(false); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenAnswer(invocation -> + { + if(capturedMeanwhile.compareAndSet(false, true)) + { + insertCapturedDelete("7"); + } + return (new BulkIndexResult().withSuccessCount(1)); + }); + + new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()); + + verify(mockClient, times(2)).deleteDocuments(anyList(), anyInt()); + assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).singleElement() + .satisfies(event -> assertThat(event.getValueString("recordId")).isEqualTo("7")); + } + + + /******************************************************************************* ** Test: with nothing left behind, recovery changes nothing. *******************************************************************************/ -- 2.54.0 From d2c5c9b1f1ba84184ad782c3b4d6ba5221d29064 Mon Sep 17 00:00:00 2001 From: James Maes Date: Fri, 9 Oct 2026 19:04:34 -0500 Subject: [PATCH 3/3] fix(#24): fail recovery when a dead run cannot be marked FAILED Recovery now writes FAILED with an update that must succeed for every stale run record, so a new full reindex refuses to start rather than run beside a misjudged live one. Tests pin each heartbeat point and the unreadable-takeover branch. Refs #24 --- .../processes/FullReindexStep.java | 75 ++++++++- .../processes/FullReindexStepTest.java | 155 ++++++++++++++++++ 2 files changed, 223 insertions(+), 7 deletions(-) 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 974b685..59b3336 100644 --- a/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java +++ b/src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java @@ -344,7 +344,9 @@ public class FullReindexStep extends AbstractIndexingStep ** handCapturedDeletesToBasepull, every AWAITING_REINDEX row up to the ** newest one read here becomes PENDING; and only then is each stale RUNNING ** record marked FAILED, so a recovery that fails partway leaves them for - ** the next one, run start included. Replaying those deletes is harmless + ** the next one, run start included. Unlike completeRunRecord, that write + ** fails the recovery (and so the full reindex calling it) when any record + ** is not updated. Replaying those deletes is harmless ** whether or not the stopped run swapped: before a swap the alias points at ** the old index, which the listener already deleted from (a missing ** document counts as success); after one, it removes the copy the run read @@ -397,15 +399,74 @@ public class FullReindexStep extends AbstractIndexingStep handCapturedDeletesToBasepull(lastCapturedId); } + failDeadRuns(runningRuns); + return (alive); + } + + + + /******************************************************************************* + ** Mark the stale run records FAILED, failing when any of them is not + ** updated: a live run misjudged as stopped learns it was taken over only + ** from these records, so recovery must not finish, and no new full reindex + ** may start, until they are written. The status is mirrored onto each + ** index row, as completeRunRecord does. + *******************************************************************************/ + private void failDeadRuns(List runningRuns) throws QException + { + if(runningRuns.isEmpty()) + { + return; + } + + Instant endTime = Instant.now(); + List updates = new ArrayList<>(); for(QRecord runRecord : runningRuns) { - QuickSearchIndexRun run = new QuickSearchIndexRun() - .withId(runRecord.getValueInteger("id")) - .withQuickSearchIndexId(runRecord.getValueInteger("quickSearchIndexId")) - .withRunType(RUN_TYPE); - completeRunRecord(run, RUN_FAILED, 0, 0, 1, "Full reindex stopped without finishing (no heartbeat since " + lastSeen(runRecord) + "); recovered automatically"); + updates.add(new QRecord() + .withValue("id", runRecord.getValue("id")) + .withValue("status", RUN_FAILED) + .withValue("endTime", endTime) + .withValue("recordsProcessed", 0) + .withValue("recordsIndexed", 0) + .withValue("errorCount", 1) + .withValue("errorMessage", truncate(staleRunMessage(runRecord), 4000))); } - return (alive); + + UpdateInput updateInput = new UpdateInput(); + updateInput.setTableName(getConfig().getQuickSearchIndexRunTableName()); + updateInput.setRecords(updates); + List results = CollectionUtils.nonNullList(new UpdateAction().execute(updateInput).getRecords()); + List written = results.stream().filter(result -> CollectionUtils.nullSafeIsEmpty(result.getErrors())).toList(); + if(written.size() < updates.size()) + { + List notWritten = results.stream().filter(result -> CollectionUtils.nullSafeHasContents(result.getErrors())).toList(); + throw (new QException("Could not mark " + (updates.size() - written.size()) + " stopped full reindex run record(s) FAILED" + + (notWritten.isEmpty() ? "" : " (first error: " + notWritten.get(0).getErrors() + ")") + + "; recovery did not finish, and no full reindex starts until it does")); + } + + for(QRecord runRecord : runningRuns) + { + Integer indexId = runRecord.getValueInteger("quickSearchIndexId"); + if(indexId != null) + { + Map values = new HashMap<>(); + values.put("lastRunStatus", RUN_TYPE + " " + RUN_FAILED); + values.put("lastErrorMessage", truncate(staleRunMessage(runRecord), 4000)); + updateIndexRow(indexId, values); + } + } + } + + + + /******************************************************************************* + ** The error message recorded on a run that stopped without finishing. + *******************************************************************************/ + private static String staleRunMessage(QRecord runRecord) + { + return ("Full reindex stopped without finishing (no heartbeat since " + lastSeen(runRecord) + "); recovered automatically"); } 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 34d76aa..f6feb7d 100644 --- a/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java +++ b/src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java @@ -42,9 +42,11 @@ 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.actions.tables.update.UpdateOutput; 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.QuickSearchQBitConfig; import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext; import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig; import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent; @@ -66,6 +68,7 @@ import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; @@ -1362,6 +1365,158 @@ class FullReindexStepTest extends BaseQuickSearchTest + /******************************************************************************* + ** Test: when the stale run records cannot be marked FAILED (every update + ** comes back with a per-row error), recovery fails and the new full + ** reindex does not start, since a live run misjudged as stopped would not + ** learn it was taken over. + *******************************************************************************/ + @Test + void testRecovery_failedRunRecordsNotWritten_newRunDoesNotStart() throws QException + { + insertTestEntities(1); + insertIndexRow(true); + updateIndexRowStatus(AbstractIndexingStep.STATUS_REBUILDING); + Integer staleRunId = insertRunningRun(queryIndexRow().getValueInteger("id"), 120, 120).getValueInteger("id"); + + try(MockedConstruction updateActions = mockConstruction(UpdateAction.class, (updateAction, context) -> + when(updateAction.execute(any())).thenAnswer(invocation -> + { + UpdateInput updateInput = invocation.getArgument(0); + UpdateOutput updateOutput = new UpdateOutput(); + updateOutput.setRecords(updateInput.getRecords().stream().map(record -> new QRecord(record).withError(new SystemErrorStatusMessage("row is locked"))).toList()); + return (updateOutput); + }))) + { + assertThatThrownBy(() -> new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput())) + .isInstanceOf(QException.class) + .hasMessageContaining("FAILED") + .hasMessageContaining("row is locked"); + } + + verify(mockClient, never()).createPhysicalIndex(anyString()); + verify(mockClient, never()).indexDocuments(anyString(), anyList(), anyInt()); + assertThat(queryAll(QuickSearchIndexRun.TABLE_NAME)).singleElement() + .satisfies(run -> + { + assertThat(run.getValueInteger("id")).isEqualTo(staleRunId); + assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING); + }); + } + + + + /******************************************************************************* + ** Test: the run keeps itself alive at every point where it could otherwise + ** go quiet: after each page, right before its pre-swap ownership check, + ** before each table's update after the swap, and before each batch of the + ** captured-delete drain. + *******************************************************************************/ + @Test + void testAllTables_heartbeatsBeforeTheOwnershipCheckEachTableAndEachDrainBatch() throws QException + { + QuickSearchQBitContext.getConfig().withSourceBatchSize(1); + insertTestEntities(1); + insertCapturedDelete("8"); + insertCapturedDelete("9"); + + List events = new ArrayList<>(); + doAnswer(invocation -> + { + events.add("refresh"); + return (null); + }).when(mockClient).refreshIndex(PHYSICAL_INDEX); + doAnswer(invocation -> + { + events.add("swap"); + return (null); + }).when(mockClient).swapAliasTo(PHYSICAL_INDEX); + when(mockClient.deleteDocuments(anyList(), anyInt())).thenAnswer(invocation -> + { + events.add("drainBatch"); + return (new BulkIndexResult().withSuccessCount(1)); + }); + + new FullReindexStep() + { + @Override + protected void heartbeat(QuickSearchIndexRun run) + { + events.add("heartbeat"); + super.heartbeat(run); + } + + + + @Override + protected void updateIndexRow(Integer indexId, Map values) throws QException + { + if(values.containsKey("lastFullReindexTime")) + { + events.add("tableDone"); + } + super.updateIndexRow(indexId, values); + } + }.run(new RunBackendStepInput(), new RunBackendStepOutput()); + + assertThat(events).containsExactly( + "heartbeat", + "refresh", "heartbeat", "swap", + "heartbeat", "tableDone", + "heartbeat", "drainBatch", "heartbeat", "drainBatch", "heartbeat"); + } + + + + /******************************************************************************* + ** Test: when a run fails and cannot read whether a recovery took it over, + ** it leaves its run records RUNNING and its tables REBUILDING, so the next + ** recovery (once the records are stale) puts everything back, watermark + ** rewind included. + *******************************************************************************/ + @Test + void testFailure_takeoverUnreadable_leavesRunRecordsRunningForRecovery() throws QException + { + insertTestEntities(1); + AtomicBoolean databaseDown = new AtomicBoolean(false); + when(mockClient.indexDocuments(anyString(), anyList(), anyInt())).thenAnswer(invocation -> + { + databaseDown.set(true); + throw (new QException("bulk failed")); + }); + + assertThatThrownBy(() -> new FullReindexStep() + { + @Override + protected QuickSearchQBitConfig getConfig() throws QException + { + /////////////////////////////////////////////////////////////////// + // only the first read after the failure (the takeover check) is // + // lost; everything after it would work // + /////////////////////////////////////////////////////////////////// + if(databaseDown.getAndSet(false)) + { + throw (new QException("database unavailable")); + } + return (super.getConfig()); + } + }.run(new RunBackendStepInput(), new RunBackendStepOutput())) + .isInstanceOf(QException.class) + .hasMessageContaining("bulk failed"); + + verify(mockClient).deletePhysicalIndex(PHYSICAL_INDEX); + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING); + QRecord run = queryAll(QuickSearchIndexRun.TABLE_NAME).get(0); + assertThat(run.getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_RUNNING); + + backdateRun(run.getValueInteger("id")); + assertThat(new FullReindexStep().recoverDeadRun(true)).isEmpty(); + assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE); + assertThat(queryRun(run.getValueInteger("id")).getValueString("status")).isEqualTo(AbstractIndexingStep.RUN_FAILED); + } + + + /******************************************************************************* ** Test: with nothing left behind, recovery changes nothing. *******************************************************************************/ -- 2.54.0