feat(tenant): business-date-scope monitor lease lanes and authorization-failure cancellation
Add an optional lane filter to the monitor task lease so callers can pull a specific lane (e.g. high) and scope authorization-failure cancellation to the failing task's business date, so a challenge on one day no longer cancels queued tasks for another day. Monitor authorization-failure cancellation now requires a business date and no-ops without one. Also stop writing the nonexistent desktop_tasks.completed_at column when canceling queued tasks for an authorization failure. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -146,6 +146,11 @@ func loadQueuedDesktopAuthorizationFailureCandidates(
|
|||||||
tx pgx.Tx,
|
tx pgx.Tx,
|
||||||
task *repository.DesktopTask,
|
task *repository.DesktopTask,
|
||||||
) ([]queuedDesktopAuthorizationFailureCandidate, error) {
|
) ([]queuedDesktopAuthorizationFailureCandidate, error) {
|
||||||
|
businessDate := desktopAuthorizationTaskBusinessDate(task)
|
||||||
|
if task != nil && task.Kind == "monitor" && businessDate == "" {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
rows, err := tx.Query(ctx, `
|
rows, err := tx.Query(ctx, `
|
||||||
SELECT desktop_id, monitor_task_id
|
SELECT desktop_id, monitor_task_id
|
||||||
FROM desktop_tasks
|
FROM desktop_tasks
|
||||||
@@ -157,9 +162,19 @@ func loadQueuedDesktopAuthorizationFailureCandidates(
|
|||||||
AND (kind = 'monitor' OR target_account_id = $5)
|
AND (kind = 'monitor' OR target_account_id = $5)
|
||||||
AND platform_id = $6
|
AND platform_id = $6
|
||||||
AND desktop_id <> $7
|
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
|
ORDER BY COALESCE(enqueued_at, created_at) ASC, created_at ASC, desktop_id ASC
|
||||||
FOR UPDATE SKIP LOCKED
|
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 {
|
if err != nil {
|
||||||
return nil, response.ErrInternal(50121, "desktop_task_authorization_failure_query_failed", "failed to select queued desktop tasks blocked by authorization failure")
|
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
|
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(
|
func finishQueuedDesktopTaskForAuthorizationFailure(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
tx pgx.Tx,
|
tx pgx.Tx,
|
||||||
@@ -193,19 +221,7 @@ func finishQueuedDesktopTaskForAuthorizationFailure(
|
|||||||
status string,
|
status string,
|
||||||
errorJSON []byte,
|
errorJSON []byte,
|
||||||
) (*repository.DesktopTask, error) {
|
) (*repository.DesktopTask, error) {
|
||||||
row := tx.QueryRow(ctx, `
|
row := tx.QueryRow(ctx, finishQueuedDesktopTaskForAuthorizationFailureSQL(),
|
||||||
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,
|
|
||||||
desktopID,
|
desktopID,
|
||||||
status,
|
status,
|
||||||
errorJSON,
|
errorJSON,
|
||||||
@@ -220,6 +236,21 @@ func finishQueuedDesktopTaskForAuthorizationFailure(
|
|||||||
return updated, nil
|
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(
|
func desktopAuthorizationFailureDispositionFromPayload(
|
||||||
task *repository.DesktopTask,
|
task *repository.DesktopTask,
|
||||||
payload map[string]any,
|
payload map[string]any,
|
||||||
|
|||||||
@@ -107,6 +107,7 @@ type LeaseDesktopTaskRequest struct {
|
|||||||
TaskID *string `json:"task_id"`
|
TaskID *string `json:"task_id"`
|
||||||
Kind *string `json:"kind" binding:"omitempty,oneof=publish monitor"`
|
Kind *string `json:"kind" binding:"omitempty,oneof=publish monitor"`
|
||||||
PlatformIDs []string `json:"platform_ids"`
|
PlatformIDs []string `json:"platform_ids"`
|
||||||
|
Lanes []string `json:"lanes"`
|
||||||
}
|
}
|
||||||
|
|
||||||
type LeaseDesktopTaskResponse struct {
|
type LeaseDesktopTaskResponse struct {
|
||||||
@@ -216,6 +217,7 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt
|
|||||||
LeaseTokenHash: tokenHash,
|
LeaseTokenHash: tokenHash,
|
||||||
}
|
}
|
||||||
eligiblePlatformIDs := normalizeDesktopTaskLeasePlatformIDs(req.PlatformIDs)
|
eligiblePlatformIDs := normalizeDesktopTaskLeasePlatformIDs(req.PlatformIDs)
|
||||||
|
eligibleLanes := normalizeDesktopTaskLeaseLanes(req.Lanes)
|
||||||
|
|
||||||
if s.logger != nil {
|
if s.logger != nil {
|
||||||
fields := []zap.Field{
|
fields := []zap.Field{
|
||||||
@@ -235,7 +237,7 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt
|
|||||||
case taskID != nil:
|
case taskID != nil:
|
||||||
task, err = s.leaseQueuedDesktopTaskByID(ctx, client, *taskID, leaseParams)
|
task, err = s.leaseQueuedDesktopTaskByID(ctx, client, *taskID, leaseParams)
|
||||||
case req.Kind != nil && strings.TrimSpace(*req.Kind) == "monitor":
|
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":
|
case req.Kind != nil && strings.TrimSpace(*req.Kind) == "publish":
|
||||||
task, err = s.leaseNextQueuedPublishTask(ctx, client, leaseParams)
|
task, err = s.leaseNextQueuedPublishTask(ctx, client, leaseParams)
|
||||||
default:
|
default:
|
||||||
@@ -523,6 +525,7 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask(
|
|||||||
client *repository.DesktopClient,
|
client *repository.DesktopClient,
|
||||||
params repository.DesktopTaskLeaseParams,
|
params repository.DesktopTaskLeaseParams,
|
||||||
eligiblePlatformIDs []string,
|
eligiblePlatformIDs []string,
|
||||||
|
eligibleLanes []string,
|
||||||
) (*repository.DesktopTask, error) {
|
) (*repository.DesktopTask, error) {
|
||||||
tx, err := beginDesktopMonitorLeaseTx(ctx, s.pool, client.ID)
|
tx, err := beginDesktopMonitorLeaseTx(ctx, s.pool, client.ID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -538,6 +541,7 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask(
|
|||||||
params.LeaseTokenHash,
|
params.LeaseTokenHash,
|
||||||
maxConcurrentMonitorPlatformsPerDesktopClient,
|
maxConcurrentMonitorPlatformsPerDesktopClient,
|
||||||
eligiblePlatformIDs,
|
eligiblePlatformIDs,
|
||||||
|
eligibleLanes,
|
||||||
)
|
)
|
||||||
task, err := scanRepositoryDesktopTask(row)
|
task, err := scanRepositoryDesktopTask(row)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -582,10 +586,14 @@ func leaseNextQueuedMonitorTaskSQL() string {
|
|||||||
AND dt.platform_id = ca.platform_id
|
AND dt.platform_id = ca.platform_id
|
||||||
AND dt.kind = 'monitor'
|
AND dt.kind = 'monitor'
|
||||||
AND dt.status = 'queued'
|
AND dt.status = 'queued'
|
||||||
AND (
|
AND (
|
||||||
COALESCE(array_length($7::text[], 1), 0) = 0
|
COALESCE(array_length($7::text[], 1), 0) = 0
|
||||||
OR dt.platform_id = ANY($7::text[])
|
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 (
|
WHERE NOT EXISTS (
|
||||||
SELECT 1
|
SELECT 1
|
||||||
FROM desktop_tasks AS active
|
FROM desktop_tasks AS active
|
||||||
@@ -701,6 +709,32 @@ func normalizeDesktopTaskLeasePlatformIDs(values []string) []string {
|
|||||||
return result
|
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 {
|
func monitorAccountLeaseLockSQL(alias string) string {
|
||||||
return fmt.Sprintf(`hashtextextended(
|
return fmt.Sprintf(`hashtextextended(
|
||||||
concat_ws(
|
concat_ws(
|
||||||
|
|||||||
@@ -462,6 +462,7 @@ func TestLeaseNextQueuedMonitorTaskSQLUsesAccountIdentityAndClientSlots(t *testi
|
|||||||
"WITH current_accounts AS",
|
"WITH current_accounts AS",
|
||||||
"client_id = $3",
|
"client_id = $3",
|
||||||
"COALESCE(array_length($7::text[], 1), 0) = 0 OR dt.platform_id = ANY($7::text[])",
|
"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",
|
"dt.target_account_id = target_account.desktop_id",
|
||||||
"target_account.account_fingerprint",
|
"target_account.account_fingerprint",
|
||||||
"target_account.platform_uid",
|
"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) {
|
func TestLeaseQueuedDesktopTaskByIDSQLUsesAccountIdentityAndClientSlots(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
@@ -647,6 +647,7 @@ func TestMonitorAuthorizationFollowupCancellationIsPlatformScopedAcrossAccounts(
|
|||||||
TargetClientID: uuid.New(),
|
TargetClientID: uuid.New(),
|
||||||
TargetAccountID: uuid.New(),
|
TargetAccountID: uuid.New(),
|
||||||
Platform: "doubao",
|
Platform: "doubao",
|
||||||
|
Payload: []byte(`{"business_date":"2026-07-22"}`),
|
||||||
}
|
}
|
||||||
|
|
||||||
rows, err := loadQueuedDesktopAuthorizationFailureCandidates(context.Background(), tx, task)
|
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 target_client_id = $4")
|
||||||
assert.Contains(t, query, "AND (kind = 'monitor' OR target_account_id = $5)")
|
assert.Contains(t, query, "AND (kind = 'monitor' OR target_account_id = $5)")
|
||||||
assert.Contains(t, query, "AND platform_id = $6")
|
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, "aborted", desktopAuthorizationFollowupTaskStatus("monitor"))
|
||||||
assert.Equal(t, "failed", desktopAuthorizationFollowupTaskStatus("publish"))
|
assert.Equal(t, "failed", desktopAuthorizationFollowupTaskStatus("publish"))
|
||||||
assert.Equal(t, "task_canceled", desktopAuthorizationFollowupEventType(&repository.DesktopTask{
|
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) {
|
func TestFailRemainingMonitoringTasksAfterFinalAccountChallengeScopesToModelAndClient(t *testing.T) {
|
||||||
targetClientID := uuid.New().String()
|
targetClientID := uuid.New().String()
|
||||||
completedAt := time.Date(2026, 7, 14, 9, 30, 0, 0, time.UTC)
|
completedAt := time.Date(2026, 7, 14, 9, 30, 0, 0, time.UTC)
|
||||||
|
|||||||
Reference in New Issue
Block a user