feat(runs): add ActionRepo.CountActiveActionsByPhase

runs/repository/impl/action.go

@@ -44,6 +44,26 @@ const (
 	defaultListenerMinReconnectInterval = 10 * time.Second
 )
 
+// terminalPhases are the phases an action ends in. FAILED and TIMED_OUT can
+// still move on when the action is retried (see UpdateActionPhase).
+var terminalPhases = []int32{
+	int32(common.ActionPhase_ACTION_PHASE_SUCCEEDED),
+	int32(common.ActionPhase_ACTION_PHASE_FAILED),
+	int32(common.ActionPhase_ACTION_PHASE_ABORTED),
+	int32(common.ActionPhase_ACTION_PHASE_TIMED_OUT),
+	int32(common.ActionPhase_ACTION_PHASE_RECOVERED),
+}
+
+// activePhases is every ActionPhase not in terminalPhases.
+var activePhases = []int32{
+	int32(common.ActionPhase_ACTION_PHASE_UNSPECIFIED),
+	int32(common.ActionPhase_ACTION_PHASE_QUEUED),
+	int32(common.ActionPhase_ACTION_PHASE_WAITING_FOR_RESOURCES),
+	int32(common.ActionPhase_ACTION_PHASE_INITIALIZING),
+	int32(common.ActionPhase_ACTION_PHASE_RUNNING),
+	int32(common.ActionPhase_ACTION_PHASE_PAUSED),
+}
+
 type notificationBuffer string
 
 // NotificationConfig configures the pending notification queues and delivery retries.
@@ -556,13 +576,6 @@ func (r *actionRepo) UpdateActionPhase(
 		int32(common.ActionPhase_ACTION_PHASE_FAILED),
 		int32(common.ActionPhase_ACTION_PHASE_TIMED_OUT),
 	}
-	terminalPhases := []int32{
-		int32(common.ActionPhase_ACTION_PHASE_SUCCEEDED),
-		int32(common.ActionPhase_ACTION_PHASE_FAILED),
-		int32(common.ActionPhase_ACTION_PHASE_ABORTED),
-		int32(common.ActionPhase_ACTION_PHASE_TIMED_OUT),
-		int32(common.ActionPhase_ACTION_PHASE_RECOVERED),
-	}
 
 	var queryBuilder strings.Builder
 	var args []interface{}
@@ -1224,6 +1237,28 @@ func (r *actionRepo) ListRootActions(ctx context.Context, project, domain string
 	return actions, nil
 }
 
+// CountActiveActionsByPhase counts the actions in an active phase, split by phase
+// and by whether the action is a root (a run).
+func (r *actionRepo) CountActiveActionsByPhase(ctx context.Context) (map[interfaces.PhaseKey]int64, error) {
+	var rows []struct {
+		Phase  int32 `db:"phase"`
+		IsRoot bool  `db:"is_root"`
+		Count  int64 `db:"count"`
+	}
+	err := sqlx.SelectContext(ctx, r.db, &rows,
+		`SELECT phase, parent_action_name IS NULL AS is_root, COUNT(*) AS count
+		FROM actions WHERE phase = ANY($1) GROUP BY 1, 2`,
+		pq.Array(activePhases))
+	if err != nil {
+		return nil, fmt.Errorf("failed to count actions by phase: %w", err)
+	}
+	counts := make(map[interfaces.PhaseKey]int64, len(rows))
+	for _, row := range rows {
+		counts[interfaces.PhaseKey{Phase: row.Phase, IsRoot: row.IsRoot}] = row.Count
+	}
+	return counts, nil
+}
+
 // startNotifyLoop acquires a dedicated DB connection for NOTIFY commands and
 // starts processing in the background. Returns an error if the initial
 // connection cannot be established.

runs/repository/impl/action_test.go

@@ -545,6 +545,82 @@ func TestListRuns_HasPausedActionFilter(t *testing.T) {
 	assert.False(t, runs[0].ParentActionName.Valid, "only the root action should be returned")
 }
 
