From 35f717f7fa1f578d1a47e5def2412aee1b402ce2 Mon Sep 17 00:00:00 2001 From: wizardchen Date: Tue, 23 Jun 2026 16:03:38 +0800 Subject: [PATCH] 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. --- .../application/service/datasource_service.go | 30 +++++- .../datasource/connector/rss/connector.go | 68 +++++++++---- .../connector/rss/connector_test.go | 98 ++++++++++++++++++- internal/datasource/connector/rss/types.go | 56 ++++++++++- internal/datasource/errors.go | 28 +++++- 5 files changed, 251 insertions(+), 29 deletions(-) diff --git a/internal/application/service/datasource_service.go b/internal/application/service/datasource_service.go index c6c88329..1f01de5a 100644 --- a/internal/application/service/datasource_service.go +++ b/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) diff --git a/internal/datasource/connector/rss/connector.go b/internal/datasource/connector/rss/connector.go index ee99b689..b84dc6ac 100644 --- a/internal/datasource/connector/rss/connector.go +++ b/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 diff --git a/internal/datasource/connector/rss/connector_test.go b/internal/datasource/connector/rss/connector_test.go index e19cbd2c..ff74c7db 100644 --- a/internal/datasource/connector/rss/connector_test.go +++ b/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 /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"}, diff --git a/internal/datasource/connector/rss/types.go b/internal/datasource/connector/rss/types.go index fca5a4f4..b6f67264 100644 --- a/internal/datasource/connector/rss/types.go +++ b/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:" 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 diff --git a/internal/datasource/errors.go b/internal/datasource/errors.go index 7e0bb2c5..426c71ec 100644 --- a/internal/datasource/errors.go +++ b/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, "; ")) +}