fix(notion): 保留正文抓取失败时的同步游标 (#3693)

internal/datasource/connector/notion/block_retry_test.go

@@ -0,0 +1,168 @@
+package notion
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/Tencent/WeKnora/internal/types"
+)
+
+const retryPageTime = "2026-09-24T10:00:00Z"
+
+type blockRetryFixture struct {
+	failed atomic.Bool
+	calls  atomic.Int32
+	mode   string
+}
+
+func (f *blockRetryFixture) serve(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Content-Type", "application/json")
+	switch r.URL.Path {
+	case "/v1/search":
+		_, _ = w.Write([]byte(`{"results":[` + retryPage("page", "workspace") + `],"has_more":false}`))
+	case "/v1/pages/page":
+		_, _ = w.Write([]byte(retryPage("page", "workspace")))
+	case "/v1/pages/healthy":
+		_, _ = w.Write([]byte(retryPage("healthy", "workspace")))
+	case "/v1/pages/child":
+		_, _ = w.Write([]byte(retryPage("child", "page_id")))
+	case "/v1/blocks/page/children":
+		f.pageBlocks(w, r)
+	case "/v1/blocks/child/children":
+		f.content(w)
+	case "/v1/blocks/healthy/children":
+		_, _ = w.Write([]byte(`{"results":[{"id":"text","type":"paragraph","paragraph":{"rich_text":[` +
+			`{"type":"text","text":{"content":"healthy content"},"plain_text":"healthy content"}]}}]}`))
+	default:
+		http.NotFound(w, r)
+	}
+}
+
+func retryPage(id, parentType string) string {
+	return `{"id":"` + id + `","object":"page","last_edited_time":"` + retryPageTime +
+		`","parent":{"type":"` + parentType + `","page_id":"page"}}`
+}
+
+func (f *blockRetryFixture) pageBlocks(w http.ResponseWriter, r *http.Request) {
+	switch f.mode {
+	case "child":
+		_, _ = w.Write([]byte(`{"results":[{"id":"child","type":"child_page",` +
+			`"child_page":{"title":"Child"}}],"has_more":false}`))
+	case "later_page":
+		if r.URL.Query().Get("start_cursor") == "later" {
+			f.content(w)
+			return
+		}
+		_, _ = w.Write([]byte(`{"results":[],"has_more":true,"next_cursor":"later"}`))
+	case "empty":
+		_, _ = w.Write([]byte(`{"results":[],"has_more":false}`))
+	default:
+		f.content(w)
+	}
+}
+
+func (f *blockRetryFixture) content(w http.ResponseWriter) {
+	f.calls.Add(1)
+	if f.failed.Load() {
+		status := http.StatusUnauthorized
+		if f.mode == "server_error" {
+			status = http.StatusServiceUnavailable
+		}
+		w.WriteHeader(status)
+		_, _ = w.Write([]byte(`{"code":"unauthorized","message":"test block request rejected"}`))
+		return
+	}
+	_, _ = w.Write([]byte(`{"results":[{"id":"text","type":"paragraph","paragraph":{"rich_text":[` +
+		`{"type":"text","text":{"content":"recovered content"},"plain_text":"recovered content"}` +
+		`]}}],"has_more":false}`))
+}
+
+func newBlockRetryFixture(t *testing.T, mode string) (*blockRetryFixture, *types.DataSourceConfig) {
+	t.Helper()
+	allowNotionTestServer(t)
+	fixture := &blockRetryFixture{mode: mode}
+	fixture.failed.Store(true)
+	server := httptest.NewServer(http.HandlerFunc(fixture.serve))
+	t.Cleanup(server.Close)
+	return fixture, makeNotionConfig(&Config{APIKey: "test-token"}, server.URL, []string{"page"})
+}
+
+func TestNotionBlockFailurePreservesRetry(t *testing.T) {
+	for _, mode := range []string{"page", "server_error", "later_page", "child"} {
+		t.Run(mode, func(t *testing.T) {
+			fixture, config := newBlockRetryFixture(t, mode)
+			oldTime := time.Date(2026, 9, 23, 10, 0, 0, 0, time.UTC)
+			previous := buildCursor(map[string]time.Time{"page": oldTime, "removed": oldTime})
+			before, err := json.Marshal(previous)
+			require.NoError(t, err)
+			connector := NewConnector()
+			items, next, err := connector.FetchIncremental(context.Background(), config, previous)
+			require.ErrorContains(t, err, "get blocks")
+			require.Nil(t, items, "failed traversal must not emit content or deletion events")
+			require.Nil(t, next, "failed traversal must not acknowledge the new edit time")
+			after, marshalErr := json.Marshal(previous)
+			require.NoError(t, marshalErr)
+			require.Equal(t, before, after)
+			require.Positive(t, fixture.calls.Load())
+
+			fixture.failed.Store(false)
+			items, next, err = connector.FetchIncremental(context.Background(), config, previous)
+			require.NoError(t, err)
+			require.NotNil(t, next)
+			var foundContent, foundDeletion bool
+			for _, item := range items {
+				foundContent = foundContent || strings.Contains(string(item.Content), "recovered content")
+				foundDeletion = foundDeletion || (item.ExternalID == "removed" && item.IsDeleted)
+			}
+			require.True(t, foundContent, "recovered page must be fetched without another source edit")
+			require.True(t, foundDeletion, "successful traversal must still detect actual deletions")
+			calls := fixture.calls.Load()
+			items, _, err = connector.FetchIncremental(context.Background(), config, next)
+			require.NoError(t, err)
+			require.Empty(t, items)
+			require.Equal(t, calls, fixture.calls.Load(), "unchanged page must not be fetched again")
+		})
+	}
+}
+
+func TestNotionFirstSyncRejectsBlockFailure(t *testing.T) {
+	fixture, config := newBlockRetryFixture(t, "page")
+	connector := NewConnector()
+	items, next, err := connector.FetchIncremental(context.Background(), config, nil)
+	require.Error(t, err)
+	require.Nil(t, items)
+	require.Nil(t, next, "first sync must not substitute time.Now for a failed page")
+	fixture.failed.Store(false)
+	items, next, err = connector.FetchIncremental(context.Background(), config, nil)
+	require.NoError(t, err)
+	require.Len(t, items, 1)
+	require.NotNil(t, next)
+}
+
+func TestNotionFullSyncRejectsBlockFailure(t *testing.T) {
+	_, config := newBlockRetryFixture(t, "child")
+	config.ResourceIDs = []string{"healthy", "page"}
+	items, err := NewConnector().FetchAll(context.Background(), config, config.ResourceIDs)
+	require.Error(t, err)
+	require.Nil(t, items, "discard earlier successful items when a later page fails")
+}
+
+func TestNotionEmptyPageStillSucceeds(t *testing.T) {
+	_, config := newBlockRetryFixture(t, "empty")
+	previous := buildCursor(map[string]time.Time{"page": {}})
+	items, next, err := NewConnector().FetchIncremental(context.Background(), config, previous)
+	require.NoError(t, err)
+	require.Len(t, items, 1)
+	require.Equal(t, "page", items[0].ExternalID)
+	require.False(t, items[0].IsDeleted)
+	require.Equal(t, "# "+defaultUntitledName+"\n", string(items[0].Content))
+	require.NotNil(t, next)
+}

