diff --git a/apps/admin-web/src/views/PublishManagementView.vue b/apps/admin-web/src/views/PublishManagementView.vue index 1b727b7..09703d9 100644 --- a/apps/admin-web/src/views/PublishManagementView.vue +++ b/apps/admin-web/src/views/PublishManagementView.vue @@ -118,6 +118,7 @@ function inlineErrorPreview(messageText: string): string { function summaryForRecord(record: PublishRecord, status: string): string { const targetLabel = record.target_type === 'enterprise_site' ? '企业站点' : '目标平台' + const errorMessage = normalizeRecordErrorMessage(record.error_message, record.platform_id) switch (status) { case 'queued': return record.target_type === 'enterprise_site' @@ -134,6 +135,9 @@ function summaryForRecord(record: PublishRecord, status: string): string { ? '本次企业站发布失败，请检查站点连接或发布错误。' : '本次发送失败，请检查账号状态或平台提示后重新提交。' case 'cancelled': + if (errorMessage && /等待通道|publish_queue_timeout/.test(errorMessage)) { + return '等待通道超时，系统已自动取消。' + } return '本次发布已取消。' case 'unknown': default: diff --git a/server/internal/scheduler/publish_lease_recovery_worker.go b/server/internal/scheduler/publish_lease_recovery_worker.go index 94f1472..36fba46 100644 --- a/server/internal/scheduler/publish_lease_recovery_worker.go +++ b/server/internal/scheduler/publish_lease_recovery_worker.go @@ -13,26 +13,36 @@ const ( defaultPublishLeaseRecoveryInterval = 3 * time.Minute defaultPublishLeaseRecoveryTimeout = 30 * time.Second defaultPublishLeaseRecoveryLimit = 1000 + defaultPublishQueuedTaskTimeout = 3 * 24 * time.Hour ) type PublishLeaseRecoveryWorker struct { - service *tenantapp.DesktopTaskService - logger *zap.Logger - interval time.Duration - timeout time.Duration - limit int + service *tenantapp.DesktopTaskService + logger *zap.Logger + interval time.Duration + timeout time.Duration + limit int + queuedTaskTimeout time.Duration } func NewPublishLeaseRecoveryWorker(service *tenantapp.DesktopTaskService, logger *zap.Logger) *PublishLeaseRecoveryWorker { return &PublishLeaseRecoveryWorker{ - service: service, - logger: logger, - interval: defaultPublishLeaseRecoveryInterval, - timeout: defaultPublishLeaseRecoveryTimeout, - limit: defaultPublishLeaseRecoveryLimit, + service: service, + logger: logger, + interval: defaultPublishLeaseRecoveryInterval, + timeout: defaultPublishLeaseRecoveryTimeout, + limit: defaultPublishLeaseRecoveryLimit, + queuedTaskTimeout: defaultPublishQueuedTaskTimeout, } } +func (w *PublishLeaseRecoveryWorker) WithQueuedTaskTimeout(timeout time.Duration) *PublishLeaseRecoveryWorker { + if w != nil && timeout > 0 { + w.queuedTaskTimeout = timeout + } + return w +} + func (w *PublishLeaseRecoveryWorker) Start(ctx context.Context) { go w.Run(ctx) } @@ -72,15 +82,40 @@ func (w *PublishLeaseRecoveryWorker) runOnce(parent context.Context) { return } - if (result.Requeued > 0 || result.Failed > 0 || result.Uncertain > 0) && w.logger != nil { + queuedTaskTimeout := w.queuedTaskTimeout + if queuedTaskTimeout <= 0 { + queuedTaskTimeout = defaultPublishQueuedTaskTimeout + } + cancelledQueued, err := w.service.CancelStaleQueuedPublishTasks(ctx, queuedTaskTimeout, w.limit) + if err != nil { + if w.logger != nil { + w.logger.Warn("publish queued task timeout cleanup failed", zap.Error(err)) + } + return + } + result.CancelledQueued = cancelledQueued + + if (result.Requeued > 0 || result.Failed > 0 || result.Uncertain > 0 || result.CancelledQueued > 0) && w.logger != nil { w.logger.Info("publish lease recovery completed", zap.Int("requeued_task_count", result.Requeued), zap.Int("failed_task_count", result.Failed), zap.Int("uncertain_task_count", result.Uncertain), + zap.Int("cancelled_queued_task_count", result.CancelledQueued), ) } } +func publishQueuedTaskTimeoutFromRun(run JobRunContext) time.Duration { + timeout := AsDuration(run.Config, "queued_task_timeout", defaultPublishQueuedTaskTimeout) + if seconds := AsInt(run.Config, "queued_task_timeout_seconds", 0); seconds > 0 { + timeout = time.Duration(seconds) * time.Second + } + if timeout <= 0 { + return defaultPublishQueuedTaskTimeout + } + return timeout +} + func (w *PublishLeaseRecoveryWorker) RunOnce(parent context.Context, run JobRunContext) (map[string]any, error) { if w == nil || w.service == nil { return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil @@ -104,11 +139,19 @@ func (w *PublishLeaseRecoveryWorker) RunOnce(parent context.Context, run JobRunC if err != nil { return map[string]any{"stage": "recover"}, err } + queuedTimeout := publishQueuedTaskTimeoutFromRun(run) + cancelledQueued, err := w.service.CancelStaleQueuedPublishTasks(ctx, queuedTimeout, limit) + if err != nil { + return map[string]any{"stage": "cancel_stale_queued"}, err + } + result.CancelledQueued = cancelledQueued return map[string]any{ - "requeued_task_count": result.Requeued, - "failed_task_count": result.Failed, - "uncertain_task_count": result.Uncertain, - "duration_ms": time.Since(startedAt).Milliseconds(), - "batch_size": limit, + "requeued_task_count": result.Requeued, + "failed_task_count": result.Failed, + "uncertain_task_count": result.Uncertain, + "cancelled_queued_task_count": result.CancelledQueued, + "queued_task_timeout_seconds": int64(queuedTimeout.Seconds()), + "duration_ms": time.Since(startedAt).Milliseconds(), + "batch_size": limit, }, nil } diff --git a/server/internal/scheduler/publish_lease_recovery_worker_test.go b/server/internal/scheduler/publish_lease_recovery_worker_test.go new file mode 100644 index 0000000..3982b4a --- /dev/null +++ b/server/internal/scheduler/publish_lease_recovery_worker_test.go @@ -0,0 +1,42 @@ +package scheduler + +import ( + "testing" + "time" +) + +func TestPublishQueuedTaskTimeoutFromRunDefaultsToThreeDays(t *testing.T) { + t.Parallel() + + got := publishQueuedTaskTimeoutFromRun(JobRunContext{}) + if got != 3*24*time.Hour { + t.Fatalf("publishQueuedTaskTimeoutFromRun() = %s, want 72h", got) + } +} + +func TestPublishQueuedTaskTimeoutFromRunReadsDurationConfig(t *testing.T) { + t.Parallel() + + got := publishQueuedTaskTimeoutFromRun(JobRunContext{ + Config: map[string]any{ + "queued_task_timeout": "72h", + }, + }) + if got != 72*time.Hour { + t.Fatalf("publishQueuedTaskTimeoutFromRun(duration) = %s, want 72h", got) + } +} + +func TestPublishQueuedTaskTimeoutFromRunReadsSecondsConfig(t *testing.T) { + t.Parallel() + + got := publishQueuedTaskTimeoutFromRun(JobRunContext{ + Config: map[string]any{ + "queued_task_timeout": "72h", + "queued_task_timeout_seconds": 3600, + }, + }) + if got != time.Hour { + t.Fatalf("publishQueuedTaskTimeoutFromRun(seconds) = %s, want 1h", got) + } +} diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 425ba6d..aa99acb 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -1987,9 +1987,10 @@ type recoveredDesktopTaskLease struct { } type PublishLeaseRecoveryResult struct { - Requeued int `json:"requeued"` - Failed int `json:"failed"` - Uncertain int `json:"uncertain"` + Requeued int `json:"requeued"` + Failed int `json:"failed"` + Uncertain int `json:"uncertain"` + CancelledQueued int `json:"cancelled_queued"` } type MonitorLeaseRecoveryResult struct { @@ -2456,6 +2457,189 @@ func (s *DesktopTaskService) RecoverExpiredPublishTasks(ctx context.Context, lim return result, nil } +func staleQueuedPublishTasksCandidateSQL() string { + return ` + SELECT + t.desktop_id, + t.workspace_id + FROM desktop_tasks AS t + JOIN desktop_publish_jobs AS j + ON j.desktop_id = t.job_id + AND j.tenant_id = t.tenant_id + AND j.workspace_id = t.workspace_id + WHERE t.kind = 'publish' + AND t.status = 'queued' + AND GREATEST( + COALESCE(t.enqueued_at, t.created_at), + COALESCE(j.scheduled_at, t.created_at) + ) < $1 + ORDER BY COALESCE(t.enqueued_at, t.created_at) ASC, + t.created_at ASC, + t.desktop_id ASC + LIMIT $2 + FOR UPDATE OF t SKIP LOCKED + ` +} + +func publishQueueTimeoutErrorPayload(timeout time.Duration) ([]byte, error) { + timeoutSeconds := int64(timeout.Seconds()) + if timeoutSeconds < 0 { + timeoutSeconds = 0 + } + message := fmt.Sprintf( + "发布任务在等待通道中超过 %s未被桌面端领取，系统已自动取消，避免任务无限期停留。请确认对应桌面端在线、账号登录正常且任务通道在线后，再重新发布。", + publishQueueTimeoutDisplayDuration(timeout), + ) + return marshalOptionalJSON(map[string]any{ + "code": "publish_queue_timeout", + "message": message, + "source": "publish_queue_timeout", + "queued_timeout_seconds": timeoutSeconds, + "queued_timeout_duration": timeout.String(), + }) +} + +func publishQueueTimeoutDisplayDuration(timeout time.Duration) string { + if timeout <= 0 { + return "配置的时间" + } + minutes := int64(timeout.Round(time.Minute) / time.Minute) + if minutes <= 0 { + return "1 分钟" + } + const ( + minutesPerHour = int64(60) + minutesPerDay = int64(24 * 60) + ) + if minutes%minutesPerDay == 0 { + return fmt.Sprintf("%d 天", minutes/minutesPerDay) + } + if minutes%minutesPerHour == 0 { + return fmt.Sprintf("%d 小时", minutes/minutesPerHour) + } + if minutes > minutesPerHour { + return fmt.Sprintf("%d 小时 %d 分钟", minutes/minutesPerHour, minutes%minutesPerHour) + } + return fmt.Sprintf("%d 分钟", minutes) +} + +func (s *DesktopTaskService) CancelStaleQueuedPublishTasks(ctx context.Context, queuedFor time.Duration, limit int) (int, error) { + if s == nil || s.pool == nil || queuedFor <= 0 { + return 0, nil + } + if limit <= 0 { + limit = 1000 + } + errorJSON, err := publishQueueTimeoutErrorPayload(queuedFor) + if err != nil { + return 0, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish queue timeout payload") + } + staleBefore := time.Now().UTC().Add(-queuedFor) + + tx, err := s.pool.Begin(ctx) + if err != nil { + return 0, response.ErrInternal(50196, "desktop_task_recovery_begin_failed", "failed to start desktop task recovery") + } + defer tx.Rollback(ctx) + + rows, err := tx.Query(ctx, staleQueuedPublishTasksCandidateSQL(), staleBefore, limit) + if err != nil { + s.logWarn("stale queued publish task query failed", err) + return 0, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select stale queued publish tasks") + } + defer rows.Close() + + candidates := make([]recoveredDesktopTaskLease, 0) + for rows.Next() { + var item recoveredDesktopTaskLease + if scanErr := rows.Scan(&item.TaskID, &item.WorkspaceID); scanErr != nil { + s.logWarn("stale queued publish task scan failed", scanErr) + return 0, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse stale queued publish task") + } + item.Kind = "publish" + item.Status = "aborted" + item.ErrorJSON = errorJSON + candidates = append(candidates, item) + } + if err := rows.Err(); err != nil { + s.logWarn("stale queued publish task rows failed", err) + return 0, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate stale queued publish tasks") + } + + publishOutcomes := make([]*desktopPublishSyncOutcome, 0, len(candidates)) + cancelled := 0 + for index := range candidates { + item := &candidates[index] + row := tx.QueryRow(ctx, ` + UPDATE desktop_tasks + SET status = 'aborted', + error = $2, + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + interrupted_at = COALESCE(interrupted_at, NOW()), + interrupt_reason = COALESCE(interrupt_reason, 'publish_queue_timeout'), + updated_at = NOW() + WHERE desktop_id = $1 + AND kind = 'publish' + AND status = 'queued' + RETURNING desktop_id, tenant_id, workspace_id, platform_id, kind, payload, status, result, error + `, item.TaskID, errorJSON) + + var updated repository.DesktopTask + if scanErr := row.Scan( + &updated.DesktopID, + &updated.TenantID, + &updated.WorkspaceID, + &updated.Platform, + &updated.Kind, + &updated.Payload, + &updated.Status, + &updated.Result, + &updated.Error, + ); scanErr != nil { + if errors.Is(scanErr, pgx.ErrNoRows) { + continue + } + s.logWarn("stale queued publish task update failed", scanErr, zap.String("task_id", item.TaskID.String())) + return cancelled, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to cancel stale queued publish task") + } + + publishOutcome, syncErr := syncDesktopPublishTaskState(ctx, tx, &updated) + if syncErr != nil { + return cancelled, syncErr + } + if publishOutcome != nil { + publishOutcomes = append(publishOutcomes, publishOutcome) + } + cancelled++ + } + + if err := tx.Commit(ctx); err != nil { + return cancelled, response.ErrInternal(50199, "desktop_task_recovery_commit_failed", "failed to commit desktop task recovery") + } + + s.afterRecoveredPublishOutcomes(ctx, publishOutcomes) + + for _, item := range candidates { + task, getErr := s.repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID) + if getErr != nil { + s.logWarn("stale queued publish task reload after commit failed", getErr, zap.String("task_id", item.TaskID.String())) + continue + } + s.publishTaskEvent(ctx, task, "task_completed") + } + + if cancelled > 0 && s.logger != nil { + s.logger.Info("stale queued publish tasks cancelled", + zap.Int("cancelled", cancelled), + zap.Duration("queued_for", queuedFor), + ) + } + + return cancelled, nil +} + func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, limit int) (MonitorLeaseRecoveryResult, error) { var result MonitorLeaseRecoveryResult if s == nil || s.pool == nil { diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index 70e5906..33a69e3 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/server/internal/tenant/app/desktop_task_service_test.go @@ -6,6 +6,7 @@ import ( "regexp" "strings" "testing" + "time" "github.com/geo-platform/tenant-api/internal/shared/response" "github.com/geo-platform/tenant-api/internal/tenant/repository" @@ -53,6 +54,71 @@ func TestDesktopTaskRecoverySelectQueryLeaseExpiryFiltersExpiredLeases(t *testin } } +func TestStaleQueuedPublishTasksCandidateSQLUsesExecutableQueueAge(t *testing.T) { + t.Parallel() + + query := normalizeSQLForDesktopTaskTest(staleQueuedPublishTasksCandidateSQL()) + for _, fragment := range []string{ + "FROM desktop_tasks AS t", + "JOIN desktop_publish_jobs AS j", + "t.kind = 'publish'", + "t.status = 'queued'", + "GREATEST( COALESCE(t.enqueued_at, t.created_at), COALESCE(j.scheduled_at, t.created_at) ) < $1", + "LIMIT $2", + "FOR UPDATE OF t SKIP LOCKED", + } { + if !strings.Contains(query, fragment) { + t.Fatalf("stale queued publish query missing %q: %s", fragment, query) + } + } +} + +func TestPublishQueueTimeoutErrorPayloadIncludesConfig(t *testing.T) { + t.Parallel() + + payload, err := publishQueueTimeoutErrorPayload(3 * 24 * time.Hour) + if err != nil { + t.Fatalf("publishQueueTimeoutErrorPayload() error = %v", err) + } + text := string(payload) + for _, fragment := range []string{ + `"code":"publish_queue_timeout"`, + `发布任务在等待通道中超过 3 天未被桌面端领取`, + `请确认对应桌面端在线、账号登录正常且任务通道在线后，再重新发布。`, + `"source":"publish_queue_timeout"`, + `"queued_timeout_seconds":259200`, + } { + if !strings.Contains(text, fragment) { + t.Fatalf("queue timeout payload missing %q: %s", fragment, text) + } + } +} + +func TestPublishQueueTimeoutDisplayDurationUsesReadableChinese(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + timeout time.Duration + want string + }{ + {name: "days", timeout: 7 * 24 * time.Hour, want: "7 天"}, + {name: "hours", timeout: 36 * time.Hour, want: "36 小时"}, + {name: "hours and minutes", timeout: 90 * time.Minute, want: "1 小时 30 分钟"}, + {name: "minutes", timeout: 45 * time.Minute, want: "45 分钟"}, + } + + for _, tc := range cases { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := publishQueueTimeoutDisplayDuration(tc.timeout); got != tc.want { + t.Fatalf("publishQueueTimeoutDisplayDuration(%s) = %q, want %q", tc.timeout, got, tc.want) + } + }) + } +} + func TestReconcileRetryPublishesTaskAvailableForAllDesktopTaskKinds(t *testing.T) { t.Parallel()