package generate import ( "context" "database/sql" "time" "github.com/jackc/pgx/v5" "go.uber.org/zap" tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app" ) const ( generationTaskAnomalyArticleFailedTaskActive = "article_failed_task_active" generationTaskAnomalyArticleCompletedTaskActive = "article_completed_task_active" generationTaskAnomalyCompletedTaskArticleMissing = "completed_task_article_not_completed" generationTaskAnomalyCompletedTaskMissingVersion = "completed_task_missing_version" generationTaskAnomalyExpiredRunningLease = "expired_running_lease" ) type generationTaskStateAnomaly struct { TaskID int64 TenantID int64 OperatorID int64 ArticleID int64 ReservationID int64 TaskType string TaskStatus string ArticleStatus string AttemptCount int LeaseOwner string AnomalyType string Severity string } func CheckGenerationTaskStateConsistency(ctx context.Context, pool generationTaskStateDB, logger *zap.Logger, cfg generationTaskStateCheckConfig) (int, error) { if pool == nil || cfg.BatchSize <= 0 { return 0, nil } if cfg.Timeout > 0 { var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, cfg.Timeout) defer cancel() } lookback := time.Now().UTC().Add(-cfg.Lookback) rows, err := pool.Query(ctx, ` WITH candidates AS ( SELECT gt.id, gt.tenant_id, gt.operator_id, gt.article_id, gt.quota_reservation_id, gt.task_type, gt.status AS task_status, a.generate_status AS article_status, gt.attempt_count, gt.lease_owner, CASE WHEN a.generate_status = 'failed' AND gt.status IN ('queued', 'running') THEN 'article_failed_task_active' WHEN a.generate_status = 'completed' AND gt.status IN ('queued', 'running') THEN 'article_completed_task_active' WHEN gt.status = 'completed' AND COALESCE(a.generate_status, '') <> 'completed' THEN 'completed_task_article_not_completed' WHEN gt.status = 'completed' AND (a.current_version_id IS NULL OR av.id IS NULL) THEN 'completed_task_missing_version' WHEN gt.status = 'running' AND (gt.lease_expires_at IS NULL OR gt.lease_expires_at < NOW()) THEN 'expired_running_lease' ELSE '' END AS anomaly_type FROM generation_tasks gt LEFT JOIN articles a ON a.id = gt.article_id AND a.tenant_id = gt.tenant_id AND a.deleted_at IS NULL LEFT JOIN article_versions av ON av.id = a.current_version_id WHERE gt.article_id IS NOT NULL AND gt.created_at >= $1 AND ( gt.status IN ('queued', 'running') OR gt.updated_at >= $1 ) AND ( (a.generate_status IN ('failed', 'completed') AND gt.status IN ('queued', 'running')) OR (gt.status = 'completed' AND COALESCE(a.generate_status, '') <> 'completed') OR (gt.status = 'completed' AND (a.current_version_id IS NULL OR av.id IS NULL)) OR (gt.status = 'running' AND (gt.lease_expires_at IS NULL OR gt.lease_expires_at < NOW())) ) ORDER BY gt.updated_at DESC, gt.id DESC LIMIT $2 ) SELECT id, tenant_id, operator_id, article_id, quota_reservation_id, task_type, task_status, article_status, attempt_count, lease_owner, anomaly_type FROM candidates WHERE anomaly_type <> '' `, lookback, cfg.BatchSize) if err != nil { return 0, err } defer rows.Close() count := 0 for rows.Next() { item, err := scanGenerationTaskStateAnomaly(rows) if err != nil { return count, err } if item.AnomalyType == "" { continue } count++ logCtx := tenantapp.GenerationTaskLogContext{ TaskID: item.TaskID, TenantID: item.TenantID, OperatorID: item.OperatorID, ArticleID: item.ArticleID, ReservationID: item.ReservationID, TaskType: item.TaskType, Stage: "state_check", AttemptCount: item.AttemptCount, LeaseOwner: item.LeaseOwner, StatusBefore: item.TaskStatus, StatusAfter: item.ArticleStatus, Result: tenantapp.GenerationTaskResultFailure, AnomalyType: item.AnomalyType, Severity: item.Severity, } tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventStateAnomaly, logCtx) tenantapp.LogGenerationTaskError(logger, "article generation state consistency anomaly detected", logCtx, zap.String("task_status", item.TaskStatus), zap.String("article_status", item.ArticleStatus), ) } if err := rows.Err(); err != nil { return count, err } return count, nil } type generationTaskStateAnomalyRow interface { Scan(dest ...any) error } type generationTaskStateDB interface { Query(context.Context, string, ...any) (pgx.Rows, error) } func scanGenerationTaskStateAnomaly(row generationTaskStateAnomalyRow) (generationTaskStateAnomaly, error) { var ( item generationTaskStateAnomaly operatorID sql.NullInt64 articleID sql.NullInt64 reservationID sql.NullInt64 articleStatus sql.NullString leaseOwner sql.NullString ) if err := row.Scan( &item.TaskID, &item.TenantID, &operatorID, &articleID, &reservationID, &item.TaskType, &item.TaskStatus, &articleStatus, &item.AttemptCount, &leaseOwner, &item.AnomalyType, ); err != nil { return generationTaskStateAnomaly{}, err } item.OperatorID = operatorID.Int64 item.ArticleID = articleID.Int64 item.ReservationID = reservationID.Int64 item.ArticleStatus = articleStatus.String item.LeaseOwner = leaseOwner.String item.Severity = generationTaskAnomalySeverity(item.AnomalyType) return item, nil } func generationTaskAnomalySeverity(anomalyType string) string { switch anomalyType { case generationTaskAnomalyCompletedTaskArticleMissing, generationTaskAnomalyCompletedTaskMissingVersion, generationTaskAnomalyArticleFailedTaskActive: return "critical" case generationTaskAnomalyArticleCompletedTaskActive, generationTaskAnomalyExpiredRunningLease: return "warning" default: return "warning" } }