root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(publish): cancel queued publish tasks when the article is unavailable

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
liangxu
2026-06-14 21:03:00 +08:00
parent c6b7090536
commit a5d89f0c48
4 changed files with 262 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/desktop_task_service.go
+96 -5
View File
@@ -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,