README.md

@@ -167,9 +167,9 @@ Without `schedulerName` the processes exist but nothing runs them; the QBit logs
 
 ## Operational tables
 
-- `quickSearchIndex`: one row per table with `enabled` (honoured by indexing and search, cached for one minute), `basepullIntervalMinutes`, `lastBasepullTime`, `lastReconcileTime`, `lastFullReindexTime`, `lastRunStatus`, `lastErrorMessage`, `recordCount`, `documentCount`, `realTimeErrorCount`, `status` (`ACTIVE` or `NEEDS_REINDEX` when the configured fields changed since the row was created). Unique on `tableName`.
+- `quickSearchIndex`: one row per table with `enabled` (honoured by indexing and search, cached for one minute), `basepullIntervalMinutes`, `lastBasepullTime`, `lastReconcileTime`, `lastFullReindexTime`, `lastRunStatus`, `lastErrorMessage`, `recordCount`, `documentCount`, `realTimeErrorCount`, `status` (`ACTIVE`, `NEEDS_REINDEX` when the configured fields changed since the row was created, or `REBUILDING` while a full reindex of all tables runs). Unique on `tableName`.
 - `quickSearchIndexRun`: `runType` (`BASEPULL`, `RECONCILE`, `FULL_REINDEX`), `status`, counts, `errorMessage`.
-- `quickSearchFailedEvent`: real-time events that could not be applied; `status` `PENDING` or `EXHAUSTED` after 10 attempts.
+- `quickSearchFailedEvent`: real-time events that could not be applied; `status` `PENDING` or `EXHAUSTED` after 10 attempts. Deletes made while a full reindex of all tables runs are also recorded here with status `AWAITING_REINDEX`: the reindex applies them to the new index after the alias swap (basepull does not replay them). While a table is `REBUILDING`, drift detection is suspended for it. If a full reindex is killed mid-run, its tables stay `REBUILDING` and deletes keep collecting as `AWAITING_REINDEX`; no data is lost, and the recovery is to run a full reindex of all tables again, which applies the collected deletes and returns the tables to `ACTIVE`.
 
 ## Extending
 

src/main/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListener.java

@@ -22,6 +22,7 @@ import java.util.ArrayList;
 import java.util.List;
 import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
 import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
 import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeEvent;
 import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeListenerInterface;
 import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeType;
@@ -33,11 +34,13 @@ import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperat
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
 import com.kingsrook.qbits.quicksearch.QuickSearchRuntime;
 import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
 import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent;
+import com.kingsrook.qbits.quicksearch.processes.AbstractIndexingStep;
 import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
 import com.kingsrook.qbits.quicksearch.publisher.IndexEventAction;
 import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher;
@@ -57,6 +60,11 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
  ** This listener never fails the host's write. When publishing fails, each
  ** affected (table, record, action) is written to quickSearchFailedEvent for
  ** the scheduled basepull to replay.
