Files
geo/server/internal/tenant/transport/desktop_task_handler.go
T
root fa51a3455f
Desktop Client Build / Resolve Build Metadata (push) Successful in 41s
Backend CI / Backend (push) Failing after 7m14s
Desktop Client Build / Build Desktop Client (push) Successful in 23m46s
Desktop Client Build / Publish Client Artifacts to NAS (push) Successful in 56s
fix(publish): prevent stuck publish queue and duplicate posting
The desktop publish queue could stall for a long time and end users assumed
the software was broken. Root cause: a hung adapter held its execution slot
with no wall-clock timeout while auto-renewing its lease forever, so the
server never reclaimed it and every queued task behind it stayed 等待发布.
Auto-recovery also risked silently re-posting a non-idempotent article.

Client (Electron):
- per-task wall-clock deadline + abort; progress-gated lease renewal that
  stops and aborts a stalled task instead of renewing it forever
- decouple the concurrency cap from CDP-induced CPU/memory pressure
  (admission gate instead of self-throttling collapse); 15s watchdog pump
- all adapter network I/O now has fetch timeouts and honors context.signal;
  bounded image-upload concurrency with per-image timeout
- surface live adapter progress, elapsed time, queue position and a
  working-vs-queued distinction in the publish view

Server (tenant-api):
- publish lease-recovery worker (every 3m) + supporting index: reclaim
  expired in_progress publish leases, requeue (<3 attempts) or terminal-fail
- 3-minute lease TTL with client-presence-gated extension; max 3 attempts

Idempotency (production-grade core):
- durable publish_submit_started_at marker, set before the irreversible
  platform submit POST; recovery and abort route a maybe-submitted task to
  unknown (manual reconcile, kept in the dedup set) instead of re-posting
- desktop UI requires explicit confirmation before retrying a possibly-
  already-published task

Verified: go build/vet/test (incl. resolvePublishRecoveryOutcome), vue-tsc,
vitest 141/141, gofmt all clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-30 19:52:17 +08:00

279 lines
7.5 KiB
Go

