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.

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")
 	}
 

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,

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)
+	}
+}

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),