From aecdb87fa36015da3f37ad4853ea6340731139f8 Mon Sep 17 00:00:00 2001 From: liangxu Date: Wed, 8 Jul 2026 22:32:06 +0800 Subject: [PATCH] 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) --- .../tenant/app/publish_job_service.go | 232 +++++++++++------- .../tenant/app/publish_job_service_test.go | 116 ++++----- .../tenant/transport/publish_job_handler.go | 1 + 3 files changed, 207 insertions(+), 142 deletions(-) diff --git a/server/internal/tenant/app/publish_job_service.go b/server/internal/tenant/app/publish_job_service.go index 16a89eb..37e21fb 100644 --- a/server/internal/tenant/app/publish_job_service.go +++ b/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, diff --git a/server/internal/tenant/app/publish_job_service_test.go b/server/internal/tenant/app/publish_job_service_test.go index 1c46394..e2988a2 100644 --- a/server/internal/tenant/app/publish_job_service_test.go +++ b/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() diff --git a/server/internal/tenant/transport/publish_job_handler.go b/server/internal/tenant/transport/publish_job_handler.go index 02d94b5..2484256 100644 --- a/server/internal/tenant/transport/publish_job_handler.go +++ b/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 {