package usecase import ( "context" "haixun-backend/internal/library/clock" "haixun-backend/internal/model/job/domain/enum" domusecase "haixun-backend/internal/model/job/domain/usecase" ) func (u *jobUseCase) RunMaintenance(ctx context.Context) (domusecase.MaintenanceResult, error) { result := domusecase.MaintenanceResult{} pending, err := u.runs.FindPendingDue(ctx, clock.NowUnixNano(), 50) if err != nil { return result, err } for _, run := range pending { if _, err := u.enqueueRun(ctx, run); err == nil { result.EnqueuedPending++ } } now := clock.NowUnixNano() cancelRuns, err := u.runs.FindCancelRequestedBefore(ctx, now, 50) if err != nil { return result, err } for _, run := range cancelRuns { template, err := u.templates.FindByType(ctx, run.TemplateType) if err != nil { continue } grace := clock.SecondsToNanos(template.CancelPolicy.GraceSeconds) if run.CancelRequestedAt == nil || now-*run.CancelRequestedAt < grace { continue } jobID := run.ID.Hex() fromStatus := string(run.Status) run.Status = enum.RunStatusCancelled run.CompletedAt = &now run.Progress.Summary = "cancelled after grace timeout" updated, err := u.runs.Update(ctx, run) if err != nil { continue } _ = u.queue.ReleaseLock(ctx, jobID, run.LockedBy) _ = u.queue.ClearCancelSignal(ctx, jobID) u.releaseDedupe(ctx, updated) _ = u.appendEvent(ctx, jobID, "status_changed", fromStatus, string(enum.RunStatusCancelled), "cancel grace timeout", nil) result.ReapedCancelGrace++ } stuckRuns, err := u.runs.FindQueuedWithLock(ctx, 50) if err != nil { return result, err } for _, run := range stuckRuns { jobID := run.ID.Hex() if run.StartedAt != nil && run.LockedUntil != nil && *run.LockedUntil > now { fromStatus := string(run.Status) patch := *run patch.Status = enum.RunStatusRunning if patch.Progress.Summary == "" || patch.Progress.Summary == "waiting for worker" { patch.Progress.Summary = "worker claimed job (status repaired)" } updated, err := u.runs.UpdateIfStatus(ctx, &patch, []enum.RunStatus{ enum.RunStatusPending, enum.RunStatusQueued, }) if err != nil { continue } _ = u.appendEvent(ctx, jobID, "status_changed", fromStatus, string(enum.RunStatusRunning), "repaired queued job with active worker lock", nil) result.RepairedStuckClaims++ _ = updated continue } fromStatus := string(run.Status) workerID := run.LockedBy patch := *run patch.LockedBy = "" patch.LockedUntil = nil patch.StartedAt = nil patch.Progress.Summary = "waiting for worker" patch.Progress.Percentage = 0 if _, err := u.runs.UpdateIfStatus(ctx, &patch, []enum.RunStatus{ enum.RunStatusPending, enum.RunStatusQueued, }); err != nil { continue } _ = u.queue.ReleaseLock(ctx, jobID, workerID) if err := u.queue.Enqueue(ctx, patch.WorkerType, jobID); err != nil { continue } _ = u.appendEvent(ctx, jobID, "status_changed", fromStatus, string(enum.RunStatusQueued), "requeued orphaned lock", nil) result.ReapedOrphanedLocks++ } expiredRuns, err := u.runs.FindRunningTimedOut(ctx, now, 50) if err != nil { return result, err } for _, run := range expiredRuns { jobID := run.ID.Hex() fromStatus := string(run.Status) run.Status = enum.RunStatusExpired run.CompletedAt = &now run.Progress.Summary = "job lock expired" updated, err := u.runs.Update(ctx, run) if err != nil { continue } _ = u.queue.ReleaseLock(ctx, jobID, run.LockedBy) _ = u.queue.ClearCancelSignal(ctx, jobID) u.releaseDedupe(ctx, updated) _ = u.appendEvent(ctx, jobID, "status_changed", fromStatus, string(enum.RunStatusExpired), "job lock expired", nil) result.ReapedExpiredLocks++ } return result, nil }