From cdbd82df4f8af14bf488c691621d05244c205364 Mon Sep 17 00:00:00 2001 From: Haruko386 Date: Fri, 14 Aug 2026 17:43:55 +0800 Subject: [PATCH] fix: return total number of docs when sync has no update (#18255) ### Summary As title --------- Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> --- internal/dao/sync_task.go | 2 +- internal/handler/connector.go | 2 +- internal/router/router.go | 1 - internal/service/sync_task_service.go | 2 +- internal/syncer/connector/registry.go | 2 +- internal/syncer/syncer_test.go | 21 ++++++++++++++++++++- 6 files changed, 24 insertions(+), 6 deletions(-) diff --git a/internal/dao/sync_task.go b/internal/dao/sync_task.go index 5bfee83981..c3f8704f8f 100644 --- a/internal/dao/sync_task.go +++ b/internal/dao/sync_task.go @@ -333,7 +333,7 @@ func (d *SyncTaskDAO) CompleteSyncTask(ctx context.Context, taskContext SyncTask "status": SyncStatusDone, "poll_range_end": entity.FlexibleTime(pollRangeEnd), "new_docs_indexed": newDocs, - "total_docs_indexed": gorm.Expr("total_docs_indexed + ?", totalDocs), + "total_docs_indexed": totalDocs, "error_msg": errorMsg, "error_count": errorCount, }) diff --git a/internal/handler/connector.go b/internal/handler/connector.go index 63d314e460..796f2ae9c0 100644 --- a/internal/handler/connector.go +++ b/internal/handler/connector.go @@ -434,7 +434,7 @@ func (h *ConnectorHandler) RebuildConnector(c *gin.Context) { common.SuccessWithData(c, ok, "success") } -// ResumeFailedSync resumes a failed connector sync task from checkpoint. +// ResumeFailedSync resumes a failed connector sync task from checkpoint. (when network outage) // @Summary Resume Failed Connector Sync // @Description Resume a failed connector sync task from its saved checkpoint // @Tags connector diff --git a/internal/router/router.go b/internal/router/router.go index 10ebdd764d..9aa64e2813 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -626,7 +626,6 @@ func (r *Router) Setup(engine *gin.Engine) { { connectors.GET("", r.connectorHandler.ListConnectors) connectors.POST("", r.connectorHandler.CreateConnector) - // Sync logs for the current user, optionally filtered by dataset. connectors.GET("/sync_logs", r.connectorHandler.ListSyncLogs) connectors.POST("/google/oauth/web/start", r.connectorHandler.StartGoogleWebOAuth) connectors.POST("/google/oauth/web/result", r.connectorHandler.PollGoogleWebOAuthResult) diff --git a/internal/service/sync_task_service.go b/internal/service/sync_task_service.go index 83fcc26370..6e975997cf 100644 --- a/internal/service/sync_task_service.go +++ b/internal/service/sync_task_service.go @@ -143,7 +143,7 @@ func (s *SyncTaskService) HandleTransientFailure(ctx context.Context, taskID, co // 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 - return s.taskDAO.CompleteSyncTask(ctx, taskContext, pollRangeEnd, changed, changed, stats.ErrorCount, stats.ErrorMsg) + return s.taskDAO.CompleteSyncTask(ctx, taskContext, pollRangeEnd, stats.Added, changed, stats.ErrorCount, stats.ErrorMsg) } // CompletePrune commits a successful PRUNE task and schedules the next one. diff --git a/internal/syncer/connector/registry.go b/internal/syncer/connector/registry.go index 6608fcc003..eb0db2287f 100644 --- a/internal/syncer/connector/registry.go +++ b/internal/syncer/connector/registry.go @@ -25,7 +25,7 @@ import ( // Factory creates a connector for a task context. type Factory func(ctx context.Context, taskContext any) (Connector, error) -// Registry maps Python connector source names to Go connector factories. +// Registry connector source names to Go connector factories. type Registry struct { mu sync.RWMutex factories map[string]Factory diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go index e87e0e96c4..388e14cd82 100644 --- a/internal/syncer/syncer_test.go +++ b/internal/syncer/syncer_test.go @@ -794,7 +794,12 @@ func TestHash128MatchesPythonGolden(t *testing.T) { func TestFingerprintSkipsUnchangedDocument(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 + if err := db.Model(&entity.SyncLogs{}).Where("id = ?", "task-1").Updates(map[string]any{ + "status": dao.SyncStatusRunning, + "total_docs_indexed": int64(1), + }).Error; err != nil { + t.Fatalf("set running task: %v", err) + } taskService := service.NewSyncTaskService(dao.NewSyncTaskDAO(db)) legacyID := service.Hash128("conn-1:source-1") store := fakeStore{ids: map[string]struct{}{legacyID: {}}, fingerprints: map[string]string{legacyID: "fp-1"}} @@ -808,6 +813,20 @@ func TestFingerprintSkipsUnchangedDocument(t *testing.T) { if sink.callCount() != 0 { t.Fatalf("sink calls = %d, want 0", sink.callCount()) } + var completed entity.SyncLogs + if err := db.First(&completed, "id = ?", "task-1").Error; err != nil { + t.Fatalf("load completed task: %v", err) + } + if completed.NewDocsIndexed != 0 || completed.TotalDocsIndexed != 0 { + t.Fatalf("completed stats new/total = %d/%d, want 0/0", completed.NewDocsIndexed, completed.TotalDocsIndexed) + } + var next entity.SyncLogs + if err := db.Where("id <> ? AND connector_id = ? AND kb_id = ? AND task_type = ? AND status = ?", "task-1", "conn-1", "kb-1", dao.TaskTypeSync, dao.SyncStatusSchedule).First(&next).Error; err != nil { + t.Fatalf("load next scheduled task: %v", err) + } + if next.TotalDocsIndexed != 1 { + t.Fatalf("next scheduled total docs indexed = %d, want 1", next.TotalDocsIndexed) + } } // TestAutoParseFlagFlowsToSink verifies connector2kb.auto_parse is preserved.