From e8f48c6d373a2f07547cef392b26eddda9821c21 Mon Sep 17 00:00:00 2001 From: liangxu Date: Wed, 15 Apr 2026 14:19:57 +0800 Subject: [PATCH] refactor(monitoring): consolidate workers and extract to scheduler process 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) --- server/cmd/tenant-api/main.go | 9 +- .../tenant/app/monitoring_callback_service.go | 15 ++ .../internal/tenant/app/monitoring_errors.go | 4 + .../tenant/app/monitoring_lease_recovery.go | 4 + .../app/monitoring_lease_recovery_worker.go | 116 ---------- .../monitoring_projection_rebuild_worker.go | 39 +++- .../app/monitoring_received_inspection.go | 10 + .../monitoring_received_inspection_worker.go | 105 --------- .../app/monitoring_result_ingest_worker.go | 39 +++- .../app/monitoring_result_recovery_worker.go | 204 ------------------ .../app/monitoring_task_result_runtime.go | 22 ++ .../tenant/app/monitoring_worker_runtime.go | 44 ++++ 12 files changed, 161 insertions(+), 450 deletions(-) delete mode 100644 server/internal/tenant/app/monitoring_lease_recovery_worker.go delete mode 100644 server/internal/tenant/app/monitoring_received_inspection_worker.go delete mode 100644 server/internal/tenant/app/monitoring_result_recovery_worker.go create mode 100644 server/internal/tenant/app/monitoring_task_result_runtime.go create mode 100644 server/internal/tenant/app/monitoring_worker_runtime.go diff --git a/server/cmd/tenant-api/main.go b/server/cmd/tenant-api/main.go index 702f7e9..2dd8758 100644 --- a/server/cmd/tenant-api/main.go +++ b/server/cmd/tenant-api/main.go @@ -28,12 +28,11 @@ func main() { workerCtx, workerCancel := context.WithCancel(context.Background()) defer workerCancel() + app.GenerationStreams.Run(workerCtx) + monitoringCallbackService := tenantapp.NewMonitoringCallbackService(app.DB, app.MonitoringDB, app.RabbitMQ, app.LLM, app.Logger) - tenantapp.NewMonitoringResultIngestWorker(monitoringCallbackService, app.RabbitMQ, app.Logger).Start(workerCtx) - tenantapp.NewMonitoringProjectionRebuildWorker(monitoringCallbackService, app.RabbitMQ, app.Logger).Start(workerCtx) - tenantapp.NewMonitoringResultRecoveryWorker(app.MonitoringDB, monitoringCallbackService, app.Logger).Start(workerCtx) - tenantapp.NewMonitoringLeaseRecoveryWorker(app.MonitoringDB, app.Logger).Start(workerCtx) - tenantapp.NewMonitoringReceivedInspectionWorker(app.MonitoringDB, app.Logger).Start(workerCtx) + tenantapp.NewMonitoringResultIngestWorker(monitoringCallbackService, app.RabbitMQ, app.Logger, app.Config.MonitoringWorkers).Start(workerCtx) + tenantapp.NewMonitoringProjectionRebuildWorker(monitoringCallbackService, app.RabbitMQ, app.Logger, app.Config.MonitoringWorkers).Start(workerCtx) transport.RegisterRoutes(app) diff --git a/server/internal/tenant/app/monitoring_callback_service.go b/server/internal/tenant/app/monitoring_callback_service.go index 798e1d7..794a68c 100644 --- a/server/internal/tenant/app/monitoring_callback_service.go +++ b/server/internal/tenant/app/monitoring_callback_service.go @@ -254,6 +254,8 @@ type monitoringTaskResultQueueMetaPatch struct { LastQueuePublishError *string `json:"last_queue_publish_error"` } +type MonitoringTaskResultQueueMetaPatch = monitoringTaskResultQueueMetaPatch + func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, installationID int64, installationToken string, taskID int64, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) { installation, err := s.authenticateInstallation(ctx, installationID, installationToken) if err != nil { @@ -1733,6 +1735,10 @@ func (s *MonitoringCallbackService) publishTaskResultEnvelope(ctx context.Contex return nil } +func (s *MonitoringCallbackService) PublishTaskResultEnvelope(ctx context.Context, payload []byte) error { + return s.publishTaskResultEnvelope(ctx, payload) +} + func (s *MonitoringCallbackService) tryPatchTaskResultQueueMeta(taskID int64, receivedAt time.Time, patch monitoringTaskResultQueueMetaPatch) { if s == nil || s.monitoringPool == nil { return @@ -1766,6 +1772,10 @@ func (s *MonitoringCallbackService) tryPatchTaskResultQueueMeta(taskID int64, re } } +func (s *MonitoringCallbackService) PatchTaskResultQueueMeta(taskID int64, receivedAt time.Time, patch MonitoringTaskResultQueueMetaPatch) { + s.tryPatchTaskResultQueueMeta(taskID, receivedAt, patch) +} + func buildMonitoringRawPayload(req MonitoringTaskResultRequest) ([]byte, []MonitoringSourceItem) { payload := make(map[string]interface{}, len(req.RawResponseJSON)+4) for key, value := range req.RawResponseJSON { @@ -2237,6 +2247,11 @@ func optionalString(value string) *string { return &text } +func ValidateMonitoringTaskResultEnvelope(payload []byte) error { + _, err := decodeMonitoringTaskResultEnvelope(payload) + return err +} + func monitoringStringValue(value *string) string { if value == nil { return "" diff --git a/server/internal/tenant/app/monitoring_errors.go b/server/internal/tenant/app/monitoring_errors.go index cef34e1..67ae3e1 100644 --- a/server/internal/tenant/app/monitoring_errors.go +++ b/server/internal/tenant/app/monitoring_errors.go @@ -7,3 +7,7 @@ func monitoringInternalError(code int, message, detail string, cause error) *res appErr.Cause = cause return appErr } + +func InternalMonitoringError(code int, message, detail string, cause error) *response.AppError { + return monitoringInternalError(code, message, detail, cause) +} diff --git a/server/internal/tenant/app/monitoring_lease_recovery.go b/server/internal/tenant/app/monitoring_lease_recovery.go index 3a2dc99..5d13983 100644 --- a/server/internal/tenant/app/monitoring_lease_recovery.go +++ b/server/internal/tenant/app/monitoring_lease_recovery.go @@ -77,3 +77,7 @@ func requeueMonitoringExpiredTasks(ctx context.Context, tx pgx.Tx, tenantID int6 } 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) +} diff --git a/server/internal/tenant/app/monitoring_lease_recovery_worker.go b/server/internal/tenant/app/monitoring_lease_recovery_worker.go deleted file mode 100644 index 9a0278c..0000000 --- a/server/internal/tenant/app/monitoring_lease_recovery_worker.go +++ /dev/null @@ -1,116 +0,0 @@ -package app - -import ( - "context" - "time" - - "github.com/jackc/pgx/v5/pgxpool" - "go.uber.org/zap" - - "github.com/geo-platform/tenant-api/internal/shared/response" -) - -const ( - defaultMonitoringLeaseRecoveryInterval = 30 * time.Minute - defaultMonitoringLeaseRecoveryTimeout = 30 * time.Second -) - -type MonitoringLeaseRecoveryWorker struct { - monitoringPool *pgxpool.Pool - logger *zap.Logger - interval time.Duration - timeout time.Duration -} - -func NewMonitoringLeaseRecoveryWorker(monitoringPool *pgxpool.Pool, logger *zap.Logger) *MonitoringLeaseRecoveryWorker { - return &MonitoringLeaseRecoveryWorker{ - monitoringPool: monitoringPool, - logger: logger, - interval: defaultMonitoringLeaseRecoveryInterval, - timeout: defaultMonitoringLeaseRecoveryTimeout, - } -} - -func (w *MonitoringLeaseRecoveryWorker) Start(ctx context.Context) { - go w.run(ctx) -} - -func (w *MonitoringLeaseRecoveryWorker) run(ctx context.Context) { - w.runOnce(ctx) - - ticker := time.NewTicker(w.interval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - w.runOnce(ctx) - } - } -} - -func (w *MonitoringLeaseRecoveryWorker) runOnce(parent context.Context) { - if w.monitoringPool == nil { - return - } - - ctx, cancel := context.WithTimeout(parent, w.timeout) - defer cancel() - - tx, err := w.monitoringPool.Begin(ctx) - if err != nil { - w.logger.Warn("monitoring lease recovery begin failed", zap.Error(err)) - return - } - defer tx.Rollback(ctx) - - expiredCount, err := expireMonitoringLeasedTasks(ctx, tx, nil, time.Now().UTC()) - if err != nil { - w.logger.Warn("monitoring lease recovery failed", monitoringWorkerErrorFields(err)...) - return - } - - if err := tx.Commit(ctx); err != nil { - w.logger.Warn("monitoring lease recovery commit failed", zap.Error(err)) - return - } - - if expiredCount > 0 { - w.logger.Info("monitoring lease recovery completed", zap.Int64("expired_task_count", expiredCount)) - } -} - -func normalizeMonitoringWorkerError(err error) error { - if err == nil { - return nil - } - if appErr, ok := err.(*response.AppError); ok { - return appErr - } - return err -} - -func monitoringWorkerErrorFields(err error) []zap.Field { - normalized := normalizeMonitoringWorkerError(err) - if normalized == nil { - return nil - } - - if appErr, ok := normalized.(*response.AppError); ok { - fields := []zap.Field{ - zap.String("error_code", appErr.Message), - zap.Int("app_code", appErr.Code), - } - if appErr.Detail != "" { - fields = append(fields, zap.String("detail", appErr.Detail)) - } - if appErr.Cause != nil { - fields = append(fields, zap.Error(appErr.Cause)) - } - return fields - } - - return []zap.Field{zap.Error(normalized)} -} diff --git a/server/internal/tenant/app/monitoring_projection_rebuild_worker.go b/server/internal/tenant/app/monitoring_projection_rebuild_worker.go index 8c601fb..357c103 100644 --- a/server/internal/tenant/app/monitoring_projection_rebuild_worker.go +++ b/server/internal/tenant/app/monitoring_projection_rebuild_worker.go @@ -10,6 +10,7 @@ import ( amqp "github.com/rabbitmq/amqp091-go" "go.uber.org/zap" + "github.com/geo-platform/tenant-api/internal/shared/config" "github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq" ) @@ -26,36 +27,54 @@ type MonitoringProjectionRebuildWorker struct { logger *zap.Logger retryInterval time.Duration processTimeout time.Duration - consumerName string + consumerPrefix string + concurrency int } -func NewMonitoringProjectionRebuildWorker(service *MonitoringCallbackService, rabbitMQClient *rabbitmq.Client, logger *zap.Logger) *MonitoringProjectionRebuildWorker { +func NewMonitoringProjectionRebuildWorker(service *MonitoringCallbackService, rabbitMQClient *rabbitmq.Client, logger *zap.Logger, cfg config.MonitoringConfig) *MonitoringProjectionRebuildWorker { + concurrency := cfg.ProjectionRebuildConcurrency + if concurrency <= 0 { + concurrency = 1 + } + return &MonitoringProjectionRebuildWorker{ rabbitMQ: rabbitMQClient, service: service, logger: logger, retryInterval: defaultMonitoringProjectionRebuildRetryInterval, processTimeout: defaultMonitoringProjectionRebuildProcessTimeout, - consumerName: fmt.Sprintf("tenant-api-monitor-projection-%d", os.Getpid()), + consumerPrefix: fmt.Sprintf("tenant-api-monitor-projection-%d", os.Getpid()), + concurrency: concurrency, } } func (w *MonitoringProjectionRebuildWorker) Start(ctx context.Context) { - go w.run(ctx) + if w == nil { + return + } + for i := 0; i < w.concurrency; i++ { + consumerName := fmt.Sprintf("%s-%d", w.consumerPrefix, i+1) + go w.run(ctx, consumerName) + } } -func (w *MonitoringProjectionRebuildWorker) run(ctx context.Context) { +func (w *MonitoringProjectionRebuildWorker) run(ctx context.Context, consumerName string) { if w.rabbitMQ == nil || w.service == nil { return } for { - if err := w.consumeOnce(ctx); err != nil && ctx.Err() == nil { + if err := w.consumeOnce(ctx, consumerName); err != nil && ctx.Err() == nil { if w.logger != nil { if errors.Is(err, errMonitoringProjectionRebuildConsumerClosed) { - w.logger.Info("monitoring projection rebuild consumer channel closed, retrying") + w.logger.Info("monitoring projection rebuild consumer channel closed, retrying", + zap.String("consumer", consumerName), + ) } else { - w.logger.Warn("monitoring projection rebuild worker stopped unexpectedly", zap.Error(err)) + w.logger.Warn("monitoring projection rebuild worker stopped unexpectedly", + zap.Error(err), + zap.String("consumer", consumerName), + ) } } } @@ -68,8 +87,8 @@ func (w *MonitoringProjectionRebuildWorker) run(ctx context.Context) { } } -func (w *MonitoringProjectionRebuildWorker) consumeOnce(ctx context.Context) error { - deliveries, ch, err := w.rabbitMQ.ConsumeMonitorProjectionRebuild(w.consumerName) +func (w *MonitoringProjectionRebuildWorker) consumeOnce(ctx context.Context, consumerName string) error { + deliveries, ch, err := w.rabbitMQ.ConsumeMonitorProjectionRebuild(consumerName) if err != nil { return err } diff --git a/server/internal/tenant/app/monitoring_received_inspection.go b/server/internal/tenant/app/monitoring_received_inspection.go index 0a1afe6..f8489b9 100644 --- a/server/internal/tenant/app/monitoring_received_inspection.go +++ b/server/internal/tenant/app/monitoring_received_inspection.go @@ -23,6 +23,8 @@ type monitoringStaleReceivedTask struct { CallbackReceivedAt time.Time } +type MonitoringStaleReceivedTask = monitoringStaleReceivedTask + func markMonitoringStaleReceivedTasks(ctx context.Context, tx pgx.Tx, now time.Time, threshold time.Duration, limit int) ([]monitoringStaleReceivedTask, error) { if limit <= 0 { limit = defaultMonitoringReceivedInspectLimit @@ -111,3 +113,11 @@ func markMonitoringStaleReceivedTasks(ctx context.Context, tx pgx.Tx, now time.T return result, nil } + +func MarkMonitoringStaleReceivedTasks(ctx context.Context, tx pgx.Tx, now time.Time, threshold time.Duration, limit int) ([]MonitoringStaleReceivedTask, error) { + items, err := markMonitoringStaleReceivedTasks(ctx, tx, now, threshold, limit) + if err != nil { + return nil, err + } + return items, nil +} diff --git a/server/internal/tenant/app/monitoring_received_inspection_worker.go b/server/internal/tenant/app/monitoring_received_inspection_worker.go deleted file mode 100644 index 535df2c..0000000 --- a/server/internal/tenant/app/monitoring_received_inspection_worker.go +++ /dev/null @@ -1,105 +0,0 @@ -package app - -import ( - "context" - "time" - - "github.com/jackc/pgx/v5/pgxpool" - "go.uber.org/zap" -) - -const ( - defaultMonitoringReceivedInspectionInterval = 5 * time.Minute - defaultMonitoringReceivedInspectionTimeout = 30 * time.Second -) - -type MonitoringReceivedInspectionWorker struct { - monitoringPool *pgxpool.Pool - logger *zap.Logger - interval time.Duration - timeout time.Duration - threshold time.Duration - limit int -} - -func NewMonitoringReceivedInspectionWorker(monitoringPool *pgxpool.Pool, logger *zap.Logger) *MonitoringReceivedInspectionWorker { - return &MonitoringReceivedInspectionWorker{ - monitoringPool: monitoringPool, - logger: logger, - interval: defaultMonitoringReceivedInspectionInterval, - timeout: defaultMonitoringReceivedInspectionTimeout, - threshold: defaultMonitoringReceivedAlertThreshold, - limit: defaultMonitoringReceivedInspectLimit, - } -} - -func (w *MonitoringReceivedInspectionWorker) Start(ctx context.Context) { - go w.run(ctx) -} - -func (w *MonitoringReceivedInspectionWorker) run(ctx context.Context) { - w.runOnce(ctx) - - ticker := time.NewTicker(w.interval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - w.runOnce(ctx) - } - } -} - -func (w *MonitoringReceivedInspectionWorker) runOnce(parent context.Context) { - if w.monitoringPool == nil { - return - } - - ctx, cancel := context.WithTimeout(parent, w.timeout) - defer cancel() - - tx, err := w.monitoringPool.Begin(ctx) - if err != nil { - w.logger.Warn("monitoring received watchdog begin failed", zap.Error(err)) - return - } - defer tx.Rollback(ctx) - - now := time.Now().UTC() - items, err := markMonitoringStaleReceivedTasks(ctx, tx, now, w.threshold, w.limit) - if err != nil { - w.logger.Warn("monitoring received watchdog failed", monitoringWorkerErrorFields(err)...) - return - } - - if err := tx.Commit(ctx); err != nil { - w.logger.Warn("monitoring received watchdog commit failed", zap.Error(err)) - return - } - - if len(items) == 0 { - return - } - - w.logger.Warn("monitoring received watchdog found overdue tasks", - zap.Int("task_count", len(items)), - zap.Duration("threshold", w.threshold), - ) - - for _, item := range items { - w.logger.Warn("monitoring task stuck in received state", - zap.Int64("task_id", item.ID), - zap.Int64("tenant_id", item.TenantID), - zap.Int64("brand_id", item.BrandID), - zap.Int64("question_id", item.QuestionID), - zap.String("collector_type", item.CollectorType), - zap.String("ai_platform_id", item.AIPlatformID), - zap.String("business_date", item.BusinessDate.Format("2006-01-02")), - zap.Time("callback_received_at", item.CallbackReceivedAt.UTC()), - zap.Duration("received_lag", now.Sub(item.CallbackReceivedAt.UTC())), - ) - } -} diff --git a/server/internal/tenant/app/monitoring_result_ingest_worker.go b/server/internal/tenant/app/monitoring_result_ingest_worker.go index 7876e34..7ccece7 100644 --- a/server/internal/tenant/app/monitoring_result_ingest_worker.go +++ b/server/internal/tenant/app/monitoring_result_ingest_worker.go @@ -10,6 +10,7 @@ import ( amqp "github.com/rabbitmq/amqp091-go" "go.uber.org/zap" + "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" ) @@ -27,36 +28,54 @@ type MonitoringResultIngestWorker struct { logger *zap.Logger retryInterval time.Duration processTimeout time.Duration - consumerName string + consumerPrefix string + concurrency int } -func NewMonitoringResultIngestWorker(service *MonitoringCallbackService, rabbitMQClient *rabbitmq.Client, logger *zap.Logger) *MonitoringResultIngestWorker { +func NewMonitoringResultIngestWorker(service *MonitoringCallbackService, rabbitMQClient *rabbitmq.Client, logger *zap.Logger, cfg config.MonitoringConfig) *MonitoringResultIngestWorker { + concurrency := cfg.ResultIngestConcurrency + if concurrency <= 0 { + concurrency = 1 + } + return &MonitoringResultIngestWorker{ rabbitMQ: rabbitMQClient, service: service, logger: logger, retryInterval: defaultMonitoringResultWorkerRetryInterval, processTimeout: defaultMonitoringResultWorkerProcessTimeout, - consumerName: fmt.Sprintf("tenant-api-monitor-result-%d", os.Getpid()), + consumerPrefix: fmt.Sprintf("tenant-api-monitor-result-%d", os.Getpid()), + concurrency: concurrency, } } func (w *MonitoringResultIngestWorker) Start(ctx context.Context) { - go w.run(ctx) + if w == nil { + return + } + for i := 0; i < w.concurrency; i++ { + consumerName := fmt.Sprintf("%s-%d", w.consumerPrefix, i+1) + go w.run(ctx, consumerName) + } } -func (w *MonitoringResultIngestWorker) run(ctx context.Context) { +func (w *MonitoringResultIngestWorker) run(ctx context.Context, consumerName string) { if w.rabbitMQ == nil || w.service == nil { return } for { - if err := w.consumeOnce(ctx); err != nil && ctx.Err() == nil { + if err := w.consumeOnce(ctx, consumerName); err != nil && ctx.Err() == nil { if w.logger != nil { if errors.Is(err, errMonitoringResultConsumerClosed) { - w.logger.Info("monitoring result consumer channel closed, retrying") + w.logger.Info("monitoring result consumer channel closed, retrying", + zap.String("consumer", consumerName), + ) } else { - w.logger.Warn("monitoring result worker stopped unexpectedly", zap.Error(err)) + w.logger.Warn("monitoring result worker stopped unexpectedly", + zap.Error(err), + zap.String("consumer", consumerName), + ) } } } @@ -69,8 +88,8 @@ func (w *MonitoringResultIngestWorker) run(ctx context.Context) { } } -func (w *MonitoringResultIngestWorker) consumeOnce(ctx context.Context) error { - deliveries, ch, err := w.rabbitMQ.ConsumeMonitorResultIngest(w.consumerName) +func (w *MonitoringResultIngestWorker) consumeOnce(ctx context.Context, consumerName string) error { + deliveries, ch, err := w.rabbitMQ.ConsumeMonitorResultIngest(consumerName) if err != nil { return err } diff --git a/server/internal/tenant/app/monitoring_result_recovery_worker.go b/server/internal/tenant/app/monitoring_result_recovery_worker.go deleted file mode 100644 index ab4c753..0000000 --- a/server/internal/tenant/app/monitoring_result_recovery_worker.go +++ /dev/null @@ -1,204 +0,0 @@ -package app - -import ( - "context" - "encoding/json" - "fmt" - "time" - - "github.com/jackc/pgx/v5" - "github.com/jackc/pgx/v5/pgxpool" - "go.uber.org/zap" -) - -const ( - defaultMonitoringResultRecoveryInterval = 1 * time.Minute - defaultMonitoringResultRecoveryTimeout = 30 * time.Second - defaultMonitoringResultRecoveryLimit = 100 -) - -type MonitoringResultRecoveryWorker struct { - monitoringPool *pgxpool.Pool - service *MonitoringCallbackService - logger *zap.Logger - interval time.Duration - timeout time.Duration - limit int -} - -type monitoringRecoverableTaskResult struct { - ID int64 - CallbackReceivedAt time.Time - RequestPayloadJSON []byte -} - -func NewMonitoringResultRecoveryWorker(monitoringPool *pgxpool.Pool, service *MonitoringCallbackService, logger *zap.Logger) *MonitoringResultRecoveryWorker { - return &MonitoringResultRecoveryWorker{ - monitoringPool: monitoringPool, - service: service, - logger: logger, - interval: defaultMonitoringResultRecoveryInterval, - timeout: defaultMonitoringResultRecoveryTimeout, - limit: defaultMonitoringResultRecoveryLimit, - } -} - -func (w *MonitoringResultRecoveryWorker) Start(ctx context.Context) { - go w.run(ctx) -} - -func (w *MonitoringResultRecoveryWorker) run(ctx context.Context) { - w.runOnce(ctx) - - ticker := time.NewTicker(w.interval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - w.runOnce(ctx) - } - } -} - -func (w *MonitoringResultRecoveryWorker) runOnce(parent context.Context) { - if w.monitoringPool == nil || w.service == nil { - return - } - - ctx, cancel := context.WithTimeout(parent, w.timeout) - defer cancel() - - tx, err := w.monitoringPool.Begin(ctx) - if err != nil { - w.logger.Warn("monitoring result recovery begin failed", zap.Error(err)) - return - } - defer tx.Rollback(ctx) - - items, err := loadMonitoringRecoverableTaskResults(ctx, tx, w.limit) - if err != nil { - w.logger.Warn("monitoring result recovery failed", monitoringWorkerErrorFields(err)...) - return - } - - if err := tx.Commit(ctx); err != nil { - w.logger.Warn("monitoring result recovery commit failed", zap.Error(err)) - return - } - - if len(items) == 0 { - return - } - - republishedCount := 0 - for _, item := range items { - if w.tryRepublish(parent, item) { - republishedCount++ - } - } - - if republishedCount > 0 && w.logger != nil { - w.logger.Info("monitoring result recovery republished tasks", zap.Int("task_count", republishedCount)) - } -} - -func (w *MonitoringResultRecoveryWorker) tryRepublish(parent context.Context, item monitoringRecoverableTaskResult) bool { - if _, err := decodeMonitoringTaskResultEnvelope(item.RequestPayloadJSON); err != nil { - w.service.tryPatchTaskResultQueueMeta(item.ID, item.CallbackReceivedAt, monitoringTaskResultQueueMetaPatch{ - QueueStatus: "invalid_payload", - QueueClaimedAt: nil, - QueueEnqueuedAt: nil, - LastQueuePublishError: optionalString(err.Error()), - }) - if w.logger != nil { - w.logger.Warn("monitoring result recovery skipped invalid payload", zap.Int64("task_id", item.ID), zap.Error(err)) - } - return false - } - - ctx, cancel := context.WithTimeout(parent, monitoringTaskResultQueueIOTimeout) - defer cancel() - - if err := w.service.publishTaskResultEnvelope(ctx, item.RequestPayloadJSON); err != nil { - w.service.tryPatchTaskResultQueueMeta(item.ID, item.CallbackReceivedAt, monitoringTaskResultQueueMetaPatch{ - QueueStatus: "publish_failed", - QueueClaimedAt: nil, - QueueEnqueuedAt: nil, - LastQueuePublishError: optionalString(err.Error()), - }) - if w.logger != nil { - w.logger.Warn("monitoring result recovery republish failed", zap.Int64("task_id", item.ID), zap.Error(err)) - } - return false - } - - enqueuedAt := time.Now().UTC().Round(0) - w.service.tryPatchTaskResultQueueMeta(item.ID, item.CallbackReceivedAt, monitoringTaskResultQueueMetaPatch{ - QueueStatus: "queued", - QueueClaimedAt: nil, - QueueEnqueuedAt: &enqueuedAt, - LastQueuePublishError: nil, - }) - - return true -} - -func loadMonitoringRecoverableTaskResults(ctx context.Context, tx pgx.Tx, limit int) ([]monitoringRecoverableTaskResult, error) { - if limit <= 0 { - limit = defaultMonitoringResultRecoveryLimit - } - - rows, err := tx.Query(ctx, ` - SELECT - id, - callback_received_at, - request_payload_json - FROM monitoring_collect_tasks - WHERE collector_type = $1 - AND status = 'received' - AND callback_received_at IS NOT NULL - AND request_payload_json IS NOT NULL - AND jsonb_typeof(request_payload_json) = 'object' - AND COALESCE(NULLIF(request_payload_json ->> 'queue_status', ''), 'pending') IN ('pending', 'publish_failed') - ORDER BY callback_received_at ASC, id ASC - FOR UPDATE SKIP LOCKED - LIMIT $2 - `, monitoringCollectorType, limit) - if err != nil { - return nil, monitoringInternalError(50041, "result_recovery_query_failed", "failed to load recoverable monitoring results", err) - } - defer rows.Close() - - result := make([]monitoringRecoverableTaskResult, 0, limit) - for rows.Next() { - var item monitoringRecoverableTaskResult - if scanErr := rows.Scan(&item.ID, &item.CallbackReceivedAt, &item.RequestPayloadJSON); scanErr != nil { - return nil, monitoringInternalError(50041, "result_recovery_scan_failed", "failed to parse recoverable monitoring results", scanErr) - } - result = append(result, item) - } - if err := rows.Err(); err != nil { - return nil, monitoringInternalError(50041, "result_recovery_scan_failed", "failed to iterate recoverable monitoring results", err) - } - - return result, nil -} - -func decodeMonitoringTaskResultEnvelope(payload []byte) (*monitoringTaskResultEnvelope, error) { - if len(payload) == 0 { - return nil, fmt.Errorf("monitoring task result payload is empty") - } - - var envelope monitoringTaskResultEnvelope - if err := json.Unmarshal(payload, &envelope); err != nil { - return nil, err - } - if envelope.TaskID <= 0 { - return nil, fmt.Errorf("monitoring task result payload missing task_id") - } - - return &envelope, nil -} diff --git a/server/internal/tenant/app/monitoring_task_result_runtime.go b/server/internal/tenant/app/monitoring_task_result_runtime.go new file mode 100644 index 0000000..b7a6f31 --- /dev/null +++ b/server/internal/tenant/app/monitoring_task_result_runtime.go @@ -0,0 +1,22 @@ +package app + +import ( + "encoding/json" + "fmt" +) + +func decodeMonitoringTaskResultEnvelope(payload []byte) (*monitoringTaskResultEnvelope, error) { + if len(payload) == 0 { + return nil, fmt.Errorf("monitoring task result payload is empty") + } + + var envelope monitoringTaskResultEnvelope + if err := json.Unmarshal(payload, &envelope); err != nil { + return nil, err + } + if envelope.TaskID <= 0 { + return nil, fmt.Errorf("monitoring task result payload missing task_id") + } + + return &envelope, nil +} diff --git a/server/internal/tenant/app/monitoring_worker_runtime.go b/server/internal/tenant/app/monitoring_worker_runtime.go new file mode 100644 index 0000000..8cb8eb7 --- /dev/null +++ b/server/internal/tenant/app/monitoring_worker_runtime.go @@ -0,0 +1,44 @@ +package app + +import ( + "go.uber.org/zap" + + "github.com/geo-platform/tenant-api/internal/shared/response" +) + +func normalizeMonitoringWorkerError(err error) error { + if err == nil { + return nil + } + if appErr, ok := err.(*response.AppError); ok { + return appErr + } + return err +} + +func monitoringWorkerErrorFields(err error) []zap.Field { + normalized := normalizeMonitoringWorkerError(err) + if normalized == nil { + return nil + } + + if appErr, ok := normalized.(*response.AppError); ok { + fields := []zap.Field{ + zap.String("error_code", appErr.Message), + zap.Int("app_code", appErr.Code), + } + if appErr.Detail != "" { + fields = append(fields, zap.String("detail", appErr.Detail)) + } + if appErr.Cause != nil { + fields = append(fields, zap.Error(appErr.Cause)) + } + return fields + } + + return []zap.Field{zap.Error(normalized)} +} + +func MonitoringWorkerErrorFields(err error) []zap.Field { + return monitoringWorkerErrorFields(err) +}