root
cbbed4b42c
The state-consistency worker now detects articles still flagged as generating whose latest generation_task is already failed or completed (with a version) and aligns the article status to the terminal task, invalidating article caches on the way out. Adds supporting partial indexes and switches UpdateGenerationTaskStatus to COALESCE started_at and completed_at so partial status updates don't wipe the timestamps.
343 lines
10 KiB
Go
343 lines
10 KiB
Go
package generate
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"time"
|
|
|
|
sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache"
|
|
"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"
|
|
generationTaskAnomalyTerminalTaskArticleActive = "terminal_task_article_generating"
|
|
)
|
|
|
|
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, cache sharedcache.Cache, 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
|
|
}
|
|
repaired, err := repairGenerationTaskArticleState(ctx, pool, logger, cache, cfg, lookback)
|
|
if err != nil {
|
|
return count, err
|
|
}
|
|
count += repaired
|
|
return count, nil
|
|
}
|
|
|
|
func repairGenerationTaskArticleState(ctx context.Context, pool generationTaskStateDB, logger *zap.Logger, cache sharedcache.Cache, cfg generationTaskStateCheckConfig, lookback time.Time) (int, error) {
|
|
if pool == nil || cfg.BatchSize <= 0 {
|
|
return 0, nil
|
|
}
|
|
rows, err := pool.Query(ctx, `
|
|
WITH recent_terminal AS (
|
|
SELECT
|
|
gt.id,
|
|
gt.tenant_id,
|
|
gt.operator_id,
|
|
gt.article_id,
|
|
gt.quota_reservation_id,
|
|
gt.task_type,
|
|
gt.status AS task_status,
|
|
gt.error_message,
|
|
gt.attempt_count,
|
|
gt.lease_owner,
|
|
gt.created_at,
|
|
gt.updated_at
|
|
FROM generation_tasks gt
|
|
WHERE gt.article_id IS NOT NULL
|
|
AND gt.status IN ('failed', 'completed')
|
|
AND gt.updated_at >= $1
|
|
ORDER BY gt.updated_at DESC, gt.id DESC
|
|
LIMIT ($2::int * 5)
|
|
),
|
|
latest_terminal AS (
|
|
SELECT DISTINCT ON (rt.article_id)
|
|
rt.id,
|
|
rt.tenant_id,
|
|
rt.operator_id,
|
|
rt.article_id,
|
|
rt.quota_reservation_id,
|
|
rt.task_type,
|
|
rt.task_status,
|
|
rt.error_message,
|
|
rt.attempt_count,
|
|
rt.lease_owner,
|
|
rt.created_at,
|
|
rt.updated_at
|
|
FROM recent_terminal rt
|
|
ORDER BY rt.article_id, rt.created_at DESC, rt.id DESC
|
|
),
|
|
candidates AS (
|
|
SELECT
|
|
lt.id,
|
|
lt.tenant_id,
|
|
lt.operator_id,
|
|
lt.article_id,
|
|
lt.quota_reservation_id,
|
|
lt.task_type,
|
|
lt.task_status,
|
|
a.generate_status AS article_status,
|
|
lt.attempt_count,
|
|
lt.lease_owner
|
|
FROM latest_terminal lt
|
|
JOIN articles a
|
|
ON a.id = lt.article_id
|
|
AND a.tenant_id = lt.tenant_id
|
|
AND a.deleted_at IS NULL
|
|
WHERE a.generate_status = 'generating'
|
|
AND (
|
|
lt.task_status = 'failed'
|
|
OR (lt.task_status = 'completed' AND a.current_version_id IS NOT NULL)
|
|
)
|
|
AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM generation_tasks newer
|
|
WHERE newer.article_id = lt.article_id
|
|
AND newer.tenant_id = lt.tenant_id
|
|
AND (newer.created_at > lt.created_at OR (newer.created_at = lt.created_at AND newer.id > lt.id))
|
|
)
|
|
ORDER BY lt.updated_at DESC, lt.id DESC
|
|
LIMIT $2
|
|
),
|
|
updated AS (
|
|
UPDATE articles a
|
|
SET generate_status = c.task_status,
|
|
updated_at = NOW()
|
|
FROM candidates c
|
|
WHERE a.id = c.article_id
|
|
AND a.tenant_id = c.tenant_id
|
|
AND a.generate_status = 'generating'
|
|
RETURNING
|
|
c.id,
|
|
c.tenant_id,
|
|
c.operator_id,
|
|
c.article_id,
|
|
c.quota_reservation_id,
|
|
c.task_type,
|
|
c.task_status,
|
|
c.article_status,
|
|
c.attempt_count,
|
|
c.lease_owner,
|
|
'terminal_task_article_generating'::text AS anomaly_type
|
|
)
|
|
SELECT id, tenant_id, operator_id, article_id, quota_reservation_id, task_type, task_status, article_status, attempt_count, lease_owner, anomaly_type
|
|
FROM updated
|
|
`, lookback, cfg.BatchSize)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
count := 0
|
|
for rows.Next() {
|
|
item, err := scanGenerationTaskStateAnomaly(rows)
|
|
if err != nil {
|
|
return count, err
|
|
}
|
|
item.AnomalyType = generationTaskAnomalyTerminalTaskArticleActive
|
|
item.Severity = generationTaskAnomalySeverity(item.AnomalyType)
|
|
count++
|
|
logCtx := tenantapp.GenerationTaskLogContext{
|
|
TaskID: item.TaskID,
|
|
TenantID: item.TenantID,
|
|
OperatorID: item.OperatorID,
|
|
ArticleID: item.ArticleID,
|
|
ReservationID: item.ReservationID,
|
|
TaskType: item.TaskType,
|
|
Stage: "state_repair",
|
|
AttemptCount: item.AttemptCount,
|
|
LeaseOwner: item.LeaseOwner,
|
|
StatusBefore: item.ArticleStatus,
|
|
StatusAfter: item.TaskStatus,
|
|
Result: tenantapp.GenerationTaskResultSuccess,
|
|
AnomalyType: item.AnomalyType,
|
|
Severity: item.Severity,
|
|
}
|
|
tenantapp.RecordGenerationTaskEvent(tenantapp.GenerationTaskEventStateAnomaly, logCtx)
|
|
tenantapp.LogGenerationTaskWarn(logger, "article generation state repaired from latest terminal task", logCtx)
|
|
stateRepairCacheInvalidator(ctx, cache, item.TenantID, &item.ArticleID)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return count, err
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
type generationTaskStateAnomalyRow interface {
|
|
Scan(dest ...any) error
|
|
}
|
|
|
|
type generationTaskStateDB interface {
|
|
Query(context.Context, string, ...any) (pgx.Rows, error)
|
|
}
|
|
|
|
var stateRepairCacheInvalidator = func(ctx context.Context, c sharedcache.Cache, tenantID int64, articleID *int64) {
|
|
tenantapp.InvalidateArticleCaches(ctx, c, tenantID, articleID)
|
|
}
|
|
|
|
func scanGenerationTaskStateAnomaly(row generationTaskStateAnomalyRow) (generationTaskStateAnomaly, error) {
|
|
var (
|
|
item generationTaskStateAnomaly
|
|
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,
|
|
generationTaskAnomalyTerminalTaskArticleActive:
|
|
return "critical"
|
|
case generationTaskAnomalyArticleCompletedTaskActive,
|
|
generationTaskAnomalyExpiredRunningLease:
|
|
return "warning"
|
|
default:
|
|
return "warning"
|
|
}
|
|
}
|