root
e8f48c6d37
Remove standalone lease-recovery, received-inspection, and result-recovery workers from tenant-api (now handled by scheduler process). Add configurable concurrency to ingest and projection-rebuild workers via MonitoringWorkers runtime config. Extract shared runtime types for worker configuration. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
84 lines
2.6 KiB
Go
84 lines
2.6 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
const (
|
|
monitoringLeaseMaxRetryCount = 1
|
|
monitoringLeaseExpiredErrPrefix = "monitoring lease expired before result callback"
|
|
)
|
|
|
|
func expireMonitoringLeasedTasks(ctx context.Context, tx pgx.Tx, tenantID *int64, now time.Time) (int64, error) {
|
|
tag, err := tx.Exec(ctx, `
|
|
UPDATE monitoring_collect_tasks
|
|
SET status = 'expired',
|
|
lease_token_hash = NULL,
|
|
leased_to_executor = NULL,
|
|
leased_at = NULL,
|
|
lease_expires_at = NULL,
|
|
error_message = COALESCE(NULLIF(error_message, ''), $1),
|
|
request_payload_json = (
|
|
CASE
|
|
WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
|
|
THEN '{}'::jsonb
|
|
ELSE request_payload_json
|
|
END
|
|
) || jsonb_build_object(
|
|
'lease_retry_count',
|
|
(
|
|
CASE
|
|
WHEN COALESCE(request_payload_json ->> 'lease_retry_count', '') ~ '^[0-9]+$'
|
|
THEN (request_payload_json ->> 'lease_retry_count')::int
|
|
ELSE 0
|
|
END
|
|
) + 1
|
|
),
|
|
updated_at = NOW()
|
|
WHERE collector_type = $2
|
|
AND status = 'leased'
|
|
AND lease_expires_at IS NOT NULL
|
|
AND lease_expires_at < $3
|
|
AND ($4::bigint IS NULL OR tenant_id = $4)
|
|
`, monitoringLeaseExpiredErrPrefix, monitoringCollectorType, now.UTC(), nullableInt64(tenantID))
|
|
if err != nil {
|
|
return 0, monitoringInternalError(50041, "lease_expire_failed", "failed to expire overdue monitoring lease tasks", err)
|
|
}
|
|
return tag.RowsAffected(), nil
|
|
}
|
|
|
|
func requeueMonitoringExpiredTasks(ctx context.Context, tx pgx.Tx, tenantID int64, businessDate time.Time) (int64, error) {
|
|
tag, err := tx.Exec(ctx, `
|
|
UPDATE monitoring_collect_tasks
|
|
SET status = 'pending',
|
|
planned_at = NOW(),
|
|
callback_received_at = NULL,
|
|
completed_at = NULL,
|
|
skip_reason = NULL,
|
|
error_message = NULL,
|
|
updated_at = NOW()
|
|
WHERE tenant_id = $1
|
|
AND collector_type = $2
|
|
AND business_date = $3::date
|
|
AND status = 'expired'
|
|
AND (
|
|
CASE
|
|
WHEN COALESCE(request_payload_json ->> 'lease_retry_count', '') ~ '^[0-9]+$'
|
|
THEN (request_payload_json ->> 'lease_retry_count')::int
|
|
ELSE 0
|
|
END
|
|
) <= $4
|
|
`, tenantID, monitoringCollectorType, businessDate.Format("2006-01-02"), monitoringLeaseMaxRetryCount)
|
|
if err != nil {
|
|
return 0, monitoringInternalError(50041, "lease_requeue_failed", "failed to recycle expired monitoring lease tasks", err)
|
|
}
|
|
return tag.RowsAffected(), nil
|
|
}
|
|
|
|
func ExpireMonitoringLeasedTasks(ctx context.Context, tx pgx.Tx, tenantID *int64, now time.Time) (int64, error) {
|
|
return expireMonitoringLeasedTasks(ctx, tx, tenantID, now)
|
|
}
|