feat[syncer]: add checkpoint resume run for RSS (#18257)

### Summary

As title

---------

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
This commit is contained in:
Haruko386
2026-08-14 17:44:10 +08:00
committed by GitHub
parent cdbd82df4f
commit a2e6ddb56d
2 changed files with 79 additions and 2 deletions

View File

@@ -117,7 +117,9 @@ func (c *RSSConnector) OpenSync(ctx context.Context, request SyncRequest) (SyncS
}
documents = append(documents, entry.toSourceDocument(c.feedURL))
}
return &rssSyncSession{documents: documents, batchSize: c.batchSize}, nil
session := &rssSyncSession{documents: documents, batchSize: c.batchSize}
session.applyResume(request.Resume)
return session, nil
}
// OpenPrune opens one complete RSS prune snapshot session.
@@ -192,7 +194,8 @@ func (s *rssSyncSession) NextBatch(ctx context.Context) (SyncBatch, error) {
if end > len(s.documents) {
end = len(s.documents)
}
batch := SyncBatch{Documents: s.documents[s.index:end]}
batchDocuments := s.documents[s.index:end]
batch := SyncBatch{Documents: batchDocuments, Checkpoint: rssSyncCheckpoint(batchDocuments[len(batchDocuments)-1])}
s.index = end
return batch, nil
}
@@ -202,6 +205,38 @@ func (s *rssSyncSession) Close() error {
return nil
}
// applyResume advances past the last committed RSS document when retrying a task.
func (s *rssSyncSession) applyResume(checkpoint *SyncCheckpoint) {
if checkpoint == nil {
return
}
sourceID := firstNonEmpty(checkpoint.SourceID, checkpoint.Cursor)
if sourceID != "" {
for index, doc := range s.documents {
if doc.SourceID == sourceID {
s.index = index + 1
return
}
}
}
if checkpoint.UpdatedAt == nil {
return
}
for s.index < len(s.documents) && s.documents[s.index].UpdatedAt.Before(*checkpoint.UpdatedAt) {
s.index++
}
}
// rssSyncCheckpoint returns a resume point after a committed RSS document.
func rssSyncCheckpoint(doc SourceDocument) *SyncCheckpoint {
updatedAt := doc.UpdatedAt
return &SyncCheckpoint{
Cursor: doc.SourceID,
SourceID: doc.SourceID,
UpdatedAt: &updatedAt,
}
}
type rssPruneSession struct {
documents []SlimDocument
batchSize int

View File

@@ -95,6 +95,48 @@ func TestRSSConnectorOpenPrune(t *testing.T) {
}
}
// TestRSSConnectorOpenSyncResumesAfterCheckpoint verifies retry resumes after committed entries.
func TestRSSConnectorOpenSyncResumesAfterCheckpoint(t *testing.T) {
connector, err := NewRSSConnector(map[string]any{"feed_url": "https://example.com/feed.xml", "batch_size": 2})
if err != nil {
t.Fatalf("NewRSSConnector failed: %v", err)
}
connector.fetchFeed = staticRSSFeed
session, err := connector.OpenSync(context.Background(), SyncRequest{FromBeginning: true, WindowEnd: mustTime(t, "2026-01-07T00:00:00Z")})
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) != 2 {
t.Fatalf("first batch len = %d, want 2", len(first.Documents))
}
if first.Checkpoint == nil || first.Checkpoint.SourceID != expectedRSSSourceID("entry-new") {
t.Fatalf("first checkpoint = %+v, want entry-new", first.Checkpoint)
}
resumed, err := connector.OpenSync(context.Background(), SyncRequest{FromBeginning: true, WindowEnd: mustTime(t, "2026-01-07T00:00:00Z"), 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 != expectedRSSSourceID("entry-future") {
t.Fatalf("resume documents = %+v, want entry-future", second.Documents)
}
if second.Checkpoint == nil || second.Checkpoint.SourceID != expectedRSSSourceID("entry-future") {
t.Fatalf("resume checkpoint = %+v, want entry-future", second.Checkpoint)
}
if _, err = resumed.NextBatch(context.Background()); !errors.Is(err, io.EOF) {
t.Fatalf("resume EOF = %v", err)
}
}
// staticRSSFeed returns a fixed RSS feed for tests.
func staticRSSFeed(ctx context.Context, feedURL string) ([]byte, error) {
return []byte(`<?xml version="1.0" encoding="UTF-8"?>