internal/datasource/connector/notion/connector.go

@@ -148,7 +148,10 @@ func (c *Connector) FetchAll(ctx context.Context, config *types.DataSourceConfig
 	for _, resourceID := range resourceIDs {
 		page, err := client.GetPage(ctx, resourceID)
 		if err == nil {
-			pageItems, _ := c.fetchPage(ctx, client, page, visited)
+			pageItems, _, err := c.fetchPage(ctx, client, page, visited)
+			if err != nil {
+				return nil, err
+			}
 			allItems = append(allItems, pageItems...)
 			continue
 		}
@@ -268,7 +271,10 @@ func (c *Connector) FetchIncremental(ctx context.Context, config *types.DataSour
 				newEditTimes[rid] = rt
 			}
 		} else {
-			items, truncated := c.fetchPage(ctx, client, pg, fetchVisited)
+			items, truncated, err := c.fetchPage(ctx, client, pg, fetchVisited)
+			if err != nil {
+				return nil, nil, err
+			}
 			queryTruncated = queryTruncated || truncated
 			changedItems = append(changedItems, items...)
 		}
@@ -334,19 +340,20 @@ func buildCursor(editTimes map[string]time.Time) *types.SyncCursor {
 // If page is nil, it will be fetched from the API.
 // The second result reports whether any query issued for this page (including
 // child databases) was truncated at a vendor limit.
+// A page block fetch error aborts the round without acknowledging its edit time.
 func (c *Connector) fetchPage(
 	ctx context.Context, client *notionClient, page *notionPage, visited map[string]bool,
-) ([]types.FetchedItem, bool) {
+) ([]types.FetchedItem, bool, error) {
 	if page == nil {
-		return nil, false
+		return nil, false, nil
 	}
 	if visited[page.ID] {
-		return nil, false
+		return nil, false, nil
 	}
 	visited[page.ID] = true
 
 	if page.InTrash {
-		return nil, false
+		return nil, false, nil
 	}
 
 	// Database records store content in properties, not blocks — delegate to
@@ -362,15 +369,14 @@ func (c *Connector) fetchPage(
 		}
 		propNames := extractPropertySchema(*page)
 		if item := c.buildRecordItem(ctx, client, *page, propNames, dbTitle); item != nil {
-			return []types.FetchedItem{*item}, false
+			return []types.FetchedItem{*item}, false, nil
 		}
-		return nil, false
+		return nil, false, nil
 	}
 
 	blocks, err := client.GetBlockChildrenAll(ctx, page.ID)
 	if err != nil {
-		logger.Warnf(ctx, "[Notion] failed to get blocks for page %s: %v", page.ID, err)
-		return nil, false
+		return nil, false, fmt.Errorf("get blocks for notion page %s: %w", page.ID, err)
 	}
 
 	resolveFileUploads(ctx, client, blocks)
@@ -445,7 +451,10 @@ func (c *Connector) fetchPage(
 				logger.Warnf(ctx, "[Notion] failed to get child page %s: %v", block.ID, err)
 				continue
 			}
-			childItems, childTruncated := c.fetchPage(ctx, client, childPage, visited)
+			childItems, childTruncated, err := c.fetchPage(ctx, client, childPage, visited)
+			if err != nil {
+				return nil, truncated || childTruncated, err
+			}
 			truncated = truncated || childTruncated
 			items = append(items, childItems...)
 		case "child_database":
@@ -455,7 +464,7 @@ func (c *Connector) fetchPage(
 		}
 	}
 
-	return items, truncated
+	return items, truncated, nil
 }
 
 // fetchDatabase syncs each database record as an individual knowledge item (full sync).

internal/datasource/connector/notion/query_truncation_test.go

@@ -233,6 +233,7 @@ func TestFetchPagePropagatesTruncationFromChildDatabase(t *testing.T) {
 		Parent: notionParent{Type: parentTypeWorkspace},
 	}
 
-	_, truncated := NewConnector().fetchPage(context.Background(), client, page, map[string]bool{})
+	_, truncated, err := NewConnector().fetchPage(context.Background(), client, page, map[string]bool{})
+	require.NoError(t, err)
 	require.True(t, truncated, "a capped child database query must reach the incremental round")
 }