diff --git a/deploy/monitoring/article-generation-alerts.yml b/deploy/monitoring/article-generation-alerts.yml new file mode 100644 index 0000000..0c692e1 --- /dev/null +++ b/deploy/monitoring/article-generation-alerts.yml @@ -0,0 +1,56 @@ +groups: + - name: article-generation-reliability + rules: + - alert: ArticleGenerationCleanupFailure + expr: increase(article_generation_task_cleanup_failures_total[5m]) > 0 + labels: + severity: critical + component: article_generation + annotations: + summary: Article generation failure cleanup is failing + description: Failure cleanup could not persist task/article/quota state. Check structured logs with event_domain=article_generation and event=cleanup_failure. + + - alert: ArticleGenerationStateAnomaly + expr: increase(article_generation_task_state_anomalies_total[5m]) > 0 + labels: + severity: critical + component: article_generation + annotations: + summary: Article generation task state anomaly detected + description: Bounded state consistency checks found task/article/version mismatch. Inspect anomaly_type, task_id, tenant_id, and article_id in structured logs. + + - alert: ArticleGenerationStateCheckFailure + expr: increase(article_generation_task_state_check_failures_total[15m]) > 0 + labels: + severity: warning + component: article_generation + annotations: + summary: Article generation state check is failing + description: The scheduler could not complete the bounded generation task state check. Inspect scheduler logs and database health. + + - alert: ArticleGenerationQueueFailure + expr: increase(article_generation_task_queue_failures_total[10m]) > 0 + labels: + severity: warning + component: article_generation + annotations: + summary: Article generation queue delivery failed + description: Queue publish or delivery reject errors were observed. Inspect RabbitMQ health and generation task structured logs. + + - alert: ArticleGenerationFailureSpike + expr: increase(article_generation_task_failures_total[15m]) >= 5 + labels: + severity: warning + component: article_generation + annotations: + summary: Article generation failures are spiking + description: Multiple article generation failures occurred in a short window. Group by stage/task_type from structured logs to identify the failing dependency. + + - alert: ArticleGenerationRecoveryFinalize + expr: increase(article_generation_task_recovery_total{action="finalize"}[15m]) > 0 + labels: + severity: warning + component: article_generation + annotations: + summary: Article generation recovery finalized stale tasks + description: Recovery had to mark stale generation tasks failed instead of normal worker completion. Inspect task_id/article_id logs and worker health. diff --git a/findings.md b/findings.md index b69037e..e266045 100644 --- a/findings.md +++ b/findings.md @@ -258,6 +258,15 @@ - Ops compliance is now implemented with dictionary/term CRUD, batch import, publish/rollback, singleton global policy, master switch, manual reviews, records, and stats under `/api/ops/compliance`. - Ops-web now has content-safety navigation and pages for global policy, dictionaries/editing, records, stats, manual-review list, and manual-review detail. - Per-tenant compliance policy customization was removed before launch: ops-web now only exposes global policy pages/APIs, and `ops.compliance_policies` is an `id=1` global singleton without per-tenant columns. + +## Article Generation Reliability Hardening - 2026-05-05 +- Incident source: article title and outline generation can succeed while article body generation fails because the queued article worker path persists task/article/version/quota state through separate transitions. +- The production root cause was a Postgres type inference failure in `UpdateGenerationTaskStatus`; the earlier hotfix cast `status` and migrated `generation_tasks.status` to `TEXT`. +- Remaining reliability gap: body-generation services still ignore the initial `UpdateGenerationTaskStatus(... running ...)` error, and several failure-cleanup paths ignore quota ledger / reservation refund errors. That means a DB error can be present but not promoted with task/article context. +- Recovery already avoids retrying a running task when the article is already `failed`, but it logs only a narrow message. The worker needs stable structured fields and alert-ready counters around retries/finalization so frontend “失败后不再重试” semantics can be verified from logs and metrics. +- State consistency checks should flag article-generation anomalies such as completed task without article version/current version, failed article with queued/running task, completed article with non-completed task, and expired running leases. +- Scheduler DB scanning should stay a bounded safety net, not the primary control loop: defaults are 5-minute interval, 10-second timeout, batch size 100, 24-hour lookback, and partial indexes on active/recent completed generation task candidates. +- Alerting should be metric-driven: Prometheus rules now cover cleanup failures, task/article/version state anomalies, state-check failures, queue failures, failure spikes, and recovery finalization. - Admin-web now has compliance API helpers, an editor-side compliance check drawer with locate/highlight and whitelist actions, and a publish modal that handles pass / needs_ack / block / manual-review flows before creating publish jobs. - Desktop publish management now surfaces scheduled-publish compliance recheck blocks through publish job metadata: `publish_job_status`, `compliance_blocked_record_id`, `compliance_blocked_at`, and summarized `compliance_blocked_reason`. - Verification passed on 2026-05-05: targeted compliance Go tests, full `go test ./...`, admin-web typecheck/build, ops-web typecheck/build, desktop-client typecheck/build, and `git diff --check`. diff --git a/progress.md b/progress.md index 61b5b6a..6b4b727 100644 --- a/progress.md +++ b/progress.md @@ -769,3 +769,24 @@ - Verification passed: - `pnpm --filter ops-web typecheck` - `go test ./internal/ops/app ./internal/ops/transport ./internal/tenant/compliance ./cmd/ops-api` + +## 2026-05-05T21:20:00+08:00 - Article generation reliability hardening + +- Started Phase 44 after the user requested large-company style safeguards: structured logs, verifiable states, no swallowed errors, and automatic alerts. +- Confirmed the remaining weak points are in article body generation worker paths, not title/outline assist paths: + - Initial task `running` updates are ignored in template/custom/imitation generation execution. + - Failure cleanup logs status update errors but still swallows quota ledger/refund errors. + - Recovery requeue/finalization logs do not carry a consistent event/stage/anomaly schema yet. +- Scope locked to generation worker/app observability and state checks, avoiding unrelated deployment/compliance dirty files. + +## 2026-05-05T23:45:00+08:00 - Article generation reliability verification + +- Added structured article-generation task log context and counters for transitions, failures, cleanup failures, queue failures, recovery actions, state anomalies, and state-check failures. +- Added bounded scheduler-owned generation state consistency checks with configurable interval/timeout/batch/lookback and partial DB indexes; this is a low-frequency safety net, not a full-table polling loop. +- Added internal JSON/Prometheus metrics endpoints and Prometheus alert rules for cleanup failures, state anomalies, state-check failures, queue failures, failure spikes, and recovery finalization. +- Fixed compile/test follow-ups: matched the state-check DB interface to `pgx.Rows` and made the Prometheus label assertion order-independent. +- Verification passed: + - `cd server && go test ./internal/shared/config ./internal/tenant/app ./internal/tenant/transport ./internal/worker/generate ./cmd/scheduler ./cmd/worker-generate` + - `cd server && go test ./...` + - `git diff --check` + - `ruby -e 'require "yaml"; YAML.load_file("deploy/monitoring/article-generation-alerts.yml"); puts "ok"'` diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index 12fb6c7..8f648b7 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -19,6 +19,7 @@ import ( internalscheduler "github.com/geo-platform/tenant-api/internal/scheduler" "github.com/geo-platform/tenant-api/internal/shared/response" tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app" + generateworker "github.com/geo-platform/tenant-api/internal/worker/generate" "github.com/gin-gonic/gin" ) @@ -91,6 +92,8 @@ func main() { monitoringLeaseRecoveryWorker := internalscheduler.NewMonitoringLeaseRecoveryWorker(app.MonitoringDB, app.Logger) monitoringReceivedInspectionWorker := internalscheduler.NewMonitoringReceivedInspectionWorker(app.MonitoringDB, app.Logger) knowledgeDeletedCleanupWorker := internalscheduler.NewKnowledgeDeletedCleanupWorker(knowledgeSvc) + generationStateCheckWorker := generateworker.NewGenerationTaskStateCheckWorker(app.DB, app.Logger, app.Config().Generation). + WithConfigProvider(app.ConfigStore) startSchedulerWorker(ctx, &workerWG, "schedule_dispatch", app.Logger.Sugar(), scheduleDispatchWorker.Run) startSchedulerWorker(ctx, &workerWG, "monitoring_daily_task", app.Logger.Sugar(), monitoringDailyTaskWorker.Run) @@ -98,6 +101,7 @@ func main() { startSchedulerWorker(ctx, &workerWG, "monitoring_lease_recovery", app.Logger.Sugar(), monitoringLeaseRecoveryWorker.Run) startSchedulerWorker(ctx, &workerWG, "monitoring_received_inspection", app.Logger.Sugar(), monitoringReceivedInspectionWorker.Run) startSchedulerWorker(ctx, &workerWG, "knowledge_deleted_cleanup", app.Logger.Sugar(), knowledgeDeletedCleanupWorker.Run) + startSchedulerWorker(ctx, &workerWG, "generation_state_check", app.Logger.Sugar(), generationStateCheckWorker.Run) var metricsServer *http.Server if app.Config().Scheduler.HTTPPort > 0 { @@ -176,6 +180,9 @@ func startSchedulerMetricsServer(app *bootstrap.App) *http.Server { app.Engine.GET("/api/internal/metrics/monitoring/daily-tasks", authMiddleware, func(c *gin.Context) { response.Success(c, tenantapp.MonitoringDailyTaskMetricsSnapshotValue()) }) + app.Engine.GET("/api/internal/metrics/generation/tasks", authMiddleware, func(c *gin.Context) { + response.Success(c, tenantapp.GenerationTaskMetricsSnapshotValue()) + }) prometheusHandler := tenantapp.MonitoringSchedulerMetricsPrometheusHandler() app.Engine.GET("/metrics", authMiddleware, func(c *gin.Context) { prometheusHandler.ServeHTTP(c.Writer, c.Request) diff --git a/server/internal/shared/config/config.go b/server/internal/shared/config/config.go index 6558d8a..0101852 100644 --- a/server/internal/shared/config/config.go +++ b/server/internal/shared/config/config.go @@ -332,16 +332,20 @@ type CacheConfig struct { } type GenerationConfig struct { - QueueSize int `mapstructure:"queue_size"` - WorkerConcurrency int `mapstructure:"worker_concurrency"` - StreamEnabled bool `mapstructure:"stream_enabled"` - ArticleTimeout time.Duration `mapstructure:"article_timeout"` - TaskLeaseTTL time.Duration `mapstructure:"task_lease_ttl"` - TaskRecoveryInterval time.Duration `mapstructure:"task_recovery_interval"` - TaskRecoveryTimeout time.Duration `mapstructure:"task_recovery_timeout"` - TaskRecoveryBatchSize int `mapstructure:"task_recovery_batch_size"` - TaskQueuedStaleAfter time.Duration `mapstructure:"task_queued_stale_after"` - TaskMaxAttempts int `mapstructure:"task_max_attempts"` + QueueSize int `mapstructure:"queue_size"` + WorkerConcurrency int `mapstructure:"worker_concurrency"` + StreamEnabled bool `mapstructure:"stream_enabled"` + ArticleTimeout time.Duration `mapstructure:"article_timeout"` + TaskLeaseTTL time.Duration `mapstructure:"task_lease_ttl"` + TaskRecoveryInterval time.Duration `mapstructure:"task_recovery_interval"` + TaskRecoveryTimeout time.Duration `mapstructure:"task_recovery_timeout"` + TaskRecoveryBatchSize int `mapstructure:"task_recovery_batch_size"` + TaskQueuedStaleAfter time.Duration `mapstructure:"task_queued_stale_after"` + TaskMaxAttempts int `mapstructure:"task_max_attempts"` + TaskStateCheckInterval time.Duration `mapstructure:"task_state_check_interval"` + TaskStateCheckTimeout time.Duration `mapstructure:"task_state_check_timeout"` + TaskStateCheckBatchSize int `mapstructure:"task_state_check_batch_size"` + TaskStateCheckLookback time.Duration `mapstructure:"task_state_check_lookback"` } func Load(configPath string) (*Config, error) { @@ -625,6 +629,18 @@ func NormalizeGenerationConfig(cfg *GenerationConfig) { if cfg.TaskMaxAttempts <= 0 { cfg.TaskMaxAttempts = 3 } + if cfg.TaskStateCheckInterval <= 0 { + cfg.TaskStateCheckInterval = 5 * time.Minute + } + if cfg.TaskStateCheckTimeout <= 0 { + cfg.TaskStateCheckTimeout = 10 * time.Second + } + if cfg.TaskStateCheckBatchSize <= 0 { + cfg.TaskStateCheckBatchSize = 100 + } + if cfg.TaskStateCheckLookback <= 0 { + cfg.TaskStateCheckLookback = 24 * time.Hour + } } func NormalizeComplianceConfig(cfg *ComplianceConfig) { @@ -731,6 +747,18 @@ func applyEnvOverrides(cfg *Config) { if n, ok := lookupIntEnv("COMPLIANCE_REVIEW_MAX_ATTEMPTS"); ok { cfg.Compliance.ReviewMaxAttempts = n } + if d, ok := lookupDurationEnv("GENERATION_TASK_STATE_CHECK_INTERVAL"); ok { + cfg.Generation.TaskStateCheckInterval = d + } + if d, ok := lookupDurationEnv("GENERATION_TASK_STATE_CHECK_TIMEOUT"); ok { + cfg.Generation.TaskStateCheckTimeout = d + } + if n, ok := lookupIntEnv("GENERATION_TASK_STATE_CHECK_BATCH_SIZE"); ok { + cfg.Generation.TaskStateCheckBatchSize = n + } + if d, ok := lookupDurationEnv("GENERATION_TASK_STATE_CHECK_LOOKBACK"); ok { + cfg.Generation.TaskStateCheckLookback = d + } if apiKey, ok := lookupNonEmptyEnv("QDRANT_API_KEY"); ok { cfg.Qdrant.APIKey = apiKey } diff --git a/server/internal/shared/config/config_test.go b/server/internal/shared/config/config_test.go index bf3ed2c..5f54f29 100644 --- a/server/internal/shared/config/config_test.go +++ b/server/internal/shared/config/config_test.go @@ -480,6 +480,18 @@ generation: if cfg.Generation.TaskMaxAttempts != 3 { t.Fatalf("expected default task max attempts 3, got %d", cfg.Generation.TaskMaxAttempts) } + if cfg.Generation.TaskStateCheckInterval != 5*time.Minute { + t.Fatalf("expected default task state check interval 5m, got %s", cfg.Generation.TaskStateCheckInterval) + } + if cfg.Generation.TaskStateCheckTimeout != 10*time.Second { + t.Fatalf("expected default task state check timeout 10s, got %s", cfg.Generation.TaskStateCheckTimeout) + } + if cfg.Generation.TaskStateCheckBatchSize != 100 { + t.Fatalf("expected default task state check batch size 100, got %d", cfg.Generation.TaskStateCheckBatchSize) + } + if cfg.Generation.TaskStateCheckLookback != 24*time.Hour { + t.Fatalf("expected default task state check lookback 24h, got %s", cfg.Generation.TaskStateCheckLookback) + } } func TestLoadAppliesGenerationRecoveryOverrides(t *testing.T) { @@ -494,6 +506,10 @@ generation: task_recovery_batch_size: 12 task_queued_stale_after: 45s task_max_attempts: 5 + task_state_check_interval: 9m + task_state_check_timeout: 11s + task_state_check_batch_size: 13 + task_state_check_lookback: 2h `) cfg, err := Load(configPath) @@ -509,11 +525,38 @@ generation: cfg.Generation.TaskRecoveryTimeout != 10*time.Second || cfg.Generation.TaskRecoveryBatchSize != 12 || cfg.Generation.TaskQueuedStaleAfter != 45*time.Second || - cfg.Generation.TaskMaxAttempts != 5 { + cfg.Generation.TaskMaxAttempts != 5 || + cfg.Generation.TaskStateCheckInterval != 9*time.Minute || + cfg.Generation.TaskStateCheckTimeout != 11*time.Second || + cfg.Generation.TaskStateCheckBatchSize != 13 || + cfg.Generation.TaskStateCheckLookback != 2*time.Hour { t.Fatalf("unexpected generation recovery config: %#v", cfg.Generation) } } +func TestLoadAppliesGenerationStateCheckEnvOverrides(t *testing.T) { + t.Setenv("GENERATION_TASK_STATE_CHECK_INTERVAL", "3m") + t.Setenv("GENERATION_TASK_STATE_CHECK_TIMEOUT", "4s") + t.Setenv("GENERATION_TASK_STATE_CHECK_BATCH_SIZE", "17") + t.Setenv("GENERATION_TASK_STATE_CHECK_LOOKBACK", "6h") + + configPath := writeTestConfig(t, ` +generation: {} +`) + + cfg, err := Load(configPath) + if err != nil { + t.Fatalf("Load() error = %v", err) + } + + if cfg.Generation.TaskStateCheckInterval != 3*time.Minute || + cfg.Generation.TaskStateCheckTimeout != 4*time.Second || + cfg.Generation.TaskStateCheckBatchSize != 17 || + cfg.Generation.TaskStateCheckLookback != 6*time.Hour { + t.Fatalf("unexpected generation state check env config: %#v", cfg.Generation) + } +} + func TestStoreReloadsConfigAndKeepsPreviousOnInvalidYAML(t *testing.T) { configPath := writeTestConfig(t, ` jwt: diff --git a/server/internal/tenant/app/article_imitation_service.go b/server/internal/tenant/app/article_imitation_service.go index 590303c..062a118 100644 --- a/server/internal/tenant/app/article_imitation_service.go +++ b/server/internal/tenant/app/article_imitation_service.go @@ -3,6 +3,7 @@ package app import ( "context" "encoding/json" + "errors" "fmt" "net/url" "strings" @@ -246,6 +247,7 @@ func (s *ArticleImitationService) Generate(ctx context.Context, req GenerateImit TenantID: actor.TenantID, ArticleID: articleID, }); err != nil { + RecordGenerationTaskEvent(GenerationTaskEventQueueFailure, articleImitationJobLogContext(job, "queue_enqueue", "queued", "failed", GenerationTaskResultFailure)) s.failGeneration(context.Background(), job, "queue_enqueue", err) return nil, response.ErrServiceUnavailable(50310, "generation_queue_unavailable", "generation queue is busy, please retry") } @@ -302,12 +304,17 @@ func (s *ArticleImitationService) executeGeneration(ctx context.Context, job art } auditRepo := repository.NewAuditRepository(s.pool) - _ = auditRepo.UpdateGenerationTaskStatus(ctx, repository.GenerationTaskStatusInput{ + if err := auditRepo.UpdateGenerationTaskStatus(ctx, repository.GenerationTaskStatusInput{ ID: job.TaskID, TenantID: job.TenantID, Status: "running", StartedAt: &now, - }) + }); err != nil { + s.failGeneration(ctx, job, "task_status_running", err) + return + } + RecordGenerationTaskEvent(GenerationTaskEventTransition, articleImitationJobLogContext(job, "task_status_running", "", GenerationTaskStatusRunning, GenerationTaskResultSuccess)) + LogGenerationTaskInfo(s.logger, "article generation task marked running", articleImitationJobLogContext(job, "task_status_running", "", GenerationTaskStatusRunning, GenerationTaskResultSuccess)) sourceURL := strings.TrimSpace(extractString(job.InputParams, "source_url")) if sourceURL == "" { @@ -458,6 +465,9 @@ func (s *ArticleImitationService) failGeneration(ctx context.Context, job articl errMsg := formatGenerationFailure(stage, genErr) now := time.Now() + logCtx := articleImitationJobLogContext(job, stage, "", "failed", GenerationTaskResultFailure) + RecordGenerationTaskEvent(GenerationTaskEventFailure, logCtx) + LogGenerationTaskWarn(s.logger, "imitation generation failed", logCtx, zap.Error(genErr)) title := strings.TrimSpace(job.InitialTitle) if title == "" { title = articleImitationFallbackTitle @@ -467,6 +477,7 @@ func (s *ArticleImitationService) failGeneration(ctx context.Context, job articl articleRepo := repository.NewArticleRepository(s.pool) quotaRepo := repository.NewQuotaRepository(s.pool) + var cleanupErr error if err := auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ ID: job.TaskID, TenantID: job.TenantID, @@ -480,6 +491,7 @@ func (s *ArticleImitationService) failGeneration(ctx context.Context, job articl zap.Int64("article_id", job.ArticleID), zap.String("stage", stage), ) + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("update generation task failed status: %w", err)) } if job.ArticleID > 0 { if err := articleRepo.UpdateArticleGenerateStatus(cleanupCtx, job.ArticleID, job.TenantID, "failed"); err != nil && s.logger != nil { @@ -489,6 +501,7 @@ func (s *ArticleImitationService) failGeneration(ctx context.Context, job articl zap.Int64("article_id", job.ArticleID), zap.String("stage", stage), ) + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("update article failed status: %w", err)) } _, _ = s.pool.Exec(cleanupCtx, ` UPDATE articles @@ -500,11 +513,13 @@ func (s *ArticleImitationService) failGeneration(ctx context.Context, job articl if job.ReservationID > 0 { balance, balanceErr := quotaRepo.GetCurrentBalance(cleanupCtx, job.TenantID, "article_generation") - if balanceErr == nil { + if balanceErr != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("load quota balance for refund: %w", balanceErr)) + } else { reason := "generation_refund" referenceType := "generation_task" taskReferenceID := job.TaskID - _, _ = quotaRepo.InsertQuotaLedger(cleanupCtx, repository.QuotaLedgerInput{ + if _, err := quotaRepo.InsertQuotaLedger(cleanupCtx, repository.QuotaLedgerInput{ TenantID: job.TenantID, OperatorID: job.OperatorID, QuotaType: "article_generation", @@ -513,14 +528,22 @@ func (s *ArticleImitationService) failGeneration(ctx context.Context, job articl Reason: &reason, ReferenceType: &referenceType, ReferenceID: &taskReferenceID, - }) + }); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("insert generation refund ledger: %w", err)) + } + } + if err := quotaRepo.RefundReservation(cleanupCtx, job.ReservationID, job.TenantID); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("refund generation reservation: %w", err)) } - _ = quotaRepo.RefundReservation(cleanupCtx, job.ReservationID, job.TenantID) } if job.ArticleID > 0 { invalidateArticleCaches(cleanupCtx, s.cache, job.TenantID, &job.ArticleID) } + if cleanupErr != nil { + RecordGenerationTaskEvent(GenerationTaskEventCleanupFailure, logCtx) + LogGenerationTaskError(s.logger, "imitation generation failure cleanup failed", logCtx, zap.Error(cleanupErr)) + } runtimeCfg := s.runtimeConfig() if runtimeCfg.StreamEnabled && s.streams != nil && job.ArticleID > 0 { @@ -528,6 +551,21 @@ func (s *ArticleImitationService) failGeneration(ctx context.Context, job articl } } +func articleImitationJobLogContext(job articleImitationJob, stage, statusBefore, statusAfter, result string) GenerationTaskLogContext { + return GenerationTaskLogContext{ + TaskID: job.TaskID, + TenantID: job.TenantID, + OperatorID: job.OperatorID, + ArticleID: job.ArticleID, + ReservationID: job.ReservationID, + TaskType: articleImitationTaskType, + Stage: stage, + StatusBefore: statusBefore, + StatusAfter: statusAfter, + Result: result, + } +} + func normalizeImitationSourceURL(raw string) (string, error) { trimmed := strings.TrimSpace(raw) if trimmed == "" { diff --git a/server/internal/tenant/app/generation_observability.go b/server/internal/tenant/app/generation_observability.go new file mode 100644 index 0000000..611eae3 --- /dev/null +++ b/server/internal/tenant/app/generation_observability.go @@ -0,0 +1,240 @@ +package app + +import ( + "strings" + "sync/atomic" + "time" + + "go.uber.org/zap" +) + +const GenerationTaskStatusRunning = "running" + +const ( + GenerationTaskEventTransition = "transition" + GenerationTaskEventFailure = "failure" + GenerationTaskEventCleanupFailure = "cleanup_failure" + GenerationTaskEventQueueFailure = "queue_failure" + GenerationTaskEventRecoveryRequeue = "recovery_requeue" + GenerationTaskEventRecoveryFinalize = "recovery_finalize" + GenerationTaskEventRecoverySkipped = "recovery_skipped" + GenerationTaskEventStateAnomaly = "state_anomaly" + GenerationTaskEventStateCheckFailure = "state_check_failure" +) + +const ( + GenerationTaskResultSuccess = "success" + GenerationTaskResultFailure = "failure" + GenerationTaskResultSkipped = "skipped" +) + +type GenerationTaskLogContext struct { + TaskID int64 + TenantID int64 + OperatorID int64 + ArticleID int64 + ReservationID int64 + TaskType string + Stage string + AttemptCount int + LeaseOwner string + StatusBefore string + StatusAfter string + Result string + AnomalyType string + Severity string + Duration time.Duration +} + +func GenerationTaskLogFields(ctx GenerationTaskLogContext) []zap.Field { + fields := []zap.Field{zap.String("event_domain", "article_generation")} + appendStringField := func(key, value string) { + if value = strings.TrimSpace(value); value != "" { + fields = append(fields, zap.String(key, value)) + } + } + appendInt64Field := func(key string, value int64) { + if value > 0 { + fields = append(fields, zap.Int64(key, value)) + } + } + + appendInt64Field("task_id", ctx.TaskID) + appendInt64Field("tenant_id", ctx.TenantID) + appendInt64Field("operator_id", ctx.OperatorID) + appendInt64Field("article_id", ctx.ArticleID) + appendInt64Field("quota_reservation_id", ctx.ReservationID) + appendStringField("task_type", ctx.TaskType) + appendStringField("stage", ctx.Stage) + if ctx.AttemptCount > 0 { + fields = append(fields, zap.Int("attempt_count", ctx.AttemptCount)) + } + appendStringField("lease_owner", ctx.LeaseOwner) + appendStringField("status_before", ctx.StatusBefore) + appendStringField("status_after", ctx.StatusAfter) + appendStringField("result", ctx.Result) + appendStringField("anomaly_type", ctx.AnomalyType) + appendStringField("severity", ctx.Severity) + if ctx.Duration > 0 { + fields = append(fields, zap.Duration("duration", ctx.Duration)) + fields = append(fields, zap.Int64("duration_ms", ctx.Duration.Milliseconds())) + } + return fields +} + +func LogGenerationTaskInfo(logger *zap.Logger, message string, ctx GenerationTaskLogContext, fields ...zap.Field) { + if logger == nil { + return + } + logger.Info(message, append(GenerationTaskLogFields(ctx), fields...)...) +} + +func LogGenerationTaskWarn(logger *zap.Logger, message string, ctx GenerationTaskLogContext, fields ...zap.Field) { + if logger == nil { + return + } + logger.Warn(message, append(GenerationTaskLogFields(ctx), fields...)...) +} + +func LogGenerationTaskError(logger *zap.Logger, message string, ctx GenerationTaskLogContext, fields ...zap.Field) { + if logger == nil { + return + } + logger.Error(message, append(GenerationTaskLogFields(ctx), fields...)...) +} + +type GenerationTaskMetricsSnapshot struct { + TransitionTotal int64 `json:"transition_total"` + FailureTotal int64 `json:"failure_total"` + CleanupFailureTotal int64 `json:"cleanup_failure_total"` + QueueFailureTotal int64 `json:"queue_failure_total"` + RecoveryRequeueTotal int64 `json:"recovery_requeue_total"` + RecoveryFinalizeTotal int64 `json:"recovery_finalize_total"` + StateAnomalyTotal int64 `json:"state_anomaly_total"` + StateCheckFailureTotal int64 `json:"state_check_failure_total"` + LastEventAt *time.Time `json:"last_event_at,omitempty"` + LastEvent string `json:"last_event,omitempty"` + LastTaskType string `json:"last_task_type,omitempty"` + LastStage string `json:"last_stage,omitempty"` + LastResult string `json:"last_result,omitempty"` + LastAnomalyType string `json:"last_anomaly_type,omitempty"` + LastSeverity string `json:"last_severity,omitempty"` +} + +type generationTaskMetricsRecorder struct { + transitionTotal atomic.Int64 + failureTotal atomic.Int64 + cleanupFailureTotal atomic.Int64 + queueFailureTotal atomic.Int64 + recoveryRequeueTotal atomic.Int64 + recoveryFinalizeTotal atomic.Int64 + stateAnomalyTotal atomic.Int64 + stateCheckFailureTotal atomic.Int64 + lastEventAtUnixNano atomic.Int64 + lastEvent atomic.Value + lastTaskType atomic.Value + lastStage atomic.Value + lastResult atomic.Value + lastAnomalyType atomic.Value + lastSeverity atomic.Value +} + +var generationTaskMetrics = newGenerationTaskMetricsRecorder() + +func newGenerationTaskMetricsRecorder() *generationTaskMetricsRecorder { + return &generationTaskMetricsRecorder{} +} + +func RecordGenerationTaskEvent(event string, ctx GenerationTaskLogContext) { + generationTaskMetrics.record(event, ctx) +} + +func GenerationTaskMetricsSnapshotValue() GenerationTaskMetricsSnapshot { + return generationTaskMetrics.snapshot() +} + +func resetGenerationTaskMetricsForTest() { + generationTaskMetrics = newGenerationTaskMetricsRecorder() +} + +func (r *generationTaskMetricsRecorder) record(event string, ctx GenerationTaskLogContext) { + if r == nil { + return + } + event = strings.TrimSpace(event) + if event == "" { + event = "unknown" + } + switch event { + case GenerationTaskEventTransition: + r.transitionTotal.Add(1) + case GenerationTaskEventFailure: + r.failureTotal.Add(1) + case GenerationTaskEventCleanupFailure: + r.cleanupFailureTotal.Add(1) + case GenerationTaskEventQueueFailure: + r.queueFailureTotal.Add(1) + case GenerationTaskEventRecoveryRequeue: + r.recoveryRequeueTotal.Add(1) + case GenerationTaskEventRecoveryFinalize: + r.recoveryFinalizeTotal.Add(1) + case GenerationTaskEventStateAnomaly: + r.stateAnomalyTotal.Add(1) + case GenerationTaskEventStateCheckFailure: + r.stateCheckFailureTotal.Add(1) + } + r.lastEventAtUnixNano.Store(time.Now().UTC().UnixNano()) + r.lastEvent.Store(event) + storeGenerationMetricString(&r.lastTaskType, ctx.TaskType) + storeGenerationMetricString(&r.lastStage, ctx.Stage) + storeGenerationMetricString(&r.lastResult, ctx.Result) + storeGenerationMetricString(&r.lastAnomalyType, ctx.AnomalyType) + storeGenerationMetricString(&r.lastSeverity, ctx.Severity) +} + +func (r *generationTaskMetricsRecorder) snapshot() GenerationTaskMetricsSnapshot { + if r == nil { + return GenerationTaskMetricsSnapshot{} + } + return GenerationTaskMetricsSnapshot{ + TransitionTotal: r.transitionTotal.Load(), + FailureTotal: r.failureTotal.Load(), + CleanupFailureTotal: r.cleanupFailureTotal.Load(), + QueueFailureTotal: r.queueFailureTotal.Load(), + RecoveryRequeueTotal: r.recoveryRequeueTotal.Load(), + RecoveryFinalizeTotal: r.recoveryFinalizeTotal.Load(), + StateAnomalyTotal: r.stateAnomalyTotal.Load(), + StateCheckFailureTotal: r.stateCheckFailureTotal.Load(), + LastEventAt: generationMetricUnixNanoTime(r.lastEventAtUnixNano.Load()), + LastEvent: loadGenerationMetricString(&r.lastEvent), + LastTaskType: loadGenerationMetricString(&r.lastTaskType), + LastStage: loadGenerationMetricString(&r.lastStage), + LastResult: loadGenerationMetricString(&r.lastResult), + LastAnomalyType: loadGenerationMetricString(&r.lastAnomalyType), + LastSeverity: loadGenerationMetricString(&r.lastSeverity), + } +} + +func storeGenerationMetricString(target *atomic.Value, value string) { + if target == nil { + return + } + target.Store(strings.TrimSpace(value)) +} + +func loadGenerationMetricString(value *atomic.Value) string { + if value == nil { + return "" + } + raw := value.Load() + text, _ := raw.(string) + return text +} + +func generationMetricUnixNanoTime(value int64) *time.Time { + if value <= 0 { + return nil + } + t := time.Unix(0, value).UTC() + return &t +} diff --git a/server/internal/tenant/app/generation_observability_test.go b/server/internal/tenant/app/generation_observability_test.go new file mode 100644 index 0000000..4a943a4 --- /dev/null +++ b/server/internal/tenant/app/generation_observability_test.go @@ -0,0 +1,101 @@ +package app + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestGenerationTaskLogFieldsUsesStableStructuredKeys(t *testing.T) { + fields := GenerationTaskLogFields(GenerationTaskLogContext{ + TaskID: 11, + TenantID: 22, + OperatorID: 33, + ArticleID: 44, + ReservationID: 55, + TaskType: "template", + Stage: "llm_generate", + AttemptCount: 2, + LeaseOwner: "worker-1", + StatusBefore: "running", + StatusAfter: "failed", + Result: GenerationTaskResultFailure, + AnomalyType: "article_failed_task_running", + Severity: "critical", + }) + + values := map[string]any{} + for _, field := range fields { + values[field.Key] = field.Interface + if field.String != "" { + values[field.Key] = field.String + } + if field.Integer != 0 { + values[field.Key] = field.Integer + } + } + + assert.Equal(t, "article_generation", values["event_domain"]) + assert.Equal(t, int64(11), values["task_id"]) + assert.Equal(t, int64(22), values["tenant_id"]) + assert.Equal(t, int64(44), values["article_id"]) + assert.Equal(t, "template", values["task_type"]) + assert.Equal(t, "llm_generate", values["stage"]) + assert.Equal(t, "failed", values["status_after"]) + assert.Equal(t, "critical", values["severity"]) +} + +func TestGenerationTaskMetricsRecordsEventsAndPrometheus(t *testing.T) { + resetGenerationTaskMetricsForTest() + t.Cleanup(resetGenerationTaskMetricsForTest) + + RecordGenerationTaskEvent(GenerationTaskEventTransition, GenerationTaskLogContext{ + TaskType: "template", + Stage: "worker_claim", + StatusAfter: GenerationTaskStatusRunning, + Result: GenerationTaskResultSuccess, + }) + RecordGenerationTaskEvent(GenerationTaskEventCleanupFailure, GenerationTaskLogContext{ + TaskType: "template", + Stage: "refund_reservation", + Result: GenerationTaskResultFailure, + Severity: "critical", + }) + RecordGenerationTaskEvent(GenerationTaskEventStateAnomaly, GenerationTaskLogContext{ + TaskType: "template", + Stage: "state_check", + AnomalyType: "completed_task_missing_version", + Severity: "critical", + }) + + snapshot := GenerationTaskMetricsSnapshotValue() + assert.Equal(t, int64(1), snapshot.TransitionTotal) + assert.Equal(t, int64(1), snapshot.CleanupFailureTotal) + assert.Equal(t, int64(1), snapshot.StateAnomalyTotal) + assert.Equal(t, "state_anomaly", snapshot.LastEvent) + assert.Equal(t, "completed_task_missing_version", snapshot.LastAnomalyType) + assert.NotNil(t, snapshot.LastEventAt) + + prometheus := GenerationTaskMetricsPrometheus() + assert.Contains(t, prometheus, "article_generation_task_transitions_total 1") + assert.Contains(t, prometheus, "article_generation_task_cleanup_failures_total 1") + assert.Contains(t, prometheus, "article_generation_task_state_anomalies_total 1") + lastEventLine := findPrometheusMetricLine(prometheus, "article_generation_task_last_event_info") + assert.Contains(t, lastEventLine, `event="state_anomaly"`) + assert.Contains(t, lastEventLine, `task_type="template"`) + assert.Contains(t, lastEventLine, `stage="state_check"`) + assert.Contains(t, lastEventLine, `result=""`) + assert.Contains(t, lastEventLine, `anomaly_type="completed_task_missing_version"`) + assert.Contains(t, lastEventLine, `severity="critical"`) + assert.True(t, strings.HasSuffix(lastEventLine, " 1"), lastEventLine) +} + +func findPrometheusMetricLine(output, metricName string) string { + for _, line := range strings.Split(output, "\n") { + if strings.HasPrefix(line, metricName+"{") || strings.HasPrefix(line, metricName+" ") { + return line + } + } + return "" +} diff --git a/server/internal/tenant/app/generation_task_runtime.go b/server/internal/tenant/app/generation_task_runtime.go index abba4f3..c742fcd 100644 --- a/server/internal/tenant/app/generation_task_runtime.go +++ b/server/internal/tenant/app/generation_task_runtime.go @@ -80,11 +80,13 @@ func HandleClaimedGenerationTaskFailure( if task.QuotaReservationID > 0 { balance, balanceErr := quotaRepo.GetCurrentBalance(cleanupCtx, task.TenantID, "article_generation") - if balanceErr == nil && task.OperatorID > 0 { + if balanceErr != nil { + failureErr = errors.Join(failureErr, fmt.Errorf("load quota balance for refund: %w", balanceErr)) + } else if task.OperatorID > 0 { reason := "generation_refund" referenceType := "generation_task" taskReferenceID := task.ID - _, _ = quotaRepo.InsertQuotaLedger(cleanupCtx, repository.QuotaLedgerInput{ + if _, err := quotaRepo.InsertQuotaLedger(cleanupCtx, repository.QuotaLedgerInput{ TenantID: task.TenantID, OperatorID: task.OperatorID, QuotaType: "article_generation", @@ -93,10 +95,14 @@ func HandleClaimedGenerationTaskFailure( Reason: &reason, ReferenceType: &referenceType, ReferenceID: &taskReferenceID, - }) + }); err != nil { + failureErr = errors.Join(failureErr, fmt.Errorf("insert generation refund ledger: %w", err)) + } } - _ = quotaRepo.RefundReservation(cleanupCtx, task.QuotaReservationID, task.TenantID) + if err := quotaRepo.RefundReservation(cleanupCtx, task.QuotaReservationID, task.TenantID); err != nil { + failureErr = errors.Join(failureErr, fmt.Errorf("refund generation reservation: %w", err)) + } } if task.ArticleID > 0 { diff --git a/server/internal/tenant/app/kol_generation_service.go b/server/internal/tenant/app/kol_generation_service.go index e999134..7faca85 100644 --- a/server/internal/tenant/app/kol_generation_service.go +++ b/server/internal/tenant/app/kol_generation_service.go @@ -327,6 +327,19 @@ func (s *KolGenerationService) Submit(ctx context.Context, actor auth.Actor, sub TenantID: actor.TenantID, ArticleID: articleID, }); err != nil { + logCtx := GenerationTaskLogContext{ + TaskID: task.ID, + TenantID: task.TenantID, + OperatorID: task.OperatorID, + ArticleID: task.ArticleID, + ReservationID: task.QuotaReservationID, + TaskType: kolGenerationTaskType, + Stage: "queue_enqueue", + StatusBefore: "queued", + StatusAfter: "failed", + Result: GenerationTaskResultFailure, + } + RecordGenerationTaskEvent(GenerationTaskEventQueueFailure, logCtx) _ = HandleKolGenerationTaskFailure(context.Background(), s.pool, s.cache, task, "queue_enqueue", err) return nil, response.ErrServiceUnavailable(50343, "generation_queue_unavailable", "generation queue is busy, please retry") } @@ -385,6 +398,18 @@ func HandleKolGenerationTaskFailure( errMsg := formatGenerationFailure(stage, taskErr) now := time.Now() + logCtx := GenerationTaskLogContext{ + TaskID: task.ID, + TenantID: task.TenantID, + OperatorID: task.OperatorID, + ArticleID: task.ArticleID, + ReservationID: task.QuotaReservationID, + TaskType: kolGenerationTaskType, + Stage: stage, + StatusAfter: "failed", + Result: GenerationTaskResultFailure, + } + RecordGenerationTaskEvent(GenerationTaskEventFailure, logCtx) auditRepo := repository.NewAuditRepository(pool) articleRepo := repository.NewArticleRepository(pool) @@ -421,11 +446,13 @@ func HandleKolGenerationTaskFailure( if task.QuotaReservationID > 0 { balance, balanceErr := quotaRepo.GetCurrentBalance(cleanupCtx, task.TenantID, "article_generation") - if balanceErr == nil && task.OperatorID > 0 { + if balanceErr != nil { + failureErr = errors.Join(failureErr, fmt.Errorf("load quota balance for refund: %w", balanceErr)) + } else if task.OperatorID > 0 { reason := "generation_refund" referenceType := "generation_task" taskReferenceID := task.ID - _, _ = quotaRepo.InsertQuotaLedger(cleanupCtx, repository.QuotaLedgerInput{ + if _, err := quotaRepo.InsertQuotaLedger(cleanupCtx, repository.QuotaLedgerInput{ TenantID: task.TenantID, OperatorID: task.OperatorID, QuotaType: "article_generation", @@ -434,11 +461,18 @@ func HandleKolGenerationTaskFailure( Reason: &reason, ReferenceType: &referenceType, ReferenceID: &taskReferenceID, - }) + }); err != nil { + failureErr = errors.Join(failureErr, fmt.Errorf("insert generation refund ledger: %w", err)) + } + } + if err := quotaRepo.RefundReservation(cleanupCtx, task.QuotaReservationID, task.TenantID); err != nil { + failureErr = errors.Join(failureErr, fmt.Errorf("refund generation reservation: %w", err)) } - _ = quotaRepo.RefundReservation(cleanupCtx, task.QuotaReservationID, task.TenantID) } InvalidateArticleCaches(cleanupCtx, cache, task.TenantID, &task.ArticleID) + if failureErr != nil { + RecordGenerationTaskEvent(GenerationTaskEventCleanupFailure, logCtx) + } return failureErr } diff --git a/server/internal/tenant/app/monitoring_metrics_prometheus.go b/server/internal/tenant/app/monitoring_metrics_prometheus.go index 3df0e19..41701ad 100644 --- a/server/internal/tenant/app/monitoring_metrics_prometheus.go +++ b/server/internal/tenant/app/monitoring_metrics_prometheus.go @@ -14,7 +14,7 @@ import ( ) func MonitoringSchedulerMetricsPrometheusHandler() http.Handler { - return monitoringPrometheusHandler(true, monitoringDailyTaskMetricsCollector{}, cache.MetricsCollector{}) + return monitoringPrometheusHandler(true, monitoringDailyTaskMetricsCollector{}, generationTaskMetricsCollector{}, cache.MetricsCollector{}) } func MonitoringDailyTaskMetricsPrometheus() string { @@ -41,6 +41,14 @@ func MonitoringHeartbeatSnapshotMetricsPrometheusHandler() http.Handler { return monitoringPrometheusHandler(false, monitoringHeartbeatSnapshotMetricsCollector{}) } +func GenerationTaskMetricsPrometheus() string { + return renderMonitoringPrometheus(monitoringPrometheusRegistry(false, generationTaskMetricsCollector{})) +} + +func GenerationTaskMetricsPrometheusHandler() http.Handler { + return monitoringPrometheusHandler(false, generationTaskMetricsCollector{}) +} + func monitoringPrometheusHandler(includeRuntime bool, metricCollectors ...prometheus.Collector) http.Handler { return promhttp.HandlerFor(monitoringPrometheusRegistry(includeRuntime, metricCollectors...), promhttp.HandlerOpts{}) } @@ -155,6 +163,33 @@ func (monitoringHeartbeatSnapshotMetricsCollector) Collect(ch chan<- prometheus. } } +type generationTaskMetricsCollector struct{} + +func (generationTaskMetricsCollector) Describe(chan<- *prometheus.Desc) {} + +func (generationTaskMetricsCollector) Collect(ch chan<- prometheus.Metric) { + snapshot := GenerationTaskMetricsSnapshotValue() + monitoringCounter(ch, "article_generation_task_transitions_total", "Total article generation task state transition attempts.", float64(snapshot.TransitionTotal), nil) + monitoringCounter(ch, "article_generation_task_failures_total", "Total article generation task failures recorded by business stage.", float64(snapshot.FailureTotal), nil) + monitoringCounter(ch, "article_generation_task_cleanup_failures_total", "Total article generation failure-cleanup errors that need operator attention.", float64(snapshot.CleanupFailureTotal), nil) + monitoringCounter(ch, "article_generation_task_queue_failures_total", "Total article generation task queue publish or delivery errors.", float64(snapshot.QueueFailureTotal), nil) + monitoringCounter(ch, "article_generation_task_recovery_total", "Total article generation recovery actions by action.", float64(snapshot.RecoveryRequeueTotal), []string{"action"}, "requeue") + monitoringCounter(ch, "article_generation_task_recovery_total", "Total article generation recovery actions by action.", float64(snapshot.RecoveryFinalizeTotal), []string{"action"}, "finalize") + monitoringCounter(ch, "article_generation_task_state_anomalies_total", "Total article generation task/article state anomalies detected by bounded consistency checks.", float64(snapshot.StateAnomalyTotal), nil) + monitoringCounter(ch, "article_generation_task_state_check_failures_total", "Total article generation state consistency check failures.", float64(snapshot.StateCheckFailureTotal), nil) + if snapshot.LastEvent != "" { + monitoringGauge(ch, "article_generation_task_last_event_info", "Last article generation task observability event.", 1, []string{"event", "task_type", "stage", "result", "anomaly_type", "severity"}, + snapshot.LastEvent, + snapshot.LastTaskType, + snapshot.LastStage, + snapshot.LastResult, + snapshot.LastAnomalyType, + snapshot.LastSeverity, + ) + } + monitoringGauge(ch, "article_generation_task_last_event_timestamp_seconds", "Unix timestamp of the last article generation task observability event.", monitoringDailyTimeSeconds(snapshot.LastEventAt), nil) +} + func monitoringCounter(ch chan<- prometheus.Metric, name, help string, value float64, labelNames []string, labelValues ...string) { monitoringMetric(ch, name, help, prometheus.CounterValue, value, labelNames, labelValues...) } diff --git a/server/internal/tenant/app/prompt_generate_service.go b/server/internal/tenant/app/prompt_generate_service.go index e2bbaea..36c7de2 100644 --- a/server/internal/tenant/app/prompt_generate_service.go +++ b/server/internal/tenant/app/prompt_generate_service.go @@ -3,6 +3,7 @@ package app import ( "context" "encoding/json" + "errors" "fmt" "strconv" "strings" @@ -445,6 +446,7 @@ func (s *PromptRuleGenerationService) enqueuePromptRuleGeneration(ctx context.Co TenantID: input.TenantID, ArticleID: articleID, }); err != nil { + RecordGenerationTaskEvent(GenerationTaskEventQueueFailure, promptRuleGenerationJobLogContext(job, "queue_enqueue", "queued", "failed", GenerationTaskResultFailure)) s.failGeneration(context.Background(), job, promptRuleGeneratingTitlePlaceholder, "queue_enqueue", err) return nil, response.ErrServiceUnavailable(50310, "generation_queue_unavailable", "generation queue is busy, please retry") } @@ -508,12 +510,17 @@ func (s *PromptRuleGenerationService) executeGeneration(ctx context.Context, job } auditRepo := repository.NewAuditRepository(s.pool) - _ = auditRepo.UpdateGenerationTaskStatus(ctx, repository.GenerationTaskStatusInput{ + if err := auditRepo.UpdateGenerationTaskStatus(ctx, repository.GenerationTaskStatusInput{ ID: job.TaskID, TenantID: job.TenantID, Status: "running", StartedAt: &now, - }) + }); err != nil { + s.failGeneration(ctx, job, title, "task_status_running", err) + return + } + RecordGenerationTaskEvent(GenerationTaskEventTransition, promptRuleGenerationJobLogContext(job, "task_status_running", "", GenerationTaskStatusRunning, GenerationTaskResultSuccess)) + LogGenerationTaskInfo(s.logger, "article generation task marked running", promptRuleGenerationJobLogContext(job, "task_status_running", "", GenerationTaskStatusRunning, GenerationTaskResultSuccess)) runtimeCfg := s.runtimeConfig() req := llm.GenerateRequest{ @@ -686,6 +693,9 @@ func (s *PromptRuleGenerationService) failGeneration(ctx context.Context, job pr errMsg := formatGenerationFailure(stage, genErr) now := time.Now() + logCtx := promptRuleGenerationJobLogContext(job, stage, "", "failed", GenerationTaskResultFailure) + RecordGenerationTaskEvent(GenerationTaskEventFailure, logCtx) + LogGenerationTaskWarn(s.logger, "prompt rule generation failed", logCtx, zap.Error(genErr)) failureTitle := strings.TrimSpace(extractString(job.InputParams, "task_name")) if failureTitle == "" { failureTitle = title @@ -695,6 +705,7 @@ func (s *PromptRuleGenerationService) failGeneration(ctx context.Context, job pr articleRepo := repository.NewArticleRepository(s.pool) quotaRepo := repository.NewQuotaRepository(s.pool) + var cleanupErr error if err := auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ ID: job.TaskID, TenantID: job.TenantID, @@ -708,6 +719,7 @@ func (s *PromptRuleGenerationService) failGeneration(ctx context.Context, job pr zap.Int64("article_id", job.ArticleID), zap.String("stage", stage), ) + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("update generation task failed status: %w", err)) } 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", @@ -716,6 +728,7 @@ func (s *PromptRuleGenerationService) failGeneration(ctx context.Context, job pr zap.Int64("article_id", job.ArticleID), zap.String("stage", stage), ) + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("update article failed status: %w", err)) } if failureTitle != "" { _, _ = s.pool.Exec(cleanupCtx, ` @@ -727,11 +740,13 @@ func (s *PromptRuleGenerationService) failGeneration(ctx context.Context, job pr } balance, balanceErr := quotaRepo.GetCurrentBalance(cleanupCtx, job.TenantID, "article_generation") - if balanceErr == nil { + if balanceErr != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("load quota balance for refund: %w", balanceErr)) + } else { reason := "generation_refund" referenceType := "generation_task" taskReferenceID := job.TaskID - _, _ = quotaRepo.InsertQuotaLedger(cleanupCtx, repository.QuotaLedgerInput{ + if _, err := quotaRepo.InsertQuotaLedger(cleanupCtx, repository.QuotaLedgerInput{ TenantID: job.TenantID, OperatorID: job.OperatorID, QuotaType: "article_generation", @@ -740,11 +755,19 @@ func (s *PromptRuleGenerationService) failGeneration(ctx context.Context, job pr Reason: &reason, ReferenceType: &referenceType, ReferenceID: &taskReferenceID, - }) + }); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("insert generation refund ledger: %w", err)) + } } - _ = quotaRepo.RefundReservation(cleanupCtx, job.ReservationID, job.TenantID) + if err := quotaRepo.RefundReservation(cleanupCtx, job.ReservationID, job.TenantID); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("refund generation reservation: %w", err)) + } invalidateArticleCaches(cleanupCtx, s.cache, job.TenantID, &job.ArticleID) + if cleanupErr != nil { + RecordGenerationTaskEvent(GenerationTaskEventCleanupFailure, logCtx) + LogGenerationTaskError(s.logger, "prompt rule generation failure cleanup failed", logCtx, zap.Error(cleanupErr)) + } runtimeCfg := s.runtimeConfig() if runtimeCfg.StreamEnabled && s.streams != nil { @@ -752,6 +775,21 @@ func (s *PromptRuleGenerationService) failGeneration(ctx context.Context, job pr } } +func promptRuleGenerationJobLogContext(job promptRuleGenerationJob, stage, statusBefore, statusAfter, result string) GenerationTaskLogContext { + return GenerationTaskLogContext{ + TaskID: job.TaskID, + TenantID: job.TenantID, + OperatorID: job.OperatorID, + ArticleID: job.ArticleID, + ReservationID: job.ReservationID, + TaskType: "custom_generation", + Stage: stage, + StatusBefore: statusBefore, + StatusAfter: statusAfter, + Result: result, + } +} + func buildPromptRuleGenerationPrompt(rule *promptRuleGenerationRecord, params map[string]interface{}, knowledgePrompt string) string { sections := []string{strings.TrimSpace(rule.PromptContent)} diff --git a/server/internal/tenant/app/template_service.go b/server/internal/tenant/app/template_service.go index 13ea82d..7cd85bd 100644 --- a/server/internal/tenant/app/template_service.go +++ b/server/internal/tenant/app/template_service.go @@ -409,6 +409,7 @@ func (s *TemplateService) Generate(ctx context.Context, templateID int64, req Ge TenantID: actor.TenantID, ArticleID: articleID, }); err != nil { + RecordGenerationTaskEvent(GenerationTaskEventQueueFailure, generationJobLogContext(job, "queue_enqueue", "queued", "failed", GenerationTaskResultFailure)) s.failGeneration(context.Background(), job, initialTitle, "queue_enqueue", err) return nil, response.ErrServiceUnavailable(50310, "generation_queue_unavailable", "generation queue is busy, please retry") } @@ -457,12 +458,17 @@ func (s *TemplateService) executeGeneration(ctx context.Context, job generationJ } auditRepo := repository.NewAuditRepository(s.pool) - _ = auditRepo.UpdateGenerationTaskStatus(ctx, repository.GenerationTaskStatusInput{ + if err := auditRepo.UpdateGenerationTaskStatus(ctx, repository.GenerationTaskStatusInput{ ID: job.TaskID, TenantID: job.TenantID, Status: "running", StartedAt: &now, - }) + }); err != nil { + s.failGeneration(ctx, job, title, "task_status_running", err) + return + } + RecordGenerationTaskEvent(GenerationTaskEventTransition, generationJobLogContext(job, "task_status_running", "", GenerationTaskStatusRunning, GenerationTaskResultSuccess)) + LogGenerationTaskInfo(s.logger, "article generation task marked running", generationJobLogContext(job, "task_status_running", "", GenerationTaskStatusRunning, GenerationTaskResultSuccess)) knowledgePrompt := "" if knowledgeGroupIDs := extractKnowledgeGroupIDs(job.Params["knowledge_group_ids"]); len(knowledgeGroupIDs) > 0 { @@ -586,11 +592,15 @@ func (s *TemplateService) failGeneration(ctx context.Context, job generationJob, errMsg := formatGenerationFailure(stage, genErr) now := time.Now() + logCtx := generationJobLogContext(job, stage, "", "failed", GenerationTaskResultFailure) + RecordGenerationTaskEvent(GenerationTaskEventFailure, logCtx) + LogGenerationTaskWarn(s.logger, "template generation failed", logCtx, zap.Error(genErr)) auditRepo := repository.NewAuditRepository(s.pool) articleRepo := repository.NewArticleRepository(s.pool) quotaRepo := repository.NewQuotaRepository(s.pool) + var cleanupErr error if err := auditRepo.UpdateGenerationTaskStatus(cleanupCtx, repository.GenerationTaskStatusInput{ ID: job.TaskID, TenantID: job.TenantID, @@ -604,6 +614,7 @@ func (s *TemplateService) failGeneration(ctx context.Context, job generationJob, zap.Int64("article_id", job.ArticleID), zap.String("stage", stage), ) + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("update generation task failed status: %w", err)) } 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", @@ -612,14 +623,17 @@ func (s *TemplateService) failGeneration(ctx context.Context, job generationJob, zap.Int64("article_id", job.ArticleID), zap.String("stage", stage), ) + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("update article failed status: %w", err)) } balance, balanceErr := quotaRepo.GetCurrentBalance(cleanupCtx, job.TenantID, "article_generation") - if balanceErr == nil { + if balanceErr != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("load quota balance for refund: %w", balanceErr)) + } else { reason := "generation_refund" referenceType := "generation_task" taskReferenceID := job.TaskID - _, _ = quotaRepo.InsertQuotaLedger(cleanupCtx, repository.QuotaLedgerInput{ + if _, err := quotaRepo.InsertQuotaLedger(cleanupCtx, repository.QuotaLedgerInput{ TenantID: job.TenantID, OperatorID: job.OperatorID, QuotaType: "article_generation", @@ -628,11 +642,19 @@ func (s *TemplateService) failGeneration(ctx context.Context, job generationJob, Reason: &reason, ReferenceType: &referenceType, ReferenceID: &taskReferenceID, - }) + }); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("insert generation refund ledger: %w", err)) + } } - _ = quotaRepo.RefundReservation(cleanupCtx, job.ReservationID, job.TenantID) + if err := quotaRepo.RefundReservation(cleanupCtx, job.ReservationID, job.TenantID); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("refund generation reservation: %w", err)) + } invalidateArticleCaches(cleanupCtx, s.cache, job.TenantID, &job.ArticleID) + if cleanupErr != nil { + RecordGenerationTaskEvent(GenerationTaskEventCleanupFailure, logCtx) + LogGenerationTaskError(s.logger, "template generation failure cleanup failed", logCtx, zap.Error(cleanupErr)) + } runtimeCfg := s.runtimeConfig() if runtimeCfg.StreamEnabled && s.streams != nil { @@ -640,6 +662,21 @@ func (s *TemplateService) failGeneration(ctx context.Context, job generationJob, } } +func generationJobLogContext(job generationJob, stage, statusBefore, statusAfter, result string) GenerationTaskLogContext { + return GenerationTaskLogContext{ + TaskID: job.TaskID, + TenantID: job.TenantID, + OperatorID: job.OperatorID, + ArticleID: job.ArticleID, + ReservationID: job.ReservationID, + TaskType: "template", + Stage: stage, + StatusBefore: statusBefore, + StatusAfter: statusAfter, + Result: result, + } +} + func formatGenerationFailure(stage string, genErr error) string { cause := "" if genErr != nil { @@ -657,6 +694,8 @@ func generationFailureStageLabel(stage string) string { switch stage { case "task_load": return "加载生成任务失败" + case "task_status_running": + return "更新生成任务运行状态失败" case "queue_enqueue": return "任务入队失败" case "llm_generate": diff --git a/server/internal/tenant/transport/internal_metrics_handler.go b/server/internal/tenant/transport/internal_metrics_handler.go index d9db296..281ef52 100644 --- a/server/internal/tenant/transport/internal_metrics_handler.go +++ b/server/internal/tenant/transport/internal_metrics_handler.go @@ -33,6 +33,13 @@ func registerInternalMetricsRoutes(app *bootstrap.App) { metrics.GET("/monitoring/heartbeat-snapshots/prometheus", func(c *gin.Context) { snapshotPrometheusHandler.ServeHTTP(c.Writer, c.Request) }) + metrics.GET("/generation/tasks", func(c *gin.Context) { + response.Success(c, tenantapp.GenerationTaskMetricsSnapshotValue()) + }) + generationTasksPrometheusHandler := tenantapp.GenerationTaskMetricsPrometheusHandler() + metrics.GET("/generation/tasks/prometheus", func(c *gin.Context) { + generationTasksPrometheusHandler.ServeHTTP(c.Writer, c.Request) + }) } func internalMetricsAuthMiddleware(app *bootstrap.App) gin.HandlerFunc { diff --git a/server/internal/worker/generate/article_generation_recovery.go b/server/internal/worker/generate/article_generation_recovery.go index a9ede9e..9381f80 100644 --- a/server/internal/worker/generate/article_generation_recovery.go +++ b/server/internal/worker/generate/article_generation_recovery.go @@ -32,6 +32,13 @@ type generationTaskRecoveryConfig struct { MaxAttempts int } +type generationTaskStateCheckConfig struct { + Interval time.Duration + Timeout time.Duration + BatchSize int + Lookback time.Duration +} + type claimedGenerationTaskLease struct { task generationTaskRecoveryRecord token string @@ -139,6 +146,17 @@ func normalizeGenerationTaskRecoveryConfig(cfg config.GenerationConfig) generati } } +func normalizeGenerationTaskStateCheckConfig(cfg config.GenerationConfig) generationTaskStateCheckConfig { + normalized := cfg + config.NormalizeGenerationConfig(&normalized) + return generationTaskStateCheckConfig{ + Interval: normalized.TaskStateCheckInterval, + Timeout: normalized.TaskStateCheckTimeout, + BatchSize: normalized.TaskStateCheckBatchSize, + Lookback: normalized.TaskStateCheckLookback, + } +} + func (w *ArticleGenerationWorker) claimGenerationTask(ctx context.Context, taskID int64, owner string, leaseTTL time.Duration) (*claimedGenerationTaskLease, bool, error) { if leaseTTL <= 0 { leaseTTL = normalizeGenerationTaskRecoveryConfig(config.GenerationConfig{}).LeaseTTL @@ -362,6 +380,20 @@ func (w *ArticleGenerationWorker) requeueStaleGenerationTasks(ctx context.Contex } continue } + logCtx := tenantapp.GenerationTaskLogContext{ + TaskID: item.ID, + TenantID: item.TenantID, + ArticleID: item.ArticleID, + TaskType: item.TaskType, + Stage: "recovery_stale_queued", + StatusBefore: "queued", + StatusAfter: "queued", + Result: tenantapp.GenerationTaskResultSuccess, + Severity: "warning", + AttemptCount: item.AttemptCount, + } + tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventRecoveryRequeue, logCtx) + tenantapp.LogGenerationTaskWarn(w.logger, "article generation stale queued task republished", logCtx) requeued++ } @@ -538,17 +570,39 @@ func (w *ArticleGenerationWorker) requeueExpiredRunningGenerationTask(ctx contex return false, err } if !shouldRequeue { + logCtx := tenantapp.GenerationTaskLogContext{ + TaskID: item.ID, + TenantID: tenantID, + ArticleID: articleID, + TaskType: taskType, + Stage: "recovery_expired_running", + StatusBefore: "running", + StatusAfter: "failed", + Result: tenantapp.GenerationTaskResultSkipped, + Severity: "warning", + } + tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventRecoveryFinalize, logCtx) 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), - ) + tenantapp.LogGenerationTaskInfo(w.logger, "article generation expired task finalized because article is already failed", logCtx) } return true, nil } if err := w.publishGenerationTask(ctx, item.ID, taskType, tenantID, articleID); err != nil { return false, err } + logCtx := tenantapp.GenerationTaskLogContext{ + TaskID: item.ID, + TenantID: tenantID, + ArticleID: articleID, + TaskType: taskType, + Stage: "recovery_expired_running", + StatusBefore: "running", + StatusAfter: "queued", + Result: tenantapp.GenerationTaskResultSuccess, + Severity: "warning", + } + tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventRecoveryRequeue, logCtx) + tenantapp.LogGenerationTaskWarn(w.logger, "article generation expired task requeued", logCtx) return true, nil } @@ -672,6 +726,21 @@ func (w *ArticleGenerationWorker) finalizeExpiredGenerationTask(ctx context.Cont if item.ArticleID > 0 { tenantapp.InvalidateArticleCaches(cacheCtx, w.cache, item.TenantID, &item.ArticleID) } + logCtx := tenantapp.GenerationTaskLogContext{ + TaskID: item.ID, + TenantID: item.TenantID, + OperatorID: item.OperatorID, + ArticleID: item.ArticleID, + ReservationID: item.QuotaReservationID, + TaskType: item.TaskType, + Stage: "recovery_finalize", + StatusAfter: "failed", + Result: tenantapp.GenerationTaskResultSuccess, + Severity: "critical", + AttemptCount: item.AttemptCount, + } + tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventRecoveryFinalize, logCtx) + tenantapp.LogGenerationTaskWarn(w.logger, "article generation stale task finalized", logCtx) return true, nil } diff --git a/server/internal/worker/generate/article_generation_recovery_test.go b/server/internal/worker/generate/article_generation_recovery_test.go index 6fae88c..4fbd6b8 100644 --- a/server/internal/worker/generate/article_generation_recovery_test.go +++ b/server/internal/worker/generate/article_generation_recovery_test.go @@ -31,6 +31,20 @@ func TestNormalizeGenerationTaskRecoveryConfigUsesGenerationDefaults(t *testing. if cfg.MaxAttempts != 3 { t.Fatalf("expected max attempts 3, got %d", cfg.MaxAttempts) } + + stateCheck := normalizeGenerationTaskStateCheckConfig(config.GenerationConfig{}) + if stateCheck.Interval != 5*time.Minute { + t.Fatalf("expected state check interval 5m, got %s", stateCheck.Interval) + } + if stateCheck.Timeout != 10*time.Second { + t.Fatalf("expected state check timeout 10s, got %s", stateCheck.Timeout) + } + if stateCheck.BatchSize != 100 { + t.Fatalf("expected state check batch size 100, got %d", stateCheck.BatchSize) + } + if stateCheck.Lookback != 24*time.Hour { + t.Fatalf("expected state check lookback 24h, got %s", stateCheck.Lookback) + } } func TestNormalizeGenerationTaskRecoveryConfigKeepsOverrides(t *testing.T) { @@ -52,6 +66,19 @@ func TestNormalizeGenerationTaskRecoveryConfigKeepsOverrides(t *testing.T) { cfg.MaxAttempts != 5 { t.Fatalf("unexpected recovery config: %#v", cfg) } + + stateCheck := normalizeGenerationTaskStateCheckConfig(config.GenerationConfig{ + TaskStateCheckInterval: 9 * time.Minute, + TaskStateCheckTimeout: 11 * time.Second, + TaskStateCheckBatchSize: 13, + TaskStateCheckLookback: 2 * time.Hour, + }) + if stateCheck.Interval != 9*time.Minute || + stateCheck.Timeout != 11*time.Second || + stateCheck.BatchSize != 13 || + stateCheck.Lookback != 2*time.Hour { + t.Fatalf("unexpected state check config: %#v", stateCheck) + } } func TestGenerationWorkerOwnerIsBounded(t *testing.T) { diff --git a/server/internal/worker/generate/article_generation_state_check.go b/server/internal/worker/generate/article_generation_state_check.go new file mode 100644 index 0000000..f782de7 --- /dev/null +++ b/server/internal/worker/generate/article_generation_state_check.go @@ -0,0 +1,191 @@ +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" + } +} diff --git a/server/internal/worker/generate/article_generation_state_check_test.go b/server/internal/worker/generate/article_generation_state_check_test.go new file mode 100644 index 0000000..7c22220 --- /dev/null +++ b/server/internal/worker/generate/article_generation_state_check_test.go @@ -0,0 +1,67 @@ +package generate + +import ( + "database/sql" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type generationTaskStateFakeRow struct { + values []any +} + +func (r generationTaskStateFakeRow) Scan(dest ...any) error { + for idx := range dest { + switch target := dest[idx].(type) { + case *int64: + *target = r.values[idx].(int64) + case *int: + *target = r.values[idx].(int) + case *string: + *target = r.values[idx].(string) + case *sql.NullInt64: + *target = r.values[idx].(sql.NullInt64) + case *sql.NullString: + *target = r.values[idx].(sql.NullString) + } + } + return nil +} + +func TestScanGenerationTaskStateAnomalyNormalizesNullableFields(t *testing.T) { + item, err := scanGenerationTaskStateAnomaly(generationTaskStateFakeRow{values: []any{ + int64(11), + int64(22), + sql.NullInt64{Int64: 33, Valid: true}, + sql.NullInt64{Int64: 44, Valid: true}, + sql.NullInt64{Int64: 55, Valid: true}, + "template", + "completed", + sql.NullString{String: "generating", Valid: true}, + 2, + sql.NullString{String: "worker-1", Valid: true}, + generationTaskAnomalyCompletedTaskArticleMissing, + }}) + require.NoError(t, err) + + assert.Equal(t, int64(11), item.TaskID) + assert.Equal(t, int64(22), item.TenantID) + assert.Equal(t, int64(33), item.OperatorID) + assert.Equal(t, int64(44), item.ArticleID) + assert.Equal(t, int64(55), item.ReservationID) + assert.Equal(t, "template", item.TaskType) + assert.Equal(t, "completed", item.TaskStatus) + assert.Equal(t, "generating", item.ArticleStatus) + assert.Equal(t, 2, item.AttemptCount) + assert.Equal(t, "worker-1", item.LeaseOwner) + assert.Equal(t, "critical", item.Severity) +} + +func TestGenerationTaskAnomalySeverity(t *testing.T) { + assert.Equal(t, "critical", generationTaskAnomalySeverity(generationTaskAnomalyCompletedTaskMissingVersion)) + assert.Equal(t, "critical", generationTaskAnomalySeverity(generationTaskAnomalyArticleFailedTaskActive)) + assert.Equal(t, "warning", generationTaskAnomalySeverity(generationTaskAnomalyExpiredRunningLease)) + assert.Equal(t, "warning", generationTaskAnomalySeverity("unknown")) +} diff --git a/server/internal/worker/generate/article_generation_state_check_worker.go b/server/internal/worker/generate/article_generation_state_check_worker.go new file mode 100644 index 0000000..18b67a5 --- /dev/null +++ b/server/internal/worker/generate/article_generation_state_check_worker.go @@ -0,0 +1,98 @@ +package generate + +import ( + "context" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "go.uber.org/zap" + + "github.com/geo-platform/tenant-api/internal/shared/config" + tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app" +) + +type GenerationTaskStateCheckWorker struct { + pool *pgxpool.Pool + logger *zap.Logger + cfg config.GenerationConfig + provider config.Provider +} + +func NewGenerationTaskStateCheckWorker(pool *pgxpool.Pool, logger *zap.Logger, cfg config.GenerationConfig) *GenerationTaskStateCheckWorker { + return &GenerationTaskStateCheckWorker{ + pool: pool, + logger: logger, + cfg: cfg, + } +} + +func (w *GenerationTaskStateCheckWorker) WithConfigProvider(provider config.Provider) *GenerationTaskStateCheckWorker { + w.provider = provider + return w +} + +func (w *GenerationTaskStateCheckWorker) Start(ctx context.Context) { + go w.Run(ctx) +} + +func (w *GenerationTaskStateCheckWorker) Run(ctx context.Context) { + if w == nil || w.pool == nil { + return + } + w.run(ctx) +} + +func (w *GenerationTaskStateCheckWorker) run(ctx context.Context) { + w.runOnce(context.Background()) + + for { + cfg := w.runtimeConfig() + timer := time.NewTimer(cfg.Interval) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + w.runOnce(context.Background()) + } + } +} + +func (w *GenerationTaskStateCheckWorker) runOnce(parent context.Context) { + cfg := w.runtimeConfig() + startedAt := time.Now() + count, err := CheckGenerationTaskStateConsistency(parent, w.pool, w.logger, cfg) + duration := time.Since(startedAt) + if err != nil { + logCtx := tenantapp.GenerationTaskLogContext{ + Stage: "state_check", + Result: tenantapp.GenerationTaskResultFailure, + Severity: "warning", + Duration: duration, + } + tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventStateCheckFailure, logCtx) + tenantapp.LogGenerationTaskWarn(w.logger, "article generation state consistency check failed", logCtx, zap.Error(err)) + return + } + if count == 0 { + return + } + tenantapp.LogGenerationTaskWarn(w.logger, "article generation state consistency anomalies detected", tenantapp.GenerationTaskLogContext{ + Stage: "state_check", + Result: tenantapp.GenerationTaskResultSuccess, + Severity: "critical", + Duration: duration, + }, zap.Int("anomaly_count", count)) +} + +func (w *GenerationTaskStateCheckWorker) runtimeConfig() generationTaskStateCheckConfig { + if w != nil && w.provider != nil { + if cfg := w.provider.Current(); cfg != nil { + return normalizeGenerationTaskStateCheckConfig(cfg.Generation) + } + } + if w != nil { + return normalizeGenerationTaskStateCheckConfig(w.cfg) + } + return normalizeGenerationTaskStateCheckConfig(config.GenerationConfig{}) +} diff --git a/server/internal/worker/generate/article_generation_worker.go b/server/internal/worker/generate/article_generation_worker.go index a08afd8..1a52088 100644 --- a/server/internal/worker/generate/article_generation_worker.go +++ b/server/internal/worker/generate/article_generation_worker.go @@ -180,6 +180,10 @@ func (w *ArticleGenerationWorker) handleDelivery(parent context.Context, deliver } task := lease.task.toClaimedTask() + owner := generationWorkerOwner(delivery.ConsumerTag) + claimLogCtx := generationWorkerTaskLogContext(task, "worker_claim", "queued", tenantapp.GenerationTaskStatusRunning, tenantapp.GenerationTaskResultSuccess, lease.task.AttemptCount, owner) + tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventTransition, claimLogCtx) + tenantapp.LogGenerationTaskInfo(w.logger, "article generation task claimed", claimLogCtx) heartbeatCtx, stopHeartbeat := context.WithCancel(parent) go w.runGenerationLeaseHeartbeat(heartbeatCtx, task.ID, lease.token, recoveryCfg.LeaseTTL) defer func() { @@ -245,6 +249,9 @@ 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) + logCtx := generationWorkerTaskLogContext(task, stage, "", "failed", tenantapp.GenerationTaskResultFailure, 0, "") + tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventFailure, logCtx) + tenantapp.LogGenerationTaskWarn(w.logger, "article generation task failed", logCtx, zap.Error(taskErr)) var err error switch { case task.TaskType == kolGenerationTaskType && w.kolWorker != nil: @@ -255,13 +262,8 @@ func (w *ArticleGenerationWorker) handleTaskFailure(ctx context.Context, task te 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), - ) + tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventCleanupFailure, logCtx) + tenantapp.LogGenerationTaskError(w.logger, "article generation task failure handling failed", logCtx, zap.Error(err)) } } @@ -280,15 +282,39 @@ func (w *ArticleGenerationWorker) rejectDelivery(delivery amqp.Delivery, requeue ) } if w.logger != nil { - w.logger.Warn("article generation processing failed", + logCtx := tenantapp.GenerationTaskLogContext{ + TaskID: taskID, + Stage: "delivery_reject", + Result: tenantapp.GenerationTaskResultFailure, + Severity: "warning", + StatusAfter: "rejected", + } + tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventQueueFailure, logCtx) + tenantapp.LogGenerationTaskWarn(w.logger, "article generation processing failed", logCtx, zap.Error(err), zap.Bool("requeue", requeue), zap.Bool("redelivered", delivery.Redelivered), - zap.Int64("task_id", taskID), ) } } +func generationWorkerTaskLogContext(task tenantapp.ClaimedGenerationTask, stage, statusBefore, statusAfter, result string, attemptCount int, leaseOwner string) tenantapp.GenerationTaskLogContext { + return tenantapp.GenerationTaskLogContext{ + TaskID: task.ID, + TenantID: task.TenantID, + OperatorID: task.OperatorID, + ArticleID: task.ArticleID, + ReservationID: task.QuotaReservationID, + TaskType: task.TaskType, + Stage: stage, + AttemptCount: attemptCount, + LeaseOwner: leaseOwner, + StatusBefore: statusBefore, + StatusAfter: statusAfter, + Result: result, + } +} + func parseJSONMap(raw []byte) (map[string]interface{}, error) { if len(raw) == 0 { return map[string]interface{}{}, nil diff --git a/server/migrations/20260505213000_generation_task_state_check_indexes.down.sql b/server/migrations/20260505213000_generation_task_state_check_indexes.down.sql new file mode 100644 index 0000000..0b8201d --- /dev/null +++ b/server/migrations/20260505213000_generation_task_state_check_indexes.down.sql @@ -0,0 +1,2 @@ +DROP INDEX IF EXISTS idx_generation_tasks_recent_completed_state_check; +DROP INDEX IF EXISTS idx_generation_tasks_active_state_check; diff --git a/server/migrations/20260505213000_generation_task_state_check_indexes.up.sql b/server/migrations/20260505213000_generation_task_state_check_indexes.up.sql new file mode 100644 index 0000000..e092697 --- /dev/null +++ b/server/migrations/20260505213000_generation_task_state_check_indexes.up.sql @@ -0,0 +1,9 @@ +CREATE INDEX IF NOT EXISTS idx_generation_tasks_active_state_check + ON generation_tasks(updated_at DESC, id DESC) + WHERE article_id IS NOT NULL + AND status IN ('queued', 'running'); + +CREATE INDEX IF NOT EXISTS idx_generation_tasks_recent_completed_state_check + ON generation_tasks(updated_at DESC, id DESC) + WHERE article_id IS NOT NULL + AND status = 'completed'; diff --git a/task_plan.md b/task_plan.md index 7b8431f..f29b505 100644 --- a/task_plan.md +++ b/task_plan.md @@ -4,7 +4,7 @@ Continue the desktop AI monitoring implementation by first adding a client-side local scheduler with durable queueing and adaptive execution policy, then adding a reusable hidden Playwright CDP execution layer that attaches to Electron Chromium and reuses existing account partitions. ## Current Phase -Phase 41 +Phase 45 ## Phases ### Phase 1: Progress Verification @@ -303,6 +303,18 @@ Phase 41 - [x] Update findings/progress and summarize completed functions plus any deferred risk - **Status:** complete +### Phase 44: Article Generation Reliability Hardening +- [x] Add structured generation task logs with stable IDs, stages, transitions, retry/recovery context, and error causes +- [x] Add generation state consistency checks plus alert-ready metrics for task/article/version mismatches +- [x] Stop swallowing generation status/cleanup errors in worker-facing paths +- **Status:** complete + +### Phase 45: Article Generation Reliability Verification +- [x] Add targeted unit tests for log fields, metrics, and anomaly classification +- [x] Run targeted Go tests for tenant generation and worker recovery packages +- [x] Commit only the reliability hardening files after verification +- **Status:** complete + ## Key Questions 1. How should the desktop client defer and locally optimize monitor-task execution without violating one-task-per-platform serialism? 2. What is the smallest durable local cache that allows same-day resume while safely dropping stale next-day monitor tasks? @@ -356,6 +368,8 @@ Phase 41 | Start k3s with a self-contained stack that mirrors the existing compose deployment | The repo already has working Docker images and compose topology; k3s should first preserve service names and runtime assumptions before introducing managed cloud replacements | | Publish deployable images only from frontend/backend CI | Deployment config has no runtime image ownership; `deploy-config-ci` stays a validation workflow while frontend/backend CI publish Gitea Registry images tagged by commit8 | | Use commit hash first 8 chars as the canonical deploy image tag | CD/offline packaging can compare the pushed commit to Registry tags and fail fast when CI has not produced the matching image | +| Treat article generation DB state scanning as a low-frequency safety net | Main reliability comes from write-path validation, structured errors, metrics, and alerts; the scheduler check is bounded by interval, timeout, lookback, batch size, and partial indexes so it does not become a DB polling dependency | +| Expose article generation alert signals through Prometheus metrics and rules | Cleanup failures, state anomalies, state-check failures, queue failures, failure spikes, and recovery finalization should page operators without relying on humans tailing logs | ## Errors Encountered | Error | Attempt | Resolution | @@ -371,6 +385,8 @@ Phase 41 | `pnpm --filter @geo/desktop-client typecheck` fails in `electron.vite.config.ts` due to a pre-existing Vite 5/8 plugin type mismatch | 1 | Verified this change with `pnpm --filter @geo/desktop-client build` and `pnpm --filter @geo/desktop-client test`; leave config dependency alignment as a separate fix | | `kubectl apply --dry-run=client` still attempted to contact the local kube-apiserver, which is not running at `127.0.0.1:26443` | 1 | Verified with `kubectl kustomize`, parsed the rendered YAML locally, and recorded that live API validation should run on the target k3s cluster | | Production compose PgBouncer smoke test could not start bundled Postgres containers because existing `server/docker-compose.yaml` containers already own `geo-postgres` / `geo-monitoring-postgres` names | 1 | Did not stop user/local DB containers; verified PgBouncer with a temporary test container attached to the existing `server_default` network and removed it after `SELECT 1` passed | +| Article generation state-check worker initially failed to compile because the local DB interface returned a custom rows type instead of `pgx.Rows` | 1 | Changed the interface to match `pgxpool.Pool.Query` directly and reran targeted/full Go tests | +| Prometheus last-event test depended on label output order | 1 | Changed the assertion to validate required labels independent of exporter ordering | ## Notes - Re-check task_plan.md before major implementation decisions.