Merge 990fbf1 into a57e76d

CHANGELOG.md

@@ -19,6 +19,7 @@
     - mssql-jdbc 7.2.1.jre8
     - hikaricp 3.3.1
     - maven-surefire 2.22.1
+- Fix workflow history cleanup to keep the actions that hold the latest values of state variables
 
 ## 5.3.3 (2019-02-04)
 

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

@@ -883,9 +883,20 @@ public int deleteWorkflowInstanceHistory(Integer workflowInstanceId, Integer his
     int deletedActions = 0;
     if (maxActionId != null) {
       params.addValue("maxActionId", maxActionId);
-      namedJdbc.update("delete from nflow_workflow_state where workflow_id = :workflowId and action_id <= :maxActionId", params);
       List<Integer> referredActionIds = namedJdbc.queryForList(
-          "select distinct parent_action_id from nflow_workflow where parent_workflow_id = :workflowId", params, Integer.class);
+          "select distinct(max(action_id)) from nflow_workflow_state where workflow_id = :workflowId group by state_key", params,
+          Integer.class);
+      if (referredActionIds.isEmpty()) {
+        namedJdbc.update("delete from nflow_workflow_state where workflow_id = :workflowId and action_id <= :maxActionId",
+            params);
+      } else {
+        params.addValue("referredActionIds", referredActionIds);
+        namedJdbc.update(
+            "delete from nflow_workflow_state where workflow_id = :workflowId and action_id <= :maxActionId and action_id not in (:referredActionIds)",
+            params);
+      }
+      referredActionIds.addAll(namedJdbc.queryForList(
+          "select distinct parent_action_id from nflow_workflow where parent_workflow_id = :workflowId", params, Integer.class));
       if (referredActionIds.isEmpty()) {
         deletedActions = namedJdbc
             .update("delete from nflow_workflow_action where workflow_id = :workflowId and id <= :maxActionId", params);

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

@@ -46,7 +46,6 @@
 
 import javax.inject.Inject;
 
-import io.nflow.engine.internal.storage.db.SQLVariants;
 import org.hamcrest.CoreMatchers;
 import org.joda.time.DateTime;
 import org.junit.Rule;
@@ -59,9 +58,10 @@
 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.WorkflowInstanceExecutor;
-import io.nflow.engine.config.db.PgDatabaseConfiguration.PostgreSQLVariants;
+import io.nflow.engine.internal.storage.db.SQLVariants;
 import io.nflow.engine.service.WorkflowInstanceInclude;
 import io.nflow.engine.workflow.instance.QueryWorkflowInstances;
 import io.nflow.engine.workflow.instance.WorkflowInstance;
@@ -907,20 +907,24 @@ public void clearingSignalInsertsAction() {
   public void deleteExpiredWorkflowHistory() {
     WorkflowInstance parentWorkflow = constructWorkflowInstanceBuilder().build();
     int parentWorkflowId = dao.insertWorkflowInstance(parentWorkflow);
-    int addChildActionId = addWorkflowAction(parentWorkflowId, new WorkflowInstance.Builder(parentWorkflow).build(), now(), now());
-    WorkflowInstance childWorkflow = constructWorkflowInstanceBuilder().setParentWorkflowId(parentWorkflowId).setParentActionId(addChildActionId).build();
+    int addChildActionId = addWorkflowAction(parentWorkflowId, new WorkflowInstance.Builder(parentWorkflow).build(), now(),
+        now());
+    WorkflowInstance childWorkflow = constructWorkflowInstanceBuilder().setParentWorkflowId(parentWorkflowId)
+        .setParentActionId(addChildActionId).build();
     int childWorkflowId = dao.insertWorkflowInstance(childWorkflow);
-    addWorkflowAction(parentWorkflowId, new WorkflowInstance.Builder(parentWorkflow).putStateVariable("deletedVariable", "deletedValue").build(), now(), now());
-    addWorkflowAction(parentWorkflowId, new WorkflowInstance.Builder(parentWorkflow).putStateVariable("preservedVariable", "preservedValue").build(), now(), now().plusDays(1));
+    addWorkflowAction(parentWorkflowId,
+        new WorkflowInstance.Builder(parentWorkflow).putStateVariable("variable", "deletedValue").build(), now(),
+        now().minusDays(1));
+    addWorkflowAction(parentWorkflowId,
+        new WorkflowInstance.Builder(parentWorkflow).putStateVariable("variable", "preservedValue").build(), now(), now());
 
     assertThat(dao.deleteWorkflowInstanceHistory(parentWorkflowId, 0), equalTo(1));
 
     parentWorkflow = dao.getWorkflowInstance(parentWorkflowId, EnumSet.allOf(WorkflowInstanceInclude.class), null);
     assertThat(parentWorkflow.getStateVariable("requestData"), equalTo("{ \"parameter\": \"abc\" }"));
-    assertThat(parentWorkflow.getStateVariable("deletedVariable"), is(nullValue()));
-    assertThat(parentWorkflow.getStateVariable("preservedVariable"), equalTo("preservedValue"));
+    assertThat(parentWorkflow.getStateVariable("variable"), equalTo("preservedValue"));
     assertThat(parentWorkflow.actions.size(), equalTo(2));
-    childWorkflow = dao.getWorkflowInstance(childWorkflowId, EnumSet.allOf(WorkflowInstanceInclude.class), null);
+    childWorkflow = dao.getWorkflowInstance(childWorkflowId, emptySet(), null);
     assertThat(childWorkflow.parentWorkflowId, equalTo(parentWorkflowId));
   }
 
@@ -947,8 +951,7 @@ private static void checkSameWorkflowInfo(WorkflowInstance i1, WorkflowInstance
 
   private int addWorkflowAction(int workflowId, final WorkflowInstance instance, DateTime started, DateTime ended) {
     final WorkflowInstanceAction action = new WorkflowInstanceAction.Builder().setExecutionStart(started).setExecutorId(42)
-        .setExecutionEnd(ended).setRetryNo(1).setType(stateExecution).setState("test")
-        .setStateText("state text")
+        .setExecutionEnd(ended).setRetryNo(1).setType(stateExecution).setState("test").setStateText("state text")
         .setWorkflowInstanceId(workflowId).build();
     int actionId = transaction.execute(new TransactionCallback<Integer>() {
       @Override

nflow-tests/src/main/java/io/nflow/tests/demo/workflow/DeleteHistoryWorkflow.java

@@ -52,12 +52,11 @@ public DeleteHistoryWorkflow() {
   public NextAction begin(StateExecution execution) {
     WorkflowInstance childWorkflow = new WorkflowInstance.Builder().setType(TYPE).build();
     execution.addChildWorkflows(childWorkflow);
-    execution.setVariable("deletedVariable", "value");
+    execution.setVariable("notDeletedVariable", "value");
     return moveToState(State.process, "Begin");
   }
 
-  public NextAction process(StateExecution execution) {
-    execution.setVariable("anotherDeletedVariable", "value");
+  public NextAction process(@SuppressWarnings("unused") StateExecution execution) {
     return moveToState(State.done, "Process");
   }
 

nflow-tests/src/test/java/io/nflow/tests/DeleteHistoryTest.java

@@ -4,7 +4,6 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
-import static org.hamcrest.Matchers.nullValue;
 import static org.junit.Assert.assertThat;
 import static org.junit.runners.MethodSorters.NAME_ASCENDING;
 
@@ -57,7 +56,7 @@ public void t03_checkInstanceHistoryIsDeleted() {
     assertThat(instance.actions.size(), is(2));
     assertThat(instance.actions.get(0).state, is(equalTo("done")));
     assertThat(instance.actions.get(1).state, is(equalTo("begin")));
-    assertThat(instance.stateVariables, is(nullValue()));
+    assertThat(instance.stateVariables.size(), is(1));
   }
 
 }