fix: add last active column to step concurrency table

cmd/hatchet-migrate/migrate/migrations/20260702145450_v1_0_123.sql

@@ -0,0 +1,181 @@
+-- +goose Up
+-- +goose StatementBegin
+
+ALTER TABLE v1_step_concurrency
+    ADD COLUMN last_active_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP;
+
+CREATE OR REPLACE FUNCTION after_v1_concurrency_slot_insert_function()
+RETURNS trigger AS $$
+BEGIN
+    WITH parent_slot AS (
+        SELECT
+            *
+        FROM
+            new_table cs
+        WHERE
+            cs.parent_strategy_id IS NOT NULL
+    ), parent_to_child_strategy_ids AS (
+        SELECT
+            wc.id AS parent_strategy_id,
+            wc.tenant_id,
+            ps.workflow_id,
+            ps.workflow_version_id,
+            ps.workflow_run_id,
+            MAX(ps.sort_id) AS sort_id,
+            MAX(ps.priority) AS priority,
+            MAX(ps.key) AS key,
+            ARRAY_AGG(DISTINCT wc.child_strategy_ids) AS child_strategy_ids
+        FROM
+            parent_slot ps
+        JOIN v1_workflow_concurrency wc ON wc.workflow_id = ps.workflow_id AND wc.workflow_version_id = ps.workflow_version_id AND wc.id = ps.parent_strategy_id
+        GROUP BY
+            wc.id,
+            wc.tenant_id,
+            ps.workflow_id,
+            ps.workflow_version_id,
+            ps.workflow_run_id
+    )
+    INSERT INTO v1_workflow_concurrency_slot (
+        sort_id,
+        tenant_id,
+        workflow_id,
+        workflow_version_id,
+        workflow_run_id,
+        strategy_id,
+        child_strategy_ids,
+        priority,
+        key
+    )
+    SELECT
+        pcs.sort_id,
+        pcs.tenant_id,
+        pcs.workflow_id,
+        pcs.workflow_version_id,
+        pcs.workflow_run_id,
+        pcs.parent_strategy_id,
+        pcs.child_strategy_ids,
+        pcs.priority,
+        pcs.key
+    FROM
+        parent_to_child_strategy_ids pcs
+    ON CONFLICT (strategy_id, workflow_version_id, workflow_run_id) DO NOTHING;
+
+    WITH inactive_strategies AS (
+        SELECT
+            strategy.*
+        FROM
+            new_table cs
+        JOIN
+            v1_step_concurrency strategy ON strategy.workflow_id = cs.workflow_id AND strategy.workflow_version_id = cs.workflow_version_id AND strategy.id = cs.strategy_id
+        WHERE
+            strategy.is_active = FALSE
+            OR strategy.last_active_at < NOW() - INTERVAL '1 hour'
+        ORDER BY
+            strategy.id
+        FOR UPDATE
+    )
+    UPDATE v1_step_concurrency strategy
+    SET is_active = TRUE, last_active_at = NOW()
+    FROM inactive_strategies
+    WHERE
+        strategy.workflow_id = inactive_strategies.workflow_id AND
+        strategy.workflow_version_id = inactive_strategies.workflow_version_id AND
+        strategy.step_id = inactive_strategies.step_id AND
+        strategy.id = inactive_strategies.id;
+
+    RETURN NULL;
+END;
+
+$$ LANGUAGE plpgsql;
+
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+
+CREATE OR REPLACE FUNCTION after_v1_concurrency_slot_insert_function()
+RETURNS trigger AS $$
+BEGIN
+    WITH parent_slot AS (
+        SELECT
+            *
+        FROM
+            new_table cs
+        WHERE
+            cs.parent_strategy_id IS NOT NULL
+    ), parent_to_child_strategy_ids AS (
+        SELECT
+            wc.id AS parent_strategy_id,
+            wc.tenant_id,
+            ps.workflow_id,
+            ps.workflow_version_id,
+            ps.workflow_run_id,
+            MAX(ps.sort_id) AS sort_id,
+            MAX(ps.priority) AS priority,
+            MAX(ps.key) AS key,
+            ARRAY_AGG(DISTINCT wc.child_strategy_ids) AS child_strategy_ids
+        FROM
+            parent_slot ps
+        JOIN v1_workflow_concurrency wc ON wc.workflow_id = ps.workflow_id AND wc.workflow_version_id = ps.workflow_version_id AND wc.id = ps.parent_strategy_id
+        GROUP BY
+            wc.id,
+            wc.tenant_id,
+            ps.workflow_id,
+            ps.workflow_version_id,
+            ps.workflow_run_id
+    )
+    INSERT INTO v1_workflow_concurrency_slot (
+        sort_id,
+        tenant_id,
+        workflow_id,
+        workflow_version_id,
+        workflow_run_id,
+        strategy_id,
+        child_strategy_ids,
+        priority,
+        key
+    )
+    SELECT
+        pcs.sort_id,
+        pcs.tenant_id,
+        pcs.workflow_id,
+        pcs.workflow_version_id,
+        pcs.workflow_run_id,
+        pcs.parent_strategy_id,
+        pcs.child_strategy_ids,
+        pcs.priority,
+        pcs.key
+    FROM
+        parent_to_child_strategy_ids pcs
+    ON CONFLICT (strategy_id, workflow_version_id, workflow_run_id) DO NOTHING;
+
+    WITH inactive_strategies AS (
+        SELECT
+            strategy.*
+        FROM
+            new_table cs
+        JOIN
+            v1_step_concurrency strategy ON strategy.workflow_id = cs.workflow_id AND strategy.workflow_version_id = cs.workflow_version_id AND strategy.id = cs.strategy_id
+        WHERE
+            strategy.is_active = FALSE
+        ORDER BY
+            strategy.id
+        FOR UPDATE
+    )
+    UPDATE v1_step_concurrency strategy
+    SET is_active = TRUE
+    FROM inactive_strategies
+    WHERE
+        strategy.workflow_id = inactive_strategies.workflow_id AND
+        strategy.workflow_version_id = inactive_strategies.workflow_version_id AND
+        strategy.step_id = inactive_strategies.step_id AND
+        strategy.id = inactive_strategies.id;
+
+    RETURN NULL;
+END;
+
+$$ LANGUAGE plpgsql;
+
+ALTER TABLE v1_step_concurrency DROP COLUMN last_active_at;
+
+-- +goose StatementEnd

