mirror/hatchet
1
0
Fork 0
mirror of https://github.com/hatchet-dev/hatchet.git synced 2026-02-20 07:09:13 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix: concurrency timeout from 5s -> 30s (#1926)

Browse Source 
* fix: concurrency timeout from 5s -> 30s

* limits in overwrite file too
This commit is contained in:
abelanger5
2025-07-01 08:05:59 -04:00
committed by GitHub
parent 0e7cc7e063
commit 646adda2a8
2 changed files with 7 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pkg/repository/v1/scheduler_concurrency.go
View File

@@ -60,7 +60,7 @@ func (c *ConcurrencyRepositoryImpl) UpdateConcurrencyStrategyIsActive(
tenantId pgtype.UUID,
strategy *sqlcv1.V1StepConcurrency,
) error {
tx, commit, rollback, err := sqlchelpers.PrepareTx(ctx, c.pool, c.l, 5000)
tx, commit, rollback, err := sqlchelpers.PrepareTx(ctx, c.pool, c.l, 30000)
if err != nil {
return err
@@ -136,7 +136,7 @@ func (c *ConcurrencyRepositoryImpl) runGroupRoundRobin(
strategy *sqlcv1.V1StepConcurrency,
) (res *RunConcurrencyResult, err error) {
tx, commit, rollback, err := sqlchelpers.PrepareTx(ctx, c.pool, c.l, 5000)
tx, commit, rollback, err := sqlchelpers.PrepareTx(ctx, c.pool, c.l, 30000)
if err != nil {
return nil, fmt.Errorf("failed to prepare transaction (strategy ID: %d): %w", strategy.ID, err)
@@ -277,7 +277,7 @@ func (c *ConcurrencyRepositoryImpl) runCancelInProgress(
strategy *sqlcv1.V1StepConcurrency,
) (res *RunConcurrencyResult, err error) {
tx, commit, rollback, err := sqlchelpers.PrepareTx(ctx, c.pool, c.l, 5000)
tx, commit, rollback, err := sqlchelpers.PrepareTx(ctx, c.pool, c.l, 30000)
if err != nil {
return nil, fmt.Errorf("failed to prepare transaction (strategy ID: %d): %w", strategy.ID, err)
@@ -496,7 +496,7 @@ func (c *ConcurrencyRepositoryImpl) runCancelNewest(
tenantId pgtype.UUID,
strategy *sqlcv1.V1StepConcurrency,
) (res *RunConcurrencyResult, err error) {
tx, commit, rollback, err := sqlchelpers.PrepareTx(ctx, c.pool, c.l, 5000)
tx, commit, rollback, err := sqlchelpers.PrepareTx(ctx, c.pool, c.l, 30000)
if err != nil {
return nil, fmt.Errorf("failed to prepare transaction (strategy ID: %d): %w", strategy.ID, err)

3
pkg/repository/v1/sqlcv1/concurrency-overwrite.sql.go
View File

@@ -39,6 +39,7 @@ WITH filled_parent_slots AS (
ORDER BY
task_id, task_inserted_at
FOR UPDATE
LIMIT 1000
), eligible_slots AS (
SELECT
cs.sort_id, cs.task_id, cs.task_inserted_at, cs.task_retry_count, cs.external_id, cs.tenant_id, cs.workflow_id, cs.workflow_version_id, cs.workflow_run_id, cs.strategy_id, cs.parent_strategy_id, cs.priority, cs.key, cs.is_filled, cs.next_parent_strategy_ids, cs.next_strategy_ids, cs.next_keys, cs.queue_to_notify, cs.schedule_timeout_at
@@ -207,6 +208,7 @@ WITH slots AS (
strategy_id = $2::bigint AND
schedule_timeout_at < NOW() AND
is_filled = FALSE
LIMIT 1000
), eligible_running_slots AS (
SELECT
task_id,
@@ -429,6 +431,7 @@ WITH slots AS (
strategy_id = $2::bigint AND
schedule_timeout_at < NOW() AND
is_filled = FALSE
LIMIT 1000
), eligible_running_slots AS (
SELECT
task_id,