package transport
import (
"errors"
"io"
"strconv"
"strings"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/geo-platform/tenant-api/internal/bootstrap"
"github.com/geo-platform/tenant-api/internal/shared/auth"
"github.com/geo-platform/tenant-api/internal/shared/response"
"github.com/geo-platform/tenant-api/internal/tenant/app"
)
type DesktopTaskHandler struct {
svc *app.DesktopTaskService
publishSvc *app.PublishJobService
}
func NewDesktopTaskHandler(a *bootstrap.App) *DesktopTaskHandler {
return &DesktopTaskHandler{
svc: app.NewDesktopTaskServiceWithConfig(
a.DB,
a.MonitoringDB,
a.RabbitMQ,
a.Logger,
a.ConfigStore,
).WithCache(a.Cache).WithRedis(a.Redis),
publishSvc: app.NewPublishJobServiceWithConfig(
a.DB,
a.RabbitMQ,
a.Redis,
a.Logger,
a.ConfigStore,
).WithCache(a.Cache),
}
}
func (h *DesktopTaskHandler) Lease(c *gin.Context) {
var req app.LeaseDesktopTaskRequest
if err := c.ShouldBindJSON(&req); err != nil && !errors.Is(err, io.EOF) {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
return
}
var routeTaskID *uuid.UUID
if rawID := c.Param("id"); rawID != "" {
parsed, err := uuid.Parse(rawID)
if err != nil {
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
return
}
routeTaskID = &parsed
}
data, err := h.svc.Lease(c.Request.Context(), MustDesktopClient(c.Request.Context()), req, routeTaskID)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopTaskHandler) ListPublish(c *gin.Context) {
req, ok := parseListPublishTasksRequest(c)
if !ok {
return
}
data, err := h.svc.ListPublishTasks(c.Request.Context(), MustDesktopClient(c.Request.Context()), req)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopTaskHandler) TenantListPublish(c *gin.Context) {
req, ok := parseListPublishTasksRequest(c)
if !ok {
return
}
data, err := h.svc.ListTenantPublishTasks(c.Request.Context(), auth.MustActor(c.Request.Context()), req)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopTaskHandler) Extend(c *gin.Context) {
desktopID, err := uuid.Parse(c.Param("id"))
if err != nil {
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
return
}
var req app.ExtendDesktopTaskRequest
if err := c.ShouldBindJSON(&req); err != nil {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
return
}
data, err := h.svc.Extend(c.Request.Context(), MustDesktopClient(c.Request.Context()), desktopID, req)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopTaskHandler) MarkPublishSubmitStarted(c *gin.Context) {
taskID, err := uuid.Parse(c.Param("id"))
if err != nil {
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
return
}
var req app.MarkPublishSubmitStartedRequest
if err := c.ShouldBindJSON(&req); err != nil {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
return
}
if err := h.svc.MarkPublishSubmitStarted(c.Request.Context(), MustDesktopClient(c.Request.Context()), taskID, req.LeaseToken); err != nil {
response.Error(c, err)
return
}
response.Success(c, gin.H{"ok": true})
}
func (h *DesktopTaskHandler) Result(c *gin.Context) {
desktopID, err := uuid.Parse(c.Param("id"))
if err != nil {
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
return
}
var req app.CompleteDesktopTaskRequest
if err := c.ShouldBindJSON(&req); err != nil {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
return
}
data, err := h.svc.Complete(c.Request.Context(), MustDesktopClient(c.Request.Context()), desktopID, req)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopTaskHandler) RetryPublish(c *gin.Context) {
desktopID, err := uuid.Parse(c.Param("id"))
if err != nil {
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
return
}
data, err := h.publishSvc.RetryByDesktopTask(c.Request.Context(), MustDesktopClient(c.Request.Context()), desktopID)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopTaskHandler) TenantRetryPublish(c *gin.Context) {
desktopID, err := uuid.Parse(c.Param("id"))
if err != nil {
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
return
}
data, err := h.publishSvc.RetryTenantDesktopTask(c.Request.Context(), auth.MustActor(c.Request.Context()), desktopID)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopTaskHandler) Cancel(c *gin.Context) {
desktopID, err := uuid.Parse(c.Param("id"))
if err != nil {
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
return
}
var req app.CancelDesktopTaskRequest
if err := c.ShouldBindJSON(&req); err != nil && !errors.Is(err, io.EOF) {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
return
}
data, err := h.svc.Cancel(c.Request.Context(), MustDesktopClient(c.Request.Context()), desktopID, req)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func parseListPublishTasksRequest(c *gin.Context) (app.ListPublishTasksRequest, bool) {
req := app.ListPublishTasksRequest{
Page: 1,
PageSize: 10,
Title: strings.TrimSpace(c.Query("title")),
}
if rawPage := c.Query("page"); rawPage != "" {
parsed, err := strconv.Atoi(rawPage)
if err != nil || parsed <= 0 {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", "page must be a positive integer"))
return req, false
}
req.Page = parsed
}
rawPageSize := c.Query("page_size")
if rawPageSize == "" {
rawPageSize = c.Query("limit")
}
if rawPageSize != "" {
parsed, err := strconv.Atoi(rawPageSize)
if err != nil || parsed <= 0 {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", "page_size must be a positive integer"))
return req, false
}
req.PageSize = parsed
}
return req, true
}
func (h *DesktopTaskHandler) TenantCancel(c *gin.Context) {
desktopID, err := uuid.Parse(c.Param("id"))
if err != nil {
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
return
}
var req app.CancelDesktopTaskRequest
if err := c.ShouldBindJSON(&req); err != nil && !errors.Is(err, io.EOF) {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
return
}
data, err := h.svc.TenantCancel(c.Request.Context(), auth.MustActor(c.Request.Context()), desktopID, req)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopTaskHandler) Reconcile(c *gin.Context) {
desktopID, err := uuid.Parse(c.Param("id"))
if err != nil {
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
return
}
var req app.ReconcileDesktopTaskRequest
if err := c.ShouldBindJSON(&req); err != nil {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
return
}
data, err := h.svc.Reconcile(c.Request.Context(), auth.MustActor(c.Request.Context()), desktopID, req)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}