diff --git a/internal/service/ingestion_task_service.go b/internal/service/ingestion_task_service.go index 1f60bde712..201957a5da 100644 --- a/internal/service/ingestion_task_service.go +++ b/internal/service/ingestion_task_service.go @@ -229,7 +229,16 @@ func (s *IngestionTaskService) StartRunning(ctx context.Context, taskID string) } return task, nil case common.STOPPING: - return s.transition(ctx, taskID, common.STOPPED) + task, err = s.transition(ctx, taskID, common.STOPPED) + if err != nil { + return nil, err + } + // The stop is finalized here without a worker (e.g. MQ redelivery of + // a task that was nacked before execution), so the Redis cancel flag + // that RequestStop set would otherwise leak until TTL and cancel the + // next run of this task at the worker's pre-start check. + clearCancelFlag(ctx, taskID) + return task, nil case common.RUNNING, common.COMPLETED, common.STOPPED, common.FAILED: return task, nil default: @@ -387,6 +396,11 @@ func (s *IngestionTaskService) CreateAndEnqueue(ctx context.Context, task *entit if err != nil { return nil, err } + // The previous run is terminal, so any leftover Redis cancel flag + // is stale: a genuine cancel of the new run can only come through + // RequestStop once the task is RUNNING again. Clear it so the + // re-queued task is not cancelled at the worker's pre-start check. + clearCancelFlag(ctx, existing.ID) if err = s.enqueueTask(existing.ID); err != nil { if rollbackErr := s.rollbackRetriedTask(ctx, existing.ID, originalStatus); rollbackErr != nil { return nil, fmt.Errorf("enqueue task %s: %w (rollback failed: %v)", existing.ID, err, rollbackErr) @@ -427,6 +441,15 @@ func (s *IngestionTaskService) rollbackCreatedTask(ctx context.Context, taskID s return err } +// clearCancelFlag removes the Redis cancel marker ({task_id}-cancel) that +// RequestStop sets for a RUNNING task. No-op when Redis is unavailable — +// the DB STOPPING status remains the fallback cancel signal. +func clearCancelFlag(ctx context.Context, taskID string) { + if rc := redis2.Get(); rc != nil { + rc.Delete(ctx, fmt.Sprintf("%s-cancel", taskID)) + } +} + func (s *IngestionTaskService) enqueueTask(taskID string) error { taskMessage := common.TaskMessage{ TaskID: taskID, diff --git a/internal/service/ingestion_task_service_test.go b/internal/service/ingestion_task_service_test.go index aaee9a481e..53c8d1f217 100644 --- a/internal/service/ingestion_task_service_test.go +++ b/internal/service/ingestion_task_service_test.go @@ -319,6 +319,30 @@ func TestStartRunningLeavesTerminalDocumentUntouched(t *testing.T) { } } +// TestStartRunningFinalizesStoppingTask locks in the redelivery path: a +// STOPPING task (cancelled after being nacked, before any worker ran it) is +// moved to STOPPED by StartRunning, so the task reaches a terminal state and +// a later re-parse can transition it back to CREATED. +func TestStartRunningFinalizesStoppingTask(t *testing.T) { + db := setupServiceTestDB(t) + pushServiceDB(t, db) + insertTestIngestionTask(t, "task-1", "user-1", "doc-1", "kb-1") + if err := db.Model(&entity.IngestionTask{}).Where("id = ?", "task-1"). + Update("status", common.STOPPING).Error; err != nil { + t.Fatalf("set STOPPING: %v", err) + } + + svc := NewIngestionTaskService() + ctx := t.Context() + task, err := svc.StartRunning(ctx, "task-1") + if err != nil { + t.Fatalf("StartRunning failed: %v", err) + } + if task.Status != common.STOPPED { + t.Fatalf("status = %q, want %q", task.Status, common.STOPPED) + } +} + func TestIngestionTaskServiceRequestStopTransitionsCreatedTaskToStopped(t *testing.T) { db := setupServiceTestDB(t) pushServiceDB(t, db)