+ **
+ ** While a full reindex rebuilds into a new physical index, the alias still
+ ** points at the old one, so a published delete never reaches the new index.
+ ** Such deletes are also written to quickSearchFailedEvent with status
+ ** AWAITING_REINDEX; the reindex applies them after the alias swap.
  *******************************************************************************/
 public class QuickSearchRecordChangeListener implements RecordChangeListenerInterface
 {
@@ -129,9 +137,33 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
 
 
    /*******************************************************************************
-    ** Publish, recording any failure durably. Never throws.
+    ** Publish, recording any failure durably, and capture the deletes if a full
+    ** reindex is rebuilding the table. Never throws.
+    **
+    ** The status is read before the deletes are published, never after: a
+    ** reindex can swap the alias and clear REBUILDING while the publish is in
+    ** flight, and a read after that would miss a delete that reached only the
+    ** old index. A read before the publish that sees no REBUILDING means either
+    ** the reindex has not started reading the table (the record, deleted and
+    ** committed, cannot be read into the new index) or the alias already points
+    ** at the new index (the publish reaches it).
     *******************************************************************************/
    static void publish(QuickSearchRuntime runtime, List<IndexEvent> indexEvents, List<IndexEvent> deleteEvents)
+   {
+      boolean rebuilding = isRebuildingBeforePublish(runtime, deleteEvents);
+      publishOrRecordFailures(runtime, indexEvents, deleteEvents);
+      if(rebuilding)
+      {
+         captureDeletesDuringRebuild(runtime, deleteEvents);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static void publishOrRecordFailures(QuickSearchRuntime runtime, List<IndexEvent> indexEvents, List<IndexEvent> deleteEvents)
    {
       IndexEventPublisher publisher;
       try
@@ -188,29 +220,13 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
 
       try
       {
-         String failedEventTable = runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME);
-         if(QContext.getQInstance() == null || QContext.getQInstance().getTable(failedEventTable) == null)
+         if(!hasTable(runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME)))
          {
             LOG.error("quickSearchFailedEvent table is not in the QInstance; failed events are lost", logPair("eventCount", events.size()));
             return;
          }
 
-         List<QRecord> rows = new ArrayList<>();
-         for(IndexEvent event : events)
-         {
-            rows.add(new QRecord()
-               .withValue("tableName", event.getTableName())
-               .withValue("recordId", event.getRecordId())
-               .withValue("action", event.getAction() == null ? null : event.getAction().name())
-               .withValue("errorMessage", truncate(message, 4000))
-               .withValue("attempts", 0)
-               .withValue("status", QuickSearchFailedEvent.STATUS_PENDING));
-         }
-
-         InsertInput insertInput = new InsertInput();
-         insertInput.setTableName(failedEventTable);
-         insertInput.setRecords(rows);
-         new InsertAction().execute(insertInput);
+         insertFailedEventRows(runtime, events, message, QuickSearchFailedEvent.STATUS_PENDING);
       }
       catch(Exception e)
       {
@@ -220,6 +236,138 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
 
 
 
+   /*******************************************************************************
+    ** Whether a full reindex is rebuilding the deletes' table. The status is
+    ** read from the database (not cached) so a reindex on another node is seen.
+    ** Runs after the delete committed. Never throws.
+    *******************************************************************************/
+   static boolean isRebuildingBeforePublish(QuickSearchRuntime runtime, List<IndexEvent> deleteEvents)
+   {
+      if(deleteEvents.isEmpty())
+      {
+         return (false);
+      }
+
+      String tableName = deleteEvents.get(0).getTableName();
+      try
+      {
+         return (hasTable(runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME)) && isRebuilding(runtime, tableName));
+      }
+      catch(Exception e)
+      {
+         LOG.error("Could not read whether a full reindex is running; deletes made during one are not captured", e, logPair("tableName", tableName));
+         return (false);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Record deletes made while a full reindex is rebuilding the table, so the
+    ** reindex can apply them to the new physical index after the alias swap.
+    ** basepull does not replay AWAITING_REINDEX rows, so it cannot consume one
+    ** against the old index before the swap. Never throws.
+    *******************************************************************************/
+   static void captureDeletesDuringRebuild(QuickSearchRuntime runtime, List<IndexEvent> deleteEvents)
+   {
+      String tableName = deleteEvents.get(0).getTableName();
+      try
+      {
+         List<QRecord> inserted = insertFailedEventRows(runtime, deleteEvents, "deleted during a full reindex", QuickSearchFailedEvent.STATUS_AWAITING_REINDEX);
+         handOverIfRebuildFinished(runtime, tableName, inserted);
+      }
+      catch(Exception e)
+      {
+         LOG.error("Could not capture deletes made during a full reindex; the next full reindex or reconcile removes them", e,
+            logPair("tableName", tableName), logPair("eventCount", deleteEvents.size()));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** The reindex may have finished, and drained its captured deletes, between
+    ** the status read and the insert. Read the status again: if the table is no
+    ** longer REBUILDING, make the rows PENDING so basepull replays them against
+    ** the current index (deletes are idempotent). If it still is, the reindex
+    ** has not drained yet and will see these rows.
+    *******************************************************************************/
+   static void handOverIfRebuildFinished(QuickSearchRuntime runtime, String tableName, List<QRecord> inserted) throws QException
+   {
+      if(inserted.isEmpty() || isRebuilding(runtime, tableName))
+      {
+         return;
+      }
+
+      List<QRecord> updates = new ArrayList<>();
+      for(QRecord row : inserted)
+      {
+         updates.add(new QRecord().withValue("id", row.getValue("id")).withValue("status", QuickSearchFailedEvent.STATUS_PENDING));
+      }
+
+      UpdateInput updateInput = new UpdateInput();
+      updateInput.setTableName(runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME));
+      updateInput.setRecords(updates);
+      new UpdateAction().execute(updateInput);
+   }
+
+
+
+   /*******************************************************************************
+    ** Whether the table's quickSearchIndex row is REBUILDING.
+    *******************************************************************************/
+   private static boolean isRebuilding(QuickSearchRuntime runtime, String tableName) throws QException
+   {
+      String indexTable = runtime.getConfig().getQuickSearchIndexTableName();
+      if(!hasTable(indexTable))
+      {
+         return (false);
+      }
+
+      QueryInput queryInput = new QueryInput();
+      queryInput.setTableName(indexTable);
+      queryInput.setFilter(new QQueryFilter(new QFilterCriteria("tableName", QCriteriaOperator.EQUALS, tableName)).withLimit(1));
+      List<QRecord> rows = new QueryAction().execute(queryInput).getRecords();
+      return (CollectionUtils.nullSafeHasContents(rows) && AbstractIndexingStep.STATUS_REBUILDING.equals(rows.get(0).getValueString("status")));
+   }
+
+
+
+   /*******************************************************************************
+    ** Insert one quickSearchFailedEvent row per event; returns the inserted rows.
+    *******************************************************************************/
+   static List<QRecord> insertFailedEventRows(QuickSearchRuntime runtime, List<IndexEvent> events, String message, String status) throws QException
+   {
+      List<QRecord> rows = new ArrayList<>();
+      for(IndexEvent event : events)
+      {
+         rows.add(new QRecord()
+            .withValue("tableName", event.getTableName())
+            .withValue("recordId", event.getRecordId())
+            .withValue("action", event.getAction() == null ? null : event.getAction().name())
+            .withValue("errorMessage", truncate(message, 4000))
+            .withValue("attempts", 0)
+            .withValue("status", status));
+      }
+
+      InsertInput insertInput = new InsertInput();
+      insertInput.setTableName(runtime.getConfig().applyPrefix(QuickSearchFailedEvent.TABLE_NAME));
+      insertInput.setRecords(rows);
+      return (CollectionUtils.nonNullList(new InsertAction().execute(insertInput).getRecords()));
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static boolean hasTable(String tableName)
+   {
+      return (QContext.getQInstance() != null && QContext.getQInstance().getTable(tableName) != null);
+   }
+
+
+
    /*******************************************************************************
     **
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/quicksearch/model/QuickSearchFailedEvent.java

@@ -34,6 +34,12 @@ public class QuickSearchFailedEvent extends QRecordEntity
    public static final String STATUS_PENDING   = "PENDING";
    public static final String STATUS_EXHAUSTED = "EXHAUSTED";
 
+   ///////////////////////////////////////////////////////////////////////////
+   // a delete made while a full reindex rebuilds into a new physical index; //
+   // the reindex applies it after the swap, and basepull does not replay it //
+   ///////////////////////////////////////////////////////////////////////////
+   public static final String STATUS_AWAITING_REINDEX = "AWAITING_REINDEX";
+
    @QField()
    private Integer id;
 

src/main/java/com/kingsrook/qbits/quicksearch/processes/AbstractIndexingStep.java

@@ -65,6 +65,7 @@ public abstract class AbstractIndexingStep implements BackendStep
 
    public static final String STATUS_ACTIVE        = "ACTIVE";
    public static final String STATUS_NEEDS_REINDEX = "NEEDS_REINDEX";
+   public static final String STATUS_REBUILDING    = "REBUILDING";
    public static final String RUN_COMPLETED        = "COMPLETED";
    public static final String RUN_FAILED           = "FAILED";
    public static final String RUN_RUNNING          = "RUNNING";
@@ -219,6 +220,8 @@ public abstract class AbstractIndexingStep implements BackendStep
    /*******************************************************************************
     ** Compare the configured fields with what the row was created with; mark
     ** the row NEEDS_REINDEX when they differ. Returns true when drift was found.
+    ** A REBUILDING row is left alone: the full reindex in progress records the
+    ** current fields, and the status tells the listener to capture deletes.
     *******************************************************************************/
    protected boolean detectDrift(QRecord row, QuickSearchableTableConfig tableConfig) throws QException
    {
@@ -227,20 +230,53 @@ public abstract class AbstractIndexingStep implements BackendStep
          return (false);
       }
 
-      String current  = buildSearchableFieldsJson(tableConfig);
-      String recorded = row.getValueString("searchableFieldsJson");
-      if(recorded == null || recorded.equals(current) || STATUS_NEEDS_REINDEX.equals(row.getValueString("status")))
+      String current = buildSearchableFieldsJson(tableConfig);
+      if(!isUnmarkedDrift(row, current))
       {
-         return (recorded != null && !recorded.equals(current));
+         return (hasDrift(row, current));
       }
 
+      ///////////////////////////////////////////////////////////////////////////
+      // the row may be a snapshot taken before a full reindex started or     //
+      // finished; decide on a fresh read, so REBUILDING is kept and fields a //
+      // reindex has since recorded are not flagged                           //
+      ///////////////////////////////////////////////////////////////////////////
+      QRecord latest = queryIndexRow(tableConfig.getTableName());
+      if(latest == null || !isUnmarkedDrift(latest, current))
+      {
+         return (latest != null && hasDrift(latest, current));
+      }
+
+      // a status change in the moment between this read and the update below is not guarded
       LOG.warn("Searchable field configuration changed since the index row was created; marking NEEDS_REINDEX", logPair("tableName", tableConfig.getTableName()));
-      updateIndexRow(row.getValueInteger("id"), Map.of("status", STATUS_NEEDS_REINDEX));
+      updateIndexRow(latest.getValueInteger("id"), Map.of("status", STATUS_NEEDS_REINDEX));
       return (true);
    }
 
 
 
+   /*******************************************************************************
+    ** Whether the row records fields that differ from the configured ones.
+    *******************************************************************************/
+   private static boolean hasDrift(QRecord row, String current)
+   {
+      String recorded = row.getValueString("searchableFieldsJson");
+      return (recorded != null && !recorded.equals(current));
+   }
+
+
+
+   /*******************************************************************************
+    ** Whether the row has drifted and is not yet NEEDS_REINDEX or REBUILDING.
+    *******************************************************************************/
+   private static boolean isUnmarkedDrift(QRecord row, String current)
+   {
+      String status = row.getValueString("status");
+      return (hasDrift(row, current) && !STATUS_NEEDS_REINDEX.equals(status) && !STATUS_REBUILDING.equals(status));
+   }
+
+
+
    /*******************************************************************************
     ** JSON array describing a table's searchable fields, weights and labels.
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStep.java

@@ -21,16 +21,34 @@ import java.io.Serializable;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.logging.QLogger;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteOutput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
+import com.kingsrook.qbits.quicksearch.QuickSearchQBitConfig;
 import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
+import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent;
 import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
+import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult;
+import com.kingsrook.qbits.quicksearch.opensearch.OpenSearchDocument;
 import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
 import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
 
@@ -45,6 +63,14 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
  ** created by an older release (outdated mapping) is upgraded. On failure the
  ** new physical index is deleted and nothing changes.
  **
+ ** While the new index is built, real-time writes still go through the alias
+ ** to the old one. Updates are caught by the next basepull (lastBasepullTime
+ ** is set to each table's start). Deletes are caught here: each table's
+ ** quickSearchIndex row is REBUILDING for the whole rebuild, which makes the
+ ** record change listener (on any node) also record each delete as an
+ ** AWAITING_REINDEX failed event. Once the rebuild's own writes are done and
+ ** the alias is swapped, those deletes are applied through the alias.
+ **
  ** With a "tableName" input, that table is rebuilt in place with the reconcile
  ** algorithm (index everything, then delete what was not re-indexed).
  *******************************************************************************/
@@ -111,15 +137,18 @@ public class FullReindexStep extends AbstractIndexingStep
       client.createPhysicalIndex(physicalIndex);
       LOG.info("Rebuilding Quick Search index", logPair("physicalIndex", physicalIndex), logPair("tables", tables.size()));
 
-      Map<QuickSearchableTableConfig, QuickSearchIndexRun> runs       = new LinkedHashMap<>();
-      Map<String, IndexCounts>                              counts     = new LinkedHashMap<>();
-      Map<String, Instant>                                  startTimes = new LinkedHashMap<>();
-      Integer                                               indexed    = 0;
+      Map<QuickSearchableTableConfig, QuickSearchIndexRun> runs        = new LinkedHashMap<>();
+      Map<String, IndexCounts>                              counts      = new LinkedHashMap<>();
+      Map<String, Instant>                                  startTimes  = new LinkedHashMap<>();
+      Map<String, String>                                   priorStatus = new LinkedHashMap<>();
+      Integer                                               indexed     = 0;
 
       boolean swapped = false;
 
       try
       {
+         markRebuilding(tables, priorStatus);
+
          for(QuickSearchableTableConfig tableConfig : tables)
          {
             String              tableName = tableConfig.getTableName();
@@ -154,10 +183,13 @@ public class FullReindexStep extends AbstractIndexingStep
             values.put("searchableFieldsJson", buildSearchableFieldsJson(tableConfig));
             values.put("status", STATUS_ACTIVE);
             updateIndexRow(runs.get(tableConfig).getQuickSearchIndexId(), values);
+            priorStatus.remove(tableName);
 
             completeRunRecord(runs.get(tableConfig), RUN_COMPLETED, tableCounts.processed(), tableCounts.indexed(), 0, null);
          }
 
+         applyCapturedDeletes(client, tables);
+
          LOG.info("Full reindex complete", logPair("physicalIndex", physicalIndex), logPair("recordsIndexed", indexed));
          return (indexed);
       }
@@ -187,6 +219,9 @@ public class FullReindexStep extends AbstractIndexingStep
             }
          }
 
+         restoreStatus(priorStatus);
+         applyCapturedDeletes(client, tables);
+
          if(e instanceof QException qException)
          {
             throw qException;
@@ -197,6 +232,248 @@ public class FullReindexStep extends AbstractIndexingStep
 
 
 
+   /*******************************************************************************
+    ** Mark each table's index row REBUILDING before any of them is read,
+    ** remembering the status it had. A stale REBUILDING left by a crashed run
+    ** is remembered as ACTIVE.
+    *******************************************************************************/
+   private void markRebuilding(List<QuickSearchableTableConfig> tables, Map<String, String> priorStatus) throws QException
+   {
+      for(QuickSearchableTableConfig tableConfig : tables)
+      {
+         QRecord row    = queryIndexRow(tableConfig.getTableName());
+         String  status = row.getValueString("status");
+         priorStatus.put(tableConfig.getTableName(), status == null || STATUS_REBUILDING.equals(status) ? STATUS_ACTIVE : status);
+         updateIndexRow(row.getValueInteger("id"), Map.of("status", STATUS_REBUILDING));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** After a failed rebuild, put back the status of each table still marked
+    ** REBUILDING. Never throws.
+    *******************************************************************************/
+   private void restoreStatus(Map<String, String> priorStatus)
+   {
+      for(Map.Entry<String, String> entry : priorStatus.entrySet())
+      {
+         try
+         {
+            updateIndexRow(queryIndexId(entry.getKey()), Map.of("status", entry.getValue()));
+         }
+         catch(Exception e)
+         {
+            LOG.warn("Could not restore the index row status after a failed full reindex", e, logPair("tableName", entry.getKey()));
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Apply the deletes the listener captured during the rebuild (failed-event
+    ** rows with status AWAITING_REINDEX) through the alias, after the rebuild's
+    ** own writes, so an in-flight page cannot bring a deleted record back.
+    ** Called once the tables are no longer REBUILDING, so the listener stops
+    ** adding rows (one that races the status change hands its own rows to
+    ** basepull).
+    **
+    ** Rows are read in id order, past the previous batch, so a batch whose rows
+    ** could not be removed or updated (QQQ reports those as per-row errors, not
+    ** exceptions) is not read again; such rows stay AWAITING_REINDEX for the next
+    ** full reindex. Never throws.
+    *******************************************************************************/
+   void applyCapturedDeletes(QuickSearchOpenSearchClient client, List<QuickSearchableTableConfig> tables)
+   {
+      List<Serializable> tableNames = new ArrayList<>();
+      tables.forEach(tableConfig -> tableNames.add(tableConfig.getTableName()));
+      if(tableNames.isEmpty())
+      {
+         return;
+      }
+
+      Integer applied = 0;
+      try
+      {
+         QuickSearchQBitConfig config           = getConfig();
+         String                failedEventTable = config.applyPrefix(QuickSearchFailedEvent.TABLE_NAME);
+         Serializable          afterId          = null;
+
+         while(true)
+         {
+            QQueryFilter filter = new QQueryFilter(
+               new QFilterCriteria("status", QCriteriaOperator.EQUALS, QuickSearchFailedEvent.STATUS_AWAITING_REINDEX),
+               new QFilterCriteria("tableName", QCriteriaOperator.IN, tableNames))
+               .withOrderBy(new QFilterOrderBy("id", true))
+               .withLimit(config.getSourceBatchSize());
+            if(afterId != null)
+            {
+               filter.addCriteria(new QFilterCriteria("id", QCriteriaOperator.GREATER_THAN, afterId));
+            }
+
+            QueryInput queryInput = new QueryInput();
+            queryInput.setTableName(failedEventTable);
+            queryInput.setFilter(filter);
+            List<QRecord> rows = new QueryAction().execute(queryInput).getRecords();
+            if(CollectionUtils.nullSafeIsEmpty(rows))
+            {
+               break;
+            }
+
+            afterId = rows.get(rows.size() - 1).getValue("id");
+            try
+            {
+               applied = applied + applyCapturedDeleteBatch(client, config, failedEventTable, rows);
+            }
+            catch(Exception e)
+            {
+               LOG.warn("Could not apply a batch of deletes captured during the full reindex; they stay AWAITING_REINDEX for the next full reindex", e,
+                  logPair("tableNames", rows.stream().map(row -> row.getValueString("tableName")).distinct().toList()), logPair("count", rows.size()));
+            }
+         }
+      }
+      catch(Exception e)
+      {
+         LOG.warn("Could not apply deletes captured during the full reindex; they stay AWAITING_REINDEX for the next full reindex", e);
+      }
+
+      LOG.info("Applied deletes captured during the full reindex", logPair("count", applied));
+   }
+
+
+
+   /*******************************************************************************
+    ** Apply one batch of captured deletes; returns how many were applied.
+    ** A record that exists again (re-created with the same key since it was
+    ** deleted) keeps its document, and its row is just removed. Deleting a
+    ** missing document is harmless. Applied rows are removed; rows the cluster
+    ** rejects become PENDING for basepull to retry.
+    *******************************************************************************/
+   private Integer applyCapturedDeleteBatch(QuickSearchOpenSearchClient client, QuickSearchQBitConfig config, String failedEventTable, List<QRecord> rows) throws QException
+   {
+      Map<String, List<QRecord>> rowsByTable = new LinkedHashMap<>();
+      for(QRecord row : rows)
+      {
+         rowsByTable.computeIfAbsent(row.getValueString("tableName"), k -> new ArrayList<>()).add(row);
+      }
+
+      List<QRecord> finished = new ArrayList<>();
+      List<QRecord> toApply  = new ArrayList<>();
+      for(Map.Entry<String, List<QRecord>> entry : rowsByTable.entrySet())
+      {
+         Set<String> existingIds = queryExistingRecordIds(entry.getKey(), entry.getValue());
+         for(QRecord row : entry.getValue())
+         {
+            if(existingIds.contains(row.getValueString("recordId")))
+            {
+               finished.add(row);
+            }
+            else
+            {
+               toApply.add(row);
+            }
+         }
+      }
+
+      String error = null;
+      if(!toApply.isEmpty())
+      {
+         List<String> documentIds = new ArrayList<>();
+         for(QRecord row : toApply)
+         {
+            documentIds.add(OpenSearchDocument.buildDocumentId(row.getValueString("tableName"), row.getValueString("recordId")));
+         }
+
+         try
+         {
+            BulkIndexResult result = client.deleteDocuments(documentIds, config.getBulkBatchSize());
+            error = result.getFailureCount() == 0 ? null : String.join("; ", result.getErrors());
+         }
+         catch(Exception e)
+         {
+            error = e.getMessage() == null ? e.toString() : e.getMessage();
+         }
+
+         if(error == null)
+         {
+            finished.addAll(toApply);
+         }
+      }
+
+      if(!finished.isEmpty())
+      {
+         DeleteInput deleteInput = new DeleteInput();
+         deleteInput.setTableName(failedEventTable);
+         deleteInput.setPrimaryKeys(finished.stream().map(row -> row.getValue("id")).toList());
+         DeleteOutput deleteOutput = new DeleteAction().execute(deleteInput);
+         if(CollectionUtils.nullSafeHasContents(deleteOutput.getRecordsWithErrors()))
+         {
+            LOG.warn("Could not remove applied captured-delete rows; they stay AWAITING_REINDEX and are applied again by the next full reindex",
+               logPair("count", deleteOutput.getRecordsWithErrors().size()), logPair("firstError", deleteOutput.getRecordsWithErrors().get(0).getErrors()));
+         }
+      }
+
+      if(error != null)
+      {
+         List<QRecord> updates = new ArrayList<>();
+         for(QRecord row : toApply)
+         {
+            updates.add(new QRecord()
+               .withValue("id", row.getValue("id"))
+               .withValue("errorMessage", truncate(error, 4000))
+               .withValue("status", QuickSearchFailedEvent.STATUS_PENDING));
+         }
+
+         UpdateInput updateInput = new UpdateInput();
+         updateInput.setTableName(failedEventTable);
+         updateInput.setRecords(updates);
+         List<QRecord> notUpdated = CollectionUtils.nonNullList(new UpdateAction().execute(updateInput).getRecords()).stream()
+            .filter(record -> CollectionUtils.nullSafeHasContents(record.getErrors()))
+            .toList();
+         LOG.warn("Could not apply deletes captured during the full reindex; basepull will retry them", logPair("count", toApply.size()), logPair("error", error));
+         if(!notUpdated.isEmpty())
+         {
+            LOG.warn("Could not hand captured deletes to basepull; they stay AWAITING_REINDEX for the next full reindex",
+               logPair("count", notUpdated.size()), logPair("firstError", notUpdated.get(0).getErrors()));
+         }
+         return (0);
+      }
+
+      return (toApply.size());
+   }
+
+
+
+   /*******************************************************************************
+    ** The record IDs, among the rows' recordIds, that exist in the source table.
+    *******************************************************************************/
+   Set<String> queryExistingRecordIds(String tableName, List<QRecord> rows) throws QException
+   {
+      QuickSearchableTableConfig tableConfig = getTableConfig(tableName);
+      if(tableConfig == null)
+      {
+         return (Set.of());
+      }
+
+      String             primaryKeyField = tableConfig.getPrimaryKeyField() == null ? "id" : tableConfig.getPrimaryKeyField();
+      List<Serializable> 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<String> 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).
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStep.java

@@ -146,7 +146,13 @@ public class ReconcileIndexStep extends AbstractIndexingStep
             values.put("recordCount", counts.processed());
             values.put("documentCount", safeCount(client, tableName));
             values.put("searchableFieldsJson", buildSearchableFieldsJson(tableConfig));
-            values.put("status", STATUS_ACTIVE);
+            ////////////////////////////////////////////////////////////////////
+            // a full reindex running alongside owns the status until it ends //
+            ////////////////////////////////////////////////////////////////////
+            if(!STATUS_REBUILDING.equals(queryIndexRow(tableName).getValueString("status")))
+            {
+               values.put("status", STATUS_ACTIVE);
+            }
             if(ReconcileIndexStep.RUN_TYPE.equals(runType) == false)
             {
                values.put("lastFullReindexTime", Instant.now());

src/test/java/com/kingsrook/qbits/quicksearch/FullReindexDeleteReplayIntegrationTest.java

@@ -0,0 +1,244 @@
+/*
+ * Copyright 2024 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.quicksearch;
+
+
+import java.io.Serializable;
+import java.util.List;
+import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType;
+import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule;
+import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore;
+import com.kingsrook.qbits.quicksearch.actions.QuickSearchAction;
+import com.kingsrook.qbits.quicksearch.actions.QuickSearchInput;
+import com.kingsrook.qbits.quicksearch.actions.QuickSearchResult;
+import com.kingsrook.qbits.quicksearch.annotations.QuickSearchField;
+import com.kingsrook.qbits.quicksearch.annotations.QuickSearchable;
+import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent;
+import com.kingsrook.qbits.quicksearch.opensearch.QuickSearchOpenSearchClient;
+import com.kingsrook.qbits.quicksearch.processes.FullReindexStep;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/*******************************************************************************
+ ** Integration test, against a real OpenSearch, that a record deleted while a
+ ** full reindex rebuilds the index is gone from the new physical index after
+ ** the alias swap, even though the delete itself only reached the old one.
+ *******************************************************************************/
+@ExtendWith(RequiresDockerCondition.class)
+@Testcontainers
+class FullReindexDeleteReplayIntegrationTest
+{
+   private static final String BACKEND_NAME = "reindexDeleteMemoryBackend";
+   private static final String TABLE_NAME   = "reindexCustomer";
+   private static final String INDEX_NAME   = "quick_search_reindex_delete_test";
+
+   @Container
+   static GenericContainer<?> opensearch = new GenericContainer<>(System.getProperty("opensearch.test.image", "opensearchproject/opensearch:2.19.6"))
+      .withExposedPorts(9200)
+      .withEnv("discovery.type", "single-node")
+      .withEnv("DISABLE_SECURITY_PLUGIN", "true")
+      .withEnv("DISABLE_INSTALL_DEMO_CONFIG", "true")
+      .waitingFor(Wait.forHttp("/_cluster/health").forStatusCode(200));
+
+
+
+   /*******************************************************************************
+    ** Entity annotated for quick search indexing.
+    *******************************************************************************/
+   @QuickSearchable(tableName = TABLE_NAME)
+   static class ReindexCustomer
+   {
+      @QuickSearchField(weight = 2)
+      private String name;
+
+      @QuickSearchField
+      private String city;
+   }
+
+
+
+   /*******************************************************************************
+    ** Build the instance, produce the QBit with real-time indexing on and the
+    ** record-security re-read off (search returns whatever the index holds),
+    ** and initialize QContext.
+    *******************************************************************************/
+   @BeforeAll
+   static void setUpAll() throws Exception
+   {
+      MemoryRecordStore.getInstance().reset();
+
+      QInstance qInstance = new QInstance();
+      qInstance.addBackend(new QBackendMetaData()
+         .withName(BACKEND_NAME)
+         .withBackendType(MemoryBackendModule.class));
+      qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
+         .withName("anonymous")
+         .withType(QAuthenticationType.FULLY_ANONYMOUS));
+      qInstance.addTable(new QTableMetaData()
+         .withName(TABLE_NAME)
+         .withBackendName(BACKEND_NAME)
+         .withPrimaryKeyField("id")
+         .withField(new QFieldMetaData("id", QFieldType.INTEGER).withIsEditable(false))
+         .withField(new QFieldMetaData("name", QFieldType.STRING))
+         .withField(new QFieldMetaData("city", QFieldType.STRING)));
+
+      QuickSearchQBitConfig config = new QuickSearchQBitConfig()
+         .withBackendName(BACKEND_NAME)
+         .withOpensearchHost(opensearch.getHost())
+         .withOpensearchPort(opensearch.getMappedPort(9200))
+         .withOpensearchIndexName(INDEX_NAME)
+         .withSearchableEntityClasses(List.of(ReindexCustomer.class))
+         .withEnableRealTimeIndexing(true)
+         .withApplyRecordSecurityLocks(false);
+
+      new QuickSearchQBitProducer().withConfig(config).produce(qInstance);
+
+      QContext.init(qInstance, new QSession());
+   }
+
+
+
+   /*******************************************************************************
+    ** Clean up QContext and QBit state after all tests complete.
+    *******************************************************************************/
+   @AfterAll
+   static void tearDownAll()
+   {
+      QContext.clear();
+      QuickSearchQBitContext.clear();
+      MemoryRecordStore.getInstance().reset();
+   }
+
+
+
+   /*******************************************************************************
+    ** Test: a record deleted after the rebuild read it (its document is already
+    ** in the new physical index) is absent from the index and from search once
+    ** the rebuild has swapped the alias.
+    *******************************************************************************/
+   @Test
+   void testDeleteDuringFullReindex_isAbsentAfterTheSwap() throws Exception
+   {
+      InsertInput insertInput = new InsertInput();
+      insertInput.setTableName(TABLE_NAME);
+      insertInput.setRecords(List.of(
+         new QRecord().withValue("id", 1).withValue("name", "Gwendolyn").withValue("city", "Rebuildton"),
+         new QRecord().withValue("id", 2).withValue("name", "Stanford").withValue("city", "Rebuildton")));
+      new InsertAction().execute(insertInput);
+      assertThat(searchRecordIds("rebuildton")).containsExactlyInAnyOrder("1", "2");
+
+      RunBackendStepOutput output = new RunBackendStepOutput();
+      new FullReindexStep()
+      {
+         ///////////////////////////////////////////////////////////////////////
+         // delete a record once its document is in the new physical index,  //
+         // while the alias still points at the old one                       //
+         ///////////////////////////////////////////////////////////////////////
+         @Override
+         protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List<QFilterCriteria> extraCriteria, String targetIndex) throws QException
+         {
+            IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex);
+
+            DeleteInput deleteInput = new DeleteInput();
+            deleteInput.setTableName(TABLE_NAME);
+            deleteInput.setPrimaryKeys(List.<Serializable>of(1));
+            new DeleteAction().execute(deleteInput);
+
+            return (counts);
+         }
+      }.run(new RunBackendStepInput(), output);
+
+      assertThat(output.getValueBoolean("aliasSwapped")).isTrue();
+      assertThat(documentRecordIds("rebuildton"))
+         .as("the deleted record's document must not survive the rebuild")
+         .containsExactly("2");
+      assertThat(searchRecordIds("rebuildton")).containsExactly("2");
+
+      QueryInput queryInput = new QueryInput();
+      queryInput.setTableName(QuickSearchFailedEvent.TABLE_NAME);
+      assertThat(new QueryAction().execute(queryInput).getRecords())
+         .as("the captured delete was applied and its row removed")
+         .isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Refresh the index, then return the record IDs search returns for the term.
+    *******************************************************************************/
+   private List<String> searchRecordIds(String term) throws QException
+   {
+      client().refreshIndex();
+      return (new QuickSearchAction().execute(new QuickSearchInput().withSearchTerm(term).withTableName(TABLE_NAME))
+         .getResults().stream()
+         .map(QuickSearchResult::getRecordId)
+         .toList());
+   }
+
+
+
+   /*******************************************************************************
+    ** Refresh the index, then return the record IDs of the documents matching
+    ** the term, straight from OpenSearch.
+    *******************************************************************************/
+   private List<String> documentRecordIds(String term) throws QException
+   {
+      client().refreshIndex();
+      return (client().search(term, List.of(TABLE_NAME), 10, 0, QuickSearchQBitContext.getDiscoveredTables())
+         .hits().hits().stream()
+         .map(hit -> hit.source().getRecordId())
+         .toList());
+   }
+
+
+
+   /*******************************************************************************
+    ** The live OpenSearch client from the QBit context.
+    *******************************************************************************/
+   private QuickSearchOpenSearchClient client()
+   {
+      return (QuickSearchQBitContext.getClient());
+   }
+
+}

src/test/java/com/kingsrook/qbits/quicksearch/listeners/QuickSearchRecordChangeListenerTest.java

@@ -41,6 +41,7 @@ import com.kingsrook.qbits.quicksearch.QuickSearchRuntime;
 import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
 import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent;
 import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex;
+import com.kingsrook.qbits.quicksearch.processes.AbstractIndexingStep;
 import com.kingsrook.qbits.quicksearch.publisher.IndexEvent;
 import com.kingsrook.qbits.quicksearch.publisher.IndexEventAction;
 import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher;
@@ -49,6 +50,7 @@ import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -230,4 +232,138 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest
       verify(publisher, never()).publishIndexEvents(anyList());
    }
 
+
+
+   @Test
+   void testDelete_duringFullReindex_isCapturedForTheNewIndex() throws QException
+   {
+      insertIndexRowWithStatus(AbstractIndexingStep.STATUS_REBUILDING);
+      Integer id = insert("widget", null);
+      delete(id);
+
+      verify(publisher, times(1)).publishDeleteEvents(anyList());
+
+      List<QRecord> captured = queryFailedEvents();
+      assertThat(captured).hasSize(1);
+      assertThat(captured.get(0).getValueString("tableName")).isEqualTo(TEST_ENTITY_TABLE);
+      assertThat(captured.get(0).getValueString("recordId")).isEqualTo(String.valueOf(id));
+      assertThat(captured.get(0).getValueString("action")).isEqualTo("DELETE");
+      assertThat(captured.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX);
+   }
+
+
+
+   @Test
+   void testDelete_publishFailureDuringFullReindex_recordsBothRows() throws QException
+   {
+      doThrow(new QException("cluster down")).when(publisher).publishDeleteEvents(anyList());
+      insertIndexRowWithStatus(AbstractIndexingStep.STATUS_REBUILDING);
+      delete(insert("widget", null));
+
+      assertThat(queryFailedEvents()).extracting(r -> r.getValueString("status"))
+         .containsExactlyInAnyOrder(QuickSearchFailedEvent.STATUS_PENDING, QuickSearchFailedEvent.STATUS_AWAITING_REINDEX);
+   }
+
+
+
+   @Test
+   void testDelete_withoutFullReindex_isNotCaptured() throws QException
+   {
+      insertIndexRowWithStatus(AbstractIndexingStep.STATUS_ACTIVE);
+      delete(insert("widget", null));
+
+      verify(publisher, times(1)).publishDeleteEvents(anyList());
+      assertThat(queryFailedEvents()).isEmpty();
+   }
+
+
+
+   @Test
+   void testCapture_rebuildFinishedBeforeInsert_handsRowsToBasepull() throws QException
+   {
+      ////////////////////////////////////////////////////////////////////////
+      // the second read sees the rebuild finished, as if it ended between //
+      // the first read and the insert                                     //
+      ////////////////////////////////////////////////////////////////////////
+      insertIndexRowWithStatus(AbstractIndexingStep.STATUS_ACTIVE);
+      List<IndexEvent> events = List.of(new IndexEvent().withAction(IndexEventAction.DELETE).withTableName(TEST_ENTITY_TABLE).withRecordId("7"));
+
+      List<QRecord> inserted = QuickSearchRecordChangeListener.insertFailedEventRows(runtime, events, "captured", QuickSearchFailedEvent.STATUS_AWAITING_REINDEX);
+      QuickSearchRecordChangeListener.handOverIfRebuildFinished(runtime, TEST_ENTITY_TABLE, inserted);
+
+      List<QRecord> rows = queryFailedEvents();
+      assertThat(rows).hasSize(1);
+      assertThat(rows.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING);
+   }
+
+
+
+   private void delete(Integer id) throws QException
+   {
+      DeleteInput deleteInput = new DeleteInput();
+      deleteInput.setTableName(TEST_ENTITY_TABLE);
+      deleteInput.setPrimaryKeys(List.of(id));
+      new DeleteAction().execute(deleteInput);
+   }
+
+
+
+   private void insertIndexRowWithStatus(String status) throws QException
+   {
+      InsertInput insertInput = new InsertInput();
+      insertInput.setTableName(QuickSearchIndex.TABLE_NAME);
+      insertInput.setRecords(List.of(new QRecord().withValue("tableName", TEST_ENTITY_TABLE).withValue("enabled", true).withValue("status", status)));
+      new InsertAction().execute(insertInput);
+      runtime.invalidateEnabledCache();
+   }
+
+
+
+   private List<QRecord> queryFailedEvents() throws QException
+   {
+      QueryInput queryInput = new QueryInput();
+      queryInput.setTableName(QuickSearchFailedEvent.TABLE_NAME);
+      return (new QueryAction().execute(queryInput).getRecords());
+   }
+
+
+
+   @Test
+   void testDelete_rebuildEndsDuringPublish_isStillCaptured() throws QException
+   {
+      ////////////////////////////////////////////////////////////////////////////
+      // the reindex swaps the alias and clears REBUILDING while the delete is //
+      // being published to the old index: the status read must come first    //
+      ////////////////////////////////////////////////////////////////////////////
+      insertIndexRowWithStatus(AbstractIndexingStep.STATUS_REBUILDING);
+      Integer id = insert("widget", null);
+      doAnswer(invocation ->
+      {
+         setIndexRowStatus(AbstractIndexingStep.STATUS_ACTIVE);
+         return (null);
+      }).when(publisher).publishDeleteEvents(anyList());
+
+      delete(id);
+
+      List<QRecord> rows = queryFailedEvents();
+      assertThat(rows).hasSize(1);
+      assertThat(rows.get(0).getValueString("recordId")).isEqualTo(String.valueOf(id));
+      assertThat(rows.get(0).getValueString("action")).isEqualTo("DELETE");
+      assertThat(rows.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING);
+   }
+
+
+
+   private void setIndexRowStatus(String status) throws QException
+   {
+      QueryInput queryInput = new QueryInput();
+      queryInput.setTableName(QuickSearchIndex.TABLE_NAME);
+      QRecord row = new QueryAction().execute(queryInput).getRecords().get(0);
+
+      UpdateInput updateInput = new UpdateInput();
+      updateInput.setTableName(QuickSearchIndex.TABLE_NAME);
+      updateInput.setRecords(List.of(new QRecord().withValue("id", row.getValue("id")).withValue("status", status)));
+      new UpdateAction().execute(updateInput);
+   }
+
 }

src/test/java/com/kingsrook/qbits/quicksearch/processes/BasepullIndexStepTest.java

@@ -864,4 +864,108 @@ class BasepullIndexStepTest extends BaseQuickSearchTest
       assertThat(new BasepullIndexStep().isDueForBasepull(index)).isTrue();
    }
 
+
+
+   /*******************************************************************************
+    ** Test: a delete captured during a full reindex is left for that reindex
+    ** to apply to the new index, not replayed against the old one.
+    *******************************************************************************/
+   @Test
+   void testReplayFailedEvents_awaitingReindexRowsAreNotReplayed() throws QException
+   {
+      insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60);
+      QRecord captured = insertFailedEvent(TEST_ENTITY_TABLE, "7", "DELETE", 0);
+      UpdateInput updateInput = new UpdateInput();
+      updateInput.setTableName(QuickSearchFailedEvent.TABLE_NAME);
+      updateInput.setRecords(List.of(new QRecord().withValue("id", captured.getValue("id")).withValue("status", QuickSearchFailedEvent.STATUS_AWAITING_REINDEX)));
+      new UpdateAction().execute(updateInput);
+
+      RunBackendStepOutput output = new RunBackendStepOutput();
+      new BasepullIndexStep().run(new RunBackendStepInput(), output);
+
+      verify(mockClient, never()).deleteDocuments(anyList(), anyInt());
+      assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).hasSize(1);
+      assertThat(output.getValueInteger("failedEventsReplayed")).isEqualTo(0);
+   }
+
+
+
+   /*******************************************************************************
+    ** Test: drift found while a full reindex is running does not replace the
+    ** REBUILDING status the listener relies on.
+    *******************************************************************************/
+   @Test
+   void testDriftDetection_duringFullReindex_keepsRebuilding() throws QException
+   {
+      QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60);
+      UpdateInput updateInput = new UpdateInput();
+      updateInput.setTableName(QuickSearchIndex.TABLE_NAME);
+      updateInput.setRecords(List.of(new QRecord().withValue("id", row.getValue("id"))
+         .withValue("status", AbstractIndexingStep.STATUS_REBUILDING)
+         .withValue("searchableFieldsJson", "[{\"fieldName\":\"name\",\"weight\":1,\"includeLabel\":false}]")));
+      new UpdateAction().execute(updateInput);
+
+      new BasepullIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
+
+      assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING);
+   }
+
+
+
+   /*******************************************************************************
+    ** Test: drift found on a row snapshot taken before a full reindex started
+    ** does not overwrite the REBUILDING status now in the database.
+    *******************************************************************************/
+   @Test
+   void testDriftDetection_staleSnapshot_keepsRebuilding() throws QException
+   {
+      QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60);
+      UpdateInput updateInput = new UpdateInput();
+      updateInput.setTableName(QuickSearchIndex.TABLE_NAME);
+      updateInput.setRecords(List.of(new QRecord().withValue("id", row.getValue("id")).withValue("status", AbstractIndexingStep.STATUS_REBUILDING)));
+      new UpdateAction().execute(updateInput);
+
+      QRecord staleSnapshot = new QRecord()
+         .withValue("id", row.getValue("id"))
+         .withValue("tableName", TEST_ENTITY_TABLE)
+         .withValue("status", AbstractIndexingStep.STATUS_ACTIVE)
+         .withValue("searchableFieldsJson", "[{\"fieldName\":\"name\",\"weight\":1,\"includeLabel\":false}]");
+      QuickSearchableTableConfig tableConfig = QuickSearchQBitContext.getDiscoveredTables().stream()
+         .filter(table -> TEST_ENTITY_TABLE.equals(table.getTableName()))
+         .findFirst().orElseThrow();
+
+      new BasepullIndexStep().detectDrift(staleSnapshot, tableConfig);
+      assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING);
+   }
+
+
+
+   /*******************************************************************************
+    ** Test: a stale snapshot shows drift, but a full reindex has since recorded
+    ** the current fields; the row is not marked NEEDS_REINDEX.
+    *******************************************************************************/
+   @Test
+   void testDriftDetection_staleSnapshot_reindexSinceRecordedCurrentFields() throws QException
+   {
+      QuickSearchableTableConfig tableConfig = QuickSearchQBitContext.getDiscoveredTables().stream()
+         .filter(table -> TEST_ENTITY_TABLE.equals(table.getTableName()))
+         .findFirst().orElseThrow();
+      QRecord row = insertIndexRow(TEST_ENTITY_TABLE, true, Instant.now(), 60);
+      UpdateInput updateInput = new UpdateInput();
+      updateInput.setTableName(QuickSearchIndex.TABLE_NAME);
+      updateInput.setRecords(List.of(new QRecord().withValue("id", row.getValue("id"))
+         .withValue("status", AbstractIndexingStep.STATUS_ACTIVE)
+         .withValue("searchableFieldsJson", AbstractIndexingStep.buildSearchableFieldsJson(tableConfig))));
+      new UpdateAction().execute(updateInput);
+
+      QRecord staleSnapshot = new QRecord()
+         .withValue("id", row.getValue("id"))
+         .withValue("tableName", TEST_ENTITY_TABLE)
+         .withValue("status", AbstractIndexingStep.STATUS_ACTIVE)
+         .withValue("searchableFieldsJson", "[{\"fieldName\":\"name\",\"weight\":1,\"includeLabel\":false}]");
+
+      assertThat(new BasepullIndexStep().detectDrift(staleSnapshot, tableConfig)).isFalse();
+      assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
+   }
+
 }

