From cbbed4b42c88e760a12c75776cfe71f4b8fa629e Mon Sep 17 00:00:00 2001 From: liangxu Date: Mon, 11 May 2026 11:11:31 +0800 Subject: [PATCH] fix(generation): repair articles stuck in generating when latest task is terminal The state-consistency worker now detects articles still flagged as generating whose latest generation_task is already failed or completed (with a version) and aligns the article status to the terminal task, invalidating article caches on the way out. Adds supporting partial indexes and switches UpdateGenerationTaskStatus to COALESCE started_at and completed_at so partial status updates don't wipe the timestamps. --- server/cmd/scheduler/main.go | 3 +- .../tenant/repository/generated/audit.sql.go | 4 +- .../tenant/repository/queries/audit.sql | 4 +- .../article_generation_state_check.go | 155 +++++++++++++++++- .../article_generation_state_check_worker.go | 9 +- ...tion_task_terminal_repair_indexes.down.sql | 3 + ...ration_task_terminal_repair_indexes.up.sql | 13 ++ 7 files changed, 183 insertions(+), 8 deletions(-) create mode 100644 server/migrations/20260511110000_generation_task_terminal_repair_indexes.down.sql create mode 100644 server/migrations/20260511110000_generation_task_terminal_repair_indexes.up.sql diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index 8f648b7..64efcb1 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -93,7 +93,8 @@ func main() { monitoringReceivedInspectionWorker := internalscheduler.NewMonitoringReceivedInspectionWorker(app.MonitoringDB, app.Logger) knowledgeDeletedCleanupWorker := internalscheduler.NewKnowledgeDeletedCleanupWorker(knowledgeSvc) generationStateCheckWorker := generateworker.NewGenerationTaskStateCheckWorker(app.DB, app.Logger, app.Config().Generation). - WithConfigProvider(app.ConfigStore) + WithConfigProvider(app.ConfigStore). + WithCache(app.Cache) startSchedulerWorker(ctx, &workerWG, "schedule_dispatch", app.Logger.Sugar(), scheduleDispatchWorker.Run) startSchedulerWorker(ctx, &workerWG, "monitoring_daily_task", app.Logger.Sugar(), monitoringDailyTaskWorker.Run) diff --git a/server/internal/tenant/repository/generated/audit.sql.go b/server/internal/tenant/repository/generated/audit.sql.go index f57a039..f5484b2 100644 --- a/server/internal/tenant/repository/generated/audit.sql.go +++ b/server/internal/tenant/repository/generated/audit.sql.go @@ -125,8 +125,8 @@ func (q *Queries) CreateTaskRecord(ctx context.Context, arg CreateTaskRecordPara const updateGenerationTaskStatus = `-- name: UpdateGenerationTaskStatus :exec UPDATE generation_tasks SET status = $1::text, error_message = $2, - started_at = $3, - completed_at = $4, + started_at = COALESCE($3, started_at), + completed_at = COALESCE($4, completed_at), lease_token = CASE WHEN $1::text IN ('completed', 'failed') THEN NULL ELSE lease_token END, lease_owner = CASE WHEN $1::text IN ('completed', 'failed') THEN NULL ELSE lease_owner END, lease_expires_at = CASE WHEN $1::text IN ('completed', 'failed') THEN NULL ELSE lease_expires_at END, diff --git a/server/internal/tenant/repository/queries/audit.sql b/server/internal/tenant/repository/queries/audit.sql index 938ae59..097ad87 100644 --- a/server/internal/tenant/repository/queries/audit.sql +++ b/server/internal/tenant/repository/queries/audit.sql @@ -40,8 +40,8 @@ WHERE tenant_id = sqlc.arg(tenant_id) -- name: UpdateGenerationTaskStatus :exec UPDATE generation_tasks SET status = sqlc.arg(status)::text, error_message = sqlc.arg(error_message), - started_at = sqlc.arg(started_at), - completed_at = sqlc.arg(completed_at), + started_at = COALESCE(sqlc.arg(started_at), started_at), + completed_at = COALESCE(sqlc.arg(completed_at), completed_at), lease_token = CASE WHEN sqlc.arg(status)::text IN ('completed', 'failed') THEN NULL ELSE lease_token END, lease_owner = CASE WHEN sqlc.arg(status)::text IN ('completed', 'failed') THEN NULL ELSE lease_owner END, lease_expires_at = CASE WHEN sqlc.arg(status)::text IN ('completed', 'failed') THEN NULL ELSE lease_expires_at END, diff --git a/server/internal/worker/generate/article_generation_state_check.go b/server/internal/worker/generate/article_generation_state_check.go index f782de7..04a782b 100644 --- a/server/internal/worker/generate/article_generation_state_check.go +++ b/server/internal/worker/generate/article_generation_state_check.go @@ -5,6 +5,7 @@ import ( "database/sql" "time" + sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache" "github.com/jackc/pgx/v5" "go.uber.org/zap" @@ -17,6 +18,7 @@ const ( generationTaskAnomalyCompletedTaskArticleMissing = "completed_task_article_not_completed" generationTaskAnomalyCompletedTaskMissingVersion = "completed_task_missing_version" generationTaskAnomalyExpiredRunningLease = "expired_running_lease" + generationTaskAnomalyTerminalTaskArticleActive = "terminal_task_article_generating" ) type generationTaskStateAnomaly struct { @@ -34,7 +36,7 @@ type generationTaskStateAnomaly struct { Severity string } -func CheckGenerationTaskStateConsistency(ctx context.Context, pool generationTaskStateDB, logger *zap.Logger, cfg generationTaskStateCheckConfig) (int, error) { +func CheckGenerationTaskStateConsistency(ctx context.Context, pool generationTaskStateDB, logger *zap.Logger, cache sharedcache.Cache, cfg generationTaskStateCheckConfig) (int, error) { if pool == nil || cfg.BatchSize <= 0 { return 0, nil } @@ -132,6 +134,150 @@ func CheckGenerationTaskStateConsistency(ctx context.Context, pool generationTas if err := rows.Err(); err != nil { return count, err } + repaired, err := repairGenerationTaskArticleState(ctx, pool, logger, cache, cfg, lookback) + if err != nil { + return count, err + } + count += repaired + return count, nil +} + +func repairGenerationTaskArticleState(ctx context.Context, pool generationTaskStateDB, logger *zap.Logger, cache sharedcache.Cache, cfg generationTaskStateCheckConfig, lookback time.Time) (int, error) { + if pool == nil || cfg.BatchSize <= 0 { + return 0, nil + } + rows, err := pool.Query(ctx, ` + WITH recent_terminal 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, + gt.error_message, + gt.attempt_count, + gt.lease_owner, + gt.created_at, + gt.updated_at + FROM generation_tasks gt + WHERE gt.article_id IS NOT NULL + AND gt.status IN ('failed', 'completed') + AND gt.updated_at >= $1 + ORDER BY gt.updated_at DESC, gt.id DESC + LIMIT ($2::int * 5) + ), + latest_terminal AS ( + SELECT DISTINCT ON (rt.article_id) + rt.id, + rt.tenant_id, + rt.operator_id, + rt.article_id, + rt.quota_reservation_id, + rt.task_type, + rt.task_status, + rt.error_message, + rt.attempt_count, + rt.lease_owner, + rt.created_at, + rt.updated_at + FROM recent_terminal rt + ORDER BY rt.article_id, rt.created_at DESC, rt.id DESC + ), + candidates AS ( + SELECT + lt.id, + lt.tenant_id, + lt.operator_id, + lt.article_id, + lt.quota_reservation_id, + lt.task_type, + lt.task_status, + a.generate_status AS article_status, + lt.attempt_count, + lt.lease_owner + FROM latest_terminal lt + JOIN articles a + ON a.id = lt.article_id + AND a.tenant_id = lt.tenant_id + AND a.deleted_at IS NULL + WHERE a.generate_status = 'generating' + AND ( + lt.task_status = 'failed' + OR (lt.task_status = 'completed' AND a.current_version_id IS NOT NULL) + ) + AND NOT EXISTS ( + SELECT 1 + FROM generation_tasks newer + WHERE newer.article_id = lt.article_id + AND newer.tenant_id = lt.tenant_id + AND (newer.created_at > lt.created_at OR (newer.created_at = lt.created_at AND newer.id > lt.id)) + ) + ORDER BY lt.updated_at DESC, lt.id DESC + LIMIT $2 + ), + updated AS ( + UPDATE articles a + SET generate_status = c.task_status, + updated_at = NOW() + FROM candidates c + WHERE a.id = c.article_id + AND a.tenant_id = c.tenant_id + AND a.generate_status = 'generating' + RETURNING + c.id, + c.tenant_id, + c.operator_id, + c.article_id, + c.quota_reservation_id, + c.task_type, + c.task_status, + c.article_status, + c.attempt_count, + c.lease_owner, + 'terminal_task_article_generating'::text AS anomaly_type + ) + SELECT id, tenant_id, operator_id, article_id, quota_reservation_id, task_type, task_status, article_status, attempt_count, lease_owner, anomaly_type + FROM updated + `, 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 + } + item.AnomalyType = generationTaskAnomalyTerminalTaskArticleActive + item.Severity = generationTaskAnomalySeverity(item.AnomalyType) + count++ + logCtx := tenantapp.GenerationTaskLogContext{ + TaskID: item.TaskID, + TenantID: item.TenantID, + OperatorID: item.OperatorID, + ArticleID: item.ArticleID, + ReservationID: item.ReservationID, + TaskType: item.TaskType, + Stage: "state_repair", + AttemptCount: item.AttemptCount, + LeaseOwner: item.LeaseOwner, + StatusBefore: item.ArticleStatus, + StatusAfter: item.TaskStatus, + Result: tenantapp.GenerationTaskResultSuccess, + AnomalyType: item.AnomalyType, + Severity: item.Severity, + } + tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventStateAnomaly, logCtx) + tenantapp.LogGenerationTaskWarn(logger, "article generation state repaired from latest terminal task", logCtx) + stateRepairCacheInvalidator(ctx, cache, item.TenantID, &item.ArticleID) + } + if err := rows.Err(); err != nil { + return count, err + } return count, nil } @@ -143,6 +289,10 @@ type generationTaskStateDB interface { Query(context.Context, string, ...any) (pgx.Rows, error) } +var stateRepairCacheInvalidator = func(ctx context.Context, c sharedcache.Cache, tenantID int64, articleID *int64) { + tenantapp.InvalidateArticleCaches(ctx, c, tenantID, articleID) +} + func scanGenerationTaskStateAnomaly(row generationTaskStateAnomalyRow) (generationTaskStateAnomaly, error) { var ( item generationTaskStateAnomaly @@ -180,7 +330,8 @@ func generationTaskAnomalySeverity(anomalyType string) string { switch anomalyType { case generationTaskAnomalyCompletedTaskArticleMissing, generationTaskAnomalyCompletedTaskMissingVersion, - generationTaskAnomalyArticleFailedTaskActive: + generationTaskAnomalyArticleFailedTaskActive, + generationTaskAnomalyTerminalTaskArticleActive: return "critical" case generationTaskAnomalyArticleCompletedTaskActive, generationTaskAnomalyExpiredRunningLease: diff --git a/server/internal/worker/generate/article_generation_state_check_worker.go b/server/internal/worker/generate/article_generation_state_check_worker.go index 18b67a5..9c9341d 100644 --- a/server/internal/worker/generate/article_generation_state_check_worker.go +++ b/server/internal/worker/generate/article_generation_state_check_worker.go @@ -4,6 +4,7 @@ import ( "context" "time" + sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache" "github.com/jackc/pgx/v5/pgxpool" "go.uber.org/zap" @@ -16,6 +17,7 @@ type GenerationTaskStateCheckWorker struct { logger *zap.Logger cfg config.GenerationConfig provider config.Provider + cache sharedcache.Cache } func NewGenerationTaskStateCheckWorker(pool *pgxpool.Pool, logger *zap.Logger, cfg config.GenerationConfig) *GenerationTaskStateCheckWorker { @@ -31,6 +33,11 @@ func (w *GenerationTaskStateCheckWorker) WithConfigProvider(provider config.Prov return w } +func (w *GenerationTaskStateCheckWorker) WithCache(cache sharedcache.Cache) *GenerationTaskStateCheckWorker { + w.cache = cache + return w +} + func (w *GenerationTaskStateCheckWorker) Start(ctx context.Context) { go w.Run(ctx) } @@ -61,7 +68,7 @@ func (w *GenerationTaskStateCheckWorker) run(ctx context.Context) { func (w *GenerationTaskStateCheckWorker) runOnce(parent context.Context) { cfg := w.runtimeConfig() startedAt := time.Now() - count, err := CheckGenerationTaskStateConsistency(parent, w.pool, w.logger, cfg) + count, err := CheckGenerationTaskStateConsistency(parent, w.pool, w.logger, w.cache, cfg) duration := time.Since(startedAt) if err != nil { logCtx := tenantapp.GenerationTaskLogContext{ diff --git a/server/migrations/20260511110000_generation_task_terminal_repair_indexes.down.sql b/server/migrations/20260511110000_generation_task_terminal_repair_indexes.down.sql new file mode 100644 index 0000000..77ce0fc --- /dev/null +++ b/server/migrations/20260511110000_generation_task_terminal_repair_indexes.down.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS idx_articles_generating_state_repair; +DROP INDEX IF EXISTS idx_generation_tasks_article_latest_state_repair; +DROP INDEX IF EXISTS idx_generation_tasks_recent_terminal_state_repair; diff --git a/server/migrations/20260511110000_generation_task_terminal_repair_indexes.up.sql b/server/migrations/20260511110000_generation_task_terminal_repair_indexes.up.sql new file mode 100644 index 0000000..38cc0f9 --- /dev/null +++ b/server/migrations/20260511110000_generation_task_terminal_repair_indexes.up.sql @@ -0,0 +1,13 @@ +CREATE INDEX IF NOT EXISTS idx_generation_tasks_recent_terminal_state_repair + ON generation_tasks(updated_at DESC, id DESC) + WHERE article_id IS NOT NULL + AND status IN ('failed', 'completed'); + +CREATE INDEX IF NOT EXISTS idx_generation_tasks_article_latest_state_repair + ON generation_tasks(article_id, tenant_id, created_at DESC, id DESC) + WHERE article_id IS NOT NULL; + +CREATE INDEX IF NOT EXISTS idx_articles_generating_state_repair + ON articles(id, tenant_id) + WHERE generate_status = 'generating' + AND deleted_at IS NULL;