Fix inconsistent ES8 workflow state refresh handling

es8-persistence/README.md

@@ -58,6 +58,9 @@ conductor.elasticsearch.indexPrefix=conductor
 # Default refresh interval applied via the component template.
 conductor.elasticsearch.indexRefreshInterval=30s
 
+# Force an immediate refresh after every ES8 write/delete operation.
+conductor.elasticsearch.refreshOnWrite=false
+
 # Path to a PEM-encoded certificate to trust for HTTPS connections.
 conductor.elasticsearch.trustCertPath=
 

es8-persistence/src/main/java/org/conductoross/conductor/es8/config/ElasticSearchProperties.java

@@ -94,13 +94,19 @@ public class ElasticSearchProperties {
     private String password;
 
     /**
-     * Whether to wait for index refresh when updating tasks and workflows. When enabled, the
-     * operation will block until the changes are visible for search. This guarantees immediate
-     * search visibility but can significantly impact performance (20-30s delays). Defaults to false
-     * for better performance.
+     * Whether to wait for the next index refresh cycle before returning from supported write
+     * operations. This preserves Elasticsearch's refresh interval behavior while still waiting for
+     * search visibility.
      */
     private boolean waitForIndexRefresh = false;
 
+    /**
+     * Whether to force an immediate refresh for every ES8 write/delete operation. When enabled, the
+     * operation returns only after Elasticsearch has refreshed the affected shard, making {@link
+     * #indexRefreshInterval} irrelevant for those operations.
+     */
+    private boolean refreshOnWrite = false;
+
     public String getUrl() {
         return url;
     }
@@ -245,6 +251,14 @@ public void setWaitForIndexRefresh(boolean waitForIndexRefresh) {
         this.waitForIndexRefresh = waitForIndexRefresh;
     }
 
+    public boolean isRefreshOnWrite() {
+        return refreshOnWrite;
+    }
+
+    public void setRefreshOnWrite(boolean refreshOnWrite) {
+        this.refreshOnWrite = refreshOnWrite;
+    }
+
     public List<URL> toURLs() {
         String clusterAddress = getUrl() == null ? "" : getUrl();
         List<URL> urls =

es8-persistence/src/main/java/org/conductoross/conductor/es8/dao/index/ElasticSearchRestDAOV8.java

@@ -318,19 +318,14 @@ public void indexWorkflow(WorkflowSummary workflow) {
         try {
             long startTime = Instant.now().toEpochMilli();
             String workflowId = workflow.getWorkflowId();
-            Refresh refresh = properties.isWaitForIndexRefresh() ? Refresh.WaitFor : null;
-            executeWithRetry(
-                    () ->
-                            elasticSearchClient.index(
-                                    i -> {
-                                        i.index(workflowIndexName)
-                                                .id(workflowId)
-                                                .document(workflow);
-                                        if (refresh != null) {
-                                            i.refresh(refresh);
-                                        }
-                                        return i;
-                                    }));
+            Refresh refresh = writeRefreshPolicy();
+            indexDocumentWithIlmFallback(
+                    workflowIndexName,
+                    WORKFLOW_DOC_TYPE,
+                    workflowId,
+                    workflow,
+                    refresh,
+                    workflow.getUpdateTime());
             long endTime = Instant.now().toEpochMilli();
             logger.debug(
                     "Time taken {} for indexing workflow: {}", endTime - startTime, workflowId);
@@ -340,6 +335,8 @@ public void indexWorkflow(WorkflowSummary workflow) {
         } catch (Exception e) {
             Monitors.error(className, "indexWorkflow");
             logger.error("Failed to index workflow: {}", workflow.getWorkflowId(), e);
+            throw new TransientException(
+                    String.format("Failed to index workflow %s", workflow.getWorkflowId()), e);
         }
     }
 
@@ -353,9 +350,14 @@ public void indexTask(TaskSummary task) {
         try {
             long startTime = Instant.now().toEpochMilli();
             String taskId = task.getTaskId();
-
-            Refresh refreshPolicy = properties.isWaitForIndexRefresh() ? Refresh.WaitFor : null;
-            indexObject(taskIndexName, TASK_DOC_TYPE, taskId, task, refreshPolicy);
+            Refresh refreshPolicy = writeRefreshPolicy();
+            indexDocumentWithIlmFallback(
+                    taskIndexName,
+                    TASK_DOC_TYPE,
+                    taskId,
+                    task,
+                    refreshPolicy,
+                    task.getUpdateTime());
             long endTime = Instant.now().toEpochMilli();
             logger.debug(
                     "Time taken {} for  indexing task:{} in workflow: {}",
@@ -367,6 +369,8 @@ public void indexTask(TaskSummary task) {
                     "indexQueue", ((ThreadPoolExecutor) executorService).getQueue().size());
         } catch (Exception e) {
             logger.error("Failed to index task: {}", task.getTaskId(), e);
+            throw new TransientException(
+                    String.format("Failed to index task %s", task.getTaskId()), e);
         }
     }
 
@@ -389,7 +393,17 @@ public void addTaskExecutionLogs(List<TaskExecLog> taskExecLogs) {
         }
 
         try {
-            executeWithRetry(() -> elasticSearchClient.bulk(b -> b.operations(operations)));
+            Refresh refreshPolicy = writeRefreshPolicy();
+            executeWithRetry(
+                    () ->
+                            elasticSearchClient.bulk(
+                                    b -> {
+                                        b.operations(operations);
+                                        if (refreshPolicy != null) {
+                                            b.refresh(refreshPolicy);
+                                        }
+                                        return b;
+                                    }));
             long endTime = Instant.now().toEpochMilli();
             logger.debug("Time taken {} for indexing taskExecutionLogs", endTime - startTime);
             Monitors.recordESIndexTime(
@@ -582,7 +596,7 @@ public void addEventExecution(EventExecution eventExecution) {
                             + "."
                             + eventExecution.getId();
 
-            indexObject(eventIndexName, EVENT_DOC_TYPE, id, eventExecution, null);
+            indexObject(eventIndexName, EVENT_DOC_TYPE, id, eventExecution, writeRefreshPolicy());
             long endTime = Instant.now().toEpochMilli();
             logger.debug(
                     "Time taken {} for indexing event execution: {}",
@@ -901,6 +915,7 @@ public void updateTask(String workflowId, String taskId, String[] keys, Object[]
 
     private enum UpdateOutcome {
         UPDATED,
+        SKIPPED_STALE,
         NOT_FOUND
     }
 
@@ -917,6 +932,14 @@ private enum DeleteOutcome {
      */
     private UpdateOutcome updateByIdWithIlmFallback(
             String indexAlias, String id, Map<String, Object> source) throws IOException {
+        if (shouldSkipStaleWrite(indexAlias, id, source.get("updateTime"))) {
+            logger.warn(
+                    "Skipping stale partial update for document {} in alias {} because incoming updateTime is older than the indexed document",
+                    id,
+                    indexAlias);
+            return UpdateOutcome.SKIPPED_STALE;
+        }
+
         UpdateOutcome aliasOutcome = updateById(indexAlias, id, source);
         if (aliasOutcome != UpdateOutcome.NOT_FOUND) {
             return aliasOutcome;
@@ -933,11 +956,18 @@ private UpdateOutcome updateByIdWithIlmFallback(
     private UpdateOutcome updateById(String indexOrAlias, String id, Map<String, Object> source)
             throws IOException {
         try {
+            Refresh refreshPolicy = writeRefreshPolicy();
             UpdateResponse<Map> response =
                     executeWithRetry(
                             () ->
                                     elasticSearchClient.update(
-                                            u -> u.index(indexOrAlias).id(id).doc(source),
+                                            u -> {
+                                                u.index(indexOrAlias).id(id).doc(source);
+                                                if (refreshPolicy != null) {
+                                                    u.refresh(refreshPolicy);
+                                                }
+                                                return u;
+                                            },
                                             Map.class));
             return response.result() == Result.NotFound
                     ? UpdateOutcome.NOT_FOUND
@@ -1001,9 +1031,18 @@ private DeleteOutcome deleteByIdWithIlmFallback(String indexAlias, String id)
 
     private DeleteOutcome deleteById(String indexName, String id) throws IOException {
         try {
+            Refresh refreshPolicy = writeRefreshPolicy();
             DeleteResponse response =
                     executeWithRetry(
-                            () -> elasticSearchClient.delete(d -> d.index(indexName).id(id)));
+                            () ->
+                                    elasticSearchClient.delete(
+                                            d -> {
+                                                d.index(indexName).id(id);
+                                                if (refreshPolicy != null) {
+                                                    d.refresh(refreshPolicy);
+                                                }
+                                                return d;
+                                            }));
             return response.result() == Result.Deleted
                     ? DeleteOutcome.DELETED
                     : DeleteOutcome.NOT_FOUND;
@@ -1035,6 +1074,74 @@ private Map getDocumentSourceByIdWithIlmFallback(String indexAlias, String id)
         return response.found() ? response.source() : null;
     }
 
+    private <T> void indexDocumentWithIlmFallback(
+            String indexAlias,
+            String docType,
+            String id,
+            T doc,
+            Refresh refreshPolicy,
+            Object incomingUpdateTime)
+            throws IOException {
+        if (shouldSkipStaleWrite(indexAlias, id, incomingUpdateTime)) {
+            logger.warn(
+                    "Skipping stale {} index write for document {} in alias {} because incoming updateTime is older than the indexed document",
+                    docType,
+                    id,
+                    indexAlias);
+            return;
+        }
+
+        Optional<String> resolvedIndex = resolveSingleIndexForId(indexAlias, id);
+        String targetIndex = resolvedIndex.orElseGet(() -> indexAlias);
+        if (resolvedIndex.isEmpty()) {
+            targetIndex = resolveWriteIndexForAlias(indexAlias).orElse(indexAlias);
+        }
+        indexObject(targetIndex, docType, id, doc, refreshPolicy);
+    }
+
+    private boolean shouldSkipStaleWrite(String indexAlias, String id, Object incomingUpdateTime)
+            throws IOException {
+        Optional<Long> incomingUpdateTimeMs = toEpochMillis(incomingUpdateTime);
+        if (incomingUpdateTimeMs.isEmpty()) {
+            return false;
+        }
+
+        Map existingSource = getDocumentSourceByIdWithIlmFallback(indexAlias, id);
+        if (existingSource == null) {
+            return false;
+        }
+
+        Optional<Long> indexedUpdateTimeMs = toEpochMillis(existingSource.get("updateTime"));
+        return indexedUpdateTimeMs.isPresent()
+                && incomingUpdateTimeMs.get() < indexedUpdateTimeMs.get();
+    }
+
+    private Optional<Long> toEpochMillis(Object value) {
+        if (value == null) {
+            return Optional.empty();
+        }
+        if (value instanceof Number number) {
+            return Optional.of(number.longValue());
+        }
+
+        String text = value.toString();
+        if (StringUtils.isBlank(text)) {
+            return Optional.empty();
+        }
+
+        try {
+            return Optional.of(Instant.parse(text).toEpochMilli());
+        } catch (Exception ignored) {
+            // fall through
+        }
+
+        try {
+            return Optional.of(Long.parseLong(text));
+        } catch (NumberFormatException ignored) {
+            return Optional.empty();
+        }
+    }
+
     private Optional<String> resolveWriteIndexForAlias(String alias) throws IOException {
         Request request = new Request(HttpMethod.GET, "/_alias/" + alias);
         try {
@@ -1210,4 +1317,11 @@ private void indexObject(
             final Refresh refreshPolicy) {
         bulkIngestionSupport.indexObject(index, docType, docId, doc, refreshPolicy);
     }
+
+    private Refresh writeRefreshPolicy() {
+        if (properties.isRefreshOnWrite()) {
+            return Refresh.True;
+        }
+        return properties.isWaitForIndexRefresh() ? Refresh.WaitFor : null;
+    }
 }

es8-persistence/src/main/java/org/conductoross/conductor/es8/dao/index/Es8BulkIngestionSupport.java

@@ -91,11 +91,18 @@ void close() {
     }
 
     void indexObject(String index, String docType, Object doc) {
-        indexObject(index, docType, null, doc, null);
+        indexObject(index, docType, null, doc, defaultRefreshPolicy());
     }
 
     void indexObject(
             String index, String docType, String docId, Object doc, Refresh refreshPolicy) {
+        Refresh effectiveRefreshPolicy =
+                refreshPolicy != null ? refreshPolicy : defaultRefreshPolicy();
+        if (effectiveRefreshPolicy != null) {
+            indexObjectSynchronously(index, docType, docId, doc, effectiveRefreshPolicy);
+            return;
+        }
+
         BulkOperation operation =
                 BulkOperation.of(
                         op ->
@@ -107,10 +114,46 @@ void indexObject(
                                             }
                                             return i;
                                         }));
-        BulkIngester<Void> ingester = getBulkIngester(docType, refreshPolicy);
+        BulkIngester<Void> ingester = getBulkIngester(docType, effectiveRefreshPolicy);
         ingester.add(operation);
     }
 
+    private Refresh defaultRefreshPolicy() {
+        if (properties.isRefreshOnWrite()) {
+            return Refresh.True;
+        }
+        return properties.isWaitForIndexRefresh() ? Refresh.WaitFor : null;
+    }
+
+    private void indexObjectSynchronously(
+            String index, String docType, String docId, Object doc, Refresh refreshPolicy) {
+        long startTime = System.currentTimeMillis();
+        try {
+            retryTemplate.execute(
+                    context -> {
+                        elasticSearchClient.index(
+                                i -> {
+                                    i.index(index).document(doc).refresh(refreshPolicy);
+                                    if (docId != null) {
+                                        i.id(docId);
+                                    }
+                                    return i;
+                                });
+                        return null;
+                    });
+            Monitors.recordESIndexTime(
+                    "index_object", docType, System.currentTimeMillis() - startTime);
+            Monitors.recordWorkerQueueSize(
+                    "indexQueue", ((ThreadPoolExecutor) executorService).getQueue().size());
+            Monitors.recordWorkerQueueSize(
+                    "logQueue", ((ThreadPoolExecutor) logExecutorService).getQueue().size());
+        } catch (Exception e) {
+            Monitors.error(className, "index");
+            throw new RuntimeException(
+                    String.format("Synchronous index failed for doc type %s", docType), e);
+        }
+    }
+
     private BulkIngester<Void> getBulkIngester(String docType, Refresh refreshPolicy) {
         Pair<String, Refresh> requestKey = new ImmutablePair<>(docType, refreshPolicy);
         return bulkIngesters.computeIfAbsent(

es8-persistence/src/test/java/org/conductoross/conductor/es8/config/ElasticSearchPropertiesTest.java

@@ -40,6 +40,19 @@ public void testWaitForIndexRefreshCanBeEnabled() {
                 properties.isWaitForIndexRefresh());
     }
 
+    @Test
+    public void testRefreshOnWriteDefaultsToFalse() {
+        ElasticSearchProperties properties = new ElasticSearchProperties();
+        assertFalse("refreshOnWrite should default to false", properties.isRefreshOnWrite());
+    }
+
+    @Test
+    public void testRefreshOnWriteCanBeEnabled() {
+        ElasticSearchProperties properties = new ElasticSearchProperties();
+        properties.setRefreshOnWrite(true);
+        assertTrue("refreshOnWrite should be configurable to true", properties.isRefreshOnWrite());
+    }
+
     @Test
     public void testDefaultUrlUsesHttpPort9200() {
         ElasticSearchProperties properties = new ElasticSearchProperties();