From a2e6ddb56d0234fedc3451c264174e5ddf842efb Mon Sep 17 00:00:00 2001
From: Haruko386 <tryeverypossible@163.com>
Date: Fri, 14 Aug 2026 17:44:10 +0800
Subject: [PATCH] feat[syncer]: add checkpoint resume run for RSS (#18257)

### Summary

As title

---------

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
---
 internal/syncer/connector/rss.go      | 39 +++++++++++++++++++++++--
 internal/syncer/connector/rss_test.go | 42 +++++++++++++++++++++++++++
 2 files changed, 79 insertions(+), 2 deletions(-)

diff --git a/internal/syncer/connector/rss.go b/internal/syncer/connector/rss.go
index dc626206e0..f49672feff 100644
--- a/internal/syncer/connector/rss.go
+++ b/internal/syncer/connector/rss.go
@@ -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
diff --git a/internal/syncer/connector/rss_test.go b/internal/syncer/connector/rss_test.go
index ad644a3f30..a293536e26 100644
--- a/internal/syncer/connector/rss_test.go
+++ b/internal/syncer/connector/rss_test.go
@@ -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"?>