feat(tenant): supersede unknown publish tasks on manual retry

Replace the in-place requeue of submit-uncertain unknown publish tasks
with an explicit supersede-and-create-new flow: manual retries mark the
old unknown task failed (manual_retry_superseded) and create a fresh task
carrying manual_retry / superseded_unknown_task_ids in its payload.

Also fold title-length validation into definitivePublishFailurePattern and
check the pattern before the publish_submit_uncertain guard so those
failures are released from the dedup set.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

server/internal/tenant/app/publish_job_service.go

@@ -74,12 +74,14 @@ type CreatePublishJobAccountRequest struct {
 }
 
 type CreatePublishJobRequest struct {
-	Title            string                           `json:"title" binding:"required"`
-	ContentRef       map[string]any                   `json:"content_ref" binding:"required"`
-	Accounts         []CreatePublishJobAccountRequest `json:"accounts" binding:"required,min=1"`
-	ScheduledAt      *time.Time                       `json:"scheduled_at"`
-	ArticleVersionID *int64                           `json:"article_version_id,omitempty"`
-	AckRecordID      *int64                           `json:"ack_record_id,omitempty"`
+	Title                 string                           `json:"title" binding:"required"`
+	ContentRef            map[string]any                   `json:"content_ref" binding:"required"`
+	Accounts              []CreatePublishJobAccountRequest `json:"accounts" binding:"required,min=1"`
+	ScheduledAt           *time.Time                       `json:"scheduled_at"`
+	ArticleVersionID      *int64                           `json:"article_version_id,omitempty"`
+	AckRecordID           *int64                           `json:"ack_record_id,omitempty"`
+	AllowSupersedeUnknown bool                             `json:"-"`
+	ManualRetry           bool                             `json:"-"`
 }
 
 type CreatePublishJobResponse struct {
@@ -220,12 +222,21 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr
 		return nil, err
 	}
 	releasedPublishOutcomes := make([]*desktopPublishSyncOutcome, 0)
