Compare commits

...

2 Commits

Author SHA1 Message Date
root ed48674ab5 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
2026-06-25 00:22:59 +08:00
root 2ec266201c fix desktop smzdm csrf publishing 2026-06-25 00:18:24 +08:00
17 changed files with 761 additions and 269 deletions
@@ -21,6 +21,8 @@ import {
encodeSmzdmForm, encodeSmzdmForm,
extractSmzdmArticleIdFromHref, extractSmzdmArticleIdFromHref,
getSmzdmCoverCropRect, getSmzdmCoverCropRect,
isSmzdmCsrfMissingMessage,
smzdmCsrfHeaders,
smzdmArticleUrl, smzdmArticleUrl,
} from './smzdm' } from './smzdm'
@@ -54,6 +56,18 @@ describe('smzdm adapter helpers', () => {
expect(buildSmzdmAwne('4242025233', 123)).toBe('m2At8ovjSz5kmnLNUFEW9g==') expect(buildSmzdmAwne('4242025233', 123)).toBe('m2At8ovjSz5kmnLNUFEW9g==')
}) })
it('builds SMZDM CSRF headers like the reference extension submit request', () => {
expect(smzdmCsrfHeaders('token-1')).toMatchObject({
_csrf_token: 'token-1',
'x-csrf-token': 'token-1',
})
})
it('recognizes SMZDM missing csrf responses', () => {
expect(isSmzdmCsrfMissingMessage('CSRF token缺失')).toBe(true)
expect(isSmzdmCsrfMissingMessage('csrf token missing')).toBe(true)
})
it('centers the cover crop at the required SMZDM ratio', () => { it('centers the cover crop at the required SMZDM ratio', () => {
const crop = getSmzdmCoverCropRect(2000, 1000) const crop = getSmzdmCoverCropRect(2000, 1000)
+229 -100
View File
@@ -2,19 +2,18 @@ import { Buffer } from 'node:buffer'
import { createCipheriv, createHash } from 'node:crypto' import { createCipheriv, createHash } from 'node:crypto'
import type { JsonValue } from '@geo/shared-types' import type { JsonValue } from '@geo/shared-types'
import { nativeImage } from 'electron'
import { PUBLISH_IMAGE_FETCH_FAILED_MESSAGE } from '../../shared/publisher-errors' import { PUBLISH_IMAGE_FETCH_FAILED_MESSAGE } from '../../shared/publisher-errors'
import { resolveDesktopApiURL } from '../transport/api-client'
import type { PublishAdapter, PublishAdapterContext } from './base' import type { PublishAdapter, PublishAdapterContext } from './base'
import { import {
adapterFetch,
extractImageSources, extractImageSources,
normalizeArticleHtml, normalizeArticleHtml,
raceWithAbort, raceWithAbort,
sessionCookieHeader, sessionCookieHeader,
sessionCookieValue,
sessionFetchJson, sessionFetchJson,
} from './common' } from './common'
import { fetchImageAssetBlob } from './media-image'
type SmzdmPublishType = 'publish' type SmzdmPublishType = 'publish'
@@ -64,6 +63,14 @@ type SmzdmCropResponse = {
}> }>
} }
type SmzdmEditorTokenResponse = {
error_code?: number | string
error_msg?: string
data?: {
token?: string
}
}
type SmzdmSubmitResponse = { type SmzdmSubmitResponse = {
error_code?: number | string error_code?: number | string
error_msg?: string error_msg?: string
@@ -75,12 +82,6 @@ type SmzdmSubmitResponse = {
} }
} }
type SmzdmImageBlob = {
blob: Blob
width: number
height: number
}
type SmzdmUploadedImage = { type SmzdmUploadedImage = {
id: string id: string
origin: string origin: string
@@ -151,6 +152,29 @@ function isSmzdmChallengeMessage(message: string): boolean {
) )
} }
export function isSmzdmCsrfMissingMessage(message: string): boolean {
return /csrf/i.test(message) && /(缺失|missing|required|invalid|校验|验证|403)/i.test(message)
}
export function smzdmCsrfHeaders(token: string): Record<string, string> {
return {
_csrf_token: token,
'x-csrf-token': token,
'x-xsrf-token': token,
'csrf-token': token,
}
}
function appendSmzdmCsrfFields(target: FormData | URLSearchParams, token: string): void {
if (!token) {
return
}
target.append('_csrf', token)
target.append('csrf_token', token)
target.append('csrfToken', token)
target.append('csrf', token)
}
async function waitForHumanPace(signal?: AbortSignal): Promise<void> { async function waitForHumanPace(signal?: AbortSignal): Promise<void> {
if (signal?.aborted) { if (signal?.aborted) {
throw new Error('adapter_aborted') throw new Error('adapter_aborted')
@@ -178,16 +202,152 @@ async function smzdmCookieHeader(context: PublishAdapterContext): Promise<string
return await sessionCookieHeader(context.session, 'smzdm.com').catch(() => '') return await sessionCookieHeader(context.session, 'smzdm.com').catch(() => '')
} }
async function smzdmCookieCsrfToken(context: PublishAdapterContext): Promise<string> {
const names = [
'csrf_token',
'csrfToken',
'_csrf',
'csrf',
'XSRF-TOKEN',
'xsrf-token',
'csrftoken',
'CSRF-TOKEN',
'SMZDM_CSRF_TOKEN',
]
for (const name of names) {
const value = await sessionCookieValue(context.session, 'smzdm.com', name).catch(() => '')
const decoded = decodeCookieText(value).trim()
if (decoded) {
return decoded
}
}
const cookies = await context.session.cookies.get({ domain: 'smzdm.com' }).catch(() => [])
const matched = cookies.find((cookie) => /(csrf|xsrf)/i.test(cookie.name) && cookie.value.trim())
return matched ? decodeCookieText(matched.value).trim() : ''
}
async function smzdmPageCsrfToken(context: PublishAdapterContext): Promise<string> {
const page = context.playwright?.page
if (!page) {
return ''
}
return await page
.evaluate(() => {
const candidates: string[] = []
const push = (value: unknown) => {
if (typeof value !== 'string') {
return
}
const trimmed = value.trim()
if (trimmed && !candidates.includes(trimmed)) {
candidates.push(trimmed)
}
}
const selectors = [
'meta[name="csrf-token"]',
'meta[name="csrf_token"]',
'meta[name="_csrf"]',
'meta[name="csrf"]',
'meta[name="x-csrf-token"]',
'meta[name="xsrf-token"]',
]
for (const selector of selectors) {
push(document.querySelector<HTMLMetaElement>(selector)?.content)
}
const inputNames = ['_csrf', 'csrf', 'csrf_token', 'csrfToken', 'x-csrf-token']
for (const name of inputNames) {
push(document.querySelector<HTMLInputElement>(`input[name="${name}"]`)?.value)
}
const globals = window as unknown as Window &
Record<string, unknown> & {
csrfToken?: string
csrf_token?: string
_csrf?: string
CSRF_TOKEN?: string
SMZDM_CSRF_TOKEN?: string
PAGE_CONFIG?: Record<string, unknown>
__INITIAL_STATE__?: Record<string, unknown>
}
push(globals.csrfToken)
push(globals.csrf_token)
push(globals._csrf)
push(globals.CSRF_TOKEN)
push(globals.SMZDM_CSRF_TOKEN)
push(globals.PAGE_CONFIG?.csrfToken)
push(globals.PAGE_CONFIG?.csrf_token)
push(globals.__INITIAL_STATE__?.csrfToken)
push(globals.__INITIAL_STATE__?.csrf_token)
const readStorage = (storage: Storage) => {
for (let index = 0; index < storage.length; index += 1) {
const key = storage.key(index) || ''
if (/(csrf|xsrf)/i.test(key)) {
push(storage.getItem(key))
}
}
}
try {
readStorage(window.localStorage)
} catch {
// Ignore inaccessible storage in restricted platform pages.
}
try {
readStorage(window.sessionStorage)
} catch {
// Ignore inaccessible storage in restricted platform pages.
}
for (const item of document.cookie.split(';')) {
const [rawName, ...rawValue] = item.split('=')
if (/(csrf|xsrf)/i.test(rawName || '')) {
push(decodeURIComponent(rawValue.join('=')))
}
}
return candidates[0] || ''
})
.catch(() => '')
}
async function smzdmCsrfToken(context: PublishAdapterContext): Promise<string> {
return (await smzdmCookieCsrfToken(context)) || (await smzdmPageCsrfToken(context))
}
async function fetchSmzdmEditorToken(context: PublishAdapterContext): Promise<string> {
const response = await sessionFetchJson<SmzdmEditorTokenResponse>(
context.session,
`${SMZDM_ORIGIN}/api/editor/get_token`,
{
method: 'GET',
credentials: 'include',
headers: await smzdmHeaders(context, {}, ''),
},
{ signal: context.signal },
).catch(() => null)
return stringId(response?.data?.token) || (await smzdmCsrfToken(context))
}
async function smzdmHeaders( async function smzdmHeaders(
context: PublishAdapterContext, context: PublishAdapterContext,
extra: Record<string, string> = {}, extra: Record<string, string> = {},
csrfToken?: string,
): Promise<Record<string, string>> { ): Promise<Record<string, string>> {
const cookie = await smzdmCookieHeader(context) const cookie = await smzdmCookieHeader(context)
const token = csrfToken ?? (await smzdmCsrfToken(context))
return { return {
accept: 'application/json, text/plain, */*', accept: 'application/json, text/plain, */*',
origin: SMZDM_ORIGIN, origin: SMZDM_ORIGIN,
referer: SMZDM_TOU_GAO_URL, referer: SMZDM_TOU_GAO_URL,
'x-requested-with': 'XMLHttpRequest',
...(cookie ? { cookie } : {}), ...(cookie ? { cookie } : {}),
...(token ? smzdmCsrfHeaders(token) : {}),
...extra, ...extra,
} }
} }
@@ -289,81 +449,6 @@ async function getArticleId(context: PublishAdapterContext): Promise<string> {
return articleId return articleId
} }
function buildAssetURLCandidates(sourceUrl: string): string[] {
const trimmed = sourceUrl.trim()
if (!trimmed) {
return []
}
if (/^(data|blob):/i.test(trimmed)) {
return [trimmed]
}
const candidates: string[] = []
const pushCandidate = (value: string) => {
if (!candidates.includes(value)) {
candidates.push(value)
}
}
try {
const parsed = new URL(trimmed, resolveDesktopApiURL('/'))
if (parsed.pathname.startsWith('/api/')) {
const apiAssetUrl = new URL(resolveDesktopApiURL(`${parsed.pathname}${parsed.search}`))
if (apiAssetUrl.pathname.startsWith('/api/public/assets/')) {
apiAssetUrl.searchParams.set('format', 'png')
}
pushCandidate(apiAssetUrl.toString())
}
pushCandidate(parsed.toString())
} catch {
pushCandidate(trimmed)
}
return candidates
}
async function fetchSmzdmImageBlob(
sourceUrl: string,
signal?: AbortSignal,
): Promise<SmzdmImageBlob | null> {
for (const candidate of buildAssetURLCandidates(sourceUrl)) {
const response = await adapterFetch(candidate, {}, { signal, timeoutMs: 10_000 }).catch(
() => null,
)
if (!response?.ok) {
continue
}
const sourceBlob = await response.blob().catch(() => null)
if (!sourceBlob || !sourceBlob.type.startsWith('image/')) {
continue
}
const buffer = Buffer.from(await sourceBlob.arrayBuffer())
const image = nativeImage.createFromBuffer(buffer)
if (image.isEmpty()) {
continue
}
const size = image.getSize()
const normalizedBlob =
sourceBlob.type.toLowerCase() === 'image/webp'
? new Blob([new Uint8Array(image.toPNG())], { type: 'image/png' })
: sourceBlob
return {
blob: normalizedBlob,
width: size.width,
height: size.height,
}
}
return null
}
export function getSmzdmCoverCropRect( export function getSmzdmCoverCropRect(
width: number, width: number,
height: number, height: number,
@@ -396,16 +481,18 @@ async function uploadImage(
articleId: string, articleId: string,
cropCover: boolean, cropCover: boolean,
): Promise<SmzdmUploadedImage | null> { ): Promise<SmzdmUploadedImage | null> {
const image = await fetchSmzdmImageBlob(sourceUrl, context.signal) const image = await fetchImageAssetBlob(sourceUrl, { signal: context.signal })
if (!image) { if (!image) {
throw new Error(cropCover ? 'smzdm_cover_fetch_failed' : 'smzdm_image_fetch_failed') throw new Error(cropCover ? 'smzdm_cover_fetch_failed' : 'smzdm_image_fetch_failed')
} }
const csrfToken = await smzdmCsrfToken(context)
const form = new FormData() const form = new FormData()
form.append('imgFile', image.blob, 'image.png') form.append('imgFile', image.blob, image.fileName)
form.append('id', 'WU_FILE_0') form.append('id', 'WU_FILE_0')
form.append('type', image.blob.type || 'image/png') form.append('type', image.blob.type || 'image/png')
form.append('article_id', articleId) form.append('article_id', articleId)
appendSmzdmCsrfFields(form, csrfToken)
const uploaded = await sessionFetchJson<SmzdmImageUploadResponse>( const uploaded = await sessionFetchJson<SmzdmImageUploadResponse>(
context.session, context.session,
@@ -413,7 +500,7 @@ async function uploadImage(
{ {
method: 'POST', method: 'POST',
credentials: 'include', credentials: 'include',
headers: await smzdmHeaders(context), headers: await smzdmHeaders(context, {}, csrfToken),
body: form, body: form,
}, },
{ signal: context.signal }, { signal: context.signal },
@@ -426,6 +513,9 @@ async function uploadImage(
if (!isSuccessfulCode(uploaded.error_code) || !uploaded.data?.url) { if (!isSuccessfulCode(uploaded.error_code) || !uploaded.data?.url) {
const message = smzdmResponseMessage(uploaded) const message = smzdmResponseMessage(uploaded)
if (isSmzdmCsrfMissingMessage(message)) {
throw new Error(`smzdm_csrf_missing:${message}`)
}
if (isSmzdmChallengeMessage(message)) { if (isSmzdmChallengeMessage(message)) {
throw new Error(`smzdm_challenge_required:${message}`) throw new Error(`smzdm_challenge_required:${message}`)
} }
@@ -436,19 +526,26 @@ async function uploadImage(
const uploadedId = stringId(uploaded.data.id) const uploadedId = stringId(uploaded.data.id)
if (uploadedId) { if (uploadedId) {
const recordBody = new URLSearchParams({
article_id: articleId,
ids: uploadedId,
})
appendSmzdmCsrfFields(recordBody, csrfToken)
void sessionFetchJson( void sessionFetchJson(
context.session, context.session,
`${SMZDM_ORIGIN}/api/editor/image_add_time/record`, `${SMZDM_ORIGIN}/api/editor/image_add_time/record`,
{ {
method: 'POST', method: 'POST',
credentials: 'include', credentials: 'include',
headers: await smzdmHeaders(context, { headers: await smzdmHeaders(
context,
{
'content-type': 'application/x-www-form-urlencoded', 'content-type': 'application/x-www-form-urlencoded',
}), },
body: new URLSearchParams({ csrfToken,
article_id: articleId, ),
ids: uploadedId, body: recordBody,
}),
}, },
{ signal: context.signal }, { signal: context.signal },
).catch(() => null) ).catch(() => null)
@@ -468,13 +565,21 @@ async function uploadImage(
{ {
method: 'POST', method: 'POST',
credentials: 'include', credentials: 'include',
headers: await smzdmHeaders(context, { headers: await smzdmHeaders(
context,
{
'content-type': 'application/x-www-form-urlencoded', 'content-type': 'application/x-www-form-urlencoded',
}), },
body: new URLSearchParams({ csrfToken,
),
body: (() => {
const body = new URLSearchParams({
article_id: articleId, article_id: articleId,
pic_url: uploaded.data.url, pic_url: uploaded.data.url,
}), })
appendSmzdmCsrfFields(body, csrfToken)
return body
})(),
}, },
{ signal: context.signal }, { signal: context.signal },
).catch(() => null) ).catch(() => null)
@@ -504,6 +609,7 @@ async function uploadImage(
cropForm.append('cut_pic_list[0][original_pic_width]', String(image.width)) cropForm.append('cut_pic_list[0][original_pic_width]', String(image.width))
cropForm.append('cut_pic_list[0][cutUrl]', original?.data?.original_url || '') cropForm.append('cut_pic_list[0][cutUrl]', original?.data?.original_url || '')
cropForm.append('cut_pic_list[0][is_head]', '1') cropForm.append('cut_pic_list[0][is_head]', '1')
appendSmzdmCsrfFields(cropForm, csrfToken)
const cropped = await sessionFetchJson<SmzdmCropResponse>( const cropped = await sessionFetchJson<SmzdmCropResponse>(
context.session, context.session,
@@ -511,7 +617,7 @@ async function uploadImage(
{ {
method: 'POST', method: 'POST',
credentials: 'include', credentials: 'include',
headers: await smzdmHeaders(context), headers: await smzdmHeaders(context, {}, csrfToken),
body: cropForm, body: cropForm,
}, },
{ signal: context.signal }, { signal: context.signal },
@@ -796,6 +902,17 @@ function failureResult(
} }
} }
if (message.startsWith('smzdm_csrf_missing') || isSmzdmCsrfMissingMessage(message)) {
return {
status: 'failed',
summary: '什么值得买 CSRF token 缺失,请重新登录或重新绑定账号后再试。',
error: {
code: 'smzdm_csrf_missing',
message,
},
}
}
if (message.startsWith('smzdm_cover_fetch_failed')) { if (message.startsWith('smzdm_cover_fetch_failed')) {
return { return {
status: 'failed', status: 'failed',
@@ -905,15 +1022,24 @@ export const smzdmAdapter: PublishAdapter = {
await context.reportProgress('smzdm.submit') await context.reportProgress('smzdm.submit')
await waitForHumanPace(context.signal) await waitForHumanPace(context.signal)
const csrfToken = await fetchSmzdmEditorToken(context)
if (!csrfToken) {
throw new Error('smzdm_csrf_missing:editor_token_missing')
}
const response = await sessionFetchJson<SmzdmSubmitResponse>( const response = await sessionFetchJson<SmzdmSubmitResponse>(
context.session, context.session,
`${SMZDM_ORIGIN}/api/editor/article/submit`, `${SMZDM_ORIGIN}/api/editor/article/submit`,
{ {
method: 'POST', method: 'POST',
credentials: 'include', credentials: 'include',
headers: await smzdmHeaders(context, { headers: await smzdmHeaders(
context,
{
'content-type': 'application/x-www-form-urlencoded', 'content-type': 'application/x-www-form-urlencoded',
}), },
csrfToken,
),
body: encodeSmzdmForm( body: encodeSmzdmForm(
buildSubmitForm({ buildSubmitForm({
articleId, articleId,
@@ -937,6 +1063,9 @@ export const smzdmAdapter: PublishAdapter = {
const topicCode = response.data?.update_topic?.error_code const topicCode = response.data?.update_topic?.error_code
if (!isSuccessfulCode(response.error_code) || !isSuccessfulCode(topicCode)) { if (!isSuccessfulCode(response.error_code) || !isSuccessfulCode(topicCode)) {
const message = response.data?.update_topic?.error_msg || smzdmResponseMessage(response) const message = response.data?.update_topic?.error_msg || smzdmResponseMessage(response)
if (isSmzdmCsrfMissingMessage(message)) {
throw new Error(`smzdm_csrf_missing:${message}`)
}
if (isSmzdmChallengeMessage(message)) { if (isSmzdmChallengeMessage(message)) {
throw new Error(`smzdm_challenge_required:${message}`) throw new Error(`smzdm_challenge_required:${message}`)
} }
@@ -94,6 +94,18 @@ describe('platform auth adapters', () => {
).toBe('auth_failure') ).toBe('auth_failure')
}) })
it('classifies SMZDM missing csrf as auth failure', () => {
const adapter = getPlatformAdapter('smzdm')
expect(
adapter.classifyFailure({
error: {
code: 'smzdm_csrf_missing',
},
}),
).toBe('auth_failure')
})
it('classifies Juejin missing login as auth failure', () => { it('classifies Juejin missing login as auth failure', () => {
const adapter = getPlatformAdapter('juejin') const adapter = getPlatformAdapter('juejin')
@@ -234,6 +234,20 @@ function classifyDongchediFailure(input: PlatformFailureInput): PlatformFailureC
return classifyFailure(input) return classifyFailure(input)
} }
function classifySmzdmFailure(input: PlatformFailureInput): PlatformFailureClassification {
const code = typeof input.error?.code === 'string' ? input.error.code : ''
if (code === 'smzdm_not_logged_in' || code === 'smzdm_csrf_missing') {
return 'auth_failure'
}
if (code === 'smzdm_challenge_required') {
return 'challenge'
}
if (code.startsWith('smzdm_')) {
return 'ok'
}
return classifyFailure(input)
}
const aiAdapter: PlatformAdapter = { const aiAdapter: PlatformAdapter = {
async probe(account, partition) { async probe(account, partition) {
return await probeAIAccountSession(account, partition) return await probeAIAccountSession(account, partition)
@@ -364,6 +378,16 @@ const dongchediAdapter: PlatformAdapter = {
classifyFailure: classifyDongchediFailure, classifyFailure: classifyDongchediFailure,
} }
const smzdmAdapter: PlatformAdapter = {
async probe(account, partition) {
return await probePublishAccountSession(account, partition)
},
async silentRefresh(account, partition) {
return await silentRefreshPublishAccountSession(account, partition)
},
classifyFailure: classifySmzdmFailure,
}
const adapterRegistry = new Map<string, PlatformAdapter>( const adapterRegistry = new Map<string, PlatformAdapter>(
[ [
...aiPlatformCatalog.map((platform) => platform.id), ...aiPlatformCatalog.map((platform) => platform.id),
@@ -404,6 +428,8 @@ const adapterRegistry = new Map<string, PlatformAdapter>(
? zolAdapter ? zolAdapter
: platform === 'dongchedi' : platform === 'dongchedi'
? dongchediAdapter ? dongchediAdapter
: platform === 'smzdm'
? smzdmAdapter
: aiPlatformCatalog.some((item) => item.id === platform) : aiPlatformCatalog.some((item) => item.id === platform)
? aiAdapter ? aiAdapter
: genericAdapter, : genericAdapter,
@@ -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,
@@ -1656,6 +1658,7 @@ func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
errorJSON []byte errorJSON []byte
createdAt time.Time createdAt time.Time
updatedAt time.Time updatedAt time.Time
publishSubmitStartedAt pgtype.Timestamptz
jobStatus string jobStatus string
blockedRecordID pgtype.Int8 blockedRecordID pgtype.Int8
blockedAt pgtype.Timestamptz blockedAt pgtype.Timestamptz
@@ -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),
@@ -1846,6 +1856,7 @@ func buildDesktopTaskView(task *repository.DesktopTask) DesktopTaskView {
DedupKey: task.DedupKey, DedupKey: task.DedupKey,
ActiveAttemptID: activeAttemptID, ActiveAttemptID: activeAttemptID,
LeaseExpiresAt: task.LeaseExpiresAt, LeaseExpiresAt: task.LeaseExpiresAt,
PublishSubmitStartedAt: task.PublishSubmitStartedAt,
Attempts: task.Attempts, Attempts: task.Attempts,
Result: json.RawMessage(task.Result), Result: json.RawMessage(task.Result),
Error: json.RawMessage(task.Error), Error: json.RawMessage(task.Error),
@@ -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)
} }
@@ -66,6 +66,7 @@ type DesktopTask struct {
InterruptedAt *time.Time InterruptedAt *time.Time
InterruptReason *string InterruptReason *string
EnqueuedAt time.Time EnqueuedAt time.Time
PublishSubmitStartedAt *time.Time
CreatedAt time.Time CreatedAt time.Time
UpdatedAt time.Time UpdatedAt time.Time
} }
@@ -384,6 +385,7 @@ func desktopTaskFromGenerated(row generated.DesktopTask) *DesktopTask {
InterruptedAt: optionalTime(row.InterruptedAt), InterruptedAt: optionalTime(row.InterruptedAt),
InterruptReason: nullableText(row.InterruptReason), InterruptReason: nullableText(row.InterruptReason),
EnqueuedAt: timeFromTimestamp(row.EnqueuedAt), EnqueuedAt: timeFromTimestamp(row.EnqueuedAt),
PublishSubmitStartedAt: optionalTime(row.PublishSubmitStartedAt),
CreatedAt: timeFromTimestamp(row.CreatedAt), CreatedAt: timeFromTimestamp(row.CreatedAt),
UpdatedAt: timeFromTimestamp(row.UpdatedAt), 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)