Set started_at in group tasks as well.

But just doing that would change current task TTL timeout behavior.
So, this commit also takes care of it.

digdag-core/src/main/java/io/digdag/core/database/DatabaseSessionStoreManager.java

@@ -1184,7 +1184,11 @@ public int trySetChildrenBlockedToReadyOrShortCircuitPlannedOrCanceled(long task
                     " when task_type = " + TaskType.GROUPING_ONLY + " then " + TaskStateCode.PLANNED_CODE +
                     " when " + bitAnd("state_flags", Integer.toString(TaskStateFlags.CANCEL_REQUESTED)) + " != 0 then " + TaskStateCode.CANCELED_CODE +
                     " else " + TaskStateCode.READY_CODE +
-                    " end" +
+                    " end, " +
+                    " started_at = case" +
+                    // Update `started_at` only when it's a group task and the column is null.
+                    // Non group tasks' `started_at` are updated by TaskControlStore.setStartedState.
+                    " when task_type = " + TaskType.GROUPING_ONLY + " then coalesce(started_at, now()) else started_at end" +
                     " where state = " + TaskStateCode.BLOCKED_CODE +
                     " and parent_id = :parentId" +
                     " and exists (" +
@@ -1550,6 +1554,9 @@ public Optional<StoredSessionAttempt> getLastAttemptIfExists(long sessionId)
         public <T> T insertRootTask(long attemptId, Task task, SessionBuilderAction<T> func)
         {
             long taskId = dao.insertTask(attemptId, task.getParentId().orNull(), task.getTaskType().get(), task.getState().get(), task.getStateFlags().get());  // tasks table don't have unique index
+            // Root task's `started_at` isn't updated by trySetChildrenBlockedToReadyOrShortCircuitPlannedOrCanceled().
+            // So this line is just for setting `started_at` of root task although there may be room for optimization.
+            dao.setStartedState(taskId, task.getState().get(), task.getState().get());
             dao.insertTaskDetails(taskId, task.getFullName(), task.getConfig().getLocal(), task.getConfig().getExport());
             dao.insertEmptyTaskStateDetails(taskId);
             return func.call(new DatabaseTaskControlStore(handle), taskId);

digdag-core/src/test/java/io/digdag/core/workflow/WorkflowExecutorTest.java

@@ -2,6 +2,7 @@
 
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -22,8 +23,11 @@
 import io.digdag.client.config.ConfigUtils;
 import io.digdag.core.DigdagEmbed;
 import io.digdag.core.repository.ProjectStoreManager;
+import io.digdag.core.session.ArchivedTask;
 import io.digdag.core.session.SessionStoreManager;
+import io.digdag.core.session.StoredSessionAttemptWithSession;
 import io.digdag.core.session.StoredTask;
+import io.digdag.core.session.TaskStateCode;
 import io.digdag.metrics.StdDigdagMetrics;
 import io.digdag.spi.metrics.DigdagMetrics;
 import io.digdag.util.RetryControl;
@@ -434,6 +438,61 @@ public void randomEnqueue()
         }
     }
 
