manerow · Oct 10, 2026 · Oct 10, 2026
@@ -103,6 +103,10 @@ private ExecutionResult doExecute(
 
     int partitionSize = jobData.getPartitionSize() != null ? jobData.getPartitionSize() : 10000;
     distributedExecutor = new DistributedSearchIndexExecutor(collectionDAO, partitionSize);
+    // A stop that landed before this executor existed had nothing to stop yet; hand it on.
+    if (stopped.get()) {
+      distributedExecutor.stop();
+    }
     distributedExecutor.performStartupRecovery();
 
     distributedExecutor.addListener(listeners);

@@ -230,6 +230,10 @@ private void runReindexing() {
 
     DistributedIndexingStrategy strategy = createDistributedStrategy();
     activeStrategy = strategy;
+    // A stop that landed before this strategy existed had nothing to stop yet; hand it on.
+    if (stopped) {
+      strategy.stop();
+    }
     registerProgressListeners(strategy);
 
     ReindexingConfiguration config = buildReindexingConfiguration();

@@ -119,7 +119,8 @@ public static boolean isCoordinatingJob(UUID jobId) {
   private final String serverId;
   private final CompositeProgressListener listeners = new CompositeProgressListener();
 
-  @Getter private SearchIndexJob currentJob;
+  // Written by the reindex thread, read by stop() on the request thread that asked for the stop.
+  @Getter private volatile SearchIndexJob currentJob;
   private DistributedJobStatsAggregator statsAggregator;
   private ExecutorService workerExecutor;
   private final Set<UUID> activePartitions = ConcurrentHashMap.newKeySet();
@@ -344,6 +345,9 @@ public ExecutionResult execute(
     }
 
     UUID jobId = currentJob.getId();
+    if (stopped.get()) {
+      return endJobStoppedBeforeStart(jobId);
+    }
     LOG.info("Server {} starting execution of job {}", serverId, jobId);
     boolean startedJob = false;
 
@@ -626,9 +630,26 @@ public ExecutionResult execute(
       LOG.debug("Removed job {} from coordinated jobs set", jobId);
     }
 
-    // Get final job state
-    currentJob = coordinator.getJobWithAggregatedStats(jobId);
+    return finalResult(jobId);
+  }
 
+  /**
+   * {@link #stop()} can land while {@link #createJob} is still building partitions, when there is
+   * no job yet to request a stop for. Starting the job anyway would flip it to RUNNING with none of
+   * our workers on it: peer servers then adopt it as an orphan and reindex what was just stopped,
+   * and nothing ever finishes the app run. Stop it before it starts instead.
+   */
+  private ExecutionResult endJobStoppedBeforeStart(UUID jobId) {
+    LOG.info("Stop was requested before job {} started; not starting it", jobId);
+    if (!isJobTerminalOrStopping(jobId)) {
+      coordinator.requestStop(jobId);
+    }
+    coordinator.releaseReindexLock(jobId);
+    return finalResult(jobId);
+  }
+
+  private ExecutionResult finalResult(UUID jobId) {
+    currentJob = coordinator.getJobWithAggregatedStats(jobId);
     return new ExecutionResult(
         currentJob.getStatus(),
         currentJob.getTotalRecords(),

@@ -803,6 +803,39 @@ void executeReturnsFailedResultWhenDistributedExecutorStartupFails() {
     }
   }
 
+  /**
+   * Regression: a stop that lands before the executor exists finds nothing to stop. The executor
+   * created afterwards never heard of it, so it created and started the job the user had stopped.
+   */
+  @Test
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  void stopBeforeExecutorExistsIsHandedToTheExecutorCreatedAfterIt() {
+    EntityRepository entityRepository = mock(EntityRepository.class);
+    EntityDAO entityDao = mock(EntityDAO.class);
+
+    when(entityRepository.getDao()).thenReturn(entityDao);
+    when(entityRepository.getReindexFilter()).thenReturn(new ListFilter(Include.ALL));
+    when(entityDao.listCount(any(ListFilter.class))).thenReturn(5);
+
+    try (MockedStatic<Entity> entityMock = mockStatic(Entity.class);
+        MockedConstruction<DistributedSearchIndexExecutor> executorConstruction =
+            mockConstruction(
+                DistributedSearchIndexExecutor.class,
+                (mock, context) ->
+                    doThrow(new RuntimeException("end the run before job creation"))
+                        .when(mock)
+                        .performStartupRecovery())) {
+      entityMock.when(() -> Entity.getEntityRepository(Entity.TABLE)).thenReturn(entityRepository);
+
+      strategy.stop();
+      strategy.execute(
+          ReindexingConfiguration.builder().entities(Set.of(Entity.TABLE)).build(),
+          context(APP_ID));
+
+      verify(executorConstruction.constructed().getFirst()).stop();
+    }
+  }
+
   @Test
   @SuppressWarnings({"rawtypes", "unchecked"})
   void executeClosesSinkAndReturnsFailedWhenDoExecuteThrowsAndSinkCloseAlsoFails()

@@ -10,6 +10,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.mockConstruction;
@@ -308,6 +309,56 @@ void stopStopsActiveStrategyAndPushesStoppedStatus() throws Exception {
     assertNotNull(appRunRecord.getEndTime());
   }
 
+  /**
+   * Regression: a stop that lands during preflight finds no strategy to stop. The strategy created
+   * afterwards never heard of it and ran the whole reindex the user had stopped.
+   */
+  @Test
+  void stopDuringPreflightIsHandedToTheStrategyCreatedAfterIt() {
+    EventPublisherJob jobData = new EventPublisherJob().withEntities(Set.of(Entity.TABLE));
+    EntityRepository entityRepository = mock(EntityRepository.class);
+    EntityDAO entityDao = mock(EntityDAO.class);
+
+    when(context.getJobName()).thenReturn("scheduled");
+    when(context.createProgressListener(jobData))
+        .thenReturn(mock(ReindexingProgressListener.class));
+    when(entityRepository.getDao()).thenReturn(entityDao);
+    when(entityDao.listCount(any())).thenReturn(5);
+    doAnswer(
+            invocation -> {
+              orchestrator.stop();
+              return null;
+            })
+        .when(searchRepository)
+        .ensureHybridSearchPipeline();
+
+    try (MockedStatic<Entity> entityMock = mockStatic(Entity.class);
+        MockedStatic<ReindexingMetrics> metricsMock = mockStatic(ReindexingMetrics.class);
+        MockedStatic<WebSocketManager> websocketMock = mockStatic(WebSocketManager.class);
+        MockedConstruction<OrphanedIndexCleaner> ignoredCleaner = mockOrphanCleaner();
+        MockedConstruction<DistributedIndexingStrategy> strategyConstruction =
+            mockConstruction(
+                DistributedIndexingStrategy.class,
+                (strategy, context1) ->
+                    when(strategy.execute(any(), any()))
+                        .thenReturn(
+                            ExecutionResult.builder()
+                                .status(ExecutionResult.Status.STOPPED)
+                                .startTime(10L)
+                                .endTime(20L)
+                                .build()))) {
+      metricsMock.when(ReindexingMetrics::getInstance).thenReturn(null);
+      websocketMock.when(WebSocketManager::getInstance).thenReturn(null);
+      entityMock.when(() -> Entity.getEntityRepository(Entity.TABLE)).thenReturn(entityRepository);
+
+      orchestrator.run(jobData);
+
+      verify(strategyConstruction.constructed().getFirst()).stop();
+      assertEquals(EventPublisherJob.Status.STOPPED, orchestrator.getJobData().getStatus());
+      assertEquals(AppRunRecord.Status.STOPPED, appRunRecord.getStatus());
+    }
+  }
+
   @Test
   void setupEntitiesExpandsAllAndCountTotalEntitiesSkipsUnsupportedTypes() throws Exception {
     EventPublisherJob jobData = new EventPublisherJob().withEntities(Set.of("all", Entity.TABLE));

@@ -320,6 +320,38 @@ void stopStopsWorkersAndRequestsCoordinatorStopOnlyOnce() throws Exception {
     verify(coordinator, times(1)).requestStop(job.getId());
   }
 
+  /**
+   * Regression: a stop that lands while {@code createJob()} is still building partitions finds no
+   * job to stop. {@code execute()} then started the job anyway, leaving it RUNNING with none of our
+   * workers on it, so peer servers adopted it as an orphan, reindexed what the user had stopped,
+   * and the app run stayed "running" for good.
+   */
+  @Test
+  void executeEndsJobStoppedBeforeItStartedInsteadOfStartingIt() throws Exception {
+    UUID jobId = UUID.randomUUID();
+    SearchIndexJob readyJob =
+        SearchIndexJob.builder().id(jobId).status(IndexJobStatus.READY).build();
+    ReindexingProgressListener listener = mock(ReindexingProgressListener.class);
+    executor.addListener(listener);
+    executor.stop();
+    setField("currentJob", readyJob);
+    when(coordinator.getJob(jobId)).thenReturn(Optional.of(readyJob));
+    when(coordinator.getJobWithAggregatedStats(jobId))
+        .thenReturn(readyJob.withStatus(IndexJobStatus.STOPPED));
+
+    DistributedSearchIndexExecutor.ExecutionResult result =
+        executor.execute(
+            mock(BulkSink.class),
+            stagedContext("table"),
+            ReindexingConfiguration.builder().entities(Set.of("table")).build());
+
+    assertEquals(IndexJobStatus.STOPPED, result.status());
+    verify(coordinator, never()).startJob(jobId);
+    verify(coordinator).requestStop(jobId);
+    verify(coordinator).releaseReindexLock(jobId);
+    verify(listener, never()).onJobStarted(any());
+  }
+
   /**
    * Regression: clicking Stop in the UI used to "do nothing" because workers blocked inside the
    * bulk-sink semaphore, slow DB queries, or {@code waitForSinkOperations} (5-min deadline)