root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(knowledge): migrate deleted-item cleanup from polling scheduler to event-driven worker

Browse Source 
Replaces the scheduler-polled KnowledgeDeletedCleanupWorker with an
event-driven KnowledgeDeletedCleanupEventWorker in tenant-api; cleanup
events are now enqueued transactionally at delete time instead of swept
periodically. Also fixes statement_timeout parameterization in
MonitoringRetentionWorker and propagates errors from cleanupDeletedKnowledgeItem.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
liangxu
2026-05-20 11:22:37 +08:00
parent e82ae56236
commit 4b09a34748
11 changed files with 474 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/knowledge_service.go
+17 -82
View File
@@ -30,7 +30,6 @@ import (
const (
defaultKnowledgeQueueSize = 128
defaultKnowledgeCleanupPoll = 15 * time.Second
defaultKnowledgeChildGroupName = "默认目录"
maxKnowledgeSnippetsPerItem = 2
maxKnowledgeTextNameRunes = 20
@@ -59,11 +58,6 @@ type knowledgeParseJob struct {
ParseTaskID int64
}
type knowledgeCleanupJob struct {
TenantID int64
ItemID int64
}
type knowledgeItemRecord struct {
ID int64
TenantID int64
@@ -459,12 +453,12 @@ func (s *KnowledgeService) DeleteGroup(ctx context.Context, groupID int64) error
return response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to delete knowledge groups")
}
if err := tx.Commit(ctx); err != nil {
return response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to commit delete transaction")
if err := s.enqueueDeletedKnowledgeCleanupEvents(ctx, tx, actor.TenantID, itemIDs); err != nil {
return response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to enqueue knowledge cleanup")
}
for _, itemID := range itemIDs {
go s.cleanupDeletedKnowledgeItemAsync(actor.TenantID, itemID)
if err := tx.Commit(ctx); err != nil {
return response.ErrInternal(50054, "knowledge_group_delete_failed", "failed to commit delete transaction")
}
if s.logger != nil {
@@ -720,6 +714,10 @@ func (s *KnowledgeService) DeleteItem(ctx context.Context, itemID int64) error {
return response.ErrInternal(50054, "knowledge_item_delete_failed", "failed to mark knowledge chunks for deletion")
}
if err := s.enqueueDeletedKnowledgeCleanupEvents(ctx, tx, actor.TenantID, []int64{itemID}); err != nil {
return response.ErrInternal(50054, "knowledge_item_delete_failed", "failed to enqueue knowledge cleanup")
}
if err := tx.Commit(ctx); err != nil {
return response.ErrInternal(50054, "knowledge_item_delete_failed", "failed to commit delete transaction")
}
@@ -730,7 +728,6 @@ func (s *KnowledgeService) DeleteItem(ctx context.Context, itemID int64) error {
zap.Int64("knowledge_item_id", itemID),
)
}
go s.cleanupDeletedKnowledgeItemAsync(actor.TenantID, itemID)
return nil
}
@@ -2558,72 +2555,6 @@ func (s *KnowledgeService) filterActiveKnowledgeSnippets(
return filtered, nil
}
func (s *KnowledgeService) cleanupDeletedKnowledgeItemAsync(tenantID, itemID int64) {
if !s.startCleanupJob(tenantID, itemID) {
return
}
go func() {
defer s.finishCleanupJob(tenantID, itemID)
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
s.cleanupDeletedKnowledgeItem(ctx, tenantID, itemID)
}()
}
func (s *KnowledgeService) scheduleDeletedKnowledgeCleanup() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
jobs, err := s.listKnowledgeCleanupCandidates(ctx, 20)
if err != nil {
return
}
for _, job := range jobs {
s.cleanupDeletedKnowledgeItemAsync(job.TenantID, job.ItemID)
}
}
func (s *KnowledgeService) RunDeletedCleanupSweep() {
if s == nil {
return
}
s.scheduleDeletedKnowledgeCleanup()
}
func (s *KnowledgeService) listKnowledgeCleanupCandidates(
ctx context.Context,
limit int,
) ([]knowledgeCleanupJob, error) {
if limit <= 0 {
limit = 20
}
rows, err := s.pool.Query(ctx, `
SELECT ki.tenant_id, ki.id
FROM knowledge_items ki
WHERE ki.deleted_at IS NOT NULL
ORDER BY ki.updated_at ASC, ki.id ASC
LIMIT $1
`, limit)
if err != nil {
return nil, err
}
defer rows.Close()
jobs := make([]knowledgeCleanupJob, 0, limit)
for rows.Next() {
var job knowledgeCleanupJob
if err := rows.Scan(&job.TenantID, &job.ItemID); err != nil {
return nil, err
}
jobs = append(jobs, job)
}
return jobs, nil
}
func (s *KnowledgeService) startCleanupJob(tenantID, itemID int64) bool {
key := fmt.Sprintf("%d:%d", tenantID, itemID)
@@ -2645,10 +2576,10 @@ func (s *KnowledgeService) finishCleanupJob(tenantID, itemID int64) {
s.cleanupMu.Unlock()
}
func (s *KnowledgeService) cleanupDeletedKnowledgeItem(ctx context.Context, tenantID, itemID int64) {
func (s *KnowledgeService) cleanupDeletedKnowledgeItem(ctx context.Context, tenantID, itemID int64) error {
item, err := s.loadKnowledgeItemForCleanup(ctx, tenantID, itemID)
if err != nil || item == nil {
return
return err
}
errs := make([]string, 0, 2)
@@ -2681,7 +2612,7 @@ func (s *KnowledgeService) cleanupDeletedKnowledgeItem(ctx context.Context, tena
)
}
_ = s.setKnowledgeCleanupError(ctx, tenantID, itemID, strings.Join(errs, "; "))
return
return fmt.Errorf("knowledge cleanup failed: %s", strings.Join(errs, "; "))
}
if err := s.deleteKnowledgeItemPermanently(ctx, tenantID, itemID); err != nil {
@@ -2693,7 +2624,7 @@ func (s *KnowledgeService) cleanupDeletedKnowledgeItem(ctx context.Context, tena
)
}
_ = s.setKnowledgeCleanupError(ctx, tenantID, itemID, err.Error())
return
return err
}
if s.logger != nil {
@@ -2703,6 +2634,7 @@ func (s *KnowledgeService) cleanupDeletedKnowledgeItem(ctx context.Context, tena
)
}
_ = s.clearKnowledgeCleanupError(ctx, tenantID, itemID)
return nil
}
func (s *KnowledgeService) loadKnowledgeItemForCleanup(
@@ -2732,7 +2664,10 @@ func (s *KnowledgeService) loadKnowledgeItemForCleanup(
&record.ItemVersion,
)
if err != nil {
return nil, nil
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
return nil, err
}
record.ContentText = contentText
record.Markdown = markdownContent