+    @Test
+    public void verifySubmittedTasks()
+            throws Exception
+    {
+        Instant startTime = Instant.now().truncatedTo(ChronoUnit.SECONDS);
+        StoredSessionAttemptWithSession attempt = runWorkflow("nested", loadYamlResource("/io/digdag/core/workflow/nested.dig"));
+        embed.getLocalSite().runUntilDone(attempt.getId());
+        Instant finishTime = Instant.now().plusSeconds(1).truncatedTo(ChronoUnit.SECONDS);
+
+        List<ArchivedTask> tasks = tm.begin(() -> embed.getLocalSite().getSessionStore().getTasksOfAttempt(attempt.getId()));
+        assertEquals(3, tasks.size());
+
+        Instant rootTaskStartedAt;
+        {
+            // Root task
+            StoredTask task = tasks.get(0);
+            assertEquals("+nested", task.getFullName());
+            assertTrue(task.getTaskType().isGroupingOnly());
+            assertEquals(TaskStateCode.SUCCESS, task.getState());
+            assertThat(task.getUpdatedAt(), is(greaterThanOrEqualTo(startTime)));
+            assertThat(task.getUpdatedAt(), is(lessThanOrEqualTo(finishTime)));
+            assertThat(task.getStartedAt().get(), is(greaterThanOrEqualTo(startTime)));
+            assertThat(task.getStartedAt().get(), is(lessThanOrEqualTo(finishTime)));
+            rootTaskStartedAt = task.getStartedAt().get().truncatedTo(ChronoUnit.SECONDS);
+        }
+
+        Instant parentTaskStartedAt;
+        {
+            // Group task
+            StoredTask task = tasks.get(1);
+            assertEquals("+nested+parent", task.getFullName());
+            assertTrue(task.getTaskType().isGroupingOnly());
+            assertEquals(TaskStateCode.SUCCESS, task.getState());
+            assertThat(task.getUpdatedAt(), is(greaterThanOrEqualTo(startTime)));
+            assertThat(task.getUpdatedAt(), is(lessThanOrEqualTo(finishTime)));
+            assertThat(task.getStartedAt().get(), is(greaterThanOrEqualTo(startTime)));
+            assertThat(task.getStartedAt().get(), is(lessThanOrEqualTo(finishTime)));
+            assertThat(task.getStartedAt().get(), is(greaterThanOrEqualTo(rootTaskStartedAt)));
+            parentTaskStartedAt = task.getStartedAt().get().truncatedTo(ChronoUnit.SECONDS);
+        }
+
+        {
+            // Nested non-group task
+            StoredTask task = tasks.get(2);
+            assertEquals("+nested+parent+child", task.getFullName());
+            assertFalse(task.getTaskType().isGroupingOnly());
+            assertEquals(TaskStateCode.SUCCESS, task.getState());
+            assertThat(task.getUpdatedAt(), is(greaterThanOrEqualTo(startTime)));
+            assertThat(task.getUpdatedAt(), is(lessThanOrEqualTo(finishTime)));
+            assertThat(task.getStartedAt().get(), is(greaterThanOrEqualTo(startTime)));
+            assertThat(task.getStartedAt().get(), is(lessThanOrEqualTo(finishTime)));
+            assertThat(task.getStartedAt().get(), is(greaterThanOrEqualTo(parentTaskStartedAt)));
+        }
+    }
+
     private Optional<String> getResult(String fileName, TemporaryFolder folder)
     {
         try {
@@ -445,15 +504,15 @@ private Optional<String> getResult(String fileName, TemporaryFolder folder)
         }
     }
 
-    private void runWorkflow(String workflowName, Config config)
+    private StoredSessionAttemptWithSession runWorkflow(String workflowName, Config config)
             throws Exception
     {
-        WorkflowTestingUtils.runWorkflow(embed, folder.getRoot().toPath(), workflowName, config);
+        return WorkflowTestingUtils.runWorkflow(embed, folder.getRoot().toPath(), workflowName, config);
     }
 
-    private void runWorkflow(DigdagEmbed embed, String workflowName, Config config)
+    private StoredSessionAttemptWithSession runWorkflow(DigdagEmbed embed, String workflowName, Config config)
             throws Exception
     {
-        WorkflowTestingUtils.runWorkflow(embed, folder.getRoot().toPath(), workflowName, config);
+        return WorkflowTestingUtils.runWorkflow(embed, folder.getRoot().toPath(), workflowName, config);
     }
 }

digdag-core/src/test/resources/io/digdag/core/workflow/nested.dig

@@ -0,0 +1,4 @@
++parent:
+
+  +child:
+    echo>: Hello

digdag-server/src/main/java/io/digdag/server/WorkflowExecutionTimeoutEnforcer.java

@@ -46,11 +46,14 @@ public class WorkflowExecutionTimeoutEnforcer
     private static final Duration DEFAULT_REAPING_INTERVAL = Duration.ofSeconds(5);
 
     // This is similar to TaskStateCode.notDoneStates() but BLOCKED, PLANNED, and READY are excluded.
-    // BLOCKED and PLANNED are excluded because there're other running tasks that should be enforced instead.
-    // READY is excluded the workflow itself is working correctly and number of threads is insufficient.
+    //
+    // - BLOCKED and PLANNED are excluded because there're other running tasks that should be enforced instead.
+    // - READY is excluded the workflow itself is working correctly and number of threads is insufficient.
+    // - GROUP_RETRY_WAITING is excluded because this status is supposed to be immediately changed to PLANNED
+    //   and long running group tasks with this state could happen only due to WorkflowExecutor trouble
+    //   not due to workflow definitions.
     private static final TaskStateCode[] TASK_TTL_ENFORCED_STATE_CODES = new TaskStateCode[] {
         TaskStateCode.RETRY_WAITING,
-        TaskStateCode.GROUP_RETRY_WAITING,
         TaskStateCode.RUNNING,
     };