zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-08-22 16:23:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: simplify sync task service and harden scheduler recovery (#18550)

Browse Source 
### Summary

As title
This commit is contained in:
Haruko386
2026-08-20 19:11:23 +08:00
committed by GitHub
parent cab70fdac3
commit 1e46cef2d5
13 changed files with 433 additions and 426 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
internal/dao/sync_task.go
View File

@@ -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
}
}

65
internal/service/sync_task_service.go
View File

@@ -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 {

9
internal/syncer/connector/builtin.go
View File

@@ -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))
})
}

25
internal/syncer/connector/registry.go
View File

@@ -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()

8
internal/syncer/connector/registry_test.go
View File

@@ -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)
}

28
internal/syncer/prune_runner.go
View File

@@ -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
}

112
internal/syncer/scheduler.go
View File

@@ -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
}
}

69
internal/syncer/sync_runner.go
View File

@@ -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

16
internal/syncer/syncer.go
View File

@@ -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(

297
internal/syncer/syncer_test.go
View File

@@ -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) {

49
internal/syncer/task_coordinator.go
View File

@@ -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

63
internal/syncer/task_worker.go
View File

@@ -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<<shift) * 30 * time.Second
}
func logTransientSyncRetry(taskContext service.SyncTaskContext, attempts int64, failed bool, err error) {
func logTransientSyncRetry(taskContext dao.SyncTaskContext, attempts int64, failed bool, err error) {
message := "sync task transient retry scheduled"
if failed {
message = "sync task failed after transient retries"

16
internal/utility/convert.go
View File

@@ -71,6 +71,22 @@ func StringPtr(s string) *string {
return &s
}
// ConfigBool reads a Python JSON bool/string flag.
func ConfigBool[M ~map[string]any](config M, 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
}
}
// ParseInt64 parses a string to int64.
// If parsing fails, it returns 0.
//