fix: poll for stuck

internal/services/ticker/ticker.go

@@ -14,8 +14,10 @@ import (
 	"github.com/hatchet-dev/hatchet/internal/integrations/alerting"
 	"github.com/hatchet-dev/hatchet/internal/msgqueue"
 	"github.com/hatchet-dev/hatchet/internal/services/partition"
+	"github.com/hatchet-dev/hatchet/internal/services/shared/tasktypes"
 	"github.com/hatchet-dev/hatchet/pkg/logger"
 	"github.com/hatchet-dev/hatchet/pkg/repository"
+	"github.com/hatchet-dev/hatchet/pkg/repository/prisma/sqlchelpers"
 )
 
 type Ticker interface {
@@ -204,6 +206,19 @@ func (t *TickerImpl) Start() (func() error, error) {
 		return nil, fmt.Errorf("could not create poll cron schedules job: %w", err)
 	}
 
+	_, err = t.s.NewJob(
+		// check every 30 seconds
+		gocron.DurationJob(time.Second*30),
+		gocron.NewTask(
+			t.runPollOrphanedStepRuns(ctx),
+		),
+	)
+
+	if err != nil {
+		cancel()
+		return nil, fmt.Errorf("could not create poll orphaned step runs job: %w", err)
+	}
+
 	_, err = t.s.NewJob(
 		// we look ahead every 5 seconds
 		gocron.DurationJob(time.Second*5),
@@ -296,6 +311,37 @@ func (t *TickerImpl) Start() (func() error, error) {
 	return cleanup, nil
 }
 
+func (t *TickerImpl) runPollOrphanedStepRuns(ctx context.Context) any {
+	return func() {
+		t.l.Debug().Msgf("ticker: polling for orphaned step runs")
+
+		orphanedStepRuns, err := t.repo.Ticker().PollOrphanedStepRuns(ctx)
+
+		if err != nil {
+			t.l.Err(err).Msg("could not poll orphaned step runs")
+		}
+
+		for _, orphanedStepRun := range orphanedStepRuns {
+			stepRun, err := t.repo.StepRun().GetStepRunForEngine(ctx,
+				sqlchelpers.UUIDToStr(orphanedStepRun.TenantId),
+				sqlchelpers.UUIDToStr(orphanedStepRun.ID),
+			)
+
+			if err != nil {
+				t.l.Err(err).Msg("could not get step run")
+			}
+
+			t.mq.AddMessage(ctx,
+				msgqueue.JOB_PROCESSING_QUEUE,
+				tasktypes.StepRunCancelToTask(stepRun, "STUCK_STEP_RUN_DETECTED", false),
+			)
+
+			t.l.Debug().Msgf("ticker: orphaned step run %s", sqlchelpers.UUIDToStr(orphanedStepRun.ID))
+		}
+
+	}
+}
+
 func (t *TickerImpl) runUpdateHeartbeat(ctx context.Context) func() {
 	return func() {
 		t.l.Debug().Msgf("ticker: updating heartbeat")

pkg/repository/prisma/dbsqlc/tickers.sql

@@ -362,3 +362,20 @@ WHERE
 	)
 	AND sr."updatedAt" < CURRENT_TIMESTAMP - INTERVAL '5 seconds'
 ;
+
+
+-- name: PollOrphanedStepRuns :many
+SELECT jr."workflowRunId", sr.*
+FROM "StepRun" sr
+JOIN "JobRun" jr ON jr."id" = sr."jobRunId"
+WHERE NOT EXISTS (
+    SELECT 1
+    FROM "SemaphoreQueueItem" sqi
+    WHERE sqi."stepRunId" = sr."id"
+) AND NOT EXISTS (
+    SELECT 1
+    FROM "TimeoutQueueItem" tqi
+    WHERE tqi."stepRunId" = sr."id"
+)
+AND sr."status" = 'RUNNING'
+AND sr."deletedAt" IS NULL;

pkg/repository/prisma/step_run.go

@@ -534,7 +534,7 @@ func (s *stepRunEngineRepository) ListStepRunsToTimeout(ctx context.Context, ten
 
 	// mark the step runs as cancelling
 	defer func() {
-		_, err = s.queries.BulkMarkStepRunsAsCancelling(ctx, s.pool, stepRunIds)
+		_, err := s.queries.BulkMarkStepRunsAsCancelling(ctx, s.pool, stepRunIds)
 
 		if err != nil {
 			s.l.Err(err).Msg("could not bulk mark step runs as cancelling")

pkg/repository/prisma/ticker.go

@@ -112,3 +112,7 @@ func (t *tickerRepository) PollTenantResourceLimitAlerts(ctx context.Context) ([
 func (t *tickerRepository) PollUnresolvedFailedStepRuns(ctx context.Context) ([]*dbsqlc.PollUnresolvedFailedStepRunsRow, error) {
 	return t.queries.PollUnresolvedFailedStepRuns(ctx, t.pool)
 }
+
+func (t *tickerRepository) PollOrphanedStepRuns(ctx context.Context) ([]*dbsqlc.PollOrphanedStepRunsRow, error) {
+	return t.queries.PollOrphanedStepRuns(ctx, t.pool)
+}

pkg/repository/ticker.go

@@ -51,6 +51,7 @@ type TickerEngineRepository interface {
 
 	PollUnresolvedFailedStepRuns(ctx context.Context) ([]*dbsqlc.PollUnresolvedFailedStepRunsRow, error)
 
+	PollOrphanedStepRuns(ctx context.Context) ([]*dbsqlc.PollOrphanedStepRunsRow, error)
 	// // AddJobRun assigns a job run to a ticker.
 	// AddJobRun(tickerId string, jobRun *db.JobRunModel) (*db.TickerModel, error)