From 8c0fbc8ffac8042761787291b857668291de3f02 Mon Sep 17 00:00:00 2001 From: liangxu Date: Mon, 11 May 2026 12:26:54 +0800 Subject: [PATCH] fix(desktop-task): drop LEFT JOIN in lease query so FOR UPDATE locks only desktop_tasks Postgres rejects FOR UPDATE on the nullable side of a LEFT JOIN, which made LeaseNextQueuedDesktopTask fail under contention. Replace the join with a NOT EXISTS subquery, scope the row lock to the dt alias, and add a regression test plus warn-level logging on lease query failures so similar issues are easier to diagnose. --- .../tenant/app/desktop_task_service.go | 10 ++++++++++ .../repository/generated/desktop_task.sql.go | 11 ++++++---- .../generated/desktop_task_sql_test.go | 20 +++++++++++++++++++ .../repository/queries/desktop_task.sql | 11 ++++++---- 4 files changed, 44 insertions(+), 8 deletions(-) create mode 100644 server/internal/tenant/repository/generated/desktop_task_sql_test.go diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 09f8e04..694f6f6 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -223,6 +223,16 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt } return nil, s.classifyLeaseError(ctx, client, taskID) } + fields := []zap.Field{ + zap.String("client_id", client.ID.String()), + } + if req.Kind != nil { + fields = append(fields, zap.String("requested_kind", strings.TrimSpace(*req.Kind))) + } + if taskID != nil { + fields = append(fields, zap.String("task_id", taskID.String())) + } + s.logWarn("desktop task lease query failed", err, fields...) return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to lease desktop task") } diff --git a/server/internal/tenant/repository/generated/desktop_task.sql.go b/server/internal/tenant/repository/generated/desktop_task.sql.go index 2cc5ec9..3bfdce2 100644 --- a/server/internal/tenant/repository/generated/desktop_task.sql.go +++ b/server/internal/tenant/repository/generated/desktop_task.sql.go @@ -558,7 +558,6 @@ const leaseNextQueuedDesktopTask = `-- name: LeaseNextQueuedDesktopTask :one WITH candidate AS ( SELECT dt.desktop_id FROM desktop_tasks AS dt - LEFT JOIN desktop_publish_jobs AS j ON j.desktop_id = dt.job_id WHERE dt.target_client_id = $3 AND dt.status = 'queued' AND ( @@ -567,8 +566,12 @@ WITH candidate AS ( ) AND ( dt.kind <> 'publish' - OR j.desktop_id IS NULL - OR j.status = 'queued' + OR NOT EXISTS ( + SELECT 1 + FROM desktop_publish_jobs AS j + WHERE j.desktop_id = dt.job_id + AND j.status <> 'queued' + ) ) ORDER BY dt.lane_weight DESC, dt.priority DESC, @@ -576,7 +579,7 @@ WITH candidate AS ( dt.created_at ASC, dt.desktop_id ASC LIMIT 1 - FOR UPDATE SKIP LOCKED + FOR UPDATE OF dt SKIP LOCKED ) UPDATE desktop_tasks AS t SET active_attempt_id = $1, diff --git a/server/internal/tenant/repository/generated/desktop_task_sql_test.go b/server/internal/tenant/repository/generated/desktop_task_sql_test.go new file mode 100644 index 0000000..7807909 --- /dev/null +++ b/server/internal/tenant/repository/generated/desktop_task_sql_test.go @@ -0,0 +1,20 @@ +package generated + +import ( + "strings" + "testing" +) + +func TestLeaseNextQueuedDesktopTaskLocksOnlyDesktopTasks(t *testing.T) { + query := leaseNextQueuedDesktopTask + + if !strings.Contains(query, "FOR UPDATE OF dt SKIP LOCKED") { + t.Fatalf("lease query must lock only desktop_tasks alias dt; query:\n%s", query) + } + if strings.Contains(query, "LEFT JOIN desktop_publish_jobs") { + t.Fatalf("lease query must not combine LEFT JOIN with FOR UPDATE; query:\n%s", query) + } + if !strings.Contains(query, "NOT EXISTS") { + t.Fatalf("lease query must filter non-queued publish jobs without joining the lock target; query:\n%s", query) + } +} diff --git a/server/internal/tenant/repository/queries/desktop_task.sql b/server/internal/tenant/repository/queries/desktop_task.sql index 2a7a832..d9faeee 100644 --- a/server/internal/tenant/repository/queries/desktop_task.sql +++ b/server/internal/tenant/repository/queries/desktop_task.sql @@ -63,7 +63,6 @@ LIMIT 1; WITH candidate AS ( SELECT dt.desktop_id FROM desktop_tasks AS dt - LEFT JOIN desktop_publish_jobs AS j ON j.desktop_id = dt.job_id WHERE dt.target_client_id = sqlc.arg(client_id) AND dt.status = 'queued' AND ( @@ -72,8 +71,12 @@ WITH candidate AS ( ) AND ( dt.kind <> 'publish' - OR j.desktop_id IS NULL - OR j.status = 'queued' + OR NOT EXISTS ( + SELECT 1 + FROM desktop_publish_jobs AS j + WHERE j.desktop_id = dt.job_id + AND j.status <> 'queued' + ) ) ORDER BY dt.lane_weight DESC, dt.priority DESC, @@ -81,7 +84,7 @@ WITH candidate AS ( dt.created_at ASC, dt.desktop_id ASC LIMIT 1 - FOR UPDATE SKIP LOCKED + FOR UPDATE OF dt SKIP LOCKED ) UPDATE desktop_tasks AS t SET active_attempt_id = sqlc.arg(attempt_id),