Do executor recovery in chunks to fix #718

CHANGELOG.md

@@ -3,9 +3,14 @@
 **Highlights**
 
 - Use Spring Boot 4 in `nflow-examples`
+- Recover workflow instances from dead executors in configurable chunks to avoid too large SQL `IN` clauses
 
 **Details**
 
+- `nflow-engine`
+  - Recover workflow instances from dead executors in chunks controlled by `nflow.db.max_sql_in_parameters`.
+    This avoids oversized SQL `IN` clauses, especially on SQL Server.
+
 ## 11.0.0 (2026-05-28)
 
 **Highlights**

nflow-engine/src/main/java/io/nflow/engine/internal/dao/WorkflowInstanceDao.java

@@ -111,6 +111,7 @@ public class WorkflowInstanceDao {
   private final long workflowInstanceQueryMaxActions;
   private final long workflowInstanceQueryMaxActionsDefault;
   private final int workflowInstanceTypeCacheSize;
+  private final int maxSqlInParameters;
   private final AtomicBoolean disableBatchUpdates = new AtomicBoolean();
   AtomicInteger instanceStateTextLength = new AtomicInteger();
   AtomicInteger actionStateTextLength = new AtomicInteger();
@@ -145,6 +146,7 @@ public WorkflowInstanceDao(SQLVariants sqlVariants, @NFlow JdbcTemplate nflowJdb
       logger.info("nFlow DB batch updates are disabled (system property nflow.db.disable_batch_updates=true)");
     }
     workflowInstanceTypeCacheSize = env.getRequiredProperty("nflow.db.workflowInstanceType.cacheSize", Integer.class);
+    maxSqlInParameters = env.getProperty("nflow.db.max_sql_in_parameters", Integer.class, 1000);
     instanceStateTextLength.set(env.getProperty("nflow.workflow.instance.state.text.length", Integer.class, -1));
     actionStateTextLength.set(env.getProperty("nflow.workflow.action.state.text.length", Integer.class, -1));
     stateVariableValueMaxLength.set(env.getProperty("nflow.workflow.state.variable.value.length", Integer.class, -1));
@@ -399,11 +401,16 @@ public void recoverWorkflowInstancesFromDeadNodes() {
     }
     WorkflowInstanceAction.Builder builder = new WorkflowInstanceAction.Builder().setExecutionStart(now()).setExecutionEnd(now())
         .setType(recovery).setStateText("Recovered");
-    for (InstanceInfo instance : getRecoverableWorkflowInstances(recoverableExecutorIds)) {
-      WorkflowInstanceAction action = builder.setState(instance.state()).setWorkflowInstanceId(instance.id()).build();
-      recoverWorkflowInstance(instance.id(), instance.executorId(), action);
+    List<Integer> recoverableExecutorIdList = new ArrayList<>(recoverableExecutorIds);
+    for (int from = 0; from < recoverableExecutorIdList.size(); from += maxSqlInParameters) {
+      List<Integer> recoverableExecutorIdChunk = recoverableExecutorIdList.subList(from,
+        min(from + maxSqlInParameters, recoverableExecutorIdList.size()));
+      for (InstanceInfo instance : getRecoverableWorkflowInstances(recoverableExecutorIdChunk)) {
+        WorkflowInstanceAction action = builder.setState(instance.state()).setWorkflowInstanceId(instance.id()).build();
+        recoverWorkflowInstance(instance.id(), instance.executorId(), action);
+      }
+      recoverableExecutorIdChunk.forEach(executorInfo::markRecovered);
     }
-    recoverableExecutorIds.forEach(executorInfo::markRecovered);
   }
 
   private List<InstanceInfo> getRecoverableWorkflowInstances(Collection<Integer> executorsIds) {

nflow-engine/src/main/resources/nflow-engine.properties

@@ -65,6 +65,7 @@ nflow.db.max_pool_size=4
 nflow.db.idle_timeout_seconds=600
 nflow.db.create_on_startup=true
 nflow.db.disable_batch_updates=false
+nflow.db.max_sql_in_parameters=1000
 nflow.db.workflowInstanceType.cacheSize=10000
 nflow.db.initialization_fail_timeout_seconds=10
 

nflow-engine/src/test/java/io/nflow/engine/internal/dao/WorkflowInstanceDaoTest.java

@@ -33,9 +33,13 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.sql.ResultSet;
@@ -53,17 +57,21 @@
 import org.joda.time.DateTime;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
+import org.mockito.ArgumentMatchers;
 import org.springframework.core.env.Environment;
 import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.core.RowCallbackHandler;
+import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.mock.env.MockEnvironment;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.support.TransactionCallback;
 import org.springframework.transaction.support.TransactionTemplate;
 
 import io.nflow.engine.config.db.PgDatabaseConfiguration.PostgreSQLVariants;
 import io.nflow.engine.internal.dao.WorkflowInstanceDao.WorkflowInstanceActionRowMapper;
+import io.nflow.engine.internal.executor.InstanceInfo;
 import io.nflow.engine.internal.executor.WorkflowInstanceExecutor;
 import io.nflow.engine.internal.storage.db.SQLVariants;
 import io.nflow.engine.service.WorkflowInstanceInclude;
@@ -748,7 +756,10 @@ public void fakePostgreSQLpollNextWorkflowInstances() {
   }
 
   private WorkflowInstanceDao preparePostgreSQLDao(JdbcTemplate jdbcTemplate) {
-    ExecutorDao eDao = mock(ExecutorDao.class);
+    return preparePostgreSQLDao(jdbcTemplate, mock(ExecutorDao.class), env);
+  }
+
+  private WorkflowInstanceDao preparePostgreSQLDao(JdbcTemplate jdbcTemplate, ExecutorDao eDao, Environment env) {
     lenient().when(eDao.getExecutorGroupCondition()).thenReturn("group matches");
     lenient().when(eDao.getExecutorId()).thenReturn(42);
     NamedParameterJdbcTemplate namedJdbc = mock(NamedParameterJdbcTemplate.class);
@@ -762,6 +773,16 @@ private WorkflowInstanceDao preparePostgreSQLDao(JdbcTemplate jdbcTemplate) {
     return d;
   }
 
+  private MockEnvironment prepareEnvironment() {
+    return new MockEnvironment()
+        .withProperty("nflow.workflow.instance.query.max.results", "1000")
+        .withProperty("nflow.workflow.instance.query.max.results.default", "1000")
+        .withProperty("nflow.workflow.instance.query.max.actions", "1000")
+        .withProperty("nflow.workflow.instance.query.max.actions.default", "1000")
+        .withProperty("nflow.db.disable_batch_updates", "false")
+        .withProperty("nflow.db.workflowInstanceType.cacheSize", "10000");
+  }
+
   @Test
   public void pollNextWorkflowInstancesWithPartialRaceCondition() throws InterruptedException {
     int batchSize = 100;
@@ -982,6 +1003,33 @@ public void recoverWorkflowInstancesFromDeadNodesSetsExecutorIdToNullAndStatusTo
     assertThat(workflowInstanceAction.stateText, is("Recovered"));
   }
 
+  @Test
+  public void recoverWorkflowInstancesFromDeadNodesSplitsExecutorIdsByConfiguredMaxSqlInParameters() {
+    JdbcTemplate j = mock(JdbcTemplate.class);
+    ExecutorDao eDao = mock(ExecutorDao.class);
+    when(eDao.getRecoverableExecutorIds()).thenReturn(asList(1, 2, 3, 4, 5));
+    when(j.query(anyString(), ArgumentMatchers.<RowMapper<InstanceInfo>> any(), any(Object[].class))).thenReturn(emptyList());
+    WorkflowInstanceDao d = preparePostgreSQLDao(j, eDao, prepareEnvironment().withProperty("nflow.db.max_sql_in_parameters", "2"));
+
+    d.recoverWorkflowInstancesFromDeadNodes();
+
+    ArgumentCaptor<String> sql = ArgumentCaptor.forClass(String.class);
+    ArgumentCaptor<Object[]> args = ArgumentCaptor.forClass(Object[].class);
+    verify(j, times(3)).query(sql.capture(), ArgumentMatchers.<RowMapper<InstanceInfo>> any(), args.capture());
+    assertThat(sql.getAllValues(), contains(
+        "select id, executor_id, state from nflow_workflow where executor_id in (?,?)",
+        "select id, executor_id, state from nflow_workflow where executor_id in (?,?)",
+        "select id, executor_id, state from nflow_workflow where executor_id in (?)"));
+    assertThat(asList(args.getAllValues().get(0)), contains((Object) 1, (Object) 2));
+    assertThat(asList(args.getAllValues().get(1)), contains((Object) 3, (Object) 4));
+    assertThat(asList(args.getAllValues().get(2)), contains((Object) 5));
+    verify(eDao).markRecovered(1);
+    verify(eDao).markRecovered(2);
+    verify(eDao).markRecovered(3);
+    verify(eDao).markRecovered(4);
+    verify(eDao).markRecovered(5);
+  }
+
   @Test
   public void settingSignalInsertsAction() {
     WorkflowInstance i = constructWorkflowInstanceBuilder().setBusinessKey("setSignalTest").build();