From ed48674ab5b50e604250eed8326affa05cd40462 Mon Sep 17 00:00:00 2001 From: liangxu Date: Thu, 25 Jun 2026 00:22:59 +0800 Subject: [PATCH] fix: allow retry after definitive publish failures --- .../src/main/publish-result-classification.ts | 72 ++++++++ .../runtime-controller-publish-retry.test.ts | 38 ++++ .../src/main/runtime-controller.ts | 77 +++----- .../renderer/views/PublishManagementView.vue | 6 +- packages/shared-types/src/index.ts | 1 + .../tenant/app/desktop_task_service.go | 95 +++++----- .../tenant/app/desktop_task_service_test.go | 6 +- .../tenant/app/publish_job_service.go | 173 +++++++++++++++++- .../tenant/app/publish_job_service_test.go | 29 +++ .../tenant/repository/desktop_task_repo.go | 126 ++++++------- .../repository/generated/desktop_task.sql.go | 4 + .../generated/desktop_task_sql_test.go | 4 + .../repository/queries/desktop_task.sql | 4 + 13 files changed, 473 insertions(+), 162 deletions(-) create mode 100644 apps/desktop-client/src/main/publish-result-classification.ts create mode 100644 apps/desktop-client/src/main/runtime-controller-publish-retry.test.ts diff --git a/apps/desktop-client/src/main/publish-result-classification.ts b/apps/desktop-client/src/main/publish-result-classification.ts new file mode 100644 index 0000000..8d9ee3b --- /dev/null +++ b/apps/desktop-client/src/main/publish-result-classification.ts @@ -0,0 +1,72 @@ +import type { JsonValue } from '@geo/shared-types' + +import type { AdapterExecutionResult } from './adapters/base' + +export function markAuthorizationFailureNonRetryable( + error: Record, + category: 'ai_platform_authorization' | 'media_account_authorization', +): Record { + return { + ...error, + retryable: false, + non_retryable: true, + failure_category: category, + } +} + +export function isAuthorizationFailureCode(code: string | null | undefined): boolean { + if (!code) { + return false + } + const normalized = code.trim().toLowerCase() + return ( + normalized === 'desktop_account_auth_expired' || + normalized === 'desktop_account_challenge_required' || + normalized === 'desktop_account_risk_control' || + normalized.endsWith('_not_logged_in') || + normalized.endsWith('_login_required') || + normalized.endsWith('_login_expired') || + normalized.endsWith('_challenge_required') + ) +} + +export function isDefinitivePublishFailure(result: AdapterExecutionResult): boolean { + if (result.status !== 'failed') { + return false + } + + const code = typeof result.error?.code === 'string' ? result.error.code.toLowerCase() : '' + const message = + typeof result.error?.message === 'string' ? result.error.message.toLowerCase() : '' + const summary = result.summary.toLowerCase() + const combined = `${code} ${message} ${summary}` + + if ( + result.error?.non_retryable === true || + result.error?.retryable === false || + isAuthorizationFailureCode(code) + ) { + return true + } + + return /(?:csrf|forbidden|\b403\b|not_logged_in|login_required|token_missing|cookie_missing|challenge_required|risk_control|captcha|验证码|滑块|人机验证|安全验证|风控|风险|参数错误|内容为空|image_fetch_failed|cover_fetch_failed|upload_failed|signature_failed)/i.test( + combined, + ) +} + +export function asSubmitUncertainExecution(result: AdapterExecutionResult): AdapterExecutionResult { + if (result.status !== 'failed' || isDefinitivePublishFailure(result)) { + return result + } + return { + ...result, + status: 'unknown', + summary: '发布可能已提交到平台，结果待人工确认，已避免自动重复发布。', + error: { + ...(result.error ?? {}), + publish_submit_uncertain: true, + original_status: 'failed', + original_summary: result.summary, + }, + } +} diff --git a/apps/desktop-client/src/main/runtime-controller-publish-retry.test.ts b/apps/desktop-client/src/main/runtime-controller-publish-retry.test.ts new file mode 100644 index 0000000..5b05b6b --- /dev/null +++ b/apps/desktop-client/src/main/runtime-controller-publish-retry.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from 'vitest' + +import type { AdapterExecutionResult } from './adapters/base' +import { + asSubmitUncertainExecution, + isDefinitivePublishFailure, +} from './publish-result-classification' + +describe('publish retry result classification', () => { + it('keeps CSRF 403 publish failures definitive after submit started', () => { + const result: AdapterExecutionResult = { + status: 'failed', + summary: '什么值得买发布失败。', + error: { + code: 'smzdm_publish_failed', + message: '{"error_code":403,"error_msg":"CSRF token invalid"}', + }, + } + + expect(isDefinitivePublishFailure(result)).toBe(true) + expect(asSubmitUncertainExecution(result).status).toBe('failed') + }) + + it('marks ambiguous submit-phase publish failures as unknown', () => { + const result: AdapterExecutionResult = { + status: 'failed', + summary: '发布失败。', + error: { + code: 'smzdm_publish_failed', + message: 'network interrupted after submit', + }, + } + + const normalized = asSubmitUncertainExecution(result) + expect(normalized.status).toBe('unknown') + expect(normalized.error?.publish_submit_uncertain).toBe(true) + }) +}) diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index 2f338c5..b48a60c 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -80,6 +80,11 @@ import { startHiddenPlaywrightReaper, } from './playwright-cdp' import { getProcessMetricsSnapshot } from './process-metrics' +import { + asSubmitUncertainExecution, + isAuthorizationFailureCode, + markAuthorizationFailureNonRetryable, +} from './publish-result-classification' import { enqueuePublishTask as enqueuePublishSchedulerTask, getPublishSchedulerSnapshot, @@ -500,34 +505,6 @@ function eligibleMonitorPlatformIds(): string[] { .filter((platform) => !blocked.has(platform)) } -function markAuthorizationFailureNonRetryable( - error: Record, - category: 'ai_platform_authorization' | 'media_account_authorization', -): Record { - return { - ...error, - retryable: false, - non_retryable: true, - failure_category: category, - } -} - -function isAuthorizationFailureCode(code: string | null | undefined): boolean { - if (!code) { - return false - } - const normalized = code.trim().toLowerCase() - return ( - normalized === 'desktop_account_auth_expired' || - normalized === 'desktop_account_challenge_required' || - normalized === 'desktop_account_risk_control' || - normalized.endsWith('_not_logged_in') || - normalized.endsWith('_login_required') || - normalized.endsWith('_login_expired') || - normalized.endsWith('_challenge_required') - ) -} - function isAuthorizationFailureMessage(message: string | null | undefined): boolean { if (!message) { return false @@ -782,7 +759,8 @@ function startRuntime(): void { return } if (playwrightCDPFatal) { - const message = '浏览器自动化通道不可用，客户端已停止采集。请重启客户端或检查安全软件/端口占用。' + const message = + '浏览器自动化通道不可用，客户端已停止采集。请重启客户端或检查安全软件/端口占用。' state.lastError = message setSchedulerPhase('paused') setSchedulerError(message) @@ -1922,23 +1900,6 @@ async function markPublishSubmitStartedIfNeeded(taskId: string, stage: string): state.activeExecutions.set(taskId, active) } -function asSubmitUncertainExecution(result: AdapterExecutionResult): AdapterExecutionResult { - if (result.status !== 'failed') { - return result - } - return { - ...result, - status: 'unknown', - summary: '发布可能已提交到平台，结果待人工确认，已避免自动重复发布。', - error: { - ...(result.error ?? {}), - publish_submit_uncertain: true, - original_status: 'failed', - original_summary: result.summary, - }, - } -} - async function retainPlaywrightPageOrThrow( task: RuntimeTaskRecord, payload: Record, @@ -2183,7 +2144,12 @@ async function pullNextTask(kind: 'publish' | 'monitor'): Promise { async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise { const leaseLimit = resolveLegacyMonitoringLeaseLimit() - if (!canRunLive(state.session) || isLeaseInFlight('monitor') || leaseLimit <= 0 || playwrightCDPFatal) { + if ( + !canRunLive(state.session) || + isLeaseInFlight('monitor') || + leaseLimit <= 0 || + playwrightCDPFatal + ) { return } @@ -2205,7 +2171,8 @@ async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise { noteSchedulerPull() enqueueLeasedMonitoringTasks(leased.tasks, routing) - state.monitorFallbackBackoffUntil = leased.tasks.length > 0 ? 0 : Date.now() + monitorPullFallbackIntervalMs + state.monitorFallbackBackoffUntil = + leased.tasks.length > 0 ? 0 : Date.now() + monitorPullFallbackIntervalMs } catch (error) { state.lastPullAt = Date.now() state.lastPullStatus = 'failed' @@ -3720,7 +3687,9 @@ async function ensurePlaywrightCDPAdmissionForRequest( return ready } -async function ensurePlaywrightCDPAdmissionForFallback(kind: 'publish' | 'monitor'): Promise { +async function ensurePlaywrightCDPAdmissionForFallback( + kind: 'publish' | 'monitor', +): Promise { if (kind === 'publish') { const queuedPlaywrightPublish = getPublishSchedulerSnapshot().tasks.some( (task) => task.state === 'queued' && adapterRequiresPlaywright('publish', task.platform), @@ -3759,7 +3728,9 @@ async function ensurePlaywrightCDPAdmission(): Promise { } } -function isPlaywrightCDPInfrastructureError(error: unknown): error is PlaywrightCDPInfrastructureError { +function isPlaywrightCDPInfrastructureError( + error: unknown, +): error is PlaywrightCDPInfrastructureError { return error instanceof PlaywrightCDPInfrastructureError || isPlaywrightCDPUnavailableError(error) } @@ -3779,7 +3750,8 @@ function deferPendingTaskRequest( } function adapterRequiresPlaywright(kind: 'publish' | 'monitor', platform: string): boolean { - const adapter = kind === 'publish' ? selectPublishAdapter(platform) : selectMonitorAdapter(platform) + const adapter = + kind === 'publish' ? selectPublishAdapter(platform) : selectMonitorAdapter(platform) return adapter?.executionMode === 'playwright' } @@ -3820,8 +3792,7 @@ function markPlaywrightCDPFatal(message: string, taskId?: string): void { playwrightCDPFatal = true const fatalMessage = - message || - '浏览器自动化通道不可用，客户端已停止采集。请重启客户端或检查安全软件/端口占用。' + message || '浏览器自动化通道不可用，客户端已停止采集。请重启客户端或检查安全软件/端口占用。' state.lastError = fatalMessage setSchedulerPhase('paused') setSchedulerError(fatalMessage) diff --git a/apps/desktop-client/src/renderer/views/PublishManagementView.vue b/apps/desktop-client/src/renderer/views/PublishManagementView.vue index 049ba07..86b40b0 100644 --- a/apps/desktop-client/src/renderer/views/PublishManagementView.vue +++ b/apps/desktop-client/src/renderer/views/PublishManagementView.vue @@ -201,7 +201,7 @@ function buildWangyihaoArticleUrl( } function normalizePublishTaskStatus(status: DesktopTaskInfo['status']): DesktopTaskInfo['status'] { - return status === 'unknown' ? 'failed' : status + return status } function normalizeTaskErrorMessage( @@ -810,7 +810,9 @@ const publishTasks = computed(() => extractString(taskError.code), task.platform, ), - submitUncertain: task.status === 'unknown' && taskError.publish_submit_uncertain === true, + submitUncertain: + task.status === 'unknown' && + (taskError.publish_submit_uncertain === true || Boolean(task.publish_submit_started_at)), complianceBlockedRecordId, complianceBlockedAt: task.compliance_blocked_at ? parseTimestamp(task.compliance_blocked_at) diff --git a/packages/shared-types/src/index.ts b/packages/shared-types/src/index.ts index 55d6edb..7d8a3db 100644 --- a/packages/shared-types/src/index.ts +++ b/packages/shared-types/src/index.ts @@ -317,6 +317,7 @@ export interface DesktopTaskInfo { dedup_key: string | null active_attempt_id: string | null lease_expires_at: string | null + publish_submit_started_at?: string | null attempts: number result: Record | null error: Record | null diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 9a8b403..cfd4a97 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -95,6 +95,7 @@ type DesktopTaskView struct { DedupKey *string `json:"dedup_key"` ActiveAttemptID *string `json:"active_attempt_id"` LeaseExpiresAt *time.Time `json:"lease_expires_at"` + PublishSubmitStartedAt *time.Time `json:"publish_submit_started_at,omitempty"` Attempts int `json:"attempts"` Result json.RawMessage `json:"result"` Error json.RawMessage `json:"error"` @@ -1563,6 +1564,7 @@ func buildListPublishTasksByStatusesQuery( t.error, t.created_at, t.updated_at, + t.publish_submit_started_at, j.status, j.compliance_blocked_record_id, j.compliance_blocked_at, @@ -1638,28 +1640,29 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) { items := make([]DesktopTaskView, 0) for rows.Next() { var ( - desktopID uuid.UUID - jobID uuid.UUID - tenantID int64 - workspaceID int64 - targetAccountID uuid.UUID - targetClientID uuid.UUID - platform string - kind string - payload []byte - status string - dedupKey pgtype.Text - activeAttemptID pgtype.UUID - leaseExpiresAt pgtype.Timestamptz - attempts int32 - resultJSON []byte - errorJSON []byte - createdAt time.Time - updatedAt time.Time - jobStatus string - blockedRecordID pgtype.Int8 - blockedAt pgtype.Timestamptz - blockedReason pgtype.Text + desktopID uuid.UUID + jobID uuid.UUID + tenantID int64 + workspaceID int64 + targetAccountID uuid.UUID + targetClientID uuid.UUID + platform string + kind string + payload []byte + status string + dedupKey pgtype.Text + activeAttemptID pgtype.UUID + leaseExpiresAt pgtype.Timestamptz + attempts int32 + resultJSON []byte + errorJSON []byte + createdAt time.Time + updatedAt time.Time + publishSubmitStartedAt pgtype.Timestamptz + jobStatus string + blockedRecordID pgtype.Int8 + blockedAt pgtype.Timestamptz + blockedReason pgtype.Text ) if err := rows.Scan( &desktopID, @@ -1680,6 +1683,7 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) { &errorJSON, &createdAt, &updatedAt, + &publishSubmitStartedAt, &jobStatus, &blockedRecordID, &blockedAt, @@ -1708,6 +1712,11 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) { value := leaseExpiresAt.Time leaseExpiresAtValue = &value } + var publishSubmitStartedAtValue *time.Time + if publishSubmitStartedAt.Valid { + value := publishSubmitStartedAt.Time + publishSubmitStartedAtValue = &value + } publishJobStatusText := strings.TrimSpace(jobStatus) var publishJobStatus *string @@ -1751,6 +1760,7 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) { DedupKey: dedupKeyText, ActiveAttemptID: activeAttemptIDText, LeaseExpiresAt: leaseExpiresAtValue, + PublishSubmitStartedAt: publishSubmitStartedAtValue, Attempts: int(attempts), Result: json.RawMessage(resultJSON), Error: json.RawMessage(errorJSON), @@ -1833,24 +1843,25 @@ func buildDesktopTaskView(task *repository.DesktopTask) DesktopTaskView { } return DesktopTaskView{ - ID: task.DesktopID.String(), - JobID: task.JobID.String(), - TenantID: task.TenantID, - WorkspaceID: task.WorkspaceID, - TargetAccountID: task.TargetAccountID.String(), - TargetClientID: task.TargetClientID.String(), - Platform: task.Platform, - Kind: task.Kind, - Payload: json.RawMessage(task.Payload), - Status: normalizeDesktopTaskViewStatus(task.Kind, task.Status), - DedupKey: task.DedupKey, - ActiveAttemptID: activeAttemptID, - LeaseExpiresAt: task.LeaseExpiresAt, - Attempts: task.Attempts, - Result: json.RawMessage(task.Result), - Error: json.RawMessage(task.Error), - CreatedAt: task.CreatedAt, - UpdatedAt: task.UpdatedAt, + ID: task.DesktopID.String(), + JobID: task.JobID.String(), + TenantID: task.TenantID, + WorkspaceID: task.WorkspaceID, + TargetAccountID: task.TargetAccountID.String(), + TargetClientID: task.TargetClientID.String(), + Platform: task.Platform, + Kind: task.Kind, + Payload: json.RawMessage(task.Payload), + Status: normalizeDesktopTaskViewStatus(task.Kind, task.Status), + DedupKey: task.DedupKey, + ActiveAttemptID: activeAttemptID, + LeaseExpiresAt: task.LeaseExpiresAt, + PublishSubmitStartedAt: task.PublishSubmitStartedAt, + Attempts: task.Attempts, + Result: json.RawMessage(task.Result), + Error: json.RawMessage(task.Error), + CreatedAt: task.CreatedAt, + UpdatedAt: task.UpdatedAt, } } @@ -1899,8 +1910,10 @@ func normalizeDesktopTaskCompletionStatusForKind(kind string, status string) str func normalizeDesktopTaskViewStatus(kind string, status string) string { switch strings.TrimSpace(kind) { - case "monitor", "publish": + case "monitor": return normalizeDesktopTaskTerminalStatus(status) + case "publish": + return strings.TrimSpace(status) default: return strings.TrimSpace(status) } diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index f46e152..d5479bb 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/server/internal/tenant/app/desktop_task_service_test.go @@ -257,14 +257,14 @@ func TestClassifyLeaseErrorDefersQueuedMonitorTask(t *testing.T) { } } -func TestNormalizeDesktopTaskViewStatusMapsMonitorAndPublishUnknownToFailed(t *testing.T) { +func TestNormalizeDesktopTaskViewStatusMapsMonitorUnknownToFailedButKeepsPublishUnknown(t *testing.T) { t.Parallel() if got := normalizeDesktopTaskViewStatus("monitor", "unknown"); got != "failed" { t.Fatalf("monitor unknown view status = %q, want failed", got) } - if got := normalizeDesktopTaskViewStatus("publish", "unknown"); got != "failed" { - t.Fatalf("publish unknown view status = %q, want failed", got) + if got := normalizeDesktopTaskViewStatus("publish", "unknown"); got != "unknown" { + t.Fatalf("publish unknown view status = %q, want unknown", got) } } diff --git a/server/internal/tenant/app/publish_job_service.go b/server/internal/tenant/app/publish_job_service.go index 08087ab..413d98c 100644 --- a/server/internal/tenant/app/publish_job_service.go +++ b/server/internal/tenant/app/publish_job_service.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "regexp" "sort" "strconv" "strings" @@ -218,6 +219,14 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr if err := lockDesktopPublishDedupKeys(ctx, tx, dedupKeys); err != nil { return nil, err } + releasedPublishOutcomes := make([]*desktopPublishSyncOutcome, 0) + if len(dedupKeys) > 0 { + outcomes, releaseErr := releaseDefinitiveFailedUnknownPublishTasks(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, dedupKeys) + if releaseErr != nil { + return nil, releaseErr + } + releasedPublishOutcomes = append(releasedPublishOutcomes, outcomes...) + } existingTasks, err := loadExistingPublishRecordTargets(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, articleID, targets) if err != nil { return nil, err @@ -338,6 +347,11 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr return nil, response.ErrInternal(50100, "desktop_publish_job_commit_failed", "failed to commit desktop publish job") } invalidateArticleCaches(ctx, s.cache, actor.TenantID, &articleID) + for _, outcome := range releasedPublishOutcomes { + if outcome != nil { + invalidateArticleCaches(ctx, s.cache, outcome.TenantID, &outcome.ArticleID) + } + } for _, task := range createdTasks { dispatchEvent := DesktopDispatchEventFromTask(task, "task_available") @@ -514,6 +528,155 @@ func lockDesktopPublishDedupKeys(ctx context.Context, tx pgx.Tx, keys []string) return nil } +func releaseDefinitiveFailedUnknownPublishTasks( + ctx context.Context, + tx pgx.Tx, + tenantID int64, + workspaceID int64, + dedupKeys []string, +) ([]*desktopPublishSyncOutcome, error) { + if len(dedupKeys) == 0 { + return nil, nil + } + + rows, err := tx.Query(ctx, ` + SELECT desktop_id, tenant_id, workspace_id, platform_id, kind, payload, status, result, error + FROM desktop_tasks + WHERE tenant_id = $1 + AND workspace_id = $2 + AND kind = 'publish' + AND status = 'unknown' + AND dedup_key = ANY($3::text[]) + FOR UPDATE + `, tenantID, workspaceID, dedupKeys) + if err != nil { + return nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to inspect stale publish deduplication tasks") + } + defer rows.Close() + + tasks := make([]repository.DesktopTask, 0) + for rows.Next() { + var task repository.DesktopTask + if scanErr := rows.Scan( + &task.DesktopID, + &task.TenantID, + &task.WorkspaceID, + &task.Platform, + &task.Kind, + &task.Payload, + &task.Status, + &task.Result, + &task.Error, + ); scanErr != nil { + return nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to parse stale publish deduplication task") + } + tasks = append(tasks, task) + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to iterate stale publish deduplication tasks") + } + + outcomes := make([]*desktopPublishSyncOutcome, 0) + for index := range tasks { + task := tasks[index] + if !publishUnknownTaskHasDefinitiveFailure(&task) { + continue + } + updated, updateErr := markUnknownPublishTaskDefinitiveFailed(ctx, tx, &task) + if updateErr != nil { + return nil, updateErr + } + outcome, syncErr := syncDesktopPublishTaskState(ctx, tx, updated) + if syncErr != nil { + return nil, syncErr + } + if outcome != nil { + outcomes = append(outcomes, outcome) + } + } + return outcomes, nil +} + +func markUnknownPublishTaskDefinitiveFailed( + ctx context.Context, + tx pgx.Tx, + task *repository.DesktopTask, +) (*repository.DesktopTask, error) { + if task == nil { + return nil, response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found") + } + row := tx.QueryRow(ctx, ` + UPDATE desktop_tasks + SET status = 'failed', + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + publish_submit_started_at = NULL, + updated_at = NOW() + WHERE desktop_id = $1 + AND workspace_id = $2 + AND status = 'unknown' + RETURNING desktop_id, tenant_id, workspace_id, platform_id, kind, payload, status, result, error + `, task.DesktopID, task.WorkspaceID) + var updated repository.DesktopTask + if err := row.Scan( + &updated.DesktopID, + &updated.TenantID, + &updated.WorkspaceID, + &updated.Platform, + &updated.Kind, + &updated.Payload, + &updated.Status, + &updated.Result, + &updated.Error, + ); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return nil, response.ErrConflict(40985, "desktop_task_reconcile_conflict", "desktop task is not waiting for reconcile") + } + return nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to release stale publish deduplication task") + } + return &updated, nil +} + +func publishUnknownTaskHasDefinitiveFailure(task *repository.DesktopTask) bool { + if task == nil || task.Kind != "publish" || task.Status != "unknown" { + return false + } + errorPayload := unmarshalJSONObject(task.Error) + if boolValueFromMap(errorPayload, "publish_submit_uncertain") { + return false + } + combined := strings.ToLower(strings.Join([]string{ + stringPointerValue(extractStringPointer(errorPayload, "code", "error_code")), + stringPointerValue(extractStringPointer(errorPayload, "message", "error_msg", "detail")), + }, " ")) + if combined == "" { + return false + } + return definitivePublishFailurePattern.MatchString(combined) +} + +func boolValueFromMap(payload map[string]any, key string) bool { + if payload == nil { + return false + } + value, ok := payload[key] + if !ok { + return false + } + switch typed := value.(type) { + case bool: + return typed + case string: + normalized := strings.TrimSpace(strings.ToLower(typed)) + return normalized == "true" || normalized == "1" || normalized == "yes" + default: + return false + } +} + +var definitivePublishFailurePattern = regexp.MustCompile(`csrf|forbidden|\b403\b|not_logged_in|login_required|token_missing|cookie_missing|challenge_required|risk_control|captcha|验证码|滑块|人机验证|安全验证|风控|风险|参数错误|内容为空|image_fetch_failed|cover_fetch_failed|upload_failed|signature_failed`) + func loadExistingPublishRecordTargets( ctx context.Context, tx pgx.Tx, @@ -567,7 +730,7 @@ func loadExistingPublishRecordTargets( AND pr.article_id = $3 AND pr.platform_account_id = ANY($4::bigint[]) AND pr.deleted_at IS NULL - AND pr.status IN ('queued', 'publishing', 'success') + AND (pr.status IN ('queued', 'publishing', 'success') OR dt.status = 'unknown') AND dt.status IN ('queued', 'in_progress', 'succeeded', 'unknown') ORDER BY pr.platform_account_id, CASE @@ -785,10 +948,18 @@ func (s *PublishJobService) requeueUnknownPublishTask( return nil, response.ErrInternal(50094, "desktop_task_reconcile_failed", "failed to reconcile desktop task") } + publishOutcome, err := syncDesktopPublishTaskState(ctx, tx, requeued) + if err != nil { + return nil, err + } + if err := tx.Commit(ctx); err != nil { return nil, response.ErrInternal(50106, "desktop_task_reconcile_commit_failed", "failed to commit desktop task reconcile") } + if publishOutcome != nil { + invalidateArticleCaches(ctx, s.cache, publishOutcome.TenantID, &publishOutcome.ArticleID) + } publishDesktopTaskLifecycleEvent(ctx, s.messaging, s.logger, requeued, "task_available") return createPublishJobResponseForRequeuedTask(requeued), nil } diff --git a/server/internal/tenant/app/publish_job_service_test.go b/server/internal/tenant/app/publish_job_service_test.go index 9d54072..263c9e3 100644 --- a/server/internal/tenant/app/publish_job_service_test.go +++ b/server/internal/tenant/app/publish_job_service_test.go @@ -203,6 +203,32 @@ func TestCreatePublishJobResponseForRequeuedTaskReportsRequeuedTaskID(t *testing } } +func TestPublishUnknownTaskHasDefinitiveFailureReleasesCSRF403(t *testing.T) { + t.Parallel() + + task := &repository.DesktopTask{ + Kind: "publish", + Status: "unknown", + Error: []byte(`{"error_code":403,"error_msg":"CSRF token invalid"}`), + } + if !publishUnknownTaskHasDefinitiveFailure(task) { + t.Fatal("unknown publish task with CSRF 403 should be treated as definitive failed") + } +} + +func TestPublishUnknownTaskHasDefinitiveFailureKeepsSubmitUncertain(t *testing.T) { + t.Parallel() + + task := &repository.DesktopTask{ + Kind: "publish", + Status: "unknown", + Error: []byte(`{"publish_submit_uncertain":true,"message":"network interrupted after submit"}`), + } + if publishUnknownTaskHasDefinitiveFailure(task) { + t.Fatal("submit-uncertain unknown publish task must stay in dedup set") + } +} + func TestLoadExistingPublishRecordTargetsDedupeUnknownTasks(t *testing.T) { t.Parallel() @@ -220,6 +246,9 @@ func TestLoadExistingPublishRecordTargetsDedupeUnknownTasks(t *testing.T) { if !strings.Contains(normalizedSQL, "dt.status IN ('queued', 'in_progress', 'succeeded', 'unknown')") { t.Fatalf("publish dedup lookup should include unknown task status; query:\n%s", tx.sql) } + if !strings.Contains(normalizedSQL, "OR dt.status = 'unknown'") { + t.Fatalf("publish dedup lookup should include unknown tasks even when publish record is failed; query:\n%s", tx.sql) + } if !strings.Contains(normalizedSQL, "AND dt.tenant_id = $1") { t.Fatalf("publish dedup lookup should constrain desktop_tasks by tenant for the payload index; query:\n%s", tx.sql) } diff --git a/server/internal/tenant/repository/desktop_task_repo.go b/server/internal/tenant/repository/desktop_task_repo.go index a9021c2..b76a2bc 100644 --- a/server/internal/tenant/repository/desktop_task_repo.go +++ b/server/internal/tenant/repository/desktop_task_repo.go @@ -37,37 +37,38 @@ type CreateDesktopPublishJobParams struct { } type DesktopTask struct { - DesktopID uuid.UUID - JobID uuid.UUID - TenantID int64 - WorkspaceID int64 - TargetAccountID uuid.UUID - TargetClientID uuid.UUID - Platform string - Kind string - Payload []byte - Status string - Priority int - Lane string - LaneWeight int - Source string - SchedulerGroupKey *string - MonitorTaskID *int64 - SupersedesTaskID *uuid.UUID - ControlFlags []byte - InterruptGeneration int - DedupKey *string - ActiveAttemptID *uuid.UUID - LeaseExpiresAt *time.Time - Attempts int - Result []byte - Error []byte - StartedAt *time.Time - InterruptedAt *time.Time - InterruptReason *string - EnqueuedAt time.Time - CreatedAt time.Time - UpdatedAt time.Time + DesktopID uuid.UUID + JobID uuid.UUID + TenantID int64 + WorkspaceID int64 + TargetAccountID uuid.UUID + TargetClientID uuid.UUID + Platform string + Kind string + Payload []byte + Status string + Priority int + Lane string + LaneWeight int + Source string + SchedulerGroupKey *string + MonitorTaskID *int64 + SupersedesTaskID *uuid.UUID + ControlFlags []byte + InterruptGeneration int + DedupKey *string + ActiveAttemptID *uuid.UUID + LeaseExpiresAt *time.Time + Attempts int + Result []byte + Error []byte + StartedAt *time.Time + InterruptedAt *time.Time + InterruptReason *string + EnqueuedAt time.Time + PublishSubmitStartedAt *time.Time + CreatedAt time.Time + UpdatedAt time.Time } type CreateDesktopTaskParams struct { @@ -355,37 +356,38 @@ func desktopPublishJobFromGenerated(row generated.DesktopPublishJob) *DesktopPub func desktopTaskFromGenerated(row generated.DesktopTask) *DesktopTask { return &DesktopTask{ - DesktopID: uuidFromPG(row.DesktopID), - JobID: uuidFromPG(row.JobID), - TenantID: row.TenantID, - WorkspaceID: row.WorkspaceID, - TargetAccountID: uuidFromPG(row.TargetAccountID), - TargetClientID: uuidFromPG(row.TargetClientID), - Platform: row.PlatformID, - Kind: row.Kind, - Payload: row.Payload, - Status: row.Status, - Priority: int(row.Priority), - Lane: row.Lane, - LaneWeight: int(row.LaneWeight), - Source: row.Source, - SchedulerGroupKey: nullableText(row.SchedulerGroupKey), - MonitorTaskID: nullableInt64(row.MonitorTaskID), - SupersedesTaskID: nullableUUID(row.SupersedesTaskID), - ControlFlags: row.ControlFlags, - InterruptGeneration: int(row.InterruptGeneration), - DedupKey: nullableText(row.DedupKey), - ActiveAttemptID: nullableUUID(row.ActiveAttemptID), - LeaseExpiresAt: optionalTime(row.LeaseExpiresAt), - Attempts: int(row.Attempts), - Result: row.Result, - Error: row.Error, - StartedAt: optionalTime(row.StartedAt), - InterruptedAt: optionalTime(row.InterruptedAt), - InterruptReason: nullableText(row.InterruptReason), - EnqueuedAt: timeFromTimestamp(row.EnqueuedAt), - CreatedAt: timeFromTimestamp(row.CreatedAt), - UpdatedAt: timeFromTimestamp(row.UpdatedAt), + DesktopID: uuidFromPG(row.DesktopID), + JobID: uuidFromPG(row.JobID), + TenantID: row.TenantID, + WorkspaceID: row.WorkspaceID, + TargetAccountID: uuidFromPG(row.TargetAccountID), + TargetClientID: uuidFromPG(row.TargetClientID), + Platform: row.PlatformID, + Kind: row.Kind, + Payload: row.Payload, + Status: row.Status, + Priority: int(row.Priority), + Lane: row.Lane, + LaneWeight: int(row.LaneWeight), + Source: row.Source, + SchedulerGroupKey: nullableText(row.SchedulerGroupKey), + MonitorTaskID: nullableInt64(row.MonitorTaskID), + SupersedesTaskID: nullableUUID(row.SupersedesTaskID), + ControlFlags: row.ControlFlags, + InterruptGeneration: int(row.InterruptGeneration), + DedupKey: nullableText(row.DedupKey), + ActiveAttemptID: nullableUUID(row.ActiveAttemptID), + LeaseExpiresAt: optionalTime(row.LeaseExpiresAt), + Attempts: int(row.Attempts), + Result: row.Result, + Error: row.Error, + StartedAt: optionalTime(row.StartedAt), + InterruptedAt: optionalTime(row.InterruptedAt), + InterruptReason: nullableText(row.InterruptReason), + EnqueuedAt: timeFromTimestamp(row.EnqueuedAt), + PublishSubmitStartedAt: optionalTime(row.PublishSubmitStartedAt), + CreatedAt: timeFromTimestamp(row.CreatedAt), + UpdatedAt: timeFromTimestamp(row.UpdatedAt), } } diff --git a/server/internal/tenant/repository/generated/desktop_task.sql.go b/server/internal/tenant/repository/generated/desktop_task.sql.go index 037dc5f..3b1a43e 100644 --- a/server/internal/tenant/repository/generated/desktop_task.sql.go +++ b/server/internal/tenant/repository/generated/desktop_task.sql.go @@ -849,6 +849,10 @@ SET status = CASE active_attempt_id = NULL, lease_token_hash = NULL, lease_expires_at = NULL, + publish_submit_started_at = CASE + WHEN $1::text = 'retry' THEN NULL + ELSE publish_submit_started_at + END, attempts = CASE WHEN $1::text = 'retry' THEN 0 ELSE attempts END, updated_at = now() WHERE desktop_id = $4 diff --git a/server/internal/tenant/repository/generated/desktop_task_sql_test.go b/server/internal/tenant/repository/generated/desktop_task_sql_test.go index 7398b37..2d90a6d 100644 --- a/server/internal/tenant/repository/generated/desktop_task_sql_test.go +++ b/server/internal/tenant/repository/generated/desktop_task_sql_test.go @@ -39,6 +39,10 @@ func TestReconcileDesktopTaskRetryResetsAttempts(t *testing.T) { if !strings.Contains(query, "attempts = CASE WHEN $1::text = 'retry' THEN 0 ELSE attempts END") { t.Fatalf("retry reconcile must reset attempts so manual retry can be leased; query:\n%s", query) } + if !strings.Contains(query, "publish_submit_started_at = CASE") || + !strings.Contains(query, "WHEN $1::text = 'retry' THEN NULL") { + t.Fatalf("retry reconcile must clear stale publish submit marker; query:\n%s", query) + } if strings.Contains(query, "attempts + CASE") { t.Fatalf("retry reconcile must not increment attempts; query:\n%s", query) } diff --git a/server/internal/tenant/repository/queries/desktop_task.sql b/server/internal/tenant/repository/queries/desktop_task.sql index a3059d3..4c487a3 100644 --- a/server/internal/tenant/repository/queries/desktop_task.sql +++ b/server/internal/tenant/repository/queries/desktop_task.sql @@ -225,6 +225,10 @@ SET status = CASE active_attempt_id = NULL, lease_token_hash = NULL, lease_expires_at = NULL, + publish_submit_started_at = CASE + WHEN sqlc.arg(status)::text = 'retry' THEN NULL + ELSE publish_submit_started_at + END, attempts = CASE WHEN sqlc.arg(status)::text = 'retry' THEN 0 ELSE attempts END, updated_at = now() WHERE desktop_id = sqlc.arg(desktop_id)