diff --git a/internal/dao/sync_task.go b/internal/dao/sync_task.go index c3f8704f8f..ae0c7f30e7 100644 --- a/internal/dao/sync_task.go +++ b/internal/dao/sync_task.go @@ -63,11 +63,6 @@ type SyncTaskContext struct { Knowledgebase entity.Knowledgebase } -// ConnectorSource returns the connector source name. -func (c SyncTaskContext) ConnectorSource() string { - return c.Connector.Source -} - // SyncTaskDAO reads and updates sync_logs tasks. type SyncTaskDAO struct { db *gorm.DB @@ -101,17 +96,36 @@ type ScheduledSyncTask struct { ConnectorConfig entity.JSONMap } -// ListScheduledTasks returns scheduled tasks with connector scheduling settings. -func (d *SyncTaskDAO) ListScheduledTasks(ctx context.Context, limit int) ([]ScheduledSyncTask, error) { +// ScheduledSyncTaskCursor identifies the last row from a scheduled task page. +type ScheduledSyncTaskCursor struct { + UpdateTime int64 + ID string +} + +// Cursor returns the keyset cursor for the task. +func (t ScheduledSyncTask) Cursor() ScheduledSyncTaskCursor { + updateTime := int64(0) + if t.UpdateTime != nil { + updateTime = *t.UpdateTime + } + return ScheduledSyncTaskCursor{UpdateTime: updateTime, ID: t.ID} +} + +// ListScheduledTasks returns one page of scheduled tasks with connector scheduling settings. +func (d *SyncTaskDAO) ListScheduledTasks(ctx context.Context, limit int, cursor *ScheduledSyncTaskCursor) ([]ScheduledSyncTask, error) { var rows []dueSyncTaskRow - if err := d.db.WithContext(ctx). + query := d.db.WithContext(ctx). Model(&entity.SyncLogs{}). Select("sync_logs.*, connector.refresh_freq AS connector_refresh_freq, connector.prune_freq AS connector_prune_freq, connector.config AS connector_config"). Joins("JOIN connector ON sync_logs.connector_id = connector.id"). Joins("JOIN connector2kb ON sync_logs.connector_id = connector2kb.connector_id AND sync_logs.kb_id = connector2kb.kb_id"). Joins("JOIN knowledgebase ON sync_logs.kb_id = knowledgebase.id"). - Where("sync_logs.status = ? AND connector.status = ? AND sync_logs.task_type IN ?", SyncStatusSchedule, SyncStatusSchedule, []string{TaskTypeSync, TaskTypePrune}). - Order("sync_logs.update_time DESC"). + Where("sync_logs.status = ? AND connector.status = ? AND sync_logs.task_type IN ?", SyncStatusSchedule, SyncStatusSchedule, []string{TaskTypeSync, TaskTypePrune}) + if cursor != nil { + query = query.Where("COALESCE(sync_logs.update_time, 0) < ? OR (COALESCE(sync_logs.update_time, 0) = ? AND sync_logs.id < ?)", cursor.UpdateTime, cursor.UpdateTime, cursor.ID) + } + if err := query. + Order("COALESCE(sync_logs.update_time, 0) DESC, sync_logs.id DESC"). Limit(limit). Scan(&rows).Error; err != nil { return nil, err @@ -371,7 +385,7 @@ func (d *SyncTaskDAO) CompletePruneTask(ctx context.Context, taskContext SyncTas if err := tx.Model(&entity.Connector{}).Where("id = ?", taskContext.Connector.ID).Update("status", SyncStatusDone).Error; err != nil { return err } - if !syncConnectorConfigBool(taskContext.Connector.Config, "sync_deleted_files") { + if !utility.ConfigBool(taskContext.Connector.Config, "sync_deleted_files") { return nil } var err error @@ -381,56 +395,6 @@ func (d *SyncTaskDAO) CompletePruneTask(ctx context.Context, taskContext SyncTas return nextTaskID, err } -// RecoverStaleRunning restores timed-out running tasks to schedule. -func (d *SyncTaskDAO) RecoverStaleRunning(ctx context.Context, now time.Time) (int64, error) { - type staleRunningTaskRow struct { - ID string `gorm:"column:id"` - ConnectorID string `gorm:"column:connector_id"` - TimeStarted *time.Time `gorm:"column:time_started"` - ConnectorTimeoutSecs int64 `gorm:"column:connector_timeout_secs"` - } - var rows []staleRunningTaskRow - if err := d.db.WithContext(ctx). - Model(&entity.SyncLogs{}). - Select("sync_logs.id, sync_logs.connector_id, sync_logs.time_started, connector.timeout_secs AS connector_timeout_secs"). - Joins("JOIN connector ON sync_logs.connector_id = connector.id"). - Where("sync_logs.status = ?", SyncStatusRunning). - Scan(&rows).Error; err != nil { - return 0, err - } - var recovered int64 - connectorIDs := map[string]struct{}{} - - for _, row := range rows { - if row.TimeStarted == nil { - continue - } - timeout := time.Duration(row.ConnectorTimeoutSecs) * time.Second - if timeout <= 0 { - timeout = time.Hour - } - if row.TimeStarted.Add(timeout).After(now) { - continue - } - if err := d.RescheduleClaimed(ctx, row.ID); err != nil { - return recovered, err - } - recovered++ - connectorIDs[row.ConnectorID] = struct{}{} - } - - if recovered == 0 { - return 0, nil - } - ids := make([]string, 0, len(connectorIDs)) - - for connectorID := range connectorIDs { - ids = append(ids, connectorID) - } - - return recovered, d.db.WithContext(ctx).Model(&entity.Connector{}).Where("id IN ? AND status = ?", ids, SyncStatusRunning).Update("status", SyncStatusSchedule).Error -} - // RecoverRunning restores running sync tasks during syncer startup. func (d *SyncTaskDAO) RecoverRunning(ctx context.Context) (int64, error) { type runningTaskRow struct { @@ -524,19 +488,3 @@ func createScheduledTask(ctx context.Context, tx *gorm.DB, connectorID, kbID, ta TotalDocsIndexed: totalDocsIndexed, }).Error } - -// syncConnectorConfigBool reads a Python JSON bool/string flag. -func syncConnectorConfigBool(config map[string]any, key string) bool { - value, ok := config[key] - if !ok { - return false - } - switch typed := value.(type) { - case bool: - return typed - case string: - return typed == "1" || typed == "true" || typed == "TRUE" - default: - return false - } -} diff --git a/internal/service/sync_task_service.go b/internal/service/sync_task_service.go index 6e975997cf..7e23b40394 100644 --- a/internal/service/sync_task_service.go +++ b/internal/service/sync_task_service.go @@ -24,19 +24,9 @@ import ( "time" ) -const ( - // TaskTypeSync is the Python-compatible SYNC task type. - TaskTypeSync = dao.TaskTypeSync - // TaskTypePrune is the Python-compatible PRUNE task type. - TaskTypePrune = dao.TaskTypePrune -) - // SyncTaskContext contains the database rows needed to execute a task. type SyncTaskContext = dao.SyncTaskContext -// ScheduledSyncTask contains one scheduled task and its connector schedule settings. -type ScheduledSyncTask = dao.ScheduledSyncTask - // SyncStats accumulates task-level document results. type SyncStats struct { Added int64 @@ -82,16 +72,6 @@ func NewSyncTaskService(taskDAO *dao.SyncTaskDAO) *SyncTaskService { return &SyncTaskService{taskDAO: taskDAO} } -// ListScheduledTasks returns scheduled tasks for one-time timer reconciliation. -func (s *SyncTaskService) ListScheduledTasks(ctx context.Context) ([]ScheduledSyncTask, error) { - return s.taskDAO.ListScheduledTasks(ctx, 4096) -} - -// GetScheduledTask returns one scheduled task for timer registration. -func (s *SyncTaskService) GetScheduledTask(ctx context.Context, taskID string) (ScheduledSyncTask, error) { - return s.taskDAO.GetScheduledTask(ctx, taskID) -} - // Claim marks a scheduled task running if no other scanner claimed it first. func (s *SyncTaskService) Claim(ctx context.Context, taskID string) (bool, error) { claimed, err := s.taskDAO.ClaimTask(ctx, taskID, time.Now().Local()) @@ -107,39 +87,6 @@ func (s *SyncTaskService) Claim(ctx context.Context, taskID string) (bool, error return true, s.taskDAO.MarkConnectorRunning(ctx, taskContext.Connector.ID) } -// GetContext loads a task execution context. -func (s *SyncTaskService) GetContext(ctx context.Context, taskID string) (SyncTaskContext, error) { - return s.taskDAO.GetTaskContext(ctx, taskID) -} - -// IsCanceled reports whether a task was canceled while a worker is running it. -func (s *SyncTaskService) IsCanceled(ctx context.Context, taskID string) (bool, error) { - return s.taskDAO.IsTaskCanceled(ctx, taskID) -} - -// RescheduleClaimed puts a claimed task back into schedule state. -func (s *SyncTaskService) RescheduleClaimed(ctx context.Context, taskID string) error { - return s.taskDAO.RescheduleClaimed(ctx, taskID) -} - -// Fail records a failed task. -func (s *SyncTaskService) Fail(ctx context.Context, taskID, connectorID string, err error) error { - message := "" - if err != nil { - message = err.Error() - } - return s.taskDAO.FailTask(ctx, taskID, connectorID, message, 1) -} - -// HandleTransientFailure retries a running task until maxRetries is reached. -func (s *SyncTaskService) HandleTransientFailure(ctx context.Context, taskID, connectorID string, err error, maxRetries int64) (int64, bool, error) { - message := "" - if err != nil { - message = err.Error() - } - return s.taskDAO.HandleTransientFailure(ctx, taskID, connectorID, message, maxRetries) -} - // CompleteSync commits a successful SYNC task and schedules the next one. func (s *SyncTaskService) CompleteSync(ctx context.Context, taskContext SyncTaskContext, pollRangeEnd time.Time, stats SyncStats) (string, error) { changed := stats.Added + stats.Updated @@ -151,18 +98,6 @@ func (s *SyncTaskService) CompletePrune(ctx context.Context, taskContext SyncTas return s.taskDAO.CompletePruneTask(ctx, taskContext, removed) } -// RecoverStaleRunning restores timed-out running tasks. -func (s *SyncTaskService) RecoverStaleRunning(ctx context.Context, now time.Time) error { - _, err := s.taskDAO.RecoverStaleRunning(ctx, now) - return err -} - -// RecoverRunning restores running tasks during syncer startup. -func (s *SyncTaskService) RecoverRunning(ctx context.Context) error { - _, err := s.taskDAO.RecoverRunning(ctx) - return err -} - // IsFromBeginning reports whether a task is a full sync. func IsFromBeginning(value *string) bool { if value == nil { diff --git a/internal/syncer/connector/builtin.go b/internal/syncer/connector/builtin.go index c15e0c04bf..00463a9aad 100644 --- a/internal/syncer/connector/builtin.go +++ b/internal/syncer/connector/builtin.go @@ -18,7 +18,6 @@ package connector import ( "context" - "fmt" "ragflow/internal/dao" ) @@ -53,11 +52,7 @@ func registerBuiltIn[T Connector](registry *Registry, source string, factory fun registry.RegisterConfigFactory(source, func(config map[string]any) (Connector, error) { return factory(config) }) - registry.Register(source, func(ctx context.Context, taskContext any) (Connector, error) { - row, ok := taskContext.(dao.SyncTaskContext) - if !ok { - return nil, fmt.Errorf("%s connector received an invalid task context", source) - } - return factory(map[string]any(row.Connector.Config)) + registry.Register(source, func(ctx context.Context, taskContext dao.SyncTaskContext) (Connector, error) { + return factory(map[string]any(taskContext.Connector.Config)) }) } diff --git a/internal/syncer/connector/registry.go b/internal/syncer/connector/registry.go index 160649ae54..93c1034c21 100644 --- a/internal/syncer/connector/registry.go +++ b/internal/syncer/connector/registry.go @@ -19,11 +19,12 @@ package connector import ( "context" "fmt" + "ragflow/internal/dao" "sync" ) // Factory creates a connector for a task context. -type Factory func(ctx context.Context, taskContext any) (Connector, error) +type Factory func(ctx context.Context, taskContext dao.SyncTaskContext) (Connector, error) // ConfigFactory creates a connector from raw connector config. type ConfigFactory func(config map[string]any) (Connector, error) @@ -58,24 +59,8 @@ func (r *Registry) RegisterConfigFactory(source string, factory ConfigFactory) { } // Open creates a connector for a task context. -func (r *Registry) Open(ctx context.Context, taskContext any) (Connector, error) { - row, ok := taskContext.(interface{ ConnectorSource() string }) - if ok { - return r.openSource(ctx, row.ConnectorSource(), taskContext) - } - sourceProvider, ok := taskContext.(interface{ Source() string }) - if ok { - return r.openSource(ctx, sourceProvider.Source(), taskContext) - } - rowContext, ok := taskContext.(struct{ Connector struct{ Source string } }) - if ok { - return r.openSource(ctx, rowContext.Connector.Source, taskContext) - } - source := "" - if value, ok := any(taskContext).(interface{ GetSource() string }); ok { - source = value.GetSource() - } - return r.openSource(ctx, source, taskContext) +func (r *Registry) Open(ctx context.Context, taskContext dao.SyncTaskContext) (Connector, error) { + return r.openSource(ctx, taskContext.Connector.Source, taskContext) } // OpenFromConfig builds a connector from a raw config map. @@ -90,7 +75,7 @@ func (r *Registry) OpenFromConfig(source string, config map[string]any) (Connect } // openSource creates a connector for a known source. -func (r *Registry) openSource(ctx context.Context, source string, taskContext any) (Connector, error) { +func (r *Registry) openSource(ctx context.Context, source string, taskContext dao.SyncTaskContext) (Connector, error) { r.mu.RLock() factory := r.factories[source] r.mu.RUnlock() diff --git a/internal/syncer/connector/registry_test.go b/internal/syncer/connector/registry_test.go index 79662fa34a..508eb224b6 100644 --- a/internal/syncer/connector/registry_test.go +++ b/internal/syncer/connector/registry_test.go @@ -2,6 +2,8 @@ package connector import ( "context" + "ragflow/internal/dao" + "ragflow/internal/entity" "strings" "testing" ) @@ -28,11 +30,13 @@ func TestRegistryOpenFromConfig(t *testing.T) { func TestRegistryOpenUsesTaskFactory(t *testing.T) { registry := NewRegistry() - registry.Register("rss", func(ctx context.Context, taskContext any) (Connector, error) { + registry.Register("rss", func(ctx context.Context, taskContext dao.SyncTaskContext) (Connector, error) { return NewRSSConnector(map[string]any{"feed_url": "https://example.com/feed.xml"}) }) - connector, err := registry.Open(context.Background(), struct{ Connector struct{ Source string } }{Connector: struct{ Source string }{Source: "rss"}}) + connector, err := registry.Open(context.Background(), dao.SyncTaskContext{ + Connector: entity.Connector{Source: "rss"}, + }) if err != nil { t.Fatalf("Open failed: %v", err) } diff --git a/internal/syncer/prune_runner.go b/internal/syncer/prune_runner.go index 3cbbeb4db8..86b45f4a5a 100644 --- a/internal/syncer/prune_runner.go +++ b/internal/syncer/prune_runner.go @@ -21,6 +21,7 @@ import ( "errors" "io" "ragflow/internal/common" + "ragflow/internal/dao" "ragflow/internal/service" syncerconnector "ragflow/internal/syncer/connector" @@ -29,17 +30,18 @@ import ( // PruneRunner executes one PRUNE task after collecting a full slim snapshot. type PruneRunner struct { + taskDAO *dao.SyncTaskDAO taskService *service.SyncTaskService pruneService *service.SyncPruneService } // NewPruneRunner creates a PRUNE runner. -func NewPruneRunner(taskService *service.SyncTaskService, pruneService *service.SyncPruneService) *PruneRunner { - return &PruneRunner{taskService: taskService, pruneService: pruneService} +func NewPruneRunner(taskDAO *dao.SyncTaskDAO, taskService *service.SyncTaskService, pruneService *service.SyncPruneService) *PruneRunner { + return &PruneRunner{taskDAO: taskDAO, taskService: taskService, pruneService: pruneService} } // Run collects the full prune snapshot before deleting stale documents. -func (r *PruneRunner) Run(ctx context.Context, taskContext service.SyncTaskContext, connector syncerconnector.Connector) (string, error) { +func (r *PruneRunner) Run(ctx context.Context, taskContext dao.SyncTaskContext, connector syncerconnector.Connector) (string, error) { if r.pruneService == nil { return "", errors.New("prune service is not configured") } @@ -53,7 +55,7 @@ func (r *PruneRunner) Run(ctx context.Context, taskContext service.SyncTaskConte if errors.Is(err, syncerconnector.ErrPruneUnsupported) { // Connectors without a slim snapshot interface (e.g. REST API) // complete PRUNE as a no-op without deleting anything. - if err := r.checkCanceled(ctx, taskContext.Task.ID); err != nil { + if err := checkTaskCanceled(r.taskDAO, ctx, taskContext.Task.ID); err != nil { return "", err } common.Warn("prune unsupported by connector, completing as no-op", zap.String("task_id", taskContext.Task.ID), zap.Error(err)) @@ -65,7 +67,7 @@ func (r *PruneRunner) Run(ctx context.Context, taskContext service.SyncTaskConte retain := map[string]struct{}{} for { - if err := r.checkCanceled(ctx, taskContext.Task.ID); err != nil { + if err := checkTaskCanceled(r.taskDAO, ctx, taskContext.Task.ID); err != nil { return "", err } batch, nextErr := session.NextBatch(ctx) @@ -80,7 +82,7 @@ func (r *PruneRunner) Run(ctx context.Context, taskContext service.SyncTaskConte } } - if err := r.checkCanceled(ctx, taskContext.Task.ID); err != nil { + if err := checkTaskCanceled(r.taskDAO, ctx, taskContext.Task.ID); err != nil { return "", err } removed, err := r.pruneService.DeleteStale(ctx, taskContext, retain) @@ -89,17 +91,3 @@ func (r *PruneRunner) Run(ctx context.Context, taskContext service.SyncTaskConte } return r.taskService.CompletePrune(ctx, taskContext, removed) } - -func (r *PruneRunner) checkCanceled(ctx context.Context, taskID string) error { - if err := ctx.Err(); err != nil { - return err - } - canceled, err := r.taskService.IsCanceled(ctx, taskID) - if err != nil { - return err - } - if canceled { - return errSyncTaskCanceled - } - return nil -} diff --git a/internal/syncer/scheduler.go b/internal/syncer/scheduler.go index 650dddc17c..e729bd58e1 100644 --- a/internal/syncer/scheduler.go +++ b/internal/syncer/scheduler.go @@ -20,14 +20,16 @@ import ( "context" "errors" "ragflow/internal/common" - "ragflow/internal/entity" - "ragflow/internal/service" + "ragflow/internal/dao" + "ragflow/internal/utility" "sync" "time" "go.uber.org/zap" ) +const scheduledTaskStartupPageSize = 4096 + // TaskEnvelope is the only payload sent through the task queue. type TaskEnvelope struct { TaskID string @@ -46,25 +48,25 @@ type SyncTaskBroker interface { // Scheduler discovers due work and enqueues task IDs for workers. type Scheduler struct { - queue chan<- TaskEnvelope - taskService *service.SyncTaskService - broker SyncTaskBroker - timerMu sync.Mutex - timers map[string]*time.Timer + queue chan<- TaskEnvelope + taskDAO *dao.SyncTaskDAO + broker SyncTaskBroker + timerMu sync.Mutex + timers map[string]*time.Timer } // NewScheduler creates a global scheduler for datasource sync tasks. -func NewScheduler(queue chan<- TaskEnvelope, taskService *service.SyncTaskService) *Scheduler { - return &Scheduler{queue: queue, taskService: taskService, timers: map[string]*time.Timer{}} +func NewScheduler(queue chan<- TaskEnvelope, taskDAO *dao.SyncTaskDAO) *Scheduler { + return &Scheduler{queue: queue, taskDAO: taskDAO, timers: map[string]*time.Timer{}} } // NewNATSScheduler creates a JetStream-driven scheduler with DB reconciliation. -func NewNATSScheduler(queue chan<- TaskEnvelope, taskService *service.SyncTaskService, broker SyncTaskBroker) *Scheduler { +func NewNATSScheduler(queue chan<- TaskEnvelope, taskDAO *dao.SyncTaskDAO, broker SyncTaskBroker) *Scheduler { return &Scheduler{ - queue: queue, - taskService: taskService, - broker: broker, - timers: map[string]*time.Timer{}, + queue: queue, + taskDAO: taskDAO, + broker: broker, + timers: map[string]*time.Timer{}, } } @@ -125,25 +127,35 @@ func (s *Scheduler) queueAvailable() int { // publishStartupTasks scans DB once for startup reconciliation. func (s *Scheduler) publishStartupTasks(ctx context.Context) error { - if err := s.taskService.RecoverRunning(ctx); err != nil { + if _, err := s.taskDAO.RecoverRunning(ctx); err != nil { return err } - tasks, err := s.taskService.ListScheduledTasks(ctx) - if err != nil { - return err - } - - for _, task := range tasks { - if err = s.ScheduleTask(ctx, task); err != nil { + var cursor *dao.ScheduledSyncTaskCursor + for { + tasks, err := s.taskDAO.ListScheduledTasks(ctx, scheduledTaskStartupPageSize, cursor) + if err != nil { return err } + if len(tasks) == 0 { + return nil + } + + for _, task := range tasks { + if err = s.ScheduleTask(ctx, task); err != nil { + return err + } + } + if len(tasks) < scheduledTaskStartupPageSize { + return nil + } + nextCursor := tasks[len(tasks)-1].Cursor() + cursor = &nextCursor } - return nil } // ScheduleTask publishes a due scheduled task or arms a one-shot timer. -func (s *Scheduler) ScheduleTask(ctx context.Context, task service.ScheduledSyncTask) error { +func (s *Scheduler) ScheduleTask(ctx context.Context, task dao.ScheduledSyncTask) error { delay, schedule := s.taskDelay(task, time.Now()) if !schedule { return nil @@ -157,14 +169,14 @@ func (s *Scheduler) ScheduleTaskAfter(ctx context.Context, taskID string, delay return nil } if delay <= 0 { - return s.publishTask(ctx, taskID) + return s.publish(ctx, taskID, false) } s.timerMu.Lock() if existing := s.timers[taskID]; existing != nil { existing.Stop() } timer := time.AfterFunc(delay, func() { - if err := s.publishTaskWakeup(ctx, taskID); err != nil && ctx.Err() == nil { + if err := s.publish(ctx, taskID, true); err != nil && ctx.Err() == nil { common.Warn("syncer scheduler timer publish failed", zap.String("task_id", taskID), zap.Error(err)) _ = s.ScheduleTaskAfter(ctx, taskID, 3*time.Second) return @@ -178,23 +190,22 @@ func (s *Scheduler) ScheduleTaskAfter(ctx context.Context, taskID string, delay return nil } -func (s *Scheduler) publishTaskWakeup(ctx context.Context, taskID string) error { +func (s *Scheduler) publish(ctx context.Context, taskID string, wakeup bool) error { if err := ctx.Err(); err != nil { return err } - if err := s.broker.PublishSyncerTaskWakeup(taskID); err != nil { - common.Warn("syncer task wakeup publish failed", zap.String("task_id", taskID), zap.Error(err)) - return err + var err error + if wakeup { + err = s.broker.PublishSyncerTaskWakeup(taskID) + } else { + err = s.broker.PublishSyncerTask(taskID) } - return nil -} - -func (s *Scheduler) publishTask(ctx context.Context, taskID string) error { - if err := ctx.Err(); err != nil { - return err - } - if err := s.broker.PublishSyncerTask(taskID); err != nil { - common.Warn("syncer task publish failed", zap.String("task_id", taskID), zap.Error(err)) + if err != nil { + message := "syncer task publish failed" + if wakeup { + message = "syncer task wakeup publish failed" + } + common.Warn(message, zap.String("task_id", taskID), zap.Error(err)) return err } return nil @@ -210,13 +221,13 @@ func (s *Scheduler) stopTimers() { } // taskDelay reports the delay before publication and whether the task must be scheduled at all. -func (s *Scheduler) taskDelay(task service.ScheduledSyncTask, now time.Time) (time.Duration, bool) { +func (s *Scheduler) taskDelay(task dao.ScheduledSyncTask, now time.Time) (time.Duration, bool) { freq := int64(0) switch task.TaskType { - case service.TaskTypeSync: + case dao.TaskTypeSync: freq = task.ConnectorRefreshFreq - case service.TaskTypePrune: - if !syncerConfigBool(task.ConnectorConfig, "sync_deleted_files") { + case dao.TaskTypePrune: + if !utility.ConfigBool(task.ConnectorConfig, "sync_deleted_files") { return 0, false } freq = task.ConnectorPruneFreq @@ -226,18 +237,3 @@ func (s *Scheduler) taskDelay(task service.ScheduledSyncTask, now time.Time) (ti } return task.UpdateDate.Add(time.Duration(freq) * time.Minute).Sub(now), true } - -func syncerConfigBool(config entity.JSONMap, key string) bool { - value, ok := config[key] - if !ok { - return false - } - switch typed := value.(type) { - case bool: - return typed - case string: - return typed == "1" || typed == "true" || typed == "TRUE" - default: - return false - } -} diff --git a/internal/syncer/sync_runner.go b/internal/syncer/sync_runner.go index f7e913ec24..7f905ff720 100644 --- a/internal/syncer/sync_runner.go +++ b/internal/syncer/sync_runner.go @@ -23,6 +23,7 @@ import ( "io" "math/rand" "ragflow/internal/common" + "ragflow/internal/dao" "ragflow/internal/service" syncerconnector "ragflow/internal/syncer/connector" "time" @@ -34,9 +35,30 @@ var errSyncTaskCanceled = errors.New("sync task canceled") const syncCancelCheckInterval = time.Second +// SyncRunnerConfig controls per-task document processing. +type SyncRunnerConfig struct { + ItemRetryCount int + ItemRetryBaseDelay time.Duration +} + +func checkTaskCanceled(taskDAO *dao.SyncTaskDAO, ctx context.Context, taskID string) error { + if err := ctx.Err(); err != nil { + return err + } + canceled, err := taskDAO.IsTaskCanceled(ctx, taskID) + if err != nil { + return err + } + if canceled { + return errSyncTaskCanceled + } + return nil +} + // SyncRunner executes one SYNC task by submitting source batches as BatchJobs. type SyncRunner struct { - config TaskCoordinatorConfig + config SyncRunnerConfig + taskDAO *dao.SyncTaskDAO taskService *service.SyncTaskService sink service.DocumentSink idResolver *service.DocumentIDResolver @@ -45,15 +67,21 @@ type SyncRunner struct { } // NewSyncRunner creates a SYNC runner. -func NewSyncRunner(config TaskCoordinatorConfig, taskService *service.SyncTaskService, sink service.DocumentSink, idResolver *service.DocumentIDResolver, queue *SyncJobQueue, checkpoints SyncCheckpointStore) *SyncRunner { +func NewSyncRunner(config SyncRunnerConfig, taskDAO *dao.SyncTaskDAO, taskService *service.SyncTaskService, sink service.DocumentSink, idResolver *service.DocumentIDResolver, queue *SyncJobQueue, checkpoints SyncCheckpointStore) *SyncRunner { + if config.ItemRetryCount <= 0 { + config.ItemRetryCount = 1 + } + if config.ItemRetryBaseDelay <= 0 { + config.ItemRetryBaseDelay = time.Second + } if checkpoints == nil { checkpoints = newMemorySyncCheckpointStore() } - return &SyncRunner{config: config, taskService: taskService, sink: sink, idResolver: idResolver, queue: queue, checkpoints: checkpoints} + return &SyncRunner{config: config, taskDAO: taskDAO, taskService: taskService, sink: sink, idResolver: idResolver, queue: queue, checkpoints: checkpoints} } // Run executes all sync batches and commits the final waterline. -func (r *SyncRunner) Run(ctx context.Context, taskContext service.SyncTaskContext, connector syncerconnector.Connector) (string, error) { +func (r *SyncRunner) Run(ctx context.Context, taskContext dao.SyncTaskContext, connector syncerconnector.Connector) (string, error) { // sink is nil means this syncer task cannot write it to document, it will fail anyway if r.sink == nil { return "", errors.New("document sink is not configured") @@ -98,7 +126,7 @@ func (r *SyncRunner) Run(ctx context.Context, taskContext service.SyncTaskContex for { // check if task has been canceled - if err := r.checkCanceled(ctx, taskContext.Task.ID); err != nil { + if err := checkTaskCanceled(r.taskDAO, ctx, taskContext.Task.ID); err != nil { return "", err } @@ -126,7 +154,7 @@ func (r *SyncRunner) Run(ctx context.Context, taskContext service.SyncTaskContex return "", err } - if err := r.checkCanceled(ctx, taskContext.Task.ID); err != nil { + if err := checkTaskCanceled(r.taskDAO, ctx, taskContext.Task.ID); err != nil { return "", err } nextTaskID, err := r.taskService.CompleteSync(ctx, taskContext, checkpointState.WindowEnd, stats) @@ -169,7 +197,7 @@ func (r *SyncRunner) collectResults(ctx context.Context, taskID string, resultCh return nil } -func (r *SyncRunner) prepareCheckpoint(ctx context.Context, taskContext service.SyncTaskContext, windowStart *time.Time, windowEnd time.Time) (syncerconnector.SyncCheckpointState, error) { +func (r *SyncRunner) prepareCheckpoint(ctx context.Context, taskContext dao.SyncTaskContext, windowStart *time.Time, windowEnd time.Time) (syncerconnector.SyncCheckpointState, error) { state, err := r.checkpoints.LoadSyncCheckpoint(ctx, taskContext.Task.ID) if err != nil { return syncerconnector.SyncCheckpointState{}, err @@ -219,7 +247,7 @@ func applyStatsToCheckpointState(state *syncerconnector.SyncCheckpointState, sta } // submitBatch submits one source batch as one BatchJob. -func (r *SyncRunner) submitBatch(ctx context.Context, taskContext service.SyncTaskContext, sourceType string, session syncerconnector.SyncSession, batch syncerconnector.SyncBatch) (<-chan syncJobResult, error) { +func (r *SyncRunner) submitBatch(ctx context.Context, taskContext dao.SyncTaskContext, sourceType string, session syncerconnector.SyncSession, batch syncerconnector.SyncBatch) (<-chan syncJobResult, error) { resultChan, err := r.queue.submit(ctx, func(jobCtx context.Context) (service.SyncStats, error) { return r.processDocuments(jobCtx, taskContext, sourceType, session, batch.Documents) }, batch.Checkpoint) @@ -246,7 +274,7 @@ func cloneSyncCheckpoint(checkpoint *syncerconnector.SyncCheckpoint) *syncerconn } // processDocuments -func (r *SyncRunner) processDocuments(ctx context.Context, taskContext service.SyncTaskContext, sourceType string, session syncerconnector.SyncSession, documents []syncerconnector.SourceDocument) (service.SyncStats, error) { +func (r *SyncRunner) processDocuments(ctx context.Context, taskContext dao.SyncTaskContext, sourceType string, session syncerconnector.SyncSession, documents []syncerconnector.SourceDocument) (service.SyncStats, error) { stats := service.SyncStats{} var firstErr error @@ -256,7 +284,7 @@ func (r *SyncRunner) processDocuments(ctx context.Context, taskContext service.S return stats, err } if lastCancelCheck.IsZero() || time.Since(lastCancelCheck) >= syncCancelCheckInterval { - if err := r.checkCanceled(ctx, taskContext.Task.ID); err != nil { + if err := checkTaskCanceled(r.taskDAO, ctx, taskContext.Task.ID); err != nil { return stats, err } lastCancelCheck = time.Now() @@ -274,7 +302,7 @@ func (r *SyncRunner) processDocuments(ctx context.Context, taskContext service.S } // processDocumentWithRetry retries transient item failures. -func (r *SyncRunner) processDocumentWithRetry(ctx context.Context, taskContext service.SyncTaskContext, sourceType string, session syncerconnector.SyncSession, sourceDocument syncerconnector.SourceDocument) (service.DocumentUpsertResult, error) { +func (r *SyncRunner) processDocumentWithRetry(ctx context.Context, taskContext dao.SyncTaskContext, sourceType string, session syncerconnector.SyncSession, sourceDocument syncerconnector.SourceDocument) (service.DocumentUpsertResult, error) { var lastErr error for attempt := 1; attempt <= r.config.ItemRetryCount; attempt++ { if err := ctx.Err(); err != nil { @@ -311,25 +339,8 @@ func (r *SyncRunner) processDocumentWithRetry(ctx context.Context, taskContext s return service.DocumentUpsertResult{}, lastErr } -// checkCanceled check if the task has been canceled -func (r *SyncRunner) checkCanceled(ctx context.Context, taskID string) error { - if err := ctx.Err(); err != nil { - return err - } - - canceled, err := r.taskService.IsCanceled(ctx, taskID) - if err != nil { - return err - } - - if canceled { - return errSyncTaskCanceled - } - return nil -} - // processDocument resolves IDs, skips unchanged fingerprints, fetches blobs, and upserts. -func (r *SyncRunner) processDocument(ctx context.Context, taskContext service.SyncTaskContext, sourceType string, session syncerconnector.SyncSession, sourceDocument syncerconnector.SourceDocument) (service.DocumentUpsertResult, error) { +func (r *SyncRunner) processDocument(ctx context.Context, taskContext dao.SyncTaskContext, sourceType string, session syncerconnector.SyncSession, sourceDocument syncerconnector.SourceDocument) (service.DocumentUpsertResult, error) { resolved, err := r.idResolver.Resolve(ctx, taskContext.Knowledgebase.ID, taskContext.Connector.ID, sourceType, sourceDocument.SourceID) if err != nil { return service.DocumentUpsertResult{}, err diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go index 7f7d38e947..22d837f026 100644 --- a/internal/syncer/syncer.go +++ b/internal/syncer/syncer.go @@ -82,14 +82,14 @@ func New(config Config, taskDAO *dao.SyncTaskDAO, registry ConnectorRegistry, si checkpoints = store } - coordinator := NewTaskCoordinator(TaskCoordinatorConfig{ + coordinator := NewTaskCoordinator(SyncRunnerConfig{ ItemRetryCount: config.ItemRetryCount, ItemRetryBaseDelay: config.ItemRetryBaseDelay, - }, taskService, registry, sink, pruneService, idResolver, executor, checkpoints) + }, taskDAO, taskService, registry, sink, pruneService, idResolver, executor, checkpoints) - scheduler := NewScheduler(queue, taskService) + scheduler := NewScheduler(queue, taskDAO) if broker, ok := messageQueue.(SyncTaskBroker); ok { - scheduler = NewNATSScheduler(queue, taskService, broker) + scheduler = NewNATSScheduler(queue, taskDAO, broker) } return &Syncer{ @@ -97,7 +97,7 @@ func New(config Config, taskDAO *dao.SyncTaskDAO, registry ConnectorRegistry, si config: config, queue: queue, scheduler: scheduler, - worker: NewTaskWorker(queue, taskService, coordinator, locker).WithScheduler(scheduler), + worker: NewTaskWorker(queue, taskDAO, taskService, coordinator, locker).WithScheduler(scheduler), executor: executor, ShutdownCh: make(chan struct{}), } @@ -152,9 +152,9 @@ func (s *Syncer) Stop() { }) } -// logSyncTaskDuration test run time, delete it soon -func logSyncTaskDuration(taskContext service.SyncTaskContext, startedAt time.Time) { - if taskContext.Task.TaskType != service.TaskTypeSync { +// logSyncTaskDuration get job run time +func logSyncTaskDuration(taskContext dao.SyncTaskContext, startedAt time.Time) { + if taskContext.Task.TaskType != dao.TaskTypeSync { return } common.Info( diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go index 8b6d9549da..1135c9e31c 100644 --- a/internal/syncer/syncer_test.go +++ b/internal/syncer/syncer_test.go @@ -19,6 +19,7 @@ package syncer import ( "context" "errors" + "fmt" "io" "ragflow/internal/common" "ragflow/internal/dao" @@ -167,14 +168,18 @@ type fakeSyncTaskBroker struct { mu sync.Mutex published []string handler func(common.TaskHandle) + onPublish func(taskID string) } func (b *fakeSyncTaskBroker) InitSyncerStream() error { return nil } func (b *fakeSyncTaskBroker) InitSyncerConsumer() error { return nil } func (b *fakeSyncTaskBroker) PublishSyncerTask(taskID string) error { b.mu.Lock() - defer b.mu.Unlock() b.published = append(b.published, taskID) + b.mu.Unlock() + if b.onPublish != nil { + b.onPublish(taskID) + } return nil } func (b *fakeSyncTaskBroker) PublishSyncerTaskWakeup(taskID string) error { @@ -190,6 +195,18 @@ func (b *fakeSyncTaskBroker) publishedIDs() []string { return append([]string(nil), b.published...) } +type closingLocker struct { + db *gorm.DB +} + +func (l closingLocker) TryLock(connectorID, kbID string) (ConnectorLockLease, bool) { + sqlDB, _ := l.db.DB() + _ = sqlDB.Close() + return ConnectorLockLease{}, false +} + +func (l closingLocker) Unlock(connectorID, kbID string) {} + // DeleteDocument records one delete. func (d *fakeDeleter) DeleteDocument(ctx context.Context, docID string) error { d.mu.Lock() @@ -311,21 +328,20 @@ func insertSyncLog(t *testing.T, db *gorm.DB, connectorID, kbID, taskID, taskTyp // newTestRegistry creates a mock connector registry. func newTestRegistry(connectors map[string]*connectormock.Connector) *syncerconnector.Registry { registry := syncerconnector.NewRegistry() - registry.Register("mock", func(ctx context.Context, taskContext any) (syncerconnector.Connector, error) { - row := taskContext.(dao.SyncTaskContext) - return connectors[row.Connector.ID], nil + registry.Register("mock", func(ctx context.Context, taskContext dao.SyncTaskContext) (syncerconnector.Connector, error) { + return connectors[taskContext.Connector.ID], nil }) return registry } // newCoordinator creates a test coordinator. -func newCoordinator(taskService *service.SyncTaskService, registry *syncerconnector.Registry, sink service.DocumentSink, pruneService *service.SyncPruneService, store service.DocumentStore) *TaskCoordinator { - return newCoordinatorWithCheckpoints(taskService, registry, sink, pruneService, store, newMemorySyncCheckpointStore()) +func newCoordinator(taskDAO *dao.SyncTaskDAO, taskService *service.SyncTaskService, registry *syncerconnector.Registry, sink service.DocumentSink, pruneService *service.SyncPruneService, store service.DocumentStore) *TaskCoordinator { + return newCoordinatorWithCheckpoints(taskDAO, taskService, registry, sink, pruneService, store, newMemorySyncCheckpointStore()) } -func newCoordinatorWithCheckpoints(taskService *service.SyncTaskService, registry *syncerconnector.Registry, sink service.DocumentSink, pruneService *service.SyncPruneService, store service.DocumentStore, checkpoints SyncCheckpointStore) *TaskCoordinator { +func newCoordinatorWithCheckpoints(taskDAO *dao.SyncTaskDAO, taskService *service.SyncTaskService, registry *syncerconnector.Registry, sink service.DocumentSink, pruneService *service.SyncPruneService, store service.DocumentStore, checkpoints SyncCheckpointStore) *TaskCoordinator { executor := NewSyncJobExecutor(SyncJobExecutorConfig{WorkerCount: 16}) - return NewTaskCoordinator(TaskCoordinatorConfig{ItemRetryCount: 1, ItemRetryBaseDelay: time.Millisecond}, taskService, registry, sink, pruneService, service.NewDocumentIDResolver(store), executor, checkpoints) + return NewTaskCoordinator(SyncRunnerConfig{ItemRetryCount: 1, ItemRetryBaseDelay: time.Millisecond}, taskDAO, taskService, registry, sink, pruneService, service.NewDocumentIDResolver(store), executor, checkpoints) } // TestClaimBlocksSameConnectorKBRunningTasks verifies DB-backed task mutual exclusion. @@ -333,7 +349,8 @@ func TestClaimBlocksSameConnectorKBRunningTasks(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) insertSyncLog(t, db, "conn-1", "kb-1", "task-2", dao.TaskTypePrune) - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) claimed, err := taskService.Claim(t.Context(), "task-1") if err != nil { t.Fatalf("claim task-1: %v", err) @@ -356,7 +373,8 @@ func TestClaimAllowsSameConnectorDifferentKB(t *testing.T) { insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) insertKnowledgebaseMapping(t, db, "conn-1", "kb-2") insertSyncLog(t, db, "conn-1", "kb-2", "task-2", dao.TaskTypeSync) - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) for _, taskID := range []string{"task-1", "task-2"} { claimed, err := taskService.Claim(t.Context(), taskID) if err != nil { @@ -370,7 +388,7 @@ func TestClaimAllowsSameConnectorDifferentKB(t *testing.T) { // TestSchedulerRequiresBroker verifies JetStream is mandatory for the scheduler. func TestSchedulerRequiresBroker(t *testing.T) { - scheduler := NewScheduler(make(chan TaskEnvelope, 1), service.NewSyncTaskService(dao.NewSyncTaskDAO(nil))) + scheduler := NewScheduler(make(chan TaskEnvelope, 1), dao.NewSyncTaskDAO(nil)) err := scheduler.Run(t.Context()) if err == nil || !strings.Contains(err.Error(), "NATS broker") { t.Fatalf("Run error = %v, want missing broker", err) @@ -381,9 +399,9 @@ func TestSchedulerRequiresBroker(t *testing.T) { func TestNATSSchedulerPublishesDueTasksWithoutClaiming(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) broker := &fakeSyncTaskBroker{} - scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskService, broker) + scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskDAO, broker) if err := scheduler.publishStartupTasks(t.Context()); err != nil { t.Fatalf("publish startup tasks: %v", err) } @@ -399,6 +417,83 @@ func TestNATSSchedulerPublishesDueTasksWithoutClaiming(t *testing.T) { } } +// TestNATSSchedulerStartupPaginatesScheduledTasks verifies startup reconciliation publishes every scheduled task. +func TestNATSSchedulerStartupPaginatesScheduledTasks(t *testing.T) { + db := setupSyncerDB(t) + total := scheduledTaskStartupPageSize + 1 + for i := 0; i < total; i++ { + taskID := fmt.Sprintf("task-%04d", i) + insertTaskContext(t, db, fmt.Sprintf("conn-%04d", i), fmt.Sprintf("kb-%04d", i), taskID, dao.TaskTypeSync) + } + + taskDAO := dao.NewSyncTaskDAO(db) + broker := &fakeSyncTaskBroker{onPublish: func(taskID string) { + claimed, err := taskDAO.ClaimTask(t.Context(), taskID, time.Now()) + if err != nil { + t.Errorf("claim published task %s: %v", taskID, err) + return + } + if !claimed { + t.Errorf("published task %s was not claimable", taskID) + } + }} + scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskDAO, broker) + + if err := scheduler.publishStartupTasks(t.Context()); err != nil { + t.Fatalf("publish startup tasks: %v", err) + } + got := broker.publishedIDs() + if len(got) != total { + t.Fatalf("published tasks = %d, want %d", len(got), total) + } + seen := make(map[string]struct{}, len(got)) + for _, taskID := range got { + seen[taskID] = struct{}{} + } + if len(seen) != total { + t.Fatalf("published unique tasks = %d, want %d", len(seen), total) + } +} + +func TestListScheduledTasksPaginatesNullUpdateTime(t *testing.T) { + db := setupSyncerDB(t) + insertTaskContext(t, db, "conn-1", "kb-1", "task-0", dao.TaskTypeSync) + insertSyncLog(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) + insertSyncLog(t, db, "conn-1", "kb-1", "task-2", dao.TaskTypeSync) + if err := db.Model(&entity.SyncLogs{}). + Where("id IN ?", []string{"task-0", "task-1", "task-2"}). + UpdateColumns(map[string]any{"update_time": gorm.Expr("NULL"), "update_date": gorm.Expr("NULL")}).Error; err != nil { + t.Fatalf("clear update timestamps: %v", err) + } + + taskDAO := dao.NewSyncTaskDAO(db) + firstPage, err := taskDAO.ListScheduledTasks(t.Context(), 2, nil) + if err != nil { + t.Fatalf("list first page: %v", err) + } + if len(firstPage) != 2 { + t.Fatalf("first page length = %d, want 2", len(firstPage)) + } + cursor := firstPage[len(firstPage)-1].Cursor() + secondPage, err := taskDAO.ListScheduledTasks(t.Context(), 2, &cursor) + if err != nil { + t.Fatalf("list second page: %v", err) + } + + seen := map[string]struct{}{} + for _, task := range append(firstPage, secondPage...) { + seen[task.ID] = struct{}{} + } + if len(seen) != 3 { + t.Fatalf("listed unique tasks = %d, want 3", len(seen)) + } + for _, taskID := range []string{"task-0", "task-1", "task-2"} { + if _, ok := seen[taskID]; !ok { + t.Fatalf("task %s was not listed", taskID) + } + } +} + // TestNATSSchedulerStartupDelaysFreshScheduledTask verifies startup keeps refresh windows without periodic DB scans. func TestNATSSchedulerStartupDelaysFreshScheduledTask(t *testing.T) { db := setupSyncerDB(t) @@ -415,9 +510,9 @@ func TestNATSSchedulerStartupDelaysFreshScheduledTask(t *testing.T) { t.Fatalf("set task update time: %v", err) } - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) broker := &fakeSyncTaskBroker{} - scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskService, broker) + scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskDAO, broker) if err := scheduler.publishStartupTasks(t.Context()); err != nil { t.Fatalf("publish startup tasks: %v", err) } @@ -442,9 +537,9 @@ func TestNATSSchedulerRecoversRunningTasksOnStartup(t *testing.T) { if err := db.Model(&entity.Connector{}).Where("id = ?", "conn-1").Update("status", dao.SyncStatusRunning).Error; err != nil { t.Fatalf("mark connector running: %v", err) } - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) broker := &fakeSyncTaskBroker{} - scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskService, broker) + scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskDAO, broker) if err := scheduler.publishStartupTasks(t.Context()); err != nil { t.Fatalf("publish startup tasks: %v", err) @@ -483,9 +578,9 @@ func TestNATSSchedulerStartupPublishesDueTasks(t *testing.T) { }).Error; err != nil { t.Fatalf("set fresh task update time: %v", err) } - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) broker := &fakeSyncTaskBroker{} - scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskService, broker) + scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskDAO, broker) if err := scheduler.publishStartupTasks(t.Context()); err != nil { t.Fatalf("publish fresh startup tasks: %v", err) @@ -512,9 +607,9 @@ func TestNATSSchedulerStartupPublishesDueTasks(t *testing.T) { // TestNATSSchedulerBuffersPushedTasks verifies enqueueHandle buffers pushed NATS handles in the local queue. func TestNATSSchedulerBuffersPushedTasks(t *testing.T) { - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(nil)) + taskDAO := dao.NewSyncTaskDAO(nil) queue := make(chan TaskEnvelope, 2) - scheduler := NewNATSScheduler(queue, taskService, &fakeSyncTaskBroker{}) + scheduler := NewNATSScheduler(queue, taskDAO, &fakeSyncTaskBroker{}) handles := []common.TaskHandle{ &fakeTaskHandle{msg: common.TaskMessage{TaskID: "task-1", TaskType: common.TaskTypeSyncer}}, &fakeTaskHandle{msg: common.TaskMessage{TaskID: "task-2", TaskType: common.TaskTypeSyncer}}, @@ -558,7 +653,7 @@ func TestWorkersRunDifferentConnectorsInParallel(t *testing.T) { "conn-2": {SyncBatches: []syncerconnector.SyncBatch{{Documents: []syncerconnector.SourceDocument{{SourceID: "b", UpdatedAt: now}}}}}, } queue := make(chan TaskEnvelope, 2) - worker := NewTaskWorker(queue, taskService, newCoordinator(taskService, newTestRegistry(connectors), sink, nil, fakeStore{}), NewConnectorLock()) + worker := NewTaskWorker(queue, taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(connectors), sink, nil, fakeStore{}), NewConnectorLock()) ctx, cancel := context.WithCancel(t.Context()) defer cancel() go worker.Run(ctx, 2) @@ -577,14 +672,16 @@ func TestWorkersRunDifferentConnectorsInParallel(t *testing.T) { func TestNATSTaskWorkerClaimsAndAcksOnSuccess(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) now := time.Now() connector := &connectormock.Connector{SyncBatches: []syncerconnector.SyncBatch{{Documents: []syncerconnector.SourceDocument{{SourceID: "a", UpdatedAt: now}}}}} handle := &fakeTaskHandle{msg: common.TaskMessage{TaskID: "task-1", TaskType: common.TaskTypeSyncer}} worker := NewTaskWorker( make(chan TaskEnvelope, 1), + taskDAO, taskService, - newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}), + newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}), NewConnectorLock(), ) worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1", Handle: handle}) @@ -608,9 +705,10 @@ func TestNATSTaskWorkerAcksUnclaimableMessage(t *testing.T) { if err := db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error; err != nil { t.Fatalf("mark running: %v", err) } - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) handle := &fakeTaskHandle{msg: common.TaskMessage{TaskID: "task-1", TaskType: common.TaskTypeSyncer}} - worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinator(taskService, newTestRegistry(nil), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(nil), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()) worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1", Handle: handle}) if handle.acks != 1 || handle.nacks != 0 { t.Fatalf("settlement acks=%d nacks=%d", handle.acks, handle.nacks) @@ -628,9 +726,10 @@ func TestNATSTaskWorkerRetriesStillScheduledUnclaimedTask(t *testing.T) { t.Fatalf("mark connector running: %v", err) } - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) - scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskService, &fakeSyncTaskBroker{}) - worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinator(taskService, newTestRegistry(nil), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()).WithScheduler(scheduler) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) + scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskDAO, &fakeSyncTaskBroker{}) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(nil), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()).WithScheduler(scheduler) handle := &fakeTaskHandle{msg: common.TaskMessage{TaskID: "task-2", TaskType: common.TaskTypeSyncer}} worker.handle(t.Context(), TaskEnvelope{TaskID: "task-2", Handle: handle}) @@ -650,6 +749,21 @@ func TestNATSTaskWorkerRetriesStillScheduledUnclaimedTask(t *testing.T) { } } +func TestNATSTaskWorkerNacksWhenLockContentionRescheduleFails(t *testing.T) { + db := setupSyncerDB(t) + insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) + handle := &fakeTaskHandle{msg: common.TaskMessage{TaskID: "task-1", TaskType: common.TaskTypeSyncer}} + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, nil, closingLocker{db: db}) + + worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1", Handle: handle}) + + if handle.acks != 0 || handle.nacks != 1 { + t.Fatalf("settlement acks=%d nacks=%d, want nack without ack", handle.acks, handle.nacks) + } +} + func TestNATSTaskWorkerDoesNotRetryCompletedUnclaimedTask(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) @@ -657,9 +771,10 @@ func TestNATSTaskWorkerDoesNotRetryCompletedUnclaimedTask(t *testing.T) { t.Fatalf("mark done: %v", err) } - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) - scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskService, &fakeSyncTaskBroker{}) - worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinator(taskService, newTestRegistry(nil), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()).WithScheduler(scheduler) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) + scheduler := NewNATSScheduler(make(chan TaskEnvelope, 1), taskDAO, &fakeSyncTaskBroker{}) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(nil), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()).WithScheduler(scheduler) handle := &fakeTaskHandle{msg: common.TaskMessage{TaskID: "task-1", TaskType: common.TaskTypeSyncer}} worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1", Handle: handle}) @@ -690,7 +805,7 @@ func TestSameConnectorDifferentKBsRunInParallel(t *testing.T) { sink := &fakeSink{delay: 100 * time.Millisecond} connector := &connectormock.Connector{SyncBatches: []syncerconnector.SyncBatch{{Documents: []syncerconnector.SourceDocument{{SourceID: "a", UpdatedAt: time.Now()}}}}} queue := make(chan TaskEnvelope, 2) - worker := NewTaskWorker(queue, taskService, newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}), NewConnectorLock()) + worker := NewTaskWorker(queue, taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}), NewConnectorLock()) ctx, cancel := context.WithCancel(t.Context()) defer cancel() go worker.Run(ctx, 2) @@ -738,7 +853,7 @@ func TestConnectorKBLockSerializesSyncAndPrune(t *testing.T) { PruneBatches: []syncerconnector.PruneBatch{{Documents: []syncerconnector.SlimDocument{{SourceID: "a"}}}}, } pruneService := service.NewSyncPruneService(&fakeDeleter{}, fakeStore{ids: map[string]struct{}{}}) - worker := NewTaskWorker(make(chan TaskEnvelope, 2), taskService, newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, pruneService, fakeStore{}), NewConnectorLock()) + worker := NewTaskWorker(make(chan TaskEnvelope, 2), taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, pruneService, fakeStore{}), NewConnectorLock()) done := make(chan struct{}) go func() { defer close(done) @@ -797,8 +912,8 @@ func TestSyncRunnerSubmitsBatchesBeforeWaiting(t *testing.T) { }, } sink := &fakeSink{delay: 80 * time.Millisecond} - coordinator := newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}) - taskContext, _ := taskService.GetContext(t.Context(), "task-1") + coordinator := newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}) + taskContext, _ := taskDAO.GetTaskContext(t.Context(), "task-1") if _, err := coordinator.Execute(t.Context(), taskContext, testLockLease()); err != nil { t.Fatalf("execute: %v", err) } @@ -812,7 +927,8 @@ func TestSyncRunnerProcessesBatchJobsInParallel(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) _ = db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) now := time.Now() connector := &connectormock.Connector{SyncBatches: []syncerconnector.SyncBatch{ {Documents: []syncerconnector.SourceDocument{{SourceID: "a", UpdatedAt: now}}}, @@ -820,8 +936,8 @@ func TestSyncRunnerProcessesBatchJobsInParallel(t *testing.T) { {Documents: []syncerconnector.SourceDocument{{SourceID: "c", UpdatedAt: now}}}, }} sink := &fakeSink{delay: 80 * time.Millisecond} - coordinator := newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}) - taskContext, _ := taskService.GetContext(t.Context(), "task-1") + coordinator := newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}) + taskContext, _ := taskDAO.GetTaskContext(t.Context(), "task-1") if _, err := coordinator.Execute(t.Context(), taskContext, testLockLease()); err != nil { t.Fatalf("execute: %v", err) } @@ -858,13 +974,14 @@ func TestFingerprintSkipsUnchangedDocument(t *testing.T) { }).Error; err != nil { t.Fatalf("set running task: %v", err) } - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) legacyID := service.Hash128("conn-1:source-1") store := fakeStore{ids: map[string]struct{}{legacyID: {}}, fingerprints: map[string]string{legacyID: "fp-1"}} connector := &connectormock.Connector{SyncBatches: []syncerconnector.SyncBatch{{Documents: []syncerconnector.SourceDocument{{SourceID: "source-1", Fingerprint: "fp-1", FetchRef: &syncerconnector.FetchReference{Key: "lazy"}, UpdatedAt: time.Now()}}}}} sink := &fakeSink{} - coordinator := newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, store) - taskContext, _ := taskService.GetContext(t.Context(), "task-1") + coordinator := newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, store) + taskContext, _ := taskDAO.GetTaskContext(t.Context(), "task-1") if _, err := coordinator.Execute(t.Context(), taskContext, testLockLease()); err != nil { t.Fatalf("execute: %v", err) } @@ -895,11 +1012,12 @@ func TestAutoParseFlagFlowsToSink(t *testing.T) { t.Fatalf("disable auto_parse: %v", err) } _ = db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) connector := &connectormock.Connector{SyncBatches: []syncerconnector.SyncBatch{{Documents: []syncerconnector.SourceDocument{{SourceID: "source-1", Blob: []byte("x"), UpdatedAt: time.Now()}}}}} sink := &fakeSink{} - coordinator := newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}) - taskContext, _ := taskService.GetContext(t.Context(), "task-1") + coordinator := newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}) + taskContext, _ := taskDAO.GetTaskContext(t.Context(), "task-1") if _, err := coordinator.Execute(t.Context(), taskContext, testLockLease()); err != nil { t.Fatalf("execute: %v", err) } @@ -914,11 +1032,12 @@ func TestCompleteSyncSchedulesNextRun(t *testing.T) { insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) _ = db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error _ = db.Model(&entity.Connector{}).Where("id = ?", "conn-1").Update("status", dao.SyncStatusRunning).Error - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) now := time.Now() connector := &connectormock.Connector{SyncBatches: []syncerconnector.SyncBatch{{Documents: []syncerconnector.SourceDocument{{SourceID: "source-1", Blob: []byte("x"), UpdatedAt: now}}}}} - coordinator := newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}) - taskContext, _ := taskService.GetContext(t.Context(), "task-1") + coordinator := newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}) + taskContext, _ := taskDAO.GetTaskContext(t.Context(), "task-1") if _, err := coordinator.Execute(t.Context(), taskContext, testLockLease()); err != nil { t.Fatalf("execute: %v", err) } @@ -947,7 +1066,8 @@ func TestCancelStopsRunningSync(t *testing.T) { insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) _ = db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error _ = db.Model(&entity.Connector{}).Where("id = ?", "conn-1").Update("status", dao.SyncStatusRunning).Error - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) now := time.Now() connector := &connectormock.Connector{SyncBatches: []syncerconnector.SyncBatch{{Documents: []syncerconnector.SourceDocument{ {SourceID: "source-1", Blob: []byte("one"), UpdatedAt: now}, @@ -967,7 +1087,7 @@ func TestCancelStopsRunningSync(t *testing.T) { }) }, } - worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}), NewConnectorLock()) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}), NewConnectorLock()) worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) if calls := sink.callCount(); calls != 1 { @@ -1003,14 +1123,15 @@ func TestSyncRunnerResultWaitHonorsCancel(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) _ = db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) - taskContext, err := taskService.GetContext(t.Context(), "task-1") + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) + taskContext, err := taskDAO.GetTaskContext(t.Context(), "task-1") if err != nil { t.Fatalf("get context: %v", err) } connector := &connectormock.Connector{SyncBatches: []syncerconnector.SyncBatch{{Documents: []syncerconnector.SourceDocument{{SourceID: "source-1", Blob: []byte("x"), UpdatedAt: time.Now()}}}}} queue := &SyncJobQueue{taskID: "task-1", jobs: make(chan *syncJob, 1)} - runner := NewSyncRunner(TaskCoordinatorConfig{ItemRetryCount: 1, ItemRetryBaseDelay: time.Millisecond}, taskService, &fakeSink{}, service.NewDocumentIDResolver(fakeStore{}), queue, newMemorySyncCheckpointStore()) + runner := NewSyncRunner(SyncRunnerConfig{ItemRetryCount: 1, ItemRetryBaseDelay: time.Millisecond}, taskDAO, taskService, &fakeSink{}, service.NewDocumentIDResolver(fakeStore{}), queue, newMemorySyncCheckpointStore()) ctx, cancel := context.WithCancel(context.Background()) result := make(chan error, 1) @@ -1041,11 +1162,12 @@ func TestBatchFailureDoesNotAdvanceWaterline(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) _ = db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) connector := &connectormock.Connector{SyncBatches: []syncerconnector.SyncBatch{{Documents: []syncerconnector.SourceDocument{{SourceID: "bad", Blob: []byte("x"), UpdatedAt: time.Now()}}}}} sink := &fakeSink{errBySourceID: map[string]error{"bad": errors.New("boom")}} queue := make(chan TaskEnvelope, 1) - worker := NewTaskWorker(queue, taskService, newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}), NewConnectorLock()) + worker := NewTaskWorker(queue, taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}), NewConnectorLock()) worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) var task entity.SyncLogs if err := db.First(&task, "id = ?", "task-1").Error; err != nil { @@ -1064,9 +1186,10 @@ func TestTransientFailureReschedulesTask(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) _ = db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) connector := &connectormock.Connector{SyncErrAt: 1} - worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()) worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) var task entity.SyncLogs @@ -1091,9 +1214,10 @@ func TestTransientFailureFailsAfterThreeRetries(t *testing.T) { if err := db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Updates(map[string]any{"status": dao.SyncStatusRunning, "error_count": int64(2)}).Error; err != nil { t.Fatalf("mark running: %v", err) } - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) connector := &connectormock.Connector{SyncErrAt: 1} - worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()) worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) var task entity.SyncLogs @@ -1116,7 +1240,8 @@ func TestTransientFetchFailureSavesCompletedBatchCheckpoint(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) _ = db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) checkpoints := newMemorySyncCheckpointStore() firstTime := time.Date(2026, 8, 12, 1, 0, 0, 0, time.UTC) connector := &connectormock.Connector{ @@ -1126,7 +1251,7 @@ func TestTransientFetchFailureSavesCompletedBatchCheckpoint(t *testing.T) { Checkpoint: &syncerconnector.SyncCheckpoint{Cursor: "cursor-1", UpdatedAt: &firstTime, SourceID: "source-1"}, }}, } - worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinatorWithCheckpoints(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}, checkpoints), NewConnectorLock()) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinatorWithCheckpoints(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}, checkpoints), NewConnectorLock()) worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) state, err := checkpoints.LoadSyncCheckpoint(t.Context(), "task-1") @@ -1157,7 +1282,8 @@ func TestSyncCheckpointDeletedAfterSuccess(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) _ = db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) checkpoints := newMemorySyncCheckpointStore() firstTime := time.Date(2026, 8, 11, 1, 0, 0, 0, time.UTC) @@ -1172,7 +1298,7 @@ func TestSyncCheckpointDeletedAfterSuccess(t *testing.T) { Checkpoint: &syncerconnector.SyncCheckpoint{Cursor: "cursor-2", UpdatedAt: &secondTime, SourceID: "source-2"}, }, }} - worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinatorWithCheckpoints(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}, checkpoints), NewConnectorLock()) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinatorWithCheckpoints(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}, checkpoints), NewConnectorLock()) worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) state, err := checkpoints.LoadSyncCheckpoint(t.Context(), "task-1") @@ -1189,7 +1315,8 @@ func TestSyncCheckpointStopsBeforeFailedBatch(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) _ = db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) checkpoints := newMemorySyncCheckpointStore() firstTime := time.Date(2026, 8, 11, 1, 0, 0, 0, time.UTC) @@ -1205,7 +1332,7 @@ func TestSyncCheckpointStopsBeforeFailedBatch(t *testing.T) { }, }} sink := &fakeSink{errBySourceID: map[string]error{"bad": errors.New("boom")}} - worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinatorWithCheckpoints(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}, checkpoints), NewConnectorLock()) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinatorWithCheckpoints(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), sink, nil, fakeStore{}, checkpoints), NewConnectorLock()) worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) state, err := checkpoints.LoadSyncCheckpoint(t.Context(), "task-1") @@ -1233,7 +1360,7 @@ func TestSyncCheckpointStopsBeforeFailedBatch(t *testing.T) { Documents: []syncerconnector.SourceDocument{{SourceID: "second-good", Blob: []byte("c"), UpdatedAt: thirdTime}}, Checkpoint: &syncerconnector.SyncCheckpoint{Cursor: "cursor-3", UpdatedAt: &thirdTime, SourceID: "second-good"}, }}} - resumeWorker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinatorWithCheckpoints(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": resumeConnector}), &fakeSink{}, nil, fakeStore{}, checkpoints), NewConnectorLock()) + resumeWorker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinatorWithCheckpoints(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": resumeConnector}), &fakeSink{}, nil, fakeStore{}, checkpoints), NewConnectorLock()) resumeWorker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) if len(resumeConnector.SyncRequests) != 1 { t.Fatalf("sync requests = %d, want 1", len(resumeConnector.SyncRequests)) @@ -1268,13 +1395,14 @@ func TestSyncRunnerClampsWaterlineToWindowEnd(t *testing.T) { t.Fatalf("mark running: %v", err) } - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) future := time.Date(2036, 6, 28, 9, 44, 0, 0, time.UTC) connector := &connectormock.Connector{SyncBatches: []syncerconnector.SyncBatch{{ Documents: []syncerconnector.SourceDocument{{SourceID: "future", Blob: []byte("future"), UpdatedAt: future}}, Checkpoint: &syncerconnector.SyncCheckpoint{Cursor: "cursor-future", UpdatedAt: &future, SourceID: "future"}, }}} - worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()) before := time.Now().UTC() worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) after := time.Now().UTC().Add(time.Second) @@ -1302,13 +1430,14 @@ func TestFullSyncWaterlineUsesWindowEnd(t *testing.T) { t.Fatalf("mark running: %v", err) } - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) oldSourceTime := time.Date(2025, 5, 17, 12, 39, 23, 0, time.UTC) connector := &connectormock.Connector{SyncBatches: []syncerconnector.SyncBatch{{ Documents: []syncerconnector.SourceDocument{{SourceID: "old", Blob: []byte("old"), UpdatedAt: oldSourceTime}}, Checkpoint: &syncerconnector.SyncCheckpoint{Cursor: "cursor-old", UpdatedAt: &oldSourceTime, SourceID: "old"}, }}} - worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskService, newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()) + worker := NewTaskWorker(make(chan TaskEnvelope, 1), taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, nil, fakeStore{}), NewConnectorLock()) before := time.Now().UTC() worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) after := time.Now().UTC().Add(time.Second) @@ -1325,38 +1454,18 @@ func TestFullSyncWaterlineUsesWindowEnd(t *testing.T) { } } -// TestRecoverStaleRunningTasks verifies timeout recovery. -func TestRecoverStaleRunningTasks(t *testing.T) { - db := setupSyncerDB(t) - insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypeSync) - started := time.Now().Add(-2 * time.Hour) - if err := db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Updates(map[string]any{"status": dao.SyncStatusRunning, "time_started": started}).Error; err != nil { - t.Fatalf("mark running: %v", err) - } - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) - if err := taskService.RecoverStaleRunning(t.Context(), time.Now()); err != nil { - t.Fatalf("recover: %v", err) - } - var task entity.SyncLogs - if err := db.First(&task, "id = ?", "task-1").Error; err != nil { - t.Fatalf("load task: %v", err) - } - if task.Status != dao.SyncStatusSchedule { - t.Fatalf("status = %s, want schedule", task.Status) - } -} - // TestPruneSourceFailureDoesNotDelete verifies incomplete source listings never delete. func TestPruneSourceFailureDoesNotDelete(t *testing.T) { db := setupSyncerDB(t) insertTaskContext(t, db, "conn-1", "kb-1", "task-1", dao.TaskTypePrune) _ = db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Update("status", dao.SyncStatusRunning).Error - taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) + taskDAO := dao.NewSyncTaskDAO(db) + taskService := service.NewSyncTaskService(taskDAO) deleter := &fakeDeleter{} pruneService := service.NewSyncPruneService(deleter, fakeStore{ids: map[string]struct{}{"stale": {}}}) connector := &connectormock.Connector{PruneErrAt: 1, PruneBatches: []syncerconnector.PruneBatch{{Documents: []syncerconnector.SlimDocument{{SourceID: "keep"}}}}} queue := make(chan TaskEnvelope, 1) - worker := NewTaskWorker(queue, taskService, newCoordinator(taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, pruneService, fakeStore{}), NewConnectorLock()) + worker := NewTaskWorker(queue, taskDAO, taskService, newCoordinator(taskDAO, taskService, newTestRegistry(map[string]*connectormock.Connector{"conn-1": connector}), &fakeSink{}, pruneService, fakeStore{}), NewConnectorLock()) worker.handle(t.Context(), TaskEnvelope{TaskID: "task-1"}) if len(deleter.deleted) != 0 { t.Fatalf("deleted %v despite incomplete snapshot", deleter.deleted) @@ -1388,7 +1497,7 @@ func TestMockSessionEOF(t *testing.T) { // TestTaskExecutionDeadlineCapsConnectorTimeout verifies tasks cannot outlive the connector lock lease. func TestTaskExecutionDeadlineCapsConnectorTimeout(t *testing.T) { now := time.Date(2026, 8, 10, 10, 0, 0, 0, time.UTC) - taskContext := service.SyncTaskContext{} + taskContext := dao.SyncTaskContext{} taskContext.Connector.TimeoutSecs = int64(connectorLockTTL.Seconds()) + 60 lease := ConnectorLockLease{ExpiresAt: now.Add(10 * time.Minute)} if got, want := taskExecutionDeadline(now, taskContext, lease), lease.ExpiresAt.Add(-connectorLockSafetyMargin); !got.Equal(want) { diff --git a/internal/syncer/task_coordinator.go b/internal/syncer/task_coordinator.go index 5a3b61abe6..97c991f6c6 100644 --- a/internal/syncer/task_coordinator.go +++ b/internal/syncer/task_coordinator.go @@ -19,6 +19,7 @@ package syncer import ( "context" "fmt" + "ragflow/internal/dao" "ragflow/internal/service" syncerconnector "ragflow/internal/syncer/connector" "time" @@ -29,25 +30,20 @@ const connectorLockSafetyMargin = 5 * time.Second // ConnectorRegistry opens registered connectors by source. type ConnectorRegistry interface { // Open creates a connector for a task context. - Open(ctx context.Context, taskContext any) (syncerconnector.Connector, error) -} - -// TaskCoordinatorConfig controls per-task document processing. -type TaskCoordinatorConfig struct { - ItemRetryCount int - ItemRetryBaseDelay time.Duration + Open(ctx context.Context, taskContext dao.SyncTaskContext) (syncerconnector.Connector, error) } // TaskCoordinator owns one task execution window. type TaskCoordinator struct { - config TaskCoordinatorConfig - taskService *service.SyncTaskService - registry ConnectorRegistry - sink service.DocumentSink - pruneService *service.SyncPruneService - idResolver *service.DocumentIDResolver - executor *SyncJobExecutor - checkpoints SyncCheckpointStore + syncRunnerConfig SyncRunnerConfig + taskDAO *dao.SyncTaskDAO + taskService *service.SyncTaskService + registry ConnectorRegistry + sink service.DocumentSink + pruneService *service.SyncPruneService + idResolver *service.DocumentIDResolver + executor *SyncJobExecutor + checkpoints SyncCheckpointStore } // TaskOutcome describes post-run scheduling work for a completed task. @@ -56,14 +52,7 @@ type TaskOutcome struct { } // NewTaskCoordinator creates a coordinator for one claimed task at a time. -func NewTaskCoordinator(config TaskCoordinatorConfig, taskService *service.SyncTaskService, registry ConnectorRegistry, sink service.DocumentSink, pruneService *service.SyncPruneService, idResolver *service.DocumentIDResolver, executor *SyncJobExecutor, checkpoints SyncCheckpointStore) *TaskCoordinator { - if config.ItemRetryCount <= 0 { - config.ItemRetryCount = 1 - } - - if config.ItemRetryBaseDelay <= 0 { - config.ItemRetryBaseDelay = time.Second - } +func NewTaskCoordinator(syncRunnerConfig SyncRunnerConfig, taskDAO *dao.SyncTaskDAO, taskService *service.SyncTaskService, registry ConnectorRegistry, sink service.DocumentSink, pruneService *service.SyncPruneService, idResolver *service.DocumentIDResolver, executor *SyncJobExecutor, checkpoints SyncCheckpointStore) *TaskCoordinator { if executor == nil { panic("task coordinator executor must not be nil") } @@ -71,11 +60,11 @@ func NewTaskCoordinator(config TaskCoordinatorConfig, taskService *service.SyncT checkpoints = newMemorySyncCheckpointStore() } - return &TaskCoordinator{config: config, taskService: taskService, registry: registry, sink: sink, pruneService: pruneService, idResolver: idResolver, executor: executor, checkpoints: checkpoints} + return &TaskCoordinator{syncRunnerConfig: syncRunnerConfig, taskDAO: taskDAO, taskService: taskService, registry: registry, sink: sink, pruneService: pruneService, idResolver: idResolver, executor: executor, checkpoints: checkpoints} } // Execute dispatches a sync_logs task by task type. -func (c *TaskCoordinator) Execute(ctx context.Context, taskContext service.SyncTaskContext, lease ConnectorLockLease) (TaskOutcome, error) { +func (c *TaskCoordinator) Execute(ctx context.Context, taskContext dao.SyncTaskContext, lease ConnectorLockLease) (TaskOutcome, error) { runCtx, cancel := context.WithDeadline(ctx, taskExecutionDeadline(time.Now(), taskContext, lease)) defer cancel() ctx = runCtx @@ -89,18 +78,18 @@ func (c *TaskCoordinator) Execute(ctx context.Context, taskContext service.SyncT } switch taskContext.Task.TaskType { - case service.TaskTypeSync: + case dao.TaskTypeSync: queue, err := c.executor.RegisterTask(ctx, taskContext.Task.ID) if err != nil { return TaskOutcome{}, err } defer queue.Close() - runner := NewSyncRunner(c.config, c.taskService, c.sink, c.idResolver, queue, c.checkpoints) + runner := NewSyncRunner(c.syncRunnerConfig, c.taskDAO, c.taskService, c.sink, c.idResolver, queue, c.checkpoints) nextTaskID, err := runner.Run(ctx, taskContext, connector) return TaskOutcome{NextTaskID: nextTaskID}, err - case service.TaskTypePrune: - runner := NewPruneRunner(c.taskService, c.pruneService) + case dao.TaskTypePrune: + runner := NewPruneRunner(c.taskDAO, c.taskService, c.pruneService) nextTaskID, err := runner.Run(ctx, taskContext, connector) return TaskOutcome{NextTaskID: nextTaskID}, err default: @@ -108,7 +97,7 @@ func (c *TaskCoordinator) Execute(ctx context.Context, taskContext service.SyncT } } -func taskExecutionDeadline(now time.Time, taskContext service.SyncTaskContext, lease ConnectorLockLease) time.Time { +func taskExecutionDeadline(now time.Time, taskContext dao.SyncTaskContext, lease ConnectorLockLease) time.Time { timeout := connectorLockTTL if seconds := taskContext.Connector.TimeoutSecs; seconds > 0 && seconds <= int64(connectorLockTTL/time.Second) { timeout = time.Duration(seconds) * time.Second diff --git a/internal/syncer/task_worker.go b/internal/syncer/task_worker.go index 3caf0b272d..6bd2f522eb 100644 --- a/internal/syncer/task_worker.go +++ b/internal/syncer/task_worker.go @@ -32,6 +32,7 @@ import ( // TaskWorker consumes claimed task envelopes and runs coordinators. type TaskWorker struct { queue <-chan TaskEnvelope + taskDAO *dao.SyncTaskDAO taskService *service.SyncTaskService coordinator *TaskCoordinator locker ConnectorLocker @@ -39,8 +40,8 @@ type TaskWorker struct { } // NewTaskWorker creates a bounded task worker pool. -func NewTaskWorker(queue <-chan TaskEnvelope, taskService *service.SyncTaskService, coordinator *TaskCoordinator, locker ConnectorLocker) *TaskWorker { - return &TaskWorker{queue: queue, taskService: taskService, coordinator: coordinator, locker: locker} +func NewTaskWorker(queue <-chan TaskEnvelope, taskDAO *dao.SyncTaskDAO, taskService *service.SyncTaskService, coordinator *TaskCoordinator, locker ConnectorLocker) *TaskWorker { + return &TaskWorker{queue: queue, taskDAO: taskDAO, taskService: taskService, coordinator: coordinator, locker: locker} } // WithScheduler attaches the event scheduler used for one-shot task publishing. @@ -98,15 +99,19 @@ func (w *TaskWorker) handle(ctx context.Context, envelope TaskEnvelope) { } // get the whole context by task_id from nats - taskContext, err := w.taskService.GetContext(ctx, envelope.TaskID) + taskContext, err := w.taskDAO.GetTaskContext(ctx, envelope.TaskID) if err != nil { if ctx.Err() != nil { // exiting - _ = w.taskService.RescheduleClaimed(context.WithoutCancel(ctx), envelope.TaskID) + if err = w.rescheduleClaimed(context.WithoutCancel(ctx), envelope.TaskID); err != nil { + common.Warn("syncer task reschedule failed after context cancellation", zap.String("task_id", envelope.TaskID), zap.Error(err)) + } nackEnvelope(envelope) return } - if failErr := w.taskService.Fail(ctx, envelope.TaskID, "", err); failErr != nil { // getContext failed - _ = w.taskService.RescheduleClaimed(context.WithoutCancel(ctx), envelope.TaskID) + if failErr := w.taskDAO.FailTask(ctx, envelope.TaskID, "", syncTaskErrorMessage(err), 1); failErr != nil { // getContext failed + if err = w.rescheduleClaimed(context.WithoutCancel(ctx), envelope.TaskID); err != nil { + common.Warn("syncer task reschedule failed after context load failure", zap.String("task_id", envelope.TaskID), zap.Error(err)) + } nackEnvelope(envelope) return } @@ -117,7 +122,11 @@ func (w *TaskWorker) handle(ctx context.Context, envelope TaskEnvelope) { // lock the connector and the KB lease, locked := w.locker.TryLock(taskContext.Connector.ID, taskContext.Knowledgebase.ID) if !locked { - _ = w.taskService.RescheduleClaimed(ctx, taskContext.Task.ID) + if err = w.rescheduleClaimed(context.WithoutCancel(ctx), taskContext.Task.ID); err != nil { + common.Warn("syncer task reschedule failed after lock contention", zap.String("task_id", taskContext.Task.ID), zap.Error(err)) + nackEnvelope(envelope) + return + } w.scheduleRetry(ctx, taskContext.Task.ID, 3*time.Second) ackEnvelope(envelope) return @@ -133,15 +142,21 @@ func (w *TaskWorker) handle(ctx context.Context, envelope TaskEnvelope) { return } if ctx.Err() != nil { // the task is canceled by system, this need to rerun - _ = w.taskService.RescheduleClaimed(context.WithoutCancel(ctx), taskContext.Task.ID) + if err = w.rescheduleClaimed(context.WithoutCancel(ctx), taskContext.Task.ID); err != nil { + common.Warn("syncer task reschedule failed after execution cancellation", zap.String("task_id", taskContext.Task.ID), zap.Error(err)) + nackEnvelope(envelope) + return + } w.scheduleRetry(context.WithoutCancel(ctx), taskContext.Task.ID, 3*time.Second) ackEnvelope(envelope) return } if isTransientSyncError(err) { - attempts, failed, transientErr := w.taskService.HandleTransientFailure(ctx, taskContext.Task.ID, taskContext.Connector.ID, err, maxTransientTaskRetries) + attempts, failed, transientErr := w.taskDAO.HandleTransientFailure(ctx, taskContext.Task.ID, taskContext.Connector.ID, syncTaskErrorMessage(err), maxTransientTaskRetries) if transientErr != nil { - _ = w.taskService.RescheduleClaimed(context.WithoutCancel(ctx), taskContext.Task.ID) + if err = w.rescheduleClaimed(context.WithoutCancel(ctx), taskContext.Task.ID); err != nil { + common.Warn("syncer task reschedule failed after transient failure handling error", zap.String("task_id", taskContext.Task.ID), zap.Error(err)) + } nackEnvelope(envelope) return } @@ -152,15 +167,17 @@ func (w *TaskWorker) handle(ctx context.Context, envelope TaskEnvelope) { ackEnvelope(envelope) return } - if failErr := w.taskService.Fail(ctx, taskContext.Task.ID, taskContext.Connector.ID, fmt.Errorf("sync task failed: %w", err)); failErr != nil { - _ = w.taskService.RescheduleClaimed(context.WithoutCancel(ctx), taskContext.Task.ID) + if failErr := w.taskDAO.FailTask(ctx, taskContext.Task.ID, taskContext.Connector.ID, syncTaskErrorMessage(fmt.Errorf("sync task failed: %w", err)), 1); failErr != nil { + if err = w.rescheduleClaimed(context.WithoutCancel(ctx), taskContext.Task.ID); err != nil { + common.Warn("syncer task reschedule failed after terminal failure handling error", zap.String("task_id", taskContext.Task.ID), zap.Error(err)) + } nackEnvelope(envelope) return } ackEnvelope(envelope) return } - logSyncTaskDuration(taskContext, startedAt) // Todo delete soon + logSyncTaskDuration(taskContext, startedAt) w.scheduleNext(ctx, outcome.NextTaskID) ackEnvelope(envelope) } @@ -169,7 +186,7 @@ func (w *TaskWorker) scheduleNext(ctx context.Context, taskID string) { if w.scheduler == nil || taskID == "" { return } - task, err := w.taskService.GetScheduledTask(ctx, taskID) + task, err := w.taskDAO.GetScheduledTask(ctx, taskID) if err != nil { common.Warn("syncer schedule next task lookup failed", zap.String("task_id", taskID), zap.Error(err)) return @@ -193,7 +210,7 @@ func (w *TaskWorker) scheduleRetryIfTaskScheduled(ctx context.Context, taskID st if w.scheduler == nil || taskID == "" { return } - taskContext, err := w.taskService.GetContext(ctx, taskID) + taskContext, err := w.taskDAO.GetTaskContext(ctx, taskID) if err != nil { common.Warn("syncer retry task lookup failed", zap.String("task_id", taskID), zap.Error(err)) return @@ -204,6 +221,20 @@ func (w *TaskWorker) scheduleRetryIfTaskScheduled(ctx context.Context, taskID st w.scheduleRetry(ctx, taskID, delay) } +func (w *TaskWorker) rescheduleClaimed(ctx context.Context, taskID string) error { + if w == nil || w.taskDAO == nil || taskID == "" { + return nil + } + return w.taskDAO.RescheduleClaimed(ctx, taskID) +} + +func syncTaskErrorMessage(err error) string { + if err == nil { + return "" + } + return err.Error() +} + // transientRetryDelay return retry delay func transientRetryDelay(attempts int64) time.Duration { if attempts < 1 { @@ -217,7 +248,7 @@ func transientRetryDelay(attempts int64) time.Duration { return time.Duration(1<