internal/services/controllers/task/controller.go

@@ -42,25 +42,26 @@ type TasksController interface {
 }
 
 type TasksControllerImpl struct {
-	mq                                    msgqueue.MessageQueue
-	pubBuffer                             *msgqueue.MQPubBuffer
-	l                                     *zerolog.Logger
-	queueLogger                           *zerolog.Logger
-	pgxStatsLogger                        *zerolog.Logger
-	repov1                                v1.Repository
-	dv                                    datautils.DataDecoderValidator
-	s                                     gocron.Scheduler
-	a                                     *hatcheterrors.Wrapped
-	p                                     *partition.Partition
-	celParser                             *cel.CELParser
-	opsPoolPollInterval                   time.Duration
-	opsPoolJitter                         time.Duration
-	timeoutTaskOperations                 *operation.TenantOperationPool
-	reassignTaskOperations                *operation.TenantOperationPool
-	retryTaskOperations                   *operation.TenantOperationPool
-	emitSleepOperations                   *operation.TenantOperationPool
-	evictExpiredIdempotencyKeysOperations *operation.TenantOperationPool
-	// deactivateStaleStepConcurrencyOperations *operation.TenantOperationPool
+	mq                                       msgqueue.MessageQueue
+	pubBuffer                                *msgqueue.MQPubBuffer
+	l                                        *zerolog.Logger
+	queueLogger                              *zerolog.Logger
+	pgxStatsLogger                           *zerolog.Logger
+	repov1                                   v1.Repository
+	dv                                       datautils.DataDecoderValidator
+	s                                        gocron.Scheduler
+	a                                        *hatcheterrors.Wrapped
+	p                                        *partition.Partition
+	celParser                                *cel.CELParser
+	opsPoolPollInterval                      time.Duration
+	opsPoolJitter                            time.Duration
+	timeoutTaskOperations                    *operation.TenantOperationPool
+	reassignTaskOperations                   *operation.TenantOperationPool
+	retryTaskOperations                      *operation.TenantOperationPool
+	emitSleepOperations                      *operation.TenantOperationPool
+	evictExpiredIdempotencyKeysOperations    *operation.TenantOperationPool
+	deactivateStaleStepConcurrencyOperations *operation.TenantOperationPool
+
 	replayEnabled       bool
 	analyzeCronInterval time.Duration
 	signaler            *signal.OLAPSignaler
@@ -286,16 +287,14 @@ func New(fs ...TasksControllerOpt) (*TasksControllerImpl, error) {
 		opts.repov1.Tasks().DefaultTaskActivityGauge,
 	))
 
-	// FIXME(mnafees): temporarily disabling this operation for the meantime
-	//
-	// t.deactivateStaleStepConcurrencyOperations = operation.NewTenantOperationPool(opts.p, opts.l, "deactivate-stale-step-concurrency", timeout, "deactivate stale step concurrency", t.deactivateStaleStepConcurrency, operation.WithPoolInterval(
-	// 	opts.repov1.IntervalSettings(),
-	// 	jitter,
-	// 	15*time.Minute,
-	// 	30*time.Minute,
-	// 	3,
-	// 	opts.repov1.Tasks().DefaultTaskActivityGauge,
-	// ))
+	t.deactivateStaleStepConcurrencyOperations = operation.NewTenantOperationPool(opts.p, opts.l, "deactivate-stale-step-concurrency", timeout, "deactivate stale step concurrency", t.deactivateStaleStepConcurrency, operation.WithPoolInterval(
+		opts.repov1.IntervalSettings(),
+		jitter,
+		15*time.Minute,
+		30*time.Minute,
+		3,
+		opts.repov1.Tasks().DefaultTaskActivityGauge,
+	))
 
 	return t, nil
 }
@@ -412,7 +411,7 @@ func (tc *TasksControllerImpl) Start() (func() error, error) {
 		tc.retryTaskOperations.Cleanup()
 		tc.emitSleepOperations.Cleanup()
 		tc.evictExpiredIdempotencyKeysOperations.Cleanup()
-		// tc.deactivateStaleStepConcurrencyOperations.Cleanup()
+		tc.deactivateStaleStepConcurrencyOperations.Cleanup()
 
 		tc.pubBuffer.Stop()
 

pkg/repository/sqlcv1/concurrency.sql

@@ -802,6 +802,8 @@ WITH tenant_step_concurrencies AS (
     FROM v1_step_concurrency sc
     WHERE sc.tenant_id = @tenantId::UUID
         AND sc.is_active = TRUE
+        -- we use 25 hours because there's a 1-hour cache on updating last_active_at
+        AND sc.last_active_at < NOW() - INTERVAL '25 hours'
         AND NOT EXISTS (
             SELECT 1 FROM v1_concurrency_slot cs
             WHERE