[fix][backend]fix backfill count (#289)

* fix backfill count

* add log

* add debug log

* no lock

* fix backfill

* add lock

* delete debug log

* go fmt

* add comment

Author: zero3233-bd

backend/modules/observability/domain/task/service/taskexe/processor/auto_evaluate.go

@@ -191,7 +191,7 @@ func (p *AutoEvaluteProcessor) OnCreateTaskChange(ctx context.Context, currentTa
 			case task.TimeUnitWeek:
 				runEndAt = runStartAt + (currentTask.Sampler.CycleInterval)*7*24*time.Hour.Milliseconds()
 			default:
-				runEndAt = runStartAt + (currentTask.Sampler.CycleInterval)*10*time.Minute.Milliseconds()
+				runEndAt = runStartAt + (currentTask.Sampler.CycleInterval)*24*time.Hour.Milliseconds()
 			}
 		}
 		err = p.OnCreateTaskRunChange(ctx, taskexe.OnCreateTaskRunChangeReq{

backend/modules/observability/domain/task/service/taskexe/tracehub/backfill.go

@@ -29,6 +29,7 @@ const (
 	pageSize                = 500
 	backfillLockKeyTemplate = "observability:tracehub:backfill:%d"
 	backfillLockMaxHold     = 24 * time.Hour
+	backfillLockTTL         = 3 * time.Minute
 )
 
 // 定时任务+锁
@@ -43,7 +44,7 @@ func (h *TraceHubServiceImpl) BackFill(ctx context.Context, event *entity.BackFi
 	)
 	if h.locker != nil && event != nil {
 		lockKey = fmt.Sprintf(backfillLockKeyTemplate, event.TaskID)
-		locked, lockCtx, cancel, lockErr := h.locker.LockWithRenew(ctx, lockKey, transformTaskStatusLockTTL, backfillLockMaxHold)
+		locked, lockCtx, cancel, lockErr := h.locker.LockWithRenew(ctx, lockKey, backfillLockTTL, backfillLockMaxHold)
 		if lockErr != nil {
 			logs.CtxError(ctx, "backfill acquire lock failed", "task_id", event.TaskID, "err", lockErr)
 			return lockErr
@@ -311,7 +312,7 @@ func (h *TraceHubServiceImpl) fetchAndSendSpans(ctx context.Context, listParam *
 			logs.CtxInfo(ctx, "completed listing spans, total_count=%d, task_id=%d", totalCount, sub.t.GetID())
 			break
 		}
-
+		listParam.PageToken = result.PageToken
 		pageToken = result.PageToken
 	}
 
@@ -417,7 +418,7 @@ func (h *TraceHubServiceImpl) applySampling(spans []*loop_span.Span, sub *spanSu
 // processSpansForBackfill handles spans for backfill
 func (h *TraceHubServiceImpl) processSpansForBackfill(ctx context.Context, spans []*loop_span.Span, sub *spanSubscriber) error {
 	// Batch processing spans for efficiency
-	const batchSize = 100
+	const batchSize = 50
 
 	for i := 0; i < len(spans); i += batchSize {
 		end := i + batchSize
@@ -432,6 +433,8 @@ func (h *TraceHubServiceImpl) processSpansForBackfill(ctx context.Context, spans
 			// Continue with the next batch without stopping due to a single failure
 			continue
 		}
+		// ml_flow rate-limited: 50/5s
+		time.Sleep(5 * time.Second)
 	}
 
 	return nil