feat[syncer]: add checkpoint resume run for Github (#18260)

### Summary As title --------- Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
2026-08-15 13:14:28 +08:00 · 2026-08-14 17:44:25 +08:00
parent a2e6ddb56d
commit d77b514d3e
2 changed files with 273 additions and 21 deletions
--- a/internal/syncer/connector/github.go
+++ b/internal/syncer/connector/github.go
@@ -99,7 +99,9 @@ func (c *GitHubConnector) OpenSync(ctx context.Context, request SyncRequest) (Sy
 	if err != nil {
 		return nil, err
 	}
-	return &githubSyncSession{connector: c, repos: repos, batchSize: c.batchSize, stage: githubStagePRs, page: 1, windowStart: request.WindowStart, windowEnd: request.WindowEnd}, nil
+	session := &githubSyncSession{connector: c, repos: repos, batchSize: c.batchSize, stage: githubStagePRs, page: 1, windowStart: request.WindowStart, windowEnd: request.WindowEnd}
+	session.applyResume(request.Resume)
+	return session, nil
 }

 // OpenPrune opens one complete GitHub prune snapshot session.
@@ -151,14 +153,15 @@ func (c *GitHubConnector) listRepoEndpoint(ctx context.Context, path string) ([]
 }

 // listPullRequestPage returns one page of GitHub pull requests.
-func (c *GitHubConnector) listPullRequestPage(ctx context.Context, fullName string, page, pageSize int, windowStart *time.Time, windowEnd time.Time) ([]SourceDocument, bool, error) {
+func (c *GitHubConnector) listPullRequestPage(ctx context.Context, fullName string, page, pageSize int, windowStart *time.Time, windowEnd time.Time) ([]githubBufferedDocument, bool, error) {
 	var batch []githubPullRequest
 	headers, err := c.getJSON(ctx, c.apiURL("/repos/"+fullName+"/pulls", githubListQuery(page, pageSize)), &batch)
 	if err != nil {
 		return nil, false, err
 	}
-	documents := make([]SourceDocument, 0, len(batch))
+	documents := make([]githubBufferedDocument, 0, len(batch))
 	doneByWindow := false
+	pageOffset := 0
 	for _, pr := range batch {
 		if beforeOrAtWindowStart(pr.UpdatedAt, windowStart) {
 			doneByWindow = true
@@ -167,21 +170,29 @@ func (c *GitHubConnector) listPullRequestPage(ctx context.Context, fullName stri
 		if afterWindowEnd(pr.UpdatedAt, windowEnd) {
 			continue
 		}
-		documents = append(documents, pr.toSourceDocument(fullName))
+		doc := pr.toSourceDocument(fullName)
+		pageOffset++
+		documents = append(documents, githubBufferedDocument{
+			document:   doc,
+			checkpoint: githubSyncCheckpoint(fullName, githubStagePRs, page, pageOffset, doc),
+			offset:     pageOffset,
+			sourceID:   doc.SourceID,
+		})
 	}
 	done := doneByWindow || !hasNextPage(headers) || len(batch) == 0
 	return documents, done, nil
 }

 // listIssuePage returns one page of GitHub issues.
-func (c *GitHubConnector) listIssuePage(ctx context.Context, fullName string, page, pageSize int, windowStart *time.Time, windowEnd time.Time) ([]SourceDocument, bool, error) {
+func (c *GitHubConnector) listIssuePage(ctx context.Context, fullName string, page, pageSize int, windowStart *time.Time, windowEnd time.Time) ([]githubBufferedDocument, bool, error) {
 	var batch []githubIssue
 	headers, err := c.getJSON(ctx, c.apiURL("/repos/"+fullName+"/issues", githubListQuery(page, pageSize)), &batch)
 	if err != nil {
 		return nil, false, err
 	}
-	documents := make([]SourceDocument, 0, len(batch))
+	documents := make([]githubBufferedDocument, 0, len(batch))
 	doneByWindow := false
+	pageOffset := 0
 	for _, issue := range batch {
 		if issue.PullRequest != nil {
 			continue
@@ -193,7 +204,14 @@ func (c *GitHubConnector) listIssuePage(ctx context.Context, fullName string, pa
 		if afterWindowEnd(issue.UpdatedAt, windowEnd) {
 			continue
 		}
-		documents = append(documents, issue.toSourceDocument(fullName))
+		doc := issue.toSourceDocument(fullName)
+		pageOffset++
+		documents = append(documents, githubBufferedDocument{
+			document:   doc,
+			checkpoint: githubSyncCheckpoint(fullName, githubStageIssues, page, pageOffset, doc),
+			offset:     pageOffset,
+			sourceID:   doc.SourceID,
+		})
 	}
 	done := doneByWindow || !hasNextPage(headers) || len(batch) == 0
 	return documents, done, nil
@@ -287,26 +305,35 @@ func (c *GitHubConnector) apiURL(path string, query url.Values) string {
 }

 type githubSyncSession struct {
-	connector   *GitHubConnector
-	repos       []githubRepo
-	repoIndex   int
-	stage       string
-	page        int
-	batchSize   int
-	windowStart *time.Time
-	windowEnd   time.Time
-	buffer      []SourceDocument
+	connector      *GitHubConnector
+	repos          []githubRepo
+	repoIndex      int
+	stage          string
+	page           int
+	batchSize      int
+	windowStart    *time.Time
+	windowEnd      time.Time
+	buffer         []githubBufferedDocument
+	resumeRepo     string
+	resumeStage    string
+	resumePage     int
+	resumeOffset   int
+	resumeSourceID string
 }

 // NextBatch returns the next GitHub document batch.
 func (s *githubSyncSession) NextBatch(ctx context.Context) (SyncBatch, error) {
 	documents := make([]SourceDocument, 0, s.batchSize)
+	var checkpoint *SyncCheckpoint
 	if len(s.buffer) > 0 {
 		n := s.batchSize
 		if n > len(s.buffer) {
 			n = len(s.buffer)
 		}
-		documents = append(documents, s.buffer[:n]...)
+		for _, buffered := range s.buffer[:n] {
+			documents = append(documents, buffered.document)
+			checkpoint = buffered.checkpoint
+		}
 		s.buffer = s.buffer[n:]
 	}

@@ -323,13 +350,19 @@ func (s *githubSyncSession) NextBatch(ctx context.Context) (SyncBatch, error) {
 		}
 		remaining := s.batchSize - len(documents)
 		if len(batch) > remaining {
-			documents = append(documents, batch[:remaining]...)
+			for _, buffered := range batch[:remaining] {
+				documents = append(documents, buffered.document)
+				checkpoint = buffered.checkpoint
+			}
 			s.buffer = append(s.buffer, batch[remaining:]...)
 			break
 		}
-		documents = append(documents, batch...)
+		for _, buffered := range batch {
+			documents = append(documents, buffered.document)
+			checkpoint = buffered.checkpoint
+		}
 	}
-	return SyncBatch{Documents: documents}, nil
+	return SyncBatch{Documents: documents, Checkpoint: checkpoint}, nil
 }

 // Close closes the GitHub sync session.
@@ -391,7 +424,7 @@ const (
 )

 // nextDocumentPage fetches one GitHub API page for sync.
-func (s *githubSyncSession) nextDocumentPage(ctx context.Context) ([]SourceDocument, error) {
+func (s *githubSyncSession) nextDocumentPage(ctx context.Context) ([]githubBufferedDocument, error) {
 	repo := s.repos[s.repoIndex]
 	switch s.stage {
 	case githubStagePRs:
@@ -403,6 +436,7 @@ func (s *githubSyncSession) nextDocumentPage(ctx context.Context) ([]SourceDocum
 		if err != nil {
 			return nil, err
 		}
+		docs = s.filterResumedDocuments(repo.FullName, githubStagePRs, s.page, docs)
 		if done {
 			s.advanceStage()
 		} else {
@@ -418,6 +452,7 @@ func (s *githubSyncSession) nextDocumentPage(ctx context.Context) ([]SourceDocum
 		if err != nil {
 			return nil, err
 		}
+		docs = s.filterResumedDocuments(repo.FullName, githubStageIssues, s.page, docs)
 		if done {
 			s.advanceRepo()
 		} else {
@@ -474,6 +509,7 @@ func (s *githubPruneSession) nextSlimPage(ctx context.Context) ([]SlimDocument,
 func (s *githubSyncSession) advanceStage() {
 	s.stage = githubStageIssues
 	s.page = 1
+	s.clearResume()
 }

 // advanceRepo moves a GitHub session to the next repository.
@@ -481,6 +517,71 @@ func (s *githubSyncSession) advanceRepo() {
 	s.repoIndex++
 	s.stage = githubStagePRs
 	s.page = 1
+	s.clearResume()
+}
+
+// applyResume advances a sync session to the last committed GitHub page.
+func (s *githubSyncSession) applyResume(checkpoint *SyncCheckpoint) {
+	if checkpoint == nil || checkpoint.Cursor == "" {
+		return
+	}
+
+	var cursor githubSyncCursor
+	if err := json.Unmarshal([]byte(checkpoint.Cursor), &cursor); err != nil {
+		return
+	}
+	if cursor.Repo == "" || cursor.Stage == "" || cursor.Page <= 0 {
+		return
+	}
+	for index, repo := range s.repos {
+		if repo.FullName != cursor.Repo {
+			continue
+		}
+		s.repoIndex = index
+		s.stage = cursor.Stage
+		s.page = cursor.Page
+		s.resumeRepo = cursor.Repo
+		s.resumeStage = cursor.Stage
+		s.resumePage = cursor.Page
+		s.resumeOffset = cursor.Offset
+		s.resumeSourceID = firstNonEmpty(cursor.SourceID, checkpoint.SourceID)
+		return
+	}
+}
+
+// filterResumedDocuments drops documents through the committed checkpoint.
+func (s *githubSyncSession) filterResumedDocuments(repo, stage string, page int, candidates []githubBufferedDocument) []githubBufferedDocument {
+	if s.resumeRepo == "" || repo != s.resumeRepo || stage != s.resumeStage || page != s.resumePage {
+		return candidates
+	}
+	if s.resumeSourceID != "" {
+		for index, candidate := range candidates {
+			if candidate.sourceID == s.resumeSourceID {
+				s.clearResume()
+				return candidates[index+1:]
+			}
+		}
+	}
+	if s.resumeOffset <= 0 {
+		s.clearResume()
+		return candidates
+	}
+	filtered := candidates[:0]
+	for _, candidate := range candidates {
+		if candidate.offset > s.resumeOffset {
+			filtered = append(filtered, candidate)
+		}
+	}
+	s.clearResume()
+	return filtered
+}
+
+func (s *githubSyncSession) clearResume() {
+	s.resumeRepo = ""
+	s.resumeStage = ""
+	s.resumePage = 0
+	s.resumeOffset = 0
+	s.resumeSourceID = ""
 }

 // advanceStage moves a GitHub prune session from PRs to issues.
@@ -500,6 +601,40 @@ type githubRepo struct {
 	FullName string `json:"full_name"`
 }

+type githubSyncCursor struct {
+	Repo     string `json:"repo"`
+	Stage    string `json:"stage"`
+	Page     int    `json:"page"`
+	Offset   int    `json:"offset"`
+	SourceID string `json:"source_id"`
+}
+
+type githubBufferedDocument struct {
+	document   SourceDocument
+	checkpoint *SyncCheckpoint
+	offset     int
+	sourceID   string
+}
+
+func githubSyncCheckpoint(repo, stage string, page, offset int, doc SourceDocument) *SyncCheckpoint {
+	cursor, err := json.Marshal(githubSyncCursor{
+		Repo:     repo,
+		Stage:    stage,
+		Page:     page,
+		Offset:   offset,
+		SourceID: doc.SourceID,
+	})
+	if err != nil {
+		return nil
+	}
+	updatedAt := doc.UpdatedAt
+	return &SyncCheckpoint{
+		Cursor:    string(cursor),
+		SourceID:  doc.SourceID,
+		UpdatedAt: &updatedAt,
+	}
+}
+
 type githubPullRequest struct {
 	HTMLURL   string        `json:"html_url"`
 	Number    int           `json:"number"`
--- a/internal/syncer/connector/github_test.go
+++ b/internal/syncer/connector/github_test.go
@@ -134,6 +134,123 @@ func TestGitHubConnectorOpenPrune(t *testing.T) {
 	}
 }

+// TestGitHubConnectorOpenSyncResumesAfterCheckpoint verifies retry skips committed GitHub documents.
+func TestGitHubConnectorOpenSyncResumesAfterCheckpoint(t *testing.T) {
+	connector, err := NewGitHubConnector(map[string]any{
+		"repository_owner":      "openai",
+		"repository_name":       "ragflow",
+		"include_pull_requests": true,
+		"include_issues":        true,
+		"batch_size":            1,
+		"credentials":           map[string]any{"github_access_token": "token"},
+	})
+	if err != nil {
+		t.Fatalf("NewGitHubConnector failed: %v", err)
+	}
+	connector.baseURL = "https://api.github.test"
+	connector.doJSON = githubFixtureDoJSON(t)
+
+	end := mustTime(t, "2026-01-04T00:00:00Z")
+	session, err := connector.OpenSync(context.Background(), SyncRequest{FromBeginning: true, WindowEnd: end})
+	if err != nil {
+		t.Fatalf("OpenSync failed: %v", err)
+	}
+	first, err := session.NextBatch(context.Background())
+	if err != nil {
+		t.Fatalf("NextBatch first failed: %v", err)
+	}
+	if len(first.Documents) != 1 || first.Documents[0].SourceID != "https://github.com/openai/ragflow/pull/7" {
+		t.Fatalf("first documents = %+v, want PR 7", first.Documents)
+	}
+	if first.Checkpoint == nil || first.Checkpoint.SourceID != "https://github.com/openai/ragflow/pull/7" {
+		t.Fatalf("first checkpoint = %+v, want PR 7", first.Checkpoint)
+	}
+
+	resumed, err := connector.OpenSync(context.Background(), SyncRequest{FromBeginning: true, WindowEnd: end, Resume: first.Checkpoint})
+	if err != nil {
+		t.Fatalf("resume OpenSync failed: %v", err)
+	}
+	second, err := resumed.NextBatch(context.Background())
+	if err != nil {
+		t.Fatalf("resume NextBatch failed: %v", err)
+	}
+	if len(second.Documents) != 1 || second.Documents[0].SourceID != "https://github.com/openai/ragflow/issues/3" {
+		t.Fatalf("resume documents = %+v, want issue 3", second.Documents)
+	}
+	if second.Checkpoint == nil || second.Checkpoint.SourceID != "https://github.com/openai/ragflow/issues/3" {
+		t.Fatalf("resume checkpoint = %+v, want issue 3", second.Checkpoint)
+	}
+	if _, err = resumed.NextBatch(context.Background()); !errors.Is(err, io.EOF) {
+		t.Fatalf("resume EOF = %v", err)
+	}
+}
+
+// TestGitHubConnectorOpenSyncResumeOffsetFallbackSkipsCommittedDocument verifies offset fallback excludes the committed offset.
+func TestGitHubConnectorOpenSyncResumeOffsetFallbackSkipsCommittedDocument(t *testing.T) {
+	connector, err := NewGitHubConnector(map[string]any{
+		"repository_owner":      "openai",
+		"repository_name":       "ragflow",
+		"include_pull_requests": true,
+		"include_issues":        true,
+		"batch_size":            1,
+		"credentials":           map[string]any{"github_access_token": "token"},
+	})
+	if err != nil {
+		t.Fatalf("NewGitHubConnector failed: %v", err)
+	}
+	connector.baseURL = "https://api.github.test"
+	connector.doJSON = githubFixtureDoJSON(t)
+
+	end := mustTime(t, "2026-01-04T00:00:00Z")
+	session, err := connector.OpenSync(context.Background(), SyncRequest{FromBeginning: true, WindowEnd: end})
+	if err != nil {
+		t.Fatalf("OpenSync failed: %v", err)
+	}
+	first, err := session.NextBatch(context.Background())
+	if err != nil {
+		t.Fatalf("NextBatch first failed: %v", err)
+	}
+	if len(first.Documents) != 1 || first.Documents[0].SourceID != "https://github.com/openai/ragflow/pull/7" {
+		t.Fatalf("first documents = %+v, want PR 7", first.Documents)
+	}
+	if first.Checkpoint == nil {
+		t.Fatalf("first checkpoint is nil")
+	}
+
+	resumeCheckpoint := cloneGitHubCheckpointWithMissingSourceID(t, first.Checkpoint)
+	resumed, err := connector.OpenSync(context.Background(), SyncRequest{FromBeginning: true, WindowEnd: end, Resume: resumeCheckpoint})
+	if err != nil {
+		t.Fatalf("resume OpenSync failed: %v", err)
+	}
+	second, err := resumed.NextBatch(context.Background())
+	if err != nil {
+		t.Fatalf("resume NextBatch failed: %v", err)
+	}
+	if len(second.Documents) != 1 || second.Documents[0].SourceID != "https://github.com/openai/ragflow/issues/3" {
+		t.Fatalf("resume documents = %+v, want issue 3", second.Documents)
+	}
+	if second.Documents[0].SourceID == first.Documents[0].SourceID {
+		t.Fatalf("committed document was redelivered: %s", second.Documents[0].SourceID)
+	}
+}
+
+func cloneGitHubCheckpointWithMissingSourceID(t *testing.T, checkpoint *SyncCheckpoint) *SyncCheckpoint {
+	t.Helper()
+	var cursor githubSyncCursor
+	if err := json.Unmarshal([]byte(checkpoint.Cursor), &cursor); err != nil {
+		t.Fatalf("decode checkpoint cursor: %v", err)
+	}
+	cursor.SourceID = ""
+	data, err := json.Marshal(cursor)
+	if err != nil {
+		t.Fatalf("encode checkpoint cursor: %v", err)
+	}
+	clone := *checkpoint
+	clone.Cursor = string(data)
+	clone.SourceID = ""
+	return &clone
+}
+
 // githubFixtureDoJSON returns a fixture GitHub JSON transport.
 func githubFixtureDoJSON(t *testing.T) func(ctx context.Context, apiURL string, out any) (http.Header, error) {
 	t.Helper()