fix(datasource): optimize RSS incremental sync and surface partial failures

Skip article-page fetches when feed signals are unchanged, preserve sync
cursor across transient feed outages, and record partial multi-feed syncs.

internal/application/service/datasource_service.go

@@ -9,6 +9,7 @@ import (
 	"mime/multipart"
 	"net/textproto"
 	"reflect"
+	"strings"
 	"time"
 
 	"github.com/Tencent/WeKnora/internal/datasource"
@@ -639,7 +640,24 @@ func (s *DataSourceService) ProcessSync(ctx context.Context, task *asynq.Task) e
 		logger.Infof(ctx, "incremental sync fetched %d items", len(items))
 	}
 
+	var fetchWarnings []string
+	var partialFetch *datasource.PartialFetchError
+	if errors.As(fetchErr, &partialFetch) {
+		fetchWarnings = partialFetch.Details
+		fetchErr = nil
+	}
+
 	if fetchErr != nil {
+		// Persist connector cursor even when fetch failed so transient outages
+		// (e.g. RSS feed downtime) do not force a full re-ingest on recovery.
+		if nextCursor != nil {
+			if cursorJSON, cerr := nextCursor.ToJSON(); cerr == nil {
+				ds.LastSyncCursor = cursorJSON
+				if uerr := s.dsRepo.UpdateSyncState(ctx, ds); uerr != nil {
+					logger.Warnf(ctx, "failed to persist sync cursor after fetch error: %v", uerr)
+				}
+			}
+		}
 		logger.Errorf(ctx, "fetch operation failed: %v", fetchErr)
 		syncLog.Status = types.SyncLogStatusFailed
 		syncLog.FinishedAt = timePtr(time.Now().UTC())
@@ -744,7 +762,17 @@ func (s *DataSourceService) ProcessSync(ctx context.Context, task *asynq.Task) e
 	}
 
 	ds.LastSyncAt = timePtr(time.Now().UTC())
-	s.updateSyncRunResult(ctx, ds, syncLog, result, resultJSON, types.SyncLogStatusSuccess, "", wasPaused)
+	syncStatus := types.SyncLogStatusSuccess
+	syncErrorMessage := ""
+	if len(fetchWarnings) > 0 {
+		syncStatus = types.SyncLogStatusPartial
+		syncErrorMessage = fmt.Sprintf("Some feeds failed: %s", strings.Join(fetchWarnings, "; "))
+		for _, w := range fetchWarnings {
+			result.Errors = append(result.Errors, w)
+		}
+		resultJSON, _ = result.ToJSON()
+	}
+	s.updateSyncRunResult(ctx, ds, syncLog, result, resultJSON, syncStatus, syncErrorMessage, wasPaused)
 
 	logger.Infof(ctx, "data source sync completed: ds=%s created=%d updated=%d deleted=%d",
 		payload.DataSourceID, syncLog.ItemsCreated, syncLog.ItemsUpdated, syncLog.ItemsDeleted)

internal/datasource/connector/rss/connector.go

@@ -144,29 +144,37 @@ func (c *Connector) FetchIncremental(
 	}
 
 	items, newCursor, err := c.walk(ctx, config, config.ResourceIDs, prev, true)