src/test/java/com/kingsrook/qbits/quicksearch/processes/FullReindexStepTest.java

@@ -19,19 +19,28 @@ package com.kingsrook.qbits.quicksearch.processes;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
 import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
 import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteOutput;
 import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.statusmessages.SystemErrorStatusMessage;
 import com.kingsrook.qbits.quicksearch.BaseQuickSearchTest;
 import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
+import com.kingsrook.qbits.quicksearch.QuickSearchableTableConfig;
+import com.kingsrook.qbits.quicksearch.model.QuickSearchFailedEvent;
 import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex;
 import com.kingsrook.qbits.quicksearch.model.QuickSearchIndexRun;
 import com.kingsrook.qbits.quicksearch.opensearch.BulkIndexResult;
@@ -41,6 +50,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.InOrder;
+import org.mockito.MockedConstruction;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.mockito.ArgumentMatchers.any;
@@ -51,6 +61,7 @@ import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -474,4 +485,216 @@ class FullReindexStepTest extends BaseQuickSearchTest
       verify(mockClient, times(3)).indexDocuments(eq(PHYSICAL_INDEX), anyList(), anyInt());
    }
 
+
+
+   /*******************************************************************************
+    ** Test: the index row reads REBUILDING while tables are read, and a delete
+    ** captured during the rebuild is applied through the alias after the swap,
+    ** then its failed-event row is removed.
+    *******************************************************************************/
+   @Test
+   void testAllTables_appliesCapturedDeletesAfterTheSwap() throws QException
+   {
+      insertTestEntities(3);
+      when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1));
+
+      List<String> statusesDuringRebuild = new ArrayList<>();
+      new FullReindexStep()
+      {
+         @Override
+         protected IndexCounts indexAllRecords(QuickSearchOpenSearchClient client, QuickSearchableTableConfig tableConfig, List<QFilterCriteria> extraCriteria, String targetIndex) throws QException
+         {
+            IndexCounts counts = super.indexAllRecords(client, tableConfig, extraCriteria, targetIndex);
+            statusesDuringRebuild.add(FullReindexStepTest.this.queryIndexRow().getValueString("status"));
+            insertCapturedDelete("7");
+            return (counts);
+         }
+      }.run(new RunBackendStepInput(), new RunBackendStepOutput());
+
+      assertThat(statusesDuringRebuild).containsExactly(AbstractIndexingStep.STATUS_REBUILDING);
+
+      InOrder inOrder = inOrder(mockClient);
+      inOrder.verify(mockClient).swapAliasTo(PHYSICAL_INDEX);
+      inOrder.verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":7")), anyInt());
+
+      assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty();
+      assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_ACTIVE);
+   }
+
+
+
+   /*******************************************************************************
+    ** Test: a captured delete the cluster rejects is handed to basepull.
+    *******************************************************************************/
+   @Test
+   void testAllTables_capturedDeleteFailure_becomesPendingForBasepull() throws QException
+   {
+      insertTestEntities(1);
+      insertCapturedDelete("9");
+      when(mockClient.deleteDocuments(anyList(), anyInt())).thenThrow(new QException("cluster down"));
+
+      new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
+
+      List<QRecord> rows = queryAll(QuickSearchFailedEvent.TABLE_NAME);
+      assertThat(rows).hasSize(1);
+      assertThat(rows.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING);
+      assertThat(rows.get(0).getValueString("errorMessage")).contains("cluster down");
+   }
+
+
+
+   /*******************************************************************************
+    ** Test: when the rebuild fails before the swap, the previous status comes
+    ** back and the captured deletes still leave the AWAITING_REINDEX state.
+    *******************************************************************************/
+   @Test
+   void testAllTables_failureBeforeSwap_restoresStatusAndDrainsCapturedDeletes() throws QException
+   {
+      insertTestEntities(1);
+      insertIndexRow(true);
+      updateIndexRowStatus(AbstractIndexingStep.STATUS_NEEDS_REINDEX);
+      insertCapturedDelete("4");
+      when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1));
+      doThrow(new QException("swap refused")).when(mockClient).swapAliasTo(PHYSICAL_INDEX);
+
+      assertThatThrownBy(() -> new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput()))
+         .isInstanceOf(QException.class);
+
+      assertThat(queryIndexRow().getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_NEEDS_REINDEX);
+      assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty();
+      verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":4")), anyInt());
+   }
+
+
+
+   /*******************************************************************************
+    ** Insert an AWAITING_REINDEX delete for testEntity, as the listener does
+    ** during a full reindex.
+    *******************************************************************************/
+   private void insertCapturedDelete(String recordId) throws QException
+   {
+      InsertInput insertInput = new InsertInput();
+      insertInput.setTableName(QuickSearchFailedEvent.TABLE_NAME);
+      insertInput.setRecords(List.of(new QRecord()
+         .withValue("tableName", TEST_ENTITY_TABLE)
+         .withValue("recordId", recordId)
+         .withValue("action", "DELETE")
+         .withValue("attempts", 0)
+         .withValue("status", QuickSearchFailedEvent.STATUS_AWAITING_REINDEX)));
+      new InsertAction().execute(insertInput);
+   }
+
+
+
+   /*******************************************************************************
+    ** Set the status on the testEntity quickSearchIndex row.
+    *******************************************************************************/
+   private void updateIndexRowStatus(String status) throws QException
+   {
+      UpdateInput updateInput = new UpdateInput();
+      updateInput.setTableName(QuickSearchIndex.TABLE_NAME);
+      updateInput.setRecords(List.of(new QRecord().withValue("id", queryIndexRow().getValue("id")).withValue("status", status)));
+      new UpdateAction().execute(updateInput);
+   }
+
+
+
+   /*******************************************************************************
+    ** Test: a captured delete for a record re-created with the same key since
+    ** is not applied (its document stays), and its row is removed.
+    *******************************************************************************/
+   @Test
+   void testAllTables_capturedDeleteOfRecreatedRecord_isNotApplied() throws QException
+   {
+      insertTestEntities(3);
+      insertCapturedDelete("2");
+      insertCapturedDelete("9");
+      when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1));
+
+      new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
+
+      verify(mockClient, times(1)).deleteDocuments(anyList(), anyInt());
+      verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":9")), anyInt());
+      assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Test: rows the delete action reports as not removed (per-row errors, not
+    ** an exception) are not read again; the drain moves on and ends.
+    *******************************************************************************/
+   @Test
+   void testAllTables_capturedRowsNotRemoved_drainStillEnds() throws QException
+   {
+      QuickSearchQBitContext.getConfig().withSourceBatchSize(1);
+      insertTestEntities(1);
+      insertCapturedDelete("8");
+      insertCapturedDelete("9");
+
+      AtomicInteger deleteCalls = new AtomicInteger(0);
+      when(mockClient.deleteDocuments(anyList(), anyInt())).thenAnswer(invocation ->
+      {
+         ///////////////////////////////////////////////////////////////////
+         // fail loudly instead of hanging if the same batch is re-read  //
+         ///////////////////////////////////////////////////////////////////
+         if(deleteCalls.incrementAndGet() > 10)
+         {
+            throw (new IllegalStateException("drain re-read the same rows"));
+         }
+         return (new BulkIndexResult().withSuccessCount(1));
+      });
+
+      try(MockedConstruction<DeleteAction> deleteActions = mockConstruction(DeleteAction.class, (deleteAction, context) ->
+      {
+         DeleteOutput output = new DeleteOutput();
+         output.setRecordsWithErrors(List.of(new QRecord().withValue("id", 1).withError(new SystemErrorStatusMessage("row is locked"))));
+         when(deleteAction.execute(any())).thenReturn(output);
+      }))
+      {
+         new FullReindexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
+      }
+
+      assertThat(deleteCalls.get()).isEqualTo(2);
+      assertThat(queryAll(QuickSearchFailedEvent.TABLE_NAME)).extracting(row -> row.getValueString("status"))
+         .containsOnly(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX);
+   }
+
+
+
+   /*******************************************************************************
+    ** Test: a batch that fails before it is applied (here, the source lookup)
+    ** stays AWAITING_REINDEX, and the later batches are still applied.
+    *******************************************************************************/
+   @Test
+   void testAllTables_failedCapturedBatch_laterBatchesStillApplied() throws QException
+   {
+      QuickSearchQBitContext.getConfig().withSourceBatchSize(1);
+      insertTestEntities(1);
+      insertCapturedDelete("8");
+      insertCapturedDelete("9");
+      when(mockClient.deleteDocuments(anyList(), anyInt())).thenReturn(new BulkIndexResult().withSuccessCount(1));
+
+      new FullReindexStep()
+      {
+         @Override
+         Set<String> queryExistingRecordIds(String tableName, List<QRecord> rows) throws QException
+         {
+            if("8".equals(rows.get(0).getValueString("recordId")))
+            {
+               throw (new QException("source table unavailable"));
+            }
+            return (super.queryExistingRecordIds(tableName, rows));
+         }
+      }.run(new RunBackendStepInput(), new RunBackendStepOutput());
+
+      verify(mockClient, times(1)).deleteDocuments(anyList(), anyInt());
+      verify(mockClient).deleteDocuments(eq(List.of(TEST_ENTITY_TABLE + ":9")), anyInt());
+
+      List<QRecord> rows = queryAll(QuickSearchFailedEvent.TABLE_NAME);
+      assertThat(rows).hasSize(1);
+      assertThat(rows.get(0).getValueString("recordId")).isEqualTo("8");
+      assertThat(rows.get(0).getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_AWAITING_REINDEX);
+   }
+
 }

