fix: allow retry after definitive publish failures
Frontend CI / Frontend (push) Successful in 3m18s
Backend CI / Backend (push) Successful in 16m16s
Desktop Client Build / Publish Client Artifacts to NAS (push) Successful in 26s
Desktop Client Build / Resolve Build Metadata (push) Successful in 26s
Desktop Client Build / Build Desktop Client (push) Successful in 25m45s

This commit is contained in:
2026-06-25 00:22:59 +08:00
parent 2ec266201c
commit ed48674ab5
13 changed files with 473 additions and 162 deletions
@@ -0,0 +1,72 @@
import type { JsonValue } from '@geo/shared-types'
import type { AdapterExecutionResult } from './adapters/base'
export function markAuthorizationFailureNonRetryable(
error: Record<string, JsonValue>,
category: 'ai_platform_authorization' | 'media_account_authorization',
): Record<string, JsonValue> {
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,
},
}
}
@@ -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)
})
})
@@ -80,6 +80,11 @@ import {
startHiddenPlaywrightReaper, startHiddenPlaywrightReaper,
} from './playwright-cdp' } from './playwright-cdp'
import { getProcessMetricsSnapshot } from './process-metrics' import { getProcessMetricsSnapshot } from './process-metrics'
import {
asSubmitUncertainExecution,
isAuthorizationFailureCode,
markAuthorizationFailureNonRetryable,
} from './publish-result-classification'
import { import {
enqueuePublishTask as enqueuePublishSchedulerTask, enqueuePublishTask as enqueuePublishSchedulerTask,
getPublishSchedulerSnapshot, getPublishSchedulerSnapshot,
@@ -500,34 +505,6 @@ function eligibleMonitorPlatformIds(): string[] {
.filter((platform) => !blocked.has(platform)) .filter((platform) => !blocked.has(platform))
} }
function markAuthorizationFailureNonRetryable(
error: Record<string, JsonValue>,
category: 'ai_platform_authorization' | 'media_account_authorization',
): Record<string, JsonValue> {
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 { function isAuthorizationFailureMessage(message: string | null | undefined): boolean {
if (!message) { if (!message) {
return false return false
@@ -782,7 +759,8 @@ function startRuntime(): void {
return return
} }
if (playwrightCDPFatal) { if (playwrightCDPFatal) {
const message = '浏览器自动化通道不可用,客户端已停止采集。请重启客户端或检查安全软件/端口占用。' const message =
'浏览器自动化通道不可用,客户端已停止采集。请重启客户端或检查安全软件/端口占用。'
state.lastError = message state.lastError = message
setSchedulerPhase('paused') setSchedulerPhase('paused')
setSchedulerError(message) setSchedulerError(message)
@@ -1922,23 +1900,6 @@ async function markPublishSubmitStartedIfNeeded(taskId: string, stage: string):
state.activeExecutions.set(taskId, active) 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( async function retainPlaywrightPageOrThrow(
task: RuntimeTaskRecord, task: RuntimeTaskRecord,
payload: Record<string, JsonValue>, payload: Record<string, JsonValue>,
@@ -2183,7 +2144,12 @@ async function pullNextTask(kind: 'publish' | 'monitor'): Promise<void> {
async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise<void> { async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise<void> {
const leaseLimit = resolveLegacyMonitoringLeaseLimit() const leaseLimit = resolveLegacyMonitoringLeaseLimit()
if (!canRunLive(state.session) || isLeaseInFlight('monitor') || leaseLimit <= 0 || playwrightCDPFatal) { if (
!canRunLive(state.session) ||
isLeaseInFlight('monitor') ||
leaseLimit <= 0 ||
playwrightCDPFatal
) {
return return
} }
@@ -2205,7 +2171,8 @@ async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise<void> {
noteSchedulerPull() noteSchedulerPull()
enqueueLeasedMonitoringTasks(leased.tasks, routing) 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) { } catch (error) {
state.lastPullAt = Date.now() state.lastPullAt = Date.now()
state.lastPullStatus = 'failed' state.lastPullStatus = 'failed'
@@ -3720,7 +3687,9 @@ async function ensurePlaywrightCDPAdmissionForRequest(
return ready return ready
} }
async function ensurePlaywrightCDPAdmissionForFallback(kind: 'publish' | 'monitor'): Promise<boolean> { async function ensurePlaywrightCDPAdmissionForFallback(
kind: 'publish' | 'monitor',
): Promise<boolean> {
if (kind === 'publish') { if (kind === 'publish') {
const queuedPlaywrightPublish = getPublishSchedulerSnapshot().tasks.some( const queuedPlaywrightPublish = getPublishSchedulerSnapshot().tasks.some(
(task) => task.state === 'queued' && adapterRequiresPlaywright('publish', task.platform), (task) => task.state === 'queued' && adapterRequiresPlaywright('publish', task.platform),
@@ -3759,7 +3728,9 @@ async function ensurePlaywrightCDPAdmission(): Promise<boolean> {
} }
} }
function isPlaywrightCDPInfrastructureError(error: unknown): error is PlaywrightCDPInfrastructureError { function isPlaywrightCDPInfrastructureError(
error: unknown,
): error is PlaywrightCDPInfrastructureError {
return error instanceof PlaywrightCDPInfrastructureError || isPlaywrightCDPUnavailableError(error) return error instanceof PlaywrightCDPInfrastructureError || isPlaywrightCDPUnavailableError(error)
} }
@@ -3779,7 +3750,8 @@ function deferPendingTaskRequest(
} }
function adapterRequiresPlaywright(kind: 'publish' | 'monitor', platform: string): boolean { 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' return adapter?.executionMode === 'playwright'
} }
@@ -3820,8 +3792,7 @@ function markPlaywrightCDPFatal(message: string, taskId?: string): void {
playwrightCDPFatal = true playwrightCDPFatal = true
const fatalMessage = const fatalMessage =
message || message || '浏览器自动化通道不可用,客户端已停止采集。请重启客户端或检查安全软件/端口占用。'
'浏览器自动化通道不可用,客户端已停止采集。请重启客户端或检查安全软件/端口占用。'
state.lastError = fatalMessage state.lastError = fatalMessage
setSchedulerPhase('paused') setSchedulerPhase('paused')
setSchedulerError(fatalMessage) setSchedulerError(fatalMessage)
@@ -201,7 +201,7 @@ function buildWangyihaoArticleUrl(
} }
function normalizePublishTaskStatus(status: DesktopTaskInfo['status']): DesktopTaskInfo['status'] { function normalizePublishTaskStatus(status: DesktopTaskInfo['status']): DesktopTaskInfo['status'] {
return status === 'unknown' ? 'failed' : status return status
} }
function normalizeTaskErrorMessage( function normalizeTaskErrorMessage(
@@ -810,7 +810,9 @@ const publishTasks = computed<PublishTaskItem[]>(() =>
extractString(taskError.code), extractString(taskError.code),
task.platform, 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, complianceBlockedRecordId,
complianceBlockedAt: task.compliance_blocked_at complianceBlockedAt: task.compliance_blocked_at
? parseTimestamp(task.compliance_blocked_at) ? parseTimestamp(task.compliance_blocked_at)
+1
View File
@@ -317,6 +317,7 @@ export interface DesktopTaskInfo {
dedup_key: string | null dedup_key: string | null
active_attempt_id: string | null active_attempt_id: string | null
lease_expires_at: string | null lease_expires_at: string | null
publish_submit_started_at?: string | null
attempts: number attempts: number
result: Record<string, JsonValue> | null result: Record<string, JsonValue> | null
error: Record<string, JsonValue> | null error: Record<string, JsonValue> | null
@@ -95,6 +95,7 @@ type DesktopTaskView struct {
DedupKey *string `json:"dedup_key"` DedupKey *string `json:"dedup_key"`
ActiveAttemptID *string `json:"active_attempt_id"` ActiveAttemptID *string `json:"active_attempt_id"`
LeaseExpiresAt *time.Time `json:"lease_expires_at"` LeaseExpiresAt *time.Time `json:"lease_expires_at"`
PublishSubmitStartedAt *time.Time `json:"publish_submit_started_at,omitempty"`
Attempts int `json:"attempts"` Attempts int `json:"attempts"`
Result json.RawMessage `json:"result"` Result json.RawMessage `json:"result"`
Error json.RawMessage `json:"error"` Error json.RawMessage `json:"error"`
@@ -1563,6 +1564,7 @@ func buildListPublishTasksByStatusesQuery(
t.error, t.error,
t.created_at, t.created_at,
t.updated_at, t.updated_at,
t.publish_submit_started_at,
j.status, j.status,
j.compliance_blocked_record_id, j.compliance_blocked_record_id,
j.compliance_blocked_at, j.compliance_blocked_at,
@@ -1638,28 +1640,29 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
items := make([]DesktopTaskView, 0) items := make([]DesktopTaskView, 0)
for rows.Next() { for rows.Next() {
var ( var (
desktopID uuid.UUID desktopID uuid.UUID
jobID uuid.UUID jobID uuid.UUID
tenantID int64 tenantID int64
workspaceID int64 workspaceID int64
targetAccountID uuid.UUID targetAccountID uuid.UUID
targetClientID uuid.UUID targetClientID uuid.UUID
platform string platform string
kind string kind string
payload []byte payload []byte
status string status string
dedupKey pgtype.Text dedupKey pgtype.Text
activeAttemptID pgtype.UUID activeAttemptID pgtype.UUID
leaseExpiresAt pgtype.Timestamptz leaseExpiresAt pgtype.Timestamptz
attempts int32 attempts int32
resultJSON []byte resultJSON []byte
errorJSON []byte errorJSON []byte
createdAt time.Time createdAt time.Time
updatedAt time.Time updatedAt time.Time
jobStatus string publishSubmitStartedAt pgtype.Timestamptz
blockedRecordID pgtype.Int8 jobStatus string
blockedAt pgtype.Timestamptz blockedRecordID pgtype.Int8
blockedReason pgtype.Text blockedAt pgtype.Timestamptz
blockedReason pgtype.Text
) )
if err := rows.Scan( if err := rows.Scan(
&desktopID, &desktopID,
@@ -1680,6 +1683,7 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
&errorJSON, &errorJSON,
&createdAt, &createdAt,
&updatedAt, &updatedAt,
&publishSubmitStartedAt,
&jobStatus, &jobStatus,
&blockedRecordID, &blockedRecordID,
&blockedAt, &blockedAt,
@@ -1708,6 +1712,11 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
value := leaseExpiresAt.Time value := leaseExpiresAt.Time
leaseExpiresAtValue = &value leaseExpiresAtValue = &value
} }
var publishSubmitStartedAtValue *time.Time
if publishSubmitStartedAt.Valid {
value := publishSubmitStartedAt.Time
publishSubmitStartedAtValue = &value
}
publishJobStatusText := strings.TrimSpace(jobStatus) publishJobStatusText := strings.TrimSpace(jobStatus)
var publishJobStatus *string var publishJobStatus *string
@@ -1751,6 +1760,7 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
DedupKey: dedupKeyText, DedupKey: dedupKeyText,
ActiveAttemptID: activeAttemptIDText, ActiveAttemptID: activeAttemptIDText,
LeaseExpiresAt: leaseExpiresAtValue, LeaseExpiresAt: leaseExpiresAtValue,
PublishSubmitStartedAt: publishSubmitStartedAtValue,
Attempts: int(attempts), Attempts: int(attempts),
Result: json.RawMessage(resultJSON), Result: json.RawMessage(resultJSON),
Error: json.RawMessage(errorJSON), Error: json.RawMessage(errorJSON),
@@ -1833,24 +1843,25 @@ func buildDesktopTaskView(task *repository.DesktopTask) DesktopTaskView {
} }
return DesktopTaskView{ return DesktopTaskView{
ID: task.DesktopID.String(), ID: task.DesktopID.String(),
JobID: task.JobID.String(), JobID: task.JobID.String(),
TenantID: task.TenantID, TenantID: task.TenantID,
WorkspaceID: task.WorkspaceID, WorkspaceID: task.WorkspaceID,
TargetAccountID: task.TargetAccountID.String(), TargetAccountID: task.TargetAccountID.String(),
TargetClientID: task.TargetClientID.String(), TargetClientID: task.TargetClientID.String(),
Platform: task.Platform, Platform: task.Platform,
Kind: task.Kind, Kind: task.Kind,
Payload: json.RawMessage(task.Payload), Payload: json.RawMessage(task.Payload),
Status: normalizeDesktopTaskViewStatus(task.Kind, task.Status), Status: normalizeDesktopTaskViewStatus(task.Kind, task.Status),
DedupKey: task.DedupKey, DedupKey: task.DedupKey,
ActiveAttemptID: activeAttemptID, ActiveAttemptID: activeAttemptID,
LeaseExpiresAt: task.LeaseExpiresAt, LeaseExpiresAt: task.LeaseExpiresAt,
Attempts: task.Attempts, PublishSubmitStartedAt: task.PublishSubmitStartedAt,
Result: json.RawMessage(task.Result), Attempts: task.Attempts,
Error: json.RawMessage(task.Error), Result: json.RawMessage(task.Result),
CreatedAt: task.CreatedAt, Error: json.RawMessage(task.Error),
UpdatedAt: task.UpdatedAt, CreatedAt: task.CreatedAt,
UpdatedAt: task.UpdatedAt,
} }
} }
@@ -1899,8 +1910,10 @@ func normalizeDesktopTaskCompletionStatusForKind(kind string, status string) str
func normalizeDesktopTaskViewStatus(kind string, status string) string { func normalizeDesktopTaskViewStatus(kind string, status string) string {
switch strings.TrimSpace(kind) { switch strings.TrimSpace(kind) {
case "monitor", "publish": case "monitor":
return normalizeDesktopTaskTerminalStatus(status) return normalizeDesktopTaskTerminalStatus(status)
case "publish":
return strings.TrimSpace(status)
default: default:
return strings.TrimSpace(status) return strings.TrimSpace(status)
} }
@@ -257,14 +257,14 @@ func TestClassifyLeaseErrorDefersQueuedMonitorTask(t *testing.T) {
} }
} }
func TestNormalizeDesktopTaskViewStatusMapsMonitorAndPublishUnknownToFailed(t *testing.T) { func TestNormalizeDesktopTaskViewStatusMapsMonitorUnknownToFailedButKeepsPublishUnknown(t *testing.T) {
t.Parallel() t.Parallel()
if got := normalizeDesktopTaskViewStatus("monitor", "unknown"); got != "failed" { if got := normalizeDesktopTaskViewStatus("monitor", "unknown"); got != "failed" {
t.Fatalf("monitor unknown view status = %q, want failed", got) t.Fatalf("monitor unknown view status = %q, want failed", got)
} }
if got := normalizeDesktopTaskViewStatus("publish", "unknown"); got != "failed" { if got := normalizeDesktopTaskViewStatus("publish", "unknown"); got != "unknown" {
t.Fatalf("publish unknown view status = %q, want failed", got) t.Fatalf("publish unknown view status = %q, want unknown", got)
} }
} }
@@ -4,6 +4,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"errors" "errors"
"regexp"
"sort" "sort"
"strconv" "strconv"
"strings" "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 { if err := lockDesktopPublishDedupKeys(ctx, tx, dedupKeys); err != nil {
return nil, err 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) existingTasks, err := loadExistingPublishRecordTargets(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, articleID, targets)
if err != nil { if err != nil {
return nil, err 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") return nil, response.ErrInternal(50100, "desktop_publish_job_commit_failed", "failed to commit desktop publish job")
} }
invalidateArticleCaches(ctx, s.cache, actor.TenantID, &articleID) 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 { for _, task := range createdTasks {
dispatchEvent := DesktopDispatchEventFromTask(task, "task_available") dispatchEvent := DesktopDispatchEventFromTask(task, "task_available")
@@ -514,6 +528,155 @@ func lockDesktopPublishDedupKeys(ctx context.Context, tx pgx.Tx, keys []string)
return nil 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( func loadExistingPublishRecordTargets(
ctx context.Context, ctx context.Context,
tx pgx.Tx, tx pgx.Tx,
@@ -567,7 +730,7 @@ func loadExistingPublishRecordTargets(
AND pr.article_id = $3 AND pr.article_id = $3
AND pr.platform_account_id = ANY($4::bigint[]) AND pr.platform_account_id = ANY($4::bigint[])
AND pr.deleted_at IS NULL 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') AND dt.status IN ('queued', 'in_progress', 'succeeded', 'unknown')
ORDER BY pr.platform_account_id, ORDER BY pr.platform_account_id,
CASE CASE
@@ -785,10 +948,18 @@ func (s *PublishJobService) requeueUnknownPublishTask(
return nil, response.ErrInternal(50094, "desktop_task_reconcile_failed", "failed to reconcile desktop task") 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 { if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50106, "desktop_task_reconcile_commit_failed", "failed to commit desktop task reconcile") 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") publishDesktopTaskLifecycleEvent(ctx, s.messaging, s.logger, requeued, "task_available")
return createPublishJobResponseForRequeuedTask(requeued), nil return createPublishJobResponseForRequeuedTask(requeued), nil
} }
@@ -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) { func TestLoadExistingPublishRecordTargetsDedupeUnknownTasks(t *testing.T) {
t.Parallel() t.Parallel()
@@ -220,6 +246,9 @@ func TestLoadExistingPublishRecordTargetsDedupeUnknownTasks(t *testing.T) {
if !strings.Contains(normalizedSQL, "dt.status IN ('queued', 'in_progress', 'succeeded', 'unknown')") { 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) 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") { 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) t.Fatalf("publish dedup lookup should constrain desktop_tasks by tenant for the payload index; query:\n%s", tx.sql)
} }
@@ -37,37 +37,38 @@ type CreateDesktopPublishJobParams struct {
} }
type DesktopTask struct { type DesktopTask struct {
DesktopID uuid.UUID DesktopID uuid.UUID
JobID uuid.UUID JobID uuid.UUID
TenantID int64 TenantID int64
WorkspaceID int64 WorkspaceID int64
TargetAccountID uuid.UUID TargetAccountID uuid.UUID
TargetClientID uuid.UUID TargetClientID uuid.UUID
Platform string Platform string
Kind string Kind string
Payload []byte Payload []byte
Status string Status string
Priority int Priority int
Lane string Lane string
LaneWeight int LaneWeight int
Source string Source string
SchedulerGroupKey *string SchedulerGroupKey *string
MonitorTaskID *int64 MonitorTaskID *int64
SupersedesTaskID *uuid.UUID SupersedesTaskID *uuid.UUID
ControlFlags []byte ControlFlags []byte
InterruptGeneration int InterruptGeneration int
DedupKey *string DedupKey *string
ActiveAttemptID *uuid.UUID ActiveAttemptID *uuid.UUID
LeaseExpiresAt *time.Time LeaseExpiresAt *time.Time
Attempts int Attempts int
Result []byte Result []byte
Error []byte Error []byte
StartedAt *time.Time StartedAt *time.Time
InterruptedAt *time.Time InterruptedAt *time.Time
InterruptReason *string InterruptReason *string
EnqueuedAt time.Time EnqueuedAt time.Time
CreatedAt time.Time PublishSubmitStartedAt *time.Time
UpdatedAt time.Time CreatedAt time.Time
UpdatedAt time.Time
} }
type CreateDesktopTaskParams struct { type CreateDesktopTaskParams struct {
@@ -355,37 +356,38 @@ func desktopPublishJobFromGenerated(row generated.DesktopPublishJob) *DesktopPub
func desktopTaskFromGenerated(row generated.DesktopTask) *DesktopTask { func desktopTaskFromGenerated(row generated.DesktopTask) *DesktopTask {
return &DesktopTask{ return &DesktopTask{
DesktopID: uuidFromPG(row.DesktopID), DesktopID: uuidFromPG(row.DesktopID),
JobID: uuidFromPG(row.JobID), JobID: uuidFromPG(row.JobID),
TenantID: row.TenantID, TenantID: row.TenantID,
WorkspaceID: row.WorkspaceID, WorkspaceID: row.WorkspaceID,
TargetAccountID: uuidFromPG(row.TargetAccountID), TargetAccountID: uuidFromPG(row.TargetAccountID),
TargetClientID: uuidFromPG(row.TargetClientID), TargetClientID: uuidFromPG(row.TargetClientID),
Platform: row.PlatformID, Platform: row.PlatformID,
Kind: row.Kind, Kind: row.Kind,
Payload: row.Payload, Payload: row.Payload,
Status: row.Status, Status: row.Status,
Priority: int(row.Priority), Priority: int(row.Priority),
Lane: row.Lane, Lane: row.Lane,
LaneWeight: int(row.LaneWeight), LaneWeight: int(row.LaneWeight),
Source: row.Source, Source: row.Source,
SchedulerGroupKey: nullableText(row.SchedulerGroupKey), SchedulerGroupKey: nullableText(row.SchedulerGroupKey),
MonitorTaskID: nullableInt64(row.MonitorTaskID), MonitorTaskID: nullableInt64(row.MonitorTaskID),
SupersedesTaskID: nullableUUID(row.SupersedesTaskID), SupersedesTaskID: nullableUUID(row.SupersedesTaskID),
ControlFlags: row.ControlFlags, ControlFlags: row.ControlFlags,
InterruptGeneration: int(row.InterruptGeneration), InterruptGeneration: int(row.InterruptGeneration),
DedupKey: nullableText(row.DedupKey), DedupKey: nullableText(row.DedupKey),
ActiveAttemptID: nullableUUID(row.ActiveAttemptID), ActiveAttemptID: nullableUUID(row.ActiveAttemptID),
LeaseExpiresAt: optionalTime(row.LeaseExpiresAt), LeaseExpiresAt: optionalTime(row.LeaseExpiresAt),
Attempts: int(row.Attempts), Attempts: int(row.Attempts),
Result: row.Result, Result: row.Result,
Error: row.Error, Error: row.Error,
StartedAt: optionalTime(row.StartedAt), StartedAt: optionalTime(row.StartedAt),
InterruptedAt: optionalTime(row.InterruptedAt), InterruptedAt: optionalTime(row.InterruptedAt),
InterruptReason: nullableText(row.InterruptReason), InterruptReason: nullableText(row.InterruptReason),
EnqueuedAt: timeFromTimestamp(row.EnqueuedAt), EnqueuedAt: timeFromTimestamp(row.EnqueuedAt),
CreatedAt: timeFromTimestamp(row.CreatedAt), PublishSubmitStartedAt: optionalTime(row.PublishSubmitStartedAt),
UpdatedAt: timeFromTimestamp(row.UpdatedAt), CreatedAt: timeFromTimestamp(row.CreatedAt),
UpdatedAt: timeFromTimestamp(row.UpdatedAt),
} }
} }
@@ -849,6 +849,10 @@ SET status = CASE
active_attempt_id = NULL, active_attempt_id = NULL,
lease_token_hash = NULL, lease_token_hash = NULL,
lease_expires_at = 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, attempts = CASE WHEN $1::text = 'retry' THEN 0 ELSE attempts END,
updated_at = now() updated_at = now()
WHERE desktop_id = $4 WHERE desktop_id = $4
@@ -39,6 +39,10 @@ func TestReconcileDesktopTaskRetryResetsAttempts(t *testing.T) {
if !strings.Contains(query, "attempts = CASE WHEN $1::text = 'retry' THEN 0 ELSE attempts END") { 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) 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") { if strings.Contains(query, "attempts + CASE") {
t.Fatalf("retry reconcile must not increment attempts; query:\n%s", query) t.Fatalf("retry reconcile must not increment attempts; query:\n%s", query)
} }
@@ -225,6 +225,10 @@ SET status = CASE
active_attempt_id = NULL, active_attempt_id = NULL,
lease_token_hash = NULL, lease_token_hash = NULL,
lease_expires_at = 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, attempts = CASE WHEN sqlc.arg(status)::text = 'retry' THEN 0 ELSE attempts END,
updated_at = now() updated_at = now()
WHERE desktop_id = sqlc.arg(desktop_id) WHERE desktop_id = sqlc.arg(desktop_id)