feat(publish): cancel stale queued publish tasks after queue timeout

Add CancelStaleQueuedPublishTasks to abort publish tasks that sit in the queue past a configurable timeout (default 3 days) without being claimed by a desktop client, so they no longer linger indefinitely as 'queued'. The lease recovery worker runs the cleanup each cycle and reports a cancelled_queued count; the timeout is configurable via job run config. Aborted tasks carry a publish_queue_timeout error payload with a readable Chinese duration, and the admin-web publish summary surfaces it as an auto-cancelled queue-timeout message. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-08 22:56:37 +08:00
parent f1334aac80
commit 832d7f507d
5 changed files with 358 additions and 19 deletions
@@ -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 {
@@ -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()