src/main/java/com/kingsrook/qbits/quicksearch/QuickSearchRuntime.java

@@ -338,7 +338,10 @@ public class QuickSearchRuntime
 
    /*******************************************************************************
     ** Start the runtime per the configured startup mode: FAIL_FAST throws when
-    ** the index cannot be prepared; DEGRADED logs and continues.
+    ** the index cannot be prepared; DEGRADED logs and continues. A LinkageError
+    ** (an optional library such as the AWS SDK missing while the client is
+    ** built) follows the same modes, with a message naming the missing class,
+    ** so a DEGRADED host still boots.
     *******************************************************************************/
    public void start() throws QException
    {
@@ -346,6 +349,18 @@ public class QuickSearchRuntime
       {
          ensureIndexReady();
       }
+      catch(LinkageError e)
+      {
+         String message = "Quick Search could not load a class to build its OpenSearch client [" + e.getMessage()
+            + "]; authMode AWS_SIGV4 needs the optional AWS SDK v2 dependencies (apache-client, sts) on the classpath";
+         if(config.getStartupMode() == QuickSearchStartupMode.DEGRADED)
+         {
+            LOG.warn(message + "; Quick Search is running degraded", e, logPair("indexName", config.getOpensearchIndexName()));
+            return;
+         }
+
+         throw (new QException(message, e));
+      }
       catch(Exception e)
       {
          if(config.getStartupMode() == QuickSearchStartupMode.DEGRADED)

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

@@ -57,9 +57,12 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
  ** is already committed, so it publishes immediately. A rolled-back write never
  ** reaches the index.
  **
- ** 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.
+ ** This listener never fails the host's write. When the re-read of updated
+ ** records or publishing fails (including a LinkageError from a missing
+ ** optional library while the client is built), each affected (table, record,
+ ** action) is written to quickSearchFailedEvent for the scheduled basepull to
+ ** replay. A failed re-read is recorded after the commit too, so a rolled-back
+ ** write leaves no row.
  **
  ** 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.
@@ -113,7 +116,27 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
       switch(event.getType())
       {
          case INSERT -> indexEvents.addAll(toIndexEvents(tableName, primaryKeyField, event.getRecords()));
-         case UPDATE -> indexEvents.addAll(toIndexEvents(tableName, primaryKeyField, fetchCurrentRecords(event, primaryKeyField)));
+         case UPDATE ->
+         {
+            try
+            {
+               indexEvents.addAll(toIndexEvents(tableName, primaryKeyField, fetchCurrentRecords(event, primaryKeyField)));
+            }
+            catch(Exception | LinkageError e)
+            {
+               ///////////////////////////////////////////////////////////////////////
+               // log now: with a transaction, nothing else is logged until commit, //
+               // and a rollback (a transaction the failed re-read aborted, too)    //
+               // would leave no trace. Then record the updated keys after commit,  //
+               // so basepull re-reads and indexes them.                            //
+               ///////////////////////////////////////////////////////////////////////
+               LOG.warn("Could not re-read updated records for indexing; recording them for replay once the write commits", e,
+                  logPair("tableName", tableName), logPair("recordCount", event.getRecords().size()));
+               List<IndexEvent> failedEvents = toIndexEvents(tableName, primaryKeyField, event.getRecords());
+               runAfterCommit(event, () -> recordFailures(runtime, failedEvents, e));
+               return;
+            }
+         }
          case DELETE -> deleteEvents.addAll(toDeleteEvents(tableName, primaryKeyField, event.getRecords()));
       }
 
@@ -122,15 +145,24 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
          return;
       }
 
-      Runnable publish = () -> publish(runtime, indexEvents, deleteEvents);
+      runAfterCommit(event, () -> publish(runtime, indexEvents, deleteEvents));
+   }
 
