From a5d89f0c4876834883345fef43e88e0bd58df220 Mon Sep 17 00:00:00 2001 From: liangxu Date: Sun, 14 Jun 2026 21:03:00 +0800 Subject: [PATCH] fix(publish): cancel queued publish tasks when the article is unavailable Stale publish jobs for deleted or invalid articles previously lingered in the queue and kept failing compliance rechecks. Now they are cancelled cleanly and batch/article publish status is recalculated. - cancel queued desktop tasks, jobs and publish records on article delete - detect deleted articles and invalid-article-version compliance errors during the client publish recheck and cancel the affected tasks - add cancelQueuedPublishTasksForUnavailableArticle helper and a unit test for the compliance-error classifier Co-Authored-By: Claude Opus 4.8 (1M context) --- server/internal/tenant/app/article_service.go | 3 + .../tenant/app/desktop_task_service.go | 101 +++++++++++- .../tenant/app/desktop_task_service_test.go | 13 ++ .../tenant/app/publish_record_support.go | 150 ++++++++++++++++++ 4 files changed, 262 insertions(+), 5 deletions(-) diff --git a/server/internal/tenant/app/article_service.go b/server/internal/tenant/app/article_service.go index 5ac1669..1f2ddbd 100644 --- a/server/internal/tenant/app/article_service.go +++ b/server/internal/tenant/app/article_service.go @@ -874,6 +874,9 @@ func (s *ArticleService) Delete(ctx context.Context, id int64) error { if err != nil || tag.RowsAffected() == 0 { return response.ErrNotFound(40411, "article_not_found", "article not found") } + if err := cancelQueuedPublishTasksForUnavailableArticle(ctx, tx, actor.TenantID, id); err != nil { + return err + } if err := repository.NewImageReferenceRepository(tx).DeleteByArticle(ctx, actor.TenantID, id); err != nil { return response.ErrInternal(50010, "delete_failed", "failed to delete article image references") } diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 79ceb0c..fc7970e 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -287,18 +287,20 @@ func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Cont j.created_by_user_id, j.article_id, j.article_version_id, - COALESCE(array_agg(DISTINCT dt.platform_id) FILTER (WHERE dt.platform_id <> ''), '{}') AS platforms + COALESCE(array_agg(DISTINCT dt.platform_id) FILTER (WHERE dt.platform_id <> ''), '{}') AS platforms, + BOOL_OR(a.deleted_at IS NOT NULL) AS article_deleted FROM desktop_tasks dt JOIN desktop_publish_jobs j ON j.desktop_id = dt.job_id + LEFT JOIN articles a ON a.id = j.article_id AND a.tenant_id = j.tenant_id WHERE dt.target_client_id = $1 AND dt.kind = 'publish' AND dt.status = 'queued' AND j.status = 'queued' AND j.article_id IS NOT NULL AND j.article_version_id IS NOT NULL - GROUP BY j.desktop_id, j.tenant_id, j.workspace_id, j.created_by_user_id, j.article_id, j.article_version_id - ORDER BY MIN(dt.created_at) - LIMIT 5 + GROUP BY j.desktop_id, j.tenant_id, j.workspace_id, j.created_by_user_id, j.article_id, j.article_version_id + ORDER BY MIN(dt.created_at) + LIMIT 5 `, client.ID) if err != nil { s.logWarn("desktop publish compliance recheck query failed", err, zap.String("client_id", client.ID.String())) @@ -314,11 +316,12 @@ func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Cont ArticleID int64 ArticleVersionID int64 Platforms []string + ArticleDeleted bool } candidates := make([]candidate, 0) for rows.Next() { var item candidate - if scanErr := rows.Scan(&item.JobID, &item.TenantID, &item.WorkspaceID, &item.CreatedByUserID, &item.ArticleID, &item.ArticleVersionID, &item.Platforms); scanErr != nil { + if scanErr := rows.Scan(&item.JobID, &item.TenantID, &item.WorkspaceID, &item.CreatedByUserID, &item.ArticleID, &item.ArticleVersionID, &item.Platforms, &item.ArticleDeleted); scanErr != nil { s.logWarn("desktop publish compliance recheck scan failed", scanErr, zap.String("client_id", client.ID.String())) return response.ErrInternal(50118, "desktop_publish_compliance_recheck_failed", "failed to scan publish job for compliance recheck") } @@ -332,6 +335,12 @@ func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Cont } for _, item := range candidates { + if item.ArticleDeleted { + if cancelErr := s.cancelUnavailableArticleQueuedPublishTasks(ctx, item.TenantID, item.ArticleID); cancelErr != nil { + return cancelErr + } + continue + } gate, gateErr := s.compliance.GateForPublish(ctx, tenantcompliance.GateForPublishRequest{ TenantID: item.TenantID, ArticleID: item.ArticleID, @@ -347,6 +356,12 @@ func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Cont continue } if gateErr != nil { + if isComplianceInvalidArticleVersionError(gateErr) { + if cancelErr := s.cancelUnavailableArticleQueuedPublishTasks(ctx, item.TenantID, item.ArticleID); cancelErr != nil { + return cancelErr + } + continue + } return response.ErrServiceUnavailable(41004, "compliance_engine_unavailable", "scheduled publish compliance recheck failed") } if gate != nil && gate.Decision == sharedcompliance.GateDecisionBlock { @@ -358,6 +373,82 @@ func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Cont return nil } +func (s *DesktopTaskService) cancelUnavailableArticleQueuedPublishTasks(ctx context.Context, tenantID, articleID int64) error { + if s == nil || s.pool == nil || tenantID <= 0 || articleID <= 0 { + return nil + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return response.ErrInternal(50122, "publish_task_cancel_unavailable_article_begin_failed", "failed to start unavailable article publish task cleanup") + } + defer tx.Rollback(ctx) + + taskRefs, err := queuedPublishTaskRefsForArticle(ctx, tx, tenantID, articleID) + if err != nil { + return err + } + if err := cancelQueuedPublishTasksForUnavailableArticle(ctx, tx, tenantID, articleID); err != nil { + return err + } + if err := tx.Commit(ctx); err != nil { + return response.ErrInternal(50123, "publish_task_cancel_unavailable_article_commit_failed", "failed to commit unavailable article publish task cleanup") + } + + for _, ref := range taskRefs { + task, getErr := s.repo.GetByDesktopID(ctx, ref.TaskID, ref.WorkspaceID) + if getErr == nil && task != nil { + s.publishTaskEvent(ctx, task, "task_canceled") + } + } + return nil +} + +type desktopTaskRef struct { + TaskID uuid.UUID + WorkspaceID int64 +} + +func queuedPublishTaskRefsForArticle(ctx context.Context, tx pgx.Tx, tenantID, articleID int64) ([]desktopTaskRef, error) { + if tx == nil || tenantID <= 0 || articleID <= 0 { + return nil, nil + } + rows, err := tx.Query(ctx, ` + SELECT dt.desktop_id, dt.workspace_id + FROM desktop_tasks dt + JOIN desktop_publish_jobs j + ON j.desktop_id = dt.job_id + AND j.tenant_id = dt.tenant_id + WHERE dt.tenant_id = $1 + AND dt.kind = 'publish' + AND dt.status = 'queued' + AND j.article_id = $2 + AND j.status = 'queued' + ORDER BY dt.created_at ASC, dt.desktop_id ASC + `, tenantID, articleID) + if err != nil { + return nil, response.ErrInternal(50124, "publish_task_cancel_unavailable_article_lookup_failed", "failed to inspect unavailable article publish tasks") + } + defer rows.Close() + + taskRefs := make([]desktopTaskRef, 0) + for rows.Next() { + var ref desktopTaskRef + if scanErr := rows.Scan(&ref.TaskID, &ref.WorkspaceID); scanErr != nil { + return nil, response.ErrInternal(50124, "publish_task_cancel_unavailable_article_lookup_failed", "failed to scan unavailable article publish tasks") + } + taskRefs = append(taskRefs, ref) + } + if err := rows.Err(); err != nil { + return nil, response.ErrInternal(50124, "publish_task_cancel_unavailable_article_lookup_failed", "failed to iterate unavailable article publish tasks") + } + return taskRefs, nil +} + +func isComplianceInvalidArticleVersionError(err error) bool { + appErr, ok := err.(*response.AppError) + return ok && appErr.Code == 41005 && appErr.Message == "invalid_article_version" +} + const desktopTaskRepositoryReturningColumns = ` t.desktop_id, t.job_id, diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index 4cbe358..5c52edf 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/server/internal/tenant/app/desktop_task_service_test.go @@ -4,6 +4,8 @@ import ( "regexp" "strings" "testing" + + "github.com/geo-platform/tenant-api/internal/shared/response" ) func TestDesktopTaskRecoverySelectQueryMatchesRecoveredLeaseScanOrder(t *testing.T) { @@ -136,6 +138,17 @@ func TestResolvePublishRecoveryOutcome(t *testing.T) { } } +func TestIsComplianceInvalidArticleVersionError(t *testing.T) { + t.Parallel() + + if !isComplianceInvalidArticleVersionError(response.ErrBadRequest(41005, "invalid_article_version", "article version does not belong to this article")) { + t.Fatal("invalid article version compliance errors should be treated as stale publish tasks") + } + if isComplianceInvalidArticleVersionError(response.ErrServiceUnavailable(41004, "compliance_engine_unavailable", "compliance gate failed closed")) { + t.Fatal("compliance engine failures should not be treated as stale publish tasks") + } +} + func recoverDesktopTaskSelectColumns(query string) []string { re := regexp.MustCompile(`(?is)select\s+(.*?)\s+from\s+desktop_tasks`) match := re.FindStringSubmatch(query) diff --git a/server/internal/tenant/app/publish_record_support.go b/server/internal/tenant/app/publish_record_support.go index 7c39999..67e5090 100644 --- a/server/internal/tenant/app/publish_record_support.go +++ b/server/internal/tenant/app/publish_record_support.go @@ -111,6 +111,156 @@ func createPublishBatchRecords( return publishBatchID, recordIDs, nil } +func cancelQueuedPublishTasksForUnavailableArticle(ctx context.Context, tx pgx.Tx, tenantID, articleID int64) error { + if tx == nil || tenantID <= 0 || articleID <= 0 { + return nil + } + errorJSON, err := marshalOptionalJSON(map[string]any{ + "code": "article_unavailable", + "message": "article is unavailable before publishing", + "source": "article_unavailable_cleanup", + }) + if err != nil { + return response.ErrInternal(50120, "publish_task_cancel_payload_failed", "failed to encode publish task cancellation payload") + } + + if _, err := tx.Exec(ctx, ` + WITH affected_records AS ( + UPDATE publish_records pr + SET status = 'cancelled', + error_message = COALESCE(NULLIF(error_message, ''), 'article_unavailable'), + updated_at = NOW() + FROM desktop_publish_jobs j + JOIN desktop_tasks dt + ON dt.job_id = j.desktop_id + AND dt.tenant_id = j.tenant_id + WHERE pr.id::text = dt.payload->>'publish_record_id' + AND pr.tenant_id = $1 + AND pr.article_id = $2 + AND pr.status IN ('queued', 'publishing', 'pending', 'running') + AND j.tenant_id = $1 + AND j.article_id = $2 + AND j.status = 'queued' + AND dt.kind = 'publish' + AND dt.status = 'queued' + RETURNING pr.publish_batch_id + ), + affected_batches AS ( + SELECT DISTINCT publish_batch_id + FROM affected_records + WHERE publish_batch_id IS NOT NULL + ), + updated_tasks AS ( + UPDATE desktop_tasks dt + SET status = 'aborted', + error = $3, + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + updated_at = NOW() + FROM desktop_publish_jobs j + WHERE dt.job_id = j.desktop_id + AND dt.tenant_id = j.tenant_id + AND dt.tenant_id = $1 + AND dt.kind = 'publish' + AND dt.status = 'queued' + AND j.article_id = $2 + AND j.status = 'queued' + RETURNING dt.job_id + ), + updated_jobs AS ( + UPDATE desktop_publish_jobs j + SET status = 'cancelled', + updated_at = NOW() + WHERE j.tenant_id = $1 + AND j.article_id = $2 + AND j.status = 'queued' + AND EXISTS ( + SELECT 1 + FROM updated_tasks ut + WHERE ut.job_id = j.desktop_id + ) + RETURNING j.desktop_id + ), + recalculated_batches AS ( + UPDATE publish_batches pb + SET status = CASE + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.publish_batch_id = pb.id + AND pr.status IN ('queued', 'publishing', 'pending', 'running') + ) THEN 'publishing' + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.publish_batch_id = pb.id + AND pr.status IN ('success', 'published', 'publish_success') + ) AND EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.publish_batch_id = pb.id + AND pr.status NOT IN ('success', 'published', 'publish_success') + ) THEN 'partial_success' + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.publish_batch_id = pb.id + AND pr.status IN ('success', 'published', 'publish_success') + ) THEN 'success' + ELSE 'failed' + END, + updated_at = NOW() + FROM affected_batches ab + WHERE pb.id = ab.publish_batch_id + RETURNING pb.id + ), + recalculated_article AS ( + UPDATE articles a + SET publish_status = CASE + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.article_id = a.id + AND pr.tenant_id = a.tenant_id + AND pr.status IN ('queued', 'publishing', 'pending', 'running') + ) THEN 'publishing' + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.article_id = a.id + AND pr.tenant_id = a.tenant_id + AND pr.status IN ('success', 'published', 'publish_success') + ) AND EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.article_id = a.id + AND pr.tenant_id = a.tenant_id + AND pr.status NOT IN ('success', 'published', 'publish_success') + ) THEN 'partial_success' + WHEN EXISTS ( + SELECT 1 + FROM publish_records pr + WHERE pr.article_id = a.id + AND pr.tenant_id = a.tenant_id + AND pr.status IN ('success', 'published', 'publish_success') + ) THEN 'success' + ELSE 'failed' + END, + updated_at = NOW() + WHERE a.tenant_id = $1 + AND a.id = $2 + AND a.deleted_at IS NULL + AND EXISTS (SELECT 1 FROM updated_jobs) + RETURNING a.id + ) + SELECT COUNT(1) FROM updated_jobs + `, tenantID, articleID, errorJSON); err != nil { + return response.ErrInternal(50121, "publish_task_cancel_deleted_article_failed", "failed to cancel queued publish tasks for deleted article") + } + return nil +} + func syncDesktopPublishTaskState( ctx context.Context, tx pgx.Tx,