diff --git a/server/internal/tenant/app/desktop_task_authorization_failure.go b/server/internal/tenant/app/desktop_task_authorization_failure.go index b922023..e40daeb 100644 --- a/server/internal/tenant/app/desktop_task_authorization_failure.go +++ b/server/internal/tenant/app/desktop_task_authorization_failure.go @@ -146,6 +146,11 @@ func loadQueuedDesktopAuthorizationFailureCandidates( tx pgx.Tx, task *repository.DesktopTask, ) ([]queuedDesktopAuthorizationFailureCandidate, error) { + businessDate := desktopAuthorizationTaskBusinessDate(task) + if task != nil && task.Kind == "monitor" && businessDate == "" { + return nil, nil + } + rows, err := tx.Query(ctx, ` SELECT desktop_id, monitor_task_id FROM desktop_tasks @@ -157,9 +162,19 @@ func loadQueuedDesktopAuthorizationFailureCandidates( AND (kind = 'monitor' OR target_account_id = $5) AND platform_id = $6 AND desktop_id <> $7 + AND ( + kind <> 'monitor' + OR COALESCE( + NULLIF(payload ->> 'business_date', ''), + NULLIF(payload ->> 'businessDate', ''), + NULLIF(payload ->> 'metric_date', ''), + NULLIF(payload ->> 'date', ''), + '' + ) = $8 + ) ORDER BY COALESCE(enqueued_at, created_at) ASC, created_at ASC, desktop_id ASC FOR UPDATE SKIP LOCKED - `, task.TenantID, task.WorkspaceID, task.Kind, task.TargetClientID, task.TargetAccountID, task.Platform, task.DesktopID) + `, task.TenantID, task.WorkspaceID, task.Kind, task.TargetClientID, task.TargetAccountID, task.Platform, task.DesktopID, businessDate) if err != nil { return nil, response.ErrInternal(50121, "desktop_task_authorization_failure_query_failed", "failed to select queued desktop tasks blocked by authorization failure") } @@ -186,6 +201,19 @@ func loadQueuedDesktopAuthorizationFailureCandidates( return candidates, nil } +func desktopAuthorizationTaskBusinessDate(task *repository.DesktopTask) string { + if task == nil || len(task.Payload) == 0 { + return "" + } + payload := unmarshalJSONObject(task.Payload) + for _, key := range []string{"business_date", "businessDate", "metric_date", "date"} { + if value := desktopAuthorizationFailureString(payload[key]); value != "" { + return value + } + } + return "" +} + func finishQueuedDesktopTaskForAuthorizationFailure( ctx context.Context, tx pgx.Tx, @@ -193,19 +221,7 @@ func finishQueuedDesktopTaskForAuthorizationFailure( status string, errorJSON []byte, ) (*repository.DesktopTask, error) { - row := tx.QueryRow(ctx, ` - UPDATE desktop_tasks AS t - SET status = $2, - result = NULL, - error = $3, - active_attempt_id = NULL, - lease_token_hash = NULL, - lease_expires_at = NULL, - completed_at = NOW(), - updated_at = NOW() - WHERE desktop_id = $1 - AND status = 'queued' - RETURNING `+desktopTaskRepositoryReturningColumns, + row := tx.QueryRow(ctx, finishQueuedDesktopTaskForAuthorizationFailureSQL(), desktopID, status, errorJSON, @@ -220,6 +236,21 @@ func finishQueuedDesktopTaskForAuthorizationFailure( return updated, nil } +func finishQueuedDesktopTaskForAuthorizationFailureSQL() string { + return ` + UPDATE desktop_tasks AS t + SET status = $2, + result = NULL, + error = $3, + active_attempt_id = NULL, + lease_token_hash = NULL, + lease_expires_at = NULL, + updated_at = NOW() + WHERE desktop_id = $1 + AND status = 'queued' + RETURNING ` + desktopTaskRepositoryReturningColumns +} + func desktopAuthorizationFailureDispositionFromPayload( task *repository.DesktopTask, payload map[string]any, diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index bd8f103..c09a2f6 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -107,6 +107,7 @@ type LeaseDesktopTaskRequest struct { TaskID *string `json:"task_id"` Kind *string `json:"kind" binding:"omitempty,oneof=publish monitor"` PlatformIDs []string `json:"platform_ids"` + Lanes []string `json:"lanes"` } type LeaseDesktopTaskResponse struct { @@ -216,6 +217,7 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt LeaseTokenHash: tokenHash, } eligiblePlatformIDs := normalizeDesktopTaskLeasePlatformIDs(req.PlatformIDs) + eligibleLanes := normalizeDesktopTaskLeaseLanes(req.Lanes) if s.logger != nil { fields := []zap.Field{ @@ -235,7 +237,7 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt case taskID != nil: task, err = s.leaseQueuedDesktopTaskByID(ctx, client, *taskID, leaseParams) case req.Kind != nil && strings.TrimSpace(*req.Kind) == "monitor": - task, err = s.leaseNextQueuedMonitorTask(ctx, client, leaseParams, eligiblePlatformIDs) + task, err = s.leaseNextQueuedMonitorTask(ctx, client, leaseParams, eligiblePlatformIDs, eligibleLanes) case req.Kind != nil && strings.TrimSpace(*req.Kind) == "publish": task, err = s.leaseNextQueuedPublishTask(ctx, client, leaseParams) default: @@ -523,6 +525,7 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask( client *repository.DesktopClient, params repository.DesktopTaskLeaseParams, eligiblePlatformIDs []string, + eligibleLanes []string, ) (*repository.DesktopTask, error) { tx, err := beginDesktopMonitorLeaseTx(ctx, s.pool, client.ID) if err != nil { @@ -538,6 +541,7 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask( params.LeaseTokenHash, maxConcurrentMonitorPlatformsPerDesktopClient, eligiblePlatformIDs, + eligibleLanes, ) task, err := scanRepositoryDesktopTask(row) if err != nil { @@ -582,10 +586,14 @@ func leaseNextQueuedMonitorTaskSQL() string { AND dt.platform_id = ca.platform_id AND dt.kind = 'monitor' AND dt.status = 'queued' - AND ( - COALESCE(array_length($7::text[], 1), 0) = 0 - OR dt.platform_id = ANY($7::text[]) - ) + AND ( + COALESCE(array_length($7::text[], 1), 0) = 0 + OR dt.platform_id = ANY($7::text[]) + ) + AND ( + COALESCE(array_length($8::text[], 1), 0) = 0 + OR dt.lane = ANY($8::text[]) + ) WHERE NOT EXISTS ( SELECT 1 FROM desktop_tasks AS active @@ -701,6 +709,32 @@ func normalizeDesktopTaskLeasePlatformIDs(values []string) []string { return result } +func normalizeDesktopTaskLeaseLanes(values []string) []string { + allowed := map[string]struct{}{ + "high": {}, + "normal": {}, + "normal_boosted": {}, + "retry": {}, + } + result := make([]string, 0, len(values)) + seen := make(map[string]struct{}, len(values)) + for _, value := range values { + normalized := strings.ToLower(strings.TrimSpace(value)) + if _, ok := allowed[normalized]; !ok { + continue + } + if _, exists := seen[normalized]; exists { + continue + } + seen[normalized] = struct{}{} + result = append(result, normalized) + } + if len(result) == 0 { + return nil + } + return result +} + func monitorAccountLeaseLockSQL(alias string) string { return fmt.Sprintf(`hashtextextended( concat_ws( diff --git a/server/internal/tenant/app/desktop_task_service_test.go b/server/internal/tenant/app/desktop_task_service_test.go index 33a69e3..1efaf7c 100644 --- a/server/internal/tenant/app/desktop_task_service_test.go +++ b/server/internal/tenant/app/desktop_task_service_test.go @@ -462,6 +462,7 @@ func TestLeaseNextQueuedMonitorTaskSQLUsesAccountIdentityAndClientSlots(t *testi "WITH current_accounts AS", "client_id = $3", "COALESCE(array_length($7::text[], 1), 0) = 0 OR dt.platform_id = ANY($7::text[])", + "COALESCE(array_length($8::text[], 1), 0) = 0 OR dt.lane = ANY($8::text[])", "dt.target_account_id = target_account.desktop_id", "target_account.account_fingerprint", "target_account.platform_uid", @@ -502,6 +503,39 @@ func TestNormalizeDesktopTaskLeasePlatformIDs(t *testing.T) { } } +func TestNormalizeDesktopTaskLeaseLanes(t *testing.T) { + t.Parallel() + + got := normalizeDesktopTaskLeaseLanes([]string{" HIGH ", "normal", "", "high", "unsupported"}) + want := []string{"high", "normal"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("normalizeDesktopTaskLeaseLanes() = %#v, want %#v", got, want) + } + if got := normalizeDesktopTaskLeaseLanes([]string{" ", "unsupported"}); got != nil { + t.Fatalf("normalizeDesktopTaskLeaseLanes(blank) = %#v, want nil", got) + } +} + +func TestFinishQueuedDesktopTaskForAuthorizationFailureSQLMatchesDesktopTaskSchema(t *testing.T) { + t.Parallel() + + normalized := normalizeSQLForDesktopTaskTest(finishQueuedDesktopTaskForAuthorizationFailureSQL()) + if strings.Contains(normalized, "completed_at") { + t.Fatalf("authorization cancellation query writes nonexistent desktop_tasks.completed_at: %s", normalized) + } + for _, fragment := range []string{ + "SET status = $2", + "active_attempt_id = NULL", + "lease_token_hash = NULL", + "lease_expires_at = NULL", + "updated_at = NOW()", + } { + if !strings.Contains(normalized, fragment) { + t.Fatalf("authorization cancellation query missing %q: %s", fragment, normalized) + } + } +} + func TestLeaseQueuedDesktopTaskByIDSQLUsesAccountIdentityAndClientSlots(t *testing.T) { t.Parallel() diff --git a/server/internal/tenant/app/monitoring_callback_service_test.go b/server/internal/tenant/app/monitoring_callback_service_test.go index 20eea5c..607d944 100644 --- a/server/internal/tenant/app/monitoring_callback_service_test.go +++ b/server/internal/tenant/app/monitoring_callback_service_test.go @@ -647,6 +647,7 @@ func TestMonitorAuthorizationFollowupCancellationIsPlatformScopedAcrossAccounts( TargetClientID: uuid.New(), TargetAccountID: uuid.New(), Platform: "doubao", + Payload: []byte(`{"business_date":"2026-07-22"}`), } rows, err := loadQueuedDesktopAuthorizationFailureCandidates(context.Background(), tx, task) @@ -659,6 +660,10 @@ func TestMonitorAuthorizationFollowupCancellationIsPlatformScopedAcrossAccounts( assert.Contains(t, query, "AND target_client_id = $4") assert.Contains(t, query, "AND (kind = 'monitor' OR target_account_id = $5)") assert.Contains(t, query, "AND platform_id = $6") + assert.Contains(t, query, "COALESCE( NULLIF(payload ->> 'business_date', '')") + assert.Contains(t, query, "= $8") + require.Len(t, tx.queryArgs, 8) + assert.Equal(t, "2026-07-22", tx.queryArgs[7]) assert.Equal(t, "aborted", desktopAuthorizationFollowupTaskStatus("monitor")) assert.Equal(t, "failed", desktopAuthorizationFollowupTaskStatus("publish")) assert.Equal(t, "task_canceled", desktopAuthorizationFollowupEventType(&repository.DesktopTask{ @@ -667,6 +672,24 @@ func TestMonitorAuthorizationFollowupCancellationIsPlatformScopedAcrossAccounts( })) } +func TestMonitorAuthorizationFollowupCancellationRequiresBusinessDate(t *testing.T) { + tx := &captureDailyMonitorTaskTx{} + task := &repository.DesktopTask{ + Kind: "monitor", + TenantID: 77, + WorkspaceID: 88, + TargetClientID: uuid.New(), + TargetAccountID: uuid.New(), + Platform: "doubao", + } + + rows, err := loadQueuedDesktopAuthorizationFailureCandidates(context.Background(), tx, task) + + require.NoError(t, err) + assert.Empty(t, rows) + assert.False(t, tx.queryCalled) +} + func TestFailRemainingMonitoringTasksAfterFinalAccountChallengeScopesToModelAndClient(t *testing.T) { targetClientID := uuid.New().String() completedAt := time.Date(2026, 7, 14, 9, 30, 0, 0, time.UTC)