root
832d7f507d
Add CancelStaleQueuedPublishTasks to abort publish tasks that sit in the queue past a configurable timeout (default 3 days) without being claimed by a desktop client, so they no longer linger indefinitely as 'queued'. The lease recovery worker runs the cleanup each cycle and reports a cancelled_queued count; the timeout is configurable via job run config. Aborted tasks carry a publish_queue_timeout error payload with a readable Chinese duration, and the admin-web publish summary surfaces it as an auto-cancelled queue-timeout message. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
560 lines
18 KiB
Go
560 lines
18 KiB
Go
package app
|
|
|
|
import (
|
|
"errors"
|
|
"reflect"
|
|
"regexp"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/geo-platform/tenant-api/internal/shared/response"
|
|
"github.com/geo-platform/tenant-api/internal/tenant/repository"
|
|
)
|
|
|
|
func TestDesktopTaskRecoverySelectQueryMatchesRecoveredLeaseScanOrder(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
query := desktopTaskRecoverySelectQuery(desktopTaskRecoveryModeStartup)
|
|
gotColumns := recoverDesktopTaskSelectColumns(query)
|
|
wantColumns := []string{
|
|
"desktop_id",
|
|
"workspace_id",
|
|
"kind",
|
|
"status",
|
|
"active_attempt_id",
|
|
"attempts",
|
|
"publish_submit_started_at",
|
|
"monitor_task_id",
|
|
}
|
|
|
|
if len(gotColumns) != len(wantColumns) {
|
|
t.Fatalf("recover select columns = %v, want %v", gotColumns, wantColumns)
|
|
}
|
|
for index := range wantColumns {
|
|
if gotColumns[index] != wantColumns[index] {
|
|
t.Fatalf("recover select columns = %v, want %v", gotColumns, wantColumns)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestDesktopTaskRecoverySelectQueryLeaseExpiryFiltersExpiredLeases(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
query := desktopTaskRecoverySelectQuery(desktopTaskRecoveryModeLeaseExpiry)
|
|
|
|
for _, fragment := range []string{
|
|
"AND lease_expires_at IS NOT NULL",
|
|
"AND lease_expires_at < NOW()",
|
|
"FOR UPDATE",
|
|
} {
|
|
if !strings.Contains(query, fragment) {
|
|
t.Fatalf("recover lease-expiry query missing %q: %s", fragment, query)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestStaleQueuedPublishTasksCandidateSQLUsesExecutableQueueAge(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
query := normalizeSQLForDesktopTaskTest(staleQueuedPublishTasksCandidateSQL())
|
|
for _, fragment := range []string{
|
|
"FROM desktop_tasks AS t",
|
|
"JOIN desktop_publish_jobs AS j",
|
|
"t.kind = 'publish'",
|
|
"t.status = 'queued'",
|
|
"GREATEST( COALESCE(t.enqueued_at, t.created_at), COALESCE(j.scheduled_at, t.created_at) ) < $1",
|
|
"LIMIT $2",
|
|
"FOR UPDATE OF t SKIP LOCKED",
|
|
} {
|
|
if !strings.Contains(query, fragment) {
|
|
t.Fatalf("stale queued publish query missing %q: %s", fragment, query)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestPublishQueueTimeoutErrorPayloadIncludesConfig(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
payload, err := publishQueueTimeoutErrorPayload(3 * 24 * time.Hour)
|
|
if err != nil {
|
|
t.Fatalf("publishQueueTimeoutErrorPayload() error = %v", err)
|
|
}
|
|
text := string(payload)
|
|
for _, fragment := range []string{
|
|
`"code":"publish_queue_timeout"`,
|
|
`发布任务在等待通道中超过 3 天未被桌面端领取`,
|
|
`请确认对应桌面端在线、账号登录正常且任务通道在线后，再重新发布。`,
|
|
`"source":"publish_queue_timeout"`,
|
|
`"queued_timeout_seconds":259200`,
|
|
} {
|
|
if !strings.Contains(text, fragment) {
|
|
t.Fatalf("queue timeout payload missing %q: %s", fragment, text)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestPublishQueueTimeoutDisplayDurationUsesReadableChinese(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cases := []struct {
|
|
name string
|
|
timeout time.Duration
|
|
want string
|
|
}{
|
|
{name: "days", timeout: 7 * 24 * time.Hour, want: "7 天"},
|
|
{name: "hours", timeout: 36 * time.Hour, want: "36 小时"},
|
|
{name: "hours and minutes", timeout: 90 * time.Minute, want: "1 小时 30 分钟"},
|
|
{name: "minutes", timeout: 45 * time.Minute, want: "45 分钟"},
|
|
}
|
|
|
|
for _, tc := range cases {
|
|
tc := tc
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
t.Parallel()
|
|
if got := publishQueueTimeoutDisplayDuration(tc.timeout); got != tc.want {
|
|
t.Fatalf("publishQueueTimeoutDisplayDuration(%s) = %q, want %q", tc.timeout, got, tc.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestReconcileRetryPublishesTaskAvailableForAllDesktopTaskKinds(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
for _, kind := range []string{"publish", "monitor"} {
|
|
if got := reconcileDesktopTaskEventType(kind, "retry"); got != "task_available" {
|
|
t.Fatalf("reconcileDesktopTaskEventType(%q, retry) = %q, want task_available", kind, got)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestDesktopTaskRecoveryPayloadPublishLeaseExpiryRequeuesInsteadOfUnknown(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
payload, reason, message, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "publish")
|
|
if err != nil {
|
|
t.Fatalf("desktopTaskRecoveryPayload returned error: %v", err)
|
|
}
|
|
if reason != "lease_expired" {
|
|
t.Fatalf("reason = %q, want lease_expired", reason)
|
|
}
|
|
if !strings.Contains(message, "re-queued for retry") {
|
|
t.Fatalf("message = %q, want re-queued for retry", message)
|
|
}
|
|
if strings.Contains(message, "unknown") {
|
|
t.Fatalf("message must not move publish recovery to unknown: %q", message)
|
|
}
|
|
if !strings.Contains(string(payload), "desktop_task_lease_expiry") {
|
|
t.Fatalf("payload missing source: %s", payload)
|
|
}
|
|
}
|
|
|
|
func TestDesktopTaskRecoveryPayloadPublishFinalFailsAfterMaxAttempts(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
payload, _, message, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "publish_final")
|
|
if err != nil {
|
|
t.Fatalf("desktopTaskRecoveryPayload returned error: %v", err)
|
|
}
|
|
if !strings.Contains(message, "marked failed") {
|
|
t.Fatalf("message = %q, want marked failed", message)
|
|
}
|
|
if !strings.Contains(message, "3 attempts") {
|
|
t.Fatalf("message = %q, want max attempts", message)
|
|
}
|
|
if !strings.Contains(string(payload), "marked failed") {
|
|
t.Fatalf("payload missing final failure message: %s", payload)
|
|
}
|
|
}
|
|
|
|
func TestDesktopTaskRecoveryPayloadMonitorLeaseExpiryRequeues(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
payload, reason, message, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "monitor")
|
|
if err != nil {
|
|
t.Fatalf("desktopTaskRecoveryPayload returned error: %v", err)
|
|
}
|
|
if reason != "lease_expired" {
|
|
t.Fatalf("reason = %q, want lease_expired", reason)
|
|
}
|
|
if !strings.Contains(message, "re-queued") {
|
|
t.Fatalf("message = %q, want re-queued", message)
|
|
}
|
|
if strings.Contains(message, "marked failed") {
|
|
t.Fatalf("message must not fail expired monitor tasks: %q", message)
|
|
}
|
|
if !strings.Contains(string(payload), "desktop_task_lease_expiry") {
|
|
t.Fatalf("payload missing source: %s", payload)
|
|
}
|
|
}
|
|
|
|
func TestReconcileNonRetryPublishesTaskReconciled(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
if got := reconcileDesktopTaskEventType("publish", "failed"); got != "task_reconciled" {
|
|
t.Fatalf("reconcileDesktopTaskEventType(publish, failed) = %q, want task_reconciled", got)
|
|
}
|
|
}
|
|
|
|
func TestResolvePublishRecoveryOutcome(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cases := []struct {
|
|
name string
|
|
submitStarted bool
|
|
attempts int
|
|
wantStatus string
|
|
wantKind string
|
|
}{
|
|
{"pre-submit first attempt requeues", false, 1, "queued", "publish"},
|
|
{"pre-submit just below cap requeues", false, desktopPublishMaxAttempts - 1, "queued", "publish"},
|
|
{"pre-submit at cap fails terminally", false, desktopPublishMaxAttempts, "failed", "publish_final"},
|
|
{"pre-submit above cap fails terminally", false, desktopPublishMaxAttempts + 1, "failed", "publish_final"},
|
|
// Submit may have happened: never auto-requeue or silently fail — route to unknown
|
|
// (manual reconcile) regardless of remaining attempts, to avoid duplicate publishing.
|
|
{"submit started first attempt is uncertain", true, 1, "unknown", "publish_uncertain"},
|
|
{"submit started at cap is still uncertain", true, desktopPublishMaxAttempts, "unknown", "publish_uncertain"},
|
|
}
|
|
|
|
for _, tc := range cases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
t.Parallel()
|
|
gotStatus, gotKind := resolvePublishRecoveryOutcome(tc.submitStarted, tc.attempts)
|
|
if gotStatus != tc.wantStatus || gotKind != tc.wantKind {
|
|
t.Fatalf("resolvePublishRecoveryOutcome(%v, %d) = (%q, %q), want (%q, %q)",
|
|
tc.submitStarted, tc.attempts, gotStatus, gotKind, tc.wantStatus, tc.wantKind)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestResolveMonitorRecoveryOutcome(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cases := []struct {
|
|
name string
|
|
mode desktopTaskRecoveryMode
|
|
attempts int
|
|
wantStatus string
|
|
wantKind string
|
|
}{
|
|
{"startup first attempt requeues", desktopTaskRecoveryModeStartup, 1, "queued", "monitor"},
|
|
{"disconnect first attempt requeues", desktopTaskRecoveryModeDisconnect, 1, "queued", "monitor"},
|
|
{"startup repeated recovery fails", desktopTaskRecoveryModeStartup, 2, "failed", "monitor_final"},
|
|
{"disconnect repeated recovery fails", desktopTaskRecoveryModeDisconnect, 2, "failed", "monitor_final"},
|
|
{"lease expiry requeues first attempt", desktopTaskRecoveryModeLeaseExpiry, 1, "queued", "monitor"},
|
|
{"lease expiry requeues repeated attempt", desktopTaskRecoveryModeLeaseExpiry, 2, "queued", "monitor"},
|
|
}
|
|
|
|
for _, tc := range cases {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
t.Parallel()
|
|
gotStatus, gotKind := resolveMonitorRecoveryOutcome(tc.mode, tc.attempts)
|
|
if gotStatus != tc.wantStatus || gotKind != tc.wantKind {
|
|
t.Fatalf("resolveMonitorRecoveryOutcome(%q, %d) = (%q, %q), want (%q, %q)",
|
|
tc.mode, tc.attempts, gotStatus, gotKind, tc.wantStatus, tc.wantKind)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestNormalizeDesktopTaskCompletionStatusForKindFailsMonitorUnknown(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
if got := normalizeDesktopTaskCompletionStatusForKind("monitor", "unknown"); got != "failed" {
|
|
t.Fatalf("monitor completion status = %q, want failed", got)
|
|
}
|
|
if got := normalizeDesktopTaskCompletionStatusForKind("publish", "unknown"); got != "unknown" {
|
|
t.Fatalf("publish completion status = %q, want unknown", got)
|
|
}
|
|
}
|
|
|
|
func TestNormalizeDesktopTaskCompletionStatusForTaskFailsDefinitivePublishUnknown(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
got := normalizeDesktopTaskCompletionStatusForTask(
|
|
"publish",
|
|
"unknown",
|
|
nil,
|
|
[]byte(`{"publish_submit_uncertain":true,"code":"toutiaohao_publish_failed","message":"标题长度应该在2-30字之间"}`),
|
|
)
|
|
if got != "failed" {
|
|
t.Fatalf("publish title validation completion status = %q, want failed", got)
|
|
}
|
|
|
|
got = normalizeDesktopTaskCompletionStatusForTask(
|
|
"publish",
|
|
"unknown",
|
|
nil,
|
|
[]byte(`{"publish_submit_uncertain":true,"message":"network interrupted after submit"}`),
|
|
)
|
|
if got != "unknown" {
|
|
t.Fatalf("ambiguous publish completion status = %q, want unknown", got)
|
|
}
|
|
|
|
got = normalizeDesktopTaskCompletionStatusForTask(
|
|
"publish",
|
|
"unknown",
|
|
[]byte(`{"manual_retry":true}`),
|
|
[]byte(`{"publish_submit_uncertain":true,"message":"network interrupted after submit"}`),
|
|
)
|
|
if got != "failed" {
|
|
t.Fatalf("manual retry publish completion status = %q, want failed", got)
|
|
}
|
|
}
|
|
|
|
func TestExpiredMonitorDesktopTasksCandidateSQLSelectsOnlyExpiredInProgressMonitorTasks(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
normalized := normalizeSQLForDesktopTaskTest(expiredMonitorDesktopTasksCandidateSQL())
|
|
for _, fragment := range []string{
|
|
"t.kind = 'monitor'",
|
|
"t.status = 'in_progress'",
|
|
"t.monitor_task_id IS NOT NULL",
|
|
"t.lease_expires_at IS NOT NULL",
|
|
"t.lease_expires_at < NOW()",
|
|
"ORDER BY t.lease_expires_at ASC",
|
|
"FOR UPDATE OF t SKIP LOCKED",
|
|
} {
|
|
if !strings.Contains(normalized, fragment) {
|
|
t.Fatalf("expired monitor candidate query missing %q: %s", fragment, normalized)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestRequeueExpiredMonitorDesktopTasksSQLReturnsTaskToQueue(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
normalized := normalizeSQLForDesktopTaskTest(requeueExpiredMonitorDesktopTasksSQL())
|
|
for _, fragment := range []string{
|
|
"SET status = 'queued'",
|
|
"active_attempt_id = NULL",
|
|
"lease_token_hash = NULL",
|
|
"lease_expires_at = NULL",
|
|
"interrupt_reason = COALESCE(interrupt_reason, $3)",
|
|
"enqueued_at = NOW()",
|
|
"WHERE desktop_id = ANY($1::uuid[])",
|
|
"AND kind = 'monitor'",
|
|
"AND status = 'in_progress'",
|
|
"RETURNING desktop_id",
|
|
} {
|
|
if !strings.Contains(normalized, fragment) {
|
|
t.Fatalf("expired monitor update query missing %q: %s", fragment, normalized)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestAbortExpiredMonitorDesktopTasksSQLTerminatesUnrecoverableTask(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
normalized := normalizeSQLForDesktopTaskTest(abortExpiredMonitorDesktopTasksSQL())
|
|
for _, fragment := range []string{
|
|
"SET status = 'aborted'",
|
|
"active_attempt_id = NULL",
|
|
"lease_token_hash = NULL",
|
|
"lease_expires_at = NULL",
|
|
"interrupt_reason = COALESCE(interrupt_reason, $3)",
|
|
"WHERE desktop_id = ANY($1::uuid[])",
|
|
"AND kind = 'monitor'",
|
|
"AND status = 'in_progress'",
|
|
"RETURNING desktop_id",
|
|
} {
|
|
if !strings.Contains(normalized, fragment) {
|
|
t.Fatalf("expired monitor abort query missing %q: %s", fragment, normalized)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestClassifyLeaseErrorDefersQueuedMonitorTask(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
task := &repository.DesktopTask{
|
|
Kind: "monitor",
|
|
Status: "queued",
|
|
}
|
|
if got := classifyDesktopTaskLeaseState(task); !errors.Is(got, errDesktopTaskLeaseDeferred) {
|
|
t.Fatalf("classifyDesktopTaskLeaseState() = %v, want deferred", got)
|
|
}
|
|
}
|
|
|
|
func TestNormalizeDesktopTaskViewStatusMapsMonitorUnknownToFailedButKeepsPublishUnknown(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
if got := normalizeDesktopTaskViewStatus("monitor", "unknown"); got != "failed" {
|
|
t.Fatalf("monitor unknown view status = %q, want failed", got)
|
|
}
|
|
if got := normalizeDesktopTaskViewStatus("publish", "unknown"); got != "unknown" {
|
|
t.Fatalf("publish unknown view status = %q, want unknown", got)
|
|
}
|
|
}
|
|
|
|
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 TestBuildListPublishTasksByStatusesQueryHidesDeletedPublishRecords(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
query, args := buildListPublishTasksByStatusesQuery(
|
|
publishTaskListOwner{TenantID: 7, WorkspaceID: 11, UserID: 23},
|
|
[]string{"succeeded", "failed"},
|
|
"合肥",
|
|
10,
|
|
20,
|
|
"t.updated_at DESC",
|
|
)
|
|
normalized := normalizeSQLForDesktopTaskTest(query)
|
|
|
|
for _, fragment := range []string{
|
|
"NOT (t.payload ? 'publish_record_id') OR EXISTS",
|
|
"FROM publish_records pr",
|
|
"pr.tenant_id = t.tenant_id",
|
|
"t.payload->>'publish_record_id')::bigint",
|
|
"pr.deleted_at IS NULL",
|
|
"LIMIT $6 OFFSET $7",
|
|
} {
|
|
if !strings.Contains(normalized, fragment) {
|
|
t.Fatalf("list publish tasks query missing %q: %s", fragment, normalized)
|
|
}
|
|
}
|
|
if len(args) != 7 {
|
|
t.Fatalf("args len = %d, want 7", len(args))
|
|
}
|
|
}
|
|
|
|
func TestBuildCountPublishTasksByStatusesQueryHidesDeletedPublishRecords(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
query, args := buildCountPublishTasksByStatusesQuery(
|
|
publishTaskListOwner{TenantID: 7, WorkspaceID: 11, UserID: 23},
|
|
[]string{"succeeded", "failed"},
|
|
"",
|
|
)
|
|
normalized := normalizeSQLForDesktopTaskTest(query)
|
|
|
|
for _, fragment := range []string{
|
|
"NOT (t.payload ? 'publish_record_id') OR EXISTS",
|
|
"FROM publish_records pr",
|
|
"pr.deleted_at IS NULL",
|
|
} {
|
|
if !strings.Contains(normalized, fragment) {
|
|
t.Fatalf("count publish tasks query missing %q: %s", fragment, normalized)
|
|
}
|
|
}
|
|
if len(args) != 5 {
|
|
t.Fatalf("args len = %d, want 5", len(args))
|
|
}
|
|
}
|
|
|
|
func TestLeaseNextQueuedMonitorTaskSQLUsesAccountIdentityAndClientSlots(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
normalized := normalizeSQLForDesktopTaskTest(leaseNextQueuedMonitorTaskSQL())
|
|
for _, fragment := range []string{
|
|
"WITH current_accounts AS",
|
|
"client_id = $3",
|
|
"COALESCE(array_length($7::text[], 1), 0) = 0 OR dt.platform_id = ANY($7::text[])",
|
|
"dt.target_account_id = target_account.desktop_id",
|
|
"target_account.account_fingerprint",
|
|
"target_account.platform_uid",
|
|
"hashtextextended",
|
|
"pg_try_advisory_xact_lock(account_slot_lock_key)",
|
|
"SET target_client_id = $3, target_account_id = candidate.lease_account_id",
|
|
"NOT EXISTS ( SELECT 1 FROM desktop_tasks AS active",
|
|
"active.platform_id = dt.platform_id",
|
|
"active.status = 'in_progress'",
|
|
"active.desktop_id <> dt.desktop_id",
|
|
"COUNT(DISTINCT active_platform.platform_id)",
|
|
"active_platform.target_client_id = $3",
|
|
"< $6::integer",
|
|
"active_client_platform.target_client_id = $3",
|
|
"active_client_platform.platform_id = dt.platform_id",
|
|
"recent.status = 'succeeded'",
|
|
"recent.updated_at >= now() - interval '2 seconds'",
|
|
"recent.status IN ('failed', 'unknown', 'aborted')",
|
|
"recent.updated_at >= now() - interval '30 seconds'",
|
|
"ORDER BY dt.lane_weight DESC, dt.priority DESC, COALESCE(dt.enqueued_at, dt.created_at) ASC",
|
|
} {
|
|
if !strings.Contains(normalized, fragment) {
|
|
t.Fatalf("monitor lease query missing %q: %s", fragment, normalized)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestNormalizeDesktopTaskLeasePlatformIDs(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
got := normalizeDesktopTaskLeasePlatformIDs([]string{" Doubao ", "qwen", "", "DOUBAO"})
|
|
want := []string{"doubao", "qwen"}
|
|
if !reflect.DeepEqual(got, want) {
|
|
t.Fatalf("normalizeDesktopTaskLeasePlatformIDs() = %#v, want %#v", got, want)
|
|
}
|
|
if got := normalizeDesktopTaskLeasePlatformIDs([]string{" ", ""}); got != nil {
|
|
t.Fatalf("normalizeDesktopTaskLeasePlatformIDs(blank) = %#v, want nil", got)
|
|
}
|
|
}
|
|
|
|
func TestLeaseQueuedDesktopTaskByIDSQLUsesAccountIdentityAndClientSlots(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
normalized := normalizeSQLForDesktopTaskTest(leaseQueuedDesktopTaskByIDSQL())
|
|
for _, fragment := range []string{
|
|
"WITH current_accounts AS",
|
|
"client_id = $4",
|
|
"dt.kind <> 'monitor' OR ca.desktop_id IS NOT NULL",
|
|
"target_account.account_fingerprint",
|
|
"target_account.platform_uid",
|
|
"SELECT dt.desktop_id, dt.kind, ca.desktop_id AS lease_account_id",
|
|
"locked_candidate AS",
|
|
"pg_try_advisory_xact_lock(account_slot_lock_key)",
|
|
"FROM locked_candidate WHERE t.desktop_id = locked_candidate.desktop_id",
|
|
"target_account_id = CASE WHEN t.kind = 'monitor' THEN locked_candidate.lease_account_id ELSE t.target_account_id END",
|
|
"dt.kind <> 'monitor' OR NOT EXISTS",
|
|
"active.platform_id = dt.platform_id",
|
|
"active.status = 'in_progress'",
|
|
"active.desktop_id <> dt.desktop_id",
|
|
"COUNT(DISTINCT active_platform.platform_id)",
|
|
"active_platform.target_client_id = $4",
|
|
"< $8::integer",
|
|
"active_client_platform.target_client_id = $4",
|
|
"active_client_platform.platform_id = dt.platform_id",
|
|
"recent.status = 'succeeded'",
|
|
"recent.updated_at >= now() - interval '2 seconds'",
|
|
"recent.status IN ('failed', 'unknown', 'aborted')",
|
|
"recent.updated_at >= now() - interval '30 seconds'",
|
|
} {
|
|
if !strings.Contains(normalized, fragment) {
|
|
t.Fatalf("task-specific lease query missing %q: %s", fragment, normalized)
|
|
}
|
|
}
|
|
}
|
|
|
|
func recoverDesktopTaskSelectColumns(query string) []string {
|
|
re := regexp.MustCompile(`(?is)select\s+(.*?)\s+from\s+desktop_tasks`)
|
|
match := re.FindStringSubmatch(query)
|
|
if len(match) != 2 {
|
|
return nil
|
|
}
|
|
|
|
rawColumns := strings.Split(match[1], ",")
|
|
columns := make([]string, 0, len(rawColumns))
|
|
for _, rawColumn := range rawColumns {
|
|
column := strings.TrimSpace(rawColumn)
|
|
column = strings.Fields(column)[0]
|
|
columns = append(columns, column)
|
|
}
|
|
return columns
|
|
}
|
|
|
|
func normalizeSQLForDesktopTaskTest(sql string) string {
|
|
return strings.Join(strings.Fields(sql), " ")
|
|
}
|