+	supersededUnknownTaskIDs := make(map[string][]string)
 	if len(dedupKeys) > 0 {
 		outcomes, releaseErr := releaseDefinitiveFailedUnknownPublishTasks(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, dedupKeys)
 		if releaseErr != nil {
 			return nil, releaseErr
 		}
 		releasedPublishOutcomes = append(releasedPublishOutcomes, outcomes...)
+		if req.AllowSupersedeUnknown {
+			outcomes, superseded, releaseErr := releaseManuallySupersededUnknownPublishTasks(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, dedupKeys)
+			if releaseErr != nil {
+				return nil, releaseErr
+			}
+			releasedPublishOutcomes = append(releasedPublishOutcomes, outcomes...)
+			supersededUnknownTaskIDs = superseded
+		}
 	}
 	existingTasks, err := loadExistingPublishRecordTargets(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, articleID, targets)
 	if err != nil {
@@ -299,7 +310,7 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr
 	createdTaskIDs := make([]string, 0, len(newTargets))
 	for _, target := range newTargets {
 		publishRecordID := publishRecordIDs[target.accountSeed.ID]
-		payloadJSON, payloadErr := marshalOptionalJSON(map[string]any{
+		payload := map[string]any{
 			"title":               req.Title,
 			"content_ref":         req.ContentRef,
 			"account_id":          target.account.DesktopID.String(),
@@ -309,7 +320,14 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr
 			"publish_batch_id":    publishBatchID,
 			"publish_record_id":   publishRecordID,
 			"platform_account_id": target.accountSeed.ID,
-		})
+		}
+		if supersededTaskIDs := supersededUnknownTaskIDs[target.dedupKey]; len(supersededTaskIDs) > 0 {
+			payload["manual_retry"] = true
+			payload["superseded_unknown_task_ids"] = supersededTaskIDs
+		} else if req.ManualRetry {
+			payload["manual_retry"] = true
+		}
+		payloadJSON, payloadErr := marshalOptionalJSON(payload)
 		if payloadErr != nil {
 			return nil, response.ErrBadRequest(40094, "invalid_desktop_task_payload", "publish job payload must be serializable")
 		}
@@ -597,10 +615,116 @@ func releaseDefinitiveFailedUnknownPublishTasks(
 	return outcomes, nil
 }
 
+type manuallySupersededUnknownPublishTask struct {
+	task     repository.DesktopTask
+	dedupKey string
+}
+
+func releaseManuallySupersededUnknownPublishTasks(
+	ctx context.Context,
+	tx pgx.Tx,
+	tenantID int64,
+	workspaceID int64,
+	dedupKeys []string,
+) ([]*desktopPublishSyncOutcome, map[string][]string, error) {
+	if len(dedupKeys) == 0 {
+		return nil, map[string][]string{}, nil
+	}
+
+	rows, err := tx.Query(ctx, `
+		SELECT desktop_id, tenant_id, workspace_id, platform_id, kind, payload, status, result, error, COALESCE(dedup_key, '')
+		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, nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to inspect manually superseded publish tasks")
+	}
+	defer rows.Close()
+
+	tasks := make([]manuallySupersededUnknownPublishTask, 0)
+	for rows.Next() {
+		var item manuallySupersededUnknownPublishTask
+		if scanErr := rows.Scan(
+			&item.task.DesktopID,
+			&item.task.TenantID,
+			&item.task.WorkspaceID,
+			&item.task.Platform,
+			&item.task.Kind,
+			&item.task.Payload,
+			&item.task.Status,
+			&item.task.Result,
+			&item.task.Error,
+			&item.dedupKey,
+		); scanErr != nil {
+			return nil, nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to parse manually superseded publish task")
+		}
+		tasks = append(tasks, item)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to iterate manually superseded publish tasks")
+	}
+
+	outcomes := make([]*desktopPublishSyncOutcome, 0, len(tasks))
+	supersededTaskIDs := make(map[string][]string, len(tasks))
+	for index := range tasks {
+		item := tasks[index]
+		updated, updateErr := markUnknownPublishTaskManuallySuperseded(ctx, tx, &item.task)
+		if updateErr != nil {
+			return nil, nil, updateErr
+		}
+		outcome, syncErr := syncDesktopPublishTaskState(ctx, tx, updated)
+		if syncErr != nil {
+			return nil, nil, syncErr
+		}
+		if outcome != nil {
+			outcomes = append(outcomes, outcome)
+		}
+		if item.dedupKey != "" {
+			supersededTaskIDs[item.dedupKey] = append(supersededTaskIDs[item.dedupKey], item.task.DesktopID.String())
+		}
+	}
+	return outcomes, supersededTaskIDs, nil
+}
+
 func markUnknownPublishTaskDefinitiveFailed(
 	ctx context.Context,
 	tx pgx.Tx,
 	task *repository.DesktopTask,
+) (*repository.DesktopTask, error) {
+	return markUnknownPublishTaskFailed(ctx, tx, task, nil)
+}
+
+func markUnknownPublishTaskManuallySuperseded(
+	ctx context.Context,
+	tx pgx.Tx,
+	task *repository.DesktopTask,
+) (*repository.DesktopTask, error) {
+	errorPayload := unmarshalJSONObject(task.Error)
+	if errorPayload == nil {
+		errorPayload = map[string]any{}
+	}
+	errorPayload["manual_retry_superseded"] = true
+	errorPayload["source"] = "manual_publish_retry"
+	if strings.TrimSpace(stringPointerValue(extractStringPointer(errorPayload, "message"))) == "" {
+		errorPayload["message"] = "用户已手动重新发起发布，旧的结果未知任务已终止。"
+	}
+	errorJSON, err := marshalOptionalJSON(errorPayload)
+	if err != nil {
+		return nil, response.ErrInternal(50131, "desktop_publish_dedup_cleanup_failed", "failed to encode manually superseded publish task")
+	}
+	return markUnknownPublishTaskFailed(ctx, tx, task, errorJSON)
+}
+
+func markUnknownPublishTaskFailed(
+	ctx context.Context,
+	tx pgx.Tx,
+	task *repository.DesktopTask,
+	errorJSON []byte,
 ) (*repository.DesktopTask, error) {
 	if task == nil {
 		return nil, response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found")
@@ -612,12 +736,13 @@ func markUnknownPublishTaskDefinitiveFailed(
 		    lease_token_hash = NULL,
 		    lease_expires_at = NULL,
 		    publish_submit_started_at = NULL,
+		    error = COALESCE($3, error),
 		    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)
+	`, task.DesktopID, task.WorkspaceID, nullableJSON(errorJSON))
 	var updated repository.DesktopTask
 	if err := row.Scan(
 		&updated.DesktopID,
@@ -643,9 +768,6 @@ func publishUnknownTaskHasDefinitiveFailure(task *repository.DesktopTask) bool {
 		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")),
@@ -653,7 +775,13 @@ func publishUnknownTaskHasDefinitiveFailure(task *repository.DesktopTask) bool {
 	if combined == "" {
 		return false
 	}
-	return definitivePublishFailurePattern.MatchString(combined)
+	if definitivePublishFailurePattern.MatchString(combined) {
+		return true
+	}
+	if boolValueFromMap(errorPayload, "publish_submit_uncertain") {
+		return false
+	}
+	return false
 }
 
 func boolValueFromMap(payload map[string]any, key string) bool {
@@ -675,7 +803,7 @@ func boolValueFromMap(payload map[string]any, key string) bool {
 	}
 }
 
-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`)
+var definitivePublishFailurePattern = regexp.MustCompile(`csrf|forbidden|\b403\b|not_logged_in|login_required|token_missing|cookie_missing|challenge_required|risk_control|captcha|验证码|滑块|人机验证|安全验证|风控|风险|参数错误|内容为空|标题.{0,12}(长度|字数|过长|太长|超过|不能超过)|title[_\s-]*(length|too[_\s-]*long|too[_\s-]*short|max)|image_fetch_failed|cover_fetch_failed|upload_failed|signature_failed`)
 
 func loadExistingPublishRecordTargets(
 	ctx context.Context,
@@ -821,10 +949,6 @@ func (s *PublishJobService) RetryByDesktopTask(
 	if task.Status == "queued" || task.Status == "in_progress" {
 		return nil, response.ErrConflict(40994, "desktop_task_retry_conflict", "desktop task is still pending or running")
 	}
-	if shouldRequeuePublishTaskInsteadOfCreatingNew(task) {
-		return s.requeueUnknownPublishTask(ctx, task)
-	}
-
 	payload := unmarshalJSONObject(task.Payload)
 	articleID, err := s.resolveRetryArticleID(ctx, client.TenantID, payload)
 	if err != nil {
@@ -853,6 +977,8 @@ func (s *PublishJobService) RetryByDesktopTask(
 		Accounts: []CreatePublishJobAccountRequest{
 			{AccountID: task.TargetAccountID.String()},
 		},
+		AllowSupersedeUnknown: true,
+		ManualRetry:           true,
 	})
 }
 
@@ -886,10 +1012,6 @@ func (s *PublishJobService) RetryTenantDesktopTask(
 	if task.Status == "queued" || task.Status == "in_progress" {
 		return nil, response.ErrConflict(40994, "desktop_task_retry_conflict", "desktop task is still pending or running")
 	}
-	if shouldRequeuePublishTaskInsteadOfCreatingNew(task) {
-		return s.requeueUnknownPublishTask(ctx, task)
-	}
-
 	payload := unmarshalJSONObject(task.Payload)
 	articleID, err := s.resolveRetryArticleID(ctx, actor.TenantID, payload)
 	if err != nil {
@@ -919,73 +1041,11 @@ func (s *PublishJobService) RetryTenantDesktopTask(
 		Accounts: []CreatePublishJobAccountRequest{
 			{AccountID: task.TargetAccountID.String()},
 		},
+		AllowSupersedeUnknown: true,
+		ManualRetry:           true,
 	})
 }
 
-func shouldRequeuePublishTaskInsteadOfCreatingNew(task *repository.DesktopTask) bool {
-	return task != nil &&
-		task.Kind == "publish" &&
-		task.Status == "unknown" &&
-		!publishUnknownTaskHasDefinitiveFailure(task)
-}
-
-func (s *PublishJobService) requeueUnknownPublishTask(
-	ctx context.Context,
-	task *repository.DesktopTask,
-) (*CreatePublishJobResponse, error) {
-	if task == nil {
-		return nil, response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found")
-	}
-	tx, err := s.pool.Begin(ctx)
-	if err != nil {
-		return nil, response.ErrInternal(50105, "desktop_task_reconcile_begin_failed", "failed to start desktop task reconcile")
-	}
-	defer tx.Rollback(ctx)
-
-	taskRepo := repository.NewDesktopTaskRepository(tx)
-	requeued, err := taskRepo.Reconcile(ctx, task.DesktopID, task.WorkspaceID, "retry", nil, nil)
-	if 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(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
-}
-
-func createPublishJobResponseForRequeuedTask(task *repository.DesktopTask) *CreatePublishJobResponse {
-	if task == nil {
-		return &CreatePublishJobResponse{
-			TaskIDs:         []string{},
-			CreatedTaskIDs:  []string{},
-			ExistingTaskIDs: []string{},
-			RequeuedTaskIDs: []string{},
-		}
-	}
-	taskID := task.DesktopID.String()
-	return &CreatePublishJobResponse{
-		JobID:           "",
-		TaskIDs:         []string{taskID},
-		CreatedTaskIDs:  []string{},
-		ExistingTaskIDs: []string{},
-		RequeuedTaskIDs: []string{taskID},
-	}
-}
-
 func (s *PublishJobService) authorizePublishTaskForClientUser(
 	ctx context.Context,
 	client *repository.DesktopClient,

server/internal/tenant/app/publish_job_service_test.go

@@ -155,62 +155,6 @@ func TestPublishTargetPlatformsSortsAndDedupes(t *testing.T) {
 	}
 }
 
-func TestShouldRequeuePublishTaskInsteadOfCreatingNewOnlyForUnknownPublish(t *testing.T) {
-	t.Parallel()
-
-	if !shouldRequeuePublishTaskInsteadOfCreatingNew(&repository.DesktopTask{
-		Kind:   "publish",
-		Status: "unknown",
-		Error:  []byte(`{"publish_submit_uncertain":true,"message":"network interrupted after submit"}`),
-	}) {
-		t.Fatal("submit-uncertain unknown publish tasks should be retried by requeueing the existing task")
-	}
-	if shouldRequeuePublishTaskInsteadOfCreatingNew(&repository.DesktopTask{
-		Kind:   "publish",
-		Status: "unknown",
-		Error:  []byte(`{"error_code":403,"error_msg":"CSRF token invalid"}`),
-	}) {
-		t.Fatal("definitive failed unknown publish tasks should create a new retry record")
-	}
-	if shouldRequeuePublishTaskInsteadOfCreatingNew(&repository.DesktopTask{
-		Kind:   "publish",
-		Status: "failed",
-	}) {
-		t.Fatal("failed publish tasks should keep create-new retry semantics")
-	}
-	if shouldRequeuePublishTaskInsteadOfCreatingNew(&repository.DesktopTask{
-		Kind:   "monitor",
-		Status: "unknown",
-	}) {
-		t.Fatal("monitor retry semantics are owned by desktop task reconcile")
-	}
-	if shouldRequeuePublishTaskInsteadOfCreatingNew(nil) {
-		t.Fatal("nil tasks should not be requeued")
-	}
-}
-
-func TestCreatePublishJobResponseForRequeuedTaskReportsRequeuedTaskID(t *testing.T) {
-	t.Parallel()
-
-	taskID := uuid.MustParse("55555555-5555-5555-5555-555555555555")
-	got := createPublishJobResponseForRequeuedTask(&repository.DesktopTask{DesktopID: taskID})
-	if got.JobID != "" {
-		t.Fatalf("JobID = %q, want empty for requeued task", got.JobID)
-	}
-	if len(got.TaskIDs) != 1 || got.TaskIDs[0] != taskID.String() {
-		t.Fatalf("TaskIDs = %v, want [%s]", got.TaskIDs, taskID)
-	}
-	if len(got.CreatedTaskIDs) != 0 {
-		t.Fatalf("CreatedTaskIDs = %v, want empty", got.CreatedTaskIDs)
-	}
-	if len(got.ExistingTaskIDs) != 0 {
-		t.Fatalf("ExistingTaskIDs = %v, want empty", got.ExistingTaskIDs)
-	}
-	if len(got.RequeuedTaskIDs) != 1 || got.RequeuedTaskIDs[0] != taskID.String() {
-		t.Fatalf("RequeuedTaskIDs = %v, want [%s]", got.RequeuedTaskIDs, taskID)
-	}
-}
-
 func TestPublishUnknownTaskHasDefinitiveFailureReleasesCSRF403(t *testing.T) {
 	t.Parallel()
 
@@ -237,6 +181,19 @@ func TestPublishUnknownTaskHasDefinitiveFailureKeepsSubmitUncertain(t *testing.T
 	}
 }
 
+func TestPublishUnknownTaskHasDefinitiveFailureReleasesTitleLengthValidation(t *testing.T) {
+	t.Parallel()
+
+	task := &repository.DesktopTask{
+		Kind:   "publish",
+		Status: "unknown",
+		Error:  []byte(`{"publish_submit_uncertain":true,"code":"toutiaohao_publish_failed","message":"标题长度应该在2-30字之间"}`),
+	}
+	if !publishUnknownTaskHasDefinitiveFailure(task) {
+		t.Fatal("title length validation should be treated as definitive failed")
+	}
+}
+
 func TestLoadExistingPublishRecordTargetsDedupeUnknownTasks(t *testing.T) {
 	t.Parallel()
 
@@ -298,6 +255,53 @@ func TestLockDesktopPublishDedupKeysSkipsEmptyList(t *testing.T) {
 	}
 }
 
+func TestReleaseManuallySupersededUnknownPublishTasksLocksUnknownDedupSet(t *testing.T) {
+	t.Parallel()
+
+	tx := &capturePublishDedupLookupTx{rows: emptyPublishDedupRows{}}
+	outcomes, superseded, err := releaseManuallySupersededUnknownPublishTasks(
+		context.Background(),
+		tx,
+		1,
+		2,
+		[]string{"dedup-a", "dedup-b"},
+	)
+	if err != nil {
+		t.Fatalf("releaseManuallySupersededUnknownPublishTasks() error = %v", err)
+	}
+	if len(outcomes) != 0 {
+		t.Fatalf("outcomes = %v, want empty", outcomes)
+	}
+	if len(superseded) != 0 {
+		t.Fatalf("superseded = %v, want empty", superseded)
+	}
+
+	normalizedSQL := normalizeSQLWhitespace(tx.sql)
+	for _, fragment := range []string{
+		"FROM desktop_tasks",
+		"tenant_id = $1",
+		"workspace_id = $2",
+		"kind = 'publish'",
+		"status = 'unknown'",
+		"dedup_key = ANY($3::text[])",
+		"FOR UPDATE",
+	} {
+		if !strings.Contains(normalizedSQL, fragment) {
+			t.Fatalf("manual supersede lookup missing %q: %s", fragment, tx.sql)
+		}
+	}
+	if !strings.Contains(normalizedSQL, "COALESCE(dedup_key, '')") {
+		t.Fatalf("manual supersede lookup should return dedup keys for new-task payload: %s", tx.sql)
+	}
+	if len(tx.args) != 3 || tx.args[0] != int64(1) || tx.args[1] != int64(2) {
+		t.Fatalf("manual supersede lookup args = %#v, want tenant/workspace/dedup keys", tx.args)
+	}
+	gotKeys, ok := tx.args[2].([]string)
+	if !ok || len(gotKeys) != 2 || gotKeys[0] != "dedup-a" || gotKeys[1] != "dedup-b" {
+		t.Fatalf("manual supersede lookup dedup keys = %#v, want [dedup-a dedup-b]", tx.args[2])
+	}
+}
+
 func TestIsPostgresUniqueViolationMatchesConstraint(t *testing.T) {
 	t.Parallel()
 

server/internal/tenant/transport/publish_job_handler.go

@@ -25,6 +25,7 @@ func (h *PublishJobHandler) Create(c *gin.Context) {
 		response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
 		return
 	}
+	req.AllowSupersedeUnknown = true
 
 	data, err := h.svc.Create(c.Request.Context(), auth.MustActor(c.Request.Context()), req)
 	if err != nil {