diff --git a/cmd/hatchet-migrate/migrate/migrations/20260702145450_v1_0_123.sql b/cmd/hatchet-migrate/migrate/migrations/20260702145450_v1_0_123.sql new file mode 100644 index 0000000000..f16ec97e0e --- /dev/null +++ b/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 diff --git a/internal/services/controllers/task/controller.go b/internal/services/controllers/task/controller.go index 5019eb61af..98b9fb187e 100644 --- a/internal/services/controllers/task/controller.go +++ b/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() diff --git a/pkg/repository/sqlcv1/concurrency.sql b/pkg/repository/sqlcv1/concurrency.sql index f9ece1109d..c741c0168b 100644 --- a/pkg/repository/sqlcv1/concurrency.sql +++ b/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 diff --git a/pkg/repository/sqlcv1/concurrency.sql.go b/pkg/repository/sqlcv1/concurrency.sql.go index 5f314d7acb..2ae5ebe414 100644 --- a/pkg/repository/sqlcv1/concurrency.sql.go +++ b/pkg/repository/sqlcv1/concurrency.sql.go @@ -115,6 +115,8 @@ WITH tenant_step_concurrencies AS ( FROM v1_step_concurrency sc WHERE sc.tenant_id = $1::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 @@ -144,7 +146,7 @@ func (q *Queries) DeactivateStaleStepConcurrency(ctx context.Context, db DBTX, t const getConcurrencyStrategyById = `-- name: GetConcurrencyStrategyById :one SELECT - sc.id, sc.parent_strategy_id, sc.workflow_id, sc.workflow_version_id, sc.step_id, sc.is_active, sc.strategy, sc.expression, sc.tenant_id, sc.max_concurrency + sc.id, sc.parent_strategy_id, sc.workflow_id, sc.workflow_version_id, sc.step_id, sc.is_active, sc.last_active_at, sc.strategy, sc.expression, sc.tenant_id, sc.max_concurrency FROM v1_step_concurrency sc WHERE @@ -167,6 +169,7 @@ func (q *Queries) GetConcurrencyStrategyById(ctx context.Context, db DBTX, arg G &i.WorkflowVersionID, &i.StepID, &i.IsActive, + &i.LastActiveAt, &i.Strategy, &i.Expression, &i.TenantID, @@ -220,7 +223,7 @@ func (q *Queries) GetWorkflowConcurrencyQueueCounts(ctx context.Context, db DBTX const listActiveConcurrencyStrategies = `-- name: ListActiveConcurrencyStrategies :many SELECT - sc.id, sc.parent_strategy_id, sc.workflow_id, sc.workflow_version_id, sc.step_id, sc.is_active, sc.strategy, sc.expression, sc.tenant_id, sc.max_concurrency + sc.id, sc.parent_strategy_id, sc.workflow_id, sc.workflow_version_id, sc.step_id, sc.is_active, sc.last_active_at, sc.strategy, sc.expression, sc.tenant_id, sc.max_concurrency FROM v1_step_concurrency sc JOIN @@ -246,6 +249,7 @@ func (q *Queries) ListActiveConcurrencyStrategies(ctx context.Context, db DBTX, &i.WorkflowVersionID, &i.StepID, &i.IsActive, + &i.LastActiveAt, &i.Strategy, &i.Expression, &i.TenantID, @@ -341,7 +345,7 @@ func (q *Queries) ListConcurrencySlotsForIndexing(ctx context.Context, db DBTX, const listConcurrencyStrategiesByStepId = `-- name: ListConcurrencyStrategiesByStepId :many SELECT - id, parent_strategy_id, workflow_id, workflow_version_id, step_id, is_active, strategy, expression, tenant_id, max_concurrency + id, parent_strategy_id, workflow_id, workflow_version_id, step_id, is_active, last_active_at, strategy, expression, tenant_id, max_concurrency FROM v1_step_concurrency WHERE @@ -370,6 +374,7 @@ func (q *Queries) ListConcurrencyStrategiesByStepId(ctx context.Context, db DBTX &i.WorkflowVersionID, &i.StepID, &i.IsActive, + &i.LastActiveAt, &i.Strategy, &i.Expression, &i.TenantID, @@ -386,7 +391,7 @@ func (q *Queries) ListConcurrencyStrategiesByStepId(ctx context.Context, db DBTX } const listConcurrencyStrategiesByWorkflowVersionId = `-- name: ListConcurrencyStrategiesByWorkflowVersionId :many -SELECT c.id, c.parent_strategy_id, c.workflow_id, c.workflow_version_id, c.step_id, c.is_active, c.strategy, c.expression, c.tenant_id, c.max_concurrency, s."readableId" AS step_readable_id +SELECT c.id, c.parent_strategy_id, c.workflow_id, c.workflow_version_id, c.step_id, c.is_active, c.last_active_at, c.strategy, c.expression, c.tenant_id, c.max_concurrency, s."readableId" AS step_readable_id FROM v1_step_concurrency c JOIN "Step" s ON s.id = c.step_id WHERE @@ -416,6 +421,7 @@ type ListConcurrencyStrategiesByWorkflowVersionIdRow struct { WorkflowVersionID uuid.UUID `json:"workflow_version_id"` StepID uuid.UUID `json:"step_id"` IsActive bool `json:"is_active"` + LastActiveAt pgtype.Timestamptz `json:"last_active_at"` Strategy V1ConcurrencyStrategy `json:"strategy"` Expression string `json:"expression"` TenantID uuid.UUID `json:"tenant_id"` @@ -439,6 +445,7 @@ func (q *Queries) ListConcurrencyStrategiesByWorkflowVersionId(ctx context.Conte &i.WorkflowVersionID, &i.StepID, &i.IsActive, + &i.LastActiveAt, &i.Strategy, &i.Expression, &i.TenantID, diff --git a/pkg/repository/sqlcv1/models.go b/pkg/repository/sqlcv1/models.go index c795e5214d..016059f636 100644 --- a/pkg/repository/sqlcv1/models.go +++ b/pkg/repository/sqlcv1/models.go @@ -3563,6 +3563,7 @@ type V1StepConcurrency struct { WorkflowVersionID uuid.UUID `json:"workflow_version_id"` StepID uuid.UUID `json:"step_id"` IsActive bool `json:"is_active"` + LastActiveAt pgtype.Timestamptz `json:"last_active_at"` Strategy V1ConcurrencyStrategy `json:"strategy"` Expression string `json:"expression"` TenantID uuid.UUID `json:"tenant_id"` diff --git a/pkg/repository/sqlcv1/workflows.sql.go b/pkg/repository/sqlcv1/workflows.sql.go index 71b9903a86..eeec44eaaa 100644 --- a/pkg/repository/sqlcv1/workflows.sql.go +++ b/pkg/repository/sqlcv1/workflows.sql.go @@ -263,7 +263,7 @@ VALUES ( $5::text, $6::uuid, $7::integer -) RETURNING id, parent_strategy_id, workflow_id, workflow_version_id, step_id, is_active, strategy, expression, tenant_id, max_concurrency +) RETURNING id, parent_strategy_id, workflow_id, workflow_version_id, step_id, is_active, last_active_at, strategy, expression, tenant_id, max_concurrency ` type CreateStepConcurrencyParams struct { @@ -294,6 +294,7 @@ func (q *Queries) CreateStepConcurrency(ctx context.Context, db DBTX, arg Create &i.WorkflowVersionID, &i.StepID, &i.IsActive, + &i.LastActiveAt, &i.Strategy, &i.Expression, &i.TenantID, @@ -648,7 +649,7 @@ WITH inserted_wcs AS ( wv."id" = $2::uuid AND j."kind" = 'DEFAULT' ) s, inserted_wcs wcs - RETURNING id, parent_strategy_id, workflow_id, workflow_version_id, step_id, is_active, strategy, expression, tenant_id, max_concurrency + RETURNING id, parent_strategy_id, workflow_id, workflow_version_id, step_id, is_active, last_active_at, strategy, expression, tenant_id, max_concurrency ) SELECT wcs.id, diff --git a/pkg/scheduling/v1/concurrency_integration_test.go b/pkg/scheduling/v1/concurrency_integration_test.go index d7ef96d102..d0ead36748 100644 --- a/pkg/scheduling/v1/concurrency_integration_test.go +++ b/pkg/scheduling/v1/concurrency_integration_test.go @@ -624,6 +624,13 @@ func TestConcurrency_ColdStrategyScheduledPromptly(t *testing.T) { concurrencyRepo := r.Scheduler().Concurrency() + // The stale-deactivation sweep only considers strategies whose last_active_at is over 25 + // hours old (last_active_at is otherwise refreshed at most once per hour on slot inserts). + // The strategy was just created with last_active_at=NOW(), so backdate it to simulate a + // strategy that has genuinely been idle for the deactivation window. + _, err = conf.Pool.Exec(ctx, `UPDATE v1_step_concurrency SET last_active_at = NOW() - INTERVAL '26 hours' WHERE id = $1`, strat.ID) + require.NoError(t, err) + // Make the strategy "cold": with no slots yet, the stale-deactivation sweep flips it to // is_active=FALSE. This mirrors a strategy that has been idle for the deactivation window. require.NoError(t, concurrencyRepo.DeactivateStaleStepConcurrency(ctx, tenantId)) diff --git a/sql/schema/v1-core.sql b/sql/schema/v1-core.sql index ce13cba780..f6707697e9 100644 --- a/sql/schema/v1-core.sql +++ b/sql/schema/v1-core.sql @@ -237,6 +237,8 @@ CREATE TABLE v1_step_concurrency ( step_id UUID NOT NULL, -- If the strategy is NONE and we've removed all concurrency slots, we can set is_active to false is_active BOOLEAN NOT NULL DEFAULT TRUE, + -- last_active_at is refreshed at most once per hour when a new slot is inserted for this strategy. + last_active_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, strategy v1_concurrency_strategy NOT NULL, expression TEXT NOT NULL, tenant_id UUID NOT NULL, @@ -907,12 +909,13 @@ BEGIN 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 + SET is_active = TRUE, last_active_at = NOW() FROM inactive_strategies WHERE strategy.workflow_id = inactive_strategies.workflow_id AND