From d77b514d3e01619fd4f6fd2656de8768d4262e33 Mon Sep 17 00:00:00 2001 From: Haruko386 Date: Fri, 14 Aug 2026 17:44:25 +0800 Subject: [PATCH] feat[syncer]: add checkpoint resume run for Github (#18260) ### Summary As title --------- Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> --- internal/syncer/connector/github.go | 177 ++++++++++++++++++++--- internal/syncer/connector/github_test.go | 117 +++++++++++++++ 2 files changed, 273 insertions(+), 21 deletions(-) diff --git a/internal/syncer/connector/github.go b/internal/syncer/connector/github.go index 7984c192c3..323f38c385 100644 --- 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"` diff --git a/internal/syncer/connector/github_test.go b/internal/syncer/connector/github_test.go index 18aa5915cf..afd40ede6d 100644 --- 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()