diff --git a/server/cmd/worker-generate/main.go b/server/cmd/worker-generate/main.go index 0bc37e5..95b8162 100644 --- a/server/cmd/worker-generate/main.go +++ b/server/cmd/worker-generate/main.go @@ -54,7 +54,7 @@ func main() { app.GenerationStreams, generationCfg, app.Config().LLM.MaxOutputTokens, - ).WithCache(app.Cache).WithConfigProvider(generationProvider) + ).WithCache(app.Cache).WithLogger(app.Logger).WithConfigProvider(generationProvider) promptRuleSvc := tenantapp.NewPromptRuleGenerationService( app.DB, @@ -82,7 +82,7 @@ func main() { app.GenerationStreams, generationCfg, app.Config().LLM.MaxOutputTokens, - ).WithCache(app.Cache).WithConfigProvider(generationProvider) + ).WithCache(app.Cache).WithLogger(app.Logger).WithConfigProvider(generationProvider) kolWorker := generateworker.NewKolGenerationWorker( app.DB, diff --git a/server/internal/tenant/app/article_imitation_service.go b/server/internal/tenant/app/article_imitation_service.go index 2d07128..590303c 100644 --- a/server/internal/tenant/app/article_imitation_service.go +++ b/server/internal/tenant/app/article_imitation_service.go @@ -9,6 +9,7 @@ import ( "time" "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" "github.com/geo-platform/tenant-api/internal/shared/auth" sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache" @@ -38,6 +39,7 @@ type ArticleImitationService struct { cfg generationRuntimeConfig configProvider runtimeConfigProvider cache sharedcache.Cache + logger *zap.Logger } type GenerateImitationRequest struct { @@ -105,6 +107,11 @@ func (s *ArticleImitationService) WithConfigProvider(provider runtimeConfigProvi return s } +func (s *ArticleImitationService) WithLogger(logger *zap.Logger) *ArticleImitationService { + s.logger = logger + return s +} + func (s *ArticleImitationService) runtimeConfig() generationRuntimeConfig { if s != nil && s.configProvider != nil { if cfg := s.configProvider.Current(); cfg != nil { @@ -460,15 +467,29 @@ func (s *ArticleImitationService) failGeneration(ctx context.Context, job articl articleRepo := repository.NewArticleRepository(s.pool) quotaRepo := repository.NewQuotaRepository(s.pool) - _ = auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ + if err := auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ ID: job.TaskID, TenantID: job.TenantID, Status: "failed", ErrorMessage: &errMsg, CompletedAt: &now, - }) + }); err != nil && s.logger != nil { + s.logger.Warn("imitation generation task failed status update failed", + zap.Error(err), + zap.Int64("task_id", job.TaskID), + zap.Int64("article_id", job.ArticleID), + zap.String("stage", stage), + ) + } if job.ArticleID > 0 { - _ = articleRepo.UpdateArticleGenerateStatus(cleanupCtx, job.ArticleID, job.TenantID, "failed") + if err := articleRepo.UpdateArticleGenerateStatus(cleanupCtx, job.ArticleID, job.TenantID, "failed"); err != nil && s.logger != nil { + s.logger.Warn("imitation generation article failed status update failed", + zap.Error(err), + zap.Int64("task_id", job.TaskID), + zap.Int64("article_id", job.ArticleID), + zap.String("stage", stage), + ) + } _, _ = s.pool.Exec(cleanupCtx, ` UPDATE articles SET wizard_state_json = jsonb_set(COALESCE(wizard_state_json, '{}'::jsonb), '{title}', to_jsonb($1::text), true), diff --git a/server/internal/tenant/app/generation_task_runtime.go b/server/internal/tenant/app/generation_task_runtime.go index a9d348d..abba4f3 100644 --- a/server/internal/tenant/app/generation_task_runtime.go +++ b/server/internal/tenant/app/generation_task_runtime.go @@ -2,6 +2,7 @@ package app import ( "context" + "errors" "fmt" "strings" "time" @@ -61,15 +62,20 @@ func HandleClaimedGenerationTaskFailure( articleRepo := repository.NewArticleRepository(pool) quotaRepo := repository.NewQuotaRepository(pool) - _ = auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ + var failureErr error + if err := auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ ID: task.ID, TenantID: task.TenantID, Status: "failed", ErrorMessage: &errMsg, CompletedAt: &now, - }) + }); err != nil { + failureErr = errors.Join(failureErr, fmt.Errorf("update generation task failed status: %w", err)) + } if task.ArticleID > 0 { - _ = articleRepo.UpdateArticleGenerateStatus(cleanupCtx, task.ArticleID, task.TenantID, "failed") + if err := articleRepo.UpdateArticleGenerateStatus(cleanupCtx, task.ArticleID, task.TenantID, "failed"); err != nil { + failureErr = errors.Join(failureErr, fmt.Errorf("update article failed status: %w", err)) + } } if task.QuotaReservationID > 0 { @@ -119,7 +125,7 @@ func HandleClaimedGenerationTaskFailure( promptRuleService.streams.Fail(task.ArticleID, task.ID, title, errMsg) } - return nil + return failureErr } func (s *TemplateService) ProcessQueuedTask(ctx context.Context, task ClaimedGenerationTask) { diff --git a/server/internal/tenant/app/kol_generation_service.go b/server/internal/tenant/app/kol_generation_service.go index dc049ff..e999134 100644 --- a/server/internal/tenant/app/kol_generation_service.go +++ b/server/internal/tenant/app/kol_generation_service.go @@ -3,6 +3,8 @@ package app import ( "context" "encoding/json" + "errors" + "fmt" "time" "github.com/jackc/pgx/v5/pgxpool" @@ -389,14 +391,17 @@ func HandleKolGenerationTaskFailure( quotaRepo := repository.NewQuotaRepository(pool) usageRepo := repository.NewKolUsageRepository(pool) + var failureErr error if task.ID > 0 { - _ = auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ + if err := auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ ID: task.ID, TenantID: task.TenantID, Status: "failed", ErrorMessage: &errMsg, CompletedAt: &now, - }) + }); err != nil { + failureErr = errors.Join(failureErr, fmt.Errorf("update generation task failed status: %w", err)) + } _ = auditRepo.UpdateTaskRecordStatus(cleanupCtx, repository.TaskRecordInput{ TenantID: task.TenantID, TaskType: kolGenerationTaskRecordType, @@ -409,7 +414,9 @@ func HandleKolGenerationTaskFailure( } if task.ArticleID > 0 { - _ = articleRepo.UpdateArticleGenerateStatus(cleanupCtx, task.ArticleID, task.TenantID, "failed") + if err := articleRepo.UpdateArticleGenerateStatus(cleanupCtx, task.ArticleID, task.TenantID, "failed"); err != nil { + failureErr = errors.Join(failureErr, fmt.Errorf("update article failed status: %w", err)) + } } if task.QuotaReservationID > 0 { @@ -433,5 +440,5 @@ func HandleKolGenerationTaskFailure( } InvalidateArticleCaches(cleanupCtx, cache, task.TenantID, &task.ArticleID) - return nil + return failureErr } diff --git a/server/internal/tenant/app/prompt_generate_service.go b/server/internal/tenant/app/prompt_generate_service.go index 10146c7..e2bbaea 100644 --- a/server/internal/tenant/app/prompt_generate_service.go +++ b/server/internal/tenant/app/prompt_generate_service.go @@ -695,14 +695,28 @@ func (s *PromptRuleGenerationService) failGeneration(ctx context.Context, job pr articleRepo := repository.NewArticleRepository(s.pool) quotaRepo := repository.NewQuotaRepository(s.pool) - _ = auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ + if err := auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ ID: job.TaskID, TenantID: job.TenantID, Status: "failed", ErrorMessage: &errMsg, CompletedAt: &now, - }) - _ = articleRepo.UpdateArticleGenerateStatus(cleanupCtx, job.ArticleID, job.TenantID, "failed") + }); err != nil && s.logger != nil { + s.logger.Warn("prompt rule generation task failed status update failed", + zap.Error(err), + zap.Int64("task_id", job.TaskID), + zap.Int64("article_id", job.ArticleID), + zap.String("stage", stage), + ) + } + if err := articleRepo.UpdateArticleGenerateStatus(cleanupCtx, job.ArticleID, job.TenantID, "failed"); err != nil && s.logger != nil { + s.logger.Warn("prompt rule generation article failed status update failed", + zap.Error(err), + zap.Int64("task_id", job.TaskID), + zap.Int64("article_id", job.ArticleID), + zap.String("stage", stage), + ) + } if failureTitle != "" { _, _ = s.pool.Exec(cleanupCtx, ` UPDATE articles diff --git a/server/internal/tenant/app/template_service.go b/server/internal/tenant/app/template_service.go index 2491e56..13ea82d 100644 --- a/server/internal/tenant/app/template_service.go +++ b/server/internal/tenant/app/template_service.go @@ -10,6 +10,7 @@ import ( "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" "github.com/geo-platform/tenant-api/internal/shared/auth" sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache" @@ -31,6 +32,7 @@ type TemplateService struct { cfg generationRuntimeConfig configProvider runtimeConfigProvider cache sharedcache.Cache + logger *zap.Logger } type generationJob struct { @@ -80,6 +82,11 @@ func (s *TemplateService) WithConfigProvider(provider runtimeConfigProvider) *Te return s } +func (s *TemplateService) WithLogger(logger *zap.Logger) *TemplateService { + s.logger = logger + return s +} + func (s *TemplateService) runtimeConfig() generationRuntimeConfig { if s != nil && s.configProvider != nil { if cfg := s.configProvider.Current(); cfg != nil { @@ -584,14 +591,28 @@ func (s *TemplateService) failGeneration(ctx context.Context, job generationJob, articleRepo := repository.NewArticleRepository(s.pool) quotaRepo := repository.NewQuotaRepository(s.pool) - _ = auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ + if err := auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ ID: job.TaskID, TenantID: job.TenantID, Status: "failed", ErrorMessage: &errMsg, CompletedAt: &now, - }) - _ = articleRepo.UpdateArticleGenerateStatus(cleanupCtx, job.ArticleID, job.TenantID, "failed") + }); err != nil && s.logger != nil { + s.logger.Warn("template generation task failed status update failed", + zap.Error(err), + zap.Int64("task_id", job.TaskID), + zap.Int64("article_id", job.ArticleID), + zap.String("stage", stage), + ) + } + if err := articleRepo.UpdateArticleGenerateStatus(cleanupCtx, job.ArticleID, job.TenantID, "failed"); err != nil && s.logger != nil { + s.logger.Warn("template generation article failed status update failed", + zap.Error(err), + zap.Int64("task_id", job.TaskID), + zap.Int64("article_id", job.ArticleID), + zap.String("stage", stage), + ) + } balance, balanceErr := quotaRepo.GetCurrentBalance(cleanupCtx, job.TenantID, "article_generation") if balanceErr == nil { diff --git a/server/internal/tenant/repository/generated/audit.sql.go b/server/internal/tenant/repository/generated/audit.sql.go index a2b88d1..f57a039 100644 --- a/server/internal/tenant/repository/generated/audit.sql.go +++ b/server/internal/tenant/repository/generated/audit.sql.go @@ -124,7 +124,7 @@ func (q *Queries) CreateTaskRecord(ctx context.Context, arg CreateTaskRecordPara } const updateGenerationTaskStatus = `-- name: UpdateGenerationTaskStatus :exec -UPDATE generation_tasks SET status = $1, error_message = $2, +UPDATE generation_tasks SET status = $1::text, error_message = $2, started_at = $3, completed_at = $4, lease_token = CASE WHEN $1::text IN ('completed', 'failed') THEN NULL ELSE lease_token END, diff --git a/server/internal/tenant/repository/queries/audit.sql b/server/internal/tenant/repository/queries/audit.sql index 9156dc0..938ae59 100644 --- a/server/internal/tenant/repository/queries/audit.sql +++ b/server/internal/tenant/repository/queries/audit.sql @@ -39,7 +39,7 @@ WHERE tenant_id = sqlc.arg(tenant_id) AND resource_id = sqlc.arg(resource_id); -- name: UpdateGenerationTaskStatus :exec -UPDATE generation_tasks SET status = sqlc.arg(status), error_message = sqlc.arg(error_message), +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), lease_token = CASE WHEN sqlc.arg(status)::text IN ('completed', 'failed') THEN NULL ELSE lease_token END, diff --git a/server/internal/tenant/transport/article_handler.go b/server/internal/tenant/transport/article_handler.go index 293266a..8ea839c 100644 --- a/server/internal/tenant/transport/article_handler.go +++ b/server/internal/tenant/transport/article_handler.go @@ -66,7 +66,7 @@ func NewArticleHandler(a *bootstrap.App) *ArticleHandler { a.GenerationStreams, a.Config().Generation, a.Config().LLM.MaxOutputTokens, - ).WithCache(a.Cache).WithConfigProvider(a.ConfigStore), + ).WithCache(a.Cache).WithLogger(a.Logger).WithConfigProvider(a.ConfigStore), streams: a.GenerationStreams, app: a, assets: NewAssetHandler(a), diff --git a/server/internal/tenant/transport/template_handler.go b/server/internal/tenant/transport/template_handler.go index 319317f..0b6fe99 100644 --- a/server/internal/tenant/transport/template_handler.go +++ b/server/internal/tenant/transport/template_handler.go @@ -42,7 +42,7 @@ func NewTemplateHandler(a *bootstrap.App) *TemplateHandler { a.GenerationStreams, a.Config().Generation, a.Config().LLM.MaxOutputTokens, - ).WithCache(a.Cache).WithConfigProvider(a.ConfigStore), + ).WithCache(a.Cache).WithLogger(a.Logger).WithConfigProvider(a.ConfigStore), } } diff --git a/server/internal/worker/generate/article_generation_recovery.go b/server/internal/worker/generate/article_generation_recovery.go index 1545043..a9ede9e 100644 --- a/server/internal/worker/generate/article_generation_recovery.go +++ b/server/internal/worker/generate/article_generation_recovery.go @@ -472,11 +472,44 @@ func (w *ArticleGenerationWorker) requeueExpiredRunningGenerationTask(ctx contex }() var ( - tenantID int64 - articleID int64 - taskType string + tenantID int64 + articleID int64 + taskType string + shouldRequeue bool ) if err := tx.QueryRow(ctx, ` + WITH existing AS ( + SELECT gt.id, + gt.tenant_id, + gt.article_id, + gt.task_type, + a.generate_status + FROM generation_tasks gt + JOIN articles a ON a.id = gt.article_id + AND a.tenant_id = gt.tenant_id + WHERE gt.id = $1 + AND gt.status = 'running' + AND ( + gt.lease_expires_at IS NULL + OR gt.lease_expires_at < NOW() + ) + FOR UPDATE OF gt + ), + finalized_failed_article AS ( + UPDATE generation_tasks gt + SET status = 'failed', + error_message = COALESCE(NULLIF(gt.error_message, ''), '文章已标记生成失败，停止后台恢复重试 [article_failed_stop_retry]'), + completed_at = COALESCE(gt.completed_at, NOW()), + lease_token = NULL, + lease_owner = NULL, + lease_expires_at = NULL, + updated_at = NOW() + FROM existing e + WHERE gt.id = e.id + AND e.generate_status = 'failed' + RETURNING e.tenant_id, e.article_id, e.task_type, false AS should_requeue + ), + requeued AS ( UPDATE generation_tasks SET status = 'queued', error_message = NULL, @@ -484,14 +517,17 @@ func (w *ArticleGenerationWorker) requeueExpiredRunningGenerationTask(ctx contex lease_owner = NULL, lease_expires_at = NULL, updated_at = NOW() - WHERE id = $1 - AND status = 'running' - AND ( - lease_expires_at IS NULL - OR lease_expires_at < NOW() - ) - RETURNING tenant_id, article_id, task_type - `, item.ID).Scan(&tenantID, &articleID, &taskType); err != nil { + FROM existing e + WHERE generation_tasks.id = e.id + AND e.generate_status <> 'failed' + RETURNING e.tenant_id, e.article_id, e.task_type, true AS should_requeue + ) + SELECT tenant_id, article_id, task_type, should_requeue + FROM finalized_failed_article + UNION ALL + SELECT tenant_id, article_id, task_type, should_requeue + FROM requeued + `, item.ID).Scan(&tenantID, &articleID, &taskType, &shouldRequeue); err != nil { if err == pgx.ErrNoRows { return false, nil } @@ -501,6 +537,15 @@ func (w *ArticleGenerationWorker) requeueExpiredRunningGenerationTask(ctx contex if err := tx.Commit(ctx); err != nil { return false, err } + if !shouldRequeue { + if w.logger != nil { + w.logger.Info("article generation expired task finalized because article is already failed", + zap.Int64("task_id", item.ID), + zap.Int64("article_id", articleID), + ) + } + return true, nil + } if err := w.publishGenerationTask(ctx, item.ID, taskType, tenantID, articleID); err != nil { return false, err } diff --git a/server/internal/worker/generate/article_generation_worker.go b/server/internal/worker/generate/article_generation_worker.go index 94915d6..a08afd8 100644 --- a/server/internal/worker/generate/article_generation_worker.go +++ b/server/internal/worker/generate/article_generation_worker.go @@ -245,13 +245,23 @@ func articleGenerationProcessTimeout(cfg config.GenerationConfig) time.Duration func (w *ArticleGenerationWorker) handleTaskFailure(ctx context.Context, task tenantapp.ClaimedGenerationTask, stage string, taskErr error) { task.TaskType = strings.TrimSpace(task.TaskType) + var err error switch { case task.TaskType == kolGenerationTaskType && w.kolWorker != nil: w.kolWorker.HandleTaskFailure(ctx, task, stage, taskErr) case task.TaskType == "imitation" && w.imitationService != nil: w.imitationService.HandleTaskFailure(ctx, task, stage, taskErr) default: - _ = tenantapp.HandleClaimedGenerationTaskFailure(ctx, w.pool, w.templateService, w.promptRuleService, task, taskErr) + err = tenantapp.HandleClaimedGenerationTaskFailure(ctx, w.pool, w.templateService, w.promptRuleService, task, taskErr) + } + if err != nil && w.logger != nil { + w.logger.Warn("article generation task failure handling failed", + zap.Error(err), + zap.Int64("task_id", task.ID), + zap.Int64("article_id", task.ArticleID), + zap.String("task_type", task.TaskType), + zap.String("stage", stage), + ) } } diff --git a/server/migrations/20260505113000_generation_task_status_text.down.sql b/server/migrations/20260505113000_generation_task_status_text.down.sql new file mode 100644 index 0000000..597b302 --- /dev/null +++ b/server/migrations/20260505113000_generation_task_status_text.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE generation_tasks + ALTER COLUMN status TYPE VARCHAR(20); diff --git a/server/migrations/20260505113000_generation_task_status_text.up.sql b/server/migrations/20260505113000_generation_task_status_text.up.sql new file mode 100644 index 0000000..8b9b132 --- /dev/null +++ b/server/migrations/20260505113000_generation_task_status_text.up.sql @@ -0,0 +1,2 @@ +ALTER TABLE generation_tasks + ALTER COLUMN status TYPE TEXT;