root
ed48674ab5
Frontend CI / Frontend (push) Successful in 3m18s
Backend CI / Backend (push) Successful in 16m16s
Desktop Client Build / Publish Client Artifacts to NAS (push) Successful in 26s
Desktop Client Build / Resolve Build Metadata (push) Successful in 26s
Desktop Client Build / Build Desktop Client (push) Successful in 25m45s
3142 lines
105 KiB
Go
3142 lines
105 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgtype"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
goredis "github.com/redis/go-redis/v9"
|
|
"go.uber.org/zap"
|
|
|
|
"github.com/geo-platform/tenant-api/internal/shared/auth"
|
|
sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache"
|
|
sharedcompliance "github.com/geo-platform/tenant-api/internal/shared/compliance"
|
|
"github.com/geo-platform/tenant-api/internal/shared/config"
|
|
"github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq"
|
|
"github.com/geo-platform/tenant-api/internal/shared/response"
|
|
tenantcompliance "github.com/geo-platform/tenant-api/internal/tenant/compliance"
|
|
"github.com/geo-platform/tenant-api/internal/tenant/repository"
|
|
)
|
|
|
|
const (
|
|
desktopPublishMaxAttempts = 3
|
|
maxConcurrentMonitorPlatformsPerDesktopClient = 2
|
|
monitorLeaseCandidateScanLimitSQL = "16"
|
|
monitorSucceededTaskCooldownSQL = "2 seconds"
|
|
monitorUnhealthyTaskCooldownSQL = "30 seconds"
|
|
)
|
|
|
|
var errDesktopTaskLeaseDeferred = errors.New("desktop task lease deferred")
|
|
|
|
type DesktopTaskService struct {
|
|
pool *pgxpool.Pool
|
|
monitoringPool *pgxpool.Pool
|
|
repo repository.DesktopTaskRepository
|
|
cache sharedcache.Cache
|
|
redis *goredis.Client
|
|
messaging *rabbitmq.Client
|
|
logger *zap.Logger
|
|
compliance *tenantcompliance.Service
|
|
}
|
|
|
|
func NewDesktopTaskService(pool *pgxpool.Pool, monitoringPool *pgxpool.Pool, messaging *rabbitmq.Client, logger *zap.Logger) *DesktopTaskService {
|
|
return &DesktopTaskService{
|
|
pool: pool,
|
|
monitoringPool: monitoringPool,
|
|
repo: repository.NewDesktopTaskRepository(pool),
|
|
messaging: messaging,
|
|
logger: logger,
|
|
compliance: tenantcompliance.NewService(pool, config.NewStaticProvider(&config.Config{
|
|
Compliance: config.ComplianceConfig{Enabled: true},
|
|
}), logger).WithRabbitMQ(messaging),
|
|
}
|
|
}
|
|
|
|
func NewDesktopTaskServiceWithConfig(pool *pgxpool.Pool, monitoringPool *pgxpool.Pool, messaging *rabbitmq.Client, logger *zap.Logger, cfg config.Provider) *DesktopTaskService {
|
|
svc := NewDesktopTaskService(pool, monitoringPool, messaging, logger)
|
|
svc.compliance = tenantcompliance.NewService(pool, cfg, logger).WithRabbitMQ(messaging)
|
|
return svc
|
|
}
|
|
|
|
func (s *DesktopTaskService) WithCache(c sharedcache.Cache) *DesktopTaskService {
|
|
s.cache = c
|
|
return s
|
|
}
|
|
|
|
func (s *DesktopTaskService) WithRedis(redis *goredis.Client) *DesktopTaskService {
|
|
if s != nil {
|
|
s.redis = redis
|
|
}
|
|
return s
|
|
}
|
|
|
|
type DesktopTaskView struct {
|
|
ID string `json:"id"`
|
|
JobID string `json:"job_id"`
|
|
TenantID int64 `json:"tenant_id"`
|
|
WorkspaceID int64 `json:"workspace_id"`
|
|
TargetAccountID string `json:"target_account_id"`
|
|
TargetClientID string `json:"target_client_id"`
|
|
Platform string `json:"platform"`
|
|
Kind string `json:"kind"`
|
|
Payload json.RawMessage `json:"payload"`
|
|
Status string `json:"status"`
|
|
PublishJobStatus *string `json:"publish_job_status,omitempty"`
|
|
ComplianceBlockedRecordID *int64 `json:"compliance_blocked_record_id,omitempty"`
|
|
ComplianceBlockedAt *time.Time `json:"compliance_blocked_at,omitempty"`
|
|
ComplianceBlockedReason *string `json:"compliance_blocked_reason,omitempty"`
|
|
DedupKey *string `json:"dedup_key"`
|
|
ActiveAttemptID *string `json:"active_attempt_id"`
|
|
LeaseExpiresAt *time.Time `json:"lease_expires_at"`
|
|
PublishSubmitStartedAt *time.Time `json:"publish_submit_started_at,omitempty"`
|
|
Attempts int `json:"attempts"`
|
|
Result json.RawMessage `json:"result"`
|
|
Error json.RawMessage `json:"error"`
|
|
CreatedAt time.Time `json:"created_at"`
|
|
UpdatedAt time.Time `json:"updated_at"`
|
|
}
|
|
|
|
type LeaseDesktopTaskRequest struct {
|
|
TaskID *string `json:"task_id"`
|
|
Kind *string `json:"kind" binding:"omitempty,oneof=publish monitor"`
|
|
PlatformIDs []string `json:"platform_ids"`
|
|
}
|
|
|
|
type LeaseDesktopTaskResponse struct {
|
|
Task *DesktopTaskView `json:"task"`
|
|
AttemptID *string `json:"attempt_id,omitempty"`
|
|
LeaseToken *string `json:"lease_token,omitempty"`
|
|
LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"`
|
|
}
|
|
|
|
type ExtendDesktopTaskRequest struct {
|
|
LeaseToken string `json:"lease_token" binding:"required"`
|
|
}
|
|
|
|
type MarkPublishSubmitStartedRequest struct {
|
|
LeaseToken string `json:"lease_token" binding:"required"`
|
|
}
|
|
|
|
type CompleteDesktopTaskRequest struct {
|
|
LeaseToken string `json:"lease_token" binding:"required"`
|
|
Status string `json:"status" binding:"required,oneof=succeeded failed unknown"`
|
|
Payload map[string]any `json:"payload"`
|
|
Error map[string]any `json:"error"`
|
|
}
|
|
|
|
type CancelDesktopTaskRequest struct {
|
|
LeaseToken *string `json:"lease_token"`
|
|
Reason *string `json:"reason"`
|
|
}
|
|
|
|
type ReconcileDesktopTaskRequest struct {
|
|
Status string `json:"status" binding:"required,oneof=succeeded failed aborted retry"`
|
|
Result map[string]any `json:"result"`
|
|
Error map[string]any `json:"error"`
|
|
}
|
|
|
|
type ListPublishTasksRequest struct {
|
|
Page int `form:"page"`
|
|
PageSize int `form:"page_size"`
|
|
Title string `form:"title"`
|
|
}
|
|
|
|
type DesktopPublishTaskList struct {
|
|
Items []DesktopTaskView `json:"items"`
|
|
Page int `json:"page"`
|
|
PageSize int `json:"page_size"`
|
|
Total int `json:"total"`
|
|
PendingCount int `json:"pending_count"`
|
|
HistoryTotal int `json:"history_total"`
|
|
}
|
|
|
|
type publishTaskListOwner struct {
|
|
TenantID int64
|
|
WorkspaceID int64
|
|
UserID int64
|
|
}
|
|
|
|
const publishTaskVisibleRecordFilterSQL = `
|
|
AND (
|
|
NOT (t.payload ? 'publish_record_id')
|
|
OR EXISTS (
|
|
SELECT 1
|
|
FROM publish_records pr
|
|
WHERE pr.tenant_id = t.tenant_id
|
|
AND pr.id = CASE
|
|
WHEN (t.payload->>'publish_record_id') ~ '^[0-9]+$'
|
|
THEN (t.payload->>'publish_record_id')::bigint
|
|
ELSE NULL
|
|
END
|
|
AND pr.deleted_at IS NULL
|
|
)
|
|
)
|
|
`
|
|
|
|
func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.DesktopClient, req LeaseDesktopTaskRequest, routeTaskID *uuid.UUID) (*LeaseDesktopTaskResponse, error) {
|
|
if client == nil {
|
|
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
|
|
}
|
|
markDesktopTaskConsumerPresent(ctx, s.redis, client.ID)
|
|
|
|
taskID := routeTaskID
|
|
if taskID == nil && req.TaskID != nil && strings.TrimSpace(*req.TaskID) != "" {
|
|
parsed, err := uuid.Parse(strings.TrimSpace(*req.TaskID))
|
|
if err != nil {
|
|
return nil, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid")
|
|
}
|
|
taskID = &parsed
|
|
}
|
|
|
|
if err := s.dropStaleMonitorTasksForClient(ctx, client); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.recoverExpiredClientTasks(ctx, client); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := s.recheckQueuedPublishTasksForClient(ctx, client); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
rawToken, tokenHash, err := newDesktopClientToken()
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50087, "desktop_task_lease_token_failed", "failed to generate desktop task lease token")
|
|
}
|
|
|
|
attemptID := uuid.New()
|
|
leaseParams := repository.DesktopTaskLeaseParams{
|
|
AttemptID: attemptID,
|
|
LeaseTokenHash: tokenHash,
|
|
}
|
|
eligiblePlatformIDs := normalizeDesktopTaskLeasePlatformIDs(req.PlatformIDs)
|
|
|
|
if s.logger != nil {
|
|
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.logger.Debug("desktop task lease requested", fields...)
|
|
}
|
|
|
|
var task *repository.DesktopTask
|
|
switch {
|
|
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)
|
|
case req.Kind != nil && strings.TrimSpace(*req.Kind) == "publish":
|
|
task, err = s.leaseNextQueuedPublishTask(ctx, client, leaseParams)
|
|
default:
|
|
task, err = s.leaseNextQueuedPublishTask(ctx, client, leaseParams)
|
|
}
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
if s.logger != nil {
|
|
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.logger.Debug("desktop task lease found no matching queued task", fields...)
|
|
}
|
|
if taskID == nil {
|
|
return &LeaseDesktopTaskResponse{}, nil
|
|
}
|
|
classifiedErr := s.classifyLeaseError(ctx, client, taskID)
|
|
if errors.Is(classifiedErr, errDesktopTaskLeaseDeferred) {
|
|
return &LeaseDesktopTaskResponse{}, nil
|
|
}
|
|
return nil, classifiedErr
|
|
}
|
|
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")
|
|
}
|
|
|
|
if _, attemptErr := s.repo.CreateAttempt(ctx, repository.CreateDesktopTaskAttemptParams{
|
|
DesktopID: attemptID,
|
|
TaskID: task.DesktopID,
|
|
ClientID: client.ID,
|
|
LeaseTokenHash: tokenHash,
|
|
}); attemptErr != nil {
|
|
s.logWarn("desktop task attempt insert failed", attemptErr, zap.String("task_id", task.DesktopID.String()))
|
|
}
|
|
|
|
s.publishTaskEvent(ctx, task, "task_leased")
|
|
|
|
if s.logger != nil {
|
|
s.logger.Info("desktop task leased",
|
|
zap.String("client_id", client.ID.String()),
|
|
zap.String("task_id", task.DesktopID.String()),
|
|
zap.String("kind", task.Kind),
|
|
zap.String("status", task.Status),
|
|
)
|
|
}
|
|
|
|
view := buildDesktopTaskView(task)
|
|
attemptIDText := attemptID.String()
|
|
return &LeaseDesktopTaskResponse{
|
|
Task: &view,
|
|
AttemptID: &attemptIDText,
|
|
LeaseToken: &rawToken,
|
|
LeaseExpiresAt: task.LeaseExpiresAt,
|
|
}, nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) recheckQueuedPublishTasksForClient(ctx context.Context, client *repository.DesktopClient) error {
|
|
if s == nil || s.pool == nil || s.compliance == nil || client == nil {
|
|
return nil
|
|
}
|
|
rows, err := s.pool.Query(ctx, `
|
|
SELECT
|
|
j.desktop_id,
|
|
j.tenant_id,
|
|
j.workspace_id,
|
|
j.created_by_user_id,
|
|
j.article_id,
|
|
j.article_version_id,
|
|
COALESCE(array_agg(DISTINCT dt.platform_id) FILTER (WHERE dt.platform_id <> ''), '{}') AS platforms,
|
|
BOOL_OR(a.deleted_at IS NOT NULL) AS article_deleted
|
|
FROM desktop_tasks dt
|
|
JOIN desktop_publish_jobs j ON j.desktop_id = dt.job_id
|
|
LEFT JOIN articles a ON a.id = j.article_id AND a.tenant_id = j.tenant_id
|
|
WHERE dt.target_client_id = $1
|
|
AND dt.kind = 'publish'
|
|
AND dt.status = 'queued'
|
|
AND j.status = 'queued'
|
|
AND j.article_id IS NOT NULL
|
|
AND j.article_version_id IS NOT NULL
|
|
GROUP BY j.desktop_id, j.tenant_id, j.workspace_id, j.created_by_user_id, j.article_id, j.article_version_id
|
|
ORDER BY MIN(dt.created_at)
|
|
LIMIT 5
|
|
`, client.ID)
|
|
if err != nil {
|
|
s.logWarn("desktop publish compliance recheck query failed", err, zap.String("client_id", client.ID.String()))
|
|
return response.ErrInternal(50118, "desktop_publish_compliance_recheck_failed", "failed to load publish jobs for compliance recheck")
|
|
}
|
|
defer rows.Close()
|
|
|
|
type candidate struct {
|
|
JobID uuid.UUID
|
|
TenantID int64
|
|
WorkspaceID int64
|
|
CreatedByUserID int64
|
|
ArticleID int64
|
|
ArticleVersionID int64
|
|
Platforms []string
|
|
ArticleDeleted bool
|
|
}
|
|
candidates := make([]candidate, 0)
|
|
for rows.Next() {
|
|
var item candidate
|
|
if scanErr := rows.Scan(&item.JobID, &item.TenantID, &item.WorkspaceID, &item.CreatedByUserID, &item.ArticleID, &item.ArticleVersionID, &item.Platforms, &item.ArticleDeleted); scanErr != nil {
|
|
s.logWarn("desktop publish compliance recheck scan failed", scanErr, zap.String("client_id", client.ID.String()))
|
|
return response.ErrInternal(50118, "desktop_publish_compliance_recheck_failed", "failed to scan publish job for compliance recheck")
|
|
}
|
|
if item.WorkspaceID == client.WorkspaceID && item.TenantID == client.TenantID {
|
|
candidates = append(candidates, item)
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
s.logWarn("desktop publish compliance recheck rows failed", err, zap.String("client_id", client.ID.String()))
|
|
return response.ErrInternal(50118, "desktop_publish_compliance_recheck_failed", "failed to iterate publish jobs for compliance recheck")
|
|
}
|
|
|
|
for _, item := range candidates {
|
|
if item.ArticleDeleted {
|
|
if cancelErr := s.cancelUnavailableArticleQueuedPublishTasks(ctx, item.TenantID, item.ArticleID); cancelErr != nil {
|
|
return cancelErr
|
|
}
|
|
continue
|
|
}
|
|
gate, gateErr := s.compliance.GateForPublish(ctx, tenantcompliance.GateForPublishRequest{
|
|
TenantID: item.TenantID,
|
|
ArticleID: item.ArticleID,
|
|
ArticleVersionID: item.ArticleVersionID,
|
|
ActorID: item.CreatedByUserID,
|
|
TargetPlatforms: item.Platforms,
|
|
TriggerSource: "scheduler_recheck",
|
|
})
|
|
if gateErr != nil && gate != nil && gate.Decision == sharedcompliance.GateDecisionBlock {
|
|
if markErr := s.compliance.MarkPublishJobBlockedByCompliance(ctx, item.JobID, gate.Result); markErr != nil {
|
|
return markErr
|
|
}
|
|
continue
|
|
}
|
|
if gateErr != nil {
|
|
if isComplianceInvalidArticleVersionError(gateErr) {
|
|
if cancelErr := s.cancelUnavailableArticleQueuedPublishTasks(ctx, item.TenantID, item.ArticleID); cancelErr != nil {
|
|
return cancelErr
|
|
}
|
|
continue
|
|
}
|
|
return response.ErrServiceUnavailable(41004, "compliance_engine_unavailable", "scheduled publish compliance recheck failed")
|
|
}
|
|
if gate != nil && gate.Decision == sharedcompliance.GateDecisionBlock {
|
|
if markErr := s.compliance.MarkPublishJobBlockedByCompliance(ctx, item.JobID, gate.Result); markErr != nil {
|
|
return markErr
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) cancelUnavailableArticleQueuedPublishTasks(ctx context.Context, tenantID, articleID int64) error {
|
|
if s == nil || s.pool == nil || tenantID <= 0 || articleID <= 0 {
|
|
return nil
|
|
}
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return response.ErrInternal(50122, "publish_task_cancel_unavailable_article_begin_failed", "failed to start unavailable article publish task cleanup")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
taskRefs, err := queuedPublishTaskRefsForArticle(ctx, tx, tenantID, articleID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := cancelQueuedPublishTasksForUnavailableArticle(ctx, tx, tenantID, articleID); err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return response.ErrInternal(50123, "publish_task_cancel_unavailable_article_commit_failed", "failed to commit unavailable article publish task cleanup")
|
|
}
|
|
|
|
for _, ref := range taskRefs {
|
|
task, getErr := s.repo.GetByDesktopID(ctx, ref.TaskID, ref.WorkspaceID)
|
|
if getErr == nil && task != nil {
|
|
s.publishTaskEvent(ctx, task, "task_canceled")
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type desktopTaskRef struct {
|
|
TaskID uuid.UUID
|
|
WorkspaceID int64
|
|
}
|
|
|
|
func queuedPublishTaskRefsForArticle(ctx context.Context, tx pgx.Tx, tenantID, articleID int64) ([]desktopTaskRef, error) {
|
|
if tx == nil || tenantID <= 0 || articleID <= 0 {
|
|
return nil, nil
|
|
}
|
|
rows, err := tx.Query(ctx, `
|
|
SELECT dt.desktop_id, dt.workspace_id
|
|
FROM desktop_tasks dt
|
|
JOIN desktop_publish_jobs j
|
|
ON j.desktop_id = dt.job_id
|
|
AND j.tenant_id = dt.tenant_id
|
|
WHERE dt.tenant_id = $1
|
|
AND dt.kind = 'publish'
|
|
AND dt.status = 'queued'
|
|
AND j.article_id = $2
|
|
AND j.status = 'queued'
|
|
ORDER BY dt.created_at ASC, dt.desktop_id ASC
|
|
`, tenantID, articleID)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50124, "publish_task_cancel_unavailable_article_lookup_failed", "failed to inspect unavailable article publish tasks")
|
|
}
|
|
defer rows.Close()
|
|
|
|
taskRefs := make([]desktopTaskRef, 0)
|
|
for rows.Next() {
|
|
var ref desktopTaskRef
|
|
if scanErr := rows.Scan(&ref.TaskID, &ref.WorkspaceID); scanErr != nil {
|
|
return nil, response.ErrInternal(50124, "publish_task_cancel_unavailable_article_lookup_failed", "failed to scan unavailable article publish tasks")
|
|
}
|
|
taskRefs = append(taskRefs, ref)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, response.ErrInternal(50124, "publish_task_cancel_unavailable_article_lookup_failed", "failed to iterate unavailable article publish tasks")
|
|
}
|
|
return taskRefs, nil
|
|
}
|
|
|
|
func isComplianceInvalidArticleVersionError(err error) bool {
|
|
appErr, ok := err.(*response.AppError)
|
|
return ok && appErr.Code == 41005 && appErr.Message == "invalid_article_version"
|
|
}
|
|
|
|
const desktopTaskRepositoryReturningColumns = `
|
|
t.desktop_id,
|
|
t.job_id,
|
|
t.tenant_id,
|
|
t.workspace_id,
|
|
t.target_account_id,
|
|
t.target_client_id,
|
|
t.platform_id,
|
|
t.kind,
|
|
t.payload,
|
|
t.status,
|
|
t.priority,
|
|
t.lane,
|
|
t.lane_weight,
|
|
t.source,
|
|
t.scheduler_group_key,
|
|
t.monitor_task_id,
|
|
t.supersedes_task_id,
|
|
t.control_flags,
|
|
t.interrupt_generation,
|
|
t.dedup_key,
|
|
t.active_attempt_id,
|
|
t.lease_expires_at,
|
|
t.attempts,
|
|
t.result,
|
|
t.error,
|
|
t.started_at,
|
|
t.interrupted_at,
|
|
t.interrupt_reason,
|
|
t.enqueued_at,
|
|
t.created_at,
|
|
t.updated_at`
|
|
|
|
type desktopTaskRepositoryScanner interface {
|
|
Scan(dest ...any) error
|
|
}
|
|
|
|
func (s *DesktopTaskService) leaseNextQueuedMonitorTask(
|
|
ctx context.Context,
|
|
client *repository.DesktopClient,
|
|
params repository.DesktopTaskLeaseParams,
|
|
eligiblePlatformIDs []string,
|
|
) (*repository.DesktopTask, error) {
|
|
tx, err := beginDesktopMonitorLeaseTx(ctx, s.pool, client.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
row := tx.QueryRow(ctx, leaseNextQueuedMonitorTaskSQL(),
|
|
client.TenantID,
|
|
client.WorkspaceID,
|
|
client.ID,
|
|
params.AttemptID,
|
|
params.LeaseTokenHash,
|
|
maxConcurrentMonitorPlatformsPerDesktopClient,
|
|
eligiblePlatformIDs,
|
|
)
|
|
task, err := scanRepositoryDesktopTask(row)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to commit desktop monitor task lease")
|
|
}
|
|
return task, nil
|
|
}
|
|
|
|
func leaseNextQueuedMonitorTaskSQL() string {
|
|
return `
|
|
WITH current_accounts AS (
|
|
SELECT desktop_id, user_id, platform_id, platform_uid, account_fingerprint
|
|
FROM platform_accounts
|
|
WHERE tenant_id = $1
|
|
AND workspace_id = $2
|
|
AND client_id = $3
|
|
AND deleted_at IS NULL
|
|
AND delete_requested_at IS NULL
|
|
),
|
|
candidate_pool AS (
|
|
SELECT dt.desktop_id,
|
|
ca.desktop_id AS lease_account_id,
|
|
dt.lane_weight,
|
|
dt.priority,
|
|
dt.enqueued_at,
|
|
dt.created_at,
|
|
` + monitorAccountLeaseLockSQL("target_account") + ` AS account_slot_lock_key
|
|
FROM current_accounts AS ca
|
|
JOIN platform_accounts AS target_account
|
|
ON target_account.tenant_id = $1
|
|
AND target_account.workspace_id = $2
|
|
AND target_account.user_id = ca.user_id
|
|
AND target_account.platform_id = ca.platform_id
|
|
AND ` + monitorAccountIdentityMatchSQL("ca", "target_account") + `
|
|
JOIN desktop_tasks AS dt
|
|
ON dt.target_account_id = target_account.desktop_id
|
|
AND dt.tenant_id = $1
|
|
AND dt.workspace_id = $2
|
|
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[])
|
|
)
|
|
WHERE NOT EXISTS (
|
|
SELECT 1
|
|
FROM desktop_tasks AS active
|
|
JOIN platform_accounts AS active_account
|
|
ON active_account.desktop_id = active.target_account_id
|
|
AND active_account.tenant_id = active.tenant_id
|
|
AND active_account.workspace_id = active.workspace_id
|
|
WHERE active.tenant_id = dt.tenant_id
|
|
AND active.workspace_id = dt.workspace_id
|
|
AND active.kind = 'monitor'
|
|
AND active.platform_id = dt.platform_id
|
|
AND active.status = 'in_progress'
|
|
AND active.desktop_id <> dt.desktop_id
|
|
AND active_account.user_id = target_account.user_id
|
|
AND ` + monitorAccountIdentityMatchSQL("active_account", "target_account") + `
|
|
)
|
|
AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM desktop_tasks AS recent
|
|
JOIN platform_accounts AS recent_account
|
|
ON recent_account.desktop_id = recent.target_account_id
|
|
AND recent_account.tenant_id = recent.tenant_id
|
|
AND recent_account.workspace_id = recent.workspace_id
|
|
WHERE recent.tenant_id = dt.tenant_id
|
|
AND recent.workspace_id = dt.workspace_id
|
|
AND recent.kind = 'monitor'
|
|
AND recent.platform_id = dt.platform_id
|
|
AND (
|
|
(recent.status = 'succeeded' AND recent.updated_at >= now() - interval '` + monitorSucceededTaskCooldownSQL + `')
|
|
OR (recent.status IN ('failed', 'unknown', 'aborted') AND recent.updated_at >= now() - interval '` + monitorUnhealthyTaskCooldownSQL + `')
|
|
)
|
|
AND recent.desktop_id <> dt.desktop_id
|
|
AND recent_account.user_id = target_account.user_id
|
|
AND ` + monitorAccountIdentityMatchSQL("recent_account", "target_account") + `
|
|
)
|
|
AND (
|
|
SELECT COUNT(DISTINCT active_platform.platform_id)
|
|
FROM desktop_tasks AS active_platform
|
|
WHERE active_platform.tenant_id = dt.tenant_id
|
|
AND active_platform.workspace_id = dt.workspace_id
|
|
AND active_platform.kind = 'monitor'
|
|
AND active_platform.target_client_id = $3
|
|
AND active_platform.status = 'in_progress'
|
|
AND active_platform.desktop_id <> dt.desktop_id
|
|
) < $6::integer
|
|
AND NOT EXISTS (
|
|
SELECT 1
|
|
FROM desktop_tasks AS active_client_platform
|
|
WHERE active_client_platform.tenant_id = dt.tenant_id
|
|
AND active_client_platform.workspace_id = dt.workspace_id
|
|
AND active_client_platform.kind = 'monitor'
|
|
AND active_client_platform.target_client_id = $3
|
|
AND active_client_platform.platform_id = dt.platform_id
|
|
AND active_client_platform.status = 'in_progress'
|
|
AND active_client_platform.desktop_id <> dt.desktop_id
|
|
)
|
|
ORDER BY dt.lane_weight DESC,
|
|
dt.priority DESC,
|
|
COALESCE(dt.enqueued_at, dt.created_at) ASC,
|
|
dt.created_at ASC,
|
|
dt.desktop_id ASC,
|
|
ca.desktop_id ASC
|
|
LIMIT ` + monitorLeaseCandidateScanLimitSQL + `
|
|
FOR UPDATE OF dt SKIP LOCKED
|
|
),
|
|
candidate AS (
|
|
SELECT desktop_id, lease_account_id
|
|
FROM candidate_pool
|
|
WHERE pg_try_advisory_xact_lock(account_slot_lock_key)
|
|
ORDER BY lane_weight DESC,
|
|
priority DESC,
|
|
COALESCE(enqueued_at, created_at) ASC,
|
|
created_at ASC,
|
|
desktop_id ASC,
|
|
lease_account_id ASC
|
|
LIMIT 1
|
|
)
|
|
UPDATE desktop_tasks AS t
|
|
SET target_client_id = $3,
|
|
target_account_id = candidate.lease_account_id,
|
|
active_attempt_id = $4,
|
|
lease_token_hash = $5,
|
|
lease_expires_at = now() + interval '10 minutes',
|
|
status = 'in_progress',
|
|
attempts = t.attempts + 1,
|
|
updated_at = now()
|
|
FROM candidate
|
|
WHERE t.desktop_id = candidate.desktop_id
|
|
RETURNING ` + desktopTaskRepositoryReturningColumns
|
|
}
|
|
|
|
func normalizeDesktopTaskLeasePlatformIDs(values []string) []string {
|
|
if len(values) == 0 {
|
|
return nil
|
|
}
|
|
|
|
result := make([]string, 0, len(values))
|
|
seen := make(map[string]struct{}, len(values))
|
|
for _, value := range values {
|
|
platformID := strings.ToLower(strings.TrimSpace(value))
|
|
if platformID == "" {
|
|
continue
|
|
}
|
|
if _, ok := seen[platformID]; ok {
|
|
continue
|
|
}
|
|
seen[platformID] = struct{}{}
|
|
result = append(result, platformID)
|
|
}
|
|
if len(result) == 0 {
|
|
return nil
|
|
}
|
|
return result
|
|
}
|
|
|
|
func monitorAccountLeaseLockSQL(alias string) string {
|
|
return fmt.Sprintf(`hashtextextended(
|
|
concat_ws(
|
|
':',
|
|
'monitor_desktop_account_slot',
|
|
%s.tenant_id::text,
|
|
%s.workspace_id::text,
|
|
%s.user_id::text,
|
|
%s.platform_id,
|
|
CASE
|
|
WHEN btrim(COALESCE(%s.account_fingerprint, '')) <> '' THEN 'fp:' || %s.account_fingerprint
|
|
WHEN btrim(COALESCE(%s.platform_uid, '')) <> '' THEN 'uid:' || %s.platform_uid
|
|
ELSE 'id:' || %s.desktop_id::text
|
|
END
|
|
),
|
|
0
|
|
)`,
|
|
alias,
|
|
alias,
|
|
alias,
|
|
alias,
|
|
alias, alias,
|
|
alias, alias,
|
|
alias,
|
|
)
|
|
}
|
|
|
|
func monitorAccountIdentityMatchSQL(leftAlias, rightAlias string) string {
|
|
return fmt.Sprintf(`(
|
|
%s.desktop_id = %s.desktop_id
|
|
OR (
|
|
btrim(COALESCE(%s.account_fingerprint, '')) <> ''
|
|
AND %s.account_fingerprint = %s.account_fingerprint
|
|
)
|
|
OR (
|
|
btrim(COALESCE(%s.platform_uid, '')) <> ''
|
|
AND %s.platform_uid = %s.platform_uid
|
|
)
|
|
)`,
|
|
leftAlias, rightAlias,
|
|
leftAlias, leftAlias, rightAlias,
|
|
leftAlias, leftAlias, rightAlias,
|
|
)
|
|
}
|
|
|
|
func (s *DesktopTaskService) leaseNextQueuedPublishTask(
|
|
ctx context.Context,
|
|
client *repository.DesktopClient,
|
|
params repository.DesktopTaskLeaseParams,
|
|
) (*repository.DesktopTask, error) {
|
|
row := s.pool.QueryRow(ctx, `
|
|
WITH candidate AS (
|
|
SELECT dt.desktop_id
|
|
FROM desktop_tasks AS dt
|
|
WHERE dt.target_client_id = $1
|
|
AND dt.kind = 'publish'
|
|
AND dt.status = 'queued'
|
|
AND dt.attempts < $4
|
|
AND EXISTS (
|
|
SELECT 1
|
|
FROM platform_accounts AS pa
|
|
WHERE pa.desktop_id = dt.target_account_id
|
|
AND pa.workspace_id = dt.workspace_id
|
|
AND pa.client_id = dt.target_client_id
|
|
AND pa.deleted_at IS NULL
|
|
AND pa.delete_requested_at IS NULL
|
|
)
|
|
AND (
|
|
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,
|
|
COALESCE(dt.enqueued_at, dt.created_at) ASC,
|
|
dt.created_at ASC,
|
|
dt.desktop_id ASC
|
|
LIMIT 1
|
|
FOR UPDATE OF dt SKIP LOCKED
|
|
)
|
|
UPDATE desktop_tasks AS t
|
|
SET active_attempt_id = $2,
|
|
lease_token_hash = $3,
|
|
lease_expires_at = now() + interval '3 minutes',
|
|
status = 'in_progress',
|
|
attempts = t.attempts + 1,
|
|
started_at = COALESCE(t.started_at, now()),
|
|
updated_at = now()
|
|
FROM candidate
|
|
WHERE t.desktop_id = candidate.desktop_id
|
|
RETURNING `+desktopTaskRepositoryReturningColumns,
|
|
client.ID,
|
|
params.AttemptID,
|
|
params.LeaseTokenHash,
|
|
desktopPublishMaxAttempts,
|
|
)
|
|
return scanRepositoryDesktopTask(row)
|
|
}
|
|
|
|
func (s *DesktopTaskService) leaseQueuedDesktopTaskByID(
|
|
ctx context.Context,
|
|
client *repository.DesktopClient,
|
|
desktopID uuid.UUID,
|
|
params repository.DesktopTaskLeaseParams,
|
|
) (*repository.DesktopTask, error) {
|
|
tx, err := beginDesktopMonitorLeaseTx(ctx, s.pool, client.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
row := tx.QueryRow(ctx, leaseQueuedDesktopTaskByIDSQL(),
|
|
desktopID,
|
|
client.WorkspaceID,
|
|
client.TenantID,
|
|
client.ID,
|
|
params.AttemptID,
|
|
params.LeaseTokenHash,
|
|
desktopPublishMaxAttempts,
|
|
maxConcurrentMonitorPlatformsPerDesktopClient,
|
|
)
|
|
task, err := scanRepositoryDesktopTask(row)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to commit desktop task lease")
|
|
}
|
|
return task, nil
|
|
}
|
|
|
|
func leaseQueuedDesktopTaskByIDSQL() string {
|
|
return `
|
|
WITH current_accounts AS (
|
|
SELECT desktop_id, user_id, platform_id, platform_uid, account_fingerprint
|
|
FROM platform_accounts
|
|
WHERE tenant_id = $3
|
|
AND workspace_id = $2
|
|
AND client_id = $4
|
|
AND deleted_at IS NULL
|
|
AND delete_requested_at IS NULL
|
|
),
|
|
candidate AS (
|
|
SELECT dt.desktop_id,
|
|
dt.kind,
|
|
ca.desktop_id AS lease_account_id,
|
|
` + monitorAccountLeaseLockSQL("target_account") + ` AS account_slot_lock_key
|
|
FROM desktop_tasks AS dt
|
|
LEFT JOIN platform_accounts AS target_account
|
|
ON dt.kind = 'monitor'
|
|
AND target_account.desktop_id = dt.target_account_id
|
|
AND target_account.tenant_id = dt.tenant_id
|
|
AND target_account.workspace_id = dt.workspace_id
|
|
LEFT JOIN current_accounts AS ca
|
|
ON dt.kind = 'monitor'
|
|
AND ca.user_id = target_account.user_id
|
|
AND ca.platform_id = dt.platform_id
|
|
AND ` + monitorAccountIdentityMatchSQL("ca", "target_account") + `
|
|
WHERE dt.desktop_id = $1
|
|
AND dt.workspace_id = $2
|
|
AND dt.status = 'queued'
|
|
AND (dt.kind <> 'publish' OR dt.attempts < $7)
|
|
AND (
|
|
dt.kind <> 'publish'
|
|
OR EXISTS (
|
|
SELECT 1
|
|
FROM platform_accounts AS pa
|
|
WHERE pa.desktop_id = dt.target_account_id
|
|
AND pa.workspace_id = dt.workspace_id
|
|
AND pa.client_id = dt.target_client_id
|
|
AND pa.deleted_at IS NULL
|
|
AND pa.delete_requested_at IS NULL
|
|
)
|
|
)
|
|
AND (
|
|
dt.kind <> 'monitor'
|
|
OR ca.desktop_id IS NOT NULL
|
|
)
|
|
AND (
|
|
dt.kind <> 'monitor'
|
|
OR NOT EXISTS (
|
|
SELECT 1
|
|
FROM desktop_tasks AS active
|
|
JOIN platform_accounts AS active_account
|
|
ON active_account.desktop_id = active.target_account_id
|
|
AND active_account.tenant_id = active.tenant_id
|
|
AND active_account.workspace_id = active.workspace_id
|
|
WHERE active.tenant_id = dt.tenant_id
|
|
AND active.workspace_id = dt.workspace_id
|
|
AND active.kind = 'monitor'
|
|
AND active.platform_id = dt.platform_id
|
|
AND active.status = 'in_progress'
|
|
AND active.desktop_id <> dt.desktop_id
|
|
AND active_account.user_id = target_account.user_id
|
|
AND ` + monitorAccountIdentityMatchSQL("active_account", "target_account") + `
|
|
)
|
|
)
|
|
AND (
|
|
dt.kind <> 'monitor'
|
|
OR NOT EXISTS (
|
|
SELECT 1
|
|
FROM desktop_tasks AS recent
|
|
JOIN platform_accounts AS recent_account
|
|
ON recent_account.desktop_id = recent.target_account_id
|
|
AND recent_account.tenant_id = recent.tenant_id
|
|
AND recent_account.workspace_id = recent.workspace_id
|
|
WHERE recent.tenant_id = dt.tenant_id
|
|
AND recent.workspace_id = dt.workspace_id
|
|
AND recent.kind = 'monitor'
|
|
AND recent.platform_id = dt.platform_id
|
|
AND (
|
|
(recent.status = 'succeeded' AND recent.updated_at >= now() - interval '` + monitorSucceededTaskCooldownSQL + `')
|
|
OR (recent.status IN ('failed', 'unknown', 'aborted') AND recent.updated_at >= now() - interval '` + monitorUnhealthyTaskCooldownSQL + `')
|
|
)
|
|
AND recent.desktop_id <> dt.desktop_id
|
|
AND recent_account.user_id = target_account.user_id
|
|
AND ` + monitorAccountIdentityMatchSQL("recent_account", "target_account") + `
|
|
)
|
|
)
|
|
AND (
|
|
dt.kind <> 'monitor'
|
|
OR (
|
|
SELECT COUNT(DISTINCT active_platform.platform_id)
|
|
FROM desktop_tasks AS active_platform
|
|
WHERE active_platform.tenant_id = dt.tenant_id
|
|
AND active_platform.workspace_id = dt.workspace_id
|
|
AND active_platform.kind = 'monitor'
|
|
AND active_platform.target_client_id = $4
|
|
AND active_platform.status = 'in_progress'
|
|
AND active_platform.desktop_id <> dt.desktop_id
|
|
) < $8::integer
|
|
)
|
|
AND (
|
|
dt.kind <> 'monitor'
|
|
OR NOT EXISTS (
|
|
SELECT 1
|
|
FROM desktop_tasks AS active_client_platform
|
|
WHERE active_client_platform.tenant_id = dt.tenant_id
|
|
AND active_client_platform.workspace_id = dt.workspace_id
|
|
AND active_client_platform.kind = 'monitor'
|
|
AND active_client_platform.target_client_id = $4
|
|
AND active_client_platform.platform_id = dt.platform_id
|
|
AND active_client_platform.status = 'in_progress'
|
|
AND active_client_platform.desktop_id <> dt.desktop_id
|
|
)
|
|
)
|
|
AND (
|
|
(
|
|
dt.kind = 'monitor'
|
|
AND dt.tenant_id = $3
|
|
)
|
|
OR (
|
|
dt.kind <> 'monitor'
|
|
AND dt.target_client_id = $4
|
|
)
|
|
)
|
|
ORDER BY ca.desktop_id ASC NULLS LAST
|
|
LIMIT 1
|
|
FOR UPDATE OF dt SKIP LOCKED
|
|
),
|
|
locked_candidate AS (
|
|
SELECT desktop_id, lease_account_id
|
|
FROM candidate
|
|
WHERE kind <> 'monitor'
|
|
OR pg_try_advisory_xact_lock(account_slot_lock_key)
|
|
LIMIT 1
|
|
)
|
|
UPDATE desktop_tasks AS t
|
|
SET target_client_id = CASE WHEN t.kind = 'monitor' THEN $4 ELSE t.target_client_id END,
|
|
target_account_id = CASE WHEN t.kind = 'monitor' THEN locked_candidate.lease_account_id ELSE t.target_account_id END,
|
|
active_attempt_id = $5,
|
|
lease_token_hash = $6,
|
|
lease_expires_at = now() + CASE WHEN t.kind = 'publish' THEN interval '3 minutes' ELSE interval '10 minutes' END,
|
|
status = 'in_progress',
|
|
attempts = t.attempts + 1,
|
|
started_at = COALESCE(t.started_at, now()),
|
|
updated_at = now()
|
|
FROM locked_candidate
|
|
WHERE t.desktop_id = locked_candidate.desktop_id
|
|
RETURNING ` + desktopTaskRepositoryReturningColumns
|
|
}
|
|
|
|
func beginDesktopMonitorLeaseTx(ctx context.Context, pool *pgxpool.Pool, clientID uuid.UUID) (pgx.Tx, error) {
|
|
if pool == nil {
|
|
return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "desktop task pool is not available")
|
|
}
|
|
tx, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to start desktop monitor task lease")
|
|
}
|
|
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock($1)`, monitorDesktopTaskClientSlotLockKey(clientID)); err != nil {
|
|
_ = tx.Rollback(ctx)
|
|
return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to lock desktop monitor task lease")
|
|
}
|
|
return tx, nil
|
|
}
|
|
|
|
func scanRepositoryDesktopTask(row desktopTaskRepositoryScanner) (*repository.DesktopTask, error) {
|
|
var (
|
|
task repository.DesktopTask
|
|
schedulerGroupKey pgtype.Text
|
|
monitorTaskID pgtype.Int8
|
|
supersedesTaskID pgtype.UUID
|
|
dedupKey pgtype.Text
|
|
activeAttemptID pgtype.UUID
|
|
leaseExpiresAt pgtype.Timestamptz
|
|
startedAt pgtype.Timestamptz
|
|
interruptedAt pgtype.Timestamptz
|
|
interruptReason pgtype.Text
|
|
enqueuedAt pgtype.Timestamptz
|
|
)
|
|
if err := row.Scan(
|
|
&task.DesktopID,
|
|
&task.JobID,
|
|
&task.TenantID,
|
|
&task.WorkspaceID,
|
|
&task.TargetAccountID,
|
|
&task.TargetClientID,
|
|
&task.Platform,
|
|
&task.Kind,
|
|
&task.Payload,
|
|
&task.Status,
|
|
&task.Priority,
|
|
&task.Lane,
|
|
&task.LaneWeight,
|
|
&task.Source,
|
|
&schedulerGroupKey,
|
|
&monitorTaskID,
|
|
&supersedesTaskID,
|
|
&task.ControlFlags,
|
|
&task.InterruptGeneration,
|
|
&dedupKey,
|
|
&activeAttemptID,
|
|
&leaseExpiresAt,
|
|
&task.Attempts,
|
|
&task.Result,
|
|
&task.Error,
|
|
&startedAt,
|
|
&interruptedAt,
|
|
&interruptReason,
|
|
&enqueuedAt,
|
|
&task.CreatedAt,
|
|
&task.UpdatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
task.SchedulerGroupKey = desktopTaskNullableText(schedulerGroupKey)
|
|
task.MonitorTaskID = desktopTaskNullableInt64(monitorTaskID)
|
|
task.SupersedesTaskID = desktopTaskNullableUUID(supersedesTaskID)
|
|
task.DedupKey = desktopTaskNullableText(dedupKey)
|
|
task.ActiveAttemptID = desktopTaskNullableUUID(activeAttemptID)
|
|
task.LeaseExpiresAt = desktopTaskNullableTime(leaseExpiresAt)
|
|
task.StartedAt = desktopTaskNullableTime(startedAt)
|
|
task.InterruptedAt = desktopTaskNullableTime(interruptedAt)
|
|
task.InterruptReason = desktopTaskNullableText(interruptReason)
|
|
if enqueuedAt.Valid {
|
|
task.EnqueuedAt = enqueuedAt.Time
|
|
}
|
|
return &task, nil
|
|
}
|
|
|
|
func desktopTaskNullableText(value pgtype.Text) *string {
|
|
if !value.Valid {
|
|
return nil
|
|
}
|
|
text := value.String
|
|
return &text
|
|
}
|
|
|
|
func desktopTaskNullableInt64(value pgtype.Int8) *int64 {
|
|
if !value.Valid {
|
|
return nil
|
|
}
|
|
number := value.Int64
|
|
return &number
|
|
}
|
|
|
|
func desktopTaskNullableUUID(value pgtype.UUID) *uuid.UUID {
|
|
if !value.Valid {
|
|
return nil
|
|
}
|
|
resolved, err := uuid.FromBytes(value.Bytes[:])
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
return &resolved
|
|
}
|
|
|
|
func desktopTaskNullableTime(value pgtype.Timestamptz) *time.Time {
|
|
if !value.Valid {
|
|
return nil
|
|
}
|
|
timestamp := value.Time
|
|
return ×tamp
|
|
}
|
|
|
|
func (s *DesktopTaskService) Extend(ctx context.Context, client *repository.DesktopClient, desktopID uuid.UUID, req ExtendDesktopTaskRequest) (*DesktopTaskView, error) {
|
|
if client == nil {
|
|
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
|
|
}
|
|
|
|
if err := s.dropStaleMonitorTasksForClient(ctx, client); err != nil {
|
|
return nil, err
|
|
}
|
|
if s.redis != nil {
|
|
presence := loadDesktopClientPresence(ctx, s.redis, []uuid.UUID{client.ID})
|
|
if presence != nil && !presence[client.ID] {
|
|
return nil, response.ErrConflict(40986, "desktop_client_offline", "desktop client presence has expired")
|
|
}
|
|
}
|
|
|
|
task, err := s.repo.ExtendLease(ctx, desktopID, HashDesktopClientToken(strings.TrimSpace(req.LeaseToken)))
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, response.ErrConflict(40983, "desktop_task_lease_invalid", "desktop task lease token is invalid or expired")
|
|
}
|
|
return nil, response.ErrInternal(50089, "desktop_task_extend_failed", "failed to extend desktop task lease")
|
|
}
|
|
|
|
s.publishTaskEvent(ctx, task, "task_extended")
|
|
|
|
view := buildDesktopTaskView(task)
|
|
return &view, nil
|
|
}
|
|
|
|
// MarkPublishSubmitStarted records a durable intent marker right before the client performs the
|
|
// irreversible platform submit POST. Lease-recovery uses it to avoid silently re-posting a
|
|
// non-idempotent article: once submit may have started, recovery routes the task to 'unknown'
|
|
// (manual reconcile) instead of auto-requeueing it. The client must receive a
|
|
// positive marker write before it performs the platform submit; a stale/lost
|
|
// lease is rejected so the client can stop before the irreversible request.
|
|
func (s *DesktopTaskService) MarkPublishSubmitStarted(ctx context.Context, client *repository.DesktopClient, taskID uuid.UUID, leaseToken string) error {
|
|
if s == nil || s.pool == nil || client == nil {
|
|
return nil
|
|
}
|
|
leaseToken = strings.TrimSpace(leaseToken)
|
|
if leaseToken == "" {
|
|
return response.ErrBadRequest(40001, "invalid_params", "lease_token is required")
|
|
}
|
|
tag, err := s.pool.Exec(ctx, `
|
|
UPDATE desktop_tasks
|
|
SET publish_submit_started_at = COALESCE(publish_submit_started_at, NOW()),
|
|
updated_at = NOW()
|
|
WHERE desktop_id = $1
|
|
AND target_client_id = $2
|
|
AND kind = 'publish'
|
|
AND status = 'in_progress'
|
|
AND lease_token_hash = $3
|
|
`, taskID, client.ID, HashDesktopClientToken(leaseToken))
|
|
if err != nil {
|
|
return response.ErrInternal(50200, "desktop_task_submit_marker_failed", "failed to mark publish submit started")
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return response.ErrConflict(40983, "desktop_task_lease_invalid", "desktop task lease token is invalid or expired")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) Complete(ctx context.Context, client *repository.DesktopClient, desktopID uuid.UUID, req CompleteDesktopTaskRequest) (*DesktopTaskView, error) {
|
|
if client == nil {
|
|
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
|
|
}
|
|
|
|
finalStatus := normalizeDesktopTaskCompletionStatus(req.Status)
|
|
resultJSON, err := marshalOptionalJSON(req.Payload)
|
|
if err != nil {
|
|
return nil, response.ErrBadRequest(40086, "invalid_desktop_task_payload", "payload must be serializable")
|
|
}
|
|
errorJSON, err := marshalOptionalJSON(req.Error)
|
|
if err != nil {
|
|
return nil, response.ErrBadRequest(40087, "invalid_desktop_task_error", "error must be serializable")
|
|
}
|
|
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50103, "desktop_task_complete_begin_failed", "failed to start desktop task completion")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
taskRepo := repository.NewDesktopTaskRepository(tx)
|
|
previousTask, _ := taskRepo.GetByDesktopID(ctx, desktopID, client.WorkspaceID)
|
|
previousAttemptID := activeAttemptID(previousTask)
|
|
if previousTask != nil {
|
|
finalStatus = normalizeDesktopTaskCompletionStatusForKind(previousTask.Kind, finalStatus)
|
|
}
|
|
|
|
task, err := taskRepo.Complete(ctx, desktopID, HashDesktopClientToken(strings.TrimSpace(req.LeaseToken)), finalStatus, resultJSON, errorJSON)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, response.ErrConflict(40983, "desktop_task_lease_invalid", "desktop task lease token is invalid or expired")
|
|
}
|
|
return nil, response.ErrInternal(50091, "desktop_task_complete_failed", "failed to complete desktop task")
|
|
}
|
|
|
|
if previousAttemptID != nil {
|
|
if finishErr := taskRepo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
|
|
TaskID: task.DesktopID,
|
|
AttemptID: *previousAttemptID,
|
|
FinalStatus: &finalStatus,
|
|
Error: errorJSON,
|
|
}); finishErr != nil {
|
|
s.logWarn("desktop task attempt finish failed after complete", finishErr, zap.String("task_id", task.DesktopID.String()))
|
|
}
|
|
}
|
|
|
|
publishOutcome, err := syncDesktopPublishTaskState(ctx, tx, task)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
bulkFailure, err := bulkFailQueuedDesktopTasksForAuthorizationFailure(ctx, tx, task, errorJSON)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return nil, response.ErrInternal(50104, "desktop_task_complete_commit_failed", "failed to commit desktop task completion")
|
|
}
|
|
|
|
if publishOutcome != nil {
|
|
invalidateArticleCaches(ctx, s.cache, publishOutcome.TenantID, &publishOutcome.ArticleID)
|
|
if syncErr := s.syncMonitoringArticleAlias(ctx, publishOutcome.ArticleAlias); syncErr != nil {
|
|
s.logWarn("monitoring article alias sync failed after publish complete", syncErr, zap.Int64("article_id", publishOutcome.ArticleID))
|
|
}
|
|
}
|
|
for _, outcome := range bulkFailure.PublishOutcomes {
|
|
invalidateArticleCaches(ctx, s.cache, outcome.TenantID, &outcome.ArticleID)
|
|
if outcome.ArticleAlias != nil {
|
|
if syncErr := s.syncMonitoringArticleAlias(ctx, outcome.ArticleAlias); syncErr != nil {
|
|
s.logWarn("monitoring article alias sync failed after bulk authorization failure", syncErr, zap.Int64("article_id", outcome.ArticleID))
|
|
}
|
|
}
|
|
}
|
|
|
|
s.publishTaskEvent(ctx, task, "task_completed")
|
|
for _, failedTask := range bulkFailure.Tasks {
|
|
s.publishTaskEvent(ctx, failedTask, "task_completed")
|
|
}
|
|
|
|
view := buildDesktopTaskView(task)
|
|
return &view, nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) Cancel(ctx context.Context, client *repository.DesktopClient, desktopID uuid.UUID, req CancelDesktopTaskRequest) (*DesktopTaskView, error) {
|
|
if client == nil {
|
|
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
|
|
}
|
|
|
|
errorJSON, err := marshalOptionalJSON(map[string]any{
|
|
"reason": optionalStringValue(req.Reason),
|
|
"source": "desktop_client",
|
|
})
|
|
if err != nil {
|
|
return nil, response.ErrBadRequest(40088, "invalid_desktop_task_cancel", "cancel payload must be serializable")
|
|
}
|
|
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50107, "desktop_task_cancel_begin_failed", "failed to start desktop task cancel")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
taskRepo := repository.NewDesktopTaskRepository(tx)
|
|
previousTask, _ := taskRepo.GetByDesktopID(ctx, desktopID, client.WorkspaceID)
|
|
previousAttemptID := activeAttemptID(previousTask)
|
|
|
|
var task *repository.DesktopTask
|
|
if req.LeaseToken != nil && strings.TrimSpace(*req.LeaseToken) != "" {
|
|
task, err = taskRepo.CancelByLease(ctx, desktopID, HashDesktopClientToken(strings.TrimSpace(*req.LeaseToken)), errorJSON)
|
|
} else {
|
|
task, err = taskRepo.CancelQueuedPublishByOwner(ctx, desktopID, client.TenantID, client.WorkspaceID, client.UserID, errorJSON)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
task, err = taskRepo.CancelByClient(ctx, desktopID, client.ID, errorJSON)
|
|
}
|
|
}
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, response.ErrConflict(40984, "desktop_task_cancel_conflict", "desktop task cannot be canceled in its current state")
|
|
}
|
|
return nil, response.ErrInternal(50092, "desktop_task_cancel_failed", "failed to cancel desktop task")
|
|
}
|
|
|
|
if previousAttemptID != nil && req.LeaseToken != nil && strings.TrimSpace(*req.LeaseToken) != "" {
|
|
if finishErr := taskRepo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
|
|
TaskID: task.DesktopID,
|
|
AttemptID: *previousAttemptID,
|
|
FinalStatus: literalStringPtr("aborted"),
|
|
Error: errorJSON,
|
|
}); finishErr != nil {
|
|
s.logWarn("desktop task attempt finish failed after cancel", finishErr, zap.String("task_id", task.DesktopID.String()))
|
|
}
|
|
}
|
|
|
|
var replacementTask *repository.DesktopTask
|
|
if shouldRequeuePreemptedMonitorTask(task, req.Reason) {
|
|
replacementTask, err = requeuePreemptedMonitorTask(ctx, tx, taskRepo, task)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return nil, response.ErrInternal(50108, "desktop_task_cancel_commit_failed", "failed to commit desktop task cancel")
|
|
}
|
|
|
|
s.publishTaskEvent(ctx, task, "task_canceled")
|
|
if replacementTask != nil {
|
|
s.publishTaskEvent(ctx, replacementTask, "task_available")
|
|
}
|
|
|
|
view := buildDesktopTaskView(task)
|
|
return &view, nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) TenantCancel(ctx context.Context, actor auth.Actor, desktopID uuid.UUID, req CancelDesktopTaskRequest) (*DesktopTaskView, error) {
|
|
workspaceID := auth.CurrentWorkspaceID(ctx)
|
|
if actor.TenantID == 0 || actor.UserID == 0 || workspaceID == 0 {
|
|
return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required")
|
|
}
|
|
|
|
errorJSON, err := marshalOptionalJSON(map[string]any{
|
|
"reason": optionalStringValue(req.Reason),
|
|
"source": "tenant_web",
|
|
})
|
|
if err != nil {
|
|
return nil, response.ErrBadRequest(40088, "invalid_desktop_task_cancel", "cancel payload must be serializable")
|
|
}
|
|
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50107, "desktop_task_cancel_begin_failed", "failed to start desktop task cancel")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
taskRepo := repository.NewDesktopTaskRepository(tx)
|
|
task, err := taskRepo.TenantCancel(ctx, desktopID, actor.TenantID, workspaceID, actor.UserID, errorJSON)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, response.ErrConflict(40984, "desktop_task_cancel_conflict", "desktop task cannot be canceled in its current state")
|
|
}
|
|
return nil, response.ErrInternal(50093, "desktop_task_tenant_cancel_failed", "failed to cancel desktop task")
|
|
}
|
|
|
|
publishOutcome, err := syncDesktopPublishTaskState(ctx, tx, task)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return nil, response.ErrInternal(50108, "desktop_task_cancel_commit_failed", "failed to commit desktop task cancel")
|
|
}
|
|
|
|
s.publishTaskEvent(ctx, task, "task_canceled")
|
|
s.afterRecoveredPublishOutcomes(ctx, []*desktopPublishSyncOutcome{publishOutcome})
|
|
|
|
view := buildDesktopTaskView(task)
|
|
return &view, nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) Reconcile(ctx context.Context, actor auth.Actor, desktopID uuid.UUID, req ReconcileDesktopTaskRequest) (*DesktopTaskView, error) {
|
|
if actor.PrimaryWorkspaceID == 0 {
|
|
return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required")
|
|
}
|
|
|
|
resultJSON, err := marshalOptionalJSON(req.Result)
|
|
if err != nil {
|
|
return nil, response.ErrBadRequest(40089, "invalid_desktop_task_reconcile_result", "result must be serializable")
|
|
}
|
|
errorJSON, err := marshalOptionalJSON(req.Error)
|
|
if err != nil {
|
|
return nil, response.ErrBadRequest(40090, "invalid_desktop_task_reconcile_error", "error must be serializable")
|
|
}
|
|
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50105, "desktop_task_reconcile_begin_failed", "failed to start desktop task reconcile")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
taskRepo := repository.NewDesktopTaskRepository(tx)
|
|
task, err := taskRepo.Reconcile(ctx, desktopID, actor.PrimaryWorkspaceID, req.Status, resultJSON, errorJSON)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, response.ErrConflict(40985, "desktop_task_reconcile_conflict", "desktop task is not waiting for reconcile")
|
|
}
|
|
return nil, response.ErrInternal(50094, "desktop_task_reconcile_failed", "failed to reconcile desktop task")
|
|
}
|
|
|
|
publishOutcome, err := syncDesktopPublishTaskState(ctx, tx, task)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return nil, response.ErrInternal(50106, "desktop_task_reconcile_commit_failed", "failed to commit desktop task reconcile")
|
|
}
|
|
|
|
if publishOutcome != nil {
|
|
invalidateArticleCaches(ctx, s.cache, publishOutcome.TenantID, &publishOutcome.ArticleID)
|
|
if syncErr := s.syncMonitoringArticleAlias(ctx, publishOutcome.ArticleAlias); syncErr != nil {
|
|
s.logWarn("monitoring article alias sync failed after publish reconcile", syncErr, zap.Int64("article_id", publishOutcome.ArticleID))
|
|
}
|
|
}
|
|
|
|
s.publishTaskEvent(ctx, task, reconcileDesktopTaskEventType(task.Kind, req.Status))
|
|
|
|
view := buildDesktopTaskView(task)
|
|
return &view, nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) ListPublishTasks(ctx context.Context, client *repository.DesktopClient, req ListPublishTasksRequest) (*DesktopPublishTaskList, error) {
|
|
if client == nil {
|
|
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
|
|
}
|
|
return s.listPublishTasksForOwner(ctx, publishTaskListOwner{
|
|
TenantID: client.TenantID,
|
|
WorkspaceID: client.WorkspaceID,
|
|
UserID: client.UserID,
|
|
}, req)
|
|
}
|
|
|
|
func (s *DesktopTaskService) ListTenantPublishTasks(ctx context.Context, actor auth.Actor, req ListPublishTasksRequest) (*DesktopPublishTaskList, error) {
|
|
workspaceID := auth.CurrentWorkspaceID(ctx)
|
|
if actor.TenantID == 0 || actor.UserID == 0 || workspaceID == 0 {
|
|
return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required")
|
|
}
|
|
return s.listPublishTasksForOwner(ctx, publishTaskListOwner{
|
|
TenantID: actor.TenantID,
|
|
WorkspaceID: workspaceID,
|
|
UserID: actor.UserID,
|
|
}, req)
|
|
}
|
|
|
|
func (s *DesktopTaskService) listPublishTasksForOwner(ctx context.Context, owner publishTaskListOwner, req ListPublishTasksRequest) (*DesktopPublishTaskList, error) {
|
|
if s.pool == nil {
|
|
return nil, response.ErrServiceUnavailable(50342, "desktop_publish_job_store_unavailable", "desktop publish job store is unavailable")
|
|
}
|
|
|
|
page := req.Page
|
|
if page <= 0 {
|
|
page = 1
|
|
}
|
|
pageSize := req.PageSize
|
|
if pageSize <= 0 || pageSize > 50 {
|
|
pageSize = 10
|
|
}
|
|
|
|
title := strings.TrimSpace(req.Title)
|
|
pendingStatuses := []string{"queued", "in_progress"}
|
|
historyStatuses := []string{"succeeded", "failed", "unknown", "aborted"}
|
|
|
|
pendingItems, err := s.listPublishTasksByStatuses(
|
|
ctx,
|
|
owner,
|
|
pendingStatuses,
|
|
title,
|
|
0,
|
|
0,
|
|
`CASE WHEN t.status = 'in_progress' THEN 0 ELSE 1 END ASC, t.created_at ASC, t.desktop_id DESC`,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
historyTotal, err := s.countPublishTasksByStatuses(ctx, owner, historyStatuses, title)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
offset := (page - 1) * pageSize
|
|
items := make([]DesktopTaskView, 0, pageSize)
|
|
total := len(pendingItems) + historyTotal
|
|
|
|
if offset < len(pendingItems) {
|
|
end := offset + pageSize
|
|
if end > len(pendingItems) {
|
|
end = len(pendingItems)
|
|
}
|
|
items = append(items, pendingItems[offset:end]...)
|
|
}
|
|
|
|
remaining := pageSize - len(items)
|
|
historyOffset := offset - len(pendingItems)
|
|
if historyOffset < 0 {
|
|
historyOffset = 0
|
|
}
|
|
|
|
if remaining > 0 && historyOffset < historyTotal {
|
|
historyItems, listErr := s.listPublishTasksByStatuses(
|
|
ctx,
|
|
owner,
|
|
historyStatuses,
|
|
title,
|
|
remaining,
|
|
historyOffset,
|
|
`t.updated_at DESC, t.desktop_id DESC`,
|
|
)
|
|
if listErr != nil {
|
|
return nil, listErr
|
|
}
|
|
items = append(items, historyItems...)
|
|
}
|
|
|
|
return &DesktopPublishTaskList{
|
|
Items: items,
|
|
Page: page,
|
|
PageSize: pageSize,
|
|
Total: total,
|
|
PendingCount: len(pendingItems),
|
|
HistoryTotal: historyTotal,
|
|
}, nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) listPublishTasksByStatuses(
|
|
ctx context.Context,
|
|
owner publishTaskListOwner,
|
|
statuses []string,
|
|
title string,
|
|
limit int,
|
|
offset int,
|
|
orderBy string,
|
|
) ([]DesktopTaskView, error) {
|
|
query, args := buildListPublishTasksByStatusesQuery(owner, statuses, title, limit, offset, orderBy)
|
|
rows, err := s.pool.Query(ctx, query, args...)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50109, "desktop_publish_tasks_query_failed", "failed to list desktop publish tasks")
|
|
}
|
|
defer rows.Close()
|
|
|
|
return scanDesktopTaskRows(rows)
|
|
}
|
|
|
|
func buildListPublishTasksByStatusesQuery(
|
|
owner publishTaskListOwner,
|
|
statuses []string,
|
|
title string,
|
|
limit int,
|
|
offset int,
|
|
orderBy string,
|
|
) (string, []any) {
|
|
query := `
|
|
SELECT
|
|
t.desktop_id,
|
|
t.job_id,
|
|
t.tenant_id,
|
|
t.workspace_id,
|
|
t.target_account_id,
|
|
t.target_client_id,
|
|
t.platform_id,
|
|
t.kind,
|
|
t.payload,
|
|
t.status,
|
|
t.dedup_key,
|
|
t.active_attempt_id,
|
|
t.lease_expires_at,
|
|
t.attempts,
|
|
t.result,
|
|
t.error,
|
|
t.created_at,
|
|
t.updated_at,
|
|
t.publish_submit_started_at,
|
|
j.status,
|
|
j.compliance_blocked_record_id,
|
|
j.compliance_blocked_at,
|
|
j.compliance_blocked_reason
|
|
FROM desktop_tasks AS t
|
|
JOIN desktop_publish_jobs AS j
|
|
ON j.desktop_id = t.job_id
|
|
AND j.workspace_id = t.workspace_id
|
|
WHERE t.tenant_id = $1
|
|
AND t.workspace_id = $2
|
|
AND j.tenant_id = $1
|
|
AND j.created_by_user_id = $3
|
|
AND t.kind = 'publish'
|
|
AND t.status = ANY($4)
|
|
AND (
|
|
$5 = ''
|
|
OR COALESCE(t.payload->>'title', j.title, '') ILIKE '%' || $5 || '%'
|
|
)
|
|
` + publishTaskVisibleRecordFilterSQL
|
|
args := []any{owner.TenantID, owner.WorkspaceID, owner.UserID, statuses, title}
|
|
|
|
if strings.TrimSpace(orderBy) != "" {
|
|
query += "\nORDER BY " + orderBy
|
|
}
|
|
if limit > 0 {
|
|
query += "\nLIMIT $6 OFFSET $7"
|
|
args = append(args, limit, offset)
|
|
}
|
|
return query, args
|
|
}
|
|
|
|
func (s *DesktopTaskService) countPublishTasksByStatuses(
|
|
ctx context.Context,
|
|
owner publishTaskListOwner,
|
|
statuses []string,
|
|
title string,
|
|
) (int, error) {
|
|
var count int
|
|
query, args := buildCountPublishTasksByStatusesQuery(owner, statuses, title)
|
|
err := s.pool.QueryRow(ctx, query, args...).Scan(&count)
|
|
if err != nil {
|
|
return 0, response.ErrInternal(50109, "desktop_publish_tasks_query_failed", "failed to list desktop publish tasks")
|
|
}
|
|
return count, nil
|
|
}
|
|
|
|
func buildCountPublishTasksByStatusesQuery(
|
|
owner publishTaskListOwner,
|
|
statuses []string,
|
|
title string,
|
|
) (string, []any) {
|
|
query := `
|
|
SELECT COUNT(1)
|
|
FROM desktop_tasks AS t
|
|
JOIN desktop_publish_jobs AS j
|
|
ON j.desktop_id = t.job_id
|
|
AND j.workspace_id = t.workspace_id
|
|
WHERE t.tenant_id = $1
|
|
AND t.workspace_id = $2
|
|
AND j.tenant_id = $1
|
|
AND j.created_by_user_id = $3
|
|
AND t.kind = 'publish'
|
|
AND t.status = ANY($4)
|
|
AND (
|
|
$5 = ''
|
|
OR COALESCE(t.payload->>'title', j.title, '') ILIKE '%' || $5 || '%'
|
|
)
|
|
` + publishTaskVisibleRecordFilterSQL
|
|
return query, []any{owner.TenantID, owner.WorkspaceID, owner.UserID, statuses, title}
|
|
}
|
|
|
|
func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
|
|
items := make([]DesktopTaskView, 0)
|
|
for rows.Next() {
|
|
var (
|
|
desktopID uuid.UUID
|
|
jobID uuid.UUID
|
|
tenantID int64
|
|
workspaceID int64
|
|
targetAccountID uuid.UUID
|
|
targetClientID uuid.UUID
|
|
platform string
|
|
kind string
|
|
payload []byte
|
|
status string
|
|
dedupKey pgtype.Text
|
|
activeAttemptID pgtype.UUID
|
|
leaseExpiresAt pgtype.Timestamptz
|
|
attempts int32
|
|
resultJSON []byte
|
|
errorJSON []byte
|
|
createdAt time.Time
|
|
updatedAt time.Time
|
|
publishSubmitStartedAt pgtype.Timestamptz
|
|
jobStatus string
|
|
blockedRecordID pgtype.Int8
|
|
blockedAt pgtype.Timestamptz
|
|
blockedReason pgtype.Text
|
|
)
|
|
if err := rows.Scan(
|
|
&desktopID,
|
|
&jobID,
|
|
&tenantID,
|
|
&workspaceID,
|
|
&targetAccountID,
|
|
&targetClientID,
|
|
&platform,
|
|
&kind,
|
|
&payload,
|
|
&status,
|
|
&dedupKey,
|
|
&activeAttemptID,
|
|
&leaseExpiresAt,
|
|
&attempts,
|
|
&resultJSON,
|
|
&errorJSON,
|
|
&createdAt,
|
|
&updatedAt,
|
|
&publishSubmitStartedAt,
|
|
&jobStatus,
|
|
&blockedRecordID,
|
|
&blockedAt,
|
|
&blockedReason,
|
|
); err != nil {
|
|
return nil, response.ErrInternal(50110, "desktop_publish_tasks_scan_failed", "failed to scan desktop publish tasks")
|
|
}
|
|
|
|
var dedupKeyText *string
|
|
if dedupKey.Valid {
|
|
value := dedupKey.String
|
|
dedupKeyText = &value
|
|
}
|
|
|
|
var activeAttemptIDText *string
|
|
if activeAttemptID.Valid {
|
|
value, convErr := uuid.FromBytes(activeAttemptID.Bytes[:])
|
|
if convErr == nil {
|
|
text := value.String()
|
|
activeAttemptIDText = &text
|
|
}
|
|
}
|
|
|
|
var leaseExpiresAtValue *time.Time
|
|
if leaseExpiresAt.Valid {
|
|
value := leaseExpiresAt.Time
|
|
leaseExpiresAtValue = &value
|
|
}
|
|
var publishSubmitStartedAtValue *time.Time
|
|
if publishSubmitStartedAt.Valid {
|
|
value := publishSubmitStartedAt.Time
|
|
publishSubmitStartedAtValue = &value
|
|
}
|
|
|
|
publishJobStatusText := strings.TrimSpace(jobStatus)
|
|
var publishJobStatus *string
|
|
if publishJobStatusText != "" {
|
|
publishJobStatus = &publishJobStatusText
|
|
}
|
|
|
|
var complianceBlockedRecordID *int64
|
|
if blockedRecordID.Valid {
|
|
value := blockedRecordID.Int64
|
|
complianceBlockedRecordID = &value
|
|
}
|
|
|
|
var complianceBlockedAt *time.Time
|
|
if blockedAt.Valid {
|
|
value := blockedAt.Time
|
|
complianceBlockedAt = &value
|
|
}
|
|
|
|
var complianceBlockedReason *string
|
|
if blockedReason.Valid {
|
|
value := blockedReason.String
|
|
complianceBlockedReason = &value
|
|
}
|
|
|
|
items = append(items, DesktopTaskView{
|
|
ID: desktopID.String(),
|
|
JobID: jobID.String(),
|
|
TenantID: tenantID,
|
|
WorkspaceID: workspaceID,
|
|
TargetAccountID: targetAccountID.String(),
|
|
TargetClientID: targetClientID.String(),
|
|
Platform: platform,
|
|
Kind: kind,
|
|
Payload: json.RawMessage(payload),
|
|
Status: normalizeDesktopTaskViewStatus(kind, status),
|
|
PublishJobStatus: publishJobStatus,
|
|
ComplianceBlockedRecordID: complianceBlockedRecordID,
|
|
ComplianceBlockedAt: complianceBlockedAt,
|
|
ComplianceBlockedReason: complianceBlockedReason,
|
|
DedupKey: dedupKeyText,
|
|
ActiveAttemptID: activeAttemptIDText,
|
|
LeaseExpiresAt: leaseExpiresAtValue,
|
|
PublishSubmitStartedAt: publishSubmitStartedAtValue,
|
|
Attempts: int(attempts),
|
|
Result: json.RawMessage(resultJSON),
|
|
Error: json.RawMessage(errorJSON),
|
|
CreatedAt: createdAt,
|
|
UpdatedAt: updatedAt,
|
|
})
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, response.ErrInternal(50109, "desktop_publish_tasks_query_failed", "failed to list desktop publish tasks")
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) classifyLeaseError(ctx context.Context, client *repository.DesktopClient, taskID *uuid.UUID) error {
|
|
if taskID == nil {
|
|
return response.ErrConflict(40986, "desktop_task_unavailable", "no desktop task is currently available")
|
|
}
|
|
|
|
task, err := s.repo.GetByDesktopID(ctx, *taskID, client.WorkspaceID)
|
|
if err != nil {
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found")
|
|
}
|
|
return response.ErrInternal(50095, "desktop_task_lookup_failed", "failed to inspect desktop task state")
|
|
}
|
|
|
|
if task.TargetClientID != client.ID && !s.clientCanLeaseMonitorTask(ctx, client, task) {
|
|
return response.ErrForbidden(40381, "desktop_task_not_target_client", "desktop task is assigned to another desktop client")
|
|
}
|
|
|
|
return classifyDesktopTaskLeaseState(task)
|
|
}
|
|
|
|
func classifyDesktopTaskLeaseState(task *repository.DesktopTask) error {
|
|
if task == nil {
|
|
return response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found")
|
|
}
|
|
if task.Kind == "monitor" && task.Status == "queued" {
|
|
return errDesktopTaskLeaseDeferred
|
|
}
|
|
if task.Status == "in_progress" {
|
|
return response.ErrConflict(40989, "desktop_task_already_in_progress", "desktop task is already in progress")
|
|
}
|
|
return response.ErrConflict(40991, "desktop_task_not_queued", "desktop task is not queued")
|
|
}
|
|
|
|
func (s *DesktopTaskService) clientCanLeaseMonitorTask(ctx context.Context, client *repository.DesktopClient, task *repository.DesktopTask) bool {
|
|
if s == nil || client == nil || task == nil || task.Kind != "monitor" || task.TargetAccountID == uuid.Nil {
|
|
return false
|
|
}
|
|
var canLease bool
|
|
if err := s.pool.QueryRow(ctx, `
|
|
SELECT EXISTS (
|
|
SELECT 1
|
|
FROM platform_accounts AS target_account
|
|
JOIN platform_accounts AS client_account
|
|
ON client_account.tenant_id = target_account.tenant_id
|
|
AND client_account.workspace_id = target_account.workspace_id
|
|
AND client_account.user_id = target_account.user_id
|
|
AND client_account.platform_id = target_account.platform_id
|
|
AND client_account.client_id = $4
|
|
AND client_account.deleted_at IS NULL
|
|
AND client_account.delete_requested_at IS NULL
|
|
AND `+monitorAccountIdentityMatchSQL("client_account", "target_account")+`
|
|
WHERE target_account.desktop_id = $1
|
|
AND target_account.tenant_id = $2
|
|
AND target_account.workspace_id = $3
|
|
)
|
|
`, task.TargetAccountID, task.TenantID, task.WorkspaceID, client.ID).Scan(&canLease); err != nil {
|
|
return false
|
|
}
|
|
return canLease
|
|
}
|
|
|
|
func buildDesktopTaskView(task *repository.DesktopTask) DesktopTaskView {
|
|
var activeAttemptID *string
|
|
if task.ActiveAttemptID != nil {
|
|
value := task.ActiveAttemptID.String()
|
|
activeAttemptID = &value
|
|
}
|
|
|
|
return DesktopTaskView{
|
|
ID: task.DesktopID.String(),
|
|
JobID: task.JobID.String(),
|
|
TenantID: task.TenantID,
|
|
WorkspaceID: task.WorkspaceID,
|
|
TargetAccountID: task.TargetAccountID.String(),
|
|
TargetClientID: task.TargetClientID.String(),
|
|
Platform: task.Platform,
|
|
Kind: task.Kind,
|
|
Payload: json.RawMessage(task.Payload),
|
|
Status: normalizeDesktopTaskViewStatus(task.Kind, task.Status),
|
|
DedupKey: task.DedupKey,
|
|
ActiveAttemptID: activeAttemptID,
|
|
LeaseExpiresAt: task.LeaseExpiresAt,
|
|
PublishSubmitStartedAt: task.PublishSubmitStartedAt,
|
|
Attempts: task.Attempts,
|
|
Result: json.RawMessage(task.Result),
|
|
Error: json.RawMessage(task.Error),
|
|
CreatedAt: task.CreatedAt,
|
|
UpdatedAt: task.UpdatedAt,
|
|
}
|
|
}
|
|
|
|
func marshalOptionalJSON(value any) ([]byte, error) {
|
|
if value == nil {
|
|
return nil, nil
|
|
}
|
|
return json.Marshal(value)
|
|
}
|
|
|
|
func optionalStringValue(value *string) string {
|
|
if value == nil {
|
|
return ""
|
|
}
|
|
return strings.TrimSpace(*value)
|
|
}
|
|
|
|
func literalStringPtr(value string) *string {
|
|
return &value
|
|
}
|
|
|
|
func normalizeDesktopTaskTerminalStatus(status string) string {
|
|
switch strings.TrimSpace(status) {
|
|
case "unknown":
|
|
return "failed"
|
|
default:
|
|
return strings.TrimSpace(status)
|
|
}
|
|
}
|
|
|
|
func normalizeDesktopTaskCompletionStatus(status string) string {
|
|
switch strings.TrimSpace(status) {
|
|
case "succeeded", "failed", "unknown":
|
|
return strings.TrimSpace(status)
|
|
default:
|
|
return strings.TrimSpace(status)
|
|
}
|
|
}
|
|
|
|
func normalizeDesktopTaskCompletionStatusForKind(kind string, status string) string {
|
|
if strings.TrimSpace(kind) == "monitor" && strings.TrimSpace(status) == "unknown" {
|
|
return "failed"
|
|
}
|
|
return strings.TrimSpace(status)
|
|
}
|
|
|
|
func normalizeDesktopTaskViewStatus(kind string, status string) string {
|
|
switch strings.TrimSpace(kind) {
|
|
case "monitor":
|
|
return normalizeDesktopTaskTerminalStatus(status)
|
|
case "publish":
|
|
return strings.TrimSpace(status)
|
|
default:
|
|
return strings.TrimSpace(status)
|
|
}
|
|
}
|
|
|
|
type desktopTaskRecoveryMode string
|
|
|
|
const (
|
|
desktopTaskRecoveryModeStartup desktopTaskRecoveryMode = "startup"
|
|
desktopTaskRecoveryModeDisconnect desktopTaskRecoveryMode = "disconnect"
|
|
desktopTaskRecoveryModeLeaseExpiry desktopTaskRecoveryMode = "lease_expired"
|
|
)
|
|
|
|
type recoveredDesktopTaskLease struct {
|
|
TaskID uuid.UUID
|
|
WorkspaceID int64
|
|
Kind string
|
|
Status string
|
|
ActiveAttempt pgtype.UUID
|
|
MonitorTaskID pgtype.Int8
|
|
Attempts int
|
|
SubmitStarted bool
|
|
ErrorJSON []byte
|
|
}
|
|
|
|
type PublishLeaseRecoveryResult struct {
|
|
Requeued int `json:"requeued"`
|
|
Failed int `json:"failed"`
|
|
Uncertain int `json:"uncertain"`
|
|
}
|
|
|
|
type MonitorLeaseRecoveryResult struct {
|
|
Requeued int `json:"requeued"`
|
|
Failed int `json:"failed"`
|
|
}
|
|
|
|
// resolvePublishRecoveryOutcome decides what to do with a publish task whose lease was lost.
|
|
//
|
|
// If the client had already entered the irreversible submit phase (publish_submit_started_at
|
|
// is set), the article may already exist on the platform. Auto-requeueing would silently
|
|
// re-post a non-idempotent article, so we route the task to 'unknown' (kept in the dedup
|
|
// active set, surfaced for manual reconcile) instead. Tasks that never reached submit are
|
|
// safe to auto-requeue while attempts remain, and give up to 'failed' once exhausted.
|
|
func resolvePublishRecoveryOutcome(submitStarted bool, attempts int) (status string, payloadKind string) {
|
|
switch {
|
|
case submitStarted:
|
|
return "unknown", "publish_uncertain"
|
|
case attempts >= desktopPublishMaxAttempts:
|
|
return "failed", "publish_final"
|
|
default:
|
|
return "queued", "publish"
|
|
}
|
|
}
|
|
|
|
func resolveMonitorRecoveryOutcome(mode desktopTaskRecoveryMode, attempts int) (status string, payloadKind string) {
|
|
if mode != desktopTaskRecoveryModeLeaseExpiry && attempts >= 2 {
|
|
return "failed", "monitor_final"
|
|
}
|
|
return "queued", "monitor"
|
|
}
|
|
|
|
func (s *DesktopTaskService) recoverClientTasksOnStartup(ctx context.Context, client *repository.DesktopClient) error {
|
|
return s.recoverClientTasks(ctx, client, desktopTaskRecoveryModeStartup)
|
|
}
|
|
|
|
func (s *DesktopTaskService) recoverClientTasksOnDisconnect(ctx context.Context, client *repository.DesktopClient) error {
|
|
return s.recoverClientTasks(ctx, client, desktopTaskRecoveryModeDisconnect)
|
|
}
|
|
|
|
func (s *DesktopTaskService) recoverExpiredClientTasks(ctx context.Context, client *repository.DesktopClient) error {
|
|
return s.recoverClientTasks(ctx, client, desktopTaskRecoveryModeLeaseExpiry)
|
|
}
|
|
|
|
func (s *DesktopTaskService) recoverClientTasks(
|
|
ctx context.Context,
|
|
client *repository.DesktopClient,
|
|
mode desktopTaskRecoveryMode,
|
|
) error {
|
|
if s == nil || client == nil || s.pool == nil {
|
|
return nil
|
|
}
|
|
|
|
publishErrorJSON, publishReason, publishMessage, err := desktopTaskRecoveryPayload(mode, "publish")
|
|
if err != nil {
|
|
return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload")
|
|
}
|
|
publishFinalErrorJSON, _, _, err := desktopTaskRecoveryPayload(mode, "publish_final")
|
|
if err != nil {
|
|
return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload")
|
|
}
|
|
publishUncertainErrorJSON, _, _, err := desktopTaskRecoveryPayload(mode, "publish_uncertain")
|
|
if err != nil {
|
|
return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload")
|
|
}
|
|
monitorErrorJSON, _, _, err := desktopTaskRecoveryPayload(mode, "monitor")
|
|
if err != nil {
|
|
return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
|
|
}
|
|
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return response.ErrInternal(50196, "desktop_task_recovery_begin_failed", "failed to start desktop task recovery")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
repo := repository.NewDesktopTaskRepository(tx)
|
|
query := desktopTaskRecoverySelectQuery(mode)
|
|
queryArgs := []any{client.ID, client.WorkspaceID}
|
|
|
|
rows, err := tx.Query(ctx, query, queryArgs...)
|
|
if err != nil {
|
|
s.logWarn("desktop task recovery query failed", err, zap.String("client_id", client.ID.String()), zap.String("mode", string(mode)))
|
|
return response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
|
|
}
|
|
defer rows.Close()
|
|
|
|
recovered := make([]recoveredDesktopTaskLease, 0)
|
|
for rows.Next() {
|
|
var item recoveredDesktopTaskLease
|
|
if scanErr := rows.Scan(
|
|
&item.TaskID,
|
|
&item.WorkspaceID,
|
|
&item.Kind,
|
|
&item.Status,
|
|
&item.ActiveAttempt,
|
|
&item.Attempts,
|
|
&item.SubmitStarted,
|
|
&item.MonitorTaskID,
|
|
); scanErr != nil {
|
|
s.logWarn("desktop task recovery scan failed", scanErr, zap.String("client_id", client.ID.String()), zap.String("mode", string(mode)))
|
|
return response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks")
|
|
}
|
|
recovered = append(recovered, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
s.logWarn("desktop task recovery rows failed", err, zap.String("client_id", client.ID.String()), zap.String("mode", string(mode)))
|
|
return response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks")
|
|
}
|
|
|
|
publishOutcomes := make([]*desktopPublishSyncOutcome, 0)
|
|
for index := range recovered {
|
|
item := &recovered[index]
|
|
if item.Kind == "monitor" {
|
|
finalStatus, monitorPayloadKind := resolveMonitorRecoveryOutcome(mode, item.Attempts)
|
|
monitorErrorForTask, monitorReasonForTask, _, payloadErr := desktopTaskRecoveryPayload(mode, monitorPayloadKind)
|
|
if payloadErr != nil {
|
|
return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
|
|
}
|
|
if finalStatus == "failed" {
|
|
if item.MonitorTaskID.Valid {
|
|
failedMonitorTaskIDs, failErr := s.failMonitorCollectTasksForDesktopRecovery(ctx, []int64{item.MonitorTaskID.Int64}, monitorErrorForTask)
|
|
if failErr != nil {
|
|
return failErr
|
|
}
|
|
if len(failedMonitorTaskIDs) == 0 {
|
|
continue
|
|
}
|
|
}
|
|
} else if item.MonitorTaskID.Valid {
|
|
requeuedMonitorTaskIDs, requeueErr := s.requeueMonitorCollectTasksForDesktopRecovery(ctx, []int64{item.MonitorTaskID.Int64}, monitorErrorForTask)
|
|
if requeueErr != nil {
|
|
return requeueErr
|
|
}
|
|
if len(requeuedMonitorTaskIDs) == 0 {
|
|
continue
|
|
}
|
|
}
|
|
if _, execErr := tx.Exec(ctx, `
|
|
UPDATE desktop_tasks
|
|
SET status = $2,
|
|
error = $3,
|
|
result = CASE WHEN $2 = 'queued' THEN NULL ELSE result END,
|
|
active_attempt_id = NULL,
|
|
lease_token_hash = NULL,
|
|
lease_expires_at = NULL,
|
|
interrupted_at = COALESCE(interrupted_at, NOW()),
|
|
interrupt_reason = $4,
|
|
enqueued_at = CASE WHEN $2 = 'queued' THEN NOW() ELSE enqueued_at END,
|
|
updated_at = NOW()
|
|
WHERE desktop_id = $1
|
|
`, item.TaskID, finalStatus, monitorErrorForTask, monitorReasonForTask); execErr != nil {
|
|
s.logWarn("desktop monitor task recovery update failed", execErr, zap.String("task_id", item.TaskID.String()), zap.String("mode", string(mode)))
|
|
return response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
|
|
}
|
|
item.Status = finalStatus
|
|
item.ErrorJSON = monitorErrorForTask
|
|
continue
|
|
}
|
|
|
|
finalStatus, payloadKind := resolvePublishRecoveryOutcome(item.SubmitStarted, item.Attempts)
|
|
nextErrorJSON := publishErrorJSON
|
|
switch payloadKind {
|
|
case "publish_final":
|
|
nextErrorJSON = publishFinalErrorJSON
|
|
case "publish_uncertain":
|
|
nextErrorJSON = publishUncertainErrorJSON
|
|
}
|
|
if _, execErr := tx.Exec(ctx, `
|
|
UPDATE desktop_tasks
|
|
SET status = $2,
|
|
error = $3,
|
|
result = CASE WHEN $2 = 'queued' THEN NULL ELSE result END,
|
|
active_attempt_id = NULL,
|
|
lease_token_hash = NULL,
|
|
lease_expires_at = NULL,
|
|
interrupted_at = COALESCE(interrupted_at, NOW()),
|
|
interrupt_reason = COALESCE(interrupt_reason, $4),
|
|
enqueued_at = CASE WHEN $2 = 'queued' THEN NOW() ELSE enqueued_at END,
|
|
updated_at = NOW()
|
|
WHERE desktop_id = $1
|
|
`, item.TaskID, finalStatus, nextErrorJSON, publishReason); execErr != nil {
|
|
s.logWarn("desktop publish task recovery update failed", execErr, zap.String("task_id", item.TaskID.String()), zap.String("mode", string(mode)))
|
|
return response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
|
|
}
|
|
item.Status = finalStatus
|
|
item.ErrorJSON = nextErrorJSON
|
|
// Only terminal 'failed' tasks finalize the publish_record; 'unknown' (may already be
|
|
// published) is left in the dedup active set for manual reconcile, 'queued' is requeued.
|
|
if finalStatus == "failed" {
|
|
task, getErr := repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
|
|
if getErr != nil {
|
|
s.logWarn("desktop publish task recovery reload failed", getErr, zap.String("task_id", item.TaskID.String()), zap.String("mode", string(mode)))
|
|
return response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
|
|
}
|
|
publishOutcome, syncErr := syncDesktopPublishTaskState(ctx, tx, task)
|
|
if syncErr != nil {
|
|
return syncErr
|
|
}
|
|
if publishOutcome != nil {
|
|
publishOutcomes = append(publishOutcomes, publishOutcome)
|
|
}
|
|
}
|
|
}
|
|
|
|
for _, item := range recovered {
|
|
if !item.ActiveAttempt.Valid {
|
|
continue
|
|
}
|
|
|
|
attemptID, convErr := uuid.FromBytes(item.ActiveAttempt.Bytes[:])
|
|
if convErr != nil {
|
|
s.logWarn("desktop recovery attempt id decode failed", convErr, zap.String("task_id", item.TaskID.String()))
|
|
continue
|
|
}
|
|
|
|
finalStatus := literalStringPtr(item.Status)
|
|
errorJSON := item.ErrorJSON
|
|
if len(errorJSON) == 0 {
|
|
errorJSON = publishErrorJSON
|
|
}
|
|
if item.Kind == "monitor" {
|
|
if item.Status == "failed" {
|
|
finalStatus = literalStringPtr("failed")
|
|
} else {
|
|
finalStatus = literalStringPtr("aborted")
|
|
}
|
|
if len(item.ErrorJSON) > 0 {
|
|
errorJSON = item.ErrorJSON
|
|
} else {
|
|
errorJSON = monitorErrorJSON
|
|
}
|
|
}
|
|
|
|
if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
|
|
TaskID: item.TaskID,
|
|
AttemptID: attemptID,
|
|
FinalStatus: finalStatus,
|
|
Error: errorJSON,
|
|
}); finishErr != nil {
|
|
s.logWarn("desktop recovery attempt finish failed", finishErr, zap.String("task_id", item.TaskID.String()))
|
|
}
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return response.ErrInternal(50199, "desktop_task_recovery_commit_failed", "failed to commit desktop task recovery")
|
|
}
|
|
|
|
s.afterRecoveredPublishOutcomes(ctx, publishOutcomes)
|
|
|
|
for _, item := range recovered {
|
|
task, getErr := s.repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
|
|
if getErr != nil {
|
|
s.logWarn("desktop recovery reload failed", getErr, zap.String("task_id", item.TaskID.String()))
|
|
continue
|
|
}
|
|
|
|
eventType := "task_completed"
|
|
if item.Status == "queued" {
|
|
eventType = "task_available"
|
|
}
|
|
s.publishTaskEvent(ctx, task, eventType)
|
|
}
|
|
|
|
if len(recovered) > 0 && s.logger != nil {
|
|
s.logger.Info("desktop tasks recovered",
|
|
zap.String("client_id", client.ID.String()),
|
|
zap.String("mode", string(mode)),
|
|
zap.Int("count", len(recovered)),
|
|
zap.String("publish_message", publishMessage),
|
|
)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) RecoverExpiredPublishTasks(ctx context.Context, limit int) (PublishLeaseRecoveryResult, error) {
|
|
var result PublishLeaseRecoveryResult
|
|
if s == nil || s.pool == nil {
|
|
return result, nil
|
|
}
|
|
if limit <= 0 {
|
|
limit = 1000
|
|
}
|
|
|
|
publishErrorJSON, publishReason, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "publish")
|
|
if err != nil {
|
|
return result, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload")
|
|
}
|
|
publishFinalErrorJSON, _, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "publish_final")
|
|
if err != nil {
|
|
return result, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload")
|
|
}
|
|
publishUncertainErrorJSON, _, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "publish_uncertain")
|
|
if err != nil {
|
|
return result, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload")
|
|
}
|
|
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return result, response.ErrInternal(50196, "desktop_task_recovery_begin_failed", "failed to start desktop task recovery")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
rows, err := tx.Query(ctx, `
|
|
SELECT
|
|
t.desktop_id,
|
|
t.workspace_id,
|
|
t.kind,
|
|
t.status,
|
|
t.active_attempt_id,
|
|
t.attempts,
|
|
(t.publish_submit_started_at IS NOT NULL) AS submit_started
|
|
FROM desktop_tasks AS t
|
|
WHERE t.kind = 'publish'
|
|
AND t.status = 'in_progress'
|
|
AND t.lease_expires_at IS NOT NULL
|
|
AND t.lease_expires_at < NOW()
|
|
ORDER BY t.lease_expires_at ASC, t.updated_at ASC, t.desktop_id ASC
|
|
LIMIT $1
|
|
FOR UPDATE OF t SKIP LOCKED
|
|
`, limit)
|
|
if err != nil {
|
|
s.logWarn("expired desktop publish recovery query failed", err)
|
|
return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
|
|
}
|
|
defer rows.Close()
|
|
|
|
recovered := make([]recoveredDesktopTaskLease, 0)
|
|
for rows.Next() {
|
|
var item recoveredDesktopTaskLease
|
|
if scanErr := rows.Scan(
|
|
&item.TaskID,
|
|
&item.WorkspaceID,
|
|
&item.Kind,
|
|
&item.Status,
|
|
&item.ActiveAttempt,
|
|
&item.Attempts,
|
|
&item.SubmitStarted,
|
|
); scanErr != nil {
|
|
s.logWarn("expired desktop publish recovery scan failed", scanErr)
|
|
return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks")
|
|
}
|
|
recovered = append(recovered, item)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
s.logWarn("expired desktop publish recovery rows failed", err)
|
|
return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks")
|
|
}
|
|
|
|
repo := repository.NewDesktopTaskRepository(tx)
|
|
publishOutcomes := make([]*desktopPublishSyncOutcome, 0)
|
|
for index := range recovered {
|
|
item := &recovered[index]
|
|
finalStatus, payloadKind := resolvePublishRecoveryOutcome(item.SubmitStarted, item.Attempts)
|
|
nextErrorJSON := publishErrorJSON
|
|
switch payloadKind {
|
|
case "publish_final":
|
|
nextErrorJSON = publishFinalErrorJSON
|
|
case "publish_uncertain":
|
|
nextErrorJSON = publishUncertainErrorJSON
|
|
}
|
|
if _, execErr := tx.Exec(ctx, `
|
|
UPDATE desktop_tasks
|
|
SET status = $2,
|
|
error = $3,
|
|
result = CASE WHEN $2 = 'queued' THEN NULL ELSE result END,
|
|
active_attempt_id = NULL,
|
|
lease_token_hash = NULL,
|
|
lease_expires_at = NULL,
|
|
interrupted_at = COALESCE(interrupted_at, NOW()),
|
|
interrupt_reason = COALESCE(interrupt_reason, $4),
|
|
enqueued_at = CASE WHEN $2 = 'queued' THEN NOW() ELSE enqueued_at END,
|
|
updated_at = NOW()
|
|
WHERE desktop_id = $1
|
|
AND status = 'in_progress'
|
|
`, item.TaskID, finalStatus, nextErrorJSON, publishReason); execErr != nil {
|
|
s.logWarn("expired desktop publish recovery update failed", execErr, zap.String("task_id", item.TaskID.String()))
|
|
return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
|
|
}
|
|
|
|
item.Status = finalStatus
|
|
item.ErrorJSON = nextErrorJSON
|
|
switch finalStatus {
|
|
case "failed":
|
|
result.Failed++
|
|
task, getErr := repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
|
|
if getErr != nil {
|
|
s.logWarn("expired desktop publish recovery reload failed", getErr, zap.String("task_id", item.TaskID.String()))
|
|
return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
|
|
}
|
|
publishOutcome, syncErr := syncDesktopPublishTaskState(ctx, tx, task)
|
|
if syncErr != nil {
|
|
return result, syncErr
|
|
}
|
|
if publishOutcome != nil {
|
|
publishOutcomes = append(publishOutcomes, publishOutcome)
|
|
}
|
|
case "unknown":
|
|
// May already be published — keep as 'unknown' (still in the dedup active set) for
|
|
// manual reconcile. Do NOT sync to a terminal publish_record state and do NOT requeue.
|
|
result.Uncertain++
|
|
default:
|
|
result.Requeued++
|
|
}
|
|
}
|
|
|
|
for _, item := range recovered {
|
|
if !item.ActiveAttempt.Valid {
|
|
continue
|
|
}
|
|
|
|
attemptID, convErr := uuid.FromBytes(item.ActiveAttempt.Bytes[:])
|
|
if convErr != nil {
|
|
s.logWarn("expired desktop publish recovery attempt id decode failed", convErr, zap.String("task_id", item.TaskID.String()))
|
|
continue
|
|
}
|
|
|
|
if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
|
|
TaskID: item.TaskID,
|
|
AttemptID: attemptID,
|
|
FinalStatus: literalStringPtr(item.Status),
|
|
Error: item.ErrorJSON,
|
|
}); finishErr != nil {
|
|
s.logWarn("expired desktop publish recovery attempt finish failed", finishErr, zap.String("task_id", item.TaskID.String()))
|
|
}
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return result, response.ErrInternal(50199, "desktop_task_recovery_commit_failed", "failed to commit desktop task recovery")
|
|
}
|
|
|
|
s.afterRecoveredPublishOutcomes(ctx, publishOutcomes)
|
|
|
|
for _, item := range recovered {
|
|
task, getErr := s.repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
|
|
if getErr != nil {
|
|
s.logWarn("expired desktop publish recovery reload after commit failed", getErr, zap.String("task_id", item.TaskID.String()))
|
|
continue
|
|
}
|
|
eventType := "task_completed"
|
|
if item.Status == "queued" {
|
|
eventType = "task_available"
|
|
}
|
|
s.publishTaskEvent(ctx, task, eventType)
|
|
}
|
|
|
|
if len(recovered) > 0 && s.logger != nil {
|
|
s.logger.Info("expired desktop publish tasks recovered",
|
|
zap.Int("requeued", result.Requeued),
|
|
zap.Int("failed", result.Failed),
|
|
zap.Int("uncertain", result.Uncertain),
|
|
)
|
|
}
|
|
|
|
return result, nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, limit int) (MonitorLeaseRecoveryResult, error) {
|
|
var result MonitorLeaseRecoveryResult
|
|
if s == nil || s.pool == nil {
|
|
return result, nil
|
|
}
|
|
if limit <= 0 {
|
|
limit = 1000
|
|
}
|
|
|
|
errorJSON, reason, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "monitor")
|
|
if err != nil {
|
|
return result, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
|
|
}
|
|
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return result, response.ErrInternal(50196, "desktop_task_recovery_begin_failed", "failed to start desktop task recovery")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
rows, err := tx.Query(ctx, expiredMonitorDesktopTasksCandidateSQL(), limit)
|
|
if err != nil {
|
|
s.logWarn("expired desktop monitor recovery query failed", err)
|
|
return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
|
|
}
|
|
defer rows.Close()
|
|
|
|
recovered := make([]recoveredDesktopTaskLease, 0)
|
|
monitorTaskIDs := make([]int64, 0)
|
|
for rows.Next() {
|
|
var item recoveredDesktopTaskLease
|
|
if scanErr := rows.Scan(
|
|
&item.TaskID,
|
|
&item.WorkspaceID,
|
|
&item.Kind,
|
|
&item.Status,
|
|
&item.ActiveAttempt,
|
|
&item.Attempts,
|
|
&item.SubmitStarted,
|
|
&item.MonitorTaskID,
|
|
); scanErr != nil {
|
|
s.logWarn("expired desktop monitor recovery scan failed", scanErr)
|
|
return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks")
|
|
}
|
|
item.Status = "queued"
|
|
item.ErrorJSON = errorJSON
|
|
recovered = append(recovered, item)
|
|
if item.MonitorTaskID.Valid {
|
|
monitorTaskIDs = append(monitorTaskIDs, item.MonitorTaskID.Int64)
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
s.logWarn("expired desktop monitor recovery rows failed", err)
|
|
return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks")
|
|
}
|
|
|
|
requeuedMonitorTaskIDs, err := s.requeueMonitorCollectTasksForDesktopRecovery(ctx, monitorTaskIDs, errorJSON)
|
|
if err != nil {
|
|
return result, err
|
|
}
|
|
requeuedMonitorTaskSet := int64Set(requeuedMonitorTaskIDs)
|
|
recoverableDesktopIDs := make([]uuid.UUID, 0, len(recovered))
|
|
recoverable := make([]recoveredDesktopTaskLease, 0, len(recovered))
|
|
for _, item := range recovered {
|
|
if !item.MonitorTaskID.Valid {
|
|
continue
|
|
}
|
|
if _, ok := requeuedMonitorTaskSet[item.MonitorTaskID.Int64]; !ok {
|
|
continue
|
|
}
|
|
recoverableDesktopIDs = append(recoverableDesktopIDs, item.TaskID)
|
|
recoverable = append(recoverable, item)
|
|
}
|
|
if len(recoverableDesktopIDs) == 0 {
|
|
return result, nil
|
|
}
|
|
|
|
updateRows, err := tx.Query(ctx, requeueExpiredMonitorDesktopTasksSQL(), recoverableDesktopIDs, errorJSON, reason)
|
|
if err != nil {
|
|
s.logWarn("expired desktop monitor recovery update failed", err)
|
|
return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
|
|
}
|
|
defer updateRows.Close()
|
|
|
|
updatedDesktopIDs := make(map[uuid.UUID]struct{}, len(recoverableDesktopIDs))
|
|
for updateRows.Next() {
|
|
var desktopID uuid.UUID
|
|
if scanErr := updateRows.Scan(&desktopID); scanErr != nil {
|
|
s.logWarn("expired desktop monitor recovery update scan failed", scanErr)
|
|
return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks")
|
|
}
|
|
updatedDesktopIDs[desktopID] = struct{}{}
|
|
}
|
|
if err := updateRows.Err(); err != nil {
|
|
s.logWarn("expired desktop monitor recovery update rows failed", err)
|
|
return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks")
|
|
}
|
|
|
|
repo := repository.NewDesktopTaskRepository(tx)
|
|
for _, item := range recoverable {
|
|
if _, ok := updatedDesktopIDs[item.TaskID]; !ok {
|
|
continue
|
|
}
|
|
if !item.ActiveAttempt.Valid {
|
|
continue
|
|
}
|
|
|
|
attemptID, convErr := uuid.FromBytes(item.ActiveAttempt.Bytes[:])
|
|
if convErr != nil {
|
|
s.logWarn("expired desktop monitor recovery attempt id decode failed", convErr, zap.String("task_id", item.TaskID.String()))
|
|
continue
|
|
}
|
|
|
|
if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
|
|
TaskID: item.TaskID,
|
|
AttemptID: attemptID,
|
|
FinalStatus: literalStringPtr("aborted"),
|
|
Error: errorJSON,
|
|
}); finishErr != nil {
|
|
s.logWarn("expired desktop monitor recovery attempt finish failed", finishErr, zap.String("task_id", item.TaskID.String()))
|
|
}
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return result, response.ErrInternal(50199, "desktop_task_recovery_commit_failed", "failed to commit desktop task recovery")
|
|
}
|
|
|
|
result.Requeued = len(updatedDesktopIDs)
|
|
|
|
for _, item := range recoverable {
|
|
if _, ok := updatedDesktopIDs[item.TaskID]; !ok {
|
|
continue
|
|
}
|
|
task, getErr := s.repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
|
|
if getErr != nil {
|
|
s.logWarn("expired desktop monitor recovery reload after commit failed", getErr, zap.String("task_id", item.TaskID.String()))
|
|
continue
|
|
}
|
|
s.publishTaskEvent(ctx, task, "task_available")
|
|
}
|
|
|
|
if result.Requeued > 0 && s.logger != nil {
|
|
s.logger.Info("expired desktop monitor tasks recovered",
|
|
zap.Int("requeued", result.Requeued),
|
|
)
|
|
}
|
|
|
|
return result, nil
|
|
}
|
|
|
|
func expiredMonitorDesktopTasksCandidateSQL() string {
|
|
return `
|
|
SELECT
|
|
t.desktop_id,
|
|
t.workspace_id,
|
|
t.kind,
|
|
t.status,
|
|
t.active_attempt_id,
|
|
t.attempts,
|
|
(t.publish_submit_started_at IS NOT NULL) AS submit_started,
|
|
t.monitor_task_id
|
|
FROM desktop_tasks AS t
|
|
WHERE t.kind = 'monitor'
|
|
AND t.status = 'in_progress'
|
|
AND t.monitor_task_id IS NOT NULL
|
|
AND t.lease_expires_at IS NOT NULL
|
|
AND t.lease_expires_at < NOW()
|
|
ORDER BY t.lease_expires_at ASC, t.updated_at ASC, t.desktop_id ASC
|
|
LIMIT $1
|
|
FOR UPDATE OF t SKIP LOCKED
|
|
`
|
|
}
|
|
|
|
func requeueExpiredMonitorDesktopTasksSQL() string {
|
|
return `
|
|
UPDATE desktop_tasks
|
|
SET status = 'queued',
|
|
error = $2,
|
|
result = NULL,
|
|
active_attempt_id = NULL,
|
|
lease_token_hash = NULL,
|
|
lease_expires_at = NULL,
|
|
interrupted_at = COALESCE(interrupted_at, NOW()),
|
|
interrupt_reason = COALESCE(interrupt_reason, $3),
|
|
enqueued_at = NOW(),
|
|
updated_at = NOW()
|
|
WHERE desktop_id = ANY($1::uuid[])
|
|
AND kind = 'monitor'
|
|
AND status = 'in_progress'
|
|
RETURNING desktop_id
|
|
`
|
|
}
|
|
|
|
func (s *DesktopTaskService) requeueMonitorCollectTasksForDesktopRecovery(ctx context.Context, monitorTaskIDs []int64, errorJSON []byte) ([]int64, error) {
|
|
if s == nil || s.monitoringPool == nil || len(monitorTaskIDs) == 0 {
|
|
return nil, nil
|
|
}
|
|
monitorTaskIDs = uniqueInt64s(monitorTaskIDs)
|
|
message := desktopTaskRecoveryMessage(errorJSON)
|
|
requestPayloadJSON, err := json.Marshal(map[string]any{
|
|
"runtime_error": json.RawMessage(errorJSON),
|
|
})
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
|
|
}
|
|
rows, err := s.monitoringPool.Query(ctx, `
|
|
WITH input AS (
|
|
SELECT unnest($1::bigint[]) AS id
|
|
),
|
|
updated AS (
|
|
UPDATE monitoring_collect_tasks
|
|
SET status = 'pending',
|
|
lease_token_hash = NULL,
|
|
leased_to_executor = NULL,
|
|
leased_at = NULL,
|
|
lease_expires_at = NULL,
|
|
callback_received_at = NULL,
|
|
completed_at = NULL,
|
|
skip_reason = NULL,
|
|
dispatch_after = NOW() + interval '30 seconds',
|
|
error_message = NULLIF($2, ''),
|
|
request_payload_json = (
|
|
CASE
|
|
WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
|
|
THEN '{}'::jsonb
|
|
ELSE request_payload_json
|
|
END
|
|
) || $3::jsonb,
|
|
updated_at = NOW()
|
|
WHERE id IN (SELECT id FROM input)
|
|
AND COALESCE(execution_owner, 'legacy') = 'desktop_tasks'
|
|
AND callback_received_at IS NULL
|
|
AND status IN ('pending', 'leased', 'expired')
|
|
RETURNING id
|
|
)
|
|
SELECT id FROM updated
|
|
`, monitorTaskIDs, message, requestPayloadJSON)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50041, "task_update_failed", "failed to requeue expired monitor desktop tasks")
|
|
}
|
|
defer rows.Close()
|
|
|
|
requeuedIDs := make([]int64, 0, len(monitorTaskIDs))
|
|
for rows.Next() {
|
|
var id int64
|
|
if scanErr := rows.Scan(&id); scanErr != nil {
|
|
return nil, response.ErrInternal(50041, "task_update_failed", "failed to parse requeued monitor desktop tasks")
|
|
}
|
|
requeuedIDs = append(requeuedIDs, id)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, response.ErrInternal(50041, "task_update_failed", "failed to iterate requeued monitor desktop tasks")
|
|
}
|
|
return requeuedIDs, nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) failMonitorCollectTasksForDesktopRecovery(ctx context.Context, monitorTaskIDs []int64, errorJSON []byte) ([]int64, error) {
|
|
if s == nil || s.monitoringPool == nil || len(monitorTaskIDs) == 0 {
|
|
return nil, nil
|
|
}
|
|
monitorTaskIDs = uniqueInt64s(monitorTaskIDs)
|
|
message := desktopTaskRecoveryMessage(errorJSON)
|
|
requestPayloadJSON, err := json.Marshal(map[string]any{
|
|
"runtime_error": json.RawMessage(errorJSON),
|
|
})
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
|
|
}
|
|
rows, err := s.monitoringPool.Query(ctx, `
|
|
WITH input AS (
|
|
SELECT unnest($1::bigint[]) AS id
|
|
),
|
|
updated AS (
|
|
UPDATE monitoring_collect_tasks
|
|
SET status = 'failed',
|
|
lease_token_hash = NULL,
|
|
leased_to_executor = NULL,
|
|
leased_at = NULL,
|
|
lease_expires_at = NULL,
|
|
callback_received_at = COALESCE(callback_received_at, NOW()),
|
|
completed_at = COALESCE(completed_at, NOW()),
|
|
skip_reason = NULL,
|
|
error_message = COALESCE(NULLIF(error_message, ''), $2),
|
|
request_payload_json = (
|
|
CASE
|
|
WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
|
|
THEN '{}'::jsonb
|
|
ELSE request_payload_json
|
|
END
|
|
) || $3::jsonb,
|
|
updated_at = NOW()
|
|
WHERE id IN (SELECT id FROM input)
|
|
AND COALESCE(execution_owner, 'legacy') = 'desktop_tasks'
|
|
AND callback_received_at IS NULL
|
|
AND status IN ('pending', 'leased', 'expired')
|
|
RETURNING id
|
|
),
|
|
already_failed AS (
|
|
SELECT t.id
|
|
FROM monitoring_collect_tasks t
|
|
JOIN input i ON i.id = t.id
|
|
WHERE COALESCE(t.execution_owner, 'legacy') = 'desktop_tasks'
|
|
AND t.status = 'failed'
|
|
AND (
|
|
t.error_message = $2
|
|
OR t.request_payload_json #>> '{runtime_error,source}' = 'desktop_task_recovery'
|
|
OR t.request_payload_json #>> '{runtime_error,source}' = 'desktop_client_offline'
|
|
)
|
|
)
|
|
SELECT id FROM updated
|
|
UNION
|
|
SELECT id FROM already_failed
|
|
`, monitorTaskIDs, message, requestPayloadJSON)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50041, "task_update_failed", "failed to fail recovered monitor desktop tasks")
|
|
}
|
|
defer rows.Close()
|
|
|
|
failedIDs := make([]int64, 0, len(monitorTaskIDs))
|
|
for rows.Next() {
|
|
var id int64
|
|
if scanErr := rows.Scan(&id); scanErr != nil {
|
|
return nil, response.ErrInternal(50041, "task_update_failed", "failed to parse failed monitor desktop tasks")
|
|
}
|
|
failedIDs = append(failedIDs, id)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, response.ErrInternal(50041, "task_update_failed", "failed to iterate failed monitor desktop tasks")
|
|
}
|
|
return failedIDs, nil
|
|
}
|
|
|
|
func uniqueInt64s(values []int64) []int64 {
|
|
if len(values) <= 1 {
|
|
return values
|
|
}
|
|
seen := make(map[int64]struct{}, len(values))
|
|
result := make([]int64, 0, len(values))
|
|
for _, value := range values {
|
|
if value == 0 {
|
|
continue
|
|
}
|
|
if _, ok := seen[value]; ok {
|
|
continue
|
|
}
|
|
seen[value] = struct{}{}
|
|
result = append(result, value)
|
|
}
|
|
return result
|
|
}
|
|
|
|
func int64Set(values []int64) map[int64]struct{} {
|
|
result := make(map[int64]struct{}, len(values))
|
|
for _, value := range values {
|
|
if value == 0 {
|
|
continue
|
|
}
|
|
result[value] = struct{}{}
|
|
}
|
|
return result
|
|
}
|
|
|
|
func desktopTaskRecoveryMessage(errorJSON []byte) string {
|
|
const fallback = "desktop monitor task recovered to failed"
|
|
if len(errorJSON) == 0 {
|
|
return fallback
|
|
}
|
|
var payload map[string]any
|
|
if err := json.Unmarshal(errorJSON, &payload); err != nil {
|
|
return fallback
|
|
}
|
|
if message, ok := payload["message"].(string); ok && strings.TrimSpace(message) != "" {
|
|
return strings.TrimSpace(message)
|
|
}
|
|
return fallback
|
|
}
|
|
|
|
func desktopTaskRecoverySelectQuery(mode desktopTaskRecoveryMode) string {
|
|
query := `
|
|
SELECT
|
|
desktop_id,
|
|
workspace_id,
|
|
kind,
|
|
status,
|
|
active_attempt_id,
|
|
attempts,
|
|
publish_submit_started_at IS NOT NULL AS submit_started,
|
|
monitor_task_id
|
|
FROM desktop_tasks
|
|
WHERE target_client_id = $1
|
|
AND workspace_id = $2
|
|
AND status = 'in_progress'
|
|
`
|
|
if mode == desktopTaskRecoveryModeLeaseExpiry {
|
|
query += `
|
|
AND lease_expires_at IS NOT NULL
|
|
AND lease_expires_at < NOW()
|
|
`
|
|
}
|
|
query += `
|
|
FOR UPDATE
|
|
`
|
|
return query
|
|
}
|
|
|
|
func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]byte, string, string, error) {
|
|
reason := strings.TrimSpace(string(mode))
|
|
message := "desktop task was recovered after the client lost the active lease"
|
|
source := "desktop_task_recovery"
|
|
isPublishFinal := kind == "publish_final"
|
|
isPublishUncertain := kind == "publish_uncertain"
|
|
isMonitorFinal := kind == "monitor_final"
|
|
|
|
if isPublishFinal {
|
|
message = fmt.Sprintf("desktop publish task exceeded %d attempts during recovery; task has been marked failed", desktopPublishMaxAttempts)
|
|
} else if isPublishUncertain {
|
|
message = "desktop publish task may have already been submitted to the platform before the lease was lost; kept as unknown for manual reconcile to avoid duplicate publishing"
|
|
} else if isMonitorFinal {
|
|
message = "desktop monitor task lost its lease; task has been marked failed to avoid repeated platform retries"
|
|
} else if kind == "monitor" {
|
|
message = "desktop monitor task lost its lease; task has been re-queued"
|
|
}
|
|
|
|
switch mode {
|
|
case desktopTaskRecoveryModeStartup:
|
|
source = "desktop_client_startup"
|
|
if isPublishFinal {
|
|
message = fmt.Sprintf("desktop client restarted while the publish task was in progress; task exceeded %d attempts and has been marked failed", desktopPublishMaxAttempts)
|
|
} else if isPublishUncertain {
|
|
message = "desktop client restarted while the publish task was mid-submit; kept as unknown for manual reconcile to avoid duplicate publishing"
|
|
} else if isMonitorFinal {
|
|
message = "desktop client restarted while the monitor task was in progress; task has been marked failed after repeated recovery"
|
|
} else if kind == "monitor" {
|
|
message = "desktop client restarted while the monitor task was in progress; task has been re-queued"
|
|
} else {
|
|
message = "desktop client restarted while the publish task was in progress; task has been re-queued for retry"
|
|
}
|
|
case desktopTaskRecoveryModeDisconnect:
|
|
source = "desktop_client_offline"
|
|
if isPublishFinal {
|
|
message = fmt.Sprintf("desktop client went offline while the publish task was in progress; task exceeded %d attempts and has been marked failed", desktopPublishMaxAttempts)
|
|
} else if isPublishUncertain {
|
|
message = "desktop client went offline while the publish task was mid-submit; kept as unknown for manual reconcile to avoid duplicate publishing"
|
|
} else if isMonitorFinal {
|
|
message = "desktop client went offline while the monitor task was in progress; task has been marked failed after repeated recovery"
|
|
} else if kind == "monitor" {
|
|
message = "desktop client went offline while the monitor task was in progress; task has been re-queued"
|
|
} else {
|
|
message = "desktop client went offline while the publish task was in progress; task has been re-queued for retry"
|
|
}
|
|
case desktopTaskRecoveryModeLeaseExpiry:
|
|
source = "desktop_task_lease_expiry"
|
|
if isPublishFinal {
|
|
message = fmt.Sprintf("desktop task lease expired before publish completion; task exceeded %d attempts and has been marked failed", desktopPublishMaxAttempts)
|
|
} else if isPublishUncertain {
|
|
message = "desktop task lease expired while the publish task was mid-submit; kept as unknown for manual reconcile to avoid duplicate publishing"
|
|
} else if isMonitorFinal {
|
|
message = "desktop task lease expired before monitor completion; task has been marked failed"
|
|
} else if kind == "monitor" {
|
|
message = "desktop task lease expired before monitor completion; task has been re-queued"
|
|
} else {
|
|
message = "desktop task lease expired before publish completion; task has been re-queued for retry"
|
|
}
|
|
}
|
|
|
|
fields := map[string]any{
|
|
"reason": reason,
|
|
"message": message,
|
|
"source": source,
|
|
}
|
|
if isPublishUncertain {
|
|
// Surfaced to the desktop UI so a manual retry of a maybe-already-published task asks for
|
|
// explicit confirmation instead of silently re-posting.
|
|
fields["publish_submit_uncertain"] = true
|
|
}
|
|
payload, err := marshalOptionalJSON(fields)
|
|
if err != nil {
|
|
return nil, "", "", err
|
|
}
|
|
return payload, reason, message, nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) dropStaleMonitorTasksForClient(ctx context.Context, client *repository.DesktopClient) error {
|
|
if s == nil || client == nil || s.pool == nil || s.monitoringPool == nil {
|
|
return nil
|
|
}
|
|
|
|
_, currentBusinessDate := monitoringBusinessDayAndDateAt(time.Now())
|
|
errorJSON, err := marshalOptionalJSON(map[string]any{
|
|
"reason": monitoringStaleTaskDropReason,
|
|
"message": monitoringStaleTaskDropError,
|
|
"current_business_date": currentBusinessDate,
|
|
"source": "business_day_rollover",
|
|
})
|
|
if err != nil {
|
|
return response.ErrInternal(50190, "desktop_task_stale_payload_failed", "failed to encode stale monitor desktop task payload")
|
|
}
|
|
|
|
tx, err := s.pool.Begin(ctx)
|
|
if err != nil {
|
|
return response.ErrInternal(50191, "desktop_task_stale_begin_failed", "failed to start stale monitor desktop cleanup")
|
|
}
|
|
defer tx.Rollback(ctx)
|
|
|
|
rows, err := tx.Query(ctx, `
|
|
WITH stale AS (
|
|
SELECT
|
|
desktop_id,
|
|
active_attempt_id
|
|
FROM desktop_tasks
|
|
WHERE target_client_id = $1
|
|
AND kind = 'monitor'
|
|
AND status IN ('queued', 'in_progress', 'unknown')
|
|
AND COALESCE(
|
|
NULLIF(payload ->> 'business_date', ''),
|
|
NULLIF(payload ->> 'businessDate', ''),
|
|
NULLIF(payload ->> 'metric_date', ''),
|
|
NULLIF(payload ->> 'date', '')
|
|
) < $2
|
|
FOR UPDATE
|
|
),
|
|
updated AS (
|
|
UPDATE desktop_tasks AS t
|
|
SET status = 'aborted',
|
|
error = $3,
|
|
active_attempt_id = NULL,
|
|
lease_token_hash = NULL,
|
|
lease_expires_at = NULL,
|
|
interrupted_at = COALESCE(interrupted_at, NOW()),
|
|
interrupt_reason = $4,
|
|
updated_at = NOW()
|
|
FROM stale
|
|
WHERE t.desktop_id = stale.desktop_id
|
|
RETURNING t.desktop_id
|
|
)
|
|
SELECT stale.desktop_id, stale.active_attempt_id
|
|
FROM stale
|
|
INNER JOIN updated ON updated.desktop_id = stale.desktop_id
|
|
`, client.ID, currentBusinessDate, errorJSON, monitoringStaleTaskDropReason)
|
|
if err != nil {
|
|
return response.ErrInternal(50192, "desktop_task_stale_query_failed", "failed to select stale monitor desktop tasks")
|
|
}
|
|
defer rows.Close()
|
|
|
|
repo := repository.NewDesktopTaskRepository(tx)
|
|
droppedTaskIDs := make([]uuid.UUID, 0)
|
|
for rows.Next() {
|
|
var (
|
|
taskID uuid.UUID
|
|
attemptID pgtype.UUID
|
|
)
|
|
if scanErr := rows.Scan(&taskID, &attemptID); scanErr != nil {
|
|
return response.ErrInternal(50193, "desktop_task_stale_scan_failed", "failed to parse stale monitor desktop tasks")
|
|
}
|
|
|
|
droppedTaskIDs = append(droppedTaskIDs, taskID)
|
|
|
|
if !attemptID.Valid {
|
|
continue
|
|
}
|
|
|
|
resolvedAttemptID, convErr := uuid.FromBytes(attemptID.Bytes[:])
|
|
if convErr != nil {
|
|
s.logWarn("desktop stale monitor attempt id decode failed", convErr, zap.String("task_id", taskID.String()))
|
|
continue
|
|
}
|
|
|
|
if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
|
|
TaskID: taskID,
|
|
AttemptID: resolvedAttemptID,
|
|
FinalStatus: literalStringPtr("aborted"),
|
|
Error: errorJSON,
|
|
}); finishErr != nil {
|
|
s.logWarn("desktop stale monitor attempt finish failed", finishErr, zap.String("task_id", taskID.String()))
|
|
}
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return response.ErrInternal(50193, "desktop_task_stale_scan_failed", "failed to iterate stale monitor desktop tasks")
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return response.ErrInternal(50194, "desktop_task_stale_commit_failed", "failed to commit stale monitor desktop cleanup")
|
|
}
|
|
|
|
for _, taskID := range droppedTaskIDs {
|
|
task, getErr := s.repo.GetByDesktopID(ctx, taskID, client.WorkspaceID)
|
|
if getErr != nil {
|
|
s.logWarn("desktop stale monitor reload failed", getErr, zap.String("task_id", taskID.String()))
|
|
continue
|
|
}
|
|
s.publishTaskEvent(ctx, task, "task_canceled")
|
|
}
|
|
|
|
if _, err := s.monitoringPool.Exec(ctx, `
|
|
UPDATE monitoring_collect_tasks
|
|
SET status = 'skipped',
|
|
lease_token_hash = NULL,
|
|
leased_to_executor = NULL,
|
|
leased_at = NULL,
|
|
lease_expires_at = NULL,
|
|
callback_received_at = COALESCE(callback_received_at, NOW()),
|
|
completed_at = COALESCE(completed_at, NOW()),
|
|
skip_reason = $4,
|
|
error_message = COALESCE(NULLIF(error_message, ''), $5),
|
|
updated_at = NOW()
|
|
WHERE tenant_id = $1
|
|
AND collector_type = $2
|
|
AND COALESCE(execution_owner, 'legacy') = 'desktop_tasks'
|
|
AND target_client_id = $3
|
|
AND business_date < $6::date
|
|
AND callback_received_at IS NULL
|
|
AND status IN ('pending', 'leased', 'expired')
|
|
`, client.TenantID, monitoringCollectorType, client.ID, monitoringStaleTaskDropReason, monitoringStaleTaskDropError, currentBusinessDate); err != nil {
|
|
return response.ErrInternal(50041, "task_skip_failed", "failed to drop stale phase2 monitoring tasks")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func activeAttemptID(task *repository.DesktopTask) *uuid.UUID {
|
|
if task == nil || task.ActiveAttemptID == nil {
|
|
return nil
|
|
}
|
|
value := *task.ActiveAttemptID
|
|
return &value
|
|
}
|
|
|
|
func shouldRequeuePreemptedMonitorTask(task *repository.DesktopTask, reason *string) bool {
|
|
if task == nil || task.Kind != "monitor" {
|
|
return false
|
|
}
|
|
return strings.EqualFold(strings.TrimSpace(optionalStringValue(reason)), "collect_now_preempt")
|
|
}
|
|
|
|
func requeuePreemptedMonitorTask(
|
|
ctx context.Context,
|
|
tx pgx.Tx,
|
|
repo repository.DesktopTaskRepository,
|
|
task *repository.DesktopTask,
|
|
) (*repository.DesktopTask, error) {
|
|
if task == nil {
|
|
return nil, nil
|
|
}
|
|
replacementID := uuid.New()
|
|
replacementJobID := uuid.New()
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO desktop_tasks (
|
|
desktop_id,
|
|
job_id,
|
|
tenant_id,
|
|
workspace_id,
|
|
target_account_id,
|
|
target_client_id,
|
|
platform_id,
|
|
kind,
|
|
payload,
|
|
status,
|
|
priority,
|
|
lane,
|
|
lane_weight,
|
|
source,
|
|
scheduler_group_key,
|
|
monitor_task_id,
|
|
supersedes_task_id,
|
|
control_flags,
|
|
interrupt_generation,
|
|
enqueued_at
|
|
)
|
|
VALUES (
|
|
$1,
|
|
$2,
|
|
$3,
|
|
$4,
|
|
$5,
|
|
$6,
|
|
$7,
|
|
'monitor',
|
|
$8,
|
|
'queued',
|
|
50,
|
|
'retry',
|
|
20,
|
|
'retry_recovery',
|
|
$9,
|
|
$10,
|
|
$11,
|
|
jsonb_build_object(
|
|
'interrupt_requested', false,
|
|
'requeued_at', NOW()::text,
|
|
'requeued_from_task_id', $11::text,
|
|
'requeued_reason', 'collect_now_preempt'
|
|
),
|
|
$12,
|
|
NOW()
|
|
)
|
|
`, replacementID, replacementJobID, task.TenantID, task.WorkspaceID, task.TargetAccountID, task.TargetClientID, task.Platform, task.Payload, nullableString(task.SchedulerGroupKey), nullableInt64(task.MonitorTaskID), task.DesktopID, task.InterruptGeneration); err != nil {
|
|
return nil, response.ErrInternal(50111, "desktop_task_requeue_failed", "failed to create replacement retry monitor desktop task")
|
|
}
|
|
|
|
next, err := repo.GetByDesktopID(ctx, replacementID, task.WorkspaceID)
|
|
if err != nil {
|
|
return nil, response.ErrInternal(50112, "desktop_task_lookup_failed", "failed to reload replacement retry monitor desktop task")
|
|
}
|
|
return next, nil
|
|
}
|
|
|
|
func (s *DesktopTaskService) publishTaskEvent(ctx context.Context, task *repository.DesktopTask, eventType string) {
|
|
publishDesktopTaskLifecycleEvent(ctx, s.messaging, s.logger, task, eventType)
|
|
}
|
|
|
|
func (s *DesktopTaskService) afterRecoveredPublishOutcomes(ctx context.Context, outcomes []*desktopPublishSyncOutcome) {
|
|
for _, outcome := range outcomes {
|
|
if outcome == nil {
|
|
continue
|
|
}
|
|
invalidateArticleCaches(ctx, s.cache, outcome.TenantID, &outcome.ArticleID)
|
|
if outcome.ArticleAlias != nil {
|
|
if syncErr := s.syncMonitoringArticleAlias(ctx, outcome.ArticleAlias); syncErr != nil {
|
|
s.logWarn("monitoring article alias sync failed after publish recovery", syncErr, zap.Int64("article_id", outcome.ArticleID))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func reconcileDesktopTaskEventType(_ string, status string) string {
|
|
if status == "retry" {
|
|
return "task_available"
|
|
}
|
|
return "task_reconciled"
|
|
}
|
|
|
|
func (s *DesktopTaskService) logWarn(message string, err error, fields ...zap.Field) {
|
|
if s.logger == nil || err == nil {
|
|
return
|
|
}
|
|
fields = append(fields, zap.Error(err))
|
|
s.logger.Warn(message, fields...)
|
|
}
|