From c34f7b6254d892f02c8c1cc56ef079dfab4491f0 Mon Sep 17 00:00:00 2001
From: wizardchen <wizardchen@tencent.com>
Date: Tue, 28 Apr 2026 17:12:21 +0800
Subject: [PATCH] Enhance wiki ingest process to handle failed operations

- Introduced a mechanism to track and requeue failed operations during the wiki ingest process.
- Added a new `requeueFailedOps` function to append failed operations back to the Redis pending list for retry in subsequent batches.
- Updated the `ProcessWikiIngest` method to collect failed operations and ensure they are retried after trimming the pending list.

This change improves the robustness of the ingest process by preventing data loss from transient failures.
---
 internal/application/service/wiki_ingest.go   | 24 ++++++++++++++++++-
 .../application/service/wiki_ingest_batch.go  |  9 +++++++
 2 files changed, 32 insertions(+), 1 deletion(-)

diff --git a/internal/application/service/wiki_ingest.go b/internal/application/service/wiki_ingest.go
index 603c77b65..984bc4381 100644
--- a/internal/application/service/wiki_ingest.go
+++ b/internal/application/service/wiki_ingest.go
@@ -322,7 +322,7 @@ func (s *wikiIngestService) peekPendingList(ctx context.Context, kbID string) ([
 		unique = append(unique, reversedUnique[i])
 	}
 
-	return unique, len(ops)
+	return unique, len(result)
 }
 
 // trimPendingList removes the first `count` items from the Redis pending list.
@@ -336,6 +336,28 @@ func (s *wikiIngestService) trimPendingList(ctx context.Context, kbID string, co
 	}
 }
 
+// requeueFailedOps appends failed operations back to the pending list so they
+// are retried in the next follow-up batch. Called after trimPendingList has
+// already removed the consumed batch head.
+func (s *wikiIngestService) requeueFailedOps(ctx context.Context, kbID string, ops []WikiPendingOp) {
+	if s.redisClient == nil {
+		return
+	}
+	pendingKey := wikiPendingKeyPrefix + kbID
+	for _, op := range ops {
+		data, err := json.Marshal(op)
+		if err != nil {
+			logger.Warnf(ctx, "wiki ingest: failed to marshal op for requeue: %v", err)
+			continue
+		}
+		if err := s.redisClient.RPush(ctx, pendingKey, string(data)).Err(); err != nil {
+			logger.Warnf(ctx, "wiki ingest: failed to requeue op %s: %v", op.KnowledgeID, err)
+			continue
+		}
+		logger.Infof(ctx, "wiki ingest: re-queued failed op %s (%s) for retry", op.KnowledgeID, op.DocTitle)
+	}
+}
+
 // docIngestResult captures per-document info for batch post-processing.
 type docIngestResult struct {
 	KnowledgeID string
diff --git a/internal/application/service/wiki_ingest_batch.go b/internal/application/service/wiki_ingest_batch.go
index c07d7736b..638a0aa2b 100644
--- a/internal/application/service/wiki_ingest_batch.go
+++ b/internal/application/service/wiki_ingest_batch.go
@@ -228,6 +228,7 @@ func (s *wikiIngestService) ProcessWikiIngest(ctx context.Context, t *asynq.Task
 
 	// 1. MAP PHASE (Parallel extraction and generation of updates)
 	var mapMu sync.Mutex
+	var failedOps []WikiPendingOp
 	slugUpdates := make(map[string][]SlugUpdate)
 	var docResults []*docIngestResult
 	var retractChangeDesc strings.Builder
@@ -310,6 +311,7 @@ func (s *wikiIngestService) ProcessWikiIngest(ctx context.Context, t *asynq.Task
 			if err != nil {
 				mapMu.Lock()
 				ingestFailed++
+				failedOps = append(failedOps, op)
 				mapMu.Unlock()
 				logger.Warnf(mapCtx, "wiki ingest: failed to map knowledge %s: %v", op.KnowledgeID, err)
 				return nil // Don't fail the whole batch
@@ -413,6 +415,13 @@ func (s *wikiIngestService) ProcessWikiIngest(ctx context.Context, t *asynq.Task
 
 	s.trimPendingList(ctx, payload.KnowledgeBaseID, peekedCount)
 
+	// Re-enqueue failed ops so they get retried in the next follow-up batch.
+	// This must happen after trim: trim removes the consumed batch head, then
+	// requeue appends the failed items to the tail for a future attempt.
+	if len(failedOps) > 0 {
+		s.requeueFailedOps(ctx, payload.KnowledgeBaseID, failedOps)
+	}
+
 	logger.Infof(ctx, "wiki ingest: batch completed for KB %s, %d ops, %d pages affected", payload.KnowledgeBaseID, len(pendingOps), len(allPagesAffected))
 
 	followUpScheduled = s.scheduleFollowUp(ctx, payload)