root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix monitoring online dispatch retries

Browse Source
This commit is contained in:
liangxu
2026-06-27 16:17:02 +08:00
parent 940ac23493
commit beb290a5a7
11 changed files with 555 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/monitoring_callback_service.go
+137 -33
View File
@@ -40,7 +40,9 @@ const (
// reports skip writes and only refresh the row at this interval so dashboards
// do not depend on per-heartbeat database churn.
monitoringHeartbeatAccessSnapshotRefreshAfter = 10 * time.Minute
monitoringRuntimeUnknownMaxDispatchAttempts = 3
monitoringRuntimeUnknownMaxDispatchAttempts = 2
monitoringAutomaticFailureMaxAttempts = 2
monitoringAutomaticFailureRetryAfter = 2 * time.Minute
)
var monitoringCitationMarkerPattern = regexp.MustCompile(`(?i)\[\s*citation\s*:\s*\d+\s*\]`)
@@ -240,27 +242,28 @@ type monitoringHeartbeatSnapshotDecision struct {
}
type monitoringCollectTask struct {
ID int64
TenantID int64
BrandID int64
QuestionID int64
QuestionHash []byte
AIPlatformID string
CollectorType string
RunMode string
BusinessDate time.Time
Status string
ExecutionOwner string
LeaseTokenHash sql.NullString
LeasedToExecutor sql.NullString
LeaseExpiresAt sql.NullTime
CallbackReceivedAt sql.NullTime
CompletedAt sql.NullTime
SkipReason sql.NullString
TriggerSource string
QuestionText string
TargetClientID sql.NullString
DispatchAttempts int
ID int64
TenantID int64
BrandID int64
QuestionID int64
QuestionHash []byte
AIPlatformID string
CollectorType string
RunMode string
BusinessDate time.Time
Status string
ExecutionOwner string
LeaseTokenHash sql.NullString
LeasedToExecutor sql.NullString
LeaseExpiresAt sql.NullTime
CallbackReceivedAt sql.NullTime
CompletedAt sql.NullTime
SkipReason sql.NullString
TriggerSource string
QuestionText string
TargetClientID sql.NullString
DispatchAttempts int
AutomaticFailureCount int
}
type monitoringDesktopExecutionLease struct {
@@ -1062,6 +1065,32 @@ type monitoringSkipOutcome struct {
RetryAfter time.Duration
}
type monitoringFailedResultOutcome struct {
TaskStatus string
Requeue bool
RetryAfter time.Duration
}
func monitoringFailedResultOutcomeForTask(task *monitoringCollectTask, req MonitoringTaskResultRequest) monitoringFailedResultOutcome {
if task == nil {
return monitoringFailedResultOutcome{TaskStatus: "failed"}
}
if !strings.EqualFold(strings.TrimSpace(task.TriggerSource), "automatic") {
return monitoringFailedResultOutcome{TaskStatus: "failed"}
}
if monitoringTaskFailureDispositionFromRequest(req).NonRetryable {
return monitoringFailedResultOutcome{TaskStatus: "failed"}
}
if task.AutomaticFailureCount < monitoringAutomaticFailureMaxAttempts-1 {
return monitoringFailedResultOutcome{
TaskStatus: "pending",
Requeue: true,
RetryAfter: monitoringAutomaticFailureRetryAfter,
}
}
return monitoringFailedResultOutcome{TaskStatus: "failed"}
}
func monitoringSkipOutcomeForReason(skipReason string, dispatchAttempts int, triggerSource string) monitoringSkipOutcome {
switch strings.TrimSpace(skipReason) {
case "runtime_unknown":
@@ -1562,6 +1591,11 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI
trigger_source,
COALESCE(dispatch_attempts, 0) AS dispatch_attempts,
COALESCE(target_client_id::text, '') AS target_client_id,
CASE
WHEN COALESCE(t.request_payload_json ->> 'automatic_failure_count', '') ~ '^[0-9]+$'
THEN (t.request_payload_json ->> 'automatic_failure_count')::int
ELSE 0
END AS automatic_failure_count,
COALESCE(NULLIF(t.question_text_snapshot, ''), (
SELECT question_text_snapshot
FROM monitoring_question_config_snapshots
@@ -1600,6 +1634,7 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI
&item.TriggerSource,
&item.DispatchAttempts,
&item.TargetClientID,
&item.AutomaticFailureCount,
&item.QuestionText,
); err != nil {
if err == pgx.ErrNoRows {
@@ -1635,6 +1670,11 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas
trigger_source,
COALESCE(dispatch_attempts, 0) AS dispatch_attempts,
COALESCE(target_client_id::text, '') AS target_client_id,
CASE
WHEN COALESCE(t.request_payload_json ->> 'automatic_failure_count', '') ~ '^[0-9]+$'
THEN (t.request_payload_json ->> 'automatic_failure_count')::int
ELSE 0
END AS automatic_failure_count,
COALESCE(NULLIF(t.question_text_snapshot, ''), (
SELECT question_text_snapshot
FROM monitoring_question_config_snapshots
@@ -1672,6 +1712,7 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas
&item.TriggerSource,
&item.DispatchAttempts,
&item.TargetClientID,
&item.AutomaticFailureCount,
&item.QuestionText,
); err != nil {
if err == pgx.ErrNoRows {
@@ -2312,12 +2353,20 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task
}
defer tx.Rollback(ctx)
runID, err := s.upsertRun(ctx, tx, task, req, runStatus, completedAt, rawPayload)
if err != nil {
return nil, err
var failedOutcome monitoringFailedResultOutcome
if runStatus == "failed" {
failedOutcome = monitoringFailedResultOutcomeForTask(task, req)
}
if runStatus == "succeeded" {
var runID int64
if !failedOutcome.Requeue {
runID, err = s.upsertRun(ctx, tx, task, req, runStatus, completedAt, rawPayload)
if err != nil {
return nil, err
}
}
if runStatus == "succeeded" && !failedOutcome.Requeue {
if err := s.upsertParseResult(ctx, tx, task, runID, parseReq, parseVersion); err != nil {
return nil, err
}
@@ -2325,10 +2374,12 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task
citationSourceCount := 0
contentCitationCount := 0
if err := s.replaceCitationFacts(ctx, tx, task.TenantID, runID); err != nil {
return nil, err
if !failedOutcome.Requeue {
if err := s.replaceCitationFacts(ctx, tx, task.TenantID, runID); err != nil {
return nil, err
}
}
if runStatus == "succeeded" {
if runStatus == "succeeded" && !failedOutcome.Requeue {
facts, buildErr := s.buildCitationFacts(ctx, tx, task.TenantID, task.AIPlatformID, sourceInputs)
if buildErr != nil {
return nil, buildErr
@@ -2346,12 +2397,16 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task
taskStatus := "completed"
if runStatus == "failed" {
taskStatus = "failed"
taskStatus = failedOutcome.TaskStatus
}
if err := s.updateCollectTaskStatus(ctx, tx, task.ID, taskStatus, completedAt, req.ErrorMessage); err != nil {
if failedOutcome.Requeue {
if err := s.requeueCollectTaskAfterFailedResult(ctx, tx, task.ID, failedOutcome.RetryAfter, req.ErrorMessage); err != nil {
return nil, err
}
} else if err := s.updateCollectTaskStatus(ctx, tx, task.ID, taskStatus, completedAt, req.ErrorMessage); err != nil {
return nil, err
}
if runStatus == "failed" {
if runStatus == "failed" && !failedOutcome.Requeue {
if _, err := s.failRemainingMonitoringTasksForAuthorizationFailure(ctx, tx, task, req, completedAt); err != nil {
return nil, err
}
@@ -2365,9 +2420,13 @@ func (s *MonitoringCallbackService) processTaskResult(ctx context.Context, task
return nil, response.ErrInternal(50041, "commit_failed", "failed to commit monitoring result ingest")
}
var runIDPtr *int64
if !failedOutcome.Requeue {
runIDPtr = &runID
}
return &MonitoringTaskResultResponse{
TaskID: task.ID,
RunID: &runID,
RunID: runIDPtr,
TaskStatus: taskStatus,
CitationSourceCount: citationSourceCount,
ContentCitationCount: contentCitationCount,
@@ -3113,6 +3172,51 @@ func (s *MonitoringCallbackService) updateCollectTaskStatus(ctx context.Context,
return nil
}
func (s *MonitoringCallbackService) requeueCollectTaskAfterFailedResult(ctx context.Context, tx pgx.Tx, taskID int64, retryAfter time.Duration, errorMessage *string) error {
if retryAfter <= 0 {
retryAfter = monitoringAutomaticFailureRetryAfter
}
if _, err := tx.Exec(ctx, `
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,
last_dispatched_at = NULL,
dispatch_after = NOW() + $2::interval,
error_message = $3,
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(
'automatic_failure_count',
(
CASE
WHEN COALESCE(request_payload_json ->> 'automatic_failure_count', '') ~ '^[0-9]+$'
THEN (request_payload_json ->> 'automatic_failure_count')::int
ELSE 0
END
) + 1,
'automatic_failure_retry_after_seconds',
$4::int,
'automatic_failure_last_requeued_at',
to_jsonb(NOW())
),
updated_at = NOW()
WHERE id = $1
`, taskID, formatPgInterval(retryAfter), nullableString(errorMessage), int(retryAfter.Seconds())); err != nil {
return response.ErrInternal(50041, "task_update_failed", "failed to requeue monitoring task after failed result")
}
return nil
}
func (s *MonitoringCallbackService) failRemainingMonitoringTasksForAuthorizationFailure(
ctx context.Context,
tx pgx.Tx,