-	if err != nil {
+	if err != nil && newCursor == nil {
 		return nil, nil, err
 	}
 
 	cursorMap := make(map[string]interface{})
-	b, err := json.Marshal(newCursor)
-	if err != nil {
-		logger.Warnf(ctx, "[RSS] marshal new cursor: %v", err)
-	} else if err := json.Unmarshal(b, &cursorMap); err != nil {
-		logger.Warnf(ctx, "[RSS] unmarshal new cursor to map: %v", err)
+	if newCursor != nil {
+		b, marshalErr := json.Marshal(newCursor)
+		if marshalErr != nil {
+			logger.Warnf(ctx, "[RSS] marshal new cursor: %v", marshalErr)
+		} else if unmarshalErr := json.Unmarshal(b, &cursorMap); unmarshalErr != nil {
+			logger.Warnf(ctx, "[RSS] unmarshal new cursor to map: %v", unmarshalErr)
+		}
 	}
 
-	return items, &types.SyncCursor{
-		LastSyncTime:    newCursor.LastSyncTime,
-		ConnectorCursor: cursorMap,
-	}, nil
+	var syncCursor *types.SyncCursor
+	if newCursor != nil {
+		syncCursor = &types.SyncCursor{
+			LastSyncTime:    newCursor.LastSyncTime,
+			ConnectorCursor: cursorMap,
+		}
+	}
+
+	return items, syncCursor, err
 }
 
 // walk is the shared implementation for FetchAll / FetchIncremental.
 //
-// When incremental is true, items whose content fingerprint matches the prior
-// cursor are omitted from the result (ingest skipped). The returned cursor
-// always reflects the full current item set so a later sync can detect changes.
+// When incremental is true, items whose feed signal and content fingerprint are
+// both unchanged are omitted from the result (ingest skipped) without fetching
+// article pages. The returned cursor always reflects the best-known item state
+// so a later sync can detect changes.
 func (c *Connector) walk(
 	ctx context.Context,
 	config *types.DataSourceConfig,
@@ -191,6 +199,7 @@ func (c *Connector) walk(
 	newCursor := &rssCursor{
 		LastSyncTime: time.Now().UTC(),
 		FeedItems:    make(map[string]map[string]string),
+		FeedSignals:  make(map[string]map[string]string),
 	}
 	var out []types.FetchedItem
 	var feedErrors []string
@@ -200,21 +209,26 @@ func (c *Connector) walk(
 		if err != nil {
 			logger.Warnf(ctx, "[RSS] sync: fetch %s failed: %v", feedURL, err)
 			feedErrors = append(feedErrors, fmt.Sprintf("%s: %v", feedURL, err))
-			newCursor.FeedItems[feedURL] = make(map[string]string)
+			copyFeedCursor(newCursor, prev, feedURL)
 			continue
 		}
 		feed, err := parser.Parse(bytes.NewReader(data))
 		if err != nil {
 			logger.Warnf(ctx, "[RSS] sync: parse %s failed: %v", feedURL, err)
 			feedErrors = append(feedErrors, fmt.Sprintf("%s: %v", feedURL, err))
-			newCursor.FeedItems[feedURL] = make(map[string]string)
+			copyFeedCursor(newCursor, prev, feedURL)
 			continue
 		}
 
 		newCursor.FeedItems[feedURL] = make(map[string]string)
+		newCursor.FeedSignals[feedURL] = make(map[string]string)
 		var prevItems map[string]string
+		var prevSignals map[string]string
 		if incremental && prev != nil {
 			prevItems = prev.FeedItems[feedURL]
+			if prev.FeedSignals != nil {
+				prevSignals = prev.FeedSignals[feedURL]
+			}
 		}
 
 		var kept, skipped int
@@ -228,8 +242,25 @@ func (c *Connector) walk(
 			}
 
 			feedContent := firstNonEmpty(item.Content, item.Description)
+			feedSig := feedSignalFingerprint(item, feedContent)
+
+			if incremental && prevItems != nil {
+				prevFP := prevItems[itemID]
+				prevSig := ""
+				if prevSignals != nil {
+					prevSig = prevSignals[itemID]
+				}
+				if prevFP != "" && feedSig == prevSig {
+					newCursor.FeedItems[feedURL][itemID] = prevFP
+					newCursor.FeedSignals[feedURL][itemID] = feedSig
+					skipped++
+					continue
+				}
+			}
+
 			resolved := c.resolveItem(ctx, cli, feed, item, feedURL, itemID, feedContent)
 			newCursor.FeedItems[feedURL][itemID] = resolved.fingerprint
+			newCursor.FeedSignals[feedURL][itemID] = feedSig
 
 			if incremental && prevItems != nil && prevItems[itemID] == resolved.fingerprint {
 				skipped++
@@ -243,8 +274,11 @@ func (c *Connector) walk(
 			feedURL, len(feed.Items), kept, skipped)
 	}
 
-	if len(out) == 0 && len(feedErrors) > 0 && len(feedErrors) == len(feedURLs) {
-		return nil, nil, fmt.Errorf("all feeds failed: %s", strings.Join(feedErrors, "; "))
+	if len(feedErrors) > 0 {
+		if len(out) == 0 && len(feedErrors) == len(feedURLs) {
+			return nil, newCursor, fmt.Errorf("all feeds failed: %s", strings.Join(feedErrors, "; "))
+		}
+		return out, newCursor, &datasource.PartialFetchError{Details: feedErrors}
 	}
 
 	return out, newCursor, nil

internal/datasource/connector/rss/connector_test.go

@@ -2,15 +2,20 @@ package rss
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"os"
 	"strings"
+	"sync/atomic"
 	"testing"
 
+	"github.com/Tencent/WeKnora/internal/datasource"
 	"github.com/Tencent/WeKnora/internal/types"
 	"github.com/Tencent/WeKnora/internal/utils"
+	"github.com/mmcdole/gofeed"
 )
 
 // TestMain whitelists loopback for SSRF so the httptest servers (127.0.0.1)
@@ -36,6 +41,8 @@ type fakeFeed struct {
 	feedTitle          string
 	itemContent        string // optional <description>/content for items
 	articleAuthHeaders []string
+	articleFetches     atomic.Int32
+	failFeed           atomic.Bool
 }
 
 func newFakeFeed(t *testing.T) *fakeFeed {
@@ -44,6 +51,7 @@ func newFakeFeed(t *testing.T) *fakeFeed {
 	mux := http.NewServeMux()
 
 	mux.HandleFunc("/article/", func(w http.ResponseWriter, r *http.Request) {
+		f.articleFetches.Add(1)
 		for k, vals := range r.Header {
 			if strings.EqualFold(k, "X-Test-Auth") && len(vals) > 0 && vals[0] != "" {
 				f.articleAuthHeaders = append(f.articleAuthHeaders, vals[0])
@@ -56,6 +64,10 @@ func newFakeFeed(t *testing.T) *fakeFeed {
 	})
 
 	mux.HandleFunc("/feed.xml", func(w http.ResponseWriter, r *http.Request) {
+		if f.failFeed.Load() {
+			w.WriteHeader(http.StatusServiceUnavailable)
+			return
+		}
 		if auth := r.Header.Get("X-Test-Auth"); f.itemContent == "needs-auth" && auth != "secret" {
 			w.WriteHeader(http.StatusUnauthorized)
 			return
@@ -168,6 +180,19 @@ func TestConfig_ParseHeaders(t *testing.T) {
 	}
 }
 
+func TestFeedSignalFingerprint(t *testing.T) {
+	item := &gofeed.Item{GUID: "g1", Link: "https://example.com/a", Title: "t"}
+	sig1 := feedSignalFingerprint(item, "body")
+	sig2 := feedSignalFingerprint(item, "body")
+	if sig1 == "" || sig1 != sig2 {
+		t.Fatalf("feed signal unstable: %q vs %q", sig1, sig2)
+	}
+	sig3 := feedSignalFingerprint(item, "changed")
+	if sig1 == sig3 {
+		t.Fatal("expected different signal when feed content changes")
+	}
+}
+
 func TestItemExternalID_ScopesByFeed(t *testing.T) {
 	got := itemExternalID("https://a.com/feed", "guid-1")
 	want := "https://a.com/feed:guid-1"
@@ -269,6 +294,35 @@ func TestConnector_FetchAll_FullTextMarkdown(t *testing.T) {
 	}
 }
 
+func TestConnector_FetchIncremental_SkipsWithoutArticleFetch(t *testing.T) {
+	f := newFakeFeed(t)
+	cfg := makeConfig(f.feedURL(), "")
+	cfg.ResourceIDs = []string{f.feedURL()}
+
+	items, cursor, err := NewConnector().FetchIncremental(context.Background(), cfg, nil)
+	if err != nil {
+		t.Fatalf("first FetchIncremental error: %v", err)
+	}
+	if len(items) != 2 {
+		t.Fatalf("expected 2 items on first sync, got %d", len(items))
+	}
+	if got := f.articleFetches.Load(); got != 2 {
+		t.Fatalf("first sync article fetches = %d, want 2", got)
+	}
+
+	f.articleFetches.Store(0)
+	items2, _, err := NewConnector().FetchIncremental(context.Background(), cfg, cursor)
+	if err != nil {
+		t.Fatalf("second FetchIncremental error: %v", err)
+	}
+	if len(items2) != 0 {
+		t.Fatalf("expected 0 items on unchanged second sync, got %d", len(items2))
+	}
+	if got := f.articleFetches.Load(); got != 0 {
+		t.Fatalf("unchanged incremental sync must not refetch articles, got %d fetches", got)
+	}
+}
+
 func TestConnector_FetchIncremental_SkipsUnchanged(t *testing.T) {
 	f := newFakeFeed(t)
 	cfg := makeConfig(f.feedURL(), "")
@@ -300,14 +354,54 @@ func TestConnector_Walk_PartialFeedFailure(t *testing.T) {
 	f := newFakeFeed(t)
 	cfg := makeConfig(f.feedURL()+", https://invalid.invalid/feed.xml", "")
 	items, err := NewConnector().FetchAll(context.Background(), cfg, nil)
-	if err != nil {
-		t.Fatalf("expected partial success, got error: %v", err)
+	var partial *datasource.PartialFetchError
+	if !errors.As(err, &partial) {
+		t.Fatalf("expected PartialFetchError, got %v", err)
+	}
+	if len(partial.Details) != 1 {
+		t.Fatalf("expected 1 feed error, got %v", partial.Details)
 	}
 	if len(items) != 2 {
 		t.Fatalf("expected 2 items from healthy feed, got %d", len(items))
 	}
 }
 
+func TestConnector_Walk_PreservesCursorOnFeedFailure(t *testing.T) {
+	f := newFakeFeed(t)
+	cfg := makeConfig(f.feedURL(), "")
+	cfg.ResourceIDs = []string{f.feedURL()}
+
+	_, cursor, err := NewConnector().FetchIncremental(context.Background(), cfg, nil)
+	if err != nil {
+		t.Fatalf("first sync error: %v", err)
+	}
+
+	f.failFeed.Store(true)
+	_, newCursor, err := NewConnector().FetchIncremental(context.Background(), cfg, cursor)
+	if err == nil {
+		t.Fatal("expected error when sole feed is unavailable")
+	}
+	if newCursor == nil || newCursor.ConnectorCursor == nil {
+		t.Fatal("expected cursor to be preserved on feed failure")
+	}
+
+	var restored rssCursor
+	b, err := json.Marshal(newCursor.ConnectorCursor)
+	if err != nil {
+		t.Fatalf("marshal cursor: %v", err)
+	}
+	if err := json.Unmarshal(b, &restored); err != nil {
+		t.Fatalf("unmarshal cursor: %v", err)
+	}
+	items := restored.FeedItems[f.feedURL()]
+	if len(items) != 2 {
+		t.Fatalf("expected preserved fingerprints for 2 items, got %d", len(items))
+	}
+	if items["guid-1"] == "" || items["guid-2"] == "" {
+		t.Fatalf("expected non-empty preserved fingerprints, got %+v", items)
+	}
+}
+
 func TestConnector_ResolveResourceAncestors_Empty(t *testing.T) {
 	got, err := NewConnector().ResolveResourceAncestors(
 		context.Background(), makeConfig("https://a.com/f", ""), []string{"x"},

internal/datasource/connector/rss/types.go

@@ -12,10 +12,10 @@
 //     run through a readability extractor; the cleaned HTML is converted to
 //     Markdown. Feed-provided content (content:encoded / description) is used
 //     as a fallback.
-//   - Incremental: a per-item fingerprint hashes the final Markdown body that
-//     would be ingested, so feed-only metadata changes and full-article edits
-//     are both detected. Deletions are NOT synced — feeds routinely drop old
-//     items, which would otherwise look like deletions.
+//   - Incremental: feed-level signals skip full-text article fetches when a feed
+//     entry is unchanged; content fingerprints detect feed-body changes. Article-
+//     only edits without feed updates may be missed until the feed entry changes.
+//     Deletions are NOT synced — feeds routinely drop old items.
 //
 // All outbound requests go through the SSRF-safe HTTP client so a malicious
 // feed cannot redirect WeKnora to internal services.
@@ -26,12 +26,14 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"maps"
 	"strings"
 	"time"
 	"unicode/utf8"
 
 	"github.com/Tencent/WeKnora/internal/datasource"
 	"github.com/Tencent/WeKnora/internal/types"
+	"github.com/mmcdole/gofeed"
 )
 
 // Config holds RSS-specific configuration.
@@ -144,9 +146,12 @@ func (c *Config) parseHeaders() map[string]string {
 //
 // FeedItems maps feedURL → itemID → fingerprint, where fingerprint is a
 // "h:<sha256-prefix>" hash of the final Markdown body that would be ingested.
+// FeedSignals maps feedURL → itemID → feed-only signal used to skip full-text
+// article fetches when the feed entry itself has not changed.
 type rssCursor struct {
 	LastSyncTime time.Time                    `json:"last_sync_time"`
 	FeedItems    map[string]map[string]string `json:"feed_items,omitempty"`
+	FeedSignals  map[string]map[string]string `json:"feed_signals,omitempty"`
 }
 
 // contentFingerprint hashes ingested Markdown for incremental change detection.
@@ -155,6 +160,49 @@ func contentFingerprint(markdown string) string {
 	return "h:" + hex.EncodeToString(sum[:])[:16]
 }
 
+// feedSignalFingerprint hashes feed-visible fields so incremental sync can skip
+// article-page fetches when the entry has not changed in the feed.
+func feedSignalFingerprint(item *gofeed.Item, feedContent string) string {
+	if item == nil {
+		return ""
+	}
+	var b strings.Builder
+	b.WriteString(item.GUID)
+	b.WriteByte('\n')
+	b.WriteString(item.Link)
+	b.WriteByte('\n')
+	b.WriteString(item.Title)
+	b.WriteByte('\n')
+	if item.UpdatedParsed != nil && !item.UpdatedParsed.IsZero() {
+		b.WriteString(item.UpdatedParsed.UTC().Format(time.RFC3339))
+	}
+	b.WriteByte('\n')
+	if item.PublishedParsed != nil && !item.PublishedParsed.IsZero() {
+		b.WriteString(item.PublishedParsed.UTC().Format(time.RFC3339))
+	}
+	b.WriteByte('\n')
+	b.WriteString(feedContent)
+	sum := sha256.Sum256([]byte(b.String()))
+	return "s:" + hex.EncodeToString(sum[:])[:16]
+}
+
+func copyFeedCursor(dst, prev *rssCursor, feedURL string) {
+	if dst == nil || prev == nil {
+		return
+	}
+	if src, ok := prev.FeedItems[feedURL]; ok && len(src) > 0 {
+		dst.FeedItems[feedURL] = maps.Clone(src)
+	}
+	if prev.FeedSignals != nil {
+		if src, ok := prev.FeedSignals[feedURL]; ok && len(src) > 0 {
+			if dst.FeedSignals == nil {
+				dst.FeedSignals = make(map[string]map[string]string)
+			}
+			dst.FeedSignals[feedURL] = maps.Clone(src)
+		}
+	}
+}
+
 // itemExternalID scopes an item ID to its feed so GUIDs cannot collide across feeds.
 func itemExternalID(feedURL, itemID string) string {
 	return feedURL + ":" + itemID

internal/datasource/errors.go

@@ -1,6 +1,10 @@
 package datasource
 
-import "errors"
+import (
+	"errors"
+	"fmt"
+	"strings"
+)
 
 // Error definitions for datasource operations
 var (
@@ -19,10 +23,10 @@ var (
 	ErrInvalidCredentials = errors.New("invalid credentials")
 
 	// Sync errors
-	ErrSyncFailed         = errors.New("sync operation failed")
-	ErrSyncCanceled       = errors.New("sync operation was canceled")
-	ErrFetchFailed        = errors.New("failed to fetch items from source")
-	ErrResourceNotFound   = errors.New("resource not found in source system")
+	ErrSyncFailed       = errors.New("sync operation failed")
+	ErrSyncCanceled     = errors.New("sync operation was canceled")
+	ErrFetchFailed      = errors.New("failed to fetch items from source")
+	ErrResourceNotFound = errors.New("resource not found in source system")
 
 	// Knowledge base errors
 	ErrKnowledgeBaseNotFound = errors.New("knowledge base not found")
@@ -30,3 +34,17 @@ var (
 	// Sync log errors
 	ErrSyncLogNotFound = errors.New("sync log not found")
 )
+
+// PartialFetchError indicates that some resources were fetched successfully but
+// others failed. The caller should process Items (if any), persist an updated
+// cursor when provided, and surface Details to the user as a partial sync.
+type PartialFetchError struct {
+	Details []string
+}
+
+func (e *PartialFetchError) Error() string {
+	if e == nil || len(e.Details) == 0 {
+		return "partial fetch: some resources failed"
+	}
+	return fmt.Sprintf("partial fetch: %s", strings.Join(e.Details, "; "))
+}