+
+
+   /*******************************************************************************
+    ** Run work once the event's transaction commits, or now when the write had
+    ** no transaction (it is already committed).
+    *******************************************************************************/
+   private static void runAfterCommit(RecordChangeEvent event, Runnable work)
+   {
       if(event.getTransaction() != null)
       {
-         event.getTransaction().addAfterCommitCallback(publish);
+         event.getTransaction().addAfterCommitCallback(work);
       }
       else
       {
-         publish.run();
+         work.run();
       }
    }
 
@@ -161,7 +193,10 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
 
 
    /*******************************************************************************
-    **
+    ** A LinkageError is caught along with Exception: building the lazy client,
+    ** or a host publisher, can hit an optional library that is not on the
+    ** classpath (the AWS SDK for AWS_SIGV4), which a DEGRADED start leaves for
+    ** the first write to find.
     *******************************************************************************/
    private static void publishOrRecordFailures(QuickSearchRuntime runtime, List<IndexEvent> indexEvents, List<IndexEvent> deleteEvents)
    {
@@ -170,7 +205,7 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
       {
          publisher = runtime.getPublisher();
       }
-      catch(Exception e)
+      catch(Exception | LinkageError e)
       {
          recordFailures(runtime, indexEvents, e);
          recordFailures(runtime, deleteEvents, e);
@@ -183,7 +218,7 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
          {
             publisher.publishIndexEvents(indexEvents);
          }
-         catch(Exception e)
+         catch(Exception | LinkageError e)
          {
             recordFailures(runtime, indexEvents, e);
          }
@@ -195,7 +230,7 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
          {
             publisher.publishDeleteEvents(deleteEvents);
          }
-         catch(Exception e)
+         catch(Exception | LinkageError e)
          {
             recordFailures(runtime, deleteEvents, e);
          }
@@ -205,9 +240,9 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
 
 
    /*******************************************************************************
-    ** Write one quickSearchFailedEvent row per event.
+    ** Write one quickSearchFailedEvent row per event. Never throws.
     *******************************************************************************/
-   static void recordFailures(QuickSearchRuntime runtime, List<IndexEvent> events, Exception cause)
+   static void recordFailures(QuickSearchRuntime runtime, List<IndexEvent> events, Throwable cause)
    {
       if(events.isEmpty())
       {
@@ -215,7 +250,7 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
       }
 
       String message = cause.getMessage() == null ? cause.getClass().getSimpleName() : cause.getMessage();
-      LOG.warn("Real-time index publish failed; recording failed events for replay", cause,
+      LOG.warn("Real-time indexing failed; recording failed events for replay", cause,
          logPair("tableName", events.get(0).getTableName()), logPair("eventCount", events.size()));
 
       try
@@ -411,7 +446,8 @@ public class QuickSearchRecordChangeListener implements RecordChangeListenerInte
    /*******************************************************************************
     ** Updated records are "as the backend returned them", which for RDBMS is the
     ** sparse input. Re-read the full rows inside the same transaction, with
-    ** display values so possible-value labels are indexed.
+    ** display values so possible-value labels are indexed (which can run host
+    ** possible-value providers).
     *******************************************************************************/
    private static List<QRecord> fetchCurrentRecords(RecordChangeEvent event, String primaryKeyField) throws QException
    {

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

@@ -0,0 +1,43 @@
+/*
+ * 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 com.kingsrook.qbits.quicksearch.opensearch.OpenSearchTransportCustomizer;
+import org.opensearch.client.transport.httpclient5.ApacheHttpClient5TransportBuilder;
+
+
+/*******************************************************************************
+ ** Test transport customizer that fails linkage while the OpenSearch client is
+ ** built, as a missing optional library (the AWS SDK for AWS_SIGV4) does.
+ *******************************************************************************/
+public class MissingLibraryTransportCustomizer implements OpenSearchTransportCustomizer
+{
+   public static final String MISSING_CLASS = "software/amazon/awssdk/http/SdkHttpClient";
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   @Override
+   public ApacheHttpClient5TransportBuilder customizeTransport(ApacheHttpClient5TransportBuilder builder)
+   {
+      throw (new NoClassDefFoundError(MISSING_CLASS));
+   }
+
+}

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

@@ -22,14 +22,18 @@ import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
 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.logging.QCollectingLogger;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
 import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
 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.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
 import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData;
 import com.kingsrook.qbits.quicksearch.model.QuickSearchIndex;
 import com.kingsrook.qbits.quicksearch.publisher.IndexEventPublisher;
 import com.kingsrook.qbits.quicksearch.publisher.SynchronousIndexEventPublisher;
+import org.apache.logging.log4j.Level;
 import org.junit.jupiter.api.Test;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatCode;
@@ -153,4 +157,38 @@ class QuickSearchRuntimeTest extends BaseQuickSearchTest
       failFast.close();
    }
 
+
+
+   @Test
+   void testStart_missingOptionalLibrary_degradedBoots_failFastThrows()
+   {
+      QCodeReference customizer = new QCodeReference(MissingLibraryTransportCustomizer.class);
+
+      QuickSearchRuntime degraded         = new QuickSearchRuntime(new QuickSearchQBitConfig().withBackendName(TEST_BACKEND_NAME).withOpensearchHost("localhost").withOpensearchPort(9200).withOpensearchIndexName("t").withStartupMode(QuickSearchStartupMode.DEGRADED).withTransportCustomizer(customizer));
+      QCollectingLogger  collectingLogger = QLogger.activateCollectingLoggerForClass(QuickSearchRuntime.class);
+      try
+      {
+         assertThatCode(degraded::start).doesNotThrowAnyException();
+      }
+      finally
+      {
+         QLogger.deactivateCollectingLoggerForClass(QuickSearchRuntime.class);
+      }
+      assertThat(degraded.isIndexReady()).isFalse();
+      assertThat(collectingLogger.getCollectedMessages()).anySatisfy(m ->
+      {
+         assertThat(m.getLevel()).isEqualTo(Level.WARN);
+         assertThat(m.getMessage()).contains(MissingLibraryTransportCustomizer.MISSING_CLASS).contains("AWS SDK v2").contains("apache-client, sts").doesNotContain("not reachable");
+      });
+      degraded.close();
+
+      QuickSearchRuntime failFast = new QuickSearchRuntime(new QuickSearchQBitConfig().withBackendName(TEST_BACKEND_NAME).withOpensearchHost("localhost").withOpensearchPort(9200).withOpensearchIndexName("t").withTransportCustomizer(customizer));
+      assertThatThrownBy(failFast::start).isInstanceOf(QException.class)
+         .hasMessageContaining(MissingLibraryTransportCustomizer.MISSING_CLASS)
+         .hasMessageContaining("AWS SDK v2 dependencies (apache-client, sts)")
+         .hasMessageNotContaining("startupMode=DEGRADED")
+         .hasCauseInstanceOf(NoClassDefFoundError.class);
+      failFast.close();
+   }
+
 }

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

@@ -17,15 +17,19 @@
 package com.kingsrook.qbits.quicksearch.listeners;
 
 
+import java.util.ArrayList;
 import java.util.List;
 import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
 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.actions.tables.listeners.RecordChangeEvent;
 import com.kingsrook.qqq.backend.core.actions.tables.listeners.RecordChangeType;
 import com.kingsrook.qqq.backend.core.context.QContext;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QCollectingLogger;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
 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.QueryInput;
@@ -35,6 +39,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
 import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
 import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData;
 import com.kingsrook.qbits.quicksearch.BaseQuickSearchTest;
+import com.kingsrook.qbits.quicksearch.MissingLibraryTransportCustomizer;
 import com.kingsrook.qbits.quicksearch.QuickSearchQBitConfig;
 import com.kingsrook.qbits.quicksearch.QuickSearchQBitContext;
 import com.kingsrook.qbits.quicksearch.QuickSearchRuntime;
@@ -45,17 +50,23 @@ 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;
+import org.apache.logging.log4j.Level;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
+import org.mockito.MockedConstruction;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.mockito.ArgumentMatchers.any;
 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.mockConstruction;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 
 /*******************************************************************************
@@ -298,6 +309,178 @@ class QuickSearchRecordChangeListenerTest extends BaseQuickSearchTest
 
 
 
+   @Test
+   void testUpdate_reReadFails_recordsFailedEventsAfterCommit() throws QException
+   {
+      QBackendTransaction transaction = new QBackendTransaction();
+      try(MockedConstruction<QueryAction> ignored = failingQueries(new QException("possible-value backend unavailable")))
+      {
+         assertThatCode(() -> new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(transaction, 7, 8))).doesNotThrowAnyException();
+      }
+
+      assertThat(queryFailedEvents()).isEmpty();
+      transaction.commit();
+
+      List<QRecord> failed = queryFailedEvents();
+      assertThat(failed).extracting(r -> r.getValueString("recordId")).containsExactlyInAnyOrder("7", "8");
+      assertThat(failed).allSatisfy(r ->
+      {
+         assertThat(r.getValueString("tableName")).isEqualTo(TEST_ENTITY_TABLE);
+         assertThat(r.getValueString("action")).isEqualTo("INDEX");
+         assertThat(r.getValueString("status")).isEqualTo(QuickSearchFailedEvent.STATUS_PENDING);
+         assertThat(r.getValueString("errorMessage")).contains("possible-value backend unavailable");
+      });
+      verify(publisher, never()).publishIndexEvents(anyList());
+   }
+
+
+
+   @Test
+   void testUpdate_reReadFailsThenRollback_logsCauseAndRecordsNothing() throws QException
+   {
+      QCollectingLogger   collectingLogger = QLogger.activateCollectingLoggerForClass(QuickSearchRecordChangeListener.class);
+      QBackendTransaction transaction      = new QBackendTransaction();
+      try(MockedConstruction<QueryAction> ignored = failingQueries(new QException("possible-value backend unavailable")))
+      {
+         new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(transaction, 7, 8));
+      }
+      finally
+      {
+         QLogger.deactivateCollectingLoggerForClass(QuickSearchRecordChangeListener.class);
+      }
+      transaction.rollback();
+
+      assertThat(queryFailedEvents()).isEmpty();
+      assertThat(collectingLogger.getCollectedMessages()).anySatisfy(m ->
+      {
+         assertThat(m.getLevel()).isEqualTo(Level.WARN);
+         assertThat(m.getMessage()).contains("Could not re-read updated records").contains(TEST_ENTITY_TABLE).contains("\"recordCount\":2").doesNotContain("changed");
+         assertThat(m.getMessage()).contains("possible-value backend unavailable");
+      });
+   }
+
+
+
+   @Test
+   void testUpdate_reReadFailsLinkage_recordsFailedEvents() throws QException
+   {
+      try(MockedConstruction<QueryAction> ignored = failingQueries(new NoClassDefFoundError("com/example/pvs/Provider")))
+      {
+         assertThatCode(() -> new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(null, 7))).doesNotThrowAnyException();
+      }
+
+      List<QRecord> failed = queryFailedEvents();
+      assertThat(failed).extracting(r -> r.getValueString("recordId")).containsExactly("7");
+      assertThat(failed.get(0).getValueString("errorMessage")).contains("com/example/pvs/Provider");
+   }
+
+
+
+   @Test
+   void testUpdate_reReadFailsWithoutTransaction_recordsImmediately() throws QException
+   {
+      try(MockedConstruction<QueryAction> ignored = failingQueries(new QException("possible-value backend unavailable")))
+      {
+         new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(null, 7));
+      }
+
+      assertThat(queryFailedEvents()).extracting(r -> r.getValueString("recordId")).containsExactly("7");
+   }
+
+
+
+   @Test
+   void testUpdate_reReadAndRecordingBothFail_logsAndDoesNotThrow()
+   {
+      QCollectingLogger collectingLogger = QLogger.activateCollectingLoggerForClass(QuickSearchRecordChangeListener.class);
+      try(MockedConstruction<QueryAction> ignoredQueries = failingQueries(new QException("possible-value backend unavailable"));
+         MockedConstruction<InsertAction> ignoredInserts = mockConstruction(InsertAction.class, (mock, context) ->
+            when(mock.execute(any(InsertInput.class))).thenThrow(new QException("failed-event table unavailable"))))
+      {
+         assertThatCode(() -> new QuickSearchRecordChangeListener().onRecordsChanged(updateEvent(null, 7))).doesNotThrowAnyException();
+      }
+      finally
+      {
+         QLogger.deactivateCollectingLoggerForClass(QuickSearchRecordChangeListener.class);
+      }
+
+      assertThat(collectingLogger.getCollectedMessages())
+         .anySatisfy(m ->
+         {
+            assertThat(m.getLevel()).isEqualTo(Level.WARN);
+            assertThat(m.getMessage()).contains("Could not re-read updated records").contains(TEST_ENTITY_TABLE).doesNotContain("changed");
+         })
+         .anySatisfy(m ->
+         {
+            assertThat(m.getLevel()).isEqualTo(Level.ERROR);
+            assertThat(m.getMessage()).contains("Could not record failed index events");
+         });
+   }
+
+
+
+   @Test
+   void testClientBuildFailsLinkage_recordsFailedEvents() throws QException
+   {
+      //////////////////////////////////////////////////////////////////
+      // no host publisher: the first write builds the client lazily, //
+      // and a missing optional library fails linkage there           //
+      //////////////////////////////////////////////////////////////////
+      runtime.getConfig().setIndexEventPublisher(null);
+      runtime.getConfig().setTransportCustomizer(new QCodeReference(MissingLibraryTransportCustomizer.class));
+
+      Integer id = insert("widget", null);
+
+      List<QRecord> failed = queryFailedEvents();
+      assertThat(failed).hasSize(1);
+      assertThat(failed.get(0).getValueString("recordId")).isEqualTo(String.valueOf(id));
+      assertThat(failed.get(0).getValueString("action")).isEqualTo("INDEX");
+      assertThat(failed.get(0).getValueString("errorMessage")).contains(MissingLibraryTransportCustomizer.MISSING_CLASS);
+   }
+
+
+
+   @Test
+   void testPublishFailsLinkage_recordsFailedEvents() throws QException
+   {
+      doThrow(new NoClassDefFoundError("com/example/broker/Client")).when(publisher).publishDeleteEvents(anyList());
+      Integer id = insert("widget", null);
+      delete(id);
+
+      List<QRecord> failed = queryFailedEvents();
+      assertThat(failed).hasSize(1);
+      assertThat(failed.get(0).getValueString("recordId")).isEqualTo(String.valueOf(id));
+      assertThat(failed.get(0).getValueString("action")).isEqualTo("DELETE");
+      assertThat(failed.get(0).getValueString("errorMessage")).contains("com/example/broker/Client");
+   }
+
+
+
+   private MockedConstruction<QueryAction> failingQueries(Throwable failure)
+   {
+      /////////////////////////////////////////////////////////
+      // read the enabled flag first, so only the listener's //
+      // re-read meets the failing QueryAction               //
+      /////////////////////////////////////////////////////////
+      runtime.isTableEnabled(TEST_ENTITY_TABLE);
+      return (mockConstruction(QueryAction.class, (mock, context) ->
+         when(mock.execute(any(QueryInput.class))).thenThrow(failure)));
+   }
+
+
+
+   private RecordChangeEvent updateEvent(QBackendTransaction transaction, Integer... ids)
+   {
+      List<QRecord> records = new ArrayList<>();
+      for(Integer id : ids)
+      {
+         records.add(new QRecord().withValue("id", id).withValue("description", "changed"));
+      }
+      return (new RecordChangeEvent().withTableName(TEST_ENTITY_TABLE).withType(RecordChangeType.UPDATE).withRecords(records).withTransaction(transaction));
+   }
+
+
+
    private void delete(Integer id) throws QException
    {
       DeleteInput deleteInput = new DeleteInput();