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

@@ -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

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)