From 4b09a34748a16e26ccdb8f01b529934a734098ee Mon Sep 17 00:00:00 2001 From: liangxu Date: Wed, 20 May 2026 11:22:37 +0800 Subject: [PATCH] refactor(knowledge): migrate deleted-item cleanup from polling scheduler to event-driven worker 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) --- server/cmd/scheduler/main.go | 7 - server/cmd/tenant-api/main.go | 13 + .../knowledge_deleted_cleanup_worker.go | 58 --- .../scheduler/monitoring_retention_worker.go | 2 +- .../app/knowledge_deleted_cleanup_worker.go | 398 ++++++++++++++++++ .../internal/tenant/app/knowledge_service.go | 99 +---- ..._knowledge_deleted_cleanup_events.down.sql | 4 + ...dd_knowledge_deleted_cleanup_events.up.sql | 28 ++ ...000_create_scheduler_control_tables.up.sql | 1 - ...dge_deleted_cleanup_scheduler_job.down.sql | 5 + ...ledge_deleted_cleanup_scheduler_job.up.sql | 8 + 11 files changed, 474 insertions(+), 149 deletions(-) delete mode 100644 server/internal/scheduler/knowledge_deleted_cleanup_worker.go create mode 100644 server/internal/tenant/app/knowledge_deleted_cleanup_worker.go create mode 100644 server/migrations/20260520130000_add_knowledge_deleted_cleanup_events.down.sql create mode 100644 server/migrations/20260520130000_add_knowledge_deleted_cleanup_events.up.sql create mode 100644 server/migrations_ops/20260520123000_remove_knowledge_deleted_cleanup_scheduler_job.down.sql create mode 100644 server/migrations_ops/20260520123000_remove_knowledge_deleted_cleanup_scheduler_job.up.sql diff --git a/server/cmd/scheduler/main.go b/server/cmd/scheduler/main.go index a1d1bf3..b62ac31 100644 --- a/server/cmd/scheduler/main.go +++ b/server/cmd/scheduler/main.go @@ -91,7 +91,6 @@ func main() { monitoringResultRecoveryWorker := internalscheduler.NewMonitoringResultRecoveryWorker(app.MonitoringDB, monitoringCallbackService, app.Logger) monitoringLeaseRecoveryWorker := internalscheduler.NewMonitoringLeaseRecoveryWorker(app.MonitoringDB, app.Logger) monitoringReceivedInspectionWorker := internalscheduler.NewMonitoringReceivedInspectionWorker(app.MonitoringDB, app.Logger) - knowledgeDeletedCleanupWorker := internalscheduler.NewKnowledgeDeletedCleanupWorker(knowledgeSvc) generationStateCheckWorker := generateworker.NewGenerationTaskStateCheckWorker(app.DB, app.Logger, app.Config().Generation). WithConfigProvider(app.ConfigStore). WithCache(app.Cache) @@ -128,12 +127,6 @@ func main() { return monitoringReceivedInspectionWorker.RunOnce(runCtx) }, }, - { - Key: "knowledge_deleted_cleanup", - Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { - return knowledgeDeletedCleanupWorker.RunOnce(runCtx) - }, - }, { Key: "generation_state_check", Run: func(runCtx context.Context, _ internalscheduler.JobRunContext) (map[string]any, error) { diff --git a/server/cmd/tenant-api/main.go b/server/cmd/tenant-api/main.go index bdc6ab0..e08f689 100644 --- a/server/cmd/tenant-api/main.go +++ b/server/cmd/tenant-api/main.go @@ -42,6 +42,19 @@ func main() { tenantapp.NewDesktopAccountHealthSinkWorker(app.DB, app.RabbitMQ, app.Logger).Start(workerCtx) imageService := tenantapp.NewImageService(app.DB, app.ObjectStorage, app.RabbitMQ, app.Logger) tenantapp.NewImageAssetWorker(imageService, app.RabbitMQ, app.Logger).Start(workerCtx) + knowledgeCleanupService := tenantapp.NewKnowledgeService( + app.DB, + app.RetrievalProvider, + app.VectorStore, + app.ObjectStorage, + app.LLM, + app.Logger, + app.Config().Retrieval, + app.Config().LLM, + 0, + 0, + ).WithConfigProvider(app.ConfigStore) + tenantapp.NewKnowledgeDeletedCleanupEventWorker(knowledgeCleanupService, app.Logger).Start(workerCtx) transport.RegisterRoutes(app) diff --git a/server/internal/scheduler/knowledge_deleted_cleanup_worker.go b/server/internal/scheduler/knowledge_deleted_cleanup_worker.go deleted file mode 100644 index 30eb50c..0000000 --- a/server/internal/scheduler/knowledge_deleted_cleanup_worker.go +++ /dev/null @@ -1,58 +0,0 @@ -package scheduler - -import ( - "context" - "time" - - tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app" -) - -type KnowledgeDeletedCleanupWorker struct { - service *tenantapp.KnowledgeService - interval time.Duration -} - -func NewKnowledgeDeletedCleanupWorker(service *tenantapp.KnowledgeService) *KnowledgeDeletedCleanupWorker { - return &KnowledgeDeletedCleanupWorker{ - service: service, - interval: 15 * time.Second, - } -} - -func (w *KnowledgeDeletedCleanupWorker) Start(ctx context.Context) { - go w.Run(ctx) -} - -func (w *KnowledgeDeletedCleanupWorker) Run(ctx context.Context) { - if w == nil || w.service == nil { - return - } - w.run(ctx) -} - -func (w *KnowledgeDeletedCleanupWorker) run(ctx context.Context) { - w.service.RunDeletedCleanupSweep() - - ticker := time.NewTicker(w.interval) - defer ticker.Stop() - - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - w.service.RunDeletedCleanupSweep() - } - } -} - -func (w *KnowledgeDeletedCleanupWorker) RunOnce(ctx context.Context) (map[string]any, error) { - if w == nil || w.service == nil { - return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil - } - startedAt := time.Now() - w.service.RunDeletedCleanupSweep() - return map[string]any{ - "duration_ms": time.Since(startedAt).Milliseconds(), - }, nil -} diff --git a/server/internal/scheduler/monitoring_retention_worker.go b/server/internal/scheduler/monitoring_retention_worker.go index 2b5c1ba..650a3a7 100644 --- a/server/internal/scheduler/monitoring_retention_worker.go +++ b/server/internal/scheduler/monitoring_retention_worker.go @@ -70,7 +70,7 @@ func (w *MonitoringRetentionWorker) RunOnce(ctx context.Context, run JobRunConte defer conn.Release() if statementTimeout > 0 { - if _, err := conn.Exec(ctx, `SET statement_timeout = $1`, int(statementTimeout.Milliseconds())); err != nil { + if _, err := conn.Exec(ctx, `SELECT set_config('statement_timeout', $1, false)`, fmt.Sprintf("%dms", statementTimeout.Milliseconds())); err != nil { return stats, fmt.Errorf("set statement timeout: %w", err) } defer conn.Exec(context.Background(), `RESET statement_timeout`) diff --git a/server/internal/tenant/app/knowledge_deleted_cleanup_worker.go b/server/internal/tenant/app/knowledge_deleted_cleanup_worker.go new file mode 100644 index 0000000..197d03d --- /dev/null +++ b/server/internal/tenant/app/knowledge_deleted_cleanup_worker.go @@ -0,0 +1,398 @@ +package app + +import ( + "context" + "database/sql" + "errors" + "fmt" + "os" + "strings" + "time" + + "github.com/jackc/pgx/v5" + "go.uber.org/zap" +) + +const ( + knowledgeDeletedCleanupNotifyChannel = "knowledge_deleted_cleanup_events" + defaultKnowledgeCleanupBatchSize = 50 + defaultKnowledgeCleanupLockTTL = 10 * time.Minute + defaultKnowledgeCleanupMaxWait = time.Hour + defaultKnowledgeCleanupRetryBase = 5 * time.Second + defaultKnowledgeCleanupRetryMax = time.Hour + defaultKnowledgeCleanupListenRetry = 5 * time.Second + defaultKnowledgeCleanupProcessTTL = 2 * time.Minute +) + +type knowledgeDeletedCleanupEvent struct { + ID int64 + TenantID int64 + ItemID int64 + AttemptCount int +} + +type KnowledgeDeletedCleanupEventWorker struct { + service *KnowledgeService + logger *zap.Logger + workerID string + batchSize int + lockTTL time.Duration + maxWait time.Duration + processTimeout time.Duration +} + +func NewKnowledgeDeletedCleanupEventWorker(service *KnowledgeService, logger *zap.Logger) *KnowledgeDeletedCleanupEventWorker { + hostname, _ := os.Hostname() + if hostname == "" { + hostname = "unknown" + } + return &KnowledgeDeletedCleanupEventWorker{ + service: service, + logger: logger, + workerID: fmt.Sprintf("%s:%d:%d", hostname, os.Getpid(), time.Now().UnixNano()), + batchSize: defaultKnowledgeCleanupBatchSize, + lockTTL: defaultKnowledgeCleanupLockTTL, + maxWait: defaultKnowledgeCleanupMaxWait, + processTimeout: defaultKnowledgeCleanupProcessTTL, + } +} + +func (w *KnowledgeDeletedCleanupEventWorker) Start(ctx context.Context) { + if w == nil { + return + } + go w.run(ctx) +} + +func (w *KnowledgeDeletedCleanupEventWorker) run(ctx context.Context) { + if w.service == nil || w.service.pool == nil { + return + } + + wake := make(chan struct{}, 1) + go w.listen(ctx, wake) + signalKnowledgeCleanup(wake) + + for { + processed, err := w.processDue(ctx) + if err != nil && ctx.Err() == nil && w.logger != nil { + w.logger.Warn("knowledge deleted cleanup worker failed", zap.Error(err)) + } + if processed > 0 { + continue + } + + waitFor := w.nextWait(ctx) + timer := time.NewTimer(waitFor) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-wake: + timer.Stop() + case <-timer.C: + } + } +} + +func (w *KnowledgeDeletedCleanupEventWorker) listen(ctx context.Context, wake chan<- struct{}) { + for { + if err := w.listenOnce(ctx, wake); err != nil && ctx.Err() == nil && w.logger != nil { + w.logger.Warn("knowledge deleted cleanup listener stopped", zap.Error(err)) + } + signalKnowledgeCleanup(wake) + + select { + case <-ctx.Done(): + return + case <-time.After(defaultKnowledgeCleanupListenRetry): + signalKnowledgeCleanup(wake) + } + } +} + +func (w *KnowledgeDeletedCleanupEventWorker) listenOnce(ctx context.Context, wake chan<- struct{}) error { + conn, err := w.service.pool.Acquire(ctx) + if err != nil { + return err + } + defer conn.Release() + + if _, err := conn.Exec(ctx, fmt.Sprintf("LISTEN %s", knowledgeDeletedCleanupNotifyChannel)); err != nil { + return err + } + defer func() { + unlistenCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, _ = conn.Exec(unlistenCtx, fmt.Sprintf("UNLISTEN %s", knowledgeDeletedCleanupNotifyChannel)) + }() + + for { + if _, err := conn.Conn().WaitForNotification(ctx); err != nil { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return nil + } + return err + } + signalKnowledgeCleanup(wake) + } +} + +func (w *KnowledgeDeletedCleanupEventWorker) processDue(parent context.Context) (int, error) { + events, err := w.service.claimDeletedCleanupEvents(parent, w.workerID, w.batchSize, w.lockTTL) + if err != nil || len(events) == 0 { + return len(events), err + } + + for _, event := range events { + ctx, cancel := context.WithTimeout(parent, w.processTimeout) + err := w.service.ProcessDeletedKnowledgeCleanup(ctx, event.TenantID, event.ItemID) + cancel() + + if err != nil { + if markErr := w.service.failDeletedCleanupEvent(parent, event.ID, err.Error(), retryDelayForKnowledgeCleanup(event.AttemptCount)); markErr != nil && w.logger != nil { + w.logger.Warn("knowledge deleted cleanup event fail mark failed", + zap.Error(markErr), + zap.Int64("event_id", event.ID), + ) + } + if w.logger != nil { + w.logger.Warn("knowledge deleted cleanup event failed", + zap.Error(err), + zap.Int64("event_id", event.ID), + zap.Int64("tenant_id", event.TenantID), + zap.Int64("knowledge_item_id", event.ItemID), + ) + } + continue + } + + if err := w.service.completeDeletedCleanupEvent(parent, event.ID); err != nil && w.logger != nil { + w.logger.Warn("knowledge deleted cleanup event complete mark failed", + zap.Error(err), + zap.Int64("event_id", event.ID), + ) + } + } + return len(events), nil +} + +func (w *KnowledgeDeletedCleanupEventWorker) nextWait(ctx context.Context) time.Duration { + waitFor := w.maxWait + if waitFor <= 0 { + waitFor = defaultKnowledgeCleanupMaxWait + } + + nextAt, err := w.service.nextDeletedCleanupDueAt(ctx, w.lockTTL) + if err != nil { + if w.logger != nil { + w.logger.Warn("knowledge deleted cleanup next due query failed", zap.Error(err)) + } + return minDuration(waitFor, time.Minute) + } + if nextAt == nil { + return waitFor + } + + until := time.Until(*nextAt) + if until <= 0 { + return time.Millisecond + } + return minDuration(until, waitFor) +} + +func (s *KnowledgeService) enqueueDeletedKnowledgeCleanupEvents(ctx context.Context, tx pgx.Tx, tenantID int64, itemIDs []int64) error { + if s == nil || tx == nil || tenantID <= 0 || len(itemIDs) == 0 { + return nil + } + + ids := normalizeInt64IDs(itemIDs) + if len(ids) == 0 { + return nil + } + + if _, err := tx.Exec(ctx, ` + INSERT INTO knowledge_deleted_cleanup_events (tenant_id, knowledge_item_id) + SELECT $1, item_id + FROM unnest($2::bigint[]) AS item_ids(item_id) + WHERE item_id > 0 + ON CONFLICT (tenant_id, knowledge_item_id) WHERE completed_at IS NULL + DO UPDATE SET status = 'pending', + available_at = NOW(), + locked_by = NULL, + locked_at = NULL, + last_error = NULL, + updated_at = NOW() + `, tenantID, ids); err != nil { + return err + } + + _, err := tx.Exec(ctx, `SELECT pg_notify($1, '')`, knowledgeDeletedCleanupNotifyChannel) + return err +} + +func (s *KnowledgeService) ProcessDeletedKnowledgeCleanup(ctx context.Context, tenantID, itemID int64) error { + if s == nil { + return nil + } + if !s.startCleanupJob(tenantID, itemID) { + return nil + } + defer s.finishCleanupJob(tenantID, itemID) + return s.cleanupDeletedKnowledgeItem(ctx, tenantID, itemID) +} + +func (s *KnowledgeService) claimDeletedCleanupEvents( + ctx context.Context, + workerID string, + batchSize int, + lockTTL time.Duration, +) ([]knowledgeDeletedCleanupEvent, error) { + if s == nil || s.pool == nil { + return nil, nil + } + if batchSize <= 0 { + batchSize = defaultKnowledgeCleanupBatchSize + } + if lockTTL <= 0 { + lockTTL = defaultKnowledgeCleanupLockTTL + } + + staleBefore := time.Now().UTC().Add(-lockTTL) + rows, err := s.pool.Query(ctx, ` + WITH next_events AS ( + SELECT id + FROM knowledge_deleted_cleanup_events + WHERE completed_at IS NULL + AND ( + (status IN ('pending', 'failed') AND available_at <= NOW()) + OR (status = 'processing' AND locked_at < $2) + ) + ORDER BY available_at ASC, id ASC + FOR UPDATE SKIP LOCKED + LIMIT $1 + ) + UPDATE knowledge_deleted_cleanup_events e + SET status = 'processing', + attempt_count = attempt_count + 1, + locked_by = $3, + locked_at = NOW(), + updated_at = NOW() + FROM next_events + WHERE e.id = next_events.id + RETURNING e.id, e.tenant_id, e.knowledge_item_id, e.attempt_count + `, batchSize, staleBefore, strings.TrimSpace(workerID)) + if err != nil { + return nil, err + } + defer rows.Close() + + events := make([]knowledgeDeletedCleanupEvent, 0, batchSize) + for rows.Next() { + var event knowledgeDeletedCleanupEvent + if err := rows.Scan(&event.ID, &event.TenantID, &event.ItemID, &event.AttemptCount); err != nil { + return nil, err + } + events = append(events, event) + } + return events, rows.Err() +} + +func (s *KnowledgeService) completeDeletedCleanupEvent(ctx context.Context, eventID int64) error { + if s == nil || s.pool == nil || eventID <= 0 { + return nil + } + _, err := s.pool.Exec(ctx, ` + UPDATE knowledge_deleted_cleanup_events + SET status = 'completed', + completed_at = NOW(), + locked_by = NULL, + locked_at = NULL, + last_error = NULL, + updated_at = NOW() + WHERE id = $1 + AND completed_at IS NULL + `, eventID) + return err +} + +func (s *KnowledgeService) failDeletedCleanupEvent(ctx context.Context, eventID int64, message string, delay time.Duration) error { + if s == nil || s.pool == nil || eventID <= 0 { + return nil + } + if delay <= 0 { + delay = defaultKnowledgeCleanupRetryBase + } + nextAt := time.Now().UTC().Add(delay) + _, err := s.pool.Exec(ctx, ` + UPDATE knowledge_deleted_cleanup_events + SET status = 'failed', + available_at = $2, + locked_by = NULL, + locked_at = NULL, + last_error = $3, + updated_at = NOW() + WHERE id = $1 + AND completed_at IS NULL + `, eventID, nextAt, strings.TrimSpace(message)) + return err +} + +func (s *KnowledgeService) nextDeletedCleanupDueAt(ctx context.Context, lockTTL time.Duration) (*time.Time, error) { + if s == nil || s.pool == nil { + return nil, nil + } + if lockTTL <= 0 { + lockTTL = defaultKnowledgeCleanupLockTTL + } + + var due sql.NullTime + err := s.pool.QueryRow(ctx, ` + SELECT MIN(due_at) + FROM ( + SELECT available_at AS due_at + FROM knowledge_deleted_cleanup_events + WHERE completed_at IS NULL + AND status IN ('pending', 'failed') + UNION ALL + SELECT locked_at + ($1::double precision * INTERVAL '1 second') AS due_at + FROM knowledge_deleted_cleanup_events + WHERE completed_at IS NULL + AND status = 'processing' + AND locked_at IS NOT NULL + ) due_events + `, lockTTL.Seconds()).Scan(&due) + if err != nil { + return nil, err + } + if !due.Valid { + return nil, nil + } + return &due.Time, nil +} + +func retryDelayForKnowledgeCleanup(attempt int) time.Duration { + if attempt <= 1 { + return defaultKnowledgeCleanupRetryBase + } + shift := attempt - 1 + if shift > 8 { + shift = 8 + } + delay := defaultKnowledgeCleanupRetryBase * time.Duration(1<= 0) +); + +CREATE UNIQUE INDEX IF NOT EXISTS uk_knowledge_deleted_cleanup_open + ON knowledge_deleted_cleanup_events (tenant_id, knowledge_item_id) + WHERE completed_at IS NULL; + +CREATE INDEX IF NOT EXISTS idx_knowledge_deleted_cleanup_pending + ON knowledge_deleted_cleanup_events (available_at, id) + WHERE completed_at IS NULL AND status IN ('pending', 'failed'); + +CREATE INDEX IF NOT EXISTS idx_knowledge_deleted_cleanup_processing + ON knowledge_deleted_cleanup_events (locked_at, id) + WHERE completed_at IS NULL AND status = 'processing'; diff --git a/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql b/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql index 802c627..b282d44 100644 --- a/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql +++ b/server/migrations_ops/20260520120000_create_scheduler_control_tables.up.sql @@ -93,7 +93,6 @@ VALUES ('monitoring_result_recovery', '监控结果恢复', 'monitoring', '恢复已回调但未成功进入结果队列的监控采集结果。', true, 'interval', 60, 'Asia/Shanghai', 30, 100, '{}'::jsonb), ('monitoring_lease_recovery', '监控租约回收', 'monitoring', '回收超过租约时间仍未回调的监控采集任务。', true, 'interval', 1800, 'Asia/Shanghai', 30, NULL, '{}'::jsonb), ('monitoring_received_inspection', '监控 received 卡死检查', 'monitoring', '检查长时间停留在 received 状态的监控任务。', true, 'interval', 300, 'Asia/Shanghai', 30, 100, '{}'::jsonb), - ('knowledge_deleted_cleanup', '知识库删除清理', 'knowledge', '清理已删除知识库条目的异步对象与向量数据。', true, 'interval', 15, 'Asia/Shanghai', 30, NULL, '{}'::jsonb), ('generation_state_check', '生成任务状态巡检', 'generation', '低频巡检文章生成任务与文章状态一致性。', true, 'interval', 300, 'Asia/Shanghai', 10, 100, '{}'::jsonb), ('monitoring_retention_cleanup', '监控历史保留清理', 'monitoring', '按业务日期清理最近 30 天窗口以前的监控历史数据。', true, 'interval', 86400, 'Asia/Shanghai', 900, 5000, '{"retention_days":30,"batch_size":5000,"max_batches_per_run":200,"dry_run":false,"statement_timeout":"5s","run_at_local":"02:30"}'::jsonb) diff --git a/server/migrations_ops/20260520123000_remove_knowledge_deleted_cleanup_scheduler_job.down.sql b/server/migrations_ops/20260520123000_remove_knowledge_deleted_cleanup_scheduler_job.down.sql new file mode 100644 index 0000000..02ff6f1 --- /dev/null +++ b/server/migrations_ops/20260520123000_remove_knowledge_deleted_cleanup_scheduler_job.down.sql @@ -0,0 +1,5 @@ +INSERT INTO ops.scheduler_jobs + (job_key, display_name, category, description, enabled, schedule_type, interval_seconds, timezone, timeout_seconds, batch_size, config) +VALUES + ('knowledge_deleted_cleanup', '知识库删除清理', 'knowledge', '清理已删除知识库条目的异步对象与向量数据。', true, 'interval', 15, 'Asia/Shanghai', 30, NULL, '{}'::jsonb) +ON CONFLICT (job_key) DO NOTHING; diff --git a/server/migrations_ops/20260520123000_remove_knowledge_deleted_cleanup_scheduler_job.up.sql b/server/migrations_ops/20260520123000_remove_knowledge_deleted_cleanup_scheduler_job.up.sql new file mode 100644 index 0000000..58fd391 --- /dev/null +++ b/server/migrations_ops/20260520123000_remove_knowledge_deleted_cleanup_scheduler_job.up.sql @@ -0,0 +1,8 @@ +DELETE FROM ops.scheduler_job_triggers +WHERE job_key = 'knowledge_deleted_cleanup'; + +DELETE FROM ops.scheduler_job_runs +WHERE job_key = 'knowledge_deleted_cleanup'; + +DELETE FROM ops.scheduler_jobs +WHERE job_key = 'knowledge_deleted_cleanup';