src/test/java/com/kingsrook/qbits/quicksearch/processes/ReconcileIndexStepTest.java

@@ -386,4 +386,28 @@ class ReconcileIndexStepTest extends BaseQuickSearchTest
       assertThat(row.getValueString("lastErrorMessage")).contains("connection refused");
    }
 
+
+
+   /*******************************************************************************
+    ** Test: a reconcile finishing while a full reindex runs leaves the
+    ** REBUILDING status in place, so deletes are still captured.
+    *******************************************************************************/
+   @Test
+   void testReconcile_duringFullReindex_keepsRebuilding() throws QException
+   {
+      insertTestEntities(1);
+      InsertInput insertInput = new InsertInput();
+      insertInput.setTableName(QuickSearchIndex.TABLE_NAME);
+      insertInput.setRecords(List.of(new QRecord()
+         .withValue("tableName", TEST_ENTITY_TABLE)
+         .withValue("enabled", true)
+         .withValue("status", AbstractIndexingStep.STATUS_REBUILDING)));
+      new InsertAction().execute(insertInput);
+
+      new ReconcileIndexStep().run(new RunBackendStepInput(), new RunBackendStepOutput());
+
+      assertThat(queryAll(QuickSearchIndex.TABLE_NAME).get(0).getValueString("status")).isEqualTo(AbstractIndexingStep.STATUS_REBUILDING);
+      assertThat(indexedRecordIds()).containsExactly("1");
+   }
+
 }