Do recovery in smaller batches

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 maxInParameters;
   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);
+    maxInParameters = env.getProperty("nflow.db.max_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));
@@ -407,12 +409,18 @@ public void recoverWorkflowInstancesFromDeadNodes() {
   }
 
   private List<InstanceInfo> getRecoverableWorkflowInstances(Collection<Integer> executorsIds) {
-    StringBuilder sql = new StringBuilder(128);
-    sql.append("select id, executor_id, state from nflow_workflow where executor_id in (");
-    executorsIds.forEach(id -> sql.append("?,"));
-    sql.setCharAt(sql.length() - 1, ')');
-    return jdbc.query(sql.toString(), (rs, rowNum) -> new InstanceInfo(rs.getLong(1), rs.getInt(2), rs.getString(3)),
-      (Object[]) executorsIds.toArray(new Integer[0]));
+    List<Integer> ids = new ArrayList<>(executorsIds);
+    List<InstanceInfo> result = new ArrayList<>();
+    for (int i = 0; i < ids.size(); i += maxInParameters) {
+      List<Integer> batch = ids.subList(i, min(i + maxInParameters, ids.size()));
+      StringBuilder sql = new StringBuilder(64);
+      sql.append("select id, executor_id, state from nflow_workflow where executor_id in (");
+      batch.forEach(id -> sql.append("?,"));
+      sql.setCharAt(sql.length() - 1, ')');
+      result.addAll(jdbc.query(sql.toString(), (rs, rowNum) -> new InstanceInfo(rs.getLong(1), rs.getInt(2), rs.getString(3)),
+          (Object[]) batch.toArray(new Integer[0])));
+    }
+    return result;
   }
 
   private void recoverWorkflowInstance(final long instanceId, int expectedExecutorId, final WorkflowInstanceAction action) {

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_in_parameters=1000
 nflow.db.workflowInstanceType.cacheSize=10000
 nflow.db.initialization_fail_timeout_seconds=10
 

nflow-tests/src/test/java/io/nflow/tests/dao/BulkRecoveryDaoTest.java

@@ -0,0 +1,99 @@
+package io.nflow.tests.dao;
+
+import static io.nflow.engine.workflow.instance.WorkflowInstance.WorkflowInstanceStatus.executing;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.joda.time.DateTime.now;
+
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+
+import jakarta.inject.Inject;
+
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestMethodOrder;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import io.nflow.engine.config.NFlow;
+import io.nflow.engine.internal.dao.ExecutorDao;
+import io.nflow.engine.internal.dao.WorkflowInstanceDao;
+import io.nflow.tests.AbstractNflowTest;
+import io.nflow.tests.extension.NflowServerConfig;
+import io.nflow.tests.extension.NflowServerExtension.BeforeServerStop;
+
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+public class BulkRecoveryDaoTest extends AbstractNflowTest {
+
+  public static NflowServerConfig server = new NflowServerConfig.Builder()
+      .springContextClass(ServerContext.class)
+      .build();
+
+  private static JdbcTemplate jdbc;
+  private static WorkflowInstanceDao workflowInstanceDao;
+  private static ExecutorDao executorDao;
+
+  public BulkRecoveryDaoTest() {
+    super(server);
+  }
+
+  @Configuration
+  static class ServerContext {
+    @Inject
+    public void init(@NFlow JdbcTemplate nflowJdbc, WorkflowInstanceDao dao, ExecutorDao executor) {
+      jdbc = nflowJdbc;
+      workflowInstanceDao = dao;
+      executorDao = executor;
+    }
+  }
+
+  @Test
+  @Order(1)
+  public void insertNineThousandDeadExecutorsAndOneWorkflow() {
+    int count = 9000;
+    Timestamp crash = new Timestamp(now().minusDays(1).getMillis());
+    Timestamp active = new Timestamp(now().minusDays(1).plusSeconds(1).getMillis());
+    Timestamp expires = new Timestamp(now().minusDays(1).plusHours(1).getMillis());
+    List<Object[]> args = new ArrayList<>(count);
+    for (int i = 0; i < count; i++) {
+      args.add(new Object[] { "localhost", 666 + i, executorDao.getExecutorGroup(), crash, active, expires });
+    }
+    jdbc.batchUpdate(
+        "insert into nflow_executor (host, pid, executor_group, started, active, expires) values (?, ?, ?, ?, ?, ?)",
+        args);
+
+    int firstCrashedExecutorId = jdbc.queryForObject(
+        "select min(id) from nflow_executor where executor_group = ? and expires < current_timestamp and recovered is null",
+        Integer.class, executorDao.getExecutorGroup());
+    jdbc.update(
+        "insert into nflow_workflow (status, type, external_id, state, executor_id, executor_group, priority) values (?, ?, ?, ?, ?, ?, 0)",
+        executing.name(), "bulkTest", "extId0", "processing", firstCrashedExecutorId, executorDao.getExecutorGroup());
+  }
+
+  @Test
+  @Order(2)
+  public void recoverWorkflowInstancesFromNineThousandDeadExecutors() {
+    workflowInstanceDao.recoverWorkflowInstancesFromDeadNodes();
+
+    int recoveredExecutors = jdbc.queryForObject(
+        "select count(*) from nflow_executor where executor_group = ? and recovered is not null",
+        Integer.class, executorDao.getExecutorGroup());
+    assertThat(recoveredExecutors, is(9000));
+
+    Integer workflowExecutorId = jdbc.queryForObject(
+        "select executor_id from nflow_workflow where executor_group = ? and type = ?",
+        Integer.class, executorDao.getExecutorGroup(), "bulkTest");
+    assertThat(workflowExecutorId, is((Integer) null));
+  }
+
+  @BeforeServerStop
+  public void cleanUp() {
+    String group = executorDao.getExecutorGroup();
+    jdbc.update("delete from nflow_workflow_action where workflow_id in (select id from nflow_workflow where executor_group = ? and type = ?)", group, "bulkTest");
+    jdbc.update("delete from nflow_workflow where executor_group = ? and type = ?", group, "bulkTest");
+    jdbc.update("delete from nflow_executor where executor_group = ? and expires < current_timestamp", group);
+  }
+}

pom.xml

@@ -109,7 +109,7 @@
     <java26ExtraArgs></java26ExtraArgs>
     <guava.version>33.6.0-jre</guava.version>
     <surefire.forkcount>1C</surefire.forkcount>
-    <asm.version>9.10</asm.version>
+    <asm.version>9.10.1</asm.version>
     <apache.cxf.version>4.2.1</apache.cxf.version>
     <build-helper.version>3.6.1</build-helper.version>
     <central-publishing-maven-plugin.version>0.10.0</central-publishing-maven-plugin.version>
@@ -124,10 +124,10 @@
     <hamcrest.version>3.0</hamcrest.version>
     <hibernate.validator.version>9.1.0.Final</hibernate.validator.version>
     <hikaricp.version>7.0.2</hikaricp.version>
-    <jackson.version>2.21</jackson.version>
-    <jackson-databind.version>3.1.3</jackson-databind.version>
-    <jackson-joda.version>3.1.3</jackson-joda.version>
-    <jackson.rsjson.version>3.1.3</jackson.rsjson.version>
+    <jackson.version>2.22</jackson.version>
+    <jackson-databind.version>3.1.4</jackson-databind.version>
+    <jackson-joda.version>3.1.4</jackson-joda.version>
+    <jackson.rsjson.version>3.1.4</jackson.rsjson.version>
     <javassist.version>3.31.0-GA</javassist.version>
     <jakarta.annotation-api.version>3.0.0</jakarta.annotation-api.version>
     <jakarta.servlet-api.version>6.1.0</jakarta.servlet-api.version>
@@ -136,10 +136,10 @@
     <jakarta.xml.bind-api.version>4.1.0-M1</jakarta.xml.bind-api.version>
     <jaxws-api.version>4.0.3</jaxws-api.version>
     <jersey.version>3.1.11</jersey.version>
-    <jetty.version>12.1.9</jetty.version>
+    <jetty.version>12.1.10</jetty.version>
     <jodatime.version>2.14.2</jodatime.version>
     <junit5.version>6.1.0</junit5.version>
-    <logback-classic.version>1.5.32</logback-classic.version>
+    <logback-classic.version>1.5.34</logback-classic.version>
     <mariadb.version>3.5.8</mariadb.version>
     <maven-assembly.version>3.8.0</maven-assembly.version>
     <maven-clean.version>3.5.0</maven-clean.version>
@@ -161,8 +161,8 @@
     <maven-shade.version>3.6.2</maven-shade.version>
     <maven-site.version>4.0.0-M16</maven-site.version>
     <maven-source.version>3.4.0</maven-source.version>
-    <maven-surefire.version>3.5.5</maven-surefire.version>
-    <metrics.version>4.2.38</metrics.version>
+    <maven-surefire.version>3.5.6</maven-surefire.version>
+    <metrics.version>4.2.39</metrics.version>
     <mockito.version>5.23.0</mockito.version>
     <mssql.version>13.4.0.jre11</mssql.version>
     <mysql.version>9.7.0</mysql.version>