+func TestCountActiveActionsByPhase(t *testing.T) {
+	db := setupActionDB(t)
+	actionRepo, err := NewActionRepo(db, testDbConfig, testNotificationConfig)
+	require.NoError(t, err)
+	ctx := context.Background()
+
+	t.Run("empty table", func(t *testing.T) {
+		counts, err := actionRepo.CountActiveActionsByPhase(ctx)
+		require.NoError(t, err)
+		assert.Empty(t, counts)
+	})
+
+	t.Run("counts only active actions", func(t *testing.T) {
+		root := func(run string, phase common.ActionPhase) *models.Action {
+			return &models.Action{Project: "proj1", Domain: "domain1", RunName: run, Name: rootActionName,
+				Phase: int32(phase)}
+		}
+		child := func(run, name string, phase common.ActionPhase) *models.Action {
+			return &models.Action{Project: "proj1", Domain: "domain1", RunName: run, Name: name,
+				ParentActionName: sql.NullString{String: rootActionName, Valid: true},
+				Phase:            int32(phase)}
+		}
+		for _, a := range []*models.Action{
+			// r1: a live run with children in live and terminal phases.
+			root("r1", common.ActionPhase_ACTION_PHASE_RUNNING),
+			child("r1", "c1", common.ActionPhase_ACTION_PHASE_QUEUED),
+			child("r1", "c2", common.ActionPhase_ACTION_PHASE_WAITING_FOR_RESOURCES),
+			child("r1", "c3", common.ActionPhase_ACTION_PHASE_INITIALIZING),
+			child("r1", "c4", common.ActionPhase_ACTION_PHASE_RUNNING),
+			child("r1", "c5", common.ActionPhase_ACTION_PHASE_SUCCEEDED),
+			child("r1", "c6", common.ActionPhase_ACTION_PHASE_FAILED),
+			// r2: a paused run is still live.
+			root("r2", common.ActionPhase_ACTION_PHASE_PAUSED),
+			// r3-r7: one run in each terminal phase; none are counted.
+			root("r3", common.ActionPhase_ACTION_PHASE_SUCCEEDED),
+			child("r3", "c1", common.ActionPhase_ACTION_PHASE_SUCCEEDED),
+			root("r4", common.ActionPhase_ACTION_PHASE_FAILED),
+			root("r5", common.ActionPhase_ACTION_PHASE_ABORTED),
+			child("r5", "c1", common.ActionPhase_ACTION_PHASE_ABORTED),
+			root("r6", common.ActionPhase_ACTION_PHASE_TIMED_OUT),
+			root("r7", common.ActionPhase_ACTION_PHASE_RECOVERED),
+		} {
+			_, err := actionRepo.CreateAction(ctx, a, false)
+			require.NoError(t, err)
+		}
+
+		counts, err := actionRepo.CountActiveActionsByPhase(ctx)
+		require.NoError(t, err)
+		key := func(phase common.ActionPhase, isRoot bool) interfaces.PhaseKey {
+			return interfaces.PhaseKey{Phase: int32(phase), IsRoot: isRoot}
+		}
+		assert.Equal(t, map[interfaces.PhaseKey]int64{
+			// Runs (root actions).
+			key(common.ActionPhase_ACTION_PHASE_RUNNING, true): 1, // r1
+			key(common.ActionPhase_ACTION_PHASE_PAUSED, true):  1, // r2
+
+			// Child actions.
+			key(common.ActionPhase_ACTION_PHASE_QUEUED, false):                1, // r1/c1
+			key(common.ActionPhase_ACTION_PHASE_WAITING_FOR_RESOURCES, false): 1, // r1/c2
+			key(common.ActionPhase_ACTION_PHASE_INITIALIZING, false):          1, // r1/c3
+			key(common.ActionPhase_ACTION_PHASE_RUNNING, false):               1, // r1/c4
+		}, counts)
+
+		// The gauge totals: runs are the root rows, actions are all rows.
+		var runs, actions int64
+		for k, n := range counts {
+			if k.IsRoot {
+				runs += n
+			}
+			actions += n
+		}
+		assert.Equal(t, int64(2), runs, "active runs: r1, r2")
+		assert.Equal(t, int64(6), actions, "active actions: r1, r2, r1/c1-c4")
+	})
+}
+
 // TestListRuns_SearchFilter runs the console's run search predicate against a real
 // database: one term must match either the run name or the task name, case-insensitively,
 // and EQUAL on a full run name must select exactly that run.

runs/repository/interfaces/action.go

@@ -54,4 +54,14 @@ type ActionRepo interface {
 
 	// Aggregation operations
 	ListRootActions(ctx context.Context, project, domain string, startDate, endDate *time.Time, limit int) ([]*models.Action, error)
+
+	// CountActiveActionsByPhase counts the actions in an active phase, split by
+	// phase and by whether the action is a root (a run).
+	CountActiveActionsByPhase(ctx context.Context) (counts map[PhaseKey]int64, err error)
+}
+
+// PhaseKey groups the counts returned by ActionRepo.CountActiveActionsByPhase.
+type PhaseKey struct {
+	Phase  int32
+	IsRoot bool
 }